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