Multi-node linclust

README.md

@@ -150,6 +150,78 @@ To search with multiple servers, call the `search` or `cluster` workflow with th
 
     RUNNER="mpirun -pernode -np 42" mmseqs search queryDB targetDB resultDB tmp
 
+#### Multi-node `linclust`/`easy-linclust` on Slurm
+
+The default (v2) `linclust`/`easy-linclust` pipeline distributes its most expensive stage,
+`kmermatcher` and the candidate-alignment stage `align2clust`, across MPI ranks while
+keeping cluster assignment (representative selection and membership) byte-identical to a
+single-node run. This is the same `--mpi-runner`/`RUNNER` mechanism described above --
+there is no separate Slurm submission layer, so `linclust` should be launched as a single
+process (not itself wrapped in `mpirun`/`srun`) with the runner passed through
+`--mpi-runner`/`RUNNER`; the workflow script wraps only its MPI-aware sub-stages
+(`kmermatcher`, `align2clust`, and, for Linclust v1, `rescorediagonal`/`align`) in that
+runner internally. Wrapping the top-level `linclust` invocation in `mpirun`/`srun` yourself
+will fail ("recursive mpirun") or, on some launchers, silently run duplicate,
+non-cooperating single-rank jobs.
+
+Requirements and recommended topology:
+
+* Build with `-DHAVE_MPI=1` (version string gets a `-MPI` suffix). A non-MPI binary
+  launched under a multi-task runner (e.g. `srun -n 4 mmseqs ...` without `-DHAVE_MPI=1`)
+  fails fast with an explicit error instead of silently letting every task race on the
+  same output files.
+* The input/output databases and the `--tmp-folder` must be on a shared POSIX filesystem
+  visible to every node (e.g. NFS, Lustre, GPFS).
+* Recommended: one MPI rank per node, with `--threads` set to the number of CPUs allocated
+  to that node (OpenMP parallelizes within each rank). Running multiple ranks per node
+  works but duplicates the memory-mapped sequence/prefilter DB cache footprint per rank, so
+  it is not the default recommendation.
+
+Example `sbatch` script requesting 4 nodes and running `linclust` once `RUNNER` is set:
+
+```sh
+#!/bin/bash
+#SBATCH --nodes=4
+#SBATCH --ntasks=4
+#SBATCH --ntasks-per-node=1
+#SBATCH --cpus-per-task=32
+
+export RUNNER="srun --mpi=pmix -n $SLURM_NTASKS --ntasks-per-node=1"
+# or: export RUNNER="mpirun -np $SLURM_NTASKS"
+
+mmseqs linclust queryDB resultDB /shared/tmp --threads $SLURM_CPUS_PER_TASK
+```
+
+`--mpi-runner "srun --mpi=pmix -n 4 --ntasks-per-node=1"` works equivalently as a
+command-line flag instead of the `RUNNER` environment variable. The exact `srun --mpi=...`
+plugin name is site-specific; check with your cluster administrator or `srun --mpi=list`.
+
+Checkpoint/resume across topology changes: `align2clust` splits its candidate-generation
+work into fixed-size, topology-independent chunks (tunable via
+`MMSEQS_ALIGN2CLUST_CHUNK_SIZE`, default 50000) and records completed chunks under a
+`_chunks` directory next to the output. If a run is interrupted, rerunning the identical
+`mmseqs linclust`/`align2clust` command resumes by regenerating only the chunks that were
+not yet completed and reusing all finished ones -- this works even if the rank count
+changes between the interrupted and resumed run (e.g. starting on 8 nodes and resuming on
+1), because chunk boundaries and ownership only depend on the (fixed) input size and chunk
+size, never on the runner or rank count. The `_chunks` directory is removed automatically
+once a run completes successfully; keep it (e.g. via `--remove-tmp-files 0`) if you want to
+inspect or reuse it after a failure.
+
+Known limitations and scaling expectations:
+
+* `--include-align-files` is currently not supported together with distributed (multi-rank)
+  candidate generation in `align2clust` and is rejected at startup in that combination; it
+  still works normally on a single rank.
+* `clusthash` and `clust` (used for the optional `--cluster-mode`/hash-based prefilter path)
+  and the final `mergeclusters`/representative-switching stages are not MPI-aware and always
+  run once from the workflow process, regardless of the runner.
+* Representative selection and cluster assignment are a single deterministic, ordered
+  reduction performed on rank 0 after all candidate chunks are collected. Scaling is
+  therefore strongest when candidate/alignment generation dominates total runtime; very
+  large numbers of nodes will eventually be bottlenecked by this final reduction step and by
+  shared-filesystem bandwidth for chunk I/O rather than by compute.
+
 ## Contributors
 
 MMseqs2 exists thanks to all the people who contribute. 

