gnunet-service-testbed_connectionpool.c

Go to the documentation of this file.

/*
  This file is part of GNUnet.
  Copyright (C) 2008--2015 GNUnet e.V.

  GNUnet is free software: you can redistribute it and/or modify it
  under the terms of the GNU Affero General Public License as published
  by the Free Software Foundation, either version 3 of the License,
  or (at your option) any later version.

  GNUnet is distributed in the hope that it will be useful, but
  WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  Affero General Public License for more details.

  You should have received a copy of the GNU Affero General Public License
  along with this program.  If not, see <http://www.gnu.org/licenses/>.

     SPDX-License-Identifier: AGPL3.0-or-later
*/

#include "gnunet-service-testbed.h"
#include "gnunet-service-testbed_connectionpool.h"
#include "testbed_api_operations.h"
#include "gnunet_transport_service.h"

#ifdef LOG
#undef LOG
#endif
#define LOG(kind, ...) \
  GNUNET_log_from (kind, "testbed-connectionpool", __VA_ARGS__)


#define CACHE_EXPIRY \
  GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 15)


struct GST_ConnectionPool_GetHandle;


struct PooledConnection
{
  struct PooledConnection *next;

  struct PooledConnection *prev;

  struct GNUNET_TRANSPORT_CoreHandle *handle_transport;

  struct GNUNET_CORE_Handle *handle_core;

  struct GNUNET_ATS_ConnectivityHandle *handle_ats_connectivity;

  struct GNUNET_TESTBED_Operation *op_transport;

  struct GNUNET_TESTBED_Operation *op_core;

  struct GNUNET_TESTBED_Operation *op_ats_connectivity;

  struct GNUNET_PeerIdentity *peer_identity;

  struct GNUNET_CONFIGURATION_Handle *cfg;

  struct GST_ConnectionPool_GetHandle *head_notify;

  struct GST_ConnectionPool_GetHandle *tail_notify;

  struct GST_ConnectionPool_GetHandle *head_waiting;

  struct GST_ConnectionPool_GetHandle *tail_waiting;

  struct GNUNET_SCHEDULER_Task *expire_task;

  struct GNUNET_SCHEDULER_Task *notify_task;

  unsigned int demand;

  int in_lru;

  int in_pool;

  uint32_t index;
};


struct GST_ConnectionPool_GetHandle
{
  struct GST_ConnectionPool_GetHandle *next;

  struct GST_ConnectionPool_GetHandle *prev;

  struct PooledConnection *entry;

  GST_connection_pool_connection_ready_cb cb;

  void *cb_cls;

  const struct GNUNET_PeerIdentity *target;

  GST_connection_pool_peer_connect_notify connect_notify_cb;

  void *connect_notify_cb_cls;

  enum GST_ConnectionPool_Service service;

  int connection_ready_called;

  int notify_waiting;
};


static struct GNUNET_CONTAINER_MultiHashMap32 *map;

static struct PooledConnection *head_lru;

static struct PooledConnection *tail_lru;

static struct PooledConnection *head_not_pooled;

static struct PooledConnection *tail_not_pooled;

static unsigned int max_size;


static void
expire_task_cancel (struct PooledConnection *entry);


static void
destroy_pooled_connection (struct PooledConnection *entry)
{
  GNUNET_assert ((NULL == entry->head_notify) && (NULL == entry->tail_notify));
  GNUNET_assert ((NULL == entry->head_waiting) &&
                 (NULL == entry->tail_waiting));
  GNUNET_assert (0 == entry->demand);
  expire_task_cancel (entry);
  if (entry->in_lru)
    GNUNET_CONTAINER_DLL_remove (head_lru, tail_lru, entry);
  if (entry->in_pool)
    GNUNET_assert (
      GNUNET_OK ==
      GNUNET_CONTAINER_multihashmap32_remove (map, entry->index, entry));
  if (NULL != entry->notify_task)
  {
    GNUNET_SCHEDULER_cancel (entry->notify_task);
    entry->notify_task = NULL;
  }
  LOG_DEBUG ("Cleaning up handles of a pooled connection\n");
  if (NULL != entry->handle_transport)
    GNUNET_assert (NULL != entry->op_transport);
  if (NULL != entry->op_transport)
  {
    GNUNET_TESTBED_operation_done (entry->op_transport);
    entry->op_transport = NULL;
  }
  if (NULL != entry->handle_ats_connectivity)
    GNUNET_assert (NULL != entry->op_ats_connectivity);
  if (NULL != entry->op_ats_connectivity)
  {
    GNUNET_TESTBED_operation_done (entry->op_ats_connectivity);
    entry->op_ats_connectivity = NULL;
  }
  if (NULL != entry->op_core)
  {
    GNUNET_TESTBED_operation_done (entry->op_core);
    entry->op_core = NULL;
  }
  GNUNET_assert (NULL == entry->handle_core);
  GNUNET_assert (NULL == entry->handle_ats_connectivity);
  GNUNET_assert (NULL == entry->handle_transport);
  GNUNET_CONFIGURATION_destroy (entry->cfg);
  GNUNET_free (entry);
}


