src/core/gnunet-service-core_sessions.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2009-2014, 2016 GNUnet e.V.
   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., 51 Franklin Street, Fifth Floor,
  18      Boston, MA 02110-1301, USA.
  19 */
  20
  21 /**
  22  * @file core/gnunet-service-core_sessions.c
  23  * @brief code for managing of 'encrypted' sessions (key exchange done)
  24  * @author Christian Grothoff
  25  */
  26 #include "platform.h"
  27 #include "gnunet-service-core.h"
  28 #include "gnunet-service-core_kx.h"
  29 #include "gnunet-service-core_typemap.h"
  30 #include "gnunet-service-core_sessions.h"
  31 #include "gnunet-service-core_clients.h"
  32 #include "gnunet_constants.h"
  33 #include "core.h"
  34
  35
  36 /**
  37  * How many encrypted messages do we queue at most?
  38  * Needed to bound memory consumption.
  39  */
  40 #define MAX_ENCRYPTED_MESSAGE_QUEUE_SIZE 4
  41
  42
  43 /**
  44  * Message ready for encryption.  This struct is followed by the
  45  * actual content of the message.
  46  */
  47 struct SessionMessageEntry
  48 {
  49
  50   /**
  51    * We keep messages in a doubly linked list.
  52    */
  53   struct SessionMessageEntry *next;
  54
  55   /**
  56    * We keep messages in a doubly linked list.
  57    */
  58   struct SessionMessageEntry *prev;
  59
  60   /**
  61    * Deadline for transmission, 1s after we received it (if we
  62    * are not corking), otherwise "now".  Note that this message
  63    * does NOT expire past its deadline.
  64    */
  65   struct GNUNET_TIME_Absolute deadline;
  66
  67   /**
  68    * How long is the message? (number of bytes following the `struct
  69    * MessageEntry`, but not including the size of `struct
  70    * MessageEntry` itself!)
  71    */
  72   size_t size;
  73
  74   /**
  75    * How important is this message.
  76    */
  77   enum GNUNET_CORE_Priority priority;
  78
  79 };
  80
  81
  82 /**
  83  * Data kept per session.
  84  */
  85 struct Session
  86 {
  87   /**
  88    * Identity of the other peer.
  89    */
  90   const struct GNUNET_PeerIdentity *peer;
  91
  92   /**
  93    * Key exchange state for this peer.
  94    */
  95   struct GSC_KeyExchangeInfo *kx;
  96
  97   /**
  98    * Head of list of requests from clients for transmission to
  99    * this peer.
 100    */
 101   struct GSC_ClientActiveRequest *active_client_request_head;
 102
 103   /**
 104    * Tail of list of requests from clients for transmission to
 105    * this peer.
 106    */
 107   struct GSC_ClientActiveRequest *active_client_request_tail;
 108
 109   /**
 110    * Head of list of messages ready for encryption.
 111    */
 112   struct SessionMessageEntry *sme_head;
 113
 114   /**
 115    * Tail of list of messages ready for encryption.
 116    */
 117   struct SessionMessageEntry *sme_tail;
 118
 119   /**
 120    * Current type map for this peer.
 121    */
 122   struct GSC_TypeMap *tmap;
 123
 124   /**
 125    * Task to transmit corked messages with a delay.
 126    */
 127   struct GNUNET_SCHEDULER_Task *cork_task;
 128
 129   /**
 130    * Task to transmit our type map.
 131    */
 132   struct GNUNET_SCHEDULER_Task *typemap_task;
 133
 134   /**
 135    * Retransmission delay we currently use for the typemap
 136    * transmissions (if not confirmed).
 137    */
 138   struct GNUNET_TIME_Relative typemap_delay;
 139
 140   /**
 141    * Is the neighbour queue empty and thus ready for us
 142    * to transmit an encrypted message?
 143    */
 144   int ready_to_transmit;
 145
 146   /**
 147    * Is this the first time we're sending the typemap? If so,
 148    * we want to send it a bit faster the second time.  0 if
 149    * we are sending for the first time, 1 if not.
 150    */
 151   int first_typemap;
 152 };
 153
 154
 155 GNUNET_NETWORK_STRUCT_BEGIN
 156
 157 /**
 158  * Message sent to confirm that a typemap was received.
 159  */
 160 struct TypeMapConfirmationMessage
 161 {
 162
 163   /**
 164    * Header with type #GNUNET_MESSAGE_TYPE_CORE_CONFIRM_TYPE_MAP.
 165    */
 166   struct GNUNET_MessageHeader header;
 167
 168   /**
 169    * Reserved, always zero.
 170    */
 171   uint32_t reserved GNUNET_PACKED;
 172
 173   /**
 174    * Hash of the (decompressed) type map that was received.
 175    */
 176   struct GNUNET_HashCode tm_hash;
 177
 178 };
 179
 180 GNUNET_NETWORK_STRUCT_END
 181
 182
 183 /**
 184  * Map of peer identities to `struct Session`.
 185  */
 186 static struct GNUNET_CONTAINER_MultiPeerMap *sessions;
 187
 188
 189 /**
 190  * Find the session for the given peer.
 191  *
 192  * @param peer identity of the peer
 193  * @return NULL if we are not connected, otherwise the
 194  *         session handle
 195  */
 196 static struct Session *
 197 find_session (const struct GNUNET_PeerIdentity *peer)
 198 {
 199   if (NULL == sessions)
 200     return NULL;
 201   return GNUNET_CONTAINER_multipeermap_get (sessions,
 202                                             peer);
 203 }
 204
 205
 206 /**
 207  * End the session with the given peer (we are no longer
 208  * connected).
 209  *
 210  * @param pid identity of peer to kill session with
 211  */
 212 void
 213 GSC_SESSIONS_end (const struct GNUNET_PeerIdentity *pid)
 214 {
 215   struct Session *session;
 216   struct GSC_ClientActiveRequest *car;
 217   struct SessionMessageEntry *sme;
 218
 219   session = find_session (pid);
 220   if (NULL == session)
 221     return;
 222   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 223               "Destroying session for peer `%s'\n",
 224               GNUNET_i2s (session->peer));
 225   if (NULL != session->cork_task)
 226   {
 227     GNUNET_SCHEDULER_cancel (session->cork_task);
 228     session->cork_task = NULL;
 229   }
 230   while (NULL != (car = session->active_client_request_head))
 231   {
 232     GNUNET_CONTAINER_DLL_remove (session->active_client_request_head,
 233                                  session->active_client_request_tail, car);
 234     GSC_CLIENTS_reject_request (car,
 235                                 GNUNET_NO);
 236   }
 237   while (NULL != (sme = session->sme_head))
 238   {
 239     GNUNET_CONTAINER_DLL_remove (session->sme_head,
 240                                  session->sme_tail,
 241                                  sme);
 242     GNUNET_free (sme);
 243   }
 244   if (NULL != session->typemap_task)
 245   {
 246     GNUNET_SCHEDULER_cancel (session->typemap_task);
 247     session->typemap_task = NULL;
 248   }
 249   GSC_CLIENTS_notify_clients_about_neighbour (session->peer,
 250                                               session->tmap,
 251                                               NULL);
 252   GNUNET_assert (GNUNET_YES ==
 253                  GNUNET_CONTAINER_multipeermap_remove (sessions,
 254                                                        session->peer,
 255                                                        session));
 256   GNUNET_STATISTICS_set (GSC_stats,
 257                          gettext_noop ("# peers connected"),
 258                          GNUNET_CONTAINER_multipeermap_size (sessions),
 259                          GNUNET_NO);
 260   GSC_TYPEMAP_destroy (session->tmap);
 261   session->tmap = NULL;
 262   GNUNET_free (session);
 263 }
 264
 265
 266 /**
 267  * Transmit our current typemap message to the other peer.
 268  * (Done periodically until the typemap is confirmed).
 269  *
 270  * @param cls the `struct Session *`
 271  */
 272 static void
 273 transmit_typemap_task (void *cls)
 274 {
 275   struct Session *session = cls;
 276   struct GNUNET_MessageHeader *hdr;
 277   struct GNUNET_TIME_Relative delay;
 278
 279   session->typemap_delay = GNUNET_TIME_STD_BACKOFF (session->typemap_delay);
 280   delay = session->typemap_delay;
 281   /* randomize a bit to avoid spont. sync */
 282   delay.rel_value_us +=
 283       GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, 1000 * 1000);
 284   session->typemap_task =
 285       GNUNET_SCHEDULER_add_delayed (delay,
 286                                     &transmit_typemap_task, session);
 287   GNUNET_STATISTICS_update (GSC_stats,
 288                             gettext_noop ("# type map refreshes sent"),
 289                             1,
 290                             GNUNET_NO);
 291   hdr = GSC_TYPEMAP_compute_type_map_message ();
 292   GSC_KX_encrypt_and_transmit (session->kx,
 293                                hdr,
 294                                ntohs (hdr->size));
 295   GNUNET_free (hdr);
 296 }
 297
 298
 299 /**
 300  * Restart the typemap task for the given session.
 301  *
 302  * @param session session to restart typemap transmission for
 303  */
 304 static void
 305 start_typemap_task (struct Session *session)
 306 {
 307   if (NULL != session->typemap_task)
 308     GNUNET_SCHEDULER_cancel (session->typemap_task);
 309   session->typemap_delay = GNUNET_TIME_UNIT_SECONDS;
 310   session->typemap_task =
 311     GNUNET_SCHEDULER_add_delayed (session->typemap_delay,
 312                                   &transmit_typemap_task,
 313                                   session);
 314 }
 315
 316
 317 /**
 318  * Create a session, a key exchange was just completed.
 319  *
 320  * @param peer peer that is now connected
 321  * @param kx key exchange that completed
 322  */
 323 void
 324 GSC_SESSIONS_create (const struct GNUNET_PeerIdentity *peer,
 325                      struct GSC_KeyExchangeInfo *kx)
 326 {
 327   struct Session *session;
 328
 329   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 330               "Creating session for peer `%4s'\n",
 331               GNUNET_i2s (peer));
 332   session = GNUNET_new (struct Session);
 333   session->tmap = GSC_TYPEMAP_create ();
 334   session->peer = peer;
 335   session->kx = kx;
 336   GNUNET_assert (GNUNET_OK ==
 337                  GNUNET_CONTAINER_multipeermap_put (sessions,
 338                                                     session->peer,
 339                                                     session,
 340                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 341   GNUNET_STATISTICS_set (GSC_stats,
 342                          gettext_noop ("# peers connected"),
 343                          GNUNET_CONTAINER_multipeermap_size (sessions),
 344                          GNUNET_NO);
 345   GSC_CLIENTS_notify_clients_about_neighbour (peer,
 346                                               NULL,
 347                                               session->tmap);
 348   start_typemap_task (session);
 349 }
 350
 351
 352 /**
 353  * The other peer has indicated that he 'lost' the session
 354  * (KX down), reinitialize the session on our end, in particular
 355  * this means to restart the typemap transmission.
 356  *
 357  * @param peer peer that is now connected
 358  */
 359 void
 360 GSC_SESSIONS_reinit (const struct GNUNET_PeerIdentity *peer)
 361 {
 362   struct Session *session;
 363
 364   session = find_session (peer);
 365   if (NULL == session)
 366   {
 367     /* KX/session is new for both sides; thus no need to restart what
 368        has not yet begun */
 369     return;
 370   }
 371   start_typemap_task (session);
 372 }
 373
 374
 375 /**
 376  * The other peer has confirmed receiving our type map,
 377  * check if it is current and if so, stop retransmitting it.
 378  *
 379  * @param peer peer that confirmed the type map
 380  * @param msg confirmation message we received
 381  */
 382 void
 383 GSC_SESSIONS_confirm_typemap (const struct GNUNET_PeerIdentity *peer,
 384                               const struct GNUNET_MessageHeader *msg)
 385 {
 386   const struct TypeMapConfirmationMessage *cmsg;
 387   struct Session *session;
 388
 389   session = find_session (peer);
 390   if (NULL == session)
 391   {
 392     GNUNET_break (0);
 393     return;
 394   }
 395   if (ntohs (msg->size) != sizeof (struct TypeMapConfirmationMessage))
 396   {
 397     GNUNET_break_op (0);
 398     return;
 399   }
 400   cmsg = (const struct TypeMapConfirmationMessage *) msg;
 401   if (GNUNET_YES !=
 402       GSC_TYPEMAP_check_hash (&cmsg->tm_hash))
 403   {
 404     /* our typemap has changed in the meantime, do not
 405        accept confirmation */
 406     GNUNET_STATISTICS_update (GSC_stats,
 407                               gettext_noop
 408                               ("# outdated typemap confirmations received"),
 409                               1, GNUNET_NO);
 410     return;
 411   }
 412   if (NULL != session->typemap_task)
 413   {
 414     GNUNET_SCHEDULER_cancel (session->typemap_task);
 415     session->typemap_task = NULL;
 416   }
 417   GNUNET_STATISTICS_update (GSC_stats,
 418                             gettext_noop
 419                             ("# valid typemap confirmations received"),
 420                             1, GNUNET_NO);
 421 }
 422
 423
 424 /**
 425  * Notify the given client about the session (client is new).
 426  *
 427  * @param cls the `struct GSC_Client`
 428  * @param key peer identity
 429  * @param value the `struct Session`
 430  * @return #GNUNET_OK (continue to iterate)
 431  */
 432 static int
 433 notify_client_about_session (void *cls,
 434                              const struct GNUNET_PeerIdentity *key,
 435                              void *value)
 436 {
 437   struct GSC_Client *client = cls;
 438   struct Session *session = value;
 439
 440   GSC_CLIENTS_notify_client_about_neighbour (client,
 441                                              session->peer,
 442                                              NULL,      /* old TMAP: none */
 443                                              session->tmap);
 444   return GNUNET_OK;
 445 }
 446
 447
 448 /**
 449  * We have a new client, notify it about all current sessions.
 450  *
 451  * @param client the new client
 452  */
 453 void
 454 GSC_SESSIONS_notify_client_about_sessions (struct GSC_Client *client)
 455 {
 456   /* notify new client about existing sessions */
 457   GNUNET_CONTAINER_multipeermap_iterate (sessions,
 458                                          &notify_client_about_session,
 459                                          client);
 460 }
 461
 462
 463 /**
 464  * Try to perform a transmission on the given session.  Will solicit
 465  * additional messages if the 'sme' queue is not full enough.
 466  *
 467  * @param session session to transmit messages from
 468  */
 469 static void
 470 try_transmission (struct Session *session);
 471
 472
 473 /**
 474  * Queue a request from a client for transmission to a particular peer.
 475  *
 476  * @param car request to queue; this handle is then shared between
 477  *         the caller (CLIENTS subsystem) and SESSIONS and must not
 478  *         be released by either until either #GSC_SESSIONS_dequeue(),
 479  *         #GSC_SESSIONS_transmit() or #GSC_CLIENTS_failed()
 480  *         have been invoked on it
 481  */
 482 void
 483 GSC_SESSIONS_queue_request (struct GSC_ClientActiveRequest *car)
 484 {
 485   struct Session *session;
 486
 487   session = find_session (&car->target);
 488   if (NULL == session)
 489   {
 490     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 491                 "Dropped client request for transmission (am disconnected)\n");
 492     GNUNET_break (0);           /* should have been rejected earlier */
 493     GSC_CLIENTS_reject_request (car,
 494                                 GNUNET_NO);
 495     return;
 496   }
 497   if (car->msize > GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE)
 498   {
 499     GNUNET_break (0);
 500     GSC_CLIENTS_reject_request (car,
 501                                 GNUNET_YES);
 502     return;
 503   }
 504   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 505               "Received client transmission request. queueing\n");
 506   GNUNET_CONTAINER_DLL_insert (session->active_client_request_head,
 507                                session->active_client_request_tail,
 508                                car);
 509   try_transmission (session);
 510 }
 511
 512
 513 /**
 514  * Dequeue a request from a client from transmission to a particular peer.
 515  *
 516  * @param car request to dequeue; this handle will then be 'owned' by
 517  *        the caller (CLIENTS sysbsystem)
 518  */
 519 void
 520 GSC_SESSIONS_dequeue_request (struct GSC_ClientActiveRequest *car)
 521 {
 522   struct Session *session;
 523
 524   if (0 ==
 525       memcmp (&car->target,
 526               &GSC_my_identity,
 527               sizeof (struct GNUNET_PeerIdentity)))
 528     return;
 529   session = find_session (&car->target);
 530   GNUNET_assert (NULL != session);
 531   GNUNET_CONTAINER_DLL_remove (session->active_client_request_head,
 532                                session->active_client_request_tail,
 533                                car);
 534   /* dequeueing of 'high' priority messages may unblock
 535      transmission for lower-priority messages, so we also
 536      need to try in this case. */
 537   try_transmission (session);
 538 }
 539
 540
 541 /**
 542  * Solicit messages for transmission, starting with those of the highest
 543  * priority.
 544  *
 545  * @param session session to solict messages for
 546  * @param msize how many bytes do we have already
 547  */
 548 static void
 549 solicit_messages (struct Session *session,
 550                   size_t msize)
 551 {
 552   struct GSC_ClientActiveRequest *car;
 553   struct GSC_ClientActiveRequest *nxt;
 554   size_t so_size;
 555   enum GNUNET_CORE_Priority pmax;
 556
 557   so_size = msize;
 558   pmax = GNUNET_CORE_PRIO_BACKGROUND;
 559   for (car = session->active_client_request_head; NULL != car; car = car->next)
 560   {
 561     if (GNUNET_YES == car->was_solicited)
 562       continue;
 563     pmax = GNUNET_MAX (pmax, car->priority);
 564   }
 565   nxt = session->active_client_request_head;
 566   while (NULL != (car = nxt))
 567   {
 568     nxt = car->next;
 569     if (car->priority < pmax)
 570       continue;
 571     if (so_size + car->msize > GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE)
 572       break;
 573     so_size += car->msize;
 574     if (GNUNET_YES == car->was_solicited)
 575       continue;
 576     car->was_solicited = GNUNET_YES;
 577     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 578                 "Soliciting message with priority %u\n",
 579                 car->priority);
 580     GSC_CLIENTS_solicit_request (car);
 581     /* The above call may *dequeue* requests and thereby
 582        clobber 'nxt'. Hence we need to restart from the
 583        head of the list. */
 584     nxt = session->active_client_request_head;
 585     so_size = msize;
 586   }
 587 }
 588
 589
 590 /**
 591  * Some messages were delayed (corked), but the timeout has now expired.
 592  * Send them now.
 593  *
 594  * @param cls `struct Session` with the messages to transmit now
 595  */
 596 static void
 597 pop_cork_task (void *cls)
 598 {
 599   struct Session *session = cls;
 600
 601   session->cork_task = NULL;
 602   try_transmission (session);
 603 }
 604
 605
 606 /**
 607  * Try to perform a transmission on the given session. Will solicit
 608  * additional messages if the 'sme' queue is not full enough or has
 609  * only low-priority messages.
 610  *
 611  * @param session session to transmit messages from
 612  */
 613 static void
 614 try_transmission (struct Session *session)
 615 {
 616   struct SessionMessageEntry *pos;
 617   size_t msize;
 618   struct GNUNET_TIME_Absolute now;
 619   struct GNUNET_TIME_Absolute min_deadline;
 620   enum GNUNET_CORE_Priority maxp;
 621   enum GNUNET_CORE_Priority maxpc;
 622   struct GSC_ClientActiveRequest *car;
 623   int excess;
 624
 625   if (GNUNET_YES != session->ready_to_transmit)
 626   {
 627     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 628                 "Not yet ready to transmit, not evaluating queue\n");
 629     return;
 630   }
 631   msize = 0;
 632   min_deadline = GNUNET_TIME_UNIT_FOREVER_ABS;
 633   /* if the peer has excess bandwidth, background traffic is allowed,
 634      otherwise not */
 635   if (MAX_ENCRYPTED_MESSAGE_QUEUE_SIZE <=
 636       GSC_NEIGHBOURS_get_queue_length (session->kx))
 637   {
 638     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 639                 "Transmission queue already very long, waiting...\n");
 640     return; /* queue already too long */
 641   }
 642   excess = GSC_NEIGHBOURS_check_excess_bandwidth (session->kx);
 643   if (GNUNET_YES == excess)
 644     maxp = GNUNET_CORE_PRIO_BACKGROUND;
 645   else
 646     maxp = GNUNET_CORE_PRIO_BEST_EFFORT;
 647   /* determine highest priority of 'ready' messages we already solicited from clients */
 648   pos = session->sme_head;
 649   while ((NULL != pos) &&
 650          (msize + pos->size <= GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE))
 651   {
 652     GNUNET_assert (pos->size < GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE);
 653     msize += pos->size;
 654     maxp = GNUNET_MAX (maxp, pos->priority);
 655     min_deadline = GNUNET_TIME_absolute_min (min_deadline,
 656                                              pos->deadline);
 657     pos = pos->next;
 658   }
 659   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 660               "Calculating transmission set with %u priority (%s) and %s earliest deadline\n",
 661               maxp,
 662               (GNUNET_YES == excess) ? "excess bandwidth" : "limited bandwidth",
 663               GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining (min_deadline),
 664                                                       GNUNET_YES));
 665
 666   if (maxp < GNUNET_CORE_PRIO_CRITICAL_CONTROL)
 667   {
 668     /* if highest already solicited priority from clients is not critical,
 669        check if there are higher-priority messages to be solicited from clients */
 670     if (GNUNET_YES == excess)
 671       maxpc = GNUNET_CORE_PRIO_BACKGROUND;
 672     else
 673       maxpc = GNUNET_CORE_PRIO_BEST_EFFORT;
 674     for (car = session->active_client_request_head; NULL != car; car = car->next)
 675     {
 676       if (GNUNET_YES == car->was_solicited)
 677         continue;
 678       maxpc = GNUNET_MAX (maxpc,
 679                           car->priority);
 680     }
 681     if (maxpc > maxp)
 682     {
 683       /* we have messages waiting for solicitation that have a higher
 684          priority than those that we already accepted; solicit the
 685          high-priority messages first */
 686       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 687                   "Soliciting messages based on priority (%u > %u)\n",
 688                   maxpc,
 689                   maxp);
 690       solicit_messages (session, 0);
 691       return;
 692     }
 693   }
 694   else
 695   {
 696     /* never solicit more, we have critical messages to process */
 697     excess = GNUNET_NO;
 698     maxpc = GNUNET_CORE_PRIO_BACKGROUND;
 699   }
 700   now = GNUNET_TIME_absolute_get ();
 701   if ( ( (GNUNET_YES == excess) ||
 702          (maxpc >= GNUNET_CORE_PRIO_BEST_EFFORT) ) &&
 703        ( (0 == msize) ||
 704          ( (msize < GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE / 2) &&
 705            (min_deadline.abs_value_us > now.abs_value_us))) )
 706   {
 707     /* not enough ready yet (tiny message & cork possible), or no messages at all,
 708        and either excess bandwidth or best-effort or higher message waiting at
 709        client; in this case, we try to solicit more */
 710     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 711                 "Soliciting messages (excess %d, maxpc %d, message size %u, deadline %s)\n",
 712                 excess,
 713                 maxpc,
 714                 (unsigned int) msize,
 715                 GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining (min_deadline),
 716                                                         GNUNET_YES));
 717     solicit_messages (session,
 718                       msize);
 719     if (msize > 0)
 720     {
 721       /* if there is data to send, just not yet, make sure we do transmit
 722        * it once the deadline is reached */
 723       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 724                   "Corking until %s\n",
 725                   GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining (min_deadline),
 726                                                           GNUNET_YES));
 727       if (NULL != session->cork_task)
 728         GNUNET_SCHEDULER_cancel (session->cork_task);
 729       session->cork_task =
 730           GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_absolute_get_remaining (min_deadline),
 731                                         &pop_cork_task,
 732                                         session);
 733     }
 734     else
 735     {
 736       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 737                   "Queue empty, waiting for solicitations\n");
 738     }
 739     return;
 740   }
 741   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 742               "Building combined plaintext buffer to transmit message!\n");
 743   /* create plaintext buffer of all messages (that fit), encrypt and
 744      transmit */
 745   {
 746     static unsigned long long total_bytes;
 747     static unsigned int total_msgs;
 748     char pbuf[msize];           /* plaintext */
 749     size_t used;
 750
 751     used = 0;
 752     while ( (NULL != (pos = session->sme_head)) &&
 753             (used + pos->size <= msize) )
 754     {
 755       GNUNET_memcpy (&pbuf[used], &pos[1], pos->size);
 756       used += pos->size;
 757       GNUNET_CONTAINER_DLL_remove (session->sme_head,
 758                                    session->sme_tail,
 759                                    pos);
 760       GNUNET_free (pos);
 761     }
 762     /* compute average payload size */
 763     total_bytes += used;
 764     total_msgs++;
 765     if (0 == total_msgs)
 766     {
 767       /* 2^32 messages, wrap around... */
 768       total_msgs = 1;
 769       total_bytes = used;
 770     }
 771     GNUNET_STATISTICS_set (GSC_stats,
 772                            "# avg payload per encrypted message",
 773                            total_bytes / total_msgs,
 774                            GNUNET_NO);
 775     /* now actually transmit... */
 776     session->ready_to_transmit = GNUNET_NO;
 777     GSC_KX_encrypt_and_transmit (session->kx,
 778                                  pbuf,
 779                                  used);
 780   }
 781 }
 782
 783
 784 /**
 785  * Send an updated typemap message to the neighbour now,
 786  * and restart typemap transmissions.
 787  *
 788  * @param cls the message
 789  * @param key neighbour's identity
 790  * @param value `struct Neighbour` of the target
 791  * @return always #GNUNET_OK
 792  */
 793 static int
 794 do_restart_typemap_message (void *cls,
 795                             const struct GNUNET_PeerIdentity *key,
 796                             void *value)
 797 {
 798   const struct GNUNET_MessageHeader *hdr = cls;
 799   struct Session *session = value;
 800   struct SessionMessageEntry *sme;
 801   uint16_t size;
 802
 803   size = ntohs (hdr->size);
 804   sme = GNUNET_malloc (sizeof (struct SessionMessageEntry) + size);
 805   GNUNET_memcpy (&sme[1],
 806                  hdr,
 807                  size);
 808   sme->size = size;
 809   sme->priority = GNUNET_CORE_PRIO_CRITICAL_CONTROL;
 810   GNUNET_CONTAINER_DLL_insert (session->sme_head,
 811                                session->sme_tail,
 812                                sme);
 813   try_transmission (session);
 814   start_typemap_task (session);
 815   return GNUNET_OK;
 816 }
 817
 818
 819 /**
 820  * Broadcast an updated typemap message to all neighbours.
 821  * Restarts the retransmissions until the typemaps are confirmed.
 822  *
 823  * @param msg message to transmit
 824  */
 825 void
 826 GSC_SESSIONS_broadcast_typemap (const struct GNUNET_MessageHeader *msg)
 827 {
 828   if (NULL == sessions)
 829     return;
 830   GNUNET_CONTAINER_multipeermap_iterate (sessions,
 831                                          &do_restart_typemap_message,
 832                                          (void *) msg);
 833 }
 834
 835
 836 /**
 837  * Traffic is being solicited for the given peer.  This means that the
 838  * message queue on the transport-level (NEIGHBOURS subsystem) is now
 839  * empty and it is now OK to transmit another (non-control) message.
 840  *
 841  * @param pid identity of peer ready to receive data
 842  */
 843 void
 844 GSC_SESSIONS_solicit (const struct GNUNET_PeerIdentity *pid)
 845 {
 846   struct Session *session;
 847
 848   session = find_session (pid);
 849   if (NULL == session)
 850     return;
 851   session->ready_to_transmit = GNUNET_YES;
 852   try_transmission (session);
 853 }
 854
 855
 856 /**
 857  * Transmit a message to a particular peer.
 858  *
 859  * @param car original request that was queued and then solicited;
 860  *            this handle will now be 'owned' by the SESSIONS subsystem
 861  * @param msg message to transmit
 862  * @param cork is corking allowed?
 863  * @param priority how important is this message
 864  */
 865 void
 866 GSC_SESSIONS_transmit (struct GSC_ClientActiveRequest *car,
 867                        const struct GNUNET_MessageHeader *msg,
 868                        int cork,
 869                        enum GNUNET_CORE_Priority priority)
 870 {
 871   struct Session *session;
 872   struct SessionMessageEntry *sme;
 873   struct SessionMessageEntry *pos;
 874   size_t msize;
 875
 876   session = find_session (&car->target);
 877   if (NULL == session)
 878     return;
 879   msize = ntohs (msg->size);
 880   sme = GNUNET_malloc (sizeof (struct SessionMessageEntry) + msize);
 881   GNUNET_memcpy (&sme[1],
 882                  msg,
 883                  msize);
 884   sme->size = msize;
 885   sme->priority = priority;
 886   if (GNUNET_YES == cork)
 887   {
 888     sme->deadline =
 889         GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_MAX_CORK_DELAY);
 890     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 891                 "Mesage corked, delaying transmission\n");
 892   }
 893   pos = session->sme_head;
 894   while ( (NULL != pos) &&
 895           (pos->priority >= sme->priority) )
 896     pos = pos->next;
 897   if (NULL == pos)
 898     GNUNET_CONTAINER_DLL_insert_tail (session->sme_head,
 899                                       session->sme_tail,
 900                                       sme);
 901   else
 902     GNUNET_CONTAINER_DLL_insert_after (session->sme_head,
 903                                        session->sme_tail,
 904                                        pos->prev,
 905                                        sme);
 906   try_transmission (session);
 907 }
 908
 909
 910 /**
 911  * We have received a typemap message from a peer, update ours.
 912  * Notifies clients about the session.
 913  *
 914  * @param peer peer this is about
 915  * @param msg typemap update message
 916  */
 917 void
 918 GSC_SESSIONS_set_typemap (const struct GNUNET_PeerIdentity *peer,
 919                           const struct GNUNET_MessageHeader *msg)
 920 {
 921   struct Session *session;
 922   struct GSC_TypeMap *nmap;
 923   struct SessionMessageEntry *sme;
 924   struct TypeMapConfirmationMessage *tmc;
 925
 926   nmap = GSC_TYPEMAP_get_from_message (msg);
 927   if (NULL == nmap)
 928     return;                     /* malformed */
 929   session = find_session (peer);
 930   if (NULL == session)
 931   {
 932     GNUNET_break (0);
 933     return;
 934   }
 935   sme = GNUNET_malloc (sizeof (struct SessionMessageEntry) +
 936                        sizeof (struct TypeMapConfirmationMessage));
 937   sme->deadline = GNUNET_TIME_absolute_get ();
 938   sme->size = sizeof (struct TypeMapConfirmationMessage);
 939   sme->priority = GNUNET_CORE_PRIO_CRITICAL_CONTROL;
 940   tmc = (struct TypeMapConfirmationMessage *) &sme[1];
 941   tmc->header.size = htons (sizeof (struct TypeMapConfirmationMessage));
 942   tmc->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_CONFIRM_TYPE_MAP);
 943   tmc->reserved = htonl (0);
 944   GSC_TYPEMAP_hash (nmap,
 945                     &tmc->tm_hash);
 946   GNUNET_CONTAINER_DLL_insert (session->sme_head,
 947                                session->sme_tail,
 948                                sme);
 949   try_transmission (session);
 950   GSC_CLIENTS_notify_clients_about_neighbour (peer,
 951                                               session->tmap,
 952                                               nmap);
 953   GSC_TYPEMAP_destroy (session->tmap);
 954   session->tmap = nmap;
 955 }
 956
 957
 958 /**
 959  * The given peer send a message of the specified type.  Make sure the
 960  * respective bit is set in its type-map and that clients are notified
 961  * about the session.
 962  *
 963  * @param peer peer this is about
 964  * @param type type of the message
 965  */
 966 void
 967 GSC_SESSIONS_add_to_typemap (const struct GNUNET_PeerIdentity *peer,
 968                              uint16_t type)
 969 {
 970   struct Session *session;
 971   struct GSC_TypeMap *nmap;
 972
 973   if (0 == memcmp (peer,
 974                    &GSC_my_identity,
 975                    sizeof (struct GNUNET_PeerIdentity)))
 976     return;
 977   session = find_session (peer);
 978   GNUNET_assert (NULL != session);
 979   if (GNUNET_YES == GSC_TYPEMAP_test_match (session->tmap, &type, 1))
 980     return;                     /* already in it */
 981   nmap = GSC_TYPEMAP_extend (session->tmap, &type, 1);
 982   GSC_CLIENTS_notify_clients_about_neighbour (peer,
 983                                               session->tmap, nmap);
 984   GSC_TYPEMAP_destroy (session->tmap);
 985   session->tmap = nmap;
 986 }
 987
 988
 989 /**
 990  * Initialize sessions subsystem.
 991  */
 992 void
 993 GSC_SESSIONS_init ()
 994 {
 995   sessions = GNUNET_CONTAINER_multipeermap_create (128,
 996                                                    GNUNET_YES);
 997 }
 998
 999
1000 /**
1001  * Helper function for #GSC_SESSIONS_done() to free all
1002  * active sessions.
1003  *
1004  * @param cls NULL
1005  * @param key identity of the connected peer
1006  * @param value the `struct Session` for the peer
1007  * @return #GNUNET_OK (continue to iterate)
1008  */
1009 static int
1010 free_session_helper (void *cls,
1011                      const struct GNUNET_PeerIdentity *key,
1012                      void *value)
1013 {
1014   /* struct Session *session = value; */
1015
1016   GSC_SESSIONS_end (key);
1017   return GNUNET_OK;
1018 }
1019
1020
1021 /**
1022  * Shutdown sessions subsystem.
1023  */
1024 void
1025 GSC_SESSIONS_done ()
1026 {
1027   if (NULL != sessions)
1028   {
1029     GNUNET_CONTAINER_multipeermap_iterate (sessions,
1030                                            &free_session_helper,
1031                                            NULL);
1032     GNUNET_CONTAINER_multipeermap_destroy (sessions);
1033     sessions = NULL;
1034   }
1035 }
1036
1037 /* end of gnunet-service-core_sessions.c */