PSYC: in-order delivery of fragments; tests for large messages

Cache message received fragments from multicast and deliver them
in the correct order to clients.

Test messages with large modifier and data payloads.

5 files changed, 390 insertions, 63 deletions

diff --git a/src/psyc/gnunet-service-psyc.c b/src/psyc/gnunet-service-psyc.c
index dcb5031f1..4cdb490a1 100644
--- a/src/psyc/gnunet-service-psyc.c
+++ b/src/psyc/gnunet-service-psyc.c
@@ -24,6 +24,8 @@
  * @author Gabor X Toth
  */
 
+#include <inttypes.h>
+
 #include "platform.h"
 #include "gnunet_util_lib.h"
 #include "gnunet_constants.h"
@@ -77,6 +79,45 @@ struct TransmitMessage
   uint8_t state;
 };
 
+
+/**
+ * Cache for received message fragments.
+ * Message fragments are only sent to clients after all modifiers arrived.
+ *
+ * chan_key -> MultiHashMap chan_msgs
+ */
+static struct GNUNET_CONTAINER_MultiHashMap *recv_cache;
+
+
+/**
+ * Entry in the chan_msgs hashmap of @a recv_cache:
+ * fragment_id -> FragmentEntry
+ */
+struct FragmentEntry
+{
+  struct GNUNET_MULTICAST_MessageHeader *mmsg;
+  uint16_t ref_count;
+};
+
+
+/**
+ * Entry in the @a recv_msgs hash map of a @a Channel.
+ * message_id -> FragmentCache
+ */
+struct FragmentCache
+{
+  /**
+   * Total size of header fragments (METHOD & MODIFIERs)
+   */
+  uint64_t header_size;
+
+  /**
+   * Fragment IDs stored in @a recv_cache.
+   */
+  struct GNUNET_CONTAINER_Heap *fragments;
+};
+
+
 /**
  * Common part of the client context for both a master and slave channel.
  */
@@ -87,6 +128,12 @@ struct Channel
   struct TransmitMessage *tmit_head;
   struct TransmitMessage *tmit_tail;
 
+  /**
+   * Received fragments not yet sent to the client.
+   * message_id -> FragmentCache
+   */
+  struct GNUNET_CONTAINER_MultiHashMap *recv_msgs;
+
   GNUNET_SCHEDULER_TaskIdentifier tmit_task;
 
   /**
@@ -213,6 +260,8 @@ shutdown_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 static void
 client_cleanup (struct Channel *ch)
 {
+  /* FIXME: fragment_cache_clear */
+
   if (ch->is_master)
   {
     struct Master *mst = (struct Master *) ch;
@@ -323,6 +372,189 @@ fragment_store_result (void *cls, int64_t result, const char *err_msg)
 }
 
 
