transport_api2_communication.c

Go to the documentation of this file.

/*
     This file is part of GNUnet.
     Copyright (C) 2018 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_protocols.h"
#include "gnunet_transport_communication_service.h"
#include "gnunet_ats_transport_service.h"
#include "transport.h"


#define DEFAULT_MAX_QUEUE_LENGTH 16


struct FlowControl {
  struct FlowControl *next;

  struct FlowControl *prev;

  GNUNET_TRANSPORT_MessageCompletedCallback cb;

  void *cb_cls;

  struct GNUNET_PeerIdentity sender;

  uint64_t id;
};


struct AckPending {
  struct AckPending *next;

  struct AckPending *prev;

  struct GNUNET_TRANSPORT_CommunicatorHandle *ch;

  struct GNUNET_PeerIdentity receiver;

  uint64_t mid;
};


struct GNUNET_TRANSPORT_CommunicatorHandle {
  struct GNUNET_TRANSPORT_AddressIdentifier *ai_head;

  struct GNUNET_TRANSPORT_AddressIdentifier *ai_tail;

  struct FlowControl *fc_head;

  struct FlowControl *fc_tail;

  struct AckPending *ap_head;

  struct AckPending *ap_tail;

  struct GNUNET_TRANSPORT_QueueHandle *queue_head;

  struct GNUNET_TRANSPORT_QueueHandle *queue_tail;

  const struct GNUNET_CONFIGURATION_Handle *cfg;

  const char *config_section;

  const char *addr_prefix;

  GNUNET_TRANSPORT_CommunicatorMqInit mq_init;

  void *mq_init_cls;

  GNUNET_TRANSPORT_CommunicatorNotify notify_cb;

  void *notify_cb_cls;

  struct GNUNET_MQ_Handle *mq;

  unsigned long long max_queue_length;

  uint64_t fc_gen;

  uint32_t aid_gen;

  uint32_t queue_gen;

  enum GNUNET_TRANSPORT_CommunicatorCharacteristics cc;
};


struct GNUNET_TRANSPORT_QueueHandle {
  struct GNUNET_TRANSPORT_QueueHandle *next;

  struct GNUNET_TRANSPORT_QueueHandle *prev;

  struct GNUNET_TRANSPORT_CommunicatorHandle *ch;

  char *address;

  struct GNUNET_MQ_Handle *mq;

  struct GNUNET_PeerIdentity peer;

  enum GNUNET_NetworkType nt;

  enum GNUNET_TRANSPORT_ConnectionStatus cs;

  uint32_t queue_id;

  uint32_t mtu;
};


struct GNUNET_TRANSPORT_AddressIdentifier {
  struct GNUNET_TRANSPORT_AddressIdentifier *next;

  struct GNUNET_TRANSPORT_AddressIdentifier *prev;

  struct GNUNET_TRANSPORT_CommunicatorHandle *ch;

  char *address;

  struct GNUNET_TIME_Relative expiration;

  uint32_t aid;

  enum GNUNET_NetworkType nt;
};


static void
reconnect(struct GNUNET_TRANSPORT_CommunicatorHandle *ch);


static void
send_add_address(struct GNUNET_TRANSPORT_AddressIdentifier *ai)
{
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_TRANSPORT_AddAddressMessage *aam;

  if (NULL == ai->ch->mq)
    return;
  env = GNUNET_MQ_msg_extra(aam,
                            strlen(ai->address) + 1,
                            GNUNET_MESSAGE_TYPE_TRANSPORT_ADD_ADDRESS);
  aam->expiration = GNUNET_TIME_relative_hton(ai->expiration);
  aam->nt = htonl((uint32_t)ai->nt);
  memcpy(&aam[1], ai->address, strlen(ai->address) + 1);
  GNUNET_MQ_send(ai->ch->mq, env);
}


static void
send_del_address(struct GNUNET_TRANSPORT_AddressIdentifier *ai)
{
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_TRANSPORT_DelAddressMessage *dam;

  if (NULL == ai->ch->mq)
    return;
  env = GNUNET_MQ_msg(dam, GNUNET_MESSAGE_TYPE_TRANSPORT_DEL_ADDRESS);
  dam->aid = htonl(ai->aid);
  GNUNET_MQ_send(ai->ch->mq, env);
}


static void
send_add_queue(struct GNUNET_TRANSPORT_QueueHandle *qh)
{
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_TRANSPORT_AddQueueMessage *aqm;

  if (NULL == qh->ch->mq)
    return;
  env = GNUNET_MQ_msg_extra(aqm,
                            strlen(qh->address) + 1,
                            GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_SETUP);
  aqm->qid = htonl(qh->queue_id);
  aqm->receiver = qh->peer;
  aqm->nt = htonl((uint32_t)qh->nt);
  aqm->mtu = htonl(qh->mtu);
  aqm->cs = htonl((uint32_t)qh->cs);
  memcpy(&aqm[1], qh->address, strlen(qh->address) + 1);
  GNUNET_MQ_send(qh->ch->mq, env);
}


static void
send_del_queue(struct GNUNET_TRANSPORT_QueueHandle *qh)
{
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_TRANSPORT_DelQueueMessage *dqm;

  if (NULL == qh->ch->mq)
    return;
  env = GNUNET_MQ_msg(dqm, GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_TEARDOWN);
  dqm->qid = htonl(qh->queue_id);
  dqm->receiver = qh->peer;
  GNUNET_MQ_send(qh->ch->mq, env);
}


static void
disconnect(struct GNUNET_TRANSPORT_CommunicatorHandle *ch)
{
  struct FlowControl *fcn;
  struct AckPending *apn;

  for (struct FlowControl *fc = ch->fc_head; NULL != fc; fc = fcn)
    {
      fcn = fc->next;
      GNUNET_CONTAINER_DLL_remove(ch->fc_head, ch->fc_tail, fc);
      fc->cb(fc->cb_cls, GNUNET_SYSERR);
      GNUNET_free(fc);
    }
  for (struct AckPending *ap = ch->ap_head; NULL != ap; ap = apn)
    {
      apn = ap->next;
      GNUNET_CONTAINER_DLL_remove(ch->ap_head, ch->ap_tail, ap);
      GNUNET_free(ap);
    }
  if (NULL == ch->mq)
    return;
  GNUNET_MQ_destroy(ch->mq);
  ch->mq = NULL;
}


static void
error_handler(void *cls, enum GNUNET_MQ_Error error)
{
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch = cls;

  GNUNET_log(GNUNET_ERROR_TYPE_INFO,
             "MQ failure %d, reconnecting to transport service.\n",
             error);
  disconnect(ch);
  /* TODO: maybe do this with exponential backoff/delay */
  reconnect(ch);
}


