@@ -5,14 +5,11 @@ use colors::storage::ColorsSerializerTrait;
55use colors:: storage:: deserializer:: ColorsDeserializer ;
66use config:: { ColorIndexType , DEFAULT_PREFETCH_AMOUNT , KEEP_FILES } ;
77use io:: compressed_read:: CompressedReadIndipendent ;
8- use io:: concurrent:: temp_reads:: creads_utils:: {
9- CompressedReadsBucketDataSerializer , DeserializedRead , NoMinimizerPosition , NoMultiplicity ,
10- NoSecondBucket ,
11- } ;
8+ use io:: concurrent:: temp_reads:: creads_utils:: { DeserializedRead , NoAlignment } ;
9+ use minimizer_bucketing:: decode_helper:: decode_sequences;
10+ use minimizer_bucketing:: split_buckets:: SplittedBucket ;
1211use nightly_quirks:: slice_group_by:: SliceGroupBy ;
13- use parallel_processor:: buckets:: SingleBucket ;
14- use parallel_processor:: buckets:: readers:: binary_reader:: ChunkedBinaryReaderIndex ;
15- use parallel_processor:: buckets:: readers:: typed_binary_reader:: TypedStreamReader ;
12+ use parallel_processor:: buckets:: MultiChunkBucket ;
1613use parallel_processor:: fast_smart_bucket_sort:: { FastSortable , SortKey , fast_smart_radix_sort} ;
1714use parallel_processor:: memory_fs:: RemoveFileMode ;
1815use parallel_processor:: phase_times_monitor:: PHASES_TIMES_MONITOR ;
@@ -29,7 +26,8 @@ pub fn colormap_reading<
2926> (
3027 k : usize ,
3128 colormap_file : PathBuf ,
32- colored_unitigs_buckets : Vec < SingleBucket > ,
29+ colored_unitigs_buckets : Vec < MultiChunkBucket > ,
30+ sub_buckets_count : usize ,
3331 single_thread_output_function : bool ,
3432 output_function : impl Fn ( & [ u8 ] , & [ ColorIndexType ] , bool ) + Send + Sync ,
3533) -> anyhow:: Result < ( ) > {
@@ -54,92 +52,94 @@ pub fn colormap_reading<
5452 let mut temp_bases = Vec :: new ( ) ;
5553 let mut temp_sequences = Vec :: new ( ) ;
5654
57- let file_index = ChunkedBinaryReaderIndex :: from_file (
58- & input. path ,
55+ let mut splitted_buckets = SplittedBucket :: generate (
56+ input. chunks . iter ( ) ,
5957 RemoveFileMode :: Remove {
6058 remove_fs : !KEEP_FILES . load ( Ordering :: Relaxed ) ,
6159 } ,
6260 DEFAULT_PREFETCH_AMOUNT ,
61+ sub_buckets_count,
6362 ) ;
6463
65- TypedStreamReader :: get_items :: <
66- CompressedReadsBucketDataSerializer <
64+ for splitted_bucket in splitted_buckets . iter_mut ( ) . map ( |b| b . iter_mut ( ) ) . flatten ( ) {
65+ decode_sequences :: <
6766 DumperKmersReferenceData < SingleKmerColorDataType < CX > > ,
68- NoSecondBucket ,
69- NoMultiplicity ,
70- NoMinimizerPosition ,
71- typenum:: consts:: U0 ,
72- > ,
73- > (
74- None ,
75- k,
76- file_index. into_chunks ( ) ,
77- |DeserializedRead {
78- extra : color_extra,
79- read,
80- ..
81- } ,
82- _| {
83- let new_read = CompressedReadIndipendent :: from_read :: < true > ( & read, & mut temp_bases) ;
84- temp_sequences. push ( ( new_read, color_extra) ) ;
85- } ,
86- ) ;
87-
88- struct ColoredUnitigsCompare < CX : ColorsManager > ( PhantomData < & ' static CX > ) ;
89- impl < CX : ColorsManager >
90- SortKey < (
91- CompressedReadIndipendent ,
9267 DumperKmersReferenceData < SingleKmerColorDataType < CX > > ,
93- ) > for ColoredUnitigsCompare < CX >
94- {
95- type KeyType = SingleKmerColorDataType < CX > ;
96- const KEY_BITS : usize = std:: mem:: size_of :: < SingleKmerColorDataType < CX > > ( ) * 8 ;
97-
98- fn compare (
99- left : & (
100- CompressedReadIndipendent ,
101- DumperKmersReferenceData < SingleKmerColorDataType < CX > > ,
102- ) ,
103- right : & (
104- CompressedReadIndipendent ,
105- DumperKmersReferenceData < SingleKmerColorDataType < CX > > ,
106- ) ,
107- ) -> std:: cmp:: Ordering {
108- left. 1 . cmp ( & right. 1 )
109- }
110-
111- fn get_shifted (
112- value : & (
68+ typenum:: consts:: U0 ,
69+ NoAlignment ,
70+ > (
71+ None ,
72+ & mut ( ) ,
73+ splitted_bucket,
74+ k,
75+ |DeserializedRead {
76+ extra : color_extra,
77+ read,
78+ ..
79+ } ,
80+ _| {
81+ let new_read =
82+ CompressedReadIndipendent :: from_read :: < true > ( & read, & mut temp_bases) ;
83+ temp_sequences. push ( ( new_read, color_extra) ) ;
84+ } ,
85+ ) ;
86+
87+ struct ColoredUnitigsCompare < CX : ColorsManager > ( PhantomData < & ' static CX > ) ;
88+ impl < CX : ColorsManager >
89+ SortKey < (
11390 CompressedReadIndipendent ,
11491 DumperKmersReferenceData < SingleKmerColorDataType < CX > > ,
115- ) ,
116- rhs : u8 ,
117- ) -> u8 {
118- value. 1 . color . get_shifted ( rhs)
92+ ) > for ColoredUnitigsCompare < CX >
93+ {
94+ type KeyType = SingleKmerColorDataType < CX > ;
95+ const KEY_BITS : usize = std:: mem:: size_of :: < SingleKmerColorDataType < CX > > ( ) * 8 ;
96+
97+ fn compare (
98+ left : & (
99+ CompressedReadIndipendent ,
100+ DumperKmersReferenceData < SingleKmerColorDataType < CX > > ,
101+ ) ,
102+ right : & (
103+ CompressedReadIndipendent ,
104+ DumperKmersReferenceData < SingleKmerColorDataType < CX > > ,
105+ ) ,
106+ ) -> std:: cmp:: Ordering {
107+ left. 1 . cmp ( & right. 1 )
108+ }
109+
110+ fn get_shifted (
111+ value : & (
112+ CompressedReadIndipendent ,
113+ DumperKmersReferenceData < SingleKmerColorDataType < CX > > ,
114+ ) ,
115+ rhs : u8 ,
116+ ) -> u8 {
117+ value. 1 . color . get_shifted ( rhs)
118+ }
119119 }
120- }
121120
122- fast_smart_radix_sort :: < _ , ColoredUnitigsCompare < CX > , false > ( & mut temp_sequences[ ..] ) ;
121+ fast_smart_radix_sort :: < _ , ColoredUnitigsCompare < CX > , false > ( & mut temp_sequences[ ..] ) ;
123122
124- for unitigs_by_color in temp_sequences. nq_group_by_mut ( |a, b| a. 1 == b. 1 ) {
125- let color = unitigs_by_color[ 0 ] . 1 . color ;
126- temp_colors_buffer. clear ( ) ;
127- colormap_decoder. get_color_mappings ( color, & mut temp_colors_buffer) ;
123+ for unitigs_by_color in temp_sequences. nq_group_by_mut ( |a, b| a. 1 == b. 1 ) {
124+ let color = unitigs_by_color[ 0 ] . 1 . color ;
125+ temp_colors_buffer. clear ( ) ;
126+ colormap_decoder. get_color_mappings ( color, & mut temp_colors_buffer) ;
128127
129- let mut same_color = false ;
128+ let mut same_color = false ;
130129
131- let _lock = if single_thread_output_function {
132- Some ( single_thread_lock. lock ( ) )
133- } else {
134- None
135- } ;
130+ let _lock = if single_thread_output_function {
131+ Some ( single_thread_lock. lock ( ) )
132+ } else {
133+ None
134+ } ;
136135
137- for unitig in unitigs_by_color {
138- let read = unitig. 0 . as_reference ( & temp_bases) ;
139- temp_decompressed_sequence. clear ( ) ;
140- temp_decompressed_sequence. extend ( read. as_bases_iter ( ) ) ;
141- output_function ( & temp_decompressed_sequence, & temp_colors_buffer, same_color) ;
142- same_color = true ;
136+ for unitig in unitigs_by_color {
137+ let read = unitig. 0 . as_reference ( & temp_bases) ;
138+ temp_decompressed_sequence. clear ( ) ;
139+ temp_decompressed_sequence. extend ( read. as_bases_iter ( ) ) ;
140+ output_function ( & temp_decompressed_sequence, & temp_colors_buffer, same_color) ;
141+ same_color = true ;
142+ }
143143 }
144144 }
145145 } ) ;
0 commit comments