Run ABCs in parallel

Author: rocallahan

Makefile

@@ -43,7 +43,12 @@ LINK_ABC := 0
 # Needed for environments that can't run executables (i.e. emscripten, wasm)
 DISABLE_SPAWN := 0
 # Needed for environments that don't have proper thread support (i.e. emscripten, wasm--for now)
+ENABLE_THREADS := 1
+ifeq ($(ENABLE_THREADS),1)
 DISABLE_ABC_THREADS := 0
+else
+DISABLE_ABC_THREADS := 1
+endif
 
 # clang sanitizers
 SANITIZER =
@@ -301,6 +306,7 @@ DISABLE_SPAWN := 1
 
 ifeq ($(ENABLE_ABC),1)
 LINK_ABC := 1
+ENABLE_THREADS := 0
 DISABLE_ABC_THREADS := 1
 endif
 
@@ -458,6 +464,11 @@ CXXFLAGS := -Og -DDEBUG $(filter-out $(OPT_LEVEL),$(CXXFLAGS))
 STRIP :=
 endif
 
+ifeq ($(ENABLE_THREADS),1)
+CXXFLAGS += -DYOSYS_ENABLE_THREADS
+LIBS += -lpthread
+endif
+
 ifeq ($(ENABLE_ABC),1)
 CXXFLAGS += -DYOSYS_ENABLE_ABC
 ifeq ($(LINK_ABC),1)
@@ -614,6 +625,7 @@ $(eval $(call add_include_file,kernel/satgen.h))
 $(eval $(call add_include_file,kernel/scopeinfo.h))
 $(eval $(call add_include_file,kernel/sexpr.h))
 $(eval $(call add_include_file,kernel/sigtools.h))
+$(eval $(call add_include_file,kernel/threading.h))
 $(eval $(call add_include_file,kernel/timinginfo.h))
 $(eval $(call add_include_file,kernel/utils.h))
 $(eval $(call add_include_file,kernel/yosys.h))
@@ -636,7 +648,7 @@ $(eval $(call add_include_file,backends/rtlil/rtlil_backend.h))
 OBJS += kernel/driver.o kernel/register.o kernel/rtlil.o kernel/log.o kernel/calc.o kernel/yosys.o kernel/io.o kernel/gzip.o
 OBJS += kernel/binding.o kernel/tclapi.o
 OBJS += kernel/cellaigs.o kernel/celledges.o kernel/cost.o kernel/satgen.o kernel/scopeinfo.o kernel/qcsat.o kernel/mem.o kernel/ffmerge.o kernel/ff.o kernel/yw.o kernel/json.o kernel/fmt.o kernel/sexpr.o
-OBJS += kernel/drivertools.o kernel/functional.o
+OBJS += kernel/drivertools.o kernel/functional.o kernel/threading.o
 ifeq ($(ENABLE_ZLIB),1)
 OBJS += kernel/fstdata.o
 endif

kernel/threading.cc

@@ -0,0 +1,44 @@
+#include "kernel/yosys_common.h"
+#include "kernel/threading.h"
+
+YOSYS_NAMESPACE_BEGIN
+
+void DeferredLogs::flush()
+{
+	for (auto &m : logs)
+		if (m.error)
+			YOSYS_NAMESPACE_PREFIX log_error("%s", m.text.c_str());
+		else
+			YOSYS_NAMESPACE_PREFIX log("%s", m.text.c_str());
+}
+
+int ThreadPool::pool_size(int reserved_cores, int max_threads)
+{
+#ifdef YOSYS_ENABLE_THREADS
+	int num_threads = std::min<int>(std::thread::hardware_concurrency() - reserved_cores, max_threads);
+        return std::max(0, num_threads);
+#else
+        return 0;
+#endif
+}
+
+ThreadPool::ThreadPool(int pool_size, std::function<void(int)> body)
+{
+#ifdef YOSYS_ENABLE_THREADS
+        threads.reserve(pool_size);
+        for (int i = 0; i < pool_size; i++)
+                threads.emplace_back([i, &body]{ body(i); });
+#else
+        log_assert(pool_size == 0);
+#endif
+}
+
+ThreadPool::~ThreadPool()
+{
+#ifdef YOSYS_ENABLE_THREADS
+	for (auto &t : threads)
+		t.join();
+#endif
+}
+
+YOSYS_NAMESPACE_END

kernel/threading.h