static void
expire (void *cls)
{
  struct PooledConnection *entry = cls;

  entry->expire_task = NULL;
  destroy_pooled_connection (entry);
}


static void
expire_task_cancel (struct PooledConnection *entry)
{
  if (NULL != entry->expire_task)
  {
    GNUNET_SCHEDULER_cancel (entry->expire_task);
    entry->expire_task = NULL;
  }
}


static void
add_to_lru (struct PooledConnection *entry)
{
  GNUNET_assert (0 == entry->demand);
  GNUNET_assert (! entry->in_lru);
  GNUNET_CONTAINER_DLL_insert_tail (head_lru, tail_lru, entry);
  entry->in_lru = GNUNET_YES;
  GNUNET_assert (NULL == entry->expire_task);
  entry->expire_task =
    GNUNET_SCHEDULER_add_delayed (CACHE_EXPIRY, &expire, entry);
}


static struct GST_ConnectionPool_GetHandle *
search_waiting (const struct PooledConnection *entry,
                struct GST_ConnectionPool_GetHandle *head)
{
  struct GST_ConnectionPool_GetHandle *gh;

  for (gh = head; NULL != gh; gh = gh->next)
  {
    switch (gh->service)
    {
    case GST_CONNECTIONPOOL_SERVICE_CORE:
      if (NULL == entry->handle_core)
        continue;
      if (NULL == entry->peer_identity)
        continue; /* CORE connection isn't ready yet */
      break;
    case GST_CONNECTIONPOOL_SERVICE_TRANSPORT:
      if (NULL == entry->handle_transport)
        continue;
      break;
    case GST_CONNECTIONPOOL_SERVICE_ATS_CONNECTIVITY:
      if (NULL == entry->handle_ats_connectivity)
        continue;
      break;
    }
    break;
  }
  return gh;
}


static void
connection_ready (void *cls)
{
  struct PooledConnection *entry = cls;
  struct GST_ConnectionPool_GetHandle *gh;
  struct GST_ConnectionPool_GetHandle *gh_next;

  GNUNET_assert (NULL != entry->notify_task);
  entry->notify_task = NULL;
  gh = search_waiting (entry, entry->head_waiting);
  GNUNET_assert (NULL != gh);
  gh_next = NULL;
  if (NULL != gh->next)
    gh_next = search_waiting (entry, gh->next);
  GNUNET_CONTAINER_DLL_remove (entry->head_waiting, entry->tail_waiting, gh);
  gh->connection_ready_called = 1;
  if (NULL != gh_next)
    entry->notify_task = GNUNET_SCHEDULER_add_now (&connection_ready, entry);
  if ((NULL != gh->target) && (NULL != gh->connect_notify_cb))
  {
    GNUNET_CONTAINER_DLL_insert_tail (entry->head_notify,
                                      entry->tail_notify,
                                      gh);
    gh->notify_waiting = 1;
  }
  LOG_DEBUG ("Connection ready for handle type %u\n", gh->service);
  gh->cb (gh->cb_cls,
          entry->handle_core,
          entry->handle_transport,
          entry->handle_ats_connectivity,
          entry->peer_identity,
          entry->cfg);
}


