src/transport/plugin_transport_udp.c

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