src/ats/ats_api_scheduling.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2010,2011 Christian Grothoff (and other contributing authors)
   4
   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.
   9
  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.
  14
  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.
  19 */
  20 /**
  21  * @file ats/ats_api_scheduling.c
  22  * @brief automatic transport selection and outbound bandwidth determination
  23  * @author Christian Grothoff
  24  * @author Matthias Wachs
  25  */
  26 #include "platform.h"
  27 #include "gnunet_ats_service.h"
  28 #include "ats.h"
  29
  30
  31 /**
  32  * Message in linked list we should send to the ATS service.  The
  33  * actual binary message follows this struct.
  34  */
  35 struct PendingMessage
  36 {
  37
  38   /**
  39    * Kept in a DLL.
  40    */
  41   struct PendingMessage *next;
  42
  43   /**
  44    * Kept in a DLL.
  45    */
  46   struct PendingMessage *prev;
  47
  48   /**
  49    * Size of the message.
  50    */
  51   size_t size;
  52
  53   /**
  54    * Is this the 'ATS_START' message?
  55    */
  56   int is_init;
  57 };
  58
  59
  60 /**
  61  * Handle to the ATS subsystem for bandwidth/transport scheduling information.
  62  */
  63 struct GNUNET_ATS_SchedulingHandle
  64 {
  65
  66   /**
  67    * Our configuration.
  68    */
  69   const struct GNUNET_CONFIGURATION_Handle *cfg;
  70
  71   /**
  72    * Callback to invoke on suggestions.
  73    */
  74   GNUNET_ATS_AddressSuggestionCallback suggest_cb;
  75
  76   /**
  77    * Closure for 'suggest_cb'.
  78    */
  79   void *suggest_cb_cls;
  80
  81   /**
  82    * Connection to ATS service.
  83    */
  84   struct GNUNET_CLIENT_Connection *client;
  85
  86   /**
  87    * Head of list of messages for the ATS service.
  88    */
  89   struct PendingMessage *pending_head;
  90
  91   /**
  92    * Tail of list of messages for the ATS service
  93    */
  94   struct PendingMessage *pending_tail;
  95
  96   /**
  97    * Current request for transmission to ATS.
  98    */
  99   struct GNUNET_CLIENT_TransmitHandle *th;
 100
 101   /**
 102    * Array of session objects (we need to translate them to numbers and back
 103    * for the protocol; the offset in the array is the session number on the
 104    * network).  Index 0 is always NULL and reserved to represent the NULL pointer.
 105    * Unused entries are also NULL.
 106    */
 107   struct Session **session_array;
 108
 109   /**
 110    * Size of the session array.
 111    */
 112   unsigned int session_array_size;
 113
 114 };
 115
 116
 117 /**
 118  * Re-establish the connection to the ATS service.
 119  *
 120  * @param sh handle to use to re-connect.
 121  */
 122 static void
 123 reconnect (struct GNUNET_ATS_SchedulingHandle *sh);
 124
 125
 126 /**
 127  * Transmit messages from the message queue to the service
 128  * (if there are any, and if we are not already trying).
 129  *
 130  * @param sh handle to use
 131  */
 132 static void
 133 do_transmit (struct GNUNET_ATS_SchedulingHandle *sh);
 134
 135
 136 /**
 137  * We can now transmit a message to ATS. Do it.
 138  *
 139  * @param cls the 'struct GNUNET_ATS_SchedulingHandle'
 140  * @param size number of bytes we can transmit to ATS
 141  * @param buf where to copy the messages
 142  * @return number of bytes copied into buf
 143  */
 144 static size_t
 145 transmit_message_to_ats (void *cls,
 146                          size_t size,
 147                          void *buf)
 148 {
 149   struct GNUNET_ATS_SchedulingHandle *sh = cls;
 150   struct PendingMessage *p;
 151   size_t ret;
 152   char *cbuf;
 153
 154   sh->th = NULL;
 155   ret = 0;
 156   cbuf = buf;
 157   while ( (NULL != (p = sh->pending_head)) &&
 158           (p->size <= size) )
 159   {
 160     memcpy (&cbuf[ret], &p[1], p->size);
 161     ret += p->size;
 162     GNUNET_CONTAINER_DLL_remove (sh->pending_head,
 163                                  sh->pending_tail,
 164                                  p);
 165     GNUNET_free (p);
 166   }
 167   do_transmit (sh);
 168   return ret;
 169 }
 170
 171
 172 /**
 173  * Transmit messages from the message queue to the service
 174  * (if there are any, and if we are not already trying).
 175  *
 176  * @param sh handle to use
 177  */
 178 static void
 179 do_transmit (struct GNUNET_ATS_SchedulingHandle *sh)
 180 {
 181   struct PendingMessage *p;
 182
 183   if (NULL != sh->th)
 184     return;
 185   if (NULL == (p = sh->pending_head))
 186     return;
 187   sh->th = GNUNET_CLIENT_notify_transmit_ready (sh->client,
 188                                                 p->size,
 189                                                 GNUNET_TIME_UNIT_FOREVER_REL,
 190                                                 GNUNET_YES,
 191                                                 &transmit_message_to_ats, sh);
 192 }
 193
 194
 195 /**
 196  * Find the session object corresponding to the given session ID.
 197  *
 198  * @param sh our handle
 199  * @param session_id current session ID
 200  * @return the session object (or NULL)
 201  */
 202 static struct Session*
 203 find_session (struct GNUNET_ATS_SchedulingHandle *sh,
 204               uint32_t session_id)
 205 {
 206   if (session_id >= sh->session_array_size)
 207   {
 208     GNUNET_break (0);
 209     return NULL;
 210   }
 211   return sh->session_array[session_id];
 212 }
 213
 214
 215 /**
 216  * Get the ID for the given session object.  If we do not have an ID for
 217  * the given session object, allocate one.
 218  *
 219  * @param sh our handle
 220  * @param session session object
 221  * @return the session id
 222  */
 223 static uint32_t
 224 get_session_id (struct GNUNET_ATS_SchedulingHandle *sh,
 225                 struct Session *session)
 226 {
 227   unsigned int i;
 228   unsigned int f;
 229
 230   f = 0;
 231   for (i=1;i<sh->session_array_size;i++)
 232   {
 233     if (session == sh->session_array[i])
 234       return i;
 235     if ( (f == 0) &&
 236          (sh->session_array[i] == NULL) )
 237       f = i;
 238   }
 239   if (f == 0)
 240   {
 241     f = sh->session_array_size;
 242     GNUNET_array_grow (sh->session_array,
 243                        sh->session_array_size,
 244                        sh->session_array_size * 2);
 245   }
 246   sh->session_array[f] = session;
 247   return f;
 248 }
 249
 250
 251 /**
 252  * Remove the session of the given session ID from the session
 253  * table (it is no longer valid).
 254  *
 255  * @param sh our handle
 256  * @param session_id identifies session that is no longer valid
 257  */
 258 static void
 259 remove_session (struct GNUNET_ATS_SchedulingHandle *sh,
 260                 uint32_t session_id)
 261 {
 262   GNUNET_assert (session_id < sh->session_array_size);
 263   sh->session_array[session_id] = NULL;
 264 }
 265
 266
 267 /**
 268  * Type of a function to call when we receive a message
 269  * from the service.
 270  *
 271  * @param cls the 'struct GNUNET_ATS_SchedulingHandle'
 272  * @param msg message received, NULL on timeout or fatal error
 273  */
 274 static void
 275 process_ats_message (void *cls,
 276                      const struct GNUNET_MessageHeader *msg)
 277 {
 278   struct GNUNET_ATS_SchedulingHandle *sh = cls;
 279   const struct AddressSuggestionMessage *m;
 280   const char *address;
 281   const char *plugin_name;
 282   uint16_t address_length;
 283
 284   if (NULL == msg)
 285   {
 286     GNUNET_CLIENT_disconnect (sh->client, GNUNET_NO);
 287     sh->client = NULL;
 288     reconnect (sh);
 289     return;
 290   }
 291   if ( (ntohs (msg->type) != GNUNET_MESSAGE_TYPE_ATS_ADDRESS_SUGGESTION) ||
 292        (ntohs (msg->size) <= sizeof (struct AddressSuggestionMessage)) )
 293   {
 294     GNUNET_break (0);
 295     GNUNET_CLIENT_disconnect (sh->client, GNUNET_NO);
 296     sh->client = NULL;
 297     reconnect (sh);
 298     return;
 299   }
 300   m = (const struct AddressSuggestionMessage*) msg;
 301   address = (const char*) &m[1];
 302   address_length = ntohs (m->address_length);
 303   plugin_name = &address[address_length];
 304   GNUNET_break (0 == ntohl (m->reserved));
 305   if ( (ntohs (m->address_length) +
 306         ntohs (m->plugin_name_length) +
 307         sizeof (struct AddressSuggestionMessage) != ntohs (msg->size))  ||
 308        (plugin_name[ntohs (m->plugin_name_length) - 1] != '\0') )
 309   {
 310     GNUNET_break (0);
 311     GNUNET_CLIENT_disconnect (sh->client, GNUNET_NO);
 312     sh->client = NULL;
 313     reconnect (sh);
 314     return;
 315   }
 316   sh->suggest_cb (sh->suggest_cb_cls,
 317                   &m->peer,
 318                   plugin_name,
 319                   address, address_length,
 320                   find_session (sh, ntohl (m->session_id)),
 321                   m->bandwidth_out,
 322                   m->bandwidth_in);
 323   GNUNET_CLIENT_receive (sh->client,
 324                          &process_ats_message, sh,
 325                          GNUNET_TIME_UNIT_FOREVER_REL);
 326 }
 327
 328
 329 /**
 330  * Re-establish the connection to the ATS service.
 331  *
 332  * @param sh handle to use to re-connect.
 333  */
 334 static void
 335 reconnect (struct GNUNET_ATS_SchedulingHandle *sh)
 336 {
 337   struct PendingMessage *p;
 338   struct ClientStartMessage *init;
 339
 340   GNUNET_assert (NULL == sh->client);
 341   sh->client = GNUNET_CLIENT_connect ("ats", sh->cfg);
 342   GNUNET_assert (NULL != sh->client);
 343   GNUNET_CLIENT_receive (sh->client,
 344                          &process_ats_message, sh,
 345                          GNUNET_TIME_UNIT_FOREVER_REL);
 346   if ( (NULL == (p = sh->pending_head)) ||
 347        (GNUNET_YES != p->is_init) )
 348   {
 349     p = GNUNET_malloc (sizeof (struct PendingMessage) +
 350                        sizeof (struct ClientStartMessage));
 351     p->size = sizeof (struct ClientStartMessage);
 352     p->is_init = GNUNET_YES;
 353     init = (struct ClientStartMessage *) &p[1];
 354     init->header.type = htons (GNUNET_MESSAGE_TYPE_ATS_START);
 355     init->header.size = htons (sizeof (struct ClientStartMessage));
 356     init->start_flag = htonl (START_FLAG_SCHEDULING);
 357     GNUNET_CONTAINER_DLL_insert (sh->pending_head,
 358                                  sh->pending_tail,
 359                                  p);
 360   }
 361   do_transmit (sh);
 362 }
 363
 364
 365 /**
 366  * Initialize the ATS subsystem.
 367  *
 368  * @param cfg configuration to use
 369  * @param suggest_cb notification to call whenever the suggestation changed
 370  * @param suggest_cb_cls closure for 'suggest_cb'
 371  * @return ats context
 372  */
 373 struct GNUNET_ATS_SchedulingHandle *
 374 GNUNET_ATS_scheduling_init (const struct GNUNET_CONFIGURATION_Handle *cfg,
 375                             GNUNET_ATS_AddressSuggestionCallback suggest_cb,
 376                             void *suggest_cb_cls)
 377 {
 378   struct GNUNET_ATS_SchedulingHandle *sh;
 379
 380   sh = GNUNET_malloc (sizeof (struct GNUNET_ATS_SchedulingHandle));
 381   sh->cfg = cfg;
 382   sh->suggest_cb = suggest_cb;
 383   sh->suggest_cb_cls = suggest_cb_cls;
 384   GNUNET_array_grow (sh->session_array,
 385                      sh->session_array_size,
 386                      4);
 387   reconnect (sh);
 388   return sh;
 389 }
 390
 391
 392 /**
 393  * Client is done with ATS scheduling, release resources.
 394  *
 395  * @param sh handle to release
 396  */
 397 void
 398 GNUNET_ATS_scheduling_done (struct GNUNET_ATS_SchedulingHandle *sh)
 399 {
 400   struct PendingMessage *p;
 401
 402   while (NULL != (p = sh->pending_head))
 403   {
 404     GNUNET_CONTAINER_DLL_remove (sh->pending_head,
 405                                  sh->pending_tail,
 406                                  p);
 407     GNUNET_free (p);
 408   }
 409   GNUNET_CLIENT_disconnect (sh->client, GNUNET_NO);
 410   GNUNET_array_grow (sh->session_array,
 411                      sh->session_array_size,
 412                      0);
 413   GNUNET_free (sh);
 414 }
 415
 416
 417 /**
 418  * We would like to establish a new connection with a peer.  ATS
 419  * should suggest a good address to begin with.
 420  *
 421  * @param sh handle
 422  * @param peer identity of the peer we need an address for
 423  */
 424 void
 425 GNUNET_ATS_suggest_address (struct GNUNET_ATS_SchedulingHandle *sh,
 426                             const struct GNUNET_PeerIdentity *peer)
 427 {
 428   struct PendingMessage *p;
 429   struct RequestAddressMessage *m;
 430
 431   p = GNUNET_malloc (sizeof (struct PendingMessage) +
 432                      sizeof (struct RequestAddressMessage));
 433   p->size = sizeof (struct RequestAddressMessage);
 434   p->is_init = GNUNET_NO;
 435   m = (struct RequestAddressMessage*) &p[1];
 436   m->header.type = htons (GNUNET_MESSAGE_TYPE_ATS_REQUEST_ADDRESS);
 437   m->header.size = htons (sizeof (struct RequestAddressMessage));
 438   m->reserved = htonl (0);
 439   m->peer = *peer;
 440   GNUNET_CONTAINER_DLL_insert_tail (sh->pending_head,
 441                                     sh->pending_tail,
 442                                     p);
 443   do_transmit (sh);
 444 }
 445
 446
 447 /**
 448  * We have updated performance statistics for a given address.  Note
 449  * that this function can be called for addresses that are currently
 450  * in use as well as addresses that are valid but not actively in use.
 451  * Furthermore, the peer may not even be connected to us right now (in
 452  * which case the call may be ignored or the information may be stored
 453  * for later use).  Update bandwidth assignments.
 454  *
 455  * @param sh handle
 456  * @param peer identity of the new peer
 457  * @param plugin_name name of the transport plugin
 458  * @param plugin_addr address  (if available)
 459  * @param plugin_addr_len number of bytes in plugin_addr
 460  * @param session session handle (if available)
 461  * @param ats performance data for the address
 462  * @param ats_count number of performance records in 'ats'
 463  */
 464 void
 465 GNUNET_ATS_address_update (struct GNUNET_ATS_SchedulingHandle *sh,
 466                            const struct GNUNET_PeerIdentity *peer,
 467                            const char *plugin_name,
 468                            const void *plugin_addr, size_t plugin_addr_len,
 469                            struct Session *session,
 470                            const struct GNUNET_TRANSPORT_ATS_Information *ats,
 471                            uint32_t ats_count)
 472 {
 473   struct PendingMessage *p;
 474   struct AddressUpdateMessage *m;
 475   struct GNUNET_TRANSPORT_ATS_Information *am;
 476   char *pm;
 477   size_t namelen;
 478   size_t msize;
 479
 480   namelen = (plugin_name == NULL) ? 0 : strlen (plugin_name) + 1;
 481   msize = sizeof (struct AddressUpdateMessage) + plugin_addr_len +
 482     ats_count * sizeof (struct GNUNET_TRANSPORT_ATS_Information) + namelen;
 483   if ( (msize >= GNUNET_SERVER_MAX_MESSAGE_SIZE) ||
 484        (plugin_addr_len  >= GNUNET_SERVER_MAX_MESSAGE_SIZE) ||
 485        (namelen  >= GNUNET_SERVER_MAX_MESSAGE_SIZE) ||
 486        (ats_count >= GNUNET_SERVER_MAX_MESSAGE_SIZE / sizeof (struct GNUNET_TRANSPORT_ATS_Information)) )
 487   {
 488     GNUNET_break (0);
 489     return;
 490   }
 491   p = GNUNET_malloc (sizeof (struct PendingMessage) +  msize);
 492   p->size = msize;
 493   p->is_init = GNUNET_NO;
 494   m = (struct AddressUpdateMessage*) &p[1];
 495   m->header.type = htons (GNUNET_MESSAGE_TYPE_ATS_ADDRESS_UPDATE);
 496   m->header.size = htons (msize);
 497   m->ats_count = htonl (ats_count);
 498   m->peer = *peer;
 499   m->address_length = htons (plugin_addr_len);
 500   m->plugin_name_length = htons (namelen);
 501   m->session_id = htonl (get_session_id (sh, session));
 502   am = (struct GNUNET_TRANSPORT_ATS_Information*) &m[1];
 503   memcpy (am, ats, ats_count * sizeof (struct GNUNET_TRANSPORT_ATS_Information));
 504   pm = (char *) &am[ats_count];
 505   memcpy (pm, plugin_addr, plugin_addr_len);
 506   memcpy (&pm[plugin_addr_len], plugin_name, namelen);
 507   GNUNET_CONTAINER_DLL_insert_tail (sh->pending_head,
 508                                     sh->pending_tail,
 509                                     p);
 510   do_transmit (sh);
 511 }
 512
 513
 514 /**
 515  * A session got destroyed, stop including it as a valid address.
 516  *
 517  * @param sh handle
 518  * @param peer identity of the peer
 519  * @param plugin_name name of the transport plugin
 520  * @param plugin_addr address  (if available)
 521  * @param plugin_addr_len number of bytes in plugin_addr
 522  * @param session session handle that is no longer valid
 523  */
 524 void
 525 GNUNET_ATS_address_destroyed (struct GNUNET_ATS_SchedulingHandle *sh,
 526                               const struct GNUNET_PeerIdentity *peer,
 527                               const char *plugin_name,
 528                               const void *plugin_addr,
 529                               size_t plugin_addr_len,
 530                               struct Session *session)
 531 {
 532   struct PendingMessage *p;
 533   struct AddressDestroyedMessage *m;
 534   char *pm;
 535   size_t namelen;
 536   size_t msize;
 537   uint32_t session_id;
 538
 539   namelen = (plugin_name == NULL) ? 0 : strlen (plugin_name) + 1;
 540   msize = sizeof (struct AddressUpdateMessage) + plugin_addr_len +
 541     namelen;
 542   if ( (msize >= GNUNET_SERVER_MAX_MESSAGE_SIZE) ||
 543        (plugin_addr_len  >= GNUNET_SERVER_MAX_MESSAGE_SIZE) ||
 544        (namelen  >= GNUNET_SERVER_MAX_MESSAGE_SIZE) )
 545   {
 546     GNUNET_break (0);
 547     return;
 548   }
 549   p = GNUNET_malloc (sizeof (struct PendingMessage) +  msize);
 550   p->size = msize;
 551   p->is_init = GNUNET_NO;
 552   m = (struct AddressDestroyedMessage*) &p[1];
 553   m->header.type = htons (GNUNET_MESSAGE_TYPE_ATS_ADDRESS_DESTROYED);
 554   m->header.size = htons (msize);
 555   m->reserved = htonl (0);
 556   m->peer = *peer;
 557   m->address_length = htons (plugin_addr_len);
 558   m->plugin_name_length = htons (namelen);
 559   m->session_id = htonl (session_id = get_session_id (sh, session));
 560   pm = (char *) &m[1];
 561   memcpy (pm, plugin_addr, plugin_addr_len);
 562   memcpy (&pm[plugin_addr_len], plugin_name, namelen);
 563   GNUNET_CONTAINER_DLL_insert_tail (sh->pending_head,
 564                                     sh->pending_tail,
 565                                     p);
 566   do_transmit (sh);
 567   remove_session (sh, session_id);
 568 }
 569
 570 /* end of ats_api_scheduling.c */