static void
peer_connect_notify_cb (void *cls,
                        const struct GNUNET_PeerIdentity *peer,
                        const enum GST_ConnectionPool_Service service)
{
  struct PooledConnection *entry = cls;
  struct GST_ConnectionPool_GetHandle *gh;
  struct GST_ConnectionPool_GetHandle *gh_next;
  GST_connection_pool_peer_connect_notify cb;
  void *cb_cls;

  for (gh = entry->head_notify; NULL != gh;)
  {
    GNUNET_assert (NULL != gh->target);
    GNUNET_assert (NULL != gh->connect_notify_cb);
    GNUNET_assert (gh->connection_ready_called);
    if (service != gh->service)
    {
      gh = gh->next;
      continue;
    }
    if (0 != memcmp (gh->target, peer, sizeof (struct GNUNET_PeerIdentity)))
    {
      gh = gh->next;
      continue;
    }
    cb = gh->connect_notify_cb;
    cb_cls = gh->connect_notify_cb_cls;
    gh_next = gh->next;
    GNUNET_CONTAINER_DLL_remove (entry->head_notify, entry->tail_notify, gh);
    gh->notify_waiting = 0;
    LOG_DEBUG ("Peer connected to peer %u at service %u\n",
               entry->index,
               gh->service);
    gh = gh_next;
    cb (cb_cls, peer);
  }
}


static void *
transport_peer_connect_notify_cb (void *cls,
                                  const struct GNUNET_PeerIdentity *peer,
                                  struct GNUNET_MQ_Handle *mq)
{
  struct PooledConnection *entry = cls;

  peer_connect_notify_cb (entry, peer, GST_CONNECTIONPOOL_SERVICE_TRANSPORT);
  return NULL;
}


static void
opstart_get_handle_transport (void *cls)
{
  struct PooledConnection *entry = cls;

  GNUNET_assert (NULL != entry);
  LOG_DEBUG ("Opening a transport connection to peer %u\n", entry->index);
  entry->handle_transport =
    GNUNET_TRANSPORT_core_connect (entry->cfg,
                                   NULL,
                                   NULL,
                                   entry,
                                   &transport_peer_connect_notify_cb,
                                   NULL,
                                   NULL);
  if (NULL == entry->handle_transport)
  {
    GNUNET_break (0);
    return;
  }
  if (0 == entry->demand)
    return;
  if (NULL != entry->notify_task)
    return;
  if (NULL != search_waiting (entry, entry->head_waiting))
  {
    entry->notify_task = GNUNET_SCHEDULER_add_now (&connection_ready, entry);
    return;
  }
}


static void
oprelease_get_handle_transport (void *cls)
{
  struct PooledConnection *entry = cls;

  if (NULL == entry->handle_transport)
    return;
  GNUNET_TRANSPORT_core_disconnect (entry->handle_transport);
  entry->handle_transport = NULL;
}


static void *
core_peer_connect_cb (void *cls,
                      const struct GNUNET_PeerIdentity *peer,
                      struct GNUNET_MQ_Handle *mq)
{
  struct PooledConnection *entry = cls;

  peer_connect_notify_cb (entry, peer, GST_CONNECTIONPOOL_SERVICE_CORE);
  return (void *) peer;
}


static void
core_startup_cb (void *cls, const struct GNUNET_PeerIdentity *my_identity)
{
  struct PooledConnection *entry = cls;

  if (NULL == my_identity)
  {
    GNUNET_break (0);
    return;
  }
  GNUNET_assert (NULL == entry->peer_identity);
  entry->peer_identity = GNUNET_new (struct GNUNET_PeerIdentity);
  *entry->peer_identity = *my_identity;
  if (0 == entry->demand)
    return;
  if (NULL != entry->notify_task)
    return;
  if (NULL != search_waiting (entry, entry->head_waiting))
  {
    entry->notify_task = GNUNET_SCHEDULER_add_now (&connection_ready, entry);
    return;
  }
}


static void
opstart_get_handle_core (void *cls)
{
  struct PooledConnection *entry = cls;

  GNUNET_assert (NULL != entry);
  LOG_DEBUG ("Opening a CORE connection to peer %u\n", entry->index);
  entry->handle_core =
    GNUNET_CORE_connect (entry->cfg,
                         entry, /* closure */
                         &core_startup_cb, /* core startup notify */
                         &core_peer_connect_cb, /* peer connect notify */
                         NULL, /* peer disconnect notify */
                         NULL);
}


static void
oprelease_get_handle_core (void *cls)
{
  struct PooledConnection *entry = cls;

  if (NULL == entry->handle_core)
    return;
  GNUNET_CORE_disconnect (entry->handle_core);
  entry->handle_core = NULL;
  GNUNET_free_non_null (entry->peer_identity);
  entry->peer_identity = NULL;
}


