trying to fix #1813 by actually respecting delay values, also some minor code cleanup
[oweals/gnunet.git] / src / transport / plugin_transport_udp.c
index df8447a12b5eb3cd54635fcb81dd4ee046161c91..ad5b165b184265e4519290fc08e10dca892a3a37 100644 (file)
 #include "gnunet_protocols.h"
 #include "gnunet_resolver_service.h"
 #include "gnunet_signatures.h"
+#include "gnunet_constants.h"
 #include "gnunet_statistics_service.h"
 #include "gnunet_transport_service.h"
 #include "gnunet_transport_plugin.h"
 #include "transport.h"
 
-#define DEBUG_UDP GNUNET_NO
+#define LOG(kind,...) GNUNET_log_from (kind, "transport-udp", __VA_ARGS__)
+
+
+#define DEBUG_UDP GNUNET_EXTRA_LOGGING
 
 /**
  * MTU for fragmentation subsystem.  Should be conservative since
@@ -89,6 +93,28 @@ struct UDPMessage
 };
 
 
+/**
+ * UDP ACK Message-Packet header (after defragmentation).
+ */
+struct UDP_ACK_Message
+{
+  /**
+   * Message header.
+   */
+  struct GNUNET_MessageHeader header;
+
+  /**
+   * Desired delay for flow control
+   */
+  uint32_t delay;
+
+  /**
+   * What is the identity of the sender
+   */
+  struct GNUNET_PeerIdentity sender;
+};
+
+
 /**
  * Network format for IPv4 addresses.
  */
@@ -132,7 +158,7 @@ struct Plugin;
  * Session with another peer.  FIXME: why not make this into
  * a regular 'struct Session' and pass it around!?
  */
-struct PeerSession
+struct Session
 {
 
   /**
@@ -150,6 +176,8 @@ struct PeerSession
    */
   const struct sockaddr *sock_addr;
 
+  size_t addrlen;
+
   /**
    * Function to call upon completion of the transmission.
    */
@@ -165,12 +193,27 @@ struct PeerSession
    */
   struct GNUNET_FRAGMENT_Context *frag;
 
+  struct GNUNET_TIME_Absolute valid_until;
+
+  GNUNET_SCHEDULER_TaskIdentifier invalidation_task;
+
+  GNUNET_SCHEDULER_TaskIdentifier delayed_cont_task;
+
+  /**
+   * Desired delay for next sending we send to other peer
+   */
+  struct GNUNET_TIME_Relative flow_delay_for_other_peer;
+
+  /**
+   * Desired delay for next sending we received from other peer
+   */
+  struct GNUNET_TIME_Absolute flow_delay_from_other_peer;
 };
 
 
 /**
  * Data structure to track defragmentation contexts based
- * on the source of the UDP traffic.  
+ * on the source of the UDP traffic.
  */
 struct ReceiveContext
 {
@@ -201,6 +244,8 @@ struct ReceiveContext
    */
   size_t addr_len;
 
+  struct GNUNET_PeerIdentity id;
+
 };
 
 
@@ -221,6 +266,12 @@ struct Plugin
    */
   struct GNUNET_CONTAINER_MultiHashMap *sessions;
 
+  /**
+   * Session of peers with whom we are currently connected,
+   * map of peer identity to 'struct PeerSession'.
+   */
+  struct GNUNET_CONTAINER_MultiHashMap *inbound_sessions;
+
   /**
    * Heap with all of our defragmentation activities.
    */
@@ -272,7 +323,7 @@ struct Plugin
   struct GNUNET_NETWORK_Handle *sockv6;
 
   /**
-   * expected delay for ACKs 
+   * expected delay for ACKs
    */
   struct GNUNET_TIME_Relative last_expected_delay;
 
@@ -288,6 +339,13 @@ struct Plugin
 
 };
 
+struct PeerSessionIteratorContext
+{
+  struct Session * result;
+  const void * addr;
+  size_t addrlen;
+};
+
 
 /**
  * Lookup the session for the given peer.
@@ -296,7 +354,7 @@ struct Plugin
  * @param peer peer's identity
  * @return NULL if we have no session
  */
