gnunet-service-zonemaster.c File Reference

publish records from namestore to GNUnet name system More...

#include "platform.h"
#include <pthread.h>
#include "gnunet_util_lib.h"
#include "gnunet_dht_service.h"
#include "gnunet_namestore_service.h"
#include "gnunet_namecache_service.h"
#include "gnunet_statistics_service.h"

Include dependency graph for gnunet-service-zonemaster.c:

Go to the source code of this file.

Data Structures
struct	RecordPublicationJob
struct	CacheOperation
	Pending operation on the namecache. More...

Macros
#define	LOG_STRERROR_FILE(kind, syscall, filename)
#define	PUBLISH_OPS_PER_EXPIRATION   4
	How often should we (re)publish each record before it expires?
#define	DELTA_INTERVAL   100
	How often do we measure the delta between desired zone iteration speed and actual speed, and tell statistics service about it?
#define	NS_BLOCK_SIZE   1000
	How many records do we fetch in one shot from the namestore?
#define	JOB_QUEUE_LIMIT   5000
	How many open jobs (and with it maximum amount of pending DHT operations) do we allow at most.
#define	NAMESTORE_MONITOR_QUEUE_LIMIT   5
	How many events may the namestore give us before it has to wait for us to keep up?
#define	INITIAL_ZONE_ITERATION_INTERVAL   GNUNET_TIME_UNIT_MILLISECONDS
	The initial interval in milliseconds btween puts in a zone iteration.
#define	MAXIMUM_ZONE_ITERATION_INTERVAL
	The upper bound for the zone iteration interval (per record).
#define	LATE_ITERATION_SPEEDUP_FACTOR   2
	The factor the current zone iteration interval is divided by for each additional new record.
#define	DHT_GNS_REPLICATION_LEVEL   5
	What replication level do we use for DHT PUT operations?

Functions
static void	free_job (struct RecordPublicationJob *job)
static void	shutdown_task (void *cls)
	Task run during shutdown.
static void	finish_cache_operation (void *cls, int32_t success, const char *emsg)
	Cache operation complete, clean up.
static void	refresh_block (const struct GNUNET_GNSRECORD_Block *block)
	Refresh the (encrypted) block in the namecache.
static void	publish_zone_namestore_next (void *cls)
	Method called periodically that triggers iteration over authoritative records.
static void	publish_zone_dht_start (void *cls)
	Periodically iterate over our zone and store everything in dht.
static void	calculate_put_interval ()
	Calculate target_iteration_velocity_per_record.
static void	note_relative_expiration (const struct GNUNET_GNSRECORD_Data *rd, unsigned int rd_count)
	Note the relative expiration times in rd, they determine how often the zone has to be republished.
static void	update_velocity (unsigned int cnt)
	Re-calculate our velocity and the desired velocity.
static void	check_zone_namestore_next ()
	Check if the current zone iteration needs to be continued by calling publish_zone_namestore_next(), and if so with what delay.
static void	cleanup_job (struct RecordPublicationJob *job)
static void	dht_put_continuation (void *cls)
	Continuation called from DHT once the PUT operation is done.
static void	dispatch_job (const struct GNUNET_CRYPTO_BlindablePrivateKey *key, const char *label, const struct GNUNET_GNSRECORD_Data *rd, unsigned int rd_count, const struct GNUNET_TIME_Absolute expire)
	Store GNS records in the DHT.
static void	notification_pipe_cb (void *cls)
static void	initiate_put_from_pipe_trigger (void *cls)
static void	zone_iteration_error (void *cls)
	We encountered an error in our zone iteration.
static void	zone_iteration_finished (void *cls)
	Zone iteration is completed.
static void	handle_record (void *cls, const struct GNUNET_CRYPTO_BlindablePrivateKey *key, const char *label, unsigned int rd_count, const struct GNUNET_GNSRECORD_Data *rd, struct GNUNET_TIME_Absolute expire)
	Function used to put all records successively into the DHT.
static void	dispatch_job_monitor (const struct GNUNET_CRYPTO_BlindablePrivateKey *key, const char *label, const struct GNUNET_GNSRECORD_Data *rd, unsigned int rd_count, struct GNUNET_TIME_Absolute expire)
	Store GNS records in the DHT.
static void	handle_monitor_event (void *cls, const struct GNUNET_CRYPTO_BlindablePrivateKey *zone, const char *label, unsigned int rd_count, const struct GNUNET_GNSRECORD_Data *rd, struct GNUNET_TIME_Absolute expire)
	Process a record that was stored in the namestore (invoked by the monitor).
static void	handle_monitor_error (void *cls)
	The zone monitor encountered an IPC error trying to to get in sync.
static void *	sign_worker (void *cls)
static void	run (void *cls, const struct GNUNET_CONFIGURATION_Handle *c, struct GNUNET_SERVICE_Handle *service)
	Perform zonemaster duties: watch namestore, publish records.
	GNUNET_SERVICE_MAIN (GNUNET_OS_project_data_gnunet(), "zonemaster", GNUNET_SERVICE_OPTION_NONE, &run, NULL, NULL, NULL, GNUNET_MQ_handler_end())
	Define "main" method using service macro.

Variables
static pthread_t *	worker
	Our workers.
static pthread_mutex_t	sign_jobs_lock
	Lock for the sign jobs queue.
static pthread_mutex_t	sign_results_lock
	Lock for the DHT put jobs queue.
static pthread_cond_t	sign_jobs_cond
	Wait condition on new sign jobs.
static int	in_shutdown = GNUNET_NO
	For threads to know we are shutting down.
static int	monitor_halted = GNUNET_NO
	Monitor halted?
static struct GNUNET_DISK_PipeHandle *	notification_pipe
	Our notification pipe.
static struct GNUNET_SCHEDULER_Task *	pipe_read_task
	Pipe read task.
static struct RecordPublicationJob *	sign_jobs_head
	The DLL for workers to retrieve open jobs that require signing of blocks.
static struct RecordPublicationJob *	sign_jobs_tail
	See above.
static struct RecordPublicationJob *	sign_results_head
	The DLL for workers to place jobs that are signed.
static struct RecordPublicationJob *	sign_results_tail
	See above.
static struct RecordPublicationJob *	dht_jobs_head
	The DLL for jobs currently in the process of being dispatched into the DHT.
static struct RecordPublicationJob *	dht_jobs_tail
	See above.
static struct GNUNET_STATISTICS_Handle *	statistics
	Handle to the statistics service.
static struct GNUNET_DHT_Handle *	dht_handle
	Our handle to the DHT.
static struct GNUNET_NAMESTORE_Handle *	namestore_handle
	Our handle to the namestore service.
static struct GNUNET_NAMESTORE_ZoneMonitor *	zmon
	Handle to monitor namestore changes to instant propagation.
static struct GNUNET_NAMECACHE_Handle *	namecache
	Our handle to the namecache service.
static int	disable_namecache
	Use the namecache? Doing so creates additional cryptographic operations whenever we touch a record.
static struct GNUNET_NAMESTORE_ZoneIterator *	namestore_iter
	Handle to iterate over our authoritative zone in namestore.
static unsigned int	job_queue_length
	Number of entries in the job queue #jobs_head.
static unsigned long long	num_public_records
	Useful for zone update for DHT put.
static unsigned long long	last_num_public_records
	Last seen record count.
static unsigned long long	put_cnt
	Number of successful put operations performed in the current measurement cycle (as measured in check_zone_namestore_next()).
static struct GNUNET_TIME_Relative	target_iteration_velocity_per_record
	What is the frequency at which we currently would like to perform DHT puts (per record)? Calculated in update_velocity() from the zone_publish_time_window() and the total number of record sets we have (so far) observed in the zone.
static struct GNUNET_TIME_Relative	min_relative_record_time
	Minimum relative expiration time of records seem during the current zone iteration.
static struct GNUNET_TIME_Relative	last_min_relative_record_time
	Minimum relative expiration time of records seem during the last zone iteration.
static struct GNUNET_TIME_Relative	zone_publish_time_window_default
	Default time window for zone iteration.
static struct GNUNET_TIME_Relative	zone_publish_time_window
	Time window for zone iteration, adjusted based on relative record expiration times in our zone.
static struct GNUNET_TIME_Absolute	last_put_100
	When did we last start measuring the DELTA_INTERVAL successful DHT puts? Used for velocity calculations.
static struct GNUNET_TIME_Relative	sub_delta
	By how much should we try to increase our per-record iteration speed (over the desired speed calculated directly from the #put_interval)? Basically this value corresponds to the per-record CPU time overhead we have.
static struct GNUNET_SCHEDULER_Task *	zone_publish_task
	zone publish task
static unsigned int	ns_iteration_left
	How many more values are left for the current query before we need to explicitly ask the namestore for more?
static int	first_zone_iteration
	GNUNET_YES if zone has never been published before
static int	cache_keys
	Optimize block insertion by caching map of private keys to public keys in memory?
static struct CacheOperation *	cop_head
	Head of cop DLL.
static struct CacheOperation *	cop_tail
	Tail of cop DLL.

Detailed Description

publish records from namestore to GNUnet name system

Author

Christian Grothoff

Definition in file gnunet-service-zonemaster.c.

Macro Definition Documentation

LOG_STRERROR_FILE

#define LOG_STRERROR_FILE	(		kind,
			syscall,
			filename
	)

Value:

                                                                   GNUNET_log_from_strerror_file (kind, "util", \
                                                                   syscall, \
                                                                   filename)

Definition at line 34 of file gnunet-service-zonemaster.c.

{
 
  struct RecordPublicationJob *next;
 
  struct RecordPublicationJob *prev;
 
  struct GNUNET_CRYPTO_BlindablePrivateKey zone;
 
  struct GNUNET_GNSRECORD_Block *block_pub;
 
  struct GNUNET_GNSRECORD_Block *block_priv;
 
  unsigned char *rdata_pub;
 
  unsigned char *rdata_priv;
 
  size_t rdata_pub_len;
 
  size_t rdata_priv_len;
 
  struct GNUNET_TIME_Absolute expire_pub;
 
  struct GNUNET_TIME_Absolute expire_priv;
 
  char *label;
 
  struct GNUNET_DHT_PutHandle *ph;
 
  struct GNUNET_TIME_Absolute start_date;
 
  struct GNUNET_GNSRECORD_EncryptionContext *ec;
 
  int have_public_records;
};
 
 
static struct RecordPublicationJob *sign_jobs_head;
 
static struct RecordPublicationJob *sign_jobs_tail;
 
static struct RecordPublicationJob *sign_results_head;
 
static struct RecordPublicationJob *sign_results_tail;
 
 
static struct RecordPublicationJob *dht_jobs_head;
 
static struct RecordPublicationJob *dht_jobs_tail;
 
 
struct CacheOperation
{
  struct CacheOperation *prev;
 
  struct CacheOperation *next;
 
  struct GNUNET_NAMECACHE_QueueEntry *qe;
 
};
 
 
static struct GNUNET_STATISTICS_Handle *statistics;
 
static struct GNUNET_DHT_Handle *dht_handle;
 
static struct GNUNET_NAMESTORE_Handle *namestore_handle;
 
static struct GNUNET_NAMESTORE_ZoneMonitor *zmon;
 
static struct GNUNET_NAMECACHE_Handle *namecache;
 
static int disable_namecache;
 
static struct GNUNET_NAMESTORE_ZoneIterator *namestore_iter;
 
static unsigned int job_queue_length;
 
static unsigned long long num_public_records;
 
static unsigned long long last_num_public_records;
 
static unsigned long long put_cnt;
 
static struct GNUNET_TIME_Relative target_iteration_velocity_per_record;
 
static struct GNUNET_TIME_Relative min_relative_record_time;
 
static struct GNUNET_TIME_Relative last_min_relative_record_time;
 
static struct GNUNET_TIME_Relative zone_publish_time_window_default;
 
static struct GNUNET_TIME_Relative zone_publish_time_window;
 
static struct GNUNET_TIME_Absolute last_put_100;
 
static struct GNUNET_TIME_Relative sub_delta;
 
static struct GNUNET_SCHEDULER_Task *zone_publish_task;
 
static unsigned int ns_iteration_left;
 
static int first_zone_iteration;
 
static int cache_keys;
 
static struct CacheOperation *cop_head;
 
