/**
* @file consensus/consensus_api.c
- * @brief
+ * @brief
* @author Florian Dold
*/
#include "platform.h"
#define LOG(kind,...) GNUNET_log_from (kind, "consensus-api",__VA_ARGS__)
-struct ElementAck
-{
- struct ElementAck *next;
- struct ElementAck *prev;
- int keep;
- struct GNUNET_CONSENSUS_Element *element;
-};
/**
* Handle for the service.
const struct GNUNET_CONFIGURATION_Handle *cfg;
/**
- * Socket (if available).
+ * Client connected to the consensus service, may be NULL if not connected.
*/
struct GNUNET_CLIENT_Connection *client;
/**
* Callback for new elements. Not called for elements added locally.
*/
- GNUNET_CONSENSUS_NewElementCallback new_element_cb;
+ GNUNET_CONSENSUS_ElementCallback new_element_cb;
/**
* Closure for new_element_cb
void *new_element_cls;
/**
- * Session identifier for the consensus session.
+ * The (local) session identifier for the consensus session.
*/
struct GNUNET_HashCode session_id;
- /**
- * Number of peers in the consensus. Optionally includes the local peer.
- */
- int num_peers;
-
- /**
- * Peer identities of peers in the consensus. Optionally includes the local peer.
- */
- struct GNUNET_PeerIdentity *peers;
-
- /**
- * Currently active transmit request.
- */
- struct GNUNET_CLIENT_TransmitHandle *th;
-
/**
* GNUNES_YES iff the join message has been sent to the service.
*/
int joined;
- /**
- * Called when the current insertion operation finishes.
- * NULL if there is no insert operation active.
- */
- GNUNET_CONSENSUS_InsertDoneCallback idc;
-
- /**
- * Closure for the insert done callback.
- */
- void *idc_cls;
-
- /**
- * An element that was requested to be inserted.
- */
- struct GNUNET_CONSENSUS_Element *insert_element;
-
/**
* Called when the conclude operation finishes or fails.
*/
*/
struct GNUNET_TIME_Absolute conclude_deadline;
- struct ElementAck *ack_head;
- struct ElementAck *ack_tail;
-
- /**
- * Set to GNUNET_YES if the begin message has been transmitted to the service
- */
- int begin_sent;
-
/**
- * Set to GNUNET_YES it the begin message should be transmitted to the service
+ * Message queue for the client.
*/
- int begin_requested;
+ struct GNUNET_MQ_Handle *mq;
};
-
-static size_t
-transmit_ack (void *cls, size_t size, void *buf);
-
-static size_t
-transmit_insert (void *cls, size_t size, void *buf);
-
-static size_t
-transmit_conclude (void *cls, size_t size, void *buf);
-
-static size_t
-transmit_begin (void *cls, size_t size, void *buf);
-
-
/**
- * Call notify_transmit_ready for ack if necessary and possible.
+ * FIXME: this should not bee necessary when the API
+ * issue has been fixed
*/
-static void
-ntr_ack (struct GNUNET_CONSENSUS_Handle *consensus)
-{
- if ((NULL == consensus->th) && (NULL != consensus->ack_head))
- {
- consensus->th =
- GNUNET_CLIENT_notify_transmit_ready (consensus->client,
- sizeof (struct GNUNET_CONSENSUS_AckMessage),
- GNUNET_TIME_UNIT_FOREVER_REL,
- GNUNET_NO, &transmit_ack, consensus);
- }
-}
-
-
-/**
- * Call notify_transmit_ready for ack if necessary and possible.
- */
-static void
-ntr_insert (struct GNUNET_CONSENSUS_Handle *consensus)
-{
- if ((NULL == consensus->th) && (NULL != consensus->insert_element))
- {
- consensus->th =
- GNUNET_CLIENT_notify_transmit_ready (consensus->client,
- sizeof (struct GNUNET_CONSENSUS_ElementMessage) +
- consensus->insert_element->size,
- GNUNET_TIME_UNIT_FOREVER_REL,
- GNUNET_NO, &transmit_insert, consensus);
- }
-}
-
-
-/**
- * Call notify_transmit_ready for ack if necessary and possible.
- */
-static void
-ntr_conclude (struct GNUNET_CONSENSUS_Handle *consensus)
+struct InsertDoneInfo
{
- if ((NULL == consensus->th) && (NULL != consensus->conclude_cb))
- {
- consensus->th =
- GNUNET_CLIENT_notify_transmit_ready (consensus->client,
- sizeof (struct GNUNET_CONSENSUS_ConcludeMessage),
- GNUNET_TIME_absolute_get_remaining (consensus->conclude_deadline),
- GNUNET_NO, &transmit_conclude, consensus);
- }
-}
-
+ GNUNET_CONSENSUS_InsertDoneCallback idc;
+ void *cls;
+};
-/**
- * Call notify_transmit_ready for ack if necessary and possible.
- */
-static void
-ntr_begin (struct GNUNET_CONSENSUS_Handle *consensus)
-{
- if ((NULL == consensus->th) && (GNUNET_YES == consensus->begin_requested) &&
- (GNUNET_NO == consensus->begin_sent))
- {
- consensus->th =
- GNUNET_CLIENT_notify_transmit_ready (consensus->client,
- sizeof (struct GNUNET_MessageHeader),
- GNUNET_TIME_UNIT_FOREVER_REL,
- GNUNET_NO, &transmit_begin, consensus);
- }
-}
/**
* Called when the server has sent is a new element
- *
- * @param consensus consensus handle
- * @param msg element message
+ *
+ * @param cls consensus handle
+ * @param mh element message
*/
static void
-handle_new_element(struct GNUNET_CONSENSUS_Handle *consensus,
- struct GNUNET_CONSENSUS_ElementMessage *msg)
+handle_new_element (void *cls,
+ const struct GNUNET_MessageHeader *mh)
{
- struct GNUNET_CONSENSUS_Element element;
- struct ElementAck *ack;
- int ret;
+ struct GNUNET_CONSENSUS_Handle *consensus = cls;
+ const struct GNUNET_CONSENSUS_ElementMessage *msg
+ = (const struct GNUNET_CONSENSUS_ElementMessage *) mh;
+ struct GNUNET_SET_Element element;
+
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "received new element\n");
element.type = msg->element_type;
- element.size = msg->header.size - sizeof (struct GNUNET_CONSENSUS_ElementMessage);
+ element.size = ntohs (msg->header.size) - sizeof (struct GNUNET_CONSENSUS_ElementMessage);
element.data = &msg[1];
- ret = consensus->new_element_cb (consensus->new_element_cls, &element);
- ack = GNUNET_malloc (sizeof (struct ElementAck));
- ack->keep = ret;
- GNUNET_CONTAINER_DLL_insert_tail (consensus->ack_head, consensus->ack_tail,ack);
-
- ntr_ack (consensus);
+ consensus->new_element_cb (consensus->new_element_cls, &element);
}
/**
* Called when the server has announced
* that the conclusion is over.
- *
- * @param consensus consensus handle
- * @param msg conclude done message
- */
-static void
-handle_conclude_done(struct GNUNET_CONSENSUS_Handle *consensus,
- struct GNUNET_CONSENSUS_ConcludeDoneMessage *msg)
-{
- GNUNET_assert (NULL != consensus->conclude_cb);
- consensus->conclude_cb(consensus->conclude_cls,
- msg->num_peers,
- (struct GNUNET_PeerIdentity *) &msg[1]);
- consensus->conclude_cb = NULL;
-}
-
-
-
-/**
- * Type of a function to call when we receive a message
- * from the service.
*
- * @param cls closure
- * @param msg message received, NULL on timeout or fatal error
+ * @param cls consensus handle
+ * @param msg conclude done message
*/
static void
-message_handler (void *cls, const struct GNUNET_MessageHeader *msg)
+handle_conclude_done (void *cls,
+ const struct GNUNET_MessageHeader *msg)
{
struct GNUNET_CONSENSUS_Handle *consensus = cls;
- LOG (GNUNET_ERROR_TYPE_INFO, "received message from consensus service\n");
-
- if (msg == NULL)
- {
- /* Error, timeout, death */
- LOG (GNUNET_ERROR_TYPE_ERROR, "error receiving\n");
- GNUNET_CLIENT_disconnect (consensus->client);
- consensus->client = NULL;
- consensus->new_element_cb (NULL, NULL);
- if (NULL != consensus->idc)
- {
- consensus->idc(consensus->idc_cls, GNUNET_NO);
- consensus->idc = NULL;
- consensus->idc_cls = NULL;
- }
- return;
- }
-
- switch (ntohs (msg->type))
- {
- case GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_RECEIVED_ELEMENT:
- handle_new_element (consensus, (struct GNUNET_CONSENSUS_ElementMessage *) msg);
- break;
- case GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE_DONE:
- handle_conclude_done (consensus, (struct GNUNET_CONSENSUS_ConcludeDoneMessage *) msg);
- break;
- default:
- LOG(GNUNET_ERROR_TYPE_WARNING, "did not understand message type sent by service, ignoring");
- }
- GNUNET_CLIENT_receive (consensus->client, &message_handler, consensus,
- GNUNET_TIME_UNIT_FOREVER_REL);
-}
-
-
-
-
-/**
- * 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.
- *
- * @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
- */
-static size_t
-transmit_ack (void *cls, size_t size, void *buf)
-{
- struct GNUNET_CONSENSUS_AckMessage *msg;
- struct GNUNET_CONSENSUS_Handle *consensus;
-
- consensus = (struct GNUNET_CONSENSUS_Handle *) cls;
-
- GNUNET_assert (NULL != consensus->ack_head);
-
- msg = (struct GNUNET_CONSENSUS_AckMessage *) buf;
- msg->keep = consensus->ack_head->keep;
- msg->header.type = htons (GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_ACK);
- msg->header.size = htons (sizeof (struct GNUNET_CONSENSUS_AckMessage));
+ GNUNET_CONSENSUS_ConcludeCallback cc;
- consensus->ack_head = consensus->ack_head->next;
-
- consensus->th = NULL;
-
- ntr_insert (consensus);
- ntr_ack (consensus);
- ntr_conclude (consensus);
-
- return sizeof (struct GNUNET_CONSENSUS_AckMessage);
-}
-
-/**
- * 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.
- *
- * @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
- */
-static size_t
-transmit_insert (void *cls, size_t size, void *buf)
-{
- struct GNUNET_CONSENSUS_ElementMessage *msg;
- struct GNUNET_CONSENSUS_Handle *consensus;
- GNUNET_CONSENSUS_InsertDoneCallback idc;
- int msize;
- void *idc_cls;
-
- GNUNET_assert (NULL != buf);
-
- consensus = cls;
-
- GNUNET_assert (NULL != consensus->insert_element);
-
- consensus->th = NULL;
-
- msg = buf;
-
- msize = sizeof (struct GNUNET_CONSENSUS_ElementMessage) +
- consensus->insert_element->size;
-
- msg->header.type = htons (GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_INSERT);
- msg->header.size = htons (msize);
- memcpy (&msg[1],
- consensus->insert_element->data,
- consensus->insert_element->size);
-
- consensus->insert_element = NULL;
-
- idc = consensus->idc;
- consensus->idc = NULL;
- idc_cls = consensus->idc_cls;
- consensus->idc_cls = NULL;
- idc (idc_cls, GNUNET_YES);
-
-
- ntr_ack (consensus);
- ntr_insert (consensus);
- ntr_conclude (consensus);
-
- return msize;
-}
-
-
-/**
- * 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.
- *
- * @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
- */
-static size_t
-transmit_join (void *cls, size_t size, void *buf)
-{
- struct GNUNET_CONSENSUS_JoinMessage *msg;
- struct GNUNET_CONSENSUS_Handle *consensus;
- int msize;
-
- GNUNET_assert (NULL != buf);
-
- LOG (GNUNET_ERROR_TYPE_DEBUG, "transmitting join message\n");
-
- consensus = cls;
- consensus->th = NULL;
- consensus->joined = 1;
-
- msg = buf;
-
- msize = sizeof (struct GNUNET_CONSENSUS_JoinMessage) +
- consensus->num_peers * sizeof (struct GNUNET_PeerIdentity);
-
- msg->header.type = htons (GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_JOIN);
- msg->header.size = htons (msize);
- msg->session_id = consensus->session_id;
- msg->num_peers = htons (consensus->num_peers);
- if (0 != msg->num_peers)
- memcpy(&msg[1],
- consensus->peers,
- consensus->num_peers * sizeof (struct GNUNET_PeerIdentity));
-
- ntr_insert (consensus);
- ntr_begin (consensus);
- ntr_conclude (consensus);
-
- GNUNET_CLIENT_receive (consensus->client, &message_handler, consensus,
- GNUNET_TIME_UNIT_FOREVER_REL);
-
- return msize;
-}
-
-
-/**
- * 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.
- *
- * @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
- */
-static size_t
-transmit_conclude (void *cls, size_t size, void *buf)
-{
- struct GNUNET_CONSENSUS_ConcludeMessage *msg;
- struct GNUNET_CONSENSUS_Handle *consensus;
- int msize;
-
- GNUNET_assert (NULL != buf);
-
- consensus = cls;
- consensus->th = NULL;
-
- msg = buf;
-
- msize = sizeof (struct GNUNET_CONSENSUS_ConcludeMessage);
-
- msg->header.type = htons (GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE);
- msg->header.size = htons (msize);
- msg->timeout =
- GNUNET_TIME_relative_hton (GNUNET_TIME_absolute_get_remaining(consensus->conclude_deadline));
-
- ntr_ack (consensus);
-
- return msize;
-}
-
-
-/**
- * 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.
- *
- * @param cls the consensus handle
- * @param size number of bytes available in buf
- * @param buf where the callee should write the message
- * @return number of bytes written to buf
- */
-static size_t
-transmit_begin (void *cls, size_t size, void *buf)
-{
- struct GNUNET_MessageHeader *msg;
- struct GNUNET_CONSENSUS_Handle *consensus;
- int msize;
-
- GNUNET_assert (NULL != buf);
-
- consensus = cls;
- consensus->th = NULL;
-
- msg = buf;
-
- msize = sizeof (struct GNUNET_MessageHeader);
-
- msg->type = htons (GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_BEGIN);
- msg->size = htons (msize);
-
- ntr_ack (consensus);
- ntr_insert (consensus);
- ntr_conclude (consensus);
-
- return msize;
+ GNUNET_assert (NULL != (cc = consensus->conclude_cb));
+ consensus->conclude_cb = NULL;
+ cc (consensus->conclude_cls);
}
/**
* Create a consensus session.
*
- * @param cfg
- * @param num_peers
+ * @param cfg configuration to use for connecting to the consensus service
+ * @param num_peers number of peers in the peers array
* @param peers array of peers participating in this consensus session
* Inclusion of the local peer is optional.
* @param session_id session identifier
unsigned int num_peers,
const struct GNUNET_PeerIdentity *peers,
const struct GNUNET_HashCode *session_id,
- GNUNET_CONSENSUS_NewElementCallback new_element_cb,
+ GNUNET_CONSENSUS_ElementCallback new_element_cb,
void *new_element_cls)
{
struct GNUNET_CONSENSUS_Handle *consensus;
- size_t join_message_size;
+ struct GNUNET_CONSENSUS_JoinMessage *join_msg;
+ struct GNUNET_MQ_Envelope *ev;
+ const static struct GNUNET_MQ_MessageHandler mq_handlers[] = {
+ {handle_new_element,
+ GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_RECEIVED_ELEMENT, 0},
+ {handle_conclude_done,
+ GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE_DONE, 0},
+ GNUNET_MQ_HANDLERS_END
+ };
consensus = GNUNET_malloc (sizeof (struct GNUNET_CONSENSUS_Handle));
consensus->cfg = cfg;
consensus->new_element_cb = new_element_cb;
consensus->new_element_cls = new_element_cls;
- consensus->num_peers = num_peers;
consensus->session_id = *session_id;
-
- if (0 == num_peers)
- {
- consensus->peers = NULL;
- }
- else if (num_peers > 0)
- {
- consensus->peers = GNUNET_memdup (peers, num_peers * sizeof (struct GNUNET_PeerIdentity));
- }
- else
- {
- GNUNET_break (0);
- }
-
consensus->client = GNUNET_CLIENT_connect ("consensus", cfg);
+ consensus->mq = GNUNET_MQ_queue_for_connection_client (consensus->client,
+ mq_handlers, NULL, consensus);
GNUNET_assert (consensus->client != NULL);
- join_message_size = (sizeof (struct GNUNET_CONSENSUS_JoinMessage)) +
- (num_peers * sizeof (struct GNUNET_PeerIdentity));
-
- consensus->th =
- GNUNET_CLIENT_notify_transmit_ready (consensus->client,
- join_message_size,
- GNUNET_TIME_UNIT_FOREVER_REL,
- GNUNET_NO, &transmit_join, consensus);
+ ev = GNUNET_MQ_msg_extra (join_msg,
+ (num_peers * sizeof (struct GNUNET_PeerIdentity)),
+ GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_JOIN);
+ join_msg->session_id = consensus->session_id;
+ join_msg->num_peers = htonl (num_peers);
+ memcpy(&join_msg[1],
+ peers,
+ num_peers * sizeof (struct GNUNET_PeerIdentity));
- GNUNET_assert (consensus->th != NULL);
+ GNUNET_MQ_send (consensus->mq, ev);
return consensus;
}
+static void
+idc_adapter (void *cls)
+{
+ struct InsertDoneInfo *i = cls;
+ i->idc (i->cls, GNUNET_OK);
+ GNUNET_free (i);
+}
/**
* Insert an element in the set being reconsiled. Must not be called after
*
* @param consensus handle for the consensus session
* @param element the element to be inserted
- * @param idc function called when we are done with this element and it
+ * @param idc function called when we are done with this element and it
* is thus allowed to call GNUNET_CONSENSUS_insert again
* @param idc_cls closure for 'idc'
*/
void
GNUNET_CONSENSUS_insert (struct GNUNET_CONSENSUS_Handle *consensus,
- const struct GNUNET_CONSENSUS_Element *element,
+ const struct GNUNET_SET_Element *element,
GNUNET_CONSENSUS_InsertDoneCallback idc,
void *idc_cls)
{
- GNUNET_assert (NULL == consensus->idc);
- GNUNET_assert (NULL == consensus->insert_element);
- GNUNET_assert (NULL == consensus->conclude_cb);
-
- consensus->idc = idc;
- consensus->idc_cls = idc_cls;
- consensus->insert_element = GNUNET_memdup(element, sizeof (struct GNUNET_CONSENSUS_Element) + element->size);
-
- if (consensus->joined == 0)
- {
- return;
- }
-
- ntr_insert (consensus);
-}
+ struct GNUNET_CONSENSUS_ElementMessage *element_msg;
+ struct GNUNET_MQ_Envelope *ev;
+ struct InsertDoneInfo *i;
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "inserting, size=%llu\n", element->size);
-/**
- * Begin reconciling elements with other peers.
- *
- * @param consensus handle for the consensus session
- */
-void
-GNUNET_CONSENSUS_begin (struct GNUNET_CONSENSUS_Handle *consensus)
-{
- GNUNET_assert (NULL == consensus->idc);
- GNUNET_assert (NULL == consensus->insert_element);
- GNUNET_assert (GNUNET_NO == consensus->begin_requested);
- GNUNET_assert (GNUNET_NO == consensus->begin_sent);
+ ev = GNUNET_MQ_msg_extra (element_msg, element->size,
+ GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_INSERT);
- consensus->begin_requested = GNUNET_YES;
+ memcpy (&element_msg[1], element->data, element->size);
- ntr_begin (consensus);
+ if (NULL != idc)
+ {
+ i = GNUNET_new (struct InsertDoneInfo);
+ i->idc = idc;
+ i->cls = idc_cls;
+ GNUNET_MQ_notify_sent (ev, idc_adapter, i);
+ }
+ GNUNET_MQ_send (consensus->mq, ev);
}
/**
- * We are finished inserting new elements into the consensus;
+ * We are done with inserting new elements into the consensus;
* try to conclude the consensus within a given time window.
+ * After conclude has been called, no further elements may be
+ * inserted by the client.
*
* @param consensus consensus session
* @param timeout timeout after which the conculde callback
GNUNET_CONSENSUS_ConcludeCallback conclude,
void *conclude_cls)
{
+ struct GNUNET_MQ_Envelope *ev;
+ struct GNUNET_CONSENSUS_ConcludeMessage *conclude_msg;
+
GNUNET_assert (NULL != conclude);
GNUNET_assert (NULL == consensus->conclude_cb);
consensus->conclude_cls = conclude_cls;
consensus->conclude_cb = conclude;
- consensus->conclude_deadline = GNUNET_TIME_relative_to_absolute(timeout);
+ ev = GNUNET_MQ_msg (conclude_msg, GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE);
+ conclude_msg->timeout = GNUNET_TIME_relative_hton (timeout);
- /* if transmitting the conclude message is not possible right now, transmit_join
- * or transmit_ack will handle it */
- ntr_conclude (consensus);
+ GNUNET_MQ_send (consensus->mq, ev);
}
GNUNET_CLIENT_disconnect (consensus->client);
consensus->client = NULL;
}
- if (NULL != consensus->peers)
- GNUNET_free (consensus->peers);
GNUNET_free (consensus);
}
projects / oweals / gnunet.git / blobdiff