2 This file is part of GNUnet.
3 (C) 2009, 2010, 2011 Christian Grothoff (and other contributing authors)
5 GNUnet is free software; you can redistribute it and/or modify
6 it under the terms of the GNU General Public License as published
7 by the Free Software Foundation; either version 3, or (at your
8 option) any later version.
10 GNUnet is distributed in the hope that it will be useful, but
11 WITHOUT ANY WARRANTY; without even the implied warranty of
12 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13 General Public License for more details.
15 You should have received a copy of the GNU General Public License
16 along with GNUnet; see the file COPYING. If not, write to the
17 Free Software Foundation, Inc., 59 Temple Place - Suite 330,
18 Boston, MA 02111-1307, USA.
22 * @file core/gnunet-service-core_neighbours.c
23 * @brief code for managing of 'encrypted' sessions (key exchange done)
24 * @author Christian Grothoff
27 #include "gnunet_service_core.h"
28 #include "gnunet_service_core_neighbours.h"
29 #include "gnunet_service_core_kx.h"
30 #include "gnunet_service_core_sessions.h"
33 * Record kept for each request for transmission issued by a
34 * client that is still pending.
36 struct ClientActiveRequest;
39 * Data kept per session.
44 * Identity of the other peer.
46 struct GNUNET_PeerIdentity peer;
49 * Head of list of requests from clients for transmission to
52 struct ClientActiveRequest *active_client_request_head;
55 * Tail of list of requests from clients for transmission to
58 struct ClientActiveRequest *active_client_request_tail;
61 * Performance data for the peer.
63 struct GNUNET_TRANSPORT_ATS_Information *ats;
66 * Information about the key exchange with the other peer.
68 struct GSC_KeyExchangeInfo *kxinfo;
72 * ID of task used for cleaning up dead neighbour entries.
74 GNUNET_SCHEDULER_TaskIdentifier dead_clean_task;
77 * ID of task used for updating bandwidth quota for this neighbour.
79 GNUNET_SCHEDULER_TaskIdentifier quota_update_task;
82 * At what time did we initially establish (as in, complete session
83 * key handshake) this connection? Should be zero if status != KEY_CONFIRMED.
85 struct GNUNET_TIME_Absolute time_established;
88 * At what time did we last receive an encrypted message from the
89 * other peer? Should be zero if status != KEY_CONFIRMED.
91 struct GNUNET_TIME_Absolute last_activity;
94 * How valueable were the messages of this peer recently?
96 unsigned long long current_preference;
99 * Number of entries in 'ats'.
101 unsigned int ats_count;
104 * Bit map indicating which of the 32 sequence numbers before the last
105 * were received (good for accepting out-of-order packets and
106 * estimating reliability of the connection)
108 unsigned int last_packets_bitmap;
111 * last sequence number received on this connection (highest)
113 uint32_t last_sequence_number_received;
116 * last sequence number transmitted
118 uint32_t last_sequence_number_sent;
121 * Available bandwidth in for this peer (current target).
123 struct GNUNET_BANDWIDTH_Value32NBO bw_in;
126 * Available bandwidth out for this peer (current target).
128 struct GNUNET_BANDWIDTH_Value32NBO bw_out;
131 * Internal bandwidth limit set for this peer (initially typically
132 * set to "-1"). Actual "bw_out" is MIN of
133 * "bpm_out_internal_limit" and "bw_out_external_limit".
135 struct GNUNET_BANDWIDTH_Value32NBO bw_out_internal_limit;
138 * External bandwidth limit set for this peer by the
139 * peer that we are communicating with. "bw_out" is MIN of
140 * "bw_out_internal_limit" and "bw_out_external_limit".
142 struct GNUNET_BANDWIDTH_Value32NBO bw_out_external_limit;
148 * Map of peer identities to 'struct Session'.
150 static struct GNUNET_CONTAINER_MultiHashMap *sessions;
154 * Session entry for "this" peer.
156 static struct Session self;
159 * Sum of all preferences among all neighbours.
161 static unsigned long long preference_sum;
167 * At what time should the connection to the given neighbour
168 * time out (given no further activity?)
170 * @param n neighbour in question
171 * @return absolute timeout
173 static struct GNUNET_TIME_Absolute
174 get_neighbour_timeout (struct Neighbour *n)
176 return GNUNET_TIME_absolute_add (n->last_activity,
177 GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT);
182 * Helper function for update_preference_sum.
185 update_preference (void *cls, const GNUNET_HashCode * key, void *value)
187 unsigned long long *ps = cls;
188 struct Neighbour *n = value;
190 n->current_preference /= 2;
191 *ps += n->current_preference;
197 * A preference value for a neighbour was update. Update
198 * the preference sum accordingly.
200 * @param inc how much was a preference value increased?
203 update_preference_sum (unsigned long long inc)
205 unsigned long long os;
208 preference_sum += inc;
209 if (preference_sum >= os)
211 /* overflow! compensate by cutting all values in half! */
213 GNUNET_CONTAINER_multihashmap_iterate (neighbours, &update_preference,
215 GNUNET_STATISTICS_set (stats, gettext_noop ("# total peer preference"),
216 preference_sum, GNUNET_NO);
221 * Find the entry for the given neighbour.
223 * @param peer identity of the neighbour
224 * @return NULL if we are not connected, otherwise the
227 static struct Neighbour *
228 find_neighbour (const struct GNUNET_PeerIdentity *peer)
230 return GNUNET_CONTAINER_multihashmap_get (neighbours, &peer->hashPubKey);
235 * Function called by transport telling us that a peer
238 * @param n the peer that changed status
241 handle_peer_status_change (struct Neighbour *n)
243 struct PeerStatusNotifyMessage *psnm;
244 char buf[GNUNET_SERVER_MAX_MESSAGE_SIZE - 1];
245 struct GNUNET_TRANSPORT_ATS_Information *ats;
248 if ((!n->is_connected) || (n->status != PEER_STATE_KEY_CONFIRMED))
251 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Peer `%4s' changed status\n",
252 GNUNET_i2s (&n->peer));
255 sizeof (struct PeerStatusNotifyMessage) +
256 n->ats_count * sizeof (struct GNUNET_TRANSPORT_ATS_Information);
257 if (size >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
260 /* recovery strategy: throw away performance data */
261 GNUNET_array_grow (n->ats, n->ats_count, 0);
263 sizeof (struct PeerStatusNotifyMessage) +
264 n->ats_count * sizeof (struct GNUNET_TRANSPORT_ATS_Information);
266 psnm = (struct PeerStatusNotifyMessage *) buf;
267 psnm->header.size = htons (size);
268 psnm->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_NOTIFY_STATUS_CHANGE);
269 psnm->timeout = GNUNET_TIME_absolute_hton (get_neighbour_timeout (n));
270 psnm->bandwidth_in = n->bw_in;
271 psnm->bandwidth_out = n->bw_out;
272 psnm->peer = n->peer;
273 psnm->ats_count = htonl (n->ats_count);
276 n->ats_count * sizeof (struct GNUNET_TRANSPORT_ATS_Information));
277 ats[n->ats_count].type = htonl (0);
278 ats[n->ats_count].value = htonl (0);
279 send_to_all_clients (&psnm->header, GNUNET_YES,
280 GNUNET_CORE_OPTION_SEND_STATUS_CHANGE);
281 GNUNET_STATISTICS_update (stats, gettext_noop ("# peer status changes"), 1,
288 * Go over our message queue and if it is not too long, go
289 * over the pending requests from clients for this
290 * neighbour and send some clients a 'READY' notification.
292 * @param n which peer to process
295 schedule_peer_messages (struct Neighbour *n)
297 struct SendMessageReady smr;
298 struct ClientActiveRequest *car;
299 struct ClientActiveRequest *pos;
301 struct MessageEntry *mqe;
302 unsigned int queue_size;
304 /* check if neighbour queue is empty enough! */
314 if (queue_size >= MAX_PEER_QUEUE_SIZE)
316 #if DEBUG_CORE_CLIENT
317 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
318 "Not considering client transmission requests: queue full\n");
320 return; /* queue still full */
322 /* find highest priority request */
323 pos = n->active_client_request_head;
327 if ((car == NULL) || (pos->priority > car->priority))
334 car = n->active_client_request_head;
337 return; /* no pending requests */
338 #if DEBUG_CORE_CLIENT
339 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
340 "Permitting client transmission request to `%s'\n",
341 GNUNET_i2s (&n->peer));
344 GNUNET_CONTAINER_DLL_remove (n->active_client_request_head,
345 n->active_client_request_tail, car);
346 GNUNET_assert (GNUNET_YES ==
347 GNUNET_CONTAINER_multihashmap_remove (c->requests,
350 smr.header.size = htons (sizeof (struct SendMessageReady));
351 smr.header.type = htons (GNUNET_MESSAGE_TYPE_CORE_SEND_READY);
352 smr.size = htons (car->msize);
353 smr.smr_id = car->smr_id;
355 send_to_client (c, &smr.header, GNUNET_NO);
362 * Free the given entry for the neighbour (it has
363 * already been removed from the list at this point).
365 * @param n neighbour to free
368 free_neighbour (struct Neighbour *n)
370 struct MessageEntry *m;
371 struct ClientActiveRequest *car;
374 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
375 "Destroying neighbour entry for peer `%4s'\n",
376 GNUNET_i2s (&n->peer));
380 GNUNET_free (n->skm);
383 while (NULL != (m = n->messages))
385 n->messages = m->next;
388 while (NULL != (m = n->encrypted_head))
390 GNUNET_CONTAINER_DLL_remove (n->encrypted_head, n->encrypted_tail, m);
393 while (NULL != (car = n->active_client_request_head))
395 GNUNET_CONTAINER_DLL_remove (n->active_client_request_head,
396 n->active_client_request_tail, car);
397 GNUNET_assert (GNUNET_YES ==
398 GNUNET_CONTAINER_multihashmap_remove (car->client->requests,
405 GNUNET_TRANSPORT_notify_transmit_ready_cancel (n->th);
408 if (n->retry_plaintext_task != GNUNET_SCHEDULER_NO_TASK)
409 GNUNET_SCHEDULER_cancel (n->retry_plaintext_task);
410 if (n->quota_update_task != GNUNET_SCHEDULER_NO_TASK)
411 GNUNET_SCHEDULER_cancel (n->quota_update_task);
412 if (n->dead_clean_task != GNUNET_SCHEDULER_NO_TASK)
413 GNUNET_SCHEDULER_cancel (n->keep_alive_task);
414 if (n->status == PEER_STATE_KEY_CONFIRMED)
415 GNUNET_STATISTICS_update (stats, gettext_noop ("# established sessions"),
417 GNUNET_array_grow (n->ats, n->ats_count, 0);
418 GNUNET_free_non_null (n->pending_ping);
419 GNUNET_free_non_null (n->pending_pong);
426 * Consider freeing the given neighbour since we may not need
427 * to keep it around anymore.
429 * @param n neighbour to consider discarding
432 consider_free_neighbour (struct Neighbour *n);
436 * Task triggered when a neighbour entry might have gotten stale.
438 * @param cls the 'struct Neighbour'
439 * @param tc scheduler context (not used)
442 consider_free_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
444 struct Neighbour *n = cls;
446 n->dead_clean_task = GNUNET_SCHEDULER_NO_TASK;
447 consider_free_neighbour (n);
452 * Consider freeing the given neighbour since we may not need
453 * to keep it around anymore.
455 * @param n neighbour to consider discarding
458 consider_free_neighbour (struct Neighbour *n)
460 struct GNUNET_TIME_Relative left;
462 if ((n->th != NULL) || (n->pitr != NULL) || (GNUNET_YES == n->is_connected))
463 return; /* no chance */
465 left = GNUNET_TIME_absolute_get_remaining (get_neighbour_timeout (n));
466 if (left.rel_value > 0)
468 if (n->dead_clean_task != GNUNET_SCHEDULER_NO_TASK)
469 GNUNET_SCHEDULER_cancel (n->dead_clean_task);
471 GNUNET_SCHEDULER_add_delayed (left, &consider_free_task, n);
474 /* actually free the neighbour... */
475 GNUNET_assert (GNUNET_YES ==
476 GNUNET_CONTAINER_multihashmap_remove (neighbours,
477 &n->peer.hashPubKey, n));
478 GNUNET_STATISTICS_set (stats, gettext_noop ("# neighbour entries allocated"),
479 GNUNET_CONTAINER_multihashmap_size (neighbours),
486 * Function called when the transport service is ready to
487 * receive an encrypted message for the respective peer
489 * @param cls neighbour to use message from
490 * @param size number of bytes we can transmit
491 * @param buf where to copy the message
492 * @return number of bytes transmitted
495 notify_encrypted_transmit_ready (void *cls, size_t size, void *buf)
497 struct Neighbour *n = cls;
498 struct MessageEntry *m;
503 m = n->encrypted_head;
507 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
508 "Encrypted message queue empty, no messages added to buffer for `%4s'\n",
509 GNUNET_i2s (&n->peer));
513 GNUNET_CONTAINER_DLL_remove (n->encrypted_head, n->encrypted_tail, m);
518 GNUNET_assert (size >= m->size);
519 memcpy (cbuf, &m[1], m->size);
521 GNUNET_BANDWIDTH_tracker_consume (&n->available_send_window, m->size);
523 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
524 "Copied message of type %u and size %u into transport buffer for `%4s'\n",
526 ntohs (((struct GNUNET_MessageHeader *) &m[1])->type),
527 (unsigned int) ret, GNUNET_i2s (&n->peer));
529 process_encrypted_neighbour_queue (n);
534 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
535 "Transmission of message of type %u and size %u failed\n",
537 ntohs (((struct GNUNET_MessageHeader *) &m[1])->type),
538 (unsigned int) m->size);
542 consider_free_neighbour (n);
543 GNUNET_STATISTICS_update (stats,
545 ("# encrypted bytes given to transport"), ret,
555 * Select messages for transmission. This heuristic uses a combination
556 * of earliest deadline first (EDF) scheduling (with bounded horizon)
557 * and priority-based discard (in case no feasible schedule exist) and
558 * speculative optimization (defer any kind of transmission until
559 * we either create a batch of significant size, 25% of max, or until
560 * we are close to a deadline). Furthermore, when scheduling the
561 * heuristic also packs as many messages into the batch as possible,
562 * starting with those with the earliest deadline. Yes, this is fun.
564 * @param n neighbour to select messages from
565 * @param size number of bytes to select for transmission
566 * @param retry_time set to the time when we should try again
567 * (only valid if this function returns zero)
568 * @return number of bytes selected, or 0 if we decided to
569 * defer scheduling overall; in that case, retry_time is set.
572 select_messages (struct Neighbour *n, size_t size,
573 struct GNUNET_TIME_Relative *retry_time)
575 struct MessageEntry *pos;
576 struct MessageEntry *min;
577 struct MessageEntry *last;
578 unsigned int min_prio;
579 struct GNUNET_TIME_Absolute t;
580 struct GNUNET_TIME_Absolute now;
581 struct GNUNET_TIME_Relative delta;
583 struct GNUNET_TIME_Relative slack; /* how long could we wait before missing deadlines? */
586 unsigned int queue_size;
587 int discard_low_prio;
589 GNUNET_assert (NULL != n->messages);
590 now = GNUNET_TIME_absolute_get ();
591 /* last entry in linked list of messages processed */
593 /* should we remove the entry with the lowest
594 * priority from consideration for scheduling at the
595 * end of the loop? */
605 discard_low_prio = GNUNET_YES;
606 while (GNUNET_YES == discard_low_prio)
610 discard_low_prio = GNUNET_NO;
611 /* calculate number of bytes available for transmission at time "t" */
612 avail = GNUNET_BANDWIDTH_tracker_get_available (&n->available_send_window);
614 /* how many bytes have we (hypothetically) scheduled so far */
616 /* maximum time we can wait before transmitting anything
617 * and still make all of our deadlines */
618 slack = GNUNET_TIME_UNIT_FOREVER_REL;
620 /* note that we use "*2" here because we want to look
621 * a bit further into the future; much more makes no
622 * sense since new message might be scheduled in the
624 while ((pos != NULL) && (off < size * 2))
626 if (pos->do_transmit == GNUNET_YES)
628 /* already removed from consideration */
632 if (discard_low_prio == GNUNET_NO)
634 delta = GNUNET_TIME_absolute_get_difference (t, pos->deadline);
635 if (delta.rel_value > 0)
637 // FIXME: HUH? Check!
640 GNUNET_BANDWIDTH_value_get_available_until (n->bw_out, delta);
642 if (avail < pos->size)
644 // FIXME: HUH? Check!
645 discard_low_prio = GNUNET_YES; /* we could not schedule this one! */
650 /* update slack, considering both its absolute deadline
651 * and relative deadlines caused by other messages
652 * with their respective load */
654 GNUNET_TIME_relative_min (slack,
655 GNUNET_BANDWIDTH_value_get_delay_for
657 if (pos->deadline.abs_value <= now.abs_value)
660 slack = GNUNET_TIME_UNIT_ZERO;
662 else if (GNUNET_YES == pos->got_slack)
664 /* should be soon now! */
666 GNUNET_TIME_relative_min (slack,
667 GNUNET_TIME_absolute_get_remaining
668 (pos->slack_deadline));
673 GNUNET_TIME_relative_min (slack,
674 GNUNET_TIME_absolute_get_difference
675 (now, pos->deadline));
676 pos->got_slack = GNUNET_YES;
677 pos->slack_deadline =
678 GNUNET_TIME_absolute_min (pos->deadline,
679 GNUNET_TIME_relative_to_absolute
680 (GNUNET_CONSTANTS_MAX_CORK_DELAY));
685 t = GNUNET_TIME_absolute_max (pos->deadline, t); // HUH? Check!
686 if (pos->priority <= min_prio)
688 /* update min for discard */
689 min_prio = pos->priority;
694 if (discard_low_prio)
696 GNUNET_assert (min != NULL);
697 /* remove lowest-priority entry from consideration */
698 min->do_transmit = GNUNET_YES; /* means: discard (for now) */
702 /* guard against sending "tiny" messages with large headers without
703 * urgent deadlines */
704 if ((slack.rel_value > GNUNET_CONSTANTS_MAX_CORK_DELAY.rel_value) &&
705 (size > 4 * off) && (queue_size <= MAX_PEER_QUEUE_SIZE - 2))
707 /* less than 25% of message would be filled with deadlines still
708 * being met if we delay by one second or more; so just wait for
709 * more data; but do not wait longer than 1s (since we don't want
710 * to delay messages for a really long time either). */
711 *retry_time = GNUNET_CONSTANTS_MAX_CORK_DELAY;
712 /* reset do_transmit values for next time */
715 pos->do_transmit = GNUNET_NO;
718 GNUNET_STATISTICS_update (stats,
720 ("# transmissions delayed due to corking"), 1,
723 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
724 "Deferring transmission for %llums due to underfull message buffer size (%u/%u)\n",
725 (unsigned long long) retry_time->rel_value, (unsigned int) off,
726 (unsigned int) size);
730 /* select marked messages (up to size) for transmission */
735 if ((pos->size <= size) && (pos->do_transmit == GNUNET_NO))
737 pos->do_transmit = GNUNET_YES; /* mark for transmission */
741 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
742 "Selecting message of size %u for transmission\n",
743 (unsigned int) pos->size);
749 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
750 "Not selecting message of size %u for transmission at this time (maximum is %u)\n",
751 (unsigned int) pos->size, size);
753 pos->do_transmit = GNUNET_NO; /* mark for not transmitting! */
758 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
759 "Selected %llu/%llu bytes of %u/%u plaintext messages for transmission to `%4s'.\n",
760 (unsigned long long) off, (unsigned long long) tsize, queue_size,
761 (unsigned int) MAX_PEER_QUEUE_SIZE, GNUNET_i2s (&n->peer));
768 * Batch multiple messages into a larger buffer.
770 * @param n neighbour to take messages from
771 * @param buf target buffer
772 * @param size size of buf
773 * @param deadline set to transmission deadline for the result
774 * @param retry_time set to the time when we should try again
775 * (only valid if this function returns zero)
776 * @param priority set to the priority of the batch
777 * @return number of bytes written to buf (can be zero)
780 batch_message (struct Neighbour *n, char *buf, size_t size,
781 struct GNUNET_TIME_Absolute *deadline,
782 struct GNUNET_TIME_Relative *retry_time, unsigned int *priority)
784 char ntmb[GNUNET_SERVER_MAX_MESSAGE_SIZE - 1];
785 struct NotifyTrafficMessage *ntm = (struct NotifyTrafficMessage *) ntmb;
786 struct MessageEntry *pos;
787 struct MessageEntry *prev;
788 struct MessageEntry *next;
793 *deadline = GNUNET_TIME_UNIT_FOREVER_ABS;
794 *retry_time = GNUNET_TIME_UNIT_FOREVER_REL;
795 if (0 == select_messages (n, size, retry_time))
798 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
799 "No messages selected, will try again in %llu ms\n",
800 retry_time->rel_value);
804 ntm->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_NOTIFY_OUTBOUND);
805 ntm->ats_count = htonl (0);
806 ntm->ats.type = htonl (0);
807 ntm->ats.value = htonl (0);
811 while ((pos != NULL) && (size >= sizeof (struct GNUNET_MessageHeader)))
814 if (GNUNET_YES == pos->do_transmit)
816 GNUNET_assert (pos->size <= size);
817 /* do notifications */
818 /* FIXME: track if we have *any* client that wants
819 * full notifications and only do this if that is
822 GNUNET_SERVER_MAX_MESSAGE_SIZE - sizeof (struct NotifyTrafficMessage))
824 memcpy (&ntm[1], &pos[1], pos->size);
826 htons (sizeof (struct NotifyTrafficMessage) +
827 sizeof (struct GNUNET_MessageHeader));
828 send_to_all_clients (&ntm->header, GNUNET_YES,
829 GNUNET_CORE_OPTION_SEND_HDR_OUTBOUND);
833 /* message too large for 'full' notifications, we do at
834 * least the 'hdr' type */
835 memcpy (&ntm[1], &pos[1], sizeof (struct GNUNET_MessageHeader));
838 htons (sizeof (struct NotifyTrafficMessage) + pos->size);
839 send_to_all_clients (&ntm->header, GNUNET_YES,
840 GNUNET_CORE_OPTION_SEND_FULL_OUTBOUND);
842 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
843 "Encrypting %u bytes with message of type %u and size %u\n",
846 ntohs (((const struct GNUNET_MessageHeader *) &pos[1])->type),
848 ntohs (((const struct GNUNET_MessageHeader *)
851 /* copy for encrypted transmission */
852 memcpy (&buf[ret], &pos[1], pos->size);
855 *priority += pos->priority;
857 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
858 "Adding plaintext message of size %u with deadline %llu ms to batch\n",
859 (unsigned int) pos->size,
861 GNUNET_TIME_absolute_get_remaining (pos->deadline).rel_value);
863 deadline->abs_value =
864 GNUNET_MIN (deadline->abs_value, pos->deadline.abs_value);
878 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
879 "Deadline for message batch is %llu ms\n",
880 GNUNET_TIME_absolute_get_remaining (*deadline).rel_value);
887 * Remove messages with deadlines that have long expired from
890 * @param n neighbour to inspect
893 discard_expired_messages (struct Neighbour *n)
895 struct MessageEntry *prev;
896 struct MessageEntry *next;
897 struct MessageEntry *pos;
898 struct GNUNET_TIME_Absolute now;
899 struct GNUNET_TIME_Relative delta;
901 unsigned int queue_length;
904 now = GNUNET_TIME_absolute_get ();
912 delta = GNUNET_TIME_absolute_get_difference (pos->deadline, now);
913 if (delta.rel_value > PAST_EXPIRATION_DISCARD_TIME.rel_value)
916 GNUNET_log (GNUNET_ERROR_TYPE_INFO,
917 "Message is %llu ms past due, discarding.\n",
924 GNUNET_STATISTICS_update (stats,
926 ("# messages discarded (expired prior to transmission)"),
935 if ( (GNUNET_YES == disc) &&
936 (queue_length == MAX_PEER_QUEUE_SIZE) )
937 schedule_peer_messages (n);
942 * Signature of the main function of a task.
945 * @param tc context information (why was this task triggered now)
948 retry_plaintext_processing (void *cls,
949 const struct GNUNET_SCHEDULER_TaskContext *tc)
951 struct Neighbour *n = cls;
953 n->retry_plaintext_task = GNUNET_SCHEDULER_NO_TASK;
954 process_plaintext_neighbour_queue (n);
959 * Check if we have plaintext messages for the specified neighbour
960 * pending, and if so, consider batching and encrypting them (and
961 * then trigger processing of the encrypted queue if needed).
963 * @param n neighbour to check.
966 process_plaintext_neighbour_queue (struct Neighbour *n)
968 char pbuf[GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE + sizeof (struct EncryptedMessage)]; /* plaintext */
970 struct EncryptedMessage *em; /* encrypted message */
971 struct EncryptedMessage *ph; /* plaintext header */
972 struct MessageEntry *me;
973 unsigned int priority;
974 struct GNUNET_TIME_Absolute deadline;
975 struct GNUNET_TIME_Relative retry_time;
976 struct GNUNET_CRYPTO_AesInitializationVector iv;
977 struct GNUNET_CRYPTO_AuthKey auth_key;
979 if (n->retry_plaintext_task != GNUNET_SCHEDULER_NO_TASK)
981 GNUNET_SCHEDULER_cancel (n->retry_plaintext_task);
982 n->retry_plaintext_task = GNUNET_SCHEDULER_NO_TASK;
986 case PEER_STATE_DOWN:
989 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
990 "Not yet connected to `%4s', deferring processing of plaintext messages.\n",
991 GNUNET_i2s (&n->peer));
994 case PEER_STATE_KEY_SENT:
995 if (n->retry_set_key_task == GNUNET_SCHEDULER_NO_TASK)
996 n->retry_set_key_task =
997 GNUNET_SCHEDULER_add_delayed (n->set_key_retry_frequency,
998 &set_key_retry_task, n);
1000 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1001 "Not yet connected to `%4s', deferring processing of plaintext messages.\n",
1002 GNUNET_i2s (&n->peer));
1005 case PEER_STATE_KEY_RECEIVED:
1006 if (n->retry_set_key_task == GNUNET_SCHEDULER_NO_TASK)
1007 n->retry_set_key_task =
1008 GNUNET_SCHEDULER_add_delayed (n->set_key_retry_frequency,
1009 &set_key_retry_task, n);
1011 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1012 "Not yet connected to `%4s', deferring processing of plaintext messages.\n",
1013 GNUNET_i2s (&n->peer));
1016 case PEER_STATE_KEY_CONFIRMED:
1017 /* ready to continue */
1020 discard_expired_messages (n);
1021 if (n->messages == NULL)
1024 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1025 "Plaintext message queue for `%4s' is empty.\n",
1026 GNUNET_i2s (&n->peer));
1028 return; /* no pending messages */
1030 if (n->encrypted_head != NULL)
1033 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1034 "Encrypted message queue for `%4s' is still full, delaying plaintext processing.\n",
1035 GNUNET_i2s (&n->peer));
1037 return; /* wait for messages already encrypted to be
1038 * processed first! */
1040 ph = (struct EncryptedMessage *) pbuf;
1041 deadline = GNUNET_TIME_UNIT_FOREVER_ABS;
1043 used = sizeof (struct EncryptedMessage);
1045 batch_message (n, &pbuf[used],
1046 GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE, &deadline,
1047 &retry_time, &priority);
1048 if (used == sizeof (struct EncryptedMessage))
1051 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1052 "No messages selected for transmission to `%4s' at this time, will try again later.\n",
1053 GNUNET_i2s (&n->peer));
1055 /* no messages selected for sending, try again later... */
1056 n->retry_plaintext_task =
1057 GNUNET_SCHEDULER_add_delayed (retry_time, &retry_plaintext_processing,
1061 GSC_KX_encrypt_and_transmit (n->kx,
1062 &pbuf[struct EncryptedMessage],
1063 used - sizeof (struct EncryptedMessage));
1064 schedule_peer_messages (n);
1071 * Check if we have encrypted messages for the specified neighbour
1072 * pending, and if so, check with the transport about sending them
1075 * @param n neighbour to check.
1078 process_encrypted_neighbour_queue (struct Neighbour *n)
1080 struct MessageEntry *m;
1083 return; /* request already pending */
1084 if (GNUNET_YES != n->is_connected)
1089 m = n->encrypted_head;
1092 /* encrypted queue empty, try plaintext instead */
1093 process_plaintext_neighbour_queue (n);
1097 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1098 "Asking transport for transmission of %u bytes to `%4s' in next %llu ms\n",
1099 (unsigned int) m->size, GNUNET_i2s (&n->peer),
1100 (unsigned long long)
1101 GNUNET_TIME_absolute_get_remaining (m->deadline).rel_value);
1104 GNUNET_TRANSPORT_notify_transmit_ready (transport, &n->peer, m->size,
1106 GNUNET_TIME_absolute_get_remaining
1108 ¬ify_encrypted_transmit_ready,
1112 /* message request too large or duplicate request */
1114 /* discard encrypted message */
1115 GNUNET_CONTAINER_DLL_remove (n->encrypted_head, n->encrypted_tail, m);
1117 process_encrypted_neighbour_queue (n);
1123 * Initialize a new 'struct Neighbour'.
1125 * @param pid ID of the new neighbour
1126 * @return handle for the new neighbour
1128 static struct Neighbour *
1129 create_neighbour (const struct GNUNET_PeerIdentity *pid)
1131 struct Neighbour *n;
1132 struct GNUNET_TIME_Absolute now;
1135 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1136 "Creating neighbour entry for peer `%4s'\n", GNUNET_i2s (pid));
1138 n = GNUNET_malloc (sizeof (struct Neighbour));
1140 GNUNET_CRYPTO_aes_create_session_key (&n->encrypt_key);
1141 now = GNUNET_TIME_absolute_get ();
1142 n->encrypt_key_created = now;
1143 n->last_activity = now;
1144 n->set_key_retry_frequency = INITIAL_SET_KEY_RETRY_FREQUENCY;
1145 n->bw_in = GNUNET_CONSTANTS_DEFAULT_BW_IN_OUT;
1146 n->bw_out = GNUNET_CONSTANTS_DEFAULT_BW_IN_OUT;
1147 n->bw_out_internal_limit = GNUNET_BANDWIDTH_value_init (UINT32_MAX);
1148 n->bw_out_external_limit = GNUNET_CONSTANTS_DEFAULT_BW_IN_OUT;
1150 GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_NONCE, UINT32_MAX);
1151 GNUNET_assert (GNUNET_OK ==
1152 GNUNET_CONTAINER_multihashmap_put (neighbours,
1153 &n->peer.hashPubKey, n,
1154 GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
1155 GNUNET_STATISTICS_set (stats, gettext_noop ("# neighbour entries allocated"),
1156 GNUNET_CONTAINER_multihashmap_size (neighbours),
1158 neighbour_quota_update (n, NULL);
1159 consider_free_neighbour (n);
1166 * We have a new client, notify it about all current sessions.
1168 * @param client the new client
1171 GSC_SESSIONS_notify_client_about_sessions (struct GSC_Client *client)
1173 /* notify new client about existing neighbours */
1174 GNUNET_CONTAINER_multihashmap_iterate (neighbours,
1175 ¬ify_client_about_neighbour, client);
1180 * Queue a request from a client for transmission to a particular peer.
1182 * @param car request to queue; this handle is then shared between
1183 * the caller (CLIENTS subsystem) and SESSIONS and must not
1184 * be released by either until either 'GNUNET_SESSIONS_dequeue',
1185 * 'GNUNET_SESSIONS_transmit' or 'GNUNET_CLIENTS_failed'
1186 * have been invoked on it
1189 GSC_SESSIONS_queue_request (struct GSC_ClientActiveRequest *car)
1191 struct Neighbour *n; // FIXME: session...
1193 n = find_neighbour (&car->peer);
1194 if ((n == NULL) || (GNUNET_YES != n->is_connected) ||
1195 (n->status != PEER_STATE_KEY_CONFIRMED))
1197 /* neighbour must have disconnected since request was issued,
1198 * ignore (client will realize it once it processes the
1199 * disconnect notification) */
1200 #if DEBUG_CORE_CLIENT
1201 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1202 "Dropped client request for transmission (am disconnected)\n");
1204 GNUNET_STATISTICS_update (stats,
1206 ("# send requests dropped (disconnected)"), 1,
1208 GSC_CLIENTS_reject_requests (car);
1211 #if DEBUG_CORE_CLIENT
1212 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1213 "Received client transmission request. queueing\n");
1215 GNUNET_CONTAINER_DLL_insert (n->active_client_request_head,
1216 n->active_client_request_tail, car);
1218 // schedule_peer_messages (n);
1223 * Dequeue a request from a client from transmission to a particular peer.
1225 * @param car request to dequeue; this handle will then be 'owned' by
1226 * the caller (CLIENTS sysbsystem)
1229 GSC_SESSIONS_dequeue_request (struct GSC_ClientActiveRequest *car)
1233 s = find_session (&car->peer);
1234 GNUNET_CONTAINER_DLL_remove (s->active_client_request_head,
1235 s->active_client_request_tail, car);
1241 * Transmit a message to a particular peer.
1243 * @param car original request that was queued and then solicited;
1244 * this handle will now be 'owned' by the SESSIONS subsystem
1245 * @param msg message to transmit
1248 GSC_SESSIONS_transmit (struct GSC_ClientActiveRequest *car,
1249 const struct GNUNET_MessageHeader *msg)
1251 struct MessageEntry *prev;
1252 struct MessageEntry *pos;
1253 struct MessageEntry *e;
1254 struct MessageEntry *min_prio_entry;
1255 struct MessageEntry *min_prio_prev;
1256 unsigned int min_prio;
1257 unsigned int queue_size;
1259 n = find_neighbour (&sm->peer);
1260 if ((n == NULL) || (GNUNET_YES != n->is_connected) ||
1261 (n->status != PEER_STATE_KEY_CONFIRMED))
1263 /* attempt to send message to peer that is not connected anymore
1264 * (can happen due to asynchrony) */
1265 GNUNET_STATISTICS_update (stats,
1267 ("# messages discarded (disconnected)"), 1,
1270 GNUNET_SERVER_receive_done (client, GNUNET_OK);
1274 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1275 "Core received `%s' request, queueing %u bytes of plaintext data for transmission to `%4s'.\n",
1276 "SEND", (unsigned int) msize, GNUNET_i2s (&sm->peer));
1278 discard_expired_messages (n);
1279 /* bound queue size */
1280 /* NOTE: this entire block to bound the queue size should be
1281 * obsolete with the new client-request code and the
1282 * 'schedule_peer_messages' mechanism; we still have this code in
1283 * here for now as a sanity check for the new mechanmism;
1284 * ultimately, we should probably simply reject SEND messages that
1285 * are not 'approved' (or provide a new core API for very unreliable
1286 * delivery that always sends with priority 0). Food for thought. */
1287 min_prio = UINT32_MAX;
1288 min_prio_entry = NULL;
1289 min_prio_prev = NULL;
1295 if (pos->priority <= min_prio)
1297 min_prio_entry = pos;
1298 min_prio_prev = prev;
1299 min_prio = pos->priority;
1305 if (queue_size >= MAX_PEER_QUEUE_SIZE)
1308 if (ntohl (sm->priority) <= min_prio)
1310 /* discard new entry; this should no longer happen! */
1313 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1314 "Queue full (%u/%u), discarding new request (%u bytes of type %u)\n",
1315 queue_size, (unsigned int) MAX_PEER_QUEUE_SIZE,
1316 (unsigned int) msize, (unsigned int) ntohs (message->type));
1318 GNUNET_STATISTICS_update (stats,
1319 gettext_noop ("# discarded CORE_SEND requests"),
1323 GNUNET_SERVER_receive_done (client, GNUNET_OK);
1326 GNUNET_assert (min_prio_entry != NULL);
1327 /* discard "min_prio_entry" */
1329 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1330 "Queue full, discarding existing older request\n");
1332 GNUNET_STATISTICS_update (stats,
1334 ("# discarded lower priority CORE_SEND requests"),
1336 if (min_prio_prev == NULL)
1337 n->messages = min_prio_entry->next;
1339 min_prio_prev->next = min_prio_entry->next;
1340 GNUNET_free (min_prio_entry);
1344 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1345 "Adding transmission request for `%4s' of size %u to queue\n",
1346 GNUNET_i2s (&sm->peer), (unsigned int) msize);
1348 e = GNUNET_malloc (sizeof (struct MessageEntry) + msize);
1349 e->deadline = GNUNET_TIME_absolute_ntoh (sm->deadline);
1350 e->priority = ntohl (sm->priority);
1352 if (GNUNET_YES != (int) ntohl (sm->cork))
1353 e->got_slack = GNUNET_YES;
1354 memcpy (&e[1], &sm[1], msize);
1356 /* insert, keep list sorted by deadline */
1359 while ((pos != NULL) && (pos->deadline.abs_value < e->deadline.abs_value))
1370 /* consider scheduling now */
1371 process_plaintext_neighbour_queue (n);
1378 GSC_NEIGHBOURS_init ()
1380 neighbours = GNUNET_CONTAINER_multihashmap_create (128);
1381 self.public_key = &my_public_key;
1382 self.peer = my_identity;
1383 self.last_activity = GNUNET_TIME_UNIT_FOREVER_ABS;
1384 self.status = PEER_STATE_KEY_CONFIRMED;
1385 self.is_connected = GNUNET_YES;
1391 GSC_NEIGHBOURS_done ()
1393 GNUNET_CONTAINER_multihashmap_iterate (neighbours, &free_neighbour_helper,
1395 GNUNET_CONTAINER_multihashmap_destroy (neighbours);
1397 GNUNET_STATISTICS_set (stats, gettext_noop ("# neighbour entries allocated"),