diff options
author | Christian Grothoff <christian@grothoff.org> | 2019-05-04 22:16:24 +0200 |
---|---|---|
committer | Christian Grothoff <christian@grothoff.org> | 2019-05-04 22:16:24 +0200 |
commit | 80f480c752fd8dfa1be51d78fce314d9f0650b50 (patch) | |
tree | 459c21a3fde3b5f66194ac9960632de53586c5f0 /src/core/gnunet-service-core_sessions.c | |
parent | 3d2a951fa12546c09809f0a4d7e789ef8e971b03 (diff) | |
download | gnunet-80f480c752fd8dfa1be51d78fce314d9f0650b50.tar.gz gnunet-80f480c752fd8dfa1be51d78fce314d9f0650b50.zip |
simplify MQ logic to always carry the same kinds of flags, and extend transport API to pass them to (TNG) service
Diffstat (limited to 'src/core/gnunet-service-core_sessions.c')
-rw-r--r-- | src/core/gnunet-service-core_sessions.c | 285 |
1 files changed, 118 insertions, 167 deletions
diff --git a/src/core/gnunet-service-core_sessions.c b/src/core/gnunet-service-core_sessions.c index 4f678c784..8ff61ec20 100644 --- a/src/core/gnunet-service-core_sessions.c +++ b/src/core/gnunet-service-core_sessions.c | |||
@@ -11,7 +11,7 @@ | |||
11 | WITHOUT ANY WARRANTY; without even the implied warranty of | 11 | WITHOUT ANY WARRANTY; without even the implied warranty of |
12 | MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU | 12 | MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU |
13 | Affero General Public License for more details. | 13 | Affero General Public License for more details. |
14 | 14 | ||
15 | You should have received a copy of the GNU Affero General Public License | 15 | You should have received a copy of the GNU Affero General Public License |
16 | along with this program. If not, see <http://www.gnu.org/licenses/>. | 16 | along with this program. If not, see <http://www.gnu.org/licenses/>. |
17 | 17 | ||
@@ -59,7 +59,7 @@ struct SessionMessageEntry | |||
59 | /** | 59 | /** |
60 | * How important is this message. | 60 | * How important is this message. |
61 | */ | 61 | */ |
62 | enum GNUNET_CORE_Priority priority; | 62 | enum GNUNET_MQ_PriorityPreferences priority; |
63 | 63 | ||
64 | /** | 64 | /** |
65 | * Flag set to #GNUNET_YES if this is a typemap message. | 65 | * Flag set to #GNUNET_YES if this is a typemap message. |
@@ -84,7 +84,6 @@ struct SessionMessageEntry | |||
84 | * MessageEntry` itself!) | 84 | * MessageEntry` itself!) |
85 | */ | 85 | */ |
86 | size_t size; | 86 | size_t size; |
87 | |||
88 | }; | 87 | }; |
89 | 88 | ||
90 | 89 | ||
@@ -177,7 +176,6 @@ struct TypeMapConfirmationMessage | |||
177 | * Hash of the (decompressed) type map that was received. | 176 | * Hash of the (decompressed) type map that was received. |
178 | */ | 177 | */ |
179 | struct GNUNET_HashCode tm_hash; | 178 | struct GNUNET_HashCode tm_hash; |
180 | |||
181 | }; | 179 | }; |
182 | 180 | ||
183 | GNUNET_NETWORK_STRUCT_END | 181 | GNUNET_NETWORK_STRUCT_END |
@@ -201,8 +199,7 @@ find_session (const struct GNUNET_PeerIdentity *peer) | |||
201 | { | 199 | { |
202 | if (NULL == sessions) | 200 | if (NULL == sessions) |
203 | return NULL; | 201 | return NULL; |
204 | return GNUNET_CONTAINER_multipeermap_get (sessions, | 202 | return GNUNET_CONTAINER_multipeermap_get (sessions, peer); |
205 | peer); | ||
206 | } | 203 | } |
207 | 204 | ||
208 | 205 | ||
@@ -233,15 +230,13 @@ GSC_SESSIONS_end (const struct GNUNET_PeerIdentity *pid) | |||
233 | while (NULL != (car = session->active_client_request_head)) | 230 | while (NULL != (car = session->active_client_request_head)) |
234 | { | 231 | { |
235 | GNUNET_CONTAINER_DLL_remove (session->active_client_request_head, | 232 | GNUNET_CONTAINER_DLL_remove (session->active_client_request_head, |
236 | session->active_client_request_tail, car); | 233 | session->active_client_request_tail, |
237 | GSC_CLIENTS_reject_request (car, | 234 | car); |
238 | GNUNET_NO); | 235 | GSC_CLIENTS_reject_request (car, GNUNET_NO); |
239 | } | 236 | } |
240 | while (NULL != (sme = session->sme_head)) | 237 | while (NULL != (sme = session->sme_head)) |
241 | { | 238 | { |
242 | GNUNET_CONTAINER_DLL_remove (session->sme_head, | 239 | GNUNET_CONTAINER_DLL_remove (session->sme_head, session->sme_tail, sme); |
243 | session->sme_tail, | ||
244 | sme); | ||
245 | GNUNET_free (sme); | 240 | GNUNET_free (sme); |
246 | } | 241 | } |
247 | if (NULL != session->typemap_task) | 242 | if (NULL != session->typemap_task) |
@@ -251,13 +246,12 @@ GSC_SESSIONS_end (const struct GNUNET_PeerIdentity *pid) | |||
251 | } | 246 | } |
252 | GSC_CLIENTS_notify_clients_about_neighbour (session->peer, | 247 | GSC_CLIENTS_notify_clients_about_neighbour (session->peer, |
253 | session->tmap, | 248 | session->tmap, |
254 | NULL); | 249 | NULL); |
255 | GNUNET_assert (GNUNET_YES == | 250 | GNUNET_assert ( |
256 | GNUNET_CONTAINER_multipeermap_remove (sessions, | 251 | GNUNET_YES == |
257 | session->peer, | 252 | GNUNET_CONTAINER_multipeermap_remove (sessions, session->peer, session)); |
258 | session)); | ||
259 | GNUNET_STATISTICS_set (GSC_stats, | 253 | GNUNET_STATISTICS_set (GSC_stats, |
260 | gettext_noop ("# peers connected"), | 254 | gettext_noop ("# peers connected"), |
261 | GNUNET_CONTAINER_multipeermap_size (sessions), | 255 | GNUNET_CONTAINER_multipeermap_size (sessions), |
262 | GNUNET_NO); | 256 | GNUNET_NO); |
263 | GSC_TYPEMAP_destroy (session->tmap); | 257 | GSC_TYPEMAP_destroy (session->tmap); |
@@ -286,20 +280,15 @@ transmit_typemap_task (void *cls) | |||
286 | delay = session->typemap_delay; | 280 | delay = session->typemap_delay; |
287 | /* randomize a bit to avoid spont. sync */ | 281 | /* randomize a bit to avoid spont. sync */ |
288 | delay.rel_value_us += | 282 | delay.rel_value_us += |
289 | GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, | 283 | GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, 1000 * 1000); |
290 | 1000 * 1000); | ||
291 | session->typemap_task = | 284 | session->typemap_task = |
292 | GNUNET_SCHEDULER_add_delayed (delay, | 285 | GNUNET_SCHEDULER_add_delayed (delay, &transmit_typemap_task, session); |
293 | &transmit_typemap_task, | ||
294 | session); | ||
295 | GNUNET_STATISTICS_update (GSC_stats, | 286 | GNUNET_STATISTICS_update (GSC_stats, |
296 | gettext_noop ("# type map refreshes sent"), | 287 | gettext_noop ("# type map refreshes sent"), |
297 | 1, | 288 | 1, |
298 | GNUNET_NO); | 289 | GNUNET_NO); |
299 | hdr = GSC_TYPEMAP_compute_type_map_message (); | 290 | hdr = GSC_TYPEMAP_compute_type_map_message (); |
300 | GSC_KX_encrypt_and_transmit (session->kx, | 291 | GSC_KX_encrypt_and_transmit (session->kx, hdr, ntohs (hdr->size)); |
301 | hdr, | ||
302 | ntohs (hdr->size)); | ||
303 | GNUNET_free (hdr); | 292 | GNUNET_free (hdr); |
304 | } | 293 | } |
305 | 294 | ||
@@ -315,10 +304,9 @@ start_typemap_task (struct Session *session) | |||
315 | if (NULL != session->typemap_task) | 304 | if (NULL != session->typemap_task) |
316 | GNUNET_SCHEDULER_cancel (session->typemap_task); | 305 | GNUNET_SCHEDULER_cancel (session->typemap_task); |
317 | session->typemap_delay = GNUNET_TIME_UNIT_SECONDS; | 306 | session->typemap_delay = GNUNET_TIME_UNIT_SECONDS; |
318 | session->typemap_task = | 307 | session->typemap_task = GNUNET_SCHEDULER_add_delayed (session->typemap_delay, |
319 | GNUNET_SCHEDULER_add_delayed (session->typemap_delay, | 308 | &transmit_typemap_task, |
320 | &transmit_typemap_task, | 309 | session); |
321 | session); | ||
322 | } | 310 | } |
323 | 311 | ||
324 | 312 | ||
@@ -342,17 +330,16 @@ GSC_SESSIONS_create (const struct GNUNET_PeerIdentity *peer, | |||
342 | session->peer = peer; | 330 | session->peer = peer; |
343 | session->kx = kx; | 331 | session->kx = kx; |
344 | GNUNET_assert (GNUNET_OK == | 332 | GNUNET_assert (GNUNET_OK == |
345 | GNUNET_CONTAINER_multipeermap_put (sessions, | 333 | GNUNET_CONTAINER_multipeermap_put ( |
346 | session->peer, | 334 | sessions, |
347 | session, | 335 | session->peer, |
348 | GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY)); | 336 | session, |
337 | GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY)); | ||
349 | GNUNET_STATISTICS_set (GSC_stats, | 338 | GNUNET_STATISTICS_set (GSC_stats, |
350 | gettext_noop ("# peers connected"), | 339 | gettext_noop ("# peers connected"), |
351 | GNUNET_CONTAINER_multipeermap_size (sessions), | 340 | GNUNET_CONTAINER_multipeermap_size (sessions), |
352 | GNUNET_NO); | 341 | GNUNET_NO); |
353 | GSC_CLIENTS_notify_clients_about_neighbour (peer, | 342 | GSC_CLIENTS_notify_clients_about_neighbour (peer, NULL, session->tmap); |
354 | NULL, | ||
355 | session->tmap); | ||
356 | start_typemap_task (session); | 343 | start_typemap_task (session); |
357 | } | 344 | } |
358 | 345 | ||
@@ -406,15 +393,15 @@ GSC_SESSIONS_confirm_typemap (const struct GNUNET_PeerIdentity *peer, | |||
406 | return; | 393 | return; |
407 | } | 394 | } |
408 | cmsg = (const struct TypeMapConfirmationMessage *) msg; | 395 | cmsg = (const struct TypeMapConfirmationMessage *) msg; |
409 | if (GNUNET_YES != | 396 | if (GNUNET_YES != GSC_TYPEMAP_check_hash (&cmsg->tm_hash)) |
410 | GSC_TYPEMAP_check_hash (&cmsg->tm_hash)) | ||
411 | { | 397 | { |
412 | /* our typemap has changed in the meantime, do not | 398 | /* our typemap has changed in the meantime, do not |
413 | accept confirmation */ | 399 | accept confirmation */ |
414 | GNUNET_STATISTICS_update (GSC_stats, | 400 | GNUNET_STATISTICS_update (GSC_stats, |
415 | gettext_noop | 401 | gettext_noop ( |
416 | ("# outdated typemap confirmations received"), | 402 | "# outdated typemap confirmations received"), |
417 | 1, GNUNET_NO); | 403 | 1, |
404 | GNUNET_NO); | ||
418 | GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, | 405 | GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, |
419 | "Got outdated typemap confirmated from peer `%s'\n", | 406 | "Got outdated typemap confirmated from peer `%s'\n", |
420 | GNUNET_i2s (session->peer)); | 407 | GNUNET_i2s (session->peer)); |
@@ -429,9 +416,10 @@ GSC_SESSIONS_confirm_typemap (const struct GNUNET_PeerIdentity *peer, | |||
429 | session->typemap_task = NULL; | 416 | session->typemap_task = NULL; |
430 | } | 417 | } |
431 | GNUNET_STATISTICS_update (GSC_stats, | 418 | GNUNET_STATISTICS_update (GSC_stats, |
432 | gettext_noop | 419 | gettext_noop ( |
433 | ("# valid typemap confirmations received"), | 420 | "# valid typemap confirmations received"), |
434 | 1, GNUNET_NO); | 421 | 1, |
422 | GNUNET_NO); | ||
435 | } | 423 | } |
436 | 424 | ||
437 | 425 | ||
@@ -453,7 +441,7 @@ notify_client_about_session (void *cls, | |||
453 | 441 | ||
454 | GSC_CLIENTS_notify_client_about_neighbour (client, | 442 | GSC_CLIENTS_notify_client_about_neighbour (client, |
455 | session->peer, | 443 | session->peer, |
456 | NULL, /* old TMAP: none */ | 444 | NULL, /* old TMAP: none */ |
457 | session->tmap); | 445 | session->tmap); |
458 | return GNUNET_OK; | 446 | return GNUNET_OK; |
459 | } | 447 | } |
@@ -503,16 +491,14 @@ GSC_SESSIONS_queue_request (struct GSC_ClientActiveRequest *car) | |||
503 | { | 491 | { |
504 | GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, | 492 | GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, |
505 | "Dropped client request for transmission (am disconnected)\n"); | 493 | "Dropped client request for transmission (am disconnected)\n"); |
506 | GNUNET_break (0); /* should have been rejected earlier */ | 494 | GNUNET_break (0); /* should have been rejected earlier */ |
507 | GSC_CLIENTS_reject_request (car, | 495 | GSC_CLIENTS_reject_request (car, GNUNET_NO); |
508 | GNUNET_NO); | ||
509 | return; | 496 | return; |
510 | } | 497 | } |
511 | if (car->msize > GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE) | 498 | if (car->msize > GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE) |
512 | { | 499 | { |
513 | GNUNET_break (0); | 500 | GNUNET_break (0); |
514 | GSC_CLIENTS_reject_request (car, | 501 | GSC_CLIENTS_reject_request (car, GNUNET_YES); |
515 | GNUNET_YES); | ||
516 | return; | 502 | return; |
517 | } | 503 | } |
518 | GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, | 504 | GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, |
@@ -535,10 +521,9 @@ GSC_SESSIONS_dequeue_request (struct GSC_ClientActiveRequest *car) | |||
535 | { | 521 | { |
536 | struct Session *session; | 522 | struct Session *session; |
537 | 523 | ||
538 | if (0 == | 524 | if (0 == memcmp (&car->target, |
539 | memcmp (&car->target, | 525 | &GSC_my_identity, |
540 | &GSC_my_identity, | 526 | sizeof (struct GNUNET_PeerIdentity))) |
541 | sizeof (struct GNUNET_PeerIdentity))) | ||
542 | return; | 527 | return; |
543 | session = find_session (&car->target); | 528 | session = find_session (&car->target); |
544 | GNUNET_assert (NULL != session); | 529 | GNUNET_assert (NULL != session); |
@@ -560,21 +545,20 @@ GSC_SESSIONS_dequeue_request (struct GSC_ClientActiveRequest *car) | |||
560 | * @param msize how many bytes do we have already | 545 | * @param msize how many bytes do we have already |
561 | */ | 546 | */ |
562 | static void | 547 | static void |
563 | solicit_messages (struct Session *session, | 548 | solicit_messages (struct Session *session, size_t msize) |
564 | size_t msize) | ||
565 | { | 549 | { |
566 | struct GSC_ClientActiveRequest *car; | 550 | struct GSC_ClientActiveRequest *car; |
567 | struct GSC_ClientActiveRequest *nxt; | 551 | struct GSC_ClientActiveRequest *nxt; |
568 | size_t so_size; | 552 | size_t so_size; |
569 | enum GNUNET_CORE_Priority pmax; | 553 | enum GNUNET_MQ_PriorityPreferences pmax; |
570 | 554 | ||
571 | so_size = msize; | 555 | so_size = msize; |
572 | pmax = GNUNET_CORE_PRIO_BACKGROUND; | 556 | pmax = GNUNET_MQ_PRIO_BACKGROUND; |
573 | for (car = session->active_client_request_head; NULL != car; car = car->next) | 557 | for (car = session->active_client_request_head; NULL != car; car = car->next) |
574 | { | 558 | { |
575 | if (GNUNET_YES == car->was_solicited) | 559 | if (GNUNET_YES == car->was_solicited) |
576 | continue; | 560 | continue; |
577 | pmax = GNUNET_MAX (pmax, car->priority); | 561 | pmax = GNUNET_MAX (pmax, car->priority & GNUNET_MQ_PRIORITY_MASK); |
578 | } | 562 | } |
579 | nxt = session->active_client_request_head; | 563 | nxt = session->active_client_request_head; |
580 | while (NULL != (car = nxt)) | 564 | while (NULL != (car = nxt)) |
@@ -631,11 +615,11 @@ try_transmission (struct Session *session) | |||
631 | size_t msize; | 615 | size_t msize; |
632 | struct GNUNET_TIME_Absolute now; | 616 | struct GNUNET_TIME_Absolute now; |
633 | struct GNUNET_TIME_Absolute min_deadline; | 617 | struct GNUNET_TIME_Absolute min_deadline; |
634 | enum GNUNET_CORE_Priority maxp; | 618 | enum GNUNET_MQ_PriorityPreferences maxp; |
635 | enum GNUNET_CORE_Priority maxpc; | 619 | enum GNUNET_MQ_PriorityPreferences maxpc; |
636 | struct GSC_ClientActiveRequest *car; | 620 | struct GSC_ClientActiveRequest *car; |
637 | int excess; | 621 | int excess; |
638 | 622 | ||
639 | msize = 0; | 623 | msize = 0; |
640 | min_deadline = GNUNET_TIME_UNIT_FOREVER_ABS; | 624 | min_deadline = GNUNET_TIME_UNIT_FOREVER_ABS; |
641 | /* if the peer has excess bandwidth, background traffic is allowed, | 625 | /* if the peer has excess bandwidth, background traffic is allowed, |
@@ -649,9 +633,9 @@ try_transmission (struct Session *session) | |||
649 | } | 633 | } |
650 | excess = GSC_NEIGHBOURS_check_excess_bandwidth (session->kx); | 634 | excess = GSC_NEIGHBOURS_check_excess_bandwidth (session->kx); |
651 | if (GNUNET_YES == excess) | 635 | if (GNUNET_YES == excess) |
652 | maxp = GNUNET_CORE_PRIO_BACKGROUND; | 636 | maxp = GNUNET_MQ_PRIO_BACKGROUND; |
653 | else | 637 | else |
654 | maxp = GNUNET_CORE_PRIO_BEST_EFFORT; | 638 | maxp = GNUNET_MQ_PRIO_BEST_EFFORT; |
655 | /* determine highest priority of 'ready' messages we already solicited from clients */ | 639 | /* determine highest priority of 'ready' messages we already solicited from clients */ |
656 | pos = session->sme_head; | 640 | pos = session->sme_head; |
657 | while ((NULL != pos) && | 641 | while ((NULL != pos) && |
@@ -659,32 +643,33 @@ try_transmission (struct Session *session) | |||
659 | { | 643 | { |
660 | GNUNET_assert (pos->size < GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE); | 644 | GNUNET_assert (pos->size < GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE); |
661 | msize += pos->size; | 645 | msize += pos->size; |
662 | maxp = GNUNET_MAX (maxp, pos->priority); | 646 | maxp = GNUNET_MAX (maxp, pos->priority & GNUNET_MQ_PRIORITY_MASK); |
663 | min_deadline = GNUNET_TIME_absolute_min (min_deadline, | 647 | min_deadline = GNUNET_TIME_absolute_min (min_deadline, pos->deadline); |
664 | pos->deadline); | ||
665 | pos = pos->next; | 648 | pos = pos->next; |
666 | } | 649 | } |
667 | GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, | 650 | GNUNET_log ( |
668 | "Calculating transmission set with %u priority (%s) and %s earliest deadline\n", | 651 | GNUNET_ERROR_TYPE_DEBUG, |
669 | maxp, | 652 | "Calculating transmission set with %u priority (%s) and %s earliest deadline\n", |
670 | (GNUNET_YES == excess) ? "excess bandwidth" : "limited bandwidth", | 653 | maxp, |
671 | GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining (min_deadline), | 654 | (GNUNET_YES == excess) ? "excess bandwidth" : "limited bandwidth", |
672 | GNUNET_YES)); | 655 | GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining ( |
673 | 656 | min_deadline), | |
674 | if (maxp < GNUNET_CORE_PRIO_CRITICAL_CONTROL) | 657 | GNUNET_YES)); |
658 | |||
659 | if (maxp < GNUNET_MQ_PRIO_CRITICAL_CONTROL) | ||
675 | { | 660 | { |
676 | /* if highest already solicited priority from clients is not critical, | 661 | /* if highest already solicited priority from clients is not critical, |
677 | check if there are higher-priority messages to be solicited from clients */ | 662 | check if there are higher-priority messages to be solicited from clients */ |
678 | if (GNUNET_YES == excess) | 663 | if (GNUNET_YES == excess) |
679 | maxpc = GNUNET_CORE_PRIO_BACKGROUND; | 664 | maxpc = GNUNET_MQ_PRIO_BACKGROUND; |
680 | else | 665 | else |
681 | maxpc = GNUNET_CORE_PRIO_BEST_EFFORT; | 666 | maxpc = GNUNET_MQ_PRIO_BEST_EFFORT; |
682 | for (car = session->active_client_request_head; NULL != car; car = car->next) | 667 | for (car = session->active_client_request_head; NULL != car; |
668 | car = car->next) | ||
683 | { | 669 | { |
684 | if (GNUNET_YES == car->was_solicited) | 670 | if (GNUNET_YES == car->was_solicited) |
685 | continue; | 671 | continue; |
686 | maxpc = GNUNET_MAX (maxpc, | 672 | maxpc = GNUNET_MAX (maxpc, car->priority & GNUNET_MQ_PRIORITY_MASK); |
687 | car->priority); | ||
688 | } | 673 | } |
689 | if (maxpc > maxp) | 674 | if (maxpc > maxp) |
690 | { | 675 | { |
@@ -703,41 +688,40 @@ try_transmission (struct Session *session) | |||
703 | { | 688 | { |
704 | /* never solicit more, we have critical messages to process */ | 689 | /* never solicit more, we have critical messages to process */ |
705 | excess = GNUNET_NO; | 690 | excess = GNUNET_NO; |
706 | maxpc = GNUNET_CORE_PRIO_BACKGROUND; | 691 | maxpc = GNUNET_MQ_PRIO_BACKGROUND; |
707 | } | 692 | } |
708 | now = GNUNET_TIME_absolute_get (); | 693 | now = GNUNET_TIME_absolute_get (); |
709 | if ( ( (GNUNET_YES == excess) || | 694 | if (((GNUNET_YES == excess) || (maxpc >= GNUNET_MQ_PRIO_BEST_EFFORT)) && |
710 | (maxpc >= GNUNET_CORE_PRIO_BEST_EFFORT) ) && | 695 | ((0 == msize) || |
711 | ( (0 == msize) || | 696 | ((msize < GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE / 2) && |
712 | ( (msize < GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE / 2) && | 697 | (min_deadline.abs_value_us > now.abs_value_us)))) |
713 | (min_deadline.abs_value_us > now.abs_value_us))) ) | ||
714 | { | 698 | { |
715 | /* not enough ready yet (tiny message & cork possible), or no messages at all, | 699 | /* not enough ready yet (tiny message & cork possible), or no messages at all, |
716 | and either excess bandwidth or best-effort or higher message waiting at | 700 | and either excess bandwidth or best-effort or higher message waiting at |
717 | client; in this case, we try to solicit more */ | 701 | client; in this case, we try to solicit more */ |
718 | GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, | 702 | GNUNET_log ( |
719 | "Soliciting messages (excess %d, maxpc %d, message size %u, deadline %s)\n", | 703 | GNUNET_ERROR_TYPE_DEBUG, |
720 | excess, | 704 | "Soliciting messages (excess %d, maxpc %d, message size %u, deadline %s)\n", |
721 | maxpc, | 705 | excess, |
722 | (unsigned int) msize, | 706 | maxpc, |
723 | GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining (min_deadline), | 707 | (unsigned int) msize, |
724 | GNUNET_YES)); | 708 | GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining ( |
725 | solicit_messages (session, | 709 | min_deadline), |
726 | msize); | 710 | GNUNET_YES)); |
711 | solicit_messages (session, msize); | ||
727 | if (msize > 0) | 712 | if (msize > 0) |
728 | { | 713 | { |
729 | /* if there is data to send, just not yet, make sure we do transmit | 714 | /* if there is data to send, just not yet, make sure we do transmit |
730 | * it once the deadline is reached */ | 715 | * it once the deadline is reached */ |
731 | GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, | 716 | GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, |
732 | "Corking until %s\n", | 717 | "Corking until %s\n", |
733 | GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining (min_deadline), | 718 | GNUNET_STRINGS_relative_time_to_string ( |
734 | GNUNET_YES)); | 719 | GNUNET_TIME_absolute_get_remaining (min_deadline), |
720 | GNUNET_YES)); | ||
735 | if (NULL != session->cork_task) | 721 | if (NULL != session->cork_task) |
736 | GNUNET_SCHEDULER_cancel (session->cork_task); | 722 | GNUNET_SCHEDULER_cancel (session->cork_task); |
737 | session->cork_task | 723 | session->cork_task = |
738 | = GNUNET_SCHEDULER_add_at (min_deadline, | 724 | GNUNET_SCHEDULER_add_at (min_deadline, &pop_cork_task, session); |
739 | &pop_cork_task, | ||
740 | session); | ||
741 | } | 725 | } |
742 | else | 726 | else |
743 | { | 727 | { |
@@ -753,26 +737,21 @@ try_transmission (struct Session *session) | |||
753 | { | 737 | { |
754 | static unsigned long long total_bytes; | 738 | static unsigned long long total_bytes; |
755 | static unsigned int total_msgs; | 739 | static unsigned int total_msgs; |
756 | char pbuf[msize]; /* plaintext */ | 740 | char pbuf[msize]; /* plaintext */ |
757 | size_t used; | 741 | size_t used; |
758 | 742 | ||
759 | used = 0; | 743 | used = 0; |
760 | while ( (NULL != (pos = session->sme_head)) && | 744 | while ((NULL != (pos = session->sme_head)) && (used + pos->size <= msize)) |
761 | (used + pos->size <= msize) ) | ||
762 | { | 745 | { |
763 | GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, | 746 | GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, |
764 | "Adding message of type %d (%d/%d) to payload for %s\n", | 747 | "Adding message of type %d (%d/%d) to payload for %s\n", |
765 | ntohs (((const struct GNUNET_MessageHeader *)&pos[1])->type), | 748 | ntohs (((const struct GNUNET_MessageHeader *) &pos[1])->type), |
766 | pos->is_typemap, | 749 | pos->is_typemap, |
767 | pos->is_typemap_confirm, | 750 | pos->is_typemap_confirm, |
768 | GNUNET_i2s (session->peer)); | 751 | GNUNET_i2s (session->peer)); |
769 | GNUNET_memcpy (&pbuf[used], | 752 | GNUNET_memcpy (&pbuf[used], &pos[1], pos->size); |
770 | &pos[1], | ||
771 | pos->size); | ||
772 | used += pos->size; | 753 | used += pos->size; |
773 | GNUNET_CONTAINER_DLL_remove (session->sme_head, | 754 | GNUNET_CONTAINER_DLL_remove (session->sme_head, session->sme_tail, pos); |
774 | session->sme_tail, | ||
775 | pos); | ||
776 | GNUNET_free (pos); | 755 | GNUNET_free (pos); |
777 | } | 756 | } |
778 | /* compute average payload size */ | 757 | /* compute average payload size */ |
@@ -789,9 +768,7 @@ try_transmission (struct Session *session) | |||
789 | total_bytes / total_msgs, | 768 | total_bytes / total_msgs, |
790 | GNUNET_NO); | 769 | GNUNET_NO); |
791 | /* now actually transmit... */ | 770 | /* now actually transmit... */ |
792 | GSC_KX_encrypt_and_transmit (session->kx, | 771 | GSC_KX_encrypt_and_transmit (session->kx, pbuf, used); |
793 | pbuf, | ||
794 | used); | ||
795 | } | 772 | } |
796 | } | 773 | } |
797 | 774 | ||
@@ -823,23 +800,17 @@ do_restart_typemap_message (void *cls, | |||
823 | { | 800 | { |
824 | if (GNUNET_YES == sme->is_typemap) | 801 | if (GNUNET_YES == sme->is_typemap) |
825 | { | 802 | { |
826 | GNUNET_CONTAINER_DLL_remove (session->sme_head, | 803 | GNUNET_CONTAINER_DLL_remove (session->sme_head, session->sme_tail, sme); |
827 | session->sme_tail, | ||
828 | sme); | ||
829 | GNUNET_free (sme); | 804 | GNUNET_free (sme); |
830 | break; | 805 | break; |
831 | } | 806 | } |
832 | } | 807 | } |
833 | sme = GNUNET_malloc (sizeof (struct SessionMessageEntry) + size); | 808 | sme = GNUNET_malloc (sizeof (struct SessionMessageEntry) + size); |
834 | sme->is_typemap = GNUNET_YES; | 809 | sme->is_typemap = GNUNET_YES; |
835 | GNUNET_memcpy (&sme[1], | 810 | GNUNET_memcpy (&sme[1], hdr, size); |
836 | hdr, | ||
837 | size); | ||
838 | sme->size = size; | 811 | sme->size = size; |
839 | sme->priority = GNUNET_CORE_PRIO_CRITICAL_CONTROL; | 812 | sme->priority = GNUNET_MQ_PRIO_CRITICAL_CONTROL; |
840 | GNUNET_CONTAINER_DLL_insert (session->sme_head, | 813 | GNUNET_CONTAINER_DLL_insert (session->sme_head, session->sme_tail, sme); |
841 | session->sme_tail, | ||
842 | sme); | ||
843 | try_transmission (session); | 814 | try_transmission (session); |
844 | start_typemap_task (session); | 815 | start_typemap_task (session); |
845 | return GNUNET_OK; | 816 | return GNUNET_OK; |
@@ -876,8 +847,8 @@ GSC_SESSIONS_solicit (const struct GNUNET_PeerIdentity *pid) | |||
876 | struct Session *session; | 847 | struct Session *session; |
877 | 848 | ||
878 | GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, | 849 | GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, |
879 | "Transport solicits for %s\n", | 850 | "Transport solicits for %s\n", |
880 | GNUNET_i2s (pid)); | 851 | GNUNET_i2s (pid)); |
881 | session = find_session (pid); | 852 | session = find_session (pid); |
882 | if (NULL == session) | 853 | if (NULL == session) |
883 | return; | 854 | return; |
@@ -891,14 +862,12 @@ GSC_SESSIONS_solicit (const struct GNUNET_PeerIdentity *pid) | |||
891 | * @param car original request that was queued and then solicited; | 862 | * @param car original request that was queued and then solicited; |
892 | * this handle will now be 'owned' by the SESSIONS subsystem | 863 | * this handle will now be 'owned' by the SESSIONS subsystem |
893 | * @param msg message to transmit | 864 | * @param msg message to transmit |
894 | * @param cork is corking allowed? | ||
895 | * @param priority how important is this message | 865 | * @param priority how important is this message |
896 | */ | 866 | */ |
897 | void | 867 | void |
898 | GSC_SESSIONS_transmit (struct GSC_ClientActiveRequest *car, | 868 | GSC_SESSIONS_transmit (struct GSC_ClientActiveRequest *car, |
899 | const struct GNUNET_MessageHeader *msg, | 869 | const struct GNUNET_MessageHeader *msg, |
900 | int cork, | 870 | enum GNUNET_MQ_PriorityPreferences priority) |
901 | enum GNUNET_CORE_Priority priority) | ||
902 | { | 871 | { |
903 | struct Session *session; | 872 | struct Session *session; |
904 | struct SessionMessageEntry *sme; | 873 | struct SessionMessageEntry *sme; |
@@ -910,21 +879,18 @@ GSC_SESSIONS_transmit (struct GSC_ClientActiveRequest *car, | |||
910 | return; | 879 | return; |
911 | msize = ntohs (msg->size); | 880 | msize = ntohs (msg->size); |
912 | sme = GNUNET_malloc (sizeof (struct SessionMessageEntry) + msize); | 881 | sme = GNUNET_malloc (sizeof (struct SessionMessageEntry) + msize); |
913 | GNUNET_memcpy (&sme[1], | 882 | GNUNET_memcpy (&sme[1], msg, msize); |
914 | msg, | ||
915 | msize); | ||
916 | sme->size = msize; | 883 | sme->size = msize; |
917 | sme->priority = priority; | 884 | sme->priority = priority; |
918 | if (GNUNET_YES == cork) | 885 | if (0 != (GNUNET_MQ_PREF_CORK_ALLOWED & priority)) |
919 | { | 886 | { |
920 | sme->deadline = | 887 | sme->deadline = |
921 | GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_MAX_CORK_DELAY); | 888 | GNUNET_TIME_relative_to_absolute (GNUNET_CONSTANTS_MAX_CORK_DELAY); |
922 | GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, | 889 | GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, |
923 | "Mesage corked, delaying transmission\n"); | 890 | "Mesage corked, delaying transmission\n"); |
924 | } | 891 | } |
925 | pos = session->sme_head; | 892 | pos = session->sme_head; |
926 | while ( (NULL != pos) && | 893 | while ((NULL != pos) && (pos->priority >= sme->priority)) |
927 | (pos->priority >= sme->priority) ) | ||
928 | pos = pos->next; | 894 | pos = pos->next; |
929 | if (NULL == pos) | 895 | if (NULL == pos) |
930 | GNUNET_CONTAINER_DLL_insert_tail (session->sme_head, | 896 | GNUNET_CONTAINER_DLL_insert_tail (session->sme_head, |
@@ -959,7 +925,7 @@ GSC_SESSIONS_set_typemap (const struct GNUNET_PeerIdentity *peer, | |||
959 | if (NULL == nmap) | 925 | if (NULL == nmap) |
960 | { | 926 | { |
961 | GNUNET_break_op (0); | 927 | GNUNET_break_op (0); |
962 | return; /* malformed */ | 928 | return; /* malformed */ |
963 | } | 929 | } |
964 | session = find_session (peer); | 930 | session = find_session (peer); |
965 | if (NULL == session) | 931 | if (NULL == session) |
@@ -975,9 +941,7 @@ GSC_SESSIONS_set_typemap (const struct GNUNET_PeerIdentity *peer, | |||
975 | { | 941 | { |
976 | if (GNUNET_YES == sme->is_typemap_confirm) | 942 | if (GNUNET_YES == sme->is_typemap_confirm) |
977 | { | 943 | { |
978 | GNUNET_CONTAINER_DLL_remove (session->sme_head, | 944 | GNUNET_CONTAINER_DLL_remove (session->sme_head, session->sme_tail, sme); |
979 | session->sme_tail, | ||
980 | sme); | ||
981 | GNUNET_free (sme); | 945 | GNUNET_free (sme); |
982 | break; | 946 | break; |
983 | } | 947 | } |
@@ -986,21 +950,16 @@ GSC_SESSIONS_set_typemap (const struct GNUNET_PeerIdentity *peer, | |||
986 | sizeof (struct TypeMapConfirmationMessage)); | 950 | sizeof (struct TypeMapConfirmationMessage)); |
987 | sme->deadline = GNUNET_TIME_absolute_get (); | 951 | sme->deadline = GNUNET_TIME_absolute_get (); |
988 | sme->size = sizeof (struct TypeMapConfirmationMessage); | 952 | sme->size = sizeof (struct TypeMapConfirmationMessage); |
989 | sme->priority = GNUNET_CORE_PRIO_CRITICAL_CONTROL; | 953 | sme->priority = GNUNET_MQ_PRIO_CRITICAL_CONTROL; |
990 | sme->is_typemap_confirm = GNUNET_YES; | 954 | sme->is_typemap_confirm = GNUNET_YES; |
991 | tmc = (struct TypeMapConfirmationMessage *) &sme[1]; | 955 | tmc = (struct TypeMapConfirmationMessage *) &sme[1]; |
992 | tmc->header.size = htons (sizeof (struct TypeMapConfirmationMessage)); | 956 | tmc->header.size = htons (sizeof (struct TypeMapConfirmationMessage)); |
993 | tmc->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_CONFIRM_TYPE_MAP); | 957 | tmc->header.type = htons (GNUNET_MESSAGE_TYPE_CORE_CONFIRM_TYPE_MAP); |
994 | tmc->reserved = htonl (0); | 958 | tmc->reserved = htonl (0); |
995 | GSC_TYPEMAP_hash (nmap, | 959 | GSC_TYPEMAP_hash (nmap, &tmc->tm_hash); |
996 | &tmc->tm_hash); | 960 | GNUNET_CONTAINER_DLL_insert (session->sme_head, session->sme_tail, sme); |
997 | GNUNET_CONTAINER_DLL_insert (session->sme_head, | ||
998 | session->sme_tail, | ||
999 | sme); | ||
1000 | try_transmission (session); | 961 | try_transmission (session); |
1001 | GSC_CLIENTS_notify_clients_about_neighbour (peer, | 962 | GSC_CLIENTS_notify_clients_about_neighbour (peer, session->tmap, nmap); |
1002 | session->tmap, | ||
1003 | nmap); | ||
1004 | GSC_TYPEMAP_destroy (session->tmap); | 963 | GSC_TYPEMAP_destroy (session->tmap); |
1005 | session->tmap = nmap; | 964 | session->tmap = nmap; |
1006 | } | 965 | } |
@@ -1021,21 +980,14 @@ GSC_SESSIONS_add_to_typemap (const struct GNUNET_PeerIdentity *peer, | |||
1021 | struct Session *session; | 980 | struct Session *session; |
1022 | struct GSC_TypeMap *nmap; | 981 | struct GSC_TypeMap *nmap; |
1023 | 982 | ||
1024 | if (0 == memcmp (peer, | 983 | if (0 == memcmp (peer, &GSC_my_identity, sizeof (struct GNUNET_PeerIdentity))) |
1025 | &GSC_my_identity, | ||
1026 | sizeof (struct GNUNET_PeerIdentity))) | ||
1027 | return; | 984 | return; |
1028 | session = find_session (peer); | 985 | session = find_session (peer); |
1029 | GNUNET_assert (NULL != session); | 986 | GNUNET_assert (NULL != session); |
1030 | if (GNUNET_YES == GSC_TYPEMAP_test_match (session->tmap, | 987 | if (GNUNET_YES == GSC_TYPEMAP_test_match (session->tmap, &type, 1)) |
1031 | &type, 1)) | 988 | return; /* already in it */ |
1032 | return; /* already in it */ | 989 | nmap = GSC_TYPEMAP_extend (session->tmap, &type, 1); |
1033 | nmap = GSC_TYPEMAP_extend (session->tmap, | 990 | GSC_CLIENTS_notify_clients_about_neighbour (peer, session->tmap, nmap); |
1034 | &type, | ||
1035 | 1); | ||
1036 | GSC_CLIENTS_notify_clients_about_neighbour (peer, | ||
1037 | session->tmap, | ||
1038 | nmap); | ||
1039 | GSC_TYPEMAP_destroy (session->tmap); | 991 | GSC_TYPEMAP_destroy (session->tmap); |
1040 | session->tmap = nmap; | 992 | session->tmap = nmap; |
1041 | } | 993 | } |
@@ -1047,8 +999,7 @@ GSC_SESSIONS_add_to_typemap (const struct GNUNET_PeerIdentity *peer, | |||
1047 | void | 999 | void |
1048 | GSC_SESSIONS_init () | 1000 | GSC_SESSIONS_init () |
1049 | { | 1001 | { |
1050 | sessions = GNUNET_CONTAINER_multipeermap_create (128, | 1002 | sessions = GNUNET_CONTAINER_multipeermap_create (128, GNUNET_YES); |
1051 | GNUNET_YES); | ||
1052 | } | 1003 | } |
1053 | 1004 | ||
1054 | 1005 | ||
@@ -1083,7 +1034,7 @@ GSC_SESSIONS_done () | |||
1083 | { | 1034 | { |
1084 | GNUNET_CONTAINER_multipeermap_iterate (sessions, | 1035 | GNUNET_CONTAINER_multipeermap_iterate (sessions, |
1085 | &free_session_helper, | 1036 | &free_session_helper, |
1086 | NULL); | 1037 | NULL); |
1087 | GNUNET_CONTAINER_multipeermap_destroy (sessions); | 1038 | GNUNET_CONTAINER_multipeermap_destroy (sessions); |
1088 | sessions = NULL; | 1039 | sessions = NULL; |
1089 | } | 1040 | } |