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