rehash/c-code/rehash-api.c

/* rehash --- a decentralised hash<->hash store
   Copyright © 2020 Maxime Devos <maxime.devos@student.kuleuven.be>

   This file is part of rehash.

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

   rehash 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 General Public License for more details.

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

/**
 * @brief Client-side API to the rehash service
 * @author Maxime Devos
 */

#include "platform.h"
#include <stdio.h>
#include <gnunet/gnunet_service_lib.h>
#include <gnunet/gnunet_dht_service.h>
#include <gnunet/gnunet_mq_lib.h>

#include "rehash_service.h"
#include "extra_gnunet_protocols.h"
#include "rehash.h"

/**
 * Type of operation in progress
 */
enum ContextType
{
  /* to be inserted into the queue
     (not visible to the API user) */
  CONTEXT_QUERY_QUEUE_ME,
  /* to be sent to rehash service */
  CONTEXT_QUERY_QUEUED,
  /* sent to rehash service, awaiting replies */
  CONTEXT_QUERY_SENT,
  /* to be inserted into the queue
     (not visible to the API user) */
  CONTEXT_PUT_QUEUE_ME,
  /* does not have request id,
     is (or is about to be) in the store queue */
  CONTEXT_PUT_QUEUED,
  /* has a request_id, is not in the store queue */
  CONTEXT_PUT_SENT,
  /* the rehash service considers the insertion
     to be completed! (Not visible to the API
     user, as the context will be freed immediately
     with REHASH_store_abort) */
  CONTEXT_PUT_COMPLETED,
  /* the context has been freed (e.g. by abortion).
     Should never be encountered in practice. */
  CONTEXT_PUT_FREED,
};

struct REHASH_QueryContext
{
  enum ContextType type;
  /* if CONTEXT_QUERY_SENT */
  uint32_t request_id;
  /* always */
  REHASH_QueryContinuation cont;
  /* always (NULL can be ok) */
  void *cls;
  /* if CONTEXT_QUERY_QUEUED */
  struct REHASH_QueryContext *prev;
  /* if CONTEXT_QUERY_QUEUED */
  struct REHASH_QueryContext *next;
  /* if CONTEXT_QUERY_QUEUE_ME or CONTEXT_QUERY_QUEUED*/
  struct GNUNET_MQ_Envelope *ev;
  /* if CONTEXT_QUERY_QUEUE_ME or CONTEXT_QUERY_QUEUED */
  struct REHASH_GetMessage *msg;
  /* always */
  struct REHASH_Handle *h;
};

struct REHASH_StoreContext
{
  enum ContextType type;
  /* if CONTEXT_PUT_SENT */
  uint32_t request_id;
  /* always */
  REHASH_StoreContinuation cont;
  /* always */
  void *cls;
  /* if CONTEXT_PUT_QUEUED */
  struct REHASH_StoreContext *prev;
  /* if CONTEXT_PUT_QUEUED */
  struct REHASH_StoreContext *next;
  /* if CONTEXT_PUT_QUEUE_ME or CONTEXT_PUT_QUEUED */
  struct GNUNET_MQ_Envelope *ev;
  /* if CONTEXT_PUT_QUEUE_ME or CONTEXT_PUT_QUEUED */
  struct REHASH_PutMessage *msg;
  /* always */
  struct REHASH_Handle *h;
};

struct REHASH_Handle
{
  const struct GNUNET_CONFIGURATION_Handle *cfg;
  struct GNUNET_MQ_Handle *mq;
  /* FIXME: GNUNET_MQ_assoc_add panics on request_id wrap-around */

  /* Linked list of queries not yet sent to the rehash service,
     in need of a request_id (CONTEXT_QUERY_QUEUED) */
  struct REHASH_QueryContext *query_queue_head;
  struct REHASH_QueryContext *query_queue_tail;
  /* Likewise (CONTEXT_PUT_QUEUED) */
  struct REHASH_StoreContext *store_queue_head;
  struct REHASH_StoreContext *store_queue_tail;
  /**
   * Exponential back-off for trying to reconnect.
   */
  struct GNUNET_TIME_Relative retry_time;
  /*
   * Task for automatically reconnecting.
   * NULL if no such task is scheduled.
   */
  struct GNUNET_SCHEDULER_Task *reconnect_task;
};

static void
process_store_queue (struct REHASH_Handle *h);
static void
process_query_queue (struct REHASH_Handle *h);
static void
retry_connect_repeatedly (void *cls);

