/*
This file is part of GNUnet.
- Copyright (C) 2008--2013 GNUnet e.V.
+ 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
* @brief API implementation for testbed barriers
* @author Sree Harsha Totakura <sreeharsha@totakura.in>
*/
-
#include "platform.h"
#include "gnunet_testbed_service.h"
#include "testbed_api.h"
-#include "testbed_api_barriers.h"
/**
* Logging shorthand
LOG (GNUNET_ERROR_TYPE_DEBUG, __VA_ARGS__);
-/**
- * Remove a barrier and it was the last one in the barrier hash map, destroy the
- * hash map
- *
- * @param barrier the barrier to remove
- */
-void
-GNUNET_TESTBED_barrier_remove_ (struct GNUNET_TESTBED_Barrier *barrier)
-{
- struct GNUNET_TESTBED_Controller *c = barrier->c;
-
- GNUNET_assert (NULL != c->barrier_map); /* No barriers present */
- GNUNET_assert (GNUNET_OK ==
- GNUNET_CONTAINER_multihashmap_remove (c->barrier_map,
- &barrier->key,
- barrier));
- GNUNET_free (barrier->name);
- GNUNET_free (barrier);
- if (0 == GNUNET_CONTAINER_multihashmap_size (c->barrier_map))
- {
- GNUNET_CONTAINER_multihashmap_destroy (c->barrier_map);
- c->barrier_map = NULL;
- }
-}
-
-
-/**
- * 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
- * @param echo GNUNET_YES to echo the barrier crossed status message back to the
- * controller
- * @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,
- int echo)
-{
- 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 == controller->barrier_map)
- controller->barrier_map = GNUNET_CONTAINER_multihashmap_create (3, GNUNET_YES);
- if (GNUNET_YES ==
- GNUNET_CONTAINER_multihashmap_contains (controller->barrier_map,
- &key))
- {
- GNUNET_break (0);
- return NULL;
- }
- LOG_DEBUG ("Initialising barrier `%s'\n", name);
- barrier = GNUNET_new (struct GNUNET_TESTBED_Barrier);
- barrier->c = controller;
- barrier->name = GNUNET_strdup (name);
- barrier->cb = cb;
- barrier->cls = cls;
- barrier->echo = echo;
- (void) memcpy (&barrier->key, &key, sizeof (struct GNUNET_HashCode));
- GNUNET_assert (GNUNET_OK ==
- GNUNET_CONTAINER_multihashmap_put (controller->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;
-}
-
-
-/**
- * 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)
-{
- return GNUNET_TESTBED_barrier_init_ (controller,
- name, quorum, cb, cls, GNUNET_YES);
-}
-
-
-/**
- * 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);
- GNUNET_TESTBED_barrier_remove_ (barrier);
-}
-
-
/**
* Barrier wait handle
*/
struct GNUNET_CONFIGURATION_Handle *cfg;
/**
- * The client connection
- */
- struct GNUNET_CLIENT_Connection *conn;
-
- /**
- * Transmit handle
+ * The testbed-barrier service message queue.
*/
- struct GNUNET_CLIENT_TransmitHandle *tx;
-
- /**
- * The message to transmit with tx
- */
- 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 (NULL == message)
- {
- GNUNET_break_op (0);
- goto fail;
- }
- 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 GNUNET_TESTBED_BARRIERSTATUS_ERROR:
- goto fail;
+ h->cb (h->cb_cls,
+ h->name,
+ GNUNET_SYSERR);
+ break;
case GNUNET_TESTBED_BARRIERSTATUS_INITIALISED:
- GNUNET_break (0); /* FIXME */
- goto destroy;
+ h->cb (h->cb_cls,
+ h->name,
+ GNUNET_SYSERR);
+ GNUNET_break (0);
+ break;
case GNUNET_TESTBED_BARRIERSTATUS_CROSSED:
- h->cb (h->cls, h->name, GNUNET_OK);
- goto destroy;
+ 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_new (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);
+ 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