@@ -0,0 +1,151 @@
+#ifdef YOSYS_ENABLE_THREADS
+#include <condition_variable>
+#include <deque>
+#include <mutex>
+#include <thread>
+#endif
+
+#include "kernel/yosys_common.h"
+#include "kernel/log.h"
+
+#ifndef YOSYS_THREADING_H
+#define YOSYS_THREADING_H
+
+YOSYS_NAMESPACE_BEGIN
+
+// Concurrent queue implementation. Not fast, but simple.
+// Multi-producer, multi-consumer, optionally bounded.
+// When YOSYS_ENABLE_THREADS is not defined, this is just a non-thread-safe non-blocking deque.
+template <typename T>
+class ConcurrentQueue
+{
+public:
+	ConcurrentQueue(int capacity = INT_MAX)
+		: capacity(capacity) {}
+	// Push an element into the queue. If it's at capacity, block until there is room.
+	void push_back(T t)
+	{
+#ifdef YOSYS_ENABLE_THREADS
+		std::unique_lock<std::mutex> lock(mutex);
+		not_full_condition.wait(lock, [this] { return static_cast<int>(contents.size()) < capacity; });
+		if (contents.empty())
+			not_empty_condition.notify_one();
+#endif
+		log_assert(!closed);
+		contents.push_back(std::move(t));
+#ifdef YOSYS_ENABLE_THREADS
+		if (static_cast<int>(contents.size()) < capacity)
+			not_full_condition.notify_one();
+#endif
+	}
+	// Signal that no more elements will be produced. `pop_front()` will return nullopt.
+	void close()
+	{
+#ifdef YOSYS_ENABLE_THREADS
+		std::unique_lock<std::mutex> lock(mutex);
+		not_empty_condition.notify_all();
+#endif
+		closed = true;
+	}
+	// Pop an element from the queue. Blocks until an element is available
+	// or the queue is closed and empty.
+	std::optional<T> pop_front()
+	{
+		return pop_front_internal(true);
+	}
+	// Pop an element from the queue. Does not block, just returns nullopt if the
+	// queue is empty.
+	std::optional<T> try_pop_front()
+	{
+		return pop_front_internal(false);
+	}
+private:
+	std::optional<T> pop_front_internal(bool wait = false)
+	{
+#ifdef YOSYS_ENABLE_THREADS
+		std::unique_lock<std::mutex> lock(mutex);
+		if (wait) {
+			not_empty_condition.wait(lock, [this] { return !contents.empty() || closed; });
+		}
+		if (contents.empty())
+			return std::nullopt;
+		if (static_cast<int>(contents.size()) == capacity)
+			not_full_condition.notify_one();
+#endif
+		T result = std::move(contents.front());
+		contents.pop_front();
+#ifdef YOSYS_ENABLE_THREADS
+		if (!contents.empty())
+			not_empty_condition.notify_one();
+#endif
+		return std::move(result);
+	}
+
+#ifdef YOSYS_ENABLE_THREADS
+	std::mutex mutex;
+	// Signals one waiter thread when the queue changes and is not full.
+	std::condition_variable not_full_condition;
+	// Signals one waiter thread when the queue changes and is not empty.
+	std::condition_variable not_empty_condition;
+#endif
+	std::deque<T> contents;
+	int capacity;
+	bool closed = false;
+};
+
+class DeferredLogs
+{
+public:
+	template <typename... Args>
+	void log(FmtString<TypeIdentity<Args>...> fmt, Args... args)
+	{
+		logs.push_back({fmt.format(args...), false});
+	}
+	template <typename... Args>
+	void log_error(FmtString<TypeIdentity<Args>...> fmt, Args... args)
+	{
+		logs.push_back({fmt.format(args...), true});
+	}
+	void flush();
+private:
+	struct Message
+	{
+		std::string text;
+		bool error;
+	};
+	std::vector<Message> logs;
+};
+
+class ThreadPool
+{
+public:
+	// Computes the number of worker threads to use.
+	// `reserved_cores` cores are set aside for other threads (e.g. work on the main thread).
+	// `max_threads` --- don't return more workers than this.
+	// The result may be 0.
+	static int pool_size(int reserved_cores, int max_threads);
+
+	// Create a pool of threads running the given closure (parameterized by thread number).
+	// `pool_size` must be the result of a `pool_size()` call.
+	ThreadPool(int pool_size, std::function<void(int)> body);
+	// Waits for all threads to terminate. Make sure those closures return!
+	~ThreadPool();
+
+	// Return the number of threads in the pool.
+	int num_threads() const
+	{
+#ifdef YOSYS_ENABLE_THREADS
+		return threads.size();
+#else
+		return 0;
+#endif
+	}
+private:
+#ifdef YOSYS_ENABLE_THREADS
+	std::vector<std::thread> threads;
+#endif
+};
+
+YOSYS_NAMESPACE_END
+
+#endif // YOSYS_THREADING_H

kernel/yosys.cc

@@ -177,7 +177,7 @@ int run_command(const std::string &command, std::function<void(const std::string
 
 	int ret = pclose(f);
 	if (ret < 0)
-		return -1;
+		return -2;
 #ifdef _WIN32
 	return ret;
 #else