gnunet-service-fs_push.c

Go to the documentation of this file.

/*
     This file is part of GNUnet.
     Copyright (C) 2011, 2016 GNUnet e.V.

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

     GNUnet is distributed in the hope that it will be useful, but
     WITHOUT ANY WARRANTY; without even the implied warranty of
     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
     Affero General Public License for more details.

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

     SPDX-License-Identifier: AGPL3.0-or-later
 */

#include "platform.h"
#include "gnunet-service-fs.h"
#include "gnunet-service-fs_cp.h"
#include "gnunet-service-fs_indexing.h"
#include "gnunet-service-fs_push.h"


#define MAX_MIGRATION_QUEUE 8

#define MIGRATION_LIST_SIZE 2

#define MIN_MIGRATION_CONTENT_LIFETIME GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_MINUTES, 30)


struct MigrationReadyBlock {
  struct MigrationReadyBlock *next;

  struct MigrationReadyBlock *prev;

  struct GNUNET_HashCode query;

  struct GNUNET_TIME_Absolute expiration;

  GNUNET_PEER_Id target_list[MIGRATION_LIST_SIZE];

  size_t size;

  unsigned int used_targets;

  enum GNUNET_BLOCK_Type type;
};


struct MigrationReadyPeer {
  struct MigrationReadyPeer *next;

  struct MigrationReadyPeer *prev;

  struct GSF_ConnectedPeer *peer;

  struct GNUNET_MQ_Envelope *env;
};


static struct MigrationReadyBlock *mig_head;

static struct MigrationReadyBlock *mig_tail;

static struct MigrationReadyPeer *peer_head;

static struct MigrationReadyPeer *peer_tail;

static struct GNUNET_DATASTORE_QueueEntry *mig_qe;

static struct GNUNET_SCHEDULER_Task *mig_task;

static struct GNUNET_TIME_Relative min_migration_delay;

static unsigned int mig_size;

static int enabled;

static int value_found;


static void
delete_migration_block(struct MigrationReadyBlock *mb)
{
  GNUNET_CONTAINER_DLL_remove(mig_head,
                              mig_tail,
                              mb);
  GNUNET_PEER_decrement_rcs(mb->target_list,
                            MIGRATION_LIST_SIZE);
  mig_size--;
  GNUNET_free(mb);
}


static void
find_content(void *cls);


static int
transmit_content(struct MigrationReadyPeer *mrp,
                 struct MigrationReadyBlock *block)
{
  struct PutMessage *msg;
  unsigned int i;
  struct GSF_PeerPerformanceData *ppd;
  int ret;

  ppd = GSF_get_peer_performance_data_(mrp->peer);
  GNUNET_assert(NULL == mrp->env);
  mrp->env = GNUNET_MQ_msg_extra(msg,
                                 block->size,
                                 GNUNET_MESSAGE_TYPE_FS_PUT);
  msg->type = htonl(block->type);
  msg->expiration = GNUNET_TIME_absolute_hton(block->expiration);
  GNUNET_memcpy(&msg[1],
                &block[1],
                block->size);
  for (i = 0; i < MIGRATION_LIST_SIZE; i++)
    {
      if (block->target_list[i] == 0)
        {
          block->target_list[i] = ppd->pid;
          GNUNET_PEER_change_rc(block->target_list[i],
                                1);
          break;
        }
    }
  if (MIGRATION_LIST_SIZE == i)
    {
      delete_migration_block(block);
      ret = GNUNET_YES;
    }
  else
    {
      ret = GNUNET_NO;
    }
  GNUNET_MQ_notify_sent(mrp->env,
                        &find_content,
                        mrp);
  GSF_peer_transmit_(mrp->peer,
                     GNUNET_NO,
                     0 /* priority */,
                     mrp->env);
  return ret;
}


static unsigned int
count_targets(struct MigrationReadyBlock *block)
{
  unsigned int i;

  for (i = 0; i < MIGRATION_LIST_SIZE; i++)
    if (block->target_list[i] == 0)
      return i;
  return i;
}


static long
score_content(struct MigrationReadyPeer *mrp,
              struct MigrationReadyBlock *block)
{
  unsigned int i;
  struct GSF_PeerPerformanceData *ppd;
  struct GNUNET_PeerIdentity id;
  struct GNUNET_HashCode hc;
  uint32_t dist;

