mirror of
https://github.com/FairRootGroup/FairMQ.git
synced 2025-10-15 17:41:45 +00:00
Compare commits
30 Commits
Author | SHA1 | Date | |
---|---|---|---|
|
d105960444 | ||
|
3aae5bae58 | ||
|
9031029d2c | ||
|
d478e050ba | ||
|
06b2b9b01f | ||
|
b3fa4f6e7e | ||
|
da5cb34416 | ||
|
226733c653 | ||
|
b06efc401e | ||
|
2500771689 | ||
|
d2aa3b6bb0 | ||
|
00df117c7c | ||
|
69faa63c5b | ||
|
b7474ae138 | ||
|
b426bf39d7 | ||
|
6780b7452c | ||
|
27277b11b4 | ||
|
cb5029f826 | ||
|
5d45d89269 | ||
|
eb9ddc81cf | ||
|
f5891d5ae3 | ||
|
3b2ad1f6f4 | ||
|
fa0bf96eb2 | ||
|
29827f0426 | ||
|
8efe7adf0e | ||
|
b747a8787c | ||
|
1a75141fc4 | ||
|
2f82eb4f09 | ||
|
92a56c26bc | ||
|
4f9aeda8ec |
86
.zenodo.json
Normal file
86
.zenodo.json
Normal file
@@ -0,0 +1,86 @@
|
||||
{
|
||||
"creators": [
|
||||
{
|
||||
"name": "Al-Turany, Mohammad"
|
||||
},
|
||||
{
|
||||
"orcid": "0000-0003-3787-1910",
|
||||
"name": "Klein, Dennis"
|
||||
},
|
||||
{
|
||||
"name": "Kollegger, Thorsten"
|
||||
},
|
||||
{
|
||||
"name": "Rybalchenko, Alexey"
|
||||
},
|
||||
{
|
||||
"name": "Winckler, Nicolas"
|
||||
}
|
||||
],
|
||||
"contributors": [
|
||||
{
|
||||
"type": "Other",
|
||||
"name": "Aphecetche, Laurent"
|
||||
},
|
||||
{
|
||||
"type": "Other",
|
||||
"name": "Binet, Sebastien"
|
||||
},
|
||||
{
|
||||
"type": "Other",
|
||||
"name": "Eulisse, Giulio"
|
||||
},
|
||||
{
|
||||
"type": "Other",
|
||||
"name": "Karabowicz, Radoslaw"
|
||||
},
|
||||
{
|
||||
"type": "Other",
|
||||
"name": "Kretz, Matthias"
|
||||
},
|
||||
{
|
||||
"type": "Other",
|
||||
"name": "Krzewicki, Mikolaj"
|
||||
},
|
||||
{
|
||||
"type": "Other",
|
||||
"name": "Lebedev, Andrey"
|
||||
},
|
||||
{
|
||||
"type": "Other",
|
||||
"name": "Mrnjavac, Teo"
|
||||
},
|
||||
{
|
||||
"type": "Other",
|
||||
"name": "Neskovic, Gvozden"
|
||||
},
|
||||
{
|
||||
"type": "Other",
|
||||
"name": "Richter, Matthias"
|
||||
},
|
||||
{
|
||||
"type": "Other",
|
||||
"orcid": "0000-0002-5321-8404",
|
||||
"name": "Tacke, Christian"
|
||||
},
|
||||
{
|
||||
"type": "Other",
|
||||
"name": "Uhlig, Florian"
|
||||
},
|
||||
{
|
||||
"type": "Other",
|
||||
"name": "Wenzel, Sandro"
|
||||
}
|
||||
],
|
||||
"description": "<p>C++ Message Queuing Library and Framework</p>",
|
||||
"related_identifiers": [
|
||||
{
|
||||
"identifier": "https://github.com/FairRootGroup/FairMQ/",
|
||||
"relation": "isSupplementTo",
|
||||
"resource_type": "software",
|
||||
"scheme": "url"
|
||||
}
|
||||
],
|
||||
"title": "FairMQ",
|
||||
"license": "LGPL-3.0-only"
|
||||
}
|
2
AUTHORS
2
AUTHORS
@@ -1,5 +1,5 @@
|
||||
Al-Turany, Mohammad
|
||||
Klein, Dennis
|
||||
Klein, Dennis [https://orcid.org/0000-0003-3787-1910]
|
||||
Kollegger, Thorsten
|
||||
Rybalchenko, Alexey
|
||||
Winckler, Nicolas
|
||||
|
@@ -1,5 +1,5 @@
|
||||
################################################################################
|
||||
# Copyright (C) 2018-2021 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH #
|
||||
# Copyright (C) 2018-2022 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH #
|
||||
# #
|
||||
# This software is distributed under the terms of the #
|
||||
# GNU Lesser General Public Licence (LGPL) version 3, #
|
||||
@@ -9,7 +9,7 @@
|
||||
|
||||
# Project ######################################################################
|
||||
cmake_minimum_required(VERSION 3.15 FATAL_ERROR)
|
||||
cmake_policy(VERSION 3.15...3.20)
|
||||
cmake_policy(VERSION 3.15...3.22)
|
||||
|
||||
list(PREPEND CMAKE_MODULE_PATH ${CMAKE_SOURCE_DIR}/cmake)
|
||||
include(GitHelper)
|
||||
|
@@ -8,6 +8,6 @@ Lebedev, Andrey
|
||||
Mrnjavac, Teo <teo.m@cern.ch>
|
||||
Neskovic, Gvozden
|
||||
Richter, Matthias
|
||||
Tacke, Christian
|
||||
Tacke, Christian [https://orcid.org/0000-0002-5321-8404]
|
||||
Uhlig, Florian
|
||||
Wenzel, Sandro
|
||||
|
10
COPYRIGHT
10
COPYRIGHT
@@ -4,19 +4,19 @@ Upstream-Contact: Mohammad Al-Turany <m.al-turany@gsi.de>
|
||||
Source: https://github.com/FairRootGroup/FairMQ
|
||||
|
||||
Files: *
|
||||
Copyright: 2012-2021, GSI Helmholtzzentrum fuer Schwerionenforschung GmbH
|
||||
Copyright: 2012-2021, [see AUTHORS file]
|
||||
Copyright: 2012-2021, [see CONTRIBUTORS file]
|
||||
Copyright: 2012-2022, GSI Helmholtzzentrum fuer Schwerionenforschung GmbH
|
||||
Copyright: 2012-2022, [see AUTHORS file]
|
||||
Copyright: 2012-2022, [see CONTRIBUTORS file]
|
||||
Comment: The copyright of individual contributors is documented in the
|
||||
Git history.
|
||||
License: LGPL-3.0-only
|
||||
|
||||
Files: extern/googletest
|
||||
Copyright: 2008-2021, Google Inc.
|
||||
Copyright: 2008-2022, Google Inc.
|
||||
License: GOOGLE
|
||||
|
||||
Files: extern/asio
|
||||
Copyright: 2003-2021, Christopher M. Kohlhoff (chris at kohlhoff dot com)
|
||||
Copyright: 2003-2022, Christopher M. Kohlhoff (chris at kohlhoff dot com)
|
||||
License: BSL-1.0
|
||||
|
||||
Files: extern/PicoSHA2
|
||||
|
@@ -72,6 +72,9 @@ endif()
|
||||
if(ENABLE_SANITIZER_THREAD)
|
||||
list(APPEND options "-DENABLE_SANITIZER_THREAD=ON")
|
||||
endif()
|
||||
if(CMAKE_CXX_COMPILER)
|
||||
list(APPEND options "-DCMAKE_CXX_COMPILER=${CMAKE_CXX_COMPILER}")
|
||||
endif()
|
||||
if(CMAKE_CXX_FLAGS)
|
||||
list(APPEND options "-DCMAKE_CXX_FLAGS=${CMAKE_CXX_FLAGS}")
|
||||
endif()
|
||||
|
6
Jenkinsfile
vendored
6
Jenkinsfile
vendored
@@ -15,9 +15,9 @@ def jobMatrix(String type, List specs) {
|
||||
ver = spec.ver
|
||||
} else { // == 'check'
|
||||
job = "${spec.name}"
|
||||
selector = 'fedora-34-x86_64'
|
||||
selector = 'fedora-35-x86_64'
|
||||
os = 'fedora'
|
||||
ver = '34'
|
||||
ver = '35'
|
||||
}
|
||||
|
||||
def label = "${job}"
|
||||
@@ -96,6 +96,7 @@ pipeline{
|
||||
[os: 'fedora', ver: '32', arch: 'x86_64', compiler: 'gcc-10', extra: all],
|
||||
[os: 'fedora', ver: '33', arch: 'x86_64', compiler: 'gcc-10', extra: all],
|
||||
[os: 'fedora', ver: '34', arch: 'x86_64', compiler: 'gcc-11', extra: all],
|
||||
[os: 'fedora', ver: '35', arch: 'x86_64', compiler: 'gcc-11', extra: all],
|
||||
[os: 'macos', ver: '11', arch: 'x86_64', compiler: 'apple-clang-12', extra: '-DHAS_ASIO=ON'],
|
||||
[os: 'macos', ver: '11', arch: 'arm64', compiler: 'apple-clang-13', extra: '-DHAS_ASIO=ON'],
|
||||
])
|
||||
@@ -106,6 +107,7 @@ pipeline{
|
||||
[name: 'static-analyzers', extra: "${all_debug} -DRUN_STATIC_ANALYSIS=ON"],
|
||||
[name: '{address,leak,ub}-sanitizers',
|
||||
extra: "${all_debug} -DENABLE_SANITIZER_ADDRESS=ON -DENABLE_SANITIZER_LEAK=ON -DENABLE_SANITIZER_UNDEFINED_BEHAVIOUR=ON -DCMAKE_CXX_FLAGS='-O1 -fno-omit-frame-pointer'"],
|
||||
[name: 'thread-sanitizer', extra: "${all_debug} -DENABLE_SANITIZER_THREAD=ON -DCMAKE_CXX_COMPILER=clang++"],
|
||||
])
|
||||
|
||||
parallel(builds + checks)
|
||||
|
32
README.md
32
README.md
@@ -1,12 +1,9 @@
|
||||
<!-- {#mainpage} -->
|
||||
# FairMQ [](COPYRIGHT) [](https://alfa-ci.gsi.de/blue/organizations/jenkins/FairRootGroup%2FFairMQ/branches) [](https://scan.coverity.com/projects/fairrootgroup-fairmq)
|
||||
# FairMQ [](COPYRIGHT)
|
||||
|
||||
C++ Message Queuing Library and Framework
|
||||
|
||||
| Release | Version | Docs |
|
||||
| :---: | :--- | :--- |
|
||||
| `stable` | [](https://github.com/FairRootGroup/FairMQ/releases/latest) | [API](https://fairrootgroup.github.io/FairMQ/latest), [Book](https://github.com/FairRootGroup/FairMQ/blob/master/README.md#documentation) |
|
||||
| `testing` | [](https://github.com/FairRootGroup/FairMQ/tags) | [Book](https://github.com/FairRootGroup/FairMQ/blob/dev/README.md#documentation) |
|
||||
Docs: [Book](https://github.com/FairRootGroup/FairMQ/blob/dev/README.md#documentation)
|
||||
|
||||
Find all FairMQ releases [here](https://github.com/FairRootGroup/FairMQ/releases).
|
||||
|
||||
@@ -24,11 +21,13 @@ FairMQ provides multiple implementations for its API (so-called "transports",
|
||||
e.g. `zeromq`, `shmem` and `ofi` (in development)) to cover a variety of use cases
|
||||
(e.g. inter-thread, inter-process, inter-node communication) and machines (e.g. Ethernet, Infiniband).
|
||||
In addition to this core functionality FairMQ provides a framework for creating "devices" - actors which
|
||||
are communicating through message passing. FairMQ does not only allow the user to use different transport but also to mix them; i.e: A Device can communicate using different transport on different channels at the same time. Device execution is modelled as a simple state machine that
|
||||
shapes the integration points for the user task. Devices also incorporate a plugin system for runtime configuration and control.
|
||||
Next to the provided devices and plugins (e.g. [DDS](https://github.com/FairRootGroup/DDS))
|
||||
the user can extend FairMQ by developing his own plugins to integrate his devices with external
|
||||
configuration and control services.
|
||||
are communicating through message passing. FairMQ does not only allow the user to use different transport
|
||||
but also to mix them; i.e: A Device can communicate using different transport on different channels at the
|
||||
same time. Device execution is modelled as a simple state machine that shapes the integration points for
|
||||
the user task. Devices also incorporate a plugin system for runtime configuration and control.
|
||||
Next to the provided [devices](https://github.com/FairRootGroup/FairMQ/tree/master/fairmq/devices) and
|
||||
[plugins](https://github.com/FairRootGroup/FairMQ/tree/master/fairmq/plugins) the user can extend FairMQ
|
||||
by developing his own plugins to integrate his devices with external configuration and control services.
|
||||
|
||||
FairMQ has been developed in the context of its mother project [FairRoot](https://github.com/FairRootGroup/FairRoot) -
|
||||
a simulation, reconstruction and analysis framework.
|
||||
@@ -47,14 +46,15 @@ cmake --build fairmq_build --target install
|
||||
|
||||
Please consult the [manpages of your CMake version](https://cmake.org/cmake/help/latest/manual/cmake.1.html) for more options.
|
||||
|
||||
If dependencies are not installed in standard system directories, you can hint the installation location via `-DCMAKE_PREFIX_PATH=...` or per dependency via `-D{DEPENDENCY}_ROOT=...`. `{DEPENDENCY}` can be `GTEST`, `BOOST`, `FAIRLOGGER`, `ZEROMQ`, `OFI`, `PMIX`, `ASIO`, `ASIOFI` or `DDS` (`*_ROOT` variables can also be environment variables).
|
||||
If dependencies are not installed in standard system directories, you can hint the installation location via
|
||||
`-DCMAKE_PREFIX_PATH=...` or per dependency via `-D{DEPENDENCY}_ROOT=...` (`*_ROOT` variables can also be environment variables).
|
||||
|
||||
## Usage
|
||||
|
||||
FairMQ ships as a CMake package, so in your `CMakeLists.txt` you can discover it like this:
|
||||
|
||||
```cmake
|
||||
find_package(FairCMakeModules 0.2 REQUIRED)
|
||||
find_package(FairCMakeModules 1.0 REQUIRED)
|
||||
include(FairFindPackage2)
|
||||
find_package2(FairMQ)
|
||||
find_package2_implicit_dependencies()
|
||||
@@ -71,14 +71,14 @@ list(PREPEND CMAKE_PREFIX_PATH /path/to/fairmq_install)
|
||||
Optionally, you can require certain FairMQ package components and a minimum version:
|
||||
|
||||
```cmake
|
||||
find_package(FairMQ 1.4.0 COMPONENTS dds_plugin)
|
||||
find_package(FairMQ 1.4.50 COMPONENTS ofi_transport)
|
||||
```
|
||||
|
||||
When building FairMQ, CMake will print a summary table of all available package components.
|
||||
|
||||
## Dependencies
|
||||
|
||||
* [asio](https://github.com/chriskohlhoff/asio) (optionally bundled)
|
||||
* [asio](https://github.com/chriskohlhoff/asio)
|
||||
* [asiofi](https://github.com/FairRootGroup/asiofi)
|
||||
* [Boost](https://www.boost.org/)
|
||||
* [CMake](https://cmake.org/)
|
||||
@@ -86,13 +86,14 @@ When building FairMQ, CMake will print a summary table of all available package
|
||||
* [Doxygen](http://www.doxygen.org/)
|
||||
* [FairCMakeModules](https://github.com/FairRootGroup/FairCMakeModules) (optionally bundled)
|
||||
* [FairLogger](https://github.com/FairRootGroup/FairLogger)
|
||||
* [Flatbuffers](https://google.github.io/flatbuffers/)
|
||||
* [GTest](https://github.com/google/googletest) (optionally bundled)
|
||||
* [PMIx](https://pmix.org/)
|
||||
* [ZeroMQ](http://zeromq.org/)
|
||||
|
||||
Which dependencies are required depends on which components are built.
|
||||
|
||||
Supported platforms: Linux and MacOS.
|
||||
Supported platform is Linux. macOS is supported on a best-effort basis.
|
||||
|
||||
## CMake options
|
||||
|
||||
@@ -102,7 +103,6 @@ On command line:
|
||||
* `-DBUILD_TESTING=OFF` disables building of tests.
|
||||
* `-DBUILD_EXAMPLES=OFF` disables building of examples.
|
||||
* `-DBUILD_OFI_TRANSPORT=ON` enables building of the experimental OFI transport.
|
||||
* `-DBUILD_DDS_PLUGIN=ON` enables building of the DDS plugin.
|
||||
* `-DBUILD_PMIX_PLUGIN=ON` enables building of the PMIx plugin.
|
||||
* `-DBUILD_DOCS=ON` enables building of API docs.
|
||||
* You can hint non-system installations for dependent packages, see the #installation-from-source section above
|
||||
|
@@ -34,9 +34,9 @@ macro(fairmq_summary_components)
|
||||
endif()
|
||||
message(STATUS " ${BWhite}ofi_transport${CR} ${ofi_summary}")
|
||||
if(BUILD_DDS_PLUGIN)
|
||||
set(dds_summary "${BGreen}YES${CR} (disable with ${BMagenta}-DBUILD_DDS_PLUGIN=OFF${CR})")
|
||||
set(dds_summary "${BGreen}YES${CR} DEPRECATED (disable with ${BMagenta}-DBUILD_DDS_PLUGIN=OFF${CR})")
|
||||
else()
|
||||
set(dds_summary "${BRed} NO${CR} (default, enable with ${BMagenta}-DBUILD_DDS_PLUGIN=ON${CR})")
|
||||
set(dds_summary "${BRed} NO${CR} DEPRECATED (default, enable with ${BMagenta}-DBUILD_DDS_PLUGIN=ON${CR})")
|
||||
endif()
|
||||
message(STATUS " ${BWhite}dds_plugin${CR} ${dds_summary}")
|
||||
if(BUILD_PMIX_PLUGIN)
|
||||
@@ -58,15 +58,15 @@ macro(fairmq_summary_components)
|
||||
endif()
|
||||
message(STATUS " ${BWhite}docs${CR} ${docs_summary}")
|
||||
if(BUILD_SDK)
|
||||
set(sdk_summary "${BGreen}YES${CR} EXPERIMENTAL (disable with ${BMagenta}-DBUILD_SDK=OFF${CR})")
|
||||
set(sdk_summary "${BGreen}YES${CR} DEPRECATED (disable with ${BMagenta}-DBUILD_SDK=OFF${CR})")
|
||||
else()
|
||||
set(sdk_summary "${BRed} NO${CR} EXPERIMENTAL (default, enable with ${BMagenta}-DBUILD_SDK=ON${CR})")
|
||||
set(sdk_summary "${BRed} NO${CR} DEPRECATED (default, enable with ${BMagenta}-DBUILD_SDK=ON${CR})")
|
||||
endif()
|
||||
message(STATUS " ${BWhite}sdk${CR} ${sdk_summary}")
|
||||
if(BUILD_SDK_COMMANDS)
|
||||
set(sdk_commands_summary "${BGreen}YES${CR} (disable with ${BMagenta}-DBUILD_SDK_COMMANDS=OFF${CR})")
|
||||
set(sdk_commands_summary "${BGreen}YES${CR} DEPRECATED (disable with ${BMagenta}-DBUILD_SDK_COMMANDS=OFF${CR})")
|
||||
else()
|
||||
set(sdk_commands_summary "${BRed} NO${CR} (default, enable with ${BMagenta}-DBUILD_SDK_COMMANDS=ON${CR})")
|
||||
set(sdk_commands_summary "${BRed} NO${CR} DEPRECATED (default, enable with ${BMagenta}-DBUILD_SDK_COMMANDS=ON${CR})")
|
||||
endif()
|
||||
message(STATUS " ${BWhite}sdk_commands${CR} ${sdk_commands_summary}")
|
||||
if(BUILD_TIDY_TOOL)
|
||||
@@ -75,6 +75,21 @@ macro(fairmq_summary_components)
|
||||
set(sdk_tidy_summary "${BRed} NO${CR} (default, enable with ${BMagenta}-DBUILD_TIDY_TOOL=ON${CR})")
|
||||
endif()
|
||||
message(STATUS " ${BWhite}tidy_tool${CR} ${sdk_tidy_summary}")
|
||||
|
||||
set(_deprecated)
|
||||
if(BUILD_SDK)
|
||||
list(APPEND _deprecated sdk)
|
||||
endif()
|
||||
if(BUILD_SDK_COMMANDS)
|
||||
list(APPEND _deprecated sdk_commands)
|
||||
endif()
|
||||
if(BUILD_DDS_PLUGIN)
|
||||
list(APPEND _deprecated dds_plugin)
|
||||
endif()
|
||||
list(JOIN _deprecated ", " _deprecated)
|
||||
if(_deprecated)
|
||||
message(DEPRECATION "You have selected to build deprecated components '${_deprecated}' which will be removed in a future release. See https://github.com/FairRootGroup/FairMQ/discussions/392 for more information. Use '-Wno-deprecated' to silence deprecation warnings.")
|
||||
endif()
|
||||
endmacro()
|
||||
|
||||
macro(fairmq_summary_static_analysis)
|
||||
|
@@ -18,7 +18,8 @@
|
||||
{
|
||||
"@type": "Person",
|
||||
"givenName": "Dennis",
|
||||
"familyName": "Klein"
|
||||
"familyName": "Klein",
|
||||
"@id": "https://orcid.org/0000-0003-3787-1910"
|
||||
},
|
||||
{
|
||||
"@type": "Person",
|
||||
@@ -92,7 +93,8 @@
|
||||
{
|
||||
"@type": "Person",
|
||||
"givenName": "Christian",
|
||||
"familyName": "Tacke"
|
||||
"familyName": "Tacke",
|
||||
"@id": "https://orcid.org/0000-0002-5321-8404"
|
||||
},
|
||||
{
|
||||
"@type": "Person",
|
||||
|
@@ -26,7 +26,6 @@ Here is an overview of the device/channel options and when they are applied:
|
||||
| `transport` | at the end of `fair::mq::State::InitializingDevice` |
|
||||
| `network-interface` | at the end of `fair::mq::State::InitializingDevice` |
|
||||
| `init-timeout` | at the end of `fair::mq::State::InitializingDevice` |
|
||||
| `max-run-time` | at the end of `fair::mq::State::InitializingDevice` |
|
||||
| `shm-segment-size` | at the end of `fair::mq::State::InitializingDevice` |
|
||||
| `shm-monitor` | at the end of `fair::mq::State::InitializingDevice` |
|
||||
| `ofi-size-hint` | at the end of `fair::mq::State::InitializingDevice` |
|
||||
|
@@ -6,10 +6,9 @@
|
||||
* copied verbatim in the file "LICENSE" *
|
||||
********************************************************************************/
|
||||
#include <fairmq/shmem/Common.h>
|
||||
#include <fairmq/shmem/UnmanagedRegion.h>
|
||||
#include <fairmq/shmem/Segment.h>
|
||||
#include <fairmq/shmem/Monitor.h>
|
||||
|
||||
#include <fairmq/shmem/Segment.h>
|
||||
#include <fairmq/shmem/UnmanagedRegion.h>
|
||||
#include <fairmq/tools/Unique.h>
|
||||
|
||||
#include <fairlogger/Logger.h>
|
||||
@@ -17,9 +16,8 @@
|
||||
#include <boost/algorithm/string.hpp>
|
||||
#include <boost/program_options.hpp>
|
||||
|
||||
#include <csignal>
|
||||
|
||||
#include <chrono>
|
||||
#include <csignal>
|
||||
#include <map>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
@@ -27,65 +25,117 @@
|
||||
using namespace std;
|
||||
using namespace boost::program_options;
|
||||
|
||||
namespace
|
||||
{
|
||||
namespace {
|
||||
volatile sig_atomic_t gStopping = 0;
|
||||
}
|
||||
volatile sig_atomic_t gResetContent = 0;
|
||||
} // namespace
|
||||
|
||||
void signalHandler(int /* signal */)
|
||||
{
|
||||
gStopping = 1;
|
||||
}
|
||||
void signalHandler(int /* signal */) { gStopping = 1; }
|
||||
|
||||
void resetContentHandler(int /* signal */) { gResetContent = 1; }
|
||||
|
||||
struct ShmManager
|
||||
{
|
||||
ShmManager(uint64_t _shmId, const vector<string>& _segments, const vector<string>& _regions)
|
||||
ShmManager(uint64_t _shmId, const vector<string>& _segments, const vector<string>& _regions, bool zero = true)
|
||||
: shmId(fair::mq::shmem::makeShmIdStr(_shmId))
|
||||
{
|
||||
for (const auto& s : _segments) {
|
||||
vector<string> segmentConf;
|
||||
boost::algorithm::split(segmentConf, s, boost::algorithm::is_any_of(","));
|
||||
if (segmentConf.size() != 2) {
|
||||
LOG(error) << "incorrect format for --segments. Expecting pairs of <id>,<size>.";
|
||||
fair::mq::shmem::Monitor::Cleanup(fair::mq::shmem::ShmId{shmId});
|
||||
throw runtime_error("incorrect format for --segments. Expecting pairs of <id>,<size>.");
|
||||
LOG(info) << "Starting ShmManager for shmId: " << shmId;
|
||||
LOG(info) << "Performing full reset...";
|
||||
FullReset();
|
||||
LOG(info) << "Done.";
|
||||
LOG(info) << "Adding managed segments...";
|
||||
AddSegments(_segments, zero);
|
||||
LOG(info) << "Done.";
|
||||
LOG(info) << "Adding unmanaged regions...";
|
||||
AddRegions(_regions, zero);
|
||||
LOG(info) << "Done.";
|
||||
LOG(info) << "Shared memory is ready for use.";
|
||||
}
|
||||
uint16_t id = stoi(segmentConf.at(0));
|
||||
uint64_t size = stoull(segmentConf.at(1));
|
||||
|
||||
void AddSegments(const vector<string>& _segments, bool zero)
|
||||
{
|
||||
for (const auto& s : _segments) {
|
||||
vector<string> conf;
|
||||
boost::algorithm::split(conf, s, boost::algorithm::is_any_of(","));
|
||||
if (conf.size() != 3) {
|
||||
LOG(error) << "incorrect format for --segments. Expecting pairs of <id>,<size><numaid>.";
|
||||
fair::mq::shmem::Monitor::Cleanup(fair::mq::shmem::ShmId{shmId});
|
||||
throw runtime_error("incorrect format for --segments. Expecting pairs of <id>,<size>,<numaid>.");
|
||||
}
|
||||
uint16_t id = stoi(conf.at(0));
|
||||
uint64_t size = stoull(conf.at(1));
|
||||
segmentCfgs.emplace_back(fair::mq::shmem::SegmentConfig{id, size, "rbtree_best_fit"});
|
||||
|
||||
auto ret = segments.emplace(id, fair::mq::shmem::Segment(shmId, id, size, fair::mq::shmem::rbTreeBestFit));
|
||||
fair::mq::shmem::Segment& segment = ret.first->second;
|
||||
LOG(info) << "Created segment " << id << " of size " << segment.GetSize() << ", starting at " << segment.GetData() << ". Locking...";
|
||||
LOG(info) << "Created segment " << id << " of size " << segment.GetSize()
|
||||
<< ", starting at " << segment.GetData() << ". Locking...";
|
||||
segment.Lock();
|
||||
LOG(info) << "Done.";
|
||||
if (zero) {
|
||||
LOG(info) << "Zeroing...";
|
||||
segment.Zero();
|
||||
LOG(info) << "Done.";
|
||||
}
|
||||
|
||||
for (const auto& r : _regions) {
|
||||
vector<string> regionConf;
|
||||
boost::algorithm::split(regionConf, r, boost::algorithm::is_any_of(","));
|
||||
if (regionConf.size() != 2) {
|
||||
LOG(error) << "incorrect format for --regions. Expecting pairs of <id>,<size>.";
|
||||
fair::mq::shmem::Monitor::Cleanup(fair::mq::shmem::ShmId{shmId});
|
||||
throw runtime_error("incorrect format for --regions. Expecting pairs of <id>,<size>.");
|
||||
}
|
||||
uint16_t id = stoi(regionConf.at(0));
|
||||
uint64_t size = stoull(regionConf.at(1));
|
||||
}
|
||||
|
||||
void AddRegions(const vector<string>& _regions, bool zero)
|
||||
{
|
||||
for (const auto& r : _regions) {
|
||||
vector<string> conf;
|
||||
boost::algorithm::split(conf, r, boost::algorithm::is_any_of(","));
|
||||
if (conf.size() != 3) {
|
||||
LOG(error) << "incorrect format for --regions. Expecting pairs of <id>,<size>,<numaid>.";
|
||||
fair::mq::shmem::Monitor::Cleanup(fair::mq::shmem::ShmId{shmId});
|
||||
throw runtime_error("incorrect format for --regions. Expecting pairs of <id>,<size>,<numaid>.");
|
||||
}
|
||||
uint16_t id = stoi(conf.at(0));
|
||||
uint64_t size = stoull(conf.at(1));
|
||||
fair::mq::RegionConfig cfg;
|
||||
cfg.id = id;
|
||||
cfg.size = size;
|
||||
regionCfgs.push_back(cfg);
|
||||
|
||||
auto ret = regions.emplace(id, make_unique<fair::mq::shmem::UnmanagedRegion>(shmId, id, size));
|
||||
fair::mq::shmem::UnmanagedRegion& region = *(ret.first->second);
|
||||
LOG(info) << "Created unamanged region " << id << " of size " << region.GetSize() << ", starting at " << region.GetData() << ". Locking...";
|
||||
LOG(info) << "Created unamanged region " << id << " of size " << region.GetSize()
|
||||
<< ", starting at " << region.GetData() << ". Locking...";
|
||||
region.Lock();
|
||||
LOG(info) << "Done.";
|
||||
if (zero) {
|
||||
LOG(info) << "Zeroing...";
|
||||
region.Zero();
|
||||
LOG(info) << "Done.";
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
bool CheckPresence()
|
||||
{
|
||||
for (const auto& sc : segmentCfgs) {
|
||||
if (!(fair::mq::shmem::Monitor::SegmentIsPresent(fair::mq::shmem::ShmId{shmId}, sc.id))) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
for (const auto& rc : regionCfgs) {
|
||||
if (!(fair::mq::shmem::Monitor::RegionIsPresent(fair::mq::shmem::ShmId{shmId}, rc.id.value()))) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
void ResetContent()
|
||||
{
|
||||
fair::mq::shmem::Monitor::ResetContent(fair::mq::shmem::ShmId{shmId});
|
||||
fair::mq::shmem::Monitor::ResetContent(fair::mq::shmem::ShmId{shmId}, segmentCfgs, regionCfgs);
|
||||
}
|
||||
|
||||
void FullReset()
|
||||
{
|
||||
segments.clear();
|
||||
regions.clear();
|
||||
fair::mq::shmem::Monitor::Cleanup(fair::mq::shmem::ShmId{shmId});
|
||||
}
|
||||
|
||||
~ShmManager()
|
||||
@@ -97,6 +147,8 @@ struct ShmManager
|
||||
std::string shmId;
|
||||
map<uint16_t, fair::mq::shmem::Segment> segments;
|
||||
map<uint16_t, unique_ptr<fair::mq::shmem::UnmanagedRegion>> regions;
|
||||
std::vector<fair::mq::shmem::SegmentConfig> segmentCfgs;
|
||||
std::vector<fair::mq::RegionConfig> regionCfgs;
|
||||
};
|
||||
|
||||
int main(int argc, char** argv)
|
||||
@@ -105,8 +157,11 @@ int main(int argc, char** argv)
|
||||
|
||||
signal(SIGINT, signalHandler);
|
||||
signal(SIGTERM, signalHandler);
|
||||
signal(SIGUSR1, resetContentHandler);
|
||||
|
||||
try {
|
||||
bool nozero = false;
|
||||
bool checkPresence = true;
|
||||
uint64_t shmId = 0;
|
||||
vector<string> segments;
|
||||
vector<string> regions;
|
||||
@@ -114,8 +169,10 @@ int main(int argc, char** argv)
|
||||
options_description desc("Options");
|
||||
desc.add_options()
|
||||
("shmid", value<uint64_t>(&shmId)->required(), "Shm id")
|
||||
("segments", value<vector<string>>(&segments)->multitoken()->composing(), "Segments, as <id>,<size> <id>,<size> <id>,<size> ...")
|
||||
("regions", value<vector<string>>(®ions)->multitoken()->composing(), "Regions, as <id>,<size> <id>,<size> <id>,<size> ...")
|
||||
("segments", value<vector<string>>(&segments)->multitoken()->composing(), "Segments, as <id>,<size>,<numaid> <id>,<size>,<numaid> <id>,<size>,<numaid> ... (numaid: -2 disabled, -1 interleave, >=0 node)")
|
||||
("regions", value<vector<string>>(®ions)->multitoken()->composing(), "Regions, as <id>,<size> <id>,<size>,<numaid> <id>,<size>,<numaid> ...")
|
||||
("nozero", value<bool>(&nozero)->default_value(false)->implicit_value(true), "Do not zero segments after initialization")
|
||||
("check-presence", value<bool>(&checkPresence)->default_value(true)->implicit_value(true), "Check periodically if configured segments/regions are still present, and cleanup and leave if they are not")
|
||||
("help,h", "Print help");
|
||||
|
||||
variables_map vm;
|
||||
@@ -128,15 +185,35 @@ int main(int argc, char** argv)
|
||||
|
||||
notify(vm);
|
||||
|
||||
ShmManager shmManager(shmId, segments, regions);
|
||||
ShmManager shmManager(shmId, segments, regions, !nozero);
|
||||
|
||||
std::thread resetContentThread([&shmManager]() {
|
||||
while (!gStopping) {
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(50));
|
||||
if (gResetContent == 1) {
|
||||
LOG(info) << "Resetting content for shmId " << shmManager.shmId;
|
||||
shmManager.ResetContent();
|
||||
gResetContent = 0;
|
||||
LOG(info) << "Done resetting content for shmId " << shmManager.shmId;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
if (checkPresence) {
|
||||
while (!gStopping) {
|
||||
if (shmManager.CheckPresence() == false) {
|
||||
LOG(error) << "Failed to find segments, exiting.";
|
||||
gStopping = true;
|
||||
}
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(500));
|
||||
}
|
||||
}
|
||||
|
||||
resetContentThread.join();
|
||||
|
||||
LOG(info) << "stopping.";
|
||||
} catch (exception& e) {
|
||||
LOG(error) << "Unhandled Exception reached the top of main: " << e.what() << ", application will now exit";
|
||||
LOG(error) << "Exception reached the top of main: " << e.what() << ", exiting";
|
||||
return 2;
|
||||
}
|
||||
|
||||
|
@@ -76,6 +76,8 @@ struct Sampler : fair::mq::Device
|
||||
|
||||
void ResetTask() override
|
||||
{
|
||||
// give some time for acks to be received
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(250));
|
||||
fRegion.reset();
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(fMtx);
|
||||
|
@@ -1,19 +1,19 @@
|
||||
/********************************************************************************
|
||||
* Copyright (C) 2012-2021 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
|
||||
* Copyright (C) 2012-2022 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
|
||||
* *
|
||||
* This software is distributed under the terms of the *
|
||||
* GNU Lesser General Public Licence (LGPL) version 3, *
|
||||
* copied verbatim in the file "LICENSE" *
|
||||
********************************************************************************/
|
||||
|
||||
#include <algorithm> // std::max
|
||||
#include <algorithm> // std::max, std::any_of
|
||||
#include <boost/algorithm/string.hpp> // join/split
|
||||
#include <chrono>
|
||||
#include <fairmq/Device.h>
|
||||
#include <fairmq/Tools.h>
|
||||
#include <future>
|
||||
#include <iomanip>
|
||||
#include <list>
|
||||
#include <memory> // std::make_unique
|
||||
#include <mutex>
|
||||
#include <thread>
|
||||
|
||||
@@ -27,7 +27,6 @@ constexpr const char* Device::DefaultTransportName;
|
||||
constexpr mq::Transport Device::DefaultTransportType;
|
||||
constexpr const char* Device::DefaultNetworkInterface;
|
||||
constexpr int Device::DefaultInitTimeout;
|
||||
constexpr uint64_t Device::DefaultMaxRunTime;
|
||||
constexpr float Device::DefaultRate;
|
||||
constexpr const char* Device::DefaultSession;
|
||||
|
||||
@@ -83,7 +82,6 @@ Device::Device(ProgOptions* config, tools::Version version)
|
||||
, fMultitransportProceed(false)
|
||||
, fVersion(version)
|
||||
, fRate(DefaultRate)
|
||||
, fMaxRunRuntimeInS(DefaultMaxRunTime)
|
||||
, fInitializationTimeoutInS(DefaultInitTimeout)
|
||||
, fTransitioning(false)
|
||||
{
|
||||
@@ -215,7 +213,6 @@ void Device::InitWrapper()
|
||||
Init();
|
||||
|
||||
fRate = fConfig->GetProperty<float>("rate", DefaultRate);
|
||||
fMaxRunRuntimeInS = fConfig->GetProperty<uint64_t>("max-run-time", DefaultMaxRunTime);
|
||||
fInitializationTimeoutInS = fConfig->GetProperty<int>("init-timeout", DefaultInitTimeout);
|
||||
|
||||
try {
|
||||
@@ -293,8 +290,10 @@ void Device::BindWrapper()
|
||||
|
||||
Bind();
|
||||
|
||||
if (!NewStatePending()) {
|
||||
ChangeState(Transition::Auto);
|
||||
}
|
||||
}
|
||||
|
||||
void Device::ConnectWrapper()
|
||||
{
|
||||
@@ -330,8 +329,10 @@ void Device::ConnectWrapper()
|
||||
|
||||
Connect();
|
||||
|
||||
if (!NewStatePending()) {
|
||||
ChangeState(Transition::Auto);
|
||||
}
|
||||
}
|
||||
|
||||
void Device::AttachChannels(vector<Channel*>& chans)
|
||||
{
|
||||
@@ -430,15 +431,28 @@ void Device::InitTaskWrapper()
|
||||
{
|
||||
InitTask();
|
||||
|
||||
if (!NewStatePending()) {
|
||||
ChangeState(Transition::Auto);
|
||||
}
|
||||
}
|
||||
|
||||
void Device::RunWrapper()
|
||||
{
|
||||
LOG(info) << "fair::mq::Device running...";
|
||||
|
||||
// start the rate logger thread
|
||||
future<void> rateLogger = async(launch::async, &Device::LogSocketRates, this);
|
||||
unique_ptr<thread> rateLogger;
|
||||
// Check if rate logging thread is needed
|
||||
const bool rateLogging = any_of(fChannels.cbegin(), fChannels.cend(), [](auto ch) {
|
||||
return any_of(ch.second.cbegin(), ch.second.cend(), [](auto sub) { return sub.fRateLogging > 0; });
|
||||
});
|
||||
|
||||
if (rateLogging) {
|
||||
rateLogger = make_unique<thread>(&Device::LogSocketRates, this);
|
||||
}
|
||||
tools::CallOnDestruction joinRateLogger([&](){
|
||||
if (rateLogging && rateLogger->joinable()) { rateLogger->join(); }
|
||||
});
|
||||
|
||||
|
||||
// notify transports to resume transfers
|
||||
for (auto& t : fTransports) {
|
||||
@@ -481,8 +495,6 @@ void Device::RunWrapper()
|
||||
PostRun();
|
||||
|
||||
cod.disable();
|
||||
|
||||
rateLogger.get();
|
||||
}
|
||||
|
||||
void Device::HandleSingleChannelInput()
|
||||
@@ -710,7 +722,6 @@ void Device::LogSocketRates()
|
||||
|
||||
chrono::time_point<chrono::high_resolution_clock> t0(chrono::high_resolution_clock::now());
|
||||
chrono::time_point<chrono::high_resolution_clock> t1;
|
||||
uint64_t secondsElapsed = 0;
|
||||
|
||||
while (!NewStatePending()) {
|
||||
WaitFor(chrono::seconds(1));
|
||||
@@ -743,7 +754,7 @@ void Device::LogSocketRates()
|
||||
bytesOut.at(i) = bytesOutNew.at(i);
|
||||
msgOut.at(i) = msgOutNew.at(i);
|
||||
|
||||
LOG(info) << setw(chanNameLen) << filteredChannelNames.at(i) << ": "
|
||||
LOG(info) << setw(static_cast<int>(chanNameLen)) << filteredChannelNames.at(i) << ": "
|
||||
<< "in: " << msgPerSecIn.at(i) << " (" << mbPerSecIn.at(i) << " MB) "
|
||||
<< "out: " << msgPerSecOut.at(i) << " (" << mbPerSecOut.at(i) << " MB)";
|
||||
}
|
||||
@@ -753,9 +764,6 @@ void Device::LogSocketRates()
|
||||
}
|
||||
|
||||
t0 = t1;
|
||||
if (fMaxRunRuntimeInS > 0 && ++secondsElapsed >= fMaxRunRuntimeInS) {
|
||||
ChangeState(Transition::Stop);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -770,8 +778,10 @@ void Device::ResetTaskWrapper()
|
||||
{
|
||||
ResetTask();
|
||||
|
||||
if (!NewStatePending()) {
|
||||
ChangeState(Transition::Auto);
|
||||
}
|
||||
}
|
||||
|
||||
void Device::ResetWrapper()
|
||||
{
|
||||
@@ -784,8 +794,10 @@ void Device::ResetWrapper()
|
||||
fChannels.clear();
|
||||
fTransports.clear();
|
||||
fTransportFactory.reset();
|
||||
if (!NewStatePending()) {
|
||||
ChangeState(Transition::Auto);
|
||||
}
|
||||
}
|
||||
|
||||
Device::~Device()
|
||||
{
|
||||
|
@@ -1,5 +1,5 @@
|
||||
/********************************************************************************
|
||||
* Copyright (C) 2021 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
|
||||
* Copyright (C) 2021-2022 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
|
||||
* *
|
||||
* This software is distributed under the terms of the *
|
||||
* GNU Lesser General Public Licence (LGPL) version 3, *
|
||||
@@ -565,7 +565,6 @@ class Device
|
||||
static constexpr mq::Transport DefaultTransportType = mq::Transport::ZMQ;
|
||||
static constexpr const char* DefaultNetworkInterface = "default";
|
||||
static constexpr int DefaultInitTimeout = 120;
|
||||
static constexpr uint64_t DefaultMaxRunTime = 0;
|
||||
static constexpr float DefaultRate = 0.;
|
||||
static constexpr const char* DefaultSession = "default";
|
||||
|
||||
|
@@ -187,9 +187,7 @@ struct Machine_ : public state_machine_def<Machine_>
|
||||
{
|
||||
unique_lock<mutex> lock(fStateMtx);
|
||||
|
||||
while (!fNewStatePending) {
|
||||
fNewStatePendingCV.wait_for(lock, chrono::milliseconds(100));
|
||||
}
|
||||
fNewStatePendingCV.wait(lock, [this]{ return fNewStatePending.load(); });
|
||||
|
||||
LOG(state) << fState << " ---> " << fNewState;
|
||||
fState = static_cast<State>(fNewState);
|
||||
|
@@ -1,5 +1,5 @@
|
||||
/********************************************************************************
|
||||
* Copyright (C) 2019 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
|
||||
* Copyright (C) 2019-2022 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
|
||||
* *
|
||||
* This software is distributed under the terms of the *
|
||||
* GNU Lesser General Public Licence (LGPL) version 3, *
|
||||
@@ -9,16 +9,14 @@
|
||||
#ifndef FAIRMQSTATEQUEUE_H_
|
||||
#define FAIRMQSTATEQUEUE_H_
|
||||
|
||||
#include <fairmq/States.h>
|
||||
|
||||
#include <queue>
|
||||
#include <mutex>
|
||||
#include <chrono>
|
||||
#include <utility> // pair
|
||||
#include <condition_variable>
|
||||
#include <fairmq/States.h>
|
||||
#include <mutex>
|
||||
#include <queue>
|
||||
#include <utility> // pair
|
||||
|
||||
namespace fair::mq
|
||||
{
|
||||
namespace fair::mq {
|
||||
|
||||
class StateQueue
|
||||
{
|
||||
@@ -33,41 +31,47 @@ class StateQueue
|
||||
fair::mq::State WaitForNext()
|
||||
{
|
||||
std::unique_lock<std::mutex> lock(fMtx);
|
||||
while (fStates.empty()) {
|
||||
fCV.wait_for(lock, std::chrono::milliseconds(50));
|
||||
fCV.wait(lock, [this] { return Predicate(); });
|
||||
return PopFrontUnsafe();
|
||||
}
|
||||
|
||||
fair::mq::State state = fStates.front();
|
||||
|
||||
if (state == fair::mq::State::Error) {
|
||||
throw DeviceErrorState("Controlled device transitioned to error state.");
|
||||
}
|
||||
|
||||
fStates.pop();
|
||||
return state;
|
||||
}
|
||||
|
||||
template<typename Rep, typename Period>
|
||||
std::pair<bool, fair::mq::State> WaitForNext(std::chrono::duration<Rep, Period> const& duration)
|
||||
template<typename Timeout>
|
||||
std::pair<bool, fair::mq::State> WaitForNext(Timeout&& duration)
|
||||
{
|
||||
std::unique_lock<std::mutex> lock(fMtx);
|
||||
fCV.wait_for(lock, duration);
|
||||
|
||||
if (fStates.empty()) {
|
||||
return { false, fair::mq::State::Ok };
|
||||
fCV.wait_for(lock, std::forward<Timeout>(duration), [this] { return Predicate(); });
|
||||
return ReturnPairUnsafe();
|
||||
}
|
||||
|
||||
fair::mq::State state = fStates.front();
|
||||
|
||||
if (state == fair::mq::State::Error) {
|
||||
throw DeviceErrorState("Controlled device transitioned to error state.");
|
||||
template<typename CustomPredicate>
|
||||
std::pair<bool, fair::mq::State> WaitForNextOrCustom(CustomPredicate&& customPredicate)
|
||||
{
|
||||
std::unique_lock<std::mutex> lock(fMtx);
|
||||
fCV.wait(lock, [this, cp = std::move(customPredicate)] { return Predicate() || cp(); });
|
||||
return ReturnPairUnsafe();
|
||||
}
|
||||
|
||||
fStates.pop();
|
||||
return { true, state };
|
||||
template<typename CustomPredicate>
|
||||
std::pair<bool, fair::mq::State> WaitForCustom(CustomPredicate&& customPredicate)
|
||||
{
|
||||
std::unique_lock<std::mutex> lock(fMtx);
|
||||
fCV.wait(lock, [cp = std::move(customPredicate)] { return cp(); });
|
||||
return ReturnPairUnsafe();
|
||||
}
|
||||
|
||||
void WaitForState(fair::mq::State state) { while (WaitForNext() != state) {} }
|
||||
void WaitForState(fair::mq::State state)
|
||||
{
|
||||
while (WaitForNext() != state) {}
|
||||
}
|
||||
|
||||
template<typename CustomPredicate>
|
||||
void WaitForStateOrCustom(fair::mq::State state, CustomPredicate customPredicate)
|
||||
{
|
||||
auto next = WaitForNextOrCustom(customPredicate);
|
||||
while (!customPredicate() && (next.first && next.second != state)) {
|
||||
next = WaitForNextOrCustom(customPredicate);
|
||||
}
|
||||
}
|
||||
|
||||
void Push(fair::mq::State state)
|
||||
{
|
||||
@@ -75,7 +79,35 @@ class StateQueue
|
||||
std::lock_guard<std::mutex> lock(fMtx);
|
||||
fStates.push(state);
|
||||
}
|
||||
fCV.notify_all();
|
||||
fCV.notify_one();
|
||||
}
|
||||
|
||||
template<typename CustomSignal>
|
||||
void Push(fair::mq::State state, CustomSignal&& signal)
|
||||
{
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(fMtx);
|
||||
fStates.push(state);
|
||||
signal();
|
||||
}
|
||||
fCV.notify_one();
|
||||
}
|
||||
|
||||
template<typename CustomSignal>
|
||||
void Notify(CustomSignal&& signal)
|
||||
{
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(fMtx);
|
||||
signal();
|
||||
}
|
||||
fCV.notify_one();
|
||||
}
|
||||
|
||||
template<typename CustomSignal>
|
||||
void Locked(CustomSignal&& signal)
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(fMtx);
|
||||
signal();
|
||||
}
|
||||
|
||||
void Clear()
|
||||
@@ -88,6 +120,27 @@ class StateQueue
|
||||
std::queue<fair::mq::State> fStates;
|
||||
std::mutex fMtx;
|
||||
std::condition_variable fCV;
|
||||
|
||||
// must be called under locked fMtx
|
||||
fair::mq::State PopFrontUnsafe()
|
||||
{
|
||||
fair::mq::State state = fStates.front();
|
||||
if (state == fair::mq::State::Error) {
|
||||
throw DeviceErrorState("Controlled device transitioned to error state.");
|
||||
}
|
||||
fStates.pop();
|
||||
return state;
|
||||
}
|
||||
|
||||
// must be called under locked fMtx
|
||||
std::pair<bool, fair::mq::State> ReturnPairUnsafe()
|
||||
{
|
||||
auto const pred = Predicate();
|
||||
return {pred, pred ? PopFrontUnsafe() : fair::mq::State::Ok};
|
||||
}
|
||||
|
||||
// must be called under locked fMtx
|
||||
bool Predicate() { return !fStates.empty(); }
|
||||
};
|
||||
|
||||
} // namespace fair::mq
|
||||
|
@@ -133,6 +133,7 @@ struct RegionConfig
|
||||
bool removeOnDestruction = true; /// remove the region on object destruction
|
||||
int creationFlags = 0; /// flags passed to the underlying transport on region creation
|
||||
int64_t userFlags = 0; /// custom flags that have no effect on the transport, but can be retrieved from the region by the user
|
||||
uint64_t size = 0; /// region size
|
||||
std::string path = ""; /// file path, if the region is backed by a file
|
||||
std::optional<uint16_t> id = std::nullopt; /// region id
|
||||
uint32_t linger = 100; /// delay in ms before region destruction to collect outstanding events
|
||||
|
@@ -1,5 +1,5 @@
|
||||
/********************************************************************************
|
||||
* Copyright (C) 2018-2020 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
|
||||
* Copyright (C) 2018-2022 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
|
||||
* *
|
||||
* This software is distributed under the terms of the *
|
||||
* GNU Lesser General Public Licence (LGPL) version 3, *
|
||||
@@ -21,7 +21,7 @@
|
||||
#define FAIRMQ_GIT_DATE "@PROJECT_GIT_DATE@"
|
||||
#define FAIRMQ_REPO_URL "https://github.com/FairRootGroup/FairMQ"
|
||||
#define FAIRMQ_LICENSE "LGPL-3.0"
|
||||
#define FAIRMQ_COPYRIGHT "2012-2021 GSI"
|
||||
#define FAIRMQ_COPYRIGHT "2012-2022 GSI"
|
||||
#define FAIRMQ_BUILD_TYPE "@CMAKE_BUILD_TYPE@"
|
||||
|
||||
#endif // FAIR_MQ_VERSION_H
|
||||
|
@@ -1,5 +1,5 @@
|
||||
/********************************************************************************
|
||||
* Copyright (C) 2017-2018 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
|
||||
* Copyright (C) 2017-2022 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
|
||||
* *
|
||||
* This software is distributed under the terms of the *
|
||||
* GNU Lesser General Public Licence (LGPL) version 3, *
|
||||
@@ -40,7 +40,7 @@ Config::Config(const string& name, Plugin::Version version, const string& mainta
|
||||
LOG(debug) << "channel-config: Parsing channel configuration";
|
||||
SetProperties(SuboptParser(GetProperty<vector<string>>("channel-config"), idForParser));
|
||||
} else {
|
||||
LOG(warn) << "fair::mq::plugins::Config: no channels configuration provided via --mq-config or --channel-config";
|
||||
LOG(info) << "fair::mq::plugins::Config: no channels configuration provided via --mq-config or --channel-config";
|
||||
}
|
||||
} catch (exception& e) {
|
||||
LOG(error) << e.what();
|
||||
@@ -62,7 +62,6 @@ Plugin::ProgOptions ConfigPluginProgramOptions()
|
||||
("transport", po::value<string >()->default_value("zeromq"), "Transport ('zeromq'/'shmem').")
|
||||
("network-interface", po::value<string >()->default_value("default"), "Network interface to bind on (e.g. eth0, ib0..., default will try to detect the interface of the default route).")
|
||||
("init-timeout", po::value<int >()->default_value(120), "Timeout for the initialization in seconds (when expecting dynamic initialization).")
|
||||
("max-run-time", po::value<uint64_t >()->default_value(0), "Maximum runtime for the Running state handler, after which state will change to Ready (in seconds, 0 for no limit).")
|
||||
("print-channels", po::value<bool >()->implicit_value(true), "Print registered channel endpoints in a machine-readable format (<channel name>:<min num subchannels>:<max num subchannels>)")
|
||||
("shm-segment-size", po::value<size_t >()->default_value(2ULL << 30), "Shared memory: size of the shared memory segment (in bytes).")
|
||||
("shm-allocation", po::value<string >()->default_value("rbtree_best_fit"), "Shared memory allocation algorithm: rbtree_best_fit/simple_seq_fit.")
|
||||
|
@@ -1,5 +1,5 @@
|
||||
/********************************************************************************
|
||||
* Copyright (C) 2017 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
|
||||
* Copyright (C) 2017-2022 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
|
||||
* *
|
||||
* This software is distributed under the terms of the *
|
||||
* GNU Lesser General Public Licence (LGPL) version 3, *
|
||||
@@ -36,7 +36,7 @@ REGISTER_FAIRMQ_PLUGIN(
|
||||
config, // Plugin name
|
||||
(Plugin::Version{FAIRMQ_VERSION_MAJOR, FAIRMQ_VERSION_MINOR, FAIRMQ_VERSION_PATCH}),
|
||||
"FairRootGroup <fairroot@gsi.de>",
|
||||
"https://github.com/FairRootGroup/FairRoot",
|
||||
"https://github.com/FairRootGroup/FairMQ",
|
||||
ConfigPluginProgramOptions
|
||||
)
|
||||
|
||||
|
@@ -1,5 +1,5 @@
|
||||
/********************************************************************************
|
||||
* Copyright (C) 2017-2018 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
|
||||
* Copyright (C) 2017-2022 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
|
||||
* *
|
||||
* This software is distributed under the terms of the *
|
||||
* GNU Lesser General Public Licence (LGPL) version 3, *
|
||||
@@ -56,11 +56,11 @@ Control::Control(const string& name, Plugin::Version version, const string& main
|
||||
SubscribeToDeviceStateChange([&](DeviceState newState) {
|
||||
LOG(trace) << "control plugin notified on new state: " << newState;
|
||||
|
||||
fStateQueue.Push(newState);
|
||||
|
||||
if (newState == DeviceState::Error) {
|
||||
fPluginShutdownRequested = true;
|
||||
fDeviceShutdownRequested = true;
|
||||
fStateQueue.Push(newState, [this]{ fDeviceShutdownRequested = true; });
|
||||
} else {
|
||||
fStateQueue.Push(newState);
|
||||
}
|
||||
});
|
||||
|
||||
@@ -99,18 +99,42 @@ Control::Control(const string& name, Plugin::Version version, const string& main
|
||||
|
||||
auto Control::RunStartupSequence() -> void
|
||||
{
|
||||
ChangeDeviceState(DeviceStateTransition::InitDevice);
|
||||
while (fStateQueue.WaitForNext() != DeviceState::InitializingDevice) {}
|
||||
ChangeDeviceState(DeviceStateTransition::CompleteInit);
|
||||
while (fStateQueue.WaitForNext() != DeviceState::Initialized) {}
|
||||
ChangeDeviceState(DeviceStateTransition::Bind);
|
||||
while (fStateQueue.WaitForNext() != DeviceState::Bound) {}
|
||||
ChangeDeviceState(DeviceStateTransition::Connect);
|
||||
while (fStateQueue.WaitForNext() != DeviceState::DeviceReady) {}
|
||||
ChangeDeviceState(DeviceStateTransition::InitTask);
|
||||
while (fStateQueue.WaitForNext() != DeviceState::Ready) {}
|
||||
ChangeDeviceState(DeviceStateTransition::Run);
|
||||
while (fStateQueue.WaitForNext() != DeviceState::Running) {}
|
||||
using Transition = DeviceStateTransition;
|
||||
using State = DeviceState;
|
||||
auto shutdownRequested = [this]{ return fDeviceShutdownRequested.load(); };
|
||||
|
||||
ChangeDeviceState(Transition::InitDevice);
|
||||
fStateQueue.WaitForStateOrCustom(State::InitializingDevice, shutdownRequested);
|
||||
if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }
|
||||
|
||||
ChangeDeviceState(Transition::CompleteInit);
|
||||
fStateQueue.WaitForStateOrCustom(State::Initialized, shutdownRequested);
|
||||
if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }
|
||||
|
||||
ChangeDeviceState(Transition::Bind);
|
||||
fStateQueue.WaitForStateOrCustom(State::Binding, shutdownRequested);
|
||||
if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }
|
||||
|
||||
fStateQueue.WaitForStateOrCustom(State::Bound, shutdownRequested);
|
||||
if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }
|
||||
|
||||
ChangeDeviceState(Transition::Connect);
|
||||
fStateQueue.WaitForStateOrCustom(State::Connecting, shutdownRequested);
|
||||
if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }
|
||||
|
||||
fStateQueue.WaitForStateOrCustom(State::DeviceReady, shutdownRequested);
|
||||
if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }
|
||||
|
||||
ChangeDeviceState(Transition::InitTask);
|
||||
fStateQueue.WaitForStateOrCustom(State::InitializingTask, shutdownRequested);
|
||||
if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }
|
||||
|
||||
fStateQueue.WaitForStateOrCustom(State::Ready, shutdownRequested);
|
||||
if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }
|
||||
|
||||
ChangeDeviceState(Transition::Run);
|
||||
fStateQueue.WaitForStateOrCustom(State::Running, shutdownRequested);
|
||||
if (fDeviceShutdownRequested) { return; /* --> shutdown sequence */ }
|
||||
}
|
||||
|
||||
auto ControlPluginProgramOptions() -> Plugin::ProgOptions
|
||||
@@ -123,10 +147,8 @@ auto ControlPluginProgramOptions() -> Plugin::ProgOptions
|
||||
return pluginOptions;
|
||||
}
|
||||
|
||||
auto Control::InteractiveMode() -> void
|
||||
try {
|
||||
RunStartupSequence();
|
||||
|
||||
auto Control::RunREPL() -> void
|
||||
{
|
||||
char input = 0; // hold the user console input
|
||||
pollfd cinfd[1];
|
||||
cinfd[0].fd = fileno(stdin);
|
||||
@@ -161,7 +183,7 @@ try {
|
||||
case 'i':
|
||||
cout << "\n --> [i] init device\n\n" << flush;
|
||||
if (ChangeDeviceState(DeviceStateTransition::InitDevice)) {
|
||||
while (fStateQueue.WaitForNext() != DeviceState::InitializingDevice) {}
|
||||
fStateQueue.WaitForState(DeviceState::InitializingDevice);
|
||||
ChangeDeviceState(DeviceStateTransition::CompleteInit);
|
||||
}
|
||||
break;
|
||||
@@ -243,7 +265,19 @@ try {
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
auto Control::InteractiveMode() -> void
|
||||
try {
|
||||
RunStartupSequence();
|
||||
|
||||
if(!fDeviceShutdownRequested) {
|
||||
RunREPL();
|
||||
}
|
||||
|
||||
if(!fDeviceShutdownRequested) {
|
||||
RunShutdownSequence();
|
||||
}
|
||||
} catch (PluginServices::DeviceControlError& e) {
|
||||
// If we are here, it means another plugin has taken control. That's fine, just print the exception message and do nothing else.
|
||||
LOG(debug) << e.what();
|
||||
@@ -366,16 +400,13 @@ auto Control::StaticMode() -> void
|
||||
try {
|
||||
RunStartupSequence();
|
||||
|
||||
{
|
||||
// Wait for next state, which is DeviceState::Ready,
|
||||
// or for device shutdown request (Ctrl-C)
|
||||
pair<bool, fair::mq::State> result;
|
||||
do {
|
||||
result = fStateQueue.WaitForNext(chrono::milliseconds(50));
|
||||
} while (result.first == false && !fDeviceShutdownRequested);
|
||||
}
|
||||
fStateQueue.WaitForNextOrCustom([this]{ return fDeviceShutdownRequested.load(); });
|
||||
|
||||
if(!fDeviceShutdownRequested) {
|
||||
RunShutdownSequence();
|
||||
}
|
||||
} catch (PluginServices::DeviceControlError& e) {
|
||||
// If we are here, it means another plugin has taken control. That's fine, just print the exception message and do nothing else.
|
||||
LOG(debug) << e.what();
|
||||
@@ -387,16 +418,12 @@ auto Control::GUIMode() -> void
|
||||
try {
|
||||
RunStartupSequence();
|
||||
|
||||
{
|
||||
// Wait for next state, which is DeviceState::Ready,
|
||||
// or for device shutdown request (Ctrl-C)
|
||||
pair<bool, fair::mq::State> result;
|
||||
do {
|
||||
result = fStateQueue.WaitForNext(chrono::milliseconds(50));
|
||||
} while (!fDeviceShutdownRequested);
|
||||
}
|
||||
// Wait for device shutdown request (Ctrl-C)
|
||||
fStateQueue.WaitForCustom([this]{ return fDeviceShutdownRequested.load(); });
|
||||
|
||||
if(!fDeviceShutdownRequested) {
|
||||
RunShutdownSequence();
|
||||
}
|
||||
} catch (PluginServices::DeviceControlError& e) {
|
||||
// If we are here, it means another plugin has taken control. That's fine, just print the
|
||||
// exception message and do nothing else.
|
||||
@@ -416,10 +443,10 @@ auto Control::SignalHandler() -> void
|
||||
LOG(info) << "Waiting for graceful device shutdown. Hit Ctrl-C again to abort immediately.";
|
||||
|
||||
// Signal and wait for controller thread, if we are controller
|
||||
fDeviceShutdownRequested = true;
|
||||
fStateQueue.Notify([this] { fDeviceShutdownRequested = true; });
|
||||
{
|
||||
unique_lock<mutex> lock(fControllerMutex);
|
||||
if (fControllerThread.joinable()) fControllerThread.join();
|
||||
if (fControllerThread.joinable()) { fControllerThread.join(); }
|
||||
}
|
||||
|
||||
if (!fDeviceHasShutdown) {
|
||||
@@ -462,6 +489,12 @@ auto Control::RunShutdownSequence() -> void
|
||||
case DeviceState::Running:
|
||||
ChangeDeviceState(DeviceStateTransition::Stop);
|
||||
break;
|
||||
case DeviceState::Binding:
|
||||
case DeviceState::Connecting:
|
||||
case DeviceState::InitializingTask:
|
||||
case DeviceState::ResettingTask:
|
||||
case DeviceState::ResettingDevice:
|
||||
ChangeDeviceState(DeviceStateTransition::Auto);
|
||||
default:
|
||||
// LOG(debug) << "Controller ignoring event: " << nextState;
|
||||
break;
|
||||
@@ -481,9 +514,9 @@ Control::~Control()
|
||||
|
||||
{
|
||||
unique_lock<mutex> lock(fControllerMutex);
|
||||
if (fControllerThread.joinable()) fControllerThread.join();
|
||||
if (fControllerThread.joinable()) { fControllerThread.join(); }
|
||||
}
|
||||
if (fSignalHandlerThread.joinable()) fSignalHandlerThread.join();
|
||||
if (fSignalHandlerThread.joinable()) { fSignalHandlerThread.join(); }
|
||||
|
||||
UnsubscribeFromDeviceStateChange();
|
||||
}
|
||||
|
@@ -1,5 +1,5 @@
|
||||
/********************************************************************************
|
||||
* Copyright (C) 2017 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
|
||||
* Copyright (C) 2017-2022 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
|
||||
* *
|
||||
* This software is distributed under the terms of the *
|
||||
* GNU Lesser General Public Licence (LGPL) version 3, *
|
||||
@@ -46,6 +46,7 @@ class Control : public Plugin
|
||||
auto GUIMode() -> void;
|
||||
auto SignalHandler() -> void;
|
||||
auto RunShutdownSequence() -> void;
|
||||
auto RunREPL() -> void;
|
||||
auto RunStartupSequence() -> void;
|
||||
|
||||
std::thread fControllerThread;
|
||||
|
@@ -28,6 +28,8 @@
|
||||
namespace fair::mq::shmem
|
||||
{
|
||||
|
||||
static constexpr uint64_t kManagementSegmentSize = 6553600;
|
||||
|
||||
struct SharedMemoryError : std::runtime_error { using std::runtime_error::runtime_error; };
|
||||
|
||||
using SimpleSeqFitSegment = boost::interprocess::basic_managed_shared_memory<char,
|
||||
@@ -58,19 +60,22 @@ struct RegionInfo
|
||||
: fPath("", alloc)
|
||||
, fCreationFlags(0)
|
||||
, fUserFlags(0)
|
||||
, fSize(0)
|
||||
, fDestroyed(false)
|
||||
{}
|
||||
|
||||
RegionInfo(const char* path, const int flags, const uint64_t userFlags, const VoidAlloc& alloc)
|
||||
RegionInfo(const char* path, int flags, uint64_t userFlags, uint64_t size, const VoidAlloc& alloc)
|
||||
: fPath(path, alloc)
|
||||
, fCreationFlags(flags)
|
||||
, fUserFlags(userFlags)
|
||||
, fSize(size)
|
||||
, fDestroyed(false)
|
||||
{}
|
||||
|
||||
Str fPath;
|
||||
int fCreationFlags;
|
||||
uint64_t fUserFlags;
|
||||
uint64_t fSize;
|
||||
bool fDestroyed;
|
||||
};
|
||||
|
||||
|
@@ -132,7 +132,7 @@ class Manager
|
||||
: fShmId64(config ? config->GetProperty<uint64_t>("shmid", makeShmIdUint64(sessionName)) : makeShmIdUint64(sessionName))
|
||||
, fShmId(makeShmIdStr(fShmId64))
|
||||
, fSegmentId(config ? config->GetProperty<uint16_t>("shm-segment-id", 0) : 0)
|
||||
, fManagementSegment(boost::interprocess::open_or_create, std::string("fmq_" + fShmId + "_mng").c_str(), 6553600)
|
||||
, fManagementSegment(boost::interprocess::open_or_create, std::string("fmq_" + fShmId + "_mng").c_str(), kManagementSegmentSize)
|
||||
, fShmVoidAlloc(fManagementSegment.get_segment_manager())
|
||||
, fShmMtx(fManagementSegment.find_or_construct<boost::interprocess::interprocess_mutex>(boost::interprocess::unique_instance)())
|
||||
, fNumObservedEvents(0)
|
||||
@@ -228,6 +228,8 @@ class Manager
|
||||
fShmSegments = fManagementSegment.find_or_construct<Uint16SegmentInfoHashMap>(unique_instance)(fShmVoidAlloc);
|
||||
fShmRegions = fManagementSegment.find_or_construct<Uint16RegionInfoHashMap>(unique_instance)(fShmVoidAlloc);
|
||||
|
||||
bool createdSegment = false;
|
||||
|
||||
try {
|
||||
std::string segmentName("fmq_" + fShmId + "_m_" + std::to_string(fSegmentId));
|
||||
auto it = fShmSegments->find(fSegmentId);
|
||||
@@ -246,6 +248,7 @@ class Manager
|
||||
if (zeroSegmentOnCreation) {
|
||||
ZeroSegment(fSegmentId);
|
||||
}
|
||||
createdSegment = true;
|
||||
} else {
|
||||
// found segment with the given id, opening
|
||||
if (it->second.fAllocationAlgorithm == AllocationAlgorithm::rbtree_best_fit) {
|
||||
@@ -278,7 +281,9 @@ class Manager
|
||||
ZeroSegment(fSegmentId);
|
||||
}
|
||||
|
||||
if (createdSegment) {
|
||||
(fEventCounter->fCount)++;
|
||||
}
|
||||
|
||||
#ifdef FAIRMQ_DEBUG_MODE
|
||||
fMsgDebug = fManagementSegment.find_or_construct<Uint16MsgDebugMapHashMap>(unique_instance)(fShmVoidAlloc);
|
||||
@@ -360,7 +365,7 @@ class Manager
|
||||
}
|
||||
bool Interrupted() { return fInterrupted.load(); }
|
||||
|
||||
std::pair<UnmanagedRegion*, uint16_t> CreateRegion(const size_t size,
|
||||
std::pair<UnmanagedRegion*, uint16_t> CreateRegion(size_t size,
|
||||
RegionCallback callback,
|
||||
RegionBulkCallback bulkCallback,
|
||||
RegionConfig cfg)
|
||||
@@ -370,7 +375,7 @@ class Manager
|
||||
std::pair<UnmanagedRegion*, uint16_t> result;
|
||||
|
||||
{
|
||||
boost::interprocess::scoped_lock<boost::interprocess::interprocess_mutex> lock(*fShmMtx);
|
||||
boost::interprocess::scoped_lock<boost::interprocess::interprocess_mutex> shmLock(*fShmMtx);
|
||||
|
||||
if (!cfg.id.has_value()) {
|
||||
RegionCounter* rc = fManagementSegment.find<RegionCounter>(unique_instance).first;
|
||||
@@ -390,23 +395,26 @@ class Manager
|
||||
|
||||
const uint16_t id = cfg.id.value();
|
||||
|
||||
UnmanagedRegion* region = nullptr;
|
||||
bool newRegionCreated = false;
|
||||
std::lock_guard<std::mutex> lock(fLocalRegionsMtx);
|
||||
auto res = fRegions.emplace(id, std::make_unique<UnmanagedRegion>(fShmId, size, false, cfg));
|
||||
bool newRegionCreated = res.second;
|
||||
UnmanagedRegion& region = *(res.first->second);
|
||||
newRegionCreated = res.second;
|
||||
region = res.first->second.get();
|
||||
// LOG(debug) << "Created region with id '" << id << "', path: '" << cfg.path << "', flags: '" << cfg.creationFlags << "'";
|
||||
|
||||
if (!newRegionCreated) {
|
||||
region.fRemote = false; // TODO: this should be more clear, refactor it.
|
||||
region->fRemote = false; // TODO: this should be more clear, refactor it.
|
||||
}
|
||||
|
||||
// start ack receiver only if a callback has been provided.
|
||||
if (callback || bulkCallback) {
|
||||
region.SetCallbacks(callback, bulkCallback);
|
||||
region.InitializeQueues();
|
||||
region.StartAckSender();
|
||||
region.StartAckReceiver();
|
||||
region->SetCallbacks(callback, bulkCallback);
|
||||
region->InitializeQueues();
|
||||
region->StartAckSender();
|
||||
region->StartAckReceiver();
|
||||
}
|
||||
result.first = &(region);
|
||||
result.first = region;
|
||||
result.second = id;
|
||||
}
|
||||
fRegionsGen += 1; // signal TL cache invalidation
|
||||
@@ -419,7 +427,7 @@ class Manager
|
||||
}
|
||||
}
|
||||
|
||||
UnmanagedRegion* GetRegion(const uint16_t id)
|
||||
UnmanagedRegion* GetRegionFromCache(uint16_t id)
|
||||
{
|
||||
// NOTE: gcc optimizations. Prevent loading tls addresses many times in the fast path
|
||||
const auto &lTlCache = fTlRegionCache;
|
||||
@@ -433,40 +441,40 @@ class Manager
|
||||
}
|
||||
}
|
||||
|
||||
boost::interprocess::scoped_lock<boost::interprocess::interprocess_mutex> shmLock(*fShmMtx);
|
||||
// slow path: check invalidation
|
||||
if (lTlCacheGen != fRegionsGen) {
|
||||
fTlRegionCache.fRegionsTLCache.clear();
|
||||
}
|
||||
|
||||
auto* lRegion = GetRegionUnsafe(id, shmLock);
|
||||
auto* lRegion = GetRegion(id);
|
||||
fTlRegionCache.fRegionsTLCache.emplace_back(std::make_tuple(lRegion, id, fShmId64));
|
||||
fTlRegionCache.fRegionsTLCacheGen = fRegionsGen;
|
||||
return lRegion;
|
||||
}
|
||||
|
||||
UnmanagedRegion* GetRegionUnsafe(const uint16_t id, boost::interprocess::scoped_lock<boost::interprocess::interprocess_mutex>& lockedShmLock)
|
||||
UnmanagedRegion* GetRegion(uint16_t id)
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(fLocalRegionsMtx);
|
||||
// remote region could actually be a local one if a message originates from this device (has been sent out and returned)
|
||||
auto it = fRegions.find(id);
|
||||
if (it != fRegions.end()) {
|
||||
return it->second.get();
|
||||
} else {
|
||||
try {
|
||||
// get region info
|
||||
RegionInfo regionInfo = fShmRegions->at(id);
|
||||
// safe to unlock now - no shm container accessed after this
|
||||
lockedShmLock.unlock();
|
||||
RegionConfig cfg;
|
||||
// get region info
|
||||
{
|
||||
boost::interprocess::scoped_lock<boost::interprocess::interprocess_mutex> shmLock(*fShmMtx);
|
||||
RegionInfo regionInfo = fShmRegions->at(id);
|
||||
cfg.id = id;
|
||||
cfg.creationFlags = regionInfo.fCreationFlags;
|
||||
cfg.path = regionInfo.fPath.c_str();
|
||||
}
|
||||
// LOG(debug) << "Located remote region with id '" << id << "', path: '" << cfg.path << "', flags: '" << cfg.creationFlags << "'";
|
||||
|
||||
auto r = fRegions.emplace(id, std::make_unique<UnmanagedRegion>(fShmId, 0, true, std::move(cfg)));
|
||||
r.first->second->InitializeQueues();
|
||||
r.first->second->StartAckSender();
|
||||
lockedShmLock.lock();
|
||||
return r.first->second.get();
|
||||
} catch (std::out_of_range& oor) {
|
||||
LOG(error) << "Could not get remote region with id '" << id << "'. Does the region creator run with the same session id?";
|
||||
@@ -479,12 +487,13 @@ class Manager
|
||||
}
|
||||
}
|
||||
|
||||
void RemoveRegion(const uint16_t id)
|
||||
void RemoveRegion(uint16_t id)
|
||||
{
|
||||
try {
|
||||
boost::interprocess::scoped_lock<boost::interprocess::interprocess_mutex> shmLock(*fShmMtx);
|
||||
std::lock_guard<std::mutex> lock(fLocalRegionsMtx);
|
||||
fRegions.at(id)->StopAcks();
|
||||
{
|
||||
boost::interprocess::scoped_lock<boost::interprocess::interprocess_mutex> lock(*fShmMtx);
|
||||
if (fRegions.at(id)->RemoveOnDestruction()) {
|
||||
fShmRegions->at(id).fDestroyed = true;
|
||||
(fEventCounter->fCount)++;
|
||||
@@ -500,45 +509,74 @@ class Manager
|
||||
std::vector<fair::mq::RegionInfo> GetRegionInfo()
|
||||
{
|
||||
std::vector<fair::mq::RegionInfo> result;
|
||||
std::map<uint64_t, RegionConfig> regionCfgs;
|
||||
|
||||
{
|
||||
boost::interprocess::scoped_lock<boost::interprocess::interprocess_mutex> shmLock(*fShmMtx);
|
||||
|
||||
for (const auto& e : *fShmSegments) {
|
||||
for (const auto& [segmentId, segmentInfo] : *fShmSegments) {
|
||||
// make sure any segments in the session are found
|
||||
GetSegment(e.first);
|
||||
GetSegment(segmentId);
|
||||
try {
|
||||
fair::mq::RegionInfo info;
|
||||
info.managed = true;
|
||||
info.id = e.first;
|
||||
info.id = segmentId;
|
||||
info.event = RegionEvent::created;
|
||||
info.ptr = boost::apply_visitor(SegmentAddress(), fSegments.at(e.first));
|
||||
info.size = boost::apply_visitor(SegmentSize(), fSegments.at(e.first));
|
||||
info.ptr = boost::apply_visitor(SegmentAddress(), fSegments.at(segmentId));
|
||||
info.size = boost::apply_visitor(SegmentSize(), fSegments.at(segmentId));
|
||||
result.push_back(info);
|
||||
} catch (const std::out_of_range& oor) {
|
||||
LOG(error) << "could not find segment with id " << e.first;
|
||||
LOG(error) << "could not find segment with id " << segmentId;
|
||||
LOG(error) << oor.what();
|
||||
}
|
||||
}
|
||||
|
||||
for (const auto& e : *fShmRegions) {
|
||||
for (const auto& [regionId, regionInfo] : *fShmRegions) {
|
||||
fair::mq::RegionInfo info;
|
||||
info.managed = false;
|
||||
info.id = e.first;
|
||||
info.flags = e.second.fUserFlags;
|
||||
info.event = e.second.fDestroyed ? RegionEvent::destroyed : RegionEvent::created;
|
||||
info.id = regionId;
|
||||
info.flags = regionInfo.fUserFlags;
|
||||
info.event = regionInfo.fDestroyed ? RegionEvent::destroyed : RegionEvent::created;
|
||||
if (info.event == RegionEvent::created) {
|
||||
auto region = GetRegionUnsafe(info.id, shmLock);
|
||||
if (region) {
|
||||
info.ptr = region->GetData();
|
||||
info.size = region->GetSize();
|
||||
} else {
|
||||
throw std::runtime_error(tools::ToString("GetRegionInfo() could not get region with id '", info.id, "'"));
|
||||
}
|
||||
RegionConfig cfg;
|
||||
cfg.id = info.id;
|
||||
cfg.creationFlags = regionInfo.fCreationFlags;
|
||||
cfg.path = regionInfo.fPath.c_str();
|
||||
regionCfgs.emplace(info.id, cfg);
|
||||
// fill the ptr+size info after shmLock is released, to avoid constructing local region under it
|
||||
} else {
|
||||
info.ptr = nullptr;
|
||||
info.size = 0;
|
||||
}
|
||||
result.push_back(info);
|
||||
}
|
||||
}
|
||||
|
||||
// do another iteration outside of shm lock, to fill ptr+size of unmanaged regions
|
||||
for (auto& info : result) {
|
||||
if (!info.managed && info.event == RegionEvent::created) {
|
||||
auto cfgIt = regionCfgs.find(info.id);
|
||||
if (cfgIt != regionCfgs.end()) {
|
||||
UnmanagedRegion* region = nullptr;
|
||||
std::lock_guard<std::mutex> lock(fLocalRegionsMtx);
|
||||
auto it = fRegions.find(info.id);
|
||||
if (it != fRegions.end()) {
|
||||
region = it->second.get();
|
||||
} else {
|
||||
auto r = fRegions.emplace(cfgIt->first, std::make_unique<UnmanagedRegion>(fShmId, 0, true, cfgIt->second));
|
||||
region = r.first->second.get();
|
||||
region->InitializeQueues();
|
||||
region->StartAckSender();
|
||||
}
|
||||
|
||||
info.ptr = region->GetData();
|
||||
info.size = region->GetSize();
|
||||
} else {
|
||||
info.ptr = nullptr;
|
||||
info.size = 0;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
@@ -800,6 +838,7 @@ class Manager
|
||||
VoidAlloc fShmVoidAlloc;
|
||||
boost::interprocess::interprocess_mutex* fShmMtx;
|
||||
|
||||
std::mutex fLocalRegionsMtx;
|
||||
std::mutex fRegionEventsMtx;
|
||||
std::condition_variable fRegionEventsCV;
|
||||
std::thread fRegionEventThread;
|
||||
|
@@ -195,7 +195,7 @@ class Message final : public fair::mq::Message
|
||||
fLocalPtr = nullptr;
|
||||
}
|
||||
} else {
|
||||
fRegionPtr = fManager.GetRegion(fMeta.fRegionId);
|
||||
fRegionPtr = fManager.GetRegionFromCache(fMeta.fRegionId);
|
||||
if (fRegionPtr) {
|
||||
fLocalPtr = reinterpret_cast<char*>(fRegionPtr->GetData()) + fMeta.fHandle;
|
||||
} else {
|
||||
@@ -365,7 +365,7 @@ class Message final : public fair::mq::Message
|
||||
void ReleaseUnmanagedRegionBlock()
|
||||
{
|
||||
if (!fRegionPtr) {
|
||||
fRegionPtr = fManager.GetRegion(fMeta.fRegionId);
|
||||
fRegionPtr = fManager.GetRegionFromCache(fMeta.fRegionId);
|
||||
}
|
||||
|
||||
if (fRegionPtr) {
|
||||
|
@@ -6,9 +6,10 @@
|
||||
* copied verbatim in the file "LICENSE" *
|
||||
********************************************************************************/
|
||||
|
||||
#include "Monitor.h"
|
||||
#include "Common.h"
|
||||
#include "UnmanagedRegion.h"
|
||||
#include "Monitor.h"
|
||||
#include "Segment.h"
|
||||
#include <fairmq/shmem/UnmanagedRegion.h>
|
||||
|
||||
#include <fairmq/tools/IO.h>
|
||||
#include <fairmq/tools/Strings.h>
|
||||
@@ -22,6 +23,7 @@
|
||||
#include <boost/interprocess/ipc/message_queue.hpp>
|
||||
|
||||
#include <csignal>
|
||||
#include <cstdio>
|
||||
#include <iostream>
|
||||
#include <iomanip>
|
||||
#include <chrono>
|
||||
@@ -267,13 +269,14 @@ bool Monitor::PrintShm(const ShmId& shmId)
|
||||
ss << "\n unmanaged regions:";
|
||||
for (const auto& r : *shmRegions) {
|
||||
ss << "\n [" << r.first << "]: " << (r.second.fDestroyed ? "destroyed" : "alive");
|
||||
ss << ", size: " << r.second.fSize;
|
||||
|
||||
try {
|
||||
boost::interprocess::message_queue q(open_only, std::string("fmq_" + std::string(shmId) + "_rgq_" + to_string(r.first)).c_str());
|
||||
ss << ", ack queue: " << q.get_num_msg() << " messages";
|
||||
} catch (bie&) {
|
||||
ss << ", ack queue: not found";
|
||||
}
|
||||
// try {
|
||||
// boost::interprocess::message_queue q(open_only, std::string("fmq_" + std::string(shmId) + "_rgq_" + to_string(r.first)).c_str());
|
||||
// ss << ", ack queue: " << q.get_num_msg() << " messages";
|
||||
// } catch (bie&) {
|
||||
// ss << ", ack queue: not found";
|
||||
// }
|
||||
}
|
||||
}
|
||||
LOGV(info, user1) << ss.str();
|
||||
@@ -414,6 +417,7 @@ void Monitor::PrintDebugInfo(const ShmId& shmId __attribute__((unused)))
|
||||
|
||||
size_t numMessages = 0;
|
||||
|
||||
if (debug) {
|
||||
for (const auto& e : *debug) {
|
||||
numMessages += e.second.size();
|
||||
}
|
||||
@@ -433,6 +437,9 @@ void Monitor::PrintDebugInfo(const ShmId& shmId __attribute__((unused)))
|
||||
<< ", at: " << setw(2) << tm->tm_hour << ":" << setw(2) << tm->tm_min << ":" << setw(2) << tm->tm_sec << "." << setw(6) << ms;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
LOG(info) << "no debug data found";
|
||||
}
|
||||
} catch (bie&) {
|
||||
LOG(info) << "no segments found";
|
||||
}
|
||||
@@ -462,12 +469,17 @@ unordered_map<uint16_t, std::vector<BufferDebugInfo>> Monitor::GetDebugInfo(cons
|
||||
|
||||
result.reserve(debug->size());
|
||||
|
||||
|
||||
if (debug) {
|
||||
for (const auto& s : *debug) {
|
||||
result[s.first].reserve(s.second.size());
|
||||
for (const auto& e : s.second) {
|
||||
result[s.first][e.first] = BufferDebugInfo(e.first, e.second.fPid, e.second.fSize, e.second.fCreationTime);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
LOG(info) << "no debug data found";
|
||||
}
|
||||
} catch (bie&) {
|
||||
LOG(info) << "no segments found";
|
||||
}
|
||||
@@ -522,6 +534,88 @@ unsigned long Monitor::GetFreeMemory(const SessionId& sessionId, uint16_t segmen
|
||||
return GetFreeMemory(shmId, segmentId);
|
||||
}
|
||||
|
||||
bool Monitor::SegmentIsPresent(const ShmId& shmId, uint16_t segmentId)
|
||||
{
|
||||
using namespace boost::interprocess;
|
||||
try {
|
||||
bipc::managed_shared_memory managementSegment(bipc::open_read_only, std::string("fmq_" + shmId.shmId + "_mng").c_str());
|
||||
Uint16SegmentInfoHashMap* shmSegments = managementSegment.find<Uint16SegmentInfoHashMap>(unique_instance).first;
|
||||
|
||||
if (!shmSegments) {
|
||||
LOG(error) << "Found management segment, but could not locate segment info";
|
||||
return false;
|
||||
}
|
||||
|
||||
auto it = shmSegments->find(segmentId);
|
||||
if (it != shmSegments->end()) {
|
||||
try {
|
||||
if (it->second.fAllocationAlgorithm == AllocationAlgorithm::rbtree_best_fit) {
|
||||
RBTreeBestFitSegment segment(open_read_only, std::string("fmq_" + shmId.shmId + "_m_" + std::to_string(segmentId)).c_str());
|
||||
} else {
|
||||
SimpleSeqFitSegment segment(open_read_only, std::string("fmq_" + shmId.shmId + "_m_" + std::to_string(segmentId)).c_str());
|
||||
}
|
||||
} catch (bie&) {
|
||||
LOG(error) << "Could not find segment with id '" << segmentId << "' for shmId '" << shmId.shmId << "'";
|
||||
return false;
|
||||
}
|
||||
} else {
|
||||
LOG(error) << "Could not find segment info for segment id '" << segmentId << "' for shmId '" << shmId.shmId << "'";
|
||||
return false;
|
||||
}
|
||||
} catch (bie&) {
|
||||
LOG(error) << "Could not find management segment for shmid '" << shmId.shmId << "'";
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
bool Monitor::SegmentIsPresent(const SessionId& sessionId, uint16_t segmentId)
|
||||
{
|
||||
ShmId shmId{makeShmIdStr(sessionId.sessionId)};
|
||||
return SegmentIsPresent(shmId, segmentId);
|
||||
}
|
||||
|
||||
bool Monitor::RegionIsPresent(const ShmId& shmId, uint16_t regionId)
|
||||
{
|
||||
using namespace boost::interprocess;
|
||||
try {
|
||||
bipc::managed_shared_memory managementSegment(bipc::open_read_only, std::string("fmq_" + shmId.shmId + "_mng").c_str());
|
||||
Uint16RegionInfoHashMap* shmRegions = managementSegment.find<Uint16RegionInfoHashMap>(bipc::unique_instance).first;
|
||||
|
||||
if (!shmRegions) {
|
||||
LOG(error) << "Found management segment, but could not locate region info";
|
||||
return false;
|
||||
}
|
||||
|
||||
std::string regionFileName("fmq_" + shmId.shmId + "_rg_" + to_string(regionId));
|
||||
|
||||
auto it = shmRegions->find(regionId);
|
||||
if (it != shmRegions->end()) {
|
||||
try {
|
||||
if (it->second.fPath.empty()) {
|
||||
shared_memory_object object(open_only, regionFileName.c_str(), read_only);
|
||||
}
|
||||
} catch (bie&) {
|
||||
LOG(error) << "Could not find region with id '" << regionId << "' for shmId '" << shmId.shmId << "'";
|
||||
return false;
|
||||
}
|
||||
} else {
|
||||
LOG(error) << "Could not find region info for region id '" << regionId << "' for shmId '" << shmId.shmId << "'";
|
||||
return false;
|
||||
}
|
||||
} catch (bie&) {
|
||||
LOG(error) << "Could not find management segment for shmid '" << shmId.shmId << "'";
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
bool Monitor::RegionIsPresent(const SessionId& sessionId, uint16_t regionId)
|
||||
{
|
||||
ShmId shmId{makeShmIdStr(sessionId.sessionId)};
|
||||
return RegionIsPresent(shmId, regionId);
|
||||
}
|
||||
|
||||
void Monitor::PrintHelp()
|
||||
{
|
||||
LOG(info) << "controls: [x] close memory, "
|
||||
@@ -552,17 +646,18 @@ std::pair<std::string, bool> Remove(const std::string& name, bool verbose)
|
||||
}
|
||||
}
|
||||
|
||||
std::vector<std::pair<std::string, bool>> Monitor::Cleanup(const ShmId& shmId, bool verbose /* = true */)
|
||||
std::vector<std::pair<std::string, bool>> Monitor::Cleanup(const ShmId& shmIdT, bool verbose /* = true */)
|
||||
{
|
||||
std::string shmId = shmIdT.shmId;
|
||||
std::vector<std::pair<std::string, bool>> result;
|
||||
|
||||
if (verbose) {
|
||||
LOG(info) << "Cleaning up for shared memory id '" << shmId.shmId << "'...";
|
||||
LOG(info) << "Cleaning up for shared memory id '" << shmId << "'...";
|
||||
}
|
||||
|
||||
string managementSegmentName("fmq_" + shmId.shmId + "_mng");
|
||||
string managementSegmentName("fmq_" + shmId + "_mng");
|
||||
try {
|
||||
bipc::managed_shared_memory managementSegment(bipc::open_only, managementSegmentName.c_str());
|
||||
bipc::managed_shared_memory managementSegment(bipc::open_read_only, managementSegmentName.c_str());
|
||||
|
||||
Uint16RegionInfoHashMap* shmRegions = managementSegment.find<Uint16RegionInfoHashMap>(bipc::unique_instance).first;
|
||||
if (shmRegions) {
|
||||
@@ -578,22 +673,21 @@ std::vector<std::pair<std::string, bool>> Monitor::Cleanup(const ShmId& shmId, b
|
||||
LOG(info) << "Found RegionInfo with path: '" << path << "', flags: " << flags << ", fDestroyed: " << info.fDestroyed << ".";
|
||||
}
|
||||
if (!path.empty()) {
|
||||
result.emplace_back(Remove<bipc::file_mapping>(path + "fmq_" + shmId.shmId + "_rg_" + to_string(id), verbose));
|
||||
result.emplace_back(Remove<bipc::file_mapping>(path + "fmq_" + shmId + "_rg_" + to_string(id), verbose));
|
||||
} else {
|
||||
result.emplace_back(Remove<bipc::shared_memory_object>("fmq_" + shmId.shmId + "_rg_" + to_string(id), verbose));
|
||||
result.emplace_back(Remove<bipc::shared_memory_object>("fmq_" + shmId + "_rg_" + to_string(id), verbose));
|
||||
}
|
||||
result.emplace_back(Remove<bipc::message_queue>("fmq_" + shmId.shmId + "_rgq_" + to_string(id), verbose));
|
||||
result.emplace_back(Remove<bipc::message_queue>("fmq_" + shmId + "_rgq_" + to_string(id), verbose));
|
||||
}
|
||||
}
|
||||
|
||||
Uint16SegmentInfoHashMap* shmSegments = managementSegment.find<Uint16SegmentInfoHashMap>(bipc::unique_instance).first;
|
||||
|
||||
if (shmSegments) {
|
||||
if (verbose) {
|
||||
LOG(info) << "Found " << shmSegments->size() << " managed segments...";
|
||||
}
|
||||
for (const auto& segment : *shmSegments) {
|
||||
result.emplace_back(Remove<bipc::shared_memory_object>("fmq_" + shmId.shmId + "_m_" + to_string(segment.first), verbose));
|
||||
result.emplace_back(Remove<bipc::shared_memory_object>("fmq_" + shmId + "_m_" + to_string(segment.first), verbose));
|
||||
}
|
||||
} else {
|
||||
if (verbose) {
|
||||
@@ -636,41 +730,57 @@ std::vector<std::pair<std::string, bool>> Monitor::CleanupFull(const SessionId&
|
||||
return CleanupFull(shmId, verbose);
|
||||
}
|
||||
|
||||
void Monitor::ResetContent(const ShmId& shmId, bool verbose /* = true */)
|
||||
void Monitor::ResetContent(const ShmId& shmIdT, bool verbose /* = true */)
|
||||
{
|
||||
std::string shmId = shmIdT.shmId;
|
||||
if (verbose) {
|
||||
cout << "Resetting segments content for shared memory id '" << shmId.shmId << "'..." << endl;
|
||||
cout << "Resetting segments content for shared memory id '" << shmId << "'..." << endl;
|
||||
}
|
||||
|
||||
string managementSegmentName("fmq_" + shmId.shmId + "_mng");
|
||||
string managementSegmentName("fmq_" + shmId + "_mng");
|
||||
try {
|
||||
using namespace boost::interprocess;
|
||||
managed_shared_memory managementSegment(open_only, managementSegmentName.c_str());
|
||||
|
||||
Uint16SegmentInfoHashMap* segmentInfos = managementSegment.find<Uint16SegmentInfoHashMap>(unique_instance).first;
|
||||
|
||||
if (segmentInfos) {
|
||||
cout << "Found info for " << segmentInfos->size() << " managed segments" << endl;
|
||||
for (const auto& s : *segmentInfos) {
|
||||
if (verbose) {
|
||||
cout << "Resetting content of segment '" << "fmq_" << shmId.shmId << "_m_" << s.first << "'..." << endl;
|
||||
cout << "Resetting content of segment '" << "fmq_" << shmId << "_m_" << s.first << "'..." << endl;
|
||||
}
|
||||
try {
|
||||
if (s.second.fAllocationAlgorithm == AllocationAlgorithm::rbtree_best_fit) {
|
||||
RBTreeBestFitSegment segment(open_only, std::string("fmq_" + shmId.shmId + "_m_" + to_string(s.first)).c_str());
|
||||
RBTreeBestFitSegment segment(open_only, std::string("fmq_" + shmId + "_m_" + to_string(s.first)).c_str());
|
||||
void* ptr = segment.get_segment_manager();
|
||||
size_t size = segment.get_segment_manager()->get_size();
|
||||
new(ptr) segment_manager<char, rbtree_best_fit<mutex_family, offset_ptr<void>>, null_index>(size);
|
||||
} else {
|
||||
SimpleSeqFitSegment segment(open_only, std::string("fmq_" + shmId.shmId + "_m_" + to_string(s.first)).c_str());
|
||||
SimpleSeqFitSegment segment(open_only, std::string("fmq_" + shmId + "_m_" + to_string(s.first)).c_str());
|
||||
void* ptr = segment.get_segment_manager();
|
||||
size_t size = segment.get_segment_manager()->get_size();
|
||||
new(ptr) segment_manager<char, simple_seq_fit<mutex_family, offset_ptr<void>>, null_index>(size);
|
||||
}
|
||||
if (verbose) {
|
||||
cout << "Done." << endl;
|
||||
}
|
||||
} catch (bie& e) {
|
||||
if (verbose) {
|
||||
cout << "Error resetting content of segment '" << std::string("fmq_" + shmId.shmId + "_m_" + to_string(s.first)) << "': " << e.what() << endl;
|
||||
cout << "Error resetting content of segment '" << std::string("fmq_" + shmId + "_m_" + to_string(s.first)) << "': " << e.what() << endl;
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
cout << "Found management segment, but cannot locate segment info, something went wrong..." << endl;
|
||||
}
|
||||
|
||||
Uint16RegionInfoHashMap* shmRegions = managementSegment.find<Uint16RegionInfoHashMap>(bipc::unique_instance).first;
|
||||
if (shmRegions) {
|
||||
for (const auto& region : *shmRegions) {
|
||||
uint16_t id = region.first;
|
||||
Remove<bipc::message_queue>("fmq_" + shmId + "_rgq_" + to_string(id), verbose);
|
||||
}
|
||||
}
|
||||
} catch (bie& e) {
|
||||
if (verbose) {
|
||||
cout << "Could not find '" << managementSegmentName << "' segment. Nothing to cleanup." << endl;
|
||||
@@ -679,7 +789,7 @@ void Monitor::ResetContent(const ShmId& shmId, bool verbose /* = true */)
|
||||
}
|
||||
|
||||
if (verbose) {
|
||||
cout << "Done resetting segment content for shared memory id '" << shmId.shmId << "'." << endl;
|
||||
cout << "Done resetting segment content for shared memory id '" << shmId << "'." << endl;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -692,6 +802,50 @@ void Monitor::ResetContent(const SessionId& sessionId, bool verbose /* = true */
|
||||
ResetContent(shmId, verbose);
|
||||
}
|
||||
|
||||
void Monitor::ResetContent(const ShmId& shmIdT, const std::vector<SegmentConfig>& segmentCfgs, const std::vector<RegionConfig>& regionCfgs, bool verbose /* = true */)
|
||||
{
|
||||
using namespace boost::interprocess;
|
||||
|
||||
std::string shmId = shmIdT.shmId;
|
||||
std::string managementSegmentName("fmq_" + shmId + "_mng");
|
||||
// delete management segment
|
||||
cout << "deleting management segment" << endl;
|
||||
Remove<bipc::shared_memory_object>(managementSegmentName, verbose);
|
||||
// recreate management segment
|
||||
cout << "recreating management segment..." << endl;
|
||||
managed_shared_memory mngSegment(create_only, managementSegmentName.c_str(), kManagementSegmentSize);
|
||||
cout << "done." << endl;
|
||||
// fill management segment with segment & region infos
|
||||
cout << "filling management segment with managed segment configs..." << endl;
|
||||
for (const auto& s : segmentCfgs) {
|
||||
if (s.allocationAlgorithm == "rbtree_best_fit") {
|
||||
Segment::Register(shmId, s.id, AllocationAlgorithm::rbtree_best_fit);
|
||||
} else if (s.allocationAlgorithm == "simple_seq_fit") {
|
||||
Segment::Register(shmId, s.id, AllocationAlgorithm::simple_seq_fit);
|
||||
} else {
|
||||
LOG(error) << "Unknown allocation algorithm provided: " << s.allocationAlgorithm;
|
||||
throw MonitorError("Unknown allocation algorithm provided: " + s.allocationAlgorithm);
|
||||
}
|
||||
}
|
||||
cout << "done." << endl;
|
||||
cout << "filling management segment with unmanaged region configs..." << endl;
|
||||
for (const auto& r : regionCfgs) {
|
||||
fair::mq::shmem::UnmanagedRegion::Register(shmId, r);
|
||||
}
|
||||
cout << "done." << endl;
|
||||
// reset managed segments
|
||||
ResetContent(shmIdT, verbose);
|
||||
}
|
||||
|
||||
void Monitor::ResetContent(const SessionId& sessionId, const std::vector<SegmentConfig>& segmentCfgs, const std::vector<RegionConfig>& regionCfgs, bool verbose /* = true */)
|
||||
{
|
||||
ShmId shmId{makeShmIdStr(sessionId.sessionId)};
|
||||
if (verbose) {
|
||||
cout << "ResetContent called with session id '" << sessionId.sessionId << "', translating to shared memory id '" << shmId.shmId << "'" << endl;
|
||||
}
|
||||
ResetContent(shmId, segmentCfgs, regionCfgs, verbose);
|
||||
}
|
||||
|
||||
Monitor::~Monitor()
|
||||
{
|
||||
if (fSignalThread.joinable()) {
|
||||
|
@@ -8,6 +8,8 @@
|
||||
#ifndef FAIR_MQ_SHMEM_MONITOR_H_
|
||||
#define FAIR_MQ_SHMEM_MONITOR_H_
|
||||
|
||||
#include <fairmq/UnmanagedRegion.h>
|
||||
|
||||
#include <fairlogger/Logger.h>
|
||||
|
||||
#include <thread>
|
||||
@@ -49,6 +51,13 @@ struct BufferDebugInfo
|
||||
uint64_t fCreationTime;
|
||||
};
|
||||
|
||||
struct SegmentConfig
|
||||
{
|
||||
uint16_t id;
|
||||
uint64_t size;
|
||||
std::string allocationAlgorithm;
|
||||
};
|
||||
|
||||
class Monitor
|
||||
{
|
||||
public:
|
||||
@@ -88,6 +97,14 @@ class Monitor
|
||||
/// @param sessionId session id
|
||||
/// Only call this when segment is not in use
|
||||
static void ResetContent(const SessionId& sessionId, bool verbose = true);
|
||||
/// @brief [EXPERIMENTAL] cleanup the content of the shem segment, without recreating it
|
||||
/// @param shmId shared memory id
|
||||
/// Only call this when segment is not in use
|
||||
static void ResetContent(const ShmId& shmId, const std::vector<SegmentConfig>& segmentCfgs, const std::vector<RegionConfig>& regionCfgs, bool verbose = true);
|
||||
/// @brief [EXPERIMENTAL] cleanup the content of the shem segment, without recreating it
|
||||
/// @param sessionId session id
|
||||
/// Only call this when segment is not in use
|
||||
static void ResetContent(const SessionId& sessionId, const std::vector<SegmentConfig>& segmentCfgs, const std::vector<RegionConfig>& regionCfgs, bool verbose = true);
|
||||
|
||||
/// @brief Outputs list of messages in shmem (if compiled with FAIRMQ_DEBUG_MODE=ON)
|
||||
/// @param shmId shmem id
|
||||
@@ -102,7 +119,7 @@ class Monitor
|
||||
/// @param sessionId session id
|
||||
static std::unordered_map<uint16_t, std::vector<BufferDebugInfo>> GetDebugInfo(const SessionId& sessionId);
|
||||
/// @brief Returns the amount of free memory in the specified segment
|
||||
/// @param sessionId shmem id
|
||||
/// @param shmId shmem id
|
||||
/// @param segmentId segment id
|
||||
/// @throws MonitorError
|
||||
static unsigned long GetFreeMemory(const ShmId& shmId, uint16_t segmentId);
|
||||
@@ -111,6 +128,23 @@ class Monitor
|
||||
/// @param segmentId segment id
|
||||
/// @throws MonitorError
|
||||
static unsigned long GetFreeMemory(const SessionId& sessionId, uint16_t segmentId);
|
||||
/// @brief Checks if a given segment can be opened
|
||||
/// @param shmId shmem id
|
||||
/// @param segmentId segment id
|
||||
static bool SegmentIsPresent(const ShmId& shmId, uint16_t segmentId);
|
||||
/// @brief Checks if a given segment can be opened
|
||||
/// @param sessionId session id
|
||||
/// @param segmentId segment id
|
||||
static bool SegmentIsPresent(const SessionId& sessionId, uint16_t segmentId);
|
||||
/// @brief Checks if a given region can be opened
|
||||
/// @param shmId shmem id
|
||||
/// @param regionId region id
|
||||
static bool RegionIsPresent(const ShmId& shmId, uint16_t regionId);
|
||||
/// @brief Checks if a given region can be opened
|
||||
/// @param sessionId session id
|
||||
/// @param regionId region id
|
||||
static bool RegionIsPresent(const SessionId& sessionId, uint16_t regionId);
|
||||
|
||||
|
||||
static bool PrintShm(const ShmId& shmId);
|
||||
static void ListAll(const std::string& path);
|
||||
|
@@ -26,6 +26,8 @@ static const RBTreeBestFit rbTreeBestFit = RBTreeBestFit();
|
||||
|
||||
struct Segment
|
||||
{
|
||||
friend class Monitor;
|
||||
|
||||
Segment(const std::string& shmId, uint16_t id, size_t size, SimpleSeqFit)
|
||||
: fSegment(SimpleSeqFitSegment(boost::interprocess::open_or_create,
|
||||
std::string("fmq_" + shmId + "_m_" + std::to_string(id)).c_str(),
|
||||
@@ -66,15 +68,12 @@ struct Segment
|
||||
static void Register(const std::string& shmId, uint16_t id, AllocationAlgorithm allocAlgo)
|
||||
{
|
||||
using namespace boost::interprocess;
|
||||
managed_shared_memory mngSegment(open_or_create, std::string("fmq_" + shmId + "_mng").c_str(), 6553600);
|
||||
managed_shared_memory mngSegment(open_or_create, std::string("fmq_" + shmId + "_mng").c_str(), kManagementSegmentSize);
|
||||
VoidAlloc alloc(mngSegment.get_segment_manager());
|
||||
|
||||
Uint16SegmentInfoHashMap* shmSegments = mngSegment.find_or_construct<Uint16SegmentInfoHashMap>(unique_instance)(alloc);
|
||||
|
||||
EventCounter* eventCounter = mngSegment.find<EventCounter>(unique_instance).first;
|
||||
if (!eventCounter) {
|
||||
eventCounter = mngSegment.construct<EventCounter>(unique_instance)(0);
|
||||
}
|
||||
EventCounter* eventCounter = mngSegment.find_or_construct<EventCounter>(unique_instance)(0);
|
||||
|
||||
bool newSegmentRegistered = shmSegments->emplace(id, allocAlgo).second;
|
||||
if (newSegmentRegistered) {
|
||||
|
@@ -41,6 +41,7 @@ struct UnmanagedRegion
|
||||
{
|
||||
friend class Message;
|
||||
friend class Manager;
|
||||
friend class Monitor;
|
||||
|
||||
UnmanagedRegion(const std::string& shmId, uint16_t id, uint64_t size)
|
||||
: UnmanagedRegion(shmId, size, false, makeRegionConfig(id))
|
||||
@@ -50,6 +51,10 @@ struct UnmanagedRegion
|
||||
: UnmanagedRegion(shmId, size, false, std::move(cfg))
|
||||
{}
|
||||
|
||||
UnmanagedRegion(const std::string& shmId, RegionConfig cfg)
|
||||
: UnmanagedRegion(shmId, cfg.size, false, std::move(cfg))
|
||||
{}
|
||||
|
||||
UnmanagedRegion(const std::string& shmId, uint64_t size, bool remote, RegionConfig cfg)
|
||||
: fRemote(remote)
|
||||
, fRemoveOnDestruction(cfg.removeOnDestruction)
|
||||
@@ -66,6 +71,9 @@ struct UnmanagedRegion
|
||||
{
|
||||
using namespace boost::interprocess;
|
||||
|
||||
// TODO: refactor this
|
||||
cfg.size = size;
|
||||
|
||||
if (!cfg.path.empty()) {
|
||||
fName = std::string(cfg.path + fName);
|
||||
|
||||
@@ -91,16 +99,25 @@ struct UnmanagedRegion
|
||||
fRegion = mapped_region(fFileMapping, read_write, 0, size, 0, cfg.creationFlags);
|
||||
} else {
|
||||
try {
|
||||
fShmemObject = shared_memory_object(open_or_create, fName.c_str(), read_write);
|
||||
if (size != 0) {
|
||||
// if opening fails, create
|
||||
try {
|
||||
fShmemObject = shared_memory_object(open_only, fName.c_str(), read_write);
|
||||
} catch (interprocess_exception& e) {
|
||||
LOG(debug) << "Could not open " << (remote ? "remote" : "local") << " shared_memory_object for region id '" << cfg.id.value() << "': " << e.what() << ", creating...";
|
||||
fShmemObject = shared_memory_object(create_only, fName.c_str(), read_write);
|
||||
fShmemObject.truncate(size);
|
||||
}
|
||||
} catch (interprocess_exception& e) {
|
||||
LOG(error) << "Failed " << (remote ? "opening" : "creating") << " shared_memory_object for region id '" << cfg.id.value() << "': " << e.what();
|
||||
throw;
|
||||
}
|
||||
|
||||
try {
|
||||
fRegion = mapped_region(fShmemObject, read_write, 0, 0, 0, cfg.creationFlags);
|
||||
if (size != 0 && size != fRegion.get_size()) {
|
||||
LOG(error) << "Created/opened region size (" << fRegion.get_size() << ") does not match configured size (" << size << ")";
|
||||
throw TransportError(tools::ToString("Created/opened region size (", fRegion.get_size(), ") does not match configured size (", size, ")"));
|
||||
}
|
||||
} catch (interprocess_exception& e) {
|
||||
LOG(error) << "Failed mapping shared_memory_object for region id '" << cfg.id.value() << "': " << e.what();
|
||||
throw;
|
||||
@@ -223,20 +240,17 @@ struct UnmanagedRegion
|
||||
return regionCfg;
|
||||
}
|
||||
|
||||
static void Register(const std::string& shmId, RegionConfig& cfg)
|
||||
static void Register(const std::string& shmId, const RegionConfig& cfg)
|
||||
{
|
||||
using namespace boost::interprocess;
|
||||
managed_shared_memory mngSegment(open_or_create, std::string("fmq_" + shmId + "_mng").c_str(), 6553600);
|
||||
managed_shared_memory mngSegment(open_or_create, std::string("fmq_" + shmId + "_mng").c_str(), kManagementSegmentSize);
|
||||
VoidAlloc alloc(mngSegment.get_segment_manager());
|
||||
|
||||
Uint16RegionInfoHashMap* shmRegions = mngSegment.find_or_construct<Uint16RegionInfoHashMap>(unique_instance)(alloc);
|
||||
|
||||
EventCounter* eventCounter = mngSegment.find<EventCounter>(unique_instance).first;
|
||||
if (!eventCounter) {
|
||||
eventCounter = mngSegment.construct<EventCounter>(unique_instance)(0);
|
||||
}
|
||||
EventCounter* eventCounter = mngSegment.find_or_construct<EventCounter>(unique_instance)(0);
|
||||
|
||||
bool newShmRegionCreated = shmRegions->emplace(cfg.id.value(), RegionInfo(cfg.path.c_str(), cfg.creationFlags, cfg.userFlags, alloc)).second;
|
||||
bool newShmRegionCreated = shmRegions->emplace(cfg.id.value(), RegionInfo(cfg.path.c_str(), cfg.creationFlags, cfg.userFlags, cfg.size, alloc)).second;
|
||||
if (newShmRegionCreated) {
|
||||
(eventCounter->fCount)++;
|
||||
}
|
||||
|
@@ -27,7 +27,11 @@ inline bool Bind(void* socket, const std::string& address, const std::string& id
|
||||
if (errno == EADDRINUSE) {
|
||||
// do not print error in this case, this is handled upstream in case no
|
||||
// connection could be established after trying a number of random ports from a range.
|
||||
size_t protocolPos = address.find(':');
|
||||
std::string protocol = address.substr(0, protocolPos);
|
||||
if (protocol == "tcp") {
|
||||
return false;
|
||||
}
|
||||
} else if (errno == EACCES) {
|
||||
// check if TCP port 1 was given, if yes then it will be handeled upstream, print debug only
|
||||
size_t protocolPos = address.find(':');
|
||||
|
9
test/ci/fedora.35.def
Normal file
9
test/ci/fedora.35.def
Normal file
@@ -0,0 +1,9 @@
|
||||
Bootstrap: docker
|
||||
From: fedora:35
|
||||
|
||||
%post
|
||||
dnf -y update
|
||||
dnf -y install https://alfa-ci.gsi.de/packages/rpm/fedora-35-x86_64/fairsoft-release-dev.rpm
|
||||
dnf -y install clang cli11-devel pmix-devel ninja-build 'dnf-command(builddep)' libasan liblsan libtsan libubsan clang-tools-extra
|
||||
dnf -y builddep fairmq
|
||||
dnf -y clean all
|
@@ -25,6 +25,26 @@ namespace
|
||||
using namespace std;
|
||||
using namespace fair::mq;
|
||||
|
||||
void RegionsSizeMismatch()
|
||||
{
|
||||
size_t session = tools::UuidHash();
|
||||
|
||||
ProgOptions config;
|
||||
config.SetProperty<string>("session", to_string(session));
|
||||
config.SetProperty<size_t>("shm-segment-size", 100000000);
|
||||
|
||||
auto factory = TransportFactory::CreateTransportFactory("shmem", tools::Uuid(), &config);
|
||||
|
||||
fair::mq::RegionConfig rCfg;
|
||||
rCfg.id = 10;
|
||||
UnmanagedRegionPtr region1 = nullptr;
|
||||
ASSERT_NO_THROW(region1 = factory->CreateUnmanagedRegion(10000, [](void*, size_t, void*) {}, rCfg));
|
||||
ASSERT_NE(region1, nullptr);
|
||||
UnmanagedRegionPtr region2 = nullptr;
|
||||
ASSERT_THROW(region2 = factory->CreateUnmanagedRegion(16000, [](void*, size_t, void*) {}, rCfg), fair::mq::TransportError);
|
||||
ASSERT_EQ(region2, nullptr);
|
||||
}
|
||||
|
||||
void RegionsCache(const string& transport, const string& address)
|
||||
{
|
||||
size_t session1 = tools::UuidHash();
|
||||
@@ -226,6 +246,11 @@ void RegionCallbacks(const string& transport, const string& _address)
|
||||
LOG(info) << "2 done.";
|
||||
}
|
||||
|
||||
TEST(RegionsSizeMismatch, shmem)
|
||||
{
|
||||
RegionsSizeMismatch();
|
||||
}
|
||||
|
||||
TEST(Cache, zeromq)
|
||||
{
|
||||
RegionsCache("zeromq", "ipc://test_region_cache");
|
||||
|
Reference in New Issue
Block a user