[GNUnet-SVN] [gnunet] branch master updated (d00b72f57 -> a03d291e3)

[gnunet]

This is an automated email from the git hooks/post-receive script.

bart-polot pushed a change to branch master
in repository gnunet.

    from d00b72f57 fix psyc service being unwilling to shutdown
     new 17047b7bc Implement data ack in CADET MQ API
     new dc803bb23 Only call window_size on incrementing size, allow ACK+1 
messages to be taken
     new e390e7ae2 Merge branch 'master' of git+ssh://gnunet.org/gnunet
     new 4d87fe352 - doxygen, debug text
     new ff9d5ad16 - fix assertion condition
     new 555b46223 fix cadet mq api, use proper function arguments
     new 8508f1ae2 - debug, indentation, legibility
     new 1aa3a2b2b - code/debug format
     new 3cfcfa48b - adjust FWD/BCK to new channel numbering
     new ec05b7e9c Fix open port, copy handlers for use on incoming channels
     new a03d291e3 Add mq test

The 11 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails.  The revisions
listed as "add" were already present in the repository and have only
been added to this reference.


Summary of changes:
 src/cadet/Makefile.am              |  18 +-
 src/cadet/cadet_api.c              | 130 ++++++++++++---
 src/cadet/test_cadet_local_mq.c    | 331 +++++++++++++++++++++++++++++++++++++
 src/include/gnunet_cadet_service.h |   3 -
 4 files changed, 455 insertions(+), 27 deletions(-)
 create mode 100644 src/cadet/test_cadet_local_mq.c

diff --git a/src/cadet/Makefile.am b/src/cadet/Makefile.am
index a5416c653..82fb7351b 100644
--- a/src/cadet/Makefile.am
+++ b/src/cadet/Makefile.am
@@ -110,11 +110,7 @@ libgnunetcadettest_a_LIBADD = \
 
 if HAVE_TESTING
 check_PROGRAMS = \
-  test_cadet_2_speed_reliable_backwards \
-  test_cadet_5_speed \
-  test_cadet_5_speed_ack \
-  test_cadet_5_speed_reliable \
-  test_cadet_5_speed_reliable_backwards \
+  test_cadet_local_mq \
   test_cadet_single \
   test_cadet_local \
   test_cadet_2_forward \
@@ -124,9 +120,14 @@ check_PROGRAMS = \
   test_cadet_2_speed_ack \
   test_cadet_2_speed_backwards \
   test_cadet_2_speed_reliable \
+  test_cadet_2_speed_reliable_backwards \
   test_cadet_5_forward \
   test_cadet_5_signal \
   test_cadet_5_keepalive \
+  test_cadet_5_speed \
+  test_cadet_5_speed_ack \
+  test_cadet_5_speed_reliable \
+  test_cadet_5_speed_reliable_backwards \
   test_cadet_5_speed_backwards
 endif
 
@@ -158,6 +159,13 @@ test_cadet_local_SOURCES = \
 test_cadet_local_LDADD = $(ld_cadet_test_lib)
 
 
+test_cadet_local_mq_SOURCES = \
+  test_cadet_local_mq.c
+test_cadet_local_mq_LDADD = \
+  libgnunetcadet.la \
+  $(top_builddir)/src/testing/libgnunettesting.la \
+  $(top_builddir)/src/util/libgnunetutil.la
+
 test_cadet_2_forward_SOURCES = \
   test_cadet.c
 test_cadet_2_forward_LDADD = $(ld_cadet_test_lib)
diff --git a/src/cadet/cadet_api.c b/src/cadet/cadet_api.c
index 7640a924a..7b9ac62b3 100644
--- a/src/cadet/cadet_api.c
+++ b/src/cadet/cadet_api.c
@@ -291,6 +291,16 @@ struct GNUNET_CADET_Channel
   struct GNUNET_MQ_Handle *mq;
 
   /**
+   * Task to allow mq to send more traffic.
+   */
+  struct GNUNET_SCHEDULER_Task *mq_cont;
+
+  /**
+   * Pending envelope in case we don't have an ACK from the service.
+   */
+  struct GNUNET_MQ_Envelope *pending_env;
+
+  /**
    * Window change handler.
    */
   GNUNET_CADET_WindowSizeEventHandler window_changes;
