src/transport/gnunet-service-transport_manipulation.c

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