static void
handle_incoming_ack(
  void *cls,
  const struct GNUNET_TRANSPORT_IncomingMessageAck *incoming_ack)
{
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch = cls;

  for (struct FlowControl *fc = ch->fc_head; NULL != fc; fc = fc->next)
    {
      if ((fc->id == incoming_ack->fc_id) &&
          (0 == memcmp(&fc->sender,
                       &incoming_ack->sender,
                       sizeof(struct GNUNET_PeerIdentity))))
        {
          GNUNET_CONTAINER_DLL_remove(ch->fc_head, ch->fc_tail, fc);
          fc->cb(fc->cb_cls, GNUNET_OK);
          GNUNET_free(fc);
          return;
        }
    }
  GNUNET_break(0);
  disconnect(ch);
  /* TODO: maybe do this with exponential backoff/delay */
  reconnect(ch);
}


static int
check_create_queue(void *cls, const struct GNUNET_TRANSPORT_CreateQueue *cq)
{
  (void)cls;
  GNUNET_MQ_check_zero_termination(cq);
  return GNUNET_OK;
}


static void
handle_create_queue(void *cls, const struct GNUNET_TRANSPORT_CreateQueue *cq)
{
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch = cls;
  const char *addr = (const char *)&cq[1];
  struct GNUNET_TRANSPORT_CreateQueueResponse *cqr;
  struct GNUNET_MQ_Envelope *env;

  if (GNUNET_OK != ch->mq_init(ch->mq_init_cls, &cq->receiver, addr))
    {
      GNUNET_log(GNUNET_ERROR_TYPE_ERROR,
                 "Address `%s' invalid for this communicator\n",
                 addr);
      env = GNUNET_MQ_msg(cqr, GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_CREATE_FAIL);
    }
  else
    {
      env = GNUNET_MQ_msg(cqr, GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_CREATE_OK);
    }
  cqr->request_id = cq->request_id;
  GNUNET_MQ_send(ch->mq, env);
}


