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);
}