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