Imported Upstream version 0.9.3upstream/0.9.3

1 files changed, 442 insertions, 130 deletions

diff --git a/src/transport/plugin_transport_unix.c b/src/transport/plugin_transport_unix.c
index 499cc23..057479d 100644
--- a/src/transport/plugin_transport_unix.c
+++ b/src/transport/plugin_transport_unix.c
@@ -42,9 +42,6 @@
 #include "gnunet_transport_plugin.h"
 #include "transport.h"
 
-#define DEBUG_UNIX GNUNET_EXTRALOGGING
-#define DETAILS GNUNET_NO
-
 #define MAX_PROBES 20
 
 /*
@@ -88,6 +85,13 @@ struct Session
   void *addr;
   size_t addrlen;
   struct GNUNET_PeerIdentity target;
+
+  /**
+   * Session timeout task
+   */
+  GNUNET_SCHEDULER_TaskIdentifier timeout_task;
+
+  struct Plugin * plugin;
 };
 
 struct UNIXMessageWrapper
@@ -188,12 +192,7 @@ struct Plugin
    */
   struct GNUNET_TRANSPORT_PluginEnvironment *env;
 
-  /*
-   * Session of peers with whom we are currently connected
-   */
-  struct PeerSession *sessions;
-
-  /*
+  /**
    * Sessions
    */
   struct GNUNET_CONTAINER_MultiHashMap *session_map;
@@ -242,30 +241,167 @@ struct Plugin
    * ATS network
    */
   struct GNUNET_ATS_Information ats_network;
+
+  unsigned int bytes_in_queue;
+  unsigned int bytes_in_sent;
+  unsigned int bytes_in_recv;
+  unsigned int bytes_discarded;
 };
 
+/**
+ * Start session timeout
+ */
+static void
+start_session_timeout (struct Session *s);
 
-static int
-get_session_delete_it (void *cls, const GNUNET_HashCode * key, void *value)
+/**
+ * Increment session timeout due to activity
+ */
+static void
+reschedule_session_timeout (struct Session *s);
+
+/**
+ * Cancel timeout
+ */
+static void
+stop_session_timeout (struct Session *s);
+
+
+static void
+unix_plugin_select (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
+
+
+static void
+reschedule_select (struct Plugin * plugin)
 {
-  struct Session *s = value;
-  struct Plugin *plugin = cls;
-  GNUNET_assert (plugin != NULL);
 
-#if DEBUG_UNIX
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Deleting session for peer `%s' `%s' \n", GNUNET_i2s (&s->target), s->addr);
-#endif
+  if (plugin->select_task != GNUNET_SCHEDULER_NO_TASK)
+  {
+    GNUNET_SCHEDULER_cancel (plugin->select_task);
+    plugin->select_task = GNUNET_SCHEDULER_NO_TASK;
+  }
+
+  if (NULL != plugin->msg_head)
+  {
+    plugin->select_task =
+      GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_DEFAULT,
+                                   GNUNET_TIME_UNIT_FOREVER_REL,
+                                   plugin->rs,
+                                   plugin->ws,
+                                   &unix_plugin_select, plugin);
+    plugin->with_ws = GNUNET_YES;
+  }
+  else
+  {
+    plugin->select_task =
+      GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_DEFAULT,
+                                   GNUNET_TIME_UNIT_FOREVER_REL,
+                                   plugin->rs,
+                                   NULL,
+                                   &unix_plugin_select, plugin);
+    plugin->with_ws = GNUNET_NO;
+  }
+}
+
+struct LookupCtx
+{
+  struct Session *s;
+  const struct sockaddr_un *addr;
+};
+
+int lookup_session_it (void *cls,
+                       const GNUNET_HashCode * key,
+                       void *value)
+{
+  struct LookupCtx *lctx = cls;
+  struct Session *t = value;
+
+  if (0 == strcmp (t->addr, lctx->addr->sun_path))
+  {
+    lctx->s = t;
+    return GNUNET_NO;
+  }
+  return GNUNET_YES;
+}
+
 
