src/testbed/gnunet-service-testbed_connectionpool.c

   1 /*
   2    This file is part of GNUnet.
   3    Copyright (C) 2008--2015 GNUnet e.V.
   4
   5    GNUnet is free software: you can redistribute it and/or modify it
   6    under the terms of the GNU Affero General Public License as published
   7    by the Free Software Foundation, either version 3 of the License,
   8    or (at your 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    Affero General Public License for more details.
  14
  15    You should have received a copy of the GNU Affero General Public License
  16    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17
  18      SPDX-License-Identifier: AGPL3.0-or-later
  19  */
  20
  21 /**
  22  * @file testbed/gnunet-service-testbed_connectionpool.c
  23  * @brief connection pooling for connections to peers' services
  24  * @author Sree Harsha Totakura <sreeharsha@totakura.in>
  25  */
  26
  27 #include "gnunet-service-testbed.h"
  28 #include "gnunet-service-testbed_connectionpool.h"
  29 #include "testbed_api_operations.h"
  30 #include "gnunet_transport_service.h"
  31
  32 /**
  33  * Redefine LOG with a changed log component string
  34  */
  35 #ifdef LOG
  36 #undef LOG
  37 #endif
  38 #define LOG(kind, ...) \
  39   GNUNET_log_from(kind, "testbed-connectionpool", __VA_ARGS__)
  40
  41
  42 /**
  43  * Time to expire a cache entry
  44  */
  45 #define CACHE_EXPIRY \
  46   GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_SECONDS, 15)
  47
  48
  49 /**
  50  * The request handle for obtaining a pooled connection
  51  */
  52 struct GST_ConnectionPool_GetHandle;
  53
  54
  55 /**
  56  * A pooled connection
  57  */
  58 struct PooledConnection {
  59   /**
  60    * Next ptr for placing this object in the DLL of least recently used pooled
  61    * connections
  62    */
  63   struct PooledConnection *next;
  64
  65   /**
  66    * Prev ptr for placing this object in the DLL of the least recently used
  67    * pooled connections
  68    */
  69   struct PooledConnection *prev;
  70
  71   /**
  72    * The transport handle to the peer corresponding to this entry; can be NULL
  73    */
  74   struct GNUNET_TRANSPORT_CoreHandle *handle_transport;
  75
  76   /**
  77    * The core handle to the peer corresponding to this entry; can be NULL
  78    */
  79   struct GNUNET_CORE_Handle *handle_core;
  80
  81   /**
  82    * The ATS handle to the peer correspondign to this entry; can be NULL.
  83    */
  84   struct GNUNET_ATS_ConnectivityHandle *handle_ats_connectivity;
  85
  86   /**
  87    * The operation handle for transport handle
  88    */
  89   struct GNUNET_TESTBED_Operation *op_transport;
  90
  91   /**
  92    * The operation handle for core handle
  93    */
  94   struct GNUNET_TESTBED_Operation *op_core;
  95
  96   /**
  97    * The operation handle for ATS handle
  98    */
  99   struct GNUNET_TESTBED_Operation *op_ats_connectivity;
 100
 101   /**
 102    * The peer identity of this peer. Will be set upon opening a connection to
 103    * the peers CORE service. Will be NULL until then and after the CORE
 104    * connection is closed
 105    */
 106   struct GNUNET_PeerIdentity *peer_identity;
 107
 108   /**
 109    * The configuration of the peer. Should be not NULL as long as the
 110    * core_handle or transport_handle are valid
 111    */
 112   struct GNUNET_CONFIGURATION_Handle *cfg;
 113
 114   /**
 115    * DLL head for the queue to serve notifications when a peer is connected
 116    */
 117   struct GST_ConnectionPool_GetHandle *head_notify;
 118
 119   /**
 120    * DLL tail for the queue to serve notifications when a peer is connected
 121    */
 122   struct GST_ConnectionPool_GetHandle *tail_notify;
 123
 124   /**
 125    * DLL head for the queue of #GST_ConnectionPool_GetHandle requests that are
 126    * waiting for this connection to be opened
 127    */
 128   struct GST_ConnectionPool_GetHandle *head_waiting;
 129
 130   /**
 131    * DLL tail for the queue of #GST_ConnectionPool_GetHandle requests that are
 132    * waiting for this connection to be opened
 133    */
 134   struct GST_ConnectionPool_GetHandle *tail_waiting;
 135
 136   /**
 137    * The task to expire this connection from the connection pool
 138    */
 139   struct GNUNET_SCHEDULER_Task *expire_task;
 140
 141   /**
 142    * The task to notify a waiting #GST_ConnectionPool_GetHandle object
 143    */
 144   struct GNUNET_SCHEDULER_Task *notify_task;
 145
 146   /**
 147    * Number of active requests using this pooled connection
 148    */
 149   unsigned int demand;
 150
 151   /**
 152    * Is this entry in LRU
 153    */
 154   int in_lru;
 155
 156   /**
 157    * Is this entry present in the connection pool
 158    */
 159   int in_pool;
 160
 161   /**
 162    * The index of this peer
 163    */
 164   uint32_t index;
 165 };
 166
 167
 168 /**
 169  * The request handle for obtaining a pooled connection
 170  */
 171 struct GST_ConnectionPool_GetHandle {
 172   /**
 173    * The next ptr for inclusion in the notification DLLs.  At first the object
 174    * is placed in the waiting DLL of the corresponding #PooledConnection
 175    * object.  After the handle is opened it is moved to the notification DLL if
 176    * @p connect_notify_cb and @p target are not NULL
 177    */
 178   struct GST_ConnectionPool_GetHandle *next;
 179
 180   /**
 181    * The prev ptr for inclusion in the notification DLLs
 182    */
 183   struct GST_ConnectionPool_GetHandle *prev;
 184
 185   /**
 186    * The pooled connection object this handle corresponds to
 187    */
 188   struct PooledConnection *entry;
 189
 190   /**
 191    * The cache callback to call when a handle is available
 192    */
 193   GST_connection_pool_connection_ready_cb cb;
 194
 195   /**
 196    * The closure for the above callback
 197    */
 198   void *cb_cls;
 199
 200   /**
 201    * The peer identity of the target peer. When this target peer is connected,
 202    * call the notify callback
 203    */
 204   const struct GNUNET_PeerIdentity *target;
 205
 206   /**
 207    * The callback to be called for serving notification that the target peer is
 208    * connected
 209    */
 210   GST_connection_pool_peer_connect_notify connect_notify_cb;
 211
 212   /**
 213    * The closure for the notify callback
 214    */
 215   void *connect_notify_cb_cls;
 216
 217   /**
 218    * The service we want to connect to
 219    */
 220   enum GST_ConnectionPool_Service service;
 221
 222   /**
 223    * Did we call the pool_connection_ready_cb already?
 224    */
 225   int connection_ready_called;
 226
 227   /**
 228    * Are we waiting for any peer connect notifications?
 229    */
 230   int notify_waiting;
 231 };
 232
 233
 234 /**
 235  * A hashmap for quickly finding connections in the connection pool
 236  */
 237 static struct GNUNET_CONTAINER_MultiHashMap32 *map;
 238
 239 /**
 240  * DLL head for maitaining the least recently used #PooledConnection objects.
 241  * The head is the least recently used object.
 242  */
 243 static struct PooledConnection *head_lru;
 244
 245 /**
 246  * DLL tail for maitaining the least recently used #PooledConnection objects
 247  */
 248 static struct PooledConnection *tail_lru;
 249
 250 /**
 251  * DLL head for maintaining #PooledConnection objects that are not added into
 252  * the connection pool as it was full at the time the object's creation
 253  * FIXME
 254  */
 255 static struct PooledConnection *head_not_pooled;
 256
 257 /**
 258  * DLL tail for maintaining #PooledConnection objects that are not added into
 259  * the connection pool as it was full at the time the object's creation
 260  */
 261 static struct PooledConnection *tail_not_pooled;
 262
 263 /**
 264  * The maximum number of entries that can be present in the connection pool
 265  */
 266 static unsigned int max_size;
 267
 268
 269 /**
 270  * Cancel the expiration task of the give #PooledConnection object
 271  *
 272  * @param entry the #PooledConnection object
 273  */
 274 static void
 275 expire_task_cancel(struct PooledConnection *entry);
 276
 277
 278 /**
 279  * Destroy a #PooledConnection object
 280  *
 281  * @param entry the #PooledConnection object
 282  */
 283 static void
 284 destroy_pooled_connection(struct PooledConnection *entry)
 285 {
 286   GNUNET_assert((NULL == entry->head_notify) && (NULL == entry->tail_notify));
 287   GNUNET_assert((NULL == entry->head_waiting) &&
 288                 (NULL == entry->tail_waiting));
 289   GNUNET_assert(0 == entry->demand);
 290   expire_task_cancel(entry);
 291   if (entry->in_lru)
 292     GNUNET_CONTAINER_DLL_remove(head_lru, tail_lru, entry);
 293   if (entry->in_pool)
 294     GNUNET_assert(
 295       GNUNET_OK ==
 296       GNUNET_CONTAINER_multihashmap32_remove(map, entry->index, entry));
 297   if (NULL != entry->notify_task)
 298     {
 299       GNUNET_SCHEDULER_cancel(entry->notify_task);
 300       entry->notify_task = NULL;
 301     }
 302   LOG_DEBUG("Cleaning up handles of a pooled connection\n");
 303   if (NULL != entry->handle_transport)
 304     GNUNET_assert(NULL != entry->op_transport);
 305   if (NULL != entry->op_transport)
 306     {
 307       GNUNET_TESTBED_operation_done(entry->op_transport);
 308       entry->op_transport = NULL;
 309     }
 310   if (NULL != entry->handle_ats_connectivity)
 311     GNUNET_assert(NULL != entry->op_ats_connectivity);
 312   if (NULL != entry->op_ats_connectivity)
 313     {
 314       GNUNET_TESTBED_operation_done(entry->op_ats_connectivity);
 315       entry->op_ats_connectivity = NULL;
 316     }
 317   if (NULL != entry->op_core)
 318     {
 319       GNUNET_TESTBED_operation_done(entry->op_core);
 320       entry->op_core = NULL;
 321     }
 322   GNUNET_assert(NULL == entry->handle_core);
 323   GNUNET_assert(NULL == entry->handle_ats_connectivity);
 324   GNUNET_assert(NULL == entry->handle_transport);
 325   GNUNET_CONFIGURATION_destroy(entry->cfg);
 326   GNUNET_free(entry);
 327 }
 328
 329
 330 /**
 331  * Expire a #PooledConnection object
 332  *
 333  * @param cls the #PooledConnection object
 334  */
 335 static void
 336 expire(void *cls)
 337 {
 338   struct PooledConnection *entry = cls;
 339
 340   entry->expire_task = NULL;
 341   destroy_pooled_connection(entry);
 342 }
 343
 344
 345 /**
 346  * Cancel the expiration task of the give #PooledConnection object
 347  *
 348  * @param entry the #PooledConnection object
 349  */
 350 static void
 351 expire_task_cancel(struct PooledConnection *entry)
 352 {
 353   if (NULL != entry->expire_task)
 354     {
 355       GNUNET_SCHEDULER_cancel(entry->expire_task);
 356       entry->expire_task = NULL;
 357     }
 358 }
 359
 360
 361 /**
 362  * Function to add a #PooledConnection object into LRU and begin the expiry task
 363  *
 364  * @param entry the #PooledConnection object
 365  */
 366 static void
 367 add_to_lru(struct PooledConnection *entry)
 368 {
 369   GNUNET_assert(0 == entry->demand);
 370   GNUNET_assert(!entry->in_lru);
 371   GNUNET_CONTAINER_DLL_insert_tail(head_lru, tail_lru, entry);
 372   entry->in_lru = GNUNET_YES;
 373   GNUNET_assert(NULL == entry->expire_task);
 374   entry->expire_task =
 375     GNUNET_SCHEDULER_add_delayed(CACHE_EXPIRY, &expire, entry);
 376 }
 377
 378
 379 /**
 380  * Function to find a #GST_ConnectionPool_GetHandle which is waiting for one of
 381  * the handles in given entry which are now available.
 382  *
 383  * @param entry the pooled connection whose active list has to be searched
 384  * @param head the starting list element in the GSTCacheGetHandle where the
 385  *          search has to be begin
 386  * @return a suitable GSTCacheGetHandle whose handle ready notify callback
 387  *           hasn't been called yet. NULL if no such suitable GSTCacheGetHandle
 388  *           is found
 389  */
 390 static struct GST_ConnectionPool_GetHandle *
 391 search_waiting(const struct PooledConnection *entry,
 392                struct GST_ConnectionPool_GetHandle *head)
 393 {
 394   struct GST_ConnectionPool_GetHandle *gh;
 395
 396   for (gh = head; NULL != gh; gh = gh->next)
 397     {
 398       switch (gh->service)
 399         {
 400         case GST_CONNECTIONPOOL_SERVICE_CORE:
 401           if (NULL == entry->handle_core)
 402             continue;
 403           if (NULL == entry->peer_identity)
 404             continue; /* CORE connection isn't ready yet */
 405           break;
 406
 407         case GST_CONNECTIONPOOL_SERVICE_TRANSPORT:
 408           if (NULL == entry->handle_transport)
 409             continue;
 410           break;
 411
 412         case GST_CONNECTIONPOOL_SERVICE_ATS_CONNECTIVITY:
 413           if (NULL == entry->handle_ats_connectivity)
 414             continue;
 415           break;
 416         }
 417       break;
 418     }
 419   return gh;
 420 }
 421
 422
 423 /**
 424  * A handle in the #PooledConnection object pointed by @a cls is ready and there
 425  * is a #GST_ConnectionPool_GetHandle object waiting in the waiting list.  This
 426  * function retrieves that object and calls the handle ready callback.  It
 427  * further schedules itself if there are similar waiting objects which can be
 428  * notified.
 429  *
 430  * @param cls the #PooledConnection object
 431  */
 432 static void
 433 connection_ready(void *cls)
 434 {
 435   struct PooledConnection *entry = cls;
 436   struct GST_ConnectionPool_GetHandle *gh;
 437   struct GST_ConnectionPool_GetHandle *gh_next;
 438
 439   GNUNET_assert(NULL != entry->notify_task);
 440   entry->notify_task = NULL;
 441   gh = search_waiting(entry, entry->head_waiting);
 442   GNUNET_assert(NULL != gh);
 443   gh_next = NULL;
 444   if (NULL != gh->next)
 445     gh_next = search_waiting(entry, gh->next);
 446   GNUNET_CONTAINER_DLL_remove(entry->head_waiting, entry->tail_waiting, gh);
 447   gh->connection_ready_called = 1;
 448   if (NULL != gh_next)
 449     entry->notify_task = GNUNET_SCHEDULER_add_now(&connection_ready, entry);
 450   if ((NULL != gh->target) && (NULL != gh->connect_notify_cb))
 451     {
 452       GNUNET_CONTAINER_DLL_insert_tail(entry->head_notify,
 453                                        entry->tail_notify,
 454                                        gh);
 455       gh->notify_waiting = 1;
 456     }
 457   LOG_DEBUG("Connection ready for handle type %u\n", gh->service);
 458   gh->cb(gh->cb_cls,
 459          entry->handle_core,
 460          entry->handle_transport,
 461          entry->handle_ats_connectivity,
 462          entry->peer_identity,
 463          entry->cfg);
 464 }
 465
 466
 467 /**
 468  * Function called from peer connect notify callbacks from CORE and TRANSPORT
 469  * connections. This function calls the pending peer connect notify callbacks
 470  * which are queued in an entry.
 471  *
 472  * @param cls the #PooledConnection object
 473  * @param peer the peer that connected
 474  * @param service the service where this notification has originated
 475  */
 476 static void
 477 peer_connect_notify_cb(void *cls,
 478                        const struct GNUNET_PeerIdentity *peer,
 479                        const enum GST_ConnectionPool_Service service)
 480 {
 481   struct PooledConnection *entry = cls;
 482   struct GST_ConnectionPool_GetHandle *gh;
 483   struct GST_ConnectionPool_GetHandle *gh_next;
 484   GST_connection_pool_peer_connect_notify cb;
 485   void *cb_cls;
 486
 487   for (gh = entry->head_notify; NULL != gh;)
 488     {
 489       GNUNET_assert(NULL != gh->target);
 490       GNUNET_assert(NULL != gh->connect_notify_cb);
 491       GNUNET_assert(gh->connection_ready_called);
 492       if (service != gh->service)
 493         {
 494           gh = gh->next;
 495           continue;
 496         }
 497       if (0 != memcmp(gh->target, peer, sizeof(struct GNUNET_PeerIdentity)))
 498         {
 499           gh = gh->next;
 500           continue;
 501         }
 502       cb = gh->connect_notify_cb;
 503       cb_cls = gh->connect_notify_cb_cls;
 504       gh_next = gh->next;
 505       GNUNET_CONTAINER_DLL_remove(entry->head_notify, entry->tail_notify, gh);
 506       gh->notify_waiting = 0;
 507       LOG_DEBUG("Peer connected to peer %u at service %u\n",
 508                 entry->index,
 509                 gh->service);
 510       gh = gh_next;
 511       cb(cb_cls, peer);
 512     }
 513 }
 514
 515
 516 /**
 517  * Function called to notify transport users that another
 518  * peer connected to us.
 519  *
 520  * @param cls the #PooledConnection object
 521  * @param peer the peer that connected
 522  * @param mq queue for sending data to @a peer
 523  * @return NULL
 524  */
 525 static void *
 526 transport_peer_connect_notify_cb(void *cls,
 527                                  const struct GNUNET_PeerIdentity *peer,
 528                                  struct GNUNET_MQ_Handle *mq)
 529 {
 530   struct PooledConnection *entry = cls;
 531
 532   peer_connect_notify_cb(entry, peer, GST_CONNECTIONPOOL_SERVICE_TRANSPORT);
 533   return NULL;
 534 }
 535
 536
 537 /**
 538  * Function called when resources for opening a connection to TRANSPORT are
 539  * available.
 540  *
 541  * @param cls the #PooledConnection object
 542  */
 543 static void
 544 opstart_get_handle_transport(void *cls)
 545 {
 546   struct PooledConnection *entry = cls;
 547
 548   GNUNET_assert(NULL != entry);
 549   LOG_DEBUG("Opening a transport connection to peer %u\n", entry->index);
 550   entry->handle_transport =
 551     GNUNET_TRANSPORT_core_connect(entry->cfg,
 552                                   NULL,
 553                                   NULL,
 554                                   entry,
 555                                   &transport_peer_connect_notify_cb,
 556                                   NULL,
 557                                   NULL);
 558   if (NULL == entry->handle_transport)
 559     {
 560       GNUNET_break(0);
 561       return;
 562     }
 563   if (0 == entry->demand)
 564     return;
 565   if (NULL != entry->notify_task)
 566     return;
 567   if (NULL != search_waiting(entry, entry->head_waiting))
 568     {
 569       entry->notify_task = GNUNET_SCHEDULER_add_now(&connection_ready, entry);
 570       return;
 571     }
 572 }
 573
 574
 575 /**
 576  * Function called when the operation responsible for opening a TRANSPORT
 577  * connection is marked as done.
 578  *
 579  * @param cls the cache entry
 580  */
 581 static void
 582 oprelease_get_handle_transport(void *cls)
 583 {
 584   struct PooledConnection *entry = cls;
 585
 586   if (NULL == entry->handle_transport)
 587     return;
 588   GNUNET_TRANSPORT_core_disconnect(entry->handle_transport);
 589   entry->handle_transport = NULL;
 590 }
 591
 592
 593 /**
 594  * Method called whenever a given peer connects at CORE level
 595  *
 596  * @param cls the #PooledConnection object
 597  * @param peer peer identity this notification is about
 598  * @param mq message queue for talking to @a peer
 599  * @return peer
 600  */
 601 static void *
 602 core_peer_connect_cb(void *cls,
 603                      const struct GNUNET_PeerIdentity *peer,
 604                      struct GNUNET_MQ_Handle *mq)
 605 {
 606   struct PooledConnection *entry = cls;
 607
 608   peer_connect_notify_cb(entry, peer, GST_CONNECTIONPOOL_SERVICE_CORE);
 609   return (void *)peer;
 610 }
 611
 612
 613 /**
 614  * Function called after #GNUNET_CORE_connect() has succeeded (or failed
 615  * for good).  Note that the private key of the peer is intentionally
 616  * not exposed here; if you need it, your process should try to read
 617  * the private key file directly (which should work if you are
 618  * authorized...).  Implementations of this function must not call
 619  * #GNUNET_CORE_disconnect() (other than by scheduling a new task to
 620  * do this later).
 621  *
 622  * @param cls the #PooledConnection object
 623  * @param my_identity ID of this peer, NULL if we failed
 624  */
 625 static void
 626 core_startup_cb(void *cls, const struct GNUNET_PeerIdentity *my_identity)
 627 {
 628   struct PooledConnection *entry = cls;
 629
 630   if (NULL == my_identity)
 631     {
 632       GNUNET_break(0);
 633       return;
 634     }
 635   GNUNET_assert(NULL == entry->peer_identity);
 636   entry->peer_identity = GNUNET_new(struct GNUNET_PeerIdentity);
 637   *entry->peer_identity = *my_identity;
 638   if (0 == entry->demand)
 639     return;
 640   if (NULL != entry->notify_task)
 641     return;
 642   if (NULL != search_waiting(entry, entry->head_waiting))
 643     {
 644       entry->notify_task = GNUNET_SCHEDULER_add_now(&connection_ready, entry);
 645       return;
 646     }
 647 }
 648
 649
 650 /**
 651  * Function called when resources for opening a connection to CORE are
 652  * available.
 653  *
 654  * @param cls the #PooledConnection object
 655  */
 656 static void
 657 opstart_get_handle_core(void *cls)
 658 {
 659   struct PooledConnection *entry = cls;
 660
 661   GNUNET_assert(NULL != entry);
 662   LOG_DEBUG("Opening a CORE connection to peer %u\n", entry->index);
 663   entry->handle_core =
 664     GNUNET_CORE_connect(entry->cfg,
 665                         entry,  /* closure */
 666                         &core_startup_cb,  /* core startup notify */
 667                         &core_peer_connect_cb,  /* peer connect notify */
 668                         NULL,  /* peer disconnect notify */
 669                         NULL);
 670 }
 671
 672
 673 /**
 674  * Function called when the operation responsible for opening a CORE
 675  * connection is marked as done.
 676  *
 677  * @param cls the #PooledConnection object
 678  */
 679 static void
 680 oprelease_get_handle_core(void *cls)
 681 {
 682   struct PooledConnection *entry = cls;
 683
 684   if (NULL == entry->handle_core)
 685     return;
 686   GNUNET_CORE_disconnect(entry->handle_core);
 687   entry->handle_core = NULL;
 688   GNUNET_free_non_null(entry->peer_identity);
 689   entry->peer_identity = NULL;
 690 }
 691
 692
 693 /**
 694  * Function called when resources for opening a connection to ATS are
 695  * available.
 696  *
 697  * @param cls the #PooledConnection object
 698  */
 699 static void
 700 opstart_get_handle_ats_connectivity(void *cls)
 701 {
 702   struct PooledConnection *entry = cls;
 703
 704   entry->handle_ats_connectivity = GNUNET_ATS_connectivity_init(entry->cfg);
 705 }
 706
 707
 708 /**
 709  * Function called when the operation responsible for opening a ATS
 710  * connection is marked as done.
 711  *
 712  * @param cls the #PooledConnection object
 713  */
 714 static void
 715 oprelease_get_handle_ats_connectivity(void *cls)
 716 {
 717   struct PooledConnection *entry = cls;
 718
 719   if (NULL == entry->handle_ats_connectivity)
 720     return;
 721   GNUNET_ATS_connectivity_done(entry->handle_ats_connectivity);
 722   entry->handle_ats_connectivity = NULL;
 723 }
 724
 725
 726 /**
 727  * This function will be called for every #PooledConnection object in @p map
 728  *
 729  * @param cls NULL
 730  * @param key current key code
 731  * @param value the #PooledConnection object
 732  * @return #GNUNET_YES if we should continue to
 733  *         iterate,
 734  *         #GNUNET_NO if not.
 735  */
 736 static int
 737 cleanup_iterator(void *cls, uint32_t key, void *value)
 738 {
 739   struct PooledConnection *entry = value;
 740
 741   GNUNET_assert(NULL != entry);
 742   destroy_pooled_connection(entry);
 743   return GNUNET_YES;
 744 }
 745
 746
 747 /**
 748  * Initialise the connection pool.
 749  *
 750  * @param size the size of the connection pool.  Each entry in the connection
 751  *   pool can handle a connection to each of the services enumerated in
 752  *   #GST_ConnectionPool_Service
 753  */
 754 void
 755 GST_connection_pool_init(unsigned int size)
 756 {
 757   max_size = size;
 758   if (0 == max_size)
 759     return;
 760   GNUNET_assert(NULL == map);
 761   map = GNUNET_CONTAINER_multihashmap32_create(((size * 3) / 4) + 1);
 762 }
 763
 764
 765 /**
 766  * Cleanup the connection pool
 767  */
 768 void
 769 GST_connection_pool_destroy()
 770 {
 771   struct PooledConnection *entry;
 772
 773   if (NULL != map)
 774     {
 775       GNUNET_assert(
 776         GNUNET_SYSERR !=
 777         GNUNET_CONTAINER_multihashmap32_iterate(map, &cleanup_iterator, NULL));
 778       GNUNET_CONTAINER_multihashmap32_destroy(map);
 779       map = NULL;
 780     }
 781   while (NULL != (entry = head_lru))
 782     {
 783       GNUNET_CONTAINER_DLL_remove(head_lru, tail_lru, entry);
 784       destroy_pooled_connection(entry);
 785     }
 786   GNUNET_assert(NULL == head_not_pooled);
 787 }
 788
 789
 790 /**
 791  * Get a connection handle to @a service.  If the connection is opened before
 792  * and the connection handle is present in the connection pool, it is returned
 793  * through @a cb.  @a peer_id is used for the lookup in the connection pool.  If
 794  * the connection handle is not present in the connection pool, a new connection
 795  * handle is opened for the @a service using @a cfg.  Additionally, @a target,
 796  * @a connect_notify_cb can be specified to get notified when @a target is
 797  * connected at @a service.
 798  *
 799  * @note @a connect_notify_cb will not be called if @a target is
 800  * already connected @a service level. Use
 801  * GNUNET_TRANSPORT_check_peer_connected() or a similar function from the
 802  * respective @a service's API to check if the target peer is already connected
 803  * or not. @a connect_notify_cb will be called only once or never (in case @a
 804  * target cannot be connected or is already connected).
 805  *
 806  * @param peer_id the index of the peer
 807  * @param cfg the configuration with which the transport handle has to be
 808  *          created if it was not present in the cache
 809  * @param service the service of interest
 810  * @param cb the callback to notify when the transport handle is available
 811  * @param cb_cls the closure for @a cb
 812  * @param target the peer identify of the peer whose connection to our TRANSPORT
 813  *          subsystem will be notified through the @a connect_notify_cb. Can be
 814  * NULL
 815  * @param connect_notify_cb the callback to call when the @a target peer is
 816  *          connected. This callback will only be called once or never again (in
 817  *          case the target peer cannot be connected). Can be NULL
 818  * @param connect_notify_cb_cls the closure for @a connect_notify_cb
 819  * @return the handle which can be used cancel or mark that the handle is no
 820  *           longer being used
 821  */
 822 struct GST_ConnectionPool_GetHandle *
 823 GST_connection_pool_get_handle(
 824   unsigned int peer_id,
 825   const struct GNUNET_CONFIGURATION_Handle *cfg,
 826   enum GST_ConnectionPool_Service service,
 827   GST_connection_pool_connection_ready_cb cb,
 828   void *cb_cls,
 829   const struct GNUNET_PeerIdentity *target,
 830   GST_connection_pool_peer_connect_notify connect_notify_cb,
 831   void *connect_notify_cb_cls)
 832 {
 833   struct GST_ConnectionPool_GetHandle *gh;
 834   struct PooledConnection *entry;
 835   struct GNUNET_TESTBED_Operation *op;
 836   void *handle;
 837   uint32_t peer_id32;
 838
 839   peer_id32 = (uint32_t)peer_id;
 840   handle = NULL;
 841   entry = NULL;
 842   if (NULL != map)
 843     entry = GNUNET_CONTAINER_multihashmap32_get(map, peer_id32);
 844   if (NULL != entry)
 845     {
 846       if (entry->in_lru)
 847         {
 848           GNUNET_assert(0 == entry->demand);
 849           expire_task_cancel(entry);
 850           GNUNET_CONTAINER_DLL_remove(head_lru, tail_lru, entry);
 851           entry->in_lru = GNUNET_NO;
 852         }
 853       switch (service)
 854         {
 855         case GST_CONNECTIONPOOL_SERVICE_TRANSPORT:
 856           handle = entry->handle_transport;
 857           if (NULL != handle)
 858             LOG_DEBUG("Found TRANSPORT handle for peer %u\n", entry->index);
 859           break;
 860
 861         case GST_CONNECTIONPOOL_SERVICE_CORE:
 862           handle = entry->handle_core;
 863           if (NULL != handle)
 864             LOG_DEBUG("Found CORE handle for peer %u\n", entry->index);
 865           break;
 866
 867         case GST_CONNECTIONPOOL_SERVICE_ATS_CONNECTIVITY:
 868           handle = entry->handle_ats_connectivity;
 869           if (NULL != handle)
 870             LOG_DEBUG("Found ATS CONNECTIVITY handle for peer %u\n", entry->index);
 871           break;
 872         }
 873     }
 874   else
 875     {
 876       entry = GNUNET_new(struct PooledConnection);
 877       entry->index = peer_id32;
 878       if ((NULL != map) &&
 879           (GNUNET_CONTAINER_multihashmap32_size(map) < max_size))
 880         {
 881           GNUNET_assert(GNUNET_OK ==
 882                         GNUNET_CONTAINER_multihashmap32_put(
 883                           map,
 884                           entry->index,
 885                           entry,
 886                           GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
 887           entry->in_pool = GNUNET_YES;
 888         }
 889       else
 890         {
 891           GNUNET_CONTAINER_DLL_insert_tail(head_not_pooled,
 892                                            tail_not_pooled,
 893                                            entry);
 894         }
 895       entry->cfg = GNUNET_CONFIGURATION_dup(cfg);
 896     }
 897   entry->demand++;
 898   gh = GNUNET_new(struct GST_ConnectionPool_GetHandle);
 899   gh->entry = entry;
 900   gh->cb = cb;
 901   gh->cb_cls = cb_cls;
 902   gh->target = target;
 903   gh->connect_notify_cb = connect_notify_cb;
 904   gh->connect_notify_cb_cls = connect_notify_cb_cls;
 905   gh->service = service;
 906   GNUNET_CONTAINER_DLL_insert(entry->head_waiting, entry->tail_waiting, gh);
 907   if (NULL != handle)
 908     {
 909       if (NULL == entry->notify_task)
 910         {
 911           if (NULL != search_waiting(entry, entry->head_waiting))
 912             entry->notify_task =
 913               GNUNET_SCHEDULER_add_now(&connection_ready, entry);
 914         }
 915       return gh;
 916     }
 917   op = NULL;
 918   switch (gh->service)
 919     {
 920     case GST_CONNECTIONPOOL_SERVICE_TRANSPORT:
 921       if (NULL != entry->op_transport)
 922         return gh; /* Operation pending */
 923       op = GNUNET_TESTBED_operation_create_(entry,
 924                                             &opstart_get_handle_transport,
 925                                             &oprelease_get_handle_transport);
 926       entry->op_transport = op;
 927       break;
 928
 929     case GST_CONNECTIONPOOL_SERVICE_CORE:
 930       if (NULL != entry->op_core)
 931         return gh; /* Operation pending */
 932       op = GNUNET_TESTBED_operation_create_(entry,
 933                                             &opstart_get_handle_core,
 934                                             &oprelease_get_handle_core);
 935       entry->op_core = op;
 936       break;
 937
 938     case GST_CONNECTIONPOOL_SERVICE_ATS_CONNECTIVITY:
 939       if (NULL != entry->op_ats_connectivity)
 940         return gh; /* Operation pending */
 941       op =
 942         GNUNET_TESTBED_operation_create_(entry,
 943                                          &opstart_get_handle_ats_connectivity,
 944                                          &oprelease_get_handle_ats_connectivity);
 945       entry->op_ats_connectivity = op;
 946       break;
 947     }
 948   GNUNET_TESTBED_operation_queue_insert_(GST_opq_openfds, op);
 949   GNUNET_TESTBED_operation_begin_wait_(op);
 950   return gh;
 951 }
 952
 953
 954 /**
 955  * Relinquish a #GST_ConnectionPool_GetHandle object.  If the connection
 956  * associated with the object is currently being used by other
 957  * #GST_ConnectionPool_GetHandle objects, it is left in the connection pool.  If
 958  * no other objects are using the connection and the connection pool is not full
 959  * then it is placed in a LRU queue.  If the connection pool is full, then
 960  * connections from the LRU queue are evicted and closed to create place for
 961  * this connection.  If the connection pool if full and the LRU queue is empty,
 962  * then the connection is closed.
 963  *
 964  * @param gh the handle
 965  */
 966 void
 967 GST_connection_pool_get_handle_done(struct GST_ConnectionPool_GetHandle *gh)
 968 {
 969   struct PooledConnection *entry;
 970
 971   if (NULL == gh)
 972     return;
 973   entry = gh->entry;
 974   LOG_DEBUG("Cleaning up get handle %p for service %u, peer %u\n",
 975             gh,
 976             gh->service,
 977             entry->index);
 978   if (!gh->connection_ready_called)
 979     {
 980       GNUNET_CONTAINER_DLL_remove(entry->head_waiting, entry->tail_waiting, gh);
 981       if ((NULL == search_waiting(entry, entry->head_waiting)) &&
 982           (NULL != entry->notify_task))
 983         {
 984           GNUNET_SCHEDULER_cancel(entry->notify_task);
 985           entry->notify_task = NULL;
 986         }
 987     }
 988   if (gh->notify_waiting)
 989     {
 990       GNUNET_CONTAINER_DLL_remove(entry->head_notify, entry->tail_notify, gh);
 991       gh->notify_waiting = 0;
 992     }
 993   GNUNET_free(gh);
 994   gh = NULL;
 995   GNUNET_assert(!entry->in_lru);
 996   if (!entry->in_pool)
 997     GNUNET_CONTAINER_DLL_remove(head_not_pooled, tail_not_pooled, entry);
 998   if (NULL != map)
 999     {
1000       if (GNUNET_YES ==
1001           GNUNET_CONTAINER_multihashmap32_contains(map, entry->index))
1002         goto unallocate;
1003       if (GNUNET_CONTAINER_multihashmap32_size(map) == max_size)
1004         {
1005           if (NULL == head_lru)
1006             goto unallocate;
1007           destroy_pooled_connection(head_lru);
1008         }
1009       GNUNET_assert(GNUNET_OK ==
1010                     GNUNET_CONTAINER_multihashmap32_put(
1011                       map,
1012                       entry->index,
1013                       entry,
1014                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
1015       entry->in_pool = GNUNET_YES;
1016     }
1017
1018 unallocate:
1019   GNUNET_assert(0 < entry->demand);
1020   entry->demand--;
1021   if (0 != entry->demand)
1022     return;
1023   if (entry->in_pool)
1024     {
1025       add_to_lru(entry);
1026       return;
1027     }
1028   destroy_pooled_connection(entry);
1029 }