static struct CacheOperation *cop_tail;
 
 
static void
free_job (struct RecordPublicationJob *job)
{
  if (job->rdata_pub != job->rdata_priv)
    GNUNET_free (job->rdata_priv);
  GNUNET_free (job->rdata_pub);
  if (job->block_pub != job->block_priv)
    GNUNET_free (job->block_priv);
  GNUNET_free (job->block_pub);
  if (NULL != job->label)
    GNUNET_free (job->label);
  if (NULL != job->ec)
  {
    GNUNET_GNSRECORD_encryption_context_destroy (job->ec);
    job->ec = NULL;
  }
  GNUNET_free (job);
}
 
 
static void
shutdown_task (void *cls)
{
  struct CacheOperation *cop;
  struct RecordPublicationJob *job;
 
  (void) cls;
  in_shutdown = GNUNET_YES;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Shutting down!\n");
  if (NULL != notification_pipe)
    GNUNET_DISK_pipe_close (notification_pipe);
  if (NULL != pipe_read_task)
    GNUNET_SCHEDULER_cancel (pipe_read_task);
  while (NULL != (cop = cop_head))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "Aborting incomplete namecache operation\n");
    GNUNET_NAMECACHE_cancel (cop->qe);
    GNUNET_CONTAINER_DLL_remove (cop_head, cop_tail, cop);
    GNUNET_free (cop);
  }
  GNUNET_assert (0 == pthread_mutex_lock (&sign_jobs_lock));
  while (NULL != (job = sign_jobs_head))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "Removing incomplete jobs\n");
    GNUNET_CONTAINER_DLL_remove (sign_jobs_head, sign_jobs_tail, job);
    job_queue_length--;
    free_job (job);
  }
  GNUNET_assert (0 == pthread_mutex_unlock (&sign_jobs_lock));
  GNUNET_assert (0 == pthread_mutex_lock (&sign_results_lock));
  while (NULL != (job = sign_results_head))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "Removing incomplete jobs\n");
    GNUNET_CONTAINER_DLL_remove (sign_results_head, sign_results_tail, job);
    free_job (job);
  }
  GNUNET_assert (0 == pthread_mutex_unlock (&sign_results_lock));
  while (NULL != (job = dht_jobs_head))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "Removing incomplete jobs\n");
    GNUNET_CONTAINER_DLL_remove (dht_jobs_head, dht_jobs_tail, job);
    if (NULL != job->ph)
      GNUNET_DHT_put_cancel (job->ph);
    free_job (job);
  }
  if (NULL != statistics)
  {
    GNUNET_STATISTICS_destroy (statistics,
                               GNUNET_NO);
    statistics = NULL;
  }
  if (NULL != zone_publish_task)
  {
    GNUNET_SCHEDULER_cancel (zone_publish_task);
    zone_publish_task = NULL;
  }
  if (NULL != namestore_iter)
  {
    GNUNET_NAMESTORE_zone_iteration_stop (namestore_iter);
    namestore_iter = NULL;
  }
  if (NULL != zmon)
  {
    GNUNET_NAMESTORE_zone_monitor_stop (zmon);
    zmon = NULL;
  }
  if (NULL != namestore_handle)
  {
    GNUNET_NAMESTORE_disconnect (namestore_handle);
    namestore_handle = NULL;
  }
  if (NULL != namecache)
  {
    GNUNET_NAMECACHE_disconnect (namecache);
    namecache = NULL;
  }
  if (NULL != dht_handle)
  {
    GNUNET_DHT_disconnect (dht_handle);
    dht_handle = NULL;
  }
}
 
 
static void
finish_cache_operation (void *cls, int32_t success, const char *emsg)
{
  struct CacheOperation *cop = cls;
 
  if (NULL != emsg)
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                _ ("Failed to replicate block in namecache: %s\n"),
                emsg);
  else
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "CACHE operation completed\n");
  GNUNET_CONTAINER_DLL_remove (cop_head, cop_tail, cop);
  GNUNET_free (cop);
}
 
 
static void
refresh_block (const struct GNUNET_GNSRECORD_Block *block)
{
  struct CacheOperation *cop;
 
  if (GNUNET_YES == disable_namecache)
  {
    GNUNET_STATISTICS_update (statistics,
                              "Namecache updates skipped (NC disabled)",
                              1,
                              GNUNET_NO);
    return;
  }
  GNUNET_assert (NULL != block);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Caching block in namecache\n");
  GNUNET_STATISTICS_update (statistics,
                            "Namecache updates pushed",
                            1,
                            GNUNET_NO);
  cop = GNUNET_new (struct CacheOperation);
  GNUNET_CONTAINER_DLL_insert (cop_head, cop_tail, cop);
  cop->qe = GNUNET_NAMECACHE_block_cache (namecache,
                                          block,
                                          &finish_cache_operation,
                                          cop);
}
 
 
static void
publish_zone_namestore_next (void *cls)
{
  (void) cls;
  zone_publish_task = NULL;
  GNUNET_assert (NULL != namestore_iter);
  GNUNET_assert (0 == ns_iteration_left);
  ns_iteration_left = NS_BLOCK_SIZE;
  GNUNET_NAMESTORE_zone_iterator_next (namestore_iter,
                                       NS_BLOCK_SIZE);
}
 
 
static void
publish_zone_dht_start (void *cls);
 
 
static void
calculate_put_interval ()
{
  /* The window follows from the record lifetimes alone: a record set has
     to be republished #PUBLISH_OPS_PER_EXPIRATION times before it
     expires, whether or not we have counted the zone yet. */
  last_min_relative_record_time
    = GNUNET_TIME_relative_min (last_min_relative_record_time,
                                min_relative_record_time);
  zone_publish_time_window
    = GNUNET_TIME_relative_min (GNUNET_TIME_relative_divide (
                                  last_min_relative_record_time,
                                  PUBLISH_OPS_PER_EXPIRATION),
                                zone_publish_time_window_default);
  if (0 == num_public_records)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
                "No records in namestore database.\n");
  }
  else
  {
    target_iteration_velocity_per_record
      = GNUNET_TIME_relative_divide (zone_publish_time_window,
                                     last_num_public_records);
  }
  target_iteration_velocity_per_record
    = GNUNET_TIME_relative_min (target_iteration_velocity_per_record,
                                MAXIMUM_ZONE_ITERATION_INTERVAL);
  GNUNET_STATISTICS_set (statistics,
                         "Minimum relative record expiration (in μs)",
                         last_min_relative_record_time.rel_value_us,
                         GNUNET_NO);
  GNUNET_STATISTICS_set (statistics,
                         "Zone publication time window (in μs)",
                         zone_publish_time_window.rel_value_us,
                         GNUNET_NO);
  GNUNET_STATISTICS_set (statistics,
                         "Target zone iteration velocity (μs)",
                         target_iteration_velocity_per_record.rel_value_us,
                         GNUNET_NO);
}
 
 
static void
note_relative_expiration (const struct GNUNET_GNSRECORD_Data *rd,
                          unsigned int rd_count)
{
  for (unsigned int i = 0; i < rd_count; i++)
  {
    if (0 == (rd[i].flags & GNUNET_GNSRECORD_RF_RELATIVE_EXPIRATION))
      continue;
    /* GNUNET_GNSRECORD_block_create will convert to absolute time;
       we just need to adjust our iteration frequency */
    min_relative_record_time.rel_value_us
      = GNUNET_MIN (rd[i].expiration_time,
                    min_relative_record_time.rel_value_us);
  }
}
 
 
static void
update_velocity (unsigned int cnt)
{
  struct GNUNET_TIME_Relative delta;
  unsigned long long pct = 0;
 
  if (0 == cnt)
    return;
  /* How fast were we really? */
  delta = GNUNET_TIME_absolute_get_duration (last_put_100);
  delta.rel_value_us /= cnt;
  last_put_100 = GNUNET_TIME_absolute_get ();
 
  /* calculate expected frequency */
  if ((num_public_records > last_num_public_records) &&
      (GNUNET_NO == first_zone_iteration))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Last record count was lower than current record count.  Reducing interval.\n");
    last_num_public_records = num_public_records
                              * LATE_ITERATION_SPEEDUP_FACTOR;
    calculate_put_interval ();
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Desired global zone iteration interval is %s/record!\n",
              GNUNET_STRINGS_relative_time_to_string (
                target_iteration_velocity_per_record,
                GNUNET_YES));
 
  /* Tell statistics actual vs. desired speed */
  GNUNET_STATISTICS_set (statistics,
                         "Current zone iteration velocity (μs/record)",
                         delta.rel_value_us,
                         GNUNET_NO);
  /* update "sub_delta" based on difference, taking
     previous sub_delta into account! */
  if (target_iteration_velocity_per_record.rel_value_us > delta.rel_value_us)
  {
    /* We were too fast, reduce sub_delta! */
    struct GNUNET_TIME_Relative corr;
 
    corr = GNUNET_TIME_relative_subtract (target_iteration_velocity_per_record,
                                          delta);
    if (sub_delta.rel_value_us > delta.rel_value_us)
    {
      /* Reduce sub_delta by corr */
      sub_delta = GNUNET_TIME_relative_subtract (sub_delta,
                                                 corr);
    }
    else
    {
      /* We're doing fine with waiting the full time, this
         should theoretically only happen if we run at
         infinite speed. */
      sub_delta = GNUNET_TIME_UNIT_ZERO;
    }
  }
  else if (target_iteration_velocity_per_record.rel_value_us <
           delta.rel_value_us)
  {
    /* We were too slow, increase sub_delta! */
    struct GNUNET_TIME_Relative corr;
 
    corr = GNUNET_TIME_relative_subtract (delta,
                                          target_iteration_velocity_per_record);
    sub_delta = GNUNET_TIME_relative_add (sub_delta,
                                          corr);
    if (sub_delta.rel_value_us >
        target_iteration_velocity_per_record.rel_value_us)
    {
      /* CPU overload detected, we cannot go at desired speed,
         as this would mean using a negative delay. */
      /* compute how much faster we would want to be for
         the desired velocity */
      if (0 == target_iteration_velocity_per_record.rel_value_us)
        pct = UINT64_MAX;     /* desired speed is infinity ... */
      else
        pct = (sub_delta.rel_value_us
               - target_iteration_velocity_per_record.rel_value_us) * 100LLU
              / target_iteration_velocity_per_record.rel_value_us;
      sub_delta = target_iteration_velocity_per_record;
    }
  }
  GNUNET_STATISTICS_set (statistics,
                         "# dispatched jobs",
                         job_queue_length,
                         GNUNET_NO);
  GNUNET_STATISTICS_set (statistics,
                         "% speed increase needed for target velocity",
                         pct,
                         GNUNET_NO);
  GNUNET_STATISTICS_set (statistics,
                         "# records processed in current iteration",
                         num_public_records,
                         GNUNET_NO);
}
 
 
static void
check_zone_namestore_next ()
{
  struct GNUNET_TIME_Relative delay;
 
  if (0 != ns_iteration_left)
    return; /* current NAMESTORE iteration not yet done */
  if (job_queue_length >= JOB_QUEUE_LIMIT)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "Job queue length exceeded (%u/%u). Pausing namestore iteration.\n",
                job_queue_length,
                JOB_QUEUE_LIMIT);
    return;
  }
  update_velocity (put_cnt);
  put_cnt = 0;
  delay = GNUNET_TIME_relative_subtract (target_iteration_velocity_per_record,
                                         sub_delta);
  /* We delay *once* per #NS_BLOCK_SIZE, so we need to multiply the
     per-record delay calculated so far with the #NS_BLOCK_SIZE */
  GNUNET_STATISTICS_set (statistics,
                         "Current artificial NAMESTORE delay (μs/record)",
                         delay.rel_value_us,
                         GNUNET_NO);
  delay = GNUNET_TIME_relative_multiply (delay,
                                         GNUNET_MIN (NS_BLOCK_SIZE,
                                                     GNUNET_MAX (
                                                       1,
                                                       last_num_public_records))
                                         );
  /* make sure we do not overshoot because of the #NS_BLOCK_SIZE factor.
     The pause is part of one pass over the zone, so it may never exceed
     the window that pass has to fit into: a record set has to be
     republished #PUBLISH_OPS_PER_EXPIRATION times before it expires, and
     with a short expiration time the whole window is shorter than
     #MAXIMUM_ZONE_ITERATION_INTERVAL. */
  delay = GNUNET_TIME_relative_min (delay,
                                    GNUNET_TIME_relative_min (
                                      MAXIMUM_ZONE_ITERATION_INTERVAL,
                                      zone_publish_time_window));
  /* no delays on first iteration */
  if (GNUNET_YES == first_zone_iteration)
    delay = GNUNET_TIME_UNIT_ZERO;
  GNUNET_assert (NULL == zone_publish_task);
  zone_publish_task = GNUNET_SCHEDULER_add_delayed (delay,
                                                    &publish_zone_namestore_next
                                                    ,
                                                    NULL);
}
 
 
static void
cleanup_job (struct RecordPublicationJob*job)
{
  GNUNET_assert (job_queue_length > 0);
  job_queue_length--;
  if (NULL == zone_publish_task)
    check_zone_namestore_next ();
  if (job_queue_length <= JOB_QUEUE_LIMIT)
  {
    if (GNUNET_YES == monitor_halted)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                  "Job queue emptied (%u/%u). Resuming monitor.\n",
                  job_queue_length,
                  JOB_QUEUE_LIMIT);
      GNUNET_NAMESTORE_zone_monitor_next (zmon, 1);
      monitor_halted = GNUNET_NO;
    }
  }
  free_job (job);
}
 
 
static void
dht_put_continuation (void *cls)
{
  struct RecordPublicationJob *job = cls;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "PUT complete; Pending jobs: %u\n", job_queue_length - 1);
  /* When we just fall under the limit, trigger monitor/iterator again
   * if halted. We can only safely trigger one, prefer iterator. */
  GNUNET_CONTAINER_DLL_remove (dht_jobs_head,
                               dht_jobs_tail,
                               job);
  cleanup_job (job);
}
 
 
static void
dispatch_job (const struct GNUNET_CRYPTO_BlindablePrivateKey *key,
              const char *label,
              const struct GNUNET_GNSRECORD_Data *rd,
              unsigned int rd_count,
              const struct GNUNET_TIME_Absolute expire)
{
  struct GNUNET_GNSRECORD_Data rd_public[rd_count];
  struct GNUNET_TIME_Absolute expire_pub;
  struct RecordPublicationJob *job;
  unsigned char *rdata_public;
  unsigned char *rdata_priv;
  size_t rdata_public_len;
  size_t rdata_private_len;
  unsigned int rd_public_count = 0;
  char *emsg;
 
  if (GNUNET_OK !=
      GNUNET_GNSRECORD_normalize_record_set (label,
                                             rd,
                                             rd_count,
                                             rd_public,
                                             &rd_public_count,
                                             &expire_pub,
                                             GNUNET_GNSRECORD_FILTER_OMIT_PRIVATE,
                                             &emsg))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "%s\n", emsg);
    GNUNET_free (emsg);
  }
 
  rdata_public_len = GNUNET_GNSRECORD_records_get_size (rd_public_count,
                                                        rd_public);
 
  rdata_private_len = GNUNET_GNSRECORD_records_get_size (rd_count,
                                                         rd);
 
  rdata_public = GNUNET_malloc (rdata_public_len);
  GNUNET_assert (GNUNET_OK == GNUNET_GNSRECORD_record_data_to_rdata (
                   rd_public_count,
                   rd_public,
                   rdata_public,
                   rdata_public_len));
  if (rd_count != rd_public_count)
  {
    rdata_priv = GNUNET_malloc (rdata_private_len);
    GNUNET_assert (GNUNET_OK == GNUNET_GNSRECORD_record_data_to_rdata (
                     rd_count,
                     rd,
                     rdata_priv,
                     rdata_private_len));
  }
  else
  {
    rdata_priv = rdata_public;
  }
  GNUNET_assert (0 == pthread_mutex_lock (&sign_jobs_lock));
  job = GNUNET_new (struct RecordPublicationJob);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Public record count: %d\n",
              rd_public_count);
  job->have_public_records = (rd_public_count > 0);
  job->rdata_pub = rdata_public;
  job->rdata_pub_len = rdata_public_len;
  job->rdata_priv = rdata_priv;
  job->rdata_priv_len = rdata_private_len;
  job->zone = *key;
  job->label = GNUNET_strdup (label);
  job->expire_pub = expire_pub;
  job->expire_priv = expire;
  job->ec = GNUNET_GNSRECORD_encryption_context_setup_owner (key);
  GNUNET_CONTAINER_DLL_insert (sign_jobs_head, sign_jobs_tail, job);
  GNUNET_assert (0 == pthread_cond_signal (&sign_jobs_cond));
  GNUNET_assert (0 == pthread_mutex_unlock (&sign_jobs_lock));
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Creating job with %u record(s) for label `%s', expiration `%s'\n",
              rd_public_count,
              label,
              GNUNET_STRINGS_absolute_time_to_string (expire));
  num_public_records++;
  return;
}
 
 
static void
notification_pipe_cb (void *cls);
 
