src/set/set_api.c

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