summaryrefslogtreecommitdiff
path: root/src/core
diff options
context:
space:
mode:
Diffstat (limited to 'src/core')
-rw-r--r--src/core/.gitignore7
-rw-r--r--src/core/Makefile.am1
-rw-r--r--src/core/core_api.c761
-rw-r--r--src/core/core_api_2.c865
-rw-r--r--src/core/core_api_monitor_peers.c2
-rw-r--r--src/core/core_api_mq.c191
-rw-r--r--src/core/test_core_api.c6
-rw-r--r--src/core/test_core_api_reliability.c6
-rw-r--r--src/core/test_core_api_send_to_self.c4
-rw-r--r--src/core/test_core_api_start_only.c12
-rw-r--r--src/core/test_core_defaults.conf6
-rw-r--r--src/core/test_core_quota_compliance.c6
12 files changed, 292 insertions, 1575 deletions
diff --git a/src/core/.gitignore b/src/core/.gitignore
index 42b7030b3..cdd1f93c2 100644
--- a/src/core/.gitignore
+++ b/src/core/.gitignore
@@ -1,2 +1,9 @@
1gnunet-service-core 1gnunet-service-core
2gnunet-core 2gnunet-core
3test_core_api
4test_core_api_reliability
5test_core_api_send_to_self
6test_core_api_start_only
7test_core_quota_compliance_asymmetric_recv_limited
8test_core_quota_compliance_asymmetric_send_limited
9test_core_quota_compliance_symmetric
diff --git a/src/core/Makefile.am b/src/core/Makefile.am
index aea64fa34..ed80bae73 100644
--- a/src/core/Makefile.am
+++ b/src/core/Makefile.am
@@ -23,7 +23,6 @@ lib_LTLIBRARIES = \
23 23
24libgnunetcore_la_SOURCES = \ 24libgnunetcore_la_SOURCES = \
25 core_api.c core.h \ 25 core_api.c core.h \
26 core_api_2.c \
27 core_api_monitor_peers.c 26 core_api_monitor_peers.c
28libgnunetcore_la_LIBADD = \ 27libgnunetcore_la_LIBADD = \
29 $(top_builddir)/src/util/libgnunetutil.la \ 28 $(top_builddir)/src/util/libgnunetutil.la \
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
diff --git a/src/core/core_api_2.c b/src/core/core_api_2.c
deleted file mode 100644
index d810bf2ec..000000000
--- a/src/core/core_api_2.c
+++ /dev/null
@@ -1,865 +0,0 @@
1/*
2 This file is part of GNUnet.
3 Copyright (C) 2009-2016 GNUnet e.V.
4
5 GNUnet is free software; you can redistribute it and/or modify
6 it under the terms of the GNU General Public License as published
7 by the Free Software Foundation; either version 3, or (at your
8 option) any later version.
9
10 GNUnet is distributed in the hope that it will be useful, but
11 WITHOUT ANY WARRANTY; without even the implied warranty of
12 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13 General Public License for more details.
14
15 You should have received a copy of the GNU General Public License
16 along with GNUnet; see the file COPYING. If not, write to the
17 Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
18 Boston, MA 02110-1301, USA.
19*/
20/**
21 * @file core/core_api_2.c
22 * @brief core service; this is the main API for encrypted P2P
23 * communications
24 * @author Christian Grothoff
25 */
26#include "platform.h"
27#include "gnunet_util_lib.h"
28#include "gnunet_constants.h"
29#include "gnunet_core_service.h"
30#include "core.h"
31
32#define LOG(kind,...) GNUNET_log_from (kind, "core-api",__VA_ARGS__)
33
34
35/**
36 * Information we track for each peer.
37 */
38struct PeerRecord
39{
40
41 /**
42 * Corresponding CORE handle.
43 */
44 struct GNUNET_CORE_Handle *h;
45
46 /**
47 * Message queue for the peer.
48 */
49 struct GNUNET_MQ_Handle *mq;
50
51 /**
52 * Message we are currently trying to pass to the CORE service
53 * for this peer (from @e mq).
54 */
55 struct GNUNET_MQ_Envelope *env;
56
57 /**
58 * Value the client returned when we connected, used
59 * as the closure in various places.
60 */
61 void *client_cls;
62
63 /**
64 * Peer the record is about.
65 */
66 struct GNUNET_PeerIdentity peer;
67
68 /**
69 * SendMessageRequest ID generator for this peer.
70 */
71 uint16_t smr_id_gen;
72
73};
74
75
76/**
77 * Context for the core service connection.
78 */
79struct GNUNET_CORE_Handle
80{
81
82 /**
83 * Configuration we're using.
84 */
85 const struct GNUNET_CONFIGURATION_Handle *cfg;
86
87 /**
88 * Closure for the various callbacks.
89 */
90 void *cls;
91
92 /**
93 * Function to call once we've handshaked with the core service.
94 */
95 GNUNET_CORE_StartupCallback init;
96
97 /**
98 * Function to call whenever we're notified about a peer connecting.
99 */
100 GNUNET_CORE_ConnecTEventHandler connects;
101
102 /**
103 * Function to call whenever we're notified about a peer disconnecting.
104 */
105 GNUNET_CORE_DisconnecTEventHandler disconnects;
106
107 /**
108 * Function handlers for messages of particular type.
109 */
110 struct GNUNET_MQ_MessageHandler *handlers;
111
112 /**
113 * Our message queue for transmissions to the service.
114 */
115 struct GNUNET_MQ_Handle *mq;
116
117 /**
118 * Hash map listing all of the peers that we are currently
119 * connected to.
120 */
121 struct GNUNET_CONTAINER_MultiPeerMap *peers;
122
123 /**
124 * Identity of this peer.
125 */
126 struct GNUNET_PeerIdentity me;
127
128 /**
129 * ID of reconnect task (if any).
130 */
131 struct GNUNET_SCHEDULER_Task *reconnect_task;
132
133 /**
134 * Current delay we use for re-trying to connect to core.
135 */
136 struct GNUNET_TIME_Relative retry_backoff;
137
138 /**
139 * Number of entries in the handlers array.
140 */
141 unsigned int hcnt;
142
143 /**
144 * Did we ever get INIT?
145 */
146 int have_init;
147
148};
149
150
151/**
152 * Our current client connection went down. Clean it up
153 * and try to reconnect!
154 *
155 * @param h our handle to the core service
156 */
157static void
158reconnect (struct GNUNET_CORE_Handle *h);
159
160
161/**
162 * Task schedule to try to re-connect to core.
163 *
164 * @param cls the `struct GNUNET_CORE_Handle`
165 * @param tc task context
166 */
167static void
168reconnect_task (void *cls)
169{
170 struct GNUNET_CORE_Handle *h = cls;
171
172 h->reconnect_task = NULL;
173 LOG (GNUNET_ERROR_TYPE_DEBUG,
174 "Connecting to CORE service after delay\n");
175 reconnect (h);
176}
177
178
179/**
180 * Notify clients about disconnect and free the entry for connected
181 * peer.
182 *
183 * @param cls the `struct GNUNET_CORE_Handle *`
184 * @param key the peer identity (not used)
185 * @param value the `struct PeerRecord` to free.
186 * @return #GNUNET_YES (continue)
187 */
188static int
189disconnect_and_free_peer_entry (void *cls,
190 const struct GNUNET_PeerIdentity *key,
191 void *value)
192{
193 struct GNUNET_CORE_Handle *h = cls;
194 struct PeerRecord *pr = value;
195
196 GNUNET_assert (pr->h == h);
197 if (NULL != h->disconnects)
198 h->disconnects (h->cls,
199 &pr->peer,
200 pr->client_cls);
201 GNUNET_assert (GNUNET_YES ==
202 GNUNET_CONTAINER_multipeermap_remove (h->peers,
203 key,
204 pr));
205 GNUNET_MQ_destroy (pr->mq);
206 GNUNET_assert (NULL == pr->mq);
207 GNUNET_free (pr);
208 return GNUNET_YES;
209}
210
211
212/**
213 * Close down any existing connection to the CORE service and
214 * try re-establishing it later.
215 *
216 * @param h our handle
217 */
218static void
219reconnect_later (struct GNUNET_CORE_Handle *h)
220{
221 GNUNET_assert (NULL == h->reconnect_task);
222 if (NULL != h->mq)
223 {
224 GNUNET_MQ_destroy (h->mq);
225 h->mq = NULL;
226 }
227 GNUNET_assert (NULL == h->reconnect_task);
228 h->reconnect_task =
229 GNUNET_SCHEDULER_add_delayed (h->retry_backoff,
230 &reconnect_task,
231 h);
232 GNUNET_CONTAINER_multipeermap_iterate (h->peers,
233 &disconnect_and_free_peer_entry,
234 h);
235 h->retry_backoff = GNUNET_TIME_STD_BACKOFF (h->retry_backoff);
236}
237
238
239/**
240 * Error handler for the message queue to the CORE service.
241 * On errors, we reconnect.
242 *
243 * @param cls closure, a `struct GNUNET_CORE_Handle *`
244 * @param error error code
245 */
246static void
247handle_mq_error (void *cls,
248 enum GNUNET_MQ_Error error)
249{
250 struct GNUNET_CORE_Handle *h = cls;
251
252 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
253 "MQ ERROR: %d\n",
254 error);
255 reconnect_later (h);
256}
257
258
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 GNUNET_assert (NULL == pr->env);
303 /* extract options from envelope */
304 env = GNUNET_MQ_get_current_envelope (mq);
305 GNUNET_break (NULL ==
306 GNUNET_MQ_env_get_options (env,
307 &flags));
308 cork = (int) (flags >> 32);
309 priority = (uint32_t) flags;
310
311 /* check message size for sanity */
312 msize = ntohs (msg->size);
313 if (msize >= GNUNET_SERVER_MAX_MESSAGE_SIZE - sizeof (struct SendMessage))
314 {
315 GNUNET_break (0);
316 GNUNET_MQ_impl_send_continue (mq);
317 return;
318 }
319
320 /* ask core for transmission */
321 LOG (GNUNET_ERROR_TYPE_DEBUG,
322 "Asking core for transmission of %u bytes to `%s'\n",
323 (unsigned int) msize,
324 GNUNET_i2s (&pr->peer));
325 env = GNUNET_MQ_msg (smr,
326 GNUNET_MESSAGE_TYPE_CORE_SEND_REQUEST);
327 smr->priority = htonl ((uint32_t) priority);
328 // smr->deadline = GNUNET_TIME_absolute_hton (deadline);
329 smr->peer = pr->peer;
330 smr->reserved = htonl (0);
331 smr->size = htons (msize);
332 smr->smr_id = htons (++pr->smr_id_gen);
333 GNUNET_MQ_send (h->mq,
334 env);
335
336 /* prepare message with actual transmission data */
337 pr->env = GNUNET_MQ_msg_nested_mh (sm,
338 GNUNET_MESSAGE_TYPE_CORE_SEND,
339 msg);
340 sm->priority = htonl ((uint32_t) priority);
341 // sm->deadline = GNUNET_TIME_absolute_hton (deadline);
342 sm->peer = pr->peer;
343 sm->cork = htonl ((uint32_t) cork);
344 sm->reserved = htonl (0);
345 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
346 "Calling get_message with buffer of %u bytes (%s)\n",
347 (unsigned int) msize,
348 cork ? "corked" : "uncorked");
349}
350
351
352/**
353 * Handle destruction of a message queue. Implementations must not
354 * free @a mq, but should take care of @a impl_state.
355 *
356 * @param mq the message queue to destroy
357 * @param impl_state state of the implementation
358 */
359static void
360core_mq_destroy_impl (struct GNUNET_MQ_Handle *mq,
361 void *impl_state)
362{
363 struct PeerRecord *pr = impl_state;
364
365 GNUNET_assert (mq == pr->mq);
366 pr->mq = NULL;
367}
368
369
370/**
371 * Implementation function that cancels the currently sent message.
372 * Should basically undo whatever #mq_send_impl() did.
373 *
374 * @param mq message queue
375 * @param impl_state state specific to the implementation
376 */
377static void
378core_mq_cancel_impl (struct GNUNET_MQ_Handle *mq,
379 void *impl_state)
380{
381 struct PeerRecord *pr = impl_state;
382
383 GNUNET_assert (NULL != pr->env);
384 GNUNET_MQ_discard (pr->env);
385 pr->env = NULL;
386}
387
388
389/**
390 * We had an error processing a message we forwarded from a peer to
391 * the CORE service. We should just complain about it but otherwise
392 * continue processing.
393 *
394 * @param cls closure
395 * @param error error code
396 */
397static void
398core_mq_error_handler (void *cls,
399 enum GNUNET_MQ_Error error)
400{
401 /* struct PeerRecord *pr = cls; */
402
403 GNUNET_break_op (0);
404}
405
406
407/**
408 * Add the given peer to the list of our connected peers
409 * and create the respective data structures and notify
410 * the application.
411 *
412 * @param h the core handle
413 * @param peer the peer that is connecting to us
414 */
415static void
416connect_peer (struct GNUNET_CORE_Handle *h,
417 const struct GNUNET_PeerIdentity *peer)
418{
419 struct PeerRecord *pr;
420 uint64_t flags;
421 const void *extra;
422
423 pr = GNUNET_new (struct PeerRecord);
424 pr->peer = *peer;
425 pr->h = h;
426 GNUNET_assert (GNUNET_YES ==
427 GNUNET_CONTAINER_multipeermap_put (h->peers,
428 &pr->peer,
429 pr,
430 GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
431 pr->mq = GNUNET_MQ_queue_for_callbacks (&core_mq_send_impl,
432 &core_mq_destroy_impl,
433 &core_mq_cancel_impl,
434 pr,
435 h->handlers,
436 &core_mq_error_handler,
437 pr);
438 /* get our default options */
439 extra = GNUNET_CORE_get_mq_options (GNUNET_NO,
440 GNUNET_CORE_PRIO_BEST_EFFORT,
441 &flags);
442 GNUNET_MQ_set_options (pr->mq,
443 flags,
444 extra);
445 if (NULL != h->connects)
446 {
447 pr->client_cls = h->connects (h->cls,
448 &pr->peer,
449 pr->mq);
450 GNUNET_MQ_set_handlers_closure (pr->mq,
451 pr->client_cls);
452 }
453}
454
455
456/**
457 * Handle init reply message received from CORE service. Notify
458 * application that we are now connected to the CORE. Also fake
459 * loopback connection.
460 *
461 * @param cls the `struct GNUNET_CORE_Handle`
462 * @param m the init reply
463 */
464static void
465handle_init_reply (void *cls,
466 const struct InitReplyMessage *m)
467{
468 struct GNUNET_CORE_Handle *h = cls;
469 GNUNET_CORE_StartupCallback init;
470
471 GNUNET_break (0 == ntohl (m->reserved));
472 h->retry_backoff = GNUNET_TIME_UNIT_MILLISECONDS;
473 if (NULL != (init = h->init))
474 {
475 /* mark so we don't call init on reconnect */
476 h->init = NULL;
477 h->me = m->my_identity;
478 LOG (GNUNET_ERROR_TYPE_DEBUG,
479 "Connected to core service of peer `%s'.\n",
480 GNUNET_i2s (&h->me));
481 h->have_init = GNUNET_YES;
482 init (h->cls,
483 &h->me);
484 }
485 else
486 {
487 LOG (GNUNET_ERROR_TYPE_DEBUG,
488 "Successfully reconnected to core service.\n");
489 if (GNUNET_NO == h->have_init)
490 {
491 h->me = m->my_identity;
492 h->have_init = GNUNET_YES;
493 }
494 else
495 {
496 GNUNET_break (0 == memcmp (&h->me,
497 &m->my_identity,
498 sizeof (struct GNUNET_PeerIdentity)));
499 }
500 }
501 /* fake 'connect to self' */
502 connect_peer (h,
503 &h->me);
504}
505
506
507/**
508 * Handle connect message received from CORE service.
509 * Notify the application about the new connection.
510 *
511 * @param cls the `struct GNUNET_CORE_Handle`
512 * @param cnm the connect message
513 */
514static void
515handle_connect_notify (void *cls,
516 const struct ConnectNotifyMessage *cnm)
517{
518 struct GNUNET_CORE_Handle *h = cls;
519 struct PeerRecord *pr;
520
521 LOG (GNUNET_ERROR_TYPE_DEBUG,
522 "Received notification about connection from `%s'.\n",
523 GNUNET_i2s (&cnm->peer));
524 if (0 == memcmp (&h->me,
525 &cnm->peer,
526 sizeof (struct GNUNET_PeerIdentity)))
527 {
528 /* connect to self!? */
529 GNUNET_break (0);
530 return;
531 }
532 pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
533 &cnm->peer);
534 if (NULL != pr)
535 {
536 GNUNET_break (0);
537 reconnect_later (h);
538 return;
539 }
540 connect_peer (h,
541 &cnm->peer);
542}
543
544
545/**
546 * Handle disconnect message received from CORE service.
547 * Notify the application about the lost connection.
548 *
549 * @param cls the `struct GNUNET_CORE_Handle`
550 * @param dnm message about the disconnect event
551 */
552static void
553handle_disconnect_notify (void *cls,
554 const struct DisconnectNotifyMessage *dnm)
555{
556 struct GNUNET_CORE_Handle *h = cls;
557 struct PeerRecord *pr;
558
559 if (0 == memcmp (&h->me,
560 &dnm->peer,
561 sizeof (struct GNUNET_PeerIdentity)))
562 {
563 /* disconnect from self!? */
564 GNUNET_break (0);
565 return;
566 }
567 GNUNET_break (0 == ntohl (dnm->reserved));
568 LOG (GNUNET_ERROR_TYPE_DEBUG,
569 "Received notification about disconnect from `%s'.\n",
570 GNUNET_i2s (&dnm->peer));
571 pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
572 &dnm->peer);
573 if (NULL == pr)
574 {
575 GNUNET_break (0);
576 reconnect_later (h);
577 return;
578 }
579 disconnect_and_free_peer_entry (h,
580 &pr->peer,
581 pr);
582}
583
584
585/**
586 * Check that message received from CORE service is well-formed.
587 *
588 * @param cls the `struct GNUNET_CORE_Handle`
589 * @param ntm the message we got
590 * @return #GNUNET_OK if the message is well-formed
591 */
592static int
593check_notify_inbound (void *cls,
594 const struct NotifyTrafficMessage *ntm)
595{
596 uint16_t msize;
597 const struct GNUNET_MessageHeader *em;
598
599 msize = ntohs (ntm->header.size) - sizeof (struct NotifyTrafficMessage);
600 if (msize < sizeof (struct GNUNET_MessageHeader))
601 {
602 GNUNET_break (0);
603 return GNUNET_SYSERR;
604 }
605 em = (const struct GNUNET_MessageHeader *) &ntm[1];
606 if (msize != ntohs (em->size))
607 {
608 GNUNET_break (0);
609 return GNUNET_SYSERR;
610 }
611 return GNUNET_OK;
612}
613
614
615/**
616 * Handle inbound message received from CORE service. If applicable,
617 * notify the application.
618 *
619 * @param cls the `struct GNUNET_CORE_Handle`
620 * @param ntm the message we got from CORE.
621 */
622static void
623handle_notify_inbound (void *cls,
624 const struct NotifyTrafficMessage *ntm)
625{
626 struct GNUNET_CORE_Handle *h = cls;
627 const struct GNUNET_MessageHeader *em;
628 struct PeerRecord *pr;
629
630 LOG (GNUNET_ERROR_TYPE_DEBUG,
631 "Received inbound message from `%s'.\n",
632 GNUNET_i2s (&ntm->peer));
633 em = (const struct GNUNET_MessageHeader *) &ntm[1];
634 pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
635 &ntm->peer);
636 if (NULL == pr)
637 {
638 GNUNET_break (0);
639 reconnect_later (h);
640 return;
641 }
642 GNUNET_MQ_inject_message (pr->mq,
643 em);
644}
645
646
647/**
648 * Handle message received from CORE service notifying us that we are
649 * now allowed to send a message to a peer. If that message is still
650 * pending, put it into the queue to be transmitted.
651 *
652 * @param cls the `struct GNUNET_CORE_Handle`
653 * @param smr the message we got
654 */
655static void
656handle_send_ready (void *cls,
657 const struct SendMessageReady *smr)
658{
659 struct GNUNET_CORE_Handle *h = cls;
660 struct PeerRecord *pr;
661
662 pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
663 &smr->peer);
664 if (NULL == pr)
665 {
666 GNUNET_break (0);
667 reconnect_later (h);
668 return;
669 }
670 LOG (GNUNET_ERROR_TYPE_DEBUG,
671 "Received notification about transmission readiness to `%s'.\n",
672 GNUNET_i2s (&smr->peer));
673 if (NULL == pr->env)
674 {
675 /* request must have been cancelled between the original request
676 * and the response from CORE, ignore CORE's readiness */
677 return;
678 }
679 if (ntohs (smr->smr_id) != pr->smr_id_gen)
680 {
681 /* READY message is for expired or cancelled message,
682 * ignore! (we should have already sent another request) */
683 return;
684 }
685
686 /* ok, all good, send message out! */
687 GNUNET_MQ_send (h->mq,
688 pr->env);
689 pr->env = NULL;
690 GNUNET_MQ_impl_send_continue (pr->mq);
691}
692
693
694/**
695 * Our current client connection went down. Clean it up and try to
696 * reconnect!
697 *
698 * @param h our handle to the core service
699 */
700static void
701reconnect (struct GNUNET_CORE_Handle *h)
702{
703 struct GNUNET_MQ_MessageHandler handlers[] = {
704 GNUNET_MQ_hd_fixed_size (init_reply,
705 GNUNET_MESSAGE_TYPE_CORE_INIT_REPLY,
706 struct InitReplyMessage,
707 h),
708 GNUNET_MQ_hd_fixed_size (connect_notify,
709 GNUNET_MESSAGE_TYPE_CORE_NOTIFY_CONNECT,
710 struct ConnectNotifyMessage,
711 h),
712 GNUNET_MQ_hd_fixed_size (disconnect_notify,
713 GNUNET_MESSAGE_TYPE_CORE_NOTIFY_DISCONNECT,
714 struct DisconnectNotifyMessage,
715 h),
716 GNUNET_MQ_hd_var_size (notify_inbound,
717 GNUNET_MESSAGE_TYPE_CORE_NOTIFY_INBOUND,
718 struct NotifyTrafficMessage,
719 h),
720 GNUNET_MQ_hd_fixed_size (send_ready,
721 GNUNET_MESSAGE_TYPE_CORE_SEND_READY,
722 struct SendMessageReady,
723 h),
724 GNUNET_MQ_handler_end ()
725 };
726 struct InitMessage *init;
727 struct GNUNET_MQ_Envelope *env;
728 uint16_t *ts;
729
730 GNUNET_assert (NULL == h->mq);
731 h->mq = GNUNET_CLIENT_connecT (h->cfg,
732 "core",
733 handlers,
734 &handle_mq_error,
735 h);
736 if (NULL == h->mq)
737 {
738 reconnect_later (h);
739 return;
740 }
741 env = GNUNET_MQ_msg_extra (init,
742 sizeof (uint16_t) * h->hcnt,
743 GNUNET_MESSAGE_TYPE_CORE_INIT);
744 LOG (GNUNET_ERROR_TYPE_INFO,
745 "(Re)connecting to CORE service\n");
746 init->options = htonl (0);
747 ts = (uint16_t *) &init[1];
748 for (unsigned int hpos = 0; hpos < h->hcnt; hpos++)
749 ts[hpos] = htons (h->handlers[hpos].type);
750 GNUNET_MQ_send (h->mq,
751 env);
752}
753
754
755/**
756 * Connect to the core service. Note that the connection may complete
757 * (or fail) asynchronously.
758 *
759 * @param cfg configuration to use
760 * @param cls closure for the various callbacks that follow (including handlers in the handlers array)
761 * @param init callback to call once we have successfully
762 * connected to the core service
763 * @param connects function to call on peer connect, can be NULL
764 * @param disconnects function to call on peer disconnect / timeout, can be NULL
765 * @param handlers callbacks for messages we care about, NULL-terminated
766 * @return handle to the core service (only useful for disconnect until @a init is called);
767 * NULL on error (in this case, init is never called)
768 */
769struct GNUNET_CORE_Handle *
770GNUNET_CORE_connecT (const struct GNUNET_CONFIGURATION_Handle *cfg,
771 void *cls,
772 GNUNET_CORE_StartupCallback init,
773 GNUNET_CORE_ConnecTEventHandler connects,
774 GNUNET_CORE_DisconnecTEventHandler disconnects,
775 const struct GNUNET_MQ_MessageHandler *handlers)
776{
777 struct GNUNET_CORE_Handle *h;
778 unsigned int hcnt;
779
780 h = GNUNET_new (struct GNUNET_CORE_Handle);
781 h->cfg = cfg;
782 h->cls = cls;
783 h->init = init;
784 h->connects = connects;
785 h->disconnects = disconnects;
786 h->peers = GNUNET_CONTAINER_multipeermap_create (128,
787 GNUNET_NO);
788 hcnt = 0;
789 if (NULL != handlers)
790 while (NULL != handlers[hcnt].cb)
791 hcnt++;
792 h->handlers = GNUNET_new_array (hcnt + 1,
793 struct GNUNET_MQ_MessageHandler);
794 if (NULL != handlers)
795 GNUNET_memcpy (h->handlers,
796 handlers,
797 hcnt * sizeof (struct GNUNET_MQ_MessageHandler));
798 h->hcnt = hcnt;
799 GNUNET_assert (hcnt <
800 (GNUNET_SERVER_MAX_MESSAGE_SIZE -
801 sizeof (struct InitMessage)) / sizeof (uint16_t));
802 LOG (GNUNET_ERROR_TYPE_DEBUG,
803 "Connecting to CORE service\n");
804 reconnect (h);
805 if (NULL == h->mq)
806 {
807 GNUNET_CORE_disconnect (h);
808 return NULL;
809 }
810 return h;
811}
812
813
814/**
815 * Disconnect from the core service.
816 *
817 * @param handle connection to core to disconnect
818 */
819void
820GNUNET_CORE_disconnecT (struct GNUNET_CORE_Handle *handle)
821{
822 LOG (GNUNET_ERROR_TYPE_DEBUG,
823 "Disconnecting from CORE service\n");
824 GNUNET_CONTAINER_multipeermap_iterate (handle->peers,
825 &disconnect_and_free_peer_entry,
826 handle);
827 GNUNET_CONTAINER_multipeermap_destroy (handle->peers);
828 handle->peers = NULL;
829 if (NULL != handle->reconnect_task)
830 {
831 GNUNET_SCHEDULER_cancel (handle->reconnect_task);
832 handle->reconnect_task = NULL;
833 }
834 if (NULL != handle->mq)
835 {
836 GNUNET_MQ_destroy (handle->mq);
837 handle->mq = NULL;
838 }
839 GNUNET_free (handle->handlers);
840 GNUNET_free (handle);
841}
842
843
844/**
845 * Obtain the message queue for a connected peer.
846 *
847 * @param h the core handle
848 * @param pid the identity of the peer to check if it has been connected to us
849 * @return NULL if peer is not connected
850 */
851struct GNUNET_MQ_Handle *
852GNUNET_CORE_get_mq (const struct GNUNET_CORE_Handle *h,
853 const struct GNUNET_PeerIdentity *pid)
854{
855 struct PeerRecord *pr;
856
857 pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
858 pid);
859 if (NULL == pr)
860 return NULL;
861 return pr->mq;
862}
863
864
865/* end of core_api.c */
diff --git a/src/core/core_api_monitor_peers.c b/src/core/core_api_monitor_peers.c
index 1455eb2b0..796fdb9d5 100644
--- a/src/core/core_api_monitor_peers.c
+++ b/src/core/core_api_monitor_peers.c
@@ -127,7 +127,7 @@ reconnect (struct GNUNET_CORE_MonitorHandle *mh)
127 if (NULL != mh->mq) 127 if (NULL != mh->mq)
128 GNUNET_MQ_destroy (mh->mq); 128 GNUNET_MQ_destroy (mh->mq);
129 /* FIXME: use backoff? */ 129 /* FIXME: use backoff? */
130 mh->mq = GNUNET_CLIENT_connecT (mh->cfg, 130 mh->mq = GNUNET_CLIENT_connect (mh->cfg,
131 "core", 131 "core",
132 handlers, 132 handlers,
133 &handle_mq_error, 133 &handle_mq_error,
diff --git a/src/core/core_api_mq.c b/src/core/core_api_mq.c
deleted file mode 100644
index 12c7a3bdd..000000000
--- a/src/core/core_api_mq.c
+++ /dev/null
@@ -1,191 +0,0 @@
1/*
2 This file is part of GNUnet.
3 Copyright (C) 2009-2014 GNUnet e.V.
4
5 GNUnet is free software; you can redistribute it and/or modify
6 it under the terms of the GNU General Public License as published
7 by the Free Software Foundation; either version 3, or (at your
8 option) any later version.
9
10 GNUnet is distributed in the hope that it will be useful, but
11 WITHOUT ANY WARRANTY; without even the implied warranty of
12 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13 General Public License for more details.
14
15 You should have received a copy of the GNU General Public License
16 along with GNUnet; see the file COPYING. If not, write to the
17 Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
18 Boston, MA 02110-1301, USA.
19*/
20/**
21 * @file core/core_api_mq.c
22 * @brief MQ support for core service
23 * @author Christian Grothoff
24 * @author Florian Dold
25 */
26#include "platform.h"
27#include "gnunet_util_lib.h"
28#include "gnunet_constants.h"
29#include "gnunet_core_service.h"
30#include "core.h"
31
32#define LOG(kind,...) GNUNET_log_from (kind, "core-api",__VA_ARGS__)
33
34
35/**
36 * Internal state of a GNUNET-MQ queue for CORE.
37 */
38struct CoreMQState
39{
40 /**
41 * Which peer does this queue target?
42 */
43 struct GNUNET_PeerIdentity target;
44
45 /**
46 * Handle to the CORE service used by this MQ.
47 */
48 struct GNUNET_CORE_Handle *core;
49
50 /**
51 * Transmission handle (if in use).
52 */
53 struct GNUNET_CORE_TransmitHandle *th;
54};
55
56
57/**
58 * Function called to notify a client about the connection
59 * begin ready to queue more data. @a buf will be
60 * NULL and @a size zero if the connection was closed for
61 * writing in the meantime.
62 *
63 * @param cls closure
64 * @param size number of bytes available in @a buf
65 * @param buf where the callee should write the message
66 * @return number of bytes written to @a buf
67 */
68static size_t
69core_mq_ntr (void *cls, size_t size,
70 void *buf)
71{
72 struct GNUNET_MQ_Handle *mq = cls;
73 struct CoreMQState *mqs = GNUNET_MQ_impl_state (mq);
74 const struct GNUNET_MessageHeader *mh = GNUNET_MQ_impl_current (mq);
75 size_t msg_size = ntohs (mh->size);
76
77 LOG (GNUNET_ERROR_TYPE_DEBUG,
78 "ntr called (size %u, type %u)\n",
79 msg_size,
80 ntohs (mh->type));
81 mqs->th = NULL;
82 if (NULL == buf)
83 {
84 LOG (GNUNET_ERROR_TYPE_DEBUG,
85 "send error\n");
86 GNUNET_MQ_inject_error (mq, GNUNET_MQ_ERROR_WRITE);
87 return 0;
88 }
89 GNUNET_memcpy (buf, mh, msg_size);
90 GNUNET_MQ_impl_send_continue (mq);
91 return msg_size;
92}
93
94
95/**
96 * Signature of functions implementing the
97 * sending functionality of a message queue.
98 *
99 * @param mq the message queue
100 * @param msg the message to send
101 * @param impl_state state of the implementation
102 */
103static void
104core_mq_send (struct GNUNET_MQ_Handle *mq,
105 const struct GNUNET_MessageHeader *msg,
106 void *impl_state)
107{
108 struct CoreMQState *mqs = impl_state;
109
110 GNUNET_assert (NULL == mqs->th);
111 LOG (GNUNET_ERROR_TYPE_DEBUG,
112 "Sending queued message (size %u)\n",
113 ntohs (msg->size));
114 mqs->th = GNUNET_CORE_notify_transmit_ready (mqs->core, GNUNET_YES, 0,
115 GNUNET_TIME_UNIT_FOREVER_REL,
116 &mqs->target,
117 ntohs (msg->size),
118 &core_mq_ntr, mq);
119}
120
121
122/**
123 * Signature of functions implementing the
124 * destruction of a message queue.
125 * Implementations must not free @a mq, but should
126 * take care of @a impl_state.
127 *
128 * @param mq the message queue to destroy
129 * @param impl_state state of the implementation
130 */
131static void
132core_mq_destroy (struct GNUNET_MQ_Handle *mq,
133 void *impl_state)
134{
135 struct CoreMQState *mqs = impl_state;
136
137 if (NULL != mqs->th)
138 {
139 GNUNET_CORE_notify_transmit_ready_cancel (mqs->th);
140 mqs->th = NULL;
141 }
142 GNUNET_free (mqs);
143}
144
145
146/**
147 * Implementation function that cancels the currently sent message.
148 *
149 * @param mq message queue
150 * @param impl_state state specific to the implementation
151 */
152static void
153core_mq_cancel (struct GNUNET_MQ_Handle *mq,
154 void *impl_state)
155{
156 struct CoreMQState *mqs = impl_state;
157
158 GNUNET_assert (NULL != mqs->th);
159 GNUNET_CORE_notify_transmit_ready_cancel (mqs->th);
160}
161
162
163/**
164 * Create a message queue for sending messages to a peer with CORE.
165 * Messages may only be queued with #GNUNET_MQ_send once the init callback has
166 * been called for the given handle.
167 * There must only be one queue per peer for each core handle.
168 * The message queue can only be used to transmit messages,
169 * not to receive them.
170 *
171 * @param h the core handle
172 * @param target the target peer for this queue, may not be NULL
173 * @return a message queue for sending messages over the core handle
174 * to the target peer
175 */
176struct GNUNET_MQ_Handle *
177GNUNET_CORE_mq_create (struct GNUNET_CORE_Handle *h,
178 const struct GNUNET_PeerIdentity *target)
179{
180 struct CoreMQState *mqs = GNUNET_new (struct CoreMQState);
181
182 mqs->core = h;
183 mqs->target = *target;
184 return GNUNET_MQ_queue_for_callbacks (&core_mq_send,
185 &core_mq_destroy,
186 &core_mq_cancel,
187 mqs,
188 NULL, NULL, NULL);
189}
190
191/* end of core_api_mq.c */
diff --git a/src/core/test_core_api.c b/src/core/test_core_api.c
index e6a113b52..847ba6e75 100644
--- a/src/core/test_core_api.c
+++ b/src/core/test_core_api.c
@@ -92,7 +92,7 @@ terminate_peer (struct PeerContext *p)
92{ 92{
93 if (NULL != p->ch) 93 if (NULL != p->ch)
94 { 94 {
95 GNUNET_CORE_disconnecT (p->ch); 95 GNUNET_CORE_disconnect (p->ch);
96 p->ch = NULL; 96 p->ch = NULL;
97 } 97 }
98 if (NULL != p->ghh) 98 if (NULL != p->ghh)
@@ -243,7 +243,7 @@ init_notify (void *cls,
243 GNUNET_assert (ok == 2); 243 GNUNET_assert (ok == 2);
244 OKPP; 244 OKPP;
245 /* connect p2 */ 245 /* connect p2 */
246 p2.ch = GNUNET_CORE_connecT (p2.cfg, 246 p2.ch = GNUNET_CORE_connect (p2.cfg,
247 &p2, 247 &p2,
248 &init_notify, 248 &init_notify,
249 &connect_notify, 249 &connect_notify,
@@ -317,7 +317,7 @@ run (void *cls,
317 (GNUNET_TIME_UNIT_SECONDS, 300), 317 (GNUNET_TIME_UNIT_SECONDS, 300),
318 &terminate_task_error, NULL); 318 &terminate_task_error, NULL);
319 p1.ch = 319 p1.ch =
320 GNUNET_CORE_connecT (p1.cfg, 320 GNUNET_CORE_connect (p1.cfg,
321 &p1, 321 &p1,
322 &init_notify, 322 &init_notify,
323 &connect_notify, 323 &connect_notify,
diff --git a/src/core/test_core_api_reliability.c b/src/core/test_core_api_reliability.c
index cd2bcad83..900c9f732 100644
--- a/src/core/test_core_api_reliability.c
+++ b/src/core/test_core_api_reliability.c
@@ -103,7 +103,7 @@ terminate_peer (struct PeerContext *p)
103{ 103{
104 if (NULL != p->ch) 104 if (NULL != p->ch)
105 { 105 {
106 GNUNET_CORE_disconnecT (p->ch); 106 GNUNET_CORE_disconnect (p->ch);
107 p->ch = NULL; 107 p->ch = NULL;
108 } 108 }
109 if (NULL != p->ghh) 109 if (NULL != p->ghh)
@@ -341,7 +341,7 @@ init_notify (void *cls,
341 OKPP; 341 OKPP;
342 /* connect p2 */ 342 /* connect p2 */
343 GNUNET_assert (NULL != 343 GNUNET_assert (NULL !=
344 (p2.ch = GNUNET_CORE_connecT (p2.cfg, 344 (p2.ch = GNUNET_CORE_connect (p2.cfg,
345 &p2, 345 &p2,
346 &init_notify, 346 &init_notify,
347 &connect_notify, 347 &connect_notify,
@@ -464,7 +464,7 @@ run (void *cls,
464 NULL); 464 NULL);
465 465
466 GNUNET_assert (NULL != 466 GNUNET_assert (NULL !=
467 (p1.ch = GNUNET_CORE_connecT (p1.cfg, 467 (p1.ch = GNUNET_CORE_connect (p1.cfg,
468 &p1, 468 &p1,
469 &init_notify, 469 &init_notify,
470 &connect_notify, 470 &connect_notify,
diff --git a/src/core/test_core_api_send_to_self.c b/src/core/test_core_api_send_to_self.c
index d29da651b..5cfc8b35f 100644
--- a/src/core/test_core_api_send_to_self.c
+++ b/src/core/test_core_api_send_to_self.c
@@ -65,7 +65,7 @@ cleanup (void *cls)
65 } 65 }
66 if (NULL != core) 66 if (NULL != core)
67 { 67 {
68 GNUNET_CORE_disconnecT (core); 68 GNUNET_CORE_disconnect (core);
69 core = NULL; 69 core = NULL;
70 } 70 }
71 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, 71 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
@@ -159,7 +159,7 @@ run (void *cls,
159 }; 159 };
160 160
161 core = 161 core =
162 GNUNET_CORE_connecT (cfg, 162 GNUNET_CORE_connect (cfg,
163 NULL, 163 NULL,
164 &init, 164 &init,
165 &connect_cb, 165 &connect_cb,
diff --git a/src/core/test_core_api_start_only.c b/src/core/test_core_api_start_only.c
index 6abc3cc89..31e300b14 100644
--- a/src/core/test_core_api_start_only.c
+++ b/src/core/test_core_api_start_only.c
@@ -74,9 +74,9 @@ static struct GNUNET_MQ_MessageHandler handlers[] = {
74static void 74static void
75shutdown_task (void *cls) 75shutdown_task (void *cls)
76{ 76{
77 GNUNET_CORE_disconnecT (p1.ch); 77 GNUNET_CORE_disconnect (p1.ch);
78 p1.ch = NULL; 78 p1.ch = NULL;
79 GNUNET_CORE_disconnecT (p2.ch); 79 GNUNET_CORE_disconnect (p2.ch);
80 p2.ch = NULL; 80 p2.ch = NULL;
81 ok = 0; 81 ok = 0;
82} 82}
@@ -91,7 +91,7 @@ init_notify (void *cls,
91 if (p == &p1) 91 if (p == &p1)
92 { 92 {
93 /* connect p2 */ 93 /* connect p2 */
94 p2.ch = GNUNET_CORE_connecT (p2.cfg, 94 p2.ch = GNUNET_CORE_connect (p2.cfg,
95 &p2, 95 &p2,
96 &init_notify, 96 &init_notify,
97 &connect_notify, 97 &connect_notify,
@@ -140,12 +140,12 @@ timeout_task (void *cls)
140 "Timeout.\n"); 140 "Timeout.\n");
141 if (NULL != p1.ch) 141 if (NULL != p1.ch)
142 { 142 {
143 GNUNET_CORE_disconnecT (p1.ch); 143 GNUNET_CORE_disconnect (p1.ch);
144 p1.ch = NULL; 144 p1.ch = NULL;
145 } 145 }
146 if (NULL != p2.ch) 146 if (NULL != p2.ch)
147 { 147 {
148 GNUNET_CORE_disconnecT (p2.ch); 148 GNUNET_CORE_disconnect (p2.ch);
149 p2.ch = NULL; 149 p2.ch = NULL;
150 } 150 }
151 ok = 42; 151 ok = 42;
@@ -168,7 +168,7 @@ run (void *cls,
168 TIMEOUT), 168 TIMEOUT),
169 &timeout_task, 169 &timeout_task,
170 NULL); 170 NULL);
171 p1.ch = GNUNET_CORE_connecT (p1.cfg, 171 p1.ch = GNUNET_CORE_connect (p1.cfg,
172 &p1, 172 &p1,
173 &init_notify, 173 &init_notify,
174 &connect_notify, 174 &connect_notify,
diff --git a/src/core/test_core_defaults.conf b/src/core/test_core_defaults.conf
index eb9a1c379..4c956987c 100644
--- a/src/core/test_core_defaults.conf
+++ b/src/core/test_core_defaults.conf
@@ -5,13 +5,7 @@
5GNUNET_TEST_HOME = /tmp/test-gnunet-core/ 5GNUNET_TEST_HOME = /tmp/test-gnunet-core/
6 6
7[nat] 7[nat]
8DISABLEV6 = YES
9ENABLE_UPNP = NO 8ENABLE_UPNP = NO
10BEHIND_NAT = NO
11ALLOW_NAT = NO
12INTERNAL_ADDRESS = 127.0.0.1
13EXTERNAL_ADDRESS = 127.0.0.1
14USE_LOCALADDR = NO
15 9
16[ats] 10[ats]
17WAN_QUOTA_IN = 1 GB 11WAN_QUOTA_IN = 1 GB
diff --git a/src/core/test_core_quota_compliance.c b/src/core/test_core_quota_compliance.c
index 4dee958f2..dcc33288d 100644
--- a/src/core/test_core_quota_compliance.c
+++ b/src/core/test_core_quota_compliance.c
@@ -117,7 +117,7 @@ terminate_peer (struct PeerContext *p)
117{ 117{
118 if (NULL != p->ch) 118 if (NULL != p->ch)
119 { 119 {
120 GNUNET_CORE_disconnecT (p->ch); 120 GNUNET_CORE_disconnect (p->ch);
121 p->ch = NULL; 121 p->ch = NULL;
122 } 122 }
123 if (NULL != p->ghh) 123 if (NULL != p->ghh)
@@ -480,7 +480,7 @@ init_notify (void *cls,
480 GNUNET_assert (ok == 2); 480 GNUNET_assert (ok == 2);
481 OKPP; 481 OKPP;
482 /* connect p2 */ 482 /* connect p2 */
483 p2.ch = GNUNET_CORE_connecT (p2.cfg, 483 p2.ch = GNUNET_CORE_connect (p2.cfg,
484 &p2, 484 &p2,
485 &init_notify, 485 &init_notify,
486 &connect_notify, 486 &connect_notify,
@@ -653,7 +653,7 @@ run (void *cls,
653 "WAN_QUOTA_OUT", 653 "WAN_QUOTA_OUT",
654 &current_quota_p2_out)); 654 &current_quota_p2_out));
655 655
656 p1.ch = GNUNET_CORE_connecT (p1.cfg, 656 p1.ch = GNUNET_CORE_connect (p1.cfg,
657 &p1, 657 &p1,
658 &init_notify, 658 &init_notify,
659 &connect_notify, 659 &connect_notify,
generated by cgit v1.2.3 (git 2.39.1) at 2024-05-20 06:44:44 +0000