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    * We keep messages in a doubly linked list.
  50    */
  51   struct SessionMessageEntry *next;
  52
  53   /**
  54    * We keep messages in a doubly linked list.
  55    */
  56   struct SessionMessageEntry *prev;
  57
  58   /**
  59    * How important is this message.
  60    */
  61   enum GNUNET_MQ_PriorityPreferences priority;
  62
  63   /**
  64    * Flag set to #GNUNET_YES if this is a typemap message.
  65    */
  66   int is_typemap;
  67
  68   /**
  69    * Flag set to #GNUNET_YES if this is a typemap confirmation message.
  70    */
  71   int is_typemap_confirm;
  72
  73   /**
  74    * Deadline for transmission, 1s after we received it (if we
  75    * are not corking), otherwise "now".  Note that this message
  76    * does NOT expire past its deadline.
  77    */
  78   struct GNUNET_TIME_Absolute deadline;
  79
  80   /**
  81    * How long is the message? (number of bytes following the `struct
  82    * MessageEntry`, but not including the size of `struct
  83    * MessageEntry` itself!)
  84    */
  85   size_t size;
  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    * 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 GNUNET_NETWORK_STRUCT_END
 180
 181
 182 /**
 183  * Map of peer identities to `struct Session`.
 184  */
 185 static struct GNUNET_CONTAINER_MultiPeerMap *sessions;
 186
 187
 188 /**
 189  * Find the session for the given peer.
 190  *
 191  * @param peer identity of the peer
 192  * @return NULL if we are not connected, otherwise the
 193  *         session handle
 194  */
 195 static struct Session *
 196 find_session (const struct GNUNET_PeerIdentity *peer)
 197 {
 198   if (NULL == sessions)
 199     return NULL;
 200   return GNUNET_CONTAINER_multipeermap_get (sessions, peer);
 201 }
 202
 203
 204 /**
 205  * End the session with the given peer (we are no longer
 206  * connected).
 207  *
 208  * @param pid identity of peer to kill session with
 209  */
 210 void
 211 GSC_SESSIONS_end (const struct GNUNET_PeerIdentity *pid)
 212 {
 213   struct Session *session;
 214   struct GSC_ClientActiveRequest *car;
 215   struct SessionMessageEntry *sme;
 216
 217   session = find_session (pid);
 218   if (NULL == session)
 219     return;
 220   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 221               "Destroying session for peer `%s'\n",
 222               GNUNET_i2s (session->peer));
 223   if (NULL != session->cork_task)
 224   {
 225     GNUNET_SCHEDULER_cancel (session->cork_task);
 226     session->cork_task = NULL;
 227   }
 228   while (NULL != (car = session->active_client_request_head))
 229   {
 230     GNUNET_CONTAINER_DLL_remove (session->active_client_request_head,
 231                                  session->active_client_request_tail,
 232                                  car);
 233     GSC_CLIENTS_reject_request (car, GNUNET_NO);
 234   }
 235   while (NULL != (sme = session->sme_head))
 236   {
 237     GNUNET_CONTAINER_DLL_remove (session->sme_head, session->sme_tail, sme);
 238     GNUNET_free (sme);
 239   }
 240   if (NULL != session->typemap_task)
 241   {
 242     GNUNET_SCHEDULER_cancel (session->typemap_task);
 243     session->typemap_task = NULL;
 244   }
 245   GSC_CLIENTS_notify_clients_about_neighbour (session->peer,
 246                                               session->tmap,
 247                                               NULL);
 248   GNUNET_assert (
 249     GNUNET_YES ==
 250     GNUNET_CONTAINER_multipeermap_remove (sessions, session->peer, session));
 251   GNUNET_STATISTICS_set (GSC_stats,
 252                          gettext_noop ("# peers connected"),
 253                          GNUNET_CONTAINER_multipeermap_size (sessions),
 254                          GNUNET_NO);
 255   GSC_TYPEMAP_destroy (session->tmap);
 256   session->tmap = NULL;
 257   GNUNET_free (session);
 258 }
 259
 260
 261 /**
 262  * Transmit our current typemap message to the other peer.
 263  * (Done periodically until the typemap is confirmed).
 264  *
 265  * @param cls the `struct Session *`
 266  */
 267 static void
 268 transmit_typemap_task (void *cls)
 269 {
 270   struct Session *session = cls;
 271   struct GNUNET_MessageHeader *hdr;
 272   struct GNUNET_TIME_Relative delay;
 273
 274   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 275               "Sending TYPEMAP to %s\n",
 276               GNUNET_i2s (session->peer));
 277   session->typemap_delay = GNUNET_TIME_STD_BACKOFF (session->typemap_delay);
 278   delay = session->typemap_delay;
 279   /* randomize a bit to avoid spont. sync */
 280   delay.rel_value_us +=
 281     GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, 1000 * 1000);
 282   session->typemap_task =
 283     GNUNET_SCHEDULER_add_delayed (delay, &transmit_typemap_task, session);
 284   GNUNET_STATISTICS_update (GSC_stats,
 285                             gettext_noop ("# type map refreshes sent"),
 286                             1,
 287                             GNUNET_NO);
 288   hdr = GSC_TYPEMAP_compute_type_map_message ();
 289   GSC_KX_encrypt_and_transmit (session->kx, hdr, ntohs (hdr->size));
 290   GNUNET_free (hdr);
 291 }
 292
 293
 294 /**
 295  * Restart the typemap task for the given session.
 296  *
 297  * @param session session to restart typemap transmission for
 298  */
 299 static void
 300 start_typemap_task (struct Session *session)
 301 {
 302   if (NULL != session->typemap_task)
 303     GNUNET_SCHEDULER_cancel (session->typemap_task);
 304   session->typemap_delay = GNUNET_TIME_UNIT_SECONDS;
 305   session->typemap_task = GNUNET_SCHEDULER_add_delayed (session->typemap_delay,
 306                                                         &transmit_typemap_task,
 307                                                         session);
 308 }
 309
 310
 311 /**
 312  * Create a session, a key exchange was just completed.
 313  *
 314  * @param peer peer that is now connected
 315  * @param kx key exchange that completed
 316  */
 317 void
 318 GSC_SESSIONS_create (const struct GNUNET_PeerIdentity *peer,
 319                      struct GSC_KeyExchangeInfo *kx)
 320 {
 321   struct Session *session;
 322
 323   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 324               "Creating session for peer `%s'\n",
 325               GNUNET_i2s (peer));
 326   session = GNUNET_new (struct Session);
 327   session->tmap = GSC_TYPEMAP_create ();
 328   session->peer = peer;
 329   session->kx = kx;
 330   GNUNET_assert (GNUNET_OK ==
 331                  GNUNET_CONTAINER_multipeermap_put (
 332                    sessions,
 333                    session->peer,
 334                    session,
 335                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 336   GNUNET_STATISTICS_set (GSC_stats,
 337                          gettext_noop ("# peers connected"),
 338                          GNUNET_CONTAINER_multipeermap_size (sessions),
 339                          GNUNET_NO);
 340   GSC_CLIENTS_notify_clients_about_neighbour (peer, NULL, session->tmap);
 341   start_typemap_task (session);
 342 }
 343
 344
 345 /**
 346  * The other peer has indicated that it 'lost' the session
 347  * (KX down), reinitialize the session on our end, in particular
 348  * this means to restart the typemap transmission.
 349  *
 350  * @param peer peer that is now connected
 351  */
 352 void
 353 GSC_SESSIONS_reinit (const struct GNUNET_PeerIdentity *peer)
 354 {
 355   struct Session *session;
 356
 357   session = find_session (peer);
 358   if (NULL == session)
 359   {
 360     /* KX/session is new for both sides; thus no need to restart what
 361        has not yet begun */
 362     return;
 363   }
 364   start_typemap_task (session);
 365 }
 366
 367
 368 /**
 369  * The other peer has confirmed receiving our type map,
 370  * check if it is current and if so, stop retransmitting it.
 371  *
 372  * @param peer peer that confirmed the type map
 373  * @param msg confirmation message we received
 374  */
 375 void
 376 GSC_SESSIONS_confirm_typemap (const struct GNUNET_PeerIdentity *peer,
 377                               const struct GNUNET_MessageHeader *msg)
 378 {
 379   const struct TypeMapConfirmationMessage *cmsg;
 380   struct Session *session;
 381
 382   session = find_session (peer);
 383   if (NULL == session)
 384   {
 385     GNUNET_break (0);
 386     return;
 387   }
 388   if (ntohs (msg->size) != sizeof(struct TypeMapConfirmationMessage))
 389   {
 390     GNUNET_break_op (0);
 391     return;
 392   }
 393   cmsg = (const struct TypeMapConfirmationMessage *) msg;
 394   if (GNUNET_YES != GSC_TYPEMAP_check_hash (&cmsg->tm_hash))
 395   {
 396     /* our typemap has changed in the meantime, do not
 397        accept confirmation */
 398     GNUNET_STATISTICS_update (GSC_stats,
 399                               gettext_noop (
 400                                 "# outdated typemap confirmations received"),
 401                               1,
 402                               GNUNET_NO);
 403     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 404                 "Got outdated typemap confirmated from peer `%s'\n",
 405                 GNUNET_i2s (session->peer));
 406     return;
 407   }
 408   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 409               "Got typemap confirmation from peer `%s'\n",
 410               GNUNET_i2s (session->peer));
 411   if (NULL != session->typemap_task)
 412   {
 413     GNUNET_SCHEDULER_cancel (session->typemap_task);
 414     session->typemap_task = NULL;
 415   }
 416   GNUNET_STATISTICS_update (GSC_stats,
 417                             gettext_noop (
 418                               "# valid typemap confirmations received"),
 419                             1,
 420                             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, GNUNET_NO);
 494     return;
 495   }
 496   if (car->msize > GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE)
 497   {
 498     GNUNET_break (0);
 499     GSC_CLIENTS_reject_request (car, GNUNET_YES);
 500     return;
 501   }
 502   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 503               "Received client transmission request. queueing\n");
 504   GNUNET_CONTAINER_DLL_insert_tail (session->active_client_request_head,
 505                                     session->active_client_request_tail,
 506                                     car);
 507   try_transmission (session);
 508 }
 509
 510
 511 /**
 512  * Dequeue a request from a client from transmission to a particular peer.
 513  *
 514  * @param car request to dequeue; this handle will then be 'owned' by
 515  *        the caller (CLIENTS sysbsystem)
 516  */
 517 void
 518 GSC_SESSIONS_dequeue_request (struct GSC_ClientActiveRequest *car)
 519 {
 520   struct Session *session;
 521
 522   if (0 == memcmp (&car->target,
 523                    &GSC_my_identity,
 524                    sizeof(struct GNUNET_PeerIdentity)))
 525     return;
 526   session = find_session (&car->target);
 527   GNUNET_assert (NULL != session);
 528   GNUNET_CONTAINER_DLL_remove (session->active_client_request_head,
 529                                session->active_client_request_tail,
 530                                car);
 531   /* dequeueing of 'high' priority messages may unblock
 532      transmission for lower-priority messages, so we also
 533      need to try in this case. */
 534   try_transmission (session);
 535 }
 536
 537
 538 /**
 539  * Solicit messages for transmission, starting with those of the highest
 540  * priority.
 541  *
 542  * @param session session to solict messages for
 543  * @param msize how many bytes do we have already
 544  */
 545 static void
 546 solicit_messages (struct Session *session, size_t msize)
 547 {
 548   struct GSC_ClientActiveRequest *car;
 549   struct GSC_ClientActiveRequest *nxt;
 550   size_t so_size;
 551   enum GNUNET_MQ_PriorityPreferences pmax;
 552
 553   so_size = msize;
 554   pmax = GNUNET_MQ_PRIO_BACKGROUND;
 555   for (car = session->active_client_request_head; NULL != car; car = car->next)
 556   {
 557     if (GNUNET_YES == car->was_solicited)
 558       continue;
 559     pmax = GNUNET_MAX (pmax, car->priority & GNUNET_MQ_PRIORITY_MASK);
 560   }
 561   nxt = session->active_client_request_head;
 562   while (NULL != (car = nxt))
 563   {
 564     nxt = car->next;
 565     if (car->priority < pmax)
 566       continue;
 567     if (so_size + car->msize > GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE)
 568       break;
 569     so_size += car->msize;
 570     if (GNUNET_YES == car->was_solicited)
 571       continue;
 572     car->was_solicited = GNUNET_YES;
 573     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 574                 "Soliciting message with priority %u\n",
 575                 car->priority);
 576     GSC_CLIENTS_solicit_request (car);
 577     /* The above call may *dequeue* requests and thereby
 578        clobber 'nxt'. Hence we need to restart from the
 579        head of the list. */
 580     nxt = session->active_client_request_head;
 581     so_size = msize;
 582   }
 583 }
 584
 585
 586 /**
 587  * Some messages were delayed (corked), but the timeout has now expired.
 588  * Send them now.
 589  *
 590  * @param cls `struct Session` with the messages to transmit now
 591  */
 592 static void
 593 pop_cork_task (void *cls)
 594 {
 595   struct Session *session = cls;
 596
 597   session->cork_task = NULL;
 598   try_transmission (session);
 599 }
 600
 601
 602 /**
 603  * Try to perform a transmission on the given session. Will solicit
 604  * additional messages if the 'sme' queue is not full enough or has
 605  * only low-priority messages.
 606  *
 607  * @param session session to transmit messages from
 608  */
 609 static void
 610 try_transmission (struct Session *session)
 611 {
 612   struct SessionMessageEntry *pos;
 613   size_t msize;
 614   struct GNUNET_TIME_Absolute now;
 615   struct GNUNET_TIME_Absolute min_deadline;
 616   enum GNUNET_MQ_PriorityPreferences maxp;
 617   enum GNUNET_MQ_PriorityPreferences maxpc;
 618   struct GSC_ClientActiveRequest *car;
 619   int excess;
 620
 621   msize = 0;
 622   min_deadline = GNUNET_TIME_UNIT_FOREVER_ABS;
 623   /* if the peer has excess bandwidth, background traffic is allowed,
 624      otherwise not */
 625   if (MAX_ENCRYPTED_MESSAGE_QUEUE_SIZE <=
 626       GSC_NEIGHBOURS_get_queue_length (session->kx))
 627   {
 628     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 629                 "Transmission queue already very long, waiting...\n");
 630     return;   /* queue already too long */
 631   }
 632   excess = GSC_NEIGHBOURS_check_excess_bandwidth (session->kx);
 633   if (GNUNET_YES == excess)
 634     maxp = GNUNET_MQ_PRIO_BACKGROUND;
 635   else
 636     maxp = GNUNET_MQ_PRIO_BEST_EFFORT;
 637   /* determine highest priority of 'ready' messages we already solicited from clients */
 638   pos = session->sme_head;
 639   while ((NULL != pos) &&
 640          (msize + pos->size <= GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE))
 641   {
 642     GNUNET_assert (pos->size < GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE);
 643     msize += pos->size;
 644     maxp = GNUNET_MAX (maxp, pos->priority & GNUNET_MQ_PRIORITY_MASK);
 645     min_deadline = GNUNET_TIME_absolute_min (min_deadline, pos->deadline);
 646     pos = pos->next;
 647   }
 648   GNUNET_log (
 649     GNUNET_ERROR_TYPE_DEBUG,
 650     "Calculating transmission set with %u priority (%s) and %s earliest deadline\n",
 651     maxp,
 652     (GNUNET_YES == excess) ? "excess bandwidth" : "limited bandwidth",
 653     GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining (
 654                                               min_deadline),
 655                                             GNUNET_YES));
 656
 657   if (maxp < GNUNET_MQ_PRIO_CRITICAL_CONTROL)
 658   {
 659     /* if highest already solicited priority from clients is not critical,
 660        check if there are higher-priority messages to be solicited from clients */
 661     if (GNUNET_YES == excess)
 662       maxpc = GNUNET_MQ_PRIO_BACKGROUND;
 663     else
 664       maxpc = GNUNET_MQ_PRIO_BEST_EFFORT;
 665     for (car = session->active_client_request_head; NULL != car;
 666          car = car->next)
 667     {
 668       if (GNUNET_YES == car->was_solicited)
 669         continue;
 670       maxpc = GNUNET_MAX (maxpc, car->priority & GNUNET_MQ_PRIORITY_MASK);
 671     }
 672     if (maxpc > maxp)
 673     {
 674       /* we have messages waiting for solicitation that have a higher
 675          priority than those that we already accepted; solicit the
 676          high-priority messages first */
 677       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 678                   "Soliciting messages based on priority (%u > %u)\n",
 679                   maxpc,
 680                   maxp);
 681       solicit_messages (session, 0);
 682       return;
 683     }
 684   }
 685   else
 686   {
 687     /* never solicit more, we have critical messages to process */
 688     excess = GNUNET_NO;
 689     maxpc = GNUNET_MQ_PRIO_BACKGROUND;
 690   }
 691   now = GNUNET_TIME_absolute_get ();
 692   if (((GNUNET_YES == excess) || (maxpc >= GNUNET_MQ_PRIO_BEST_EFFORT)) &&
 693       ((0 == msize) ||
 694        ((msize < GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE / 2) &&
 695         (min_deadline.abs_value_us > now.abs_value_us))))
 696   {
 697     /* not enough ready yet (tiny message & cork possible), or no messages at all,
 698        and either excess bandwidth or best-effort or higher message waiting at
 699        client; in this case, we try to solicit more */
 700     GNUNET_log (
 701       GNUNET_ERROR_TYPE_DEBUG,
 702       "Soliciting messages (excess %d, maxpc %d, message size %u, deadline %s)\n",
 703       excess,
 704       maxpc,
 705       (unsigned int) msize,
 706       GNUNET_STRINGS_relative_time_to_string (
 707         GNUNET_TIME_absolute_get_remaining (
 708           min_deadline),
 709         GNUNET_YES));
 710     solicit_messages (session, msize);
 711     if (msize > 0)
 712     {
 713       /* if there is data to send, just not yet, make sure we do transmit
 714        * it once the deadline is reached */
 715       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 716                   "Corking until %s\n",
 717                   GNUNET_STRINGS_relative_time_to_string (
 718                     GNUNET_TIME_absolute_get_remaining (min_deadline),
 719                     GNUNET_YES));
 720       if (NULL != session->cork_task)
 721         GNUNET_SCHEDULER_cancel (session->cork_task);
 722       session->cork_task =
 723         GNUNET_SCHEDULER_add_at (min_deadline, &pop_cork_task, session);
 724     }
 725     else
 726     {
 727       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 728                   "Queue empty, waiting for solicitations\n");
 729     }
 730     return;
 731   }
 732   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 733               "Building combined plaintext buffer to transmit message!\n");
 734   /* create plaintext buffer of all messages (that fit), encrypt and
 735      transmit */
 736   {
 737     static unsigned long long total_bytes;
 738     static unsigned int total_msgs;
 739     char pbuf[msize]; /* plaintext */
 740     size_t used;
 741
 742     used = 0;
 743     while ((NULL != (pos = session->sme_head)) && (used + pos->size <= msize))
 744     {
 745       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 746                   "Adding message of type %d (%d/%d) to payload for %s\n",
 747                   ntohs (((const struct GNUNET_MessageHeader *) &pos[1])->type),
 748                   pos->is_typemap,
 749                   pos->is_typemap_confirm,
 750                   GNUNET_i2s (session->peer));
 751       GNUNET_memcpy (&pbuf[used], &pos[1], pos->size);
 752       used += pos->size;
 753       GNUNET_CONTAINER_DLL_remove (session->sme_head, session->sme_tail, pos);
 754       GNUNET_free (pos);
 755     }
 756     /* compute average payload size */
 757     total_bytes += used;
 758     total_msgs++;
 759     if (0 == total_msgs)
 760     {
 761       /* 2^32 messages, wrap around... */
 762       total_msgs = 1;
 763       total_bytes = used;
 764     }
 765     GNUNET_STATISTICS_set (GSC_stats,
 766                            "# avg payload per encrypted message",
 767                            total_bytes / total_msgs,
 768                            GNUNET_NO);
 769     /* now actually transmit... */
 770     GSC_KX_encrypt_and_transmit (session->kx, pbuf, used);
 771   }
 772 }
 773
 774
 775 /**
 776  * Send an updated typemap message to the neighbour now,
 777  * and restart typemap transmissions.
 778  *
 779  * @param cls the message
 780  * @param key neighbour's identity
 781  * @param value `struct Neighbour` of the target
 782  * @return always #GNUNET_OK
 783  */
 784 static int
 785 do_restart_typemap_message (void *cls,
 786                             const struct GNUNET_PeerIdentity *key,
 787                             void *value)
 788 {
 789   const struct GNUNET_MessageHeader *hdr = cls;
 790   struct Session *session = value;
 791   struct SessionMessageEntry *sme;
 792   uint16_t size;
 793
 794   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 795               "Restarting sending TYPEMAP to %s\n",
 796               GNUNET_i2s (session->peer));
 797   size = ntohs (hdr->size);
 798   for (sme = session->sme_head; NULL != sme; sme = sme->next)
 799   {
 800     if (GNUNET_YES == sme->is_typemap)
 801     {
 802       GNUNET_CONTAINER_DLL_remove (session->sme_head, session->sme_tail, sme);
 803       GNUNET_free (sme);
 804       break;
 805     }
 806   }
 807   sme = GNUNET_malloc (sizeof(struct SessionMessageEntry) + size);
 808   sme->is_typemap = GNUNET_YES;
 809   GNUNET_memcpy (&sme[1], hdr, size);
 810   sme->size = size;
 811   sme->priority = GNUNET_MQ_PRIO_CRITICAL_CONTROL;
 812   GNUNET_CONTAINER_DLL_insert (session->sme_head, session->sme_tail, 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   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 849               "Transport solicits for %s\n",
 850               GNUNET_i2s (pid));
 851   session = find_session (pid);
 852   if (NULL == session)
 853     return;
 854   try_transmission (session);
 855 }
 856
 857
 858 /**
 859  * Transmit a message to a particular peer.
 860  *
 861  * @param car original request that was queued and then solicited;
 862  *            this handle will now be 'owned' by the SESSIONS subsystem
 863  * @param msg message to transmit
 864  * @param priority how important is this message
 865  */
 866 void
 867 GSC_SESSIONS_transmit (struct GSC_ClientActiveRequest *car,
 868                        const struct GNUNET_MessageHeader *msg,
 869                        enum GNUNET_MQ_PriorityPreferences 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], msg, msize);
 882   sme->size = msize;
 883   sme->priority = priority;
 884   if (0 != (GNUNET_MQ_PREF_CORK_ALLOWED & priority))
 885   {
 886     sme->deadline =
 887       GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_MAX_CORK_DELAY);
 888     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 889                 "Mesage corked, delaying transmission\n");
 890   }
 891   pos = session->sme_head;
 892   while ((NULL != pos) && (pos->priority >= sme->priority))
 893     pos = pos->next;
 894   if (NULL == pos)
 895     GNUNET_CONTAINER_DLL_insert_tail (session->sme_head,
 896                                       session->sme_tail,
 897                                       sme);
 898   else
 899     GNUNET_CONTAINER_DLL_insert_after (session->sme_head,
 900                                        session->sme_tail,
 901                                        pos->prev,
 902                                        sme);
 903   try_transmission (session);
 904 }
 905
 906
 907 /**
 908  * We have received a typemap message from a peer, update ours.
 909  * Notifies clients about the session.
 910  *
 911  * @param peer peer this is about
 912  * @param msg typemap update message
 913  */
 914 void
 915 GSC_SESSIONS_set_typemap (const struct GNUNET_PeerIdentity *peer,
 916                           const struct GNUNET_MessageHeader *msg)
 917 {
 918   struct Session *session;
 919   struct GSC_TypeMap *nmap;
 920   struct SessionMessageEntry *sme;
 921   struct TypeMapConfirmationMessage *tmc;
 922
 923   nmap = GSC_TYPEMAP_get_from_message (msg);
 924   if (NULL == nmap)
 925   {
 926     GNUNET_break_op (0);
 927     return;   /* malformed */
 928   }
 929   session = find_session (peer);
 930   if (NULL == session)
 931   {
 932     GSC_TYPEMAP_destroy (nmap);
 933     GNUNET_break (0);
 934     return;
 935   }
 936   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 937               "Received TYPEMAP from %s\n",
 938               GNUNET_i2s (session->peer));
 939   for (sme = session->sme_head; NULL != sme; sme = sme->next)
 940   {
 941     if (GNUNET_YES == sme->is_typemap_confirm)
 942     {
 943       GNUNET_CONTAINER_DLL_remove (session->sme_head, session->sme_tail, sme);
 944       GNUNET_free (sme);
 945       break;
 946     }
 947   }
 948   sme = GNUNET_malloc (sizeof(struct SessionMessageEntry)
 949                        + sizeof(struct TypeMapConfirmationMessage));
 950   sme->deadline = GNUNET_TIME_absolute_get ();
 951   sme->size = sizeof(struct TypeMapConfirmationMessage);
 952   sme->priority = GNUNET_MQ_PRIO_CRITICAL_CONTROL;
 953   sme->is_typemap_confirm = GNUNET_YES;
 954   tmc = (struct TypeMapConfirmationMessage *) &sme[1];
 955   tmc->header.size = htons (sizeof(struct TypeMapConfirmationMessage));
 956   tmc->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_CONFIRM_TYPE_MAP);
 957   tmc->reserved = htonl (0);
 958   GSC_TYPEMAP_hash (nmap, &tmc->tm_hash);
 959   GNUNET_CONTAINER_DLL_insert (session->sme_head, session->sme_tail, sme);
 960   try_transmission (session);
 961   GSC_CLIENTS_notify_clients_about_neighbour (peer, session->tmap, nmap);
 962   GSC_TYPEMAP_destroy (session->tmap);
 963   session->tmap = nmap;
 964 }
 965
 966
 967 /**
 968  * The given peer send a message of the specified type.  Make sure the
 969  * respective bit is set in its type-map and that clients are notified
 970  * about the session.
 971  *
 972  * @param peer peer this is about
 973  * @param type type of the message
 974  */
 975 void
 976 GSC_SESSIONS_add_to_typemap (const struct GNUNET_PeerIdentity *peer,
 977                              uint16_t type)
 978 {
 979   struct Session *session;
 980   struct GSC_TypeMap *nmap;
 981
 982   if (0 == memcmp (peer, &GSC_my_identity, sizeof(struct GNUNET_PeerIdentity)))
 983     return;
 984   session = find_session (peer);
 985   GNUNET_assert (NULL != session);
 986   if (GNUNET_YES == GSC_TYPEMAP_test_match (session->tmap, &type, 1))
 987     return; /* already in it */
 988   nmap = GSC_TYPEMAP_extend (session->tmap, &type, 1);
 989   GSC_CLIENTS_notify_clients_about_neighbour (peer, session->tmap, nmap);
 990   GSC_TYPEMAP_destroy (session->tmap);
 991   session->tmap = nmap;
 992 }
 993
 994
 995 /**
 996  * Initialize sessions subsystem.
 997  */
 998 void
 999 GSC_SESSIONS_init ()
