fix data race on seal vs batch

This commit is contained in:
Peter Zhang 2025-11-24 08:48:17 +08:00
parent 1117cd65e9
commit ef04124d09
2 changed files with 24 additions and 19 deletions

View File

@ -270,17 +270,17 @@ impl FlowWrite for FlowStore {
.get_entry_batch(chunk_index)? .get_entry_batch(chunk_index)?
.unwrap_or_else(|| EntryBatch::new(chunk_index)); .unwrap_or_else(|| EntryBatch::new(chunk_index));
let completed_seals = batch.insert_data( let _ = batch.insert_data(
(chunk.start_index % self.config.batch_size as u64) as usize, (chunk.start_index % self.config.batch_size as u64) as usize,
chunk.data, chunk.data,
)?; )?;
if self.seal_manager.seal_worker_available() { if self.seal_manager.seal_worker_available() && batch.is_complete() {
completed_seals.into_iter().for_each(|x| { for seal_index in 0..SEALS_PER_LOAD {
to_seal_set.insert( to_seal_set.insert(
chunk_index as usize * SEALS_PER_LOAD + x as usize, chunk_index as usize * SEALS_PER_LOAD + seal_index,
self.seal_manager.to_seal_version(), self.seal_manager.to_seal_version(),
); );
}); }
} }
batch_list.push((chunk_index, batch)); batch_list.push((chunk_index, batch));

View File

@ -38,6 +38,12 @@ impl UnsealedDataList {
} }
} }
impl Default for UnsealedDataList {
fn default() -> Self {
Self::new()
}
}
use super::SealAnswer; use super::SealAnswer;
pub use chunk_data::EntryBatchData; pub use chunk_data::EntryBatchData;
use seal::SealInfo; use seal::SealInfo;
@ -60,6 +66,11 @@ impl EntryBatch {
pub fn is_empty(&self) -> bool { pub fn is_empty(&self) -> bool {
self.data.is_empty() self.data.is_empty()
} }
/// Check if the batch data is complete (all data has been filled)
pub fn is_complete(&self) -> bool {
matches!(self.data, EntryBatchData::Complete(_))
}
} }
impl EntryBatch { impl EntryBatch {
@ -192,9 +203,9 @@ impl EntryBatch {
// If complete, return the entire unsealed data as a single chunk // If complete, return the entire unsealed data as a single chunk
let mut result = UnsealedDataList::new(); let mut result = UnsealedDataList::new();
let mut complete_data = Vec::with_capacity(BYTES_PER_LOAD); let mut complete_data = Vec::with_capacity(BYTES_PER_LOAD);
for bit in 0..SEALS_PER_LOAD { for bit in 0..SEALS_PER_LOAD {
let start_byte = bit as usize * BYTES_PER_SEAL; let start_byte = bit * BYTES_PER_SEAL;
if self.seal.is_sealed(bit as u16) { if self.seal.is_sealed(bit as u16) {
// If sealed, get the slice, unseal it, and append // If sealed, get the slice, unseal it, and append
let mut data_slice = self.data.get(start_byte, BYTES_PER_SEAL)?.to_vec(); let mut data_slice = self.data.get(start_byte, BYTES_PER_SEAL)?.to_vec();
@ -206,7 +217,7 @@ impl EntryBatch {
complete_data.extend_from_slice(data_slice); complete_data.extend_from_slice(data_slice);
} }
} }
result.add_chunk(complete_data); result.add_chunk(complete_data);
Some(result) Some(result)
} }
@ -215,14 +226,12 @@ impl EntryBatch {
debug!("Building unsealed data from incomplete known data"); debug!("Building unsealed data from incomplete known data");
let mut result = UnsealedDataList::new(); let mut result = UnsealedDataList::new();
for partial_batch in &incomplete_data.known_data { for partial_batch in &incomplete_data.known_data {
let start_sector = partial_batch.start_sector; let start_sector = partial_batch.start_sector;
let data_len = partial_batch.data.len(); let data_len = partial_batch.data.len();
debug!( debug!(
"Processing partial batch: start_sector={} data_len={}", "Processing partial batch: start_sector={} data_len={}",
start_sector, start_sector, data_len
data_len
); );
if data_len == 0 { if data_len == 0 {
@ -241,14 +250,12 @@ impl EntryBatch {
debug!( debug!(
"Partial batch spans seals: start_seal_index={} end_seal_index={}", "Partial batch spans seals: start_seal_index={} end_seal_index={}",
start_seal_index, start_seal_index, end_seal_index
end_seal_index
); );
debug!( debug!(
"Partial batch byte range: {} to {}", "Partial batch byte range: {} to {}",
partial_start_byte, partial_start_byte, partial_end_byte
partial_end_byte
); );
// Iterate through each seal that this partial batch spans // Iterate through each seal that this partial batch spans
@ -262,9 +269,7 @@ impl EntryBatch {
debug!( debug!(
"Processing seal_start_byte={} seal_end_byte={} is_full_seal={}", "Processing seal_start_byte={} seal_end_byte={} is_full_seal={}",
seal_start_byte, seal_start_byte, seal_end_byte, is_full_seal
seal_end_byte,
is_full_seal
); );
if is_full_seal && self.seal.is_sealed(seal_index) { if is_full_seal && self.seal.is_sealed(seal_index) {
@ -285,7 +290,7 @@ impl EntryBatch {
partial_batch_data.extend_from_slice(overlap_data); partial_batch_data.extend_from_slice(overlap_data);
} }
} }
// Add the complete unsealed data for this partial batch as one chunk // Add the complete unsealed data for this partial batch as one chunk
result.add_chunk(partial_batch_data); result.add_chunk(partial_batch_data);
} }