src/core/gnunet-service-core_clients.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2009, 2010, 2011 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 core/gnunet-service-core_clients.c
  23  * @brief code for managing interactions with clients of core service
  24  * @author Christian Grothoff
  25  */
  26 #include "platform.h"
  27 #include "gnunet_util_lib.h"
  28 #include "gnunet_transport_service.h"
  29 #include "gnunet_service_core.h"
  30 #include "gnunet_service_core_clients.h"
  31 #include "gnunet_service_core_sessions.h"
  32
  33
  34 /**
  35  * Data structure for each client connected to the core service.
  36  */
  37 struct Client
  38 {
  39   /**
  40    * Clients are kept in a linked list.
  41    */
  42   struct Client *next;
  43
  44   /**
  45    * Handle for the client with the server API.
  46    */
  47   struct GNUNET_SERVER_Client *client_handle;
  48
  49   /**
  50    * Array of the types of messages this peer cares
  51    * about (with "tcnt" entries).  Allocated as part
  52    * of this client struct, do not free!
  53    */
  54   const uint16_t *types;
  55
  56   /**
  57    * Map of peer identities to active transmission requests of this
  58    * client to the peer (of type 'struct ClientActiveRequest').
  59    */
  60   struct GNUNET_CONTAINER_MultiHashMap *requests;
  61
  62   /**
  63    * Options for messages this client cares about,
  64    * see GNUNET_CORE_OPTION_ values.
  65    */
  66   uint32_t options;
  67
  68   /**
  69    * Number of types of incoming messages this client
  70    * specifically cares about.  Size of the "types" array.
  71    */
  72   unsigned int tcnt;
  73
  74 };
  75
  76
  77 /**
  78  * Record kept for each request for transmission issued by a
  79  * client that is still pending.
  80  */
  81 struct ClientActiveRequest
  82 {
  83
  84   /**
  85    * Active requests are kept in a doubly-linked list of
  86    * the respective target peer.
  87    */
  88   struct ClientActiveRequest *next;
  89
  90   /**
  91    * Active requests are kept in a doubly-linked list of
  92    * the respective target peer.
  93    */
  94   struct ClientActiveRequest *prev;
  95
  96   /**
  97    * Handle to the client.
  98    */
  99   struct Client *client;
 100
 101   /**
 102    * By what time would the client want to see this message out?
 103    */
 104   struct GNUNET_TIME_Absolute deadline;
 105
 106   /**
 107    * How important is this request.
 108    */
 109   uint32_t priority;
 110
 111   /**
 112    * How many more requests does this client have?
 113    */
 114   uint32_t queue_size;
 115
 116   /**
 117    * How many bytes does the client intend to send?
 118    */
 119   uint16_t msize;
 120
 121   /**
 122    * Unique request ID (in big endian).
 123    */
 124   uint16_t smr_id;
 125
 126 };
 127
 128
 129
 130 /**
 131  * Linked list of our clients.
 132  */
 133 static struct Client *clients;
 134
 135 /**
 136  * Context for notifications we need to send to our clients.
 137  */
 138 static struct GNUNET_SERVER_NotificationContext *notifier;
 139
 140
 141 /**
 142  * Our message stream tokenizer (for encrypted payload).
 143  */
 144 static struct GNUNET_SERVER_MessageStreamTokenizer *mst;
 145
 146
 147
 148 /**
 149  * Send a message to one of our clients.
 150  *
 151  * @param client target for the message
 152  * @param msg message to transmit
 153  * @param can_drop could this message be dropped if the
 154  *        client's queue is getting too large?
 155  */
 156 static void
 157 send_to_client (struct Client *client, const struct GNUNET_MessageHeader *msg,
 158                 int can_drop)
 159 {
 160 #if DEBUG_CORE_CLIENT
 161   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 162               "Preparing to send %u bytes of message of type %u to client.\n",
 163               (unsigned int) ntohs (msg->size),
 164               (unsigned int) ntohs (msg->type));
 165 #endif
 166   GNUNET_SERVER_notification_context_unicast (notifier, client->client_handle,
 167                                               msg, can_drop);
 168 }
 169
 170
 171
 172
 173
 174 /**
 175  * Send a message to all of our current clients that have
 176  * the right options set.
 177  *
 178  * @param msg message to multicast
 179  * @param can_drop can this message be discarded if the queue is too long
 180  * @param options mask to use
 181  */
 182 static void
 183 send_to_all_clients (const struct GNUNET_MessageHeader *msg, int can_drop,
 184                      int options)
 185 {
 186   struct Client *c;
 187
 188   c = clients;
 189   while (c != NULL)
 190   {
 191     if (0 != (c->options & options))
 192     {
 193 #if DEBUG_CORE_CLIENT > 1
 194       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 195                   "Sending message of type %u to client.\n",
 196                   (unsigned int) ntohs (msg->type));
 197 #endif
 198       send_to_client (c, msg, can_drop);
 199     }
 200     c = c->next;
 201   }
 202 }
 203
 204
 205
 206 /**
 207  * Handle CORE_SEND_REQUEST message.
 208  */
 209 static void
 210 handle_client_send_request (void *cls, struct GNUNET_SERVER_Client *client,
 211                             const struct GNUNET_MessageHeader *message)
 212 {
 213   const struct SendMessageRequest *req;
 214   struct Neighbour *n;
 215   struct Client *c;
 216   struct ClientActiveRequest *car;
 217
 218   req = (const struct SendMessageRequest *) message;
 219   if (0 ==
 220       memcmp (&req->peer, &my_identity, sizeof (struct GNUNET_PeerIdentity)))
 221     n = &self;
 222   else
 223     n = find_neighbour (&req->peer);
 224   if ((n == NULL) || (GNUNET_YES != n->is_connected) ||
 225       (n->status != PEER_STATE_KEY_CONFIRMED))
 226   {
 227     /* neighbour must have disconnected since request was issued,
 228      * ignore (client will realize it once it processes the
 229      * disconnect notification) */
 230 #if DEBUG_CORE_CLIENT
 231     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 232                 "Dropped client request for transmission (am disconnected)\n");
 233 #endif
 234     GNUNET_STATISTICS_update (stats,
 235                               gettext_noop
 236                               ("# send requests dropped (disconnected)"), 1,
 237                               GNUNET_NO);
 238     GNUNET_SERVER_receive_done (client, GNUNET_OK);
 239     return;
 240   }
 241   c = clients;
 242   while ((c != NULL) && (c->client_handle != client))
 243     c = c->next;
 244   if (c == NULL)
 245   {
 246     /* client did not send INIT first! */
 247     GNUNET_break (0);
 248     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 249     return;
 250   }
 251   if (c->requests == NULL)
 252     c->requests = GNUNET_CONTAINER_multihashmap_create (16);
 253 #if DEBUG_CORE_CLIENT
 254   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 255               "Received client transmission request. queueing\n");
 256 #endif
 257   car = GNUNET_CONTAINER_multihashmap_get (c->requests, &req->peer.hashPubKey);
 258   if (car == NULL)
 259   {
 260     /* create new entry */
 261     car = GNUNET_malloc (sizeof (struct ClientActiveRequest));
 262     GNUNET_assert (GNUNET_OK ==
 263                    GNUNET_CONTAINER_multihashmap_put (c->requests,
 264                                                       &req->peer.hashPubKey,
 265                                                       car,
 266                                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
 267     GNUNET_CONTAINER_DLL_insert (n->active_client_request_head,
 268                                  n->active_client_request_tail, car);
 269     car->client = c;
 270   }
 271   car->deadline = GNUNET_TIME_absolute_ntoh (req->deadline);
 272   car->priority = ntohl (req->priority);
 273   car->queue_size = ntohl (req->queue_size);
 274   car->msize = ntohs (req->size);
 275   car->smr_id = req->smr_id;
 276   schedule_peer_messages (n);
 277   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 278 }
 279
 280
 281 /**
 282  * Notify client about an existing connection to one of our neighbours.
 283  */
 284 static int
 285 notify_client_about_neighbour (void *cls, const GNUNET_HashCode * key,
 286                                void *value)
 287 {
 288   struct Client *c = cls;
 289   struct Neighbour *n = value;
 290   size_t size;
 291   char buf[GNUNET_SERVER_MAX_MESSAGE_SIZE - 1];
 292   struct GNUNET_TRANSPORT_ATS_Information *ats;
 293   struct ConnectNotifyMessage *cnm;
 294
 295   size =
 296       sizeof (struct ConnectNotifyMessage) +
 297       (n->ats_count) * sizeof (struct GNUNET_TRANSPORT_ATS_Information);
 298   if (size >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
 299   {
 300     GNUNET_break (0);
 301     /* recovery strategy: throw away performance data */
 302     GNUNET_array_grow (n->ats, n->ats_count, 0);
 303     size =
 304         sizeof (struct ConnectNotifyMessage) +
 305         (n->ats_count) * sizeof (struct GNUNET_TRANSPORT_ATS_Information);
 306   }
 307   cnm = (struct ConnectNotifyMessage *) buf;
 308   cnm->header.size = htons (size);
 309   cnm->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_NOTIFY_CONNECT);
 310   cnm->ats_count = htonl (n->ats_count);
 311   ats = &cnm->ats;
 312   memcpy (ats, n->ats,
 313           sizeof (struct GNUNET_TRANSPORT_ATS_Information) * n->ats_count);
 314   ats[n->ats_count].type = htonl (GNUNET_TRANSPORT_ATS_ARRAY_TERMINATOR);
 315   ats[n->ats_count].value = htonl (0);
 316   if (n->status == PEER_STATE_KEY_CONFIRMED)
 317   {
 318 #if DEBUG_CORE_CLIENT
 319     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending `%s' message to client.\n",
 320                 "NOTIFY_CONNECT");
 321 #endif
 322     cnm->peer = n->peer;
 323     send_to_client (c, &cnm->header, GNUNET_NO);
 324   }
 325   return GNUNET_OK;
 326 }
 327
 328
 329
 330 /**
 331  * Handle CORE_INIT request.
 332  */
 333 static void
 334 handle_client_init (void *cls, struct GNUNET_SERVER_Client *client,
 335                     const struct GNUNET_MessageHeader *message)
 336 {
 337   const struct InitMessage *im;
 338   struct InitReplyMessage irm;
 339   struct Client *c;
 340   uint16_t msize;
 341   const uint16_t *types;
 342   uint16_t *wtypes;
 343   unsigned int i;
 344
 345 #if DEBUG_CORE_CLIENT
 346   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 347               "Client connecting to core service with `%s' message\n", "INIT");
 348 #endif
 349   /* check that we don't have an entry already */
 350   c = clients;
 351   while (c != NULL)
 352   {
 353     if (client == c->client_handle)
 354     {
 355       GNUNET_break (0);
 356       GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 357       return;
 358     }
 359     c = c->next;
 360   }
 361   msize = ntohs (message->size);
 362   if (msize < sizeof (struct InitMessage))
 363   {
 364     GNUNET_break (0);
 365     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 366     return;
 367   }
 368   GNUNET_SERVER_notification_context_add (notifier, client);
 369   im = (const struct InitMessage *) message;
 370   types = (const uint16_t *) &im[1];
 371   msize -= sizeof (struct InitMessage);
 372   c = GNUNET_malloc (sizeof (struct Client) + msize);
 373   c->client_handle = client;
 374   c->next = clients;
 375   clients = c;
 376   c->tcnt = msize / sizeof (uint16_t);
 377   c->types = (const uint16_t *) &c[1];
 378   wtypes = (uint16_t *) & c[1];
 379   for (i = 0; i < c->tcnt; i++)
 380   {
 381     wtypes[i] = ntohs (types[i]);
 382     my_type_map[wtypes[i] / 32] |= (1 << (wtypes[i] % 32));
 383   }
 384   if (c->tcnt > 0)
 385     broadcast_my_type_map ();
 386   c->options = ntohl (im->options);
 387 #if DEBUG_CORE_CLIENT
 388   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 389               "Client %p is interested in %u message types\n", c,
 390               (unsigned int) c->tcnt);
 391 #endif
 392   /* send init reply message */
 393   irm.header.size = htons (sizeof (struct InitReplyMessage));
 394   irm.header.type = htons (GNUNET_MESSAGE_TYPE_CORE_INIT_REPLY);
 395   irm.reserved = htonl (0);
 396   memcpy (&irm.publicKey, &my_public_key,
 397           sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded));
 398 #if DEBUG_CORE_CLIENT
 399   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending `%s' message to client.\n",
 400               "INIT_REPLY");
 401 #endif
 402   send_to_client (c, &irm.header, GNUNET_NO);
 403   if (0 != (c->options & GNUNET_CORE_OPTION_SEND_CONNECT))
 404   {
 405     /* notify new client about existing neighbours */
 406     GNUNET_CONTAINER_multihashmap_iterate (neighbours,
 407                                            &notify_client_about_neighbour, c);
 408   }
 409   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 410 }
 411
 412
 413 /**
 414  * Free client request records.
 415  *
 416  * @param cls NULL
 417  * @param key identity of peer for which this is an active request
 418  * @param value the 'struct ClientActiveRequest' to free
 419  * @return GNUNET_YES (continue iteration)
 420  */
 421 static int
 422 destroy_active_client_request (void *cls, const GNUNET_HashCode * key,
 423                                void *value)
 424 {
 425   struct ClientActiveRequest *car = value;
 426   struct Neighbour *n;
 427   struct GNUNET_PeerIdentity peer;
 428
 429   peer.hashPubKey = *key;
 430   n = find_neighbour (&peer);
 431   GNUNET_assert (NULL != n);
 432   GNUNET_CONTAINER_DLL_remove (n->active_client_request_head,
 433                                n->active_client_request_tail, car);
 434   GNUNET_free (car);
 435   return GNUNET_YES;
 436 }
 437
 438
 439 /**
 440  * A client disconnected, clean up.
 441  *
 442  * @param cls closure
 443  * @param client identification of the client
 444  */
 445 static void
 446 handle_client_disconnect (void *cls, struct GNUNET_SERVER_Client *client)
 447 {
 448   struct Client *pos;
 449   struct Client *prev;
 450   unsigned int i;
 451   const uint16_t *wtypes;
 452
 453   if (client == NULL)
 454     return;
 455 #if DEBUG_CORE_CLIENT
 456   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 457               "Client %p has disconnected from core service.\n", client);
 458 #endif
 459   prev = NULL;
 460   pos = clients;
 461   while (pos != NULL)
 462   {
 463     if (client == pos->client_handle)
 464       break;
 465     prev = pos;
 466     pos = pos->next;
 467   }
 468   if (pos == NULL)
 469   {
 470     /* client never sent INIT */
 471     return;
 472   }
 473   if (prev == NULL)
 474     clients = pos->next;
 475   else
 476     prev->next = pos->next;
 477   if (pos->requests != NULL)
 478   {
 479     GNUNET_CONTAINER_multihashmap_iterate (pos->requests,
 480                                            &destroy_active_client_request,
 481                                            NULL);
 482     GNUNET_CONTAINER_multihashmap_destroy (pos->requests);
 483   }
 484   GNUNET_free (pos);
 485
 486   /* rebuild my_type_map */
 487   memset (my_type_map, 0, sizeof (my_type_map));
 488   for (pos = clients; NULL != pos; pos = pos->next)
 489   {
 490     wtypes = (const uint16_t *) &pos[1];
 491     for (i = 0; i < pos->tcnt; i++)
 492       my_type_map[wtypes[i] / 32] |= (1 << (wtypes[i] % 32));
 493   }
 494   broadcast_my_type_map ();
 495 }
 496
 497
 498
 499
 500
 501 /**
 502  * Handle CORE_SEND request.
 503  *
 504  * @param cls unused
 505  * @param client the client issuing the request
 506  * @param message the "struct SendMessage"
 507  */
 508 static void
 509 handle_client_send (void *cls, struct GNUNET_SERVER_Client *client,
 510                     const struct GNUNET_MessageHeader *message)
 511 {
 512   const struct SendMessage *sm;
 513   struct Neighbour *n;
 514   struct MessageEntry *prev;
 515   struct MessageEntry *pos;
 516   struct MessageEntry *e;
 517   struct MessageEntry *min_prio_entry;
 518   struct MessageEntry *min_prio_prev;
 519   unsigned int min_prio;
 520   unsigned int queue_size;
 521   uint16_t msize;
 522
 523   msize = ntohs (message->size);
 524   if (msize <
 525       sizeof (struct SendMessage) + sizeof (struct GNUNET_MessageHeader))
 526   {
 527     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 528                 "msize is %u, should be at least %u (in %s:%d)\n", msize,
 529                 sizeof (struct SendMessage) +
 530                 sizeof (struct GNUNET_MessageHeader), __FILE__, __LINE__);
 531     GNUNET_break (0);
 532     if (client != NULL)
 533       GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 534     return;
 535   }
 536   sm = (const struct SendMessage *) message;
 537   msize -= sizeof (struct SendMessage);
 538   if (0 ==
 539       memcmp (&sm->peer, &my_identity, sizeof (struct GNUNET_PeerIdentity)))
 540   {
 541     /* loopback */
 542     GNUNET_SERVER_mst_receive (mst, &self, (const char *) &sm[1], msize,
 543                                GNUNET_YES, GNUNET_NO);
 544     if (client != NULL)
 545       GNUNET_SERVER_receive_done (client, GNUNET_OK);
 546     return;
 547   }
 548   n = find_neighbour (&sm->peer);
 549   if ((n == NULL) || (GNUNET_YES != n->is_connected) ||
 550       (n->status != PEER_STATE_KEY_CONFIRMED))
 551   {
 552     /* attempt to send message to peer that is not connected anymore
 553      * (can happen due to asynchrony) */
 554     GNUNET_STATISTICS_update (stats,
 555                               gettext_noop
 556                               ("# messages discarded (disconnected)"), 1,
 557                               GNUNET_NO);
 558     if (client != NULL)
 559       GNUNET_SERVER_receive_done (client, GNUNET_OK);
 560     return;
 561   }
 562 #if DEBUG_CORE
 563   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 564               "Core received `%s' request, queueing %u bytes of plaintext data for transmission to `%4s'.\n",
 565               "SEND", (unsigned int) msize, GNUNET_i2s (&sm->peer));
 566 #endif
 567   discard_expired_messages (n);
 568   /* bound queue size */
 569   /* NOTE: this entire block to bound the queue size should be
 570    * obsolete with the new client-request code and the
 571    * 'schedule_peer_messages' mechanism; we still have this code in
 572    * here for now as a sanity check for the new mechanmism;
 573    * ultimately, we should probably simply reject SEND messages that
 574    * are not 'approved' (or provide a new core API for very unreliable
 575    * delivery that always sends with priority 0).  Food for thought. */
 576   min_prio = UINT32_MAX;
 577   min_prio_entry = NULL;
 578   min_prio_prev = NULL;
 579   queue_size = 0;
 580   prev = NULL;
 581   pos = n->messages;
 582   while (pos != NULL)
 583   {
 584     if (pos->priority <= min_prio)
 585     {
 586       min_prio_entry = pos;
 587       min_prio_prev = prev;
 588       min_prio = pos->priority;
 589     }
 590     queue_size++;
 591     prev = pos;
 592     pos = pos->next;
 593   }
 594   if (queue_size >= MAX_PEER_QUEUE_SIZE)
 595   {
 596     /* queue full */
 597     if (ntohl (sm->priority) <= min_prio)
 598     {
 599       /* discard new entry; this should no longer happen! */
 600       GNUNET_break (0);
 601 #if DEBUG_CORE
 602       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 603                   "Queue full (%u/%u), discarding new request (%u bytes of type %u)\n",
 604                   queue_size, (unsigned int) MAX_PEER_QUEUE_SIZE,
 605                   (unsigned int) msize, (unsigned int) ntohs (message->type));
 606 #endif
 607       GNUNET_STATISTICS_update (stats,
 608                                 gettext_noop ("# discarded CORE_SEND requests"),
 609                                 1, GNUNET_NO);
 610
 611       if (client != NULL)
 612         GNUNET_SERVER_receive_done (client, GNUNET_OK);
 613       return;
 614     }
 615     GNUNET_assert (min_prio_entry != NULL);
 616     /* discard "min_prio_entry" */
 617 #if DEBUG_CORE
 618     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 619                 "Queue full, discarding existing older request\n");
 620 #endif
 621     GNUNET_STATISTICS_update (stats,
 622                               gettext_noop
 623                               ("# discarded lower priority CORE_SEND requests"),
 624                               1, GNUNET_NO);
 625     if (min_prio_prev == NULL)
 626       n->messages = min_prio_entry->next;
 627     else
 628       min_prio_prev->next = min_prio_entry->next;
 629     GNUNET_free (min_prio_entry);
 630   }
 631
 632 #if DEBUG_CORE
 633   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 634               "Adding transmission request for `%4s' of size %u to queue\n",
 635               GNUNET_i2s (&sm->peer), (unsigned int) msize);
 636 #endif
 637   GNUNET_break (0 == ntohl (sm->reserved));
 638   e = GNUNET_malloc (sizeof (struct MessageEntry) + msize);
 639   e->deadline = GNUNET_TIME_absolute_ntoh (sm->deadline);
 640   e->priority = ntohl (sm->priority);
 641   e->size = msize;
 642   if (GNUNET_YES != (int) ntohl (sm->cork))
 643     e->got_slack = GNUNET_YES;
 644   memcpy (&e[1], &sm[1], msize);
 645
 646   /* insert, keep list sorted by deadline */
 647   prev = NULL;
 648   pos = n->messages;
 649   while ((pos != NULL) && (pos->deadline.abs_value < e->deadline.abs_value))
 650   {
 651     prev = pos;
 652     pos = pos->next;
 653   }
 654   if (prev == NULL)
 655     n->messages = e;
 656   else
 657     prev->next = e;
 658   e->next = pos;
 659
 660   /* consider scheduling now */
 661   process_plaintext_neighbour_queue (n);
 662   if (client != NULL)
 663     GNUNET_SERVER_receive_done (client, GNUNET_OK);
 664 }
 665
 666
 667 /**
 668  * Helper function for handle_client_iterate_peers.
 669  *
 670  * @param cls the 'struct GNUNET_SERVER_TransmitContext' to queue replies
 671  * @param key identity of the connected peer
 672  * @param value the 'struct Neighbour' for the peer
 673  * @return GNUNET_OK (continue to iterate)
 674  */
 675 static int
 676 queue_connect_message (void *cls, const GNUNET_HashCode * key, void *value)
 677 {
 678   struct GNUNET_SERVER_TransmitContext *tc = cls;
 679   struct Neighbour *n = value;
 680   char buf[GNUNET_SERVER_MAX_MESSAGE_SIZE - 1];
 681   struct GNUNET_TRANSPORT_ATS_Information *ats;
 682   size_t size;
 683   struct ConnectNotifyMessage *cnm;
 684
 685   cnm = (struct ConnectNotifyMessage *) buf;
 686   if (n->status != PEER_STATE_KEY_CONFIRMED)
 687     return GNUNET_OK;
 688   size =
 689       sizeof (struct ConnectNotifyMessage) +
 690       (n->ats_count) * sizeof (struct GNUNET_TRANSPORT_ATS_Information);
 691   if (size >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
 692   {
 693     GNUNET_break (0);
 694     /* recovery strategy: throw away performance data */
 695     GNUNET_array_grow (n->ats, n->ats_count, 0);
 696     size =
 697         sizeof (struct PeerStatusNotifyMessage) +
 698         n->ats_count * sizeof (struct GNUNET_TRANSPORT_ATS_Information);
 699   }
 700   cnm = (struct ConnectNotifyMessage *) buf;
 701   cnm->header.size = htons (size);
 702   cnm->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_NOTIFY_CONNECT);
 703   cnm->ats_count = htonl (n->ats_count);
 704   ats = &cnm->ats;
 705   memcpy (ats, n->ats,
 706           n->ats_count * sizeof (struct GNUNET_TRANSPORT_ATS_Information));
 707   ats[n->ats_count].type = htonl (GNUNET_TRANSPORT_ATS_ARRAY_TERMINATOR);
 708   ats[n->ats_count].value = htonl (0);
 709 #if DEBUG_CORE_CLIENT
 710   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending `%s' message to client.\n",
 711               "NOTIFY_CONNECT");
 712 #endif
 713   cnm->peer = n->peer;
 714   GNUNET_SERVER_transmit_context_append_message (tc, &cnm->header);
 715   return GNUNET_OK;
 716 }
 717
 718
 719 /**
 720  * Handle CORE_ITERATE_PEERS request.
 721  *
 722  * @param cls unused
 723  * @param client client sending the iteration request
 724  * @param message iteration request message
 725  */
 726 static void
 727 handle_client_iterate_peers (void *cls, struct GNUNET_SERVER_Client *client,
 728                              const struct GNUNET_MessageHeader *message)
 729 {
 730   struct GNUNET_MessageHeader done_msg;
 731   struct GNUNET_SERVER_TransmitContext *tc;
 732   int msize;
 733
 734   /* notify new client about existing neighbours */
 735
 736   msize = ntohs (message->size);
 737   tc = GNUNET_SERVER_transmit_context_create (client);
 738   if (msize == sizeof (struct GNUNET_MessageHeader))
 739     GNUNET_CONTAINER_multihashmap_iterate (neighbours, &queue_connect_message,
 740                                            tc);
 741   else
 742     GNUNET_break (0);
 743
 744   done_msg.size = htons (sizeof (struct GNUNET_MessageHeader));
 745   done_msg.type = htons (GNUNET_MESSAGE_TYPE_CORE_ITERATE_PEERS_END);
 746   GNUNET_SERVER_transmit_context_append_message (tc, &done_msg);
 747   GNUNET_SERVER_transmit_context_run (tc, GNUNET_TIME_UNIT_FOREVER_REL);
 748 }
 749
 750
 751 /**
 752  * Handle CORE_PEER_CONNECTED request.  Notify client about existing neighbours.
 753  *
 754  * @param cls unused
 755  * @param client client sending the iteration request
 756  * @param message iteration request message
 757  */
 758 static void
 759 handle_client_have_peer (void *cls, struct GNUNET_SERVER_Client *client,
 760                          const struct GNUNET_MessageHeader *message)
 761 {
 762   struct GNUNET_MessageHeader done_msg;
 763   struct GNUNET_SERVER_TransmitContext *tc;
 764   struct GNUNET_PeerIdentity *peer;
 765
 766   tc = GNUNET_SERVER_transmit_context_create (client);
 767   peer = (struct GNUNET_PeerIdentity *) &message[1];
 768   GNUNET_CONTAINER_multihashmap_get_multiple (neighbours, &peer->hashPubKey,
 769                                               &queue_connect_message, tc);
 770   done_msg.size = htons (sizeof (struct GNUNET_MessageHeader));
 771   done_msg.type = htons (GNUNET_MESSAGE_TYPE_CORE_ITERATE_PEERS_END);
 772   GNUNET_SERVER_transmit_context_append_message (tc, &done_msg);
 773   GNUNET_SERVER_transmit_context_run (tc, GNUNET_TIME_UNIT_FOREVER_REL);
 774 }
 775
 776
 777 /**
 778  * Handle REQUEST_INFO request.
 779  *
 780  * @param cls unused
 781  * @param client client sending the request
 782  * @param message iteration request message
 783  */
 784 static void
 785 handle_client_request_info (void *cls, struct GNUNET_SERVER_Client *client,
 786                             const struct GNUNET_MessageHeader *message)
 787 {
 788   const struct RequestInfoMessage *rcm;
 789   struct Client *pos;
 790   struct Neighbour *n;
 791   struct ConfigurationInfoMessage cim;
 792   int32_t want_reserv;
 793   int32_t got_reserv;
 794   unsigned long long old_preference;
 795   struct GNUNET_TIME_Relative rdelay;
 796
 797   rdelay = GNUNET_TIME_relative_get_zero ();
 798 #if DEBUG_CORE_CLIENT
 799   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Core service receives `%s' request.\n",
 800               "REQUEST_INFO");
 801 #endif
 802   pos = clients;
 803   while (pos != NULL)
 804   {
 805     if (client == pos->client_handle)
 806       break;
 807     pos = pos->next;
 808   }
 809   if (pos == NULL)
 810   {
 811     GNUNET_break (0);
 812     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
 813     return;
 814   }
 815
 816   rcm = (const struct RequestInfoMessage *) message;
 817   n = find_neighbour (&rcm->peer);
 818   memset (&cim, 0, sizeof (cim));
 819   if ((n != NULL) && (GNUNET_YES == n->is_connected))
 820   {
 821     want_reserv = ntohl (rcm->reserve_inbound);
 822     if (n->bw_out_internal_limit.value__ != rcm->limit_outbound.value__)
 823     {
 824       n->bw_out_internal_limit = rcm->limit_outbound;
 825       if (n->bw_out.value__ !=
 826           GNUNET_BANDWIDTH_value_min (n->bw_out_internal_limit,
 827                                       n->bw_out_external_limit).value__)
 828       {
 829         n->bw_out =
 830             GNUNET_BANDWIDTH_value_min (n->bw_out_internal_limit,
 831                                         n->bw_out_external_limit);
 832         GNUNET_BANDWIDTH_tracker_update_quota (&n->available_recv_window,
 833                                                n->bw_out);
 834         GNUNET_TRANSPORT_set_quota (transport, &n->peer, n->bw_in, n->bw_out);
 835         handle_peer_status_change (n);
 836       }
 837     }
 838     if (want_reserv < 0)
 839     {
 840       got_reserv = want_reserv;
 841     }
 842     else if (want_reserv > 0)
 843     {
 844       rdelay =
 845           GNUNET_BANDWIDTH_tracker_get_delay (&n->available_recv_window,
 846                                               want_reserv);
 847       if (rdelay.rel_value == 0)
 848         got_reserv = want_reserv;
 849       else
 850         got_reserv = 0;         /* all or nothing */
 851     }
 852     else
 853       got_reserv = 0;
 854     GNUNET_BANDWIDTH_tracker_consume (&n->available_recv_window, got_reserv);
 855     old_preference = n->current_preference;
 856     n->current_preference += GNUNET_ntohll (rcm->preference_change);
 857     if (old_preference > n->current_preference)
 858     {
 859       /* overflow; cap at maximum value */
 860       n->current_preference = ULLONG_MAX;
 861     }
 862     update_preference_sum (n->current_preference - old_preference);
 863 #if DEBUG_CORE_QUOTA
 864     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 865                 "Received reservation request for %d bytes for peer `%4s', reserved %d bytes, suggesting delay of %llu ms\n",
 866                 (int) want_reserv, GNUNET_i2s (&rcm->peer), (int) got_reserv,
 867                 (unsigned long long) rdelay.rel_value);
 868 #endif
 869     cim.reserved_amount = htonl (got_reserv);
 870     cim.reserve_delay = GNUNET_TIME_relative_hton (rdelay);
 871     cim.bw_out = n->bw_out;
 872     cim.preference = n->current_preference;
 873   }
 874   else
 875   {
 876     /* Technically, this COULD happen (due to asynchronous behavior),
 877      * but it should be rare, so we should generate an info event
 878      * to help diagnosis of serious errors that might be masked by this */
 879     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 880                 _
 881                 ("Client asked for preference change with peer `%s', which is not connected!\n"),
 882                 GNUNET_i2s (&rcm->peer));
 883     GNUNET_SERVER_receive_done (client, GNUNET_OK);
 884     return;
 885   }
 886   cim.header.size = htons (sizeof (struct ConfigurationInfoMessage));
 887   cim.header.type = htons (GNUNET_MESSAGE_TYPE_CORE_CONFIGURATION_INFO);
 888   cim.peer = rcm->peer;
 889   cim.rim_id = rcm->rim_id;
 890 #if DEBUG_CORE_CLIENT
 891   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending `%s' message to client.\n",
 892               "CONFIGURATION_INFO");
 893 #endif
 894   send_to_client (pos, &cim.header, GNUNET_NO);
 895   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 896 }
 897
 898
 899
 900
 901 /**
 902  * Send a P2P message to a client.
 903  *
 904  * @param sender who sent us the message?
 905  * @param client who should we give the message to?
 906  * @param m contains the message to transmit
 907  * @param msize number of bytes in buf to transmit
 908  */
 909 static void
 910 send_p2p_message_to_client (struct Neighbour *sender, struct Client *client,
 911                             const void *m, size_t msize)
 912 {
 913   size_t size =
 914       msize + sizeof (struct NotifyTrafficMessage) +
 915       (sender->ats_count) * sizeof (struct GNUNET_TRANSPORT_ATS_Information);
 916   char buf[size];
 917   struct NotifyTrafficMessage *ntm;
 918   struct GNUNET_TRANSPORT_ATS_Information *ats;
 919
 920   GNUNET_assert (GNUNET_YES == sender->is_connected);
 921   GNUNET_break (sender->status == PEER_STATE_KEY_CONFIRMED);
 922   if (size >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
 923   {
 924     GNUNET_break (0);
 925     /* recovery strategy: throw performance data away... */
 926     GNUNET_array_grow (sender->ats, sender->ats_count, 0);
 927     size =
 928         msize + sizeof (struct NotifyTrafficMessage) +
 929         (sender->ats_count) * sizeof (struct GNUNET_TRANSPORT_ATS_Information);
 930   }
 931 #if DEBUG_CORE
 932   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 933               "Core service passes message from `%4s' of type %u to client.\n",
 934               GNUNET_i2s (&sender->peer),
 935               (unsigned int)
 936               ntohs (((const struct GNUNET_MessageHeader *) m)->type));
 937 #endif
 938   ntm = (struct NotifyTrafficMessage *) buf;
 939   ntm->header.size = htons (size);
 940   ntm->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_NOTIFY_INBOUND);
 941   ntm->ats_count = htonl (sender->ats_count);
 942   ntm->peer = sender->peer;
 943   ats = &ntm->ats;
 944   memcpy (ats, sender->ats,
 945           sizeof (struct GNUNET_TRANSPORT_ATS_Information) * sender->ats_count);
 946   ats[sender->ats_count].type = htonl (GNUNET_TRANSPORT_ATS_ARRAY_TERMINATOR);
 947   ats[sender->ats_count].value = htonl (0);
 948   memcpy (&ats[sender->ats_count + 1], m, msize);
 949   send_to_client (client, &ntm->header, GNUNET_YES);
 950 }
 951
 952
 953
 954
 955 /**
 956  * Deliver P2P message to interested clients.
 957  *
 958  * @param sender peer who sent us the message
 959  * @param m the message
 960  */
 961 void
 962 GSC_CLIENTS_deliver_message (const struct GNUNET_PeerIdentity *sender,
 963                              const struct GNUNET_MessageHeader *m)
 964 {
 965   struct Neighbour *sender = client;
 966   size_t msize = ntohs (m->size);
 967   char buf[256];
 968   struct Client *cpos;
 969   uint16_t type;
 970   unsigned int tpos;
 971   int deliver_full;
 972   int dropped;
 973
 974   GNUNET_break (sender->status == PEER_STATE_KEY_CONFIRMED);
 975   type = ntohs (m->type);
 976 #if DEBUG_CORE > 1
 977   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 978               "Received encapsulated message of type %u and size %u from `%4s'\n",
 979               (unsigned int) type, ntohs (m->size), GNUNET_i2s (&sender->peer));
 980 #endif
 981   GNUNET_snprintf (buf, sizeof (buf),
 982                    gettext_noop ("# bytes of messages of type %u received"),
 983                    (unsigned int) type);
 984   GNUNET_STATISTICS_update (stats, buf, msize, GNUNET_NO);
 985   if ((GNUNET_MESSAGE_TYPE_CORE_BINARY_TYPE_MAP == type) ||
 986       (GNUNET_MESSAGE_TYPE_CORE_COMPRESSED_TYPE_MAP == type))
 987   {
 988     /* FIXME: update message type map for 'Neighbour' */
 989     return;
 990   }
 991   dropped = GNUNET_YES;
 992   cpos = clients;
 993   while (cpos != NULL)
 994   {
 995     deliver_full = GNUNET_NO;
 996     if (0 != (cpos->options & GNUNET_CORE_OPTION_SEND_FULL_INBOUND))
 997       deliver_full = GNUNET_YES;
 998     else
 999     {
1000       for (tpos = 0; tpos < cpos->tcnt; tpos++)
1001       {
1002         if (type != cpos->types[tpos])
1003           continue;
1004         deliver_full = GNUNET_YES;
1005         break;
1006       }
1007     }
1008     if (GNUNET_YES == deliver_full)
1009     {
1010       send_p2p_message_to_client (sender, cpos, m, msize);
1011       dropped = GNUNET_NO;
1012     }
1013     else if (cpos->options & GNUNET_CORE_OPTION_SEND_HDR_INBOUND)
1014     {
1015       send_p2p_message_to_client (sender, cpos, m,
1016                                   sizeof (struct GNUNET_MessageHeader));
1017     }
1018     cpos = cpos->next;
1019   }
1020   if (dropped == GNUNET_YES)
1021   {
1022 #if DEBUG_CORE
1023     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1024                 "Message of type %u from `%4s' not delivered to any client.\n",
1025                 (unsigned int) type, GNUNET_i2s (&sender->peer));
1026 #endif
1027     GNUNET_STATISTICS_update (stats,
1028                               gettext_noop
1029                               ("# messages not delivered to any client"), 1,
1030                               GNUNET_NO);
1031   }
1032 }
1033
1034
1035
1036 void
1037 GSC_CLIENTS_init (struct GNUNET_SERVER_Handle *server)
1038 {
1039   static const struct GNUNET_SERVER_MessageHandler handlers[] = {
1040     {&handle_client_init, NULL,
1041      GNUNET_MESSAGE_TYPE_CORE_INIT, 0},
1042     {&handle_client_iterate_peers, NULL,
1043      GNUNET_MESSAGE_TYPE_CORE_ITERATE_PEERS,
1044      sizeof (struct GNUNET_MessageHeader)},
1045     {&handle_client_have_peer, NULL,
1046      GNUNET_MESSAGE_TYPE_CORE_PEER_CONNECTED,
1047      sizeof (struct GNUNET_MessageHeader) +
1048      sizeof (struct GNUNET_PeerIdentity)},
1049     {&handle_client_request_info, NULL,
1050      GNUNET_MESSAGE_TYPE_CORE_REQUEST_INFO,
1051      sizeof (struct RequestInfoMessage)},
1052     {&handle_client_send_request, NULL,
1053      GNUNET_MESSAGE_TYPE_CORE_SEND_REQUEST,
1054      sizeof (struct SendMessageRequest)},
1055     {&handle_client_send, NULL,
1056      GNUNET_MESSAGE_TYPE_CORE_SEND, 0},
1057     {NULL, NULL, 0, 0}
1058   };
1059
1060   /* setup notification */
1061   notifier =
1062       GNUNET_SERVER_notification_context_create (server, MAX_NOTIFY_QUEUE);
1063   GNUNET_SERVER_disconnect_notify (server, &handle_client_disconnect, NULL);
1064   GNUNET_SERVER_add_handlers (server, handlers);
1065   mst = GNUNET_SERVER_mst_create (&deliver_message, NULL);
1066 }
1067
1068
1069 void
1070 GSC_CLIENTS_done ()
1071 {
1072   struct Client *c;
1073
1074   while (NULL != (c = clients))
1075     handle_client_disconnect (NULL, c->client_handle);
1076   GNUNET_SERVER_notification_context_destroy (notifier);
1077   notifier = NULL;
1078   if (mst != NULL)
1079     {
1080       GNUNET_SERVER_mst_destroy (mst);
1081       mst = NULL;
1082     }
1083
1084 }