multicast/psyc/social: message acks & scheduling

2 files changed, 72 insertions, 57 deletions

diff --git a/src/psyc/gnunet-service-psyc.c b/src/psyc/gnunet-service-psyc.c
index ceb0ca95b..e710b41a5 100644
--- a/src/psyc/gnunet-service-psyc.c
+++ b/src/psyc/gnunet-service-psyc.c
@@ -107,17 +107,6 @@ struct TransmitMessage
    */
   uint16_t last_ptype;
 
-  /**
-   * @see enum MessageState
-   */
-  uint8_t state;
-
-  /**
-   * Whether a message ACK has already been sent to the client.
-   * #GNUNET_YES or #GNUNET_NO
-   */
-  uint8_t ack_sent;
-
   /* Followed by message */
 };
 
@@ -281,11 +270,6 @@ struct Channel
   uint32_t tmit_mod_value_size;
 
   /**
-   * @see enum MessageState
-   */
-  uint8_t tmit_state;
-
-  /**
    * Is this a channel master (#GNUNET_YES), or slave (#GNUNET_NO)?
    */
   uint8_t is_master;
@@ -438,6 +422,15 @@ static uint64_t
 message_queue_drop (struct Channel *chn);
 
 
+static void
+schedule_transmit_message (void *cls,
+                           const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  struct Channel *chn = cls;
+  transmit_message (chn);
+}
+
+
 /**
  * Task run during shutdown.
  *
@@ -1145,8 +1138,8 @@ fragment_queue_insert (struct Channel *chn,
       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                   "%p Header of message %" PRIu64 " is NOT complete yet: "
                   "%" PRIu64 " != %" PRIu64 "\n",
-                  chn, GNUNET_ntohll (mmsg->message_id), frag_offset,
-                  fragq->header_size);
+                  chn, GNUNET_ntohll (mmsg->message_id),
+                  frag_offset, fragq->header_size);
     }
   }
 
@@ -1159,8 +1152,8 @@ fragment_queue_insert (struct Channel *chn,
       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                   "%p Message %" PRIu64 " is NOT complete yet: "
                   "%" PRIu64 " != %" PRIu64 "\n",
-                  chn, GNUNET_ntohll (mmsg->message_id), frag_offset,
-                  fragq->size);
+                  chn, GNUNET_ntohll (mmsg->message_id),
+                  frag_offset, fragq->size);
     break;
 
   case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL:
@@ -1486,17 +1479,26 @@ mcast_recv_message (void *cls, const struct GNUNET_MULTICAST_MessageHeader *mmsg
   uint16_t size = ntohs (mmsg->header.size);
 
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "%p Received multicast message of size %u.\n",
-              chn, size);
+              "%p Received multicast message of size %u. "
+              "fragment_id=%" PRIu64 ", message_id=%" PRIu64
+              ", fragment_offset=%" PRIu64 ", flags=%" PRIu64 "\n",
+              chn, size,
+              GNUNET_ntohll (mmsg->fragment_id),
+              GNUNET_ntohll (mmsg->message_id),
+              GNUNET_ntohll (mmsg->fragment_offset),
+              GNUNET_ntohll (mmsg->flags));
 
   GNUNET_PSYCSTORE_fragment_store (store, &chn->pub_key, mmsg, 0,
                                    &store_recv_fragment_store_result, chn);
 
   uint16_t first_ptype = 0, last_ptype = 0;
-  if (GNUNET_SYSERR
-      == GNUNET_PSYC_receive_check_parts (size - sizeof (*mmsg),
-                                          (const char *) &mmsg[1],
-                                          &first_ptype, &last_ptype))
+  int check = GNUNET_PSYC_receive_check_parts (size - sizeof (*mmsg),
+                                               (const char *) &mmsg[1],
+                                               &first_ptype, &last_ptype);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Message check result %d, first part type %u, last part type %u\n",
+              chn, check, first_ptype, last_ptype);
+  if (GNUNET_SYSERR == check)
   {
     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                 "%p Dropping incoming multicast message with invalid parts.\n",
@@ -1505,10 +1507,6 @@ mcast_recv_message (void *cls, const struct GNUNET_MULTICAST_MessageHeader *mmsg
     return;
   }
 
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Message parts: first: type %u, last: type %u\n",
-              first_ptype, last_ptype);
-
   fragment_queue_insert (chn, mmsg, first_ptype, last_ptype);
   message_queue_run (chn);
 }
@@ -1965,6 +1963,8 @@ transmit_notify (void *cls, size_t *data_size, void *data)
   {
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                 "%p transmit_notify: nothing to send.\n", chn);
+    if (NULL != tmit_msg && *data_size < tmit_msg->size)
+      GNUNET_break (0);
     *data_size = 0;
     return GNUNET_NO;
   }
@@ -1975,9 +1975,13 @@ transmit_notify (void *cls, size_t *data_size, void *data)
   *data_size = tmit_msg->size;
   memcpy (data, &tmit_msg[1], *data_size);
 
-  int ret = (MSG_STATE_END < chn->tmit_state) ? GNUNET_NO : GNUNET_YES;
+  int ret
+    = (tmit_msg->last_ptype < GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END)
+    ? GNUNET_NO
+    : GNUNET_YES;
 
-  if (NULL != tmit_msg->client && GNUNET_NO == tmit_msg->ack_sent)
+  /* FIXME: handle disconnecting clients */
+  if (NULL != tmit_msg->client)
     send_message_ack (chn, tmit_msg->client);
 
   GNUNET_CONTAINER_DLL_remove (chn->tmit_head, chn->tmit_tail, tmit_msg);
