Shared mem support / production improvements for o2sim_parallel

This commit provides several important changes towards a production
ready parallel o2sim (as described at CHEP18).

Most notably, this concerns the support of a shared memory treatment
of the hits in order to avoid costly ROOT serializations to communicate between
simulation workers and the hit IO process.
The solution put forward here works by creating a single sufficiantly large shared
memory region in which individual simulation workers get assigned a private space.
The HitMerger can read from these spaces and write hits to file.

Care is being taken in order to ensure that the shared memory is mapped under the same
address space in all processes in order to be able to share complex C++ objects.

If this mapping fails, the system will automatically turn to
using zmq + ROOT serialization mode.

Some changes on the hit types needed to be done:
  - usage of a special allocator
  - move hits to classes instead of "using statements" in order to work with allocators
  - creation/destruction of hits is handled by some special allocation/deallocation functions

The commit contains other minor changes that move the o2sim_parallel a lot
closer to a production ready version.

Beware, that this is a first version. Some future cleanups/refactorings/improvements
are expected. Importantly, we now have a working demonstrator
for sharing complex data amongst multiple processes.

Other minor changes include:
  - make forked sim mode default
  - better signal handling in driver executable
  - add passive detectors by default
  - make chunksize configurable
  - o2sim and o2sim_parallel give identical results due to special seeding
    which is applied for both

Author: sawenzel

Common/SimConfig/include/SimConfig/SimConfig.h

@@ -38,6 +38,8 @@ struct SimConfigData {
                                              // Foo.parameter1=x,Bar.parameter2=y,Baz.paramter3=hello
                                              // (can be used to **loosly** change any configuration parameter from
                                              //  command-line)
+  int mPrimaryChunkSize;                     // defining max granularity for input primaries of a sim job
+  int mInternalChunkSize;
   ClassDefNV(SimConfigData, 1);
 };
 
@@ -93,6 +95,8 @@ class SimConfig
   std::string getLogVerbosity() const { return mConfigData.mLogVerbosity; }
   std::string getLogSeverity() const { return mConfigData.mLogSeverity; }
   std::string getKeyValueString() const { return mConfigData.mKeyValueTokens; }
+  int getPrimChunkSize() const { return mConfigData.mPrimaryChunkSize; }
+  int getInternalChunkSize() const { return mConfigData.mInternalChunkSize; }
 
  private:
   SimConfigData mConfigData; //!

Common/SimConfig/src/SimConfig.cxx

@@ -36,7 +36,7 @@ void SimConfig::initOptions(boost::program_options::options_description& options
     "outPrefix,o", bpo::value<std::string>()->default_value("o2sim"), "prefix of output files")(
     "logseverity", bpo::value<std::string>()->default_value("INFO"), "severity level for FairLogger")(
     "logverbosity", bpo::value<std::string>()->default_value("low"), "level of verbosity for FairLogger (low, medium, high, veryhigh)")(
-    "configKeyValues", bpo::value<std::string>()->default_value(""), "comma separated key=value strings (e.g.: 'TPC.gasDensity=1,...");
+    "configKeyValues", bpo::value<std::string>()->default_value(""), "comma separated key=value strings (e.g.: 'TPC.gasDensity=1,...")("chunkSize", bpo::value<unsigned int>()->default_value(10000), "max size of primary chunk (subevent) distributed by server")("chunkSizeI", bpo::value<int>()->default_value(-1), "internalChunkSize");
 }
 
 bool SimConfig::resetFromParsedMap(boost::program_options::variables_map const& vm)
@@ -50,6 +50,13 @@ bool SimConfig::resetFromParsedMap(boost::program_options::variables_map const&
     for (int d = DetID::First; d <= DetID::Last; ++d) {
       active.emplace_back(DetID::getName(d));
     }
+    // add passive components manually (make a PassiveDetID for them!)
+    active.emplace_back("HALL");
+    active.emplace_back("MAG");
+    active.emplace_back("DIPO");
+    active.emplace_back("PIPE");
+    active.emplace_back("ABSO");
+    active.emplace_back("SHIL");
   }
   // now we take out detectors listed as skipped
   auto& skipped = vm["skipModules"].as<std::vector<std::string>>();
@@ -73,6 +80,8 @@ bool SimConfig::resetFromParsedMap(boost::program_options::variables_map const&
   mConfigData.mLogSeverity = vm["logseverity"].as<std::string>();
   mConfigData.mLogVerbosity = vm["logverbosity"].as<std::string>();
   mConfigData.mKeyValueTokens = vm["configKeyValues"].as<std::string>();
+  mConfigData.mPrimaryChunkSize = vm["chunkSize"].as<unsigned int>();
+  mConfigData.mInternalChunkSize = vm["chunkSizeI"].as<int>();
   return true;
 }
 