  ppd = GSF_get_peer_performance_data_(mrp->peer);
  for (i = 0; i < MIGRATION_LIST_SIZE; i++)
    if (block->target_list[i] == ppd->pid)
      return -1;
  GNUNET_assert(0 != ppd->pid);
  GNUNET_PEER_resolve(ppd->pid,
                      &id);
  GNUNET_CRYPTO_hash(&id,
                     sizeof(struct GNUNET_PeerIdentity),
                     &hc);
  dist = GNUNET_CRYPTO_hash_distance_u32(&block->query,
                                         &hc);
  /* closer distance, higher score: */
  return UINT32_MAX - dist;
}


static void
consider_gathering(void);


static void
find_content(void *cls)
{
  struct MigrationReadyPeer *mrp = cls;
  struct MigrationReadyBlock *pos;
  long score;
  long best_score;
  struct MigrationReadyBlock *best;

  mrp->env = NULL;
  best = NULL;
  best_score = -1;
  pos = mig_head;
  while (NULL != pos)
    {
      score = score_content(mrp, pos);
      if (score > best_score)
        {
          best_score = score;
          best = pos;
        }
      pos = pos->next;
    }
  if (NULL == best)
    {
      if (mig_size < MAX_MIGRATION_QUEUE)
        {
          GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
                     "No content found for pushing, waiting for queue to fill\n");
          return;               /* will fill up eventually... */
        }
      GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
                 "No suitable content found, purging content from full queue\n");
      /* failed to find migration target AND
       * queue is full, purge most-forwarded
       * block from queue to make room for more */
      pos = mig_head;
      while (NULL != pos)
        {
          score = count_targets(pos);
          if (score >= best_score)
            {
              best_score = score;
              best = pos;
            }
          pos = pos->next;
        }
      GNUNET_assert(NULL != best);
      delete_migration_block(best);
      consider_gathering();
      return;
    }
  GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
             "Preparing to push best content to peer\n");
  transmit_content(mrp,
                   best);
}


static void
gather_migration_blocks(void *cls);


static void
consider_gathering()
{
  struct GNUNET_TIME_Relative delay;

  if (NULL == GSF_dsh)
    return;
  if (NULL != mig_qe)
    return;
  if (NULL != mig_task)
    return;
  if (mig_size >= MAX_MIGRATION_QUEUE)
    return;
  delay = GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_SECONDS,
                                        mig_size);
  delay = GNUNET_TIME_relative_divide(delay,
                                      MAX_MIGRATION_QUEUE);
  delay = GNUNET_TIME_relative_max(delay,
                                   min_migration_delay);
  if (GNUNET_NO == value_found)
    {
      /* wait at least 5s if the datastore is empty */
      delay = GNUNET_TIME_relative_max(delay,
                                       GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_SECONDS,
                                                                     5));
    }
  GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
             "Scheduling gathering task (queue size: %u)\n",
             mig_size);
  mig_task = GNUNET_SCHEDULER_add_delayed(delay,
                                          &gather_migration_blocks,
                                          NULL);
}


static void
process_migration_content(void *cls,
                          const struct GNUNET_HashCode *key,
                          size_t size,
                          const void *data,
                          enum GNUNET_BLOCK_Type type,
                          uint32_t priority,
                          uint32_t anonymity,
                          uint32_t replication,
                          struct GNUNET_TIME_Absolute expiration,
                          uint64_t uid)
{
  struct MigrationReadyBlock *mb;
  struct MigrationReadyPeer *pos;

  mig_qe = NULL;
  if (NULL == key)
    {
      GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
                 "No content found for migration...\n");
      consider_gathering();
      return;
    }
  value_found = GNUNET_YES;
  if (GNUNET_TIME_absolute_get_remaining(expiration).rel_value_us <
      MIN_MIGRATION_CONTENT_LIFETIME.rel_value_us)
    {
      /* content will expire soon, don't bother */
      consider_gathering();
      return;
    }
  if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
    {
      if (GNUNET_OK !=
          GNUNET_FS_handle_on_demand_block(key,
                                           size,
                                           data,
                                           type,
                                           priority,
                                           anonymity,
                                           replication,
                                           expiration,
                                           uid,
                                           &process_migration_content,
                                           NULL))
        consider_gathering();
      return;
    }
  GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
             "Retrieved block `%s' of type %u for migration (queue size: %u/%u)\n",
             GNUNET_h2s(key),
             type, mig_size + 1,
             MAX_MIGRATION_QUEUE);
  mb = GNUNET_malloc(sizeof(struct MigrationReadyBlock) + size);
  mb->query = *key;
  mb->expiration = expiration;
  mb->size = size;
  mb->type = type;
  GNUNET_memcpy(&mb[1], data, size);
  GNUNET_CONTAINER_DLL_insert_after(mig_head,
                                    mig_tail,
                                    mig_tail,
                                    mb);
  mig_size++;
  for (pos = peer_head; NULL != pos; pos = pos->next)
    {
      GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
                 "Preparing to push best content to peer %s\n",
                 GNUNET_i2s(GSF_connected_peer_get_identity2_(pos->peer)));
      if ((NULL == pos->env) &&
          (GNUNET_YES == transmit_content(pos,
                                          mb)))
        {
          break;              /* 'mb' was freed! */
        }
    }
  consider_gathering();
}