static void
initiate_put_from_pipe_trigger (void *cls)
{
  struct GNUNET_HashCode query;
  struct RecordPublicationJob *job;
  const struct GNUNET_DISK_FileHandle *np_fh;
  char buf[100];
  ssize_t nf_count;
 
  pipe_read_task = NULL;
  np_fh = GNUNET_DISK_pipe_handle (notification_pipe,
                                   GNUNET_DISK_PIPE_END_READ);
  pipe_read_task =
    GNUNET_SCHEDULER_add_read_file (
      GNUNET_TIME_UNIT_FOREVER_REL,
      np_fh,
      notification_pipe_cb,
      NULL);
  /* empty queue */
  nf_count = GNUNET_DISK_file_read (np_fh, buf, sizeof (buf));
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Read %lld notifications from pipe\n",
              (long long) nf_count);
  while (true)
  {
    GNUNET_assert (0 == pthread_mutex_lock (&sign_results_lock));
    if (NULL == sign_results_head)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "No more results. Back to sleep.\n");
      GNUNET_assert (0 == pthread_mutex_unlock (&sign_results_lock));
      return;
    }
    job = sign_results_head;
    GNUNET_CONTAINER_DLL_remove (sign_results_head, sign_results_tail, job);
    GNUNET_assert (0 == pthread_mutex_unlock (&sign_results_lock));
    GNUNET_GNSRECORD_query_from_private_key (&job->zone,
                                             job->label,
                                             &query);
    // It is possible that the public block size is 0 (no public blocks)
    // Do not bother with the DHT in that case
    if (job->have_public_records)
    {
      size_t block_size = GNUNET_GNSRECORD_block_get_size (job->block_pub);
      job->ph = GNUNET_DHT_put (dht_handle,
                                &query,
                                DHT_GNS_REPLICATION_LEVEL,
                                GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,
                                GNUNET_BLOCK_TYPE_GNS_NAMERECORD,
                                block_size,
                                job->block_pub,
                                job->expire_pub,
                                &dht_put_continuation,
                                job);
      if (NULL == job->ph)
      {
        /* Drop this job, but keep draining the queue: returning here
           leaves every block the workers already signed stranded, and
           losing the job_queue_length slot eventually halts both the
           zone iteration and the monitor for good. */
        GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                    "Could not perform DHT PUT, is the DHT running?\n");
        cleanup_job (job);
        continue;
      }
      GNUNET_STATISTICS_update (statistics,
                                "DHT put operations initiated",
                                1,
                                GNUNET_NO);
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Storing record(s) for label `%s' in DHT under key %s\n",
                  job->label,
                  GNUNET_h2s (&query));
      refresh_block (job->block_priv);
      GNUNET_CONTAINER_DLL_insert (dht_jobs_head, dht_jobs_tail, job);
    }
    else
    {
      // Private blocks may still be available and must be updated
      // in the cache
      refresh_block (job->block_priv);
      cleanup_job (job);
    }
  }
}
 
 
static void
zone_iteration_error (void *cls)
{
  (void) cls;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Got disconnected from namestore database, retrying.\n");
  namestore_iter = NULL;
  /* We end up here on error/disconnect/shutdown, so potentially
     while a zone publish task or a DHT put is still running; hence
     we need to cancel those. */
  if (NULL != zone_publish_task)
  {
    GNUNET_SCHEDULER_cancel (zone_publish_task);
    zone_publish_task = NULL;
  }
  zone_publish_task = GNUNET_SCHEDULER_add_now (&publish_zone_dht_start,
                                                NULL);
}
 
 
static void
zone_iteration_finished (void *cls)
{
  (void) cls;
  /* we're done with one iteration, calculate when to do the next one */
  namestore_iter = NULL;
  last_num_public_records = num_public_records;
  first_zone_iteration = GNUNET_NO;
  last_min_relative_record_time = min_relative_record_time;
  calculate_put_interval ();
  /* reset for next iteration */
  min_relative_record_time
    = GNUNET_TIME_UNIT_FOREVER_REL;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Zone iteration finished. Adjusted zone iteration interval to %s\n",
              GNUNET_STRINGS_relative_time_to_string (
                target_iteration_velocity_per_record,
                GNUNET_YES));
  GNUNET_STATISTICS_set (statistics,
                         "Target zone iteration velocity (μs)",
                         target_iteration_velocity_per_record.rel_value_us,
                         GNUNET_NO);
  GNUNET_STATISTICS_set (statistics,
                         "Number of public records in DHT",
                         last_num_public_records,
                         GNUNET_NO);
  GNUNET_assert (NULL == zone_publish_task);
  if (0 == last_num_public_records)
  {
    zone_publish_task = GNUNET_SCHEDULER_add_delayed (
      target_iteration_velocity_per_record,
      &publish_zone_dht_start,
      NULL);
  }
  else
  {
    zone_publish_task = GNUNET_SCHEDULER_add_now (&publish_zone_dht_start,
                                                  NULL);
  }
}
 
 
static void
handle_record (void *cls,
               const struct GNUNET_CRYPTO_BlindablePrivateKey *key,
               const char *label,
               unsigned int rd_count,
               const struct GNUNET_GNSRECORD_Data *rd,
               struct GNUNET_TIME_Absolute expire)
{
  (void) cls;
  ns_iteration_left--;
  if (0 == rd_count)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Record set empty, moving to next record set\n");
    check_zone_namestore_next ();
    return;
  }
  note_relative_expiration (rd,
                            rd_count);
 
  /* We got a set of records to publish */
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Starting DHT PUT\n");
  put_cnt++;
  if (0 == put_cnt % DELTA_INTERVAL)
    update_velocity (DELTA_INTERVAL);
  dispatch_job (key,
                label,
                rd,
                rd_count,
                expire);
  job_queue_length++;
  check_zone_namestore_next ();
}
 
 
static void
publish_zone_dht_start (void *cls)
{
  (void) cls;
  zone_publish_task = NULL;
  GNUNET_STATISTICS_update (statistics,
                            "Full zone iterations launched",
                            1,
                            GNUNET_NO);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Starting DHT zone update!\n");
  /* start counting again */
  num_public_records = 0;
  GNUNET_assert (NULL == namestore_iter);
  ns_iteration_left = 1;
  namestore_iter
    = GNUNET_NAMESTORE_zone_iteration_start2 (namestore_handle,
                                              NULL, /* All zones */
                                              &zone_iteration_error,
                                              NULL,
                                              &handle_record,
                                              NULL,
                                              &zone_iteration_finished,
                                              NULL,
                                              GNUNET_GNSRECORD_FILTER_NONE);
  GNUNET_assert (NULL != namestore_iter);
}
 
 
static void
dispatch_job_monitor (const struct GNUNET_CRYPTO_BlindablePrivateKey *key,
                      const char *label,
                      const struct GNUNET_GNSRECORD_Data *rd,
                      unsigned int rd_count,
                      struct GNUNET_TIME_Absolute expire)
{
  dispatch_job (key, label, rd, rd_count, expire);
}
 
 
static void
handle_monitor_event (void *cls,
                      const struct GNUNET_CRYPTO_BlindablePrivateKey *zone,
                      const char *label,
                      unsigned int rd_count,
                      const struct GNUNET_GNSRECORD_Data *rd,
                      struct GNUNET_TIME_Absolute expire)
{
  (void) cls;
  GNUNET_STATISTICS_update (statistics,
                            "Namestore monitor events received",
                            1,
                            GNUNET_NO);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received %u records for label `%s' via namestore monitor\n",
              rd_count,
              label);
  if (0 == rd_count)
  {
    GNUNET_NAMESTORE_zone_monitor_next (zmon,
                                        1);
    return;   /* nothing to do */
  }
  note_relative_expiration (rd,
                            rd_count);
  if ( (GNUNET_TIME_relative_cmp (min_relative_record_time,
                                  <,
                                  last_min_relative_record_time)) &&
       (NULL == namestore_iter) &&
       (NULL != zone_publish_task) )
  {
    /* This record set expires sooner than anything we saw while
       iterating, so the next iteration -- which is what republishes it --
       has to happen sooner than the one we scheduled back then.  Without
       this a record with a lifetime below
       #MAXIMUM_ZONE_ITERATION_INTERVAL is gone from the DHT long before
       it is put there again. */
    last_min_relative_record_time = min_relative_record_time;
    calculate_put_interval ();
    GNUNET_SCHEDULER_cancel (zone_publish_task);
    zone_publish_task
      = GNUNET_SCHEDULER_add_delayed (target_iteration_velocity_per_record,
                                      &publish_zone_dht_start,
                                      NULL);
  }
  dispatch_job_monitor (zone,
                        label,
                        rd,
                        rd_count,
                        expire);
  job_queue_length++;
  if (job_queue_length >= JOB_QUEUE_LIMIT)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "Job queue length exceeded (%u/%u). Halting monitor.\n",
                job_queue_length,
                JOB_QUEUE_LIMIT);
    monitor_halted = GNUNET_YES;
    return;
  }
  GNUNET_NAMESTORE_zone_monitor_next (zmon,
                                      1);
}
 
 
static void
handle_monitor_error (void *cls)
{
  (void) cls;
  GNUNET_STATISTICS_update (statistics,
                            "Namestore monitor errors encountered",
                            1,
                            GNUNET_NO);
}
 
 
static void*
sign_worker (void *cls)
{
  struct RecordPublicationJob *job;
  const struct GNUNET_DISK_FileHandle *fh;
 
  fh = GNUNET_DISK_pipe_handle (notification_pipe, GNUNET_DISK_PIPE_END_WRITE);
  while (GNUNET_YES != in_shutdown)
  {
    GNUNET_assert (0 == pthread_mutex_lock (&sign_jobs_lock));
    while (NULL == sign_jobs_head)
      GNUNET_assert (0 == pthread_cond_wait (&sign_jobs_cond, &sign_jobs_lock));
    if (GNUNET_YES == in_shutdown)
    {
      GNUNET_assert (0 == pthread_mutex_unlock (&sign_jobs_lock));
      return NULL;
    }
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Taking on Job for %s\n", sign_jobs_head->label);
    job = sign_jobs_head;
    GNUNET_CONTAINER_DLL_remove (sign_jobs_head, sign_jobs_tail, job);
    GNUNET_assert (0 == pthread_mutex_unlock (&sign_jobs_lock));
    job->ec->seal (job->ec->cls,
                   job->label,
                   job->expire_pub,
                   job->rdata_pub,
                   job->rdata_pub_len,
                   &job->block_pub);
    if (job->rdata_pub != job->rdata_priv)
      job->ec->seal (job->ec->cls,
                     job->label,
                     job->expire_priv,
                     job->rdata_priv,
                     job->rdata_priv_len,
                     &job->block_priv);
    else
      job->block_priv = job->block_pub;
    GNUNET_assert (0 == pthread_mutex_lock (&sign_results_lock));
    GNUNET_CONTAINER_DLL_insert (sign_results_head, sign_results_tail, job);
    GNUNET_assert (0 == pthread_mutex_unlock (&sign_results_lock));
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Done, notifying main thread through pipe!\n");
    GNUNET_DISK_file_write (fh, "!", 1);
  }
  return NULL;
}
 
 
static void
notification_pipe_cb (void *cls)
{
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received wake up notification through pipe, checking results\n");
  GNUNET_SCHEDULER_add_now (&initiate_put_from_pipe_trigger, NULL);
}
 
 
static void
run (void *cls,
     const struct GNUNET_CONFIGURATION_Handle *c,
     struct GNUNET_SERVICE_Handle *service)
{
  unsigned long long max_parallel_bg_queries = 128;
  const struct GNUNET_DISK_FileHandle *np_fh;
 
