Limit max process count in phase 1

Allow specifying maximum number of simultaneous plot processes
entering phase 1 for staggered plotting.
Temporary files are used as flags: one file per process.
Sample cli usage:
mkdir -p /home/user/.config/chia/run
./ProofOfSpace create -k 24 --p1maxproc 2 --runtimedir /home/user/.config/chia/run

python-bindings/chiapos.cpp

@@ -49,7 +49,9 @@ PYBIND11_MODULE(chiapos, m)
                uint32_t num_buckets,
                uint32_t stripe_size,
                uint8_t num_threads,
-               bool nobitfield) {
+               bool nobitfield,
+               const std::string runtime_dir,
+               uint32_t phase1_max_processes) {
                 std::string memo_str(memo);
                 const uint8_t *memo_ptr = reinterpret_cast<const uint8_t *>(memo_str.data());
                 std::string id_str(id);
@@ -68,7 +70,9 @@ PYBIND11_MODULE(chiapos, m)
                                       num_buckets,
                                       stripe_size,
                                       num_threads,
-                                      nobitfield);
+                                      nobitfield,
+                                      runtime_dir,
+                                      phase1_max_processes);
                 } catch (const std::exception &e) {
                     std::cout << "Caught plotting error: " << e.what() << std::endl;
                     throw e;

src/cli.cpp

@@ -82,6 +82,8 @@ int main(int argc, char *argv[]) try {
     string memo = "0102030405";
     string id = "022fb42c08c12de3a6af053880199806532e79515f94e83461612101f9412f9e";
     bool nobitfield = false;
+    uint32_t phase1_max_processes = 0;
+    string runtimedir = ".";
     uint32_t buffmegabytes = 0;
 
     options.allow_unrecognised_options().add_options()(
@@ -96,6 +98,8 @@ int main(int argc, char *argv[]) try {
         "m, memo", "Memo to insert into the plot", cxxopts::value<string>(memo))(
         "i, id", "Unique 32-byte seed for the plot", cxxopts::value<string>(id))(
         "e, nobitfield", "Disable bitfield", cxxopts::value<bool>(nobitfield))(
+        "p1maxproc", "Phase 1 max process count", cxxopts::value<uint32_t>(phase1_max_processes))(
+        "runtimedir", "Runtime directory", cxxopts::value<string>(runtimedir))(
         "b, buffer",
         "Megabytes to be used as buffer for sorting and plotting",
         cxxopts::value<uint32_t>(buffmegabytes))("help", "Print help");
@@ -145,7 +149,9 @@ int main(int argc, char *argv[]) try {
                 num_buckets,
                 num_stripes,
                 num_threads,
-                nobitfield);
+                nobitfield,
+                runtimedir,
+                phase1_max_processes);
     } else if (operation == "prove") {
         if (argc < 3) {
             HelpAndQuit(options);

src/disk_util.hpp

@@ -39,6 +39,9 @@
 #include "chia_filesystem.hpp"
 #include "util.hpp"
 
+using namespace std::chrono;
+using namespace std::chrono_literals;
+
 namespace DiskUtil {
 
 #if !defined(__APPLE__) && !defined(_WIN32)
@@ -134,7 +137,6 @@ namespace DiskUtil {
             return -1;
         }
         while (0 != flock(dir_fd, LOCK_EX | LOCK_NB)) {
-            using namespace std::chrono_literals;
             if (EWOULDBLOCK == errno) {
                 std::this_thread::sleep_for(10s);
             } else {
@@ -190,8 +192,6 @@ class DirectoryLock
     bool Lock()
     {
         if (fd_ == -1) {
-            using namespace std::chrono;
-
             std::cout << "Acquiring directory lock: " << dirname_ << std::endl;
 
             steady_clock::time_point start = steady_clock::now();
@@ -223,6 +223,128 @@ class DirectoryLock
     std::string dirname_;
 };
 
+class MultiFileLock
+{
+public:
+    MultiFileLock(const std::string &runtime_dir, const std::string &lock_name,
+        int max_slots, bool lock = true)
+    {
+        runtime_dir_ = runtime_dir;
+        lock_name_ = lock_name;
+
+        std::ostringstream prefix;
+        prefix << "." << lock_name << "-lock";
+        prefix_ = prefix.str();
+
+        max_slots_ = max_slots;
+        if (lock) {
+            Lock();
+        }
+    }
+
+    MultiFileLock(const MultiFileLock&) = delete;
+
+    virtual ~MultiFileLock()
+    {
+        Unlock();
+    }
+
+    bool Lock()
+    {
+#ifdef _WIN32
+        return false;
+#else
+        if (max_slots_ < 1 || fd_ != -1) {
+            return false;
+        }
+
+        std::cout << "Acquiring " << lock_name_ << " lock" << std::endl;
+
+        steady_clock::time_point start = steady_clock::now();
+        while (!TryLock()) {
+            std::this_thread::sleep_for(20s);
+        }
+        steady_clock::time_point end = steady_clock::now();
+
+        std::cout << "Lock acquired (took "
+            << duration_cast<seconds>(end - start).count()
+            << " sec)" << std::endl;
+
+        return true;
+#endif
+    }
+
+    bool Unlock()
+    {
+#ifdef _WIN32
+        return false;
+#else
+        if (fd_ == -1) {
+            return false;
+        }
+
+        std::cout << "Releasing " << lock_name_ << " lock" << std::endl;
+
+        if (-1 == flock(fd_, LOCK_UN)) {
+            std::cerr << "Failed to unlock the file: " << strerror(errno)
+                << std::endl;
+            return false;
+        }
+
+        if (-1 == close(fd_)) {
+            std::cerr << "Failed to close the file during unlocking: "
+                << strerror(errno) << std::endl;
+            return false;
+        }
+
+        fd_ = -1;
+
+        return true;
+#endif
+    }
+
+private:
+#ifndef _WIN32
+    bool TryLock()
+    {
+        for (int current_slot = 0; current_slot < max_slots_; ++current_slot) {
+            fs::path path(runtime_dir_);
+            std::ostringstream filename;
+            filename << prefix_ << "-" << current_slot;
+            path.append(filename.str());
+
+            std::string fullname = path.string();
+
+            fd_ = open(fullname.c_str(), O_CREAT | O_RDONLY | O_NOCTTY, 0666);
+            if (fd_ == -1) {
+                std::cerr << "Unable to open file for locking: " << fullname
+                    << ". Error: " << strerror(errno) << std::endl;
+                return false;
+            }
+            if (0 == flock(fd_, LOCK_EX | LOCK_NB)) {
+                return true;
+            }
+            if (EWOULDBLOCK != errno) {
+                std::cerr << "Error while trying to lock " << fullname << ": "
+                    << strerror(errno) << std::endl;
+            }
+            if (-1 == close(fd_)) {
+                std::cerr << "Failed to close " << fullname << ": "
+                    << strerror(errno) << std::endl;
+            }
+            fd_ = -1;
+        }
+        return false;
+    }
+#endif
+
+    int fd_ = -1;
+    std::string runtime_dir_;
+    std::string lock_name_;
+    std::string prefix_;
+    int max_slots_ = 0;
+};
+
 #endif // SRC_CPP_DISK_UTIL_HPP_
 
 

src/plotter_disk.hpp

@@ -72,7 +72,9 @@ class DiskPlotter {
         uint32_t num_buckets_input = 0,
         uint64_t stripe_size_input = 0,
         uint8_t num_threads_input = 0,
-        bool nobitfield = false)
+        bool nobitfield = false,
+        std::string runtime_dir = ".",
+        uint32_t phase1_max_processes = 0)
     {
         // Increases the open file limit, we will open a lot of files.
 #ifndef _WIN32
@@ -216,12 +218,18 @@ class DiskPlotter {
 
             assert(id_len == kIdLen);
 
+            if (phase1_max_processes > 0) {
+                std::cout << std::endl;
+            }
+            MultiFileLock lock(runtime_dir, "phase1", phase1_max_processes);
+
             std::cout << std::endl
                       << "Starting phase 1/4: Forward Propagation into tmp files... "
                       << Timer::GetNow();
 
-            Timer p1;
             Timer all_phases;
+
+            Timer p1;
             std::vector<uint64_t> table_sizes = RunPhase1(
                 tmp_1_disks,
                 k,
@@ -235,6 +243,8 @@ class DiskPlotter {
                 num_threads);
             p1.PrintElapsed("Time for phase 1 =");
 
+            lock.Unlock();
+
             uint64_t finalsize=0;
 
             if(nobitfield)

tests/test_python_bindings.py

@@ -59,6 +59,8 @@ def test_k_21(self):
             8192,
             8,
             False,
+            ".",
+            0
         )
         pl = None