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