  (void) cls;
  (void) service;
  pthread_mutex_init (&sign_jobs_lock, NULL);
  pthread_mutex_init (&sign_results_lock, NULL);
  pthread_cond_init (&sign_jobs_cond, NULL);
  last_put_100 = GNUNET_TIME_absolute_get ();  /* first time! */
  min_relative_record_time
    = GNUNET_TIME_UNIT_FOREVER_REL;
  target_iteration_velocity_per_record = INITIAL_ZONE_ITERATION_INTERVAL;
  namestore_handle = GNUNET_NAMESTORE_connect (c);
  if (NULL == namestore_handle)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                _ ("Failed to connect to the namestore!\n"));
    GNUNET_SCHEDULER_shutdown ();
    return;
  }
  disable_namecache = GNUNET_CONFIGURATION_get_value_yesno (c,
                                                            "namecache",
                                                            "DISABLE");
  if (GNUNET_NO == disable_namecache)
  {
    namecache = GNUNET_NAMECACHE_connect (c);
    if (NULL == namecache)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                  _ ("Failed to connect to the namecache!\n"));
      GNUNET_SCHEDULER_shutdown ();
      return;
    }
  }
  cache_keys = GNUNET_CONFIGURATION_get_value_yesno (c,
                                                     "namestore",
                                                     "CACHE_KEYS");
  zone_publish_time_window_default = GNUNET_DHT_DEFAULT_REPUBLISH_FREQUENCY;
  if (GNUNET_OK ==
      GNUNET_CONFIGURATION_get_value_time (c,
                                           "zonemaster",
                                           "ZONE_PUBLISH_TIME_WINDOW",
                                           &zone_publish_time_window_default))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Time window for zone iteration: %s\n",
                GNUNET_STRINGS_relative_time_to_string (
                  zone_publish_time_window_default,
                  GNUNET_YES));
  }
  zone_publish_time_window = zone_publish_time_window_default;
  if (GNUNET_OK ==
      GNUNET_CONFIGURATION_get_value_number (c,
                                             "zonemaster",
                                             "MAX_PARALLEL_BACKGROUND_QUERIES",
                                             &max_parallel_bg_queries))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Number of allowed parallel background queries: %llu\n",
                max_parallel_bg_queries);
  }
  if (0 == max_parallel_bg_queries)
    max_parallel_bg_queries = 1;
  dht_handle = GNUNET_DHT_connect (c,
                                   (unsigned int) max_parallel_bg_queries);
  if (NULL == dht_handle)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                _ ("Could not connect to DHT!\n"));
    GNUNET_SCHEDULER_add_now (&shutdown_task,
                              NULL);
    return;
  }
 
  /* Schedule periodic put for our records. */
  first_zone_iteration = GNUNET_YES;
  statistics = GNUNET_STATISTICS_create ("zonemaster",
                                         c);
  GNUNET_STATISTICS_set (statistics,
                         "Target zone iteration velocity (μs)",
                         target_iteration_velocity_per_record.rel_value_us,
                         GNUNET_NO);
  zone_publish_task = GNUNET_SCHEDULER_add_now (&publish_zone_dht_start,
                                                NULL);
  zmon = GNUNET_NAMESTORE_zone_monitor_start2 (c,
                                               NULL,
                                               GNUNET_NO,
                                               &handle_monitor_error,
                                               NULL,
                                               &handle_monitor_event,
                                               NULL,
                                               NULL /* sync_cb */,
                                               NULL,
                                               GNUNET_GNSRECORD_FILTER_NONE);
  GNUNET_NAMESTORE_zone_monitor_next (zmon,
                                      NAMESTORE_MONITOR_QUEUE_LIMIT - 1);
  GNUNET_break (NULL != zmon);
 
  GNUNET_SCHEDULER_add_shutdown (&shutdown_task,
                                 NULL);
 
  notification_pipe = GNUNET_DISK_pipe (GNUNET_DISK_PF_NONE);
  np_fh = GNUNET_DISK_pipe_handle (
    notification_pipe,
    GNUNET_DISK_PIPE_END_READ);
  pipe_read_task = GNUNET_SCHEDULER_add_read_file (GNUNET_TIME_UNIT_FOREVER_REL,
                                                   np_fh,
                                                   notification_pipe_cb, NULL);
 
  {
    long long unsigned int worker_count = 1;
    if (GNUNET_OK !=
        GNUNET_CONFIGURATION_get_value_number (c,
                                               "zonemaster",
                                               "WORKER_COUNT",
                                               &worker_count))
    {
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Number of workers not defined falling back to 1\n");
    }
    worker = GNUNET_malloc (sizeof (pthread_t) * worker_count);
    for (int i = 0; i < worker_count; i++)
    {
      if (0 !=
          pthread_create (&worker[i],
                          NULL,
                          &sign_worker,
                          NULL))
      {
        GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING,
                             "pthread_create");
        GNUNET_SCHEDULER_shutdown ();
      }
      GNUNET_STATISTICS_update (statistics,
                                "Workers running",
                                1,
                                GNUNET_NO);
    }
  }
}
 
 
GNUNET_SERVICE_MAIN
  (GNUNET_OS_project_data_gnunet (),
  "zonemaster",
  GNUNET_SERVICE_OPTION_NONE,
  &run,
  NULL,
  NULL,
  NULL,
  GNUNET_MQ_handler_end ());
 
 
/* end of gnunet-service-zonemaster.c */

PUBLISH_OPS_PER_EXPIRATION

#define PUBLISH_OPS_PER_EXPIRATION   4

How often should we (re)publish each record before it expires?

Definition at line 44 of file gnunet-service-zonemaster.c.

DELTA_INTERVAL

#define DELTA_INTERVAL   100

How often do we measure the delta between desired zone iteration speed and actual speed, and tell statistics service about it?

Definition at line 51 of file gnunet-service-zonemaster.c.

NS_BLOCK_SIZE

#define NS_BLOCK_SIZE   1000

How many records do we fetch in one shot from the namestore?

Definition at line 56 of file gnunet-service-zonemaster.c.

JOB_QUEUE_LIMIT

#define JOB_QUEUE_LIMIT   5000

How many open jobs (and with it maximum amount of pending DHT operations) do we allow at most.

Definition at line 61 of file gnunet-service-zonemaster.c.

NAMESTORE_MONITOR_QUEUE_LIMIT

#define NAMESTORE_MONITOR_QUEUE_LIMIT   5

How many events may the namestore give us before it has to wait for us to keep up?

Definition at line 67 of file gnunet-service-zonemaster.c.

INITIAL_ZONE_ITERATION_INTERVAL

#define INITIAL_ZONE_ITERATION_INTERVAL   GNUNET_TIME_UNIT_MILLISECONDS

The initial interval in milliseconds btween puts in a zone iteration.

Definition at line 73 of file gnunet-service-zonemaster.c.

MAXIMUM_ZONE_ITERATION_INTERVAL

#define MAXIMUM_ZONE_ITERATION_INTERVAL

Value:

          GNUNET_TIME_relative_multiply ( \
          GNUNET_TIME_UNIT_MINUTES, 15)

The upper bound for the zone iteration interval (per record).

Definition at line 79 of file gnunet-service-zonemaster.c.

LATE_ITERATION_SPEEDUP_FACTOR

#define LATE_ITERATION_SPEEDUP_FACTOR   2

The factor the current zone iteration interval is divided by for each additional new record.

Definition at line 86 of file gnunet-service-zonemaster.c.

DHT_GNS_REPLICATION_LEVEL

#define DHT_GNS_REPLICATION_LEVEL   5

What replication level do we use for DHT PUT operations?

Definition at line 91 of file gnunet-service-zonemaster.c.

Function Documentation

free_job()

static void free_job ( struct RecordPublicationJob *  job )

static

Definition at line 412 of file gnunet-service-zonemaster.c.

{
  if (job->rdata_pub != job->rdata_priv)
    GNUNET_free (job->rdata_priv);
  GNUNET_free (job->rdata_pub);
  if (job->block_pub != job->block_priv)
    GNUNET_free (job->block_priv);
  GNUNET_free (job->block_pub);
  if (NULL != job->label)
    GNUNET_free (job->label);
  if (NULL != job->ec)
  {
    GNUNET_GNSRECORD_encryption_context_destroy (job->ec);
    job->ec = NULL;
  }
  GNUNET_free (job);
}

References GNUNET_free, GNUNET_GNSRECORD_encryption_context_destroy(), and job.

Referenced by cleanup_job(), and shutdown_task().

Here is the call graph for this function:

Here is the caller graph for this function:

shutdown_task()

static void shutdown_task ( void *  cls )

static

Task run during shutdown.

Parameters

cls	unused
tc	unused

Definition at line 438 of file gnunet-service-zonemaster.c.

{
  struct CacheOperation *cop;
  struct RecordPublicationJob *job;
 
  (void) cls;
  in_shutdown = GNUNET_YES;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Shutting down!\n");
  if (NULL != notification_pipe)
    GNUNET_DISK_pipe_close (notification_pipe);
  if (NULL != pipe_read_task)
    GNUNET_SCHEDULER_cancel (pipe_read_task);
  while (NULL != (cop = cop_head))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "Aborting incomplete namecache operation\n");
    GNUNET_NAMECACHE_cancel (cop->qe);
    GNUNET_CONTAINER_DLL_remove (cop_head, cop_tail, cop);
    GNUNET_free (cop);
  }
  GNUNET_assert (0 == pthread_mutex_lock (&sign_jobs_lock));
  while (NULL != (job = sign_jobs_head))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "Removing incomplete jobs\n");
    GNUNET_CONTAINER_DLL_remove (sign_jobs_head, sign_jobs_tail, job);
    job_queue_length--;
    free_job (job);
  }
  GNUNET_assert (0 == pthread_mutex_unlock (&sign_jobs_lock));
  GNUNET_assert (0 == pthread_mutex_lock (&sign_results_lock));
  while (NULL != (job = sign_results_head))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "Removing incomplete jobs\n");
    GNUNET_CONTAINER_DLL_remove (sign_results_head, sign_results_tail, job);
    free_job (job);
  }
  GNUNET_assert (0 == pthread_mutex_unlock (&sign_results_lock));
  while (NULL != (job = dht_jobs_head))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "Removing incomplete jobs\n");
    GNUNET_CONTAINER_DLL_remove (dht_jobs_head, dht_jobs_tail, job);
    if (NULL != job->ph)
      GNUNET_DHT_put_cancel (job->ph);
    free_job (job);
  }
  if (NULL != statistics)
  {
    GNUNET_STATISTICS_destroy (statistics,
                               GNUNET_NO);
    statistics = NULL;
  }
  if (NULL != zone_publish_task)
  {
    GNUNET_SCHEDULER_cancel (zone_publish_task);
    zone_publish_task = NULL;
  }
  if (NULL != namestore_iter)
  {
    GNUNET_NAMESTORE_zone_iteration_stop (namestore_iter);
    namestore_iter = NULL;
  }
  if (NULL != zmon)
  {
    GNUNET_NAMESTORE_zone_monitor_stop (zmon);
    zmon = NULL;
  }
  if (NULL != namestore_handle)
  {
    GNUNET_NAMESTORE_disconnect (namestore_handle);
    namestore_handle = NULL;
  }
  if (NULL != namecache)
  {
    GNUNET_NAMECACHE_disconnect (namecache);
    namecache = NULL;
  }
  if (NULL != dht_handle)
  {
    GNUNET_DHT_disconnect (dht_handle);
    dht_handle = NULL;
  }
}

References cop_head, cop_tail, dht_handle, dht_jobs_head, dht_jobs_tail, free_job(), GNUNET_assert, GNUNET_CONTAINER_DLL_remove, GNUNET_DHT_disconnect(), GNUNET_DHT_put_cancel(), GNUNET_DISK_pipe_close(), GNUNET_ERROR_TYPE_DEBUG, GNUNET_ERROR_TYPE_WARNING, GNUNET_free, GNUNET_log, GNUNET_NAMECACHE_cancel(), GNUNET_NAMECACHE_disconnect(), GNUNET_NAMESTORE_disconnect(), GNUNET_NAMESTORE_zone_iteration_stop(), GNUNET_NAMESTORE_zone_monitor_stop(), GNUNET_NO, GNUNET_SCHEDULER_cancel(), GNUNET_STATISTICS_destroy(), GNUNET_YES, in_shutdown, job, job_queue_length, namecache, namestore_handle, namestore_iter, notification_pipe, pipe_read_task, CacheOperation::qe, sign_jobs_head, sign_jobs_lock, sign_jobs_tail, sign_results_head, sign_results_lock, sign_results_tail, statistics, zmon, and zone_publish_task.

Here is the call graph for this function:

finish_cache_operation()

static void finish_cache_operation ( void *  cls, int32_t  success, const char *  emsg  )

static

Cache operation complete, clean up.

Parameters

cls	the struct CacheOperation
success	success
emsg	error messages

Definition at line 534 of file gnunet-service-zonemaster.c.

{
  struct CacheOperation *cop = cls;
 
  if (NULL != emsg)
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                _ ("Failed to replicate block in namecache: %s\n"),
                emsg);
  else
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "CACHE operation completed\n");
  GNUNET_CONTAINER_DLL_remove (cop_head, cop_tail, cop);
  GNUNET_free (cop);
}

References _, cop_head, cop_tail, GNUNET_CONTAINER_DLL_remove, GNUNET_ERROR_TYPE_DEBUG, GNUNET_ERROR_TYPE_WARNING, GNUNET_free, and GNUNET_log.

Referenced by refresh_block().

Here is the caller graph for this function:

refresh_block()

static void refresh_block ( const struct GNUNET_GNSRECORD_Block *  block )

static

Refresh the (encrypted) block in the namecache.

Parameters

zone_key	private key of the zone
name	label for the records
rd_count	number of records
rd	records stored under the given name

Definition at line 558 of file gnunet-service-zonemaster.c.

{
  struct CacheOperation *cop;
 
  if (GNUNET_YES == disable_namecache)
  {
    GNUNET_STATISTICS_update (statistics,
                              "Namecache updates skipped (NC disabled)",
                              1,
                              GNUNET_NO);
    return;
  }
  GNUNET_assert (NULL != block);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Caching block in namecache\n");
  GNUNET_STATISTICS_update (statistics,
                            "Namecache updates pushed",
                            1,
                            GNUNET_NO);
  cop = GNUNET_new (struct CacheOperation);
  GNUNET_CONTAINER_DLL_insert (cop_head, cop_tail, cop);
  cop->qe = GNUNET_NAMECACHE_block_cache (namecache,
                                          block,
                                          &finish_cache_operation,
                                          cop);
}

References cop_head, cop_tail, disable_namecache, finish_cache_operation(), GNUNET_assert, GNUNET_CONTAINER_DLL_insert, GNUNET_ERROR_TYPE_DEBUG, GNUNET_log, GNUNET_NAMECACHE_block_cache(), GNUNET_new, GNUNET_NO, GNUNET_STATISTICS_update(), GNUNET_YES, namecache, CacheOperation::qe, and statistics.

Referenced by initiate_put_from_pipe_trigger().

Here is the call graph for this function:

Here is the caller graph for this function:

