alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::io::{self, BufWriter, Write};
2use std::os::unix::fs::FileExt;
3use std::sync::Mutex;
4
5use gix::ObjectId;
6
7use crate::error::PackError;
8use crate::ids::{Crc32, PackOffset};
9
10const V2_SIGNATURE: &[u8] = &[0xff, 0x74, 0x4f, 0x63];
11const V2_VERSION: u32 = 2;
12const HIGH_BIT: u32 = 0x8000_0000;
13const LARGE_OFFSET_THRESHOLD: u64 = 0x7fff_ffff;
14const BUCKETS: usize = 256;
15const BUCKET_BUF: usize = 64 * 1024;
16const CRC_LEN: usize = 4;
17const OFFSET_LEN: usize = 8;
18
19struct Record {
20 id: ObjectId,
21 crc32: Crc32,
22 offset: PackOffset,
23}
24
25type Bucket = Mutex<Option<BufWriter<std::fs::File>>>;
26
27pub(crate) struct Spool {
28 buckets: Vec<Bucket>,
29 record_len: usize,
30 hash_len: usize,
31}
32
33impl Spool {
34 pub(crate) fn new(kind: gix::hash::Kind) -> Self {
35 let hash_len = kind.len_in_bytes();
36 Self {
37 buckets: (0..BUCKETS).map(|_| Mutex::new(None)).collect(),
38 record_len: hash_len + CRC_LEN + OFFSET_LEN,
39 hash_len,
40 }
41 }
42
43 pub(crate) fn push(&self, id: ObjectId, crc32: Crc32, offset: PackOffset) -> io::Result<()> {
44 let mut guard = self.buckets[id.first_byte() as usize]
45 .lock()
46 .expect("spool bucket poisoned");
47 let writer = match guard.as_mut() {
48 Some(writer) => writer,
49 None => guard.insert(BufWriter::with_capacity(BUCKET_BUF, tempfile::tempfile()?)),
50 };
51 writer.write_all(id.as_slice())?;
52 writer.write_all(&crc32.get().to_be_bytes())?;
53 writer.write_all(&offset.get().to_be_bytes())
54 }
55
56 fn cumulative_fanout(&self) -> Result<[u32; 256], PackError> {
57 let mut fanout = [0u32; 256];
58 self.buckets.iter().enumerate().try_for_each(
59 |(bucket, cell)| -> Result<(), PackError> {
60 let mut guard = cell.lock().expect("spool bucket poisoned");
61 fanout[bucket] = match guard.as_mut() {
62 None => 0,
63 Some(writer) => {
64 writer.flush()?;
65 (writer.get_ref().metadata()?.len() as usize / self.record_len) as u32
66 }
67 };
68 Ok(())
69 },
70 )?;
71 fanout.iter_mut().fold(0u32, |acc, count| {
72 *count += acc;
73 *count
74 });
75 Ok(fanout)
76 }
77
78 fn visit_sorted(
79 &self,
80 mut visit: impl FnMut(&Record) -> Result<(), PackError>,
81 ) -> Result<(), PackError> {
82 self.buckets.iter().try_for_each(|cell| {
83 let mut records = self.read_bucket(cell)?;
84 records.sort_unstable_by_key(|record| record.id);
85 records.iter().try_for_each(&mut visit)
86 })
87 }
88
89 fn read_bucket(&self, cell: &Bucket) -> Result<Vec<Record>, PackError> {
90 let mut guard = cell.lock().expect("spool bucket poisoned");
91 let bytes = match guard.as_mut() {
92 None => Vec::new(),
93 Some(writer) => {
94 writer.flush()?;
95 let file = writer.get_ref();
96 let len = file.metadata()?.len() as usize;
97 let mut bytes = vec![0u8; len];
98 file.read_exact_at(&mut bytes, 0)?;
99 bytes
100 }
101 };
102 drop(guard);
103 bytes
104 .chunks_exact(self.record_len)
105 .map(|chunk| {
106 let (id, rest) = chunk.split_at(self.hash_len);
107 Ok(Record {
108 id: ObjectId::try_from(id)
109 .map_err(|error| PackError::Pack(format!("spool record oid: {error}")))?,
110 crc32: Crc32::new(u32::from_be_bytes(
111 rest[..CRC_LEN].try_into().expect("crc slice"),
112 )),
113 offset: PackOffset::new(u64::from_be_bytes(
114 rest[CRC_LEN..].try_into().expect("offset slice"),
115 )),
116 })
117 })
118 .collect()
119 }
120}
121
122fn feed(out: &mut dyn Write, hasher: &mut gix_hash::Hasher, buf: &[u8]) -> io::Result<()> {
123 hasher.update(buf);
124 out.write_all(buf)
125}
126
127pub(crate) fn write_v2_index(
128 out: &mut dyn Write,
129 records: &Spool,
130 pack_hash: &ObjectId,
131 kind: gix::hash::Kind,
132) -> Result<ObjectId, PackError> {
133 let mut hasher = gix_hash::hasher(kind);
134 feed(out, &mut hasher, V2_SIGNATURE)?;
135 feed(out, &mut hasher, &V2_VERSION.to_be_bytes())?;
136
137 records
138 .cumulative_fanout()?
139 .iter()
140 .try_for_each(|count| feed(out, &mut hasher, &count.to_be_bytes()))?;
141 records.visit_sorted(|record| Ok(feed(out, &mut hasher, record.id.as_slice())?))?;
142 records
143 .visit_sorted(|record| Ok(feed(out, &mut hasher, &record.crc32.get().to_be_bytes())?))?;
144
145 let mut large_offsets = Vec::<u64>::new();
146 records.visit_sorted(|record| {
147 let encoded = if record.offset.get() > LARGE_OFFSET_THRESHOLD {
148 let position = large_offsets.len() as u32;
149 large_offsets.push(record.offset.get());
150 position | HIGH_BIT
151 } else {
152 record.offset.get() as u32
153 };
154 Ok(feed(out, &mut hasher, &encoded.to_be_bytes())?)
155 })?;
156 large_offsets
157 .iter()
158 .try_for_each(|offset| feed(out, &mut hasher, &offset.to_be_bytes()))?;
159
160 feed(out, &mut hasher, pack_hash.as_slice())?;
161
162 let index_hash = hasher
163 .try_finalize()
164 .map_err(|error| PackError::Pack(format!("finalize index hash: {error}")))?;
165 out.write_all(index_hash.as_slice())?;
166 out.flush()?;
167 Ok(index_hash)
168}
169
170#[cfg(test)]
171mod tests {
172 use super::*;
173
174 fn oid(seed: u8) -> ObjectId {
175 let mut raw = [0u8; 20];
176 raw[0] = seed;
177 raw[19] = seed;
178 ObjectId::try_from(raw.as_slice()).unwrap()
179 }
180
181 #[test]
182 fn large_offsets_round_trip_through_the_index_reader() {
183 let records = [
184 (oid(0x02), 0x1111_1111u32, 12u64),
185 (oid(0x40), 0x2222_2222, LARGE_OFFSET_THRESHOLD),
186 (oid(0x80), 0x3333_3333, 0x1_2345_6789),
187 (oid(0xc0), 0x4444_4444, LARGE_OFFSET_THRESHOLD + 1),
188 ];
189 let spool = Spool::new(gix::hash::Kind::Sha1);
190 records.iter().for_each(|(id, crc32, offset)| {
191 spool
192 .push(*id, Crc32::new(*crc32), PackOffset::new(*offset))
193 .unwrap()
194 });
195 let pack_hash = oid(0xaa);
196
197 let mut buf = Vec::new();
198 let index_hash =
199 write_v2_index(&mut buf, &spool, &pack_hash, gix::hash::Kind::Sha1).unwrap();
200
201 let dir = tempfile::tempdir().unwrap();
202 let path = dir.path().join("pack-under-test.idx");
203 std::fs::write(&path, &buf).unwrap();
204 let index = gix_pack::index::File::at(&path, gix::hash::Kind::Sha1).unwrap();
205
206 assert_eq!(index.num_objects(), records.len() as u32);
207 assert_eq!(index.index_checksum(), index_hash);
208 assert_eq!(index.pack_checksum(), pack_hash);
209 records.iter().for_each(|(id, crc32, offset)| {
210 let at = index
211 .lookup(*id)
212 .expect("written oid resolves in the index");
213 assert_eq!(index.oid_at_index(at), id.as_ref());
214 assert_eq!(index.pack_offset_at_index(at), *offset);
215 assert_eq!(index.crc32_at_index(at), Some(*crc32));
216 });
217 }
218}