RooFit::MultiProcess & TestStatistics part 2b: MultiProcess

builtins/zeromq/CMakeLists.txt

@@ -0,0 +1,48 @@
+include(ExternalProject)
+
+set(ZeroMQ_PREFIX ${CMAKE_CURRENT_BINARY_DIR}/BUILTIN_ZeroMQ-prefix)
+set(ZeroMQ_LIBNAME ${CMAKE_STATIC_LIBRARY_PREFIX}zmq${CMAKE_STATIC_LIBRARY_SUFFIX})
+
+set(ZeroMQ_LIBRARY ${ZeroMQ_PREFIX}/lib/${ZeroMQ_LIBNAME} CACHE INTERNAL "" FORCE)
+set(ZeroMQ_LIBRARIES ${ZeroMQ_LIBRARY} CACHE INTERNAL "" FORCE)
+
+ExternalProject_Add(BUILTIN_ZeroMQ
+    URL https://github.com/zeromq/libzmq/archive/refs/tags/v4.3.2.tar.gz
+    CMAKE_ARGS
+        -DCMAKE_INSTALL_PREFIX=<INSTALL_DIR>
+        -DWITH_PERF_TOOL=OFF
+        -DZMQ_BUILD_TESTS=OFF
+        -DPOLLER=select
+        # TODO: remove -w flag when updating to v4.3.5 or higher, current warnings should be gone by then
+        -DCMAKE_CXX_FLAGS=${CMAKE_CXX_FLAGS}\ -w
+    BUILD_BYPRODUCTS ${ZeroMQ_LIBRARIES}
+    )
+
+ExternalProject_Get_Property(BUILTIN_ZeroMQ install_dir)
+
+set(ZeroMQ_FOUND TRUE CACHE BOOL "" FORCE)
+
+set(ZeroMQ_INCLUDE_DIR  ${ZeroMQ_PREFIX}/include CACHE INTERNAL "" FORCE)
+set(ZeroMQ_INCLUDE_DIRS ${ZeroMQ_PREFIX}/include CACHE INTERNAL "" FORCE)
+
+add_library(ZeroMQ INTERFACE)
+target_include_directories(ZeroMQ INTERFACE $<BUILD_INTERFACE:${ZeroMQ_INCLUDE_DIR}>)
+target_link_libraries(ZeroMQ INTERFACE $<BUILD_INTERFACE:${ZeroMQ_LIBRARIES}>)
+add_dependencies(ZeroMQ BUILTIN_ZeroMQ)
+
+add_library(ZeroMQ::ZeroMQ ALIAS ZeroMQ)
+add_library(libzmq ALIAS ZeroMQ)
+
+set(ZeroMQ_DIR ${ZeroMQ_PREFIX}/share/cmake/ZeroMQ CACHE INTERNAL "" FORCE)
+
+# also need source and build directories for ppoll in RooFitZMQ
+set(ZeroMQ_SOURCE_DIR ${ZeroMQ_PREFIX}/src/BUILTIN_ZeroMQ/src CACHE INTERNAL "" FORCE)
+set(ZeroMQ_BUILD_DIR ${ZeroMQ_PREFIX}/src/BUILTIN_ZeroMQ-build CACHE INTERNAL "" FORCE)
+
+ExternalProject_Add(BUILTIN_cppzmq
+    URL https://github.com/zeromq/cppzmq/archive/refs/tags/v4.7.1.tar.gz
+    CMAKE_ARGS
+        -DCMAKE_INSTALL_PREFIX=${install_dir}
+        -DCPPZMQ_BUILD_TESTS=OFF
+    DEPENDS BUILTIN_ZeroMQ
+    )

cmake/modules/RootBuildOptions.cmake

@@ -107,6 +107,7 @@ ROOT_BUILD_OPTION(builtin_vdt OFF "Build VDT internally (requires network)")
 ROOT_BUILD_OPTION(builtin_veccore OFF "Build VecCore internally (requires network)")
 ROOT_BUILD_OPTION(builtin_xrootd OFF "Build XRootD internally (requires network)")
 ROOT_BUILD_OPTION(builtin_xxhash OFF "Build bundled copy of xxHash")
