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
   6   it under the terms of the GNU General Public License as published
   7   by the Free Software Foundation; either version 3, or (at your
   8   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   General Public License for more details.
  14
  15   You should have received a copy of the GNU General Public License
  16   along with GNUnet; see the file COPYING.  If not, write to the
  17   Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
  18   Boston, MA 02110-1301, USA.
  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
  31 /**
  32  * Redefine LOG with a changed log component string
  33  */
  34 #ifdef LOG
  35 #undef LOG
  36 #endif
  37 #define LOG(kind,...)                                   \
  38   GNUNET_log_from (kind, "testbed-connectionpool", __VA_ARGS__)
  39
  40
  41 /**
  42  * Time to expire a cache entry
  43  */
  44 #define CACHE_EXPIRY                            \
  45   GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 15)
  46
  47
  48 /**
  49  * The request handle for obtaining a pooled connection
  50  */
  51 struct GST_ConnectionPool_GetHandle;
  52
  53
  54 /**
  55  * A pooled connection
  56  */
  57 struct PooledConnection
  58 {
  59   /**
  60    * Next ptr for placing this object in the DLL of least recently used pooled
  61    * connections
  62    */
  63   struct PooledConnection *next;
  64
  65   /**
  66    * Prev ptr for placing this object in the DLL of the least recently used
  67    * pooled connections
  68    */
  69   struct PooledConnection *prev;
  70
  71   /**
  72    * The transport handle to the peer corresponding to this entry; can be NULL
  73    */
  74   struct GNUNET_TRANSPORT_Handle *handle_transport;
  75
  76   /**
  77    * The core handle to the peer corresponding to this entry; can be NULL
  78    */
  79   struct GNUNET_CORE_Handle *handle_core;
  80
  81   /**
  82    * The ATS handle to the peer correspondign to this entry; can be NULL.
  83    */
  84   struct GNUNET_ATS_ConnectivityHandle *handle_ats_connectivity;
  85
  86   /**
  87    * The operation handle for transport handle
  88    */
  89   struct GNUNET_TESTBED_Operation *op_transport;
  90
  91   /**
  92    * The operation handle for core handle
  93    */
  94   struct GNUNET_TESTBED_Operation *op_core;
  95
  96   /**
  97    * The operation handle for ATS handle
  98    */
  99   struct GNUNET_TESTBED_Operation *op_ats_connectivity;
 100
 101   /**
 102    * The peer identity of this peer. Will be set upon opening a connection to
 103    * the peers CORE service. Will be NULL until then and after the CORE
 104    * connection is closed
 105    */
 106   struct GNUNET_PeerIdentity *peer_identity;
 107
 108   /**
 109    * The configuration of the peer. Should be not NULL as long as the core_handle
 110    * or transport_handle are valid
 111    */
 112   struct GNUNET_CONFIGURATION_Handle *cfg;
 113
 114   /**
 115    * DLL head for the queue to serve notifications when a peer is connected
 116    */
 117   struct GST_ConnectionPool_GetHandle *head_notify;
 118
 119   /**
 120    * DLL tail for the queue to serve notifications when a peer is connected
 121    */
 122   struct GST_ConnectionPool_GetHandle *tail_notify;
 123
 124   /**
 125    * DLL head for the queue of #GST_ConnectionPool_GetHandle requests that are
 126    * waiting for this connection to be opened
 127    */
 128   struct GST_ConnectionPool_GetHandle *head_waiting;
 129
 130   /**
 131    * DLL tail for the queue of #GST_ConnectionPool_GetHandle requests that are
 132    * waiting for this connection to be opened
 133    */
 134   struct GST_ConnectionPool_GetHandle *tail_waiting;
 135
 136   /**
 137    * The task to expire this connection from the connection pool
 138    */
 139   struct GNUNET_SCHEDULER_Task * expire_task;
 140
 141   /**
 142    * The task to notify a waiting #GST_ConnectionPool_GetHandle object
 143    */
 144   struct GNUNET_SCHEDULER_Task * notify_task;
 145
 146   /**
 147    * Number of active requests using this pooled connection
 148    */
 149   unsigned int demand;
 150
 151   /**
 152    * Is this entry in LRU
 153    */
 154   int in_lru;
 155
 156   /**
 157    * Is this entry present in the connection pool
 158    */
 159   int in_pool;
 160
 161   /**
 162    * The index of this peer
 163    */
 164   uint32_t index;
 165 };
 166
 167
 168 /**
 169  * The request handle for obtaining a pooled connection
 170  */
 171 struct GST_ConnectionPool_GetHandle
 172 {
 173   /**
 174    * The next ptr for inclusion in the notification DLLs.  At first the object
 175    * is placed in the waiting DLL of the corresponding #PooledConnection
 176    * object.  After the handle is opened it is moved to the notification DLL if
 177    * @p connect_notify_cb and @p target are not NULL
 178    */
 179   struct GST_ConnectionPool_GetHandle *next;
 180
 181   /**
 182    * The prev ptr for inclusion in the notification DLLs
 183    */
 184   struct GST_ConnectionPool_GetHandle *prev;
 185
 186   /**
 187    * The pooled connection object this handle corresponds to
 188    */
 189   struct PooledConnection *entry;
 190
 191   /**
 192    * The cache callback to call when a handle is available
 193    */
 194   GST_connection_pool_connection_ready_cb cb;
 195
 196   /**
 197    * The closure for the above callback
 198    */
 199   void *cb_cls;
 200
 201   /**
 202    * The peer identity of the target peer. When this target peer is connected,
 203    * call the notify callback
 204    */
 205   const struct GNUNET_PeerIdentity *target;
 206
 207   /**
 208    * The callback to be called for serving notification that the target peer is
 209    * connected
 210    */
 211   GST_connection_pool_peer_connect_notify connect_notify_cb;
 212
 213   /**
 214    * The closure for the notify callback
 215    */
 216   void *connect_notify_cb_cls;
 217
 218   /**
 219    * The service we want to connect to
 220    */
 221   enum GST_ConnectionPool_Service service;
 222
 223   /**
 224    * Did we call the pool_connection_ready_cb already?
 225    */
 226   int connection_ready_called;
 227
 228   /**
 229    * Are we waiting for any peer connect notifications?
 230    */
 231   int notify_waiting;
 232 };
 233
 234
 235 /**
 236  * A hashmap for quickly finding connections in the connection pool
 237  */
 238 static struct GNUNET_CONTAINER_MultiHashMap32 *map;
 239
 240 /**
 241  * DLL head for maitaining the least recently used #PooledConnection objects.
 242  * The head is the least recently used object.
 243  */
 244 static struct PooledConnection *head_lru;
 245
 246 /**
 247  * DLL tail for maitaining the least recently used #PooledConnection objects
 248  */
 249 static struct PooledConnection *tail_lru;
 250
 251 /**
 252  * DLL head for maintaining #PooledConnection objects that are not added into
 253  * the connection pool as it was full at the time the object's creation
 254  * FIXME
 255  */
 256 static struct PooledConnection *head_not_pooled;
 257
 258 /**
 259  * DLL tail for maintaining #PooledConnection objects that are not added into
 260  * the connection pool as it was full at the time the object's creation
 261  */
 262 static struct PooledConnection *tail_not_pooled;
 263
 264 /**
 265  * The maximum number of entries that can be present in the connection pool
 266  */
 267 static unsigned int max_size;
 268
 269
 270 /**
 271  * Cancel the expiration task of the give #PooledConnection object
 272  *
 273  * @param entry the #PooledConnection object
 274  */
 275 static void
 276 expire_task_cancel (struct PooledConnection *entry);
 277
 278
 279 /**
 280  * Destroy a #PooledConnection object
 281  *
 282  * @param entry the #PooledConnection object
 283  */
 284 static void
 285 destroy_pooled_connection (struct PooledConnection *entry)
 286 {
 287   GNUNET_assert ((NULL == entry->head_notify) && (NULL == entry->tail_notify));
 288   GNUNET_assert ((NULL == entry->head_waiting) && (NULL ==
 289                                                    entry->tail_waiting));
 290   GNUNET_assert (0 == entry->demand);
 291   expire_task_cancel (entry);
 292   if (entry->in_lru)
 293     GNUNET_CONTAINER_DLL_remove (head_lru, tail_lru, entry);
 294   if (entry->in_pool)
 295     GNUNET_assert (GNUNET_OK ==
 296                    GNUNET_CONTAINER_multihashmap32_remove (map,
 297                                                            entry->index,
 298                                                            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 = GNUNET_SCHEDULER_add_delayed (CACHE_EXPIRY,
 377                                                      &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 notified.
 428  *
 429  * @param cls the #PooledConnection object
 430  */
 431 static void
 432 connection_ready (void *cls)
 433 {
 434   struct PooledConnection *entry = cls;
 435   struct GST_ConnectionPool_GetHandle *gh;
 436   struct GST_ConnectionPool_GetHandle *gh_next;
 437
 438   GNUNET_assert (NULL != entry->notify_task);
 439   entry->notify_task = NULL;
 440   gh = search_waiting (entry, entry->head_waiting);
 441   GNUNET_assert (NULL != gh);
 442   gh_next = NULL;
 443   if (NULL != gh->next)
 444     gh_next = search_waiting (entry, gh->next);
 445   GNUNET_CONTAINER_DLL_remove (entry->head_waiting,
 446                                entry->tail_waiting,
 447                                gh);
 448   gh->connection_ready_called = 1;
 449   if (NULL != gh_next)
 450     entry->notify_task = GNUNET_SCHEDULER_add_now (&connection_ready,
 451                                                    entry);
 452   if ( (NULL != gh->target) &&
 453        (NULL != gh->connect_notify_cb) )
 454   {
 455     GNUNET_CONTAINER_DLL_insert_tail (entry->head_notify,
 456                                       entry->tail_notify,
 457                                       gh);
 458     gh->notify_waiting = 1;
 459   }
 460   LOG_DEBUG ("Connection ready for handle type %u\n",
 461              gh->service);
 462   gh->cb (gh->cb_cls,
 463           entry->handle_core,
 464           entry->handle_transport,
 465           entry->handle_ats_connectivity,
 466           entry->peer_identity);
 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  */
 528 static void
 529 transport_peer_connect_notify_cb (void *cls,
 530                                   const struct GNUNET_PeerIdentity *peer)
 531 {
 532   struct PooledConnection *entry = cls;
 533
 534   peer_connect_notify_cb (entry,
 535                           peer,
 536                           GST_CONNECTIONPOOL_SERVICE_TRANSPORT);
 537 }
 538
 539
 540 /**
 541  * Function called when resources for opening a connection to TRANSPORT are
 542  * available.
 543  *
 544  * @param cls the #PooledConnection object
 545  */
 546 static void
 547 opstart_get_handle_transport (void *cls)
 548 {
 549   struct PooledConnection *entry = cls;
 550
 551   GNUNET_assert (NULL != entry);
 552   LOG_DEBUG ("Opening a transport connection to peer %u\n", entry->index);
 553   entry->handle_transport =
 554       GNUNET_TRANSPORT_connect (entry->cfg, NULL, entry, NULL,
 555                                 &transport_peer_connect_notify_cb, NULL);
 556   if (NULL == entry->handle_transport)
 557   {
 558     GNUNET_break (0);
 559     return;
 560   }
 561   if (0 == entry->demand)
 562     return;
 563   if (NULL != entry->notify_task)
 564     return;
 565   if (NULL != search_waiting (entry, entry->head_waiting))
 566   {
 567     entry->notify_task = GNUNET_SCHEDULER_add_now (&connection_ready, entry);
 568     return;
 569   }
 570 }
 571
 572
 573 /**
 574  * Function called when the operation responsible for opening a TRANSPORT
 575  * connection is marked as done.
 576  *
 577  * @param cls the cache entry
 578  */
 579 static void
 580 oprelease_get_handle_transport (void *cls)
 581 {
 582   struct PooledConnection *entry = cls;
 583
 584   if (NULL == entry->handle_transport)
 585     return;
 586   GNUNET_TRANSPORT_disconnect (entry->handle_transport);
 587   entry->handle_transport = NULL;
 588 }
 589
 590
 591 /**
 592  * Method called whenever a given peer connects at CORE level
 593  *
 594  * @param cls the #PooledConnection object
 595  * @param peer peer identity this notification is about
 596  */
 597 static void
 598 core_peer_connect_cb (void *cls, const struct GNUNET_PeerIdentity *peer)
 599 {
 600   struct PooledConnection *entry = cls;
 601
 602   peer_connect_notify_cb (entry, peer, GST_CONNECTIONPOOL_SERVICE_CORE);
 603 }
 604
 605
 606 /**
 607  * Function called after #GNUNET_CORE_connect() has succeeded (or failed
 608  * for good).  Note that the private key of the peer is intentionally
 609  * not exposed here; if you need it, your process should try to read
 610  * the private key file directly (which should work if you are
 611  * authorized...).  Implementations of this function must not call
 612  * #GNUNET_CORE_disconnect() (other than by scheduling a new task to
 613  * do this later).
 614  *
 615  * @param cls the #PooledConnection object
 616  * @param my_identity ID of this peer, NULL if we failed
 617  */
 618 static void
 619 core_startup_cb (void *cls,
 620                  const struct GNUNET_PeerIdentity *my_identity)
 621 {
 622   struct PooledConnection *entry = cls;
 623
 624   if (NULL == my_identity)
 625   {
 626     GNUNET_break (0);
 627     return;
 628   }
 629   GNUNET_assert (NULL == entry->peer_identity);
 630   entry->peer_identity = GNUNET_new (struct GNUNET_PeerIdentity);
 631   memcpy (entry->peer_identity,
 632           my_identity,
 633           sizeof (struct GNUNET_PeerIdentity));
 634   if (0 == entry->demand)
 635     return;
 636   if (NULL != entry->notify_task)
 637     return;
 638   if (NULL != search_waiting (entry, entry->head_waiting))
 639   {
 640     entry->notify_task = GNUNET_SCHEDULER_add_now (&connection_ready, entry);
 641     return;
 642   }
 643 }
 644
 645
 646 /**
 647  * Function called when resources for opening a connection to CORE are
 648  * available.
 649  *
 650  * @param cls the #PooledConnection object
 651  */
 652 static void
 653 opstart_get_handle_core (void *cls)
 654 {
 655   struct PooledConnection *entry = cls;
 656   const struct GNUNET_CORE_MessageHandler no_handlers[] = {
 657     {NULL, 0, 0}
 658   };
 659
 660   GNUNET_assert (NULL != entry);
 661   LOG_DEBUG ("Opening a CORE connection to peer %u\n", entry->index);
 662   entry->handle_core =
 663       GNUNET_CORE_connect (entry->cfg, entry,        /* closure */
 664                            &core_startup_cb, /* core startup notify */
 665                            &core_peer_connect_cb,    /* peer connect notify */
 666                            NULL,     /* peer disconnect notify */
 667                            NULL,     /* inbound notify */
 668                            GNUNET_NO,        /* inbound header only? */
 669                            NULL,     /* outbound notify */
 670                            GNUNET_NO,        /* outbound header only? */
 671                            no_handlers);
 672 }
 673
 674
 675 /**
 676  * Function called when the operation responsible for opening a CORE
 677  * connection is marked as done.
 678  *
 679  * @param cls the #PooledConnection object
 680  */
 681 static void
 682 oprelease_get_handle_core (void *cls)
 683 {
 684   struct PooledConnection *entry = cls;
 685
 686   if (NULL == entry->handle_core)
 687     return;
 688   GNUNET_CORE_disconnect (entry->handle_core);
 689   entry->handle_core = NULL;
 690   GNUNET_free_non_null (entry->peer_identity);
 691   entry->peer_identity = NULL;
 692 }
 693
 694
 695 /**
 696  * Function called when resources for opening a connection to ATS are
 697  * available.
 698  *
 699  * @param cls the #PooledConnection object
 700  */
 701 static void
 702 opstart_get_handle_ats_connectivity (void *cls)
 703 {
 704   struct PooledConnection *entry = cls;
 705
 706   entry->handle_ats_connectivity =
 707     GNUNET_ATS_connectivity_init (entry->cfg);
 708 }
 709
 710
 711 /**
 712  * Function called when the operation responsible for opening a ATS
 713  * connection is marked as done.
 714  *
 715  * @param cls the #PooledConnection object
 716  */
 717 static void
 718 oprelease_get_handle_ats_connectivity (void *cls)
 719 {
 720   struct PooledConnection *entry = cls;
 721
 722   if (NULL == entry->handle_ats_connectivity)
 723     return;
 724   GNUNET_ATS_connectivity_done (entry->handle_ats_connectivity);
 725   entry->handle_ats_connectivity = NULL;
 726 }
 727
 728
 729 /**
 730  * This function will be called for every #PooledConnection object in @p map
 731  *
 732  * @param cls NULL
 733  * @param key current key code
 734  * @param value the #PooledConnection object
 735  * @return #GNUNET_YES if we should continue to
 736  *         iterate,
 737  *         #GNUNET_NO if not.
 738  */
 739 static int
 740 cleanup_iterator (void *cls,
 741                   uint32_t key,
 742                   void *value)
 743 {
 744   struct PooledConnection *entry = value;
 745
 746   GNUNET_assert (NULL != entry);
 747   destroy_pooled_connection (entry);
 748   return GNUNET_YES;
 749 }
 750
 751
 752 /**
 753  * Initialise the connection pool.
 754  *
 755  * @param size the size of the connection pool.  Each entry in the connection
 756  *   pool can handle a connection to each of the services enumerated in
 757  *   #GST_ConnectionPool_Service
 758  */
 759 void
 760 GST_connection_pool_init (unsigned int size)
 761 {
 762   max_size = size;
 763   if (0 == max_size)
 764     return;
 765   GNUNET_assert (NULL == map);
 766   map = GNUNET_CONTAINER_multihashmap32_create (((size * 3) / 4) + 1);
 767 }
 768
 769
 770 /**
 771  * Cleanup the connection pool
 772  */
 773 void
 774 GST_connection_pool_destroy ()
 775 {
 776   struct PooledConnection *entry;
 777
 778   if (NULL != map)
 779   {
 780     GNUNET_assert (GNUNET_SYSERR !=
 781                    GNUNET_CONTAINER_multihashmap32_iterate (map,
 782                                                             &cleanup_iterator,
 783                                                             NULL));
 784     GNUNET_CONTAINER_multihashmap32_destroy (map);
 785     map = NULL;
 786   }
 787   while (NULL != (entry = head_lru))
 788   {
 789     GNUNET_CONTAINER_DLL_remove (head_lru, tail_lru, entry);
 790     destroy_pooled_connection (entry);
 791   }
 792   GNUNET_assert (NULL == head_not_pooled);
 793 }
 794
 795
 796 /**
 797  * Get a connection handle to @a service.  If the connection is opened before
 798  * and the connection handle is present in the connection pool, it is returned
 799  * through @a cb.  @a peer_id is used for the lookup in the connection pool.  If
 800  * the connection handle is not present in the connection pool, a new connection
 801  * handle is opened for the @a service using @a cfg.  Additionally, @a target,
 802  * @a connect_notify_cb can be specified to get notified when @a target is
 803  * connected at @a service.
 804  *
 805  * @note @a connect_notify_cb will not be called if @a target is
 806  * already connected @a service level. Use
 807  * GNUNET_TRANSPORT_check_peer_connected() or a similar function from the
 808  * respective @a service's API to check if the target peer is already connected or
 809  * not. @a connect_notify_cb will be called only once or never (in case @a target
 810  * cannot be connected or is already connected).
 811  *
 812  * @param peer_id the index of the peer
 813  * @param cfg the configuration with which the transport handle has to be
 814  *          created if it was not present in the cache
 815  * @param service the service of interest
 816  * @param cb the callback to notify when the transport handle is available
 817  * @param cb_cls the closure for @a cb
 818  * @param target the peer identify of the peer whose connection to our TRANSPORT
 819  *          subsystem will be notified through the @a connect_notify_cb. Can be NULL
 820  * @param connect_notify_cb the callback to call when the @a target peer is
 821  *          connected. This callback will only be called once or never again (in
 822  *          case the target peer cannot be connected). Can be NULL
 823  * @param connect_notify_cb_cls the closure for @a connect_notify_cb
 824  * @return the handle which can be used cancel or mark that the handle is no
 825  *           longer being used
 826  */
 827 struct GST_ConnectionPool_GetHandle *
 828 GST_connection_pool_get_handle (unsigned int peer_id,
 829                                 const struct GNUNET_CONFIGURATION_Handle *cfg,
 830                                 enum GST_ConnectionPool_Service service,
 831                                 GST_connection_pool_connection_ready_cb cb,
 832                                 void *cb_cls,
 833                                 const struct GNUNET_PeerIdentity *target,
 834                                 GST_connection_pool_peer_connect_notify connect_notify_cb,
 835                                 void *connect_notify_cb_cls)
 836 {
 837   struct GST_ConnectionPool_GetHandle *gh;
 838   struct PooledConnection *entry;
 839   struct GNUNET_TESTBED_Operation *op;
 840   void *handle;
 841   uint32_t peer_id32;
 842
 843   peer_id32 = (uint32_t) peer_id;
 844   handle = NULL;
 845   entry = NULL;
 846   if (NULL != map)
 847     entry = GNUNET_CONTAINER_multihashmap32_get (map, peer_id32);
 848   if (NULL != entry)
 849   {
 850     if (entry->in_lru)
 851     {
 852       GNUNET_assert (0 == entry->demand);
 853       expire_task_cancel (entry);
 854       GNUNET_CONTAINER_DLL_remove (head_lru, tail_lru, entry);
 855       entry->in_lru = GNUNET_NO;
 856     }
 857     switch (service)
 858     {
 859     case GST_CONNECTIONPOOL_SERVICE_TRANSPORT:
 860       handle = entry->handle_transport;
 861       if (NULL != handle)
 862         LOG_DEBUG ("Found TRANSPORT handle for peer %u\n",
 863                    entry->index);
 864       break;
 865     case GST_CONNECTIONPOOL_SERVICE_CORE:
 866       handle = entry->handle_core;
 867       if (NULL != handle)
 868         LOG_DEBUG ("Found CORE handle for peer %u\n",
 869                    entry->index);
 870       break;
 871     case GST_CONNECTIONPOOL_SERVICE_ATS_CONNECTIVITY:
 872       handle = entry->handle_ats_connectivity;
 873       if (NULL != handle)
 874         LOG_DEBUG ("Found ATS CONNECTIVITY handle for peer %u\n",
 875                    entry->index);
 876       break;
 877     }
 878   }
 879   else
 880   {
 881     entry = GNUNET_new (struct PooledConnection);
 882     entry->index = peer_id32;
 883     if ((NULL != map)
 884         && (GNUNET_CONTAINER_multihashmap32_size (map) < max_size))
 885     {
 886       GNUNET_assert (GNUNET_OK ==
 887                      GNUNET_CONTAINER_multihashmap32_put (map,
 888                                                           entry->index,
 889                                                           entry,
 890                                                           GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
 891       entry->in_pool = GNUNET_YES;
 892     }
 893     else
 894     {
 895       GNUNET_CONTAINER_DLL_insert_tail (head_not_pooled,
 896                                         tail_not_pooled,
 897                                         entry);
 898     }
 899     entry->cfg = GNUNET_CONFIGURATION_dup (cfg);
 900   }
 901   entry->demand++;
 902   gh = GNUNET_new (struct GST_ConnectionPool_GetHandle);
 903   gh->entry = entry;
 904   gh->cb = cb;
 905   gh->cb_cls = cb_cls;
 906   gh->target = target;
 907   gh->connect_notify_cb = connect_notify_cb;
 908   gh->connect_notify_cb_cls = connect_notify_cb_cls;
 909   gh->service = service;
 910   GNUNET_CONTAINER_DLL_insert (entry->head_waiting,
 911                                entry->tail_waiting,
 912                                gh);
 913   if (NULL != handle)
 914   {
 915     if (NULL == entry->notify_task)
 916     {
 917       if (NULL != search_waiting (entry, entry->head_waiting))
 918         entry->notify_task = GNUNET_SCHEDULER_add_now (&connection_ready,
 919                                                        entry);
 920     }
 921     return gh;
 922   }
 923   op = NULL;
 924   switch (gh->service)
 925   {
 926   case GST_CONNECTIONPOOL_SERVICE_TRANSPORT:
 927     if (NULL != entry->op_transport)
 928       return gh;                /* Operation pending */
 929     op = GNUNET_TESTBED_operation_create_ (entry,
 930                                            &opstart_get_handle_transport,
 931                                            &oprelease_get_handle_transport);
 932     entry->op_transport = op;
 933     break;
 934   case GST_CONNECTIONPOOL_SERVICE_CORE:
 935     if (NULL != entry->op_core)
 936       return gh;                /* Operation pending */
 937     op = GNUNET_TESTBED_operation_create_ (entry,
 938                                            &opstart_get_handle_core,
 939                                            &oprelease_get_handle_core);
 940     entry->op_core = op;
 941     break;
 942   case GST_CONNECTIONPOOL_SERVICE_ATS_CONNECTIVITY:
 943     if (NULL != entry->op_ats_connectivity)
 944       return gh;                /* Operation pending */
 945     op = GNUNET_TESTBED_operation_create_ (entry,
 946                                            &opstart_get_handle_ats_connectivity,
 947                                            &oprelease_get_handle_ats_connectivity);
 948     entry->op_ats_connectivity = op;
 949     break;
 950   }
 951   GNUNET_TESTBED_operation_queue_insert_ (GST_opq_openfds,
 952                                           op);
 953   GNUNET_TESTBED_operation_begin_wait_ (op);
 954   return gh;
 955 }
 956
 957
 958 /**
 959  * Relinquish a #GST_ConnectionPool_GetHandle object.  If the connection
 960  * associated with the object is currently being used by other
 961  * #GST_ConnectionPool_GetHandle objects, it is left in the connection pool.  If
 962  * no other objects are using the connection and the connection pool is not full
 963  * then it is placed in a LRU queue.  If the connection pool is full, then
 964  * connections from the LRU queue are evicted and closed to create place for this
 965  * connection.  If the connection pool if full and the LRU queue is empty, then
 966  * the connection is closed.
 967  *
 968  * @param gh the handle
 969  */
 970 void
 971 GST_connection_pool_get_handle_done (struct GST_ConnectionPool_GetHandle *gh)
 972 {
 973   struct PooledConnection *entry;
 974
 975   if (NULL == gh)
 976     return;
 977   entry = gh->entry;
 978   LOG_DEBUG ("Cleaning up get handle %p for service %u, peer %u\n",
 979              gh,
 980              gh->service, entry->index);
 981   if (! gh->connection_ready_called)
 982   {
 983     GNUNET_CONTAINER_DLL_remove (entry->head_waiting,
 984                                  entry->tail_waiting,
 985                                  gh);
 986     if ( (NULL == search_waiting (entry, entry->head_waiting)) &&
 987          (NULL != entry->notify_task) )
 988     {
 989       GNUNET_SCHEDULER_cancel (entry->notify_task);
 990       entry->notify_task = NULL;
 991     }
 992   }
 993   if (gh->notify_waiting)
 994   {
 995     GNUNET_CONTAINER_DLL_remove (entry->head_notify,
 996                                  entry->tail_notify,
 997                                  gh);
 998     gh->notify_waiting = 0;
 999   }
1000   GNUNET_free (gh);
1001   gh = NULL;
1002   GNUNET_assert (! entry->in_lru);
1003   if (! entry->in_pool)
1004     GNUNET_CONTAINER_DLL_remove (head_not_pooled,
1005                                  tail_not_pooled,
1006                                  entry);
1007   if (NULL != map)
1008   {
1009     if (GNUNET_YES == GNUNET_CONTAINER_multihashmap32_contains (map,
1010                                                                 entry->index))
1011       goto unallocate;
1012     if (GNUNET_CONTAINER_multihashmap32_size (map) == max_size)
1013     {
1014       if (NULL == head_lru)
1015         goto unallocate;
1016       destroy_pooled_connection (head_lru);
1017     }
1018     GNUNET_assert (GNUNET_OK ==
1019                    GNUNET_CONTAINER_multihashmap32_put (map,
1020                                                         entry->index,
1021                                                         entry,
1022                                                         GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
1023     entry->in_pool = GNUNET_YES;
1024   }
1025
1026  unallocate:
1027   GNUNET_assert (0 < entry->demand);
1028   entry->demand--;
1029   if (0 != entry->demand)
1030     return;
1031   if (entry->in_pool)
1032   {
1033     add_to_lru (entry);
1034     return;
1035   }
1036   destroy_pooled_connection (entry);
1037 }