set_api.c 35 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255
  1. /*
  2. This file is part of GNUnet.
  3. Copyright (C) 2012-2016 GNUnet e.V.
  4. GNUnet is free software: you can redistribute it and/or modify it
  5. under the terms of the GNU Affero General Public License as published
  6. by the Free Software Foundation, either version 3 of the License,
  7. or (at your option) any later version.
  8. GNUnet is distributed in the hope that it will be useful, but
  9. WITHOUT ANY WARRANTY; without even the implied warranty of
  10. MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  11. Affero General Public License for more details.
  12. You should have received a copy of the GNU Affero General Public License
  13. along with this program. If not, see <http://www.gnu.org/licenses/>.
  14. SPDX-License-Identifier: AGPL3.0-or-later
  15. */
  16. /**
  17. * @file set/set_api.c
  18. * @brief api for the set service
  19. * @author Florian Dold
  20. * @author Christian Grothoff
  21. */
  22. #include "platform.h"
  23. #include "gnunet_util_lib.h"
  24. #include "gnunet_protocols.h"
  25. #include "gnunet_set_service.h"
  26. #include "set.h"
  27. #define LOG(kind,...) GNUNET_log_from (kind, "set-api",__VA_ARGS__)
  28. struct SetCopyRequest
  29. {
  30. struct SetCopyRequest *next;
  31. struct SetCopyRequest *prev;
  32. void *cls;
  33. GNUNET_SET_CopyReadyCallback cb;
  34. };
  35. /**
  36. * Opaque handle to a set.
  37. */
  38. struct GNUNET_SET_Handle
  39. {
  40. /**
  41. * Message queue for @e client.
  42. */
  43. struct GNUNET_MQ_Handle *mq;
  44. /**
  45. * Linked list of operations on the set.
  46. */
  47. struct GNUNET_SET_OperationHandle *ops_head;
  48. /**
  49. * Linked list of operations on the set.
  50. */
  51. struct GNUNET_SET_OperationHandle *ops_tail;
  52. /**
  53. * Callback for the current iteration over the set,
  54. * NULL if no iterator is active.
  55. */
  56. GNUNET_SET_ElementIterator iterator;
  57. /**
  58. * Closure for @e iterator
  59. */
  60. void *iterator_cls;
  61. /**
  62. * Should the set be destroyed once all operations are gone?
  63. * #GNUNET_SYSERR if #GNUNET_SET_destroy() must raise this flag,
  64. * #GNUNET_YES if #GNUNET_SET_destroy() did raise this flag.
  65. */
  66. int destroy_requested;
  67. /**
  68. * Has the set become invalid (e.g. service died)?
  69. */
  70. int invalid;
  71. /**
  72. * Both client and service count the number of iterators
  73. * created so far to match replies with iterators.
  74. */
  75. uint16_t iteration_id;
  76. /**
  77. * Configuration, needed when creating (lazy) copies.
  78. */
  79. const struct GNUNET_CONFIGURATION_Handle *cfg;
  80. /**
  81. * Doubly linked list of copy requests.
  82. */
  83. struct SetCopyRequest *copy_req_head;
  84. /**
  85. * Doubly linked list of copy requests.
  86. */
  87. struct SetCopyRequest *copy_req_tail;
  88. };
  89. /**
  90. * Handle for a set operation request from another peer.
  91. */
  92. struct GNUNET_SET_Request
  93. {
  94. /**
  95. * Id of the request, used to identify the request when
  96. * accepting/rejecting it.
  97. */
  98. uint32_t accept_id;
  99. /**
  100. * Has the request been accepted already?
  101. * #GNUNET_YES/#GNUNET_NO
  102. */
  103. int accepted;
  104. };
  105. /**
  106. * Handle to an operation. Only known to the service after committing
  107. * the handle with a set.
  108. */
  109. struct GNUNET_SET_OperationHandle
  110. {
  111. /**
  112. * Function to be called when we have a result,
  113. * or an error.
  114. */
  115. GNUNET_SET_ResultIterator result_cb;
  116. /**
  117. * Closure for @e result_cb.
  118. */
  119. void *result_cls;
  120. /**
  121. * Local set used for the operation,
  122. * NULL if no set has been provided by conclude yet.
  123. */
  124. struct GNUNET_SET_Handle *set;
  125. /**
  126. * Message sent to the server on calling conclude,
  127. * NULL if conclude has been called.
  128. */
  129. struct GNUNET_MQ_Envelope *conclude_mqm;
  130. /**
  131. * Address of the request if in the conclude message,
  132. * used to patch the request id into the message when the set is known.
  133. */
  134. uint32_t *request_id_addr;
  135. /**
  136. * Handles are kept in a linked list.
  137. */
  138. struct GNUNET_SET_OperationHandle *prev;
  139. /**
  140. * Handles are kept in a linked list.
  141. */
  142. struct GNUNET_SET_OperationHandle *next;
  143. /**
  144. * Request ID to identify the operation within the set.
  145. */
  146. uint32_t request_id;
  147. };
  148. /**
  149. * Opaque handle to a listen operation.
  150. */
  151. struct GNUNET_SET_ListenHandle
  152. {
  153. /**
  154. * Message queue for the client.
  155. */
  156. struct GNUNET_MQ_Handle* mq;
  157. /**
  158. * Configuration handle for the listener, stored
  159. * here to be able to reconnect transparently on
  160. * connection failure.
  161. */
  162. const struct GNUNET_CONFIGURATION_Handle *cfg;
  163. /**
  164. * Function to call on a new incoming request,
  165. * or on error.
  166. */
  167. GNUNET_SET_ListenCallback listen_cb;
  168. /**
  169. * Closure for @e listen_cb.
  170. */
  171. void *listen_cls;
  172. /**
  173. * Application ID we listen for.
  174. */
  175. struct GNUNET_HashCode app_id;
  176. /**
  177. * Time to wait until we try to reconnect on failure.
  178. */
  179. struct GNUNET_TIME_Relative reconnect_backoff;
  180. /**
  181. * Task for reconnecting when the listener fails.
  182. */
  183. struct GNUNET_SCHEDULER_Task *reconnect_task;
  184. /**
  185. * Operation we listen for.
  186. */
  187. enum GNUNET_SET_OperationType operation;
  188. };
  189. /* mutual recursion with handle_copy_lazy */
  190. static struct GNUNET_SET_Handle *
  191. create_internal (const struct GNUNET_CONFIGURATION_Handle *cfg,
  192. enum GNUNET_SET_OperationType op,
  193. const uint32_t *cookie);
  194. /**
  195. * Handle element for iteration over the set. Notifies the
  196. * iterator and sends an acknowledgement to the service.
  197. *
  198. * @param cls the `struct GNUNET_SET_Handle *`
  199. * @param msg the message
  200. */
  201. static void
  202. handle_copy_lazy (void *cls,
  203. const struct GNUNET_SET_CopyLazyResponseMessage *msg)
  204. {
  205. struct GNUNET_SET_Handle *set = cls;
  206. struct SetCopyRequest *req;
  207. struct GNUNET_SET_Handle *new_set;
  208. req = set->copy_req_head;
  209. if (NULL == req)
  210. {
  211. /* Service sent us unsolicited lazy copy response */
  212. GNUNET_break (0);
  213. return;
  214. }
  215. LOG (GNUNET_ERROR_TYPE_DEBUG,
  216. "Handling response to lazy copy\n");
  217. GNUNET_CONTAINER_DLL_remove (set->copy_req_head,
  218. set->copy_req_tail,
  219. req);
  220. // We pass none as operation here, since it doesn't matter when
  221. // cloning.
  222. new_set = create_internal (set->cfg,
  223. GNUNET_SET_OPERATION_NONE,
  224. &msg->cookie);
  225. req->cb (req->cls, new_set);
  226. GNUNET_free (req);
  227. }
  228. /**
  229. * Check that the given @a msg is well-formed.
  230. *
  231. * @param cls closure
  232. * @param msg message to check
  233. * @return #GNUNET_OK if message is well-formed
  234. */
  235. static int
  236. check_iter_element (void *cls,
  237. const struct GNUNET_SET_IterResponseMessage *msg)
  238. {
  239. /* minimum size was already checked, everything else is OK! */
  240. return GNUNET_OK;
  241. }
  242. /**
  243. * Handle element for iteration over the set. Notifies the
  244. * iterator and sends an acknowledgement to the service.
  245. *
  246. * @param cls the `struct GNUNET_SET_Handle *`
  247. * @param mh the message
  248. */
  249. static void
  250. handle_iter_element (void *cls,
  251. const struct GNUNET_SET_IterResponseMessage *msg)
  252. {
  253. struct GNUNET_SET_Handle *set = cls;
  254. GNUNET_SET_ElementIterator iter = set->iterator;
  255. struct GNUNET_SET_Element element;
  256. struct GNUNET_SET_IterAckMessage *ack_msg;
  257. struct GNUNET_MQ_Envelope *ev;
  258. uint16_t msize;
  259. LOG (GNUNET_ERROR_TYPE_DEBUG,
  260. "Received element in set iteration\n");
  261. msize = ntohs (msg->header.size);
  262. if (set->iteration_id != ntohs (msg->iteration_id))
  263. {
  264. /* element from a previous iteration, skip! */
  265. iter = NULL;
  266. }
  267. if (NULL != iter)
  268. {
  269. element.size = msize - sizeof (struct GNUNET_SET_IterResponseMessage);
  270. element.element_type = ntohs (msg->element_type);
  271. element.data = &msg[1];
  272. iter (set->iterator_cls,
  273. &element);
  274. }
  275. ev = GNUNET_MQ_msg (ack_msg,
  276. GNUNET_MESSAGE_TYPE_SET_ITER_ACK);
  277. ack_msg->send_more = htonl ((NULL != iter));
  278. GNUNET_MQ_send (set->mq, ev);
  279. }
  280. /**
  281. * Handle message signalling conclusion of iteration over the set.
  282. * Notifies the iterator that we are done.
  283. *
  284. * @param cls the set
  285. * @param mh the message
  286. */
  287. static void
  288. handle_iter_done (void *cls,
  289. const struct GNUNET_MessageHeader *mh)
  290. {
  291. struct GNUNET_SET_Handle *set = cls;
  292. GNUNET_SET_ElementIterator iter = set->iterator;
  293. if (NULL == iter)
  294. {
  295. /* FIXME: if this is true, could cancel+start a fresh one
  296. cause elements to go to the wrong iteration? */
  297. LOG (GNUNET_ERROR_TYPE_INFO,
  298. "Service completed set iteration that was already cancelled\n");
  299. return;
  300. }
  301. LOG (GNUNET_ERROR_TYPE_DEBUG,
  302. "Set iteration completed\n");
  303. set->destroy_requested = GNUNET_SYSERR;
  304. set->iterator = NULL;
  305. set->iteration_id++;
  306. iter (set->iterator_cls,
  307. NULL);
  308. if (GNUNET_SYSERR == set->destroy_requested)
  309. set->destroy_requested = GNUNET_NO;
  310. if (GNUNET_YES == set->destroy_requested)
  311. GNUNET_SET_destroy (set);
  312. }
  313. /**
  314. * Check that the given @a msg is well-formed.
  315. *
  316. * @param cls closure
  317. * @param msg message to check
  318. * @return #GNUNET_OK if message is well-formed
  319. */
  320. static int
  321. check_result (void *cls,
  322. const struct GNUNET_SET_ResultMessage *msg)
  323. {
  324. /* minimum size was already checked, everything else is OK! */
  325. return GNUNET_OK;
  326. }
  327. /**
  328. * Handle result message for a set operation.
  329. *
  330. * @param cls the set
  331. * @param mh the message
  332. */
  333. static void
  334. handle_result (void *cls,
  335. const struct GNUNET_SET_ResultMessage *msg)
  336. {
  337. struct GNUNET_SET_Handle *set = cls;
  338. struct GNUNET_SET_OperationHandle *oh;
  339. struct GNUNET_SET_Element e;
  340. enum GNUNET_SET_Status result_status;
  341. int destroy_set;
  342. GNUNET_assert (NULL != set->mq);
  343. result_status = (enum GNUNET_SET_Status) ntohs (msg->result_status);
  344. LOG (GNUNET_ERROR_TYPE_DEBUG,
  345. "Got result message with status %d\n",
  346. result_status);
  347. oh = GNUNET_MQ_assoc_get (set->mq,
  348. ntohl (msg->request_id));
  349. if (NULL == oh)
  350. {
  351. /* 'oh' can be NULL if we canceled the operation, but the service
  352. did not get the cancel message yet. */
  353. GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
  354. "Ignoring result from canceled operation\n");
  355. return;
  356. }
  357. switch (result_status)
  358. {
  359. case GNUNET_SET_STATUS_OK:
  360. case GNUNET_SET_STATUS_ADD_LOCAL:
  361. case GNUNET_SET_STATUS_ADD_REMOTE:
  362. goto do_element;
  363. case GNUNET_SET_STATUS_FAILURE:
  364. case GNUNET_SET_STATUS_DONE:
  365. goto do_final;
  366. case GNUNET_SET_STATUS_HALF_DONE:
  367. /* not used anymore */
  368. GNUNET_assert (0);
  369. }
  370. do_final:
  371. LOG (GNUNET_ERROR_TYPE_DEBUG,
  372. "Treating result as final status\n");
  373. GNUNET_MQ_assoc_remove (set->mq,
  374. ntohl (msg->request_id));
  375. GNUNET_CONTAINER_DLL_remove (set->ops_head,
  376. set->ops_tail,
  377. oh);
  378. /* Need to do this calculation _before_ the result callback,
  379. as IF the application still has a valid set handle, it
  380. may trigger destruction of the set during the callback. */
  381. destroy_set = (GNUNET_YES == set->destroy_requested) &&
  382. (NULL == set->ops_head);
  383. if (NULL != oh->result_cb)
  384. {
  385. oh->result_cb (oh->result_cls,
  386. NULL,
  387. GNUNET_ntohll (msg->current_size),
  388. result_status);
  389. }
  390. else
  391. {
  392. LOG (GNUNET_ERROR_TYPE_DEBUG,
  393. "No callback for final status\n");
  394. }
  395. if (destroy_set)
  396. GNUNET_SET_destroy (set);
  397. GNUNET_free (oh);
  398. return;
  399. do_element:
  400. LOG (GNUNET_ERROR_TYPE_DEBUG,
  401. "Treating result as element\n");
  402. e.data = &msg[1];
  403. e.size = ntohs (msg->header.size) - sizeof (struct GNUNET_SET_ResultMessage);
  404. e.element_type = ntohs (msg->element_type);
  405. if (NULL != oh->result_cb)
  406. oh->result_cb (oh->result_cls,
  407. &e,
  408. GNUNET_ntohll (msg->current_size),
  409. result_status);
  410. }
  411. /**
  412. * Destroy the given set operation.
  413. *
  414. * @param oh set operation to destroy
  415. */
  416. static void
  417. set_operation_destroy (struct GNUNET_SET_OperationHandle *oh)
  418. {
  419. struct GNUNET_SET_Handle *set = oh->set;
  420. struct GNUNET_SET_OperationHandle *h_assoc;
  421. if (NULL != oh->conclude_mqm)
  422. GNUNET_MQ_discard (oh->conclude_mqm);
  423. /* is the operation already commited? */
  424. if (NULL != set)
  425. {
  426. GNUNET_CONTAINER_DLL_remove (set->ops_head,
  427. set->ops_tail,
  428. oh);
  429. h_assoc = GNUNET_MQ_assoc_remove (set->mq,
  430. oh->request_id);
  431. GNUNET_assert ( (NULL == h_assoc) ||
  432. (h_assoc == oh) );
  433. }
  434. GNUNET_free (oh);
  435. }
  436. /**
  437. * Cancel the given set operation. We need to send an explicit cancel
  438. * message, as all operations one one set communicate using one
  439. * handle.
  440. *
  441. * @param oh set operation to cancel
  442. */
  443. void
  444. GNUNET_SET_operation_cancel (struct GNUNET_SET_OperationHandle *oh)
  445. {
  446. struct GNUNET_SET_Handle *set = oh->set;
  447. struct GNUNET_SET_CancelMessage *m;
  448. struct GNUNET_MQ_Envelope *mqm;
  449. LOG (GNUNET_ERROR_TYPE_DEBUG,
  450. "Cancelling SET operation\n");
  451. if (NULL != set)
  452. {
  453. mqm = GNUNET_MQ_msg (m, GNUNET_MESSAGE_TYPE_SET_CANCEL);
  454. m->request_id = htonl (oh->request_id);
  455. GNUNET_MQ_send (set->mq, mqm);
  456. }
  457. set_operation_destroy (oh);
  458. if ( (NULL != set) &&
  459. (GNUNET_YES == set->destroy_requested) &&
  460. (NULL == set->ops_head) )
  461. {
  462. LOG (GNUNET_ERROR_TYPE_DEBUG,
  463. "Destroying set after operation cancel\n");
  464. GNUNET_SET_destroy (set);
  465. }
  466. }
  467. /**
  468. * We encountered an error communicating with the set service while
  469. * performing a set operation. Report to the application.
  470. *
  471. * @param cls the `struct GNUNET_SET_Handle`
  472. * @param error error code
  473. */
  474. static void
  475. handle_client_set_error (void *cls,
  476. enum GNUNET_MQ_Error error)
  477. {
  478. struct GNUNET_SET_Handle *set = cls;
  479. GNUNET_SET_ElementIterator iter = set->iterator;
  480. LOG (GNUNET_ERROR_TYPE_ERROR,
  481. "Handling client set error %d\n",
  482. error);
  483. while (NULL != set->ops_head)
  484. {
  485. if ( (NULL != set->ops_head->result_cb) &&
  486. (GNUNET_NO == set->destroy_requested) )
  487. set->ops_head->result_cb (set->ops_head->result_cls,
  488. NULL,
  489. 0,
  490. GNUNET_SET_STATUS_FAILURE);
  491. set_operation_destroy (set->ops_head);
  492. }
  493. set->iterator = NULL;
  494. set->iteration_id++;
  495. set->invalid = GNUNET_YES;
  496. if (NULL != iter)
  497. iter (set->iterator_cls,
  498. NULL);
  499. }
  500. /**
  501. * FIXME.
  502. */
  503. static struct GNUNET_SET_Handle *
  504. create_internal (const struct GNUNET_CONFIGURATION_Handle *cfg,
  505. enum GNUNET_SET_OperationType op,
  506. const uint32_t *cookie)
  507. {
  508. struct GNUNET_SET_Handle *set = GNUNET_new (struct GNUNET_SET_Handle);
  509. struct GNUNET_MQ_MessageHandler mq_handlers[] = {
  510. GNUNET_MQ_hd_var_size (result,
  511. GNUNET_MESSAGE_TYPE_SET_RESULT,
  512. struct GNUNET_SET_ResultMessage,
  513. set),
  514. GNUNET_MQ_hd_var_size (iter_element,
  515. GNUNET_MESSAGE_TYPE_SET_ITER_ELEMENT,
  516. struct GNUNET_SET_IterResponseMessage,
  517. set),
  518. GNUNET_MQ_hd_fixed_size (iter_done,
  519. GNUNET_MESSAGE_TYPE_SET_ITER_DONE,
  520. struct GNUNET_MessageHeader,
  521. set),
  522. GNUNET_MQ_hd_fixed_size (copy_lazy,
  523. GNUNET_MESSAGE_TYPE_SET_COPY_LAZY_RESPONSE,
  524. struct GNUNET_SET_CopyLazyResponseMessage,
  525. set),
  526. GNUNET_MQ_handler_end ()
  527. };
  528. struct GNUNET_MQ_Envelope *mqm;
  529. struct GNUNET_SET_CreateMessage *create_msg;
  530. struct GNUNET_SET_CopyLazyConnectMessage *copy_msg;
  531. set->cfg = cfg;
  532. set->mq = GNUNET_CLIENT_connect (cfg,
  533. "set",
  534. mq_handlers,
  535. &handle_client_set_error,
  536. set);
  537. if (NULL == set->mq)
  538. {
  539. GNUNET_free (set);
  540. return NULL;
  541. }
  542. if (NULL == cookie)
  543. {
  544. LOG (GNUNET_ERROR_TYPE_DEBUG,
  545. "Creating new set (operation %u)\n",
  546. op);
  547. mqm = GNUNET_MQ_msg (create_msg,
  548. GNUNET_MESSAGE_TYPE_SET_CREATE);
  549. create_msg->operation = htonl (op);
  550. }
  551. else
  552. {
  553. LOG (GNUNET_ERROR_TYPE_DEBUG,
  554. "Creating new set (lazy copy)\n",
  555. op);
  556. mqm = GNUNET_MQ_msg (copy_msg,
  557. GNUNET_MESSAGE_TYPE_SET_COPY_LAZY_CONNECT);
  558. copy_msg->cookie = *cookie;
  559. }
  560. GNUNET_MQ_send (set->mq,
  561. mqm);
  562. return set;
  563. }
  564. /**
  565. * Create an empty set, supporting the specified operation.
  566. *
  567. * @param cfg configuration to use for connecting to the
  568. * set service
  569. * @param op operation supported by the set
  570. * Note that the operation has to be specified
  571. * beforehand, as certain set operations need to maintain
  572. * data structures spefific to the operation
  573. * @return a handle to the set
  574. */
  575. struct GNUNET_SET_Handle *
  576. GNUNET_SET_create (const struct GNUNET_CONFIGURATION_Handle *cfg,
  577. enum GNUNET_SET_OperationType op)
  578. {
  579. struct GNUNET_SET_Handle *set;
  580. set = create_internal (cfg,
  581. op,
  582. NULL);
  583. LOG (GNUNET_ERROR_TYPE_DEBUG,
  584. "Creating set %p for operation %d\n",
  585. set,
  586. op);
  587. return set;
  588. }
  589. /**
  590. * Add an element to the given set. After the element has been added
  591. * (in the sense of being transmitted to the set service), @a cont
  592. * will be called. Multiple calls to GNUNET_SET_add_element() can be
  593. * queued.
  594. *
  595. * @param set set to add element to
  596. * @param element element to add to the set
  597. * @param cont continuation called after the element has been added
  598. * @param cont_cls closure for @a cont
  599. * @return #GNUNET_OK on success, #GNUNET_SYSERR if the
  600. * set is invalid (e.g. the set service crashed)
  601. */
  602. int
  603. GNUNET_SET_add_element (struct GNUNET_SET_Handle *set,
  604. const struct GNUNET_SET_Element *element,
  605. GNUNET_SET_Continuation cont,
  606. void *cont_cls)
  607. {
  608. struct GNUNET_MQ_Envelope *mqm;
  609. struct GNUNET_SET_ElementMessage *msg;
  610. LOG (GNUNET_ERROR_TYPE_DEBUG,
  611. "adding element of type %u to set %p\n",
  612. (unsigned int) element->element_type,
  613. set);
  614. GNUNET_assert (NULL != set);
  615. if (GNUNET_YES == set->invalid)
  616. {
  617. if (NULL != cont)
  618. cont (cont_cls);
  619. return GNUNET_SYSERR;
  620. }
  621. mqm = GNUNET_MQ_msg_extra (msg,
  622. element->size,
  623. GNUNET_MESSAGE_TYPE_SET_ADD);
  624. msg->element_type = htons (element->element_type);
  625. GNUNET_memcpy (&msg[1],
  626. element->data,
  627. element->size);
  628. GNUNET_MQ_notify_sent (mqm,
  629. cont, cont_cls);
  630. GNUNET_MQ_send (set->mq, mqm);
  631. return GNUNET_OK;
  632. }
  633. /**
  634. * Remove an element to the given set. After the element has been
  635. * removed (in the sense of the request being transmitted to the set
  636. * service), @a cont will be called. Multiple calls to
  637. * GNUNET_SET_remove_element() can be queued
  638. *
  639. * @param set set to remove element from
  640. * @param element element to remove from the set
  641. * @param cont continuation called after the element has been removed
  642. * @param cont_cls closure for @a cont
  643. * @return #GNUNET_OK on success, #GNUNET_SYSERR if the
  644. * set is invalid (e.g. the set service crashed)
  645. */
  646. int
  647. GNUNET_SET_remove_element (struct GNUNET_SET_Handle *set,
  648. const struct GNUNET_SET_Element *element,
  649. GNUNET_SET_Continuation cont,
  650. void *cont_cls)
  651. {
  652. struct GNUNET_MQ_Envelope *mqm;
  653. struct GNUNET_SET_ElementMessage *msg;
  654. LOG (GNUNET_ERROR_TYPE_DEBUG,
  655. "Removing element from set %p\n",
  656. set);
  657. if (GNUNET_YES == set->invalid)
  658. {
  659. if (NULL != cont)
  660. cont (cont_cls);
  661. return GNUNET_SYSERR;
  662. }
  663. mqm = GNUNET_MQ_msg_extra (msg,
  664. element->size,
  665. GNUNET_MESSAGE_TYPE_SET_REMOVE);
  666. msg->element_type = htons (element->element_type);
  667. GNUNET_memcpy (&msg[1],
  668. element->data,
  669. element->size);
  670. GNUNET_MQ_notify_sent (mqm,
  671. cont, cont_cls);
  672. GNUNET_MQ_send (set->mq, mqm);
  673. return GNUNET_OK;
  674. }
  675. /**
  676. * Destroy the set handle if no operations are left, mark the set
  677. * for destruction otherwise.
  678. *
  679. * @param set set handle to destroy
  680. */
  681. void
  682. GNUNET_SET_destroy (struct GNUNET_SET_Handle *set)
  683. {
  684. /* destroying set while iterator is active is currently
  685. not supported; we should expand the API to allow
  686. clients to explicitly cancel the iteration! */
  687. GNUNET_assert (NULL != set);
  688. if ( (NULL != set->ops_head) ||
  689. (NULL != set->iterator) ||
  690. (GNUNET_SYSERR == set->destroy_requested) )
  691. {
  692. LOG (GNUNET_ERROR_TYPE_DEBUG,
  693. "Set operations are pending, delaying set destruction\n");
  694. set->destroy_requested = GNUNET_YES;
  695. return;
  696. }
  697. LOG (GNUNET_ERROR_TYPE_DEBUG,
  698. "Really destroying set\n");
  699. if (NULL != set->mq)
  700. {
  701. GNUNET_MQ_destroy (set->mq);
  702. set->mq = NULL;
  703. }
  704. GNUNET_free (set);
  705. }
  706. /**
  707. * Prepare a set operation to be evaluated with another peer.
  708. * The evaluation will not start until the client provides
  709. * a local set with #GNUNET_SET_commit().
  710. *
  711. * @param other_peer peer with the other set
  712. * @param app_id hash for the application using the set
  713. * @param context_msg additional information for the request
  714. * @param result_mode specified how results will be returned,
  715. * see `enum GNUNET_SET_ResultMode`.
  716. * @param result_cb called on error or success
  717. * @param result_cls closure for @e result_cb
  718. * @return a handle to cancel the operation
  719. */
  720. struct GNUNET_SET_OperationHandle *
  721. GNUNET_SET_prepare (const struct GNUNET_PeerIdentity *other_peer,
  722. const struct GNUNET_HashCode *app_id,
  723. const struct GNUNET_MessageHeader *context_msg,
  724. enum GNUNET_SET_ResultMode result_mode,
  725. struct GNUNET_SET_Option options[],
  726. GNUNET_SET_ResultIterator result_cb,
  727. void *result_cls)
  728. {
  729. struct GNUNET_MQ_Envelope *mqm;
  730. struct GNUNET_SET_OperationHandle *oh;
  731. struct GNUNET_SET_EvaluateMessage *msg;
  732. struct GNUNET_SET_Option *opt;
  733. LOG (GNUNET_ERROR_TYPE_DEBUG,
  734. "Client prepares set operation (%d)\n",
  735. result_mode);
  736. oh = GNUNET_new (struct GNUNET_SET_OperationHandle);
  737. oh->result_cb = result_cb;
  738. oh->result_cls = result_cls;
  739. mqm = GNUNET_MQ_msg_nested_mh (msg,
  740. GNUNET_MESSAGE_TYPE_SET_EVALUATE,
  741. context_msg);
  742. msg->app_id = *app_id;
  743. msg->result_mode = htonl (result_mode);
  744. msg->target_peer = *other_peer;
  745. for (opt = options; opt->type != 0; opt++)
  746. {
  747. switch (opt->type)
  748. {
  749. case GNUNET_SET_OPTION_BYZANTINE:
  750. msg->byzantine = GNUNET_YES;
  751. msg->byzantine_lower_bound = opt->v.num;
  752. break;
  753. case GNUNET_SET_OPTION_FORCE_FULL:
  754. msg->force_full = GNUNET_YES;
  755. break;
  756. case GNUNET_SET_OPTION_FORCE_DELTA:
  757. msg->force_delta = GNUNET_YES;
  758. break;
  759. default:
  760. LOG (GNUNET_ERROR_TYPE_ERROR,
  761. "Option with type %d not recognized\n", (int) opt->type);
  762. }
  763. }
  764. oh->conclude_mqm = mqm;
  765. oh->request_id_addr = &msg->request_id;
  766. return oh;
  767. }
  768. /**
  769. * Connect to the set service in order to listen for requests.
  770. *
  771. * @param cls the `struct GNUNET_SET_ListenHandle *` to connect
  772. */
  773. static void
  774. listen_connect (void *cls);
  775. /**
  776. * Check validity of request message for a listen operation
  777. *
  778. * @param cls the listen handle
  779. * @param msg the message
  780. * @return #GNUNET_OK if the message is well-formed
  781. */
  782. static int
  783. check_request (void *cls,
  784. const struct GNUNET_SET_RequestMessage *msg)
  785. {
  786. const struct GNUNET_MessageHeader *context_msg;
  787. if (ntohs (msg->header.size) == sizeof (*msg))
  788. return GNUNET_OK; /* no context message is OK */
  789. context_msg = GNUNET_MQ_extract_nested_mh (msg);
  790. if (NULL == context_msg)
  791. {
  792. /* malformed context message is NOT ok */
  793. GNUNET_break_op (0);
  794. return GNUNET_SYSERR;
  795. }
  796. return GNUNET_OK;
  797. }
  798. /**
  799. * Handle request message for a listen operation
  800. *
  801. * @param cls the listen handle
  802. * @param msg the message
  803. */
  804. static void
  805. handle_request (void *cls,
  806. const struct GNUNET_SET_RequestMessage *msg)
  807. {
  808. struct GNUNET_SET_ListenHandle *lh = cls;
  809. struct GNUNET_SET_Request req;
  810. const struct GNUNET_MessageHeader *context_msg;
  811. struct GNUNET_MQ_Envelope *mqm;
  812. struct GNUNET_SET_RejectMessage *rmsg;
  813. LOG (GNUNET_ERROR_TYPE_DEBUG,
  814. "Processing incoming operation request with id %u\n",
  815. ntohl (msg->accept_id));
  816. /* we got another valid request => reset the backoff */
  817. lh->reconnect_backoff = GNUNET_TIME_UNIT_MILLISECONDS;
  818. req.accept_id = ntohl (msg->accept_id);
  819. req.accepted = GNUNET_NO;
  820. context_msg = GNUNET_MQ_extract_nested_mh (msg);
  821. /* calling #GNUNET_SET_accept() in the listen cb will set req->accepted */
  822. lh->listen_cb (lh->listen_cls,
  823. &msg->peer_id,
  824. context_msg,
  825. &req);
  826. if (GNUNET_YES == req.accepted)
  827. return; /* the accept-case is handled in #GNUNET_SET_accept() */
  828. LOG (GNUNET_ERROR_TYPE_DEBUG,
  829. "Rejected request %u\n",
  830. ntohl (msg->accept_id));
  831. mqm = GNUNET_MQ_msg (rmsg,
  832. GNUNET_MESSAGE_TYPE_SET_REJECT);
  833. rmsg->accept_reject_id = msg->accept_id;
  834. GNUNET_MQ_send (lh->mq, mqm);
  835. }
  836. /**
  837. * Our connection with the set service encountered an error,
  838. * re-initialize with exponential back-off.
  839. *
  840. * @param cls the `struct GNUNET_SET_ListenHandle *`
  841. * @param error reason for the disconnect
  842. */
  843. static void
  844. handle_client_listener_error (void *cls,
  845. enum GNUNET_MQ_Error error)
  846. {
  847. struct GNUNET_SET_ListenHandle *lh = cls;
  848. LOG (GNUNET_ERROR_TYPE_DEBUG,
  849. "Listener broke down (%d), re-connecting\n",
  850. (int) error);
  851. GNUNET_MQ_destroy (lh->mq);
  852. lh->mq = NULL;
  853. lh->reconnect_task = GNUNET_SCHEDULER_add_delayed (lh->reconnect_backoff,
  854. &listen_connect,
  855. lh);
  856. lh->reconnect_backoff = GNUNET_TIME_STD_BACKOFF (lh->reconnect_backoff);
  857. }
  858. /**
  859. * Connect to the set service in order to listen for requests.
  860. *
  861. * @param cls the `struct GNUNET_SET_ListenHandle *` to connect
  862. */
  863. static void
  864. listen_connect (void *cls)
  865. {
  866. struct GNUNET_SET_ListenHandle *lh = cls;
  867. struct GNUNET_MQ_MessageHandler mq_handlers[] = {
  868. GNUNET_MQ_hd_var_size (request,
  869. GNUNET_MESSAGE_TYPE_SET_REQUEST,
  870. struct GNUNET_SET_RequestMessage,
  871. lh),
  872. GNUNET_MQ_handler_end ()
  873. };
  874. struct GNUNET_MQ_Envelope *mqm;
  875. struct GNUNET_SET_ListenMessage *msg;
  876. lh->reconnect_task = NULL;
  877. GNUNET_assert (NULL == lh->mq);
  878. lh->mq = GNUNET_CLIENT_connect (lh->cfg,
  879. "set",
  880. mq_handlers,
  881. &handle_client_listener_error,
  882. lh);
  883. if (NULL == lh->mq)
  884. return;
  885. mqm = GNUNET_MQ_msg (msg, GNUNET_MESSAGE_TYPE_SET_LISTEN);
  886. msg->operation = htonl (lh->operation);
  887. msg->app_id = lh->app_id;
  888. GNUNET_MQ_send (lh->mq,
  889. mqm);
  890. }
  891. /**
  892. * Wait for set operation requests for the given application id
  893. *
  894. * @param cfg configuration to use for connecting to
  895. * the set service, needs to be valid for the lifetime of the listen handle
  896. * @param operation operation we want to listen for
  897. * @param app_id id of the application that handles set operation requests
  898. * @param listen_cb called for each incoming request matching the operation
  899. * and application id
  900. * @param listen_cls handle for @a listen_cb
  901. * @return a handle that can be used to cancel the listen operation
  902. */
  903. struct GNUNET_SET_ListenHandle *
  904. GNUNET_SET_listen (const struct GNUNET_CONFIGURATION_Handle *cfg,
  905. enum GNUNET_SET_OperationType operation,
  906. const struct GNUNET_HashCode *app_id,
  907. GNUNET_SET_ListenCallback listen_cb,
  908. void *listen_cls)
  909. {
  910. struct GNUNET_SET_ListenHandle *lh;
  911. LOG (GNUNET_ERROR_TYPE_DEBUG,
  912. "Starting listener for app %s\n",
  913. GNUNET_h2s (app_id));
  914. lh = GNUNET_new (struct GNUNET_SET_ListenHandle);
  915. lh->listen_cb = listen_cb;
  916. lh->listen_cls = listen_cls;
  917. lh->cfg = cfg;
  918. lh->operation = operation;
  919. lh->app_id = *app_id;
  920. lh->reconnect_backoff = GNUNET_TIME_UNIT_MILLISECONDS;
  921. listen_connect (lh);
  922. if (NULL == lh->mq)
  923. {
  924. GNUNET_free (lh);
  925. return NULL;
  926. }
  927. return lh;
  928. }
  929. /**
  930. * Cancel the given listen operation.
  931. *
  932. * @param lh handle for the listen operation
  933. */
  934. void
  935. GNUNET_SET_listen_cancel (struct GNUNET_SET_ListenHandle *lh)
  936. {
  937. LOG (GNUNET_ERROR_TYPE_DEBUG,
  938. "Canceling listener %s\n",
  939. GNUNET_h2s (&lh->app_id));
  940. if (NULL != lh->mq)
  941. {
  942. GNUNET_MQ_destroy (lh->mq);
  943. lh->mq = NULL;
  944. }
  945. if (NULL != lh->reconnect_task)
  946. {
  947. GNUNET_SCHEDULER_cancel (lh->reconnect_task);
  948. lh->reconnect_task = NULL;
  949. }
  950. GNUNET_free (lh);
  951. }
  952. /**
  953. * Accept a request we got via #GNUNET_SET_listen. Must be called during
  954. * #GNUNET_SET_listen, as the 'struct GNUNET_SET_Request' becomes invalid
  955. * afterwards.
  956. * Call #GNUNET_SET_commit to provide the local set to use for the operation,
  957. * and to begin the exchange with the remote peer.
  958. *
  959. * @param request request to accept
  960. * @param result_mode specified how results will be returned,
  961. * see `enum GNUNET_SET_ResultMode`.
  962. * @param result_cb callback for the results
  963. * @param result_cls closure for @a result_cb
  964. * @return a handle to cancel the operation
  965. */
  966. struct GNUNET_SET_OperationHandle *
  967. GNUNET_SET_accept (struct GNUNET_SET_Request *request,
  968. enum GNUNET_SET_ResultMode result_mode,
  969. struct GNUNET_SET_Option options[],
  970. GNUNET_SET_ResultIterator result_cb,
  971. void *result_cls)
  972. {
  973. struct GNUNET_MQ_Envelope *mqm;
  974. struct GNUNET_SET_OperationHandle *oh;
  975. struct GNUNET_SET_AcceptMessage *msg;
  976. GNUNET_assert (GNUNET_NO == request->accepted);
  977. LOG (GNUNET_ERROR_TYPE_DEBUG,
  978. "Client accepts set operation (%d) with id %u\n",
  979. result_mode,
  980. request->accept_id);
  981. request->accepted = GNUNET_YES;
  982. mqm = GNUNET_MQ_msg (msg,
  983. GNUNET_MESSAGE_TYPE_SET_ACCEPT);
  984. msg->accept_reject_id = htonl (request->accept_id);
  985. msg->result_mode = htonl (result_mode);
  986. oh = GNUNET_new (struct GNUNET_SET_OperationHandle);
  987. oh->result_cb = result_cb;
  988. oh->result_cls = result_cls;
  989. oh->conclude_mqm = mqm;
  990. oh->request_id_addr = &msg->request_id;
  991. return oh;
  992. }
  993. /**
  994. * Commit a set to be used with a set operation.
  995. * This function is called once we have fully constructed
  996. * the set that we want to use for the operation. At this
  997. * time, the P2P protocol can then begin to exchange the
  998. * set information and call the result callback with the
  999. * result information.
  1000. *
  1001. * @param oh handle to the set operation
  1002. * @param set the set to use for the operation
  1003. * @return #GNUNET_OK on success, #GNUNET_SYSERR if the
  1004. * set is invalid (e.g. the set service crashed)
  1005. */
  1006. int
  1007. GNUNET_SET_commit (struct GNUNET_SET_OperationHandle *oh,
  1008. struct GNUNET_SET_Handle *set)
  1009. {
  1010. if (NULL != oh->set)
  1011. {
  1012. /* Some other set was already committed for this
  1013. * operation, there is a logic bug in the client of this API */
  1014. GNUNET_break (0);
  1015. return GNUNET_OK;
  1016. }
  1017. GNUNET_assert (NULL != set);
  1018. if (GNUNET_YES == set->invalid)
  1019. return GNUNET_SYSERR;
  1020. LOG (GNUNET_ERROR_TYPE_DEBUG,
  1021. "Client commits to SET\n");
  1022. GNUNET_assert (NULL != oh->conclude_mqm);
  1023. oh->set = set;
  1024. GNUNET_CONTAINER_DLL_insert (set->ops_head,
  1025. set->ops_tail,
  1026. oh);
  1027. oh->request_id = GNUNET_MQ_assoc_add (set->mq,
  1028. oh);
  1029. *oh->request_id_addr = htonl (oh->request_id);
  1030. GNUNET_MQ_send (set->mq,
  1031. oh->conclude_mqm);
  1032. oh->conclude_mqm = NULL;
  1033. oh->request_id_addr = NULL;
  1034. return GNUNET_OK;
  1035. }
  1036. /**
  1037. * Iterate over all elements in the given set. Note that this
  1038. * operation involves transferring every element of the set from the
  1039. * service to the client, and is thus costly.
  1040. *
  1041. * @param set the set to iterate over
  1042. * @param iter the iterator to call for each element
  1043. * @param iter_cls closure for @a iter
  1044. * @return #GNUNET_YES if the iteration started successfuly,
  1045. * #GNUNET_NO if another iteration is active
  1046. * #GNUNET_SYSERR if the set is invalid (e.g. the server crashed, disconnected)
  1047. */
  1048. int
  1049. GNUNET_SET_iterate (struct GNUNET_SET_Handle *set,
  1050. GNUNET_SET_ElementIterator iter,
  1051. void *iter_cls)
  1052. {
  1053. struct GNUNET_MQ_Envelope *ev;
  1054. GNUNET_assert (NULL != iter);
  1055. if (GNUNET_YES == set->invalid)
  1056. return GNUNET_SYSERR;
  1057. if (NULL != set->iterator)
  1058. return GNUNET_NO;
  1059. LOG (GNUNET_ERROR_TYPE_DEBUG,
  1060. "Iterating over set\n");
  1061. set->iterator = iter;
  1062. set->iterator_cls = iter_cls;
  1063. ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_ITER_REQUEST);
  1064. GNUNET_MQ_send (set->mq, ev);
  1065. return GNUNET_YES;
  1066. }
  1067. void
  1068. GNUNET_SET_copy_lazy (struct GNUNET_SET_Handle *set,
  1069. GNUNET_SET_CopyReadyCallback cb,
  1070. void *cls)
  1071. {
  1072. struct GNUNET_MQ_Envelope *ev;
  1073. struct SetCopyRequest *req;
  1074. LOG (GNUNET_ERROR_TYPE_DEBUG,
  1075. "Creating lazy copy of set\n");
  1076. ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_COPY_LAZY_PREPARE);
  1077. GNUNET_MQ_send (set->mq, ev);
  1078. req = GNUNET_new (struct SetCopyRequest);
  1079. req->cb = cb;
  1080. req->cls = cls;
  1081. GNUNET_CONTAINER_DLL_insert (set->copy_req_head,
  1082. set->copy_req_tail,
  1083. req);
  1084. }
  1085. /**
  1086. * Create a copy of an element. The copy
  1087. * must be GNUNET_free-d by the caller.
  1088. *
  1089. * @param element the element to copy
  1090. * @return the copied element
  1091. */
  1092. struct GNUNET_SET_Element *
  1093. GNUNET_SET_element_dup (const struct GNUNET_SET_Element *element)
  1094. {
  1095. struct GNUNET_SET_Element *copy;
  1096. copy = GNUNET_malloc (element->size + sizeof (struct GNUNET_SET_Element));
  1097. copy->size = element->size;
  1098. copy->element_type = element->element_type;
  1099. copy->data = &copy[1];
  1100. GNUNET_memcpy (&copy[1],
  1101. element->data,
  1102. copy->size);
  1103. return copy;
  1104. }
  1105. /**
  1106. * Hash a set element.
  1107. *
  1108. * @param element the element that should be hashed
  1109. * @param[out] ret_hash a pointer to where the hash of @a element
  1110. * should be stored
  1111. */
  1112. void
  1113. GNUNET_SET_element_hash (const struct GNUNET_SET_Element *element,
  1114. struct GNUNET_HashCode *ret_hash)
  1115. {
  1116. struct GNUNET_HashContext *ctx = GNUNET_CRYPTO_hash_context_start ();
  1117. /* It's not guaranteed that the element data is always after the element header,
  1118. so we need to hash the chunks separately. */
  1119. GNUNET_CRYPTO_hash_context_read (ctx, &element->size, sizeof (uint16_t));
  1120. GNUNET_CRYPTO_hash_context_read (ctx, &element->element_type, sizeof (uint16_t));
  1121. GNUNET_CRYPTO_hash_context_read (ctx, element->data, element->size);
  1122. GNUNET_CRYPTO_hash_context_finish (ctx, ret_hash);
  1123. }
  1124. /* end of set_api.c */