/*
This file is part of GNUnet.
- (C) 2008--2013 Christian Grothoff (and other contributing authors)
+ Copyright (C) 2008--2013, 2016 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
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., 59 Temple Place - Suite 330,
- Boston, MA 02111-1307, USA.
+ Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
+ Boston, MA 02110-1301, USA.
*/
/**
* @file testbed/testbed_api_barriers.c
* @brief API implementation for testbed barriers
- * @author Sree Harsha Totakura <sreeharsha@totakura.in>
+ * @author Sree Harsha Totakura <sreeharsha@totakura.in>
*/
-
#include "platform.h"
#include "gnunet_testbed_service.h"
#include "testbed_api.h"
#define LOG_DEBUG(...) \
LOG (GNUNET_ERROR_TYPE_DEBUG, __VA_ARGS__);
-/**
- * Handle for barrier
- */
-struct GNUNET_TESTBED_Barrier
-{
- /**
- * hashcode identifying this barrier in the hashmap
- */
- struct GNUNET_HashCode key;
-
- /**
- * The controller handle given while initiliasing this barrier
- */
- struct GNUNET_TESTBED_Controller *c;
-
- /**
- * The name of the barrier
- */
- char *name;
-
- /**
- * The continuation callback to call when we have a status update on this
- */
- GNUNET_TESTBED_barrier_status_cb cb;
-
- /**
- * the closure for the above callback
- */
- void *cls;
-
-};
-
-
-/**
- * handle for hashtable of barrier handles
- */
-static struct GNUNET_CONTAINER_MultiHashMap *barrier_map;
-
-
-/**
- * Remove a barrier and it was the last one in the barrier hash map, destroy the
- * hash map
- *
- * @param barrier the barrier to remove
- */
-static void
-barrier_remove (struct GNUNET_TESTBED_Barrier *barrier)
-{
- GNUNET_assert (NULL != barrier_map); /* No barriers present */
- GNUNET_assert (GNUNET_OK ==
- GNUNET_CONTAINER_multihashmap_remove (barrier_map,
- &barrier->key,
- barrier));
- GNUNET_free (barrier->name);
- GNUNET_free (barrier);
- if (0 == GNUNET_CONTAINER_multihashmap_size (barrier_map))
- {
- GNUNET_CONTAINER_multihashmap_destroy (barrier_map);
- barrier_map = NULL;
- }
-}
-
-
-/**
- * Handler for GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_STATUS messages
- *
- * @param c the controller handle to determine the connection this message
- * belongs to
- * @param msg the barrier status message
- * @return GNUNET_OK to keep the connection active; GNUNET_SYSERR to tear it
- * down signalling an error
- */
-int
-GNUNET_TESTBED_handle_barrier_status_ (struct GNUNET_TESTBED_Controller *c,
- const struct GNUNET_TESTBED_BarrierStatusMsg
- *msg)
-{
- struct GNUNET_TESTBED_Barrier *barrier;
- char *emsg;
- const char *name;
- struct GNUNET_HashCode key;
- size_t emsg_len;
- int status;
- uint16_t msize;
- uint16_t name_len;
-
- emsg = NULL;
- barrier = NULL;
- msize = ntohs (msg->header.size);
- name = msg->data;
- name_len = ntohs (msg->name_len);
- LOG_DEBUG ("Received BARRIER_STATUS msg\n");
- if (sizeof (struct GNUNET_TESTBED_BarrierStatusMsg) + name_len + 1 > msize)
- {
- GNUNET_break_op (0);
- return GNUNET_SYSERR;
- }
- if ('\0' != name[name_len])
- {
- GNUNET_break_op (0);
- return GNUNET_SYSERR;
- }
- status = ntohs (msg->status);
- if (BARRIER_STATUS_ERROR == status)
- {
- status = -1;
- emsg_len = msize - (sizeof (struct GNUNET_TESTBED_BarrierStatusMsg) + name_len
- + 1);
- if (0 == emsg_len)
- {
- GNUNET_break_op (0);
- return GNUNET_SYSERR;
- }
- emsg_len++;
- emsg = GNUNET_malloc (emsg_len);
- emsg_len--;
- emsg[emsg_len] = '\0';
- (void) memcpy (emsg, msg->data + name_len + 1, emsg_len);
- }
- if (NULL == barrier_map)
- {
- GNUNET_break_op (0);
- goto cleanup;
- }
- GNUNET_CRYPTO_hash (name, name_len, &key);
- barrier = GNUNET_CONTAINER_multihashmap_get (barrier_map, &key);
- if (NULL == barrier)
- {
- GNUNET_break_op (0);
- goto cleanup;
- }
- GNUNET_assert (NULL != barrier->cb);
- barrier->cb (barrier->cls, name, barrier, status, emsg);
- if (BARRIER_STATUS_INITIALISED == status)
- return GNUNET_OK; /* just initialised; skip cleanup */
-
- cleanup:
- GNUNET_free_non_null (emsg);
- if (NULL != barrier)
- barrier_remove (barrier);
- return GNUNET_OK;
-}
-
-
-/**
- * Initialise a barrier and call the given callback when the required percentage
- * of peers (quorum) reach the barrier OR upon error.
- *
- * @param controller the handle to the controller
- * @param name identification name of the barrier
- * @param quorum the percentage of peers that is required to reach the barrier.
- * Peers signal reaching a barrier by calling
- * GNUNET_TESTBED_barrier_reached().
- * @param cb the callback to call when the barrier is reached or upon error.
- * Cannot be NULL.
- * @param cls closure for the above callback
- * @return barrier handle; NULL upon error
- */
-struct GNUNET_TESTBED_Barrier *
-GNUNET_TESTBED_barrier_init (struct GNUNET_TESTBED_Controller *controller,
- const char *name,
- unsigned int quorum,
- GNUNET_TESTBED_barrier_status_cb cb, void *cls)
-{
- struct GNUNET_TESTBED_BarrierInit *msg;
- struct GNUNET_TESTBED_Barrier *barrier;
- struct GNUNET_HashCode key;
- size_t name_len;
- uint16_t msize;
-
- GNUNET_assert (quorum <= 100);
- GNUNET_assert (NULL != cb);
- name_len = strlen (name);
- GNUNET_assert (0 < name_len);
- GNUNET_CRYPTO_hash (name, name_len, &key);
- if (NULL == barrier_map)
- barrier_map = GNUNET_CONTAINER_multihashmap_create (3, GNUNET_YES);
- if (GNUNET_YES ==
- GNUNET_CONTAINER_multihashmap_contains (barrier_map, &key))
- {
- GNUNET_break (0);
- return NULL;
- }
- LOG_DEBUG ("Initialising barrier `%s'\n", name);
- barrier = GNUNET_malloc (sizeof (struct GNUNET_TESTBED_Barrier));
- barrier->c = controller;
- barrier->name = GNUNET_strdup (name);
- barrier->cb = cb;
- barrier->cls = cls;
- (void) memcpy (&barrier->key, &key, sizeof (struct GNUNET_HashCode));
- GNUNET_assert (GNUNET_OK ==
- GNUNET_CONTAINER_multihashmap_put (barrier_map, &barrier->key,
- barrier,
- GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
- msize = name_len + sizeof (struct GNUNET_TESTBED_BarrierInit);
- msg = GNUNET_malloc (msize);
- msg->header.size = htons (msize);
- msg->header.type = htons (GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_INIT);
- msg->quorum = (uint8_t) quorum;
- (void) memcpy (msg->name, barrier->name, name_len);
- GNUNET_TESTBED_queue_message_ (barrier->c, &msg->header);
- return barrier;
-}
-
-
-/**
- * Cancel a barrier.
- *
- * @param barrier the barrier handle
- */
-void
-GNUNET_TESTBED_barrier_cancel (struct GNUNET_TESTBED_Barrier *barrier)
-{
- struct GNUNET_TESTBED_BarrierCancel *msg;
- uint16_t msize;
-
- msize = sizeof (struct GNUNET_TESTBED_BarrierCancel) + strlen (barrier->name);
- msg = GNUNET_malloc (msize);
- msg->header.size = htons (msize);
- msg->header.type = htons (GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_CANCEL);
- (void) memcpy (msg->name, barrier->name, strlen (barrier->name));
- GNUNET_TESTBED_queue_message_ (barrier->c, &msg->header);
- barrier_remove (barrier);
-}
-
/**
* Barrier wait handle
* The name of the barrier
*/
char *name;
-
- /**
- * Then configuration used for the client connection
- */
- struct GNUNET_CONFIGURATION_Handle *cfg;
-
- /**
- * The client connection
- */
- struct GNUNET_CLIENT_Connection *conn;
/**
- * Transmit handle
+ * Then configuration used for the client connection
*/
- struct GNUNET_CLIENT_TransmitHandle *tx;
+ struct GNUNET_CONFIGURATION_Handle *cfg;
/**
- * The message to transmit with tx
+ * The testbed-barrier service message queue.
*/
- struct GNUNET_MessageHeader *msg;
+ struct GNUNET_MQ_Handle *mq;
/**
* The barrier wait callback
GNUNET_TESTBED_barrier_wait_cb cb;
/**
- * The closure for the above callback
+ * The closure for @e cb.
*/
- void *cls;
+ void *cb_cls;
};
+
/**
- * Function to destroy barrier wait handle
+ * Check if barrier status message is well-formed.
*
- * @param h the handle to destroy
+ * @param cls closure
+ * @param msg received message
+ * @return #GNUNET_OK if the message is well-formed.
*/
-static void
-destroy_handle (struct GNUNET_TESTBED_BarrierWaitHandle *h)
+static int
+check_status (void *cls,
+ const struct GNUNET_TESTBED_BarrierStatusMsg *msg)
{
- GNUNET_free (h->name);
- if (NULL != h->tx)
- GNUNET_CLIENT_notify_transmit_ready_cancel (h->tx);
- if (NULL != h->conn)
- GNUNET_CLIENT_disconnect (h->conn);
- if (NULL != h->msg)
- GNUNET_free (h->msg);
- GNUNET_CONFIGURATION_destroy (h->cfg);
- GNUNET_free (h);
+ /* FIXME: this fails to actually check that the message
+ follows the protocol spec (0-terminations!). However,
+ not critical as #handle_status() doesn't interpret the
+ variable-size part anyway right now. */
+ return GNUNET_OK;
}
* from the service.
*
* @param cls closure
- * @param message received message; NULL on timeout or fatal error
+ * @param msg received message
*/
static void
-receive_handler (void *cls, const struct GNUNET_MessageHeader *message)
+handle_status (void *cls,
+ const struct GNUNET_TESTBED_BarrierStatusMsg *msg)
{
struct GNUNET_TESTBED_BarrierWaitHandle *h = cls;
- const struct GNUNET_TESTBED_BarrierStatusMsg *msg;
- uint16_t msize;
-
- if (GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_STATUS != ntohs (message->type))
- {
- GNUNET_break_op (0);
- goto fail;
- }
- msize = ntohs (message->size);
- if (msize <= sizeof (struct GNUNET_TESTBED_BarrierStatusMsg))
- {
- GNUNET_break_op (0);
- goto fail;
- }
- msg = (const struct GNUNET_TESTBED_BarrierStatusMsg *) message;
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Got barrier status %d\n",
+ (int) ntohs (msg->status));
switch (ntohs (msg->status))
{
- case BARRIER_STATUS_ERROR:
- goto fail;
- case BARRIER_STATUS_INITIALISED:
- GNUNET_break (0); /* FIXME */
- goto destroy;
- case BARRIER_STATUS_CROSSED:
- h->cb (h->cls, h->name, GNUNET_OK);
- goto destroy;
+ case GNUNET_TESTBED_BARRIERSTATUS_ERROR:
+ h->cb (h->cb_cls,
+ h->name,
+ GNUNET_SYSERR);
+ break;
+ case GNUNET_TESTBED_BARRIERSTATUS_INITIALISED:
+ h->cb (h->cb_cls,
+ h->name,
+ GNUNET_SYSERR);
+ GNUNET_break (0);
+ break;
+ case GNUNET_TESTBED_BARRIERSTATUS_CROSSED:
+ h->cb (h->cb_cls,
+ h->name,
+ GNUNET_OK);
+ break;
default:
GNUNET_break_op (0);
+ h->cb (h->cb_cls,
+ h->name,
+ GNUNET_SYSERR);
+ break;
}
-
- fail:
- h->cb (h->cls, h->name, GNUNET_SYSERR);
-
- destroy:
- destroy_handle (h);
+ GNUNET_TESTBED_barrier_wait_cancel (h);
}
/**
- * Function called to notify a client about the connection
- * begin ready to queue more data. "buf" will be
- * NULL and "size" zero if the connection was closed for
- * writing in the meantime.
+ * Generic error handler, called with the appropriate error code and
+ * the same closure specified at the creation of the message queue.
+ * Not every message queue implementation supports an error handler.
*
- * @param cls closure
- * @param size number of bytes available in buf
- * @param buf where the callee should write the message
- * @return number of bytes written to buf
+ * @param cls closure with the `struct GNUNET_TESTBED_BarrierWaitHandle *`
+ * @param error error code
*/
-static size_t
-transmit_notify (void *cls, size_t size, void *buf)
+static void
+mq_error_handler (void *cls,
+ enum GNUNET_MQ_Error error)
{
struct GNUNET_TESTBED_BarrierWaitHandle *h = cls;
- uint16_t msize;
-
- h->tx = NULL;
- if ((0 == size) || (NULL == buf))
- {
- destroy_handle (h);
- return 0;
- }
- msize = htons (h->msg->size);
- GNUNET_assert (msize <= size);
- (void) memcpy (buf, h->msg, msize);
- GNUNET_free (h->msg);
- h->msg = NULL;
- GNUNET_CLIENT_receive (h->conn, &receive_handler, h, GNUNET_TIME_UNIT_FOREVER_REL);
- return msize;
+
+ h->cb (h->cb_cls,
+ h->name,
+ GNUNET_SYSERR);
+ GNUNET_TESTBED_barrier_wait_cancel (h);
}
*
* @param name the name of the barrier
* @param cb the barrier wait callback
- * @param cls the closure for the above callback
+ * @param cb_cls the closure for @a cb
* @return barrier wait handle which can be used to cancel the waiting at
* anytime before the callback is called. NULL upon error.
*/
struct GNUNET_TESTBED_BarrierWaitHandle *
GNUNET_TESTBED_barrier_wait (const char *name,
GNUNET_TESTBED_barrier_wait_cb cb,
- void *cls)
+ void *cb_cls)
{
+ struct GNUNET_TESTBED_BarrierWaitHandle *h
+ = GNUNET_new (struct GNUNET_TESTBED_BarrierWaitHandle);
+ struct GNUNET_MQ_MessageHandler handlers[] = {
+ GNUNET_MQ_hd_var_size (status,
+ GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_STATUS,
+ struct GNUNET_TESTBED_BarrierStatusMsg,
+ h),
+ GNUNET_MQ_handler_end ()
+ };
+ struct GNUNET_MQ_Envelope *env;
struct GNUNET_TESTBED_BarrierWait *msg;
- struct GNUNET_CONFIGURATION_Handle *cfg;
- struct GNUNET_TESTBED_BarrierWaitHandle *h;
- char *cfg_filename;
+ const char *cfg_filename;
size_t name_len;
- uint16_t msize;
GNUNET_assert (NULL != cb);
- GNUNET_assert (NULL != name);
cfg_filename = getenv (ENV_TESTBED_CONFIG);
if (NULL == cfg_filename)
{
- LOG (GNUNET_ERROR_TYPE_ERROR, "Are you running under testbed?\n");
+ LOG (GNUNET_ERROR_TYPE_ERROR,
+ "Are you running under testbed?\n");
+ GNUNET_free (h);
return NULL;
}
- cfg = GNUNET_CONFIGURATION_create ();
- if (GNUNET_OK != GNUNET_CONFIGURATION_load (cfg, cfg_filename))
+ h->cfg = GNUNET_CONFIGURATION_create ();
+ if (GNUNET_OK !=
+ GNUNET_CONFIGURATION_load (h->cfg,
+ cfg_filename))
{
- LOG (GNUNET_ERROR_TYPE_ERROR, "Unable to load configuration from file `%s'\n",
+ LOG (GNUNET_ERROR_TYPE_ERROR,
+ "Unable to load configuration from file `%s'\n",
cfg_filename);
- GNUNET_CONFIGURATION_destroy (cfg);
+ GNUNET_CONFIGURATION_destroy (h->cfg);
+ GNUNET_free (h);
return NULL;
}
- h = GNUNET_malloc (sizeof (struct GNUNET_TESTBED_BarrierWaitHandle));
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Waiting on barrier `%s'\n",
+ name);
h->name = GNUNET_strdup (name);
- h->cfg = cfg;
- h->conn = GNUNET_CLIENT_connect ("testbed-barrier", h->cfg);
h->cb = cb;
- h->cls = cls;
- if (NULL == h->conn)
+ h->cb_cls = cb_cls;
+ h->mq = GNUNET_CLIENT_connect (h->cfg,
+ "testbed-barrier",
+ handlers,
+ &mq_error_handler,
+ h);
+ if (NULL == h->mq)
{
- LOG (GNUNET_ERROR_TYPE_ERROR, "Unable to connect to local testbed-barrier service\n");
- destroy_handle (h);
+ LOG (GNUNET_ERROR_TYPE_ERROR,
+ "Unable to connect to local testbed-barrier service\n");
+ GNUNET_TESTBED_barrier_wait_cancel (h);
return NULL;
}
- name_len = strlen (name);
- msize = sizeof (struct GNUNET_TESTBED_BarrierWait) + name_len;
- msg = GNUNET_malloc (msize);
- msg->header.type = htons (GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_WAIT);
- msg->header.size = htons (msize);
- (void) memcpy (msg->name, name, name_len);
- h->msg = &msg->header;
- h->tx =
- GNUNET_CLIENT_notify_transmit_ready (h->conn, msize,
- GNUNET_TIME_UNIT_FOREVER_REL,
- GNUNET_NO,
- &transmit_notify,
- h);
+ name_len = strlen (name); /* NOTE: unusual to not have 0-termination, change? */
+ env = GNUNET_MQ_msg_extra (msg,
+ name_len,
+ GNUNET_MESSAGE_TYPE_TESTBED_BARRIER_WAIT);
+ GNUNET_memcpy (msg->name,
+ name,
+ name_len);
+ GNUNET_MQ_send (h->mq,
+ env);
return h;
}
void
GNUNET_TESTBED_barrier_wait_cancel (struct GNUNET_TESTBED_BarrierWaitHandle *h)
{
- destroy_handle (h);
+ if (NULL != h->mq)
+ {
+ GNUNET_MQ_destroy (h->mq);
+ h->mq = NULL;
+ }
+ GNUNET_free (h->name);
+ GNUNET_CONFIGURATION_destroy (h->cfg);
+ GNUNET_free (h);
}
/* end of testbed_api_barriers.c */
projects / oweals / gnunet.git / blobdiff