static void
opstart_get_handle_ats_connectivity (void *cls)
{
  struct PooledConnection *entry = cls;

  entry->handle_ats_connectivity = GNUNET_ATS_connectivity_init (entry->cfg);
}


static void
oprelease_get_handle_ats_connectivity (void *cls)
{
  struct PooledConnection *entry = cls;

  if (NULL == entry->handle_ats_connectivity)
    return;
  GNUNET_ATS_connectivity_done (entry->handle_ats_connectivity);
  entry->handle_ats_connectivity = NULL;
}


static int
cleanup_iterator (void *cls, uint32_t key, void *value)
{
  struct PooledConnection *entry = value;

  GNUNET_assert (NULL != entry);
  destroy_pooled_connection (entry);
  return GNUNET_YES;
}


void
GST_connection_pool_init (unsigned int size)
{
  max_size = size;
  if (0 == max_size)
    return;
  GNUNET_assert (NULL == map);
  map = GNUNET_CONTAINER_multihashmap32_create (((size * 3) / 4) + 1);
}


void
GST_connection_pool_destroy ()
{
  struct PooledConnection *entry;

  if (NULL != map)
  {
    GNUNET_assert (
      GNUNET_SYSERR !=
      GNUNET_CONTAINER_multihashmap32_iterate (map, &cleanup_iterator, NULL));
    GNUNET_CONTAINER_multihashmap32_destroy (map);
    map = NULL;
  }
  while (NULL != (entry = head_lru))
  {
    GNUNET_CONTAINER_DLL_remove (head_lru, tail_lru, entry);
    destroy_pooled_connection (entry);
  }
  GNUNET_assert (NULL == head_not_pooled);
}


