aboutsummaryrefslogtreecommitdiff
path: root/src/dht/dht_api.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/dht/dht_api.c')
-rw-r--r--src/dht/dht_api.c964
1 files changed, 477 insertions, 487 deletions
diff --git a/src/dht/dht_api.c b/src/dht/dht_api.c
index 69ac041cc..516c96479 100644
--- a/src/dht/dht_api.c
+++ b/src/dht/dht_api.c
@@ -11,12 +11,12 @@
11 WITHOUT ANY WARRANTY; without even the implied warranty of 11 WITHOUT ANY WARRANTY; without even the implied warranty of
12 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 12 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13 Affero General Public License for more details. 13 Affero General Public License for more details.
14 14
15 You should have received a copy of the GNU Affero General Public License 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/>. 16 along with this program. If not, see <http://www.gnu.org/licenses/>.
17 17
18 SPDX-License-Identifier: AGPL3.0-or-later 18 SPDX-License-Identifier: AGPL3.0-or-later
19*/ 19 */
20 20
21/** 21/**
22 * @file dht/dht_api.c 22 * @file dht/dht_api.c
@@ -34,14 +34,13 @@
34#include "gnunet_dht_service.h" 34#include "gnunet_dht_service.h"
35#include "dht.h" 35#include "dht.h"
36 36
37#define LOG(kind,...) GNUNET_log_from (kind, "dht-api",__VA_ARGS__) 37#define LOG(kind, ...) GNUNET_log_from(kind, "dht-api", __VA_ARGS__)
38 38
39 39
40/** 40/**
41 * Handle to a PUT request. 41 * Handle to a PUT request.
42 */ 42 */
43struct GNUNET_DHT_PutHandle 43struct GNUNET_DHT_PutHandle {
44{
45 /** 44 /**
46 * Kept in a DLL. 45 * Kept in a DLL.
47 */ 46 */
@@ -71,15 +70,12 @@ struct GNUNET_DHT_PutHandle
71 * Envelope from the PUT operation. 70 * Envelope from the PUT operation.
72 */ 71 */
73 struct GNUNET_MQ_Envelope *env; 72 struct GNUNET_MQ_Envelope *env;
74
75}; 73};
76 74
77/** 75/**
78 * Handle to a GET request 76 * Handle to a GET request
79 */ 77 */
80struct GNUNET_DHT_GetHandle 78struct GNUNET_DHT_GetHandle {
81{
82
83 /** 79 /**
84 * Iterator to call on data receipt 80 * Iterator to call on data receipt
85 */ 81 */
@@ -142,15 +138,13 @@ struct GNUNET_DHT_GetHandle
142 * end of the positions that are actually used. 138 * end of the positions that are actually used.
143 */ 139 */
144 unsigned int seen_results_end; 140 unsigned int seen_results_end;
145
146}; 141};
147 142
148 143
149/** 144/**
150 * Handle to a monitoring request. 145 * Handle to a monitoring request.
151 */ 146 */
152struct GNUNET_DHT_MonitorHandle 147struct GNUNET_DHT_MonitorHandle {
153{
154 /** 148 /**
155 * DLL. 149 * DLL.
156 */ 150 */
@@ -195,16 +189,13 @@ struct GNUNET_DHT_MonitorHandle
195 * Closure for @e get_cb, @e put_cb and @e get_resp_cb. 189 * Closure for @e get_cb, @e put_cb and @e get_resp_cb.
196 */ 190 */
197 void *cb_cls; 191 void *cb_cls;
198
199}; 192};
200 193
201 194
202/** 195/**
203 * Connection to the DHT service. 196 * Connection to the DHT service.
204 */ 197 */
205struct GNUNET_DHT_Handle 198struct GNUNET_DHT_Handle {
206{
207
208 /** 199 /**
209 * Configuration to use. 200 * Configuration to use.
210 */ 201 */
@@ -256,8 +247,6 @@ struct GNUNET_DHT_Handle
256 * Generator for unique ids. 247 * Generator for unique ids.
257 */ 248 */
258 uint64_t uid_gen; 249 uint64_t uid_gen;
259
260
261}; 250};
262 251
263 252
@@ -268,7 +257,7 @@ struct GNUNET_DHT_Handle
268 * @return #GNUNET_YES on success, #GNUNET_NO on failure. 257 * @return #GNUNET_YES on success, #GNUNET_NO on failure.
269 */ 258 */
270static int 259static int
271try_connect (struct GNUNET_DHT_Handle *h); 260try_connect(struct GNUNET_DHT_Handle *h);
272 261
273 262
274/** 263/**
@@ -277,25 +266,25 @@ try_connect (struct GNUNET_DHT_Handle *h);
277 * @param gh GET to generate messages for. 266 * @param gh GET to generate messages for.
278 */ 267 */
279static void 268static void
280send_get (struct GNUNET_DHT_GetHandle *gh) 269send_get(struct GNUNET_DHT_GetHandle *gh)
281{ 270{
282 struct GNUNET_DHT_Handle *h = gh->dht_handle; 271 struct GNUNET_DHT_Handle *h = gh->dht_handle;
283 struct GNUNET_MQ_Envelope *env; 272 struct GNUNET_MQ_Envelope *env;
284 struct GNUNET_DHT_ClientGetMessage *get_msg; 273 struct GNUNET_DHT_ClientGetMessage *get_msg;
285 274
286 env = GNUNET_MQ_msg_extra (get_msg, 275 env = GNUNET_MQ_msg_extra(get_msg,
287 gh->xquery_size, 276 gh->xquery_size,
288 GNUNET_MESSAGE_TYPE_DHT_CLIENT_GET); 277 GNUNET_MESSAGE_TYPE_DHT_CLIENT_GET);
289 get_msg->options = htonl ((uint32_t) gh->options); 278 get_msg->options = htonl((uint32_t)gh->options);
290 get_msg->desired_replication_level = htonl (gh->desired_replication_level); 279 get_msg->desired_replication_level = htonl(gh->desired_replication_level);
291 get_msg->type = htonl (gh->type); 280 get_msg->type = htonl(gh->type);
292 get_msg->key = gh->key; 281 get_msg->key = gh->key;
293 get_msg->unique_id = gh->unique_id; 282 get_msg->unique_id = gh->unique_id;
294 GNUNET_memcpy (&get_msg[1], 283 GNUNET_memcpy(&get_msg[1],
295 &gh[1], 284 &gh[1],
296 gh->xquery_size); 285 gh->xquery_size);
297 GNUNET_MQ_send (h->mq, 286 GNUNET_MQ_send(h->mq,
298 env); 287 env);
299} 288}
300 289
301 290
@@ -309,8 +298,8 @@ send_get (struct GNUNET_DHT_GetHandle *gh)
309 * @param transmission_offset_start at which offset should we start? 298 * @param transmission_offset_start at which offset should we start?
310 */ 299 */
311static void 300static void
312send_get_known_results (struct GNUNET_DHT_GetHandle *gh, 301send_get_known_results(struct GNUNET_DHT_GetHandle *gh,
313 unsigned int transmission_offset_start) 302 unsigned int transmission_offset_start)
314{ 303{
315 struct GNUNET_DHT_Handle *h = gh->dht_handle; 304 struct GNUNET_DHT_Handle *h = gh->dht_handle;
316 struct GNUNET_MQ_Envelope *env; 305 struct GNUNET_MQ_Envelope *env;
@@ -319,26 +308,26 @@ send_get_known_results (struct GNUNET_DHT_GetHandle *gh,
319 unsigned int max; 308 unsigned int max;
320 unsigned int transmission_offset; 309 unsigned int transmission_offset;
321 310
322 max = (GNUNET_MAX_MESSAGE_SIZE - sizeof (*msg)) 311 max = (GNUNET_MAX_MESSAGE_SIZE - sizeof(*msg))
323 / sizeof (struct GNUNET_HashCode); 312 / sizeof(struct GNUNET_HashCode);
324 transmission_offset = transmission_offset_start; 313 transmission_offset = transmission_offset_start;
325 while (transmission_offset < gh->seen_results_end) 314 while (transmission_offset < gh->seen_results_end)
326 { 315 {
327 delta = gh->seen_results_end - transmission_offset; 316 delta = gh->seen_results_end - transmission_offset;
328 if (delta > max) 317 if (delta > max)
329 delta = max; 318 delta = max;
330 env = GNUNET_MQ_msg_extra (msg, 319 env = GNUNET_MQ_msg_extra(msg,
331 delta * sizeof (struct GNUNET_HashCode), 320 delta * sizeof(struct GNUNET_HashCode),
332 GNUNET_MESSAGE_TYPE_DHT_CLIENT_GET_RESULTS_KNOWN); 321 GNUNET_MESSAGE_TYPE_DHT_CLIENT_GET_RESULTS_KNOWN);
333 msg->key = gh->key; 322 msg->key = gh->key;
334 msg->unique_id = gh->unique_id; 323 msg->unique_id = gh->unique_id;
335 GNUNET_memcpy (&msg[1], 324 GNUNET_memcpy(&msg[1],
336 &gh->seen_results[transmission_offset], 325 &gh->seen_results[transmission_offset],
337 sizeof (struct GNUNET_HashCode) * delta); 326 sizeof(struct GNUNET_HashCode) * delta);
338 GNUNET_MQ_send (h->mq, 327 GNUNET_MQ_send(h->mq,
339 env); 328 env);
340 transmission_offset += delta; 329 transmission_offset += delta;
341 } 330 }
342} 331}
343 332
344 333
@@ -352,19 +341,19 @@ send_get_known_results (struct GNUNET_DHT_GetHandle *gh,
352 * @return #GNUNET_YES (always) 341 * @return #GNUNET_YES (always)
353 */ 342 */
354static int 343static int
355add_get_request_to_pending (void *cls, 344add_get_request_to_pending(void *cls,
356 const struct GNUNET_HashCode *key, 345 const struct GNUNET_HashCode *key,
357 void *value) 346 void *value)
358{ 347{
359 struct GNUNET_DHT_Handle *handle = cls; 348 struct GNUNET_DHT_Handle *handle = cls;
360 struct GNUNET_DHT_GetHandle *gh = value; 349 struct GNUNET_DHT_GetHandle *gh = value;
361 350
362 LOG (GNUNET_ERROR_TYPE_DEBUG, 351 LOG(GNUNET_ERROR_TYPE_DEBUG,
363 "Retransmitting request related to %s to DHT %p\n", 352 "Retransmitting request related to %s to DHT %p\n",
364 GNUNET_h2s (key), 353 GNUNET_h2s(key),
365 handle); 354 handle);
366 send_get (gh); 355 send_get(gh);
367 send_get_known_results (gh, 0); 356 send_get_known_results(gh, 0);
368 return GNUNET_YES; 357 return GNUNET_YES;
369} 358}
370 359
@@ -375,25 +364,25 @@ add_get_request_to_pending (void *cls,
375 * @param mh monitor handle to generate start message for 364 * @param mh monitor handle to generate start message for
376 */ 365 */
377static void 366static void
378send_monitor_start (struct GNUNET_DHT_MonitorHandle *mh) 367send_monitor_start(struct GNUNET_DHT_MonitorHandle *mh)
379{ 368{
380 struct GNUNET_DHT_Handle *h = mh->dht_handle; 369 struct GNUNET_DHT_Handle *h = mh->dht_handle;
381 struct GNUNET_MQ_Envelope *env; 370 struct GNUNET_MQ_Envelope *env;
382 struct GNUNET_DHT_MonitorStartStopMessage *m; 371 struct GNUNET_DHT_MonitorStartStopMessage *m;
383 372
384 env = GNUNET_MQ_msg (m, 373 env = GNUNET_MQ_msg(m,
385 GNUNET_MESSAGE_TYPE_DHT_MONITOR_START); 374 GNUNET_MESSAGE_TYPE_DHT_MONITOR_START);
386 m->type = htonl (mh->type); 375 m->type = htonl(mh->type);
387 m->get = htons (NULL != mh->get_cb); 376 m->get = htons(NULL != mh->get_cb);
388 m->get_resp = htons (NULL != mh->get_resp_cb); 377 m->get_resp = htons(NULL != mh->get_resp_cb);
389 m->put = htons (NULL != mh->put_cb); 378 m->put = htons(NULL != mh->put_cb);
390 if (NULL != mh->key) 379 if (NULL != mh->key)
391 { 380 {
392 m->filter_key = htons(1); 381 m->filter_key = htons(1);
393 m->key = *mh->key; 382 m->key = *mh->key;
394 } 383 }
395 GNUNET_MQ_send (h->mq, 384 GNUNET_MQ_send(h->mq,
396 env); 385 env);
397} 386}
398 387
399 388
@@ -403,31 +392,31 @@ send_monitor_start (struct GNUNET_DHT_MonitorHandle *mh)
403 * @param cls a `struct GNUNET_DHT_Handle` 392 * @param cls a `struct GNUNET_DHT_Handle`
404 */ 393 */
405static void 394static void
406try_reconnect (void *cls) 395try_reconnect(void *cls)
407{ 396{
408 struct GNUNET_DHT_Handle *h = cls; 397 struct GNUNET_DHT_Handle *h = cls;
409 struct GNUNET_DHT_MonitorHandle *mh; 398 struct GNUNET_DHT_MonitorHandle *mh;
410 399
411 LOG (GNUNET_ERROR_TYPE_DEBUG, 400 LOG(GNUNET_ERROR_TYPE_DEBUG,
412 "Reconnecting with DHT %p\n", 401 "Reconnecting with DHT %p\n",
413 h); 402 h);
414 h->retry_time = GNUNET_TIME_STD_BACKOFF (h->retry_time); 403 h->retry_time = GNUNET_TIME_STD_BACKOFF(h->retry_time);
415 h->reconnect_task = NULL; 404 h->reconnect_task = NULL;
416 if (GNUNET_YES != try_connect (h)) 405 if (GNUNET_YES != try_connect(h))
417 { 406 {
418 LOG (GNUNET_ERROR_TYPE_WARNING, 407 LOG(GNUNET_ERROR_TYPE_WARNING,
419 "DHT reconnect failed!\n"); 408 "DHT reconnect failed!\n");
420 h->reconnect_task 409 h->reconnect_task
421 = GNUNET_SCHEDULER_add_delayed (h->retry_time, 410 = GNUNET_SCHEDULER_add_delayed(h->retry_time,
422 &try_reconnect, 411 &try_reconnect,
423 h); 412 h);
424 return; 413 return;
425 } 414 }
426 GNUNET_CONTAINER_multihashmap_iterate (h->active_requests, 415 GNUNET_CONTAINER_multihashmap_iterate(h->active_requests,
427 &add_get_request_to_pending, 416 &add_get_request_to_pending,
428 h); 417 h);
429 for (mh = h->monitor_head; NULL != mh; mh = mh->next) 418 for (mh = h->monitor_head; NULL != mh; mh = mh->next)
430 send_monitor_start (mh); 419 send_monitor_start(mh);
431} 420}
432 421
433 422
@@ -437,7 +426,7 @@ try_reconnect (void *cls)
437 * @param h handle to dht to (possibly) disconnect and reconnect 426 * @param h handle to dht to (possibly) disconnect and reconnect
438 */ 427 */
439static void 428static void
440do_disconnect (struct GNUNET_DHT_Handle *h) 429do_disconnect(struct GNUNET_DHT_Handle *h)
441{ 430{
442 struct GNUNET_DHT_PutHandle *ph; 431 struct GNUNET_DHT_PutHandle *ph;
443 GNUNET_SCHEDULER_TaskCallback cont; 432 GNUNET_SCHEDULER_TaskCallback cont;
@@ -445,27 +434,27 @@ do_disconnect (struct GNUNET_DHT_Handle *h)
445 434
446 if (NULL == h->mq) 435 if (NULL == h->mq)
447 return; 436 return;
448 GNUNET_MQ_destroy (h->mq); 437 GNUNET_MQ_destroy(h->mq);
449 h->mq = NULL; 438 h->mq = NULL;
450 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, 439 GNUNET_log(GNUNET_ERROR_TYPE_DEBUG,
451 "Disconnecting from DHT service, will try to reconnect in %s\n", 440 "Disconnecting from DHT service, will try to reconnect in %s\n",
452 GNUNET_STRINGS_relative_time_to_string (h->retry_time, 441 GNUNET_STRINGS_relative_time_to_string(h->retry_time,
453 GNUNET_YES)); 442 GNUNET_YES));
454 /* notify client about all PUTs that (may) have failed due to disconnect */ 443 /* notify client about all PUTs that (may) have failed due to disconnect */
455 while (NULL != (ph = h->put_head)) 444 while (NULL != (ph = h->put_head))
456 { 445 {
457 cont = ph->cont; 446 cont = ph->cont;
458 cont_cls = ph->cont_cls; 447 cont_cls = ph->cont_cls;
459 ph->env = NULL; 448 ph->env = NULL;
460 GNUNET_DHT_put_cancel (ph); 449 GNUNET_DHT_put_cancel(ph);
461 if (NULL != cont) 450 if (NULL != cont)
462 cont (cont_cls); 451 cont(cont_cls);
463 } 452 }
464 GNUNET_assert (NULL == h->reconnect_task); 453 GNUNET_assert(NULL == h->reconnect_task);
465 h->reconnect_task 454 h->reconnect_task
466 = GNUNET_SCHEDULER_add_delayed (h->retry_time, 455 = GNUNET_SCHEDULER_add_delayed(h->retry_time,
467 &try_reconnect, 456 &try_reconnect,
468 h); 457 h);
469} 458}
470 459
471 460
@@ -478,12 +467,12 @@ do_disconnect (struct GNUNET_DHT_Handle *h)
478 * @param error error code 467 * @param error error code
479 */ 468 */
480static void 469static void
481mq_error_handler (void *cls, 470mq_error_handler(void *cls,
482 enum GNUNET_MQ_Error error) 471 enum GNUNET_MQ_Error error)
483{ 472{
484 struct GNUNET_DHT_Handle *h = cls; 473 struct GNUNET_DHT_Handle *h = cls;
485 474
486 do_disconnect (h); 475 do_disconnect(h);
487} 476}
488 477
489 478
@@ -496,18 +485,18 @@ mq_error_handler (void *cls,
496 * #GNUNET_SYSERR if the message is malformed. 485 * #GNUNET_SYSERR if the message is malformed.
497 */ 486 */
498static int 487static int
499check_monitor_get (void *cls, 488check_monitor_get(void *cls,
500 const struct GNUNET_DHT_MonitorGetMessage *msg) 489 const struct GNUNET_DHT_MonitorGetMessage *msg)
501{ 490{
502 uint32_t plen = ntohl (msg->get_path_length); 491 uint32_t plen = ntohl(msg->get_path_length);
503 uint16_t msize = ntohs (msg->header.size) - sizeof (*msg); 492 uint16_t msize = ntohs(msg->header.size) - sizeof(*msg);
504 493
505 if ( (plen > UINT16_MAX) || 494 if ((plen > UINT16_MAX) ||
506 (plen * sizeof (struct GNUNET_PeerIdentity) != msize) ) 495 (plen * sizeof(struct GNUNET_PeerIdentity) != msize))
507 { 496 {
508 GNUNET_break (0); 497 GNUNET_break(0);
509 return GNUNET_SYSERR; 498 return GNUNET_SYSERR;
510 } 499 }
511 return GNUNET_OK; 500 return GNUNET_OK;
512} 501}
513 502
@@ -519,31 +508,31 @@ check_monitor_get (void *cls,
519 * @param msg Monitor get message from the service. 508 * @param msg Monitor get message from the service.
520 */ 509 */
521static void 510static void
522handle_monitor_get (void *cls, 511handle_monitor_get(void *cls,
523 const struct GNUNET_DHT_MonitorGetMessage *msg) 512 const struct GNUNET_DHT_MonitorGetMessage *msg)
524{ 513{
525 struct GNUNET_DHT_Handle *handle = cls; 514 struct GNUNET_DHT_Handle *handle = cls;
526 struct GNUNET_DHT_MonitorHandle *mh; 515 struct GNUNET_DHT_MonitorHandle *mh;
527 516
528 for (mh = handle->monitor_head; NULL != mh; mh = mh->next) 517 for (mh = handle->monitor_head; NULL != mh; mh = mh->next)
529 { 518 {
530 if (NULL == mh->get_cb) 519 if (NULL == mh->get_cb)
531 continue; 520 continue;
532 if ( ( (GNUNET_BLOCK_TYPE_ANY == mh->type) || 521 if (((GNUNET_BLOCK_TYPE_ANY == mh->type) ||
533 (mh->type == ntohl (msg->type)) ) && 522 (mh->type == ntohl(msg->type))) &&
534 ( (NULL == mh->key) || 523 ((NULL == mh->key) ||
535 (0 == memcmp (mh->key, 524 (0 == memcmp(mh->key,
536 &msg->key, 525 &msg->key,
537 sizeof (struct GNUNET_HashCode))) ) ) 526 sizeof(struct GNUNET_HashCode)))))
538 mh->get_cb (mh->cb_cls, 527 mh->get_cb(mh->cb_cls,
539 ntohl (msg->options), 528 ntohl(msg->options),
540 (enum GNUNET_BLOCK_Type) ntohl(msg->type), 529 (enum GNUNET_BLOCK_Type)ntohl(msg->type),
541 ntohl (msg->hop_count), 530 ntohl(msg->hop_count),
542 ntohl (msg->desired_replication_level), 531 ntohl(msg->desired_replication_level),
543 ntohl (msg->get_path_length), 532 ntohl(msg->get_path_length),
544 (struct GNUNET_PeerIdentity *) &msg[1], 533 (struct GNUNET_PeerIdentity *)&msg[1],
545 &msg->key); 534 &msg->key);
546 } 535 }
547} 536}
548 537
549 538
@@ -556,19 +545,19 @@ handle_monitor_get (void *cls,
556 * #GNUNET_SYSERR if the message is malformed. 545 * #GNUNET_SYSERR if the message is malformed.
557 */ 546 */
558static int 547static int
559check_monitor_get_resp (void *cls, 548check_monitor_get_resp(void *cls,
560 const struct GNUNET_DHT_MonitorGetRespMessage *msg) 549 const struct GNUNET_DHT_MonitorGetRespMessage *msg)
561{ 550{
562 size_t msize = ntohs (msg->header.size) - sizeof (*msg); 551 size_t msize = ntohs(msg->header.size) - sizeof(*msg);
563 uint32_t getl = ntohl (msg->get_path_length); 552 uint32_t getl = ntohl(msg->get_path_length);
564 uint32_t putl = ntohl (msg->put_path_length); 553 uint32_t putl = ntohl(msg->put_path_length);
565 554
566 if ( (getl + putl < getl) || 555 if ((getl + putl < getl) ||
567 ( (msize / sizeof (struct GNUNET_PeerIdentity)) < getl + putl) ) 556 ((msize / sizeof(struct GNUNET_PeerIdentity)) < getl + putl))
568 { 557 {
569 GNUNET_break (0); 558 GNUNET_break(0);
570 return GNUNET_SYSERR; 559 return GNUNET_SYSERR;
571 } 560 }
572 return GNUNET_OK; 561 return GNUNET_OK;
573} 562}
574 563
@@ -580,38 +569,38 @@ check_monitor_get_resp (void *cls,
580 * @param msg monitor get response message from the service 569 * @param msg monitor get response message from the service
581 */ 570 */
582static void 571static void
583handle_monitor_get_resp (void *cls, 572handle_monitor_get_resp(void *cls,
584 const struct GNUNET_DHT_MonitorGetRespMessage *msg) 573 const struct GNUNET_DHT_MonitorGetRespMessage *msg)
585{ 574{
586 struct GNUNET_DHT_Handle *handle = cls; 575 struct GNUNET_DHT_Handle *handle = cls;
587 size_t msize = ntohs (msg->header.size) - sizeof (*msg); 576 size_t msize = ntohs(msg->header.size) - sizeof(*msg);
588 const struct GNUNET_PeerIdentity *path; 577 const struct GNUNET_PeerIdentity *path;
589 uint32_t getl = ntohl (msg->get_path_length); 578 uint32_t getl = ntohl(msg->get_path_length);
590 uint32_t putl = ntohl (msg->put_path_length); 579 uint32_t putl = ntohl(msg->put_path_length);
591 struct GNUNET_DHT_MonitorHandle *mh; 580 struct GNUNET_DHT_MonitorHandle *mh;
592 581
593 path = (const struct GNUNET_PeerIdentity *) &msg[1]; 582 path = (const struct GNUNET_PeerIdentity *)&msg[1];
594 for (mh = handle->monitor_head; NULL != mh; mh = mh->next) 583 for (mh = handle->monitor_head; NULL != mh; mh = mh->next)
595 { 584 {
596 if (NULL == mh->get_resp_cb) 585 if (NULL == mh->get_resp_cb)
597 continue; 586 continue;
598 if ( ( (GNUNET_BLOCK_TYPE_ANY == mh->type) || 587 if (((GNUNET_BLOCK_TYPE_ANY == mh->type) ||
599 (mh->type == ntohl(msg->type)) ) && 588 (mh->type == ntohl(msg->type))) &&
600 ( (NULL == mh->key) || 589 ((NULL == mh->key) ||
601 (0 == memcmp (mh->key, 590 (0 == memcmp(mh->key,
602 &msg->key, 591 &msg->key,
603 sizeof (struct GNUNET_HashCode))) ) ) 592 sizeof(struct GNUNET_HashCode)))))
604 mh->get_resp_cb (mh->cb_cls, 593 mh->get_resp_cb(mh->cb_cls,
605 (enum GNUNET_BLOCK_Type) ntohl (msg->type), 594 (enum GNUNET_BLOCK_Type)ntohl(msg->type),
606 path, 595 path,
607 getl, 596 getl,
608 &path[getl], 597 &path[getl],
609 putl, 598 putl,
610 GNUNET_TIME_absolute_ntoh(msg->expiration_time), 599 GNUNET_TIME_absolute_ntoh(msg->expiration_time),
611 &msg->key, 600 &msg->key,
612 (const void *) &path[getl + putl], 601 (const void *)&path[getl + putl],
613 msize - sizeof (struct GNUNET_PeerIdentity) * (putl + getl)); 602 msize - sizeof(struct GNUNET_PeerIdentity) * (putl + getl));
614 } 603 }
615} 604}
616 605
617 606
@@ -624,19 +613,19 @@ handle_monitor_get_resp (void *cls,
624 * #GNUNET_SYSERR if the message is malformed. 613 * #GNUNET_SYSERR if the message is malformed.
625 */ 614 */
626static int 615static int
627check_monitor_put (void *cls, 616check_monitor_put(void *cls,
628 const struct GNUNET_DHT_MonitorPutMessage *msg) 617 const struct GNUNET_DHT_MonitorPutMessage *msg)
629{ 618{
630 size_t msize; 619 size_t msize;
631 uint32_t putl; 620 uint32_t putl;
632 621
633 msize = ntohs (msg->header.size) - sizeof (*msg); 622 msize = ntohs(msg->header.size) - sizeof(*msg);
634 putl = ntohl (msg->put_path_length); 623 putl = ntohl(msg->put_path_length);
635 if ((msize / sizeof (struct GNUNET_PeerIdentity)) < putl) 624 if ((msize / sizeof(struct GNUNET_PeerIdentity)) < putl)
636 { 625 {
637 GNUNET_break (0); 626 GNUNET_break(0);
638 return GNUNET_SYSERR; 627 return GNUNET_SYSERR;
639 } 628 }
640 return GNUNET_OK; 629 return GNUNET_OK;
641} 630}
642 631
@@ -648,38 +637,38 @@ check_monitor_put (void *cls,
648 * @param msg Monitor put message from the service. 637 * @param msg Monitor put message from the service.
649 */ 638 */
650static void 639static void
651handle_monitor_put (void *cls, 640handle_monitor_put(void *cls,
652 const struct GNUNET_DHT_MonitorPutMessage *msg) 641 const struct GNUNET_DHT_MonitorPutMessage *msg)
653{ 642{
654 struct GNUNET_DHT_Handle *handle = cls; 643 struct GNUNET_DHT_Handle *handle = cls;
655 size_t msize = ntohs (msg->header.size) - sizeof (*msg); 644 size_t msize = ntohs(msg->header.size) - sizeof(*msg);
656 uint32_t putl = ntohl (msg->put_path_length); 645 uint32_t putl = ntohl(msg->put_path_length);
657 const struct GNUNET_PeerIdentity *path; 646 const struct GNUNET_PeerIdentity *path;
658 struct GNUNET_DHT_MonitorHandle *mh; 647 struct GNUNET_DHT_MonitorHandle *mh;
659 648
660 path = (const struct GNUNET_PeerIdentity *) &msg[1]; 649 path = (const struct GNUNET_PeerIdentity *)&msg[1];
661 for (mh = handle->monitor_head; NULL != mh; mh = mh->next) 650 for (mh = handle->monitor_head; NULL != mh; mh = mh->next)
662 { 651 {
663 if (NULL == mh->put_cb) 652 if (NULL == mh->put_cb)
664 continue; 653 continue;
665 if ( ( (GNUNET_BLOCK_TYPE_ANY == mh->type) || 654 if (((GNUNET_BLOCK_TYPE_ANY == mh->type) ||
666 (mh->type == ntohl(msg->type)) ) && 655 (mh->type == ntohl(msg->type))) &&
667 ( (NULL == mh->key) || 656 ((NULL == mh->key) ||
668 (0 == memcmp (mh->key, 657 (0 == memcmp(mh->key,
669 &msg->key, 658 &msg->key,
670 sizeof (struct GNUNET_HashCode))) ) ) 659 sizeof(struct GNUNET_HashCode)))))
671 mh->put_cb (mh->cb_cls, 660 mh->put_cb(mh->cb_cls,
672 ntohl (msg->options), 661 ntohl(msg->options),
673 (enum GNUNET_BLOCK_Type) ntohl(msg->type), 662 (enum GNUNET_BLOCK_Type)ntohl(msg->type),
674 ntohl (msg->hop_count), 663 ntohl(msg->hop_count),
675 ntohl (msg->desired_replication_level), 664 ntohl(msg->desired_replication_level),
676 putl, 665 putl,
677 path, 666 path,
678 GNUNET_TIME_absolute_ntoh(msg->expiration_time), 667 GNUNET_TIME_absolute_ntoh(msg->expiration_time),
679 &msg->key, 668 &msg->key,
680 (const void *) &path[putl], 669 (const void *)&path[putl],
681 msize - sizeof (struct GNUNET_PeerIdentity) * putl); 670 msize - sizeof(struct GNUNET_PeerIdentity) * putl);
682 } 671 }
683} 672}
684 673
685 674
@@ -692,25 +681,25 @@ handle_monitor_put (void *cls,
692 * #GNUNET_SYSERR if the message is malformed. 681 * #GNUNET_SYSERR if the message is malformed.
693 */ 682 */
694static int 683static int
695check_client_result (void *cls, 684check_client_result(void *cls,
696 const struct GNUNET_DHT_ClientResultMessage *msg) 685 const struct GNUNET_DHT_ClientResultMessage *msg)
697{ 686{
698 size_t msize = ntohs (msg->header.size) - sizeof (*msg); 687 size_t msize = ntohs(msg->header.size) - sizeof(*msg);
699 uint32_t put_path_length = ntohl (msg->put_path_length); 688 uint32_t put_path_length = ntohl(msg->put_path_length);
700 uint32_t get_path_length = ntohl (msg->get_path_length); 689 uint32_t get_path_length = ntohl(msg->get_path_length);
701 size_t meta_length; 690 size_t meta_length;
702 691
703 meta_length = 692 meta_length =
704 sizeof (struct GNUNET_PeerIdentity) * (get_path_length + put_path_length); 693 sizeof(struct GNUNET_PeerIdentity) * (get_path_length + put_path_length);
705 if ( (msize < meta_length) || 694 if ((msize < meta_length) ||
706 (get_path_length > 695 (get_path_length >
707 GNUNET_MAX_MESSAGE_SIZE / sizeof (struct GNUNET_PeerIdentity)) || 696 GNUNET_MAX_MESSAGE_SIZE / sizeof(struct GNUNET_PeerIdentity)) ||
708 (put_path_length > 697 (put_path_length >
709 GNUNET_MAX_MESSAGE_SIZE / sizeof (struct GNUNET_PeerIdentity)) ) 698 GNUNET_MAX_MESSAGE_SIZE / sizeof(struct GNUNET_PeerIdentity)))
710 { 699 {
711 GNUNET_break (0); 700 GNUNET_break(0);
712 return GNUNET_SYSERR; 701 return GNUNET_SYSERR;
713 } 702 }
714 return GNUNET_OK; 703 return GNUNET_OK;
715} 704}
716 705
@@ -724,15 +713,15 @@ check_client_result (void *cls,
724 * @return #GNUNET_YES to continue to iterate over all results 713 * @return #GNUNET_YES to continue to iterate over all results
725 */ 714 */
726static int 715static int
727process_client_result (void *cls, 716process_client_result(void *cls,
728 const struct GNUNET_HashCode *key, 717 const struct GNUNET_HashCode *key,
729 void *value) 718 void *value)
730{ 719{
731 const struct GNUNET_DHT_ClientResultMessage *crm = cls; 720 const struct GNUNET_DHT_ClientResultMessage *crm = cls;
732 struct GNUNET_DHT_GetHandle *get_handle = value; 721 struct GNUNET_DHT_GetHandle *get_handle = value;
733 size_t msize = ntohs (crm->header.size) - sizeof (*crm); 722 size_t msize = ntohs(crm->header.size) - sizeof(*crm);
734 uint32_t put_path_length = ntohl (crm->put_path_length); 723 uint32_t put_path_length = ntohl(crm->put_path_length);
735 uint32_t get_path_length = ntohl (crm->get_path_length); 724 uint32_t get_path_length = ntohl(crm->get_path_length);
736 const struct GNUNET_PeerIdentity *put_path; 725 const struct GNUNET_PeerIdentity *put_path;
737 const struct GNUNET_PeerIdentity *get_path; 726 const struct GNUNET_PeerIdentity *get_path;
738 struct GNUNET_HashCode hc; 727 struct GNUNET_HashCode hc;
@@ -741,59 +730,59 @@ process_client_result (void *cls,
741 const void *data; 730 const void *data;
742 731
743 if (crm->unique_id != get_handle->unique_id) 732 if (crm->unique_id != get_handle->unique_id)
744 { 733 {
745 /* UID mismatch */ 734 /* UID mismatch */
746 LOG (GNUNET_ERROR_TYPE_DEBUG, 735 LOG(GNUNET_ERROR_TYPE_DEBUG,
747 "Ignoring reply for %s: UID mismatch: %llu/%llu\n", 736 "Ignoring reply for %s: UID mismatch: %llu/%llu\n",
748 GNUNET_h2s (key), 737 GNUNET_h2s(key),
749 crm->unique_id, 738 crm->unique_id,
750 get_handle->unique_id); 739 get_handle->unique_id);
751 return GNUNET_YES; 740 return GNUNET_YES;
752 } 741 }
753 /* FIXME: might want to check that type matches */ 742 /* FIXME: might want to check that type matches */
754 meta_length = 743 meta_length =
755 sizeof (struct GNUNET_PeerIdentity) * (get_path_length + put_path_length); 744 sizeof(struct GNUNET_PeerIdentity) * (get_path_length + put_path_length);
756 data_length = msize - meta_length; 745 data_length = msize - meta_length;
757 put_path = (const struct GNUNET_PeerIdentity *) &crm[1]; 746 put_path = (const struct GNUNET_PeerIdentity *)&crm[1];
758 get_path = &put_path[put_path_length]; 747 get_path = &put_path[put_path_length];
759 { 748 {
760 char *pp; 749 char *pp;
761 char *gp; 750 char *gp;
762 751
763 gp = GNUNET_STRINGS_pp2s (get_path, 752 gp = GNUNET_STRINGS_pp2s(get_path,
764 get_path_length); 753 get_path_length);
765 pp = GNUNET_STRINGS_pp2s (put_path, 754 pp = GNUNET_STRINGS_pp2s(put_path,
766 put_path_length); 755 put_path_length);
767 LOG (GNUNET_ERROR_TYPE_DEBUG, 756 LOG(GNUNET_ERROR_TYPE_DEBUG,
768 "Giving %u byte reply for %s to application (GP: %s, PP: %s)\n", 757 "Giving %u byte reply for %s to application (GP: %s, PP: %s)\n",
769 (unsigned int) data_length, 758 (unsigned int)data_length,
770 GNUNET_h2s (key), 759 GNUNET_h2s(key),
771 gp, 760 gp,
772 pp); 761 pp);
773 GNUNET_free (gp); 762 GNUNET_free(gp);
774 GNUNET_free (pp); 763 GNUNET_free(pp);
775 } 764 }
776 data = &get_path[get_path_length]; 765 data = &get_path[get_path_length];
777 /* remember that we've seen this result */ 766 /* remember that we've seen this result */
778 GNUNET_CRYPTO_hash (data, 767 GNUNET_CRYPTO_hash(data,
779 data_length, 768 data_length,
780 &hc); 769 &hc);
781 if (get_handle->seen_results_size == get_handle->seen_results_end) 770 if (get_handle->seen_results_size == get_handle->seen_results_end)
782 GNUNET_array_grow (get_handle->seen_results, 771 GNUNET_array_grow(get_handle->seen_results,
783 get_handle->seen_results_size, 772 get_handle->seen_results_size,
784 get_handle->seen_results_size * 2 + 1); 773 get_handle->seen_results_size * 2 + 1);
785 get_handle->seen_results[get_handle->seen_results_end++] = hc; 774 get_handle->seen_results[get_handle->seen_results_end++] = hc;
786 /* no need to block it explicitly, service already knows about it! */ 775 /* no need to block it explicitly, service already knows about it! */
787 get_handle->iter (get_handle->iter_cls, 776 get_handle->iter(get_handle->iter_cls,
788 GNUNET_TIME_absolute_ntoh (crm->expiration), 777 GNUNET_TIME_absolute_ntoh(crm->expiration),
789 key, 778 key,
790 get_path, 779 get_path,
791 get_path_length, 780 get_path_length,
792 put_path, 781 put_path,
793 put_path_length, 782 put_path_length,
794 ntohl (crm->type), 783 ntohl(crm->type),
795 data_length, 784 data_length,
796 data); 785 data);
797 return GNUNET_YES; 786 return GNUNET_YES;
798} 787}
799 788
@@ -805,15 +794,15 @@ process_client_result (void *cls,
805 * @param msg Monitor put message from the service. 794 * @param msg Monitor put message from the service.
806 */ 795 */
807static void 796static void
808handle_client_result (void *cls, 797handle_client_result(void *cls,
809 const struct GNUNET_DHT_ClientResultMessage *msg) 798 const struct GNUNET_DHT_ClientResultMessage *msg)
810{ 799{
811 struct GNUNET_DHT_Handle *handle = cls; 800 struct GNUNET_DHT_Handle *handle = cls;
812 801
813 GNUNET_CONTAINER_multihashmap_get_multiple (handle->active_requests, 802 GNUNET_CONTAINER_multihashmap_get_multiple(handle->active_requests,
814 &msg->key, 803 &msg->key,
815 &process_client_result, 804 &process_client_result,
816 (void *) msg); 805 (void *)msg);
817} 806}
818 807
819 808
@@ -823,7 +812,7 @@ handle_client_result (void *cls,
823 * @param cls The DHT handle. 812 * @param cls The DHT handle.
824 */ 813 */
825static void 814static void
826handle_put_cont (void *cls) 815handle_put_cont(void *cls)
827{ 816{
828 struct GNUNET_DHT_PutHandle *ph = cls; 817 struct GNUNET_DHT_PutHandle *ph = cls;
829 GNUNET_SCHEDULER_TaskCallback cont; 818 GNUNET_SCHEDULER_TaskCallback cont;
@@ -832,9 +821,9 @@ handle_put_cont (void *cls)
832 cont = ph->cont; 821 cont = ph->cont;
833 cont_cls = ph->cont_cls; 822 cont_cls = ph->cont_cls;
834 ph->env = NULL; 823 ph->env = NULL;
835 GNUNET_DHT_put_cancel (ph); 824 GNUNET_DHT_put_cancel(ph);
836 if (NULL != cont) 825 if (NULL != cont)
837 cont (cont_cls); 826 cont(cont_cls);
838} 827}
839 828
840 829
@@ -845,40 +834,41 @@ handle_put_cont (void *cls)
845 * @return #GNUNET_YES on success, #GNUNET_NO on failure. 834 * @return #GNUNET_YES on success, #GNUNET_NO on failure.
846 */ 835 */
847static int 836static int
848try_connect (struct GNUNET_DHT_Handle *h) 837try_connect(struct GNUNET_DHT_Handle *h)
849{ 838{
850 struct GNUNET_MQ_MessageHandler handlers[] = { 839 struct GNUNET_MQ_MessageHandler handlers[] = {
851 GNUNET_MQ_hd_var_size (monitor_get, 840 GNUNET_MQ_hd_var_size(monitor_get,
852 GNUNET_MESSAGE_TYPE_DHT_MONITOR_GET, 841 GNUNET_MESSAGE_TYPE_DHT_MONITOR_GET,
853 struct GNUNET_DHT_MonitorGetMessage, 842 struct GNUNET_DHT_MonitorGetMessage,
854 h), 843 h),
855 GNUNET_MQ_hd_var_size (monitor_get_resp, 844 GNUNET_MQ_hd_var_size(monitor_get_resp,
856 GNUNET_MESSAGE_TYPE_DHT_MONITOR_GET_RESP, 845 GNUNET_MESSAGE_TYPE_DHT_MONITOR_GET_RESP,
857 struct GNUNET_DHT_MonitorGetRespMessage, 846 struct GNUNET_DHT_MonitorGetRespMessage,
858 h), 847 h),
859 GNUNET_MQ_hd_var_size (monitor_put, 848 GNUNET_MQ_hd_var_size(monitor_put,
860 GNUNET_MESSAGE_TYPE_DHT_MONITOR_PUT, 849 GNUNET_MESSAGE_TYPE_DHT_MONITOR_PUT,
861 struct GNUNET_DHT_MonitorPutMessage, 850 struct GNUNET_DHT_MonitorPutMessage,
862 h), 851 h),
863 GNUNET_MQ_hd_var_size (client_result, 852 GNUNET_MQ_hd_var_size(client_result,
864 GNUNET_MESSAGE_TYPE_DHT_CLIENT_RESULT, 853 GNUNET_MESSAGE_TYPE_DHT_CLIENT_RESULT,
865 struct GNUNET_DHT_ClientResultMessage, 854 struct GNUNET_DHT_ClientResultMessage,
866 h), 855 h),
867 GNUNET_MQ_handler_end () 856 GNUNET_MQ_handler_end()
868 }; 857 };
858
869 if (NULL != h->mq) 859 if (NULL != h->mq)
870 return GNUNET_OK; 860 return GNUNET_OK;
871 h->mq = GNUNET_CLIENT_connect (h->cfg, 861 h->mq = GNUNET_CLIENT_connect(h->cfg,
872 "dht", 862 "dht",
873 handlers, 863 handlers,
874 &mq_error_handler, 864 &mq_error_handler,
875 h); 865 h);
876 if (NULL == h->mq) 866 if (NULL == h->mq)
877 { 867 {
878 LOG (GNUNET_ERROR_TYPE_WARNING, 868 LOG(GNUNET_ERROR_TYPE_WARNING,
879 "Failed to connect to the DHT service!\n"); 869 "Failed to connect to the DHT service!\n");
880 return GNUNET_NO; 870 return GNUNET_NO;
881 } 871 }
882 return GNUNET_YES; 872 return GNUNET_YES;
883} 873}
884 874
@@ -892,24 +882,24 @@ try_connect (struct GNUNET_DHT_Handle *h)
892 * @return handle to the DHT service, or NULL on error 882 * @return handle to the DHT service, or NULL on error
893 */ 883 */
894struct GNUNET_DHT_Handle * 884struct GNUNET_DHT_Handle *
895GNUNET_DHT_connect (const struct GNUNET_CONFIGURATION_Handle *cfg, 885GNUNET_DHT_connect(const struct GNUNET_CONFIGURATION_Handle *cfg,
896 unsigned int ht_len) 886 unsigned int ht_len)
897{ 887{
898 struct GNUNET_DHT_Handle *handle; 888 struct GNUNET_DHT_Handle *handle;
899 889
900 handle = GNUNET_new (struct GNUNET_DHT_Handle); 890 handle = GNUNET_new(struct GNUNET_DHT_Handle);
901 handle->cfg = cfg; 891 handle->cfg = cfg;
902 handle->uid_gen 892 handle->uid_gen
903 = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK, 893 = GNUNET_CRYPTO_random_u64(GNUNET_CRYPTO_QUALITY_WEAK,
904 UINT64_MAX); 894 UINT64_MAX);
905 handle->active_requests 895 handle->active_requests
906 = GNUNET_CONTAINER_multihashmap_create (ht_len, 896 = GNUNET_CONTAINER_multihashmap_create(ht_len,
907 GNUNET_YES); 897 GNUNET_YES);
908 if (GNUNET_NO == try_connect (handle)) 898 if (GNUNET_NO == try_connect(handle))
909 { 899 {
910 GNUNET_DHT_disconnect (handle); 900 GNUNET_DHT_disconnect(handle);
911 return NULL; 901 return NULL;
912 } 902 }
913 return handle; 903 return handle;
914} 904}
915 905
@@ -920,30 +910,30 @@ GNUNET_DHT_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
920 * @param handle handle of the DHT connection to stop 910 * @param handle handle of the DHT connection to stop
921 */ 911 */
922void 912void
923GNUNET_DHT_disconnect (struct GNUNET_DHT_Handle *handle) 913GNUNET_DHT_disconnect(struct GNUNET_DHT_Handle *handle)
924{ 914{
925 struct GNUNET_DHT_PutHandle *ph; 915 struct GNUNET_DHT_PutHandle *ph;
926 916
927 GNUNET_assert (0 == 917 GNUNET_assert(0 ==
928 GNUNET_CONTAINER_multihashmap_size (handle->active_requests)); 918 GNUNET_CONTAINER_multihashmap_size(handle->active_requests));
929 while (NULL != (ph = handle->put_head)) 919 while (NULL != (ph = handle->put_head))
930 { 920 {
931 if (NULL != ph->cont) 921 if (NULL != ph->cont)
932 ph->cont (ph->cont_cls); 922 ph->cont(ph->cont_cls);
933 GNUNET_DHT_put_cancel (ph); 923 GNUNET_DHT_put_cancel(ph);
934 } 924 }
935 if (NULL != handle->mq) 925 if (NULL != handle->mq)
936 { 926 {
937 GNUNET_MQ_destroy (handle->mq); 927 GNUNET_MQ_destroy(handle->mq);
938 handle->mq = NULL; 928 handle->mq = NULL;
939 } 929 }
940 if (NULL != handle->reconnect_task) 930 if (NULL != handle->reconnect_task)
941 { 931 {
942 GNUNET_SCHEDULER_cancel (handle->reconnect_task); 932 GNUNET_SCHEDULER_cancel(handle->reconnect_task);
943 handle->reconnect_task = NULL; 933 handle->reconnect_task = NULL;
944 } 934 }
945 GNUNET_CONTAINER_multihashmap_destroy (handle->active_requests); 935 GNUNET_CONTAINER_multihashmap_destroy(handle->active_requests);
946 GNUNET_free (handle); 936 GNUNET_free(handle);
947} 937}
948 938
949 939
@@ -968,59 +958,59 @@ GNUNET_DHT_disconnect (struct GNUNET_DHT_Handle *handle)
968 * @param cont_cls closure for @a cont 958 * @param cont_cls closure for @a cont
969 */ 959 */
970struct GNUNET_DHT_PutHandle * 960struct GNUNET_DHT_PutHandle *
971GNUNET_DHT_put (struct GNUNET_DHT_Handle *handle, 961GNUNET_DHT_put(struct GNUNET_DHT_Handle *handle,
972 const struct GNUNET_HashCode *key, 962 const struct GNUNET_HashCode *key,
973 uint32_t desired_replication_level, 963 uint32_t desired_replication_level,
974 enum GNUNET_DHT_RouteOption options, 964 enum GNUNET_DHT_RouteOption options,
975 enum GNUNET_BLOCK_Type type, 965 enum GNUNET_BLOCK_Type type,
976 size_t size, 966 size_t size,
977 const void *data, 967 const void *data,
978 struct GNUNET_TIME_Absolute exp, 968 struct GNUNET_TIME_Absolute exp,
979 GNUNET_SCHEDULER_TaskCallback cont, 969 GNUNET_SCHEDULER_TaskCallback cont,
980 void *cont_cls) 970 void *cont_cls)
981{ 971{
982 struct GNUNET_MQ_Envelope *env; 972 struct GNUNET_MQ_Envelope *env;
983 struct GNUNET_DHT_ClientPutMessage *put_msg; 973 struct GNUNET_DHT_ClientPutMessage *put_msg;
984 size_t msize; 974 size_t msize;
985 struct GNUNET_DHT_PutHandle *ph; 975 struct GNUNET_DHT_PutHandle *ph;
986 976
987 msize = sizeof (struct GNUNET_DHT_ClientPutMessage) + size; 977 msize = sizeof(struct GNUNET_DHT_ClientPutMessage) + size;
988 if ((msize >= GNUNET_MAX_MESSAGE_SIZE) || 978 if ((msize >= GNUNET_MAX_MESSAGE_SIZE) ||
989 (size >= GNUNET_MAX_MESSAGE_SIZE)) 979 (size >= GNUNET_MAX_MESSAGE_SIZE))
990 { 980 {
991 GNUNET_break (0); 981 GNUNET_break(0);
992 return NULL; 982 return NULL;
993 } 983 }
994 if (NULL == handle->mq) 984 if (NULL == handle->mq)
995 return NULL; 985 return NULL;
996 LOG (GNUNET_ERROR_TYPE_DEBUG, 986 LOG(GNUNET_ERROR_TYPE_DEBUG,
997 "Sending PUT for %s to DHT via %p\n", 987 "Sending PUT for %s to DHT via %p\n",
998 GNUNET_h2s (key), 988 GNUNET_h2s(key),
999 handle); 989 handle);
1000 ph = GNUNET_new (struct GNUNET_DHT_PutHandle); 990 ph = GNUNET_new(struct GNUNET_DHT_PutHandle);
1001 ph->dht_handle = handle; 991 ph->dht_handle = handle;
1002 ph->cont = cont; 992 ph->cont = cont;
1003 ph->cont_cls = cont_cls; 993 ph->cont_cls = cont_cls;
1004 GNUNET_CONTAINER_DLL_insert_tail (handle->put_head, 994 GNUNET_CONTAINER_DLL_insert_tail(handle->put_head,
1005 handle->put_tail, 995 handle->put_tail,
1006 ph); 996 ph);
1007 env = GNUNET_MQ_msg_extra (put_msg, 997 env = GNUNET_MQ_msg_extra(put_msg,
1008 size, 998 size,
1009 GNUNET_MESSAGE_TYPE_DHT_CLIENT_PUT); 999 GNUNET_MESSAGE_TYPE_DHT_CLIENT_PUT);
1010 GNUNET_MQ_notify_sent (env, 1000 GNUNET_MQ_notify_sent(env,
1011 &handle_put_cont, 1001 &handle_put_cont,
1012 ph); 1002 ph);
1013 ph->env = env; 1003 ph->env = env;
1014 put_msg->type = htonl ((uint32_t) type); 1004 put_msg->type = htonl((uint32_t)type);
1015 put_msg->options = htonl ((uint32_t) options); 1005 put_msg->options = htonl((uint32_t)options);
1016 put_msg->desired_replication_level = htonl (desired_replication_level); 1006 put_msg->desired_replication_level = htonl(desired_replication_level);
1017 put_msg->expiration = GNUNET_TIME_absolute_hton (exp); 1007 put_msg->expiration = GNUNET_TIME_absolute_hton(exp);
1018 put_msg->key = *key; 1008 put_msg->key = *key;
1019 GNUNET_memcpy (&put_msg[1], 1009 GNUNET_memcpy(&put_msg[1],
1020 data, 1010 data,
1021 size); 1011 size);
1022 GNUNET_MQ_send (handle->mq, 1012 GNUNET_MQ_send(handle->mq,
1023 env); 1013 env);
1024 return ph; 1014 return ph;
1025} 1015}
1026 1016
@@ -1037,18 +1027,18 @@ GNUNET_DHT_put (struct GNUNET_DHT_Handle *handle,
1037 * @param ph put operation to cancel ('cont' will no longer be called) 1027 * @param ph put operation to cancel ('cont' will no longer be called)
1038 */ 1028 */
1039void 1029void
1040GNUNET_DHT_put_cancel (struct GNUNET_DHT_PutHandle *ph) 1030GNUNET_DHT_put_cancel(struct GNUNET_DHT_PutHandle *ph)
1041{ 1031{
1042 struct GNUNET_DHT_Handle *handle = ph->dht_handle; 1032 struct GNUNET_DHT_Handle *handle = ph->dht_handle;
1043 1033
1044 if (NULL != ph->env) 1034 if (NULL != ph->env)
1045 GNUNET_MQ_notify_sent (ph->env, 1035 GNUNET_MQ_notify_sent(ph->env,
1046 NULL, 1036 NULL,
1047 NULL); 1037 NULL);
1048 GNUNET_CONTAINER_DLL_remove (handle->put_head, 1038 GNUNET_CONTAINER_DLL_remove(handle->put_head,
1049 handle->put_tail, 1039 handle->put_tail,
1050 ph); 1040 ph);
1051 GNUNET_free (ph); 1041 GNUNET_free(ph);
1052} 1042}
1053 1043
1054 1044
@@ -1069,32 +1059,32 @@ GNUNET_DHT_put_cancel (struct GNUNET_DHT_PutHandle *ph)
1069 * @return handle to stop the async get 1059 * @return handle to stop the async get
1070 */ 1060 */
1071struct GNUNET_DHT_GetHandle * 1061struct GNUNET_DHT_GetHandle *
1072GNUNET_DHT_get_start (struct GNUNET_DHT_Handle *handle, 1062GNUNET_DHT_get_start(struct GNUNET_DHT_Handle *handle,
1073 enum GNUNET_BLOCK_Type type, 1063 enum GNUNET_BLOCK_Type type,
1074 const struct GNUNET_HashCode *key, 1064 const struct GNUNET_HashCode *key,
1075 uint32_t desired_replication_level, 1065 uint32_t desired_replication_level,
1076 enum GNUNET_DHT_RouteOption options, 1066 enum GNUNET_DHT_RouteOption options,
1077 const void *xquery, 1067 const void *xquery,
1078 size_t xquery_size, 1068 size_t xquery_size,
1079 GNUNET_DHT_GetIterator iter, 1069 GNUNET_DHT_GetIterator iter,
1080 void *iter_cls) 1070 void *iter_cls)
1081{ 1071{
1082 struct GNUNET_DHT_GetHandle *gh; 1072 struct GNUNET_DHT_GetHandle *gh;
1083 size_t msize; 1073 size_t msize;
1084 1074
1085 msize = sizeof (struct GNUNET_DHT_ClientGetMessage) + xquery_size; 1075 msize = sizeof(struct GNUNET_DHT_ClientGetMessage) + xquery_size;
1086 if ((msize >= GNUNET_MAX_MESSAGE_SIZE) || 1076 if ((msize >= GNUNET_MAX_MESSAGE_SIZE) ||
1087 (xquery_size >= GNUNET_MAX_MESSAGE_SIZE)) 1077 (xquery_size >= GNUNET_MAX_MESSAGE_SIZE))
1088 { 1078 {
1089 GNUNET_break (0); 1079 GNUNET_break(0);
1090 return NULL; 1080 return NULL;
1091 } 1081 }
1092 LOG (GNUNET_ERROR_TYPE_DEBUG, 1082 LOG(GNUNET_ERROR_TYPE_DEBUG,
1093 "Sending query for %s to DHT %p\n", 1083 "Sending query for %s to DHT %p\n",
1094 GNUNET_h2s (key), 1084 GNUNET_h2s(key),
1095 handle); 1085 handle);
1096 gh = GNUNET_malloc (sizeof (struct GNUNET_DHT_GetHandle) + 1086 gh = GNUNET_malloc(sizeof(struct GNUNET_DHT_GetHandle) +
1097 xquery_size); 1087 xquery_size);
1098 gh->iter = iter; 1088 gh->iter = iter;
1099 gh->iter_cls = iter_cls; 1089 gh->iter_cls = iter_cls;
1100 gh->dht_handle = handle; 1090 gh->dht_handle = handle;
@@ -1104,15 +1094,15 @@ GNUNET_DHT_get_start (struct GNUNET_DHT_Handle *handle,
1104 gh->desired_replication_level = desired_replication_level; 1094 gh->desired_replication_level = desired_replication_level;
1105 gh->type = type; 1095 gh->type = type;
1106 gh->options = options; 1096 gh->options = options;
1107 GNUNET_memcpy (&gh[1], 1097 GNUNET_memcpy(&gh[1],
1108 xquery, 1098 xquery,
1109 xquery_size); 1099 xquery_size);
1110 GNUNET_CONTAINER_multihashmap_put (handle->active_requests, 1100 GNUNET_CONTAINER_multihashmap_put(handle->active_requests,
1111 &gh->key, 1101 &gh->key,
1112 gh, 1102 gh,
1113 GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE); 1103 GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
1114 if (NULL != handle->mq) 1104 if (NULL != handle->mq)
1115 send_get (gh); 1105 send_get(gh);
1116 return gh; 1106 return gh;
1117} 1107}
1118 1108
@@ -1128,9 +1118,9 @@ GNUNET_DHT_get_start (struct GNUNET_DHT_Handle *handle,
1128 * to be blocked 1118 * to be blocked
1129 */ 1119 */
1130void 1120void
1131GNUNET_DHT_get_filter_known_results (struct GNUNET_DHT_GetHandle *get_handle, 1121GNUNET_DHT_get_filter_known_results(struct GNUNET_DHT_GetHandle *get_handle,
1132 unsigned int num_results, 1122 unsigned int num_results,
1133 const struct GNUNET_HashCode *results) 1123 const struct GNUNET_HashCode *results)
1134{ 1124{
1135 unsigned int needed; 1125 unsigned int needed;
1136 unsigned int had; 1126 unsigned int had;
@@ -1138,16 +1128,16 @@ GNUNET_DHT_get_filter_known_results (struct GNUNET_DHT_GetHandle *get_handle,
1138 had = get_handle->seen_results_end; 1128 had = get_handle->seen_results_end;
1139 needed = had + num_results; 1129 needed = had + num_results;
1140 if (needed > get_handle->seen_results_size) 1130 if (needed > get_handle->seen_results_size)
1141 GNUNET_array_grow (get_handle->seen_results, 1131 GNUNET_array_grow(get_handle->seen_results,
1142 get_handle->seen_results_size, 1132 get_handle->seen_results_size,
1143 needed); 1133 needed);
1144 GNUNET_memcpy (&get_handle->seen_results[get_handle->seen_results_end], 1134 GNUNET_memcpy(&get_handle->seen_results[get_handle->seen_results_end],
1145 results, 1135 results,
1146 num_results * sizeof (struct GNUNET_HashCode)); 1136 num_results * sizeof(struct GNUNET_HashCode));
1147 get_handle->seen_results_end += num_results; 1137 get_handle->seen_results_end += num_results;
1148 if (NULL != get_handle->dht_handle->mq) 1138 if (NULL != get_handle->dht_handle->mq)
1149 send_get_known_results (get_handle, 1139 send_get_known_results(get_handle,
1150 had); 1140 had);
1151} 1141}
1152 1142
1153 1143
@@ -1157,35 +1147,35 @@ GNUNET_DHT_get_filter_known_results (struct GNUNET_DHT_GetHandle *get_handle,
1157 * @param get_handle handle to the GET operation to stop 1147 * @param get_handle handle to the GET operation to stop
1158 */ 1148 */
1159void 1149void
1160GNUNET_DHT_get_stop (struct GNUNET_DHT_GetHandle *get_handle) 1150GNUNET_DHT_get_stop(struct GNUNET_DHT_GetHandle *get_handle)
1161{ 1151{
1162 struct GNUNET_DHT_Handle *handle = get_handle->dht_handle; 1152 struct GNUNET_DHT_Handle *handle = get_handle->dht_handle;
1163 1153
1164 LOG (GNUNET_ERROR_TYPE_DEBUG, 1154 LOG(GNUNET_ERROR_TYPE_DEBUG,
1165 "Sending STOP for %s to DHT via %p\n", 1155 "Sending STOP for %s to DHT via %p\n",
1166 GNUNET_h2s (&get_handle->key), 1156 GNUNET_h2s(&get_handle->key),
1167 handle); 1157 handle);
1168 if (NULL != handle->mq) 1158 if (NULL != handle->mq)
1169 { 1159 {
1170 struct GNUNET_MQ_Envelope *env; 1160 struct GNUNET_MQ_Envelope *env;
1171 struct GNUNET_DHT_ClientGetStopMessage *stop_msg; 1161 struct GNUNET_DHT_ClientGetStopMessage *stop_msg;
1172 1162
1173 env = GNUNET_MQ_msg (stop_msg, 1163 env = GNUNET_MQ_msg(stop_msg,
1174 GNUNET_MESSAGE_TYPE_DHT_CLIENT_GET_STOP); 1164 GNUNET_MESSAGE_TYPE_DHT_CLIENT_GET_STOP);
1175 stop_msg->reserved = htonl (0); 1165 stop_msg->reserved = htonl(0);
1176 stop_msg->unique_id = get_handle->unique_id; 1166 stop_msg->unique_id = get_handle->unique_id;
1177 stop_msg->key = get_handle->key; 1167 stop_msg->key = get_handle->key;
1178 GNUNET_MQ_send (handle->mq, 1168 GNUNET_MQ_send(handle->mq,
1179 env); 1169 env);
1180 } 1170 }
1181 GNUNET_assert (GNUNET_YES == 1171 GNUNET_assert(GNUNET_YES ==
1182 GNUNET_CONTAINER_multihashmap_remove (handle->active_requests, 1172 GNUNET_CONTAINER_multihashmap_remove(handle->active_requests,
1183 &get_handle->key, 1173 &get_handle->key,
1184 get_handle)); 1174 get_handle));
1185 GNUNET_array_grow (get_handle->seen_results, 1175 GNUNET_array_grow(get_handle->seen_results,
1186 get_handle->seen_results_end, 1176 get_handle->seen_results_end,
1187 0); 1177 0);
1188 GNUNET_free (get_handle); 1178 GNUNET_free(get_handle);
1189} 1179}
1190 1180
1191 1181
@@ -1202,17 +1192,17 @@ GNUNET_DHT_get_stop (struct GNUNET_DHT_GetHandle *get_handle)
1202 * @return Handle to stop monitoring. 1192 * @return Handle to stop monitoring.
1203 */ 1193 */
1204struct GNUNET_DHT_MonitorHandle * 1194struct GNUNET_DHT_MonitorHandle *
1205GNUNET_DHT_monitor_start (struct GNUNET_DHT_Handle *handle, 1195GNUNET_DHT_monitor_start(struct GNUNET_DHT_Handle *handle,
1206 enum GNUNET_BLOCK_Type type, 1196 enum GNUNET_BLOCK_Type type,
1207 const struct GNUNET_HashCode *key, 1197 const struct GNUNET_HashCode *key,
1208 GNUNET_DHT_MonitorGetCB get_cb, 1198 GNUNET_DHT_MonitorGetCB get_cb,
1209 GNUNET_DHT_MonitorGetRespCB get_resp_cb, 1199 GNUNET_DHT_MonitorGetRespCB get_resp_cb,
1210 GNUNET_DHT_MonitorPutCB put_cb, 1200 GNUNET_DHT_MonitorPutCB put_cb,
1211 void *cb_cls) 1201 void *cb_cls)
1212{ 1202{
1213 struct GNUNET_DHT_MonitorHandle *mh; 1203 struct GNUNET_DHT_MonitorHandle *mh;
1214 1204
1215 mh = GNUNET_new (struct GNUNET_DHT_MonitorHandle); 1205 mh = GNUNET_new(struct GNUNET_DHT_MonitorHandle);
1216 mh->get_cb = get_cb; 1206 mh->get_cb = get_cb;
1217 mh->get_resp_cb = get_resp_cb; 1207 mh->get_resp_cb = get_resp_cb;
1218 mh->put_cb = put_cb; 1208 mh->put_cb = put_cb;
@@ -1220,15 +1210,15 @@ GNUNET_DHT_monitor_start (struct GNUNET_DHT_Handle *handle,
1220 mh->type = type; 1210 mh->type = type;
1221 mh->dht_handle = handle; 1211 mh->dht_handle = handle;
1222 if (NULL != key) 1212 if (NULL != key)
1223 { 1213 {
1224 mh->key = GNUNET_new (struct GNUNET_HashCode); 1214 mh->key = GNUNET_new(struct GNUNET_HashCode);
1225 *mh->key = *key; 1215 *mh->key = *key;
1226 } 1216 }
1227 GNUNET_CONTAINER_DLL_insert (handle->monitor_head, 1217 GNUNET_CONTAINER_DLL_insert(handle->monitor_head,
1228 handle->monitor_tail, 1218 handle->monitor_tail,
1229 mh); 1219 mh);
1230 if (NULL != handle->mq) 1220 if (NULL != handle->mq)
1231 send_monitor_start (mh); 1221 send_monitor_start(mh);
1232 return mh; 1222 return mh;
1233} 1223}
1234 1224
@@ -1241,30 +1231,30 @@ GNUNET_DHT_monitor_start (struct GNUNET_DHT_Handle *handle,
1241 * On return get_handle will no longer be valid, caller must not use again!!! 1231 * On return get_handle will no longer be valid, caller must not use again!!!
1242 */ 1232 */
1243void 1233void
1244GNUNET_DHT_monitor_stop (struct GNUNET_DHT_MonitorHandle *mh) 1234GNUNET_DHT_monitor_stop(struct GNUNET_DHT_MonitorHandle *mh)
1245{ 1235{
1246 struct GNUNET_DHT_Handle *handle = mh->dht_handle; 1236 struct GNUNET_DHT_Handle *handle = mh->dht_handle;
1247 struct GNUNET_DHT_MonitorStartStopMessage *m; 1237 struct GNUNET_DHT_MonitorStartStopMessage *m;
1248 struct GNUNET_MQ_Envelope *env; 1238 struct GNUNET_MQ_Envelope *env;
1249 1239
1250 GNUNET_CONTAINER_DLL_remove (handle->monitor_head, 1240 GNUNET_CONTAINER_DLL_remove(handle->monitor_head,
1251 handle->monitor_tail, 1241 handle->monitor_tail,
1252 mh); 1242 mh);
1253 env = GNUNET_MQ_msg (m, 1243 env = GNUNET_MQ_msg(m,
1254 GNUNET_MESSAGE_TYPE_DHT_MONITOR_STOP); 1244 GNUNET_MESSAGE_TYPE_DHT_MONITOR_STOP);
1255 m->type = htonl (mh->type); 1245 m->type = htonl(mh->type);
1256 m->get = htons (NULL != mh->get_cb); 1246 m->get = htons(NULL != mh->get_cb);
1257 m->get_resp = htons(NULL != mh->get_resp_cb); 1247 m->get_resp = htons(NULL != mh->get_resp_cb);
1258 m->put = htons (NULL != mh->put_cb); 1248 m->put = htons(NULL != mh->put_cb);
1259 if (NULL != mh->key) 1249 if (NULL != mh->key)
1260 { 1250 {
1261 m->filter_key = htons (1); 1251 m->filter_key = htons(1);
1262 m->key = *mh->key; 1252 m->key = *mh->key;
1263 } 1253 }
1264 GNUNET_MQ_send (handle->mq, 1254 GNUNET_MQ_send(handle->mq,
1265 env); 1255 env);
1266 GNUNET_free_non_null (mh->key); 1256 GNUNET_free_non_null(mh->key);
1267 GNUNET_free (mh); 1257 GNUNET_free(mh);
1268} 1258}
1269 1259
1270 1260