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