Github-Mirror/seaweedfs
1
0 Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-11-02 10:17:10 +01:00
Code Issues Packages Projects Releases Wiki Activity

volume server: ec.decode reads the .ecx from the index dir it was copied to (#11536)

Browse Source 
* volume server: ec.decode reads the .ecx from the index dir it was copied to

VolumeEcShardsCopy writes the .ecx/.ecj into the receiver's -dir.idx, so
with a split data/index dir the decode target has no .ecx beside its
shards. VolumeEcShardsToVolume sized the .dat from the right .ecx but
built the .idx from the data dir, failing with NotFound after the .dat
was already published. It now reads .ecx/.ecj from where the EC volume
opened them and writes the .idx beside the .dat, where Go leaves it.

The live-entry check and the .dat size also ignored deletions recorded
only in the .ecj, which Go folds into the .ecx (RebuildEcxFile) first:
a fully deleted volume was decoded instead of reported as having no live
entries, and deleted tail needles were copied into the .dat. Both now
treat journaled ids as deleted, without rewriting the sealed .ecx.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

* volume server: ec.decode keeps the decoded volume writable and reads every .ecj

The rebuilt .idx copied a journaled tail needle's .ecx row verbatim after
the .dat was cut short before it, so the mount saw a row past EOF and
marked the decoded volume read-only. Rows of deleted needles the .dat no
longer holds are now dropped, and each journaled needle still in the .dat
gets one tombstone instead of one per journal entry.

VolumeEcShardsCopy appends journals collected from other holders into
the idx dir, but the decode read only the .ecj beside the .ecx, which
sits in the data dir when this server generated the shards. It now
reads both, once, in bounded chunks via the loader EcVolume uses.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

* volume server: test ec.decode drops a sealed .ecx tail tombstone

Covers the other half of the rule added in the previous commit: a tail
needle tombstoned in the .ecx itself (Go's RebuildEcxFile) is cut from
the .dat, and its row must not reach the rebuilt .idx either.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Eliah RusinandClaude Opus 5.5 authored and GitHub committed 2026-10-01 20:19:17 +08:00
1 parent b8f074b7d3
commit 62481f1673
3 files changed
+447 -91
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

seaweed-volume/src/server/grpc_server.rs
+294 -26
View File
@@ -3876,36 +3876,13 @@ impl VolumeServer for VolumeGrpcService {
}
}
// Read the .ecx index to check for live entries
let ecx_path = ec_vol.ecx_file_name();
let ecx_data =
std::fs::read(&ecx_path).map_err(|e| Status::internal(format!("read ecx: {}", e)))?;
let entry_count = ecx_data.len() / NEEDLE_MAP_ENTRY_SIZE;
let mut has_live = false;
for i in 0..entry_count {
let start = i * NEEDLE_MAP_ENTRY_SIZE;
let (_, _, size) =
idx_entry_from_bytes(&ecx_data[start..start + NEEDLE_MAP_ENTRY_SIZE]);
if !size.is_deleted() {
has_live = true;
break;
}
}
if !has_live {
return Err(Status::failed_precondition(format!(
"ec volume {} has no live entries",
req.volume_id
)));
}
// Reconstruct the volume from EC shards. Use the EcVolume's
// own dir for the produced .dat (matches the volume's home
// disk) and its `ecx_actual_dir` for the .ecx lookup, while
// reading each shard from its real on-disk location.
let dat_dir = ec_vol.dir.clone();
let ecx_dir = ec_vol.ecx_actual_dir().to_string();
let idx_dir = ec_vol.dir_idx.clone();
let collection = ec_vol.collection.clone();
let vif_dat_file_size = ec_vol.dat_file_size;
let (large_block_size, small_block_size) =
@@ -3918,6 +3895,28 @@ impl VolumeServer for VolumeGrpcService {
.collect();
drop(store);
// Deletions journaled beside the .ecx, or collected by
// VolumeEcShardsCopy into the idx dir, count as deleted throughout.
let deleted = crate::storage::erasure_coding::ec_decoder::read_ecj_deletions(
&[&ecx_dir, &idx_dir],
&collection,
vid,
)
.map_err(|e| Status::internal(format!("read ecj: {}", e)))?;
let has_live = crate::storage::erasure_coding::ec_decoder::has_live_needles(
&ecx_dir,
&collection,
vid,
&deleted,
)
.map_err(|e| Status::internal(format!("HasLiveNeedles: {}", e)))?;
if !has_live {
return Err(Status::failed_precondition(format!(
"ec volume {} has no live entries",
req.volume_id
)));
}
// Calculate .dat file size from .ecx entries (.ec00 lives on
// its own disk, .ecx on the index disk).
let dat_file_size =
@@ -3926,6 +3925,7 @@ impl VolumeServer for VolumeGrpcService {
&ecx_dir,
&collection,
vid,
&deleted,
)
.map_err(|e| Status::internal(format!("FindDatFileSize: {}", e)))?;
@@ -3950,11 +3950,15 @@ impl VolumeServer for VolumeGrpcService {
)
.map_err(|e| Status::internal(format!("WriteDatFile: {}", e)))?;
// Write .idx file from .ecx and .ecj files (lives on idx dir).
crate::storage::erasure_coding::ec_decoder::write_idx_file_from_ec_index(
// Write .idx from the .ecx wherever it lives, beside the .dat where
// the mount looks first (Go moves it there after the rebuild).
crate::storage::erasure_coding::ec_decoder::write_idx_file_from_ec_index_with_dirs(
&ecx_dir,
&dat_dir,
&collection,
vid,
&deleted,
dat_file_size,
)
.map_err(|e| Status::internal(format!("WriteIdxFileFromEcIndex: {}", e)))?;
@@ -11082,4 +11086,268 @@ mod tests {
resp.error
);
}
/// A split -dir/-dir.idx decode target with the shards in the data dir.
/// The .ecx is moved to the idx dir, as VolumeEcShardsCopy leaves it, unless
/// `ecx_in_data_dir` (where VolumeEcShardsGenerate writes it). `journal` goes
/// into the idx-dir .ecj, where VolumeEcShardsCopy appends collected ones.
async fn make_split_idx_ec_decode_service(
journal: &[u64],
ecx_in_data_dir: bool,
) -> (VolumeGrpcService, TempDir, String, String, u64) {
let tmp = TempDir::new().unwrap();
let data = tmp.path().join("data").to_str().unwrap().to_string();
let idx = tmp.path().join("idx").to_str().unwrap().to_string();
std::fs::create_dir_all(&data).unwrap();
std::fs::create_dir_all(&idx).unwrap();
let mut store = Store::new(NeedleMapKind::InMemory);
store
.add_location(
&data,
&idx,
10,
DiskType::HardDrive,
MinFreeSpace::Percent(0.0),
Vec::new(),
)
.unwrap();
store
.add_volume(VolumeId(1), DiskType::HardDrive, &VolumeSpec::default())
.unwrap();
let mut size_before_needle_3 = 0;
{
let (_, volume) = store.find_volume_mut(VolumeId(1)).unwrap();
for id in 1..=3u64 {
if id == 3 {
size_before_needle_3 = volume.dat_file_size().unwrap();
}
let body = format!("split idx decode needle {id}").into_bytes();
let mut needle = Needle {
id: NeedleId(id),
cookie: Cookie(id as u32),
data_size: body.len() as u32,
data: body,
..Needle::default()
};
volume.write_needle(&mut needle, true, false).unwrap();
}
volume.sync_to_disk().unwrap();
}
let service = split_disk_grpc_service(store);
service
.volume_ec_shards_generate(Request::new(
volume_server_pb::VolumeEcShardsGenerateRequest {
volume_id: 1,
collection: String::new(),
},
))
.await
.unwrap();
let mut store = service.state.store.write().unwrap();
store.unmount_volume(VolumeId(1)).unwrap();
std::fs::remove_file(format!("{data}/1.dat")).unwrap();
std::fs::remove_file(format!("{idx}/1.idx")).unwrap();
let ecx_dir = if ecx_in_data_dir {
data.clone()
} else {
std::fs::rename(format!("{data}/1.ecx"), format!("{idx}/1.ecx")).unwrap();
idx.clone()
};
let mut ecj = Vec::new();
for id in journal {
let mut buf = [0u8; NEEDLE_ID_SIZE];
NeedleId(*id).to_bytes(&mut buf);
ecj.extend_from_slice(&buf);
}
std::fs::write(format!("{idx}/1.ecj"), ecj).unwrap();
for shard_id in 0..14 {
store.mount_ec_shard(VolumeId(1), "", shard_id, "").unwrap();
}
assert_eq!(
store.find_ec_volume(VolumeId(1)).unwrap().ecx_actual_dir(),
ecx_dir,
"precondition: the mounted EC volume reads its .ecx from {ecx_dir}"
);
drop(store);
(service, tmp, data, idx, size_before_needle_3)
}
fn ec_shards_to_volume_request() -> Request<volume_server_pb::VolumeEcShardsToVolumeRequest> {
let mut request = Request::new(volume_server_pb::VolumeEcShardsToVolumeRequest {
volume_id: 1,
collection: String::new(),
..Default::default()
});
request
.extensions_mut()
.insert(tonic::transport::server::TcpConnectInfo {
local_addr: None,
remote_addr: Some("127.0.0.1:65000".parse().unwrap()),
});
request
}
/// Mounts the decoded volume and checks its (file, deleted) counts, that it
/// is writable, that `live` needles read back and `deleted` ones stay deleted.
fn assert_decoded_volume(
service: &VolumeGrpcService,
counts: (i64, i64),
live: &[u64],
deleted: &[u64],
) {
let mut store = service.state.store.write().unwrap();
let all_shards: Vec<ShardId> = (0..14).collect();
store.unmount_ec_shards(VolumeId(1), &all_shards);
store
.mount_volume(VolumeId(1), "", DiskType::HardDrive)
.unwrap();
let (_, v) = store.find_volume(VolumeId(1)).unwrap();
assert_eq!((v.file_count(), v.deleted_count()), counts);
assert!(
!v.is_no_write_or_delete(),
"no .idx row may point past the end of the decoded .dat"
);
for &id in live {
let mut n = Needle {
id: NeedleId(id),
..Needle::default()
};
store.read_volume_needle(VolumeId(1), &mut n).unwrap();
assert_eq!(n.data, format!("split idx decode needle {id}").into_bytes());
}
for &id in deleted {
let mut n = Needle {
id: NeedleId(id),
..Needle::default()
};
assert!(
store.read_volume_needle(VolumeId(1), &mut n).is_err(),
"needle {id} was deleted through the .ecj and must stay deleted"
);
}
let body = b"written after decode".to_vec();
let mut n = Needle {
id: NeedleId(4),
cookie: Cookie(4),
data_size: body.len() as u32,
data: body.clone(),
..Needle::default()
};
store
.write_volume_needle(VolumeId(1), &mut n, true)
.unwrap();
let mut n = Needle {
id: NeedleId(4),
..Needle::default()
};
store.read_volume_needle(VolumeId(1), &mut n).unwrap();
assert_eq!(n.data, body);
}
/// ec.decode used to read the .ecx/.ecj from the data dir when writing the
/// .idx, failing with NotFound after the .dat was already published, and
/// ignored .ecj deletions when sizing the .dat (Go folds them in first).
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_volume_ec_shards_to_volume_reads_ecx_from_split_idx_dir() {
// The lowest key (the first .idx row) and the .dat tail, each journaled
// many times over.
let journal: Vec<u64> = std::iter::repeat_n([1, 3], 500).flatten().collect();
let (service, _tmp, data, idx, size_before_needle_3) =
make_split_idx_ec_decode_service(&journal, false).await;
let ecx = std::fs::read(format!("{idx}/1.ecx")).unwrap();
service
.volume_ec_shards_to_volume(ec_shards_to_volume_request())
.await
.unwrap();
// The journaled tail needle is not decoded.
assert_eq!(
std::fs::metadata(format!("{data}/1.dat")).unwrap().len(),
size_before_needle_3
);
// The .idx sits beside the .dat. It keeps the .ecx rows the .dat holds
// and tombstones each journaled one once, however often it was journaled.
let mut expected_idx = ecx[..2 * NEEDLE_MAP_ENTRY_SIZE].to_vec();
crate::storage::idx::write_index_entry(
&mut expected_idx,
NeedleId(1),
Offset::default(),
TOMBSTONE_FILE_SIZE,
)
.unwrap();
assert_eq!(
std::fs::read(format!("{data}/1.idx")).unwrap(),
expected_idx
);
assert!(!std::path::Path::new(&format!("{idx}/1.idx")).exists());
// Needle 1 is still in the .dat, so it counts as garbage; needle 3 is not.
assert_decoded_volume(&service, (2, 1), &[2], &[1, 3]);
}
/// Collected journals land in the idx dir even when the .ecx is beside the
/// shards, so the decode reads deletions from both.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_volume_ec_shards_to_volume_reads_idx_dir_ecj_with_data_dir_ecx() {
let (service, _tmp, data, _idx, size_before_needle_3) =
make_split_idx_ec_decode_service(&[3], true).await;
service
.volume_ec_shards_to_volume(ec_shards_to_volume_request())
.await
.unwrap();
assert_eq!(
std::fs::metadata(format!("{data}/1.dat")).unwrap().len(),
size_before_needle_3
);
assert_decoded_volume(&service, (2, 0), &[1, 2], &[3]);
}
/// A tail needle tombstoned in the .ecx itself (Go's RebuildEcxFile) is cut
/// from the .dat the same way, so its row must not survive either.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_volume_ec_shards_to_volume_drops_sealed_tail_tombstone() {
let (service, _tmp, data, idx, size_before_needle_3) =
make_split_idx_ec_decode_service(&[], false).await;
let ecx_path = format!("{idx}/1.ecx");
let mut ecx = std::fs::read(&ecx_path).unwrap();
let size_at = 2 * NEEDLE_MAP_ENTRY_SIZE + NEEDLE_ID_SIZE + OFFSET_SIZE;
TOMBSTONE_FILE_SIZE.to_bytes(&mut ecx[size_at..size_at + SIZE_SIZE]);
std::fs::write(&ecx_path, &ecx).unwrap();
service
.volume_ec_shards_to_volume(ec_shards_to_volume_request())
.await
.unwrap();
assert_eq!(
std::fs::metadata(format!("{data}/1.dat")).unwrap().len(),
size_before_needle_3
);
assert_eq!(
std::fs::read(format!("{data}/1.idx")).unwrap(),
ecx[..2 * NEEDLE_MAP_ENTRY_SIZE]
);
assert_decoded_volume(&service, (2, 0), &[1, 2], &[3]);
}
/// Deletions only in the .ecj count toward "no live entries", as after Go's
/// RebuildEcxFile, so the caller purges the shards instead of decoding.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_volume_ec_shards_to_volume_all_journaled_has_no_live_entries() {
let (service, _tmp, data, _idx, _) =
make_split_idx_ec_decode_service(&[1, 2, 3], false).await;
let err = service
.volume_ec_shards_to_volume(ec_shards_to_volume_request())
.await
.unwrap_err();
assert_eq!(err.code(), tonic::Code::FailedPrecondition, "{err:?}");
assert!(err.message().contains("has no live entries"), "{err:?}");
assert!(!std::path::Path::new(&format!("{data}/1.dat")).exists());
}
}
seaweed-volume/src/storage/erasure_coding/ec_decoder.rs
+129 -33
View File
@@ -3,10 +3,12 @@
//! Rebuilds the original .dat + .idx files from data shards (.ec00-.ec09)
//! and the sorted index (.ecx) + deletion journal (.ecj).
use std::collections::HashSet;
use std::fs::File;
use std::io::{self, Read, Write};
use crate::storage::erasure_coding::ec_shard::*;
use crate::storage::erasure_coding::ec_volume::read_ecj_ids;
use crate::storage::idx;
use crate::storage::needle::needle::get_actual_size;
use crate::storage::super_block::SUPER_BLOCK_SIZE;
@@ -20,18 +22,21 @@ use crate::storage::volume::{fsync_dir, volume_file_name};
/// `dir` is used both for reading `.ec00` and `.ecx`. For split-disk
/// reconciled volumes call [`find_dat_file_size_with_dirs`] instead.
pub fn find_dat_file_size(dir: &str, collection: &str, volume_id: VolumeId) -> io::Result<i64> {
find_dat_file_size_with_dirs(dir, dir, collection, volume_id)
let deleted = read_ecj_deletions(&[dir], collection, volume_id)?;
find_dat_file_size_with_dirs(dir, dir, collection, volume_id, &deleted)
}
/// Like [`find_dat_file_size`] but lets the caller pass separate dirs
/// for `.ec00` (the data shard) and `.ecx` (the sealed index). This
/// is the form needed when shards are split across data dirs and the
/// `.ecx` lives on a sibling disk's idx dir (#9252).
/// `.ecx` lives on a sibling disk's idx dir (#9252). Needles in `deleted`
/// count as deleted.
pub fn find_dat_file_size_with_dirs(
ec00_dir: &str,
ecx_dir: &str,
collection: &str,
volume_id: VolumeId,
deleted: &HashSet<NeedleId>,
) -> io::Result<i64> {
let ec00_base = volume_file_name(ec00_dir, collection, volume_id);
let ecx_base = volume_file_name(ecx_dir, collection, volume_id);
@@ -53,9 +58,9 @@ pub fn find_dat_file_size_with_dirs(
for i in 0..entry_count {
let start = i * NEEDLE_MAP_ENTRY_SIZE;
let (_, offset, size) =
let (key, offset, size) =
idx_entry_from_bytes(&ecx_data[start..start + NEEDLE_MAP_ENTRY_SIZE]);
if size.is_deleted() {
if size.is_deleted() || deleted.contains(&key) {
continue;
}
let entry_stop = offset.to_actual_offset() + get_actual_size(size, version);
@@ -67,6 +72,48 @@ pub fn find_dat_file_size_with_dirs(
Ok(dat_size)
}
/// Whether the `.ecx` in `ecx_dir` indexes a needle deleted neither there nor
/// in `deleted`.
pub fn has_live_needles(
ecx_dir: &str,
collection: &str,
volume_id: VolumeId,
deleted: &HashSet<NeedleId>,
) -> io::Result<bool> {
let ecx_base = volume_file_name(ecx_dir, collection, volume_id);
let ecx_data = std::fs::read(format!("{}.ecx", ecx_base))?;
let (entries, _) = ecx_data.as_chunks::<NEEDLE_MAP_ENTRY_SIZE>();
Ok(entries.iter().any(|entry| {
let (key, _, size) = idx_entry_from_bytes(entry);
!size.is_deleted() && !deleted.contains(&key)
}))
}
/// Distinct needle ids journaled in the `.ecj` of any of `dirs`. Go folds the
/// journal into the `.ecx` (RebuildEcxFile) before a decode; reading it leaves
/// the sealed index untouched. Only NotFound means "no journal".
pub fn read_ecj_deletions(
dirs: &[&str],
collection: &str,
volume_id: VolumeId,
) -> io::Result<HashSet<NeedleId>> {
let mut ids = HashSet::new();
for (i, dir) in dirs.iter().enumerate() {
if dirs[..i].contains(dir) {
continue;
}
let path = format!("{}.ecj", volume_file_name(dir, collection, volume_id));
let file = match File::open(&path) {
Ok(file) => file,
Err(e) if e.kind() == io::ErrorKind::NotFound => continue,
Err(e) => return Err(e),
};
let len = file.metadata()?.len();
read_ecj_ids(&file, len, &mut ids)?;
}
Ok(ids)
}
/// What it takes to rebuild a volume's .dat from its EC data shards.
///
/// Mirrors Go's `WriteDatFile(baseFileName, datFileSize,
@@ -267,51 +314,61 @@ pub fn write_dat_file_from_shards(spec: &DatRebuild<'_>) -> io::Result<()> {
/// Write .idx file from .ecx index + .ecj deletion journal.
///
/// Copies sorted .ecx entries to .idx, then appends tombstones for
/// deleted needles from .ecj.
/// See [`write_idx_file_from_ec_index_with_dirs`]; everything lives in `dir`.
pub fn write_idx_file_from_ec_index(
dir: &str,
collection: &str,
volume_id: VolumeId,
) -> io::Result<()> {
let base = volume_file_name(dir, collection, volume_id);
let ecx_path = format!("{}.ecx", base);
let ecj_path = format!("{}.ecj", base);
let idx_path = format!("{}.idx", base);
let deleted = read_ecj_deletions(&[dir], collection, volume_id)?;
let dat_file_size = find_dat_file_size_with_dirs(dir, dir, collection, volume_id, &deleted)?;
write_idx_file_from_ec_index_with_dirs(dir, dir, collection, volume_id, &deleted, dat_file_size)
}
/// Write the `.idx` for a `.dat` decoded to `dat_file_size` bytes, from the
/// `.ecx` in `ecx_dir`, into `idx_dir`.
///
/// Copies the `.ecx` rows, then appends one tombstone per row whose needle is
/// in `deleted`. A deleted needle at or past `dat_file_size` was cut from the
/// `.dat`, so its row is dropped: a row pointing past the end of the `.dat`
/// makes the volume load read-only.
pub fn write_idx_file_from_ec_index_with_dirs(
ecx_dir: &str,
idx_dir: &str,
collection: &str,
volume_id: VolumeId,
deleted: &HashSet<NeedleId>,
dat_file_size: i64,
) -> io::Result<()> {
let ecx_path = format!("{}.ecx", volume_file_name(ecx_dir, collection, volume_id));
let idx_path = format!("{}.idx", volume_file_name(idx_dir, collection, volume_id));
// Write to a temp file and atomically rename into place, so a crash
// mid-write never leaves a partial .idx at the final name beside the
// source shards.
let tmp_path = format!("{}.tmp", idx_path);
let write_result = (|| -> io::Result<()> {
// Copy .ecx to the temp .idx
std::fs::copy(&ecx_path, &tmp_path)?;
// Append deletions from .ecj as tombstones. Read the journal directly
// and treat only NotFound as "no journal": Path::exists would also
// swallow a permission/IO error and silently skip deletions, which
// would resurrect deleted needles as live.
let mut idx_file = std::fs::OpenOptions::new().append(true).open(&tmp_path)?;
match std::fs::read(&ecj_path) {
Ok(ecj_data) => {
let count = ecj_data.len() / NEEDLE_ID_SIZE;
for i in 0..count {
let start = i * NEEDLE_ID_SIZE;
let needle_id = NeedleId::from_bytes(&ecj_data[start..start + NEEDLE_ID_SIZE]);
idx::write_index_entry(
&mut idx_file,
needle_id,
Offset::default(),
TOMBSTONE_FILE_SIZE,
)?;
}
let mut ecx_file = File::open(&ecx_path)?;
let mut idx_file = io::BufWriter::new(File::create(&tmp_path)?);
let mut tombstoned = Vec::new();
idx::walk_index_file(&mut ecx_file, 0, |key, offset, size| {
let is_deleted = size.is_deleted() || deleted.contains(&key);
if is_deleted && offset.to_actual_offset() >= dat_file_size {
return Ok(());
}
Err(e) if e.kind() == io::ErrorKind::NotFound => {}
Err(e) => return Err(e),
idx::write_index_entry(&mut idx_file, key, offset, size)?;
if !size.is_deleted() && deleted.contains(&key) {
tombstoned.push(key);
}
Ok(())
})?;
for key in tombstoned {
idx::write_index_entry(&mut idx_file, key, Offset::default(), TOMBSTONE_FILE_SIZE)?;
}
// fsync, rename, then fsync the dir so the decoded .idx is durable and
// atomically published before the caller deletes the source shards.
let idx_file = idx_file.into_inner().map_err(|e| e.into_error())?;
idx_file.sync_all()?;
drop(idx_file);
// Windows rename does not replace an existing file on every version;
@@ -694,4 +751,43 @@ mod tests {
.is_err()
);
}
/// A journal many chunks long that repeats a few ids reads back as those
/// ids, from each dir once, whatever its length.
#[test]
fn test_read_ecj_deletions_collects_distinct_ids_across_dirs() {
let tmp = TempDir::new().unwrap();
let data = tmp.path().join("data");
let idx = tmp.path().join("idx");
let missing = tmp.path().join("missing");
std::fs::create_dir_all(&data).unwrap();
std::fs::create_dir_all(&idx).unwrap();
let (data, idx, missing) = (
data.to_str().unwrap(),
idx.to_str().unwrap(),
missing.to_str().unwrap(),
);
let entry = |id: u64| {
let mut buf = [0u8; NEEDLE_ID_SIZE];
NeedleId(id).to_bytes(&mut buf);
buf
};
// Past two load chunks of three repeating ids, then an id only in the
// last chunk and a torn trailing record.
let mut ecj = Vec::new();
while ecj.len() <= 2 * (1 << 20) {
for id in [1, 2, 3] {
ecj.extend_from_slice(&entry(id));
}
}
ecj.extend_from_slice(&entry(7));
ecj.extend_from_slice(&entry(8)[..3]);
std::fs::write(format!("{idx}/1.ecj"), &ecj).unwrap();
std::fs::write(format!("{data}/1.ecj"), entry(9)).unwrap();
let ids = read_ecj_deletions(&[data, idx, idx, missing], "", VolumeId(1)).unwrap();
let expected: HashSet<NeedleId> = [1, 2, 3, 7, 9].into_iter().map(NeedleId).collect();
assert_eq!(ids, expected);
}
}
seaweed-volume/src/storage/erasure_coding/ec_volume.rs
+24 -32
View File
@@ -48,6 +48,29 @@ pub(crate) struct ShardLocationCache {
/// A multiple of `NEEDLE_ID_SIZE`; 1 MiB is 131072 entries per syscall.
const ECJ_LOAD_CHUNK_BYTES: usize = 1 << 20;
/// Adds every whole needle id in the first `len` bytes of `ecj_file` to `ids`,
/// reading `ECJ_LOAD_CHUNK_BYTES` at a time; a trailing partial record is
/// ignored.
pub(crate) fn read_ecj_ids(
ecj_file: &File,
len: u64,
ids: &mut HashSet<NeedleId>,
) -> io::Result<()> {
let mut buf = vec![0u8; std::cmp::min(ECJ_LOAD_CHUNK_BYTES as u64, len) as usize];
let mut off: u64 = 0;
while off + NEEDLE_ID_SIZE as u64 <= len {
let mut want = std::cmp::min(ECJ_LOAD_CHUNK_BYTES as u64, len - off) as usize;
want -= want % NEEDLE_ID_SIZE;
// Positional read: the loader's handle is shared with journal appends.
read_exact_at(ecj_file, &mut buf[..want], off)?;
for entry in buf[..want].as_chunks::<NEEDLE_ID_SIZE>().0 {
ids.insert(NeedleId::from_bytes(entry));
}
off += want as u64;
}
Ok(())
}
/// An erasure-coded volume managing its local shards and index.
pub struct EcVolume {
pub volume_id: VolumeId,
@@ -810,38 +833,7 @@ impl EcVolume {
// held the `deleted_needles` write lock for the whole scan, which on a
// bloated journal is the entire (unbounded) startup.
let mut loaded: HashSet<NeedleId> = HashSet::new();
let mut buf = vec![0u8; ECJ_LOAD_CHUNK_BYTES];
let end = self.ecj_file_size as u64;
let mut off: u64 = 0;
while off + NEEDLE_ID_SIZE as u64 <= end {
// Whole entries only; a trailing partial record is ignored, as the
// per-entry loop did by construction.
let mut want = std::cmp::min(ECJ_LOAD_CHUNK_BYTES as u64, end - off) as usize;
want -= want % NEEDLE_ID_SIZE;
if want == 0 {
break;
}
#[cfg(unix)]
{
use std::os::unix::fs::FileExt;
ecj_file.read_exact_at(&mut buf[..want], off)?;
}
#[cfg(windows)]
{
// Positional read so concurrent readers of the shared .ecj
// handle can't interleave seek/read. Mirrors the
// read_exact_at helper at the bottom of this file.
read_exact_at(ecj_file, &mut buf[..want], off)?;
}
#[cfg(not(any(unix, windows)))]
{
compile_error!("Platform not supported: only unix and windows are supported");
}
for entry in buf[..want].chunks_exact(NEEDLE_ID_SIZE) {
loaded.insert(NeedleId::from_bytes(entry));
}
off += want as u64;
}
read_ecj_ids(ecj_file, self.ecj_file_size as u64, &mut loaded)?;
let mut set = self
.deleted_needles