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, void *cont_cls)
  65 {
  66   char buf[GNUNET_SERVER_MAX_MESSAGE_SIZE];
  67   struct GNUNET_DHT_PutMessage *put_msg;
  68
  69   if (size >= sizeof (buf) - sizeof (struct GNUNET_DHT_PutMessage))
  70   {
  71     GNUNET_break (0);
  72     return;
  73   }
  74   put_msg = (struct GNUNET_DHT_PutMessage *) buf;
  75   put_msg->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_PUT);
  76   put_msg->header.size = htons (sizeof (struct GNUNET_DHT_PutMessage) + size);
  77   put_msg->type = htonl ((uint32_t) type);
  78   put_msg->expiration = GNUNET_TIME_absolute_hton (exp);
  79   memcpy (&put_msg[1], data, size);
  80   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
  81               "Starting route for %u byte `%s' message of type %u \n",
  82               (unsigned int) (sizeof (struct GNUNET_DHT_PutMessage) + size),
  83               "PUT", type);
  84   GNUNET_break (NULL ==
  85                 GNUNET_DHT_route_start (handle,
  86                                         key,
  87                                         desired_replication_level, options,
  88                                         &put_msg->header,
  89                                         timeout, NULL, NULL, cont, cont_cls));
  90 }
  91
  92
  93
  94 /**
  95  * Handle to control a get operation.
  96  */
  97 struct GNUNET_DHT_GetHandle
  98 {
  99   /**
 100    * Handle to the actual route operation for the get
 101    */
 102   struct GNUNET_DHT_RouteHandle *route_handle;
 103
 104   /**
 105    * Iterator to call on data receipt
 106    */
 107   GNUNET_DHT_GetIterator iter;
 108
 109   /**
 110    * Closure for the iterator callback
 111    */
 112   void *iter_cls;
 113
 114 };
 115
 116
 117
 118 /**
 119  * Iterator called on each result obtained from a generic route
 120  * operation
 121  *
 122  * @param cls the 'struct GNUNET_DHT_GetHandle'
 123  * @param key key that was used
 124  * @param outgoing_path path of the message from this peer
 125  *                      to the target
 126  * @param reply response
 127  */
 128 static void
 129 get_reply_iterator (void *cls,
 130                     const GNUNET_HashCode * key,
 131                     const struct GNUNET_PeerIdentity *const *outgoing_path,
 132                     const struct GNUNET_MessageHeader *reply)
 133 {
 134   struct GNUNET_DHT_GetHandle *get_handle = cls;
 135   const struct GNUNET_DHT_GetResultMessage *result;
 136   const struct GNUNET_PeerIdentity **put_path;
 137   size_t payload;
 138   char *path_offset;
 139   const struct GNUNET_PeerIdentity *pos;
 140   unsigned int i;
 141   uint16_t put_path_length;
 142   uint16_t data_size;
 143
 144   if (ntohs (reply->type) != GNUNET_MESSAGE_TYPE_DHT_GET_RESULT)
 145   {
 146     GNUNET_break (0);
 147     return;
 148   }
 149
 150   GNUNET_assert (ntohs (reply->size) >=
 151                  sizeof (struct GNUNET_DHT_GetResultMessage));
 152   result = (const struct GNUNET_DHT_GetResultMessage *) reply;
 153
 154   put_path = NULL;
 155   put_path_length = ntohs (result->put_path_length);
 156   if (put_path_length > 0)
 157   {
 158     data_size =
 159         ntohs (result->header.size) -
 160         (put_path_length * sizeof (struct GNUNET_PeerIdentity)) -
 161         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 =
 168         GNUNET_malloc ((put_path_length +
 169                         1) * sizeof (struct GNUNET_PeerIdentity *));
 170     for (i = 0; i < put_path_length; i++)
 171     {
 172       put_path[i] = pos;
 173       pos++;
 174     }
 175     put_path[put_path_length] = NULL;
 176   }
 177
 178   payload = ntohs (reply->size) - sizeof (struct GNUNET_DHT_GetResultMessage);
 179   get_handle->iter (get_handle->iter_cls,
 180                     GNUNET_TIME_absolute_ntoh (result->expiration),
 181                     key,
 182                     outgoing_path,
 183                     put_path, ntohs (result->type), payload, &result[1]);
 184   GNUNET_free_non_null (put_path);
 185 }
 186
 187
 188
 189 /**
 190  * Perform an asynchronous GET operation on the DHT identified. See
 191  * also "GNUNET_BLOCK_evaluate".
 192  *
 193  * @param handle handle to the DHT service
 194  * @param timeout how long to wait for transmission of this request to the service
 195  * @param type expected type of the response object
 196  * @param key the key to look up
 197  * @param desired_replication_level estimate of how many
 198                   nearest peers this request should reach
 199  * @param options routing options for this message
 200  * @param bf bloom filter associated with query (can be NULL)
 201  * @param bf_mutator mutation value for bf
 202  * @param xquery extended query data (can be NULL, depending on type)
 203  * @param xquery_size number of bytes in xquery
 204  * @param iter function to call on each result
 205  * @param iter_cls closure for iter
 206  *
 207  * @return handle to stop the async get
 208  */
 209 struct GNUNET_DHT_GetHandle *
 210 GNUNET_DHT_get_start (struct GNUNET_DHT_Handle *handle,
 211                       struct GNUNET_TIME_Relative timeout,
 212                       enum GNUNET_BLOCK_Type type,
 213                       const GNUNET_HashCode * key,
 214                       uint32_t desired_replication_level,
 215                       enum GNUNET_DHT_RouteOption options,
 216                       const struct GNUNET_CONTAINER_BloomFilter *bf,
 217                       int32_t bf_mutator,
 218                       const void *xquery,
 219                       size_t xquery_size,
 220                       GNUNET_DHT_GetIterator iter, void *iter_cls)
 221 {
 222   struct GNUNET_DHT_GetHandle *get_handle;
 223   char buf[GNUNET_SERVER_MAX_MESSAGE_SIZE - 1];
 224   struct GNUNET_DHT_GetMessage *get_msg;
 225   size_t bf_size;
 226
 227   bf_size = GNUNET_CONTAINER_bloomfilter_get_size (bf);
 228   if ((sizeof (buf) <=
 229        sizeof (struct GNUNET_DHT_GetMessage) + xquery_size + bf_size) ||
 230       (sizeof (buf) <= bf_size))
 231   {
 232     GNUNET_break (0);
 233     return NULL;
 234   }
 235   get_handle = GNUNET_malloc (sizeof (struct GNUNET_DHT_GetHandle));
 236   get_handle->iter = iter;
 237   get_handle->iter_cls = iter_cls;
 238   get_msg = (struct GNUNET_DHT_GetMessage *) buf;
 239   get_msg->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_GET);
 240   get_msg->header.size = htons (sizeof (struct GNUNET_DHT_GetMessage) +
 241                                 xquery_size + bf_size);
 242   get_msg->type = htonl ((uint32_t) type);
 243   get_msg->bf_mutator = bf_mutator;
 244   get_msg->xquery_size = htons ((uint16_t) xquery_size);
 245   get_msg->bf_size = htons (bf_size);
 246   if (xquery != NULL)
 247     memcpy (&buf[sizeof (struct GNUNET_DHT_GetMessage)], xquery, xquery_size);
 248   else
 249     GNUNET_assert (xquery_size == 0);
 250   (void) GNUNET_CONTAINER_bloomfilter_get_raw_data (bf,
 251                                                     &buf[sizeof
 252                                                          (struct
 253                                                           GNUNET_DHT_GetMessage)
 254                                                          + xquery_size],
 255                                                     bf_size);
 256   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 257               "Starting route for %u byte `%s' message\n",
 258               (unsigned int) (sizeof (struct GNUNET_DHT_GetMessage) +
 259                               xquery_size + bf_size), "GET");
 260   get_handle->route_handle =
 261       GNUNET_DHT_route_start (handle, key, desired_replication_level, options,
 262                               &get_msg->header, timeout, &get_reply_iterator,
 263                               get_handle, NULL, NULL);
 264   GNUNET_break (NULL != get_handle->route_handle);
 265   return get_handle;
 266 }
 267
 268
 269 /**
 270  * Stop async DHT-get.
 271  *
 272  * @param get_handle handle to the GET operation to stop
 273  *
 274  * On return get_handle will no longer be valid, caller
 275  * must not use again!!!
 276  */
 277 void
 278 GNUNET_DHT_get_stop (struct GNUNET_DHT_GetHandle *get_handle)
 279 {
 280   GNUNET_DHT_route_stop (get_handle->route_handle);
 281   GNUNET_free (get_handle);
 282 }
 283
 284
 285 /* end of dht_api_get_put.c */