src/fs/gnunet-service-fs_cadet_client.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2012, 2013 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 fs/gnunet-service-fs_cadet_client.c
  23  * @brief non-anonymous file-transfer
  24  * @author Christian Grothoff
  25  *
  26  * TODO:
  27  * - PORT is set to old application type, unsure if we should keep
  28  *   it that way (fine for now)
  29  */
  30 #include "platform.h"
  31 #include "gnunet_constants.h"
  32 #include "gnunet_util_lib.h"
  33 #include "gnunet_cadet_service.h"
  34 #include "gnunet_protocols.h"
  35 #include "gnunet_applications.h"
  36 #include "gnunet-service-fs.h"
  37 #include "gnunet-service-fs_indexing.h"
  38 #include "gnunet-service-fs_cadet.h"
  39
  40
  41 /**
  42  * After how long do we reset connections without replies?
  43  */
  44 #define CLIENT_RETRY_TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 30)
  45
  46
  47 /**
  48  * Handle for a cadet to another peer.
  49  */
  50 struct CadetHandle;
  51
  52
  53 /**
  54  * Handle for a request that is going out via cadet API.
  55  */
  56 struct GSF_CadetRequest
  57 {
  58
  59   /**
  60    * DLL.
  61    */
  62   struct GSF_CadetRequest *next;
  63
  64   /**
  65    * DLL.
  66    */
  67   struct GSF_CadetRequest *prev;
  68
  69   /**
  70    * Which cadet is this request associated with?
  71    */
  72   struct CadetHandle *mh;
  73
  74   /**
  75    * Function to call with the result.
  76    */
  77   GSF_CadetReplyProcessor proc;
  78
  79   /**
  80    * Closure for 'proc'
  81    */
  82   void *proc_cls;
  83
  84   /**
  85    * Query to transmit to the other peer.
  86    */
  87   struct GNUNET_HashCode query;
  88
  89   /**
  90    * Desired type for the reply.
  91    */
  92   enum GNUNET_BLOCK_Type type;
  93
  94   /**
  95    * Did we transmit this request already? #GNUNET_YES if we are
  96    * in the 'waiting_map', #GNUNET_NO if we are in the 'pending' DLL.
  97    */
  98   int was_transmitted;
  99 };
 100
 101
 102 /**
 103  * Handle for a cadet to another peer.
 104  */
 105 struct CadetHandle
 106 {
 107   /**
 108    * Head of DLL of pending requests on this cadet.
 109    */
 110   struct GSF_CadetRequest *pending_head;
 111
 112   /**
 113    * Tail of DLL of pending requests on this cadet.
 114    */
 115   struct GSF_CadetRequest *pending_tail;
 116
 117   /**
 118    * Map from query to `struct GSF_CadetRequest`s waiting for
 119    * a reply.
 120    */
 121   struct GNUNET_CONTAINER_MultiHashMap *waiting_map;
 122
 123   /**
 124    * Channel to the other peer.
 125    */
 126   struct GNUNET_CADET_Channel *channel;
 127
 128   /**
 129    * Handle for active write operation, or NULL.
 130    */
 131   struct GNUNET_CADET_TransmitHandle *wh;
 132
 133   /**
 134    * Which peer does this cadet go to?
 135    */
 136   struct GNUNET_PeerIdentity target;
 137
 138   /**
 139    * Task to kill inactive cadets (we keep them around for
 140    * a few seconds to give the application a chance to give
 141    * us another query).
 142    */
 143   GNUNET_SCHEDULER_TaskIdentifier timeout_task;
 144
 145   /**
 146    * Task to reset cadets that had errors (asynchronously,
 147    * as we may not be able to do it immediately during a
 148    * callback from the cadet API).
 149    */
 150   GNUNET_SCHEDULER_TaskIdentifier reset_task;
 151
 152 };
 153
 154
 155 /**
 156  * Cadet channel for creating outbound channels.
 157  */
 158 static struct GNUNET_CADET_Handle *cadet_handle;
 159
 160 /**
 161  * Map from peer identities to 'struct CadetHandles' with cadet
 162  * channels to those peers.
 163  */
 164 static struct GNUNET_CONTAINER_MultiPeerMap *cadet_map;
 165
 166
 167 /* ********************* client-side code ************************* */
 168
 169
 170 /**
 171  * Transmit pending requests via the cadet.
 172  *
 173  * @param mh cadet to process
 174  */
 175 static void
 176 transmit_pending (struct CadetHandle *mh);
 177
 178
 179 /**
 180  * Iterator called on each entry in a waiting map to
 181  * move it back to the pending list.
 182  *
 183  * @param cls the `struct CadetHandle`
 184  * @param key the key of the entry in the map (the query)
 185  * @param value the `struct GSF_CadetRequest` to move to pending
 186  * @return #GNUNET_YES (continue to iterate)
 187  */
 188 static int
 189 move_to_pending (void *cls,
 190                  const struct GNUNET_HashCode *key,
 191                  void *value)
 192 {
 193   struct CadetHandle *mh = cls;
 194   struct GSF_CadetRequest *sr = value;
 195
 196   GNUNET_assert (GNUNET_YES ==
 197                  GNUNET_CONTAINER_multihashmap_remove (mh->waiting_map,
 198                                                        key,
 199                                                        value));
 200   GNUNET_CONTAINER_DLL_insert (mh->pending_head,
 201                                mh->pending_tail,
 202                                sr);
 203   sr->was_transmitted = GNUNET_NO;
 204   return GNUNET_YES;
 205 }
 206
 207
 208 /**
 209  * We had a serious error, tear down and re-create cadet from scratch.
 210  *
 211  * @param mh cadet to reset
 212  */
 213 static void
 214 reset_cadet (struct CadetHandle *mh)
 215 {
 216   struct GNUNET_CADET_Channel *channel = mh->channel;
 217
 218   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 219               "Resetting cadet channel to %s\n",
 220               GNUNET_i2s (&mh->target));
 221   mh->channel = NULL;
 222
 223   if (NULL != channel)
 224   {
 225     /* Avoid loop */
 226     if ( NULL != mh->wh)
 227         {
 228         GNUNET_CADET_cancel_notify(mh->wh);
 229                 mh->wh = NULL;
 230         }
 231     GNUNET_CADET_channel_destroy (channel);
 232   }
 233   GNUNET_CONTAINER_multihashmap_iterate (mh->waiting_map,
 234                                          &move_to_pending,
 235                                          mh);
 236   mh->channel = GNUNET_CADET_channel_create (cadet_handle,
 237                                           mh,
 238                                           &mh->target,
 239                                           GNUNET_APPLICATION_TYPE_FS_BLOCK_TRANSFER,
 240                                           GNUNET_CADET_OPTION_RELIABLE);
 241   transmit_pending (mh);
 242 }
 243
 244
 245 /**
 246  * Task called when it is time to destroy an inactive cadet channel.
 247  *
 248  * @param cls the `struct CadetHandle` to tear down
 249  * @param tc scheduler context, unused
 250  */
 251 static void
 252 cadet_timeout (void *cls,
 253               const struct GNUNET_SCHEDULER_TaskContext *tc)
 254 {
 255   struct CadetHandle *mh = cls;
 256   struct GNUNET_CADET_Channel *tun;
 257
 258   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 259               "Timeout on cadet channel to %s\n",
 260               GNUNET_i2s (&mh->target));
 261   mh->timeout_task = GNUNET_SCHEDULER_NO_TASK;
 262   tun = mh->channel;
 263   mh->channel = NULL;
 264   GNUNET_CADET_channel_destroy (tun);
 265 }
 266
 267
 268 /**
 269  * Task called when it is time to reset an cadet.
 270  *
 271  * @param cls the `struct CadetHandle` to tear down
 272  * @param tc scheduler context, unused
 273  */
 274 static void
 275 reset_cadet_task (void *cls,
 276                  const struct GNUNET_SCHEDULER_TaskContext *tc)
 277 {
 278   struct CadetHandle *mh = cls;
 279
 280   mh->reset_task = GNUNET_SCHEDULER_NO_TASK;
 281   reset_cadet (mh);
 282 }
 283
 284
 285 /**
 286  * We had a serious error, tear down and re-create cadet from scratch,
 287  * but do so asynchronously.
 288  *
 289  * @param mh cadet to reset
 290  */
 291 static void
 292 reset_cadet_async (struct CadetHandle *mh)
 293 {
 294   if (GNUNET_SCHEDULER_NO_TASK != mh->reset_task)
 295     GNUNET_SCHEDULER_cancel (mh->reset_task);
 296   mh->reset_task = GNUNET_SCHEDULER_add_now (&reset_cadet_task,
 297                                              mh);
 298 }
 299
 300
 301 /**
 302  * Functions of this signature are called whenever we are ready to transmit
 303  * query via a cadet.
 304  *
 305  * @param cls the struct CadetHandle for which we did the write call
 306  * @param size the number of bytes that can be written to @a buf
 307  * @param buf where to write the message
 308  * @return number of bytes written to @a buf
 309  */
 310 static size_t
 311 transmit_sqm (void *cls,
 312               size_t size,
 313               void *buf)
 314 {
 315   struct CadetHandle *mh = cls;
 316   struct CadetQueryMessage sqm;
 317   struct GSF_CadetRequest *sr;
 318
 319   mh->wh = NULL;
 320   if (NULL == buf)
 321   {
 322     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 323                 "Cadet channel to %s failed during transmission attempt, rebuilding\n",
 324                 GNUNET_i2s (&mh->target));
 325     reset_cadet_async (mh);
 326     return 0;
 327   }
 328   sr = mh->pending_head;
 329   if (NULL == sr)
 330     return 0;
 331   GNUNET_assert (size >= sizeof (struct CadetQueryMessage));
 332   GNUNET_CONTAINER_DLL_remove (mh->pending_head,
 333                                mh->pending_tail,
 334                                sr);
 335   GNUNET_assert (GNUNET_OK ==
 336                  GNUNET_CONTAINER_multihashmap_put (mh->waiting_map,
 337                                                     &sr->query,
 338                                                     sr,
 339                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
 340   sr->was_transmitted = GNUNET_YES;
 341   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 342               "Sending query for %s via cadet to %s\n",
 343               GNUNET_h2s (&sr->query),
 344               GNUNET_i2s (&mh->target));
 345   sqm.header.size = htons (sizeof (sqm));
 346   sqm.header.type = htons (GNUNET_MESSAGE_TYPE_FS_CADET_QUERY);
 347   sqm.type = htonl (sr->type);
 348   sqm.query = sr->query;
 349   memcpy (buf, &sqm, sizeof (sqm));
 350   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 351               "Successfully transmitted %u bytes via cadet to %s\n",
 352               (unsigned int) size,
 353               GNUNET_i2s (&mh->target));
 354   transmit_pending (mh);
 355   return sizeof (sqm);
 356 }
 357
 358
 359 /**
 360  * Transmit pending requests via the cadet.
 361  *
 362  * @param mh cadet to process
 363  */
 364 static void
 365 transmit_pending (struct CadetHandle *mh)
 366 {
 367   if (NULL == mh->channel)
 368     return;
 369   if (NULL != mh->wh)
 370     return;
 371   mh->wh = GNUNET_CADET_notify_transmit_ready (mh->channel, GNUNET_YES /* allow cork */,
 372                                               GNUNET_TIME_UNIT_FOREVER_REL,
 373                                               sizeof (struct CadetQueryMessage),
 374                                               &transmit_sqm, mh);
 375 }
 376
 377
 378 /**
 379  * Closure for handle_reply().
 380  */
 381 struct HandleReplyClosure
 382 {
 383
 384   /**
 385    * Reply payload.
 386    */
 387   const void *data;
 388
 389   /**
 390    * Expiration time for the block.
 391    */
 392   struct GNUNET_TIME_Absolute expiration;
 393
 394   /**
 395    * Number of bytes in 'data'.
 396    */
 397   size_t data_size;
 398
 399   /**
 400    * Type of the block.
 401    */
 402   enum GNUNET_BLOCK_Type type;
 403
 404   /**
 405    * Did we have a matching query?
 406    */
 407   int found;
 408 };
 409
 410
 411 /**
 412  * Iterator called on each entry in a waiting map to
 413  * process a result.
 414  *
 415  * @param cls the `struct HandleReplyClosure`
 416  * @param key the key of the entry in the map (the query)
 417  * @param value the `struct GSF_CadetRequest` to handle result for
 418  * @return #GNUNET_YES (continue to iterate)
 419  */
 420 static int
 421 handle_reply (void *cls,
 422               const struct GNUNET_HashCode *key,
 423               void *value)
 424 {
 425   struct HandleReplyClosure *hrc = cls;
 426   struct GSF_CadetRequest *sr = value;
 427
 428   sr->proc (sr->proc_cls,
 429             hrc->type,
 430             hrc->expiration,
 431             hrc->data_size,
 432             hrc->data);
 433   sr->proc = NULL;
 434   GSF_cadet_query_cancel (sr);
 435   hrc->found = GNUNET_YES;
 436   return GNUNET_YES;
 437 }
 438
 439
 440 /**
 441  * Functions with this signature are called whenever a complete reply
 442  * is received.
 443  *
 444  * @param cls closure with the `struct CadetHandle`
 445  * @param channel channel handle
 446  * @param channel_ctx channel context
 447  * @param message the actual message
 448  * @return #GNUNET_OK on success, #GNUNET_SYSERR to stop further processing
 449  */
 450 static int
 451 reply_cb (void *cls,
 452           struct GNUNET_CADET_Channel *channel,
 453           void **channel_ctx,
 454           const struct GNUNET_MessageHeader *message)
 455 {
 456   struct CadetHandle *mh = *channel_ctx;
 457   const struct CadetReplyMessage *srm;
 458   struct HandleReplyClosure hrc;
 459   uint16_t msize;
 460   enum GNUNET_BLOCK_Type type;
 461   struct GNUNET_HashCode query;
 462
 463   msize = ntohs (message->size);
 464   if (sizeof (struct CadetReplyMessage) > msize)
 465   {
 466     GNUNET_break_op (0);
 467     reset_cadet_async (mh);
 468     return GNUNET_SYSERR;
 469   }
 470   srm = (const struct CadetReplyMessage *) message;
 471   msize -= sizeof (struct CadetReplyMessage);
 472   type = (enum GNUNET_BLOCK_Type) ntohl (srm->type);
 473   if (GNUNET_YES !=
 474       GNUNET_BLOCK_get_key (GSF_block_ctx,
 475                             type,
 476                             &srm[1], msize, &query))
 477   {
 478     GNUNET_break_op (0);
 479     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 480                 "Received bogus reply of type %u with %u bytes via cadet from peer %s\n",
 481                 type,
 482                 msize,
 483                 GNUNET_i2s (&mh->target));
 484     reset_cadet_async (mh);
 485     return GNUNET_SYSERR;
 486   }
 487   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 488               "Received reply `%s' via cadet from peer %s\n",
 489               GNUNET_h2s (&query),
 490               GNUNET_i2s (&mh->target));
 491   GNUNET_CADET_receive_done (channel);
 492   GNUNET_STATISTICS_update (GSF_stats,
 493                             gettext_noop ("# replies received via cadet"), 1,
 494                             GNUNET_NO);
 495   hrc.data = &srm[1];
 496   hrc.data_size = msize;
 497   hrc.expiration = GNUNET_TIME_absolute_ntoh (srm->expiration);
 498   hrc.type = type;
 499   hrc.found = GNUNET_NO;
 500   GNUNET_CONTAINER_multihashmap_get_multiple (mh->waiting_map,
 501                                               &query,
 502                                               &handle_reply,
 503                                               &hrc);
 504   if (GNUNET_NO == hrc.found)
 505   {
 506     GNUNET_STATISTICS_update (GSF_stats,
 507                               gettext_noop ("# replies received via cadet dropped"), 1,
 508                               GNUNET_NO);
 509     return GNUNET_OK;
 510   }
 511   return GNUNET_OK;
 512 }
 513
 514
 515 /**
 516  * Get (or create) a cadet to talk to the given peer.
 517  *
 518  * @param target peer we want to communicate with
 519  */
 520 static struct CadetHandle *
 521 get_cadet (const struct GNUNET_PeerIdentity *target)
 522 {
 523   struct CadetHandle *mh;
 524
 525   mh = GNUNET_CONTAINER_multipeermap_get (cadet_map,
 526                                           target);
 527   if (NULL != mh)
 528   {
 529     if (GNUNET_SCHEDULER_NO_TASK != mh->timeout_task)
 530     {
 531       GNUNET_SCHEDULER_cancel (mh->timeout_task);
 532       mh->timeout_task = GNUNET_SCHEDULER_NO_TASK;
 533     }
 534     return mh;
 535   }
 536   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 537               "Creating cadet channel to %s\n",
 538               GNUNET_i2s (target));
 539   mh = GNUNET_new (struct CadetHandle);
 540   mh->reset_task = GNUNET_SCHEDULER_add_delayed (CLIENT_RETRY_TIMEOUT,
 541                                                  &reset_cadet_task,
 542                                                  mh);
 543   mh->waiting_map = GNUNET_CONTAINER_multihashmap_create (16, GNUNET_YES);
 544   mh->target = *target;
 545   GNUNET_assert (GNUNET_OK ==
 546                  GNUNET_CONTAINER_multipeermap_put (cadet_map,
 547                                                     &mh->target,
 548                                                     mh,
 549                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 550   mh->channel = GNUNET_CADET_channel_create (cadet_handle,
 551                                             mh,
 552                                             &mh->target,
 553                                             GNUNET_APPLICATION_TYPE_FS_BLOCK_TRANSFER,
 554                                             GNUNET_CADET_OPTION_RELIABLE);
 555   GNUNET_assert (mh ==
 556                  GNUNET_CONTAINER_multipeermap_get (cadet_map,
 557                                                     target));
 558   return mh;
 559 }
 560
 561
 562 /**
 563  * Look for a block by directly contacting a particular peer.
 564  *
 565  * @param target peer that should have the block
 566  * @param query hash to query for the block
 567  * @param type desired type for the block
 568  * @param proc function to call with result
 569  * @param proc_cls closure for @a proc
 570  * @return handle to cancel the operation
 571  */
 572 struct GSF_CadetRequest *
 573 GSF_cadet_query (const struct GNUNET_PeerIdentity *target,
 574                 const struct GNUNET_HashCode *query,
 575                 enum GNUNET_BLOCK_Type type,
 576                 GSF_CadetReplyProcessor proc, void *proc_cls)
 577 {
 578   struct CadetHandle *mh;
 579   struct GSF_CadetRequest *sr;
 580
 581   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 582               "Preparing to send query for %s via cadet to %s\n",
 583               GNUNET_h2s (query),
 584               GNUNET_i2s (target));
 585   mh = get_cadet (target);
 586   sr = GNUNET_new (struct GSF_CadetRequest);
 587   sr->mh = mh;
 588   sr->proc = proc;
 589   sr->proc_cls = proc_cls;
 590   sr->type = type;
 591   sr->query = *query;
 592   GNUNET_CONTAINER_DLL_insert (mh->pending_head,
 593                                mh->pending_tail,
 594                                sr);
 595   transmit_pending (mh);
 596   return sr;
 597 }
 598
 599
 600 /**
 601  * Cancel an active request; must not be called after 'proc'
 602  * was calld.
 603  *
 604  * @param sr request to cancel
 605  */
 606 void
 607 GSF_cadet_query_cancel (struct GSF_CadetRequest *sr)
 608 {
 609   struct CadetHandle *mh = sr->mh;
 610   GSF_CadetReplyProcessor p;
 611
 612   p = sr->proc;
 613   sr->proc = NULL;
 614   if (NULL != p)
 615   {
 616     /* signal failure / cancellation to callback */
 617     p (sr->proc_cls, GNUNET_BLOCK_TYPE_ANY,
 618        GNUNET_TIME_UNIT_ZERO_ABS,
 619        0, NULL);
 620   }
 621   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 622               "Cancelled query for %s via cadet to %s\n",
 623               GNUNET_h2s (&sr->query),
 624               GNUNET_i2s (&sr->mh->target));
 625   if (GNUNET_YES == sr->was_transmitted)
 626     GNUNET_assert (GNUNET_OK ==
 627                    GNUNET_CONTAINER_multihashmap_remove (mh->waiting_map,
 628                                                          &sr->query,
 629                                                          sr));
 630   else
 631     GNUNET_CONTAINER_DLL_remove (mh->pending_head,
 632                                  mh->pending_tail,
 633                                  sr);
 634   GNUNET_free (sr);
 635   if ( (0 == GNUNET_CONTAINER_multihashmap_size (mh->waiting_map)) &&
 636        (NULL == mh->pending_head) )
 637     mh->timeout_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
 638                                                      &cadet_timeout,
 639                                                      mh);
 640 }
 641
 642
 643 /**
 644  * Iterator called on each entry in a waiting map to
 645  * call the 'proc' continuation and release associated
 646  * resources.
 647  *
 648  * @param cls the `struct CadetHandle`
 649  * @param key the key of the entry in the map (the query)
 650  * @param value the `struct GSF_CadetRequest` to clean up
 651  * @return #GNUNET_YES (continue to iterate)
 652  */
 653 static int
 654 free_waiting_entry (void *cls,
 655                     const struct GNUNET_HashCode *key,
 656                     void *value)
 657 {
 658   struct GSF_CadetRequest *sr = value;
 659
 660   GSF_cadet_query_cancel (sr);
 661   return GNUNET_YES;
 662 }
 663
 664
 665 /**
 666  * Function called by cadet when a client disconnects.
 667  * Cleans up our `struct CadetClient` of that channel.
 668  *
 669  * @param cls NULL
 670  * @param channel channel of the disconnecting client
 671  * @param channel_ctx our `struct CadetClient`
 672  */
 673 static void
 674 cleaner_cb (void *cls,
 675             const struct GNUNET_CADET_Channel *channel,
 676             void *channel_ctx)
 677 {
 678   struct CadetHandle *mh = channel_ctx;
 679   struct GSF_CadetRequest *sr;
 680
 681   if (NULL == mh->channel)
 682     return; /* being destroyed elsewhere */
 683   GNUNET_assert (channel == mh->channel);
 684   mh->channel = NULL;
 685   while (NULL != (sr = mh->pending_head))
 686     GSF_cadet_query_cancel (sr);
 687   /* first remove `mh` from the `cadet_map`, so that if the
 688      callback from `free_waiting_entry()` happens to re-issue
 689      the request, we don't immediately have it back in the
 690      `waiting_map`. */
 691   GNUNET_assert (GNUNET_OK ==
 692                  GNUNET_CONTAINER_multipeermap_remove (cadet_map,
 693                                                        &mh->target,
 694                                                        mh));
 695   GNUNET_CONTAINER_multihashmap_iterate (mh->waiting_map,
 696                                          &free_waiting_entry,
 697                                          mh);
 698   if (NULL != mh->wh)
 699     GNUNET_CADET_notify_transmit_ready_cancel (mh->wh);
 700   if (GNUNET_SCHEDULER_NO_TASK != mh->timeout_task)
 701     GNUNET_SCHEDULER_cancel (mh->timeout_task);
 702   if (GNUNET_SCHEDULER_NO_TASK != mh->reset_task)
 703     GNUNET_SCHEDULER_cancel (mh->reset_task);
 704   GNUNET_assert (0 ==
 705                  GNUNET_CONTAINER_multihashmap_size (mh->waiting_map));
 706   GNUNET_CONTAINER_multihashmap_destroy (mh->waiting_map);
 707   GNUNET_free (mh);
 708 }
 709
 710
 711 /**
 712  * Initialize subsystem for non-anonymous file-sharing.
 713  */
 714 void
 715 GSF_cadet_start_client ()
 716 {
 717   static const struct GNUNET_CADET_MessageHandler handlers[] = {
 718     { &reply_cb, GNUNET_MESSAGE_TYPE_FS_CADET_REPLY, 0 },
 719     { NULL, 0, 0 }
 720   };
 721
 722   cadet_map = GNUNET_CONTAINER_multipeermap_create (16, GNUNET_YES);
 723   cadet_handle = GNUNET_CADET_connect (GSF_cfg,
 724                                      NULL,
 725                                      NULL,
 726                                      &cleaner_cb,
 727                                      handlers,
 728                                      NULL);
 729 }
 730
 731
 732 /**
 733  * Function called on each active cadets to shut them down.
 734  *
 735  * @param cls NULL
 736  * @param key target peer, unused
 737  * @param value the `struct CadetHandle` to destroy
 738  * @return #GNUNET_YES (continue to iterate)
 739  */
 740 static int
 741 release_cadets (void *cls,
 742                const struct GNUNET_PeerIdentity *key,
 743                void *value)
 744 {
 745   struct CadetHandle *mh = value;
 746
 747   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 748               "Timeout on cadet channel to %s\n",
 749               GNUNET_i2s (&mh->target));
 750   if (NULL != mh->channel)
 751     GNUNET_CADET_channel_destroy (mh->channel);
 752   return GNUNET_YES;
 753 }
 754
 755
 756 /**
 757  * Shutdown subsystem for non-anonymous file-sharing.
 758  */
 759 void
 760 GSF_cadet_stop_client ()
 761 {
 762   GNUNET_CONTAINER_multipeermap_iterate (cadet_map,
 763                                          &release_cadets,
 764                                          NULL);
 765   GNUNET_CONTAINER_multipeermap_destroy (cadet_map);
 766   cadet_map = NULL;
 767   if (NULL != cadet_handle)
 768   {
 769     GNUNET_CADET_disconnect (cadet_handle);
 770     cadet_handle = NULL;
 771   }
 772 }
 773
 774
 775 /* end of gnunet-service-fs_cadet_client.c */