-struct PeerSession *
+static struct Session *
 find_session (struct Plugin *plugin, const struct GNUNET_PeerIdentity *peer)
 {
   return GNUNET_CONTAINER_multihashmap_get (plugin->sessions,
@@ -304,6 +362,127 @@ find_session (struct Plugin *plugin, const struct GNUNET_PeerIdentity *peer)
 }
 
 
+static int 
+inbound_session_iterator (void *cls,
+                         const GNUNET_HashCode * key,
+                         void *value)
+{
+  struct PeerSessionIteratorContext *psc = cls;
+  struct Session *s = value;
+  if (s->addrlen == psc->addrlen)
+  {
+    if (0 == memcmp (&s[1], psc->addr, s->addrlen))
+      psc->result = s;
+  }
+  if (psc->result != NULL)
+    return GNUNET_NO;
+  return GNUNET_YES;
+}
+
+
+/**
+ * Lookup the session for the given peer.
+ *
+ * @param plugin the plugin
+ * @param peer peer's identity
+ * @return NULL if we have no session
+ */
+static struct Session *
+find_inbound_session (struct Plugin *plugin,
+                      const struct GNUNET_PeerIdentity *peer,
+                      const void * addr, size_t addrlen)
+{
+  struct PeerSessionIteratorContext psc;
+  psc.result = NULL;
+  psc.addrlen = addrlen;
+  psc.addr = addr;
+
+  GNUNET_CONTAINER_multihashmap_get_multiple(plugin->inbound_sessions, &peer->hashPubKey, &inbound_session_iterator, &psc);
+  return psc.result;
+}
+
+
+static int 
+inbound_session_by_addr_iterator (void *cls,
+                                 const GNUNET_HashCode * key,
+                                 void *value)
+{
+  struct PeerSessionIteratorContext *psc = cls;
+  struct Session *s = value;
+  if (s->addrlen == psc->addrlen)
+  {
+    if (0 == memcmp (&s[1], psc->addr, s->addrlen))
+      psc->result = s;
+  }
+  if (psc->result != NULL)
+    return GNUNET_NO;
+  else
+    return GNUNET_YES;
+};
+
+/**
+ * Lookup the session for the given peer just by address.
+ *
+ * @param plugin the plugin
+ * @param addr address
+ * @param addrlen address length
+ * @return NULL if we have no session
+ */
+static struct Session *
+find_inbound_session_by_addr (struct Plugin *plugin, const void * addr, size_t addrlen)
+{
+  struct PeerSessionIteratorContext psc;
+  psc.result = NULL;
+  psc.addrlen = addrlen;
+  psc.addr = addr;
+
+  GNUNET_CONTAINER_multihashmap_iterate (plugin->inbound_sessions, &inbound_session_by_addr_iterator, &psc);
+  return psc.result;
+}
+
+
+/**
+ * Destroy a session, plugin is being unloaded.
+ *
+ * @param cls unused
+ * @param key hash of public key of target peer
+ * @param value a 'struct PeerSession*' to clean up
+ * @return GNUNET_OK (continue to iterate)
+ */
+static int
+destroy_session (void *cls, const GNUNET_HashCode * key, void *value)
+{
+  struct Session *peer_session = value;
+
+  if (peer_session->frag != NULL)
+    GNUNET_FRAGMENT_context_destroy (peer_session->frag);
+  if (GNUNET_SCHEDULER_NO_TASK != peer_session->delayed_cont_task)
+    GNUNET_SCHEDULER_cancel (peer_session->delayed_cont_task);
+  GNUNET_free (peer_session);
+  return GNUNET_OK;
+}
+
+/**
+ * Destroy a session, plugin is being unloaded.
+ *
+ * @param cls unused
+ * @param key hash of public key of target peer
+ * @param value a 'struct PeerSession*' to clean up
+ * @return GNUNET_OK (continue to iterate)
+ */
+static int
+destroy_inbound_session (void *cls, const GNUNET_HashCode * key, void *value)
+{
+  struct Session *s = value;
+
+  if (s->invalidation_task != GNUNET_SCHEDULER_NO_TASK)
+    GNUNET_SCHEDULER_cancel(s->invalidation_task);
+  if (GNUNET_SCHEDULER_NO_TASK != s->delayed_cont_task)
+    GNUNET_SCHEDULER_cancel (s->delayed_cont_task);
+  GNUNET_free (s);
+  return GNUNET_OK;
+}
+
 /**
  * Disconnect from a remote node.  Clean up session if we have one for this peer
  *
@@ -315,7 +494,7 @@ static void
 udp_disconnect (void *cls, const struct GNUNET_PeerIdentity *target)
 {
   struct Plugin *plugin = cls;
-  struct PeerSession *session;
+  struct Session *session;
 
   session = find_session (plugin, target);
   if (NULL == session)
@@ -324,8 +503,19 @@ udp_disconnect (void *cls, const struct GNUNET_PeerIdentity *target)
                  GNUNET_CONTAINER_multihashmap_remove (plugin->sessions,
                                                        &target->hashPubKey,
                                                        session));
+
+  GNUNET_CONTAINER_multihashmap_get_multiple (plugin->inbound_sessions,
+      &target->hashPubKey,
+      &destroy_inbound_session, NULL);
+  GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR, "sendto");
+  LOG (GNUNET_ERROR_TYPE_ERROR,
+              "UDP DISCONNECT\n");
+
   plugin->last_expected_delay = GNUNET_FRAGMENT_context_destroy (session->frag);
-  session->cont (session->cont_cls, target, GNUNET_SYSERR);
+  if (GNUNET_SCHEDULER_NO_TASK != session->delayed_cont_task)
+    GNUNET_SCHEDULER_cancel (session->delayed_cont_task);
+  if (session->cont != NULL)
+    session->cont (session->cont_cls, target, GNUNET_SYSERR);
   GNUNET_free (session);
 }
 
@@ -334,24 +524,13 @@ udp_disconnect (void *cls, const struct GNUNET_PeerIdentity *target)
  * Actually send out the message.
  *
  * @param plugin the plugin
- * @param send_handle which handle to send message on
- * @param target who should receive this message (ignored by UDP)
- * @param msgbuf one or more GNUNET_MessageHeader(s) strung together
- * @param msgbuf_size the size of the msgbuf to send
- * @param priority how important is the message (ignored by UDP)
- * @param timeout when should we time out (give up) if we can not transmit?
- * @param addr the addr to send the message to, needs to be a sockaddr for us
- * @param addrlen the len of addr
- * @param cont continuation to call once the message has
- *        been transmitted (or if the transport is ready
- *        for the next transmission call; or if the
- *        peer disconnected...)
- * @param cont_cls closure for cont
+ * @param sa the address to send the message to
+ * @param msg message to transmit
  * @return the number of bytes written
  */
 static ssize_t
-udp_send (struct Plugin *plugin,
-          const struct sockaddr *sa, const struct GNUNET_MessageHeader *msg)
+udp_send (struct Plugin *plugin, const struct sockaddr *sa,
+          const struct GNUNET_MessageHeader *msg)
 {
   ssize_t sent;
   size_t slen;
@@ -362,18 +541,14 @@ udp_send (struct Plugin *plugin,
     if (NULL == plugin->sockv4)
       return 0;
     sent =
-        GNUNET_NETWORK_socket_sendto (plugin->sockv4,
-                                      msg,
-                                      ntohs (msg->size),
+        GNUNET_NETWORK_socket_sendto (plugin->sockv4, msg, ntohs (msg->size),
                                       sa, slen = sizeof (struct sockaddr_in));
     break;
   case AF_INET6:
     if (NULL == plugin->sockv6)
       return 0;
     sent =
-        GNUNET_NETWORK_socket_sendto (plugin->sockv6,
-                                      msg,
-                                      ntohs (msg->size),
+        GNUNET_NETWORK_socket_sendto (plugin->sockv6, msg, ntohs (msg->size),
                                       sa, slen = sizeof (struct sockaddr_in6));
     break;
   default:
@@ -381,14 +556,18 @@ udp_send (struct Plugin *plugin,
     return 0;
   }
   if (GNUNET_SYSERR == sent)
-    GNUNET_log_strerror (GNUNET_ERROR_TYPE_INFO, "sendto");
-#if DEBUG_UDP
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+  {
+    GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR, "sendto");
+    LOG (GNUNET_ERROR_TYPE_ERROR,
+                "UDP transmited %u-byte message to %s (%d: %s)\n",
+                (unsigned int) ntohs (msg->size), GNUNET_a2s (sa, slen),
+                (int) sent, (sent < 0) ? STRERROR (errno) : "ok");
+
+  }
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
               "UDP transmited %u-byte message to %s (%d: %s)\n",
-              (unsigned int) ntohs (msg->size),
-              GNUNET_a2s (sa, slen),
+              (unsigned int) ntohs (msg->size), GNUNET_a2s (sa, slen),
               (int) sent, (sent < 0) ? STRERROR (errno) : "ok");
-#endif
   return sent;
 }
 
@@ -405,13 +584,98 @@ udp_send (struct Plugin *plugin,
 static void
 send_fragment (void *cls, const struct GNUNET_MessageHeader *msg)
 {
-  struct PeerSession *session = cls;
+  struct Session *session = cls;
 
   udp_send (session->plugin, session->sock_addr, msg);
   GNUNET_FRAGMENT_context_transmission_done (session->frag);
 }
 
 
+static struct Session *
+create_session (struct Plugin *plugin, const struct GNUNET_PeerIdentity *target,
+    const void *addr, size_t addrlen,
+    GNUNET_TRANSPORT_TransmitContinuation cont, void *cont_cls)
+{
+  struct Session * peer_session;
+  const struct IPv4UdpAddress *t4;
+  const struct IPv6UdpAddress *t6;
+  struct sockaddr_in *v4;
+  struct sockaddr_in6 *v6;
+  size_t len;
+
+  switch (addrlen)
+  {
+  case sizeof (struct IPv4UdpAddress):
+    if (NULL == plugin->sockv4)
+    {
+      return NULL;
+    }
+    t4 = addr;
+    peer_session =
+        GNUNET_malloc (sizeof (struct Session) +
+                       sizeof (struct sockaddr_in));
+    len = sizeof (struct sockaddr_in);
+    v4 = (struct sockaddr_in *) &peer_session[1];
+    v4->sin_family = AF_INET;
+#if HAVE_SOCKADDR_IN_SIN_LEN
+    v4->sin_len = sizeof (struct sockaddr_in);
+#endif
+    v4->sin_port = t4->u4_port;
+    v4->sin_addr.s_addr = t4->ipv4_addr;
+    break;
+  case sizeof (struct IPv6UdpAddress):
+    if (NULL == plugin->sockv6)
+    {
+      return NULL;
+    }
+    t6 = addr;
+    peer_session =
+        GNUNET_malloc (sizeof (struct Session) +
+                       sizeof (struct sockaddr_in6));
+    len = sizeof (struct sockaddr_in6);
+    v6 = (struct sockaddr_in6 *) &peer_session[1];
+    v6->sin6_family = AF_INET6;
+#if HAVE_SOCKADDR_IN_SIN_LEN
+    v6->sin6_len = sizeof (struct sockaddr_in6);
+#endif
+    v6->sin6_port = t6->u6_port;
+    v6->sin6_addr = t6->ipv6_addr;
+    break;
+  default:
+    /* Must have a valid address to send to */
+    GNUNET_break_op (0);
+    return NULL;
+  }
+
+  peer_session->valid_until = GNUNET_TIME_absolute_get_zero ();
+  peer_session->invalidation_task = GNUNET_SCHEDULER_NO_TASK;
+  peer_session->addrlen = len;
+  peer_session->target = *target;
+  peer_session->plugin = plugin;
+  peer_session->sock_addr = (const struct sockaddr *) &peer_session[1];
+  peer_session->cont = cont;
+  peer_session->cont_cls = cont_cls;
+
+  return peer_session;
+}
+
+static const char *
+udp_address_to_string (void *cls, const void *addr, size_t addrlen);
+
+
+static void
+udp_call_continuation (void *cls,
+                      const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  struct Session *s = cls;
+  GNUNET_TRANSPORT_TransmitContinuation cont = s->cont;
+
+  s->delayed_cont_task = GNUNET_SCHEDULER_NO_TASK;
+  s->cont = NULL;
+  cont (s->cont_cls, &s->target, GNUNET_OK);
+}
+
+
 /**
  * Function that can be used by the transport service to transmit
  * a message using the plugin.
@@ -437,95 +701,126 @@ send_fragment (void *cls, const struct GNUNET_MessageHeader *msg)
  *         still be transmitted later!)
  */
 static ssize_t
-udp_plugin_send (void *cls,
-                 const struct GNUNET_PeerIdentity *target,
-                 const char *msgbuf,
-                 size_t msgbuf_size,
-                 unsigned int priority,
-                 struct GNUNET_TIME_Relative timeout,
-                 struct Session *session,
-                 const void *addr,
-                 size_t addrlen,
-                 int force_address,
+udp_plugin_send (void *cls, const struct GNUNET_PeerIdentity *target,
+                 const char *msgbuf, size_t msgbuf_size, unsigned int priority,
+                 struct GNUNET_TIME_Relative timeout, struct Session *session,
+                 const void *addr, size_t addrlen, int force_address,
                  GNUNET_TRANSPORT_TransmitContinuation cont, void *cont_cls)
 {
   struct Plugin *plugin = cls;
-  struct PeerSession *peer_session;
+  struct Session *peer_session;
+  struct Session *s;
   const struct IPv4UdpAddress *t4;
   const struct IPv6UdpAddress *t6;
-  struct sockaddr_in *v4;
-  struct sockaddr_in6 *v6;
   size_t mlen = msgbuf_size + sizeof (struct UDPMessage);
   char mbuf[mlen];
   struct UDPMessage *udp;
+  struct GNUNET_TIME_Relative delta;
 
-  if (force_address == GNUNET_SYSERR)
-    return GNUNET_SYSERR;
-  GNUNET_assert (NULL == session);
   if (mlen >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
   {
     GNUNET_break (0);
     return GNUNET_SYSERR;
   }
-  switch (addrlen)
+
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+      "UDP transmits %u-byte message to `%s' using address `%s' session 0x%X mode %i\n",
+      msgbuf_size, GNUNET_i2s (target),
+      udp_address_to_string (NULL, addr, addrlen),
+      session, force_address);
+
+  if ((force_address == GNUNET_SYSERR) && (session == NULL))
+    return GNUNET_SYSERR;
+
+  s = NULL;
+  /* safety check: comparing address to address stored in session */
+  if ((session != NULL) && (addr != NULL) && (addrlen != 0))
   {
-  case sizeof (struct IPv4UdpAddress):
-    if (NULL == plugin->sockv4)
-    {
-      cont (cont_cls, target, GNUNET_SYSERR);
-      return 0;
-    }
-    t4 = addr;
-    peer_session =
-        GNUNET_malloc (sizeof (struct PeerSession) +
-                       sizeof (struct sockaddr_in));
-    v4 = (struct sockaddr_in *) &peer_session[1];
-    v4->sin_family = AF_INET;
-#if HAVE_SOCKADDR_IN_SIN_LEN
-    v4->sin_len = sizeof (struct sockaddr_in);
-#endif
-    v4->sin_port = t4->u4_port;
-    v4->sin_addr.s_addr = t4->ipv4_addr;
-    break;
-  case sizeof (struct IPv6UdpAddress):
-    if (NULL == plugin->sockv6)
+    s = session;
+    GNUNET_assert (GNUNET_YES == GNUNET_CONTAINER_multihashmap_contains_value (
+        plugin->inbound_sessions, &target->hashPubKey, s));
+
+    if  (0 != memcmp (&s->target, target, sizeof (struct GNUNET_PeerIdentity)))
+      return GNUNET_SYSERR;
+    switch (addrlen)
     {
-      cont (cont_cls, target, GNUNET_SYSERR);
-      return 0;
+    case sizeof (struct IPv4UdpAddress):
+      if (NULL == plugin->sockv4)
+      {
+        if (cont != NULL)
+          cont (cont_cls, target, GNUNET_SYSERR);
+        return GNUNET_SYSERR;
+      }
+      t4 = addr;
+      if (s->addrlen != (sizeof (struct sockaddr_in)))
+        return GNUNET_SYSERR;
+      struct sockaddr_in *a4 = (struct sockaddr_in *) s->sock_addr;
+      GNUNET_assert (a4->sin_port == t4->u4_port);
+      GNUNET_assert (0 == memcmp(&a4->sin_addr, &t4->ipv4_addr, sizeof (struct in_addr)));
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
+                  "Session 0x%X successfully checked!\n", session);
+      break;
+    case sizeof (struct IPv6UdpAddress):
+      if (NULL == plugin->sockv6)
+      {
+        if (cont != NULL)
+          cont (cont_cls, target, GNUNET_SYSERR);
+        return GNUNET_SYSERR;
+      }
+      t6 = addr;
+      GNUNET_assert (s->addrlen == sizeof (struct sockaddr_in6));
+      struct sockaddr_in6 *a6 = (struct sockaddr_in6 *) s->sock_addr;
+      GNUNET_assert (a6->sin6_port == t6->u6_port);
+      GNUNET_assert (0 == memcmp(&a6->sin6_addr, &t6->ipv6_addr, sizeof (struct in6_addr)));
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
+                  "Session 0x%X successfully checked!\n", session);
+      break;
+    default:
+      /* Must have a valid address to send to */
+      GNUNET_break_op (0);
     }
-    t6 = addr;
-    peer_session =
-        GNUNET_malloc (sizeof (struct PeerSession) +
-                       sizeof (struct sockaddr_in6));
-    v6 = (struct sockaddr_in6 *) &peer_session[1];
-    v6->sin6_family = AF_INET6;
-#if HAVE_SOCKADDR_IN_SIN_LEN
-    v6->sin6_len = sizeof (struct sockaddr_in6);
-#endif
-    v6->sin6_port = t6->u6_port;
-    v6->sin6_addr = t6->ipv6_addr;
-    break;
-  default:
-    /* Must have a valid address to send to */
-    GNUNET_break_op (0);
+  }
+//session_invalid:
+  if ((addr == NULL) || (addrlen == 0))
     return GNUNET_SYSERR;
+  peer_session = create_session (plugin, target, addr, addrlen, cont, cont_cls);
+  if (peer_session == NULL)
+  {
+    if (cont != NULL)
+      cont (cont_cls, target, GNUNET_SYSERR);
+    return GNUNET_SYSERR;;
   }
+
+  /* Message */
   udp = (struct UDPMessage *) mbuf;
   udp->header.size = htons (mlen);
   udp->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_MESSAGE);
   udp->reserved = htonl (0);
   udp->sender = *plugin->env->my_identity;
   memcpy (&udp[1], msgbuf, msgbuf_size);
-  peer_session->target = *target;
-  peer_session->plugin = plugin;
-  peer_session->sock_addr = (const struct sockaddr *) &peer_session[1];
-  peer_session->cont = cont;
-  peer_session->cont_cls = cont_cls;
+
+  if (s != NULL)  
+    delta = GNUNET_TIME_absolute_get_remaining (s->flow_delay_from_other_peer);
+  else
+    delta = GNUNET_TIME_UNIT_ZERO;
   if (mlen <= UDP_MTU)
   {
     mlen = udp_send (plugin, peer_session->sock_addr, &udp->header);
-    cont (cont_cls, target, (mlen > 0) ? GNUNET_OK : GNUNET_SYSERR);
-    GNUNET_free (peer_session);
+    if (cont != NULL)
+    {
+      if ( (delta.rel_value > 0) &&
+          (mlen > 0) )
+      {
+       s->cont = cont;
+       s->cont_cls = cont_cls;
+       s->delayed_cont_task = GNUNET_SCHEDULER_add_delayed (delta,
+                                                            &udp_call_continuation,
+                                                            s);
+      }
+      else
+       cont (cont_cls, target, (mlen > 0) ? GNUNET_OK : GNUNET_SYSERR);
+    }
+    GNUNET_free_non_null (peer_session);
   }
   else
   {
@@ -534,13 +829,12 @@ udp_plugin_send (void *cls,
                                                       &target->hashPubKey,
                                                       peer_session,
                                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
-    peer_session->frag = GNUNET_FRAGMENT_context_create (plugin->env->stats,
-                                                         UDP_MTU,
-                                                         &plugin->tracker,
-                                                         plugin->last_expected_delay,
-                                                         &udp->header,
-                                                         &send_fragment,
-                                                         peer_session);
+    peer_session->frag =
+        GNUNET_FRAGMENT_context_create (plugin->env->stats, UDP_MTU,
+                                        &plugin->tracker,
+                                        plugin->last_expected_delay,
+                                        &udp->header, &send_fragment,
+                                        peer_session);
   }
   return mlen;
 }
@@ -565,11 +859,13 @@ struct SourceInformation
    * Number of bytes in source address.
    */
   size_t args;
+
+  struct Session * session;
 };
 
 
 /**
- * Message tokenizer has broken up an incomming message. Pass it on 
+ * Message tokenizer has broken up an incomming message. Pass it on
  * to the service.
  *
  * @param cls the 'struct Plugin'
@@ -577,22 +873,36 @@ struct SourceInformation
  * @param hdr the actual message
  */
 static void
-process_inbound_tokenized_messages (void *cls,
-                                    void *client,
+process_inbound_tokenized_messages (void *cls, void *client,
                                     const struct GNUNET_MessageHeader *hdr)
 {
   struct Plugin *plugin = cls;
   struct SourceInformation *si = client;
-  struct GNUNET_TRANSPORT_ATS_Information distance[2];
+  struct GNUNET_ATS_Information distance;
+  struct GNUNET_TIME_Relative delay;
 
   /* setup ATS */
-  distance[0].type = htonl (GNUNET_TRANSPORT_ATS_QUALITY_NET_DISTANCE);
-  distance[0].value = htonl (1);
-  distance[1].type = htonl (GNUNET_TRANSPORT_ATS_ARRAY_TERMINATOR);
-  distance[1].value = htonl (0);
+  distance.type = htonl (GNUNET_ATS_QUALITY_NET_DISTANCE);
+  distance.value = htonl (1);
+
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+                   "Giving Session %X %s  to transport\n", si->session, GNUNET_i2s(&si->session->target));
+  delay = plugin->env->receive (plugin->env->cls, &si->sender, hdr, &distance, 1, si->session,
+                        si->arg, si->args);
+  si->session->flow_delay_for_other_peer = delay;
+}
 
-  plugin->env->receive (plugin->env->cls,
-                        &si->sender, hdr, distance, 2, NULL, si->arg, si->args);
+static void
+invalidation_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  struct Session * s = cls;
+  s->invalidation_task = GNUNET_SCHEDULER_NO_TASK;
+  LOG (GNUNET_ERROR_TYPE_ERROR,
+                   "Session %X (`%s') is now invalid\n", s, GNUNET_a2s (s->sock_addr,s->addrlen));
+
+  s->plugin->env->session_end(s->plugin->env->cls, &s->target, s);
+  GNUNET_assert (GNUNET_YES == GNUNET_CONTAINER_multihashmap_remove(s->plugin->inbound_sessions, &s->target.hashPubKey, s));
+  GNUNET_free (s);
 }
 
 
