transport_api_monitor_plugins.c

Go to the documentation of this file.

/*
     This file is part of GNUnet.
     Copyright (C) 2014, 2016 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 "platform.h"
#include "gnunet_util_lib.h"
#include "gnunet_arm_service.h"
#include "gnunet_hello_lib.h"
#include "gnunet_protocols.h"
#include "gnunet_transport_service.h"
#include "transport.h"


struct GNUNET_TRANSPORT_PluginMonitor
{

  struct GNUNET_MQ_Handle *mq;

  const struct GNUNET_CONFIGURATION_Handle *cfg;

  GNUNET_TRANSPORT_SessionMonitorCallback cb;

  void *cb_cls;

  struct GNUNET_CONTAINER_MultiHashMap32 *sessions;

  struct GNUNET_TIME_Relative backoff;

  struct GNUNET_SCHEDULER_Task *reconnect_task;

};


struct GNUNET_TRANSPORT_PluginSession
{
  uint64_t session_id;

  void *client_ctx;
};



static void
do_plugin_connect (void *cls);


static int
free_entry (void *cls,
            uint32_t key,
            void *value)
{
  struct GNUNET_TRANSPORT_PluginMonitor *pm = cls;
  struct GNUNET_TRANSPORT_PluginSession *ps = value;

  pm->cb (pm->cb_cls,
          ps,
          &ps->client_ctx,
          NULL);
  GNUNET_break (GNUNET_YES ==
                GNUNET_CONTAINER_multihashmap32_remove (pm->sessions,
                                                        key,
                                                        ps));
  GNUNET_break (NULL == ps->client_ctx);
  GNUNET_free (ps);
  return GNUNET_OK;
}


static void
reconnect_plugin_ctx (struct GNUNET_TRANSPORT_PluginMonitor *pm)
{
  GNUNET_MQ_destroy (pm->mq);
  pm->mq = NULL;
  GNUNET_CONTAINER_multihashmap32_iterate (pm->sessions,
                                           &free_entry,
                                           pm);
  pm->backoff = GNUNET_TIME_STD_BACKOFF (pm->backoff);
  pm->reconnect_task = GNUNET_SCHEDULER_add_delayed (pm->backoff,
                                                     &do_plugin_connect,
                                                     pm);
}


static uint32_t
wrap_id (uint64_t id)
{
  return ((uint32_t) id) ^ ((uint32_t) (id >> 32));
}


struct SearchContext
{

  struct GNUNET_TRANSPORT_PluginSession *ps;

  uint64_t session_id;

};


static int
locate_by_id (void *cls,
              uint32_t key,
              void *value)
{
  struct SearchContext *sc = cls;
  struct GNUNET_TRANSPORT_PluginSession *ps = value;

  if (sc->session_id == ps->session_id)
  {
    sc->ps = ps;
    return GNUNET_SYSERR;
  }
  return GNUNET_OK;
}


static int
check_event (void *cls,
             const struct TransportPluginMonitorMessage *tpmm)
{
  const char *pname;
  size_t pname_len;
  size_t paddr_len;

  pname = (const char *) &tpmm[1];
  pname_len = ntohs (tpmm->plugin_name_len);
  paddr_len = ntohs (tpmm->plugin_address_len);
  if ( (pname_len +
        paddr_len +
        sizeof (struct TransportPluginMonitorMessage) != ntohs (tpmm->header.size)) ||
       ( (0 != pname_len) &&
         ('\0' != pname[pname_len - 1]) ) )
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  return GNUNET_OK;
}


static void
handle_event (void *cls,
              const struct TransportPluginMonitorMessage *tpmm)
{
  struct GNUNET_TRANSPORT_PluginMonitor *pm = cls;
  struct GNUNET_TRANSPORT_PluginSession *ps;
  const char *pname;
  const void *paddr;
  enum GNUNET_TRANSPORT_SessionState ss;
  size_t pname_len;
  size_t paddr_len;
  struct GNUNET_TRANSPORT_SessionInfo info;
  struct GNUNET_HELLO_Address addr;
  struct SearchContext rv;

  pname = (const char *) &tpmm[1];
  pname_len = ntohs (tpmm->plugin_name_len);
  paddr_len = ntohs (tpmm->plugin_address_len);
  paddr = &pname[pname_len];
  ps = NULL;
  ss = (enum GNUNET_TRANSPORT_SessionState) ntohs (tpmm->session_state);
  if (GNUNET_TRANSPORT_SS_INIT == ss)
  {
    ps = GNUNET_new (struct GNUNET_TRANSPORT_PluginSession);
    ps->session_id = tpmm->session_id;
    (void) GNUNET_CONTAINER_multihashmap32_put (pm->sessions,
                                                wrap_id (tpmm->session_id),
                                                ps,
                                                GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);

  }
  else
  {
    rv.session_id = tpmm->session_id;
    rv.ps = NULL;
    (void) GNUNET_CONTAINER_multihashmap32_get_multiple (pm->sessions,
                                                         wrap_id (tpmm->session_id),
                                                         &locate_by_id,
                                                         &rv);
    ps = rv.ps;
    if (NULL == ps)
    {
      GNUNET_break (0);
      reconnect_plugin_ctx (pm);
      return;
    }
  }
  info.state = ss;
  info.is_inbound = (int16_t) ntohs (tpmm->is_inbound);
  info.num_msg_pending = ntohl (tpmm->msgs_pending);
  info.num_bytes_pending = ntohl (tpmm->bytes_pending);
  info.receive_delay = GNUNET_TIME_absolute_ntoh (tpmm->delay);
  info.session_timeout = GNUNET_TIME_absolute_ntoh (tpmm->timeout);
  info.address = &addr;
  addr.peer = tpmm->peer;
  addr.address = (0 == paddr_len) ? NULL : paddr;
  addr.address_length = paddr_len;
  addr.transport_name = (0 == pname_len) ? NULL : pname;
  addr.local_info = GNUNET_HELLO_ADDRESS_INFO_NONE;
  pm->cb (pm->cb_cls,
          ps,
          &ps->client_ctx,
          &info);

  if (GNUNET_TRANSPORT_SS_DONE == ss)
  {
    GNUNET_break (NULL == ps->client_ctx);
    GNUNET_assert (GNUNET_YES ==
                   GNUNET_CONTAINER_multihashmap32_remove (pm->sessions,
                                                           wrap_id (tpmm->session_id),
                                                           ps));
    GNUNET_free (ps);
  }
}


static void
handle_sync (void *cls,
             const struct GNUNET_MessageHeader *msg)
{
  struct GNUNET_TRANSPORT_PluginMonitor *pm = cls;

  /* we are in sync, notify callback */
  pm->cb (pm->cb_cls,
          NULL,
          NULL,
          NULL);
}


