src/transport/gnunet-service-transport_manipulation.c

   1 /*
   2  This file is part of GNUnet.
   3  Copyright (C) 2010-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 transport/gnunet-service-transport_manipulation.c
  23  * @brief transport component manipulation traffic for simulation
  24  * @author Christian Grothoff
  25  * @author Matthias Wachs
  26  */
  27 #include "platform.h"
  28 #include "gnunet-service-transport_blacklist.h"
  29 #include "gnunet-service-transport_clients.h"
  30 #include "gnunet-service-transport_hello.h"
  31 #include "gnunet-service-transport_neighbours.h"
  32 #include "gnunet-service-transport_plugins.h"
  33 #include "gnunet-service-transport_validation.h"
  34 #include "gnunet-service-transport.h"
  35 #include "transport.h"
  36
  37 enum TRAFFIC_METRIC_DIRECTION
  38 {
  39   TM_SEND = 0, TM_RECEIVE = 1, TM_BOTH = 2
  40 };
  41
  42
  43 /**
  44  * Struct containing information about manipulations to a specific peer
  45  */
  46 struct TM_Peer;
  47
  48 /**
  49  * Manipulation entry
  50  */
  51 struct PropManipulationEntry
  52 {
  53   /**
  54    * Next in DLL
  55    */
  56   struct PropManipulationEntry *next;
  57
  58   /**
  59    * Previous in DLL
  60    */
  61   struct PropManipulationEntry *prev;
  62
  63   /**
  64    * ATS type in HBO
  65    */
  66   uint32_t type;
  67
  68   /**
  69    * Value in HBO
  70    */
  71   uint32_t metrics[TM_BOTH];
  72
  73 };
  74
  75 /**
  76  * Struct containing information about manipulations to a specific peer
  77  */
  78 struct TM_Peer
  79 {
  80   /**
  81    * Peer ID
  82    */
  83   struct GNUNET_PeerIdentity peer;
  84
  85   struct PropManipulationEntry *head;
  86   struct PropManipulationEntry *tail;
  87
  88   /**
  89    * Peer specific manipulation metrics
  90    */
  91   uint32_t metrics[TM_BOTH][GNUNET_ATS_QualityPropertiesCount];
  92
  93   /**
  94    * Task to schedule delayed sendding
  95    */
  96   struct GNUNET_SCHEDULER_Task * send_delay_task;
  97
  98   /**
  99    * Send queue DLL head
 100    */
 101   struct DelayQueueEntry *send_head;
 102
 103   /**
 104    * Send queue DLL tail
 105    */
 106   struct DelayQueueEntry *send_tail;
 107 };
 108
 109
 110 struct GST_ManipulationHandle
 111 {
 112   /**
 113    * Hashmap contain all peers currently manipulated
 114    */
 115   struct GNUNET_CONTAINER_MultiPeerMap *peers;
 116
 117   /**
 118    * Peer containing information for general manipulation
 119    */
 120   struct TM_Peer general;
 121 };
 122
 123 /**
 124  * Entry in the delay queue for an outbound delayed message
 125  */
 126 struct DelayQueueEntry
 127 {
 128   /**
 129    * Next in DLL
 130    */
 131   struct DelayQueueEntry *prev;
 132
 133   /**
 134    * Previous in DLL
 135    */
 136   struct DelayQueueEntry *next;
 137
 138   /**
 139    * Peer this entry is belonging to
 140    * if (NULL == tmp): enqueued in generic DLL and scheduled by generic_send_delay_task
 141    * else: enqueued in tmp->send_head and tmp->send_tail and scheduled by tmp->send_delay_task
 142    */
 143   struct TM_Peer *tmp;
 144
 145   /**
 146    * Peer ID
 147    */
 148   struct GNUNET_PeerIdentity id;
 149
 150   /**
 151    * Absolute time when to send
 152    */
 153   struct GNUNET_TIME_Absolute sent_at;
 154
 155   /**
 156    * The message
 157    */
 158   void *msg;
 159
 160   /**
 161    * The message size
 162    */
 163   size_t msg_size;
 164
 165   /**
 166    * Message timeout
 167    */
 168   struct GNUNET_TIME_Relative timeout;
 169
 170   /**
 171    * Transports send continuation
 172    */
 173   GST_NeighbourSendContinuation cont;
 174
 175   /**
 176    * Transports send continuation cls
 177    */
 178   void *cont_cls;
 179 };
 180
 181 struct GST_ManipulationHandle man_handle;
 182
 183 /**
 184  * DLL head for delayed messages based on general delay
 185  */
 186 struct DelayQueueEntry *generic_dqe_head;
 187
 188 /**
 189  * DLL tail for delayed messages based on general delay
 190  */
 191 struct DelayQueueEntry *generic_dqe_tail;
 192
 193 /**
 194  * Task to schedule delayed sending based on general delay
 195  */
 196 struct GNUNET_SCHEDULER_Task * generic_send_delay_task;
 197
 198
 199 static void
 200 set_metric(struct TM_Peer *dest, int direction, uint32_t type, uint32_t value)
 201 {
 202   struct PropManipulationEntry *cur;
 203   for (cur = dest->head; NULL != cur; cur = cur->next)
 204     {
 205       if (cur->type == type)
 206         break;
 207     }
 208   if (NULL == cur)
 209     {
 210       cur = GNUNET_new (struct PropManipulationEntry);
 211       GNUNET_CONTAINER_DLL_insert(dest->head, dest->tail, cur);
 212       cur->type = type;
 213       cur->metrics[TM_SEND] = UINT32_MAX;
 214       cur->metrics[TM_RECEIVE] = UINT32_MAX;
 215     }
 216
 217   switch (direction)
 218     {
 219   case TM_BOTH:
 220     cur->metrics[TM_SEND] = value;
 221     cur->metrics[TM_RECEIVE] = value;
 222     break;
 223   case TM_SEND:
 224     cur->metrics[TM_SEND] = value;
 225     break;
 226   case TM_RECEIVE:
 227     cur->metrics[TM_RECEIVE] = value;
 228     break;
 229   default:
 230     break;
 231     }
 232 }
 233
 234
 235 static uint32_t
 236 find_metric(struct TM_Peer *dest, uint32_t type, int direction)
 237 {
 238   struct PropManipulationEntry *cur;
 239
 240   for (cur = dest->head; NULL != cur; cur = cur->next)
 241     {
 242       if (cur->type == type)
 243         return cur->metrics[direction];
 244
 245     }
 246   return UINT32_MAX;
 247 }
 248
 249
 250 /**
 251  * Clean up metrics for a peer
 252  */
 253 static void
 254 free_metric(struct TM_Peer *dest)
 255 {
 256   struct PropManipulationEntry *cur;
 257   struct PropManipulationEntry *next;
 258
 259   for (cur = dest->head; NULL != cur; cur = next)
 260     {
 261       next = cur->next;
 262       GNUNET_CONTAINER_DLL_remove(dest->head, dest->tail, cur);
 263       GNUNET_free(cur);
 264     }
 265 }
 266
 267
 268 /**
 269  * Set traffic metric to manipulate
 270  *
 271  * @param cls closure
 272  * @param client client sending message
 273  * @param message containing information
 274  */
 275 void
 276 GST_manipulation_set_metric(void *cls, struct GNUNET_SERVER_Client *client,
 277     const struct GNUNET_MessageHeader *message)
 278 {
 279   struct TrafficMetricMessage *tm = (struct TrafficMetricMessage *) message;
 280   struct GNUNET_PeerIdentity dummy;
 281   struct GNUNET_ATS_Information *ats;
 282   struct TM_Peer *tmp;
 283   uint32_t type;
 284   uint32_t value;
 285   uint16_t direction;
 286   int c;
 287   int c2;
 288
 289   if (0 == ntohs(tm->ats_count))
 290     GNUNET_SERVER_receive_done(client, GNUNET_SYSERR);
 291
 292   direction = TM_BOTH;
 293   switch (ntohs(tm->direction))
 294     {
 295   case 1:
 296     direction = TM_SEND;
 297     break;
 298   case 2:
 299     direction = TM_RECEIVE;
 300     break;
 301   case 3:
 302     direction = TM_BOTH;
 303     break;
 304   default:
 305     break;
 306     }
 307
 308   memset(&dummy, '\0', sizeof(struct GNUNET_PeerIdentity));
 309   if (0 == memcmp(&tm->peer, &dummy, sizeof(struct GNUNET_PeerIdentity)))
 310     {
 311       GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
 312           "Received traffic metrics for all peers \n");
 313
 314       ats = (struct GNUNET_ATS_Information *) &tm[1];
 315       for (c = 0; c < ntohs(tm->ats_count); c++)
 316         {
 317           type = htonl(ats[c].type);
 318           value = htonl(ats[c].value);
 319           set_metric(&man_handle.general, direction, type, value);
 320         }
 321       return;
 322     }
 323
 324   GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
 325       "Received traffic metrics for peer `%s'\n", GNUNET_i2s(&tm->peer));
 326
 327   if (NULL
 328       == (tmp = GNUNET_CONTAINER_multipeermap_get(man_handle.peers, &tm->peer)))
 329     {
 330       tmp = GNUNET_new (struct TM_Peer);
 331       tmp->peer = (tm->peer);
 332       for (c = 0; c < TM_BOTH; c++)
 333         {
 334           for (c2 = 0; c2 < GNUNET_ATS_QualityPropertiesCount; c2++)
 335             {
 336               tmp->metrics[c][c2] = UINT32_MAX;
 337             }
 338         }
 339       GNUNET_CONTAINER_multipeermap_put(man_handle.peers, &tm->peer, tmp,
 340           GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
 341     }
 342
 343   ats = (struct GNUNET_ATS_Information *) &tm[1];
 344   for (c = 0; c < ntohs(tm->ats_count); c++)
 345     {
 346       type = htonl(ats[c].type);
 347       value = htonl(ats[c].value);
 348       set_metric(tmp, direction, type, value);
 349     }
 350
 351   GNUNET_SERVER_receive_done(client, GNUNET_OK);
 352 }
 353
 354
 355 static void
 356 send_delayed(void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 357 {
 358   struct DelayQueueEntry *dqe = cls;
 359   struct DelayQueueEntry *next;
 360   struct TM_Peer *tmp = dqe->tmp;
 361   struct GNUNET_TIME_Relative delay;
 362
 363   if (NULL != tmp)
 364     {
 365       GNUNET_break (GNUNET_YES ==
 366                     GST_neighbours_test_connected (&dqe->id));
 367       tmp->send_delay_task = NULL;
 368       GNUNET_CONTAINER_DLL_remove (tmp->send_head,
 369                                    tmp->send_tail,
 370                                    dqe);
 371       GST_neighbours_send (&dqe->id,
 372                            dqe->msg,
 373                            dqe->msg_size,
 374                            dqe->timeout,
 375                            dqe->cont,
 376                            dqe->cont_cls);
 377
 378       next = tmp->send_head;
 379       if (NULL != next)
 380         {
 381           /* More delayed messages */
 382           delay = GNUNET_TIME_absolute_get_remaining(next->sent_at);
 383           tmp->send_delay_task = GNUNET_SCHEDULER_add_delayed(delay,
 384               &send_delayed, next);
 385         }
 386     }
 387   else
 388     {
 389       /* Remove from generic queue */
 390       GNUNET_break(GNUNET_YES == GST_neighbours_test_connected (&dqe->id));
 391       generic_send_delay_task = NULL;
 392       GNUNET_CONTAINER_DLL_remove(generic_dqe_head, generic_dqe_tail, dqe);
 393       GST_neighbours_send(&dqe->id, dqe->msg, dqe->msg_size, dqe->timeout,
 394           dqe->cont, dqe->cont_cls);
 395       next = generic_dqe_head;
 396       if (NULL != next)
 397         {
 398           /* More delayed messages */
 399           delay = GNUNET_TIME_absolute_get_remaining(next->sent_at);
 400           generic_send_delay_task = GNUNET_SCHEDULER_add_delayed(delay,
 401               &send_delayed, next);
 402         }
 403     }
 404   GNUNET_free(dqe);
 405 }
 406
 407
 408 /**
 409  * Adapter function between transport's send function and transport plugins.
 410  * Delays message transmission if an artificial delay is configured.
 411  *
 412  * @param target the peer the message to send to
 413  * @param msg the message received
 414  * @param msg_size message size
 415  * @param timeout timeout
 416  * @param cont the continuation to call after sending
 417  * @param cont_cls cls for @a cont
 418  */
 419 void
 420 GST_manipulation_send (const struct GNUNET_PeerIdentity *target,
 421                        const void *msg,
 422                        size_t msg_size,
 423                        struct GNUNET_TIME_Relative timeout,
 424                        GST_NeighbourSendContinuation cont,
 425                        void *cont_cls)
 426 {
 427   struct TM_Peer *tmp;
 428   struct DelayQueueEntry *dqe;
 429   struct GNUNET_TIME_Relative delay;
 430   int do_delay;
 431
 432   do_delay = GNUNET_NO;
 433   if (NULL != (tmp =
 434                GNUNET_CONTAINER_multipeermap_get (man_handle.peers,
 435                                                   target)))
 436   {
 437     GNUNET_break (GNUNET_YES ==
 438                   GST_neighbours_test_connected(target));
 439     /* check for peer-specific delay */
 440     if (UINT32_MAX !=
 441         find_metric (tmp,
 442                      GNUNET_ATS_QUALITY_NET_DELAY,
 443                      TM_SEND))
 444     {
 445       /* We have a delay */
 446       delay.rel_value_us = find_metric(tmp, GNUNET_ATS_QUALITY_NET_DELAY,
 447                                        TM_SEND);
 448       do_delay = GNUNET_YES;
 449     }
 450   }
 451   else if (UINT32_MAX !=
 452            find_metric(&man_handle.general,
 453                        GNUNET_ATS_QUALITY_NET_DELAY,
 454                        TM_SEND))
 455   {
 456     GNUNET_break (GNUNET_YES ==
 457                   GST_neighbours_test_connected (target));
 458     /* We have a delay */
 459     delay.rel_value_us = find_metric (&man_handle.general,
 460                                       GNUNET_ATS_QUALITY_NET_DELAY,
 461                                       TM_SEND);
 462     do_delay = GNUNET_YES;
 463   }
 464   if (GNUNET_NO == do_delay)
 465   {
 466     /* Normal sending */
 467     GST_neighbours_send (target,
 468                          msg,
 469                          msg_size,
 470                          timeout,
 471                          cont, cont_cls);
 472     return;
 473   }
 474   dqe = GNUNET_malloc (sizeof (struct DelayQueueEntry) + msg_size);
 475   dqe->id = *target;
 476   dqe->tmp = tmp;
 477   dqe->sent_at = GNUNET_TIME_relative_to_absolute (delay);
 478   dqe->cont = cont;
 479   dqe->cont_cls = cont_cls;
 480   dqe->msg = &dqe[1];
 481   dqe->msg_size = msg_size;
 482   dqe->timeout = timeout;
 483   memcpy (dqe->msg,
 484           msg,
 485           msg_size);
 486   if (NULL == tmp)
 487   {
 488     GNUNET_CONTAINER_DLL_insert_tail (generic_dqe_head,
 489                                       generic_dqe_tail,
 490                                       dqe);
 491     if (NULL == generic_send_delay_task)
 492       generic_send_delay_task = GNUNET_SCHEDULER_add_delayed (delay,
 493                                                               &send_delayed,
 494                                                               dqe);
 495   }
 496   else
 497   {
 498     GNUNET_CONTAINER_DLL_insert_tail (tmp->send_head,
 499                                       tmp->send_tail,
 500                                       dqe);
 501     if (NULL == tmp->send_delay_task)
 502       tmp->send_delay_task = GNUNET_SCHEDULER_add_delayed (delay,
 503                                                            &send_delayed,
 504                                                            dqe);
 505   }
 506   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 507               "Delaying %u byte message to peer `%s' with peer specific delay for %s\n",
 508               msg_size,
 509               GNUNET_i2s (target),
 510               GNUNET_STRINGS_relative_time_to_string (delay,
 511                                                       GNUNET_YES));
 512 }
 513
 514
 515 /**
 516  * Function that will be called to manipulate ATS information according to
 517  * current manipulation settings
 518  *
 519  * @param peer the peer
 520  * @param address binary address
 521  * @param session the session
 522  * @param ats the ats information
 523  * @param ats_count the number of ats information
 524  */
 525 struct GNUNET_ATS_Information *
 526 GST_manipulation_manipulate_metrics (const struct GNUNET_HELLO_Address *address,
 527                                      struct Session *session,
 528                                      const struct GNUNET_ATS_Information *ats,
 529                                      uint32_t ats_count)
 530 {
 531   const struct GNUNET_PeerIdentity *peer = &address->peer;
 532   struct GNUNET_ATS_Information *ats_new;
 533   struct TM_Peer *tmp;
 534   uint32_t m_tmp;
 535   uint32_t g_tmp;
 536   uint32_t d;
 537
 538   if (0 == ats_count)
 539     return NULL;
 540   ats_new = GNUNET_malloc (sizeof (struct GNUNET_ATS_Information) * ats_count);
 541   tmp = GNUNET_CONTAINER_multipeermap_get (man_handle.peers, peer);
 542   for (d = 0; d < ats_count; d++)
 543   {
 544     ats_new[d] = ats[d];
 545     m_tmp = UINT32_MAX;
 546     if (NULL != tmp)
 547       m_tmp = find_metric (tmp, ntohl(ats[d].type),
 548                            TM_RECEIVE);
 549     g_tmp = find_metric (&man_handle.general,
 550                          ntohl(ats[d].type),
 551                          TM_RECEIVE);
 552
 553     if (UINT32_MAX != g_tmp)
 554       ats_new[d].value = htonl(g_tmp);
 555     if (UINT32_MAX != m_tmp)
 556       ats_new[d].value = htonl(m_tmp);
 557   }
 558   return ats_new;
 559 }
 560
 561
 562 /**
 563  * Adapter function between transport plugins and transport receive function
 564  * manipulation delays for next send.
 565  *
 566  * @param cls the closure for transport
 567  * @param address the address and the peer the message was received from
 568  * @param message the message received
 569  * @param session the session the message was received on
 570  * @return manipulated delay for next receive
 571  */
 572 struct GNUNET_TIME_Relative
 573 GST_manipulation_recv (void *cls,
 574                        const struct GNUNET_HELLO_Address *address,
 575                        struct Session *session,
 576                        const struct GNUNET_MessageHeader *message)
 577 {
 578   struct TM_Peer *tmp;
 579   uint32_t p_recv_delay;
 580   uint32_t g_recv_delay;
 581   struct GNUNET_TIME_Relative quota_delay;
 582   struct GNUNET_TIME_Relative m_delay;
 583
 584   g_recv_delay = find_metric(&man_handle.general, GNUNET_ATS_QUALITY_NET_DELAY,
 585       TM_RECEIVE);
 586   if ((g_recv_delay >= GNUNET_TIME_UNIT_ZERO.rel_value_us)
 587       && (UINT32_MAX != g_recv_delay))
 588     m_delay.rel_value_us = g_recv_delay; /* Global delay */
 589   else
 590     m_delay = GNUNET_TIME_UNIT_ZERO;
 591
 592   if (NULL != (tmp = GNUNET_CONTAINER_multipeermap_get(man_handle.peers, &address->peer)))
 593     {
 594       /* Manipulate receive delay */
 595       p_recv_delay = find_metric(tmp, GNUNET_ATS_QUALITY_NET_DELAY, TM_RECEIVE);
 596       if (UINT32_MAX != p_recv_delay)
 597         m_delay.rel_value_us = p_recv_delay; /* Peer specific delay */
 598     }
 599
 600   quota_delay = GST_receive_callback(cls, address, session, message);
 601
 602   if (quota_delay.rel_value_us > m_delay.rel_value_us)
 603     m_delay = quota_delay;
 604
 605   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 606               "Delaying next receive for peer `%s' for %s\n",
 607               GNUNET_i2s (&address->peer),
 608               GNUNET_STRINGS_relative_time_to_string (m_delay,
 609                                                       GNUNET_YES));
 610   return m_delay;
 611 }
 612
 613
 614 /**
 615  * Initialize traffic manipulation
 616  *
 617  * @param GST_cfg configuration handle
 618  */
 619 void
 620 GST_manipulation_init(const struct GNUNET_CONFIGURATION_Handle *GST_cfg)
 621 {
 622   unsigned long long tmp;
 623   struct GNUNET_TIME_Relative delay;
 624
 625   if ( (GNUNET_OK ==
 626         GNUNET_CONFIGURATION_get_value_number(GST_cfg,
 627                                               "transport",
 628                                               "MANIPULATE_DISTANCE_IN",
 629                                               &tmp)) &&
 630        (tmp > 0) )
 631   {
 632     GNUNET_log(GNUNET_ERROR_TYPE_INFO,
 633                "Setting inbound distance_in to %llu\n",
 634                (unsigned long long) tmp);
 635     set_metric (&man_handle.general,
 636                 TM_RECEIVE,
 637                 GNUNET_ATS_QUALITY_NET_DISTANCE,
 638                 tmp);
 639   }
 640
 641   if ((GNUNET_OK
 642       == GNUNET_CONFIGURATION_get_value_number(GST_cfg, "transport",
 643           "MANIPULATE_DISTANCE_OUT", &tmp)) && (tmp > 0))
 644     {
 645       GNUNET_log(GNUNET_ERROR_TYPE_INFO,
 646           "Setting outbound distance_in to %llu\n", (unsigned long long) tmp);
 647       set_metric(&man_handle.general, TM_SEND, GNUNET_ATS_QUALITY_NET_DISTANCE,
 648           tmp);
 649     }
 650
 651   if ((GNUNET_OK
 652       == GNUNET_CONFIGURATION_get_value_time(GST_cfg, "transport",
 653           "MANIPULATE_DELAY_IN", &delay)) && (delay.rel_value_us > 0))
 654     {
 655       GNUNET_log(GNUNET_ERROR_TYPE_INFO,
 656           "Delaying inbound traffic for %s\n", GNUNET_STRINGS_relative_time_to_string (delay, GNUNET_YES));
 657       set_metric(&man_handle.general, TM_RECEIVE, GNUNET_ATS_QUALITY_NET_DELAY,
 658           delay.rel_value_us);
 659     }
 660   if ((GNUNET_OK
 661       == GNUNET_CONFIGURATION_get_value_time(GST_cfg, "transport",
 662           "MANIPULATE_DELAY_OUT", &delay)) && (delay.rel_value_us > 0))
 663     {
 664       GNUNET_log(GNUNET_ERROR_TYPE_INFO,
 665           "Delaying outbound traffic for %s\n", GNUNET_STRINGS_relative_time_to_string (delay, GNUNET_YES));
 666       set_metric(&man_handle.general, TM_SEND, GNUNET_ATS_QUALITY_NET_DELAY,
 667           delay.rel_value_us);
 668     }
 669   man_handle.peers = GNUNET_CONTAINER_multipeermap_create(10, GNUNET_NO);
 670 }
 671
 672
 673 static int
 674 free_tmps (void *cls,
 675            const struct GNUNET_PeerIdentity *key,
 676            void *value)
 677 {
 678   struct TM_Peer *tmp = value;
 679   struct DelayQueueEntry *dqe;
 680
 681   if (NULL == tmp)
 682     return GNUNET_OK;
 683   GNUNET_break (GNUNET_YES ==
 684                 GNUNET_CONTAINER_multipeermap_remove (man_handle.peers,
 685                                                       key,
 686                                                       value));
 687   free_metric (tmp);
 688   while (NULL != (dqe = tmp->send_head))
 689   {
 690     GNUNET_CONTAINER_DLL_remove (tmp->send_head,
 691                                  tmp->send_tail,
 692                                  dqe);
 693     if (NULL != dqe->cont)
 694       dqe->cont (dqe->cont_cls,
 695                  GNUNET_SYSERR,
 696                  dqe->msg_size,
 697                  0);
 698     GNUNET_free (dqe);
 699   }
 700   if (NULL != tmp->send_delay_task)
 701   {
 702     GNUNET_SCHEDULER_cancel(tmp->send_delay_task);
 703     tmp->send_delay_task = NULL;
 704   }
 705   GNUNET_free(tmp);
 706   return GNUNET_OK;
 707 }
 708
 709
 710 /**
 711  * Notify manipulation about disconnect so it can discard queued messages
 712  *
 713  * @param peer the disconnecting peer
 714  */
 715 void
 716 GST_manipulation_peer_disconnect (const struct GNUNET_PeerIdentity *peer)
 717 {
 718   struct TM_Peer *tmp;
 719   struct DelayQueueEntry *dqe;
 720   struct DelayQueueEntry *next;
 721
 722   if (NULL != (tmp = GNUNET_CONTAINER_multipeermap_get(man_handle.peers, peer)))
 723   {
 724     while (NULL != (dqe = tmp->send_head))
 725     {
 726       GNUNET_CONTAINER_DLL_remove (tmp->send_head,
 727                                    tmp->send_tail,
 728                                    dqe);
 729       if (NULL != dqe->cont)
 730         dqe->cont (dqe->cont_cls,
 731                    GNUNET_SYSERR,
 732                    dqe->msg_size,
 733                    0);
 734       GNUNET_free(dqe);
 735     }
 736   }
 737   else if (UINT32_MAX != find_metric (&man_handle.general,
 738                                       GNUNET_ATS_QUALITY_NET_DELAY,
 739                                       TM_SEND))
 740   {
 741     next = generic_dqe_head;
 742     while (NULL != (dqe = next))
 743     {
 744       next = dqe->next;
 745       if (0 == memcmp(peer, &dqe->id, sizeof(dqe->id)))
 746       {
 747         GNUNET_CONTAINER_DLL_remove (generic_dqe_head,
 748                                      generic_dqe_tail,
 749                                      dqe);
 750         if (NULL != dqe->cont)
 751           dqe->cont (dqe->cont_cls,
 752                      GNUNET_SYSERR,
 753                      dqe->msg_size,
 754                      0);
 755         GNUNET_free(dqe);
 756       }
 757     }
 758     if (NULL != generic_send_delay_task)
 759     {
 760       GNUNET_SCHEDULER_cancel (generic_send_delay_task);
 761       generic_send_delay_task = NULL;
 762       if (NULL != generic_dqe_head)
 763         generic_send_delay_task
 764           = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_absolute_get_remaining(generic_dqe_head->sent_at),
 765                                           &send_delayed,
 766                                           generic_dqe_head);
 767     }
 768   }
 769 }
 770
 771
 772 /**
 773  * Stop traffic manipulation
 774  */
 775 void
 776 GST_manipulation_stop()
 777 {
 778   struct DelayQueueEntry *cur;
 779
 780   GNUNET_CONTAINER_multipeermap_iterate (man_handle.peers,
 781                                          &free_tmps,
 782                                          NULL);
 783   GNUNET_CONTAINER_multipeermap_destroy (man_handle.peers);
 784
 785   while (NULL != (cur = generic_dqe_head))
 786   {
 787     GNUNET_CONTAINER_DLL_remove (generic_dqe_head,
 788                                  generic_dqe_tail,
 789                                  cur);
 790     if (NULL != cur->cont)
 791       cur->cont (cur->cont_cls,
 792                  GNUNET_SYSERR,
 793                  cur->msg_size,
 794                  0);
 795     GNUNET_free (cur);
 796   }
 797   if (NULL != generic_send_delay_task)
 798   {
 799     GNUNET_SCHEDULER_cancel (generic_send_delay_task);
 800     generic_send_delay_task = NULL;
 801   }
 802   free_metric (&man_handle.general);
 803   man_handle.peers = NULL;
 804 }
 805
 806 /* end of file gnunet-service-transport_manipulation.c */