*/
#include "platform.h"
#include "gnunet_util_lib.h"
+#include "gnunet_statistics_service.h"
#include "gnunet_transport_service.h"
-#include "gnunet_service_core.h"
-#include "gnunet_service_core_clients.h"
-#include "gnunet_service_core_sessions.h"
+#include "gnunet-service-core.h"
+#include "gnunet-service-core_clients.h"
+#include "gnunet-service-core_sessions.h"
+#include "gnunet-service-core_typemap.h"
+#include "core.h"
+
+#define DEBUG_CONNECTS GNUNET_YES
+
+/**
+ * How many messages do we queue up at most for optional
+ * notifications to a client? (this can cause notifications
+ * about outgoing messages to be dropped).
+ */
+#define MAX_NOTIFY_QUEUE 1024
/**
* Data structure for each client connected to the core service.
*/
-struct Client
+struct GSC_Client
{
/**
* Clients are kept in a linked list.
*/
- struct Client *next;
+ struct GSC_Client *next;
/**
* Clients are kept in a linked list.
*/
- struct Client *prev;
+ struct GSC_Client *prev;
/**
* Handle for the client with the server API.
/**
* Map of peer identities to active transmission requests of this
- * client to the peer (of type 'struct ClientActiveRequest').
+ * client to the peer (of type 'struct GSC_ClientActiveRequest').
*/
struct GNUNET_CONTAINER_MultiHashMap *requests;
+#if DEBUG_CONNECTS
+ /**
+ * Map containing all peers that this client knows we're connected to.
+ */
+ struct GNUNET_CONTAINER_MultiHashMap *connectmap;
+#endif
+
/**
* Options for messages this client cares about,
* see GNUNET_CORE_OPTION_ values.
/**
- * Record kept for each request for transmission issued by a
- * client that is still pending.
+ * Head of linked list of our clients.
*/
-struct ClientActiveRequest
-{
-
- /**
- * Active requests are kept in a doubly-linked list of
- * the respective target peer.
- */
- struct ClientActiveRequest *next;
-
- /**
- * Active requests are kept in a doubly-linked list of
- * the respective target peer.
- */
- struct ClientActiveRequest *prev;
-
- /**
- * Handle to the client.
- */
- struct Client *client;
-
- /**
- * By what time would the client want to see this message out?
- */
- struct GNUNET_TIME_Absolute deadline;
-
- /**
- * How important is this request.
- */
- uint32_t priority;
-
- /**
- * How many more requests does this client have?
- */
- uint32_t queue_size;
-
- /**
- * How many bytes does the client intend to send?
- */
- uint16_t msize;
-
- /**
- * Unique request ID (in big endian).
- */
- uint16_t smr_id;
-
-};
-
-
+static struct GSC_Client *client_head;
/**
- * Linked list of our clients.
+ * Tail of linked list of our clients.
*/
-static struct Client *clients;
+static struct GSC_Client *client_tail;
/**
* Context for notifications we need to send to our clients.
*/
static struct GNUNET_SERVER_NotificationContext *notifier;
+/**
+ * Tokenizer for messages received from clients.
+ */
+static struct GNUNET_SERVER_MessageStreamTokenizer *client_mst;
+
/**
- * Our message stream tokenizer (for encrypted payload).
+ * Lookup our client struct given the server's client handle.
+ *
+ * @param client server client handle to look up
+ * @return our client handle for the client
*/
-static struct GNUNET_SERVER_MessageStreamTokenizer *mst;
+static struct GSC_Client *
+find_client (struct GNUNET_SERVER_Client *client)
+{
+ struct GSC_Client *c;
+ c = client_head;
+ while ((c != NULL) && (c->client_handle != client))
+ c = c->next;
+ return c;
+}
/**
* client's queue is getting too large?
*/
static void
-send_to_client (struct Client *client, const struct GNUNET_MessageHeader *msg,
- int can_drop)
+send_to_client (struct GSC_Client *client,
+ const struct GNUNET_MessageHeader *msg, int can_drop)
{
-#if DEBUG_CORE_CLIENT
+#if DEBUG_CORE
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Preparing to send %u bytes of message of type %u to client.\n",
(unsigned int) ntohs (msg->size),
}
-
-
-
/**
- * Send a message to all of our current clients that have
- * the right options set.
+ * Send a message to one of our clients.
*
- * @param msg message to multicast
- * @param can_drop can this message be discarded if the queue is too long
- * @param options mask to use
+ * @param client target for the message
+ * @param msg message to transmit
+ * @param can_drop could this message be dropped if the
+ * client's queue is getting too large?
*/
-static void
-send_to_all_clients (const struct GNUNET_MessageHeader *msg, int can_drop,
- int options)
+void
+GSC_CLIENTS_send_to_client (struct GNUNET_SERVER_Client *client,
+ const struct GNUNET_MessageHeader *msg,
+ int can_drop)
{
- struct Client *c;
+ struct GSC_Client *c;
- c = clients;
- while (c != NULL)
+ c = find_client (client);
+ if (NULL == c)
{
- if (0 != (c->options & options))
- {
-#if DEBUG_CORE_CLIENT > 1
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Sending message of type %u to client.\n",
- (unsigned int) ntohs (msg->type));
-#endif
- send_to_client (c, msg, can_drop);
- }
- c = c->next;
+ GNUNET_break (0);
+ return;
}
+ send_to_client (c, msg, can_drop);
}
-
/**
- * Handle CORE_SEND_REQUEST message.
+ * Test if the client is interested in messages of the given type.
+ *
+ * @param type message type
+ * @param c client to test
+ * @return GNUNET_YES if 'c' is interested, GNUNET_NO if not.
*/
-static void
-handle_client_send_request (void *cls, struct GNUNET_SERVER_Client *client,
- const struct GNUNET_MessageHeader *message)
+static int
+type_match (uint16_t type, struct GSC_Client *c)
{
- const struct SendMessageRequest *req;
- struct Neighbour *n;
- struct Client *c;
- struct ClientActiveRequest *car;
+ unsigned int i;
- req = (const struct SendMessageRequest *) message;
- if (0 ==
- memcmp (&req->peer, &my_identity, sizeof (struct GNUNET_PeerIdentity)))
- n = &self;
- else
- n = find_neighbour (&req->peer);
- if ((n == NULL) || (GNUNET_YES != n->is_connected) ||
- (n->status != PEER_STATE_KEY_CONFIRMED))
- {
- /* neighbour must have disconnected since request was issued,
- * ignore (client will realize it once it processes the
- * disconnect notification) */
-#if DEBUG_CORE_CLIENT
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Dropped client request for transmission (am disconnected)\n");
-#endif
- GNUNET_STATISTICS_update (stats,
- gettext_noop
- ("# send requests dropped (disconnected)"), 1,
- GNUNET_NO);
- GNUNET_SERVER_receive_done (client, GNUNET_OK);
- return;
- }
- c = clients;
- while ((c != NULL) && (c->client_handle != client))
- c = c->next;
- if (c == NULL)
- {
- /* client did not send INIT first! */
- GNUNET_break (0);
- GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
- return;
- }
- if (c->requests == NULL)
- c->requests = GNUNET_CONTAINER_multihashmap_create (16);
-#if DEBUG_CORE_CLIENT
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Received client transmission request. queueing\n");
-#endif
- car = GNUNET_CONTAINER_multihashmap_get (c->requests, &req->peer.hashPubKey);
- if (car == NULL)
- {
- /* create new entry */
- car = GNUNET_malloc (sizeof (struct ClientActiveRequest));
- GNUNET_assert (GNUNET_OK ==
- GNUNET_CONTAINER_multihashmap_put (c->requests,
- &req->peer.hashPubKey,
- car,
- GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
- GNUNET_CONTAINER_DLL_insert (n->active_client_request_head,
- n->active_client_request_tail, car);
- car->client = c;
- }
- car->deadline = GNUNET_TIME_absolute_ntoh (req->deadline);
- car->priority = ntohl (req->priority);
- car->queue_size = ntohl (req->queue_size);
- car->msize = ntohs (req->size);
- car->smr_id = req->smr_id;
- schedule_peer_messages (n);
- GNUNET_SERVER_receive_done (client, GNUNET_OK);
+ if (c->tcnt == 0)
+ return GNUNET_YES; /* peer without handlers matches ALL */
+ for (i = 0; i < c->tcnt; i++)
+ if (type == c->types[i])
+ return GNUNET_YES;
+ return GNUNET_NO;
}
/**
- * Notify client about an existing connection to one of our neighbours.
+ * Send a message to all of our current clients that have the right
+ * options set.
+ *
+ * @param msg message to multicast
+ * @param can_drop can this message be discarded if the queue is too long
+ * @param options mask to use
+ * @param type type of the embedded message, 0 for none
*/
-static int
-notify_client_about_neighbour (void *cls, const GNUNET_HashCode * key,
- void *value)
+static void
+send_to_all_clients (const struct GNUNET_PeerIdentity *sender,
+ const struct GNUNET_MessageHeader *msg, int can_drop,
+ int options, uint16_t type)
{
- struct Client *c = cls;
- struct Neighbour *n = value;
- size_t size;
- char buf[GNUNET_SERVER_MAX_MESSAGE_SIZE - 1];
- struct GNUNET_TRANSPORT_ATS_Information *ats;
- struct ConnectNotifyMessage *cnm;
+ struct GSC_Client *c;
- size =
- sizeof (struct ConnectNotifyMessage) +
- (n->ats_count) * sizeof (struct GNUNET_TRANSPORT_ATS_Information);
- if (size >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
- {
- GNUNET_break (0);
- /* recovery strategy: throw away performance data */
- GNUNET_array_grow (n->ats, n->ats_count, 0);
- size =
- sizeof (struct ConnectNotifyMessage) +
- (n->ats_count) * sizeof (struct GNUNET_TRANSPORT_ATS_Information);
- }
- cnm = (struct ConnectNotifyMessage *) buf;
- cnm->header.size = htons (size);
- cnm->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_NOTIFY_CONNECT);
- cnm->ats_count = htonl (n->ats_count);
- ats = &cnm->ats;
- memcpy (ats, n->ats,
- sizeof (struct GNUNET_TRANSPORT_ATS_Information) * n->ats_count);
- ats[n->ats_count].type = htonl (GNUNET_TRANSPORT_ATS_ARRAY_TERMINATOR);
- ats[n->ats_count].value = htonl (0);
- if (n->status == PEER_STATE_KEY_CONFIRMED)
+ for (c = client_head; c != NULL; c = c->next)
{
-#if DEBUG_CORE_CLIENT
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending `%s' message to client.\n",
- "NOTIFY_CONNECT");
+ if ((0 == (options & GNUNET_CORE_OPTION_SEND_FULL_INBOUND)) &&
+ (GNUNET_YES == type_match (type, c)))
+ continue; /* not the full message, but we'd like the full one! */
+ if ((0 == (c->options & options)) && (GNUNET_YES != type_match (type, c)))
+ continue; /* neither options nor type match permit the message */
+#if DEBUG_CORE
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Sending message to client interested in messages of type %u.\n",
+ (unsigned int) type);
#endif
- cnm->peer = n->peer;
- send_to_client (c, &cnm->header, GNUNET_NO);
+#if DEBUG_CONNECTS
+ GNUNET_assert (GNUNET_YES ==
+ GNUNET_CONTAINER_multihashmap_contains (c->connectmap,
+ &sender->hashPubKey));
+#endif
+ send_to_client (c, msg, can_drop);
}
- return GNUNET_OK;
}
-
/**
* Handle CORE_INIT request.
+ *
+ * @param cls unused
+ * @param client new client that sent INIT
+ * @param message the 'struct InitMessage' (presumably)
*/
static void
handle_client_init (void *cls, struct GNUNET_SERVER_Client *client,
{
const struct InitMessage *im;
struct InitReplyMessage irm;
- struct Client *c;
+ struct GSC_Client *c;
uint16_t msize;
const uint16_t *types;
uint16_t *wtypes;
unsigned int i;
-#if DEBUG_CORE_CLIENT
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Client connecting to core service with `%s' message\n", "INIT");
-#endif
/* check that we don't have an entry already */
- c = clients;
- while (c != NULL)
+ c = find_client (client);
+ if (NULL != c)
{
- if (client == c->client_handle)
- {
- GNUNET_break (0);
- GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
- return;
- }
- c = c->next;
+ GNUNET_break (0);
+ GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+ return;
}
msize = ntohs (message->size);
if (msize < sizeof (struct InitMessage))
im = (const struct InitMessage *) message;
types = (const uint16_t *) &im[1];
msize -= sizeof (struct InitMessage);
- c = GNUNET_malloc (sizeof (struct Client) + msize);
+ c = GNUNET_malloc (sizeof (struct GSC_Client) + msize);
c->client_handle = client;
- c->next = clients;
- clients = c;
c->tcnt = msize / sizeof (uint16_t);
+ c->options = ntohl (im->options);
c->types = (const uint16_t *) &c[1];
+#if DEBUG_CONNECTS
+ c->connectmap = GNUNET_CONTAINER_multihashmap_create (16);
+ GNUNET_assert (GNUNET_YES ==
+ GNUNET_CONTAINER_multihashmap_put (c->connectmap,
+ &GSC_my_identity.hashPubKey,
+ NULL,
+ GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
+#endif
+
wtypes = (uint16_t *) & c[1];
for (i = 0; i < c->tcnt; i++)
- {
wtypes[i] = ntohs (types[i]);
- my_type_map[wtypes[i] / 32] |= (1 << (wtypes[i] % 32));
- }
- if (c->tcnt > 0)
- broadcast_my_type_map ();
- c->options = ntohl (im->options);
-#if DEBUG_CORE_CLIENT
+ GSC_TYPEMAP_add (wtypes, c->tcnt);
+ GNUNET_CONTAINER_DLL_insert (client_head, client_tail, c);
+#if DEBUG_CORE
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Client %p is interested in %u message types\n", c,
+ "Client connecting to core service is interested in %u message types\n",
(unsigned int) c->tcnt);
#endif
/* send init reply message */
irm.header.size = htons (sizeof (struct InitReplyMessage));
irm.header.type = htons (GNUNET_MESSAGE_TYPE_CORE_INIT_REPLY);
irm.reserved = htonl (0);
- memcpy (&irm.publicKey, &my_public_key,
- sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded));
-#if DEBUG_CORE_CLIENT
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending `%s' message to client.\n",
- "INIT_REPLY");
-#endif
+ irm.my_identity = GSC_my_identity;
send_to_client (c, &irm.header, GNUNET_NO);
- if (0 != (c->options & GNUNET_CORE_OPTION_SEND_CONNECT))
- {
- /* notify new client about existing neighbours */
- GNUNET_CONTAINER_multihashmap_iterate (neighbours,
- ¬ify_client_about_neighbour, c);
- }
+ GSC_SESSIONS_notify_client_about_sessions (c);
GNUNET_SERVER_receive_done (client, GNUNET_OK);
}
/**
- * Free client request records.
- *
- * @param cls NULL
- * @param key identity of peer for which this is an active request
- * @param value the 'struct ClientActiveRequest' to free
- * @return GNUNET_YES (continue iteration)
- */
-static int
-destroy_active_client_request (void *cls, const GNUNET_HashCode * key,
- void *value)
-{
- struct ClientActiveRequest *car = value;
- struct Neighbour *n;
- struct GNUNET_PeerIdentity peer;
-
- peer.hashPubKey = *key;
- n = find_neighbour (&peer);
- GNUNET_assert (NULL != n);
- GNUNET_CONTAINER_DLL_remove (n->active_client_request_head,
- n->active_client_request_tail, car);
- GNUNET_free (car);
- return GNUNET_YES;
-}
-
-
-/**
- * A client disconnected, clean up.
+ * Handle CORE_SEND_REQUEST message.
*
- * @param cls closure
- * @param client identification of the client
+ * @param cls unused
+ * @param client new client that sent CORE_SEND_REQUEST
+ * @param message the 'struct SendMessageRequest' (presumably)
*/
static void
-handle_client_disconnect (void *cls, struct GNUNET_SERVER_Client *client)
+handle_client_send_request (void *cls, struct GNUNET_SERVER_Client *client,
+ const struct GNUNET_MessageHeader *message)
{
- struct Client *pos;
- struct Client *prev;
- unsigned int i;
- const uint16_t *wtypes;
+ const struct SendMessageRequest *req;
+ struct GSC_Client *c;
+ struct GSC_ClientActiveRequest *car;
- if (client == NULL)
+ req = (const struct SendMessageRequest *) message;
+ c = find_client (client);
+ if (c == NULL)
+ {
+ /* client did not send INIT first! */
+ GNUNET_break (0);
+ GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
return;
-#if DEBUG_CORE_CLIENT
+ }
+ if (c->requests == NULL)
+ c->requests = GNUNET_CONTAINER_multihashmap_create (16);
+#if DEBUG_CORE
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Client %p has disconnected from core service.\n", client);
+ "Client asked for transmission to `%s'\n",
+ GNUNET_i2s (&req->peer));
#endif
- prev = NULL;
- pos = clients;
- while (pos != NULL)
- {
- if (client == pos->client_handle)
- break;
- prev = pos;
- pos = pos->next;
- }
- if (pos == NULL)
+ car = GNUNET_CONTAINER_multihashmap_get (c->requests, &req->peer.hashPubKey);
+ if (car == NULL)
{
- /* client never sent INIT */
- return;
+ /* create new entry */
+ car = GNUNET_malloc (sizeof (struct GSC_ClientActiveRequest));
+ GNUNET_assert (GNUNET_OK ==
+ GNUNET_CONTAINER_multihashmap_put (c->requests,
+ &req->peer.hashPubKey,
+ car,
+ GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
+ car->client_handle = c;
}
- if (prev == NULL)
- clients = pos->next;
else
- prev->next = pos->next;
- if (pos->requests != NULL)
{
- GNUNET_CONTAINER_multihashmap_iterate (pos->requests,
- &destroy_active_client_request,
- NULL);
- GNUNET_CONTAINER_multihashmap_destroy (pos->requests);
- }
- GNUNET_free (pos);
-
- /* rebuild my_type_map */
- memset (my_type_map, 0, sizeof (my_type_map));
- for (pos = clients; NULL != pos; pos = pos->next)
- {
- wtypes = (const uint16_t *) &pos[1];
- for (i = 0; i < pos->tcnt; i++)
- my_type_map[wtypes[i] / 32] |= (1 << (wtypes[i] % 32));
+ GSC_SESSIONS_dequeue_request (car);
}
- broadcast_my_type_map ();
+ car->target = req->peer;
+ car->deadline = GNUNET_TIME_absolute_ntoh (req->deadline);
+ car->priority = ntohl (req->priority);
+ car->msize = ntohs (req->size);
+ car->smr_id = req->smr_id;
+ car->was_solicited = GNUNET_NO;
+ if (0 ==
+ memcmp (&req->peer, &GSC_my_identity,
+ sizeof (struct GNUNET_PeerIdentity)))
+ GSC_CLIENTS_solicit_request (car);
+ else
+ GSC_SESSIONS_queue_request (car);
+ GNUNET_SERVER_receive_done (client, GNUNET_OK);
}
+/**
+ * Closure for the 'client_tokenizer_callback'.
+ */
+struct TokenizerContext
+{
+
+ /**
+ * Active request handle for the message.
+ */
+ struct GSC_ClientActiveRequest *car;
+
+ /**
+ * Is corking allowed (set only once we have the real message).
+ */
+ int cork;
+};
/**
const struct GNUNET_MessageHeader *message)
{
const struct SendMessage *sm;
- struct Neighbour *n;
- struct MessageEntry *prev;
- struct MessageEntry *pos;
- struct MessageEntry *e;
- struct MessageEntry *min_prio_entry;
- struct MessageEntry *min_prio_prev;
- unsigned int min_prio;
- unsigned int queue_size;
+ struct GSC_Client *c;
+ struct TokenizerContext tc;
uint16_t msize;
msize = ntohs (message->size);
if (msize <
sizeof (struct SendMessage) + sizeof (struct GNUNET_MessageHeader))
{
- GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
- "msize is %u, should be at least %u (in %s:%d)\n", msize,
- sizeof (struct SendMessage) +
- sizeof (struct GNUNET_MessageHeader), __FILE__, __LINE__);
GNUNET_break (0);
- if (client != NULL)
- GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+ GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
return;
}
sm = (const struct SendMessage *) message;
msize -= sizeof (struct SendMessage);
- if (0 ==
- memcmp (&sm->peer, &my_identity, sizeof (struct GNUNET_PeerIdentity)))
+ GNUNET_break (0 == ntohl (sm->reserved));
+ c = find_client (client);
+ if (c == NULL)
{
- /* loopback */
- GNUNET_SERVER_mst_receive (mst, &self, (const char *) &sm[1], msize,
- GNUNET_YES, GNUNET_NO);
- if (client != NULL)
- GNUNET_SERVER_receive_done (client, GNUNET_OK);
+ /* client did not send INIT first! */
+ GNUNET_break (0);
+ GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
return;
}
- n = find_neighbour (&sm->peer);
- if ((n == NULL) || (GNUNET_YES != n->is_connected) ||
- (n->status != PEER_STATE_KEY_CONFIRMED))
+ tc.car =
+ GNUNET_CONTAINER_multihashmap_get (c->requests, &sm->peer.hashPubKey);
+ if (NULL == tc.car)
{
- /* attempt to send message to peer that is not connected anymore
- * (can happen due to asynchrony) */
- GNUNET_STATISTICS_update (stats,
+ /* Must have been that we first approved the request, then got disconnected
+ * (which triggered removal of the 'car') and now the client gives us a message
+ * just *before* the client learns about the disconnect. Theoretically, we
+ * might also now be *again* connected. So this can happen (but should be
+ * rare). If it does happen, the message is discarded. */
+ GNUNET_STATISTICS_update (GSC_stats,
gettext_noop
- ("# messages discarded (disconnected)"), 1,
- GNUNET_NO);
- if (client != NULL)
- GNUNET_SERVER_receive_done (client, GNUNET_OK);
+ ("# messages discarded (session disconnected)"),
+ 1, GNUNET_NO);
+ GNUNET_SERVER_receive_done (client, GNUNET_OK);
return;
}
+ GNUNET_assert (GNUNET_YES ==
+ GNUNET_CONTAINER_multihashmap_remove (c->requests,
+ &sm->peer.hashPubKey,
+ tc.car));
+ tc.cork = ntohl (sm->cork);
#if DEBUG_CORE
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Core received `%s' request, queueing %u bytes of plaintext data for transmission to `%4s'.\n",
- "SEND", (unsigned int) msize, GNUNET_i2s (&sm->peer));
-#endif
- discard_expired_messages (n);
- /* bound queue size */
- /* NOTE: this entire block to bound the queue size should be
- * obsolete with the new client-request code and the
- * 'schedule_peer_messages' mechanism; we still have this code in
- * here for now as a sanity check for the new mechanmism;
- * ultimately, we should probably simply reject SEND messages that
- * are not 'approved' (or provide a new core API for very unreliable
- * delivery that always sends with priority 0). Food for thought. */
- min_prio = UINT32_MAX;
- min_prio_entry = NULL;
- min_prio_prev = NULL;
- queue_size = 0;
- prev = NULL;
- pos = n->messages;
- while (pos != NULL)
- {
- if (pos->priority <= min_prio)
- {
- min_prio_entry = pos;
- min_prio_prev = prev;
- min_prio = pos->priority;
- }
- queue_size++;
- prev = pos;
- pos = pos->next;
- }
- if (queue_size >= MAX_PEER_QUEUE_SIZE)
- {
- /* queue full */
- if (ntohl (sm->priority) <= min_prio)
- {
- /* discard new entry; this should no longer happen! */
- GNUNET_break (0);
-#if DEBUG_CORE
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Queue full (%u/%u), discarding new request (%u bytes of type %u)\n",
- queue_size, (unsigned int) MAX_PEER_QUEUE_SIZE,
- (unsigned int) msize, (unsigned int) ntohs (message->type));
+ "Client asked for transmission of %u bytes to `%s' %s\n", msize,
+ GNUNET_i2s (&sm->peer), tc.cork ? "now" : "");
#endif
- GNUNET_STATISTICS_update (stats,
- gettext_noop ("# discarded CORE_SEND requests"),
- 1, GNUNET_NO);
+ GNUNET_SERVER_mst_receive (client_mst, &tc, (const char *) &sm[1], msize,
+ GNUNET_YES, GNUNET_NO);
+ if (0 !=
+ memcmp (&tc.car->target, &GSC_my_identity,
+ sizeof (struct GNUNET_PeerIdentity)))
+ GSC_SESSIONS_dequeue_request (tc.car);
+ GNUNET_free (tc.car);
+ GNUNET_SERVER_receive_done (client, GNUNET_OK);
+}
- if (client != NULL)
- GNUNET_SERVER_receive_done (client, GNUNET_OK);
- return;
- }
- GNUNET_assert (min_prio_entry != NULL);
- /* discard "min_prio_entry" */
+
+/**
+ * Functions with this signature are called whenever a complete
+ * message is received by the tokenizer. Used by the 'client_mst' for
+ * dispatching messages from clients to either the SESSION subsystem
+ * or other CLIENT (for loopback).
+ *
+ * @param cls closure
+ * @param client reservation request ('struct GSC_ClientActiveRequest')
+ * @param message the actual message
+ */
+static void
+client_tokenizer_callback (void *cls, void *client,
+ const struct GNUNET_MessageHeader *message)
+{
+ struct TokenizerContext *tc = client;
+ struct GSC_ClientActiveRequest *car = tc->car;
+
+ if (0 ==
+ memcmp (&car->target, &GSC_my_identity,
+ sizeof (struct GNUNET_PeerIdentity)))
+ {
#if DEBUG_CORE
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Queue full, discarding existing older request\n");
+ "Delivering message of type %u to myself\n",
+ ntohs (message->type));
#endif
- GNUNET_STATISTICS_update (stats,
- gettext_noop
- ("# discarded lower priority CORE_SEND requests"),
- 1, GNUNET_NO);
- if (min_prio_prev == NULL)
- n->messages = min_prio_entry->next;
- else
- min_prio_prev->next = min_prio_entry->next;
- GNUNET_free (min_prio_entry);
+ GSC_CLIENTS_deliver_message (&GSC_my_identity, NULL, 0, message,
+ ntohs (message->size),
+ GNUNET_CORE_OPTION_SEND_FULL_INBOUND |
+ GNUNET_CORE_OPTION_SEND_FULL_OUTBOUND);
+ GSC_CLIENTS_deliver_message (&GSC_my_identity, NULL, 0, message,
+ sizeof (struct GNUNET_MessageHeader),
+ GNUNET_CORE_OPTION_SEND_HDR_INBOUND |
+ GNUNET_CORE_OPTION_SEND_HDR_OUTBOUND);
}
-
+ else
+ {
#if DEBUG_CORE
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Adding transmission request for `%4s' of size %u to queue\n",
- GNUNET_i2s (&sm->peer), (unsigned int) msize);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Delivering message of type %u to %s\n", ntohs (message->type),
+ GNUNET_i2s (&car->target));
#endif
- GNUNET_break (0 == ntohl (sm->reserved));
- e = GNUNET_malloc (sizeof (struct MessageEntry) + msize);
- e->deadline = GNUNET_TIME_absolute_ntoh (sm->deadline);
- e->priority = ntohl (sm->priority);
- e->size = msize;
- if (GNUNET_YES != (int) ntohl (sm->cork))
- e->got_slack = GNUNET_YES;
- memcpy (&e[1], &sm[1], msize);
-
- /* insert, keep list sorted by deadline */
- prev = NULL;
- pos = n->messages;
- while ((pos != NULL) && (pos->deadline.abs_value < e->deadline.abs_value))
- {
- prev = pos;
- pos = pos->next;
+ GSC_SESSIONS_transmit (car, message, tc->cork);
}
- if (prev == NULL)
- n->messages = e;
- else
- prev->next = e;
- e->next = pos;
-
- /* consider scheduling now */
- process_plaintext_neighbour_queue (n);
- if (client != NULL)
- GNUNET_SERVER_receive_done (client, GNUNET_OK);
}
/**
- * Helper function for handle_client_iterate_peers.
+ * Free client request records.
*
- * @param cls the 'struct GNUNET_SERVER_TransmitContext' to queue replies
- * @param key identity of the connected peer
- * @param value the 'struct Neighbour' for the peer
- * @return GNUNET_OK (continue to iterate)
+ * @param cls NULL
+ * @param key identity of peer for which this is an active request
+ * @param value the 'struct GSC_ClientActiveRequest' to free
+ * @return GNUNET_YES (continue iteration)
*/
static int
-queue_connect_message (void *cls, const GNUNET_HashCode * key, void *value)
+destroy_active_client_request (void *cls, const GNUNET_HashCode * key,
+ void *value)
{
- struct GNUNET_SERVER_TransmitContext *tc = cls;
- struct Neighbour *n = value;
- char buf[GNUNET_SERVER_MAX_MESSAGE_SIZE - 1];
- struct GNUNET_TRANSPORT_ATS_Information *ats;
- size_t size;
- struct ConnectNotifyMessage *cnm;
-
- cnm = (struct ConnectNotifyMessage *) buf;
- if (n->status != PEER_STATE_KEY_CONFIRMED)
- return GNUNET_OK;
- size =
- sizeof (struct ConnectNotifyMessage) +
- (n->ats_count) * sizeof (struct GNUNET_TRANSPORT_ATS_Information);
- if (size >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
- {
- GNUNET_break (0);
- /* recovery strategy: throw away performance data */
- GNUNET_array_grow (n->ats, n->ats_count, 0);
- size =
- sizeof (struct PeerStatusNotifyMessage) +
- n->ats_count * sizeof (struct GNUNET_TRANSPORT_ATS_Information);
- }
- cnm = (struct ConnectNotifyMessage *) buf;
- cnm->header.size = htons (size);
- cnm->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_NOTIFY_CONNECT);
- cnm->ats_count = htonl (n->ats_count);
- ats = &cnm->ats;
- memcpy (ats, n->ats,
- n->ats_count * sizeof (struct GNUNET_TRANSPORT_ATS_Information));
- ats[n->ats_count].type = htonl (GNUNET_TRANSPORT_ATS_ARRAY_TERMINATOR);
- ats[n->ats_count].value = htonl (0);
-#if DEBUG_CORE_CLIENT
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending `%s' message to client.\n",
- "NOTIFY_CONNECT");
-#endif
- cnm->peer = n->peer;
- GNUNET_SERVER_transmit_context_append_message (tc, &cnm->header);
- return GNUNET_OK;
+ struct GSC_ClientActiveRequest *car = value;
+
+ GNUNET_assert (GNUNET_YES ==
+ GNUNET_CONTAINER_multihashmap_remove (car->
+ client_handle->requests,
+ &car->target.hashPubKey,
+ car));
+ GSC_SESSIONS_dequeue_request (car);
+ GNUNET_free (car);
+ return GNUNET_YES;
}
/**
- * Handle CORE_ITERATE_PEERS request.
+ * A client disconnected, clean up.
*
- * @param cls unused
- * @param client client sending the iteration request
- * @param message iteration request message
+ * @param cls closure
+ * @param client identification of the client
*/
static void
-handle_client_iterate_peers (void *cls, struct GNUNET_SERVER_Client *client,
- const struct GNUNET_MessageHeader *message)
+handle_client_disconnect (void *cls, struct GNUNET_SERVER_Client *client)
{
- struct GNUNET_MessageHeader done_msg;
- struct GNUNET_SERVER_TransmitContext *tc;
- int msize;
+ struct GSC_Client *c;
- /* notify new client about existing neighbours */
+ if (client == NULL)
+ return;
+#if DEBUG_CORE
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Client %p has disconnected from core service.\n", client);
+#endif
+ c = find_client (client);
+ if (c == NULL)
+ return; /* client never sent INIT */
+ GNUNET_CONTAINER_DLL_remove (client_head, client_tail, c);
+ if (c->requests != NULL)
+ {
+ GNUNET_CONTAINER_multihashmap_iterate (c->requests,
+ &destroy_active_client_request,
+ NULL);
+ GNUNET_CONTAINER_multihashmap_destroy (c->requests);
+ }
+#if DEBUG_CONNECTS
+ GNUNET_CONTAINER_multihashmap_destroy (c->connectmap);
+#endif
+ GSC_TYPEMAP_remove (c->types, c->tcnt);
+ GNUNET_free (c);
+}
- msize = ntohs (message->size);
- tc = GNUNET_SERVER_transmit_context_create (client);
- if (msize == sizeof (struct GNUNET_MessageHeader))
- GNUNET_CONTAINER_multihashmap_iterate (neighbours, &queue_connect_message,
- tc);
- else
- GNUNET_break (0);
- done_msg.size = htons (sizeof (struct GNUNET_MessageHeader));
- done_msg.type = htons (GNUNET_MESSAGE_TYPE_CORE_ITERATE_PEERS_END);
- GNUNET_SERVER_transmit_context_append_message (tc, &done_msg);
- GNUNET_SERVER_transmit_context_run (tc, GNUNET_TIME_UNIT_FOREVER_REL);
+/**
+ * Tell a client that we are ready to receive the message.
+ *
+ * @param car request that is now ready; the responsibility
+ * for the handle remains shared between CLIENTS
+ * and SESSIONS after this call.
+ */
+void
+GSC_CLIENTS_solicit_request (struct GSC_ClientActiveRequest *car)
+{
+ struct GSC_Client *c;
+ struct SendMessageReady smr;
+
+ c = car->client_handle;
+ smr.header.size = htons (sizeof (struct SendMessageReady));
+ smr.header.type = htons (GNUNET_MESSAGE_TYPE_CORE_SEND_READY);
+ smr.size = htons (car->msize);
+ smr.smr_id = car->smr_id;
+ smr.peer = car->target;
+#if DEBUG_CONNECTS
+ GNUNET_assert (GNUNET_YES ==
+ GNUNET_CONTAINER_multihashmap_contains (c->connectmap,
+ &car->
+ target.hashPubKey));
+#endif
+ send_to_client (c, &smr.header, GNUNET_NO);
}
/**
- * Handle CORE_PEER_CONNECTED request. Notify client about existing neighbours.
+ * Tell a client that we will never be ready to receive the
+ * given message in time (disconnect or timeout).
*
- * @param cls unused
- * @param client client sending the iteration request
- * @param message iteration request message
+ * @param car request that now permanently failed; the
+ * responsibility for the handle is now returned
+ * to CLIENTS (SESSIONS is done with it).
*/
-static void
-handle_client_have_peer (void *cls, struct GNUNET_SERVER_Client *client,
- const struct GNUNET_MessageHeader *message)
+void
+GSC_CLIENTS_reject_request (struct GSC_ClientActiveRequest *car)
{
- struct GNUNET_MessageHeader done_msg;
- struct GNUNET_SERVER_TransmitContext *tc;
- struct GNUNET_PeerIdentity *peer;
-
- tc = GNUNET_SERVER_transmit_context_create (client);
- peer = (struct GNUNET_PeerIdentity *) &message[1];
- GNUNET_CONTAINER_multihashmap_get_multiple (neighbours, &peer->hashPubKey,
- &queue_connect_message, tc);
- done_msg.size = htons (sizeof (struct GNUNET_MessageHeader));
- done_msg.type = htons (GNUNET_MESSAGE_TYPE_CORE_ITERATE_PEERS_END);
- GNUNET_SERVER_transmit_context_append_message (tc, &done_msg);
- GNUNET_SERVER_transmit_context_run (tc, GNUNET_TIME_UNIT_FOREVER_REL);
+ GNUNET_assert (GNUNET_YES ==
+ GNUNET_CONTAINER_multihashmap_remove (car->
+ client_handle->requests,
+ &car->target.hashPubKey,
+ car));
+ GNUNET_free (car);
}
/**
- * Handle REQUEST_INFO request.
+ * Notify a particular client about a change to existing connection to
+ * one of our neighbours (check if the client is interested). Called
+ * from 'GSC_SESSIONS_notify_client_about_sessions'.
*
- * @param cls unused
- * @param client client sending the request
- * @param message iteration request message
+ * @param client client to notify
+ * @param neighbour identity of the neighbour that changed status
+ * @param atsi performance information about neighbour
+ * @param atsi_count number of entries in 'ats' array
+ * @param tmap_old previous type map for the neighbour, NULL for disconnect
+ * @param tmap_new updated type map for the neighbour, NULL for disconnect
+ * @param is_new GNUNET_YES if this is a completely new neighbour
*/
-static void
-handle_client_request_info (void *cls, struct GNUNET_SERVER_Client *client,
- const struct GNUNET_MessageHeader *message)
+void
+GSC_CLIENTS_notify_client_about_neighbour (struct GSC_Client *client,
+ const struct GNUNET_PeerIdentity
+ *neighbour,
+ const struct GNUNET_ATS_Information
+ *atsi, unsigned int atsi_count,
+ const struct GSC_TypeMap *tmap_old,
+ const struct GSC_TypeMap *tmap_new)
{
- const struct RequestInfoMessage *rcm;
- struct Client *pos;
- struct Neighbour *n;
- struct ConfigurationInfoMessage cim;
- int32_t want_reserv;
- int32_t got_reserv;
- unsigned long long old_preference;
- struct GNUNET_TIME_Relative rdelay;
-
- rdelay = GNUNET_TIME_relative_get_zero ();
-#if DEBUG_CORE_CLIENT
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Core service receives `%s' request.\n",
- "REQUEST_INFO");
-#endif
- pos = clients;
- while (pos != NULL)
- {
- if (client == pos->client_handle)
- break;
- pos = pos->next;
- }
- if (pos == NULL)
+ struct ConnectNotifyMessage *cnm;
+ size_t size;
+ char buf[GNUNET_SERVER_MAX_MESSAGE_SIZE - 1];
+ struct GNUNET_ATS_Information *a;
+ struct DisconnectNotifyMessage dcm;
+ int old_match;
+ int new_match;
+
+ old_match = GSC_TYPEMAP_test_match (tmap_old, client->types, client->tcnt);
+ new_match = GSC_TYPEMAP_test_match (tmap_new, client->types, client->tcnt);
+ if (old_match == new_match)
{
- GNUNET_break (0);
- GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
- return;
+ GNUNET_assert (old_match ==
+ GNUNET_CONTAINER_multihashmap_contains (client->connectmap,
+ &neighbour->hashPubKey));
+ return; /* no change */
}
-
- rcm = (const struct RequestInfoMessage *) message;
- n = find_neighbour (&rcm->peer);
- memset (&cim, 0, sizeof (cim));
- if ((n != NULL) && (GNUNET_YES == n->is_connected))
+ if (old_match == GNUNET_NO)
{
- want_reserv = ntohl (rcm->reserve_inbound);
- if (n->bw_out_internal_limit.value__ != rcm->limit_outbound.value__)
- {
- n->bw_out_internal_limit = rcm->limit_outbound;
- if (n->bw_out.value__ !=
- GNUNET_BANDWIDTH_value_min (n->bw_out_internal_limit,
- n->bw_out_external_limit).value__)
- {
- n->bw_out =
- GNUNET_BANDWIDTH_value_min (n->bw_out_internal_limit,
- n->bw_out_external_limit);
- GNUNET_BANDWIDTH_tracker_update_quota (&n->available_recv_window,
- n->bw_out);
- GNUNET_TRANSPORT_set_quota (transport, &n->peer, n->bw_in, n->bw_out);
- handle_peer_status_change (n);
- }
- }
- if (want_reserv < 0)
- {
- got_reserv = want_reserv;
- }
- else if (want_reserv > 0)
- {
- rdelay =
- GNUNET_BANDWIDTH_tracker_get_delay (&n->available_recv_window,
- want_reserv);
- if (rdelay.rel_value == 0)
- got_reserv = want_reserv;
- else
- got_reserv = 0; /* all or nothing */
- }
- else
- got_reserv = 0;
- GNUNET_BANDWIDTH_tracker_consume (&n->available_recv_window, got_reserv);
- old_preference = n->current_preference;
- n->current_preference += GNUNET_ntohll (rcm->preference_change);
- if (old_preference > n->current_preference)
+ /* send connect */
+#if DEBUG_CONNECTS
+ GNUNET_assert (GNUNET_NO ==
+ GNUNET_CONTAINER_multihashmap_contains (client->connectmap,
+ &neighbour->hashPubKey));
+ GNUNET_assert (GNUNET_YES ==
+ GNUNET_CONTAINER_multihashmap_put (client->connectmap,
+ &neighbour->hashPubKey,
+ NULL,
+ GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
+#endif
+ size =
+ sizeof (struct ConnectNotifyMessage) +
+ (atsi_count) * sizeof (struct GNUNET_ATS_Information);
+ if (size >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
{
- /* overflow; cap at maximum value */
- n->current_preference = ULLONG_MAX;
+ GNUNET_break (0);
+ /* recovery strategy: throw away performance data */
+ atsi_count = 0;
+ size = sizeof (struct ConnectNotifyMessage);
}
- update_preference_sum (n->current_preference - old_preference);
-#if DEBUG_CORE_QUOTA
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Received reservation request for %d bytes for peer `%4s', reserved %d bytes, suggesting delay of %llu ms\n",
- (int) want_reserv, GNUNET_i2s (&rcm->peer), (int) got_reserv,
- (unsigned long long) rdelay.rel_value);
+ cnm = (struct ConnectNotifyMessage *) buf;
+ cnm->header.size = htons (size);
+ cnm->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_NOTIFY_CONNECT);
+ cnm->ats_count = htonl (atsi_count);
+ a = (struct GNUNET_ATS_Information *) &cnm[1];
+ memcpy (a, atsi, sizeof (struct GNUNET_ATS_Information) * atsi_count);
+#if DEBUG_CORE
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending `%s' message to client.\n",
+ "NOTIFY_CONNECT");
#endif
- cim.reserved_amount = htonl (got_reserv);
- cim.reserve_delay = GNUNET_TIME_relative_hton (rdelay);
- cim.bw_out = n->bw_out;
- cim.preference = n->current_preference;
+ cnm->peer = *neighbour;
+ send_to_client (client, &cnm->header, GNUNET_NO);
}
else
{
- /* Technically, this COULD happen (due to asynchronous behavior),
- * but it should be rare, so we should generate an info event
- * to help diagnosis of serious errors that might be masked by this */
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- _
- ("Client asked for preference change with peer `%s', which is not connected!\n"),
- GNUNET_i2s (&rcm->peer));
- GNUNET_SERVER_receive_done (client, GNUNET_OK);
- return;
- }
- cim.header.size = htons (sizeof (struct ConfigurationInfoMessage));
- cim.header.type = htons (GNUNET_MESSAGE_TYPE_CORE_CONFIGURATION_INFO);
- cim.peer = rcm->peer;
- cim.rim_id = rcm->rim_id;
-#if DEBUG_CORE_CLIENT
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending `%s' message to client.\n",
- "CONFIGURATION_INFO");
+ /* send disconnect */
+#if DEBUG_CONNECTS
+ GNUNET_assert (GNUNET_YES ==
+ GNUNET_CONTAINER_multihashmap_contains (client->connectmap,
+ &neighbour->hashPubKey));
+ GNUNET_assert (GNUNET_YES ==
+ GNUNET_CONTAINER_multihashmap_remove (client->connectmap,
+ &neighbour->hashPubKey,
+ NULL));
#endif
- send_to_client (pos, &cim.header, GNUNET_NO);
- GNUNET_SERVER_receive_done (client, GNUNET_OK);
+ dcm.header.size = htons (sizeof (struct DisconnectNotifyMessage));
+ dcm.header.type = htons (GNUNET_MESSAGE_TYPE_CORE_NOTIFY_DISCONNECT);
+ dcm.reserved = htonl (0);
+ dcm.peer = *neighbour;
+ send_to_client (client, &dcm.header, GNUNET_NO);
+ }
}
+/**
+ * Notify all clients about a change to existing session.
+ * Called from SESSIONS whenever there is a change in sessions
+ * or types processed by the respective peer.
+ *
+ * @param neighbour identity of the neighbour that changed status
+ * @param atsi performance information about neighbour
+ * @param atsi_count number of entries in 'ats' array
+ * @param tmap_old previous type map for the neighbour, NULL for disconnect
+ * @param tmap_new updated type map for the neighbour, NULL for disconnect
+ */
+void
+GSC_CLIENTS_notify_clients_about_neighbour (const struct GNUNET_PeerIdentity
+ *neighbour,
+ const struct GNUNET_ATS_Information
+ *atsi, unsigned int atsi_count,
+ const struct GSC_TypeMap *tmap_old,
+ const struct GSC_TypeMap *tmap_new)
+{
+ struct GSC_Client *c;
+
+ for (c = client_head; c != NULL; c = c->next)
+ GSC_CLIENTS_notify_client_about_neighbour (c, neighbour, atsi, atsi_count,
+ tmap_old, tmap_new);
+}
/**
- * Send a P2P message to a client.
+ * Deliver P2P message to interested clients. Caller must have checked
+ * that the sending peer actually lists the given message type as one
+ * of its types.
*
- * @param sender who sent us the message?
- * @param client who should we give the message to?
- * @param m contains the message to transmit
- * @param msize number of bytes in buf to transmit
+ * @param sender peer who sent us the message
+ * @param atsi performance information about neighbour
+ * @param atsi_count number of entries in 'ats' array
+ * @param msg the message
+ * @param msize number of bytes to transmit
+ * @param options options for checking which clients should
+ * receive the message
*/
-static void
-send_p2p_message_to_client (struct Neighbour *sender, struct Client *client,
- const void *m, size_t msize)
+void
+GSC_CLIENTS_deliver_message (const struct GNUNET_PeerIdentity *sender,
+ const struct GNUNET_ATS_Information *atsi,
+ unsigned int atsi_count,
+ const struct GNUNET_MessageHeader *msg,
+ uint16_t msize, int options)
{
size_t size =
msize + sizeof (struct NotifyTrafficMessage) +
- (sender->ats_count) * sizeof (struct GNUNET_TRANSPORT_ATS_Information);
+ atsi_count * sizeof (struct GNUNET_ATS_Information);
char buf[size];
struct NotifyTrafficMessage *ntm;
- struct GNUNET_TRANSPORT_ATS_Information *ats;
+ struct GNUNET_ATS_Information *a;
- GNUNET_assert (GNUNET_YES == sender->is_connected);
- GNUNET_break (sender->status == PEER_STATE_KEY_CONFIRMED);
+ if (0 == options)
+ {
+ GNUNET_snprintf (buf, sizeof (buf),
+ gettext_noop ("# bytes of messages of type %u received"),
+ (unsigned int) ntohs (msg->type));
+ GNUNET_STATISTICS_update (GSC_stats, buf, msize, GNUNET_NO);
+ }
if (size >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
{
GNUNET_break (0);
/* recovery strategy: throw performance data away... */
- GNUNET_array_grow (sender->ats, sender->ats_count, 0);
- size =
- msize + sizeof (struct NotifyTrafficMessage) +
- (sender->ats_count) * sizeof (struct GNUNET_TRANSPORT_ATS_Information);
+ atsi_count = 0;
+ size = msize + sizeof (struct NotifyTrafficMessage);
}
#if DEBUG_CORE
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Core service passes message from `%4s' of type %u to client.\n",
- GNUNET_i2s (&sender->peer),
- (unsigned int)
- ntohs (((const struct GNUNET_MessageHeader *) m)->type));
+ GNUNET_i2s (sender), (unsigned int) ntohs (msg->type));
#endif
+ GSC_SESSIONS_add_to_typemap (sender, ntohs (msg->type));
ntm = (struct NotifyTrafficMessage *) buf;
ntm->header.size = htons (size);
ntm->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_NOTIFY_INBOUND);
- ntm->ats_count = htonl (sender->ats_count);
- ntm->peer = sender->peer;
- ats = &ntm->ats;
- memcpy (ats, sender->ats,
- sizeof (struct GNUNET_TRANSPORT_ATS_Information) * sender->ats_count);
- ats[sender->ats_count].type = htonl (GNUNET_TRANSPORT_ATS_ARRAY_TERMINATOR);
- ats[sender->ats_count].value = htonl (0);
- memcpy (&ats[sender->ats_count + 1], m, msize);
- send_to_client (client, &ntm->header, GNUNET_YES);
-}
-
-
-
-
-/**
- * Notify client about a change to existing connection to one of our neighbours.
- *
- * @param neighbour identity of the neighbour that changed status
- * @param tmap updated type map for the neighbour, NULL for disconnect
- */
-void
-GDS_CLIENTS_notify_clients_about_neighbour (const struct GNUNET_PeerIdentity *neighbour,
- const struct GSC_TypeMap *tmap)
-{
+ ntm->ats_count = htonl (atsi_count);
+ ntm->peer = *sender;
+ a = &ntm->ats;
+ memcpy (a, atsi, sizeof (struct GNUNET_ATS_Information) * atsi_count);
+ a[atsi_count].type = htonl (GNUNET_ATS_ARRAY_TERMINATOR);
+ a[atsi_count].value = htonl (0);
+ memcpy (&a[atsi_count + 1], msg, msize);
+ send_to_all_clients (sender, &ntm->header, GNUNET_YES, options,
+ ntohs (msg->type));
}
/**
- * Deliver P2P message to interested clients.
+ * Initialize clients subsystem.
*
- * @param sender peer who sent us the message
- * @param m the message
+ * @param server handle to server clients connect to
*/
-void
-GSC_CLIENTS_deliver_message (const struct GNUNET_PeerIdentity *sender,
- const struct GNUNET_MessageHeader *m)
-{
- struct Neighbour *sender = client;
- size_t msize = ntohs (m->size);
- char buf[256];
- struct Client *cpos;
- uint16_t type;
- unsigned int tpos;
- int deliver_full;
- int dropped;
-
- GNUNET_break (sender->status == PEER_STATE_KEY_CONFIRMED);
- type = ntohs (m->type);
-#if DEBUG_CORE > 1
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Received encapsulated message of type %u and size %u from `%4s'\n",
- (unsigned int) type, ntohs (m->size), GNUNET_i2s (&sender->peer));
-#endif
- GNUNET_snprintf (buf, sizeof (buf),
- gettext_noop ("# bytes of messages of type %u received"),
- (unsigned int) type);
- GNUNET_STATISTICS_update (stats, buf, msize, GNUNET_NO);
- if ((GNUNET_MESSAGE_TYPE_CORE_BINARY_TYPE_MAP == type) ||
- (GNUNET_MESSAGE_TYPE_CORE_COMPRESSED_TYPE_MAP == type))
- {
- /* FIXME: update message type map for 'Neighbour' */
- return;
- }
- dropped = GNUNET_YES;
- cpos = clients;
- while (cpos != NULL)
- {
- deliver_full = GNUNET_NO;
- if (0 != (cpos->options & GNUNET_CORE_OPTION_SEND_FULL_INBOUND))
- deliver_full = GNUNET_YES;
- else
- {
- for (tpos = 0; tpos < cpos->tcnt; tpos++)
- {
- if (type != cpos->types[tpos])
- continue;
- deliver_full = GNUNET_YES;
- break;
- }
- }
- if (GNUNET_YES == deliver_full)
- {
- send_p2p_message_to_client (sender, cpos, m, msize);
- dropped = GNUNET_NO;
- }
- else if (cpos->options & GNUNET_CORE_OPTION_SEND_HDR_INBOUND)
- {
- send_p2p_message_to_client (sender, cpos, m,
- sizeof (struct GNUNET_MessageHeader));
- }
- cpos = cpos->next;
- }
- if (dropped == GNUNET_YES)
- {
-#if DEBUG_CORE
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Message of type %u from `%4s' not delivered to any client.\n",
- (unsigned int) type, GNUNET_i2s (&sender->peer));
-#endif
- GNUNET_STATISTICS_update (stats,
- gettext_noop
- ("# messages not delivered to any client"), 1,
- GNUNET_NO);
- }
-}
-
-
-
void
GSC_CLIENTS_init (struct GNUNET_SERVER_Handle *server)
{
static const struct GNUNET_SERVER_MessageHandler handlers[] = {
{&handle_client_init, NULL,
GNUNET_MESSAGE_TYPE_CORE_INIT, 0},
- {&handle_client_iterate_peers, NULL,
+ {&GSC_SESSIONS_handle_client_iterate_peers, NULL,
GNUNET_MESSAGE_TYPE_CORE_ITERATE_PEERS,
sizeof (struct GNUNET_MessageHeader)},
- {&handle_client_have_peer, NULL,
+ {&GSC_SESSIONS_handle_client_have_peer, NULL,
GNUNET_MESSAGE_TYPE_CORE_PEER_CONNECTED,
sizeof (struct GNUNET_MessageHeader) +
sizeof (struct GNUNET_PeerIdentity)},
- {&handle_client_request_info, NULL,
- GNUNET_MESSAGE_TYPE_CORE_REQUEST_INFO,
- sizeof (struct RequestInfoMessage)},
{&handle_client_send_request, NULL,
GNUNET_MESSAGE_TYPE_CORE_SEND_REQUEST,
sizeof (struct SendMessageRequest)},
};
/* setup notification */
+ client_mst = GNUNET_SERVER_mst_create (&client_tokenizer_callback, NULL);
notifier =
GNUNET_SERVER_notification_context_create (server, MAX_NOTIFY_QUEUE);
GNUNET_SERVER_disconnect_notify (server, &handle_client_disconnect, NULL);
GNUNET_SERVER_add_handlers (server, handlers);
- mst = GNUNET_SERVER_mst_create (&deliver_message, NULL);
}
+/**
+ * Shutdown clients subsystem.
+ */
void
GSC_CLIENTS_done ()
{
- struct Client *c;
+ struct GSC_Client *c;
- while (NULL != (c = clients))
+ while (NULL != (c = client_head))
handle_client_disconnect (NULL, c->client_handle);
- GNUNET_SERVER_notification_context_destroy (notifier);
- notifier = NULL;
- if (mst != NULL)
- {
- GNUNET_SERVER_mst_destroy (mst);
- mst = NULL;
- }
-
+ if (NULL != notifier)
+ {
+ GNUNET_SERVER_notification_context_destroy (notifier);
+ notifier = NULL;
+ }
+ GNUNET_SERVER_mst_destroy (client_mst);
+ client_mst = NULL;
}
+
+/* end of gnunet-service-core_clients.c */
projects / oweals / gnunet.git / blobdiff