static void
mq_error_handler (void *cls,
                  enum GNUNET_MQ_Error error)
{
  struct GNUNET_TRANSPORT_PluginMonitor *pm = cls;

  reconnect_plugin_ctx (pm);
}


static void
do_plugin_connect (void *cls)
{
  struct GNUNET_TRANSPORT_PluginMonitor *pm = cls;
  struct GNUNET_MQ_MessageHandler handlers[] = {
    GNUNET_MQ_hd_var_size (event,
                           GNUNET_MESSAGE_TYPE_TRANSPORT_MONITOR_PLUGIN_EVENT,
                           struct TransportPluginMonitorMessage,
                           pm),
    GNUNET_MQ_hd_fixed_size (sync,
                             GNUNET_MESSAGE_TYPE_TRANSPORT_MONITOR_PLUGIN_SYNC,
                             struct GNUNET_MessageHeader,
                             pm),
    GNUNET_MQ_handler_end ()
  };
  struct GNUNET_MessageHeader *msg;
  struct GNUNET_MQ_Envelope *env;

  pm->reconnect_task = NULL;
  pm->mq = GNUNET_CLIENT_connect (pm->cfg,
                                  "transport",
                                  handlers,
                                  &mq_error_handler,
                                  pm);
  if (NULL == pm->mq)
    return;
  env = GNUNET_MQ_msg (msg,
                       GNUNET_MESSAGE_TYPE_TRANSPORT_MONITOR_PLUGIN_START);
  GNUNET_MQ_send (pm->mq,
                  env);
}


struct GNUNET_TRANSPORT_PluginMonitor *
GNUNET_TRANSPORT_monitor_plugins (const struct GNUNET_CONFIGURATION_Handle *cfg,
                                  GNUNET_TRANSPORT_SessionMonitorCallback cb,
                                  void *cb_cls)
{
  struct GNUNET_TRANSPORT_PluginMonitor *pm;

  pm = GNUNET_new (struct GNUNET_TRANSPORT_PluginMonitor);
  pm->cb = cb;
  pm->cb_cls = cb_cls;
  pm->cfg = cfg;
  do_plugin_connect (pm);
  if (NULL == pm->mq)
  {
    GNUNET_free (pm);
    return NULL;
  }
  pm->sessions = GNUNET_CONTAINER_multihashmap32_create (128);
  return pm;
}


void
GNUNET_TRANSPORT_monitor_plugins_cancel (struct GNUNET_TRANSPORT_PluginMonitor *pm)
{
  if (NULL != pm->mq)
  {
    GNUNET_MQ_destroy (pm->mq);
    pm->mq = NULL;
  }
  if (NULL != pm->reconnect_task)
  {
    GNUNET_SCHEDULER_cancel (pm->reconnect_task);
    pm->reconnect_task = NULL;
  }
  GNUNET_CONTAINER_multihashmap32_iterate (pm->sessions,
                                           &free_entry,
                                           pm);
  GNUNET_CONTAINER_multihashmap32_destroy (pm->sessions);
  GNUNET_free (pm);
}


/* end of transport_api_monitor_plugins.c */