@@ -601,12 +911,11 @@ process_inbound_tokenized_messages (void *cls,
  *
  * @param plugin plugin context
  * @param msg the message
- * @param sender_addr sender address 
+ * @param sender_addr sender address
  * @param sender_addr_len number of bytes in sender_addr
  */
 static void
-process_udp_message (struct Plugin *plugin,
-                     const struct UDPMessage *msg,
+process_udp_message (struct Plugin *plugin, const struct UDPMessage *msg,
                      const struct sockaddr *sender_addr,
                      socklen_t sender_addr_len)
 {
@@ -650,21 +959,61 @@ process_udp_message (struct Plugin *plugin,
     return;
   }
 #if DEBUG_UDP
-  GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
-                   "udp",
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
                    "Received message with %u bytes from peer `%s' at `%s'\n",
                    (unsigned int) ntohs (msg->header.size),
-                   GNUNET_i2s (&msg->sender),
-                   GNUNET_a2s (sender_addr, sender_addr_len));
+                   GNUNET_i2s (&msg->sender), GNUNET_a2s (sender_addr,
+                                                          sender_addr_len));
 #endif
 
+  /* create a session for inbound connections */
+  const struct UDPMessage * udp_msg = (const struct UDPMessage *) msg;
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+              "Lookup inbound UDP sessions for peer `%s' address `%s'\n",
+              GNUNET_i2s (&udp_msg->sender),
+              udp_address_to_string(NULL, arg, args));
+
+  struct Session * s = NULL;
+  s = find_inbound_session (plugin, &udp_msg->sender, sender_addr, sender_addr_len);
+
+  if (s != NULL)
+  {
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+                "Found existing inbound UDP sessions 0x%X for peer `%s' address `%s'\n",
+                s,
+                GNUNET_i2s (&s->target),
+                udp_address_to_string(NULL, arg, args));
+  }
+  else
+  {
+    s = create_session (plugin, &udp_msg->sender, arg, args, NULL, NULL);
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+                "Creating inbound UDP sessions 0x%X for peer `%s' address `%s'\n",
+                s,
+                GNUNET_i2s (&s->target),
+                udp_address_to_string(NULL, arg, args));
+
+    GNUNET_assert (GNUNET_OK == GNUNET_CONTAINER_multihashmap_put (plugin->inbound_sessions,
+                                                      &s->target.hashPubKey,
+                                                      s,
+                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
+  }
+  s->valid_until = GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT);
+  if (s->invalidation_task != GNUNET_SCHEDULER_NO_TASK)
+  {
+    GNUNET_SCHEDULER_cancel(s->invalidation_task);
+    s->invalidation_task = GNUNET_SCHEDULER_NO_TASK;
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+                "Rescheduling %X' `%s'\n",
+                s, udp_address_to_string(NULL, arg, args));
+  }
+  s->invalidation_task = GNUNET_SCHEDULER_add_delayed(GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT, &invalidation_task, s);
   /* iterate over all embedded messages */
   si.sender = msg->sender;
   si.arg = arg;
   si.args = args;