Common/Utils/CMakeLists.txt

@@ -7,13 +7,15 @@ set(SRCS
   src/TreeStreamRedirector.cxx
   src/RootChain.cxx
   src/CompStream.cxx
+  src/ShmManager.cxx
 )
 
 Set(HEADERS
   include/${MODULE_NAME}/TreeStream.h
   include/${MODULE_NAME}/TreeStreamRedirector.h
   include/${MODULE_NAME}/RootChain.h
   include/${MODULE_NAME}/BoostSerializer.h
+  include/${MODULE_NAME}/ShmManager.h
 )
 
 Set(LINKDEF src/CommonUtilsLinkDef.h)

Common/Utils/include/CommonUtils/ShmAllocator.h

@@ -0,0 +1,142 @@
+// Copyright CERN and copyright holders of ALICE O2. This software is
+// distributed under the terms of the GNU General Public License v3 (GPL
+// Version 3), copied verbatim in the file "COPYING".
+//
+// See http://alice-o2.web.cern.ch/license for full licensing information.
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+/*
+ * ShmAllocator.h
+ *
+ *  Created on: Jun 18, 2018
+ *      Author: swenzel
+ */
+
+#ifndef COMMON_UTILS_INCLUDE_COMMONUTILS_SHMALLOCATOR_H_
+#define COMMON_UTILS_INCLUDE_COMMONUTILS_SHMALLOCATOR_H_
+
+#include "CommonUtils/ShmManager.h"
+#include <cassert>
+
+namespace o2
+{
+namespace utils
+{
+
+// An allocator placing objects in shared memory as organized by ShmManager.
+// Allocator used in a few std::vector so that they create stuff in shared mem
+// The only non-trivial things are in methods allocate/deallocate
+template <typename T>
+class ShmAllocator
+{
+ public:
+  typedef T value_type;
+  typedef std::size_t size_type;
+  typedef std::ptrdiff_t difference_type;
+
+  typedef T* pointer;
+  typedef const T* const_pointer;
+
+  typedef T& reference;
+  typedef const T& const_reference;
+
+ public:
+  inline ShmAllocator() throw() {}
+
+  template <typename T2>
+  inline ShmAllocator(const ShmAllocator<T2>&) throw()
+  {
+  }
+
+  inline ~ShmAllocator() throw() {}
+
+  inline pointer adress(reference r) { return &r; }
+
+  inline const_pointer adress(const_reference r) const { return &r; }
+
+  // the actually important functions:
+  inline pointer allocate(size_type n)
+  {
+    auto& instance = ShmManager::Instance();
+    if (instance.readyToAllocate()) {
+      return (pointer)ShmManager::Instance().getmemblock(sizeof(value_type) * n);
+    }
+    return (pointer)malloc(sizeof(value_type) * n);
+  }
+  inline void deallocate(pointer p, size_type s)
+  {
+    auto& instance = ShmManager::Instance();
+    if (instance.readyToAllocate()) {
+      ShmManager::Instance().freememblock(p, s);
+    } else {
+      free(p);
+    }
+  }
+
+  inline void construct(pointer p, const value_type& value)
+  {
+    new (p) value_type(value);
+  }
+
+  template <class U, class... Args>
+  void construct(U* p, Args&&... args)
+  {
+    ::new ((void*)p) U(std::forward<Args>(args)...);
+  }
+
+  inline void destroy(pointer p) { p->~value_type(); }
+
+  inline size_type max_size() const throw() { return size_type(-1) / sizeof(value_type); }
+
+  template <typename T2>
+  struct rebind {
+    typedef ShmAllocator<T2> other;
+  };
+
+  bool operator!=(const ShmAllocator<T>& other) const { return !(*this == other); }
+
+  // Returns true if and only if storage allocated from *this
+  // can be deallocated from other, and vice versa.
+  // Always returns true for stateless allocators.
+  bool operator==(const ShmAllocator<T>& /*other*/) const { return true; }
+};
+
+template <typename T>
+std::vector<T>* createSimVector()
+{
+  using vector_t = std::vector<T>;
+  auto& instance = o2::utils::ShmManager::Instance();
+  if (instance.isOperational() && instance.readyToAllocate()) {
+    auto placement = instance.hasSegment() ? instance.getmemblock(sizeof(vector_t)) : malloc(sizeof(vector_t));
+    return new (placement) vector_t;
+  } else {
+    return new vector_t;
+  }
+}
+
+template <typename T>
+void freeSimVector(std::vector<T>* ptr)
+{
+  return;
+  using vector_t = std::vector<T>;
+#ifdef USESHM
+  auto& instance = o2::utils::ShmManager::Instance();
+  ptr->clear();
+  ptr->shrink_to_fit();
+  if (instance.hasSegment() && instance.isPointerOk(ptr)) {
+    instance.freememblock(ptr);
+  } else {
+    free(ptr);
+  }
+// at this moment we have to trust that std::
+#else
+  delete ptr;
+#endif
+}
+
+} // end namespace utils
+} // end namespace o2
+#endif /* COMMON_UTILS_INCLUDE_COMMONUTILS_SHMALLOCATOR_H_ */