@@ -1985,7 +1989,7 @@ transmit_notify (void *cls, size_t *data_size, void *data)
 
   if (NULL != chn->tmit_head)
   {
-    transmit_message (chn);
+    GNUNET_SCHEDULER_add_now (schedule_transmit_message, chn);
   }
   else if (GNUNET_YES == chn->is_disconnected
            && tmit_msg->last_ptype < GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END)
@@ -2037,10 +2041,12 @@ slave_transmit_notify (void *cls, size_t *data_size, void *data)
 static void
 master_transmit_message (struct Master *mst)
 {
+  if (NULL == mst->chn.tmit_head)
+    return;
   if (NULL == mst->tmit_handle)
   {
     mst->tmit_handle
-      = GNUNET_MULTICAST_origin_to_all (mst->origin, mst->max_message_id,
+      = GNUNET_MULTICAST_origin_to_all (mst->origin, mst->chn.tmit_head->id,
                                         mst->max_group_generation,
                                         master_transmit_notify, mst);
   }
@@ -2057,10 +2063,12 @@ master_transmit_message (struct Master *mst)
 static void
 slave_transmit_message (struct Slave *slv)
 {
+  if (NULL == slv->chn.tmit_head)
+    return;
   if (NULL == slv->tmit_handle)
   {
     slv->tmit_handle
-      = GNUNET_MULTICAST_member_to_origin (slv->member, slv->max_request_id,
+      = GNUNET_MULTICAST_member_to_origin (slv->member, slv->chn.tmit_head->id,
                                            slave_transmit_notify, slv);
   }
   else
@@ -2090,6 +2098,9 @@ master_queue_message (struct Master *mst, struct TransmitMessage *tmit_msg)
   if (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD == tmit_msg->first_ptype)
   {
     tmit_msg->id = ++mst->max_message_id;
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                "%p master_queue_message: message_id=%" PRIu64 "\n",
+                mst, tmit_msg->id);
     struct GNUNET_PSYC_MessageMethod *pmeth
       = (struct GNUNET_PSYC_MessageMethod *) &tmit_msg[1];
 
@@ -2159,7 +2170,6 @@ queue_message (struct Channel *chn,
   memcpy (&tmit_msg[1], data, data_size);
   tmit_msg->client = client;
   tmit_msg->size = data_size;
-  tmit_msg->state = chn->tmit_state;
   tmit_msg->first_ptype = first_ptype;
   tmit_msg->last_ptype = last_ptype;
 
diff --git a/src/psyc/test_psyc.c b/src/psyc/test_psyc.c
index e2e6cfc87..1ce9074d5 100644
--- a/src/psyc/test_psyc.c
+++ b/src/psyc/test_psyc.c
@@ -225,7 +225,8 @@ master_message_part_cb (void *cls,
   if (NULL == msg)
   {
     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                "Error while receiving message %" PRIu64 "\n", message_id);
+                "Test #%d: Error while master is receiving part of message #%" PRIu64 ".\n",
+                test, message_id);
     return;
   }
 
@@ -243,7 +244,8 @@ master_message_part_cb (void *cls,
     if (GNUNET_PSYC_MESSAGE_REQUEST != flags)
     {
       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                  "Unexpected request flags: %x" PRIu32 "\n", flags);
+                  "Test #%d: Unexpected request flags: %x" PRIu32 "\n",
+                  test, flags);
       GNUNET_assert (0);
       return;
     }
@@ -297,7 +299,8 @@ slave_message_part_cb (void *cls,
   if (NULL == msg)
   {
     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                "Error while receiving message " PRIu64 "\n", message_id);
+                "Test #%d: Error while slave is receiving part of message #%" PRIu64 ".\n",
+                test, message_id);
     return;
   }
 
@@ -322,7 +325,7 @@ slave_message_part_cb (void *cls,
     {
       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                   "Test #%d: Unexpected flags for historic message: %x" PRIu32 "\n",
-                  flags);
+                  test, flags);
       GNUNET_assert (0);
       return;
     }
@@ -575,9 +578,9 @@ 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: %u bytes available, "
+              "Test #%d: Transmit notify data: %u bytes available, "
               "processing fragment %u/%u (size %u).\n",
-              *data_size, tmit->n + 1, tmit->data_count, size);
+              test, *data_size, tmit->n + 1, tmit->data_count, size);
   if (*data_size < size)
   {
     *data_size = 0;
@@ -587,7 +590,8 @@ tmit_notify_data (void *cls, uint16_t *data_size, void *data)
 
   if (GNUNET_YES != tmit->paused && 0 < tmit->data_delay[tmit->n])
   {
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Transmission paused.\n");
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "Test #%d: Transmission paused.\n", test);
     tmit->paused = GNUNET_YES;
     GNUNET_SCHEDULER_add_delayed (
       GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
@@ -611,9 +615,9 @@ tmit_notify_mod (void *cls, uint16_t *data_size, void *data, uint8_t *oper,
 {
   struct TransmitClosure *tmit = cls;
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Transmit notify modifier: %lu bytes available, "
+              "Test #%d: Transmit notify modifier: %lu bytes available, "
               "%u modifiers left to process.\n",
-              *data_size, GNUNET_ENV_environment_get_count (tmit->env));
+              test, *data_size, GNUNET_ENV_environment_get_count (tmit->env));
 
   uint16_t name_size = 0;
   size_t value_size = 0;
@@ -688,9 +692,9 @@ slave_join ();
 void
 slave_transmit ()
 {
-
-  GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Slave sending request to master.\n");
   test = TEST_SLAVE_TRANSMIT;
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "Test #%d: Slave sending request to master.\n", test);
 
   tmit = GNUNET_new (struct TransmitClosure);
   tmit->env = GNUNET_ENV_environment_create ();
@@ -772,7 +776,7 @@ join_decision_cb (void *cls,
                   const struct GNUNET_PSYC_Message *join_msg)
 {
   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-              "Slave got join decision: %d\n", is_admitted);
+              "Test #%d: Slave got join decision: %d\n", test, is_admitted);
 
   switch (test)
   {
@@ -804,8 +808,8 @@ join_request_cb (void *cls,
   struct GNUNET_HashCode slave_key_hash;
   GNUNET_CRYPTO_hash (slave_key, sizeof (*slave_key), &slave_key_hash);
   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-              "Got join request #%u from %s.\n",
-              join_req_count, GNUNET_h2s (&slave_key_hash));
+              "Test #%d: Got join request #%u from %s.\n",
+              test, join_req_count, GNUNET_h2s (&slave_key_hash));
 
   /* Reject first request */
   int is_admitted = (0 < join_req_count++) ? GNUNET_YES : GNUNET_NO;
@@ -817,8 +821,8 @@ static void
 slave_connect_cb (void *cls, int result, uint64_t max_message_id)
 {
   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-              "Slave connected: %d, max_message_id: %" PRIu64 "\n",
-              result, max_message_id);
+              "Test #%d: Slave connected: %d, max_message_id: %" PRIu64 "\n",
+              test, result, max_message_id);
   GNUNET_assert (TEST_SLAVE_JOIN_REJECT == test || TEST_SLAVE_JOIN_ACCEPT == test);
   GNUNET_assert (GNUNET_OK == result || GNUNET_NO == result);
 }
