@@ -24,17 +24,20 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
2424#include " output/output_format.h"
2525#include " output/output.h"
2626#include " legacy/pipeline.h"
27- #include " util/async_buffer .h"
27+ #include " search/hit_buffer .h"
2828#include " util/parallel/thread_pool.h"
2929#include " extend.h"
3030#include " util/util.h"
31+ // #include "util/sort/sort.h"
3132#ifdef WITH_DNA
3233#include " ../dna/extension.h"
3334#endif
3435#include " util/util.h"
3536#define _REENTRANT
3637#include " ips4o/ips4o.hpp"
3738#include " data/queries.h"
39+ #include " data/sequence_file.h"
40+ #include " search/hit_buffer.h"
3841
3942using std::get;
4043using std::tuple;
@@ -47,12 +50,6 @@ using std::vector;
4750
4851DpStat dp_stat;
4952
50- namespace Extension {
51-
52- TextBuffer* pipeline_short (BlockId query, Search::Hit* begin, Search::Hit* end, Search::Config& cfg, Statistics& stats);
53-
54- }
55-
5653static vector<int64_t > make_partition (Search::Hit* begin, Search::Hit* end) {
5754 vector<int64_t > partition;
5855 partition.reserve (div_up (end - begin, (ptrdiff_t )config.min_task_trace_pts ) + 1 );
@@ -124,9 +121,9 @@ struct HitIterator {
124121static TextBuffer* legacy_pipeline (const HitIterator::Hits& hits, Search::Config& cfg, Statistics &stat) {
125122 if (hits.end == hits.begin ) {
126123 TextBuffer *buf = nullptr ;
127- if (!cfg.blocked_processing && *cfg.output_format != OutputFormat::daa && config. report_unaligned != 0 ) {
124+ if (!cfg.blocked_processing && *cfg.output_format != OutputFormat::daa && cfg. output_format -> report_unaligned () ) {
128125 buf = new TextBuffer;
129- Output::Info info{ cfg.query ->seq_info (hits.query ), true , cfg.db .get (), *buf, {}, Util::Seq::AccessionParsing (), cfg.db ->sequence_count (), cfg.db ->letters () };
126+ Output::Info info{ cfg.query ->seq_info (hits.query ), true , cfg.db .get (), *buf, Util::Seq::AccessionParsing (), cfg.db ->sequence_count (), cfg.db ->letters () };
130127 cfg.output_format ->print_query_intro (info);
131128 cfg.output_format ->print_query_epilog (info);
132129 }
@@ -161,6 +158,7 @@ static TextBuffer* legacy_pipeline(const HitIterator::Hits& hits, Search::Config
161158static void align_worker (HitIterator* hit_it, Search::Config* cfg, int64_t next)
162159{
163160 try {
161+ std::pmr::monotonic_buffer_resource pool;
164162 const vector<HitIterator::Hits> hits = hit_it->fetch (next);
165163 assert (!hits.empty ());
166164 Statistics stat;
@@ -173,23 +171,18 @@ static void align_worker(HitIterator* hit_it, Search::Config* cfg, int64_t next)
173171 output_sink->push (h->query , buf);
174172 continue ;
175173 }
176- if (config.pipeline_short ) {
177- TextBuffer* buf = Extension::pipeline_short (h->query , h->begin , h->end , *cfg, stat);
178- output_sink->push (h->query , buf);
179- continue ;
180- }
181174 if (h->begin == nullptr && !HitIterator::single_query ()) {
182175 output_sink->push (h->query , nullptr );
183176 continue ;
184177 }
185178
186- pair< vector<Extension::Match>, Extension::Stats > matches =
179+ vector<Extension::Match> matches =
187180#ifdef WITH_DNA
188181 align_mode.mode == AlignMode::blastn ? Dna::extend (*cfg, cfg->query ->seqs ()[h->query ]) :
189182#endif
190- Extension::extend (h->query , h->begin , h->end , *cfg, stat, parallel ? DP ::Flags::PARALLEL : DP ::Flags::NONE );
191- TextBuffer* buf = cfg->blocked_processing ? Extension::generate_intermediate_output (matches. first , h->query , *cfg) : Extension::generate_output (matches. first , matches. second , h->query , stat, *cfg);
192- if (!matches.first . empty () && cfg->track_aligned_queries ) {
183+ Extension::extend (h->query , h->begin , h->end , *cfg, stat, parallel ? DP ::Flags::PARALLEL : DP ::Flags::NONE , pool );
184+ TextBuffer* buf = cfg->blocked_processing ? Extension::generate_intermediate_output (matches, h->query , *cfg) : Extension::generate_output (matches, h->query , stat, *cfg);
185+ if (!matches.empty () && cfg->track_aligned_queries ) {
193186 std::lock_guard<std::mutex> lock (query_aligned_mtx);
194187 if (!query_aligned[h->query ]) {
195188 query_aligned[h->query ] = true ;
@@ -198,7 +191,7 @@ static void align_worker(HitIterator* hit_it, Search::Config* cfg, int64_t next)
198191 }
199192 if (!config.unaligned_targets .empty ()) {
200193 lock_guard<mutex> lock (cfg->aligned_targets_mtx );
201- for (const Extension::Match &m : matches. first ) {
194+ for (const Extension::Match &m : matches) {
202195 const OId oid = cfg->target ->block_id2oid (m.target_block_id );
203196 if (!cfg->aligned_targets [oid])
204197 cfg->aligned_targets [oid] = true ;
@@ -220,42 +213,48 @@ void align_queries(Consumer* output_file, Search::Config& cfg)
220213 const int64_t mem_limit = Util::String::interpret_number (config.memory_limit .get (" 16G"
221214
222215 pair<BlockId, BlockId> query_range;
223- TaskTimer timer (nullptr , 3 );
216+ TaskTimer timer (" Allocating memory " 3 );
224217
225218 if (!cfg.blocked_processing && !cfg.iterated ())
226219 cfg.db ->init_random_access (cfg.current_query_block , 0 , false );
227220
228221 int64_t res_size = cfg.query ->mem_size () + cfg.target ->mem_size (), last_size = 0 ;
229- cfg.seed_hit_buf ->load (std::min (mem_limit - res_size - cfg.seed_hit_buf ->bin_size (1 ) * (int64_t )sizeof (Search::Hit), config.trace_pt_fetch_size ));
222+ cfg.seed_hit_buf ->alloc_buffer ();
223+ // cfg.seed_hit_buf->load(std::min(mem_limit - res_size - cfg.seed_hit_buf->bin_size(1) * (int64_t)sizeof(Search::Hit), config.trace_pt_fetch_size));
230224
231225 while (true ) {
232226 timer.go (" Loading trace points"
233- tuple<vector<Search::Hit>*, BlockId, BlockId> input = cfg.seed_hit_buf ->retrieve ();
227+ cfg.seed_hit_buf ->load (std::min (mem_limit - res_size - cfg.seed_hit_buf ->bin_size (1 ) * (int64_t )sizeof (Search::Hit), config.trace_pt_fetch_size ));
228+ tuple<Search::Hit*, int64_t , BlockId, BlockId> input = cfg.seed_hit_buf ->retrieve ();
234229 if (get<0 >(input) == nullptr )
235230 break ;
236231 statistics.inc (Statistics::TIME_LOAD_SEED_HITS , timer.microseconds ());
237- vector<Search::Hit>* hit_buf = get<0 >(input);
238- res_size += hit_buf->size () * sizeof (Search::Hit);
239- query_range = { get<1 >(input), get<2 >(input) };
240- cfg.seed_hit_buf ->load (std::min (mem_limit - res_size, config.trace_pt_fetch_size ));
232+ timer.finish ();
233+ Search::Hit* hit_buf = get<0 >(input);
234+ const int64_t hit_count = get<1 >(input);
235+ log_stream << " Processing " " trace points."
236+ res_size += hit_count * sizeof (Search::Hit);
237+ query_range = { get<2 >(input), get<3 >(input) };
238+ // cfg.seed_hit_buf->load(std::min(mem_limit - res_size, config.trace_pt_fetch_size));
241239
242240 if (res_size + last_size > mem_limit)
243241 log_stream << " Warning: resident size (" " ) exceeds memory limit."
244242
245243 timer.go (" Sorting trace points"
246244#ifdef NDEBUG
247- ips4o::parallel::sort (hit_buf->begin (), hit_buf->end (), std::less<Search::Hit>(), config.threads_ );
245+ // sort::sort_parallel_blocked_inplace(hit_buf, hit_buf + hit_count, std::less<Search::Hit>(), config.threads_);
246+ ips4o::parallel::sort (hit_buf, hit_buf + hit_count, std::less<Search::Hit>(), config.threads_ );
248247#else
249- std::sort (hit_buf-> begin () , hit_buf-> end () );
248+ std::sort (hit_buf, hit_buf + hit_count );
250249#endif
251250 statistics.inc (Statistics::TIME_SORT_SEED_HITS , timer.microseconds ());
252251
253252 timer.go (" Computing partition"
254- const vector<int64_t > partition = make_partition (hit_buf-> data () , hit_buf-> data () + hit_buf-> size () );
253+ const vector<int64_t > partition = make_partition (hit_buf, hit_buf + hit_count );
255254
256255 timer.go (" Computing alignments"
257- HitIterator hit_it (query_range.first , query_range.second , hit_buf-> data () , hit_buf-> data () + hit_buf-> size () , partition.begin (), (int64_t )partition.size () - 1 );
258- OutputWriter writer{output_file, cfg.output_format ->query_separator };
256+ HitIterator hit_it (query_range.first , query_range.second , hit_buf, hit_buf + hit_count , partition.begin (), (int64_t )partition.size () - 1 );
257+ OutputWriter writer{output_file, cfg.blocked_processing ? ' \0 ' : cfg. output_format ->query_separator };
259258 output_sink.reset (new ReorderQueue<TextBuffer*, OutputWriter>(query_range.first , writer));
260259 unique_ptr<thread> heartbeat;
261260 if (config.verbosity >= 3 && config.load_balancing == Config::query_parallel && !config.swipe_all && config.heartbeat )
@@ -273,12 +272,13 @@ void align_queries(Consumer* output_file, Search::Config& cfg)
273272 timer.go (" Deallocating buffers"
274273 cfg.thread_pool .reset ();
275274 output_sink.reset ();
276- last_size = hit_buf-> size () * sizeof (Search::Hit);
275+ last_size = hit_count * sizeof (Search::Hit);
277276 res_size -= last_size;
278- delete hit_buf;
279277 }
280278 statistics.max (Statistics::SEARCH_TEMP_SPACE , cfg.seed_hit_buf ->total_disk_size ());
281279
280+ timer.go (" Freeing memory"
281+ cfg.seed_hit_buf ->free_buffer ();
282282 if (!cfg.blocked_processing && !cfg.iterated ())
283283 cfg.db ->end_random_access (false );
284284}
0 commit comments