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