src/transport/plugin_transport_udp.c

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