Compare commits

..

16 Commits

Author SHA1 Message Date
Alexey Rybalchenko
9f8a3553ba Avoid deadlock in FairMQChannel operator=, handle self-assignment 2020-01-16 17:17:52 +01:00
Alexey Rybalchenko
692ec4e997 Fix CIDs 350447, 321250 (uncaught exception) 2020-01-16 17:17:52 +01:00
Alexey Rybalchenko
b6d9c949ae Fix CID 350448 (uncaught exception) 2020-01-16 17:17:52 +01:00
Alexey Rybalchenko
b6791856f9 Fix CID 350451 (uncaught exception) 2020-01-16 17:17:52 +01:00
Alexey Rybalchenko
a1e0814a92 Fix CIDs 350452, 323467 (missing_lock) 2020-01-16 17:17:52 +01:00
Alexey Rybalchenko
38bb14e556 Fix CID 350453 (uninit_member) 2020-01-16 17:17:52 +01:00
Alexey Rybalchenko
7187953604 Fix CID 350455 (uncaught exception) 2020-01-16 17:17:52 +01:00
Alexey Rybalchenko
c290c16896 PMIx: Add commands to plugin and command ui 2020-01-06 20:20:18 +01:00
Dennis Klein
fd2bac3e22 Modernize ctor
https://clang.llvm.org/extra/clang-tidy/checks/modernize-pass-by-value.html#pass-by-value-in-constructors
2020-01-06 20:20:18 +01:00
Alexey Rybalchenko
8e3f25851c Pass by const ref 2020-01-06 20:20:18 +01:00
Alexey Rybalchenko
ce937ca03e Bump linger time for multipart test 2019-12-20 14:04:58 +01:00
Alexey Rybalchenko
684e711b8b Shmem: track number of message objects, throw if non-zero at reset 2019-12-20 14:04:58 +01:00
Alexey Rybalchenko
5b5fecc994 Extend multipart tests to include single part, transfer across >1 channel 2019-12-20 14:04:58 +01:00
Alexey Rybalchenko
462a93b58b Add multipart arguments to the benchmark script 2019-12-20 14:04:58 +01:00
Alexey Rybalchenko
a2cff5b7bb Shmem: simplify message/socket and refactor to use namespaces 2019-12-20 14:04:58 +01:00
Alexey Rybalchenko
b2e027478e shmem: properly initialize received multipart messages 2019-12-16 21:17:06 +01:00
47 changed files with 1809 additions and 881 deletions

View File

