gnunet-service-peerstore.c

Go to the documentation of this file.

/*
     This file is part of GNUnet.
     Copyright (C) 2014, 2015, 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_peerstore_service.h"
#include "gnunet_protocols.h"
#include "gnunet_util_lib.h"
#include "peerstore.h"
#include "gnunet_peerstore_plugin.h"
#include "peerstore_common.h"
#include "gnunet_hello_uri_lib.h"
 
 
#define EXPIRED_RECORDS_CLEANUP_INTERVAL 300 /* 5mins */
 
#define MAX_RECORDS_PER_ROUND 256
 
 
struct PeerstoreClient;
 
struct Monitor
{
  struct Monitor *next;
 
  struct Monitor *prev;
 
  struct PeerstoreClient *pc;
 
  struct GNUNET_SCHEDULER_Task *task;
 
  struct GNUNET_SCHEDULER_Task *sa_wait_warning;
 
  struct GNUNET_TIME_Absolute sa_waiting_start;
 
  uint64_t seq;
 
  uint64_t limit;
 
  uint64_t iteration_cnt;
 
  int in_first_iteration;
 
  int peer_set;
 
  char *sub_system;
 
  struct GNUNET_PeerIdentity peer;
 
  char *key;
};
 
struct Iteration
{
  struct Iteration *next;
 
  struct Iteration *prev;
 
  struct PeerstoreClient *pc;
 
  uint64_t seq;
 
  uint32_t request_id;
 
  uint32_t offset;
 
  int send_end;
 
  char *sub_system;
 
  struct GNUNET_PeerIdentity peer;
 
  char *key;
 
  int peer_set;
};
 
struct PeerstoreClient
{
  struct GNUNET_SERVICE_Client *client;
 
  struct GNUNET_MQ_Handle *mq;
 
  struct Iteration *op_head;
 
  struct Iteration *op_tail;
};
 
 
struct StoreRecordContext
{
  struct GNUNET_PEERSTORE_Record *record;
 
  uint32_t rid;
 
  struct PeerstoreClient *pc;
};
 
static const struct GNUNET_CONFIGURATION_Handle *cfg;
 
static char *db_lib_name;
 
static struct GNUNET_PEERSTORE_PluginFunctions *db;
 
static struct GNUNET_SCHEDULER_Task *expire_task;
 
static struct Monitor *monitors_head;
 
static struct Monitor *monitors_tail;
 
static struct GNUNET_NotificationContext *monitor_nc;
 
