-/* code for managing of 'encrypted' sessions (key exchange done) */
+/*
+ This file is part of GNUnet.
+ (C) 2009, 2010, 2011 Christian Grothoff (and other contributing authors)
+
+ 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.
+*/
+/**
+ * @file core/gnunet-service-core_sessions.c
+ * @brief code for managing of 'encrypted' sessions (key exchange done)
+ * @author Christian Grothoff
+ */
+#include "platform.h"
+#include "gnunet-service-core.h"
+#include "gnunet-service-core_neighbours.h"
+#include "gnunet-service-core_kx.h"
+#include "gnunet-service-core_typemap.h"
+#include "gnunet-service-core_sessions.h"
+#include "gnunet-service-core_clients.h"
+#include "gnunet_constants.h"
/**
- * Record kept for each request for transmission issued by a
- * client that is still pending.
+ * How often do we transmit our typemap?
*/
-struct ClientActiveRequest;
+#define TYPEMAP_FREQUENCY GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 5)
+
/**
- * Data kept per session.
+ * Message ready for encryption. This struct is followed by the
+ * actual content of the message.
*/
-struct Session
+struct SessionMessageEntry
{
- /**
- * Identity of the other peer.
- */
- struct GNUNET_PeerIdentity peer;
/**
- * Head of list of requests from clients for transmission to
- * this peer.
- */
- struct ClientActiveRequest *active_client_request_head;
-
- /**
- * Tail of list of requests from clients for transmission to
- * this peer.
+ * We keep messages in a doubly linked list.
*/
- struct ClientActiveRequest *active_client_request_tail;
+ struct SessionMessageEntry *next;
/**
- * Performance data for the peer.
+ * We keep messages in a doubly linked list.
*/
- struct GNUNET_TRANSPORT_ATS_Information *ats;
+ struct SessionMessageEntry *prev;
/**
- * Information about the key exchange with the other peer.
+ * Deadline for transmission, 1s after we received it (if we
+ * are not corking), otherwise "now". Note that this message
+ * does NOT expire past its deadline.
*/
- struct GSC_KeyExchangeInfo *kxinfo;
+ struct GNUNET_TIME_Absolute deadline;
/**
- * ID of task used for sending keep-alive pings.
+ * How long is the message? (number of bytes following the "struct
+ * MessageEntry", but not including the size of "struct
+ * MessageEntry" itself!)
*/
- GNUNET_SCHEDULER_TaskIdentifier keep_alive_task;
+ size_t size;
- /**
- * ID of task used for cleaning up dead neighbour entries.
- */
- GNUNET_SCHEDULER_TaskIdentifier dead_clean_task;
+};
- /**
- * ID of task used for updating bandwidth quota for this neighbour.
- */
- GNUNET_SCHEDULER_TaskIdentifier quota_update_task;
+/**
+ * Data kept per session.
+ */
+struct Session
+{
/**
- * At what time did we initially establish (as in, complete session
- * key handshake) this connection? Should be zero if status != KEY_CONFIRMED.
+ * Identity of the other peer.
*/
- struct GNUNET_TIME_Absolute time_established;
+ struct GNUNET_PeerIdentity peer;
/**
- * At what time did we last receive an encrypted message from the
- * other peer? Should be zero if status != KEY_CONFIRMED.
+ * Head of list of requests from clients for transmission to
+ * this peer.
*/
- struct GNUNET_TIME_Absolute last_activity;
+ struct GSC_ClientActiveRequest *active_client_request_head;
/**
- * How valueable were the messages of this peer recently?
+ * Tail of list of requests from clients for transmission to
+ * this peer.
*/
- unsigned long long current_preference;
+ struct GSC_ClientActiveRequest *active_client_request_tail;
/**
- * Number of entries in 'ats'.
+ * Head of list of messages ready for encryption.
*/
- unsigned int ats_count;
+ struct SessionMessageEntry *sme_head;
/**
- * Bit map indicating which of the 32 sequence numbers before the last
- * were received (good for accepting out-of-order packets and
- * estimating reliability of the connection)
+ * Tail of list of messages ready for encryption.
*/
- unsigned int last_packets_bitmap;
+ struct SessionMessageEntry *sme_tail;
/**
- * last sequence number received on this connection (highest)
+ * Information about the key exchange with the other peer.
*/
- uint32_t last_sequence_number_received;
+ struct GSC_KeyExchangeInfo *kxinfo;
/**
- * last sequence number transmitted
+ * Current type map for this peer.
*/
- uint32_t last_sequence_number_sent;
+ struct GSC_TypeMap *tmap;
/**
- * Available bandwidth in for this peer (current target).
+ * At what time did we initially establish this session?
+ * (currently unused, should be integrated with ATS in the
+ * future...).
*/
- struct GNUNET_BANDWIDTH_Value32NBO bw_in;
+ struct GNUNET_TIME_Absolute time_established;
/**
- * Available bandwidth out for this peer (current target).
+ * Task to transmit corked messages with a delay.
*/
- struct GNUNET_BANDWIDTH_Value32NBO bw_out;
+ GNUNET_SCHEDULER_TaskIdentifier cork_task;
/**
- * Internal bandwidth limit set for this peer (initially typically
- * set to "-1"). Actual "bw_out" is MIN of
- * "bpm_out_internal_limit" and "bw_out_external_limit".
+ * Task to transmit our type map.
*/
- struct GNUNET_BANDWIDTH_Value32NBO bw_out_internal_limit;
+ GNUNET_SCHEDULER_TaskIdentifier typemap_task;
/**
- * External bandwidth limit set for this peer by the
- * peer that we are communicating with. "bw_out" is MIN of
- * "bw_out_internal_limit" and "bw_out_external_limit".
+ * Is the neighbour queue empty and thus ready for us
+ * to transmit an encrypted message?
*/
- struct GNUNET_BANDWIDTH_Value32NBO bw_out_external_limit;
+ int ready_to_transmit;
};
/**
- * Session entry for "this" peer.
+ * Find the session for the given peer.
+ *
+ * @param peer identity of the peer
+ * @return NULL if we are not connected, otherwise the
+ * session handle
*/
-static struct Session self;
+static struct Session *
+find_session (const struct GNUNET_PeerIdentity *peer)
+{
+ return GNUNET_CONTAINER_multihashmap_get (sessions, &peer->hashPubKey);
+}
+
/**
- * Sum of all preferences among all neighbours.
+ * End the session with the given peer (we are no longer
+ * connected).
+ *
+ * @param pid identity of peer to kill session with
*/
-static unsigned long long preference_sum;
+void
+GSC_SESSIONS_end (const struct GNUNET_PeerIdentity *pid)
+{
+ struct Session *session;
+ struct GSC_ClientActiveRequest *car;
+ struct SessionMessageEntry *sme;
+ session = find_session (pid);
+ if (NULL == session)
+ return;
+#if DEBUG_CORE
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Destroying session for peer `%4s'\n",
+ GNUNET_i2s (&session->peer));
+#endif
+ if (GNUNET_SCHEDULER_NO_TASK != session->cork_task)
+ {
+ GNUNET_SCHEDULER_cancel (session->cork_task);
+ session->cork_task = GNUNET_SCHEDULER_NO_TASK;
+ }
+ while (NULL != (car = session->active_client_request_head))
+ {
+ GNUNET_CONTAINER_DLL_remove (session->active_client_request_head,
+ session->active_client_request_tail, car);
+ GSC_CLIENTS_reject_request (car);
+ }
+ while (NULL != (sme = session->sme_head))
+ {
+ GNUNET_CONTAINER_DLL_remove (session->sme_head, session->sme_tail, sme);
+ GNUNET_free (sme);
+ }
+ GNUNET_SCHEDULER_cancel (session->typemap_task);
+ GSC_CLIENTS_notify_clients_about_neighbour (&session->peer, NULL,
+ 0 /* FIXME: ATSI */ ,
+ session->tmap, NULL);
+ GNUNET_assert (GNUNET_YES ==
+ GNUNET_CONTAINER_multihashmap_remove (sessions,
+ &session->
+ peer.hashPubKey,
+ session));
+ GNUNET_STATISTICS_set (GSC_stats, gettext_noop ("# entries in session map"),
+ GNUNET_CONTAINER_multihashmap_size (sessions),
+ GNUNET_NO);
+ GSC_TYPEMAP_destroy (session->tmap);
+ session->tmap = NULL;
+ GNUNET_free (session);
+}
-// FIXME.........
/**
- * At what time should the connection to the given neighbour
- * time out (given no further activity?)
+ * Transmit our current typemap message to the other peer.
+ * (Done periodically in case an update got lost).
*
- * @param n neighbour in question
- * @return absolute timeout
+ * @param cls the 'struct Session*'
+ * @param tc unused
*/
-static struct GNUNET_TIME_Absolute
-get_neighbour_timeout (struct Neighbour *n)
+static void
+transmit_typemap_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
{
- return GNUNET_TIME_absolute_add (n->last_activity,
- GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT);
+ struct Session *session = cls;
+ struct GNUNET_MessageHeader *hdr;
+ struct GNUNET_TIME_Relative delay;
+
+ delay = TYPEMAP_FREQUENCY;
+ /* randomize a bit to avoid spont. sync */
+ delay.rel_value +=
+ GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, 1000);
+ session->typemap_task =
+ GNUNET_SCHEDULER_add_delayed (delay, &transmit_typemap_task, session);
+ GNUNET_STATISTICS_update (GSC_stats,
+ gettext_noop ("# type map refreshes sent"), 1,
+ GNUNET_NO);
+ hdr = GSC_TYPEMAP_compute_type_map_message ();
+ GSC_KX_encrypt_and_transmit (session->kxinfo, hdr, ntohs (hdr->size));
+ GNUNET_free (hdr);
}
/**
- * Helper function for update_preference_sum.
+ * Create a session, a key exchange was just completed.
+ *
+ * @param peer peer that is now connected
+ * @param kx key exchange that completed
*/
-static int
-update_preference (void *cls, const GNUNET_HashCode * key, void *value)
+void
+GSC_SESSIONS_create (const struct GNUNET_PeerIdentity *peer,
+ struct GSC_KeyExchangeInfo *kx)
{
- unsigned long long *ps = cls;
- struct Neighbour *n = value;
+ struct Session *session;
- n->current_preference /= 2;
- *ps += n->current_preference;
- return GNUNET_OK;
+#if DEBUG_CORE
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Creating session for peer `%4s'\n",
+ GNUNET_i2s (peer));
+#endif
+ session = GNUNET_malloc (sizeof (struct Session));
+ session->tmap = GSC_TYPEMAP_create ();
+ session->peer = *peer;
+ session->kxinfo = kx;
+ session->time_established = GNUNET_TIME_absolute_get ();
+ session->typemap_task =
+ GNUNET_SCHEDULER_add_now (&transmit_typemap_task, session);
+ GNUNET_assert (GNUNET_OK ==
+ GNUNET_CONTAINER_multihashmap_put (sessions, &peer->hashPubKey,
+ session,
+ GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
+ GNUNET_STATISTICS_set (GSC_stats, gettext_noop ("# entries in session map"),
+ GNUNET_CONTAINER_multihashmap_size (sessions),
+ GNUNET_NO);
+ GSC_CLIENTS_notify_clients_about_neighbour (peer, NULL, 0 /* FIXME: ATSI */ ,
+ NULL, session->tmap);
}
/**
- * A preference value for a neighbour was update. Update
- * the preference sum accordingly.
+ * Notify the given client about the session (client is new).
*
- * @param inc how much was a preference value increased?
+ * @param cls the 'struct GSC_Client'
+ * @param key peer identity
+ * @param value the 'struct Session'
+ * @return GNUNET_OK (continue to iterate)
*/
-static void
-update_preference_sum (unsigned long long inc)
+static int
+notify_client_about_session (void *cls, const GNUNET_HashCode * key,
+ void *value)
{
- unsigned long long os;
-
- os = preference_sum;
- preference_sum += inc;
- if (preference_sum >= os)
- return; /* done! */
- /* overflow! compensate by cutting all values in half! */
- preference_sum = 0;
- GNUNET_CONTAINER_multihashmap_iterate (neighbours, &update_preference,
- &preference_sum);
- GNUNET_STATISTICS_set (stats, gettext_noop ("# total peer preference"),
- preference_sum, GNUNET_NO);
+ struct GSC_Client *client = cls;
+ struct Session *session = value;
+
+ GSC_CLIENTS_notify_client_about_neighbour (client, &session->peer, NULL, 0, /* FIXME: ATS!? */
+ NULL, /* old TMAP: none */
+ session->tmap);
+ return GNUNET_OK;
}
/**
- * Find the entry for the given neighbour.
+ * We have a new client, notify it about all current sessions.
*
- * @param peer identity of the neighbour
- * @return NULL if we are not connected, otherwise the
- * neighbour's entry.
+ * @param client the new client
*/
-static struct Neighbour *
-find_neighbour (const struct GNUNET_PeerIdentity *peer)
+void
+GSC_SESSIONS_notify_client_about_sessions (struct GSC_Client *client)
{
- return GNUNET_CONTAINER_multihashmap_get (neighbours, &peer->hashPubKey);
+ /* notify new client about existing sessions */
+ GNUNET_CONTAINER_multihashmap_iterate (sessions, ¬ify_client_about_session,
+ client);
}
/**
- * Function called by transport telling us that a peer
- * changed status.
+ * Try to perform a transmission on the given session. Will solicit
+ * additional messages if the 'sme' queue is not full enough.
*
- * @param n the peer that changed status
+ * @param session session to transmit messages from
*/
static void
-handle_peer_status_change (struct Neighbour *n)
+try_transmission (struct Session *session);
+
+
+/**
+ * Queue a request from a client for transmission to a particular peer.
+ *
+ * @param car request to queue; this handle is then shared between
+ * the caller (CLIENTS subsystem) and SESSIONS and must not
+ * be released by either until either 'GNUNET_SESSIONS_dequeue',
+ * 'GNUNET_SESSIONS_transmit' or 'GNUNET_CLIENTS_failed'
+ * have been invoked on it
+ */
+void
+GSC_SESSIONS_queue_request (struct GSC_ClientActiveRequest *car)
{
- struct PeerStatusNotifyMessage *psnm;
- char buf[GNUNET_SERVER_MAX_MESSAGE_SIZE - 1];
- struct GNUNET_TRANSPORT_ATS_Information *ats;
- size_t size;
+ struct Session *session;
- if ((!n->is_connected) || (n->status != PEER_STATE_KEY_CONFIRMED))
- return;
-#if DEBUG_CORE > 1
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Peer `%4s' changed status\n",
- GNUNET_i2s (&n->peer));
+ session = find_session (&car->target);
+ if (session == NULL)
+ {
+#if DEBUG_CORE
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Dropped client request for transmission (am disconnected)\n");
#endif
- size =
- sizeof (struct PeerStatusNotifyMessage) +
- n->ats_count * sizeof (struct GNUNET_TRANSPORT_ATS_Information);
- if (size >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
+ GNUNET_break (0); /* should have been rejected earlier */
+ GSC_CLIENTS_reject_request (car);
+ return;
+ }
+ if (car->msize > GNUNET_CONSTANTS_MAX_ENCRYPTED_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);
+ GSC_CLIENTS_reject_request (car);
+ return;
}
- psnm = (struct PeerStatusNotifyMessage *) buf;
- psnm->header.size = htons (size);
- psnm->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_NOTIFY_STATUS_CHANGE);
- psnm->timeout = GNUNET_TIME_absolute_hton (get_neighbour_timeout (n));
- psnm->bandwidth_in = n->bw_in;
- psnm->bandwidth_out = n->bw_out;
- psnm->peer = n->peer;
- psnm->ats_count = htonl (n->ats_count);
- ats = &psnm->ats;
- memcpy (ats, n->ats,
- n->ats_count * sizeof (struct GNUNET_TRANSPORT_ATS_Information));
- ats[n->ats_count].type = htonl (0);
- ats[n->ats_count].value = htonl (0);
- send_to_all_clients (&psnm->header, GNUNET_YES,
- GNUNET_CORE_OPTION_SEND_STATUS_CHANGE);
- GNUNET_STATISTICS_update (stats, gettext_noop ("# peer status changes"), 1,
- GNUNET_NO);
+#if DEBUG_CORE
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Received client transmission request. queueing\n");
+#endif
+ GNUNET_CONTAINER_DLL_insert (session->active_client_request_head,
+ session->active_client_request_tail, car);
+ try_transmission (session);
}
+/**
+ * Dequeue a request from a client from transmission to a particular peer.
+ *
+ * @param car request to dequeue; this handle will then be 'owned' by
+ * the caller (CLIENTS sysbsystem)
+ */
+void
+GSC_SESSIONS_dequeue_request (struct GSC_ClientActiveRequest *car)
+{
+ struct Session *s;
+
+ if (0 ==
+ memcmp (&car->target, &GSC_my_identity,
+ sizeof (struct GNUNET_PeerIdentity)))
+ return;
+ s = find_session (&car->target);
+ GNUNET_assert (NULL != s);
+ GNUNET_CONTAINER_DLL_remove (s->active_client_request_head,
+ s->active_client_request_tail, car);
+}
+
/**
- * Go over our message queue and if it is not too long, go
- * over the pending requests from clients for this
- * neighbour and send some clients a 'READY' notification.
+ * Discard all expired active transmission requests from clients.
*
- * @param n which peer to process
+ * @param session session to clean up
*/
static void
-schedule_peer_messages (struct Neighbour *n)
+discard_expired_requests (struct Session *session)
{
- struct SendMessageReady smr;
- struct ClientActiveRequest *car;
- struct ClientActiveRequest *pos;
- struct Client *c;
- struct MessageEntry *mqe;
- unsigned int queue_size;
-
- /* check if neighbour queue is empty enough! */
- if (n != &self)
+ struct GSC_ClientActiveRequest *pos;
+ struct GSC_ClientActiveRequest *nxt;
+ struct GNUNET_TIME_Absolute now;
+
+ now = GNUNET_TIME_absolute_get ();
+ pos = NULL;
+ nxt = session->active_client_request_head;
+ while (NULL != nxt)
{
- queue_size = 0;
- mqe = n->messages;
- while (mqe != NULL)
- {
- queue_size++;
- mqe = mqe->next;
- }
- if (queue_size >= MAX_PEER_QUEUE_SIZE)
- {
-#if DEBUG_CORE_CLIENT
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Not considering client transmission requests: queue full\n");
-#endif
- return; /* queue still full */
- }
- /* find highest priority request */
- pos = n->active_client_request_head;
- car = NULL;
- while (pos != NULL)
+ pos = nxt;
+ nxt = pos->next;
+ if ((pos->deadline.abs_value < now.abs_value) &&
+ (GNUNET_YES != pos->was_solicited))
{
- if ((car == NULL) || (pos->priority > car->priority))
- car = pos;
- pos = pos->next;
+ GNUNET_STATISTICS_update (GSC_stats,
+ gettext_noop
+ ("# messages discarded (expired prior to transmission)"),
+ 1, GNUNET_NO);
+ GNUNET_CONTAINER_DLL_remove (session->active_client_request_head,
+ session->active_client_request_tail, pos);
+ GSC_CLIENTS_reject_request (pos);
}
}
- else
+}
+
+
+/**
+ * Solicit messages for transmission.
+ *
+ * @param session session to solict messages for
+ */
+static void
+solicit_messages (struct Session *session)
+{
+ struct GSC_ClientActiveRequest *car;
+ struct GSC_ClientActiveRequest *nxt;
+ size_t so_size;
+
+ discard_expired_requests (session);
+ so_size = 0;
+ nxt = session->active_client_request_head;
+ while (NULL != (car = nxt))
{
- car = n->active_client_request_head;
+ nxt = car->next;
+ if (so_size + car->msize > GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE)
+ break;
+ so_size += car->msize;
+ if (car->was_solicited == GNUNET_YES)
+ continue;
+ car->was_solicited = GNUNET_YES;
+ GSC_CLIENTS_solicit_request (car);
}
- if (car == NULL)
- return; /* no pending requests */
-#if DEBUG_CORE_CLIENT
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Permitting client transmission request to `%s'\n",
- GNUNET_i2s (&n->peer));
-#endif
- c = car->client;
- GNUNET_CONTAINER_DLL_remove (n->active_client_request_head,
- n->active_client_request_tail, car);
- GNUNET_assert (GNUNET_YES ==
- GNUNET_CONTAINER_multihashmap_remove (c->requests,
- &n->peer.hashPubKey,
- car));
- 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 = n->peer;
- send_to_client (c, &smr.header, GNUNET_NO);
- GNUNET_free (car);
}
+/**
+ * Some messages were delayed (corked), but the timeout has now expired.
+ * Send them now.
+ *
+ * @param cls 'struct Session' with the messages to transmit now
+ * @param tc scheduler context (unused)
+ */
+static void
+pop_cork_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+ struct Session *session = cls;
+
+ session->cork_task = GNUNET_SCHEDULER_NO_TASK;
+ try_transmission (session);
+}
+
/**
- * Free the given entry for the neighbour (it has
- * already been removed from the list at this point).
+ * Try to perform a transmission on the given session. Will solicit
+ * additional messages if the 'sme' queue is not full enough.
*
- * @param n neighbour to free
+ * @param session session to transmit messages from
*/
static void
-free_neighbour (struct Neighbour *n)
+try_transmission (struct Session *session)
{
- struct MessageEntry *m;
- struct ClientActiveRequest *car;
+ struct SessionMessageEntry *pos;
+ size_t msize;
+ struct GNUNET_TIME_Absolute now;
+ struct GNUNET_TIME_Absolute min_deadline;
-#if DEBUG_CORE
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Destroying neighbour entry for peer `%4s'\n",
- GNUNET_i2s (&n->peer));
-#endif
- if (n->skm != NULL)
- {
- GNUNET_free (n->skm);
- n->skm = NULL;
- }
- while (NULL != (m = n->messages))
- {
- n->messages = m->next;
- GNUNET_free (m);
- }
- while (NULL != (m = n->encrypted_head))
+ if (GNUNET_YES != session->ready_to_transmit)
+ return;
+ msize = 0;
+ min_deadline = GNUNET_TIME_UNIT_FOREVER_ABS;
+ /* check 'ready' messages */
+ pos = session->sme_head;
+ while ((NULL != pos) &&
+ (msize + pos->size <= GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE))
{
- GNUNET_CONTAINER_DLL_remove (n->encrypted_head, n->encrypted_tail, m);
- GNUNET_free (m);
+ GNUNET_assert (pos->size < GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE);
+ msize += pos->size;
+ min_deadline = GNUNET_TIME_absolute_min (min_deadline, pos->deadline);
+ pos = pos->next;
}
- while (NULL != (car = n->active_client_request_head))
+ now = GNUNET_TIME_absolute_get ();
+ if ((msize == 0) ||
+ ((msize < GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE / 2) &&
+ (min_deadline.abs_value > now.abs_value)))
{
- GNUNET_CONTAINER_DLL_remove (n->active_client_request_head,
- n->active_client_request_tail, car);
- GNUNET_assert (GNUNET_YES ==
- GNUNET_CONTAINER_multihashmap_remove (car->client->requests,
- &n->peer.hashPubKey,
- car));
- GNUNET_free (car);
+ /* not enough ready yet, try to solicit more */
+ solicit_messages (session);
+ if (msize > 0)
+ {
+ /* if there is data to send, just not yet, make sure we do transmit
+ * it once the deadline is reached */
+ if (session->cork_task != GNUNET_SCHEDULER_NO_TASK)
+ GNUNET_SCHEDULER_cancel (session->cork_task);
+ session->cork_task =
+ GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_absolute_get_remaining
+ (min_deadline), &pop_cork_task,
+ session);
+ }
+ return;
}
- if (NULL != n->th)
+ /* create plaintext buffer of all messages, encrypt and transmit */
{
- GNUNET_TRANSPORT_notify_transmit_ready_cancel (n->th);
- n->th = NULL;
+ static unsigned long long total_bytes;
+ static unsigned int total_msgs;
+ char pbuf[msize]; /* plaintext */
+ size_t used;
+
+ used = 0;
+ while ((NULL != (pos = session->sme_head)) && (used + pos->size <= msize))
+ {
+ memcpy (&pbuf[used], &pos[1], pos->size);
+ used += pos->size;
+ GNUNET_CONTAINER_DLL_remove (session->sme_head, session->sme_tail, pos);
+ GNUNET_free (pos);
+ }
+ /* compute average payload size */
+ total_bytes += used;
+ total_msgs++;
+ if (0 == total_msgs)
+ {
+ /* 2^32 messages, wrap around... */
+ total_msgs = 1;
+ total_bytes = used;
+ }
+ GNUNET_STATISTICS_set (GSC_stats, "# avg payload per encrypted message",
+ total_bytes / total_msgs, GNUNET_NO);
+ /* now actually transmit... */
+ session->ready_to_transmit = GNUNET_NO;
+ GSC_KX_encrypt_and_transmit (session->kxinfo, pbuf, used);
}
- if (n->retry_plaintext_task != GNUNET_SCHEDULER_NO_TASK)
- GNUNET_SCHEDULER_cancel (n->retry_plaintext_task);
- if (n->quota_update_task != GNUNET_SCHEDULER_NO_TASK)
- GNUNET_SCHEDULER_cancel (n->quota_update_task);
- if (n->dead_clean_task != GNUNET_SCHEDULER_NO_TASK)
- GNUNET_SCHEDULER_cancel (n->dead_clean_task);
- if (n->keep_alive_task != GNUNET_SCHEDULER_NO_TASK)
- GNUNET_SCHEDULER_cancel (n->keep_alive_task);
- if (n->status == PEER_STATE_KEY_CONFIRMED)
- GNUNET_STATISTICS_update (stats, gettext_noop ("# established sessions"),
- -1, GNUNET_NO);
- GNUNET_array_grow (n->ats, n->ats_count, 0);
- GNUNET_free_non_null (n->pending_ping);
- GNUNET_free_non_null (n->pending_pong);
- GNUNET_free (n);
}
-
/**
- * Task triggered when a neighbour entry is about to time out
- * (and we should prevent this by sending a PING).
+ * Send a message to the neighbour now.
*
- * @param cls the 'struct Neighbour'
- * @param tc scheduler context (not used)
+ * @param cls the message
+ * @param key neighbour's identity
+ * @param value 'struct Neighbour' of the target
+ * @return always GNUNET_OK
*/
-static void
-send_keep_alive (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+static int
+do_send_message (void *cls, const GNUNET_HashCode * key, void *value)
{
- struct Neighbour *n = cls;
- struct GNUNET_TIME_Relative retry;
- struct GNUNET_TIME_Relative left;
- struct MessageEntry *me;
- struct PingMessage pp;
- struct PingMessage *pm;
- struct GNUNET_CRYPTO_AesInitializationVector iv;
-
- n->keep_alive_task = GNUNET_SCHEDULER_NO_TASK;
- /* send PING */
- me = GNUNET_malloc (sizeof (struct MessageEntry) +
- sizeof (struct PingMessage));
- me->deadline = GNUNET_TIME_relative_to_absolute (MAX_PING_DELAY);
- me->priority = PING_PRIORITY;
- me->size = sizeof (struct PingMessage);
- GNUNET_CONTAINER_DLL_insert_after (n->encrypted_head, n->encrypted_tail,
- n->encrypted_tail, me);
- pm = (struct PingMessage *) &me[1];
- pm->header.size = htons (sizeof (struct PingMessage));
- pm->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_PING);
- pm->iv_seed =
- GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_NONCE, UINT32_MAX);
- derive_iv (&iv, &n->encrypt_key, pm->iv_seed, &n->peer);
- pp.challenge = n->ping_challenge;
- pp.target = n->peer;
-#if DEBUG_HANDSHAKE
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Encrypting `%s' message with challenge %u for `%4s' using key %u, IV %u (salt %u).\n",
- "PING", (unsigned int) n->ping_challenge, GNUNET_i2s (&n->peer),
- (unsigned int) n->encrypt_key.crc32, GNUNET_CRYPTO_crc32_n (&iv,
- sizeof
- (iv)),
- pm->iv_seed);
-#endif
- do_encrypt (n, &iv, &pp.target, &pm->target,
- sizeof (struct PingMessage) - ((void *) &pm->target -
- (void *) pm));
- process_encrypted_neighbour_queue (n);
- /* reschedule PING job */
- left = GNUNET_TIME_absolute_get_remaining (get_neighbour_timeout (n));
- retry =
- GNUNET_TIME_relative_max (GNUNET_TIME_relative_divide (left, 2),
- MIN_PING_FREQUENCY);
- n->keep_alive_task =
- GNUNET_SCHEDULER_add_delayed (retry, &send_keep_alive, n);
-
+ const struct GNUNET_MessageHeader *hdr = cls;
+ struct Session *session = value;
+ struct SessionMessageEntry *m;
+ uint16_t size;
+
+ size = ntohs (hdr->size);
+ m = GNUNET_malloc (sizeof (struct SessionMessageEntry) + size);
+ memcpy (&m[1], hdr, size);
+ m->size = size;
+ GNUNET_CONTAINER_DLL_insert (session->sme_head, session->sme_tail, m);
+ try_transmission (session);
+ return GNUNET_OK;
}
+
/**
- * Consider freeing the given neighbour since we may not need
- * to keep it around anymore.
+ * Broadcast a message to all neighbours.
*
- * @param n neighbour to consider discarding
+ * @param msg message to transmit
*/
-static void
-consider_free_neighbour (struct Neighbour *n);
+void
+GSC_SESSIONS_broadcast (const struct GNUNET_MessageHeader *msg)
+{
+ if (NULL == sessions)
+ return;
+ GNUNET_CONTAINER_multihashmap_iterate (sessions, &do_send_message,
+ (void *) msg);
+}
/**
- * Task triggered when a neighbour entry might have gotten stale.
+ * Traffic is being solicited for the given peer. This means that the
+ * message queue on the transport-level (NEIGHBOURS subsystem) is now
+ * empty and it is now OK to transmit another (non-control) message.
*
- * @param cls the 'struct Neighbour'
- * @param tc scheduler context (not used)
+ * @param pid identity of peer ready to receive data
*/
-static void
-consider_free_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+void
+GSC_SESSIONS_solicit (const struct GNUNET_PeerIdentity *pid)
{
- struct Neighbour *n = cls;
+ struct Session *session;
- n->dead_clean_task = GNUNET_SCHEDULER_NO_TASK;
- consider_free_neighbour (n);
+ session = find_session (pid);
+ if (NULL == session)
+ return;
+ session->ready_to_transmit = GNUNET_YES;
+ try_transmission (session);
}
/**
- * Consider freeing the given neighbour since we may not need
- * to keep it around anymore.
+ * Transmit a message to a particular peer.
*
- * @param n neighbour to consider discarding
+ * @param car original request that was queued and then solicited;
+ * this handle will now be 'owned' by the SESSIONS subsystem
+ * @param msg message to transmit
+ * @param cork is corking allowed?
*/
-static void
-consider_free_neighbour (struct Neighbour *n)
+void
+GSC_SESSIONS_transmit (struct GSC_ClientActiveRequest *car,
+ const struct GNUNET_MessageHeader *msg, int cork)
{
- struct GNUNET_TIME_Relative left;
-
- if ((n->th != NULL) || (n->pitr != NULL) || (GNUNET_YES == n->is_connected))
- return; /* no chance */
+ struct Session *session;
+ struct SessionMessageEntry *sme;
+ size_t msize;
- left = GNUNET_TIME_absolute_get_remaining (get_neighbour_timeout (n));
- if (left.rel_value > 0)
- {
- if (n->dead_clean_task != GNUNET_SCHEDULER_NO_TASK)
- GNUNET_SCHEDULER_cancel (n->dead_clean_task);
- n->dead_clean_task =
- GNUNET_SCHEDULER_add_delayed (left, &consider_free_task, n);
+ session = find_session (&car->target);
+ if (NULL == session)
return;
- }
- /* actually free the neighbour... */
- GNUNET_assert (GNUNET_YES ==
- GNUNET_CONTAINER_multihashmap_remove (neighbours,
- &n->peer.hashPubKey, n));
- GNUNET_STATISTICS_set (stats, gettext_noop ("# neighbour entries allocated"),
- GNUNET_CONTAINER_multihashmap_size (neighbours),
- GNUNET_NO);
- free_neighbour (n);
+ msize = ntohs (msg->size);
+ sme = GNUNET_malloc (sizeof (struct SessionMessageEntry) + msize);
+ memcpy (&sme[1], msg, msize);
+ sme->size = msize;
+ if (GNUNET_YES == cork)
+ sme->deadline =
+ GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_MAX_CORK_DELAY);
+ GNUNET_CONTAINER_DLL_insert_tail (session->sme_head, session->sme_tail, sme);
+ try_transmission (session);
}
/**
- * Function called when the transport service is ready to
- * receive an encrypted message for the respective peer
+ * Helper function for GSC_SESSIONS_handle_client_iterate_peers.
*
- * @param cls neighbour to use message from
- * @param size number of bytes we can transmit
- * @param buf where to copy the message
- * @return number of bytes transmitted
+ * @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)
*/
-static size_t
-notify_encrypted_transmit_ready (void *cls, size_t size, void *buf)
+#include "core.h"
+static int
+queue_connect_message (void *cls, const GNUNET_HashCode * key, void *value)
{
- struct Neighbour *n = cls;
- struct MessageEntry *m;
- size_t ret;
- char *cbuf;
-
- n->th = NULL;
- m = n->encrypted_head;
- if (m == NULL)
- {
-#if DEBUG_CORE
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Encrypted message queue empty, no messages added to buffer for `%4s'\n",
- GNUNET_i2s (&n->peer));
-#endif
- return 0;
- }
- GNUNET_CONTAINER_DLL_remove (n->encrypted_head, n->encrypted_tail, m);
- ret = 0;
- cbuf = buf;
- if (buf != NULL)
- {
- GNUNET_assert (size >= m->size);
- memcpy (cbuf, &m[1], m->size);
- ret = m->size;
- GNUNET_BANDWIDTH_tracker_consume (&n->available_send_window, m->size);
-#if DEBUG_CORE
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Copied message of type %u and size %u into transport buffer for `%4s'\n",
- (unsigned int)
- ntohs (((struct GNUNET_MessageHeader *) &m[1])->type),
- (unsigned int) ret, GNUNET_i2s (&n->peer));
-#endif
- process_encrypted_neighbour_queue (n);
- }
- else
- {
-#if DEBUG_CORE
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Transmission of message of type %u and size %u failed\n",
- (unsigned int)
- ntohs (((struct GNUNET_MessageHeader *) &m[1])->type),
- (unsigned int) m->size);
-#endif
- }
- GNUNET_free (m);
- consider_free_neighbour (n);
- GNUNET_STATISTICS_update (stats,
- gettext_noop
- ("# encrypted bytes given to transport"), ret,
- GNUNET_NO);
- return ret;
+ struct GNUNET_SERVER_TransmitContext *tc = cls;
+ struct Session *session = value;
+ struct ConnectNotifyMessage cnm;
+
+ /* FIXME: code duplication with clients... */
+ cnm.header.size = htons (sizeof (struct ConnectNotifyMessage));
+ cnm.header.type = htons (GNUNET_MESSAGE_TYPE_CORE_NOTIFY_CONNECT);
+ // FIXME: full ats...
+ cnm.ats_count = htonl (0);
+ cnm.peer = session->peer;
+ GNUNET_SERVER_transmit_context_append_message (tc, &cnm.header);
+ return GNUNET_OK;
}
/**
- * Check if we have encrypted messages for the specified neighbour
- * pending, and if so, check with the transport about sending them
- * out.
+ * Handle CORE_ITERATE_PEERS request. For this request type, the client
+ * does not have to have transmitted an INIT request. All current peers
+ * are returned, regardless of which message types they accept.
*
- * @param n neighbour to check.
+ * @param cls unused
+ * @param client client sending the iteration request
+ * @param message iteration request message
*/
-static void
-process_encrypted_neighbour_queue (struct Neighbour *n)
+void
+GSC_SESSIONS_handle_client_iterate_peers (void *cls,
+ struct GNUNET_SERVER_Client *client,
+ const struct GNUNET_MessageHeader
+ *message)
+{
+ struct GNUNET_MessageHeader done_msg;
+ struct GNUNET_SERVER_TransmitContext *tc;
+
+ tc = GNUNET_SERVER_transmit_context_create (client);
+ GNUNET_CONTAINER_multihashmap_iterate (sessions, &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);
+}
+
+
+/**
+ * Handle CORE_PEER_CONNECTED request. Notify client about connection
+ * to the given neighbour. For this request type, the client does not
+ * have to have transmitted an INIT request. All current peers are
+ * returned, regardless of which message types they accept.
+ *
+ * @param cls unused
+ * @param client client sending the iteration request
+ * @param message iteration request message
+ */
+void
+GSC_SESSIONS_handle_client_have_peer (void *cls,
+ struct GNUNET_SERVER_Client *client,
+ const struct GNUNET_MessageHeader
+ *message)
{
- struct MessageEntry *m;
+ struct GNUNET_MessageHeader done_msg;
+ struct GNUNET_SERVER_TransmitContext *tc;
+ const struct GNUNET_PeerIdentity *peer;
+
+ peer = (const struct GNUNET_PeerIdentity *) &message[1]; // YUCK!
+ tc = GNUNET_SERVER_transmit_context_create (client);
+ GNUNET_CONTAINER_multihashmap_get_multiple (sessions, &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);
+}
+
- if (n->th != NULL)
- return; /* request already pending */
- if (GNUNET_YES != n->is_connected)
+/**
+ * We've received a typemap message from a peer, update ours.
+ * Notifies clients about the session.
+ *
+ * @param peer peer this is about
+ * @param msg typemap update message
+ */
+void
+GSC_SESSIONS_set_typemap (const struct GNUNET_PeerIdentity *peer,
+ const struct GNUNET_MessageHeader *msg)
+{
+ struct Session *session;
+ struct GSC_TypeMap *nmap;
+
+ nmap = GSC_TYPEMAP_get_from_message (msg);
+ if (NULL == nmap)
+ return; /* malformed */
+ session = find_session (peer);
+ if (NULL == session)
{
GNUNET_break (0);
return;
}
- m = n->encrypted_head;
- if (m == NULL)
- {
- /* encrypted queue empty, try plaintext instead */
- process_plaintext_neighbour_queue (n);
- return;
- }
-#if DEBUG_CORE > 1
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Asking transport for transmission of %u bytes to `%4s' in next %llu ms\n",
- (unsigned int) m->size, GNUNET_i2s (&n->peer),
- (unsigned long long)
- GNUNET_TIME_absolute_get_remaining (m->deadline).rel_value);
-#endif
- n->th =
- GNUNET_TRANSPORT_notify_transmit_ready (transport, &n->peer, m->size,
- m->priority,
- GNUNET_TIME_absolute_get_remaining
- (m->deadline),
- ¬ify_encrypted_transmit_ready,
- n);
- if (n->th == NULL)
- {
- /* message request too large or duplicate request */
- GNUNET_break (0);
- /* discard encrypted message */
- GNUNET_CONTAINER_DLL_remove (n->encrypted_head, n->encrypted_tail, m);
- GNUNET_free (m);
- process_encrypted_neighbour_queue (n);
- }
+ GSC_CLIENTS_notify_clients_about_neighbour (peer, NULL, 0, /* FIXME: ATS */
+ session->tmap, nmap);
+ GSC_TYPEMAP_destroy (session->tmap);
+ session->tmap = nmap;
}
/**
- * Initialize a new 'struct Neighbour'.
+ * The given peer send a message of the specified type. Make sure the
+ * respective bit is set in its type-map and that clients are notified
+ * about the session.
*
- * @param pid ID of the new neighbour
- * @return handle for the new neighbour
+ * @param peer peer this is about
+ * @param type type of the message
*/
-static struct Neighbour *
-create_neighbour (const struct GNUNET_PeerIdentity *pid)
+void
+GSC_SESSIONS_add_to_typemap (const struct GNUNET_PeerIdentity *peer,
+ uint16_t type)
{
- struct Neighbour *n;
- struct GNUNET_TIME_Absolute now;
+ struct Session *session;
+ struct GSC_TypeMap *nmap;
-#if DEBUG_CORE
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Creating neighbour entry for peer `%4s'\n", GNUNET_i2s (pid));
-#endif
- n = GNUNET_malloc (sizeof (struct Neighbour));
- n->peer = *pid;
- GNUNET_CRYPTO_aes_create_session_key (&n->encrypt_key);
- now = GNUNET_TIME_absolute_get ();
- n->encrypt_key_created = now;
- n->last_activity = now;
- n->set_key_retry_frequency = INITIAL_SET_KEY_RETRY_FREQUENCY;
- n->bw_in = GNUNET_CONSTANTS_DEFAULT_BW_IN_OUT;
- n->bw_out = GNUNET_CONSTANTS_DEFAULT_BW_IN_OUT;
- n->bw_out_internal_limit = GNUNET_BANDWIDTH_value_init (UINT32_MAX);
- n->bw_out_external_limit = GNUNET_CONSTANTS_DEFAULT_BW_IN_OUT;
- n->ping_challenge =
- GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_NONCE, UINT32_MAX);
- GNUNET_assert (GNUNET_OK ==
- GNUNET_CONTAINER_multihashmap_put (neighbours,
- &n->peer.hashPubKey, n,
- GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
- GNUNET_STATISTICS_set (stats, gettext_noop ("# neighbour entries allocated"),
- GNUNET_CONTAINER_multihashmap_size (neighbours),
- GNUNET_NO);
- neighbour_quota_update (n, NULL);
- consider_free_neighbour (n);
- return n;
+ if (0 == memcmp (peer, &GSC_my_identity, sizeof (struct GNUNET_PeerIdentity)))
+ return;
+ session = find_session (peer);
+ GNUNET_assert (NULL != session);
+ if (GNUNET_YES == GSC_TYPEMAP_test_match (session->tmap, &type, 1))
+ return; /* already in it */
+ nmap = GSC_TYPEMAP_extend (session->tmap, &type, 1);
+ GSC_CLIENTS_notify_clients_about_neighbour (peer, NULL, 0, /* FIXME: ATS */
+ session->tmap, nmap);
+ GSC_TYPEMAP_destroy (session->tmap);
+ session->tmap = nmap;
}
-int
-GSC_NEIGHBOURS_init ()
+/**
+ * Initialize sessions subsystem.
+ */
+void
+GSC_SESSIONS_init ()
+{
+ sessions = GNUNET_CONTAINER_multihashmap_create (128);
+}
+
+
+/**
+ * Helper function for GSC_SESSIONS_handle_client_iterate_peers.
+ *
+ * @param cls NULL
+ * @param key identity of the connected peer
+ * @param value the 'struct Session' for the peer
+ * @return GNUNET_OK (continue to iterate)
+ */
+static int
+free_session_helper (void *cls, const GNUNET_HashCode * key, void *value)
{
- neighbours = GNUNET_CONTAINER_multihashmap_create (128);
- self.public_key = &my_public_key;
- self.peer = my_identity;
- self.last_activity = GNUNET_TIME_UNIT_FOREVER_ABS;
- self.status = PEER_STATE_KEY_CONFIRMED;
- self.is_connected = GNUNET_YES;
+ struct Session *session = value;
+
+ GSC_SESSIONS_end (&session->peer);
return GNUNET_OK;
}
+/**
+ * Shutdown sessions subsystem.
+ */
void
-GSC_NEIGHBOURS_done ()
+GSC_SESSIONS_done ()
{
- GNUNET_CONTAINER_multihashmap_iterate (neighbours, &free_neighbour_helper,
- NULL);
- GNUNET_CONTAINER_multihashmap_destroy (neighbours);
- neighbours = NULL;
- GNUNET_STATISTICS_set (stats, gettext_noop ("# neighbour entries allocated"),
- 0, GNUNET_NO);
+ GNUNET_CONTAINER_multihashmap_iterate (sessions, &free_session_helper, NULL);
+ GNUNET_CONTAINER_multihashmap_destroy (sessions);
+ sessions = NULL;
}
+
+/* end of gnunet-service-core_sessions.c */
projects / oweals / gnunet.git / blobdiff