| 244 | |
| 245 | #[test] |
| 246 | fn concurrent_group_commit() { |
| 247 | let dir = tempfile::tempdir().unwrap(); |
| 248 | let path = dir.path().join("test.wal"); |
| 249 | |
| 250 | let writer = Arc::new(Mutex::new( |
| 251 | WalWriter::open_without_direct_io(&path).unwrap(), |
| 252 | )); |
| 253 | let gc = Arc::new(GroupCommitter::new()); |
| 254 | |
| 255 | let mut handles = Vec::new(); |
| 256 | |
| 257 | for i in 0..10 { |
| 258 | let w = Arc::clone(&writer); |
| 259 | let g = Arc::clone(&gc); |
| 260 | handles.push(thread::spawn(move || { |
| 261 | let payload = format!("record-{i}"); |
| 262 | let result = g |
| 263 | .submit( |
| 264 | &w, |
| 265 | PendingWrite { |
| 266 | record_type: RecordType::Put as u32, |
| 267 | tenant_id: 1, |
| 268 | vshard_id: 0, |
| 269 | database_id: 0, |
| 270 | payload: payload.into_bytes(), |
| 271 | }, |
| 272 | ) |
| 273 | .unwrap(); |
| 274 | assert!(result.durable); |
| 275 | result.lsn |
| 276 | })); |
| 277 | } |
| 278 | |
| 279 | let lsns: Vec<u64> = handles.into_iter().map(|h| h.join().unwrap()).collect(); |
| 280 | |
| 281 | // All should have gotten durable results. |
| 282 | assert!(lsns.iter().all(|l| *l > 0)); |
| 283 | |
| 284 | // All 10 records should be in the WAL. |
| 285 | let reader = WalReader::open(&path).unwrap(); |
| 286 | let records: Vec<_> = reader |
| 287 | .records() |
| 288 | .collect::<crate::error::Result<_>>() |
| 289 | .unwrap(); |
| 290 | assert_eq!(records.len(), 10); |
| 291 | } |
| 292 | } |