+static struct Session *
+lookup_session (struct Plugin *plugin, struct GNUNET_PeerIdentity *sender, const struct sockaddr_un *addr)
+{
+  struct LookupCtx lctx;
+
+  GNUNET_assert (NULL != plugin);
+  GNUNET_assert (NULL != sender);
+  GNUNET_assert (NULL != addr);
+
+  lctx.s = NULL;
+  lctx.addr = addr;
+
+  GNUNET_CONTAINER_multihashmap_get_multiple (plugin->session_map, &sender->hashPubKey, &lookup_session_it, &lctx);
+
+  return lctx.s;
+}
+
+/**
+ * Functions with this signature are called whenever we need
+ * to close a session due to a disconnect or failure to
+ * establish a connection.
+ *
+ * @param s session to close down
+ */
+static void
+disconnect_session (struct Session *s)
+{
+  struct UNIXMessageWrapper *msgw;
+  struct UNIXMessageWrapper *next;
+  struct Plugin * plugin = s->plugin;
+  int removed;
+  GNUNET_assert (plugin != NULL);
+  GNUNET_assert (s != NULL);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Disconnecting session for peer `%s' `%s' \n", GNUNET_i2s (&s->target), s->addr);
+  stop_session_timeout (s);
   plugin->env->session_end (plugin->env->cls, &s->target, s);
 
+  msgw = plugin->msg_head;
+  removed = GNUNET_NO;
+  next = plugin->msg_head;
+  while (NULL != next)
+  {
+    msgw = next;
+    next = msgw->next;
+    if (msgw->session != s)
+      continue;
+    GNUNET_CONTAINER_DLL_remove (plugin->msg_head, plugin->msg_tail, msgw);
+    if (NULL != msgw->cont)
+      msgw->cont (msgw->cont_cls,  &msgw->session->target, GNUNET_SYSERR);
+    GNUNET_free (msgw->msg);
+    GNUNET_free (msgw);
+    removed = GNUNET_YES;    
+  }
+  if ((GNUNET_YES == removed) && (NULL == plugin->msg_head))
+    reschedule_select (plugin);
+
   GNUNET_assert (GNUNET_YES ==
-		 GNUNET_CONTAINER_multihashmap_remove(plugin->session_map, &s->target.hashPubKey, s));
+                 GNUNET_CONTAINER_multihashmap_remove(plugin->session_map, &s->target.hashPubKey, s));
+
+  GNUNET_STATISTICS_set(plugin->env->stats,
+                        "# UNIX sessions active",
+                        GNUNET_CONTAINER_multihashmap_size(plugin->session_map),
+                        GNUNET_NO);
 
   GNUNET_free (s);
+}
 
+static int
+get_session_delete_it (void *cls, const GNUNET_HashCode * key, void *value)
+{
+  struct Session *s = value;
+  disconnect_session (s);
   return GNUNET_YES;
 }
 
+
 /**
  * Disconnect from a remote node.  Clean up session if we have one for this peer
  *
@@ -273,7 +409,7 @@ get_session_delete_it (void *cls, const GNUNET_HashCode * key, void *value)
  * @param target the peeridentity of the peer to disconnect
  * @return GNUNET_OK on success, GNUNET_SYSERR if the operation failed
  */
-void
+static void
 unix_disconnect (void *cls, const struct GNUNET_PeerIdentity *target)
 {
   struct Plugin *plugin = cls;
@@ -314,28 +450,14 @@ unix_transport_server_stop (void *cls)
     plugin->select_task = GNUNET_SCHEDULER_NO_TASK;
   }
 
-  GNUNET_break (GNUNET_OK ==
-                GNUNET_NETWORK_socket_close (plugin->unix_sock.desc));
-  plugin->unix_sock.desc = NULL;
-  plugin->with_ws = GNUNET_NO;
-  return GNUNET_OK;
-}
-
-
-struct PeerSession *
-find_session (struct Plugin *plugin, const struct GNUNET_PeerIdentity *peer)
-{
-  struct PeerSession *pos;
-
-  pos = plugin->sessions;
-  while (pos != NULL)
+  if (NULL != plugin->unix_sock.desc)
   {
-    if (memcmp (&pos->target, peer, sizeof (struct GNUNET_PeerIdentity)) == 0)
-      return pos;
-    pos = pos->next;
+    GNUNET_break (GNUNET_OK ==
+                  GNUNET_NETWORK_socket_close (plugin->unix_sock.desc));
+    plugin->unix_sock.desc = NULL;
+    plugin->with_ws = GNUNET_NO;
   }
-
-  return pos;
+  return GNUNET_OK;
 }
 
 
