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