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    * Task to trigger reconnect.
 111    */
 112   GNUNET_SCHEDULER_TaskIdentifier task;
 113
 114   /**
 115    * Size of the session array.
 116    */
 117   unsigned int session_array_size;
 118
 119 };
 120
 121
 122 /**
 123  * Re-establish the connection to the ATS service.
 124  *
 125  * @param sh handle to use to re-connect.
 126  */
 127 static void
 128 reconnect (struct GNUNET_ATS_SchedulingHandle *sh);
 129
 130
 131
 132 /**
 133  * Re-establish the connection to the ATS service.
 134  *
 135  * @param cls handle to use to re-connect.
 136  * @param tc scheduler context
 137  */
 138 static void
 139 reconnect_task (void *cls,
 140                 const struct GNUNET_SCHEDULER_TaskContext *tc)
 141 {
 142   struct GNUNET_ATS_SchedulingHandle *sh = cls;
 143
 144   sh->task = GNUNET_SCHEDULER_NO_TASK;
 145   reconnect (sh);
 146 }
 147
 148
 149 /**
 150  * Transmit messages from the message queue to the service
 151  * (if there are any, and if we are not already trying).
 152  *
 153  * @param sh handle to use
 154  */
 155 static void
 156 do_transmit (struct GNUNET_ATS_SchedulingHandle *sh);
 157
 158
 159 /**
 160  * Type of a function to call when we receive a message
 161  * from the service.
 162  *
 163  * @param cls the 'struct GNUNET_ATS_SchedulingHandle'
 164  * @param msg message received, NULL on timeout or fatal error
 165  */
 166 static void
 167 process_ats_message (void *cls,
 168                      const struct GNUNET_MessageHeader *msg);
 169
 170
 171 /**
 172  * We can now transmit a message to ATS. Do it.
 173  *
 174  * @param cls the 'struct GNUNET_ATS_SchedulingHandle'
 175  * @param size number of bytes we can transmit to ATS
 176  * @param buf where to copy the messages
 177  * @return number of bytes copied into buf
 178  */
 179 static size_t
 180 transmit_message_to_ats (void *cls,
 181                          size_t size,
 182                          void *buf)
 183 {
 184   struct GNUNET_ATS_SchedulingHandle *sh = cls;
 185   struct PendingMessage *p;
 186   size_t ret;
 187   char *cbuf;
 188
 189   sh->th = NULL;
 190   ret = 0;
 191   cbuf = buf;
 192   while ( (NULL != (p = sh->pending_head)) &&
 193           (p->size <= size) )
 194   {
 195     memcpy (&cbuf[ret], &p[1], p->size);
 196     ret += p->size;
 197     size -= p->size;
 198     GNUNET_CONTAINER_DLL_remove (sh->pending_head,
 199                                  sh->pending_tail,
 200                                  p);
 201     if (GNUNET_YES == p->is_init)
 202       GNUNET_CLIENT_receive (sh->client,
 203                              &process_ats_message, sh,
 204                              GNUNET_TIME_UNIT_FOREVER_REL);
 205     GNUNET_free (p);
 206   }
 207   do_transmit (sh);
 208   return ret;
 209 }
 210
 211
 212 /**
 213  * Transmit messages from the message queue to the service
 214  * (if there are any, and if we are not already trying).
 215  *
 216  * @param sh handle to use
 217  */
 218 static void
 219 do_transmit (struct GNUNET_ATS_SchedulingHandle *sh)
 220 {
 221   struct PendingMessage *p;
 222
 223   if (NULL != sh->th)
 224     return;
 225   if (NULL == (p = sh->pending_head))
 226     return;
 227   sh->th = GNUNET_CLIENT_notify_transmit_ready (sh->client,
 228                                                 p->size,
 229                                                 GNUNET_TIME_UNIT_FOREVER_REL,
 230                                                 GNUNET_YES,
 231                                                 &transmit_message_to_ats, sh);
 232 }
 233
 234
 235 /**
 236  * Find the session object corresponding to the given session ID.
 237  *
 238  * @param sh our handle
 239  * @param session_id current session ID
 240  * @return the session object (or NULL)
 241  */
 242 static struct Session*
 243 find_session (struct GNUNET_ATS_SchedulingHandle *sh,
 244               uint32_t session_id)
 245 {
 246   if (session_id >= sh->session_array_size)
 247   {
 248     GNUNET_break (0);
 249     return NULL;
 250   }
 251   return sh->session_array[session_id];
 252 }
 253
 254
 255 /**
 256  * Get the ID for the given session object.  If we do not have an ID for
 257  * the given session object, allocate one.
 258  *
 259  * @param sh our handle
 260  * @param session session object
 261  * @return the session id
 262  */
 263 static uint32_t
 264 get_session_id (struct GNUNET_ATS_SchedulingHandle *sh,
 265                 struct Session *session)
 266 {
 267   unsigned int i;
 268   unsigned int f;
 269
 270   f = 0;
 271   for (i=1;i<sh->session_array_size;i++)
 272   {
 273     if (session == sh->session_array[i])
 274       return i;
 275     if ( (f == 0) &&
 276          (sh->session_array[i] == NULL) )
 277       f = i;
 278   }
 279   if (f == 0)
 280   {
 281     f = sh->session_array_size;
 282     GNUNET_array_grow (sh->session_array,
 283                        sh->session_array_size,
 284                        sh->session_array_size * 2);
 285   }
 286   sh->session_array[f] = session;
 287   return f;
 288 }
 289
 290
 291 /**
 292  * Remove the session of the given session ID from the session
 293  * table (it is no longer valid).
 294  *
 295  * @param sh our handle
 296  * @param session_id identifies session that is no longer valid
 297  */
 298 static void
 299 remove_session (struct GNUNET_ATS_SchedulingHandle *sh,
 300                 uint32_t session_id)
 301 {
 302   GNUNET_assert (session_id < sh->session_array_size);
 303   sh->session_array[session_id] = NULL;
 304 }
 305
 306
 307 /**
 308  * Type of a function to call when we receive a message
 309  * from the service.
 310  *
 311  * @param cls the 'struct GNUNET_ATS_SchedulingHandle'
 312  * @param msg message received, NULL on timeout or fatal error
 313  */
 314 static void
 315 process_ats_message (void *cls,
 316                      const struct GNUNET_MessageHeader *msg)
 317 {
 318   struct GNUNET_ATS_SchedulingHandle *sh = cls;
 319   const struct AddressSuggestionMessage *m;
 320   const struct GNUNET_ATS_Information *atsi;
 321   const char *address;
 322   const char *plugin_name;
 323   uint16_t address_length;
 324   uint16_t plugin_name_length;
 325   uint32_t ats_count;
 326
 327   if (NULL == msg)
 328   {
 329     GNUNET_CLIENT_disconnect (sh->client, GNUNET_NO);
 330     sh->client = NULL;
 331     sh->task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
 332                                              &reconnect_task, sh);
 333     return;
 334   }
 335   if ( (ntohs (msg->type) != GNUNET_MESSAGE_TYPE_ATS_ADDRESS_SUGGESTION) ||
 336        (ntohs (msg->size) <= sizeof (struct AddressSuggestionMessage)) )
 337   {
 338     GNUNET_break (0);
 339     GNUNET_CLIENT_disconnect (sh->client, GNUNET_NO);
 340     sh->client = NULL;
 341     sh->task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
 342                                              &reconnect_task, sh);
 343     return;
 344   }
 345   m = (const struct AddressSuggestionMessage*) msg;
 346   ats_count = ntohl (m->ats_count);
 347   address_length = ntohs (m->address_length);
 348   atsi = (const struct GNUNET_ATS_Information*) &m[1];
 349   address = (const char*) &atsi[ats_count];
 350   plugin_name = &address[address_length];
 351   plugin_name_length = ntohs (m->plugin_name_length);
 352   if ( (address_length +
 353         plugin_name_length +
 354         ats_count * sizeof (struct GNUNET_ATS_Information) +
 355         sizeof (struct AddressSuggestionMessage) != ntohs (msg->size))  ||
 356        (ats_count > GNUNET_SERVER_MAX_MESSAGE_SIZE / sizeof (struct GNUNET_ATS_Information)) ||
 357        (plugin_name[plugin_name_length - 1] != '\0') )
 358   {
 359     GNUNET_break (0);
 360     GNUNET_CLIENT_disconnect (sh->client, GNUNET_NO);
 361     sh->client = NULL;
 362     sh->task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
 363                                              &reconnect_task, sh);
 364     return;
 365   }
 366   sh->suggest_cb (sh->suggest_cb_cls,
 367                   &m->peer,
 368                   plugin_name,
 369                   address, address_length,
 370                   find_session (sh, ntohl (m->session_id)),
 371                   m->bandwidth_out,
 372                   m->bandwidth_in,
 373                   atsi,
 374                   ats_count);
 375   GNUNET_CLIENT_receive (sh->client,
 376                          &process_ats_message, sh,
 377                          GNUNET_TIME_UNIT_FOREVER_REL);
 378 }
 379
 380
 381 /**
 382  * Re-establish the connection to the ATS service.
 383  *
 384  * @param sh handle to use to re-connect.
 385  */
 386 static void
 387 reconnect (struct GNUNET_ATS_SchedulingHandle *sh)
 388 {
 389   struct PendingMessage *p;
 390   struct ClientStartMessage *init;
 391
 392   GNUNET_assert (NULL == sh->client);
 393   sh->client = GNUNET_CLIENT_connect ("ats", sh->cfg);
 394   GNUNET_assert (NULL != sh->client);
 395   if ( (NULL == (p = sh->pending_head)) ||
 396        (GNUNET_YES != p->is_init) )
 397   {
 398     p = GNUNET_malloc (sizeof (struct PendingMessage) +
 399                        sizeof (struct ClientStartMessage));
 400     p->size = sizeof (struct ClientStartMessage);
 401     p->is_init = GNUNET_YES;
 402     init = (struct ClientStartMessage *) &p[1];
 403     init->header.type = htons (GNUNET_MESSAGE_TYPE_ATS_START);
 404     init->header.size = htons (sizeof (struct ClientStartMessage));
 405     init->start_flag = htonl (START_FLAG_SCHEDULING);
 406     GNUNET_CONTAINER_DLL_insert (sh->pending_head,
 407                                  sh->pending_tail,
 408                                  p);
 409   }
 410   do_transmit (sh);
 411 }
 412
 413
 414 /**
 415  * Initialize the ATS subsystem.
 416  *
 417  * @param cfg configuration to use
 418  * @param suggest_cb notification to call whenever the suggestation changed
 419  * @param suggest_cb_cls closure for 'suggest_cb'
 420  * @return ats context
 421  */
 422 struct GNUNET_ATS_SchedulingHandle *
 423 GNUNET_ATS_scheduling_init (const struct GNUNET_CONFIGURATION_Handle *cfg,
 424                             GNUNET_ATS_AddressSuggestionCallback suggest_cb,
 425                             void *suggest_cb_cls)
 426 {
 427   struct GNUNET_ATS_SchedulingHandle *sh;
 428
 429   sh = GNUNET_malloc (sizeof (struct GNUNET_ATS_SchedulingHandle));
 430   sh->cfg = cfg;
 431   sh->suggest_cb = suggest_cb;
 432   sh->suggest_cb_cls = suggest_cb_cls;
 433   GNUNET_array_grow (sh->session_array,
 434                      sh->session_array_size,
 435                      4);
 436   reconnect (sh);
 437   return sh;
 438 }
 439
 440
 441 /**
 442  * Client is done with ATS scheduling, release resources.
 443  *
 444  * @param sh handle to release
 445  */
 446 void
 447 GNUNET_ATS_scheduling_done (struct GNUNET_ATS_SchedulingHandle *sh)
 448 {
 449   struct PendingMessage *p;
 450
 451   while (NULL != (p = sh->pending_head))
 452   {
 453     GNUNET_CONTAINER_DLL_remove (sh->pending_head,
 454                                  sh->pending_tail,
 455                                  p);
 456     GNUNET_free (p);
 457   }
 458   if (NULL != sh->client)
 459   {
 460     GNUNET_CLIENT_disconnect (sh->client, GNUNET_NO);
 461     sh->client = NULL;
 462   }
 463   if (GNUNET_SCHEDULER_NO_TASK != sh->task)
 464   {
 465     GNUNET_SCHEDULER_cancel (sh->task);
 466     sh->task = GNUNET_SCHEDULER_NO_TASK;
 467   }
 468   GNUNET_array_grow (sh->session_array,
 469                      sh->session_array_size,
 470                      0);
 471   GNUNET_free (sh);
 472 }
 473
 474
 475 /**
 476  * We would like to establish a new connection with a peer.  ATS
 477  * should suggest a good address to begin with.
 478  *
 479  * @param sh handle
 480  * @param peer identity of the peer we need an address for
 481  */
 482 void
 483 GNUNET_ATS_suggest_address (struct GNUNET_ATS_SchedulingHandle *sh,
 484                             const struct GNUNET_PeerIdentity *peer)
 485 {
 486   struct PendingMessage *p;
 487   struct RequestAddressMessage *m;
 488
 489   p = GNUNET_malloc (sizeof (struct PendingMessage) +
 490                      sizeof (struct RequestAddressMessage));
 491   p->size = sizeof (struct RequestAddressMessage);
 492   p->is_init = GNUNET_NO;
 493   m = (struct RequestAddressMessage*) &p[1];
 494   m->header.type = htons (GNUNET_MESSAGE_TYPE_ATS_REQUEST_ADDRESS);
 495   m->header.size = htons (sizeof (struct RequestAddressMessage));
 496   m->reserved = htonl (0);
 497   m->peer = *peer;
 498   GNUNET_CONTAINER_DLL_insert_tail (sh->pending_head,
 499                                     sh->pending_tail,
 500                                     p);
 501   do_transmit (sh);
 502 }
 503
 504
 505 /**
 506  * We have updated performance statistics for a given address.  Note
 507  * that this function can be called for addresses that are currently
 508  * in use as well as addresses that are valid but not actively in use.
 509  * Furthermore, the peer may not even be connected to us right now (in
 510  * which case the call may be ignored or the information may be stored
 511  * for later use).  Update bandwidth assignments.
 512  *
 513  * @param sh handle
 514  * @param peer identity of the new peer
 515  * @param plugin_name name of the transport plugin
 516  * @param plugin_addr address  (if available)
 517  * @param plugin_addr_len number of bytes in plugin_addr
 518  * @param session session handle (if available)
 519  * @param ats performance data for the address
 520  * @param ats_count number of performance records in 'ats'
 521  */
 522 void
 523 GNUNET_ATS_address_update (struct GNUNET_ATS_SchedulingHandle *sh,
 524                            const struct GNUNET_PeerIdentity *peer,
 525                            const char *plugin_name,
 526                            const void *plugin_addr, size_t plugin_addr_len,
 527                            struct Session *session,
 528                            const struct GNUNET_ATS_Information *ats,
 529                            uint32_t ats_count)
 530 {
 531   struct PendingMessage *p;
 532   struct AddressUpdateMessage *m;
 533   struct GNUNET_ATS_Information *am;
 534   char *pm;
 535   size_t namelen;
 536   size_t msize;
 537
 538   namelen = (plugin_name == NULL) ? 0 : strlen (plugin_name) + 1;
 539   msize = sizeof (struct AddressUpdateMessage) + plugin_addr_len +
 540     ats_count * sizeof (struct GNUNET_ATS_Information) + namelen;
 541   if ( (msize >= GNUNET_SERVER_MAX_MESSAGE_SIZE) ||
 542        (plugin_addr_len  >= GNUNET_SERVER_MAX_MESSAGE_SIZE) ||
 543        (namelen  >= GNUNET_SERVER_MAX_MESSAGE_SIZE) ||
 544        (ats_count >= GNUNET_SERVER_MAX_MESSAGE_SIZE / sizeof (struct GNUNET_ATS_Information)) )
 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 AddressUpdateMessage*) &p[1];
 553   m->header.type = htons (GNUNET_MESSAGE_TYPE_ATS_ADDRESS_UPDATE);
 554   m->header.size = htons (msize);
 555   m->ats_count = htonl (ats_count);
 556   m->peer = *peer;
 557   m->address_length = htons (plugin_addr_len);
 558   m->plugin_name_length = htons (namelen);
 559   m->session_id = htonl (get_session_id (sh, session));
 560   am = (struct GNUNET_ATS_Information*) &m[1];
 561   memcpy (am, ats, ats_count * sizeof (struct GNUNET_ATS_Information));
 562   pm = (char *) &am[ats_count];
 563   memcpy (pm, plugin_addr, plugin_addr_len);
 564   memcpy (&pm[plugin_addr_len], plugin_name, namelen);
 565   GNUNET_CONTAINER_DLL_insert_tail (sh->pending_head,
 566                                     sh->pending_tail,
 567                                     p);
 568   do_transmit (sh);
 569 }
 570
 571
 572 /**
 573  * A session got destroyed, stop including it as a valid address.
 574  *
 575  * @param sh handle
 576  * @param peer identity of the peer
 577  * @param plugin_name name of the transport plugin
 578  * @param plugin_addr address  (if available)
 579  * @param plugin_addr_len number of bytes in plugin_addr
 580  * @param session session handle that is no longer valid
 581  */
 582 void
 583 GNUNET_ATS_address_destroyed (struct GNUNET_ATS_SchedulingHandle *sh,
 584                               const struct GNUNET_PeerIdentity *peer,
 585                               const char *plugin_name,
 586                               const void *plugin_addr,
 587                               size_t plugin_addr_len,
 588                               struct Session *session)
 589 {
 590   struct PendingMessage *p;
 591   struct AddressDestroyedMessage *m;
 592   char *pm;
 593   size_t namelen;
 594   size_t msize;
 595   uint32_t session_id;
 596
 597   namelen = (plugin_name == NULL) ? 0 : strlen (plugin_name) + 1;
 598   msize = sizeof (struct AddressUpdateMessage) + plugin_addr_len +
 599     namelen;
 600   if ( (msize >= GNUNET_SERVER_MAX_MESSAGE_SIZE) ||
 601        (plugin_addr_len  >= GNUNET_SERVER_MAX_MESSAGE_SIZE) ||
 602        (namelen  >= GNUNET_SERVER_MAX_MESSAGE_SIZE) )
 603   {
 604     GNUNET_break (0);
 605     return;
 606   }
 607   p = GNUNET_malloc (sizeof (struct PendingMessage) +  msize);
 608   p->size = msize;
 609   p->is_init = GNUNET_NO;
 610   m = (struct AddressDestroyedMessage*) &p[1];
 611   m->header.type = htons (GNUNET_MESSAGE_TYPE_ATS_ADDRESS_DESTROYED);
 612   m->header.size = htons (msize);
 613   m->reserved = htonl (0);
 614   m->peer = *peer;
 615   m->address_length = htons (plugin_addr_len);
 616   m->plugin_name_length = htons (namelen);
 617   m->session_id = htonl (session_id = get_session_id (sh, session));
 618   pm = (char *) &m[1];
 619   memcpy (pm, plugin_addr, plugin_addr_len);
 620   memcpy (&pm[plugin_addr_len], plugin_name, namelen);
 621   GNUNET_CONTAINER_DLL_insert_tail (sh->pending_head,
 622                                     sh->pending_tail,
 623                                     p);
 624   do_transmit (sh);
 625   remove_session (sh, session_id);
 626 }
 627
 628 /* end of ats_api_scheduling.c */