src/transport/plugin_transport_udp.c

   1 /*
   2  This file is part of GNUnet
   3  (C) 2010-2014 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 #define UDP_SESSION_TIME_OUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 60)
  46
  47 /**
  48  * Number of messages we can defragment in parallel.  We only really
  49  * defragment 1 message at a time, but if messages get re-ordered, we
  50  * may want to keep knowledge about the previous message to avoid
  51  * discarding the current message in favor of a single fragment of a
  52  * previous message.  3 should be good since we don't expect massive
  53  * message reorderings with UDP.
  54  */
  55 #define UDP_MAX_MESSAGES_IN_DEFRAG 3
  56
  57 /**
  58  * We keep a defragmentation queue per sender address.  How many
  59  * sender addresses do we support at the same time? Memory consumption
  60  * is roughly a factor of 32k * UDP_MAX_MESSAGES_IN_DEFRAG times this
  61  * value. (So 128 corresponds to 12 MB and should suffice for
  62  * connecting to roughly 128 peers via UDP).
  63  */
  64 #define UDP_MAX_SENDER_ADDRESSES_WITH_DEFRAG 128
  65
  66
  67 /**
  68  * Closure for #append_port().
  69  */
  70 struct PrettyPrinterContext
  71 {
  72   /**
  73    * DLL
  74    */
  75   struct PrettyPrinterContext *next;
  76
  77   /**
  78    * DLL
  79    */
  80   struct PrettyPrinterContext *prev;
  81
  82   /**
  83    * Our plugin.
  84    */
  85   struct Plugin *plugin;
  86
  87   /**
  88    * Resolver handle
  89    */
  90   struct GNUNET_RESOLVER_RequestHandle *resolver_handle;
  91
  92   /**
  93    * Function to call with the result.
  94    */
  95   GNUNET_TRANSPORT_AddressStringCallback asc;
  96
  97   /**
  98    * Clsoure for @e asc.
  99    */
 100   void *asc_cls;
 101
 102   /**
 103    * Timeout task
 104    */
 105   struct GNUNET_SCHEDULER_Task * timeout_task;
 106
 107   /**
 108    * IPv6 address
 109    */
 110   int ipv6;
 111
 112   /**
 113    * Options
 114    */
 115   uint32_t options;
 116
 117   /**
 118    * Port to add after the IP address.
 119    */
 120   uint16_t port;
 121
 122 };
 123
 124
 125 /**
 126  * Session with another peer.
 127  */
 128 struct Session
 129 {
 130   /**
 131    * Which peer is this session for?
 132    */
 133   struct GNUNET_PeerIdentity target;
 134
 135   /**
 136    * Plugin this session belongs to.
 137    */
 138   struct Plugin *plugin;
 139
 140   /**
 141    * Context for dealing with fragments.
 142    */
 143   struct UDP_FragmentationContext *frag_ctx;
 144
 145   /**
 146    * Desired delay for next sending we send to other peer
 147    */
 148   struct GNUNET_TIME_Relative flow_delay_for_other_peer;
 149
 150   /**
 151    * Desired delay for next sending we received from other peer
 152    */
 153   struct GNUNET_TIME_Absolute flow_delay_from_other_peer;
 154
 155   /**
 156    * Session timeout task
 157    */
 158   struct GNUNET_SCHEDULER_Task * timeout_task;
 159
 160   /**
 161    * When does this session time out?
 162    */
 163   struct GNUNET_TIME_Absolute timeout;
 164
 165   /**
 166    * expected delay for ACKs
 167    */
 168   struct GNUNET_TIME_Relative last_expected_ack_delay;
 169
 170   /**
 171    * desired delay between UDP messages
 172    */
 173   struct GNUNET_TIME_Relative last_expected_msg_delay;
 174
 175   /**
 176    * Address metrics (as set by the "update_address_metrics" by
 177    * the environment).
 178    */
 179   struct GNUNET_ATS_Information ats;
 180
 181   /**
 182    * Our own address.
 183    */
 184   struct GNUNET_HELLO_Address *address;
 185
 186   /**
 187    * Number of bytes waiting for transmission to this peer.
 188    */
 189   unsigned long long bytes_in_queue;
 190
 191   /**
 192    * Number of messages waiting for transmission to this peer.
 193    */
 194   unsigned int msgs_in_queue;
 195
 196   /**
 197    * Reference counter to indicate that this session is
 198    * currently being used and must not be destroyed;
 199    * setting @e in_destroy will destroy it as soon as
 200    * possible.
 201    */
 202   unsigned int rc;
 203
 204   /**
 205    * Is this session about to be destroyed (sometimes we cannot
 206    * destroy a session immediately as below us on the stack
 207    * there might be code that still uses it; in this case,
 208    * @e rc is non-zero).
 209    */
 210   int in_destroy;
 211 };
 212
 213
 214 /**
 215  * Closure for #process_inbound_tokenized_messages().
 216  */
 217 struct SourceInformation
 218 {
 219   /**
 220    * Sender identity.
 221    */
 222   struct GNUNET_PeerIdentity sender;
 223
 224   /**
 225    * Source address.
 226    */
 227   const void *arg;
 228
 229   /**
 230    * Associated session.
 231    */
 232   struct Session *session;
 233
 234   /**
 235    * Number of bytes in source address.
 236    */
 237   size_t args;
 238
 239 };
 240
 241 /**
 242  * Closure for #find_receive_context().
 243  */
 244 struct FindReceiveContext
 245 {
 246   /**
 247    * Where to store the result.
 248    */
 249   struct DefragContext *rc;
 250
 251   /**
 252    * Address to find.
 253    */
 254   const struct sockaddr *addr;
 255
 256   /**
 257    *
 258    */
 259   struct Session *session;
 260
 261   /**
 262    * Number of bytes in @e addr.
 263    */
 264   socklen_t addr_len;
 265
 266 };
 267
 268 /**
 269  * Data structure to track defragmentation contexts based
 270  * on the source of the UDP traffic.
 271  */
 272 struct DefragContext
 273 {
 274
 275   /**
 276    * Defragmentation context.
 277    */
 278   struct GNUNET_DEFRAGMENT_Context *defrag;
 279
 280   /**
 281    * Source address this receive context is for (allocated at the
 282    * end of the struct).
 283    */
 284   const struct sockaddr *src_addr;
 285
 286   /**
 287    * Reference to master plugin struct.
 288    */
 289   struct Plugin *plugin;
 290
 291   /**
 292    * Node in the defrag heap.
 293    */
 294   struct GNUNET_CONTAINER_HeapNode *hnode;
 295
 296   /**
 297    * Length of @e src_addr.
 298    */
 299   size_t addr_len;
 300 };
 301
 302
 303 /**
 304  * Context to send fragmented messages
 305  */
 306 struct UDP_FragmentationContext
 307 {
 308   /**
 309    * Next in linked list
 310    */
 311   struct UDP_FragmentationContext *next;
 312
 313   /**
 314    * Previous in linked list
 315    */
 316   struct UDP_FragmentationContext *prev;
 317
 318   /**
 319    * The plugin
 320    */
 321   struct Plugin *plugin;
 322
 323   /**
 324    * Handle for GNUNET_FRAGMENT context
 325    */
 326   struct GNUNET_FRAGMENT_Context *frag;
 327
 328   /**
 329    * The session this fragmentation context belongs to
 330    */
 331   struct Session *session;
 332
 333   /**
 334    * Function to call upon completion of the transmission.
 335    */
 336   GNUNET_TRANSPORT_TransmitContinuation cont;
 337
 338   /**
 339    * Closure for @e cont.
 340    */
 341   void *cont_cls;
 342
 343   /**
 344    * Message timeout
 345    */
 346   struct GNUNET_TIME_Absolute timeout;
 347
 348   /**
 349    * Payload size of original unfragmented message
 350    */
 351   size_t payload_size;
 352
 353   /**
 354    * Bytes used to send all fragments on wire including UDP overhead
 355    */
 356   size_t on_wire_size;
 357
 358   /**
 359    * FIXME.
 360    */
 361   unsigned int fragments_used;
 362
 363 };
 364
 365
 366 /**
 367  * Message types included in a `struct UDP_MessageWrapper`
 368  */
 369 enum UDP_MessageType
 370 {
 371   /**
 372    * Uninitialized (error)
 373    */
 374   UMT_UNDEFINED = 0,
 375
 376   /**
 377    * Fragment of a message.
 378    */
 379   UMT_MSG_FRAGMENTED = 1,
 380
 381   /**
 382    *
 383    */
 384   UMT_MSG_FRAGMENTED_COMPLETE = 2,
 385
 386   /**
 387    * Unfragmented message.
 388    */
 389   UMT_MSG_UNFRAGMENTED = 3,
 390
 391   /**
 392    * Receipt confirmation.
 393    */
 394   UMT_MSG_ACK = 4
 395
 396 };
 397
 398
 399 /**
 400  * Information we track for each message in the queue.
 401  */
 402 struct UDP_MessageWrapper
 403 {
 404   /**
 405    * Session this message belongs to
 406    */
 407   struct Session *session;
 408
 409   /**
 410    * DLL of messages
 411    * previous element
 412    */
 413   struct UDP_MessageWrapper *prev;
 414
 415   /**
 416    * DLL of messages
 417    * previous element
 418    */
 419   struct UDP_MessageWrapper *next;
 420
 421   /**
 422    * Message with size msg_size including UDP specific overhead
 423    */
 424   char *msg_buf;
 425
 426   /**
 427    * Function to call upon completion of the transmission.
 428    */
 429   GNUNET_TRANSPORT_TransmitContinuation cont;
 430
 431   /**
 432    * Closure for @e cont.
 433    */
 434   void *cont_cls;
 435
 436   /**
 437    * Fragmentation context
 438    * frag_ctx == NULL if transport <= MTU
 439    * frag_ctx != NULL if transport > MTU
 440    */
 441   struct UDP_FragmentationContext *frag_ctx;
 442
 443   /**
 444    * Message timeout
 445    */
 446   struct GNUNET_TIME_Absolute timeout;
 447
 448   /**
 449    * Size of UDP message to send including UDP specific overhead
 450    */
 451   size_t msg_size;
 452
 453   /**
 454    * Payload size of original message
 455    */
 456   size_t payload_size;
 457
 458   /**
 459    * Message type
 460    */
 461   enum UDP_MessageType msg_type;
 462
 463 };
 464
 465
 466 /**
 467  * UDP ACK Message-Packet header (after defragmentation).
 468  */
 469 struct UDP_ACK_Message
 470 {
 471   /**
 472    * Message header.
 473    */
 474   struct GNUNET_MessageHeader header;
 475
 476   /**
 477    * Desired delay for flow control
 478    */
 479   uint32_t delay;
 480
 481   /**
 482    * What is the identity of the sender
 483    */
 484   struct GNUNET_PeerIdentity sender;
 485
 486 };
 487
 488
 489 /**
 490  * If a session monitor is attached, notify it about the new
 491  * session state.
 492  *
 493  * @param plugin our plugin
 494  * @param session session that changed state
 495  * @param state new state of the session
 496  */
 497 static void
 498 notify_session_monitor (struct Plugin *plugin,
 499                         struct Session *session,
 500                         enum GNUNET_TRANSPORT_SessionState state)
 501 {
 502   struct GNUNET_TRANSPORT_SessionInfo info;
 503
 504   if (NULL == plugin->sic)
 505     return;
 506   memset (&info, 0, sizeof (info));
 507   info.state = state;
 508   info.is_inbound = GNUNET_SYSERR; /* hard to say */
 509   info.num_msg_pending = session->msgs_in_queue;
 510   info.num_bytes_pending = session->bytes_in_queue;
 511   /* info.receive_delay remains zero as this is not supported by UDP
 512      (cannot selectively not receive from 'some' peer while continuing
 513      to receive from others) */
 514   info.session_timeout = session->timeout;
 515   info.address = session->address;
 516   plugin->sic (plugin->sic_cls,
 517                session,
 518                &info);
 519 }
 520
 521
 522 /**
 523  * We have been notified that our readset has something to read.  We don't
 524  * know which socket needs to be read, so we have to check each one
 525  * Then reschedule this function to be called again once more is available.
 526  *
 527  * @param cls the plugin handle
 528  * @param tc the scheduling context (for rescheduling this function again)
 529  */
 530 static void
 531 udp_plugin_select (void *cls,
 532                    const struct GNUNET_SCHEDULER_TaskContext *tc);
 533
 534
 535 /**
 536  * We have been notified that our readset has something to read.  We don't
 537  * know which socket needs to be read, so we have to check each one
 538  * Then reschedule this function to be called again once more is available.
 539  *
 540  * @param cls the plugin handle
 541  * @param tc the scheduling context (for rescheduling this function again)
 542  */
 543 static void
 544 udp_plugin_select_v6 (void *cls,
 545                       const struct GNUNET_SCHEDULER_TaskContext *tc);
 546
 547
 548 /**
 549  * (re)schedule select tasks for this plugin.
 550  *
 551  * @param plugin plugin to reschedule
 552  */
 553 static void
 554 schedule_select (struct Plugin *plugin)
 555 {
 556   struct GNUNET_TIME_Relative min_delay;
 557   struct UDP_MessageWrapper *udpw;
 558
 559   if ((GNUNET_YES == plugin->enable_ipv4) && (NULL != plugin->sockv4))
 560   {
 561     /* Find a message ready to send:
 562      * Flow delay from other peer is expired or not set (0) */
 563     min_delay = GNUNET_TIME_UNIT_FOREVER_REL;
 564     for (udpw = plugin->ipv4_queue_head; NULL != udpw; udpw = udpw->next)
 565       min_delay = GNUNET_TIME_relative_min (min_delay,
 566           GNUNET_TIME_absolute_get_remaining (
 567               udpw->session->flow_delay_from_other_peer));
 568
 569     if (plugin->select_task != NULL )
 570       GNUNET_SCHEDULER_cancel (plugin->select_task);
 571
 572     /* Schedule with:
 573      * - write active set if message is ready
 574      * - timeout minimum delay */
 575     plugin->select_task = GNUNET_SCHEDULER_add_select (
 576         GNUNET_SCHEDULER_PRIORITY_DEFAULT,
 577         (0 == min_delay.rel_value_us) ?
 578             GNUNET_TIME_UNIT_FOREVER_REL : min_delay, plugin->rs_v4,
 579         (0 == min_delay.rel_value_us) ? plugin->ws_v4 : NULL,
 580         &udp_plugin_select, plugin);
 581   }
 582   if ((GNUNET_YES == plugin->enable_ipv6) && (NULL != plugin->sockv6))
 583   {
 584     min_delay = GNUNET_TIME_UNIT_FOREVER_REL;
 585     for (udpw = plugin->ipv6_queue_head; NULL != udpw; udpw = udpw->next)
 586       min_delay = GNUNET_TIME_relative_min (min_delay,
 587           GNUNET_TIME_absolute_get_remaining (
 588               udpw->session->flow_delay_from_other_peer));
 589
 590     if (NULL != plugin->select_task_v6)
 591       GNUNET_SCHEDULER_cancel (plugin->select_task_v6);
 592     plugin->select_task_v6 = GNUNET_SCHEDULER_add_select (
 593         GNUNET_SCHEDULER_PRIORITY_DEFAULT,
 594         (0 == min_delay.rel_value_us) ?
 595             GNUNET_TIME_UNIT_FOREVER_REL : min_delay, plugin->rs_v6,
 596         (0 == min_delay.rel_value_us) ? plugin->ws_v6 : NULL,
 597         &udp_plugin_select_v6, plugin);
 598   }
 599 }
 600
 601
 602 /**
 603  * Function called for a quick conversion of the binary address to
 604  * a numeric address.  Note that the caller must not free the
 605  * address and that the next call to this function is allowed
 606  * to override the address again.
 607  *
 608  * @param cls closure
 609  * @param addr binary address
 610  * @param addrlen length of the address
 611  * @return string representing the same address
 612  */
 613 const char *
 614 udp_address_to_string (void *cls,
 615                        const void *addr,
 616                        size_t addrlen)
 617 {
 618   static char rbuf[INET6_ADDRSTRLEN + 10];
 619   char buf[INET6_ADDRSTRLEN];
 620   const void *sb;
 621   struct in_addr a4;
 622   struct in6_addr a6;
 623   const struct IPv4UdpAddress *t4;
 624   const struct IPv6UdpAddress *t6;
 625   int af;
 626   uint16_t port;
 627   uint32_t options;
 628
 629   if ((NULL != addr) && (addrlen == sizeof(struct IPv6UdpAddress)))
 630   {
 631     t6 = addr;
 632     af = AF_INET6;
 633     options = ntohl (t6->options);
 634     port = ntohs (t6->u6_port);
 635     memcpy (&a6, &t6->ipv6_addr, sizeof(a6));
 636     sb = &a6;
 637   }
 638   else if ((NULL != addr) && (addrlen == sizeof(struct IPv4UdpAddress)))
 639   {
 640     t4 = addr;
 641     af = AF_INET;
 642     options = ntohl (t4->options);
 643     port = ntohs (t4->u4_port);
 644     memcpy (&a4, &t4->ipv4_addr, sizeof(a4));
 645     sb = &a4;
 646   }
 647   else
 648   {
 649     return NULL;
 650   }
 651   inet_ntop (af, sb, buf, INET6_ADDRSTRLEN);
 652
 653   GNUNET_snprintf (rbuf, sizeof(rbuf),
 654       (af == AF_INET6) ? "%s.%u.[%s]:%u" : "%s.%u.%s:%u", PLUGIN_NAME, options,
 655       buf, port);
 656   return rbuf;
 657 }
 658
 659
 660 /**
 661  * Function called to convert a string address to
 662  * a binary address.
 663  *
 664  * @param cls closure ('struct Plugin*')
 665  * @param addr string address
 666  * @param addrlen length of the address
 667  * @param buf location to store the buffer
 668  * @param added location to store the number of bytes in the buffer.
 669  *        If the function returns #GNUNET_SYSERR, its contents are undefined.
 670  * @return #GNUNET_OK on success, #GNUNET_SYSERR on failure
 671  */
 672 static int
 673 udp_string_to_address (void *cls,
 674                        const char *addr,
 675                        uint16_t addrlen,
 676                        void **buf,
 677                        size_t *added)
 678 {
 679   struct sockaddr_storage socket_address;
 680   char *address;
 681   char *plugin;
 682   char *optionstr;
 683   uint32_t options;
 684
 685   /* Format tcp.options.address:port */
 686   address = NULL;
 687   plugin = NULL;
 688   optionstr = NULL;
 689
 690   if ((NULL == addr) || (addrlen == 0))
 691   {
 692     GNUNET_break(0);
 693     return GNUNET_SYSERR;
 694   }
 695   if ('\0' != addr[addrlen - 1])
 696   {
 697     GNUNET_break(0);
 698     return GNUNET_SYSERR;
 699   }
 700   if (strlen (addr) != addrlen - 1)
 701   {
 702     GNUNET_break(0);
 703     return GNUNET_SYSERR;
 704   }
 705   plugin = GNUNET_strdup (addr);
 706   optionstr = strchr (plugin, '.');
 707   if (NULL == optionstr)
 708   {
 709     GNUNET_break(0);
 710     GNUNET_free(plugin);
 711     return GNUNET_SYSERR;
 712   }
 713   optionstr[0] = '\0';
 714   optionstr++;
 715   options = atol (optionstr);
 716   address = strchr (optionstr, '.');
 717   if (NULL == address)
 718   {
 719     GNUNET_break(0);
 720     GNUNET_free(plugin);
 721     return GNUNET_SYSERR;
 722   }
 723   address[0] = '\0';
 724   address++;
 725
 726   if (GNUNET_OK !=
 727       GNUNET_STRINGS_to_address_ip (address, strlen (address),
 728                                     &socket_address))
 729   {
 730     GNUNET_break(0);
 731     GNUNET_free(plugin);
 732     return GNUNET_SYSERR;
 733   }
 734
 735   GNUNET_free(plugin);
 736
 737   switch (socket_address.ss_family)
 738   {
 739   case AF_INET:
 740   {
 741     struct IPv4UdpAddress *u4;
 742     struct sockaddr_in *in4 = (struct sockaddr_in *) &socket_address;
 743     u4 = GNUNET_new (struct IPv4UdpAddress);
 744     u4->options = htonl (options);
 745     u4->ipv4_addr = in4->sin_addr.s_addr;
 746     u4->u4_port = in4->sin_port;
 747     *buf = u4;
 748     *added = sizeof(struct IPv4UdpAddress);
 749     return GNUNET_OK;
 750   }
 751   case AF_INET6:
 752   {
 753     struct IPv6UdpAddress *u6;
 754     struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) &socket_address;
 755     u6 = GNUNET_new (struct IPv6UdpAddress);
 756     u6->options = htonl (options);
 757     u6->ipv6_addr = in6->sin6_addr;
 758     u6->u6_port = in6->sin6_port;
 759     *buf = u6;
 760     *added = sizeof(struct IPv6UdpAddress);
 761     return GNUNET_OK;
 762   }
 763   default:
 764     GNUNET_break(0);
 765     return GNUNET_SYSERR;
 766   }
 767 }
 768
 769
 770 /**
 771  * Append our port and forward the result.
 772  *
 773  * @param cls a `struct PrettyPrinterContext *`
 774  * @param hostname result from DNS resolver
 775  */
 776 static void
 777 append_port (void *cls,
 778              const char *hostname)
 779 {
 780   struct PrettyPrinterContext *ppc = cls;
 781   struct Plugin *plugin = ppc->plugin;
 782   char *ret;
 783
 784   if (NULL == hostname)
 785   {
 786     /* Final call, done */
 787     ppc->asc (ppc->asc_cls,
 788               NULL,
 789               GNUNET_OK);
 790     GNUNET_CONTAINER_DLL_remove (plugin->ppc_dll_head,
 791                                  plugin->ppc_dll_tail,
 792                                  ppc);
 793     ppc->resolver_handle = NULL;
 794     GNUNET_free (ppc);
 795     return;
 796   }
 797   if (GNUNET_YES == ppc->ipv6)
 798     GNUNET_asprintf (&ret,
 799                      "%s.%u.[%s]:%d",
 800                      PLUGIN_NAME,
 801                      ppc->options,
 802                      hostname,
 803                      ppc->port);
 804   else
 805     GNUNET_asprintf (&ret,
 806                      "%s.%u.%s:%d",
 807                      PLUGIN_NAME,
 808                      ppc->options,
 809                      hostname,
 810                      ppc->port);
 811   ppc->asc (ppc->asc_cls,
 812             ret,
 813             GNUNET_OK);
 814   GNUNET_free (ret);
 815 }
 816
 817
 818 /**
 819  * Convert the transports address to a nice, human-readable
 820  * format.
 821  *
 822  * @param cls closure with the `struct Plugin *`
 823  * @param type name of the transport that generated the address
 824  * @param addr one of the addresses of the host, NULL for the last address
 825  *        the specific address format depends on the transport
 826  * @param addrlen length of the address
 827  * @param numeric should (IP) addresses be displayed in numeric form?
 828  * @param timeout after how long should we give up?
 829  * @param asc function to call on each string
 830  * @param asc_cls closure for @a asc
 831  */
 832 static void
 833 udp_plugin_address_pretty_printer (void *cls,
 834                                    const char *type,
 835                                    const void *addr,
 836                                    size_t addrlen,
 837                                    int numeric,
 838                                    struct GNUNET_TIME_Relative timeout,
 839                                    GNUNET_TRANSPORT_AddressStringCallback asc,
 840                                    void *asc_cls)
 841 {
 842   struct Plugin *plugin = cls;
 843   struct PrettyPrinterContext *ppc;
 844   const void *sb;
 845   size_t sbs;
 846   struct sockaddr_in a4;
 847   struct sockaddr_in6 a6;
 848   const struct IPv4UdpAddress *u4;
 849   const struct IPv6UdpAddress *u6;
 850   uint16_t port;
 851   uint32_t options;
 852
 853   if (addrlen == sizeof(struct IPv6UdpAddress))
 854   {
 855     u6 = addr;
 856     memset (&a6, 0, sizeof(a6));
 857     a6.sin6_family = AF_INET6;
 858 #if HAVE_SOCKADDR_IN_SIN_LEN
 859     a6.sin6_len = sizeof (a6);
 860 #endif
 861     a6.sin6_port = u6->u6_port;
 862     memcpy (&a6.sin6_addr, &u6->ipv6_addr, sizeof(struct in6_addr));
 863     port = ntohs (u6->u6_port);
 864     options = ntohl (u6->options);
 865     sb = &a6;
 866     sbs = sizeof(a6);
 867   }
 868   else if (addrlen == sizeof(struct IPv4UdpAddress))
 869   {
 870     u4 = addr;
 871     memset (&a4, 0, sizeof(a4));
 872     a4.sin_family = AF_INET;
 873 #if HAVE_SOCKADDR_IN_SIN_LEN
 874     a4.sin_len = sizeof (a4);
 875 #endif
 876     a4.sin_port = u4->u4_port;
 877     a4.sin_addr.s_addr = u4->ipv4_addr;
 878     port = ntohs (u4->u4_port);
 879     options = ntohl (u4->options);
 880     sb = &a4;
 881     sbs = sizeof(a4);
 882   }
 883   else
 884   {
 885     /* invalid address */
 886     GNUNET_break_op (0);
 887     asc (asc_cls, NULL , GNUNET_SYSERR);
 888     asc (asc_cls, NULL, GNUNET_OK);
 889     return;
 890   }
 891   ppc = GNUNET_new (struct PrettyPrinterContext);
 892   ppc->plugin = plugin;
 893   ppc->asc = asc;
 894   ppc->asc_cls = asc_cls;
 895   ppc->port = port;
 896   ppc->options = options;
 897   if (addrlen == sizeof(struct IPv6UdpAddress))
 898     ppc->ipv6 = GNUNET_YES;
 899   else
 900     ppc->ipv6 = GNUNET_NO;
 901   GNUNET_CONTAINER_DLL_insert (plugin->ppc_dll_head,
 902                                plugin->ppc_dll_tail,
 903                                ppc);
 904   ppc->resolver_handle
 905     = GNUNET_RESOLVER_hostname_get (sb,
 906                                     sbs,
 907                                     ! numeric,
 908                                     timeout,
 909                                     &append_port, ppc);
 910 }
 911
 912
 913 /**
 914  * FIXME.
 915  */
 916 static void
 917 call_continuation (struct UDP_MessageWrapper *udpw,
 918                    int result)
 919 {
 920   struct Session *session = udpw->session;
 921   struct Plugin *plugin = session->plugin;
 922   size_t overhead;
 923
 924   LOG (GNUNET_ERROR_TYPE_DEBUG,
 925        "Calling continuation for %u byte message to `%s' with result %s\n",
 926        udpw->payload_size, GNUNET_i2s (&udpw->session->target),
 927        (GNUNET_OK == result) ? "OK" : "SYSERR");
 928
 929   if (udpw->msg_size >= udpw->payload_size)
 930     overhead = udpw->msg_size - udpw->payload_size;
 931   else
 932     overhead = udpw->msg_size;
 933
 934   switch (result)
 935   {
 936   case GNUNET_OK:
 937     switch (udpw->msg_type)
 938     {
 939     case UMT_MSG_UNFRAGMENTED:
 940       if (NULL != udpw->cont)
 941       {
 942         /* Transport continuation */
 943         udpw->cont (udpw->cont_cls, &udpw->session->target, result,
 944             udpw->payload_size, udpw->msg_size);
 945       }
 946       GNUNET_STATISTICS_update (plugin->env->stats,
 947           "# UDP, unfragmented msgs, messages, sent, success", 1, GNUNET_NO);
 948       GNUNET_STATISTICS_update (plugin->env->stats,
 949           "# UDP, unfragmented msgs, bytes payload, sent, success",
 950           udpw->payload_size, GNUNET_NO);
 951       GNUNET_STATISTICS_update (plugin->env->stats,
 952           "# UDP, unfragmented msgs, bytes overhead, sent, success", overhead,
 953           GNUNET_NO);
 954       GNUNET_STATISTICS_update (plugin->env->stats,
 955           "# UDP, total, bytes overhead, sent", overhead, GNUNET_NO);
 956       GNUNET_STATISTICS_update (plugin->env->stats,
 957           "# UDP, total, bytes payload, sent", udpw->payload_size, GNUNET_NO);
 958       break;
 959     case UMT_MSG_FRAGMENTED_COMPLETE:
 960       GNUNET_assert(NULL != udpw->frag_ctx);
 961       if (udpw->frag_ctx->cont != NULL )
 962         udpw->frag_ctx->cont (udpw->frag_ctx->cont_cls, &udpw->session->target,
 963             GNUNET_OK, udpw->frag_ctx->payload_size,
 964             udpw->frag_ctx->on_wire_size);
 965       GNUNET_STATISTICS_update (plugin->env->stats,
 966           "# UDP, fragmented msgs, messages, sent, success", 1, GNUNET_NO);
 967       GNUNET_STATISTICS_update (plugin->env->stats,
 968           "# UDP, fragmented msgs, bytes payload, sent, success",
 969           udpw->payload_size, GNUNET_NO);
 970       GNUNET_STATISTICS_update (plugin->env->stats,
 971           "# UDP, fragmented msgs, bytes overhead, sent, success", overhead,
 972           GNUNET_NO);
 973       GNUNET_STATISTICS_update (plugin->env->stats,
 974           "# UDP, total, bytes overhead, sent", overhead, GNUNET_NO);
 975       GNUNET_STATISTICS_update (plugin->env->stats,
 976           "# UDP, total, bytes payload, sent", udpw->payload_size, GNUNET_NO);
 977       GNUNET_STATISTICS_update (plugin->env->stats,
 978           "# UDP, fragmented msgs, messages, pending", -1, GNUNET_NO);
 979       break;
 980     case UMT_MSG_FRAGMENTED:
 981       /* Fragmented message: enqueue next fragment */
 982       if (NULL != udpw->cont)
 983         udpw->cont (udpw->cont_cls, &udpw->session->target, result,
 984             udpw->payload_size, udpw->msg_size);
 985       GNUNET_STATISTICS_update (plugin->env->stats,
 986           "# UDP, fragmented msgs, fragments, sent, success", 1, GNUNET_NO);
 987       GNUNET_STATISTICS_update (plugin->env->stats,
 988           "# UDP, fragmented msgs, fragments bytes, sent, success",
 989           udpw->msg_size, GNUNET_NO);
 990       break;
 991     case UMT_MSG_ACK:
 992       /* No continuation */
 993       GNUNET_STATISTICS_update (plugin->env->stats,
 994           "# UDP, ACK msgs, messages, sent, success", 1, GNUNET_NO);
 995       GNUNET_STATISTICS_update (plugin->env->stats,
 996           "# UDP, ACK msgs, bytes overhead, sent, success", overhead,
 997           GNUNET_NO);
 998       GNUNET_STATISTICS_update (plugin->env->stats,
 999           "# UDP, total, bytes overhead, sent", overhead, GNUNET_NO);
1000       break;
1001     default:
1002       GNUNET_break(0);
1003       break;
1004     }
1005     break;
1006   case GNUNET_SYSERR:
1007     switch (udpw->msg_type)
1008     {
1009     case UMT_MSG_UNFRAGMENTED:
1010       /* Unfragmented message: failed to send */
1011       if (NULL != udpw->cont)
1012         udpw->cont (udpw->cont_cls, &udpw->session->target, result,
1013             udpw->payload_size, overhead);
1014       GNUNET_STATISTICS_update (plugin->env->stats,
1015           "# UDP, unfragmented msgs, messages, sent, failure", 1, GNUNET_NO);
1016       GNUNET_STATISTICS_update (plugin->env->stats,
1017           "# UDP, unfragmented msgs, bytes payload, sent, failure",
1018           udpw->payload_size, GNUNET_NO);
1019       GNUNET_STATISTICS_update (plugin->env->stats,
1020           "# UDP, unfragmented msgs, bytes overhead, sent, failure", overhead,
1021           GNUNET_NO);
1022       break;
1023     case UMT_MSG_FRAGMENTED_COMPLETE:
1024       GNUNET_assert(NULL != udpw->frag_ctx);
1025       if (udpw->frag_ctx->cont != NULL )
1026         udpw->frag_ctx->cont (udpw->frag_ctx->cont_cls, &udpw->session->target,
1027             GNUNET_SYSERR, udpw->frag_ctx->payload_size,
1028             udpw->frag_ctx->on_wire_size);
1029       GNUNET_STATISTICS_update (plugin->env->stats,
1030           "# UDP, fragmented msgs, messages, sent, failure", 1, GNUNET_NO);
1031       GNUNET_STATISTICS_update (plugin->env->stats,
1032           "# UDP, fragmented msgs, bytes payload, sent, failure",
1033           udpw->payload_size, GNUNET_NO);
1034       GNUNET_STATISTICS_update (plugin->env->stats,
1035           "# UDP, fragmented msgs, bytes payload, sent, failure", overhead,
1036           GNUNET_NO);
1037       GNUNET_STATISTICS_update (plugin->env->stats,
1038           "# UDP, fragmented msgs, bytes payload, sent, failure", overhead,
1039           GNUNET_NO);
1040       GNUNET_STATISTICS_update (plugin->env->stats,
1041           "# UDP, fragmented msgs, messages, pending", -1, GNUNET_NO);
1042       break;
1043     case UMT_MSG_FRAGMENTED:
1044       GNUNET_assert(NULL != udpw->frag_ctx);
1045       /* Fragmented message: failed to send */
1046       GNUNET_STATISTICS_update (plugin->env->stats,
1047           "# UDP, fragmented msgs, fragments, sent, failure", 1, GNUNET_NO);
1048       GNUNET_STATISTICS_update (plugin->env->stats,
1049           "# UDP, fragmented msgs, fragments bytes, sent, failure",
1050           udpw->msg_size, GNUNET_NO);
1051       break;
1052     case UMT_MSG_ACK:
1053       /* ACK message: failed to send */
1054       GNUNET_STATISTICS_update (plugin->env->stats,
1055           "# UDP, ACK msgs, messages, sent, failure", 1, GNUNET_NO);
1056       break;
1057     default:
1058       GNUNET_break(0);
1059       break;
1060     }
1061     break;
1062   default:
1063     GNUNET_break(0);
1064     break;
1065   }
1066 }
1067
1068
1069 /**
1070  * Check if the given port is plausible (must be either our listen
1071  * port or our advertised port).  If it is neither, we return
1072  * #GNUNET_SYSERR.
1073  *
1074  * @param plugin global variables
1075  * @param in_port port number to check
1076  * @return #GNUNET_OK if port is either open_port or adv_port
1077  */
1078 static int
1079 check_port (struct Plugin *plugin,
1080             uint16_t in_port)
1081 {
1082   if ((in_port == plugin->port) || (in_port == plugin->aport))
1083     return GNUNET_OK;
1084   return GNUNET_SYSERR;
1085 }
1086
1087
1088 /**
1089  * Function that will be called to check if a binary address for this
1090  * plugin is well-formed and corresponds to an address for THIS peer
1091  * (as per our configuration).  Naturally, if absolutely necessary,
1092  * plugins can be a bit conservative in their answer, but in general
1093  * plugins should make sure that the address does not redirect
1094  * traffic to a 3rd party that might try to man-in-the-middle our
1095  * traffic.
1096  *
1097  * @param cls closure, should be our handle to the Plugin
1098  * @param addr pointer to the address
1099  * @param addrlen length of @a addr
1100  * @return #GNUNET_OK if this is a plausible address for this peer
1101  *         and transport, #GNUNET_SYSERR if not
1102  */
1103 static int
1104 udp_plugin_check_address (void *cls,
1105                           const void *addr,
1106                           size_t addrlen)
1107 {
1108   struct Plugin *plugin = cls;
1109   struct IPv4UdpAddress *v4;
1110   struct IPv6UdpAddress *v6;
1111
1112   if ((addrlen != sizeof(struct IPv4UdpAddress))
1113       && (addrlen != sizeof(struct IPv6UdpAddress)))
1114   {
1115     GNUNET_break_op(0);
1116     return GNUNET_SYSERR;
1117   }
1118   if (addrlen == sizeof(struct IPv4UdpAddress))
1119   {
1120     v4 = (struct IPv4UdpAddress *) addr;
1121     if (GNUNET_OK != check_port (plugin, ntohs (v4->u4_port)))
1122       return GNUNET_SYSERR;
1123     if (GNUNET_OK
1124         != GNUNET_NAT_test_address (plugin->nat, &v4->ipv4_addr,
1125             sizeof(struct in_addr)))
1126       return GNUNET_SYSERR;
1127   }
1128   else
1129   {
1130     v6 = (struct IPv6UdpAddress *) addr;
1131     if (IN6_IS_ADDR_LINKLOCAL (&v6->ipv6_addr))
1132     {
1133       GNUNET_break_op(0);
1134       return GNUNET_SYSERR;
1135     }
1136     if (GNUNET_OK != check_port (plugin, ntohs (v6->u6_port)))
1137       return GNUNET_SYSERR;
1138     if (GNUNET_OK !=
1139         GNUNET_NAT_test_address (plugin->nat,
1140                                  &v6->ipv6_addr,
1141                                  sizeof(struct in6_addr)))
1142       return GNUNET_SYSERR;
1143   }
1144   return GNUNET_OK;
1145 }
1146
1147
1148 /**
1149  * Function to free last resources associated with a session.
1150  *
1151  * @param s session to free
1152  */
1153 static void
1154 free_session (struct Session *s)
1155 {
1156   if (NULL != s->frag_ctx)
1157   {
1158     GNUNET_FRAGMENT_context_destroy (s->frag_ctx->frag, NULL, NULL );
1159     GNUNET_free(s->frag_ctx);
1160     s->frag_ctx = NULL;
1161   }
1162   GNUNET_free(s);
1163 }
1164
1165
1166 /**
1167  * Remove a message from the transmission queue.
1168  *
1169  * @param plugin the UDP plugin
1170  * @param udpw message wrapper to queue
1171  */
1172 static void
1173 dequeue (struct Plugin *plugin,
1174          struct UDP_MessageWrapper *udpw)
1175 {
1176   struct Session *session = udpw->session;
1177
1178   if (plugin->bytes_in_buffer < udpw->msg_size)
1179   {
1180     GNUNET_break (0);
1181   }
1182   else
1183   {
1184     GNUNET_STATISTICS_update (plugin->env->stats,
1185                               "# UDP, total, bytes in buffers",
1186                               -(long long) udpw->msg_size,
1187                               GNUNET_NO);
1188     plugin->bytes_in_buffer -= udpw->msg_size;
1189   }
1190   GNUNET_STATISTICS_update (plugin->env->stats,
1191                             "# UDP, total, msgs in buffers",
1192                             -1, GNUNET_NO);
1193   if (udpw->session->address->address_length == sizeof(struct IPv4UdpAddress))
1194     GNUNET_CONTAINER_DLL_remove (plugin->ipv4_queue_head,
1195                                  plugin->ipv4_queue_tail,
1196                                  udpw);
1197   else if (udpw->session->address->address_length == sizeof(struct IPv6UdpAddress))
1198     GNUNET_CONTAINER_DLL_remove (plugin->ipv6_queue_head,
1199                                  plugin->ipv6_queue_tail,
1200                                  udpw);
1201   else
1202   {
1203     GNUNET_break (0);
1204     return;
1205   }
1206   GNUNET_assert (session->msgs_in_queue > 0);
1207   session->msgs_in_queue--;
1208   GNUNET_assert (session->bytes_in_queue >= udpw->msg_size);
1209   session->bytes_in_queue -= udpw->msg_size;
1210 }
1211
1212
1213 /**
1214  * FIXME.
1215  */
1216 static void
1217 fragmented_message_done (struct UDP_FragmentationContext *fc,
1218                          int result)
1219 {
1220   struct Plugin *plugin = fc->plugin;
1221   struct Session *s = fc->session;
1222   struct UDP_MessageWrapper *udpw;
1223   struct UDP_MessageWrapper *tmp;
1224   struct UDP_MessageWrapper dummy;
1225
1226   LOG (GNUNET_ERROR_TYPE_DEBUG,
1227        "%p : Fragmented message removed with result %s\n",
1228        fc,
1229        (result == GNUNET_SYSERR) ? "FAIL" : "SUCCESS");
1230
1231   /* Call continuation for fragmented message */
1232   memset (&dummy, 0, sizeof(dummy));
1233   dummy.msg_type = UMT_MSG_FRAGMENTED_COMPLETE;
1234   dummy.msg_size = s->frag_ctx->on_wire_size;
1235   dummy.payload_size = s->frag_ctx->payload_size;
1236   dummy.frag_ctx = s->frag_ctx;
1237   dummy.cont = NULL;
1238   dummy.cont_cls = NULL;
1239   dummy.session = s;
1240   call_continuation (&dummy, result);
1241   /* Remove leftover fragments from queue */
1242   if (s->address->address_length == sizeof(struct IPv6UdpAddress))
1243   {
1244     udpw = plugin->ipv6_queue_head;
1245     while (NULL != udpw)
1246     {
1247       tmp = udpw->next;
1248       if ((udpw->frag_ctx != NULL )&& (udpw->frag_ctx == s->frag_ctx)){
1249       dequeue (plugin, udpw);
1250       call_continuation (udpw, GNUNET_SYSERR);
1251       GNUNET_free (udpw);
1252     }
1253       udpw = tmp;
1254     }
1255   }
1256   if (s->address->address_length == sizeof(struct IPv4UdpAddress))
1257   {
1258     udpw = plugin->ipv4_queue_head;
1259     while (udpw != NULL )
1260     {
1261       tmp = udpw->next;
1262       if ((NULL != udpw->frag_ctx) && (udpw->frag_ctx == s->frag_ctx))
1263       {
1264         dequeue (plugin, udpw);
1265         call_continuation (udpw, GNUNET_SYSERR);
1266         GNUNET_free(udpw);
1267       }
1268       udpw = tmp;
1269     }
1270   }
1271   notify_session_monitor (s->plugin,
1272                           s,
1273                           GNUNET_TRANSPORT_SS_UPDATE);
1274   /* Destroy fragmentation context */
1275   GNUNET_FRAGMENT_context_destroy (fc->frag,
1276                                    &s->last_expected_msg_delay,
1277                                    &s->last_expected_ack_delay);
1278   s->frag_ctx = NULL;
1279   GNUNET_free (fc);
1280 }
1281
1282
1283 /**
1284  * Scan the heap for a receive context with the given address.
1285  *
1286  * @param cls the `struct FindReceiveContext`
1287  * @param node internal node of the heap
1288  * @param element value stored at the node (a 'struct ReceiveContext')
1289  * @param cost cost associated with the node
1290  * @return #GNUNET_YES if we should continue to iterate,
1291  *         #GNUNET_NO if not.
1292  */
1293 static int
1294 find_receive_context (void *cls,
1295                       struct GNUNET_CONTAINER_HeapNode *node,
1296                       void *element,
1297                       GNUNET_CONTAINER_HeapCostType cost)
1298 {
1299   struct FindReceiveContext *frc = cls;
1300   struct DefragContext *e = element;
1301
1302   if ((frc->addr_len == e->addr_len)
1303       && (0 == memcmp (frc->addr, e->src_addr, frc->addr_len)))
1304   {
1305     frc->rc = e;
1306     return GNUNET_NO;
1307   }
1308   return GNUNET_YES;
1309 }
1310
1311
1312 /**
1313  * Functions with this signature are called whenever we need
1314  * to close a session due to a disconnect or failure to
1315  * establish a connection.
1316  *
1317  * @param cls closure with the `struct Plugin`
1318  * @param s session to close down
1319  * @return #GNUNET_OK on success
1320  */
1321 static int
1322 udp_disconnect_session (void *cls,
1323                         struct Session *s)
1324 {
1325   struct Plugin *plugin = cls;
1326   struct UDP_MessageWrapper *udpw;
1327   struct UDP_MessageWrapper *next;
1328   struct FindReceiveContext frc;
1329
1330   GNUNET_assert (GNUNET_YES != s->in_destroy);
1331   LOG(GNUNET_ERROR_TYPE_DEBUG,
1332       "Session %p to peer `%s' address ended\n", s,
1333       GNUNET_i2s (&s->target),
1334       udp_address_to_string (NULL,
1335                              s->address->address,
1336                              s->address->address_length));
1337   /* stop timeout task */
1338   if (NULL != s->timeout_task)
1339   {
1340     GNUNET_SCHEDULER_cancel (s->timeout_task);
1341     s->timeout_task = NULL;
1342   }
1343   if (NULL != s->frag_ctx)
1344   {
1345     /* Remove fragmented message due to disconnect */
1346     fragmented_message_done (s->frag_ctx, GNUNET_SYSERR);
1347   }
1348
1349   frc.rc = NULL;
1350   frc.addr = s->address->address;
1351   frc.addr_len = s->address->address_length;
1352   /* Lookup existing receive context for this address */
1353   if (NULL != plugin->defrag_ctxs)
1354   {
1355     GNUNET_CONTAINER_heap_iterate (plugin->defrag_ctxs,
1356                                    &find_receive_context,
1357                                    &frc);
1358     if (NULL != frc.rc)
1359     {
1360       struct DefragContext *d_ctx = frc.rc;
1361
1362       GNUNET_CONTAINER_heap_remove_node (d_ctx->hnode);
1363       GNUNET_DEFRAGMENT_context_destroy (d_ctx->defrag);
1364       GNUNET_free (d_ctx);
1365     }
1366   }
1367   next = plugin->ipv4_queue_head;
1368   while (NULL != (udpw = next))
1369   {
1370     next = udpw->next;
1371     if (udpw->session == s)
1372     {
1373       dequeue (plugin, udpw);
1374       call_continuation (udpw, GNUNET_SYSERR);
1375       GNUNET_free(udpw);
1376     }
1377   }
1378   next = plugin->ipv6_queue_head;
1379   while (NULL != (udpw = next))
1380   {
1381     next = udpw->next;
1382     if (udpw->session == s)
1383     {
1384       dequeue (plugin, udpw);
1385       call_continuation (udpw, GNUNET_SYSERR);
1386       GNUNET_free(udpw);
1387     }
1388   }
1389   notify_session_monitor (s->plugin,
1390                           s,
1391                           GNUNET_TRANSPORT_SS_DONE);
1392   plugin->env->session_end (plugin->env->cls,
1393                             s->address,
1394                             s);
1395
1396   if (NULL != s->frag_ctx)
1397   {
1398     if (NULL != s->frag_ctx->cont)
1399     {
1400       s->frag_ctx->cont (s->frag_ctx->cont_cls,
1401                          &s->target,
1402                          GNUNET_SYSERR,
1403                          s->frag_ctx->payload_size,
1404                          s->frag_ctx->on_wire_size);
1405       LOG (GNUNET_ERROR_TYPE_DEBUG,
1406            "Calling continuation for fragemented message to `%s' with result SYSERR\n",
1407            GNUNET_i2s (&s->target));
1408     }
1409   }
1410
1411   GNUNET_assert (GNUNET_YES ==
1412                  GNUNET_CONTAINER_multipeermap_remove (plugin->sessions,
1413                                                        &s->target,
1414                                                        s));
1415   GNUNET_STATISTICS_set (plugin->env->stats,
1416                          "# UDP sessions active",
1417                          GNUNET_CONTAINER_multipeermap_size (plugin->sessions),
1418                          GNUNET_NO);
1419   if (s->rc > 0)
1420   {
1421     s->in_destroy = GNUNET_YES;
1422   }
1423   else
1424   {
1425     GNUNET_HELLO_address_free (s->address);
1426     free_session (s);
1427   }
1428   return GNUNET_OK;
1429 }
1430
1431
1432 /**
1433  * Function that is called to get the keepalive factor.
1434  * #GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT is divided by this number to
1435  * calculate the interval between keepalive packets.
1436  *
1437  * @param cls closure with the `struct Plugin`
1438  * @return keepalive factor
1439  */
1440 static unsigned int
1441 udp_query_keepalive_factor (void *cls)
1442 {
1443   return 15;
1444 }
1445
1446
1447 /**
1448  * Destroy a session, plugin is being unloaded.
1449  *
1450  * @param cls the `struct Plugin`
1451  * @param key hash of public key of target peer
1452  * @param value a `struct PeerSession *` to clean up
1453  * @return #GNUNET_OK (continue to iterate)
1454  */
1455 static int
1456 disconnect_and_free_it (void *cls,
1457                         const struct GNUNET_PeerIdentity *key,
1458                         void *value)
1459 {
1460   struct Plugin *plugin = cls;
1461
1462   udp_disconnect_session (plugin, value);
1463   return GNUNET_OK;
1464 }
1465
1466
1467 /**
1468  * Disconnect from a remote node.  Clean up session if we have one for
1469  * this peer.
1470  *
1471  * @param cls closure for this call (should be handle to Plugin)
1472  * @param target the peeridentity of the peer to disconnect
1473  * @return #GNUNET_OK on success, #GNUNET_SYSERR if the operation failed
1474  */
1475 static void
1476 udp_disconnect (void *cls,
1477                 const struct GNUNET_PeerIdentity *target)
1478 {
1479   struct Plugin *plugin = cls;
1480
1481   LOG (GNUNET_ERROR_TYPE_DEBUG,
1482        "Disconnecting from peer `%s'\n",
1483        GNUNET_i2s (target));
1484   /* Clean up sessions */
1485   GNUNET_CONTAINER_multipeermap_get_multiple (plugin->sessions,
1486                                               target,
1487                                               &disconnect_and_free_it,
1488                                               plugin);
1489 }
1490
1491
1492 /**
1493  * Session was idle, so disconnect it
1494  *
1495  * @param cls the `struct Session` to time out
1496  * @param tc scheduler context
1497  */
1498 static void
1499 session_timeout (void *cls,
1500                  const struct GNUNET_SCHEDULER_TaskContext *tc)
1501 {
1502   struct Session *s = cls;
1503   struct Plugin *plugin = s->plugin;
1504   struct GNUNET_TIME_Relative left;
1505
1506   s->timeout_task = NULL;
1507   left = GNUNET_TIME_absolute_get_remaining (s->timeout);
1508   if (left.rel_value_us > 0)
1509   {
1510     /* not actually our turn yet, but let's at least update
1511        the monitor, it may think we're about to die ... */
1512     notify_session_monitor (s->plugin,
1513                             s,
1514                             GNUNET_TRANSPORT_SS_UPDATE);
1515     s->timeout_task = GNUNET_SCHEDULER_add_delayed (left,
1516                                                     &session_timeout,
1517                                                     s);
1518     return;
1519   }
1520   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1521               "Session %p was idle for %s, disconnecting\n",
1522               s,
1523               GNUNET_STRINGS_relative_time_to_string (UDP_SESSION_TIME_OUT,
1524                                                       GNUNET_YES));
1525   /* call session destroy function */
1526   udp_disconnect_session (plugin, s);
1527 }
1528
1529
1530 /**
1531  * Increment session timeout due to activity
1532  *
1533  * @param s session to reschedule timeout activity for
1534  */
1535 static void
1536 reschedule_session_timeout (struct Session *s)
1537 {
1538   if (GNUNET_YES == s->in_destroy)
1539     return;
1540   GNUNET_assert(NULL != s->timeout_task);
1541   s->timeout = GNUNET_TIME_relative_to_absolute (UDP_SESSION_TIME_OUT);
1542 }
1543
1544
1545 /**
1546  * FIXME.
1547  */
1548 static struct Session *
1549 create_session (struct Plugin *plugin,
1550                 const struct GNUNET_HELLO_Address *address)
1551 {
1552   struct Session *s;
1553
1554   s = GNUNET_new (struct Session);
1555   s->plugin = plugin;
1556   s->address = GNUNET_HELLO_address_copy (address);
1557   s->target = address->peer;
1558   s->last_expected_ack_delay = GNUNET_TIME_relative_multiply (
1559       GNUNET_TIME_UNIT_MILLISECONDS, 250);
1560   s->last_expected_msg_delay = GNUNET_TIME_UNIT_MILLISECONDS;
1561   s->flow_delay_from_other_peer = GNUNET_TIME_UNIT_ZERO_ABS;
1562   s->flow_delay_for_other_peer = GNUNET_TIME_UNIT_ZERO;
1563   s->timeout = GNUNET_TIME_relative_to_absolute (UDP_SESSION_TIME_OUT);
1564   s->timeout_task = GNUNET_SCHEDULER_add_delayed (UDP_SESSION_TIME_OUT,
1565                                                   &session_timeout, s);
1566   return s;
1567 }
1568
1569
1570 /**
1571  * Function obtain the network type for a session
1572  *
1573  * @param cls closure ('struct Plugin*')
1574  * @param session the session
1575  * @return the network type
1576  */
1577 static enum GNUNET_ATS_Network_Type
1578 udp_get_network (void *cls,
1579                  struct Session *session)
1580 {
1581   return ntohl (session->ats.value);
1582 }
1583
1584
1585 /**
1586  * Closure for #session_cmp_it().
1587  */
1588 struct SessionCompareContext
1589 {
1590   /**
1591    * Set to session matching the address.
1592    */
1593   struct Session *res;
1594
1595   /**
1596    * Address we are looking for.
1597    */
1598   const struct GNUNET_HELLO_Address *address;
1599 };
1600
1601
1602 /**
1603  * Find a session with a matching address.
1604  *
1605  * @param cls the `struct SessionCompareContext *`
1606  * @param key peer identity (unused)
1607  * @param value the `struct Session *`
1608  * @return #GNUNET_NO if we found the session, #GNUNET_OK if not
1609  */
1610 static int
1611 session_cmp_it (void *cls,
1612                 const struct GNUNET_PeerIdentity *key,
1613                 void *value)
1614 {
1615   struct SessionCompareContext *cctx = cls;
1616   const struct GNUNET_HELLO_Address *address = cctx->address;
1617   struct Session *s = value;
1618
1619   LOG (GNUNET_ERROR_TYPE_DEBUG,
1620        "Comparing address %s <-> %s\n",
1621        udp_address_to_string (NULL,
1622                               address->address,
1623                               address->address_length),
1624        udp_address_to_string (NULL,
1625                               s->address->address,
1626                               s->address->address_length));
1627   if (0 == GNUNET_HELLO_address_cmp(s->address, cctx->address))
1628   {
1629     cctx->res = s;
1630     return GNUNET_NO;
1631   }
1632   return GNUNET_YES;
1633 }
1634
1635
1636 /**
1637  * Creates a new outbound session the transport service will use to
1638  * send data to the peer
1639  *
1640  * @param cls the plugin
1641  * @param address the address
1642  * @return the session or NULL of max connections exceeded
1643  */
1644 static struct Session *
1645 udp_plugin_lookup_session (void *cls,
1646                            const struct GNUNET_HELLO_Address *address)
1647 {
1648   struct Plugin * plugin = cls;
1649   struct IPv6UdpAddress *udp_a6;
1650   struct IPv4UdpAddress *udp_a4;
1651   struct SessionCompareContext cctx;
1652
1653   if ( (NULL == address->address) ||
1654        ((address->address_length != sizeof (struct IPv4UdpAddress)) &&
1655         (address->address_length != sizeof (struct IPv6UdpAddress))))
1656   {
1657     LOG (GNUNET_ERROR_TYPE_WARNING,
1658          _("Trying to create session for address of unexpected length %u (should be %u or %u)\n"),
1659          address->address_length,
1660          sizeof (struct IPv4UdpAddress),
1661          sizeof (struct IPv6UdpAddress));
1662     return NULL;
1663   }
1664
1665   if (address->address_length == sizeof(struct IPv4UdpAddress))
1666   {
1667     if (plugin->sockv4 == NULL)
1668       return NULL;
1669     udp_a4 = (struct IPv4UdpAddress *) address->address;
1670     if (udp_a4->u4_port == 0)
1671       return NULL;
1672   }
1673
1674   if (address->address_length == sizeof(struct IPv6UdpAddress))
1675   {
1676     if (plugin->sockv6 == NULL)
1677       return NULL;
1678     udp_a6 = (struct IPv6UdpAddress *) address->address;
1679     if (udp_a6->u6_port == 0)
1680       return NULL;
1681   }
1682
1683   /* check if session already exists */
1684   cctx.address = address;
1685   cctx.res = NULL;
1686   LOG (GNUNET_ERROR_TYPE_DEBUG,
1687        "Looking for existing session for peer `%s' `%s' \n",
1688        GNUNET_i2s (&address->peer),
1689        udp_address_to_string(NULL, address->address, address->address_length));
1690   GNUNET_CONTAINER_multipeermap_get_multiple (plugin->sessions, &address->peer,
1691       session_cmp_it, &cctx);
1692   if (cctx.res != NULL )
1693   {
1694     LOG (GNUNET_ERROR_TYPE_DEBUG,
1695          "Found existing session %p\n",
1696          cctx.res);
1697     return cctx.res;
1698   }
1699   return NULL;
1700 }
1701
1702
1703 /**
1704  * Context to lookup a session based on a IP address
1705  */
1706 struct LookupContext
1707 {
1708   /**
1709    * The result
1710    */
1711   struct Session *res;
1712
1713   /**
1714    * The socket address
1715    */
1716   const struct sockaddr *address;
1717
1718   /**
1719    * The socket address length
1720    */
1721   size_t addr_len;
1722
1723   /**
1724    * Is a fragmentation context required for the session
1725    */
1726   int must_have_frag_ctx;
1727 };
1728
1729
1730 /**
1731  * Find a session with a matching address.
1732  * FIXME: very similar code to #udp_plugin_lookup_session() above.
1733  * Unify?
1734  *
1735  * @param cls the `struct LookupContext *`
1736  * @param key peer identity (unused)
1737  * @param value the `struct Session *`
1738  * @return #GNUNET_NO if we found the session, #GNUNET_OK if not
1739  */
1740 static int
1741 lookup_session_by_sockaddr_it (void *cls,
1742                                const struct GNUNET_PeerIdentity *key,
1743                                void *value)
1744 {
1745   struct LookupContext *l_ctx = cls;
1746   struct Session *s = value;
1747   struct IPv4UdpAddress u4;
1748   struct IPv6UdpAddress u6;
1749   void *arg;
1750   size_t args;
1751
1752   /* convert address */
1753   switch (l_ctx->address->sa_family)
1754   {
1755   case AF_INET:
1756     GNUNET_assert(l_ctx->addr_len == sizeof(struct sockaddr_in));
1757     memset (&u4, 0, sizeof(u4));
1758     u6.options = htonl (0);
1759     u4.ipv4_addr = ((struct sockaddr_in *) l_ctx->address)->sin_addr.s_addr;
1760     u4.u4_port = ((struct sockaddr_in *) l_ctx->address)->sin_port;
1761     arg = &u4;
1762     args = sizeof(u4);
1763     break;
1764   case AF_INET6:
1765     GNUNET_assert(l_ctx->addr_len == sizeof(struct sockaddr_in6));
1766     memset (&u6, 0, sizeof(u6));
1767     u6.options = htonl (0);
1768     u6.ipv6_addr = ((struct sockaddr_in6 *) l_ctx->address)->sin6_addr;
1769     u6.u6_port = ((struct sockaddr_in6 *) l_ctx->address)->sin6_port;
1770     arg = &u6;
1771     args = sizeof(u6);
1772     break;
1773   default:
1774     GNUNET_break(0);
1775     return GNUNET_YES;
1776   }
1777   if ( (GNUNET_YES == l_ctx->must_have_frag_ctx) &&
1778        (NULL == s->frag_ctx))
1779     return GNUNET_YES;
1780
1781   /* Does not compare peer identities but addresses */
1782   if ((args == s->address->address_length) &&
1783       (0 == memcmp (arg, s->address->address, args)))
1784   {
1785     l_ctx->res = s;
1786     return GNUNET_NO;
1787   }
1788   return GNUNET_YES;
1789 }
1790
1791
1792 static struct Session *
1793 udp_plugin_create_session (void *cls,
1794                            const struct GNUNET_HELLO_Address *address)
1795 {
1796   struct Plugin *plugin = cls;
1797   struct Session *s;
1798   struct IPv4UdpAddress *udp_v4;
1799   struct IPv6UdpAddress *udp_v6;
1800
1801   s = create_session (plugin, address);
1802   if (sizeof (struct IPv4UdpAddress) == address->address_length)
1803   {
1804     struct sockaddr_in v4;
1805
1806     udp_v4 = (struct IPv4UdpAddress *) address->address;
1807     memset (&v4, '\0', sizeof (v4));
1808     v4.sin_family = AF_INET;
1809 #if HAVE_SOCKADDR_IN_SIN_LEN
1810     v4.sin_len = sizeof (struct sockaddr_in);
1811 #endif
1812     v4.sin_port = udp_v4->u4_port;
1813     v4.sin_addr.s_addr = udp_v4->ipv4_addr;
1814     s->ats.type = htonl (GNUNET_ATS_NETWORK_TYPE);
1815     s->ats.value = htonl (plugin->env->get_address_type (plugin->env->cls,
1816                                                          (const struct sockaddr *) &v4,
1817                                                          sizeof (v4)));
1818   }
1819   else if (sizeof (struct IPv6UdpAddress) == address->address_length)
1820   {
1821     struct sockaddr_in6 v6;
1822     udp_v6 = (struct IPv6UdpAddress *) address->address;
1823     memset (&v6, '\0', sizeof (v6));
1824     v6.sin6_family = AF_INET6;
1825 #if HAVE_SOCKADDR_IN_SIN_LEN
1826     v6.sin6_len = sizeof (struct sockaddr_in6);
1827 #endif
1828     v6.sin6_port = udp_v6->u6_port;
1829     v6.sin6_addr = udp_v6->ipv6_addr;
1830     s->ats.type = htonl (GNUNET_ATS_NETWORK_TYPE);
1831     s->ats.value = htonl (plugin->env->get_address_type (plugin->env->cls,
1832                                                          (const struct sockaddr *) &v6,
1833                                                          sizeof (v6)));
1834   }
1835
1836   if (NULL == s)
1837     return NULL; /* protocol not supported or address invalid */
1838   LOG(GNUNET_ERROR_TYPE_DEBUG,
1839       "Creating new %s session %p for peer `%s' address `%s'\n",
1840       GNUNET_HELLO_address_check_option (address, GNUNET_HELLO_ADDRESS_INFO_INBOUND) ? "inbound" : "outbound",
1841       s, GNUNET_i2s (&address->peer),
1842       udp_address_to_string( NULL,address->address,address->address_length));
1843   GNUNET_assert(
1844       GNUNET_OK == GNUNET_CONTAINER_multipeermap_put (plugin->sessions, &s->target, s, GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
1845   GNUNET_STATISTICS_set (plugin->env->stats, "# UDP sessions active",
1846       GNUNET_CONTAINER_multipeermap_size (plugin->sessions), GNUNET_NO);
1847   return s;
1848 }
1849
1850
1851 /**
1852  * Function that will be called whenever the transport service wants to
1853  * notify the plugin that a session is still active and in use and
1854  * therefore the session timeout for this session has to be updated
1855  *
1856  * @param cls closure
1857  * @param peer which peer was the session for
1858  * @param session which session is being updated
1859  */
1860 static void
1861 udp_plugin_update_session_timeout (void *cls,
1862                                    const struct GNUNET_PeerIdentity *peer,
1863                                    struct Session *session)
1864 {
1865   struct Plugin *plugin = cls;
1866
1867   if (GNUNET_YES !=
1868       GNUNET_CONTAINER_multipeermap_contains_value (plugin->sessions,
1869                                                     peer,
1870                                                     session))
1871   {
1872     GNUNET_break(0);
1873     return;
1874   }
1875   /* Reschedule session timeout */
1876   reschedule_session_timeout (session);
1877 }
1878
1879
1880 /**
1881  * Creates a new outbound session the transport service will use to send data to the
1882  * peer
1883  *
1884  * @param cls the plugin
1885  * @param address the address
1886  * @return the session or NULL of max connections exceeded
1887  */
1888 static struct Session *
1889 udp_plugin_get_session (void *cls,
1890                         const struct GNUNET_HELLO_Address *address)
1891 {
1892   struct Session *s;
1893
1894   if (NULL == address)
1895   {
1896     GNUNET_break(0);
1897     return NULL;
1898   }
1899   if ( (address->address_length != sizeof(struct IPv4UdpAddress)) &&
1900        (address->address_length != sizeof(struct IPv6UdpAddress)) )
1901     return NULL;
1902
1903   /* otherwise create new */
1904   if (NULL != (s = udp_plugin_lookup_session (cls, address)))
1905     return s;
1906   return udp_plugin_create_session (cls, address);
1907 }
1908
1909
1910 /**
1911  * Enqueue a message for transmission.
1912  *
1913  * @param plugin the UDP plugin
1914  * @param udpw message wrapper to queue
1915  */
1916 static void
1917 enqueue (struct Plugin *plugin,
1918          struct UDP_MessageWrapper *udpw)
1919 {
1920   struct Session *session = udpw->session;
1921
1922   if (plugin->bytes_in_buffer + udpw->msg_size > INT64_MAX)
1923   {
1924     GNUNET_break (0);
1925   }
1926   else
1927   {
1928     GNUNET_STATISTICS_update (plugin->env->stats,
1929         "# UDP, total, bytes in buffers", udpw->msg_size, GNUNET_NO);
1930     plugin->bytes_in_buffer += udpw->msg_size;
1931   }
1932   GNUNET_STATISTICS_update (plugin->env->stats,
1933                             "# UDP, total, msgs in buffers",
1934                             1, GNUNET_NO);
1935   if (udpw->session->address->address_length == sizeof (struct IPv4UdpAddress))
1936     GNUNET_CONTAINER_DLL_insert(plugin->ipv4_queue_head,
1937                                 plugin->ipv4_queue_tail,
1938                                 udpw);
1939   else if (udpw->session->address->address_length == sizeof (struct IPv6UdpAddress))
1940     GNUNET_CONTAINER_DLL_insert (plugin->ipv6_queue_head,
1941                                  plugin->ipv6_queue_tail,
1942                                  udpw);
1943   else
1944   {
1945     GNUNET_break (0);
1946     return;
1947   }
1948   session->msgs_in_queue++;
1949   session->bytes_in_queue += udpw->msg_size;
1950 }
1951
1952
1953 /**
1954  * Fragment message was transmitted via UDP, let fragmentation know
1955  * to send the next fragment now.
1956  *
1957  * @param cls the `struct UDPMessageWrapper *` of the fragment
1958  * @param target destination peer (ignored)
1959  * @param result #GNUNET_OK on success (ignored)
1960  * @param payload bytes payload sent
1961  * @param physical bytes physical sent
1962  */
1963 static void
1964 send_next_fragment (void *cls,
1965                     const struct GNUNET_PeerIdentity *target,
1966                     int result,
1967                     size_t payload,
1968                     size_t physical)
1969 {
1970   struct UDP_MessageWrapper *udpw = cls;
1971
1972   GNUNET_FRAGMENT_context_transmission_done (udpw->frag_ctx->frag);
1973 }
1974
1975
1976 /**
1977  * Function that is called with messages created by the fragmentation
1978  * module.  In the case of the 'proc' callback of the
1979  * #GNUNET_FRAGMENT_context_create() function, this function must
1980  * eventually call #GNUNET_FRAGMENT_context_transmission_done().
1981  *
1982  * @param cls closure, the 'struct FragmentationContext'
1983  * @param msg the message that was created
1984  */
1985 static void
1986 enqueue_fragment (void *cls,
1987                   const struct GNUNET_MessageHeader *msg)
1988 {
1989   struct UDP_FragmentationContext *frag_ctx = cls;
1990   struct Plugin *plugin = frag_ctx->plugin;
1991   struct UDP_MessageWrapper * udpw;
1992   size_t msg_len = ntohs (msg->size);
1993
1994   LOG (GNUNET_ERROR_TYPE_DEBUG,
1995        "Enqueuing fragment with %u bytes\n",
1996        msg_len);
1997   frag_ctx->fragments_used++;
1998   udpw = GNUNET_malloc (sizeof (struct UDP_MessageWrapper) + msg_len);
1999   udpw->session = frag_ctx->session;
2000   udpw->msg_buf = (char *) &udpw[1];
2001   udpw->msg_size = msg_len;
2002   udpw->payload_size = msg_len; /*FIXME: minus fragment overhead */
2003   udpw->cont = &send_next_fragment;
2004   udpw->cont_cls = udpw;
2005   udpw->timeout = frag_ctx->timeout;
2006   udpw->frag_ctx = frag_ctx;
2007   udpw->msg_type = UMT_MSG_FRAGMENTED;
2008   memcpy (udpw->msg_buf, msg, msg_len);
2009   enqueue (plugin, udpw);
2010   schedule_select (plugin);
2011 }
2012
2013
2014 /**
2015  * Function that can be used by the transport service to transmit
2016  * a message using the plugin.   Note that in the case of a
2017  * peer disconnecting, the continuation MUST be called
2018  * prior to the disconnect notification itself.  This function
2019  * will be called with this peer's HELLO message to initiate
2020  * a fresh connection to another peer.
2021  *
2022  * @param cls closure
2023  * @param s which session must be used
2024  * @param msgbuf the message to transmit
2025  * @param msgbuf_size number of bytes in 'msgbuf'
2026  * @param priority how important is the message (most plugins will
2027  *                 ignore message priority and just FIFO)
2028  * @param to how long to wait at most for the transmission (does not
2029  *                require plugins to discard the message after the timeout,
2030  *                just advisory for the desired delay; most plugins will ignore
2031  *                this as well)
2032  * @param cont continuation to call once the message has
2033  *        been transmitted (or if the transport is ready
2034  *        for the next transmission call; or if the
2035  *        peer disconnected...); can be NULL
2036  * @param cont_cls closure for cont
2037  * @return number of bytes used (on the physical network, with overheads);
2038  *         -1 on hard errors (i.e. address invalid); 0 is a legal value
2039  *         and does NOT mean that the message was not transmitted (DV)
2040  */
2041 static ssize_t
2042 udp_plugin_send (void *cls,
2043                  struct Session *s,
2044                  const char *msgbuf,
2045                  size_t msgbuf_size,
2046                  unsigned int priority,
2047                  struct GNUNET_TIME_Relative to,
2048                  GNUNET_TRANSPORT_TransmitContinuation cont,
2049                  void *cont_cls)
2050 {
2051   struct Plugin *plugin = cls;
2052   size_t udpmlen = msgbuf_size + sizeof(struct UDPMessage);
2053   struct UDP_FragmentationContext * frag_ctx;
2054   struct UDP_MessageWrapper * udpw;
2055   struct UDPMessage *udp;
2056   char mbuf[udpmlen];
2057   GNUNET_assert(plugin != NULL);
2058   GNUNET_assert(s != NULL);
2059
2060   if ( (s->address->address_length == sizeof(struct IPv6UdpAddress)) &&
2061        (plugin->sockv6 == NULL) )
2062     return GNUNET_SYSERR;
2063   if ( (s->address->address_length == sizeof(struct IPv4UdpAddress)) &&
2064        (plugin->sockv4 == NULL) )
2065     return GNUNET_SYSERR;
2066   if (udpmlen >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
2067   {
2068     GNUNET_break(0);
2069     return GNUNET_SYSERR;
2070   }
2071   if (GNUNET_YES !=
2072       GNUNET_CONTAINER_multipeermap_contains_value (plugin->sessions,
2073                                                     &s->target,
2074                                                     s))
2075   {
2076     GNUNET_break(0);
2077     return GNUNET_SYSERR;
2078   }
2079   LOG (GNUNET_ERROR_TYPE_DEBUG,
2080        "UDP transmits %u-byte message to `%s' using address `%s'\n",
2081        udpmlen,
2082        GNUNET_i2s (&s->target),
2083        udp_address_to_string (NULL,
2084                               s->address->address,
2085                               s->address->address_length));
2086
2087   /* Message */
2088   udp = (struct UDPMessage *) mbuf;
2089   udp->header.size = htons (udpmlen);
2090   udp->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_MESSAGE);
2091   udp->reserved = htonl (0);
2092   udp->sender = *plugin->env->my_identity;
2093
2094   /* We do not update the session time out here!
2095    * Otherwise this session will not timeout since we send keep alive before
2096    * session can timeout
2097    *
2098    * For UDP we update session timeout only on receive, this will cover keep
2099    * alives, since remote peer will reply with keep alive response!
2100    */
2101   if (udpmlen <= UDP_MTU)
2102   {
2103     /* unfragmented message */
2104     udpw = GNUNET_malloc (sizeof (struct UDP_MessageWrapper) + udpmlen);
2105     udpw->session = s;
2106     udpw->msg_buf = (char *) &udpw[1];
2107     udpw->msg_size = udpmlen; /* message size with UDP overhead */
2108     udpw->payload_size = msgbuf_size; /* message size without UDP overhead */
2109     udpw->timeout = GNUNET_TIME_absolute_add (GNUNET_TIME_absolute_get (), to);
2110     udpw->cont = cont;
2111     udpw->cont_cls = cont_cls;
2112     udpw->frag_ctx = NULL;
2113     udpw->msg_type = UMT_MSG_UNFRAGMENTED;
2114     memcpy (udpw->msg_buf, udp, sizeof(struct UDPMessage));
2115     memcpy (&udpw->msg_buf[sizeof(struct UDPMessage)], msgbuf, msgbuf_size);
2116     enqueue (plugin, udpw);
2117
2118     GNUNET_STATISTICS_update (plugin->env->stats,
2119         "# UDP, unfragmented msgs, messages, attempt", 1, GNUNET_NO);
2120     GNUNET_STATISTICS_update (plugin->env->stats,
2121                               "# UDP, unfragmented msgs, bytes payload, attempt",
2122                               udpw->payload_size,
2123                               GNUNET_NO);
2124   }
2125   else
2126   {
2127     /* fragmented message */
2128     if (s->frag_ctx != NULL)
2129       return GNUNET_SYSERR;
2130     memcpy (&udp[1], msgbuf, msgbuf_size);
2131     frag_ctx = GNUNET_new (struct UDP_FragmentationContext);
2132     frag_ctx->plugin = plugin;
2133     frag_ctx->session = s;
2134     frag_ctx->cont = cont;
2135     frag_ctx->cont_cls = cont_cls;
2136     frag_ctx->timeout = GNUNET_TIME_absolute_add (GNUNET_TIME_absolute_get (),
2137         to);
2138     frag_ctx->payload_size = msgbuf_size; /* unfragmented message size without UDP overhead */
2139     frag_ctx->on_wire_size = 0; /* bytes with UDP and fragmentation overhead */
2140     frag_ctx->frag = GNUNET_FRAGMENT_context_create (plugin->env->stats,
2141                                                      UDP_MTU,
2142                                                      &plugin->tracker,
2143                                                      s->last_expected_msg_delay,
2144                                                      s->last_expected_ack_delay,
2145                                                      &udp->header,
2146                                                      &enqueue_fragment,
2147                                                      frag_ctx);
2148     s->frag_ctx = frag_ctx;
2149     GNUNET_STATISTICS_update (plugin->env->stats,
2150                               "# UDP, fragmented msgs, messages, pending",
2151                               1,
2152                               GNUNET_NO);
2153     GNUNET_STATISTICS_update (plugin->env->stats,
2154                               "# UDP, fragmented msgs, messages, attempt",
2155                               1,
2156                               GNUNET_NO);
2157     GNUNET_STATISTICS_update (plugin->env->stats,
2158                               "# UDP, fragmented msgs, bytes payload, attempt",
2159                               frag_ctx->payload_size,
2160                               GNUNET_NO);
2161   }
2162   notify_session_monitor (s->plugin,
2163                           s,
2164                           GNUNET_TRANSPORT_SS_UPDATE);
2165   schedule_select (plugin);
2166   return udpmlen;
2167 }
2168
2169
2170 /**
2171  * Our external IP address/port mapping has changed.
2172  *
2173  * @param cls closure, the `struct LocalAddrList`
2174  * @param add_remove #GNUNET_YES to mean the new public IP address, #GNUNET_NO to mean
2175  *     the previous (now invalid) one
2176  * @param addr either the previous or the new public IP address
2177  * @param addrlen actual lenght of the address
2178  */
2179 static void
2180 udp_nat_port_map_callback (void *cls,
2181                            int add_remove,
2182                            const struct sockaddr *addr,
2183                            socklen_t addrlen)
2184 {
2185   struct Plugin *plugin = cls;
2186   struct GNUNET_HELLO_Address *address;
2187   struct IPv4UdpAddress u4;
2188   struct IPv6UdpAddress u6;
2189   void *arg;
2190   size_t args;
2191
2192   LOG (GNUNET_ERROR_TYPE_INFO,
2193        "NAT notification to %s address `%s'\n",
2194        (GNUNET_YES == add_remove) ? "add" : "remove",
2195        GNUNET_a2s (addr, addrlen));
2196
2197   /* convert 'address' to our internal format */
2198   switch (addr->sa_family)
2199   {
2200   case AF_INET:
2201     GNUNET_assert(addrlen == sizeof(struct sockaddr_in));
2202     memset (&u4, 0, sizeof(u4));
2203     u4.options = htonl (plugin->myoptions);
2204     u4.ipv4_addr = ((struct sockaddr_in *) addr)->sin_addr.s_addr;
2205     u4.u4_port = ((struct sockaddr_in *) addr)->sin_port;
2206     if (0 == ((struct sockaddr_in *) addr)->sin_port)
2207       return;
2208     arg = &u4;
2209     args = sizeof(struct IPv4UdpAddress);
2210     break;
2211   case AF_INET6:
2212     GNUNET_assert(addrlen == sizeof(struct sockaddr_in6));
2213     memset (&u6, 0, sizeof(u6));
2214     u6.options = htonl (plugin->myoptions);
2215     if (0 == ((struct sockaddr_in6 *) addr)->sin6_port)
2216       return;
2217     memcpy (&u6.ipv6_addr, &((struct sockaddr_in6 *) addr)->sin6_addr,
2218         sizeof(struct in6_addr));
2219     u6.u6_port = ((struct sockaddr_in6 *) addr)->sin6_port;
2220     arg = &u6;
2221     args = sizeof(struct IPv6UdpAddress);
2222     break;
2223   default:
2224     GNUNET_break(0);
2225     return;
2226   }
2227   /* modify our published address list */
2228   address = GNUNET_HELLO_address_allocate (plugin->env->my_identity,
2229                                            PLUGIN_NAME,
2230                                            arg, args,
2231                                            GNUNET_HELLO_ADDRESS_INFO_NONE);
2232   plugin->env->notify_address (plugin->env->cls, add_remove, address);
2233   GNUNET_HELLO_address_free (address);
2234 }
2235
2236
2237 /**
2238  * Message tokenizer has broken up an incomming message. Pass it on
2239  * to the service.
2240  *
2241  * @param cls the `struct Plugin *`
2242  * @param client the `struct SourceInformation *`
2243  * @param hdr the actual message
2244  * @return #GNUNET_OK (always)
2245  */
2246 static int
2247 process_inbound_tokenized_messages (void *cls,
2248                                     void *client,
2249                                     const struct GNUNET_MessageHeader *hdr)
2250 {
2251   struct Plugin *plugin = cls;
2252   struct SourceInformation *si = client;
2253   struct GNUNET_TIME_Relative delay;
2254
2255   GNUNET_assert(si->session != NULL);
2256   if (GNUNET_YES == si->session->in_destroy)
2257     return GNUNET_OK;
2258   /* setup ATS */
2259   GNUNET_break (ntohl (si->session->ats.value) != GNUNET_ATS_NET_UNSPECIFIED);
2260   reschedule_session_timeout (si->session);
2261   delay = plugin->env->receive (plugin->env->cls,
2262                                 si->session->address,
2263                                 si->session,
2264                                 hdr);
2265   plugin->env->update_address_metrics (plugin->env->cls,
2266                                        si->session->address,
2267                                        si->session,
2268                                        &si->session->ats, 1);
2269   si->session->flow_delay_for_other_peer = delay;
2270   return GNUNET_OK;
2271 }
2272
2273
2274 /**
2275  * We've received a UDP Message.  Process it (pass contents to main service).
2276  *
2277  * @param plugin plugin context
2278  * @param msg the message
2279  * @param sender_addr sender address
2280  * @param sender_addr_len number of bytes in @a sender_addr
2281  */
2282 static void
2283 process_udp_message (struct Plugin *plugin,
2284                      const struct UDPMessage *msg,
2285                      const struct sockaddr *sender_addr,
2286                      socklen_t sender_addr_len)
2287 {
2288   struct SourceInformation si;
2289   struct Session * s;
2290   struct GNUNET_HELLO_Address *address;
2291   struct IPv4UdpAddress u4;
2292   struct IPv6UdpAddress u6;
2293   const void *arg;
2294   size_t args;
2295
2296   if (0 != ntohl (msg->reserved))
2297   {
2298     GNUNET_break_op(0);
2299     return;
2300   }
2301   if (ntohs (msg->header.size)
2302       < sizeof(struct GNUNET_MessageHeader) + sizeof(struct UDPMessage))
2303   {
2304     GNUNET_break_op(0);
2305     return;
2306   }
2307
2308   /* convert address */
2309   switch (sender_addr->sa_family)
2310   {
2311   case AF_INET:
2312     GNUNET_assert(sender_addr_len == sizeof(struct sockaddr_in));
2313     memset (&u4, 0, sizeof(u4));
2314     u6.options = htonl (0);
2315     u4.ipv4_addr = ((struct sockaddr_in *) sender_addr)->sin_addr.s_addr;
2316     u4.u4_port = ((struct sockaddr_in *) sender_addr)->sin_port;
2317     arg = &u4;
2318     args = sizeof(u4);
2319     break;
2320   case AF_INET6:
2321     GNUNET_assert(sender_addr_len == sizeof(struct sockaddr_in6));
2322     memset (&u6, 0, sizeof(u6));
2323     u6.options = htonl (0);
2324     u6.ipv6_addr = ((struct sockaddr_in6 *) sender_addr)->sin6_addr;
2325     u6.u6_port = ((struct sockaddr_in6 *) sender_addr)->sin6_port;
2326     arg = &u6;
2327     args = sizeof(u6);
2328     break;
2329   default:
2330     GNUNET_break(0);
2331     return;
2332   }
2333   LOG(GNUNET_ERROR_TYPE_DEBUG,
2334       "Received message with %u bytes from peer `%s' at `%s'\n",
2335       (unsigned int ) ntohs (msg->header.size), GNUNET_i2s (&msg->sender),
2336       GNUNET_a2s (sender_addr, sender_addr_len));
2337
2338   address = GNUNET_HELLO_address_allocate ( &msg->sender, PLUGIN_NAME,
2339                                             arg, args,
2340                                             GNUNET_HELLO_ADDRESS_INFO_INBOUND);
2341   if ( (NULL == (s = udp_plugin_lookup_session (plugin, address))) &&
2342        (GNUNET_YES != s->in_destroy) )
2343   {
2344     s = udp_plugin_create_session (plugin, address);
2345     plugin->env->session_start (NULL, address, s, NULL, 0);
2346     notify_session_monitor (s->plugin,
2347                             s,
2348                             GNUNET_TRANSPORT_SS_INIT);
2349     notify_session_monitor (s->plugin,
2350                             s,
2351                             GNUNET_TRANSPORT_SS_UP);
2352   }
2353   GNUNET_free (address);
2354
2355   /* iterate over all embedded messages */
2356   si.session = s;
2357   si.sender = msg->sender;
2358   si.arg = arg;
2359   si.args = args;
2360   s->rc++;
2361   GNUNET_SERVER_mst_receive (plugin->mst,
2362                              &si,
2363                              (const char *) &msg[1],
2364                              ntohs (msg->header.size) - sizeof(struct UDPMessage),
2365                              GNUNET_YES,
2366                              GNUNET_NO);
2367   s->rc--;
2368   if ((0 == s->rc) && (GNUNET_YES == s->in_destroy))
2369     free_session (s);
2370 }
2371
2372
2373 /**
2374  * Process a defragmented message.
2375  *
2376  * @param cls the `struct DefragContext *`
2377  * @param msg the message
2378  */
2379 static void
2380 fragment_msg_proc (void *cls,
2381                    const struct GNUNET_MessageHeader *msg)
2382 {
2383   struct DefragContext *rc = cls;
2384
2385   if (ntohs (msg->type) != GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_MESSAGE)
2386   {
2387     GNUNET_break(0);
2388     return;
2389   }
2390   if (ntohs (msg->size) < sizeof(struct UDPMessage))
2391   {
2392     GNUNET_break(0);
2393     return;
2394   }
2395   process_udp_message (rc->plugin,
2396                        (const struct UDPMessage *) msg,
2397                        rc->src_addr,
2398                        rc->addr_len);
2399 }
2400
2401
2402 /**
2403  * Transmit an acknowledgement.
2404  *
2405  * @param cls the `struct DefragContext *`
2406  * @param id message ID (unused)
2407  * @param msg ack to transmit
2408  */
2409 static void
2410 ack_proc (void *cls,
2411           uint32_t id,
2412           const struct GNUNET_MessageHeader *msg)
2413 {
2414   struct DefragContext *rc = cls;
2415   size_t msize = sizeof(struct UDP_ACK_Message) + ntohs (msg->size);
2416   struct UDP_ACK_Message *udp_ack;
2417   uint32_t delay = 0;
2418   struct UDP_MessageWrapper *udpw;
2419   struct Session *s;
2420   struct LookupContext l_ctx;
2421
2422   l_ctx.address = rc->src_addr;
2423   l_ctx.addr_len = rc->addr_len;
2424   l_ctx.must_have_frag_ctx = GNUNET_NO;
2425   l_ctx.res = NULL;
2426   GNUNET_CONTAINER_multipeermap_iterate (rc->plugin->sessions,
2427                                          &lookup_session_by_sockaddr_it,
2428                                          &l_ctx);
2429   s = l_ctx.res;
2430   if (NULL == s)
2431   {
2432     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
2433         "Trying to transmit ACK to peer `%s' but not session found!\n",
2434         GNUNET_a2s(rc->src_addr, rc->addr_len));
2435
2436     GNUNET_CONTAINER_heap_remove_node (rc->hnode);
2437     GNUNET_DEFRAGMENT_context_destroy (rc->defrag);
2438     GNUNET_free (rc);
2439
2440     return;
2441   }
2442   if (s->flow_delay_for_other_peer.rel_value_us <= UINT32_MAX)
2443     delay = s->flow_delay_for_other_peer.rel_value_us;
2444
2445   LOG (GNUNET_ERROR_TYPE_DEBUG,
2446        "Sending ACK to `%s' including delay of %s\n",
2447        GNUNET_a2s (rc->src_addr, (rc->src_addr->sa_family == AF_INET) ? sizeof (struct sockaddr_in) : sizeof (struct sockaddr_in6)),
2448        GNUNET_STRINGS_relative_time_to_string (s->flow_delay_for_other_peer,
2449                                                GNUNET_YES));
2450   udpw = GNUNET_malloc (sizeof (struct UDP_MessageWrapper) + msize);
2451   udpw->msg_size = msize;
2452   udpw->payload_size = 0;
2453   udpw->session = s;
2454   udpw->timeout = GNUNET_TIME_UNIT_FOREVER_ABS;
2455   udpw->msg_buf = (char *) &udpw[1];
2456   udpw->msg_type = UMT_MSG_ACK;
2457   udp_ack = (struct UDP_ACK_Message *) udpw->msg_buf;
2458   udp_ack->header.size = htons ((uint16_t) msize);
2459   udp_ack->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_ACK);
2460   udp_ack->delay = htonl (delay);
2461   udp_ack->sender = *rc->plugin->env->my_identity;
2462   memcpy (&udp_ack[1], msg, ntohs (msg->size));
2463   enqueue (rc->plugin, udpw);
2464   notify_session_monitor (s->plugin,
2465                           s,
2466                           GNUNET_TRANSPORT_SS_UPDATE);
2467   schedule_select (rc->plugin);
2468 }
2469
2470
2471 /**
2472  * FIXME.
2473  */
2474 static void
2475 read_process_msg (struct Plugin *plugin,
2476                   const struct GNUNET_MessageHeader *msg,
2477                   const struct sockaddr *addr,
2478                   socklen_t fromlen)
2479 {
2480   if (ntohs (msg->size) < sizeof(struct UDPMessage))
2481   {
2482     GNUNET_break_op(0);
2483     return;
2484   }
2485   process_udp_message (plugin,
2486                        (const struct UDPMessage *) msg,
2487                        addr,
2488                        fromlen);
2489 }
2490
2491
2492 /**
2493  * FIXME.
2494  */
2495 static void
2496 read_process_ack (struct Plugin *plugin,
2497                   const struct GNUNET_MessageHeader *msg,
2498                   const struct sockaddr *addr,
2499                   socklen_t fromlen)
2500 {
2501   const struct GNUNET_MessageHeader *ack;
2502   const struct UDP_ACK_Message *udp_ack;
2503   struct LookupContext l_ctx;
2504   struct Session *s;
2505   struct GNUNET_TIME_Relative flow_delay;
2506
2507   if (ntohs (msg->size)
2508       < sizeof(struct UDP_ACK_Message) + sizeof(struct GNUNET_MessageHeader))
2509   {
2510     GNUNET_break_op(0);
2511     return;
2512   }
2513   udp_ack = (const struct UDP_ACK_Message *) msg;
2514
2515   /* Lookup session based on sockaddr */
2516   l_ctx.address = addr;
2517   l_ctx.addr_len = fromlen;
2518   l_ctx.res = NULL;
2519   l_ctx.must_have_frag_ctx = GNUNET_YES;
2520   GNUNET_CONTAINER_multipeermap_iterate (plugin->sessions,
2521       &lookup_session_by_sockaddr_it, &l_ctx);
2522   s = l_ctx.res;
2523   if ((NULL == s) || (NULL == s->frag_ctx))
2524   {
2525     return;
2526   }
2527
2528   flow_delay.rel_value_us = (uint64_t) ntohl (udp_ack->delay);
2529   LOG(GNUNET_ERROR_TYPE_DEBUG, "We received a sending delay of %s\n",
2530       GNUNET_STRINGS_relative_time_to_string (flow_delay, GNUNET_YES));
2531   s->flow_delay_from_other_peer = GNUNET_TIME_relative_to_absolute (flow_delay);
2532
2533   ack = (const struct GNUNET_MessageHeader *) &udp_ack[1];
2534   if (ntohs (ack->size) != ntohs (msg->size) - sizeof(struct UDP_ACK_Message))
2535   {
2536     GNUNET_break_op(0);
2537     return;
2538   }
2539
2540   if (0
2541       != memcmp (&l_ctx.res->target, &udp_ack->sender,
2542           sizeof(struct GNUNET_PeerIdentity)))
2543     GNUNET_break(0);
2544   if (GNUNET_OK != GNUNET_FRAGMENT_process_ack (s->frag_ctx->frag, ack))
2545   {
2546     LOG(GNUNET_ERROR_TYPE_DEBUG,
2547         "UDP processes %u-byte acknowledgement from `%s' at `%s'\n",
2548         (unsigned int ) ntohs (msg->size), GNUNET_i2s (&udp_ack->sender),
2549         GNUNET_a2s (addr, fromlen));
2550     /* Expect more ACKs to arrive */
2551     return;
2552   }
2553
2554   LOG(GNUNET_ERROR_TYPE_DEBUG, "Message full ACK'ed\n",
2555       (unsigned int ) ntohs (msg->size), GNUNET_i2s (&udp_ack->sender),
2556       GNUNET_a2s (addr, fromlen));
2557
2558   /* Remove fragmented message after successful sending */
2559   fragmented_message_done (s->frag_ctx, GNUNET_OK);
2560 }
2561
2562
2563 /**
2564  * FIXME.
2565  */
2566 static void
2567 read_process_fragment (struct Plugin *plugin,
2568                        const struct GNUNET_MessageHeader *msg,
2569                        const struct sockaddr *addr,
2570                        socklen_t fromlen)
2571 {
2572   struct DefragContext *d_ctx;
2573   struct GNUNET_TIME_Absolute now;
2574   struct FindReceiveContext frc;
2575
2576   frc.rc = NULL;
2577   frc.addr = addr;
2578   frc.addr_len = fromlen;
2579
2580   LOG(GNUNET_ERROR_TYPE_DEBUG, "UDP processes %u-byte fragment from `%s'\n",
2581       (unsigned int ) ntohs (msg->size), GNUNET_a2s (addr, fromlen));
2582   /* Lookup existing receive context for this address */
2583   GNUNET_CONTAINER_heap_iterate (plugin->defrag_ctxs,
2584       &find_receive_context, &frc);
2585   now = GNUNET_TIME_absolute_get ();
2586   d_ctx = frc.rc;
2587
2588   if (d_ctx == NULL )
2589   {
2590     /* Create a new defragmentation context */
2591     d_ctx = GNUNET_malloc (sizeof (struct DefragContext) + fromlen);
2592     memcpy (&d_ctx[1], addr, fromlen);
2593     d_ctx->src_addr = (const struct sockaddr *) &d_ctx[1];
2594     d_ctx->addr_len = fromlen;
2595     d_ctx->plugin = plugin;
2596     d_ctx->defrag = GNUNET_DEFRAGMENT_context_create (plugin->env->stats,
2597         UDP_MTU, UDP_MAX_MESSAGES_IN_DEFRAG, d_ctx, &fragment_msg_proc,
2598         &ack_proc);
2599     d_ctx->hnode = GNUNET_CONTAINER_heap_insert (plugin->defrag_ctxs, d_ctx,
2600         (GNUNET_CONTAINER_HeapCostType) now.abs_value_us);
2601     LOG(GNUNET_ERROR_TYPE_DEBUG,
2602         "Created new defragmentation context for %u-byte fragment from `%s'\n",
2603         (unsigned int ) ntohs (msg->size), GNUNET_a2s (addr, fromlen));
2604   }
2605   else
2606   {
2607     LOG(GNUNET_ERROR_TYPE_DEBUG,
2608         "Found existing defragmentation context for %u-byte fragment from `%s'\n",
2609         (unsigned int ) ntohs (msg->size), GNUNET_a2s (addr, fromlen));
2610   }
2611
2612   if (GNUNET_OK == GNUNET_DEFRAGMENT_process_fragment (d_ctx->defrag, msg))
2613   {
2614     /* keep this 'rc' from expiring */
2615     GNUNET_CONTAINER_heap_update_cost (plugin->defrag_ctxs, d_ctx->hnode,
2616         (GNUNET_CONTAINER_HeapCostType) now.abs_value_us);
2617   }
2618   if (GNUNET_CONTAINER_heap_get_size (plugin->defrag_ctxs) >
2619   UDP_MAX_SENDER_ADDRESSES_WITH_DEFRAG)
2620   {
2621     /* remove 'rc' that was inactive the longest */
2622     d_ctx = GNUNET_CONTAINER_heap_remove_root (plugin->defrag_ctxs);
2623     GNUNET_assert(NULL != d_ctx);
2624     GNUNET_DEFRAGMENT_context_destroy (d_ctx->defrag);
2625     GNUNET_free(d_ctx);
2626   }
2627 }
2628
2629
2630 /**
2631  * Read and process a message from the given socket.
2632  *
2633  * @param plugin the overall plugin
2634  * @param rsock socket to read from
2635  */
2636 static void
2637 udp_select_read (struct Plugin *plugin,
2638                  struct GNUNET_NETWORK_Handle *rsock)
2639 {
2640   socklen_t fromlen;
2641   struct sockaddr_storage addr;
2642   char buf[65536] GNUNET_ALIGN;
2643   ssize_t size;
2644   const struct GNUNET_MessageHeader *msg;
2645
2646   fromlen = sizeof(addr);
2647   memset (&addr, 0, sizeof(addr));
2648   size = GNUNET_NETWORK_socket_recvfrom (rsock, buf, sizeof(buf),
2649                                          (struct sockaddr *) &addr, &fromlen);
2650 #if MINGW
2651   /* On SOCK_DGRAM UDP sockets recvfrom might fail with a
2652    * WSAECONNRESET error to indicate that previous sendto() (yes, sendto!)
2653    * on this socket has failed.
2654    * Quote from MSDN:
2655    *   WSAECONNRESET - The virtual circuit was reset by the remote side
2656    *   executing a hard or abortive close. The application should close
2657    *   the socket; it is no longer usable. On a UDP-datagram socket this
2658    *   error indicates a previous send operation resulted in an ICMP Port
2659    *   Unreachable message.
2660    */
2661   if ( (-1 == size) && (ECONNRESET == errno) )
2662   return;
2663 #endif
2664   if (-1 == size)
2665   {
2666     LOG (GNUNET_ERROR_TYPE_DEBUG,
2667          "UDP failed to receive data: %s\n",
2668          STRERROR (errno));
2669     /* Connection failure or something. Not a protocol violation. */
2670     return;
2671   }
2672   if (size < sizeof(struct GNUNET_MessageHeader))
2673   {
2674     LOG (GNUNET_ERROR_TYPE_WARNING,
2675          "UDP got %u bytes from %s, which is not enough for a GNUnet message header\n",
2676          (unsigned int ) size,
2677          GNUNET_a2s ((const struct sockaddr *) &addr, fromlen));
2678     /* _MAY_ be a connection failure (got partial message) */
2679     /* But it _MAY_ also be that the other side uses non-GNUnet protocol. */
2680     GNUNET_break_op(0);
2681     return;
2682   }
2683   msg = (const struct GNUNET_MessageHeader *) buf;
2684
2685   LOG (GNUNET_ERROR_TYPE_DEBUG,
2686        "UDP received %u-byte message from `%s' type %u\n",
2687        (unsigned int) size,
2688        GNUNET_a2s ((const struct sockaddr *) &addr, fromlen),
2689        ntohs (msg->type));
2690
2691   if (size != ntohs (msg->size))
2692   {
2693     LOG (GNUNET_ERROR_TYPE_WARNING,
2694          "UDP malformed message header from %s\n",
2695          (unsigned int ) size,
2696          GNUNET_a2s ((const struct sockaddr *) &addr, fromlen));
2697     GNUNET_break_op(0);
2698     return;
2699   }
2700   GNUNET_STATISTICS_update (plugin->env->stats,
2701                             "# UDP, total, bytes, received",
2702                             size,
2703                             GNUNET_NO);
2704
2705   switch (ntohs (msg->type))
2706   {
2707   case GNUNET_MESSAGE_TYPE_TRANSPORT_BROADCAST_BEACON:
2708     if (GNUNET_YES == plugin->enable_broadcasting_receiving)
2709       udp_broadcast_receive (plugin, buf, size, (const struct sockaddr *) &addr,
2710           fromlen);
2711     return;
2712   case GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_MESSAGE:
2713     read_process_msg (plugin, msg, (const struct sockaddr *) &addr, fromlen);
2714     return;
2715   case GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_ACK:
2716     read_process_ack (plugin, msg, (const struct sockaddr *) &addr, fromlen);
2717     return;
2718   case GNUNET_MESSAGE_TYPE_FRAGMENT:
2719     read_process_fragment (plugin, msg, (const struct sockaddr *) &addr,
2720         fromlen);
2721     return;
2722   default:
2723     GNUNET_break_op(0);
2724     return;
2725   }
2726 }
2727
2728
2729 /**
2730  * FIXME.
2731  */
2732 static struct UDP_MessageWrapper *
2733 remove_timeout_messages_and_select (struct UDP_MessageWrapper *head,
2734                                     struct GNUNET_NETWORK_Handle *sock)
2735 {
2736   struct UDP_MessageWrapper *udpw = NULL;
2737   struct GNUNET_TIME_Relative remaining;
2738   struct Session *session;
2739   struct Plugin *plugin;
2740   int removed;
2741
2742   removed = GNUNET_NO;
2743   udpw = head;
2744   while (NULL != udpw)
2745   {
2746     session = udpw->session;
2747     plugin = session->plugin;
2748     /* Find messages with timeout */
2749     remaining = GNUNET_TIME_absolute_get_remaining (udpw->timeout);
2750     if (GNUNET_TIME_UNIT_ZERO.rel_value_us == remaining.rel_value_us)
2751     {
2752       /* Message timed out */
2753       switch (udpw->msg_type)
2754       {
2755       case UMT_MSG_UNFRAGMENTED:
2756         GNUNET_STATISTICS_update (plugin->env->stats,
2757                                   "# UDP, total, bytes, sent, timeout",
2758                                   udpw->msg_size,
2759                                   GNUNET_NO);
2760         GNUNET_STATISTICS_update (plugin->env->stats,
2761                                   "# UDP, total, messages, sent, timeout",
2762                                   1,
2763                                   GNUNET_NO);
2764         GNUNET_STATISTICS_update (plugin->env->stats,
2765                                   "# UDP, unfragmented msgs, messages, sent, timeout",
2766                                   1,
2767                                   GNUNET_NO);
2768         GNUNET_STATISTICS_update (plugin->env->stats,
2769                                   "# UDP, unfragmented msgs, bytes, sent, timeout",
2770                                   udpw->payload_size,
2771                                   GNUNET_NO);
2772         /* Not fragmented message */
2773         LOG (GNUNET_ERROR_TYPE_DEBUG,
2774              "Message for peer `%s' with size %u timed out\n",
2775              GNUNET_i2s (&udpw->session->target),
2776              udpw->payload_size);
2777         call_continuation (udpw, GNUNET_SYSERR);
2778         /* Remove message */
2779         removed = GNUNET_YES;
2780         dequeue (plugin, udpw);
2781         GNUNET_free(udpw);
2782         break;
2783       case UMT_MSG_FRAGMENTED:
2784         /* Fragmented message */
2785         GNUNET_STATISTICS_update (plugin->env->stats,
2786                                   "# UDP, total, bytes, sent, timeout",
2787                                   udpw->frag_ctx->on_wire_size,
2788                                   GNUNET_NO);
2789         GNUNET_STATISTICS_update (plugin->env->stats,
2790                                   "# UDP, total, messages, sent, timeout",
2791                                   1,
2792                                   GNUNET_NO);
2793         call_continuation (udpw, GNUNET_SYSERR);
2794         LOG (GNUNET_ERROR_TYPE_DEBUG,
2795              "Fragment for message for peer `%s' with size %u timed out\n",
2796              GNUNET_i2s (&udpw->session->target),
2797             udpw->frag_ctx->payload_size);
2798
2799         GNUNET_STATISTICS_update (plugin->env->stats,
2800                                   "# UDP, fragmented msgs, messages, sent, timeout",
2801                                   1,
2802                                   GNUNET_NO);
2803         GNUNET_STATISTICS_update (plugin->env->stats,
2804                                   "# UDP, fragmented msgs, bytes, sent, timeout",
2805                                   udpw->frag_ctx->payload_size,
2806                                   GNUNET_NO);
2807         /* Remove fragmented message due to timeout */
2808         fragmented_message_done (udpw->frag_ctx, GNUNET_SYSERR);
2809         break;
2810       case UMT_MSG_ACK:
2811         GNUNET_STATISTICS_update (plugin->env->stats,
2812                                   "# UDP, total, bytes, sent, timeout",
2813                                   udpw->msg_size,
2814                                   GNUNET_NO);
2815         GNUNET_STATISTICS_update (plugin->env->stats,
2816                                   "# UDP, total, messages, sent, timeout",
2817                                   1,
2818                                   GNUNET_NO);
2819         LOG (GNUNET_ERROR_TYPE_DEBUG,
2820              "ACK Message for peer `%s' with size %u timed out\n",
2821              GNUNET_i2s (&udpw->session->target),
2822              udpw->payload_size);
2823         call_continuation (udpw, GNUNET_SYSERR);
2824         removed = GNUNET_YES;
2825         dequeue (plugin, udpw);
2826         GNUNET_free(udpw);
2827         break;
2828       default:
2829         break;
2830       }
2831       if (sock == plugin->sockv4)
2832         udpw = plugin->ipv4_queue_head;
2833       else if (sock == plugin->sockv6)
2834         udpw = plugin->ipv6_queue_head;
2835       else
2836       {
2837         GNUNET_break(0); /* should never happen */
2838         udpw = NULL;
2839       }
2840       GNUNET_STATISTICS_update (plugin->env->stats,
2841                                 "# messages discarded due to timeout",
2842                                 1,
2843                                 GNUNET_NO);
2844     }
2845     else
2846     {
2847       /* Message did not time out, check flow delay */
2848       remaining = GNUNET_TIME_absolute_get_remaining (udpw->session->flow_delay_from_other_peer);
2849       if (GNUNET_TIME_UNIT_ZERO.rel_value_us == remaining.rel_value_us)
2850       {
2851         /* this message is not delayed */
2852         LOG (GNUNET_ERROR_TYPE_DEBUG,
2853              "Message for peer `%s' (%u bytes) is not delayed \n",
2854              GNUNET_i2s (&udpw->session->target),
2855              udpw->payload_size);
2856         break; /* Found message to send, break */
2857       }
2858       else
2859       {
2860         /* Message is delayed, try next */
2861         LOG (GNUNET_ERROR_TYPE_DEBUG,
2862              "Message for peer `%s' (%u bytes) is delayed for %s\n",
2863              GNUNET_i2s (&udpw->session->target), udpw->payload_size,
2864              GNUNET_STRINGS_relative_time_to_string (remaining, GNUNET_YES));
2865         udpw = udpw->next;
2866       }
2867     }
2868   }
2869   if (GNUNET_YES == removed)
2870     notify_session_monitor (session->plugin,
2871                             session,
2872                             GNUNET_TRANSPORT_SS_UPDATE);
2873   return udpw;
2874 }
2875
2876
2877 /**
2878  * FIXME.
2879  */
2880 static void
2881 analyze_send_error (struct Plugin *plugin,
2882                     const struct sockaddr *sa,
2883                     socklen_t slen, int error)
2884 {
2885   enum GNUNET_ATS_Network_Type type;
2886
2887   type = plugin->env->get_address_type (plugin->env->cls, sa, slen);
2888   if (((GNUNET_ATS_NET_LAN == type)
2889        || (GNUNET_ATS_NET_WAN == type))
2890       && ((ENETUNREACH == errno)|| (ENETDOWN == errno)))
2891   {
2892     if (slen == sizeof (struct sockaddr_in))
2893     {
2894       /* IPv4: "Network unreachable" or "Network down"
2895        *
2896        * This indicates we do not have connectivity
2897        */
2898       LOG (GNUNET_ERROR_TYPE_WARNING | GNUNET_ERROR_TYPE_BULK,
2899            _("UDP could not transmit message to `%s': "
2900              "Network seems down, please check your network configuration\n"),
2901            GNUNET_a2s (sa, slen));
2902     }
2903     if (slen == sizeof (struct sockaddr_in6))
2904     {
2905       /* IPv6: "Network unreachable" or "Network down"
2906        *
2907        * This indicates that this system is IPv6 enabled, but does not
2908        * have a valid global IPv6 address assigned or we do not have
2909        * connectivity
2910        */
2911       LOG (GNUNET_ERROR_TYPE_WARNING | GNUNET_ERROR_TYPE_BULK,
2912            _("UDP could not transmit IPv6 message! "
2913              "Please check your network configuration and disable IPv6 if your "
2914              "connection does not have a global IPv6 address\n"));
2915     }
2916   }
2917   else
2918   {
2919     LOG (GNUNET_ERROR_TYPE_WARNING,
2920          "UDP could not transmit message to `%s': `%s'\n",
2921          GNUNET_a2s (sa, slen), STRERROR (error));
2922   }
2923 }
2924
2925
2926 /**
2927  * FIXME.
2928  */
2929 static size_t
2930 udp_select_send (struct Plugin *plugin,
2931                  struct GNUNET_NETWORK_Handle *sock)
2932 {
2933   ssize_t sent;
2934   socklen_t slen;
2935   struct sockaddr *a;
2936   const struct IPv4UdpAddress *u4;
2937   struct sockaddr_in a4;
2938   const struct IPv6UdpAddress *u6;
2939   struct sockaddr_in6 a6;
2940   struct UDP_MessageWrapper *udpw;
2941
2942   /* Find message to send */
2943   udpw = remove_timeout_messages_and_select ((sock == plugin->sockv4)
2944                                              ? plugin->ipv4_queue_head
2945                                              : plugin->ipv6_queue_head,
2946                                              sock);
2947   if (NULL == udpw)
2948     return 0; /* No message to send */
2949
2950   if (sizeof (struct IPv4UdpAddress) == udpw->session->address->address_length)
2951   {
2952     u4 = udpw->session->address->address;
2953     memset (&a4, 0, sizeof(a4));
2954     a4.sin_family = AF_INET;
2955 #if HAVE_SOCKADDR_IN_SIN_LEN
2956     a4.sin_len = sizeof (a4);
2957 #endif
2958     a4.sin_port = u4->u4_port;
2959     memcpy (&a4.sin_addr, &u4->ipv4_addr, sizeof(struct in_addr));
2960     a = (struct sockaddr *) &a4;
2961     slen = sizeof (a4);
2962   }
2963   else if (sizeof (struct IPv6UdpAddress) == udpw->session->address->address_length)
2964   {
2965     u6 = udpw->session->address->address;
2966     memset (&a6, 0, sizeof(a6));
2967     a6.sin6_family = AF_INET6;
2968 #if HAVE_SOCKADDR_IN_SIN_LEN
2969     a6.sin6_len = sizeof (a6);
2970 #endif
2971     a6.sin6_port = u6->u6_port;
2972     memcpy (&a6.sin6_addr, &u6->ipv6_addr, sizeof(struct in6_addr));
2973     a = (struct sockaddr *) &a6;
2974     slen = sizeof (a6);
2975   }
2976   else
2977   {
2978     call_continuation (udpw, GNUNET_OK);
2979     dequeue (plugin, udpw);
2980     notify_session_monitor (plugin,
2981                             udpw->session,
2982                             GNUNET_TRANSPORT_SS_UPDATE);
2983     GNUNET_free (udpw);
2984     return GNUNET_SYSERR;
2985   }
2986
2987   sent = GNUNET_NETWORK_socket_sendto (sock,
2988                                        udpw->msg_buf,
2989                                        udpw->msg_size,
2990                                        a,
2991                                        slen);
2992   if (GNUNET_SYSERR == sent)
2993   {
2994     /* Failure */
2995     analyze_send_error (plugin, a, slen, errno);
2996     call_continuation (udpw, GNUNET_SYSERR);
2997     GNUNET_STATISTICS_update (plugin->env->stats,
2998         "# UDP, total, bytes, sent, failure", sent, GNUNET_NO);
2999     GNUNET_STATISTICS_update (plugin->env->stats,
3000         "# UDP, total, messages, sent, failure", 1, GNUNET_NO);
3001   }
3002   else
3003   {
3004     /* Success */
3005     LOG(GNUNET_ERROR_TYPE_DEBUG,
3006         "UDP transmitted %u-byte message to  `%s' `%s' (%d: %s)\n",
3007         (unsigned int ) (udpw->msg_size), GNUNET_i2s (&udpw->session->target),
3008         GNUNET_a2s (a, slen), (int ) sent,
3009         (sent < 0) ? STRERROR (errno) : "ok");
3010     GNUNET_STATISTICS_update (plugin->env->stats,
3011         "# UDP, total, bytes, sent, success", sent, GNUNET_NO);
3012     GNUNET_STATISTICS_update (plugin->env->stats,
3013         "# UDP, total, messages, sent, success", 1, GNUNET_NO);
3014     if (NULL != udpw->frag_ctx)
3015       udpw->frag_ctx->on_wire_size += udpw->msg_size;
3016     call_continuation (udpw, GNUNET_OK);
3017   }
3018   dequeue (plugin, udpw);
3019   notify_session_monitor (plugin,
3020                           udpw->session,
3021                           GNUNET_TRANSPORT_SS_UPDATE);
3022   GNUNET_free(udpw);
3023   return sent;
3024 }
3025
3026
3027 /**
3028  * We have been notified that our readset has something to read.  We don't
3029  * know which socket needs to be read, so we have to check each one
3030  * Then reschedule this function to be called again once more is available.
3031  *
3032  * @param cls the plugin handle
3033  * @param tc the scheduling context (for rescheduling this function again)
3034  */
3035 static void
3036 udp_plugin_select (void *cls,
3037                    const struct GNUNET_SCHEDULER_TaskContext *tc)
3038 {
3039   struct Plugin *plugin = cls;
3040
3041   plugin->select_task = NULL;
3042   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
3043     return;
3044   if ((0 != (tc->reason & GNUNET_SCHEDULER_REASON_READ_READY))
3045       && (NULL != plugin->sockv4)
3046       && (GNUNET_NETWORK_fdset_isset (tc->read_ready, plugin->sockv4)))
3047     udp_select_read (plugin, plugin->sockv4);
3048   if ((0 != (tc->reason & GNUNET_SCHEDULER_REASON_WRITE_READY))
3049       && (NULL != plugin->sockv4) && (NULL != plugin->ipv4_queue_head)
3050       && (GNUNET_NETWORK_fdset_isset (tc->write_ready, plugin->sockv4)))
3051     udp_select_send (plugin, plugin->sockv4);
3052   schedule_select (plugin);
3053 }
3054
3055
3056 /**
3057  * We have been notified that our readset has something to read.  We don't
3058  * know which socket needs to be read, so we have to check each one
3059  * Then reschedule this function to be called again once more is available.
3060  *
3061  * @param cls the plugin handle
3062  * @param tc the scheduling context (for rescheduling this function again)
3063  */
3064 static void
3065 udp_plugin_select_v6 (void *cls,
3066                       const struct GNUNET_SCHEDULER_TaskContext *tc)
3067 {
3068   struct Plugin *plugin = cls;
3069
3070   plugin->select_task_v6 = NULL;
3071   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
3072     return;
3073   if (((tc->reason & GNUNET_SCHEDULER_REASON_READ_READY) != 0)
3074       && (NULL != plugin->sockv6)
3075       && (GNUNET_NETWORK_fdset_isset (tc->read_ready, plugin->sockv6)))
3076     udp_select_read (plugin, plugin->sockv6);
3077   if ((0 != (tc->reason & GNUNET_SCHEDULER_REASON_WRITE_READY))
3078       && (NULL != plugin->sockv6) && (plugin->ipv6_queue_head != NULL )&&
3079       (GNUNET_NETWORK_fdset_isset (tc->write_ready, plugin->sockv6)) )udp_select_send (plugin, plugin->sockv6);
3080   schedule_select (plugin);
3081 }
3082
3083
3084 /**
3085  * Setup the UDP sockets (for IPv4 and IPv6) for the plugin.
3086  *
3087  * @param plugin the plugin to initialize
3088  * @param bind_v6 IPv6 address to bind to (can be NULL, for 'any')
3089  * @param bind_v4 IPv4 address to bind to (can be NULL, for 'any')
3090  * @return number of sockets that were successfully bound
3091  */
3092 static int
3093 setup_sockets (struct Plugin *plugin,
3094                const struct sockaddr_in6 *bind_v6,
3095                const struct sockaddr_in *bind_v4)
3096 {
3097   int tries;
3098   int sockets_created = 0;
3099   struct sockaddr_in6 server_addrv6;
3100   struct sockaddr_in server_addrv4;
3101   struct sockaddr *server_addr;
3102   struct sockaddr *addrs[2];
3103   socklen_t addrlens[2];
3104   socklen_t addrlen;
3105   int eno;
3106
3107   /* Create IPv6 socket */
3108   eno = EINVAL;
3109   if (GNUNET_YES == plugin->enable_ipv6)
3110   {
3111     plugin->sockv6 = GNUNET_NETWORK_socket_create (PF_INET6, SOCK_DGRAM, 0);
3112     if (NULL == plugin->sockv6)
3113     {
3114       LOG(GNUNET_ERROR_TYPE_WARNING,
3115           "Disabling IPv6 since it is not supported on this system!\n");
3116       plugin->enable_ipv6 = GNUNET_NO;
3117     }
3118     else
3119     {
3120       memset (&server_addrv6, '\0', sizeof(struct sockaddr_in6));
3121 #if HAVE_SOCKADDR_IN_SIN_LEN
3122       server_addrv6.sin6_len = sizeof (struct sockaddr_in6);
3123 #endif
3124       server_addrv6.sin6_family = AF_INET6;
3125       if (NULL != bind_v6)
3126         server_addrv6.sin6_addr = bind_v6->sin6_addr;
3127       else
3128         server_addrv6.sin6_addr = in6addr_any;
3129
3130       if (0 == plugin->port) /* autodetect */
3131         server_addrv6.sin6_port = htons (
3132             GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_STRONG, 33537)
3133                 + 32000);
3134       else
3135         server_addrv6.sin6_port = htons (plugin->port);
3136       addrlen = sizeof(struct sockaddr_in6);
3137       server_addr = (struct sockaddr *) &server_addrv6;
3138
3139       tries = 0;
3140       while (tries < 10)
3141       {
3142         LOG(GNUNET_ERROR_TYPE_DEBUG,
3143             "Binding to IPv6 `%s'\n",
3144             GNUNET_a2s (server_addr, addrlen));
3145         /* binding */
3146         if (GNUNET_OK
3147             == GNUNET_NETWORK_socket_bind (plugin->sockv6, server_addr,
3148                 addrlen))
3149           break;
3150         eno = errno;
3151         if (0 != plugin->port)
3152         {
3153           tries = 10; /* fail */
3154           break; /* bind failed on specific port */
3155         }
3156         /* autodetect */
3157         server_addrv6.sin6_port = htons (
3158             GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_STRONG, 33537)
3159                 + 32000);
3160         tries++;
3161       }
3162       if (tries >= 10)
3163       {
3164         GNUNET_NETWORK_socket_close (plugin->sockv6);
3165         plugin->enable_ipv6 = GNUNET_NO;
3166         plugin->sockv6 = NULL;
3167       }
3168
3169       if (plugin->sockv6 != NULL )
3170       {
3171         LOG (GNUNET_ERROR_TYPE_DEBUG,
3172              "IPv6 socket created on port %s\n",
3173              GNUNET_a2s (server_addr, addrlen));
3174         addrs[sockets_created] = (struct sockaddr *) &server_addrv6;
3175         addrlens[sockets_created] = sizeof(struct sockaddr_in6);
3176         sockets_created++;
3177       }
3178       else
3179       {
3180         LOG (GNUNET_ERROR_TYPE_ERROR,
3181              "Failed to bind UDP socket to %s: %s\n",
3182              GNUNET_a2s (server_addr, addrlen),
3183              STRERROR (eno));
3184       }
3185     }
3186   }
3187
3188   /* Create IPv4 socket */
3189   eno = EINVAL;
3190   plugin->sockv4 = GNUNET_NETWORK_socket_create (PF_INET, SOCK_DGRAM, 0);
3191   if (NULL == plugin->sockv4)
3192   {
3193     GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING,
3194                          "socket");
3195     LOG(GNUNET_ERROR_TYPE_WARNING,
3196         "Disabling IPv4 since it is not supported on this system!\n");
3197     plugin->enable_ipv4 = GNUNET_NO;
3198   }
3199   else
3200   {
3201     memset (&server_addrv4, '\0', sizeof(struct sockaddr_in));
3202 #if HAVE_SOCKADDR_IN_SIN_LEN
3203     server_addrv4.sin_len = sizeof (struct sockaddr_in);
3204 #endif
3205     server_addrv4.sin_family = AF_INET;
3206     if (NULL != bind_v4)
3207       server_addrv4.sin_addr = bind_v4->sin_addr;
3208     else
3209       server_addrv4.sin_addr.s_addr = INADDR_ANY;
3210
3211     if (0 == plugin->port)
3212       /* autodetect */
3213       server_addrv4.sin_port = htons (GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_STRONG,
3214                                                                 33537)
3215                                       + 32000);
3216     else
3217       server_addrv4.sin_port = htons (plugin->port);
3218
3219     addrlen = sizeof(struct sockaddr_in);
3220     server_addr = (struct sockaddr *) &server_addrv4;
3221
3222     tries = 0;
3223     while (tries < 10)
3224     {
3225       LOG (GNUNET_ERROR_TYPE_DEBUG,
3226            "Binding to IPv4 `%s'\n",
3227            GNUNET_a2s (server_addr, addrlen));
3228
3229       /* binding */
3230       if (GNUNET_OK
3231           == GNUNET_NETWORK_socket_bind (plugin->sockv4, server_addr, addrlen))
3232         break;
3233       eno = errno;
3234       if (0 != plugin->port)
3235       {
3236         tries = 10; /* fail */
3237         break; /* bind failed on specific port */
3238       }
3239
3240       /* autodetect */
3241       server_addrv4.sin_port = htons (
3242           GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_STRONG, 33537)
3243               + 32000);
3244       tries++;
3245     }
3246
3247     if (tries >= 10)
3248     {
3249       GNUNET_NETWORK_socket_close (plugin->sockv4);
3250       plugin->enable_ipv4 = GNUNET_NO;
3251       plugin->sockv4 = NULL;
3252     }
3253
3254     if (NULL != plugin->sockv4)
3255     {
3256       LOG(GNUNET_ERROR_TYPE_DEBUG, "IPv4 socket created on port %s\n",
3257           GNUNET_a2s (server_addr, addrlen));
3258       addrs[sockets_created] = (struct sockaddr *) &server_addrv4;
3259       addrlens[sockets_created] = sizeof(struct sockaddr_in);
3260       sockets_created++;
3261     }
3262     else
3263     {
3264       LOG (GNUNET_ERROR_TYPE_ERROR,
3265            _("Failed to bind UDP socket to %s: %s\n"),
3266            GNUNET_a2s (server_addr, addrlen),
3267            STRERROR (eno));
3268     }
3269   }
3270
3271   if (0 == sockets_created)
3272   {
3273     LOG(GNUNET_ERROR_TYPE_WARNING, _("Failed to open UDP sockets\n"));
3274     return 0; /* No sockets created, return */
3275   }
3276
3277   /* Create file descriptors */
3278   if (plugin->enable_ipv4 == GNUNET_YES)
3279   {
3280     plugin->rs_v4 = GNUNET_NETWORK_fdset_create ();
3281     plugin->ws_v4 = GNUNET_NETWORK_fdset_create ();
3282     GNUNET_NETWORK_fdset_zero (plugin->rs_v4);
3283     GNUNET_NETWORK_fdset_zero (plugin->ws_v4);
3284     if (NULL != plugin->sockv4)
3285     {
3286       GNUNET_NETWORK_fdset_set (plugin->rs_v4, plugin->sockv4);
3287       GNUNET_NETWORK_fdset_set (plugin->ws_v4, plugin->sockv4);
3288     }
3289   }
3290
3291   if (plugin->enable_ipv6 == GNUNET_YES)
3292   {
3293     plugin->rs_v6 = GNUNET_NETWORK_fdset_create ();
3294     plugin->ws_v6 = GNUNET_NETWORK_fdset_create ();
3295     GNUNET_NETWORK_fdset_zero (plugin->rs_v6);
3296     GNUNET_NETWORK_fdset_zero (plugin->ws_v6);
3297     if (NULL != plugin->sockv6)
3298     {
3299       GNUNET_NETWORK_fdset_set (plugin->rs_v6, plugin->sockv6);
3300       GNUNET_NETWORK_fdset_set (plugin->ws_v6, plugin->sockv6);
3301     }
3302   }
3303
3304   schedule_select (plugin);
3305   plugin->nat = GNUNET_NAT_register (plugin->env->cfg,
3306                                      GNUNET_NO,
3307                                      plugin->port,
3308                                      sockets_created,
3309                                      (const struct sockaddr **) addrs,
3310                                      addrlens,
3311                                      &udp_nat_port_map_callback,
3312                                      NULL,
3313                                      plugin);
3314
3315   return sockets_created;
3316 }
3317
3318
3319 /**
3320  * Return information about the given session to the
3321  * monitor callback.
3322  *
3323  * @param cls the `struct Plugin` with the monitor callback (`sic`)
3324  * @param peer peer we send information about
3325  * @param value our `struct Session` to send information about
3326  * @return #GNUNET_OK (continue to iterate)
3327  */
3328 static int
3329 send_session_info_iter (void *cls,
3330                         const struct GNUNET_PeerIdentity *peer,
3331                         void *value)
3332 {
3333   struct Plugin *plugin = cls;
3334   struct Session *session = value;
3335
3336   notify_session_monitor (plugin,
3337                           session,
3338                           GNUNET_TRANSPORT_SS_INIT);
3339   notify_session_monitor (plugin,
3340                           session,
3341                           GNUNET_TRANSPORT_SS_UP);
3342   return GNUNET_OK;
3343 }
3344
3345
3346 /**
3347  * Begin monitoring sessions of a plugin.  There can only
3348  * be one active monitor per plugin (i.e. if there are
3349  * multiple monitors, the transport service needs to
3350  * multiplex the generated events over all of them).
3351  *
3352  * @param cls closure of the plugin
3353  * @param sic callback to invoke, NULL to disable monitor;
3354  *            plugin will being by iterating over all active
3355  *            sessions immediately and then enter monitor mode
3356  * @param sic_cls closure for @a sic
3357  */
3358 static void
3359 udp_plugin_setup_monitor (void *cls,
3360                           GNUNET_TRANSPORT_SessionInfoCallback sic,
3361                           void *sic_cls)
3362 {
3363   struct Plugin *plugin = cls;
3364
3365   plugin->sic = sic;
3366   plugin->sic_cls = sic_cls;
3367   if (NULL != sic)
3368   {
3369     GNUNET_CONTAINER_multipeermap_iterate (plugin->sessions,
3370                                            &send_session_info_iter,
3371                                            plugin);
3372     /* signal end of first iteration */
3373     sic (sic_cls, NULL, NULL);
3374   }
3375 }
3376
3377
3378 /**
3379  * The exported method. Makes the core api available via a global and
3380  * returns the udp transport API.
3381  *
3382  * @param cls our `struct GNUNET_TRANSPORT_PluginEnvironment`
3383  * @return our `struct GNUNET_TRANSPORT_PluginFunctions`
3384  */
3385 void *
3386 libgnunet_plugin_transport_udp_init (void *cls)
3387 {
3388   struct GNUNET_TRANSPORT_PluginEnvironment *env = cls;
3389   struct GNUNET_TRANSPORT_PluginFunctions *api;
3390   struct Plugin *p;
3391   unsigned long long port;
3392   unsigned long long aport;
3393   unsigned long long udp_max_bps;
3394   unsigned long long enable_v6;
3395   unsigned long long enable_broadcasting;
3396   unsigned long long enable_broadcasting_recv;
3397   char *bind4_address;
3398   char *bind6_address;
3399   char *fancy_interval;
3400   struct GNUNET_TIME_Relative interval;
3401   struct sockaddr_in server_addrv4;
3402   struct sockaddr_in6 server_addrv6;
3403   int res;
3404   int have_bind4;
3405   int have_bind6;
3406
3407   if (NULL == env->receive)
3408   {
3409     /* run in 'stub' mode (i.e. as part of gnunet-peerinfo), don't fully
3410      initialze the plugin or the API */
3411     api = GNUNET_new (struct GNUNET_TRANSPORT_PluginFunctions);
3412     api->cls = NULL;
3413     api->address_pretty_printer = &udp_plugin_address_pretty_printer;
3414     api->address_to_string = &udp_address_to_string;
3415     api->string_to_address = &udp_string_to_address;
3416     return api;
3417   }
3418
3419   /* Get port number: port == 0 : autodetect a port,
3420    * > 0 : use this port, not given : 2086 default */
3421   if (GNUNET_OK !=
3422       GNUNET_CONFIGURATION_get_value_number (env->cfg, "transport-udp",
3423                                              "PORT", &port))
3424     port = 2086;
3425   if (GNUNET_OK !=
3426       GNUNET_CONFIGURATION_get_value_number (env->cfg, "transport-udp",
3427                                              "ADVERTISED_PORT", &aport))
3428     aport = port;
3429   if (port > 65535)
3430   {
3431     LOG (GNUNET_ERROR_TYPE_WARNING,
3432          _("Given `%s' option is out of range: %llu > %u\n"),
3433          "PORT", port,
3434          65535);
3435     return NULL;
3436   }
3437
3438   /* Protocols */
3439   if (GNUNET_YES ==
3440       GNUNET_CONFIGURATION_get_value_yesno (env->cfg, "nat", "DISABLEV6"))
3441     enable_v6 = GNUNET_NO;
3442   else
3443     enable_v6 = GNUNET_YES;
3444
3445   /* Addresses */
3446   have_bind4 = GNUNET_NO;
3447   memset (&server_addrv4, 0, sizeof(server_addrv4));
3448   if (GNUNET_YES ==
3449       GNUNET_CONFIGURATION_get_value_string (env->cfg, "transport-udp",
3450                                              "BINDTO", &bind4_address))
3451   {
3452     LOG (GNUNET_ERROR_TYPE_DEBUG,
3453          "Binding udp plugin to specific address: `%s'\n",
3454          bind4_address);
3455     if (1 != inet_pton (AF_INET,
3456                         bind4_address,
3457                         &server_addrv4.sin_addr))
3458     {
3459       GNUNET_free (bind4_address);
3460       return NULL;
3461     }
3462     have_bind4 = GNUNET_YES;
3463   }
3464   GNUNET_free_non_null(bind4_address);
3465   have_bind6 = GNUNET_NO;
3466   memset (&server_addrv6, 0, sizeof(server_addrv6));
3467   if (GNUNET_YES ==
3468       GNUNET_CONFIGURATION_get_value_string (env->cfg, "transport-udp",
3469                                              "BINDTO6", &bind6_address))
3470   {
3471     LOG (GNUNET_ERROR_TYPE_DEBUG,
3472          "Binding udp plugin to specific address: `%s'\n",
3473          bind6_address);
3474     if (1 != inet_pton (AF_INET6,
3475                         bind6_address,
3476                         &server_addrv6.sin6_addr))
3477     {
3478       LOG (GNUNET_ERROR_TYPE_ERROR,
3479            _("Invalid IPv6 address: `%s'\n"),
3480            bind6_address);
3481       GNUNET_free (bind6_address);
3482       return NULL;
3483     }
3484     have_bind6 = GNUNET_YES;
3485   }
3486   GNUNET_free_non_null (bind6_address);
3487
3488   /* Enable neighbour discovery */
3489   enable_broadcasting = GNUNET_CONFIGURATION_get_value_yesno (env->cfg,
3490       "transport-udp", "BROADCAST");
3491   if (enable_broadcasting == GNUNET_SYSERR)
3492     enable_broadcasting = GNUNET_NO;
3493
3494   enable_broadcasting_recv = GNUNET_CONFIGURATION_get_value_yesno (env->cfg,
3495       "transport-udp", "BROADCAST_RECEIVE");
3496   if (enable_broadcasting_recv == GNUNET_SYSERR)
3497     enable_broadcasting_recv = GNUNET_YES;
3498
3499   if (GNUNET_SYSERR ==
3500       GNUNET_CONFIGURATION_get_value_string (env->cfg, "transport-udp",
3501                                              "BROADCAST_INTERVAL",
3502                                              &fancy_interval))
3503   {
3504     interval = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 10);
3505   }
3506   else
3507   {
3508     if (GNUNET_SYSERR ==
3509         GNUNET_STRINGS_fancy_time_to_relative (fancy_interval, &interval))
3510     {
3511       interval = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 30);
3512     }
3513     GNUNET_free(fancy_interval);
3514   }
3515
3516   /* Maximum datarate */
3517   if (GNUNET_OK !=
3518       GNUNET_CONFIGURATION_get_value_number (env->cfg, "transport-udp",
3519                                              "MAX_BPS", &udp_max_bps))
3520   {
3521     udp_max_bps = 1024 * 1024 * 50; /* 50 MB/s == infinity for practical purposes */
3522   }
3523
3524   p = GNUNET_new (struct Plugin);
3525   p->port = port;
3526   p->aport = aport;
3527   p->broadcast_interval = interval;
3528   p->enable_ipv6 = enable_v6;
3529   p->enable_ipv4 = GNUNET_YES; /* default */
3530   p->enable_broadcasting = enable_broadcasting;
3531   p->enable_broadcasting_receiving = enable_broadcasting_recv;
3532   p->env = env;
3533   p->sessions = GNUNET_CONTAINER_multipeermap_create (10, GNUNET_NO);
3534   p->defrag_ctxs = GNUNET_CONTAINER_heap_create (
3535       GNUNET_CONTAINER_HEAP_ORDER_MIN);
3536   p->mst = GNUNET_SERVER_mst_create (&process_inbound_tokenized_messages, p);
3537   GNUNET_BANDWIDTH_tracker_init (&p->tracker, NULL, NULL,
3538       GNUNET_BANDWIDTH_value_init ((uint32_t) udp_max_bps), 30);
3539   LOG(GNUNET_ERROR_TYPE_DEBUG,
3540       "Setting up sockets\n");
3541   res = setup_sockets (p,
3542                        (GNUNET_YES == have_bind6) ? &server_addrv6 : NULL,
3543                        (GNUNET_YES == have_bind4) ? &server_addrv4 : NULL);
3544   if ((res == 0) || ((p->sockv4 == NULL )&& (p->sockv6 == NULL)))
3545   {
3546     LOG (GNUNET_ERROR_TYPE_ERROR,
3547         _("Failed to create network sockets, plugin failed\n"));
3548     GNUNET_CONTAINER_multipeermap_destroy (p->sessions);
3549     GNUNET_CONTAINER_heap_destroy (p->defrag_ctxs);
3550     GNUNET_SERVER_mst_destroy (p->mst);
3551     GNUNET_free (p);
3552     return NULL;
3553   }
3554
3555   /* Setup broadcasting and receiving beacons */
3556   setup_broadcast (p, &server_addrv6, &server_addrv4);
3557
3558   api = GNUNET_new (struct GNUNET_TRANSPORT_PluginFunctions);
3559   api->cls = p;
3560   api->send = NULL;
3561   api->disconnect_session = &udp_disconnect_session;
3562   api->query_keepalive_factor = &udp_query_keepalive_factor;
3563   api->disconnect_peer = &udp_disconnect;
3564   api->address_pretty_printer = &udp_plugin_address_pretty_printer;
3565   api->address_to_string = &udp_address_to_string;
3566   api->string_to_address = &udp_string_to_address;
3567   api->check_address = &udp_plugin_check_address;
3568   api->get_session = &udp_plugin_get_session;
3569   api->send = &udp_plugin_send;
3570   api->get_network = &udp_get_network;
3571   api->update_session_timeout = &udp_plugin_update_session_timeout;
3572   api->setup_monitor = &udp_plugin_setup_monitor;
3573   return api;
3574 }
3575
3576
3577 /**
3578  * Function called on each entry in the defragmentation heap to
3579  * clean it up.
3580  *
3581  * @param cls NULL
3582  * @param node node in the heap (to be removed)
3583  * @param element a `struct DefragContext` to be cleaned up
3584  * @param cost unused
3585  * @return #GNUNET_YES
3586  */
3587 static int
3588 heap_cleanup_iterator (void *cls,
3589                        struct GNUNET_CONTAINER_HeapNode *node,
3590                        void *element,
3591                        GNUNET_CONTAINER_HeapCostType cost)
3592 {
3593   struct DefragContext *d_ctx = element;
3594
3595   GNUNET_CONTAINER_heap_remove_node (node);
3596   GNUNET_DEFRAGMENT_context_destroy (d_ctx->defrag);
3597   GNUNET_free (d_ctx);
3598   return GNUNET_YES;
3599 }
3600
3601
3602 /**
3603  * The exported method. Makes the core api available via a global and
3604  * returns the udp transport API.
3605  *
3606  * @param cls our `struct GNUNET_TRANSPORT_PluginEnvironment`
3607  * @return NULL
3608  */
3609 void *
3610 libgnunet_plugin_transport_udp_done (void *cls)
3611 {
3612   struct GNUNET_TRANSPORT_PluginFunctions *api = cls;
3613   struct Plugin *plugin = api->cls;
3614   struct PrettyPrinterContext *cur;
3615   struct PrettyPrinterContext *next;
3616   struct UDP_MessageWrapper *udpw;
3617
3618   if (NULL == plugin)
3619   {
3620     GNUNET_free(api);
3621     return NULL;
3622   }
3623   stop_broadcast (plugin);
3624   if (plugin->select_task != NULL)
3625   {
3626     GNUNET_SCHEDULER_cancel (plugin->select_task);
3627     plugin->select_task = NULL;
3628   }
3629   if (plugin->select_task_v6 != NULL)
3630   {
3631     GNUNET_SCHEDULER_cancel (plugin->select_task_v6);
3632     plugin->select_task_v6 = NULL;
3633   }
3634
3635   /* Closing sockets */
3636   if (GNUNET_YES == plugin->enable_ipv4)
3637   {
3638     if (NULL != plugin->sockv4)
3639     {
3640       GNUNET_break (GNUNET_OK ==
3641                     GNUNET_NETWORK_socket_close (plugin->sockv4));
3642       plugin->sockv4 = NULL;
3643     }
3644     GNUNET_NETWORK_fdset_destroy (plugin->rs_v4);
3645     GNUNET_NETWORK_fdset_destroy (plugin->ws_v4);
3646   }
3647   if (GNUNET_YES == plugin->enable_ipv6)
3648   {
3649     if (NULL != plugin->sockv6)
3650     {
3651       GNUNET_break (GNUNET_OK ==
3652                     GNUNET_NETWORK_socket_close (plugin->sockv6));
3653       plugin->sockv6 = NULL;
3654
3655       GNUNET_NETWORK_fdset_destroy (plugin->rs_v6);
3656       GNUNET_NETWORK_fdset_destroy (plugin->ws_v6);
3657     }
3658   }
3659   if (NULL != plugin->nat)
3660   {
3661     GNUNET_NAT_unregister (plugin->nat);
3662     plugin->nat = NULL;
3663   }
3664   if (NULL != plugin->defrag_ctxs)
3665   {
3666     GNUNET_CONTAINER_heap_iterate (plugin->defrag_ctxs,
3667                                    &heap_cleanup_iterator, NULL);
3668     GNUNET_CONTAINER_heap_destroy (plugin->defrag_ctxs);
3669     plugin->defrag_ctxs = NULL;
3670   }
3671   if (NULL != plugin->mst)
3672   {
3673     GNUNET_SERVER_mst_destroy (plugin->mst);
3674     plugin->mst = NULL;
3675   }
3676
3677   /* Clean up leftover messages */
3678   udpw = plugin->ipv4_queue_head;
3679   while (NULL != udpw)
3680   {
3681     struct UDP_MessageWrapper *tmp = udpw->next;
3682     dequeue (plugin, udpw);
3683     call_continuation (udpw, GNUNET_SYSERR);
3684     GNUNET_free(udpw);
3685     udpw = tmp;
3686   }
3687   udpw = plugin->ipv6_queue_head;
3688   while (NULL != udpw)
3689   {
3690     struct UDP_MessageWrapper *tmp = udpw->next;
3691     dequeue (plugin, udpw);
3692     call_continuation (udpw, GNUNET_SYSERR);
3693     GNUNET_free(udpw);
3694     udpw = tmp;
3695   }
3696
3697   /* Clean up sessions */
3698   LOG (GNUNET_ERROR_TYPE_DEBUG,
3699        "Cleaning up sessions\n");
3700   GNUNET_CONTAINER_multipeermap_iterate (plugin->sessions,
3701                                          &disconnect_and_free_it, plugin);
3702   GNUNET_CONTAINER_multipeermap_destroy (plugin->sessions);
3703
3704   next = plugin->ppc_dll_head;
3705   for (cur = next; NULL != cur; cur = next)
3706   {
3707     GNUNET_break(0);
3708     next = cur->next;
3709     GNUNET_CONTAINER_DLL_remove (plugin->ppc_dll_head,
3710                                  plugin->ppc_dll_tail,
3711                                  cur);
3712     GNUNET_RESOLVER_request_cancel (cur->resolver_handle);
3713     GNUNET_free (cur);
3714   }
3715   GNUNET_free (plugin);
3716   GNUNET_free (api);
3717   return NULL;
3718 }
3719
3720 /* end of plugin_transport_udp.c */