src/fs/gnunet-service-fs_put.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2011 GNUnet e.V.
   4
   5      GNUnet is free software: you can redistribute it and/or modify it
   6      under the terms of the GNU Affero General Public License as published
   7      by the Free Software Foundation, either version 3 of the License,
   8      or (at your option) any later version.
   9
  10      GNUnet is distributed in the hope that it will be useful, but
  11      WITHOUT ANY WARRANTY; without even the implied warranty of
  12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13      Affero General Public License for more details.
  14 */
  15
  16 /**
  17  * @file fs/gnunet-service-fs_put.c
  18  * @brief API to PUT zero-anonymity index data from our datastore into the DHT
  19  * @author Christian Grothoff
  20  */
  21 #include "platform.h"
  22 #include "gnunet-service-fs.h"
  23 #include "gnunet-service-fs_put.h"
  24
  25
  26 /**
  27  * How often do we at most PUT content into the DHT?
  28  */
  29 #define MAX_DHT_PUT_FREQ GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5)
  30
  31 /**
  32  * How many replicas do we try to create per PUT?
  33  */
  34 #define DEFAULT_PUT_REPLICATION 5
  35
  36
  37 /**
  38  * Context for each zero-anonymity iterator.
  39  */
  40 struct PutOperator
  41 {
  42
  43   /**
  44    * Request to datastore for DHT PUTs (or NULL).
  45    */
  46   struct GNUNET_DATASTORE_QueueEntry *dht_qe;
  47
  48   /**
  49    * Type we request from the datastore.
  50    */
  51   enum GNUNET_BLOCK_Type dht_put_type;
  52
  53   /**
  54    * Handle to PUT operation.
  55    */
  56   struct GNUNET_DHT_PutHandle *dht_put;
  57
  58   /**
  59    * ID of task that collects blocks for DHT PUTs.
  60    */
  61   struct GNUNET_SCHEDULER_Task * dht_task;
  62
  63   /**
  64    * How many entires with zero anonymity of our type do we currently
  65    * estimate to have in the database?
  66    */
  67   uint64_t zero_anonymity_count_estimate;
  68
  69   /**
  70    * Count of results received from the database.
  71    */
  72   uint64_t result_count;
  73
  74   /**
  75    * Next UID to request when iterating the database.
  76    */
  77   uint64_t next_uid;
  78 };
  79
  80
  81 /**
  82  * ANY-terminated list of our operators (one per type
  83  * of block that we're putting into the DHT).
  84  */
  85 static struct PutOperator operators[] = {
  86   {NULL, GNUNET_BLOCK_TYPE_FS_UBLOCK, 0, 0, 0},
  87   {NULL, GNUNET_BLOCK_TYPE_ANY, 0, 0, 0}
  88 };
  89
  90
  91 /**
  92  * Task that is run periodically to obtain blocks for DHT PUTs.
  93  *
  94  * @param cls type of blocks to gather
  95  * @param tc scheduler context (unused)
  96  */
  97 static void
  98 gather_dht_put_blocks (void *cls);
  99
 100
 101 /**
 102  * Calculate when to run the next PUT operation and schedule it.
 103  *
 104  * @param po put operator to schedule
 105  */
 106 static void
 107 schedule_next_put (struct PutOperator *po)
 108 {
 109   struct GNUNET_TIME_Relative delay;
 110
 111   if (po->zero_anonymity_count_estimate > 0)
 112   {
 113     delay =
 114         GNUNET_TIME_relative_divide (GNUNET_DHT_DEFAULT_REPUBLISH_FREQUENCY,
 115                                      po->zero_anonymity_count_estimate);
 116     delay = GNUNET_TIME_relative_min (delay, MAX_DHT_PUT_FREQ);
 117   }
 118   else
 119   {
 120     /* if we have NO zero-anonymity content yet, wait 5 minutes for some to
 121      * (hopefully) appear */
 122     delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 5);
 123   }
 124   po->dht_task =
 125       GNUNET_SCHEDULER_add_delayed (delay, &gather_dht_put_blocks, po);
 126 }
 127
 128
 129 /**
 130  * Continuation called after DHT PUT operation has finished.
 131  *
 132  * @param cls type of blocks to gather
 133  */
 134 static void
 135 delay_dht_put_blocks (void *cls)
 136 {
 137   struct PutOperator *po = cls;
 138
 139   po->dht_put = NULL;
 140   schedule_next_put (po);
 141 }
 142
 143
 144 /**
 145  * Task that is run periodically to obtain blocks for DHT PUTs.
 146  *
 147  * @param cls type of blocks to gather
 148  */
 149 static void
 150 delay_dht_put_task (void *cls)
 151 {
 152   struct PutOperator *po = cls;
 153
 154   po->dht_task = NULL;
 155   schedule_next_put (po);
 156 }
 157
 158
 159 /**
 160  * Store content in DHT.
 161  *
 162  * @param cls closure
 163  * @param key key for the content
 164  * @param size number of bytes in data
 165  * @param data content stored
 166  * @param type type of the content
 167  * @param priority priority of the content
 168  * @param anonymity anonymity-level for the content
 169  * @param replication replication-level for the content
 170  * @param expiration expiration time for the content
 171  * @param uid unique identifier for the datum;
 172  *        maybe 0 if no unique identifier is available
 173  */
 174 static void
 175 process_dht_put_content (void *cls,
 176                          const struct GNUNET_HashCode * key,
 177                          size_t size,
 178                          const void *data,
 179                          enum GNUNET_BLOCK_Type type,
 180                          uint32_t priority,
 181                          uint32_t anonymity,
 182                          uint32_t replication,
 183                          struct GNUNET_TIME_Absolute expiration,
 184                          uint64_t uid)
 185 {
 186   struct PutOperator *po = cls;
 187
 188   po->dht_qe = NULL;
 189   if (key == NULL)
 190   {
 191     po->zero_anonymity_count_estimate = po->result_count;
 192     po->result_count = 0;
 193     po->next_uid = 0;
 194     po->dht_task = GNUNET_SCHEDULER_add_now (&delay_dht_put_task, po);
 195     return;
 196   }
 197   po->result_count++;
 198   po->next_uid = uid + 1;
 199   po->zero_anonymity_count_estimate =
 200     GNUNET_MAX (po->result_count, po->zero_anonymity_count_estimate);
 201   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 202               "Retrieved block `%s' of type %u for DHT PUT\n", GNUNET_h2s (key),
 203               type);
 204   po->dht_put = GNUNET_DHT_put (GSF_dht,
 205                                 key,
 206                                 DEFAULT_PUT_REPLICATION,
 207                                 GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,
 208                                 type,
 209                                 size,
 210                                 data,
 211                                 expiration,
 212                                 &delay_dht_put_blocks,
 213                                 po);
 214 }
 215
 216
 217 /**
 218  * Task that is run periodically to obtain blocks for DHT PUTs.
 219  *
 220  * @param cls type of blocks to gather
 221  */
 222 static void
 223 gather_dht_put_blocks (void *cls)
 224 {
 225   struct PutOperator *po = cls;
 226
 227   po->dht_task = NULL;
 228   po->dht_qe =
 229       GNUNET_DATASTORE_get_zero_anonymity (GSF_dsh,
 230                                            po->next_uid,
 231                                            0,
 232                                            UINT_MAX,
 233                                            po->dht_put_type,
 234                                            &process_dht_put_content,
 235                                            po);
 236   if (NULL == po->dht_qe)
 237     po->dht_task = GNUNET_SCHEDULER_add_now (&delay_dht_put_task, po);
 238 }
 239
 240
 241 /**
 242  * Setup the module.
 243  */
 244 void
 245 GSF_put_init_ ()
 246 {
 247   unsigned int i;
 248
 249   i = 0;
 250   while (operators[i].dht_put_type != GNUNET_BLOCK_TYPE_ANY)
 251   {
 252     operators[i].dht_task =
 253         GNUNET_SCHEDULER_add_now (&gather_dht_put_blocks, &operators[i]);
 254     i++;
 255   }
 256 }
 257
 258
 259 /**
 260  * Shutdown the module.
 261  */
 262 void
 263 GSF_put_done_ ()
 264 {
 265   struct PutOperator *po;
 266   unsigned int i;
 267
 268   i = 0;
 269   while ((po = &operators[i])->dht_put_type != GNUNET_BLOCK_TYPE_ANY)
 270   {
 271     if (NULL != po->dht_task)
 272     {
 273       GNUNET_SCHEDULER_cancel (po->dht_task);
 274       po->dht_task = NULL;
 275     }
 276     if (NULL != po->dht_put)
 277     {
 278       GNUNET_DHT_put_cancel (po->dht_put);
 279       po->dht_put = NULL;
 280     }
 281     if (NULL != po->dht_qe)
 282     {
 283       GNUNET_DATASTORE_cancel (po->dht_qe);
 284       po->dht_qe = NULL;
 285     }
 286     i++;
 287   }
 288 }
 289
 290 /* end of gnunet-service-fs_put.c */