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