static int
check_send_msg(void *cls, const struct GNUNET_TRANSPORT_SendMessageTo *smt)
{
  (void)cls;
  GNUNET_MQ_check_boxed_message(smt);
  return GNUNET_OK;
}


static void
send_ack(struct GNUNET_TRANSPORT_CommunicatorHandle *ch,
         int status,
         const struct GNUNET_PeerIdentity *receiver,
         uint64_t mid)
{
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_TRANSPORT_SendMessageToAck *ack;

  env = GNUNET_MQ_msg(ack, GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_MSG_ACK);
  ack->status = htonl(status);
  ack->mid = mid;
  ack->receiver = *receiver;
  GNUNET_MQ_send(ch->mq, env);
}


static void
send_ack_cb(void *cls)
{
  struct AckPending *ap = cls;
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch = ap->ch;

  GNUNET_CONTAINER_DLL_remove(ch->ap_head, ch->ap_tail, ap);
  send_ack(ch, GNUNET_OK, &ap->receiver, ap->mid);
  GNUNET_free(ap);
}


static void
handle_send_msg(void *cls, const struct GNUNET_TRANSPORT_SendMessageTo *smt)
{
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch = cls;
  const struct GNUNET_MessageHeader *mh;
  struct GNUNET_MQ_Envelope *env;
  struct AckPending *ap;
  struct GNUNET_TRANSPORT_QueueHandle *qh;

  for (qh = ch->queue_head; NULL != qh; qh = qh->next)
    if ((qh->queue_id == smt->qid) &&
        (0 == memcmp(&qh->peer,
                     &smt->receiver,
                     sizeof(struct GNUNET_PeerIdentity))))
      break;
  if (NULL == qh)
    {
      /* queue is already gone, tell transport this one failed */
      GNUNET_log(GNUNET_ERROR_TYPE_INFO,
                 "Transmission failed, queue no longer exists.\n");
      send_ack(ch, GNUNET_NO, &smt->receiver, smt->mid);
      return;
    }
  ap = GNUNET_new(struct AckPending);
  ap->ch = ch;
  ap->receiver = smt->receiver;
  ap->mid = smt->mid;
  GNUNET_CONTAINER_DLL_insert(ch->ap_head, ch->ap_tail, ap);
  mh = (const struct GNUNET_MessageHeader *)&smt[1];
  env = GNUNET_MQ_msg_copy(mh);
  GNUNET_MQ_notify_sent(env, &send_ack_cb, ap);
  GNUNET_MQ_send(qh->mq, env);
}


static int
check_backchannel_incoming(
  void *cls,
  const struct GNUNET_TRANSPORT_CommunicatorBackchannelIncoming *bi)
{
  (void)cls;
  GNUNET_MQ_check_boxed_message(bi);
  return GNUNET_OK;
}


static void
handle_backchannel_incoming(
  void *cls,
  const struct GNUNET_TRANSPORT_CommunicatorBackchannelIncoming *bi)
{
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch = cls;

  if (NULL != ch->notify_cb)
    ch->notify_cb(ch->notify_cb_cls,
                  &bi->pid,
                  (const struct GNUNET_MessageHeader *)&bi[1]);
  else
    GNUNET_log(
      GNUNET_ERROR_TYPE_INFO,
      _("Dropped backchanel message: handler not provided by communicator\n"));
}


