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_core_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 core_handle
 111    * 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) && (NULL ==
 290                                                    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 (GNUNET_OK ==
 297                    GNUNET_CONTAINER_multihashmap32_remove (map,
 298                                                            entry->index,
 299                                                            entry));
 300   if (NULL != entry->notify_task)
 301   {
 302     GNUNET_SCHEDULER_cancel (entry->notify_task);
 303     entry->notify_task = NULL;
 304   }
 305   LOG_DEBUG ("Cleaning up handles of a pooled connection\n");
 306   if (NULL != entry->handle_transport)
 307     GNUNET_assert (NULL != entry->op_transport);
 308   if (NULL != entry->op_transport)
 309   {
 310     GNUNET_TESTBED_operation_done (entry->op_transport);
 311     entry->op_transport = NULL;
 312   }
 313   if (NULL != entry->handle_ats_connectivity)
 314     GNUNET_assert (NULL != entry->op_ats_connectivity);
 315   if (NULL != entry->op_ats_connectivity)
 316   {
 317     GNUNET_TESTBED_operation_done (entry->op_ats_connectivity);
 318     entry->op_ats_connectivity = NULL;
 319   }
 320   if (NULL != entry->op_core)
 321   {
 322     GNUNET_TESTBED_operation_done (entry->op_core);
 323     entry->op_core = NULL;
 324   }
 325   GNUNET_assert (NULL == entry->handle_core);
 326   GNUNET_assert (NULL == entry->handle_ats_connectivity);
 327   GNUNET_assert (NULL == entry->handle_transport);
 328   GNUNET_CONFIGURATION_destroy (entry->cfg);
 329   GNUNET_free (entry);
 330 }
 331
 332
 333 /**
 334  * Expire a #PooledConnection object
 335  *
 336  * @param cls the #PooledConnection object
 337  */
 338 static void
 339 expire (void *cls)
 340 {
 341   struct PooledConnection *entry = cls;
 342
 343   entry->expire_task = NULL;
 344   destroy_pooled_connection (entry);
 345 }
 346
 347
 348 /**
 349  * Cancel the expiration task of the give #PooledConnection object
 350  *
 351  * @param entry the #PooledConnection object
 352  */
 353 static void
 354 expire_task_cancel (struct PooledConnection *entry)
 355 {
 356   if (NULL != entry->expire_task)
 357   {
 358     GNUNET_SCHEDULER_cancel (entry->expire_task);
 359     entry->expire_task = NULL;
 360   }
 361 }
 362
 363
 364 /**
 365  * Function to add a #PooledConnection object into LRU and begin the expiry task
 366  *
 367  * @param entry the #PooledConnection object
 368  */
 369 static void
 370 add_to_lru (struct PooledConnection *entry)
 371 {
 372   GNUNET_assert (0 == entry->demand);
 373   GNUNET_assert (!entry->in_lru);
 374   GNUNET_CONTAINER_DLL_insert_tail (head_lru, tail_lru, entry);
 375   entry->in_lru = GNUNET_YES;
 376   GNUNET_assert (NULL == entry->expire_task);
 377   entry->expire_task = GNUNET_SCHEDULER_add_delayed (CACHE_EXPIRY,
 378                                                      &expire, entry);
 379 }
 380
 381
 382 /**
 383  * Function to find a #GST_ConnectionPool_GetHandle which is waiting for one of
 384  * the handles in given entry which are now available.
 385  *
 386  * @param entry the pooled connection whose active list has to be searched
 387  * @param head the starting list element in the GSTCacheGetHandle where the
 388  *          search has to be begin
 389  * @return a suitable GSTCacheGetHandle whose handle ready notify callback
 390  *           hasn't been called yet. NULL if no such suitable GSTCacheGetHandle
 391  *           is found
 392  */
 393 static struct GST_ConnectionPool_GetHandle *
 394 search_waiting (const struct PooledConnection *entry,
 395                 struct GST_ConnectionPool_GetHandle *head)
 396 {
 397   struct GST_ConnectionPool_GetHandle *gh;
 398
 399   for (gh = head; NULL != gh; gh = gh->next)
 400   {
 401     switch (gh->service)
 402     {
 403     case GST_CONNECTIONPOOL_SERVICE_CORE:
 404       if (NULL == entry->handle_core)
 405         continue;
 406       if (NULL == entry->peer_identity)
 407         continue;               /* CORE connection isn't ready yet */
 408       break;
 409     case GST_CONNECTIONPOOL_SERVICE_TRANSPORT:
 410       if (NULL == entry->handle_transport)
 411         continue;
 412       break;
 413     case GST_CONNECTIONPOOL_SERVICE_ATS_CONNECTIVITY:
 414       if (NULL == entry->handle_ats_connectivity)
 415         continue;
 416       break;
 417     }
 418     break;
 419   }
 420   return gh;
 421 }
 422
 423
 424 /**
 425  * A handle in the #PooledConnection object pointed by @a cls is ready and there
 426  * is a #GST_ConnectionPool_GetHandle object waiting in the waiting list.  This
 427  * function retrieves that object and calls the handle ready callback.  It
 428  * further schedules itself if there are similar waiting objects which can be 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,
 447                                entry->tail_waiting,
 448                                gh);
 449   gh->connection_ready_called = 1;
 450   if (NULL != gh_next)
 451     entry->notify_task = GNUNET_SCHEDULER_add_now (&connection_ready,
 452                                                    entry);
 453   if ( (NULL != gh->target) &&
 454        (NULL != gh->connect_notify_cb) )
 455   {
 456     GNUNET_CONTAINER_DLL_insert_tail (entry->head_notify,
 457                                       entry->tail_notify,
 458                                       gh);
 459     gh->notify_waiting = 1;
 460   }
 461   LOG_DEBUG ("Connection ready for handle type %u\n",
 462              gh->service);
 463   gh->cb (gh->cb_cls,
 464           entry->handle_core,
 465           entry->handle_transport,
 466           entry->handle_ats_connectivity,
 467           entry->peer_identity,
 468           entry->cfg);
 469 }
 470
 471
 472 /**
 473  * Function called from peer connect notify callbacks from CORE and TRANSPORT
 474  * connections. This function calls the pending peer connect notify callbacks
 475  * which are queued in an entry.
 476  *
 477  * @param cls the #PooledConnection object
 478  * @param peer the peer that connected
 479  * @param service the service where this notification has originated
 480  */
 481 static void
 482 peer_connect_notify_cb (void *cls,
 483                         const struct GNUNET_PeerIdentity *peer,
 484                         const enum GST_ConnectionPool_Service service)
 485 {
 486   struct PooledConnection *entry = cls;
 487   struct GST_ConnectionPool_GetHandle *gh;
 488   struct GST_ConnectionPool_GetHandle *gh_next;
 489   GST_connection_pool_peer_connect_notify cb;
 490   void *cb_cls;
 491
 492   for (gh = entry->head_notify; NULL != gh;)
 493   {
 494     GNUNET_assert (NULL != gh->target);
 495     GNUNET_assert (NULL != gh->connect_notify_cb);
 496     GNUNET_assert (gh->connection_ready_called);
 497     if (service != gh->service)
 498     {
 499       gh = gh->next;
 500       continue;
 501     }
 502     if (0 != memcmp (gh->target,
 503                      peer,
 504                      sizeof (struct GNUNET_PeerIdentity)))
 505     {
 506       gh = gh->next;
 507       continue;
 508     }
 509     cb = gh->connect_notify_cb;
 510     cb_cls = gh->connect_notify_cb_cls;
 511     gh_next = gh->next;
 512     GNUNET_CONTAINER_DLL_remove (entry->head_notify, entry->tail_notify, gh);
 513     gh->notify_waiting = 0;
 514     LOG_DEBUG ("Peer connected to peer %u at service %u\n",
 515                entry->index,
 516                gh->service);
 517     gh = gh_next;
 518     cb (cb_cls, peer);
 519   }
 520 }
 521
 522
 523 /**
 524  * Function called to notify transport users that another
 525  * peer connected to us.
 526  *
 527  * @param cls the #PooledConnection object
 528  * @param peer the peer that connected
 529  * @param mq queue for sending data to @a peer
 530  * @return NULL
 531  */
 532 static void *
 533 transport_peer_connect_notify_cb (void *cls,
 534                                   const struct GNUNET_PeerIdentity *peer,
 535                                   struct GNUNET_MQ_Handle *mq)
 536 {
 537   struct PooledConnection *entry = cls;
 538
 539   peer_connect_notify_cb (entry,
 540                           peer,
 541                           GST_CONNECTIONPOOL_SERVICE_TRANSPORT);
 542   return NULL;
 543 }
 544
 545
 546 /**
 547  * Function called when resources for opening a connection to TRANSPORT are
 548  * available.
 549  *
 550  * @param cls the #PooledConnection object
 551  */
 552 static void
 553 opstart_get_handle_transport (void *cls)
 554 {
 555   struct PooledConnection *entry = cls;
 556
 557   GNUNET_assert (NULL != entry);
 558   LOG_DEBUG ("Opening a transport connection to peer %u\n",
 559              entry->index);
 560   entry->handle_transport =
 561       GNUNET_TRANSPORT_core_connect (entry->cfg,
 562                                      NULL,
 563                                      NULL,
 564                                      entry,
 565                                      &transport_peer_connect_notify_cb,
 566                                      NULL,
 567                                      NULL);
 568   if (NULL == entry->handle_transport)
 569   {
 570     GNUNET_break (0);
 571     return;
 572   }
 573   if (0 == entry->demand)
 574     return;
 575   if (NULL != entry->notify_task)
 576     return;
 577   if (NULL != search_waiting (entry, entry->head_waiting))
 578   {
 579     entry->notify_task = GNUNET_SCHEDULER_add_now (&connection_ready, entry);
 580     return;
 581   }
 582 }
 583
 584
 585 /**
 586  * Function called when the operation responsible for opening a TRANSPORT
 587  * connection is marked as done.
 588  *
 589  * @param cls the cache entry
 590  */
 591 static void
 592 oprelease_get_handle_transport (void *cls)
 593 {
 594   struct PooledConnection *entry = cls;
 595
 596   if (NULL == entry->handle_transport)
 597     return;
 598   GNUNET_TRANSPORT_core_disconnect (entry->handle_transport);
 599   entry->handle_transport = NULL;
 600 }
 601
 602
 603 /**
 604  * Method called whenever a given peer connects at CORE level
 605  *
 606  * @param cls the #PooledConnection object
 607  * @param peer peer identity this notification is about
 608  * @param mq message queue for talking to @a peer
 609  * @return peer
 610  */
 611 static void *
 612 core_peer_connect_cb (void *cls,
 613                       const struct GNUNET_PeerIdentity *peer,
 614                       struct GNUNET_MQ_Handle *mq)
 615 {
 616   struct PooledConnection *entry = cls;
 617
 618   peer_connect_notify_cb (entry,
 619                           peer,
 620                           GST_CONNECTIONPOOL_SERVICE_CORE);
 621   return (void *) peer;
 622 }
 623
 624
 625 /**
 626  * Function called after #GNUNET_CORE_connect() has succeeded (or failed
 627  * for good).  Note that the private key of the peer is intentionally
 628  * not exposed here; if you need it, your process should try to read
 629  * the private key file directly (which should work if you are
 630  * authorized...).  Implementations of this function must not call
 631  * #GNUNET_CORE_disconnect() (other than by scheduling a new task to
 632  * do this later).
 633  *
 634  * @param cls the #PooledConnection object
 635  * @param my_identity ID of this peer, NULL if we failed
 636  */
 637 static void
 638 core_startup_cb (void *cls,
 639                  const struct GNUNET_PeerIdentity *my_identity)
 640 {
 641   struct PooledConnection *entry = cls;
 642
 643   if (NULL == my_identity)
 644   {
 645     GNUNET_break (0);
 646     return;
 647   }
 648   GNUNET_assert (NULL == entry->peer_identity);
 649   entry->peer_identity = GNUNET_new (struct GNUNET_PeerIdentity);
 650   *entry->peer_identity = *my_identity;
 651   if (0 == entry->demand)
 652     return;
 653   if (NULL != entry->notify_task)
 654     return;
 655   if (NULL != search_waiting (entry, entry->head_waiting))
 656   {
 657     entry->notify_task = GNUNET_SCHEDULER_add_now (&connection_ready, entry);
 658     return;
 659   }
 660 }
 661
 662
 663 /**
 664  * Function called when resources for opening a connection to CORE are
 665  * available.
 666  *
 667  * @param cls the #PooledConnection object
 668  */
 669 static void
 670 opstart_get_handle_core (void *cls)
 671 {
 672   struct PooledConnection *entry = cls;
 673
 674   GNUNET_assert (NULL != entry);
 675   LOG_DEBUG ("Opening a CORE connection to peer %u\n",
 676              entry->index);
 677   entry->handle_core
 678     = GNUNET_CORE_connect (entry->cfg,
 679                            entry,        /* closure */
 680                            &core_startup_cb, /* core startup notify */
 681                            &core_peer_connect_cb,    /* peer connect notify */
 682                            NULL,     /* peer disconnect notify */
 683                            NULL);
 684 }
 685
 686
 687 /**
 688  * Function called when the operation responsible for opening a CORE
 689  * connection is marked as done.
 690  *
 691  * @param cls the #PooledConnection object
 692  */
 693 static void
 694 oprelease_get_handle_core (void *cls)
 695 {
 696   struct PooledConnection *entry = cls;
 697
 698   if (NULL == entry->handle_core)
 699     return;
 700   GNUNET_CORE_disconnect (entry->handle_core);
 701   entry->handle_core = NULL;
 702   GNUNET_free_non_null (entry->peer_identity);
 703   entry->peer_identity = NULL;
 704 }
 705
 706
 707 /**
 708  * Function called when resources for opening a connection to ATS are
 709  * available.
 710  *
 711  * @param cls the #PooledConnection object
 712  */
 713 static void
 714 opstart_get_handle_ats_connectivity (void *cls)
 715 {
 716   struct PooledConnection *entry = cls;
 717
 718   entry->handle_ats_connectivity =
 719     GNUNET_ATS_connectivity_init (entry->cfg);
 720 }
 721
 722
 723 /**
 724  * Function called when the operation responsible for opening a ATS
 725  * connection is marked as done.
 726  *
 727  * @param cls the #PooledConnection object
 728  */
 729 static void
 730 oprelease_get_handle_ats_connectivity (void *cls)
 731 {
 732   struct PooledConnection *entry = cls;
 733
 734   if (NULL == entry->handle_ats_connectivity)
 735     return;
 736   GNUNET_ATS_connectivity_done (entry->handle_ats_connectivity);
 737   entry->handle_ats_connectivity = NULL;
 738 }
 739
 740
 741 /**
 742  * This function will be called for every #PooledConnection object in @p map
 743  *
 744  * @param cls NULL
 745  * @param key current key code
 746  * @param value the #PooledConnection object
 747  * @return #GNUNET_YES if we should continue to
 748  *         iterate,
 749  *         #GNUNET_NO if not.
 750  */
 751 static int
 752 cleanup_iterator (void *cls,
 753                   uint32_t key,
 754                   void *value)
 755 {
 756   struct PooledConnection *entry = value;
 757
 758   GNUNET_assert (NULL != entry);
 759   destroy_pooled_connection (entry);
 760   return GNUNET_YES;
 761 }
 762
 763
 764 /**
 765  * Initialise the connection pool.
 766  *
 767  * @param size the size of the connection pool.  Each entry in the connection
 768  *   pool can handle a connection to each of the services enumerated in
 769  *   #GST_ConnectionPool_Service
 770  */
 771 void
 772 GST_connection_pool_init (unsigned int size)
 773 {
 774   max_size = size;
 775   if (0 == max_size)
 776     return;
 777   GNUNET_assert (NULL == map);
 778   map = GNUNET_CONTAINER_multihashmap32_create (((size * 3) / 4) + 1);
 779 }
 780
 781
 782 /**
 783  * Cleanup the connection pool
 784  */
 785 void
 786 GST_connection_pool_destroy ()
 787 {
 788   struct PooledConnection *entry;
 789
 790   if (NULL != map)
 791   {
 792     GNUNET_assert (GNUNET_SYSERR !=
 793                    GNUNET_CONTAINER_multihashmap32_iterate (map,
 794                                                             &cleanup_iterator,
 795                                                             NULL));
 796     GNUNET_CONTAINER_multihashmap32_destroy (map);
 797     map = NULL;
 798   }
 799   while (NULL != (entry = head_lru))
 800   {
 801     GNUNET_CONTAINER_DLL_remove (head_lru, tail_lru, entry);
 802     destroy_pooled_connection (entry);
 803   }
 804   GNUNET_assert (NULL == head_not_pooled);
 805 }
 806
 807
 808 /**
 809  * Get a connection handle to @a service.  If the connection is opened before
 810  * and the connection handle is present in the connection pool, it is returned
 811  * through @a cb.  @a peer_id is used for the lookup in the connection pool.  If
 812  * the connection handle is not present in the connection pool, a new connection
 813  * handle is opened for the @a service using @a cfg.  Additionally, @a target,
 814  * @a connect_notify_cb can be specified to get notified when @a target is
 815  * connected at @a service.
 816  *
 817  * @note @a connect_notify_cb will not be called if @a target is
 818  * already connected @a service level. Use
 819  * GNUNET_TRANSPORT_check_peer_connected() or a similar function from the
 820  * respective @a service's API to check if the target peer is already connected or
 821  * not. @a connect_notify_cb will be called only once or never (in case @a target
 822  * cannot be connected or is already connected).
 823  *
 824  * @param peer_id the index of the peer
 825  * @param cfg the configuration with which the transport handle has to be
 826  *          created if it was not present in the cache
 827  * @param service the service of interest
 828  * @param cb the callback to notify when the transport handle is available
 829  * @param cb_cls the closure for @a cb
 830  * @param target the peer identify of the peer whose connection to our TRANSPORT
 831  *          subsystem will be notified through the @a connect_notify_cb. Can be NULL
 832  * @param connect_notify_cb the callback to call when the @a target peer is
 833  *          connected. This callback will only be called once or never again (in
 834  *          case the target peer cannot be connected). Can be NULL
 835  * @param connect_notify_cb_cls the closure for @a connect_notify_cb
 836  * @return the handle which can be used cancel or mark that the handle is no
 837  *           longer being used
 838  */
 839 struct GST_ConnectionPool_GetHandle *
 840 GST_connection_pool_get_handle (unsigned int peer_id,
 841                                 const struct GNUNET_CONFIGURATION_Handle *cfg,
 842                                 enum GST_ConnectionPool_Service service,
 843                                 GST_connection_pool_connection_ready_cb cb,
 844                                 void *cb_cls,
 845                                 const struct GNUNET_PeerIdentity *target,
 846                                 GST_connection_pool_peer_connect_notify connect_notify_cb,
 847                                 void *connect_notify_cb_cls)
 848 {
 849   struct GST_ConnectionPool_GetHandle *gh;
 850   struct PooledConnection *entry;
 851   struct GNUNET_TESTBED_Operation *op;
 852   void *handle;
 853   uint32_t peer_id32;
 854
 855   peer_id32 = (uint32_t) peer_id;
 856   handle = NULL;
 857   entry = NULL;
 858   if (NULL != map)
 859     entry = GNUNET_CONTAINER_multihashmap32_get (map, peer_id32);
 860   if (NULL != entry)
 861   {
 862     if (entry->in_lru)
 863     {
 864       GNUNET_assert (0 == entry->demand);
 865       expire_task_cancel (entry);
 866       GNUNET_CONTAINER_DLL_remove (head_lru, tail_lru, entry);
 867       entry->in_lru = GNUNET_NO;
 868     }
 869     switch (service)
 870     {
 871     case GST_CONNECTIONPOOL_SERVICE_TRANSPORT:
 872       handle = entry->handle_transport;
 873       if (NULL != handle)
 874         LOG_DEBUG ("Found TRANSPORT handle for peer %u\n",
 875                    entry->index);
 876       break;
 877     case GST_CONNECTIONPOOL_SERVICE_CORE:
 878       handle = entry->handle_core;
 879       if (NULL != handle)
 880         LOG_DEBUG ("Found CORE handle for peer %u\n",
 881                    entry->index);
 882       break;
 883     case GST_CONNECTIONPOOL_SERVICE_ATS_CONNECTIVITY:
 884       handle = entry->handle_ats_connectivity;
 885       if (NULL != handle)
 886         LOG_DEBUG ("Found ATS CONNECTIVITY handle for peer %u\n",
 887                    entry->index);
 888       break;
 889     }
 890   }
 891   else
 892   {
 893     entry = GNUNET_new (struct PooledConnection);
 894     entry->index = peer_id32;
 895     if ((NULL != map)
 896         && (GNUNET_CONTAINER_multihashmap32_size (map) < max_size))
 897     {
 898       GNUNET_assert (GNUNET_OK ==
 899                      GNUNET_CONTAINER_multihashmap32_put (map,
 900                                                           entry->index,
 901                                                           entry,
 902                                                           GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
 903       entry->in_pool = GNUNET_YES;
 904     }
 905     else
 906     {
 907       GNUNET_CONTAINER_DLL_insert_tail (head_not_pooled,
 908                                         tail_not_pooled,
 909                                         entry);
 910     }
 911     entry->cfg = GNUNET_CONFIGURATION_dup (cfg);
 912   }
 913   entry->demand++;
 914   gh = GNUNET_new (struct GST_ConnectionPool_GetHandle);
 915   gh->entry = entry;
 916   gh->cb = cb;
 917   gh->cb_cls = cb_cls;
 918   gh->target = target;
 919   gh->connect_notify_cb = connect_notify_cb;
 920   gh->connect_notify_cb_cls = connect_notify_cb_cls;
 921   gh->service = service;
 922   GNUNET_CONTAINER_DLL_insert (entry->head_waiting,
 923                                entry->tail_waiting,
 924                                gh);
 925   if (NULL != handle)
 926   {
 927     if (NULL == entry->notify_task)
 928     {
 929       if (NULL != search_waiting (entry, entry->head_waiting))
 930         entry->notify_task = GNUNET_SCHEDULER_add_now (&connection_ready,
 931                                                        entry);
 932     }
 933     return gh;
 934   }
 935   op = NULL;
 936   switch (gh->service)
 937   {
 938   case GST_CONNECTIONPOOL_SERVICE_TRANSPORT:
 939     if (NULL != entry->op_transport)
 940       return gh;                /* Operation pending */
 941     op = GNUNET_TESTBED_operation_create_ (entry,
 942                                            &opstart_get_handle_transport,
 943                                            &oprelease_get_handle_transport);
 944     entry->op_transport = op;
 945     break;
 946   case GST_CONNECTIONPOOL_SERVICE_CORE:
 947     if (NULL != entry->op_core)
 948       return gh;                /* Operation pending */
 949     op = GNUNET_TESTBED_operation_create_ (entry,
 950                                            &opstart_get_handle_core,
 951                                            &oprelease_get_handle_core);
 952     entry->op_core = op;
 953     break;
 954   case GST_CONNECTIONPOOL_SERVICE_ATS_CONNECTIVITY:
 955     if (NULL != entry->op_ats_connectivity)
 956       return gh;                /* Operation pending */
 957     op = GNUNET_TESTBED_operation_create_ (entry,
 958                                            &opstart_get_handle_ats_connectivity,
 959                                            &oprelease_get_handle_ats_connectivity);
 960     entry->op_ats_connectivity = op;
 961     break;
 962   }
 963   GNUNET_TESTBED_operation_queue_insert_ (GST_opq_openfds,
 964                                           op);
 965   GNUNET_TESTBED_operation_begin_wait_ (op);
 966   return gh;
 967 }
 968
 969
 970 /**
 971  * Relinquish a #GST_ConnectionPool_GetHandle object.  If the connection
 972  * associated with the object is currently being used by other
 973  * #GST_ConnectionPool_GetHandle objects, it is left in the connection pool.  If
 974  * no other objects are using the connection and the connection pool is not full
 975  * then it is placed in a LRU queue.  If the connection pool is full, then
 976  * connections from the LRU queue are evicted and closed to create place for this
 977  * connection.  If the connection pool if full and the LRU queue is empty, then
 978  * the connection is closed.
 979  *
 980  * @param gh the handle
 981  */
 982 void
 983 GST_connection_pool_get_handle_done (struct GST_ConnectionPool_GetHandle *gh)
 984 {
 985   struct PooledConnection *entry;
 986
 987   if (NULL == gh)
 988     return;
 989   entry = gh->entry;
 990   LOG_DEBUG ("Cleaning up get handle %p for service %u, peer %u\n",
 991              gh,
 992              gh->service, entry->index);
 993   if (! gh->connection_ready_called)
 994   {
 995     GNUNET_CONTAINER_DLL_remove (entry->head_waiting,
 996                                  entry->tail_waiting,
 997                                  gh);
 998     if ( (NULL == search_waiting (entry, entry->head_waiting)) &&
 999          (NULL != entry->notify_task) )
1000     {
1001       GNUNET_SCHEDULER_cancel (entry->notify_task);
1002       entry->notify_task = NULL;
1003     }
1004   }
1005   if (gh->notify_waiting)
1006   {
1007     GNUNET_CONTAINER_DLL_remove (entry->head_notify,
1008                                  entry->tail_notify,
1009                                  gh);
1010     gh->notify_waiting = 0;
1011   }
1012   GNUNET_free (gh);
1013   gh = NULL;
1014   GNUNET_assert (! entry->in_lru);
1015   if (! entry->in_pool)
1016     GNUNET_CONTAINER_DLL_remove (head_not_pooled,
1017                                  tail_not_pooled,
1018                                  entry);
1019   if (NULL != map)
1020   {
1021     if (GNUNET_YES == GNUNET_CONTAINER_multihashmap32_contains (map,
1022                                                                 entry->index))
1023       goto unallocate;
1024     if (GNUNET_CONTAINER_multihashmap32_size (map) == max_size)
1025     {
1026       if (NULL == head_lru)
1027         goto unallocate;
1028       destroy_pooled_connection (head_lru);
1029     }
1030     GNUNET_assert (GNUNET_OK ==
1031                    GNUNET_CONTAINER_multihashmap32_put (map,
1032                                                         entry->index,
1033                                                         entry,
1034                                                         GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
1035     entry->in_pool = GNUNET_YES;
1036   }
1037
1038  unallocate:
1039   GNUNET_assert (0 < entry->demand);
1040   entry->demand--;
1041   if (0 != entry->demand)
1042     return;
1043   if (entry->in_pool)
1044   {
1045     add_to_lru (entry);
1046     return;
1047   }
1048   destroy_pooled_connection (entry);
1049 }