Experiment with using taskflow for parallel presolving

Author: arcondello

.gitmodules

@@ -2,3 +2,6 @@
 	path = testscpp/Catch2
 	url = https://github.com/catchorg/Catch2.git
 	branch = v2.x
+[submodule "extern/taskflow"]
+	path = extern/taskflow
+	url = git@github.com:taskflow/taskflow.git

MANIFEST.in

@@ -1,3 +1,5 @@
 include pyproject.toml
 recursive-include dwave/preprocessing/include/ *.hpp *.h
 recursive-include dwave/preprocessing *.pyx *.pxd *.pyx.src
+graft extern/taskflow/taskflow
+include extern/taskflow/LICENSE

dwave/preprocessing/include/dwave/presolve.h

@@ -14,12 +14,17 @@
 
 #pragma once
 
+#include <chrono>
+#include <thread>
+
+
 #include <algorithm>
 #include <unordered_map>
 #include <utility>
 #include <vector>
 
-#include "dimod/constrained_quadratic_model.h"
+#include <dimod/constrained_quadratic_model.h>
+#include <taskflow/taskflow.hpp>
 
 namespace dwave {
 namespace presolve {
@@ -134,10 +139,24 @@ class Presolver {
     /// Apply any loaded presolve techniques. Acts of the model() in-place.
     void apply();
 
+    void apply_parallel() {
+        tf::Executor executor;
+        tf::Taskflow taskflow;
+
+        taskflow.for_each_index(0, static_cast<int>(model_.num_constraints()), 1,
+                                [&](int i) { do_slow_thing(i); });
+
+        executor.run(taskflow).wait();
+    }
+
     /// Detach the constrained quadratic model and return it.
     /// This clears the model from the presolver.
     model_type detach_model();
 
+    void do_slow_thing(index_type c) {
+        std::this_thread::sleep_for(std::chrono::seconds(1));
+    }
+
     /// Load the default presolve techniques.
     void load_default_presolvers();
 

dwave/preprocessing/libcpp.pxd

@@ -29,6 +29,7 @@ cdef extern from "dwave/presolve.h" namespace "dwave::presolve" nogil:
         Presolver()
         Presolver(model_type)
         void apply() except+
+        void apply_parallel() except+
         model_type detach_model()
         void load_default_presolvers()
         model_type& model()

dwave/preprocessing/presolve/cypresolve.pyx

@@ -47,6 +47,13 @@ cdef class cyPresolver:
         self.cpppresolver.apply()
         self._model_num_variables = self.cpppresolver.model().num_variables()
 
+    def apply_parallel(self):
+        import time
+
+        t = time.perf_counter()
+        self.cpppresolver.apply_parallel()
+        return time.perf_counter() - t
+
     def clear_model(self):
         """Clear the held model. This is useful to save memory."""
         self.cpppresolver.detach_model()

extern/taskflow

@@ -23,13 +23,13 @@
 from distutils.command.build_ext import build_ext as _build_ext
 
 extra_compile_args = {
-    'msvc': ['/EHsc'],
-    'unix': ['-std=c++11'],
+    'msvc': ['/EHsc', '-pthread'],
+    'unix': ['-std=c++17', '-pthread'],
 }
 
 extra_link_args = {
     'msvc': [],
-    'unix': ['-std=c++11'],
+    'unix': ['-std=c++17'],
 }
 
 
@@ -60,6 +60,7 @@ def build_extensions(self):
     include_dirs=[
         numpy.get_include(),
         dimod.get_include(),
+        "extern/taskflow/",
         ],
     install_requires=[
         'numpy>=1.20.0,<2.0.0',  # keep synced with circle-ci, pyproject.toml

setup.py

@@ -20,6 +20,16 @@
 from dwave.preprocessing import Presolver, InfeasibleModelError
 
 
+class TestExperiment(unittest.TestCase):
+    def test_parallel(self):
+        cqm = dimod.CQM()
+        for _ in range(100):
+            cqm.add_constraint(dimod.BQM("BINARY") == 1)
+
+        presolver = Presolver(cqm)
+        self.assertLessEqual(presolver.apply_parallel(), 55)  # should take 100s
+
+
 class TestPresolver(unittest.TestCase):
     def test_bug0(self):
         random = np.random.RandomState(0)

tests/test_presolve.py

@@ -2,6 +2,7 @@ ROOT := ../
 SRC := $(ROOT)/dwave/preprocessing/
 CATCH2 := $(ROOT)/testscpp/Catch2/single_include/
 DIMOD := $(shell python -c 'import dimod; print(dimod.get_include())')
+TASKFLOW := $(ROOT)/extern/taskflow/
 
 all: catch2 test_main test_main_parallel tests tests_parallel
 
@@ -12,12 +13,12 @@ tests_parallel: test_main_parallel.out
 	./test_main_parallel
 
 test_main: test_main.cpp
-	g++ -std=c++11 -Wall -c test_main.cpp
-	g++ -std=c++11 -Wall test_main.o tests/*.cpp -o test_main -I $(SRC)/include/ -I $(DIMOD) -I $(CATCH2)
+	g++ -std=c++17 -Wall -pthread -c test_main.cpp
+	g++ -std=c++17 -Wall -pthread test_main.o tests/*.cpp -o test_main -I $(SRC)/include/ -I $(DIMOD) -I $(CATCH2) -I $(TASKFLOW)
 
 test_main_parallel: test_main.cpp
-	g++ -std=c++11 -fopenmp -Wall -c test_main.cpp -o test_main_parallel.o
-	g++ -std=c++11 -fopenmp -Wall test_main_parallel.o tests/*.cpp -o test_main_parallel -I $(SRC)/include/ -I $(DIMOD) -I $(CATCH2)
+	g++ -std=c++17 -fopenmp -Wall -c test_main.cpp -o test_main_parallel.o
+	g++ -std=c++17 -fopenmp -Wall test_main_parallel.o tests/*.cpp -o test_main_parallel -I $(SRC)/include/ -I $(DIMOD) -I $(CATCH2) -I $(TASKFLOW)
 
 catch2:
 	git submodule init

testscpp/Makefile

@@ -350,4 +350,14 @@ SCENARIO("constrained quadratic models can be presolved") {
     }
 }
 
+TEST_CASE("experiment") {
+    GIVEN("a CQM with a bunch of constraints") {
+        auto cqm = dimod::ConstrainedQuadraticModel<double>();
+        cqm.add_constraints(10);
+
+        auto presolver = presolve::Presolver<double>(std::move(cqm));
+        presolver.apply_parallel();
+    }
+}
+
 }  // namespace dwave