data/workflow/linclust.sh

@@ -23,13 +23,13 @@ if [ "$LINCLUST_MODULE" = "linclust2" ]; then
     if [ -n "$CLUSTHASH" ]; then
         if notExists "${TMP_PATH}/input_clusthash.dbtype"; then
             # shellcheck disable=SC2086
-            $RUNNER "$MMSEQS" clusthash "$INPUT" "${TMP_PATH}/input_clusthash" ${CLUSTHASH_PAR} \
+            "$MMSEQS" clusthash "$INPUT" "${TMP_PATH}/input_clusthash" ${CLUSTHASH_PAR} \
                 || fail "clusthash died"
         fi
 
         if notExists "${TMP_PATH}/input_clusthash_clust.dbtype"; then
             # shellcheck disable=SC2086
-            $RUNNER "$MMSEQS" clust "$INPUT" "${TMP_PATH}/input_clusthash" "${TMP_PATH}/input_clusthash_clust" ${CLUSTHASH_CLUST_PAR} \
+            "$MMSEQS" clust "$INPUT" "${TMP_PATH}/input_clusthash" "${TMP_PATH}/input_clusthash_clust" ${CLUSTHASH_CLUST_PAR} \
                 || fail "clusthash-based clust died"
         fi
 
@@ -144,13 +144,13 @@ elif [ "$LINCLUST_MODULE" = "linclust1" ]; then
     if [ -n "$CLUSTHASH" ]; then
         if notExists "${TMP_PATH}/input_clusthash.dbtype"; then
             # shellcheck disable=SC2086
-            $RUNNER "$MMSEQS" clusthash "$INPUT" "${TMP_PATH}/input_clusthash" ${CLUSTHASH_PAR} \
+            "$MMSEQS" clusthash "$INPUT" "${TMP_PATH}/input_clusthash" ${CLUSTHASH_PAR} \
                     || fail "clust hash died"
         fi
 
         if notExists "${TMP_PATH}/input_clusthash_clust.dbtype"; then
             # shellcheck disable=SC2086
-            $RUNNER "$MMSEQS" clust "$INPUT" "${TMP_PATH}/input_clusthash" "${TMP_PATH}/input_clusthash_clust" ${CLUSTHASH_CLUST_PAR} \
+            "$MMSEQS" clust "$INPUT" "${TMP_PATH}/input_clusthash" "${TMP_PATH}/input_clusthash_clust" ${CLUSTHASH_CLUST_PAR} \
                     || fail "clust hash based clust died"
         fi
 

src/alignment/Align2ClustCheckpoint.cpp