1000 {
1001   sessions = GNUNET_CONTAINER_multipeermap_create (128, GNUNET_YES);
1002 }
1003
1004
1005 /**
1006  * Helper function for #GSC_SESSIONS_done() to free all
1007  * active sessions.
1008  *
1009  * @param cls NULL
1010  * @param key identity of the connected peer
1011  * @param value the `struct Session` for the peer
1012  * @return #GNUNET_OK (continue to iterate)
1013  */
1014 static int
1015 free_session_helper (void *cls,
1016                      const struct GNUNET_PeerIdentity *key,
1017                      void *value)
1018 {
1019   /* struct Session *session = value; */
1020
1021   GSC_SESSIONS_end (key);
1022   return GNUNET_OK;
1023 }
1024
1025
1026 /**
1027  * Shutdown sessions subsystem.
1028  */
1029 void
1030 GSC_SESSIONS_done ()
1031 {
1032   if (NULL != sessions)
1033   {
1034     GNUNET_CONTAINER_multipeermap_iterate (sessions,
1035                                            &free_session_helper,
1036                                            NULL);
1037     GNUNET_CONTAINER_multipeermap_destroy (sessions);
1038     sessions = NULL;
1039   }
1040 }
1041
1042
1043 /* end of gnunet-service-core_sessions.c */