static int
check_client_result(void *cls, const struct REHASH_ResultMessage *msg)
{
  if (ntohl(msg->output_length) == ntohs(msg->header.size) - sizeof(*msg))
    return GNUNET_OK;
  else
    return GNUNET_SYSERR;
};

static void
handle_client_result(void *cls, const struct REHASH_ResultMessage *msg)
{
  /* TODO call callbacks */
  GNUNET_assert (0);
};

static void
handle_put_done (void *cls, const struct REHASH_PutStatusMessage *msg)
{
  struct REHASH_Handle *h;
  struct REHASH_StoreContext *ctx;
  REHASH_StoreContinuation cont;
  void *cont_cls;
  void *put_cls;
  uint32_t flags;

  flags = ntohl(msg->flags);
  /* Detect unsupported required flags.
     TODO don't abort. */
  GNUNET_assert (0 == ((flags & 0xffff) & ~(REHASH_PUT_COMPLETED)));
  /* flags > 0xffff don't have to be recognised. */
  if (! (flags & REHASH_PUT_COMPLETED))
    /* nothing interesting happened */
    return;

  h = cls;
  ctx = GNUNET_MQ_assoc_get (h->mq, msg->request_id);
  /* TODO handle this case gracefully */
  GNUNET_assert (ctx != NULL);
  /* TODO strictly speaking this acces can be a strict-aliasing
     violation */
  GNUNET_assert (ctx->type == CONTEXT_PUT_SENT);
  cont = ctx->cont;
  cont_cls = ctx->cls;

  /* First free the structure, then call the callback.
     That way misuse of a completed query in the callback
     will be caught with segfaults (probabilistically) */
  ctx->type = CONTEXT_PUT_COMPLETED;
  REHASH_store_abort (ctx);
  cont (cont_cls);
}

static void
mq_error_handler (void *h, enum GNUNET_MQ_Error e)
{
  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
	      "Connection to rehash service lost. Trying to reconnect\n");
  struct REHASH_Handle *handle = h;
  struct GNUNET_MQ_Handle *mq = handle->mq;
  /* FIXME investigate why this happens */
  GNUNET_assert (e == GNUNET_MQ_ERROR_READ);
  /* FIXME handle other cases as well */
  GNUNET_assert (handle->reconnect_task == NULL);
  handle->mq = NULL;
  GNUNET_MQ_destroy (mq);
  retry_connect_repeatedly (handle);
}

int
REHASH_connected (struct REHASH_Handle *h)
{
  GNUNET_assert (h != NULL);
  if (h->mq != NULL)
    return GNUNET_YES;
  return GNUNET_NO;
}

/**
 * Call when connected successfully
 * to handle backlog due to an offline rehash service.
 */
static void
when_successfully_connected (struct REHASH_Handle *handle)
{
  struct GNUNET_SCHEDULER_Task *reconnect_task;
  /* Handle backlog */
  /* FIXME stores and queries will need to be sent again */
  process_query_queue (handle);
  process_store_queue (handle);
  handle->retry_time = GNUNET_TIME_relative_get_zero_ ();
  if (reconnect_task = handle->reconnect_task) {
    handle->reconnect_task = NULL;
    GNUNET_SCHEDULER_cancel (reconnect_task);
  }
}

/**
 * Try to connect to the rehash service.
 * If successful, cancel the reconnect_task (if any)
 * if any and perform backlog.
 */
static int
try_connect (struct REHASH_Handle *handle)
{
  struct GNUNET_MQ_MessageHandler handlers[] = {
    GNUNET_MQ_hd_var_size (client_result,
                           GNUNET_MESSAGE_TYPE_REHASH_CLIENT_RESULT,
                           struct REHASH_ResultMessage,
                           handle),
    GNUNET_MQ_hd_fixed_size (put_done,
			     GNUNET_MESSAGE_TYPE_REHASH_PUT_DONE,
			     struct REHASH_PutStatusMessage,
			     handle),
    GNUNET_MQ_handler_end ()
  };
  if (NULL != handle->mq) {
    GNUNET_assert (handle->reconnect_task == NULL);
    return GNUNET_YES;
  }
  handle->mq = GNUNET_CLIENT_connect (handle->cfg,
				      "rehash",
				      handlers,
				      &mq_error_handler,
				      handle);
  if (handle->mq != NULL)
    {
      when_successfully_connected (handle);
      return GNUNET_YES;
    }
  else
    return GNUNET_NO;
}

