multicast: send messages between client lib & service

4 files changed, 151 insertions, 138 deletions

diff --git a/src/psyc/Makefile.am b/src/psyc/Makefile.am
index 5f3abfbb5..212c383ac 100644
--- a/src/psyc/Makefile.am
+++ b/src/psyc/Makefile.am
@@ -42,14 +42,14 @@ gnunet_service_psyc_SOURCES = \
  gnunet-service-psyc.c \
  psyc_common.c
 gnunet_service_psyc_LDADD = \
-  $(top_builddir)/src/statistics/libgnunetstatistics.la \
   $(top_builddir)/src/util/libgnunetutil.la \
+  $(top_builddir)/src/statistics/libgnunetstatistics.la \
   $(top_builddir)/src/multicast/libgnunetmulticast.la \
   $(top_builddir)/src/psycstore/libgnunetpsycstore.la \
   $(GN_LIBINTL)
 gnunet_service_psyc_DEPENDENCIES = \
-  $(top_builddir)/src/statistics/libgnunetstatistics.la \
   $(top_builddir)/src/util/libgnunetutil.la \
+  $(top_builddir)/src/statistics/libgnunetstatistics.la \
   $(top_builddir)/src/multicast/libgnunetmulticast.la \
   $(top_builddir)/src/psycstore/libgnunetpsycstore.la
 gnunet_service_psyc_CFLAGS = $(AM_CFLAGS)
diff --git a/src/psyc/gnunet-service-psyc.c b/src/psyc/gnunet-service-psyc.c
index 3a29c8ffd..70322adaa 100644
--- a/src/psyc/gnunet-service-psyc.c
+++ b/src/psyc/gnunet-service-psyc.c
@@ -58,10 +58,16 @@ static struct GNUNET_SERVER_NotificationContext *nc;
 static struct GNUNET_PSYCSTORE_Handle *store;
 
 /**
- * All connected masters and slaves.
+ * All connected masters.
  * Channel's pub_key_hash -> struct Channel
  */