@@ -827,8 +831,8 @@ slave_connect_cb (void *cls, int result, uint64_t max_message_id)
 static void
 slave_join (int t)
 {
-  GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Joining slave.\n");
   test = t;
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Test #%d: Joining slave.\n");
 
   struct GNUNET_PeerIdentity origin = this_peer;
   struct GNUNET_ENV_Environment *env = GNUNET_ENV_environment_create ();
@@ -852,8 +856,9 @@ slave_join (int t)
 void
 master_transmit ()
 {
-  GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Master sending message to all.\n");
   test = TEST_MASTER_TRANSMIT;
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "Test #%d: Master sending message to all.\n", test);
   end_count = 0;
 
   uint32_t i, j;
@@ -907,8 +912,8 @@ void
 master_start_cb (void *cls, int result, uint64_t max_message_id)
 {
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Master started: %d, max_message_id: %" PRIu64 "\n",
-              result, max_message_id);
+              "Test #%d: Master started: %d, max_message_id: %" PRIu64 "\n",
+              test, result, max_message_id);
   GNUNET_assert (TEST_MASTER_START == test);
   GNUNET_assert (GNUNET_OK == result || GNUNET_NO == result);
   slave_join (TEST_SLAVE_JOIN_REJECT);
@@ -918,8 +923,8 @@ master_start_cb (void *cls, int result, uint64_t max_message_id)
 void
 master_start ()
 {
-  GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Starting master.\n");
   test = TEST_MASTER_START;
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Test #%d: Starting master.\n", test);
   mst = GNUNET_PSYC_master_start (cfg, channel_key, GNUNET_PSYC_CHANNEL_PRIVATE,
                                   &master_start_cb, &join_request_cb,
                                   &master_message_cb, &master_message_part_cb,