+static void
+message_to_client (struct Channel *ch,
+                   const struct GNUNET_MULTICAST_MessageHeader *mmsg)
+{
+  uint16_t size = ntohs (mmsg->header.size);
+  struct GNUNET_PSYC_MessageHeader *pmsg;
+  uint16_t psize = sizeof (*pmsg) + size - sizeof (*mmsg);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Sending message to client. "
+              "fragment_id: %" PRIu64 ", message_id: %" PRIu64 "\n",
+              ch, GNUNET_ntohll (mmsg->fragment_id),
+              GNUNET_ntohll (mmsg->message_id));
+
+  pmsg = GNUNET_malloc (psize);
+  pmsg->header.size = htons (psize);
+  pmsg->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE);
+  pmsg->message_id = mmsg->message_id;
+
+  memcpy (&pmsg[1], &mmsg[1], size - sizeof (*mmsg));
+
+  GNUNET_SERVER_notification_context_add (nc, ch->client);
+  GNUNET_SERVER_notification_context_unicast (nc, ch->client,
+                                              (const struct GNUNET_MessageHeader *) pmsg,
+                                              GNUNET_NO);
+  GNUNET_free (pmsg);
+}
+
+
+/**
+ * Convert an uint64_t in network byte order to a HashCode
+ * that can be used as key in a MultiHashMap
+ */
+static inline void
+hash_key_from_nll (struct GNUNET_HashCode *key, uint64_t n)
+{
+  /* use little-endian order, as idx_of MultiHashMap casts key to unsigned int */
+
+  n = ((n <<  8) & 0xFF00FF00FF00FF00ULL) | ((n >>  8) & 0x00FF00FF00FF00FFULL);
+  n = ((n << 16) & 0xFFFF0000FFFF0000ULL) | ((n >> 16) & 0x0000FFFF0000FFFFULL);
+
+  *key = (struct GNUNET_HashCode) {{ 0 }};
+  *((uint64_t *) key)
+    = (n << 32) | (n >> 32);
+}
+
+
+/**
+ * Convert an uint64_t in host byte order to a HashCode
+ * that can be used as key in a MultiHashMap
+ */
+static inline void
+hash_key_from_hll (struct GNUNET_HashCode *key, uint64_t n)
+{
+#if __BYTE_ORDER == __BIG_ENDIAN
+  hash_key_from_nll (key, n);
+#elif __BYTE_ORDER == __LITTLE_ENDIAN
+  *key = (struct GNUNET_HashCode) {{ 0 }};
+  *((uint64_t *) key) = n;
+#else
+  #error byteorder undefined
+#endif
+}
+
+
+static void
+fragment_cache_insert (struct Channel *ch,
+                       const struct GNUNET_HashCode *chan_key_hash,
+                       const struct GNUNET_HashCode *msg_id,
+                       struct FragmentCache *frag_cache,
+                       const struct GNUNET_MULTICAST_MessageHeader *mmsg,
+                       uint16_t last_part_type)
+{
+  uint16_t size = ntohs (mmsg->header.size);
+  struct GNUNET_CONTAINER_MultiHashMap
+    *chan_msgs = GNUNET_CONTAINER_multihashmap_get (recv_cache, chan_key_hash);
+
+  if (NULL == frag_cache)
+  {
+    frag_cache = GNUNET_new (struct FragmentCache);
+    frag_cache->fragments
+      = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
+
+    if (NULL == ch->recv_msgs)
+    {
+      ch->recv_msgs = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
+    }
+    GNUNET_CONTAINER_multihashmap_put (ch->recv_msgs, msg_id, frag_cache,
+                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
+
+    if (NULL == chan_msgs)
+    {
+      chan_msgs = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
+      GNUNET_CONTAINER_multihashmap_put (recv_cache, chan_key_hash, chan_msgs,
+                                         GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
+    }
+  }
+
+  struct GNUNET_HashCode *frag_id = GNUNET_new (struct GNUNET_HashCode);
+  hash_key_from_nll (frag_id, mmsg->fragment_id);
+  struct FragmentEntry
+    *frag_entry = GNUNET_CONTAINER_multihashmap_get (chan_msgs, frag_id);
+  if (NULL == frag_entry)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "%p Adding message fragment to cache. "
+                "fragment_id: %" PRIu64 ", "
+                "header_size: %" PRIu64 " + %" PRIu64 ").\n",
+                ch, GNUNET_ntohll (mmsg->fragment_id),
+                frag_cache->header_size, size);
+    frag_entry = GNUNET_new (struct FragmentEntry);
+    frag_entry->ref_count = 1;
+    frag_entry->mmsg = GNUNET_malloc (size);
+    memcpy (frag_entry->mmsg, mmsg, size);
+    GNUNET_CONTAINER_multihashmap_put (chan_msgs, frag_id, frag_entry,
+                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
+  }
+  else
+  {
+    frag_entry->ref_count++;
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "%p Message fragment already in cache. "
+                "fragment_id: %" PRIu64 ", ref_count: %u\n",
+                ch, GNUNET_ntohll (mmsg->fragment_id), frag_entry->ref_count);
+  }
+
+  switch (last_part_type)
+  {
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD:
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER:
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT:
+    frag_cache->header_size += size;
+  }
+  GNUNET_CONTAINER_heap_insert (frag_cache->fragments, frag_id,
+                                GNUNET_ntohll (mmsg->fragment_id));
+}
+
+
+static void
+fragment_cache_clear (struct Channel *ch,
+                      const struct GNUNET_HashCode *chan_key_hash,
+                      const struct GNUNET_HashCode *msg_id,
+                      struct FragmentCache *frag_cache,
+                      uint8_t send_to_client)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Clearing message fragment cache.\n", ch);
+
+  struct GNUNET_CONTAINER_MultiHashMap
+    *chan_msgs = GNUNET_CONTAINER_multihashmap_get (recv_cache, chan_key_hash);
+  GNUNET_assert (NULL != chan_msgs);
+  struct GNUNET_HashCode *frag_id;
+
+  while ((frag_id = GNUNET_CONTAINER_heap_remove_root (frag_cache->fragments)))
+  {
+    struct FragmentEntry
+      *frag_entry = GNUNET_CONTAINER_multihashmap_get (chan_msgs, frag_id);
+    if (frag_entry != NULL)
+    {
+      if (GNUNET_YES == send_to_client)
+      {
+        message_to_client (ch, frag_entry->mmsg);
+      }
+      if (1 == frag_entry->ref_count)
+      {
+        GNUNET_CONTAINER_multihashmap_remove (chan_msgs, frag_id, frag_entry);
+        GNUNET_free (frag_entry->mmsg);
+        GNUNET_free (frag_entry);
+      }
+      else
+      {
+        frag_entry->ref_count--;
+      }
+    }
+    GNUNET_free (frag_id);
+  }
+
+  GNUNET_CONTAINER_multihashmap_remove (ch->recv_msgs, msg_id, frag_cache);
+  GNUNET_CONTAINER_heap_destroy (frag_cache->fragments);
+  GNUNET_free (frag_cache);
+}
+
+
 /**
  * Incoming message fragment from multicast.
  *
@@ -358,11 +590,15 @@ message_cb (struct Channel *ch,
                                    rcb, rcb_cls);
 #endif
 
-    const struct GNUNET_MULTICAST_MessageHeader *mmsg
-      = (const struct GNUNET_MULTICAST_MessageHeader *) msg;
+    const struct GNUNET_MULTICAST_MessageHeader
+      *mmsg = (const struct GNUNET_MULTICAST_MessageHeader *) msg;
 
-    if (GNUNET_YES != GNUNET_PSYC_check_message_parts (size - sizeof (*mmsg),
-                                                       (const char *) &mmsg[1]))
+    uint16_t ptype = GNUNET_PSYC_message_last_part (size - sizeof (*mmsg),
+                                                    (const char *) &mmsg[1]);
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "Last message part type %u\n", ptype);
+
+    if (GNUNET_NO == ptype)
     {
       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                   "%p Received message with invalid parts from multicast. "
@@ -371,20 +607,55 @@ message_cb (struct Channel *ch,
       break;
     }
 
-    struct GNUNET_PSYC_MessageHeader *pmsg;
-    uint16_t psize = sizeof (*pmsg) + size - sizeof (*mmsg);
-    pmsg = GNUNET_malloc (psize);
-    pmsg->header.size = htons (psize);
-    pmsg->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE);
-    pmsg->message_id = mmsg->message_id;
+    struct GNUNET_HashCode msg_id;
+    hash_key_from_nll (&msg_id, mmsg->message_id);
 
-    memcpy (&pmsg[1], &mmsg[1], size - sizeof (*mmsg));
+    struct FragmentCache *frag_cache = NULL;
+    if (NULL != ch->recv_msgs)
+      frag_cache = GNUNET_CONTAINER_multihashmap_get (ch->recv_msgs, &msg_id);
 
-    GNUNET_SERVER_notification_context_add (nc, ch->client);
-    GNUNET_SERVER_notification_context_unicast (nc, ch->client,
-                                                (const struct GNUNET_MessageHeader *) pmsg,
-                                                GNUNET_NO);
-    GNUNET_free (pmsg);
+    switch (ptype)
+    {
+    case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA:
+    case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END:
+      /* FIXME: check state flag / max_state_message_id */
+      if (NULL == frag_cache)
+      {
+        message_to_client (ch, mmsg);
+        break;
+      }
+      else
+      {
+        if (GNUNET_ntohll (mmsg->fragment_offset) == frag_cache->header_size)
+        { /* first data fragment after the header, send cached fragments */
+          fragment_cache_clear (ch, chan_key_hash, &msg_id, frag_cache, GNUNET_YES);
+          message_to_client (ch, mmsg);
+          break;
+        }
+        else
+        { /* still missing fragments from the header, cache data fragment */
+          /* fall thru */
+        }
+      }
+
+    case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD:
+    case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER:
+    case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT:
+      /* not all modifiers arrived yet, cache fragment */
+      fragment_cache_insert (ch, chan_key_hash, &msg_id, frag_cache, mmsg, ptype);
+      break;
+
+    case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL:
+      if (NULL != frag_cache)
+      { /* fragments not yet sent to client, remove from cache */
+        fragment_cache_clear (ch, chan_key_hash, &msg_id, frag_cache, GNUNET_NO);
+      }
+      else
+      {
+        message_to_client (ch, mmsg);
+      }
+      break;
+    }
     break;
   }
   default:
@@ -457,8 +728,9 @@ request_cb (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *member_key,
     const struct GNUNET_MULTICAST_RequestHeader *req
       = (const struct GNUNET_MULTICAST_RequestHeader *) msg;
 
-    if (GNUNET_YES != GNUNET_PSYC_check_message_parts (size - sizeof (*req),
-                                                       (const char *) &req[1]))
+    /* FIXME: see message_cb() */
+    if (GNUNET_NO == GNUNET_PSYC_message_last_part (size - sizeof (*req),
+                                                    (const char *) &req[1]))
     {
       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                   "%p Dropping message with invalid parts "
@@ -826,7 +1098,7 @@ handle_psyc_message (void *cls, struct GNUNET_SERVER_Client *client,
   if (GNUNET_YES != ch->ready)
   {
     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-                "%p Ignoring message from client, channel is not ready yet.\n",
+                "%p Dropping message from client, channel is not ready yet.\n",
                 ch);
     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
     return;
@@ -912,11 +1184,12 @@ run (void *cls, struct GNUNET_SERVER_Handle *server,
   store = GNUNET_PSYCSTORE_connect (cfg);
   stats = GNUNET_STATISTICS_create ("psyc", cfg);
   clients = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
+  recv_cache = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
   nc = GNUNET_SERVER_notification_context_create (server, 1);
   GNUNET_SERVER_add_handlers (server, handlers);
   GNUNET_SERVER_disconnect_notify (server, &client_disconnect, NULL);
-  GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL, &shutdown_task,
-                                NULL);
+  GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL,
+                                &shutdown_task, NULL);
 }
 
 
diff --git a/src/psyc/psyc.h b/src/psyc/psyc.h
index 1ffda5d08..940412a32 100644
--- a/src/psyc/psyc.h
+++ b/src/psyc/psyc.h
@@ -31,8 +31,8 @@
 #include "gnunet_psyc_service.h"
 
 
-int
-GNUNET_PSYC_check_message_parts (uint16_t data_size, const char *data);
+uint16_t
+GNUNET_PSYC_message_last_part (uint16_t data_size, const char *data);
 
 void
 GNUNET_PSYC_log_message (enum GNUNET_ErrorType kind,
diff --git a/src/psyc/psyc_api.c b/src/psyc/psyc_api.c
index 8a1c9ffaa..16e8106d4 100644
--- a/src/psyc/psyc_api.c
+++ b/src/psyc/psyc_api.c
@@ -336,7 +336,7 @@ recv_error (struct GNUNET_PSYC_Channel *ch)
 
 
 /**
- * Queue an incoming message part for transmission to the PSYC service.
+ * Queue a message part for transmission to the PSYC service.
  *
  * The message part is added to the current message buffer.
  * When this buffer is full, it is added to the transmission queue.
@@ -390,7 +390,7 @@ queue_message (struct GNUNET_PSYC_Channel *ch,
     op->msg->size = sizeof (*op->msg) + size;
     memcpy (&op->msg[1], msg, size);
   }
- 
+
   if (NULL != op
       && (GNUNET_YES == end
           || (GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD
@@ -433,12 +433,12 @@ channel_transmit_mod (struct GNUNET_PSYC_Channel *ch)
     max_data_size = data_size = GNUNET_PSYC_MODIFIER_MAX_PAYLOAD;
     msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER);
     msg->size = sizeof (struct GNUNET_PSYC_MessageModifier);
-    notify_ret = ch->tmit.notify_mod (ch->tmit.notify_cls,
-                                      &data_size, &mod[1], &mod->oper);
+    notify_ret = ch->tmit.notify_mod (ch->tmit.notify_cls, &data_size, &mod[1],
+                                      &mod->oper, &mod->value_size);
     mod->name_size = strnlen ((char *) &mod[1], data_size);
     if (mod->name_size < data_size)
     {
-      mod->value_size = htons (data_size - 1 - mod->name_size);
+      mod->value_size = htonl (mod->value_size);
       mod->name_size = htons (mod->name_size);
     }
     else if (0 < data_size)
@@ -451,10 +451,10 @@ channel_transmit_mod (struct GNUNET_PSYC_Channel *ch)
   case MSG_STATE_MOD_CONT:
   {
     max_data_size = data_size = GNUNET_PSYC_MOD_CONT_MAX_PAYLOAD;
-    msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT);    
+    msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT);
     msg->size = sizeof (struct GNUNET_MessageHeader);
     notify_ret = ch->tmit.notify_mod (ch->tmit.notify_cls,
-                                      &data_size, &msg[1], NULL);
+                                      &data_size, &msg[1], NULL, NULL);
     break;
   }
   default:
@@ -669,6 +669,8 @@ handle_psyc_message (struct GNUNET_PSYC_Channel *ch,
     ch->recv_message_id = GNUNET_ntohll (msg->message_id);
     ch->recv_flags = flags;
     ch->recv_slave_key = msg->slave_key;
+    ch->recv_mod_value_size = 0;
+    ch->recv_mod_value_size_expected = 0;
   }
   else if (GNUNET_ntohll (msg->message_id) != ch->recv_message_id)
   {
@@ -703,7 +705,7 @@ handle_psyc_message (struct GNUNET_PSYC_Channel *ch,
     if (psize < sizeof (*pmsg) || sizeof (*msg) + pos + psize > size)
     {
       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-                  "Discarding message of type %u with invalid size %u.\n",
+                  "Dropping message of type %u with invalid size %u.\n",
                   ptype, psize);
       recv_error (ch);
       return;
@@ -753,7 +755,8 @@ handle_psyc_message (struct GNUNET_PSYC_Channel *ch,
       if (MSG_STATE_START != ch->recv_state)
       {
         LOG (GNUNET_ERROR_TYPE_WARNING,
-             "Discarding out of order message method.\n");
+             "Dropping out of order message method (%u).\n",
+             ch->recv_state);
         /* It is normal to receive an incomplete message right after connecting,
          * but should not happen later.
          * FIXME: add a check for this condition.
@@ -766,7 +769,7 @@ handle_psyc_message (struct GNUNET_PSYC_Channel *ch,
       if ('\0' != *((char *) meth + psize - 1))
       {
         LOG (GNUNET_ERROR_TYPE_WARNING,
-             "Discarding message with malformed method. "
+             "Dropping message with malformed method. "
              "Message ID: %" PRIu64 "\n", ch->recv_message_id);
         GNUNET_break_op (0);
         recv_error (ch);
@@ -782,7 +785,8 @@ handle_psyc_message (struct GNUNET_PSYC_Channel *ch,
             || MSG_STATE_MOD_CONT == ch->recv_state))
       {
         LOG (GNUNET_ERROR_TYPE_WARNING,
-             "Discarding out of order message modifier.\n");
+             "Dropping out of order message modifier (%u).\n",
+             ch->recv_state);
         GNUNET_break_op (0);
         recv_error (ch);
         return;
@@ -792,14 +796,14 @@ handle_psyc_message (struct GNUNET_PSYC_Channel *ch,
         = (struct GNUNET_PSYC_MessageModifier *) pmsg;
 
       uint16_t name_size = ntohs (mod->name_size);
-      ch->recv_mod_value_size_expected = ntohs (mod->value_size);
+      ch->recv_mod_value_size_expected = ntohl (mod->value_size);
       ch->recv_mod_value_size = psize - sizeof (*mod) - name_size - 1;
 
       if (psize < sizeof (*mod) + name_size + 1
           || '\0' != *((char *) &mod[1] + name_size)
           || ch->recv_mod_value_size_expected < ch->recv_mod_value_size)
       {
-        LOG (GNUNET_ERROR_TYPE_WARNING, "Discarding malformed modifier.\n");
+        LOG (GNUNET_ERROR_TYPE_WARNING, "Dropping malformed modifier.\n");
         GNUNET_break_op (0);
         recv_error (ch);
         return;
@@ -816,7 +820,11 @@ handle_psyc_message (struct GNUNET_PSYC_Channel *ch,
           || ch->recv_mod_value_size_expected < ch->recv_mod_value_size)
       {
         LOG (GNUNET_ERROR_TYPE_WARNING,
-             "Discarding out of order message modifier continuation.\n");
+             "Dropping out of order message modifier continuation "
+             "!(%u == %u || %u == %u) || %lu < %lu.\n",
+             MSG_STATE_MODIFIER, ch->recv_state,
+             MSG_STATE_MOD_CONT, ch->recv_state,
+             ch->recv_mod_value_size_expected, ch->recv_mod_value_size);
         GNUNET_break_op (0);
         recv_error (ch);
         return;
@@ -829,7 +837,11 @@ handle_psyc_message (struct GNUNET_PSYC_Channel *ch,
           || ch->recv_mod_value_size_expected != ch->recv_mod_value_size)
       {
         LOG (GNUNET_ERROR_TYPE_WARNING,
-             "Discarding out of order message data fragment.\n");
+             "Dropping out of order message data fragment "
+             "(%u < %u || %lu != %lu).\n",
+             ch->recv_state, MSG_STATE_METHOD,
+             ch->recv_mod_value_size_expected, ch->recv_mod_value_size);
+
         GNUNET_break_op (0);
         recv_error (ch);
         return;
@@ -1412,7 +1424,7 @@ GNUNET_PSYC_slave_transmit (struct GNUNET_PSYC_Slave *slave,
  * @param th Handle of the request that is being resumed.
  */
 void
-GNUNET_PSYC_slave_transmit_resume (struct GNUNET_PSYC_MasterTransmitHandle *th)
+GNUNET_PSYC_slave_transmit_resume (struct GNUNET_PSYC_SlaveTransmitHandle *th)
 {
   channel_transmit_resume ((struct GNUNET_PSYC_ChannelTransmitHandle *) th);
 }
diff --git a/src/psyc/psyc_common.c b/src/psyc/psyc_common.c
index 7368011fc..bf5643ff2 100644
--- a/src/psyc/psyc_common.c
+++ b/src/psyc/psyc_common.c
@@ -33,28 +33,33 @@
  * @param data_size  Size of @a data.
  * @param data       Data.
  *
- * @return GNUNET_YES or GNUNET_NO
+ * @return Message type number
+ *         or GNUNET_NO if the message contains invalid or no parts.
  */
-int
-GNUNET_PSYC_check_message_parts (uint16_t data_size, const char *data)
+uint16_t
+GNUNET_PSYC_message_last_part (uint16_t data_size, const char *data)
 {
   const struct GNUNET_MessageHeader *pmsg;
+  uint16_t ptype = GNUNET_NO;
   uint16_t psize = 0;
   uint16_t pos = 0;
 
-  for (pos = 0; data_size + pos < data_size; pos += psize)
+  for (pos = 0; pos < data_size; pos += psize)
   {
     pmsg = (const struct GNUNET_MessageHeader *) (data + pos);
     psize = ntohs (pmsg->size);
-    if (psize < sizeof (*pmsg) || data_size + pos + psize > data_size)
+    ptype = ntohs (pmsg->type);
+    if (psize < sizeof (*pmsg) || pos + psize > data_size
+        || ptype < GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD
+        || GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL < ptype)
     {
       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-                  "Invalid message part of type %u and size %u.",
-                  ntohs (pmsg->type), psize);
+                  "Invalid message part of type %u and size %u.\n",
+                  ptype, psize);
       return GNUNET_NO;
     }
   }
-  return GNUNET_YES;
+  return ptype;
 }
 
 
