aboutsummaryrefslogtreecommitdiff
path: root/src/core/core_api.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/core_api.c')
-rw-r--r--src/core/core_api.c761
1 files changed, 267 insertions, 494 deletions
diff --git a/src/core/core_api.c b/src/core/core_api.c
index fda49e259..ace80b952 100644
--- a/src/core/core_api.c
+++ b/src/core/core_api.c
@@ -33,79 +33,32 @@
33 33
34 34
35/** 35/**
36 * Handle for a transmission request. 36 * Information we track for each peer.
37 */ 37 */
38struct GNUNET_CORE_TransmitHandle 38struct PeerRecord
39{ 39{
40 40
41 /** 41 /**
42 * Corresponding peer record. 42 * Corresponding CORE handle.
43 */
44 struct PeerRecord *peer;
45
46 /**
47 * Function that will be called to get the actual request
48 * (once we are ready to transmit this request to the core).
49 * The function will be called with a NULL buffer to signal
50 * timeout.
51 */
52 GNUNET_CONNECTION_TransmitReadyNotify get_message;
53
54 /**
55 * Closure for @e get_message.
56 */
57 void *get_message_cls;
58
59 /**
60 * Deadline for the transmission (the request does not get cancelled
61 * at this time, this is merely how soon the application wants this out).
62 */
63 struct GNUNET_TIME_Absolute deadline;
64
65 /**
66 * When did this request get queued?
67 */
68 struct GNUNET_TIME_Absolute request_time;
69
70 /**
71 * How important is this message?
72 */
73 enum GNUNET_CORE_Priority priority;
74
75 /**
76 * Is corking allowed?
77 */
78 int cork;
79
80 /**
81 * Size of this request.
82 */ 43 */
83 uint16_t msize; 44 struct GNUNET_CORE_Handle *h;
84 45
85 /** 46 /**
86 * Send message request ID for this request. 47 * Message queue for the peer.
87 */ 48 */
88 uint16_t smr_id; 49 struct GNUNET_MQ_Handle *mq;
89
90};
91
92
93/**
94 * Information we track for each peer.
95 */
96struct PeerRecord
97{
98 50
99 /** 51 /**
100 * Corresponding CORE handle. 52 * Message we are currently trying to pass to the CORE service
53 * for this peer (from @e mq).
101 */ 54 */
102 struct GNUNET_CORE_Handle *ch; 55 struct GNUNET_MQ_Envelope *env;
103 56
104 /** 57 /**
105 * Pending request, if any. 'th->peer' is set to NULL if the 58 * Value the client returned when we connected, used
106 * request is not active. 59 * as the closure in various places.
107 */ 60 */
108 struct GNUNET_CORE_TransmitHandle th; 61 void *client_cls;
109 62
110 /** 63 /**
111 * Peer the record is about. 64 * Peer the record is about.
@@ -152,19 +105,9 @@ struct GNUNET_CORE_Handle
152 GNUNET_CORE_DisconnectEventHandler disconnects; 105 GNUNET_CORE_DisconnectEventHandler disconnects;
153 106
154 /** 107 /**
155 * Function to call whenever we receive an inbound message.
156 */
157 GNUNET_CORE_MessageCallback inbound_notify;
158
159 /**
160 * Function to call whenever we receive an outbound message.
161 */
162 GNUNET_CORE_MessageCallback outbound_notify;
163
164 /**
165 * Function handlers for messages of particular type. 108 * Function handlers for messages of particular type.
166 */ 109 */
167 struct GNUNET_CORE_MessageHandler *handlers; 110 struct GNUNET_MQ_MessageHandler *handlers;
168 111
169 /** 112 /**
170 * Our message queue for transmissions to the service. 113 * Our message queue for transmissions to the service.
@@ -198,24 +141,6 @@ struct GNUNET_CORE_Handle
198 unsigned int hcnt; 141 unsigned int hcnt;
199 142
200 /** 143 /**
201 * For inbound notifications without a specific handler, do
202 * we expect to only receive headers?
203 */
204 int inbound_hdr_only;
205
206 /**
207 * For outbound notifications without a specific handler, do
208 * we expect to only receive headers?
209 */
210 int outbound_hdr_only;
211
212 /**
213 * Are we currently disconnected and hence unable to forward
214 * requests?
215 */
216 int currently_down;
217
218 /**
219 * Did we ever get INIT? 144 * Did we ever get INIT?
220 */ 145 */
221 int have_init; 146 int have_init;
@@ -266,25 +191,19 @@ disconnect_and_free_peer_entry (void *cls,
266 void *value) 191 void *value)
267{ 192{
268 struct GNUNET_CORE_Handle *h = cls; 193 struct GNUNET_CORE_Handle *h = cls;
269 struct GNUNET_CORE_TransmitHandle *th;
270 struct PeerRecord *pr = value; 194 struct PeerRecord *pr = value;
271 195
196 GNUNET_assert (pr->h == h);
272 if (NULL != h->disconnects) 197 if (NULL != h->disconnects)
273 h->disconnects (h->cls, 198 h->disconnects (h->cls,
274 &pr->peer); 199 &pr->peer,
275 /* all requests should have been cancelled, clean up anyway, just in case */ 200 pr->client_cls);
276 th = &pr->th;
277 if (NULL != th->peer)
278 {
279 GNUNET_break (0);
280 th->peer = NULL;
281 }
282 /* done with 'voluntary' cleanups, now on to normal freeing */
283 GNUNET_assert (GNUNET_YES == 201 GNUNET_assert (GNUNET_YES ==
284 GNUNET_CONTAINER_multipeermap_remove (h->peers, 202 GNUNET_CONTAINER_multipeermap_remove (h->peers,
285 key, 203 key,
286 pr)); 204 pr));
287 GNUNET_assert (pr->ch == h); 205 GNUNET_MQ_destroy (pr->mq);
206 GNUNET_assert (NULL == pr->mq);
288 GNUNET_free (pr); 207 GNUNET_free (pr);
289 return GNUNET_YES; 208 return GNUNET_YES;
290} 209}
@@ -305,8 +224,7 @@ reconnect_later (struct GNUNET_CORE_Handle *h)
305 GNUNET_MQ_destroy (h->mq); 224 GNUNET_MQ_destroy (h->mq);
306 h->mq = NULL; 225 h->mq = NULL;
307 } 226 }
308 h->currently_down = GNUNET_YES; 227 GNUNET_assert (NULL == h->reconnect_task);
309 GNUNET_assert (h->reconnect_task == NULL);
310 h->reconnect_task = 228 h->reconnect_task =
311 GNUNET_SCHEDULER_add_delayed (h->retry_backoff, 229 GNUNET_SCHEDULER_add_delayed (h->retry_backoff,
312 &reconnect_task, 230 &reconnect_task,
@@ -319,9 +237,8 @@ reconnect_later (struct GNUNET_CORE_Handle *h)
319 237
320 238
321/** 239/**
322 * Generic error handler, called with the appropriate error code and 240 * Error handler for the message queue to the CORE service.
323 * the same closure specified at the creation of the message queue. 241 * On errors, we reconnect.
324 * Not every message queue implementation supports an error handler.
325 * 242 *
326 * @param cls closure, a `struct GNUNET_CORE_Handle *` 243 * @param cls closure, a `struct GNUNET_CORE_Handle *`
327 * @param error error code 244 * @param error error code
@@ -340,6 +257,209 @@ handle_mq_error (void *cls,
340 257
341 258
342/** 259/**
260 * Inquire with CORE what options should be set for a message
261 * so that it is transmitted with the given @a priority and
262 * the given @a cork value.
263 *
264 * @param cork desired corking
265 * @param priority desired message priority
266 * @param[out] flags set to `flags` value for #GNUNET_MQ_set_options()
267 * @return `extra` argument to give to #GNUNET_MQ_set_options()
268 */
269const void *
270GNUNET_CORE_get_mq_options (int cork,
271 enum GNUNET_CORE_Priority priority,
272 uint64_t *flags)
273{
274 *flags = ((uint64_t) priority) + (((uint64_t) cork) << 32);
275 return NULL;
276}
277
278
279/**
280 * Implement sending functionality of a message queue for
281 * us sending messages to a peer.
282 *
283 * @param mq the message queue
284 * @param msg the message to send
285 * @param impl_state state of the implementation
286 */
287static void
288core_mq_send_impl (struct GNUNET_MQ_Handle *mq,
289 const struct GNUNET_MessageHeader *msg,
290 void *impl_state)
291{
292 struct PeerRecord *pr = impl_state;
293 struct GNUNET_CORE_Handle *h = pr->h;
294 struct SendMessageRequest *smr;
295 struct SendMessage *sm;
296 struct GNUNET_MQ_Envelope *env;
297 uint16_t msize;
298 uint64_t flags;
299 int cork;
300 enum GNUNET_CORE_Priority priority;
301
302 if (NULL == h->mq)
303 {
304 /* We're currently reconnecting, pretend this worked */
305 GNUNET_MQ_impl_send_continue (mq);
306 return;
307 }
308 GNUNET_assert (NULL == pr->env);
309 /* extract options from envelope */
310 env = GNUNET_MQ_get_current_envelope (mq);
311 GNUNET_break (NULL ==
312 GNUNET_MQ_env_get_options (env,
313 &flags));
314 cork = (int) (flags >> 32);
315 priority = (uint32_t) flags;
316
317 /* check message size for sanity */
318 msize = ntohs (msg->size);
319 if (msize >= GNUNET_SERVER_MAX_MESSAGE_SIZE - sizeof (struct SendMessage))
320 {
321 GNUNET_break (0);
322 GNUNET_MQ_impl_send_continue (mq);
323 return;
324 }
325
326 /* ask core for transmission */
327 LOG (GNUNET_ERROR_TYPE_DEBUG,
328 "Asking core for transmission of %u bytes to `%s'\n",
329 (unsigned int) msize,
330 GNUNET_i2s (&pr->peer));
331 env = GNUNET_MQ_msg (smr,
332 GNUNET_MESSAGE_TYPE_CORE_SEND_REQUEST);
333 smr->priority = htonl ((uint32_t) priority);
334 // smr->deadline = GNUNET_TIME_absolute_hton (deadline);
335 smr->peer = pr->peer;
336 smr->reserved = htonl (0);
337 smr->size = htons (msize);
338 smr->smr_id = htons (++pr->smr_id_gen);
339 GNUNET_MQ_send (h->mq,
340 env);
341
342 /* prepare message with actual transmission data */
343 pr->env = GNUNET_MQ_msg_nested_mh (sm,
344 GNUNET_MESSAGE_TYPE_CORE_SEND,
345 msg);
346 sm->priority = htonl ((uint32_t) priority);
347 // sm->deadline = GNUNET_TIME_absolute_hton (deadline);
348 sm->peer = pr->peer;
349 sm->cork = htonl ((uint32_t) cork);
350 sm->reserved = htonl (0);
351 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
352 "Calling get_message with buffer of %u bytes (%s)\n",
353 (unsigned int) msize,
354 cork ? "corked" : "uncorked");
355}
356
357
358/**
359 * Handle destruction of a message queue. Implementations must not
360 * free @a mq, but should take care of @a impl_state.
361 *
362 * @param mq the message queue to destroy
363 * @param impl_state state of the implementation
364 */
365static void
366core_mq_destroy_impl (struct GNUNET_MQ_Handle *mq,
367 void *impl_state)
368{
369 struct PeerRecord *pr = impl_state;
370
371 GNUNET_assert (mq == pr->mq);
372 pr->mq = NULL;
373}
374
375
376/**
377 * Implementation function that cancels the currently sent message.
378 * Should basically undo whatever #mq_send_impl() did.
379 *
380 * @param mq message queue
381 * @param impl_state state specific to the implementation
382 */
383static void
384core_mq_cancel_impl (struct GNUNET_MQ_Handle *mq,
385 void *impl_state)
386{
387 struct PeerRecord *pr = impl_state;
388
389 GNUNET_assert (NULL != pr->env);
390 GNUNET_MQ_discard (pr->env);
391 pr->env = NULL;
392}
393
394
395/**
396 * We had an error processing a message we forwarded from a peer to
397 * the CORE service. We should just complain about it but otherwise
398 * continue processing.
399 *
400 * @param cls closure
401 * @param error error code
402 */
403static void
404core_mq_error_handler (void *cls,
405 enum GNUNET_MQ_Error error)
406{
407 /* struct PeerRecord *pr = cls; */
408
409 GNUNET_break_op (0);
410}
411
412
413/**
414 * Add the given peer to the list of our connected peers
415 * and create the respective data structures and notify
416 * the application.
417 *
418 * @param h the core handle
419 * @param peer the peer that is connecting to us
420 */
421static void
422connect_peer (struct GNUNET_CORE_Handle *h,
423 const struct GNUNET_PeerIdentity *peer)
424{
425 struct PeerRecord *pr;
426 uint64_t flags;
427 const void *extra;
428
429 pr = GNUNET_new (struct PeerRecord);
430 pr->peer = *peer;
431 pr->h = h;
432 GNUNET_assert (GNUNET_YES ==
433 GNUNET_CONTAINER_multipeermap_put (h->peers,
434 &pr->peer,
435 pr,
436 GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
437 pr->mq = GNUNET_MQ_queue_for_callbacks (&core_mq_send_impl,
438 &core_mq_destroy_impl,
439 &core_mq_cancel_impl,
440 pr,
441 h->handlers,
442 &core_mq_error_handler,
443 pr);
444 /* get our default options */
445 extra = GNUNET_CORE_get_mq_options (GNUNET_NO,
446 GNUNET_CORE_PRIO_BEST_EFFORT,
447 &flags);
448 GNUNET_MQ_set_options (pr->mq,
449 flags,
450 extra);
451 if (NULL != h->connects)
452 {
453 pr->client_cls = h->connects (h->cls,
454 &pr->peer,
455 pr->mq);
456 GNUNET_MQ_set_handlers_closure (pr->mq,
457 pr->client_cls);
458 }
459}
460
461
462/**
343 * Handle init reply message received from CORE service. Notify 463 * Handle init reply message received from CORE service. Notify
344 * application that we are now connected to the CORE. Also fake 464 * application that we are now connected to the CORE. Also fake
345 * loopback connection. 465 * loopback connection.
@@ -353,11 +473,8 @@ handle_init_reply (void *cls,
353{ 473{
354 struct GNUNET_CORE_Handle *h = cls; 474 struct GNUNET_CORE_Handle *h = cls;
355 GNUNET_CORE_StartupCallback init; 475 GNUNET_CORE_StartupCallback init;
356 struct PeerRecord *pr;
357 476
358 GNUNET_break (0 == ntohl (m->reserved)); 477 GNUNET_break (0 == ntohl (m->reserved));
359 GNUNET_break (GNUNET_YES == h->currently_down);
360 h->currently_down = GNUNET_NO;
361 h->retry_backoff = GNUNET_TIME_UNIT_MILLISECONDS; 478 h->retry_backoff = GNUNET_TIME_UNIT_MILLISECONDS;
362 if (NULL != (init = h->init)) 479 if (NULL != (init = h->init))
363 { 480 {
@@ -388,17 +505,8 @@ handle_init_reply (void *cls,
388 } 505 }
389 } 506 }
390 /* fake 'connect to self' */ 507 /* fake 'connect to self' */
391 pr = GNUNET_new (struct PeerRecord); 508 connect_peer (h,
392 pr->peer = h->me; 509 &h->me);
393 pr->ch = h;
394 GNUNET_assert (GNUNET_YES ==
395 GNUNET_CONTAINER_multipeermap_put (h->peers,
396 &h->me,
397 pr,
398 GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
399 if (NULL != h->connects)
400 h->connects (h->cls,
401 &pr->peer);
402} 510}
403 511
404 512
@@ -416,7 +524,6 @@ handle_connect_notify (void *cls,
416 struct GNUNET_CORE_Handle *h = cls; 524 struct GNUNET_CORE_Handle *h = cls;
417 struct PeerRecord *pr; 525 struct PeerRecord *pr;
418 526
419 GNUNET_break (GNUNET_NO == h->currently_down);
420 LOG (GNUNET_ERROR_TYPE_DEBUG, 527 LOG (GNUNET_ERROR_TYPE_DEBUG,
421 "Received notification about connection from `%s'.\n", 528 "Received notification about connection from `%s'.\n",
422 GNUNET_i2s (&cnm->peer)); 529 GNUNET_i2s (&cnm->peer));
@@ -436,17 +543,8 @@ handle_connect_notify (void *cls,
436 reconnect_later (h); 543 reconnect_later (h);
437 return; 544 return;
438 } 545 }
439 pr = GNUNET_new (struct PeerRecord); 546 connect_peer (h,
440 pr->peer = cnm->peer; 547 &cnm->peer);
441 pr->ch = h;
442 GNUNET_assert (GNUNET_YES ==
443 GNUNET_CONTAINER_multipeermap_put (h->peers,
444 &cnm->peer,
445 pr,
446 GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
447 if (NULL != h->connects)
448 h->connects (h->cls,
449 &pr->peer);
450} 548}
451 549
452 550
@@ -459,17 +557,16 @@ handle_connect_notify (void *cls,
459 */ 557 */
460static void 558static void
461handle_disconnect_notify (void *cls, 559handle_disconnect_notify (void *cls,
462 const struct DisconnectNotifyMessage * dnm) 560 const struct DisconnectNotifyMessage *dnm)
463{ 561{
464 struct GNUNET_CORE_Handle *h = cls; 562 struct GNUNET_CORE_Handle *h = cls;
465 struct PeerRecord *pr; 563 struct PeerRecord *pr;
466 564
467 GNUNET_break (GNUNET_NO == h->currently_down);
468 if (0 == memcmp (&h->me, 565 if (0 == memcmp (&h->me,
469 &dnm->peer, 566 &dnm->peer,
470 sizeof (struct GNUNET_PeerIdentity))) 567 sizeof (struct GNUNET_PeerIdentity)))
471 { 568 {
472 /* connection to self!? */ 569 /* disconnect from self!? */
473 GNUNET_break (0); 570 GNUNET_break (0);
474 return; 571 return;
475 } 572 }
@@ -486,7 +583,7 @@ handle_disconnect_notify (void *cls,
486 return; 583 return;
487 } 584 }
488 disconnect_and_free_peer_entry (h, 585 disconnect_and_free_peer_entry (h,
489 &dnm->peer, 586 &pr->peer,
490 pr); 587 pr);
491} 588}
492 589
@@ -502,11 +599,9 @@ static int
502check_notify_inbound (void *cls, 599check_notify_inbound (void *cls,
503 const struct NotifyTrafficMessage *ntm) 600 const struct NotifyTrafficMessage *ntm)
504{ 601{
505 struct GNUNET_CORE_Handle *h = cls;
506 uint16_t msize; 602 uint16_t msize;
507 const struct GNUNET_MessageHeader *em; 603 const struct GNUNET_MessageHeader *em;
508 604
509 GNUNET_break (GNUNET_NO == h->currently_down);
510 msize = ntohs (ntm->header.size) - sizeof (struct NotifyTrafficMessage); 605 msize = ntohs (ntm->header.size) - sizeof (struct NotifyTrafficMessage);
511 if (msize < sizeof (struct GNUNET_MessageHeader)) 606 if (msize < sizeof (struct GNUNET_MessageHeader))
512 { 607 {
@@ -514,8 +609,7 @@ check_notify_inbound (void *cls,
514 return GNUNET_SYSERR; 609 return GNUNET_SYSERR;
515 } 610 }
516 em = (const struct GNUNET_MessageHeader *) &ntm[1]; 611 em = (const struct GNUNET_MessageHeader *) &ntm[1];
517 if ( (GNUNET_NO == h->inbound_hdr_only) && 612 if (msize != ntohs (em->size))
518 (msize != ntohs (em->size)) )
519 { 613 {
520 GNUNET_break (0); 614 GNUNET_break (0);
521 return GNUNET_SYSERR; 615 return GNUNET_SYSERR;
@@ -538,120 +632,21 @@ handle_notify_inbound (void *cls,
538 struct GNUNET_CORE_Handle *h = cls; 632 struct GNUNET_CORE_Handle *h = cls;
539 const struct GNUNET_MessageHeader *em; 633 const struct GNUNET_MessageHeader *em;
540 struct PeerRecord *pr; 634 struct PeerRecord *pr;
541 uint16_t et;
542 635
543 GNUNET_break (GNUNET_NO == h->currently_down);
544 em = (const struct GNUNET_MessageHeader *) &ntm[1];
545 et = ntohs (em->type);
546 LOG (GNUNET_ERROR_TYPE_DEBUG, 636 LOG (GNUNET_ERROR_TYPE_DEBUG,
547 "Received inbound message of type %d from `%s'.\n", 637 "Received inbound message from `%s'.\n",
548 (int) et,
549 GNUNET_i2s (&ntm->peer)); 638 GNUNET_i2s (&ntm->peer));
550 for (unsigned int hpos = 0; NULL != h->handlers[hpos].callback; hpos++)
551 {
552 const struct GNUNET_CORE_MessageHandler *mh;
553
554 mh = &h->handlers[hpos];
555 if (mh->type != et)
556 continue;
557 if ( (mh->expected_size != ntohs (em->size)) &&
558 (0 != mh->expected_size) )
559 {
560 LOG (GNUNET_ERROR_TYPE_ERROR,
561 "Unexpected message size %u for message of type %u from peer `%s'\n",
562 htons (em->size),
563 mh->type,
564 GNUNET_i2s (&ntm->peer));
565 GNUNET_break_op (0);
566 continue;
567 }
568 pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
569 &ntm->peer);
570 if (NULL == pr)
571 {
572 GNUNET_break (0);
573 reconnect_later (h);
574 return;
575 }
576 if (GNUNET_OK !=
577 h->handlers[hpos].callback (h->cls,
578 &ntm->peer,
579 em))
580 {
581 /* error in processing, do not process other messages! */
582 break;
583 }
584 }
585 if (NULL != h->inbound_notify)
586 h->inbound_notify (h->cls,
587 &ntm->peer,
588 em);
589}
590
591
592/**
593 * Check that message received from CORE service is well-formed.
594 *
595 * @param cls the `struct GNUNET_CORE_Handle`
596 * @param ntm the message we got
597 * @return #GNUNET_OK if the message is well-formed
598 */
599static int
600check_notify_outbound (void *cls,
601 const struct NotifyTrafficMessage *ntm)
602{
603 struct GNUNET_CORE_Handle *h = cls;
604 uint16_t msize;
605 const struct GNUNET_MessageHeader *em;
606
607 GNUNET_break (GNUNET_NO == h->currently_down);
608 LOG (GNUNET_ERROR_TYPE_DEBUG,
609 "Received outbound message from `%s'.\n",
610 GNUNET_i2s (&ntm->peer));
611 msize = ntohs (ntm->header.size) - sizeof (struct NotifyTrafficMessage);
612 if (msize < sizeof (struct GNUNET_MessageHeader))
613 {
614 GNUNET_break (0);
615 return GNUNET_SYSERR;
616 }
617 em = (const struct GNUNET_MessageHeader *) &ntm[1];
618 if ( (GNUNET_NO == h->outbound_hdr_only) &&
619 (msize != ntohs (em->size)) )
620 {
621 GNUNET_break (0);
622 return GNUNET_SYSERR;
623 }
624 return GNUNET_OK;
625}
626
627
628/**
629 * Handle outbound message received from CORE service. If applicable,
630 * notify the application.
631 *
632 * @param cls the `struct GNUNET_CORE_Handle`
633 * @param ntm the message we got
634 */
635static void
636handle_notify_outbound (void *cls,
637 const struct NotifyTrafficMessage *ntm)
638{
639 struct GNUNET_CORE_Handle *h = cls;
640 const struct GNUNET_MessageHeader *em;
641
642 GNUNET_break (GNUNET_NO == h->currently_down);
643 em = (const struct GNUNET_MessageHeader *) &ntm[1]; 639 em = (const struct GNUNET_MessageHeader *) &ntm[1];
644 LOG (GNUNET_ERROR_TYPE_DEBUG, 640 pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
645 "Received notification about transmission to `%s'.\n", 641 &ntm->peer);
646 GNUNET_i2s (&ntm->peer)); 642 if (NULL == pr)
647 if (NULL == h->outbound_notify)
648 { 643 {
649 GNUNET_break (0); 644 GNUNET_break (0);
645 reconnect_later (h);
650 return; 646 return;
651 } 647 }
652 h->outbound_notify (h->cls, 648 GNUNET_MQ_inject_message (pr->mq,
653 &ntm->peer, 649 em);
654 em);
655} 650}
656 651
657 652
@@ -661,7 +656,7 @@ handle_notify_outbound (void *cls,
661 * pending, put it into the queue to be transmitted. 656 * pending, put it into the queue to be transmitted.
662 * 657 *
663 * @param cls the `struct GNUNET_CORE_Handle` 658 * @param cls the `struct GNUNET_CORE_Handle`
664 * @param ntm the message we got 659 * @param smr the message we got
665 */ 660 */
666static void 661static void
667handle_send_ready (void *cls, 662handle_send_ready (void *cls,
@@ -669,16 +664,7 @@ handle_send_ready (void *cls,
669{ 664{
670 struct GNUNET_CORE_Handle *h = cls; 665 struct GNUNET_CORE_Handle *h = cls;
671 struct PeerRecord *pr; 666 struct PeerRecord *pr;
672 struct GNUNET_CORE_TransmitHandle *th;
673 struct SendMessage *sm;
674 struct GNUNET_MQ_Envelope *env;
675 struct GNUNET_TIME_Relative delay;
676 struct GNUNET_TIME_Relative overdue;
677 unsigned int ret;
678 unsigned int priority;
679 int cork;
680 667
681 GNUNET_break (GNUNET_NO == h->currently_down);
682 pr = GNUNET_CONTAINER_multipeermap_get (h->peers, 668 pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
683 &smr->peer); 669 &smr->peer);
684 if (NULL == pr) 670 if (NULL == pr)
@@ -690,72 +676,24 @@ handle_send_ready (void *cls,
690 LOG (GNUNET_ERROR_TYPE_DEBUG, 676 LOG (GNUNET_ERROR_TYPE_DEBUG,
691 "Received notification about transmission readiness to `%s'.\n", 677 "Received notification about transmission readiness to `%s'.\n",
692 GNUNET_i2s (&smr->peer)); 678 GNUNET_i2s (&smr->peer));
693 if (NULL == pr->th.peer) 679 if (NULL == pr->env)
694 { 680 {
695 /* request must have been cancelled between the original request 681 /* request must have been cancelled between the original request
696 * and the response from CORE, ignore CORE's readiness */ 682 * and the response from CORE, ignore CORE's readiness */
697 return; 683 return;
698 } 684 }
699 th = &pr->th; 685 if (ntohs (smr->smr_id) != pr->smr_id_gen)
700 if (ntohs (smr->smr_id) != th->smr_id)
701 { 686 {
702 /* READY message is for expired or cancelled message, 687 /* READY message is for expired or cancelled message,
703 * ignore! (we should have already sent another request) */ 688 * ignore! (we should have already sent another request) */
704 return; 689 return;
705 } 690 }
691
706 /* ok, all good, send message out! */ 692 /* ok, all good, send message out! */
707 th->peer = NULL;
708 env = GNUNET_MQ_msg_extra (sm,
709 th->msize,
710 GNUNET_MESSAGE_TYPE_CORE_SEND);
711 sm->priority = htonl ((uint32_t) th->priority);
712 sm->deadline = GNUNET_TIME_absolute_hton (th->deadline);
713 sm->peer = pr->peer;
714 sm->cork = htonl ((uint32_t) (cork = th->cork));
715 sm->reserved = htonl (0);
716 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
717 "Calling get_message with buffer of %u bytes (%s)\n",
718 (unsigned int) th->msize,
719 cork ? "corked" : "uncorked");
720 /* FIXME: this is ugly and a bit brutal, but "get_message"
721 may call GNUNET_CORE_notify_transmit_ready() which
722 may call GNUNET_MQ_send() as well, and we MUST get this
723 message out before the next SEND_REQUEST. So we queue
724 it (even though incomplete) and then---relying on MQ being
725 nice and not actually touching 'env' until much later---
726 fill it afterwards. This is horrible style, and once
727 the core_api abandons GNUNET_CORE_notify_transmit_ready
728 in favor of an MQ-style API, this hack should no longer
729 be required */
730 GNUNET_MQ_send (h->mq, 693 GNUNET_MQ_send (h->mq,
731 env); 694 pr->env);
732 delay = GNUNET_TIME_absolute_get_duration (th->request_time); 695 pr->env = NULL;
733 overdue = GNUNET_TIME_absolute_get_duration (th->deadline); 696 GNUNET_MQ_impl_send_continue (pr->mq);
734 priority = th->priority;
735 ret = th->get_message (th->get_message_cls,
736 th->msize,
737 &sm[1]);
738 /* after this point, 'th' should not be used anymore, it
739 may now be about another message! */
740 sm->header.size = htons (ret + sizeof (struct SendMessage));
741 if (overdue.rel_value_us > GNUNET_CONSTANTS_LATENCY_WARN.rel_value_us)
742 LOG (GNUNET_ERROR_TYPE_WARNING,
743 "Transmitting overdue %u bytes to `%s' at priority %u with %s delay %s\n",
744 ret,
745 GNUNET_i2s (&pr->peer),
746 priority,
747 GNUNET_STRINGS_relative_time_to_string (delay,
748 GNUNET_YES),
749 (cork) ? " (corked)" : " (uncorked)");
750 else
751 LOG (GNUNET_ERROR_TYPE_DEBUG,
752 "Transmitting %u bytes to `%s' at priority %u with %s delay %s\n",
753 ret,
754 GNUNET_i2s (&pr->peer),
755 priority,
756 GNUNET_STRINGS_relative_time_to_string (delay,
757 GNUNET_YES),
758 (cork) ? " (corked)" : " (uncorked)");
759} 697}
760 698
761 699
@@ -785,10 +723,6 @@ reconnect (struct GNUNET_CORE_Handle *h)
785 GNUNET_MESSAGE_TYPE_CORE_NOTIFY_INBOUND, 723 GNUNET_MESSAGE_TYPE_CORE_NOTIFY_INBOUND,
786 struct NotifyTrafficMessage, 724 struct NotifyTrafficMessage,
787 h), 725 h),
788 GNUNET_MQ_hd_var_size (notify_outbound,
789 GNUNET_MESSAGE_TYPE_CORE_NOTIFY_OUTBOUND,
790 struct NotifyTrafficMessage,
791 h),
792 GNUNET_MQ_hd_fixed_size (send_ready, 726 GNUNET_MQ_hd_fixed_size (send_ready,
793 GNUNET_MESSAGE_TYPE_CORE_SEND_READY, 727 GNUNET_MESSAGE_TYPE_CORE_SEND_READY,
794 struct SendMessageReady, 728 struct SendMessageReady,
@@ -797,12 +731,10 @@ reconnect (struct GNUNET_CORE_Handle *h)
797 }; 731 };
798 struct InitMessage *init; 732 struct InitMessage *init;
799 struct GNUNET_MQ_Envelope *env; 733 struct GNUNET_MQ_Envelope *env;
800 uint32_t opt;
801 uint16_t *ts; 734 uint16_t *ts;
802 735
803 GNUNET_assert (NULL == h->mq); 736 GNUNET_assert (NULL == h->mq);
804 GNUNET_assert (GNUNET_YES == h->currently_down); 737 h->mq = GNUNET_CLIENT_connect (h->cfg,
805 h->mq = GNUNET_CLIENT_connecT (h->cfg,
806 "core", 738 "core",
807 handlers, 739 handlers,
808 &handle_mq_error, 740 &handle_mq_error,
@@ -815,25 +747,9 @@ reconnect (struct GNUNET_CORE_Handle *h)
815 env = GNUNET_MQ_msg_extra (init, 747 env = GNUNET_MQ_msg_extra (init,
816 sizeof (uint16_t) * h->hcnt, 748 sizeof (uint16_t) * h->hcnt,
817 GNUNET_MESSAGE_TYPE_CORE_INIT); 749 GNUNET_MESSAGE_TYPE_CORE_INIT);
818 opt = GNUNET_CORE_OPTION_NOTHING;
819 if (NULL != h->inbound_notify)
820 {
821 if (h->inbound_hdr_only)
822 opt |= GNUNET_CORE_OPTION_SEND_HDR_INBOUND;
823 else
824 opt |= GNUNET_CORE_OPTION_SEND_FULL_INBOUND;
825 }
826 if (NULL != h->outbound_notify)
827 {
828 if (h->outbound_hdr_only)
829 opt |= GNUNET_CORE_OPTION_SEND_HDR_OUTBOUND;
830 else
831 opt |= GNUNET_CORE_OPTION_SEND_FULL_OUTBOUND;
832 }
833 LOG (GNUNET_ERROR_TYPE_INFO, 750 LOG (GNUNET_ERROR_TYPE_INFO,
834 "(Re)connecting to CORE service, monitoring messages of type %u\n", 751 "(Re)connecting to CORE service\n");
835 opt); 752 init->options = htonl (0);
836 init->options = htonl (opt);
837 ts = (uint16_t *) &init[1]; 753 ts = (uint16_t *) &init[1];
838 for (unsigned int hpos = 0; hpos < h->hcnt; hpos++) 754 for (unsigned int hpos = 0; hpos < h->hcnt; hpos++)
839 ts[hpos] = htons (h->handlers[hpos].type); 755 ts[hpos] = htons (h->handlers[hpos].type);
@@ -852,14 +768,6 @@ reconnect (struct GNUNET_CORE_Handle *h)
852 * connected to the core service 768 * connected to the core service
853 * @param connects function to call on peer connect, can be NULL 769 * @param connects function to call on peer connect, can be NULL
854 * @param disconnects function to call on peer disconnect / timeout, can be NULL 770 * @param disconnects function to call on peer disconnect / timeout, can be NULL
855 * @param inbound_notify function to call for all inbound messages, can be NULL
856 * @param inbound_hdr_only set to #GNUNET_YES if inbound_notify will only read the
857 * GNUNET_MessageHeader and hence we do not need to give it the full message;
858 * can be used to improve efficiency, ignored if @a inbound_notify is NULL
859 * @param outbound_notify function to call for all outbound messages, can be NULL
860 * @param outbound_hdr_only set to #GNUNET_YES if outbound_notify will only read the
861 * GNUNET_MessageHeader and hence we do not need to give it the full message
862 * can be used to improve efficiency, ignored if @a outbound_notify is NULL
863 * @param handlers callbacks for messages we care about, NULL-terminated 771 * @param handlers callbacks for messages we care about, NULL-terminated
864 * @return handle to the core service (only useful for disconnect until @a init is called); 772 * @return handle to the core service (only useful for disconnect until @a init is called);
865 * NULL on error (in this case, init is never called) 773 * NULL on error (in this case, init is never called)
@@ -870,11 +778,7 @@ GNUNET_CORE_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
870 GNUNET_CORE_StartupCallback init, 778 GNUNET_CORE_StartupCallback init,
871 GNUNET_CORE_ConnectEventHandler connects, 779 GNUNET_CORE_ConnectEventHandler connects,
872 GNUNET_CORE_DisconnectEventHandler disconnects, 780 GNUNET_CORE_DisconnectEventHandler disconnects,
873 GNUNET_CORE_MessageCallback inbound_notify, 781 const struct GNUNET_MQ_MessageHandler *handlers)
874 int inbound_hdr_only,
875 GNUNET_CORE_MessageCallback outbound_notify,
876 int outbound_hdr_only,
877 const struct GNUNET_CORE_MessageHandler *handlers)
878{ 782{
879 struct GNUNET_CORE_Handle *h; 783 struct GNUNET_CORE_Handle *h;
880 unsigned int hcnt; 784 unsigned int hcnt;
@@ -885,22 +789,18 @@ GNUNET_CORE_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
885 h->init = init; 789 h->init = init;
886 h->connects = connects; 790 h->connects = connects;
887 h->disconnects = disconnects; 791 h->disconnects = disconnects;
888 h->inbound_notify = inbound_notify; 792 h->peers = GNUNET_CONTAINER_multipeermap_create (128,
889 h->outbound_notify = outbound_notify; 793 GNUNET_NO);
890 h->inbound_hdr_only = inbound_hdr_only;
891 h->outbound_hdr_only = outbound_hdr_only;
892 h->currently_down = GNUNET_YES;
893 h->peers = GNUNET_CONTAINER_multipeermap_create (128, GNUNET_NO);
894 hcnt = 0; 794 hcnt = 0;
895 if (NULL != handlers) 795 if (NULL != handlers)
896 while (NULL != handlers[hcnt].callback) 796 while (NULL != handlers[hcnt].cb)
897 hcnt++; 797 hcnt++;
898 h->handlers = GNUNET_new_array (hcnt + 1, 798 h->handlers = GNUNET_new_array (hcnt + 1,
899 struct GNUNET_CORE_MessageHandler); 799 struct GNUNET_MQ_MessageHandler);
900 if (NULL != handlers) 800 if (NULL != handlers)
901 GNUNET_memcpy (h->handlers, 801 GNUNET_memcpy (h->handlers,
902 handlers, 802 handlers,
903 hcnt * sizeof (struct GNUNET_CORE_MessageHandler)); 803 hcnt * sizeof (struct GNUNET_MQ_MessageHandler));
904 h->hcnt = hcnt; 804 h->hcnt = hcnt;
905 GNUNET_assert (hcnt < 805 GNUNET_assert (hcnt <
906 (GNUNET_SERVER_MAX_MESSAGE_SIZE - 806 (GNUNET_SERVER_MAX_MESSAGE_SIZE -
@@ -918,9 +818,7 @@ GNUNET_CORE_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
918 818
919 819
920/** 820/**
921 * Disconnect from the core service. This function can only 821 * Disconnect from the core service.
922 * be called *after* all pending #GNUNET_CORE_notify_transmit_ready()
923 * requests have been explicitly canceled.
924 * 822 *
925 * @param handle connection to core to disconnect 823 * @param handle connection to core to disconnect
926 */ 824 */
@@ -950,148 +848,23 @@ GNUNET_CORE_disconnect (struct GNUNET_CORE_Handle *handle)
950 848
951 849
952/** 850/**
953 * Ask the core to call @a notify once it is ready to transmit the 851 * Obtain the message queue for a connected peer.
954 * given number of bytes to the specified @a target. Must only be
955 * called after a connection to the respective peer has been
956 * established (and the client has been informed about this). You may
957 * have one request of this type pending for each connected peer at
958 * any time. If a peer disconnects, the application MUST call
959 * #GNUNET_CORE_notify_transmit_ready_cancel on the respective
960 * transmission request, if one such request is pending.
961 * 852 *
962 * @param handle connection to core service 853 * @param h the core handle
963 * @param cork is corking allowed for this transmission? 854 * @param pid the identity of the peer to check if it has been connected to us
964 * @param priority how important is the message? 855 * @return NULL if peer is not connected
965 * @param maxdelay how long can the message wait? Only effective if @a cork is #GNUNET_YES
966 * @param target who should receive the message, never NULL (can be this peer's identity for loopback)
967 * @param notify_size how many bytes of buffer space does @a notify want?
968 * @param notify function to call when buffer space is available;
969 * will be called with NULL on timeout; clients MUST cancel
970 * all pending transmission requests DURING the disconnect
971 * handler
972 * @param notify_cls closure for @a notify
973 * @return non-NULL if the notify callback was queued,
974 * NULL if we can not even queue the request (request already pending);
975 * if NULL is returned, @a notify will NOT be called.
976 */ 856 */
977struct GNUNET_CORE_TransmitHandle * 857struct GNUNET_MQ_Handle *
978GNUNET_CORE_notify_transmit_ready (struct GNUNET_CORE_Handle *handle, 858GNUNET_CORE_get_mq (const struct GNUNET_CORE_Handle *h,
979 int cork, 859 const struct GNUNET_PeerIdentity *pid)
980 enum GNUNET_CORE_Priority priority,
981 struct GNUNET_TIME_Relative maxdelay,
982 const struct GNUNET_PeerIdentity *target,
983 size_t notify_size,
984 GNUNET_CONNECTION_TransmitReadyNotify notify,
985 void *notify_cls)
986{ 860{
987 struct PeerRecord *pr; 861 struct PeerRecord *pr;
988 struct GNUNET_CORE_TransmitHandle *th;
989 struct SendMessageRequest *smr;
990 struct GNUNET_MQ_Envelope *env;
991 862
992 if (NULL == handle->mq) 863 pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
993 { 864 pid);
994 GNUNET_break (0); /* SEE #4588: do not call NTR from disconnect notification! */
995 return NULL;
996 }
997 GNUNET_assert (NULL != notify);
998 if ( (notify_size > GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE) ||
999 (notify_size + sizeof (struct SendMessage) >= GNUNET_SERVER_MAX_MESSAGE_SIZE) )
1000 {
1001 GNUNET_break (0);
1002 return NULL;
1003 }
1004 LOG (GNUNET_ERROR_TYPE_DEBUG,
1005 "Asking core for transmission of %u bytes to `%s'%s\n",
1006 (unsigned int) notify_size,
1007 GNUNET_i2s (target),
1008 cork ? " (corked)" : "");
1009 pr = GNUNET_CONTAINER_multipeermap_get (handle->peers,
1010 target);
1011 if (NULL == pr) 865 if (NULL == pr)
1012 {
1013 /* attempt to send to peer that is not connected */
1014 GNUNET_break (0);
1015 return NULL; 866 return NULL;
1016 } 867 return pr->mq;
1017 if (NULL != pr->th.peer)
1018 {
1019 /* attempting to queue a second request for the same destination */
1020 GNUNET_break (0);
1021 return NULL;
1022 }
1023 th = &pr->th;
1024 memset (th,
1025 0,
1026 sizeof (struct GNUNET_CORE_TransmitHandle));
1027 th->peer = pr;
1028 th->get_message = notify;
1029 th->get_message_cls = notify_cls;
1030 th->request_time = GNUNET_TIME_absolute_get ();
1031 if (GNUNET_YES == cork)
1032 th->deadline = GNUNET_TIME_relative_to_absolute (maxdelay);
1033 else
1034 th->deadline = th->request_time;
1035 th->priority = priority;
1036 th->msize = notify_size;
1037 th->cork = cork;
1038 if (NULL == handle->mq)
1039 return th; /* see #4588 (hack until we transition core fully to MQ) */
1040 env = GNUNET_MQ_msg (smr,
1041 GNUNET_MESSAGE_TYPE_CORE_SEND_REQUEST);
1042 smr->priority = htonl ((uint32_t) th->priority);
1043 smr->deadline = GNUNET_TIME_absolute_hton (th->deadline);
1044 smr->peer = pr->peer;
1045 smr->reserved = htonl (0);
1046 smr->size = htons (th->msize);
1047 smr->smr_id = htons (th->smr_id = pr->smr_id_gen++);
1048 GNUNET_MQ_send (handle->mq,
1049 env);
1050 LOG (GNUNET_ERROR_TYPE_DEBUG,
1051 "Transmission request added to queue\n");
1052 return th;
1053}
1054
1055
1056/**
1057 * Cancel the specified transmission-ready notification.
1058 *
1059 * @param th handle that was returned by #GNUNET_CORE_notify_transmit_ready().
1060 */
1061void
1062GNUNET_CORE_notify_transmit_ready_cancel (struct GNUNET_CORE_TransmitHandle *th)
1063{
1064 struct PeerRecord *pr = th->peer;
1065
1066 LOG (GNUNET_ERROR_TYPE_DEBUG,
1067 "Aborting transmission request to core for %u bytes to `%s'\n",
1068 (unsigned int) th->msize,
1069 GNUNET_i2s (&pr->peer));
1070 th->peer = NULL;
1071}
1072
1073
1074/**
1075 * Check if the given peer is currently connected. This function is for special
1076 * cirumstances (GNUNET_TESTBED uses it), normal users of the CORE API are
1077 * expected to track which peers are connected based on the connect/disconnect
1078 * callbacks from #GNUNET_CORE_connect(). This function is NOT part of the
1079 * 'versioned', 'official' API. The difference between this function and the
1080 * function GNUNET_CORE_is_peer_connected() is that this one returns
1081 * synchronously after looking in the CORE API cache. The function
1082 * GNUNET_CORE_is_peer_connected() sends a message to the CORE service and hence
1083 * its response is given asynchronously.
1084 *
1085 * @param h the core handle
1086 * @param pid the identity of the peer to check if it has been connected to us
1087 * @return #GNUNET_YES if the peer is connected to us; #GNUNET_NO if not
1088 */
1089int
1090GNUNET_CORE_is_peer_connected_sync (const struct GNUNET_CORE_Handle *h,
1091 const struct GNUNET_PeerIdentity *pid)
1092{
1093 return GNUNET_CONTAINER_multipeermap_contains (h->peers,
1094 pid);
1095} 868}
1096 869
1097 870
generated by cgit v1.2.3 (git 2.39.1) at 2024-06-11 20:57:02 +0000