static void
reconnect(struct GNUNET_TRANSPORT_CommunicatorHandle *ch)
{
  struct GNUNET_MQ_MessageHandler handlers[] =
  { GNUNET_MQ_hd_fixed_size(incoming_ack,
                            GNUNET_MESSAGE_TYPE_TRANSPORT_INCOMING_MSG_ACK,
                            struct GNUNET_TRANSPORT_IncomingMessageAck,
                            ch),
    GNUNET_MQ_hd_var_size(create_queue,
                          GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_CREATE,
                          struct GNUNET_TRANSPORT_CreateQueue,
                          ch),
    GNUNET_MQ_hd_var_size(send_msg,
                          GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_MSG,
                          struct GNUNET_TRANSPORT_SendMessageTo,
                          ch),
    GNUNET_MQ_hd_var_size(
      backchannel_incoming,
      GNUNET_MESSAGE_TYPE_TRANSPORT_COMMUNICATOR_BACKCHANNEL_INCOMING,
      struct GNUNET_TRANSPORT_CommunicatorBackchannelIncoming,
      ch),
    GNUNET_MQ_handler_end() };
  struct GNUNET_TRANSPORT_CommunicatorAvailableMessage *cam;
  struct GNUNET_MQ_Envelope *env;

  ch->mq =
    GNUNET_CLIENT_connect(ch->cfg, "transport", handlers, &error_handler, ch);
  if (NULL == ch->mq)
    return;
  env = GNUNET_MQ_msg_extra(cam,
                            strlen(ch->addr_prefix) + 1,
                            GNUNET_MESSAGE_TYPE_TRANSPORT_NEW_COMMUNICATOR);
  cam->cc = htonl((uint32_t)ch->cc);
  memcpy(&cam[1], ch->addr_prefix, strlen(ch->addr_prefix) + 1);
  GNUNET_MQ_send(ch->mq, env);
  for (struct GNUNET_TRANSPORT_AddressIdentifier *ai = ch->ai_head; NULL != ai;
       ai = ai->next)
    send_add_address(ai);
  for (struct GNUNET_TRANSPORT_QueueHandle *qh = ch->queue_head; NULL != qh;
       qh = qh->next)
    send_add_queue(qh);
}


struct GNUNET_TRANSPORT_CommunicatorHandle *
GNUNET_TRANSPORT_communicator_connect(
  const struct GNUNET_CONFIGURATION_Handle *cfg,
  const char *config_section,
  const char *addr_prefix,
  enum GNUNET_TRANSPORT_CommunicatorCharacteristics cc,
  GNUNET_TRANSPORT_CommunicatorMqInit mq_init,
  void *mq_init_cls,
  GNUNET_TRANSPORT_CommunicatorNotify notify_cb,
  void *notify_cb_cls)
{
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch;

  ch = GNUNET_new(struct GNUNET_TRANSPORT_CommunicatorHandle);
  ch->cfg = cfg;
  ch->config_section = config_section;
  ch->addr_prefix = addr_prefix;
  ch->mq_init = mq_init;
  ch->mq_init_cls = mq_init_cls;
  ch->notify_cb = notify_cb;
  ch->notify_cb_cls = notify_cb_cls;
  ch->cc = cc;
  reconnect(ch);
  if (GNUNET_OK !=
      GNUNET_CONFIGURATION_get_value_number(cfg,
                                            config_section,
                                            "MAX_QUEUE_LENGTH",
                                            &ch->max_queue_length))
    ch->max_queue_length = DEFAULT_MAX_QUEUE_LENGTH;
  if (NULL == ch->mq)
    {
      GNUNET_free(ch);
      return NULL;
    }
  return ch;
}


void
GNUNET_TRANSPORT_communicator_disconnect(
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch)
{
  disconnect(ch);
  while (NULL != ch->ai_head)
    {
      GNUNET_break(0); /* communicator forgot to remove address, warn! */
      GNUNET_TRANSPORT_communicator_address_remove(ch->ai_head);
    }
  GNUNET_free(ch);
}


/* ************************* Receiving *************************** */


int
GNUNET_TRANSPORT_communicator_receive(
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch,
  const struct GNUNET_PeerIdentity *sender,
  const struct GNUNET_MessageHeader *msg,
  struct GNUNET_TIME_Relative expected_addr_validity,
  GNUNET_TRANSPORT_MessageCompletedCallback cb,
  void *cb_cls)
{
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_TRANSPORT_IncomingMessage *im;
  uint16_t msize;

  if (NULL == ch->mq)
    return GNUNET_SYSERR;
  if ((NULL == cb) && (GNUNET_MQ_get_length(ch->mq) >= ch->max_queue_length))
    {
      GNUNET_log(
        GNUNET_ERROR_TYPE_WARNING,
        "Dropping message: transprot is too slow, queue length %llu exceeded\n",
        ch->max_queue_length);
      return GNUNET_NO;
    }

  msize = ntohs(msg->size);
  env =
    GNUNET_MQ_msg_extra(im, msize, GNUNET_MESSAGE_TYPE_TRANSPORT_INCOMING_MSG);
  if (NULL == env)
    {
      GNUNET_break(0);
      return GNUNET_SYSERR;
    }
  im->expected_address_validity =
    GNUNET_TIME_relative_hton(expected_addr_validity);
  im->sender = *sender;
  memcpy(&im[1], msg, msize);
  if (NULL != cb)
    {
      struct FlowControl *fc;

      im->fc_on = htonl(GNUNET_YES);
      im->fc_id = ch->fc_gen++;
      fc = GNUNET_new(struct FlowControl);
      fc->sender = *sender;
      fc->id = im->fc_id;
      fc->cb = cb;
      fc->cb_cls = cb_cls;
      GNUNET_CONTAINER_DLL_insert(ch->fc_head, ch->fc_tail, fc);
    }
  GNUNET_MQ_send(ch->mq, env);
  return GNUNET_OK;
}