@@ -89,7 +94,8 @@ GNUNET_PSYC_log_message (enum GNUNET_ErrorType kind,
     uint16_t name_size = ntohs (mod->name_size);
     char oper = ' ' < mod->oper ? mod->oper : ' ';
     GNUNET_log (kind, "\t%c%.*s\t%.*s\n", oper, name_size, &mod[1],
-                ntohs (mod->value_size), ((char *) &mod[1]) + name_size + 1);
+                size - sizeof (*mod) - name_size - 1,
+                ((char *) &mod[1]) + name_size + 1);
     break;
   }
   case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT:
diff --git a/src/psyc/test_psyc.c b/src/psyc/test_psyc.c
index 88947be60..360d56c06 100644
--- a/src/psyc/test_psyc.c
+++ b/src/psyc/test_psyc.c
@@ -35,7 +35,7 @@
 #include "gnunet_env_lib.h"
 #include "gnunet_psyc_service.h"
 
-#define TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 10)
+#define TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 30)
 
 #define DEBUG_SERVICE 1
 
@@ -72,6 +72,7 @@ struct TransmitClosure
   char *data[16];
   const char *mod_value;
   size_t mod_value_size;
+  uint8_t data_delay[16];
   uint8_t data_count;
   uint8_t paused;
   uint8_t n;