@@ -315,6 +325,8 @@ struct GNUNET_CADET_Port
 
   /**
    * Port ID.
+   *
+   * @deprecated
    */
   struct GNUNET_HashCode *hash;
 
@@ -547,7 +559,11 @@ destroy_channel (struct GNUNET_CADET_Channel *ch)
   GNUNET_CONTAINER_DLL_remove (h->channels_head,
                                h->channels_tail,
                                ch);
-
+  if (NULL != ch->mq_cont)
+  {
+    GNUNET_SCHEDULER_cancel (ch->mq_cont);
+    ch->mq_cont = NULL;
+  }
   /* signal channel destruction */
   if (0 != ch->peer)
   {
@@ -630,10 +646,37 @@ remove_from_queue (struct GNUNET_CADET_TransmitHandle *th)
 }
 
 
+/**
+ * Notify the application about a change in the window size (if needed).
+ *
+ * @param ch Channel to notify about.
+ */
+static void
+notify_window_size (struct GNUNET_CADET_Channel *ch)
+{
+  if (NULL != ch->window_changes)
+  {
+    ch->window_changes (ch->ctx, ch, ch->allow_send);
+  }
+}
+
 
/******************************************************************************/
 /***********************      MQ API CALLBACKS     
****************************/
 
/******************************************************************************/
 
+/**
+ * Allow the MQ implementation to send the next message.
+ *
+ * @param cls Closure (channel whose mq to activate).
+ */
+static void
+cadet_mq_send_continue (void *cls)
+{
+  struct GNUNET_CADET_Channel *ch = cls;
+
+  ch->mq_cont = NULL;
+  GNUNET_MQ_impl_send_continue (ch->mq);
+}
 
 /**
  * Implement sending functionality of a message queue for
@@ -679,8 +722,21 @@ cadet_mq_send_impl (struct GNUNET_MQ_Handle *mq,
                                  GNUNET_MESSAGE_TYPE_CADET_LOCAL_DATA,
                                  msg);
   cadet_msg->ccn = ch->ccn;
-  GNUNET_MQ_send (h->mq, env);
-  GNUNET_MQ_impl_send_continue (mq);
+
+  if (0 < ch->allow_send)
+  {
+    /* Service has allowed this message, just send it and continue accepting */
+    GNUNET_MQ_send (h->mq, env);
+    ch->allow_send--;
+    ch->mq_cont = GNUNET_SCHEDULER_add_now (&cadet_mq_send_continue, ch);
+    // notify_window_size (ch); /* FIXME add "verbose" setting? */
+  }
+  else
+  {
+    /* Service has NOT allowed this message, queue it and wait for an ACK */
+    GNUNET_assert (NULL == ch->pending_env);
+    ch->pending_env = env;
+  }
 }
 
 
