#include "gnunet_peerinfo_service.h"
#include "gnunet_protocols.h"
#include "gnunet_signatures.h"
+#include "gnunet_statistics_service.h"
#include "gnunet_transport_service.h"
#include "core.h"
#define DEBUG_HANDSHAKE GNUNET_NO
+#define DEBUG_CORE_QUOTA GNUNET_NO
+
/**
* Receive and send buffer windows grow over time. For
* how long can 'unused' bandwidth accumulate before we
- * need to cap it? (specified in ms).
+ * need to cap it? (specified in seconds).
*/
-#define MAX_WINDOW_TIME (5 * 60 * 1000)
+#define MAX_WINDOW_TIME_S (5 * 60)
/**
* How many messages do we queue up at most for optional
#define MAX_NOTIFY_QUEUE 16
/**
- * Minimum of bytes per minute (out) to assign to any connected peer.
- * Should be rather low; values larger than DEFAULT_BPM_IN_OUT make no
+ * Minimum bandwidth (out) to assign to any connected peer.
+ * Should be rather low; values larger than DEFAULT_BW_IN_OUT make no
* sense.
*/
-#define MIN_BPM_PER_PEER GNUNET_CONSTANTS_DEFAULT_BPM_IN_OUT
-
-/**
- * What is the smallest change (in number of bytes per minute)
- * that we consider significant enough to bother triggering?
- */
-#define MIN_BPM_CHANGE 32
+#define MIN_BANDWIDTH_PER_PEER GNUNET_CONSTANTS_DEFAULT_BW_IN_OUT
/**
* After how much time past the "official" expiration time do
*/
#define PAST_EXPIRATION_DISCARD_TIME GNUNET_TIME_UNIT_SECONDS
+/**
+ * How long do we delay messages to get larger packet sizes (CORKing)?
+ */
+#define MAX_CORK_DELAY GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 1)
+
/**
* What is the maximum delay for a SET_KEY message?
*/
-#define MAX_SET_KEY_DELAY GNUNET_TIME_UNIT_SECONDS
+#define MAX_SET_KEY_DELAY GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5)
/**
* What how long do we wait for SET_KEY confirmation initially?
*/
-#define INITIAL_SET_KEY_RETRY_FREQUENCY GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 3)
+#define INITIAL_SET_KEY_RETRY_FREQUENCY GNUNET_TIME_relative_multiply (MAX_SET_KEY_DELAY, 3)
/**
* What is the maximum delay for a PING message?
*/
-#define MAX_PING_DELAY GNUNET_TIME_UNIT_SECONDS
+#define MAX_PING_DELAY GNUNET_TIME_relative_multiply (MAX_SET_KEY_DELAY, 2)
/**
* What is the maximum delay for a PONG message?
*/
-#define MAX_PONG_DELAY GNUNET_TIME_UNIT_SECONDS
+#define MAX_PONG_DELAY GNUNET_TIME_relative_multiply (MAX_PING_DELAY, 2)
+
+/**
+ * What is the minimum frequency for a PING message?
+ */
+#define MIN_PING_FREQUENCY GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5)
/**
* How often do we recalculate bandwidth quotas?
*/
-#define QUOTA_UPDATE_FREQUENCY GNUNET_TIME_UNIT_SECONDS
+#define QUOTA_UPDATE_FREQUENCY GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5)
/**
* What is the priority for a SET_KEY message?
#define PONG_PRIORITY 0xFFFFFF
/**
- * How many messages do we queue per peer at most?
+ * How many messages do we queue per peer at most? Must be at
+ * least two.
*/
#define MAX_PEER_QUEUE_SIZE 16
* Number of bytes (at the beginning) of "struct EncryptedMessage"
* that are NOT encrypted.
*/
-#define ENCRYPTED_HEADER_SIZE (sizeof(struct GNUNET_MessageHeader) + sizeof(uint32_t) + sizeof(GNUNET_HashCode))
+#define ENCRYPTED_HEADER_SIZE (sizeof(struct GNUNET_MessageHeader) + sizeof(uint32_t))
/**
struct GNUNET_MessageHeader header;
/**
- * Always zero.
+ * Random value used for IV generation. ENCRYPTED_HEADER_SIZE must
+ * be set to the offset of the *next* field.
*/
- uint32_t reserved GNUNET_PACKED;
+ uint32_t iv_seed GNUNET_PACKED;
/**
- * Hash of the plaintext, used to verify message integrity;
- * ALSO used as the IV for the symmetric cipher! Everything
- * after this hash will be encrypted. ENCRYPTED_HEADER_SIZE
- * must be set to the offset of the next field.
+ * Hash of the plaintext (starting at 'sequence_number'), used to
+ * verify message integrity. Everything after this hash (including
+ * this hash itself) will be encrypted.
*/
GNUNET_HashCode plaintext_hash;
uint32_t sequence_number GNUNET_PACKED;
/**
- * Desired bandwidth (how much we should send to this
- * peer / how much is the sender willing to receive),
- * in bytes per minute.
+ * Desired bandwidth (how much we should send to this peer / how
+ * much is the sender willing to receive)?
*/
- uint32_t inbound_bpm_limit GNUNET_PACKED;
+ struct GNUNET_BANDWIDTH_Value32NBO inbound_bw_limit;
/**
* Timestamp. Used to prevent reply of ancient messages
};
+
/**
* We're sending an (encrypted) PING to the other peer to check if he
* can decrypt. The other peer should respond with a PONG with the
struct PingMessage
{
/**
- * Message type is either CORE_PING or CORE_PONG.
+ * Message type is CORE_PING.
*/
struct GNUNET_MessageHeader header;
};
+
+/**
+ * Response to a PING. Includes data from the original PING
+ * plus initial bandwidth quota information.
+ */
+struct PongMessage
+{
+ /**
+ * Message type is CORE_PONG.
+ */
+ struct GNUNET_MessageHeader header;
+
+ /**
+ * Random number proochosen to make reply harder. Must be
+ * first field after header (this is where we start to encrypt!).
+ */
+ uint32_t challenge GNUNET_PACKED;
+
+ /**
+ * Must be zero.
+ */
+ uint32_t reserved GNUNET_PACKED;
+
+ /**
+ * Desired bandwidth (how much we should send to this
+ * peer / how much is the sender willing to receive).
+ */
+ struct GNUNET_BANDWIDTH_Value32NBO inbound_bw_limit;
+
+ /**
+ * Intended target of the PING, used primarily to check
+ * that decryption actually worked.
+ */
+ struct GNUNET_PeerIdentity target;
+};
+
+
/**
* Message transmitted to set (or update) a session key.
*/
{
/**
- * We keep messages in a linked list (for now).
+ * We keep messages in a doubly linked list.
*/
struct MessageEntry *next;
+ /**
+ * We keep messages in a doubly linked list.
+ */
+ struct MessageEntry *prev;
+
/**
* By when are we supposed to transmit this message?
*/
struct GNUNET_TIME_Absolute deadline;
+ /**
+ * By when are we supposed to transmit this message (after
+ * giving slack)?
+ */
+ struct GNUNET_TIME_Absolute slack_deadline;
+
/**
* How important is this message to us?
*/
* (or the SET_KEY). We keep it here until we have a key
* to decrypt it. NULL if no PONG is pending.
*/
- struct PingMessage *pending_pong;
+ struct PongMessage *pending_pong;
/**
* Non-NULL if we are currently looking up HELLOs for this peer.
*/
GNUNET_SCHEDULER_TaskIdentifier quota_update_task;
+ /**
+ * ID of task used for sending keep-alive pings.
+ */
+ GNUNET_SCHEDULER_TaskIdentifier keep_alive_task;
+
/**
* ID of task used for cleaning up dead neighbour entries.
*/
struct GNUNET_TIME_Relative set_key_retry_frequency;
/**
- * Time of our last update to the "available_send_window".
- */
- struct GNUNET_TIME_Absolute last_asw_update;
-
- /**
- * Time of our last update to the "available_recv_window".
+ * Tracking bandwidth for sending to this peer.
*/
- struct GNUNET_TIME_Absolute last_arw_update;
+ struct GNUNET_BANDWIDTH_Tracker available_send_window;
/**
- * Number of bytes that we are eligible to transmit to this
- * peer at this point. Incremented every minute by max_out_bpm,
- * bounded by max_bpm (no back-log larger than MAX_BUF_FACT minutes,
- * bandwidth-hogs are sampled at a frequency of about 78s!);
- * may get negative if we have VERY high priority content.
+ * Tracking bandwidth for receiving from this peer.
*/
- long long available_send_window;
-
- /**
- * How much downstream capacity of this peer has been reserved for
- * our traffic? (Our clients can request that a certain amount of
- * bandwidth is available for replies to them; this value is used to
- * make sure that this reserved amount of bandwidth is actually
- * available).
- */
- long long available_recv_window;
+ struct GNUNET_BANDWIDTH_Tracker available_recv_window;
/**
* How valueable were the messages of this peer recently?
*/
unsigned int last_packets_bitmap;
- /**
- * Number of messages in the message queue for this peer.
- */
- unsigned int message_queue_size;
-
/**
* last sequence number received on this connection (highest)
*/
/**
* Available bandwidth in for this peer (current target).
*/
- uint32_t bpm_in;
+ struct GNUNET_BANDWIDTH_Value32NBO bw_in;
/**
* Available bandwidth out for this peer (current target).
*/
- uint32_t bpm_out;
+ struct GNUNET_BANDWIDTH_Value32NBO bw_out;
/**
- * Internal bandwidth limit set for this peer (initially
- * typically set to "-1"). "bpm_out" is MAX of
- * "bpm_out_internal_limit" and "bpm_out_external_limit".
+ * 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".
*/
- uint32_t bpm_out_internal_limit;
+ struct GNUNET_BANDWIDTH_Value32NBO bw_out_internal_limit;
/**
* External bandwidth limit set for this peer by the
- * peer that we are communicating with. "bpm_out" is MAX of
- * "bpm_out_internal_limit" and "bpm_out_external_limit".
+ * peer that we are communicating with. "bw_out" is MIN of
+ * "bw_out_internal_limit" and "bw_out_external_limit".
*/
- uint32_t bpm_out_external_limit;
+ struct GNUNET_BANDWIDTH_Value32NBO bw_out_external_limit;
/**
* What was our PING challenge number (for this peer)?
* Are we currently connected to this neighbour?
*/
int is_connected;
+
};
* about (with "tcnt" entries). Allocated as part
* of this client struct, do not free!
*/
- uint16_t *types;
+ const uint16_t *types;
/**
* Options for messages this client cares about,
*/
static struct Neighbour *neighbours;
+/**
+ * For creating statistics.
+ */
+static struct GNUNET_STATISTICS_Handle *stats;
+
/**
* Sum of all preferences among all neighbours.
*/
static unsigned int neighbour_count;
/**
- * How much inbound bandwidth are we supposed to be using?
+ * How much inbound bandwidth are we supposed to be using per second?
+ * FIXME: this value is not used!
*/
-static unsigned long long bandwidth_target_in;
+static unsigned long long bandwidth_target_in_bps;
/**
- * How much outbound bandwidth are we supposed to be using?
+ * How much outbound bandwidth are we supposed to be using per second?
*/
-static unsigned long long bandwidth_target_out;
+static unsigned long long bandwidth_target_out_bps;
preference_sum += n->current_preference;
n = n->next;
}
-}
-
-
-/**
- * Recalculate the number of bytes we expect to
- * receive or transmit in a given window.
- *
- * @param force force an update now (even if not much time has passed)
- * @param window pointer to the byte counter (updated)
- * @param ts pointer to the timestamp (updated)
- * @param bpm number of bytes per minute that should
- * be added to the window.
- */
-static void
-update_window (int force,
- long long *window,
- struct GNUNET_TIME_Absolute *ts, unsigned int bpm)
-{
- struct GNUNET_TIME_Relative since;
-
- since = GNUNET_TIME_absolute_get_duration (*ts);
- if ( (force == GNUNET_NO) &&
- (since.value < 60 * 1000) )
- return; /* not even a minute has passed */
- *ts = GNUNET_TIME_absolute_get ();
- *window += (bpm * since.value) / 60 / 1000;
- if (*window > MAX_WINDOW_TIME * bpm)
- *window = MAX_WINDOW_TIME * bpm;
+ GNUNET_STATISTICS_set (stats, gettext_noop ("# total peer preference"), preference_sum, GNUNET_NO);
}
while (c != NULL)
{
if (0 != (c->options & options))
- send_to_client (c, msg, can_drop);
+ {
+#if DEBUG_CORE_CLIENT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Sending message of type %u to client.\n",
+ ntohs (msg->type));
+#endif
+ send_to_client (c, msg, can_drop);
+ }
c = c->next;
}
}
struct Client *c;
uint16_t msize;
const uint16_t *types;
+ uint16_t *wtypes;
struct Neighbour *n;
struct ConnectNotifyMessage cnm;
+ unsigned int i;
#if DEBUG_CORE_CLIENT
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
c->client_handle = client;
c->next = clients;
clients = c;
- memcpy (&c[1], types, msize);
- c->types = (uint16_t *) & c[1];
- c->options = ntohl (im->options);
c->tcnt = msize / sizeof (uint16_t);
+ c->types = (const uint16_t *) &c[1];
+ wtypes = (uint16_t *) &c[1];
+ for (i=0;i<c->tcnt;i++)
+ wtypes[i] = ntohs (types[i]);
+ c->options = ntohl (im->options);
+#if DEBUG_CORE_CLIENT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Client %p is interested in %u message types\n",
+ c,
+ c->tcnt);
+#endif
/* send init reply message */
irm.header.size = htons (sizeof (struct InitReplyMessage));
irm.header.type = htons (GNUNET_MESSAGE_TYPE_CORE_INIT_REPLY);
"Sending `%s' message to client.\n", "INIT_REPLY");
#endif
send_to_client (c, &irm.header, GNUNET_NO);
- /* notify new client about existing neighbours */
- cnm.header.size = htons (sizeof (struct ConnectNotifyMessage));
- cnm.header.type = htons (GNUNET_MESSAGE_TYPE_CORE_NOTIFY_CONNECT);
- n = neighbours;
- while (n != NULL)
+ if (0 != (c->options & GNUNET_CORE_OPTION_SEND_CONNECT))
{
- if (n->status == PEER_STATE_KEY_CONFIRMED)
+ /* notify new client about existing neighbours */
+ cnm.header.size = htons (sizeof (struct ConnectNotifyMessage));
+ cnm.header.type = htons (GNUNET_MESSAGE_TYPE_CORE_NOTIFY_CONNECT);
+ n = neighbours;
+ while (n != NULL)
{
+ if (n->status == PEER_STATE_KEY_CONFIRMED)
+ {
#if DEBUG_CORE_CLIENT
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Sending `%s' message to client.\n", "NOTIFY_CONNECT");
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Sending `%s' message to client.\n", "NOTIFY_CONNECT");
#endif
- cnm.distance = htonl (n->last_distance);
- cnm.latency = GNUNET_TIME_relative_hton (n->last_latency);
- cnm.peer = n->peer;
- send_to_client (c, &cnm.header, GNUNET_NO);
+ cnm.distance = htonl (n->last_distance);
+ cnm.latency = GNUNET_TIME_relative_hton (n->last_latency);
+ cnm.peer = n->peer;
+ send_to_client (c, &cnm.header, GNUNET_NO);
+ }
+ n = n->next;
}
- n = n->next;
}
GNUNET_SERVER_receive_done (client, GNUNET_OK);
}
return;
#if DEBUG_CORE_CLIENT
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Client has disconnected from core service.\n");
+ "Client %p has disconnected from core service.\n",
+ client);
#endif
prev = NULL;
pos = clients;
const struct RequestInfoMessage *rcm;
struct Neighbour *n;
struct ConfigurationInfoMessage cim;
- int reserv;
+ int32_t want_reserv;
+ int32_t got_reserv;
unsigned long long old_preference;
struct GNUNET_SERVER_TransmitContext *tc;
memset (&cim, 0, sizeof (cim));
if (n != NULL)
{
- update_window (GNUNET_YES,
- &n->available_send_window,
- &n->last_asw_update,
- n->bpm_out);
- n->bpm_out_internal_limit = ntohl (rcm->limit_outbound_bpm);
- n->bpm_out = GNUNET_MAX (n->bpm_out_internal_limit,
- n->bpm_out_external_limit);
- reserv = ntohl (rcm->reserve_inbound);
- if (reserv < 0)
+ 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;
+ 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,
+ GNUNET_TIME_UNIT_FOREVER_REL,
+ NULL, NULL);
+ }
+ if (want_reserv < 0)
{
- n->available_recv_window += reserv;
+ got_reserv = want_reserv;
}
- else if (reserv > 0)
+ else if (want_reserv > 0)
{
- update_window (GNUNET_NO,
- &n->available_recv_window,
- &n->last_arw_update, n->bpm_in);
- if (n->available_recv_window < reserv)
- reserv = n->available_recv_window;
- n->available_recv_window -= reserv;
+ if (GNUNET_BANDWIDTH_tracker_get_delay (&n->available_recv_window,
+ want_reserv).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)
n->current_preference = (unsigned long long) -1;
}
update_preference_sum (n->current_preference - old_preference);
- cim.reserved_amount = htonl (reserv);
- cim.bpm_in = htonl (n->bpm_in);
- cim.bpm_out = htonl (n->bpm_out);
+#if DEBUG_CORE_QUOTA
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Received reservation request for %d bytes for peer `%4s', reserved %d bytes\n",
+ (int) want_reserv,
+ GNUNET_i2s (&rcm->peer),
+ (int) got_reserv);
+#endif
+ cim.reserved_amount = htonl (got_reserv);
+ cim.bw_in = n->bw_in;
+ cim.bw_out = n->bw_out;
cim.preference = n->current_preference;
}
cim.header.size = htons (sizeof (struct ConfigurationInfoMessage));
}
while (NULL != (m = n->encrypted_head))
{
- n->encrypted_head = m->next;
+ GNUNET_CONTAINER_DLL_remove (n->encrypted_head,
+ n->encrypted_tail,
+ m);
GNUNET_free (m);
}
if (NULL != n->th)
- GNUNET_TRANSPORT_notify_transmit_ready_cancel (n->th);
+ {
+ GNUNET_TRANSPORT_notify_transmit_ready_cancel (n->th);
+ n->th = NULL;
+ }
if (n->retry_plaintext_task != GNUNET_SCHEDULER_NO_TASK)
GNUNET_SCHEDULER_cancel (sched, n->retry_plaintext_task);
if (n->retry_set_key_task != GNUNET_SCHEDULER_NO_TASK)
GNUNET_SCHEDULER_cancel (sched, n->quota_update_task);
if (n->dead_clean_task != GNUNET_SCHEDULER_NO_TASK)
GNUNET_SCHEDULER_cancel (sched, n->dead_clean_task);
+ if (n->keep_alive_task != GNUNET_SCHEDULER_NO_TASK)
+ GNUNET_SCHEDULER_cancel (sched, n->keep_alive_task);
+ if (n->status == PEER_STATE_KEY_CONFIRMED)
+ GNUNET_STATISTICS_update (stats, gettext_noop ("# peers connected"), -1, GNUNET_NO);
GNUNET_free_non_null (n->public_key);
GNUNET_free_non_null (n->pending_ping);
GNUNET_free_non_null (n->pending_pong);
}
+/**
+ * Check if we have encrypted messages for the specified neighbour
+ * pending, and if so, check with the transport about sending them
+ * out.
+ *
+ * @param n neighbour to check.
+ */
+static void process_encrypted_neighbour_queue (struct Neighbour *n);
+
+
+/**
+ * Encrypt size bytes from in and write the result to out. Use the
+ * key for outbound traffic of the given neighbour.
+ *
+ * @param n neighbour we are sending to
+ * @param iv initialization vector to use
+ * @param in ciphertext
+ * @param out plaintext
+ * @param size size of in/out
+ * @return GNUNET_OK on success
+ */
+static int
+do_encrypt (struct Neighbour *n,
+ const GNUNET_HashCode * iv,
+ const void *in, void *out, size_t size)
+{
+ if (size != (uint16_t) size)
+ {
+ GNUNET_break (0);
+ return GNUNET_NO;
+ }
+ GNUNET_assert (size ==
+ GNUNET_CRYPTO_aes_encrypt (in,
+ (uint16_t) size,
+ &n->encrypt_key,
+ (const struct
+ GNUNET_CRYPTO_AesInitializationVector
+ *) iv, out));
+ GNUNET_STATISTICS_update (stats, gettext_noop ("# bytes encrypted"), size, GNUNET_NO);
+#if DEBUG_CORE
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Encrypted %u bytes for `%4s' using key %u\n", size,
+ GNUNET_i2s (&n->peer), n->encrypt_key.crc32);
+#endif
+ return GNUNET_OK;
+}
+
+
/**
* Consider freeing the given neighbour since we may not need
* to keep it around anymore.
consider_free_neighbour (struct Neighbour *n);
+/**
+ * Task triggered when a neighbour entry is about to time out
+ * (and we should prevent this by sending a PING).
+ *
+ * @param cls the 'struct Neighbour'
+ * @param tc scheduler context (not used)
+ */
+static void
+send_keep_alive (void *cls,
+ const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+ struct Neighbour *n = cls;
+ struct GNUNET_TIME_Relative retry;
+ struct GNUNET_TIME_Relative left;
+ struct MessageEntry *me;
+ struct PingMessage pp;
+ struct PingMessage *pm;
+
+ 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);
+ pp.challenge = htonl (n->ping_challenge);
+ pp.target = n->peer;
+#if DEBUG_CORE
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Encrypting `%s' and `%s' messages for `%4s'.\n",
+ "SET_KEY", "PING", GNUNET_i2s (&n->peer));
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Sending `%s' to `%4s' with challenge %u encrypted using key %u\n",
+ "PING",
+ GNUNET_i2s (&n->peer), n->ping_challenge, n->encrypt_key.crc32);
+#endif
+ do_encrypt (n,
+ &n->peer.hashPubKey,
+ &pp.challenge,
+ &pm->challenge,
+ sizeof (struct PingMessage) -
+ sizeof (struct GNUNET_MessageHeader));
+ process_encrypted_neighbour_queue (n);
+ /* reschedule PING job */
+ left = GNUNET_TIME_absolute_get_remaining (GNUNET_TIME_absolute_add (n->last_activity,
+ GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT));
+ retry = GNUNET_TIME_relative_max (GNUNET_TIME_relative_divide (left, 2),
+ MIN_PING_FREQUENCY);
+ n->keep_alive_task
+ = GNUNET_SCHEDULER_add_delayed (sched,
+ retry,
+ &send_keep_alive,
+ n);
+
+}
+
+
/**
* Task triggered when a neighbour entry might have gotten stale.
*
const struct GNUNET_SCHEDULER_TaskContext *tc)
{
struct Neighbour *n = cls;
+
n->dead_clean_task = GNUNET_SCHEDULER_NO_TASK;
consider_free_neighbour (n);
}
return; /* no chance */
left = GNUNET_TIME_absolute_get_remaining (GNUNET_TIME_absolute_add (n->last_activity,
- MAX_PONG_DELAY));
+ GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT));
if (left.value > 0)
{
if (n->dead_clean_task != GNUNET_SCHEDULER_NO_TASK)
prev->next = n->next;
GNUNET_assert (neighbour_count > 0);
neighbour_count--;
+ GNUNET_STATISTICS_set (stats, gettext_noop ("# active neighbours"), neighbour_count, GNUNET_NO);
free_neighbour (n);
}
-/**
- * Check if we have encrypted messages for the specified neighbour
- * pending, and if so, check with the transport about sending them
- * out.
- *
- * @param n neighbour to check.
- */
-static void process_encrypted_neighbour_queue (struct Neighbour *n);
-
-
/**
* Function called when the transport service is ready to
* receive an encrypted message for the respective peer
n->th = NULL;
GNUNET_assert (NULL != (m = n->encrypted_head));
- n->encrypted_head = m->next;
- if (m->next == NULL)
- n->encrypted_tail = NULL;
+ 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;
- n->available_send_window -= m->size;
- process_encrypted_neighbour_queue (n);
-
+ 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",
ntohs (((struct GNUNET_MessageHeader *) &m[1])->type),
ret, GNUNET_i2s (&n->peer));
#endif
+ process_encrypted_neighbour_queue (n);
}
else
{
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+#if DEBUG_CORE
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Transmission of message of type %u and size %u failed\n",
ntohs (((struct GNUNET_MessageHeader *) &m[1])->type),
m->size);
+#endif
}
GNUNET_free (m);
consider_free_neighbour (n);
if (n->th != NULL)
return; /* request already pending */
- if (n->encrypted_head == NULL)
+ m = n->encrypted_head;
+ if (m == NULL)
{
/* encrypted queue empty, try plaintext instead */
process_plaintext_neighbour_queue (n);
#if DEBUG_CORE
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Asking transport for transmission of %u bytes to `%4s' in next %llu ms\n",
- n->encrypted_head->size,
+ m->size,
GNUNET_i2s (&n->peer),
- GNUNET_TIME_absolute_get_remaining (n->
- encrypted_head->deadline).
+ GNUNET_TIME_absolute_get_remaining (m->deadline).
value);
#endif
n->th =
GNUNET_TRANSPORT_notify_transmit_ready (transport, &n->peer,
- n->encrypted_head->size,
- n->encrypted_head->priority,
+ m->size,
+ m->priority,
GNUNET_TIME_absolute_get_remaining
- (n->encrypted_head->deadline),
+ (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_assert (NULL != (m = n->encrypted_head));
- n->encrypted_head = m->next;
- if (m->next == NULL)
- n->encrypted_tail = NULL;
+ GNUNET_CONTAINER_DLL_remove (n->encrypted_head,
+ n->encrypted_tail,
+ m);
GNUNET_free (m);
process_encrypted_neighbour_queue (n);
}
GNUNET_break (0);
return GNUNET_SYSERR;
}
+ GNUNET_STATISTICS_update (stats, gettext_noop ("# bytes decrypted"), size, GNUNET_NO);
#if DEBUG_CORE
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Decrypted %u bytes from `%4s' using key %u\n",
}
-/**
- * Encrypt size bytes from in and write the result to out. Use the
- * key for outbound traffic of the given neighbour.
- *
- * @param n neighbour we are sending to
- * @param iv initialization vector to use
- * @param in ciphertext
- * @param out plaintext
- * @param size size of in/out
- * @return GNUNET_OK on success
- */
-static int
-do_encrypt (struct Neighbour *n,
- const GNUNET_HashCode * iv,
- const void *in, void *out, size_t size)
-{
- if (size != (uint16_t) size)
- {
- GNUNET_break (0);
- return GNUNET_NO;
- }
- GNUNET_assert (size ==
- GNUNET_CRYPTO_aes_encrypt (in,
- (uint16_t) size,
- &n->encrypt_key,
- (const struct
- GNUNET_CRYPTO_AesInitializationVector
- *) iv, out));
-#if DEBUG_CORE
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Encrypted %u bytes for `%4s' using key %u\n", size,
- GNUNET_i2s (&n->peer), n->encrypt_key.crc32);
-#endif
- return GNUNET_OK;
-}
-
-
/**
* Select messages for transmission. This heuristic uses a combination
* of earliest deadline first (EDF) scheduling (with bounded horizon)
unsigned int min_prio;
struct GNUNET_TIME_Absolute t;
struct GNUNET_TIME_Absolute now;
- uint64_t delta;
+ struct GNUNET_TIME_Relative delta;
uint64_t avail;
- unsigned long long slack; /* how long could we wait before missing deadlines? */
+ struct GNUNET_TIME_Relative slack; /* how long could we wait before missing deadlines? */
size_t off;
+ uint64_t tsize;
+ unsigned int queue_size;
int discard_low_prio;
GNUNET_assert (NULL != n->messages);
/* should we remove the entry with the lowest
priority from consideration for scheduling at the
end of the loop? */
+ queue_size = 0;
+ tsize = 0;
+ pos = n->messages;
+ while (pos != NULL)
+ {
+ queue_size++;
+ tsize += pos->size;
+ pos = pos->next;
+ }
discard_low_prio = GNUNET_YES;
while (GNUNET_YES == discard_low_prio)
{
min_prio = -1;
discard_low_prio = GNUNET_NO;
/* calculate number of bytes available for transmission at time "t" */
- update_window (GNUNET_NO,
- &n->available_send_window,
- &n->last_asw_update,
- n->bpm_out);
- avail = n->available_send_window;
- t = n->last_asw_update;
+ avail = GNUNET_BANDWIDTH_tracker_get_available (&n->available_send_window);
+ t = now;
/* how many bytes have we (hypothetically) scheduled so far */
off = 0;
/* maximum time we can wait before transmitting anything
and still make all of our deadlines */
- slack = -1;
-
+ slack = MAX_CORK_DELAY;
pos = n->messages;
/* note that we use "*2" here because we want to look
a bit further into the future; much more makes no
sense since new message might be scheduled in the
meantime... */
while ((pos != NULL) && (off < size * 2))
- {
+ {
if (pos->do_transmit == GNUNET_YES)
{
/* already removed from consideration */
}
if (discard_low_prio == GNUNET_NO)
{
- delta = pos->deadline.value;
- if (delta < t.value)
- delta = 0;
- else
- delta = t.value - delta;
- avail += delta * n->bpm_out / 1000 / 60;
+ delta = GNUNET_TIME_absolute_get_difference (t, pos->deadline);
+ if (delta.value > 0)
+ {
+ // FIXME: HUH? Check!
+ t = pos->deadline;
+ avail += GNUNET_BANDWIDTH_value_get_available_until (n->bw_out,
+ delta);
+ }
if (avail < pos->size)
{
+ // FIXME: HUH? Check!
discard_low_prio = GNUNET_YES; /* we could not schedule this one! */
}
else
/* update slack, considering both its absolute deadline
and relative deadlines caused by other messages
with their respective load */
- slack = GNUNET_MIN (slack, avail / n->bpm_out);
- if ( (pos->deadline.value < now.value) ||
- (GNUNET_YES == pos->got_slack) )
+ slack = GNUNET_TIME_relative_min (slack,
+ GNUNET_BANDWIDTH_value_get_delay_for (n->bw_out,
+ avail));
+ if (pos->deadline.value <= now.value)
{
- slack = 0;
+ /* now or never */
+ slack = GNUNET_TIME_UNIT_ZERO;
+ }
+ else if (GNUNET_YES == pos->got_slack)
+ {
+ /* should be soon now! */
+ slack = GNUNET_TIME_relative_min (slack,
+ GNUNET_TIME_absolute_get_remaining (pos->slack_deadline));
}
else
{
slack =
- GNUNET_MIN (slack, pos->deadline.value - now.value);
+ GNUNET_TIME_relative_min (slack,
+ GNUNET_TIME_absolute_get_difference (now, pos->deadline));
pos->got_slack = GNUNET_YES;
+ pos->slack_deadline = GNUNET_TIME_absolute_min (pos->deadline,
+ GNUNET_TIME_relative_to_absolute (MAX_CORK_DELAY));
}
}
}
-
off += pos->size;
- t.value = GNUNET_MAX (pos->deadline.value, t.value);
+ t = GNUNET_TIME_absolute_max (pos->deadline, t); // HUH? Check!
if (pos->priority <= min_prio)
{
/* update min for discard */
}
/* guard against sending "tiny" messages with large headers without
urgent deadlines */
- if ( (slack > 1000) && (size > 4 * off) )
+ if ( (slack.value > 0) &&
+ (size > 4 * off) &&
+ (queue_size <= MAX_PEER_QUEUE_SIZE - 2) )
{
/* less than 25% of message would be filled with deadlines still
being met if we delay by one second or more; so just wait for
more data; but do not wait longer than 1s (since we don't want
to delay messages for a really long time either). */
- retry_time->value = 1000;
+ *retry_time = MAX_CORK_DELAY;
/* reset do_transmit values for next time */
while (pos != last)
{
}
#if DEBUG_CORE
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Deferring transmission for 1s due to underfull message buffer size\n");
+ "Deferring transmission for %llums due to underfull message buffer size (%u/%u)\n",
+ (unsigned long long) slack.value,
+ (unsigned int) off,
+ (unsigned int) size);
#endif
return 0;
}
}
#if DEBUG_CORE
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Selected %u bytes of plaintext messages for transmission to `%4s'.\n",
- off, GNUNET_i2s (&n->peer));
+ "Selected %u/%u bytes of %u/%u plaintext messages for transmission to `%4s'.\n",
+ off, tsize,
+ queue_size, MAX_PEER_QUEUE_SIZE,
+ GNUNET_i2s (&n->peer));
#endif
return off;
}
*retry_time = GNUNET_TIME_UNIT_FOREVER_REL;
if (0 == select_messages (n, size, retry_time))
{
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+#if DEBUG_CORE
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"No messages selected, will try again in %llu ms\n",
retry_time->value);
+#endif
return 0;
}
ntm->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_NOTIFY_OUTBOUND);
*priority += pos->priority;
#if DEBUG_CORE
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Adding plaintext message with deadline %llu ms to batch\n",
+ "Adding plaintext message of size %u with deadline %llu ms to batch\n",
+ pos->size,
GNUNET_TIME_absolute_get_remaining (pos->deadline).value);
#endif
deadline->value = GNUNET_MIN (deadline->value, pos->deadline.value);
{
struct Neighbour *n = cls;
+#if DEBUG_CORE
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Retrying key transmission to `%4s'\n",
+ GNUNET_i2s (&n->peer));
+#endif
n->retry_set_key_task = GNUNET_SCHEDULER_NO_TASK;
n->set_key_retry_frequency =
GNUNET_TIME_relative_multiply (n->set_key_retry_frequency, 2);
unsigned int priority;
struct GNUNET_TIME_Absolute deadline;
struct GNUNET_TIME_Relative retry_time;
+ GNUNET_HashCode iv;
if (n->retry_plaintext_task != GNUNET_SCHEDULER_NO_TASK)
{
&retry_plaintext_processing, n);
return;
}
+#if DEBUG_CORE_QUOTA
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Sending %u b/s as new limit to peer `%4s'\n",
+ (unsigned int) ntohl (n->bw_in.value__),
+ GNUNET_i2s (&n->peer));
+#endif
+ ph->iv_seed = htonl (GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, -1));
ph->sequence_number = htonl (++n->last_sequence_number_sent);
- ph->inbound_bpm_limit = htonl (n->bpm_in);
+ ph->inbound_bw_limit = n->bw_in;
ph->timestamp = GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get ());
/* setup encryption message header */
em = (struct EncryptedMessage *) &me[1];
em->header.size = htons (used);
em->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_ENCRYPTED_MESSAGE);
- em->reserved = htonl (0);
+ em->iv_seed = ph->iv_seed;
esize = used - ENCRYPTED_HEADER_SIZE;
- GNUNET_CRYPTO_hash (&ph->sequence_number, esize, &em->plaintext_hash);
+ GNUNET_CRYPTO_hash (&ph->sequence_number,
+ esize - sizeof (GNUNET_HashCode),
+ &ph->plaintext_hash);
+ GNUNET_CRYPTO_hash (&ph->iv_seed, sizeof (uint32_t), &iv);
/* encrypt */
#if DEBUG_CORE
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
#endif
GNUNET_assert (GNUNET_OK ==
do_encrypt (n,
- &em->plaintext_hash,
- &ph->sequence_number,
- &em->sequence_number, esize));
+ &iv,
+ &ph->plaintext_hash,
+ &em->plaintext_hash, esize));
/* append to transmission list */
- if (n->encrypted_tail == NULL)
- n->encrypted_head = me;
- else
- n->encrypted_tail->next = me;
- n->encrypted_tail = me;
+ GNUNET_CONTAINER_DLL_insert_after (n->encrypted_head,
+ n->encrypted_tail,
+ n->encrypted_tail,
+ me);
process_encrypted_neighbour_queue (n);
}
n->next = neighbours;
neighbours = n;
neighbour_count++;
+ GNUNET_STATISTICS_set (stats, gettext_noop ("# active neighbours"), neighbour_count, GNUNET_NO);
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->bpm_in = GNUNET_CONSTANTS_DEFAULT_BPM_IN_OUT;
- n->bpm_out = GNUNET_CONSTANTS_DEFAULT_BPM_IN_OUT;
- n->bpm_out_internal_limit = (uint32_t) - 1;
- n->bpm_out_external_limit = GNUNET_CONSTANTS_DEFAULT_BPM_IN_OUT;
+ 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_t) - 1);
+ n->bw_out_external_limit = GNUNET_CONSTANTS_DEFAULT_BW_IN_OUT;
n->ping_challenge = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
(uint32_t) - 1);
- schedule_quota_update (n);
+ neighbour_quota_update (n, NULL);
return n;
}
-
-
-
/**
* Handle CORE_SEND request.
*
const struct GNUNET_MessageHeader *message)
{
const struct SendMessage *sm;
- const struct GNUNET_MessageHeader *mh;
struct Neighbour *n;
struct MessageEntry *prev;
struct MessageEntry *pos;
}
sm = (const struct SendMessage *) message;
msize -= sizeof (struct SendMessage);
- mh = (const struct GNUNET_MessageHeader *) &sm[1];
- if (msize != ntohs (mh->size))
- {
- GNUNET_break (0);
- if (client != NULL)
- GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
- return;
- }
n = find_neighbour (&sm->peer);
if (n == NULL)
n = create_neighbour (&sm->peer);
/* discard new entry */
#if DEBUG_CORE
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Queue full, discarding new request\n");
+ "Queue full (%u/%u), discarding new request (%u bytes of type %u)\n",
+ queue_size,
+ MAX_PEER_QUEUE_SIZE,
+ msize,
+ ntohs (message->type));
#endif
if (client != NULL)
GNUNET_SERVER_receive_done (client, GNUNET_OK);
#if DEBUG_CORE
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Adding transmission request for `%4s' to queue\n",
- GNUNET_i2s (&sm->peer));
+ "Adding transmission request for `%4s' of size %u to queue\n",
+ GNUNET_i2s (&sm->peer),
+ msize);
#endif
e = GNUNET_malloc (sizeof (struct MessageEntry) + msize);
e->deadline = GNUNET_TIME_absolute_ntoh (sm->deadline);
e->priority = ntohl (sm->priority);
e->size = msize;
- memcpy (&e[1], mh, msize);
+ memcpy (&e[1], &sm[1], msize);
/* insert, keep list sorted by deadline */
prev = NULL;
}
+/**
+ * Function called when the transport service is ready to
+ * receive a message. Only resets 'n->th' to NULL.
+ *
+ * @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
+ */
+static size_t
+notify_transport_connect_done (void *cls, size_t size, void *buf)
+{
+ struct Neighbour *n = cls;
+
+ n->th = NULL;
+ if (buf == NULL)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ _("Failed to connect to `%4s': transport failed to connect\n"),
+ GNUNET_i2s (&n->peer));
+ return 0;
+ }
+ send_key (n);
+ return 0;
+}
+
+
/**
* Handle CORE_REQUEST_CONNECT request.
*
{
const struct ConnectMessage *cm = (const struct ConnectMessage*) message;
struct Neighbour *n;
+ struct GNUNET_TIME_Relative timeout;
+ if (0 == memcmp (&cm->peer, &my_identity, sizeof (struct GNUNET_PeerIdentity)))
+ {
+ GNUNET_break (0);
+ GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+ return;
+ }
GNUNET_SERVER_receive_done (client, GNUNET_OK);
n = find_neighbour (&cm->peer);
if (n == NULL)
if ( (n->is_connected) ||
(n->th != NULL) )
return; /* already connected, or at least trying */
+ GNUNET_STATISTICS_update (stats, gettext_noop ("# connection requests received"), 1, GNUNET_NO);
#if DEBUG_CORE
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Core received `%s' request for `%4s', will try to establish connection\n",
"REQUEST_CONNECT",
GNUNET_i2s (&cm->peer));
#endif
+ timeout = GNUNET_TIME_relative_ntoh (cm->timeout);
/* ask transport to connect to the peer */
- /* FIXME: timeout zero OK? */
n->th = GNUNET_TRANSPORT_notify_transmit_ready (transport,
&cm->peer,
- 0, 0,
- GNUNET_TIME_UNIT_ZERO,
- NULL,
- NULL);
+ sizeof (struct GNUNET_MessageHeader), 0,
+ timeout,
+ ¬ify_transport_connect_done,
+ n);
+ GNUNET_break (NULL != n->th);
}
n->pitr = NULL;
if (n->public_key != NULL)
{
+ GNUNET_STATISTICS_update (stats,
+ gettext_noop ("# SETKEY messages deferred (need public key)"),
+ -1,
+ GNUNET_NO);
send_key (n);
}
else
{
+ GNUNET_STATISTICS_update (stats,
+ gettext_noop ("# Delayed connecting due to lack of public key"),
+ 1,
+ GNUNET_NO);
if (GNUNET_SCHEDULER_NO_TASK == n->retry_set_key_task)
n->retry_set_key_task
= GNUNET_SCHEDULER_add_delayed (sched,
GNUNET_malloc (sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded));
if (GNUNET_OK != GNUNET_HELLO_get_key (hello, n->public_key))
{
+ GNUNET_STATISTICS_update (stats,
+ gettext_noop ("# Error extracting public key from HELLO"),
+ 1,
+ GNUNET_NO);
GNUNET_free (n->public_key);
n->public_key = NULL;
#if DEBUG_CORE
me->deadline = GNUNET_TIME_relative_to_absolute (MAX_SET_KEY_DELAY);
me->priority = SET_KEY_PRIORITY;
me->size = sizeof (struct SetKeyMessage);
- if (n->encrypted_head == NULL)
- n->encrypted_head = me;
- else
- n->encrypted_tail->next = me;
- n->encrypted_tail = me;
+ GNUNET_CONTAINER_DLL_insert_after (n->encrypted_head,
+ n->encrypted_tail,
+ n->encrypted_tail,
+ me);
sm = (struct SetKeyMessage *) &me[1];
sm->header.size = htons (sizeof (struct SetKeyMessage));
sm->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_SET_KEY);
me->deadline = GNUNET_TIME_relative_to_absolute (MAX_PING_DELAY);
me->priority = PING_PRIORITY;
me->size = sizeof (struct PingMessage);
- n->encrypted_tail->next = me;
- n->encrypted_tail = me;
+ 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);
handle_ping (struct Neighbour *n, const struct PingMessage *m)
{
struct PingMessage t;
- struct PingMessage *tp;
+ struct PongMessage tx;
+ struct PongMessage *tp;
struct MessageEntry *me;
#if DEBUG_CORE
"Target of `%s' request is `%4s'.\n",
"PING", GNUNET_i2s (&t.target));
#endif
+ GNUNET_STATISTICS_update (stats, gettext_noop ("# ping messages decrypted"), 1, GNUNET_NO);
if (0 != memcmp (&t.target,
&my_identity, sizeof (struct GNUNET_PeerIdentity)))
{
return;
}
me = GNUNET_malloc (sizeof (struct MessageEntry) +
- sizeof (struct PingMessage));
- if (n->encrypted_tail != NULL)
- n->encrypted_tail->next = me;
- else
- {
- n->encrypted_tail = me;
- n->encrypted_head = me;
- }
+ sizeof (struct PongMessage));
+ GNUNET_CONTAINER_DLL_insert_after (n->encrypted_head,
+ n->encrypted_tail,
+ n->encrypted_tail,
+ me);
me->deadline = GNUNET_TIME_relative_to_absolute (MAX_PONG_DELAY);
me->priority = PONG_PRIORITY;
- me->size = sizeof (struct PingMessage);
- tp = (struct PingMessage *) &me[1];
+ me->size = sizeof (struct PongMessage);
+ tx.reserved = htonl (0);
+ tx.inbound_bw_limit = n->bw_in;
+ tx.challenge = t.challenge;
+ tx.target = t.target;
+ tp = (struct PongMessage *) &me[1];
tp->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_PONG);
- tp->header.size = htons (sizeof (struct PingMessage));
+ tp->header.size = htons (sizeof (struct PongMessage));
do_encrypt (n,
&my_identity.hashPubKey,
- &t.challenge,
+ &tx.challenge,
&tp->challenge,
- sizeof (struct PingMessage) -
+ sizeof (struct PongMessage) -
sizeof (struct GNUNET_MessageHeader));
#if DEBUG_CORE
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
* @param m the encrypted PONG message itself
*/
static void
-handle_pong (struct Neighbour *n, const struct PingMessage *m)
+handle_pong (struct Neighbour *n,
+ const struct PongMessage *m)
{
- struct PingMessage t;
+ struct PongMessage t;
struct ConnectNotifyMessage cnm;
#if DEBUG_CORE
&n->peer.hashPubKey,
&m->challenge,
&t.challenge,
- sizeof (struct PingMessage) -
+ sizeof (struct PongMessage) -
sizeof (struct GNUNET_MessageHeader)))
return;
+ if (0 != ntohl (t.reserved))
+ {
+ GNUNET_break_op (0);
+ return;
+ }
#if DEBUG_CORE
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Decrypted `%s' from `%4s' with challenge %u using key %u\n",
return;
case PEER_STATE_KEY_RECEIVED:
n->status = PEER_STATE_KEY_CONFIRMED;
+ if (n->bw_out_external_limit.value__ != t.inbound_bw_limit.value__)
+ {
+ n->bw_out_external_limit = t.inbound_bw_limit;
+ n->bw_out = GNUNET_BANDWIDTH_value_min (n->bw_out_external_limit,
+ n->bw_out_internal_limit);
+ GNUNET_BANDWIDTH_tracker_update_quota (&n->available_send_window,
+ n->bw_out);
+ GNUNET_TRANSPORT_set_quota (transport,
+ &n->peer,
+ n->bw_in,
+ n->bw_out,
+ GNUNET_TIME_UNIT_FOREVER_REL,
+ NULL, NULL);
+ }
#if DEBUG_CORE
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Confirmed key via `%s' message for peer `%4s'\n",
"PONG", GNUNET_i2s (&n->peer));
-#endif
+#endif
if (n->retry_set_key_task != GNUNET_SCHEDULER_NO_TASK)
{
GNUNET_SCHEDULER_cancel (sched, n->retry_set_key_task);
cnm.peer = n->peer;
send_to_all_clients (&cnm.header, GNUNET_YES, GNUNET_CORE_OPTION_SEND_CONNECT);
process_encrypted_neighbour_queue (n);
- break;
+ /* fall-through! */
case PEER_STATE_KEY_CONFIRMED:
- /* duplicate PONG? */
+ n->last_activity = GNUNET_TIME_absolute_get ();
+ if (n->keep_alive_task != GNUNET_SCHEDULER_NO_TASK)
+ GNUNET_SCHEDULER_cancel (sched, n->keep_alive_task);
+ n->keep_alive_task
+ = GNUNET_SCHEDULER_add_delayed (sched,
+ GNUNET_TIME_relative_divide (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT, 2),
+ &send_keep_alive,
+ n);
break;
default:
GNUNET_break (0);
struct GNUNET_TIME_Absolute t;
struct GNUNET_CRYPTO_AesSessionKey k;
struct PingMessage *ping;
- struct PingMessage *pong;
+ struct PongMessage *pong;
enum PeerStateMachine sender_status;
#if DEBUG_CORE
0,
GNUNET_TIME_UNIT_MINUTES,
&process_hello_retry_handle_set_key, n);
+ GNUNET_STATISTICS_update (stats, gettext_noop ("# SETKEY messages deferred (need public key)"), 1, GNUNET_NO);
return;
}
if (0 != memcmp (&m->target,
sizeof (struct GNUNET_PeerIdentity)))
{
GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- _("Received `%s' message that was not for me. Ignoring.\n"),
- "SET_KEY");
+ _("Received `%s' message that was for `%s', not for me. Ignoring.\n"),
+ "SET_KEY",
+ GNUNET_i2s (&m->target));
return;
}
if ((ntohl (m->purpose.size) !=
return;
}
#if DEBUG_CORE
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Decrypting key material.\n");
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Decrypting key material.\n");
#endif
if ((GNUNET_CRYPTO_rsa_decrypt (my_private_key,
&m->encrypted_key,
uint16_t type;
unsigned int tpos;
int deliver_full;
+ int dropped;
type = ntohs (m->type);
-#if DEBUG_HANDSHAKE
- fprintf (stderr,
- "Received encapsulated message of type %u from `%4s'\n",
- type,
- GNUNET_i2s (&sender->peer));
+#if DEBUG_CORE
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Received encapsulated message of type %u from `%4s'\n",
+ type,
+ GNUNET_i2s (&sender->peer));
#endif
+ dropped = GNUNET_YES;
cpos = clients;
while (cpos != NULL)
{
}
}
if (GNUNET_YES == deliver_full)
- send_p2p_message_to_client (sender, cpos, m, msize);
+ {
+ 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));
+ {
+ 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",
+ type,
+ GNUNET_i2s (&sender->peer));
+#endif
+ /* FIXME: stats... */
+ }
}
size_t off;
uint32_t snum;
struct GNUNET_TIME_Absolute t;
+ GNUNET_HashCode iv;
#if DEBUG_CORE
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Core service receives `%s' request from `%4s'.\n",
"ENCRYPTED_MESSAGE", GNUNET_i2s (&n->peer));
#endif
+ GNUNET_CRYPTO_hash (&m->iv_seed, sizeof (uint32_t), &iv);
/* decrypt */
if (GNUNET_OK !=
do_decrypt (n,
+ &iv,
&m->plaintext_hash,
- &m->sequence_number,
- &buf[ENCRYPTED_HEADER_SIZE], size - ENCRYPTED_HEADER_SIZE))
+ &buf[ENCRYPTED_HEADER_SIZE],
+ size - ENCRYPTED_HEADER_SIZE))
return;
pt = (struct EncryptedMessage *) buf;
/* validate hash */
GNUNET_CRYPTO_hash (&pt->sequence_number,
- size - ENCRYPTED_HEADER_SIZE, &ph);
- if (0 != memcmp (&ph, &m->plaintext_hash, sizeof (GNUNET_HashCode)))
+ size - ENCRYPTED_HEADER_SIZE - sizeof (GNUNET_HashCode), &ph);
+ if (0 != memcmp (&ph,
+ &pt->plaintext_hash,
+ sizeof (GNUNET_HashCode)))
{
/* checksum failed */
GNUNET_break_op (0);
}
/* process decrypted message(s) */
- update_window (GNUNET_YES,
- &n->available_send_window,
- &n->last_asw_update,
- n->bpm_out);
- n->bpm_out_external_limit = ntohl (pt->inbound_bpm_limit);
- n->bpm_out = GNUNET_MAX (n->bpm_out_external_limit,
- n->bpm_out_internal_limit);
+ if (n->bw_out_external_limit.value__ != pt->inbound_bw_limit.value__)
+ {
+#if DEBUG_CORE_SET_QUOTA
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Received %u b/s as new inbound limit for peer `%4s'\n",
+ (unsigned int) ntohl (pt->inbound_bw_limit.value__),
+ GNUNET_i2s (&n->peer));
+#endif
+ n->bw_out_external_limit = pt->inbound_bw_limit;
+ n->bw_out = GNUNET_BANDWIDTH_value_min (n->bw_out_external_limit,
+ n->bw_out_internal_limit);
+ GNUNET_BANDWIDTH_tracker_update_quota (&n->available_send_window,
+ n->bw_out);
+ GNUNET_TRANSPORT_set_quota (transport,
+ &n->peer,
+ n->bw_in,
+ n->bw_out,
+ GNUNET_TIME_UNIT_FOREVER_REL,
+ NULL, NULL);
+ }
n->last_activity = GNUNET_TIME_absolute_get ();
+ if (n->keep_alive_task != GNUNET_SCHEDULER_NO_TASK)
+ GNUNET_SCHEDULER_cancel (sched, n->keep_alive_task);
+ n->keep_alive_task
+ = GNUNET_SCHEDULER_add_delayed (sched,
+ GNUNET_TIME_relative_divide (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT, 2),
+ &send_keep_alive,
+ n);
off = sizeof (struct EncryptedMessage);
deliver_messages (n, buf, size, off);
}
#endif
n = find_neighbour (peer);
if (n == NULL)
- {
- GNUNET_break (0);
- return;
- }
- GNUNET_break (n->is_connected);
+ n = create_neighbour (peer);
+ if (n == NULL)
+ return;
n->last_latency = latency;
n->last_distance = distance;
up = (n->status == PEER_STATE_KEY_CONFIRMED);
GNUNET_break_op (0);
return;
}
+ GNUNET_STATISTICS_update (stats, gettext_noop ("# session keys received"), 1, GNUNET_NO);
handle_set_key (n, (const struct SetKeyMessage *) message);
break;
case GNUNET_MESSAGE_TYPE_CORE_ENCRYPTED_MESSAGE:
(n->status != PEER_STATE_KEY_CONFIRMED))
{
GNUNET_break_op (0);
+ /* blacklist briefly (?); might help recover (?) */
+ GNUNET_TRANSPORT_blacklist (sched, cfg,
+ &n->peer,
+ GNUNET_TIME_UNIT_SECONDS,
+ GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
+ 5),
+ NULL, NULL);
return;
}
handle_encrypted_message (n, (const struct EncryptedMessage *) message);
GNUNET_break_op (0);
return;
}
+ GNUNET_STATISTICS_update (stats, gettext_noop ("# ping messages received"), 1, GNUNET_NO);
if ((n->status != PEER_STATE_KEY_RECEIVED) &&
(n->status != PEER_STATE_KEY_CONFIRMED))
{
handle_ping (n, (const struct PingMessage *) message);
break;
case GNUNET_MESSAGE_TYPE_CORE_PONG:
- if (size != sizeof (struct PingMessage))
+ if (size != sizeof (struct PongMessage))
{
GNUNET_break_op (0);
return;
}
+ GNUNET_STATISTICS_update (stats, gettext_noop ("# pong messages received"), 1, GNUNET_NO);
if ( (n->status != PEER_STATE_KEY_RECEIVED) &&
(n->status != PEER_STATE_KEY_CONFIRMED) )
{
"PONG", GNUNET_i2s (&n->peer));
#endif
GNUNET_free_non_null (n->pending_pong);
- n->pending_pong = GNUNET_malloc (sizeof (struct PingMessage));
- memcpy (n->pending_pong, message, sizeof (struct PingMessage));
+ n->pending_pong = GNUNET_malloc (sizeof (struct PongMessage));
+ memcpy (n->pending_pong, message, sizeof (struct PongMessage));
return;
}
- handle_pong (n, (const struct PingMessage *) message);
+ handle_pong (n, (const struct PongMessage *) message);
break;
default:
GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
now = GNUNET_TIME_absolute_get ();
n->last_activity = now;
if (!up)
- n->time_established = now;
+ {
+ GNUNET_STATISTICS_update (stats, gettext_noop ("# peers connected"), 1, GNUNET_NO);
+ n->time_established = now;
+ }
+ if (n->keep_alive_task != GNUNET_SCHEDULER_NO_TASK)
+ GNUNET_SCHEDULER_cancel (sched, n->keep_alive_task);
+ n->keep_alive_task
+ = GNUNET_SCHEDULER_add_delayed (sched,
+ GNUNET_TIME_relative_divide (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT, 2),
+ &send_keep_alive,
+ n);
}
}
const struct GNUNET_SCHEDULER_TaskContext *tc)
{
struct Neighbour *n = cls;
- uint32_t q_in;
+ struct GNUNET_BANDWIDTH_Value32NBO q_in;
double pref_rel;
double share;
unsigned long long distributable;
+ uint64_t need_per_peer;
+ uint64_t need_per_second;
n->quota_update_task = GNUNET_SCHEDULER_NO_TASK;
/* calculate relative preference among all neighbours;
divides by a bit more to avoid division by zero AND to
account for possibility of new neighbours joining any time
AND to convert to double... */
- pref_rel = n->current_preference / (1.0 + preference_sum);
+ if (preference_sum == 0)
+ {
+ pref_rel = 1.0 / (double) neighbour_count;
+ }
+ else
+ {
+ pref_rel = n->current_preference / preference_sum;
+ }
+ need_per_peer = GNUNET_BANDWIDTH_value_get_available_until (MIN_BANDWIDTH_PER_PEER,
+ GNUNET_TIME_UNIT_SECONDS);
+ need_per_second = need_per_peer * neighbour_count;
distributable = 0;
- if (bandwidth_target_out > neighbour_count * MIN_BPM_PER_PEER)
- distributable = bandwidth_target_out - neighbour_count * MIN_BPM_PER_PEER;
+ if (bandwidth_target_out_bps > need_per_second)
+ distributable = bandwidth_target_out_bps - need_per_second;
share = distributable * pref_rel;
- q_in = MIN_BPM_PER_PEER + (unsigned long long) share;
+ if (share + need_per_peer > ( (uint32_t)-1))
+ q_in = GNUNET_BANDWIDTH_value_init ((uint32_t) -1);
+ else
+ q_in = GNUNET_BANDWIDTH_value_init (need_per_peer + (uint32_t) share);
/* check if we want to disconnect for good due to inactivity */
if ( (GNUNET_TIME_absolute_get_duration (n->last_activity).value > GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT.value) &&
(GNUNET_TIME_absolute_get_duration (n->time_established).value > GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT.value) )
{
#if DEBUG_CORE
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Forcing disconnect of `%4s' due to inactivity (?).\n",
- GNUNET_i2s (&n->peer));
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Forcing disconnect of `%4s' due to inactivity (?).\n",
+ GNUNET_i2s (&n->peer));
#endif
- q_in = 0; /* force disconnect */
+ q_in = GNUNET_BANDWIDTH_value_init (0); /* force disconnect */
}
- if ( (n->bpm_in + MIN_BPM_CHANGE < q_in) ||
- (n->bpm_in - MIN_BPM_CHANGE > q_in) )
+#if DEBUG_CORE_QUOTA
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Current quota for `%4s' is %u/%llu b/s in (old: %u b/s) / %u out (%u internal)\n",
+ GNUNET_i2s (&n->peer),
+ (unsigned int) ntohl (q_in.value__),
+ bandwidth_target_out_bps,
+ (unsigned int) ntohl (n->bw_in.value__),
+ (unsigned int) ntohl (n->bw_out.value__),
+ (unsigned int) ntohl (n->bw_out_internal_limit.value__));
+#endif
+ if (n->bw_in.value__ != q_in.value__)
{
- n->bpm_in = q_in;
+ n->bw_in = q_in;
GNUNET_TRANSPORT_set_quota (transport,
&n->peer,
- n->bpm_in,
- n->bpm_out,
+ n->bw_in,
+ n->bw_out,
GNUNET_TIME_UNIT_FOREVER_REL,
NULL, NULL);
}
unsigned int distance)
{
struct Neighbour *n;
- struct GNUNET_TIME_Absolute now;
struct ConnectNotifyMessage cnm;
n = find_neighbour (peer);
{
n = create_neighbour (peer);
}
- now = GNUNET_TIME_absolute_get ();
n->is_connected = GNUNET_YES;
n->last_latency = latency;
n->last_distance = distance;
- n->last_asw_update = now;
- n->last_arw_update = now;
+ GNUNET_BANDWIDTH_tracker_init (&n->available_send_window,
+ n->bw_out,
+ MAX_WINDOW_TIME_S);
+ GNUNET_BANDWIDTH_tracker_init (&n->available_recv_window,
+ n->bw_in,
+ MAX_WINDOW_TIME_S);
#if DEBUG_CORE
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Received connection from `%4s'.\n",
cnm.latency = GNUNET_TIME_relative_hton (n->last_latency);
cnm.peer = *peer;
send_to_all_clients (&cnm.header, GNUNET_YES, GNUNET_CORE_OPTION_SEND_PRE_CONNECT);
- send_key (n);
+ GNUNET_TRANSPORT_set_quota (transport,
+ &n->peer,
+ n->bw_in,
+ n->bw_out,
+ GNUNET_TIME_UNIT_FOREVER_REL,
+ NULL, NULL);
+ send_key (n);
}
"Peer `%4s' disconnected from us.\n", GNUNET_i2s (peer));
#endif
n = find_neighbour (peer);
+ if (n == NULL)
+ {
+ GNUNET_break (0);
+ return;
+ }
GNUNET_break (n->is_connected);
cnm.header.size = htons (sizeof (struct DisconnectNotifyMessage));
cnm.header.type = htons (GNUNET_MESSAGE_TYPE_CORE_NOTIFY_DISCONNECT);
neighbour_count--;
free_neighbour (n);
}
+ GNUNET_STATISTICS_set (stats, gettext_noop ("# active neighbours"), neighbour_count, GNUNET_NO);
GNUNET_SERVER_notification_context_destroy (notifier);
notifier = NULL;
while (NULL != (c = clients))
handle_client_disconnect (NULL, c->client_handle);
if (my_private_key != NULL)
GNUNET_CRYPTO_rsa_key_free (my_private_key);
+ if (stats != NULL)
+ GNUNET_STATISTICS_destroy (stats, GNUNET_NO);
}
struct GNUNET_SERVER_Handle *serv,
const struct GNUNET_CONFIGURATION_Handle *c)
{
-#if 0
- unsigned long long qin;
- unsigned long long qout;
- unsigned long long tneigh;
-#endif
char *keyfile;
sched = s;
GNUNET_CONFIGURATION_get_value_number (c,
"CORE",
"TOTAL_QUOTA_IN",
- &bandwidth_target_in)) ||
+ &bandwidth_target_in_bps)) ||
(GNUNET_OK !=
GNUNET_CONFIGURATION_get_value_number (c,
"CORE",
"TOTAL_QUOTA_OUT",
- &bandwidth_target_out)) ||
+ &bandwidth_target_out_bps)) ||
(GNUNET_OK !=
GNUNET_CONFIGURATION_get_value_filename (c,
"GNUNETD",
&handle_transport_notify_connect,
&handle_transport_notify_disconnect);
GNUNET_assert (NULL != transport);
+ stats = GNUNET_STATISTICS_create (sched, "core", cfg);
GNUNET_SCHEDULER_add_delayed (sched,
GNUNET_TIME_UNIT_FOREVER_REL,
&cleaning_task, NULL);
projects / oweals / gnunet.git / blobdiff