mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-22 06:22:38 +00:00
1144bf15d4
* rust volume: group-commit fsync uploads on the direct write path Each fsync=true upload used to run its own .dat/.idx sync when the write queue was off, which is the default. Route durable uploads through the per-volume write workers so a batch shares one sync, matching Go's writeNeedle2: non-fsync writes, and writes while the server is stopping, stay on the synchronous path. The queue is now always constructed and SEAWEED_WRITE_QUEUE is gone. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * rust volume: document post_handler write routing Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
738 lines
28 KiB
Rust
738 lines
28 KiB
Rust
//! Async batched write processing for the volume server.
|
|
//!
|
|
//! Durable (fsync) uploads are submitted here instead of each handler calling
|
|
//! `write_needle` directly — Go's `writeNeedle2` makes the same split, so the
|
|
//! queue is always constructed. Each volume gets its own FIFO worker (Go's
|
|
//! volume_write.go keeps one batching worker per volume), which drains its
|
|
//! channel in batches of up to 128 entries or 4 MiB and shares the .dat/.idx
|
|
//! flushes across durable writes (see `Volume::write_needles_grouped`).
|
|
//! Writes to different volumes proceed in parallel; a volume's ordering and
|
|
//! exclusion still come from its data-file write lease.
|
|
|
|
use std::collections::HashMap;
|
|
use std::collections::hash_map::Entry;
|
|
use std::sync::{Arc, Mutex};
|
|
|
|
use tokio::sync::{OwnedSemaphorePermit, Semaphore, mpsc, oneshot};
|
|
use tracing::debug;
|
|
|
|
use crate::storage::needle::needle::Needle;
|
|
use crate::storage::types::{Size, VolumeId};
|
|
use crate::storage::volume::VolumeError;
|
|
|
|
use super::volume_server::VolumeServerState;
|
|
|
|
/// Result of a single write operation: (offset, size, is_unchanged).
|
|
pub type WriteResult = Result<(u64, Size, bool), VolumeError>;
|
|
|
|
/// The needles queued for one volume, each with the durability it asked for.
|
|
type VolumeBatch = Vec<(Needle, bool, oneshot::Sender<WriteResult>)>;
|
|
|
|
/// A request to write a needle, submitted to the write queue.
|
|
pub struct WriteRequest {
|
|
pub volume_id: VolumeId,
|
|
pub needle: Needle,
|
|
pub fsync: bool,
|
|
pub response_tx: oneshot::Sender<WriteResult>,
|
|
/// The global queued-slot permit, owned by the request itself: a caller
|
|
/// that cancels `submit` after the send must not free the slot while the
|
|
/// worker still holds the needle.
|
|
pub queue_permit: Option<OwnedSemaphorePermit>,
|
|
}
|
|
|
|
/// Maximum number of write requests to batch together.
|
|
const MAX_BATCH_SIZE: usize = 128;
|
|
|
|
/// Maximum bytes to accumulate per batch before breaking (matches Go's 4MB limit).
|
|
/// This prevents large writes from accumulating unbounded latency.
|
|
const MAX_BATCH_BYTES: usize = 4 * 1024 * 1024;
|
|
|
|
/// Cap on volumes doing grouped I/O at once, so many writers cannot crowd
|
|
/// readers off a slow disk the way unlimited parallel fsyncs would.
|
|
const MAX_CONCURRENT_VOLUME_WRITES: usize = 8;
|
|
|
|
/// How often an idle worker re-checks that its volume is still mounted.
|
|
/// Workers live for their volume's queue lifetime (Go parity), so the only
|
|
/// retirement is the volume leaving the store.
|
|
const IDLE_WORKER_POLL: std::time::Duration = std::time::Duration::from_secs(60);
|
|
|
|
/// Handle for submitting write requests to the per-volume workers.
|
|
#[derive(Clone)]
|
|
pub struct WriteQueue {
|
|
state: Arc<VolumeServerState>,
|
|
/// Per-volume channel capacity; backpressure kicks in when full.
|
|
capacity: usize,
|
|
io_permits: Arc<Semaphore>,
|
|
/// Bounds queued+in-flight writes across all volumes — per-volume channels
|
|
/// alone would multiply the old single queue's capacity by volume count.
|
|
queue_permits: Arc<Semaphore>,
|
|
workers: Arc<Mutex<HashMap<VolumeId, mpsc::Sender<WriteRequest>>>>,
|
|
/// Submissions accepted, for tests asserting which writes route here.
|
|
#[cfg(test)]
|
|
submits: Arc<std::sync::atomic::AtomicU64>,
|
|
}
|
|
|
|
impl WriteQueue {
|
|
/// Create a new write queue. Per-volume workers are spawned lazily on the
|
|
/// first submission for that volume.
|
|
pub fn new(state: Arc<VolumeServerState>, capacity: usize) -> Self {
|
|
WriteQueue {
|
|
state,
|
|
capacity,
|
|
io_permits: Arc::new(Semaphore::new(MAX_CONCURRENT_VOLUME_WRITES)),
|
|
queue_permits: Arc::new(Semaphore::new(capacity)),
|
|
workers: Arc::new(Mutex::new(HashMap::new())),
|
|
#[cfg(test)]
|
|
submits: Arc::new(std::sync::atomic::AtomicU64::new(0)),
|
|
}
|
|
}
|
|
|
|
/// How many writes were submitted, for tests asserting upload routing.
|
|
#[cfg(test)]
|
|
pub fn submits_for_test(&self) -> u64 {
|
|
self.submits.load(std::sync::atomic::Ordering::Relaxed)
|
|
}
|
|
|
|
/// Sender for the volume's worker, spawning one on first use.
|
|
fn worker_for(&self, volume_id: VolumeId) -> mpsc::Sender<WriteRequest> {
|
|
let mut workers = self.workers.lock().unwrap();
|
|
match workers.entry(volume_id) {
|
|
Entry::Occupied(e) => e.get().clone(),
|
|
Entry::Vacant(e) => {
|
|
let (tx, rx) = mpsc::channel(self.capacity);
|
|
tokio::spawn(
|
|
VolumeWriteWorker {
|
|
rx,
|
|
vid: volume_id,
|
|
state: self.state.clone(),
|
|
io_permits: self.io_permits.clone(),
|
|
workers: self.workers.clone(),
|
|
}
|
|
.run(),
|
|
);
|
|
e.insert(tx.clone());
|
|
tx
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Submit a write request and wait for the result.
|
|
///
|
|
/// Returns `Err` if the worker has shut down or the response channel was dropped.
|
|
pub async fn submit(&self, volume_id: VolumeId, needle: Needle, fsync: bool) -> WriteResult {
|
|
// Workers spawn only for mounted volumes — a bogus id never creates
|
|
// a channel, so volume-id sprays can't grow the worker map.
|
|
#[cfg(test)]
|
|
self.submits
|
|
.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
|
{
|
|
let store = self.state.store.read().unwrap();
|
|
if store.find_volume(volume_id).is_none() {
|
|
return Err(VolumeError::NotFound);
|
|
}
|
|
}
|
|
// The permit travels inside the request, so the global queued bound
|
|
// covers the request until the worker drops it — including after a
|
|
// canceled submit.
|
|
let queue_permit = self
|
|
.queue_permits
|
|
.clone()
|
|
.acquire_owned()
|
|
.await
|
|
.map_err(|_| broken_pipe("write queue shut down"))?;
|
|
let (response_tx, response_rx) = oneshot::channel();
|
|
let mut request = WriteRequest {
|
|
volume_id,
|
|
needle,
|
|
fsync,
|
|
response_tx,
|
|
queue_permit: Some(queue_permit),
|
|
};
|
|
|
|
// A send fails only against a retiring worker's closed channel; the
|
|
// request was never accepted, so retrying it behind a fresh worker
|
|
// keeps same-volume order. The volume recheck bounds the spin.
|
|
loop {
|
|
let tx = self.worker_for(volume_id);
|
|
match tx.send(request).await {
|
|
Ok(()) => break,
|
|
Err(mpsc::error::SendError(returned)) => {
|
|
request = returned;
|
|
{
|
|
let store = self.state.store.read().unwrap();
|
|
if store.find_volume(volume_id).is_none() {
|
|
return Err(VolumeError::NotFound);
|
|
}
|
|
}
|
|
tokio::task::yield_now().await;
|
|
}
|
|
}
|
|
}
|
|
|
|
// Wait for the worker to process our request.
|
|
match response_rx.await {
|
|
Ok(result) => result,
|
|
Err(_) => Err(broken_pipe("write queue worker dropped response channel")),
|
|
}
|
|
}
|
|
}
|
|
|
|
fn broken_pipe(msg: &'static str) -> VolumeError {
|
|
VolumeError::Io(std::io::Error::new(std::io::ErrorKind::BrokenPipe, msg))
|
|
}
|
|
|
|
/// FIFO worker draining one volume's write channel in batches. Like Go's
|
|
/// per-volume async writer it runs for the volume's queue lifetime — exiting
|
|
/// only when the channel closes or its volume leaves the store — so no
|
|
/// handoff can ever split one volume's write order across two workers.
|
|
struct VolumeWriteWorker {
|
|
rx: mpsc::Receiver<WriteRequest>,
|
|
vid: VolumeId,
|
|
state: Arc<VolumeServerState>,
|
|
io_permits: Arc<Semaphore>,
|
|
workers: Arc<Mutex<HashMap<VolumeId, mpsc::Sender<WriteRequest>>>>,
|
|
}
|
|
|
|
impl VolumeWriteWorker {
|
|
async fn run(mut self) {
|
|
loop {
|
|
let first = match tokio::time::timeout(IDLE_WORKER_POLL, self.rx.recv()).await {
|
|
Ok(Some(req)) => req,
|
|
// recv() yields None only once every sender is gone — the map
|
|
// holds a sender per volume, so that is exactly queue shutdown.
|
|
Ok(None) => return,
|
|
Err(_) if self.volume_gone() => {
|
|
self.retire().await;
|
|
return;
|
|
}
|
|
Err(_) => continue,
|
|
};
|
|
self.process_from(first).await;
|
|
}
|
|
}
|
|
|
|
/// The volume's handle left the store, so the worker can go too; an idle
|
|
/// worker for a mounted volume stays (Go keeps the writer for the mount).
|
|
fn volume_gone(&self) -> bool {
|
|
let store = self.state.store.read().unwrap();
|
|
store.find_volume(self.vid).is_none()
|
|
}
|
|
|
|
/// Retire in a fixed order that keeps same-volume FIFO: close the channel
|
|
/// so racing senders fail fast (they retry and wait for a fresh worker),
|
|
/// drain what was already accepted, and only then give up the map slot —
|
|
/// a replacement worker can never start while this one still holds queued
|
|
/// writes.
|
|
async fn retire(&mut self) {
|
|
self.rx.close();
|
|
while let Some(first) = self.rx.recv().await {
|
|
self.process_from(first).await;
|
|
}
|
|
self.workers.lock().unwrap().remove(&self.vid);
|
|
}
|
|
|
|
/// Collect a batch around `first` and process it under the I/O bound.
|
|
async fn process_from(&mut self, first: WriteRequest) {
|
|
// Drain as many additional requests as available, up to MAX_BATCH_SIZE
|
|
// or MAX_BATCH_BYTES (matches Go: 128 requests or 4MB, whichever comes first).
|
|
let mut batch = Vec::with_capacity(MAX_BATCH_SIZE);
|
|
let mut batch_bytes: usize = first.needle.data.len();
|
|
batch.push(first);
|
|
while batch.len() < MAX_BATCH_SIZE && batch_bytes < MAX_BATCH_BYTES {
|
|
match self.rx.try_recv() {
|
|
Ok(req) => {
|
|
batch_bytes += req.needle.data.len();
|
|
batch.push(req);
|
|
}
|
|
Err(_) => break,
|
|
}
|
|
}
|
|
debug!("processing write batch of {} requests", batch.len());
|
|
|
|
// Bound the number of volumes doing grouped I/O at once, then run
|
|
// the batch in spawn_blocking since write_needle does file I/O.
|
|
let Ok(_permit) = self.io_permits.clone().acquire_owned().await else {
|
|
return;
|
|
};
|
|
let state = self.state.clone();
|
|
let _ = tokio::task::spawn_blocking(move || {
|
|
process_batch(state, batch);
|
|
})
|
|
.await;
|
|
}
|
|
}
|
|
|
|
/// Process a batch of write requests, grouped by volume ID.
|
|
///
|
|
/// Groups writes by volume to minimize the number of store lock acquisitions,
|
|
/// then sends results back via each request's oneshot channel.
|
|
fn process_batch(state: Arc<VolumeServerState>, batch: Vec<WriteRequest>) {
|
|
// Group requests by volume ID for efficient processing.
|
|
// We use a Vec of (VolumeId, Vec<(Needle, Sender)>) to preserve order
|
|
// and avoid requiring Hash on VolumeId.
|
|
let mut groups: Vec<(VolumeId, VolumeBatch)> = Vec::new();
|
|
|
|
for req in batch {
|
|
let vid = req.volume_id;
|
|
if let Some(group) = groups.iter_mut().find(|(v, _)| *v == vid) {
|
|
group.1.push((req.needle, req.fsync, req.response_tx));
|
|
} else {
|
|
groups.push((vid, vec![(req.needle, req.fsync, req.response_tx)]));
|
|
}
|
|
}
|
|
|
|
// Resolve each group's volume handle under a brief store read, then run
|
|
// the group's I/O with the guard released — one volume's write must not
|
|
// block admission on another volume's.
|
|
for (vid, entries) in groups {
|
|
let (mut writes, senders): (Vec<_>, Vec<_>) = entries
|
|
.into_iter()
|
|
.map(|(needle, fsync, response_tx)| ((needle, fsync), response_tx))
|
|
.unzip();
|
|
let vol = {
|
|
let store = state.store.read().unwrap();
|
|
store.writable_volume(vid)
|
|
};
|
|
let results = match vol {
|
|
Ok(vol) => vol.write_needles_grouped(&mut writes),
|
|
Err(e) => writes
|
|
.iter()
|
|
.map(|_| match e {
|
|
VolumeError::ReadOnly(vid) => Err(VolumeError::ReadOnly(vid)),
|
|
_ => Err(VolumeError::NotFound),
|
|
})
|
|
.collect(),
|
|
};
|
|
for (response_tx, result) in senders.into_iter().zip(results) {
|
|
// Send result back; ignore error if receiver dropped.
|
|
let _ = response_tx.send(result);
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::storage::types::VolumeId;
|
|
|
|
/// Helper to create a minimal VolumeServerState for testing.
|
|
fn make_test_state() -> Arc<VolumeServerState> {
|
|
use crate::security::{Guard, SigningKey};
|
|
use crate::server::volume_server::RuntimeMetricsConfig;
|
|
use crate::storage::needle_map::NeedleMapKind;
|
|
use crate::storage::store::Store;
|
|
use std::sync::RwLock;
|
|
use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU32};
|
|
|
|
let store = Store::new(NeedleMapKind::InMemory);
|
|
let guard = Guard::new(&[], SigningKey(vec![]), 0, SigningKey(vec![]), 0);
|
|
|
|
Arc::new(VolumeServerState {
|
|
store: RwLock::new(store),
|
|
guard: RwLock::new(guard),
|
|
is_stopping: RwLock::new(false),
|
|
maintenance: AtomicBool::new(false),
|
|
state_version: AtomicU32::new(0),
|
|
concurrent_upload_limit: 0,
|
|
concurrent_download_limit: 0,
|
|
inflight_upload_data_timeout: std::time::Duration::ZERO,
|
|
inflight_download_data_timeout: std::time::Duration::ZERO,
|
|
inflight_upload_bytes: AtomicI64::new(0),
|
|
inflight_download_bytes: AtomicI64::new(0),
|
|
upload_notify: tokio::sync::Notify::new(),
|
|
download_notify: tokio::sync::Notify::new(),
|
|
data_center: String::new(),
|
|
rack: String::new(),
|
|
file_size_limit_bytes: 0,
|
|
maintenance_byte_per_second: 0,
|
|
vacuum_commit_slots: None,
|
|
vacuum_commits_waiting: std::sync::atomic::AtomicU32::new(0),
|
|
is_heartbeating: AtomicBool::new(false),
|
|
has_master: false,
|
|
pre_stop_seconds: 0,
|
|
volume_state_notify: tokio::sync::Notify::new(),
|
|
write_queue: std::sync::OnceLock::new(),
|
|
read_mode: crate::config::ReadMode::Local,
|
|
allow_untrusted_remote_endpoints: false,
|
|
master_url: String::new(),
|
|
master_urls: Vec::new(),
|
|
seed_master_set: std::collections::HashSet::new(),
|
|
current_master_url: tokio::sync::RwLock::new(String::new()),
|
|
self_url: String::new(),
|
|
http_client: reqwest::Client::new(),
|
|
outgoing_http_scheme: "http".to_string(),
|
|
outgoing_grpc_tls: None,
|
|
metrics_runtime: std::sync::RwLock::new(RuntimeMetricsConfig::default()),
|
|
metrics_notify: tokio::sync::Notify::new(),
|
|
fix_jpg_orientation: false,
|
|
has_slow_read: true,
|
|
read_buffer_size_bytes: 4 * 1024 * 1024,
|
|
security_file: String::new(),
|
|
cli_white_list: vec![],
|
|
state_file_path: String::new(),
|
|
ec_decodes_in_flight: std::sync::Mutex::new(std::collections::HashSet::new()),
|
|
ec_decode_tail: std::sync::Mutex::new(std::collections::HashSet::new()),
|
|
ec_decode_tail_notify: tokio::sync::Notify::new(),
|
|
})
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_write_queue_submit_no_volume() {
|
|
// Submit a write to a non-existent volume -- should return VolumeError::NotFound.
|
|
let state = make_test_state();
|
|
let queue = WriteQueue::new(state, MAX_BATCH_SIZE);
|
|
|
|
let needle = Needle {
|
|
id: 1.into(),
|
|
cookie: 0x12345678.into(),
|
|
data: vec![1, 2, 3],
|
|
data_size: 3,
|
|
..Needle::default()
|
|
};
|
|
|
|
let result = queue.submit(VolumeId(999), needle, false).await;
|
|
assert!(result.is_err());
|
|
match result {
|
|
Err(VolumeError::NotFound) => {} // expected
|
|
other => panic!("expected NotFound, got {:?}", other),
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_write_queue_concurrent_submissions() {
|
|
// Submit multiple concurrent writes -- all should complete (with errors since no volume).
|
|
let state = make_test_state();
|
|
let queue = WriteQueue::new(state, MAX_BATCH_SIZE);
|
|
|
|
let mut handles = Vec::new();
|
|
for i in 0..10u64 {
|
|
let q = queue.clone();
|
|
handles.push(tokio::spawn(async move {
|
|
let needle = Needle {
|
|
id: i.into(),
|
|
cookie: 0xABCD.into(),
|
|
data: vec![i as u8; 10],
|
|
data_size: 10,
|
|
..Needle::default()
|
|
};
|
|
q.submit(VolumeId(1), needle, false).await
|
|
}));
|
|
}
|
|
|
|
for handle in handles {
|
|
let result = handle.await.unwrap();
|
|
// All should fail with NotFound since there's no volume 1
|
|
assert!(matches!(result, Err(VolumeError::NotFound)));
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_write_queue_batching() {
|
|
// Verify that many concurrent writes get processed (testing the batching path).
|
|
let state = make_test_state();
|
|
let queue = WriteQueue::new(state, MAX_BATCH_SIZE);
|
|
|
|
// Submit MAX_BATCH_SIZE requests concurrently
|
|
let mut handles = Vec::new();
|
|
for i in 0..MAX_BATCH_SIZE as u64 {
|
|
let q = queue.clone();
|
|
handles.push(tokio::spawn(async move {
|
|
let needle = Needle {
|
|
id: i.into(),
|
|
cookie: 0x1111.into(),
|
|
data: vec![0u8; 4],
|
|
data_size: 4,
|
|
..Needle::default()
|
|
};
|
|
q.submit(VolumeId(42), needle, false).await
|
|
}));
|
|
}
|
|
|
|
let mut results = Vec::new();
|
|
for handle in handles {
|
|
results.push(handle.await.unwrap());
|
|
}
|
|
|
|
// All should complete (with NotFound errors since no volume exists)
|
|
assert_eq!(results.len(), MAX_BATCH_SIZE);
|
|
for r in results {
|
|
assert!(matches!(r, Err(VolumeError::NotFound)));
|
|
}
|
|
}
|
|
|
|
/// The queue hands a volume's batch to the grouped path, so ten durable
|
|
/// writes cost one .dat sync and one .idx sync, not ten of each.
|
|
#[test]
|
|
fn test_process_batch_group_commits_fsync_writes() {
|
|
use crate::config::MinFreeSpace;
|
|
use crate::storage::types::{DiskType, NeedleId};
|
|
use crate::storage::volume::VolumeSpec;
|
|
|
|
let tmp = tempfile::TempDir::new().unwrap();
|
|
let dir = tmp.path().to_str().unwrap();
|
|
let state = make_test_state();
|
|
{
|
|
let mut store = state.store.write().unwrap();
|
|
store
|
|
.add_location(
|
|
dir,
|
|
dir,
|
|
10,
|
|
DiskType::HardDrive,
|
|
MinFreeSpace::Percent(1.0),
|
|
Vec::new(),
|
|
)
|
|
.unwrap();
|
|
store
|
|
.add_volume(VolumeId(1), DiskType::HardDrive, &VolumeSpec::default())
|
|
.unwrap();
|
|
}
|
|
|
|
let mut receivers = Vec::new();
|
|
let batch = (1..=10u64)
|
|
.map(|id| {
|
|
let (response_tx, response_rx) = oneshot::channel();
|
|
receivers.push(response_rx);
|
|
WriteRequest {
|
|
volume_id: VolumeId(1),
|
|
needle: Needle {
|
|
id: NeedleId(id),
|
|
cookie: 0x1111.into(),
|
|
data: vec![id as u8; 8],
|
|
data_size: 8,
|
|
..Needle::default()
|
|
},
|
|
fsync: true,
|
|
response_tx,
|
|
queue_permit: None,
|
|
}
|
|
})
|
|
.collect();
|
|
process_batch(state.clone(), batch);
|
|
|
|
for mut rx in receivers {
|
|
assert!(matches!(rx.try_recv().unwrap(), Ok((_, _, false))));
|
|
}
|
|
let store = state.store.read().unwrap();
|
|
let (_, vol) = store.find_volume(VolumeId(1)).unwrap();
|
|
assert_eq!(vol.sync_counts_for_test(), (1, 1));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_write_queue_dropped_sender() {
|
|
// When the queue is dropped, subsequent submits should fail gracefully.
|
|
let state = make_test_state();
|
|
let queue = WriteQueue::new(state, 1);
|
|
|
|
// Clone then drop the original -- the worker keeps running via its rx handle.
|
|
let queue2 = queue.clone();
|
|
drop(queue);
|
|
|
|
// This should still work since the worker is alive.
|
|
let needle = Needle {
|
|
id: 1.into(),
|
|
cookie: 0.into(),
|
|
data: vec![],
|
|
data_size: 0,
|
|
..Needle::default()
|
|
};
|
|
let result = queue2.submit(VolumeId(1), needle, false).await;
|
|
assert!(result.is_err()); // NotFound is fine -- the point is it doesn't panic
|
|
}
|
|
|
|
/// Issue #791: a write parked inside its .dat append must not hold the
|
|
/// store guard, so another volume still resolves and writes. The hook
|
|
/// runs under the volume's data-file lease, so the assertion is that the
|
|
/// store — not the per-volume serialization — stays open.
|
|
#[test]
|
|
fn test_process_batch_does_not_block_other_volumes() {
|
|
use crate::config::MinFreeSpace;
|
|
use crate::storage::types::{DiskType, NeedleId};
|
|
use crate::storage::volume::{VolumeSpec, needle_write_hook};
|
|
use std::sync::Barrier;
|
|
|
|
let tmp = tempfile::TempDir::new().unwrap();
|
|
let dir = tmp.path().to_str().unwrap();
|
|
let state = make_test_state();
|
|
{
|
|
let mut store = state.store.write().unwrap();
|
|
store
|
|
.add_location(
|
|
dir,
|
|
dir,
|
|
10,
|
|
DiskType::HardDrive,
|
|
MinFreeSpace::Percent(1.0),
|
|
Vec::new(),
|
|
)
|
|
.unwrap();
|
|
for vid in [VolumeId(1), VolumeId(2)] {
|
|
store
|
|
.add_volume(vid, DiskType::HardDrive, &VolumeSpec::default())
|
|
.unwrap();
|
|
}
|
|
}
|
|
|
|
let in_append = Arc::new(Barrier::new(2));
|
|
let release = Arc::new(Barrier::new(2));
|
|
let _reg = needle_write_hook::register(NeedleId(101), {
|
|
let in_append = in_append.clone();
|
|
let release = release.clone();
|
|
move |_| {
|
|
in_append.wait();
|
|
release.wait();
|
|
}
|
|
});
|
|
|
|
let (response_tx, mut response_rx) = oneshot::channel();
|
|
let batch = vec![WriteRequest {
|
|
volume_id: VolumeId(1),
|
|
needle: Needle {
|
|
id: NeedleId(101),
|
|
cookie: 0x1111.into(),
|
|
data: vec![7u8; 8],
|
|
data_size: 8,
|
|
..Needle::default()
|
|
},
|
|
fsync: false,
|
|
response_tx,
|
|
queue_permit: None,
|
|
}];
|
|
let writer = {
|
|
let state = state.clone();
|
|
std::thread::spawn(move || process_batch(state, batch))
|
|
};
|
|
|
|
in_append.wait(); // volume 1's write is inside its .dat append.
|
|
|
|
// The store still admits a lifecycle writer and a volume-2 write.
|
|
drop(state.store.try_write().unwrap());
|
|
let vol2 = state
|
|
.store
|
|
.read()
|
|
.unwrap()
|
|
.writable_volume(VolumeId(2))
|
|
.unwrap();
|
|
let mut n2 = Needle {
|
|
id: NeedleId(9),
|
|
cookie: 0x2222.into(),
|
|
data: vec![9u8; 8],
|
|
data_size: 8,
|
|
..Needle::default()
|
|
};
|
|
vol2.write_needle(&mut n2, true, false).unwrap();
|
|
|
|
// Same-volume exclusion is intact: volume 1's data-file lease is held.
|
|
let (_, vol1) = state.store.read().unwrap().find_volume(VolumeId(1)).unwrap();
|
|
assert!(vol1.instance().try_write_lock().is_none());
|
|
|
|
release.wait();
|
|
writer.join().unwrap();
|
|
assert!(response_rx.try_recv().unwrap().is_ok());
|
|
assert!(vol1.instance().try_write_lock().is_some());
|
|
}
|
|
|
|
/// A write parked inside volume 1's .dat append must not keep later
|
|
/// batches for other volumes queued behind it: each volume drains through
|
|
/// its own FIFO worker, while same-volume writes still serialize.
|
|
/// NeedleId(102) is reserved for this test — hooks are keyed by needle id
|
|
/// process-wide and parallel tests must not share one.
|
|
#[tokio::test]
|
|
async fn test_write_queue_does_not_block_other_volumes() {
|
|
use crate::config::MinFreeSpace;
|
|
use crate::storage::types::{DiskType, NeedleId};
|
|
use crate::storage::volume::{VolumeSpec, needle_write_hook};
|
|
use std::sync::Barrier;
|
|
|
|
let tmp = tempfile::TempDir::new().unwrap();
|
|
let dir = tmp.path().to_str().unwrap();
|
|
let state = make_test_state();
|
|
{
|
|
let mut store = state.store.write().unwrap();
|
|
store
|
|
.add_location(
|
|
dir,
|
|
dir,
|
|
10,
|
|
DiskType::HardDrive,
|
|
MinFreeSpace::Percent(1.0),
|
|
Vec::new(),
|
|
)
|
|
.unwrap();
|
|
for vid in [VolumeId(1), VolumeId(2)] {
|
|
store
|
|
.add_volume(vid, DiskType::HardDrive, &VolumeSpec::default())
|
|
.unwrap();
|
|
}
|
|
}
|
|
let queue = WriteQueue::new(state, MAX_BATCH_SIZE);
|
|
|
|
let in_append = Arc::new(Barrier::new(2));
|
|
let release = Arc::new(Barrier::new(2));
|
|
let _reg = needle_write_hook::register(NeedleId(102), {
|
|
let in_append = in_append.clone();
|
|
let release = release.clone();
|
|
move |_| {
|
|
in_append.wait();
|
|
release.wait();
|
|
}
|
|
});
|
|
|
|
let needle_a = Needle {
|
|
id: NeedleId(102),
|
|
cookie: 0x1111.into(),
|
|
data: vec![7u8; 8],
|
|
data_size: 8,
|
|
..Needle::default()
|
|
};
|
|
let a1 = tokio::spawn({
|
|
let queue = queue.clone();
|
|
async move { queue.submit(VolumeId(1), needle_a, false).await }
|
|
});
|
|
// The hook parks on a blocking thread, so wait on the barrier off the
|
|
// runtime worker.
|
|
let parked = {
|
|
let in_append = in_append.clone();
|
|
tokio::task::spawn_blocking(move || in_append.wait())
|
|
};
|
|
parked.await.unwrap(); // volume 1's write is inside its .dat append.
|
|
|
|
// Volume 2's worker is independent: its submit completes now.
|
|
let needle_b = Needle {
|
|
id: NeedleId(9),
|
|
cookie: 0x2222.into(),
|
|
data: vec![9u8; 8],
|
|
data_size: 8,
|
|
..Needle::default()
|
|
};
|
|
queue.submit(VolumeId(2), needle_b, true).await.unwrap();
|
|
|
|
// Same-volume FIFO is preserved: a second volume-1 write stays queued.
|
|
let needle_a2 = Needle {
|
|
id: NeedleId(2),
|
|
cookie: 0x3333.into(),
|
|
data: vec![8u8; 8],
|
|
data_size: 8,
|
|
..Needle::default()
|
|
};
|
|
let mut a2 = tokio::spawn({
|
|
let queue = queue.clone();
|
|
async move { queue.submit(VolumeId(1), needle_a2, false).await }
|
|
});
|
|
assert!(
|
|
tokio::time::timeout(std::time::Duration::from_millis(200), &mut a2)
|
|
.await
|
|
.is_err(),
|
|
"a same-volume write must queue behind the stalled one"
|
|
);
|
|
|
|
let released = {
|
|
let release = release.clone();
|
|
tokio::task::spawn_blocking(move || release.wait())
|
|
};
|
|
released.await.unwrap();
|
|
assert!(a1.await.unwrap().is_ok());
|
|
assert!(a2.await.unwrap().is_ok());
|
|
}
|
|
}
|