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
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 3, or (at your
   8      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      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
  18      Boston, MA 02110-1301, USA.
  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  * @param success GNUNET_OK if the PUT was transmitted,
 139  *                GNUNET_NO on timeout,
 140  *                GNUNET_SYSERR on disconnect from service
 141  *                after the PUT message was transmitted
 142  *                (so we don't know if it was received or not)
 143  */
 144 static void
 145 delay_dht_put_blocks (void *cls, int success)
 146 {
 147   struct PutOperator *po = cls;
 148
 149   po->dht_put = NULL;
 150   schedule_next_put (po);
 151 }
 152
 153
 154 /**
 155  * Task that is run periodically to obtain blocks for DHT PUTs.
 156  *
 157  * @param cls type of blocks to gather
 158  */
 159 static void
 160 delay_dht_put_task (void *cls)
 161 {
 162   struct PutOperator *po = cls;
 163
 164   po->dht_task = NULL;
 165   schedule_next_put (po);
 166 }
 167
 168
 169 /**
 170  * Store content in DHT.
 171  *
 172  * @param cls closure
 173  * @param key key for the content
 174  * @param size number of bytes in data
 175  * @param data content stored
 176  * @param type type of the content
 177  * @param priority priority of the content
 178  * @param anonymity anonymity-level for the content
 179  * @param expiration expiration time for the content
 180  * @param uid unique identifier for the datum;
 181  *        maybe 0 if no unique identifier is available
 182  */
 183 static void
 184 process_dht_put_content (void *cls,
 185                          const struct GNUNET_HashCode * key,
 186                          size_t size,
 187                          const void *data,
 188                          enum GNUNET_BLOCK_Type type,
 189                          uint32_t priority,
 190                          uint32_t anonymity,
 191                          struct GNUNET_TIME_Absolute expiration,
 192                          uint64_t uid)
 193 {
 194   struct PutOperator *po = cls;
 195
 196   po->dht_qe = NULL;
 197   if (key == NULL)
 198   {
 199     po->zero_anonymity_count_estimate = po->result_count;
 200     po->result_count = 0;
 201     po->next_uid = 0;
 202     po->dht_task = GNUNET_SCHEDULER_add_now (&delay_dht_put_task, po);
 203     return;
 204   }
 205   po->result_count++;
 206   po->next_uid = uid + 1;
 207   po->zero_anonymity_count_estimate =
 208     GNUNET_MAX (po->result_count, po->zero_anonymity_count_estimate);
 209   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 210               "Retrieved block `%s' of type %u for DHT PUT\n", GNUNET_h2s (key),
 211               type);
 212   po->dht_put = GNUNET_DHT_put (GSF_dht,
 213                                 key,
 214                                 DEFAULT_PUT_REPLICATION,
 215                                 GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,
 216                                 type,
 217                                 size,
 218                                 data,
 219                                 expiration,
 220                                 &delay_dht_put_blocks,
 221                                 po);
 222 }
 223
 224
 225 /**
 226  * Task that is run periodically to obtain blocks for DHT PUTs.
 227  *
 228  * @param cls type of blocks to gather
 229  */
 230 static void
 231 gather_dht_put_blocks (void *cls)
 232 {
 233   struct PutOperator *po = cls;
 234
 235   po->dht_task = NULL;
 236   po->dht_qe =
 237       GNUNET_DATASTORE_get_zero_anonymity (GSF_dsh,
 238                                            po->next_uid,
 239                                            0,
 240                                            UINT_MAX,
 241                                            po->dht_put_type,
 242                                            &process_dht_put_content,
 243                                            po);
 244   if (NULL == po->dht_qe)
 245     po->dht_task = GNUNET_SCHEDULER_add_now (&delay_dht_put_task, po);
 246 }
 247
 248
 249 /**
 250  * Setup the module.
 251  */
 252 void
 253 GSF_put_init_ ()
 254 {
 255   unsigned int i;
 256
 257   i = 0;
 258   while (operators[i].dht_put_type != GNUNET_BLOCK_TYPE_ANY)
 259   {
 260     operators[i].dht_task =
 261         GNUNET_SCHEDULER_add_now (&gather_dht_put_blocks, &operators[i]);
 262     i++;
 263   }
 264 }
 265
 266
 267 /**
 268  * Shutdown the module.
 269  */
 270 void
 271 GSF_put_done_ ()
 272 {
 273   struct PutOperator *po;
 274   unsigned int i;
 275
 276   i = 0;
 277   while ((po = &operators[i])->dht_put_type != GNUNET_BLOCK_TYPE_ANY)
 278   {
 279     if (NULL != po->dht_task)
 280     {
 281       GNUNET_SCHEDULER_cancel (po->dht_task);
 282       po->dht_task = NULL;
 283     }
 284     if (NULL != po->dht_put)
 285     {
 286       GNUNET_DHT_put_cancel (po->dht_put);
 287       po->dht_put = NULL;
 288     }
 289     if (NULL != po->dht_qe)
 290     {
 291       GNUNET_DATASTORE_cancel (po->dht_qe);
 292       po->dht_qe = NULL;
 293     }
 294     i++;
 295   }
 296 }
 297
 298 /* end of gnunet-service-fs_put.c */