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 static void
 199 set_metric(struct TM_Peer *dest, int direction, uint32_t type, uint32_t value)
 200 {
 201   struct PropManipulationEntry *cur;
 202   for (cur = dest->head; NULL != cur; cur = cur->next)
 203     {
 204       if (cur->type == type)
 205         break;
 206     }
 207   if (NULL == cur)
 208     {
 209       cur = GNUNET_new (struct PropManipulationEntry);
 210       GNUNET_CONTAINER_DLL_insert(dest->head, dest->tail, cur);
 211       cur->type = type;
 212       cur->metrics[TM_SEND] = UINT32_MAX;
 213       cur->metrics[TM_RECEIVE] = UINT32_MAX;
 214     }
 215
 216   switch (direction)
 217     {
 218   case TM_BOTH:
 219     cur->metrics[TM_SEND] = value;
 220     cur->metrics[TM_RECEIVE] = value;
 221     break;
 222   case TM_SEND:
 223     cur->metrics[TM_SEND] = value;
 224     break;
 225   case TM_RECEIVE:
 226     cur->metrics[TM_RECEIVE] = value;
 227     break;
 228   default:
 229     break;
 230     }
 231
 232 }
 233
 234 static uint32_t
 235 find_metric(struct TM_Peer *dest, uint32_t type, int direction)
 236 {
 237   struct PropManipulationEntry *cur;
 238
 239   for (cur = dest->head; NULL != cur; cur = cur->next)
 240     {
 241       if (cur->type == type)
 242         return cur->metrics[direction];
 243
 244     }
 245   return UINT32_MAX;
 246 }
 247
 248 /**
 249  * Clean up metrics for a peer
 250  */
 251
 252 static void
 253 free_metric(struct TM_Peer *dest)
 254 {
 255   struct PropManipulationEntry *cur;
 256   struct PropManipulationEntry *next;
 257
 258   for (cur = dest->head; NULL != cur; cur = next)
 259     {
 260       next = cur->next;
 261       GNUNET_CONTAINER_DLL_remove(dest->head, dest->tail, cur);
 262       GNUNET_free(cur);
 263     }
 264 }
 265
 266 /**
 267  * Set traffic metric to manipulate
 268  *
 269  * @param cls closure
 270  * @param client client sending message
 271  * @param message containing information
 272  */
 273 void
 274 GST_manipulation_set_metric(void *cls, struct GNUNET_SERVER_Client *client,
 275     const struct GNUNET_MessageHeader *message)
 276 {
 277   struct TrafficMetricMessage *tm = (struct TrafficMetricMessage *) message;
 278   struct GNUNET_PeerIdentity dummy;
 279   struct GNUNET_ATS_Information *ats;
 280   struct TM_Peer *tmp;
 281   uint32_t type;
 282   uint32_t value;
 283   uint16_t direction;
 284   int c;
 285   int c2;
 286
 287   if (0 == ntohs(tm->ats_count))
 288     GNUNET_SERVER_receive_done(client, GNUNET_SYSERR);
 289
 290   direction = TM_BOTH;
 291   switch (ntohs(tm->direction))
 292     {
 293   case 1:
 294     direction = TM_SEND;
 295     break;
 296   case 2:
 297     direction = TM_RECEIVE;
 298     break;
 299   case 3:
 300     direction = TM_BOTH;
 301     break;
 302   default:
 303     break;
 304     }
 305
 306   memset(&dummy, '\0', sizeof(struct GNUNET_PeerIdentity));
 307   if (0 == memcmp(&tm->peer, &dummy, sizeof(struct GNUNET_PeerIdentity)))
 308     {
 309       GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
 310           "Received traffic metrics for all peers \n");
 311
 312       ats = (struct GNUNET_ATS_Information *) &tm[1];
 313       for (c = 0; c < ntohs(tm->ats_count); c++)
 314         {
 315           type = htonl(ats[c].type);
 316           value = htonl(ats[c].value);
 317           set_metric(&man_handle.general, direction, type, value);
 318         }
 319       return;
 320     }
 321
 322   GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
 323       "Received traffic metrics for peer `%s'\n", GNUNET_i2s(&tm->peer));
 324
 325   if (NULL
 326       == (tmp = GNUNET_CONTAINER_multipeermap_get(man_handle.peers, &tm->peer)))
 327     {
 328       tmp = GNUNET_new (struct TM_Peer);
 329       tmp->peer = (tm->peer);
 330       for (c = 0; c < TM_BOTH; c++)
 331         {
 332           for (c2 = 0; c2 < GNUNET_ATS_QualityPropertiesCount; c2++)
 333             {
 334               tmp->metrics[c][c2] = UINT32_MAX;
 335             }
 336         }
 337       GNUNET_CONTAINER_multipeermap_put(man_handle.peers, &tm->peer, tmp,
 338           GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
 339     }
 340
 341   ats = (struct GNUNET_ATS_Information *) &tm[1];
 342   for (c = 0; c < ntohs(tm->ats_count); c++)
 343     {
 344       type = htonl(ats[c].type);
 345       value = htonl(ats[c].value);
 346       set_metric(tmp, direction, type, value);
 347     }
 348
 349   GNUNET_SERVER_receive_done(client, GNUNET_OK);
 350 }
 351
 352 static void
 353 send_delayed(void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 354 {
 355   struct DelayQueueEntry *dqe = cls;
 356   struct DelayQueueEntry *next;
 357   struct TM_Peer *tmp = dqe->tmp;
 358   struct GNUNET_TIME_Relative delay;
 359
 360   if (NULL != tmp)
 361     {
 362       GNUNET_break(GNUNET_YES == GST_neighbours_test_connected (&dqe->id));
 363       tmp->send_delay_task = NULL;
 364       GNUNET_CONTAINER_DLL_remove(tmp->send_head, tmp->send_tail, dqe);
 365       GST_neighbours_send(&dqe->id, dqe->msg, dqe->msg_size, dqe->timeout,
 366           dqe->cont, dqe->cont_cls);
 367
 368       next = tmp->send_head;
 369       if (NULL != next)
 370         {
 371           /* More delayed messages */
 372           delay = GNUNET_TIME_absolute_get_remaining(next->sent_at);
 373           tmp->send_delay_task = GNUNET_SCHEDULER_add_delayed(delay,
 374               &send_delayed, next);
 375         }
 376     }
 377   else
 378     {
 379       /* Remove from generic queue */
 380       GNUNET_break(GNUNET_YES == GST_neighbours_test_connected (&dqe->id));
 381       generic_send_delay_task = NULL;
 382       GNUNET_CONTAINER_DLL_remove(generic_dqe_head, generic_dqe_tail, dqe);
 383       GST_neighbours_send(&dqe->id, dqe->msg, dqe->msg_size, dqe->timeout,
 384           dqe->cont, dqe->cont_cls);
 385       next = generic_dqe_head;
 386       if (NULL != next)
 387         {
 388           /* More delayed messages */
 389           delay = GNUNET_TIME_absolute_get_remaining(next->sent_at);
 390           generic_send_delay_task = GNUNET_SCHEDULER_add_delayed(delay,
 391               &send_delayed, next);
 392         }
 393     }
 394   GNUNET_free(dqe);
 395 }
 396
 397 /**
 398  * Adapter function between transport's send function and transport plugins
 399  *
 400  * @param target the peer the message to send to
 401  * @param msg the message received
 402  * @param msg_size message size
 403  * @param timeout timeout
 404  * @param cont the continuation to call after sending
 405  * @param cont_cls cls for continuation
 406  */
 407 void
 408 GST_manipulation_send(const struct GNUNET_PeerIdentity *target, const void *msg,
 409     size_t msg_size, struct GNUNET_TIME_Relative timeout,
 410     GST_NeighbourSendContinuation cont, void *cont_cls)
 411 {
 412   struct TM_Peer *tmp;
 413   struct DelayQueueEntry *dqe;
 414   struct GNUNET_TIME_Relative delay;
 415
 416   if (NULL
 417       != (tmp = GNUNET_CONTAINER_multipeermap_get(man_handle.peers, target)))
 418     {
 419       GNUNET_break(GNUNET_YES == GST_neighbours_test_connected(target));
 420       /* Manipulate here */
 421       /* Delay */
 422       if (UINT32_MAX != find_metric(tmp, GNUNET_ATS_QUALITY_NET_DELAY, TM_SEND))
 423         {
 424           /* We have a delay */
 425           delay.rel_value_us = find_metric(tmp, GNUNET_ATS_QUALITY_NET_DELAY,
 426               TM_SEND);
 427           dqe = GNUNET_malloc (sizeof (struct DelayQueueEntry) + msg_size);
 428           dqe->id = *target;
 429           dqe->tmp = tmp;
 430           dqe->sent_at = GNUNET_TIME_absolute_add(GNUNET_TIME_absolute_get(),
 431               delay);
 432           dqe->cont = cont;
 433           dqe->cont_cls = cont_cls;
 434           dqe->msg = &dqe[1];
 435           dqe->msg_size = msg_size;
 436           dqe->timeout = timeout;
 437           memcpy(dqe->msg, msg, msg_size);
 438           GNUNET_CONTAINER_DLL_insert_tail(tmp->send_head, tmp->send_tail, dqe);
 439           if (NULL == tmp->send_delay_task)
 440             tmp->send_delay_task = GNUNET_SCHEDULER_add_delayed(delay,
 441                 &send_delayed, dqe);
 442           GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
 443               "Delaying %u byte message to peer `%s' with generic delay for %ms\n", msg_size, GNUNET_i2s (target), GNUNET_STRINGS_relative_time_to_string (delay, GNUNET_YES));
 444           return;
 445         }
 446     }
 447   else if (UINT32_MAX
 448       != find_metric(&man_handle.general, GNUNET_ATS_QUALITY_NET_DELAY,
 449           TM_SEND))
 450     {
 451       GNUNET_break(GNUNET_YES == GST_neighbours_test_connected(target));
 452       /* We have a delay */
 453       delay.rel_value_us = find_metric(&man_handle.general,
 454           GNUNET_ATS_QUALITY_NET_DELAY, TM_SEND);
 455       dqe = GNUNET_malloc (sizeof (struct DelayQueueEntry) + msg_size);
 456       dqe->id = *target;
 457       dqe->tmp = NULL;
 458       dqe->sent_at = GNUNET_TIME_absolute_add(GNUNET_TIME_absolute_get(),
 459           delay);
 460       dqe->cont = cont;
 461       dqe->cont_cls = cont_cls;
 462       dqe->msg = &dqe[1];
 463       dqe->msg_size = msg_size;
 464       dqe->timeout = timeout;
 465       memcpy(dqe->msg, msg, msg_size);
 466       GNUNET_CONTAINER_DLL_insert_tail(generic_dqe_head, generic_dqe_tail, dqe);
 467       if (NULL == generic_send_delay_task)
 468         {
 469           generic_send_delay_task = GNUNET_SCHEDULER_add_delayed(delay,
 470               &send_delayed, dqe);
 471         }
 472       GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
 473           "Delaying %u byte message to peer `%s' with peer specific delay for %s\n", msg_size, GNUNET_i2s (target), GNUNET_STRINGS_relative_time_to_string (delay, GNUNET_YES));
 474       return;
 475     }
 476
 477   /* Normal sending */
 478   GST_neighbours_send(target, msg, msg_size, timeout, cont, cont_cls);
 479 }
 480
 481 /**
 482  * Function that will be called to manipulate ATS information according to
 483  * current manipulation settings
 484  *
 485  * @param peer the peer
 486  * @param address binary address
 487  * @param session the session
 488  * @param ats the ats information
 489  * @param ats_count the number of ats information
 490  */
 491 struct GNUNET_ATS_Information *
 492 GST_manipulation_manipulate_metrics(const struct GNUNET_HELLO_Address *address,
 493                                     struct Session *session,
 494                                     const struct GNUNET_ATS_Information *ats,
 495                                     uint32_t ats_count)
 496 {
 497   const struct GNUNET_PeerIdentity *peer = &address->peer;
 498   struct GNUNET_ATS_Information *ats_new;
 499   struct TM_Peer *tmp;
 500   uint32_t m_tmp;
 501   uint32_t g_tmp;
 502   uint32_t d;
 503
 504   if (0 == ats_count)
 505     return NULL;
 506   ats_new = GNUNET_malloc (sizeof (struct GNUNET_ATS_Information) * ats_count);
 507   tmp = GNUNET_CONTAINER_multipeermap_get (man_handle.peers, peer);
 508   for (d = 0; d < ats_count; d++)
 509   {
 510     ats_new[d] = ats[d];
 511     m_tmp = UINT32_MAX;
 512     if (NULL != tmp)
 513       m_tmp = find_metric(tmp, ntohl(ats[d].type), TM_RECEIVE);
 514     g_tmp = find_metric(&man_handle.general, ntohl(ats[d].type), TM_RECEIVE);
 515
 516     if (UINT32_MAX != g_tmp)
 517       ats_new[d].value = htonl(g_tmp);
 518     if (UINT32_MAX != m_tmp)
 519       ats_new[d].value = htonl(m_tmp);
 520   }
 521   return ats_new;
 522 }
 523
 524
 525 /**
 526  * Adapter function between transport plugins and transport receive function
 527  * manipulation delays for next send.
 528  *
 529  * @param cls the closure for transport
 530  * @param address the address and the peer the message was received from
 531  * @param message the message received
 532  * @param session the session the message was received on
 533  * @return manipulated delay for next receive
 534  */
 535 struct GNUNET_TIME_Relative
 536 GST_manipulation_recv (void *cls,
 537                        const struct GNUNET_HELLO_Address *address,
 538                        struct Session *session,
 539                        const struct GNUNET_MessageHeader *message)
 540 {
 541   struct TM_Peer *tmp;
 542   uint32_t p_recv_delay;
 543   uint32_t g_recv_delay;
 544   struct GNUNET_TIME_Relative quota_delay;
 545   struct GNUNET_TIME_Relative m_delay;
 546
 547   g_recv_delay = find_metric(&man_handle.general, GNUNET_ATS_QUALITY_NET_DELAY,
 548       TM_RECEIVE);
 549   if ((g_recv_delay >= GNUNET_TIME_UNIT_ZERO.rel_value_us)
 550       && (UINT32_MAX != g_recv_delay))
 551     m_delay.rel_value_us = g_recv_delay; /* Global delay */
 552   else
 553     m_delay = GNUNET_TIME_UNIT_ZERO;
 554
 555   if (NULL != (tmp = GNUNET_CONTAINER_multipeermap_get(man_handle.peers, &address->peer)))
 556     {
 557       /* Manipulate receive delay */
 558       p_recv_delay = find_metric(tmp, GNUNET_ATS_QUALITY_NET_DELAY, TM_RECEIVE);
 559       if (UINT32_MAX != p_recv_delay)
 560         m_delay.rel_value_us = p_recv_delay; /* Peer specific delay */
 561     }
 562
 563   quota_delay = GST_receive_callback(cls, address, session, message);
 564
 565   if (quota_delay.rel_value_us > m_delay.rel_value_us)
 566     m_delay = quota_delay;
 567
 568   GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
 569       "Delaying next receive for peer `%s' for %s\n",
 570       GNUNET_i2s (&address->peer),
 571       GNUNET_STRINGS_relative_time_to_string (m_delay, GNUNET_YES));
 572   return m_delay;
 573
 574 }
 575
 576 /**
 577  * Initialize traffic manipulation
 578  *
 579  * @param GST_cfg configuration handle
 580  */
 581 void
 582 GST_manipulation_init(const struct GNUNET_CONFIGURATION_Handle *GST_cfg)
 583 {
 584   unsigned long long tmp;
 585   struct GNUNET_TIME_Relative delay;
 586
 587   if ((GNUNET_OK
 588       == GNUNET_CONFIGURATION_get_value_number(GST_cfg, "transport",
 589           "MANIPULATE_DISTANCE_IN", &tmp)) && (tmp > 0))
 590     {
 591       GNUNET_log(GNUNET_ERROR_TYPE_INFO,
 592           "Setting inbound distance_in to %llu\n", (unsigned long long) tmp);
 593       set_metric(&man_handle.general, TM_RECEIVE,
 594           GNUNET_ATS_QUALITY_NET_DISTANCE, tmp);
 595     }
 596
 597   if ((GNUNET_OK
 598       == GNUNET_CONFIGURATION_get_value_number(GST_cfg, "transport",
 599           "MANIPULATE_DISTANCE_OUT", &tmp)) && (tmp > 0))
 600     {
 601       GNUNET_log(GNUNET_ERROR_TYPE_INFO,
 602           "Setting outbound distance_in to %llu\n", (unsigned long long) tmp);
 603       set_metric(&man_handle.general, TM_SEND, GNUNET_ATS_QUALITY_NET_DISTANCE,
 604           tmp);
 605     }
 606
 607   if ((GNUNET_OK
 608       == GNUNET_CONFIGURATION_get_value_time(GST_cfg, "transport",
 609           "MANIPULATE_DELAY_IN", &delay)) && (delay.rel_value_us > 0))
 610     {
 611       GNUNET_log(GNUNET_ERROR_TYPE_INFO,
 612           "Delaying inbound traffic for %s\n", GNUNET_STRINGS_relative_time_to_string (delay, GNUNET_YES));
 613       set_metric(&man_handle.general, TM_RECEIVE, GNUNET_ATS_QUALITY_NET_DELAY,
 614           delay.rel_value_us);
 615     }
 616   if ((GNUNET_OK
 617       == GNUNET_CONFIGURATION_get_value_time(GST_cfg, "transport",
 618           "MANIPULATE_DELAY_OUT", &delay)) && (delay.rel_value_us > 0))
 619     {
 620       GNUNET_log(GNUNET_ERROR_TYPE_INFO,
 621           "Delaying outbound traffic for %s\n", GNUNET_STRINGS_relative_time_to_string (delay, GNUNET_YES));
 622       set_metric(&man_handle.general, TM_SEND, GNUNET_ATS_QUALITY_NET_DELAY,
 623           delay.rel_value_us);
 624     }
 625   man_handle.peers = GNUNET_CONTAINER_multipeermap_create(10, GNUNET_NO);
 626 }
 627
 628 static int
 629 free_tmps(void *cls, const struct GNUNET_PeerIdentity *key, void *value)
 630 {
 631   struct DelayQueueEntry *dqe;
 632   struct DelayQueueEntry *next;
 633
 634   if (NULL != value)
 635     {
 636       struct TM_Peer *tmp = (struct TM_Peer *) value;
 637
 638       if (GNUNET_YES
 639           != GNUNET_CONTAINER_multipeermap_remove(man_handle.peers, key, value))
 640         GNUNET_break(0);
 641       free_metric(tmp);
 642       next = tmp->send_head;
 643       while (NULL != (dqe = next))
 644         {
 645           next = dqe->next;
 646           GNUNET_CONTAINER_DLL_remove(tmp->send_head, tmp->send_tail, dqe);
 647           if (NULL != dqe->cont)
 648             dqe->cont(dqe->cont_cls, GNUNET_SYSERR, dqe->msg_size, 0);
 649           GNUNET_free(dqe);
 650         }
 651       if (NULL != tmp->send_delay_task)
 652         {
 653           GNUNET_SCHEDULER_cancel(tmp->send_delay_task);
 654           tmp->send_delay_task = NULL;
 655         }
 656       GNUNET_free(tmp);
 657     }
 658   return GNUNET_OK;
 659 }
 660
 661 /**
 662  * Notify manipulation about disconnect so it can discard queued messages
 663  *
 664  * @param peer the disconnecting peer
 665  */
 666 void
 667 GST_manipulation_peer_disconnect(const struct GNUNET_PeerIdentity *peer)
 668 {
 669   struct TM_Peer *tmp;
 670   struct DelayQueueEntry *dqe;
 671   struct DelayQueueEntry *next;
 672
 673   if (NULL != (tmp = GNUNET_CONTAINER_multipeermap_get(man_handle.peers, peer)))
 674     {
 675       next = tmp->send_head;
 676       while (NULL != (dqe = next))
 677         {
 678           next = dqe->next;
 679           GNUNET_CONTAINER_DLL_remove(tmp->send_head, tmp->send_tail, dqe);
 680           if (NULL != dqe->cont)
 681             dqe->cont(dqe->cont_cls, GNUNET_SYSERR, dqe->msg_size, 0);
 682           GNUNET_free(dqe);
 683         }
 684     }
 685   else if (UINT32_MAX
 686       != find_metric(&man_handle.general, GNUNET_ATS_QUALITY_NET_DELAY,
 687           TM_SEND))
 688     {
 689       next = generic_dqe_head;
 690       while (NULL != (dqe = next))
 691         {
 692           next = dqe->next;
 693           if (0 == memcmp(peer, &dqe->id, sizeof(dqe->id)))
 694             {
 695               GNUNET_CONTAINER_DLL_remove(generic_dqe_head, generic_dqe_tail,
 696                   dqe);
 697               if (NULL != dqe->cont)
 698                 dqe->cont(dqe->cont_cls, GNUNET_SYSERR, dqe->msg_size, 0);
 699               GNUNET_free(dqe);
 700             }
 701         }
 702       if (NULL != generic_send_delay_task)
 703         {
 704           GNUNET_SCHEDULER_cancel(generic_send_delay_task);
 705           generic_send_delay_task = NULL;
 706           if (NULL != generic_dqe_head)
 707             generic_send_delay_task = GNUNET_SCHEDULER_add_delayed(
 708                 GNUNET_TIME_absolute_get_remaining(generic_dqe_head->sent_at),
 709                 &send_delayed, generic_dqe_head);
 710         }
 711     }
 712 }
 713
 714 /**
 715  * Stop traffic manipulation
 716  */
 717 void
 718 GST_manipulation_stop()
 719 {
 720   struct DelayQueueEntry *cur;
 721   struct DelayQueueEntry *next;
 722   GNUNET_CONTAINER_multipeermap_iterate(man_handle.peers, &free_tmps, NULL);
 723   GNUNET_CONTAINER_multipeermap_destroy(man_handle.peers);
 724
 725   next = generic_dqe_head;
 726   while (NULL != (cur = next))
 727     {
 728       next = cur->next;
 729       GNUNET_CONTAINER_DLL_remove(generic_dqe_head, generic_dqe_tail, cur);
 730       if (NULL != cur->cont)
 731         cur->cont(cur->cont_cls, GNUNET_SYSERR, cur->msg_size, 0);
 732       GNUNET_free(cur);
 733     }
 734   if (NULL != generic_send_delay_task)
 735     {
 736       GNUNET_SCHEDULER_cancel(generic_send_delay_task);
 737       generic_send_delay_task = NULL;
 738     }
 739
 740   free_metric(&man_handle.general);
 741   man_handle.peers = NULL;
 742 }
 743
 744 /* end of file gnunet-service-transport_manipulation.c */