use super::reply_pack::PACKET_DATA;
use super::reply_pack::padding;
use std::fs::File;
use std::io::{BufWriter, Read, Seek, SeekFrom, Write as _};
use std::os::unix::fs::FileExt as _;
use anyhow::Context as _;
use knot_types::ObjectFormat;
use super::reply_pack::clone_range;
/// the padding puts the rest of `done` on a block boundary, which a share needs.
pub fn behind(acks: &[u8], done: &File, out: &File) -> anyhow::Result<()> {
const PACKFILE: &[u8] = b"000dpackfile\n";
let len = done.metadata()?.len();
let first = BLOCK.min(len);
let mut start = vec![0; first as usize];
done.read_exact_at(&mut start, 0)
.context("read a done reply")?;
anyhow::ensure!(start.starts_with(PACKFILE), "a done reply without its pack");
let padding = padding(acks.len() as u64);
let mut head = acks.to_vec();
head.extend_from_slice(PACKFILE);
head.extend_from_slice(&padding);
head.extend_from_slice(&start[PACKFILE.len()..]);
out.write_all_at(&head, 0)
.context("write a round's reply")?;
let shift = (acks.len() + padding.len()) as u64;
if len > BLOCK {
clone_range(done, BLOCK, out, shift + BLOCK, len - BLOCK)
.context("share a done reply's bytes")?;
}
Ok(())
}
pub(crate) fn pack_entries_of(
pack: &std::path::Path,
format: ObjectFormat,
each: &mut dyn FnMut(std::sync::Arc<Vec<u8>>) -> anyhow::Result<()>,
) -> anyhow::Result<()> {
let mut file = File::open(pack).with_context(|| format!("open {}", pack.display()))?;
let entries = (file.metadata()?.len())
.checked_sub(12 + hash_len(format) as u64)
.with_context(|| format!("{} is too short for a pack", pack.display()))?;
file.seek(SeekFrom::Start(12))?;
let mut entries = file.take(entries);
loop {
let mut chunk = vec![0; 1 << 20];
let read =
(entries.read(&mut chunk)).with_context(|| format!("read {}", pack.display()))?;
if read == 0 {
return Ok(());
}
chunk.truncate(read);
each(std::sync::Arc::new(chunk))?;
}
}
pub(crate) fn objects_of(pack: &[u8]) -> u32 {
pack.get(8..12).map_or(0, |count| {
u32::from_be_bytes(count.try_into().expect("four bytes"))
})
}
pub(crate) fn objects_in(pack: &std::path::Path) -> anyhow::Result<u32> {
let mut header = [0; 12];
File::open(pack)
.and_then(|mut file| file.read_exact(&mut header))
.with_context(|| format!("read {}", pack.display()))?;
Ok(u32::from_be_bytes(header[8..].try_into()?))
}
pub(crate) fn pack_header(count: u32) -> Vec<u8> {
gix_pack::data::header::encode(gix_pack::data::Version::V2, count).to_vec()
}
pub(crate) const BLOCK: u64 = 4096;
pub(crate) struct Sink<'a> {
pub(super) out: BufWriter<Shared<'a>>,
pending: Vec<u8>,
pub(super) hasher: Option<Hasher>,
hashed: u64,
}
/// below `shared` nothing is written, since the other reply has those bytes at the same offsets.
pub(crate) struct Shared<'a> {
file: &'a File,
at: u64,
shared: u64,
head: Vec<u8>,
}
impl std::io::Write for Shared<'_> {
fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
let below = (self.shared.saturating_sub(self.at) as usize).min(bytes.len());
let kept = (BLOCK.saturating_sub(self.at) as usize).min(below);
self.head.extend_from_slice(&bytes[..kept]);
self.at += below as u64;
self.file.write_all_at(&bytes[below..], self.at)?;
self.at += (bytes.len() - below) as u64;
Ok(bytes.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
impl<'a> Sink<'a> {
pub(crate) fn new(
prefix: &[u8],
file: &'a File,
hasher: Option<Hasher>,
shared: u64,
) -> anyhow::Result<Self> {
let shared = Shared {
file,
at: 0,
shared,
head: Vec::new(),
};
let mut out = BufWriter::with_capacity(1 << 20, shared);
out.write_all(prefix).context("write a reply")?;
Ok(Self {
out,
pending: Vec::with_capacity(2 * PACKET_DATA),
hasher,
hashed: 0,
})
}
pub(crate) fn pack(&mut self, bytes: &[u8]) -> anyhow::Result<()> {
if let Some(hasher) = &mut self.hasher {
hasher.update(bytes);
}
self.hashed += bytes.len() as u64;
self.pending.extend_from_slice(bytes);
let full = self.pending.len() / PACKET_DATA * PACKET_DATA;
for chunk in self.pending[..full].chunks(PACKET_DATA) {
packet(&mut self.out, chunk).context("write a reply")?;
}
self.pending.drain(..full);
Ok(())
}
/// no mark unless the hash has caught up with every whole block written.
pub(crate) fn mark(&self) -> Option<(u64, Midstate)> {
let (blocks, state) = self.hasher.as_ref()?.state();
Some((self.hashed, state)).filter(|_| blocks == self.hashed / 64)
}
/// the caller writes this head at 0 once the shared bytes are in.
pub(crate) fn finish(mut self, checksum: &[u8]) -> anyhow::Result<Vec<u8>> {
self.hasher = None;
self.pack(checksum)?;
if !self.pending.is_empty() {
packet(&mut self.out, &self.pending).context("write a reply")?;
}
self.out.write_all(b"0000").context("write a reply")?;
let shared = self.out.into_inner().map_err(|error| error.into_error())?;
Ok(shared.head)
}
}
pub fn reflink_or_copy(source: &File, target: &File) -> std::io::Result<()> {
reflink(source, target).or_else(|_| std::io::copy(&mut &*source, &mut &*target).map(drop))
}
/// errors without reflink support, which is why callers fall back to a copy.
pub fn reflink(source: &File, target: &File) -> std::io::Result<()> {
#[cfg(target_os = "linux")]
{
use std::os::fd::AsRawFd as _;
// SAFETY: both descriptors are open for the duration of the call.
let status = unsafe { libc::ioctl(target.as_raw_fd(), libc::FICLONE, source.as_raw_fd()) };
if status == 0 {
return Ok(());
}
Err(std::io::Error::last_os_error())
}
#[cfg(not(target_os = "linux"))]
{
let _ = (source, target);
Err(std::io::ErrorKind::Unsupported.into())
}
}
pub(crate) fn hash_len(format: ObjectFormat) -> usize {
format.kind().len_in_bytes()
}
pub(crate) fn pack_entries(pack: &[u8], hash_len: usize) -> anyhow::Result<(u32, &[u8])> {
anyhow::ensure!(
pack.len() >= 12 + hash_len && pack.starts_with(b"PACK\0\0\0\x02"),
"the new objects aren't a version 2 pack"
);
let count = u32::from_be_bytes([pack[8], pack[9], pack[10], pack[11]]);
Ok((count, &pack[12..pack.len() - hash_len]))
}
pub(crate) fn packet(out: &mut impl std::io::Write, data: &[u8]) -> std::io::Result<()> {
write!(out, "{:04x}", data.len() + 5)?;
out.write_all(&[1])?;
out.write_all(data)
}
/// hashes whole 64-byte blocks, which lets the midstate be saved and resumed.
#[derive(Clone)]
pub(crate) struct Hasher {
pub(super) state: Midstate,
pub(super) blocks: u64,
carry: Vec<u8>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum Midstate {
Sha1([u32; 5]),
Sha256([u32; 8]),
}
impl Midstate {
pub(crate) fn new(format: ObjectFormat) -> Self {
if format == ObjectFormat::SHA1 {
Self::Sha1([0x67452301, 0xefcdab89, 0x98badcfe, 0x10325476, 0xc3d2e1f0])
} else {
Self::Sha256([
0x6a09e667, 0xbb67ae85, 0x3c6ef372, 0xa54ff53a, 0x510e527f, 0x9b05688c, 0x1f83d9ab,
0x5be0cd19,
])
}
}
pub(crate) fn compress(&mut self, blocks: &[[u8; 64]]) {
match self {
Self::Sha1(state) => {
// SAFETY: a `Block` is a `GenericArray<u8, U64>`, which is a
// transparent `[u8; 64]`
let blocks = unsafe {
&*(blocks as *const [[u8; 64]]
as *const [sha1::digest::core_api::Block<sha1::Sha1Core>])
};
sha1::compress(state, blocks);
}
Self::Sha256(state) => sha2::block_api::compress256(state, blocks),
}
}
pub(crate) fn words(&self) -> &[u32] {
match self {
Self::Sha1(state) => state,
Self::Sha256(state) => state,
}
}
pub(crate) fn digest(&self) -> Vec<u8> {
self.words()
.iter()
.flat_map(|word| word.to_be_bytes())
.collect()
}
pub(crate) fn to_hex(self) -> String {
self.words()
.iter()
.map(|word| format!("{word:08x}"))
.collect()
}
pub(crate) fn from_hex(hex: &str) -> Option<Self> {
let words: Vec<u32> = (0..hex.len())
.step_by(8)
.map(|at| u32::from_str_radix(hex.get(at..at + 8)?, 16).ok())
.collect::<Option<_>>()?;
match words.len() {
5 => Some(Self::Sha1(words.try_into().ok()?)),
8 => Some(Self::Sha256(words.try_into().ok()?)),
_ => None,
}
}
}
impl Hasher {
pub(crate) fn new(format: ObjectFormat) -> Self {
Self::resume(Midstate::new(format), 0)
}
pub(crate) fn resume(state: Midstate, blocks: u64) -> Self {
Self {
state,
blocks,
carry: Vec::with_capacity(64),
}
}
pub(crate) fn update(&mut self, mut data: &[u8]) {
if !self.carry.is_empty() {
let taken = (64 - self.carry.len()).min(data.len());
self.carry.extend_from_slice(&data[..taken]);
data = &data[taken..];
if self.carry.len() < 64 {
return;
}
let block: [u8; 64] = self.carry[..].try_into().expect("a whole block");
self.state.compress(&[block]);
self.blocks += 1;
self.carry.clear();
}
let (blocks, rest) = data.as_chunks::<64>();
self.state.compress(blocks);
self.blocks += blocks.len() as u64;
self.carry.extend_from_slice(rest);
}
pub(crate) fn state(&self) -> (u64, Midstate) {
(self.blocks, self.state)
}
pub(crate) fn finish(mut self) -> Vec<u8> {
let bits = (self.blocks * 64 + self.carry.len() as u64) * 8;
let mut rest = std::mem::take(&mut self.carry);
rest.push(0x80);
while rest.len() % 64 != 56 {
rest.push(0);
}
rest.extend_from_slice(&bits.to_be_bytes());
self.state.compress(rest.as_chunks::<64>().0);
self.state.digest()
}
}