@@ -259,13 +260,16 @@ transmit_resume (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Transmission resumed.\n");
   struct TransmitClosure *tmit = cls;
-  tmit->paused = GNUNET_NO;
-  GNUNET_PSYC_master_transmit_resume (tmit->mst_tmit);
+  if (NULL != tmit->mst_tmit)
+    GNUNET_PSYC_master_transmit_resume (tmit->mst_tmit);
+  else
+    GNUNET_PSYC_slave_transmit_resume (tmit->slv_tmit);
 }
 
 
 static int
-tmit_notify_mod (void *cls, uint16_t *data_size, void *data, uint8_t *oper)
+tmit_notify_mod (void *cls, uint16_t *data_size, void *data, uint8_t *oper,
+                 uint32_t *full_value_size)
 {
   struct TransmitClosure *tmit = cls;
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
@@ -288,6 +292,8 @@ tmit_notify_mod (void *cls, uint16_t *data_size, void *data, uint8_t *oper)
       return GNUNET_YES;
     }
 
+    GNUNET_assert (value_size < UINT32_MAX);
+    *full_value_size = value_size;
     *oper = op;
     name_size = strlen (name);
 
@@ -351,7 +357,7 @@ tmit_notify_data (void *cls, uint16_t *data_size, void *data)
 
   uint16_t size = strlen (tmit->data[tmit->n]);
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Transmit notify data: %lu bytes available, "
+              "Transmit notify data: %u bytes available, "
               "processing fragment %u/%u (size %u).\n",
               *data_size, tmit->n + 1, tmit->data_count, size);
   if (*data_size < size)