@@ -369,7 +491,7 @@ unix_real_send (void *cls,
                 size_t addrlen, GNUNET_TRANSPORT_TransmitContinuation cont,
                 void *cont_cls)
 {
-
+  struct Plugin *plugin = cls;
   ssize_t sent;
   const void *sb;
   size_t sbs;
@@ -377,34 +499,31 @@ unix_real_send (void *cls,
   size_t slen;
   int retry;
 
+  GNUNET_assert (NULL != plugin);
+
   if (send_handle == NULL)
   {
-#if DEBUG_UNIX
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                "unix_real_send with send_handle NULL!\n");
-#endif
-    /* failed to open send socket for AF */
+    /* We do not have a send handle */
+    GNUNET_break (0);
     if (cont != NULL)
       cont (cont_cls, target, GNUNET_SYSERR);
-    return 0;
+    return -1;
   }
   if ((addr == NULL) || (addrlen == 0))
   {
-#if DEBUG_UNIX
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                "unix_real_send called without address, returning!\n");
-#endif
+    /* Can never send if we don't have an address */
+    GNUNET_break (0);
     if (cont != NULL)
       cont (cont_cls, target, GNUNET_SYSERR);
-    return 0;                   /* Can never send if we don't have an address!! */
+    return -1;
   }
 
+  /* Prepare address */
   memset (&un, 0, sizeof (un));
   un.sun_family = AF_UNIX;
   slen = strlen (addr) + 1;
   if (slen >= sizeof (un.sun_path))
     slen = sizeof (un.sun_path) - 1;
-  sent = 0;
   GNUNET_assert (slen < sizeof (un.sun_path));
   memcpy (un.sun_path, addr, slen);
   un.sun_path[slen] = '\0';
@@ -417,11 +536,17 @@ unix_real_send (void *cls,
 #endif
   sb = (struct sockaddr *) &un;
   sbs = slen;
+
+  /* Send the data */
+  sent = 0;
   retry = GNUNET_NO;
   sent = GNUNET_NETWORK_socket_sendto (send_handle, msgbuf, msgbuf_size, sb, sbs);
 
   if ((GNUNET_SYSERR == sent) && ((errno == EAGAIN) || (errno == ENOBUFS)))
-    retry = GNUNET_YES;
+  {
+    /* We have to retry later: retry */
+    return 0;
+  }
 
   if ((GNUNET_SYSERR == sent) && (errno == EMSGSIZE))
   {
@@ -436,23 +561,38 @@ unix_real_send (void *cls,
     {
       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                   "Trying to increase socket buffer size from %i to %i for message size %i\n",
-                  size, ((msgbuf_size / 1000) + 2) * 1000, msgbuf_size);
+                  size,
+                  ((msgbuf_size / 1000) + 2) * 1000,
+                  msgbuf_size);
       size = ((msgbuf_size / 1000) + 2) * 1000;
       if (GNUNET_NETWORK_socket_setsockopt
           ((struct GNUNET_NETWORK_Handle *) send_handle, SOL_SOCKET, SO_SNDBUF,
            &size, sizeof (size)) == GNUNET_OK)
-        retry = GNUNET_YES;
+      {
+        /* Increased buffer size, retry sending */
+        return 0;
+      }
       else
+      {
+        /* Could not increase buffer size: error, no retry */
         GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR, "setsockopt");
+        return -1;
+      }
+    }
+    else
+    {
+      /* Buffer is bigger than message:  error, no retry
+       * This should never happen!*/
+      GNUNET_break (0);
+      return -1;
     }
   }
 
-#if DEBUG_UNIX
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "UNIX transmit %u-byte message to %s (%d: %s)\n",
               (unsigned int) msgbuf_size, GNUNET_a2s (sb, sbs), (int) sent,
               (sent < 0) ? STRERROR (errno) : "ok");