struct GST_ConnectionPool_GetHandle *
GST_connection_pool_get_handle (
  unsigned int peer_id,
  const struct GNUNET_CONFIGURATION_Handle *cfg,
  enum GST_ConnectionPool_Service service,
  GST_connection_pool_connection_ready_cb cb,
  void *cb_cls,
  const struct GNUNET_PeerIdentity *target,
  GST_connection_pool_peer_connect_notify connect_notify_cb,
  void *connect_notify_cb_cls)
{
  struct GST_ConnectionPool_GetHandle *gh;
  struct PooledConnection *entry;
  struct GNUNET_TESTBED_Operation *op;
  void *handle;
  uint32_t peer_id32;

  peer_id32 = (uint32_t) peer_id;
  handle = NULL;
  entry = NULL;
  if (NULL != map)
    entry = GNUNET_CONTAINER_multihashmap32_get (map, peer_id32);
  if (NULL != entry)
  {
    if (entry->in_lru)
    {
      GNUNET_assert (0 == entry->demand);
      expire_task_cancel (entry);
      GNUNET_CONTAINER_DLL_remove (head_lru, tail_lru, entry);
      entry->in_lru = GNUNET_NO;
    }
    switch (service)
    {
    case GST_CONNECTIONPOOL_SERVICE_TRANSPORT:
      handle = entry->handle_transport;
      if (NULL != handle)
        LOG_DEBUG ("Found TRANSPORT handle for peer %u\n", entry->index);
      break;
    case GST_CONNECTIONPOOL_SERVICE_CORE:
      handle = entry->handle_core;
      if (NULL != handle)
        LOG_DEBUG ("Found CORE handle for peer %u\n", entry->index);
      break;
    case GST_CONNECTIONPOOL_SERVICE_ATS_CONNECTIVITY:
      handle = entry->handle_ats_connectivity;
      if (NULL != handle)
        LOG_DEBUG ("Found ATS CONNECTIVITY handle for peer %u\n", entry->index);
      break;
    }
  }
  else
  {
    entry = GNUNET_new (struct PooledConnection);
    entry->index = peer_id32;
    if ((NULL != map) &&
        (GNUNET_CONTAINER_multihashmap32_size (map) < max_size))
    {
      GNUNET_assert (GNUNET_OK ==
                     GNUNET_CONTAINER_multihashmap32_put (
                       map,
                       entry->index,
                       entry,
                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
      entry->in_pool = GNUNET_YES;
    }
    else
    {
      GNUNET_CONTAINER_DLL_insert_tail (head_not_pooled,
                                        tail_not_pooled,
                                        entry);
    }
    entry->cfg = GNUNET_CONFIGURATION_dup (cfg);
  }
  entry->demand++;
  gh = GNUNET_new (struct GST_ConnectionPool_GetHandle);
  gh->entry = entry;
  gh->cb = cb;
  gh->cb_cls = cb_cls;
  gh->target = target;
  gh->connect_notify_cb = connect_notify_cb;
  gh->connect_notify_cb_cls = connect_notify_cb_cls;
  gh->service = service;
  GNUNET_CONTAINER_DLL_insert (entry->head_waiting, entry->tail_waiting, gh);
  if (NULL != handle)
  {
    if (NULL == entry->notify_task)
    {
      if (NULL != search_waiting (entry, entry->head_waiting))
        entry->notify_task =
          GNUNET_SCHEDULER_add_now (&connection_ready, entry);
    }
    return gh;
  }
  op = NULL;
  switch (gh->service)
  {
  case GST_CONNECTIONPOOL_SERVICE_TRANSPORT:
    if (NULL != entry->op_transport)
      return gh; /* Operation pending */
    op = GNUNET_TESTBED_operation_create_ (entry,
                                           &opstart_get_handle_transport,
                                           &oprelease_get_handle_transport);
    entry->op_transport = op;
    break;
  case GST_CONNECTIONPOOL_SERVICE_CORE:
    if (NULL != entry->op_core)
      return gh; /* Operation pending */
    op = GNUNET_TESTBED_operation_create_ (entry,
                                           &opstart_get_handle_core,
                                           &oprelease_get_handle_core);
    entry->op_core = op;
    break;
  case GST_CONNECTIONPOOL_SERVICE_ATS_CONNECTIVITY:
    if (NULL != entry->op_ats_connectivity)
      return gh; /* Operation pending */
    op =
      GNUNET_TESTBED_operation_create_ (entry,
                                        &opstart_get_handle_ats_connectivity,
                                        &oprelease_get_handle_ats_connectivity);
    entry->op_ats_connectivity = op;
    break;
  }
  GNUNET_TESTBED_operation_queue_insert_ (GST_opq_openfds, op);
  GNUNET_TESTBED_operation_begin_wait_ (op);
  return gh;
}


void
GST_connection_pool_get_handle_done (struct GST_ConnectionPool_GetHandle *gh)
{
  struct PooledConnection *entry;

  if (NULL == gh)
    return;
  entry = gh->entry;
  LOG_DEBUG ("Cleaning up get handle %p for service %u, peer %u\n",
             gh,
             gh->service,
             entry->index);
  if (! gh->connection_ready_called)
  {
    GNUNET_CONTAINER_DLL_remove (entry->head_waiting, entry->tail_waiting, gh);
    if ((NULL == search_waiting (entry, entry->head_waiting)) &&
        (NULL != entry->notify_task))
    {
      GNUNET_SCHEDULER_cancel (entry->notify_task);
      entry->notify_task = NULL;
    }
  }
  if (gh->notify_waiting)
  {
    GNUNET_CONTAINER_DLL_remove (entry->head_notify, entry->tail_notify, gh);
    gh->notify_waiting = 0;
  }
  GNUNET_free (gh);
  gh = NULL;
  GNUNET_assert (! entry->in_lru);
  if (! entry->in_pool)
    GNUNET_CONTAINER_DLL_remove (head_not_pooled, tail_not_pooled, entry);
  if (NULL != map)
  {
    if (GNUNET_YES ==
        GNUNET_CONTAINER_multihashmap32_contains (map, entry->index))
      goto unallocate;
    if (GNUNET_CONTAINER_multihashmap32_size (map) == max_size)
    {
      if (NULL == head_lru)
        goto unallocate;
      destroy_pooled_connection (head_lru);
    }
    GNUNET_assert (GNUNET_OK ==
                   GNUNET_CONTAINER_multihashmap32_put (
                     map,
                     entry->index,
                     entry,
                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
    entry->in_pool = GNUNET_YES;
  }

unallocate:
  GNUNET_assert (0 < entry->demand);
  entry->demand--;
  if (0 != entry->demand)
    return;
  if (entry->in_pool)
  {
    add_to_lru (entry);
    return;
  }
  destroy_pooled_connection (entry);
}