publish_zone_namestore_next()

static void publish_zone_namestore_next ( void *  cls )

static

Method called periodically that triggers iteration over authoritative records.

Parameters

cls

NULL

Definition at line 591 of file gnunet-service-zonemaster.c.

{
  (void) cls;
  zone_publish_task = NULL;
  GNUNET_assert (NULL != namestore_iter);
  GNUNET_assert (0 == ns_iteration_left);
  ns_iteration_left = NS_BLOCK_SIZE;
  GNUNET_NAMESTORE_zone_iterator_next (namestore_iter,
                                       NS_BLOCK_SIZE);
}

References GNUNET_assert, GNUNET_NAMESTORE_zone_iterator_next(), namestore_iter, NS_BLOCK_SIZE, ns_iteration_left, and zone_publish_task.

Referenced by check_zone_namestore_next().

Here is the call graph for this function:

Here is the caller graph for this function:

publish_zone_dht_start()

static void publish_zone_dht_start ( void *  cls )

static

Periodically iterate over our zone and store everything in dht.

Periodically iterate over all zones and store everything in DHT.

Parameters

cls

NULL

Definition at line 1197 of file gnunet-service-zonemaster.c.

{
  (void) cls;
  zone_publish_task = NULL;
  GNUNET_STATISTICS_update (statistics,
                            "Full zone iterations launched",
                            1,
                            GNUNET_NO);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Starting DHT zone update!\n");
  /* start counting again */
  num_public_records = 0;
  GNUNET_assert (NULL == namestore_iter);
  ns_iteration_left = 1;
  namestore_iter
    = GNUNET_NAMESTORE_zone_iteration_start2 (namestore_handle,
                                              NULL, /* All zones */
                                              &zone_iteration_error,
                                              NULL,
                                              &handle_record,
                                              NULL,
                                              &zone_iteration_finished,
                                              NULL,
                                              GNUNET_GNSRECORD_FILTER_NONE);
  GNUNET_assert (NULL != namestore_iter);
}

References GNUNET_assert, GNUNET_ERROR_TYPE_DEBUG, GNUNET_GNSRECORD_FILTER_NONE, GNUNET_log, GNUNET_NAMESTORE_zone_iteration_start2(), GNUNET_NO, GNUNET_STATISTICS_update(), handle_record(), namestore_handle, namestore_iter, ns_iteration_left, num_public_records, statistics, zone_iteration_error(), zone_iteration_finished(), and zone_publish_task.

Referenced by handle_monitor_event(), run(), zone_iteration_error(), and zone_iteration_finished().

Here is the call graph for this function:

Here is the caller graph for this function:

calculate_put_interval()

static void calculate_put_interval ( )

static

Calculate target_iteration_velocity_per_record.

If no records are known (startup) or none present we can safely set the interval to the value for a single record

Definition at line 616 of file gnunet-service-zonemaster.c.

{
  /* The window follows from the record lifetimes alone: a record set has
     to be republished #PUBLISH_OPS_PER_EXPIRATION times before it
     expires, whether or not we have counted the zone yet. */
  last_min_relative_record_time
    = GNUNET_TIME_relative_min (last_min_relative_record_time,
                                min_relative_record_time);
  zone_publish_time_window
    = GNUNET_TIME_relative_min (GNUNET_TIME_relative_divide (
                                  last_min_relative_record_time,
                                  PUBLISH_OPS_PER_EXPIRATION),
                                zone_publish_time_window_default);
  if (0 == num_public_records)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
                "No records in namestore database.\n");
  }
  else
  {
    target_iteration_velocity_per_record
      = GNUNET_TIME_relative_divide (zone_publish_time_window,
                                     last_num_public_records);
  }
  target_iteration_velocity_per_record
    = GNUNET_TIME_relative_min (target_iteration_velocity_per_record,
                                MAXIMUM_ZONE_ITERATION_INTERVAL);
  GNUNET_STATISTICS_set (statistics,
                         "Minimum relative record expiration (in μs)",
                         last_min_relative_record_time.rel_value_us,
                         GNUNET_NO);
  GNUNET_STATISTICS_set (statistics,
                         "Zone publication time window (in μs)",
                         zone_publish_time_window.rel_value_us,
                         GNUNET_NO);
  GNUNET_STATISTICS_set (statistics,
                         "Target zone iteration velocity (μs)",
                         target_iteration_velocity_per_record.rel_value_us,
                         GNUNET_NO);
}

References GNUNET_ERROR_TYPE_BULK, GNUNET_ERROR_TYPE_DEBUG, GNUNET_log, GNUNET_NO, GNUNET_STATISTICS_set(), GNUNET_TIME_relative_divide(), GNUNET_TIME_relative_min(), last_min_relative_record_time, last_num_public_records, MAXIMUM_ZONE_ITERATION_INTERVAL, min_relative_record_time, num_public_records, PUBLISH_OPS_PER_EXPIRATION, GNUNET_TIME_Relative::rel_value_us, statistics, target_iteration_velocity_per_record, zone_publish_time_window, and zone_publish_time_window_default.

Referenced by handle_monitor_event(), update_velocity(), and zone_iteration_finished().

Here is the call graph for this function:

Here is the caller graph for this function:

note_relative_expiration()

static void note_relative_expiration ( const struct GNUNET_GNSRECORD_Data *  rd, unsigned int  rd_count  )

static

Note the relative expiration times in rd, they determine how often the zone has to be republished.

Parameters

rd	record set
rd_count	number of records in rd

Definition at line 671 of file gnunet-service-zonemaster.c.

{
  for (unsigned int i = 0; i < rd_count; i++)
  {
    if (0 == (rd[i].flags & GNUNET_GNSRECORD_RF_RELATIVE_EXPIRATION))
      continue;
    /* GNUNET_GNSRECORD_block_create will convert to absolute time;
       we just need to adjust our iteration frequency */
    min_relative_record_time.rel_value_us
      = GNUNET_MIN (rd[i].expiration_time,
                    min_relative_record_time.rel_value_us);
  }
}

References GNUNET_GNSRECORD_RF_RELATIVE_EXPIRATION, GNUNET_MIN, min_relative_record_time, rd, rd_count, and GNUNET_TIME_Relative::rel_value_us.

Referenced by handle_monitor_event(), and handle_record().

Here is the caller graph for this function:

update_velocity()

static void update_velocity ( unsigned int  cnt )

static

Re-calculate our velocity and the desired velocity.

We have succeeded in making DELTA_INTERVAL puts, so now calculate the new desired delay between puts.

Parameters

cnt	how many records were processed since the last call?

Definition at line 695 of file gnunet-service-zonemaster.c.

{
  struct GNUNET_TIME_Relative delta;
  unsigned long long pct = 0;
 
  if (0 == cnt)
    return;
  /* How fast were we really? */
  delta = GNUNET_TIME_absolute_get_duration (last_put_100);
  delta.rel_value_us /= cnt;
  last_put_100 = GNUNET_TIME_absolute_get ();
 
  /* calculate expected frequency */
  if ((num_public_records > last_num_public_records) &&
      (GNUNET_NO == first_zone_iteration))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Last record count was lower than current record count.  Reducing interval.\n");
    last_num_public_records = num_public_records
                              * LATE_ITERATION_SPEEDUP_FACTOR;
    calculate_put_interval ();
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Desired global zone iteration interval is %s/record!\n",
              GNUNET_STRINGS_relative_time_to_string (
                target_iteration_velocity_per_record,
                GNUNET_YES));
 
  /* Tell statistics actual vs. desired speed */
  GNUNET_STATISTICS_set (statistics,
                         "Current zone iteration velocity (μs/record)",
                         delta.rel_value_us,
                         GNUNET_NO);
  /* update "sub_delta" based on difference, taking
     previous sub_delta into account! */
  if (target_iteration_velocity_per_record.rel_value_us > delta.rel_value_us)
  {
    /* We were too fast, reduce sub_delta! */
    struct GNUNET_TIME_Relative corr;
 
    corr = GNUNET_TIME_relative_subtract (target_iteration_velocity_per_record,
                                          delta);
    if (sub_delta.rel_value_us > delta.rel_value_us)
    {
      /* Reduce sub_delta by corr */
      sub_delta = GNUNET_TIME_relative_subtract (sub_delta,
                                                 corr);
    }
    else
    {
      /* We're doing fine with waiting the full time, this
         should theoretically only happen if we run at
         infinite speed. */
      sub_delta = GNUNET_TIME_UNIT_ZERO;
    }
  }
  else if (target_iteration_velocity_per_record.rel_value_us <
           delta.rel_value_us)
  {
    /* We were too slow, increase sub_delta! */
    struct GNUNET_TIME_Relative corr;
 
    corr = GNUNET_TIME_relative_subtract (delta,
                                          target_iteration_velocity_per_record);
    sub_delta = GNUNET_TIME_relative_add (sub_delta,
                                          corr);
    if (sub_delta.rel_value_us >
        target_iteration_velocity_per_record.rel_value_us)
    {
      /* CPU overload detected, we cannot go at desired speed,
         as this would mean using a negative delay. */
      /* compute how much faster we would want to be for
         the desired velocity */
      if (0 == target_iteration_velocity_per_record.rel_value_us)
        pct = UINT64_MAX;     /* desired speed is infinity ... */
      else
        pct = (sub_delta.rel_value_us
               - target_iteration_velocity_per_record.rel_value_us) * 100LLU
              / target_iteration_velocity_per_record.rel_value_us;
      sub_delta = target_iteration_velocity_per_record;
    }
  }
  GNUNET_STATISTICS_set (statistics,
                         "# dispatched jobs",
                         job_queue_length,
                         GNUNET_NO);
  GNUNET_STATISTICS_set (statistics,
                         "% speed increase needed for target velocity",
                         pct,
                         GNUNET_NO);
  GNUNET_STATISTICS_set (statistics,
                         "# records processed in current iteration",
                         num_public_records,
                         GNUNET_NO);
}

References calculate_put_interval(), delta, first_zone_iteration, GNUNET_ERROR_TYPE_DEBUG, GNUNET_log, GNUNET_NO, GNUNET_STATISTICS_set(), GNUNET_STRINGS_relative_time_to_string(), GNUNET_TIME_absolute_get(), GNUNET_TIME_absolute_get_duration(), GNUNET_TIME_relative_add(), GNUNET_TIME_relative_subtract(), GNUNET_TIME_UNIT_ZERO, GNUNET_YES, job_queue_length, last_num_public_records, last_put_100, LATE_ITERATION_SPEEDUP_FACTOR, num_public_records, GNUNET_TIME_Relative::rel_value_us, statistics, sub_delta, and target_iteration_velocity_per_record.

Referenced by check_zone_namestore_next(), and handle_record().

Here is the call graph for this function:

Here is the caller graph for this function:

check_zone_namestore_next()

static void check_zone_namestore_next ( )

static

Check if the current zone iteration needs to be continued by calling publish_zone_namestore_next(), and if so with what delay.

Definition at line 797 of file gnunet-service-zonemaster.c.

{
  struct GNUNET_TIME_Relative delay;
 
  if (0 != ns_iteration_left)
    return; /* current NAMESTORE iteration not yet done */
  if (job_queue_length >= JOB_QUEUE_LIMIT)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "Job queue length exceeded (%u/%u). Pausing namestore iteration.\n",
                job_queue_length,
                JOB_QUEUE_LIMIT);
    return;
  }
  update_velocity (put_cnt);
  put_cnt = 0;
  delay = GNUNET_TIME_relative_subtract (target_iteration_velocity_per_record,
                                         sub_delta);
  /* We delay *once* per #NS_BLOCK_SIZE, so we need to multiply the
     per-record delay calculated so far with the #NS_BLOCK_SIZE */
  GNUNET_STATISTICS_set (statistics,
                         "Current artificial NAMESTORE delay (μs/record)",
                         delay.rel_value_us,
                         GNUNET_NO);
  delay = GNUNET_TIME_relative_multiply (delay,
                                         GNUNET_MIN (NS_BLOCK_SIZE,
                                                     GNUNET_MAX (
                                                       1,
                                                       last_num_public_records))
                                         );
  /* make sure we do not overshoot because of the #NS_BLOCK_SIZE factor.
     The pause is part of one pass over the zone, so it may never exceed
     the window that pass has to fit into: a record set has to be
     republished #PUBLISH_OPS_PER_EXPIRATION times before it expires, and
     with a short expiration time the whole window is shorter than
     #MAXIMUM_ZONE_ITERATION_INTERVAL. */
  delay = GNUNET_TIME_relative_min (delay,
                                    GNUNET_TIME_relative_min (
                                      MAXIMUM_ZONE_ITERATION_INTERVAL,
                                      zone_publish_time_window));
  /* no delays on first iteration */
  if (GNUNET_YES == first_zone_iteration)
    delay = GNUNET_TIME_UNIT_ZERO;
  GNUNET_assert (NULL == zone_publish_task);
  zone_publish_task = GNUNET_SCHEDULER_add_delayed (delay,
                                                    &publish_zone_namestore_next
                                                    ,
                                                    NULL);
}

References first_zone_iteration, GNUNET_assert, GNUNET_ERROR_TYPE_WARNING, GNUNET_log, GNUNET_MAX, GNUNET_MIN, GNUNET_NO, GNUNET_SCHEDULER_add_delayed(), GNUNET_STATISTICS_set(), GNUNET_TIME_relative_min(), GNUNET_TIME_relative_multiply(), GNUNET_TIME_relative_subtract(), GNUNET_TIME_UNIT_ZERO, GNUNET_YES, job_queue_length, JOB_QUEUE_LIMIT, last_num_public_records, MAXIMUM_ZONE_ITERATION_INTERVAL, NS_BLOCK_SIZE, ns_iteration_left, publish_zone_namestore_next(), put_cnt, GNUNET_TIME_Relative::rel_value_us, statistics, sub_delta, target_iteration_velocity_per_record, update_velocity(), zone_publish_task, and zone_publish_time_window.

Referenced by cleanup_job(), and handle_record().

Here is the call graph for this function:

Here is the caller graph for this function:

cleanup_job()

