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 replication replication-level for the content
 180  * @param expiration expiration time for the content
 181  * @param uid unique identifier for the datum;
 182  *        maybe 0 if no unique identifier is available
 183  */
 184 static void
 185 process_dht_put_content (void *cls,
 186                          const struct GNUNET_HashCode * key,
 187                          size_t size,
 188                          const void *data,
 189                          enum GNUNET_BLOCK_Type type,
 190                          uint32_t priority,
 191                          uint32_t anonymity,
 192                          uint32_t replication,
 193                          struct GNUNET_TIME_Absolute expiration,
 194                          uint64_t uid)
 195 {
 196   struct PutOperator *po = cls;
 197
 198   po->dht_qe = NULL;
 199   if (key == NULL)
 200   {
 201     po->zero_anonymity_count_estimate = po->result_count;
 202     po->result_count = 0;
 203     po->next_uid = 0;
 204     po->dht_task = GNUNET_SCHEDULER_add_now (&delay_dht_put_task, po);
 205     return;
 206   }
 207   po->result_count++;
 208   po->next_uid = uid + 1;
 209   po->zero_anonymity_count_estimate =
 210     GNUNET_MAX (po->result_count, po->zero_anonymity_count_estimate);
 211   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 212               "Retrieved block `%s' of type %u for DHT PUT\n", GNUNET_h2s (key),
 213               type);
 214   po->dht_put = GNUNET_DHT_put (GSF_dht,
 215                                 key,
 216                                 DEFAULT_PUT_REPLICATION,
 217                                 GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,
 218                                 type,
 219                                 size,
 220                                 data,
 221                                 expiration,
 222                                 &delay_dht_put_blocks,
 223                                 po);
 224 }
 225
 226
 227 /**
 228  * Task that is run periodically to obtain blocks for DHT PUTs.
 229  *
 230  * @param cls type of blocks to gather
 231  */
 232 static void
 233 gather_dht_put_blocks (void *cls)
 234 {
 235   struct PutOperator *po = cls;
 236
 237   po->dht_task = NULL;
 238   po->dht_qe =
 239       GNUNET_DATASTORE_get_zero_anonymity (GSF_dsh,
 240                                            po->next_uid,
 241                                            0,
 242                                            UINT_MAX,
 243                                            po->dht_put_type,
 244                                            &process_dht_put_content,
 245                                            po);
 246   if (NULL == po->dht_qe)
 247     po->dht_task = GNUNET_SCHEDULER_add_now (&delay_dht_put_task, po);
 248 }
 249
 250
 251 /**
 252  * Setup the module.
 253  */
 254 void
 255 GSF_put_init_ ()
 256 {
 257   unsigned int i;
 258
 259   i = 0;
 260   while (operators[i].dht_put_type != GNUNET_BLOCK_TYPE_ANY)
 261   {
 262     operators[i].dht_task =
 263         GNUNET_SCHEDULER_add_now (&gather_dht_put_blocks, &operators[i]);
 264     i++;
 265   }
 266 }
 267
 268
 269 /**
 270  * Shutdown the module.
 271  */
 272 void
 273 GSF_put_done_ ()
 274 {
 275   struct PutOperator *po;
 276   unsigned int i;
 277
 278   i = 0;
 279   while ((po = &operators[i])->dht_put_type != GNUNET_BLOCK_TYPE_ANY)
 280   {
 281     if (NULL != po->dht_task)
 282     {
 283       GNUNET_SCHEDULER_cancel (po->dht_task);
 284       po->dht_task = NULL;
 285     }
 286     if (NULL != po->dht_put)
 287     {
 288       GNUNET_DHT_put_cancel (po->dht_put);
 289       po->dht_put = NULL;
 290     }
 291     if (NULL != po->dht_qe)
 292     {
 293       GNUNET_DATASTORE_cancel (po->dht_qe);
 294       po->dht_qe = NULL;
 295     }
 296     i++;
 297   }
 298 }
 299
 300 /* end of gnunet-service-fs_put.c */