Use SCQ

Author: eseiler

include/fpgalign/contrib/slotted_cart_queue.hpp

@@ -20,7 +20,7 @@ struct hit
 };
 
 void search(config const & config);
-std::vector<hit> ibf(config const & config);
-void fmindex(config const & config, std::vector<hit> hits);
+std::vector<hit> ibf(config const & config, size_t & todo_bin_count);
+void fmindex(config const & config, std::vector<hit> hits, size_t const todo_bin_count);
 
 } // namespace search

include/fpgalign/search/search.hpp

@@ -2,8 +2,12 @@
 // SPDX-FileCopyrightText: 2016-2025 Knut Reinert & MPI für molekulare Genetik
 // SPDX-License-Identifier: BSD-3-Clause
 
+#include <fmt/format.h>
+
 #include <seqan3/io/sequence_file/input.hpp>
 
+#include <hibf/contrib/std/enumerate_view.hpp>
+
 #include <fmindex-collection/fmindex/BiFMIndex.h>
 
 #include <fpgalign/build/build.hpp>
@@ -16,44 +20,41 @@ struct dna4_traits : seqan3::sequence_file_input_default_traits_dna
     using sequence_alphabet = seqan3::dna4;
 };
 
-void fmindex(config const & config)
+std::vector<std::vector<uint8_t>> read_reference(std::vector<std::string> const & bin_paths)
 {
-    std::vector<std::vector<uint8_t>> const reference = [&config]()
-    {
-        using sequence_file_t = seqan3::sequence_file_input<dna4_traits, seqan3::fields<seqan3::field::seq>>;
+    std::vector<std::vector<uint8_t>> reference{};
 
-        auto const bin_pathss = parse_input(config);
-
-        std::vector<std::vector<uint8_t>> result{};
-
-        for (auto && bin_paths : bin_pathss)
+    for (auto const & bin_path : bin_paths)
+    {
+        seqan3::sequence_file_input<dna4_traits, seqan3::fields<seqan3::field::seq>> fin{bin_path};
+        for (auto && record : fin)
         {
-            for (auto && bin_path : bin_paths)
-            {
-                sequence_file_t fin{bin_path};
-                for (auto && record : fin)
-                {
-                    result.push_back({});
-                    std::ranges::copy(record.sequence()
-                                          | std::views::transform(
-                                              [](auto const & in)
-                                              {
-                                                  return seqan3::to_rank(in);
-                                              }),
-                                      std::back_inserter(result.back()));
-                }
-            }
+            reference.push_back({});
+            std::ranges::copy(record.sequence()
+                                  | std::views::transform(
+                                      [](auto const & in)
+                                      {
+                                          return seqan3::to_rank(in);
+                                      }),
+                              std::back_inserter(reference.back()));
         }
+    }
 
-        return result;
-    }();
-
-    fmc::BiFMIndex<4> index{reference, /*samplingRate*/ 16, config.threads};
+    return reference;
+}
 
+void fmindex(config const & config)
+{
+    for (auto && [id, bin_paths] : seqan::stl::views::enumerate(parse_input(config)))
     {
-        std::ofstream os{config.output_path.string() + ".fmindex", std::ios::binary};
-        cereal::BinaryOutputArchive oarchive{os};
-        oarchive(index);
+        auto reference = read_reference(bin_paths);
+        fmc::BiFMIndex<4> index{reference, /*samplingRate*/ 16, config.threads};
+
+        {
+            std::ofstream os{fmt::format("{}.{}.fmindex", config.output_path.c_str(), id), std::ios::binary};
+            cereal::BinaryOutputArchive oarchive{os};
+            oarchive(index);
+        }
     }
 }
 

src/build/fmindex.cpp

@@ -2,45 +2,90 @@
 // SPDX-FileCopyrightText: 2016-2025 Knut Reinert & MPI für molekulare Genetik
 // SPDX-License-Identifier: BSD-3-Clause
 
+#include <fmt/format.h>
+
 #include <seqan3/io/sequence_file/input.hpp>
 
+#include <hibf/contrib/std/enumerate_view.hpp>
+
 #include <fmindex-collection/fmindex/BiFMIndex.h>
 #include <fmindex-collection/search/search.h>
 