static void cleanup_job ( struct RecordPublicationJob *  job )

static

Definition at line 849 of file gnunet-service-zonemaster.c.

{
  GNUNET_assert (job_queue_length > 0);
  job_queue_length--;
  if (NULL == zone_publish_task)
    check_zone_namestore_next ();
  if (job_queue_length <= JOB_QUEUE_LIMIT)
  {
    if (GNUNET_YES == monitor_halted)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                  "Job queue emptied (%u/%u). Resuming monitor.\n",
                  job_queue_length,
                  JOB_QUEUE_LIMIT);
      GNUNET_NAMESTORE_zone_monitor_next (zmon, 1);
      monitor_halted = GNUNET_NO;
    }
  }
  free_job (job);
}

References check_zone_namestore_next(), free_job(), GNUNET_assert, GNUNET_ERROR_TYPE_WARNING, GNUNET_log, GNUNET_NAMESTORE_zone_monitor_next(), GNUNET_NO, GNUNET_YES, job, job_queue_length, JOB_QUEUE_LIMIT, monitor_halted, zmon, and zone_publish_task.

Referenced by dht_put_continuation(), and initiate_put_from_pipe_trigger().

Here is the call graph for this function:

Here is the caller graph for this function:

dht_put_continuation()

static void dht_put_continuation ( void *  cls )

static

Continuation called from DHT once the PUT operation is done.

Definition at line 876 of file gnunet-service-zonemaster.c.

{
  struct RecordPublicationJob *job = cls;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "PUT complete; Pending jobs: %u\n", job_queue_length - 1);
  /* When we just fall under the limit, trigger monitor/iterator again
   * if halted. We can only safely trigger one, prefer iterator. */
  GNUNET_CONTAINER_DLL_remove (dht_jobs_head,
                               dht_jobs_tail,
                               job);
  cleanup_job (job);
}

References cleanup_job(), dht_jobs_head, dht_jobs_tail, GNUNET_CONTAINER_DLL_remove, GNUNET_ERROR_TYPE_DEBUG, GNUNET_log, job, and job_queue_length.

Referenced by initiate_put_from_pipe_trigger().

Here is the call graph for this function:

Here is the caller graph for this function:

dispatch_job()

static void dispatch_job ( const struct GNUNET_CRYPTO_BlindablePrivateKey *  key, const char *  label, const struct GNUNET_GNSRECORD_Data *  rd, unsigned int  rd_count, const struct GNUNET_TIME_Absolute  expire  )

static

Store GNS records in the DHT.

Parameters

key	key of the zone
label	label to store under
rd_public	public record data
rd_public_count	number of records in rd_public
ma	handle for the put operation

Returns

DHT PUT handle, NULL on error

Definition at line 902 of file gnunet-service-zonemaster.c.

{
  struct GNUNET_GNSRECORD_Data rd_public[rd_count];
  struct GNUNET_TIME_Absolute expire_pub;
  struct RecordPublicationJob *job;
  unsigned char *rdata_public;
  unsigned char *rdata_priv;
  size_t rdata_public_len;
  size_t rdata_private_len;
  unsigned int rd_public_count = 0;
  char *emsg;
 
  if (GNUNET_OK !=
      GNUNET_GNSRECORD_normalize_record_set (label,
                                             rd,
                                             rd_count,
                                             rd_public,
                                             &rd_public_count,
                                             &expire_pub,
                                             GNUNET_GNSRECORD_FILTER_OMIT_PRIVATE,
                                             &emsg))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "%s\n", emsg);
    GNUNET_free (emsg);
  }
 
  rdata_public_len = GNUNET_GNSRECORD_records_get_size (rd_public_count,
                                                        rd_public);
 
  rdata_private_len = GNUNET_GNSRECORD_records_get_size (rd_count,
                                                         rd);
 
  rdata_public = GNUNET_malloc (rdata_public_len);
  GNUNET_assert (GNUNET_OK == GNUNET_GNSRECORD_record_data_to_rdata (
                   rd_public_count,
                   rd_public,
                   rdata_public,
                   rdata_public_len));
  if (rd_count != rd_public_count)
  {
    rdata_priv = GNUNET_malloc (rdata_private_len);
    GNUNET_assert (GNUNET_OK == GNUNET_GNSRECORD_record_data_to_rdata (
                     rd_count,
                     rd,
                     rdata_priv,
                     rdata_private_len));
  }
  else
  {
    rdata_priv = rdata_public;
  }
  GNUNET_assert (0 == pthread_mutex_lock (&sign_jobs_lock));
  job = GNUNET_new (struct RecordPublicationJob);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Public record count: %d\n",
              rd_public_count);
  job->have_public_records = (rd_public_count > 0);
  job->rdata_pub = rdata_public;
  job->rdata_pub_len = rdata_public_len;
  job->rdata_priv = rdata_priv;
  job->rdata_priv_len = rdata_private_len;
  job->zone = *key;
  job->label = GNUNET_strdup (label);
  job->expire_pub = expire_pub;
  job->expire_priv = expire;
  job->ec = GNUNET_GNSRECORD_encryption_context_setup_owner (key);
  GNUNET_CONTAINER_DLL_insert (sign_jobs_head, sign_jobs_tail, job);
  GNUNET_assert (0 == pthread_cond_signal (&sign_jobs_cond));
  GNUNET_assert (0 == pthread_mutex_unlock (&sign_jobs_lock));
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Creating job with %u record(s) for label `%s', expiration `%s'\n",
              rd_public_count,
              label,
              GNUNET_STRINGS_absolute_time_to_string (expire));
  num_public_records++;
  return;
}

References expire, RecordPublicationJob::expire_pub, GNUNET_assert, GNUNET_CONTAINER_DLL_insert, GNUNET_ERROR_TYPE_DEBUG, GNUNET_ERROR_TYPE_ERROR, GNUNET_free, GNUNET_GNSRECORD_encryption_context_setup_owner(), GNUNET_GNSRECORD_FILTER_OMIT_PRIVATE, GNUNET_GNSRECORD_normalize_record_set(), GNUNET_GNSRECORD_record_data_to_rdata(), GNUNET_GNSRECORD_records_get_size(), GNUNET_log, GNUNET_malloc, GNUNET_new, GNUNET_OK, GNUNET_strdup, GNUNET_STRINGS_absolute_time_to_string(), job, key, RecordPublicationJob::label, num_public_records, rd, rd_count, RecordPublicationJob::rdata_priv, sign_jobs_cond, sign_jobs_head, sign_jobs_lock, and sign_jobs_tail.

Referenced by dispatch_job_monitor(), and handle_record().

Here is the call graph for this function:

Here is the caller graph for this function:

notification_pipe_cb()

static void notification_pipe_cb ( void *  cls )

static

Definition at line 1386 of file gnunet-service-zonemaster.c.

{
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received wake up notification through pipe, checking results\n");
  GNUNET_SCHEDULER_add_now (&initiate_put_from_pipe_trigger, NULL);
}

References GNUNET_ERROR_TYPE_DEBUG, GNUNET_log, GNUNET_SCHEDULER_add_now(), and initiate_put_from_pipe_trigger().

Referenced by initiate_put_from_pipe_trigger(), and run().

Here is the call graph for this function:

Here is the caller graph for this function:

initiate_put_from_pipe_trigger()

static void initiate_put_from_pipe_trigger ( void *  cls )

static

Definition at line 989 of file gnunet-service-zonemaster.c.

{
  struct GNUNET_HashCode query;
  struct RecordPublicationJob *job;
  const struct GNUNET_DISK_FileHandle *np_fh;
  char buf[100];
  ssize_t nf_count;
 
  pipe_read_task = NULL;
  np_fh = GNUNET_DISK_pipe_handle (notification_pipe,
                                   GNUNET_DISK_PIPE_END_READ);
  pipe_read_task =
    GNUNET_SCHEDULER_add_read_file (
      GNUNET_TIME_UNIT_FOREVER_REL,
      np_fh,
      notification_pipe_cb,
      NULL);
  /* empty queue */
  nf_count = GNUNET_DISK_file_read (np_fh, buf, sizeof (buf));
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Read %lld notifications from pipe\n",
              (long long) nf_count);
  while (true)
  {
    GNUNET_assert (0 == pthread_mutex_lock (&sign_results_lock));
    if (NULL == sign_results_head)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "No more results. Back to sleep.\n");
      GNUNET_assert (0 == pthread_mutex_unlock (&sign_results_lock));
      return;
    }
    job = sign_results_head;
    GNUNET_CONTAINER_DLL_remove (sign_results_head, sign_results_tail, job);
    GNUNET_assert (0 == pthread_mutex_unlock (&sign_results_lock));
    GNUNET_GNSRECORD_query_from_private_key (&job->zone,
                                             job->label,
                                             &query);
    // It is possible that the public block size is 0 (no public blocks)
    // Do not bother with the DHT in that case
    if (job->have_public_records)
    {
      size_t block_size = GNUNET_GNSRECORD_block_get_size (job->block_pub);
      job->ph = GNUNET_DHT_put (dht_handle,
                                &query,
                                DHT_GNS_REPLICATION_LEVEL,
                                GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,
                                GNUNET_BLOCK_TYPE_GNS_NAMERECORD,
                                block_size,
                                job->block_pub,
                                job->expire_pub,
                                &dht_put_continuation,
                                job);
      if (NULL == job->ph)
      {
        /* Drop this job, but keep draining the queue: returning here
           leaves every block the workers already signed stranded, and
           losing the job_queue_length slot eventually halts both the
           zone iteration and the monitor for good. */
        GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                    "Could not perform DHT PUT, is the DHT running?\n");
        cleanup_job (job);
        continue;
      }
      GNUNET_STATISTICS_update (statistics,
                                "DHT put operations initiated",
                                1,
                                GNUNET_NO);
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Storing record(s) for label `%s' in DHT under key %s\n",
                  job->label,
                  GNUNET_h2s (&query));
      refresh_block (job->block_priv);
      GNUNET_CONTAINER_DLL_insert (dht_jobs_head, dht_jobs_tail, job);
    }
    else
    {
      // Private blocks may still be available and must be updated
      // in the cache
      refresh_block (job->block_priv);
      cleanup_job (job);
    }
  }
}

References cleanup_job(), DHT_GNS_REPLICATION_LEVEL, dht_handle, dht_jobs_head, dht_jobs_tail, dht_put_continuation(), GNUNET_assert, GNUNET_BLOCK_TYPE_GNS_NAMERECORD, GNUNET_CONTAINER_DLL_insert, GNUNET_CONTAINER_DLL_remove, GNUNET_DHT_put(), GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE, GNUNET_DISK_file_read(), GNUNET_DISK_PIPE_END_READ, GNUNET_DISK_pipe_handle(), GNUNET_ERROR_TYPE_DEBUG, GNUNET_ERROR_TYPE_WARNING, GNUNET_GNSRECORD_block_get_size(), GNUNET_GNSRECORD_query_from_private_key(), GNUNET_h2s(), GNUNET_log, GNUNET_NO, GNUNET_SCHEDULER_add_read_file(), GNUNET_STATISTICS_update(), GNUNET_TIME_UNIT_FOREVER_REL, job, notification_pipe, notification_pipe_cb(), pipe_read_task, refresh_block(), sign_results_head, sign_results_lock, sign_results_tail, and statistics.

Referenced by notification_pipe_cb().

Here is the call graph for this function:

Here is the caller graph for this function:

zone_iteration_error()

static void zone_iteration_error ( void *  cls )

static

We encountered an error in our zone iteration.

Parameters

cls

NULL

Definition at line 1080 of file gnunet-service-zonemaster.c.

{
  (void) cls;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Got disconnected from namestore database, retrying.\n");
  namestore_iter = NULL;
  /* We end up here on error/disconnect/shutdown, so potentially
     while a zone publish task or a DHT put is still running; hence
     we need to cancel those. */
  if (NULL != zone_publish_task)
  {
    GNUNET_SCHEDULER_cancel (zone_publish_task);
    zone_publish_task = NULL;
  }
  zone_publish_task = GNUNET_SCHEDULER_add_now (&publish_zone_dht_start,
                                                NULL);
}

References GNUNET_ERROR_TYPE_DEBUG, GNUNET_log, GNUNET_SCHEDULER_add_now(), GNUNET_SCHEDULER_cancel(), namestore_iter, publish_zone_dht_start(), and zone_publish_task.

Referenced by publish_zone_dht_start().

Here is the call graph for this function:

Here is the caller graph for this function:

zone_iteration_finished()

static void zone_iteration_finished ( void *  cls )

static

Zone iteration is completed.

Parameters

cls

NULL

Definition at line 1105 of file gnunet-service-zonemaster.c.

{
  (void) cls;
  /* we're done with one iteration, calculate when to do the next one */
  namestore_iter = NULL;
  last_num_public_records = num_public_records;
  first_zone_iteration = GNUNET_NO;
  last_min_relative_record_time = min_relative_record_time;
  calculate_put_interval ();
  /* reset for next iteration */
  min_relative_record_time
    = GNUNET_TIME_UNIT_FOREVER_REL;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Zone iteration finished. Adjusted zone iteration interval to %s\n",
              GNUNET_STRINGS_relative_time_to_string (
                target_iteration_velocity_per_record,
                GNUNET_YES));
  GNUNET_STATISTICS_set (statistics,
                         "Target zone iteration velocity (μs)",
                         target_iteration_velocity_per_record.rel_value_us,
                         GNUNET_NO);
  GNUNET_STATISTICS_set (statistics,
                         "Number of public records in DHT",
                         last_num_public_records,
                         GNUNET_NO);
  GNUNET_assert (NULL == zone_publish_task);
  if (0 == last_num_public_records)
  {
    zone_publish_task = GNUNET_SCHEDULER_add_delayed (
      target_iteration_velocity_per_record,
      &publish_zone_dht_start,
      NULL);
  }
  else
  {
    zone_publish_task = GNUNET_SCHEDULER_add_now (&publish_zone_dht_start,
                                                  NULL);
  }
}

References calculate_put_interval(), first_zone_iteration, GNUNET_assert, GNUNET_ERROR_TYPE_DEBUG, GNUNET_log, GNUNET_NO, GNUNET_SCHEDULER_add_delayed(), GNUNET_SCHEDULER_add_now(), GNUNET_STATISTICS_set(), GNUNET_STRINGS_relative_time_to_string(), GNUNET_TIME_UNIT_FOREVER_REL, GNUNET_YES, last_min_relative_record_time, last_num_public_records, min_relative_record_time, namestore_iter, num_public_records, publish_zone_dht_start(), GNUNET_TIME_Relative::rel_value_us, statistics, target_iteration_velocity_per_record, and zone_publish_task.

Referenced by publish_zone_dht_start().

Here is the call graph for this function:

Here is the caller graph for this function:

handle_record()

static void handle_record ( void *  cls, const struct GNUNET_CRYPTO_BlindablePrivateKey *  key, const char *  label, unsigned int  rd_count, const struct GNUNET_GNSRECORD_Data *  rd, struct GNUNET_TIME_Absolute  expire  )

static

Function used to put all records successively into the DHT.

Parameters

cls	the closure (NULL)
key	the private key of the authority (ours)
label	the name of the records, NULL once the iteration is done
rd_count	the number of records in rd
rd	the record data

Definition at line 1156 of file gnunet-service-zonemaster.c.

{
  (void) cls;
  ns_iteration_left--;
  if (0 == rd_count)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Record set empty, moving to next record set\n");
    check_zone_namestore_next ();
    return;
  }
  note_relative_expiration (rd,
                            rd_count);
 
  /* We got a set of records to publish */
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Starting DHT PUT\n");
  put_cnt++;
  if (0 == put_cnt % DELTA_INTERVAL)
    update_velocity (DELTA_INTERVAL);
  dispatch_job (key,
                label,
                rd,
                rd_count,
                expire);
  job_queue_length++;
  check_zone_namestore_next ();
}

References check_zone_namestore_next(), DELTA_INTERVAL, dispatch_job(), expire, GNUNET_ERROR_TYPE_DEBUG, GNUNET_log, job_queue_length, key, note_relative_expiration(), ns_iteration_left, put_cnt, rd, rd_count, and update_velocity().

Referenced by publish_zone_dht_start().

Here is the call graph for this function:

Here is the caller graph for this function:

dispatch_job_monitor()

static void dispatch_job_monitor ( const struct GNUNET_CRYPTO_BlindablePrivateKey *  key, const char *  label, const struct GNUNET_GNSRECORD_Data *  rd, unsigned int  rd_count, struct GNUNET_TIME_Absolute  expire  )

static

Store GNS records in the DHT.

Parameters

key	key of the zone
label	label to store under
rd_public	public record data
rd_public_count	number of records in rd_public

Definition at line 1234 of file gnunet-service-zonemaster.c.

{
  dispatch_job (key, label, rd, rd_count, expire);
}

References dispatch_job(), expire, key, rd, and rd_count.

Referenced by handle_monitor_event().

Here is the call graph for this function:

Here is the caller graph for this function:

handle_monitor_event()

static void handle_monitor_event ( void *  cls, const struct GNUNET_CRYPTO_BlindablePrivateKey *  zone, const char *  label, unsigned int  rd_count, const struct GNUNET_GNSRECORD_Data *  rd, struct GNUNET_TIME_Absolute  expire  )

static

Process a record that was stored in the namestore (invoked by the monitor).

Parameters

cls	closure, NULL
zone	private key of the zone; NULL on disconnect
label	label of the records; NULL on disconnect
rd_count	number of entries in rd array, 0 if label was deleted
rd	array of records with data to store
expire	expiration of this record set

Definition at line 1256 of file gnunet-service-zonemaster.c.

{
  (void) cls;
  GNUNET_STATISTICS_update (statistics,
                            "Namestore monitor events received",
                            1,
                            GNUNET_NO);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received %u records for label `%s' via namestore monitor\n",
              rd_count,
              label);
  if (0 == rd_count)
  {
    GNUNET_NAMESTORE_zone_monitor_next (zmon,
                                        1);
    return;   /* nothing to do */
  }
  note_relative_expiration (rd,
                            rd_count);
  if ( (GNUNET_TIME_relative_cmp (min_relative_record_time,
                                  <,
                                  last_min_relative_record_time)) &&
       (NULL == namestore_iter) &&
       (NULL != zone_publish_task) )
  {
    /* This record set expires sooner than anything we saw while
       iterating, so the next iteration -- which is what republishes it --
       has to happen sooner than the one we scheduled back then.  Without
       this a record with a lifetime below
       #MAXIMUM_ZONE_ITERATION_INTERVAL is gone from the DHT long before
       it is put there again. */
    last_min_relative_record_time = min_relative_record_time;
    calculate_put_interval ();
    GNUNET_SCHEDULER_cancel (zone_publish_task);
    zone_publish_task
      = GNUNET_SCHEDULER_add_delayed (target_iteration_velocity_per_record,
                                      &publish_zone_dht_start,
                                      NULL);
  }
  dispatch_job_monitor (zone,
                        label,
                        rd,
                        rd_count,
                        expire);
  job_queue_length++;
  if (job_queue_length >= JOB_QUEUE_LIMIT)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "Job queue length exceeded (%u/%u). Halting monitor.\n",
                job_queue_length,
                JOB_QUEUE_LIMIT);
    monitor_halted = GNUNET_YES;
    return;
  }
  GNUNET_NAMESTORE_zone_monitor_next (zmon,
                                      1);
}