-static struct GNUNET_CONTAINER_MultiHashMap *clients;
+static struct GNUNET_CONTAINER_MultiHashMap *masters;
+
+/**
+ * All connected slaves.
+ * Channel's pub_key_hash -> struct Channel
+ */
+static struct GNUNET_CONTAINER_MultiHashMap *slaves;
 
 
 /**
@@ -158,7 +164,7 @@ struct FragmentQueue
 
 
 /**
- * Common part of the client context for both a master and slave channel.
+ * Common part of the client context for both a channel master and slave.
  */
 struct Channel
 {
@@ -266,7 +272,7 @@ struct Master
   /**
    * Transmit handle for multicast.
    */
-  struct GNUNET_MULTICAST_OriginMessageHandle *tmit_handle;
+  struct GNUNET_MULTICAST_OriginTransmitHandle *tmit_handle;
 
   /**
    * Last message ID transmitted to this channel.
@@ -307,7 +313,7 @@ struct Slave
   /**
    * Private key of the slave.
    */
-  struct GNUNET_CRYPTO_EddsaPrivateKey slave_key;
+  struct GNUNET_CRYPTO_EddsaPrivateKey priv_key;
 
   /**
    * Handle for the multicast member.
@@ -317,7 +323,7 @@ struct Slave
   /**
    * Transmit handle for multicast.
    */
-  struct GNUNET_MULTICAST_MemberRequestHandle *tmit_handle;
+  struct GNUNET_MULTICAST_MemberTransmitHandle *tmit_handle;
 
   /**
    * Peer identity of the origin.
@@ -382,7 +388,7 @@ client_cleanup (struct Channel *ch)
     struct Master *mst = (struct Master *) ch;
     if (NULL != mst->origin)
       GNUNET_MULTICAST_origin_stop (mst->origin);
-    GNUNET_CONTAINER_multihashmap_remove (clients, &ch->pub_key_hash, mst);
+    GNUNET_CONTAINER_multihashmap_remove (masters, &ch->pub_key_hash, ch);
   }
   else
   {
@@ -393,6 +399,7 @@ client_cleanup (struct Channel *ch)
       GNUNET_free (slv->relays);
     if (NULL != slv->member)
       GNUNET_MULTICAST_member_part (slv->member);
+    GNUNET_CONTAINER_multihashmap_remove (slaves, &ch->pub_key_hash, ch);
   }
 
   GNUNET_free (ch);
@@ -975,7 +982,7 @@ request_cb (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *slave_key,
                                                           NULL, NULL))
     {
       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-                  "%p Dropping message with invalid parts "
+                  "%p Dropping request with invalid parts "
                   "received from multicast.\n", ch);
       GNUNET_break_op (0);
       break;
@@ -1017,6 +1024,7 @@ master_counters_cb (void *cls, int result, uint64_t max_fragment_id,
 {
   struct Master *mst = cls;
   struct Channel *ch = &mst->channel;
+
   struct CountersResult *res = GNUNET_malloc (sizeof (*res));
   res->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MASTER_START_ACK);
   res->header.size = htons (sizeof (*res));
@@ -1031,12 +1039,20 @@ master_counters_cb (void *cls, int result, uint64_t max_fragment_id,
     mst->max_group_generation = max_group_generation;
     mst->origin
       = GNUNET_MULTICAST_origin_start (cfg, &mst->priv_key,
-                                       max_fragment_id + 1,
+                                       max_fragment_id,
                                        join_cb, membership_test_cb,
                                        replay_fragment_cb, replay_message_cb,
                                        request_cb, message_cb, ch);
     ch->ready = GNUNET_YES;
   }
+  else
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                "%p GNUNET_PSYCSTORE_counters_get() "
+                "returned %d for channel %s.\n",
+                ch, result, GNUNET_h2s (&ch->pub_key_hash));
+  }
+
   GNUNET_SERVER_notification_context_add (nc, ch->client);
   GNUNET_SERVER_notification_context_unicast (nc, ch->client, &res->header,
                                               GNUNET_NO);
@@ -1054,6 +1070,7 @@ slave_counters_cb (void *cls, int result, uint64_t max_fragment_id,
 {
   struct Slave *slv = cls;
   struct Channel *ch = &slv->channel;
+
   struct CountersResult *res = GNUNET_malloc (sizeof (*res));
   res->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN_ACK);
   res->header.size = htons (sizeof (*res));
@@ -1065,7 +1082,7 @@ slave_counters_cb (void *cls, int result, uint64_t max_fragment_id,
     ch->max_message_id = max_message_id;
     ch->max_state_message_id = max_state_message_id;
     slv->member
-      = GNUNET_MULTICAST_member_join (cfg, &ch->pub_key, &slv->slave_key,
+      = GNUNET_MULTICAST_member_join (cfg, &ch->pub_key, &slv->priv_key,
                                       &slv->origin,
                                       slv->relay_count, slv->relays,
                                       slv->join_req, join_cb,
@@ -1074,6 +1091,13 @@ slave_counters_cb (void *cls, int result, uint64_t max_fragment_id,
                                       message_cb, ch);
     ch->ready = GNUNET_YES;
   }
+  else
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                "%p GNUNET_PSYCSTORE_counters_get() "
+                "returned %d for channel %s.\n",
+                ch, result, GNUNET_h2s (&ch->pub_key_hash));
+  }
 
   GNUNET_SERVER_notification_context_add (nc, ch->client);
   GNUNET_SERVER_notification_context_unicast (nc, ch->client, &res->header,
@@ -1118,9 +1142,9 @@ handle_master_start (void *cls, struct GNUNET_SERVER_Client *client,
 
   GNUNET_PSYCSTORE_counters_get (store, &ch->pub_key, master_counters_cb, mst);
 
-  GNUNET_SERVER_client_set_user_context (client, &mst->channel);
-  GNUNET_CONTAINER_multihashmap_put (clients, &ch->pub_key_hash, mst,
+  GNUNET_CONTAINER_multihashmap_put (masters, &ch->pub_key_hash, ch,
                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
+  GNUNET_SERVER_client_set_user_context (client, ch);
   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 }
 
@@ -1135,7 +1159,7 @@ handle_slave_join (void *cls, struct GNUNET_SERVER_Client *client,
   const struct SlaveJoinRequest *req
     = (const struct SlaveJoinRequest *) msg;
   struct Slave *slv = GNUNET_new (struct Slave);
-  slv->slave_key = req->slave_key;
+  slv->priv_key = req->slave_key;
   slv->origin = req->origin;
   slv->relay_count = ntohl (req->relay_count);
   if (0 < slv->relay_count)
@@ -1163,6 +1187,8 @@ handle_slave_join (void *cls, struct GNUNET_SERVER_Client *client,
 
   GNUNET_PSYCSTORE_counters_get (store, &ch->pub_key, slave_counters_cb, slv);
 
+  GNUNET_CONTAINER_multihashmap_put (slaves, &ch->pub_key_hash, ch,
+                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
   GNUNET_SERVER_client_set_user_context (client, &slv->channel);
   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 }
@@ -1183,8 +1209,7 @@ send_message_ack (struct Channel *ch)
   res.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_ACK);
 
   GNUNET_SERVER_notification_context_add (nc, ch->client);
-  GNUNET_SERVER_notification_context_unicast (nc, ch->client, &res,
-                                              GNUNET_NO);
+  GNUNET_SERVER_notification_context_unicast (nc, ch->client, &res, GNUNET_NO);
 }
 
 
@@ -1554,7 +1579,8 @@ run (void *cls, struct GNUNET_SERVER_Handle *server,
   cfg = c;
   store = GNUNET_PSYCSTORE_connect (cfg);
   stats = GNUNET_STATISTICS_create ("psyc", cfg);
-  clients = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
+  masters = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
+  slaves = 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);
diff --git a/src/psyc/psyc_api.c b/src/psyc/psyc_api.c
index 22f1da069..85f86ceaa 100644
--- a/src/psyc/psyc_api.c
+++ b/src/psyc/psyc_api.c
@@ -41,11 +41,10 @@
 
 #define LOG(kind,...) GNUNET_log_from (kind, "psyc-api",__VA_ARGS__)
 
-struct OperationHandle
+struct MessageQueue
 {
-  struct OperationHandle *prev;
-  struct OperationHandle *next;
-  struct GNUNET_MessageHeader *msg;
+  struct MessageQueue *prev;
+  struct MessageQueue *next;
 };
 
 
@@ -87,19 +86,19 @@ struct GNUNET_PSYC_Channel
   struct GNUNET_CLIENT_TransmitHandle *th;
 
   /**
-   * Head of operations to transmit.
+   * Head of messages to transmit to the service.
    */
-  struct OperationHandle *tmit_head;
+  struct MessageQueue *tmit_head;
 
   /**
-   * Tail of operations to transmit.
+   * Tail of operations to transmit to the service.
    */
-  struct OperationHandle *tmit_tail;
+  struct MessageQueue *tmit_tail;
 
   /**
-   * Message being transmitted to the PSYC service.
+   * Message currently being transmitted to the service.
    */
-  struct OperationHandle *tmit_msg;
+  struct MessageQueue *tmit_msg;
 
   /**
    * Message to send on reconnect.
@@ -201,8 +200,6 @@ struct GNUNET_PSYC_Master
   struct GNUNET_PSYC_Channel ch;
 
   GNUNET_PSYC_MasterStartCallback start_cb;
-
-  uint64_t max_message_id;
 };
 
 
@@ -214,8 +211,6 @@ struct GNUNET_PSYC_Slave
   struct GNUNET_PSYC_Channel ch;
 
   GNUNET_PSYC_SlaveJoinCallback join_cb;
-
-  uint64_t max_message_id;
 };
 
 
@@ -269,30 +264,30 @@ channel_transmit_data (struct GNUNET_PSYC_Channel *ch);
 /**
  * Reschedule a connect attempt to the service.
  *
- * @param c channel to reconnect
+ * @param ch  Channel to reconnect.
  */
 static void
-reschedule_connect (struct GNUNET_PSYC_Channel *c)
+reschedule_connect (struct GNUNET_PSYC_Channel *ch)
 {
-  GNUNET_assert (c->reconnect_task == GNUNET_SCHEDULER_NO_TASK);
+  GNUNET_assert (ch->reconnect_task == GNUNET_SCHEDULER_NO_TASK);
 
-  if (NULL != c->th)
+  if (NULL != ch->th)
   {
-    GNUNET_CLIENT_notify_transmit_ready_cancel (c->th);
-    c->th = NULL;
+    GNUNET_CLIENT_notify_transmit_ready_cancel (ch->th);
+    ch->th = NULL;
   }
-  if (NULL != c->client)
+  if (NULL != ch->client)
   {
-    GNUNET_CLIENT_disconnect (c->client);
-    c->client = NULL;
+    GNUNET_CLIENT_disconnect (ch->client);
+    ch->client = NULL;
   }
-  c->in_receive = GNUNET_NO;
+  ch->in_receive = GNUNET_NO;
   LOG (GNUNET_ERROR_TYPE_DEBUG,
        "Scheduling task to reconnect to PSYC service in %s.\n",
-       GNUNET_STRINGS_relative_time_to_string (c->reconnect_delay, GNUNET_YES));
-  c->reconnect_task =
-      GNUNET_SCHEDULER_add_delayed (c->reconnect_delay, &reconnect, c);
-  c->reconnect_delay = GNUNET_TIME_STD_BACKOFF (c->reconnect_delay);
+       GNUNET_STRINGS_relative_time_to_string (ch->reconnect_delay, GNUNET_YES));
+  ch->reconnect_task =
+      GNUNET_SCHEDULER_add_delayed (ch->reconnect_delay, &reconnect, ch);
+  ch->reconnect_delay = GNUNET_TIME_STD_BACKOFF (ch->reconnect_delay);
 }
 
 
