src/core/agc_compressor.cpp
// *******************************************************************************************
// Start compressing threads - reproducibility mode 
Start compressing threads - reproducibility mode
Start compressing threads - reproducibility mode
void CAGCCompressor::start_compressing_rep_threads(vector<thread>& v_threads, my_barrier &bar, const uint32_t n_t)
{
    v_threads.clear();
    for (uint32_t i = 0; i < n_t; ++i)
    {
        v_threads.emplace_back([&, i, n_t]() {
            auto zstd_cctx = ZSTD_createCCtx();
            auto zstd_dctx = ZSTD_createDCtx();
            uint32_t thread_id = i;
            while (!pq_contigs_desc_working->IsCompleted())
            {
                task_t task;
                if (!pq_contigs_desc_working->PopLarge(task))
                    continue;
                if (get<0>(task) == contig_processing_stage_t::registration)
                {
                    bar.arrive_and_wait();
                    if (thread_id == 0)
                        register_segments();
                    bar.arrive_and_wait();
                    store_segments(zstd_cctx, zstd_dctx);
                    bar.arrive_and_wait();
                    if (thread_id == 0)
                    {
                        buffered_seg_part.clear();
                        if(n_t == 1)
                            out_archive->FlushOutBuffers();
                            pq_contigs_desc_working = pq_contigs_desc;
                    }
                    else if(thread_id == 1)
                        out_archive->FlushOutBuffers();
                    bar.arrive_and_wait();
                    continue;
                }
                if (get<0>(task) == contig_processing_stage_t::new_splitters)
                {
                    bar.arrive_and_wait();
                    if (thread_id == 0)
                    {
                        for(auto &v : vv_splitters)
                        { 
                            for (auto& x : v)
                                hs_splitters.insert_fast(x);
                            v.clear();
                        }
                        for (auto& x : v_raw_contigs)
                            pq_contigs_desc_aux->Emplace(make_tuple(contig_processing_stage_t::hard_contigs, get<0>(x), get<1>(x), move(get<2>(x))), 1, get<2>(x).size());
                        v_raw_contigs.clear();
                        for(uint32_t i = 0; i < n_t; ++i)
                            pq_contigs_desc_aux->Emplace(make_tuple(contig_processing_stage_t::registration, "", "", contig_t()), 0, 0);
                        pq_contigs_desc_working = pq_contigs_desc_aux;
                    }
                    bar.arrive_and_wait();
                    continue;
                }
                if(get<0>(task) == contig_processing_stage_t::all_contigs)
                    preprocess_raw_contig(get<3>(task));
                size_t ctg_size = get<3>(task).size();
                if (compress_contig_rep(get<0>(task), get<1>(task), get<2>(task), get<3>(task), zstd_cctx, zstd_dctx, thread_id))
                {
                    processed_bases += ctg_size;
                    if (verbosity > 0 && is_app_mode)
                        cerr << "Compressed: " + to_string(processed_bases / 1'000'000) + " Mb\r";
                    fflush(stdout);
                }
                else
                {
                    lock_guard<mutex> lck(mtx_raw_contigs);
                    v_raw_contigs.emplace_back(get<1>(task), get<2>(task), move(get<3>(task)));
                }
                get<3>(task).clear();
                get<3>(task).shrink_to_fit();
            }
            ZSTD_freeCCtx(zstd_cctx);
            ZSTD_freeDCtx(zstd_dctx);
            });
    }
}
void CAGCCompressor :: start_compressing_rep_threads ( vector < thread > & v_threads , my_barrier & bar , const uint32_t n_t ) { v_threads . clear ( ) ; for ( uint32_t i = 0 ; i < n_t ; ++ i ) { v_threads . emplace_back ( [ & , i , n_t ] ( ) { auto zstd_cctx = ZSTD_createCCtx ( ) ; auto zstd_dctx = ZSTD_createDCtx ( ) ; uint32_t thread_id = i ; while ( ! pq_contigs_desc_working -> IsCompleted ( ) ) { task_t task ; if ( ! pq_contigs_desc_working -> PopLarge ( task ) ) continue ; if ( get < 0 > ( task ) == contig_processing_stage_t :: registration ) { bar . arrive_and_wait ( ) ; if ( thread_id == 0 ) register_segments ( ) ; bar . arrive_and_wait ( ) ; store_segments ( zstd_cctx , zstd_dctx ) ; bar . arrive_and_wait ( ) ; if ( thread_id == 0 ) { buffered_seg_part . clear ( ) ; if ( n_t == 1 ) out_archive -> FlushOutBuffers ( ) ; pq_contigs_desc_working = pq_contigs_desc ; } else if ( thread_id == 1 ) out_archive -> FlushOutBuffers ( ) ; bar . arrive_and_wait ( ) ; continue ; } if ( get < 0 > ( task ) == contig_processing_stage_t :: new_splitters ) { bar . arrive_and_wait ( ) ; if ( thread_id == 0 ) { for ( auto & v : vv_splitters ) { for ( auto & x : v ) hs_splitters . insert_fast ( x ) ; v . clear ( ) ; } for ( auto & x : v_raw_contigs ) pq_contigs_desc_aux -> Emplace ( make_tuple ( contig_processing_stage_t :: hard_contigs , get < 0 > ( x ) , get < 1 > ( x ) , move ( get < 2 > ( x ) ) ) , 1 , get < 2 > ( x ) . size ( ) ) ; v_raw_contigs . clear ( ) ; for ( uint32_t i = 0 ; i < n_t ; ++ i ) pq_contigs_desc_aux -> Emplace ( make_tuple ( contig_processing_stage_t :: registration , " " , " " , contig_t ( ) ) , 0 , 0 ) ; pq_contigs_desc_working = pq_contigs_desc_aux ; } bar . arrive_and_wait ( ) ; continue ; } if ( get < 0 > ( task ) == contig_processing_stage_t :: all_contigs ) preprocess_raw_contig ( get < 3 > ( task ) ) ; size_t ctg_size = get < 3 > ( task ) . size ( ) ; if ( compress_contig_rep ( get < 0 > ( task ) , get < 1 > ( task ) , get < 2 > ( task ) , get < 3 > ( task ) , zstd_cctx , zstd_dctx , thread_id ) ) { processed_bases += ctg_size ; if ( verbosity > 0 && is_app_mode ) cerr << " " + to_string ( processed_bases / 1'000'000 ) + " \r " ; fflush ( stdout ) ; } else { lock_guard < mutex > lck ( mtx_raw_contigs ) ; v_raw_contigs . emplace_back ( get < 1 > ( task ) , get < 2 > ( task ) , move ( get < 3 > ( task ) ) ) ; } get < 3 > ( task ) . clear ( ) ; get < 3 > ( task ) . shrink_to_fit ( ) ; } ZSTD_freeCCtx ( zstd_cctx ) ; ZSTD_freeDCtx ( zstd_dctx ) ; } ) ; } }
Start compressing threads - reproducibility mode
Start compressing threads - reproducibility mode
// synchronization token received //                        if(adaptive_compression) // Add new splitters