static void
gather_migration_blocks(void *cls)
{
  mig_task = NULL;
  if (mig_size >= MAX_MIGRATION_QUEUE)
    return;
  if (NULL == GSF_dsh)
    return;
  GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
             "Asking datastore for content for replication (queue size: %u)\n",
             mig_size);
  value_found = GNUNET_NO;
  mig_qe = GNUNET_DATASTORE_get_for_replication(GSF_dsh,
                                                0,
                                                UINT_MAX,
                                                &process_migration_content,
                                                NULL);
  if (NULL == mig_qe)
    consider_gathering();
}


void
GSF_push_start_(struct GSF_ConnectedPeer *peer)
{
  struct MigrationReadyPeer *mrp;

  if (GNUNET_YES != enabled)
    return;
  for (mrp = peer_head; NULL != mrp; mrp = mrp->next)
    if (mrp->peer == peer)
      break;
  if (NULL != mrp)
    {
      /* same peer added twice, must not happen */
      GNUNET_break(0);
      return;
    }

  GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
             "Adding peer %s to list for pushing\n",
             GNUNET_i2s(GSF_connected_peer_get_identity2_(peer)));

  mrp = GNUNET_new(struct MigrationReadyPeer);
  mrp->peer = peer;
  find_content(mrp);
  GNUNET_CONTAINER_DLL_insert(peer_head,
                              peer_tail,
                              mrp);
}


void
GSF_push_stop_(struct GSF_ConnectedPeer *peer)
{
  struct MigrationReadyPeer *pos;

  for (pos = peer_head; NULL != pos; pos = pos->next)
    if (pos->peer == peer)
      break;
  if (NULL == pos)
    return;
  if (NULL != pos->env)
    GNUNET_MQ_send_cancel(pos->env);
  GNUNET_CONTAINER_DLL_remove(peer_head,
                              peer_tail,
                              pos);
  GNUNET_free(pos);
}


void
GSF_push_init_()
{
  enabled =
    GNUNET_CONFIGURATION_get_value_yesno(GSF_cfg,
                                         "FS",
                                         "CONTENT_PUSHING");
  if (GNUNET_YES != enabled)
    return;

  if (GNUNET_OK !=
      GNUNET_CONFIGURATION_get_value_time(GSF_cfg,
                                          "fs",
                                          "MIN_MIGRATION_DELAY",
                                          &min_migration_delay))
    {
      GNUNET_log_config_invalid(GNUNET_ERROR_TYPE_WARNING,
                                "fs",
                                "MIN_MIGRATION_DELAY",
                                _("time required, content pushing disabled"));
      return;
    }
  consider_gathering();
}


void
GSF_push_done_()
{
  if (NULL != mig_task)
    {
      GNUNET_SCHEDULER_cancel(mig_task);
      mig_task = NULL;
    }
  if (NULL != mig_qe)
    {
      GNUNET_DATASTORE_cancel(mig_qe);
      mig_qe = NULL;
    }
  while (NULL != mig_head)
    delete_migration_block(mig_head);
  GNUNET_assert(0 == mig_size);
}

/* end of gnunet-service-fs_push.c */