+ROOT_BUILD_OPTION(builtin_zeromq ON "Build ZeroMQ internally (requires network)")
 ROOT_BUILD_OPTION(builtin_zlib OFF "Build bundled copy of zlib")
 ROOT_BUILD_OPTION(builtin_zstd OFF "Build included libzstd, or use system libzstd")
 ROOT_BUILD_OPTION(ccache OFF "Enable ccache usage for speeding up builds")
@@ -299,6 +300,7 @@ if(builtin_all)
   set(builtin_veccore_defvalue ON)
   set(builtin_xrootd_defvalue ON)
   set(builtin_xxhash_defvalue ON)
+  set(builtin_zeromq_defvalue ON)
   set(builtin_zlib_defvalue ON)
   set(builtin_zstd_defvalue ON)
 endif()

cmake/modules/SearchInstalledSoftware.cmake

@@ -1767,6 +1767,13 @@ if (mpi)
   endif()
 endif()
 
+#---Check for ZeroMQ-----------------------------------------------------------
+
+if(builtin_zeromq)
+  list(APPEND ROOT_BUILTINS ZeroMQ)
+  add_subdirectory(builtins/zeromq)
+endif()
+
 #---Download googletest--------------------------------------------------------------
 if (testing)
   # FIXME: Remove our version of gtest in roottest. We can reuse this one.

roofit/CMakeLists.txt

@@ -5,6 +5,10 @@
 # For the list of contributors see $ROOTSYS/README/CREDITS.
 
 add_subdirectory(batchcompute)
+if (builtin_zeromq)
+  add_subdirectory(roofitZMQ)
+  add_subdirectory(multiprocess)
+endif(builtin_zeromq)
 add_subdirectory(roofitcore)
 add_subdirectory(roofit)
 if(mathmore)

roofit/multiprocess/CMakeLists.txt

@@ -0,0 +1,40 @@
+############################################################################
+# CMakeLists.txt file for building ROOT RooFitMultiProcess package
+# @author Patrick Bos, Netherlands eScience Center
+############################################################################
+
+#FIND_PACKAGE(ZeroMQ REQUIRED)
+
+ROOT_STANDARD_LIBRARY_PACKAGE(RooFitMultiProcess
+    HEADERS
+        RooFit/MultiProcess/worker.h
+        RooFit/MultiProcess/Messenger_decl.h
+        RooFit/MultiProcess/Messenger.h
+        RooFit/MultiProcess/ProcessManager.h
+        RooFit/MultiProcess/JobManager.h
+        RooFit/MultiProcess/util.h
+        RooFit/MultiProcess/Queue.h
+        RooFit/MultiProcess/types.h
+        RooFit/MultiProcess/Job.h
+    SOURCES
+        src/worker.cxx
+        src/Messenger.cxx
+        src/ProcessManager.cxx
+        src/util.cxx
+        src/Queue.cxx
+        src/JobManager.cxx
+        src/Job.cxx
+    DICTIONARY_OPTIONS
+        "-writeEmptyRootPCM"
+    DEPENDENCIES
+        RooFitZMQ
+    LIBRARIES
+        ZeroMQ
+)
+
+target_link_libraries(RooFitMultiProcess PRIVATE ZeroMQ)
+target_include_directories(RooFitMultiProcess PUBLIC $<BUILD_INTERFACE:${ZeroMQ_INCLUDE_DIR}>)
+
+if(testing)
+    ROOT_ADD_TEST_SUBDIRECTORY(test)
+endif()

roofit/multiprocess/inc/LinkDef.h

@@ -0,0 +1 @@
+// ceci n'est pas un fichier

roofit/multiprocess/inc/RooFit/MultiProcess/Job.h

