2432 2553 | EligibilityGroup::Pool,
|
2433 2554 | None,
|
2434 2555 | None,
|
2435 2556 | ))
|
2436 2557 | }
|
2437 2558 |
|
2438 2559 | fn connection_info(id: u64) -> Arc<ConnectionInfo> {
|
2439 2560 | ConnectionInfo::for_test(ConnectionId::new(id), PartitionId::from_index(1))
|
2440 2561 | }
|
2441 2562 |
|
2563 + | /// Races local HTTP/1 selection with close.
|
2564 + | ///
|
2565 + | /// The sender must have one winner and no installed record may survive.
|
2442 2566 | #[test]
|
2443 2567 | fn h1_selection_linearizes_against_close() {
|
2444 2568 | loom::model(|| {
|
2445 2569 | let cell = unbounded_cell();
|
2446 2570 | let (connection, _physical) = ConnectionState::unbounded(connection_info(1));
|
2447 2571 | OriginCell::insert_idle_h1(&cell, connection.clone(), H1Sender::test(11));
|
2448 2572 | let close = H1CloseHandle::new(&cell, &connection);
|
2449 2573 |
|
2450 2574 | let select_cell = cell.clone();
|
2451 2575 | let select = loom::thread::spawn(move || OriginCell::select_h1(&select_cell));
|
2452 2576 | let close = loom::thread::spawn(move || close.close(CloseReason::Poisoned));
|
2453 2577 |
|
2454 2578 | let selected = select.join().unwrap();
|
2455 2579 | assert!(close.join().unwrap());
|
2456 2580 | drop(selected);
|
2457 2581 |
|
2458 2582 | assert_eq!((0, 0), cell.h1_counts());
|
2459 2583 | assert_eq!(Some(CloseReason::Poisoned), connection.probe().close_reason);
|
2460 2584 | });
|
2461 2585 | }
|
2462 2586 |
|
2587 + | /// Races a reusable HTTP/1 return with close.
|
2588 + | ///
|
2589 + | /// Close must prevent the returning sender from becoming selectable again.
|
2463 2590 | #[test]
|
2464 2591 | fn h1_return_linearizes_against_close() {
|
2465 2592 | loom::model(|| {
|
2466 2593 | let cell = unbounded_cell();
|
2467 2594 | let (connection, _physical) = ConnectionState::unbounded(connection_info(1));
|
2468 2595 | let selection =
|
2469 2596 | OriginCell::insert_selected_h1(&cell, connection.clone(), H1Sender::test(11));
|
2470 2597 | let exchange = selection.into_exchange();
|
2471 2598 | let close = H1CloseHandle::new(&cell, &connection);
|
2472 2599 |
|
2473 2600 | let returning = loom::thread::spawn(move || exchange.offer_for_reuse());
|
2474 2601 | let closing = loom::thread::spawn(move || close.close(CloseReason::Poisoned));
|
2475 2602 | returning.join().unwrap();
|
2476 2603 | assert!(closing.join().unwrap());
|
2477 2604 |
|
2478 2605 | assert_eq!((0, 0), cell.h1_counts());
|
2479 2606 | assert_eq!(Some(CloseReason::Poisoned), connection.probe().close_reason);
|
2480 2607 | });
|
2481 2608 | }
|
2482 2609 |
|
2610 + | /// Races local sender delivery with cancellation of its target waiter.
|
2611 + | ///
|
2612 + | /// Cancellation may win, but the exclusive sender must return exactly once.
|
2483 2613 | #[test]
|
2484 2614 | fn h1_delivery_and_waiter_cancellation_preserve_the_sender() {
|
2485 2615 | loom::model(|| {
|
2486 2616 | let cell = unbounded_cell();
|
2487 2617 | let (connection, _physical) = ConnectionState::unbounded(connection_info(1));
|
2488 2618 | let selection = OriginCell::insert_selected_h1(&cell, connection, H1Sender::test(11));
|
2489 2619 | let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
2490 2620 |
|
2491 2621 | let returning = loom::thread::spawn(move || drop(selection));
|
2492 2622 | let cancel_cell = cell.clone();
|
2493 2623 | let cancelling =
|
2494 2624 | loom::thread::spawn(move || OriginCell::cancel_waiter(&cancel_cell, waiter));
|
2495 2625 | returning.join().unwrap();
|
2496 2626 | assert!(cancelling.join().unwrap());
|
2497 2627 |
|
2498 2628 | assert_eq!((1, 1), cell.h1_counts());
|
2499 2629 | assert_eq!(0, cell.probe().retained);
|
2500 2630 | assert_eq!(
|
2501 2631 | 11,
|
2502 2632 | OriginCell::select_h1(&cell)
|
2503 2633 | .expect("sender was lost during return/cancel")
|
2504 2634 | .test_sender_id()
|
2505 2635 | );
|
2506 2636 | });
|
2507 2637 | }
|
2508 2638 |
|
2639 + | /// Races an idle return with completion of an already-started establishment.
|
2640 + | ///
|
2641 + | /// Exactly one result satisfies the waiter and the losing sender remains reusable.
|
2509 2642 | #[test]
|
2510 2643 | fn returned_h1_and_establishment_race_for_one_waiter() {
|
2511 2644 | loom::model(|| {
|
2512 2645 | let cell = unbounded_cell();
|
2513 2646 | let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
2514 2647 | let AcquisitionStep::StartEstablishment(permit) = cell
|
2515 2648 | .take_ready_event(waiter)
|
2516 2649 | .expect("unbounded miss did not start establishment")
|
2517 2650 | else {
|
2518 2651 | panic!("unbounded miss completed before establishment started");
|
2519 2652 | };
|
2520 2653 | assert!(cell.start_establishment(waiter));
|
2521 2654 | drop(permit);
|
2522 2655 |
|
2523 2656 | let (returning_connection, _returning_physical) =
|
2524 2657 | ConnectionState::unbounded(connection_info(1));
|
2525 2658 | let returning =
|
2526 2659 | OriginCell::insert_selected_h1(&cell, returning_connection, H1Sender::test(11));
|
2527 2660 | let (fresh_connection, _fresh_physical) =
|
2528 2661 | ConnectionState::unbounded(connection_info(2));
|
2529 2662 | let fresh = OriginCell::insert_selected_h1(&cell, fresh_connection, H1Sender::test(22));
|
2530 2663 |
|
2531 2664 | let returning = loom::thread::spawn(move || drop(returning));
|
2532 2665 | let completion_cell = cell.clone();
|
2533 2666 | let completing = loom::thread::spawn(move || {
|
2534 2667 | completion_cell.complete_establishment(waiter, AcquisitionOutcome::H1(fresh));
|
2535 2668 | });
|
2536 2669 | returning.join().unwrap();
|
2537 2670 | completing.join().unwrap();
|
2538 2671 |
|
2539 2672 | assert_eq!((2, 1), cell.h1_counts());
|
2540 2673 | let winner = cell
|
2541 2674 | .take_ready_h1(waiter)
|
2542 2675 | .expect("both acquisition outcomes were lost");
|
2543 2676 | assert!(matches!(winner.test_sender_id(), 11 | 22));
|
2544 2677 | drop(winner);
|
2545 2678 | assert_eq!((2, 2), cell.h1_counts());
|
2546 2679 | });
|
2547 2680 | }
|
2548 2681 |
|
2682 + | /// Races first establishment commitment with a reusable HTTP/1 return.
|
2683 + | ///
|
2684 + | /// A return that wins before commitment must satisfy the waiter without being displaced.
|
2549 2685 | #[test]
|
2550 2686 | fn first_establishment_poll_races_returned_h1() {
|
2551 2687 | loom::model(|| {
|
2552 2688 | let cell = unbounded_cell();
|
2553 2689 | let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
2554 2690 | let AcquisitionStep::StartEstablishment(permit) = cell
|
2555 2691 | .take_ready_event(waiter)
|
2556 2692 | .expect("unbounded miss did not prepare establishment")
|
2557 2693 | else {
|
2558 2694 | panic!("unbounded miss completed before establishment was prepared");
|
2559 2695 | };
|
2560 2696 |
|
2561 2697 | let (returning_connection, _returning_physical) =
|
2562 2698 | ConnectionState::unbounded(connection_info(1));
|
2563 2699 | let returning =
|
2564 2700 | OriginCell::insert_selected_h1(&cell, returning_connection, H1Sender::test(11));
|
2565 2701 | let return_thread = loom::thread::spawn(move || drop(returning));
|
2566 2702 | let start_cell = cell.clone();
|
2567 2703 | let start_thread = loom::thread::spawn(move || start_cell.start_establishment(waiter));
|
2568 2704 |
|
2569 2705 | return_thread.join().unwrap();
|
2570 2706 | let started = start_thread.join().unwrap();
|
2571 2707 | drop(permit);
|
2572 2708 |
|
2573 2709 | if started {
|
2574 2710 | let (fresh_connection, _fresh_physical) =
|
2575 2711 | ConnectionState::unbounded(connection_info(2));
|
2576 2712 | let fresh =
|
2577 2713 | OriginCell::insert_selected_h1(&cell, fresh_connection, H1Sender::test(22));
|
2578 2714 | cell.complete_establishment(waiter, AcquisitionOutcome::H1(fresh));
|
2579 2715 | }
|
2580 2716 |
|
2581 2717 | let winner = cell
|
2582 2718 | .take_ready_h1(waiter)
|
2583 2719 | .expect("returned H1 was lost at the first-poll boundary");
|
2584 2720 | assert_eq!(11, winner.test_sender_id());
|
2585 2721 | drop(winner);
|
2586 2722 | assert_eq!(
|
2587 2723 | (usize::from(started) + 1, usize::from(started) + 1),
|
2588 2724 | cell.h1_counts()
|
2589 2725 | );
|
2590 2726 | });
|
2591 2727 | }
|
2592 2728 |
|
2729 + | /// Races local HTTP/1 return with bounded-capacity delivery.
|
2730 + | ///
|
2731 + | /// The sender wins the waiter while the unused capacity returns to admission.
|
2593 2732 | #[test]
|
2594 2733 | fn h1_return_and_capacity_delivery_complete_one_waiter() {
|
2595 2734 | loom::model(|| {
|
2596 2735 | let (admission, cell) = bounded_cell_with_limit(2);
|
2597 2736 | let installed_lease = OriginAdmission::lease_for_test(&admission);
|
2598 2737 | let (connection, _physical) =
|
2599 2738 | ConnectionState::bounded(connection_info(1), installed_lease);
|
2600 2739 | let returning = OriginCell::insert_selected_h1(&cell, connection, H1Sender::test(11));
|
2601 2740 | let (waiter, demand) =
|
2602 2741 | cell.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
|
2603 - | let mut delivery =
|
2742 + | let delivery =
|
2604 2743 | OriginAdmission::submit_without_running(&admission, cell.id().partition(), demand)
|
2605 2744 | .expect("published demand did not reserve capacity");
|
2606 - | assert!(
|
2607 - | delivery.resolve_payload_for_test(),
|
2608 - | "capacity delivery did not materialize"
|
2609 - | );
|
2610 2745 |
|
2611 - | let delivery_cell = cell.clone();
|
2612 2746 | let delivering = loom::thread::spawn(move || {
|
2613 - | drop(OriginCell::receive_delivery(&delivery_cell, delivery));
|
2747 + | OriginAdmission::run_action_chain(Some(AdmissionAction::Deliver(delivery)));
|
2614 2748 | });
|
2615 2749 | let returning = loom::thread::spawn(move || drop(returning));
|
2616 2750 | delivering.join().unwrap();
|
2617 2751 | returning.join().unwrap();
|
2618 2752 |
|
2619 2753 | assert_eq!(1, admission.available_capacity_for_test());
|
2620 2754 | let selection = cell
|
2621 2755 | .take_ready_h1(waiter)
|
2622 2756 | .expect("return and capacity delivery lost the HTTP/1 sender");
|
2623 2757 | assert_eq!(11, selection.test_sender_id());
|
2624 2758 | drop(selection);
|
2625 2759 | admission.clear_modeled_cells_for_test();
|
2626 2760 | });
|
2627 2761 | }
|
2628 2762 |
|
2763 + | /// Races a retained peer match with cancellation of its requesting waiter.
|
2764 + | ///
|
2765 + | /// Three preemptions cover either actor entering first and a complete
|
2766 + | /// reservation round trip across the admission and cell locks.
|
2629 2767 | #[test]
|
2630 2768 | fn peer_match_and_request_cancellation_preserve_the_sender() {
|
2631 2769 | let mut model = loom::model::Builder::new();
|
2632 - | model.preemption_bound = Some(2);
|
2770 + | model.preemption_bound.get_or_insert(3);
|
2633 2771 | model.check(|| {
|
2634 2772 | let (admission, connection_cell, requesting_cell) =
|
2635 2773 | bounded_peer_cells(EligibilityGroup::Pool, EligibilityGroup::Pool);
|
2636 2774 | let lease = OriginAdmission::lease_for_test(&admission);
|
2637 2775 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
|
2638 2776 | let returning =
|
2639 2777 | OriginCell::insert_selected_h1(&connection_cell, connection, H1Sender::test(11));
|
2640 2778 | let waiter =
|
2641 2779 | OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H1Compatible);
|
2642 2780 |
|
2643 2781 | // Loom's default 4 KiB coroutine stack is too small for the full
|
2644 2782 | // sender-return and admission-settlement path modeled here.
|
2645 2783 | let returning = loom::thread::Builder::new()
|
2646 2784 | .stack_size(16 * 1024)
|
2647 2785 | .spawn(move || drop(returning))
|
2648 2786 | .unwrap();
|
2649 2787 | let cancel_requesting_cell = requesting_cell.clone();
|
2650 2788 | let cancelling = loom::thread::Builder::new()
|
2651 2789 | .stack_size(16 * 1024)
|
2652 2790 | .spawn(move || OriginCell::cancel_waiter(&cancel_requesting_cell, waiter))
|
2653 2791 | .unwrap();
|
2654 2792 | returning.join().unwrap();
|
2655 2793 | assert!(cancelling.join().unwrap());
|
2656 2794 |
|
2657 2795 | assert_eq!((1, 1), connection_cell.h1_counts());
|
2658 2796 | assert_eq!(0, requesting_cell.probe().retained);
|
2659 2797 | assert_eq!(0, admission.ordered_demand_count_for_test());
|
2660 2798 | admission.clear_modeled_cells_for_test();
|
2661 2799 | });
|
2662 2800 | }
|
2663 2801 |
|
2802 + | /// Races a busy idle-probe reply with cancellation of its demand.
|
2803 + | ///
|
2804 + | /// The supplier returns to its index, no fallback reservation survives,
|
2805 + | /// and the active sender remains owned by its connection cell.
|
2806 + | #[test]
|
2807 + | fn busy_idle_probe_and_cancellation_leave_no_retained_match() {
|
2808 + | let mut model = loom::model::Builder::new();
|
2809 + | model.preemption_bound.get_or_insert(3);
|
2810 + | model.check(|| {
|
2811 + | let admission = OriginAdmission::for_test(NonZeroUsize::new(2).unwrap());
|
2812 + | let origin = OriginKey::from_parts(Scheme::HTTPS, "example.com", None).unwrap();
|
2813 + | let first = OriginAdmission::register_cell(
|
2814 + | &admission,
|
2815 + | Arc::new(OriginCell::new(
|
2816 + | PartitionId::from_index(1),
|
2817 + | origin.clone(),
|
2818 + | EligibilityGroup::Pool,
|
2819 + | Some(admission.clone()),
|
2820 + | None,
|
2821 + | )),
|
2822 + | );
|
2823 + | let second = OriginAdmission::register_cell(
|
2824 + | &admission,
|
2825 + | Arc::new(OriginCell::new(
|
2826 + | PartitionId::from_index(2),
|
2827 + | origin.clone(),
|
2828 + | EligibilityGroup::Pool,
|
2829 + | Some(admission.clone()),
|
2830 + | None,
|
2831 + | )),
|
2832 + | );
|
2833 + | let requesting_cell = OriginAdmission::register_cell(
|
2834 + | &admission,
|
2835 + | Arc::new(OriginCell::new(
|
2836 + | PartitionId::from_index(3),
|
2837 + | origin,
|
2838 + | EligibilityGroup::Pool,
|
2839 + | Some(admission.clone()),
|
2840 + | None,
|
2841 + | )),
|
2842 + | );
|
2843 + | let first_lease = OriginAdmission::lease_for_test(&admission);
|
2844 + | let (first_connection, _first_physical) =
|
2845 + | ConnectionState::bounded(connection_info(1), first_lease);
|
2846 + | let first_returning =
|
2847 + | OriginCell::insert_selected_h1(&first, first_connection, H1Sender::test(11));
|
2848 + | let second_lease = OriginAdmission::lease_for_test(&admission);
|
2849 + | let (second_connection, _second_physical) =
|
2850 + | ConnectionState::bounded(connection_info(2), second_lease);
|
2851 + | let second_returning =
|
2852 + | OriginCell::insert_selected_h1(&second, second_connection, H1Sender::test(22));
|
2853 + | let (waiter, demand) =
|
2854 + | requesting_cell.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
|
2855 + | let action = OriginAdmission::submit_action_without_running(
|
2856 + | &admission,
|
2857 + | requesting_cell.id().partition(),
|
2858 + | demand,
|
2859 + | )
|
2860 + | .expect("peer demand did not prepare an idle probe");
|
2861 + |
|
2862 + | let probing = loom::thread::Builder::new()
|
2863 + | .stack_size(16 * 1024)
|
2864 + | .spawn(move || OriginAdmission::run_action_chain(Some(action)))
|
2865 + | .unwrap();
|
2866 + | let cancel_cell = requesting_cell.clone();
|
2867 + | let cancelling = loom::thread::Builder::new()
|
2868 + | .stack_size(16 * 1024)
|
2869 + | .spawn(move || OriginCell::cancel_waiter(&cancel_cell, waiter))
|
2870 + | .unwrap();
|
2871 + | probing.join().unwrap();
|
2872 + | assert!(cancelling.join().unwrap());
|
2873 + | drop(first_returning);
|
2874 + | drop(second_returning);
|
2875 + |
|
2876 + | assert_eq!(vec![11], first.h1_idle_sender_ids());
|
2877 + | assert_eq!(vec![22], second.h1_idle_sender_ids());
|
2878 + | assert!(first.state.lock().h1.peer_reservation_available());
|
2879 + | assert!(second.state.lock().h1.peer_reservation_available());
|
2880 + | assert_eq!(0, admission.ordered_demand_count_for_test());
|
2881 + | admission.clear_modeled_cells_for_test();
|
2882 + | });
|
2883 + | }
|
2884 + |
|
2885 + | /// Races cancellation with a later idle supplier winning bounded probing.
|
2886 + | ///
|
2887 + | /// Whether cancellation or candidate delivery wins, both sender identities
|
2888 + | /// return to their owning cells exactly once. Cancellation returns `true`
|
2889 + | /// when it removes either a waiting record or a ready, unclaimed selection;
|
2890 + | /// in the latter case it also owns cleanup of that selection.
|
2891 + | #[test]
|
2892 + | fn later_idle_probe_and_cancellation_preserve_each_sender() {
|
2893 + | let mut model = loom::model::Builder::new();
|
2894 + | model.preemption_bound.get_or_insert(3);
|
2895 + | model.check(|| {
|
2896 + | let admission = OriginAdmission::for_test(NonZeroUsize::new(2).unwrap());
|
2897 + | let origin = OriginKey::from_parts(Scheme::HTTPS, "example.com", None).unwrap();
|
2898 + | let first = OriginAdmission::register_cell(
|
2899 + | &admission,
|
2900 + | Arc::new(OriginCell::new(
|
2901 + | PartitionId::from_index(1),
|
2902 + | origin.clone(),
|
2903 + | EligibilityGroup::Pool,
|
2904 + | Some(admission.clone()),
|
2905 + | None,
|
2906 + | )),
|
2907 + | );
|
2908 + | let second = OriginAdmission::register_cell(
|
2909 + | &admission,
|
2910 + | Arc::new(OriginCell::new(
|
2911 + | PartitionId::from_index(2),
|
2912 + | origin.clone(),
|
2913 + | EligibilityGroup::Pool,
|
2914 + | Some(admission.clone()),
|
2915 + | None,
|
2916 + | )),
|
2917 + | );
|
2918 + | let requester = OriginAdmission::register_cell(
|
2919 + | &admission,
|
2920 + | Arc::new(OriginCell::new(
|
2921 + | PartitionId::from_index(3),
|
2922 + | origin,
|
2923 + | EligibilityGroup::Pool,
|
2924 + | Some(admission.clone()),
|
2925 + | None,
|
2926 + | )),
|
2927 + | );
|
2928 + | let first_lease = OriginAdmission::lease_for_test(&admission);
|
2929 + | let (first_connection, _first_physical) =
|
2930 + | ConnectionState::bounded(connection_info(1), first_lease);
|
2931 + | let returning =
|
2932 + | OriginCell::insert_selected_h1(&first, first_connection, H1Sender::test(11));
|
2933 + | let second_lease = OriginAdmission::lease_for_test(&admission);
|
2934 + | let (second_connection, _second_physical) =
|
2935 + | ConnectionState::bounded(connection_info(2), second_lease);
|
2936 + | OriginCell::insert_idle_h1(&second, second_connection, H1Sender::test(22));
|
2937 + | let (waiter, demand) =
|
2938 + | requester.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
|
2939 + | let action = OriginAdmission::submit_action_without_running(
|
2940 + | &admission,
|
2941 + | requester.id().partition(),
|
2942 + | demand,
|
2943 + | )
|
2944 + | .expect("peer demand did not prepare an idle probe");
|
2945 + |
|
2946 + | let probing = loom::thread::Builder::new()
|
2947 + | .stack_size(16 * 1024)
|
2948 + | .spawn(move || OriginAdmission::run_action_chain(Some(action)))
|
2949 + | .unwrap();
|
2950 + | let cancel_cell = requester.clone();
|
2951 + | let cancelling = loom::thread::Builder::new()
|
2952 + | .stack_size(16 * 1024)
|
2953 + | .spawn(move || OriginCell::cancel_waiter(&cancel_cell, waiter))
|
2954 + | .unwrap();
|
2955 + | probing.join().unwrap();
|
2956 + | assert!(cancelling.join().unwrap());
|
2957 + | drop(returning);
|
2958 + |
|
2959 + | let mut sender_ids = first.h1_idle_sender_ids();
|
2960 + | sender_ids.extend(second.h1_idle_sender_ids());
|
2961 + | sender_ids.extend(requester.h1_idle_sender_ids());
|
2962 + | sender_ids.sort_unstable();
|
2963 + | assert_eq!(vec![11, 22], sender_ids);
|
2964 + | assert!(first.state.lock().h1.peer_reservation_available());
|
2965 + | assert!(second.state.lock().h1.peer_reservation_available());
|
2966 + | assert_eq!(0, admission.ordered_demand_count_for_test());
|
2967 + | admission.clear_modeled_cells_for_test();
|
2968 + | });
|
2969 + | }
|
2970 + |
|
2971 + | /// Races cancellation with the second idle probe and exact fallback reservation.
|
2972 + | ///
|
2973 + | /// Three busy suppliers force two distinct idle-only probes followed by
|
2974 + | /// reservation of the third supplier. Cancellation begins after the second
|
2975 + | /// probe is selected, so every explored schedule crosses the full bounded
|
2976 + | /// probe path while racing its final supplier transition.
|
2977 + | #[test]
|
2978 + | fn second_idle_probe_and_cancellation_preserve_all_senders() {
|
2979 + | let mut model = loom::model::Builder::new();
|
2980 + | model.preemption_bound.get_or_insert(3);
|
2981 + | model.check(|| {
|
2982 + | let admission = OriginAdmission::for_test(NonZeroUsize::new(3).unwrap());
|
2983 + | let origin = OriginKey::from_parts(Scheme::HTTPS, "example.com", None).unwrap();
|
2984 + | let first = OriginAdmission::register_cell(
|
2985 + | &admission,
|
2986 + | Arc::new(OriginCell::new(
|
2987 + | PartitionId::from_index(1),
|
2988 + | origin.clone(),
|
2989 + | EligibilityGroup::Pool,
|
2990 + | Some(admission.clone()),
|
2991 + | None,
|
2992 + | )),
|
2993 + | );
|
2994 + | let second = OriginAdmission::register_cell(
|
2995 + | &admission,
|
2996 + | Arc::new(OriginCell::new(
|
2997 + | PartitionId::from_index(2),
|
2998 + | origin.clone(),
|
2999 + | EligibilityGroup::Pool,
|
3000 + | Some(admission.clone()),
|
3001 + | None,
|
3002 + | )),
|
3003 + | );
|
3004 + | let third = OriginAdmission::register_cell(
|
3005 + | &admission,
|
3006 + | Arc::new(OriginCell::new(
|
3007 + | PartitionId::from_index(3),
|
3008 + | origin.clone(),
|
3009 + | EligibilityGroup::Pool,
|
3010 + | Some(admission.clone()),
|
3011 + | None,
|
3012 + | )),
|
3013 + | );
|
3014 + | let requester = OriginAdmission::register_cell(
|
3015 + | &admission,
|
3016 + | Arc::new(OriginCell::new(
|
3017 + | PartitionId::from_index(4),
|
3018 + | origin,
|
3019 + | EligibilityGroup::Pool,
|
3020 + | Some(admission.clone()),
|
3021 + | None,
|
3022 + | )),
|
3023 + | );
|
3024 + |
|
3025 + | let first_lease = OriginAdmission::lease_for_test(&admission);
|
3026 + | let (first_connection, _first_physical) =
|
3027 + | ConnectionState::bounded(connection_info(1), first_lease);
|
3028 + | let first_returning =
|
3029 + | OriginCell::insert_selected_h1(&first, first_connection, H1Sender::test(11));
|
3030 + | let second_lease = OriginAdmission::lease_for_test(&admission);
|
3031 + | let (second_connection, _second_physical) =
|
3032 + | ConnectionState::bounded(connection_info(2), second_lease);
|
3033 + | let second_returning =
|
3034 + | OriginCell::insert_selected_h1(&second, second_connection, H1Sender::test(22));
|
3035 + | let third_lease = OriginAdmission::lease_for_test(&admission);
|
3036 + | let (third_connection, _third_physical) =
|
3037 + | ConnectionState::bounded(connection_info(3), third_lease);
|
3038 + | let third_returning =
|
3039 + | OriginCell::insert_selected_h1(&third, third_connection, H1Sender::test(33));
|
3040 + |
|
3041 + | let (waiter, demand) =
|
3042 + | requester.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
|
3043 + | let first_probe = OriginAdmission::submit_action_without_running(
|
3044 + | &admission,
|
3045 + | requester.id().partition(),
|
3046 + | demand,
|
3047 + | )
|
3048 + | .expect("peer demand did not prepare its first idle probe");
|
3049 + | assert!(matches!(&first_probe, AdmissionAction::ProbeH1Supplier(_)));
|
3050 + |
|
3051 + | let probes = Arc::new(AtomicUsize::new(0));
|
3052 + | let second_probe_started = Arc::new(AtomicBool::new(false));
|
3053 + | let probing_count = probes.clone();
|
3054 + | let probing_started = second_probe_started.clone();
|
3055 + | let probing = loom::thread::Builder::new()
|
3056 + | .stack_size(16 * 1024)
|
3057 + | .spawn(move || {
|
3058 + | probing_count.fetch_add(1, Ordering::SeqCst);
|
3059 + | let second_probe = first_probe
|
3060 + | .run_once_for_test()
|
3061 + | .expect("first miss did not prepare a second idle probe");
|
3062 + | assert!(matches!(&second_probe, AdmissionAction::ProbeH1Supplier(_)));
|
3063 + | probing_count.fetch_add(1, Ordering::SeqCst);
|
3064 + | probing_started.store(true, Ordering::SeqCst);
|
3065 + | let fallback = second_probe.run_once_for_test();
|
3066 + | OriginAdmission::run_action_chain(fallback);
|
3067 + | })
|
3068 + | .unwrap();
|
3069 + | let cancel_cell = requester.clone();
|
3070 + | let cancelling = loom::thread::Builder::new()
|
3071 + | .stack_size(16 * 1024)
|
3072 + | .spawn(move || {
|
3073 + | while !second_probe_started.load(Ordering::SeqCst) {
|
3074 + | loom::thread::yield_now();
|
3075 + | }
|
3076 + | OriginCell::cancel_waiter(&cancel_cell, waiter)
|
3077 + | })
|
3078 + | .unwrap();
|
3079 + | probing.join().unwrap();
|
3080 + | assert!(cancelling.join().unwrap());
|
3081 + | assert_eq!(2, probes.load(Ordering::SeqCst));
|
3082 + |
|
3083 + | drop(first_returning);
|
3084 + | drop(second_returning);
|
3085 + | drop(third_returning);
|
3086 + |
|
3087 + | let mut sender_ids = first.h1_idle_sender_ids();
|
3088 + | sender_ids.extend(second.h1_idle_sender_ids());
|
3089 + | sender_ids.extend(third.h1_idle_sender_ids());
|
3090 + | sender_ids.extend(requester.h1_idle_sender_ids());
|
3091 + | sender_ids.sort_unstable();
|
3092 + | assert_eq!(vec![11, 22, 33], sender_ids);
|
3093 + | assert!(first.state.lock().h1.peer_reservation_available());
|
3094 + | assert!(second.state.lock().h1.peer_reservation_available());
|
3095 + | assert!(third.state.lock().h1.peer_reservation_available());
|
3096 + | assert_eq!(0, admission.ordered_demand_count_for_test());
|
3097 + | assert_eq!(0, admission.available_capacity_for_test());
|
3098 + | admission.clear_modeled_cells_for_test();
|
3099 + | });
|
3100 + | }
|
3101 + |
|
3102 + | /// Races a probed supplier return with installation of the fallback reservation.
|
3103 + | ///
|
3104 + | /// The first supplier is busy when probed. Its sender returns after the
|
3105 + | /// miss while admission installs the exact reservation on the second
|
3106 + | /// supplier. The fallback must satisfy the waiter without losing,
|
3107 + | /// duplicating, or reserving the returning sender.
|
3108 + | #[test]
|
3109 + | fn supplier_return_between_probe_miss_and_fallback_preserves_both_senders() {
|
3110 + | let mut model = loom::model::Builder::new();
|
3111 + | model.preemption_bound.get_or_insert(3);
|
3112 + | model.check(|| {
|
3113 + | let admission = OriginAdmission::for_test(NonZeroUsize::new(2).unwrap());
|
3114 + | let origin = OriginKey::from_parts(Scheme::HTTPS, "example.com", None).unwrap();
|
3115 + | let first = OriginAdmission::register_cell(
|
3116 + | &admission,
|
3117 + | Arc::new(OriginCell::new(
|
3118 + | PartitionId::from_index(1),
|
3119 + | origin.clone(),
|
3120 + | EligibilityGroup::Pool,
|
3121 + | Some(admission.clone()),
|
3122 + | None,
|
3123 + | )),
|
3124 + | );
|
3125 + | let second = OriginAdmission::register_cell(
|
3126 + | &admission,
|
3127 + | Arc::new(OriginCell::new(
|
3128 + | PartitionId::from_index(2),
|
3129 + | origin.clone(),
|
3130 + | EligibilityGroup::Pool,
|
3131 + | Some(admission.clone()),
|
3132 + | None,
|
3133 + | )),
|
3134 + | );
|
3135 + | let requester = OriginAdmission::register_cell(
|
3136 + | &admission,
|
3137 + | Arc::new(OriginCell::new(
|
3138 + | PartitionId::from_index(3),
|
3139 + | origin,
|
3140 + | EligibilityGroup::Pool,
|
3141 + | Some(admission.clone()),
|
3142 + | None,
|
3143 + | )),
|
3144 + | );
|
3145 + |
|
3146 + | let first_lease = OriginAdmission::lease_for_test(&admission);
|
3147 + | let (first_connection, _first_physical) =
|
3148 + | ConnectionState::bounded(connection_info(1), first_lease);
|
3149 + | let returning =
|
3150 + | OriginCell::insert_selected_h1(&first, first_connection, H1Sender::test(11));
|
3151 + |
|
3152 + | let second_lease = OriginAdmission::lease_for_test(&admission);
|
3153 + | let (second_connection, _second_physical) =
|
3154 + | ConnectionState::bounded(connection_info(2), second_lease);
|
3155 + | OriginCell::insert_idle_h1(&second, second_connection, H1Sender::test(22));
|
3156 + |
|
3157 + | let (waiter, demand) =
|
3158 + | requester.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
|
3159 + | let probe = OriginAdmission::submit_action_without_running(
|
3160 + | &admission,
|
3161 + | requester.id().partition(),
|
3162 + | demand,
|
3163 + | )
|
3164 + | .expect("peer demand did not prepare an idle probe");
|
3165 + | let fallback = probe
|
3166 + | .run_once_for_test()
|
3167 + | .expect("busy probe did not prepare the fallback reservation");
|
3168 + |
|
3169 + | let reserving = loom::thread::Builder::new()
|
3170 + | .stack_size(16 * 1024)
|
3171 + | .spawn(move || OriginAdmission::run_action_chain(Some(fallback)))
|
3172 + | .unwrap();
|
3173 + | let returning = loom::thread::Builder::new()
|
3174 + | .stack_size(16 * 1024)
|
3175 + | .spawn(move || drop(returning))
|
3176 + | .unwrap();
|
3177 + | reserving.join().unwrap();
|
3178 + | returning.join().unwrap();
|
3179 + |
|
3180 + | let selected = requester
|
3181 + | .take_ready_h1(waiter)
|
3182 + | .expect("fallback reservation did not satisfy the waiter");
|
3183 + | assert_eq!(22, selected.test_sender_id());
|
3184 + | drop(selected);
|
3185 + |
|
3186 + | let mut sender_ids = first.h1_idle_sender_ids();
|
3187 + | sender_ids.extend(second.h1_idle_sender_ids());
|
3188 + | sender_ids.extend(requester.h1_idle_sender_ids());
|
3189 + | sender_ids.sort_unstable();
|
3190 + | assert_eq!(vec![11, 22], sender_ids);
|
3191 + | assert!(first.state.lock().h1.peer_reservation_available());
|
3192 + | assert!(second.state.lock().h1.peer_reservation_available());
|
3193 + | assert_eq!(0, admission.ordered_demand_count_for_test());
|
3194 + | admission.clear_modeled_cells_for_test();
|
3195 + | });
|
3196 + | }
|
3197 + |
|
3198 + | /// Races borrowed-sender delivery with a local sender return.
|
3199 + | ///
|
3200 + | /// Both exclusive sender identities must remain installed exactly once.
|
3201 + | /// Three preemptions cover either actor entering first and one crossing
|
3202 + | /// returning through both lock domains.
|
2664 3203 | #[test]
|
2665 3204 | fn borrowed_delivery_racing_a_local_return_preserves_both_senders() {
|
2666 3205 | let mut model = loom::model::Builder::new();
|
2667 - | model.preemption_bound = Some(2);
|
3206 + | model.preemption_bound.get_or_insert(3);
|
2668 3207 | model.check(|| {
|
2669 3208 | let (admission, connection_cell, requesting_cell) =
|
2670 3209 | bounded_peer_cells_with_limit(2, EligibilityGroup::Pool, EligibilityGroup::Pool);
|
2671 3210 | let owning_lease = OriginAdmission::lease_for_test(&admission);
|
2672 3211 | let (owned_connection, _owning_physical) =
|
2673 3212 | ConnectionState::bounded(connection_info(1), owning_lease);
|
2674 3213 | OriginCell::insert_idle_h1(&connection_cell, owned_connection, H1Sender::test(11));
|
2675 3214 | let requesting_lease = OriginAdmission::lease_for_test(&admission);
|
2676 3215 | let (requesting_connection, _requesting_physical) =
|
2677 3216 | ConnectionState::bounded(connection_info(2), requesting_lease);
|
2678 3217 | let local = OriginCell::insert_selected_h1(
|
2679 3218 | &requesting_cell,
|
2680 3219 | requesting_connection,
|
2681 3220 | H1Sender::test(22),
|
2682 3221 | );
|
2683 3222 | let (waiter, demand) =
|
2684 3223 | requesting_cell.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
|
2685 3224 | let install = OriginAdmission::submit_action_without_running(
|
2686 3225 | &admission,
|
2687 3226 | requesting_cell.id().partition(),
|
2688 3227 | demand,
|
2689 3228 | )
|
2690 3229 | .expect("peer demand did not prepare an H1 match");
|
2691 - | let delivery = install
|
2692 - | .run_once_for_test()
|
2693 - | .expect("H1 supplier reservation did not prepare a delivery");
|
2694 - |
|
2695 3230 | let delivering = loom::thread::spawn(move || {
|
2696 - | OriginAdmission::run_action_chain(Some(delivery));
|
3231 + | OriginAdmission::run_action_chain(Some(install));
|
2697 3232 | });
|
2698 3233 | let returning = loom::thread::spawn(move || drop(local));
|
2699 3234 | delivering.join().unwrap();
|
2700 3235 | returning.join().unwrap();
|
2701 3236 |
|
2702 3237 | let selected = requesting_cell
|
2703 3238 | .take_ready_h1(waiter)
|
2704 3239 | .expect("delivery/return race did not satisfy the requesting cell waiter");
|
2705 3240 | drop(selected);
|
3241 + | let mut sender_ids = connection_cell.h1_idle_sender_ids();
|
3242 + | sender_ids.extend(requesting_cell.h1_idle_sender_ids());
|
3243 + | sender_ids.sort_unstable();
|
3244 + | assert_eq!(
|
3245 + | vec![11, 22],
|
3246 + | sender_ids,
|
3247 + | "delivery/return race duplicated or lost an HTTP/1 sender"
|
3248 + | );
|
2706 3249 | assert_eq!(
|
2707 3250 | 2,
|
2708 3251 | connection_cell.h1_counts().0 + requesting_cell.h1_counts().0
|
2709 3252 | );
|
2710 3253 | assert_eq!(
|
2711 3254 | 2,
|
2712 3255 | connection_cell.h1_counts().1 + requesting_cell.h1_counts().1
|
2713 3256 | );
|
2714 3257 | assert!(connection_cell.state.lock().h1.peer_reservation_available());
|
2715 3258 | admission.clear_modeled_cells_for_test();
|
2716 3259 | });
|
2717 3260 | }
|
2718 3261 |
|
3262 + | /// Races peer-sender materialization with close of the owning cell record.
|
3263 + | ///
|
3264 + | /// The request must receive the sender or establishment capacity, never
|
3265 + | /// become stranded. Three preemptions cover the delivery-close-delivery
|
3266 + | /// schedule across the two cell locks and admission lock.
|
2719 3267 | #[test]
|
2720 3268 | fn borrow_materialization_races_owning_cell_close_without_stranding_request() {
|
2721 3269 | let mut model = loom::model::Builder::new();
|
2722 - | model.preemption_bound = Some(2);
|
3270 + | model.preemption_bound.get_or_insert(3);
|
2723 3271 | model.check(|| {
|
2724 3272 | let (admission, connection_cell, requesting_cell) =
|
2725 3273 | bounded_peer_cells(EligibilityGroup::Pool, EligibilityGroup::Pool);
|
2726 3274 | let lease = OriginAdmission::lease_for_test(&admission);
|
2727 3275 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
|
2728 3276 | OriginCell::insert_idle_h1(&connection_cell, connection.clone(), H1Sender::test(11));
|
2729 3277 | let close = H1CloseHandle::new(&connection_cell, &connection);
|
2730 3278 | let (waiter, demand) =
|
2731 3279 | requesting_cell.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
|
2732 3280 | let install = OriginAdmission::submit_action_without_running(
|
2733 3281 | &admission,
|
2734 3282 | requesting_cell.id().partition(),
|
2735 3283 | demand,
|
2736 3284 | )
|
2737 3285 | .expect("peer demand did not prepare an H1 match");
|
2738 - | let delivery = install
|
2739 - | .run_once_for_test()
|
2740 - | .expect("H1 supplier reservation did not prepare a delivery");
|
2741 - |
|
2742 3286 | let delivering = loom::thread::spawn(move || {
|
2743 - | OriginAdmission::run_action_chain(Some(delivery));
|
3287 + | OriginAdmission::run_action_chain(Some(install));
|
2744 3288 | });
|
2745 3289 | let closing = loom::thread::spawn(move || close.close(CloseReason::Poisoned));
|
2746 3290 | delivering.join().unwrap();
|
2747 3291 | closing.join().unwrap();
|
2748 3292 |
|
2749 3293 | let event = requesting_cell
|
2750 3294 | .take_ready_event(waiter)
|
2751 3295 | .expect("borrow/close race stranded its requesting cell waiter");
|
2752 3296 | match event {
|
2753 3297 | AcquisitionStep::Resolved(AcquisitionOutcome::H1(selection)) => {
|
2754 3298 | drop(selection);
|
2755 3299 | }
|
2756 3300 | AcquisitionStep::StartEstablishment(permit) => {
|
2757 3301 | assert!(OriginCell::cancel_waiter(&requesting_cell, waiter));
|
2758 3302 | drop(permit);
|
2759 3303 | }
|
2760 3304 | AcquisitionStep::Resolved(AcquisitionOutcome::Failed(error)) => {
|
2761 3305 | panic!("unexpected establishment failure: {error}")
|
2762 3306 | }
|
2763 3307 | AcquisitionStep::Resolved(AcquisitionOutcome::RetryAcquisition) => {
|
2764 3308 | panic!("HTTP/1 reuse model requested reacquisition")
|
2765 3309 | }
|
2766 3310 | AcquisitionStep::Resolved(AcquisitionOutcome::H2(_)) => {
|
2767 3311 | panic!("HTTP/1 reuse model received an HTTP/2 activation")
|
2768 3312 | }
|
2769 3313 | }
|
2770 3314 | assert_eq!(0, requesting_cell.probe().retained);
|
2771 3315 | assert_eq!(1, admission.available_capacity_for_test());
|
2772 3316 | admission.clear_modeled_cells_for_test();
|
2773 3317 | });
|
2774 3318 | }
|
2775 3319 |
|
3320 + | /// Races HTTP/1 capacity reclaim with independent close of the same connection.
|
3321 + | ///
|
3322 + | /// Capacity must return exactly once. Three preemptions cover close and
|
3323 + | /// reclaim entering in either order and the resulting admission delivery.
|
2776 3324 | #[test]
|
2777 3325 | fn reclaim_and_connection_close_release_exactly_one_capacity_slot() {
|
2778 3326 | let mut model = loom::model::Builder::new();
|
2779 - | model.preemption_bound = Some(2);
|
3327 + | model.preemption_bound.get_or_insert(3);
|
2780 3328 | model.check(|| {
|
2781 3329 | let (admission, connection_cell, requesting_cell) = bounded_peer_cells(
|
2782 3330 | EligibilityGroup::Partition(PartitionId::from_index(1)),
|
2783 3331 | EligibilityGroup::Partition(PartitionId::from_index(2)),
|
2784 3332 | );
|
2785 3333 | let lease = OriginAdmission::lease_for_test(&admission);
|
2786 3334 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
|
2787 3335 | let returning = OriginCell::insert_selected_h1(
|
2788 3336 | &connection_cell,
|
2789 3337 | connection.clone(),
|
2790 3338 | H1Sender::test(11),
|
2791 3339 | );
|
2792 3340 | let close = H1CloseHandle::new(&connection_cell, &connection);
|
2793 3341 | let waiter =
|
2794 3342 | OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H1Compatible);
|
2795 3343 | let local_waiter =
|
2796 3344 | OriginCell::register_waiter(&connection_cell, ProtocolRequirement::H1Compatible);
|
2797 3345 |
|
2798 3346 | // Loom's default 4 KiB coroutine stack is too small for the full
|
2799 3347 | // sender-return and connection-close paths modeled here.
|
2800 3348 | let returning = loom::thread::Builder::new()
|
2801 3349 | .name("returning H1 sender".into())
|
2802 3350 | .stack_size(16 * 1024)
|
2803 3351 | .spawn(move || drop(returning))
|
2804 3352 | .unwrap();
|
2805 3353 | let closing = loom::thread::Builder::new()
|
2806 3354 | .name("closing H1 connection".into())
|
2807 3355 | .stack_size(16 * 1024)
|
2808 3356 | .spawn(move || close.close(CloseReason::Poisoned))
|
2809 3357 | .unwrap();
|
2810 3358 | returning.join().unwrap();
|
2811 3359 | let close_won = closing.join().unwrap();
|
2812 3360 |
|
2813 3361 | assert_eq!(
|
2814 3362 | !close_won,
|
2815 3363 | connection_cell.state.lock().h1.local_turn_owed(),
|
2816 3364 | "only a completed reclaim may earn the connection-owning cell a local fairness turn"
|
2817 3365 | );
|
2818 3366 |
|
2819 3367 | let replacement = OriginCell::take_ready_lease(&requesting_cell, waiter)
|
2820 3368 | .expect("close/reclaim race did not deliver released capacity");
|
2821 3369 | assert_eq!((0, 0), connection_cell.h1_counts());
|
2822 3370 | assert_eq!(0, admission.available_capacity_for_test());
|
2823 3371 | assert!(OriginCell::cancel_waiter(&connection_cell, local_waiter));
|
2824 3372 | drop(replacement);
|
2825 3373 | assert_eq!(1, admission.available_capacity_for_test());
|
2826 3374 | admission.clear_modeled_cells_for_test();
|
2827 3375 | });
|
2828 3376 | }
|
2829 3377 |
|
3378 + | /// Races bounded-capacity delivery with cancellation of its target waiter.
|
3379 + | ///
|
3380 + | /// Refusal must refunnel the permit after the cell lock is released.
|
2830 3381 | #[test]
|
2831 3382 | fn delivery_and_cancellation_race_refunnels_capacity() {
|
2832 3383 | loom::model(|| {
|
2833 3384 | let (admission, cell) = bounded_cell();
|
2834 3385 | let (first, demand) =
|
2835 3386 | cell.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
|
2836 - | let mut delivery =
|
3387 + | let delivery =
|
2837 3388 | OriginAdmission::submit_without_running(&admission, cell.id().partition(), demand)
|
2838 3389 | .unwrap();
|
2839 - | assert!(
|
2840 - | delivery.resolve_payload_for_test(),
|
2841 - | "capacity delivery did not materialize"
|
2842 - | );
|
2843 3390 |
|
2844 - | let delivery_cell = cell.clone();
|
2845 3391 | let deliver = loom::thread::spawn(move || {
|
2846 - | drop(OriginCell::receive_delivery(&delivery_cell, delivery));
|
3392 + | OriginAdmission::run_action_chain(Some(AdmissionAction::Deliver(delivery)));
|
2847 3393 | });
|
2848 3394 | let cancel_cell = cell.clone();
|
2849 3395 | let cancel =
|
2850 3396 | loom::thread::spawn(move || OriginCell::cancel_waiter(&cancel_cell, first));
|
2851 3397 | deliver.join().unwrap();
|
2852 3398 | cancel.join().unwrap();
|
2853 3399 |
|
2854 3400 | assert_eq!(1, admission.available_capacity_for_test());
|
2855 3401 | admission.clear_modeled_cells_for_test();
|
2856 3402 | });
|
2857 3403 | }
|
2858 3404 |
|
3405 + | /// Races cell-local peer route attachment with waiter cancellation.
|
3406 + | ///
|
3407 + | /// This model isolates the requesting-cell transition. The production
|
3408 + | /// admission preparation and settlement path is exercised separately above.
|
2859 3409 | #[test]
|
2860 - | fn h2_route_installation_races_requesting_cell_cancellation() {
|
3410 + | fn h2_route_attachment_linearizes_against_requesting_cell_cancellation() {
|
2861 3411 | let mut model = loom::model::Builder::new();
|
2862 - | model.preemption_bound = Some(2);
|
3412 + | model.preemption_bound.get_or_insert(3);
|
2863 3413 | model.check(|| {
|
2864 3414 | let (admission, connection_cell, requesting_cell) =
|
2865 3415 | bounded_peer_cells(EligibilityGroup::Pool, EligibilityGroup::Pool);
|
2866 3416 | let lease = OriginAdmission::lease_for_test(&admission);
|
2867 3417 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
|
2868 3418 | let generation = OriginCell::install_h2_for_test(&connection_cell, connection, 1, None);
|
2869 3419 | let (waiter, demand) =
|
2870 3420 | requesting_cell.register_waiter_without_publish(ProtocolRequirement::H2Required);
|
2871 3421 | let route = h2::H2Route::new(&connection_cell, generation);
|
2872 3422 |
|
2873 3423 | let install_cell = requesting_cell.clone();
|
2874 3424 | let installing = loom::thread::spawn(move || {
|
2875 3425 | OriginCell::attach_h2_route(
|
2876 3426 | &install_cell,
|
2877 3427 | route,
|
2878 3428 | &EligibilityGroup::Pool,
|
2879 3429 | demand.id_for_test(),
|
2880 3430 | )
|
2881 3431 | });
|
2882 3432 | let cancel_cell = requesting_cell.clone();
|
2883 3433 | let cancelling =
|
2884 3434 | loom::thread::spawn(move || OriginCell::cancel_waiter(&cancel_cell, waiter));
|
2885 3435 | let _installed = installing.join().unwrap();
|
2886 3436 | assert!(cancelling.join().unwrap());
|
2887 3437 |
|
2888 3438 | assert_eq!(0, requesting_cell.probe().retained);
|
2889 - | assert_eq!(0, admission.ordered_demand_count_for_test());
|
2890 3439 | assert_eq!(0, admission.available_capacity_for_test());
|
2891 3440 | assert!(OriginCell::close_h2(
|
2892 3441 | &connection_cell,
|
2893 3442 | generation,
|
2894 3443 | CloseReason::PoolDropped,
|
2895 3444 | ));
|
2896 3445 | assert_eq!(1, admission.available_capacity_for_test());
|
2897 3446 | admission.clear_modeled_cells_for_test();
|
2898 3447 | });
|
2899 3448 | }
|
2900 3449 |
|
3450 + | /// Races route publication, generation close, and peer route service.
|
3451 + | ///
|
3452 + | /// The three-preemption bound permits each actor to intervene before route
|
3453 + | /// settlement returns through admission to the requesting cell.
|
2901 3454 | #[test]
|
2902 3455 | fn h2_route_acknowledgement_races_generation_close_and_route_service() {
|
2903 3456 | let mut model = loom::model::Builder::new();
|
2904 - | model.preemption_bound = Some(3);
|
3457 + | model.preemption_bound.get_or_insert(3);
|
2905 3458 | model.check(|| {
|
2906 3459 | let (admission, connection_cell, requesting_cell) =
|
2907 3460 | bounded_peer_cells(EligibilityGroup::Pool, EligibilityGroup::Pool);
|
2908 3461 | let lease = OriginAdmission::lease_for_test(&admission);
|
2909 3462 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
|
2910 3463 | let generation = OriginCell::install_h2_for_test(&connection_cell, connection, 1, None);
|
2911 3464 | let (waiter, demand) =
|
2912 3465 | requesting_cell.register_waiter_without_publish(ProtocolRequirement::H2Required);
|
2913 3466 | let action = OriginAdmission::submit_action_without_running(
|
2914 3467 | &admission,
|
2915 3468 | requesting_cell.id().partition(),
|
2916 3469 | demand,
|
2917 3470 | )
|
2918 3471 | .expect("peer demand did not prepare a route");
|
2919 3472 |
|
2920 3473 | let publishing = loom::thread::spawn(move || {
|
2921 3474 | OriginAdmission::run_action_chain(Some(action));
|
2922 3475 | });
|
2923 3476 | let closing_cell = connection_cell.clone();
|
2924 3477 | let closing = loom::thread::spawn(move || {
|
2925 3478 | OriginCell::close_h2(&closing_cell, generation, CloseReason::Poisoned)
|
2926 3479 | });
|
2927 3480 | let servicing_cell = requesting_cell.clone();
|
2928 3481 | let servicing = loom::thread::spawn(move || {
|
2929 3482 | OriginCell::offer_peer_h2(&servicing_cell);
|
2930 3483 | });
|
2931 3484 |
|
2932 3485 | publishing.join().unwrap();
|
2933 3486 | assert!(closing.join().unwrap());
|
2934 3487 | servicing.join().unwrap();
|
2935 3488 |
|
2936 3489 | assert!(
|
2937 3490 | OriginCell::cancel_waiter(&requesting_cell, waiter),
|
2938 3491 | "route acknowledgement lost the live demand"
|
2939 3492 | );
|
2940 3493 | assert_eq!(0, requesting_cell.probe().retained);
|
2941 3494 | assert_eq!(0, admission.ordered_demand_count_for_test());
|
2942 3495 | assert_eq!(1, admission.available_capacity_for_test());
|
2943 3496 | admission.clear_modeled_cells_for_test();
|
2944 3497 | });
|
2945 3498 | }
|
2946 3499 |
|
3500 + | /// Races peer route publication with close of the routed generation.
|
3501 + | ///
|
3502 + | /// Three preemptions cover either actor entering first and the stale route
|
3503 + | /// acknowledgement returning to admission.
|
2947 3504 | #[test]
|
2948 3505 | fn h2_route_close_race_preserves_capacity_ownership() {
|
2949 3506 | let mut model = loom::model::Builder::new();
|
2950 - | model.preemption_bound = Some(2);
|
3507 + | model.preemption_bound.get_or_insert(3);
|
2951 3508 | model.check(|| {
|
2952 3509 | let (admission, connection_cell, requesting_cell) =
|
2953 3510 | bounded_peer_cells(EligibilityGroup::Pool, EligibilityGroup::Pool);
|
2954 3511 | let lease = OriginAdmission::lease_for_test(&admission);
|
2955 3512 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
|
2956 3513 | let generation = OriginCell::install_h2_for_test(&connection_cell, connection, 1, None);
|
2957 3514 | let (waiter, demand) =
|
2958 3515 | requesting_cell.register_waiter_without_publish(ProtocolRequirement::H2Required);
|
2959 3516 | let action = OriginAdmission::submit_action_without_running(
|
2960 3517 | &admission,
|