src/transport/plugin_transport_udp_new.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_new.h"
  30 #include "gnunet_hello_lib.h"
  31 #include "gnunet_util_lib.h"
  32 #include "gnunet_fragmentation_lib.h"
  33 #include "gnunet_nat_lib.h"
  34 #include "gnunet_protocols.h"
  35 #include "gnunet_resolver_service.h"
  36 #include "gnunet_signatures.h"
  37 #include "gnunet_constants.h"
  38 #include "gnunet_statistics_service.h"
  39 #include "gnunet_transport_service.h"
  40 #include "gnunet_transport_plugin.h"
  41 #include "transport.h"
  42
  43 #define LOG(kind,...) GNUNET_log_from (kind, "transport-udp", __VA_ARGS__)
  44
  45
  46 /**
  47  * Number of messages we can defragment in parallel.  We only really
  48  * defragment 1 message at a time, but if messages get re-ordered, we
  49  * may want to keep knowledge about the previous message to avoid
  50  * discarding the current message in favor of a single fragment of a
  51  * previous message.  3 should be good since we don't expect massive
  52  * message reorderings with UDP.
  53  */
  54 #define UDP_MAX_MESSAGES_IN_DEFRAG 3
  55
  56 /**
  57  * We keep a defragmentation queue per sender address.  How many
  58  * sender addresses do we support at the same time? Memory consumption
  59  * is roughly a factor of 32k * UDP_MAX_MESSAGES_IN_DEFRAG times this
  60  * value. (So 128 corresponds to 12 MB and should suffice for
  61  * connecting to roughly 128 peers via UDP).
  62  */
  63 #define UDP_MAX_SENDER_ADDRESSES_WITH_DEFRAG 128
  64
  65
  66
  67 /**
  68  * Closure for 'append_port'.
  69  */
  70 struct PrettyPrinterContext
  71 {
  72   /**
  73    * Function to call with the result.
  74    */
  75   GNUNET_TRANSPORT_AddressStringCallback asc;
  76
  77   /**
  78    * Clsoure for 'asc'.
  79    */
  80   void *asc_cls;
  81
  82   /**
  83    * Port to add after the IP address.
  84    */
  85   uint16_t port;
  86 };
  87
  88 struct Session
  89 {
  90   /**
  91    * Which peer is this session for?
  92    */
  93   struct GNUNET_PeerIdentity target;
  94
  95   /**
  96    * Address of the other peer
  97    */
  98   const struct sockaddr *sock_addr;
  99
 100   size_t addrlen;
 101
 102   /**
 103    * Desired delay for next sending we send to other peer
 104    */
 105   struct GNUNET_TIME_Relative flow_delay_for_other_peer;
 106
 107   /**
 108    * Desired delay for next sending we received from other peer
 109    */
 110   struct GNUNET_TIME_Absolute flow_delay_from_other_peer;
 111
 112
 113   struct GNUNET_ATS_Information ats;
 114
 115   struct FragmentationContext * frag_ctx;
 116 };
 117
 118
 119 struct SessionCompareContext
 120 {
 121   struct Session *res;
 122   const struct GNUNET_HELLO_Address *addr;
 123 };
 124
 125
 126 /**
 127  * Closure for 'process_inbound_tokenized_messages'
 128  */
 129 struct SourceInformation
 130 {
 131   /**
 132    * Sender identity.
 133    */
 134   struct GNUNET_PeerIdentity sender;
 135
 136   /**
 137    * Source address.
 138    */
 139   const void *arg;
 140
 141   /**
 142    * Number of bytes in source address.
 143    */
 144   size_t args;
 145
 146   struct Session *session;
 147 };
 148
 149
 150 /**
 151  * Closure for 'find_receive_context'.
 152  */
 153 struct FindReceiveContext
 154 {
 155   /**
 156    * Where to store the result.
 157    */
 158   struct DefragContext *rc;
 159
 160   /**
 161    * Address to find.
 162    */
 163   const struct sockaddr *addr;
 164
 165   /**
 166    * Number of bytes in 'addr'.
 167    */
 168   socklen_t addr_len;
 169
 170   struct Session *session;
 171 };
 172
 173
 174
 175 /**
 176  * Data structure to track defragmentation contexts based
 177  * on the source of the UDP traffic.
 178  */
 179 struct DefragContext
 180 {
 181
 182   /**
 183    * Defragmentation context.
 184    */
 185   struct GNUNET_DEFRAGMENT_Context *defrag;
 186
 187   /**
 188    * Source address this receive context is for (allocated at the
 189    * end of the struct).
 190    */
 191   const struct sockaddr *src_addr;
 192
 193   /**
 194    * Reference to master plugin struct.
 195    */
 196   struct Plugin *plugin;
 197
 198   /**
 199    * Node in the defrag heap.
 200    */
 201   struct GNUNET_CONTAINER_HeapNode *hnode;
 202
 203   /**
 204    * Length of 'src_addr'
 205    */
 206   size_t addr_len;
 207 };
 208
 209
 210
 211 /**
 212  * Closure for 'process_inbound_tokenized_messages'
 213  */
 214 struct FragmentationContext
 215 {
 216   struct FragmentationContext * next;
 217   struct FragmentationContext * prev;
 218
 219   struct Plugin * plugin;
 220   struct GNUNET_FRAGMENT_Context * frag;
 221   struct Session * session;
 222
 223   struct GNUNET_TIME_Absolute timeout;
 224
 225
 226   /**
 227    * Function to call upon completion of the transmission.
 228    */
 229   GNUNET_TRANSPORT_TransmitContinuation cont;
 230
 231   /**
 232    * Closure for 'cont'.
 233    */
 234   void *cont_cls;
 235
 236   size_t bytes_to_send;
 237 };
 238
 239
 240 struct UDPMessageWrapper
 241 {
 242   struct Session *session;
 243   struct UDPMessageWrapper *prev;
 244   struct UDPMessageWrapper *next;
 245   char *udp;
 246   size_t msg_size;
 247
 248   struct GNUNET_TIME_Absolute timeout;
 249
 250   /**
 251    * Function to call upon completion of the transmission.
 252    */
 253   GNUNET_TRANSPORT_TransmitContinuation cont;
 254
 255   /**
 256    * Closure for 'cont'.
 257    */
 258   void *cont_cls;
 259
 260   struct FragmentationContext *frag_ctx;
 261
 262 };
 263
 264
 265 /**
 266  * UDP ACK Message-Packet header (after defragmentation).
 267  */
 268 struct UDP_ACK_Message
 269 {
 270   /**
 271    * Message header.
 272    */
 273   struct GNUNET_MessageHeader header;
 274
 275   /**
 276    * Desired delay for flow control
 277    */
 278   uint32_t delay;
 279
 280   /**
 281    * What is the identity of the sender
 282    */
 283   struct GNUNET_PeerIdentity sender;
 284
 285 };
 286
 287
 288
 289 /**
 290  * Function called for a quick conversion of the binary address to
 291  * a numeric address.  Note that the caller must not free the
 292  * address and that the next call to this function is allowed
 293  * to override the address again.
 294  *
 295  * @param cls closure
 296  * @param addr binary address
 297  * @param addrlen length of the address
 298  * @return string representing the same address
 299  */
 300 const char *
 301 udp_address_to_string (void *cls, const void *addr, size_t addrlen)
 302 {
 303   static char rbuf[INET6_ADDRSTRLEN + 10];
 304   char buf[INET6_ADDRSTRLEN];
 305   const void *sb;
 306   struct in_addr a4;
 307   struct in6_addr a6;
 308   const struct IPv4UdpAddress *t4;
 309   const struct IPv6UdpAddress *t6;
 310   int af;
 311   uint16_t port;
 312
 313   if (addrlen == sizeof (struct IPv6UdpAddress))
 314   {
 315     t6 = addr;
 316     af = AF_INET6;
 317     port = ntohs (t6->u6_port);
 318     memcpy (&a6, &t6->ipv6_addr, sizeof (a6));
 319     sb = &a6;
 320   }
 321   else if (addrlen == sizeof (struct IPv4UdpAddress))
 322   {
 323     t4 = addr;
 324     af = AF_INET;
 325     port = ntohs (t4->u4_port);
 326     memcpy (&a4, &t4->ipv4_addr, sizeof (a4));
 327     sb = &a4;
 328   }
 329   else
 330   {
 331     GNUNET_break_op (0);
 332     return NULL;
 333   }
 334   inet_ntop (af, sb, buf, INET6_ADDRSTRLEN);
 335   GNUNET_snprintf (rbuf, sizeof (rbuf), (af == AF_INET6) ? "[%s]:%u" : "%s:%u",
 336                    buf, port);
 337   return rbuf;
 338 }
 339
 340
 341 /**
 342  * Append our port and forward the result.
 343  *
 344  * @param cls a 'struct PrettyPrinterContext'
 345  * @param hostname result from DNS resolver
 346  */
 347 static void
 348 append_port (void *cls, const char *hostname)
 349 {
 350   struct PrettyPrinterContext *ppc = cls;
 351   char *ret;
 352
 353   if (hostname == NULL)
 354   {
 355     ppc->asc (ppc->asc_cls, NULL);
 356     GNUNET_free (ppc);
 357     return;
 358   }
 359   GNUNET_asprintf (&ret, "%s:%d", hostname, ppc->port);
 360   ppc->asc (ppc->asc_cls, ret);
 361   GNUNET_free (ret);
 362 }
 363
 364
 365 /**
 366  * Convert the transports address to a nice, human-readable
 367  * format.
 368  *
 369  * @param cls closure
 370  * @param type name of the transport that generated the address
 371  * @param addr one of the addresses of the host, NULL for the last address
 372  *        the specific address format depends on the transport
 373  * @param addrlen length of the address
 374  * @param numeric should (IP) addresses be displayed in numeric form?
 375  * @param timeout after how long should we give up?
 376  * @param asc function to call on each string
 377  * @param asc_cls closure for asc
 378  */
 379 static void
 380 udp_plugin_address_pretty_printer (void *cls, const char *type,
 381                                    const void *addr, size_t addrlen,
 382                                    int numeric,
 383                                    struct GNUNET_TIME_Relative timeout,
 384                                    GNUNET_TRANSPORT_AddressStringCallback asc,
 385                                    void *asc_cls)
 386 {
 387   struct PrettyPrinterContext *ppc;
 388   const void *sb;
 389   size_t sbs;
 390   struct sockaddr_in a4;
 391   struct sockaddr_in6 a6;
 392   const struct IPv4UdpAddress *u4;
 393   const struct IPv6UdpAddress *u6;
 394   uint16_t port;
 395
 396   if (addrlen == sizeof (struct IPv6UdpAddress))
 397   {
 398     u6 = addr;
 399     memset (&a6, 0, sizeof (a6));
 400     a6.sin6_family = AF_INET6;
 401 #if HAVE_SOCKADDR_IN_SIN_LEN
 402     a6.sin6_len = sizeof (a6);
 403 #endif
 404     a6.sin6_port = u6->u6_port;
 405     memcpy (&a6.sin6_addr, &u6->ipv6_addr, sizeof (struct in6_addr));
 406     port = ntohs (u6->u6_port);
 407     sb = &a6;
 408     sbs = sizeof (a6);
 409   }
 410   else if (addrlen == sizeof (struct IPv4UdpAddress))
 411   {
 412     u4 = addr;
 413     memset (&a4, 0, sizeof (a4));
 414     a4.sin_family = AF_INET;
 415 #if HAVE_SOCKADDR_IN_SIN_LEN
 416     a4.sin_len = sizeof (a4);
 417 #endif
 418     a4.sin_port = u4->u4_port;
 419     a4.sin_addr.s_addr = u4->ipv4_addr;
 420     port = ntohs (u4->u4_port);
 421     sb = &a4;
 422     sbs = sizeof (a4);
 423   }
 424   else
 425   {
 426     /* invalid address */
 427     GNUNET_break_op (0);
 428     asc (asc_cls, NULL);
 429     return;
 430   }
 431   ppc = GNUNET_malloc (sizeof (struct PrettyPrinterContext));
 432   ppc->asc = asc;
 433   ppc->asc_cls = asc_cls;
 434   ppc->port = port;
 435   GNUNET_RESOLVER_hostname_get (sb, sbs, !numeric, timeout, &append_port, ppc);
 436 }
 437
 438
 439 /**
 440  * Check if the given port is plausible (must be either our listen
 441  * port or our advertised port).  If it is neither, we return
 442  * GNUNET_SYSERR.
 443  *
 444  * @param plugin global variables
 445  * @param in_port port number to check
 446  * @return GNUNET_OK if port is either open_port or adv_port
 447  */
 448 static int
 449 check_port (struct Plugin *plugin, uint16_t in_port)
 450 {
 451   if ((in_port == plugin->port) || (in_port == plugin->aport))
 452     return GNUNET_OK;
 453   return GNUNET_SYSERR;
 454 }
 455
 456
 457
 458 /**
 459  * Function that will be called to check if a binary address for this
 460  * plugin is well-formed and corresponds to an address for THIS peer
 461  * (as per our configuration).  Naturally, if absolutely necessary,
 462  * plugins can be a bit conservative in their answer, but in general
 463  * plugins should make sure that the address does not redirect
 464  * traffic to a 3rd party that might try to man-in-the-middle our
 465  * traffic.
 466  *
 467  * @param cls closure, should be our handle to the Plugin
 468  * @param addr pointer to the address
 469  * @param addrlen length of addr
 470  * @return GNUNET_OK if this is a plausible address for this peer
 471  *         and transport, GNUNET_SYSERR if not
 472  *
 473  */
 474 static int
 475 udp_plugin_check_address (void *cls, const void *addr, size_t addrlen)
 476 {
 477   struct Plugin *plugin = cls;
 478   struct IPv4UdpAddress *v4;
 479   struct IPv6UdpAddress *v6;
 480
 481   if ((addrlen != sizeof (struct IPv4UdpAddress)) &&
 482       (addrlen != sizeof (struct IPv6UdpAddress)))
 483   {
 484     GNUNET_break_op (0);
 485     return GNUNET_SYSERR;
 486   }
 487   if (addrlen == sizeof (struct IPv4UdpAddress))
 488   {
 489     v4 = (struct IPv4UdpAddress *) addr;
 490     if (GNUNET_OK != check_port (plugin, ntohs (v4->u4_port)))
 491       return GNUNET_SYSERR;
 492     if (GNUNET_OK !=
 493         GNUNET_NAT_test_address (plugin->nat, &v4->ipv4_addr,
 494                                  sizeof (struct in_addr)))
 495       return GNUNET_SYSERR;
 496   }
 497   else
 498   {
 499     v6 = (struct IPv6UdpAddress *) addr;
 500     if (IN6_IS_ADDR_LINKLOCAL (&v6->ipv6_addr))
 501     {
 502       GNUNET_break_op (0);
 503       return GNUNET_SYSERR;
 504     }
 505     if (GNUNET_OK != check_port (plugin, ntohs (v6->u6_port)))
 506       return GNUNET_SYSERR;
 507     if (GNUNET_OK !=
 508         GNUNET_NAT_test_address (plugin->nat, &v6->ipv6_addr,
 509                                  sizeof (struct in6_addr)))
 510       return GNUNET_SYSERR;
 511   }
 512   return GNUNET_OK;
 513 }
 514
 515
 516 /**
 517  * Destroy a session, plugin is being unloaded.
 518  *
 519  * @param cls unused
 520  * @param key hash of public key of target peer
 521  * @param value a 'struct PeerSession*' to clean up
 522  * @return GNUNET_OK (continue to iterate)
 523  */
 524 static int
 525 disconnect_and_free_it (void *cls, const GNUNET_HashCode * key, void *value)
 526 {
 527   struct Plugin *plugin = cls;
 528   struct Session *s = value;
 529   struct UDPMessageWrapper *udpw;
 530
 531 #if DEBUG_UDP
 532   LOG (GNUNET_ERROR_TYPE_DEBUG,
 533        "Session %p to peer `%s' address ended \n",
 534          s,
 535          GNUNET_i2s (&s->target),
 536          GNUNET_a2s (s->sock_addr, s->addrlen));
 537 #endif
 538   plugin->env->session_end (plugin->env->cls, &s->target, s);
 539
 540   while (s->frag_ctx != NULL)
 541   {
 542     GNUNET_FRAGMENT_context_destroy(s->frag_ctx->frag);
 543     GNUNET_free (s->frag_ctx);
 544     s->frag_ctx = NULL;
 545   }
 546
 547   udpw = plugin->msg_head;
 548   while (udpw != NULL)
 549   {
 550     if (udpw->session == s)
 551     {
 552       GNUNET_CONTAINER_DLL_remove(plugin->msg_head, plugin->msg_tail, udpw);
 553
 554       if (udpw->cont != NULL)
 555         udpw->cont (udpw->cont_cls, &s->target, GNUNET_SYSERR);
 556       GNUNET_free (udpw);
 557     }
 558     udpw = plugin->msg_head;
 559   }
 560
 561   GNUNET_assert (GNUNET_YES ==
 562                  GNUNET_CONTAINER_multihashmap_remove (plugin->sessions,
 563                                                        &s->target.hashPubKey,
 564                                                        s));
 565
 566
 567   GNUNET_free (s);
 568   return GNUNET_OK;
 569 }
 570
 571
 572 /**
 573  * Disconnect from a remote node.  Clean up session if we have one for this peer
 574  *
 575  * @param cls closure for this call (should be handle to Plugin)
 576  * @param target the peeridentity of the peer to disconnect
 577  * @return GNUNET_OK on success, GNUNET_SYSERR if the operation failed
 578  */
 579 static void
 580 udp_disconnect (void *cls, const struct GNUNET_PeerIdentity *target)
 581 {
 582   struct Plugin *plugin = cls;
 583   GNUNET_assert (plugin != NULL);
 584
 585   GNUNET_assert (target != NULL);
 586 #if DEBUG_UDP
 587   LOG (GNUNET_ERROR_TYPE_DEBUG,
 588        "Disconnecting from peer `%s'\n", GNUNET_i2s (target));
 589 #endif
 590   /* Clean up sessions */
 591   GNUNET_CONTAINER_multihashmap_get_multiple (plugin->sessions, &target->hashPubKey, &disconnect_and_free_it, plugin);
 592
 593   LOG (GNUNET_ERROR_TYPE_DEBUG,
 594        "FREEED SESSIONS from peer `%s'\n", GNUNET_i2s (target));
 595 }
 596
 597 static struct Session *
 598 create_session (struct Plugin *plugin, const struct GNUNET_PeerIdentity *target,
 599                 const void *addr, size_t addrlen,
 600                 GNUNET_TRANSPORT_TransmitContinuation cont, void *cont_cls)
 601 {
 602   struct Session *s;
 603   const struct IPv4UdpAddress *t4;
 604   const struct IPv6UdpAddress *t6;
 605   struct sockaddr_in *v4;
 606   struct sockaddr_in6 *v6;
 607   size_t len;
 608
 609   switch (addrlen)
 610   {
 611   case sizeof (struct IPv4UdpAddress):
 612     if (NULL == plugin->sockv4)
 613     {
 614       return NULL;
 615     }
 616     t4 = addr;
 617     s = GNUNET_malloc (sizeof (struct Session) + sizeof (struct sockaddr_in));
 618     len = sizeof (struct sockaddr_in);
 619     v4 = (struct sockaddr_in *) &s[1];
 620     v4->sin_family = AF_INET;
 621 #if HAVE_SOCKADDR_IN_SIN_LEN
 622     v4->sin_len = sizeof (struct sockaddr_in);
 623 #endif
 624     v4->sin_port = t4->u4_port;
 625     v4->sin_addr.s_addr = t4->ipv4_addr;
 626     s->ats = plugin->env->get_address_type (plugin->env->cls, (const struct sockaddr *) v4, sizeof (struct sockaddr_in));
 627     break;
 628   case sizeof (struct IPv6UdpAddress):
 629     if (NULL == plugin->sockv6)
 630     {
 631       return NULL;
 632     }
 633     t6 = addr;
 634     s =
 635         GNUNET_malloc (sizeof (struct Session) + sizeof (struct sockaddr_in6));
 636     len = sizeof (struct sockaddr_in6);
 637     v6 = (struct sockaddr_in6 *) &s[1];
 638     v6->sin6_family = AF_INET6;
 639 #if HAVE_SOCKADDR_IN_SIN_LEN
 640     v6->sin6_len = sizeof (struct sockaddr_in6);
 641 #endif
 642     v6->sin6_port = t6->u6_port;
 643     v6->sin6_addr = t6->ipv6_addr;
 644     s->ats = plugin->env->get_address_type (plugin->env->cls, (const struct sockaddr *) v6, sizeof (struct sockaddr_in6));
 645     break;
 646   default:
 647     /* Must have a valid address to send to */
 648     GNUNET_break_op (0);
 649     return NULL;
 650   }
 651
 652   s->addrlen = len;
 653   s->target = *target;
 654   s->sock_addr = (const struct sockaddr *) &s[1];
 655   s->flow_delay_for_other_peer = GNUNET_TIME_relative_get_zero();
 656   s->flow_delay_from_other_peer = GNUNET_TIME_absolute_get_zero();
 657
 658   return s;
 659 }
 660
 661 static int session_cmp_it (void *cls,
 662                            const GNUNET_HashCode * key,
 663                            void *value)
 664 {
 665   struct SessionCompareContext * cctx = cls;
 666   const struct GNUNET_HELLO_Address *address = cctx->addr;
 667   struct Session *s = value;
 668
 669   socklen_t s_addrlen = s->addrlen;
 670
 671 #if VERBOSE_UDP
 672   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Comparing  address %s <-> %s\n",
 673       udp_address_to_string (NULL, (void *) address->address, address->address_length),
 674       GNUNET_a2s (s->sock_addr, s->addrlen));
 675 #endif
 676
 677   if ((address->address_length == sizeof (struct IPv4UdpAddress)) &&
 678       (s_addrlen == sizeof (struct sockaddr_in)))
 679   {
 680     struct IPv4UdpAddress * u4 = NULL;
 681     u4 = (struct IPv4UdpAddress *) address->address;
 682     const struct sockaddr_in *s4 = (const struct sockaddr_in *) s->sock_addr;
 683     if ((0 == memcmp ((const void *) &u4->ipv4_addr,(const void *) &s4->sin_addr, sizeof (struct in_addr))) &&
 684         (u4->u4_port == s4->sin_port))
 685     {
 686       cctx->res = s;
 687       return GNUNET_NO;
 688     }
 689
 690   }
 691   if ((address->address_length == sizeof (struct IPv6UdpAddress)) &&
 692       (s_addrlen == sizeof (struct sockaddr_in6)))
 693   {
 694     struct IPv6UdpAddress * u6 = NULL;
 695     u6 = (struct IPv6UdpAddress *) address->address;
 696     const struct sockaddr_in6 *s6 = (const struct sockaddr_in6 *) s->sock_addr;
 697     if ((0 == memcmp (&u6->ipv6_addr, &s6->sin6_addr, sizeof (struct in6_addr))) &&
 698         (u6->u6_port == s6->sin6_port))
 699     {
 700       cctx->res = s;
 701       return GNUNET_NO;
 702     }
 703   }
 704
 705
 706   return GNUNET_YES;
 707 }
 708
 709
 710 /**
 711  * Creates a new outbound session the transport service will use to send data to the
 712  * peer
 713  *
 714  * @param cls the plugin
 715  * @param address the address
 716  * @return the session or NULL of max connections exceeded
 717  */
 718 static struct Session *
 719 udp_plugin_get_session (void *cls,
 720                   const struct GNUNET_HELLO_Address *address)
 721 {
 722   struct Session * s = NULL;
 723   struct Plugin * plugin = cls;
 724
 725   GNUNET_assert (plugin != NULL);
 726   GNUNET_assert (address != NULL);
 727
 728   if ((address->address == NULL) ||
 729       ((address->address_length != sizeof (struct IPv4UdpAddress)) &&
 730       (address->address_length != sizeof (struct IPv6UdpAddress))))
 731   {
 732     GNUNET_break (0);
 733     return NULL;
 734   }
 735
 736   /* check if session already exists */
 737   struct SessionCompareContext cctx;
 738   cctx.addr = address;
 739   cctx.res = NULL;
 740 #if VERBOSE_UDP
 741   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Looking for existing session for peer `%s' `%s' \n", GNUNET_i2s (&address->peer), udp_address_to_string(NULL, address->address, address->address_length));
 742 #endif
 743   GNUNET_CONTAINER_multihashmap_get_multiple(plugin->sessions, &address->peer.hashPubKey, session_cmp_it, &cctx);
 744   if (cctx.res != NULL)
 745   {
 746 #if VERBOSE_UDP
 747     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Found existing session %p\n", cctx.res);
 748 #endif
 749     return cctx.res;
 750   }
 751
 752   /* otherwise create new */
 753   s = create_session (plugin,
 754       &address->peer,
 755       address->address,
 756       address->address_length,
 757       NULL, NULL);
 758 #if VERBOSE
 759     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 760               "Creating new session %p for peer `%s' address `%s'\n",
 761               s,
 762               GNUNET_i2s(&address->peer),
 763               udp_address_to_string(NULL,address->address,address->address_length));
 764 #endif
 765   GNUNET_assert (GNUNET_OK ==
 766                  GNUNET_CONTAINER_multihashmap_put (plugin->sessions,
 767                                                     &s->target.hashPubKey,
 768                                                     s,
 769                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
 770
 771   return s;
 772 }
 773
 774 /**
 775  * Function that is called with messages created by the fragmentation
 776  * module.  In the case of the 'proc' callback of the
 777  * GNUNET_FRAGMENT_context_create function, this function must
 778  * eventually call 'GNUNET_FRAGMENT_context_transmission_done'.
 779  *
 780  * @param cls closure, the 'struct FragmentationContext'
 781  * @param msg the message that was created
 782  */
 783 static void
 784 enqueue_fragment (void *cls, const struct GNUNET_MessageHeader *msg)
 785 {
 786   struct FragmentationContext *frag_ctx = cls;
 787   struct Plugin *plugin = frag_ctx->plugin;
 788   struct UDPMessageWrapper * udpw;
 789
 790   size_t msg_len = ntohs (msg->size);
 791
 792 #if VERBOSE_UDP
 793   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Enqueuing fragment with %u bytes %u\n", msg_len , sizeof (struct UDPMessageWrapper));
 794 #endif
 795
 796   udpw = GNUNET_malloc (sizeof (struct UDPMessageWrapper) + msg_len);
 797   udpw->session = frag_ctx->session;
 798   udpw->udp = (char *) &udpw[1];
 799
 800   udpw->msg_size = msg_len;
 801   udpw->cont = frag_ctx->cont;
 802   udpw->cont_cls = frag_ctx->cont_cls;
 803   udpw->timeout = frag_ctx->timeout;
 804   udpw->frag_ctx = frag_ctx;
 805   memcpy (udpw->udp, msg, msg_len);
 806
 807   GNUNET_CONTAINER_DLL_insert(plugin->msg_head, plugin->msg_tail, udpw);
 808 }
 809
 810
 811 /**
 812  * Function that can be used by the transport service to transmit
 813  * a message using the plugin.   Note that in the case of a
 814  * peer disconnecting, the continuation MUST be called
 815  * prior to the disconnect notification itself.  This function
 816  * will be called with this peer's HELLO message to initiate
 817  * a fresh connection to another peer.
 818  *
 819  * @param cls closure
 820  * @param session which session must be used
 821  * @param msgbuf the message to transmit
 822  * @param msgbuf_size number of bytes in 'msgbuf'
 823  * @param priority how important is the message (most plugins will
 824  *                 ignore message priority and just FIFO)
 825  * @param to how long to wait at most for the transmission (does not
 826  *                require plugins to discard the message after the timeout,
 827  *                just advisory for the desired delay; most plugins will ignore
 828  *                this as well)
 829  * @param cont continuation to call once the message has
 830  *        been transmitted (or if the transport is ready
 831  *        for the next transmission call; or if the
 832  *        peer disconnected...); can be NULL
 833  * @param cont_cls closure for cont
 834  * @return number of bytes used (on the physical network, with overheads);
 835  *         -1 on hard errors (i.e. address invalid); 0 is a legal value
 836  *         and does NOT mean that the message was not transmitted (DV)
 837  */
 838 static ssize_t
 839 udp_plugin_send (void *cls,
 840                   struct Session *s,
 841                   const char *msgbuf, size_t msgbuf_size,
 842                   unsigned int priority,
 843                   struct GNUNET_TIME_Relative to,
 844                   GNUNET_TRANSPORT_TransmitContinuation cont, void *cont_cls)
 845 {
 846   struct Plugin *plugin = cls;
 847   size_t mlen = msgbuf_size + sizeof (struct UDPMessage);
 848
 849   struct UDPMessageWrapper * udpw;
 850   struct UDPMessage *udp;
 851   char mbuf[mlen];
 852   GNUNET_assert (plugin != NULL);
 853   GNUNET_assert (s != NULL);
 854
 855   if (mlen >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
 856   {
 857     GNUNET_break (0);
 858     return GNUNET_SYSERR;
 859   }
 860
 861   if (GNUNET_YES != GNUNET_CONTAINER_multihashmap_contains_value(plugin->sessions, &s->target.hashPubKey, s))
 862   {
 863     GNUNET_break (0);
 864     return GNUNET_SYSERR;
 865   }
 866
 867   LOG (GNUNET_ERROR_TYPE_DEBUG,
 868        "UDP transmits %u-byte message to `%s' using address `%s'\n",
 869          msgbuf_size,
 870          GNUNET_i2s (&s->target),
 871          GNUNET_a2s(s->sock_addr, s->addrlen));
 872
 873   /* Message */
 874   udp = (struct UDPMessage *) mbuf;
 875   udp->header.size = htons (mlen);
 876   udp->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_MESSAGE);
 877   udp->reserved = htonl (0);
 878   udp->sender = *plugin->env->my_identity;
 879
 880   if (mlen <= UDP_MTU)
 881   {
 882     udpw = GNUNET_malloc (sizeof (struct UDPMessageWrapper) + mlen);
 883     udpw->session = s;
 884     udpw->udp = (char *) &udpw[1];
 885     udpw->msg_size = mlen;
 886     udpw->timeout = GNUNET_TIME_absolute_add(GNUNET_TIME_absolute_get(), to);
 887     udpw->cont = cont;
 888     udpw->cont_cls = cont_cls;
 889     udpw->frag_ctx = NULL;
 890
 891     memcpy (udpw->udp, udp, sizeof (struct UDPMessage));
 892     memcpy (&udpw->udp[sizeof (struct UDPMessage)], msgbuf, msgbuf_size);
 893
 894     GNUNET_CONTAINER_DLL_insert(plugin->msg_head, plugin->msg_tail, udpw);
 895   }
 896   else
 897   {
 898     LOG (GNUNET_ERROR_TYPE_DEBUG,
 899          "UDP has to fragment message \n");
 900     if  (s->frag_ctx != NULL)
 901       return GNUNET_SYSERR;
 902     memcpy (&udp[1], msgbuf, msgbuf_size);
 903     struct FragmentationContext * frag_ctx = GNUNET_malloc(sizeof (struct FragmentationContext));
 904
 905     frag_ctx->plugin = plugin;
 906     frag_ctx->session = s;
 907     frag_ctx->cont = cont;
 908     frag_ctx->cont_cls = cont_cls;
 909     frag_ctx->timeout = GNUNET_TIME_absolute_add(GNUNET_TIME_absolute_get(), to);
 910     frag_ctx->bytes_to_send = mlen;
 911     frag_ctx->frag = GNUNET_FRAGMENT_context_create (plugin->env->stats,
 912               UDP_MTU,
 913               &plugin->tracker,
 914               plugin->last_expected_delay,
 915               &udp->header,
 916               &enqueue_fragment,
 917               frag_ctx);
 918
 919     s->frag_ctx = frag_ctx;
 920
 921   }
 922   return mlen;
 923 }
 924
 925 static ssize_t udp_plugin_send_wrapper (void *cls,
 926                                         const struct
 927                                         GNUNET_PeerIdentity *
 928                                         target,
 929                                         const char *msgbuf,
 930                                         size_t msgbuf_size,
 931                                         uint32_t priority,
 932                                         struct
 933                                         GNUNET_TIME_Relative
 934                                         timeout,
 935                                         struct Session * session,
 936                                         const void *addr,
 937                                         size_t addrlen,
 938                                         int force_address,
 939                                         GNUNET_TRANSPORT_TransmitContinuation
 940                                         cont, void *cont_cls)
 941 {
 942   int ret;
 943   struct Session * s = NULL;
 944   struct GNUNET_HELLO_Address * ha = NULL;
 945
 946   ha = GNUNET_HELLO_address_allocate(target, "", addr,addrlen);
 947   GNUNET_assert (ha != NULL);
 948   s = udp_plugin_get_session(cls, ha);
 949   GNUNET_assert (s != NULL);
 950   GNUNET_free (ha);
 951   ret = udp_plugin_send (cls, s, msgbuf, msgbuf_size, priority, timeout, cont, cont_cls);
 952
 953   return ret;
 954 }
 955
 956
 957 /**
 958  * Our external IP address/port mapping has changed.
 959  *
 960  * @param cls closure, the 'struct LocalAddrList'
 961  * @param add_remove GNUNET_YES to mean the new public IP address, GNUNET_NO to mean
 962  *     the previous (now invalid) one
 963  * @param addr either the previous or the new public IP address
 964  * @param addrlen actual lenght of the address
 965  */
 966 static void
 967 udp_nat_port_map_callback (void *cls, int add_remove,
 968                            const struct sockaddr *addr, socklen_t addrlen)
 969 {
 970   struct Plugin *plugin = cls;
 971   struct IPv4UdpAddress u4;
 972   struct IPv6UdpAddress u6;
 973   void *arg;
 974   size_t args;
 975
 976   /* convert 'addr' to our internal format */
 977   switch (addr->sa_family)
 978   {
 979   case AF_INET:
 980     GNUNET_assert (addrlen == sizeof (struct sockaddr_in));
 981     u4.ipv4_addr = ((struct sockaddr_in *) addr)->sin_addr.s_addr;
 982     u4.u4_port = ((struct sockaddr_in *) addr)->sin_port;
 983     arg = &u4;
 984     args = sizeof (u4);
 985     break;
 986   case AF_INET6:
 987     GNUNET_assert (addrlen == sizeof (struct sockaddr_in6));
 988     memcpy (&u6.ipv6_addr, &((struct sockaddr_in6 *) addr)->sin6_addr,
 989             sizeof (struct in6_addr));
 990     u6.u6_port = ((struct sockaddr_in6 *) addr)->sin6_port;
 991     arg = &u6;
 992     args = sizeof (u6);
 993     break;
 994   default:
 995     GNUNET_break (0);
 996     return;
 997   }
 998   /* modify our published address list */
 999   plugin->env->notify_address (plugin->env->cls, add_remove, arg, args);
1000 }
1001
1002
1003
1004 /**
1005  * Message tokenizer has broken up an incomming message. Pass it on
1006  * to the service.
1007  *
1008  * @param cls the 'struct Plugin'
1009  * @param client the 'struct SourceInformation'
1010  * @param hdr the actual message
1011  */
1012 static void
1013 process_inbound_tokenized_messages (void *cls, void *client,
1014                                     const struct GNUNET_MessageHeader *hdr)
1015 {
1016   struct Plugin *plugin = cls;
1017   struct SourceInformation *si = client;
1018   struct GNUNET_ATS_Information ats[2];
1019   struct GNUNET_TIME_Relative delay;
1020
1021   GNUNET_assert (si->session != NULL);
1022   /* setup ATS */
1023   ats[0].type = htonl (GNUNET_ATS_QUALITY_NET_DISTANCE);
1024   ats[0].value = htonl (1);
1025   ats[1] = si->session->ats;
1026   GNUNET_break (ntohl(ats[1].value) != GNUNET_ATS_NET_UNSPECIFIED);
1027
1028   delay = plugin->env->receive (plugin->env->cls,
1029                 &si->sender,
1030                 hdr,
1031                 (const struct GNUNET_ATS_Information *) &ats, 2,
1032                 NULL,
1033                 si->arg,
1034                 si->args);
1035   si->session->flow_delay_for_other_peer = delay;
1036 }
1037
1038
1039 /**
1040  * We've received a UDP Message.  Process it (pass contents to main service).
1041  *
1042  * @param plugin plugin context
1043  * @param msg the message
1044  * @param sender_addr sender address
1045  * @param sender_addr_len number of bytes in sender_addr
1046  */
1047 static void
1048 process_udp_message (struct Plugin *plugin, const struct UDPMessage *msg,
1049                      const struct sockaddr *sender_addr,
1050                      socklen_t sender_addr_len)
1051 {
1052   struct SourceInformation si;
1053   struct Session * s = NULL;
1054   struct IPv4UdpAddress u4;
1055   struct IPv6UdpAddress u6;
1056   struct GNUNET_ATS_Information ats;
1057   const void *arg;
1058   size_t args;
1059
1060   if (0 != ntohl (msg->reserved))
1061   {
1062     GNUNET_break_op (0);
1063     return;
1064   }
1065   if (ntohs (msg->header.size) <
1066       sizeof (struct GNUNET_MessageHeader) + sizeof (struct UDPMessage))
1067   {
1068     GNUNET_break_op (0);
1069     return;
1070   }
1071
1072   ats.type = htonl (GNUNET_ATS_NETWORK_TYPE);
1073   ats.value = htonl (GNUNET_ATS_NET_UNSPECIFIED);
1074   /* convert address */
1075   switch (sender_addr->sa_family)
1076   {
1077   case AF_INET:
1078     GNUNET_assert (sender_addr_len == sizeof (struct sockaddr_in));
1079     u4.ipv4_addr = ((struct sockaddr_in *) sender_addr)->sin_addr.s_addr;
1080     u4.u4_port = ((struct sockaddr_in *) sender_addr)->sin_port;
1081     arg = &u4;
1082     args = sizeof (u4);
1083     break;
1084   case AF_INET6:
1085     GNUNET_assert (sender_addr_len == sizeof (struct sockaddr_in6));
1086     u6.ipv6_addr = ((struct sockaddr_in6 *) sender_addr)->sin6_addr;
1087     u6.u6_port = ((struct sockaddr_in6 *) sender_addr)->sin6_port;
1088     arg = &u6;
1089     args = sizeof (u6);
1090     break;
1091   default:
1092     GNUNET_break (0);
1093     return;
1094   }
1095 #if DEBUG_UDP
1096   LOG (GNUNET_ERROR_TYPE_DEBUG,
1097        "Received message with %u bytes from peer `%s' at `%s'\n",
1098        (unsigned int) ntohs (msg->header.size), GNUNET_i2s (&msg->sender),
1099        GNUNET_a2s (sender_addr, sender_addr_len));
1100 #endif
1101
1102   struct GNUNET_HELLO_Address * address = GNUNET_HELLO_address_allocate(&msg->sender, "udp", arg, args);
1103   s = udp_plugin_get_session(plugin, address);
1104   GNUNET_free (address);
1105
1106   /* iterate over all embedded messages */
1107   si.session = s;
1108   si.sender = msg->sender;
1109   si.arg = arg;
1110   si.args = args;
1111
1112   GNUNET_SERVER_mst_receive (plugin->mst, &si, (const char *) &msg[1],
1113                              ntohs (msg->header.size) -
1114                              sizeof (struct UDPMessage), GNUNET_YES, GNUNET_NO);
1115 }
1116
1117
1118 /**
1119  * Scan the heap for a receive context with the given address.
1120  *
1121  * @param cls the 'struct FindReceiveContext'
1122  * @param node internal node of the heap
1123  * @param element value stored at the node (a 'struct ReceiveContext')
1124  * @param cost cost associated with the node
1125  * @return GNUNET_YES if we should continue to iterate,
1126  *         GNUNET_NO if not.
1127  */
1128 static int
1129 find_receive_context (void *cls, struct GNUNET_CONTAINER_HeapNode *node,
1130                       void *element, GNUNET_CONTAINER_HeapCostType cost)
1131 {
1132   struct FindReceiveContext *frc = cls;
1133   struct DefragContext *e = element;
1134
1135   if ((frc->addr_len == e->addr_len) &&
1136       (0 == memcmp (frc->addr, e->src_addr, frc->addr_len)))
1137   {
1138     frc->rc = e;
1139     return GNUNET_NO;
1140   }
1141   return GNUNET_YES;
1142 }
1143
1144
1145 /**
1146  * Process a defragmented message.
1147  *
1148  * @param cls the 'struct ReceiveContext'
1149  * @param msg the message
1150  */
1151 static void
1152 fragment_msg_proc (void *cls, const struct GNUNET_MessageHeader *msg)
1153 {
1154   struct DefragContext *rc = cls;
1155
1156   if (ntohs (msg->type) != GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_MESSAGE)
1157   {
1158     GNUNET_break (0);
1159     return;
1160   }
1161   if (ntohs (msg->size) < sizeof (struct UDPMessage))
1162   {
1163     GNUNET_break (0);
1164     return;
1165   }
1166   process_udp_message (rc->plugin, (const struct UDPMessage *) msg,
1167                        rc->src_addr, rc->addr_len);
1168 }
1169
1170 struct LookupContext
1171 {
1172   const struct sockaddr * addr;
1173   size_t addrlen;
1174
1175   struct Session *res;
1176 };
1177
1178 static int
1179 lookup_session_by_addr_it (void *cls, const GNUNET_HashCode * key, void *value)
1180 {
1181   struct LookupContext *l_ctx = cls;
1182   struct Session * s = value;
1183
1184   if ((s->addrlen == l_ctx->addrlen) &&
1185       (0 == memcmp (s->sock_addr, l_ctx->addr, s->addrlen)))
1186   {
1187     l_ctx->res = s;
1188     return GNUNET_NO;
1189   }
1190   return GNUNET_YES;
1191 }
1192
1193 /**
1194  * Transmit an acknowledgement.
1195  *
1196  * @param cls the 'struct ReceiveContext'
1197  * @param id message ID (unused)
1198  * @param msg ack to transmit
1199  */
1200 static void
1201 ack_proc (void *cls, uint32_t id, const struct GNUNET_MessageHeader *msg)
1202 {
1203   struct DefragContext *rc = cls;
1204
1205   size_t msize = sizeof (struct UDP_ACK_Message) + ntohs (msg->size);
1206   struct UDP_ACK_Message *udp_ack;
1207   uint32_t delay = 0;
1208   struct UDPMessageWrapper *udpw;
1209   struct Session *s;
1210
1211   struct LookupContext l_ctx;
1212   l_ctx.addr = rc->src_addr;
1213   l_ctx.addrlen = rc->addr_len;
1214   l_ctx.res = NULL;
1215   GNUNET_CONTAINER_multihashmap_iterate (rc->plugin->sessions,
1216       &lookup_session_by_addr_it,
1217       &l_ctx);
1218   s = l_ctx.res;
1219
1220   GNUNET_assert (s != NULL);
1221
1222   if (s->flow_delay_for_other_peer.rel_value <= UINT32_MAX)
1223     delay = s->flow_delay_for_other_peer.rel_value;
1224
1225 #if DEBUG_UDP
1226   LOG (GNUNET_ERROR_TYPE_DEBUG,
1227        "Sending ACK to `%s' including delay of %u ms\n",
1228        GNUNET_a2s (rc->src_addr,
1229                    (rc->src_addr->sa_family ==
1230                     AF_INET) ? sizeof (struct sockaddr_in) : sizeof (struct
1231                                                                      sockaddr_in6)),
1232        delay);
1233 #endif
1234   udpw = GNUNET_malloc (sizeof (struct UDPMessageWrapper) + msize);
1235   udpw->cont = NULL;
1236   udpw->cont_cls = NULL;
1237   udpw->frag_ctx = NULL;
1238   udpw->msg_size = msize;
1239   udpw->session = s;
1240   udpw->timeout = GNUNET_TIME_absolute_get_forever();
1241   udpw->udp = (char *)&udpw[1];
1242
1243   udp_ack = (struct UDP_ACK_Message *) udpw->udp;
1244   udp_ack->header.size = htons ((uint16_t) msize);
1245   udp_ack->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_ACK);
1246   udp_ack->delay = htonl (delay);
1247   udp_ack->sender = *rc->plugin->env->my_identity;
1248   memcpy (&udp_ack[1], msg, ntohs (msg->size));
1249
1250   GNUNET_CONTAINER_DLL_insert(rc->plugin->msg_head, rc->plugin->msg_tail, udpw);
1251 }
1252
1253
1254 static void read_process_msg (struct Plugin *plugin,
1255     const struct GNUNET_MessageHeader *msg,
1256     char *addr,
1257     socklen_t fromlen)
1258 {
1259   if (ntohs (msg->size) < sizeof (struct UDPMessage))
1260   {
1261     GNUNET_break_op (0);
1262     return;
1263   }
1264   process_udp_message (plugin, (const struct UDPMessage *) msg,
1265                        (const struct sockaddr *) addr, fromlen);
1266   return;
1267 }
1268
1269 static void read_process_ack (struct Plugin *plugin,
1270     const struct GNUNET_MessageHeader *msg,
1271     char *addr,
1272     socklen_t fromlen)
1273 {
1274   const struct GNUNET_MessageHeader *ack;
1275   const struct UDP_ACK_Message *udp_ack;
1276   struct LookupContext l_ctx;
1277   struct Session *s = NULL;
1278   struct GNUNET_TIME_Relative flow_delay;
1279
1280   if (ntohs (msg->size) <
1281       sizeof (struct UDP_ACK_Message) + sizeof (struct GNUNET_MessageHeader))
1282   {
1283     GNUNET_break_op (0);
1284     return;
1285   }
1286
1287   udp_ack = (const struct UDP_ACK_Message *) msg;
1288
1289   l_ctx.addr = (const struct sockaddr *) addr;
1290   l_ctx.addrlen = fromlen;
1291   l_ctx.res = NULL;
1292   GNUNET_CONTAINER_multihashmap_iterate (plugin->sessions,
1293       &lookup_session_by_addr_it,
1294       &l_ctx);
1295   s = l_ctx.res;
1296
1297   if ((s == NULL) || (s->frag_ctx == NULL))
1298     return;
1299
1300   flow_delay.rel_value = (uint64_t) ntohl (udp_ack->delay);
1301   LOG (GNUNET_ERROR_TYPE_DEBUG, "We received a sending delay of %llu\n",
1302        flow_delay.rel_value);
1303   s->flow_delay_from_other_peer =
1304       GNUNET_TIME_relative_to_absolute (flow_delay);
1305
1306   ack = (const struct GNUNET_MessageHeader *) &udp_ack[1];
1307   if (ntohs (ack->size) !=
1308       ntohs (msg->size) - sizeof (struct UDP_ACK_Message))
1309   {
1310     GNUNET_break_op (0);
1311     return;
1312   }
1313
1314   if (GNUNET_OK != GNUNET_FRAGMENT_process_ack (s->frag_ctx->frag, ack))
1315   {
1316 #if DEBUG_UDP
1317   LOG (GNUNET_ERROR_TYPE_DEBUG,
1318        "UDP processes %u-byte acknowledgement from `%s' at `%s'\n",
1319        (unsigned int) ntohs (msg->size), GNUNET_i2s (&udp_ack->sender),
1320        GNUNET_a2s ((const struct sockaddr *) addr, fromlen));
1321 #endif
1322     return;
1323   }
1324
1325 #if DEBUG_UDP
1326   LOG (GNUNET_ERROR_TYPE_DEBUG,
1327        "FULL MESSAGE ACKed\n",
1328        (unsigned int) ntohs (msg->size), GNUNET_i2s (&udp_ack->sender),
1329        GNUNET_a2s ((const struct sockaddr *) addr, fromlen));
1330 #endif
1331   plugin->last_expected_delay = GNUNET_FRAGMENT_context_destroy (s->frag_ctx->frag);
1332
1333   struct UDPMessageWrapper * udpw = plugin->msg_head;
1334   while (udpw!= NULL)
1335   {
1336     if ((udpw->frag_ctx != NULL) && (udpw->frag_ctx == s->frag_ctx))
1337     {
1338       GNUNET_CONTAINER_DLL_remove(plugin->msg_head, plugin->msg_tail, udpw);
1339       GNUNET_free (udpw);
1340     }
1341     udpw = udpw->next;
1342   }
1343
1344   if (s->frag_ctx->cont != NULL)
1345     s->frag_ctx->cont
1346     (s->frag_ctx->cont_cls, &udp_ack->sender, GNUNET_OK);
1347   GNUNET_free (s->frag_ctx);
1348   s->frag_ctx = NULL;
1349   return;
1350 }
1351
1352 static void read_process_fragment (struct Plugin *plugin,
1353     const struct GNUNET_MessageHeader *msg,
1354     char *addr,
1355     socklen_t fromlen)
1356 {
1357   struct DefragContext *d_ctx;
1358   struct GNUNET_TIME_Absolute now;
1359   struct FindReceiveContext frc;
1360
1361
1362   frc.rc = NULL;
1363   frc.addr = (const struct sockaddr *) addr;
1364   frc.addr_len = fromlen;
1365
1366 #if DEBUG_UDP
1367   LOG (GNUNET_ERROR_TYPE_DEBUG, "UDP processes %u-byte fragment from `%s'\n",
1368        (unsigned int) ntohs (msg->size),
1369        GNUNET_a2s ((const struct sockaddr *) addr, fromlen));
1370 #endif
1371
1372   /* Lookup existing receive context for this address */
1373   GNUNET_CONTAINER_heap_iterate (plugin->defrag_ctxs,
1374                                  &find_receive_context,
1375                                  &frc);
1376   now = GNUNET_TIME_absolute_get ();
1377   d_ctx = frc.rc;
1378
1379   if (d_ctx == NULL)
1380   {
1381     /* Create a new defragmentation context */
1382     d_ctx = GNUNET_malloc (sizeof (struct DefragContext) + fromlen);
1383     memcpy (&d_ctx[1], addr, fromlen);
1384     d_ctx->src_addr = (const struct sockaddr *) &d_ctx[1];
1385     d_ctx->addr_len = fromlen;
1386     d_ctx->plugin = plugin;
1387     d_ctx->defrag =
1388         GNUNET_DEFRAGMENT_context_create (plugin->env->stats, UDP_MTU,
1389                                           UDP_MAX_MESSAGES_IN_DEFRAG, d_ctx,
1390                                           &fragment_msg_proc, &ack_proc);
1391     d_ctx->hnode =
1392         GNUNET_CONTAINER_heap_insert (plugin->defrag_ctxs, d_ctx,
1393                                       (GNUNET_CONTAINER_HeapCostType)
1394                                       now.abs_value);
1395 #if DEBUG_UDP
1396   LOG (GNUNET_ERROR_TYPE_DEBUG, "Created new defragmentation context for %u-byte fragment from `%s'\n",
1397        (unsigned int) ntohs (msg->size),
1398        GNUNET_a2s ((const struct sockaddr *) addr, fromlen));
1399 #endif
1400   }
1401   else
1402   {
1403 #if DEBUG_UDP
1404   LOG (GNUNET_ERROR_TYPE_DEBUG, "Found existing defragmentation context for %u-byte fragment from `%s'\n",
1405        (unsigned int) ntohs (msg->size),
1406        GNUNET_a2s ((const struct sockaddr *) addr, fromlen));
1407 #endif
1408   }
1409
1410   if (GNUNET_OK == GNUNET_DEFRAGMENT_process_fragment (d_ctx->defrag, msg))
1411   {
1412     /* keep this 'rc' from expiring */
1413     GNUNET_CONTAINER_heap_update_cost (plugin->defrag_ctxs, d_ctx->hnode,
1414                                        (GNUNET_CONTAINER_HeapCostType)
1415                                        now.abs_value);
1416   }
1417   if (GNUNET_CONTAINER_heap_get_size (plugin->defrag_ctxs) >
1418       UDP_MAX_SENDER_ADDRESSES_WITH_DEFRAG)
1419   {
1420     /* remove 'rc' that was inactive the longest */
1421     d_ctx = GNUNET_CONTAINER_heap_remove_root (plugin->defrag_ctxs);
1422     GNUNET_assert (NULL != d_ctx);
1423     GNUNET_DEFRAGMENT_context_destroy (d_ctx->defrag);
1424     GNUNET_free (d_ctx);
1425   }
1426 }
1427
1428 /**
1429  * Read and process a message from the given socket.
1430  *
1431  * @param plugin the overall plugin
1432  * @param rsock socket to read from
1433  */
1434 static void
1435 udp_select_read (struct Plugin *plugin, struct GNUNET_NETWORK_Handle *rsock)
1436 {
1437   socklen_t fromlen;
1438   char addr[32];
1439   char buf[65536];
1440   ssize_t size;
1441   const struct GNUNET_MessageHeader *msg;
1442
1443   fromlen = sizeof (addr);
1444   memset (&addr, 0, sizeof (addr));
1445   size = GNUNET_NETWORK_socket_recvfrom (rsock, buf, sizeof (buf),
1446                                       (struct sockaddr *) &addr, &fromlen);
1447
1448   if (size < sizeof (struct GNUNET_MessageHeader))
1449   {
1450     GNUNET_break_op (0);
1451     return;
1452   }
1453   msg = (const struct GNUNET_MessageHeader *) buf;
1454
1455   LOG (GNUNET_ERROR_TYPE_DEBUG,
1456        "UDP received %u-byte message from `%s' type %i\n", (unsigned int) size,
1457        GNUNET_a2s ((const struct sockaddr *) addr, fromlen), ntohs (msg->type));
1458
1459   if (size != ntohs (msg->size))
1460   {
1461     GNUNET_break_op (0);
1462     return;
1463   }
1464
1465   switch (ntohs (msg->type))
1466   {
1467   case GNUNET_MESSAGE_TYPE_TRANSPORT_BROADCAST_BEACON:
1468     udp_broadcast_receive (plugin, &buf, size, addr, fromlen);
1469     return;
1470
1471   case GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_MESSAGE:
1472     read_process_msg (plugin, msg, addr, fromlen);
1473     return;
1474
1475   case GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_ACK:
1476     read_process_ack (plugin, msg, addr, fromlen);;
1477     return;
1478
1479   case GNUNET_MESSAGE_TYPE_FRAGMENT:
1480     read_process_fragment (plugin, msg, addr, fromlen);
1481     return;
1482
1483   default:
1484     GNUNET_break_op (0);
1485     return;
1486   }
1487 }
1488
1489 size_t
1490 udp_select_send (struct Plugin *plugin)
1491 {
1492   ssize_t sent;
1493   size_t slen;
1494   struct GNUNET_TIME_Absolute max;
1495   struct GNUNET_TIME_Absolute ;
1496
1497   struct UDPMessageWrapper *udpw = plugin->msg_head;
1498   const struct sockaddr * sa = udpw->session->sock_addr;
1499
1500   max = GNUNET_TIME_absolute_max(udpw->timeout, GNUNET_TIME_absolute_get());
1501
1502   while (udpw != NULL)
1503   {
1504     if (max.abs_value != udpw->timeout.abs_value)
1505     {
1506       /* Message timed out */
1507
1508       if (udpw->cont != NULL)
1509         udpw->cont (udpw->cont_cls, &udpw->session->target, GNUNET_SYSERR);
1510       if (udpw->frag_ctx != NULL)
1511       {
1512 #if DEBUG_UDP
1513         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Fragmented message for peer `%s' with size %u timed out\n",
1514             GNUNET_i2s(&udpw->session->target), udpw->frag_ctx->bytes_to_send);
1515 #endif
1516         GNUNET_FRAGMENT_context_destroy(udpw->frag_ctx->frag);
1517       }
1518       else
1519       {
1520 #if DEBUG_UDP
1521         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Message for peer `%s' with size %u timed out\n",
1522             GNUNET_i2s(&udpw->session->target), udpw->msg_size);
1523 #endif
1524       }
1525
1526       GNUNET_CONTAINER_DLL_remove(plugin->msg_head, plugin->msg_tail, udpw);
1527       GNUNET_free (udpw);
1528       udpw = plugin->msg_head;
1529     }
1530     else
1531     {
1532       struct GNUNET_TIME_Relative delta = GNUNET_TIME_absolute_get_remaining (udpw->session->flow_delay_from_other_peer);
1533       if (delta.rel_value == 0)
1534       {
1535         /* this message is not delayed */
1536         GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Message is not delayed for %llu \n",
1537             delta);
1538         break;
1539       }
1540       else
1541       {
1542         /* this message is delayed, try next */
1543         GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Message is delayed for %llu \n",
1544             delta);
1545         udpw = udpw->next;
1546       }
1547     }
1548
1549   }
1550
1551   if (udpw == NULL)
1552   {
1553     /* No message left */
1554     return 0;
1555   }
1556
1557   switch (sa->sa_family)
1558   {
1559   case AF_INET:
1560     if (NULL == plugin->sockv4)
1561       return 0;
1562     sent =
1563         GNUNET_NETWORK_socket_sendto (plugin->sockv4, udpw->udp, udpw->msg_size,
1564                                       sa, slen = sizeof (struct sockaddr_in));
1565     break;
1566   case AF_INET6:
1567     if (NULL == plugin->sockv6)
1568       return 0;
1569     sent =
1570         GNUNET_NETWORK_socket_sendto (plugin->sockv6, udpw->udp, udpw->msg_size,
1571                                       sa, slen = sizeof (struct sockaddr_in6));
1572     break;
1573   default:
1574     GNUNET_break (0);
1575     return 0;
1576   }
1577   if (GNUNET_SYSERR == sent)
1578   {
1579     GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR, "sendto");
1580     LOG (GNUNET_ERROR_TYPE_DEBUG,
1581          "UDP transmitted %u-byte message to %s (%d: %s)\n",
1582          (unsigned int) (udpw->msg_size), GNUNET_a2s (sa, slen), (int) sent,
1583          (sent < 0) ? STRERROR (errno) : "ok");
1584     if (udpw->cont != NULL)
1585       udpw->cont (udpw->cont_cls, &udpw->session->target, GNUNET_SYSERR);
1586   }
1587   LOG (GNUNET_ERROR_TYPE_DEBUG,
1588        "UDP transmitted %u-byte message to %s (%d: %s)\n",
1589        (unsigned int) (udpw->msg_size), GNUNET_a2s (sa, slen), (int) sent,
1590        (sent < 0) ? STRERROR (errno) : "ok");
1591
1592   /* This was just a message fragment */
1593   if (udpw->frag_ctx != NULL)
1594   {
1595     GNUNET_FRAGMENT_context_transmission_done (udpw->frag_ctx->frag);
1596   }
1597   /* This was a complete message*/
1598   else
1599   {
1600     if (udpw->cont != NULL)
1601       udpw->cont (udpw->cont_cls, &udpw->session->target, GNUNET_OK);
1602   }
1603
1604   GNUNET_CONTAINER_DLL_remove(plugin->msg_head, plugin->msg_tail, udpw);
1605   GNUNET_free (udpw);
1606
1607   return sent;
1608 }
1609
1610 /**
1611  * We have been notified that our readset has something to read.  We don't
1612  * know which socket needs to be read, so we have to check each one
1613  * Then reschedule this function to be called again once more is available.
1614  *
1615  * @param cls the plugin handle
1616  * @param tc the scheduling context (for rescheduling this function again)
1617  */
1618 static void
1619 udp_plugin_select (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1620 {
1621   struct Plugin *plugin = cls;
1622
1623   plugin->select_task = GNUNET_SCHEDULER_NO_TASK;
1624   if ((tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN) != 0)
1625     return;
1626
1627   if ((tc->reason & GNUNET_SCHEDULER_REASON_READ_READY) != 0)
1628   {
1629     if ((NULL != plugin->sockv4) &&
1630       (GNUNET_NETWORK_fdset_isset (tc->read_ready, plugin->sockv4)))
1631         udp_select_read (plugin, plugin->sockv4);
1632     if ((NULL != plugin->sockv6) &&
1633       (GNUNET_NETWORK_fdset_isset (tc->read_ready, plugin->sockv6)))
1634         udp_select_read (plugin, plugin->sockv6);
1635   }
1636
1637   if ((tc->reason & GNUNET_SCHEDULER_REASON_WRITE_READY) != 0)
1638   {
1639     if (plugin->msg_head != NULL)
1640       udp_select_send (plugin);
1641   }
1642
1643   plugin->select_task = GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_DEFAULT,
1644                                    GNUNET_SCHEDULER_NO_TASK,
1645                                    GNUNET_TIME_UNIT_FOREVER_REL, plugin->rs,
1646                                    plugin->ws, &udp_plugin_select, plugin);
1647
1648 }
1649
1650
1651 static int
1652 setup_sockets (struct Plugin *plugin, struct sockaddr_in6 *serverAddrv6, struct sockaddr_in *serverAddrv4)
1653 {
1654   int tries;
1655   int sockets_created = 0;
1656   struct sockaddr *serverAddr;
1657   struct sockaddr *addrs[2];
1658   socklen_t addrlens[2];
1659   socklen_t addrlen;
1660
1661   /* Create IPv6 socket */
1662   if (plugin->enable_ipv6 == GNUNET_YES)
1663   {
1664     plugin->sockv6 = GNUNET_NETWORK_socket_create (PF_INET6, SOCK_DGRAM, 0);
1665     if (NULL == plugin->sockv6)
1666     {
1667       GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING, "socket");
1668     }
1669     else
1670     {
1671 #if HAVE_SOCKADDR_IN_SIN_LEN
1672       serverAddrv6.sin6_len = sizeof (serverAddrv6);
1673 #endif
1674       serverAddrv6->sin6_family = AF_INET6;
1675       serverAddrv6->sin6_addr = in6addr_any;
1676       serverAddrv6->sin6_port = htons (plugin->port);
1677       addrlen = sizeof (struct sockaddr_in6);
1678       serverAddr = (struct sockaddr *) serverAddrv6;
1679 #if DEBUG_UDP
1680       LOG (GNUNET_ERROR_TYPE_DEBUG, "Binding to IPv6 port %d\n",
1681            ntohs (serverAddrv6->sin6_port));
1682 #endif
1683       tries = 0;
1684       while (GNUNET_NETWORK_socket_bind (plugin->sockv6, serverAddr, addrlen) !=
1685              GNUNET_OK)
1686       {
1687         serverAddrv6->sin6_port = htons (GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_STRONG, 33537) + 32000);        /* Find a good, non-root port */
1688 #if DEBUG_UDP
1689         LOG (GNUNET_ERROR_TYPE_DEBUG,
1690              "IPv6 Binding failed, trying new port %d\n",
1691              ntohs (serverAddrv6->sin6_port));
1692 #endif
1693         tries++;
1694         if (tries > 10)
1695         {
1696           GNUNET_NETWORK_socket_close (plugin->sockv6);
1697           plugin->sockv6 = NULL;
1698           break;
1699         }
1700       }
1701       if (plugin->sockv6 != NULL)
1702       {
1703 #if DEBUG_UDP
1704         LOG (GNUNET_ERROR_TYPE_DEBUG,
1705              "IPv6 socket created on port %d\n",
1706              ntohs (serverAddrv6->sin6_port));
1707 #endif
1708         addrs[sockets_created] = (struct sockaddr *) serverAddrv6;
1709         addrlens[sockets_created] = sizeof (struct sockaddr_in6);
1710         sockets_created++;
1711       }
1712     }
1713   }
1714
1715   /* Create IPv4 socket */
1716   plugin->sockv4 = GNUNET_NETWORK_socket_create (PF_INET, SOCK_DGRAM, 0);
1717   if (NULL == plugin->sockv4)
1718   {
1719     GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING, "socket");
1720   }
1721   else
1722   {
1723 #if HAVE_SOCKADDR_IN_SIN_LEN
1724     serverAddrv4.sin_len = sizeof (serverAddrv4);
1725 #endif
1726     serverAddrv4->sin_family = AF_INET;
1727     serverAddrv4->sin_addr.s_addr = INADDR_ANY;
1728     serverAddrv4->sin_port = htons (plugin->port);
1729     addrlen = sizeof (struct sockaddr_in);
1730     serverAddr = (struct sockaddr *) serverAddrv4;
1731
1732 #if DEBUG_UDP
1733     LOG (GNUNET_ERROR_TYPE_DEBUG, "Binding to IPv4 port %d\n",
1734          ntohs (serverAddrv4->sin_port));
1735 #endif
1736     tries = 0;
1737     while (GNUNET_NETWORK_socket_bind (plugin->sockv4, serverAddr, addrlen) !=
1738            GNUNET_OK)
1739     {
1740       serverAddrv4->sin_port = htons (GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_STRONG, 33537) + 32000);   /* Find a good, non-root port */
1741 #if DEBUG_UDP
1742       LOG (GNUNET_ERROR_TYPE_DEBUG, "IPv4 Binding failed, trying new port %d\n",
1743            ntohs (serverAddrv4->sin_port));
1744 #endif
1745       tries++;
1746       if (tries > 10)
1747       {
1748         GNUNET_NETWORK_socket_close (plugin->sockv4);
1749         plugin->sockv4 = NULL;
1750         break;
1751       }
1752     }
1753     if (plugin->sockv4 != NULL)
1754     {
1755       addrs[sockets_created] = (struct sockaddr *) serverAddrv4;
1756       addrlens[sockets_created] = sizeof (struct sockaddr_in);
1757       sockets_created++;
1758     }
1759   }
1760
1761   /* Create file descriptors */
1762   plugin->rs = GNUNET_NETWORK_fdset_create ();
1763   plugin->ws = GNUNET_NETWORK_fdset_create ();
1764   GNUNET_NETWORK_fdset_zero (plugin->rs);
1765   GNUNET_NETWORK_fdset_zero (plugin->ws);
1766   if (NULL != plugin->sockv4)
1767   {
1768     GNUNET_NETWORK_fdset_set (plugin->rs, plugin->sockv4);
1769     GNUNET_NETWORK_fdset_set (plugin->ws, plugin->sockv4);
1770   }
1771   if (NULL != plugin->sockv6)
1772   {
1773     GNUNET_NETWORK_fdset_set (plugin->rs, plugin->sockv6);
1774     GNUNET_NETWORK_fdset_set (plugin->ws, plugin->sockv6);
1775   }
1776
1777   if (sockets_created == 0)
1778     GNUNET_log (GNUNET_ERROR_TYPE_WARNING, _("Failed to open UDP sockets\n"));
1779
1780   plugin->select_task =
1781       GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_DEFAULT,
1782                                    GNUNET_SCHEDULER_NO_TASK,
1783                                    GNUNET_TIME_UNIT_FOREVER_REL, plugin->rs,
1784                                    plugin->ws, &udp_plugin_select, plugin);
1785
1786   plugin->nat = GNUNET_NAT_register (plugin->env->cfg,
1787                            GNUNET_NO, plugin->port,
1788                            sockets_created,
1789                            (const struct sockaddr **) addrs, addrlens,
1790                            &udp_nat_port_map_callback, NULL, plugin);
1791
1792   return sockets_created;
1793 }
1794
1795
1796 /**
1797  * The exported method. Makes the core api available via a global and
1798  * returns the udp transport API.
1799  *
1800  * @param cls our 'struct GNUNET_TRANSPORT_PluginEnvironment'
1801  * @return our 'struct GNUNET_TRANSPORT_PluginFunctions'
1802  */
1803 void *
1804 libgnunet_plugin_transport_udp_init (void *cls)
1805 {
1806   struct GNUNET_TRANSPORT_PluginEnvironment *env = cls;
1807   struct GNUNET_TRANSPORT_PluginFunctions *api;
1808   struct Plugin *plugin;
1809
1810   unsigned long long port;
1811   unsigned long long aport;
1812   unsigned long long broadcast;
1813   unsigned long long udp_max_bps;
1814   unsigned long long enable_v6;
1815   char * bind4_address;
1816   char * bind6_address;
1817   struct GNUNET_TIME_Relative interval;
1818
1819   struct sockaddr_in serverAddrv4;
1820   struct sockaddr_in6 serverAddrv6;
1821
1822   int res;
1823
1824   /* Get port number */
1825   if (GNUNET_OK !=
1826       GNUNET_CONFIGURATION_get_value_number (env->cfg, "transport-udp", "PORT",
1827                                              &port))
1828     port = 2086;
1829   if (GNUNET_OK !=
1830       GNUNET_CONFIGURATION_get_value_number (env->cfg, "transport-udp",
1831                                              "ADVERTISED_PORT", &aport))
1832     aport = port;
1833   if (port > 65535)
1834   {
1835     LOG (GNUNET_ERROR_TYPE_WARNING,
1836          _("Given `%s' option is out of range: %llu > %u\n"), "PORT", port,
1837          65535);
1838     return NULL;
1839   }
1840
1841   /* Protocols */
1842   if ((GNUNET_YES ==
1843        GNUNET_CONFIGURATION_get_value_yesno (env->cfg, "nat",
1844                                              "DISABLEV6")))
1845   {
1846     enable_v6 = GNUNET_NO;
1847   }
1848   else
1849     enable_v6 = GNUNET_YES;
1850
1851
1852   /* Addresses */
1853   memset (&serverAddrv6, 0, sizeof (serverAddrv6));
1854   memset (&serverAddrv4, 0, sizeof (serverAddrv4));
1855
1856   if (GNUNET_YES ==
1857       GNUNET_CONFIGURATION_get_value_string (env->cfg, "transport-udp",
1858                                              "BINDTO", &bind4_address))
1859   {
1860     LOG (GNUNET_ERROR_TYPE_DEBUG,
1861          "Binding udp plugin to specific address: `%s'\n",
1862          bind4_address);
1863     if (1 != inet_pton (AF_INET, bind4_address, &serverAddrv4.sin_addr))
1864     {
1865       GNUNET_free (bind4_address);
1866       return NULL;
1867     }
1868   }
1869
1870   if (GNUNET_YES ==
1871       GNUNET_CONFIGURATION_get_value_string (env->cfg, "transport-udp",
1872                                              "BINDTO6", &bind6_address))
1873   {
1874     LOG (GNUNET_ERROR_TYPE_DEBUG,
1875          "Binding udp plugin to specific address: `%s'\n",
1876          bind6_address);
1877     if (1 !=
1878         inet_pton (AF_INET6, bind6_address, &serverAddrv6.sin6_addr))
1879     {
1880       LOG (GNUNET_ERROR_TYPE_ERROR, _("Invalid IPv6 address: `%s'\n"),
1881            bind6_address);
1882       GNUNET_free_non_null (bind4_address);
1883       GNUNET_free (bind6_address);
1884       return NULL;
1885     }
1886   }
1887
1888
1889   /* Enable neighbour discovery */
1890   broadcast = GNUNET_CONFIGURATION_get_value_yesno (env->cfg, "transport-udp",
1891                                             "BROADCAST");
1892   if (broadcast == GNUNET_SYSERR)
1893     broadcast = GNUNET_NO;
1894
1895   if (GNUNET_SYSERR == GNUNET_CONFIGURATION_get_value_time (env->cfg, "transport-udp",
1896                                            "BROADCAST_INTERVAL", &interval))
1897   {
1898     interval = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 10);
1899   }
1900
1901   /* Maximum datarate */
1902   if (GNUNET_OK != GNUNET_CONFIGURATION_get_value_number (env->cfg, "transport-udp",
1903                                              "MAX_BPS", &udp_max_bps))
1904   {
1905     udp_max_bps = 1024 * 1024 * 50;     /* 50 MB/s == infinity for practical purposes */
1906   }
1907
1908   plugin = GNUNET_malloc (sizeof (struct Plugin));
1909   api = GNUNET_malloc (sizeof (struct GNUNET_TRANSPORT_PluginFunctions));
1910
1911   GNUNET_BANDWIDTH_tracker_init (&plugin->tracker,
1912                                  GNUNET_BANDWIDTH_value_init ((uint32_t)udp_max_bps), 30);
1913
1914
1915   plugin->sessions = GNUNET_CONTAINER_multihashmap_create (10);
1916   plugin->defrag_ctxs = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
1917   plugin->mst = GNUNET_SERVER_mst_create (&process_inbound_tokenized_messages, plugin);
1918   plugin->port = port;
1919   plugin->aport = aport;
1920   plugin->last_expected_delay = GNUNET_TIME_UNIT_SECONDS;
1921   plugin->broadcast_interval = interval;
1922   plugin->enable_ipv6 = enable_v6;
1923   plugin->env = env;
1924
1925   api->cls = plugin;
1926   api->send = NULL;
1927   api->disconnect = &udp_disconnect;
1928   api->address_pretty_printer = &udp_plugin_address_pretty_printer;
1929   api->address_to_string = &udp_address_to_string;
1930   api->check_address = &udp_plugin_check_address;
1931   api->get_session = &udp_plugin_get_session;
1932   api->send = &udp_plugin_send_wrapper;
1933   api->send_with_session = &udp_plugin_send;
1934
1935   LOG (GNUNET_ERROR_TYPE_DEBUG, "Setting up sockets\n");
1936   res = setup_sockets (plugin, &serverAddrv6, &serverAddrv4);
1937   if ((res == 0) || ((plugin->sockv4 == NULL) && (plugin->sockv6 == NULL)))
1938   {
1939     LOG (GNUNET_ERROR_TYPE_ERROR, "Failed to create network sockets, plugin failed\n");
1940     GNUNET_free (plugin);
1941     GNUNET_free (api);
1942     return NULL;
1943   }
1944
1945   LOG (GNUNET_ERROR_TYPE_DEBUG, "Starting broadcasting\n");
1946   if (broadcast == GNUNET_YES)
1947     setup_broadcast (plugin, &serverAddrv6, &serverAddrv4);
1948
1949
1950   GNUNET_free_non_null (bind4_address);
1951   GNUNET_free_non_null (bind6_address);
1952   return api;
1953 }
1954
1955 int heap_cleanup_iterator (void *cls,
1956                           struct GNUNET_CONTAINER_HeapNode *
1957                           node, void *element,
1958                           GNUNET_CONTAINER_HeapCostType
1959                           cost)
1960 {
1961   struct DefragContext * d_ctx = element;
1962
1963   GNUNET_CONTAINER_heap_remove_node (node);
1964   GNUNET_DEFRAGMENT_context_destroy(d_ctx->defrag);
1965   GNUNET_free (d_ctx);
1966
1967   return GNUNET_YES;
1968 }
1969
1970
1971 /**
1972  * The exported method. Makes the core api available via a global and
1973  * returns the udp transport API.
1974  *
1975  * @param cls our 'struct GNUNET_TRANSPORT_PluginEnvironment'
1976  * @return our 'struct GNUNET_TRANSPORT_PluginFunctions'
1977  */
1978 void *
1979 libgnunet_plugin_transport_udp_done (void *cls)
1980 {
1981   struct GNUNET_TRANSPORT_PluginFunctions *api = cls;
1982   struct Plugin *plugin = api->cls;
1983   stop_broadcast (plugin);
1984
1985   if (plugin->select_task != GNUNET_SCHEDULER_NO_TASK)
1986   {
1987     GNUNET_SCHEDULER_cancel (plugin->select_task);
1988     plugin->select_task = GNUNET_SCHEDULER_NO_TASK;
1989   }
1990
1991   /* Closing sockets */
1992   if (plugin->sockv4 != NULL)
1993   {
1994     GNUNET_break (GNUNET_OK == GNUNET_NETWORK_socket_close (plugin->sockv4));
1995     plugin->sockv4 = NULL;
1996   }
1997   if (plugin->sockv6 != NULL)
1998   {
1999     GNUNET_break (GNUNET_OK == GNUNET_NETWORK_socket_close (plugin->sockv6));
2000     plugin->sockv6 = NULL;
2001   }
2002   GNUNET_NETWORK_fdset_destroy (plugin->rs);
2003   GNUNET_NETWORK_fdset_destroy (plugin->ws);
2004   GNUNET_NAT_unregister (plugin->nat);
2005
2006   if (plugin->defrag_ctxs != NULL)
2007   {
2008     GNUNET_CONTAINER_heap_iterate(plugin->defrag_ctxs,
2009         heap_cleanup_iterator, NULL);
2010     GNUNET_CONTAINER_heap_destroy(plugin->defrag_ctxs);
2011     plugin->defrag_ctxs = NULL;
2012   }
2013   if (plugin->mst != NULL)
2014   {
2015     GNUNET_SERVER_mst_destroy(plugin->mst);
2016     plugin->mst = NULL;
2017   }
2018
2019   /* Clean up leftover messages */
2020   struct UDPMessageWrapper *udpw = plugin->msg_head;
2021   while (udpw != NULL)
2022   {
2023     struct UDPMessageWrapper *tmp = udpw->next;
2024     GNUNET_CONTAINER_DLL_remove(plugin->msg_head, plugin->msg_tail, udpw);
2025     if (udpw->cont != NULL)
2026       udpw->cont (udpw->cont_cls, &udpw->session->target, GNUNET_SYSERR);
2027     GNUNET_free (udpw);
2028     udpw = tmp;
2029   }
2030
2031   /* Clean up sessions */
2032 #if DEBUG_UDP
2033   LOG (GNUNET_ERROR_TYPE_DEBUG,
2034        "Cleaning up sessions\n");
2035 #endif
2036   GNUNET_CONTAINER_multihashmap_iterate (plugin->sessions, &disconnect_and_free_it, plugin);
2037   GNUNET_CONTAINER_multihashmap_destroy (plugin->sessions);
2038
2039   plugin->nat = NULL;
2040   GNUNET_free (plugin);
2041   GNUNET_free (api);
2042   return NULL;
2043 }
2044
2045
2046 /* end of plugin_transport_udp.c */