[DO NOT MERGE] hpcSPAdes rebase, cleanup & implementation

src/CMakeListsInternal.txt

@@ -11,9 +11,11 @@ if (SPADES_BUILD_INTERNAL)
   add_subdirectory(test/debruijn)
   add_subdirectory(test/examples)
   add_subdirectory(test/adt)
+  add_subdirectory(test/mpi)
 else()
   add_subdirectory(test/include_test EXCLUDE_FROM_ALL)
   add_subdirectory(test/debruijn EXCLUDE_FROM_ALL)
+  add_subdirectory(test/mpi EXCLUDE_FROM_ALL)
   add_subdirectory(test/adt EXCLUDE_FROM_ALL)
   add_subdirectory(test/examples EXCLUDE_FROM_ALL)
 endif()

src/cmake/includes.cmake

@@ -9,6 +9,7 @@ include_directories(SYSTEM "${Boost_INCLUDE_DIRS}")
 if (SPADES_USE_TCMALLOC)
   include_directories("${GOOGLE_PERFTOOLS_INCLUDE_DIR}")
 endif()
+
 if (SPADES_USE_JEMALLOC)
   include_directories("$<TARGET_FILE_DIR:jemalloc-static>/../include")
 endif()

src/cmake/proj.cmake

@@ -8,7 +8,7 @@
 
 # Side-by-side subprojects layout: automatically set the
 # SPADES_EXTERNAL_${project}_SOURCE_DIR using SPADES_ALL_PROJECTS