@@ -0,0 +1,190 @@
+#include "Align2ClustCheckpoint.h"
+#include "Parameters.h"
+#include "FileUtil.h"
+#include "Util.h"
+#include "Debug.h"
+
+#include <cstdio>
+#include <cstring>
+#include <sstream>
+
+#include <dirent.h>
+#include <sys/stat.h>
+#include <unistd.h>
+
+static std::string computeFingerprintHex(const Parameters &par, size_t dbSize, size_t endRange,
+                                          int mode, size_t chunkSize) {
+    // Every field below can change the *result* of clustering and therefore must be
+    // part of the fingerprint. Rank count, thread count, and the runner used to
+    // launch the job are intentionally excluded so the fingerprint (and thus the
+    // checkpoint directory) stays identical across topology changes.
+    std::ostringstream key;
+    key << "db1=" << par.db1 << ";db1size=" << dbSize
+        << ";db2=" << par.db2 << ";endRange=" << endRange
+        << ";mode=" << mode << ";chunkSize=" << chunkSize
+        << ";covMode=" << par.covMode << ";covThr=" << par.covThr
+        << ";seqIdThr=" << par.seqIdThr << ";seqIdMode=" << par.seqIdMode
+        << ";evalThr=" << par.evalThr << ";alnLenThr=" << par.alnLenThr
+        << ";alignmentMode=" << par.alignmentMode
+        << ";compBiasCorrection=" << par.compBiasCorrection
+        << ";compBiasCorrectionScale=" << par.compBiasCorrectionScale
+        << ";gapOpen=" << par.gapOpen.values.aminoacid()
+        << ";gapExtend=" << par.gapExtend.values.aminoacid()
+        << ";zdrop=" << par.zdrop
+        << ";scoringMatrixFile=" << par.scoringMatrixFile.values.aminoacid()
+        << ";addBacktrace=" << par.addBacktrace
+        << ";includeAlignFiles=" << par.includeAlignFiles
+        << ";filterCluDBFile=" << par.filterCluDBFile
+        << ";filterSeqDBFile=" << par.filterSeqDBFile;
+
+    const std::string keyStr = key.str();
+    const size_t hash = Util::hash<const char>(keyStr.data(), keyStr.size());
+
+    char buffer[32];
+    snprintf(buffer, sizeof(buffer), "%016zx", hash);
+    return std::string(buffer);
+}
+
+static void ensureDirExists(const std::string &dir) {
+    if (FileUtil::directoryExists(dir.c_str())) {
+        return;
+    }
+    if (FileUtil::makeDir(dir.c_str()) == false && FileUtil::directoryExists(dir.c_str()) == false) {
+        Debug(Debug::ERROR) << "Could not create directory " << dir << "\n";
+        EXIT(EXIT_FAILURE);
+    }
+}
+
+Align2ClustCheckpoint::Align2ClustCheckpoint(const Parameters &par, const std::string &baseOutputPath,
+                                             size_t dbSize, size_t endRange, int mode, size_t chunkSize)
+    : parentDir_(baseOutputPath + "_chunks"),
+      fingerprint_(computeFingerprintHex(par, dbSize, endRange, mode, chunkSize)),
+      chunkDir_(parentDir_ + "/" + fingerprint_) {
+}
+
+void Align2ClustCheckpoint::ensureChunkDirExists() const {
+    ensureDirExists(parentDir_);
+    ensureDirExists(chunkDir_);
+}
+
+std::string Align2ClustCheckpoint::dataPath(size_t chunkIndex) const {
+    return chunkDir_ + "/chunk_" + SSTR(chunkIndex) + ".bin";
+}
+
+std::string Align2ClustCheckpoint::donePath(size_t chunkIndex) const {
+    return chunkDir_ + "/chunk_" + SSTR(chunkIndex) + ".done";
+}
+
+bool Align2ClustCheckpoint::isChunkDone(size_t chunkIndex) const {
+    return FileUtil::fileExists(donePath(chunkIndex).c_str());
+}
+
+void Align2ClustCheckpoint::writeChunk(size_t chunkIndex, const std::vector<ClusterResult> &results) const {
+    const std::string finalPath = dataPath(chunkIndex);
+    const std::string tmpPath = finalPath + ".tmp";
+
+    FILE *fp = fopen(tmpPath.c_str(), "wb");
+    if (fp == NULL) {
+        Debug(Debug::ERROR) << "Could not open " << tmpPath << " for writing\n";
+        EXIT(EXIT_FAILURE);
+    }
+
+    const uint64_t resultCount = static_cast<uint64_t>(results.size());
+    fwrite(&resultCount, sizeof(uint64_t), 1, fp);
+    for (const ClusterResult &result : results) {
+        const uint64_t sequenceIdx = static_cast<uint64_t>(result.sequenceIdx);
+        const uint64_t representativeId = static_cast<uint64_t>(result.representativeId);
+        const uint64_t prefSize = static_cast<uint64_t>(result.prefSize);
+        const uint64_t memberCount = static_cast<uint64_t>(result.memberIds.size());
+        fwrite(&sequenceIdx, sizeof(uint64_t), 1, fp);
+        fwrite(&representativeId, sizeof(uint64_t), 1, fp);
+        fwrite(&prefSize, sizeof(uint64_t), 1, fp);
+        fwrite(&memberCount, sizeof(uint64_t), 1, fp);
+        for (DBLocalId memberId : result.memberIds) {
+            const uint64_t member = static_cast<uint64_t>(memberId);
+            fwrite(&member, sizeof(uint64_t), 1, fp);
+        }
+    }
+    fflush(fp);
+    fsync(fileno(fp));
+    fclose(fp);
+
+    if (rename(tmpPath.c_str(), finalPath.c_str()) != 0) {
+        Debug(Debug::ERROR) << "Could not rename " << tmpPath << " to " << finalPath << "\n";
+        EXIT(EXIT_FAILURE);
+    }
+
+    // The empty ".done" marker is created only after the data file has been fully
+    // written and atomically renamed into place, so isChunkDone() never observes a
+    // partially written chunk.
+    FILE *doneFp = fopen(donePath(chunkIndex).c_str(), "wb");
+    if (doneFp == NULL) {
+        Debug(Debug::ERROR) << "Could not create " << donePath(chunkIndex) << "\n";
+        EXIT(EXIT_FAILURE);
+    }
+    fclose(doneFp);
+}
+
+std::vector<ClusterResult> Align2ClustCheckpoint::readChunk(size_t chunkIndex) const {
+    std::vector<ClusterResult> results;
+
+    const std::string path = dataPath(chunkIndex);
+    FILE *fp = fopen(path.c_str(), "rb");
+    if (fp == NULL) {
+        Debug(Debug::ERROR) << "Could not open " << path << " for reading\n";
+        EXIT(EXIT_FAILURE);
+    }
+
+    uint64_t resultCount = 0;
+    if (fread(&resultCount, sizeof(uint64_t), 1, fp) != 1) {
+        Debug(Debug::ERROR) << "Could not read chunk header from " << path << "\n";
+        EXIT(EXIT_FAILURE);
+    }
+    results.reserve(resultCount);
+    for (uint64_t i = 0; i < resultCount; ++i) {
+        uint64_t sequenceIdx = 0, representativeId = 0, prefSize = 0, memberCount = 0;
+        if (fread(&sequenceIdx, sizeof(uint64_t), 1, fp) != 1 ||
+            fread(&representativeId, sizeof(uint64_t), 1, fp) != 1 ||
+            fread(&prefSize, sizeof(uint64_t), 1, fp) != 1 ||
+            fread(&memberCount, sizeof(uint64_t), 1, fp) != 1) {
+            Debug(Debug::ERROR) << "Truncated chunk file " << path << "\n";
+            EXIT(EXIT_FAILURE);
+        }
+
+        ClusterResult result;
+        result.sequenceIdx = static_cast<size_t>(sequenceIdx);
+        result.representativeId = static_cast<DBLocalId>(representativeId);
+        result.prefSize = static_cast<size_t>(prefSize);
+        result.memberIds.resize(memberCount);
+        for (uint64_t m = 0; m < memberCount; ++m) {
+            uint64_t member = 0;
+            if (fread(&member, sizeof(uint64_t), 1, fp) != 1) {
+                Debug(Debug::ERROR) << "Truncated chunk file " << path << "\n";
+                EXIT(EXIT_FAILURE);
+            }
+            result.memberIds[m] = static_cast<DBLocalId>(member);
+        }
+        results.push_back(std::move(result));
+    }
+    fclose(fp);
+    return results;
+}
+
+void Align2ClustCheckpoint::cleanup() const {
+    DIR *dir = opendir(chunkDir_.c_str());
+    if (dir != NULL) {
+        struct dirent *entry;
+        while ((entry = readdir(dir)) != NULL) {
+            const std::string name = entry->d_name;
+            if (name == "." || name == "..") {
+                continue;
+            }
+            FileUtil::remove((chunkDir_ + "/" + name).c_str());
+        }
+        closedir(dir);
+    }
+    rmdir(chunkDir_.c_str());
+    // Only removes parentDir_ if this was the last fingerprint subdirectory in it;
+    // harmless no-op (ENOTEMPTY) otherwise.
+    rmdir(parentDir_.c_str());
+}