-  GNUNET_SERVER_mst_receive (plugin->mst,
-                             &si,
-                             (const char *) &msg[1],
+  si.session = s;
+  GNUNET_SERVER_mst_receive (plugin->mst, &si, (const char *) &msg[1],
                              ntohs (msg->header.size) -
                              sizeof (struct UDPMessage), GNUNET_YES, GNUNET_NO);
 }
@@ -691,8 +1040,7 @@ fragment_msg_proc (void *cls, const struct GNUNET_MessageHeader *msg)
     GNUNET_break (0);
     return;
   }
-  process_udp_message (rc->plugin,
-                       (const struct UDPMessage *) msg,
+  process_udp_message (rc->plugin, (const struct UDPMessage *) msg,
                        rc->src_addr, rc->addr_len);
 }
 
@@ -708,26 +1056,38 @@ static void
 ack_proc (void *cls, uint32_t id, const struct GNUNET_MessageHeader *msg)
 {
   struct ReceiveContext *rc = cls;
-  size_t msize = sizeof (struct UDPMessage) + ntohs (msg->size);
+
+  size_t msize = sizeof (struct UDP_ACK_Message) + ntohs (msg->size);
   char buf[msize];
-  struct UDPMessage *udp;
+  struct UDP_ACK_Message *udp_ack;
+  uint32_t delay = 0;
+
+  struct Session *s;
+  s = find_inbound_session_by_addr (rc->plugin, rc->src_addr, rc->addr_len);
+  if (s != NULL)
+  {
+    if (s->flow_delay_for_other_peer.rel_value <= UINT32_MAX)
+      delay = s->flow_delay_for_other_peer.rel_value;
+    else
+      delay = UINT32_MAX;
+  }
+
 
 #if DEBUG_UDP
-  GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG,
-                   "udp",
-                   "Sending ACK to `%s'\n",
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "Sending ACK to `%s' including delay of %u ms\n",
                    GNUNET_a2s (rc->src_addr,
-                               (rc->src_addr->sa_family == AF_INET)
-                               ? sizeof (struct sockaddr_in)
-                               : sizeof (struct sockaddr_in6)));
+                               (rc->src_addr->sa_family ==
+                                AF_INET) ? sizeof (struct sockaddr_in) :
+                               sizeof (struct sockaddr_in6)),
+                               delay);
 #endif