-set(SPADES_ALL_PROJECTS "spades;hammer;ionhammer;corrector;spaligner;spades_tools;binspreader;pathracer")
+set(SPADES_ALL_PROJECTS "spades;hammer;ionhammer;corrector;spaligner;spades_tools;binspreader;pathracer;hpcspades")
 set(SPADES_EXTRA_PROJECTS "mts;online_vis;cds_subgraphs")
 set(SPADES_KNOWN_PROJECTS "${SPADES_ALL_PROJECTS};${SPADES_EXTRA_PROJECTS}")
 set(SPADES_ENABLE_PROJECTS "" CACHE STRING

src/common/alignment/long_read_mapper.hpp

@@ -55,6 +55,20 @@ class LongReadMapper: public SequenceMapperListener {
         return g_;
     }
 
+    void Serialize(std::ostream &os) const override {
+        storage_.BinWrite(os);
+    }
+
+    void Deserialize(std::istream &is) override {
+        storage_.BinRead(is);
+    }
+
+    void MergeFromStream(std::istream &is) override {
+        PathStorage<Graph> remote(g_);
+        remote.BinRead(is);
+        storage_.AddStorage(remote);
+    }
+
 private:
 
     void ProcessSingleRead(size_t thread_index, const omnigraph::MappingPath<EdgeId>& mapping, const io::SingleRead& r);

src/common/alignment/rna/ss_coverage_filler.hpp

@@ -13,6 +13,8 @@
 #include "alignment/sequence_mapper_notifier.hpp"
 #include "assembly_graph/paths/mapping_path.hpp"
 
+#include "io/binary/binary.hpp"
+
 namespace debruijn_graph {
 
 class SSCoverageFiller: public SequenceMapperListener {
@@ -65,6 +67,22 @@ class SSCoverageFiller: public SequenceMapperListener {
             storage_.IncreaseKmerCount(it.first, size_t(it.second));
         tmp_storages_[thread_index].Clear();
     }
+
+    void Serialize(std::ostream &os) const override {
+        io::binary::BinWrite(os, storage_);
+    }
+
+    void Deserialize(std::istream &is) override {
+        io::binary::BinRead(is, storage_);
+    }
+
+    void MergeFromStream(std::istream &is) override {
+        SSCoverageStorage remote(g_);
+        io::binary::BinRead(is, remote);
+        for (const auto& it : remote) {
+            storage_.IncreaseKmerCount(it.first, size_t(it.second));
+        }
+    }
 };
 
 

src/common/alignment/sequence_mapper_notifier.cpp

@@ -13,9 +13,8 @@
 #include "io/reads/read_stream_vector.hpp"
 
 namespace debruijn_graph {
-
 SequenceMapperNotifier::SequenceMapperNotifier(size_t lib_count)
-    : listeners_(lib_count) 
+    : listeners_(lib_count)
 {}
 
 void SequenceMapperNotifier::Subscribe(SequenceMapperListener* listener, size_t lib_index) {

src/common/alignment/sequence_mapper_notifier.hpp

@@ -16,6 +16,7 @@
 #include "io/reads/paired_read.hpp"
 #include "io/reads/read_stream_vector.hpp"
 #include "utils/perf/timetracer.hpp"
+#include "utils/stl_utils.hpp"
 
 #include <string>
 #include <vector>
@@ -36,7 +37,23 @@ class SequenceMapperListener {
     virtual void ProcessSingleRead(size_t /* thread_index */, const io::SingleReadSeq& /* r */, const omnigraph::MappingPath<EdgeId>& /* read */) {}
 
     virtual void MergeBuffer(size_t /* thread_index */) {}
-
+
+    virtual void Serialize(std::ostream&) const {
+        VERIFY_MSG(false, "Serialize() is not implemented");
+    }
+
+    virtual void Deserialize(std::istream&) {
+        VERIFY_MSG(false, "Deserialize() is not implemented");
+    }
+
+    virtual void MergeFromStream(std::istream&) {
+        VERIFY_MSG(false, "MergeFromStream() is not implemented");
+    }
+
+    virtual const std::string name() const {
+        return utils::type_name(typeid(*this).name());
+    }
+
     virtual ~SequenceMapperListener() {}
 };
 
@@ -56,8 +73,7 @@ class SequenceMapperNotifier {
                         const SequenceMapperT& mapper, size_t threads_count = 0) {
         return ProcessLibrary(streams, 0, mapper, threads_count);
     }
-
-  private:
+
     template<class ReadType>
     void ProcessLibrary(io::ReadStreamList<ReadType>& streams,
                         size_t lib_index, const SequenceMapperT& mapper, size_t threads_count = 0) {
@@ -68,7 +84,7 @@ class SequenceMapperNotifier {
             threads_count = streams.size();
 
         streams.reset();
-        NotifyStartProcessLibrary(lib_index, threads_count);
+        NotifyStartProcessLibrary(lib_index, streams.size());
         size_t counter = 0, n = 15;
 
         #pragma omp parallel for num_threads(threads_count) shared(counter)
@@ -97,9 +113,10 @@ class SequenceMapperNotifier {
             counter += size;
         }
 
-        for (size_t i = 0; i < threads_count; ++i)
+        for (size_t i = 0; i < streams.size(); ++i)
             NotifyMergeBuffer(lib_index, i);
 
+        streams.close();
         INFO("Total " << counter << " reads processed");
         NotifyStopProcessLibrary(lib_index);
     }
@@ -114,7 +131,47 @@ class SequenceMapperNotifier {
 
     void NotifyMergeBuffer(size_t ilib, size_t ithread) const;
 
-    std::vector<std::vector<SequenceMapperListener*> > listeners_;  //first vector's size = count libs
+protected:
+    std::vector<ListenersContainer> listeners_;  //first vector's size = count libs
+};
+
+class MapLibBase {
+public:
+    virtual void operator() (const std::vector<SequenceMapperListener*>& listeners, const SequenceMapper<Graph>& mapper, io::ReadStreamList<io::PairedRead>& streams) const = 0;
+    virtual void operator() (const std::vector<SequenceMapperListener*>& listeners, const SequenceMapper<Graph>& mapper, io::ReadStreamList<io::SingleRead>& streams) const = 0;
+    virtual void operator() (const std::vector<SequenceMapperListener*>& listeners, const SequenceMapper<Graph>& mapper, io::ReadStreamList<io::SingleReadSeq>& streams) const = 0;
+    virtual void operator() (const std::vector<SequenceMapperListener*>& listeners, const SequenceMapper<Graph>& mapper, io::ReadStreamList<io::PairedReadSeq>& streams) const = 0;
+
+    template<class Streams>
+    void operator() (SequenceMapperListener* listener, const SequenceMapper<Graph>& mapper, Streams& streams) const {
+        this->operator() (std::vector<SequenceMapperListener*>(1, listener), mapper, streams);
+    }
+};
+
+class MapLibFunc : public MapLibBase {
+public:
+    void operator() (const std::vector<SequenceMapperListener*>& listeners, const SequenceMapper<Graph>& mapper, io::ReadStreamList<io::PairedRead>& streams) const override {
+        MapLib(listeners, mapper, streams);
+    }
+    void operator() (const std::vector<SequenceMapperListener*>& listeners, const SequenceMapper<Graph>& mapper, io::ReadStreamList<io::SingleRead>& streams) const override {
+        MapLib(listeners, mapper, streams);
+    }
+    void operator() (const std::vector<SequenceMapperListener*>& listeners, const SequenceMapper<Graph>& mapper, io::ReadStreamList<io::SingleReadSeq>& streams) const override {
+        MapLib(listeners, mapper, streams);
+    }
+    void operator() (const std::vector<SequenceMapperListener*>& listeners, const SequenceMapper<Graph>& mapper, io::ReadStreamList<io::PairedReadSeq>& streams) const override {
+        MapLib(listeners, mapper, streams);
+    }
+
+private:
+    template<class ReadType>
+    void MapLib(const std::vector<SequenceMapperListener*>& listeners, const SequenceMapper<Graph>& mapper, io::ReadStreamList<ReadType>& streams) const {
+        SequenceMapperNotifier notifier;
+        for (auto listener: listeners) {
+            notifier.Subscribe(listener);
+        }
+        notifier.ProcessLibrary(streams, mapper);
+    }
 };
 
 } // namespace debruijn_graph