Common/Utils/include/CommonUtils/ShmManager.h

@@ -0,0 +1,121 @@
+// Copyright CERN and copyright holders of ALICE O2. This software is
+// distributed under the terms of the GNU General Public License v3 (GPL
+// Version 3), copied verbatim in the file "COPYING".
+//
+// See http://alice-o2.web.cern.ch/license for full licensing information.
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+/*
+ * ShmManager.h
+ *
+ *  Created on: Jun 17, 2018
+ *      Author: swenzel
+ */
+
+#ifndef COMMON_UTILS_INCLUDE_COMMONUTILS_SHMMANAGER_H_
+#define COMMON_UTILS_INCLUDE_COMMONUTILS_SHMMANAGER_H_
+
+#include <list>
+#include <cstddef>
+
+#include <boost/interprocess/managed_external_buffer.hpp>
+#include <boost/interprocess/allocators/allocator.hpp>
+
+#define USESHM 1
+
+namespace o2
+{
+namespace utils
+{
+
+// the size dedicated to each attached worker/process
+constexpr size_t SHMPOOLSIZE = 1024 * 1024 * 200; // 200MB
+
+// some meta info stored at the beginning of the global shared mem segment
+struct ShmMetaInfo {
+  unsigned long long allocedbytes = 0;
+  std::atomic<int> counter = 0; // atomic counter .. counter number of attached processes
+                                // and used to assign a subregion to the attached processes
+  std::atomic<int> failures = 0;
+};
+
+// Class creating -- or attaching to -- a shared memory pool
+// and manages allocations within the pool
+// This is used in the parallel simulation in order
+// to put hits directly in shared mem; I hope this can be replaced/refactored
+// to use directly functionality by FairMQ some day.
+// For the moment a wrapper around boost allocators ... enhancing them with some state.
+class ShmManager
+{
+ public:
+  static ShmManager& Instance()
+  {
+    static ShmManager instance;
+    return instance;
+  }
+
+  // creates a global shared mem region
+  // to be used by "nsubsegments" simulation processes
+  bool createGlobalSegment(int nsubsegments = 1);
+
+  // create the local segment
+  // this will occupy a subregion of an already created global shared mem segment
+  void occupySegment();
+
+  // simply attaches to the global segment
+  bool attachToGlobalSegment();
+
+  // the equivalent of malloc
+  void* getmemblock(size_t size);
+  // the equivalent of free
+  void freememblock(void*, std::size_t = 1);
+
+  void release();
+  int getShmID() const { return mShmID; }
+  bool hasSegment() const { return mShmID != -1; }
+  bool readyToAllocate() const { return mShmID != -1 && mBufferPtr; }
+
+  // returns if pointer is part of the shm region under control of this manager
+  bool isPointerOk(void* ptr) const
+  {
+    return mBufferPtr && getPointerOffset(ptr) < SHMPOOLSIZE;
+  }
+
+  // returns if shared mem setup is correctly setup/operational
+  // used to decide whether to communicate via shared mem at runtime or via
+  // TMessages /etc/
+  bool isOperational() const { return mSegInfoPtr && mSegInfoPtr->failures == 0; /* mIsOperational; */ }
+
+  void disable()
+  {
+    if (mSegInfoPtr) {
+      mSegInfoPtr->failures.fetch_add(1);
+    };
+  }
+
+  void printSegInfo() const;
+
+ private:
+  ShmManager();
+  ~ShmManager();
+  int mShmID = -1;                    // id of shared mem created or used
+  void* mBufferPtr = nullptr;         // the mapped/start ptr of the buffer to use
+  void* mSegPtr = nullptr;            // address of the segment start
+  ShmMetaInfo* mSegInfoPtr = nullptr; // pointing to the meta information object
+  bool mIsMaster = false;             // true if the manager who allocated the region
+  bool mIsOperational = false;
+  // helper function
+  void* tryAttach(bool& success);
+  size_t getPointerOffset(void* ptr) const { return (size_t)((char*)ptr - (char*)mBufferPtr); }
+
+  boost::interprocess::wmanaged_external_buffer* boostmanagedbuffer;
+  boost::interprocess::allocator<char, boost::interprocess::wmanaged_external_buffer::segment_manager>* boostallocator;
+};
+
+} // namespace utils
+} // namespace o2
+
+#endif /* COMMON_UTILS_INCLUDE_COMMONUTILS_SHMMANAGER_H_ */