-  udp = (struct UDPMessage *) buf;
-  udp->header.size = htons ((uint16_t) msize);
-  udp->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_ACK);
-  udp->reserved = htonl (0);
-  udp->sender = *rc->plugin->env->my_identity;
-  memcpy (&udp[1], msg, ntohs (msg->size));
-  (void) udp_send (rc->plugin, rc->src_addr, &udp->header);
+  udp_ack = (struct UDP_ACK_Message *) buf;
+  udp_ack->header.size = htons ((uint16_t) msize);
+  udp_ack->header.type = htons (GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_ACK);
+  udp_ack->delay = htonl (delay);
+  udp_ack->sender = *rc->plugin->env->my_identity;
+  memcpy (&udp_ack[1], msg, ntohs (msg->size));
+  (void) udp_send (rc->plugin, rc->src_addr, &udp_ack->header);
 }
 
 
@@ -750,6 +1110,8 @@ struct FindReceiveContext
    * Number of bytes in 'addr'.
    */
   socklen_t addr_len;
+
+  struct Session * session;
 };
 
 
@@ -764,8 +1126,7 @@ struct FindReceiveContext
  *         GNUNET_NO if not.
  */
 static int
-find_receive_context (void *cls,
-                      struct GNUNET_CONTAINER_HeapNode *node,
+find_receive_context (void *cls, struct GNUNET_CONTAINER_HeapNode *node,
                       void *element, GNUNET_CONTAINER_HeapCostType cost)
 {
   struct FindReceiveContext *frc = cls;
@@ -796,28 +1157,30 @@ udp_read (struct Plugin *plugin, struct GNUNET_NETWORK_Handle *rsock)
   ssize_t ret;
   const struct GNUNET_MessageHeader *msg;
   const struct GNUNET_MessageHeader *ack;
-  struct PeerSession *peer_session;
-  const struct UDPMessage *udp;
+  struct Session *peer_session;
+  const struct UDP_ACK_Message *udp_ack;
   struct ReceiveContext *rc;
   struct GNUNET_TIME_Absolute now;
   struct FindReceiveContext frc;
+  struct Session * s = NULL;
+  struct GNUNET_TIME_Relative flow_delay;
 
   fromlen = sizeof (addr);
   memset (&addr, 0, sizeof (addr));
-  ret = GNUNET_NETWORK_socket_recvfrom (rsock, buf, sizeof (buf),
-                                        (struct sockaddr *) &addr, &fromlen);
+  ret =
+      GNUNET_NETWORK_socket_recvfrom (rsock, buf, sizeof (buf),
+                                      (struct sockaddr *) &addr, &fromlen);
   if (ret < sizeof (struct GNUNET_MessageHeader))
   {
     GNUNET_break_op (0);
     return;
   }
-#if DEBUG_UDP
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "UDP received %u-byte message from `%s'\n",
-              (unsigned int) ret,
-              GNUNET_a2s ((const struct sockaddr *) addr, fromlen));
-#endif
   msg = (const struct GNUNET_MessageHeader *) buf;
+
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+              "UDP received %u-byte message from `%s' type %i\n", (unsigned int) ret,
+              GNUNET_a2s ((const struct sockaddr *) addr, fromlen), ntohs(msg->type));
+
   if (ret != ntohs (msg->size))
   {
     GNUNET_break_op (0);
@@ -831,42 +1194,46 @@ udp_read (struct Plugin *plugin, struct GNUNET_NETWORK_Handle *rsock)
       GNUNET_break_op (0);
       return;
     }