References calculate_put_interval(), dispatch_job_monitor(), expire, GNUNET_ERROR_TYPE_DEBUG, GNUNET_ERROR_TYPE_WARNING, GNUNET_log, GNUNET_NAMESTORE_zone_monitor_next(), GNUNET_NO, GNUNET_SCHEDULER_add_delayed(), GNUNET_SCHEDULER_cancel(), GNUNET_STATISTICS_update(), GNUNET_TIME_relative_cmp, GNUNET_YES, job_queue_length, JOB_QUEUE_LIMIT, last_min_relative_record_time, min_relative_record_time, monitor_halted, namestore_iter, note_relative_expiration(), publish_zone_dht_start(), rd, rd_count, statistics, target_iteration_velocity_per_record, zmon, and zone_publish_task.

Referenced by run().

Here is the call graph for this function:

Here is the caller graph for this function:

handle_monitor_error()

static void handle_monitor_error ( void *  cls )

static

The zone monitor encountered an IPC error trying to to get in sync.

Restart from the beginning.

Parameters

cls

NULL

Definition at line 1327 of file gnunet-service-zonemaster.c.

{
  (void) cls;
  GNUNET_STATISTICS_update (statistics,
                            "Namestore monitor errors encountered",
                            1,
                            GNUNET_NO);
}

References GNUNET_NO, GNUNET_STATISTICS_update(), and statistics.

Referenced by run().

Here is the call graph for this function:

Here is the caller graph for this function:

sign_worker()

static void * sign_worker ( void *  cls )

static

Definition at line 1338 of file gnunet-service-zonemaster.c.

{
  struct RecordPublicationJob *job;
  const struct GNUNET_DISK_FileHandle *fh;
 
  fh = GNUNET_DISK_pipe_handle (notification_pipe, GNUNET_DISK_PIPE_END_WRITE);
  while (GNUNET_YES != in_shutdown)
  {
    GNUNET_assert (0 == pthread_mutex_lock (&sign_jobs_lock));
    while (NULL == sign_jobs_head)
      GNUNET_assert (0 == pthread_cond_wait (&sign_jobs_cond, &sign_jobs_lock));
    if (GNUNET_YES == in_shutdown)
    {
      GNUNET_assert (0 == pthread_mutex_unlock (&sign_jobs_lock));
      return NULL;
    }
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Taking on Job for %s\n", sign_jobs_head->label);
    job = sign_jobs_head;
    GNUNET_CONTAINER_DLL_remove (sign_jobs_head, sign_jobs_tail, job);
    GNUNET_assert (0 == pthread_mutex_unlock (&sign_jobs_lock));
    job->ec->seal (job->ec->cls,
                   job->label,
                   job->expire_pub,
                   job->rdata_pub,
                   job->rdata_pub_len,
                   &job->block_pub);
    if (job->rdata_pub != job->rdata_priv)
      job->ec->seal (job->ec->cls,
                     job->label,
                     job->expire_priv,
                     job->rdata_priv,
                     job->rdata_priv_len,
                     &job->block_priv);
    else
      job->block_priv = job->block_pub;
    GNUNET_assert (0 == pthread_mutex_lock (&sign_results_lock));
    GNUNET_CONTAINER_DLL_insert (sign_results_head, sign_results_tail, job);
    GNUNET_assert (0 == pthread_mutex_unlock (&sign_results_lock));
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Done, notifying main thread through pipe!\n");
    GNUNET_DISK_file_write (fh, "!", 1);
  }
  return NULL;
}

References GNUNET_assert, GNUNET_CONTAINER_DLL_insert, GNUNET_CONTAINER_DLL_remove, GNUNET_DISK_file_write(), GNUNET_DISK_PIPE_END_WRITE, GNUNET_DISK_pipe_handle(), GNUNET_ERROR_TYPE_DEBUG, GNUNET_log, GNUNET_YES, in_shutdown, job, RecordPublicationJob::label, notification_pipe, sign_jobs_cond, sign_jobs_head, sign_jobs_lock, sign_jobs_tail, sign_results_head, sign_results_lock, and sign_results_tail.

Referenced by run().

Here is the call graph for this function:

Here is the caller graph for this function:

run()

static void run ( void *  cls, const struct GNUNET_CONFIGURATION_Handle *  c, struct GNUNET_SERVICE_Handle *  service  )

static

Perform zonemaster duties: watch namestore, publish records.

Parameters

cls	closure
server	the initialized server
c	configuration to use

Start worker

Definition at line 1402 of file gnunet-service-zonemaster.c.

{
  unsigned long long max_parallel_bg_queries = 128;
  const struct GNUNET_DISK_FileHandle *np_fh;
 
  (void) cls;
  (void) service;
  pthread_mutex_init (&sign_jobs_lock, NULL);
  pthread_mutex_init (&sign_results_lock, NULL);
  pthread_cond_init (&sign_jobs_cond, NULL);
  last_put_100 = GNUNET_TIME_absolute_get ();  /* first time! */
  min_relative_record_time
    = GNUNET_TIME_UNIT_FOREVER_REL;
  target_iteration_velocity_per_record = INITIAL_ZONE_ITERATION_INTERVAL;
  namestore_handle = GNUNET_NAMESTORE_connect (c);
  if (NULL == namestore_handle)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                _ ("Failed to connect to the namestore!\n"));
    GNUNET_SCHEDULER_shutdown ();
    return;
  }
  disable_namecache = GNUNET_CONFIGURATION_get_value_yesno (c,
                                                            "namecache",
                                                            "DISABLE");
  if (GNUNET_NO == disable_namecache)
  {
    namecache = GNUNET_NAMECACHE_connect (c);
    if (NULL == namecache)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                  _ ("Failed to connect to the namecache!\n"));
      GNUNET_SCHEDULER_shutdown ();
      return;
    }
  }
  cache_keys = GNUNET_CONFIGURATION_get_value_yesno (c,
                                                     "namestore",
                                                     "CACHE_KEYS");
  zone_publish_time_window_default = GNUNET_DHT_DEFAULT_REPUBLISH_FREQUENCY;
  if (GNUNET_OK ==
      GNUNET_CONFIGURATION_get_value_time (c,
                                           "zonemaster",
                                           "ZONE_PUBLISH_TIME_WINDOW",
                                           &zone_publish_time_window_default))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Time window for zone iteration: %s\n",
                GNUNET_STRINGS_relative_time_to_string (
                  zone_publish_time_window_default,
                  GNUNET_YES));
  }
  zone_publish_time_window = zone_publish_time_window_default;
  if (GNUNET_OK ==
      GNUNET_CONFIGURATION_get_value_number (c,
                                             "zonemaster",
                                             "MAX_PARALLEL_BACKGROUND_QUERIES",
                                             &max_parallel_bg_queries))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Number of allowed parallel background queries: %llu\n",
                max_parallel_bg_queries);
  }
  if (0 == max_parallel_bg_queries)
    max_parallel_bg_queries = 1;
  dht_handle = GNUNET_DHT_connect (c,
                                   (unsigned int) max_parallel_bg_queries);
  if (NULL == dht_handle)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                _ ("Could not connect to DHT!\n"));
    GNUNET_SCHEDULER_add_now (&shutdown_task,
                              NULL);
    return;
  }
 
  /* Schedule periodic put for our records. */
  first_zone_iteration = GNUNET_YES;
  statistics = GNUNET_STATISTICS_create ("zonemaster",
                                         c);
  GNUNET_STATISTICS_set (statistics,
                         "Target zone iteration velocity (μs)",
                         target_iteration_velocity_per_record.rel_value_us,
                         GNUNET_NO);
  zone_publish_task = GNUNET_SCHEDULER_add_now (&publish_zone_dht_start,
                                                NULL);
  zmon = GNUNET_NAMESTORE_zone_monitor_start2 (c,
                                               NULL,
                                               GNUNET_NO,
                                               &handle_monitor_error,
                                               NULL,
                                               &handle_monitor_event,
                                               NULL,
                                               NULL /* sync_cb */,
                                               NULL,
                                               GNUNET_GNSRECORD_FILTER_NONE);
  GNUNET_NAMESTORE_zone_monitor_next (zmon,
                                      NAMESTORE_MONITOR_QUEUE_LIMIT - 1);
  GNUNET_break (NULL != zmon);
 
  GNUNET_SCHEDULER_add_shutdown (&shutdown_task,
                                 NULL);
 
  notification_pipe = GNUNET_DISK_pipe (GNUNET_DISK_PF_NONE);
  np_fh = GNUNET_DISK_pipe_handle (
    notification_pipe,
    GNUNET_DISK_PIPE_END_READ);
  pipe_read_task = GNUNET_SCHEDULER_add_read_file (GNUNET_TIME_UNIT_FOREVER_REL,
                                                   np_fh,
                                                   notification_pipe_cb, NULL);
 
  {
    long long unsigned int worker_count = 1;
    if (GNUNET_OK !=
        GNUNET_CONFIGURATION_get_value_number (c,
                                               "zonemaster",
                                               "WORKER_COUNT",
                                               &worker_count))
    {
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Number of workers not defined falling back to 1\n");
    }
    worker = GNUNET_malloc (sizeof (pthread_t) * worker_count);
    for (int i = 0; i < worker_count; i++)
    {
      if (0 !=
          pthread_create (&worker[i],
                          NULL,
                          &sign_worker,
                          NULL))
      {
        GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING,
                             "pthread_create");
        GNUNET_SCHEDULER_shutdown ();
      }
      GNUNET_STATISTICS_update (statistics,
                                "Workers running",
                                1,
                                GNUNET_NO);
    }
  }
}

