src/cadet/gnunet-service-cadet_core.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2017 GNUnet e.V.
   4
   5      GNUnet is free software; you can redistribute it and/or modify
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 3, or (at your
   8      option) any later version.
   9
  10      GNUnet is distributed in the hope that it will be useful, but
  11      WITHOUT ANY WARRANTY; without even the implied warranty of
  12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
  18      Boston, MA 02110-1301, USA.
  19 */
  20
  21 /**
  22  * @file cadet/gnunet-service-cadet_core.c
  23  * @brief cadet service; interaction with CORE service
  24  * @author Bartlomiej Polot
  25  * @author Christian Grothoff
  26  *
  27  * All functions in this file should use the prefix GCO (Gnunet Cadet cOre (bottom))
  28  *
  29  * TODO:
  30  * - Optimization: given BROKEN messages, destroy paths (?)
  31  */
  32 #include "platform.h"
  33 #include "gnunet-service-cadet_core.h"
  34 #include "gnunet-service-cadet_paths.h"
  35 #include "gnunet-service-cadet_peer.h"
  36 #include "gnunet-service-cadet_connection.h"
  37 #include "gnunet-service-cadet_tunnels.h"
  38 #include "gnunet_core_service.h"
  39 #include "gnunet_statistics_service.h"
  40 #include "cadet_protocol.h"
  41
  42
  43 #define LOG(level, ...) GNUNET_log_from(level,"cadet-cor",__VA_ARGS__)
  44
  45 /**
  46  * Information we keep per direction for a route.
  47  */
  48 struct RouteDirection;
  49
  50
  51 /**
  52  * Set of CadetRoutes that have exactly the same number of messages
  53  * in their buffer.  Used so we can efficiently find all of those
  54  * routes that have the current maximum of messages in the buffer (in
  55  * case we have to purge).
  56  */
  57 struct Rung
  58 {
  59
  60   /**
  61    * Rung of RouteDirections with one more buffer entry each.
  62    */
  63   struct Rung *next;
  64
  65   /**
  66    * Rung of RouteDirections with one less buffer entry each.
  67    */
  68   struct Rung *prev;
  69
  70   /**
  71    * DLL of route directions with a number of buffer entries matching this rung.
  72    */
  73   struct RouteDirection *rd_head;
  74
  75   /**
  76    * DLL of route directions with a number of buffer entries matching this rung.
  77    */
  78   struct RouteDirection *rd_tail;
  79
  80   /**
  81    * Total number of route directions in this rung.
  82    */
  83   unsigned int num_routes;
  84
  85   /**
  86    * Number of messages route directions at this rung have
  87    * in their buffer.
  88    */
  89   unsigned int rung_off;
  90 };
  91
  92
  93 /**
  94  * Information we keep per direction for a route.
  95  */
  96 struct RouteDirection
  97 {
  98
  99   /**
 100    * DLL of other route directions within the same `struct Rung`.
 101    */
 102   struct RouteDirection *prev;
 103
 104   /**
 105    * DLL of other route directions within the same `struct Rung`.
 106    */
 107   struct RouteDirection *next;
 108
 109   /**
 110    * Rung of this route direction (matches length of the buffer DLL).
 111    */
 112   struct Rung *rung;
 113
 114   /**
 115    * Head of DLL of envelopes we have in the buffer for this direction.
 116    */
 117   struct GNUNET_MQ_Envelope *env_head;
 118
 119   /**
 120    * Tail of DLL of envelopes we have in the buffer for this direction.
 121    */
 122   struct GNUNET_MQ_Envelope *env_tail;
 123
 124   /**
 125    * Target peer.
 126    */
 127   struct CadetPeer *hop;
 128
 129   /**
 130    * Route this direction is part of.
 131    */
 132   struct CadetRoute *my_route;
 133
 134   /**
 135    * Message queue manager for @e hop.
 136    */
 137   struct GCP_MessageQueueManager *mqm;
 138
 139   /**
 140    * Is @e mqm currently ready for transmission?
 141    */
 142   int is_ready;
 143
 144 };
 145
 146
 147 /**
 148  * Description of a segment of a `struct CadetConnection` at the
 149  * intermediate peers.  Routes are basically entries in a peer's
 150  * routing table for forwarding traffic.  At both endpoints, the
 151  * routes are terminated by a `struct CadetConnection`, which knows
 152  * the complete `struct CadetPath` that is formed by the individual
 153  * routes.
 154  */
 155 struct CadetRoute
 156 {
 157
 158   /**
 159    * Information about the next hop on this route.
 160    */
 161   struct RouteDirection next;
 162
 163   /**
 164    * Information about the previous hop on this route.
 165    */
 166   struct RouteDirection prev;
 167
 168   /**
 169    * Unique identifier for the connection that uses this route.
 170    */
 171   struct GNUNET_CADET_ConnectionTunnelIdentifier cid;
 172
 173   /**
 174    * When was this route last in use?
 175    */
 176   struct GNUNET_TIME_Absolute last_use;
 177
 178   /**
 179    * Position of this route in the #route_heap.
 180    */
 181   struct GNUNET_CONTAINER_HeapNode *hn;
 182
 183   /**
 184    * Options for the route, control buffering.
 185    */
 186   enum GNUNET_CADET_ChannelOption options;
 187 };
 188
 189
 190 /**
 191  * Handle to the CORE service.
 192  */
 193 static struct GNUNET_CORE_Handle *core;
 194
 195 /**
 196  * Routes on which this peer is an intermediate.
 197  */
 198 static struct GNUNET_CONTAINER_MultiShortmap *routes;
 199
 200 /**
 201  * Heap of routes, MIN-sorted by last activity.
 202  */
 203 static struct GNUNET_CONTAINER_Heap *route_heap;
 204
 205 /**
 206  * Rung zero (always pointed to by #rung_head).
 207  */
 208 static struct Rung rung_zero;
 209
 210 /**
 211  * DLL of rungs, with the head always point to a rung of
 212  * route directions with no messages in the queue.
 213  */
 214 static struct Rung *rung_head = &rung_zero;
 215
 216 /**
 217  * Tail of the #rung_head DLL.
 218  */
 219 static struct Rung *rung_tail = &rung_zero;
 220
 221 /**
 222  * Maximum number of concurrent routes this peer will support.
 223  */
 224 static unsigned long long max_routes;
 225
 226 /**
 227  * Maximum number of envelopes we will buffer at this peer.
 228  */
 229 static unsigned long long max_buffers;
 230
 231 /**
 232  * Current number of envelopes we have buffered at this peer.
 233  */
 234 static unsigned long long cur_buffers;
 235
 236 /**
 237  * Task to timeout routes.
 238  */
 239 static struct GNUNET_SCHEDULER_Task *timeout_task;
 240
 241
 242 /**
 243  * Get the route corresponding to a hash.
 244  *
 245  * @param cid hash generated from the connection identifier
 246  */
 247 static struct CadetRoute *
 248 get_route (const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid)
 249 {
 250   return GNUNET_CONTAINER_multishortmap_get (routes,
 251                                              &cid->connection_of_tunnel);
 252 }
 253
 254
 255 /**
 256  * Lower the rung in which @a dir is by 1.
 257  *
 258  * @param dir direction to lower in rung.
 259  */
 260 static void
 261 lower_rung (struct RouteDirection *dir)
 262 {
 263   struct Rung *rung = dir->rung;
 264   struct Rung *prev;
 265
 266   GNUNET_CONTAINER_DLL_remove (rung->rd_head,
 267                                rung->rd_tail,
 268                                dir);
 269   prev = rung->prev;
 270   GNUNET_assert (NULL != prev);
 271   if (prev->rung_off != rung->rung_off - 1)
 272   {
 273     prev = GNUNET_new (struct Rung);
 274     prev->rung_off = rung->rung_off - 1;
 275     GNUNET_CONTAINER_DLL_insert_after (rung_head,
 276                                        rung_tail,
 277                                        rung->prev,
 278                                        prev);
 279   }
 280   GNUNET_assert (NULL != prev);
 281   GNUNET_CONTAINER_DLL_insert (prev->rd_head,
 282                                prev->rd_tail,
 283                                dir);
 284   dir->rung = prev;
 285 }
 286
 287
 288 /**
 289  * Discard the buffer @a env from the route direction @a dir and
 290  * move @a dir down a rung.
 291  *
 292  * @param dir direction that contains the @a env in the buffer
 293  * @param env envelope to discard
 294  */
 295 static void
 296 discard_buffer (struct RouteDirection *dir,
 297                 struct GNUNET_MQ_Envelope *env)
 298 {
 299   GNUNET_MQ_dll_remove (&dir->env_head,
 300                         &dir->env_tail,
 301                         env);
 302   cur_buffers--;
 303   GNUNET_MQ_discard (env);
 304   lower_rung (dir);
 305   GNUNET_STATISTICS_set (stats,
 306                          "# buffer use",
 307                          cur_buffers,
 308                          GNUNET_NO);
 309 }
 310
 311
 312 /**
 313  * Discard all messages from the highest rung, to make space.
 314  */
 315 static void
 316 discard_all_from_rung_tail ()
 317 {
 318   struct Rung *tail = rung_tail;
 319   struct RouteDirection *dir;
 320
 321   while (NULL != (dir = tail->rd_head))
 322   {
 323     LOG (GNUNET_ERROR_TYPE_DEBUG,
 324          "Queue full due new message %s on connection %s, dropping old message\n",
 325          GNUNET_sh2s (&dir->my_route->cid.connection_of_tunnel));
 326     GNUNET_STATISTICS_update (stats,
 327                               "# messages dropped due to full buffer",
 328                               1,
 329                               GNUNET_NO);
 330     discard_buffer (dir,
 331                     dir->env_head);
 332   }
 333   GNUNET_CONTAINER_DLL_remove (rung_head,
 334                                rung_tail,
 335                                tail);
 336   GNUNET_free (tail);
 337 }
 338
 339
 340 /**
 341  * We message @a msg from @a prev.  Find its route by @a cid and
 342  * forward to the next hop.  Drop and signal broken route if we do not
 343  * have a route.
 344  *
 345  * @param prev previous hop (sender)
 346  * @param cid connection identifier, tells us which route to use
 347  * @param msg the message to forward
 348  */
 349 static void
 350 route_message (struct CadetPeer *prev,
 351                const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid,
 352                const struct GNUNET_MessageHeader *msg)
 353 {
 354   struct CadetRoute *route;
 355   struct RouteDirection *dir;
 356   struct Rung *rung;
 357   struct Rung *nxt;
 358   struct GNUNET_MQ_Envelope *env;
 359
 360   route = get_route (cid);
 361   if (NULL == route)
 362   {
 363     struct GNUNET_MQ_Envelope *env;
 364     struct GNUNET_CADET_ConnectionBrokenMessage *bm;
 365
 366     LOG (GNUNET_ERROR_TYPE_DEBUG,
 367          "Failed to route message of type %u from %s on connection %s: no route\n",
 368          ntohs (msg->type),
 369          GCP_2s (prev),
 370          GNUNET_sh2s (&cid->connection_of_tunnel));
 371     switch (ntohs (msg->type))
 372     {
 373     case GNUNET_MESSAGE_TYPE_CADET_CONNECTION_DESTROY:
 374     case GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN:
 375       /* No need to respond to these! */
 376       return;
 377     }
 378     env = GNUNET_MQ_msg (bm,
 379                          GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN);
 380     bm->cid = *cid;
 381     bm->peer1 = my_full_id;
 382     GCP_send_ooo (prev,
 383                   env);
 384     return;
 385   }
 386   route->last_use = GNUNET_TIME_absolute_get ();
 387   GNUNET_CONTAINER_heap_update_cost (route->hn,
 388                                      route->last_use.abs_value_us);
 389   dir = (prev == route->prev.hop) ? &route->next : &route->prev;
 390   if (GNUNET_YES == dir->is_ready)
 391   {
 392     LOG (GNUNET_ERROR_TYPE_DEBUG,
 393          "Routing message of type %u from %s to %s on connection %s\n",
 394          ntohs (msg->type),
 395          GCP_2s (prev),
 396          GNUNET_i2s (GCP_get_id (dir->hop)),
 397          GNUNET_sh2s (&cid->connection_of_tunnel));
 398     dir->is_ready = GNUNET_NO;
 399     GCP_send (dir->mqm,
 400               GNUNET_MQ_msg_copy (msg));
 401     return;
 402   }
 403   /* Check if buffering is disallowed, and if so, make sure we only queue
 404      one message per direction. */
 405   if ( (0 != (route->options & GNUNET_CADET_OPTION_NOBUFFER)) &&
 406        (NULL != dir->env_head) )
 407     discard_buffer (dir,
 408                     dir->env_head);
 409   rung = dir->rung;
 410   if (cur_buffers == max_buffers)
 411   {
 412     /* Need to make room. */
 413     if (NULL != rung->next)
 414     {
 415       /* Easy case, drop messages from route directions in highest rung */
 416       discard_all_from_rung_tail ();
 417     }
 418     else
 419     {
 420       /* We are in the highest rung, drop our own! */
 421       LOG (GNUNET_ERROR_TYPE_DEBUG,
 422            "Queue full due new message %s on connection %s, dropping old message\n",
 423            GNUNET_sh2s (&dir->my_route->cid.connection_of_tunnel));
 424       GNUNET_STATISTICS_update (stats,
 425                                 "# messages dropped due to full buffer",
 426                                 1,
 427                                 GNUNET_NO);
 428       discard_buffer (dir,
 429                       dir->env_head);
 430       rung = dir->rung;
 431     }
 432   }
 433   /* remove 'dir' from current rung */
 434   GNUNET_CONTAINER_DLL_remove (rung->rd_head,
 435                                rung->rd_tail,
 436                                dir);
 437   /* make 'nxt' point to the next higher rung, creat if necessary */
 438   nxt = rung->next;
 439   if ( (NULL == nxt) ||
 440        (rung->rung_off + 1 != nxt->rung_off) )
 441   {
 442     nxt = GNUNET_new (struct Rung);
 443     nxt->rung_off = rung->rung_off + 1;
 444     GNUNET_CONTAINER_DLL_insert_after (rung_head,
 445                                        rung_tail,
 446                                        rung,
 447                                        nxt);
 448   }
 449   /* insert 'dir' into next higher rung */
 450   GNUNET_CONTAINER_DLL_insert (nxt->rd_head,
 451                                nxt->rd_tail,
 452                                dir);
 453   dir->rung = nxt;
 454
 455   /* add message into 'dir' buffer */
 456   LOG (GNUNET_ERROR_TYPE_DEBUG,
 457        "Queueing new message of type %u from %s to %s on connection %s\n",
 458        ntohs (msg->type),
 459        GCP_2s (prev),
 460        GNUNET_i2s (GCP_get_id (dir->hop)),
 461        GNUNET_sh2s (&cid->connection_of_tunnel));
 462   env = GNUNET_MQ_msg_copy (msg);
 463   GNUNET_MQ_dll_insert_tail (&dir->env_head,
 464                              &dir->env_tail,
 465                              env);
 466   cur_buffers++;
 467   GNUNET_STATISTICS_set (stats,
 468                          "# buffer use",
 469                          cur_buffers,
 470                          GNUNET_NO);
 471   /* Clean up 'rung' if now empty (and not head) */
 472   if ( (NULL == rung->rd_head) &&
 473        (rung != rung_head) )
 474   {
 475     GNUNET_CONTAINER_DLL_remove (rung_head,
 476                                  rung_tail,
 477                                  rung);
 478     GNUNET_free (rung);
 479   }
 480 }
 481
 482
 483 /**
 484  * Check if the create_connection message has the appropriate size.
 485  *
 486  * @param cls Closure (unused).
 487  * @param msg Message to check.
 488  *
 489  * @return #GNUNET_YES if size is correct, #GNUNET_NO otherwise.
 490  */
 491 static int
 492 check_connection_create (void *cls,
 493                          const struct GNUNET_CADET_ConnectionCreateMessage *msg)
 494 {
 495   uint16_t size = ntohs (msg->header.size) - sizeof (*msg);
 496
 497   if (0 != (size % sizeof (struct GNUNET_PeerIdentity)))
 498   {
 499     GNUNET_break_op (0);
 500     return GNUNET_NO;
 501   }
 502   return GNUNET_YES;
 503 }
 504
 505
 506 /**
 507  * Free internal data of a route direction.
 508  *
 509  * @param dir direction to destroy (do NOT free memory of 'dir' itself)
 510  */
 511 static void
 512 destroy_direction (struct RouteDirection *dir)
 513 {
 514   struct GNUNET_MQ_Envelope *env;
 515
 516   while (NULL != (env = dir->env_head))
 517   {
 518     GNUNET_STATISTICS_update (stats,
 519                               "# messages dropped due to route destruction",
 520                               1,
 521                               GNUNET_NO);
 522     discard_buffer (dir,
 523                     env);
 524   }
 525   if (NULL != dir->mqm)
 526   {
 527     GCP_request_mq_cancel (dir->mqm,
 528                            NULL);
 529     dir->mqm = NULL;
 530   }
 531   GNUNET_CONTAINER_DLL_remove (rung_head->rd_head,
 532                                rung_head->rd_tail,
 533                                dir);
 534 }
 535
 536
 537 /**
 538  * Destroy our state for @a route.
 539  *
 540  * @param route route to destroy
 541  */
 542 static void
 543 destroy_route (struct CadetRoute *route)
 544 {
 545   LOG (GNUNET_ERROR_TYPE_DEBUG,
 546        "Destroying route from %s to %s of connection %s\n",
 547        GNUNET_i2s  (GCP_get_id (route->prev.hop)),
 548        GNUNET_i2s2 (GCP_get_id (route->next.hop)),
 549        GNUNET_sh2s (&route->cid.connection_of_tunnel));
 550   GNUNET_assert (route ==
 551                  GNUNET_CONTAINER_heap_remove_node (route->hn));
 552   GNUNET_assert (GNUNET_YES ==
 553                  GNUNET_CONTAINER_multishortmap_remove (routes,
 554                                                         &route->cid.connection_of_tunnel,
 555                                                         route));
 556   GNUNET_STATISTICS_set (stats,
 557                          "# routes",
 558                          GNUNET_CONTAINER_multishortmap_size (routes),
 559                          GNUNET_NO);
 560   destroy_direction (&route->prev);
 561   destroy_direction (&route->next);
 562   GNUNET_free (route);
 563 }
 564
 565
 566 /**
 567  * Send message that a route is broken between @a peer1 and @a peer2.
 568  *
 569  * @param target where to send the message
 570  * @param cid connection identifier to use
 571  * @param peer1 one of the peers where a link is broken
 572  * @param peer2 another one of the peers where a link is broken
 573  */
 574 static void
 575 send_broken (struct RouteDirection *target,
 576              const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid,
 577              const struct GNUNET_PeerIdentity *peer1,
 578              const struct GNUNET_PeerIdentity *peer2)
 579 {
 580   struct GNUNET_MQ_Envelope *env;
 581   struct GNUNET_CADET_ConnectionBrokenMessage *bm;
 582
 583   if (NULL == target->mqm)
 584     return; /* Can't send notification, connection is down! */
 585   LOG (GNUNET_ERROR_TYPE_DEBUG,
 586        "Notifying %s about BROKEN route at %s-%s of connection %s\n",
 587        GCP_2s (target->hop),
 588        GNUNET_i2s (peer1),
 589        GNUNET_i2s2 (peer2),
 590        GNUNET_sh2s (&cid->connection_of_tunnel));
 591
 592   env = GNUNET_MQ_msg (bm,
 593                        GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN);
 594   bm->cid = *cid;
 595   if (NULL != peer1)
 596     bm->peer1 = *peer1;
 597   if (NULL != peer2)
 598     bm->peer2 = *peer2;
 599   GCP_request_mq_cancel (target->mqm,
 600                          env);
 601   target->mqm = NULL;
 602 }
 603
 604
 605 /**
 606  * Function called to check if any routes have timed out, and if
 607  * so, to clean them up.  Finally, schedules itself again at the
 608  * earliest time where there might be more work.
 609  *
 610  * @param cls NULL
 611  */
 612 static void
 613 timeout_cb (void *cls)
 614 {
 615   struct CadetRoute *r;
 616   struct GNUNET_TIME_Relative linger;
 617   struct GNUNET_TIME_Absolute exp;
 618
 619   timeout_task = NULL;
 620   linger = GNUNET_TIME_relative_multiply (keepalive_period,
 621                                           3);
 622   while (NULL != (r = GNUNET_CONTAINER_heap_peek (route_heap)))
 623   {
 624     exp = GNUNET_TIME_absolute_add (r->last_use,
 625                                     linger);
 626     if (0 != GNUNET_TIME_absolute_get_duration (exp).rel_value_us)
 627     {
 628       /* Route not yet timed out, wait until it does. */
 629       timeout_task = GNUNET_SCHEDULER_add_at (exp,
 630                                               &timeout_cb,
 631                                               NULL);
 632       return;
 633     }
 634     send_broken (&r->prev,
 635                  &r->cid,
 636                  NULL,
 637                  NULL);
 638     send_broken (&r->next,
 639                  &r->cid,
 640                  NULL,
 641                  NULL);
 642     destroy_route (r);
 643   }
 644   /* No more routes left, so no need for a #timeout_task */
 645 }
 646
 647
 648 /**
 649  * Function called when the message queue to the previous hop
 650  * becomes available/unavailable.  We expect this function to
 651  * be called immediately when we register, and then again
 652  * later if the connection ever goes down.
 653  *
 654  * @param cls the `struct RouteDirection`
 655  * @param available #GNUNET_YES if sending is now possible,
 656  *                  #GNUNET_NO if sending is no longer possible
 657  *                  #GNUNET_SYSERR if sending is no longer possible
 658  *                                 and the last envelope was discarded
 659  */
 660 static void
 661 dir_ready_cb (void *cls,
 662               int ready)
 663 {
 664   struct RouteDirection *dir = cls;
 665   struct CadetRoute *route = dir->my_route;
 666   struct RouteDirection *odir;
 667
 668   if (GNUNET_YES == ready)
 669   {
 670     struct GNUNET_MQ_Envelope *env;
 671
 672     dir->is_ready = GNUNET_YES;
 673     if (NULL != (env = dir->env_head))
 674     {
 675       GNUNET_MQ_dll_remove (&dir->env_head,
 676                             &dir->env_tail,
 677                             env);
 678       cur_buffers--;
 679       GNUNET_STATISTICS_set (stats,
 680                              "# buffer use",
 681                              cur_buffers,
 682                              GNUNET_NO);
 683       lower_rung (dir);
 684       dir->is_ready = GNUNET_NO;
 685       GCP_send (dir->mqm,
 686                 env);
 687     }
 688     return;
 689   }
 690   odir = (dir == &route->next) ? &route->prev : &route->next;
 691   send_broken (&route->next,
 692                &route->cid,
 693                GCP_get_id (odir->hop),
 694                &my_full_id);
 695   destroy_route (route);
 696 }
 697
 698
 699 /**
 700  * Initialize one of the directions of a route.
 701  *
 702  * @param route route the direction belongs to
 703  * @param dir direction to initialize
 704  * @param hop next hop on in the @a dir
 705  */
 706 static void
 707 dir_init (struct RouteDirection *dir,
 708           struct CadetRoute *route,
 709           struct CadetPeer *hop)
 710 {
 711   dir->hop = hop;
 712   dir->my_route = route;
 713   dir->mqm = GCP_request_mq (hop,
 714                              &dir_ready_cb,
 715                              dir);
 716   GNUNET_CONTAINER_DLL_insert (rung_head->rd_head,
 717                                rung_head->rd_tail,
 718                                dir);
 719   dir->rung = rung_head;
 720   GNUNET_assert (GNUNET_YES == dir->is_ready);
 721 }
 722
 723
 724 /**
 725  * We could not create the desired route.  Send a
 726  * #GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN
 727  * message to @a target.
 728  *
 729  * @param target who should receive the message
 730  * @param cid identifier of the connection/route that failed
 731  * @param failure_at neighbour with which we failed to route,
 732  *        or NULL.
 733  */
 734 static void
 735 send_broken_without_mqm (struct CadetPeer *target,
 736                          const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid,
 737                          const struct GNUNET_PeerIdentity *failure_at)
 738 {
 739   struct GNUNET_MQ_Envelope *env;
 740   struct GNUNET_CADET_ConnectionBrokenMessage *bm;
 741
 742   env = GNUNET_MQ_msg (bm,
 743                        GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN);
 744   bm->cid = *cid;
 745   bm->peer1 = my_full_id;
 746   if (NULL != failure_at)
 747     bm->peer2 = *failure_at;
 748   GCP_send_ooo (target,
 749                 env);
 750 }
 751
 752
 753 /**
 754  * Handle for #GNUNET_MESSAGE_TYPE_CADET_CONNECTION_CREATE
 755  *
 756  * @param cls Closure (CadetPeer for neighbor that sent the message).
 757  * @param msg Message itself.
 758  */
 759 static void
 760 handle_connection_create (void *cls,
 761                           const struct GNUNET_CADET_ConnectionCreateMessage *msg)
 762 {
 763   struct CadetPeer *sender = cls;
 764   struct CadetPeer *next;
 765   const struct GNUNET_PeerIdentity *pids = (const struct GNUNET_PeerIdentity *) &msg[1];
 766   struct CadetRoute *route;
 767   uint16_t size = ntohs (msg->header.size) - sizeof (*msg);
 768   unsigned int path_length;
 769   unsigned int off;
 770   enum GNUNET_CADET_ChannelOption options;
 771
 772   options = (enum GNUNET_CADET_ChannelOption) ntohl (msg->options);
 773   path_length = size / sizeof (struct GNUNET_PeerIdentity);
 774   /* Initiator is at offset 0. */
 775   for (off=1;off<path_length;off++)
 776     if (0 == memcmp (&my_full_id,
 777                      &pids[off],
 778                      sizeof (struct GNUNET_PeerIdentity)))
 779       break;
 780   if (off == path_length)
 781   {
 782     /* We are not on the path, bogus request */
 783     GNUNET_break_op (0);
 784     return;
 785   }
 786   /* Check previous hop */
 787   if (sender != GCP_get (&pids[off - 1],
 788                          GNUNET_NO))
 789   {
 790     /* sender is not on the path, not allowed */
 791     GNUNET_break_op (0);
 792     return;
 793   }
 794   if (NULL !=
 795       get_route (&msg->cid))
 796   {
 797     /* Duplicate CREATE, pass it on, previous one might have been lost! */
 798     LOG (GNUNET_ERROR_TYPE_DEBUG,
 799          "Passing on duplicate CADET_CONNECTION_CREATE message on connection %s\n",
 800          GNUNET_sh2s (&msg->cid.connection_of_tunnel));
 801     route_message (sender,
 802                    &msg->cid,
 803                    &msg->header);
 804     return;
 805   }
 806   if (off == path_length - 1)
 807   {
 808     /* We are the destination, create connection */
 809     struct CadetConnection *cc;
 810     struct CadetPeerPath *path;
 811     struct CadetPeer *origin;
 812
 813     cc = GCC_lookup (&msg->cid);
 814     if (NULL != cc)
 815     {
 816       LOG (GNUNET_ERROR_TYPE_DEBUG,
 817            "Received duplicate CADET_CONNECTION_CREATE message on connection %s\n",
 818            GNUNET_sh2s (&msg->cid.connection_of_tunnel));
 819       GCC_handle_duplicate_create (cc);
 820       return;
 821     }
 822
 823     origin = GCP_get (&pids[0],
 824                       GNUNET_YES);
 825     LOG (GNUNET_ERROR_TYPE_DEBUG,
 826          "Received CADET_CONNECTION_CREATE message from %s for connection %s, building inverse path\n",
 827          GCP_2s (origin),
 828          GNUNET_sh2s (&msg->cid.connection_of_tunnel));
 829     path = GCPP_get_path_from_route (path_length - 1,
 830                                      pids);
 831     if (GNUNET_OK !=
 832         GCT_add_inbound_connection (GCP_get_tunnel (origin,
 833                                                     GNUNET_YES),
 834                                     &msg->cid,
 835                                     (enum GNUNET_CADET_ChannelOption) ntohl (msg->options),
 836                                     path))
 837     {
 838       /* Send back BROKEN: duplicate connection on the same path,
 839          we will use the other one. */
 840       LOG (GNUNET_ERROR_TYPE_DEBUG,
 841            "Received CADET_CONNECTION_CREATE from %s for %s, but %s already has a connection. Sending BROKEN\n",
 842            GCP_2s (sender),
 843            GNUNET_sh2s (&msg->cid.connection_of_tunnel),
 844            GCPP_2s (path));
 845       send_broken_without_mqm (sender,
 846                                &msg->cid,
 847                                NULL);
 848       return;
 849     }
 850     return;
 851   }
 852   /* We are merely a hop on the way, check if we can support the route */
 853   next = GCP_get (&pids[off + 1],
 854                   GNUNET_NO);
 855   if ( (NULL == next) ||
 856        (GNUNET_NO == GCP_has_core_connection (next)) )
 857   {
 858     /* unworkable, send back BROKEN notification */
 859     LOG (GNUNET_ERROR_TYPE_DEBUG,
 860          "Received CADET_CONNECTION_CREATE from %s for %s. Next hop %s:%u is down. Sending BROKEN\n",
 861          GCP_2s (sender),
 862          GNUNET_sh2s (&msg->cid.connection_of_tunnel),
 863          GNUNET_i2s (&pids[off + 1]),
 864          off + 1);
 865     send_broken_without_mqm (sender,
 866                              &msg->cid,
 867                              &pids[off + 1]);
 868     return;
 869   }
 870   if (max_routes <= GNUNET_CONTAINER_multishortmap_size (routes))
 871   {
 872     LOG (GNUNET_ERROR_TYPE_DEBUG,
 873          "Received CADET_CONNECTION_CREATE from %s for %s. We have reached our route limit. Sending BROKEN\n",
 874          GCP_2s (sender),
 875          GNUNET_sh2s (&msg->cid.connection_of_tunnel));
 876     send_broken_without_mqm (sender,
 877                              &msg->cid,
 878                              &pids[off - 1]);
 879     return;
 880   }
 881
 882   /* Workable route, create routing entry */
 883   LOG (GNUNET_ERROR_TYPE_DEBUG,
 884        "Received CADET_CONNECTION_CREATE from %s for %s. Next hop %s:%u is up. Creating route\n",
 885        GCP_2s (sender),
 886        GNUNET_sh2s (&msg->cid.connection_of_tunnel),
 887        GNUNET_i2s (&pids[off + 1]),
 888        off + 1);
 889   route = GNUNET_new (struct CadetRoute);
 890   route->options = options;
 891   route->cid = msg->cid;
 892   route->last_use = GNUNET_TIME_absolute_get ();
 893   dir_init (&route->prev,
 894             route,
 895             sender);
 896   dir_init (&route->next,
 897             route,
 898             next);
 899   GNUNET_assert (GNUNET_OK ==
 900                  GNUNET_CONTAINER_multishortmap_put (routes,
 901                                                      &route->cid.connection_of_tunnel,
 902                                                      route,
 903                                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 904   GNUNET_STATISTICS_set (stats,
 905                          "# routes",
 906                          GNUNET_CONTAINER_multishortmap_size (routes),
 907                          GNUNET_NO);
 908   route->hn = GNUNET_CONTAINER_heap_insert (route_heap,
 909                                             route,
 910                                             route->last_use.abs_value_us);
 911   if (NULL == timeout_task)
 912     timeout_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply (keepalive_period,
 913                                                                                 3),
 914                                                  &timeout_cb,
 915                                                  NULL);
 916 }
 917
 918
 919 /**
 920  * Handle for #GNUNET_MESSAGE_TYPE_CADET_CONNECTION_CREATE_ACK
 921  *
 922  * @param cls Closure (CadetPeer for neighbor that sent the message).
 923  * @param msg Message itself.
 924  */
 925 static void
 926 handle_connection_create_ack (void *cls,
 927                               const struct GNUNET_CADET_ConnectionCreateAckMessage *msg)
 928 {
 929   struct CadetPeer *peer = cls;
 930   struct CadetConnection *cc;
 931
 932   /* First, check if ACK belongs to a connection that ends here. */
 933   cc = GCC_lookup (&msg->cid);
 934   if (NULL != cc)
 935   {
 936     /* verify ACK came from the right direction */
 937     struct CadetPeerPath *path = GCC_get_path (cc);
 938
 939     if (peer !=
 940         GCPP_get_peer_at_offset (path,
 941                                  0))
 942     {
 943       /* received ACK from unexpected direction, ignore! */
 944       GNUNET_break_op (0);
 945       return;
 946     }
 947     LOG (GNUNET_ERROR_TYPE_DEBUG,
 948          "Received CONNECTION_CREATE_ACK for connection %s.\n",
 949          GNUNET_sh2s (&msg->cid.connection_of_tunnel));
 950     GCC_handle_connection_create_ack (cc);
 951     return;
 952   }
 953
 954   /* We're just an intermediary peer, route the message along its path */
 955   route_message (peer,
 956                  &msg->cid,
 957                  &msg->header);
 958 }
 959
 960
 961 /**
 962  * Handle for #GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN
 963  *
 964  * @param cls Closure (CadetPeer for neighbor that sent the message).
 965  * @param msg Message itself.
 966  * @deprecated duplicate logic with #handle_destroy(); dedup!
 967  */
 968 static void
 969 handle_connection_broken (void *cls,
 970                           const struct GNUNET_CADET_ConnectionBrokenMessage *msg)
 971 {
 972   struct CadetPeer *peer = cls;
 973   struct CadetConnection *cc;
 974   struct CadetRoute *route;
 975
 976   /* First, check if message belongs to a connection that ends here. */
 977   cc = GCC_lookup (&msg->cid);
 978   if (NULL != cc)
 979   {
 980     /* verify message came from the right direction */
 981     struct CadetPeerPath *path = GCC_get_path (cc);
 982
 983     if (peer !=
 984         GCPP_get_peer_at_offset (path,
 985                                  0))
 986     {
 987       /* received message from unexpected direction, ignore! */
 988       GNUNET_break_op (0);
 989       return;
 990     }
 991     LOG (GNUNET_ERROR_TYPE_DEBUG,
 992          "Received CONNECTION_BROKEN for connection %s. Destroying it.\n",
 993          GNUNET_sh2s (&msg->cid.connection_of_tunnel));
 994     GCC_destroy_without_core (cc);
 995
 996     /* FIXME: also destroy the path up to the specified link! */
 997     return;
 998   }
 999
1000   /* We're just an intermediary peer, route the message along its path */
1001   route_message (peer,
1002                  &msg->cid,
1003                  &msg->header);
1004   route = get_route (&msg->cid);
1005   if (NULL != route)
1006     destroy_route (route);
1007   /* FIXME: also destroy paths we MAY have up to the specified link! */
1008 }
1009
1010
1011 /**
1012  * Handle for #GNUNET_MESSAGE_TYPE_CADET_CONNECTION_DESTROY
1013  *
1014  * @param cls Closure (CadetPeer for neighbor that sent the message).
1015  * @param msg Message itself.
1016  */
1017 static void
1018 handle_connection_destroy (void *cls,
1019                            const struct GNUNET_CADET_ConnectionDestroyMessage *msg)
1020 {
1021   struct CadetPeer *peer = cls;
1022   struct CadetConnection *cc;
1023   struct CadetRoute *route;
1024
1025   /* First, check if message belongs to a connection that ends here. */
1026   cc = GCC_lookup (&msg->cid);
1027   if (NULL != cc)
1028   {
1029     /* verify message came from the right direction */
1030     struct CadetPeerPath *path = GCC_get_path (cc);
1031
1032     if (peer !=
1033         GCPP_get_peer_at_offset (path,
1034                                  0))
1035     {
1036       /* received message from unexpected direction, ignore! */
1037       GNUNET_break_op (0);
1038       return;
1039     }
1040     LOG (GNUNET_ERROR_TYPE_DEBUG,
1041          "Received CONNECTION_DESTROY for connection %s. Destroying connection.\n",
1042          GNUNET_sh2s (&msg->cid.connection_of_tunnel));
1043
1044     GCC_destroy_without_core (cc);
1045     return;
1046   }
1047
1048   /* We're just an intermediary peer, route the message along its path */
1049   LOG (GNUNET_ERROR_TYPE_DEBUG,
1050        "Received CONNECTION_DESTROY for connection %s. Destroying route.\n",
1051        GNUNET_sh2s (&msg->cid.connection_of_tunnel));
1052   route_message (peer,
1053                  &msg->cid,
1054                  &msg->header);
1055   route = get_route (&msg->cid);
1056   if (NULL != route)
1057     destroy_route (route);
1058 }
1059
1060
1061 /**
1062  * Handle for #GNUNET_MESSAGE_TYPE_CADET_TUNNEL_KX
1063  *
1064  * @param cls Closure (CadetPeer for neighbor that sent the message).
1065  * @param msg Message itself.
1066  */
1067 static void
1068 handle_tunnel_kx (void *cls,
1069                   const struct GNUNET_CADET_TunnelKeyExchangeMessage *msg)
1070 {
1071   struct CadetPeer *peer = cls;
1072   struct CadetConnection *cc;
1073
1074   /* First, check if message belongs to a connection that ends here. */
1075   cc = GCC_lookup (&msg->cid);
1076   if (NULL != cc)
1077   {
1078     /* verify message came from the right direction */
1079     struct CadetPeerPath *path = GCC_get_path (cc);
1080
1081     if (peer !=
1082         GCPP_get_peer_at_offset (path,
1083                                  0))
1084     {
1085       /* received message from unexpected direction, ignore! */
1086       GNUNET_break_op (0);
1087       return;
1088     }
1089     GCC_handle_kx (cc,
1090                    msg);
1091     return;
1092   }
1093
1094   /* We're just an intermediary peer, route the message along its path */
1095   route_message (peer,
1096                  &msg->cid,
1097                  &msg->header);
1098 }
1099
1100
1101 /**
1102  * Handle for #GNUNET_MESSAGE_TYPE_CADET_TUNNEL_KX_AUTH
1103  *
1104  * @param cls Closure (CadetPeer for neighbor that sent the message).
1105  * @param msg Message itself.
1106  */
1107 static void
1108 handle_tunnel_kx_auth (void *cls,
1109                        const struct GNUNET_CADET_TunnelKeyExchangeAuthMessage *msg)
1110 {
1111   struct CadetPeer *peer = cls;
1112   struct CadetConnection *cc;
1113
1114   /* First, check if message belongs to a connection that ends here. */
1115   cc = GCC_lookup (&msg->kx.cid);
1116   if (NULL != cc)
1117   {
1118     /* verify message came from the right direction */
1119     struct CadetPeerPath *path = GCC_get_path (cc);
1120
1121     if (peer !=
1122         GCPP_get_peer_at_offset (path,
1123                                  0))
1124     {
1125       /* received message from unexpected direction, ignore! */
1126       GNUNET_break_op (0);
1127       return;
1128     }
1129     GCC_handle_kx_auth (cc,
1130                         msg);
1131     return;
1132   }
1133
1134   /* We're just an intermediary peer, route the message along its path */
1135   route_message (peer,
1136                  &msg->kx.cid,
1137                  &msg->kx.header);
1138 }
1139
1140
1141 /**
1142  * Check if the encrypted message has the appropriate size.
1143  *
1144  * @param cls Closure (unused).
1145  * @param msg Message to check.
1146  *
1147  * @return #GNUNET_YES if size is correct, #GNUNET_NO otherwise.
1148  */
1149 static int
1150 check_tunnel_encrypted (void *cls,
1151                         const struct GNUNET_CADET_TunnelEncryptedMessage *msg)
1152 {
1153   return GNUNET_YES;
1154 }
1155
1156
1157 /**
1158  * Handle for #GNUNET_MESSAGE_TYPE_CADET_TUNNEL_ENCRYPTED.
1159  *
1160  * @param cls Closure (CadetPeer for neighbor that sent the message).
1161  * @param msg Message itself.
1162  */
1163 static void
1164 handle_tunnel_encrypted (void *cls,
1165                          const struct GNUNET_CADET_TunnelEncryptedMessage *msg)
1166 {
1167   struct CadetPeer *peer = cls;
1168   struct CadetConnection *cc;
1169
1170   /* First, check if message belongs to a connection that ends here. */
1171   cc = GCC_lookup (&msg->cid);
1172   if (NULL != cc)
1173   {
1174     /* verify message came from the right direction */
1175     struct CadetPeerPath *path = GCC_get_path (cc);
1176
1177     if (peer !=
1178         GCPP_get_peer_at_offset (path,
1179                                  0))
1180     {
1181       /* received message from unexpected direction, ignore! */
1182       GNUNET_break_op (0);
1183       return;
1184     }
1185     GCC_handle_encrypted (cc,
1186                           msg);
1187     return;
1188   }
1189   /* We're just an intermediary peer, route the message along its path */
1190   route_message (peer,
1191                  &msg->cid,
1192                  &msg->header);
1193 }
1194
1195
1196 /**
1197  * Function called after #GNUNET_CORE_connect has succeeded (or failed
1198  * for good).  Note that the private key of the peer is intentionally
1199  * not exposed here; if you need it, your process should try to read
1200  * the private key file directly (which should work if you are
1201  * authorized...).  Implementations of this function must not call
1202  * #GNUNET_CORE_disconnect (other than by scheduling a new task to
1203  * do this later).
1204  *
1205  * @param cls closure
1206  * @param my_identity ID of this peer, NULL if we failed
1207  */
1208 static void
1209 core_init_cb (void *cls,
1210               const struct GNUNET_PeerIdentity *my_identity)
1211 {
1212   if (NULL == my_identity)
1213   {
1214     GNUNET_break (0);
1215     return;
1216   }
1217   GNUNET_break (0 ==
1218                 memcmp (my_identity,
1219                         &my_full_id,
1220                         sizeof (struct GNUNET_PeerIdentity)));
1221 }
1222
1223
1224 /**
1225  * Method called whenever a given peer connects.
1226  *
1227  * @param cls closure
1228  * @param peer peer identity this notification is about
1229  */
1230 static void *
1231 core_connect_cb (void *cls,
1232                  const struct GNUNET_PeerIdentity *peer,
1233                  struct GNUNET_MQ_Handle *mq)
1234 {
1235   struct CadetPeer *cp;
1236
1237   LOG (GNUNET_ERROR_TYPE_DEBUG,
1238        "CORE connection to peer %s was established.\n",
1239        GNUNET_i2s (peer));
1240   cp = GCP_get (peer,
1241                 GNUNET_YES);
1242   GCP_set_mq (cp,
1243               mq);
1244   return cp;
1245 }
1246
1247
1248 /**
1249  * Method called whenever a peer disconnects.
1250  *
1251  * @param cls closure
1252  * @param peer peer identity this notification is about
1253  */
1254 static void
1255 core_disconnect_cb (void *cls,
1256                     const struct GNUNET_PeerIdentity *peer,
1257                     void *peer_cls)
1258 {
1259   struct CadetPeer *cp = peer_cls;
1260
1261   LOG (GNUNET_ERROR_TYPE_DEBUG,
1262        "CORE connection to peer %s went down.\n",
1263        GNUNET_i2s (peer));
1264   GCP_set_mq (cp,
1265               NULL);
1266 }
1267
1268
1269 /**
1270  * Initialize the CORE subsystem.
1271  *
1272  * @param c Configuration.
1273  */
1274 void
1275 GCO_init (const struct GNUNET_CONFIGURATION_Handle *c)
1276 {
1277   struct GNUNET_MQ_MessageHandler handlers[] = {
1278     GNUNET_MQ_hd_var_size (connection_create,
1279                            GNUNET_MESSAGE_TYPE_CADET_CONNECTION_CREATE,
1280                            struct GNUNET_CADET_ConnectionCreateMessage,
1281                            NULL),
1282     GNUNET_MQ_hd_fixed_size (connection_create_ack,
1283                              GNUNET_MESSAGE_TYPE_CADET_CONNECTION_CREATE_ACK,
1284                              struct GNUNET_CADET_ConnectionCreateAckMessage,
1285                              NULL),
1286     GNUNET_MQ_hd_fixed_size (connection_broken,
1287                              GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN,
1288                              struct GNUNET_CADET_ConnectionBrokenMessage,
1289                              NULL),
1290     GNUNET_MQ_hd_fixed_size (connection_destroy,
1291                              GNUNET_MESSAGE_TYPE_CADET_CONNECTION_DESTROY,
1292                              struct GNUNET_CADET_ConnectionDestroyMessage,
1293                              NULL),
1294     GNUNET_MQ_hd_fixed_size (tunnel_kx,
1295                              GNUNET_MESSAGE_TYPE_CADET_TUNNEL_KX,
1296                              struct GNUNET_CADET_TunnelKeyExchangeMessage,
1297                              NULL),
1298     GNUNET_MQ_hd_fixed_size (tunnel_kx_auth,
1299                              GNUNET_MESSAGE_TYPE_CADET_TUNNEL_KX_AUTH,
1300                              struct GNUNET_CADET_TunnelKeyExchangeAuthMessage,
1301                              NULL),
1302     GNUNET_MQ_hd_var_size (tunnel_encrypted,
1303                            GNUNET_MESSAGE_TYPE_CADET_TUNNEL_ENCRYPTED,
1304                            struct GNUNET_CADET_TunnelEncryptedMessage,
1305                            NULL),
1306     GNUNET_MQ_handler_end ()
1307   };
1308
1309   if (GNUNET_OK !=
1310       GNUNET_CONFIGURATION_get_value_number (c,
1311                                              "CADET",
1312                                              "MAX_ROUTES",
1313                                              &max_routes))
1314     max_routes = 5000;
1315   if (GNUNET_OK !=
1316       GNUNET_CONFIGURATION_get_value_number (c,
1317                                              "CADET",
1318                                              "MAX_MSGS_QUEUE",
1319                                              &max_buffers))
1320     max_buffers = 10000;
1321   routes = GNUNET_CONTAINER_multishortmap_create (1024,
1322                                                   GNUNET_NO);
1323   route_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
1324   core = GNUNET_CORE_connect (c,
1325                               NULL,
1326                               &core_init_cb,
1327                               &core_connect_cb,
1328                               &core_disconnect_cb,
1329                               handlers);
1330 }
1331
1332
1333 /**
1334  * Shut down the CORE subsystem.
1335  */
1336 void
1337 GCO_shutdown ()
1338 {
1339   if (NULL != core)
1340   {
1341     GNUNET_CORE_disconnect (core);
1342     core = NULL;
1343   }
1344   GNUNET_assert (0 == GNUNET_CONTAINER_multishortmap_size (routes));
1345   GNUNET_CONTAINER_multishortmap_destroy (routes);
1346   routes = NULL;
1347   GNUNET_CONTAINER_heap_destroy (route_heap);
1348   route_heap = NULL;
1349   if (NULL != timeout_task)
1350   {
1351     GNUNET_SCHEDULER_cancel (timeout_task);
1352     timeout_task = NULL;
1353   }
1354 }
1355
1356 /* end of gnunet-cadet-service_core.c */