-    process_udp_message (plugin,
-                         (const struct UDPMessage *) msg,
+    process_udp_message (plugin, (const struct UDPMessage *) msg,
                          (const struct sockaddr *) addr, fromlen);
     return;
   case GNUNET_MESSAGE_TYPE_TRANSPORT_UDP_ACK:
+
     if (ntohs (msg->size) <
-        sizeof (struct UDPMessage) + sizeof (struct GNUNET_MessageHeader))
+        sizeof (struct UDP_ACK_Message) + sizeof (struct GNUNET_MessageHeader))
     {
       GNUNET_break_op (0);
       return;
     }
-    udp = (const struct UDPMessage *) msg;
-    if (ntohl (udp->reserved) != 0)
+    udp_ack = (const struct UDP_ACK_Message *) msg;
+    s = find_inbound_session(plugin, &udp_ack->sender, addr, fromlen);
+    if (s != NULL)
     {
-      GNUNET_break_op (0);
-      return;
+      flow_delay.rel_value = (uint64_t) ntohl(udp_ack->delay);
+
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
+                  "We received a sending delay of %llu\n", flow_delay.rel_value);
+
+      s->flow_delay_from_other_peer = GNUNET_TIME_relative_to_absolute (flow_delay);
     }
-    ack = (const struct GNUNET_MessageHeader *) &udp[1];
-    if (ntohs (ack->size) != ntohs (msg->size) - sizeof (struct UDPMessage))
+    ack = (const struct GNUNET_MessageHeader *) &udp_ack[1];
+    if (ntohs (ack->size) != ntohs (msg->size) - sizeof (struct UDP_ACK_Message))
     {
       GNUNET_break_op (0);
       return;
     }
 #if DEBUG_UDP
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
                 "UDP processes %u-byte acknowledgement from `%s' at `%s'\n",
-                (unsigned int) ntohs (msg->size),
-                GNUNET_i2s (&udp->sender),
+                (unsigned int) ntohs (msg->size), GNUNET_i2s (&udp_ack->sender),
                 GNUNET_a2s ((const struct sockaddr *) addr, fromlen));
 #endif
 
-    peer_session = find_session (plugin, &udp->sender);
+    peer_session = find_session (plugin, &udp_ack->sender);
     if (NULL == peer_session)
     {
 #if DEBUG_UDP
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
                   "Session for ACK not found, dropping ACK!\n");
 #endif
       return;
@@ -875,20 +1242,21 @@ udp_read (struct Plugin *plugin, struct GNUNET_NETWORK_Handle *rsock)
       return;
     GNUNET_assert (GNUNET_OK ==
                    GNUNET_CONTAINER_multihashmap_remove (plugin->sessions,
-                                                         &udp->
+                                                         &udp_ack->
                                                          sender.hashPubKey,
                                                          peer_session));
     plugin->last_expected_delay =
         GNUNET_FRAGMENT_context_destroy (peer_session->frag);
-    peer_session->cont (peer_session->cont_cls, &udp->sender, GNUNET_OK);
+    if (peer_session->cont != NULL)
+      peer_session->cont (peer_session->cont_cls, &udp_ack->sender, GNUNET_OK);
     GNUNET_free (peer_session);
     return;
   case GNUNET_MESSAGE_TYPE_FRAGMENT:
     frc.rc = NULL;
     frc.addr = (const struct sockaddr *) addr;
     frc.addr_len = fromlen;
-    GNUNET_CONTAINER_heap_iterate (plugin->defrags,
-                                   &find_receive_context, &frc);
+    GNUNET_CONTAINER_heap_iterate (plugin->defrags, &find_receive_context,
+                                   &frc);
     now = GNUNET_TIME_absolute_get ();
     rc = frc.rc;
     if (rc == NULL)
@@ -899,19 +1267,17 @@ udp_read (struct Plugin *plugin, struct GNUNET_NETWORK_Handle *rsock)
       rc->src_addr = (const struct sockaddr *) &rc[1];
       rc->addr_len = fromlen;
       rc->plugin = plugin;
-      rc->defrag = GNUNET_DEFRAGMENT_context_create (plugin->env->stats,
-                                                     UDP_MTU,
-                                                     UDP_MAX_MESSAGES_IN_DEFRAG,
-                                                     rc,
-                                                     &fragment_msg_proc,
-                                                     &ack_proc);
-      rc->hnode = GNUNET_CONTAINER_heap_insert (plugin->defrags,
-                                                rc,
-                                                (GNUNET_CONTAINER_HeapCostType)
-                                                now.abs_value);
+      rc->defrag =
+          GNUNET_DEFRAGMENT_context_create (plugin->env->stats, UDP_MTU,
+                                            UDP_MAX_MESSAGES_IN_DEFRAG, rc,
+                                            &fragment_msg_proc, &ack_proc);
+      rc->hnode =
+          GNUNET_CONTAINER_heap_insert (plugin->defrags, rc,
+                                        (GNUNET_CONTAINER_HeapCostType)
+                                        now.abs_value);
     }
 #if DEBUG_UDP
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
                 "UDP processes %u-byte fragment from `%s'\n",
                 (unsigned int) ntohs (msg->size),
                 GNUNET_a2s ((const struct sockaddr *) addr, fromlen));