static void
shutdown_task (void *cls)
{
  struct Monitor *mc;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Shutting down peerstore, bye.\n");
  /* Stop everything that could still touch the database plugin BEFORE
     unloading it. */
  if (NULL != expire_task)
  {
    GNUNET_SCHEDULER_cancel (expire_task);
    expire_task = NULL;
  }
  while (NULL != (mc = monitors_head))
  {
    GNUNET_CONTAINER_DLL_remove (monitors_head,
                                 monitors_tail,
                                 mc);
    if (NULL != mc->task)
    {
      GNUNET_SCHEDULER_cancel (mc->task);
      mc->task = NULL;
    }
    if (NULL != mc->sa_wait_warning)
    {
      GNUNET_SCHEDULER_cancel (mc->sa_wait_warning);
      mc->sa_wait_warning = NULL;
    }
    GNUNET_free (mc->sub_system);
    GNUNET_free (mc->key);
    GNUNET_free (mc);
  }
  if (NULL != monitor_nc)
  {
    GNUNET_notification_context_destroy (monitor_nc);
    monitor_nc = NULL;
  }
  if (NULL != db_lib_name)
  {
    GNUNET_break (NULL ==
                  GNUNET_PLUGIN_unload (db_lib_name,
                                        db));
    GNUNET_free (db_lib_name);
    db_lib_name = NULL;
  }
  /* NOTE: with GNUNET_SERVICE_OPTION_SOFT_SHUTDOWN clients may still be
     connected and sending requests at this point, so every handler that
     uses @e db MUST check it for NULL. */
  db = NULL;
}
 
 
/* Forward declaration */
static void
expire_records_continuation (void *cls, int success);
 
 
static void
cleanup_expired_records (void *cls)
{
  int ret;
 
  expire_task = NULL;
  GNUNET_assert (NULL != db);
  ret = db->expire_records (db->cls,
                            GNUNET_TIME_absolute_get (),
                            &expire_records_continuation,
                            NULL);
  if (GNUNET_OK != ret)
  {
    GNUNET_assert (NULL == expire_task);
    expire_task = GNUNET_SCHEDULER_add_delayed (
      GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
                                     EXPIRED_RECORDS_CLEANUP_INTERVAL),
      &cleanup_expired_records,
      NULL);
  }
}
 
 
static void
expire_records_continuation (void *cls, int success)
{
  if (success > 0)
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%d records expired.\n", success);
  GNUNET_assert (NULL == expire_task);
  expire_task = GNUNET_SCHEDULER_add_delayed (
    GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
                                   EXPIRED_RECORDS_CLEANUP_INTERVAL),
    &cleanup_expired_records,
    NULL);
}
 
 
static void
monitor_sync (struct Monitor *mc)
{
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_MessageHeader *sync;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Syncing zone monitor %p\n", mc);
 
  env = GNUNET_MQ_msg (sync, GNUNET_MESSAGE_TYPE_PEERSTORE_MONITOR_SYNC);
  GNUNET_MQ_send (mc->pc->mq, env);
  /* mark iteration done */
  mc->in_first_iteration = GNUNET_NO;
  mc->iteration_cnt = 0;
}
 
 
static void
watch_notifier (struct GNUNET_PEERSTORE_Record *record)
{
  struct GNUNET_MQ_Envelope *env;
  struct Monitor *mc;
 
  // FIXME this is very inefficient, we may want to use a hash
  // map again.
  for (mc = monitors_head; NULL != mc; mc = mc->next)
  {
    if ((GNUNET_YES == mc->peer_set) &&
        (0 != memcmp (&mc->peer, &record->peer, sizeof (record->peer))))
      continue;
    if ((NULL != mc->sub_system) &&
        (0 != strcmp (mc->sub_system, record->sub_system)))
      continue;
    if ((NULL != mc->key) &&
        (0 != strcmp (mc->key, record->key)))
      continue;
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Found watcher %p to update.\n", mc);
    if (0 == mc->limit)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                  "Watcher %p reached its limit, skipping.\n", mc);
      continue;
    }
    env = PEERSTORE_create_record_mq_envelope (
      0,
      record->sub_system,
      &record->peer,
      record->key,
      record->value,
      record->value_size,
      record->expiry,
      0,
      GNUNET_MESSAGE_TYPE_PEERSTORE_RECORD);
    GNUNET_MQ_send (mc->pc->mq, env);
    mc->limit--;
  }
}
 
 
struct IterationProcResult
{
  struct Iteration *ic;
 
  uint64_t limit;
 
};
 
