2 This file is part of GNUnet.
3 Copyright (C) 2010-2015 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.
21 * @file ats/ats_api_scheduling.c
22 * @brief automatic transport selection and outbound bandwidth determination
23 * @author Christian Grothoff
24 * @author Matthias Wachs
27 * - we could avoid a linear scan over the
28 * active addresses in some cases, so if
29 * there is need, we can still optimize here
30 * - we might want to split off the logic to
31 * determine LAN vs. WAN, as it has nothing
32 * to do with accessing the ATS service.
35 #include "gnunet_ats_service.h"
39 * How frequently do we scan the interfaces for changes to the addresses?
41 #define INTERFACE_PROCESSING_INTERVAL GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 2)
43 #define LOG(kind,...) GNUNET_log_from(kind, "ats-scheduling-api", __VA_ARGS__)
46 * Session ID we use if there is no session / slot.
52 * Information we track per address, incoming or outgoing. It also
53 * doesn't matter if we have a session, any address that ATS is
54 * allowed to suggest right now should be tracked.
56 struct GNUNET_ATS_AddressRecord
60 * Scheduling handle this address record belongs to.
62 struct GNUNET_ATS_SchedulingHandle *sh;
67 struct GNUNET_HELLO_Address *address;
70 * Session handle. NULL if we have an address but no
71 * active session for this address.
73 struct Session *session;
76 * Array with performance data about the address.
78 struct GNUNET_ATS_Information *ats;
81 * Number of entries in @e ats.
86 * Which slot (index) in the session array does
87 * this record correspond to? FIXME:
88 * FIXME: a linear search on this is really crappy!
89 * Maybe switch to a 64-bit global counter and be
90 * done with it? Or does that then cause too much
91 * trouble on the ATS-service side?
96 * We're about to destroy this address record, just ATS does
97 * not know this yet. Once ATS confirms its destruction,
105 * Handle to the ATS subsystem for bandwidth/transport scheduling information.
107 struct GNUNET_ATS_SchedulingHandle
113 const struct GNUNET_CONFIGURATION_Handle *cfg;
116 * Callback to invoke on suggestions.
118 GNUNET_ATS_AddressSuggestionCallback suggest_cb;
121 * Closure for @e suggest_cb.
123 void *suggest_cb_cls;
126 * Connection to ATS service.
128 struct GNUNET_CLIENT_Connection *client;
131 * Message queue for sending requests to the ATS service.
133 struct GNUNET_MQ_Handle *mq;
136 * Array of session objects (we need to translate them to numbers and back
137 * for the protocol; the offset in the array is the session number on the
138 * network). Index 0 is always NULL and reserved to represent the NULL pointer.
139 * Unused entries are also NULL.
141 struct GNUNET_ATS_AddressRecord **session_array;
144 * Task to trigger reconnect.
146 struct GNUNET_SCHEDULER_Task *task;
149 * Reconnect backoff delay.
151 struct GNUNET_TIME_Relative backoff;
154 * Size of the @e session_array.
156 unsigned int session_array_size;
162 * Re-establish the connection to the ATS service.
164 * @param sh handle to use to re-connect.
167 reconnect (struct GNUNET_ATS_SchedulingHandle *sh);
171 * Re-establish the connection to the ATS service.
173 * @param cls handle to use to re-connect.
174 * @param tc scheduler context
177 reconnect_task (void *cls,
178 const struct GNUNET_SCHEDULER_TaskContext *tc)
180 struct GNUNET_ATS_SchedulingHandle *sh = cls;
188 * Disconnect from ATS and then reconnect.
190 * @param sh our handle
193 force_reconnect (struct GNUNET_ATS_SchedulingHandle *sh)
197 GNUNET_MQ_destroy (sh->mq);
200 if (NULL != sh->client)
202 GNUNET_CLIENT_disconnect (sh->client);
205 sh->suggest_cb (sh->suggest_cb_cls,
207 GNUNET_BANDWIDTH_ZERO,
208 GNUNET_BANDWIDTH_ZERO);
209 sh->backoff = GNUNET_TIME_STD_BACKOFF (sh->backoff);
210 sh->task = GNUNET_SCHEDULER_add_delayed (sh->backoff,
217 * Find the session object corresponding to the given session ID.
219 * @param sh our handle
220 * @param session_id current session ID
221 * @param peer peer the session belongs to
222 * @return the session object (or NULL)
224 static struct GNUNET_ATS_AddressRecord *
225 find_session (struct GNUNET_ATS_SchedulingHandle *sh,
227 const struct GNUNET_PeerIdentity *peer)
229 struct GNUNET_ATS_AddressRecord *ar;
231 if (session_id >= sh->session_array_size)
238 ar = sh->session_array[session_id];
244 if (NULL == ar->address)
246 /* address was destroyed in the meantime, this can happen
247 as we communicate asynchronously with the ATS service. */
250 if (0 != memcmp (peer,
252 sizeof (struct GNUNET_PeerIdentity)))
255 force_reconnect (sh);
263 * Get an available session ID.
265 * @param sh our handle
266 * @return an unused slot, but never NOT_FOUND (0)
269 find_empty_session_slot (struct GNUNET_ATS_SchedulingHandle *sh)
275 while ( ( (NOT_FOUND == off) ||
276 (NULL != sh->session_array[off % sh->session_array_size]) ) &&
277 (i < sh->session_array_size) )
282 if ( (NOT_FOUND != off % sh->session_array_size) &&
283 (NULL == sh->session_array[off % sh->session_array_size]) )
285 i = sh->session_array_size;
286 GNUNET_array_grow (sh->session_array,
287 sh->session_array_size,
288 sh->session_array_size * 2);
294 * Get the ID for the given session object.
296 * @param sh our handle
297 * @param session session object
298 * @param address the address we are looking for
299 * @return the session id or NOT_FOUND for error
302 find_session_id (struct GNUNET_ATS_SchedulingHandle *sh,
303 struct Session *session,
304 const struct GNUNET_HELLO_Address *address)
313 for (i = 1; i < sh->session_array_size; i++)
314 if ( (NULL != sh->session_array[i]) &&
315 ( (session == sh->session_array[i]->session) ||
316 (NULL == sh->session_array[i]->session) ) &&
317 (0 == GNUNET_HELLO_address_cmp (address,
318 sh->session_array[i]->address)) )
325 * Release the session slot from the session table (ATS service is
326 * also done using it).
328 * @param sh our handle
329 * @param session_id identifies session that is no longer valid
332 release_session (struct GNUNET_ATS_SchedulingHandle *sh,
335 struct GNUNET_ATS_AddressRecord *ar;
337 if (NOT_FOUND == session_id)
339 if (session_id >= sh->session_array_size)
342 force_reconnect (sh);
345 /* this slot should have been removed from remove_session before */
346 ar = sh->session_array[session_id];
347 if (NULL != ar->session)
350 force_reconnect (sh);
353 GNUNET_HELLO_address_free (ar->address);
355 sh->session_array[session_id] = NULL;
360 * Type of a function to call when we receive a session release
361 * message from the service.
363 * @param cls the `struct GNUNET_ATS_SchedulingHandle`
364 * @param msg message received, NULL on timeout or fatal error
367 process_ats_session_release_message (void *cls,
368 const struct GNUNET_MessageHeader *msg)
370 struct GNUNET_ATS_SchedulingHandle *sh = cls;
371 const struct SessionReleaseMessage *srm;
373 srm = (const struct SessionReleaseMessage *) msg;
374 /* Note: peer field in srm not necessary right now,
375 but might be good to have in the future */
377 ntohl (srm->session_id));
382 * Type of a function to call when we receive a address suggestion
383 * message from the service.
385 * @param cls the `struct GNUNET_ATS_SchedulingHandle`
386 * @param msg message received, NULL on timeout or fatal error
389 process_ats_address_suggestion_message (void *cls,
390 const struct GNUNET_MessageHeader *msg)
392 struct GNUNET_ATS_SchedulingHandle *sh = cls;
393 const struct AddressSuggestionMessage *m;
394 struct GNUNET_ATS_AddressRecord *ar;
397 m = (const struct AddressSuggestionMessage *) msg;
398 session_id = ntohl (m->session_id);
402 force_reconnect (sh);
405 ar = find_session (sh, session_id, &m->peer);
409 force_reconnect (sh);
412 if (NULL == sh->suggest_cb)
414 if (GNUNET_YES == ar->in_destroy)
416 /* ignore suggestion, as this address is dying, unless BW is 0,
417 in that case signal 'disconnect' via BW 0 */
418 if ( (0 == ntohl (m->bandwidth_out.value__)) &&
419 (0 == ntohl (m->bandwidth_in.value__)) )
421 LOG (GNUNET_ERROR_TYPE_DEBUG,
422 "ATS suggests disconnect from peer `%s' with BW %u/%u\n",
423 GNUNET_i2s (&ar->address->peer),
424 (unsigned int) ntohl (m->bandwidth_out.value__),
425 (unsigned int) ntohl (m->bandwidth_in.value__));
426 sh->suggest_cb (sh->suggest_cb_cls,
435 if ( (NULL == ar->session) &&
436 (GNUNET_HELLO_address_check_option (ar->address,
437 GNUNET_HELLO_ADDRESS_INFO_INBOUND)) )
442 sh->backoff = GNUNET_TIME_UNIT_ZERO;
443 LOG (GNUNET_ERROR_TYPE_DEBUG,
444 "ATS suggests address slot %u for peer `%s' using plugin %s\n",
446 GNUNET_i2s (&ar->address->peer),
447 ar->address->transport_name);
448 sh->suggest_cb (sh->suggest_cb_cls,
458 * We encountered an error handling the MQ to the
459 * ATS service. Reconnect.
461 * @param cls the `struct GNUNET_ATS_SchedulingHandle`
462 * @param error details about the error
465 error_handler (void *cls,
466 enum GNUNET_MQ_Error error)
468 struct GNUNET_ATS_SchedulingHandle *sh = cls;
470 LOG (GNUNET_ERROR_TYPE_WARNING,
471 "ATS connection died (code %d), reconnecting\n",
473 force_reconnect (sh);
478 * Generate and transmit the `struct AddressAddMessage` for the given
481 * @param sh the scheduling handle to use for transmission
482 * @param ar the address to inform the ATS service about
485 send_add_address_message (struct GNUNET_ATS_SchedulingHandle *sh,
486 const struct GNUNET_ATS_AddressRecord *ar)
488 struct GNUNET_MQ_Envelope *ev;
489 struct AddressAddMessage *m;
490 struct GNUNET_ATS_Information *am;
496 return; /* disconnected, skip for now */
497 namelen = (NULL == ar->address->transport_name)
499 : strlen (ar->address->transport_name) + 1;
500 msize = ar->address->address_length +
501 ar->ats_count * sizeof (struct GNUNET_ATS_Information) + namelen;
503 ev = GNUNET_MQ_msg_extra (m, msize, GNUNET_MESSAGE_TYPE_ATS_ADDRESS_ADD);
504 m->ats_count = htonl (ar->ats_count);
505 m->peer = ar->address->peer;
506 m->address_length = htons (ar->address->address_length);
507 m->address_local_info = htonl ((uint32_t) ar->address->local_info);
508 m->plugin_name_length = htons (namelen);
509 m->session_id = htonl (ar->slot);
511 LOG (GNUNET_ERROR_TYPE_DEBUG,
512 "Adding address for peer `%s', plugin `%s', session %p slot %u\n",
513 GNUNET_i2s (&ar->address->peer),
514 ar->address->transport_name,
517 am = (struct GNUNET_ATS_Information *) &m[1];
520 ar->ats_count * sizeof (struct GNUNET_ATS_Information));
521 pm = (char *) &am[ar->ats_count];
523 ar->address->address,
524 ar->address->address_length);
525 if (NULL != ar->address->transport_name)
526 memcpy (&pm[ar->address->address_length],
527 ar->address->transport_name,
529 GNUNET_MQ_send (sh->mq, ev);
534 * Re-establish the connection to the ATS service.
536 * @param sh handle to use to re-connect.
539 reconnect (struct GNUNET_ATS_SchedulingHandle *sh)
541 static const struct GNUNET_MQ_MessageHandler handlers[] =
542 { { &process_ats_session_release_message,
543 GNUNET_MESSAGE_TYPE_ATS_SESSION_RELEASE,
544 sizeof (struct SessionReleaseMessage) },
545 { &process_ats_address_suggestion_message,
546 GNUNET_MESSAGE_TYPE_ATS_ADDRESS_SUGGESTION,
547 sizeof (struct AddressSuggestionMessage) },
549 struct GNUNET_MQ_Envelope *ev;
550 struct ClientStartMessage *init;
552 struct GNUNET_ATS_AddressRecord *ar;
554 GNUNET_assert (NULL == sh->client);
555 sh->client = GNUNET_CLIENT_connect ("ats", sh->cfg);
556 if (NULL == sh->client)
558 force_reconnect (sh);
561 sh->mq = GNUNET_MQ_queue_for_connection_client (sh->client,
565 ev = GNUNET_MQ_msg (init,
566 GNUNET_MESSAGE_TYPE_ATS_START);
567 init->start_flag = htonl (START_FLAG_SCHEDULING);
568 GNUNET_MQ_send (sh->mq, ev);
571 for (i=0;i<sh->session_array_size;i++)
573 ar = sh->session_array[i];
576 send_add_address_message (sh, ar);
584 * Initialize the ATS subsystem.
586 * @param cfg configuration to use
587 * @param suggest_cb notification to call whenever the suggestation changed
588 * @param suggest_cb_cls closure for @a suggest_cb
589 * @return ats context
591 struct GNUNET_ATS_SchedulingHandle *
592 GNUNET_ATS_scheduling_init (const struct GNUNET_CONFIGURATION_Handle *cfg,
593 GNUNET_ATS_AddressSuggestionCallback suggest_cb,
594 void *suggest_cb_cls)
596 struct GNUNET_ATS_SchedulingHandle *sh;
598 sh = GNUNET_new (struct GNUNET_ATS_SchedulingHandle);
600 sh->suggest_cb = suggest_cb;
601 sh->suggest_cb_cls = suggest_cb_cls;
602 GNUNET_array_grow (sh->session_array,
603 sh->session_array_size,
611 * Client is done with ATS scheduling, release resources.
613 * @param sh handle to release
616 GNUNET_ATS_scheduling_done (struct GNUNET_ATS_SchedulingHandle *sh)
622 GNUNET_MQ_destroy (sh->mq);
625 if (NULL != sh->client)
627 GNUNET_CLIENT_disconnect (sh->client);
630 if (NULL != sh->task)
632 GNUNET_SCHEDULER_cancel (sh->task);
635 for (i=0;i<sh->session_array_size;i++)
637 GNUNET_free_non_null (sh->session_array[i]);
638 sh->session_array[i] = NULL;
640 GNUNET_array_grow (sh->session_array,
641 sh->session_array_size,
648 * Test if a address and a session is known to ATS
650 * @param sh the scheduling handle
651 * @param address the address
652 * @param session the session
653 * @return #GNUNET_YES or #GNUNET_NO
656 GNUNET_ATS_session_known (struct GNUNET_ATS_SchedulingHandle *sh,
657 const struct GNUNET_HELLO_Address *address,
658 struct Session *session)
662 if (NOT_FOUND != find_session_id (sh,
665 return GNUNET_YES; /* Exists */
671 * We have a new address ATS should know. Addresses have to be added
672 * with this function before they can be: updated, set in use and
676 * @param address the address
677 * @param session session handle, can be NULL
678 * @param ats performance data for the address
679 * @param ats_count number of performance records in @a ats
680 * @return handle to the address representation inside ATS, NULL
681 * on error (i.e. ATS knows this exact address already)
683 struct GNUNET_ATS_AddressRecord *
684 GNUNET_ATS_address_add (struct GNUNET_ATS_SchedulingHandle *sh,
685 const struct GNUNET_HELLO_Address *address,
686 struct Session *session,
687 const struct GNUNET_ATS_Information *ats,
690 struct GNUNET_ATS_AddressRecord *ar;
697 /* we need a valid address */
701 namelen = (NULL == address->transport_name)
703 : strlen (address->transport_name) + 1;
704 msize = address->address_length +
705 ats_count * sizeof (struct GNUNET_ATS_Information) + namelen;
706 if ((msize + sizeof (struct AddressUpdateMessage) >= GNUNET_SERVER_MAX_MESSAGE_SIZE) ||
707 (address->address_length >= GNUNET_SERVER_MAX_MESSAGE_SIZE) ||
708 (namelen >= GNUNET_SERVER_MAX_MESSAGE_SIZE) ||
710 GNUNET_SERVER_MAX_MESSAGE_SIZE / sizeof (struct GNUNET_ATS_Information)))
712 /* address too large for us, this should not happen */
722 /* Already existing, nothing todo, but this should not happen */
726 s = find_empty_session_slot (sh);
727 ar = GNUNET_new (struct GNUNET_ATS_AddressRecord);
730 ar->session = session;
731 ar->address = GNUNET_HELLO_address_copy (address);
732 GNUNET_array_grow (ar->ats,
737 ats_count * sizeof (struct GNUNET_ATS_Information));
738 sh->session_array[s] = ar;
739 send_add_address_message (sh, ar);
745 * An address was used to initiate a session.
747 * @param ar address record to update information for
748 * @param session session handle
751 GNUNET_ATS_address_add_session (struct GNUNET_ATS_AddressRecord *ar,
752 struct Session *session)
754 GNUNET_break (NULL == ar->session);
755 ar->session = session;
760 * A session was destroyed, disassociate it from the
761 * given address record. If this was an incoming
762 * addess, destroy the address as well.
764 * @param ar address record to update information for
765 * @param session session handle
766 * @return #GNUNET_YES if the @a ar was destroyed because
767 * it was an incoming address,
768 * #GNUNET_NO if the @ar was kept because we can
769 * use it still to establish a new session
772 GNUNET_ATS_address_del_session (struct GNUNET_ATS_AddressRecord *ar,
773 struct Session *session)
775 GNUNET_break (session == ar->session);
777 if (GNUNET_HELLO_address_check_option (ar->address,
778 GNUNET_HELLO_ADDRESS_INFO_INBOUND))
780 GNUNET_ATS_address_destroy (ar);
788 * We have updated performance statistics for a given address. Note
789 * that this function can be called for addresses that are currently
790 * in use as well as addresses that are valid but not actively in use.
791 * Furthermore, the peer may not even be connected to us right now (in
792 * which case the call may be ignored or the information may be stored
793 * for later use). Update bandwidth assignments.
795 * @param ar address record to update information for
796 * @param ats performance data for the address
797 * @param ats_count number of performance records in @a ats
800 GNUNET_ATS_address_update (struct GNUNET_ATS_AddressRecord *ar,
801 const struct GNUNET_ATS_Information *ats,
804 struct GNUNET_ATS_SchedulingHandle *sh = ar->sh;
805 struct GNUNET_MQ_Envelope *ev;
806 struct AddressUpdateMessage *m;
807 struct GNUNET_ATS_Information *am;
810 LOG (GNUNET_ERROR_TYPE_DEBUG,
811 "Updating address for peer `%s', plugin `%s', session %p slot %u\n",
812 GNUNET_i2s (&ar->address->peer),
813 ar->address->transport_name,
816 GNUNET_array_grow (ar->ats,
821 ats_count * sizeof (struct GNUNET_ATS_Information));
824 return; /* disconnected, skip for now */
825 msize = ar->ats_count * sizeof (struct GNUNET_ATS_Information);
826 ev = GNUNET_MQ_msg_extra (m, msize, GNUNET_MESSAGE_TYPE_ATS_ADDRESS_UPDATE);
827 m->ats_count = htonl (ar->ats_count);
828 m->peer = ar->address->peer;
829 m->session_id = htonl (ar->slot);
830 am = (struct GNUNET_ATS_Information *) &m[1];
833 ar->ats_count * sizeof (struct GNUNET_ATS_Information));
834 GNUNET_MQ_send (sh->mq, ev);
840 * An address got destroyed, stop using it as a valid address.
842 * @param ar address to destroy
845 GNUNET_ATS_address_destroy (struct GNUNET_ATS_AddressRecord *ar)
847 struct GNUNET_ATS_SchedulingHandle *sh = ar->sh;
848 struct GNUNET_MQ_Envelope *ev;
849 struct AddressDestroyedMessage *m;
851 LOG (GNUNET_ERROR_TYPE_DEBUG,
852 "Deleting address for peer `%s', plugin `%s', slot %u session %p\n",
853 GNUNET_i2s (&ar->address->peer),
854 ar->address->transport_name,
857 GNUNET_break (NULL == ar->session);
859 ar->in_destroy = GNUNET_YES;
860 GNUNET_array_grow (ar->ats,
865 ev = GNUNET_MQ_msg (m, GNUNET_MESSAGE_TYPE_ATS_ADDRESS_DESTROYED);
866 m->session_id = htonl (ar->slot);
867 m->peer = ar->address->peer;
868 GNUNET_MQ_send (sh->mq, ev);
872 /* end of ats_api_scheduling.c */