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 if (0 == addrlen)
 518   {
 519     asc (asc_cls, "<inbound connection>");
 520     asc (asc_cls, NULL);
 521     return;
 522   }
 523   else
 524   {
 525     /* invalid address */
 526     GNUNET_break_op (0);
 527     asc (asc_cls, NULL);
 528     return;
 529   }
 530   ppc = GNUNET_malloc (sizeof (struct PrettyPrinterContext));
 531   ppc->asc = asc;
 532   ppc->asc_cls = asc_cls;
 533   ppc->port = port;
 534   GNUNET_RESOLVER_hostname_get (sb, sbs, !numeric, timeout, &append_port, ppc);
 535 }
 536
 537
 538 /**
 539  * Check if the given port is plausible (must be either our listen
 540  * port or our advertised port).  If it is neither, we return
 541  * GNUNET_SYSERR.
 542  *
 543  * @param plugin global variables
 544  * @param in_port port number to check
 545  * @return GNUNET_OK if port is either open_port or adv_port
 546  */
 547 static int
 548 check_port (struct Plugin *plugin, uint16_t in_port)
 549 {
 550   if ((in_port == plugin->port) || (in_port == plugin->aport))
 551     return GNUNET_OK;
 552   return GNUNET_SYSERR;
 553 }
 554
 555
 556
 557 /**
 558  * Function that will be called to check if a binary address for this
 559  * plugin is well-formed and corresponds to an address for THIS peer
 560  * (as per our configuration).  Naturally, if absolutely necessary,
 561  * plugins can be a bit conservative in their answer, but in general
 562  * plugins should make sure that the address does not redirect
 563  * traffic to a 3rd party that might try to man-in-the-middle our
 564  * traffic.
 565  *
 566  * @param cls closure, should be our handle to the Plugin
 567  * @param addr pointer to the address
 568  * @param addrlen length of addr
 569  * @return GNUNET_OK if this is a plausible address for this peer
 570  *         and transport, GNUNET_SYSERR if not
 571  *
 572  */
 573 static int
 574 udp_plugin_check_address (void *cls, const void *addr, size_t addrlen)
 575 {
 576   struct Plugin *plugin = cls;
 577   struct IPv4UdpAddress *v4;
 578   struct IPv6UdpAddress *v6;
 579
 580   if ((addrlen != sizeof (struct IPv4UdpAddress)) &&
 581       (addrlen != sizeof (struct IPv6UdpAddress)))
 582   {
 583     GNUNET_break_op (0);
 584     return GNUNET_SYSERR;
 585   }
 586   if (addrlen == sizeof (struct IPv4UdpAddress))
 587   {
 588     v4 = (struct IPv4UdpAddress *) addr;
 589     if (GNUNET_OK != check_port (plugin, ntohs (v4->u4_port)))
 590       return GNUNET_SYSERR;
 591     if (GNUNET_OK !=
 592         GNUNET_NAT_test_address (plugin->nat, &v4->ipv4_addr,
 593                                  sizeof (struct in_addr)))
 594       return GNUNET_SYSERR;
 595   }
 596   else
 597   {
 598     v6 = (struct IPv6UdpAddress *) addr;
 599     if (IN6_IS_ADDR_LINKLOCAL (&v6->ipv6_addr))
 600     {
 601       GNUNET_break_op (0);
 602       return GNUNET_SYSERR;
 603     }
 604     if (GNUNET_OK != check_port (plugin, ntohs (v6->u6_port)))
 605       return GNUNET_SYSERR;
 606     if (GNUNET_OK !=
 607         GNUNET_NAT_test_address (plugin->nat, &v6->ipv6_addr,
 608                                  sizeof (struct in6_addr)))
 609       return GNUNET_SYSERR;
 610   }
 611   return GNUNET_OK;
 612 }
 613
 614
 615 /**
 616  * Destroy a session, plugin is being unloaded.
 617  *
 618  * @param cls unused
 619  * @param key hash of public key of target peer
 620  * @param value a 'struct PeerSession*' to clean up
 621  * @return GNUNET_OK (continue to iterate)
 622  */
 623 static int
 624 disconnect_and_free_it (void *cls, const GNUNET_HashCode * key, void *value)
 625 {
 626   struct Plugin *plugin = cls;
 627   struct Session *s = value;
 628   struct UDPMessageWrapper *udpw;
 629   struct UDPMessageWrapper *next;
 630
 631   LOG (GNUNET_ERROR_TYPE_DEBUG,
 632        "Session %p to peer `%s' address ended \n",
 633          s,
 634          GNUNET_i2s (&s->target),
 635          GNUNET_a2s (s->sock_addr, s->addrlen));
 636   if (s->frag_ctx != NULL)
 637   {
 638     GNUNET_FRAGMENT_context_destroy(s->frag_ctx->frag);
 639     GNUNET_free (s->frag_ctx);
 640     s->frag_ctx = NULL;
 641   }
 642
 643   udpw = plugin->ipv4_queue_head;
 644   while (udpw != NULL)
 645   {
 646     next = udpw->next;
 647     if (udpw->session == s)
 648     {
 649       GNUNET_CONTAINER_DLL_remove(plugin->ipv4_queue_head, plugin->ipv4_queue_tail, udpw);
 650
 651       if (udpw->cont != NULL)
 652         udpw->cont (udpw->cont_cls, &s->target, GNUNET_SYSERR);
 653       GNUNET_free (udpw);
 654     }
 655     udpw = next;
 656   }
 657
 658   udpw = plugin->ipv6_queue_head;
 659   while (udpw != NULL)
 660   {
 661     next = udpw->next;
 662     if (udpw->session == s)
 663     {
 664       GNUNET_CONTAINER_DLL_remove(plugin->ipv6_queue_head, plugin->ipv6_queue_tail, udpw);
 665
 666       if (udpw->cont != NULL)
 667         udpw->cont (udpw->cont_cls, &s->target, GNUNET_SYSERR);
 668       GNUNET_free (udpw);
 669     }
 670     udpw = next;
 671   }
 672
 673   plugin->env->session_end (plugin->env->cls, &s->target, s);
 674
 675   GNUNET_assert (GNUNET_YES ==
 676                  GNUNET_CONTAINER_multihashmap_remove (plugin->sessions,
 677                                                        &s->target.hashPubKey,
 678                                                        s));
 679
 680   GNUNET_STATISTICS_set(plugin->env->stats,
 681                         "# UDP sessions active",
 682                         GNUNET_CONTAINER_multihashmap_size(plugin->sessions),
 683                         GNUNET_NO);
 684
 685   GNUNET_free (s);
 686   return GNUNET_OK;
 687 }
 688
 689
 690 /**
 691  * Disconnect from a remote node.  Clean up session if we have one for this peer
 692  *
 693  * @param cls closure for this call (should be handle to Plugin)
 694  * @param target the peeridentity of the peer to disconnect
 695  * @return GNUNET_OK on success, GNUNET_SYSERR if the operation failed
 696  */
 697 static void
 698 udp_disconnect (void *cls, const struct GNUNET_PeerIdentity *target)
 699 {
 700   struct Plugin *plugin = cls;
 701   GNUNET_assert (plugin != NULL);
 702
 703   GNUNET_assert (target != NULL);
 704   LOG (GNUNET_ERROR_TYPE_DEBUG,
 705        "Disconnecting from peer `%s'\n", GNUNET_i2s (target));
 706   /* Clean up sessions */
 707   GNUNET_CONTAINER_multihashmap_get_multiple (plugin->sessions, &target->hashPubKey, &disconnect_and_free_it, plugin);
 708 }
 709
 710 static struct Session *
 711 create_session (struct Plugin *plugin, const struct GNUNET_PeerIdentity *target,
 712                 const void *addr, size_t addrlen,
 713                 GNUNET_TRANSPORT_TransmitContinuation cont, void *cont_cls)
 714 {
 715   struct Session *s;
 716   const struct IPv4UdpAddress *t4;
 717   const struct IPv6UdpAddress *t6;
 718   struct sockaddr_in *v4;
 719   struct sockaddr_in6 *v6;
 720   size_t len;
 721
 722   switch (addrlen)
 723   {
 724   case sizeof (struct IPv4UdpAddress):
 725     if (NULL == plugin->sockv4)
 726     {
 727       return NULL;
 728     }
 729     t4 = addr;
 730     s = GNUNET_malloc (sizeof (struct Session) + sizeof (struct sockaddr_in));
 731     len = sizeof (struct sockaddr_in);
 732     v4 = (struct sockaddr_in *) &s[1];
 733     v4->sin_family = AF_INET;
 734 #if HAVE_SOCKADDR_IN_SIN_LEN
 735     v4->sin_len = sizeof (struct sockaddr_in);
 736 #endif
 737     v4->sin_port = t4->u4_port;
 738     v4->sin_addr.s_addr = t4->ipv4_addr;
 739     s->ats = plugin->env->get_address_type (plugin->env->cls, (const struct sockaddr *) v4, sizeof (struct sockaddr_in));
 740     break;
 741   case sizeof (struct IPv6UdpAddress):
 742     if (NULL == plugin->sockv6)
 743     {
 744       return NULL;
 745     }
 746     t6 = addr;
 747     s =
 748         GNUNET_malloc (sizeof (struct Session) + sizeof (struct sockaddr_in6));
 749     len = sizeof (struct sockaddr_in6);
 750     v6 = (struct sockaddr_in6 *) &s[1];
 751     v6->sin6_family = AF_INET6;
 752 #if HAVE_SOCKADDR_IN_SIN_LEN
 753     v6->sin6_len = sizeof (struct sockaddr_in6);
 754 #endif
 755     v6->sin6_port = t6->u6_port;
 756     v6->sin6_addr = t6->ipv6_addr;
 757     s->ats = plugin->env->get_address_type (plugin->env->cls, (const struct sockaddr *) v6, sizeof (struct sockaddr_in6));
 758     break;
 759   default:
 760     /* Must have a valid address to send to */
 761     GNUNET_break_op (0);
 762     return NULL;
 763   }
 764
 765   s->addrlen = len;
 766   s->target = *target;
 767   s->sock_addr = (const struct sockaddr *) &s[1];
 768   s->flow_delay_for_other_peer = GNUNET_TIME_relative_get_zero();
 769   s->flow_delay_from_other_peer = GNUNET_TIME_absolute_get_zero();
 770   s->last_expected_delay = GNUNET_TIME_UNIT_SECONDS;
 771
 772   return s;
 773 }
 774
 775 static int session_cmp_it (void *cls,
 776                            const GNUNET_HashCode * key,
 777                            void *value)
 778 {
 779   struct SessionCompareContext * cctx = cls;
 780   const struct GNUNET_HELLO_Address *address = cctx->addr;
 781   struct Session *s = value;
 782
 783   socklen_t s_addrlen = s->addrlen;
 784
 785 #if VERBOSE_UDP
 786   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Comparing  address %s <-> %s\n",
 787       udp_address_to_string (NULL, (void *) address->address, address->address_length),
 788       GNUNET_a2s (s->sock_addr, s->addrlen));
 789 #endif
 790
 791   if ((address->address_length == sizeof (struct IPv4UdpAddress)) &&
 792       (s_addrlen == sizeof (struct sockaddr_in)))
 793   {
 794     struct IPv4UdpAddress * u4 = NULL;
 795     u4 = (struct IPv4UdpAddress *) address->address;
 796     const struct sockaddr_in *s4 = (const struct sockaddr_in *) s->sock_addr;
 797     if ((0 == memcmp ((const void *) &u4->ipv4_addr,(const void *) &s4->sin_addr, sizeof (struct in_addr))) &&
 798         (u4->u4_port == s4->sin_port))
 799     {
 800       cctx->res = s;
 801       return GNUNET_NO;
 802     }
 803
 804   }
 805   if ((address->address_length == sizeof (struct IPv6UdpAddress)) &&
 806       (s_addrlen == sizeof (struct sockaddr_in6)))
 807   {
 808     struct IPv6UdpAddress * u6 = NULL;
 809     u6 = (struct IPv6UdpAddress *) address->address;
 810     const struct sockaddr_in6 *s6 = (const struct sockaddr_in6 *) s->sock_addr;
 811     if ((0 == memcmp (&u6->ipv6_addr, &s6->sin6_addr, sizeof (struct in6_addr))) &&
 812         (u6->u6_port == s6->sin6_port))
 813     {
 814       cctx->res = s;
 815       return GNUNET_NO;
 816     }
 817   }
 818
 819
 820   return GNUNET_YES;
 821 }
 822
 823
 824 /**
 825  * Creates a new outbound session the transport service will use to send data to the
 826  * peer
 827  *
 828  * @param cls the plugin
 829  * @param address the address
 830  * @return the session or NULL of max connections exceeded
 831  */
 832 static struct Session *
 833 udp_plugin_get_session (void *cls,
 834                   const struct GNUNET_HELLO_Address *address)
 835 {
 836   struct Session * s = NULL;
 837   struct Plugin * plugin = cls;
 838   struct IPv6UdpAddress * udp_a6;
 839   struct IPv4UdpAddress * udp_a4;
 840
 841   GNUNET_assert (plugin != NULL);
 842   GNUNET_assert (address != NULL);
 843
 844
 845   if ((address->address == NULL) ||
 846       ((address->address_length != sizeof (struct IPv4UdpAddress)) &&
 847       (address->address_length != sizeof (struct IPv6UdpAddress))))
 848   {
 849     GNUNET_break (0);
 850     return NULL;
 851   }
 852
 853   if (address->address_length == sizeof (struct IPv4UdpAddress))
 854   {
 855     if (plugin->sockv4 == NULL)
 856       return NULL;
 857     udp_a4 = (struct IPv4UdpAddress *) address->address;
 858     if (udp_a4->u4_port == 0)
 859       return NULL;
 860   }
 861
 862   if (address->address_length == sizeof (struct IPv6UdpAddress))
 863   {
 864     if (plugin->sockv6 == NULL)
 865       return NULL;
 866     udp_a6 = (struct IPv6UdpAddress *) address->address;
 867     if (udp_a6->u6_port == 0)
 868       return NULL;
 869   }
 870
 871   /* check if session already exists */
 872   struct SessionCompareContext cctx;
 873   cctx.addr = address;
 874   cctx.res = NULL;
 875 #if VERBOSE_UDP
 876   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));
 877 #endif
 878   GNUNET_CONTAINER_multihashmap_get_multiple(plugin->sessions, &address->peer.hashPubKey, session_cmp_it, &cctx);
 879   if (cctx.res != NULL)
 880   {
 881 #if VERBOSE_UDP
 882     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Found existing session %p\n", cctx.res);
 883 #endif
 884     return cctx.res;
 885   }
 886
 887   /* otherwise create new */
 888   s = create_session (plugin,
 889       &address->peer,
 890       address->address,
 891       address->address_length,
 892       NULL, NULL);
 893 #if VERBOSE
 894     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 895               "Creating new session %p for peer `%s' address `%s'\n",
 896               s,
 897               GNUNET_i2s(&address->peer),
 898               udp_address_to_string(NULL,address->address,address->address_length));
 899 #endif
 900   GNUNET_assert (GNUNET_OK ==
 901                  GNUNET_CONTAINER_multihashmap_put (plugin->sessions,
 902                                                     &s->target.hashPubKey,
 903                                                     s,
 904                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
 905
 906   GNUNET_STATISTICS_set(plugin->env->stats,
 907                         "# UDP sessions active",
 908                         GNUNET_CONTAINER_multihashmap_size(plugin->sessions),
 909                         GNUNET_NO);
 910
 911   return s;
 912 }
 913
 914 static void enqueue (struct Plugin *plugin, struct UDPMessageWrapper * udpw)
 915 {
 916
 917   if (udpw->session->addrlen == sizeof (struct sockaddr_in))
 918     GNUNET_CONTAINER_DLL_insert(plugin->ipv4_queue_head, plugin->ipv4_queue_tail, udpw);
 919   if (udpw->session->addrlen == sizeof (struct sockaddr_in6))
 920     GNUNET_CONTAINER_DLL_insert(plugin->ipv6_queue_head, plugin->ipv6_queue_tail, udpw);
 921 }
 922
 923 /**
 924  * Function that is called with messages created by the fragmentation
 925  * module.  In the case of the 'proc' callback of the
 926  * GNUNET_FRAGMENT_context_create function, this function must
 927  * eventually call 'GNUNET_FRAGMENT_context_transmission_done'.
 928  *
 929  * @param cls closure, the 'struct FragmentationContext'
 930  * @param msg the message that was created
 931  */
 932 static void
 933 enqueue_fragment (void *cls, const struct GNUNET_MessageHeader *msg)
 934 {
 935   struct FragmentationContext *frag_ctx = cls;
 936   struct Plugin *plugin = frag_ctx->plugin;
 937   struct UDPMessageWrapper * udpw;
 938   struct Session *s;
 939
 940   size_t msg_len = ntohs (msg->size);
 941
 942 #if VERBOSE_UDP
 943   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Enqueuing fragment with %u bytes %u\n", msg_len , sizeof (struct UDPMessageWrapper));
 944 #endif
 945
 946   udpw = GNUNET_malloc (sizeof (struct UDPMessageWrapper) + msg_len);
 947   udpw->session = frag_ctx->session;
 948   s = udpw->session;
 949   udpw->udp = (char *) &udpw[1];
 950
 951   udpw->msg_size = msg_len;
 952   udpw->cont = frag_ctx->cont;
 953   udpw->cont_cls = frag_ctx->cont_cls;
 954   udpw->timeout = frag_ctx->timeout;
 955   udpw->frag_ctx = frag_ctx;
 956   memcpy (udpw->udp, msg, msg_len);
 957
 958   enqueue (plugin, udpw);
 959
 960
 961   if (s->addrlen == sizeof (struct sockaddr_in))
 962   {
 963     if (plugin->with_v4_ws == GNUNET_NO)
 964     {
 965       if (plugin->select_task != GNUNET_SCHEDULER_NO_TASK)
 966         GNUNET_SCHEDULER_cancel(plugin->select_task);
 967
 968       plugin->select_task =
 969           GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_DEFAULT,
 970                                        GNUNET_TIME_UNIT_FOREVER_REL,
 971                                        plugin->rs_v4,
 972                                        plugin->ws_v4,
 973                                        &udp_plugin_select, plugin);
 974       plugin->with_v4_ws = GNUNET_YES;
 975     }
 976   }
 977
 978   else if (s->addrlen == sizeof (struct sockaddr_in6))
 979   {
 980     if (plugin->with_v6_ws == GNUNET_NO)
 981     {
 982       if (plugin->select_task_v6 != GNUNET_SCHEDULER_NO_TASK)
 983         GNUNET_SCHEDULER_cancel(plugin->select_task_v6);
 984
 985       plugin->select_task_v6 =
 986           GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_DEFAULT,
 987                                        GNUNET_TIME_UNIT_FOREVER_REL,
 988                                        plugin->rs_v6,
 989                                        plugin->ws_v6,
 990                                        &udp_plugin_select_v6, plugin);
 991       plugin->with_v6_ws = GNUNET_YES;
 992     }
 993   }
 994
 995 }
 996
 997
 998
 999