/** The reconnect task */
static void
retry_connect_repeatedly (void *cls)
{
  struct REHASH_Handle *handle;
  handle = cls;
  /* No reason for cancelling ourself! */
  /* TODO: should we free ourselves? */
  handle->reconnect_task = NULL;
  if (GNUNET_YES == try_connect (handle))
    /* success! */
    return;
  /* Increase back-off and try again later */
  handle->retry_time = GNUNET_TIME_STD_BACKOFF (handle->retry_time);
  handle->reconnect_task
    = GNUNET_SCHEDULER_add_delayed (handle->retry_time,
				    &retry_connect_repeatedly,
				    handle);
}

void
REHASH_retry_connect (struct REHASH_Handle *handle)
{
  GNUNET_assert (handle != NULL);
  try_connect (handle);
}

struct REHASH_Handle *
REHASH_connect (const struct GNUNET_CONFIGURATION_Handle *cfg)
{
  struct REHASH_Handle *handle;
  handle = GNUNET_new (struct REHASH_Handle);
  *handle = (struct REHASH_Handle) { };
  handle->cfg = cfg;
  retry_connect_repeatedly (handle);
  return handle;
}

void
REHASH_disconnect (struct REHASH_Handle *h)
{
  if (h->reconnect_task)
      GNUNET_SCHEDULER_cancel (h->reconnect_task);
  /* TODO memory for GNUNET_Configuration */
  /* TODO delete outstanding requests */
  if (h->mq)
    GNUNET_MQ_destroy (h->mq);
  GNUNET_free (h);
}

void
REHASH_store_abort (struct REHASH_StoreContext *cls)
{
  GNUNET_assert (cls != NULL);
  switch (cls->type)
    {
    case CONTEXT_PUT_QUEUED:
      /* Not yet communicated to the rehash service,
         so simply remove from the queue and free
         associated memory. */
      GNUNET_CONTAINER_DLL_remove (cls->h->store_queue_head,
				   cls->h->store_queue_tail, cls);
      GNUNET_assert (cls->ev != NULL);
      GNUNET_assert (cls->msg != NULL);
      GNUNET_free (cls->ev);
      /* cls->msg is a part of cls->ev (TODO verify),
         so don't deallocate that */
      break;
    case CONTEXT_PUT_SENT:
      /* The store request has already been send */
      /* These should not be used if CONTEXT_PUT_SENT. */
      GNUNET_assert (cls->ev == NULL);
      GNUNET_assert (cls->msg == NULL);
      GNUNET_assert (cls->prev == NULL);
      GNUNET_assert (cls->next == NULL);
      /* No use after free! */
      GNUNET_MQ_assoc_remove (cls->h->mq, cls->request_id);
      /* TODO inform rehash service we aren't interested anymore */
      break;
    case CONTEXT_PUT_COMPLETED:
      GNUNET_assert (cls->ev == NULL);
      GNUNET_assert (cls->msg == NULL);
      GNUNET_assert (cls->prev == NULL);
      GNUNET_assert (cls->next == NULL);
      /* No use after free / no leaking memory! */
      /* The rehash service won't recognise the request_id
         anymore. */
      /* TODO: abort / completed races */
      GNUNET_MQ_assoc_remove (cls->h->mq, cls->request_id);
      break;
    case CONTEXT_PUT_FREED:
      /* Warning: this case cannot always be detected!
         Strictly speaking, this is caught by the ‘default’
         case as well, but this is more informative. */
      GNUNET_assert (0 && "tried to abort a freed store context");
      break;
    default:
      GNUNET_assert (0 && "invalid cls->type");
    }
  cls->type = CONTEXT_PUT_FREED;
  GNUNET_free (cls);
}

/* TODO perhaps less copy-pasting is possible?
   Let's wait with that until everything is implemented,
   perhaps by then there will be divergence. */
