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