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     case GST_CONNECTIONPOOL_SERVICE_TRANSPORT:
 409       if (NULL == entry->handle_transport)
 410         continue;
 411       break;
 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     case GST_CONNECTIONPOOL_SERVICE_CORE:
 861       handle = entry->handle_core;
 862       if (NULL != handle)
 863         LOG_DEBUG ("Found CORE handle for peer %u\n", entry->index);
 864       break;
 865     case GST_CONNECTIONPOOL_SERVICE_ATS_CONNECTIVITY:
 866       handle = entry->handle_ats_connectivity;
 867       if (NULL != handle)
 868         LOG_DEBUG ("Found ATS CONNECTIVITY handle for peer %u\n", entry->index);
 869       break;
 870     }
 871   }
 872   else
 873   {
 874     entry = GNUNET_new (struct PooledConnection);
 875     entry->index = peer_id32;
 876     if ((NULL != map) &&
 877         (GNUNET_CONTAINER_multihashmap32_size (map) < max_size))
 878     {
 879       GNUNET_assert (GNUNET_OK ==
 880                      GNUNET_CONTAINER_multihashmap32_put (
 881                        map,
 882                        entry->index,
 883                        entry,
 884                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
 885       entry->in_pool = GNUNET_YES;
 886     }
 887     else
 888     {
 889       GNUNET_CONTAINER_DLL_insert_tail (head_not_pooled,
 890                                         tail_not_pooled,
 891                                         entry);
 892     }
 893     entry->cfg = GNUNET_CONFIGURATION_dup (cfg);
 894   }
 895   entry->demand++;
 896   gh = GNUNET_new (struct GST_ConnectionPool_GetHandle);
 897   gh->entry = entry;
 898   gh->cb = cb;
 899   gh->cb_cls = cb_cls;
 900   gh->target = target;
 901   gh->connect_notify_cb = connect_notify_cb;
 902   gh->connect_notify_cb_cls = connect_notify_cb_cls;
 903   gh->service = service;
 904   GNUNET_CONTAINER_DLL_insert (entry->head_waiting, entry->tail_waiting, gh);
 905   if (NULL != handle)
 906   {
 907     if (NULL == entry->notify_task)
 908     {
 909       if (NULL != search_waiting (entry, entry->head_waiting))
 910         entry->notify_task =
 911           GNUNET_SCHEDULER_add_now (&connection_ready, entry);
 912     }
 913     return gh;
 914   }
 915   op = NULL;
 916   switch (gh->service)
 917   {
 918   case GST_CONNECTIONPOOL_SERVICE_TRANSPORT:
 919     if (NULL != entry->op_transport)
 920       return gh; /* Operation pending */
 921     op = GNUNET_TESTBED_operation_create_ (entry,
 922                                            &opstart_get_handle_transport,
 923                                            &oprelease_get_handle_transport);
 924     entry->op_transport = op;
 925     break;
 926   case GST_CONNECTIONPOOL_SERVICE_CORE:
 927     if (NULL != entry->op_core)
 928       return gh; /* Operation pending */
 929     op = GNUNET_TESTBED_operation_create_ (entry,
 930                                            &opstart_get_handle_core,
 931                                            &oprelease_get_handle_core);
 932     entry->op_core = op;
 933     break;
 934   case GST_CONNECTIONPOOL_SERVICE_ATS_CONNECTIVITY:
 935     if (NULL != entry->op_ats_connectivity)
 936       return gh; /* Operation pending */
 937     op =
 938       GNUNET_TESTBED_operation_create_ (entry,
 939                                         &opstart_get_handle_ats_connectivity,
 940                                         &oprelease_get_handle_ats_connectivity);
 941     entry->op_ats_connectivity = op;
 942     break;
 943   }
 944   GNUNET_TESTBED_operation_queue_insert_ (GST_opq_openfds, op);
 945   GNUNET_TESTBED_operation_begin_wait_ (op);
 946   return gh;
 947 }
 948
 949
 950 /**
 951  * Relinquish a #GST_ConnectionPool_GetHandle object.  If the connection
 952  * associated with the object is currently being used by other
 953  * #GST_ConnectionPool_GetHandle objects, it is left in the connection pool.  If
 954  * no other objects are using the connection and the connection pool is not full
 955  * then it is placed in a LRU queue.  If the connection pool is full, then
 956  * connections from the LRU queue are evicted and closed to create place for
 957  * this connection.  If the connection pool if full and the LRU queue is empty,
 958  * then the connection is closed.
 959  *
 960  * @param gh the handle
 961  */
 962 void
 963 GST_connection_pool_get_handle_done (struct GST_ConnectionPool_GetHandle *gh)
 964 {
 965   struct PooledConnection *entry;
 966
 967   if (NULL == gh)
 968     return;
 969   entry = gh->entry;
 970   LOG_DEBUG ("Cleaning up get handle %p for service %u, peer %u\n",
 971              gh,
 972              gh->service,
 973              entry->index);
 974   if (! gh->connection_ready_called)
 975   {
 976     GNUNET_CONTAINER_DLL_remove (entry->head_waiting, entry->tail_waiting, gh);
 977     if ((NULL == search_waiting (entry, entry->head_waiting)) &&
 978         (NULL != entry->notify_task))
 979     {
 980       GNUNET_SCHEDULER_cancel (entry->notify_task);
 981       entry->notify_task = NULL;
 982     }
 983   }
 984   if (gh->notify_waiting)
 985   {
 986     GNUNET_CONTAINER_DLL_remove (entry->head_notify, entry->tail_notify, gh);
 987     gh->notify_waiting = 0;
 988   }
 989   GNUNET_free (gh);
 990   gh = NULL;
 991   GNUNET_assert (! entry->in_lru);
 992   if (! entry->in_pool)
 993     GNUNET_CONTAINER_DLL_remove (head_not_pooled, tail_not_pooled, entry);
 994   if (NULL != map)
 995   {
 996     if (GNUNET_YES ==
 997         GNUNET_CONTAINER_multihashmap32_contains (map, entry->index))
 998       goto unallocate;
 999     if (GNUNET_CONTAINER_multihashmap32_size (map) == max_size)
1000     {
1001       if (NULL == head_lru)
1002         goto unallocate;
1003       destroy_pooled_connection (head_lru);
1004     }
1005     GNUNET_assert (GNUNET_OK ==
1006                    GNUNET_CONTAINER_multihashmap32_put (
1007                      map,
1008                      entry->index,
1009                      entry,
1010                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
1011     entry->in_pool = GNUNET_YES;
1012   }
1013
1014 unallocate:
1015   GNUNET_assert (0 < entry->demand);
1016   entry->demand--;
1017   if (0 != entry->demand)
1018     return;
1019   if (entry->in_pool)
1020   {
1021     add_to_lru (entry);
1022     return;
1023   }
1024   destroy_pooled_connection (entry);
1025 }