| 238 | |
| 239 | #[test] |
| 240 | fn concurrent_serialization() { |
| 241 | // Verify that reserve() on the same sequence blocks until commit/rollback. |
| 242 | let mgr = Arc::new(GapFreeManager::new()); |
| 243 | let counter = Arc::new(AtomicI64::new(0)); |
| 244 | let order = Arc::new(AtomicI64::new(0)); |
| 245 | |
| 246 | let mgr2 = Arc::clone(&mgr); |
| 247 | let counter2 = Arc::clone(&counter); |
| 248 | let order2 = Arc::clone(&order); |
| 249 | |
| 250 | // First reservation — held for a bit. |
| 251 | let h1 = mgr |
| 252 | .reserve("1:test", || Ok(counter.fetch_add(1, Ordering::SeqCst) + 1)) |
| 253 | .unwrap(); |
| 254 | assert_eq!(h1.value, 1); |
| 255 | order.store(1, Ordering::SeqCst); |
| 256 | |
| 257 | // Second reservation in another thread — should block. |
| 258 | let t = std::thread::spawn(move || { |
| 259 | let h2 = mgr2 |
| 260 | .reserve("1:test", || Ok(counter2.fetch_add(1, Ordering::SeqCst) + 1)) |
| 261 | .unwrap(); |
| 262 | // When we get here, h1 must have been committed (order == 2). |
| 263 | assert!(order2.load(Ordering::SeqCst) >= 2); |
| 264 | assert_eq!(h2.value, 2); |
| 265 | mgr2.commit(&h2); |
| 266 | }); |
| 267 | |
| 268 | // Small sleep to let thread start and block on reserve(). |
| 269 | std::thread::sleep(std::time::Duration::from_millis(50)); |
| 270 | order.store(2, Ordering::SeqCst); |
| 271 | mgr.commit(&h1); |
| 272 | |
| 273 | t.join().unwrap(); |
| 274 | assert_eq!(counter.load(Ordering::SeqCst), 2); |
| 275 | } |
| 276 | } |