@@ -306,7 +301,7 @@ transmit_next (struct GNUNET_PSYC_Channel *ch);
 
 
 /**
- * Reset data stored related to the last received message.
+ * Reset stored data related to the last received message.
  */
 static void
 recv_reset (struct GNUNET_PSYC_Channel *ch)
@@ -356,51 +351,53 @@ queue_message (struct GNUNET_PSYC_Channel *ch,
        "Queueing message of type %u and size %u (end: %u)).\n",
        ntohs (msg->type), size, end);
 
-  struct OperationHandle *op = ch->tmit_msg;
-  if (NULL != op)
+  struct MessageQueue *mq = ch->tmit_msg;
+  struct GNUNET_MessageHeader *qmsg = NULL;
+  if (NULL != mq)
   {
+    qmsg = (struct GNUNET_MessageHeader *) &mq[1];
     if (NULL == msg
-        || GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD < op->msg->size + size)
+        || GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD < qmsg->size + size)
     {
       /* End of message or buffer is full, add it to transmission queue
        * and start with empty buffer */
-      op->msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE);
-      op->msg->size = htons (op->msg->size);
-      GNUNET_CONTAINER_DLL_insert_tail (ch->tmit_head, ch->tmit_tail, op);
-      ch->tmit_msg = op = NULL;
+      qmsg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE);
+      qmsg->size = htons (qmsg->size);
+      GNUNET_CONTAINER_DLL_insert_tail (ch->tmit_head, ch->tmit_tail, mq);
+      ch->tmit_msg = mq = NULL;
       ch->tmit_ack_pending++;
     }
     else
     {
       /* Message fits in current buffer, append */
-      ch->tmit_msg = op
-        = GNUNET_realloc (op, sizeof (*op) + op->msg->size + size);
-      op->msg = (struct GNUNET_MessageHeader *) &op[1];
-      memcpy ((char *) op->msg + op->msg->size, msg, size);
-      op->msg->size += size;
+      ch->tmit_msg
+        = mq = GNUNET_realloc (mq, sizeof (*mq) + qmsg->size + size);
+      qmsg = (struct GNUNET_MessageHeader *) &mq[1];
+      memcpy ((char *) qmsg + qmsg->size, msg, size);
+      qmsg->size += size;
     }
   }
 