+#include <fpgalign/contrib/slotted_cart_queue.hpp>
 #include <fpgalign/search/search.hpp>
 
 namespace search
 {
 
-void fmindex(config const & config, std::vector<hit> hits)
+fmc::BiFMIndex<4> load_index(config const & config, size_t const id)
 {
     fmc::BiFMIndex<4> index{};
 
     {
-        std::ifstream os{config.input_path.string() + ".fmindex", std::ios::binary};
+        std::ifstream os{fmt::format("{}.{}.fmindex", config.input_path.c_str(), id), std::ios::binary};
         cereal::BinaryInputArchive iarchive{os};
         iarchive(index);
     }
 
-#pragma omp parallel for num_threads(config.threads)
-    for (size_t i = 0; i < hits.size(); ++i)
+    return index;
+}
+
+void fmindex(config const & config, std::vector<hit> hits, size_t const todo_bin_count)
+{
+    // todo bin count
+    // todo capacity
+    // each slot = 1 bin
+    // a cart is full if it has 5 elements (hits)
+    scq::slotted_cart_queue<size_t> queue{{.slots = todo_bin_count, .carts = todo_bin_count, .capacity = 5}};
+    size_t thread_id{};
+
+    auto get_thread = [&]()
     {
-        auto & [id, seq, bins] = hits[i];
-        auto callback = [&](auto cursor, size_t)
-        {
-            for (auto j : cursor)
+        return std::jthread(
+            [&, thread_id = thread_id++]()
             {
-                auto [entry, offset] = index.locate(j);
-                auto [seqId, pos] = entry;
-#pragma omp critical
+                while (true)
                 {
-                    std::cout << '[' << id << "] found hit in seqNo " << seqId << " Pos " << pos + offset << '\n';
+                    scq::cart_future<size_t> cart = queue.dequeue();
+                    if (!cart.valid())
+                        return;
+                    auto [slot, span] = cart.get();
+                    auto index = load_index(config, slot.value);
+                    for (auto idx : span)
+                    {
+                        auto & [id, seq, bins] = hits[idx];
+
+                        auto callback = [&](auto cursor, size_t)
+                        {
+                            for (auto j : cursor)
+                            {
+                                auto [entry, offset] = index.locate(j);
+                                auto [seqId, pos] = entry;
+                                {
+                                    fmt::print("[{}][{}] found hit in bin {} in seqNo {} at Pos {}\n",
+                                               thread_id,
+                                               id,
+                                               slot.value,
+                                               seqId,
+                                               pos + offset);
+                                }
+                            }
+                        };
+
+                        fmc::search<true>(index, seq, config.errors, callback);
+                    }
                 }
-            }
-        };
-        fmc::search<true>(index, seq, config.errors, callback);
-        (void)bins;
-    }
+            });
+    };
+
+    std::vector<std::jthread> worker(config.threads);
+    std::ranges::generate(worker, get_thread);
+
+    for (auto && [idx, hit] : seqan::stl::views::enumerate(hits))
+        for (auto bin : hit.bins)
+            queue.enqueue(scq::slot_id{bin}, idx);
+
+    queue.close();
 }
 
 } // namespace search

src/search/fmindex.cpp

@@ -44,7 +44,7 @@ threshold::threshold get_thresholder(config const & config)
 using seqfile_t = seqan3::sequence_file_input<dna4_traits, seqan3::fields<seqan3::field::id, seqan3::field::seq>>;
 using record_t = typename seqfile_t::record_type;
 
-std::vector<hit> ibf(config const & config)
+std::vector<hit> ibf(config const & config, size_t & todo_bin_count)
 {
     seqan::hibf::interleaved_bloom_filter ibf{};
 
@@ -53,6 +53,7 @@ std::vector<hit> ibf(config const & config)
         cereal::BinaryInputArchive iarchive{os};
         iarchive(ibf);
     }
+    todo_bin_count = ibf.bin_count();
 
     std::vector<record_t> records = [&]()
     {

src/search/ibf.cpp

@@ -9,8 +9,9 @@ namespace search
 
 void search(config const & config)
 {
-    std::vector<hit> hits = ibf(config);
-    fmindex(config, std::move(hits));
+    size_t todo_bin_count{};
+    std::vector<hit> hits = ibf(config, todo_bin_count);
+    fmindex(config, std::move(hits), todo_bin_count);
 }
 
 } // namespace search

src/search/search.cpp

@@ -11,7 +11,6 @@ include (test/config)
 # This includes `test/data/datasources.cmake`, which makes test data available to the tests.
 include (data/datasources.cmake)
 
-add_app_test (api_fastq_coversion_test.cpp)
-add_app_test (cli_fastq_coversion_test.cpp)
+add_app_test (fpgalign_test.cpp)
 
 message (STATUS "You can run `make check` to build and run tests.")

test/CMakeLists.txt

@@ -0,0 +1,12 @@
+// SPDX-FileCopyrightText: 2006-2025 Knut Reinert & Freie Universität Berlin
+// SPDX-FileCopyrightText: 2016-2025 Knut Reinert & MPI für molekulare Genetik
+// SPDX-License-Identifier: BSD-3-Clause
+
+#include "app_test.hpp"
+
+// To prevent issues when running multiple CLI tests in parallel, give each CLI test unique names:
+struct fpgalign : public app_test
+{};
+
+TEST_F(fpgalign, no_options)
+{}