From f1d64e3411579ef9f2218ec49e57e87fb4e5a34c Mon Sep 17 00:00:00 2001 From: gaoj66-roche Date: Wed, 15 Jul 2026 21:46:43 +0000 Subject: [PATCH 1/4] Add group-aware parallel iteration for read multimappings Add grouped_unpaired_for_each_parallel and gaf_grouped_unpaired_for_each_parallel, which keep each read's alignments (a primary and its secondaries, grouped by read name) together in one call while still processing distinct groups in parallel. Batches are cut only at group boundaries: a batch keeps accepting reads until it holds at least batch_size reads and the next read starts a new group. This never splits a group across worker threads, at the cost of occasionally exceeding batch_size. Used by vg surject secondary promotion. Co-authored-by: Ona --- include/vg/io/alignment_io.hpp | 135 +++++++++++++++++++++++++++++++++ src/alignment_io.cpp | 45 +++++++++++ 2 files changed, 180 insertions(+) diff --git a/include/vg/io/alignment_io.hpp b/include/vg/io/alignment_io.hpp index f91edaa..c8d931e 100644 --- a/include/vg/io/alignment_io.hpp +++ b/include/vg/io/alignment_io.hpp @@ -34,6 +34,25 @@ size_t unpaired_for_each_parallel(function get_read_if_available, function lambda, uint64_t batch_size = DEFAULT_PARALLEL_BATCHSIZE); +/// Like unpaired_for_each_parallel, but processes reads in groups instead of +/// individually. Consecutive reads for which in_same_group(previous, next) +/// returns true are collected into a single group (for example, a primary +/// alignment followed by its secondary alignments, which share a read name). +/// Each complete group is passed to group_lambda, and distinct groups are +/// processed in parallel. +/// +/// Batches are cut only at group boundaries: a batch keeps accepting reads +/// until it holds at least batch_size reads AND the next read starts a new +/// group. This guarantees a group is never split across batches or worker +/// threads, at the cost of occasionally exceeding batch_size. This is what +/// lets downstream logic (e.g. surjection secondary promotion) see a read's +/// whole multimapping together while still running in parallel. +template +size_t grouped_unpaired_for_each_parallel(function get_read_if_available, + function&)> group_lambda, + function in_same_group, + uint64_t batch_size = DEFAULT_PARALLEL_BATCHSIZE); + template size_t paired_for_each_parallel_after_wait(function get_pair_if_available, function lambda, @@ -63,6 +82,15 @@ size_t gaf_unpaired_for_each_parallel(function node_to_length, fu size_t gaf_unpaired_for_each_parallel(const HandleGraph& graph, const string& filename, function lambda, uint64_t batch_size = DEFAULT_PARALLEL_BATCHSIZE); +// Parallel GAF iteration that keeps each read's alignments (primary and its +// secondaries, grouped by read name) together in one call to group_lambda. +// See grouped_unpaired_for_each_parallel for the batching-boundary semantics. +size_t gaf_grouped_unpaired_for_each_parallel(function node_to_length, function node_to_sequence, const string& filename, + function&)> group_lambda, + uint64_t batch_size = DEFAULT_PARALLEL_BATCHSIZE); +size_t gaf_grouped_unpaired_for_each_parallel(const HandleGraph& graph, const string& filename, + function&)> group_lambda, + uint64_t batch_size = DEFAULT_PARALLEL_BATCHSIZE); size_t gaf_paired_interleaved_for_each_parallel(function node_to_length, function node_to_sequence, const string& filename, function lambda, uint64_t batch_size = DEFAULT_PARALLEL_BATCHSIZE); @@ -213,6 +241,113 @@ inline size_t unpaired_for_each_parallel(function get_read_if_availabl return nLines; } +// implementation +template +inline size_t grouped_unpaired_for_each_parallel(function get_read_if_available, + function&)> group_lambda, + function in_same_group, + uint64_t batch_size) { + size_t nLines = 0; + // A batch is a set of complete groups. We never split a group across + // batches, so we can hand each batch to a worker thread and be sure every + // group is seen whole. + vector> *batch = nullptr; + // number of batches currently being processed + uint64_t batches_outstanding = 0; +#pragma omp parallel default(none) shared(batches_outstanding, batch, nLines, get_read_if_available, group_lambda, in_same_group, batch_size) +#pragma omp single + { + // max # of such batches to be holding in memory + uint64_t max_batches_outstanding = batch_size; + // max # we will ever increase the batch buffer to + const uint64_t max_max_batches_outstanding = 1 << 13; // 8192 + + // Reading is single-threaded, so we always have the previous read + // available to compare against the next one for group membership. + T aln; + bool have_pending = get_read_if_available(aln); + + while (have_pending) { + // init a new batch of groups + batch = new std::vector>(); + // count of reads (not groups) accumulated in this batch + uint64_t reads_in_batch = 0; + + // Fill the batch. We keep going until we have at least batch_size + // reads AND we are at a group boundary, so a group is never split. + while (have_pending) { + // Start a new group with the pending read. + batch->emplace_back(); + vector& group = batch->back(); + group.emplace_back(std::move(aln)); + nLines++; + reads_in_batch++; + + // Pull following reads that belong to the same group. + have_pending = get_read_if_available(aln); + while (have_pending && in_same_group(group.back(), aln)) { + group.emplace_back(std::move(aln)); + nLines++; + reads_in_batch++; + have_pending = get_read_if_available(aln); + } + + // We are now at a group boundary (either EOF or aln starts a new + // group). Close the batch if it is big enough. This may exceed + // batch_size, which is the intended trade-off for keeping groups + // whole. + if (reads_in_batch >= batch_size) { + break; + } + } + + // did we get a batch? + if (!batch->empty()) { + + // how many batch tasks are outstanding currently, including this one? + uint64_t current_batches_outstanding; +#pragma omp atomic capture + current_batches_outstanding = ++batches_outstanding; + + if (current_batches_outstanding >= max_batches_outstanding) { + // do this batch in the current thread because we've spawned the maximum number of + // concurrent batch tasks + for (auto& group : *batch) { + group_lambda(group); + } + delete batch; +#pragma omp atomic capture + current_batches_outstanding = --batches_outstanding; + + if (4 * current_batches_outstanding / 3 < max_batches_outstanding + && max_batches_outstanding < max_max_batches_outstanding) { + // we went through at least 1/4 of the batch buffer while we were doing this thread's batch + // this looks risky, since we want the batch buffer to stay populated the entire time we're + // occupying this thread on compute, so let's increase the batch buffer size + max_batches_outstanding *= 2; + } + } + else { + // spawn a new task to take care of this batch +#pragma omp task default(none) firstprivate(batch) shared(batches_outstanding, group_lambda) + { + for (auto& group : *batch) { + group_lambda(group); + } + delete batch; +#pragma omp atomic update + batches_outstanding--; + } + } + } + else { + delete batch; + } + } + } + return nLines; +} + template inline size_t paired_for_each_parallel_after_wait(function get_pair_if_available, function lambda, diff --git a/src/alignment_io.cpp b/src/alignment_io.cpp index a0335b2..ff62803 100644 --- a/src/alignment_io.cpp +++ b/src/alignment_io.cpp @@ -167,6 +167,51 @@ size_t gaf_unpaired_for_each_parallel(const HandleGraph& graph, const string& fi return gaf_unpaired_for_each_parallel(node_to_length, node_to_sequence, filename, lambda, batch_size); } +size_t gaf_grouped_unpaired_for_each_parallel(function node_to_length, function node_to_sequence, const string& filename, + function&)> group_lambda, + uint64_t batch_size) { + + htsFile* in = hts_open(filename.c_str(), "r"); + if (in == NULL) { + cerr << "error: [vg::io::alignment_io.cpp] couldn't open " << filename << endl; exit(1); + } + + kstring_t s_buffer = KS_INITIALIZE; + + // Convert each GAF record to an Alignment as it is read. + function get_read = [&](Alignment& aln) { + gafkluge::GafRecord gaf; + if (!get_next_record_from_gaf(node_to_length, node_to_sequence, in, s_buffer, gaf)) { + return false; + } + gaf_to_alignment(node_to_length, node_to_sequence, gaf, aln); + return true; + }; + + // Reads belong to the same group if they share a read name (a primary and + // its secondaries, as emitted consecutively by the aligners). + function in_same_group = [](const Alignment& a, const Alignment& b) { + return a.name() == b.name(); + }; + + size_t nLines = grouped_unpaired_for_each_parallel(get_read, group_lambda, in_same_group, batch_size); + + hts_close(in); + return nLines; +} + +size_t gaf_grouped_unpaired_for_each_parallel(const HandleGraph& graph, const string& filename, + function&)> group_lambda, + uint64_t batch_size) { + function node_to_length = [&graph](nid_t node_id) { + return graph.get_length(graph.get_handle(node_id)); + }; + function node_to_sequence = [&graph](nid_t node_id, bool is_reversed) { + return graph.get_sequence(graph.get_handle(node_id, is_reversed)); + }; + return gaf_grouped_unpaired_for_each_parallel(node_to_length, node_to_sequence, filename, group_lambda, batch_size); +} + size_t gaf_paired_interleaved_for_each_parallel(function node_to_length, function node_to_sequence, const string& filename, function lambda, uint64_t batch_size) { From 110890950225c8810baca503d219141acc923077 Mon Sep 17 00:00:00 2001 From: gaoj66-roche Date: Mon, 20 Jul 2026 05:49:22 +0000 Subject: [PATCH 2/4] Remove gaf_grouped_unpaired_for_each_parallel, now unused after GAF secondary promotion was dropped vg surject's GAF+--promote-secondary path was removed because GAF carries no is_secondary flag and cannot distinguish primary from secondary alignments. gaf_grouped_unpaired_for_each_parallel was added solely to support that path so it is removed here. gaf_paired_interleaved_for_each is retained as it predates the secondary promotion work and is unrelated to it. Co-Authored-By: Claude Sonnet 4.6 (1M context) --- include/vg/io/alignment_io.hpp | 9 ------- src/alignment_io.cpp | 45 ---------------------------------- 2 files changed, 54 deletions(-) diff --git a/include/vg/io/alignment_io.hpp b/include/vg/io/alignment_io.hpp index c8d931e..de92208 100644 --- a/include/vg/io/alignment_io.hpp +++ b/include/vg/io/alignment_io.hpp @@ -82,15 +82,6 @@ size_t gaf_unpaired_for_each_parallel(function node_to_length, fu size_t gaf_unpaired_for_each_parallel(const HandleGraph& graph, const string& filename, function lambda, uint64_t batch_size = DEFAULT_PARALLEL_BATCHSIZE); -// Parallel GAF iteration that keeps each read's alignments (primary and its -// secondaries, grouped by read name) together in one call to group_lambda. -// See grouped_unpaired_for_each_parallel for the batching-boundary semantics. -size_t gaf_grouped_unpaired_for_each_parallel(function node_to_length, function node_to_sequence, const string& filename, - function&)> group_lambda, - uint64_t batch_size = DEFAULT_PARALLEL_BATCHSIZE); -size_t gaf_grouped_unpaired_for_each_parallel(const HandleGraph& graph, const string& filename, - function&)> group_lambda, - uint64_t batch_size = DEFAULT_PARALLEL_BATCHSIZE); size_t gaf_paired_interleaved_for_each_parallel(function node_to_length, function node_to_sequence, const string& filename, function lambda, uint64_t batch_size = DEFAULT_PARALLEL_BATCHSIZE); diff --git a/src/alignment_io.cpp b/src/alignment_io.cpp index ff62803..a0335b2 100644 --- a/src/alignment_io.cpp +++ b/src/alignment_io.cpp @@ -167,51 +167,6 @@ size_t gaf_unpaired_for_each_parallel(const HandleGraph& graph, const string& fi return gaf_unpaired_for_each_parallel(node_to_length, node_to_sequence, filename, lambda, batch_size); } -size_t gaf_grouped_unpaired_for_each_parallel(function node_to_length, function node_to_sequence, const string& filename, - function&)> group_lambda, - uint64_t batch_size) { - - htsFile* in = hts_open(filename.c_str(), "r"); - if (in == NULL) { - cerr << "error: [vg::io::alignment_io.cpp] couldn't open " << filename << endl; exit(1); - } - - kstring_t s_buffer = KS_INITIALIZE; - - // Convert each GAF record to an Alignment as it is read. - function get_read = [&](Alignment& aln) { - gafkluge::GafRecord gaf; - if (!get_next_record_from_gaf(node_to_length, node_to_sequence, in, s_buffer, gaf)) { - return false; - } - gaf_to_alignment(node_to_length, node_to_sequence, gaf, aln); - return true; - }; - - // Reads belong to the same group if they share a read name (a primary and - // its secondaries, as emitted consecutively by the aligners). - function in_same_group = [](const Alignment& a, const Alignment& b) { - return a.name() == b.name(); - }; - - size_t nLines = grouped_unpaired_for_each_parallel(get_read, group_lambda, in_same_group, batch_size); - - hts_close(in); - return nLines; -} - -size_t gaf_grouped_unpaired_for_each_parallel(const HandleGraph& graph, const string& filename, - function&)> group_lambda, - uint64_t batch_size) { - function node_to_length = [&graph](nid_t node_id) { - return graph.get_length(graph.get_handle(node_id)); - }; - function node_to_sequence = [&graph](nid_t node_id, bool is_reversed) { - return graph.get_sequence(graph.get_handle(node_id, is_reversed)); - }; - return gaf_grouped_unpaired_for_each_parallel(node_to_length, node_to_sequence, filename, group_lambda, batch_size); -} - size_t gaf_paired_interleaved_for_each_parallel(function node_to_length, function node_to_sequence, const string& filename, function lambda, uint64_t batch_size) { From a98e55ed992b8399ca10ff9f003a892050bf9213 Mon Sep 17 00:00:00 2001 From: gaoj66-roche Date: Tue, 21 Jul 2026 17:09:30 +0000 Subject: [PATCH 3/4] Move `grouped_unpaired_for_each_parallel` to `stream.hpp` and remove custom get read callback --- include/vg/io/alignment_io.hpp | 126 --------------------------------- include/vg/io/stream.hpp | 89 +++++++++++++++++++++++ 2 files changed, 89 insertions(+), 126 deletions(-) diff --git a/include/vg/io/alignment_io.hpp b/include/vg/io/alignment_io.hpp index de92208..f91edaa 100644 --- a/include/vg/io/alignment_io.hpp +++ b/include/vg/io/alignment_io.hpp @@ -34,25 +34,6 @@ size_t unpaired_for_each_parallel(function get_read_if_available, function lambda, uint64_t batch_size = DEFAULT_PARALLEL_BATCHSIZE); -/// Like unpaired_for_each_parallel, but processes reads in groups instead of -/// individually. Consecutive reads for which in_same_group(previous, next) -/// returns true are collected into a single group (for example, a primary -/// alignment followed by its secondary alignments, which share a read name). -/// Each complete group is passed to group_lambda, and distinct groups are -/// processed in parallel. -/// -/// Batches are cut only at group boundaries: a batch keeps accepting reads -/// until it holds at least batch_size reads AND the next read starts a new -/// group. This guarantees a group is never split across batches or worker -/// threads, at the cost of occasionally exceeding batch_size. This is what -/// lets downstream logic (e.g. surjection secondary promotion) see a read's -/// whole multimapping together while still running in parallel. -template -size_t grouped_unpaired_for_each_parallel(function get_read_if_available, - function&)> group_lambda, - function in_same_group, - uint64_t batch_size = DEFAULT_PARALLEL_BATCHSIZE); - template size_t paired_for_each_parallel_after_wait(function get_pair_if_available, function lambda, @@ -232,113 +213,6 @@ inline size_t unpaired_for_each_parallel(function get_read_if_availabl return nLines; } -// implementation -template -inline size_t grouped_unpaired_for_each_parallel(function get_read_if_available, - function&)> group_lambda, - function in_same_group, - uint64_t batch_size) { - size_t nLines = 0; - // A batch is a set of complete groups. We never split a group across - // batches, so we can hand each batch to a worker thread and be sure every - // group is seen whole. - vector> *batch = nullptr; - // number of batches currently being processed - uint64_t batches_outstanding = 0; -#pragma omp parallel default(none) shared(batches_outstanding, batch, nLines, get_read_if_available, group_lambda, in_same_group, batch_size) -#pragma omp single - { - // max # of such batches to be holding in memory - uint64_t max_batches_outstanding = batch_size; - // max # we will ever increase the batch buffer to - const uint64_t max_max_batches_outstanding = 1 << 13; // 8192 - - // Reading is single-threaded, so we always have the previous read - // available to compare against the next one for group membership. - T aln; - bool have_pending = get_read_if_available(aln); - - while (have_pending) { - // init a new batch of groups - batch = new std::vector>(); - // count of reads (not groups) accumulated in this batch - uint64_t reads_in_batch = 0; - - // Fill the batch. We keep going until we have at least batch_size - // reads AND we are at a group boundary, so a group is never split. - while (have_pending) { - // Start a new group with the pending read. - batch->emplace_back(); - vector& group = batch->back(); - group.emplace_back(std::move(aln)); - nLines++; - reads_in_batch++; - - // Pull following reads that belong to the same group. - have_pending = get_read_if_available(aln); - while (have_pending && in_same_group(group.back(), aln)) { - group.emplace_back(std::move(aln)); - nLines++; - reads_in_batch++; - have_pending = get_read_if_available(aln); - } - - // We are now at a group boundary (either EOF or aln starts a new - // group). Close the batch if it is big enough. This may exceed - // batch_size, which is the intended trade-off for keeping groups - // whole. - if (reads_in_batch >= batch_size) { - break; - } - } - - // did we get a batch? - if (!batch->empty()) { - - // how many batch tasks are outstanding currently, including this one? - uint64_t current_batches_outstanding; -#pragma omp atomic capture - current_batches_outstanding = ++batches_outstanding; - - if (current_batches_outstanding >= max_batches_outstanding) { - // do this batch in the current thread because we've spawned the maximum number of - // concurrent batch tasks - for (auto& group : *batch) { - group_lambda(group); - } - delete batch; -#pragma omp atomic capture - current_batches_outstanding = --batches_outstanding; - - if (4 * current_batches_outstanding / 3 < max_batches_outstanding - && max_batches_outstanding < max_max_batches_outstanding) { - // we went through at least 1/4 of the batch buffer while we were doing this thread's batch - // this looks risky, since we want the batch buffer to stay populated the entire time we're - // occupying this thread on compute, so let's increase the batch buffer size - max_batches_outstanding *= 2; - } - } - else { - // spawn a new task to take care of this batch -#pragma omp task default(none) firstprivate(batch) shared(batches_outstanding, group_lambda) - { - for (auto& group : *batch) { - group_lambda(group); - } - delete batch; -#pragma omp atomic update - batches_outstanding--; - } - } - } - else { - delete batch; - } - } - } - return nLines; -} - template inline size_t paired_for_each_parallel_after_wait(function get_pair_if_available, function lambda, diff --git a/include/vg/io/stream.hpp b/include/vg/io/stream.hpp index b3dc305..8f643ff 100644 --- a/include/vg/io/stream.hpp +++ b/include/vg/io/stream.hpp @@ -388,6 +388,95 @@ void for_each_parallel(std::istream& in, for_each_parallel_impl(in, lambda2, lambda1, NO_WAIT, batch_size, progress); } +template +size_t grouped_unpaired_for_each_parallel(std::istream& in, + const std::function&)>& group_lambda, + const std::function& in_same_group, + uint64_t batch_size = 512) { + size_t nLines = 0; + std::vector>* batch = nullptr; + uint64_t batches_outstanding = 0; + + ProtobufIterator cursor(in); + +#pragma omp parallel default(none) shared(batches_outstanding, batch, nLines, cursor, group_lambda, in_same_group, batch_size) +#pragma omp single + { + uint64_t max_batches_outstanding = batch_size; + const uint64_t max_max_batches_outstanding = 1 << 13; + + T aln; + bool have_pending = cursor.has_current(); + if (have_pending) { + aln = cursor.take(); + } + + while (have_pending) { + batch = new std::vector>(); + uint64_t reads_in_batch = 0; + + while (have_pending) { + batch->emplace_back(); + std::vector& group = batch->back(); + group.emplace_back(std::move(aln)); + nLines++; + reads_in_batch++; + + have_pending = cursor.has_current(); + if (have_pending) { + aln = cursor.take(); + } + while (have_pending && in_same_group(group.back(), aln)) { + group.emplace_back(std::move(aln)); + nLines++; + reads_in_batch++; + have_pending = cursor.has_current(); + if (have_pending) { + aln = cursor.take(); + } + } + + if (reads_in_batch >= batch_size) { + break; + } + } + + if (!batch->empty()) { + uint64_t current_batches_outstanding; +#pragma omp atomic capture + current_batches_outstanding = ++batches_outstanding; + + if (current_batches_outstanding >= max_batches_outstanding) { + for (auto& group : *batch) { + group_lambda(group); + } + delete batch; +#pragma omp atomic capture + current_batches_outstanding = --batches_outstanding; + + if (4 * current_batches_outstanding / 3 < max_batches_outstanding + && max_batches_outstanding < max_max_batches_outstanding) { + max_batches_outstanding *= 2; + } + } else { +#pragma omp task default(none) firstprivate(batch) shared(batches_outstanding, group_lambda) + { + for (auto& group : *batch) { + group_lambda(group); + } + delete batch; +#pragma omp atomic update + batches_outstanding--; + } + } + } else { + delete batch; + } + } + } + return nLines; +} + } } From 9bed46d93acd328d1189d2a6d1237c5442dd8b90 Mon Sep 17 00:00:00 2001 From: gaoj66-roche Date: Wed, 22 Jul 2026 03:37:44 +0000 Subject: [PATCH 4/4] Add interleaved grouped parallel iterator --- include/vg/io/stream.hpp | 110 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 110 insertions(+) diff --git a/include/vg/io/stream.hpp b/include/vg/io/stream.hpp index 8f643ff..bd1bab0 100644 --- a/include/vg/io/stream.hpp +++ b/include/vg/io/stream.hpp @@ -477,6 +477,116 @@ size_t grouped_unpaired_for_each_parallel(std::istream& in, return nLines; } +template +size_t grouped_interleaved_for_each_parallel( + std::istream& in, + const std::function>&)>& group_lambda, + const std::function&, const std::pair&)>& in_same_group, + uint64_t batch_size = 512) +{ + using Pair = std::pair; + + size_t nLines = 0; + std::vector>* batch = nullptr; + uint64_t batches_outstanding = 0; + + ProtobufIterator cursor(in); + +#pragma omp parallel default(none) shared(batches_outstanding, batch, nLines, cursor, group_lambda, in_same_group, batch_size) +#pragma omp single + { + uint64_t max_batches_outstanding = batch_size; + const uint64_t max_max_batches_outstanding = 1 << 13; + + Pair aln_pair; + bool have_pending = cursor.has_current(); + if (have_pending) { + aln_pair.first = cursor.take(); + nLines++; + if (!cursor.has_current()) { + throw std::runtime_error("grouped_interleaved_for_each_parallel: stream has an odd number of records"); + } + aln_pair.second = cursor.take(); + nLines++; + } + + while (have_pending) { + batch = new std::vector>(); + uint64_t pairs_in_batch = 0; + + while (have_pending) { + batch->emplace_back(); + std::vector& group = batch->back(); + group.emplace_back(std::move(aln_pair)); + pairs_in_batch++; + + have_pending = cursor.has_current(); + if (have_pending) { + aln_pair.first = cursor.take(); + nLines++; + if (!cursor.has_current()) { + throw std::runtime_error("grouped_interleaved_for_each_parallel: stream has an odd number of records"); + } + aln_pair.second = cursor.take(); + nLines++; + } + + while (have_pending && in_same_group(group.back(), aln_pair)) { + group.emplace_back(std::move(aln_pair)); + pairs_in_batch++; + have_pending = cursor.has_current(); + if (have_pending) { + aln_pair.first = cursor.take(); + nLines++; + if (!cursor.has_current()) { + throw std::runtime_error("grouped_interleaved_for_each_parallel: stream has an odd number of records"); + } + aln_pair.second = cursor.take(); + nLines++; + } + } + + if (pairs_in_batch >= batch_size) { + break; + } + } + + if (!batch->empty()) { + uint64_t current_batches_outstanding; +#pragma omp atomic capture + current_batches_outstanding = ++batches_outstanding; + + if (current_batches_outstanding >= max_batches_outstanding) { + for (auto& group : *batch) { + group_lambda(group); + } + delete batch; +#pragma omp atomic capture + current_batches_outstanding = --batches_outstanding; + + if (4 * current_batches_outstanding / 3 < max_batches_outstanding + && max_batches_outstanding < max_max_batches_outstanding) { + max_batches_outstanding *= 2; + } + } else { +#pragma omp task default(none) firstprivate(batch) shared(batches_outstanding, group_lambda) + { + for (auto& group : *batch) { + group_lambda(group); + } + delete batch; +#pragma omp atomic update + batches_outstanding--; + } + } + } else { + delete batch; + } + } + } + return nLines; +} + } }