@@ -920,8 +1286,7 @@ udp_read (struct Plugin *plugin, struct GNUNET_NETWORK_Handle *rsock)
     if (GNUNET_OK == GNUNET_DEFRAGMENT_process_fragment (rc->defrag, msg))
     {
       /* keep this 'rc' from expiring */
-      GNUNET_CONTAINER_heap_update_cost (plugin->defrags,
-                                         rc->hnode,
+      GNUNET_CONTAINER_heap_update_cost (plugin->defrags, rc->hnode,
                                          (GNUNET_CONTAINER_HeapCostType)
                                          now.abs_value);
     }
@@ -1026,8 +1391,8 @@ udp_plugin_check_address (void *cls, const void *addr, size_t addrlen)
     if (GNUNET_OK != check_port (plugin, ntohs (v4->u4_port)))
       return GNUNET_SYSERR;
     if (GNUNET_OK !=
-        GNUNET_NAT_test_address (plugin->nat,
-                                 &v4->ipv4_addr, sizeof (struct in_addr)))
+        GNUNET_NAT_test_address (plugin->nat, &v4->ipv4_addr,
+                                 sizeof (struct in_addr)))
       return GNUNET_SYSERR;
   }
   else
@@ -1041,8 +1406,8 @@ udp_plugin_check_address (void *cls, const void *addr, size_t addrlen)
     if (GNUNET_OK != check_port (plugin, ntohs (v6->u6_port)))
       return GNUNET_SYSERR;
     if (GNUNET_OK !=
-        GNUNET_NAT_test_address (plugin->nat,
-                                 &v6->ipv6_addr, sizeof (struct in6_addr)))
+        GNUNET_NAT_test_address (plugin->nat, &v6->ipv6_addr,
+                                 sizeof (struct in6_addr)))
       return GNUNET_SYSERR;
   }
   return GNUNET_OK;
@@ -1095,7 +1460,8 @@ udp_address_to_string (void *cls, const void *addr, size_t addrlen)
     return NULL;
   }
   inet_ntop (af, sb, buf, INET6_ADDRSTRLEN);
-  GNUNET_snprintf (rbuf, sizeof (rbuf), "%s:%u", buf, port);
+  GNUNET_snprintf (rbuf, sizeof (rbuf), (af == AF_INET6) ? "[%s]:%u" : "%s:%u",
+                  buf, port);
   return rbuf;
 }
 
@@ -1161,10 +1527,8 @@ append_port (void *cls, const char *hostname)
  * @param asc_cls closure for asc
  */
 static void