@@ -361,17 +367,18 @@ tmit_notify_data (void *cls, uint16_t *data_size, void *data)
     return GNUNET_SYSERR;
   }
 
-  if (GNUNET_YES == tmit->paused && tmit->n == tmit->data_count - 1)
+  if (GNUNET_YES != tmit->paused && 0 < tmit->data_delay[tmit->n])
   {
-    /* Send last fragment later. */
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Transmission paused.\n");
     tmit->paused = GNUNET_YES;
-    GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
-                                  (GNUNET_TIME_UNIT_SECONDS, 3),
-                                  &transmit_resume, tmit);
+    GNUNET_SCHEDULER_add_delayed (
+      GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
+                                     tmit->data_delay[tmit->n]),
+      &transmit_resume, tmit);
     *data_size = 0;
     return GNUNET_NO;
   }
+  tmit->paused = GNUNET_NO;
 
   *data_size = size;
   memcpy (data, tmit->data[tmit->n], size);
@@ -416,8 +423,9 @@ slave_join ()
   GNUNET_ENV_environment_add (env, GNUNET_ENV_OP_ASSIGN,
                               "_foo_bar", "foo bar baz", 11);
   slv = GNUNET_PSYC_slave_join (cfg, &channel_pub_key, slave_key, &origin,
-                                16, relays, &slave_message, &join_request, &slave_joined,
-                                NULL, "_request_join", env, "some data", 9);
+                                16, relays, &slave_message, &join_request,
+                                &slave_joined, NULL, "_request_join", env,
+                                "some data", 9);
   GNUNET_ENV_environment_destroy (env);
 }
 
