src/fs/gnunet-service-fs_cadet_client.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2012, 2013 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 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   struct GNUNET_SCHEDULER_Task * 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   struct GNUNET_SCHEDULER_Task * 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_notify_transmit_ready_cancel (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  */
 250 static void
 251 cadet_timeout (void *cls)
 252 {
 253   struct CadetHandle *mh = cls;
 254   struct GNUNET_CADET_Channel *tun;
 255
 256   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 257               "Timeout on cadet channel to %s\n",
 258               GNUNET_i2s (&mh->target));
 259   mh->timeout_task = NULL;
 260   tun = mh->channel;
 261   mh->channel = NULL;
 262   if(NULL != tun)
 263         GNUNET_CADET_channel_destroy (tun);
 264 }
 265
 266
 267 /**
 268  * Task called when it is time to reset an cadet.
 269  *
 270  * @param cls the `struct CadetHandle` to tear down
 271  */
 272 static void
 273 reset_cadet_task (void *cls)
 274 {
 275   struct CadetHandle *mh = cls;
 276
 277   mh->reset_task = NULL;
 278   reset_cadet (mh);
 279 }
 280
 281
 282 /**
 283  * We had a serious error, tear down and re-create cadet from scratch,
 284  * but do so asynchronously.
 285  *
 286  * @param mh cadet to reset
 287  */
 288 static void
 289 reset_cadet_async (struct CadetHandle *mh)
 290 {
 291   if (NULL != mh->reset_task)
 292     GNUNET_SCHEDULER_cancel (mh->reset_task);
 293   mh->reset_task = GNUNET_SCHEDULER_add_now (&reset_cadet_task,
 294                                              mh);
 295 }
 296
 297
 298 /**
 299  * Functions of this signature are called whenever we are ready to transmit
 300  * query via a cadet.
 301  *
 302  * @param cls the struct CadetHandle for which we did the write call
 303  * @param size the number of bytes that can be written to @a buf
 304  * @param buf where to write the message
 305  * @return number of bytes written to @a buf
 306  */
 307 static size_t
 308 transmit_sqm (void *cls,
 309               size_t size,
 310               void *buf)
 311 {
 312   struct CadetHandle *mh = cls;
 313   struct CadetQueryMessage sqm;
 314   struct GSF_CadetRequest *sr;
 315
 316   mh->wh = NULL;
 317   if (NULL == buf)
 318   {
 319     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 320                 "Cadet channel to %s failed during transmission attempt, rebuilding\n",
 321                 GNUNET_i2s (&mh->target));
 322     reset_cadet_async (mh);
 323     return 0;
 324   }
 325   sr = mh->pending_head;
 326   if (NULL == sr)
 327     return 0;
 328   GNUNET_assert (size >= sizeof (struct CadetQueryMessage));
 329   GNUNET_CONTAINER_DLL_remove (mh->pending_head,
 330                                mh->pending_tail,
 331                                sr);
 332   GNUNET_assert (GNUNET_OK ==
 333                  GNUNET_CONTAINER_multihashmap_put (mh->waiting_map,
 334                                                     &sr->query,
 335                                                     sr,
 336                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
 337   sr->was_transmitted = GNUNET_YES;
 338   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 339               "Sending query for %s via cadet to %s\n",
 340               GNUNET_h2s (&sr->query),
 341               GNUNET_i2s (&mh->target));
 342   sqm.header.size = htons (sizeof (sqm));
 343   sqm.header.type = htons (GNUNET_MESSAGE_TYPE_FS_CADET_QUERY);
 344   sqm.type = htonl (sr->type);
 345   sqm.query = sr->query;
 346   GNUNET_memcpy (buf, &sqm, sizeof (sqm));
 347   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 348               "Successfully transmitted %u bytes via cadet to %s\n",
 349               (unsigned int) size,
 350               GNUNET_i2s (&mh->target));
 351   transmit_pending (mh);
 352   return sizeof (sqm);
 353 }
 354
 355
 356 /**
 357  * Transmit pending requests via the cadet.
 358  *
 359  * @param mh cadet to process
 360  */
 361 static void
 362 transmit_pending (struct CadetHandle *mh)
 363 {
 364   if (NULL == mh->channel)
 365     return;
 366   if (NULL != mh->wh)
 367     return;
 368   mh->wh = GNUNET_CADET_notify_transmit_ready (mh->channel, GNUNET_YES /* allow cork */,
 369                                               GNUNET_TIME_UNIT_FOREVER_REL,
 370                                               sizeof (struct CadetQueryMessage),
 371                                               &transmit_sqm, mh);
 372 }
 373
 374
 375 /**
 376  * Closure for handle_reply().
 377  */
 378 struct HandleReplyClosure
 379 {
 380
 381   /**
 382    * Reply payload.
 383    */
 384   const void *data;
 385
 386   /**
 387    * Expiration time for the block.
 388    */
 389   struct GNUNET_TIME_Absolute expiration;
 390
 391   /**
 392    * Number of bytes in 'data'.
 393    */
 394   size_t data_size;
 395
 396   /**
 397    * Type of the block.
 398    */
 399   enum GNUNET_BLOCK_Type type;
 400
 401   /**
 402    * Did we have a matching query?
 403    */
 404   int found;
 405 };
 406
 407
 408 /**
 409  * Iterator called on each entry in a waiting map to
 410  * process a result.
 411  *
 412  * @param cls the `struct HandleReplyClosure`
 413  * @param key the key of the entry in the map (the query)
 414  * @param value the `struct GSF_CadetRequest` to handle result for
 415  * @return #GNUNET_YES (continue to iterate)
 416  */
 417 static int
 418 handle_reply (void *cls,
 419               const struct GNUNET_HashCode *key,
 420               void *value)
 421 {
 422   struct HandleReplyClosure *hrc = cls;
 423   struct GSF_CadetRequest *sr = value;
 424
 425   sr->proc (sr->proc_cls,
 426             hrc->type,
 427             hrc->expiration,
 428             hrc->data_size,
 429             hrc->data);
 430   sr->proc = NULL;
 431   GSF_cadet_query_cancel (sr);
 432   hrc->found = GNUNET_YES;
 433   return GNUNET_YES;
 434 }
 435
 436
 437 /**
 438  * Functions with this signature are called whenever a complete reply
 439  * is received.
 440  *
 441  * @param cls closure with the `struct CadetHandle`
 442  * @param channel channel handle
 443  * @param channel_ctx channel context
 444  * @param message the actual message
 445  * @return #GNUNET_OK on success, #GNUNET_SYSERR to stop further processing
 446  */
 447 static int
 448 reply_cb (void *cls,
 449           struct GNUNET_CADET_Channel *channel,
 450           void **channel_ctx,
 451           const struct GNUNET_MessageHeader *message)
 452 {
 453   struct CadetHandle *mh = *channel_ctx;
 454   const struct CadetReplyMessage *srm;
 455   struct HandleReplyClosure hrc;
 456   uint16_t msize;
 457   enum GNUNET_BLOCK_Type type;
 458   struct GNUNET_HashCode query;
 459
 460   msize = ntohs (message->size);
 461   if (sizeof (struct CadetReplyMessage) > msize)
 462   {
 463     GNUNET_break_op (0);
 464     reset_cadet_async (mh);
 465     return GNUNET_SYSERR;
 466   }
 467   srm = (const struct CadetReplyMessage *) message;
 468   msize -= sizeof (struct CadetReplyMessage);
 469   type = (enum GNUNET_BLOCK_Type) ntohl (srm->type);
 470   if (GNUNET_YES !=
 471       GNUNET_BLOCK_get_key (GSF_block_ctx,
 472                             type,
 473                             &srm[1], msize, &query))
 474   {
 475     GNUNET_break_op (0);
 476     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 477                 "Received bogus reply of type %u with %u bytes via cadet from peer %s\n",
 478                 type,
 479                 msize,
 480                 GNUNET_i2s (&mh->target));
 481     reset_cadet_async (mh);
 482     return GNUNET_SYSERR;
 483   }
 484   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 485               "Received reply `%s' via cadet from peer %s\n",
 486               GNUNET_h2s (&query),
 487               GNUNET_i2s (&mh->target));
 488   GNUNET_CADET_receive_done (channel);
 489   GNUNET_STATISTICS_update (GSF_stats,
 490                             gettext_noop ("# replies received via cadet"), 1,
 491                             GNUNET_NO);
 492   hrc.data = &srm[1];
 493   hrc.data_size = msize;
 494   hrc.expiration = GNUNET_TIME_absolute_ntoh (srm->expiration);
 495   hrc.type = type;
 496   hrc.found = GNUNET_NO;
 497   GNUNET_CONTAINER_multihashmap_get_multiple (mh->waiting_map,
 498                                               &query,
 499                                               &handle_reply,
 500                                               &hrc);
 501   if (GNUNET_NO == hrc.found)
 502   {
 503     GNUNET_STATISTICS_update (GSF_stats,
 504                               gettext_noop ("# replies received via cadet dropped"), 1,
 505                               GNUNET_NO);
 506     return GNUNET_OK;
 507   }
 508   return GNUNET_OK;
 509 }
 510
 511
 512 /**
 513  * Get (or create) a cadet to talk to the given peer.
 514  *
 515  * @param target peer we want to communicate with
 516  */
 517 static struct CadetHandle *
 518 get_cadet (const struct GNUNET_PeerIdentity *target)
 519 {
 520   struct CadetHandle *mh;
 521
 522   mh = GNUNET_CONTAINER_multipeermap_get (cadet_map,
 523                                           target);
 524   if (NULL != mh)
 525   {
 526     if (NULL != mh->timeout_task)
 527     {
 528       GNUNET_SCHEDULER_cancel (mh->timeout_task);
 529       mh->timeout_task = NULL;
 530     }
 531     return mh;
 532   }
 533   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 534               "Creating cadet channel to %s\n",
 535               GNUNET_i2s (target));
 536   mh = GNUNET_new (struct CadetHandle);
 537   mh->reset_task = GNUNET_SCHEDULER_add_delayed (CLIENT_RETRY_TIMEOUT,
 538                                                  &reset_cadet_task,
 539                                                  mh);
 540   mh->waiting_map = GNUNET_CONTAINER_multihashmap_create (16, GNUNET_YES);
 541   mh->target = *target;
 542   GNUNET_assert (GNUNET_OK ==
 543                  GNUNET_CONTAINER_multipeermap_put (cadet_map,
 544                                                     &mh->target,
 545                                                     mh,
 546                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 547   mh->channel = GNUNET_CADET_channel_create (cadet_handle,
 548                                             mh,
 549                                             &mh->target,
 550                                             GNUNET_APPLICATION_TYPE_FS_BLOCK_TRANSFER,
 551                                             GNUNET_CADET_OPTION_RELIABLE);
 552   GNUNET_assert (mh ==
 553                  GNUNET_CONTAINER_multipeermap_get (cadet_map,
 554                                                     target));
 555   return mh;
 556 }
 557
 558
 559 /**
 560  * Look for a block by directly contacting a particular peer.
 561  *
 562  * @param target peer that should have the block
 563  * @param query hash to query for the block
 564  * @param type desired type for the block
 565  * @param proc function to call with result
 566  * @param proc_cls closure for @a proc
 567  * @return handle to cancel the operation
 568  */
 569 struct GSF_CadetRequest *
 570 GSF_cadet_query (const struct GNUNET_PeerIdentity *target,
 571                 const struct GNUNET_HashCode *query,
 572                 enum GNUNET_BLOCK_Type type,
 573                 GSF_CadetReplyProcessor proc, void *proc_cls)
 574 {
 575   struct CadetHandle *mh;
 576   struct GSF_CadetRequest *sr;
 577
 578   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 579               "Preparing to send query for %s via cadet to %s\n",
 580               GNUNET_h2s (query),
 581               GNUNET_i2s (target));
 582   mh = get_cadet (target);
 583   sr = GNUNET_new (struct GSF_CadetRequest);
 584   sr->mh = mh;
 585   sr->proc = proc;
 586   sr->proc_cls = proc_cls;
 587   sr->type = type;
 588   sr->query = *query;
 589   GNUNET_CONTAINER_DLL_insert (mh->pending_head,
 590                                mh->pending_tail,
 591                                sr);
 592   transmit_pending (mh);
 593   return sr;
 594 }
 595
 596
 597 /**
 598  * Cancel an active request; must not be called after 'proc'
 599  * was calld.
 600  *
 601  * @param sr request to cancel
 602  */
 603 void
 604 GSF_cadet_query_cancel (struct GSF_CadetRequest *sr)
 605 {
 606   struct CadetHandle *mh = sr->mh;
 607   GSF_CadetReplyProcessor p;
 608
 609   p = sr->proc;
 610   sr->proc = NULL;
 611   if (NULL != p)
 612   {
 613     /* signal failure / cancellation to callback */
 614     p (sr->proc_cls, GNUNET_BLOCK_TYPE_ANY,
 615        GNUNET_TIME_UNIT_ZERO_ABS,
 616        0, NULL);
 617   }
 618   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 619               "Cancelled query for %s via cadet to %s\n",
 620               GNUNET_h2s (&sr->query),
 621               GNUNET_i2s (&sr->mh->target));
 622   if (GNUNET_YES == sr->was_transmitted)
 623     GNUNET_assert (GNUNET_OK ==
 624                    GNUNET_CONTAINER_multihashmap_remove (mh->waiting_map,
 625                                                          &sr->query,
 626                                                          sr));
 627   else
 628     GNUNET_CONTAINER_DLL_remove (mh->pending_head,
 629                                  mh->pending_tail,
 630                                  sr);
 631   GNUNET_free (sr);
 632   if ( (0 == GNUNET_CONTAINER_multihashmap_size (mh->waiting_map)) &&
 633        (NULL == mh->pending_head) )
 634     mh->timeout_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
 635                                                      &cadet_timeout,
 636                                                      mh);
 637 }
 638
 639
 640 /**
 641  * Iterator called on each entry in a waiting map to
 642  * call the 'proc' continuation and release associated
 643  * resources.
 644  *
 645  * @param cls the `struct CadetHandle`
 646  * @param key the key of the entry in the map (the query)
 647  * @param value the `struct GSF_CadetRequest` to clean up
 648  * @return #GNUNET_YES (continue to iterate)
 649  */
 650 static int
 651 free_waiting_entry (void *cls,
 652                     const struct GNUNET_HashCode *key,
 653                     void *value)
 654 {
 655   struct GSF_CadetRequest *sr = value;
 656
 657   GSF_cadet_query_cancel (sr);
 658   return GNUNET_YES;
 659 }
 660
 661
 662 /**
 663  * Function called by cadet when a client disconnects.
 664  * Cleans up our `struct CadetClient` of that channel.
 665  *
 666  * @param cls NULL
 667  * @param channel channel of the disconnecting client
 668  * @param channel_ctx our `struct CadetClient`
 669  */
 670 static void
 671 cleaner_cb (void *cls,
 672             const struct GNUNET_CADET_Channel *channel,
 673             void *channel_ctx)
 674 {
 675   struct CadetHandle *mh = channel_ctx;
 676   struct GSF_CadetRequest *sr;
 677
 678   if (NULL == mh->channel)
 679     return; /* being destroyed elsewhere */
 680   GNUNET_assert (channel == mh->channel);
 681   mh->channel = NULL;
 682   while (NULL != (sr = mh->pending_head))
 683     GSF_cadet_query_cancel (sr);
 684   /* first remove `mh` from the `cadet_map`, so that if the
 685      callback from `free_waiting_entry()` happens to re-issue
 686      the request, we don't immediately have it back in the
 687      `waiting_map`. */
 688   GNUNET_assert (GNUNET_OK ==
 689                  GNUNET_CONTAINER_multipeermap_remove (cadet_map,
 690                                                        &mh->target,
 691                                                        mh));
 692   GNUNET_CONTAINER_multihashmap_iterate (mh->waiting_map,
 693                                          &free_waiting_entry,
 694                                          mh);
 695   if (NULL != mh->wh)
 696     GNUNET_CADET_notify_transmit_ready_cancel (mh->wh);
 697   if (NULL != mh->timeout_task)
 698     GNUNET_SCHEDULER_cancel (mh->timeout_task);
 699   if (NULL != mh->reset_task)
 700     GNUNET_SCHEDULER_cancel (mh->reset_task);
 701   GNUNET_assert (0 ==
 702                  GNUNET_CONTAINER_multihashmap_size (mh->waiting_map));
 703   GNUNET_CONTAINER_multihashmap_destroy (mh->waiting_map);
 704   GNUNET_free (mh);
 705 }
 706
 707
 708 /**
 709  * Initialize subsystem for non-anonymous file-sharing.
 710  */
 711 void
 712 GSF_cadet_start_client ()
 713 {
 714   static const struct GNUNET_CADET_MessageHandler handlers[] = {
 715     { &reply_cb, GNUNET_MESSAGE_TYPE_FS_CADET_REPLY, 0 },
 716     { NULL, 0, 0 }
 717   };
 718
 719   cadet_map = GNUNET_CONTAINER_multipeermap_create (16, GNUNET_YES);
 720   cadet_handle = GNUNET_CADET_connect (GSF_cfg,
 721                                      NULL,
 722                                      NULL,
 723                                      &cleaner_cb,
 724                                      handlers,
 725                                      NULL);
 726 }
 727
 728
 729 /**
 730  * Function called on each active cadets to shut them down.
 731  *
 732  * @param cls NULL
 733  * @param key target peer, unused
 734  * @param value the `struct CadetHandle` to destroy
 735  * @return #GNUNET_YES (continue to iterate)
 736  */
 737 static int
 738 release_cadets (void *cls,
 739                const struct GNUNET_PeerIdentity *key,
 740                void *value)
 741 {
 742   struct CadetHandle *mh = value;
 743
 744   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 745               "Timeout on cadet channel to %s\n",
 746               GNUNET_i2s (&mh->target));
 747   if (NULL != mh->channel)
 748     GNUNET_CADET_channel_destroy (mh->channel);
 749   return GNUNET_YES;
 750 }
 751
 752
 753 /**
 754  * Shutdown subsystem for non-anonymous file-sharing.
 755  */
 756 void
 757 GSF_cadet_stop_client ()
 758 {
 759   GNUNET_CONTAINER_multipeermap_iterate (cadet_map,
 760                                          &release_cadets,
 761                                          NULL);
 762   GNUNET_CONTAINER_multipeermap_destroy (cadet_map);
 763   cadet_map = NULL;
 764   if (NULL != cadet_handle)
 765   {
 766     GNUNET_CADET_disconnect (cadet_handle);
 767     cadet_handle = NULL;
 768   }
 769 }
 770
 771
 772 /* end of gnunet-service-fs_cadet_client.c */