src/transport/plugin_transport_udp.c

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