/*
This file is part of GNUnet.
- (C) 2009-2014 Christian Grothoff (and other contributing authors)
+ Copyright (C) 2009-2014, 2016 GNUnet e.V.
GNUnet is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published
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.
*/
/**
*/
#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"
#include "core.h"
+/**
+ * How many encrypted messages do we queue at most?
+ * Needed to bound memory consumption.
+ */
+#define MAX_ENCRYPTED_MESSAGE_QUEUE_SIZE 4
+
+
/**
* Message ready for encryption. This struct is followed by the
* actual content of the message.
struct GNUNET_TIME_Absolute deadline;
/**
- * How long is the message? (number of bytes following the "struct
- * MessageEntry", but not including the size of "struct
- * MessageEntry" itself!)
+ * How long is the message? (number of bytes following the `struct
+ * MessageEntry`, but not including the size of `struct
+ * MessageEntry` itself!)
*/
size_t size;
/**
* Identity of the other peer.
*/
- struct GNUNET_PeerIdentity peer;
+ const struct GNUNET_PeerIdentity *peer;
+ /**
+ * Key exchange state for this peer.
+ */
+ struct GSC_KeyExchangeInfo *kx;
+
/**
* Head of list of requests from clients for transmission to
* this peer.
*/
struct SessionMessageEntry *sme_tail;
- /**
- * Information about the key exchange with the other peer.
- */
- struct GSC_KeyExchangeInfo *kxinfo;
-
/**
* Current type map for this peer.
*/
/**
* Task to transmit corked messages with a delay.
*/
- GNUNET_SCHEDULER_TaskIdentifier cork_task;
+ struct GNUNET_SCHEDULER_Task *cork_task;
/**
* Task to transmit our type map.
*/
- GNUNET_SCHEDULER_TaskIdentifier typemap_task;
+ struct GNUNET_SCHEDULER_Task *typemap_task;
/**
* Retransmission delay we currently use for the typemap
static struct Session *
find_session (const struct GNUNET_PeerIdentity *peer)
{
- return GNUNET_CONTAINER_multipeermap_get (sessions, peer);
+ if (NULL == sessions)
+ return NULL;
+ return GNUNET_CONTAINER_multipeermap_get (sessions,
+ peer);
}
if (NULL == session)
return;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Destroying session for peer `%4s'\n",
- GNUNET_i2s (&session->peer));
- if (GNUNET_SCHEDULER_NO_TASK != session->cork_task)
+ "Destroying session for peer `%s'\n",
+ GNUNET_i2s (session->peer));
+ if (NULL != session->cork_task)
{
GNUNET_SCHEDULER_cancel (session->cork_task);
- session->cork_task = GNUNET_SCHEDULER_NO_TASK;
+ session->cork_task = NULL;
}
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);
+ GSC_CLIENTS_reject_request (car,
+ GNUNET_NO);
}
while (NULL != (sme = session->sme_head))
{
sme);
GNUNET_free (sme);
}
- if (GNUNET_SCHEDULER_NO_TASK != session->typemap_task)
+ if (NULL != session->typemap_task)
{
GNUNET_SCHEDULER_cancel (session->typemap_task);
- session->typemap_task = GNUNET_SCHEDULER_NO_TASK;
+ session->typemap_task = NULL;
}
- GSC_CLIENTS_notify_clients_about_neighbour (&session->peer,
- session->tmap, NULL);
+ GSC_CLIENTS_notify_clients_about_neighbour (session->peer,
+ session->tmap,
+ NULL);
GNUNET_assert (GNUNET_YES ==
GNUNET_CONTAINER_multipeermap_remove (sessions,
- &session->peer,
+ session->peer,
session));
- GNUNET_STATISTICS_set (GSC_stats, gettext_noop ("# peers connected"),
+ GNUNET_STATISTICS_set (GSC_stats,
+ gettext_noop ("# peers connected"),
GNUNET_CONTAINER_multipeermap_size (sessions),
GNUNET_NO);
GSC_TYPEMAP_destroy (session->tmap);
* (Done periodically until the typemap is confirmed).
*
* @param cls the `struct Session *`
- * @param tc unused
*/
static void
-transmit_typemap_task (void *cls,
- const struct GNUNET_SCHEDULER_TaskContext *tc)
+transmit_typemap_task (void *cls)
{
struct Session *session = cls;
struct GNUNET_MessageHeader *hdr;
GNUNET_SCHEDULER_add_delayed (delay,
&transmit_typemap_task, session);
GNUNET_STATISTICS_update (GSC_stats,
- gettext_noop ("# type map refreshes sent"), 1,
+ 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));
+ GSC_KX_encrypt_and_transmit (session->kx,
+ hdr,
+ ntohs (hdr->size));
GNUNET_free (hdr);
}
static void
start_typemap_task (struct Session *session)
{
- if (GNUNET_SCHEDULER_NO_TASK != session->typemap_task)
+ if (NULL != session->typemap_task)
GNUNET_SCHEDULER_cancel (session->typemap_task);
session->typemap_delay = GNUNET_TIME_UNIT_SECONDS;
session->typemap_task =
GNUNET_i2s (peer));
session = GNUNET_new (struct Session);
session->tmap = GSC_TYPEMAP_create ();
- session->peer = *peer;
- session->kxinfo = kx;
+ session->peer = peer;
+ session->kx = kx;
GNUNET_assert (GNUNET_OK ==
GNUNET_CONTAINER_multipeermap_put (sessions,
- &session->peer,
+ session->peer,
session,
GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
- GNUNET_STATISTICS_set (GSC_stats, gettext_noop ("# peers connected"),
+ GNUNET_STATISTICS_set (GSC_stats,
+ gettext_noop ("# peers connected"),
GNUNET_CONTAINER_multipeermap_size (sessions),
GNUNET_NO);
GSC_CLIENTS_notify_clients_about_neighbour (peer,
1, GNUNET_NO);
return;
}
- if (GNUNET_SCHEDULER_NO_TASK != session->typemap_task)
+ if (NULL != session->typemap_task)
{
GNUNET_SCHEDULER_cancel (session->typemap_task);
- session->typemap_task = GNUNET_SCHEDULER_NO_TASK;
+ session->typemap_task = NULL;
}
GNUNET_STATISTICS_update (GSC_stats,
gettext_noop
struct Session *session = value;
GSC_CLIENTS_notify_client_about_neighbour (client,
- &session->peer,
+ session->peer,
NULL, /* old TMAP: none */
session->tmap);
return GNUNET_OK;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Dropped client request for transmission (am disconnected)\n");
GNUNET_break (0); /* should have been rejected earlier */
- GSC_CLIENTS_reject_request (car);
+ GSC_CLIENTS_reject_request (car,
+ GNUNET_NO);
return;
}
if (car->msize > GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE)
{
GNUNET_break (0);
- GSC_CLIENTS_reject_request (car);
+ GSC_CLIENTS_reject_request (car,
+ GNUNET_YES);
return;
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Received client transmission request. queueing\n");
GNUNET_CONTAINER_DLL_insert (session->active_client_request_head,
- session->active_client_request_tail, car);
+ session->active_client_request_tail,
+ car);
try_transmission (session);
}
struct Session *session;
if (0 ==
- memcmp (&car->target, &GSC_my_identity,
+ memcmp (&car->target,
+ &GSC_my_identity,
sizeof (struct GNUNET_PeerIdentity)))
return;
session = find_session (&car->target);
GNUNET_CONTAINER_DLL_remove (session->active_client_request_head,
session->active_client_request_tail,
car);
-}
-
-
-/**
- * Discard all expired active transmission requests from clients.
- *
- * @param session session to clean up
- */
-static void
-discard_expired_requests (struct Session *session)
-{
- 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)
- {
- pos = nxt;
- nxt = pos->next;
- if ((pos->deadline.abs_value_us < now.abs_value_us) &&
- (GNUNET_YES != pos->was_solicited))
- {
- 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);
- }
- }
+ /* dequeueing of 'high' priority messages may unblock
+ transmission for lower-priority messages, so we also
+ need to try in this case. */
+ try_transmission (session);
}
size_t so_size;
enum GNUNET_CORE_Priority pmax;
- discard_expired_requests (session);
so_size = msize;
pmax = GNUNET_CORE_PRIO_BACKGROUND;
for (car = session->active_client_request_head; NULL != car; car = car->next)
if (GNUNET_YES == car->was_solicited)
continue;
car->was_solicited = GNUNET_YES;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Soliciting message with priority %u\n",
+ car->priority);
GSC_CLIENTS_solicit_request (car);
+ /* The above call may *dequeue* requests and thereby
+ clobber 'nxt'. Hence we need to restart from the
+ head of the list. */
+ nxt = session->active_client_request_head;
+ so_size = msize;
}
}
* 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)
+pop_cork_task (void *cls)
{
struct Session *session = cls;
- session->cork_task = GNUNET_SCHEDULER_NO_TASK;
+ session->cork_task = NULL;
try_transmission (session);
}
int excess;
if (GNUNET_YES != session->ready_to_transmit)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Not yet ready to transmit, not evaluating queue\n");
return;
+ }
msize = 0;
min_deadline = GNUNET_TIME_UNIT_FOREVER_ABS;
/* if the peer has excess bandwidth, background traffic is allowed,
otherwise not */
- excess = GSC_NEIGHBOURS_check_excess_bandwidth (&session->peer);
+ if (MAX_ENCRYPTED_MESSAGE_QUEUE_SIZE <=
+ GSC_NEIGHBOURS_get_queue_length (session->kx))
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Transmission queue already very long, waiting...\n");
+ return; /* queue already too long */
+ }
+ excess = GSC_NEIGHBOURS_check_excess_bandwidth (session->kx);
if (GNUNET_YES == excess)
maxp = GNUNET_CORE_PRIO_BACKGROUND;
else
GNUNET_assert (pos->size < GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE);
msize += pos->size;
maxp = GNUNET_MAX (maxp, pos->priority);
- min_deadline = GNUNET_TIME_absolute_min (min_deadline, pos->deadline);
+ min_deadline = GNUNET_TIME_absolute_min (min_deadline,
+ pos->deadline);
pos = pos->next;
}
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Calculating transmission set with %u priority (%s) and %s earliest deadline\n",
+ maxp,
+ (GNUNET_YES == excess) ? "excess bandwidth" : "limited bandwidth",
+ GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining (min_deadline),
+ GNUNET_YES));
+
if (maxp < GNUNET_CORE_PRIO_CRITICAL_CONTROL)
{
/* if highest already solicited priority from clients is not critical,
{
if (GNUNET_YES == car->was_solicited)
continue;
- maxpc = GNUNET_MAX (maxpc, car->priority);
+ maxpc = GNUNET_MAX (maxpc,
+ car->priority);
}
if (maxpc > maxp)
{
/* we have messages waiting for solicitation that have a higher
priority than those that we already accepted; solicit the
high-priority messages first */
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Soliciting messages based on priority (%u > %u)\n",
+ maxpc,
+ maxp);
solicit_messages (session, 0);
return;
}
{
/* never solicit more, we have critical messages to process */
excess = GNUNET_NO;
+ maxpc = GNUNET_CORE_PRIO_BACKGROUND;
}
now = GNUNET_TIME_absolute_get ();
if ( ( (GNUNET_YES == excess) ||
/* not enough ready yet (tiny message & cork possible), or no messages at all,
and either excess bandwidth or best-effort or higher message waiting at
client; in this case, we try to solicit more */
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Soliciting messages (excess %d, maxpc %d, message size %u, deadline %s)\n",
+ excess,
+ maxpc,
+ (unsigned int) msize,
+ GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining (min_deadline),
+ GNUNET_YES));
solicit_messages (session,
msize);
if (msize > 0)
{
/* if there is data to send, just not yet, make sure we do transmit
* it once the deadline is reached */
- if (GNUNET_SCHEDULER_NO_TASK != session->cork_task)
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Corking until %s\n",
+ GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining (min_deadline),
+ GNUNET_YES));
+ if (NULL != session->cork_task)
GNUNET_SCHEDULER_cancel (session->cork_task);
session->cork_task =
- GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_absolute_get_remaining
- (min_deadline), &pop_cork_task,
+ GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_absolute_get_remaining (min_deadline),
+ &pop_cork_task,
session);
}
+ else
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Queue empty, waiting for solicitations\n");
+ }
return;
}
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Building combined plaintext buffer to transmit message!\n");
/* create plaintext buffer of all messages (that fit), encrypt and
transmit */
{
while ( (NULL != (pos = session->sme_head)) &&
(used + pos->size <= msize) )
{
- memcpy (&pbuf[used], &pos[1], pos->size);
+ GNUNET_memcpy (&pbuf[used], &pos[1], pos->size);
used += pos->size;
GNUNET_CONTAINER_DLL_remove (session->sme_head,
session->sme_tail,
}
GNUNET_STATISTICS_set (GSC_stats,
"# avg payload per encrypted message",
- total_bytes / total_msgs, GNUNET_NO);
+ total_bytes / total_msgs,
+ GNUNET_NO);
/* now actually transmit... */
session->ready_to_transmit = GNUNET_NO;
- GSC_KX_encrypt_and_transmit (session->kxinfo, pbuf, used);
+ GSC_KX_encrypt_and_transmit (session->kx,
+ pbuf,
+ used);
}
}
size = ntohs (hdr->size);
sme = GNUNET_malloc (sizeof (struct SessionMessageEntry) + size);
- memcpy (&sme[1], hdr, size);
+ GNUNET_memcpy (&sme[1],
+ hdr,
+ size);
sme->size = size;
sme->priority = GNUNET_CORE_PRIO_CRITICAL_CONTROL;
GNUNET_CONTAINER_DLL_insert (session->sme_head,
return;
msize = ntohs (msg->size);
sme = GNUNET_malloc (sizeof (struct SessionMessageEntry) + msize);
- memcpy (&sme[1], msg, msize);
+ GNUNET_memcpy (&sme[1],
+ msg,
+ msize);
sme->size = msize;
sme->priority = priority;
if (GNUNET_YES == cork)
+ {
sme->deadline =
GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_MAX_CORK_DELAY);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Mesage corked, delaying transmission\n");
+ }
pos = session->sme_head;
while ( (NULL != pos) &&
- (pos->priority > sme->priority) )
+ (pos->priority >= sme->priority) )
pos = pos->next;
if (NULL == pos)
GNUNET_CONTAINER_DLL_insert_tail (session->sme_head,
const struct GNUNET_PeerIdentity *key,
void *value)
{
- struct Session *session = value;
+ /* struct Session *session = value; */
- GSC_SESSIONS_end (&session->peer);
+ GSC_SESSIONS_end (key);
return GNUNET_OK;
}
if (NULL != sessions)
{
GNUNET_CONTAINER_multipeermap_iterate (sessions,
- &free_session_helper, NULL);
+ &free_session_helper,
+ NULL);
GNUNET_CONTAINER_multipeermap_destroy (sessions);
sessions = NULL;
}
projects / oweals / gnunet.git / blobdiff