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