diff options
Diffstat (limited to 'src/service/transport/transport_api2_core.c')
-rw-r--r-- | src/service/transport/transport_api2_core.c | 825 |
1 files changed, 825 insertions, 0 deletions
diff --git a/src/service/transport/transport_api2_core.c b/src/service/transport/transport_api2_core.c new file mode 100644 index 000000000..3eb6c651e --- /dev/null +++ b/src/service/transport/transport_api2_core.c | |||
@@ -0,0 +1,825 @@ | |||
1 | /* | ||
2 | This file is part of GNUnet. | ||
3 | Copyright (C) 2009-2013, 2016, 2018 GNUnet e.V. | ||
4 | |||
5 | GNUnet is free software: you can redistribute it and/or modify it | ||
6 | under the terms of the GNU Affero General Public License as published | ||
7 | by the Free Software Foundation, either version 3 of the License, | ||
8 | or (at your 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 | Affero General Public License for more details. | ||
14 | |||
15 | You should have received a copy of the GNU Affero General Public License | ||
16 | along with this program. If not, see <http://www.gnu.org/licenses/>. | ||
17 | |||
18 | SPDX-License-Identifier: AGPL3.0-or-later | ||
19 | */ | ||
20 | |||
21 | /** | ||
22 | * @file transport/transport_api_core.c | ||
23 | * @brief library to access the transport service for message exchange | ||
24 | * @author Christian Grothoff | ||
25 | */ | ||
26 | #include "platform.h" | ||
27 | #include "gnunet_util_lib.h" | ||
28 | #include "gnunet_constants.h" | ||
29 | #include "gnunet_arm_service.h" | ||
30 | #include "gnunet_protocols.h" | ||
31 | #include "gnunet_transport_core_service.h" | ||
32 | #include "transport.h" | ||
33 | |||
34 | #define LOG(kind, ...) GNUNET_log_from (kind, "transport-api-core", __VA_ARGS__) | ||
35 | |||
36 | /** | ||
37 | * How large to start with for the hashmap of neighbours. | ||
38 | */ | ||
39 | #define STARTING_NEIGHBOURS_SIZE 16 | ||
40 | |||
41 | /** | ||
42 | * Window size. How many messages to the same target do we pass | ||
43 | * to TRANSPORT without a SEND_OK in between? Small values limit | ||
44 | * thoughput, large values will increase latency. | ||
45 | * | ||
46 | * FIXME-OPTIMIZE: find out what good values are experimentally, | ||
47 | * maybe set adaptively (i.e. to observed available bandwidth). | ||
48 | */ | ||
49 | #define SEND_WINDOW_SIZE 4 | ||
50 | |||
51 | |||
52 | /** | ||
53 | * Entry in hash table of all of our current (connected) neighbours. | ||
54 | */ | ||
55 | struct Neighbour | ||
56 | { | ||
57 | /** | ||
58 | * Identity of this neighbour. | ||
59 | */ | ||
60 | struct GNUNET_PeerIdentity id; | ||
61 | |||
62 | /** | ||
63 | * Overall transport handle. | ||
64 | */ | ||
65 | struct GNUNET_TRANSPORT_CoreHandle *h; | ||
66 | |||
67 | /** | ||
68 | * Active message queue for the peer. | ||
69 | */ | ||
70 | struct GNUNET_MQ_Handle *mq; | ||
71 | |||
72 | /** | ||
73 | * Envelope with the message we are currently transmitting (or NULL). | ||
74 | */ | ||
75 | struct GNUNET_MQ_Envelope *env; | ||
76 | |||
77 | /** | ||
78 | * Closure for @e mq handlers. | ||
79 | */ | ||
80 | void *handlers_cls; | ||
81 | |||
82 | /** | ||
83 | * How many messages can we still send to this peer before we should | ||
84 | * throttle? | ||
85 | */ | ||
86 | unsigned int ready_window; | ||
87 | |||
88 | /** | ||
89 | * Used to indicate our status if @e env is non-NULL. Set to | ||
90 | * #GNUNET_YES if we did pass a message to the MQ and are waiting | ||
91 | * for the call to #notify_send_done(). Set to #GNUNET_NO if the @e | ||
92 | * ready_window is 0 and @e env is waiting for a | ||
93 | * #GNUNET_MESSAGE_TYPE_TRANSPORT_RECV_OK? | ||
94 | */ | ||
95 | int16_t awaiting_done; | ||
96 | |||
97 | /** | ||
98 | * Size of the message in @e env. | ||
99 | */ | ||
100 | uint16_t env_size; | ||
101 | }; | ||
102 | |||
103 | |||
104 | /** | ||
105 | * Handle for the transport service (includes all of the | ||
106 | * state for the transport service). | ||
107 | */ | ||
108 | struct GNUNET_TRANSPORT_CoreHandle | ||
109 | { | ||
110 | /** | ||
111 | * Closure for the callbacks. | ||
112 | */ | ||
113 | void *cls; | ||
114 | |||
115 | /** | ||
116 | * Functions to call for received data (template for | ||
117 | * new message queues). | ||
118 | */ | ||
119 | struct GNUNET_MQ_MessageHandler *handlers; | ||
120 | |||
121 | /** | ||
122 | * function to call on connect events | ||
123 | */ | ||
124 | GNUNET_TRANSPORT_NotifyConnect nc_cb; | ||
125 | |||
126 | /** | ||
127 | * function to call on disconnect events | ||
128 | */ | ||
129 | GNUNET_TRANSPORT_NotifyDisconnect nd_cb; | ||
130 | |||
131 | /** | ||
132 | * My client connection to the transport service. | ||
133 | */ | ||
134 | struct GNUNET_MQ_Handle *mq; | ||
135 | |||
136 | /** | ||
137 | * My configuration. | ||
138 | */ | ||
139 | const struct GNUNET_CONFIGURATION_Handle *cfg; | ||
140 | |||
141 | /** | ||
142 | * Hash map of the current connected neighbours of this peer. | ||
143 | * Maps peer identities to `struct Neighbour` entries. | ||
144 | */ | ||
145 | struct GNUNET_CONTAINER_MultiPeerMap *neighbours; | ||
146 | |||
147 | /** | ||
148 | * Peer identity as assumed by this process, or all zeros. | ||
149 | */ | ||
150 | struct GNUNET_PeerIdentity self; | ||
151 | |||
152 | /** | ||
153 | * ID of the task trying to reconnect to the service. | ||
154 | */ | ||
155 | struct GNUNET_SCHEDULER_Task *reconnect_task; | ||
156 | |||
157 | /** | ||
158 | * Delay until we try to reconnect. | ||
159 | */ | ||
160 | struct GNUNET_TIME_Relative reconnect_delay; | ||
161 | |||
162 | /** | ||
163 | * Should we check that @e self matches what the service thinks? | ||
164 | * (if #GNUNET_NO, then @e self is all zeros!). | ||
165 | */ | ||
166 | int check_self; | ||
167 | }; | ||
168 | |||
169 | |||
170 | /** | ||
171 | * Function that will schedule the job that will try | ||
172 | * to connect us again to the client. | ||
173 | * | ||
174 | * @param h transport service to reconnect | ||
175 | */ | ||
176 | static void | ||
177 | disconnect_and_schedule_reconnect (struct GNUNET_TRANSPORT_CoreHandle *h); | ||
178 | |||
179 | |||
180 | /** | ||
181 | * Get the neighbour list entry for the given peer | ||
182 | * | ||
183 | * @param h our context | ||
184 | * @param peer peer to look up | ||
185 | * @return NULL if no such peer entry exists | ||
186 | */ | ||
187 | static struct Neighbour * | ||
188 | neighbour_find (struct GNUNET_TRANSPORT_CoreHandle *h, | ||
189 | const struct GNUNET_PeerIdentity *peer) | ||
190 | { | ||
191 | return GNUNET_CONTAINER_multipeermap_get (h->neighbours, peer); | ||
192 | } | ||
193 | |||
194 | |||
195 | /** | ||
196 | * Iterator over hash map entries, for deleting state of a neighbour. | ||
197 | * | ||
198 | * @param cls the `struct GNUNET_TRANSPORT_CoreHandle *` | ||
199 | * @param key peer identity | ||
200 | * @param value value in the hash map, the neighbour entry to delete | ||
201 | * @return #GNUNET_YES if we should continue to | ||
202 | * iterate, | ||
203 | * #GNUNET_NO if not. | ||
204 | */ | ||
205 | static int | ||
206 | neighbour_delete (void *cls, const struct GNUNET_PeerIdentity *key, void *value) | ||
207 | { | ||
208 | struct GNUNET_TRANSPORT_CoreHandle *handle = cls; | ||
209 | struct Neighbour *n = value; | ||
210 | |||
211 | LOG (GNUNET_ERROR_TYPE_DEBUG, | ||
212 | "Dropping entry for neighbour `%s'.\n", | ||
213 | GNUNET_i2s (key)); | ||
214 | if (NULL != handle->nd_cb) | ||
215 | handle->nd_cb (handle->cls, &n->id, n->handlers_cls); | ||
216 | if (NULL != n->env) | ||
217 | { | ||
218 | GNUNET_MQ_send_cancel (n->env); | ||
219 | n->env = NULL; | ||
220 | } | ||
221 | GNUNET_MQ_destroy (n->mq); | ||
222 | GNUNET_assert (NULL == n->mq); | ||
223 | GNUNET_assert ( | ||
224 | GNUNET_YES == | ||
225 | GNUNET_CONTAINER_multipeermap_remove (handle->neighbours, key, n)); | ||
226 | GNUNET_free (n); | ||
227 | return GNUNET_YES; | ||
228 | } | ||
229 | |||
230 | |||
231 | /** | ||
232 | * Generic error handler, called with the appropriate | ||
233 | * error code and the same closure specified at the creation of | ||
234 | * the message queue. | ||
235 | * Not every message queue implementation supports an error handler. | ||
236 | * | ||
237 | * @param cls closure with the `struct GNUNET_TRANSPORT_CoreHandle *` | ||
238 | * @param error error code | ||
239 | */ | ||
240 | static void | ||
241 | mq_error_handler (void *cls, enum GNUNET_MQ_Error error) | ||
242 | { | ||
243 | struct GNUNET_TRANSPORT_CoreHandle *h = cls; | ||
244 | |||
245 | LOG (GNUNET_ERROR_TYPE_DEBUG, | ||
246 | "Error %u received from transport service, disconnecting temporarily.\n", | ||
247 | error); | ||
248 | disconnect_and_schedule_reconnect (h); | ||
249 | } | ||
250 | |||
251 | |||
252 | /** | ||
253 | * A message from the handler's message queue to a neighbour was | ||
254 | * transmitted. Now trigger (possibly delayed) notification of the | ||
255 | * neighbour's message queue that we are done and thus ready for | ||
256 | * the next message. Note that the MQ being ready is independent | ||
257 | * of the send window, as we may queue many messages and simply | ||
258 | * not pass them to TRANSPORT if the send window is insufficient. | ||
259 | * | ||
260 | * @param cls the `struct Neighbour` where the message was sent | ||
261 | */ | ||
262 | static void | ||
263 | notify_send_done (void *cls) | ||
264 | { | ||
265 | struct Neighbour *n = cls; | ||
266 | |||
267 | n->awaiting_done = GNUNET_NO; | ||
268 | n->env = NULL; | ||
269 | if (0 < n->ready_window) | ||
270 | GNUNET_MQ_impl_send_continue (n->mq); | ||
271 | } | ||
272 | |||
273 | |||
274 | /** | ||
275 | * We have an envelope waiting for transmission at @a n, and | ||
276 | * our transmission window is positive. Perform the transmission. | ||
277 | * | ||
278 | * @param n neighbour to perform transmission for | ||
279 | */ | ||
280 | static void | ||
281 | do_send (struct Neighbour *n) | ||
282 | { | ||
283 | GNUNET_assert (0 < n->ready_window); | ||
284 | GNUNET_assert (NULL != n->env); | ||
285 | n->ready_window--; | ||
286 | n->awaiting_done = GNUNET_YES; | ||
287 | GNUNET_MQ_notify_sent (n->env, ¬ify_send_done, n); | ||
288 | GNUNET_MQ_send (n->h->mq, n->env); | ||
289 | LOG (GNUNET_ERROR_TYPE_DEBUG, | ||
290 | "Passed message of type %u for neighbour `%s' to TRANSPORT. ready_window %u\n", | ||
291 | ntohs (GNUNET_MQ_env_get_msg (n->env)->type), | ||
292 | GNUNET_i2s (&n->id), | ||
293 | n->ready_window); | ||
294 | } | ||
295 | |||
296 | |||
297 | /** | ||
298 | * Implement sending functionality of a message queue. | ||
299 | * Called one message at a time. Should send the @a msg | ||
300 | * to the transport service and then notify the queue | ||
301 | * once we are ready for the next one. | ||
302 | * | ||
303 | * @param mq the message queue | ||
304 | * @param msg the message to send | ||
305 | * @param impl_state state of the implementation | ||
306 | */ | ||
307 | static void | ||
308 | mq_send_impl (struct GNUNET_MQ_Handle *mq, | ||
309 | const struct GNUNET_MessageHeader *msg, | ||
310 | void *impl_state) | ||
311 | { | ||
312 | struct Neighbour *n = impl_state; | ||
313 | struct OutboundMessage *obm; | ||
314 | uint16_t msize; | ||
315 | |||
316 | msize = ntohs (msg->size); | ||
317 | if (msize >= GNUNET_MAX_MESSAGE_SIZE - sizeof(*obm)) | ||
318 | { | ||
319 | GNUNET_break (0); | ||
320 | GNUNET_MQ_impl_send_continue (mq); | ||
321 | return; | ||
322 | } | ||
323 | LOG (GNUNET_ERROR_TYPE_DEBUG, | ||
324 | "CORE requested transmission of message of type %u to neighbour `%s'.\n", | ||
325 | ntohs (msg->type), | ||
326 | GNUNET_i2s (&n->id)); | ||
327 | |||
328 | GNUNET_assert (NULL == n->env); | ||
329 | n->env = | ||
330 | GNUNET_MQ_msg_nested_mh (obm, GNUNET_MESSAGE_TYPE_TRANSPORT_SEND, msg); | ||
331 | n->env_size = ntohs (msg->size); | ||
332 | { | ||
333 | struct GNUNET_MQ_Envelope *env; | ||
334 | |||
335 | env = GNUNET_MQ_get_current_envelope (mq); | ||
336 | obm->priority = htonl ((uint32_t) GNUNET_MQ_env_get_options (env)); | ||
337 | } | ||
338 | obm->peer = n->id; | ||
339 | if (0 == n->ready_window) | ||
340 | { | ||
341 | LOG (GNUNET_ERROR_TYPE_DEBUG, | ||
342 | "Flow control delays transmission to CORE until we see SEND_OK.\n"); | ||
343 | return; /* can't send yet, need to wait for SEND_OK */ | ||
344 | } | ||
345 | do_send (n); | ||
346 | } | ||
347 | |||
348 | |||
349 | /** | ||
350 | * Handle destruction of a message queue. Implementations must not | ||
351 | * free @a mq, but should take care of @a impl_state. | ||
352 | * | ||
353 | * @param mq the message queue to destroy | ||
354 | * @param impl_state state of the implementation | ||
355 | */ | ||
356 | static void | ||
357 | mq_destroy_impl (struct GNUNET_MQ_Handle *mq, void *impl_state) | ||
358 | { | ||
359 | struct Neighbour *n = impl_state; | ||
360 | |||
361 | GNUNET_assert (mq == n->mq); | ||
362 | n->mq = NULL; | ||
363 | } | ||
364 | |||
365 | |||
366 | /** | ||
367 | * Implementation function that cancels the currently sent message. | ||
368 | * Should basically undo whatever #mq_send_impl() did. | ||
369 | * | ||
370 | * @param mq message queue | ||
371 | * @param impl_state state specific to the implementation | ||
372 | */ | ||
373 | static void | ||
374 | mq_cancel_impl (struct GNUNET_MQ_Handle *mq, void *impl_state) | ||
375 | { | ||
376 | struct Neighbour *n = impl_state; | ||
377 | |||
378 | n->ready_window++; | ||
379 | if (GNUNET_YES == n->awaiting_done) | ||
380 | { | ||
381 | GNUNET_MQ_send_cancel (n->env); | ||
382 | n->env = NULL; | ||
383 | n->awaiting_done = GNUNET_NO; | ||
384 | } | ||
385 | else | ||
386 | { | ||
387 | GNUNET_assert (0 == n->ready_window); | ||
388 | n->env = NULL; | ||
389 | } | ||
390 | } | ||
391 | |||
392 | |||
393 | /** | ||
394 | * We had an error processing a message we forwarded from a peer to | ||
395 | * the CORE service. We should just complain about it but otherwise | ||
396 | * continue processing. | ||
397 | * | ||
398 | * @param cls closure | ||
399 | * @param error error code | ||
400 | */ | ||
401 | static void | ||
402 | peer_mq_error_handler (void *cls, enum GNUNET_MQ_Error error) | ||
403 | { | ||
404 | struct Neighbour *n = cls; | ||
405 | |||
406 | if (GNUNET_MQ_ERROR_MALFORMED == error) | ||
407 | GNUNET_break_op (0); | ||
408 | //TODO Look into bug #7887 | ||
409 | |||
410 | GNUNET_TRANSPORT_core_receive_continue (n->h, &n->id); | ||
411 | } | ||
412 | |||
413 | |||
414 | /** | ||
415 | * Function we use for handling incoming connect messages. | ||
416 | * | ||
417 | * @param cls closure, a `struct GNUNET_TRANSPORT_Handle *` | ||
418 | * @param cim message received | ||
419 | */ | ||
420 | static void | ||
421 | handle_connect (void *cls, const struct ConnectInfoMessage *cim) | ||
422 | { | ||
423 | struct GNUNET_TRANSPORT_CoreHandle *h = cls; | ||
424 | struct Neighbour *n; | ||
425 | |||
426 | LOG (GNUNET_ERROR_TYPE_DEBUG, | ||
427 | "Receiving CONNECT message for `%s'\n", | ||
428 | GNUNET_i2s (&cim->id)); | ||
429 | n = neighbour_find (h, &cim->id); | ||
430 | if (NULL != n) | ||
431 | { | ||
432 | GNUNET_break (0); | ||
433 | disconnect_and_schedule_reconnect (h); | ||
434 | return; | ||
435 | } | ||
436 | n = GNUNET_new (struct Neighbour); | ||
437 | n->id = cim->id; | ||
438 | n->h = h; | ||
439 | n->ready_window = SEND_WINDOW_SIZE; | ||
440 | GNUNET_assert (GNUNET_OK == | ||
441 | GNUNET_CONTAINER_multipeermap_put ( | ||
442 | h->neighbours, | ||
443 | &n->id, | ||
444 | n, | ||
445 | GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY)); | ||
446 | |||
447 | n->mq = GNUNET_MQ_queue_for_callbacks (&mq_send_impl, | ||
448 | &mq_destroy_impl, | ||
449 | &mq_cancel_impl, | ||
450 | n, | ||
451 | h->handlers, | ||
452 | &peer_mq_error_handler, | ||
453 | n); | ||
454 | if (NULL != h->nc_cb) | ||
455 | { | ||
456 | n->handlers_cls = h->nc_cb (h->cls, &n->id, n->mq); | ||
457 | GNUNET_MQ_set_handlers_closure (n->mq, n->handlers_cls); | ||
458 | } | ||
459 | } | ||
460 | |||
461 | |||
462 | /** | ||
463 | * Function we use for handling incoming disconnect messages. | ||
464 | * | ||
465 | * @param cls closure, a `struct GNUNET_TRANSPORT_CoreHandle *` | ||
466 | * @param dim message received | ||
467 | */ | ||
468 | static void | ||
469 | handle_disconnect (void *cls, const struct DisconnectInfoMessage *dim) | ||
470 | { | ||
471 | struct GNUNET_TRANSPORT_CoreHandle *h = cls; | ||
472 | struct Neighbour *n; | ||
473 | |||
474 | GNUNET_break (ntohl (dim->reserved) == 0); | ||
475 | LOG (GNUNET_ERROR_TYPE_DEBUG, | ||
476 | "Receiving DISCONNECT message for `%s'.\n", | ||
477 | GNUNET_i2s (&dim->peer)); | ||
478 | n = neighbour_find (h, &dim->peer); | ||
479 | if (NULL == n) | ||
480 | { | ||
481 | GNUNET_break (0); | ||
482 | disconnect_and_schedule_reconnect (h); | ||
483 | return; | ||
484 | } | ||
485 | GNUNET_assert (GNUNET_YES == neighbour_delete (h, &dim->peer, n)); | ||
486 | } | ||
487 | |||
488 | |||
489 | /** | ||
490 | * Function we use for handling incoming send-ok messages. | ||
491 | * | ||
492 | * @param cls closure, a `struct GNUNET_TRANSPORT_CoreHandle *` | ||
493 | * @param okm message received | ||
494 | */ | ||
495 | static void | ||
496 | handle_send_ok (void *cls, const struct SendOkMessage *okm) | ||
497 | { | ||
498 | struct GNUNET_TRANSPORT_CoreHandle *h = cls; | ||
499 | struct Neighbour *n; | ||
500 | |||
501 | LOG (GNUNET_ERROR_TYPE_DEBUG, | ||
502 | "Receiving SEND_OK message for transmission to %s\n", | ||
503 | GNUNET_i2s (&okm->peer)); | ||
504 | |||
505 | n = neighbour_find (h, &okm->peer); | ||
506 | |||
507 | if (NULL == n) | ||
508 | { | ||
509 | /* We should never get a 'SEND_OK' for a peer that we are not | ||
510 | connected to */ | ||
511 | GNUNET_break (0); | ||
512 | disconnect_and_schedule_reconnect (h); | ||
513 | return; | ||
514 | } | ||
515 | |||
516 | if ((GNUNET_NO == n->awaiting_done) && | ||
517 | (NULL != n->env) && | ||
518 | (0 == n->ready_window)) | ||
519 | { | ||
520 | n->ready_window++; | ||
521 | do_send (n); | ||
522 | return; | ||
523 | } | ||
524 | else if ((GNUNET_NO == n->awaiting_done) && | ||
525 | (0 == n->ready_window)) | ||
526 | { | ||
527 | n->ready_window++; | ||
528 | GNUNET_MQ_impl_send_continue (n->mq); | ||
529 | return; | ||
530 | } | ||
531 | n->ready_window++; | ||
532 | } | ||
533 | |||
534 | |||
535 | /** | ||
536 | * Function we use for checking incoming "inbound" messages. | ||
537 | * | ||
538 | * @param cls closure, a `struct GNUNET_TRANSPORT_CoreHandle *` | ||
539 | * @param im message received | ||
540 | */ | ||
541 | static int | ||
542 | check_recv (void *cls, const struct InboundMessage *im) | ||
543 | { | ||
544 | const struct GNUNET_MessageHeader *imm; | ||
545 | uint16_t size; | ||
546 | |||
547 | GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, | ||
548 | "check_recv\n"); | ||
549 | size = ntohs (im->header.size) - sizeof(*im); | ||
550 | if (size < sizeof(struct GNUNET_MessageHeader)) | ||
551 | { | ||
552 | GNUNET_break (0); | ||
553 | return GNUNET_SYSERR; | ||
554 | } | ||
555 | imm = (const struct GNUNET_MessageHeader *) &im[1]; | ||
556 | if (ntohs (imm->size) != size) | ||
557 | { | ||
558 | GNUNET_break (0); | ||
559 | return GNUNET_SYSERR; | ||
560 | } | ||
561 | return GNUNET_OK; | ||
562 | } | ||
563 | |||
564 | |||
565 | /** | ||
566 | * Function we use for handling incoming messages. | ||
567 | * | ||
568 | * @param cls closure, a `struct GNUNET_TRANSPORT_CoreHandle *` | ||
569 | * @param im message received | ||
570 | */ | ||
571 | static void | ||
572 | handle_recv (void *cls, const struct InboundMessage *im) | ||
573 | { | ||
574 | struct GNUNET_TRANSPORT_CoreHandle *h = cls; | ||
575 | const struct GNUNET_MessageHeader *imm = | ||
576 | (const struct GNUNET_MessageHeader *) &im[1]; | ||
577 | struct Neighbour *n; | ||
578 | |||
579 | LOG (GNUNET_ERROR_TYPE_DEBUG, | ||
580 | "Received message of type %u with %u bytes from `%s'.\n", | ||
581 | (unsigned int) ntohs (imm->type), | ||
582 | (unsigned int) ntohs (imm->size), | ||
583 | GNUNET_i2s (&im->peer)); | ||
584 | n = neighbour_find (h, &im->peer); | ||
585 | if (NULL == n) | ||
586 | { | ||
587 | GNUNET_break (0); | ||
588 | disconnect_and_schedule_reconnect (h); | ||
589 | return; | ||
590 | } | ||
591 | GNUNET_MQ_inject_message (n->mq, imm); | ||
592 | } | ||
593 | |||
594 | |||
595 | /** | ||
596 | * Try again to connect to transport service. | ||
597 | * | ||
598 | * @param cls the handle to the transport service | ||
599 | */ | ||
600 | static void | ||
601 | reconnect (void *cls) | ||
602 | { | ||
603 | struct GNUNET_TRANSPORT_CoreHandle *h = cls; | ||
604 | struct GNUNET_MQ_MessageHandler handlers[] = | ||
605 | { GNUNET_MQ_hd_fixed_size (connect, | ||
606 | GNUNET_MESSAGE_TYPE_TRANSPORT_CONNECT, | ||
607 | struct ConnectInfoMessage, | ||
608 | h), | ||
609 | GNUNET_MQ_hd_fixed_size (disconnect, | ||
610 | GNUNET_MESSAGE_TYPE_TRANSPORT_DISCONNECT, | ||
611 | struct DisconnectInfoMessage, | ||
612 | h), | ||
613 | GNUNET_MQ_hd_fixed_size (send_ok, | ||
614 | GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_OK, | ||
615 | struct SendOkMessage, | ||
616 | h), | ||
617 | GNUNET_MQ_hd_var_size (recv, | ||
618 | GNUNET_MESSAGE_TYPE_TRANSPORT_RECV, | ||
619 | struct InboundMessage, | ||
620 | h), | ||
621 | GNUNET_MQ_handler_end () }; | ||
622 | struct GNUNET_MQ_Envelope *env; | ||
623 | struct StartMessage *s; | ||
624 | uint32_t options; | ||
625 | |||
626 | h->reconnect_task = NULL; | ||
627 | LOG (GNUNET_ERROR_TYPE_DEBUG, "Connecting to transport service.\n"); | ||
628 | GNUNET_assert (NULL == h->mq); | ||
629 | h->mq = | ||
630 | GNUNET_CLIENT_connect (h->cfg, "transport", handlers, &mq_error_handler, h); | ||
631 | if (NULL == h->mq) | ||
632 | return; | ||
633 | env = GNUNET_MQ_msg (s, GNUNET_MESSAGE_TYPE_TRANSPORT_START); | ||
634 | options = 0; | ||
635 | if (h->check_self) | ||
636 | options |= 1; | ||
637 | if (NULL != h->handlers) | ||
638 | options |= 2; | ||
639 | s->options = htonl (options); | ||
640 | s->self = h->self; | ||
641 | GNUNET_MQ_send (h->mq, env); | ||
642 | } | ||
643 | |||
644 | |||
645 | /** | ||
646 | * Disconnect from the transport service. | ||
647 | * | ||
648 | * @param h transport service to reconnect | ||
649 | */ | ||
650 | static void | ||
651 | disconnect (struct GNUNET_TRANSPORT_CoreHandle *h) | ||
652 | { | ||
653 | GNUNET_CONTAINER_multipeermap_iterate (h->neighbours, &neighbour_delete, h); | ||
654 | if (NULL != h->mq) | ||
655 | { | ||
656 | GNUNET_MQ_destroy (h->mq); | ||
657 | h->mq = NULL; | ||
658 | } | ||
659 | } | ||
660 | |||
661 | |||
662 | /** | ||
663 | * Function that will schedule the job that will try | ||
664 | * to connect us again to the client. | ||
665 | * | ||
666 | * @param h transport service to reconnect | ||
667 | */ | ||
668 | static void | ||
669 | disconnect_and_schedule_reconnect (struct GNUNET_TRANSPORT_CoreHandle *h) | ||
670 | { | ||
671 | GNUNET_assert (NULL == h->reconnect_task); | ||
672 | disconnect (h); | ||
673 | LOG (GNUNET_ERROR_TYPE_DEBUG, | ||
674 | "Scheduling task to reconnect to transport service in %s.\n", | ||
675 | GNUNET_STRINGS_relative_time_to_string (h->reconnect_delay, GNUNET_YES)); | ||
676 | h->reconnect_task = | ||
677 | GNUNET_SCHEDULER_add_delayed (h->reconnect_delay, &reconnect, h); | ||
678 | h->reconnect_delay = GNUNET_TIME_STD_BACKOFF (h->reconnect_delay); | ||
679 | } | ||
680 | |||
681 | |||
682 | /** | ||
683 | * Checks if a given peer is connected to us and get the message queue. | ||
684 | * | ||
685 | * @param handle connection to transport service | ||
686 | * @param peer the peer to check | ||
687 | * @return NULL if disconnected, otherwise message queue for @a peer | ||
688 | */ | ||
689 | struct GNUNET_MQ_Handle * | ||
690 | GNUNET_TRANSPORT_core_get_mq (struct GNUNET_TRANSPORT_CoreHandle *handle, | ||
691 | const struct GNUNET_PeerIdentity *peer) | ||
692 | { | ||
693 | struct Neighbour *n; | ||
694 | |||
695 | n = neighbour_find (handle, peer); | ||
696 | if (NULL == n) | ||
697 | return NULL; | ||
698 | return n->mq; | ||
699 | } | ||
700 | |||
701 | |||
702 | /** | ||
703 | * Notification from the CORE service to the TRANSPORT service | ||
704 | * that the CORE service has finished processing a message from | ||
705 | * TRANSPORT (via the @code{handlers} of #GNUNET_TRANSPORT_core_connect()) | ||
706 | * and that it is thus now OK for TRANSPORT to send more messages | ||
707 | * for @a pid. | ||
708 | * | ||
709 | * Used to provide flow control, this is our equivalent to | ||
710 | * #GNUNET_SERVICE_client_continue() of an ordinary service. | ||
711 | * | ||
712 | * Note that due to the use of a window, TRANSPORT may send multiple | ||
713 | * messages destined for the same peer even without an intermediate | ||
714 | * call to this function. However, CORE must still call this function | ||
715 | * once per message received, as otherwise eventually the window will | ||
716 | * be full and TRANSPORT will stop providing messages to CORE for @a | ||
717 | * pid. | ||
718 | * | ||
719 | * @param ch core handle | ||
720 | * @param pid which peer was the message from that was fully processed by CORE | ||
721 | */ | ||
722 | void | ||
723 | GNUNET_TRANSPORT_core_receive_continue (struct GNUNET_TRANSPORT_CoreHandle *ch, | ||
724 | const struct GNUNET_PeerIdentity *pid) | ||
725 | { | ||
726 | struct GNUNET_MQ_Envelope *env; | ||
727 | struct RecvOkMessage *rok; | ||
728 | |||
729 | LOG (GNUNET_ERROR_TYPE_DEBUG, | ||
730 | "Message for %s finished CORE processing, sending RECV_OK.\n", | ||
731 | GNUNET_i2s (pid)); | ||
732 | if (NULL == ch->mq) | ||
733 | return; | ||
734 | env = GNUNET_MQ_msg (rok, GNUNET_MESSAGE_TYPE_TRANSPORT_RECV_OK); | ||
735 | rok->increase_window_delta = htonl (1); | ||
736 | rok->peer = *pid; | ||
737 | GNUNET_MQ_send (ch->mq, env); | ||
738 | } | ||
739 | |||
740 | |||
741 | /** | ||
742 | * Connect to the transport service. Note that the connection may | ||
743 | * complete (or fail) asynchronously. | ||
744 | * | ||
745 | * @param cfg configuration to use | ||
746 | * @param self our own identity (API should check that it matches | ||
747 | * the identity found by transport), or NULL (no check) | ||
748 | * @param cls closure for the callbacks | ||
749 | * @param rec receive function to call | ||
750 | * @param nc function to call on connect events | ||
751 | * @param nd function to call on disconnect events | ||
752 | * @return NULL on error | ||
753 | */ | ||
754 | struct GNUNET_TRANSPORT_CoreHandle * | ||
755 | GNUNET_TRANSPORT_core_connect (const struct GNUNET_CONFIGURATION_Handle *cfg, | ||
756 | const struct GNUNET_PeerIdentity *self, | ||
757 | const struct GNUNET_MQ_MessageHandler *handlers, | ||
758 | void *cls, | ||
759 | GNUNET_TRANSPORT_NotifyConnect nc, | ||
760 | GNUNET_TRANSPORT_NotifyDisconnect nd) | ||
761 | { | ||
762 | struct GNUNET_TRANSPORT_CoreHandle *h; | ||
763 | unsigned int i; | ||
764 | |||
765 | h = GNUNET_new (struct GNUNET_TRANSPORT_CoreHandle); | ||
766 | if (NULL != self) | ||
767 | { | ||
768 | h->self = *self; | ||
769 | h->check_self = GNUNET_YES; | ||
770 | } | ||
771 | h->cfg = cfg; | ||
772 | h->cls = cls; | ||
773 | h->nc_cb = nc; | ||
774 | h->nd_cb = nd; | ||
775 | h->reconnect_delay = GNUNET_TIME_UNIT_ZERO; | ||
776 | if (NULL != handlers) | ||
777 | { | ||
778 | for (i = 0; NULL != handlers[i].cb; i++) | ||
779 | ; | ||
780 | h->handlers = GNUNET_new_array (i + 1, struct GNUNET_MQ_MessageHandler); | ||
781 | GNUNET_memcpy (h->handlers, | ||
782 | handlers, | ||
783 | i * sizeof(struct GNUNET_MQ_MessageHandler)); | ||
784 | } | ||
785 | LOG (GNUNET_ERROR_TYPE_DEBUG, "Connecting to transport service\n"); | ||
786 | reconnect (h); | ||
787 | if (NULL == h->mq) | ||
788 | { | ||
789 | GNUNET_free (h->handlers); | ||
790 | GNUNET_free (h); | ||
791 | return NULL; | ||
792 | } | ||
793 | h->neighbours = | ||
794 | GNUNET_CONTAINER_multipeermap_create (STARTING_NEIGHBOURS_SIZE, GNUNET_YES); | ||
795 | return h; | ||
796 | } | ||
797 | |||
798 | |||
799 | /** | ||
800 | * Disconnect from the transport service. | ||
801 | * | ||
802 | * @param handle handle to the service as returned from | ||
803 | * #GNUNET_TRANSPORT_core_connect() | ||
804 | */ | ||
805 | void | ||
806 | GNUNET_TRANSPORT_core_disconnect (struct GNUNET_TRANSPORT_CoreHandle *handle) | ||
807 | { | ||
808 | LOG (GNUNET_ERROR_TYPE_DEBUG, "Transport disconnect called!\n"); | ||
809 | /* this disconnects all neighbours... */ | ||
810 | disconnect (handle); | ||
811 | /* and now we stop trying to connect again... */ | ||
812 | if (NULL != handle->reconnect_task) | ||
813 | { | ||
814 | GNUNET_SCHEDULER_cancel (handle->reconnect_task); | ||
815 | handle->reconnect_task = NULL; | ||
816 | } | ||
817 | GNUNET_CONTAINER_multipeermap_destroy (handle->neighbours); | ||
818 | handle->neighbours = NULL; | ||
819 | GNUNET_free (handle->handlers); | ||
820 | handle->handlers = NULL; | ||
821 | GNUNET_free (handle); | ||
822 | } | ||
823 | |||
824 | |||
825 | /* end of transport_api_core.c */ | ||