1use crate::hash::{FastBuildHasher, hash_u64};
9use crate::state::{ColorCoordinate, UnitigColor};
10use scc::{HashMap as SccHashMap, hash_map::Entry as SccEntry};
11use std::fs::{self, File, OpenOptions};
12use std::io::{BufReader, BufWriter, Read, Seek, Write};
13use std::os::unix::fs::FileExt;
14use std::path::{Path, PathBuf};
15use std::sync::{
16 Mutex,
17 atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering},
18};
19
20const COLOR_BUFFER_BYTES: usize = 128 * 1024;
21
22pub fn reverse_color_runs(runs: &[UnitigColor], vertex_count: u32) -> Vec<UnitigColor> {
23 let mut reversed = Vec::with_capacity(runs.len());
24 for index in (0..runs.len()).rev() {
25 let end = runs
26 .get(index + 1)
27 .map_or(vertex_count, |next| next.offset());
28 reversed.push(UnitigColor::new(
29 vertex_count - end,
30 ColorCoordinate::from_u40(runs[index].coordinate()),
31 ));
32 }
33 reversed
34}
35
36pub fn reverse_color_runs_in_place(runs: &mut [UnitigColor], vertex_count: u32) {
37 if runs.is_empty() {
38 return;
39 }
40 runs.reverse();
41 for index in (1..runs.len()).rev() {
42 runs[index] = UnitigColor::new(
43 vertex_count - runs[index - 1].offset(),
44 ColorCoordinate::from_u40(runs[index].coordinate()),
45 );
46 }
47 runs[0] = UnitigColor::new(0, ColorCoordinate::from_u40(runs[0].coordinate()));
48}
49
50pub fn append_color_runs(
51 output: &mut Vec<UnitigColor>,
52 output_vertex_count: u32,
53 runs: &[UnitigColor],
54 unitig_vertex_count: u32,
55 reverse: bool,
56) {
57 if reverse {
58 output.reserve(runs.len());
59 for index in (0..runs.len()).rev() {
60 if index == runs.len() - 1
61 && output
62 .last()
63 .is_some_and(|left| left.coordinate() == runs[index].coordinate())
64 {
65 continue;
66 }
67 let end = runs
68 .get(index + 1)
69 .map_or(unitig_vertex_count, |next| next.offset());
70 output.push(UnitigColor::new(
71 output_vertex_count + unitig_vertex_count - end - 1,
72 ColorCoordinate::from_u40(runs[index].coordinate()),
73 ));
74 }
75 } else {
76 let start = usize::from(
77 output
78 .last()
79 .zip(runs.first())
80 .is_some_and(|(left, right)| left.coordinate() == right.coordinate()),
81 );
82 output.reserve(runs.len().saturating_sub(start));
83 for run in &runs[start..] {
84 output.push(UnitigColor::new(
85 output_vertex_count + run.offset() - 1,
86 ColorCoordinate::from_u40(run.coordinate()),
87 ));
88 }
89 }
90}
91
92pub fn rotate_cycle_color_runs(
93 runs: &[UnitigColor],
94 vertex_count: u32,
95 pivot: u32,
96 reverse: bool,
97) -> Vec<UnitigColor> {
98 if runs.is_empty() || vertex_count == 0 {
99 return Vec::new();
100 }
101 let oriented = if reverse {
102 reverse_color_runs(runs, vertex_count)
103 } else {
104 runs.to_vec()
105 };
106 let mut colors = vec![0u64; vertex_count as usize];
107 for (index, run) in oriented.iter().enumerate() {
108 let end = oriented
109 .get(index + 1)
110 .map_or(vertex_count, |next| next.offset());
111 colors[run.offset() as usize..end as usize].fill(run.coordinate());
112 }
113 colors.rotate_left((pivot % vertex_count) as usize);
114 compress_color_coordinates(&colors)
115}
116
117fn compress_color_coordinates(colors: &[u64]) -> Vec<UnitigColor> {
118 let mut runs = Vec::new();
119 for (offset, &coordinate) in colors.iter().enumerate() {
120 if runs
121 .last()
122 .is_none_or(|previous: &UnitigColor| previous.coordinate() != coordinate)
123 {
124 runs.push(UnitigColor::new(
125 offset as u32,
126 ColorCoordinate::from_u40(coordinate),
127 ));
128 }
129 }
130 runs
131}
132
133pub struct ColorRunSidecarWriter {
134 run_path: PathBuf,
135 runs: BufWriter<File>,
136 run_bytes: u64,
137 run_count: u64,
138 unitigs: u64,
139}
140
141impl ColorRunSidecarWriter {
142 pub fn create(path_prefix: impl AsRef<Path>) -> Result<Self, ColorError> {
143 let run_path = path_prefix.as_ref().with_extension("color-runs");
144 let run_file = OpenOptions::new()
145 .create(true)
146 .truncate(true)
147 .read(true)
148 .write(true)
149 .open(&run_path)
150 .map_err(|source| ColorError::Io {
151 path: run_path.clone(),
152 source,
153 })?;
154 Ok(Self {
155 run_path,
156 runs: BufWriter::with_capacity(COLOR_BUFFER_BYTES, run_file),
157 run_bytes: 0,
158 run_count: 0,
159 unitigs: 0,
160 })
161 }
162
163 pub fn position(&self) -> u64 {
164 self.run_bytes
165 }
166
167 pub fn write_unitig(&mut self, colors: &[UnitigColor]) -> Result<(), ColorError> {
168 let count = u32::try_from(colors.len()).map_err(|_| ColorError::TooManyColorRuns)?;
169 let mut encoded_count = Vec::with_capacity(5);
170 append_varint_u32(&mut encoded_count, count);
171 self.runs
172 .write_all(&encoded_count)
173 .map_err(|source| ColorError::Io {
174 path: self.run_path.clone(),
175 source,
176 })?;
177 for color in colors {
178 self.runs
179 .write_all(&color.raw().to_le_bytes())
180 .map_err(|source| ColorError::Io {
181 path: self.run_path.clone(),
182 source,
183 })?;
184 }
185 self.run_bytes += encoded_count.len() as u64 + u64::from(count) * 8;
186 self.run_count += u64::from(count);
187 self.unitigs += 1;
188 Ok(())
189 }
190
191 pub fn finish(mut self) -> Result<ColorRunSidecar, ColorError> {
192 self.runs.flush().map_err(|source| ColorError::Io {
193 path: self.run_path.clone(),
194 source,
195 })?;
196 self.runs.into_inner().map_err(|error| ColorError::Io {
197 path: self.run_path.clone(),
198 source: error.into_error(),
199 })?;
200 Ok(ColorRunSidecar {
201 run_path: self.run_path,
202 unitigs: self.unitigs,
203 runs: self.run_count,
204 })
205 }
206}
207
208pub struct ConcurrentColorRunSidecarWriter {
209 run_path: PathBuf,
210 runs: File,
211 run_bytes: AtomicU64,
212 run_count: AtomicU64,
213 unitigs: AtomicU64,
214}
215
216impl ConcurrentColorRunSidecarWriter {
217 pub fn create(path_prefix: impl AsRef<Path>) -> Result<Self, ColorError> {
218 let run_path = path_prefix.as_ref().with_extension("color-runs");
219 let runs = OpenOptions::new()
220 .create(true)
221 .truncate(true)
222 .read(true)
223 .write(true)
224 .open(&run_path)
225 .map_err(|source| ColorError::Io {
226 path: run_path.clone(),
227 source,
228 })?;
229 Ok(Self {
230 run_path,
231 runs,
232 run_bytes: AtomicU64::new(0),
233 run_count: AtomicU64::new(0),
234 unitigs: AtomicU64::new(0),
235 })
236 }
237
238 pub fn write_unitigs(&self, unitigs: &[Vec<UnitigColor>]) -> Result<u64, ColorError> {
239 let run_count = unitigs.iter().map(Vec::len).sum::<usize>();
240 let byte_len = unitigs
241 .len()
242 .checked_mul(5)
243 .and_then(|bytes| bytes.checked_add(run_count.checked_mul(8)?))
244 .ok_or(ColorError::TooManyColorRuns)?;
245 let mut encoded = Vec::with_capacity(byte_len);
246 for colors in unitigs {
247 let count = u32::try_from(colors.len()).map_err(|_| ColorError::TooManyColorRuns)?;
248 append_varint_u32(&mut encoded, count);
249 for color in colors {
250 encoded.extend_from_slice(&color.raw().to_le_bytes());
251 }
252 }
253 let offset = self
254 .run_bytes
255 .fetch_add(encoded.len() as u64, Ordering::Relaxed);
256 self.runs
257 .write_all_at(&encoded, offset)
258 .map_err(|source| ColorError::Io {
259 path: self.run_path.clone(),
260 source,
261 })?;
262 self.run_count
263 .fetch_add(run_count as u64, Ordering::Relaxed);
264 self.unitigs
265 .fetch_add(unitigs.len() as u64, Ordering::Relaxed);
266 Ok(offset)
267 }
268
269 pub fn finish(self) -> Result<ColorRunSidecar, ColorError> {
270 self.runs.sync_data().map_err(|source| ColorError::Io {
271 path: self.run_path.clone(),
272 source,
273 })?;
274 Ok(ColorRunSidecar {
275 run_path: self.run_path,
276 unitigs: self.unitigs.load(Ordering::Relaxed),
277 runs: self.run_count.load(Ordering::Relaxed),
278 })
279 }
280}
281
282#[derive(Debug, Clone)]
283pub struct ColorRunSidecar {
284 pub run_path: PathBuf,
285 pub unitigs: u64,
286 pub runs: u64,
287}
288
289impl ColorRunSidecar {
290 pub fn reader_at(&self, byte_offset: u64) -> Result<ColorRunStreamReader, ColorError> {
291 let mut file = File::open(&self.run_path).map_err(|source| ColorError::Io {
292 path: self.run_path.clone(),
293 source,
294 })?;
295 file.seek(std::io::SeekFrom::Start(byte_offset))
296 .map_err(|source| ColorError::Io {
297 path: self.run_path.clone(),
298 source,
299 })?;
300 Ok(ColorRunStreamReader {
301 path: self.run_path.clone(),
302 input: BufReader::with_capacity(COLOR_BUFFER_BYTES, file),
303 })
304 }
305
306 pub fn read_unitig(&self, unitig: u64) -> Result<Vec<UnitigColor>, ColorError> {
307 if unitig >= self.unitigs {
308 return Err(ColorError::MalformedUnitigIndex(unitig));
309 }
310 let mut reader = self.reader_at(0)?;
311 for index in 0..=unitig {
312 let colors = reader.read_next()?;
313 if index == unitig {
314 return Ok(colors);
315 }
316 }
317 Err(ColorError::MalformedUnitigIndex(unitig))
318 }
319}
320
321pub struct ColorRunStreamReader {
322 path: PathBuf,
323 input: BufReader<File>,
324}
325
326impl ColorRunStreamReader {
327 pub fn read_next(&mut self) -> Result<Vec<UnitigColor>, ColorError> {
328 let mut colors = Vec::new();
329 self.read_next_into(&mut colors)?;
330 Ok(colors)
331 }
332
333 pub fn read_next_into(&mut self, colors: &mut Vec<UnitigColor>) -> Result<(), ColorError> {
334 let count = read_varint(&mut self.input).map_err(|source| ColorError::Io {
335 path: self.path.clone(),
336 source,
337 })?;
338 colors.clear();
339 colors.reserve(count as usize);
340 for _ in 0..count {
341 let mut raw = [0u8; 8];
342 self.input
343 .read_exact(&mut raw)
344 .map_err(|source| ColorError::Io {
345 path: self.path.clone(),
346 source,
347 })?;
348 let raw = u64::from_le_bytes(raw);
349 colors.push(UnitigColor::new(
350 (raw & 0xff_ffff) as u32,
351 ColorCoordinate::from_u40(raw >> 24),
352 ));
353 }
354 Ok(())
355 }
356}
357
358pub struct ConcurrentColorRepository {
359 dir: PathBuf,
360 num_colors: u32,
363 table: AtomicColorTable,
364 overflow: SccHashMap<u64, ColorCoordinate, FastBuildHasher>,
365 workers: Vec<Mutex<ColorWorkerWriter>>,
366}
367
368struct AtomicColorSlot {
369 key: AtomicU64,
370 value: AtomicU64,
371}
372
373struct AtomicColorTable {
374 slots: Vec<AtomicColorSlot>,
375 mask: usize,
376 entries: AtomicUsize,
377 saturation_entries: usize,
378 active_insertions: AtomicUsize,
379 saturated: AtomicBool,
380 quiesced: AtomicBool,
382}
383
384enum AtomicColorEntry<'a> {
385 Occupied(ColorCoordinate),
386 Vacant(&'a AtomicColorSlot),
387 Full,
388}
389
390fn max_color_table_slots() -> usize {
402 const DEFAULT_MAX_SLOTS: usize = 64 * 1024 * 1024;
403 std::env::var("CF3_RS_COLOR_TABLE_SLOTS")
404 .ok()
405 .and_then(|value| value.parse::<usize>().ok())
406 .filter(|value| value.is_power_of_two())
407 .unwrap_or(DEFAULT_MAX_SLOTS)
408}
409
410impl AtomicColorTable {
411 const EMPTY_KEY: u64 = u64::MAX;
412 const PENDING_VALUE: u64 = u64::MAX;
413
414 fn with_expected_entries(expected: usize) -> Self {
415 let capacity = expected
416 .max(8)
417 .saturating_mul(4)
418 .div_ceil(3)
419 .next_power_of_two()
420 .min(max_color_table_slots());
421 Self {
422 slots: (0..capacity)
423 .map(|_| AtomicColorSlot {
424 key: AtomicU64::new(Self::EMPTY_KEY),
425 value: AtomicU64::new(Self::PENDING_VALUE),
426 })
427 .collect(),
428 mask: capacity - 1,
429 entries: AtomicUsize::new(0),
430 saturation_entries: capacity * 3 / 4,
431 active_insertions: AtomicUsize::new(0),
432 saturated: AtomicBool::new(false),
433 quiesced: AtomicBool::new(false),
434 }
435 }
436
437 #[inline]
438 fn entry(&self, key: u64) -> AtomicColorEntry<'_> {
439 if key == Self::EMPTY_KEY {
440 return AtomicColorEntry::Full;
441 }
442 if let Some(coordinate) = self.get(key) {
449 return AtomicColorEntry::Occupied(coordinate);
450 }
451 if self.saturated.load(Ordering::Acquire) {
454 return AtomicColorEntry::Full;
455 }
456 self.active_insertions.fetch_add(1, Ordering::AcqRel);
457 if self.saturated.load(Ordering::Acquire) {
458 self.active_insertions.fetch_sub(1, Ordering::Release);
459 return AtomicColorEntry::Full;
460 }
461 let hash = hash_u64(key, 0);
462 let mut index = hash as usize & self.mask;
463 for _ in 0..self.slots.len() {
464 let slot = &self.slots[index];
465 let observed = slot.key.load(Ordering::Acquire);
466 if observed == Self::EMPTY_KEY {
467 if slot
468 .key
469 .compare_exchange(Self::EMPTY_KEY, key, Ordering::AcqRel, Ordering::Relaxed)
470 .is_ok()
471 {
472 return AtomicColorEntry::Vacant(slot);
473 }
474 continue;
475 }
476 if observed == key {
477 let mut value = slot.value.load(Ordering::Acquire);
478 while value == Self::PENDING_VALUE {
479 std::hint::spin_loop();
480 value = slot.value.load(Ordering::Acquire);
481 }
482 self.active_insertions.fetch_sub(1, Ordering::Release);
483 return AtomicColorEntry::Occupied(ColorCoordinate::from_u40(value));
484 }
485 index = (index + 1) & self.mask;
486 }
487 self.saturated.store(true, Ordering::Release);
488 self.active_insertions.fetch_sub(1, Ordering::Release);
489 AtomicColorEntry::Full
490 }
491
492 #[inline]
493 fn get(&self, key: u64) -> Option<ColorCoordinate> {
494 if key == Self::EMPTY_KEY {
495 return None;
496 }
497 let hash = hash_u64(key, 0);
498 let mut index = hash as usize & self.mask;
499 for _ in 0..self.slots.len() {
500 let slot = &self.slots[index];
501 let observed = slot.key.load(Ordering::Acquire);
502 if observed == Self::EMPTY_KEY {
503 return None;
504 }
505 if observed == key {
506 let mut value = slot.value.load(Ordering::Acquire);
507 while value == Self::PENDING_VALUE {
508 std::hint::spin_loop();
509 value = slot.value.load(Ordering::Acquire);
510 }
511 return Some(ColorCoordinate::from_u40(value));
512 }
513 index = (index + 1) & self.mask;
514 }
515 None
516 }
517
518 fn publish(&self, slot: &AtomicColorSlot, coordinate: ColorCoordinate) {
519 slot.value.store(coordinate.as_u40(), Ordering::Release);
520 if self.entries.fetch_add(1, Ordering::AcqRel) + 1 >= self.saturation_entries {
521 self.saturated.store(true, Ordering::Release);
522 }
523 self.active_insertions.fetch_sub(1, Ordering::Release);
524 }
525
526 fn abort(&self, slot: &AtomicColorSlot) {
527 slot.value.store(Self::PENDING_VALUE, Ordering::Relaxed);
528 slot.key.store(Self::EMPTY_KEY, Ordering::Release);
529 self.active_insertions.fetch_sub(1, Ordering::Release);
530 }
531
532 #[inline]
533 fn is_saturated(&self) -> bool {
534 self.saturated.load(Ordering::Acquire)
535 }
536
537 fn wait_until_quiescent(&self) {
544 if self.quiesced.load(Ordering::Acquire) {
545 return;
546 }
547 while self.active_insertions.load(Ordering::Acquire) != 0 {
548 std::hint::spin_loop();
549 }
550 self.quiesced.store(true, Ordering::Release);
551 }
552}
553
554struct ColorWorkerWriter {
555 records: u32,
556 output: BufWriter<File>,
557 bits: BitWriter,
559}
560
561impl ConcurrentColorRepository {
562 pub(crate) fn worker_count(&self) -> usize {
563 self.workers.len()
564 }
565
566 pub fn create(
567 dir: impl AsRef<Path>,
568 workers: usize,
569 expected_colors: usize,
570 num_colors: u32,
571 ) -> Result<Self, ColorError> {
572 if workers == 0 || workers > 256 {
573 return Err(ColorError::InvalidWorkerCount(workers));
574 }
575 let dir = dir.as_ref().to_path_buf();
576 if dir.exists() {
577 fs::remove_dir_all(&dir).map_err(|source| ColorError::Io {
578 path: dir.clone(),
579 source,
580 })?;
581 }
582 fs::create_dir_all(&dir).map_err(|source| ColorError::Io {
583 path: dir.clone(),
584 source,
585 })?;
586 let mut worker_writers = Vec::with_capacity(workers);
587 for worker in 0..workers {
588 let path = color_worker_path(&dir, worker);
589 let file = File::create(&path).map_err(|source| ColorError::Io {
590 path: path.clone(),
591 source,
592 })?;
593 worker_writers.push(Mutex::new(ColorWorkerWriter {
594 records: 0,
595 output: BufWriter::with_capacity(COLOR_BUFFER_BYTES, file),
596 bits: BitWriter::default(),
597 }));
598 }
599 Ok(Self {
600 dir,
601 num_colors,
602 table: AtomicColorTable::with_expected_entries(expected_colors),
603 overflow: SccHashMap::with_capacity_and_hasher(8, FastBuildHasher::default()),
604 workers: worker_writers,
605 })
606 }
607
608 pub fn resolve_or_insert(
609 &self,
610 color_hash: u64,
611 sources: &[u32],
612 worker: usize,
613 ) -> Result<ColorCoordinate, ColorError> {
614 if sources.is_empty() {
615 return self.get(color_hash).ok_or(ColorError::MalformedSourceSet);
616 }
617 if !sources.windows(2).all(|pair| pair[0] < pair[1]) {
618 return Err(ColorError::MalformedSourceSet);
619 }
620 if worker >= self.workers.len() {
621 return Err(ColorError::InvalidWorkerCount(worker + 1));
622 }
623 match self.table.entry(color_hash) {
624 AtomicColorEntry::Occupied(coordinate) => Ok(coordinate),
625 AtomicColorEntry::Vacant(slot) => {
626 let mut writer = self.workers[worker]
627 .lock()
628 .map_err(|_| ColorError::PoisonedWriter)?;
629 let index = writer.records;
630 if let Err(source) = write_color_record(&mut writer, sources, self.num_colors) {
631 self.table.abort(slot);
632 return Err(ColorError::Io {
633 path: color_worker_path(&self.dir, worker),
634 source,
635 });
636 }
637 let Some(next_records) = writer.records.checked_add(1) else {
638 self.table.abort(slot);
639 return Err(ColorError::TooManyColors);
640 };
641 writer.records = next_records;
642 let coordinate = ColorCoordinate::discovered(worker as u64, index as u64);
643 self.table.publish(slot, coordinate);
644 Ok(coordinate)
645 }
646 AtomicColorEntry::Full => {
647 self.table.wait_until_quiescent();
648 if let Some(coordinate) = self.table.get(color_hash) {
649 return Ok(coordinate);
650 }
651 match self.overflow.entry_sync(color_hash) {
652 SccEntry::Occupied(entry) => Ok(*entry.get()),
653 SccEntry::Vacant(entry) => {
654 let mut writer = self.workers[worker]
655 .lock()
656 .map_err(|_| ColorError::PoisonedWriter)?;
657 let index = writer.records;
658 write_color_record(&mut writer, sources, self.num_colors).map_err(
659 |source| ColorError::Io {
660 path: color_worker_path(&self.dir, worker),
661 source,
662 },
663 )?;
664 writer.records = writer
665 .records
666 .checked_add(1)
667 .ok_or(ColorError::TooManyColors)?;
668 let coordinate = ColorCoordinate::discovered(worker as u64, index as u64);
669 entry.insert_entry(coordinate);
670 Ok(coordinate)
671 }
672 }
673 }
674 }
675 }
676
677 pub fn get(&self, color_hash: u64) -> Option<ColorCoordinate> {
678 if self.table.is_saturated() {
679 self.table.wait_until_quiescent();
680 }
681 self.table.get(color_hash).or_else(|| {
682 self.overflow
683 .read_sync(&color_hash, |_, coordinate| *coordinate)
684 })
685 }
686
687 pub fn finish(&self) -> Result<ColorRepositoryManifest, ColorError> {
688 eprintln!(
689 "cuttlefish: color repository table capacity {}, overflow entries {}",
690 self.table.slots.len(),
691 self.overflow.len()
692 );
693 let mut records = Vec::with_capacity(self.workers.len());
694 for (worker, writer) in self.workers.iter().enumerate() {
695 let mut writer = writer.lock().map_err(|_| ColorError::PoisonedWriter)?;
696 writer.output.flush().map_err(|source| ColorError::Io {
697 path: color_worker_path(&self.dir, worker),
698 source,
699 })?;
700 records.push(writer.records);
701 }
702 let num_colors = self.num_colors;
703 let manifest = ColorRepositoryManifest {
704 dir: self.dir.clone(),
705 records,
706 num_colors,
707 };
708 manifest.write()?;
709 Ok(manifest)
710 }
711}
712
713#[derive(Debug, Clone, PartialEq, Eq)]
714pub struct ColorRepositoryManifest {
715 pub dir: PathBuf,
716 pub records: Vec<u32>,
717 pub num_colors: u32,
719}
720
721impl ColorRepositoryManifest {
722 fn write(&self) -> Result<(), ColorError> {
723 let path = self.dir.join("manifest.tsv");
724 let mut output = BufWriter::new(File::create(&path).map_err(|source| ColorError::Io {
725 path: path.clone(),
726 source,
727 })?);
728 writeln!(output, "worker\trecords\tpath").map_err(|source| ColorError::Io {
729 path: path.clone(),
730 source,
731 })?;
732 for (worker, records) in self.records.iter().enumerate() {
733 writeln!(output, "{worker}\t{records}\t{:03}.colors", worker,).map_err(|source| {
734 ColorError::Io {
735 path: path.clone(),
736 source,
737 }
738 })?;
739 }
740 output
741 .flush()
742 .map_err(|source| ColorError::Io { path, source })
743 }
744
745 pub fn write_metadata(
746 &self,
747 k: u16,
748 fasta_path: &Path,
749 sources: &[PathBuf],
750 ) -> Result<(), ColorError> {
751 let metadata_path = self.dir.join("metadata.tsv");
752 let mut metadata =
753 BufWriter::new(
754 File::create(&metadata_path).map_err(|source| ColorError::Io {
755 path: metadata_path.clone(),
756 source,
757 })?,
758 );
759 writeln!(metadata, "format\tcf3rs-color-repository-v2")
760 .and_then(|_| writeln!(metadata, "k\t{k}"))
761 .and_then(|_| writeln!(metadata, "fasta\t{}", fasta_path.display()))
762 .and_then(|_| writeln!(metadata, "coordinate\tworker:u8,index:u32"))
763 .and_then(|_| {
764 writeln!(
765 metadata,
766 "encoding\thybrid-elias-delta: sparse gaps, bitmap, complement gaps"
767 )
768 })
769 .and_then(|_| writeln!(metadata, "source_count\t{}", sources.len()))
770 .map_err(|source| ColorError::Io {
771 path: metadata_path.clone(),
772 source,
773 })?;
774 for (index, source_path) in sources.iter().enumerate() {
775 writeln!(metadata, "source\t{}\t{}", index + 1, source_path.display()).map_err(
776 |source| ColorError::Io {
777 path: metadata_path.clone(),
778 source,
779 },
780 )?;
781 }
782 metadata.flush().map_err(|source| ColorError::Io {
783 path: metadata_path,
784 source,
785 })
786 }
787
788 pub fn read_color(&self, coordinate: ColorCoordinate) -> Result<Vec<u32>, ColorError> {
789 let worker = coordinate.worker();
790 let target = coordinate.index();
791 if worker >= self.records.len() || target >= self.records[worker] {
792 return Err(ColorError::MalformedCoordinate(coordinate.as_u40()));
793 }
794 let path = color_worker_path(&self.dir, worker);
795 let mut input = BufReader::with_capacity(
796 COLOR_BUFFER_BYTES,
797 File::open(&path).map_err(|source| ColorError::Io {
798 path: path.clone(),
799 source,
800 })?,
801 );
802 for index in 0..=target {
803 let sources = read_color_record(&mut input, self.num_colors).map_err(|source| {
804 ColorError::Io {
805 path: path.clone(),
806 source,
807 }
808 })?;
809 if index == target {
810 return Ok(sources);
811 }
812 }
813 Err(ColorError::MalformedCoordinate(coordinate.as_u40()))
814 }
815}
816
817fn color_worker_path(dir: &Path, worker: usize) -> PathBuf {
818 dir.join(format!("{worker:03}.colors"))
819}
820
821fn write_color_record(
826 writer: &mut ColorWorkerWriter,
827 sources: &[u32],
828 num_colors: u32,
829) -> std::io::Result<()> {
830 writer.bits.clear();
831 encode_source_set(&mut writer.bits, sources, num_colors);
832 let encoded = writer.bits.finish();
833 write_varint(&mut writer.output, encoded.len() as u32)?;
834 writer.output.write_all(encoded)
835}
836
837fn read_color_record(input: &mut impl Read, num_colors: u32) -> std::io::Result<Vec<u32>> {
839 let len = read_varint(input)? as usize;
840 let mut encoded = vec![0u8; len];
841 input.read_exact(&mut encoded)?;
842 decode_source_set(&mut BitReader::new(&encoded), num_colors)
843}
844
845#[derive(Default)]
847struct BitWriter {
848 bytes: Vec<u8>,
849 accumulator: u64,
850 pending: u32,
851 bitmap: Vec<u64>,
854}
855
856impl BitWriter {
857 fn clear(&mut self) {
858 self.bytes.clear();
859 self.accumulator = 0;
860 self.pending = 0;
861 }
862
863 fn push(&mut self, value: u64, count: u32) {
864 debug_assert!(count <= 64);
865 if count == 0 {
866 return;
867 }
868 let value = value & mask(count);
869 let free = 64 - self.pending;
872 if count < free {
873 self.accumulator |= value << self.pending;
874 self.pending += count;
875 return;
876 }
877 self.accumulator |= value << self.pending;
878 self.bytes
879 .extend_from_slice(&self.accumulator.to_le_bytes());
880 self.accumulator = if free == 64 { 0 } else { value >> free };
881 self.pending = count - free;
882 }
883
884 fn finish(&mut self) -> &[u8] {
886 while self.pending > 0 {
889 self.bytes.push(self.accumulator as u8);
890 self.accumulator >>= 8;
891 self.pending = self.pending.saturating_sub(8);
892 }
893 self.accumulator = 0;
894 &self.bytes
895 }
896}
897
898struct BitReader<'a> {
900 bytes: &'a [u8],
901 position: usize,
902}
903
904impl<'a> BitReader<'a> {
905 fn new(bytes: &'a [u8]) -> Self {
906 Self { bytes, position: 0 }
907 }
908
909 fn bit(&mut self) -> std::io::Result<u64> {
910 let byte = self
911 .bytes
912 .get(self.position / 8)
913 .ok_or_else(|| std::io::Error::from(std::io::ErrorKind::UnexpectedEof))?;
914 let bit = (byte >> (self.position % 8)) & 1;
915 self.position += 1;
916 Ok(u64::from(bit))
917 }
918
919 fn take(&mut self, count: u32) -> std::io::Result<u64> {
920 let mut value = 0u64;
921 for index in 0..count {
922 value |= self.bit()? << index;
923 }
924 Ok(value)
925 }
926
927 fn skip_zeros(&mut self) -> std::io::Result<u64> {
928 let mut zeros = 0u64;
929 while self.bit()? == 0 {
930 zeros += 1;
931 }
932 Ok(zeros)
933 }
934}
935
936fn mask(bits: u32) -> u64 {
937 if bits >= 64 {
938 u64::MAX
939 } else {
940 (1u64 << bits) - 1
941 }
942}
943
944fn msb(value: u64) -> u32 {
946 63 - value.leading_zeros()
947}
948
949fn write_gamma(out: &mut BitWriter, value: u64) {
950 let shifted = value + 1;
951 let bits = msb(shifted);
952 out.push(1u64 << bits, bits + 1);
954 out.push(shifted & mask(bits), bits);
955}
956
957fn read_gamma(input: &mut BitReader<'_>) -> std::io::Result<u64> {
958 let bits = input.skip_zeros()? as u32;
959 Ok((input.take(bits)? | (1u64 << bits)) - 1)
960}
961
962fn write_delta(out: &mut BitWriter, value: u64) {
963 let shifted = value + 1;
964 let bits = msb(shifted);
965 write_gamma(out, u64::from(bits));
966 out.push(shifted & mask(bits), bits);
967}
968
969fn read_delta(input: &mut BitReader<'_>) -> std::io::Result<u64> {
970 let bits = read_gamma(input)? as u32;
971 Ok((input.take(bits)? | (1u64 << bits)) - 1)
972}
973
974fn encode_source_set(out: &mut BitWriter, sources: &[u32], num_colors: u32) {
984 let len = sources.len() as u64;
985 write_delta(out, len);
986 if sources.is_empty() {
987 return;
988 }
989 let sparse_threshold = u64::from(num_colors) / 4;
990 let dense_threshold = u64::from(num_colors) * 3 / 4;
991 if len < sparse_threshold {
992 write_delta(out, u64::from(sources[0]));
993 for pair in sources.windows(2) {
994 write_delta(out, u64::from(pair[1] - (pair[0] + 1)));
995 }
996 } else if len < dense_threshold {
997 let words = (num_colors as usize).div_ceil(64);
998 out.bitmap.clear();
999 out.bitmap.resize(words, 0);
1000 for &source in sources {
1001 out.bitmap[source as usize / 64] |= 1u64 << (source % 64);
1002 }
1003 let mut remaining = num_colors;
1004 for index in 0..words {
1005 let word = out.bitmap[index];
1006 let width = remaining.min(64);
1007 out.push(word, width);
1008 remaining -= width;
1009 }
1010 } else {
1011 let mut previous = u32::MAX;
1019 let mut absent = 0u32;
1020 for &source in sources {
1021 while absent < source {
1022 let gap = if previous == u32::MAX {
1023 absent
1024 } else {
1025 absent - (previous + 1)
1026 };
1027 write_delta(out, u64::from(gap));
1028 previous = absent;
1029 absent += 1;
1030 }
1031 absent = source + 1;
1032 }
1033 while absent < num_colors {
1034 let gap = if previous == u32::MAX {
1035 absent
1036 } else {
1037 absent - (previous + 1)
1038 };
1039 write_delta(out, u64::from(gap));
1040 previous = absent;
1041 absent += 1;
1042 }
1043 }
1044}
1045
1046fn decode_source_set(input: &mut BitReader<'_>, num_colors: u32) -> std::io::Result<Vec<u32>> {
1048 let len = read_delta(input)? as usize;
1049 if len == 0 {
1050 return Ok(Vec::new());
1051 }
1052 let sparse_threshold = u64::from(num_colors) / 4;
1053 let dense_threshold = u64::from(num_colors) * 3 / 4;
1054 let mut sources = Vec::with_capacity(len);
1055 if (len as u64) < sparse_threshold {
1056 let mut previous = read_delta(input)? as u32;
1057 sources.push(previous);
1058 for _ in 1..len {
1059 let gap = read_delta(input)? as u32;
1060 let value = previous
1061 .checked_add(1)
1062 .and_then(|base| base.checked_add(gap))
1063 .ok_or_else(|| std::io::Error::from(std::io::ErrorKind::InvalidData))?;
1064 sources.push(value);
1065 previous = value;
1066 }
1067 } else if (len as u64) < dense_threshold {
1068 let mut remaining = num_colors;
1069 let mut base = 0u32;
1070 while remaining > 0 {
1071 let width = remaining.min(64);
1072 let word = input.take(width)?;
1073 for offset in 0..width {
1074 if word >> offset & 1 == 1 {
1075 sources.push(base + offset);
1076 }
1077 }
1078 base += width;
1079 remaining -= width;
1080 }
1081 } else {
1082 let absent_count = num_colors as usize - len;
1084 let mut absent = Vec::with_capacity(absent_count);
1085 let mut previous: Option<u32> = None;
1086 for _ in 0..absent_count {
1087 let gap = read_delta(input)? as u32;
1088 let value = match previous {
1089 None => gap,
1090 Some(prior) => prior + 1 + gap,
1091 };
1092 absent.push(value);
1093 previous = Some(value);
1094 }
1095 let mut next_absent = absent.into_iter().peekable();
1096 for value in 0..num_colors {
1097 if next_absent.peek() == Some(&value) {
1098 next_absent.next();
1099 } else {
1100 sources.push(value);
1101 }
1102 }
1103 }
1104 Ok(sources)
1105}
1106
1107fn write_varint(output: &mut impl Write, mut value: u32) -> std::io::Result<()> {
1108 while value >= 0x80 {
1109 output.write_all(&[((value as u8) & 0x7f) | 0x80])?;
1110 value >>= 7;
1111 }
1112 output.write_all(&[value as u8])
1113}
1114
1115pub(crate) fn write_unitig_color_runs(
1116 output: &mut impl Write,
1117 colors: &[UnitigColor],
1118) -> std::io::Result<()> {
1119 let count = u32::try_from(colors.len())
1120 .map_err(|_| std::io::Error::from(std::io::ErrorKind::InvalidInput))?;
1121 write_varint(output, count)?;
1122 for color in colors {
1123 output.write_all(&color.raw().to_le_bytes())?;
1124 }
1125 Ok(())
1126}
1127
1128pub(crate) fn read_unitig_color_runs(
1129 input: &mut impl Read,
1130 colors: &mut Vec<UnitigColor>,
1131) -> std::io::Result<()> {
1132 let count = read_varint(input)?;
1133 colors.clear();
1134 colors.reserve(count as usize);
1135 for _ in 0..count {
1136 let mut raw = [0u8; 8];
1137 input.read_exact(&mut raw)?;
1138 let raw = u64::from_le_bytes(raw);
1139 colors.push(UnitigColor::new(
1140 (raw & 0xff_ffff) as u32,
1141 ColorCoordinate::from_u40(raw >> 24),
1142 ));
1143 }
1144 Ok(())
1145}
1146
1147fn append_varint_u32(output: &mut Vec<u8>, mut value: u32) {
1148 while value >= 0x80 {
1149 output.push(((value as u8) & 0x7f) | 0x80);
1150 value >>= 7;
1151 }
1152 output.push(value as u8);
1153}
1154
1155fn read_varint(input: &mut impl Read) -> std::io::Result<u32> {
1156 let mut value = 0u32;
1157 for shift in (0..35).step_by(7) {
1158 let mut byte = [0u8; 1];
1159 input.read_exact(&mut byte)?;
1160 value |= u32::from(byte[0] & 0x7f) << shift;
1161 if byte[0] & 0x80 == 0 {
1162 return Ok(value);
1163 }
1164 }
1165 Err(std::io::Error::from(std::io::ErrorKind::InvalidData))
1166}
1167
1168#[derive(Debug)]
1169pub enum ColorError {
1170 Io {
1171 path: PathBuf,
1172 source: std::io::Error,
1173 },
1174 InvalidWorkerCount(usize),
1175 MalformedSourceSet,
1176 MalformedCoordinate(u64),
1177 TooManyColors,
1178 TooManyColorRuns,
1179 MalformedUnitigIndex(u64),
1180 PoisonedWriter,
1181}
1182
1183impl std::fmt::Display for ColorError {
1184 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1185 match self {
1186 Self::Io { path, source } => write!(f, "{}: {source}", path.display()),
1187 Self::InvalidWorkerCount(count) => write!(f, "invalid color worker count: {count}"),
1188 Self::MalformedSourceSet => write!(f, "color source set is empty or not sorted"),
1189 Self::MalformedCoordinate(coord) => write!(f, "invalid color coordinate: {coord}"),
1190 Self::TooManyColors => write!(f, "worker color repository exceeds 2^32 entries"),
1191 Self::TooManyColorRuns => write!(f, "a local unitig exceeds 2^32 color runs"),
1192 Self::MalformedUnitigIndex(unitig) => {
1193 write!(f, "invalid local-unitig color index: {unitig}")
1194 }
1195 Self::PoisonedWriter => write!(f, "color repository writer lock is poisoned"),
1196 }
1197 }
1198}
1199
1200impl std::error::Error for ColorError {}
1201
1202#[cfg(test)]
1203mod tests {
1204 #[test]
1207 fn hybrid_color_sets_round_trip_across_regimes() {
1208 let num_colors = 512u32;
1209 let sparse = num_colors / 4;
1210 let dense = num_colors * 3 / 4;
1211 let mut cases: Vec<Vec<u32>> = vec![
1212 vec![0],
1213 vec![511],
1214 vec![0, 511],
1215 vec![0, 1, 2, 3],
1216 (0..num_colors).collect(),
1217 ];
1218 for size in [
1220 1,
1221 sparse - 1,
1222 sparse,
1223 sparse + 1,
1224 num_colors / 2,
1225 dense - 1,
1226 dense,
1227 dense + 1,
1228 num_colors - 1,
1229 ] {
1230 cases.push((0..size).collect());
1231 let stride = (num_colors / size.max(1)).max(1);
1233 cases.push(
1234 (0..size)
1235 .map(|index| (index * stride) % num_colors)
1236 .collect::<std::collections::BTreeSet<_>>()
1237 .into_iter()
1238 .collect(),
1239 );
1240 }
1241
1242 for sources in cases {
1243 let mut writer = BitWriter::default();
1244 encode_source_set(&mut writer, &sources, num_colors);
1245 let encoded = writer.finish().to_vec();
1246 let decoded = decode_source_set(&mut BitReader::new(&encoded), num_colors).unwrap();
1247 assert_eq!(
1248 decoded,
1249 sources,
1250 "round trip failed for {} sources",
1251 sources.len()
1252 );
1253 }
1254 }
1255
1256 #[test]
1259 fn bit_writer_round_trips_arbitrary_widths() {
1260 for start in 0..64u32 {
1261 let mut writer = BitWriter::default();
1262 writer.push(0, start);
1264 let values: Vec<(u64, u32)> = (1..=64u32)
1265 .map(|width| (0xdead_beef_cafe_babeu64 & mask(width), width))
1266 .collect();
1267 for &(value, width) in &values {
1268 writer.push(value, width);
1269 }
1270 let encoded = writer.finish().to_vec();
1271 let mut reader = BitReader::new(&encoded);
1272 assert_eq!(reader.take(start).unwrap(), 0);
1273 for &(value, width) in &values {
1274 assert_eq!(
1275 reader.take(width).unwrap(),
1276 value,
1277 "width {width} at start {start}"
1278 );
1279 }
1280 }
1281 }
1282
1283 #[test]
1284 fn elias_codes_round_trip() {
1285 let mut writer = BitWriter::default();
1286 let values: Vec<u64> = (0..64)
1287 .chain([1000, 65535, 1 << 20, u32::MAX as u64])
1288 .collect();
1289 for &value in &values {
1290 write_gamma(&mut writer, value);
1291 write_delta(&mut writer, value);
1292 }
1293 let encoded = writer.finish().to_vec();
1294 let mut reader = BitReader::new(&encoded);
1295 for &value in &values {
1296 assert_eq!(read_gamma(&mut reader).unwrap(), value);
1297 assert_eq!(read_delta(&mut reader).unwrap(), value);
1298 }
1299 }
1300
1301 #[test]
1304 fn dense_sets_encode_smaller_than_their_membership() {
1305 let num_colors = 4096u32;
1306 let sources: Vec<u32> = (0..num_colors).filter(|value| value % 64 != 0).collect();
1307 let mut writer = BitWriter::default();
1308 encode_source_set(&mut writer, &sources, num_colors);
1309 let encoded_len = writer.finish().len();
1310 assert!(
1311 encoded_len < sources.len() / 4,
1312 "dense set took {encoded_len} bytes for {} members",
1313 sources.len()
1314 );
1315 }
1316
1317 use super::*;
1318 use std::sync::Arc;
1319
1320 #[test]
1321 fn saturated_atomic_table_bypasses_full_table_probes() {
1322 let table = AtomicColorTable::with_expected_entries(8);
1323 assert_eq!(std::mem::size_of::<AtomicColorSlot>(), 16);
1324 for key in 0..table.saturation_entries as u64 {
1325 match table.entry(key) {
1326 AtomicColorEntry::Vacant(slot) => {
1327 table.publish(slot, ColorCoordinate::discovered(0, key));
1328 }
1329 AtomicColorEntry::Occupied(_) | AtomicColorEntry::Full => {
1330 panic!("primary color table saturated before all slots were populated")
1331 }
1332 }
1333 }
1334 assert!(matches!(
1335 table.entry(table.saturation_entries as u64),
1336 AtomicColorEntry::Full
1337 ));
1338 assert!(table.is_saturated());
1339
1340 table.slots[0]
1342 .value
1343 .store(AtomicColorTable::PENDING_VALUE, Ordering::Relaxed);
1344 assert!(matches!(table.entry(u64::MAX), AtomicColorEntry::Full));
1345 }
1346
1347 #[test]
1348 fn atomic_table_publishes_one_value_to_concurrent_duplicates() {
1349 let table = Arc::new(AtomicColorTable::with_expected_entries(8));
1350 let coordinates = std::thread::scope(|scope| {
1351 let mut handles = Vec::new();
1352 for _ in 0..16 {
1353 let table = Arc::clone(&table);
1354 handles.push(scope.spawn(move || match table.entry(17) {
1355 AtomicColorEntry::Vacant(slot) => {
1356 let coordinate = ColorCoordinate::discovered(3, 9);
1357 table.publish(slot, coordinate);
1358 coordinate
1359 }
1360 AtomicColorEntry::Occupied(coordinate) => coordinate,
1361 AtomicColorEntry::Full => panic!("small color table unexpectedly saturated"),
1362 }));
1363 }
1364 handles
1365 .into_iter()
1366 .map(|handle| handle.join().unwrap())
1367 .collect::<Vec<_>>()
1368 });
1369 assert!(
1370 coordinates
1371 .iter()
1372 .all(|coordinate| *coordinate == coordinates[0])
1373 );
1374 assert_eq!(table.entries.load(Ordering::Relaxed), 1);
1375 }
1376
1377 #[test]
1378 fn repository_deduplicates_hashes_and_round_trips_sparse_sets() {
1379 let dir = std::env::temp_dir().join(format!(
1380 "cf3-color-repo-{}-{:?}",
1381 std::process::id(),
1382 std::thread::current().id()
1383 ));
1384 let repository = ConcurrentColorRepository::create(&dir, 2, 8, 64).unwrap();
1385 let first = repository.resolve_or_insert(17, &[1, 2, 150], 0).unwrap();
1386 let duplicate = repository.resolve_or_insert(17, &[1, 2, 150], 1).unwrap();
1387 assert_eq!(first, duplicate);
1388 let second = repository.resolve_or_insert(23, &[3, 1000], 1).unwrap();
1389 let manifest = repository.finish().unwrap();
1390 assert_eq!(manifest.read_color(first).unwrap(), [1, 2, 150]);
1391 assert_eq!(manifest.read_color(second).unwrap(), [3, 1000]);
1392 manifest
1393 .write_metadata(31, Path::new("graph.fa"), &[PathBuf::from("source.fa")])
1394 .unwrap();
1395 let metadata = fs::read_to_string(dir.join("metadata.tsv")).unwrap();
1396 assert!(metadata.contains("format\tcf3rs-color-repository-v2"));
1397 assert!(metadata.contains("source\t1\tsource.fa"));
1398 let manifest_text = fs::read_to_string(dir.join("manifest.tsv")).unwrap();
1399 assert!(manifest_text.contains("000.colors"));
1400 assert!(!manifest_text.contains(&dir.display().to_string()));
1401 fs::remove_dir_all(dir).unwrap();
1402 }
1403
1404 #[test]
1405 fn repository_deduplicates_across_primary_overflow_handoff() {
1406 let dir = std::env::temp_dir().join(format!(
1407 "cf3-color-overflow-{}-{:?}",
1408 std::process::id(),
1409 std::thread::current().id()
1410 ));
1411 let repository = ConcurrentColorRepository::create(&dir, 2, 8, 64).unwrap();
1412 let mut coordinates = Vec::new();
1413 for key in 0..24u64 {
1414 coordinates.push(
1415 repository
1416 .resolve_or_insert(key, &[key as u32 + 1], key as usize % 2)
1417 .unwrap(),
1418 );
1419 }
1420 assert!(repository.table.is_saturated());
1421 for key in 0..24u64 {
1422 assert_eq!(
1423 repository
1424 .resolve_or_insert(key, &[key as u32 + 1], (key as usize + 1) % 2)
1425 .unwrap(),
1426 coordinates[key as usize]
1427 );
1428 }
1429 let manifest = repository.finish().unwrap();
1430 for (key, coordinate) in coordinates.into_iter().enumerate() {
1431 assert_eq!(manifest.read_color(coordinate).unwrap(), [key as u32 + 1]);
1432 }
1433 fs::remove_dir_all(dir).unwrap();
1434 }
1435
1436 #[test]
1437 fn color_runs_reverse_append_and_rotate_like_vertex_colors() {
1438 let a = ColorCoordinate::discovered(0, 1);
1439 let b = ColorCoordinate::discovered(0, 2);
1440 let c = ColorCoordinate::discovered(0, 3);
1441 let runs = vec![UnitigColor::new(0, a), UnitigColor::new(2, b)];
1442 let reversed = reverse_color_runs(&runs, 5);
1443 assert_eq!(
1444 reversed
1445 .iter()
1446 .map(|run| (run.offset(), run.coordinate()))
1447 .collect::<Vec<_>>(),
1448 [(0, b.as_u40()), (3, a.as_u40())]
1449 );
1450
1451 let mut joined = vec![UnitigColor::new(0, c), UnitigColor::new(3, a)];
1452 append_color_runs(&mut joined, 4, &runs, 5, false);
1453 assert_eq!(
1454 joined
1455 .iter()
1456 .map(|run| (run.offset(), run.coordinate()))
1457 .collect::<Vec<_>>(),
1458 [(0, c.as_u40()), (3, a.as_u40()), (5, b.as_u40())]
1459 );
1460
1461 let rotated = rotate_cycle_color_runs(&runs, 5, 2, false);
1462 assert_eq!(
1463 rotated
1464 .iter()
1465 .map(|run| (run.offset(), run.coordinate()))
1466 .collect::<Vec<_>>(),
1467 [(0, b.as_u40()), (3, a.as_u40())]
1468 );
1469 }
1470
1471 #[test]
1472 fn color_run_sidecar_supports_range_streams() {
1473 let dir = std::env::temp_dir().join(format!(
1474 "cf3-color-runs-{}-{:?}",
1475 std::process::id(),
1476 std::thread::current().id()
1477 ));
1478 fs::create_dir_all(&dir).unwrap();
1479 let mut writer = ColorRunSidecarWriter::create(dir.join("local")).unwrap();
1480 let coordinate = ColorCoordinate::discovered(1, 9);
1481 writer.write_unitig(&[]).unwrap();
1482 let second_offset = writer.position();
1483 writer
1484 .write_unitig(&[
1485 UnitigColor::new(0, coordinate),
1486 UnitigColor::new(7, coordinate),
1487 ])
1488 .unwrap();
1489 let sidecar = writer.finish().unwrap();
1490 assert!(sidecar.read_unitig(0).unwrap().is_empty());
1491 assert_eq!(
1492 sidecar
1493 .reader_at(second_offset)
1494 .unwrap()
1495 .read_next()
1496 .unwrap()
1497 .len(),
1498 2
1499 );
1500 assert_eq!(
1501 sidecar
1502 .read_unitig(1)
1503 .unwrap()
1504 .iter()
1505 .map(|color| color.raw())
1506 .collect::<Vec<_>>(),
1507 [
1508 UnitigColor::new(0, coordinate).raw(),
1509 UnitigColor::new(7, coordinate).raw()
1510 ]
1511 );
1512 fs::remove_dir_all(dir).unwrap();
1513 }
1514}