1000 /**
1001  * Function that can be used by the transport service to transmit
1002  * a message using the plugin.   Note that in the case of a
1003  * peer disconnecting, the continuation MUST be called
1004  * prior to the disconnect notification itself.  This function
1005  * will be called with this peer's HELLO message to initiate
1006  * a fresh connection to another peer.
1007  *
1008  * @param cls closure
1009  * @param s which session must be used
1010  * @param msgbuf the message to transmit
1011  * @param msgbuf_size number of bytes in 'msgbuf'
1012  * @param priority how important is the message (most plugins will
1013  *                 ignore message priority and just FIFO)
1014  * @param to how long to wait at most for the transmission (does not
1015  *                require plugins to discard the message after the timeout,
1016  *                just advisory for the desired delay; most plugins will ignore
1017  *                this as well)
1018  * @param cont continuation to call once the message has
1019  *        been transmitted (or if the transport is ready
1020  *        for the next transmission call; or if the
1021  *        peer disconnected...); can be NULL
1022  * @param cont_cls closure for cont
1023  * @return number of bytes used (on the physical network, with overheads);
1024  *         -1 on hard errors (i.e. address invalid); 0 is a legal value
1025  *         and does NOT mean that the message was not transmitted (DV)
1026  */
1027 static ssize_t
1028 udp_plugin_send (void *cls,
1029                   struct Session *s,
1030                   const char *msgbuf, size_t msgbuf_size,
1031                   unsigned int priority,
1032                   struct GNUNET_TIME_Relative to,
1033                   GNUNET_TRANSPORT_TransmitContinuation cont, void *cont_cls)
1034 {
1035   struct Plugin *plugin = cls;
1036   size_t mlen = msgbuf_size + sizeof (struct UDPMessage);
1037
1038   struct UDPMessageWrapper * udpw;
1039   struct UDPMessage *udp;
1040   char mbuf[mlen];
1041   GNUNET_assert (plugin != NULL);
1042   GNUNET_assert (s != NULL);
1043
1044   if ((s->addrlen == sizeof (struct sockaddr_in6)) && (plugin->sockv6 == NULL))
1045     return GNUNET_SYSERR;
1046
1047    if ((s->addrlen == sizeof (struct sockaddr_in)) && (plugin->sockv4 == NULL))
1048      return GNUNET_SYSERR;
1049
1050
1051   if (mlen >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
1052   {
1053     GNUNET_break (0);
1054     return GNUNET_SYSERR;
1055   }
1056
1057   if (GNUNET_YES != GNUNET_CONTAINER_multihashmap_contains_value(plugin->sessions, &s->target.hashPubKey, s))
1058   {
1059     GNUNET_break (0);
1060     return GNUNET_SYSERR;
1061   }
1062
1063   LOG (GNUNET_ERROR_TYPE_DEBUG,
1064        "UDP transmits %u-byte message to `%s' using address `%s'\n",
1065          msgbuf_size,
1066          GNUNET_i2s (&s->target),
1067          GNUNET_a2s(s->sock_addr, s->addrlen));
1068
1069   /* Message */
1070   udp = (struct UDPMessage *) mbuf;
1071   udp->header.size = htons (mlen);
1072   udp->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_MESSAGE);
1073   udp->reserved = htonl (0);
1074   udp->sender = *plugin->env->my_identity;
1075
1076   if (mlen <= UDP_MTU)
1077   {
1078     udpw = GNUNET_malloc (sizeof (struct UDPMessageWrapper) + mlen);
1079     udpw->session = s;
1080     udpw->udp = (char *) &udpw[1];
1081     udpw->msg_size = mlen;
1082     udpw->timeout = GNUNET_TIME_absolute_add(GNUNET_TIME_absolute_get(), to);
1083     udpw->cont = cont;
1084     udpw->cont_cls = cont_cls;
1085     udpw->frag_ctx = NULL;
1086
1087     memcpy (udpw->udp, udp, sizeof (struct UDPMessage));
1088     memcpy (&udpw->udp[sizeof (struct UDPMessage)], msgbuf, msgbuf_size);
1089
1090     enqueue (plugin, udpw);
1091   }
1092   else
1093   {
1094     LOG (GNUNET_ERROR_TYPE_DEBUG,
1095          "UDP has to fragment message \n");
1096     if  (s->frag_ctx != NULL)
1097       return GNUNET_SYSERR;
1098     memcpy (&udp[1], msgbuf, msgbuf_size);
1099     struct FragmentationContext * frag_ctx = GNUNET_malloc(sizeof (struct FragmentationContext));
1100
1101     frag_ctx->plugin = plugin;
1102     frag_ctx->session = s;
1103     frag_ctx->cont = cont;
1104     frag_ctx->cont_cls = cont_cls;
1105     frag_ctx->timeout = GNUNET_TIME_absolute_add(GNUNET_TIME_absolute_get(), to);
1106     frag_ctx->bytes_to_send = mlen;
1107     frag_ctx->frag = GNUNET_FRAGMENT_context_create (plugin->env->stats,
1108               UDP_MTU,
1109               &plugin->tracker,
1110               s->last_expected_delay,
1111               &udp->header,
1112               &enqueue_fragment,
1113               frag_ctx);
1114
1115     s->frag_ctx = frag_ctx;
1116
1117   }
1118
1119   if (s->addrlen == sizeof (struct sockaddr_in))
1120   {
1121     if (plugin->with_v4_ws == GNUNET_NO)
1122     {
1123       if (plugin->select_task != GNUNET_SCHEDULER_NO_TASK)
1124         GNUNET_SCHEDULER_cancel(plugin->select_task);
1125
1126       plugin->select_task =
1127           GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_DEFAULT,
1128                                        GNUNET_TIME_UNIT_FOREVER_REL,
1129                                        plugin->rs_v4,
1130                                        plugin->ws_v4,
1131                                        &udp_plugin_select, plugin);
1132       plugin->with_v4_ws = GNUNET_YES;
1133     }
1134   }
1135
1136   else if (s->addrlen == sizeof (struct sockaddr_in6))
1137   {
1138     if (plugin->with_v6_ws == GNUNET_NO)
1139     {
1140       if (plugin->select_task_v6 != GNUNET_SCHEDULER_NO_TASK)
1141         GNUNET_SCHEDULER_cancel(plugin->select_task_v6);
1142
1143       plugin->select_task_v6 =
1144         GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_DEFAULT,
1145                                      GNUNET_TIME_UNIT_FOREVER_REL,
1146                                      plugin->rs_v6,
1147                                      plugin->ws_v6,
1148                                      &udp_plugin_select_v6, plugin);
1149       plugin->with_v6_ws = GNUNET_YES;
1150     }
1151   }
1152
1153   return mlen;
1154 }
1155
1156
1157 /**
1158  * Our external IP address/port mapping has changed.
1159  *
1160  * @param cls closure, the 'struct LocalAddrList'
1161  * @param add_remove GNUNET_YES to mean the new public IP address, GNUNET_NO to mean
1162  *     the previous (now invalid) one
1163  * @param addr either the previous or the new public IP address
1164  * @param addrlen actual lenght of the address
1165  */
1166 static void
1167 udp_nat_port_map_callback (void *cls, int add_remove,
1168                            const struct sockaddr *addr, socklen_t addrlen)
1169 {
1170   struct Plugin *plugin = cls;
1171   struct IPv4UdpAddress u4;
1172   struct IPv6UdpAddress u6;
1173   void *arg;
1174   size_t args;
1175
1176   /* convert 'addr' to our internal format */
1177   switch (addr->sa_family)
1178   {
1179   case AF_INET:
1180     GNUNET_assert (addrlen == sizeof (struct sockaddr_in));
1181     u4.ipv4_addr = ((struct sockaddr_in *) addr)->sin_addr.s_addr;
1182     u4.u4_port = ((struct sockaddr_in *) addr)->sin_port;
1183     arg = &u4;
1184     args = sizeof (u4);
1185     break;
1186   case AF_INET6:
1187     GNUNET_assert (addrlen == sizeof (struct sockaddr_in6));
1188     memcpy (&u6.ipv6_addr, &((struct sockaddr_in6 *) addr)->sin6_addr,
1189             sizeof (struct in6_addr));
1190     u6.u6_port = ((struct sockaddr_in6 *) addr)->sin6_port;
1191     arg = &u6;
1192     args = sizeof (u6);
1193     break;
1194   default:
1195     GNUNET_break (0);
1196     return;
1197   }
1198   /* modify our published address list */
1199   plugin->env->notify_address (plugin->env->cls, add_remove, arg, args);
1200 }
1201
1202
1203
1204 /**
1205  * Message tokenizer has broken up an incomming message. Pass it on
1206  * to the service.
1207  *
1208  * @param cls the 'struct Plugin'
1209  * @param client the 'struct SourceInformation'
1210  * @param hdr the actual message
1211  */
1212 static void
1213 process_inbound_tokenized_messages (void *cls, void *client,
1214                                     const struct GNUNET_MessageHeader *hdr)
1215 {
1216   struct Plugin *plugin = cls;
1217   struct SourceInformation *si = client;
1218   struct GNUNET_ATS_Information ats[2];
1219   struct GNUNET_TIME_Relative delay;
1220
1221   GNUNET_assert (si->session != NULL);
1222   /* setup ATS */
1223   ats[0].type = htonl (GNUNET_ATS_QUALITY_NET_DISTANCE);
1224   ats[0].value = htonl (1);
1225   ats[1] = si->session->ats;
1226   GNUNET_break (ntohl(ats[1].value) != GNUNET_ATS_NET_UNSPECIFIED);
1227
1228   delay = plugin->env->receive (plugin->env->cls,
1229                 &si->sender,
1230                 hdr,
1231                 (const struct GNUNET_ATS_Information *) &ats, 2,
1232                 NULL,
1233                 si->arg,
1234                 si->args);
1235   si->session->flow_delay_for_other_peer = delay;
1236 }
1237
1238
1239 /**
1240  * We've received a UDP Message.  Process it (pass contents to main service).
1241  *
1242  * @param plugin plugin context
1243  * @param msg the message
1244  * @param sender_addr sender address
1245  * @param sender_addr_len number of bytes in sender_addr
1246  */
1247 static void
1248 process_udp_message (struct Plugin *plugin, const struct UDPMessage *msg,
1249                      const struct sockaddr *sender_addr,
1250                      socklen_t sender_addr_len)
1251 {
1252   struct SourceInformation si;
1253   struct Session * s = NULL;
1254   struct IPv4UdpAddress u4;
1255   struct IPv6UdpAddress u6;
1256   const void *arg;
1257   size_t args;
1258
1259   if (0 != ntohl (msg->reserved))
1260   {
1261     GNUNET_break_op (0);
1262     return;
1263   }
1264   if (ntohs (msg->header.size) <
1265       sizeof (struct GNUNET_MessageHeader) + sizeof (struct UDPMessage))
1266   {
1267     GNUNET_break_op (0);
1268     return;
1269   }
1270
1271   /* convert address */
1272   switch (sender_addr->sa_family)
1273   {
1274   case AF_INET:
1275     GNUNET_assert (sender_addr_len == sizeof (struct sockaddr_in));
1276     u4.ipv4_addr = ((struct sockaddr_in *) sender_addr)->sin_addr.s_addr;
1277     u4.u4_port = ((struct sockaddr_in *) sender_addr)->sin_port;
1278     arg = &u4;
1279     args = sizeof (u4);
1280     break;
1281   case AF_INET6:
1282     GNUNET_assert (sender_addr_len == sizeof (struct sockaddr_in6));
1283     u6.ipv6_addr = ((struct sockaddr_in6 *) sender_addr)->sin6_addr;
1284     u6.u6_port = ((struct sockaddr_in6 *) sender_addr)->sin6_port;
1285     arg = &u6;
1286     args = sizeof (u6);
1287     break;
1288   default:
1289     GNUNET_break (0);
1290     return;
1291   }
1292   LOG (GNUNET_ERROR_TYPE_DEBUG,
1293        "Received message with %u bytes from peer `%s' at `%s'\n",
1294        (unsigned int) ntohs (msg->header.size), GNUNET_i2s (&msg->sender),
1295        GNUNET_a2s (sender_addr, sender_addr_len));
1296
1297   struct GNUNET_HELLO_Address * address = GNUNET_HELLO_address_allocate(&msg->sender, "udp", arg, args);
1298   s = udp_plugin_get_session(plugin, address);
1299   GNUNET_free (address);
1300
1301   /* iterate over all embedded messages */
1302   si.session = s;
1303   si.sender = msg->sender;
1304   si.arg = arg;
1305   si.args = args;
1306
1307   GNUNET_SERVER_mst_receive (plugin->mst, &si, (const char *) &msg[1],
1308                              ntohs (msg->header.size) -
1309                              sizeof (struct UDPMessage), GNUNET_YES, GNUNET_NO);
1310 }
1311
1312
1313 /**
1314  * Scan the heap for a receive context with the given address.
1315  *
1316  * @param cls the 'struct FindReceiveContext'
1317  * @param node internal node of the heap
1318  * @param element value stored at the node (a 'struct ReceiveContext')
1319  * @param cost cost associated with the node
1320  * @return GNUNET_YES if we should continue to iterate,
1321  *         GNUNET_NO if not.
1322  */
1323 static int
1324 find_receive_context (void *cls, struct GNUNET_CONTAINER_HeapNode *node,
1325                       void *element, GNUNET_CONTAINER_HeapCostType cost)
1326 {
1327   struct FindReceiveContext *frc = cls;
1328   struct DefragContext *e = element;
1329
1330   if ((frc->addr_len == e->addr_len) &&
1331       (0 == memcmp (frc->addr, e->src_addr, frc->addr_len)))
1332   {
1333     frc->rc = e;
1334     return GNUNET_NO;
1335   }
1336   return GNUNET_YES;
1337 }
1338
1339
1340 /**
1341  * Process a defragmented message.
1342  *
1343  * @param cls the 'struct ReceiveContext'
1344  * @param msg the message
1345  */
1346 static void
1347 fragment_msg_proc (void *cls, const struct GNUNET_MessageHeader *msg)
1348 {
1349   struct DefragContext *rc = cls;
1350
1351   if (ntohs (msg->type) != GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_MESSAGE)
1352   {
1353     GNUNET_break (0);
1354     return;
1355   }
1356   if (ntohs (msg->size) < sizeof (struct UDPMessage))
1357   {
1358     GNUNET_break (0);
1359     return;
1360   }
1361   process_udp_message (rc->plugin, (const struct UDPMessage *) msg,
1362                        rc->src_addr, rc->addr_len);
1363 }
1364
1365 struct LookupContext
1366 {
1367   const struct sockaddr * addr;
1368   size_t addrlen;
1369
1370   struct Session *res;
1371 };
1372
1373 static int
1374 lookup_session_by_addr_it (void *cls, const GNUNET_HashCode * key, void *value)
1375 {
1376   struct LookupContext *l_ctx = cls;
1377   struct Session * s = value;
1378
1379   if ((s->addrlen == l_ctx->addrlen) &&
1380       (0 == memcmp (s->sock_addr, l_ctx->addr, s->addrlen)))
1381   {
1382     l_ctx->res = s;
1383     return GNUNET_NO;
1384   }
1385   return GNUNET_YES;
1386 }
1387
1388 /**
1389  * Transmit an acknowledgement.
1390  *
1391  * @param cls the 'struct ReceiveContext'
1392  * @param id message ID (unused)
1393  * @param msg ack to transmit
1394  */
1395 static void
1396 ack_proc (void *cls, uint32_t id, const struct GNUNET_MessageHeader *msg)
1397 {
1398   struct DefragContext *rc = cls;
1399
1400   size_t msize = sizeof (struct UDP_ACK_Message) + ntohs (msg->size);
1401   struct UDP_ACK_Message *udp_ack;
1402   uint32_t delay = 0;
1403   struct UDPMessageWrapper *udpw;
1404   struct Session *s;
1405
1406   struct LookupContext l_ctx;
1407   l_ctx.addr = rc->src_addr;
1408   l_ctx.addrlen = rc->addr_len;
1409   l_ctx.res = NULL;
1410   GNUNET_CONTAINER_multihashmap_iterate (rc->plugin->sessions,
1411       &lookup_session_by_addr_it,
1412       &l_ctx);
1413   s = l_ctx.res;
1414
1415   if (NULL == s)
1416     return;
1417
1418   if (s->flow_delay_for_other_peer.rel_value <= UINT32_MAX)
1419     delay = s->flow_delay_for_other_peer.rel_value;
1420
1421   LOG (GNUNET_ERROR_TYPE_DEBUG,
1422        "Sending ACK to `%s' including delay of %u ms\n",
1423        GNUNET_a2s (rc->src_addr,
1424                    (rc->src_addr->sa_family ==
1425                     AF_INET) ? sizeof (struct sockaddr_in) : sizeof (struct
1426                                                                      sockaddr_in6)),
1427        delay);
1428   udpw = GNUNET_malloc (sizeof (struct UDPMessageWrapper) + msize);
1429   udpw->cont = NULL;
1430   udpw->cont_cls = NULL;
1431   udpw->frag_ctx = NULL;
1432   udpw->msg_size = msize;
1433   udpw->session = s;
1434   udpw->timeout = GNUNET_TIME_absolute_get_forever();
1435   udpw->udp = (char *)&udpw[1];
1436
1437   udp_ack = (struct UDP_ACK_Message *) udpw->udp;
1438   udp_ack->header.size = htons ((uint16_t) msize);
1439   udp_ack->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_ACK);
1440   udp_ack->delay = htonl (delay);
1441   udp_ack->sender = *rc->plugin->env->my_identity;
1442   memcpy (&udp_ack[1], msg, ntohs (msg->size));
1443
1444   enqueue (rc->plugin, udpw);
1445 }
1446
1447
1448 static void read_process_msg (struct Plugin *plugin,
1449     const struct GNUNET_MessageHeader *msg,
1450     char *addr,
1451     socklen_t fromlen)
1452 {
1453   if (ntohs (msg->size) < sizeof (struct UDPMessage))
1454   {
1455     GNUNET_break_op (0);
1456     return;
1457   }
1458   process_udp_message (plugin, (const struct UDPMessage *) msg,
1459                        (const struct sockaddr *) addr, fromlen);
1460   return;
1461 }
1462
1463 static void read_process_ack (struct Plugin *plugin,
1464     const struct GNUNET_MessageHeader *msg,
1465     char *addr,
1466     socklen_t fromlen)
1467 {
1468   const struct GNUNET_MessageHeader *ack;
1469   const struct UDP_ACK_Message *udp_ack;
1470   struct LookupContext l_ctx;
1471   struct Session *s = NULL;
1472   struct GNUNET_TIME_Relative flow_delay;
1473
1474   if (ntohs (msg->size) <
1475       sizeof (struct UDP_ACK_Message) + sizeof (struct GNUNET_MessageHeader))
1476   {
1477     GNUNET_break_op (0);
1478     return;
1479   }
1480
1481   udp_ack = (const struct UDP_ACK_Message *) msg;
1482
1483   l_ctx.addr = (const struct sockaddr *) addr;
1484   l_ctx.addrlen = fromlen;
1485   l_ctx.res = NULL;
1486   GNUNET_CONTAINER_multihashmap_iterate (plugin->sessions,
1487       &lookup_session_by_addr_it,
1488       &l_ctx);
1489   s = l_ctx.res;
1490
1491   if ((s == NULL) || (s->frag_ctx == NULL))
1492     return;
1493
1494   flow_delay.rel_value = (uint64_t) ntohl (udp_ack->delay);
1495   LOG (GNUNET_ERROR_TYPE_DEBUG, "We received a sending delay of %llu\n",
1496        flow_delay.rel_value);
1497   s->flow_delay_from_other_peer =
1498       GNUNET_TIME_relative_to_absolute (flow_delay);
1499
1500   ack = (const struct GNUNET_MessageHeader *) &udp_ack[1];
1501   if (ntohs (ack->size) !=
1502       ntohs (msg->size) - sizeof (struct UDP_ACK_Message))
1503   {
1504     GNUNET_break_op (0);
1505     return;
1506   }
1507
1508   if (GNUNET_OK != GNUNET_FRAGMENT_process_ack (s->frag_ctx->frag, ack))
1509   {
1510   LOG (GNUNET_ERROR_TYPE_DEBUG,
1511        "UDP processes %u-byte acknowledgement from `%s' at `%s'\n",
1512        (unsigned int) ntohs (msg->size), GNUNET_i2s (&udp_ack->sender),
1513        GNUNET_a2s ((const struct sockaddr *) addr, fromlen));
1514     return;
1515   }
1516
1517   LOG (GNUNET_ERROR_TYPE_DEBUG,
1518        "FULL MESSAGE ACKed\n",
1519        (unsigned int) ntohs (msg->size), GNUNET_i2s (&udp_ack->sender),
1520        GNUNET_a2s ((const struct sockaddr *) addr, fromlen));
1521   s->last_expected_delay = GNUNET_FRAGMENT_context_destroy (s->frag_ctx->frag);
1522
1523   struct UDPMessageWrapper * udpw = NULL;
1524   if (s->addrlen == sizeof (struct sockaddr_in6))
1525   {
1526     udpw = plugin->ipv6_queue_head;
1527     while (udpw!= NULL)
1528     {
1529       if ((udpw->frag_ctx != NULL) && (udpw->frag_ctx == s->frag_ctx))
1530       {
1531         GNUNET_CONTAINER_DLL_remove(plugin->ipv6_queue_head, plugin->ipv6_queue_tail, udpw);
1532         GNUNET_free (udpw);
1533       }
1534       udpw = udpw->next;
1535     }
1536   }
1537   if (s->addrlen == sizeof (struct sockaddr_in))
1538   {
1539     udpw = plugin->ipv4_queue_head;
1540     while (udpw!= NULL)
1541     {
1542       if ((udpw->frag_ctx != NULL) && (udpw->frag_ctx == s->frag_ctx))
1543       {
1544         GNUNET_CONTAINER_DLL_remove(plugin->ipv4_queue_head, plugin->ipv4_queue_tail, udpw);
1545         GNUNET_free (udpw);
1546       }
1547       udpw = udpw->next;
1548     }
1549   }
1550
1551   if (s->frag_ctx->cont != NULL)
1552     s->frag_ctx->cont
1553     (s->frag_ctx->cont_cls, &udp_ack->sender, GNUNET_OK);
1554   GNUNET_free (s->frag_ctx);
1555   s->frag_ctx = NULL;
1556   return;
1557 }
1558
1559 static void read_process_fragment (struct Plugin *plugin,
1560     const struct GNUNET_MessageHeader *msg,
1561     char *addr,
1562     socklen_t fromlen)
1563 {
1564   struct DefragContext *d_ctx;
1565   struct GNUNET_TIME_Absolute now;
1566   struct FindReceiveContext frc;
1567
1568
1569   frc.rc = NULL;
1570   frc.addr = (const struct sockaddr *) addr;
1571   frc.addr_len = fromlen;
1572
1573   LOG (GNUNET_ERROR_TYPE_DEBUG, "UDP processes %u-byte fragment from `%s'\n",
1574        (unsigned int) ntohs (msg->size),
1575        GNUNET_a2s ((const struct sockaddr *) addr, fromlen));
1576   /* Lookup existing receive context for this address */
1577   GNUNET_CONTAINER_heap_iterate (plugin->defrag_ctxs,
1578                                  &find_receive_context,
1579                                  &frc);
1580   now = GNUNET_TIME_absolute_get ();
1581   d_ctx = frc.rc;
1582
1583   if (d_ctx == NULL)
1584   {
1585     /* Create a new defragmentation context */
1586     d_ctx = GNUNET_malloc (sizeof (struct DefragContext) + fromlen);
1587     memcpy (&d_ctx[1], addr, fromlen);
1588     d_ctx->src_addr = (const struct sockaddr *) &d_ctx[1];
1589     d_ctx->addr_len = fromlen;
1590     d_ctx->plugin = plugin;
1591     d_ctx->defrag =
1592         GNUNET_DEFRAGMENT_context_create (plugin->env->stats, UDP_MTU,
1593                                           UDP_MAX_MESSAGES_IN_DEFRAG, d_ctx,
1594                                           &fragment_msg_proc, &ack_proc);
1595     d_ctx->hnode =
1596         GNUNET_CONTAINER_heap_insert (plugin->defrag_ctxs, d_ctx,
1597                                       (GNUNET_CONTAINER_HeapCostType)
1598                                       now.abs_value);
1599   LOG (GNUNET_ERROR_TYPE_DEBUG, "Created new 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   else
1604   {
1605   LOG (GNUNET_ERROR_TYPE_DEBUG, "Found existing defragmentation context for %u-byte fragment from `%s'\n",
1606        (unsigned int) ntohs (msg->size),
1607        GNUNET_a2s ((const struct sockaddr *) addr, fromlen));
1608   }
1609
1610   if (GNUNET_OK == GNUNET_DEFRAGMENT_process_fragment (d_ctx->defrag, msg))
1611   {
1612     /* keep this 'rc' from expiring */
1613     GNUNET_CONTAINER_heap_update_cost (plugin->defrag_ctxs, d_ctx->hnode,
1614                                        (GNUNET_CONTAINER_HeapCostType)
1615                                        now.abs_value);
1616   }
1617   if (GNUNET_CONTAINER_heap_get_size (plugin->defrag_ctxs) >
1618       UDP_MAX_SENDER_ADDRESSES_WITH_DEFRAG)
1619   {
1620     /* remove 'rc' that was inactive the longest */
1621     d_ctx = GNUNET_CONTAINER_heap_remove_root (plugin->defrag_ctxs);
1622     GNUNET_assert (NULL != d_ctx);
1623     GNUNET_DEFRAGMENT_context_destroy (d_ctx->defrag);
1624     GNUNET_free (d_ctx);
1625   }
1626 }
1627
1628 /**
1629  * Read and process a message from the given socket.
1630  *
1631  * @param plugin the overall plugin
1632  * @param rsock socket to read from
1633  */
1634 static void
1635 udp_select_read (struct Plugin *plugin, struct GNUNET_NETWORK_Handle *rsock)
1636 {
1637   socklen_t fromlen;
1638   char addr[32];
1639   char buf[65536] GNUNET_ALIGN;
1640   ssize_t size;
1641   const struct GNUNET_MessageHeader *msg;
1642
1643   fromlen = sizeof (addr);
1644   memset (&addr, 0, sizeof (addr));
1645   size = GNUNET_NETWORK_socket_recvfrom (rsock, buf, sizeof (buf),
1646                                       (struct sockaddr *) &addr, &fromlen);
1647
1648   if (size < sizeof (struct GNUNET_MessageHeader))
1649   {
1650     GNUNET_break_op (0);
1651     return;
1652   }
1653   msg = (const struct GNUNET_MessageHeader *) buf;
1654
1655   LOG (GNUNET_ERROR_TYPE_DEBUG,
1656        "UDP received %u-byte message from `%s' type %i\n", (unsigned int) size,
1657        GNUNET_a2s ((const struct sockaddr *) addr, fromlen), ntohs (msg->type));
1658
1659   if (size != ntohs (msg->size))
1660   {
1661     GNUNET_break_op (0);
1662     return;
1663   }
1664
1665   switch (ntohs (msg->type))
1666   {
1667   case GNUNET_MESSAGE_TYPE_TRANSPORT_BROADCAST_BEACON:
1668     udp_broadcast_receive (plugin, &buf, size, addr, fromlen);
1669     return;
1670
1671   case GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_MESSAGE:
1672     read_process_msg (plugin, msg, addr, fromlen);
1673     return;
1674
1675   case GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_ACK:
1676     read_process_ack (plugin, msg, addr, fromlen);;
1677     return;
1678
1679   case GNUNET_MESSAGE_TYPE_FRAGMENT:
1680     read_process_fragment (plugin, msg, addr, fromlen);
1681     return;
1682
1683   default:
1684     GNUNET_break_op (0);
1685     return;
1686   }
1687 }
1688
1689 size_t
1690 udp_select_send (struct Plugin *plugin, struct GNUNET_NETWORK_Handle *sock)
1691 {
1692   ssize_t sent;
1693   size_t slen;
1694   struct GNUNET_TIME_Absolute max;
1695   struct GNUNET_TIME_Absolute ;
1696
1697   struct UDPMessageWrapper *udpw = NULL;
1698
1699   if (sock == plugin->sockv4)
1700   {
1701     udpw = plugin->ipv4_queue_head;
1702   }
1703   else if (sock == plugin->sockv6)
1704   {
1705     udpw = plugin->ipv6_queue_head;
1706   }
1707   else
1708   {
1709     GNUNET_break (0);
1710     return 0;
1711   }
1712
1713   const struct sockaddr * sa = udpw->session->sock_addr;
1714   slen = udpw->session->addrlen;
1715
1716   max = GNUNET_TIME_absolute_max(udpw->timeout, GNUNET_TIME_absolute_get());
1717
1718   while (udpw != NULL)
1719   {
1720     if (max.abs_value != udpw->timeout.abs_value)
1721     {
1722       /* Message timed out */
1723
1724       if (udpw->cont != NULL)
1725         udpw->cont (udpw->cont_cls, &udpw->session->target, GNUNET_SYSERR);
1726       if (udpw->frag_ctx != NULL)
1727       {
1728         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Fragmented message for peer `%s' with size %u timed out\n",
1729             GNUNET_i2s(&udpw->session->target), udpw->frag_ctx->bytes_to_send);
1730         udpw->session->last_expected_delay = GNUNET_FRAGMENT_context_destroy(udpw->frag_ctx->frag);
1731         GNUNET_free (udpw->frag_ctx);
1732         udpw->session->frag_ctx = NULL;
1733       }
1734       else
1735       {
1736         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Message for peer `%s' with size %u timed out\n",
1737             GNUNET_i2s(&udpw->session->target), udpw->msg_size);
1738       }
1739
1740       if (sock == plugin->sockv4)
1741       {
1742         GNUNET_CONTAINER_DLL_remove(plugin->ipv4_queue_head, plugin->ipv4_queue_tail, udpw);
1743         GNUNET_free (udpw);
1744         udpw = plugin->ipv4_queue_head;
1745       }
1746       else if (sock == plugin->sockv6)
1747       {
1748         GNUNET_CONTAINER_DLL_remove(plugin->ipv6_queue_head, plugin->ipv6_queue_tail, udpw);
1749         GNUNET_free (udpw);
1750         udpw = plugin->ipv6_queue_head;
1751       }
1752     }
1753     else
1754     {
1755       struct GNUNET_TIME_Relative delta = GNUNET_TIME_absolute_get_remaining (udpw->session->flow_delay_from_other_peer);
1756       if (delta.rel_value == 0)
1757       {
1758         /* this message is not delayed */
1759         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Message for peer `%s' (%u bytes) is not delayed \n",
1760             GNUNET_i2s(&udpw->session->target), udpw->msg_size);
1761         break;
1762       }
1763       else
1764       {
1765         /* this message is delayed, try next */
1766         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Message for peer `%s' (%u bytes) is delayed for %llu \n",
1767             GNUNET_i2s(&udpw->session->target), udpw->msg_size,
1768             delta);
1769         udpw = udpw->next;
1770       }
1771     }
1772   }
1773
1774   if (udpw == NULL)
1775   {
1776     /* No message left */
1777     return 0;
1778   }
1779
1780   sent = GNUNET_NETWORK_socket_sendto (sock, udpw->udp, udpw->msg_size, sa, slen);
1781
1782   if (GNUNET_SYSERR == sent)
1783   {
1784     LOG (GNUNET_ERROR_TYPE_ERROR,
1785          "UDP could not transmit %u-byte message to `%s': `%s'\n",
1786          (unsigned int) (udpw->msg_size), GNUNET_a2s (sa, slen),
1787          STRERROR (errno));
1788     if (udpw->cont != NULL)
1789       udpw->cont (udpw->cont_cls, &udpw->session->target, GNUNET_SYSERR);
1790   }
1791   else
1792   {
1793     LOG (GNUNET_ERROR_TYPE_DEBUG,
1794          "UDP transmitted %u-byte message to `%s' (%d: %s)\n",
1795          (unsigned int) (udpw->msg_size), GNUNET_a2s (sa, slen), (int) sent,
1796          (sent < 0) ? STRERROR (errno) : "ok");
1797   }
1798   /* This was just a message fragment */
1799   if (udpw->frag_ctx != NULL)
1800   {
1801     GNUNET_FRAGMENT_context_transmission_done (udpw->frag_ctx->frag);
1802   }
1803   /* This was a complete message*/
1804   else
1805   {
1806     if (udpw->cont != NULL)
1807       udpw->cont (udpw->cont_cls, &udpw->session->target, GNUNET_OK);
1808   }
1809
1810   if (sock == plugin->sockv4)
1811     GNUNET_CONTAINER_DLL_remove(plugin->ipv4_queue_head, plugin->ipv4_queue_tail, udpw);
1812   else if (sock == plugin->sockv6)
1813     GNUNET_CONTAINER_DLL_remove(plugin->ipv6_queue_head, plugin->ipv6_queue_tail, udpw);
1814   GNUNET_free (udpw);
1815   udpw = NULL;
1816
1817   return sent;
1818 }
1819
1820 /**
1821  * We have been notified that our readset has something to read.  We don't
1822  * know which socket needs to be read, so we have to check each one
1823  * Then reschedule this function to be called again once more is available.
1824  *
1825  * @param cls the plugin handle
1826  * @param tc the scheduling context (for rescheduling this function again)
1827  */
1828 static void
1829 udp_plugin_select (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1830 {
1831   struct Plugin *plugin = cls;
1832
1833   plugin->select_task = GNUNET_SCHEDULER_NO_TASK;
1834   if ((tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN) != 0)
1835     return;
1836   plugin->with_v4_ws = GNUNET_NO;
1837
1838   if ((tc->reason & GNUNET_SCHEDULER_REASON_READ_READY) != 0)
1839   {
1840     if ((NULL != plugin->sockv4) &&
1841       (GNUNET_NETWORK_fdset_isset (tc->read_ready, plugin->sockv4)))
1842         udp_select_read (plugin, plugin->sockv4);
1843
1844   }
1845
1846   if ((tc->reason & GNUNET_SCHEDULER_REASON_WRITE_READY) != 0)
1847   {
1848     if ((NULL != plugin->sockv4) && (plugin->ipv4_queue_head != NULL) &&
1849       (GNUNET_NETWORK_fdset_isset (tc->write_ready, plugin->sockv4)))
1850       {
1851         udp_select_send (plugin, plugin->sockv4);
1852       }
1853   }
1854
1855   if (plugin->select_task != GNUNET_SCHEDULER_NO_TASK)
1856     GNUNET_SCHEDULER_cancel (plugin->select_task);
1857   plugin->select_task = GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_DEFAULT,
1858                                    GNUNET_TIME_UNIT_FOREVER_REL,
1859                                    plugin->rs_v4,
1860                                    (plugin->ipv4_queue_head != NULL) ? plugin->ws_v4 : NULL,
1861                                    &udp_plugin_select, plugin);
1862   if (plugin->ipv4_queue_head != NULL)
1863     plugin->with_v4_ws = GNUNET_YES;
1864   else
1865     plugin->with_v4_ws = GNUNET_NO;
1866 }
1867
1868
1869 /**
1870  * We have been notified that our readset has something to read.  We don't
1871  * know which socket needs to be read, so we have to check each one
1872  * Then reschedule this function to be called again once more is available.
1873  *
1874  * @param cls the plugin handle
1875  * @param tc the scheduling context (for rescheduling this function again)
1876  */
1877 static void
1878 udp_plugin_select_v6 (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1879 {
1880   struct Plugin *plugin = cls;
1881
1882   plugin->select_task_v6 = GNUNET_SCHEDULER_NO_TASK;
1883   if ((tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN) != 0)
1884     return;
1885
1886   plugin->with_v6_ws = GNUNET_NO;
1887   if ((tc->reason & GNUNET_SCHEDULER_REASON_READ_READY) != 0)
1888   {
1889     if ((NULL != plugin->sockv6) &&
1890       (GNUNET_NETWORK_fdset_isset (tc->read_ready, plugin->sockv6)))
1891         udp_select_read (plugin, plugin->sockv6);
1892   }
1893
1894   if ((tc->reason & GNUNET_SCHEDULER_REASON_WRITE_READY) != 0)
1895   {
1896     if ((NULL != plugin->sockv6) && (plugin->ipv6_queue_head != NULL) &&
1897       (GNUNET_NETWORK_fdset_isset (tc->write_ready, plugin->sockv6)))
1898       {
1899         udp_select_send (plugin, plugin->sockv6);
1900       }
1901   }
1902   if (plugin->select_task_v6 != GNUNET_SCHEDULER_NO_TASK)
1903     GNUNET_SCHEDULER_cancel (plugin->select_task_v6);
1904   plugin->select_task_v6 = GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_DEFAULT,
1905                                    GNUNET_TIME_UNIT_FOREVER_REL,
1906                                    plugin->rs_v6,
1907                                    (plugin->ipv6_queue_head != NULL) ? plugin->ws_v6 : NULL,
1908                                    &udp_plugin_select_v6, plugin);
1909   if (plugin->ipv6_queue_head != NULL)
1910     plugin->with_v6_ws = GNUNET_YES;
1911   else
1912     plugin->with_v6_ws = GNUNET_NO;
1913 }
1914
1915
1916 static int
1917 setup_sockets (struct Plugin *plugin, struct sockaddr_in6 *serverAddrv6, struct sockaddr_in *serverAddrv4)
1918 {
1919   int tries;
1920   int sockets_created = 0;
1921   struct sockaddr *serverAddr;
1922   struct sockaddr *addrs[2];
1923   socklen_t addrlens[2];
1924   socklen_t addrlen;
1925
1926   /* Create IPv6 socket */
1927   if (plugin->enable_ipv6 == GNUNET_YES)
1928   {
1929     plugin->sockv6 = GNUNET_NETWORK_socket_create (PF_INET6, SOCK_DGRAM, 0);
1930     if (NULL == plugin->sockv6)
1931     {
1932       GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Disabling IPv6 since it is not supported on this system!\n");
1933       plugin->enable_ipv6 = GNUNET_NO;
1934     }
1935     else
1936     {
1937 #if HAVE_SOCKADDR_IN_SIN_LEN
1938       serverAddrv6->sin6_len = sizeof (serverAddrv6);
1939 #endif
1940       serverAddrv6->sin6_family = AF_INET6;
1941       serverAddrv6->sin6_addr = in6addr_any;
1942       serverAddrv6->sin6_port = htons (plugin->port);
1943       addrlen = sizeof (struct sockaddr_in6);
1944       serverAddr = (struct sockaddr *) serverAddrv6;
1945       LOG (GNUNET_ERROR_TYPE_DEBUG, "Binding to IPv6 port %d\n",
1946            ntohs (serverAddrv6->sin6_port));
1947       tries = 0;
1948       while (GNUNET_NETWORK_socket_bind (plugin->sockv6, serverAddr, addrlen) !=
1949              GNUNET_OK)
1950       {
1951         serverAddrv6->sin6_port = htons (GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_STRONG, 33537) + 32000);        /* Find a good, non-root port */
1952         LOG (GNUNET_ERROR_TYPE_DEBUG,
1953              "IPv6 Binding failed, trying new port %d\n",
1954              ntohs (serverAddrv6->sin6_port));
1955         tries++;
1956         if (tries > 10)
1957         {
1958           GNUNET_NETWORK_socket_close (plugin->sockv6);
1959           plugin->sockv6 = NULL;
1960           break;
1961         }
1962       }
1963       if (plugin->sockv6 != NULL)
1964       {
1965         LOG (GNUNET_ERROR_TYPE_DEBUG,
1966              "IPv6 socket created on port %d\n",
1967              ntohs (serverAddrv6->sin6_port));
1968         addrs[sockets_created] = (struct sockaddr *) serverAddrv6;
1969         addrlens[sockets_created] = sizeof (struct sockaddr_in6);
1970         sockets_created++;
1971       }
1972     }
1973   }
1974
1975   /* Create IPv4 socket */
1976   plugin->sockv4 = GNUNET_NETWORK_socket_create (PF_INET, SOCK_DGRAM, 0);
1977   if (NULL == plugin->sockv4)
1978   {
1979     GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING, "socket");
1980   }
1981   else
1982   {
1983 #if HAVE_SOCKADDR_IN_SIN_LEN
1984     serverAddrv4->sin_len = sizeof (serverAddrv4);
1985 #endif
1986     serverAddrv4->sin_family = AF_INET;
1987     serverAddrv4->sin_addr.s_addr = INADDR_ANY;
1988     serverAddrv4->sin_port = htons (plugin->port);
1989     addrlen = sizeof (struct sockaddr_in);
1990     serverAddr = (struct sockaddr *) serverAddrv4;
1991
1992     LOG (GNUNET_ERROR_TYPE_DEBUG, "Binding to IPv4 port %d\n",
1993          ntohs (serverAddrv4->sin_port));
1994     tries = 0;
1995     while (GNUNET_NETWORK_socket_bind (plugin->sockv4, serverAddr, addrlen) !=
1996            GNUNET_OK)
1997     {
1998       serverAddrv4->sin_port = htons (GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_STRONG, 33537) + 32000);   /* Find a good, non-root port */
1999       LOG (GNUNET_ERROR_TYPE_DEBUG, "IPv4 Binding failed, trying new port %d\n",
2000            ntohs (serverAddrv4->sin_port));
2001       tries++;
2002       if (tries > 10)
2003       {
2004         GNUNET_NETWORK_socket_close (plugin->sockv4);
2005         plugin->sockv4 = NULL;
2006         break;
2007       }
2008     }
2009     if (plugin->sockv4 != NULL)
2010     {
2011       addrs[sockets_created] = (struct sockaddr *) serverAddrv4;
2012       addrlens[sockets_created] = sizeof (struct sockaddr_in);
2013       sockets_created++;
2014     }
2015   }
2016
2017   /* Create file descriptors */
2018   plugin->rs_v4 = GNUNET_NETWORK_fdset_create ();
2019   plugin->ws_v4 = GNUNET_NETWORK_fdset_create ();
2020   GNUNET_NETWORK_fdset_zero (plugin->rs_v4);
2021   GNUNET_NETWORK_fdset_zero (plugin->ws_v4);
2022   if (NULL != plugin->sockv4)
2023   {
2024     GNUNET_NETWORK_fdset_set (plugin->rs_v4, plugin->sockv4);
2025     GNUNET_NETWORK_fdset_set (plugin->ws_v4, plugin->sockv4);
2026   }
2027
2028   if (sockets_created == 0)
2029     GNUNET_log (GNUNET_ERROR_TYPE_WARNING, _("Failed to open UDP sockets\n"));
2030
2031   plugin->select_task =
2032       GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_DEFAULT,
2033                                    GNUNET_TIME_UNIT_FOREVER_REL,
2034                                    plugin->rs_v4,
2035                                    NULL,
2036                                    &udp_plugin_select, plugin);
2037   plugin->with_v4_ws = GNUNET_NO;
2038
2039   if (plugin->enable_ipv6 == GNUNET_YES)
2040   {
2041     plugin->rs_v6 = GNUNET_NETWORK_fdset_create ();
2042     plugin->ws_v6 = GNUNET_NETWORK_fdset_create ();
2043     GNUNET_NETWORK_fdset_zero (plugin->rs_v6);
2044     GNUNET_NETWORK_fdset_zero (plugin->ws_v6);
2045     if (NULL != plugin->sockv6)
2046     {
2047       GNUNET_NETWORK_fdset_set (plugin->rs_v6, plugin->sockv6);
2048       GNUNET_NETWORK_fdset_set (plugin->ws_v6, plugin->sockv6);
2049     }
2050
2051     plugin->select_task_v6 =
2052         GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_DEFAULT,
2053                                      GNUNET_TIME_UNIT_FOREVER_REL,
2054                                      plugin->rs_v6,
2055                                      NULL,
2056                                      &udp_plugin_select_v6, plugin);
2057     plugin->with_v6_ws = GNUNET_NO;
2058   }
2059
2060   plugin->nat = GNUNET_NAT_register (plugin->env->cfg,
2061                            GNUNET_NO, plugin->port,
2062                            sockets_created,
2063                            (const struct sockaddr **) addrs, addrlens,
2064                            &udp_nat_port_map_callback, NULL, plugin);
2065
2066   return sockets_created;
2067 }
2068
2069
2070 /**
2071  * The exported method. Makes the core api available via a global and
2072  * returns the udp transport API.
2073  *
2074  * @param cls our 'struct GNUNET_TRANSPORT_PluginEnvironment'
2075  * @return our 'struct GNUNET_TRANSPORT_PluginFunctions'
2076  */
2077 void *
2078 libgnunet_plugin_transport_udp_init (void *cls)
2079 {
2080   struct GNUNET_TRANSPORT_PluginEnvironment *env = cls;
2081   struct GNUNET_TRANSPORT_PluginFunctions *api;
2082   struct Plugin *plugin;
2083
2084   unsigned long long port;
2085   unsigned long long aport;
2086   unsigned long long broadcast;
2087   unsigned long long udp_max_bps;
2088   unsigned long long enable_v6;
2089   char * bind4_address;
2090   char * bind6_address;
2091   struct GNUNET_TIME_Relative interval;
2092
2093   struct sockaddr_in serverAddrv4;
2094   struct sockaddr_in6 serverAddrv6;
2095
2096   int res;
2097
2098   if (NULL == env->receive)
2099   {
2100     /* run in 'stub' mode (i.e. as part of gnunet-peerinfo), don't fully
2101        initialze the plugin or the API */
2102     api = GNUNET_malloc (sizeof (struct GNUNET_TRANSPORT_PluginFunctions));
2103     api->cls = NULL;
2104     api->address_pretty_printer = &udp_plugin_address_pretty_printer;
2105     api->address_to_string = &udp_address_to_string;
2106     api->string_to_address = &udp_string_to_address;
2107     return api;
2108   }
2109
2110   GNUNET_assert( NULL != env->stats);
2111
2112   /* Get port number */
2113   if (GNUNET_OK !=
2114       GNUNET_CONFIGURATION_get_value_number (env->cfg, "transport-udp", "PORT",
2115                                              &port))
2116     port = 2086;
2117   if (GNUNET_OK !=
2118       GNUNET_CONFIGURATION_get_value_number (env->cfg, "transport-udp",
2119                                              "ADVERTISED_PORT", &aport))
2120     aport = port;
2121   if (port > 65535)
2122   {
2123     LOG (GNUNET_ERROR_TYPE_WARNING,
2124          _("Given `%s' option is out of range: %llu > %u\n"), "PORT", port,
2125          65535);
2126     return NULL;
2127   }
2128
2129   /* Protocols */
2130   if ((GNUNET_YES ==
2131        GNUNET_CONFIGURATION_get_value_yesno (env->cfg, "nat",
2132                                              "DISABLEV6")))
2133   {
2134     enable_v6 = GNUNET_NO;
2135   }
2136   else
2137     enable_v6 = GNUNET_YES;
2138
2139
2140   /* Addresses */
2141   memset (&serverAddrv6, 0, sizeof (serverAddrv6));
2142   memset (&serverAddrv4, 0, sizeof (serverAddrv4));
2143
2144   if (GNUNET_YES ==
2145       GNUNET_CONFIGURATION_get_value_string (env->cfg, "transport-udp",
2146                                              "BINDTO", &bind4_address))
2147   {
2148     LOG (GNUNET_ERROR_TYPE_DEBUG,
2149          "Binding udp plugin to specific address: `%s'\n",
2150          bind4_address);
2151     if (1 != inet_pton (AF_INET, bind4_address, &serverAddrv4.sin_addr))
2152     {
2153       GNUNET_free (bind4_address);
2154       return NULL;
2155     }
2156   }
2157
2158   if (GNUNET_YES ==
2159       GNUNET_CONFIGURATION_get_value_string (env->cfg, "transport-udp",
2160                                              "BINDTO6", &bind6_address))
2161   {
2162     LOG (GNUNET_ERROR_TYPE_DEBUG,
2163          "Binding udp plugin to specific address: `%s'\n",
2164          bind6_address);
2165     if (1 !=
2166         inet_pton (AF_INET6, bind6_address, &serverAddrv6.sin6_addr))
2167     {
2168       LOG (GNUNET_ERROR_TYPE_ERROR, _("Invalid IPv6 address: `%s'\n"),
2169            bind6_address);
2170       GNUNET_free_non_null (bind4_address);
2171       GNUNET_free (bind6_address);
2172       return NULL;
2173     }
2174   }
2175
2176
2177   /* Enable neighbour discovery */
2178   broadcast = GNUNET_CONFIGURATION_get_value_yesno (env->cfg, "transport-udp",
2179                                             "BROADCAST");
2180   if (broadcast == GNUNET_SYSERR)
2181     broadcast = GNUNET_NO;
2182
2183   if (GNUNET_SYSERR == GNUNET_CONFIGURATION_get_value_time (env->cfg, "transport-udp",
2184                                            "BROADCAST_INTERVAL", &interval))
2185   {
2186     interval = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 10);
2187   }
2188
2189   /* Maximum datarate */
2190   if (GNUNET_OK != GNUNET_CONFIGURATION_get_value_number (env->cfg, "transport-udp",
2191                                              "MAX_BPS", &udp_max_bps))
2192   {
2193     udp_max_bps = 1024 * 1024 * 50;     /* 50 MB/s == infinity for practical purposes */
2194   }
2195
2196   plugin = GNUNET_malloc (sizeof (struct Plugin));
2197   api = GNUNET_malloc (sizeof (struct GNUNET_TRANSPORT_PluginFunctions));
2198
2199   GNUNET_BANDWIDTH_tracker_init (&plugin->tracker,
2200                                  GNUNET_BANDWIDTH_value_init ((uint32_t)udp_max_bps), 30);
2201
2202
2203   plugin->sessions = GNUNET_CONTAINER_multihashmap_create (10);
2204   plugin->defrag_ctxs = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
2205   plugin->mst = GNUNET_SERVER_mst_create (&process_inbound_tokenized_messages, plugin);
2206   plugin->port = port;
2207   plugin->aport = aport;
2208   plugin->broadcast_interval = interval;
2209   plugin->enable_ipv6 = enable_v6;
2210   plugin->env = env;
2211
2212   api->cls = plugin;
2213   api->send = NULL;
2214   api->disconnect = &udp_disconnect;
2215   api->address_pretty_printer = &udp_plugin_address_pretty_printer;
2216   api->address_to_string = &udp_address_to_string;
2217   api->string_to_address = &udp_string_to_address;
2218   api->check_address = &udp_plugin_check_address;
2219   api->get_session = &udp_plugin_get_session;
2220   api->send = &udp_plugin_send;
2221
2222   LOG (GNUNET_ERROR_TYPE_DEBUG, "Setting up sockets\n");
2223   res = setup_sockets (plugin, &serverAddrv6, &serverAddrv4);
2224   if ((res == 0) || ((plugin->sockv4 == NULL) && (plugin->sockv6 == NULL)))
2225   {
2226     LOG (GNUNET_ERROR_TYPE_ERROR, "Failed to create network sockets, plugin failed\n");
2227     GNUNET_free (plugin);
2228     GNUNET_free (api);
2229     return NULL;
2230   }
2231
2232   LOG (GNUNET_ERROR_TYPE_DEBUG, "Starting broadcasting\n");
2233   if (broadcast == GNUNET_YES)
2234     setup_broadcast (plugin, &serverAddrv6, &serverAddrv4);
2235
2236
2237   GNUNET_free_non_null (bind4_address);
2238   GNUNET_free_non_null (bind6_address);
2239   return api;
2240 }
2241
2242
2243 static int
2244 heap_cleanup_iterator (void *cls,
2245                        struct GNUNET_CONTAINER_HeapNode *
2246                        node, void *element,
2247                        GNUNET_CONTAINER_HeapCostType
2248                        cost)
2249 {
2250   struct DefragContext * d_ctx = element;
2251
2252   GNUNET_CONTAINER_heap_remove_node (node);
2253   GNUNET_DEFRAGMENT_context_destroy(d_ctx->defrag);
2254   GNUNET_free (d_ctx);
2255
2256   return GNUNET_YES;
2257 }
2258
2259
2260 /**
2261  * The exported method. Makes the core api available via a global and
2262  * returns the udp transport API.
2263  *
2264  * @param cls our 'struct GNUNET_TRANSPORT_PluginEnvironment'
2265  * @return NULL
2266  */
2267 void *
2268 libgnunet_plugin_transport_udp_done (void *cls)
2269 {
2270   struct GNUNET_TRANSPORT_PluginFunctions *api = cls;
2271   struct Plugin *plugin = api->cls;
2272
2273   if (NULL == plugin)
2274   {
2275     GNUNET_free (api);
2276     return NULL;
2277   }
2278
2279   stop_broadcast (plugin);
2280
2281   if (plugin->select_task != GNUNET_SCHEDULER_NO_TASK)
2282   {
2283     GNUNET_SCHEDULER_cancel (plugin->select_task);
2284     plugin->select_task = GNUNET_SCHEDULER_NO_TASK;
2285   }
2286   if (plugin->select_task_v6 != GNUNET_SCHEDULER_NO_TASK)
2287   {
2288     GNUNET_SCHEDULER_cancel (plugin->select_task_v6);
2289     plugin->select_task_v6 = GNUNET_SCHEDULER_NO_TASK;
2290   }
2291
2292   /* Closing sockets */
2293   if (plugin->sockv4 != NULL)
2294   {
2295     GNUNET_break (GNUNET_OK == GNUNET_NETWORK_socket_close (plugin->sockv4));
2296     plugin->sockv4 = NULL;
2297   }
2298   GNUNET_NETWORK_fdset_destroy (plugin->rs_v4);
2299   GNUNET_NETWORK_fdset_destroy (plugin->ws_v4);
2300
2301   if (plugin->sockv6 != NULL)
2302   {
2303     GNUNET_break (GNUNET_OK == GNUNET_NETWORK_socket_close (plugin->sockv6));
2304     plugin->sockv6 = NULL;
2305
2306     GNUNET_NETWORK_fdset_destroy (plugin->rs_v6);
2307     GNUNET_NETWORK_fdset_destroy (plugin->ws_v6);
2308   }
2309
2310   GNUNET_NAT_unregister (plugin->nat);
2311
2312   if (plugin->defrag_ctxs != NULL)
2313   {
2314     GNUNET_CONTAINER_heap_iterate(plugin->defrag_ctxs,
2315         heap_cleanup_iterator, NULL);
2316     GNUNET_CONTAINER_heap_destroy(plugin->defrag_ctxs);
2317     plugin->defrag_ctxs = NULL;
2318   }
2319   if (plugin->mst != NULL)
2320   {
2321     GNUNET_SERVER_mst_destroy(plugin->mst);
2322     plugin->mst = NULL;
2323   }
2324
2325   /* Clean up leftover messages */
2326   struct UDPMessageWrapper * udpw;
2327   udpw = plugin->ipv4_queue_head;
2328   while (udpw != NULL)
2329   {
2330     struct UDPMessageWrapper *tmp = udpw->next;
2331     GNUNET_CONTAINER_DLL_remove(plugin->ipv4_queue_head, plugin->ipv4_queue_tail, udpw);
2332     if (udpw->cont != NULL)
2333       udpw->cont (udpw->cont_cls, &udpw->session->target, GNUNET_SYSERR);
2334     GNUNET_free (udpw);
2335     udpw = tmp;
2336   }
2337   udpw = plugin->ipv6_queue_head;
2338   while (udpw != NULL)
2339   {
2340     struct UDPMessageWrapper *tmp = udpw->next;
2341     GNUNET_CONTAINER_DLL_remove(plugin->ipv6_queue_head, plugin->ipv6_queue_tail, udpw);
2342     if (udpw->cont != NULL)
2343       udpw->cont (udpw->cont_cls, &udpw->session->target, GNUNET_SYSERR);
2344     GNUNET_free (udpw);
2345     udpw = tmp;
2346   }
2347
2348   /* Clean up sessions */
2349   LOG (GNUNET_ERROR_TYPE_DEBUG,
2350        "Cleaning up sessions\n");
2351   GNUNET_CONTAINER_multihashmap_iterate (plugin->sessions, &disconnect_and_free_it, plugin);
2352   GNUNET_CONTAINER_multihashmap_destroy (plugin->sessions);
2353
2354   plugin->nat = NULL;
2355   GNUNET_free (plugin);
2356   GNUNET_free (api);
2357   return NULL;
2358 }
2359
2360
2361 /* end of plugin_transport_udp.c */