/*
This file is part of GNUnet.
- (C) 2011 Christian Grothoff (and other contributing authors)
+ 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., 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 cadet/cadet_api.c
* @brief cadet api: client implementation of new cadet service
#include "cadet_protocol.h"
#define LOG(kind,...) GNUNET_log_from (kind, "cadet-api",__VA_ARGS__)
+#define DATA_OVERHEAD sizeof(struct GNUNET_CADET_LocalData)
/******************************************************************************/
/************************ DATA STRUCTURES ****************************/
/**
* Task triggering a timeout, can be NO_TASK if the timeout is FOREVER.
*/
- GNUNET_SCHEDULER_TaskIdentifier timeout_task;
+ struct GNUNET_SCHEDULER_Task * timeout_task;
/**
* Size of 'data' -- or the desired size of 'notify' if 'data' is NULL.
/**
* Task for trying to reconnect.
*/
- GNUNET_SCHEDULER_TaskIdentifier reconnect_task;
+ struct GNUNET_SCHEDULER_Task * reconnect_task;
/**
* Callback for an info task (only one active at a time).
*
* @param th Transmission handle.
*
- * @return GNUNET_YES if it is a payload packet,
- * GNUNET_NO if it is a cadet management packet.
+ * @return #GNUNET_YES if it is a payload packet,
+ * #GNUNET_NO if it is a cadet management packet.
*/
static int
th_is_payload (struct GNUNET_CADET_TransmitHandle *th)
*
* @param h Cadet handle.
*
- * @return The size of the first ready message in the queue,
+ * @return The size of the first ready message in the queue, including overhead.
* 0 if there is none.
*/
static size_t
for (th = h->th_head; NULL != th; th = th->next)
{
ch = th->channel;
- if (GNUNET_NO == th_is_payload (th))
- {
- LOG (GNUNET_ERROR_TYPE_DEBUG, "# message internal\n");
+ if (GNUNET_NO == th_is_payload (th) || GNUNET_YES == ch->allow_send)
return th->size;
- }
- if (GNUNET_YES == ch->allow_send)
- {
- LOG (GNUNET_ERROR_TYPE_DEBUG, "# message payload ok\n");
- return th->size;
- }
}
return 0;
}
GNUNET_CONTAINER_DLL_remove (h->th_head, h->th_tail, th);
/* clean up request */
- if (GNUNET_SCHEDULER_NO_TASK != th->timeout_task)
+ if (NULL != th->timeout_task)
GNUNET_SCHEDULER_cancel (th->timeout_task);
GNUNET_free (th);
}
* Notify client that the transmission has timed out
*
* @param cls closure
- * @param tc task context
*/
static void
-timeout_transmission (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+timeout_transmission (void *cls)
{
struct GNUNET_CADET_TransmitHandle *th = cls;
- struct GNUNET_CADET_Handle *cadet;
+ struct GNUNET_CADET_Handle *cadet = th->channel->cadet;
- cadet = th->channel->cadet;
- GNUNET_CONTAINER_DLL_remove (cadet->th_head, cadet->th_tail, th);
+ th->timeout_task = NULL;
th->channel->packet_size = 0;
+ GNUNET_CONTAINER_DLL_remove (cadet->th_head, cadet->th_tail, th);
if (GNUNET_YES == th_is_payload (th))
- th->notify (th->notify_cls, 0, NULL);
+ GNUNET_break (0 == th->notify (th->notify_cls, 0, NULL));
GNUNET_free (th);
if ((0 == message_ready_size (cadet)) && (NULL != cadet->th))
{
/**
* Reconnect callback: tries to reconnect again after a failer previous
- * reconnecttion
+ * reconnection.
+ *
* @param cls closure (cadet handle)
- * @param tc task context
*/
static void
-reconnect_cbk (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
+reconnect_cbk (void *cls);
/**
/**
* Reconnect callback: tries to reconnect again after a failer previous
* reconnecttion
+ *
* @param cls closure (cadet handle)
- * @param tc task context
*/
static void
-reconnect_cbk (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+reconnect_cbk (void *cls)
{
struct GNUNET_CADET_Handle *h = cls;
- h->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
- if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
- return;
+ h->reconnect_task = NULL;
do_reconnect (h);
}
*
* @param h handle to the cadet
*
- * @return GNUNET_YES in case of sucess, GNUNET_NO otherwise (service down...)
+ * @return #GNUNET_YES in case of sucess, #GNUNET_NO otherwise (service down...)
*/
static void
reconnect (struct GNUNET_CADET_Handle *h)
{
struct GNUNET_CADET_Channel *ch;
- struct GNUNET_CADET_Channel *next;
LOG (GNUNET_ERROR_TYPE_DEBUG,
"Requested RECONNECT, destroying all channels\n");
h->in_receive = GNUNET_NO;
- for (ch = h->channels_head; NULL != ch; ch = next)
- {
- next = ch->next;
+ for (ch = h->channels_head; NULL != ch; ch = h->channels_head)
destroy_channel (ch, GNUNET_YES);
- }
- if (GNUNET_SCHEDULER_NO_TASK == h->reconnect_task)
+ if (NULL == h->reconnect_task)
h->reconnect_task = GNUNET_SCHEDULER_add_delayed (h->reconnect_time,
&reconnect_cbk, h);
}
*/
static void
process_channel_created (struct GNUNET_CADET_Handle *h,
- const struct GNUNET_CADET_ChannelMessage *msg)
+ const struct GNUNET_CADET_ChannelMessage *msg)
{
struct GNUNET_CADET_Channel *ch;
CADET_ChannelNumber chid;
struct GNUNET_CADET_Channel *ch;
CADET_ChannelNumber chid;
- LOG (GNUNET_ERROR_TYPE_DEBUG, "Channel Destroy received from service\n");
chid = ntohl (msg->channel_id);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Channel %X Destroy from service\n", chid);
ch = retrieve_channel (h, chid);
if (NULL == ch)
LOG (GNUNET_ERROR_TYPE_DEBUG, "channel %X unknown\n", chid);
return;
}
- LOG (GNUNET_ERROR_TYPE_DEBUG, " destroying channel %X\n", ch->chid);
destroy_channel (ch, GNUNET_YES);
}
size = ntohs (message->size);
LOG (GNUNET_ERROR_TYPE_DEBUG, " %u bytes\n", size);
- if (NULL == ch)
- {
- /* Channel was ignored/destroyed, probably service didn't get it yet */
- LOG (GNUNET_ERROR_TYPE_DEBUG, " ignored!\n");
- return;
- }
type = ntohs (payload->type);
size = ntohs (payload->size);
LOG (GNUNET_ERROR_TYPE_DEBUG, " payload type %s\n", GC_m2s (type));
process_get_peer (struct GNUNET_CADET_Handle *h,
const struct GNUNET_MessageHeader *message)
{
- struct GNUNET_CADET_LocalInfoTunnel *msg;
+ struct GNUNET_CADET_LocalInfoPeer *msg;
+ struct GNUNET_PeerIdentity *id;
+ unsigned int epaths;
+ unsigned int paths;
+ unsigned int path_length;
+ unsigned int i;
+ int neighbor;
size_t esize;
size_t msize;
- unsigned int ch_n;
- unsigned int c_n;
- struct GNUNET_CADET_Hash *conns;
- CADET_ChannelNumber *chns;
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Get Tunnel messasge received\n");
- if (NULL == h->info_cb.tunnel_cb)
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Info Peer messasge received\n");
+ if (NULL == h->info_cb.peer_cb)
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " ignored\n");
return;
}
/* Verify message sanity */
- msg = (struct GNUNET_CADET_LocalInfoTunnel *) message;
- msize = ntohs (message->size);
- esize = sizeof (struct GNUNET_CADET_LocalInfoTunnel);
- if (esize > msize)
+ msg = (struct GNUNET_CADET_LocalInfoPeer *) message;
+ esize = ntohs (message->size);
+ msize = sizeof (struct GNUNET_CADET_LocalInfoPeer);
+ if (esize < msize)
{
GNUNET_break_op (0);
- h->info_cb.tunnel_cb (h->info_cls, NULL, 0, 0, NULL, NULL, 0, 0);
+ h->info_cb.peer_cb (h->info_cls, NULL, 0, 0, 0, NULL);
goto clean_cls;
}
- ch_n = ntohl (msg->channels);
- c_n = ntohl (msg->connections);
- esize += ch_n * sizeof (CADET_ChannelNumber);
- esize += c_n * sizeof (struct GNUNET_CADET_Hash);
+ epaths = (unsigned int) ntohs (msg->paths);
+ paths = 0;
+ path_length = 0;
+ neighbor = GNUNET_NO;
+ id = (struct GNUNET_PeerIdentity *) &msg[1];
+ for (i = 0; msize < esize; i++)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " %s\n", GNUNET_i2s (&id[i]));
+ msize += sizeof (struct GNUNET_PeerIdentity);
+ path_length++;
+ if (0 == memcmp (&id[i], &msg->destination,
+ sizeof (struct GNUNET_PeerIdentity)))
+ {
+ if (1 == path_length)
+ neighbor = GNUNET_YES;
+ path_length = 0;
+ paths++;
+ }
+ }
if (msize != esize)
{
GNUNET_break_op (0);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "m:%u, e: %u (%u ch, %u conn)\n",
- msize, esize, ch_n, c_n);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%u (%u ch, %u conn)\n",
- sizeof (struct GNUNET_CADET_LocalInfoTunnel),
- sizeof (CADET_ChannelNumber), sizeof (struct GNUNET_HashCode));
- h->info_cb.tunnel_cb (h->info_cls, NULL, 0, 0, NULL, NULL, 0, 0);
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "m:%u, e: %u\n",
+ (unsigned int) msize,
+ (unsigned int) esize);
+ h->info_cb.peer_cb (h->info_cls, NULL, 0, 0, 0, NULL);
+ goto clean_cls;
+ }
+ if (paths != epaths)
+ {
+ GNUNET_break_op (0);
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "p:%u, e: %u\n", paths, epaths);
+ h->info_cb.peer_cb (h->info_cls, NULL, 0, 0, 0, NULL);
goto clean_cls;
}
/* Call Callback with tunnel info. */
- conns = (struct GNUNET_CADET_Hash *) &msg[1];
- chns = (CADET_ChannelNumber *) &conns[c_n];
- h->info_cb.tunnel_cb (h->info_cls, &msg->destination,
- ch_n, c_n, chns, conns,
- ntohs (msg->estate), ntohs (msg->cstate));
+ id = (struct GNUNET_PeerIdentity *) &msg[1];
+ h->info_cb.peer_cb (h->info_cls,
+ &msg->destination,
+ (int) ntohs (msg->tunnel),
+ neighbor,
+ paths,
+ id);
clean_cls:
- h->info_cb.tunnel_cb = NULL;
+ h->info_cb.peer_cb = NULL;
h->info_cls = NULL;
}
if (msize != esize)
{
GNUNET_break_op (0);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "m:%u, e: %u (%u ch, %u conn)\n",
- msize, esize, ch_n, c_n);
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%u (%u ch, %u conn)\n",
- sizeof (struct GNUNET_CADET_LocalInfoTunnel),
- sizeof (CADET_ChannelNumber), sizeof (struct GNUNET_HashCode));
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "m:%u, e: %u (%u ch, %u conn)\n",
+ (unsigned int) msize,
+ (unsigned int) esize,
+ ch_n,
+ c_n);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%u (%u ch, %u conn)\n",
+ (unsigned int) sizeof (struct GNUNET_CADET_LocalInfoTunnel),
+ (unsigned int) sizeof (CADET_ChannelNumber),
+ (unsigned int) sizeof (struct GNUNET_HashCode));
h->info_cb.tunnel_cb (h->info_cls, NULL, 0, 0, NULL, NULL, 0, 0);
goto clean_cls;
}
size_t nsize;
LOG (GNUNET_ERROR_TYPE_DEBUG, "\n");
- LOG (GNUNET_ERROR_TYPE_DEBUG, "# Send packet() Buffer %u\n", size);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "# Send callback, buffer %u\n", size);
if ((0 == size) || (NULL == buf))
{
LOG (GNUNET_ERROR_TYPE_DEBUG, "# Received NULL send callback on %p\n", h);
struct GNUNET_CADET_LocalData *dmsg;
struct GNUNET_MessageHeader *mh;
- LOG (GNUNET_ERROR_TYPE_DEBUG, "# payload\n");
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "# payload, %u bytes on %X (%p)\n",
+ th->size, ch->chid, ch);
if (GNUNET_NO == ch->allow_send)
{
- /* This channel is not ready to transmit yet, try next message */
+ /* This channel is not ready to transmit yet, Try the next message */
next = th->next;
continue;
}
GNUNET_assert (size >= th->size);
dmsg = (struct GNUNET_CADET_LocalData *) cbuf;
mh = (struct GNUNET_MessageHeader *) &dmsg[1];
- psize = th->notify (th->notify_cls,
- size - sizeof (struct GNUNET_CADET_LocalData),
- mh);
+ psize = th->notify (th->notify_cls, size - DATA_OVERHEAD, mh);
+
if (psize > 0)
{
- psize += sizeof (struct GNUNET_CADET_LocalData);
+ GNUNET_assert (sizeof (struct GNUNET_MessageHeader) <= psize);
+ psize += DATA_OVERHEAD;
GNUNET_assert (size >= psize);
+ dmsg->header.type = htons (GNUNET_MESSAGE_TYPE_CADET_LOCAL_DATA);
dmsg->header.size = htons (psize);
dmsg->id = htonl (ch->chid);
- dmsg->header.type = htons (GNUNET_MESSAGE_TYPE_CADET_LOCAL_DATA);
- LOG (GNUNET_ERROR_TYPE_DEBUG, "# payload type %s\n",
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "# sending, type %s\n",
GC_m2s (ntohs (mh->type)));
- ch->allow_send = GNUNET_NO;
+ ch->allow_send = GNUNET_NO;
}
else
{
}
else
{
- struct GNUNET_MessageHeader *mh = (struct GNUNET_MessageHeader *) &th[1];
+ const struct GNUNET_MessageHeader *mh;
+ mh = (const struct GNUNET_MessageHeader *) &th[1];
LOG (GNUNET_ERROR_TYPE_DEBUG, "# cadet internal traffic, type %s\n",
GC_m2s (ntohs (mh->type)));
memcpy (cbuf, &th[1], th->size);
psize = th->size;
}
- if (th->timeout_task != GNUNET_SCHEDULER_NO_TASK)
+ GNUNET_assert (GNUNET_CONSTANTS_MAX_CADET_MESSAGE_SIZE >= psize);
+ if (th->timeout_task != NULL)
GNUNET_SCHEDULER_cancel (th->timeout_task);
+ next = th->next;
GNUNET_CONTAINER_DLL_remove (h->th_head, h->th_tail, th);
GNUNET_free (th);
- next = h->th_head;
nsize = message_ready_size (h);
cbuf += psize;
size -= psize;
else
{
if (NULL != h->th_head)
- LOG (GNUNET_ERROR_TYPE_DEBUG, "# can't transmit any more\n");
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "# nothing ready to transmit\n");
else
LOG (GNUNET_ERROR_TYPE_DEBUG, "# nothing left to transmit\n");
}
GNUNET_CLIENT_receive (h->client, &msg_received, h,
GNUNET_TIME_UNIT_FOREVER_REL);
}
- LOG (GNUNET_ERROR_TYPE_DEBUG, "# Send packet() END\n");
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "# Send callback() END\n");
return tsize;
}
th->channel = channel;
memcpy (&th[1], msg, msize);
add_to_queue (h, th);
- LOG (GNUNET_ERROR_TYPE_DEBUG, " queued\n");
if (NULL != h->th)
return;
LOG (GNUNET_ERROR_TYPE_DEBUG, " calling ntfy tmt rdy for %u bytes\n", msize);
h->ports = ports;
h->next_chid = GNUNET_CADET_LOCAL_CHANNEL_ID_CLI;
h->reconnect_time = GNUNET_TIME_UNIT_MILLISECONDS;
- h->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
+ h->reconnect_task = NULL;
if (NULL != ports && ports[0] != 0 && NULL == new_channel)
{
break;
default:
GNUNET_break (0);
- LOG (GNUNET_ERROR_TYPE_ERROR, "unexpected msg %u\n",
- ntohs(msg->type));
+ LOG (GNUNET_ERROR_TYPE_ERROR, "unexpected unsent msg %s\n",
+ GC_m2s (ntohs(msg->type)));
}
GNUNET_CONTAINER_DLL_remove (handle->th_head, handle->th_tail, th);
GNUNET_CLIENT_disconnect (handle->client);
handle->client = NULL;
}
- if (GNUNET_SCHEDULER_NO_TASK != handle->reconnect_task)
+ if (NULL != handle->reconnect_task)
{
GNUNET_SCHEDULER_cancel(handle->reconnect_task);
- handle->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
+ handle->reconnect_task = NULL;
}
GNUNET_free (handle);
}
msg.port = htonl (port);
msg.peer = *peer;
msg.opt = htonl (options);
- ch->allow_send = 0;
+ ch->allow_send = GNUNET_NO;
send_packet (h, &msg.header, ch);
return ch;
}
if (th->channel == channel)
{
aux = th->next;
- /* FIXME call the handler? */
if (GNUNET_YES == th_is_payload (th))
+ {
+ /* applications should cancel before destroying channel */
+ GNUNET_break (0);
+ LOG (GNUNET_ERROR_TYPE_WARNING,
+ "Channel destroyed without cancelling transmission requests\n");
th->notify (th->notify_cls, 0, NULL);
- GNUNET_CONTAINER_DLL_remove (h->th_head, h->th_tail, th);
- GNUNET_free (th);
+ }
+ GNUNET_CADET_notify_transmit_ready_cancel (th);
th = aux;
}
else
struct GNUNET_CADET_TransmitHandle *
GNUNET_CADET_notify_transmit_ready (struct GNUNET_CADET_Channel *channel, int cork,
- struct GNUNET_TIME_Relative maxdelay,
- size_t notify_size,
- GNUNET_CONNECTION_TransmitReadyNotify notify,
- void *notify_cls)
+ struct GNUNET_TIME_Relative maxdelay,
+ size_t notify_size,
+ GNUNET_CONNECTION_TransmitReadyNotify notify,
+ void *notify_cls)
{
struct GNUNET_CADET_TransmitHandle *th;
GNUNET_assert (NULL != channel);
+ GNUNET_assert (GNUNET_CONSTANTS_MAX_CADET_MESSAGE_SIZE >= notify_size);
LOG (GNUNET_ERROR_TYPE_DEBUG, "CADET NOTIFY TRANSMIT READY\n");
LOG (GNUNET_ERROR_TYPE_DEBUG, " on channel %X\n", channel->chid);
LOG (GNUNET_ERROR_TYPE_DEBUG, " allow_send %d\n", channel->allow_send);
th = GNUNET_new (struct GNUNET_CADET_TransmitHandle);
th->channel = channel;
th->timeout = GNUNET_TIME_relative_to_absolute (maxdelay);
- th->size = notify_size + sizeof (struct GNUNET_CADET_LocalData);
+ th->size = notify_size + DATA_OVERHEAD;
channel->packet_size = th->size;
LOG (GNUNET_ERROR_TYPE_DEBUG, " total size %u\n", th->size);
th->notify = notify;
{
struct GNUNET_CADET_Handle *cadet;
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "CADET NOTIFY TRANSMIT READY CANCEL\n");
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " on channel %X (%p)\n",
+ th->channel->chid, th->channel);
+ LOG (GNUNET_ERROR_TYPE_DEBUG, " size %u bytes\n", th->size);
th->channel->packet_size = 0;
cadet = th->channel->cadet;
- if (th->timeout_task != GNUNET_SCHEDULER_NO_TASK)
+ if (th->timeout_task != NULL)
GNUNET_SCHEDULER_cancel (th->timeout_task);
GNUNET_CONTAINER_DLL_remove (cadet->th_head, cadet->th_tail, th);
GNUNET_free (th);
GNUNET_CLIENT_notify_transmit_ready_cancel (cadet->th);
cadet->th = NULL;
}
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "CADET NOTIFY TRANSMIT READY CANCEL END\n");
}
*/
static void
cadet_mq_send_impl (struct GNUNET_MQ_Handle *mq,
- const struct GNUNET_MessageHeader *msg, void *impl_state)
+ const struct GNUNET_MessageHeader *msg,
+ void *impl_state)
{
struct CadetMQState *state = impl_state;
GNUNET_NO,
GNUNET_TIME_UNIT_FOREVER_REL,
ntohs (msg->size),
- cadet_mq_ntr, mq);
+ &cadet_mq_ntr, mq);
}
* @param impl_state state of the implementation
*/
static void
-cadet_mq_destroy_impl (struct GNUNET_MQ_Handle *mq, void *impl_state)
+cadet_mq_destroy_impl (struct GNUNET_MQ_Handle *mq,
+ void *impl_state)
{
struct CadetMQState *state = impl_state;
state = GNUNET_new (struct CadetMQState);
state->channel = channel;
- mq = GNUNET_MQ_queue_for_callbacks (cadet_mq_send_impl,
- cadet_mq_destroy_impl,
+ mq = GNUNET_MQ_queue_for_callbacks (&cadet_mq_send_impl,
+ &cadet_mq_destroy_impl,
NULL, /* FIXME: cancel impl. */
state,
NULL, /* no msg handlers */
NULL); /* no handler cls */
return mq;
}
-
projects / oweals / gnunet.git / blobdiff