/* ************************* Discovery *************************** */


struct GNUNET_TRANSPORT_QueueHandle *
GNUNET_TRANSPORT_communicator_mq_add(
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch,
  const struct GNUNET_PeerIdentity *peer,
  const char *address,
  uint32_t mtu,
  enum GNUNET_NetworkType nt,
  enum GNUNET_TRANSPORT_ConnectionStatus cs,
  struct GNUNET_MQ_Handle *mq)
{
  struct GNUNET_TRANSPORT_QueueHandle *qh;

  qh = GNUNET_new(struct GNUNET_TRANSPORT_QueueHandle);
  qh->ch = ch;
  qh->peer = *peer;
  qh->address = GNUNET_strdup(address);
  qh->nt = nt;
  qh->mtu = mtu;
  qh->cs = cs;
  qh->mq = mq;
  qh->queue_id = ch->queue_gen++;
  GNUNET_CONTAINER_DLL_insert(ch->queue_head, ch->queue_tail, qh);
  send_add_queue(qh);
  return qh;
}


void
GNUNET_TRANSPORT_communicator_mq_del(struct GNUNET_TRANSPORT_QueueHandle *qh)
{
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch = qh->ch;

  send_del_queue(qh);
  GNUNET_CONTAINER_DLL_remove(ch->queue_head, ch->queue_tail, qh);
  GNUNET_MQ_destroy(qh->mq);
  GNUNET_free(qh->address);
  GNUNET_free(qh);
}


struct GNUNET_TRANSPORT_AddressIdentifier *
GNUNET_TRANSPORT_communicator_address_add(
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch,
  const char *address,
  enum GNUNET_NetworkType nt,
  struct GNUNET_TIME_Relative expiration)
{
  struct GNUNET_TRANSPORT_AddressIdentifier *ai;

  ai = GNUNET_new(struct GNUNET_TRANSPORT_AddressIdentifier);
  ai->ch = ch;
  ai->address = GNUNET_strdup(address);
  ai->nt = nt;
  ai->expiration = expiration;
  ai->aid = ch->aid_gen++;
  GNUNET_CONTAINER_DLL_insert(ch->ai_head, ch->ai_tail, ai);
  send_add_address(ai);
  return ai;
}


void
GNUNET_TRANSPORT_communicator_address_remove(
  struct GNUNET_TRANSPORT_AddressIdentifier *ai)
{
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch = ai->ch;

  send_del_address(ai);
  GNUNET_CONTAINER_DLL_remove(ch->ai_head, ch->ai_tail, ai);
  GNUNET_free(ai->address);
  GNUNET_free(ai);
}


/* ************************* Backchannel *************************** */


void
GNUNET_TRANSPORT_communicator_notify(
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch,
  const struct GNUNET_PeerIdentity *pid,
  const char *comm,
  const struct GNUNET_MessageHeader *header)
{
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_TRANSPORT_CommunicatorBackchannel *cb;
  size_t slen = strlen(comm) + 1;
  uint16_t mlen = ntohs(header->size);

  GNUNET_assert(mlen + slen + sizeof(*cb) < UINT16_MAX);
  env =
    GNUNET_MQ_msg_extra(cb,
                        slen + mlen,
                        GNUNET_MESSAGE_TYPE_TRANSPORT_COMMUNICATOR_BACKCHANNEL);
  cb->pid = *pid;
  memcpy(&cb[1], header, mlen);
  memcpy(((char *)&cb[1]) + mlen, comm, slen);
  GNUNET_MQ_send(ch->mq, env);
}


/* end of transport_api2_communication.c */