src/core/gnunet-service-core_neighbours.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2009, 2010, 2011 Christian Grothoff (and other contributing authors)
   4
   5      GNUnet is free software; you can redistribute it and/or modify
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 3, or (at your
   8      option) any later version.
   9
  10      GNUnet is distributed in the hope that it will be useful, but
  11      WITHOUT ANY WARRANTY; without even the implied warranty of
  12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18      Boston, MA 02111-1307, USA.
  19 */
  20
  21 /**
  22  * @file core/gnunet-service-core_neighbours.c
  23  * @brief code for managing low-level 'plaintext' connections with transport (key exchange may or may not be done yet)
  24  * @author Christian Grothoff
  25  */
  26 #include "platform.h"
  27 #include "gnunet_util_lib.h"
  28 #include "gnunet_transport_service.h"
  29 #include "gnunet_service_core.h"
  30 #include "gnunet_service_core_neighbours.h"
  31 #include "gnunet_service_core_kx.h"
  32 #include "gnunet_service_core_sessions.h"
  33
  34
  35 /**
  36  * Message ready for transmission via transport service.  This struct
  37  * is followed by the actual content of the message.
  38  */
  39 struct MessageEntry
  40 {
  41
  42   /**
  43    * We keep messages in a doubly linked list.
  44    */
  45   struct MessageEntry *next;
  46
  47   /**
  48    * We keep messages in a doubly linked list.
  49    */
  50   struct MessageEntry *prev;
  51
  52   /**
  53    * By when are we supposed to transmit this message?
  54    */
  55   struct GNUNET_TIME_Absolute deadline;
  56
  57   /**
  58    * How long is the message? (number of bytes following the "struct
  59    * MessageEntry", but not including the size of "struct
  60    * MessageEntry" itself!)
  61    */
  62   size_t size;
  63
  64 };
  65
  66
  67 /**
  68  * Data kept per transport-connected peer.
  69  */
  70 struct Neighbour
  71 {
  72
  73   /**
  74    * Head of the batched message queue (already ordered, transmit
  75    * starting with the head).
  76    */
  77   struct MessageEntry *message_head;
  78
  79   /**
  80    * Tail of the batched message queue (already ordered, append new
  81    * messages to tail).
  82    */
  83   struct MessageEntry *message_tail;
  84
  85   /**
  86    * Handle for pending requests for transmission to this peer
  87    * with the transport service.  NULL if no request is pending.
  88    */
  89   struct GNUNET_TRANSPORT_TransmitHandle *th;
  90
  91   /**
  92    * Information about the key exchange with the other peer.
  93    */
  94   struct GSC_KeyExchangeInfo *kxinfo;
  95
  96   /**
  97    * Identity of the other peer.
  98    */
  99   struct GNUNET_PeerIdentity peer;
 100
 101   /**
 102    * ID of task used for re-trying plaintext scheduling.
 103    */
 104   GNUNET_SCHEDULER_TaskIdentifier retry_plaintext_task;
 105
 106   /**
 107    * Tracking bandwidth for sending to this peer.
 108    */
 109   struct GNUNET_BANDWIDTH_Tracker available_send_window;
 110
 111   /**
 112    * Tracking bandwidth for sending to this peer.
 113    */
 114   struct GNUNET_BANDWIDTH_Tracker available_recv_window;
 115
 116 };
 117
 118
 119 /**
 120  * Map of peer identities to 'struct Neighbour'.
 121  */
 122 static struct GNUNET_CONTAINER_MultiHashMap *neighbours;
 123
 124 /**
 125  * Transport service.
 126  */
 127 static struct GNUNET_TRANSPORT_Handle *transport;
 128
 129
 130 /**
 131  * Find the entry for the given neighbour.
 132  *
 133  * @param peer identity of the neighbour
 134  * @return NULL if we are not connected, otherwise the
 135  *         neighbour's entry.
 136  */
 137 static struct Neighbour *
 138 find_neighbour (const struct GNUNET_PeerIdentity *peer)
 139 {
 140   return GNUNET_CONTAINER_multihashmap_get (neighbours, &peer->hashPubKey);
 141 }
 142
 143
 144 /**
 145  * Free the given entry for the neighbour.
 146  *
 147  * @param n neighbour to free
 148  */
 149 static void
 150 free_neighbour (struct Neighbour *n)
 151 {
 152   struct MessageEntry *m;
 153
 154 #if DEBUG_CORE
 155   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 156               "Destroying neighbour entry for peer `%4s'\n",
 157               GNUNET_i2s (&n->peer));
 158 #endif
 159   while (NULL != (m = n->message_head))
 160   {
 161     GNUNET_CONTAINER_DLL_remove (n->message_head, n->message_tail, m);
 162     GNUNET_free (m);
 163   }
 164   if (NULL != n->th)
 165   {
 166     GNUNET_TRANSPORT_notify_transmit_ready_cancel (n->th);
 167     n->th = NULL;
 168   }
 169   GSC_SESSIONS_end (&n->peer);
 170   if (NULL != n->kx)
 171   {
 172     GSC_KX_stop (n->kx);
 173     n->kx = NULL;
 174   }
 175   if (n->retry_plaintext_task != GNUNET_SCHEDULER_NO_TASK)
 176   {
 177     GNUNET_SCHEDULER_cancel (n->retry_plaintext_task);
 178     n->retry_plaintext_task = GNUNET_SCHEDULER_NO_TASK;
 179   }
 180   GNUNET_assert (GNUNET_OK ==
 181                  GNUNET_CONTAINER_multihashmap_remove (neighbours,
 182                                                        &n->peer.hashPubKey, n));
 183   GNUNET_STATISTICS_set (stats, gettext_noop ("# neighbour entries allocated"),
 184                          GNUNET_CONTAINER_multihashmap_size (neighbours),
 185                          GNUNET_NO);
 186   GNUNET_free (n);
 187 }
 188
 189
 190 /**
 191  * Check if we have encrypted messages for the specified neighbour
 192  * pending, and if so, check with the transport about sending them
 193  * out.
 194  *
 195  * @param n neighbour to check.
 196  */
 197 static void
 198 process_queue (struct Neighbour *n);
 199
 200
 201 /**
 202  * Function called when the transport service is ready to receive a
 203  * message for the respective peer
 204  *
 205  * @param cls neighbour to use message from
 206  * @param size number of bytes we can transmit
 207  * @param buf where to copy the message
 208  * @return number of bytes transmitted
 209  */
 210 static size_t
 211 transmit_ready (void *cls, size_t size, void *buf)
 212 {
 213   struct Neighbour *n = cls;
 214   struct MessageEntry *m;
 215   size_t ret;
 216   char *cbuf;
 217
 218   n->th = NULL;
 219   m = n->message_head;
 220   if (m == NULL)
 221   {
 222     GNUNET_break (0);
 223     return 0;
 224   }
 225   GNUNET_CONTAINER_DLL_remove (n->encrypted_head, n->encrypted_tail, m);
 226   if (buf == NULL)
 227   {
 228 #if DEBUG_CORE
 229     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 230                 "Transmission of message of type %u and size %u failed\n",
 231                 (unsigned int)
 232                 ntohs (((struct GNUNET_MessageHeader *) &m[1])->type),
 233                 (unsigned int) m->size);
 234 #endif
 235     GNUNET_free (m);
 236     process_queue (n);
 237     return 0;
 238   }
 239   ret = 0;
 240   cbuf = buf;
 241   GNUNET_assert (size >= m->size);
 242   memcpy (cbuf, &m[1], m->size);
 243   ret = m->size;
 244   GNUNET_BANDWIDTH_tracker_consume (&n->available_send_window, m->size);
 245 #if DEBUG_CORE
 246   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 247               "Copied message of type %u and size %u into transport buffer for `%4s'\n",
 248               (unsigned int)
 249               ntohs (((struct GNUNET_MessageHeader *) &m[1])->type),
 250               (unsigned int) ret, GNUNET_i2s (&n->peer));
 251 #endif
 252   GNUNET_free (m);
 253   process_queue (n);
 254   GNUNET_STATISTICS_update (GSC_stats,
 255                             gettext_noop
 256                             ("# encrypted bytes given to transport"), ret,
 257                             GNUNET_NO);
 258   return ret;
 259 }
 260
 261
 262 /**
 263  * Check if we have messages for the specified neighbour pending, and
 264  * if so, check with the transport about sending them out.
 265  *
 266  * @param n neighbour to check.
 267  */
 268 static void
 269 process_queue (struct Neighbour *n)
 270 {
 271   struct MessageEntry *m;
 272
 273   if (n->th != NULL)
 274     return;                     /* request already pending */
 275   m = n->message_head;
 276   if (m == NULL)
 277   {
 278     /* notify sessions that the queue is empty and more messages
 279        could thus be queued now */
 280     GSC_SESSIONS_solicit (&n->peer);
 281     return;
 282   }
 283 #if DEBUG_CORE > 1
 284   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 285               "Asking transport for transmission of %u bytes to `%4s' in next %llu ms\n",
 286               (unsigned int) m->size, GNUNET_i2s (&n->peer),
 287               (unsigned long long)
 288               GNUNET_TIME_absolute_get_remaining (m->deadline).rel_value);
 289 #endif
 290   n->th =
 291        GNUNET_TRANSPORT_notify_transmit_ready (transport, &n->peer, m->size,
 292                                                m->priority,
 293                                                GNUNET_TIME_absolute_get_remaining
 294                                                (m->deadline),
 295                                                &transmit_ready,
 296                                                n);
 297   if (n->th != NULL)
 298     return;
 299   /* message request too large or duplicate request */
 300   GNUNET_break (0);
 301   /* discard encrypted message */
 302   GNUNET_CONTAINER_DLL_remove (n->encrypted_head, n->encrypted_tail, m);
 303   GNUNET_free (m);
 304   process_queue (n);
 305 }
 306
 307
 308
 309 /**
 310  * Function called by transport to notify us that
 311  * a peer connected to us (on the network level).
 312  *
 313  * @param cls closure
 314  * @param peer the peer that connected
 315  * @param ats performance data
 316  * @param ats_count number of entries in ats (excluding 0-termination)
 317  */
 318 static void
 319 handle_transport_notify_connect (void *cls,
 320                                  const struct GNUNET_PeerIdentity *peer,
 321                                  const struct GNUNET_TRANSPORT_ATS_Information
 322                                  *ats, uint32_t ats_count)
 323 {
 324   struct Neighbour *n;
 325
 326   if (0 == memcmp (peer, &my_identity, sizeof (struct GNUNET_PeerIdentity)))
 327   {
 328     GNUNET_break (0);
 329     return;
 330   }
 331   n = find_neighbour (peer);
 332   if (n != NULL)
 333   {
 334     /* duplicate connect notification!? */
 335     GNUNET_break (0);
 336     return;
 337   }
 338 #if DEBUG_CORE
 339   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Received connection from `%4s'.\n",
 340               GNUNET_i2s (peer));
 341 #endif
 342   n = GNUNET_malloc (sizeof (struct Neighbour));
 343   n->peer = *pid;
 344   GNUNET_BANDWIDTH_tracker_init (&n->available_send_window,
 345                                  GNUNET_CONSTANTS_DEFAULT_BW_IN_OUT,
 346                                  MAX_WINDOW_TIME_S);
 347   GNUNET_BANDWIDTH_tracker_init (&n->available_recv_window,
 348                                  GNUNET_CONSTANTS_DEFAULT_BW_IN_OUT,
 349                                  MAX_WINDOW_TIME_S);
 350   GNUNET_assert (GNUNET_OK ==
 351                  GNUNET_CONTAINER_multihashmap_put (neighbours,
 352                                                     &n->peer.hashPubKey, n,
 353                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 354   GNUNET_STATISTICS_set (stats, gettext_noop ("# neighbour entries allocated"),
 355                          GNUNET_CONTAINER_multihashmap_size (neighbours),
 356                          GNUNET_NO);
 357   GNUNET_TRANSPORT_set_quota (transport, peer,
 358                               GNUNET_CONSTANTS_DEFAULT_BW_IN_OUT,
 359                               GNUNET_CONSTANTS_DEFAULT_BW_IN_OUT);
 360   n->kx = GSC_KX_start (pid);
 361 }
 362
 363
 364 /**
 365  * Function called by transport telling us that a peer
 366  * disconnected.
 367  *
 368  * @param cls closure
 369  * @param peer the peer that disconnected
 370  */
 371 static void
 372 handle_transport_notify_disconnect (void *cls,
 373                                     const struct GNUNET_PeerIdentity *peer)
 374 {
 375   struct Neighbour *n;
 376
 377 #if DEBUG_CORE
 378   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 379               "Peer `%4s' disconnected from us; received notification from transport.\n",
 380               GNUNET_i2s (peer));
 381 #endif
 382   n = find_neighbour (peer);
 383   if (n == NULL)
 384   {
 385     GNUNET_break (0);
 386     return;
 387   }
 388   free_neighbour (n);
 389 }
 390
 391
 392 /**
 393  * Function called by the transport for each received message.
 394  *
 395  * @param cls closure
 396  * @param peer (claimed) identity of the other peer
 397  * @param message the message
 398  * @param ats performance data
 399  * @param ats_count number of entries in ats (excluding 0-termination)
 400  */
 401 static void
 402 handle_transport_receive (void *cls, const struct GNUNET_PeerIdentity *peer,
 403                           const struct GNUNET_MessageHeader *message,
 404                           const struct GNUNET_TRANSPORT_ATS_Information *ats,
 405                           uint32_t ats_count)
 406 {
 407   struct Neighbour *n;
 408   uint16_t type;
 409
 410 #if DEBUG_CORE > 1
 411   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 412               "Received message of type %u from `%4s', demultiplexing.\n",
 413               (unsigned int) ntohs (message->type), GNUNET_i2s (peer));
 414 #endif
 415   if (0 == memcmp (peer, &my_identity, sizeof (struct GNUNET_PeerIdentity)))
 416   {
 417     GNUNET_break (0);
 418     return;
 419   }
 420   n = find_neighbour (peer);
 421   if (n == NULL)
 422   {
 423     /* received message from peer that is not connected!? */
 424     GNUNET_break (0);
 425     return;
 426   }
 427   type = ntohs (message->type);
 428   switch (type)
 429   {
 430   case GNUNET_MESSAGE_TYPE_CORE_SET_KEY:
 431     GSC_KX_handle_set_key (n->kxinfo, message, ats, ats_count);
 432     break;
 433   case GNUNET_MESSAGE_TYPE_CORE_PING:
 434     GSC_KX_handle_ping (n->kxinfo, message, ats, ats_count);
 435     break;
 436   case GNUNET_MESSAGE_TYPE_CORE_PONG:
 437     GSC_KX_handle_pong (n->kxinfo, message, ats, ats_count);
 438     break;
 439   case GNUNET_MESSAGE_TYPE_CORE_ENCRYPTED_MESSAGE:
 440     GSC_KX_handle_encrypted_message (peer,
 441                                      n->kxinfo,
 442                                      message, ats,
 443                                      ats_count);
 444     break;
 445   default:
 446     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 447                 _("Unsupported message of type %u received.\n"),
 448                 (unsigned int) type);
 449     return;
 450   }
 451 }
 452
 453
 454 /**
 455  * Transmit the given message to the given target.
 456  *
 457  * @param target peer that should receive the message (must be connected)
 458  * @param msg message to transmit
 459  * @param timeout by when should the transmission be done?
 460  */
 461 void
 462 GDS_NEIGHBOURS_transmit (const struct GNUNET_PeerIdentity *target,
 463                          const struct GNUNET_MessageHeader *msg,
 464                          struct GNUNET_TIME_Relative timeout)
 465 {
 466   struct MessageEntry *me;
 467   struct Neighbour *n;
 468   size_t msize;
 469
 470   n = find_neighbour (target);
 471   if (NULL == n)
 472   {
 473     GNUNET_break (0);
 474     return;
 475   }
 476   msize = ntohs (msg->size);
 477   me = GNUNET_malloc (sizeof (struct MessageEntry) + msize);
 478   me->deadline = GNUNET_TIME_relative_to_absolute (timeout);
 479   me->size = msize;
 480   memcpy (&me[1], msg, msize);
 481   GNUNET_CONTAINER_DLL_insert (n->message_head,
 482                                n->message_tail,
 483                                me);
 484   process_queue (n);
 485 }
 486
 487
 488 /**
 489  * Initialize neighbours subsystem.
 490  */
 491 int
 492 GSC_NEIGHBOURS_init ()
 493 {
 494   neighbours = GNUNET_CONTAINER_multihashmap_create (128);
 495   transport =
 496       GNUNET_TRANSPORT_connect (GSC_cfg,
 497                                 &GSC_my_identity, NULL,
 498                                 &handle_transport_receive,
 499                                 &handle_transport_notify_connect,
 500                                 &handle_transport_notify_disconnect);
 501   if (NULL == transport)
 502   {
 503     GNUNET_CONTAINER_multihashmap_destroy (neighbours);
 504     neighbours = NULL;
 505     return GNUNET_SYSERR;
 506   }
 507   return GNUNET_OK;
 508 }
 509
 510
 511 /**
 512  * Wrapper around 'free_neighbour'.
 513  *
 514  * @param cls unused
 515  * @param key peer identity
 516  * @param value the 'struct Neighbour' to free
 517  * @return GNUNET_OK (continue to iterate)
 518  */
 519 static int
 520 free_neighbour_helper (void *cls, const GNUNET_HashCode * key, void *value)
 521 {
 522   struct Neighbour *n = value;
 523
 524   free_neighbour (n);
 525   return GNUNET_OK;
 526 }
 527
 528
 529 /**
 530  * Shutdown neighbours subsystem.
 531  */
 532 void
 533 GSC_NEIGHBOURS_done ()
 534 {
 535   if (NULL == transport)
 536     return;
 537   GNUNET_CONTAINER_multihashmap_iterate (neighbours, &free_neighbour_helper,
 538                                          NULL);
 539   GNUNET_TRANSPORT_disconnect (transport);
 540   transport = NULL;
 541   GNUNET_CONTAINER_multihashmap_destroy (neighbours);
 542   neighbours = NULL;
 543 }
 544
 545 /* end of gnunet-service-core_neighbours.c */
 546