void
REHASH_query_abort (struct REHASH_QueryContext *cls)
{
  GNUNET_assert (cls != NULL);
  switch (cls->type)
    {
    case CONTEXT_QUERY_QUEUED:
      /* Not yet communicated to the rehash service,
         so simply remove from the queue and free
         associated memory. */
      GNUNET_CONTAINER_DLL_remove (cls->h->query_queue_head,
				   cls->h->query_queue_tail, cls);
      GNUNET_assert (cls->ev != NULL);
      GNUNET_assert (cls->msg != NULL);
      GNUNET_free (cls->ev);
      /* cls->msg is a part of cls->ev (TODO verify),
         so don't deallocate that */
      GNUNET_free (cls);
      break;
    case CONTEXT_QUERY_SENT:
      /* The store request has already been send,
         so it cannot be removed from the query queue */
      /* These should not be used if CONTEXT_PUT_SENT. */
      GNUNET_assert (cls->ev == NULL);
      GNUNET_assert (cls->msg == NULL);
      GNUNET_assert (cls->prev == NULL);
      GNUNET_assert (cls->next == NULL);
      /* No use after free! */
      GNUNET_MQ_assoc_remove (cls->h->mq, cls->request_id);
      /* TODO inform rehash service we aren't interested anymore */
      break;
    default:
      GNUNET_assert (0 && "invalid cls->type");
    }
}
/**
 * Insert @a q into the queue
 *
 * q->request_id and q->next should not be set
 *
 * @param h handle to the rehash service
 * @param q query to queue
 */
static void
queue_query (struct REHASH_Handle *h, struct REHASH_QueryContext *q)
{
  GNUNET_assert (q->type == CONTEXT_QUERY_QUEUE_ME);
  GNUNET_assert (q->ev != NULL);
  GNUNET_assert (q->msg != NULL);
  GNUNET_assert (q->prev == NULL);
  GNUNET_assert (q->next == NULL);
  GNUNET_assert (q->h == h);
  q->type = CONTEXT_QUERY_QUEUED;
  GNUNET_CONTAINER_DLL_insert (h->query_queue_head, h->query_queue_tail, q);
}

/**
 * Insert @a q into the queue
 *
 * q->request_id and q->next should not be set
 *
 * @param h handle to the rehash service
 * @param q query to queue
 */
static void
queue_store (struct REHASH_Handle *h, struct REHASH_StoreContext *q)
{
  GNUNET_assert (q->type == CONTEXT_PUT_QUEUE_ME);
  GNUNET_assert (q->ev != NULL);
  GNUNET_assert (q->msg != NULL);
  GNUNET_assert (q->prev == NULL);
  GNUNET_assert (q->next == NULL);
  GNUNET_assert (q->h == h);
  q->type = CONTEXT_PUT_QUEUED;
  GNUNET_CONTAINER_DLL_insert (h->store_queue_head, h->store_queue_tail, q);
}

/* TODO call these two functions */

/**
 * Remove entries from the query queue and try to send
 * them to the rehash service. q->request_id and
 * q->msg->request are not expected to be set.
 *
 * @param h handle to the rehash service
 */
static void
process_query_queue (struct REHASH_Handle *h)
{
  struct REHASH_QueryContext *current;
  struct REHASH_QueryContext *next;
  struct GNUNET_MQ_Envelope *ev;
  struct REHASH_GetMessage *msg;
  if (h->mq == NULL)
    /* rehash service is offline */
    return;
  current = h->query_queue_head;
  while (current != NULL)
    {
      GNUNET_assert
	(current->type == CONTEXT_QUERY_QUEUED);
      current->request_id = GNUNET_MQ_assoc_add (h->mq, current);
      current->msg->request_id = current->request_id;
      GNUNET_MQ_send (h->mq, current->ev);
      /* These shouldn't be used anymore
         (TODO does GNUNET_MQ_send deallocate?) */
      current->ev = NULL;
      current->msg = NULL;
      next = current->next;
      /* likewise */
      current->prev = NULL;
      current->next = NULL;
      current->type = CONTEXT_QUERY_SENT;
      current = next;
    }
  h->query_queue_head = NULL;
  h->query_queue_tail = NULL;
}


/**
 * Remove entries from the store queue and try to send
 * them to the rehash service. q->request_id and
 * q->msg->request are not expected to be set.
 *
 * @param h handle to the rehash service
 */