@@ -725,6 +781,7 @@ cadet_mq_error_handler (void *cls,
  * @param mq message queue
  * @param impl_state state specific to the implementation
  */
+
 static void
 cadet_mq_cancel_impl (struct GNUNET_MQ_Handle *mq,
                      void *impl_state)
@@ -849,11 +906,13 @@ handle_channel_created (void *cls,
     /** @deprecated */
     /* Old style API */
     ch->ctx = port->handler (port->cls,
-                            ch,
-                            &msg->peer,
-                            port->hash,
-                            ch->options);
-  } else {
+                             ch,
+                             &msg->peer,
+                             port->hash,
+                             ch->options);
+  }
+  else
+  {
     /* MQ API */
     GNUNET_assert (NULL != port->connects);
     ch->window_changes = port->window_changes;
@@ -865,7 +924,7 @@ handle_channel_created (void *cls,
                                             port->handlers,
                                             &cadet_mq_error_handler,
                                             ch);
-    ch->ctx = port->connects (port->cadet->cls,
+    ch->ctx = port->connects (port->cls,
                               ch,
                               &msg->peer);
     GNUNET_MQ_set_handlers_closure (ch->mq, ch->ctx);
@@ -955,6 +1014,7 @@ handle_local_data (void *cls,
   const struct GNUNET_CADET_MessageHandler *handler;
   struct GNUNET_CADET_Channel *ch;
   uint16_t type;
+  int fwd;
 
   ch = retrieve_channel (h,
                          message->ccn);
@@ -967,16 +1027,20 @@ handle_local_data (void *cls,
 
   payload = (struct GNUNET_MessageHeader *) &message[1];
   type = ntohs (payload->type);
+  fwd = ntohl (ch->ccn.channel_of_client) <= GNUNET_CADET_LOCAL_CHANNEL_ID_CLI;
   LOG (GNUNET_ERROR_TYPE_DEBUG,
        "Got a %s data on channel %s [%X] of type %s (%u)\n",
-       GC_f2s (ntohl (ch->ccn.channel_of_client) >=
-               GNUNET_CADET_LOCAL_CHANNEL_ID_CLI),
+       GC_f2s (fwd),
        GNUNET_i2s (GNUNET_PEER_resolve2 (ch->peer)),
        ntohl (message->ccn.channel_of_client),
        GC_m2s (type),
        type);
   if (NULL != ch->mq)
   {
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "injecting msg %s into mq %p\n",
+         GC_m2s (ntohs (payload->type)),
+         ch->mq);
     GNUNET_MQ_inject_message (ch->mq, payload);
     return;
   }
@@ -1012,8 +1076,6 @@ handle_local_data (void *cls,
  *
  * @param h Cadet handle.
  * @param message Message itself.
- *
- * FIXME either delete or port to MQ
  */
 static void
 handle_local_ack (void *cls,
@@ -1034,10 +1096,31 @@ handle_local_ack (void *cls,
     return;
   }
   ch->allow_send++;
-  LOG (GNUNET_ERROR_TYPE_DEBUG,
-       "Got an ACK on channel %X, allow send now %u!\n",
-       ntohl (ch->ccn.channel_of_client),
-       ch->allow_send);
+  if (NULL != ch->mq)
+  {
+    if (NULL == ch->pending_env)
+    {
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
+           "Got an ACK on mq channel %X, allow send now %u!\n",
+           ntohl (ch->ccn.channel_of_client),
+           ch->allow_send);
+      notify_window_size (ch);
+    }
+    else
+    {
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
+           "Got an ACK on mq channel %X, sending pending message!\n",
+           ntohl (ch->ccn.channel_of_client));
+      GNUNET_MQ_send (h->mq, ch->pending_env);
+      ch->allow_send--;
+      ch->pending_env = NULL;
+      ch->mq_cont = GNUNET_SCHEDULER_add_now (&cadet_mq_send_continue, ch);
+    }
+    return;
+  }
+
+  /** @deprecated */
+  /* Old style API */
   for (th = h->th_head; NULL != th; th = th->next)
   {
     if ( (th->channel == ch) &&
@@ -2455,11 +2538,20 @@ GNUNET_CADET_open_porT (struct GNUNET_CADET_Handle *h,
   p->cls = connects_cls;
   p->window_changes = window_changes;
   p->disconnects = disconnects;
-  p->handlers = handlers;
+  if (NULL != handlers)
+  {
+    unsigned int i;
+    for (i=0;NULL != handlers[i].cb; i++) ;
+    p->handlers = GNUNET_new_array (i + 1,
+                                    struct GNUNET_MQ_MessageHandler);
+    GNUNET_memcpy ((struct GNUNET_MQ_MessageHandler *) p->handlers,
+                   handlers,
+                   i * sizeof (struct GNUNET_MQ_MessageHandler));
+  }
 
   GNUNET_assert (GNUNET_OK ==
                 GNUNET_CONTAINER_multihashmap_put (h->ports,
-                                                   p->hash,
+                                                   &p->id,
                                                    p,
                                                    
GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 
diff --git a/src/cadet/test_cadet_local_mq.c b/src/cadet/test_cadet_local_mq.c
new file mode 100644
index 000000000..988ec725e
--- /dev/null
+++ b/src/cadet/test_cadet_local_mq.c
@@ -0,0 +1,331 @@
+/*
+     This file is part of GNUnet.
+     Copyright (C) 2011 GNUnet e.V.
+
+     GNUnet is free software; you can redistribute it and/or modify
+     it under the terms of the GNU General Public License as published
+     by the Free Software Foundation; either version 3, or (at your
+     option) any later version.
+
+     GNUnet is distributed in the hope that it will be useful, but
+     WITHOUT ANY WARRANTY; without even the implied warranty of
+     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+     General Public License for more details.
+
+     You should have received a copy of the GNU General Public License
+     along with GNUnet; see the file COPYING.  If not, write to the
+     Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
+     Boston, MA 02110-1301, USA.
+*/
+
+/**
+ * @file cadet/test_cadet_local.c
+ * @brief test cadet local: test of cadet channels with just one peer
+ * @author Bartlomiej Polot
+ */
+
+#include "platform.h"
+#include "gnunet_util_lib.h"
+#include "gnunet_dht_service.h"
+#include "gnunet_testing_lib.h"
+#include "gnunet_cadet_service.h"
+
+#define TEST_MESSAGE_TYPE 1
+#define TEST_PORT_ID 1
+
+/**
+ * Test message structure.
+ */
+struct GNUNET_CADET_TestMsg
+{
+  /**
+   * Type: #TEST_MESSAGE_TYPE
+   *
+   * Size: sizeof(struct GNUNET_CADET_TestMsg)
+   */
+  struct GNUNET_MessageHeader header;
+
+  /**
+   * Test payload.
+   */
+  uint64_t payload;
+};
+
+struct GNUNET_TESTING_Peer *me;
+
+static struct GNUNET_CADET_Handle *cadet_peer_1;
+
+static struct GNUNET_CADET_Handle *cadet_peer_2;
+
+static struct GNUNET_CADET_Channel *ch;
+
+static int result = GNUNET_OK;
+
+static int got_data = GNUNET_NO;
+
+static struct GNUNET_SCHEDULER_Task *abort_task;
+
+static struct GNUNET_SCHEDULER_Task *connect_task;
+
+
+/**
+ * Connect to other client and send data
+ *
+ * @param cls Closue (unused).
+ */
+static void
+do_connect (void *cls);
+
+
+/**
+ * Shutdown nicely
+ */
+static void
+do_shutdown (void *cls)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "shutdown\n");
+  if (NULL != abort_task)
+  {
+    GNUNET_SCHEDULER_cancel (abort_task);
+    abort_task = NULL;
+  }
+  if (NULL != ch)
+  {
+    GNUNET_CADET_channel_destroy (ch);
+    ch = NULL;
+  }
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Disconnect client 1\n");
+  if (NULL != cadet_peer_1)
+  {
+    GNUNET_CADET_disconnect (cadet_peer_1);
+    cadet_peer_1 = NULL;
+  }
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Disconnect client 2\n");
+  if (NULL != cadet_peer_2)
+  {
+    GNUNET_CADET_disconnect (cadet_peer_2);
+    cadet_peer_2 = NULL;
+  }
+  if (NULL != connect_task)
+  {
+    GNUNET_SCHEDULER_cancel (connect_task);
+    connect_task = NULL;
+  }
+}
+
+
+/**
+ * Something went wrong and timed out. Kill everything and set error flag
+ */
+static void
+do_abort (void *cls)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "ABORT from line %ld\n", (long) cls);
+  result = GNUNET_SYSERR;
+  abort_task = NULL;
+  GNUNET_SCHEDULER_shutdown ();
+}
+
+/**
+ * Method called whenever a peer connects to a port in MQ-based CADET.
+ *
+ * @param cls Closure from #GNUNET_CADET_open_porT.
+ * @param channel New handle to the channel.
+ * @param source Peer that started this channel.
+ * @return Closure for the incoming @a channel. It's given to:
+ *         - The #GNUNET_CADET_DisconnectEventHandler (given to
+ *           #GNUNET_CADET_open_porT) when the channel dies.
+ *         - Each the #GNUNET_MQ_MessageCallback handlers for each message
+ *           received on the @a channel.
+ */
+static void *
+connected (void *cls,
+           struct GNUNET_CADET_Channel *channel,
+           const struct GNUNET_PeerIdentity *source)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "connected %s, cls: %p\n",
+              GNUNET_i2s(source), cls);
+  return channel;
+}
+
+/**
+ * Function called whenever an MQ-channel is destroyed, even if the destruction
+ * was requested by #GNUNET_CADET_channel_destroy.
+ * It must NOT call #GNUNET_CADET_channel_destroy on the channel.
+ *
+ * It should clean up any associated state, including cancelling any pending
+ * transmission on this channel.
+ *
+ * @param cls Channel closure.
+ * @param channel Connection to the other end (henceforth invalid).
+ */
+static void
+disconnected (void *cls,
+              const struct GNUNET_CADET_Channel *channel)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "disconnected cls: %p\n",
+              cls);
+  if (channel == ch)
+    ch = NULL;
+}
+
+
+/**
+ * Handle test data
+ *
+ * @param h     The cadet handle
+ * @param msg   A message with the details of the new incoming channel
+ */
+static void
+handle_data_received (void *cls,
+                      const struct GNUNET_CADET_TestMsg *msg)
+{
+  uint64_t payload;
+
+  payload = GNUNET_ntohll (msg->payload);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+             "Data callback payload %lu with cls: %p! Shutting down.\n",
+              payload, cls);
+  GNUNET_assert (42 == payload);
+  got_data = GNUNET_YES;
+  GNUNET_SCHEDULER_shutdown ();
+}
+
+
+/**
+ * Signature of the main function of a task.
+ *
+ * @param cls Closure (unused).
+ */
+static void
+message_sent (void *cls)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "message sent\n");
+}
+
+
+/**
+ * Connect to other client and send data
+ *
+ * @param cls Closure (unused).
+ */
+static void
+do_connect (void *cls)
+{
+  struct GNUNET_PeerIdentity id;
+  struct GNUNET_MQ_Handle *mq;
+  struct GNUNET_MQ_Envelope *env;
+  struct GNUNET_CADET_TestMsg *msg;
+
+  struct GNUNET_MQ_MessageHandler handlers[] = {
+    GNUNET_MQ_hd_fixed_size (data_received,
+                             TEST_MESSAGE_TYPE,
+                             struct GNUNET_CADET_TestMsg,
+                             cadet_peer_1),
+    GNUNET_MQ_handler_end ()
+  };
+
+  connect_task = NULL;
+  GNUNET_TESTING_peer_get_identity (me, &id);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "creating channel\n");
+  ch = GNUNET_CADET_channel_creatE (cadet_peer_1, /* cadet handle */
+                                    NULL,         /* channel cls */
+                                    &id,          /* destination */
+                                    GC_u2h (TEST_MESSAGE_TYPE), /* port */
+                                    GNUNET_CADET_OPTION_DEFAULT, /* opt */
+                                    NULL,          /* window change */
+                                    &disconnected, /* disconnect handler */
+                                    handlers       /* traffic handlers */
+                                   );
+  env = GNUNET_MQ_msg (msg, TEST_MESSAGE_TYPE);
+  msg->payload = GNUNET_htonll (42);
+  mq = GNUNET_CADET_get_mq (ch);
+  GNUNET_MQ_notify_sent (env, &message_sent, NULL);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "sending message\n");
+  GNUNET_MQ_send (mq, env);
+}
+
+
+/**
+ * Initialize framework and start test
+ *
+ * @param cls Closure (unused).
+ * @param cfg Configuration handle.
+ * @param peer Testing peer handle.
+ */
+static void
+run (void *cls,
+     const struct GNUNET_CONFIGURATION_Handle *cfg,
+     struct GNUNET_TESTING_Peer *peer)
+{
+  struct GNUNET_TIME_Relative start_delay;
+  me = peer;
+  GNUNET_SCHEDULER_add_shutdown (&do_shutdown,
+                                 NULL);
+  abort_task =
+      GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
+                                    (GNUNET_TIME_UNIT_SECONDS, 15),
+                                   &do_abort,
+                                    (void *) (long) __LINE__);
+  cadet_peer_1 = GNUNET_CADET_connecT (cfg);
+  cadet_peer_2 = GNUNET_CADET_connecT (cfg);
+
+  if ( (NULL == cadet_peer_1) ||
+       (NULL == cadet_peer_2) )
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+               "Couldn't connect to cadet\n");
+    result = GNUNET_SYSERR;
+    GNUNET_SCHEDULER_shutdown ();
+    return;
+  }
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "CADET 1: %p\n", cadet_peer_1);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "CADET 2: %p\n", cadet_peer_2);
+  struct GNUNET_MQ_MessageHandler handlers[] = {
+    GNUNET_MQ_hd_fixed_size (data_received,
+                             TEST_MESSAGE_TYPE,
+                             struct GNUNET_CADET_TestMsg,
+                             cadet_peer_2),
+    GNUNET_MQ_handler_end ()
+  };
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "handlers 2: %p\n", handlers);
+  GNUNET_CADET_open_porT (cadet_peer_2,          /* cadet handle */
+                          GC_u2h (TEST_PORT_ID), /* port id */
+                          &connected,            /* connect handler */
+                          (void *) 2L,           /* handle for #connected */
+                          NULL,                  /* window size handler */
+                          &disconnected,         /* disconnect handler */
+                          handlers);             /* traffic handlers */
+  start_delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 2);
+  if (NULL == connect_task)
+    connect_task = GNUNET_SCHEDULER_add_delayed (start_delay,
+                                                 &do_connect,
+                                                 NULL);
+}
+
+
+/**
+ * Main
+ */
+int
+main (int argc, char *argv[])
+{
+  if (0 != GNUNET_TESTING_peer_run ("test-cadet-local",
+                                    "test_cadet.conf",
+                                &run, NULL))
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "run failed\n");
+    return 2;
+  }
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Final result: %d\n", result);
+  return (result == GNUNET_OK) ? 0 : 1;
+}
+
+/* end of test_cadet_local_1.c */
diff --git a/src/include/gnunet_cadet_service.h 
b/src/include/gnunet_cadet_service.h
index 1434180f4..1b3aac7c9 100644
--- a/src/include/gnunet_cadet_service.h
+++ b/src/include/gnunet_cadet_service.h
@@ -702,9 +702,6 @@ GC_u2h (uint32_t port);
  * @param cls Closure from #GNUNET_CADET_open_porT.
  * @param channel New handle to the channel.
  * @param source Peer that started this channel.
- * FIXME: Add port that this channel is created for, or is cls enough?
- *        Port cannot be closed yet, #handle_channel_create would have
- *        rejected it.
  * @return Closure for the incoming @a channel. It's given to:
  *         - The #GNUNET_CADET_DisconnectEventHandler (given to
  *           #GNUNET_CADET_open_porT) when the channel dies.

-- 
To stop receiving notification emails like this one, please contact
address@hidden