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