-#endif
+
   /* Calling continuation */
   if (cont != NULL)
   {
@@ -465,14 +605,20 @@ unix_real_send (void *cls,
   /* return number of bytes successfully sent */
   if (sent > 0)
     return sent;
+  if (sent == 0)
+  {
+    /* That should never happen */
+    GNUNET_break (0);
+    return -1;
+  }
   /* failed and retry: return 0 */
   if ((GNUNET_SYSERR == sent) && (retry == GNUNET_YES))
     return 0;
   /* failed and no retry: return -1 */
   if ((GNUNET_SYSERR == sent) && (retry == GNUNET_NO))
     return -1;
-
-  return sent;
+  /* default */
+  return -1;
 }
 
 struct gsi_ctx
@@ -482,15 +628,14 @@ struct gsi_ctx
   struct Session *res;
 };
 
+
 static int
 get_session_it (void *cls, const GNUNET_HashCode * key, void *value)
 {
   struct gsi_ctx *gsi = cls;
   struct Session *s = value;
 
-#if DEBUG_UNIX
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Comparing session %s %s\n", gsi->address, s->addr);
-#endif
   if ((gsi->addrlen == s->addrlen) &&
       (0 == memcmp (gsi->address, s->addr, s->addrlen)))
   {
@@ -527,27 +672,29 @@ unix_plugin_get_session (void *cls,
   GNUNET_CONTAINER_multihashmap_get_multiple (plugin->session_map, &address->peer.hashPubKey, &get_session_it, &gsi);
   if (gsi.res != NULL)
   {
-#if DEBUG_UNIX
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Found existing session\n");
-#endif
     return gsi.res;
   }
 
   /* Create a new session */
-
   s = GNUNET_malloc (sizeof (struct Session) + address->address_length);
   s->addr = &s[1];
   s->addrlen = address->address_length;
+  s->plugin = plugin;
   memcpy(s->addr, address->address, s->addrlen);
   memcpy(&s->target, &address->peer, sizeof (struct GNUNET_PeerIdentity));
 
+  start_session_timeout (s);
+
   GNUNET_CONTAINER_multihashmap_put (plugin->session_map,
       &address->peer.hashPubKey, s,
       GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
-#if DEBUG_UNIX
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Creating new session\n");
-#endif
 
+  GNUNET_STATISTICS_set(plugin->env->stats,
+                        "# UNIX sessions active",
+                        GNUNET_CONTAINER_multihashmap_size(plugin->session_map),
+                        GNUNET_NO);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Creating new session\n");
   return s;
 }
 
@@ -609,7 +756,11 @@ unix_plugin_send (void *cls,
   if (GNUNET_OK != GNUNET_CONTAINER_multihashmap_contains_value(plugin->session_map,
       &session->target.hashPubKey, session))
   {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Invalid session for peer `%s' `%s'\n",
+                GNUNET_i2s (&session->target),
+                (char *) session->addr);
     GNUNET_break (0);
+
     return GNUNET_SYSERR;
   }
 
@@ -621,6 +772,8 @@ unix_plugin_send (void *cls,
           sizeof (struct GNUNET_PeerIdentity));
   memcpy (&message[1], msgbuf, msgbuf_size);
 
+  reschedule_session_timeout (session);
+
   wrapper = GNUNET_malloc (sizeof (struct UNIXMessageWrapper));
   wrapper->msg = message;
   wrapper->msgsize = ssize;
@@ -632,24 +785,15 @@ unix_plugin_send (void *cls,
 
   GNUNET_CONTAINER_DLL_insert(plugin->msg_head, plugin->msg_tail, wrapper);
 
-#if DEBUG_UNIX
+  plugin->bytes_in_queue += ssize;
+  GNUNET_STATISTICS_set (plugin->env->stats,"# UNIX bytes in send queue",
+      plugin->bytes_in_queue, GNUNET_NO);
+
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sent %d bytes to `%s'\n", ssize,
               (char *) session->addr);
-#endif
-
   if (plugin->with_ws == GNUNET_NO)
   {
-    if (plugin->select_task != GNUNET_SCHEDULER_NO_TASK)
-      GNUNET_SCHEDULER_cancel(plugin->select_task);
-
-    plugin->select_task =
-        GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_DEFAULT,
-                                     GNUNET_SCHEDULER_NO_TASK,
-                                     GNUNET_TIME_UNIT_FOREVER_REL,
-                                     plugin->rs,
-                                     plugin->ws,
-                                     &unix_plugin_select, plugin);
-    plugin->with_ws = GNUNET_YES;
+    reschedule_select (plugin);
   }
   return ssize;
 }
@@ -670,6 +814,8 @@ unix_demultiplexer (struct Plugin *plugin, struct GNUNET_PeerIdentity *sender,
                     const struct sockaddr_un *un, size_t fromlen)
 {
   struct GNUNET_ATS_Information ats[2];
+  struct Session *s = NULL;
+  struct GNUNET_HELLO_Address * addr;
 
   ats[0].type = htonl (GNUNET_ATS_QUALITY_NET_DISTANCE);
   ats[0].value = htonl (UNIX_DIRECT_DISTANCE);
@@ -678,20 +824,30 @@ unix_demultiplexer (struct Plugin *plugin, struct GNUNET_PeerIdentity *sender,
 
   GNUNET_assert (fromlen >= sizeof (struct sockaddr_un));
 
-#if DEBUG_UNIX
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Received message from %s\n",
               un->sun_path);
-#endif
+
+  plugin->bytes_in_recv += ntohs(currhdr->size);
+  GNUNET_STATISTICS_set (plugin->env->stats,"# UNIX bytes received",
+      plugin->bytes_in_recv, GNUNET_NO);
+
+  addr = GNUNET_HELLO_address_allocate(sender, "unix", un->sun_path, strlen (un->sun_path) + 1);
+  s = lookup_session (plugin, sender, un);
+  if (NULL == s)
+    s = unix_plugin_get_session (plugin, addr);
+  reschedule_session_timeout (s);
+
   plugin->env->receive (plugin->env->cls, sender, currhdr,
                         (const struct GNUNET_ATS_Information *) &ats, 2,
-                        NULL, un->sun_path, strlen (un->sun_path) + 1);
+                        s, un->sun_path, strlen (un->sun_path) + 1);
+  GNUNET_free (addr);
 }
 
 
 static void
 unix_plugin_select_read (struct Plugin * plugin)
 {
-  char buf[65536];
+  char buf[65536] GNUNET_ALIGN;
   struct UNIXMessage *msg;
   struct GNUNET_PeerIdentity sender;
   struct sockaddr_un un;
@@ -723,10 +879,8 @@ unix_plugin_select_read (struct Plugin * plugin)
 #if LINUX
     un.sun_path[0] = '/';
 #endif
-#if DEBUG_UNIX
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Read %d bytes from socket %s\n", ret,
                 &un.sun_path[0]);
-#endif
   }
 
   GNUNET_assert (AF_UNIX == (un.sun_family));
@@ -752,14 +906,17 @@ unix_plugin_select_read (struct Plugin * plugin)
       GNUNET_break_op (0);
       break;
     }
+
     unix_demultiplexer (plugin, &sender, currhdr, &un, sizeof (un));
     offset += csize;
   }
 }
 
+
 static void
 unix_plugin_select_write (struct Plugin * plugin)
 {
+  static int retry_counter;
   int sent = 0;
   struct UNIXMessageWrapper * msgw = plugin->msg_head;
 
@@ -774,37 +931,69 @@ unix_plugin_select_write (struct Plugin * plugin)
                          msgw->session->addrlen,
                          msgw->cont, msgw->cont_cls);
 
-  /* successfully sent bytes */
-  if (sent > 0)
+  if (sent == 0)
+  {
+    /* failed and retry */
+    retry_counter++;
+    GNUNET_STATISTICS_set (plugin->env->stats,"# UNIX retry attempt",
+        retry_counter, GNUNET_NO);
+    return;
+  }
+
+  if (retry_counter > 0 )
+  {
+    /* no retry: reset counter */
+    retry_counter = 0;
+    GNUNET_STATISTICS_set (plugin->env->stats,"# UNIX retry attempt",
+        retry_counter, GNUNET_NO);
+  }
+
+  if (sent == -1)
   {
+    /* failed and no retry */
     GNUNET_CONTAINER_DLL_remove(plugin->msg_head, plugin->msg_tail, msgw);
+
+    GNUNET_assert (plugin->bytes_in_queue >= msgw->msgsize);
+    plugin->bytes_in_queue -= msgw->msgsize;
+    GNUNET_STATISTICS_set (plugin->env->stats,"# UNIX bytes in send queue",
+        plugin->bytes_in_queue, GNUNET_NO);
+    plugin->bytes_discarded += msgw->msgsize;
+    GNUNET_STATISTICS_set (plugin->env->stats,"# UNIX bytes discarded",
+        plugin->bytes_discarded, GNUNET_NO);
+
     GNUNET_free (msgw->msg);
     GNUNET_free (msgw);
     return;
   }
 
-  /* failed and no retry */
-  if (sent == -1)
+  if (sent > 0)
   {
+    /* successfully sent bytes */
     GNUNET_CONTAINER_DLL_remove(plugin->msg_head, plugin->msg_tail, msgw);
+
+    GNUNET_assert (plugin->bytes_in_queue >= msgw->msgsize);
+    plugin->bytes_in_queue -= msgw->msgsize;
+    GNUNET_STATISTICS_set (plugin->env->stats,"# UNIX bytes in send queue",
+        plugin->bytes_in_queue, GNUNET_NO);
+    plugin->bytes_in_sent += msgw->msgsize;
+    GNUNET_STATISTICS_set (plugin->env->stats,"# UNIX bytes sent",
+        plugin->bytes_in_sent, GNUNET_NO);
+
     GNUNET_free (msgw->msg);
     GNUNET_free (msgw);
     return;
   }
 
-  /* failed and retry */
-  if (sent == 0)
-    return;
 }
 
-/*
- * @param cls the plugin handle
- * @param tc the scheduling context (for rescheduling this function again)
- *
+
+/**
  * We have been notified that our writeset has something to read.  We don't
  * know which socket needs to be read, so we have to check each one
  * Then reschedule this function to be called again once more is available.
  *
+ * @param cls the plugin handle
+ * @param tc the scheduling context (for rescheduling this function again)
  */
 static void
 unix_plugin_select (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
@@ -815,9 +1004,9 @@ unix_plugin_select (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
   if ((tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN) != 0)
     return;
 
-  plugin->with_ws = GNUNET_NO;
   if ((tc->reason & GNUNET_SCHEDULER_REASON_WRITE_READY) != 0)
   {
+    /* Ready to send data */
     GNUNET_assert (GNUNET_NETWORK_fdset_isset
                    (tc->write_ready, plugin->unix_sock.desc));
     if (plugin->msg_head != NULL)
@@ -826,30 +1015,22 @@ unix_plugin_select (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 
   if ((tc->reason & GNUNET_SCHEDULER_REASON_READ_READY) != 0)
   {
+    /* Ready to receive data */
     GNUNET_assert (GNUNET_NETWORK_fdset_isset
                    (tc->read_ready, plugin->unix_sock.desc));
     unix_plugin_select_read (plugin);
   }
 
-  if (plugin->select_task != GNUNET_SCHEDULER_NO_TASK)
-    GNUNET_SCHEDULER_cancel (plugin->select_task);
-  plugin->select_task =
-      GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_DEFAULT,
-                                   GNUNET_SCHEDULER_NO_TASK,
-                                   GNUNET_TIME_UNIT_FOREVER_REL,
-                                   plugin->rs,
-                                   (plugin->msg_head != NULL) ? plugin->ws : NULL,
-                                   &unix_plugin_select, plugin);
-  if (plugin->msg_head != NULL)
-    plugin->with_ws = GNUNET_YES;
+  reschedule_select (plugin);
 }
 
+
 /**
  * Create a slew of UNIX sockets.  If possible, use IPv6 and IPv4.
  *
  * @param cls closure for server start, should be a struct Plugin *
  * @return number of sockets created or GNUNET_SYSERR on error
-*/
+ */
 static int
 unix_transport_server_start (void *cls)
 {
@@ -893,10 +1074,8 @@ unix_transport_server_start (void *cls)
     plugin->unix_sock.desc = NULL;
     return GNUNET_SYSERR;
   }
-#if DEBUG_UNIX
   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "unix", "Bound to `%s'\n",
                    &un.sun_path[0]);
-#endif
   plugin->rs = GNUNET_NETWORK_fdset_create ();
   plugin->ws = GNUNET_NETWORK_fdset_create ();
   GNUNET_NETWORK_fdset_zero (plugin->rs);
@@ -904,14 +1083,7 @@ unix_transport_server_start (void *cls)
   GNUNET_NETWORK_fdset_set (plugin->rs, plugin->unix_sock.desc);
   GNUNET_NETWORK_fdset_set (plugin->ws, plugin->unix_sock.desc);
 
-  plugin->select_task =
-      GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_DEFAULT,
-                                   GNUNET_SCHEDULER_NO_TASK,
-                                   GNUNET_TIME_UNIT_FOREVER_REL,
-                                   plugin->rs,
-                                   NULL,
-                                   &unix_plugin_select, plugin);
-  plugin->with_ws = GNUNET_NO;
+  reschedule_select (plugin);
 
   return 1;
 }
@@ -936,12 +1108,9 @@ unix_transport_server_start (void *cls)
 static int
 unix_check_address (void *cls, const void *addr, size_t addrlen)
 {
-
-#if DEBUG_UNIX
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "Informing transport service about my address `%s'\n",
               (char *) addr);
-#endif
   return GNUNET_OK;
 }
 