static void
iterate_proc (void *cls,
              uint64_t seq,
              const struct GNUNET_PEERSTORE_Record *record,
              const char *emsg)
{
  struct IterationProcResult *proc = cls;
  struct GNUNET_MQ_Envelope *env;
 
  if (NULL != emsg)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Error iterating over peerstore: `%s'", emsg);
    return;
  }
  if (NULL == record)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Iteration done\n");
    return;
  }
  if (0 == proc->limit)
  {
    /* what is this!? should never happen */
    GNUNET_break (0);
    return;
  }
  proc->ic->seq = seq;
  env = PEERSTORE_create_record_mq_envelope (
    proc->ic->request_id,
    record->sub_system,
    &record->peer,
    record->key,
    record->value,
    record->value_size,
    record->expiry,
    0,
    GNUNET_MESSAGE_TYPE_PEERSTORE_RECORD);
  GNUNET_MQ_send (proc->ic->pc->mq, env);
  proc->limit--;
}
 
 
static void
destroy_iteration (struct Iteration *ic)
{
  GNUNET_free (ic->key);
  GNUNET_free (ic->sub_system);
  GNUNET_free (ic);
}
 
 
static void
iteration_done_client_continue (struct Iteration *ic)
{
  struct GNUNET_MQ_Envelope *env;
  struct PeerstoreResultMessage *endmsg;
 
  GNUNET_SERVICE_client_continue (ic->pc->client);
  if (! ic->send_end)
    return;
  /* No more records */
 
  env = GNUNET_MQ_msg (endmsg, GNUNET_MESSAGE_TYPE_PEERSTORE_ITERATE_END);
  endmsg->rid = htonl (ic->request_id);
  endmsg->result = htonl (GNUNET_OK);
  GNUNET_MQ_send (ic->pc->mq, env);
  GNUNET_CONTAINER_DLL_remove (ic->pc->op_head, ic->pc->op_tail, ic);
  destroy_iteration (ic);
  return;
}
 
 
static void
run_iteration_round (struct Iteration *ic, uint64_t limit)
{
  struct IterationProcResult proc;
  struct GNUNET_TIME_Absolute start;
  struct GNUNET_TIME_Relative duration;
 
  if (NULL == db)
  {
    /* post-shutdown, see #shutdown_task() */
    GNUNET_SERVICE_client_drop (ic->pc->client);
    return;
  }
  memset (&proc, 0, sizeof(proc));
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Asked to return up to %llu records at position %llu\n",
              (unsigned long long) limit,
              (unsigned long long) ic->seq);
  proc.ic = ic;
  proc.limit = limit;
  start = GNUNET_TIME_absolute_get ();
  GNUNET_break (GNUNET_SYSERR !=
                db->iterate_records (db->cls,
                                     ic->sub_system,
                                     (GNUNET_YES == ic->peer_set) ? &ic->peer :
                                     NULL,
                                     ic->key,
                                     ic->seq,
                                     proc.limit,
                                     &iterate_proc,
                                     &proc));
  duration = GNUNET_TIME_absolute_get_duration (start);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Returned %llu records in %s\n",
              (unsigned long long) (limit - proc.limit),
              GNUNET_STRINGS_relative_time_to_string (duration, GNUNET_YES));
  if (0 == proc.limit)
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Returned %llu results, more results available\n",
                (unsigned long long) limit);
  ic->send_end = (0 != proc.limit);
  iteration_done_client_continue (ic);
}
 
 
static int
check_iterate_start (void *cls, const struct
                     PeerstoreIterationStartMessage *srm)
{
  const char *ptr = (const char *) &srm[1];
  uint16_t ss_size;
  uint16_t key_size;
  uint16_t size;
 
  ss_size = ntohs (srm->sub_system_size);
  key_size = ntohs (srm->key_size);
  size = ntohs (srm->header.size);
 
  if (size < key_size + ss_size + sizeof(*srm))
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  /* the handler GNUNET_strdup()s these, so they MUST be 0-terminated
     within the space they claim */
  if ((0 != ss_size) && ('\0' != ptr[ss_size - 1]))
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  if ((0 != key_size) && ('\0' != ptr[ss_size + key_size - 1]))
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  return GNUNET_OK;
}
 
 
static void
handle_iterate_start (void *cls, const struct PeerstoreIterationStartMessage *
                      srm)
{
  struct Iteration *ic = GNUNET_new (struct Iteration);
  uint16_t ss_size;
  char *ptr;
 
  ss_size = ntohs (srm->sub_system_size);
 
  ic->pc = cls;
  ic->request_id = ntohl (srm->rid);
  ic->offset = 0;
  ic->peer_set = (ntohs (srm->peer_set)) ? GNUNET_YES : GNUNET_NO;
  if (GNUNET_YES == ic->peer_set)
    ic->peer = srm->peer;
  ptr = (char*) &srm[1];
  if (0 < ss_size)
    ic->sub_system = GNUNET_strdup (ptr);
  ptr += ss_size;
  if (0 < ntohs (srm->key_size))
    ic->key = GNUNET_strdup (ptr);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Iterate request: ss `%s', peer `%s', key `%s'\n",
              ic->sub_system,
              GNUNET_i2s (&ic->peer),
              (NULL == ic->key) ? "NULL" : ic->key);
  GNUNET_CONTAINER_DLL_insert (ic->pc->op_head,
                               ic->pc->op_tail,
                               ic);
  run_iteration_round (ic, 1);
}
 
 
static void
handle_iterate_stop (void *cls,
                     const struct PeerstoreIterationStopMessage *zis_msg)
{
  struct PeerstoreClient *pc = cls;
  struct Iteration *ic;
  uint32_t rid;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received ITERATION_STOP message\n");
  /* uint32_t, and #handle_iterate_start()/#handle_iterate_next() read it
     with ntohl().  Reading it with ntohs() here matched only rid 0, so a
     stopped iteration was never actually found: we logged "Iteration
     already finished" and leaked the `struct Iteration'. */
  rid = ntohl (zis_msg->rid);
  for (ic = pc->op_head; NULL != ic; ic = ic->next)
    if (ic->request_id == rid)
      break;
  if (NULL == ic)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Iteration already finished.\n");
    GNUNET_SERVICE_client_continue (pc->client);
    return;
  }
  GNUNET_CONTAINER_DLL_remove (pc->op_head, pc->op_tail, ic);
  destroy_iteration (ic);
  GNUNET_SERVICE_client_continue (pc->client);
}
 
 
static void
handle_iterate_next (void *cls,
                     const struct PeerstoreIterationNextMessage *is_msg)
{
  struct PeerstoreClient *pc = cls;
  struct Iteration *ic;
  uint32_t rid;
  uint64_t limit;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received ITERATION_NEXT message\n");
  rid = ntohl (is_msg->rid);
  limit = GNUNET_ntohll (is_msg->limit);
  if (0 == limit)
  {
    GNUNET_break (0);
    GNUNET_SERVICE_client_drop (pc->client);
    return;
  }
  if (limit > MAX_RECORDS_PER_ROUND)
    limit = MAX_RECORDS_PER_ROUND;
  for (ic = pc->op_head; NULL != ic; ic = ic->next)
    if (ic->request_id == rid)
      break;
  if (NULL == ic)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Not in iteration...\n");
    GNUNET_break (0);
    GNUNET_SERVICE_client_drop (pc->client);
    return;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Found iteration...\n");
  run_iteration_round (ic, limit);
}
 
 
static void
monitor_iteration_next (void *cls);
 
 
static void
monitor_iterate_cb (void *cls,
                    uint64_t seq,
                    const struct GNUNET_PEERSTORE_Record *record,
                    const char *emsg)
{
  struct Monitor *mc = cls;
  struct GNUNET_MQ_Envelope *env;
 
  if (NULL != emsg)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Error iterating over peerstore: `%s'", emsg);
    return;
  }
  if (NULL == record)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Iteration done\n");
    return;
  }
  GNUNET_assert (0 != seq);
  mc->seq = seq;
  if (0 == mc->limit)
  {
    /* what is this!? should never happen */
    GNUNET_break (0);
    return;
  }
  env = PEERSTORE_create_record_mq_envelope (
    0,
    record->sub_system,
    &record->peer,
    record->key,
    record->value,
    record->value_size,
    record->expiry,
    0,
    GNUNET_MESSAGE_TYPE_PEERSTORE_RECORD);
  GNUNET_MQ_send (mc->pc->mq,
                  env);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Sent records.\n");
  mc->limit--;
  mc->iteration_cnt--;
  if ( (0 == mc->iteration_cnt) &&
       (0 != mc->limit) )
  {
    /* We are done with the current iteration batch, AND the
       client would right now accept more, so go again! */
    GNUNET_assert (NULL == mc->task);
    mc->task = GNUNET_SCHEDULER_add_now (&monitor_iteration_next,
                                         mc);
  }
}
 
 
static void
monitor_iteration_next (void *cls)
{
  struct Monitor *mc = cls;
  int ret;
 
  mc->task = NULL;
  if (NULL == db)
    return; /* post-shutdown, see #shutdown_task() */
  GNUNET_assert (0 == mc->iteration_cnt);
  if (mc->limit > 16)
    mc->iteration_cnt = mc->limit / 2;   /* leave half for monitor events */
  else
    mc->iteration_cnt = mc->limit;   /* use it all */
  if (mc->iteration_cnt > MAX_RECORDS_PER_ROUND)
    mc->iteration_cnt = MAX_RECORDS_PER_ROUND;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Running iteration\n");
  ret = db->iterate_records (db->cls,
                             mc->sub_system,
                             (GNUNET_YES == mc->peer_set) ? &mc->peer : NULL,
                             mc->key,
                             mc->seq,
                             mc->iteration_cnt,
                             &monitor_iterate_cb,
                             mc);
  if (GNUNET_SYSERR == ret)
  {
    if (NULL != mc->task)
    {
      GNUNET_SCHEDULER_cancel (mc->task);
      mc->task = NULL;
    }
    GNUNET_CONTAINER_DLL_remove (monitors_head,
                                 monitors_tail,
                                 mc);
    GNUNET_free (mc->key);
    GNUNET_free (mc->sub_system);
    GNUNET_SERVICE_client_drop (mc->pc->client);
    GNUNET_free (mc);
    return;
  }
  if (GNUNET_NO == ret)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Zone empty... syncing\n");
    /* empty zone */
    monitor_sync (mc);
    return;
  }
}
 
 
static int
check_monitor_start (void *cls,
                     const struct PeerstoreMonitorStartMessage *srm)
{
  const char *ptr = (const char *) &srm[1];
  uint16_t ss_size;
  uint16_t key_size;
  uint16_t size;
 
  ss_size = ntohs (srm->sub_system_size);
  key_size = ntohs (srm->key_size);
  size = ntohs (srm->header.size);
  if (size < key_size + ss_size + sizeof(*srm))
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  /* the handler GNUNET_strdup()s these, so they MUST be 0-terminated
     within the space they claim */
  if ((0 != ss_size) && ('\0' != ptr[ss_size - 1]))
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  if ((0 != key_size) && ('\0' != ptr[ss_size + key_size - 1]))
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  return GNUNET_OK;
}
 
 
static void
handle_monitor_start (void *cls,
                      const struct PeerstoreMonitorStartMessage *msm)
{
  struct PeerstoreClient *pc = cls;
  struct Monitor *mc;
  uint16_t ss_size;
  char *ptr;
 
  if (NULL == monitor_nc)
  {
    GNUNET_break (0);
    GNUNET_SERVICE_client_drop (pc->client);
    return; /* post-shutdown */
  }
  /* The MONITOR_NEXT message carries no request ID, so #handle_monitor_next()
     can only ever address the first monitor of a client; refuse to create a
     second one rather than silently mis-crediting it. */
  for (mc = monitors_head; NULL != mc; mc = mc->next)
    if (mc->pc == pc)
    {
      GNUNET_break (0);
      GNUNET_SERVICE_client_drop (pc->client);
      return;
    }
  mc = GNUNET_new (struct Monitor);
  ss_size = ntohs (msm->sub_system_size);
  mc->pc = cls;
  mc->peer_set = (ntohs (msm->peer_set)) ? GNUNET_YES : GNUNET_NO;
  if (GNUNET_YES == mc->peer_set)
    mc->peer = msm->peer;
  ptr = (char*) &msm[1];
  if (0 < ss_size)
    mc->sub_system = GNUNET_strdup (ptr);
  ptr += ss_size;
  if (0 < ntohs (msm->key_size))
    mc->key = GNUNET_strdup (ptr);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Monitor request: ss `%s', peer `%s', key `%s'\n",
              mc->sub_system,
              GNUNET_i2s (&mc->peer),
              (NULL == mc->key) ? "NULL" : mc->key);
  mc->in_first_iteration = (GNUNET_YES == ntohs (msm->iterate_first));
  mc->limit = 1;
  mc->peer_set = (ntohs (msm->peer_set)) ? GNUNET_YES : GNUNET_NO;
  GNUNET_CONTAINER_DLL_insert (monitors_head,
                               monitors_tail,
                               mc);
  GNUNET_SERVICE_client_mark_monitor (mc->pc->client);
  GNUNET_SERVICE_client_continue (mc->pc->client);
  GNUNET_notification_context_add (monitor_nc,
                                   mc->pc->mq);
  if (mc->in_first_iteration)
    mc->task = GNUNET_SCHEDULER_add_now (&monitor_iteration_next,
                                         mc);
  else
    monitor_sync (mc);
}
 
 
static void
handle_monitor_next (void *cls,
                     const struct PeerstoreMonitorNextMessage *nm)
{
  struct PeerstoreClient *pc = cls;
  struct Monitor *mc;
  uint64_t inc;
 
  inc = GNUNET_ntohll (nm->limit);
  if (inc > MAX_RECORDS_PER_ROUND)
    inc = MAX_RECORDS_PER_ROUND;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received MONITOR_NEXT message with limit %llu\n",
              (unsigned long long) inc);
  for (mc = monitors_head; NULL != mc; mc = mc->next)
    if (mc->pc == pc)
      break;
  if (NULL == mc)
  {
    GNUNET_break (0);
    GNUNET_SERVICE_client_drop (pc->client);
    return;
  }
  GNUNET_SERVICE_client_continue (pc->client);
  if (mc->limit + inc < mc->limit)
  {
    GNUNET_break (0);
    GNUNET_SERVICE_client_drop (pc->client);
    return;
  }
  mc->limit += inc;
  if ((mc->in_first_iteration) && (mc->limit == inc))
  {
    /* We are still iterating, and the previous iteration must
       have stopped due to the client's limit, so continue it! */
    GNUNET_assert (NULL == mc->task);
    mc->task = GNUNET_SCHEDULER_add_now (&monitor_iteration_next, mc);
  }
  GNUNET_assert (mc->iteration_cnt <= mc->limit);
}
 
 
static void
store_record_continuation (void *cls, int success)
{
  struct StoreRecordContext *src = cls;
  struct PeerstoreResultMessage *msg;
  struct GNUNET_MQ_Envelope *env;
 
  env = GNUNET_MQ_msg (msg, GNUNET_MESSAGE_TYPE_PEERSTORE_STORE_RESULT);
  msg->rid = src->rid;
  msg->result = htonl (success);
  GNUNET_MQ_send (src->pc->mq, env);
  watch_notifier (src->record);
  GNUNET_SERVICE_client_continue (src->pc->client);
  PEERSTORE_destroy_record (src->record);
  GNUNET_free (src);
}
 
 
static int
check_store (void *cls, const struct PeerstoreRecordMessage *srm)
{
  struct GNUNET_PEERSTORE_Record *record;
 
  record = PEERSTORE_parse_record_message (srm);
  if (NULL == record)
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  if ((NULL == record->sub_system) || (NULL == record->key))
  {
    GNUNET_break (0);
    PEERSTORE_destroy_record (record);
    return GNUNET_SYSERR;
  }
  PEERSTORE_destroy_record (record);
  return GNUNET_OK;
}
 
 
static void
handle_store (void *cls, const struct PeerstoreRecordMessage *srm)
{
  struct PeerstoreClient *pc = cls;
  struct StoreRecordContext *src;
 
  if (NULL == db)
  {
    /* post-shutdown, see #shutdown_task() */
    GNUNET_SERVICE_client_drop (pc->client);
    return;
  }
  src = GNUNET_new (struct StoreRecordContext);
  src->record = PEERSTORE_parse_record_message (srm);
  GNUNET_log (
    GNUNET_ERROR_TYPE_DEBUG,
    "Received a store request. Sub system `%s' Peer `%s Key `%s' Options: %u.\n",
    src->record->sub_system,
    GNUNET_i2s (&src->record->peer),
    src->record->key,
    (uint32_t) ntohl (srm->options));
  src->rid = srm->rid;
  src->pc = pc;
  if (GNUNET_OK != db->store_record (db->cls,
                                     src->record->sub_system,
                                     &src->record->peer,
                                     src->record->key,
                                     src->record->value,
                                     src->record->value_size,
                                     src->record->expiry,
                                     ntohl (srm->options),
                                     &store_record_continuation,
                                     src))
  {
    GNUNET_break (0);
    PEERSTORE_destroy_record (src->record);
    GNUNET_free (src);
    GNUNET_SERVICE_client_drop (pc->client);
    return;
  }
}
 
 
static void *
client_connect_cb (void *cls,
                   struct GNUNET_SERVICE_Client *client,
                   struct GNUNET_MQ_Handle *mq)
{
  struct PeerstoreClient *pc;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "A client %p connected\n", client);
  pc = GNUNET_new (struct PeerstoreClient);
  pc->client = client;
  pc->mq = mq;
  return pc;
}
 
 
static void
client_disconnect_cb (void *cls,
                      struct GNUNET_SERVICE_Client *client,
                      void *app_cls)
{
  struct PeerstoreClient *pc = app_cls;
  struct Iteration *iter;
  struct Monitor *mo;
 
  (void) cls;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Client %p disconnected.\n",
              client);
  for (mo = monitors_head; NULL != mo; mo = mo->next)
  {
    if (pc != mo->pc)
      continue;
    GNUNET_CONTAINER_DLL_remove (monitors_head,
                                 monitors_tail,
                                 mo);
    if (NULL != mo->task)
    {
      GNUNET_SCHEDULER_cancel (mo->task);
      mo->task = NULL;
    }
    if (NULL != mo->sa_wait_warning)
    {
      GNUNET_SCHEDULER_cancel (mo->sa_wait_warning);
      mo->sa_wait_warning = NULL;
    }
    GNUNET_free (mo->sub_system);
    GNUNET_free (mo->key);
    GNUNET_free (mo);
    // Restart from monitors_head
    mo = monitors_head;
  }
  while (NULL != (iter = pc->op_head))
  {
    GNUNET_CONTAINER_DLL_remove (pc->op_head,
                                 pc->op_tail,
                                 iter);
    destroy_iteration (iter);
  }
  GNUNET_free (pc);
}
 
 
static void
store_hello_continuation (void *cls, int success)
{
  (void) cls;
 
  if (GNUNET_OK != success)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Error storing bootstrap hello!\n");
    GNUNET_break (0);
  }
}
 
 
static int
hosts_directory_scan_callback (void *cls, const char *fullname)
{
  ssize_t size_total;
  char buffer[GNUNET_MAX_MESSAGE_SIZE - 1] GNUNET_ALIGN;
  const struct GNUNET_MessageHeader *hello;
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_HELLO_Parser *parser;
  const struct GNUNET_PeerIdentity *pid;
  struct GNUNET_TIME_Absolute et;
  (void) cls;
 
  if (GNUNET_YES != GNUNET_DISK_file_test (fullname))
    return GNUNET_OK; /* ignore non-files */
 
  size_total = GNUNET_DISK_fn_read (fullname, buffer, sizeof(buffer) - 1);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Read %d bytes from `%s'\n",
              (int) size_total,
              fullname);
  if ((size_total < 0) ||
      (((size_t) size_total) < sizeof(struct GNUNET_MessageHeader)))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                _ ("Failed to parse HELLO in file `%s': %s\n"),
                fullname,
                "File has invalid size");
    return GNUNET_OK;
  }
  buffer[size_total] = '\0';
  parser = GNUNET_HELLO_parser_from_url (buffer);
  if (NULL == parser)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Unable to parse HELLO url\n");
    return GNUNET_OK;
  }
  env = GNUNET_HELLO_parser_to_env (parser);
  hello = GNUNET_MQ_env_get_msg (env);
  pid = GNUNET_HELLO_parser_get_id (parser);
  et = GNUNET_HELLO_get_expiration_time_from_msg (hello);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "store contrib hello for peer %s\n",
              GNUNET_i2s (pid));
 
  if (GNUNET_OK != db->store_record (db->cls,
                                     "peerstore",
                                     pid,
                                     GNUNET_PEERSTORE_HELLO_KEY,
                                     hello,
                                     ntohs (hello->size),
                                     et,
                                     GNUNET_PEERSTORE_STOREOPTION_MULTIPLE,
                                     &store_hello_continuation,
                                     NULL))
  {
    GNUNET_break (0);
  }
  GNUNET_MQ_discard (env);
  GNUNET_HELLO_parser_free (parser);
  return GNUNET_OK;
}
 
 
static void
run (void *cls,
     const struct GNUNET_CONFIGURATION_Handle *c,
     struct GNUNET_SERVICE_Handle *service)
{
  char *database;
  int use_included;
  char *ip;
  char *peerdir;
 
