src/transport/plugin_transport_udp.c

   1 /*
   2  This file is part of GNUnet
   3  Copyright (C) 2010-2017 GNUnet e.V.
   4
   5  GNUnet is free software: you can redistribute it and/or modify it
   6  under the terms of the GNU Affero General Public License as published
   7  by the Free Software Foundation, either version 3 of the License,
   8  or (at your 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  Affero General Public License for more details.
  14
  15  You should have received a copy of the GNU Affero General Public License
  16  along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17  */
  18
  19 /**
  20  * @file transport/plugin_transport_udp.c
  21  * @brief Implementation of the UDP transport protocol
  22  * @author Christian Grothoff
  23  * @author Nathan Evans
  24  * @author Matthias Wachs
  25  */
  26 #include "platform.h"
  27 #include "plugin_transport_udp.h"
  28 #include "gnunet_hello_lib.h"
  29 #include "gnunet_util_lib.h"
  30 #include "gnunet_fragmentation_lib.h"
  31 #include "gnunet_nat_service.h"
  32 #include "gnunet_protocols.h"
  33 #include "gnunet_resolver_service.h"
  34 #include "gnunet_signatures.h"
  35 #include "gnunet_constants.h"
  36 #include "gnunet_statistics_service.h"
  37 #include "gnunet_transport_service.h"
  38 #include "gnunet_transport_plugin.h"
  39 #include "transport.h"
  40
  41 #define LOG(kind,...) GNUNET_log_from (kind, "transport-udp", __VA_ARGS__)
  42
  43 /**
  44  * After how much inactivity should a UDP session time out?
  45  */
  46 #define UDP_SESSION_TIME_OUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 60)
  47
  48 /**
  49  * Number of messages we can defragment in parallel.  We only really
  50  * defragment 1 message at a time, but if messages get re-ordered, we
  51  * may want to keep knowledge about the previous message to avoid
  52  * discarding the current message in favor of a single fragment of a
  53  * previous message.  3 should be good since we don't expect massive
  54  * message reorderings with UDP.
  55  */
  56 #define UDP_MAX_MESSAGES_IN_DEFRAG 3
  57
  58 /**
  59  * We keep a defragmentation queue per sender address.  How many
  60  * sender addresses do we support at the same time? Memory consumption
  61  * is roughly a factor of 32k * #UDP_MAX_MESSAGES_IN_DEFRAG times this
  62  * value. (So 128 corresponds to 12 MB and should suffice for
  63  * connecting to roughly 128 peers via UDP).
  64  */
  65 #define UDP_MAX_SENDER_ADDRESSES_WITH_DEFRAG 128
  66
  67
  68 /**
  69  * UDP Message-Packet header (after defragmentation).
  70  */
  71 struct UDPMessage
  72 {
  73   /**
  74    * Message header.
  75    */
  76   struct GNUNET_MessageHeader header;
  77
  78   /**
  79    * Always zero for now.
  80    */
  81   uint32_t reserved;
  82
  83   /**
  84    * What is the identity of the sender
  85    */
  86   struct GNUNET_PeerIdentity sender;
  87
  88 };
  89
  90
  91 /**
  92  * Closure for #append_port().
  93  */
  94 struct PrettyPrinterContext
  95 {
  96   /**
  97    * DLL
  98    */
  99   struct PrettyPrinterContext *next;
 100
 101   /**
 102    * DLL
 103    */
 104   struct PrettyPrinterContext *prev;
 105
 106   /**
 107    * Our plugin.
 108    */
 109   struct Plugin *plugin;
 110
 111   /**
 112    * Resolver handle
 113    */
 114   struct GNUNET_RESOLVER_RequestHandle *resolver_handle;
 115
 116   /**
 117    * Function to call with the result.
 118    */
 119   GNUNET_TRANSPORT_AddressStringCallback asc;
 120
 121   /**
 122    * Clsoure for @e asc.
 123    */
 124   void *asc_cls;
 125
 126   /**
 127    * Timeout task
 128    */
 129   struct GNUNET_SCHEDULER_Task *timeout_task;
 130
 131   /**
 132    * Is this an IPv6 address?
 133    */
 134   int ipv6;
 135
 136   /**
 137    * Options
 138    */
 139   uint32_t options;
 140
 141   /**
 142    * Port to add after the IP address.
 143    */
 144   uint16_t port;
 145
 146 };
 147
 148
 149 /**
 150  * Session with another peer.
 151  */
 152 struct GNUNET_ATS_Session
 153 {
 154   /**
 155    * Which peer is this session for?
 156    */
 157   struct GNUNET_PeerIdentity target;
 158
 159   /**
 160    * Tokenizer for inbound messages.
 161    */
 162   struct GNUNET_MessageStreamTokenizer *mst;
 163
 164   /**
 165    * Plugin this session belongs to.
 166    */
 167   struct Plugin *plugin;
 168
 169   /**
 170    * Context for dealing with fragments.
 171    */
 172   struct UDP_FragmentationContext *frag_ctx;
 173
 174   /**
 175    * Desired delay for next sending we send to other peer
 176    */
 177   struct GNUNET_TIME_Relative flow_delay_for_other_peer;
 178
 179   /**
 180    * Desired delay for transmissions we received from other peer.
 181    * This is for full messages, the value needs to be adjusted for
 182    * fragmented messages.
 183    */
 184   struct GNUNET_TIME_Relative flow_delay_from_other_peer;
 185
 186   /**
 187    * Session timeout task
 188    */
 189   struct GNUNET_SCHEDULER_Task *timeout_task;
 190
 191   /**
 192    * When does this session time out?
 193    */
 194   struct GNUNET_TIME_Absolute timeout;
 195
 196   /**
 197    * What time did we last transmit?
 198    */
 199   struct GNUNET_TIME_Absolute last_transmit_time;
 200
 201   /**
 202    * expected delay for ACKs
 203    */
 204   struct GNUNET_TIME_Relative last_expected_ack_delay;
 205
 206   /**
 207    * desired delay between UDP messages
 208    */
 209   struct GNUNET_TIME_Relative last_expected_msg_delay;
 210
 211   /**
 212    * Our own address.
 213    */
 214   struct GNUNET_HELLO_Address *address;
 215
 216   /**
 217    * Number of bytes waiting for transmission to this peer.
 218    */
 219   unsigned long long bytes_in_queue;
 220
 221   /**
 222    * Number of messages waiting for transmission to this peer.
 223    */
 224   unsigned int msgs_in_queue;
 225
 226   /**
 227    * Reference counter to indicate that this session is
 228    * currently being used and must not be destroyed;
 229    * setting @e in_destroy will destroy it as soon as
 230    * possible.
 231    */
 232   unsigned int rc;
 233
 234   /**
 235    * Network type of the address.
 236    */
 237   enum GNUNET_ATS_Network_Type scope;
 238
 239   /**
 240    * Is this session about to be destroyed (sometimes we cannot
 241    * destroy a session immediately as below us on the stack
 242    * there might be code that still uses it; in this case,
 243    * @e rc is non-zero).
 244    */
 245   int in_destroy;
 246 };
 247
 248
 249
 250 /**
 251  * Data structure to track defragmentation contexts based
 252  * on the source of the UDP traffic.
 253  */
 254 struct DefragContext
 255 {
 256
 257   /**
 258    * Defragmentation context.
 259    */
 260   struct GNUNET_DEFRAGMENT_Context *defrag;
 261
 262   /**
 263    * Reference to master plugin struct.
 264    */
 265   struct Plugin *plugin;
 266
 267   /**
 268    * Node in the defrag heap.
 269    */
 270   struct GNUNET_CONTAINER_HeapNode *hnode;
 271
 272   /**
 273    * Source address this receive context is for (allocated at the
 274    * end of the struct).
 275    */
 276   const union UdpAddress *udp_addr;
 277
 278   /**
 279    * Who's message(s) are we defragmenting here?
 280    * Only initialized once we succeeded and
 281    * @e have_sender is set.
 282    */
 283   struct GNUNET_PeerIdentity sender;
 284
 285   /**
 286    * Length of @e udp_addr.
 287    */
 288   size_t udp_addr_len;
 289
 290   /**
 291    * Network type the address belongs to.
 292    */
 293   enum GNUNET_ATS_Network_Type network_type;
 294
 295   /**
 296    * Has the @e sender field been initialized yet?
 297    */
 298   int have_sender;
 299 };
 300
 301
 302 /**
 303  * Context to send fragmented messages
 304  */
 305 struct UDP_FragmentationContext
 306 {
 307   /**
 308    * Next in linked list
 309    */
 310   struct UDP_FragmentationContext *next;
 311
 312   /**
 313    * Previous in linked list
 314    */
 315   struct UDP_FragmentationContext *prev;
 316
 317   /**
 318    * The plugin
 319    */
 320   struct Plugin *plugin;
 321
 322   /**
 323    * Handle for fragmentation.
 324    */
 325   struct GNUNET_FRAGMENT_Context *frag;
 326
 327   /**
 328    * The session this fragmentation context belongs to
 329    */
 330   struct GNUNET_ATS_Session *session;
 331
 332   /**
 333    * Function to call upon completion of the transmission.
 334    */
 335   GNUNET_TRANSPORT_TransmitContinuation cont;
 336
 337   /**
 338    * Closure for @e cont.
 339    */
 340   void *cont_cls;
 341
 342   /**
 343    * Start time.
 344    */
 345   struct GNUNET_TIME_Absolute start_time;
 346
 347   /**
 348    * Transmission time for the next fragment.  Incremented by
 349    * the @e flow_delay_from_other_peer for each fragment when
 350    * we setup the fragments.
 351    */
 352   struct GNUNET_TIME_Absolute next_frag_time;
 353
 354   /**
 355    * Desired delay for transmissions we received from other peer.
 356    * Adjusted to be per fragment (UDP_MTU), even though on the
 357    * wire it was for "full messages".
 358    */
 359   struct GNUNET_TIME_Relative flow_delay_from_other_peer;
 360
 361   /**
 362    * Message timeout
 363    */
 364   struct GNUNET_TIME_Absolute timeout;
 365
 366   /**
 367    * Payload size of original unfragmented message
 368    */
 369   size_t payload_size;
 370
 371   /**
 372    * Bytes used to send all fragments on wire including UDP overhead
 373    */
 374   size_t on_wire_size;
 375
 376 };
 377
 378
 379 /**
 380  * Function called when a message is removed from the
 381  * transmission queue.
 382  *
 383  * @param cls closure
 384  * @param udpw message wrapper finished
 385  * @param result #GNUNET_OK on success (message was sent)
 386  *               #GNUNET_SYSERR if the target disconnected
 387  *               or we had a timeout or other trouble sending
 388  */
 389 typedef void
 390 (*QueueContinuation) (void *cls,
 391                       struct UDP_MessageWrapper *udpw,
 392                       int result);
 393
 394
 395 /**
 396  * Information we track for each message in the queue.
 397  */
 398 struct UDP_MessageWrapper
 399 {
 400   /**
 401    * Session this message belongs to
 402    */
 403   struct GNUNET_ATS_Session *session;
 404
 405   /**
 406    * DLL of messages, previous element
 407    */
 408   struct UDP_MessageWrapper *prev;
 409
 410   /**
 411    * DLL of messages, next element
 412    */
 413   struct UDP_MessageWrapper *next;
 414
 415   /**
 416    * Message with @e msg_size bytes including UDP-specific overhead.
 417    */
 418   char *msg_buf;
 419
 420   /**
 421    * Function to call once the message wrapper is being removed
 422    * from the queue (with success or failure).
 423    */
 424   QueueContinuation qc;
 425
 426   /**
 427    * Closure for @e qc.
 428    */
 429   void *qc_cls;
 430
 431   /**
 432    * External continuation to call upon completion of the
 433    * transmission, NULL if this queue entry is not for a
 434    * message from the application.
 435    */
 436   GNUNET_TRANSPORT_TransmitContinuation cont;
 437
 438   /**
 439    * Closure for @e cont.
 440    */
 441   void *cont_cls;
 442
 443   /**
 444    * Fragmentation context.
 445    * frag_ctx == NULL if transport <= MTU
 446    * frag_ctx != NULL if transport > MTU
 447    */
 448   struct UDP_FragmentationContext *frag_ctx;
 449
 450   /**
 451    * Message enqueue time.
 452    */
 453   struct GNUNET_TIME_Absolute start_time;
 454
 455   /**
 456    * Desired transmission time for this message, based on the
 457    * flow limiting information we got from the other peer.
 458    */
 459   struct GNUNET_TIME_Absolute transmission_time;
 460
 461   /**
 462    * Message timeout.
 463    */
 464   struct GNUNET_TIME_Absolute timeout;
 465
 466   /**
 467    * Size of UDP message to send, including UDP-specific overhead.
 468    */
 469   size_t msg_size;
 470
 471   /**
 472    * Payload size of original message.
 473    */
 474   size_t payload_size;
 475
 476 };
 477
 478
 479 GNUNET_NETWORK_STRUCT_BEGIN
 480
 481 /**
 482  * UDP ACK Message-Packet header.
 483  */
 484 struct UDP_ACK_Message
 485 {
 486   /**
 487    * Message header.
 488    */
 489   struct GNUNET_MessageHeader header;
 490
 491   /**
 492    * Desired delay for flow control, in us (in NBO).
 493    * A value of UINT32_MAX indicates that the other
 494    * peer wants us to disconnect.
 495    */
 496   uint32_t delay GNUNET_PACKED;
 497
 498   /**
 499    * What is the identity of the sender
 500    */
 501   struct GNUNET_PeerIdentity sender;
 502
 503 };
 504
 505 GNUNET_NETWORK_STRUCT_END
 506
 507
 508 /* ************************* Monitoring *********** */
 509
 510
 511 /**
 512  * If a session monitor is attached, notify it about the new
 513  * session state.
 514  *
 515  * @param plugin our plugin
 516  * @param session session that changed state
 517  * @param state new state of the session
 518  */
 519 static void
 520 notify_session_monitor (struct Plugin *plugin,
 521                         struct GNUNET_ATS_Session *session,
 522                         enum GNUNET_TRANSPORT_SessionState state)
 523 {
 524   struct GNUNET_TRANSPORT_SessionInfo info;
 525
 526   if (NULL == plugin->sic)
 527     return;
 528   if (GNUNET_YES == session->in_destroy)
 529     return; /* already destroyed, just RC>0 left-over actions */
 530   memset (&info,
 531           0,
 532           sizeof (info));
 533   info.state = state;
 534   info.is_inbound = GNUNET_SYSERR; /* hard to say */
 535   info.num_msg_pending = session->msgs_in_queue;
 536   info.num_bytes_pending = session->bytes_in_queue;
 537   /* info.receive_delay remains zero as this is not supported by UDP
 538      (cannot selectively not receive from 'some' peer while continuing
 539      to receive from others) */
 540   info.session_timeout = session->timeout;
 541   info.address = session->address;
 542   plugin->sic (plugin->sic_cls,
 543                session,
 544                &info);
 545 }
 546
 547
 548 /**
 549  * Return information about the given session to the monitor callback.
 550  *
 551  * @param cls the `struct Plugin` with the monitor callback (`sic`)
 552  * @param peer peer we send information about
 553  * @param value our `struct GNUNET_ATS_Session` to send information about
 554  * @return #GNUNET_OK (continue to iterate)
 555  */
 556 static int
 557 send_session_info_iter (void *cls,
 558                         const struct GNUNET_PeerIdentity *peer,
 559                         void *value)
 560 {
 561   struct Plugin *plugin = cls;
 562   struct GNUNET_ATS_Session *session = value;
 563
 564   notify_session_monitor (plugin,
 565                           session,
 566                           GNUNET_TRANSPORT_SS_INIT);
 567   notify_session_monitor (plugin,
 568                           session,
 569                           GNUNET_TRANSPORT_SS_UP);
 570   return GNUNET_OK;
 571 }
 572
 573
 574 /**
 575  * Begin monitoring sessions of a plugin.  There can only
 576  * be one active monitor per plugin (i.e. if there are
 577  * multiple monitors, the transport service needs to
 578  * multiplex the generated events over all of them).
 579  *
 580  * @param cls closure of the plugin
 581  * @param sic callback to invoke, NULL to disable monitor;
 582  *            plugin will being by iterating over all active
 583  *            sessions immediately and then enter monitor mode
 584  * @param sic_cls closure for @a sic
 585  */
 586 static void
 587 udp_plugin_setup_monitor (void *cls,
 588                           GNUNET_TRANSPORT_SessionInfoCallback sic,
 589                           void *sic_cls)
 590 {
 591   struct Plugin *plugin = cls;
 592
 593   plugin->sic = sic;
 594   plugin->sic_cls = sic_cls;
 595   if (NULL != sic)
 596   {
 597     GNUNET_CONTAINER_multipeermap_iterate (plugin->sessions,
 598                                            &send_session_info_iter,
 599                                            plugin);
 600     /* signal end of first iteration */
 601     sic (sic_cls,
 602          NULL,
 603          NULL);
 604   }
 605 }
 606
 607
 608 /* ****************** Little Helpers ****************** */
 609
 610
 611 /**
 612  * Function to free last resources associated with a session.
 613  *
 614  * @param s session to free
 615  */
 616 static void
 617 free_session (struct GNUNET_ATS_Session *s)
 618 {
 619   if (NULL != s->address)
 620   {
 621     GNUNET_HELLO_address_free (s->address);
 622     s->address = NULL;
 623   }
 624   if (NULL != s->frag_ctx)
 625   {
 626     GNUNET_FRAGMENT_context_destroy (s->frag_ctx->frag,
 627                                      NULL,
 628                                      NULL);
 629     GNUNET_free (s->frag_ctx);
 630     s->frag_ctx = NULL;
 631   }
 632   if (NULL != s->mst)
 633   {
 634     GNUNET_MST_destroy (s->mst);
 635     s->mst = NULL;
 636   }
 637   GNUNET_free (s);
 638 }
 639
 640
 641 /**
 642  * Function that is called to get the keepalive factor.
 643  * #GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT is divided by this number to
 644  * calculate the interval between keepalive packets.
 645  *
 646  * @param cls closure with the `struct Plugin`
 647  * @return keepalive factor
 648  */
 649 static unsigned int
 650 udp_query_keepalive_factor (void *cls)
 651 {
 652   return 15;
 653 }
 654
 655
 656 /**
 657  * Function obtain the network type for a session
 658  *
 659  * @param cls closure (`struct Plugin *`)
 660  * @param session the session
 661  * @return the network type
 662  */
 663 static enum GNUNET_ATS_Network_Type
 664 udp_plugin_get_network (void *cls,
 665                         struct GNUNET_ATS_Session *session)
 666 {
 667   return session->scope;
 668 }
 669
 670
 671 /**
 672  * Function obtain the network type for an address.
 673  *
 674  * @param cls closure (`struct Plugin *`)
 675  * @param address the address
 676  * @return the network type
 677  */
 678 static enum GNUNET_ATS_Network_Type
 679 udp_plugin_get_network_for_address (void *cls,
 680                                     const struct GNUNET_HELLO_Address *address)
 681 {
 682   struct Plugin *plugin = cls;
 683   size_t addrlen;
 684   struct sockaddr_in a4;
 685   struct sockaddr_in6 a6;
 686   const struct IPv4UdpAddress *u4;
 687   const struct IPv6UdpAddress *u6;
 688   const void *sb;
 689   size_t sbs;
 690
 691   addrlen = address->address_length;
 692   if (addrlen == sizeof(struct IPv6UdpAddress))
 693   {
 694     GNUNET_assert (NULL != address->address); /* make static analysis happy */
 695     u6 = address->address;
 696     memset (&a6, 0, sizeof(a6));
 697 #if HAVE_SOCKADDR_IN_SIN_LEN
 698     a6.sin6_len = sizeof (a6);
 699 #endif
 700     a6.sin6_family = AF_INET6;
 701     a6.sin6_port = u6->u6_port;
 702     GNUNET_memcpy (&a6.sin6_addr, &u6->ipv6_addr, sizeof(struct in6_addr));
 703     sb = &a6;
 704     sbs = sizeof(a6);
 705   }
 706   else if (addrlen == sizeof(struct IPv4UdpAddress))
 707   {
 708     GNUNET_assert (NULL != address->address); /* make static analysis happy */
 709     u4 = address->address;
 710     memset (&a4, 0, sizeof(a4));
 711 #if HAVE_SOCKADDR_IN_SIN_LEN
 712     a4.sin_len = sizeof (a4);
 713 #endif
 714     a4.sin_family = AF_INET;
 715     a4.sin_port = u4->u4_port;
 716     a4.sin_addr.s_addr = u4->ipv4_addr;
 717     sb = &a4;
 718     sbs = sizeof(a4);
 719   }
 720   else
 721   {
 722     GNUNET_break (0);
 723     return GNUNET_ATS_NET_UNSPECIFIED;
 724   }
 725   return plugin->env->get_address_type (plugin->env->cls,
 726                                         sb,
 727                                         sbs);
 728 }
 729
 730
 731 /* ******************* Event loop ******************** */
 732
 733 /**
 734  * We have been notified that our readset has something to read.  We don't
 735  * know which socket needs to be read, so we have to check each one
 736  * Then reschedule this function to be called again once more is available.
 737  *
 738  * @param cls the plugin handle
 739  */
 740 static void
 741 udp_plugin_select_v4 (void *cls);
 742
 743
 744 /**
 745  * We have been notified that our readset has something to read.  We don't
 746  * know which socket needs to be read, so we have to check each one
 747  * Then reschedule this function to be called again once more is available.
 748  *
 749  * @param cls the plugin handle
 750  */
 751 static void
 752 udp_plugin_select_v6 (void *cls);
 753
 754
 755 /**
 756  * (re)schedule IPv4-select tasks for this plugin.
 757  *
 758  * @param plugin plugin to reschedule
 759  */
 760 static void
 761 schedule_select_v4 (struct Plugin *plugin)
 762 {
 763   struct GNUNET_TIME_Relative min_delay;
 764   struct GNUNET_TIME_Relative delay;
 765   struct UDP_MessageWrapper *udpw;
 766   struct UDP_MessageWrapper *min_udpw;
 767
 768   if ( (GNUNET_YES == plugin->enable_ipv4) &&
 769        (NULL != plugin->sockv4) )
 770   {
 771     /* Find a message ready to send:
 772      * Flow delay from other peer is expired or not set (0) */
 773     min_delay = GNUNET_TIME_UNIT_FOREVER_REL;
 774     min_udpw = NULL;
 775     for (udpw = plugin->ipv4_queue_head; NULL != udpw; udpw = udpw->next)
 776     {
 777       delay = GNUNET_TIME_absolute_get_remaining (udpw->transmission_time);
 778       if (delay.rel_value_us < min_delay.rel_value_us)
 779       {
 780         min_delay = delay;
 781         min_udpw = udpw;
 782       }
 783     }
 784     if (NULL != plugin->select_task_v4)
 785       GNUNET_SCHEDULER_cancel (plugin->select_task_v4);
 786     if (NULL != min_udpw)
 787     {
 788       if (min_delay.rel_value_us > GNUNET_CONSTANTS_LATENCY_WARN.rel_value_us)
 789       {
 790         GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 791                     "Calculated flow delay for UDPv4 at %s for %s\n",
 792                     GNUNET_STRINGS_relative_time_to_string (min_delay,
 793                                                             GNUNET_YES),
 794                     GNUNET_i2s (&min_udpw->session->target));
 795       }
 796       else
 797       {
 798         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 799                     "Calculated flow delay for UDPv4 at %s for %s\n",
 800                     GNUNET_STRINGS_relative_time_to_string (min_delay,
 801                                                             GNUNET_YES),
 802                     GNUNET_i2s (&min_udpw->session->target));
 803       }
 804     }
 805     plugin->select_task_v4
 806       = GNUNET_SCHEDULER_add_read_net (min_delay,
 807                                        plugin->sockv4,
 808                                        &udp_plugin_select_v4,
 809                                        plugin);
 810   }
 811 }
 812
 813
 814 /**
 815  * (re)schedule IPv6-select tasks for this plugin.
 816  *
 817  * @param plugin plugin to reschedule
 818  */
 819 static void
 820 schedule_select_v6 (struct Plugin *plugin)
 821 {
 822   struct GNUNET_TIME_Relative min_delay;
 823   struct GNUNET_TIME_Relative delay;
 824   struct UDP_MessageWrapper *udpw;
 825   struct UDP_MessageWrapper *min_udpw;
 826
 827   if ( (GNUNET_YES == plugin->enable_ipv6) &&
 828        (NULL != plugin->sockv6) )
 829   {
 830     min_delay = GNUNET_TIME_UNIT_FOREVER_REL;
 831     min_udpw = NULL;
 832     for (udpw = plugin->ipv6_queue_head; NULL != udpw; udpw = udpw->next)
 833     {
 834       delay = GNUNET_TIME_absolute_get_remaining (udpw->transmission_time);
 835       if (delay.rel_value_us < min_delay.rel_value_us)
 836       {
 837         min_delay = delay;
 838         min_udpw = udpw;
 839       }
 840     }
 841     if (NULL != plugin->select_task_v6)
 842       GNUNET_SCHEDULER_cancel (plugin->select_task_v6);
 843     if (NULL != min_udpw)
 844     {
 845       if (min_delay.rel_value_us > GNUNET_CONSTANTS_LATENCY_WARN.rel_value_us)
 846       {
 847         GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 848                     "Calculated flow delay for UDPv6 at %s for %s\n",
 849                     GNUNET_STRINGS_relative_time_to_string (min_delay,
 850                                                             GNUNET_YES),
 851                     GNUNET_i2s (&min_udpw->session->target));
 852       }
 853       else
 854       {
 855         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 856                     "Calculated flow delay for UDPv6 at %s for %s\n",
 857                     GNUNET_STRINGS_relative_time_to_string (min_delay,
 858                                                             GNUNET_YES),
 859                     GNUNET_i2s (&min_udpw->session->target));
 860       }
 861     }
 862     plugin->select_task_v6
 863       = GNUNET_SCHEDULER_add_read_net (min_delay,
 864                                        plugin->sockv6,
 865                                        &udp_plugin_select_v6,
 866                                        plugin);
 867   }
 868 }
 869
 870
 871 /* ******************* Address to string and back ***************** */
 872
 873
 874 /**
 875  * Function called for a quick conversion of the binary address to
 876  * a numeric address.  Note that the caller must not free the
 877  * address and that the next call to this function is allowed
 878  * to override the address again.
 879  *
 880  * @param cls closure
 881  * @param addr binary address (a `union UdpAddress`)
 882  * @param addrlen length of the @a addr
 883  * @return string representing the same address
 884  */
 885 const char *
 886 udp_address_to_string (void *cls,
 887                        const void *addr,
 888                        size_t addrlen)
 889 {
 890   static char rbuf[INET6_ADDRSTRLEN + 10];
 891   char buf[INET6_ADDRSTRLEN];
 892   const void *sb;
 893   struct in_addr a4;
 894   struct in6_addr a6;
 895   const struct IPv4UdpAddress *t4;
 896   const struct IPv6UdpAddress *t6;
 897   int af;
 898   uint16_t port;
 899   uint32_t options;
 900
 901   if (NULL == addr)
 902   {
 903     GNUNET_break_op (0);
 904     return NULL;
 905   }
 906
 907   if (addrlen == sizeof(struct IPv6UdpAddress))
 908   {
 909     t6 = addr;
 910     af = AF_INET6;
 911     options = ntohl (t6->options);
 912     port = ntohs (t6->u6_port);
 913     a6 = t6->ipv6_addr;
 914     sb = &a6;
 915   }
 916   else if (addrlen == sizeof(struct IPv4UdpAddress))
 917   {
 918     t4 = addr;
 919     af = AF_INET;
 920     options = ntohl (t4->options);
 921     port = ntohs (t4->u4_port);
 922     a4.s_addr = t4->ipv4_addr;
 923     sb = &a4;
 924   }
 925   else
 926   {
 927     GNUNET_break_op (0);
 928     return NULL;
 929   }
 930   inet_ntop (af,
 931              sb,
 932              buf,
 933              INET6_ADDRSTRLEN);
 934   GNUNET_snprintf (rbuf,
 935                    sizeof(rbuf),
 936                    (af == AF_INET6)
 937                    ? "%s.%u.[%s]:%u"
 938                    : "%s.%u.%s:%u",
 939                    PLUGIN_NAME,
 940                    options,
 941                    buf,
 942                    port);
 943   return rbuf;
 944 }
 945
 946
 947 /**
 948  * Function called to convert a string address to a binary address.
 949  *
 950  * @param cls closure (`struct Plugin *`)
 951  * @param addr string address
 952  * @param addrlen length of the address
 953  * @param buf location to store the buffer
 954  * @param added location to store the number of bytes in the buffer.
 955  *        If the function returns #GNUNET_SYSERR, its contents are undefined.
 956  * @return #GNUNET_OK on success, #GNUNET_SYSERR on failure
 957  */
 958 static int
 959 udp_string_to_address (void *cls,
 960                        const char *addr,
 961                        uint16_t addrlen,
 962                        void **buf,
 963                        size_t *added)
 964 {
 965   struct sockaddr_storage socket_address;
 966   char *address;
 967   char *plugin;
 968   char *optionstr;
 969   uint32_t options;
 970
 971   /* Format tcp.options.address:port */
 972   address = NULL;
 973   plugin = NULL;
 974   optionstr = NULL;
 975
 976   if ((NULL == addr) || (0 == addrlen))
 977   {
 978     GNUNET_break (0);
 979     return GNUNET_SYSERR;
 980   }
 981   if ('\0' != addr[addrlen - 1])
 982   {
 983     GNUNET_break (0);
 984     return GNUNET_SYSERR;
 985   }
 986   if (strlen (addr) != addrlen - 1)
 987   {
 988     GNUNET_break (0);
 989     return GNUNET_SYSERR;
 990   }
 991   plugin = GNUNET_strdup (addr);
 992   optionstr = strchr (plugin, '.');
 993   if (NULL == optionstr)
 994   {
 995     GNUNET_break (0);
 996     GNUNET_free (plugin);
 997     return GNUNET_SYSERR;
 998   }
 999   optionstr[0] = '\0';
1000   optionstr++;
1001   options = atol (optionstr);
1002   address = strchr (optionstr, '.');
1003   if (NULL == address)
1004   {
1005     GNUNET_break (0);
1006     GNUNET_free (plugin);
1007     return GNUNET_SYSERR;
1008   }
1009   address[0] = '\0';
1010   address++;
1011
1012   if (GNUNET_OK !=
1013       GNUNET_STRINGS_to_address_ip (address,
1014                                     strlen (address),
1015                                     &socket_address))
1016   {
1017     GNUNET_break (0);
1018     GNUNET_free (plugin);
1019     return GNUNET_SYSERR;
1020   }
1021   GNUNET_free(plugin);
1022
1023   switch (socket_address.ss_family)
1024   {
1025   case AF_INET:
1026     {
1027       struct IPv4UdpAddress *u4;
1028       const struct sockaddr_in *in4 = (const struct sockaddr_in *) &socket_address;
1029
1030       u4 = GNUNET_new (struct IPv4UdpAddress);
1031       u4->options = htonl (options);
1032       u4->ipv4_addr = in4->sin_addr.s_addr;
1033       u4->u4_port = in4->sin_port;
1034       *buf = u4;
1035       *added = sizeof (struct IPv4UdpAddress);
1036       return GNUNET_OK;
1037     }
1038   case AF_INET6:
1039     {
1040       struct IPv6UdpAddress *u6;
1041       const struct sockaddr_in6 *in6 = (const struct sockaddr_in6 *) &socket_address;
1042
1043       u6 = GNUNET_new (struct IPv6UdpAddress);
1044       u6->options = htonl (options);
1045       u6->ipv6_addr = in6->sin6_addr;
1046       u6->u6_port = in6->sin6_port;
1047       *buf = u6;
1048       *added = sizeof (struct IPv6UdpAddress);
1049       return GNUNET_OK;
1050     }
1051   default:
1052     GNUNET_break (0);
1053     return GNUNET_SYSERR;
1054   }
1055 }
1056
1057
1058 /**
1059  * Append our port and forward the result.
1060  *
1061  * @param cls a `struct PrettyPrinterContext *`
1062  * @param hostname result from DNS resolver
1063  */
1064 static void
1065 append_port (void *cls,
1066              const char *hostname)
1067 {
1068   struct PrettyPrinterContext *ppc = cls;
1069   struct Plugin *plugin = ppc->plugin;
1070   char *ret;
1071
1072   if (NULL == hostname)
1073   {
1074     /* Final call, done */
1075     GNUNET_CONTAINER_DLL_remove (plugin->ppc_dll_head,
1076                                  plugin->ppc_dll_tail,
1077                                  ppc);
1078     ppc->resolver_handle = NULL;
1079     ppc->asc (ppc->asc_cls,
1080               NULL,
1081               GNUNET_OK);
1082     GNUNET_free (ppc);
1083     return;
1084   }
1085   if (GNUNET_YES == ppc->ipv6)
1086     GNUNET_asprintf (&ret,
1087                      "%s.%u.[%s]:%d",
1088                      PLUGIN_NAME,
1089                      ppc->options,
1090                      hostname,
1091                      ppc->port);
1092   else
1093     GNUNET_asprintf (&ret,
1094                      "%s.%u.%s:%d",
1095                      PLUGIN_NAME,
1096                      ppc->options,
1097                      hostname,
1098                      ppc->port);
1099   ppc->asc (ppc->asc_cls,
1100             ret,
1101             GNUNET_OK);
1102   GNUNET_free (ret);
1103 }
1104
1105
1106 /**
1107  * Convert the transports address to a nice, human-readable format.
1108  *
1109  * @param cls closure with the `struct Plugin *`
1110  * @param type name of the transport that generated the address
1111  * @param addr one of the addresses of the host, NULL for the last address
1112  *        the specific address format depends on the transport;
1113  *        a `union UdpAddress`
1114  * @param addrlen length of the address
1115  * @param numeric should (IP) addresses be displayed in numeric form?
1116  * @param timeout after how long should we give up?
1117  * @param asc function to call on each string
1118  * @param asc_cls closure for @a asc
1119  */
1120 static void
1121 udp_plugin_address_pretty_printer (void *cls,
1122                                    const char *type,
1123                                    const void *addr,
1124                                    size_t addrlen,
1125                                    int numeric,
1126                                    struct GNUNET_TIME_Relative timeout,
1127                                    GNUNET_TRANSPORT_AddressStringCallback asc,
1128                                    void *asc_cls)
1129 {
1130   struct Plugin *plugin = cls;
1131   struct PrettyPrinterContext *ppc;
1132   const struct sockaddr *sb;
1133   size_t sbs;
1134   struct sockaddr_in a4;
1135   struct sockaddr_in6 a6;
1136   const struct IPv4UdpAddress *u4;
1137   const struct IPv6UdpAddress *u6;
1138   uint16_t port;
1139   uint32_t options;
1140
1141   if (addrlen == sizeof(struct IPv6UdpAddress))
1142   {
1143     u6 = addr;
1144     memset (&a6,
1145             0,
1146             sizeof (a6));
1147     a6.sin6_family = AF_INET6;
1148 #if HAVE_SOCKADDR_IN_SIN_LEN
1149     a6.sin6_len = sizeof (a6);
1150 #endif
1151     a6.sin6_port = u6->u6_port;
1152     a6.sin6_addr = u6->ipv6_addr;
1153     port = ntohs (u6->u6_port);
1154     options = ntohl (u6->options);
1155     sb = (const struct sockaddr *) &a6;
1156     sbs = sizeof (a6);
1157   }
1158   else if (addrlen == sizeof (struct IPv4UdpAddress))
1159   {
1160     u4 = addr;
1161     memset (&a4,
1162             0,
1163             sizeof(a4));
1164     a4.sin_family = AF_INET;
1165 #if HAVE_SOCKADDR_IN_SIN_LEN
1166     a4.sin_len = sizeof (a4);
1167 #endif
1168     a4.sin_port = u4->u4_port;
1169     a4.sin_addr.s_addr = u4->ipv4_addr;
1170     port = ntohs (u4->u4_port);
1171     options = ntohl (u4->options);
1172     sb = (const struct sockaddr *) &a4;
1173     sbs = sizeof(a4);
1174   }
1175   else
1176   {
1177     /* invalid address */
1178     GNUNET_break_op (0);
1179     asc (asc_cls,
1180          NULL,
1181          GNUNET_SYSERR);
1182     asc (asc_cls,
1183          NULL,
1184          GNUNET_OK);
1185     return;
1186   }
1187   ppc = GNUNET_new (struct PrettyPrinterContext);
1188   ppc->plugin = plugin;
1189   ppc->asc = asc;
1190   ppc->asc_cls = asc_cls;
1191   ppc->port = port;
1192   ppc->options = options;
1193   if (addrlen == sizeof (struct IPv6UdpAddress))
1194     ppc->ipv6 = GNUNET_YES;
1195   else
1196     ppc->ipv6 = GNUNET_NO;
1197   GNUNET_CONTAINER_DLL_insert (plugin->ppc_dll_head,
1198                                plugin->ppc_dll_tail,
1199                                ppc);
1200   ppc->resolver_handle
1201     = GNUNET_RESOLVER_hostname_get (sb,
1202                                     sbs,
1203                                     ! numeric,
1204                                     timeout,
1205                                     &append_port,
1206                                     ppc);
1207 }
1208
1209
1210 /**
1211  * Check if the given port is plausible (must be either our listen
1212  * port or our advertised port).  If it is neither, we return
1213  * #GNUNET_SYSERR.
1214  *
1215  * @param plugin global variables
1216  * @param in_port port number to check
1217  * @return #GNUNET_OK if port is either our open or advertised port
1218  */
1219 static int
1220 check_port (const struct Plugin *plugin,
1221             uint16_t in_port)
1222 {
1223   if ( (plugin->port == in_port) ||
1224        (plugin->aport == in_port) )
1225     return GNUNET_OK;
1226   return GNUNET_SYSERR;
1227 }
1228
1229
1230 /**
1231  * Function that will be called to check if a binary address for this
1232  * plugin is well-formed and corresponds to an address for THIS peer
1233  * (as per our configuration).  Naturally, if absolutely necessary,
1234  * plugins can be a bit conservative in their answer, but in general
1235  * plugins should make sure that the address does not redirect
1236  * traffic to a 3rd party that might try to man-in-the-middle our
1237  * traffic.
1238  *
1239  * @param cls closure, should be our handle to the Plugin
1240  * @param addr pointer to a `union UdpAddress`
1241  * @param addrlen length of @a addr
1242  * @return #GNUNET_OK if this is a plausible address for this peer
1243  *         and transport, #GNUNET_SYSERR if not
1244  */
1245 static int
1246 udp_plugin_check_address (void *cls,
1247                           const void *addr,
1248                           size_t addrlen)
1249 {
1250   struct Plugin *plugin = cls;
1251   const struct IPv4UdpAddress *v4;
1252   const struct IPv6UdpAddress *v6;
1253
1254   if (sizeof(struct IPv4UdpAddress) == addrlen)
1255   {
1256     struct sockaddr_in s4;
1257
1258     v4 = (const struct IPv4UdpAddress *) addr;
1259     if (GNUNET_OK != check_port (plugin,
1260                                  ntohs (v4->u4_port)))
1261       return GNUNET_SYSERR;
1262     memset (&s4, 0, sizeof (s4));
1263     s4.sin_family = AF_INET;
1264 #if HAVE_SOCKADDR_IN_SIN_LEN
1265     s4.sin_len = sizeof (s4);
1266 #endif
1267     s4.sin_port = v4->u4_port;
1268     s4.sin_addr.s_addr = v4->ipv4_addr;
1269
1270     if (GNUNET_OK !=
1271         GNUNET_NAT_test_address (plugin->nat,
1272                                  &s4,
1273                                  sizeof (struct sockaddr_in)))
1274       return GNUNET_SYSERR;
1275   }
1276   else if (sizeof(struct IPv6UdpAddress) == addrlen)
1277   {
1278     struct sockaddr_in6 s6;
1279
1280     v6 = (const struct IPv6UdpAddress *) addr;
1281     if (IN6_IS_ADDR_LINKLOCAL (&v6->ipv6_addr))
1282       return GNUNET_OK; /* plausible, if unlikely... */
1283     memset (&s6, 0, sizeof (s6));
1284     s6.sin6_family = AF_INET6;
1285 #if HAVE_SOCKADDR_IN_SIN_LEN
1286     s6.sin6_len = sizeof (s6);
1287 #endif
1288     s6.sin6_port = v6->u6_port;
1289     s6.sin6_addr = v6->ipv6_addr;
1290
1291     if (GNUNET_OK !=
1292         GNUNET_NAT_test_address (plugin->nat,
1293                                  &s6,
1294                                  sizeof(struct sockaddr_in6)))
1295       return GNUNET_SYSERR;
1296   }
1297   else
1298   {
1299     GNUNET_break_op (0);
1300     return GNUNET_SYSERR;
1301   }
1302   return GNUNET_OK;
1303 }
1304
1305
1306 /**
1307  * Our external IP address/port mapping has changed.
1308  *
1309  * @param cls closure, the `struct Plugin`
1310  * @param add_remove #GNUNET_YES to mean the new public IP address,
1311  *                   #GNUNET_NO to mean the previous (now invalid) one
1312  * @param ac address class the address belongs to
1313  * @param addr either the previous or the new public IP address
1314  * @param addrlen actual length of the @a addr
1315  */
1316 static void
1317 udp_nat_port_map_callback (void *cls,
1318                            int add_remove,
1319                            enum GNUNET_NAT_AddressClass ac,
1320                            const struct sockaddr *addr,
1321                            socklen_t addrlen)
1322 {
1323   struct Plugin *plugin = cls;
1324   struct GNUNET_HELLO_Address *address;
1325   struct IPv4UdpAddress u4;
1326   struct IPv6UdpAddress u6;
1327   void *arg;
1328   size_t args;
1329
1330   LOG (GNUNET_ERROR_TYPE_DEBUG,
1331        (GNUNET_YES == add_remove)
1332        ? "NAT notification to add address `%s'\n"
1333        : "NAT notification to remove address `%s'\n",
1334        GNUNET_a2s (addr,
1335                    addrlen));
1336   /* convert 'address' to our internal format */
1337   switch (addr->sa_family)
1338   {
1339   case AF_INET:
1340     {
1341       const struct sockaddr_in *i4;
1342
1343       GNUNET_assert (sizeof(struct sockaddr_in) == addrlen);
1344       i4 = (const struct sockaddr_in *) addr;
1345       if (0 == ntohs (i4->sin_port))
1346         return; /* Port = 0 means unmapped, ignore these for UDP. */
1347       memset (&u4,
1348               0,
1349               sizeof(u4));
1350       u4.options = htonl (plugin->myoptions);
1351       u4.ipv4_addr = i4->sin_addr.s_addr;
1352       u4.u4_port = i4->sin_port;
1353       arg = &u4;
1354       args = sizeof (struct IPv4UdpAddress);
1355       break;
1356     }
1357   case AF_INET6:
1358     {
1359       const struct sockaddr_in6 *i6;
1360
1361       GNUNET_assert (sizeof(struct sockaddr_in6) == addrlen);
1362       i6 = (const struct sockaddr_in6 *) addr;
1363       if (0 == ntohs (i6->sin6_port))
1364         return; /* Port = 0 means unmapped, ignore these for UDP. */
1365       memset (&u6,
1366               0,
1367               sizeof(u6));
1368       u6.options = htonl (plugin->myoptions);
1369       u6.ipv6_addr = i6->sin6_addr;
1370       u6.u6_port = i6->sin6_port;
1371       arg = &u6;
1372       args = sizeof (struct IPv6UdpAddress);
1373       break;
1374     }
1375   default:
1376     GNUNET_break (0);
1377     return;
1378   }
1379   /* modify our published address list */
1380   /* TODO: use 'ac' here in the future... */
1381   address = GNUNET_HELLO_address_allocate (plugin->env->my_identity,
1382                                            PLUGIN_NAME,
1383                                            arg,
1384                                            args,
1385                                            GNUNET_HELLO_ADDRESS_INFO_NONE);
1386   plugin->env->notify_address (plugin->env->cls,
1387                                add_remove,
1388                                address);
1389   GNUNET_HELLO_address_free (address);
1390 }
1391
1392
1393 /* ********************* Finding sessions ******************* */
1394
1395
1396 /**
1397  * Closure for #session_cmp_it().
1398  */
1399 struct GNUNET_ATS_SessionCompareContext
1400 {
1401   /**
1402    * Set to session matching the address.
1403    */
1404   struct GNUNET_ATS_Session *res;
1405
1406   /**
1407    * Address we are looking for.
1408    */
1409   const struct GNUNET_HELLO_Address *address;
1410 };
1411
1412
1413 /**
1414  * Find a session with a matching address.
1415  *
1416  * @param cls the `struct GNUNET_ATS_SessionCompareContext *`
1417  * @param key peer identity (unused)
1418  * @param value the `struct GNUNET_ATS_Session *`
1419  * @return #GNUNET_NO if we found the session, #GNUNET_OK if not
1420  */
1421 static int
1422 session_cmp_it (void *cls,
1423                 const struct GNUNET_PeerIdentity *key,
1424                 void *value)
1425 {
1426   struct GNUNET_ATS_SessionCompareContext *cctx = cls;
1427   struct GNUNET_ATS_Session *s = value;
1428
1429   if (0 == GNUNET_HELLO_address_cmp (s->address,
1430                                      cctx->address))
1431   {
1432     GNUNET_assert (GNUNET_NO == s->in_destroy);
1433     cctx->res = s;
1434     return GNUNET_NO;
1435   }
1436   return GNUNET_OK;
1437 }
1438
1439
1440 /**
1441  * Locate an existing session the transport service is using to
1442  * send data to another peer.  Performs some basic sanity checks
1443  * on the address and then tries to locate a matching session.
1444  *
1445  * @param cls the plugin
1446  * @param address the address we should locate the session by
1447  * @return the session if it exists, or NULL if it is not found
1448  */
1449 static struct GNUNET_ATS_Session *
1450 udp_plugin_lookup_session (void *cls,
1451                            const struct GNUNET_HELLO_Address *address)
1452 {
1453   struct Plugin *plugin = cls;
1454   const struct IPv6UdpAddress *udp_a6;
1455   const struct IPv4UdpAddress *udp_a4;
1456   struct GNUNET_ATS_SessionCompareContext cctx;
1457
1458   if (NULL == address->address)
1459   {
1460     GNUNET_break (0);
1461     return NULL;
1462   }
1463   if (sizeof(struct IPv4UdpAddress) == address->address_length)
1464   {
1465     if (NULL == plugin->sockv4)
1466       return NULL;
1467     udp_a4 = (const struct IPv4UdpAddress *) address->address;
1468     if (0 == udp_a4->u4_port)
1469     {
1470       GNUNET_break (0);
1471       return NULL;
1472     }
1473   }
1474   else if (sizeof(struct IPv6UdpAddress) == address->address_length)
1475   {
1476     if (NULL == plugin->sockv6)
1477       return NULL;
1478     udp_a6 = (const struct IPv6UdpAddress *) address->address;
1479     if (0 == udp_a6->u6_port)
1480     {
1481       GNUNET_break (0);
1482       return NULL;
1483     }
1484   }
1485   else
1486   {
1487     GNUNET_break (0);
1488     return NULL;
1489   }
1490
1491   /* check if session already exists */
1492   cctx.address = address;
1493   cctx.res = NULL;
1494   LOG (GNUNET_ERROR_TYPE_DEBUG,
1495        "Looking for existing session for peer `%s' with address `%s'\n",
1496        GNUNET_i2s (&address->peer),
1497        udp_address_to_string (plugin,
1498                               address->address,
1499                               address->address_length));
1500   GNUNET_CONTAINER_multipeermap_get_multiple (plugin->sessions,
1501                                               &address->peer,
1502                                               &session_cmp_it,
1503                                               &cctx);
1504   if (NULL == cctx.res)
1505     return NULL;
1506   LOG (GNUNET_ERROR_TYPE_DEBUG,
1507        "Found existing session %p\n",
1508        cctx.res);
1509   return cctx.res;
1510 }
1511
1512
1513 /* ********************** Timeout ****************** */
1514
1515
1516 /**
1517  * Increment session timeout due to activity.
1518  *
1519  * @param s session to reschedule timeout activity for
1520  */
1521 static void
1522 reschedule_session_timeout (struct GNUNET_ATS_Session *s)
1523 {
1524   if (GNUNET_YES == s->in_destroy)
1525     return;
1526   GNUNET_assert (NULL != s->timeout_task);
1527   s->timeout = GNUNET_TIME_relative_to_absolute (UDP_SESSION_TIME_OUT);
1528 }
1529
1530
1531
1532 /**
1533  * Function that will be called whenever the transport service wants to
1534  * notify the plugin that a session is still active and in use and
1535  * therefore the session timeout for this session has to be updated
1536  *
1537  * @param cls closure with the `struct Plugin`
1538  * @param peer which peer was the session for
1539  * @param session which session is being updated
1540  */
1541 static void
1542 udp_plugin_update_session_timeout (void *cls,
1543                                    const struct GNUNET_PeerIdentity *peer,
1544                                    struct GNUNET_ATS_Session *session)
1545 {
1546   struct Plugin *plugin = cls;
1547
1548   if (GNUNET_YES !=
1549       GNUNET_CONTAINER_multipeermap_contains_value (plugin->sessions,
1550                                                     peer,
1551                                                     session))
1552   {
1553     GNUNET_break (0);
1554     return;
1555   }
1556   /* Reschedule session timeout */
1557   reschedule_session_timeout (session);
1558 }
1559
1560
1561 /* ************************* Sending ************************ */
1562
1563
1564 /**
1565  * Remove the given message from the transmission queue and
1566  * update all applicable statistics.
1567  *
1568  * @param plugin the UDP plugin
1569  * @param udpw message wrapper to dequeue
1570  */
1571 static void
1572 dequeue (struct Plugin *plugin,
1573          struct UDP_MessageWrapper *udpw)
1574 {
1575   struct GNUNET_ATS_Session *session = udpw->session;
1576
1577   if (plugin->bytes_in_buffer < udpw->msg_size)
1578   {
1579     GNUNET_break (0);
1580   }
1581   else
1582   {
1583     GNUNET_STATISTICS_update (plugin->env->stats,
1584                               "# UDP, total bytes in send buffers",
1585                               - (long long) udpw->msg_size,
1586                               GNUNET_NO);
1587     plugin->bytes_in_buffer -= udpw->msg_size;
1588   }
1589   GNUNET_STATISTICS_update (plugin->env->stats,
1590                             "# UDP, total messages in send buffers",
1591                             -1,
1592                             GNUNET_NO);
1593   if (sizeof(struct IPv4UdpAddress) == udpw->session->address->address_length)
1594   {
1595     GNUNET_CONTAINER_DLL_remove (plugin->ipv4_queue_head,
1596                                  plugin->ipv4_queue_tail,
1597                                  udpw);
1598   }
1599   else if (sizeof(struct IPv6UdpAddress) == udpw->session->address->address_length)
1600   {
1601     GNUNET_CONTAINER_DLL_remove (plugin->ipv6_queue_head,
1602                                  plugin->ipv6_queue_tail,
1603                                  udpw);
1604   }
1605   else
1606   {
1607     GNUNET_break (0);
1608     return;
1609   }
1610   GNUNET_assert (session->msgs_in_queue > 0);
1611   session->msgs_in_queue--;
1612   GNUNET_assert (session->bytes_in_queue >= udpw->msg_size);
1613   session->bytes_in_queue -= udpw->msg_size;
1614 }
1615
1616
1617 /**
1618  * Enqueue a message for transmission and update statistics.
1619  *
1620  * @param plugin the UDP plugin
1621  * @param udpw message wrapper to queue
1622  */
1623 static void
1624 enqueue (struct Plugin *plugin,
1625          struct UDP_MessageWrapper *udpw)
1626 {
1627   struct GNUNET_ATS_Session *session = udpw->session;
1628
1629   if (GNUNET_YES == session->in_destroy)
1630   {
1631     GNUNET_break (0);
1632     GNUNET_free (udpw);
1633     return;
1634   }
1635   if (plugin->bytes_in_buffer > INT64_MAX - udpw->msg_size)
1636   {
1637     GNUNET_break (0);
1638   }
1639   else
1640   {
1641     GNUNET_STATISTICS_update (plugin->env->stats,
1642                               "# UDP, total bytes in send buffers",
1643                               udpw->msg_size,
1644                               GNUNET_NO);
1645     plugin->bytes_in_buffer += udpw->msg_size;
1646   }
1647   GNUNET_STATISTICS_update (plugin->env->stats,
1648                             "# UDP, total messages in send buffers",
1649                             1,
1650                             GNUNET_NO);
1651   if (sizeof (struct IPv4UdpAddress) == udpw->session->address->address_length)
1652   {
1653     GNUNET_CONTAINER_DLL_insert(plugin->ipv4_queue_head,
1654                                 plugin->ipv4_queue_tail,
1655                                 udpw);
1656   }
1657   else if (sizeof (struct IPv6UdpAddress) == udpw->session->address->address_length)
1658   {
1659     GNUNET_CONTAINER_DLL_insert (plugin->ipv6_queue_head,
1660                                  plugin->ipv6_queue_tail,
1661                                  udpw);
1662   }
1663   else
1664   {
1665     GNUNET_break (0);
1666     udpw->cont (udpw->cont_cls,
1667                 &session->target,
1668                 GNUNET_SYSERR,
1669                 udpw->msg_size,
1670                 0);
1671     GNUNET_free (udpw);
1672     return;
1673   }
1674   session->msgs_in_queue++;
1675   session->bytes_in_queue += udpw->msg_size;
1676 }
1677
1678
1679 /**
1680  * We have completed our (attempt) to transmit a message that had to
1681  * be fragmented -- either because we got an ACK saying that all
1682  * fragments were received, or because of timeout / disconnect.  Clean
1683  * up our state.
1684  *
1685  * @param frag_ctx fragmentation context to clean up
1686  * @param result #GNUNET_OK if we succeeded (got ACK),
1687  *               #GNUNET_SYSERR if the transmission failed
1688  */
1689 static void
1690 fragmented_message_done (struct UDP_FragmentationContext *frag_ctx,
1691                          int result)
1692 {
1693   struct Plugin *plugin = frag_ctx->plugin;
1694   struct GNUNET_ATS_Session *s = frag_ctx->session;
1695   struct UDP_MessageWrapper *udpw;
1696   struct UDP_MessageWrapper *tmp;
1697   size_t overhead;
1698   struct GNUNET_TIME_Relative delay;
1699
1700   LOG (GNUNET_ERROR_TYPE_DEBUG,
1701        "%p: Fragmented message removed with result %s\n",
1702        frag_ctx,
1703        (result == GNUNET_SYSERR) ? "FAIL" : "SUCCESS");
1704   /* Call continuation for fragmented message */
1705   if (frag_ctx->on_wire_size >= frag_ctx->payload_size)
1706     overhead = frag_ctx->on_wire_size - frag_ctx->payload_size;
1707   else
1708     overhead = frag_ctx->on_wire_size;
1709   delay = GNUNET_TIME_absolute_get_duration (frag_ctx->start_time);
1710   if (delay.rel_value_us > GNUNET_CONSTANTS_LATENCY_WARN.rel_value_us)
1711   {
1712     LOG (GNUNET_ERROR_TYPE_WARNING,
1713          "Fragmented message acknowledged after %s (expected at %s)\n",
1714          GNUNET_STRINGS_relative_time_to_string (delay,
1715                                                  GNUNET_YES),
1716          GNUNET_STRINGS_absolute_time_to_string (frag_ctx->next_frag_time));
1717   }
1718   else
1719   {
1720     LOG (GNUNET_ERROR_TYPE_DEBUG,
1721          "Fragmented message acknowledged after %s (expected at %s)\n",
1722          GNUNET_STRINGS_relative_time_to_string (delay,
1723                                                  GNUNET_YES),
1724          GNUNET_STRINGS_absolute_time_to_string (frag_ctx->next_frag_time));
1725   }
1726
1727   if (NULL != frag_ctx->cont)
1728     frag_ctx->cont (frag_ctx->cont_cls,
1729                     &s->target,
1730                     result,
1731                     s->frag_ctx->payload_size,
1732                     frag_ctx->on_wire_size);
1733   GNUNET_STATISTICS_update (plugin->env->stats,
1734                             "# UDP, fragmented messages active",
1735                             -1,
1736                             GNUNET_NO);
1737
1738   if (GNUNET_OK == result)
1739   {
1740     GNUNET_STATISTICS_update (plugin->env->stats,
1741                               "# UDP, fragmented msgs, messages, sent, success",
1742                               1,
1743                               GNUNET_NO);
1744     GNUNET_STATISTICS_update (plugin->env->stats,
1745                               "# UDP, fragmented msgs, bytes payload, sent, success",
1746                               s->frag_ctx->payload_size,
1747                               GNUNET_NO);
1748     GNUNET_STATISTICS_update (plugin->env->stats,
1749                               "# UDP, fragmented msgs, bytes overhead, sent, success",
1750                               overhead,
1751                               GNUNET_NO);
1752     GNUNET_STATISTICS_update (plugin->env->stats,
1753                               "# UDP, total, bytes overhead, sent",
1754                               overhead,
1755                               GNUNET_NO);
1756     GNUNET_STATISTICS_update (plugin->env->stats,
1757                               "# UDP, total, bytes payload, sent",
1758                               s->frag_ctx->payload_size,
1759                               GNUNET_NO);
1760   }
1761   else
1762   {
1763     GNUNET_STATISTICS_update (plugin->env->stats,
1764                               "# UDP, fragmented msgs, messages, sent, failure",
1765                               1,
1766                               GNUNET_NO);
1767     GNUNET_STATISTICS_update (plugin->env->stats,
1768                               "# UDP, fragmented msgs, bytes payload, sent, failure",
1769                               s->frag_ctx->payload_size,
1770                               GNUNET_NO);
1771     GNUNET_STATISTICS_update (plugin->env->stats,
1772                               "# UDP, fragmented msgs, bytes payload, sent, failure",
1773                               overhead,
1774                               GNUNET_NO);
1775     GNUNET_STATISTICS_update (plugin->env->stats,
1776                               "# UDP, fragmented msgs, bytes payload, sent, failure",
1777                               overhead,
1778                               GNUNET_NO);
1779   }
1780
1781   /* Remove remaining fragments from queue, no need to transmit those
1782      any longer. */
1783   if (s->address->address_length == sizeof(struct IPv6UdpAddress))
1784   {
1785     udpw = plugin->ipv6_queue_head;
1786     while (NULL != udpw)
1787     {
1788       tmp = udpw->next;
1789       if ( (udpw->frag_ctx != NULL) &&
1790            (udpw->frag_ctx == frag_ctx) )
1791       {
1792         dequeue (plugin,
1793                  udpw);
1794         GNUNET_free (udpw);
1795       }
1796       udpw = tmp;
1797     }
1798   }
1799   if (s->address->address_length == sizeof(struct IPv4UdpAddress))
1800   {
1801     udpw = plugin->ipv4_queue_head;
1802     while (NULL != udpw)
1803     {
1804       tmp = udpw->next;
1805       if ( (NULL != udpw->frag_ctx) &&
1806            (udpw->frag_ctx == frag_ctx) )
1807       {
1808         dequeue (plugin,
1809                  udpw);
1810         GNUNET_free (udpw);
1811       }
1812       udpw = tmp;
1813     }
1814   }
1815   notify_session_monitor (s->plugin,
1816                           s,
1817                           GNUNET_TRANSPORT_SS_UPDATE);
1818   GNUNET_FRAGMENT_context_destroy (frag_ctx->frag,
1819                                    &s->last_expected_msg_delay,
1820                                    &s->last_expected_ack_delay);
1821   s->frag_ctx = NULL;
1822   GNUNET_free (frag_ctx);
1823 }
1824
1825
1826 /**
1827  * We are finished with a fragment in the message queue.
1828  * Notify the continuation and update statistics.
1829  *
1830  * @param cls the `struct Plugin *`
1831  * @param udpw the queue entry
1832  * @param result #GNUNET_OK on success, #GNUNET_SYSERR on failure
1833  */
1834 static void
1835 qc_fragment_sent (void *cls,
1836                   struct UDP_MessageWrapper *udpw,
1837                   int result)
1838 {
1839   struct Plugin *plugin = cls;
1840
1841   GNUNET_assert (NULL != udpw->frag_ctx);
1842   if (GNUNET_OK == result)
1843   {
1844     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1845                 "Fragment of message with %u bytes transmitted to %s\n",
1846                 (unsigned int) udpw->payload_size,
1847                 GNUNET_i2s (&udpw->session->target));
1848     GNUNET_FRAGMENT_context_transmission_done (udpw->frag_ctx->frag);
1849     GNUNET_STATISTICS_update (plugin->env->stats,
1850                               "# UDP, fragmented msgs, fragments, sent, success",
1851                               1,
1852                               GNUNET_NO);
1853     GNUNET_STATISTICS_update (plugin->env->stats,
1854                               "# UDP, fragmented msgs, fragments bytes, sent, success",
1855                               udpw->msg_size,
1856                               GNUNET_NO);
1857   }
1858   else
1859   {
1860     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1861                 "Failed to transmit fragment of message with %u bytes to %s\n",
1862                 (unsigned int) udpw->payload_size,
1863                 GNUNET_i2s (&udpw->session->target));
1864     fragmented_message_done (udpw->frag_ctx,
1865                              GNUNET_SYSERR);
1866     GNUNET_STATISTICS_update (plugin->env->stats,
1867                               "# UDP, fragmented msgs, fragments, sent, failure",
1868                               1,
1869                               GNUNET_NO);
1870     GNUNET_STATISTICS_update (plugin->env->stats,
1871                               "# UDP, fragmented msgs, fragments bytes, sent, failure",
1872                               udpw->msg_size,
1873                               GNUNET_NO);
1874   }
1875 }
1876
1877
1878 /**
1879  * Function that is called with messages created by the fragmentation
1880  * module.  In the case of the `proc` callback of the
1881  * #GNUNET_FRAGMENT_context_create() function, this function must
1882  * eventually call #GNUNET_FRAGMENT_context_transmission_done().
1883  *
1884  * @param cls closure, the `struct UDP_FragmentationContext`
1885  * @param msg the message that was created
1886  */
1887 static void
1888 enqueue_fragment (void *cls,
1889                   const struct GNUNET_MessageHeader *msg)
1890 {
1891   struct UDP_FragmentationContext *frag_ctx = cls;
1892   struct Plugin *plugin = frag_ctx->plugin;
1893   struct UDP_MessageWrapper *udpw;
1894   struct GNUNET_ATS_Session *session = frag_ctx->session;
1895   size_t msg_len = ntohs (msg->size);
1896
1897   LOG (GNUNET_ERROR_TYPE_DEBUG,
1898        "Enqueuing fragment with %u bytes\n",
1899        msg_len);
1900   udpw = GNUNET_malloc (sizeof (struct UDP_MessageWrapper) + msg_len);
1901   udpw->session = session;
1902   udpw->msg_buf = (char *) &udpw[1];
1903   udpw->msg_size = msg_len;
1904   udpw->payload_size = msg_len; /* FIXME: minus fragment overhead */
1905   udpw->timeout = frag_ctx->timeout;
1906   udpw->start_time = frag_ctx->start_time;
1907   udpw->transmission_time = frag_ctx->next_frag_time;
1908   frag_ctx->next_frag_time
1909     = GNUNET_TIME_absolute_add (frag_ctx->next_frag_time,
1910                                 frag_ctx->flow_delay_from_other_peer);
1911   udpw->frag_ctx = frag_ctx;
1912   udpw->qc = &qc_fragment_sent;
1913   udpw->qc_cls = plugin;
1914   GNUNET_memcpy (udpw->msg_buf,
1915                  msg,
1916                  msg_len);
1917   enqueue (plugin,
1918            udpw);
1919   if (session->address->address_length == sizeof (struct IPv4UdpAddress))
1920     schedule_select_v4 (plugin);
1921   else
1922     schedule_select_v6 (plugin);
1923 }
1924
1925
1926 /**
1927  * We are finished with a message from the message queue.
1928  * Notify the continuation and update statistics.
1929  *
1930  * @param cls the `struct Plugin *`
1931  * @param udpw the queue entry
1932  * @param result #GNUNET_OK on success, #GNUNET_SYSERR on failure
1933  */
1934 static void
1935 qc_message_sent (void *cls,
1936                  struct UDP_MessageWrapper *udpw,
1937                  int result)
1938 {
1939   struct Plugin *plugin = cls;
1940   size_t overhead;
1941   struct GNUNET_TIME_Relative delay;
1942
1943   if (udpw->msg_size >= udpw->payload_size)
1944     overhead = udpw->msg_size - udpw->payload_size;
1945   else
1946     overhead = udpw->msg_size;
1947
1948   if (NULL != udpw->cont)
1949   {
1950     delay = GNUNET_TIME_absolute_get_duration (udpw->start_time);
1951     if (delay.rel_value_us > GNUNET_CONSTANTS_LATENCY_WARN.rel_value_us)
1952     {
1953       LOG (GNUNET_ERROR_TYPE_WARNING,
1954            "Message sent via UDP with delay of %s\n",
1955            GNUNET_STRINGS_relative_time_to_string (delay,
1956                                                    GNUNET_YES));
1957     }
1958     else
1959     {
1960       LOG (GNUNET_ERROR_TYPE_DEBUG,
1961            "Message sent via UDP with delay of %s\n",
1962            GNUNET_STRINGS_relative_time_to_string (delay,
1963                                                    GNUNET_YES));
1964     }
1965     udpw->cont (udpw->cont_cls,
1966                 &udpw->session->target,
1967                 result,
1968                 udpw->payload_size,
1969                 overhead);
1970   }
1971   if (GNUNET_OK == result)
1972   {
1973     GNUNET_STATISTICS_update (plugin->env->stats,
1974                               "# UDP, unfragmented msgs, messages, sent, success",
1975                               1,
1976                               GNUNET_NO);
1977     GNUNET_STATISTICS_update (plugin->env->stats,
1978                               "# UDP, unfragmented msgs, bytes payload, sent, success",
1979                               udpw->payload_size,
1980                               GNUNET_NO);
1981     GNUNET_STATISTICS_update (plugin->env->stats,
1982                               "# UDP, unfragmented msgs, bytes overhead, sent, success",
1983                               overhead,
1984                               GNUNET_NO);
1985     GNUNET_STATISTICS_update (plugin->env->stats,
1986                               "# UDP, total, bytes overhead, sent",
1987                               overhead,
1988                               GNUNET_NO);
1989     GNUNET_STATISTICS_update (plugin->env->stats,
1990                               "# UDP, total, bytes payload, sent",
1991                               udpw->payload_size,
1992                               GNUNET_NO);
1993   }
1994   else
1995   {
1996     GNUNET_STATISTICS_update (plugin->env->stats,
1997                               "# UDP, unfragmented msgs, messages, sent, failure",
1998                               1,
1999                               GNUNET_NO);
2000     GNUNET_STATISTICS_update (plugin->env->stats,
2001                               "# UDP, unfragmented msgs, bytes payload, sent, failure",
2002                               udpw->payload_size,
2003                               GNUNET_NO);
2004     GNUNET_STATISTICS_update (plugin->env->stats,
2005                               "# UDP, unfragmented msgs, bytes overhead, sent, failure",
2006                               overhead,
2007                               GNUNET_NO);
2008   }
2009 }
2010
2011
2012 /**
2013  * Function that can be used by the transport service to transmit a
2014  * message using the plugin.  Note that in the case of a peer
2015  * disconnecting, the continuation MUST be called prior to the
2016  * disconnect notification itself.  This function will be called with
2017  * this peer's HELLO message to initiate a fresh connection to another
2018  * peer.
2019  *
2020  * @param cls closure
2021  * @param s which session must be used
2022  * @param msgbuf the message to transmit
2023  * @param msgbuf_size number of bytes in @a msgbuf
2024  * @param priority how important is the message (most plugins will
2025  *                 ignore message priority and just FIFO)
2026  * @param to how long to wait at most for the transmission (does not
2027  *                require plugins to discard the message after the timeout,
2028  *                just advisory for the desired delay; most plugins will ignore
2029  *                this as well)
2030  * @param cont continuation to call once the message has
2031  *        been transmitted (or if the transport is ready
2032  *        for the next transmission call; or if the
2033  *        peer disconnected...); can be NULL
2034  * @param cont_cls closure for @a cont
2035  * @return number of bytes used (on the physical network, with overheads);
2036  *         -1 on hard errors (i.e. address invalid); 0 is a legal value
2037  *         and does NOT mean that the message was not transmitted (DV)
2038  */
2039 static ssize_t
2040 udp_plugin_send (void *cls,
2041                  struct GNUNET_ATS_Session *s,
2042                  const char *msgbuf,
2043                  size_t msgbuf_size,
2044                  unsigned int priority,
2045                  struct GNUNET_TIME_Relative to,
2046                  GNUNET_TRANSPORT_TransmitContinuation cont,
2047                  void *cont_cls)
2048 {
2049   struct Plugin *plugin = cls;
2050   size_t udpmlen = msgbuf_size + sizeof(struct UDPMessage);
2051   struct UDP_FragmentationContext *frag_ctx;
2052   struct UDP_MessageWrapper *udpw;
2053   struct UDPMessage *udp;
2054   char mbuf[udpmlen] GNUNET_ALIGN;
2055   struct GNUNET_TIME_Relative latency;
2056
2057   if ( (sizeof(struct IPv6UdpAddress) == s->address->address_length) &&
2058        (NULL == plugin->sockv6) )
2059     return GNUNET_SYSERR;
2060   if ( (sizeof(struct IPv4UdpAddress) == s->address->address_length) &&
2061        (NULL == plugin->sockv4) )
2062     return GNUNET_SYSERR;
2063   if (udpmlen >= GNUNET_MAX_MESSAGE_SIZE)
2064   {
2065     GNUNET_break (0);
2066     return GNUNET_SYSERR;
2067   }
2068   if (GNUNET_YES !=
2069       GNUNET_CONTAINER_multipeermap_contains_value (plugin->sessions,
2070                                                     &s->target,
2071                                                     s))
2072   {
2073     GNUNET_break (0);
2074     return GNUNET_SYSERR;
2075   }
2076   LOG (GNUNET_ERROR_TYPE_DEBUG,
2077        "UDP transmits %u-byte message to `%s' using address `%s'\n",
2078        udpmlen,
2079        GNUNET_i2s (&s->target),
2080        udp_address_to_string (plugin,
2081                               s->address->address,
2082                               s->address->address_length));
2083
2084   udp = (struct UDPMessage *) mbuf;
2085   udp->header.size = htons (udpmlen);
2086   udp->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_MESSAGE);
2087   udp->reserved = htonl (0);
2088   udp->sender = *plugin->env->my_identity;
2089
2090   /* We do not update the session time out here!  Otherwise this
2091    * session will not timeout since we send keep alive before session
2092    * can timeout.
2093    *
2094    * For UDP we update session timeout only on receive, this will
2095    * cover keep alives, since remote peer will reply with keep alive
2096    * responses!
2097    */
2098   if (udpmlen <= UDP_MTU)
2099   {
2100     /* unfragmented message */
2101     udpw = GNUNET_malloc (sizeof (struct UDP_MessageWrapper) + udpmlen);
2102     udpw->session = s;
2103     udpw->msg_buf = (char *) &udpw[1];
2104     udpw->msg_size = udpmlen; /* message size with UDP overhead */
2105     udpw->payload_size = msgbuf_size; /* message size without UDP overhead */
2106     udpw->start_time = GNUNET_TIME_absolute_get ();
2107     udpw->timeout = GNUNET_TIME_relative_to_absolute (to);
2108     udpw->transmission_time = s->last_transmit_time;
2109     s->last_transmit_time
2110       = GNUNET_TIME_absolute_add (s->last_transmit_time,
2111                                   s->flow_delay_from_other_peer);
2112     udpw->cont = cont;
2113     udpw->cont_cls = cont_cls;
2114     udpw->frag_ctx = NULL;
2115     udpw->qc = &qc_message_sent;
2116     udpw->qc_cls = plugin;
2117     GNUNET_memcpy (udpw->msg_buf,
2118             udp,
2119             sizeof (struct UDPMessage));
2120     GNUNET_memcpy (&udpw->msg_buf[sizeof(struct UDPMessage)],
2121             msgbuf,
2122             msgbuf_size);
2123     enqueue (plugin,
2124              udpw);
2125     GNUNET_STATISTICS_update (plugin->env->stats,
2126                               "# UDP, unfragmented messages queued total",
2127                               1,
2128                               GNUNET_NO);
2129     GNUNET_STATISTICS_update (plugin->env->stats,
2130                               "# UDP, unfragmented bytes payload queued total",
2131                               msgbuf_size,
2132                               GNUNET_NO);
2133     if (s->address->address_length == sizeof (struct IPv4UdpAddress))
2134       schedule_select_v4 (plugin);
2135     else
2136       schedule_select_v6 (plugin);
2137   }
2138   else
2139   {
2140     /* fragmented message */
2141     if (NULL != s->frag_ctx)
2142       return GNUNET_SYSERR;
2143     GNUNET_memcpy (&udp[1],
2144             msgbuf,
2145             msgbuf_size);
2146     frag_ctx = GNUNET_new (struct UDP_FragmentationContext);
2147     frag_ctx->plugin = plugin;
2148     frag_ctx->session = s;
2149     frag_ctx->cont = cont;
2150     frag_ctx->cont_cls = cont_cls;
2151     frag_ctx->start_time = GNUNET_TIME_absolute_get ();
2152     frag_ctx->next_frag_time = s->last_transmit_time;
2153     frag_ctx->flow_delay_from_other_peer
2154       = GNUNET_TIME_relative_divide (s->flow_delay_from_other_peer,
2155                                      1 + (msgbuf_size /
2156                                           UDP_MTU));
2157     frag_ctx->timeout = GNUNET_TIME_relative_to_absolute (to);
2158     frag_ctx->payload_size = msgbuf_size; /* unfragmented message size without UDP overhead */
2159     frag_ctx->on_wire_size = 0; /* bytes with UDP and fragmentation overhead */
2160     frag_ctx->frag = GNUNET_FRAGMENT_context_create (plugin->env->stats,
2161                                                      UDP_MTU,
2162                                                      &plugin->tracker,
2163                                                      s->last_expected_msg_delay,
2164                                                      s->last_expected_ack_delay,
2165                                                      &udp->header,
2166                                                      &enqueue_fragment,
2167                                                      frag_ctx);
2168     s->frag_ctx = frag_ctx;
2169     s->last_transmit_time = frag_ctx->next_frag_time;
2170     latency = GNUNET_TIME_absolute_get_remaining (s->last_transmit_time);
2171     if (latency.rel_value_us > GNUNET_CONSTANTS_LATENCY_WARN.rel_value_us)
2172       LOG (GNUNET_ERROR_TYPE_WARNING,
2173            "Enqueued fragments will take %s for transmission to %s (queue size: %u)\n",
2174            GNUNET_STRINGS_relative_time_to_string (latency,
2175                                                    GNUNET_YES),
2176            GNUNET_i2s (&s->target),
2177            (unsigned int) s->msgs_in_queue);
2178     else
2179       LOG (GNUNET_ERROR_TYPE_DEBUG,
2180            "Enqueued fragments will take %s for transmission to %s (queue size: %u)\n",
2181            GNUNET_STRINGS_relative_time_to_string (latency,
2182                                                    GNUNET_YES),
2183            GNUNET_i2s (&s->target),
2184            (unsigned int) s->msgs_in_queue);
2185
2186     GNUNET_STATISTICS_update (plugin->env->stats,
2187                               "# UDP, fragmented messages active",
2188                               1,
2189                               GNUNET_NO);
2190     GNUNET_STATISTICS_update (plugin->env->stats,
2191                               "# UDP, fragmented messages, total",
2192                               1,
2193                               GNUNET_NO);
2194     GNUNET_STATISTICS_update (plugin->env->stats,
2195                               "# UDP, fragmented bytes (payload)",
2196                               frag_ctx->payload_size,
2197                               GNUNET_NO);
2198   }
2199   notify_session_monitor (s->plugin,
2200                           s,
2201                           GNUNET_TRANSPORT_SS_UPDATE);
2202   return udpmlen;
2203 }
2204
2205
2206 /* ********************** Receiving ********************** */
2207
2208
2209 /**
2210  * Closure for #find_receive_context().
2211  */
2212 struct FindReceiveContext
2213 {
2214   /**
2215    * Where to store the result.
2216    */
2217   struct DefragContext *rc;
2218
2219   /**
2220    * Session associated with this context.
2221    */
2222   struct GNUNET_ATS_Session *session;
2223
2224   /**
2225    * Address to find.
2226    */
2227   const union UdpAddress *udp_addr;
2228
2229   /**
2230    * Number of bytes in @e udp_addr.
2231    */
2232   size_t udp_addr_len;
2233
2234 };
2235
2236
2237 /**
2238  * Scan the heap for a receive context with the given address.
2239  *
2240  * @param cls the `struct FindReceiveContext`
2241  * @param node internal node of the heap
2242  * @param element value stored at the node (a `struct ReceiveContext`)
2243  * @param cost cost associated with the node
2244  * @return #GNUNET_YES if we should continue to iterate,
2245  *         #GNUNET_NO if not.
2246  */
2247 static int
2248 find_receive_context (void *cls,
2249                       struct GNUNET_CONTAINER_HeapNode *node,
2250                       void *element,
2251                       GNUNET_CONTAINER_HeapCostType cost)
2252 {
2253   struct FindReceiveContext *frc = cls;
2254   struct DefragContext *e = element;
2255
2256   if ( (frc->udp_addr_len == e->udp_addr_len) &&
2257        (0 == memcmp (frc->udp_addr,
2258                      e->udp_addr,
2259                      frc->udp_addr_len)) )
2260   {
2261     frc->rc = e;
2262     return GNUNET_NO;
2263   }
2264   return GNUNET_YES;
2265 }
2266
2267
2268 /**
2269  * Functions with this signature are called whenever we need to close
2270  * a session due to a disconnect or failure to establish a connection.
2271  *
2272  * @param cls closure with the `struct Plugin`
2273  * @param s session to close down
2274  * @return #GNUNET_OK on success
2275  */
2276 static int
2277 udp_disconnect_session (void *cls,
2278                         struct GNUNET_ATS_Session *s)
2279 {
2280   struct Plugin *plugin = cls;
2281   struct UDP_MessageWrapper *udpw;
2282   struct UDP_MessageWrapper *next;
2283   struct FindReceiveContext frc;
2284
2285   GNUNET_assert (GNUNET_YES != s->in_destroy);
2286   LOG (GNUNET_ERROR_TYPE_DEBUG,
2287        "Session %p to peer `%s' at address %s ended\n",
2288        s,
2289        GNUNET_i2s (&s->target),
2290        udp_address_to_string (plugin,
2291                               s->address->address,
2292                               s->address->address_length));
2293   if (NULL != s->timeout_task)
2294   {
2295     GNUNET_SCHEDULER_cancel (s->timeout_task);
2296     s->timeout_task = NULL;
2297   }
2298   if (NULL != s->frag_ctx)
2299   {
2300     /* Remove fragmented message due to disconnect */
2301     fragmented_message_done (s->frag_ctx,
2302                              GNUNET_SYSERR);
2303   }
2304   GNUNET_assert (GNUNET_YES ==
2305                  GNUNET_CONTAINER_multipeermap_remove (plugin->sessions,
2306                                                        &s->target,
2307                                                        s));
2308   frc.rc = NULL;
2309   frc.udp_addr = s->address->address;
2310   frc.udp_addr_len = s->address->address_length;
2311   /* Lookup existing receive context for this address */
2312   if (NULL != plugin->defrag_ctxs)
2313   {
2314     GNUNET_CONTAINER_heap_iterate (plugin->defrag_ctxs,
2315                                    &find_receive_context,
2316                                    &frc);
2317     if (NULL != frc.rc)
2318     {
2319       struct DefragContext *d_ctx = frc.rc;
2320
2321       GNUNET_CONTAINER_heap_remove_node (d_ctx->hnode);
2322       GNUNET_DEFRAGMENT_context_destroy (d_ctx->defrag);
2323       GNUNET_free (d_ctx);
2324     }
2325   }
2326   s->in_destroy = GNUNET_YES;
2327   next = plugin->ipv4_queue_head;
2328   while (NULL != (udpw = next))
2329   {
2330     next = udpw->next;
2331     if (udpw->session == s)
2332     {
2333       dequeue (plugin,
2334                udpw);
2335       udpw->qc (udpw->qc_cls,
2336                 udpw,
2337                 GNUNET_SYSERR);
2338       GNUNET_free (udpw);
2339     }
2340   }
2341   next = plugin->ipv6_queue_head;
2342   while (NULL != (udpw = next))
2343   {
2344     next = udpw->next;
2345     if (udpw->session == s)
2346     {
2347       dequeue (plugin,
2348                udpw);
2349       udpw->qc (udpw->qc_cls,
2350                 udpw,
2351                 GNUNET_SYSERR);
2352       GNUNET_free (udpw);
2353     }
2354   }
2355   if ( (NULL != s->frag_ctx) &&
2356        (NULL != s->frag_ctx->cont) )
2357   {
2358     /* The 'frag_ctx' itself will be freed in #free_session() a bit
2359        later, as it might be in use right now */
2360     LOG (GNUNET_ERROR_TYPE_DEBUG,
2361          "Calling continuation for fragemented message to `%s' with result SYSERR\n",
2362          GNUNET_i2s (&s->target));
2363     s->frag_ctx->cont (s->frag_ctx->cont_cls,
2364                        &s->target,
2365                        GNUNET_SYSERR,
2366                        s->frag_ctx->payload_size,
2367                        s->frag_ctx->on_wire_size);
2368   }
2369   notify_session_monitor (s->plugin,
2370                           s,
2371                           GNUNET_TRANSPORT_SS_DONE);
2372   plugin->env->session_end (plugin->env->cls,
2373                             s->address,
2374                             s);
2375   GNUNET_STATISTICS_set (plugin->env->stats,
2376                          "# UDP sessions active",
2377                          GNUNET_CONTAINER_multipeermap_size (plugin->sessions),
2378                          GNUNET_NO);
2379   if (0 == s->rc)
2380     free_session (s);
2381   return GNUNET_OK;
2382 }
2383
2384
2385 /**
2386  * Handle a #GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_ACK message.
2387  *
2388  * @param plugin the UDP plugin
2389  * @param msg the (presumed) UDP ACK message
2390  * @param udp_addr sender address
2391  * @param udp_addr_len number of bytes in @a udp_addr
2392  */
2393 static void
2394 read_process_ack (struct Plugin *plugin,
2395                   const struct GNUNET_MessageHeader *msg,
2396                   const union UdpAddress *udp_addr,
2397                   socklen_t udp_addr_len)
2398 {
2399   const struct GNUNET_MessageHeader *ack;
2400   const struct UDP_ACK_Message *udp_ack;
2401   struct GNUNET_HELLO_Address *address;
2402   struct GNUNET_ATS_Session *s;
2403   struct GNUNET_TIME_Relative flow_delay;
2404
2405   /* check message format */
2406   if (ntohs (msg->size)
2407       < sizeof(struct UDP_ACK_Message) + sizeof(struct GNUNET_MessageHeader))
2408   {
2409     GNUNET_break_op (0);
2410     return;
2411   }
2412   udp_ack = (const struct UDP_ACK_Message *) msg;
2413   ack = (const struct GNUNET_MessageHeader *) &udp_ack[1];
2414   if (ntohs (ack->size) != ntohs (msg->size) - sizeof(struct UDP_ACK_Message))
2415   {
2416     GNUNET_break_op(0);
2417     return;
2418   }
2419
2420   /* Locate session */
2421   address = GNUNET_HELLO_address_allocate (&udp_ack->sender,
2422                                            PLUGIN_NAME,
2423                                            udp_addr,
2424                                            udp_addr_len,
2425                                            GNUNET_HELLO_ADDRESS_INFO_NONE);
2426   s = udp_plugin_lookup_session (plugin,
2427                                  address);
2428   if (NULL == s)
2429   {
2430     LOG (GNUNET_ERROR_TYPE_WARNING,
2431          "UDP session of address %s for ACK not found\n",
2432          udp_address_to_string (plugin,
2433                                 address->address,
2434                                 address->address_length));
2435     GNUNET_HELLO_address_free (address);
2436     return;
2437   }
2438   if (NULL == s->frag_ctx)
2439   {
2440     LOG (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
2441          "Fragmentation context of address %s for ACK (%s) not found\n",
2442          udp_address_to_string (plugin,
2443                                 address->address,
2444                                 address->address_length),
2445          GNUNET_FRAGMENT_print_ack (ack));
2446     GNUNET_HELLO_address_free (address);
2447     return;
2448   }
2449   GNUNET_HELLO_address_free (address);
2450
2451   /* evaluate flow delay: how long should we wait between messages? */
2452   if (UINT32_MAX == ntohl (udp_ack->delay))
2453   {
2454     /* Other peer asked for us to terminate the session */
2455     LOG (GNUNET_ERROR_TYPE_INFO,
2456          "Asked to disconnect UDP session of %s\n",
2457          GNUNET_i2s (&udp_ack->sender));
2458     udp_disconnect_session (plugin,
2459                             s);
2460     return;
2461   }
2462   flow_delay.rel_value_us = (uint64_t) ntohl (udp_ack->delay);
2463   if (flow_delay.rel_value_us > GNUNET_CONSTANTS_LATENCY_WARN.rel_value_us)
2464     LOG (GNUNET_ERROR_TYPE_WARNING,
2465          "We received a sending delay of %s for %s\n",
2466          GNUNET_STRINGS_relative_time_to_string (flow_delay,
2467                                                  GNUNET_YES),
2468          GNUNET_i2s (&udp_ack->sender));
2469   else
2470     LOG (GNUNET_ERROR_TYPE_DEBUG,
2471          "We received a sending delay of %s for %s\n",
2472          GNUNET_STRINGS_relative_time_to_string (flow_delay,
2473                                                  GNUNET_YES),
2474          GNUNET_i2s (&udp_ack->sender));
2475   /* Flow delay is for the reassembled packet, however, our delay
2476      is per packet, so we need to adjust: */
2477   s->flow_delay_from_other_peer = flow_delay;
2478
2479   /* Handle ACK */
2480   if (GNUNET_OK !=
2481       GNUNET_FRAGMENT_process_ack (s->frag_ctx->frag,
2482                                    ack))
2483   {
2484     LOG (GNUNET_ERROR_TYPE_DEBUG,
2485          "UDP processes %u-byte acknowledgement from `%s' at `%s'\n",
2486          (unsigned int) ntohs (msg->size),
2487          GNUNET_i2s (&udp_ack->sender),
2488          udp_address_to_string (plugin,
2489                                 udp_addr,
2490                                 udp_addr_len));
2491     /* Expect more ACKs to arrive */
2492     return;
2493   }
2494
2495   /* Remove fragmented message after successful sending */
2496   LOG (GNUNET_ERROR_TYPE_DEBUG,
2497        "Message from %s at %s full ACK'ed\n",
2498        GNUNET_i2s (&udp_ack->sender),
2499        udp_address_to_string (plugin,
2500                               udp_addr,
2501                               udp_addr_len));
2502   fragmented_message_done (s->frag_ctx,
2503                            GNUNET_OK);
2504 }
2505
2506
2507 /**
2508  * Message tokenizer has broken up an incomming message. Pass it on
2509  * to the service.
2510  *
2511  * @param cls the `struct GNUNET_ATS_Session *`
2512  * @param hdr the actual message
2513  * @return #GNUNET_OK (always)
2514  */
2515 static int
2516 process_inbound_tokenized_messages (void *cls,
2517                                     const struct GNUNET_MessageHeader *hdr)
2518 {
2519   struct GNUNET_ATS_Session *session = cls;
2520   struct Plugin *plugin = session->plugin;
2521
2522   if (GNUNET_YES == session->in_destroy)
2523     return GNUNET_OK;
2524   reschedule_session_timeout (session);
2525   session->flow_delay_for_other_peer
2526     = plugin->env->receive (plugin->env->cls,
2527                             session->address,
2528                             session,
2529                             hdr);
2530   return GNUNET_OK;
2531 }
2532
2533
2534 /**
2535  * Destroy a session, plugin is being unloaded.
2536  *
2537  * @param cls the `struct Plugin`
2538  * @param key hash of public key of target peer
2539  * @param value a `struct PeerSession *` to clean up
2540  * @return #GNUNET_OK (continue to iterate)
2541  */
2542 static int
2543 disconnect_and_free_it (void *cls,
2544                         const struct GNUNET_PeerIdentity *key,
2545                         void *value)
2546 {
2547   struct Plugin *plugin = cls;
2548
2549   udp_disconnect_session (plugin,
2550                           value);
2551   return GNUNET_OK;
2552 }
2553
2554
2555 /**
2556  * Disconnect from a remote node.  Clean up session if we have one for
2557  * this peer.
2558  *
2559  * @param cls closure for this call (should be handle to Plugin)
2560  * @param target the peeridentity of the peer to disconnect
2561  * @return #GNUNET_OK on success, #GNUNET_SYSERR if the operation failed
2562  */
2563 static void
2564 udp_disconnect (void *cls,
2565                 const struct GNUNET_PeerIdentity *target)
2566 {
2567   struct Plugin *plugin = cls;
2568
2569   LOG (GNUNET_ERROR_TYPE_DEBUG,
2570        "Disconnecting from peer `%s'\n",
2571        GNUNET_i2s (target));
2572   GNUNET_CONTAINER_multipeermap_get_multiple (plugin->sessions,
2573                                               target,
2574                                               &disconnect_and_free_it,
2575                                               plugin);
2576 }
2577
2578
2579 /**
2580  * Session was idle, so disconnect it.
2581  *
2582  * @param cls the `struct GNUNET_ATS_Session` to time out
2583  */
2584 static void
2585 session_timeout (void *cls)
2586 {
2587   struct GNUNET_ATS_Session *s = cls;
2588   struct Plugin *plugin = s->plugin;
2589   struct GNUNET_TIME_Relative left;
2590
2591   s->timeout_task = NULL;
2592   left = GNUNET_TIME_absolute_get_remaining (s->timeout);
2593   if (left.rel_value_us > 0)
2594   {
2595     /* not actually our turn yet, but let's at least update
2596        the monitor, it may think we're about to die ... */
2597     notify_session_monitor (s->plugin,
2598                             s,
2599                             GNUNET_TRANSPORT_SS_UPDATE);
2600     s->timeout_task = GNUNET_SCHEDULER_add_delayed (left,
2601                                                     &session_timeout,
2602                                                     s);
2603     return;
2604   }
2605   LOG (GNUNET_ERROR_TYPE_DEBUG,
2606        "Session %p was idle for %s, disconnecting\n",
2607        s,
2608        GNUNET_STRINGS_relative_time_to_string (UDP_SESSION_TIME_OUT,
2609                                                GNUNET_YES));
2610   /* call session destroy function */
2611   udp_disconnect_session (plugin,
2612                           s);
2613 }
2614
2615
2616 /**
2617  * Allocate a new session for the given endpoint address.
2618  * Note that this function does not inform the service
2619  * of the new session, this is the responsibility of the
2620  * caller (if needed).
2621  *
2622  * @param cls the `struct Plugin`
2623  * @param address address of the other peer to use
2624  * @param network_type network type the address belongs to
2625  * @return NULL on error, otherwise session handle
2626  */
2627 static struct GNUNET_ATS_Session *
2628 udp_plugin_create_session (void *cls,
2629                            const struct GNUNET_HELLO_Address *address,
2630                            enum GNUNET_ATS_Network_Type network_type)
2631 {
2632   struct Plugin *plugin = cls;
2633   struct GNUNET_ATS_Session *s;
2634
2635   s = GNUNET_new (struct GNUNET_ATS_Session);
2636   s->mst = GNUNET_MST_create (&process_inbound_tokenized_messages,
2637                               s);
2638   s->plugin = plugin;
2639   s->address = GNUNET_HELLO_address_copy (address);
2640   s->target = address->peer;
2641   s->last_transmit_time = GNUNET_TIME_absolute_get ();
2642   s->last_expected_ack_delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
2643                                                               250);
2644   s->last_expected_msg_delay = GNUNET_TIME_UNIT_MILLISECONDS;
2645   s->flow_delay_from_other_peer = GNUNET_TIME_UNIT_ZERO;
2646   s->flow_delay_for_other_peer = GNUNET_TIME_UNIT_ZERO;
2647   s->timeout = GNUNET_TIME_relative_to_absolute (UDP_SESSION_TIME_OUT);
2648   s->timeout_task = GNUNET_SCHEDULER_add_delayed (UDP_SESSION_TIME_OUT,
2649                                                   &session_timeout,
2650                                                   s);
2651   s->scope = network_type;
2652
2653   LOG (GNUNET_ERROR_TYPE_DEBUG,
2654        "Creating new session %p for peer `%s' address `%s'\n",
2655        s,
2656        GNUNET_i2s (&address->peer),
2657        udp_address_to_string (plugin,
2658                               address->address,
2659                               address->address_length));
2660   GNUNET_assert (GNUNET_OK ==
2661                  GNUNET_CONTAINER_multipeermap_put (plugin->sessions,
2662                                                     &s->target,
2663                                                     s,
2664                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
2665   GNUNET_STATISTICS_set (plugin->env->stats,
2666                          "# UDP sessions active",
2667                          GNUNET_CONTAINER_multipeermap_size (plugin->sessions),
2668                          GNUNET_NO);
2669   notify_session_monitor (plugin,
2670                           s,
2671                           GNUNET_TRANSPORT_SS_INIT);
2672   return s;
2673 }
2674
2675
2676 /**
2677  * Creates a new outbound session the transport service will use to
2678  * send data to the peer.
2679  *
2680  * @param cls the `struct Plugin *`
2681  * @param address the address
2682  * @return the session or NULL of max connections exceeded
2683  */
2684 static struct GNUNET_ATS_Session *
2685 udp_plugin_get_session (void *cls,
2686                         const struct GNUNET_HELLO_Address *address)
2687 {
2688   struct Plugin *plugin = cls;
2689   struct GNUNET_ATS_Session *s;
2690   enum GNUNET_ATS_Network_Type network_type = GNUNET_ATS_NET_UNSPECIFIED;
2691   const struct IPv4UdpAddress *udp_v4;
2692   const struct IPv6UdpAddress *udp_v6;
2693
2694   if (NULL == address)
2695   {
2696     GNUNET_break (0);
2697     return NULL;
2698   }
2699   if ( (address->address_length != sizeof(struct IPv4UdpAddress)) &&
2700        (address->address_length != sizeof(struct IPv6UdpAddress)) )
2701   {
2702     GNUNET_break_op (0);
2703     return NULL;
2704   }
2705   if (NULL != (s = udp_plugin_lookup_session (cls,
2706                                               address)))
2707     return s;
2708
2709   /* need to create new session */
2710   if (sizeof (struct IPv4UdpAddress) == address->address_length)
2711   {
2712     struct sockaddr_in v4;
2713
2714     udp_v4 = (const struct IPv4UdpAddress *) address->address;
2715     memset (&v4, '\0', sizeof (v4));
2716     v4.sin_family = AF_INET;
2717 #if HAVE_SOCKADDR_IN_SIN_LEN
2718     v4.sin_len = sizeof (struct sockaddr_in);
2719 #endif
2720     v4.sin_port = udp_v4->u4_port;
2721     v4.sin_addr.s_addr = udp_v4->ipv4_addr;
2722     network_type = plugin->env->get_address_type (plugin->env->cls,
2723                                                   (const struct sockaddr *) &v4,
2724                                                   sizeof (v4));
2725   }
2726   if (sizeof (struct IPv6UdpAddress) == address->address_length)
2727   {
2728     struct sockaddr_in6 v6;
2729
2730     udp_v6 = (const struct IPv6UdpAddress *) address->address;
2731     memset (&v6, '\0', sizeof (v6));
2732     v6.sin6_family = AF_INET6;
2733 #if HAVE_SOCKADDR_IN_SIN_LEN
2734     v6.sin6_len = sizeof (struct sockaddr_in6);
2735 #endif
2736     v6.sin6_port = udp_v6->u6_port;
2737     v6.sin6_addr = udp_v6->ipv6_addr;
2738     network_type = plugin->env->get_address_type (plugin->env->cls,
2739                                                   (const struct sockaddr *) &v6,
2740                                                   sizeof (v6));
2741   }
2742   GNUNET_break (GNUNET_ATS_NET_UNSPECIFIED != network_type);
2743   return udp_plugin_create_session (cls,
2744                                     address,
2745                                     network_type);
2746 }
2747
2748
2749 /**
2750  * We've received a UDP Message.  Process it (pass contents to main service).
2751  *
2752  * @param plugin plugin context
2753  * @param msg the message
2754  * @param udp_addr sender address
2755  * @param udp_addr_len number of bytes in @a udp_addr
2756  * @param network_type network type the address belongs to
2757  */
2758 static void
2759 process_udp_message (struct Plugin *plugin,
2760                      const struct UDPMessage *msg,
2761                      const union UdpAddress *udp_addr,
2762                      size_t udp_addr_len,
2763                      enum GNUNET_ATS_Network_Type network_type)
2764 {
2765   struct GNUNET_ATS_Session *s;
2766   struct GNUNET_HELLO_Address *address;
2767
2768   GNUNET_break (GNUNET_ATS_NET_UNSPECIFIED != network_type);
2769   if (0 != ntohl (msg->reserved))
2770   {
2771     GNUNET_break_op(0);
2772     return;
2773   }
2774   if (ntohs (msg->header.size)
2775       < sizeof(struct GNUNET_MessageHeader) + sizeof(struct UDPMessage))
2776   {
2777     GNUNET_break_op(0);
2778     return;
2779   }
2780
2781   address = GNUNET_HELLO_address_allocate (&msg->sender,
2782                                            PLUGIN_NAME,
2783                                            udp_addr,
2784                                            udp_addr_len,
2785                                            GNUNET_HELLO_ADDRESS_INFO_NONE);
2786   if (NULL ==
2787       (s = udp_plugin_lookup_session (plugin,
2788                                       address)))
2789   {
2790     s = udp_plugin_create_session (plugin,
2791                                    address,
2792                                    network_type);
2793     plugin->env->session_start (plugin->env->cls,
2794                                 address,
2795                                 s,
2796                                 s->scope);
2797     notify_session_monitor (plugin,
2798                             s,
2799                             GNUNET_TRANSPORT_SS_UP);
2800   }
2801   GNUNET_free (address);
2802
2803   s->rc++;
2804   GNUNET_MST_from_buffer (s->mst,
2805                           (const char *) &msg[1],
2806                           ntohs (msg->header.size) - sizeof(struct UDPMessage),
2807                           GNUNET_YES,
2808                           GNUNET_NO);
2809   s->rc--;
2810   if ( (0 == s->rc) &&
2811        (GNUNET_YES == s->in_destroy) )
2812     free_session (s);
2813 }
2814
2815
2816 /**
2817  * Process a defragmented message.
2818  *
2819  * @param cls the `struct DefragContext *`
2820  * @param msg the message
2821  */
2822 static void
2823 fragment_msg_proc (void *cls,
2824                    const struct GNUNET_MessageHeader *msg)
2825 {
2826   struct DefragContext *dc = cls;
2827   const struct UDPMessage *um;
2828
2829   if (ntohs (msg->type) != GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_MESSAGE)
2830   {
2831     GNUNET_break_op (0);
2832     return;
2833   }
2834   if (ntohs (msg->size) < sizeof(struct UDPMessage))
2835   {
2836     GNUNET_break_op (0);
2837     return;
2838   }
2839   um = (const struct UDPMessage *) msg;
2840   dc->sender = um->sender;
2841   dc->have_sender = GNUNET_YES;
2842   process_udp_message (dc->plugin,
2843                        um,
2844                        dc->udp_addr,
2845                        dc->udp_addr_len,
2846                        dc->network_type);
2847 }
2848
2849
2850 /**
2851  * We finished sending an acknowledgement.  Update
2852  * statistics.
2853  *
2854  * @param cls the `struct Plugin`
2855  * @param udpw message queue entry of the ACK
2856  * @param result #GNUNET_OK if the transmission worked,
2857  *               #GNUNET_SYSERR if we failed to send the ACK
2858  */
2859 static void
2860 ack_message_sent (void *cls,
2861                   struct UDP_MessageWrapper *udpw,
2862                   int result)
2863 {
2864   struct Plugin *plugin = cls;
2865
2866   if (GNUNET_OK == result)
2867   {
2868     GNUNET_STATISTICS_update (plugin->env->stats,
2869                               "# UDP, ACK messages sent",
2870                               1,
2871                               GNUNET_NO);
2872   }
2873   else
2874   {
2875     GNUNET_STATISTICS_update (plugin->env->stats,
2876                               "# UDP, ACK transmissions failed",
2877                               1,
2878                               GNUNET_NO);
2879   }
2880 }
2881
2882
2883 /**
2884  * Transmit an acknowledgement.
2885  *
2886  * @param cls the `struct DefragContext *`
2887  * @param id message ID (unused)
2888  * @param msg ack to transmit
2889  */
2890 static void
2891 ack_proc (void *cls,
2892           uint32_t id,
2893           const struct GNUNET_MessageHeader *msg)
2894 {
2895   struct DefragContext *rc = cls;
2896   struct Plugin *plugin = rc->plugin;
2897   size_t msize = sizeof(struct UDP_ACK_Message) + ntohs (msg->size);
2898   struct UDP_ACK_Message *udp_ack;
2899   uint32_t delay;
2900   struct UDP_MessageWrapper *udpw;
2901   struct GNUNET_ATS_Session *s;
2902   struct GNUNET_HELLO_Address *address;
2903
2904   if (GNUNET_NO == rc->have_sender)
2905   {
2906     /* tried to defragment but never succeeded, hence will not ACK */
2907     /* This can happen if we just lost msgs */
2908     GNUNET_STATISTICS_update (plugin->env->stats,
2909                               "# UDP, fragments discarded without ACK",
2910                               1,
2911                               GNUNET_NO);
2912     return;
2913   }
2914   address = GNUNET_HELLO_address_allocate (&rc->sender,
2915                                            PLUGIN_NAME,
2916                                            rc->udp_addr,
2917                                            rc->udp_addr_len,
2918                                            GNUNET_HELLO_ADDRESS_INFO_NONE);
2919   s = udp_plugin_lookup_session (plugin,
2920                                  address);
2921   GNUNET_HELLO_address_free (address);
2922   if (NULL == s)
2923   {
2924     LOG (GNUNET_ERROR_TYPE_ERROR,
2925          "Trying to transmit ACK to peer `%s' but no session found!\n",
2926          udp_address_to_string (plugin,
2927                                 rc->udp_addr,
2928                                 rc->udp_addr_len));
2929     GNUNET_CONTAINER_heap_remove_node (rc->hnode);
2930     GNUNET_DEFRAGMENT_context_destroy (rc->defrag);
2931     GNUNET_free (rc);
2932     GNUNET_STATISTICS_update (plugin->env->stats,
2933                               "# UDP, ACK transmissions failed",
2934                               1,
2935                               GNUNET_NO);
2936     return;
2937   }
2938   if (GNUNET_TIME_UNIT_FOREVER_REL.rel_value_us ==
2939       s->flow_delay_for_other_peer.rel_value_us)
2940     delay = UINT32_MAX;
2941   else if (s->flow_delay_for_other_peer.rel_value_us < UINT32_MAX)
2942     delay = s->flow_delay_for_other_peer.rel_value_us;
2943   else
2944     delay = UINT32_MAX - 1; /* largest value we can communicate */
2945   LOG (GNUNET_ERROR_TYPE_DEBUG,
2946        "Sending ACK to `%s' including delay of %s\n",
2947        udp_address_to_string (plugin,
2948                               rc->udp_addr,
2949                               rc->udp_addr_len),
2950        GNUNET_STRINGS_relative_time_to_string (s->flow_delay_for_other_peer,
2951                                                GNUNET_YES));
2952   udpw = GNUNET_malloc (sizeof (struct UDP_MessageWrapper) + msize);
2953   udpw->msg_size = msize;
2954   udpw->payload_size = 0;
2955   udpw->session = s;
2956   udpw->start_time = GNUNET_TIME_absolute_get ();
2957   udpw->timeout = GNUNET_TIME_UNIT_FOREVER_ABS;
2958   udpw->msg_buf = (char *) &udpw[1];
2959   udpw->qc = &ack_message_sent;
2960   udpw->qc_cls = plugin;
2961   udp_ack = (struct UDP_ACK_Message *) udpw->msg_buf;
2962   udp_ack->header.size = htons ((uint16_t) msize);
2963   udp_ack->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_ACK);
2964   udp_ack->delay = htonl (delay);
2965   udp_ack->sender = *plugin->env->my_identity;
2966   GNUNET_memcpy (&udp_ack[1],
2967           msg,
2968           ntohs (msg->size));
2969   enqueue (plugin,
2970            udpw);
2971   notify_session_monitor (plugin,
2972                           s,
2973                           GNUNET_TRANSPORT_SS_UPDATE);
2974   if (s->address->address_length == sizeof (struct IPv4UdpAddress))
2975     schedule_select_v4 (plugin);
2976   else
2977     schedule_select_v6 (plugin);
2978 }
2979
2980
2981 /**
2982  * We received a fragment, process it.
2983  *
2984  * @param plugin our plugin
2985  * @param msg a message of type #GNUNET_MESSAGE_TYPE_FRAGMENT
2986  * @param udp_addr sender address
2987  * @param udp_addr_len number of bytes in @a udp_addr
2988  * @param network_type network type the address belongs to
2989  */
2990 static void
2991 read_process_fragment (struct Plugin *plugin,
2992                        const struct GNUNET_MessageHeader *msg,
2993                        const union UdpAddress *udp_addr,
2994                        size_t udp_addr_len,
2995                        enum GNUNET_ATS_Network_Type network_type)
2996 {
2997   struct DefragContext *d_ctx;
2998   struct GNUNET_TIME_Absolute now;
2999   struct FindReceiveContext frc;
3000
3001   frc.rc = NULL;
3002   frc.udp_addr = udp_addr;
3003   frc.udp_addr_len = udp_addr_len;
3004
3005   /* Lookup existing receive context for this address */
3006   GNUNET_CONTAINER_heap_iterate (plugin->defrag_ctxs,
3007                                  &find_receive_context,
3008                                  &frc);
3009   now = GNUNET_TIME_absolute_get ();
3010   d_ctx = frc.rc;
3011
3012   if (NULL == d_ctx)
3013   {
3014     /* Create a new defragmentation context */
3015     d_ctx = GNUNET_malloc (sizeof (struct DefragContext) + udp_addr_len);
3016     GNUNET_memcpy (&d_ctx[1],
3017             udp_addr,
3018             udp_addr_len);
3019     d_ctx->udp_addr = (const union UdpAddress *) &d_ctx[1];
3020     d_ctx->udp_addr_len = udp_addr_len;
3021     d_ctx->network_type = network_type;
3022     d_ctx->plugin = plugin;
3023     d_ctx->defrag = GNUNET_DEFRAGMENT_context_create (plugin->env->stats,
3024                                                       UDP_MTU,
3025                                                       UDP_MAX_MESSAGES_IN_DEFRAG,
3026                                                       d_ctx,
3027                                                       &fragment_msg_proc,
3028                                                       &ack_proc);
3029     d_ctx->hnode = GNUNET_CONTAINER_heap_insert (plugin->defrag_ctxs,
3030                                                  d_ctx,
3031                                                  (GNUNET_CONTAINER_HeapCostType) now.abs_value_us);
3032     LOG (GNUNET_ERROR_TYPE_DEBUG,
3033          "Created new defragmentation context for %u-byte fragment from `%s'\n",
3034          (unsigned int) ntohs (msg->size),
3035          udp_address_to_string (plugin,
3036                                 udp_addr,
3037                                 udp_addr_len));
3038   }
3039   else
3040   {
3041     LOG (GNUNET_ERROR_TYPE_DEBUG,
3042          "Found existing defragmentation context for %u-byte fragment from `%s'\n",
3043          (unsigned int) ntohs (msg->size),
3044          udp_address_to_string (plugin,
3045                                 udp_addr,
3046                                 udp_addr_len));
3047   }
3048
3049   if (GNUNET_OK ==
3050       GNUNET_DEFRAGMENT_process_fragment (d_ctx->defrag,
3051                                           msg))
3052   {
3053     /* keep this 'rc' from expiring */
3054     GNUNET_CONTAINER_heap_update_cost (d_ctx->hnode,
3055                                        (GNUNET_CONTAINER_HeapCostType) now.abs_value_us);
3056   }
3057   if (GNUNET_CONTAINER_heap_get_size (plugin->defrag_ctxs) >
3058       UDP_MAX_SENDER_ADDRESSES_WITH_DEFRAG)
3059   {
3060     /* remove 'rc' that was inactive the longest */
3061     d_ctx = GNUNET_CONTAINER_heap_remove_root (plugin->defrag_ctxs);
3062     GNUNET_assert (NULL != d_ctx);
3063     GNUNET_DEFRAGMENT_context_destroy (d_ctx->defrag);
3064     GNUNET_free (d_ctx);
3065     GNUNET_STATISTICS_update (plugin->env->stats,
3066                               "# UDP, Defragmentations aborted",
3067                               1,
3068                               GNUNET_NO);
3069   }
3070 }
3071
3072
3073 /**
3074  * Read and process a message from the given socket.
3075  *
3076  * @param plugin the overall plugin
3077  * @param rsock socket to read from
3078  */
3079 static void
3080 udp_select_read (struct Plugin *plugin,
3081                  struct GNUNET_NETWORK_Handle *rsock)
3082 {
3083   socklen_t fromlen;
3084   struct sockaddr_storage addr;
3085   char buf[65536] GNUNET_ALIGN;
3086   ssize_t size;
3087   const struct GNUNET_MessageHeader *msg;
3088   struct IPv4UdpAddress v4;
3089   struct IPv6UdpAddress v6;
3090   const struct sockaddr *sa;
3091   const struct sockaddr_in *sa4;
3092   const struct sockaddr_in6 *sa6;
3093   const union UdpAddress *int_addr;
3094   size_t int_addr_len;
3095   enum GNUNET_ATS_Network_Type network_type;
3096
3097   fromlen = sizeof (addr);
3098   memset (&addr,
3099           0,
3100           sizeof(addr));
3101   size = GNUNET_NETWORK_socket_recvfrom (rsock,
3102                                          buf,
3103                                          sizeof (buf),
3104                                          (struct sockaddr *) &addr,
3105                                          &fromlen);
3106   sa = (const struct sockaddr *) &addr;
3107 #if MINGW
3108   /* On SOCK_DGRAM UDP sockets recvfrom might fail with a
3109    * WSAECONNRESET error to indicate that previous sendto() (yes, sendto!)
3110    * on this socket has failed.
3111    * Quote from MSDN:
3112    *   WSAECONNRESET - The virtual circuit was reset by the remote side
3113    *   executing a hard or abortive close. The application should close
3114    *   the socket; it is no longer usable. On a UDP-datagram socket this
3115    *   error indicates a previous send operation resulted in an ICMP Port
3116    *   Unreachable message.
3117    */
3118   if ( (-1 == size) &&
3119        (ECONNRESET == errno) )
3120     return;
3121 #endif
3122   if (-1 == size)
3123   {
3124     LOG (GNUNET_ERROR_TYPE_DEBUG,
3125          "UDP failed to receive data: %s\n",
3126          STRERROR (errno));
3127     /* Connection failure or something. Not a protocol violation. */
3128     return;
3129   }
3130
3131   /* Check if this is a STUN packet */
3132   if (GNUNET_NO !=
3133       GNUNET_NAT_stun_handle_packet (plugin->nat,
3134                                      (const struct sockaddr *) &addr,
3135                                      fromlen,
3136                                      buf,
3137                                      size))
3138     return; /* was STUN, do not process further */
3139
3140   if (size < sizeof(struct GNUNET_MessageHeader))
3141   {
3142     LOG (GNUNET_ERROR_TYPE_WARNING,
3143          "UDP got %u bytes from %s, which is not enough for a GNUnet message header\n",
3144          (unsigned int ) size,
3145          GNUNET_a2s (sa,
3146                      fromlen));
3147     /* _MAY_ be a connection failure (got partial message) */
3148     /* But it _MAY_ also be that the other side uses non-GNUnet protocol. */
3149     GNUNET_break_op (0);
3150     return;
3151   }
3152
3153   msg = (const struct GNUNET_MessageHeader *) buf;
3154   LOG (GNUNET_ERROR_TYPE_DEBUG,
3155        "UDP received %u-byte message from `%s' type %u\n",
3156        (unsigned int) size,
3157        GNUNET_a2s (sa,
3158                    fromlen),
3159        ntohs (msg->type));
3160   if (size != ntohs (msg->size))
3161   {
3162     LOG (GNUNET_ERROR_TYPE_WARNING,
3163          "UDP malformed message (size %u) header from %s\n",
3164          (unsigned int) size,
3165          GNUNET_a2s (sa,
3166                      fromlen));
3167     GNUNET_break_op (0);
3168     return;
3169   }
3170   GNUNET_STATISTICS_update (plugin->env->stats,
3171                             "# UDP, total bytes received",
3172                             size,
3173                             GNUNET_NO);
3174   network_type = plugin->env->get_address_type (plugin->env->cls,
3175                                                 sa,
3176                                                 fromlen);
3177   switch (sa->sa_family)
3178   {
3179   case AF_INET:
3180     sa4 = (const struct sockaddr_in *) &addr;
3181     v4.options = 0;
3182     v4.ipv4_addr = sa4->sin_addr.s_addr;
3183     v4.u4_port = sa4->sin_port;
3184     int_addr = (union UdpAddress *) &v4;
3185     int_addr_len = sizeof (v4);
3186     break;
3187   case AF_INET6:
3188     sa6 = (const struct sockaddr_in6 *) &addr;
3189     v6.options = 0;
3190     v6.ipv6_addr = sa6->sin6_addr;
3191     v6.u6_port = sa6->sin6_port;
3192     int_addr = (union UdpAddress *) &v6;
3193     int_addr_len = sizeof (v6);
3194     break;
3195   default:
3196     GNUNET_break (0);
3197     return;
3198   }
3199
3200   switch (ntohs (msg->type))
3201   {
3202   case GNUNET_MESSAGE_TYPE_TRANSPORT_BROADCAST_BEACON:
3203     if (GNUNET_YES == plugin->enable_broadcasting_receiving)
3204       udp_broadcast_receive (plugin,
3205                              buf,
3206                              size,
3207                              int_addr,
3208                              int_addr_len,
3209                              network_type);
3210     return;
3211   case GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_MESSAGE:
3212     if (ntohs (msg->size) < sizeof(struct UDPMessage))
3213     {
3214       GNUNET_break_op(0);
3215       return;
3216     }
3217     process_udp_message (plugin,
3218                          (const struct UDPMessage *) msg,
3219                          int_addr,
3220                          int_addr_len,
3221                          network_type);
3222     return;
3223   case GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_ACK:
3224     read_process_ack (plugin,
3225                       msg,
3226                       int_addr,
3227                       int_addr_len);
3228     return;
3229   case GNUNET_MESSAGE_TYPE_FRAGMENT:
3230     read_process_fragment (plugin,
3231                            msg,
3232                            int_addr,
3233                            int_addr_len,
3234                            network_type);
3235     return;
3236   default:
3237     GNUNET_break_op(0);
3238     return;
3239   }
3240 }
3241
3242
3243 /**
3244  * Removes messages from the transmission queue that have
3245  * timed out, and then selects a message that should be
3246  * transmitted next.
3247  *
3248  * @param plugin the UDP plugin
3249  * @param sock which socket should we process the queue for (v4 or v6)
3250  * @return message selected for transmission, or NULL for none
3251  */
3252 static struct UDP_MessageWrapper *
3253 remove_timeout_messages_and_select (struct Plugin *plugin,
3254                                     struct GNUNET_NETWORK_Handle *sock)
3255 {
3256   struct UDP_MessageWrapper *udpw;
3257   struct GNUNET_TIME_Relative remaining;
3258   struct GNUNET_ATS_Session *session;
3259   int removed;
3260
3261   removed = GNUNET_NO;
3262   udpw = (sock == plugin->sockv4)
3263     ? plugin->ipv4_queue_head
3264     : plugin->ipv6_queue_head;
3265   while (NULL != udpw)
3266   {
3267     session = udpw->session;
3268     /* Find messages with timeout */
3269     remaining = GNUNET_TIME_absolute_get_remaining (udpw->timeout);
3270     if (GNUNET_TIME_UNIT_ZERO.rel_value_us == remaining.rel_value_us)
3271     {
3272       /* Message timed out */
3273       removed = GNUNET_YES;
3274       dequeue (plugin,
3275                udpw);
3276       udpw->qc (udpw->qc_cls,
3277                 udpw,
3278                 GNUNET_SYSERR);
3279       GNUNET_free (udpw);
3280
3281       if (sock == plugin->sockv4)
3282       {
3283         udpw = plugin->ipv4_queue_head;
3284       }
3285       else if (sock == plugin->sockv6)
3286       {
3287         udpw = plugin->ipv6_queue_head;
3288       }
3289       else
3290       {
3291         GNUNET_break (0); /* should never happen */
3292         udpw = NULL;
3293       }
3294       GNUNET_STATISTICS_update (plugin->env->stats,
3295                                 "# messages discarded due to timeout",
3296                                 1,
3297                                 GNUNET_NO);
3298     }
3299     else
3300     {
3301       /* Message did not time out, check transmission time */
3302       remaining = GNUNET_TIME_absolute_get_remaining (udpw->transmission_time);
3303       if (0 == remaining.rel_value_us)
3304       {
3305         /* this message is not delayed */
3306         LOG (GNUNET_ERROR_TYPE_DEBUG,
3307              "Message for peer `%s' (%u bytes) is not delayed \n",
3308              GNUNET_i2s (&udpw->session->target),
3309              udpw->payload_size);
3310         break; /* Found message to send, break */
3311       }
3312       else
3313       {
3314         /* Message is delayed, try next */
3315         LOG (GNUNET_ERROR_TYPE_DEBUG,
3316              "Message for peer `%s' (%u bytes) is delayed for %s\n",
3317              GNUNET_i2s (&udpw->session->target),
3318              udpw->payload_size,
3319              GNUNET_STRINGS_relative_time_to_string (remaining,
3320                                                      GNUNET_YES));
3321         udpw = udpw->next;
3322       }
3323     }
3324   }
3325   if (GNUNET_YES == removed)
3326     notify_session_monitor (session->plugin,
3327                             session,
3328                             GNUNET_TRANSPORT_SS_UPDATE);
3329   return udpw;
3330 }
3331
3332
3333 /**
3334  * We failed to transmit a message via UDP. Generate
3335  * a descriptive error message.
3336  *
3337  * @param plugin our plugin
3338  * @param sa target address we were trying to reach
3339  * @param slen number of bytes in @a sa
3340  * @param error the errno value returned from the sendto() call
3341  */
3342 static void
3343 analyze_send_error (struct Plugin *plugin,
3344                     const struct sockaddr *sa,
3345                     socklen_t slen,
3346                     int error)
3347 {
3348   enum GNUNET_ATS_Network_Type type;
3349
3350   type = plugin->env->get_address_type (plugin->env->cls,
3351                                         sa,
3352                                         slen);
3353   if ( ( (GNUNET_ATS_NET_LAN == type) ||
3354          (GNUNET_ATS_NET_WAN == type) ) &&
3355        ( (ENETUNREACH == errno) ||
3356          (ENETDOWN == errno) ) )
3357   {
3358     if (slen == sizeof (struct sockaddr_in))
3359     {
3360       /* IPv4: "Network unreachable" or "Network down"
3361        *
3362        * This indicates we do not have connectivity
3363        */
3364       LOG (GNUNET_ERROR_TYPE_WARNING | GNUNET_ERROR_TYPE_BULK,
3365            _("UDP could not transmit message to `%s': "
3366              "Network seems down, please check your network configuration\n"),
3367            GNUNET_a2s (sa,
3368                        slen));
3369     }
3370     if (slen == sizeof (struct sockaddr_in6))
3371     {
3372       /* IPv6: "Network unreachable" or "Network down"
3373        *
3374        * This indicates that this system is IPv6 enabled, but does not
3375        * have a valid global IPv6 address assigned or we do not have
3376        * connectivity
3377        */
3378       LOG (GNUNET_ERROR_TYPE_WARNING | GNUNET_ERROR_TYPE_BULK,
3379            _("UDP could not transmit IPv6 message! "
3380              "Please check your network configuration and disable IPv6 if your "
3381              "connection does not have a global IPv6 address\n"));
3382     }
3383   }
3384   else
3385   {
3386     LOG (GNUNET_ERROR_TYPE_WARNING,
3387          "UDP could not transmit message to `%s': `%s'\n",
3388          GNUNET_a2s (sa,
3389                      slen),
3390          STRERROR (error));
3391   }
3392 }
3393
3394
3395 /**
3396  * It is time to try to transmit a UDP message.  Select one
3397  * and send.
3398  *
3399  * @param plugin the plugin
3400  * @param sock which socket (v4/v6) to send on
3401  */
3402 static void
3403 udp_select_send (struct Plugin *plugin,
3404                  struct GNUNET_NETWORK_Handle *sock)
3405 {
3406   ssize_t sent;
3407   socklen_t slen;
3408   const struct sockaddr *a;
3409   const struct IPv4UdpAddress *u4;
3410   struct sockaddr_in a4;
3411   const struct IPv6UdpAddress *u6;
3412   struct sockaddr_in6 a6;
3413   struct UDP_MessageWrapper *udpw;
3414
3415   /* Find message(s) to send */
3416   while (NULL != (udpw = remove_timeout_messages_and_select (plugin,
3417                                                              sock)))
3418   {
3419     if (sizeof (struct IPv4UdpAddress) == udpw->session->address->address_length)
3420     {
3421       u4 = udpw->session->address->address;
3422       memset (&a4,
3423               0,
3424               sizeof(a4));
3425       a4.sin_family = AF_INET;
3426 #if HAVE_SOCKADDR_IN_SIN_LEN
3427       a4.sin_len = sizeof (a4);
3428 #endif
3429       a4.sin_port = u4->u4_port;
3430       a4.sin_addr.s_addr = u4->ipv4_addr;
3431       a = (const struct sockaddr *) &a4;
3432       slen = sizeof (a4);
3433     }
3434     else if (sizeof (struct IPv6UdpAddress) == udpw->session->address->address_length)
3435     {
3436       u6 = udpw->session->address->address;
3437       memset (&a6,
3438               0,
3439               sizeof(a6));
3440       a6.sin6_family = AF_INET6;
3441 #if HAVE_SOCKADDR_IN_SIN_LEN
3442       a6.sin6_len = sizeof (a6);
3443 #endif
3444       a6.sin6_port = u6->u6_port;
3445       a6.sin6_addr = u6->ipv6_addr;
3446       a = (const struct sockaddr *) &a6;
3447       slen = sizeof (a6);
3448     }
3449     else
3450     {
3451       GNUNET_break (0);
3452       dequeue (plugin,
3453                udpw);
3454       udpw->qc (udpw->qc_cls,
3455                 udpw,
3456                 GNUNET_SYSERR);
3457       notify_session_monitor (plugin,
3458                               udpw->session,
3459                               GNUNET_TRANSPORT_SS_UPDATE);
3460       GNUNET_free (udpw);
3461       continue;
3462     }
3463     sent = GNUNET_NETWORK_socket_sendto (sock,
3464                                          udpw->msg_buf,
3465                                          udpw->msg_size,
3466                                          a,
3467                                          slen);
3468     udpw->session->last_transmit_time
3469       = GNUNET_TIME_absolute_max (GNUNET_TIME_absolute_get (),
3470                                   udpw->session->last_transmit_time);
3471     dequeue (plugin,
3472              udpw);
3473     if (GNUNET_SYSERR == sent)
3474     {
3475       /* Failure */
3476       analyze_send_error (plugin,
3477                           a,
3478                           slen,
3479                           errno);
3480       udpw->qc (udpw->qc_cls,
3481                 udpw,
3482                 GNUNET_SYSERR);
3483       GNUNET_STATISTICS_update (plugin->env->stats,
3484                                 "# UDP, total, bytes, sent, failure",
3485                                 sent,
3486                                 GNUNET_NO);
3487       GNUNET_STATISTICS_update (plugin->env->stats,
3488                                 "# UDP, total, messages, sent, failure",
3489                                 1,
3490                                 GNUNET_NO);
3491     }
3492     else
3493     {
3494       /* Success */
3495       LOG (GNUNET_ERROR_TYPE_DEBUG,
3496            "UDP transmitted %u-byte message to  `%s' `%s' (%d: %s)\n",
3497            (unsigned int) (udpw->msg_size),
3498            GNUNET_i2s (&udpw->session->target),
3499            GNUNET_a2s (a,
3500                        slen),
3501            (int ) sent,
3502            (sent < 0) ? STRERROR (errno) : "ok");
3503       GNUNET_STATISTICS_update (plugin->env->stats,
3504                                 "# UDP, total, bytes, sent, success",
3505                                 sent,
3506                                 GNUNET_NO);
3507       GNUNET_STATISTICS_update (plugin->env->stats,
3508                                 "# UDP, total, messages, sent, success",
3509                                 1,
3510                                 GNUNET_NO);
3511       if (NULL != udpw->frag_ctx)
3512         udpw->frag_ctx->on_wire_size += udpw->msg_size;
3513       udpw->qc (udpw->qc_cls,
3514                 udpw,
3515                 GNUNET_OK);
3516     }
3517     notify_session_monitor (plugin,
3518                             udpw->session,
3519                             GNUNET_TRANSPORT_SS_UPDATE);
3520     GNUNET_free (udpw);
3521   }
3522 }
3523
3524
3525 /* ***************** Event loop (part 2) *************** */
3526
3527
3528 /**
3529  * We have been notified that our readset has something to read.  We don't
3530  * know which socket needs to be read, so we have to check each one
3531  * Then reschedule this function to be called again once more is available.
3532  *
3533  * @param cls the plugin handle
3534  */
3535 static void
3536 udp_plugin_select_v4 (void *cls)
3537 {
3538   struct Plugin *plugin = cls;
3539   const struct GNUNET_SCHEDULER_TaskContext *tc;
3540
3541   plugin->select_task_v4 = NULL;
3542   if (NULL == plugin->sockv4)
3543     return;
3544   tc = GNUNET_SCHEDULER_get_task_context ();
3545   if ((0 != (tc->reason & GNUNET_SCHEDULER_REASON_READ_READY)) &&
3546       (GNUNET_NETWORK_fdset_isset (tc->read_ready,
3547                                    plugin->sockv4)))
3548     udp_select_read (plugin,
3549                      plugin->sockv4);
3550   udp_select_send (plugin,
3551                    plugin->sockv4);
3552   schedule_select_v4 (plugin);
3553 }
3554
3555
3556 /**
3557  * We have been notified that our readset has something to read.  We don't
3558  * know which socket needs to be read, so we have to check each one
3559  * Then reschedule this function to be called again once more is available.
3560  *
3561  * @param cls the plugin handle
3562  */
3563 static void
3564 udp_plugin_select_v6 (void *cls)
3565 {
3566   struct Plugin *plugin = cls;
3567   const struct GNUNET_SCHEDULER_TaskContext *tc;
3568
3569   plugin->select_task_v6 = NULL;
3570   if (NULL == plugin->sockv6)
3571     return;
3572   tc = GNUNET_SCHEDULER_get_task_context ();
3573   if ( (0 != (tc->reason & GNUNET_SCHEDULER_REASON_READ_READY)) &&
3574        (GNUNET_NETWORK_fdset_isset (tc->read_ready,
3575                                     plugin->sockv6)) )
3576     udp_select_read (plugin,
3577                      plugin->sockv6);
3578
3579   udp_select_send (plugin,
3580                    plugin->sockv6);
3581   schedule_select_v6 (plugin);
3582 }
3583
3584
3585 /* ******************* Initialization *************** */
3586
3587
3588 /**
3589  * Setup the UDP sockets (for IPv4 and IPv6) for the plugin.
3590  *
3591  * @param plugin the plugin to initialize
3592  * @param bind_v6 IPv6 address to bind to (can be NULL, for 'any')
3593  * @param bind_v4 IPv4 address to bind to (can be NULL, for 'any')
3594  * @return number of sockets that were successfully bound
3595  */
3596 static unsigned int
3597 setup_sockets (struct Plugin *plugin,
3598                const struct sockaddr_in6 *bind_v6,
3599                const struct sockaddr_in *bind_v4)
3600 {
3601   int tries;
3602   unsigned int sockets_created = 0;
3603   struct sockaddr_in6 server_addrv6;
3604   struct sockaddr_in server_addrv4;
3605   const struct sockaddr *server_addr;
3606   const struct sockaddr *addrs[2];
3607   socklen_t addrlens[2];
3608   socklen_t addrlen;
3609   int eno;
3610
3611   /* Create IPv6 socket */
3612   eno = EINVAL;
3613   if (GNUNET_YES == plugin->enable_ipv6)
3614   {
3615     plugin->sockv6 = GNUNET_NETWORK_socket_create (PF_INET6,
3616                                                    SOCK_DGRAM,
3617                                                    0);
3618     if (NULL == plugin->sockv6)
3619     {
3620       LOG (GNUNET_ERROR_TYPE_INFO,
3621            _("Disabling IPv6 since it is not supported on this system!\n"));
3622       plugin->enable_ipv6 = GNUNET_NO;
3623     }
3624     else
3625     {
3626       memset (&server_addrv6,
3627               0,
3628               sizeof(struct sockaddr_in6));
3629 #if HAVE_SOCKADDR_IN_SIN_LEN
3630       server_addrv6.sin6_len = sizeof (struct sockaddr_in6);
3631 #endif
3632       server_addrv6.sin6_family = AF_INET6;
3633       if (NULL != bind_v6)
3634         server_addrv6.sin6_addr = bind_v6->sin6_addr;
3635       else
3636         server_addrv6.sin6_addr = in6addr_any;
3637
3638       if (0 == plugin->port) /* autodetect */
3639         server_addrv6.sin6_port
3640           = htons (GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_STRONG,
3641                                              33537)
3642                    + 32000);
3643       else
3644         server_addrv6.sin6_port = htons (plugin->port);
3645       addrlen = sizeof (struct sockaddr_in6);
3646       server_addr = (const struct sockaddr *) &server_addrv6;
3647
3648       tries = 0;
3649       while (tries < 10)
3650       {
3651         LOG(GNUNET_ERROR_TYPE_DEBUG,
3652             "Binding to IPv6 `%s'\n",
3653             GNUNET_a2s (server_addr,
3654                         addrlen));
3655         /* binding */
3656         if (GNUNET_OK ==
3657             GNUNET_NETWORK_socket_bind (plugin->sockv6,
3658                                         server_addr,
3659                                         addrlen))
3660           break;
3661         eno = errno;
3662         if (0 != plugin->port)
3663         {
3664           tries = 10; /* fail immediately */
3665           break; /* bind failed on specific port */
3666         }
3667         /* autodetect */
3668         server_addrv6.sin6_port
3669           = htons (GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_STRONG,
3670                                              33537)
3671                    + 32000);
3672         tries++;
3673       }
3674       if (tries >= 10)
3675       {
3676         GNUNET_NETWORK_socket_close (plugin->sockv6);
3677         plugin->enable_ipv6 = GNUNET_NO;
3678         plugin->sockv6 = NULL;
3679       }
3680       else
3681       {
3682         plugin->port = ntohs (server_addrv6.sin6_port);
3683       }
3684       if (NULL != plugin->sockv6)
3685       {
3686         LOG (GNUNET_ERROR_TYPE_DEBUG,
3687              "IPv6 UDP socket created listinging at %s\n",
3688              GNUNET_a2s (server_addr,
3689                          addrlen));
3690         addrs[sockets_created] = server_addr;
3691         addrlens[sockets_created] = addrlen;
3692         sockets_created++;
3693       }
3694       else
3695       {
3696         LOG (GNUNET_ERROR_TYPE_WARNING,
3697              _("Failed to bind UDP socket to %s: %s\n"),
3698              GNUNET_a2s (server_addr,
3699                          addrlen),
3700              STRERROR (eno));
3701       }
3702     }
3703   }
3704
3705   /* Create IPv4 socket */
3706   eno = EINVAL;
3707   plugin->sockv4 = GNUNET_NETWORK_socket_create (PF_INET,
3708                                                  SOCK_DGRAM,
3709                                                  0);
3710   if (NULL == plugin->sockv4)
3711   {
3712     GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING,
3713                          "socket");
3714     LOG (GNUNET_ERROR_TYPE_INFO,
3715          _("Disabling IPv4 since it is not supported on this system!\n"));
3716     plugin->enable_ipv4 = GNUNET_NO;
3717   }
3718   else
3719   {
3720     memset (&server_addrv4,
3721             0,
3722             sizeof(struct sockaddr_in));
3723 #if HAVE_SOCKADDR_IN_SIN_LEN
3724     server_addrv4.sin_len = sizeof (struct sockaddr_in);
3725 #endif
3726     server_addrv4.sin_family = AF_INET;
3727     if (NULL != bind_v4)
3728       server_addrv4.sin_addr = bind_v4->sin_addr;
3729     else
3730       server_addrv4.sin_addr.s_addr = INADDR_ANY;
3731
3732     if (0 == plugin->port)
3733       /* autodetect */
3734       server_addrv4.sin_port
3735         = htons (GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_STRONG,
3736                                            33537)
3737                  + 32000);
3738     else
3739       server_addrv4.sin_port = htons (plugin->port);
3740
3741     addrlen = sizeof (struct sockaddr_in);
3742     server_addr = (const struct sockaddr *) &server_addrv4;
3743
3744     tries = 0;
3745     while (tries < 10)
3746     {
3747       LOG (GNUNET_ERROR_TYPE_DEBUG,
3748            "Binding to IPv4 `%s'\n",
3749            GNUNET_a2s (server_addr,
3750                        addrlen));
3751
3752       /* binding */
3753       if (GNUNET_OK ==
3754           GNUNET_NETWORK_socket_bind (plugin->sockv4,
3755                                       server_addr,
3756                                       addrlen))
3757         break;
3758       eno = errno;
3759       if (0 != plugin->port)
3760       {
3761         tries = 10; /* fail */
3762         break; /* bind failed on specific port */
3763       }
3764
3765       /* autodetect */
3766       server_addrv4.sin_port
3767         = htons (GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_STRONG,
3768                                            33537)
3769                  + 32000);
3770       tries++;
3771     }
3772     if (tries >= 10)
3773     {
3774       GNUNET_NETWORK_socket_close (plugin->sockv4);
3775       plugin->enable_ipv4 = GNUNET_NO;
3776       plugin->sockv4 = NULL;
3777     }
3778     else
3779     {
3780       plugin->port = ntohs (server_addrv4.sin_port);
3781     }
3782
3783     if (NULL != plugin->sockv4)
3784     {
3785       LOG (GNUNET_ERROR_TYPE_DEBUG,
3786            "IPv4 socket created on port %s\n",
3787            GNUNET_a2s (server_addr,
3788                        addrlen));
3789       addrs[sockets_created] = server_addr;
3790       addrlens[sockets_created] = addrlen;
3791       sockets_created++;
3792     }
3793     else
3794     {
3795       LOG (GNUNET_ERROR_TYPE_ERROR,
3796            _("Failed to bind UDP socket to %s: %s\n"),
3797            GNUNET_a2s (server_addr,
3798                        addrlen),
3799            STRERROR (eno));
3800     }
3801   }
3802
3803   if (0 == sockets_created)
3804   {
3805     LOG (GNUNET_ERROR_TYPE_WARNING,
3806          _("Failed to open UDP sockets\n"));
3807     return 0; /* No sockets created, return */
3808   }
3809   schedule_select_v4 (plugin);
3810   schedule_select_v6 (plugin);
3811   plugin->nat = GNUNET_NAT_register (plugin->env->cfg,
3812                                      "transport-udp",
3813                                      IPPROTO_UDP,
3814                                      sockets_created,
3815                                      addrs,
3816                                      addrlens,
3817                                      &udp_nat_port_map_callback,
3818                                      NULL,
3819                                      plugin);
3820   return sockets_created;
3821 }
3822
3823
3824 /**
3825  * The exported method. Makes the core api available via a global and
3826  * returns the udp transport API.
3827  *
3828  * @param cls our `struct GNUNET_TRANSPORT_PluginEnvironment`
3829  * @return our `struct GNUNET_TRANSPORT_PluginFunctions`
3830  */
3831 void *
3832 libgnunet_plugin_transport_udp_init (void *cls)
3833 {
3834   struct GNUNET_TRANSPORT_PluginEnvironment *env = cls;
3835   struct GNUNET_TRANSPORT_PluginFunctions *api;
3836   struct Plugin *p;
3837   unsigned long long port;
3838   unsigned long long aport;
3839   unsigned long long udp_max_bps;
3840   int enable_v6;
3841   int enable_broadcasting;
3842   int enable_broadcasting_recv;
3843   char *bind4_address;
3844   char *bind6_address;
3845   struct GNUNET_TIME_Relative interval;
3846   struct sockaddr_in server_addrv4;
3847   struct sockaddr_in6 server_addrv6;
3848   unsigned int res;
3849   int have_bind4;
3850   int have_bind6;
3851
3852   if (NULL == env->receive)
3853   {
3854     /* run in 'stub' mode (i.e. as part of gnunet-peerinfo), don't fully
3855      initialze the plugin or the API */
3856     api = GNUNET_new (struct GNUNET_TRANSPORT_PluginFunctions);
3857     api->cls = NULL;
3858     api->address_pretty_printer = &udp_plugin_address_pretty_printer;
3859     api->address_to_string = &udp_address_to_string;
3860     api->string_to_address = &udp_string_to_address;
3861     return api;
3862   }
3863
3864   /* Get port number: port == 0 : autodetect a port,
3865    * > 0 : use this port, not given : 2086 default */
3866   if (GNUNET_OK !=
3867       GNUNET_CONFIGURATION_get_value_number (env->cfg,
3868                                              "transport-udp",
3869                                              "PORT",
3870                                              &port))
3871     port = 2086;
3872   if (port > 65535)
3873   {
3874     GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
3875                                "transport-udp",
3876                                "PORT",
3877                                _("must be in [0,65535]"));
3878     return NULL;
3879   }
3880   if (GNUNET_OK !=
3881       GNUNET_CONFIGURATION_get_value_number (env->cfg,
3882                                              "transport-udp",
3883                                              "ADVERTISED_PORT",
3884                                              &aport))
3885     aport = port;
3886   if (aport > 65535)
3887   {
3888     GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
3889                                "transport-udp",
3890                                "ADVERTISED_PORT",
3891                                _("must be in [0,65535]"));
3892     return NULL;
3893   }
3894
3895   if (GNUNET_YES ==
3896       GNUNET_CONFIGURATION_get_value_yesno (env->cfg,
3897                                             "nat",
3898                                             "DISABLEV6"))
3899     enable_v6 = GNUNET_NO;
3900   else
3901     enable_v6 = GNUNET_YES;
3902
3903   have_bind4 = GNUNET_NO;
3904   memset (&server_addrv4,
3905           0,
3906           sizeof (server_addrv4));
3907   if (GNUNET_YES ==
3908       GNUNET_CONFIGURATION_get_value_string (env->cfg,
3909                                              "transport-udp",
3910                                              "BINDTO",
3911                                              &bind4_address))
3912   {
3913     LOG (GNUNET_ERROR_TYPE_DEBUG,
3914          "Binding UDP plugin to specific address: `%s'\n",
3915          bind4_address);
3916     if (1 != inet_pton (AF_INET,
3917                         bind4_address,
3918                         &server_addrv4.sin_addr))
3919     {
3920       GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
3921                                  "transport-udp",
3922                                  "BINDTO",
3923                                  _("must be valid IPv4 address"));
3924       GNUNET_free (bind4_address);
3925       return NULL;
3926     }
3927     have_bind4 = GNUNET_YES;
3928   }
3929   GNUNET_free_non_null (bind4_address);
3930   have_bind6 = GNUNET_NO;
3931   memset (&server_addrv6,
3932           0,
3933           sizeof (server_addrv6));
3934   if (GNUNET_YES ==
3935       GNUNET_CONFIGURATION_get_value_string (env->cfg,
3936                                              "transport-udp",
3937                                              "BINDTO6",
3938                                              &bind6_address))
3939   {
3940     LOG (GNUNET_ERROR_TYPE_DEBUG,
3941          "Binding udp plugin to specific address: `%s'\n",
3942          bind6_address);
3943     if (1 != inet_pton (AF_INET6,
3944                         bind6_address,
3945                         &server_addrv6.sin6_addr))
3946     {
3947       GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_ERROR,
3948                                  "transport-udp",
3949                                  "BINDTO6",
3950                                  _("must be valid IPv6 address"));
3951       GNUNET_free (bind6_address);
3952       return NULL;
3953     }
3954     have_bind6 = GNUNET_YES;
3955   }
3956   GNUNET_free_non_null (bind6_address);
3957
3958   enable_broadcasting = GNUNET_CONFIGURATION_get_value_yesno (env->cfg,
3959                                                               "transport-udp",
3960                                                               "BROADCAST");
3961   if (enable_broadcasting == GNUNET_SYSERR)
3962     enable_broadcasting = GNUNET_NO;
3963
3964   enable_broadcasting_recv = GNUNET_CONFIGURATION_get_value_yesno (env->cfg,
3965                                                                    "transport-udp",
3966                                                                    "BROADCAST_RECEIVE");
3967   if (enable_broadcasting_recv == GNUNET_SYSERR)
3968     enable_broadcasting_recv = GNUNET_YES;
3969
3970   if (GNUNET_SYSERR ==
3971       GNUNET_CONFIGURATION_get_value_time (env->cfg,
3972                                            "transport-udp",
3973                                            "BROADCAST_INTERVAL",
3974                                            &interval))
3975   {
3976     interval = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
3977                                               10);
3978   }
3979   if (GNUNET_OK !=
3980       GNUNET_CONFIGURATION_get_value_number (env->cfg,
3981                                              "transport-udp",
3982                                              "MAX_BPS",
3983                                              &udp_max_bps))
3984   {
3985     /* 50 MB/s == infinity for practical purposes */
3986     udp_max_bps = 1024 * 1024 * 50;
3987   }
3988
3989   p = GNUNET_new (struct Plugin);
3990   p->port = port;
3991   p->aport = aport;
3992   p->broadcast_interval = interval;
3993   p->enable_ipv6 = enable_v6;
3994   p->enable_ipv4 = GNUNET_YES; /* default */
3995   p->enable_broadcasting = enable_broadcasting;
3996   p->enable_broadcasting_receiving = enable_broadcasting_recv;
3997   p->env = env;
3998   p->sessions = GNUNET_CONTAINER_multipeermap_create (16,
3999                                                       GNUNET_NO);
4000   p->defrag_ctxs = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
4001   GNUNET_BANDWIDTH_tracker_init (&p->tracker,
4002                                  NULL,
4003                                  NULL,
4004                                  GNUNET_BANDWIDTH_value_init ((uint32_t) udp_max_bps),
4005                                  30);
4006   res = setup_sockets (p,
4007                        (GNUNET_YES == have_bind6) ? &server_addrv6 : NULL,
4008                        (GNUNET_YES == have_bind4) ? &server_addrv4 : NULL);
4009   if ( (0 == res) ||
4010        ( (NULL == p->sockv4) &&
4011          (NULL == p->sockv6) ) )
4012   {
4013     LOG (GNUNET_ERROR_TYPE_ERROR,
4014         _("Failed to create UDP network sockets\n"));
4015     GNUNET_CONTAINER_multipeermap_destroy (p->sessions);
4016     GNUNET_CONTAINER_heap_destroy (p->defrag_ctxs);
4017     if (NULL != p->nat)
4018       GNUNET_NAT_unregister (p->nat);
4019     GNUNET_free (p);
4020     return NULL;
4021   }
4022
4023   /* Setup broadcasting and receiving beacons */
4024   setup_broadcast (p,
4025                    &server_addrv6,
4026                    &server_addrv4);
4027
4028   api = GNUNET_new (struct GNUNET_TRANSPORT_PluginFunctions);
4029   api->cls = p;
4030   api->disconnect_session = &udp_disconnect_session;
4031   api->query_keepalive_factor = &udp_query_keepalive_factor;
4032   api->disconnect_peer = &udp_disconnect;
4033   api->address_pretty_printer = &udp_plugin_address_pretty_printer;
4034   api->address_to_string = &udp_address_to_string;
4035   api->string_to_address = &udp_string_to_address;
4036   api->check_address = &udp_plugin_check_address;
4037   api->get_session = &udp_plugin_get_session;
4038   api->send = &udp_plugin_send;
4039   api->get_network = &udp_plugin_get_network;
4040   api->get_network_for_address = &udp_plugin_get_network_for_address;
4041   api->update_session_timeout = &udp_plugin_update_session_timeout;
4042   api->setup_monitor = &udp_plugin_setup_monitor;
4043   return api;
4044 }
4045
4046
4047 /**
4048  * Function called on each entry in the defragmentation heap to
4049  * clean it up.
4050  *
4051  * @param cls NULL
4052  * @param node node in the heap (to be removed)
4053  * @param element a `struct DefragContext` to be cleaned up
4054  * @param cost unused
4055  * @return #GNUNET_YES
4056  */
4057 static int
4058 heap_cleanup_iterator (void *cls,
4059                        struct GNUNET_CONTAINER_HeapNode *node,
4060                        void *element,
4061                        GNUNET_CONTAINER_HeapCostType cost)
4062 {
4063   struct DefragContext *d_ctx = element;
4064
4065   GNUNET_CONTAINER_heap_remove_node (node);
4066   GNUNET_DEFRAGMENT_context_destroy (d_ctx->defrag);
4067   GNUNET_free (d_ctx);
4068   return GNUNET_YES;
4069 }
4070
4071
4072 /**
4073  * The exported method. Makes the core api available via a global and
4074  * returns the udp transport API.
4075  *
4076  * @param cls our `struct GNUNET_TRANSPORT_PluginEnvironment`
4077  * @return NULL
4078  */
4079 void *
4080 libgnunet_plugin_transport_udp_done (void *cls)
4081 {
4082   struct GNUNET_TRANSPORT_PluginFunctions *api = cls;
4083   struct Plugin *plugin = api->cls;
4084   struct PrettyPrinterContext *cur;
4085   struct UDP_MessageWrapper *udpw;
4086
4087   if (NULL == plugin)
4088   {
4089     GNUNET_free (api);
4090     return NULL;
4091   }
4092   stop_broadcast (plugin);
4093   if (NULL != plugin->select_task_v4)
4094   {
4095     GNUNET_SCHEDULER_cancel (plugin->select_task_v4);
4096     plugin->select_task_v4 = NULL;
4097   }
4098   if (NULL != plugin->select_task_v6)
4099   {
4100     GNUNET_SCHEDULER_cancel (plugin->select_task_v6);
4101     plugin->select_task_v6 = NULL;
4102   }
4103   if (NULL != plugin->sockv4)
4104   {
4105     GNUNET_break (GNUNET_OK ==
4106                   GNUNET_NETWORK_socket_close (plugin->sockv4));
4107     plugin->sockv4 = NULL;
4108   }
4109   if (NULL != plugin->sockv6)
4110   {
4111     GNUNET_break (GNUNET_OK ==
4112                   GNUNET_NETWORK_socket_close (plugin->sockv6));
4113     plugin->sockv6 = NULL;
4114   }
4115   if (NULL != plugin->nat)
4116   {
4117     GNUNET_NAT_unregister (plugin->nat);
4118     plugin->nat = NULL;
4119   }
4120   if (NULL != plugin->defrag_ctxs)
4121   {
4122     GNUNET_CONTAINER_heap_iterate (plugin->defrag_ctxs,
4123                                    &heap_cleanup_iterator,
4124                                    NULL);
4125     GNUNET_CONTAINER_heap_destroy (plugin->defrag_ctxs);
4126     plugin->defrag_ctxs = NULL;
4127   }
4128   while (NULL != (udpw = plugin->ipv4_queue_head))
4129   {
4130     dequeue (plugin,
4131              udpw);
4132     udpw->qc (udpw->qc_cls,
4133               udpw,
4134               GNUNET_SYSERR);
4135     GNUNET_free (udpw);
4136   }
4137   while (NULL != (udpw = plugin->ipv6_queue_head))
4138   {
4139     dequeue (plugin,
4140              udpw);
4141     udpw->qc (udpw->qc_cls,
4142               udpw,
4143               GNUNET_SYSERR);
4144     GNUNET_free (udpw);
4145   }
4146   GNUNET_CONTAINER_multipeermap_iterate (plugin->sessions,
4147                                          &disconnect_and_free_it,
4148                                          plugin);
4149   GNUNET_CONTAINER_multipeermap_destroy (plugin->sessions);
4150
4151   while (NULL != (cur = plugin->ppc_dll_head))
4152   {
4153     GNUNET_break (0);
4154     GNUNET_CONTAINER_DLL_remove (plugin->ppc_dll_head,
4155                                  plugin->ppc_dll_tail,
4156                                  cur);
4157     GNUNET_RESOLVER_request_cancel (cur->resolver_handle);
4158     if (NULL != cur->timeout_task)
4159     {
4160       GNUNET_SCHEDULER_cancel (cur->timeout_task);
4161       cur->timeout_task = NULL;
4162     }
4163     GNUNET_free (cur);
4164   }
4165   GNUNET_free (plugin);
4166   GNUNET_free (api);
4167   return NULL;
4168 }
4169
4170 /* end of plugin_transport_udp.c */