| 43 | |
| 44 | #[test] |
| 45 | fn bridge_throughput_flat_memory() { |
| 46 | let (mut producer, mut consumer) = RingBuffer::channel::<TestMessage>(8192); |
| 47 | |
| 48 | let total_produced = Arc::new(AtomicU64::new(0)); |
| 49 | let total_consumed = Arc::new(AtomicU64::new(0)); |
| 50 | |
| 51 | let consumed = Arc::clone(&total_consumed); |
| 52 | |
| 53 | // Consumer thread (simulates TPC core). |
| 54 | let consumer_handle = thread::spawn(move || { |
| 55 | let mut count = 0u64; |
| 56 | let mut last_id = 0u64; |
| 57 | |
| 58 | loop { |
| 59 | match consumer.try_pop() { |
| 60 | Ok(msg) => { |
| 61 | // Verify ordering. |
| 62 | assert!( |
| 63 | msg.id > last_id || last_id == 0, |
| 64 | "out-of-order: got {} after {}", |
| 65 | msg.id, |
| 66 | last_id |
| 67 | ); |
| 68 | last_id = msg.id; |
| 69 | count += 1; |
| 70 | consumed.store(count, Ordering::Relaxed); |
| 71 | |
| 72 | if count >= MESSAGE_COUNT { |
| 73 | break; |
| 74 | } |
| 75 | } |
| 76 | Err(nodedb_bridge::BridgeError::Empty) => { |
| 77 | thread::yield_now(); |
| 78 | } |
| 79 | Err(nodedb_bridge::BridgeError::Disconnected { .. }) => break, |
| 80 | Err(e) => panic!("unexpected error: {e}"), |
| 81 | } |
| 82 | } |
| 83 | |
| 84 | count |
| 85 | }); |
| 86 | |
| 87 | // Producer thread (simulates Tokio Control Plane). |
| 88 | let start = Instant::now(); |
| 89 | let mut pushed = 0u64; |
| 90 | let mut full_spins = 0u64; |
| 91 | |
| 92 | while pushed < MESSAGE_COUNT { |
| 93 | let msg = TestMessage::new(pushed + 1); |
| 94 | match producer.try_push(msg) { |
| 95 | Ok(()) => { |
| 96 | pushed += 1; |
| 97 | total_produced.store(pushed, Ordering::Relaxed); |
| 98 | } |
| 99 | Err(nodedb_bridge::BridgeError::Full { .. }) => { |
| 100 | full_spins += 1; |
| 101 | thread::yield_now(); |
| 102 | } |