*/
/**
- * @file core/gnunet-service-core_neighbours.c
+ * @file core/gnunet-service-core_sessions.c
* @brief code for managing of 'encrypted' sessions (key exchange done)
* @author Christian Grothoff
*/
* Record kept for each request for transmission issued by a
* client that is still pending.
*/
-struct ClientActiveRequest;
+struct GSC_ClientActiveRequest;
/**
* Data kept per session.
* Head of list of requests from clients for transmission to
* this peer.
*/
- struct ClientActiveRequest *active_client_request_head;
+ struct GSC_ClientActiveRequest *active_client_request_head;
/**
* Tail of list of requests from clients for transmission to
* this peer.
*/
- struct ClientActiveRequest *active_client_request_tail;
+ struct GSC_ClientActiveRequest *active_client_request_tail;
/**
* Performance data for the peer.
*/
struct GSC_KeyExchangeInfo *kxinfo;
-
/**
* ID of task used for cleaning up dead neighbour entries.
*/
*/
unsigned int ats_count;
- /**
- * Bit map indicating which of the 32 sequence numbers before the last
- * were received (good for accepting out-of-order packets and
- * estimating reliability of the connection)
- */
- unsigned int last_packets_bitmap;
-
- /**
- * last sequence number received on this connection (highest)
- */
- uint32_t last_sequence_number_received;
-
- /**
- * last sequence number transmitted
- */
- uint32_t last_sequence_number_sent;
-
/**
* Available bandwidth in for this peer (current target).
*/
static void
schedule_peer_messages (struct Neighbour *n)
{
- struct SendMessageReady smr;
- struct ClientActiveRequest *car;
- struct ClientActiveRequest *pos;
+ struct GSC_ClientActiveRequest *car;
+ struct GSC_ClientActiveRequest *pos;
struct Client *c;
struct MessageEntry *mqe;
unsigned int queue_size;
free_neighbour (struct Neighbour *n)
{
struct MessageEntry *m;
- struct ClientActiveRequest *car;
+ struct GSC_ClientActiveRequest *car;
#if DEBUG_CORE
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
#endif
n = GNUNET_malloc (sizeof (struct Neighbour));
n->peer = *pid;
- GNUNET_CRYPTO_aes_create_session_key (&n->encrypt_key);
- now = GNUNET_TIME_absolute_get ();
- n->encrypt_key_created = now;
- n->last_activity = now;
- n->set_key_retry_frequency = INITIAL_SET_KEY_RETRY_FREQUENCY;
+ n->last_activity = GNUNET_TIME_absolute_get ();
n->bw_in = GNUNET_CONSTANTS_DEFAULT_BW_IN_OUT;
n->bw_out = GNUNET_CONSTANTS_DEFAULT_BW_IN_OUT;
n->bw_out_internal_limit = GNUNET_BANDWIDTH_value_init (UINT32_MAX);
+/**
+ * Send a message to the neighbour.
+ *
+ * @param cls the message
+ * @param key neighbour's identity
+ * @param value 'struct Neighbour' of the target
+ * @return always GNUNET_OK
+ */
+static int
+do_send_message (void *cls, const GNUNET_HashCode * key, void *value)
+{
+ struct GNUNET_MessageHeader *hdr = cls;
+ struct Neighbour *n = value;
+ struct MessageEntry *m;
+ uint16_t size;
+
+ size = ntohs (hdr->size);
+ m = GNUNET_malloc (sizeof (struct MessageEntry) + size);
+ memcpy (&m[1], hdr, size);
+ m->deadline = GNUNET_TIME_UNIT_FOREVER_ABS;
+ m->slack_deadline = GNUNET_TIME_UNIT_FOREVER_ABS;
+ m->priority = UINT_MAX;
+ m->sender_status = n->status;
+ m->size = size;
+ GNUNET_CONTAINER_DLL_insert (n->message_head,
+ n->message_tail,
+ m);
+ return GNUNET_OK;
+}
+
+
+/**
+ * Broadcast a message to all neighbours.
+ *
+ * @param msg message to transmit
+ */
+void
+GSC_SESSIONS_broadcast (const struct GNUNET_MessageHeader *msg)
+{
+ if (NULL == sessions)
+ return;
+ GNUNET_CONTAINER_multihashmap_iterate (sessions,
+ &do_send_message, msg);
+}
+
+
+/**
+ * Helper function for GSC_SESSIONS_handle_client_iterate_peers.
+ *
+ * @param cls the 'struct GNUNET_SERVER_TransmitContext' to queue replies
+ * @param key identity of the connected peer
+ * @param value the 'struct Neighbour' for the peer
+ * @return GNUNET_OK (continue to iterate)
+ */
+static int
+queue_connect_message (void *cls, const GNUNET_HashCode * key, void *value)
+{
+ struct GNUNET_SERVER_TransmitContext *tc = cls;
+ struct Neighbour *n = value;
+ char buf[GNUNET_SERVER_MAX_MESSAGE_SIZE - 1];
+ struct GNUNET_TRANSPORT_ATS_Information *ats;
+ size_t size;
+ struct ConnectNotifyMessage *cnm;
+
+ cnm = (struct ConnectNotifyMessage *) buf;
+ if (n->status != PEER_STATE_KEY_CONFIRMED)
+ return GNUNET_OK;
+ size =
+ sizeof (struct ConnectNotifyMessage) +
+ (n->ats_count) * sizeof (struct GNUNET_TRANSPORT_ATS_Information);
+ if (size >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
+ {
+ GNUNET_break (0);
+ /* recovery strategy: throw away performance data */
+ GNUNET_array_grow (n->ats, n->ats_count, 0);
+ size =
+ sizeof (struct PeerStatusNotifyMessage) +
+ n->ats_count * sizeof (struct GNUNET_TRANSPORT_ATS_Information);
+ }
+ cnm = (struct ConnectNotifyMessage *) buf;
+ cnm->header.size = htons (size);
+ cnm->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_NOTIFY_CONNECT);
+ cnm->ats_count = htonl (n->ats_count);
+ ats = &cnm->ats;
+ memcpy (ats, n->ats,
+ n->ats_count * sizeof (struct GNUNET_TRANSPORT_ATS_Information));
+ ats[n->ats_count].type = htonl (GNUNET_TRANSPORT_ATS_ARRAY_TERMINATOR);
+ ats[n->ats_count].value = htonl (0);
+#if DEBUG_CORE_CLIENT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending `%s' message to client.\n",
+ "NOTIFY_CONNECT");
+#endif
+ cnm->peer = n->peer;
+ GNUNET_SERVER_transmit_context_append_message (tc, &cnm->header);
+ return GNUNET_OK;
+}
+
+
+/**
+ * End the session with the given peer (we are no longer
+ * connected).
+ *
+ * @param pid identity of peer to kill session with
+ */
+void
+GSC_SESSIONS_end (const struct GNUNET_PeerIdentity *pid)
+{
+}
+
+
+/**
+ * Traffic is being solicited for the given peer. This means that the
+ * message queue on the transport-level (NEIGHBOURS subsystem) is now
+ * empty and it is now OK to transmit another (non-control) message.
+ *
+ * @param pid identity of peer ready to receive data
+ */
+void
+GSC_SESSIONS_solicit (const struct GNUNET_PeerIdentity *pid)
+{
+}
+
+
+/**
+ * Transmit a message to a particular peer.
+ *
+ * @param car original request that was queued and then solicited,
+ * ownership does not change (dequeue will be called soon).
+ * @param msg message to transmit
+ */
+void
+GSC_SESSIONS_transmit (struct GSC_ClientActiveRequest *car,
+ const struct GNUNET_MessageHeader *msg)
+{
+}
+
+
+/**
+ * We have a new client, notify it about all current sessions.
+ *
+ * @param client the new client
+ */
+void
+GSC_SESSIONS_notify_client_about_sessions (struct GSC_Client *client)
+{
+}
+
+
+/**
+ * Handle CORE_ITERATE_PEERS request. For this request type, the client
+ * does not have to have transmitted an INIT request. All current peers
+ * are returned, regardless of which message types they accept.
+ *
+ * @param cls unused
+ * @param client client sending the iteration request
+ * @param message iteration request message
+ */
+void
+GSC_SESSIONS_handle_client_iterate_peers (void *cls, struct GNUNET_SERVER_Client *client,
+ const struct GNUNET_MessageHeader *message)
+{
+ struct GNUNET_MessageHeader done_msg;
+ struct GNUNET_SERVER_TransmitContext *tc;
+
+ tc = GNUNET_SERVER_transmit_context_create (client);
+ GNUNET_CONTAINER_multihashmap_iterate (neighbours, &queue_connect_message,
+ tc);
+ done_msg.size = htons (sizeof (struct GNUNET_MessageHeader));
+ done_msg.type = htons (GNUNET_MESSAGE_TYPE_CORE_ITERATE_PEERS_END);
+ GNUNET_SERVER_transmit_context_append_message (tc, &done_msg);
+ GNUNET_SERVER_transmit_context_run (tc, GNUNET_TIME_UNIT_FOREVER_REL);
+}
+
+
+/**
+ * Handle CORE_PEER_CONNECTED request. Notify client about connection
+ * to the given neighbour. For this request type, the client does not
+ * have to have transmitted an INIT request. All current peers are
+ * returned, regardless of which message types they accept.
+ *
+ * @param cls unused
+ * @param client client sending the iteration request
+ * @param message iteration request message
+ */
+void
+GSC_SESSIONS_handle_client_have_peer (void *cls, struct GNUNET_SERVER_Client *client,
+ const struct GNUNET_MessageHeader *message)
+{
+ struct GNUNET_MessageHeader done_msg;
+ struct GNUNET_SERVER_TransmitContext *tc;
+ const struct GNUNET_PeerIdentity *peer;
+
+ peer = (const struct GNUNET_PeerIdentity *) &message[1]; // YUCK!
+ tc = GNUNET_SERVER_transmit_context_create (client);
+ GNUNET_CONTAINER_multihashmap_get_multiple (neighbours, &peer->hashPubKey,
+ &queue_connect_message, tc);
+ done_msg.size = htons (sizeof (struct GNUNET_MessageHeader));
+ done_msg.type = htons (GNUNET_MESSAGE_TYPE_CORE_ITERATE_PEERS_END);
+ GNUNET_SERVER_transmit_context_append_message (tc, &done_msg);
+ GNUNET_SERVER_transmit_context_run (tc, GNUNET_TIME_UNIT_FOREVER_REL);
+}
+
+
+
+/**
+ * Handle REQUEST_INFO request. For this request type, the client must
+ * have transmitted an INIT first.
+ *
+ * @param cls unused
+ * @param client client sending the request
+ * @param message iteration request message
+ */
+void
+GSC_SESSIONS_handle_client_request_info (void *cls, struct GNUNET_SERVER_Client *client,
+ const struct GNUNET_MessageHeader *message)
+{
+ const struct RequestInfoMessage *rcm;
+ struct GSC_Client *pos;
+ struct Neighbour *n;
+ struct ConfigurationInfoMessage cim;
+ int32_t want_reserv;
+ int32_t got_reserv;
+ unsigned long long old_preference;
+ struct GNUNET_TIME_Relative rdelay;
+
+ rdelay = GNUNET_TIME_relative_get_zero ();
+#if DEBUG_CORE_CLIENT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Core service receives `%s' request.\n",
+ "REQUEST_INFO");
+#endif
+ rcm = (const struct RequestInfoMessage *) message;
+ n = find_neighbour (&rcm->peer);
+ memset (&cim, 0, sizeof (cim));
+ if ((n != NULL) && (GNUNET_YES == n->is_connected))
+ {
+ want_reserv = ntohl (rcm->reserve_inbound);
+ if (n->bw_out_internal_limit.value__ != rcm->limit_outbound.value__)
+ {
+ n->bw_out_internal_limit = rcm->limit_outbound;
+ if (n->bw_out.value__ !=
+ GNUNET_BANDWIDTH_value_min (n->bw_out_internal_limit,
+ n->bw_out_external_limit).value__)
+ {
+ n->bw_out =
+ GNUNET_BANDWIDTH_value_min (n->bw_out_internal_limit,
+ n->bw_out_external_limit);
+ GNUNET_BANDWIDTH_tracker_update_quota (&n->available_recv_window,
+ n->bw_out);
+ GNUNET_TRANSPORT_set_quota (transport, &n->peer, n->bw_in, n->bw_out);
+ handle_peer_status_change (n);
+ }
+ }
+ if (want_reserv < 0)
+ {
+ got_reserv = want_reserv;
+ }
+ else if (want_reserv > 0)
+ {
+ rdelay =
+ GNUNET_BANDWIDTH_tracker_get_delay (&n->available_recv_window,
+ want_reserv);
+ if (rdelay.rel_value == 0)
+ got_reserv = want_reserv;
+ else
+ got_reserv = 0; /* all or nothing */
+ }
+ else
+ got_reserv = 0;
+ GNUNET_BANDWIDTH_tracker_consume (&n->available_recv_window, got_reserv);
+ old_preference = n->current_preference;
+ n->current_preference += GNUNET_ntohll (rcm->preference_change);
+ if (old_preference > n->current_preference)
+ {
+ /* overflow; cap at maximum value */
+ n->current_preference = ULLONG_MAX;
+ }
+ update_preference_sum (n->current_preference - old_preference);
+#if DEBUG_CORE_QUOTA
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Received reservation request for %d bytes for peer `%4s', reserved %d bytes, suggesting delay of %llu ms\n",
+ (int) want_reserv, GNUNET_i2s (&rcm->peer), (int) got_reserv,
+ (unsigned long long) rdelay.rel_value);
+#endif
+ cim.reserved_amount = htonl (got_reserv);
+ cim.reserve_delay = GNUNET_TIME_relative_hton (rdelay);
+ cim.bw_out = n->bw_out;
+ cim.preference = n->current_preference;
+ }
+ else
+ {
+ /* Technically, this COULD happen (due to asynchronous behavior),
+ * but it should be rare, so we should generate an info event
+ * to help diagnosis of serious errors that might be masked by this */
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ _
+ ("Client asked for preference change with peer `%s', which is not connected!\n"),
+ GNUNET_i2s (&rcm->peer));
+ GNUNET_SERVER_receive_done (client, GNUNET_OK);
+ return;
+ }
+ cim.header.size = htons (sizeof (struct ConfigurationInfoMessage));
+ cim.header.type = htons (GNUNET_MESSAGE_TYPE_CORE_CONFIGURATION_INFO);
+ cim.peer = rcm->peer;
+ cim.rim_id = rcm->rim_id;
+#if DEBUG_CORE_CLIENT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending `%s' message to client.\n",
+ "CONFIGURATION_INFO");
+#endif
+ GSC_CLIENTS_send_to_client (client, &cim.header, GNUNET_NO);
+ GNUNET_SERVER_receive_done (client, GNUNET_OK);
+}
+
+
+/**
+ * Create a session, a key exchange was just completed.
+ *
+ * @param peer peer that is now connected
+ * @param kx key exchange that completed
+ */
+void
+GSC_SESSIONS_create (const struct GNUNET_PeerIdentity *peer,
+ struct GSC_KeyExchangeInfo *kx)
+{
+ {
+ struct GNUNET_MessageHeader *hdr;
+
+ hdr = compute_type_map_message ();
+ send_type_map_to_neighbour (hdr, &n->peer.hashPubKey, n);
+ GNUNET_free (hdr);
+ }
+ 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);
+ }
+#if DEBUG_CORE
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Confirmed key via `%s' message for peer `%4s'\n", "PONG",
+ GNUNET_i2s (&n->peer));
+#endif
+
+
+ size =
+ sizeof (struct ConnectNotifyMessage) +
+ (n->ats_count) * sizeof (struct GNUNET_TRANSPORT_ATS_Information);
+ if (size >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
+ {
+ GNUNET_break (0);
+ /* recovery strategy: throw away performance data */
+ GNUNET_array_grow (n->ats, n->ats_count, 0);
+ size =
+ sizeof (struct PeerStatusNotifyMessage) +
+ n->ats_count * sizeof (struct GNUNET_TRANSPORT_ATS_Information);
+ }
+ cnm = (struct ConnectNotifyMessage *) buf;
+ cnm->header.size = htons (size);
+ cnm->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_NOTIFY_CONNECT);
+ cnm->ats_count = htonl (n->ats_count);
+ cnm->peer = n->peer;
+ mats = &cnm->ats;
+ memcpy (mats, n->ats,
+ n->ats_count * sizeof (struct GNUNET_TRANSPORT_ATS_Information));
+ mats[n->ats_count].type = htonl (GNUNET_TRANSPORT_ATS_ARRAY_TERMINATOR);
+ mats[n->ats_count].value = htonl (0);
+ send_to_all_clients (&cnm->header, GNUNET_NO,
+ GNUNET_CORE_OPTION_SEND_CONNECT);
+ process_encrypted_neighbour_queue (n);
+ n->last_activity = GNUNET_TIME_absolute_get ();
+
+ if (n->status == PEER_STATE_KEY_CONFIRMED)
+ {
+ now = GNUNET_TIME_absolute_get ();
+ n->last_activity = now;
+ changed = GNUNET_YES;
+ if (!up)
+ {
+ GNUNET_STATISTICS_update (stats, gettext_noop ("# established sessions"),
+ 1, GNUNET_NO);
+ n->time_established = now;
+ }
+ if (n->keep_alive_task != GNUNET_SCHEDULER_NO_TASK)
+ GNUNET_SCHEDULER_cancel (n->keep_alive_task);
+ n->keep_alive_task =
+ GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_divide
+ (GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT,
+ 2), &send_keep_alive, n);
+ }
+
+
+}
+
+
+/**
+ * Update information about a session.
+ *
+ * @param peer peer who's session should be updated
+ * @param bw_out new outbound bandwidth limit for the peer
+ * @param atsi performance information
+ * @param atsi_count number of performance records supplied
+ */
+void
+GSC_SESSIONS_update (const struct GNUNET_PeerIdentity *peer,
+ struct GNUNET_BANDWIDTH_Value32NBO bw_out,
+ const struct GNUNET_TRANSPORT_ATS_Information *atsi,
+ uint32_t atsi_count)
+{
+ if (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);
+ }
+
+}
+
+
+/**
+ * Initialize sessions subsystem.
+ */
int
-GSC_NEIGHBOURS_init ()
+GSC_SESSIONS_init ()
{
neighbours = GNUNET_CONTAINER_multihashmap_create (128);
self.public_key = &my_public_key;
}
+/**
+ * Shutdown sessions subsystem.
+ */
void
-GSC_NEIGHBOURS_done ()
+GSC_SESSIONS_done ()
{
GNUNET_CONTAINER_multihashmap_iterate (neighbours, &free_neighbour_helper,
NULL);
projects / oweals / gnunet.git / blobdiff