@@ -968,16 +1137,59 @@ unix_plugin_address_pretty_printer (void *cls, const char *type,
                                     GNUNET_TRANSPORT_AddressStringCallback asc,
                                     void *asc_cls)
 {
-  if ((addr != NULL) && (addrlen > 0))
+  if ((NULL != addr) && (addrlen > 0))
+  {
     asc (asc_cls, (const char *) addr);
+  }
   else
   {
     GNUNET_break (0);
-    asc (asc_cls, "Invalid UNIX address");
+    asc (asc_cls, "<invalid UNIX address>");
   }
+  asc (asc_cls, NULL);
+}
+
 
+/**
+ * Function called to convert a string address to
+ * a binary address.
+ *
+ * @param cls closure ('struct Plugin*')
+ * @param addr string address
+ * @param addrlen length of the address (strlen(addr) + '\0')
+ * @param buf location to store the buffer
+ *        If the function returns GNUNET_SYSERR, its contents are undefined.
+ * @param added length of created address
+ * @return GNUNET_OK on success, GNUNET_SYSERR on failure
+ */
+static int
+unix_string_to_address (void *cls, const char *addr, uint16_t addrlen,
+    void **buf, size_t *added)
+{
+  if ((NULL == addr) || (0 == addrlen))
+  {
+    GNUNET_break (0);
+    return GNUNET_SYSERR;
+  }
+
+  if ('\0' != addr[addrlen - 1])
+  {
+    GNUNET_break (0);
+    return GNUNET_SYSERR;
+  }
+
+  if (strlen (addr) != addrlen - 1)
+  {
+    GNUNET_break (0);
+    return GNUNET_SYSERR;
+  }
+
+  (*buf) = strdup (addr);
+  (*added) = strlen (addr) + 1;
+  return GNUNET_OK;
 }
 
