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