References _, cache_keys, dht_handle, disable_namecache, first_zone_iteration, GNUNET_break, GNUNET_CONFIGURATION_get_value_number(), GNUNET_CONFIGURATION_get_value_time(), GNUNET_CONFIGURATION_get_value_yesno(), GNUNET_DHT_connect(), GNUNET_DHT_DEFAULT_REPUBLISH_FREQUENCY, GNUNET_DISK_PF_NONE, GNUNET_DISK_pipe(), GNUNET_DISK_PIPE_END_READ, GNUNET_DISK_pipe_handle(), GNUNET_ERROR_TYPE_DEBUG, GNUNET_ERROR_TYPE_ERROR, GNUNET_ERROR_TYPE_WARNING, GNUNET_GNSRECORD_FILTER_NONE, GNUNET_log, GNUNET_log_strerror, GNUNET_malloc, GNUNET_NAMECACHE_connect(), GNUNET_NAMESTORE_connect(), GNUNET_NAMESTORE_zone_monitor_next(), GNUNET_NAMESTORE_zone_monitor_start2(), GNUNET_NO, GNUNET_OK, GNUNET_SCHEDULER_add_now(), GNUNET_SCHEDULER_add_read_file(), GNUNET_SCHEDULER_add_shutdown(), GNUNET_SCHEDULER_shutdown(), GNUNET_STATISTICS_create(), GNUNET_STATISTICS_set(), GNUNET_STATISTICS_update(), GNUNET_STRINGS_relative_time_to_string(), GNUNET_TIME_absolute_get(), GNUNET_TIME_UNIT_FOREVER_REL, GNUNET_YES, handle_monitor_error(), handle_monitor_event(), INITIAL_ZONE_ITERATION_INTERVAL, last_put_100, min_relative_record_time, namecache, namestore_handle, NAMESTORE_MONITOR_QUEUE_LIMIT, notification_pipe, notification_pipe_cb(), pipe_read_task, publish_zone_dht_start(), GNUNET_TIME_Relative::rel_value_us, service, shutdown_task, sign_jobs_cond, sign_jobs_lock, sign_results_lock, sign_worker(), statistics, target_iteration_velocity_per_record, worker, zmon, zone_publish_task, zone_publish_time_window, and zone_publish_time_window_default.

Here is the call graph for this function:

GNUNET_SERVICE_MAIN()

GNUNET_SERVICE_MAIN	(	GNUNET_OS_project_data_gnunet()	,
		"zonemaster"	,
		GNUNET_SERVICE_OPTION_NONE	,
		&	run,
		NULL	,
		NULL	,
		NULL	,
		GNUNET_MQ_handler_end()
	)

Define "main" method using service macro.

Variable Documentation

worker

pthread_t* worker

static

Our workers.

Definition at line 96 of file gnunet-service-zonemaster.c.

Referenced by run().

sign_jobs_lock

pthread_mutex_t sign_jobs_lock

static

Lock for the sign jobs queue.

Definition at line 101 of file gnunet-service-zonemaster.c.

Referenced by dispatch_job(), run(), shutdown_task(), and sign_worker().

sign_results_lock

pthread_mutex_t sign_results_lock

static

Lock for the DHT put jobs queue.

Definition at line 106 of file gnunet-service-zonemaster.c.

Referenced by initiate_put_from_pipe_trigger(), run(), shutdown_task(), and sign_worker().

sign_jobs_cond

pthread_cond_t sign_jobs_cond

static

Wait condition on new sign jobs.

Definition at line 111 of file gnunet-service-zonemaster.c.

Referenced by dispatch_job(), run(), and sign_worker().

in_shutdown

int in_shutdown = GNUNET_NO

static

For threads to know we are shutting down.

Definition at line 116 of file gnunet-service-zonemaster.c.

Referenced by shutdown_task(), and sign_worker().

monitor_halted

int monitor_halted = GNUNET_NO

static

Monitor halted?

Definition at line 121 of file gnunet-service-zonemaster.c.

Referenced by cleanup_job(), and handle_monitor_event().

notification_pipe

struct GNUNET_DISK_PipeHandle* notification_pipe

static

Our notification pipe.

Definition at line 126 of file gnunet-service-zonemaster.c.

Referenced by initiate_put_from_pipe_trigger(), run(), shutdown_task(), and sign_worker().

pipe_read_task

struct GNUNET_SCHEDULER_Task* pipe_read_task

static

Pipe read task.

Definition at line 131 of file gnunet-service-zonemaster.c.

Referenced by initiate_put_from_pipe_trigger(), run(), and shutdown_task().

sign_jobs_head

struct RecordPublicationJob* sign_jobs_head

static

The DLL for workers to retrieve open jobs that require signing of blocks.

Definition at line 223 of file gnunet-service-zonemaster.c.

Referenced by dispatch_job(), shutdown_task(), and sign_worker().

sign_jobs_tail

struct RecordPublicationJob* sign_jobs_tail

static

See above.

Definition at line 228 of file gnunet-service-zonemaster.c.

Referenced by dispatch_job(), shutdown_task(), and sign_worker().

sign_results_head

struct RecordPublicationJob* sign_results_head

static

The DLL for workers to place jobs that are signed.

Definition at line 233 of file gnunet-service-zonemaster.c.

Referenced by initiate_put_from_pipe_trigger(), shutdown_task(), and sign_worker().

sign_results_tail

struct RecordPublicationJob* sign_results_tail

static

See above.

Definition at line 238 of file gnunet-service-zonemaster.c.

Referenced by initiate_put_from_pipe_trigger(), shutdown_task(), and sign_worker().

dht_jobs_head

struct RecordPublicationJob* dht_jobs_head

static

The DLL for jobs currently in the process of being dispatched into the DHT.

Definition at line 244 of file gnunet-service-zonemaster.c.

Referenced by dht_put_continuation(), initiate_put_from_pipe_trigger(), and shutdown_task().

dht_jobs_tail

struct RecordPublicationJob* dht_jobs_tail

static

See above.

Definition at line 249 of file gnunet-service-zonemaster.c.

Referenced by dht_put_continuation(), initiate_put_from_pipe_trigger(), and shutdown_task().

statistics

struct GNUNET_STATISTICS_Handle* statistics

static

Handle to the statistics service.

Definition at line 278 of file gnunet-service-zonemaster.c.

Referenced by calculate_put_interval(), check_zone_namestore_next(), handle_monitor_error(), handle_monitor_event(), initiate_put_from_pipe_trigger(), publish_zone_dht_start(), refresh_block(), run(), shutdown_task(), update_velocity(), and zone_iteration_finished().

dht_handle

struct GNUNET_DHT_Handle* dht_handle

static

Our handle to the DHT.

Definition at line 283 of file gnunet-service-zonemaster.c.

Referenced by initiate_put_from_pipe_trigger(), run(), and shutdown_task().

namestore_handle

struct GNUNET_NAMESTORE_Handle* namestore_handle

static

Our handle to the namestore service.

Definition at line 288 of file gnunet-service-zonemaster.c.

Referenced by publish_zone_dht_start(), run(), and shutdown_task().

zmon

struct GNUNET_NAMESTORE_ZoneMonitor* zmon

static

Handle to monitor namestore changes to instant propagation.

Definition at line 293 of file gnunet-service-zonemaster.c.

Referenced by cleanup_job(), handle_monitor_event(), run(), and shutdown_task().

namecache

struct GNUNET_NAMECACHE_Handle* namecache

static

Our handle to the namecache service.

Definition at line 298 of file gnunet-service-zonemaster.c.

Referenced by refresh_block(), run(), and shutdown_task().

disable_namecache

int disable_namecache

static

Use the namecache? Doing so creates additional cryptographic operations whenever we touch a record.

Definition at line 304 of file gnunet-service-zonemaster.c.

Referenced by refresh_block(), and run().

namestore_iter

struct GNUNET_NAMESTORE_ZoneIterator* namestore_iter

static

Handle to iterate over our authoritative zone in namestore.

Definition at line 309 of file gnunet-service-zonemaster.c.

Referenced by handle_monitor_event(), publish_zone_dht_start(), publish_zone_namestore_next(), shutdown_task(), zone_iteration_error(), and zone_iteration_finished().

job_queue_length

unsigned int job_queue_length

static

Number of entries in the job queue #jobs_head.

Definition at line 314 of file gnunet-service-zonemaster.c.

Referenced by check_zone_namestore_next(), cleanup_job(), dht_put_continuation(), handle_monitor_event(), handle_record(), shutdown_task(), and update_velocity().

num_public_records

unsigned long long num_public_records

static

Useful for zone update for DHT put.

Definition at line 319 of file gnunet-service-zonemaster.c.

Referenced by calculate_put_interval(), dispatch_job(), publish_zone_dht_start(), update_velocity(), and zone_iteration_finished().

last_num_public_records

unsigned long long last_num_public_records

static

Last seen record count.

Definition at line 324 of file gnunet-service-zonemaster.c.

Referenced by calculate_put_interval(), check_zone_namestore_next(), update_velocity(), and zone_iteration_finished().

put_cnt

unsigned long long put_cnt

static

Number of successful put operations performed in the current measurement cycle (as measured in check_zone_namestore_next()).

Definition at line 330 of file gnunet-service-zonemaster.c.

Referenced by check_zone_namestore_next(), and handle_record().

target_iteration_velocity_per_record

struct GNUNET_TIME_Relative target_iteration_velocity_per_record

static

What is the frequency at which we currently would like to perform DHT puts (per record)? Calculated in update_velocity() from the zone_publish_time_window() and the total number of record sets we have (so far) observed in the zone.

Definition at line 339 of file gnunet-service-zonemaster.c.

Referenced by calculate_put_interval(), check_zone_namestore_next(), handle_monitor_event(), run(), update_velocity(), and zone_iteration_finished().

min_relative_record_time

struct GNUNET_TIME_Relative min_relative_record_time

static

Minimum relative expiration time of records seem during the current zone iteration.

Definition at line 345 of file gnunet-service-zonemaster.c.

Referenced by calculate_put_interval(), handle_monitor_event(), note_relative_expiration(), run(), and zone_iteration_finished().

last_min_relative_record_time

struct GNUNET_TIME_Relative last_min_relative_record_time

static

Minimum relative expiration time of records seem during the last zone iteration.

Definition at line 351 of file gnunet-service-zonemaster.c.

Referenced by calculate_put_interval(), handle_monitor_event(), and zone_iteration_finished().

zone_publish_time_window_default

struct GNUNET_TIME_Relative zone_publish_time_window_default

static

Default time window for zone iteration.

Definition at line 356 of file gnunet-service-zonemaster.c.

Referenced by calculate_put_interval(), and run().

zone_publish_time_window

struct GNUNET_TIME_Relative zone_publish_time_window

static

Time window for zone iteration, adjusted based on relative record expiration times in our zone.

Definition at line 362 of file gnunet-service-zonemaster.c.

Referenced by calculate_put_interval(), check_zone_namestore_next(), and run().

last_put_100

struct GNUNET_TIME_Absolute last_put_100

static

When did we last start measuring the DELTA_INTERVAL successful DHT puts? Used for velocity calculations.

Definition at line 368 of file gnunet-service-zonemaster.c.

Referenced by run(), and update_velocity().

sub_delta

struct GNUNET_TIME_Relative sub_delta

static

By how much should we try to increase our per-record iteration speed (over the desired speed calculated directly from the #put_interval)? Basically this value corresponds to the per-record CPU time overhead we have.

Definition at line 376 of file gnunet-service-zonemaster.c.

Referenced by check_zone_namestore_next(), and update_velocity().

zone_publish_task

struct GNUNET_SCHEDULER_Task* zone_publish_task

static

zone publish task

Definition at line 381 of file gnunet-service-zonemaster.c.

Referenced by check_zone_namestore_next(), cleanup_job(), handle_monitor_event(), publish_zone_dht_start(), publish_zone_namestore_next(), run(), shutdown_task(), zone_iteration_error(), and zone_iteration_finished().

ns_iteration_left

unsigned int ns_iteration_left

static

How many more values are left for the current query before we need to explicitly ask the namestore for more?

Definition at line 387 of file gnunet-service-zonemaster.c.

Referenced by check_zone_namestore_next(), handle_record(), publish_zone_dht_start(), and publish_zone_namestore_next().

first_zone_iteration

int first_zone_iteration

static

GNUNET_YES if zone has never been published before

Definition at line 392 of file gnunet-service-zonemaster.c.

Referenced by check_zone_namestore_next(), run(), update_velocity(), and zone_iteration_finished().

cache_keys

int cache_keys

static

Optimize block insertion by caching map of private keys to public keys in memory?

Definition at line 398 of file gnunet-service-zonemaster.c.

Referenced by run().

cop_head

struct CacheOperation* cop_head

static

Head of cop DLL.

Definition at line 403 of file gnunet-service-zonemaster.c.

Referenced by finish_cache_operation(), refresh_block(), and shutdown_task().

cop_tail

struct CacheOperation* cop_tail

static

Tail of cop DLL.

Definition at line 408 of file gnunet-service-zonemaster.c.

Referenced by finish_cache_operation(), refresh_block(), and shutdown_task().