+
 /**
  * Function called for a quick conversion of the binary address to
  * a numeric address.  Note that the caller must not free the
@@ -994,10 +1206,10 @@ unix_address_to_string (void *cls, const void *addr, size_t addrlen)
 {
   if ((addr != NULL) && (addrlen > 0))
     return (const char *) addr;
-  else
-    return NULL;
+  return NULL;
 }
 
+
 /**
  * Notify transport service about address
  *
@@ -1014,6 +1226,85 @@ address_notification (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
                                strlen (plugin->unix_socket_path) + 1);
 }
 
+
+/**
+ * Session was idle, so disconnect it
+ */
+static void
+session_timeout (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  GNUNET_assert (NULL != cls);
+  struct Session *s = cls;
+
+  s->timeout_task = GNUNET_SCHEDULER_NO_TASK;
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Session %p was idle for %llu, disconnecting\n",
+      s, GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT.rel_value);
+
+  /* call session destroy function */
+  disconnect_session(s);
+
+}
+
+/**
+ * Start session timeout
+ */
+static void
+start_session_timeout (struct Session *s)
+{
+  GNUNET_assert (NULL != s);
+  GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == s->timeout_task);
+
+  s->timeout_task =  GNUNET_SCHEDULER_add_delayed (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT,
+                                                   &session_timeout,
+                                                   s);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Timeout for session %p set to %llu\n",
+      s, GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT.rel_value);
+}
+
+/**
+ * Increment session timeout due to activity
+ */
+static void
+reschedule_session_timeout (struct Session *s)
+{
+  GNUNET_assert (NULL != s);
+  GNUNET_assert (GNUNET_SCHEDULER_NO_TASK != s->timeout_task);
+
+  GNUNET_SCHEDULER_cancel (s->timeout_task);
+  s->timeout_task =  GNUNET_SCHEDULER_add_delayed (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT,
+                                                   &session_timeout,
+                                                   s);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Timeout rescheduled for session %p set to %llu\n",
+      s, GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT.rel_value);
+}
+
+/**
+ * Cancel timeout
+ */
+static void
+stop_session_timeout (struct Session *s)
+{
+  GNUNET_assert (NULL != s);
+
+  if (GNUNET_SCHEDULER_NO_TASK != s->timeout_task)
+  {
+    GNUNET_SCHEDULER_cancel (s->timeout_task);
+    s->timeout_task = GNUNET_SCHEDULER_NO_TASK;
+
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Timeout rescheduled for session %p canceled\n",
+      s, GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT.rel_value);
+  }
+  else
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Timeout for session %p was not active\n",
+      s);
+  }
+}
+
+
 /**
  * The exported method. Makes the core api available via a global and
  * returns the unix transport API.
@@ -1027,6 +1318,19 @@ libgnunet_plugin_transport_unix_init (void *cls)
   struct Plugin *plugin;
   int sockets_created;
 
+  if (NULL == env->receive)
+  {
+    /* run in 'stub' mode (i.e. as part of gnunet-peerinfo), don't fully
+       initialze the plugin or the API */
+    api = GNUNET_malloc (sizeof (struct GNUNET_TRANSPORT_PluginFunctions));
+    api->cls = NULL;
+    api->address_pretty_printer = &unix_plugin_address_pretty_printer;
+    api->address_to_string = &unix_address_to_string;
+    api->string_to_address = &unix_string_to_address;
+    return api;
+  }
+  GNUNET_assert( NULL != env->stats);
+
   if (GNUNET_OK !=
       GNUNET_CONFIGURATION_get_value_number (env->cfg, "transport-unix", "PORT",
                                              &port))