src/alignment/Align2ClustCheckpoint.h

@@ -0,0 +1,70 @@
+#ifndef MMSEQS_ALIGN2CLUST_CHECKPOINT_H
+#define MMSEQS_ALIGN2CLUST_CHECKPOINT_H
+
+#include "Align2ClustReducer.h"
+
+#include <cstddef>
+#include <string>
+#include <vector>
+
+class Parameters;
+
+// Manages on-disk chunk checkpoint state for a (potentially MPI-distributed)
+// align2clust run.
+//
+// The checkpoint directory is named after a fingerprint computed from every
+// parameter that affects the *result* of clustering (thresholds, alignment
+// mode, input database identity/size, chunk size, ...). It deliberately
+// excludes anything about *how* the work is distributed (rank count, thread
+// count, MPI runner): those never change the fingerprint, so the exact same
+// checkpoint directory -- and therefore the exact same per-chunk completion
+// state -- is reused whether a run is resumed on the same topology, a
+// different number of nodes, or a single node. A run interrupted on N nodes
+// can always be resumed on M nodes (including M == 1) as long as the
+// semantic clustering parameters are unchanged.
+//
+// Known limitation: input database identity is fingerprinted by path and
+// sequence count, not file content checksum. Replacing db1/db2 in place with
+// different content of the same size, between an interrupted run and its
+// resume, will not be detected. Documented as an out-of-scope hardening item.
+class Align2ClustCheckpoint {
+public:
+    // baseOutputPath is typically par.db3 (the align2clust cluster-result DB
+    // path); the checkpoint directory is created as a sibling of it.
+    Align2ClustCheckpoint(const Parameters &par, const std::string &baseOutputPath,
+                          size_t dbSize, size_t endRange, int mode, size_t chunkSize);
+
+    const std::string &getChunkDir() const { return chunkDir_; }
+    const std::string &getFingerprint() const { return fingerprint_; }
+
+    // Ensures the checkpoint directory (and its parent) exist on disk.
+    void ensureChunkDirExists() const;
+
+    bool isChunkDone(size_t chunkIndex) const;
+
+    // Serializes results to disk and marks the chunk done. The data file is
+    // written to a temporary name and rename()'d into place (atomic on a
+    // POSIX filesystem) before the ".done" marker is created, so a chunk is
+    // only ever observed as done once its data is fully and correctly on
+    // disk; a crash mid-write leaves no ".done" marker and the chunk is
+    // regenerated on the next run.
+    void writeChunk(size_t chunkIndex, const std::vector<ClusterResult> &results) const;
+
+    std::vector<ClusterResult> readChunk(size_t chunkIndex) const;
+
+    // Removes the entire checkpoint directory (and its parent, if left
+    // empty). Call only after every chunk has been consumed and the final
+    // clustering result has been written successfully -- an interrupted run
+    // must keep its checkpoint directory so it can be resumed.
+    void cleanup() const;
+
+private:
+    std::string dataPath(size_t chunkIndex) const;
+    std::string donePath(size_t chunkIndex) const;
+
+    std::string parentDir_;
+    std::string fingerprint_;
+    std::string chunkDir_;
+};
+
+#endif