src/dht/dht_api_get_put.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2009, 2010 Christian Grothoff (and other contributing authors)
   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., 59 Temple Place - Suite 330,
  18      Boston, MA 02111-1307, USA.
  19 */
  20
  21 /**
  22  * @file dht/dht_api_get_put.c
  23  * @brief library to perform DHT gets and puts
  24  * @author Christian Grothoff
  25  * @author Nathan Evans
  26  */
  27
  28 #include "platform.h"
  29 #include "gnunet_constants.h"
  30 #include "gnunet_arm_service.h"
  31 #include "gnunet_protocols.h"
  32 #include "gnunet_util_lib.h"
  33 #include "gnunet_dht_service.h"
  34 #include "dht.h"
  35
  36
  37 /**
  38  * Perform a PUT operation storing data in the DHT.
  39  *
  40  * @param handle handle to DHT service
  41  * @param key the key to store under
  42  * @param desired_replication_level estimate of how many
  43  *                nearest peers this request should reach
  44  * @param options routing options for this message
  45  * @param type type of the value
  46  * @param size number of bytes in data; must be less than 64k
  47  * @param data the data to store
  48  * @param exp desired expiration time for the value
  49  * @param timeout how long to wait for transmission of this request
  50  * @param cont continuation to call when done (transmitting request to service)
  51  * @param cont_cls closure for cont
  52  * @return GNUNET_YES if put message is queued for transmission
  53  */
  54 void
  55 GNUNET_DHT_put (struct GNUNET_DHT_Handle *handle,
  56                 const GNUNET_HashCode * key,
  57                 uint32_t desired_replication_level,
  58                 enum GNUNET_DHT_RouteOption options,
  59                 enum GNUNET_BLOCK_Type type,
  60                 size_t size,
  61                 const char *data,
  62                 struct GNUNET_TIME_Absolute exp,
  63                 struct GNUNET_TIME_Relative timeout,
  64                 GNUNET_SCHEDULER_Task cont,
  65                 void *cont_cls)
  66 {
  67   char buf[GNUNET_SERVER_MAX_MESSAGE_SIZE];
  68   struct GNUNET_DHT_PutMessage *put_msg;
  69
  70   if (size >= sizeof (buf) - sizeof (struct GNUNET_DHT_PutMessage))
  71     {
  72       GNUNET_break (0);
  73       return;
  74     }
  75   put_msg = (struct GNUNET_DHT_PutMessage*) buf;
  76   put_msg->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_PUT);
  77   put_msg->header.size = htons (sizeof (struct GNUNET_DHT_PutMessage) + size);
  78   put_msg->type = htonl ((uint32_t)type);
  79   put_msg->expiration = GNUNET_TIME_absolute_hton (exp);
  80   memcpy (&put_msg[1], data, size);
  81   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
  82               "Starting route for %u byte `%s' message of type %u \n",
  83               (unsigned int) (sizeof (struct GNUNET_DHT_PutMessage) + size),
  84               "PUT", type);
  85   GNUNET_break (NULL ==
  86                 GNUNET_DHT_route_start (handle,
  87                                         key,
  88                                         desired_replication_level, options,
  89                                         &put_msg->header,
  90                                         timeout,
  91                                         NULL, NULL,
  92                                         cont, cont_cls));
  93 }
  94
  95
  96
  97 /**
  98  * Handle to control a get operation.
  99  */
 100 struct GNUNET_DHT_GetHandle
 101 {
 102   /**
 103    * Handle to the actual route operation for the get
 104    */
 105   struct GNUNET_DHT_RouteHandle *route_handle;
 106
 107   /**
 108    * Iterator to call on data receipt
 109    */
 110   GNUNET_DHT_GetIterator iter;
 111
 112   /**
 113    * Closure for the iterator callback
 114    */
 115   void *iter_cls;
 116
 117 };
 118
 119
 120
 121 /**
 122  * Iterator called on each result obtained from a generic route
 123  * operation
 124  *
 125  * @param cls the 'struct GNUNET_DHT_GetHandle'
 126  * @param key key that was used
 127  * @param outgoing_path path of the message from this peer
 128  *                      to the target
 129  * @param reply response
 130  */
 131 static void
 132 get_reply_iterator (void *cls,
 133                     const GNUNET_HashCode *key,
 134                     const struct GNUNET_PeerIdentity * const *outgoing_path,
 135                     const struct GNUNET_MessageHeader *reply)
 136 {
 137   struct GNUNET_DHT_GetHandle *get_handle = cls;
 138   const struct GNUNET_DHT_GetResultMessage *result;
 139   const struct GNUNET_PeerIdentity **put_path;
 140   size_t payload;
 141   char *path_offset;
 142   const struct GNUNET_PeerIdentity *pos;
 143   unsigned int i;
 144   uint16_t put_path_length;
 145   uint16_t data_size;
 146
 147   if (ntohs (reply->type) != GNUNET_MESSAGE_TYPE_DHT_GET_RESULT)
 148     {
 149       GNUNET_break (0);
 150       return;
 151     }
 152
 153   GNUNET_assert (ntohs (reply->size) >=
 154                  sizeof (struct GNUNET_DHT_GetResultMessage));
 155   result = (const struct GNUNET_DHT_GetResultMessage *) reply;
 156
 157   put_path = NULL;
 158   put_path_length = ntohs(result->put_path_length);
 159   if (put_path_length > 0)
 160     {
 161       data_size = ntohs(result->header.size) - (put_path_length * sizeof(struct GNUNET_PeerIdentity)) - sizeof(struct GNUNET_DHT_GetResultMessage);
 162       path_offset = (char *)&result[1];
 163       //GNUNET_log(GNUNET_ERROR_TYPE_WARNING, "In get_reply_iterator, result->header.size is %d, put_path_length %d, offset is %d, data_size is %d\n", ntohs(result->header.size), put_path_length, ntohs(result->header.size) - (put_path_length * sizeof(struct GNUNET_PeerIdentity)), data_size);
 164       path_offset += data_size;
 165       pos = (const struct GNUNET_PeerIdentity *)path_offset;
 166       //GNUNET_log(GNUNET_ERROR_TYPE_WARNING, "Found put peer %s\n", GNUNET_i2s((const struct GNUNET_PeerIdentity *)path_offset));
 167       put_path = GNUNET_malloc ((put_path_length + 1) * sizeof (struct GNUNET_PeerIdentity*));
 168       for (i = 0; i < put_path_length; i++)
 169         {
 170           put_path[i] = pos;
 171           pos++;
 172         }
 173       put_path[put_path_length] = NULL;
 174     }
 175
 176   payload = ntohs (reply->size) - sizeof(struct GNUNET_DHT_GetResultMessage);
 177   get_handle->iter (get_handle->iter_cls,
 178                     GNUNET_TIME_absolute_ntoh (result->expiration),
 179                     key,
 180                     outgoing_path,
 181                     put_path,
 182                     ntohs (result->type),
 183                     payload,
 184                     &result[1]);
 185   GNUNET_free_non_null(put_path);
 186 }
 187
 188
 189
 190 /**
 191  * Perform an asynchronous GET operation on the DHT identified. See
 192  * also "GNUNET_BLOCK_evaluate".
 193  *
 194  * @param handle handle to the DHT service
 195  * @param timeout how long to wait for transmission of this request to the service
 196  * @param type expected type of the response object
 197  * @param key the key to look up
 198  * @param desired_replication_level estimate of how many
 199                   nearest peers this request should reach
 200  * @param options routing options for this message
 201  * @param bf bloom filter associated with query (can be NULL)
 202  * @param bf_mutator mutation value for bf
 203  * @param xquery extended query data (can be NULL, depending on type)
 204  * @param xquery_size number of bytes in xquery
 205  * @param iter function to call on each result
 206  * @param iter_cls closure for iter
 207  *
 208  * @return handle to stop the async get
 209  */
 210 struct GNUNET_DHT_GetHandle *
 211 GNUNET_DHT_get_start (struct GNUNET_DHT_Handle *handle,
 212                       struct GNUNET_TIME_Relative timeout,
 213                       enum GNUNET_BLOCK_Type type,
 214                       const GNUNET_HashCode * key,
 215                       uint32_t desired_replication_level,
 216                       enum GNUNET_DHT_RouteOption options,
 217                       const struct GNUNET_CONTAINER_BloomFilter *bf,
 218                       int32_t bf_mutator,
 219                       const void *xquery,
 220                       size_t xquery_size,
 221                       GNUNET_DHT_GetIterator iter,
 222                       void *iter_cls)
 223 {
 224   struct GNUNET_DHT_GetHandle *get_handle;
 225   char buf[GNUNET_SERVER_MAX_MESSAGE_SIZE - 1];
 226   struct GNUNET_DHT_GetMessage *get_msg;
 227   size_t bf_size;
 228
 229   bf_size = GNUNET_CONTAINER_bloomfilter_get_size (bf);
 230   if ( (sizeof (buf) <=
 231         sizeof (struct GNUNET_DHT_GetMessage) + xquery_size + bf_size) ||
 232        (sizeof (buf) <= bf_size))
 233     {
 234       GNUNET_break (0);
 235       return NULL;
 236     }
 237   get_handle = GNUNET_malloc (sizeof (struct GNUNET_DHT_GetHandle));
 238   get_handle->iter = iter;
 239   get_handle->iter_cls = iter_cls;
 240   get_msg = (struct GNUNET_DHT_GetMessage*) buf;
 241   get_msg->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_GET);
 242   get_msg->header.size = htons (sizeof (struct GNUNET_DHT_GetMessage) +
 243                                 xquery_size +
 244                                 bf_size);
 245   get_msg->type = htonl ((uint32_t) type);
 246   get_msg->bf_mutator = bf_mutator;
 247   get_msg->xquery_size = htons ((uint16_t) xquery_size);
 248   get_msg->bf_size = htons (bf_size);
 249   if (xquery != NULL)
 250     memcpy (&buf[sizeof(struct GNUNET_DHT_GetMessage)],
 251             xquery,
 252             xquery_size);
 253   else
 254     GNUNET_assert (xquery_size == 0);
 255   (void) GNUNET_CONTAINER_bloomfilter_get_raw_data (bf,
 256                                                     &buf[sizeof(struct GNUNET_DHT_GetMessage) + xquery_size],
 257                                                     bf_size);
 258   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 259               "Starting route for %u byte `%s' message\n",
 260               (unsigned int) (sizeof (struct GNUNET_DHT_GetMessage) + xquery_size + bf_size) ,
 261               "GET");
 262   get_handle->route_handle =
 263     GNUNET_DHT_route_start (handle,
 264                             key,
 265                             desired_replication_level,
 266                             options,
 267                             &get_msg->header,
 268                             timeout,
 269                             &get_reply_iterator, get_handle,
 270                             NULL, NULL);
 271   GNUNET_break (NULL != get_handle->route_handle);
 272   return get_handle;
 273 }
 274
 275
 276 /**
 277  * Stop async DHT-get.
 278  *
 279  * @param get_handle handle to the GET operation to stop
 280  *
 281  * On return get_handle will no longer be valid, caller
 282  * must not use again!!!
 283  */
 284 void
 285 GNUNET_DHT_get_stop (struct GNUNET_DHT_GetHandle *get_handle)
 286 {
 287   GNUNET_DHT_route_stop (get_handle->route_handle);
 288   GNUNET_free (get_handle);
 289 }
 290
 291
 292 /* end of dht_api_get_put.c */