@@ -427,17 +435,45 @@ master_transmit ()
 {
   GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Master sending message to all.\n");
   test = TEST_MASTER_TRANSMIT;
+  uint32_t i, j;
+
+  char *name_max = "_test_max";
+  uint8_t name_max_size = sizeof ("_test_max");
+  char *val_max = GNUNET_malloc (GNUNET_PSYC_MODIFIER_MAX_PAYLOAD);
+  for (i = 0; i < GNUNET_PSYC_MODIFIER_MAX_PAYLOAD; i++)
+    val_max[i] = (0 == i % 10000) ? '0' + i / 10000 : '.';
+
+  char *name_cont = "_test_cont";
+  uint8_t name_cont_size = sizeof ("_test_cont");
+  char *val_cont = GNUNET_malloc (GNUNET_PSYC_MODIFIER_MAX_PAYLOAD
+                                  + GNUNET_PSYC_MOD_CONT_MAX_PAYLOAD);
+  for (i = 0; i < GNUNET_PSYC_MODIFIER_MAX_PAYLOAD - name_cont_size; i++)
+    val_cont[i] = (0 == i % 10000) ? '0' + i / 10000 : ':';
+  for (j = 0; j < GNUNET_PSYC_MOD_CONT_MAX_PAYLOAD; j++, i++)
+    val_cont[i] = (0 == j % 10000) ? '0' + j / 10000 : '!';
 
   tmit = GNUNET_new (struct TransmitClosure);
   tmit->env = GNUNET_ENV_environment_create ();
   GNUNET_ENV_environment_add (tmit->env, GNUNET_ENV_OP_ASSIGN,
                               "_foo", "bar baz", 7);
   GNUNET_ENV_environment_add (tmit->env, GNUNET_ENV_OP_ASSIGN,
+                              name_max, val_max,
+                              GNUNET_PSYC_MODIFIER_MAX_PAYLOAD
+                              - name_max_size);
+  GNUNET_ENV_environment_add (tmit->env, GNUNET_ENV_OP_ASSIGN,
                               "_foo_bar", "foo bar baz", 11);
+  GNUNET_ENV_environment_add (tmit->env, GNUNET_ENV_OP_ASSIGN,
+                              name_cont, val_cont,
+                              GNUNET_PSYC_MODIFIER_MAX_PAYLOAD - name_cont_size
+                              + GNUNET_PSYC_MOD_CONT_MAX_PAYLOAD);
   tmit->data[0] = "foo";
-  tmit->data[1] = "foo bar";
-  tmit->data[2] = "foo bar baz";
-  tmit->data_count = 3;
+  tmit->data[1] =  GNUNET_malloc (GNUNET_PSYC_DATA_MAX_PAYLOAD + 1);
+  for (i = 0; i < GNUNET_PSYC_DATA_MAX_PAYLOAD; i++)
+    tmit->data[1][i] = (0 == i % 10000) ? '0' + i / 10000 : '_';
+  tmit->data[2] = "foo bar";
+  tmit->data[3] = "foo bar baz";
+  tmit->data_delay[1] = 3;
+  tmit->data_count = 4;
   tmit->mst_tmit
     = GNUNET_PSYC_master_transmit (mst, "_notice_test", tmit_notify_mod,
                                    tmit_notify_data, tmit,