static void
process_store_queue (struct REHASH_Handle *h)
{
  /* See process_query_queue */
  struct REHASH_StoreContext *current;
  struct REHASH_StoreContext *next;
  struct GNUNET_MQ_Envelope *ev;
  if (h->mq == NULL)
    return;
  current = h->store_queue_head;
  while (current != NULL)
    {
      GNUNET_assert
	(current->type == CONTEXT_PUT_QUEUED);
      current->request_id = GNUNET_MQ_assoc_add (h->mq, current);
      current->msg->request_id = current->request_id;
      GNUNET_MQ_send (h->mq, current->ev);
      current->ev = NULL;
      current->msg = NULL;
      next = current->next;
      current->prev = NULL;
      current->next = NULL;
      current->type = CONTEXT_PUT_SENT;
      current = next;
    }
  h->store_queue_head = NULL;
  h->store_queue_tail = NULL;
}

struct REHASH_QueryContext *
REHASH_query_start (struct REHASH_Handle *h,
		    enum GNUNET_FS_SearchOptions options,
		    uint32_t anonymity,
		    enum REHASH_Hash_Type in_type,
		    enum REHASH_Hash_Type out_type,
		    const char *input,
		    size_t input_length,
		    REHASH_QueryContinuation cont,
		    void *cls)
{
  /* TODO stub */
  struct REHASH_QueryContext *ret;
  char *msg_input;

  /* TODO proper error messages, less magic */
  /* Prevent buffer overflows! */
  GNUNET_assert (input_length <= 64);

  ret = GNUNET_new (struct REHASH_QueryContext);
  *ret = (struct REHASH_QueryContext) {};
  ret->type = CONTEXT_QUERY_QUEUE_ME;
  /* TODO clobber ret->request_id */
  ret->cont = cont;
  ret->cls = cls;
  /* TODO some refcounting mechanism */
  ret->h = h;

  ret->ev = GNUNET_MQ_msg_extra (ret->msg, input_length, GNUNET_MESSAGE_TYPE_REHASH_CLIENT_GET);
  /* TODO clobber ret->msg->request_id */
  ret->msg->options = htonl (options);
  ret->msg->anonymity_level = htonl (anonymity);
  /* FIXME!  */
  ret->msg->in_type = htonl (out_type);
  ret->msg->out_type = htonl (out_type);
  ret->msg->input_length = htonl ((uint32_t) input_length);
  msg_input = (char *) &ret->msg[1];
  memcpy (msg_input, input, input_length);

  queue_query (h, ret);
  process_query_queue (h);
  return ret;
}

struct REHASH_StoreContext *
REHASH_store_start (struct REHASH_Handle *h,
		    const struct GNUNET_FS_BlockOptions *options,
		    enum REHASH_Hash_Type in_type,
		    enum REHASH_Hash_Type out_type,
		    const char *input,
		    size_t input_length,
		    const char *output,
		    size_t output_length,
		    REHASH_StoreContinuation cont,
		    void *cls)
{
  struct REHASH_StoreContext *ret;

  /* TODO proper error messages, less magic */
  /* Prevent buffer overflows! */
  GNUNET_assert (input_length <= 64);
  GNUNET_assert (output_length <= 64);

  ret = GNUNET_new (struct REHASH_StoreContext);
  *ret = (struct REHASH_StoreContext) {};
  ret->type = CONTEXT_PUT_QUEUE_ME;
  /* TODO clobber ret->request_id for debugging */
  ret->cont = cont;
  ret->cls = cls;
  ret->h = h;

  ret->ev = GNUNET_MQ_msg_extra
    (ret->msg, input_length + output_length,
     GNUNET_MESSAGE_TYPE_REHASH_CLIENT_PUT);
  ret->msg->header.size
    = htons (sizeof(*ret->msg) + input_length + output_length);
  /* TODO clobber msg->request_id */
  ret->msg->expiration_time = GNUNET_TIME_absolute_hton (options->expiration_time);
  ret->msg->anonymity_level = htonl (options->anonymity_level);
  ret->msg->content_priority = htonl (options->content_priority);
  ret->msg->replication_level = htonl (options->replication_level);
  ret->msg->in_type = htonl (in_type);
  ret->msg->in_type = htonl (out_type);
  ret->msg->input_length = htonl ((uint32_t) input_length);
  ret->msg->output_length = htonl ((uint32_t) output_length);
  /* Include input / output */
  memcpy(&ret->msg[1], input, input_length);
  memcpy(input_length + (char *) &ret->msg[1], output, output_length);
  /* FIXME include input / output */

  /* TODO maybe send directly if service is online */
  queue_store (h, ret);
  process_store_queue (h);

  /* TODO: check memory allocation / freeing */
  /* TODO: progress / abort / ... */
  return ret;
}