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