src/set/gnunet-service-set.c

   1 /*
   2       This file is part of GNUnet
   3       (C) 2013 Christian Grothoff (and other contributing authors)
   4
   5       GNUnet is free software; you can redistribute it and/or modify
   6       it under the terms of the GNU General Public License as published
   7       by the Free Software Foundation; either version 3, or (at your
   8       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       General Public License for more details.
  14
  15       You should have received a copy of the GNU General Public License
  16       along with GNUnet; see the file COPYING.  If not, write to the
  17       Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18       Boston, MA 02111-1307, USA.
  19 */
  20
  21 /**
  22  * @file set/gnunet-service-set.c
  23  * @brief two-peer set operations
  24  * @author Florian Dold
  25  */
  26 #include "gnunet-service-set.h"
  27 #include "set_protocol.h"
  28
  29
  30 /**
  31  * Peer that has connected to us, but is not yet evaluating a set operation.
  32  * Once the peer has sent a request, and the client has
  33  * accepted or rejected it, this information will be deleted.
  34  */
  35 struct Incoming
  36 {
  37   /**
  38    * Incoming peers are held in a linked list
  39    */
  40   struct Incoming *next;
  41
  42   /**
  43    * Incoming peers are held in a linked list
  44    */
  45   struct Incoming *prev;
  46
  47   /**
  48    * Detail information about the operation.
  49    * NULL as long as we did not receive the operation
  50    * request from the remote peer.
  51    */
  52   struct OperationSpecification *spec;
  53
  54   /**
  55    * The identity of the requesting peer.  Needs to
  56    * be stored here as the op spec might not have been created yet.
  57    */
  58   struct GNUNET_PeerIdentity peer;
  59
  60   /**
  61    * Tunnel to the peer.
  62    */
  63   struct GNUNET_MESH_Tunnel *tunnel;
  64
  65   /**
  66    * Unique request id for the request from
  67    * a remote peer, sent to the client, which will
  68    * accept or reject the request.
  69    * Set to '0' iff the request has not been
  70    * suggested yet.
  71    */
  72   uint32_t suggest_id;
  73
  74   /**
  75    * Timeout task, if the incoming peer has not been accepted
  76    * after the timeout, it will be disconnected.
  77    */
  78   GNUNET_SCHEDULER_TaskIdentifier timeout_task;
  79
  80   /**
  81    * Tunnel context, needs to be stored here as a client's accept will change
  82    * the tunnel context.
  83    */
  84   struct TunnelContext *tc;
  85 };
  86
  87
  88 /**
  89  * A listener is inhabited by a client, and
  90  * waits for evaluation requests from remote peers.
  91  */
  92 struct Listener
  93 {
  94   /**
  95    * Listeners are held in a doubly linked list.
  96    */
  97   struct Listener *next;
  98
  99   /**
 100    * Listeners are held in a doubly linked list.
 101    */
 102   struct Listener *prev;
 103
 104   /**
 105    * Client that owns the listener.
 106    * Only one client may own a listener.
 107    */
 108   struct GNUNET_SERVER_Client *client;
 109
 110   /**
 111    * Message queue for the client
 112    */
 113   struct GNUNET_MQ_Handle *client_mq;
 114
 115   /**
 116    * The type of the operation.
 117    */
 118   enum GNUNET_SET_OperationType operation;
 119
 120   /**
 121    * Application ID for the operation, used to distinguish
 122    * multiple operations of the same type with the same peer.
 123    */
 124   struct GNUNET_HashCode app_id;
 125 };
 126
 127
 128 /**
 129  * Configuration of our local peer.
 130  */
 131 static const struct GNUNET_CONFIGURATION_Handle *configuration;
 132
 133 /**
 134  * Handle to the mesh service, used
 135  * to listen for and connect to remote peers.
 136  */
 137 static struct GNUNET_MESH_Handle *mesh;
 138
 139 /**
 140  * Sets are held in a doubly linked list.
 141  */
 142 static struct Set *sets_head;
 143
 144 /**
 145  * Sets are held in a doubly linked list.
 146  */
 147 static struct Set *sets_tail;
 148
 149 /**
 150  * Listeners are held in a doubly linked list.
 151  */
 152 static struct Listener *listeners_head;
 153
 154 /**
 155  * Listeners are held in a doubly linked list.
 156  */
 157 static struct Listener *listeners_tail;
 158
 159 /**
 160  * Incoming sockets from remote peers are
 161  * held in a doubly linked list.
 162  */
 163 static struct Incoming *incoming_head;
 164
 165 /**
 166  * Incoming sockets from remote peers are
 167  * held in a doubly linked list.
 168  */
 169 static struct Incoming *incoming_tail;
 170
 171 /**
 172  * Counter for allocating unique IDs for clients,
 173  * used to identify incoming operation requests from remote peers,
 174  * that the client can choose to accept or refuse.
 175  */
 176 static uint32_t suggest_id = 1;
 177
 178
 179 /**
 180  * Get set that is owned by the given client, if any.
 181  *
 182  * @param client client to look for
 183  * @return set that the client owns, NULL if the client
 184  *         does not own a set
 185  */
 186 static struct Set *
 187 set_get (struct GNUNET_SERVER_Client *client)
 188 {
 189   struct Set *set;
 190
 191   for (set = sets_head; NULL != set; set = set->next)
 192     if (set->client == client)
 193       return set;
 194   return NULL;
 195 }
 196
 197
 198 /**
 199  * Get the listener associated with the given client, if any.
 200  *
 201  * @param client the client
 202  * @return listener associated with the client, NULL
 203  *         if there isn't any
 204  */
 205 static struct Listener *
 206 listener_get (struct GNUNET_SERVER_Client *client)
 207 {
 208   struct Listener *listener;
 209
 210   for (listener = listeners_head; NULL != listener; listener = listener->next)
 211     if (listener->client == client)
 212       return listener;
 213   return NULL;
 214 }
 215
 216
 217 /**
 218  * Get the incoming socket associated with the given id.
 219  *
 220  * @param id id to look for
 221  * @return the incoming socket associated with the id,
 222  *         or NULL if there is none
 223  */
 224 static struct Incoming *
 225 get_incoming (uint32_t id)
 226 {
 227   struct Incoming *incoming;
 228
 229   for (incoming = incoming_head; NULL != incoming; incoming = incoming->next)
 230     if (incoming->suggest_id == id)
 231       return incoming;
 232   return NULL;
 233 }
 234
 235
 236 /**
 237  * Destroy a listener, free all resources associated with it.
 238  *
 239  * @param listener listener to destroy
 240  */
 241 static void
 242 listener_destroy (struct Listener *listener)
 243 {
 244   /* If the client is not dead yet, destroy it.
 245    * The client's destroy callback will destroy the listener again. */
 246   if (NULL != listener->client)
 247   {
 248     struct GNUNET_SERVER_Client *client = listener->client;
 249     listener->client = NULL;
 250     GNUNET_SERVER_client_disconnect (client);
 251     return;
 252   }
 253   if (NULL != listener->client_mq)
 254   {
 255     GNUNET_MQ_destroy (listener->client_mq);
 256     listener->client_mq = NULL;
 257   }
 258   GNUNET_CONTAINER_DLL_remove (listeners_head, listeners_tail, listener);
 259   GNUNET_free (listener);
 260 }
 261
 262
 263 /**
 264  * Iterator over hash map entries.
 265  *
 266  * @param cls closure
 267  * @param key current key code
 268  * @param value value in the hash map
 269  * @return GNUNET_YES if we should continue to
 270  *         iterate,
 271  *         GNUNET_NO if not.
 272  */
 273 static int
 274 destroy_elements_iterator (void *cls,
 275                            const struct GNUNET_HashCode * key,
 276                            void *value)
 277 {
 278   struct ElementEntry *ee = value;
 279
 280   GNUNET_free (ee);
 281   return GNUNET_YES;
 282 }
 283
 284
 285 /**
 286  * Destroy a set, and free all resources associated with it.
 287  *
 288  * @param set the set to destroy
 289  */
 290 static void
 291 set_destroy (struct Set *set)
 292 {
 293   /* If the client is not dead yet, destroy it.
 294    * The client's destroy callback will destroy the set again.
 295    * We do this so that the tunnel end handler still has a valid set handle
 296    * to destroy. */
 297   if (NULL != set->client)
 298   {
 299     struct GNUNET_SERVER_Client *client = set->client;
 300     set->client = NULL;
 301     GNUNET_SERVER_client_disconnect (client);
 302     return;
 303   }
 304   GNUNET_assert (NULL != set->state);
 305   set->vt->destroy_set (set->state);
 306   set->state = NULL;
 307   if (NULL != set->client_mq)
 308   {
 309     GNUNET_MQ_destroy (set->client_mq);
 310     set->client_mq = NULL;
 311   }
 312   if (NULL != set->iter)
 313   {
 314     GNUNET_CONTAINER_multihashmap_iterator_destroy (set->iter);
 315     set->iter = NULL;
 316   }
 317   GNUNET_CONTAINER_DLL_remove (sets_head, sets_tail, set);
 318   if (NULL != set->elements)
 319   {
 320     GNUNET_CONTAINER_multihashmap_iterate (set->elements,
 321                                            destroy_elements_iterator, NULL);
 322     GNUNET_CONTAINER_multihashmap_destroy (set->elements);
 323     set->elements = NULL;
 324   }
 325   GNUNET_free (set);
 326 }
 327
 328
 329 /**
 330  * Clean up after a client after it is
 331  * disconnected (either by us or by itself)
 332  *
 333  * @param cls closure, unused
 334  * @param client the client to clean up after
 335  */
 336 static void
 337 handle_client_disconnect (void *cls, struct GNUNET_SERVER_Client *client)
 338 {
 339   struct Set *set;
 340   struct Listener *listener;
 341
 342   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "client disconnected, cleaning up\n");
 343
 344   set = set_get (client);
 345   if (NULL != set)
 346   {
 347     set->client = NULL;
 348     set_destroy (set);
 349     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "(client's set destroyed)\n");
 350   }
 351   listener = listener_get (client);
 352   if (NULL != listener)
 353   {
 354     listener->client = NULL;
 355     listener_destroy (listener);
 356     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "(client's listener destroyed)\n");
 357   }
 358 }
 359
 360
 361 /**
 362  * Destroy an incoming request from a remote peer
 363  *
 364  * @param incoming remote request to destroy
 365  */
 366 static void
 367 incoming_destroy (struct Incoming *incoming)
 368 {
 369   GNUNET_CONTAINER_DLL_remove (incoming_head, incoming_tail, incoming);
 370   if (GNUNET_SCHEDULER_NO_TASK != incoming->timeout_task)
 371   {
 372     GNUNET_SCHEDULER_cancel (incoming->timeout_task);
 373     incoming->timeout_task = GNUNET_SCHEDULER_NO_TASK;
 374   }
 375   if (NULL != incoming->tunnel)
 376   {
 377     struct GNUNET_MESH_Tunnel *t = incoming->tunnel;
 378     incoming->tunnel = NULL;
 379     GNUNET_MESH_tunnel_destroy (t);
 380     return;
 381   }
 382   GNUNET_free (incoming);
 383 }
 384
 385
 386 static struct Listener *
 387 listener_get_by_target (enum GNUNET_SET_OperationType op,
 388                         const struct GNUNET_HashCode *app_id)
 389 {
 390   struct Listener *l;
 391
 392   for (l = listeners_head; NULL != l; l = l->next)
 393   {
 394     if (l->operation != op)
 395       continue;
 396     if (0 != GNUNET_CRYPTO_hash_cmp (app_id, &l->app_id))
 397       continue;
 398     return l;
 399   }
 400   return NULL;
 401 }
 402
 403
 404 /**
 405  * Suggest the given request to the listener,
 406  * who can accept or reject the request.
 407  *
 408  * @param incoming the incoming peer with the request to suggest
 409  * @param listener the listener to suggest the request to
 410  */
 411 static void
 412 incoming_suggest (struct Incoming *incoming, struct Listener *listener)
 413 {
 414   struct GNUNET_MQ_Envelope *mqm;
 415   struct GNUNET_SET_RequestMessage *cmsg;
 416
 417   GNUNET_assert (0 == incoming->suggest_id);
 418   GNUNET_assert (NULL != incoming->spec);
 419   incoming->suggest_id = suggest_id++;
 420
 421   GNUNET_SCHEDULER_cancel (incoming->timeout_task);
 422   incoming->timeout_task = GNUNET_SCHEDULER_NO_TASK;
 423   mqm = GNUNET_MQ_msg_nested_mh (cmsg, GNUNET_MESSAGE_TYPE_SET_REQUEST,
 424                                  incoming->spec->context_msg);
 425   GNUNET_assert (NULL != mqm);
 426   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "suggesting request with accept id %u\n",
 427               incoming->suggest_id);
 428   cmsg->accept_id = htonl (incoming->suggest_id);
 429   cmsg->peer_id = incoming->spec->peer;
 430   GNUNET_MQ_send (listener->client_mq, mqm);
 431 }
 432
 433
 434 /**
 435  * Handle a request for a set operation from
 436  * another peer.
 437  *
 438  * @param op the operation state
 439  * @param mh the received message
 440  * @return GNUNET_OK if the tunnel should be kept alive,
 441  *         GNUNET_SYSERR to destroy the tunnel
 442  */
 443 static int
 444 handle_incoming_msg (struct OperationState *op,
 445                      const struct GNUNET_MessageHeader *mh)
 446 {
 447   struct Incoming *incoming = (struct Incoming *) op;
 448   const struct OperationRequestMessage *msg = (const struct OperationRequestMessage *) mh;
 449   struct Listener *listener;
 450   struct OperationSpecification *spec;
 451
 452   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "got op request\n");
 453
 454   if (GNUNET_MESSAGE_TYPE_SET_P2P_OPERATION_REQUEST != ntohs (mh->type))
 455   {
 456     GNUNET_break_op (0);
 457     return GNUNET_SYSERR;
 458   }
 459
 460   if (NULL != incoming->spec)
 461   {
 462     /* double operation request */
 463     GNUNET_break_op (0);
 464     return GNUNET_SYSERR;
 465   }
 466
 467   spec = GNUNET_new (struct OperationSpecification);
 468   spec->context_msg = GNUNET_MQ_extract_nested_mh (msg);
 469   if (NULL != spec->context_msg)
 470     spec->context_msg = GNUNET_copy_message (spec->context_msg);
 471   spec->operation = ntohl (msg->operation);
 472   spec->app_id = msg->app_id;
 473   spec->salt = ntohl (msg->salt);
 474   spec->peer = incoming->peer;
 475
 476   incoming->spec = spec;
 477
 478   if ( (NULL != spec->context_msg) &&
 479        (ntohs (spec->context_msg->size) > GNUNET_SET_CONTEXT_MESSAGE_MAX_SIZE) )
 480   {
 481     GNUNET_break_op (0);
 482     return GNUNET_SYSERR;
 483   }
 484
 485   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "received P2P operation request (op %u, app %s)\n",
 486               ntohl (msg->operation), GNUNET_h2s (&msg->app_id));
 487   listener = listener_get_by_target (ntohl (msg->operation), &msg->app_id);
 488   if (NULL == listener)
 489   {
 490     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 491                 "no listener matches incoming request, waiting with timeout\n");
 492     return GNUNET_OK;
 493   }
 494   incoming_suggest (incoming, listener);
 495   return GNUNET_OK;
 496 }
 497
 498
 499 static void
 500 send_client_element (struct Set *set)
 501 {
 502   int ret;
 503   struct ElementEntry *ee;
 504   struct GNUNET_MQ_Envelope *ev;
 505
 506   GNUNET_assert (NULL != set->iter);
 507   ret = GNUNET_CONTAINER_multihashmap_iterator_next (set->iter, NULL, (const void **) &ee);
 508   if (GNUNET_NO == ret)
 509   {
 510     ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_SET_ITER_DONE);
 511   }
 512   else
 513   {
 514     struct GNUNET_SET_IterResponseMessage *msg;
 515
 516     GNUNET_assert (NULL != ee);
 517     ev = GNUNET_MQ_msg_extra (msg, ee->element.size, GNUNET_MESSAGE_TYPE_SET_ITER_ELEMENT);
 518     memcpy (&msg[1], ee->element.data, ee->element.size);
 519     msg->element_type = ee->element.type;
 520   }
 521   GNUNET_MQ_send (set->client_mq, ev);
 522 }
 523
 524
 525 /**
 526  * Called when a client wants to iterate the elements of a set.
 527  *
 528  * @param cls unused
 529  * @param client client that sent the message
 530  * @param m message sent by the client
 531  */
 532 static void
 533 handle_client_iterate (void *cls,
 534                        struct GNUNET_SERVER_Client *client,
 535                        const struct GNUNET_MessageHeader *m)
 536 {
 537   struct Set *set;
 538
 539   set = set_get (client);
 540   if (NULL == set)
 541   {
 542     GNUNET_break (0);
 543     GNUNET_SERVER_client_disconnect (client);
 544     return;
 545   }
 546
 547   if (NULL != set->iter)
 548   {
 549     GNUNET_break (0);
 550     GNUNET_SERVER_client_disconnect (client);
 551     return;
 552   }
 553   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "iterating union set with %u elements\n",
 554               GNUNET_CONTAINER_multihashmap_size (set->elements));
 555   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 556   set->iter = GNUNET_CONTAINER_multihashmap_iterator_create (set->elements);
 557   send_client_element (set);
 558 }
 559
 560
 561 /**
 562  * Called when a client wants to create a new set.
 563  *
 564  * @param cls unused
 565  * @param client client that sent the message
 566  * @param m message sent by the client
 567  */
 568 static void
 569 handle_client_create (void *cls,
 570                       struct GNUNET_SERVER_Client *client,
 571                       const struct GNUNET_MessageHeader *m)
 572 {
 573   struct GNUNET_SET_CreateMessage *msg = (struct GNUNET_SET_CreateMessage *) m;
 574   struct Set *set;
 575
 576   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "client created new set (operation %u)\n",
 577               ntohs (msg->operation));
 578
 579   if (NULL != set_get (client))
 580   {
 581     GNUNET_break (0);
 582     GNUNET_SERVER_client_disconnect (client);
 583     return;
 584   }
 585
 586   set = GNUNET_new (struct Set);
 587
 588   switch (ntohs (msg->operation))
 589   {
 590     case GNUNET_SET_OPERATION_INTERSECTION:
 591 //      set->vt = _GSS_intersection_vt ();
 592       break;
 593     case GNUNET_SET_OPERATION_UNION:
 594       set->vt = _GSS_union_vt ();
 595       break;
 596     default:
 597       GNUNET_free (set);
 598       GNUNET_break (0);
 599       GNUNET_SERVER_client_disconnect (client);
 600       return;
 601   }
 602
 603   set->state = set->vt->create ();
 604   set->elements = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
 605   set->client = client;
 606   set->client_mq = GNUNET_MQ_queue_for_server_client (client);
 607   GNUNET_CONTAINER_DLL_insert (sets_head, sets_tail, set);
 608   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 609 }
 610
 611
 612 /**
 613  * Called when a client wants to create a new listener.
 614  *
 615  * @param cls unused
 616  * @param client client that sent the message
 617  * @param m message sent by the client
 618  */
 619 static void
 620 handle_client_listen (void *cls,
 621                       struct GNUNET_SERVER_Client *client,
 622                       const struct GNUNET_MessageHeader *m)
 623 {
 624   struct GNUNET_SET_ListenMessage *msg = (struct GNUNET_SET_ListenMessage *) m;
 625   struct Listener *listener;
 626   struct Incoming *incoming;
 627
 628   if (NULL != listener_get (client))
 629   {
 630     GNUNET_break (0);
 631     GNUNET_SERVER_client_disconnect (client);
 632     return;
 633   }
 634   listener = GNUNET_new (struct Listener);
 635   listener->client = client;
 636   listener->client_mq = GNUNET_MQ_queue_for_server_client (client);
 637   listener->app_id = msg->app_id;
 638   listener->operation = ntohl (msg->operation);
 639   GNUNET_CONTAINER_DLL_insert_tail (listeners_head, listeners_tail, listener);
 640   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "new listener created (op %u, app %s)\n",
 641               listener->operation, GNUNET_h2s (&listener->app_id));
 642   for (incoming = incoming_head; NULL != incoming; incoming = incoming->next)
 643   {
 644     if (NULL == incoming->spec)
 645     {
 646       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "request has no spec yet\n");
 647       continue;
 648     }
 649     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "considering (op: %u, app: %s, suggest: %u)\n",
 650                 incoming->spec->operation, GNUNET_h2s (&incoming->spec->app_id), incoming->suggest_id);
 651
 652     if (0 != incoming->suggest_id)
 653       continue;
 654     if (listener->operation != incoming->spec->operation)
 655       continue;
 656     if (0 != GNUNET_CRYPTO_hash_cmp (&listener->app_id, &incoming->spec->app_id))
 657       continue;
 658     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "request suggested\n");
 659     incoming_suggest (incoming, listener);
 660   }
 661   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "considered all incoming requests\n");
 662   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 663 }
 664
 665
 666 /**
 667  * Called when the client wants to reject an operation
 668  * request from another peer.
 669  *
 670  * @param cls unused
 671  * @param client client that sent the message
 672  * @param m message sent by the client
 673  */
 674 static void
 675 handle_client_reject (void *cls,
 676                       struct GNUNET_SERVER_Client *client,
 677                       const struct GNUNET_MessageHeader *m)
 678 {
 679   struct Incoming *incoming;
 680   const struct GNUNET_SET_AcceptRejectMessage *msg;
 681
 682   msg = (const struct GNUNET_SET_AcceptRejectMessage *) m;
 683   GNUNET_break (0 == ntohl (msg->request_id));
 684
 685   incoming = get_incoming (ntohl (msg->accept_reject_id));
 686   if (NULL == incoming)
 687   {
 688     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 689     return;
 690   }
 691   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "peer request rejected by client\n");
 692   GNUNET_MESH_tunnel_destroy (incoming->tunnel);
 693   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 694 }
 695
 696
 697 /**
 698  * Called when a client wants to add an element to a
 699  * set it inhabits.
 700  *
 701  * @param cls unused
 702  * @param client client that sent the message
 703  * @param m message sent by the client
 704  */
 705 static void
 706 handle_client_add_remove (void *cls,
 707                           struct GNUNET_SERVER_Client *client,
 708                           const struct GNUNET_MessageHeader *m)
 709 {
 710   struct Set *set;
 711   const struct GNUNET_SET_ElementMessage *msg;
 712   struct GNUNET_SET_Element el;
 713   struct ElementEntry *ee;
 714
 715   set = set_get (client);
 716   if (NULL == set)
 717   {
 718     GNUNET_break (0);
 719     GNUNET_SERVER_client_disconnect (client);
 720     return;
 721   }
 722   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 723   msg = (const struct GNUNET_SET_ElementMessage *) m;
 724   el.size = ntohs (m->size) - sizeof *msg;
 725   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 726               "client ins/rem element of size %u\n", el.size);
 727   el.data = &msg[1];
 728   if (GNUNET_MESSAGE_TYPE_SET_REMOVE == ntohs (m->type))
 729   {
 730     struct GNUNET_HashCode hash;
 731
 732     GNUNET_CRYPTO_hash (el.data, el.size, &hash);
 733     ee = GNUNET_CONTAINER_multihashmap_get (set->elements, &hash);
 734     if (NULL == ee)
 735     {
 736       GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "client tried to remove non-existing element\n");
 737       return;
 738     }
 739     if (GNUNET_YES == ee->removed)
 740     {
 741       GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "client tried to remove element twice\n");
 742       return;
 743     }
 744     ee->removed = GNUNET_YES;
 745     ee->generation_removed = set->current_generation;
 746     set->vt->remove (set->state, ee);
 747   }
 748   else
 749   {
 750     struct ElementEntry *ee_dup;
 751
 752     ee = GNUNET_malloc (el.size + sizeof *ee);
 753     ee->element.size = el.size;
 754     memcpy (&ee[1], el.data, el.size);
 755     ee->element.data = &ee[1];
 756     ee->generation_added = set->current_generation;
 757     ee->remote = GNUNET_NO;
 758     GNUNET_CRYPTO_hash (ee->element.data, el.size, &ee->element_hash);
 759     ee_dup = GNUNET_CONTAINER_multihashmap_get (set->elements,
 760                                                 &ee->element_hash);
 761     if (NULL != ee_dup)
 762     {
 763       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "element inserted twice, ignoring\n");
 764       GNUNET_free (ee);
 765       return;
 766     }
 767     GNUNET_CONTAINER_multihashmap_put (set->elements, &ee->element_hash, ee,
 768                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
 769     set->vt->add (set->state, ee);
 770   }
 771 }
 772
 773
 774 /**
 775  * Called when a client wants to evaluate a set operation with another peer.
 776  *
 777  * @param cls unused
 778  * @param client client that sent the message
 779  * @param m message sent by the client
 780  */
 781 static void
 782 handle_client_evaluate (void *cls,
 783                         struct GNUNET_SERVER_Client *client,
 784                         const struct GNUNET_MessageHeader *m)
 785 {
 786   struct Set *set;
 787   struct TunnelContext *tc;
 788   struct GNUNET_MESH_Tunnel *tunnel;
 789   struct GNUNET_SET_EvaluateMessage *msg;
 790   struct OperationSpecification *spec;
 791
 792   set = set_get (client);
 793   if (NULL == set)
 794   {
 795     GNUNET_break (0);
 796     GNUNET_SERVER_client_disconnect (client);
 797     return;
 798   }
 799
 800   msg = (struct GNUNET_SET_EvaluateMessage *) m;
 801   tc = GNUNET_new (struct TunnelContext);
 802   spec = GNUNET_new (struct OperationSpecification);
 803   spec->operation = set->operation;
 804   spec->app_id = msg->app_id;
 805   spec->salt = ntohl (msg->salt);
 806   spec->peer = msg->target_peer;
 807   spec->set = set;
 808   spec->client_request_id = ntohl (msg->request_id);
 809   spec->context_msg = GNUNET_MQ_extract_nested_mh (msg);
 810   if (NULL != spec->context_msg)
 811     spec->context_msg = GNUNET_copy_message (spec->context_msg);
 812
 813   tunnel = GNUNET_MESH_tunnel_create (mesh, tc, &msg->target_peer,
 814                                       GNUNET_APPLICATION_TYPE_SET,
 815                                       GNUNET_YES,
 816                                       GNUNET_YES);
 817
 818   set->vt->evaluate (spec, tunnel, tc);
 819
 820   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 821 }
 822
 823
 824 /**
 825  * Handle an ack from a client.
 826  *
 827  * @param cls unused
 828  * @param client the client
 829  * @param m the message
 830  */
 831 static void
 832 handle_client_iter_ack (void *cls,
 833                    struct GNUNET_SERVER_Client *client,
 834                    const struct GNUNET_MessageHeader *m)
 835 {
 836   struct Set *set;
 837
 838   set = set_get (client);
 839   if (NULL == set)
 840   {
 841     GNUNET_break (0);
 842     GNUNET_SERVER_client_disconnect (client);
 843     return;
 844   }
 845
 846   if (NULL == set->iter)
 847   {
 848     GNUNET_break (0);
 849     GNUNET_SERVER_client_disconnect (client);
 850     return;
 851   }
 852
 853   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 854   send_client_element (set);
 855 }
 856
 857
 858 /**
 859  * Handle a request from the client to accept
 860  * a set operation that came from a remote peer.
 861  *
 862  * @param cls unused
 863  * @param client the client
 864  * @param mh the message
 865  */
 866 static void
 867 handle_client_cancel (void *cls,
 868                       struct GNUNET_SERVER_Client *client,
 869                       const struct GNUNET_MessageHeader *mh)
 870 {
 871   const struct GNUNET_SET_CancelMessage *msg =
 872       (const struct GNUNET_SET_CancelMessage *) mh;
 873   struct Set *set;
 874
 875   set = set_get (client);
 876   if (NULL == set)
 877   {
 878     GNUNET_break (0);
 879     GNUNET_SERVER_client_disconnect (client);
 880     return;
 881   }
 882   /* FIXME: maybe cancel should return success/error code? */
 883   set->vt->cancel (set->state, ntohl (msg->request_id));
 884 }
 885
 886
 887 /**
 888  * Handle a request from the client to accept
 889  * a set operation that came from a remote peer.
 890  *
 891  * @param cls unused
 892  * @param client the client
 893  * @param mh the message
 894  */
 895 static void
 896 handle_client_accept (void *cls,
 897                       struct GNUNET_SERVER_Client *client,
 898                       const struct GNUNET_MessageHeader *mh)
 899 {
 900   struct Set *set;
 901   struct Incoming *incoming;
 902   struct GNUNET_SET_AcceptRejectMessage *msg = (struct GNUNET_SET_AcceptRejectMessage *) mh;
 903
 904   incoming = get_incoming (ntohl (msg->accept_reject_id));
 905
 906   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "client accepting %u\n", ntohl (msg->accept_reject_id));
 907
 908   if (NULL == incoming)
 909   {
 910     GNUNET_break (0);
 911     GNUNET_SERVER_client_disconnect (client);
 912     return;
 913   }
 914
 915   set = set_get (client);
 916
 917   if (NULL == set)
 918   {
 919     GNUNET_break (0);
 920     GNUNET_SERVER_client_disconnect (client);
 921     return;
 922   }
 923
 924   incoming->spec->set = set;
 925   incoming->spec->client_request_id = ntohl (msg->request_id);
 926   set->vt->accept (incoming->spec, incoming->tunnel, incoming->tc);
 927   /* tunnel ownership goes to operation */
 928   incoming->tunnel = NULL;
 929   incoming_destroy (incoming);
 930   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 931 }
 932
 933
 934 /**
 935  * Called to clean up, after a shutdown has been requested.
 936  *
 937  * @param cls closure
 938  * @param tc context information (why was this task triggered now)
 939  */
 940 static void
 941 shutdown_task (void *cls,
 942                const struct GNUNET_SCHEDULER_TaskContext *tc)
 943 {
 944   while (NULL != incoming_head)
 945     incoming_destroy (incoming_head);
 946
 947   while (NULL != listeners_head)
 948     listener_destroy (listeners_head);
 949
 950   while (NULL != sets_head)
 951     set_destroy (sets_head);
 952
 953
 954   /* it's important to destroy mesh at the end, as tunnels
 955    * must be destroyed first! */
 956   if (NULL != mesh)
 957   {
 958     GNUNET_MESH_disconnect (mesh);
 959     mesh = NULL;
 960   }
 961
 962   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "handled shutdown request\n");
 963 }
 964
 965
 966 /**
 967  * Signature of the main function of a task.
 968  *
 969  * @param cls closure
 970  * @param tc context information (why was this task triggered now)
 971  */
 972 static void
 973 incoming_timeout_cb (void *cls,
 974                      const struct GNUNET_SCHEDULER_TaskContext *tc)
 975 {
 976   struct Incoming *incoming = cls;
 977
 978   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
 979     return;
 980
 981   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "remote peer timed out\n");
 982   incoming_destroy (incoming);
 983 }
 984
 985
 986 static void
 987 handle_incoming_disconnect (struct OperationState *op_state)
 988 {
 989   struct Incoming *incoming = (struct Incoming *) op_state;
 990   if (NULL == incoming->tunnel)
 991     return;
 992
 993   incoming_destroy (incoming);
 994 }
 995
 996
 997 /**
 998  * Method called whenever another peer has added us to a tunnel
 999  * the other peer initiated.
1000  * Only called (once) upon reception of data with a message type which was
1001  * subscribed to in GNUNET_MESH_connect. A call to GNUNET_MESH_tunnel_destroy
1002  * causes te tunnel to be ignored and no further notifications are sent about
1003  * the same tunnel.
1004  *
1005  * @param cls closure
1006  * @param tunnel new handle to the tunnel
1007  * @param initiator peer that started the tunnel
1008  * @param port Port this tunnel is for.
1009  * @return initial tunnel context for the tunnel
1010  *         (can be NULL -- that's not an error)
1011  */
1012 static void *
1013 tunnel_new_cb (void *cls,
1014                struct GNUNET_MESH_Tunnel *tunnel,
1015                const struct GNUNET_PeerIdentity *initiator,
1016                uint32_t port)
1017 {
1018   struct Incoming *incoming;
1019   static const struct SetVT incoming_vt = {
1020     .msg_handler = handle_incoming_msg,
1021     .peer_disconnect = handle_incoming_disconnect
1022   };
1023
1024   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "new incoming tunnel\n");
1025
1026   GNUNET_assert (port == GNUNET_APPLICATION_TYPE_SET);
1027   incoming = GNUNET_new (struct Incoming);
1028   incoming->peer = *initiator;
1029   incoming->tunnel = tunnel;
1030   incoming->tc = GNUNET_new (struct TunnelContext);;
1031   incoming->tc->vt = &incoming_vt;
1032   incoming->tc->op = (struct OperationState *) incoming;
1033   incoming->timeout_task =
1034       GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES, incoming_timeout_cb, incoming);
1035   GNUNET_CONTAINER_DLL_insert_tail (incoming_head, incoming_tail, incoming);
1036
1037   return incoming->tc;
1038 }
1039
1040
1041 /**
1042  * Function called whenever a tunnel is destroyed.  Should clean up
1043  * any associated state.
1044  * GNUNET_MESH_tunnel_destroy. It must NOT call GNUNET_MESH_tunnel_destroy on
1045  * the tunnel.
1046  *
1047  * @param cls closure (set from GNUNET_MESH_connect)
1048  * @param tunnel connection to the other end (henceforth invalid)
1049  * @param tunnel_ctx place where local state associated
1050  *                   with the tunnel is stored
1051  */
1052 static void
1053 tunnel_end_cb (void *cls,
1054                const struct GNUNET_MESH_Tunnel *tunnel, void *tunnel_ctx)
1055 {
1056   struct TunnelContext *ctx = tunnel_ctx;
1057
1058   ctx->vt->peer_disconnect (ctx->op);
1059   /* mesh will never call us with the context again! */
1060   GNUNET_free (tunnel_ctx);
1061 }
1062
1063
1064 /**
1065  * Functions with this signature are called whenever a message is
1066  * received.
1067  *
1068  * Each time the function must call GNUNET_MESH_receive_done on the tunnel
1069  * in order to receive the next message. This doesn't need to be immediate:
1070  * can be delayed if some processing is done on the message.
1071  *
1072  * @param cls Closure (set from GNUNET_MESH_connect).
1073  * @param tunnel Connection to the other end.
1074  * @param tunnel_ctx Place to store local state associated with the tunnel.
1075  * @param message The actual message.
1076  *
1077  * @return GNUNET_OK to keep the tunnel open,
1078  *         GNUNET_SYSERR to close it (signal serious error).
1079  */
1080 static int
1081 dispatch_p2p_message (void *cls,
1082                       struct GNUNET_MESH_Tunnel *tunnel,
1083                       void **tunnel_ctx,
1084                       const struct GNUNET_MessageHeader *message)
1085 {
1086   struct TunnelContext *tc = *tunnel_ctx;
1087   int ret;
1088
1089   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "dispatching mesh message (type: %u)\n",
1090               ntohs (message->type));
1091   /* do this before the handler, as the handler might kill the tunnel */
1092   GNUNET_MESH_receive_done (tunnel);
1093   ret = tc->vt->msg_handler (tc->op, message);
1094   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "handled mesh message (type: %u)\n",
1095               ntohs (message->type));
1096   return ret;
1097 }
1098
1099
1100 /**
1101  * Function called by the service's run
1102  * method to run service-specific setup code.
1103  *
1104  * @param cls closure
1105  * @param server the initialized server
1106  * @param cfg configuration to use
1107  */
1108 static void
1109 run (void *cls, struct GNUNET_SERVER_Handle *server,
1110      const struct GNUNET_CONFIGURATION_Handle *cfg)
1111 {
1112   static const struct GNUNET_SERVER_MessageHandler server_handlers[] = {
1113     {handle_client_accept, NULL, GNUNET_MESSAGE_TYPE_SET_ACCEPT,
1114         sizeof (struct GNUNET_SET_AcceptRejectMessage)},
1115     {handle_client_iter_ack, NULL, GNUNET_MESSAGE_TYPE_SET_ITER_ACK, 0},
1116     {handle_client_add_remove, NULL, GNUNET_MESSAGE_TYPE_SET_ADD, 0},
1117     {handle_client_create, NULL, GNUNET_MESSAGE_TYPE_SET_CREATE,
1118         sizeof (struct GNUNET_SET_CreateMessage)},
1119     {handle_client_iterate, NULL, GNUNET_MESSAGE_TYPE_SET_ITER_REQUEST,
1120         sizeof (struct GNUNET_MessageHeader)},
1121     {handle_client_evaluate, NULL, GNUNET_MESSAGE_TYPE_SET_EVALUATE, 0},
1122     {handle_client_listen, NULL, GNUNET_MESSAGE_TYPE_SET_LISTEN,
1123         sizeof (struct GNUNET_SET_ListenMessage)},
1124     {handle_client_reject, NULL, GNUNET_MESSAGE_TYPE_SET_REJECT,
1125         sizeof (struct GNUNET_SET_AcceptRejectMessage)},
1126     {handle_client_add_remove, NULL, GNUNET_MESSAGE_TYPE_SET_REMOVE, 0},
1127     {handle_client_cancel, NULL, GNUNET_MESSAGE_TYPE_SET_REMOVE,
1128         sizeof (struct GNUNET_SET_CancelMessage)},
1129     {NULL, NULL, 0, 0}
1130   };
1131   static const struct GNUNET_MESH_MessageHandler mesh_handlers[] = {
1132     {dispatch_p2p_message, GNUNET_MESSAGE_TYPE_SET_P2P_OPERATION_REQUEST, 0},
1133     {dispatch_p2p_message, GNUNET_MESSAGE_TYPE_SET_P2P_IBF, 0},
1134     {dispatch_p2p_message, GNUNET_MESSAGE_TYPE_SET_P2P_ELEMENTS, 0},
1135     {dispatch_p2p_message, GNUNET_MESSAGE_TYPE_SET_P2P_DONE, 0},
1136     {dispatch_p2p_message, GNUNET_MESSAGE_TYPE_SET_P2P_ELEMENT_REQUESTS, 0},
1137     {dispatch_p2p_message, GNUNET_MESSAGE_TYPE_SET_P2P_SE, 0},
1138     {NULL, 0, 0}
1139   };
1140   static const uint32_t mesh_ports[] = {GNUNET_APPLICATION_TYPE_SET, 0};
1141
1142   configuration = cfg;
1143   GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL,
1144                                 &shutdown_task, NULL);
1145   GNUNET_SERVER_disconnect_notify (server, &handle_client_disconnect, NULL);
1146   GNUNET_SERVER_add_handlers (server, server_handlers);
1147
1148   mesh = GNUNET_MESH_connect (cfg, NULL, tunnel_new_cb, tunnel_end_cb,
1149                               mesh_handlers, mesh_ports);
1150   if (NULL == mesh)
1151   {
1152     GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "could not connect to mesh\n");
1153     return;
1154   }
1155
1156   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "started\n");
1157 }
1158
1159
1160 /**
1161  * The main function for the set service.
1162  *
1163  * @param argc number of arguments from the command line
1164  * @param argv command line arguments
1165  * @return 0 ok, 1 on error
1166  */
1167 int
1168 main (int argc, char *const *argv)
1169 {
1170   int ret;
1171   ret = GNUNET_SERVICE_run (argc, argv, "set",
1172                             GNUNET_SERVICE_OPTION_NONE, &run, NULL);
1173   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "exit (%d)\n", GNUNET_OK != ret);
1174   return (GNUNET_OK == ret) ? 0 : 1;
1175 }
1176