@@ -1046,6 +1350,7 @@ libgnunet_plugin_transport_unix_init (void *cls)
   api->address_pretty_printer = &unix_plugin_address_pretty_printer;
   api->address_to_string = &unix_address_to_string;
   api->check_address = &unix_check_address;
+  api->string_to_address = &unix_string_to_address;
   sockets_created = unix_transport_server_start (plugin);
   if (sockets_created == 0)
     GNUNET_log (GNUNET_ERROR_TYPE_WARNING, _("Failed to open UNIX sockets\n"));
@@ -1062,13 +1367,20 @@ libgnunet_plugin_transport_unix_done (void *cls)
   struct GNUNET_TRANSPORT_PluginFunctions *api = cls;
   struct Plugin *plugin = api->cls;
 
+  if (NULL == plugin)
+  {
+    GNUNET_free (api);
+    return NULL;
+  }
   unix_transport_server_stop (plugin);
 
   GNUNET_CONTAINER_multihashmap_iterate (plugin->session_map, &get_session_delete_it, plugin);
   GNUNET_CONTAINER_multihashmap_destroy (plugin->session_map);
 
-  GNUNET_NETWORK_fdset_destroy (plugin->rs);
-  GNUNET_NETWORK_fdset_destroy (plugin->ws);
+  if (NULL != plugin->rs)
+    GNUNET_NETWORK_fdset_destroy (plugin->rs);
+  if (NULL != plugin->ws)
+    GNUNET_NETWORK_fdset_destroy (plugin->ws);
   GNUNET_free (plugin->unix_socket_path);
   GNUNET_free (plugin);
   GNUNET_free (api);