src/cadet/gnunet-service-cadet-new_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  * - properly implement GLOBAL message buffer, instead of per-route buffers
  31  * - do NOT use buffering if the route options say no buffer!
  32  * - Optimization: given BROKEN messages, destroy paths (?)
  33  */
  34 #include "platform.h"
  35 #include "gnunet-service-cadet-new_core.h"
  36 #include "gnunet-service-cadet-new_paths.h"
  37 #include "gnunet-service-cadet-new_peer.h"
  38 #include "gnunet-service-cadet-new_connection.h"
  39 #include "gnunet-service-cadet-new_tunnels.h"
  40 #include "gnunet_core_service.h"
  41 #include "gnunet_statistics_service.h"
  42 #include "cadet_protocol.h"
  43
  44
  45 #define LOG(level, ...) GNUNET_log_from(level,"cadet-cor",__VA_ARGS__)
  46
  47
  48 /**
  49  * Number of messages we are willing to buffer per route.
  50  * FIXME: have global buffer pool instead!
  51  */
  52 #define ROUTE_BUFFER_SIZE 8
  53
  54
  55 /**
  56  * Information we keep per direction for a route.
  57  */
  58 struct RouteDirection
  59 {
  60   /**
  61    * Target peer.
  62    */
  63   struct CadetPeer *hop;
  64
  65   /**
  66    * Route this direction is part of.
  67    */
  68   struct CadetRoute *my_route;
  69
  70   /**
  71    * Message queue manager for @e hop.
  72    */
  73   struct GCP_MessageQueueManager *mqm;
  74
  75   /**
  76    * Cyclic message buffer to @e hop.
  77    */
  78   struct GNUNET_MQ_Envelope *out_buffer[ROUTE_BUFFER_SIZE];
  79
  80   /**
  81    * Next write offset to use to append messages to @e out_buffer.
  82    */
  83   unsigned int out_wpos;
  84
  85   /**
  86    * Next read offset to use to retrieve messages from @e out_buffer.
  87    */
  88   unsigned int out_rpos;
  89
  90   /**
  91    * Is @e mqm currently ready for transmission?
  92    */
  93   int is_ready;
  94
  95 };
  96
  97
  98 /**
  99  * Description of a segment of a `struct CadetConnection` at the
 100  * intermediate peers.  Routes are basically entries in a peer's
 101  * routing table for forwarding traffic.  At both endpoints, the
 102  * routes are terminated by a `struct CadetConnection`, which knows
 103  * the complete `struct CadetPath` that is formed by the individual
 104  * routes.
 105  */
 106 struct CadetRoute
 107 {
 108
 109   /**
 110    * Information about the next hop on this route.
 111    */
 112   struct RouteDirection next;
 113
 114   /**
 115    * Information about the previous hop on this route.
 116    */
 117   struct RouteDirection prev;
 118
 119   /**
 120    * Unique identifier for the connection that uses this route.
 121    */
 122   struct GNUNET_CADET_ConnectionTunnelIdentifier cid;
 123
 124   /**
 125    * When was this route last in use?
 126    */
 127   struct GNUNET_TIME_Absolute last_use;
 128
 129   /**
 130    * Position of this route in the #route_heap.
 131    */
 132   struct GNUNET_CONTAINER_HeapNode *hn;
 133
 134   /**
 135    * Options for the route, control buffering.
 136    */
 137   enum GNUNET_CADET_ChannelOption options;
 138 };
 139
 140
 141 /**
 142  * Handle to the CORE service.
 143  */
 144 static struct GNUNET_CORE_Handle *core;
 145
 146 /**
 147  * Routes on which this peer is an intermediate.
 148  */
 149 static struct GNUNET_CONTAINER_MultiShortmap *routes;
 150
 151 /**
 152  * Heap of routes, MIN-sorted by last activity.
 153  */
 154 static struct GNUNET_CONTAINER_Heap *route_heap;
 155
 156 /**
 157  * Maximum number of concurrent routes this peer will support.
 158  */
 159 static unsigned long long max_routes;
 160
 161 /**
 162  * Task to timeout routes.
 163  */
 164 static struct GNUNET_SCHEDULER_Task *timeout_task;
 165
 166
 167 /**
 168  * Get the route corresponding to a hash.
 169  *
 170  * @param cid hash generated from the connection identifier
 171  */
 172 static struct CadetRoute *
 173 get_route (const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid)
 174 {
 175   return GNUNET_CONTAINER_multishortmap_get (routes,
 176                                              &cid->connection_of_tunnel);
 177 }
 178
 179
 180 /**
 181  * We message @a msg from @a prev.  Find its route by @a cid and
 182  * forward to the next hop.  Drop and signal broken route if we do not
 183  * have a route.
 184  *
 185  * @param prev previous hop (sender)
 186  * @param cid connection identifier, tells us which route to use
 187  * @param msg the message to forward
 188  */
 189 static void
 190 route_message (struct CadetPeer *prev,
 191                const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid,
 192                const struct GNUNET_MessageHeader *msg)
 193 {
 194   struct CadetRoute *route;
 195   struct RouteDirection *dir;
 196   struct GNUNET_MQ_Envelope *env;
 197
 198   route = get_route (cid);
 199   if (NULL == route)
 200   {
 201     struct GNUNET_MQ_Envelope *env;
 202     struct GNUNET_CADET_ConnectionBrokenMessage *bm;
 203
 204     LOG (GNUNET_ERROR_TYPE_DEBUG,
 205          "Failed to route message of type %u from %s on connection %s: no route\n",
 206          ntohs (msg->type),
 207          GCP_2s (prev),
 208          GNUNET_sh2s (&cid->connection_of_tunnel));
 209     env = GNUNET_MQ_msg (bm,
 210                          GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN);
 211     bm->cid = *cid;
 212     bm->peer1 = my_full_id;
 213     GCP_send_ooo (prev,
 214                   env);
 215     return;
 216   }
 217   route->last_use = GNUNET_TIME_absolute_get ();
 218   GNUNET_CONTAINER_heap_update_cost (route->hn,
 219                                      route->last_use.abs_value_us);
 220   dir = (prev == route->prev.hop) ? &route->next : &route->prev;
 221   if (GNUNET_YES == dir->is_ready)
 222   {
 223     LOG (GNUNET_ERROR_TYPE_DEBUG,
 224          "Routing message of type %u from %s to %s on connection %s\n",
 225          ntohs (msg->type),
 226          GCP_2s (prev),
 227          GNUNET_i2s (GCP_get_id (dir->hop)),
 228          GNUNET_sh2s (&cid->connection_of_tunnel));
 229     dir->is_ready = GNUNET_NO;
 230     GCP_send (dir->mqm,
 231               GNUNET_MQ_msg_copy (msg));
 232     return;
 233   }
 234   env = dir->out_buffer[dir->out_wpos];
 235   if (NULL != env)
 236   {
 237     /* Queue full, drop earliest message in queue */
 238     LOG (GNUNET_ERROR_TYPE_DEBUG,
 239          "Queue full due to new message of type %u from %s to %s on connection %s, dropping old message\n",
 240          ntohs (msg->type),
 241          GCP_2s (prev),
 242          GNUNET_i2s (GCP_get_id (dir->hop)),
 243          GNUNET_sh2s (&cid->connection_of_tunnel));
 244     GNUNET_STATISTICS_update (stats,
 245                               "# messages dropped due to full buffer",
 246                               1,
 247                               GNUNET_NO);
 248   GNUNET_assert (dir->out_rpos == dir->out_wpos);
 249     GNUNET_MQ_discard (env);
 250     dir->out_rpos++;
 251     if (ROUTE_BUFFER_SIZE == dir->out_rpos)
 252       dir->out_rpos = 0;
 253   }
 254   LOG (GNUNET_ERROR_TYPE_DEBUG,
 255        "Queueing new message of type %u from %s to %s on connection %s\n",
 256        ntohs (msg->type),
 257        GCP_2s (prev),
 258        GNUNET_i2s (GCP_get_id (dir->hop)),
 259        GNUNET_sh2s (&cid->connection_of_tunnel));
 260   env = GNUNET_MQ_msg_copy (msg);
 261   dir->out_buffer[dir->out_wpos] = env;
 262   dir->out_wpos++;
 263   if (ROUTE_BUFFER_SIZE == dir->out_wpos)
 264     dir->out_wpos = 0;
 265 }
 266
 267
 268 /**
 269  * Check if the create_connection message has the appropriate size.
 270  *
 271  * @param cls Closure (unused).
 272  * @param msg Message to check.
 273  *
 274  * @return #GNUNET_YES if size is correct, #GNUNET_NO otherwise.
 275  */
 276 static int
 277 check_connection_create (void *cls,
 278                          const struct GNUNET_CADET_ConnectionCreateMessage *msg)
 279 {
 280   uint16_t size = ntohs (msg->header.size) - sizeof (*msg);
 281
 282   if (0 != (size % sizeof (struct GNUNET_PeerIdentity)))
 283   {
 284     GNUNET_break_op (0);
 285     return GNUNET_NO;
 286   }
 287   return GNUNET_YES;
 288 }
 289
 290
 291 /**
 292  * Free internal data of a route direction.
 293  *
 294  * @param dir direction to destroy (do NOT free memory of 'dir' itself)
 295  */
 296 static void
 297 destroy_direction (struct RouteDirection *dir)
 298 {
 299   for (unsigned int i=0;i<ROUTE_BUFFER_SIZE;i++)
 300     if (NULL != dir->out_buffer[i])
 301     {
 302       GNUNET_MQ_discard (dir->out_buffer[i]);
 303       dir->out_buffer[i] = NULL;
 304     }
 305   if (NULL != dir->mqm)
 306   {
 307     GCP_request_mq_cancel (dir->mqm,
 308                            NULL);
 309     dir->mqm = NULL;
 310   }
 311 }
 312
 313
 314 /**
 315  * Destroy our state for @a route.
 316  *
 317  * @param route route to destroy
 318  */
 319 static void
 320 destroy_route (struct CadetRoute *route)
 321 {
 322   LOG (GNUNET_ERROR_TYPE_DEBUG,
 323        "Destroying route from %s to %s of connection %s\n",
 324        GNUNET_i2s  (GCP_get_id (route->prev.hop)),
 325        GNUNET_i2s2 (GCP_get_id (route->next.hop)),
 326        GNUNET_sh2s (&route->cid.connection_of_tunnel));
 327   GNUNET_assert (route ==
 328                  GNUNET_CONTAINER_heap_remove_node (route->hn));
 329   destroy_direction (&route->prev);
 330   destroy_direction (&route->next);
 331   GNUNET_free (route);
 332 }
 333
 334
 335 /**
 336  * Send message that a route is broken between @a peer1 and @a peer2.
 337  *
 338  * @param target where to send the message
 339  * @param cid connection identifier to use
 340  * @param peer1 one of the peers where a link is broken
 341  * @param peer2 another one of the peers where a link is broken
 342  */
 343 static void
 344 send_broken (struct RouteDirection *target,
 345              const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid,
 346              const struct GNUNET_PeerIdentity *peer1,
 347              const struct GNUNET_PeerIdentity *peer2)
 348 {
 349   struct GNUNET_MQ_Envelope *env;
 350   struct GNUNET_CADET_ConnectionBrokenMessage *bm;
 351
 352   if (NULL == target->mqm)
 353     return; /* Can't send notification, connection is down! */
 354   LOG (GNUNET_ERROR_TYPE_DEBUG,
 355        "Notifying %s about BROKEN route at %s-%s of connection %s\n",
 356        GCP_2s (target->hop),
 357        GNUNET_i2s (peer1),
 358        GNUNET_i2s2 (peer2),
 359        GNUNET_sh2s (&cid->connection_of_tunnel));
 360
 361   env = GNUNET_MQ_msg (bm,
 362                        GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN);
 363   bm->cid = *cid;
 364   if (NULL != peer1)
 365     bm->peer1 = *peer1;
 366   if (NULL != peer2)
 367     bm->peer2 = *peer2;
 368
 369   GCP_request_mq_cancel (target->mqm,
 370                          env);
 371   target->mqm = NULL;
 372 }
 373
 374
 375 /**
 376  * Function called to check if any routes have timed out, and if
 377  * so, to clean them up.  Finally, schedules itself again at the
 378  * earliest time where there might be more work.
 379  *
 380  * @param cls NULL
 381  */
 382 static void
 383 timeout_cb (void *cls)
 384 {
 385   struct CadetRoute *r;
 386   struct GNUNET_TIME_Relative linger;
 387   struct GNUNET_TIME_Absolute exp;
 388
 389   timeout_task = NULL;
 390   linger = GNUNET_TIME_relative_multiply (keepalive_period,
 391                                           3);
 392   while (NULL != (r = GNUNET_CONTAINER_heap_peek (route_heap)))
 393   {
 394     exp = GNUNET_TIME_absolute_add (r->last_use,
 395                                     linger);
 396     if (0 != GNUNET_TIME_absolute_get_duration (exp).rel_value_us)
 397     {
 398       /* Route not yet timed out, wait until it does. */
 399       timeout_task = GNUNET_SCHEDULER_add_at (exp,
 400                                               &timeout_cb,
 401                                               NULL);
 402       return;
 403     }
 404     send_broken (&r->prev,
 405                  &r->cid,
 406                  NULL,
 407                  NULL);
 408     send_broken (&r->next,
 409                  &r->cid,
 410                  NULL,
 411                  NULL);
 412     destroy_route (r);
 413   }
 414   /* No more routes left, so no need for a #timeout_task */
 415 }
 416
 417
 418 /**
 419  * Function called when the message queue to the previous hop
 420  * becomes available/unavailable.  We expect this function to
 421  * be called immediately when we register, and then again
 422  * later if the connection ever goes down.
 423  *
 424  * @param cls the `struct RouteDirection`
 425  * @param available #GNUNET_YES if sending is now possible,
 426  *                  #GNUNET_NO if sending is no longer possible
 427  *                  #GNUNET_SYSERR if sending is no longer possible
 428  *                                 and the last envelope was discarded
 429  */
 430 static void
 431 dir_ready_cb (void *cls,
 432               int ready)
 433 {
 434   struct RouteDirection *dir = cls;
 435   struct CadetRoute *route = dir->my_route;
 436   struct RouteDirection *odir;
 437
 438   if (GNUNET_YES == ready)
 439   {
 440     struct GNUNET_MQ_Envelope *env;
 441
 442     dir->is_ready = GNUNET_YES;
 443     if (NULL != (env = dir->out_buffer[dir->out_rpos]))
 444     {
 445       dir->out_buffer[dir->out_rpos] = NULL;
 446       dir->out_rpos++;
 447       if (ROUTE_BUFFER_SIZE == dir->out_rpos)
 448         dir->out_rpos = 0;
 449       dir->is_ready = GNUNET_NO;
 450       GCP_send (dir->mqm,
 451                 env);
 452     }
 453     return;
 454   }
 455   odir = (dir == &route->next) ? &route->prev : &route->next;
 456   send_broken (&route->next,
 457                &route->cid,
 458                GCP_get_id (odir->hop),
 459                &my_full_id);
 460   destroy_route (route);
 461 }
 462
 463
 464 /**
 465  * Initialize one of the directions of a route.
 466  *
 467  * @param route route the direction belongs to
 468  * @param dir direction to initialize
 469  * @param hop next hop on in the @a dir
 470  */
 471 static void
 472 dir_init (struct RouteDirection *dir,
 473           struct CadetRoute *route,
 474           struct CadetPeer *hop)
 475 {
 476   dir->hop = hop;
 477   dir->my_route = route;
 478   dir->mqm = GCP_request_mq (hop,
 479                              &dir_ready_cb,
 480                              dir);
 481   GNUNET_assert (GNUNET_YES == dir->is_ready);
 482 }
 483
 484
 485 /**
 486  * We could not create the desired route.  Send a
 487  * #GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN
 488  * message to @a target.
 489  *
 490  * @param target who should receive the message
 491  * @param cid identifier of the connection/route that failed
 492  * @param failure_at neighbour with which we failed to route,
 493  *        or NULL.
 494  */
 495 static void
 496 send_broken_without_mqm (struct CadetPeer *target,
 497                          const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid,
 498                          const struct GNUNET_PeerIdentity *failure_at)
 499 {
 500   struct GNUNET_MQ_Envelope *env;
 501   struct GNUNET_CADET_ConnectionBrokenMessage *bm;
 502
 503   env = GNUNET_MQ_msg (bm,
 504                        GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN);
 505   bm->cid = *cid;
 506   bm->peer1 = my_full_id;
 507   if (NULL != failure_at)
 508     bm->peer2 = *failure_at;
 509   GCP_send_ooo (target,
 510                 env);
 511 }
 512
 513
 514 /**
 515  * Handle for #GNUNET_MESSAGE_TYPE_CADET_CONNECTION_CREATE
 516  *
 517  * @param cls Closure (CadetPeer for neighbor that sent the message).
 518  * @param msg Message itself.
 519  */
 520 static void
 521 handle_connection_create (void *cls,
 522                           const struct GNUNET_CADET_ConnectionCreateMessage *msg)
 523 {
 524   struct CadetPeer *sender = cls;
 525   struct CadetPeer *next;
 526   const struct GNUNET_PeerIdentity *pids = (const struct GNUNET_PeerIdentity *) &msg[1];
 527   struct CadetRoute *route;
 528   uint16_t size = ntohs (msg->header.size) - sizeof (*msg);
 529   unsigned int path_length;
 530   unsigned int off;
 531   enum GNUNET_CADET_ChannelOption options;
 532
 533   options = (enum GNUNET_CADET_ChannelOption) ntohl (msg->options);
 534   path_length = size / sizeof (struct GNUNET_PeerIdentity);
 535   /* Initiator is at offset 0. */
 536   for (off=1;off<path_length;off++)
 537     if (0 == memcmp (&my_full_id,
 538                      &pids[off],
 539                      sizeof (struct GNUNET_PeerIdentity)))
 540       break;
 541   if (off == path_length)
 542   {
 543     /* We are not on the path, bogus request */
 544     GNUNET_break_op (0);
 545     return;
 546   }
 547   /* Check previous hop */
 548   if (sender != GCP_get (&pids[off - 1],
 549                          GNUNET_NO))
 550   {
 551     /* sender is not on the path, not allowed */
 552     GNUNET_break_op (0);
 553     return;
 554   }
 555   if (NULL !=
 556       get_route (&msg->cid))
 557   {
 558     /* Duplicate CREATE, pass it on, previous one might have been lost! */
 559     LOG (GNUNET_ERROR_TYPE_DEBUG,
 560          "Passing on duplicate CADET_CONNECTION_CREATE message on connection %s\n",
 561          GNUNET_sh2s (&msg->cid.connection_of_tunnel));
 562     route_message (sender,
 563                    &msg->cid,
 564                    &msg->header);
 565     return;
 566   }
 567   if (off == path_length - 1)
 568   {
 569     /* We are the destination, create connection */
 570     struct CadetConnection *cc;
 571     struct CadetPeerPath *path;
 572     struct CadetPeer *origin;
 573
 574     cc = GNUNET_CONTAINER_multishortmap_get (connections,
 575                                              &msg->cid.connection_of_tunnel);
 576     if (NULL != cc)
 577     {
 578       LOG (GNUNET_ERROR_TYPE_DEBUG,
 579            "Received duplicate CADET_CONNECTION_CREATE message on connection %s\n",
 580            GNUNET_sh2s (&msg->cid.connection_of_tunnel));
 581       GCC_handle_duplicate_create (cc);
 582       return;
 583     }
 584
 585     origin = GCP_get (&pids[0],
 586                       GNUNET_YES);
 587     LOG (GNUNET_ERROR_TYPE_DEBUG,
 588          "Received CADET_CONNECTION_CREATE message from %s for connection %s, building inverse path\n",
 589          GCP_2s (origin),
 590          GNUNET_sh2s (&msg->cid.connection_of_tunnel));
 591     path = GCPP_get_path_from_route (path_length - 1,
 592                                      pids);
 593     if (GNUNET_OK !=
 594         GCT_add_inbound_connection (GCP_get_tunnel (origin,
 595                                                     GNUNET_YES),
 596                                     &msg->cid,
 597                                     (enum GNUNET_CADET_ChannelOption) ntohl (msg->options),
 598                                     path))
 599     {
 600       /* Send back BROKEN: duplicate connection on the same path,
 601          we will use the other one. */
 602       LOG (GNUNET_ERROR_TYPE_DEBUG,
 603            "Received CADET_CONNECTION_CREATE from %s for %s, but %s already has a connection. Sending BROKEN\n",
 604            GCP_2s (sender),
 605            GNUNET_sh2s (&msg->cid.connection_of_tunnel),
 606            GCPP_2s (path));
 607       send_broken_without_mqm (sender,
 608                                &msg->cid,
 609                                NULL);
 610       return;
 611     }
 612     return;
 613   }
 614   /* We are merely a hop on the way, check if we can support the route */
 615   next = GCP_get (&pids[off + 1],
 616                   GNUNET_NO);
 617   if ( (NULL == next) ||
 618        (GNUNET_NO == GCP_has_core_connection (next)) )
 619   {
 620     /* unworkable, send back BROKEN notification */
 621     LOG (GNUNET_ERROR_TYPE_DEBUG,
 622          "Received CADET_CONNECTION_CREATE from %s for %s. Next hop %s:%u is down. Sending BROKEN\n",
 623          GCP_2s (sender),
 624          GNUNET_sh2s (&msg->cid.connection_of_tunnel),
 625          GNUNET_i2s (&pids[off + 1]),
 626          off + 1);
 627     send_broken_without_mqm (sender,
 628                              &msg->cid,
 629                              &pids[off + 1]);
 630     return;
 631   }
 632   if (max_routes <= GNUNET_CONTAINER_multishortmap_size (routes))
 633   {
 634     LOG (GNUNET_ERROR_TYPE_DEBUG,
 635          "Received CADET_CONNECTION_CREATE from %s for %s. We have reached our route limit. Sending BROKEN\n",
 636          GCP_2s (sender),
 637          GNUNET_sh2s (&msg->cid.connection_of_tunnel));
 638     send_broken_without_mqm (sender,
 639                              &msg->cid,
 640                              &pids[off - 1]);
 641     return;
 642   }
 643
 644   /* Workable route, create routing entry */
 645   LOG (GNUNET_ERROR_TYPE_DEBUG,
 646        "Received CADET_CONNECTION_CREATE from %s for %s. Next hop %s:%u is up. Creating route\n",
 647        GCP_2s (sender),
 648        GNUNET_sh2s (&msg->cid.connection_of_tunnel),
 649        GNUNET_i2s (&pids[off + 1]),
 650        off + 1);
 651   route = GNUNET_new (struct CadetRoute);
 652   route->options = options;
 653   route->cid = msg->cid;
 654   route->last_use = GNUNET_TIME_absolute_get ();
 655   dir_init (&route->prev,
 656             route,
 657             sender);
 658   dir_init (&route->next,
 659             route,
 660             next);
 661   GNUNET_assert (GNUNET_OK ==
 662                  GNUNET_CONTAINER_multishortmap_put (routes,
 663                                                      &route->cid.connection_of_tunnel,
 664                                                      route,
 665                                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 666   route->hn = GNUNET_CONTAINER_heap_insert (route_heap,
 667                                             route,
 668                                             route->last_use.abs_value_us);
 669   if (NULL == timeout_task)
 670     timeout_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply (keepalive_period,
 671                                                                                 3),
 672                                                  &timeout_cb,
 673                                                  NULL);
 674 }
 675
 676
 677 /**
 678  * Handle for #GNUNET_MESSAGE_TYPE_CADET_CONNECTION_CREATE_ACK
 679  *
 680  * @param cls Closure (CadetPeer for neighbor that sent the message).
 681  * @param msg Message itself.
 682  */
 683 static void
 684 handle_connection_create_ack (void *cls,
 685                               const struct GNUNET_CADET_ConnectionCreateAckMessage *msg)
 686 {
 687   struct CadetPeer *peer = cls;
 688   struct CadetConnection *cc;
 689
 690   /* First, check if ACK belongs to a connection that ends here. */
 691   cc = GNUNET_CONTAINER_multishortmap_get (connections,
 692                                            &msg->cid.connection_of_tunnel);
 693   if (NULL != cc)
 694   {
 695     /* verify ACK came from the right direction */
 696     struct CadetPeerPath *path = GCC_get_path (cc);
 697
 698     if (peer !=
 699         GCPP_get_peer_at_offset (path,
 700                                  0))
 701     {
 702       /* received ACK from unexpected direction, ignore! */
 703       GNUNET_break_op (0);
 704       return;
 705     }
 706     LOG (GNUNET_ERROR_TYPE_DEBUG,
 707          "Received CONNECTION_CREATE_ACK for connection %s.\n",
 708          GNUNET_sh2s (&msg->cid.connection_of_tunnel));
 709     GCC_handle_connection_create_ack (cc);
 710     return;
 711   }
 712
 713   /* We're just an intermediary peer, route the message along its path */
 714   route_message (peer,
 715                  &msg->cid,
 716                  &msg->header);
 717 }
 718
 719
 720 /**
 721  * Handle for #GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN
 722  *
 723  * @param cls Closure (CadetPeer for neighbor that sent the message).
 724  * @param msg Message itself.
 725  * @deprecated duplicate logic with #handle_destroy(); dedup!
 726  */
 727 static void
 728 handle_connection_broken (void *cls,
 729                           const struct GNUNET_CADET_ConnectionBrokenMessage *msg)
 730 {
 731   struct CadetPeer *peer = cls;
 732   struct CadetConnection *cc;
 733   struct CadetRoute *route;
 734
 735   /* First, check if message belongs to a connection that ends here. */
 736   cc = GNUNET_CONTAINER_multishortmap_get (connections,
 737                                            &msg->cid.connection_of_tunnel);
 738   if (NULL != cc)
 739   {
 740     /* verify message came from the right direction */
 741     struct CadetPeerPath *path = GCC_get_path (cc);
 742
 743     if (peer !=
 744         GCPP_get_peer_at_offset (path,
 745                                  0))
 746     {
 747       /* received message from unexpected direction, ignore! */
 748       GNUNET_break_op (0);
 749       return;
 750     }
 751     LOG (GNUNET_ERROR_TYPE_DEBUG,
 752          "Received CONNECTION_BROKEN for connection %s. Destroying it.\n",
 753          GNUNET_sh2s (&msg->cid.connection_of_tunnel));
 754     GCC_destroy_without_core (cc);
 755
 756     /* FIXME: also destroy the path up to the specified link! */
 757     return;
 758   }
 759
 760   /* We're just an intermediary peer, route the message along its path */
 761   route = get_route (&msg->cid);
 762   route_message (peer,
 763                  &msg->cid,
 764                  &msg->header);
 765   destroy_route (route);
 766   /* FIXME: also destroy paths we MAY have up to the specified link! */
 767 }
 768
 769
 770 /**
 771  * Handle for #GNUNET_MESSAGE_TYPE_CADET_CONNECTION_DESTROY
 772  *
 773  * @param cls Closure (CadetPeer for neighbor that sent the message).
 774  * @param msg Message itself.
 775  */
 776 static void
 777 handle_connection_destroy (void *cls,
 778                            const struct GNUNET_CADET_ConnectionDestroyMessage *msg)
 779 {
 780   struct CadetPeer *peer = cls;
 781   struct CadetConnection *cc;
 782   struct CadetRoute *route;
 783
 784   /* First, check if message belongs to a connection that ends here. */
 785   cc = GNUNET_CONTAINER_multishortmap_get (connections,
 786                                            &msg->cid.connection_of_tunnel);
 787   if (NULL != cc)
 788   {
 789     /* verify message came from the right direction */
 790     struct CadetPeerPath *path = GCC_get_path (cc);
 791
 792     if (peer !=
 793         GCPP_get_peer_at_offset (path,
 794                                  0))
 795     {
 796       /* received message from unexpected direction, ignore! */
 797       GNUNET_break_op (0);
 798       return;
 799     }
 800     LOG (GNUNET_ERROR_TYPE_DEBUG,
 801          "Received CONNECTION_DESTROY for connection %s. Destroying connection.\n",
 802          GNUNET_sh2s (&msg->cid.connection_of_tunnel));
 803
 804     GCC_destroy_without_core (cc);
 805     return;
 806   }
 807
 808   /* We're just an intermediary peer, route the message along its path */
 809   LOG (GNUNET_ERROR_TYPE_DEBUG,
 810        "Received CONNECTION_DESTROY for connection %s. Destroying route.\n",
 811        GNUNET_sh2s (&msg->cid.connection_of_tunnel));
 812   route = get_route (&msg->cid);
 813   route_message (peer,
 814                  &msg->cid,
 815                  &msg->header);
 816   destroy_route (route);
 817 }
 818
 819
 820 /**
 821  * Handle for #GNUNET_MESSAGE_TYPE_CADET_TUNNEL_KX
 822  *
 823  * @param cls Closure (CadetPeer for neighbor that sent the message).
 824  * @param msg Message itself.
 825  */
 826 static void
 827 handle_tunnel_kx (void *cls,
 828                   const struct GNUNET_CADET_TunnelKeyExchangeMessage *msg)
 829 {
 830   struct CadetPeer *peer = cls;
 831   struct CadetConnection *cc;
 832
 833   /* First, check if message belongs to a connection that ends here. */
 834   cc = GNUNET_CONTAINER_multishortmap_get (connections,
 835                                            &msg->cid.connection_of_tunnel);
 836   if (NULL != cc)
 837   {
 838     /* verify message came from the right direction */
 839     struct CadetPeerPath *path = GCC_get_path (cc);
 840
 841     if (peer !=
 842         GCPP_get_peer_at_offset (path,
 843                                  0))
 844     {
 845       /* received message from unexpected direction, ignore! */
 846       GNUNET_break_op (0);
 847       return;
 848     }
 849     GCC_handle_kx (cc,
 850                    msg);
 851     return;
 852   }
 853
 854   /* We're just an intermediary peer, route the message along its path */
 855   route_message (peer,
 856                  &msg->cid,
 857                  &msg->header);
 858 }
 859
 860
 861 /**
 862  * Handle for #GNUNET_MESSAGE_TYPE_CADET_TUNNEL_KX_AUTH
 863  *
 864  * @param cls Closure (CadetPeer for neighbor that sent the message).
 865  * @param msg Message itself.
 866  */
 867 static void
 868 handle_tunnel_kx_auth (void *cls,
 869                        const struct GNUNET_CADET_TunnelKeyExchangeAuthMessage *msg)
 870 {
 871   struct CadetPeer *peer = cls;
 872   struct CadetConnection *cc;
 873
 874   /* First, check if message belongs to a connection that ends here. */
 875   cc = GNUNET_CONTAINER_multishortmap_get (connections,
 876                                            &msg->kx.cid.connection_of_tunnel);
 877   if (NULL != cc)
 878   {
 879     /* verify message came from the right direction */
 880     struct CadetPeerPath *path = GCC_get_path (cc);
 881
 882     if (peer !=
 883         GCPP_get_peer_at_offset (path,
 884                                  0))
 885     {
 886       /* received message from unexpected direction, ignore! */
 887       GNUNET_break_op (0);
 888       return;
 889     }
 890     GCC_handle_kx_auth (cc,
 891                         msg);
 892     return;
 893   }
 894
 895   /* We're just an intermediary peer, route the message along its path */
 896   route_message (peer,
 897                  &msg->kx.cid,
 898                  &msg->kx.header);
 899 }
 900
 901
 902 /**
 903  * Check if the encrypted message has the appropriate size.
 904  *
 905  * @param cls Closure (unused).
 906  * @param msg Message to check.
 907  *
 908  * @return #GNUNET_YES if size is correct, #GNUNET_NO otherwise.
 909  */
 910 static int
 911 check_tunnel_encrypted (void *cls,
 912                         const struct GNUNET_CADET_TunnelEncryptedMessage *msg)
 913 {
 914   return GNUNET_YES;
 915 }
 916
 917
 918 /**
 919  * Handle for #GNUNET_MESSAGE_TYPE_CADET_TUNNEL_ENCRYPTED.
 920  *
 921  * @param cls Closure (CadetPeer for neighbor that sent the message).
 922  * @param msg Message itself.
 923  */
 924 static void
 925 handle_tunnel_encrypted (void *cls,
 926                          const struct GNUNET_CADET_TunnelEncryptedMessage *msg)
 927 {
 928   struct CadetPeer *peer = cls;
 929   struct CadetConnection *cc;
 930
 931   /* First, check if message belongs to a connection that ends here. */
 932   cc = GNUNET_CONTAINER_multishortmap_get (connections,
 933                                            &msg->cid.connection_of_tunnel);
 934   if (NULL != cc)
 935   {
 936     /* verify message 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 message from unexpected direction, ignore! */
 944       GNUNET_break_op (0);
 945       return;
 946     }
 947     GCC_handle_encrypted (cc,
 948                           msg);
 949     return;
 950   }
 951   /* We're just an intermediary peer, route the message along its path */
 952   route_message (peer,
 953                  &msg->cid,
 954                  &msg->header);
 955 }
 956
 957
 958 /**
 959  * Function called after #GNUNET_CORE_connect has succeeded (or failed
 960  * for good).  Note that the private key of the peer is intentionally
 961  * not exposed here; if you need it, your process should try to read
 962  * the private key file directly (which should work if you are
 963  * authorized...).  Implementations of this function must not call
 964  * #GNUNET_CORE_disconnect (other than by scheduling a new task to
 965  * do this later).
 966  *
 967  * @param cls closure
 968  * @param my_identity ID of this peer, NULL if we failed
 969  */
 970 static void
 971 core_init_cb (void *cls,
 972               const struct GNUNET_PeerIdentity *my_identity)
 973 {
 974   if (NULL == my_identity)
 975   {
 976     GNUNET_break (0);
 977     return;
 978   }
 979   GNUNET_break (0 ==
 980                 memcmp (my_identity,
 981                         &my_full_id,
 982                         sizeof (struct GNUNET_PeerIdentity)));
 983 }
 984
 985
 986 /**
 987  * Method called whenever a given peer connects.
 988  *
 989  * @param cls closure
 990  * @param peer peer identity this notification is about
 991  */
 992 static void *
 993 core_connect_cb (void *cls,
 994                  const struct GNUNET_PeerIdentity *peer,
 995                  struct GNUNET_MQ_Handle *mq)
 996 {
 997   struct CadetPeer *cp;
 998
 999   LOG (GNUNET_ERROR_TYPE_DEBUG,
1000        "CORE connection to peer %s was established.\n",
1001        GNUNET_i2s (peer));
1002   cp = GCP_get (peer,
1003                 GNUNET_YES);
1004   GCP_set_mq (cp,
1005               mq);
1006   return cp;
1007 }
1008
1009
1010 /**
1011  * Method called whenever a peer disconnects.
1012  *
1013  * @param cls closure
1014  * @param peer peer identity this notification is about
1015  */
1016 static void
1017 core_disconnect_cb (void *cls,
1018                     const struct GNUNET_PeerIdentity *peer,
1019                     void *peer_cls)
1020 {
1021   struct CadetPeer *cp = peer_cls;
1022
1023   LOG (GNUNET_ERROR_TYPE_DEBUG,
1024        "CORE connection to peer %s went down.\n",
1025        GNUNET_i2s (peer));
1026   GCP_set_mq (cp,
1027               NULL);
1028 }
1029
1030
1031 /**
1032  * Initialize the CORE subsystem.
1033  *
1034  * @param c Configuration.
1035  */
1036 void
1037 GCO_init (const struct GNUNET_CONFIGURATION_Handle *c)
1038 {
1039   struct GNUNET_MQ_MessageHandler handlers[] = {
1040     GNUNET_MQ_hd_var_size (connection_create,
1041                            GNUNET_MESSAGE_TYPE_CADET_CONNECTION_CREATE,
1042                            struct GNUNET_CADET_ConnectionCreateMessage,
1043                            NULL),
1044     GNUNET_MQ_hd_fixed_size (connection_create_ack,
1045                              GNUNET_MESSAGE_TYPE_CADET_CONNECTION_CREATE_ACK,
1046                              struct GNUNET_CADET_ConnectionCreateAckMessage,
1047                              NULL),
1048     GNUNET_MQ_hd_fixed_size (connection_broken,
1049                              GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN,
1050                              struct GNUNET_CADET_ConnectionBrokenMessage,
1051                              NULL),
1052     GNUNET_MQ_hd_fixed_size (connection_destroy,
1053                              GNUNET_MESSAGE_TYPE_CADET_CONNECTION_DESTROY,
1054                              struct GNUNET_CADET_ConnectionDestroyMessage,
1055                              NULL),
1056     GNUNET_MQ_hd_fixed_size (tunnel_kx,
1057                              GNUNET_MESSAGE_TYPE_CADET_TUNNEL_KX,
1058                              struct GNUNET_CADET_TunnelKeyExchangeMessage,
1059                              NULL),
1060     GNUNET_MQ_hd_fixed_size (tunnel_kx_auth,
1061                              GNUNET_MESSAGE_TYPE_CADET_TUNNEL_KX_AUTH,
1062                              struct GNUNET_CADET_TunnelKeyExchangeAuthMessage,
1063                              NULL),
1064     GNUNET_MQ_hd_var_size (tunnel_encrypted,
1065                            GNUNET_MESSAGE_TYPE_CADET_TUNNEL_ENCRYPTED,
1066                            struct GNUNET_CADET_TunnelEncryptedMessage,
1067                            NULL),
1068     GNUNET_MQ_handler_end ()
1069   };
1070
1071   if (GNUNET_OK !=
1072       GNUNET_CONFIGURATION_get_value_number (c,
1073                                              "CADET",
1074                                              "MAX_ROUTES",
1075                                              &max_routes))
1076     max_routes = 10000;
1077   routes = GNUNET_CONTAINER_multishortmap_create (1024,
1078                                                   GNUNET_NO);
1079   route_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
1080   core = GNUNET_CORE_connect (c,
1081                               NULL,
1082                               &core_init_cb,
1083                               &core_connect_cb,
1084                               &core_disconnect_cb,
1085                               handlers);
1086 }
1087
1088
1089 /**
1090  * Shut down the CORE subsystem.
1091  */
1092 void
1093 GCO_shutdown ()
1094 {
1095   if (NULL != core)
1096   {
1097     GNUNET_CORE_disconnect (core);
1098     core = NULL;
1099   }
1100   GNUNET_assert (0 == GNUNET_CONTAINER_multishortmap_size (routes));
1101   GNUNET_CONTAINER_multishortmap_destroy (routes);
1102   routes = NULL;
1103   GNUNET_CONTAINER_heap_destroy (route_heap);
1104   route_heap = NULL;
1105   if (NULL != timeout_task)
1106   {
1107     GNUNET_SCHEDULER_cancel (timeout_task);
1108     timeout_task = NULL;
1109   }
1110 }
1111
1112 /* end of gnunet-cadet-service_core.c */