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