-  if (NULL == op && NULL != msg)
+  if (NULL == mq && NULL != msg)
   {
     /* Empty buffer, copy over message. */
-    ch->tmit_msg = op
-      = GNUNET_malloc (sizeof (*op) + sizeof (*op->msg) + size);
-    op->msg = (struct GNUNET_MessageHeader *) &op[1];
-    op->msg->size = sizeof (*op->msg) + size;
-    memcpy (&op->msg[1], msg, size);
+    ch->tmit_msg
+      = mq = GNUNET_malloc (sizeof (*mq) + sizeof (*qmsg) + size);
+    qmsg = (struct GNUNET_MessageHeader *) &mq[1];
+    qmsg->size = sizeof (*qmsg) + size;
+    memcpy (&qmsg[1], msg, size);
   }
 
-  if (NULL != op
+  if (NULL != mq
       && (GNUNET_YES == end
           || (GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD
-              < op->msg->size + sizeof (struct GNUNET_MessageHeader))))
+              < qmsg->size + sizeof (struct GNUNET_MessageHeader))))
   {
     /* End of message or buffer is full, add it to transmission queue. */
-    op->msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE);
-    op->msg->size = htons (op->msg->size);
-    GNUNET_CONTAINER_DLL_insert_tail (ch->tmit_head, ch->tmit_tail, op);
-    ch->tmit_msg = op = NULL;
+    qmsg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE);
+    qmsg->size = htons (qmsg->size);
+    GNUNET_CONTAINER_DLL_insert_tail (ch->tmit_head, ch->tmit_tail, mq);
+    ch->tmit_msg = mq = NULL;
     ch->tmit_ack_pending++;
   }
 
