src/transport/plugin_transport_udp.c

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