@@ -0,0 +1,130 @@
+/*
+ * Project: RooFit
+ * Authors:
+ *   PB, Patrick Bos, Netherlands eScience Center, p.bos@esciencecenter.nl
+ *   IP, Inti Pelupessy, Netherlands eScience Center, i.pelupessy@esciencecenter.nl
+ *
+ * Copyright (c) 2016-2019, Netherlands eScience Center
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ */
+#ifndef ROOT_ROOFIT_MultiProcess_Job_decl
+#define ROOT_ROOFIT_MultiProcess_Job_decl
+
+#include <string>
+#include <zmq.hpp>
+
+namespace RooFit {
+namespace MultiProcess {
+
+// forward declaration
+class JobManager;
+
+/*
+ * @brief interface class for defining the actual work that must be done
+ *
+ * Think of "job" as in "employment", e.g. the job of a baker, which
+ * involves *tasks* like baking and selling bread. The Job must define the
+ * tasks through its execution (evaluate_task), based on a task index argument.
+ *
+ * Classes inheriting from Job must implement the pure virtual methods:
+ * - void evaluate_task(std::size_t task)
+ * - void update_real(std::size_t ix, double val, bool is_constant)
+ * - void update_bool(std::size_t ix, bool value)
+ * - void receive_task_result_on_queue(std::size_t task, std::size_t worker_id)
+ * - void send_back_results_from_queue_to_master()
+ * - void clear_results()
+ * - void receive_results_on_master()
+ * - void send_back_task_result_from_worker(std::size_t task)
+ *
+ * The latter five deal with sending results back from workers to master.
+ * update_real is used to sync parameters that changed on master to the workers
+ * before the next set of tasks is queued; it must have a part that updates the
+ * actual parameters on the worker process, but can optionally also be used to
+ * send them from the master to the queue (Q2W is handled by the Queue::loop).
+ * An example implementation:
+
+
+
+ *
+ * The type of result from each task is strongly dependent on the Job at hand
+ * and so Job does not provide a default results vector or anything like this.
+ * It is up to the inheriting class to implement this. We would have liked a
+ * template parameter task_result_t, so that we could also provide a default
+ * "boilerplate" calculate function to show a typical Job use-case of all the
+ * above infrastructure. This is not trivial, because the JobManager has to
+ * keep a list of Job pointers, so if there would be different template
+ * instantiations of Jobs, this would complicate this list. As an example,
+ * a boilerplate calculation could look something like this:
+ *
+return_type CertainJob::evaluate() {
+   if (get_manager()->process_manager().is_master()) {
+      // update parameters that changed since last calculation (or creation if first time)
+      for (size_t ix = 0; ix < changed_parameters.size(); ++ix) {  // changed_parameters would be a member of CertainJob
+         auto msg = RooFit::MultiProcess::M2Q::update_real;
+         get_manager()->messenger().send_from_master_to_queue(msg, id, ix,
+                                                              changed_parameters[ix].getVal(),
+                                                              changed_parameters[ix].isConstant());
+      }
+
+      // master fills queue with tasks
+      for (std::size_t ix = 0; ix < N_tasks; ++ix) {  // N_tasks would also be a member of CertainJob
+         JobTask job_task(id, ix);
+         get_manager()->queue().add(job_task);
+      }
+
+      // wait for task results back from workers to master
+      gather_worker_results();
+
+      // possibly do some final manipulation of the results that were gathered
+   }
+   return result;  // or not, it could be stored and accessed later; also here, result would be a member of CertainJob
+}
+ *
+ * Child classes should refrain from direct access to the JobManager instance
+ * (through JobManager::instance), but rather use the here provided
+ * Job::get_manager(). This function starts the worker_loop on the worker when
+ * first called, meaning that the workers will not
+ */
+class Job {
+public:
+   explicit Job();
+   Job(const Job &other);
+
+   ~Job();
+
+   virtual void evaluate_task(std::size_t task) = 0;
+   virtual void update_real(std::size_t ix, double val, bool is_constant) = 0;
+   virtual void update_bool(std::size_t ix, bool value) = 0;
+   virtual void update_state();
+
+   virtual void send_back_task_result_from_worker(std::size_t task) = 0;
+   virtual void receive_task_result_on_queue(std::size_t task, std::size_t worker_id) = 0;
+   virtual void send_back_results_from_queue_to_master() = 0;
+   // after results have been retrieved, they may need to be cleared to ensure
+   // they won't be retrieved the next time again, e.g. when using a map to
+   // collect results; if not needed it can just be left empty
+   virtual void clear_results() = 0;
+   virtual void receive_results_on_master() = 0;
+   virtual bool receive_task_result_on_master(const zmq::message_t & message) = 0;
+
+   void gather_worker_results();
+
+protected:
+   JobManager *get_manager();
+
+   std::size_t id;
+
+private:
+   // do not use _manager directly, it must first be initialized! use get_manager()
+   JobManager *_manager = nullptr;
+};
+
+} // namespace MultiProcess
+} // namespace RooFit
+
+#endif // ROOT_ROOFIT_MultiProcess_Job_decl