@@ -577,6 +574,7 @@ channel_transmit_data (struct GNUNET_PSYC_Channel *ch)
  * @param notify_data Function to call to obtain fragments of the data.
  * @param notify_cls Closure for @a notify_mod and @a notify_data.
  * @param flags Flags for the message being transmitted.
+ *
  * @return Transmission handle, NULL on error (i.e. more than one request queued).
  */
 static struct GNUNET_PSYC_ChannelTransmitHandle *
@@ -593,14 +591,14 @@ channel_transmit (struct GNUNET_PSYC_Channel *ch,
 
   size_t size = strlen (method_name) + 1;
   struct GNUNET_PSYC_MessageMethod *pmeth;
-  struct OperationHandle *op;
-
-  ch->tmit_msg = op = GNUNET_malloc (sizeof (*op) + sizeof (*op->msg)
-                                     + sizeof (*pmeth) + size);
-  op->msg = (struct GNUNET_MessageHeader *) &op[1];
-  op->msg->size = sizeof (*op->msg) + sizeof (*pmeth) + size;
-
-  pmeth = (struct GNUNET_PSYC_MessageMethod *) &op->msg[1];
+  struct GNUNET_MessageHeader *qmsg;
+  struct MessageQueue *
+    mq = ch->tmit_msg = GNUNET_malloc (sizeof (*mq) + sizeof (*qmsg)
+                                       + sizeof (*pmeth) + size);
+  qmsg = (struct GNUNET_MessageHeader *) &mq[1];
+  qmsg->size = sizeof (*qmsg) + sizeof (*pmeth) + size;
+
+  pmeth = (struct GNUNET_PSYC_MessageMethod *) &qmsg[1];
   pmeth->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD);
   pmeth->header.size = htons (sizeof (*pmeth) + size);
   pmeth->flags = htonl (flags);
