src/core/agc_compressor.cpp
// *******************************************************************************************
// Add sample files - reproducibility mode
Add sample files - reproducibility mode
Add sample files - reproducibility mode
bool CAGCCompressor::AddSampleFilesRep(vector<pair<string, string>> _v_sample_file_name, const uint32_t no_threads)
{
    if (_v_sample_file_name.empty())
        return true;
    reproducibility_mode = true;
    processed_bases = 0;
    size_t queue_capacity = max(2ull << 30, no_threads * (192ull << 20));
    pq_contigs_desc = make_shared<CBoundedPQueue<task_t>>(1, queue_capacity);
    pq_contigs_desc_aux = make_shared<CBoundedPQueue<task_t>>(1, ~0ull);
    pq_contigs_desc_working = pq_contigs_desc;
    uint32_t no_workers = (no_threads < 8) ? no_threads : no_threads - 1;
    vector<thread> v_threads;
    v_threads.reserve((size_t)no_workers);
    my_barrier bar(no_workers);
    start_compressing_rep_threads(v_threads, bar, no_workers);
    pair<string, string> sf;
    CGenomeIO gio;
    string id;
    contig_t contig;
    size_t sample_priority = ~0ull;
    size_t cnt_contigs_in_sample = 0;
    const size_t max_no_contigs_before_synchronization = 1000ull;
    const size_t min_size_before_synchronization = 1ull << 30;
    size_t total_size_from_synchronization = min_size_before_synchronization;       
    for(auto sf : _v_sample_file_name)
    {
        if (!gio.Open(sf.second, false))
        {
            cerr << "Cannot open file: " << sf.second << endl;
            continue;
        }
        while (gio.ReadContigRaw(id, contig))
        {
            if (concatenated_genomes)
            {
                if (cnt_contigs_in_sample++ >= max_no_contigs_before_synchronization)
                {
                    for (uint32_t i = 0; i < no_workers; ++i)
                        pq_contigs_desc->Emplace(make_tuple(contig_processing_stage_t::all_contigs, "", "", contig_t()), sample_priority, 0);
                    --sample_priority;
                    cnt_contigs_in_sample = 0;
                }
                if (collection_desc.register_sample_contig("", id))
                    pq_contigs_desc->Emplace(make_tuple(contig_processing_stage_t::all_contigs, "", id, move(contig)), sample_priority, contig.size());
                else
                    cerr << "Error: Pair sample_name:contig_name " << id << ":" << id << " is already in the archive!\n";
            }
            else
            {
                total_size_from_synchronization += contig.size();
                if (collection_desc.register_sample_contig(sf.first, id))
                    pq_contigs_desc->Emplace(make_tuple(contig_processing_stage_t::all_contigs, sf.first, id, move(contig)), sample_priority, contig.size());
                else
                    cerr << "Error: Pair sample_name:contig_name " << sf.first << ":" << id << " is already in the archive!\n";
            }
        }
        {
            total_size_from_synchronization = 0;
            for (uint32_t i = 0; i < no_workers; ++i)
                pq_contigs_desc->Emplace(make_tuple(
                    adaptive_compression ? contig_processing_stage_t::new_splitters : contig_processing_stage_t::registration, 
                    "", "", contig_t()), sample_priority, 0);
        }
        --sample_priority;
        gio.Close();
    }
/*    if (total_size_from_synchronization > 0)
    {
        for (uint32_t i = 0; i < no_workers; ++i)
            pq_contigs_desc->Emplace(make_tuple(
                adaptive_compression ? contig_processing_stage_t::new_splitters : contig_processing_stage_t::registration,
                "", "", contig_t()), sample_priority, 0);
    }*/
    pq_contigs_desc->MarkCompleted();
    join_threads(v_threads);
    out_archive->FlushOutBuffers();
    pq_contigs_desc.reset();
    pq_contigs_desc_aux.reset();
    pq_contigs_desc_working.reset();
    no_samples_in_archive += _v_sample_file_name.size();
    return true;
}
bool CAGCCompressor :: AddSampleFilesRep ( vector < pair < string , string > > _v_sample_file_name , const uint32_t no_threads ) { if ( _v_sample_file_name . empty ( ) ) return true ; reproducibility_mode = true ; processed_bases = 0 ; size_t queue_capacity = max ( 2ull << 30 , no_threads * ( 192ull << 20 ) ) ; pq_contigs_desc = make_shared < CBoundedPQueue < task_t > > ( 1 , queue_capacity ) ; pq_contigs_desc_aux = make_shared < CBoundedPQueue < task_t > > ( 1 , ~ 0ull ) ; pq_contigs_desc_working = pq_contigs_desc ; uint32_t no_workers = ( no_threads < 8 ) ? no_threads : no_threads - 1 ; vector < thread > v_threads ; v_threads . reserve ( ( size_t ) no_workers ) ; my_barrier bar ( no_workers ) ; start_compressing_rep_threads ( v_threads , bar , no_workers ) ; pair < string , string > sf ; CGenomeIO gio ; string id ; contig_t contig ; size_t sample_priority = ~ 0ull ; size_t cnt_contigs_in_sample = 0 ; const size_t max_no_contigs_before_synchronization = 1000ull ; const size_t min_size_before_synchronization = 1ull << 30 ; size_t total_size_from_synchronization = min_size_before_synchronization ; for ( auto sf : _v_sample_file_name ) { if ( ! gio . Open ( sf . second , false ) ) { cerr << " " << sf . second << endl ; continue ; } while ( gio . ReadContigRaw ( id , contig ) ) { if ( concatenated_genomes ) { if ( cnt_contigs_in_sample ++ >= max_no_contigs_before_synchronization ) { for ( uint32_t i = 0 ; i < no_workers ; ++ i ) pq_contigs_desc -> Emplace ( make_tuple ( contig_processing_stage_t :: all_contigs , " " , " " , contig_t ( ) ) , sample_priority , 0 ) ; -- sample_priority ; cnt_contigs_in_sample = 0 ; } if ( collection_desc . register_sample_contig ( " " , id ) ) pq_contigs_desc -> Emplace ( make_tuple ( contig_processing_stage_t :: all_contigs , " " , id , move ( contig ) ) , sample_priority , contig . size ( ) ) ; else cerr << " " << id << " " << id << " \n " ; } else { total_size_from_synchronization += contig . size ( ) ; if ( collection_desc . register_sample_contig ( sf . first , id ) ) pq_contigs_desc -> Emplace ( make_tuple ( contig_processing_stage_t :: all_contigs , sf . first , id , move ( contig ) ) , sample_priority , contig . size ( ) ) ; else cerr << " " << sf . first << " " << id << " \n " ; } } { total_size_from_synchronization = 0 ; for ( uint32_t i = 0 ; i < no_workers ; ++ i ) pq_contigs_desc -> Emplace ( make_tuple ( adaptive_compression ? contig_processing_stage_t :: new_splitters : contig_processing_stage_t :: registration , " " , " " , contig_t ( ) ) , sample_priority , 0 ) ; } -- sample_priority ; gio . Close ( ) ; } pq_contigs_desc -> MarkCompleted ( ) ; join_threads ( v_threads ) ; out_archive -> FlushOutBuffers ( ) ; pq_contigs_desc . reset ( ) ; pq_contigs_desc_aux . reset ( ) ; pq_contigs_desc_working . reset ( ) ; no_samples_in_archive += _v_sample_file_name . size ( ) ; return true ; }
Add sample files - reproducibility mode
Add sample files - reproducibility mode
// Reading Input //    const size_t min_size_before_synchronization = 0; // To ensure that synchronization will be after first genome // Send synchronization tokens //        if (concatenated_genomes || total_size_from_synchronization >= min_size_before_synchronization) // Send synchronization tokens /*    if (total_size_from_synchronization > 0)
    {
        // Send synchronization tokens
        for (uint32_t i = 0; i < no_workers; ++i)
            pq_contigs_desc->Emplace(make_tuple(
                adaptive_compression ? contig_processing_stage_t::new_splitters : contig_processing_stage_t::registration,
                "", "", contig_t()), sample_priority, 0);
    }*/