-udp_plugin_address_pretty_printer (void *cls,
-                                   const char *type,
-                                   const void *addr,
-                                   size_t addrlen,
+udp_plugin_address_pretty_printer (void *cls, const char *type,
+                                   const void *addr, size_t addrlen,
                                    int numeric,
                                    struct GNUNET_TIME_Relative timeout,
                                    GNUNET_TRANSPORT_AddressStringCallback asc,
@@ -1232,8 +1596,7 @@ udp_plugin_address_pretty_printer (void *cls,
  * @param addrlen actual lenght of the address
  */
 static void
-udp_nat_port_map_callback (void *cls,
-                           int add_remove,
+udp_nat_port_map_callback (void *cls, int add_remove,
                            const struct sockaddr *addr, socklen_t addrlen)
 {
   struct Plugin *plugin = cls;
@@ -1254,8 +1617,7 @@ udp_nat_port_map_callback (void *cls,
     break;
   case AF_INET6:
     GNUNET_assert (addrlen == sizeof (struct sockaddr_in6));
-    memcpy (&u6.ipv6_addr,
-            &((struct sockaddr_in6 *) addr)->sin6_addr,
+    memcpy (&u6.ipv6_addr, &((struct sockaddr_in6 *) addr)->sin6_addr,
             sizeof (struct in6_addr));
     u6.u6_port = ((struct sockaddr_in6 *) addr)->sin6_port;
     arg = &u6;
@@ -1296,24 +1658,22 @@ libgnunet_plugin_transport_udp_init (void *cls)
   unsigned long long udp_max_bps;
 
   if (GNUNET_OK !=
-      GNUNET_CONFIGURATION_get_value_number (env->cfg,
-                                             "transport-udp", "PORT", &port))
+      GNUNET_CONFIGURATION_get_value_number (env->cfg, "transport-udp", "PORT",
+                                             &port))
     port = 2086;
   if (GNUNET_OK !=
-      GNUNET_CONFIGURATION_get_value_number (env->cfg,
-                                             "transport-udp",
+      GNUNET_CONFIGURATION_get_value_number (env->cfg, "transport-udp",
                                              "MAX_BPS", &udp_max_bps))
     udp_max_bps = 1024 * 1024 * 50;     /* 50 MB/s == infinity for practical purposes */
   if (GNUNET_OK !=
-      GNUNET_CONFIGURATION_get_value_number (env->cfg,
-                                             "transport-udp",
+      GNUNET_CONFIGURATION_get_value_number (env->cfg, "transport-udp",
                                              "ADVERTISED_PORT", &aport))
     aport = port;
   if (port > 65535)
   {
-    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-                _("Given `%s' option is out of range: %llu > %u\n"),
-                "PORT", port, 65535);
+    LOG (GNUNET_ERROR_TYPE_WARNING,
+                _("Given `%s' option is out of range: %llu > %u\n"), "PORT",
+                port, 65535);
     return NULL;
   }
   memset (&serverAddrv6, 0, sizeof (serverAddrv6));
@@ -1336,12 +1696,11 @@ libgnunet_plugin_transport_udp_init (void *cls)
   api->address_to_string = &udp_address_to_string;
   api->check_address = &udp_plugin_check_address;
 
-  if (GNUNET_YES == GNUNET_CONFIGURATION_get_value_string (env->cfg,
-                                                           "transport-udp",
-                                                           "BINDTO",
-                                                           &plugin->bind4_address))
+  if (GNUNET_YES ==
+      GNUNET_CONFIGURATION_get_value_string (env->cfg, "transport-udp",
+                                             "BINDTO", &plugin->bind4_address))
   {
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
                 "Binding udp plugin to specific address: `%s'\n",
                 plugin->bind4_address);
     if (1 != inet_pton (AF_INET, plugin->bind4_address, &serverAddrv4.sin_addr))
@@ -1353,19 +1712,18 @@ libgnunet_plugin_transport_udp_init (void *cls)
     }
   }
 
-  if (GNUNET_YES == GNUNET_CONFIGURATION_get_value_string (env->cfg,
-                                                           "transport-udp",
-                                                           "BINDTO6",
-                                                           &plugin->bind6_address))
+  if (GNUNET_YES ==
+      GNUNET_CONFIGURATION_get_value_string (env->cfg, "transport-udp",
+                                             "BINDTO6", &plugin->bind6_address))
   {
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
                 "Binding udp plugin to specific address: `%s'\n",
                 plugin->bind6_address);
-    if (1 != inet_pton (AF_INET6,
-                        plugin->bind6_address, &serverAddrv6.sin6_addr))
+    if (1 !=
+        inet_pton (AF_INET6, plugin->bind6_address, &serverAddrv6.sin6_addr))
     {
-      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                  _("Invalid IPv6 address: `%s'\n"), plugin->bind6_address);
+      LOG (GNUNET_ERROR_TYPE_ERROR, _("Invalid IPv6 address: `%s'\n"),
+                  plugin->bind6_address);
       GNUNET_free_non_null (plugin->bind4_address);
       GNUNET_free (plugin->bind6_address);
       GNUNET_free (plugin);
@@ -1379,10 +1737,13 @@ libgnunet_plugin_transport_udp_init (void *cls)
   plugin->sessions =
       GNUNET_CONTAINER_multihashmap_create (UDP_MAX_SENDER_ADDRESSES_WITH_DEFRAG
                                             * 2);
+  plugin->inbound_sessions =
+      GNUNET_CONTAINER_multihashmap_create (UDP_MAX_SENDER_ADDRESSES_WITH_DEFRAG
+                                            * 2);
   sockets_created = 0;
   if ((GNUNET_YES !=
-       GNUNET_CONFIGURATION_get_value_yesno (plugin->env->cfg,
-                                             "nat", "DISABLEV6")))
+       GNUNET_CONFIGURATION_get_value_yesno (plugin->env->cfg, "nat",
+                                             "DISABLEV6")))
   {
     plugin->sockv6 = GNUNET_NETWORK_socket_create (PF_INET6, SOCK_DGRAM, 0);
     if (NULL == plugin->sockv6)
@@ -1400,16 +1761,16 @@ libgnunet_plugin_transport_udp_init (void *cls)
       addrlen = sizeof (serverAddrv6);
       serverAddr = (struct sockaddr *) &serverAddrv6;
 #if DEBUG_UDP
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                  "Binding to IPv6 port %d\n", ntohs (serverAddrv6.sin6_port));
+      LOG (GNUNET_ERROR_TYPE_DEBUG, "Binding to IPv6 port %d\n",
+                  ntohs (serverAddrv6.sin6_port));
 #endif
       tries = 0;
-      while (GNUNET_NETWORK_socket_bind (plugin->sockv6,
-                                         serverAddr, addrlen) != GNUNET_OK)
+      while (GNUNET_NETWORK_socket_bind (plugin->sockv6, serverAddr, addrlen) !=
+             GNUNET_OK)
       {
         serverAddrv6.sin6_port = htons (GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_STRONG, 33537) + 32000);        /* Find a good, non-root port */
 #if DEBUG_UDP
-        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+        LOG (GNUNET_ERROR_TYPE_DEBUG,
                     "IPv6 Binding failed, trying new port %d\n",
                     ntohs (serverAddrv6.sin6_port));
 #endif
@@ -1430,8 +1791,8 @@ libgnunet_plugin_transport_udp_init (void *cls)
     }
   }
 
-  plugin->mst = GNUNET_SERVER_mst_create (&process_inbound_tokenized_messages,
-                                          plugin);
+  plugin->mst =
+      GNUNET_SERVER_mst_create (&process_inbound_tokenized_messages, plugin);
   plugin->sockv4 = GNUNET_NETWORK_socket_create (PF_INET, SOCK_DGRAM, 0);
   if (NULL == plugin->sockv4)
   {
@@ -1448,8 +1809,8 @@ libgnunet_plugin_transport_udp_init (void *cls)
     addrlen = sizeof (serverAddrv4);
     serverAddr = (struct sockaddr *) &serverAddrv4;
 #if DEBUG_UDP
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                "Binding to IPv4 port %d\n", ntohs (serverAddrv4.sin_port));
+    LOG (GNUNET_ERROR_TYPE_DEBUG, "Binding to IPv4 port %d\n",
+                ntohs (serverAddrv4.sin_port));
 #endif
     tries = 0;
     while (GNUNET_NETWORK_socket_bind (plugin->sockv4, serverAddr, addrlen) !=
@@ -1457,7 +1818,7 @@ libgnunet_plugin_transport_udp_init (void *cls)
     {
       serverAddrv4.sin_port = htons (GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_STRONG, 33537) + 32000);   /* Find a good, non-root port */
 #if DEBUG_UDP
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
                   "IPv4 Binding failed, trying new port %d\n",
                   ntohs (serverAddrv4.sin_port));
 #endif
@@ -1491,35 +1852,13 @@ libgnunet_plugin_transport_udp_init (void *cls)
                                    NULL, &udp_plugin_select, plugin);
   if (sockets_created == 0)
     GNUNET_log (GNUNET_ERROR_TYPE_WARNING, _("Failed to open UDP sockets\n"));
-  plugin->nat = GNUNET_NAT_register (env->cfg,
-                                     GNUNET_NO,
-                                     port,
-                                     sockets_created,
-                                     (const struct sockaddr **) addrs, addrlens,
-                                     &udp_nat_port_map_callback, NULL, plugin);
+  plugin->nat =
+      GNUNET_NAT_register (env->cfg, GNUNET_NO, port, sockets_created,
+                           (const struct sockaddr **) addrs, addrlens,
+                           &udp_nat_port_map_callback, NULL, plugin);
   return api;
 }
 
-
-/**
- * Destroy a session, plugin is being unloaded.
- *
- * @param cls unused
- * @param key hash of public key of target peer
- * @param value a 'struct PeerSession*' to clean up
- * @return GNUNET_OK (continue to iterate)
- */
-static int
-destroy_session (void *cls, const GNUNET_HashCode * key, void *value)
-{
-  struct PeerSession *peer_session = value;
-
-  GNUNET_FRAGMENT_context_destroy (peer_session->frag);
-  GNUNET_free (peer_session);
-  return GNUNET_OK;
-}
-
-
 /**
  * Shutdown the plugin.
  *
@@ -1534,10 +1873,14 @@ libgnunet_plugin_transport_udp_done (void *cls)
   struct ReceiveContext *rc;
 
   /* FIXME: clean up heap and hashmap */
-  GNUNET_CONTAINER_multihashmap_iterate (plugin->sessions,
-                                         &destroy_session, NULL);
+  GNUNET_CONTAINER_multihashmap_iterate (plugin->sessions, &destroy_session,
+                                         NULL);
   GNUNET_CONTAINER_multihashmap_destroy (plugin->sessions);
   plugin->sessions = NULL;
+  GNUNET_CONTAINER_multihashmap_iterate (plugin->inbound_sessions, &destroy_inbound_session,
+                                         NULL);
+  GNUNET_CONTAINER_multihashmap_destroy (plugin->inbound_sessions);
+  plugin->inbound_sessions = NULL;
   while (NULL != (rc = GNUNET_CONTAINER_heap_remove_root (plugin->defrags)))
   {
     GNUNET_DEFRAGMENT_context_destroy (rc->defrag);