@@ -928,7 +926,7 @@ message_handler (void *cls,
 
   if (NULL == msg)
   {
-    // timeout / disconnected from server, reconnect
+    // timeout / disconnected from service, reconnect
     reschedule_connect (ch);
     return;
   }
@@ -970,17 +968,15 @@ message_handler (void *cls,
   case GNUNET_MESSAGE_TYPE_PSYC_MASTER_START_ACK:
   {
     struct CountersResult *cres = (struct CountersResult *) msg;
-    mst->max_message_id = GNUNET_ntohll (cres->max_message_id);
     if (NULL != mst->start_cb)
-      mst->start_cb (ch->cb_cls, mst->max_message_id);
+      mst->start_cb (ch->cb_cls, GNUNET_ntohll (cres->max_message_id));
     break;
   }
   case GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN_ACK:
   {
     struct CountersResult *cres = (struct CountersResult *) msg;
-    slv->max_message_id = GNUNET_ntohll (cres->max_message_id);
     if (NULL != slv->join_cb)
-      slv->join_cb (ch->cb_cls, slv->max_message_id);
+      slv->join_cb (ch->cb_cls, GNUNET_ntohll (cres->max_message_id));
     break;
   }
   case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_ACK:
@@ -1005,31 +1001,32 @@ message_handler (void *cls,
 /**
  * Transmit next message to service.
  *
- * @param cls The 'struct GNUNET_PSYC_Channel'.
- * @param size Number of bytes available in buf.
- * @param buf Where to copy the message.
- * @return Number of bytes copied to buf.
+ * @param cls   The struct GNUNET_PSYC_Channel.
+ * @param size  Number of bytes available in @a buf.
+ * @param buf   Where to copy the message.
+ *
+ * @return Number of bytes copied to @a buf.
  */
 static size_t
 send_next_message (void *cls, size_t size, void *buf)
 {
-  struct GNUNET_PSYC_Channel *ch = cls;
-  struct OperationHandle *op = ch->tmit_head;
-  size_t ret;
   LOG (GNUNET_ERROR_TYPE_DEBUG, "send_next_message()\n");
-  ch->th = NULL;
-  if (NULL == op->msg)
+  struct GNUNET_PSYC_Channel *ch = cls;
+  struct MessageQueue *mq = ch->tmit_head;
+  if (NULL == mq)
     return 0;
-  ret = ntohs (op->msg->size);
+  struct GNUNET_MessageHeader *qmsg = (struct GNUNET_MessageHeader *) &mq[1];
+  size_t ret = ntohs (qmsg->size);
+  ch->th = NULL;
   if (ret > size)
   {
     reschedule_connect (ch);
     return 0;
   }
-  memcpy (buf, op->msg, ret);
+  memcpy (buf, qmsg, ret);
 
-  GNUNET_CONTAINER_DLL_remove (ch->tmit_head, ch->tmit_tail, op);
-  GNUNET_free (op);
+  GNUNET_CONTAINER_DLL_remove (ch->tmit_head, ch->tmit_tail, mq);
+  GNUNET_free (mq);
 
   if (NULL != ch->tmit_head)
     transmit_next (ch);
@@ -1056,12 +1053,13 @@ transmit_next (struct GNUNET_PSYC_Channel *ch)
   if (NULL != ch->th || NULL == ch->client)
     return;
 
-  struct OperationHandle *op = ch->tmit_head;
-  if (NULL == op)
+  struct MessageQueue *mq = ch->tmit_head;
+  if (NULL == mq)
     return;
+  struct GNUNET_MessageHeader *qmsg = (struct GNUNET_MessageHeader *) &mq[1];
 
   ch->th = GNUNET_CLIENT_notify_transmit_ready (ch->client,
-                                                ntohs (op->msg->size),
+                                                ntohs (qmsg->size),
                                                 GNUNET_TIME_UNIT_FOREVER_REL,
                                                 GNUNET_NO,
                                                 &send_next_message,
@@ -1087,15 +1085,14 @@ reconnect (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
   GNUNET_assert (NULL == ch->client);
   ch->client = GNUNET_CLIENT_connect ("psyc", ch->cfg);
   GNUNET_assert (NULL != ch->client);
+  uint16_t reconn_size = ntohs (ch->reconnect_msg->size);
 
   if (NULL == ch->tmit_head ||
-      ch->tmit_head->msg->type != ch->reconnect_msg->type)
+      0 != memcmp (&ch->tmit_head[1], ch->reconnect_msg, reconn_size))
   {
-    uint16_t reconn_size = ntohs (ch->reconnect_msg->size);
-    struct OperationHandle *op = GNUNET_malloc (sizeof (*op) + reconn_size);
-    memcpy (&op[1], ch->reconnect_msg, reconn_size);
-    op->msg = (struct GNUNET_MessageHeader *) &op[1];
-    GNUNET_CONTAINER_DLL_insert (ch->tmit_head, ch->tmit_tail, op);
+    struct MessageQueue *mq = GNUNET_malloc (sizeof (*mq) + reconn_size);
+    memcpy (&mq[1], ch->reconnect_msg, reconn_size);
+    GNUNET_CONTAINER_DLL_insert (ch->tmit_head, ch->tmit_tail, mq);
   }
   transmit_next (ch);
 }
@@ -1104,7 +1101,7 @@ reconnect (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 /**
  * Disconnect from the PSYC service.
  *
- * @param c Channel handle to disconnect
+ * @param c  Channel handle to disconnect.
  */
 static void
 disconnect (void *c)
@@ -1167,6 +1164,7 @@ disconnect (void *c)
  * @param join_cb Function to invoke when a peer wants to join.
  * @param master_started_cb Function to invoke after the channel master started.
  * @param cls Closure for @a master_started_cb and @a join_cb.
+ *
  * @return Handle for the channel master, NULL on error.
  */
 struct GNUNET_PSYC_Master *
@@ -1187,17 +1185,16 @@ GNUNET_PSYC_master_start (const struct GNUNET_CONFIGURATION_Handle *cfg,
   req->channel_key = *channel_key;
   req->policy = policy;
 
+  mst->start_cb = master_started_cb;
+  ch->message_cb = message_cb;
+  ch->join_cb = join_cb;
+  ch->cb_cls = cls;
   ch->cfg = cfg;
   ch->is_master = GNUNET_YES;
   ch->reconnect_msg = (struct GNUNET_MessageHeader *) req;
   ch->reconnect_delay = GNUNET_TIME_UNIT_ZERO;
   ch->reconnect_task = GNUNET_SCHEDULER_add_now (&reconnect, mst);
 
-  ch->message_cb = message_cb;
-  ch->join_cb = join_cb;
-  ch->cb_cls = cls;
-  mst->start_cb = master_started_cb;
-
   return mst;
 }
 
@@ -1260,6 +1257,7 @@ GNUNET_PSYC_join_decision (struct GNUNET_PSYC_JoinHandle *jh,
  * @param notify_data Function to call to obtain fragments of the data.
  * @param notify_cls Closure for @a notify_mod and @a notify_data.
  * @param flags Flags for the message being transmitted.
+ *
  * @return Transmission handle, NULL on error (i.e. more than one request queued).
  */
 struct GNUNET_PSYC_MasterTransmitHandle *
@@ -1330,6 +1328,7 @@ GNUNET_PSYC_master_transmit_cancel (struct GNUNET_PSYC_MasterTransmitHandle *th)
  * @param env Environment containing transient variables for the request, or NULL.
  * @param data Payload for the join message.
  * @param data_size Number of bytes in @a data.
+ *
  * @return Handle for the slave, NULL on error.
  */
 struct GNUNET_PSYC_Slave *
@@ -1361,6 +1360,7 @@ GNUNET_PSYC_slave_join (const struct GNUNET_CONFIGURATION_Handle *cfg,
   req->relay_count = htonl (relay_count);
   memcpy (&req[1], relays, relay_count * sizeof (*relays));
 
+  slv->join_cb = slave_joined_cb;
   ch->message_cb = message_cb;
   ch->join_cb = join_cb;
   ch->cb_cls = cls;
@@ -1371,7 +1371,6 @@ GNUNET_PSYC_slave_join (const struct GNUNET_CONFIGURATION_Handle *cfg,
   ch->reconnect_delay = GNUNET_TIME_UNIT_ZERO;
   ch->reconnect_task = GNUNET_SCHEDULER_add_now (&reconnect, slv);
 
-  slv->join_cb = slave_joined_cb;
   return slv;
 }
 
@@ -1401,6 +1400,7 @@ GNUNET_PSYC_slave_part (struct GNUNET_PSYC_Slave *slave)
  * @param notify_data Function to call to obtain fragments of the data.
  * @param notify_cls Closure for @a notify.
  * @param flags Flags for the message being transmitted.
+ *
  * @return Transmission handle, NULL on error (i.e. more than one request
  *         queued).
  */
@@ -1447,6 +1447,7 @@ GNUNET_PSYC_slave_transmit_cancel (struct GNUNET_PSYC_SlaveTransmitHandle *th)
  * APIs.
  *
  * @param master Channel master handle.
+ *
  * @return Channel handle, valid for as long as @a master is valid.
  */
 struct GNUNET_PSYC_Channel *
@@ -1460,6 +1461,7 @@ GNUNET_PSYC_master_get_channel (struct GNUNET_PSYC_Master *master)
  * Convert @a slave to a @e channel handle to access the @e channel APIs.
  *
  * @param slave Slave handle.
+ *
  * @return Channel handle, valid for as long as @a slave is valid.
  */
 struct GNUNET_PSYC_Channel *
@@ -1497,18 +1499,16 @@ GNUNET_PSYC_channel_slave_add (struct GNUNET_PSYC_Channel *channel,
                                uint64_t effective_since)
 {
   struct ChannelSlaveAdd *slvadd;
-  struct OperationHandle *op = GNUNET_malloc (sizeof (*op) + sizeof (*slvadd));
-
-  slvadd = (struct ChannelSlaveAdd *) &op[1];
-  op->msg = (struct GNUNET_MessageHeader *) slvadd;
+  struct MessageQueue *mq = GNUNET_malloc (sizeof (*mq) + sizeof (*slvadd));
 
+  slvadd = (struct ChannelSlaveAdd *) &mq[1];
   slvadd->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_SLAVE_ADD);
   slvadd->header.size = htons (sizeof (*slvadd));
   slvadd->announced_at = GNUNET_htonll (announced_at);
   slvadd->effective_since = GNUNET_htonll (effective_since);
   GNUNET_CONTAINER_DLL_insert_tail (channel->tmit_head,
                                     channel->tmit_tail,
-                                    op);
+                                    mq);
   transmit_next (channel);
 }
 
@@ -1540,16 +1540,15 @@ GNUNET_PSYC_channel_slave_remove (struct GNUNET_PSYC_Channel *channel,
                                   uint64_t announced_at)
 {
   struct ChannelSlaveRemove *slvrm;
-  struct OperationHandle *op = GNUNET_malloc (sizeof (*op) + sizeof (*slvrm));
+  struct MessageQueue *mq = GNUNET_malloc (sizeof (*mq) + sizeof (*slvrm));
 
-  slvrm = (struct ChannelSlaveRemove *) &op[1];
-  op->msg = (struct GNUNET_MessageHeader *) slvrm;
+  slvrm = (struct ChannelSlaveRemove *) &mq[1];
   slvrm->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_SLAVE_RM);
   slvrm->header.size = htons (sizeof (*slvrm));
   slvrm->announced_at = GNUNET_htonll (announced_at);
   GNUNET_CONTAINER_DLL_insert_tail (channel->tmit_head,
                                     channel->tmit_tail,
-                                    op);
+                                    mq);
   transmit_next (channel);
 }
 
@@ -1573,6 +1572,7 @@ GNUNET_PSYC_channel_slave_remove (struct GNUNET_PSYC_Channel *channel,
  *        has been called, the client must not call
  *        GNUNET_PSYC_channel_story_tell_cancel() anymore.
  * @param cls Closure for the callbacks.
+ *
  * @return Handle to cancel story telling operation.
  */
 struct GNUNET_PSYC_Story *
@@ -1615,6 +1615,7 @@ GNUNET_PSYC_channel_story_tell_cancel (struct GNUNET_PSYC_Story *story)
  * @param cb Function called once when a matching state variable is found.
  *        Not called if there's no matching state variable.
  * @param cb_cls Closure for the callbacks.
+ *
  * @return Handle that can be used to cancel the query operation.
  */
 struct GNUNET_PSYC_StateQuery *
@@ -1641,6 +1642,7 @@ GNUNET_PSYC_channel_state_get (struct GNUNET_PSYC_Channel *channel,
  * @param name_prefix Prefix of the state variable name to match.
  * @param cb Function to call with the matching state variables.
  * @param cb_cls Closure for the callbacks.
+ *
  * @return Handle that can be used to cancel the query operation.
  */
 struct GNUNET_PSYC_StateQuery *
diff --git a/src/psyc/test_psyc.conf b/src/psyc/test_psyc.conf
index 7a1eb8404..70a408ae3 100644
--- a/src/psyc/test_psyc.conf
+++ b/src/psyc/test_psyc.conf
@@ -1,17 +1,2 @@
 [arm]
-UNIXPATH = $GNUNET_RUNTIME_DIR/test-gnunet-service-arm.sock
-DEFAULTSERVICES = psyc
-
-[psyc]
-AUTOSTART = YES
-BINARY = gnunet-service-psyc
-UNIXPATH = $GNUNET_RUNTIME_DIR/test-gnunet-service-psyc.sock
-UNIX_MATCH_UID = NO
-UNIX_MATCH_GID = YES
-
-[psycstore]
-AUTOSTART = YES
-BINARY = gnunet-service-psycstore
-UNIXPATH = $GNUNET_RUNTIME_DIR/test-gnunet-service-psycstore.sock
-UNIX_MATCH_UID = NO
-UNIX_MATCH_GID = YES
+DEFAULTSERVICES = psyc psycstore multicast