  cfg = c;
 
  if (GNUNET_OK !=
      GNUNET_CONFIGURATION_get_value_string (cfg,
                                             "peerstore",
                                             "DATABASE",
                                             &database))
  {
    GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR,
                               "peerstore",
                               "DATABASE");
    GNUNET_SCHEDULER_shutdown ();
    return;
  }
  GNUNET_asprintf (&db_lib_name,
                   "libgnunet_plugin_peerstore_%s",
                   database);
  db = GNUNET_PLUGIN_load (GNUNET_OS_project_data_gnunet (),
                           db_lib_name,
                           (void *) cfg);
  GNUNET_free (database);
  if (NULL == db)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                _ ("Could not load database backend `%s'\n"),
                db_lib_name);
    GNUNET_SCHEDULER_shutdown ();
    return;
  }
  expire_task = GNUNET_SCHEDULER_add_now (&cleanup_expired_records,
                                          NULL);
  use_included = GNUNET_CONFIGURATION_get_value_yesno (cfg,
                                                       "peerstore",
                                                       "USE_INCLUDED_HELLOS");
  if (GNUNET_SYSERR == use_included)
    use_included = GNUNET_NO;
  if (GNUNET_YES == use_included)
  {
    ip = GNUNET_OS_installation_get_path (GNUNET_OS_project_data_gnunet (),
                                          GNUNET_OS_IPK_DATADIR);
    GNUNET_asprintf (&peerdir, "%shellos", ip);
    GNUNET_free (ip);
 
    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                _ ("Importing HELLOs from `%s'\n"),
                peerdir);
    GNUNET_DISK_directory_scan (peerdir,
                                &hosts_directory_scan_callback,
                                NULL);
    GNUNET_free (peerdir);
  }
  else
  {
    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                _ ("Skipping import of included HELLOs\n"));
  }
  monitor_nc = GNUNET_notification_context_create (1);
  GNUNET_SCHEDULER_add_shutdown (&shutdown_task,
                                 NULL);
}
 
 
GNUNET_SERVICE_MAIN (
  GNUNET_OS_project_data_gnunet (),
  "peerstore", GNUNET_SERVICE_OPTION_SOFT_SHUTDOWN, &run, &client_connect_cb,
  &client_disconnect_cb, NULL,
  GNUNET_MQ_hd_var_size (store, GNUNET_MESSAGE_TYPE_PEERSTORE_STORE,
                         struct PeerstoreRecordMessage, NULL),
  GNUNET_MQ_hd_var_size (iterate_start,
                         GNUNET_MESSAGE_TYPE_PEERSTORE_ITERATE_START,
                         struct PeerstoreIterationStartMessage, NULL),
  GNUNET_MQ_hd_fixed_size (iterate_stop,
                           GNUNET_MESSAGE_TYPE_PEERSTORE_ITERATE_STOP,
                           struct PeerstoreIterationStopMessage, NULL),
  GNUNET_MQ_hd_fixed_size (iterate_next,
                           GNUNET_MESSAGE_TYPE_PEERSTORE_ITERATE_NEXT,
                           struct PeerstoreIterationNextMessage,
                           NULL),
  GNUNET_MQ_hd_var_size (monitor_start,
                         GNUNET_MESSAGE_TYPE_PEERSTORE_MONITOR_START,
                         struct PeerstoreMonitorStartMessage,
                         NULL),
  GNUNET_MQ_hd_fixed_size (monitor_next,
                           GNUNET_MESSAGE_TYPE_PEERSTORE_MONITOR_NEXT,
                           struct PeerstoreMonitorNextMessage,
                           NULL),
  GNUNET_MQ_handler_end ());
 
 
/* end of gnunet-service-peerstore.c */