roofit/multiprocess/inc/RooFit/MultiProcess/JobManager.h

@@ -0,0 +1,91 @@
+/*
+ * Project: RooFit
+ * Authors:
+ *   PB, Patrick Bos, Netherlands eScience Center, p.bos@esciencecenter.nl
+ *   IP, Inti Pelupessy, Netherlands eScience Center, i.pelupessy@esciencecenter.nl
+ *
+ * Copyright (c) 2016-2019, Netherlands eScience Center
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ */
+#ifndef ROOT_ROOFIT_MultiProcess_JobManager
+#define ROOT_ROOFIT_MultiProcess_JobManager
+
+#include <memory>  // unique_ptr
+#include <map>
+
+#include "RooFit/MultiProcess/types.h"
+
+namespace RooFit {
+namespace MultiProcess {
+
+// forward definitions
+class ProcessManager;
+class Messenger;
+class Queue;
+class Job;
+
+/*
+ * @brief Main point of access for all MultiProcess infrastructure
+ *
+ * This class mainly serves as the access point to the multi-process infrastructure
+ * for Jobs. It is meant to be used as a singleton that holds and connects the other
+ * infrastructural classes: the messenger, process manager, worker and queue loops.
+ *
+ * It is important that the user of this class, particularly the one that calls
+ * instance() first, calls activate_loops() soon after, because everything that is
+ * done in between instance() and activate_loops() will be executed on all processes.
+ * This may be useful in some cases, but in general, one will probably want to always
+ * use the JobManager in its full capacity, including the queue and worker loops.
+ * This is the way the Job class uses this class, see
+ *
+ * The default number of processes is set using std::thread::hardware_concurrency().
+ * To change it, simply set JobManager::default_N_workers to a different value
+ * before creation of a new instance.
+ */
+
+class JobManager {
+public:
+   static JobManager *instance();
+   static bool is_instantiated();
+
+   ~JobManager();
+
+   static std::size_t add_job_object(Job *job_object);
+   static Job *get_job_object(std::size_t job_object_id);
+   static bool remove_job_object(std::size_t job_object_id);
+
+   ProcessManager & process_manager() const;
+   Messenger & messenger() const;
+   Queue & queue() const;
+
+   void retrieve(std::size_t requesting_job_id);
+   void results_from_queue_to_master();
+
+   void activate();
+   bool is_activated() const;
+
+private:
+   explicit JobManager(std::size_t N_workers);
+
+   std::unique_ptr<ProcessManager> process_manager_ptr;
+   std::unique_ptr<Messenger> messenger_ptr;
+   std::unique_ptr<Queue> queue_ptr;
+   bool activated = false;
+
+   static std::map<std::size_t, Job *> job_objects;
+   static std::size_t job_counter;
+   static std::unique_ptr <JobManager> _instance;
+
+public:
+   static unsigned int default_N_workers;  // no need for getters/setters, just public
+};
+
+} // namespace MultiProcess
+} // namespace RooFit
+
+#endif // ROOT_ROOFIT_MultiProcess_JobManager