@@ -18,7 +18,7 @@ get_git_version()
project(FairMQ VERSION ${PROJECT_VERSION} LANGUAGES CXX)
message(STATUS "${BWhite}${PROJECT_NAME}${CR} ${PROJECT_GIT_VERSION} from ${PROJECT_DATE}")
if(BUILD_OFI_TRANSPORT OR BUILD_SDK)
if(BUILD_OFI_TRANSPORT OR BUILD_SDK OR BUILD_PMIX_PLUGIN)
set(PROJECT_MIN_CXX_STANDARD 14)
else()
set(PROJECT_MIN_CXX_STANDARD 11)
@@ -44,7 +44,7 @@ fairmq_build_option(BUILD_SDK_COMMANDS "Build the FairMQ SDK commands."
fairmq_build_option(BUILD_DDS_PLUGIN "Build DDS plugin."
DEFAULT OFF REQUIRES "BUILD_FAIRMQ;BUILD_SDK_COMMANDS")
fairmq_build_option(BUILD_PMIX_PLUGIN "Build PMIx plugin."
DEFAULT OFF REQUIRES "BUILD_FAIRMQ")
DEFAULT OFF REQUIRES "BUILD_FAIRMQ;BUILD_SDK_COMMANDS")
fairmq_build_option(BUILD_EXAMPLES "Build FairMQ examples."
DEFAULT ON REQUIRES "BUILD_FAIRMQ")
fairmq_build_option(BUILD_SDK "Build the FairMQ controller SDK."
@@ -397,9 +397,9 @@ else()
endif()
message(STATUS " ${BWhite}dds_plugin${CR} ${dds_summary}")
if(BUILD_PMIX_PLUGIN)
set(pmix_summary "${BGreen}YES${CR} (disable with ${BMagenta}-DBUILD_PMIX_PLUGIN=OFF${CR})")
set(pmix_summary "${BGreen}YES${CR} EXPERIMENTAL (requires C++14) (disable with ${BMagenta}-DBUILD_PMIX_PLUGIN=OFF${CR})")
else()
set(pmix_summary "${BRed} NO${CR} (default, enable with ${BMagenta}-DBUILD_PMIX_PLUGIN=ON${CR})")
set(pmix_summary "${BRed} NO${CR} EXPERIMENTAL (requires C++14) (default, enable with ${BMagenta}-DBUILD_PMIX_PLUGIN=ON${CR})")
endif()
message(STATUS " ${BWhite}pmix_plugin${CR} ${pmix_summary}")
if(BUILD_EXAMPLES)

View File

@@ -20,7 +20,7 @@ SAMPLER+=" --verbosity veryhigh"
SAMPLER+=" --session $SESSION"
SAMPLER+=" --max-iterations 1"
SAMPLER+=" --control static --color false"
SAMPLER+=" --channel-config name=data,type=pair,method=connect,rateLogging=0,address=tcp://127.0.0.1:5555"
SAMPLER+=" --channel-config name=data,type=pair,method=connect,rateLogging=0,address=tcp://127.0.0.1:5555,linger=1000"
@CMAKE_CURRENT_BINARY_DIR@/$SAMPLER &
SAMPLER_PID=$!

View File

@@ -6,16 +6,6 @@
# copied verbatim in the file "LICENSE" #
################################################################################
####################
# external plugins #
####################
if(BUILD_DDS_PLUGIN)
add_subdirectory(plugins/DDS)
endif()
if(BUILD_PMIX_PLUGIN)
add_subdirectory(plugins/PMIx)
endif()
if(BUILD_FAIRMQ OR BUILD_SDK)
###########
# Version #
@@ -192,11 +182,11 @@ if(BUILD_FAIRMQ)
plugins/Builtin.h
plugins/config/Config.h
plugins/Control.h
shmem/FairMQMessageSHM.h
shmem/FairMQPollerSHM.h
shmem/FairMQUnmanagedRegionSHM.h
shmem/FairMQSocketSHM.h
shmem/FairMQTransportFactorySHM.h
shmem/Message.h
shmem/Poller.h
shmem/UnmanagedRegion.h
shmem/Socket.h
shmem/TransportFactory.h
shmem/Common.h
shmem/Manager.h
shmem/Region.h
@@ -253,11 +243,11 @@ if(BUILD_FAIRMQ)
SuboptParser.cxx
plugins/config/Config.cxx
plugins/Control.cxx
shmem/FairMQMessageSHM.cxx
shmem/FairMQPollerSHM.cxx
shmem/FairMQUnmanagedRegionSHM.cxx
shmem/FairMQSocketSHM.cxx
shmem/FairMQTransportFactorySHM.cxx
shmem/Message.cxx
shmem/Poller.cxx
shmem/UnmanagedRegion.cxx
shmem/Socket.cxx
shmem/TransportFactory.cxx
shmem/Manager.cxx
shmem/Region.cxx
zeromq/FairMQMessageZMQ.cxx
@@ -473,3 +463,13 @@ endif()
if(BUILD_SDK)
add_subdirectory(sdk)
endif()
####################
# external plugins #
####################
if(BUILD_DDS_PLUGIN)
add_subdirectory(plugins/DDS)
endif()
if(BUILD_PMIX_PLUGIN)
add_subdirectory(plugins/PMIx)
endif()

View File

@@ -75,7 +75,7 @@ void DeviceRunner::SubscribeForConfigChange()
fair::Logger::SetConsoleColor(val);
}
});
fConfig.Subscribe<string>("device-runner", [&](const std::string& key, const std::string val) {
fConfig.Subscribe<string>("device-runner", [&](const std::string& key, const std::string& val) {
if (key == "severity") {
fair::Logger::SetConsoleSeverity(val);
} else if (key == "file-severity") {

View File

@@ -20,7 +20,7 @@ using namespace std;
using namespace fair::mq;
template<typename T>
T GetPropertyOrDefault(const fair::mq::Properties& m, const string& k, const T& ifNotFound) noexcept
T GetPropertyOrDefault(const fair::mq::Properties& m, const string& k, const T& ifNotFound)
{
if (m.count(k)) {
return boost::any_cast<T>(m.at(k));
@@ -133,26 +133,37 @@ FairMQChannel::FairMQChannel(const FairMQChannel& chan, const string& newName)
FairMQChannel& FairMQChannel::operator=(const FairMQChannel& chan)
{
fTransportFactory = nullptr;
fTransportType = chan.fTransportType;
fSocket = nullptr;
fName = chan.fName;
fType = chan.fType;
fMethod = chan.fMethod;
fAddress = chan.fAddress;
fSndBufSize = chan.fSndBufSize;
fRcvBufSize = chan.fRcvBufSize;
fSndKernelSize = chan.fSndKernelSize;
fRcvKernelSize = chan.fRcvKernelSize;
fLinger = chan.fLinger;
fRateLogging = chan.fRateLogging;
fPortRangeMin = chan.fPortRangeMin;
fPortRangeMax = chan.fPortRangeMax;
fAutoBind = chan.fAutoBind;
fIsValid = false;
fMultipart = chan.fMultipart;
fModified = chan.fModified;
fReset = false;
if (this == &chan) {
return *this;
}
{
// TODO: replace this with std::scoped_lock (c++17)
lock(fMtx, chan.fMtx);
lock_guard<mutex> lock1(fMtx, adopt_lock);
lock_guard<mutex> lock2(chan.fMtx, adopt_lock);
fTransportFactory = nullptr;
fTransportType = chan.fTransportType;
fSocket = nullptr;
fName = chan.fName;
fType = chan.fType;
fMethod = chan.fMethod;
fAddress = chan.fAddress;
fSndBufSize = chan.fSndBufSize;
fRcvBufSize = chan.fRcvBufSize;
fSndKernelSize = chan.fSndKernelSize;
fRcvKernelSize = chan.fRcvKernelSize;
fLinger = chan.fLinger;
fRateLogging = chan.fRateLogging;
fPortRangeMin = chan.fPortRangeMin;
fPortRangeMax = chan.fPortRangeMax;
fAutoBind = chan.fAutoBind;
fIsValid = false;
fMultipart = chan.fMultipart;
fModified = chan.fModified;
fReset = false;
}
return *this;
}

View File

@@ -31,8 +31,8 @@ namespace fair
namespace mq
{
using PollerPtr = std::unique_ptr<FairMQPoller>;
using Poller = FairMQPoller;
using PollerPtr = FairMQPollerPtr;
struct PollerError : std::runtime_error { using std::runtime_error::runtime_error; };
} /* namespace mq */

View File

@@ -22,7 +22,7 @@ class FairMQSocket
FairMQSocket() {}
FairMQSocket(FairMQTransportFactory* fac): fTransport(fac) {}
virtual std::string GetId() = 0;
virtual std::string GetId() const = 0;
virtual bool Bind(const std::string& address) = 0;
virtual bool Connect(const std::string& address) = 0;

View File

@@ -8,7 +8,7 @@
#include <FairMQTransportFactory.h>
#include <zeromq/FairMQTransportFactoryZMQ.h>
#include <shmem/FairMQTransportFactorySHM.h>
#include <fairmq/shmem/TransportFactory.h>
#ifdef BUILD_NANOMSG_TRANSPORT
#include <nanomsg/FairMQTransportFactoryNN.h>
#endif /* BUILD_NANOMSG_TRANSPORT */
@@ -44,7 +44,7 @@ auto FairMQTransportFactory::CreateTransportFactory(const std::string& type, con
}
else if (type == "shmem")
{
return make_shared<FairMQTransportFactorySHM>(finalId, config);
return make_shared<fair::mq::shmem::TransportFactory>(finalId, config);
}
#ifdef BUILD_NANOMSG_TRANSPORT
else if (type == "nanomsg")

View File

@@ -137,6 +137,7 @@ namespace fair
namespace mq
{
using TransportFactory = FairMQTransportFactory;
struct TransportFactoryError : std::runtime_error { using std::runtime_error::runtime_error; };
} /* namespace mq */

View File

@@ -31,7 +31,9 @@ namespace fair
namespace mq
{
using UnmanagedRegionPtr = std::unique_ptr<FairMQUnmanagedRegion>;
using RegionCallback = FairMQRegionCallback;
using UnmanagedRegion = FairMQUnmanagedRegion;
using UnmanagedRegionPtr = FairMQUnmanagedRegionPtr;
} /* namespace mq */
} /* namespace fair */

View File

@@ -1,5 +1,5 @@
/********************************************************************************
* Copyright (C) 2017 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* Copyright (C) 2017-2019 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence (LGPL) version 3, *
@@ -8,15 +8,20 @@
#include <fairmq/Plugin.h>
#include <FairMQLogger.h>
#include <utility>
using namespace std;
fair::mq::Plugin::Plugin(const string name, const Version version, const string maintainer, const string homepage, PluginServices* pluginServices)
: fkName{name}
fair::mq::Plugin::Plugin(string name,
Version version,
string maintainer,
string homepage,
PluginServices* pluginServices)
: fkName(std::move(name))
, fkVersion(version)
, fkMaintainer{maintainer}
, fkHomepage{homepage}
, fPluginServices{pluginServices}
, fkMaintainer(std::move(maintainer))
, fkHomepage(std::move(homepage))
, fPluginServices(pluginServices)
{
LOG(debug) << "Loaded plugin: " << *this;
}

View File

@@ -1,5 +1,5 @@
/********************************************************************************
* Copyright (C) 2017 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* Copyright (C) 2017-2019 GSI Helmholtzzentrum fuer Schwerionenforschung GmbH *
* *
* This software is distributed under the terms of the *
* GNU Lesser General Public Licence (LGPL) version 3, *
@@ -44,7 +44,11 @@ class Plugin
using Version = tools::Version;
Plugin() = delete;
Plugin(const std::string name, const Version version, const std::string maintainer, const std::string homepage, PluginServices* pluginServices);
Plugin(std::string name,
Version version,
std::string maintainer,
std::string homepage,
PluginServices* pluginServices);
Plugin(const Plugin&) = delete;
Plugin operator=(const Plugin&) = delete;

View File

@@ -79,7 +79,9 @@ struct Machine_ : public state_machine_def<Machine_>
{
public:
Machine_()
: fLastTransitionResult(true)
: fState(State::Ok)
, fNewState(State::Ok)
, fLastTransitionResult(true)
, fNewStatePending(false)
{}

View File

@@ -91,11 +91,6 @@ FairMQSocketNN::FairMQSocketNN(const string& type, const string& name, const str
LOG(debug) << "Created socket " << GetId();
}
string FairMQSocketNN::GetId()
{
return fId;
}
bool FairMQSocketNN::Bind(const string& address)
{
// LOG(info) << "bind socket " << fId << " on " << address;

View File

@@ -23,7 +23,7 @@ class FairMQSocketNN final : public FairMQSocket
FairMQSocketNN(const FairMQSocketNN&) = delete;
FairMQSocketNN operator=(const FairMQSocketNN&) = delete;
std::string GetId() override;
std::string GetId() const override { return fId; }
bool Bind(const std::string& address) override;
bool Connect(const std::string& address) override;

View File

@@ -43,7 +43,7 @@ class Socket final : public fair::mq::Socket
Socket(const Socket&) = delete;
Socket operator=(const Socket&) = delete;
auto GetId() -> std::string { return fId; }
auto GetId() const -> std::string override { return fId; }
auto Bind(const std::string& address) -> bool override;
auto Connect(const std::string& address) -> bool override;

View File

@@ -10,9 +10,10 @@ set(plugin FairMQPlugin_pmix)
add_library(${plugin} SHARED
${CMAKE_CURRENT_SOURCE_DIR}/PMIxPlugin.cxx
${CMAKE_CURRENT_SOURCE_DIR}/PMIxPlugin.h
${CMAKE_CURRENT_SOURCE_DIR}/PMIxCommands.h
${CMAKE_CURRENT_SOURCE_DIR}/PMIx.hpp
)
target_link_libraries(${plugin} FairMQ PMIx::libpmix)
target_link_libraries(${plugin} PUBLIC FairMQ PMIx::libpmix PRIVATE Commands)
target_include_directories(${plugin} PRIVATE ${CMAKE_CURRENT_BINARY_DIR})
set_target_properties(${plugin} PROPERTIES
CXX_VISIBILITY_PRESET hidden
@@ -20,7 +21,12 @@ set_target_properties(${plugin} PROPERTIES
SOVERSION "${PROJECT_VERSION_MAJOR}.${PROJECT_VERSION_MINOR}"
)
install(TARGETS ${plugin}
set(exe fairmq-pmix-command-ui)
add_executable(${exe} ${CMAKE_CURRENT_SOURCE_DIR}/runPMIxCommandUI.cxx)
target_link_libraries(${exe} FairMQ Commands StateMachine PMIx::libpmix)
target_include_directories(${exe} PRIVATE ${CMAKE_CURRENT_BINARY_DIR})
install(TARGETS ${plugin} ${exe}
EXPORT ${PROJECT_EXPORT_SET}
LIBRARY DESTINATION ${PROJECT_INSTALL_LIBDIR}
RUNTIME DESTINATION ${PROJECT_INSTALL_BINDIR}

View File

@@ -9,17 +9,18 @@
#ifndef PMIX_HPP
#define PMIX_HPP
#include <array>
#include <cstring>
#include <functional>
#include <limits>
#include <memory>
#include <ostream>
#include <pmix.h>
#include <sstream>
#include <stdexcept>
#include <string.h>
#include <type_traits>
#include <utility>
#include <vector>
#include <FairMQLogger.h>
// C++ PMIx v2.2 API
namespace pmix
@@ -69,7 +70,7 @@ struct proc : pmix_proc_t
friend std::ostream& operator<<(std::ostream& os, const proc& p)
{
return os << "nspace=" << p.nspace << ",rank=" << p.rank;
return os << p.nspace << "_" << p.rank;
}
};
@@ -80,7 +81,6 @@ struct value : pmix_value_t
value(const value& rhs)
{
LOG(warn) << "copy ctor";
status rc;
auto lhs(static_cast<pmix_value_t*>(this));
PMIX_VALUE_XFER(rc, lhs, static_cast<pmix_value_t*>(const_cast<value*>(&rhs)));
@@ -111,6 +111,11 @@ struct value : pmix_value_t
{
PMIX_VALUE_LOAD(static_cast<pmix_value_t*>(this), &val, PMIX_INT);
}
explicit value(pmix_data_array_t* val)
{
PMIX_VALUE_LOAD(static_cast<pmix_value_t*>(this), val, PMIX_DATA_ARRAY);
}
};
struct info : pmix_info_t
@@ -134,9 +139,15 @@ struct info : pmix_info_t
}
}
friend std::ostream& operator<<(std::ostream& os, const info& p)
friend std::ostream& operator<<(std::ostream& os, const info& i)
{
return os << "key=" << p.key << ",value='" << p.value.data.string << "'";
return os << "key=" << i.key << ",value='" << i.value.data.string << "'";
}
info(const info& rhs)
{
PMIX_INFO_XFER(static_cast<pmix_info_t*>(this),
static_cast<pmix_info_t*>(const_cast<info*>(&rhs)));
}
};
@@ -144,6 +155,7 @@ struct pdata : pmix_pdata_t
{
pdata() { PMIX_PDATA_CONSTRUCT(static_cast<pmix_pdata_t*>(this)); }
~pdata() { PMIX_PDATA_DESTRUCT(static_cast<pmix_pdata_t*>(this)); }
pdata(const pdata& rhs)
{
PMIX_PDATA_XFER(static_cast<pmix_pdata_t*>(this),
@@ -171,7 +183,7 @@ auto init(const std::vector<info>& info = {}) -> proc
auto initialized() -> bool { return !!PMIx_Initialized(); }
auto get_version() -> const char* { return PMIx_Get_version(); }
auto get_version() -> std::string { return {PMIx_Get_version()}; }
auto finalize(const std::vector<info>& info = {}) -> void
{
@@ -213,6 +225,92 @@ auto lookup(std::vector<pdata>& pdata, const std::vector<info>& info = {}) -> vo
}
}
std::string get_info(const std::string& name, pmix::proc& process)
{
pmix_value_t* v;
pmix::status rc = PMIx_Get(&process, name.c_str(), nullptr, 0, &v);
if (rc == PMIX_SUCCESS) {
std::stringstream ss;
switch (v->type) {
case PMIX_SIZE: ss << static_cast<size_t>(v->data.size) << " (size_t)"; break;
case PMIX_INT: ss << static_cast<int>(v->data.integer) << " (int)"; break;
case PMIX_INT8: ss << static_cast<int8_t>(v->data.int8) << " (int8_t)"; break;
case PMIX_INT16: ss << static_cast<int16_t>(v->data.int16) << " (int16_t)"; break;
case PMIX_INT32: ss << static_cast<int32_t>(v->data.int32) << " (int32_t)"; break;
case PMIX_INT64: ss << static_cast<int64_t>(v->data.int64) << " (int64_t)"; break;
case PMIX_UINT: ss << static_cast<unsigned int>(v->data.uint) << " (unsigned int)"; break;
case PMIX_UINT8: ss << static_cast<uint8_t>(v->data.uint8) << " (uint8_t)"; break;
case PMIX_UINT16: ss << static_cast<uint16_t>(v->data.uint16) << " (uint16_t)"; break;
case PMIX_UINT32: ss << static_cast<uint32_t>(v->data.uint32) << " (uint32_t)"; break;
case PMIX_UINT64: ss << static_cast<uint64_t>(v->data.uint64) << " (uint64_t)"; break;
case PMIX_FLOAT: ss << static_cast<float>(v->data.fval) << " (float)"; break;
case PMIX_DOUBLE: ss << static_cast<double>(v->data.dval) << " (double)"; break;
case PMIX_PID: ss << static_cast<pid_t>(v->data.pid) << " (pid_t)"; break;
case PMIX_STRING: ss << static_cast<char*>(v->data.string) << " (string)"; break;
case PMIX_PROC_RANK: ss << static_cast<uint32_t>(v->data.rank) << " (pmix_rank_t)"; break;
case PMIX_PROC: ss << "proc.nspace: " << static_cast<pmix_proc_t*>(v->data.proc)->nspace
<< ", proc.rank: " << static_cast<pmix_proc_t*>(v->data.proc)->rank << " (pmix_proc_t*)"; break;
default:
ss << "unknown type: " << v->type;
break;
}
return ss.str();
} else if (rc == PMIX_ERR_NOT_FOUND) {
// LOG(error) << "PMIx_Get failed: PMIX_ERR_NOT_FOUND";
return "";
} else {
// LOG(error) << "PMIx_Get failed: " << rc;
return "<undefined>";
}
}
std::string get_value_str(const pmix_value_t& v)
{
switch (v.type) {
case PMIX_BOOL: return std::to_string(static_cast<bool>(v.data.flag));
case PMIX_SIZE: return std::to_string(static_cast<size_t>(v.data.size));
case PMIX_INT: return std::to_string(static_cast<int>(v.data.integer));
case PMIX_INT8: return std::to_string(static_cast<int8_t>(v.data.int8));
case PMIX_INT16: return std::to_string(static_cast<int16_t>(v.data.int16));
case PMIX_INT32: return std::to_string(static_cast<int32_t>(v.data.int32));
case PMIX_INT64: return std::to_string(static_cast<int64_t>(v.data.int64));
case PMIX_UINT: return std::to_string(static_cast<unsigned int>(v.data.uint));
case PMIX_UINT8: return std::to_string(static_cast<uint8_t>(v.data.uint8));
case PMIX_UINT16: return std::to_string(static_cast<uint16_t>(v.data.uint16));
case PMIX_UINT32: return std::to_string(static_cast<uint32_t>(v.data.uint32));
case PMIX_UINT64: return std::to_string(static_cast<uint64_t>(v.data.uint64));
case PMIX_FLOAT: return std::to_string(static_cast<float>(v.data.fval));
case PMIX_DOUBLE: return std::to_string(static_cast<double>(v.data.dval));
case PMIX_PID: return std::to_string(static_cast<pid_t>(v.data.pid));
case PMIX_STRING: return static_cast<char*>(v.data.string);
case PMIX_PROC_RANK: return std::to_string(static_cast<uint32_t>(v.data.rank));
case PMIX_POINTER: { std::stringstream ss; ss << static_cast<void*>(v.data.ptr); return ss.str(); }
case PMIX_DATA_ARRAY: {
if (v.data.darray->type == PMIX_PROC) {
std::stringstream ss;
ss << "[";
for (size_t i = 0; i < v.data.darray->size; ++i) {
ss << static_cast<pmix_proc_t*>(static_cast<pmix_data_array_t*>(v.data.darray)->array)[0].nspace;
ss << "_";
ss << static_cast<pmix_proc_t*>(static_cast<pmix_data_array_t*>(v.data.darray)->array)[0].rank;
if (i < v.data.darray->size - 1) {
ss << ",";
}
}
ss << "]";
return ss.str();
} else {
return "UNKNOWN TYPE IN DATA ARRAY";
}
}
default: return "UNKNOWN TYPE";
}
}
} /* namespace pmix */
#endif /* PMIX_HPP */

View File

@@ -0,0 +1,291 @@
/********************************************************************************
* Copyright (C) 2019 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" *
********************************************************************************/
#ifndef PMIXCOMMANDS_H
#define PMIXCOMMANDS_H
#include "PMIx.hpp"
#include <FairMQLogger.h>
#include <fairmq/tools/Semaphore.h>
#include <fairmq/tools/CppSTL.h>
#include <string>
namespace pmix
{
std::array<std::string, 47> typeNames =
{
{
"PMIX_UNDEF",
"PMIX_BOOL",
"PMIX_BYTE",
"PMIX_STRING",
"PMIX_SIZE",
"PMIX_PID",
"PMIX_INT",
"PMIX_INT8",
"PMIX_INT16",
"PMIX_INT32",
"PMIX_INT64",
"PMIX_UINT",
"PMIX_UINT8",
"PMIX_UINT16",
"PMIX_UINT32",
"PMIX_UINT64",
"PMIX_FLOAT",
"PMIX_DOUBLE",
"PMIX_TIMEVAL",
"PMIX_TIME",
"PMIX_STATUS",
"PMIX_VALUE",
"PMIX_PROC",
"PMIX_APP",
"PMIX_INFO",
"PMIX_PDATA",
"PMIX_BUFFER",
"PMIX_BYTE_OBJECT",
"PMIX_KVAL",
"PMIX_MODEX",
"PMIX_PERSIST",
"PMIX_POINTER",
"PMIX_SCOPE",
"PMIX_DATA_RANGE",
"PMIX_COMMAND",
"PMIX_INFO_DIRECTIVES",
"PMIX_DATA_TYPE",
"PMIX_PROC_STATE",
"PMIX_PROC_INFO",
"PMIX_DATA_ARRAY",
"PMIX_PROC_RANK",
"PMIX_QUERY",
"PMIX_COMPRESSED_STRING",
"PMIX_ALLOC_DIRECTIVE",
"PMIX_INFO_ARRAY",
"PMIX_IOF_CHANNEL",
"PMIX_ENVAR"
}
};
enum class Command : int
{
general = PMIX_EXTERNAL_ERR_BASE,
error = PMIX_EXTERNAL_ERR_BASE - 1
};
class Commands
{
public:
Commands(const proc& process)
: fProcess(process)
, fSubscribed(false)
{
}
~Commands()
{
Unsubscribe();
}
void Subscribe(std::function<void(const std::string& msg, const proc& sender)> callback)
{
using namespace std::placeholders;
LOG(debug) << "PMIxCommands: Subscribing...";
fCallback = callback;
std::array<pmix::status, 1> codes;
codes[0] = static_cast<int>(pmix::Command::general);
PMIX_INFO_LOAD(&(fInfos[0]), PMIX_EVENT_RETURN_OBJECT, this, PMIX_POINTER);
PMIx_Register_event_handler(codes.data(), codes.size(),
fInfos.data(), fInfos.size(),
&Commands::Handler,
&Commands::EventHandlerRegistration,
this);
fBlocker.Wait();
LOG(debug) << "PMIxCommands: Subscribing complete!";
}
void Unsubscribe()
{
if (fSubscribed) {
LOG(debug) << "PMIxCommands: Unsubscribing...";
PMIx_Deregister_event_handler(fHandlerRef, &Commands::EventHandlerDeregistration, this);
fBlocker.Wait();
LOG(debug) << "PMIxCommands: Unsubscribing complete!";
} else {
LOG(debug) << "Unsubscribe() is called while no subscription is active";
}
}
struct Holder
{
Holder() : fData(nullptr) {}
~Holder() { PMIX_DATA_ARRAY_FREE(fData); }
std::vector<pmix::info> fInfos;
pmix_data_array_t* fData;
};
void Send(const std::string& msg)
{
std::vector<pmix::info>* infos = new std::vector<pmix::info>();
infos->emplace_back("fairmq.cmd", msg);
PMIx_Notify_event(static_cast<int>(pmix::Command::general),
&fProcess,
PMIX_RANGE_NAMESPACE,
infos->data(), infos->size(),
&Commands::OpCompleteCallback<std::vector<pmix::info>>,
infos);
}
void Send(const std::string& msg, rank rank)
{
pmix::proc destination(fProcess);
destination.rank = rank;
Send(msg, {destination});
}
void Send(const std::string& msg, const std::vector<proc>& destination)
{
std::unique_ptr<Holder> holder = fair::mq::tools::make_unique<Holder>();
PMIX_DATA_ARRAY_CREATE(holder->fData, destination.size(), PMIX_PROC);
memcpy(holder->fData->array, destination.data(), destination.size() * sizeof(pmix_proc_t));
// LOG(warn) << "OLOG: " << msg << " > " << static_cast<pmix_proc_t*>(holder->fData->array)[0].nspace << ": " << static_cast<pmix_proc_t*>(holder->fData->array)[0].rank;
holder->fInfos.emplace_back(PMIX_EVENT_CUSTOM_RANGE, holder->fData);
// LOG(warn) << msg << " // packed range: " << static_cast<pmix_proc_t*>(static_cast<pmix_data_array_t*>(holder->fInfos.at(0).value.data.darray)->array)[0].nspace << "_" << static_cast<pmix_proc_t*>(static_cast<pmix_data_array_t*>(holder->fInfos.at(0).value.data.darray)->array)[0].rank;
// LOG(warn) << msg << " // packed range.type: " << pmix::typeNames.at(holder->fInfos.at(0).value.type);
// LOG(warn) << msg << " // packed range.array.type: " << pmix::typeNames.at(static_cast<pmix_data_array_t*>(holder->fInfos.at(0).value.data.darray)->type);
// LOG(warn) << msg << " // packed range.array.size: " << static_cast<pmix_data_array_t*>(holder->fInfos.at(0).value.data.darray)->size;
// LOG(warn) << holder->fInfos.size();
holder->fInfos.emplace_back("fairmq.cmd", msg);
// LOG(warn) << msg << " // packed msg: " << holder->fInfos.at(1).value.data.string;
// LOG(warn) << msg << " // packed msg.type: " << pmix::typeNames.at(holder->fInfos.at(1).value.type);
// LOG(warn) << holder->fInfos.size();
PMIx_Notify_event(static_cast<int>(pmix::Command::general),
&fProcess,
PMIX_RANGE_CUSTOM,
holder->fInfos.data(), holder->fInfos.size(),
&Commands::OpCompleteCallback<Holder>,
holder.get());
holder.release();
}
private:
static void EventHandlerRegistration(pmix_status_t s, size_t handlerRef, void* obj)
{
if (s == PMIX_SUCCESS) {
LOG(debug) << "Successfully registered event handler, reference = " << static_cast<unsigned long>(handlerRef);
static_cast<Commands*>(obj)->fHandlerRef = handlerRef;
static_cast<Commands*>(obj)->fSubscribed = true;
} else {
LOG(error) << "Could not register PMIx event handler, status = " << s;
}
static_cast<Commands*>(obj)->fBlocker.Signal();
}
static void EventHandlerDeregistration(pmix_status_t s, void* obj)
{
if (s == PMIX_SUCCESS) {
LOG(debug) << "Successfully deregistered event handler, reference = " << static_cast<Commands*>(obj)->fHandlerRef;
static_cast<Commands*>(obj)->fSubscribed = false;
} else {
LOG(error) << "Could not deregister PMIx event handler, reference = " << static_cast<Commands*>(obj)->fHandlerRef << ", status = " << s;
}
static_cast<Commands*>(obj)->fBlocker.Signal();
}
template<typename T>
static void OpCompleteCallback(pmix_status_t s, void* data)
{
if (s == PMIX_SUCCESS) {
// LOG(info) << "Operation completed successfully";
} else {
LOG(error) << "Could not complete operation, status = " << s;
}
if (data) {
// LOG(warn) << "Destroying event data...";
delete static_cast<T*>(data);
}
}
static void Handler(size_t handlerId,
pmix_status_t s,
const pmix_proc_t* src,
pmix_info_t info[], size_t ninfo,
pmix_info_t[] /* results */, size_t nresults,
pmix_event_notification_cbfunc_fn_t cbfunc,
void* cbdata)
{
std::stringstream ss;
ss << "Event handler called with "
<< "status: " << s << ", "
<< "source: " << src->nspace << "_" << src->rank << ", "
<< "ninfo: " << ninfo << ", "
<< "nresults: " << nresults << ", "
<< "handlerId: " << handlerId;
std::string msg;
Commands* obj = nullptr;
if (ninfo > 0) {
ss << ":\n";
for (size_t i = 0; i < ninfo; ++i) {
ss << " [" << i << "]: key: '" << info[i].key
<< "', value: '" << pmix::get_value_str(info[i].value)
<< "', value.type: '" << pmix::typeNames.at(info[i].value.type)
<< "', flags: " << info[i].flags;
if (std::strcmp(info[i].key, "fairmq.cmd") == 0) {
msg = pmix::get_value_str(info[i].value);
}
if (std::strcmp(info[i].key, PMIX_EVENT_RETURN_OBJECT) == 0) {
obj = static_cast<Commands*>(info[i].value.data.ptr);
}
if (i < ninfo - 1) {
ss << "\n";
}
}
}
if (obj != nullptr) {
if (static_cast<Commands*>(obj)->fProcess.rank != src->rank) {
// LOG(warn) << ss.str();
static_cast<Commands*>(obj)->fCallback(msg, proc(const_cast<char*>(src->nspace), rank(src->rank)));
} else {
// LOG(trace) << "suppressing message from itself";
}
} else {
LOG(ERROR) << "ERROR";
}
if (cbfunc != nullptr) {
cbfunc(PMIX_SUCCESS, nullptr, 0, nullptr, nullptr, cbdata);
}
}
const proc& fProcess;
size_t fHandlerRef;
std::function<void(const std::string& msg, const proc& sender)> fCallback;
std::array<pmix_info_t, 1> fInfos;
bool fSubscribed;
fair::mq::tools::SharedSemaphore fBlocker;
};
} /* namespace pmix */
#endif /* PMIXCOMMANDS_H */

View File

@@ -8,8 +8,15 @@
#include "PMIxPlugin.h"
#include <fairmq/sdk/commands/Commands.h>
#include <fairmq/Tools.h>
#include <sstream>
#include <stdexcept>
#include <cstdint> // UINT32_MAX
using namespace std;
using namespace fair::mq::sdk::cmd;
namespace fair
{
@@ -18,47 +25,99 @@ namespace mq
namespace plugins
{
PMIxPlugin::PMIxPlugin(const std::string& name,
PMIxPlugin::PMIxPlugin(const string& name,
const Plugin::Version version,
const std::string& maintainer,
const std::string& homepage,
const string& maintainer,
const string& homepage,
PluginServices* pluginServices)
: Plugin(name, version, maintainer, homepage, pluginServices)
, fProcess(Init())
, fPid(getpid())
, fPMIxClient(tools::ToString("PMIx client(pid=", fPid, ") "))
, fDeviceId(string(fProcess.nspace) + "_" + to_string(fProcess.rank))
, fCommands(fProcess)
, fLastExternalController(UINT32_MAX)
, fExitingAckedByLastExternalController(false)
, fCurrentState(DeviceState::Idle)
, fLastState(DeviceState::Idle)
{
Init();
SetProperty<std::string>("id", std::string(fProc.nspace) + "_" + std::to_string(fProc.rank));
Fence();
TakeDeviceControl();
LOG(debug) << PMIxClient() << "pmix::init() OK: " << fProcess << ", version=" << pmix::get_version();
SetProperty<string>("id", fDeviceId);
SubscribeToDeviceStateChange([&](DeviceState newState) {
Fence("pmix::init");
SubscribeForCommands();
Fence("subscribed");
// fCommands.Send("test1");
// fCommands.Send("test2", 0);
// fCommands.Send("test3", 0);
// LOG(info) << "PMIX_EXTERNAL_ERR_BASE: " << PMIX_EXTERNAL_ERR_BASE;
// job level infos
// LOG(info) << "PMIX_SESSION_ID: " << pmix::getInfo(PMIX_SESSION_ID, fProcess);
// LOG(info) << "PMIX_UNIV_SIZE: " << pmix::getInfo(PMIX_UNIV_SIZE, fProcess);
// LOG(info) << "PMIX_JOB_SIZE: " << pmix::getInfo(PMIX_JOB_SIZE, fProcess);
// LOG(info) << "PMIX_JOB_NUM_APPS: " << pmix::getInfo(PMIX_JOB_NUM_APPS, fProcess);
// LOG(info) << "PMIX_APP_SIZE: " << pmix::getInfo(PMIX_APP_SIZE, fProcess);
// LOG(info) << "PMIX_MAX_PROCS: " << pmix::getInfo(PMIX_MAX_PROCS, fProcess);
// LOG(info) << "PMIX_NUM_NODES: " << pmix::getInfo(PMIX_NUM_NODES, fProcess);
// LOG(info) << "PMIX_CLUSTER_ID: " << pmix::getInfo(PMIX_CLUSTER_ID, fProcess);
// LOG(info) << "PMIX_NSPACE: " << pmix::getInfo(PMIX_NSPACE, fProcess);
// LOG(info) << "PMIX_JOBID: " << pmix::getInfo(PMIX_JOBID, fProcess);
// LOG(info) << "PMIX_NODE_LIST: " << pmix::getInfo(PMIX_NODE_LIST, fProcess);
// LOG(info) << "PMIX_ALLOCATED_NODELIST: " << pmix::getInfo(PMIX_ALLOCATED_NODELIST, fProcess);
// LOG(info) << "PMIX_NPROC_OFFSET: " << pmix::getInfo(PMIX_NPROC_OFFSET, fProcess);
// LOG(info) << "PMIX_LOCALLDR: " << pmix::getInfo(PMIX_LOCALLDR, fProcess);
// LOG(info) << "PMIX_APPLDR: " << pmix::getInfo(PMIX_APPLDR, fProcess);
// // per-node information
// LOG(info) << "PMIX_NODE_SIZE: " << pmix::getInfo(PMIX_NODE_SIZE, fProcess);
// LOG(info) << "PMIX_LOCAL_SIZE: " << pmix::getInfo(PMIX_LOCAL_SIZE, fProcess);
// LOG(info) << "PMIX_AVAIL_PHYS_MEMORY: " << pmix::getInfo(PMIX_AVAIL_PHYS_MEMORY, fProcess);
// // per-process information
// LOG(info) << "PMIX_PROCID: " << pmix::getInfo(PMIX_PROCID, fProcess);
// LOG(info) << "PMIX_APPNUM: " << pmix::getInfo(PMIX_APPNUM, fProcess);
// LOG(info) << "PMIX_LOCAL_RANK: " << pmix::getInfo(PMIX_LOCAL_RANK, fProcess);
// LOG(info) << "PMIX_NODE_RANK: " << pmix::getInfo(PMIX_NODE_RANK, fProcess);
// LOG(info) << "PMIX_RANK: " << pmix::getInfo(PMIX_RANK, fProcess);
// LOG(info) << "PMIX_GLOBAL_RANK: " << pmix::getInfo(PMIX_GLOBAL_RANK, fProcess);
// LOG(info) << "PMIX_APP_RANK: " << pmix::getInfo(PMIX_APP_RANK, fProcess);
SubscribeToDeviceStateChange([this](DeviceState newState) {
switch (newState) {
case DeviceState::Idle:
Fence();
break;
case DeviceState::Bound:
Publish();
Fence();
break;
case DeviceState::Connecting:
Lookup();
break;
case DeviceState::DeviceReady:
Fence();
break;
case DeviceState::Ready:
Fence();
break;
case DeviceState::Exiting:
UnsubscribeFromDeviceStateChange();
break;
default:
break;
case DeviceState::Bound:
Publish();
break;
case DeviceState::Connecting:
Lookup();
break;
case DeviceState::Exiting:
ReleaseDeviceControl();
UnsubscribeFromDeviceStateChange();
break;
default:
break;
}
lock_guard<mutex> lock{fStateChangeSubscriberMutex};
fLastState = fCurrentState;
fCurrentState = newState;
for (auto subscriberId : fStateChangeSubscribers) {
LOG(debug) << "Publishing state-change: " << fLastState << "->" << newState << " to " << subscriberId;
Cmds cmds(make<StateChange>(fDeviceId, 0, fLastState, fCurrentState));
fCommands.Send(cmds.Serialize(Format::JSON), static_cast<pmix::rank>(subscriberId));
}
});
}
PMIxPlugin::~PMIxPlugin()
{
LOG(debug) << "Destroying PMIxPlugin";
ReleaseDeviceControl();
fCommands.Unsubscribe();
while (pmix::initialized()) {
try {
pmix::finalize();
@@ -69,33 +128,112 @@ PMIxPlugin::~PMIxPlugin()
}
}
auto PMIxPlugin::PMIxClient() const -> std::string
auto PMIxPlugin::SubscribeForCommands() -> void
{
std::stringstream ss;
ss << "PMIx client(pid=" << fPid << ") ";
return ss.str();
fCommands.Subscribe([this](const string& cmdStr, const pmix::proc& sender) {
// LOG(info) << "PMIx Plugin received message: '" << cmdStr << "', from " << sender;
Cmds inCmds;
inCmds.Deserialize(cmdStr, Format::JSON);
for (const auto& cmd : inCmds) {
LOG(info) << "Received command type: '" << cmd->GetType() << "' from " << sender;
switch (cmd->GetType()) {
case Type::check_state:
fCommands.Send(Cmds(make<CurrentState>(fDeviceId, GetCurrentDeviceState()))
.Serialize(Format::JSON),
{sender});
break;
case Type::change_state: {
Transition transition = static_cast<ChangeState&>(*cmd).GetTransition();
if (ChangeDeviceState(transition)) {
fCommands.Send(
Cmds(make<TransitionStatus>(fDeviceId, Result::Ok, transition))
.Serialize(Format::JSON),
{sender});
} else {
fCommands.Send(
Cmds(make<TransitionStatus>(fDeviceId, Result::Failure, transition))
.Serialize(Format::JSON),
{sender});
}
{
lock_guard<mutex> lock{fStateChangeSubscriberMutex};
fLastExternalController = sender.rank;
}
}
break;
case Type::subscribe_to_state_change: {
{
lock_guard<mutex> lock{fStateChangeSubscriberMutex};
fStateChangeSubscribers.insert(sender.rank);
}
LOG(debug) << "Publishing state-change: " << fLastState << "->" << fCurrentState
<< " to " << sender;
Cmds outCmds(make<StateChangeSubscription>(fDeviceId, Result::Ok),
make<StateChange>(fDeviceId, 0, fLastState, fCurrentState));
fCommands.Send(outCmds.Serialize(Format::JSON), {sender});
}
break;
case Type::unsubscribe_from_state_change: {
{
lock_guard<mutex> lock{fStateChangeSubscriberMutex};
fStateChangeSubscribers.erase(sender.rank);
}
fCommands.Send(Cmds(make<StateChangeUnsubscription>(fDeviceId, Result::Ok))
.Serialize(Format::JSON),
{sender});
}
break;
case Type::state_change_exiting_received: {
{
lock_guard<mutex> lock{fStateChangeSubscriberMutex};
if (fLastExternalController == sender.rank) {
fExitingAckedByLastExternalController = true;
}
}
fExitingAcked.notify_one();
}
break;
case Type::dump_config: {
stringstream ss;
for (const auto& k: GetPropertyKeys()) {
ss << fDeviceId << ": " << k << " -> " << GetPropertyAsString(k) << "\n";
}
fCommands.Send(Cmds(make<Config>(fDeviceId, ss.str())).Serialize(Format::JSON),
{sender});
}
break;
default:
LOG(warn) << "Unexpected/unknown command received: " << cmdStr;
LOG(warn) << "Origin: " << sender;
break;
}
}
});
}
auto PMIxPlugin::Init() -> void
auto PMIxPlugin::Init() -> pmix::proc
{
if (!pmix::initialized()) {
fProc = pmix::init();
LOG(debug) << PMIxClient() << "pmix::init() OK: " << fProc
<< ",version=" << pmix::get_version();
return pmix::init();
} else {
throw runtime_error("trying to initialize PMIx while it is already initialized");
}
}
auto PMIxPlugin::Publish() -> void
{
auto channels(GetChannelInfo());
std::vector<pmix::info> info;
vector<pmix::info> info;
for (const auto& c : channels) {
std::string methodKey{"chans." + c.first + "." + std::to_string(c.second - 1) + ".method"};
if (GetProperty<std::string>(methodKey) == "bind") {
string methodKey("chans." + c.first + "." + to_string(c.second - 1) + ".method");
if (GetProperty<string>(methodKey) == "bind") {
for (int i = 0; i < c.second; ++i) {
std::string addressKey{"chans." + c.first + "." + std::to_string(i) + ".address"};
info.emplace_back(addressKey, GetProperty<std::string>(addressKey));
string addressKey("chans." + c.first + "." + to_string(i) + ".address");
info.emplace_back(addressKey, GetProperty<string>(addressKey));
LOG(debug) << PMIxClient() << info.back();
}
}
@@ -103,32 +241,37 @@ auto PMIxPlugin::Publish() -> void
if (info.size() > 0) {
pmix::publish(info);
LOG(debug) << PMIxClient() << "pmix::publish() OK: published "
<< info.size() << " binding channels.";
LOG(debug) << PMIxClient() << "pmix::publish() OK: published " << info.size()
<< " binding channels.";
}
}
auto PMIxPlugin::Fence() -> void
{
pmix::proc all(fProc);
pmix::proc all(fProcess);
all.rank = pmix::rank::wildcard;
pmix::fence({all});
LOG(debug) << PMIxClient() << "pmix::fence() OK";
}
auto PMIxPlugin::Fence(const std::string& label) -> void
{
Fence(label);
LOG(debug) << PMIxClient() << "pmix::fence() [" << label << "] OK";
}
auto PMIxPlugin::Lookup() -> void
{
auto channels(GetChannelInfo());
for (const auto& c : channels) {
std::string methodKey{"chans." + c.first + "." + std::to_string(c.second - 1) + ".method"};
if (GetProperty<std::string>(methodKey) == "connect") {
string methodKey("chans." + c.first + "." + to_string(c.second - 1) + ".method");
if (GetProperty<string>(methodKey) == "connect") {
for (int i = 0; i < c.second; ++i) {
std::vector<pmix::pdata> pdata;
std::string addressKey{"chans." + c.first + "." + std::to_string(i) + ".address"};
vector<pmix::pdata> pdata;
string addressKey("chans." + c.first + "." + to_string(i) + ".address");
pdata.emplace_back();
pdata.back().set_key(addressKey);
std::vector<pmix::info> info;
vector<pmix::info> info;
info.emplace_back(PMIX_WAIT, static_cast<int>(pdata.size()));
if (pdata.size() > 0) {
@@ -141,11 +284,11 @@ auto PMIxPlugin::Lookup() -> void
LOG(debug) << PMIxClient() << "pmix::lookup() not found: key=" << p.key;
} else if (p.value.type == PMIX_STRING) {
LOG(debug) << PMIxClient() << "pmix::lookup() found:"
<< " key=" << p.key << ",value=" << p.value.data.string;
SetProperty<std::string>(p.key, p.value.data.string);
<< " key=" << p.key << ",value=" << p.value.data.string;
SetProperty<string>(p.key, p.value.data.string);
} else {
LOG(debug) << PMIxClient() << "pmix::lookup() wrong type returned: "
<< "key=" << p.key << ",type=" << p.value.type;
<< "key=" << p.key << ",type=" << p.value.type;
}
}
}
@@ -153,6 +296,14 @@ auto PMIxPlugin::Lookup() -> void
}
}
auto PMIxPlugin::WaitForExitingAck() -> void
{
unique_lock<mutex> lock(fStateChangeSubscriberMutex);
fExitingAcked.wait_for(lock, chrono::milliseconds(1000), [this]() {
return fExitingAckedByLastExternalController;
});
}
} /* namespace plugins */
} /* namespace mq */
} /* namespace fair */

View File

@@ -10,6 +10,7 @@
#define FAIR_MQ_PLUGINS_PMIX
#include "PMIx.hpp"
#include "PMIxCommands.h"
#include <fairmq/Plugin.h>
#include <fairmq/Version.h>
@@ -39,23 +40,40 @@ class PMIxPlugin : public Plugin
const std::string& homepage,
PluginServices* pluginServices);
~PMIxPlugin();
auto PMIxClient() const -> std::string;
auto PMIxClient() const -> std::string { return fPMIxClient; };
private:
pmix::proc fProc;
pmix::proc fProcess;
pid_t fPid;
std::string fPMIxClient;
std::string fDeviceId;
pmix::Commands fCommands;
auto Init() -> void;
std::set<uint32_t> fStateChangeSubscribers;
uint32_t fLastExternalController;
bool fExitingAckedByLastExternalController;
std::condition_variable fExitingAcked;
std::mutex fStateChangeSubscriberMutex;
DeviceState fCurrentState;
DeviceState fLastState;
auto Init() -> pmix::proc;
auto Publish() -> void;
auto Fence() -> void;
auto Fence(const std::string& label) -> void;
auto Lookup() -> void;
auto SubscribeForCommands() -> void;
auto WaitForExitingAck() -> void;
};
Plugin::ProgOptions PMIxProgramOptions()
{
boost::program_options::options_description options{"PMIx Plugin"};
options.add_options()(
"pmix-dummy", boost::program_options::value<int>()->default_value(0), "Dummy.");
boost::program_options::options_description options("PMIx Plugin");
options.add_options()
("pmix-dummy", boost::program_options::value<int>()->default_value(0), "Dummy.");
return options;
}

View File

@@ -0,0 +1,218 @@
/********************************************************************************
* Copyright (C) 2014-2019 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 <fairmq/sdk/commands/Commands.h>
#include <fairmq/States.h>
#include <fairlogger/Logger.h>
#include "PMIx.hpp"
#include "PMIxCommands.h"
#include <boost/program_options.hpp>
#include <condition_variable>
#include <algorithm>
#include <cstdlib>
#include <iostream>
#include <mutex>
#include <string>
#include <termios.h> // raw mode console input
#include <thread>
#include <utility>
#include <unistd.h>
#include <vector>
using namespace std;
using namespace fair::mq::sdk::cmd;
namespace bpo = boost::program_options;
const std::map<fair::mq::Transition, fair::mq::State> expected =
{
{ fair::mq::Transition::InitDevice, fair::mq::State::InitializingDevice },
{ fair::mq::Transition::CompleteInit, fair::mq::State::Initialized },
{ fair::mq::Transition::Bind, fair::mq::State::Bound },
{ fair::mq::Transition::Connect, fair::mq::State::DeviceReady },
{ fair::mq::Transition::InitTask, fair::mq::State::Ready },
{ fair::mq::Transition::Run, fair::mq::State::Running },
{ fair::mq::Transition::Stop, fair::mq::State::Ready },
{ fair::mq::Transition::ResetTask, fair::mq::State::DeviceReady },
{ fair::mq::Transition::ResetDevice, fair::mq::State::Idle },
{ fair::mq::Transition::End, fair::mq::State::Exiting }
};
struct StateSubscription
{
pmix::Commands& fCommands;
explicit StateSubscription(pmix::Commands& commands)
: fCommands(commands)
{
fCommands.Send(Cmds(make<SubscribeToStateChange>()).Serialize(Format::JSON));
}
~StateSubscription()
{
fCommands.Send(Cmds(make<UnsubscribeFromStateChange>()).Serialize(Format::JSON));
this_thread::sleep_for(chrono::milliseconds(100)); // give PMIx a chance to complete request
}
};
struct MiniTopo
{
explicit MiniTopo(unsigned int n)
: fState(n, fair::mq::State::Ok)
{}
void WaitFor(const fair::mq::State state)
{
std::unique_lock<std::mutex> lk(fMtx);
fCV.wait(lk, [&](){
unsigned int count = std::count_if(fState.cbegin(), fState.cend(), [=](const auto& s) {
return s == state;
});
bool result = count == fState.size();
cout << "expecting " << state << " for " << fState.size() << " devices, found " << count << ", condition: " << result << endl;
return result;
});
}
void Update(uint32_t rank, const fair::mq::State state)
{
try {
{
std::lock_guard<std::mutex> lk(fMtx);
fState.at(rank - 1) = state;
}
fCV.notify_one();
} catch (const std::exception& e) {
LOG(error) << "Exception in Update: " << e.what();
}
}
private:
vector<fair::mq::State> fState;
std::mutex fMtx;
std::condition_variable fCV;
};
int main(int argc, char* argv[])
{
try {
unsigned int numDevices;
bpo::options_description options("Common options");
options.add_options()
("number-devices,n", bpo::value<unsigned int>(&numDevices)->default_value(0), "Number of devices (will be removed in the future)")
("help,h", "Produce help message");
bpo::variables_map vm;
bpo::store(bpo::command_line_parser(argc, argv).options(options).run(), vm);
if (vm.count("help")) {
cout << "FairMQ DDS Command UI" << endl << options << endl;
cout << "Commands: [c] check state, [o] dump config, [h] help, [r] run, [s] stop, [t] reset task, [d] reset device, [q] end, [j] init task, [i] init device, [k] complete init, [b] bind, [x] connect" << endl;
return EXIT_SUCCESS;
}
bpo::notify(vm);
fair::Logger::SetConsoleSeverity(fair::Severity::debug);
fair::Logger::SetConsoleColor(true);
fair::Logger::SetVerbosity(fair::Verbosity::low);
pmix::proc process;
if (!pmix::initialized()) {
process = pmix::init();
LOG(warn) << "pmix::init() OK: " << process << ", version=" << pmix::get_version();
}
pmix::proc all(process);
all.rank = pmix::rank::wildcard;
pmix::fence({all});
LOG(warn) << "pmix::fence() [pmix::init] OK";
MiniTopo topo(numDevices);
pmix::Commands commands(process);
commands.Subscribe([&](const string& msg, const pmix::proc& sender) {
// LOG(info) << "Received '" << msg << "' from " << sender;
Cmds cmds;
cmds.Deserialize(msg, Format::JSON);
// cout << "Received " << cmds.Size() << " command(s) with total size of " << msg.length() << " bytes: " << endl;
for (const auto& cmd : cmds) {
// cout << " > " << cmd->GetType() << endl;
switch (cmd->GetType()) {
case Type::state_change: {
cout << "Received state_change from " << static_cast<StateChange&>(*cmd).GetDeviceId() << ": " << static_cast<StateChange&>(*cmd).GetLastState() << "->" << static_cast<StateChange&>(*cmd).GetCurrentState() << endl;
topo.Update(sender.rank, static_cast<StateChange&>(*cmd).GetCurrentState());
if (static_cast<StateChange&>(*cmd).GetCurrentState() == fair::mq::State::Exiting) {
commands.Send(Cmds(make<StateChangeExitingReceived>()).Serialize(Format::JSON), {sender});
}
}
break;
case Type::state_change_subscription:
if (static_cast<StateChangeSubscription&>(*cmd).GetResult() != Result::Ok) {
cout << "State change subscription failed for " << static_cast<StateChangeSubscription&>(*cmd).GetDeviceId() << endl;
}
break;
case Type::state_change_unsubscription:
if (static_cast<StateChangeUnsubscription&>(*cmd).GetResult() != Result::Ok) {
cout << "State change unsubscription failed for " << static_cast<StateChangeUnsubscription&>(*cmd).GetDeviceId() << endl;
}
break;
case Type::transition_status: {
if (static_cast<TransitionStatus&>(*cmd).GetResult() == Result::Ok) {
cout << "Device " << static_cast<TransitionStatus&>(*cmd).GetDeviceId() << " started to transition with " << static_cast<TransitionStatus&>(*cmd).GetTransition() << endl;
} else {
cout << "Device " << static_cast<TransitionStatus&>(*cmd).GetDeviceId() << " cannot transition with " << static_cast<TransitionStatus&>(*cmd).GetTransition() << endl;
}
}
break;
case Type::current_state:
cout << "Device " << static_cast<CurrentState&>(*cmd).GetDeviceId() << " is in " << static_cast<CurrentState&>(*cmd).GetCurrentState() << " state" << endl;
break;
case Type::config:
cout << "Received config for device " << static_cast<Config&>(*cmd).GetDeviceId() << ":\n" << static_cast<Config&>(*cmd).GetConfig() << endl;
break;
default:
cout << "Unexpected/unknown command received: " << cmd->GetType() << endl;
cout << "Origin: " << sender << endl;
break;
}
}
});
pmix::fence({all});
LOG(warn) << "pmix::fence() [subscribed] OK";
StateSubscription stateSubscription(commands);
for (auto transition : { fair::mq::Transition::InitDevice,
fair::mq::Transition::CompleteInit,
fair::mq::Transition::Bind,
fair::mq::Transition::Connect,
fair::mq::Transition::InitTask,
fair::mq::Transition::Run,
fair::mq::Transition::Stop,
fair::mq::Transition::ResetTask,
fair::mq::Transition::ResetDevice,
fair::mq::Transition::End }) {
commands.Send(Cmds(make<ChangeState>(transition)).Serialize(Format::JSON));
topo.WaitFor(expected.at(transition));
}
} catch (exception& e) {
LOG(error) << "Error: " << e.what();
return EXIT_FAILURE;
}
LOG(warn) << "exiting";
return EXIT_SUCCESS;
}

View File

@@ -6,6 +6,8 @@ maxIterations="0"
msgSize="1000000"
transport="zeromq"
affinity="false"
multipart="false"
numParts="1"
affinitySamp=""
affinitySink=""
@@ -26,6 +28,14 @@ if [[ $4 =~ ^[a-z]+$ ]]; then
affinity=$4
fi
if [[ $5 =~ ^[a-z]+$ ]]; then
multipart=$5
fi
if [[ $6 =~ ^[0-9]+$ ]]; then
numParts=$6
fi
echo "Starting benchmark with following settings:"
@@ -58,7 +68,8 @@ SAMPLER+=" --id bsampler1"
SAMPLER+=" --transport $transport"
SAMPLER+=" --severity debug"
SAMPLER+=" --msg-size $msgSize"
SAMPLER+=" --num-parts 1"
SAMPLER+=" --multipart $multipart"
SAMPLER+=" --num-parts $numParts"
# SAMPLER+=" --msg-rate 1000"
SAMPLER+=" --max-iterations $maxIterations"
SAMPLER+=" --channel-config name=data,type=pair,method=bind,address=tcp://127.0.0.1:5555"
@@ -73,7 +84,7 @@ SINK+=" --id sink1"
#SINK+=" --control static"
SINK+=" --transport $transport"
SINK+=" --severity debug"
SINK+=" --multipart false"
SINK+=" --multipart $multipart"
SINK+=" --max-iterations $maxIterations"
SINK+=" --channel-config name=data,type=pair,method=connect,address=tcp://127.0.0.1:5555"
xterm -geometry 90x50+550+0 -hold -e $affinitySink @CMAKE_CURRENT_BINARY_DIR@/$SINK &

View File

@@ -77,8 +77,8 @@ struct MetaHeader
{
size_t fSize;
size_t fRegionId;
boost::interprocess::managed_shared_memory::handle_t fHandle;
size_t fHint;
boost::interprocess::managed_shared_memory::handle_t fHandle;
};
struct RegionBlock

View File

@@ -1,281 +0,0 @@
/********************************************************************************
* Copyright (C) 2014 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 "Common.h"
#include "Region.h"
#include "FairMQMessageSHM.h"
#include "FairMQUnmanagedRegionSHM.h"
#include "FairMQLogger.h"
#include <boost/date_time/posix_time/posix_time.hpp>
#include <cstdlib>
using namespace std;
using namespace fair::mq::shmem;
namespace bipc = ::boost::interprocess;
namespace bpt = ::boost::posix_time;
atomic<bool> FairMQMessageSHM::fInterrupted(false);
fair::mq::Transport FairMQMessageSHM::fTransportType = fair::mq::Transport::SHM;
FairMQMessageSHM::FairMQMessageSHM(Manager& manager, FairMQTransportFactory* factory)
: FairMQMessage{factory}
, fManager(manager)
, fMessage()
, fQueued(false)
, fMetaCreated(false)
, fRegionId(0)
, fRegionPtr(nullptr)
, fHandle(-1)
, fSize(0)
, fHint(0)
, fLocalPtr(nullptr)
{
if (zmq_msg_init(&fMessage) != 0) {
LOG(error) << "failed initializing message, reason: " << zmq_strerror(errno);
}
fMetaCreated = true;
}
FairMQMessageSHM::FairMQMessageSHM(Manager& manager, const size_t size, FairMQTransportFactory* factory)
: FairMQMessage{factory}
, fManager(manager)
, fMessage()
, fQueued(false)
, fMetaCreated(false)
, fRegionId(0)
, fRegionPtr(nullptr)
, fHandle(-1)
, fSize(0)
, fHint(0)
, fLocalPtr(nullptr)
{
InitializeChunk(size);
}
FairMQMessageSHM::FairMQMessageSHM(Manager& manager, void* data, const size_t size, fairmq_free_fn* ffn, void* hint, FairMQTransportFactory* factory)
: FairMQMessage{factory}
, fManager(manager)
, fMessage()
, fQueued(false)
, fMetaCreated(false)
, fRegionId(0)
, fRegionPtr(nullptr)
, fHandle(-1)
, fSize(0)
, fHint(0)
, fLocalPtr(nullptr)
{
if (InitializeChunk(size)) {
memcpy(fLocalPtr, data, size);
if (ffn) {
ffn(data, hint);
} else {
free(data);
}
}
}
FairMQMessageSHM::FairMQMessageSHM(Manager& manager, FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint, FairMQTransportFactory* factory)
: FairMQMessage{factory}
, fManager(manager)
, fMessage()
, fQueued(false)
, fMetaCreated(false)
, fRegionId(static_cast<FairMQUnmanagedRegionSHM*>(region.get())->fRegionId)
, fRegionPtr(nullptr)
, fHandle(-1)
, fSize(size)
, fHint(reinterpret_cast<size_t>(hint))
, fLocalPtr(static_cast<char*>(data))
{
if (reinterpret_cast<const char*>(data) >= reinterpret_cast<const char*>(region->GetData()) ||
reinterpret_cast<const char*>(data) <= reinterpret_cast<const char*>(region->GetData()) + region->GetSize()) {
fHandle = (bipc::managed_shared_memory::handle_t)(reinterpret_cast<const char*>(data) - reinterpret_cast<const char*>(region->GetData()));
if (zmq_msg_init_size(&fMessage, sizeof(MetaHeader)) != 0) {
LOG(error) << "failed initializing meta message, reason: " << zmq_strerror(errno);
} else {
MetaHeader header;
header.fSize = size;
header.fHandle = fHandle;
header.fRegionId = fRegionId;
header.fHint = fHint;
memcpy(zmq_msg_data(&fMessage), &header, sizeof(MetaHeader));
fMetaCreated = true;
}
} else {
LOG(error) << "trying to create region message with data from outside the region";
throw runtime_error("trying to create region message with data from outside the region");
}
}
bool FairMQMessageSHM::InitializeChunk(const size_t size)
{
while (fHandle < 0) {
try {
bipc::managed_shared_memory::size_type actualSize = size;
char* hint = 0; // unused for bipc::allocate_new
fLocalPtr = fManager.Segment().allocation_command<char>(bipc::allocate_new, size, actualSize, hint);
} catch (bipc::bad_alloc& ba) {
// LOG(warn) << "Shared memory full...";
this_thread::sleep_for(chrono::milliseconds(50));
if (fInterrupted) {
return false;
} else {
continue;
}
}
fHandle = fManager.Segment().get_handle_from_address(fLocalPtr);
}
fSize = size;
if (zmq_msg_init_size(&fMessage, sizeof(MetaHeader)) != 0) {
LOG(error) << "failed initializing meta message, reason: " << zmq_strerror(errno);
return false;
}
MetaHeader header;
header.fSize = size;
header.fHandle = fHandle;
header.fRegionId = fRegionId;
header.fHint = fHint;
memcpy(zmq_msg_data(&fMessage), &header, sizeof(MetaHeader));
fMetaCreated = true;
return true;
}
void FairMQMessageSHM::Rebuild()
{
CloseMessage();
fQueued = false;
if (zmq_msg_init(&fMessage) != 0) {
LOG(error) << "failed initializing message, reason: " << zmq_strerror(errno);
}
fMetaCreated = true;
}
void FairMQMessageSHM::Rebuild(const size_t size)
{
CloseMessage();
fQueued = false;
InitializeChunk(size);
}
void FairMQMessageSHM::Rebuild(void* data, const size_t size, fairmq_free_fn* ffn, void* hint)
{
CloseMessage();
fQueued = false;
if (InitializeChunk(size)) {
memcpy(fLocalPtr, data, size);
if (ffn) {
ffn(data, hint);
} else {
free(data);
}
}
}
void* FairMQMessageSHM::GetData() const
{
if (!fLocalPtr) {
if (fRegionId == 0) {
if (fSize > 0) {
fLocalPtr = reinterpret_cast<char*>(fManager.Segment().get_address_from_handle(fHandle));
} else {
fLocalPtr = nullptr;
}
} else {
fRegionPtr = fManager.GetRemoteRegion(fRegionId);
if (fRegionPtr) {
fLocalPtr = reinterpret_cast<char*>(fRegionPtr->fRegion.get_address()) + fHandle;
} else {
// LOG(warn) << "could not get pointer from a region message";
fLocalPtr = nullptr;
}
}
}
return fLocalPtr;
}
bool FairMQMessageSHM::SetUsedSize(const size_t size)
{
if (size == fSize) {
return true;
} else if (size <= fSize) {
try {
bipc::managed_shared_memory::size_type shrunkSize = size;
fLocalPtr = fManager.Segment().allocation_command<char>(bipc::shrink_in_place, fSize + 128, shrunkSize, fLocalPtr);
fSize = size;
// update meta header
MetaHeader* hdrPtr = static_cast<MetaHeader*>(zmq_msg_data(&fMessage));
hdrPtr->fSize = fSize;
return true;
} catch (bipc::interprocess_exception& e) {
LOG(info) << "could not set used size: " << e.what();
return false;
}
} else {
LOG(error) << "cannot set used size higher than original.";
return false;
}
}
void FairMQMessageSHM::Copy(const FairMQMessage& msg)
{
if (fHandle < 0) {
bipc::managed_shared_memory::handle_t otherHandle = static_cast<const FairMQMessageSHM&>(msg).fHandle;
if (otherHandle) {
if (InitializeChunk(msg.GetSize())) {
memcpy(GetData(), msg.GetData(), msg.GetSize());
}
} else {
LOG(error) << "copy fail: source message not initialized!";
}
} else {
LOG(error) << "copy fail: target message already initialized!";
}
}
void FairMQMessageSHM::CloseMessage()
{
if (fHandle >= 0 && !fQueued) {
if (fRegionId == 0) {
fManager.Segment().deallocate(fManager.Segment().get_address_from_handle(fHandle));
fHandle = -1;
} else {
if (!fRegionPtr) {
fRegionPtr = fManager.GetRemoteRegion(fRegionId);
}
if (fRegionPtr) {
fRegionPtr->ReleaseBlock({fHandle, fSize, fHint});
} else {
LOG(warn) << "region ack queue for id " << fRegionId << " no longer exist. Not sending ack";
}
}
}
if (fMetaCreated) {
if (zmq_msg_close(&fMessage) != 0) {
LOG(error) << "failed closing message, reason: " << zmq_strerror(errno);
}
fMetaCreated = false;
}
}

View File

@@ -1,72 +0,0 @@
/********************************************************************************
* Copyright (C) 2014 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" *
********************************************************************************/
#ifndef FAIRMQMESSAGESHM_H_
#define FAIRMQMESSAGESHM_H_
#include <fairmq/shmem/Manager.h>
#include "FairMQMessage.h"
#include "FairMQUnmanagedRegion.h"
#include <zmq.h>
#include <boost/interprocess/mapped_region.hpp>
#include <cstddef> // size_t
#include <atomic>
class FairMQSocketSHM;
class FairMQMessageSHM final : public FairMQMessage
{
friend class FairMQSocketSHM;
public:
FairMQMessageSHM(fair::mq::shmem::Manager& manager, FairMQTransportFactory* factory = nullptr);
FairMQMessageSHM(fair::mq::shmem::Manager& manager, const size_t size, FairMQTransportFactory* factory = nullptr);
FairMQMessageSHM(fair::mq::shmem::Manager& manager, void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr, FairMQTransportFactory* factory = nullptr);
FairMQMessageSHM(fair::mq::shmem::Manager& manager, FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint = 0, FairMQTransportFactory* factory = nullptr);
FairMQMessageSHM(const FairMQMessageSHM&) = delete;
FairMQMessageSHM operator=(const FairMQMessageSHM&) = delete;
void Rebuild() override;
void Rebuild(const size_t size) override;
void Rebuild(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr) override;
void* GetData() const override;
size_t GetSize() const override { return fSize; }
bool SetUsedSize(const size_t size) override;
fair::mq::Transport GetType() const override { return fTransportType; }
void Copy(const FairMQMessage& msg) override;
~FairMQMessageSHM() override { CloseMessage(); }
private:
fair::mq::shmem::Manager& fManager;
zmq_msg_t fMessage;
bool fQueued;
bool fMetaCreated;
static std::atomic<bool> fInterrupted;
static fair::mq::Transport fTransportType;
size_t fRegionId;
mutable fair::mq::shmem::Region* fRegionPtr;
boost::interprocess::managed_shared_memory::handle_t fHandle;
size_t fSize;
size_t fHint;
mutable char* fLocalPtr;
bool InitializeChunk(const size_t size);
zmq_msg_t* GetMessage() { return &fMessage; }
void CloseMessage();
};
#endif /* FAIRMQMESSAGESHM_H_ */

View File

@@ -1,67 +0,0 @@
/********************************************************************************
* Copyright (C) 2016-2017 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" *
********************************************************************************/
#ifndef FAIRMQTRANSPORTFACTORYSHM_H_
#define FAIRMQTRANSPORTFACTORYSHM_H_
#include <fairmq/shmem/Manager.h>
#include <fairmq/shmem/Common.h>
#include "FairMQTransportFactory.h"
#include "FairMQMessageSHM.h"
#include "FairMQSocketSHM.h"
#include "FairMQPollerSHM.h"
#include "FairMQUnmanagedRegionSHM.h"
#include <fairmq/ProgOptions.h>
#include <vector>
#include <string>
#include <thread>
#include <atomic>
class FairMQTransportFactorySHM final : public FairMQTransportFactory
{
public:
FairMQTransportFactorySHM(const std::string& id = "", const fair::mq::ProgOptions* config = nullptr);
FairMQTransportFactorySHM(const FairMQTransportFactorySHM&) = delete;
FairMQTransportFactorySHM operator=(const FairMQTransportFactorySHM&) = delete;
FairMQMessagePtr CreateMessage() override;
FairMQMessagePtr CreateMessage(const size_t size) override;
FairMQMessagePtr CreateMessage(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr) override;
FairMQMessagePtr CreateMessage(FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint = 0) override;
FairMQSocketPtr CreateSocket(const std::string& type, const std::string& name) override;
FairMQPollerPtr CreatePoller(const std::vector<FairMQChannel>& channels) const override;
FairMQPollerPtr CreatePoller(const std::vector<FairMQChannel*>& channels) const override;
FairMQPollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList) const override;
FairMQUnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, FairMQRegionCallback callback = nullptr, const std::string& path = "", int flags = 0) const override;
fair::mq::Transport GetType() const override;
void Interrupt() override { FairMQSocketSHM::Interrupt(); }
void Resume() override { FairMQSocketSHM::Resume(); }
void Reset() override {}
~FairMQTransportFactorySHM() override;
private:
void SendHeartbeats();
static fair::mq::Transport fTransportType;
std::string fDeviceId;
std::string fShmId;
void* fZMQContext;
std::unique_ptr<fair::mq::shmem::Manager> fManager;
std::thread fHeartbeatThread;
std::atomic<bool> fSendHeartbeats;
};
#endif /* FAIRMQTRANSPORTFACTORYSHM_H_ */

View File

@@ -6,8 +6,9 @@
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include <fairmq/shmem/Manager.h>
#include <fairmq/shmem/Common.h>
#include "Manager.h"
#include "Common.h"
#include <fairmq/tools/CppSTL.h>
#include <fairmq/tools/Strings.h>
@@ -83,7 +84,7 @@ void Manager::StartMonitor(const std::string& id)
this_thread::sleep_for(chrono::milliseconds(10));
if (++numTries > 1000) {
LOG(error) << "Did not get response from fairmq-shmmonitor after " << 10 * 1000 << " milliseconds. Exiting.";
throw runtime_error(fair::mq::tools::ToString("Did not get response from fairmq-shmmonitor after ", 10 * 1000, " milliseconds. Exiting."));
throw runtime_error(tools::ToString("Did not get response from fairmq-shmmonitor after ", 10 * 1000, " milliseconds. Exiting."));
}
}
} while (true);
@@ -109,7 +110,7 @@ void Manager::Resume()
}
}
bipc::mapped_region* Manager::CreateRegion(const size_t size, const uint64_t id, FairMQRegionCallback callback, const std::string& path /* = "" */, int flags /* = 0 */)
bipc::mapped_region* Manager::CreateRegion(const size_t size, const uint64_t id, RegionCallback callback, const std::string& path /* = "" */, int flags /* = 0 */)
{
auto it = fRegions.find(id);
if (it != fRegions.end()) {
@@ -125,7 +126,7 @@ bipc::mapped_region* Manager::CreateRegion(const size_t size, const uint64_t id,
}
// LOG(debug) << "Created region with id '" << id << "', path: '" << path << "', flags: '" << flags << "'";
auto r = fRegions.emplace(id, fair::mq::tools::make_unique<Region>(*this, id, size, false, callback, path, flags));
auto r = fRegions.emplace(id, tools::make_unique<Region>(*this, id, size, false, callback, path, flags));
r.first->second->StartReceivingAcks();
@@ -158,7 +159,7 @@ Region* Manager::GetRemoteRegion(const uint64_t id)
}
// LOG(debug) << "Located remote region with id '" << id << "', path: '" << path << "', flags: '" << flags << "'";
auto r = fRegions.emplace(id, fair::mq::tools::make_unique<Region>(*this, id, 0, true, nullptr, path, flags));
auto r = fRegions.emplace(id, tools::make_unique<Region>(*this, id, 0, true, nullptr, path, flags));
return r.first->second.get();
} catch (bie& e) {
LOG(warn) << "Could not get remote region for id: " << id;
@@ -192,7 +193,7 @@ Manager::~Manager()
{
bool lastRemoved = false;
{
try {
bipc::scoped_lock<bipc::named_mutex> lock(fShmMtx);
(fDeviceCounter->fCount)--;
@@ -205,6 +206,8 @@ Manager::~Manager()
} else {
LOG(debug) << "other segment users present (" << fDeviceCounter->fCount << "), not removing it.";
}
} catch(bie& e) {
LOG(error) << "error while acquiring lock in Manager destructor: " << e.what();
}
if (lastRemoved) {

View File

@@ -6,7 +6,7 @@
* copied verbatim in the file "LICENSE" *
********************************************************************************/
/**
* FairMQShmManager.h
* Manager.h
*
* @since 2016-04-08
* @author A. Rybalchenko
@@ -15,11 +15,10 @@
#ifndef FAIR_MQ_SHMEM_MANAGER_H_
#define FAIR_MQ_SHMEM_MANAGER_H_
#include <fairmq/shmem/Region.h>
#include <fairmq/shmem/Common.h>
#include "Region.h"
#include "Common.h"
#include "FairMQLogger.h"
#include "FairMQMessage.h"
#include <FairMQLogger.h>
#include <boost/interprocess/managed_shared_memory.hpp>
#include <boost/interprocess/ipc/message_queue.hpp>
@@ -64,7 +63,7 @@ class Manager
int IncrementDeviceCounter();
int DecrementDeviceCounter();
boost::interprocess::mapped_region* CreateRegion(const size_t size, const uint64_t id, FairMQRegionCallback callback, const std::string& path = "", int flags = 0);
boost::interprocess::mapped_region* CreateRegion(const size_t size, const uint64_t id, RegionCallback callback, const std::string& path = "", int flags = 0);
Region* GetRemoteRegion(const uint64_t id);
void RemoveRegion(const uint64_t id);
@@ -77,7 +76,7 @@ class Manager
boost::interprocess::managed_shared_memory fSegment;
boost::interprocess::managed_shared_memory fManagementSegment;
boost::interprocess::named_mutex fShmMtx;
fair::mq::shmem::DeviceCounter* fDeviceCounter;
DeviceCounter* fDeviceCounter;
static std::unordered_map<uint64_t, std::unique_ptr<Region>> fRegions;
};

251
fairmq/shmem/Message.cxx Normal file
View File

@@ -0,0 +1,251 @@
/********************************************************************************
* Copyright (C) 2014 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 "Region.h"
#include "Message.h"
#include "UnmanagedRegion.h"
#include "TransportFactory.h"
#include <FairMQLogger.h>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <cstdlib>
using namespace std;
namespace bipc = ::boost::interprocess;
namespace bpt = ::boost::posix_time;
namespace fair
{
namespace mq
{
namespace shmem
{
atomic<bool> Message::fInterrupted(false);
Transport Message::fTransportType = Transport::SHM;
Message::Message(Manager& manager, FairMQTransportFactory* factory)
: fair::mq::Message{factory}
, fManager(manager)
, fQueued(false)
, fMeta{0, 0, 0, -1}
, fRegionPtr(nullptr)
, fLocalPtr(nullptr)
{
static_cast<TransportFactory*>(GetTransport())->IncrementMsgCounter();
}
Message::Message(Manager& manager, const size_t size, FairMQTransportFactory* factory)
: fair::mq::Message{factory}
, fManager(manager)
, fQueued(false)
, fMeta{0, 0, 0, -1}
, fRegionPtr(nullptr)
, fLocalPtr(nullptr)
{
InitializeChunk(size);
static_cast<TransportFactory*>(GetTransport())->IncrementMsgCounter();
}
Message::Message(Manager& manager, MetaHeader& hdr, FairMQTransportFactory* factory)
: fair::mq::Message{factory}
, fManager(manager)
, fQueued(false)
, fMeta{hdr}
, fRegionPtr(nullptr)
, fLocalPtr(nullptr)
{
static_cast<TransportFactory*>(GetTransport())->IncrementMsgCounter();
}
Message::Message(Manager& manager, void* data, const size_t size, fairmq_free_fn* ffn, void* hint, FairMQTransportFactory* factory)
: fair::mq::Message{factory}
, fManager(manager)
, fQueued(false)
, fMeta{0, 0, 0, -1}
, fRegionPtr(nullptr)
, fLocalPtr(nullptr)
{
if (InitializeChunk(size)) {
std::memcpy(fLocalPtr, data, size);
if (ffn) {
ffn(data, hint);
} else {
free(data);
}
}
static_cast<TransportFactory*>(GetTransport())->IncrementMsgCounter();
}
Message::Message(Manager& manager, UnmanagedRegionPtr& region, void* data, const size_t size, void* hint, FairMQTransportFactory* factory)
: fair::mq::Message{factory}
, fManager(manager)
, fQueued(false)
, fMeta{size, static_cast<UnmanagedRegion*>(region.get())->fRegionId, reinterpret_cast<size_t>(hint), -1}
, fRegionPtr(nullptr)
, fLocalPtr(static_cast<char*>(data))
{
if (reinterpret_cast<const char*>(data) >= reinterpret_cast<const char*>(region->GetData()) ||
reinterpret_cast<const char*>(data) <= reinterpret_cast<const char*>(region->GetData()) + region->GetSize()) {
fMeta.fHandle = (bipc::managed_shared_memory::handle_t)(reinterpret_cast<const char*>(data) - reinterpret_cast<const char*>(region->GetData()));
} else {
LOG(error) << "trying to create region message with data from outside the region";
throw runtime_error("trying to create region message with data from outside the region");
}
static_cast<TransportFactory*>(GetTransport())->IncrementMsgCounter();
}
bool Message::InitializeChunk(const size_t size)
{
while (fMeta.fHandle < 0) {
try {
bipc::managed_shared_memory::size_type actualSize = size;
char* hint = 0; // unused for bipc::allocate_new
fLocalPtr = fManager.Segment().allocation_command<char>(bipc::allocate_new, size, actualSize, hint);
} catch (bipc::bad_alloc& ba) {
// LOG(warn) << "Shared memory full...";
this_thread::sleep_for(chrono::milliseconds(50));
if (fInterrupted) {
return false;
} else {
continue;
}
}
fMeta.fHandle = fManager.Segment().get_handle_from_address(fLocalPtr);
}
fMeta.fSize = size;
return true;
}
void Message::Rebuild()
{
CloseMessage();
fQueued = false;
}
void Message::Rebuild(const size_t size)
{
CloseMessage();
fQueued = false;
InitializeChunk(size);
}
void Message::Rebuild(void* data, const size_t size, fairmq_free_fn* ffn, void* hint)
{
CloseMessage();
fQueued = false;
if (InitializeChunk(size)) {
std::memcpy(fLocalPtr, data, size);
if (ffn) {
ffn(data, hint);
} else {
free(data);
}
}
}
void* Message::GetData() const
{
if (!fLocalPtr) {
if (fMeta.fRegionId == 0) {
if (fMeta.fSize > 0) {
fLocalPtr = reinterpret_cast<char*>(fManager.Segment().get_address_from_handle(fMeta.fHandle));
} else {
fLocalPtr = nullptr;
}
} else {
fRegionPtr = fManager.GetRemoteRegion(fMeta.fRegionId);
if (fRegionPtr) {
fLocalPtr = reinterpret_cast<char*>(fRegionPtr->fRegion.get_address()) + fMeta.fHandle;
} else {
// LOG(warn) << "could not get pointer from a region message";
fLocalPtr = nullptr;
}
}
}
return fLocalPtr;
}
bool Message::SetUsedSize(const size_t size)
{
if (size == fMeta.fSize) {
return true;
} else if (size <= fMeta.fSize) {
try {
bipc::managed_shared_memory::size_type shrunkSize = size;
fLocalPtr = fManager.Segment().allocation_command<char>(bipc::shrink_in_place, fMeta.fSize + 128, shrunkSize, fLocalPtr);
fMeta.fSize = size;
return true;
} catch (bipc::interprocess_exception& e) {
LOG(info) << "could not set used size: " << e.what();
return false;
}
} else {
LOG(error) << "cannot set used size higher than original.";
return false;
}
}
void Message::Copy(const fair::mq::Message& msg)
{
if (fMeta.fHandle < 0) {
bipc::managed_shared_memory::handle_t otherHandle = static_cast<const Message&>(msg).fMeta.fHandle;
if (otherHandle) {
if (InitializeChunk(msg.GetSize())) {
std::memcpy(GetData(), msg.GetData(), msg.GetSize());
}
} else {
LOG(error) << "copy fail: source message not initialized!";
}
} else {
LOG(error) << "copy fail: target message already initialized!";
}
}
void Message::CloseMessage()
{
if (fMeta.fHandle >= 0 && !fQueued) {
if (fMeta.fRegionId == 0) {
fManager.Segment().deallocate(fManager.Segment().get_address_from_handle(fMeta.fHandle));
fMeta.fHandle = -1;
} else {
if (!fRegionPtr) {
fRegionPtr = fManager.GetRemoteRegion(fMeta.fRegionId);
}
if (fRegionPtr) {
fRegionPtr->ReleaseBlock({fMeta.fHandle, fMeta.fSize, fMeta.fHint});
} else {
LOG(warn) << "region ack queue for id " << fMeta.fRegionId << " no longer exist. Not sending ack";
}
}
}
static_cast<TransportFactory*>(GetTransport())->DecrementMsgCounter();
}
Message::~Message()
{
try {
CloseMessage();
} catch(SharedMemoryError& sme) {
LOG(error) << "error closing message: " << sme.what();
} catch(bipc::lock_exception& le) {
LOG(error) << "error closing message: " << le.what();
}
}
}
}
}

79
fairmq/shmem/Message.h Normal file
View File

@@ -0,0 +1,79 @@
/********************************************************************************
* Copyright (C) 2014 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" *
********************************************************************************/
#ifndef FAIR_MQ_SHMEM_MESSAGE_H_
#define FAIR_MQ_SHMEM_MESSAGE_H_
#include "Common.h"
#include "Manager.h"
#include <FairMQMessage.h>
#include <FairMQUnmanagedRegion.h>
#include <boost/interprocess/mapped_region.hpp>
#include <cstddef> // size_t
#include <atomic>
namespace fair
{
namespace mq
{
namespace shmem
{
class Socket;
class Message final : public fair::mq::Message
{
friend class Socket;
public:
Message(Manager& manager, FairMQTransportFactory* factory = nullptr);
Message(Manager& manager, const size_t size, FairMQTransportFactory* factory = nullptr);
Message(Manager& manager, void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr, FairMQTransportFactory* factory = nullptr);
Message(Manager& manager, UnmanagedRegionPtr& region, void* data, const size_t size, void* hint = 0, FairMQTransportFactory* factory = nullptr);
Message(Manager& manager, MetaHeader& hdr, FairMQTransportFactory* factory = nullptr);
Message(const Message&) = delete;
Message operator=(const Message&) = delete;
void Rebuild() override;
void Rebuild(const size_t size) override;
void Rebuild(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr) override;
void* GetData() const override;
size_t GetSize() const override { return fMeta.fSize; }
bool SetUsedSize(const size_t size) override;
Transport GetType() const override { return fTransportType; }
void Copy(const fair::mq::Message& msg) override;
~Message() override;
private:
Manager& fManager;
bool fQueued;
MetaHeader fMeta;
mutable Region* fRegionPtr;
mutable char* fLocalPtr;
static std::atomic<bool> fInterrupted;
static Transport fTransportType;
bool InitializeChunk(const size_t size);
void CloseMessage();
};
}
}
}
#endif /* FAIR_MQ_SHMEM_MESSAGE_H_ */

View File

@@ -6,8 +6,9 @@
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include <fairmq/shmem/Monitor.h>
#include <fairmq/shmem/Common.h>
#include "Monitor.h"
#include "Common.h"
#include <fairmq/Tools.h>
#include <boost/interprocess/managed_shared_memory.hpp>
@@ -275,7 +276,7 @@ void Monitor::CheckSegment()
unsigned int numDevices = 0;
if (fInteractive) {
fair::mq::shmem::DeviceCounter* dc = managementSegment.find<fair::mq::shmem::DeviceCounter>(bipc::unique_instance).first;
DeviceCounter* dc = managementSegment.find<DeviceCounter>(bipc::unique_instance).first;
if (dc) {
numDevices = dc->fCount;
}
@@ -462,6 +463,8 @@ void Monitor::Cleanup(const string& shmId)
RemoveObject(managementSegmentName.c_str());
} catch (bie&) {
cout << "Did not find '" << managementSegmentName << "' shared memory segment. No regions to cleanup." << endl;
} catch(std::out_of_range& oor) {
cout << "Could not locate element in the region map, out of range: " << oor.what() << endl;
}
RemoveObject("fmq_" + shmId + "_main");

View File

@@ -6,21 +6,29 @@
* copied verbatim in the file "LICENSE" *
********************************************************************************/
/**
* FairMQPollerSHM.cxx
* Poller.cxx
*
* @since 2014-01-23
* @author A. Rybalchenko
*/
#include "FairMQPollerSHM.h"
#include "FairMQSocketSHM.h"
#include "FairMQLogger.h"
#include "Poller.h"
#include "Socket.h"
#include <FairMQLogger.h>
#include <zmq.h>
using namespace std;
FairMQPollerSHM::FairMQPollerSHM(const vector<FairMQChannel>& channels)
namespace fair
{
namespace mq
{
namespace shmem
{
Poller::Poller(const vector<FairMQChannel>& channels)
: fItems()
, fNumItems(0)
, fOffsetMap()
@@ -30,19 +38,19 @@ FairMQPollerSHM::FairMQPollerSHM(const vector<FairMQChannel>& channels)
for (int i = 0; i < fNumItems; ++i)
{
fItems[i].socket = static_cast<const FairMQSocketSHM*>(&(channels.at(i).GetSocket()))->GetSocket();
fItems[i].socket = static_cast<const Socket*>(&(channels.at(i).GetSocket()))->GetSocket();
fItems[i].fd = 0;
fItems[i].revents = 0;
int type = 0;
size_t size = sizeof(type);
zmq_getsockopt(static_cast<const FairMQSocketSHM*>(&(channels.at(i).GetSocket()))->GetSocket(), ZMQ_TYPE, &type, &size);
zmq_getsockopt(static_cast<const Socket*>(&(channels.at(i).GetSocket()))->GetSocket(), ZMQ_TYPE, &type, &size);
SetItemEvents(fItems[i], type);
}
}
FairMQPollerSHM::FairMQPollerSHM(const vector<FairMQChannel*>& channels)
Poller::Poller(const vector<FairMQChannel*>& channels)
: fItems()
, fNumItems(0)
, fOffsetMap()
@@ -52,19 +60,19 @@ FairMQPollerSHM::FairMQPollerSHM(const vector<FairMQChannel*>& channels)
for (int i = 0; i < fNumItems; ++i)
{
fItems[i].socket = static_cast<const FairMQSocketSHM*>(&(channels.at(i)->GetSocket()))->GetSocket();
fItems[i].socket = static_cast<const Socket*>(&(channels.at(i)->GetSocket()))->GetSocket();
fItems[i].fd = 0;
fItems[i].revents = 0;
int type = 0;
size_t size = sizeof(type);
zmq_getsockopt(static_cast<const FairMQSocketSHM*>(&(channels.at(i)->GetSocket()))->GetSocket(), ZMQ_TYPE, &type, &size);
zmq_getsockopt(static_cast<const Socket*>(&(channels.at(i)->GetSocket()))->GetSocket(), ZMQ_TYPE, &type, &size);
SetItemEvents(fItems[i], type);
}
}
FairMQPollerSHM::FairMQPollerSHM(const unordered_map<string, vector<FairMQChannel>>& channelsMap, const vector<string>& channelList)
Poller::Poller(const unordered_map<string, vector<FairMQChannel>>& channelsMap, const vector<string>& channelList)
: fItems()
, fNumItems(0)
, fOffsetMap()
@@ -89,19 +97,19 @@ FairMQPollerSHM::FairMQPollerSHM(const unordered_map<string, vector<FairMQChanne
{
index = fOffsetMap[channel] + i;
fItems[index].socket = static_cast<const FairMQSocketSHM*>(&(channelsMap.at(channel).at(i).GetSocket()))->GetSocket();
fItems[index].socket = static_cast<const Socket*>(&(channelsMap.at(channel).at(i).GetSocket()))->GetSocket();
fItems[index].fd = 0;
fItems[index].revents = 0;
int type = 0;
size_t size = sizeof(type);
zmq_getsockopt(static_cast<const FairMQSocketSHM*>(&(channelsMap.at(channel).at(i).GetSocket()))->GetSocket(), ZMQ_TYPE, &type, &size);
zmq_getsockopt(static_cast<const Socket*>(&(channelsMap.at(channel).at(i).GetSocket()))->GetSocket(), ZMQ_TYPE, &type, &size);
SetItemEvents(fItems[index], type);
}
}
}
catch (const std::out_of_range& oor)
catch (const out_of_range& oor)
{
LOG(error) << "at least one of the provided channel keys for poller initialization is invalid";
LOG(error) << "out of range error: " << oor.what() << '\n';
@@ -109,7 +117,7 @@ FairMQPollerSHM::FairMQPollerSHM(const unordered_map<string, vector<FairMQChanne
}
}
void FairMQPollerSHM::SetItemEvents(zmq_pollitem_t& item, const int type)
void Poller::SetItemEvents(zmq_pollitem_t& item, const int type)
{
if (type == ZMQ_REQ || type == ZMQ_REP || type == ZMQ_PAIR || type == ZMQ_DEALER || type == ZMQ_ROUTER)
{
@@ -130,7 +138,7 @@ void FairMQPollerSHM::SetItemEvents(zmq_pollitem_t& item, const int type)
}
}
void FairMQPollerSHM::Poll(const int timeout)
void Poller::Poll(const int timeout)
{
if (zmq_poll(fItems, fNumItems, timeout) < 0)
{
@@ -141,12 +149,12 @@ void FairMQPollerSHM::Poll(const int timeout)
else
{
LOG(error) << "polling failed, reason: " << zmq_strerror(errno);
throw std::runtime_error("polling failed");
throw runtime_error("polling failed");
}
}
}
bool FairMQPollerSHM::CheckInput(const int index)
bool Poller::CheckInput(const int index)
{
if (fItems[index].revents & ZMQ_POLLIN)
{
@@ -156,7 +164,7 @@ bool FairMQPollerSHM::CheckInput(const int index)
return false;
}
bool FairMQPollerSHM::CheckOutput(const int index)
bool Poller::CheckOutput(const int index)
{
if (fItems[index].revents & ZMQ_POLLOUT)
{
@@ -166,7 +174,7 @@ bool FairMQPollerSHM::CheckOutput(const int index)
return false;
}
bool FairMQPollerSHM::CheckInput(const string& channelKey, const int index)
bool Poller::CheckInput(const string& channelKey, const int index)
{
try
{
@@ -177,7 +185,7 @@ bool FairMQPollerSHM::CheckInput(const string& channelKey, const int index)
return false;
}
catch (const std::out_of_range& oor)
catch (const out_of_range& oor)
{
LOG(error) << "invalid channel key: \"" << channelKey << "\"";
LOG(error) << "out of range error: " << oor.what() << '\n';
@@ -185,7 +193,7 @@ bool FairMQPollerSHM::CheckInput(const string& channelKey, const int index)
}
}
bool FairMQPollerSHM::CheckOutput(const string& channelKey, const int index)
bool Poller::CheckOutput(const string& channelKey, const int index)
{
try
{
@@ -196,7 +204,7 @@ bool FairMQPollerSHM::CheckOutput(const string& channelKey, const int index)
return false;
}
catch (const std::out_of_range& oor)
catch (const out_of_range& oor)
{
LOG(error) << "Invalid channel key: \"" << channelKey << "\"";
LOG(error) << "out of range error: " << oor.what() << '\n';
@@ -204,7 +212,11 @@ bool FairMQPollerSHM::CheckOutput(const string& channelKey, const int index)
}
}
FairMQPollerSHM::~FairMQPollerSHM()
Poller::~Poller()
{
delete[] fItems;
}
}
}
}

View File

@@ -5,32 +5,35 @@
* GNU Lesser General Public Licence (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#ifndef FAIRMQPOLLERSHM_H_
#define FAIRMQPOLLERSHM_H_
#ifndef FAIR_MQ_SHMEM_POLLER_H_
#define FAIR_MQ_SHMEM_POLLER_H_
#include <zmq.h>
#include <FairMQPoller.h>
#include <FairMQChannel.h>
#include <vector>
#include <unordered_map>
#include <zmq.h>
#include "FairMQPoller.h"
#include "FairMQChannel.h"
#include "FairMQTransportFactorySHM.h"
class FairMQChannel;
class FairMQPollerSHM final : public FairMQPoller
namespace fair
{
namespace mq
{
namespace shmem
{
friend class FairMQChannel;
friend class FairMQTransportFactorySHM;
class Poller final : public fair::mq::Poller
{
public:
FairMQPollerSHM(const std::vector<FairMQChannel>& channels);
FairMQPollerSHM(const std::vector<FairMQChannel*>& channels);
FairMQPollerSHM(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList);
Poller(const std::vector<FairMQChannel>& channels);
Poller(const std::vector<FairMQChannel*>& channels);
Poller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList);
FairMQPollerSHM(const FairMQPollerSHM&) = delete;
FairMQPollerSHM operator=(const FairMQPollerSHM&) = delete;
Poller(const Poller&) = delete;
Poller operator=(const Poller&) = delete;
void SetItemEvents(zmq_pollitem_t& item, const int type);
@@ -40,7 +43,7 @@ class FairMQPollerSHM final : public FairMQPoller
bool CheckInput(const std::string& channelKey, const int index) override;
bool CheckOutput(const std::string& channelKey, const int index) override;
~FairMQPollerSHM() override;
~Poller() override;
private:
zmq_pollitem_t* fItems;
@@ -49,4 +52,8 @@ class FairMQPollerSHM final : public FairMQPoller
std::unordered_map<std::string, int> fOffsetMap;
};
#endif /* FAIRMQPOLLERSHM_H_ */
}
}
}
#endif /* FAIR_MQ_SHMEM_POLLER_H_ */

View File

@@ -6,17 +6,18 @@
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include <fairmq/shmem/Region.h>
#include <fairmq/shmem/Common.h>
#include <fairmq/shmem/Manager.h>
#include "Region.h"
#include "Common.h"
#include "Manager.h"
#include <fairmq/tools/CppSTL.h>
#include <fairmq/tools/Strings.h>
#include <boost/filesystem.hpp>
#include <boost/process.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <cerrno>
#include <cerrno>
#include <chrono>
using namespace std;
@@ -31,7 +32,7 @@ namespace mq
namespace shmem
{
Region::Region(Manager& manager, uint64_t id, uint64_t size, bool remote, FairMQRegionCallback callback, const string& path /* = "" */, int flags /* = 0 */)
Region::Region(Manager& manager, uint64_t id, uint64_t size, bool remote, RegionCallback callback, const string& path /* = "" */, int flags /* = 0 */)
: fManager(manager)
, fRemote(remote)
, fStop(false)

View File

@@ -6,7 +6,7 @@
* copied verbatim in the file "LICENSE" *
********************************************************************************/
/**
* FairMQShmManager.h
* Region.h
*
* @since 2016-04-08
* @author A. Rybalchenko
@@ -15,10 +15,10 @@
#ifndef FAIR_MQ_SHMEM_REGION_H_
#define FAIR_MQ_SHMEM_REGION_H_
#include "FairMQLogger.h"
#include "FairMQUnmanagedRegion.h"
#include "Common.h"
#include <fairmq/shmem/Common.h>
#include <FairMQLogger.h>
#include <FairMQUnmanagedRegion.h>
#include <boost/interprocess/managed_shared_memory.hpp>
#include <boost/interprocess/file_mapping.hpp>
@@ -40,7 +40,7 @@ class Manager;
struct Region
{
Region(Manager& manager, uint64_t id, uint64_t size, bool remote, FairMQRegionCallback callback = nullptr, const std::string& path = "", int flags = 0);
Region(Manager& manager, uint64_t id, uint64_t size, bool remote, RegionCallback callback = nullptr, const std::string& path = "", int flags = 0);
Region() = delete;
@@ -75,7 +75,7 @@ struct Region
std::thread fReceiveAcksWorker;
std::thread fSendAcksWorker;
FairMQRegionCallback fCallback;
RegionCallback fCallback;
};
} // namespace shmem

View File

@@ -5,12 +5,14 @@
* GNU Lesser General Public Licence (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include "Common.h"
#include "FairMQSocketSHM.h"
#include "FairMQMessageSHM.h"
#include "FairMQUnmanagedRegionSHM.h"
#include "FairMQLogger.h"
#include "Common.h"
#include "Socket.h"
#include "Message.h"
#include "UnmanagedRegion.h"
#include "TransportFactory.h"
#include <FairMQLogger.h>
#include <fairmq/Tools.h>
#include <zmq.h>
@@ -18,13 +20,31 @@
#include <stdexcept>
using namespace std;
using namespace fair::mq::shmem;
using namespace fair::mq;
atomic<bool> FairMQSocketSHM::fInterrupted(false);
namespace fair
{
namespace mq
{
namespace shmem
{
FairMQSocketSHM::FairMQSocketSHM(Manager& manager, const string& type, const string& name, const string& id /*= ""*/, void* context, FairMQTransportFactory* fac /*=nullptr*/)
: FairMQSocket{fac}
atomic<bool> Socket::fInterrupted(false);
struct ZMsg
{
ZMsg() { int rc __attribute__((unused)) = zmq_msg_init(&fMsg); assert(rc == 0); }
explicit ZMsg(size_t size) { int rc __attribute__((unused)) = zmq_msg_init_size(&fMsg, size); assert(rc == 0); }
~ZMsg() { int rc __attribute__((unused)) = zmq_msg_close(&fMsg); assert(rc == 0); }
void* Data() { return zmq_msg_data(&fMsg); }
size_t Size() { return zmq_msg_size(&fMsg); }
zmq_msg_t* Msg() { return &fMsg; }
zmq_msg_t fMsg;
};
Socket::Socket(Manager& manager, const string& type, const string& name, const string& id /*= ""*/, void* context, FairMQTransportFactory* fac /*=nullptr*/)
: fair::mq::Socket{fac}
, fSocket(nullptr)
, fManager(manager)
, fId(id + "." + name + "." + type)
@@ -40,7 +60,7 @@ FairMQSocketSHM::FairMQSocketSHM(Manager& manager, const string& type, const str
if (fSocket == nullptr) {
LOG(error) << "Failed creating socket " << fId << ", reason: " << zmq_strerror(errno);
exit(EXIT_FAILURE);
throw SocketError(tools::ToString("Failed creating socket ", fId, ", reason: ", zmq_strerror(errno)));
}
if (zmq_setsockopt(fSocket, ZMQ_IDENTITY, fId.c_str(), fId.length()) != 0) {
@@ -72,16 +92,14 @@ FairMQSocketSHM::FairMQSocketSHM(Manager& manager, const string& type, const str
if (type == "sub" || type == "pub") {
LOG(error) << "PUB/SUB socket type is not supported for shared memory transport";
throw fair::mq::SocketError("PUB/SUB socket type is not supported for shared memory transport");
throw SocketError("PUB/SUB socket type is not supported for shared memory transport");
}
LOG(debug) << "Created socket " << GetId();
}
bool FairMQSocketSHM::Bind(const string& address)
bool Socket::Bind(const string& address)
{
// LOG(info) << "binding socket " << fId << " on " << address;
if (zmq_bind(fSocket, address.c_str()) != 0) {
if (errno == EADDRINUSE) {
// do not print error in this case, this is handled by FairMQDevice in case no connection could be established after trying a number of random ports from a range.
@@ -93,19 +111,17 @@ bool FairMQSocketSHM::Bind(const string& address)
return true;
}
bool FairMQSocketSHM::Connect(const string& address)
bool Socket::Connect(const string& address)
{
// LOG(info) << "connecting socket " << fId << " on " << address;
if (zmq_connect(fSocket, address.c_str()) != 0) {
LOG(error) << "Failed connecting socket " << fId << ", reason: " << zmq_strerror(errno);
return false;
}
return true;
}
int FairMQSocketSHM::Send(FairMQMessagePtr& msg, const int timeout)
int Socket::Send(MessagePtr& msg, const int timeout)
{
int flags = 0;
if (timeout == 0) {
@@ -113,16 +129,17 @@ int FairMQSocketSHM::Send(FairMQMessagePtr& msg, const int timeout)
}
int elapsed = 0;
Message* shmMsg = static_cast<Message*>(msg.get());
ZMsg zmqMsg(sizeof(MetaHeader));
std::memcpy(zmqMsg.Data(), &(shmMsg->fMeta), sizeof(MetaHeader));
while (true && !fInterrupted) {
int nbytes = zmq_msg_send(static_cast<FairMQMessageSHM*>(msg.get())->GetMessage(), fSocket, flags);
if (nbytes == 0) {
int nbytes = zmq_msg_send(zmqMsg.Msg(), fSocket, flags);
if (nbytes > 0) {
shmMsg->fQueued = true;
++fMessagesTx;
return nbytes;
} else if (nbytes > 0) {
static_cast<FairMQMessageSHM*>(msg.get())->fQueued = true;
size_t size = msg->GetSize();
fBytesTx += size;
++fMessagesTx;
return size;
} else if (zmq_errno() == EAGAIN) {
if (!fInterrupted && ((flags & ZMQ_DONTWAIT) == 0)) {
@@ -140,7 +157,7 @@ int FairMQSocketSHM::Send(FairMQMessagePtr& msg, const int timeout)
LOG(info) << "terminating socket " << fId;
return -1;
} else {
LOG(error) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(error) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno) << ", nbytes = " << nbytes;
return nbytes;
}
}
@@ -148,7 +165,7 @@ int FairMQSocketSHM::Send(FairMQMessagePtr& msg, const int timeout)
return -1;
}
int FairMQSocketSHM::Receive(FairMQMessagePtr& msg, const int timeout)
int Socket::Receive(MessagePtr& msg, const int timeout)
{
int flags = 0;
if (timeout == 0) {
@@ -156,28 +173,18 @@ int FairMQSocketSHM::Receive(FairMQMessagePtr& msg, const int timeout)
}
int elapsed = 0;
ZMsg zmqMsg;
while (true) {
FairMQMessageSHM* shmMsg = static_cast<FairMQMessageSHM*>(msg.get());
zmq_msg_t* zmqMsg = shmMsg->GetMessage();
int nbytes = zmq_msg_recv(zmqMsg, fSocket, flags);
if (nbytes == 0) {
++fMessagesRx;
return nbytes;
} else if (nbytes > 0) {
Message* shmMsg = static_cast<Message*>(msg.get());
int nbytes = zmq_msg_recv(zmqMsg.Msg(), fSocket, flags);
if (nbytes > 0) {
// check for number of received messages. must be 1
const auto numMsgs = nbytes / sizeof(MetaHeader);
if (numMsgs > 1) {
LOG(error) << "Receiving SHM multipart with a single message receive call";
}
assert((nbytes / sizeof(MetaHeader)) == 1);
assert(numMsgs == 1);
MetaHeader* hdr = static_cast<MetaHeader*>(zmq_msg_data(zmqMsg));
MetaHeader* hdr = static_cast<MetaHeader*>(zmqMsg.Data());
size_t size = hdr->fSize;
shmMsg->fHandle = hdr->fHandle;
shmMsg->fSize = size;
shmMsg->fRegionId = hdr->fRegionId;
shmMsg->fHint = hdr->fHint;
shmMsg->fMeta = *hdr;
fBytesRx += size;
++fMessagesRx;
@@ -198,99 +205,74 @@ int FairMQSocketSHM::Receive(FairMQMessagePtr& msg, const int timeout)
LOG(info) << "terminating socket " << fId;
return -1;
} else {
LOG(error) << "Failed receiving on socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(error) << "Failed receiving on socket " << fId << ", reason: " << zmq_strerror(errno) << ", nbytes = " << nbytes;
return nbytes;
}
}
}
int64_t FairMQSocketSHM::Send(vector<FairMQMessagePtr>& msgVec, const int timeout)
int64_t Socket::Send(vector<MessagePtr>& msgVec, const int timeout)
{
int flags = 0;
if (timeout == 0) {
flags = ZMQ_DONTWAIT;
}
const unsigned int vecSize = msgVec.size();
int elapsed = 0;
if (vecSize == 1) {
return Send(msgVec.back(), timeout);
}
// put it into zmq message
zmq_msg_t zmqMsg;
zmq_msg_init_size(&zmqMsg, vecSize * sizeof(MetaHeader));
const unsigned int vecSize = msgVec.size();
ZMsg zmqMsg(vecSize * sizeof(MetaHeader));
// prepare the message with shm metas
MetaHeader* metas = static_cast<MetaHeader*>(zmq_msg_data(&zmqMsg));
MetaHeader* metas = static_cast<MetaHeader*>(zmqMsg.Data());
for (auto& msg : msgVec) {
zmq_msg_t* metaMsg = static_cast<FairMQMessageSHM*>(msg.get())->GetMessage();
if (zmq_msg_size(metaMsg) > 0) {
memcpy(metas++, zmq_msg_data(metaMsg), sizeof(MetaHeader));
} else {
// if the message is empty, create meta data to reflect this
// (always creating meta data for empty messages would add an unnecessary allocation for the receive case, so we do it lazily here)
MetaHeader hdr;
hdr.fSize = 0;
hdr.fHandle = -1;
hdr.fRegionId = 0;
hdr.fHint = 0;
memcpy(metas++, &hdr, sizeof(MetaHeader));
}
Message* shmMsg = static_cast<Message*>(msg.get());
std::memcpy(metas++, &(shmMsg->fMeta), sizeof(MetaHeader));
}
while (!fInterrupted) {
int64_t totalSize = 0;
int nbytes = zmq_msg_send(&zmqMsg, fSocket, flags);
if (nbytes == 0) {
zmq_msg_close(&zmqMsg);
return nbytes;
} else if (nbytes > 0) {
int nbytes = zmq_msg_send(zmqMsg.Msg(), fSocket, flags);
if (nbytes > 0) {
assert(static_cast<unsigned int>(nbytes) == (vecSize * sizeof(MetaHeader))); // all or nothing
for (auto& msg : msgVec) {
FairMQMessageSHM* shmMsg = static_cast<FairMQMessageSHM*>(msg.get());
Message* shmMsg = static_cast<Message*>(msg.get());
shmMsg->fQueued = true;
totalSize += shmMsg->fSize;
totalSize += shmMsg->fMeta.fSize;
}
// store statistics on how many messages have been sent
fMessagesTx++;
fBytesTx += totalSize;
zmq_msg_close(&zmqMsg);
return totalSize;
} else if (zmq_errno() == EAGAIN) {
if (!fInterrupted && ((flags & ZMQ_DONTWAIT) == 0)) {
if (timeout > 0) {
elapsed += fSndTimeout;
if (elapsed >= timeout) {
zmq_msg_close(&zmqMsg);
return -2;
}
}
continue;
} else {
zmq_msg_close(&zmqMsg);
return -2;
}
} else if (zmq_errno() == ETERM) {
zmq_msg_close(&zmqMsg);
LOG(info) << "terminating socket " << fId;
return -1;
} else {
zmq_msg_close(&zmqMsg);
LOG(error) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(error) << "Failed sending on socket " << fId << ", reason: " << zmq_strerror(errno) << ", nbytes = " << nbytes;
return nbytes;
}
}
zmq_msg_close(&zmqMsg);
return -1;
}
int64_t FairMQSocketSHM::Receive(vector<FairMQMessagePtr>& msgVec, const int timeout)
int64_t Socket::Receive(vector<MessagePtr>& msgVec, const int timeout)
{
int flags = 0;
if (timeout == 0) {
@@ -298,18 +280,14 @@ int64_t FairMQSocketSHM::Receive(vector<FairMQMessagePtr>& msgVec, const int tim
}
int elapsed = 0;
zmq_msg_t zmqMsg;
zmq_msg_init(&zmqMsg);
ZMsg zmqMsg;
while (!fInterrupted) {
int64_t totalSize = 0;
int nbytes = zmq_msg_recv(&zmqMsg, fSocket, flags);
if (nbytes == 0) {
zmq_msg_close(&zmqMsg);
return 0;
} else if (nbytes > 0) {
MetaHeader* hdrVec = static_cast<MetaHeader*>(zmq_msg_data(&zmqMsg));
const auto hdrVecSize = zmq_msg_size(&zmqMsg);
int nbytes = zmq_msg_recv(zmqMsg.Msg(), fSocket, flags);
if (nbytes > 0) {
MetaHeader* hdrVec = static_cast<MetaHeader*>(zmqMsg.Data());
const auto hdrVecSize = zmqMsg.Size();
assert(hdrVecSize > 0);
assert(hdrVecSize % sizeof(MetaHeader) == 0);
@@ -318,20 +296,9 @@ int64_t FairMQSocketSHM::Receive(vector<FairMQMessagePtr>& msgVec, const int tim
msgVec.reserve(numMessages);
for (size_t m = 0; m < numMessages; m++) {
// get the meta data pointer
MetaHeader* hdr = &hdrVec[m];
// create new message (part)
msgVec.emplace_back(tools::make_unique<FairMQMessageSHM>(fManager, GetTransport()));
FairMQMessageSHM* shmMsg = static_cast<FairMQMessageSHM*>(msgVec.back().get());
// fill the zmq buffer with the delivered meta data
memcpy(zmq_msg_data(shmMsg->GetMessage()), hdr, sizeof(MetaHeader));
// set the message members with the meta data
shmMsg->fHandle = hdr->fHandle;
shmMsg->fSize = hdr->fSize;
shmMsg->fRegionId = hdr->fRegionId;
shmMsg->fHint = hdr->fHint;
msgVec.emplace_back(tools::make_unique<Message>(fManager, hdrVec[m], GetTransport()));
Message* shmMsg = static_cast<Message*>(msgVec.back().get());
totalSize += shmMsg->GetSize();
}
@@ -339,34 +306,29 @@ int64_t FairMQSocketSHM::Receive(vector<FairMQMessagePtr>& msgVec, const int tim
fMessagesRx++;
fBytesRx += totalSize;
zmq_msg_close(&zmqMsg);
return totalSize;
} else if (zmq_errno() == EAGAIN) {
if (!fInterrupted && ((flags & ZMQ_DONTWAIT) == 0)) {
if (timeout > 0) {
elapsed += fRcvTimeout;
if (elapsed >= timeout) {
zmq_msg_close(&zmqMsg);
return -2;
}
}
continue;
} else {
zmq_msg_close(&zmqMsg);
return -2;
}
} else {
zmq_msg_close(&zmqMsg);
LOG(error) << "Failed receiving on socket " << fId << ", reason: " << zmq_strerror(errno);
LOG(error) << "Failed receiving on socket " << fId << ", reason: " << zmq_strerror(errno) << ", nbytes = " << nbytes;
return nbytes;
}
}
zmq_msg_close(&zmqMsg);
return -1;
}
void FairMQSocketSHM::Close()
void Socket::Close()
{
// LOG(debug) << "Closing socket " << fId;
@@ -381,42 +343,42 @@ void FairMQSocketSHM::Close()
fSocket = nullptr;
}
void FairMQSocketSHM::Interrupt()
void Socket::Interrupt()
{
Manager::Interrupt();
FairMQMessageSHM::fInterrupted = true;
Message::fInterrupted = true;
fInterrupted = true;
}
void FairMQSocketSHM::Resume()
void Socket::Resume()
{
Manager::Resume();
FairMQMessageSHM::fInterrupted = false;
Message::fInterrupted = false;
fInterrupted = false;
}
void FairMQSocketSHM::SetOption(const string& option, const void* value, size_t valueSize)
void Socket::SetOption(const string& option, const void* value, size_t valueSize)
{
if (zmq_setsockopt(fSocket, GetConstant(option), value, valueSize) < 0) {
LOG(error) << "Failed setting socket option, reason: " << zmq_strerror(errno);
}
}
void FairMQSocketSHM::GetOption(const string& option, void* value, size_t* valueSize)
void Socket::GetOption(const string& option, void* value, size_t* valueSize)
{
if (zmq_getsockopt(fSocket, GetConstant(option), value, valueSize) < 0) {
LOG(error) << "Failed getting socket option, reason: " << zmq_strerror(errno);
}
}
void FairMQSocketSHM::SetLinger(const int value)
void Socket::SetLinger(const int value)
{
if (zmq_setsockopt(fSocket, ZMQ_LINGER, &value, sizeof(value)) < 0) {
throw SocketError(tools::ToString("failed setting ZMQ_LINGER, reason: ", zmq_strerror(errno)));
}
}
int FairMQSocketSHM::GetLinger() const
int Socket::GetLinger() const
{
int value = 0;
size_t valueSize = sizeof(value);
@@ -426,14 +388,14 @@ int FairMQSocketSHM::GetLinger() const
return value;
}
void FairMQSocketSHM::SetSndBufSize(const int value)
void Socket::SetSndBufSize(const int value)
{
if (zmq_setsockopt(fSocket, ZMQ_SNDHWM, &value, sizeof(value)) < 0) {
throw SocketError(tools::ToString("failed setting ZMQ_SNDHWM, reason: ", zmq_strerror(errno)));
}
}
int FairMQSocketSHM::GetSndBufSize() const
int Socket::GetSndBufSize() const
{
int value = 0;
size_t valueSize = sizeof(value);
@@ -443,14 +405,14 @@ int FairMQSocketSHM::GetSndBufSize() const
return value;
}
void FairMQSocketSHM::SetRcvBufSize(const int value)
void Socket::SetRcvBufSize(const int value)
{
if (zmq_setsockopt(fSocket, ZMQ_RCVHWM, &value, sizeof(value)) < 0) {
throw SocketError(tools::ToString("failed setting ZMQ_RCVHWM, reason: ", zmq_strerror(errno)));
}
}
int FairMQSocketSHM::GetRcvBufSize() const
int Socket::GetRcvBufSize() const
{
int value = 0;
size_t valueSize = sizeof(value);
@@ -460,14 +422,14 @@ int FairMQSocketSHM::GetRcvBufSize() const
return value;
}
void FairMQSocketSHM::SetSndKernelSize(const int value)
void Socket::SetSndKernelSize(const int value)
{
if (zmq_setsockopt(fSocket, ZMQ_SNDBUF, &value, sizeof(value)) < 0) {
throw SocketError(tools::ToString("failed getting ZMQ_SNDBUF, reason: ", zmq_strerror(errno)));
}
}
int FairMQSocketSHM::GetSndKernelSize() const
int Socket::GetSndKernelSize() const
{
int value = 0;
size_t valueSize = sizeof(value);
@@ -477,14 +439,14 @@ int FairMQSocketSHM::GetSndKernelSize() const
return value;
}
void FairMQSocketSHM::SetRcvKernelSize(const int value)
void Socket::SetRcvKernelSize(const int value)
{
if (zmq_setsockopt(fSocket, ZMQ_RCVBUF, &value, sizeof(value)) < 0) {
throw SocketError(tools::ToString("failed getting ZMQ_RCVBUF, reason: ", zmq_strerror(errno)));
}
}
int FairMQSocketSHM::GetRcvKernelSize() const
int Socket::GetRcvKernelSize() const
{
int value = 0;
size_t valueSize = sizeof(value);
@@ -494,7 +456,7 @@ int FairMQSocketSHM::GetRcvKernelSize() const
return value;
}
int FairMQSocketSHM::GetConstant(const string& constant)
int Socket::GetConstant(const string& constant)
{
if (constant == "") return 0;
if (constant == "sub") return ZMQ_SUB;
@@ -522,3 +484,7 @@ int FairMQSocketSHM::GetConstant(const string& constant)
return -1;
}
}
}
}

View File

@@ -5,34 +5,42 @@
* GNU Lesser General Public Licence (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#ifndef FAIRMQSOCKETSHM_H_
#define FAIRMQSOCKETSHM_H_
#ifndef FAIR_MQ_SHMEM_SOCKET_H_
#define FAIR_MQ_SHMEM_SOCKET_H_
#include "FairMQSocket.h"
#include "FairMQMessage.h"
#include "Manager.h"
#include <fairmq/shmem/Manager.h>
#include <FairMQSocket.h>
#include <FairMQMessage.h>
#include <atomic>
#include <memory> // unique_ptr
class FairMQTransportFactory;
class FairMQSocketSHM final : public FairMQSocket
namespace fair
{
namespace mq
{
namespace shmem
{
class Socket final : public fair::mq::Socket
{
public:
FairMQSocketSHM(fair::mq::shmem::Manager& manager, const std::string& type, const std::string& name, const std::string& id = "", void* context = nullptr, FairMQTransportFactory* fac = nullptr);
FairMQSocketSHM(const FairMQSocketSHM&) = delete;
FairMQSocketSHM operator=(const FairMQSocketSHM&) = delete;
Socket(Manager& manager, const std::string& type, const std::string& name, const std::string& id = "", void* context = nullptr, FairMQTransportFactory* fac = nullptr);
Socket(const Socket&) = delete;
Socket operator=(const Socket&) = delete;
std::string GetId() override { return fId; }
std::string GetId() const override { return fId; }
bool Bind(const std::string& address) override;
bool Connect(const std::string& address) override;
int Send(FairMQMessagePtr& msg, const int timeout = -1) override;
int Receive(FairMQMessagePtr& msg, const int timeout = -1) override;
int64_t Send(std::vector<std::unique_ptr<FairMQMessage>>& msgVec, const int timeout = -1) override;
int64_t Receive(std::vector<std::unique_ptr<FairMQMessage>>& msgVec, const int timeout = -1) override;
int Send(MessagePtr& msg, const int timeout = -1) override;
int Receive(MessagePtr& msg, const int timeout = -1) override;
int64_t Send(std::vector<MessagePtr>& msgVec, const int timeout = -1) override;
int64_t Receive(std::vector<MessagePtr>& msgVec, const int timeout = -1) override;
void* GetSocket() const { return fSocket; }
@@ -62,11 +70,11 @@ class FairMQSocketSHM final : public FairMQSocket
static int GetConstant(const std::string& constant);
~FairMQSocketSHM() override { Close(); }
~Socket() override { Close(); }
private:
void* fSocket;
fair::mq::shmem::Manager& fManager;
Manager& fManager;
std::string fId;
std::atomic<unsigned long> fBytesTx;
std::atomic<unsigned long> fBytesRx;
@@ -79,4 +87,8 @@ class FairMQSocketSHM final : public FairMQSocket
int fRcvTimeout;
};
#endif /* FAIRMQSOCKETSHM_H_ */
}
}
}
#endif /* FAIR_MQ_SHMEM_SOCKET_H_ */

View File

@@ -6,9 +6,9 @@
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include "FairMQLogger.h"
#include "FairMQTransportFactorySHM.h"
#include "TransportFactory.h"
#include <FairMQLogger.h>
#include <fairmq/Tools.h>
#include <zmq.h>
@@ -26,21 +26,28 @@
#include <cstdlib> // getenv
using namespace std;
using namespace fair::mq::shmem;
namespace bpt = ::boost::posix_time;
namespace bipc = ::boost::interprocess;
fair::mq::Transport FairMQTransportFactorySHM::fTransportType = fair::mq::Transport::SHM;
namespace fair
{
namespace mq
{
namespace shmem
{
FairMQTransportFactorySHM::FairMQTransportFactorySHM(const string& id, const fair::mq::ProgOptions* config)
: FairMQTransportFactory(id)
Transport TransportFactory::fTransportType = Transport::SHM;
TransportFactory::TransportFactory(const string& id, const ProgOptions* config)
: fair::mq::TransportFactory(id)
, fDeviceId(id)
, fShmId()
, fZMQContext(nullptr)
, fManager(nullptr)
, fHeartbeatThread()
, fSendHeartbeats(true)
, fMsgCounter(0)
{
int major, minor, patch;
zmq_version(&major, &minor, &patch);
@@ -49,7 +56,7 @@ FairMQTransportFactorySHM::FairMQTransportFactorySHM(const string& id, const fai
fZMQContext = zmq_ctx_new();
if (!fZMQContext) {
throw runtime_error(fair::mq::tools::ToString("failed creating context, reason: ", zmq_strerror(errno)));
throw runtime_error(tools::ToString("failed creating context, reason: ", zmq_strerror(errno)));
}
int numIoThreads = 1;
@@ -62,7 +69,7 @@ FairMQTransportFactorySHM::FairMQTransportFactorySHM(const string& id, const fai
segmentSize = config->GetProperty<size_t>("shm-segment-size", segmentSize);
autolaunchMonitor = config->GetProperty<bool>("shm-monitor", autolaunchMonitor);
} else {
LOG(debug) << "fair::mq::ProgOptions not available! Using defaults.";
LOG(debug) << "ProgOptions not available! Using defaults.";
}
fShmId = buildShmIdFromSessionIdAndUserId(sessionName);
@@ -81,18 +88,18 @@ FairMQTransportFactorySHM::FairMQTransportFactorySHM(const string& id, const fai
Manager::StartMonitor(fShmId);
}
fManager = fair::mq::tools::make_unique<Manager>(fShmId, segmentSize);
fManager = tools::make_unique<Manager>(fShmId, segmentSize);
} catch (bipc::interprocess_exception& e) {
LOG(error) << "Could not initialize shared memory transport: " << e.what();
throw runtime_error(fair::mq::tools::ToString("Could not initialize shared memory transport: ", e.what()));
throw runtime_error(tools::ToString("Could not initialize shared memory transport: ", e.what()));
}
fSendHeartbeats = true;
fHeartbeatThread = thread(&FairMQTransportFactorySHM::SendHeartbeats, this);
fHeartbeatThread = thread(&TransportFactory::SendHeartbeats, this);
}
void FairMQTransportFactorySHM::SendHeartbeats()
void TransportFactory::SendHeartbeats()
{
string controlQueueName("fmq_" + fShmId + "_cq");
while (fSendHeartbeats) {
@@ -111,58 +118,67 @@ void FairMQTransportFactorySHM::SendHeartbeats()
}
}
FairMQMessagePtr FairMQTransportFactorySHM::CreateMessage()
MessagePtr TransportFactory::CreateMessage()
{
return unique_ptr<FairMQMessage>(new FairMQMessageSHM(*fManager, this));
return tools::make_unique<Message>(*fManager, this);
}
FairMQMessagePtr FairMQTransportFactorySHM::CreateMessage(const size_t size)
MessagePtr TransportFactory::CreateMessage(const size_t size)
{
return unique_ptr<FairMQMessage>(new FairMQMessageSHM(*fManager, size, this));
return tools::make_unique<Message>(*fManager, size, this);
}
FairMQMessagePtr FairMQTransportFactorySHM::CreateMessage(void* data, const size_t size, fairmq_free_fn* ffn, void* hint)
MessagePtr TransportFactory::CreateMessage(void* data, const size_t size, fairmq_free_fn* ffn, void* hint)
{
return unique_ptr<FairMQMessage>(new FairMQMessageSHM(*fManager, data, size, ffn, hint, this));
return tools::make_unique<Message>(*fManager, data, size, ffn, hint, this);
}
FairMQMessagePtr FairMQTransportFactorySHM::CreateMessage(FairMQUnmanagedRegionPtr& region, void* data, const size_t size, void* hint)
MessagePtr TransportFactory::CreateMessage(UnmanagedRegionPtr& region, void* data, const size_t size, void* hint)
{
return unique_ptr<FairMQMessage>(new FairMQMessageSHM(*fManager, region, data, size, hint, this));
return tools::make_unique<Message>(*fManager, region, data, size, hint, this);
}
FairMQSocketPtr FairMQTransportFactorySHM::CreateSocket(const string& type, const string& name)
SocketPtr TransportFactory::CreateSocket(const string& type, const string& name)
{
assert(fZMQContext);
return unique_ptr<FairMQSocket>(new FairMQSocketSHM(*fManager, type, name, GetId(), fZMQContext, this));
return tools::make_unique<Socket>(*fManager, type, name, GetId(), fZMQContext, this);
}
FairMQPollerPtr FairMQTransportFactorySHM::CreatePoller(const vector<FairMQChannel>& channels) const
PollerPtr TransportFactory::CreatePoller(const vector<FairMQChannel>& channels) const
{
return unique_ptr<FairMQPoller>(new FairMQPollerSHM(channels));
return tools::make_unique<Poller>(channels);
}
FairMQPollerPtr FairMQTransportFactorySHM::CreatePoller(const vector<FairMQChannel*>& channels) const
PollerPtr TransportFactory::CreatePoller(const vector<FairMQChannel*>& channels) const
{
return unique_ptr<FairMQPoller>(new FairMQPollerSHM(channels));
return tools::make_unique<Poller>(channels);
}
FairMQPollerPtr FairMQTransportFactorySHM::CreatePoller(const unordered_map<string, vector<FairMQChannel>>& channelsMap, const vector<string>& channelList) const
PollerPtr TransportFactory::CreatePoller(const unordered_map<string, vector<FairMQChannel>>& channelsMap, const vector<string>& channelList) const
{
return unique_ptr<FairMQPoller>(new FairMQPollerSHM(channelsMap, channelList));
return tools::make_unique<Poller>(channelsMap, channelList);
}
FairMQUnmanagedRegionPtr FairMQTransportFactorySHM::CreateUnmanagedRegion(const size_t size, FairMQRegionCallback callback, const std::string& path /* = "" */, int flags /* = 0 */) const
UnmanagedRegionPtr TransportFactory::CreateUnmanagedRegion(const size_t size, RegionCallback callback, const std::string& path /* = "" */, int flags /* = 0 */) const
{
return unique_ptr<FairMQUnmanagedRegion>(new FairMQUnmanagedRegionSHM(*fManager, size, callback, path, flags));
return tools::make_unique<UnmanagedRegion>(*fManager, size, callback, path, flags);
}
fair::mq::Transport FairMQTransportFactorySHM::GetType() const
Transport TransportFactory::GetType() const
{
return fTransportType;
}
FairMQTransportFactorySHM::~FairMQTransportFactorySHM()
void TransportFactory::Reset()
{
if (fMsgCounter.load() != 0) {
LOG(error) << "Message counter during Reset expected to be 0, found: " << fMsgCounter.load();
throw MessageError(tools::ToString("Message counter during Reset expected to be 0, found: ", fMsgCounter.load()));
}
}
TransportFactory::~TransportFactory()
{
LOG(debug) << "Destroying Shared Memory transport...";
fSendHeartbeats = false;
@@ -181,3 +197,7 @@ FairMQTransportFactorySHM::~FairMQTransportFactorySHM()
LOG(error) << "context not available for shutdown";
}
}
} // namespace shmem
} // namespace mq
} // namespace fair

View File

@@ -0,0 +1,82 @@
/********************************************************************************
* Copyright (C) 2016-2017 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" *
********************************************************************************/
#ifndef FAIR_MQ_SHMEM_TRANSPORTFACTORY_H_
#define FAIR_MQ_SHMEM_TRANSPORTFACTORY_H_
#include "Manager.h"
#include "Common.h"
#include "Message.h"
#include "Socket.h"
#include "Poller.h"
#include "UnmanagedRegion.h"
#include <FairMQTransportFactory.h>
#include <fairmq/ProgOptions.h>
#include <vector>
#include <string>
#include <thread>
#include <atomic>
namespace fair
{
namespace mq
{
namespace shmem
{
class TransportFactory final : public fair::mq::TransportFactory
{
public:
TransportFactory(const std::string& id = "", const ProgOptions* config = nullptr);
TransportFactory(const TransportFactory&) = delete;
TransportFactory operator=(const TransportFactory&) = delete;
MessagePtr CreateMessage() override;
MessagePtr CreateMessage(const size_t size) override;
MessagePtr CreateMessage(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr) override;
MessagePtr CreateMessage(UnmanagedRegionPtr& region, void* data, const size_t size, void* hint = 0) override;
SocketPtr CreateSocket(const std::string& type, const std::string& name) override;
PollerPtr CreatePoller(const std::vector<FairMQChannel>& channels) const override;
PollerPtr CreatePoller(const std::vector<FairMQChannel*>& channels) const override;
PollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList) const override;
UnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, RegionCallback callback = nullptr, const std::string& path = "", int flags = 0) const override;
Transport GetType() const override;
void Interrupt() override { Socket::Interrupt(); }
void Resume() override { Socket::Resume(); }
void Reset() override;
void IncrementMsgCounter() { ++fMsgCounter; }
void DecrementMsgCounter() { --fMsgCounter; }
~TransportFactory() override;
private:
void SendHeartbeats();
static Transport fTransportType;
std::string fDeviceId;
std::string fShmId;
void* fZMQContext;
std::unique_ptr<Manager> fManager;
std::thread fHeartbeatThread;
std::atomic<bool> fSendHeartbeats;
std::atomic<int32_t> fMsgCounter;
};
} // namespace shmem
} // namespace mq
} // namespace fair
#endif /* FAIR_MQ_SHMEM_TRANSPORTFACTORY_H_ */

View File

@@ -7,15 +7,20 @@
********************************************************************************/
#include "Common.h"
#include "FairMQUnmanagedRegionSHM.h"
#include "UnmanagedRegion.h"
using namespace std;
using namespace fair::mq::shmem;
namespace bipc = ::boost::interprocess;
FairMQUnmanagedRegionSHM::FairMQUnmanagedRegionSHM(Manager& manager, const size_t size, FairMQRegionCallback callback, const std::string& path /* = "" */, int flags /* = 0 */)
namespace fair
{
namespace mq
{
namespace shmem
{
UnmanagedRegion::UnmanagedRegion(Manager& manager, const size_t size, RegionCallback callback, const std::string& path /* = "" */, int flags /* = 0 */)
: fManager(manager)
, fRegion(nullptr)
, fRegionId(0)
@@ -41,3 +46,7 @@ FairMQUnmanagedRegionSHM::FairMQUnmanagedRegionSHM(Manager& manager, const size_
throw;
}
}
}
}
}

View File

@@ -6,13 +6,13 @@
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#ifndef FAIRMQUNMANAGEDREGIONSHM_H_
#define FAIRMQUNMANAGEDREGIONSHM_H_
#ifndef FAIR_MQ_SHMEM_UNMANAGEDREGION_H_
#define FAIR_MQ_SHMEM_UNMANAGEDREGION_H_
#include <fairmq/shmem/Manager.h>
#include "Manager.h"
#include "FairMQUnmanagedRegion.h"
#include "FairMQLogger.h"
#include <FairMQUnmanagedRegion.h>
#include <FairMQLogger.h>
#include <boost/interprocess/shared_memory_object.hpp>
#include <boost/interprocess/mapped_region.hpp>
@@ -20,23 +20,37 @@
#include <cstddef> // size_t
#include <string>
class FairMQUnmanagedRegionSHM final : public FairMQUnmanagedRegion
namespace fair
{
friend class FairMQSocketSHM;
friend class FairMQMessageSHM;
namespace mq
{
namespace shmem
{
class Message;
class Socket;
class UnmanagedRegion final : public fair::mq::UnmanagedRegion
{
friend class Message;
friend class Socket;
public:
FairMQUnmanagedRegionSHM(fair::mq::shmem::Manager& manager, const size_t size, FairMQRegionCallback callback = nullptr, const std::string& path = "", int flags = 0);
UnmanagedRegion(Manager& manager, const size_t size, RegionCallback callback = nullptr, const std::string& path = "", int flags = 0);
void* GetData() const override { return fRegion->get_address(); }
size_t GetSize() const override { return fRegion->get_size(); }
~FairMQUnmanagedRegionSHM() override { fManager.RemoveRegion(fRegionId); }
~UnmanagedRegion() override { fManager.RemoveRegion(fRegionId); }
private:
fair::mq::shmem::Manager& fManager;
Manager& fManager;
boost::interprocess::mapped_region* fRegion;
uint64_t fRegionId;
};
#endif /* FAIRMQUNMANAGEDREGIONSHM_H_ */
}
}
}
#endif /* FAIR_MQ_SHMEM_UNMANAGEDREGION_H_ */

View File

@@ -5,8 +5,8 @@
* GNU Lesser General Public Licence (LGPL) version 3, *
* copied verbatim in the file "LICENSE" *
********************************************************************************/
#include <fairmq/shmem/Monitor.h>
#include <fairmq/shmem/Common.h>
#include "Monitor.h"
#include "Common.h"
#include <boost/program_options.hpp>

View File

@@ -74,11 +74,6 @@ FairMQSocketZMQ::FairMQSocketZMQ(const string& type, const string& name, const s
LOG(debug) << "Created socket " << GetId();
}
string FairMQSocketZMQ::GetId()
{
return fId;
}
bool FairMQSocketZMQ::Bind(const string& address)
{
// LOG(info) << "bind socket " << fId << " on " << address;

View File

@@ -24,7 +24,7 @@ class FairMQSocketZMQ final : public FairMQSocket
FairMQSocketZMQ(const FairMQSocketZMQ&) = delete;
FairMQSocketZMQ operator=(const FairMQSocketZMQ&) = delete;
std::string GetId() override;
std::string GetId() const override { return fId; }
bool Bind(const std::string& address) override;
bool Connect(const std::string& address) override;

View File

@@ -11,6 +11,7 @@
#include <gtest/gtest.h>
#include <boost/process.hpp>
#include <fairmq/Tools.h>
#include <FairMQDevice.h>
#include <string>
#include <thread>
@@ -23,6 +24,39 @@ using namespace std;
using namespace fair::mq::test;
using namespace fair::mq::tools;
class BadDevice : public FairMQDevice
{
public:
BadDevice()
{
fDeviceThread = thread([&](){
EXPECT_THROW(RunStateMachine(), fair::mq::MessageError);
});
SetTransport("shmem");
ChangeState(fair::mq::Transition::InitDevice);
WaitForState(fair::mq::State::InitializingDevice);
ChangeState(fair::mq::Transition::CompleteInit);
WaitForState(fair::mq::State::Initialized);
parts.AddPart(NewMessage());
}
~BadDevice()
{
ChangeState(fair::mq::Transition::ResetDevice);
if (fDeviceThread.joinable()) {
fDeviceThread.join();
}
}
private:
thread fDeviceThread;
FairMQParts parts;
};
void RunErrorStateIn(const string& state, const string& control, const string& input = "")
{
size_t session{fair::mq::tools::UuidHash()};
@@ -118,4 +152,9 @@ TEST(ErrorState, interactive_InReset)
EXPECT_EXIT(RunErrorStateIn("Reset", "interactive", "q"), ::testing::ExitedWithCode(1), "");
}
TEST(ErrorState, OrphanMessages)
{
BadDevice badDevice;
}
} // namespace

View File

@@ -25,45 +25,86 @@ namespace
using namespace std;
auto RunSingleThreadedMultipart(string transport, string address) -> void {
auto RunSingleThreadedMultipart(string transport, string address1, string address2) -> void {
size_t session{fair::mq::tools::UuidHash()};
fair::mq::ProgOptions config;
config.SetProperty<string>("session", std::to_string(session));
auto factory = FairMQTransportFactory::CreateTransportFactory(transport, fair::mq::tools::Uuid(), &config);
FairMQTransportFactory* factoryptr = factory.get();
FairMQChannel push("Push", "push", factory);
ASSERT_TRUE(push.Bind(address));
FairMQChannel pull("Pull", "pull", factory);
pull.Connect(address);
FairMQChannel push1("Push1", "push", factory);
ASSERT_TRUE(push1.Bind(address1));
FairMQChannel pull1("Pull1", "pull", factory);
ASSERT_TRUE(pull1.Connect(address1));
FairMQChannel push2("Push2", "push", factory);
ASSERT_TRUE(push2.Bind(address2));
FairMQChannel pull2("Pull2", "pull", factory);
ASSERT_TRUE(pull2.Connect(address2));
// TODO validate that fTransportFactory is not nullptr
// TODO validate that fSocket is not nullptr
ASSERT_TRUE(push.Validate());
ASSERT_TRUE(pull.Validate());
ASSERT_TRUE(push1.Validate());
ASSERT_TRUE(pull1.Validate());
ASSERT_TRUE(push2.Validate());
ASSERT_TRUE(pull2.Validate());
{
auto sentMsg = FairMQParts{};
sentMsg.AddPart(push.NewSimpleMessage("1"));
sentMsg.AddPart(push.NewSimpleMessage("2"));
sentMsg.AddPart(push.NewSimpleMessage("3"));
FairMQParts multiplePartsOut;
multiplePartsOut.AddPart(push1.NewSimpleMessage("1"));
multiplePartsOut.AddPart(push1.NewSimpleMessage("2"));
multiplePartsOut.AddPart(push1.NewSimpleMessage("3"));
ASSERT_GE(push1.Send(multiplePartsOut), 0);
ASSERT_GE(push.Send(sentMsg), 0);
FairMQParts singlePartOut;
singlePartOut.AddPart(push1.NewSimpleMessage("4"));
ASSERT_GE(push1.Send(singlePartOut), 0);
}
auto receivedMsg = FairMQParts{};
ASSERT_GE(pull.Receive(receivedMsg), 0);
FairMQParts multipleParts;
ASSERT_GE(pull1.Receive(multipleParts), 0);
stringstream out;
for_each(receivedMsg.cbegin(), receivedMsg.cend(), [&out,&factoryptr](const FairMQMessagePtr& part) {
out << string{static_cast<char*>(part->GetData()), part->GetSize()};
ASSERT_EQ(part->GetTransport(),factoryptr);
stringstream multiple;
for_each(multipleParts.cbegin(), multipleParts.cend(), [&multiple, &factory](const FairMQMessagePtr& part) {
multiple << string{static_cast<char*>(part->GetData()), part->GetSize()};
ASSERT_EQ(part->GetTransport(), factory.get());
});
ASSERT_EQ(out.str(), "123");
ASSERT_EQ(multiple.str(), "123");
FairMQParts singlePart;
ASSERT_GE(pull1.Receive(singlePart), 0);
stringstream single;
for_each(singlePart.cbegin(), singlePart.cend(), [&single](const FairMQMessagePtr& part) {
single << string{static_cast<char*>(part->GetData()), part->GetSize()};
});
ASSERT_EQ(single.str(), "4");
ASSERT_GE(push2.Send(singlePart), 0);
ASSERT_GE(push2.Send(multipleParts), 0);
{
FairMQParts singlePartIn;
ASSERT_GE(pull2.Receive(singlePartIn), 0);
stringstream singleIn;
for_each(singlePartIn.cbegin(), singlePartIn.cend(), [&singleIn](const FairMQMessagePtr& part) {
singleIn << string{static_cast<char*>(part->GetData()), part->GetSize()};
});
ASSERT_EQ(singleIn.str(), "4");
FairMQParts multiplePartsIn;
ASSERT_GE(pull2.Receive(multiplePartsIn), 0);
stringstream multipleIn;
for_each(multiplePartsIn.cbegin(), multiplePartsIn.cend(), [&multipleIn, &factory](const FairMQMessagePtr& part) {
multipleIn << string{static_cast<char*>(part->GetData()), part->GetSize()};
ASSERT_EQ(part->GetTransport(), factory.get());
});
ASSERT_EQ(multipleIn.str(), "123");
}
}
auto RunMultiThreadedMultipart(string transport, string address) -> void
auto RunMultiThreadedMultipart(string transport, string address1) -> void
{
size_t session{fair::mq::tools::UuidHash()};
@@ -71,31 +112,33 @@ auto RunMultiThreadedMultipart(string transport, string address) -> void
config.SetProperty<string>("session", std::to_string(session));
config.SetProperty<int>("io-threads", 1);
config.SetProperty<size_t>("shm-segment-size", 20000000);
auto factory = FairMQTransportFactory::CreateTransportFactory(transport, fair::mq::tools::Uuid(), &config);
FairMQChannel push("Push", "push", factory);
ASSERT_TRUE(push.Bind(address));
FairMQChannel pull("Pull", "pull", factory);
pull.Connect(address);
auto pusher = thread{[&push](){
ASSERT_TRUE(push.Validate());
FairMQChannel push1("Push1", "push", factory);
ASSERT_TRUE(push1.Bind(address1));
FairMQChannel pull1("Pull1", "pull", factory);
ASSERT_TRUE(pull1.Connect(address1));
auto sentMsg = FairMQParts{};
sentMsg.AddPart(push.NewSimpleMessage("1"));
sentMsg.AddPart(push.NewSimpleMessage("2"));
sentMsg.AddPart(push.NewSimpleMessage("3"));
auto pusher = thread{[&push1](){
ASSERT_TRUE(push1.Validate());
ASSERT_GE(push.Send(sentMsg), 0);
FairMQParts sent;
sent.AddPart(push1.NewSimpleMessage("1"));
sent.AddPart(push1.NewSimpleMessage("2"));
sent.AddPart(push1.NewSimpleMessage("3"));
ASSERT_GE(push1.Send(sent), 0);
}};
auto puller = thread{[&pull](){
ASSERT_TRUE(pull.Validate());
auto puller = thread{[&pull1](){
ASSERT_TRUE(pull1.Validate());
auto receivedMsg = FairMQParts{};
ASSERT_GE(pull.Receive(receivedMsg), 0);
FairMQParts received;
ASSERT_GE(pull1.Receive(received), 0);
stringstream out;
for_each(receivedMsg.cbegin(), receivedMsg.cend(), [&out](const FairMQMessagePtr& part) {
for_each(received.cbegin(), received.cend(), [&out](const FairMQMessagePtr& part) {
out << string{static_cast<char*>(part->GetData()), part->GetSize()};
});
ASSERT_EQ(out.str(), "123");
@@ -107,69 +150,69 @@ auto RunMultiThreadedMultipart(string transport, string address) -> void
TEST(PushPull, Multipart_ST_inproc_zeromq)
{
RunSingleThreadedMultipart("zeromq", "inproc://test");
RunSingleThreadedMultipart("zeromq", "inproc://test1", "inproc://test2");
}
TEST(PushPull, Multipart_ST_inproc_shmem)
{
RunSingleThreadedMultipart("shmem", "inproc://test");
RunSingleThreadedMultipart("shmem", "inproc://test1", "inproc://test2");
}
#ifdef BUILD_NANOMSG_TRANSPORT
TEST(PushPull, Multipart_ST_inproc_nanomsg)
{
RunSingleThreadedMultipart("nanomsg", "inproc://test");
RunSingleThreadedMultipart("nanomsg", "inproc://test1", "inproc://test2");
}
#endif /* BUILD_NANOMSG_TRANSPORT */
TEST(PushPull, Multipart_ST_ipc_zeromq)
{
RunSingleThreadedMultipart("zeromq", "ipc://test_Multipart_ST_ipc_zeromq");
RunSingleThreadedMultipart("zeromq", "ipc://test_Multipart_ST_ipc_zeromq_1", "ipc://test_Multipart_ST_ipc_zeromq_2");
}
TEST(PushPull, Multipart_ST_ipc_shmen)
{
RunSingleThreadedMultipart("shmem", "ipc://test_Multipart_ST_ipc_shmen");
RunSingleThreadedMultipart("shmem", "ipc://test_Multipart_ST_ipc_shmen_1", "ipc://test_Multipart_ST_ipc_shmen_2");
}
#ifdef BUILD_NANOMSG_TRANSPORT
TEST(PushPull, Multipart_ST_ipc_nanomsg)
{
RunSingleThreadedMultipart("nanomsg", "ipc://test_Multipart_ST_ipc_nanomsg");
RunSingleThreadedMultipart("nanomsg", "ipc://test_Multipart_ST_ipc_nanomsg_1", "ipc://test_Multipart_ST_ipc_nanomsg_2");
}
#endif /* BUILD_NANOMSG_TRANSPORT */
TEST(PushPull, Multipart_MT_inproc_zeromq)
{
RunMultiThreadedMultipart("zeromq", "inproc://test");
RunMultiThreadedMultipart("zeromq", "inproc://test_1");
}
TEST(PushPull, Multipart_MT_inproc_shmem)
{
RunMultiThreadedMultipart("shmem", "inproc://test");
RunMultiThreadedMultipart("shmem", "inproc://test_1");
}
#ifdef BUILD_NANOMSG_TRANSPORT
TEST(PushPull, Multipart_MT_inproc_nanomsg)
{
RunMultiThreadedMultipart("nanomsg", "inproc://test");
RunMultiThreadedMultipart("nanomsg", "inproc://test_1");
}
#endif /* BUILD_NANOMSG_TRANSPORT */
TEST(PushPull, Multipart_MT_ipc_zeromq)
{
RunMultiThreadedMultipart("zeromq", "ipc://test_Multipart_MT_ipc_zeromq");
RunMultiThreadedMultipart("zeromq", "ipc://test_Multipart_MT_ipc_zeromq_1");
}
TEST(PushPull, Multipart_MT_ipc_shmem)
{
RunMultiThreadedMultipart("shmem", "ipc://test_Multipart_MT_ipc_shmem");
RunMultiThreadedMultipart("shmem", "ipc://test_Multipart_MT_ipc_shmem_1");
}
#ifdef BUILD_NANOMSG_TRANSPORT
TEST(PushPull, Multipart_MT_ipc_nanomsg)
{
RunMultiThreadedMultipart("nanomsg", "ipc://test_Multipart_MT_ipc_nanomsg");
RunMultiThreadedMultipart("nanomsg", "ipc://test_Multipart_MT_ipc_nanomsg_1");
}
#endif /* BUILD_NANOMSG_TRANSPORT */