Add primitive pipelined integrity stage
This commit is contained in:
parent
1c1042e640
commit
0a02eea478
|
@ -120,6 +120,19 @@ impl<RP: RawPile> RawPile for RawPileIntegrityChecker<RP> {
|
||||||
settings: StoragePipelineSettings,
|
settings: StoragePipelineSettings,
|
||||||
controller_send: Sender<ControllerMessage>,
|
controller_send: Sender<ControllerMessage>,
|
||||||
) -> anyhow::Result<Sender<(ChunkId, Vec<u8>)>> {
|
) -> anyhow::Result<Sender<(ChunkId, Vec<u8>)>> {
|
||||||
todo!()
|
// TODO primitive implementation but good enough for now.
|
||||||
|
// May want metrics later?
|
||||||
|
let next_stage = self.underlying.build_storage_pipeline(settings, controller_send)?;
|
||||||
|
let (input, receiver) = crossbeam_channel::bounded::<(ChunkId, Vec<u8>)>(64);
|
||||||
|
std::thread::spawn(move || {
|
||||||
|
while let Ok((chunk_id, mut chunk)) = receiver.recv() {
|
||||||
|
let mut hasher = twox_hash::XxHash64::with_seed(XXH64_SEED);
|
||||||
|
hasher.write(&chunk);
|
||||||
|
let computed_hash = hasher.finish().to_be_bytes();
|
||||||
|
chunk.extend_from_slice(&computed_hash);
|
||||||
|
next_stage.send((chunk_id, chunk)).unwrap();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
Ok(input)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in New Issue