Compare commits

..

16 Commits

Author SHA1 Message Date
Alexey Rybalchenko
2b3e38d9a4 shmmonitor: non-interactive mode checks and quits 2021-02-10 10:36:08 +01:00
Alexey Rybalchenko
c6b13cd3a1 Fix shmem::Message::SetUsedSize(0) 2021-01-25 13:46:40 +01:00
Alexey Rybalchenko
c5487a11ed Remove custom implementation for enum hashing 2021-01-25 13:46:40 +01:00
Alexey Rybalchenko
4a09154a91 17-ify namespaces 2021-01-25 13:46:40 +01:00
Alexey Rybalchenko
d9a5e82160 Cleanup tools includes 2021-01-25 13:46:40 +01:00
Alexey Rybalchenko
751c53171c Replace tools::make_unique with std::make_unique 2021-01-25 13:46:40 +01:00
Alexey Rybalchenko
6815c9c172 zmq: implement alignment 2021-01-13 12:36:32 +01:00
Alexey Rybalchenko
02a3980343 Remove useless parameter from implementation 2021-01-13 12:36:32 +01:00
Alexey Rybalchenko
38b34785e0 format & remove unused variable 2021-01-13 12:36:32 +01:00
Alexey Rybalchenko
be94ceb7a7 zmq: simplify SetUsedSize implementation 2021-01-13 12:36:32 +01:00
Alexey Rybalchenko
afadbb53e4 zmq: correct accounting for msg size > 2GB 2021-01-13 12:36:32 +01:00
Alexey Rybalchenko
749d28a3b5 DeviceRunner: check FAIRMQ_SEVERITY env var for severity 2021-01-13 12:36:32 +01:00
Alexey Rybalchenko
29f45fa77d Rename TransferResult to TransferCode 2021-01-13 12:36:32 +01:00
Alexey Rybalchenko
ea746b17d0 Remove deprecated methods 2021-01-13 12:36:32 +01:00
Alexey Rybalchenko
636846fcdb Bump C++ standard requirement to C++17 2021-01-13 12:36:32 +01:00
Alexey Rybalchenko
f46d446d52 shm: do mlock/zeroing under shmem lock 2020-12-04 13:27:45 +01:00
150 changed files with 774 additions and 1339 deletions

View File

@@ -18,11 +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 OR BUILD_PMIX_PLUGIN)
set(PROJECT_MIN_CXX_STANDARD 14)
else()
set(PROJECT_MIN_CXX_STANDARD 11)
endif()
set(PROJECT_MIN_CXX_STANDARD 17)
set_fairmq_defaults()
@@ -262,7 +258,7 @@ install_cmake_package()
# Summary ######################################################################
message(STATUS " ")
message(STATUS " ${Cyan}CXX STANDARD${CR} ${BGreen}C++${CMAKE_CXX_STANDARD}${CR} (>= C++${PROJECT_MIN_CXX_STANDARD}, change with ${BMagenta}-DCMAKE_CXX_STANDARD=17${CR})")
message(STATUS " ${Cyan}CXX STANDARD${CR} ${BGreen}C++${CMAKE_CXX_STANDARD}${CR} (>= C++${PROJECT_MIN_CXX_STANDARD}, change with ${BMagenta}-DCMAKE_CXX_STANDARD=20${CR})")
if(CMAKE_CXX_FLAGS)
message(STATUS " ")
message(STATUS " ${Cyan}GLOBAL CXX FLAGS${CR} ${BGreen}${CMAKE_CXX_FLAGS}${CR}")
@@ -365,9 +361,9 @@ else()
endif()
message(STATUS " ${BWhite}tests${CR} ${tests_summary}")
if(BUILD_OFI_TRANSPORT)
set(ofi_summary "${BGreen}YES${CR} EXPERIMENTAL (requires C++14) (disable with ${BMagenta}-DBUILD_OFI_TRANSPORT=OFF${CR})")
set(ofi_summary "${BGreen}YES${CR} EXPERIMENTAL (disable with ${BMagenta}-DBUILD_OFI_TRANSPORT=OFF${CR})")
else()
set(ofi_summary "${BRed} NO${CR} EXPERIMENTAL (requires C++14) (default, enable with ${BMagenta}-DBUILD_OFI_TRANSPORT=ON${CR})")
set(ofi_summary "${BRed} NO${CR} EXPERIMENTAL (default, enable with ${BMagenta}-DBUILD_OFI_TRANSPORT=ON${CR})")
endif()
message(STATUS " ${BWhite}ofi_transport${CR} ${ofi_summary}")
if(BUILD_DDS_PLUGIN)
@@ -377,9 +373,9 @@ else()
endif()
message(STATUS " ${BWhite}dds_plugin${CR} ${dds_summary}")
if(BUILD_PMIX_PLUGIN)
set(pmix_summary "${BGreen}YES${CR} EXPERIMENTAL (requires C++14) (disable with ${BMagenta}-DBUILD_PMIX_PLUGIN=OFF${CR})")
set(pmix_summary "${BGreen}YES${CR} EXPERIMENTAL (disable with ${BMagenta}-DBUILD_PMIX_PLUGIN=OFF${CR})")
else()
set(pmix_summary "${BRed} NO${CR} EXPERIMENTAL (requires C++14) (default, enable with ${BMagenta}-DBUILD_PMIX_PLUGIN=ON${CR})")
set(pmix_summary "${BRed} NO${CR} EXPERIMENTAL (default, enable with ${BMagenta}-DBUILD_PMIX_PLUGIN=ON${CR})")
endif()
message(STATUS " ${BWhite}pmix_plugin${CR} ${pmix_summary}")
if(BUILD_EXAMPLES)
@@ -395,9 +391,9 @@ else()
endif()
message(STATUS " ${BWhite}docs${CR} ${docs_summary}")
if(BUILD_SDK)
set(sdk_summary "${BGreen}YES${CR} EXPERIMENTAL (required C++14) (disable with ${BMagenta}-DBUILD_SDK=OFF${CR})")
set(sdk_summary "${BGreen}YES${CR} EXPERIMENTAL (disable with ${BMagenta}-DBUILD_SDK=OFF${CR})")
else()
set(sdk_summary "${BRed} NO${CR} EXPERIMENTAL (required C++14) (default, enable with ${BMagenta}-DBUILD_SDK=ON${CR})")
set(sdk_summary "${BRed} NO${CR} EXPERIMENTAL (default, enable with ${BMagenta}-DBUILD_SDK=ON${CR})")
endif()
message(STATUS " ${BWhite}sdk${CR} ${sdk_summary}")
if(BUILD_SDK_COMMANDS)
@@ -432,9 +428,9 @@ message(STATUS " ")
message(STATUS " ${Cyan}RUN STATIC ANALYSIS ${static_ana_summary}")
message(STATUS " ")
if(FAIRMQ_DEBUG_MODE)
message(STATUS " ${Cyan}DEBUG_MODE${CR} ${BGreen}${FAIRMQ_DEBUG_MODE}${CR} (disable with ${BMagenta}-DFAIRMQ_DEBUG_MODE=OFF${CR})")
message(STATUS " ${Cyan}DEBUG MODE${CR} ${BGreen}${FAIRMQ_DEBUG_MODE}${CR} (disable with ${BMagenta}-DFAIRMQ_DEBUG_MODE=OFF${CR})")
else()
message(STATUS " ${Cyan}DEBUG_MODE${CR} ${BRed}${FAIRMQ_DEBUG_MODE}${CR} (enable with ${BMagenta}-DFAIRMQ_DEBUG_MODE=ON${CR})")
message(STATUS " ${Cyan}DEBUG MODE${CR} ${BRed}${FAIRMQ_DEBUG_MODE}${CR} (enable with ${BMagenta}-DFAIRMQ_DEBUG_MODE=ON${CR})")
endif()
message(STATUS " ")
################################################################################

View File

@@ -56,6 +56,10 @@ bool DeviceRunner::HandleGeneralOptions(const fair::mq::ProgOptions& config, boo
fair::Logger::SetConsoleSeverity("nolog");
} else {
fair::Logger::SetConsoleColor(color);
auto envFairMQSeverity = getenv("FAIRMQ_SEVERITY");
if (envFairMQSeverity) {
severity = envFairMQSeverity;
}
if (severity != "") {
fair::Logger::SetConsoleSeverity(severity);
}

View File

@@ -19,8 +19,8 @@
#include <string>
#include <vector>
namespace fair {
namespace mq {
namespace fair::mq
{
/**
* @class DeviceRunner DeviceRunner.h <fairmq/DeviceRunner.h>
@@ -88,7 +88,6 @@ struct ModifyRawCmdLineArgs : Event<DeviceRunner&> {};
struct InstantiateDevice : Event<DeviceRunner&> {};
} /* namespace hooks */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq
#endif /* FAIR_MQ_DEVICERUNNER_H */

View File

@@ -21,9 +21,7 @@
#include <boost/functional/hash.hpp>
#include <boost/signals2.hpp>
namespace fair
{
namespace mq
namespace fair::mq
{
// Inherit from this base event type to create custom event types
@@ -137,7 +135,6 @@ class EventManager
}
}; /* class EventManager */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq
#endif /* FAIR_MQ_EVENTMANAGER_H */

View File

@@ -256,7 +256,7 @@ class FairMQChannel
/// Sends a message to the socket queue.
/// @param msg Constant reference of unique_ptr to a FairMQMessage
/// @param sndTimeoutInMs send timeout in ms. -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot send)
/// @return Number of bytes that have been queued, TransferResult::timeout if timed out, TransferResult::error if there was an error, TransferResult::interrupted if interrupted (e.g. by requested state change)
/// @return Number of bytes that have been queued, TransferCode::timeout if timed out, TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by requested state change)
int64_t Send(FairMQMessagePtr& msg, int sndTimeoutInMs = -1)
{
CheckSendCompatibility(msg);
@@ -266,7 +266,7 @@ class FairMQChannel
/// Receives a message from the socket queue.
/// @param msg Constant reference of unique_ptr to a FairMQMessage
/// @param rcvTimeoutInMs receive timeout in ms. -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot receive)
/// @return Number of bytes that have been received, TransferResult::timeout if timed out, TransferResult::error if there was an error, TransferResult::interrupted if interrupted (e.g. by requested state change)
/// @return Number of bytes that have been received, TransferCode::timeout if timed out, TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by requested state change)
int64_t Receive(FairMQMessagePtr& msg, int rcvTimeoutInMs = -1)
{
CheckReceiveCompatibility(msg);
@@ -276,7 +276,7 @@ class FairMQChannel
/// Send a vector of messages
/// @param msgVec message vector reference
/// @param sndTimeoutInMs send timeout in ms. -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot send)
/// @return Number of bytes that have been queued, TransferResult::timeout if timed out, TransferResult::error if there was an error, TransferResult::interrupted if interrupted (e.g. by requested state change)
/// @return Number of bytes that have been queued, TransferCode::timeout if timed out, TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by requested state change)
int64_t Send(std::vector<FairMQMessagePtr>& msgVec, int sndTimeoutInMs = -1)
{
CheckSendCompatibility(msgVec);
@@ -286,7 +286,7 @@ class FairMQChannel
/// Receive a vector of messages
/// @param msgVec message vector reference
/// @param rcvTimeoutInMs receive timeout in ms. -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot receive)
/// @return Number of bytes that have been received, TransferResult::timeout if timed out, TransferResult::error if there was an error, TransferResult::interrupted if interrupted (e.g. by requested state change)
/// @return Number of bytes that have been received, TransferCode::timeout if timed out, TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by requested state change)
int64_t Receive(std::vector<FairMQMessagePtr>& msgVec, int rcvTimeoutInMs = -1)
{
CheckReceiveCompatibility(msgVec);
@@ -296,7 +296,7 @@ class FairMQChannel
/// Send FairMQParts
/// @param parts FairMQParts reference
/// @param sndTimeoutInMs send timeout in ms. -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot send)
/// @return Number of bytes that have been queued, TransferResult::timeout if timed out, TransferResult::error if there was an error, TransferResult::interrupted if interrupted (e.g. by requested state change)
/// @return Number of bytes that have been queued, TransferCode::timeout if timed out, TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by requested state change)
int64_t Send(FairMQParts& parts, int sndTimeoutInMs = -1)
{
return Send(parts.fParts, sndTimeoutInMs);
@@ -305,7 +305,7 @@ class FairMQChannel
/// Receive FairMQParts
/// @param parts FairMQParts reference
/// @param rcvTimeoutInMs receive timeout in ms. -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot receive)
/// @return Number of bytes that have been received, TransferResult::timeout if timed out, TransferResult::error if there was an error, TransferResult::interrupted if interrupted (e.g. by requested state change)
/// @return Number of bytes that have been received, TransferCode::timeout if timed out, TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by requested state change)
int64_t Receive(FairMQParts& parts, int rcvTimeoutInMs = -1)
{
return Receive(parts.fParts, rcvTimeoutInMs);

View File

@@ -24,34 +24,6 @@
using namespace std;
using namespace fair::mq;
static map<Transition, State> backwardsCompatibilityWaitForEndOfStateHelper =
{
{ Transition::InitDevice, State::InitializingDevice },
{ Transition::CompleteInit, State::Initialized },
{ Transition::Bind, State::Bound },
{ Transition::Connect, State::DeviceReady },
{ Transition::InitTask, State::Ready },
{ Transition::Run, State::Ready },
{ Transition::Stop, State::Ready },
{ Transition::ResetTask, State::DeviceReady },
{ Transition::ResetDevice, State::Idle }
};
static map<int, Transition> backwardsCompatibilityChangeStateHelper =
{
{ FairMQDevice::Event::INIT_DEVICE, Transition::InitDevice },
{ FairMQDevice::Event::internal_DEVICE_READY, Transition::Auto },
{ FairMQDevice::Event::INIT_TASK, Transition::InitTask },
{ FairMQDevice::Event::internal_READY, Transition::Auto },
{ FairMQDevice::Event::RUN, Transition::Run },
{ FairMQDevice::Event::STOP, Transition::Stop },
{ FairMQDevice::Event::RESET_TASK, Transition::ResetTask },
{ FairMQDevice::Event::RESET_DEVICE, Transition::ResetDevice },
{ FairMQDevice::Event::internal_IDLE, Transition::Auto },
{ FairMQDevice::Event::END, Transition::End },
{ FairMQDevice::Event::ERROR_FOUND, Transition::ErrorFound }
};
constexpr const char* FairMQDevice::DefaultId;
constexpr int FairMQDevice::DefaultIOThreads;
constexpr const char* FairMQDevice::DefaultTransportName;
@@ -103,7 +75,7 @@ FairMQDevice::FairMQDevice(ProgOptions* config, const tools::Version version)
: fTransportFactory(nullptr)
, fTransports()
, fChannels()
, fInternalConfig(config ? nullptr : tools::make_unique<ProgOptions>())
, fInternalConfig(config ? nullptr : make_unique<ProgOptions>())
, fConfig(config ? config : fInternalConfig.get())
, fId(DefaultId)
, fDefaultTransportType(DefaultTransportType)
@@ -244,16 +216,6 @@ void FairMQDevice::TransitionTo(const fair::mq::State s)
}
}
bool FairMQDevice::ChangeState(const int transition)
{
return ChangeState(backwardsCompatibilityChangeStateHelper.at(transition));
}
void FairMQDevice::WaitForEndOfState(Transition transition)
{
WaitForState(backwardsCompatibilityWaitForEndOfStateHelper.at(transition));
}
void FairMQDevice::InitWrapper()
{
// run initialization once CompleteInit transition is requested

View File

@@ -41,51 +41,16 @@ using FairMQChannelMap = std::unordered_map<std::string, std::vector<FairMQChann
using InputMsgCallback = std::function<bool(FairMQMessagePtr&, int)>;
using InputMultipartCallback = std::function<bool(FairMQParts&, int)>;
namespace fair
{
namespace mq
namespace fair::mq
{
struct OngoingTransition : std::runtime_error { using std::runtime_error::runtime_error; };
}
}
class FairMQDevice
{
friend class FairMQChannel;
public:
// backwards-compatibility enum for old state machine interface, todo: delete this
enum Event
{
INIT_DEVICE,
internal_DEVICE_READY,
INIT_TASK,
internal_READY,
RUN,
STOP,
RESET_TASK,
RESET_DEVICE,
internal_IDLE,
END,
ERROR_FOUND
};
// backwards-compatibility enum for old state machine interface, todo: delete this
enum State
{
OK,
Error,
IDLE,
INITIALIZING_DEVICE,
DEVICE_READY,
INITIALIZING_TASK,
READY,
RUNNING,
RESETTING_TASK,
RESETTING_DEVICE,
EXITING
};
/// Default constructor
FairMQDevice();
/// Constructor with external fair::mq::ProgOptions
@@ -128,7 +93,7 @@ class FairMQDevice
/// @param chan channel name
/// @param i channel index
/// @param sndTimeoutInMs send timeout in ms, -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot send)
/// @return Number of bytes that have been queued, TransferResult::timeout if timed out, TransferResult::error if there was an error, TransferResult::interrupted if interrupted (e.g. by requested state change)
/// @return Number of bytes that have been queued, TransferCode::timeout if timed out, TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by requested state change)
int64_t Send(FairMQMessagePtr& msg, const std::string& channel, const int index = 0, int sndTimeoutInMs = -1)
{
return GetChannel(channel, index).Send(msg, sndTimeoutInMs);
@@ -139,7 +104,7 @@ class FairMQDevice
/// @param chan channel name
/// @param i channel index
/// @param rcvTimeoutInMs receive timeout in ms, -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot receive)
/// @return Number of bytes that have been received, TransferResult::timeout if timed out, TransferResult::error if there was an error, TransferResult::interrupted if interrupted (e.g. by requested state change)
/// @return Number of bytes that have been received, TransferCode::timeout if timed out, TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by requested state change)
int64_t Receive(FairMQMessagePtr& msg, const std::string& channel, const int index = 0, int rcvTimeoutInMs = -1)
{
return GetChannel(channel, index).Receive(msg, rcvTimeoutInMs);
@@ -150,7 +115,7 @@ class FairMQDevice
/// @param chan channel name
/// @param i channel index
/// @param sndTimeoutInMs send timeout in ms, -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot send)
/// @return Number of bytes that have been queued, TransferResult::timeout if timed out, TransferResult::error if there was an error, TransferResult::interrupted if interrupted (e.g. by requested state change)
/// @return Number of bytes that have been queued, TransferCode::timeout if timed out, TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by requested state change)
int64_t Send(FairMQParts& parts, const std::string& channel, const int index = 0, int sndTimeoutInMs = -1)
{
return GetChannel(channel, index).Send(parts.fParts, sndTimeoutInMs);
@@ -161,7 +126,7 @@ class FairMQDevice
/// @param chan channel name
/// @param i channel index
/// @param rcvTimeoutInMs receive timeout in ms, -1 will wait forever (or until interrupt (e.g. via state change)), 0 will not wait (return immediately if cannot receive)
/// @return Number of bytes that have been received, TransferResult::timeout if timed out, TransferResult::error if there was an error, TransferResult::interrupted if interrupted (e.g. by requested state change)
/// @return Number of bytes that have been received, TransferCode::timeout if timed out, TransferCode::error if there was an error, TransferCode::interrupted if interrupted (e.g. by requested state change)
int64_t Receive(FairMQParts& parts, const std::string& channel, const int index = 0, int rcvTimeoutInMs = -1)
{
return GetChannel(channel, index).Receive(parts.fParts, rcvTimeoutInMs);
@@ -449,8 +414,6 @@ class FairMQDevice
/// Called in the RUNNING state once after executing the Run()/ConditionalRun() method
virtual void PostRun() {}
virtual void Pause() __attribute__((deprecated("PAUSE state is removed. This method is never called. To pause Run, go to READY with STOP transition and back to RUNNING with RUN to resume."))) {}
/// Resets the user task (to be overloaded in child classes)
virtual void ResetTask() {}
@@ -461,11 +424,6 @@ class FairMQDevice
bool ChangeState(const fair::mq::Transition transition) { return fStateMachine.ChangeState(transition); }
bool ChangeState(const std::string& transition) { return fStateMachine.ChangeState(fair::mq::GetTransition(transition)); }
bool ChangeState(const int transition) __attribute__((deprecated("Use ChangeState(const fair::mq::Transition transition).")));
void WaitForEndOfState(const fair::mq::Transition transition) __attribute__((deprecated("Use WaitForState(fair::mq::State expectedState).")));
void WaitForEndOfState(const std::string& transition) __attribute__((deprecated("Use WaitForState(fair::mq::State expectedState)."))) { WaitForState(transition); }
fair::mq::State WaitForNextState() { return fStateQueue.WaitForNext(); }
void WaitForState(fair::mq::State state) { fStateQueue.WaitForState(state); }
void WaitForState(const std::string& state) { WaitForState(fair::mq::GetState(state)); }
@@ -478,9 +436,6 @@ class FairMQDevice
void SubscribeToNewTransition(const std::string& key, std::function<void(const fair::mq::Transition)> callback) { fStateMachine.SubscribeToNewTransition(key, callback); }
void UnsubscribeFromNewTransition(const std::string& key) { fStateMachine.UnsubscribeFromNewTransition(key); }
bool CheckCurrentState(const int /* state */) const __attribute__((deprecated("Use NewStatePending()."))) { return !fStateMachine.NewStatePending(); }
bool CheckCurrentState(const std::string& /* state */) const __attribute__((deprecated("Use NewStatePending()."))) { return !fStateMachine.NewStatePending(); }
/// Returns true is a new state has been requested, signaling the current handler to stop.
bool NewStatePending() const { return fStateMachine.NewStatePending(); }

View File

@@ -18,9 +18,7 @@
using fairmq_free_fn = void(void* data, void* hint);
class FairMQTransportFactory;
namespace fair
{
namespace mq
namespace fair::mq
{
struct Alignment
@@ -29,8 +27,7 @@ struct Alignment
explicit operator size_t() const { return alignment; }
};
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq
class FairMQMessage
{
@@ -39,7 +36,9 @@ class FairMQMessage
FairMQMessage(FairMQTransportFactory* factory) : fTransport(factory) {}
virtual void Rebuild() = 0;
virtual void Rebuild(fair::mq::Alignment alignment) = 0;
virtual void Rebuild(const size_t size) = 0;
virtual void Rebuild(const size_t size, fair::mq::Alignment alignment) = 0;
virtual void Rebuild(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr) = 0;
virtual void* GetData() const = 0;
@@ -61,9 +60,7 @@ class FairMQMessage
using FairMQMessagePtr = std::unique_ptr<FairMQMessage>;
namespace fair
{
namespace mq
namespace fair::mq
{
using Message = FairMQMessage;
@@ -71,7 +68,6 @@ using MessagePtr = FairMQMessagePtr;
struct MessageError : std::runtime_error { using std::runtime_error::runtime_error; };
struct MessageBadAlloc : std::runtime_error { using std::runtime_error::runtime_error; };
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq
#endif /* FAIRMQMESSAGE_H_ */

View File

@@ -26,16 +26,13 @@ class FairMQPoller
using FairMQPollerPtr = std::unique_ptr<FairMQPoller>;
namespace fair
{
namespace mq
namespace fair::mq
{
using Poller = FairMQPoller;
using PollerPtr = FairMQPollerPtr;
struct PollerError : std::runtime_error { using std::runtime_error::runtime_error; };
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq
#endif /* FAIRMQPOLLER_H_ */

View File

@@ -19,20 +19,18 @@
class FairMQTransportFactory;
namespace fair
{
namespace mq
namespace fair::mq
{
enum class TransferResult : int
enum class TransferCode : int
{
success = 0,
error = -1,
timeout = -2,
interrupted = -3
};
} // namespace mq
} // namespace fair
} // namespace fair::mq
class FairMQSocket
{
@@ -86,16 +84,13 @@ class FairMQSocket
using FairMQSocketPtr = std::unique_ptr<FairMQSocket>;
namespace fair
{
namespace mq
namespace fair::mq
{
using Socket = FairMQSocket;
using SocketPtr = FairMQSocketPtr;
struct SocketError : std::runtime_error { using std::runtime_error::runtime_error; };
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq
#endif /* FAIRMQSOCKET_H_ */

View File

@@ -13,7 +13,8 @@
#include <fairmq/ofi/TransportFactory.h>
#endif
#include <FairMQLogger.h>
#include <fairmq/Tools.h>
#include <fairmq/tools/Unique.h>
#include <fairmq/tools/Strings.h>
#include <fairlogger/Logger.h>

View File

@@ -24,7 +24,7 @@
#include <cstddef> // size_t
class FairMQChannel;
namespace fair { namespace mq { class ProgOptions; } }
namespace fair::mq { class ProgOptions; }
class FairMQTransportFactory
{
@@ -171,15 +171,12 @@ class FairMQTransportFactory
}
};
namespace fair
{
namespace mq
namespace fair::mq
{
using TransportFactory = FairMQTransportFactory;
struct TransportFactoryError : std::runtime_error { using std::runtime_error::runtime_error; };
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq
#endif /* FAIRMQTRANSPORTFACTORY_H_ */

View File

@@ -104,9 +104,7 @@ inline std::ostream& operator<<(std::ostream& os, const FairMQRegionEvent& event
}
}
namespace fair
{
namespace mq
namespace fair::mq
{
using RegionCallback = FairMQRegionCallback;
@@ -118,7 +116,6 @@ using RegionBlock = FairMQRegionBlock;
using UnmanagedRegion = FairMQUnmanagedRegion;
using UnmanagedRegionPtr = FairMQUnmanagedRegionPtr;
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq
#endif /* FAIRMQUNMANAGEDREGION_H_ */

View File

@@ -15,7 +15,6 @@
#include "JSONParser.h"
#include "FairMQChannel.h"
#include <fairmq/PropertyOutput.h>
#include <fairmq/tools/CppSTL.h>
#include <fairmq/tools/Strings.h>
#include <fairlogger/Logger.h>
@@ -31,9 +30,7 @@ using namespace fair::mq;
using namespace fair::mq::tools;
using namespace boost::property_tree;
namespace fair
{
namespace mq
namespace fair::mq
{
fair::mq::Properties PtreeParser(const ptree& pt, const string& id)
@@ -186,5 +183,4 @@ void SubChannelParser(const ptree& channelTree, fair::mq::Properties& properties
}
} // helper namespace
} // namespace mq
} // namespace fair
} // namespace fair::mq

View File

@@ -21,9 +21,7 @@
#include <stdexcept>
#include <string>
namespace fair
{
namespace mq
namespace fair::mq
{
struct ParserError : std::runtime_error { using std::runtime_error::runtime_error; };
@@ -41,7 +39,6 @@ void SubChannelParser(const boost::property_tree::ptree& tree, fair::mq::Propert
} // helper namespace
} // namespace mq
} // namespace fair
} // namespace fair::mq
#endif /* FAIR_MQ_JSONPARSER_H */

View File

@@ -15,8 +15,8 @@
#include <fairmq/FairMQTransportFactory.h>
#include <fairmq/MemoryResources.h>
namespace fair {
namespace mq {
namespace fair::mq
{
using BytePmrAllocator = pmr::polymorphic_allocator<fair::mq::byte>;
@@ -62,5 +62,4 @@ FairMQMessagePtr getMessage(ContainerT &&container_, FairMQMemoryResource *targe
return message;
}
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq

View File

@@ -26,8 +26,8 @@ class FairMQTransportFactory;
#include <stdexcept>
#include <utility>
namespace fair {
namespace mq {
namespace fair::mq
{
using byte = unsigned char;
namespace pmr = boost::container::pmr;
@@ -107,7 +107,6 @@ class ChannelResource : public FairMQMemoryResource
};
};
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq
#endif /* FAIR_MQ_MEMORY_RESOURCES_H */

View File

@@ -9,7 +9,6 @@
#ifndef FAIR_MQ_PLUGIN_H
#define FAIR_MQ_PLUGIN_H
#include <fairmq/tools/CppSTL.h>
#include <fairmq/tools/Version.h>
#include <fairmq/PluginServices.h>
@@ -25,9 +24,7 @@
#include <tuple>
#include <utility>
namespace fair
{
namespace mq
namespace fair::mq
{
/**
@@ -134,13 +131,12 @@ class Plugin
PluginServices* fPluginServices;
}; /* class Plugin */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq
#define REGISTER_FAIRMQ_PLUGIN(KLASS, NAME, VERSION, MAINTAINER, HOMEPAGE, PROGOPTIONS) \
static auto Make_##NAME##_Plugin(fair::mq::PluginServices* pluginServices) -> std::unique_ptr<fair::mq::Plugin> \
{ \
return fair::mq::tools::make_unique<KLASS>(std::string{#NAME}, VERSION, std::string{MAINTAINER}, std::string{HOMEPAGE}, pluginServices); \
return std::make_unique<KLASS>(std::string{#NAME}, VERSION, std::string{MAINTAINER}, std::string{HOMEPAGE}, pluginServices); \
} \
BOOST_DLL_ALIAS(Make_##NAME##_Plugin, make_##NAME##_plugin) \
BOOST_DLL_ALIAS(PROGOPTIONS, get_##NAME##_plugin_progoptions)

View File

@@ -11,7 +11,6 @@
#include <fairmq/Plugin.h>
#include <fairmq/PluginServices.h>
#include <fairmq/tools/CppSTL.h>
#include <fairmq/tools/Strings.h>
#define BOOST_FILESYSTEM_VERSION 3
@@ -31,9 +30,7 @@
#include <vector>
#include <utility> // forward
namespace fair
{
namespace mq
namespace fair::mq
{
/**
@@ -80,7 +77,7 @@ class PluginManager
auto ForEachPluginProgOptions(std::function<void (boost::program_options::options_description)> func) const -> void { for(const auto& pair : fPluginProgOptions) { func(pair.second); } }
template<typename... Args>
auto EmplacePluginServices(Args&&... args) -> void { fPluginServices = fair::mq::tools::make_unique<PluginServices>(std::forward<Args>(args)...); }
auto EmplacePluginServices(Args&&... args) -> void { fPluginServices = std::make_unique<PluginServices>(std::forward<Args>(args)...); }
auto WaitForPluginsToReleaseDeviceControl() -> void { fPluginServices->WaitForReleaseDeviceControl(); }
@@ -126,7 +123,6 @@ class PluginManager
std::map<std::string, boost::program_options::options_description> fPluginProgOptions;
}; /* class PluginManager */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq
#endif /* FAIR_MQ_PLUGINMANAGER_H */

View File

@@ -26,9 +26,7 @@
#include <unordered_map>
#include <vector>
namespace fair
{
namespace mq
namespace fair::mq
{
/**
@@ -281,7 +279,6 @@ class PluginServices
std::condition_variable fReleaseDeviceControlCondition;
}; /* class PluginServices */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq
#endif /* FAIR_MQ_PLUGINSERVICES_H */

View File

@@ -39,9 +39,7 @@ struct ValInfo
string origin;
};
namespace fair
{
namespace mq
namespace fair::mq
{
ValInfo ConvertVarValToValInfo(const po::variable_value& v)
@@ -449,6 +447,4 @@ void ProgOptions::PrintOptionsRaw() const
}
}
} // namespace mq
} // namespace fair
} // namespace fair::mq

View File

@@ -26,9 +26,7 @@
#include <vector>
#include <stdexcept>
namespace fair
{
namespace mq
namespace fair::mq
{
struct PropertyNotFoundError : std::runtime_error { using std::runtime_error::runtime_error; };
@@ -270,7 +268,6 @@ class ProgOptions
mutable std::mutex fMtx;
};
} // namespace mq
} // namespace fair
} // namespace fair::mq
#endif /* FAIR_MQ_PROGOPTIONS_H */

View File

@@ -9,13 +9,10 @@
#ifndef FAIR_MQ_PROGOPTIONSFWD_H
#define FAIR_MQ_PROGOPTIONSFWD_H
namespace fair
{
namespace mq
namespace fair::mq
{
class ProgOptions;
}
}
using FairMQProgOptions = fair::mq::ProgOptions;

View File

@@ -7,17 +7,13 @@
********************************************************************************/
#include <fairmq/Properties.h>
#include <fairmq/Tools.h>
#include <boost/filesystem.hpp>
using namespace std;
using namespace fair::mq::tools;
using boost::any_cast;
namespace fair
{
namespace mq
namespace fair::mq
{
template<class T>
@@ -145,5 +141,4 @@ unordered_map<type_index, void(*)(const EventManager&, const string&, const Prop
{ type_index(typeid(vector<boost::filesystem::path>)), [](const EventManager& em, const string& k, const Property& p) { em.Emit<PropertyChange, vector<boost::filesystem::path>>(k, any_cast<vector<boost::filesystem::path>>(p)); } },
};
} // namespace mq
} // namespace fair
} // namespace fair::mq

View File

@@ -22,9 +22,7 @@
#include <typeinfo>
#include <utility> // pair
namespace fair
{
namespace mq
namespace fair::mq
{
using Property = boost::any;
@@ -72,7 +70,6 @@ class PropertyHelper
static std::unordered_map<std::type_index, std::function<std::pair<std::string, std::string>(const Property&)>> fTypeInfos;
};
}
}
#endif /* FAIR_MQ_PROPERTIES_H */

View File

@@ -35,9 +35,7 @@ using namespace boost::msm::front;
using namespace boost::msm::back;
namespace bmpl = boost::mpl;
namespace fair
{
namespace mq
namespace fair::mq
{
namespace fsm
{
@@ -239,8 +237,7 @@ struct Machine_ : public state_machine_def<Machine_>
using FairMQFSM = state_machine<Machine_>;
} // namespace fsm
} // namespace mq
} // namespace fair
} // namespace fair::mq
using namespace fair::mq::fsm;
using namespace fair::mq;

View File

@@ -16,9 +16,7 @@
#include <functional>
#include <stdexcept>
namespace fair
{
namespace mq
namespace fair::mq
{
class StateMachine
@@ -56,7 +54,6 @@ class StateMachine
std::shared_ptr<void> fFsm;
};
} // namespace mq
} // namespace fair
} // namespace fair::mq
#endif /* FAIRMQSTATEMACHINE_H_ */

View File

@@ -17,9 +17,7 @@
#include <utility> // pair
#include <condition_variable>
namespace fair
{
namespace mq
namespace fair::mq
{
class StateQueue
@@ -88,7 +86,6 @@ class StateQueue
std::condition_variable fCV;
};
} // namespace mq
} // namespace fair
} // namespace fair::mq
#endif /* FAIRMQSTATEQUEUE_H_ */

View File

@@ -13,9 +13,7 @@
using namespace std;
namespace fair
{
namespace mq
namespace fair::mq
{
array<string, 16> stateNames =
@@ -114,5 +112,4 @@ Transition GetTransition(const string& transition)
return transitions.at(transition);
}
} // namespace mq
} // namespace fair
} // namespace fair::mq

View File

@@ -13,9 +13,7 @@
#include <ostream>
#include <stdexcept>
namespace fair
{
namespace mq
namespace fair::mq
{
enum class State : int
@@ -64,7 +62,6 @@ struct DeviceErrorState : std::runtime_error { using std::runtime_error::runtime
inline std::ostream& operator<<(std::ostream& os, const State& state) { return os << GetStateName(state); }
inline std::ostream& operator<<(std::ostream& os, const Transition& transition) { return os << GetTransitionName(transition); }
} // namespace mq
} // namespace fair
} // namespace fair::mq
#endif /* FAIRMQSTATES_H_ */

View File

@@ -23,9 +23,7 @@
using boost::property_tree::ptree;
using namespace std;
namespace fair
{
namespace mq
namespace fair::mq
{
enum channelOptionKeyIds
@@ -120,4 +118,3 @@ Properties SuboptParser(const vector<string>& channelConfig, const string& devic
}
}
}

View File

@@ -19,9 +19,7 @@
#include <vector>
#include <string>
namespace fair
{
namespace mq
namespace fair::mq
{
/**
@@ -43,7 +41,6 @@ namespace mq
Properties SuboptParser(const std::vector<std::string>& channelConfig, const std::string& deviceId);
}
}
#endif /* FAIR_MQ_SUBOPTPARSER_H */

View File

@@ -9,7 +9,6 @@
#ifndef FAIR_MQ_TRANSPORTS_H
#define FAIR_MQ_TRANSPORTS_H
#include <fairmq/tools/CppSTL.h>
#include <fairmq/tools/Strings.h>
#include <memory>
@@ -17,9 +16,7 @@
#include <string>
#include <unordered_map>
namespace fair
{
namespace mq
namespace fair::mq
{
enum class Transport
@@ -32,20 +29,9 @@ enum class Transport
struct TransportError : std::runtime_error { using std::runtime_error::runtime_error; };
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq
namespace std
{
template<>
struct hash<fair::mq::Transport> : fair::mq::tools::HashEnum<fair::mq::Transport> {};
} /* namespace std */
namespace fair
{
namespace mq
namespace fair::mq
{
static std::unordered_map<std::string, Transport> TransportTypes {
@@ -74,7 +60,6 @@ try {
throw TransportError(tools::ToString("Unknown transport provided: ", transport));
}
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq
#endif /* FAIR_MQ_TRANSPORTS_H */

View File

@@ -13,7 +13,6 @@
#include "FairMQDevice.h"
#include "tools/RateLimit.h"
#include <atomic>
#include <chrono>
#include <cstddef> // size_t
#include <cstdint> // uint64_t
@@ -106,15 +105,12 @@ class FairMQBenchmarkSampler : public FairMQDevice
LOG(info) << "Done " << fNumIterations << " iterations in " << std::chrono::duration<double, std::milli>(tEnd - tStart).count() << "ms.";
}
virtual ~FairMQBenchmarkSampler() {}
protected:
bool fMultipart;
bool fMemSet;
size_t fNumParts;
size_t fMsgSize;
size_t fMsgAlignment;
std::atomic<int> fMsgCounter;
float fMsgRate;
uint64_t fNumIterations;
uint64_t fMaxIterations;

View File

@@ -7,7 +7,7 @@
********************************************************************************/
#include <fairmq/ofi/Context.h>
#include <fairmq/Tools.h>
#include <fairmq/tools/Strings.h>
#include <FairMQLogger.h>
#include <asiofi/version.hpp>
@@ -22,11 +22,7 @@
#include <string.h>
#include <sys/socket.h>
namespace fair
{
namespace mq
{
namespace ofi
namespace fair::mq::ofi
{
using namespace std;
@@ -134,6 +130,4 @@ auto Context::MakeSendMessage(size_t size) -> MessagePtr
return fSendFactory.CreateMessage(size);
}
} /* namespace ofi */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::ofi

View File

@@ -24,11 +24,7 @@
#include <thread>
#include <vector>
namespace fair
{
namespace mq
{
namespace ofi
namespace fair::mq::ofi
{
enum class ConnectionType : bool { Bind, Connect };
@@ -88,8 +84,6 @@ class Context
struct ContextError : std::runtime_error { using std::runtime_error::runtime_error; };
} /* namespace ofi */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::ofi
#endif /* FAIR_MQ_OFI_CONTEXT_H */

View File

@@ -17,8 +17,8 @@
#include <memory>
#include <type_traits>
namespace boost {
namespace asio {
namespace boost::asio
{
template<typename PodType>
auto buffer(const PodType& obj) -> boost::asio::const_buffer
@@ -26,12 +26,10 @@ auto buffer(const PodType& obj) -> boost::asio::const_buffer
return boost::asio::const_buffer(static_cast<const void*>(&obj), sizeof(PodType));
}
} // namespace asio
} // namespace boost
} // namespace boost::asio
namespace fair {
namespace mq {
namespace ofi {
namespace fair::mq::ofi
{
enum class ControlMessageType
{
@@ -109,8 +107,6 @@ auto MakeControlMessage(Args&&... args) -> ControlMessage
return ctrl;
}
} // namespace ofi
} // namespace mq
} // namespace fair
} // namespace fair::mq::ofi
#endif /* FAIR_MQ_OFI_CONTROLMESSAGES_H */

View File

@@ -7,7 +7,6 @@
********************************************************************************/
#include <fairmq/ofi/Message.h>
#include <fairmq/Tools.h>
#include <FairMQLogger.h>
#include <asiofi.hpp>
@@ -15,11 +14,7 @@
#include <cstdlib>
#include <zmq.h>
namespace fair
{
namespace mq
{
namespace ofi
namespace fair::mq::ofi
{
using namespace std;
@@ -110,6 +105,12 @@ auto Message::Rebuild() -> void
fHint = nullptr;
}
auto Message::Rebuild(Alignment /* alignment */) -> void
{
// TODO: implement alignment
Rebuild();
}
auto Message::Rebuild(const size_t size) -> void
{
if (fFreeFunction) {
@@ -131,6 +132,12 @@ auto Message::Rebuild(const size_t size) -> void
fHint = nullptr;
}
auto Message::Rebuild(const size_t size, Alignment /* alignment */) -> void
{
// TODO: implement alignment
Rebuild(size);
}
auto Message::Rebuild(void* /*data*/, const size_t size, fairmq_free_fn* ffn, void* hint) -> void
{
if (fFreeFunction) {
@@ -190,6 +197,4 @@ Message::~Message()
}
}
} /* namespace ofi */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::ofi

View File

@@ -17,11 +17,7 @@
#include <cstddef> // size_t
#include <zmq.h>
namespace fair
{
namespace mq
{
namespace ofi
namespace fair::mq::ofi
{
/**
@@ -52,7 +48,9 @@ class Message final : public fair::mq::Message
Message operator=(const Message&) = delete;
auto Rebuild() -> void override;
auto Rebuild(Alignment alignment) -> void override;
auto Rebuild(const size_t size) -> void override;
auto Rebuild(const size_t size, Alignment alignment) -> void override;
auto Rebuild(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr) -> void override;
auto GetData() const -> void* override;
@@ -75,8 +73,6 @@ class Message final : public fair::mq::Message
boost::container::pmr::memory_resource* fPmr;
}; /* class Message */
} /* namespace ofi */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::ofi
#endif /* FAIR_MQ_OFI_MESSAGE_H */

View File

@@ -8,16 +8,12 @@
#include <fairmq/ofi/Poller.h>
#include <fairmq/ofi/Socket.h>
#include <fairmq/Tools.h>
#include <fairmq/tools/Strings.h>
#include <FairMQLogger.h>
#include <zmq.h>
namespace fair
{
namespace mq
{
namespace ofi
namespace fair::mq::ofi
{
using namespace std;
@@ -153,6 +149,4 @@ Poller::~Poller()
delete[] fItems;
}
} /* namespace ofi */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::ofi

View File

@@ -18,11 +18,7 @@
#include <zmq.h>
namespace fair
{
namespace mq
{
namespace ofi
namespace fair::mq::ofi
{
class TransportFactory;
@@ -63,8 +59,6 @@ class Poller final : public FairMQPoller
std::unordered_map<std::string, int> fOffsetMap;
}; /* class Poller */
} /* namespace ofi */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::ofi
#endif /* FAIR_MQ_OFI_POLLER_H */

View File

@@ -9,7 +9,7 @@
#include <fairmq/ofi/ControlMessages.h>
#include <fairmq/ofi/Socket.h>
#include <fairmq/ofi/TransportFactory.h>
#include <fairmq/Tools.h>
#include <fairmq/tools/Strings.h>
#include <FairMQLogger.h>
#include <asiofi.hpp>
@@ -25,11 +25,7 @@
#include <mutex>
#include <queue>
namespace fair
{
namespace mq
{
namespace ofi
namespace fair::mq::ofi
{
using namespace std;
@@ -74,16 +70,16 @@ auto Socket::InitOfi(Address addr) -> void
hints.set_provider("verbs");
}
if (fRemoteAddr == addr) {
fOfiInfo = tools::make_unique<asiofi::info>(addr.Ip.c_str(), std::to_string(addr.Port).c_str(), 0, hints);
fOfiInfo = make_unique<asiofi::info>(addr.Ip.c_str(), std::to_string(addr.Port).c_str(), 0, hints);
} else {
fOfiInfo = tools::make_unique<asiofi::info>(addr.Ip.c_str(), std::to_string(addr.Port).c_str(), FI_SOURCE, hints);
fOfiInfo = make_unique<asiofi::info>(addr.Ip.c_str(), std::to_string(addr.Port).c_str(), FI_SOURCE, hints);
}
LOG(debug) << "OFI transport (" << fId << "): " << *fOfiInfo;
fOfiFabric = tools::make_unique<asiofi::fabric>(*fOfiInfo);
fOfiFabric = make_unique<asiofi::fabric>(*fOfiInfo);
fOfiDomain = tools::make_unique<asiofi::domain>(*fOfiFabric);
fOfiDomain = make_unique<asiofi::domain>(*fOfiFabric);
}
}
@@ -96,7 +92,7 @@ try {
InitOfi(fLocalAddr);
fPassiveEndpoint = tools::make_unique<asiofi::passive_endpoint>(fContext.GetIoContext(), *fOfiFabric);
fPassiveEndpoint = make_unique<asiofi::passive_endpoint>(fContext.GetIoContext(), *fOfiFabric);
//fPassiveEndpoint->set_local_address(Context::ConvertAddress(fLocalAddr));
BindControlEndpoint();
@@ -128,7 +124,7 @@ auto Socket::BindControlEndpoint() -> void
fPassiveEndpoint->listen([&](asiofi::info&& info) {
LOG(debug) << "OFI transport (" << fId
<< "): control band connection request received. Accepting ...";
fControlEndpoint = tools::make_unique<asiofi::connected_endpoint>(
fControlEndpoint = make_unique<asiofi::connected_endpoint>(
fContext.GetIoContext(), *fOfiDomain, info);
fControlEndpoint->enable();
fControlEndpoint->accept([&]() {
@@ -148,7 +144,7 @@ auto Socket::BindDataEndpoint() -> void
fPassiveEndpoint->listen([&](asiofi::info&& info) {
LOG(debug) << "OFI transport (" << fId
<< "): data band connection request received. Accepting ...";
fDataEndpoint = tools::make_unique<asiofi::connected_endpoint>(
fDataEndpoint = make_unique<asiofi::connected_endpoint>(
fContext.GetIoContext(), *fOfiDomain, info);
fDataEndpoint->enable();
fDataEndpoint->accept([&]() {
@@ -215,7 +211,7 @@ auto Socket::ConnectEndpoint(std::unique_ptr<asiofi::connected_endpoint>& endpoi
std::string band(type == Band::Control ? "control" : "data");
endpoint = tools::make_unique<asiofi::connected_endpoint>(fContext.GetIoContext(), *fOfiDomain);
endpoint = make_unique<asiofi::connected_endpoint>(fContext.GetIoContext(), *fOfiDomain);
endpoint->enable();
LOG(debug) << "OFI transport (" << fId << "): Sending " << band << " band connection request to " << fRemoteAddr;
@@ -284,7 +280,7 @@ try {
return size;
} catch (const std::exception& e) {
LOG(error) << e.what();
return static_cast<int64_t>(TransferResult::error);
return static_cast<int64_t>(TransferCode::error);
}
auto Socket::SendQueueReader() -> void
@@ -431,7 +427,7 @@ try {
return size;
} catch (const std::exception& e) {
LOG(error) << e.what();
return static_cast<int>(TransferResult::error);
return static_cast<int>(TransferCode::error);
}
auto Socket::Receive(std::vector<MessagePtr>& msgVec, const int /*timeout*/) -> int64_t
@@ -456,7 +452,7 @@ try {
return size;
} catch (const std::exception& e) {
LOG(error) << e.what();
return static_cast<int64_t>(TransferResult::error);
return static_cast<int64_t>(TransferCode::error);
}
auto Socket::RecvControlQueueReader() -> void
@@ -683,6 +679,4 @@ Socket::~Socket()
}
}
} /* namespace ofi */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::ofi

View File

@@ -23,11 +23,7 @@
#include <mutex>
namespace fair
{
namespace mq
{
namespace ofi
namespace fair::mq::ofi
{
/**
@@ -120,8 +116,6 @@ class Socket final : public fair::mq::Socket
struct SilentSocketError : SocketError { using SocketError::SocketError; };
} /* namespace ofi */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::ofi
#endif /* FAIR_MQ_OFI_SOCKET_H */

View File

@@ -10,15 +10,10 @@
#include <fairmq/ofi/Poller.h>
#include <fairmq/ofi/Socket.h>
#include <fairmq/ofi/TransportFactory.h>
#include <fairmq/Tools.h>
#include <stdexcept>
namespace fair
{
namespace mq
{
namespace ofi
namespace fair::mq::ofi
{
using namespace std;
@@ -122,6 +117,4 @@ auto TransportFactory::GetType() const -> Transport
return Transport::OFI;
}
} /* namespace ofi */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::ofi

View File

@@ -15,11 +15,7 @@
#include <asiofi.hpp>
namespace fair
{
namespace mq
{
namespace ofi
namespace fair::mq::ofi
{
/**
@@ -69,8 +65,6 @@ class TransportFactory final : public FairMQTransportFactory
asiofi::allocated_pool_resource fMemoryResource;
}; /* class TransportFactory */
} /* namespace ofi */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::ofi
#endif /* FAIR_MQ_OFI_TRANSPORTFACTORY_H */

View File

@@ -43,11 +43,7 @@ namespace
}
}
namespace fair
{
namespace mq
{
namespace plugins
namespace fair::mq::plugins
{
Control::Control(const string& name, const Plugin::Version version, const string& maintainer, const string& homepage, PluginServices* pluginServices)
@@ -475,6 +471,4 @@ Control::~Control()
UnsubscribeFromDeviceStateChange();
}
} /* namespace plugins */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::plugins

View File

@@ -21,11 +21,7 @@
#include <atomic>
#include <stdexcept>
namespace fair
{
namespace mq
{
namespace plugins
namespace fair::mq::plugins
{
class Control : public Plugin
@@ -68,8 +64,6 @@ REGISTER_FAIRMQ_PLUGIN(
// boost::optional<boost::program_options::options_description>
)
} /* namespace plugins */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::plugins
#endif /* FAIR_MQ_PLUGINS_CONTROL */

View File

@@ -22,11 +22,7 @@
using namespace std;
using fair::mq::tools::ToString;
namespace fair
{
namespace mq
{
namespace plugins
namespace fair::mq::plugins
{
DDS::DDS(const string& name,
@@ -458,6 +454,4 @@ DDS::~DDS()
}
}
} /* namespace plugins */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::plugins

View File

@@ -32,11 +32,7 @@
#include <utility> // pair
#include <vector>
namespace fair
{
namespace mq
{
namespace plugins
namespace fair::mq::plugins
{
struct DDSConfig
@@ -196,8 +192,6 @@ REGISTER_FAIRMQ_PLUGIN(
DDSProgramOptions // custom program options for the plugin
)
} /* namespace plugins */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::plugins
#endif /* FAIR_MQ_PLUGINS_DDS */

View File

@@ -13,7 +13,7 @@
#include <FairMQLogger.h>
#include <fairmq/tools/Semaphore.h>
#include <fairmq/tools/CppSTL.h>
#include <memory> // make_unique
#include <string>
namespace pmix
@@ -156,7 +156,7 @@ class Commands
void Send(const std::string& msg, const std::vector<proc>& destination)
{
std::unique_ptr<Holder> holder = fair::mq::tools::make_unique<Holder>();
std::unique_ptr<Holder> holder = std::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));

View File

@@ -9,7 +9,7 @@
#include "PMIxPlugin.h"
#include <fairmq/sdk/commands/Commands.h>
#include <fairmq/Tools.h>
#include <fairmq/tools/Strings.h>
#include <sstream>
#include <stdexcept>
@@ -18,11 +18,7 @@
using namespace std;
using namespace fair::mq::sdk::cmd;
namespace fair
{
namespace mq
{
namespace plugins
namespace fair::mq::plugins
{
PMIxPlugin::PMIxPlugin(const string& name,
@@ -304,6 +300,4 @@ auto PMIxPlugin::WaitForExitingAck() -> void
});
}
} /* namespace plugins */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::plugins

View File

@@ -24,11 +24,7 @@
#include <unistd.h>
#include <vector>
namespace fair
{
namespace mq
{
namespace plugins
namespace fair::mq::plugins
{
class PMIxPlugin : public Plugin
@@ -88,8 +84,6 @@ REGISTER_FAIRMQ_PLUGIN(
PMIxProgramOptions // custom program options for the plugin
)
} /* namespace plugins */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::plugins
#endif /* FAIR_MQ_PLUGINS_PMIX */

View File

@@ -15,11 +15,7 @@
using namespace std;
namespace fair
{
namespace mq
{
namespace plugins
namespace fair::mq::plugins
{
Config::Config(const string& name, const Plugin::Version version, const string& maintainer, const string& homepage, PluginServices* pluginServices)
@@ -89,6 +85,4 @@ Config::~Config()
UnsubscribeFromDeviceStateChange();
}
} /* namespace plugins */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::plugins

View File

@@ -14,11 +14,7 @@
#include <string>
namespace fair
{
namespace mq
{
namespace plugins
namespace fair::mq::plugins
{
class Config : public Plugin
@@ -40,8 +36,6 @@ REGISTER_FAIRMQ_PLUGIN(
ConfigPluginProgramOptions
)
} /* namespace plugins */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::plugins
#endif /* FAIR_MQ_PLUGINS_CONFIG */

View File

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

View File

@@ -29,9 +29,8 @@
#define FAIR_LOG LOG
#endif /* ifndef FAIR_LOG */
namespace fair {
namespace mq {
namespace sdk {
namespace fair::mq::sdk
{
template<typename... SignatureArgTypes>
struct AsioAsyncOpImplBase
@@ -219,9 +218,6 @@ struct AsioAsyncOp<Executor,
}
};
} /* namespace sdk */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::sdk
#endif /* FAIR_MQ_SDK_ASIOASYNCOP_H */

View File

@@ -14,9 +14,8 @@
#include <memory>
#include <utility>
namespace fair {
namespace mq {
namespace sdk {
namespace fair::mq::sdk
{
using DefaultExecutor = asio::any_io_executor;
using DefaultAllocator = std::allocator<int>;
@@ -69,8 +68,6 @@ class AsioBase
AllocatorType fAllocator;
};
} /* namespace sdk */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::sdk
#endif /* FAIR_MQ_SDK_ASIOBASE_H */

View File

@@ -16,9 +16,8 @@
#include <chrono>
#include <cstdint>
namespace fair {
namespace mq {
namespace sdk {
namespace fair::mq::sdk
{
/**
* @class DDSAgent <fairmq/sdk/DDSAgent.h>
@@ -74,8 +73,6 @@ class DDSAgent
std::string fUsername;
};
} // namespace sdk
} // namespace mq
} // namespace fair
} // namespace fair::mq::sdk
#endif /* FAIR_MQ_SDK_DDSSAGENT_H */

View File

@@ -14,9 +14,8 @@
#include <ostream>
#include <cstdint>
namespace fair {
namespace mq {
namespace sdk {
namespace fair::mq::sdk
{
/**
* @class DDSCollection <fairmq/sdk/DDSCollection.h>
@@ -42,8 +41,6 @@ class DDSCollection
Id fId;
};
} // namespace sdk
} // namespace mq
} // namespace fair
} // namespace fair::mq::sdk
#endif /* FAIR_MQ_SDK_DDSCOLLECTION_H */

View File

@@ -11,14 +11,13 @@
#include <cstdlib>
#include <dds/dds.h>
#include <fairlogger/Logger.h>
#include <fairmq/Tools.h>
#include <fairmq/tools/InstanceLimit.h>
#include <fairmq/sdk/DDSInfo.h>
#include <sstream>
#include <utility>
namespace fair {
namespace mq {
namespace sdk {
namespace fair::mq::sdk
{
struct DDSEnvironment::Impl
{
@@ -133,6 +132,4 @@ auto operator<<(std::ostream& os, DDSEnvironment env) -> std::ostream&
<< "$DDS_CONFIG_HOME: " << env.GetConfigHome() / DDSEnvironment::Path(".DDS");
}
} // namespace sdk
} // namespace mq
} // namespace fair
} // namespace fair::mq::sdk

View File

@@ -13,9 +13,8 @@
#include <memory>
#include <ostream>
namespace fair {
namespace mq {
namespace sdk {
namespace fair::mq::sdk
{
/**
* @class DDSEnvironment DDSSession.h <fairmq/sdk/DDSSession.h>
@@ -40,8 +39,6 @@ class DDSEnvironment
using DDSEnv = DDSEnvironment;
} // namespace sdk
} // namespace mq
} // namespace fair
} // namespace fair::mq::sdk
#endif /* FAIR_MQ_SDK_DDSENVIRONMENT_H */

View File

@@ -11,23 +11,11 @@
#include <string>
namespace dds {
namespace tools_api {
namespace dds::tools_api { class CSession; }
namespace dds::topology_api { class CTopology; }
class CSession;
} // namespace tools_api
namespace topology_api {
class CTopology;
} // namespace topology_api
} // namespace dds
namespace fair {
namespace mq {
namespace sdk {
namespace fair::mq::sdk
{
const std::string DDSVersion("@DDS_VERSION@");
const std::string DDSInstallPrefix("@DDS_INSTALL_PREFIX@");
@@ -36,8 +24,6 @@ const std::string DDSIncludeDir("@DDS_INCDIR@");
const std::string DDSLibraryDir("@DDS_LIBDIR@");
const std::string DDSPluginDir("@DDS_PLUGINDIR@");
} // namespace sdk
} // namespace mq
} // namespace fair
} // namespace fair::mq::sdk
#endif /* FAIR_MQ_SDK_DDSINFO_H */

View File

@@ -14,7 +14,8 @@
#include <cstdlib>
#include <dds/dds.h>
#include <fairlogger/Logger.h>
#include <fairmq/Tools.h>
#include <fairmq/tools/Strings.h>
#include <fairmq/tools/Semaphore.h>
#include <fairmq/sdk/DDSAgent.h>
#include <fairmq/sdk/DDSEnvironment.h>
#include <fairmq/sdk/DDSTopology.h>
@@ -24,9 +25,8 @@
#include <utility>
#include <vector>
namespace fair {
namespace mq {
namespace sdk {
namespace fair::mq::sdk
{
auto operator<<(std::ostream& os, DDSRMSPlugin plugin) -> std::ostream&
{
@@ -383,6 +383,4 @@ auto getMostRecentRunningDDSSession(DDSEnv env) -> DDSSession
return DDSSession(DDSSession::Id(sessionId), std::move(env));
}
} // namespace sdk
} // namespace mq
} // namespace fair
} // namespace fair::mq::sdk

View File

@@ -24,9 +24,8 @@
#include <functional>
#include <vector>
namespace fair {
namespace mq {
namespace sdk {
namespace fair::mq::sdk
{
/**
* @enum DDSRMSPlugin DDSSession.h <fairmq/sdk/DDSSession.h>
@@ -114,8 +113,6 @@ class DDSSession
auto getMostRecentRunningDDSSession(DDSEnv env = {}) -> DDSSession;
} // namespace sdk
} // namespace mq
} // namespace fair
} // namespace fair::mq::sdk
#endif /* FAIR_MQ_SDK_DDSSESSION_H */

View File

@@ -14,9 +14,8 @@
#include <ostream>
#include <cstdint>
namespace fair {
namespace mq {
namespace sdk {
namespace fair::mq::sdk
{
/**
* @class DDSTask <fairmq/sdk/DDSTask.h>
@@ -45,8 +44,6 @@ class DDSTask
DDSCollection::Id fCollectionId;
};
} // namespace sdk
} // namespace mq
} // namespace fair
} // namespace fair::mq::sdk
#endif /* FAIR_MQ_SDK_DDSTASK_H */

View File

@@ -11,16 +11,14 @@
#include <boost/range/iterator_range.hpp>
#include <dds/dds.h>
#include <fairlogger/Logger.h>
#include <fairmq/Tools.h>
#include <fairmq/sdk/DDSEnvironment.h>
#include <memory>
#include <sstream>
#include <stdexcept>
#include <utility>
namespace fair {
namespace mq {
namespace sdk {
namespace fair::mq::sdk
{
struct DDSTopology::Impl
{
@@ -113,6 +111,4 @@ try {
return os << "DDS topology: " << t.GetName();
}
} // namespace sdk
} // namespace mq
} // namespace fair
} // namespace fair::mq::sdk

View File

@@ -18,9 +18,8 @@
#include <string>
#include <vector>
namespace fair {
namespace mq {
namespace sdk {
namespace fair::mq::sdk
{
/**
* @class DDSTopology DDSTopology.h <fairmq/sdk/DDSTopology.h>
@@ -71,8 +70,6 @@ class DDSTopology
using DDSTopo = DDSTopology;
} // namespace sdk
} // namespace mq
} // namespace fair
} // namespace fair::mq::sdk
#endif /* FAIR_MQ_SDK_DDSTOPOLOGY_H */

View File

@@ -8,8 +8,8 @@
#include "Error.h"
namespace fair {
namespace mq {
namespace fair::mq
{
const char* ErrorCategory::name() const noexcept
{
@@ -40,5 +40,4 @@ const ErrorCategory errorCategory{};
std::error_code MakeErrorCode(ErrorCode e) { return {static_cast<int>(e), errorCategory}; }
} // namespace mq
} // namespace fair
} // namespace fair::mq

View File

@@ -13,9 +13,11 @@
#include <stdexcept>
#include <system_error>
namespace fair {
namespace mq {
namespace sdk {
namespace fair::mq
{
namespace sdk
{
struct RuntimeError : ::std::runtime_error
{
@@ -45,15 +47,15 @@ struct ErrorCategory : std::error_category
std::string message(int ev) const override;
};
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq
namespace std {
namespace std
{
template<>
struct is_error_code_enum<fair::mq::ErrorCode> : true_type
{};
} // namespace std
} // namespace std
#endif /* FAIR_MQ_SDK_ERROR_H */

View File

@@ -10,9 +10,8 @@
#include <dds/dds.h>
namespace fair {
namespace mq {
namespace sdk {
namespace fair::mq::sdk
{
/// @brief Helper to (Re)Construct a FairMQ topology based on already existing native DDS API objects
/// @param nativeSession Existing and initialized CSession (either via create() or attach())
@@ -27,6 +26,4 @@ auto MakeTopology(dds::topology_api::CTopology nativeTopo,
return {DDSTopo(std::move(nativeTopo), env), DDSSession(std::move(nativeSession), env), blockUntilConnected};
}
} // namespace sdk
} // namespace mq
} // namespace fair
} // namespace fair::mq::sdk

View File

@@ -48,9 +48,8 @@
#include <utility>
#include <vector>
namespace fair {
namespace mq {
namespace sdk {
namespace fair::mq::sdk
{
using DeviceId = std::string;
using DeviceState = fair::mq::State;
@@ -1366,8 +1365,6 @@ auto MakeTopology(dds::topology_api::CTopology nativeTopo,
DDSEnv env = {},
bool blockUntilConnected = false) -> Topology;
} // namespace sdk
} // namespace mq
} // namespace fair
} // namespace fair::mq::sdk
#endif /* FAIR_MQ_SDK_TOPOLOGY_H */

View File

@@ -13,8 +13,7 @@
#include <asio/associated_executor.hpp>
#include <type_traits>
namespace asio {
namespace detail {
namespace asio::detail {
/// Specialize to match our coding conventions
template<typename T, typename Executor>
@@ -44,7 +43,6 @@ struct associated_allocator_impl<T,
}
};
} /* namespace detail */
} /* namespace asio */
} /* namespace asio::detail */
#endif /* FAIR_MQ_SDK_TRAITS_H */

View File

@@ -17,10 +17,8 @@
using namespace std;
namespace fair {
namespace mq {
namespace sdk {
namespace cmd {
namespace fair::mq::sdk::cmd
{
array<Result, 2> fbResultToResult =
{
@@ -219,37 +217,37 @@ string Cmds::Serialize(const Format type) const
switch (cmd->GetType()) {
case Type::check_state: {
cmdBuilder = tools::make_unique<FBCommandBuilder>(fbb);
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
}
break;
case Type::change_state: {
cmdBuilder = tools::make_unique<FBCommandBuilder>(fbb);
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_transition(GetFBTransition(static_cast<ChangeState&>(*cmd).GetTransition()));
}
break;
case Type::dump_config: {
cmdBuilder = tools::make_unique<FBCommandBuilder>(fbb);
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
}
break;
break;
case Type::subscribe_to_state_change: {
auto _cmd = static_cast<SubscribeToStateChange&>(*cmd);
cmdBuilder = tools::make_unique<FBCommandBuilder>(fbb);
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_interval(_cmd.GetInterval());
}
break;
case Type::unsubscribe_from_state_change: {
cmdBuilder = tools::make_unique<FBCommandBuilder>(fbb);
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
}
break;
case Type::state_change_exiting_received: {
cmdBuilder = tools::make_unique<FBCommandBuilder>(fbb);
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
}
break;
case Type::get_properties: {
auto _cmd = static_cast<GetProperties&>(*cmd);
auto query = fbb.CreateString(_cmd.GetQuery());
cmdBuilder = tools::make_unique<FBCommandBuilder>(fbb);
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_request_id(_cmd.GetRequestId());
cmdBuilder->add_property_query(query);
}
@@ -263,21 +261,21 @@ string Cmds::Serialize(const Format type) const
propsVector.push_back(CreateFBProperty(fbb, key, val));
}
auto props = fbb.CreateVector(propsVector);
cmdBuilder = tools::make_unique<FBCommandBuilder>(fbb);
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_request_id(_cmd.GetRequestId());
cmdBuilder->add_properties(props);
}
break;
case Type::subscription_heartbeat: {
auto _cmd = static_cast<SubscriptionHeartbeat&>(*cmd);
cmdBuilder = tools::make_unique<FBCommandBuilder>(fbb);
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_interval(_cmd.GetInterval());
}
break;
case Type::current_state: {
auto _cmd = static_cast<CurrentState&>(*cmd);
auto deviceId = fbb.CreateString(_cmd.GetDeviceId());
cmdBuilder = tools::make_unique<FBCommandBuilder>(fbb);
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_device_id(deviceId);
cmdBuilder->add_current_state(GetFBState(_cmd.GetCurrentState()));
}
@@ -285,7 +283,7 @@ string Cmds::Serialize(const Format type) const
case Type::transition_status: {
auto _cmd = static_cast<TransitionStatus&>(*cmd);
auto deviceId = fbb.CreateString(_cmd.GetDeviceId());
cmdBuilder = tools::make_unique<FBCommandBuilder>(fbb);
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_device_id(deviceId);
cmdBuilder->add_task_id(_cmd.GetTaskId());
cmdBuilder->add_result(GetFBResult(_cmd.GetResult()));
@@ -297,7 +295,7 @@ string Cmds::Serialize(const Format type) const
auto _cmd = static_cast<Config&>(*cmd);
auto deviceId = fbb.CreateString(_cmd.GetDeviceId());
auto config = fbb.CreateString(_cmd.GetConfig());
cmdBuilder = tools::make_unique<FBCommandBuilder>(fbb);
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_device_id(deviceId);
cmdBuilder->add_config_string(config);
}
@@ -305,7 +303,7 @@ string Cmds::Serialize(const Format type) const
case Type::state_change_subscription: {
auto _cmd = static_cast<StateChangeSubscription&>(*cmd);
auto deviceId = fbb.CreateString(_cmd.GetDeviceId());
cmdBuilder = tools::make_unique<FBCommandBuilder>(fbb);
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_device_id(deviceId);
cmdBuilder->add_task_id(_cmd.GetTaskId());
cmdBuilder->add_result(GetFBResult(_cmd.GetResult()));
@@ -314,7 +312,7 @@ string Cmds::Serialize(const Format type) const
case Type::state_change_unsubscription: {
auto _cmd = static_cast<StateChangeUnsubscription&>(*cmd);
auto deviceId = fbb.CreateString(_cmd.GetDeviceId());
cmdBuilder = tools::make_unique<FBCommandBuilder>(fbb);
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_device_id(deviceId);
cmdBuilder->add_task_id(_cmd.GetTaskId());
cmdBuilder->add_result(GetFBResult(_cmd.GetResult()));
@@ -323,7 +321,7 @@ string Cmds::Serialize(const Format type) const
case Type::state_change: {
auto _cmd = static_cast<StateChange&>(*cmd);
auto deviceId = fbb.CreateString(_cmd.GetDeviceId());
cmdBuilder = tools::make_unique<FBCommandBuilder>(fbb);
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_device_id(deviceId);
cmdBuilder->add_task_id(_cmd.GetTaskId());
cmdBuilder->add_last_state(GetFBState(_cmd.GetLastState()));
@@ -342,7 +340,7 @@ string Cmds::Serialize(const Format type) const
propsVector.push_back(prop);
}
auto props = fbb.CreateVector(propsVector);
cmdBuilder = tools::make_unique<FBCommandBuilder>(fbb);
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_device_id(deviceId);
cmdBuilder->add_request_id(_cmd.GetRequestId());
cmdBuilder->add_result(GetFBResult(_cmd.GetResult()));
@@ -352,7 +350,7 @@ string Cmds::Serialize(const Format type) const
case Type::properties_set: {
auto _cmd = static_cast<PropertiesSet&>(*cmd);
auto deviceId = fbb.CreateString(_cmd.GetDeviceId());
cmdBuilder = tools::make_unique<FBCommandBuilder>(fbb);
cmdBuilder = make_unique<FBCommandBuilder>(fbb);
cmdBuilder->add_device_id(deviceId);
cmdBuilder->add_request_id(_cmd.GetRequestId());
cmdBuilder->add_result(GetFBResult(_cmd.GetResult()));
@@ -478,7 +476,4 @@ void Cmds::Deserialize(const string& str, const Format type)
}
}
} // namespace cmd
} // namespace sdk
} // namespace mq
} // namespace fair
} // namespace fair::mq::sdk::cmd

View File

@@ -10,7 +10,6 @@
#define FAIR_MQ_SDK_COMMANDFACTORY
#include <fairmq/States.h>
#include <fairmq/tools/CppSTL.h>
#include <vector>
#include <string>
@@ -18,13 +17,7 @@
#include <type_traits>
#include <stdexcept>
namespace fair
{
namespace mq
{
namespace sdk
{
namespace cmd
namespace fair::mq::sdk::cmd
{
enum class Format : int {
@@ -354,7 +347,7 @@ struct PropertiesSet : Cmd {
template<typename C, typename... Args>
std::unique_ptr<Cmd> make(Args&&... args)
{
return fair::mq::tools::make_unique<C>(std::forward<Args>(args)...);
return std::make_unique<C>(std::forward<Args>(args)...);
}
struct Cmds
@@ -415,9 +408,6 @@ std::string GetTypeName(const Type type);
inline std::ostream& operator<<(std::ostream& os, const Result& result) { return os << GetResultName(result); }
inline std::ostream& operator<<(std::ostream& os, const Type& type) { return os << GetTypeName(type); }
} /* namespace cmd */
} /* namespace sdk */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::sdk::cmd
#endif /* FAIR_MQ_SDK_COMMANDFACTORY */

View File

@@ -8,14 +8,11 @@
#include <string>
namespace fair {
namespace mq {
namespace sdk {
namespace fair::mq::sdk
{
namespace cmd {
constexpr auto commandsFormatDefFbs = R"(@commands_format_def_fbs@)";
} // namespace cmd
} // namespace sdk
} // namespace mq
} // namespace fair
} // namespace fair::mq::sdk

View File

@@ -9,7 +9,7 @@
#include <fairmq/sdk/commands/Commands.h>
#include <fairmq/States.h>
#include <fairmq/SDK.h>
#include <fairmq/Tools.h>
#include <fairmq/tools/Strings.h>
#include <boost/program_options.hpp>

View File

@@ -28,11 +28,7 @@
#include <unistd.h>
#include <sys/types.h>
namespace fair
{
namespace mq
{
namespace shmem
namespace fair::mq::shmem
{
struct SharedMemoryError : std::runtime_error { using std::runtime_error::runtime_error; };
@@ -267,9 +263,8 @@ struct SegmentAllocateAligned : public boost::static_visitor<void*>
struct SegmentBufferShrink : public boost::static_visitor<char*>
{
SegmentBufferShrink(const size_t _old_size, const size_t _new_size, char* _local_ptr)
: old_size(_old_size)
, new_size(_new_size)
SegmentBufferShrink(const size_t _new_size, char* _local_ptr)
: new_size(_new_size)
, local_ptr(_local_ptr)
{}
@@ -277,10 +272,9 @@ struct SegmentBufferShrink : public boost::static_visitor<char*>
char* operator()(S& s) const
{
boost::interprocess::managed_shared_memory::size_type shrunk_size = new_size;
return s.template allocation_command<char>(boost::interprocess::shrink_in_place, old_size + 128, shrunk_size, local_ptr);
return s.template allocation_command<char>(boost::interprocess::shrink_in_place, new_size + 128, shrunk_size, local_ptr);
}
const size_t old_size;
const size_t new_size;
mutable char* local_ptr;
};
@@ -295,8 +289,6 @@ struct SegmentDeallocate : public boost::static_visitor<>
void* ptr;
};
} // namespace shmem
} // namespace mq
} // namespace fair
} // namespace fair::mq::shmem
#endif /* FAIR_MQ_SHMEM_COMMON_H_ */

View File

@@ -22,7 +22,6 @@
#include <FairMQLogger.h>
#include <FairMQMessage.h>
#include <fairmq/ProgOptions.h>
#include <fairmq/tools/CppSTL.h>
#include <fairmq/tools/Strings.h>
#include <boost/date_time/posix_time/posix_time.hpp>
@@ -36,6 +35,7 @@
#include <cstdlib> // getenv
#include <condition_variable>
#include <memory> // make_unique
#include <mutex>
#include <set>
#include <sstream>
@@ -48,11 +48,7 @@
#include <sys/mman.h> // mlock
namespace fair
{
namespace mq
{
namespace shmem
namespace fair::mq::shmem
{
class Manager
@@ -143,23 +139,19 @@ class Manager
} catch(interprocess_exception& bie) {
LOG(error) << "something went wrong: " << bie.what();
}
}
if (mlockSegment) {
LOG(debug) << "Locking the managed segment memory pages...";
if (mlock(boost::apply_visitor(SegmentAddress{}, fSegments.at(fSegmentId)), boost::apply_visitor(SegmentSize{}, fSegments.at(fSegmentId))) == -1) {
LOG(error) << "Could not lock the managed segment memory. Code: " << errno << ", reason: " << strerror(errno);
if (mlockSegment) {
LOG(debug) << "Locking the managed segment memory pages...";
if (mlock(boost::apply_visitor(SegmentAddress{}, fSegments.at(fSegmentId)), boost::apply_visitor(SegmentSize{}, fSegments.at(fSegmentId))) == -1) {
LOG(error) << "Could not lock the managed segment memory. Code: " << errno << ", reason: " << strerror(errno);
}
LOG(debug) << "Successfully locked the managed segment memory pages.";
}
if (zeroSegment) {
LOG(debug) << "Zeroing the managed segment free memory...";
boost::apply_visitor(SegmentMemoryZeroer{}, fSegments.at(fSegmentId));
LOG(debug) << "Successfully zeroed the managed segment free memory.";
}
LOG(debug) << "Successfully locked the managed segment memory pages.";
}
if (zeroSegment) {
LOG(debug) << "Zeroing the managed segment free memory...";
boost::apply_visitor(SegmentMemoryZeroer{}, fSegments.at(fSegmentId));
LOG(debug) << "Successfully zeroed the managed segment free memory.";
}
{
boost::interprocess::scoped_lock<boost::interprocess::named_mutex> lock(fShmMtx);
fShmRegions = fManagementSegment.find_or_construct<Uint16RegionInfoHashMap>(unique_instance)(fShmVoidAlloc);
@@ -278,7 +270,7 @@ class Manager
// create region info
fShmRegions->emplace(id, RegionInfo(path.c_str(), flags, userFlags, fShmVoidAlloc));
auto r = fRegions.emplace(id, tools::make_unique<Region>(fShmId, id, size, false, callback, bulkCallback, path, flags));
auto r = fRegions.emplace(id, std::make_unique<Region>(fShmId, id, size, false, callback, bulkCallback, path, flags));
// LOG(debug) << "Created region with id '" << id << "', path: '" << path << "', flags: '" << flags << "'";
r.first->second->StartReceivingAcks();
@@ -316,7 +308,7 @@ class Manager
int flags = regionInfo.fFlags;
// LOG(debug) << "Located remote region with id '" << id << "', path: '" << path << "', flags: '" << flags << "'";
auto r = fRegions.emplace(id, tools::make_unique<Region>(fShmId, id, 0, true, nullptr, nullptr, path, flags));
auto r = fRegions.emplace(id, std::make_unique<Region>(fShmId, id, 0, true, nullptr, nullptr, path, flags));
return r.first->second.get();
} catch (std::out_of_range& oor) {
LOG(error) << "Could not get remote region with id '" << id << "'. Does the region creator run with the same session id?";
@@ -569,9 +561,9 @@ class Manager
#endif
}
char* ShrinkInPlace(size_t oldSize, size_t newSize, char* localPtr, uint16_t segmentId)
char* ShrinkInPlace(size_t newSize, char* localPtr, uint16_t segmentId)
{
return boost::apply_visitor(SegmentBufferShrink{oldSize, newSize, localPtr}, fSegments.at(segmentId));
return boost::apply_visitor(SegmentBufferShrink{newSize, localPtr}, fSegments.at(segmentId));
}
uint16_t GetSegmentId() const { return fSegmentId; }
@@ -647,8 +639,6 @@ class Manager
bool fThrowOnBadAlloc;
};
} // namespace shmem
} // namespace mq
} // namespace fair
} // namespace fair::mq::shmem
#endif /* FAIR_MQ_SHMEM_MANAGER_H_ */

View File

@@ -12,7 +12,6 @@
#include "Manager.h"
#include "Region.h"
#include "UnmanagedRegion.h"
#include <fairmq/Tools.h>
#include <FairMQLogger.h>
#include <FairMQMessage.h>
#include <FairMQUnmanagedRegion.h>
@@ -25,11 +24,7 @@
#include <sys/types.h> // getpid
#include <unistd.h> // pid_t
namespace fair
{
namespace mq
{
namespace shmem
namespace fair::mq::shmem
{
class Socket;
@@ -50,11 +45,12 @@ class Message final : public fair::mq::Message
fManager.IncrementMsgCounter();
}
Message(Manager& manager, Alignment /* alignment */, FairMQTransportFactory* factory = nullptr)
Message(Manager& manager, Alignment alignment, FairMQTransportFactory* factory = nullptr)
: fair::mq::Message(factory)
, fManager(manager)
, fQueued(false)
, fMeta{0, 0, 0, fManager.GetSegmentId(), -1}
, fAlignment(alignment.alignment)
, fRegionPtr(nullptr)
, fLocalPtr(nullptr)
{
@@ -78,10 +74,11 @@ class Message final : public fair::mq::Message
, fManager(manager)
, fQueued(false)
, fMeta{0, 0, 0, fManager.GetSegmentId(), -1}
, fAlignment(alignment.alignment)
, fRegionPtr(nullptr)
, fLocalPtr(nullptr)
{
InitializeChunk(size, static_cast<size_t>(alignment));
InitializeChunk(size, fAlignment);
fManager.IncrementMsgCounter();
}
@@ -142,6 +139,13 @@ class Message final : public fair::mq::Message
fQueued = false;
}
void Rebuild(Alignment alignment) override
{
CloseMessage();
fQueued = false;
fAlignment = alignment.alignment;
}
void Rebuild(const size_t size) override
{
CloseMessage();
@@ -149,6 +153,14 @@ class Message final : public fair::mq::Message
InitializeChunk(size);
}
void Rebuild(const size_t size, Alignment alignment) override
{
CloseMessage();
fQueued = false;
fAlignment = alignment.alignment;
InitializeChunk(size, fAlignment);
}
void Rebuild(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr) override
{
CloseMessage();
@@ -194,9 +206,12 @@ class Message final : public fair::mq::Message
{
if (newSize == fMeta.fSize) {
return true;
} else if (newSize == 0) {
Deallocate();
return true;
} else if (newSize <= fMeta.fSize) {
try {
fLocalPtr = fManager.ShrinkInPlace(fMeta.fSize, newSize, fLocalPtr, fMeta.fSegmentId);
fLocalPtr = fManager.ShrinkInPlace(newSize, fLocalPtr, fMeta.fSegmentId);
fMeta.fSize = newSize;
return true;
} catch (boost::interprocess::interprocess_exception& e) {
@@ -242,6 +257,7 @@ class Message final : public fair::mq::Message
Manager& fManager;
bool fQueued;
MetaHeader fMeta;
size_t fAlignment; // TODO: put this to debug mode
mutable Region* fRegionPtr;
mutable char* fLocalPtr;
@@ -255,7 +271,7 @@ class Message final : public fair::mq::Message
return fLocalPtr;
}
void CloseMessage()
void Deallocate()
{
if (fMeta.fHandle >= 0 && !fQueued) {
if (fMeta.fRegionId == 0) {
@@ -276,13 +292,17 @@ class Message final : public fair::mq::Message
}
fLocalPtr = nullptr;
fMeta.fSize = 0;
}
fManager.DecrementMsgCounter();
void CloseMessage()
{
Deallocate();
fAlignment = 0;
fManager.DecrementMsgCounter(); // TODO: put this to debug mode
}
};
}
}
}
} // namespace fair::mq::shmem
#endif /* FAIR_MQ_SHMEM_MESSAGE_H_ */

View File

@@ -9,7 +9,7 @@
#include "Monitor.h"
#include "Common.h"
#include <fairmq/Tools.h>
#include <fairmq/tools/Strings.h>
#include <fairlogger/Logger.h>
#include <boost/interprocess/managed_shared_memory.hpp>
@@ -42,12 +42,29 @@ namespace
volatile sig_atomic_t gSignalStatus = 0;
}
namespace fair
namespace fair::mq::shmem
{
namespace mq
{
namespace shmem
struct TerminalConfig
{
TerminalConfig()
{
termios t;
tcgetattr(STDIN_FILENO, &t); // get the current terminal I/O structure
t.c_lflag &= ~ICANON; // disable canonical input
t.c_lflag &= ~ECHO; // do not echo input chars
tcsetattr(STDIN_FILENO, TCSANOW, &t); // apply the new settings
}
~TerminalConfig()
{
termios t;
tcgetattr(STDIN_FILENO, &t); // get the current terminal I/O structure
t.c_lflag |= ICANON; // re-enable canonical input
t.c_lflag |= ECHO; // echo input chars
tcsetattr(STDIN_FILENO, TCSANOW, &t); // apply the new settings
}
};
void signalHandler(int signal)
{
@@ -120,10 +137,7 @@ void Monitor::Run()
if (fInteractive) {
Interactive();
} else {
while (!fTerminating) {
this_thread::sleep_for(chrono::milliseconds(fIntervalInMS));
CheckSegment();
}
CheckSegment();
}
if (!fViewOnly) {
@@ -158,27 +172,6 @@ void Monitor::MonitorHeartbeats()
RemoveQueue(fControlQueueName);
}
struct TerminalConfig
{
TerminalConfig()
{
termios t;
tcgetattr(STDIN_FILENO, &t); // get the current terminal I/O structure
t.c_lflag &= ~ICANON; // disable canonical input
t.c_lflag &= ~ECHO; // do not echo input chars
tcsetattr(STDIN_FILENO, TCSANOW, &t); // apply the new settings
}
~TerminalConfig()
{
termios t;
tcgetattr(STDIN_FILENO, &t); // get the current terminal I/O structure
t.c_lflag |= ICANON; // re-enable canonical input
t.c_lflag |= ECHO; // echo input chars
tcsetattr(STDIN_FILENO, TCSANOW, &t); // apply the new settings
}
};
void Monitor::Interactive()
{
char c;
@@ -575,6 +568,4 @@ Monitor::~Monitor()
}
}
} // namespace shmem
} // namespace mq
} // namespace fair
} // namespace fair::mq::shmem

View File

@@ -17,11 +17,7 @@
#include <utility> // pair
#include <vector>
namespace fair
{
namespace mq
{
namespace shmem
namespace fair::mq::shmem
{
struct SessionId
@@ -120,8 +116,6 @@ class Monitor
std::unordered_map<std::string, std::chrono::high_resolution_clock::time_point> fDeviceHeartbeats;
};
} // namespace shmem
} // namespace mq
} // namespace fair
} // namespace fair::mq::shmem
#endif /* FAIR_MQ_SHMEM_MONITOR_H_ */

View File

@@ -9,7 +9,7 @@
#define FAIR_MQ_SHMEM_POLLER_H_
#include "Socket.h"
#include <fairmq/Tools.h>
#include <fairmq/tools/Strings.h>
#include <FairMQChannel.h>
#include <FairMQLogger.h>
#include <FairMQPoller.h>
@@ -21,9 +21,8 @@
class FairMQChannel;
namespace fair {
namespace mq {
namespace shmem {
namespace fair::mq::shmem
{
class Poller final : public fair::mq::Poller
{
@@ -201,8 +200,6 @@ class Poller final : public fair::mq::Poller
std::unordered_map<std::string, int> fOffsetMap;
};
} // namespace shmem
} // namespace mq
} // namespace fair
} // namespace fair::mq::shmem
#endif /* FAIR_MQ_SHMEM_POLLER_H_ */

View File

@@ -19,7 +19,6 @@
#include <FairMQLogger.h>
#include <FairMQUnmanagedRegion.h>
#include <fairmq/tools/CppSTL.h>
#include <fairmq/tools/Strings.h>
#include <boost/filesystem.hpp>
@@ -32,6 +31,7 @@
#include <algorithm> // min
#include <atomic>
#include <thread>
#include <memory> // make_unique
#include <mutex>
#include <condition_variable>
#include <unordered_map>
@@ -39,11 +39,7 @@
#include <chrono>
#include <ios>
namespace fair
{
namespace mq
{
namespace shmem
namespace fair::mq::shmem
{
struct Region
@@ -113,9 +109,9 @@ struct Region
using namespace boost::interprocess;
if (fRemote) {
fQueue = tools::make_unique<message_queue>(open_only, fQueueName.c_str());
fQueue = std::make_unique<message_queue>(open_only, fQueueName.c_str());
} else {
fQueue = tools::make_unique<message_queue>(create_only, fQueueName.c_str(), 1024, fAckBunchSize * sizeof(RegionBlock));
fQueue = std::make_unique<message_queue>(create_only, fQueueName.c_str(), 1024, fAckBunchSize * sizeof(RegionBlock));
}
LOG(debug) << "shmem: initialized region queue: " << fQueueName;
}
@@ -123,7 +119,7 @@ struct Region
void StartSendingAcks() { fAcksSender = std::thread(&Region::SendAcks, this); }
void SendAcks()
{
std::unique_ptr<RegionBlock[]> blocks = tools::make_unique<RegionBlock[]>(fAckBunchSize);
std::unique_ptr<RegionBlock[]> blocks = std::make_unique<RegionBlock[]>(fAckBunchSize);
size_t blocksToSend = 0;
while (true) {
@@ -165,7 +161,7 @@ struct Region
{
unsigned int priority;
boost::interprocess::message_queue::size_type recvdSize;
std::unique_ptr<RegionBlock[]> blocks = tools::make_unique<RegionBlock[]>(fAckBunchSize);
std::unique_ptr<RegionBlock[]> blocks = std::make_unique<RegionBlock[]>(fAckBunchSize);
std::vector<fair::mq::RegionBlock> result;
result.reserve(fAckBunchSize);
@@ -276,8 +272,6 @@ struct Region
RegionBulkCallback fBulkCallback;
};
} // namespace shmem
} // namespace mq
} // namespace fair
} // namespace fair::mq::shmem
#endif /* FAIR_MQ_SHMEM_REGION_H_ */

View File

@@ -15,19 +15,16 @@
#include <FairMQSocket.h>
#include <FairMQMessage.h>
#include <FairMQLogger.h>
#include <fairmq/Tools.h>
#include <fairmq/tools/Strings.h>
#include <zmq.h>
#include <atomic>
#include <memory> // make_unique
class FairMQTransportFactory;
namespace fair
{
namespace mq
{
namespace shmem
namespace fair::mq::shmem
{
struct ZMsg
@@ -148,10 +145,10 @@ class Socket final : public fair::mq::Socket
{
if (zmq_errno() == ETERM) {
LOG(debug) << "Terminating socket " << fId;
return static_cast<int>(TransferResult::error);
return static_cast<int>(TransferCode::error);
} else {
LOG(error) << "Failed transfer on socket " << fId << ", reason: " << zmq_strerror(errno);
return static_cast<int>(TransferResult::error);
return static_cast<int>(TransferCode::error);
}
}
@@ -177,18 +174,18 @@ class Socket final : public fair::mq::Socket
return size;
} else if (zmq_errno() == EAGAIN || zmq_errno() == EINTR) {
if (fManager.Interrupted()) {
return static_cast<int>(TransferResult::interrupted);
return static_cast<int>(TransferCode::interrupted);
} else if (ShouldRetry(flags, timeout, elapsed)) {
continue;
} else {
return static_cast<int>(TransferResult::timeout);
return static_cast<int>(TransferCode::timeout);
}
} else {
return HandleErrors();
}
}
return static_cast<int>(TransferResult::error);
return static_cast<int>(TransferCode::error);
}
int64_t Receive(MessagePtr& msg, const int timeout = -1) override
@@ -222,11 +219,11 @@ class Socket final : public fair::mq::Socket
return size;
} else if (zmq_errno() == EAGAIN || zmq_errno() == EINTR) {
if (fManager.Interrupted()) {
return static_cast<int>(TransferResult::interrupted);
return static_cast<int>(TransferCode::interrupted);
} else if (ShouldRetry(flags, timeout, elapsed)) {
continue;
} else {
return static_cast<int>(TransferResult::timeout);
return static_cast<int>(TransferCode::timeout);
}
} else {
return HandleErrors();
@@ -273,18 +270,18 @@ class Socket final : public fair::mq::Socket
return totalSize;
} else if (zmq_errno() == EAGAIN || zmq_errno() == EINTR) {
if (fManager.Interrupted()) {
return static_cast<int>(TransferResult::interrupted);
return static_cast<int>(TransferCode::interrupted);
} else if (ShouldRetry(flags, timeout, elapsed)) {
continue;
} else {
return static_cast<int>(TransferResult::timeout);
return static_cast<int>(TransferCode::timeout);
}
} else {
return HandleErrors();
}
}
return static_cast<int>(TransferResult::error);
return static_cast<int>(TransferCode::error);
}
int64_t Receive(std::vector<MessagePtr>& msgVec, const int timeout = -1) override
@@ -317,7 +314,7 @@ class Socket final : public fair::mq::Socket
for (size_t m = 0; m < numMessages; m++) {
// create new message (part)
msgVec.emplace_back(tools::make_unique<Message>(fManager, hdrVec[m], GetTransport()));
msgVec.emplace_back(std::make_unique<Message>(fManager, hdrVec[m], GetTransport()));
Message* shmMsg = static_cast<Message*>(msgVec.back().get());
totalSize += shmMsg->GetSize();
}
@@ -329,18 +326,18 @@ class Socket final : public fair::mq::Socket
return totalSize;
} else if (zmq_errno() == EAGAIN || zmq_errno() == EINTR) {
if (fManager.Interrupted()) {
return static_cast<int>(TransferResult::interrupted);
return static_cast<int>(TransferCode::interrupted);
} else if (ShouldRetry(flags, timeout, elapsed)) {
continue;
} else {
return static_cast<int>(TransferResult::timeout);
return static_cast<int>(TransferCode::timeout);
}
} else {
return HandleErrors();
}
}
return static_cast<int>(TransferResult::error);
return static_cast<int>(TransferCode::error);
}
void* GetSocket() const { return fSocket; }
@@ -385,8 +382,7 @@ class Socket final : public fair::mq::Socket
{
size_t eventsSize = sizeof(uint32_t);
if (zmq_getsockopt(fSocket, ZMQ_EVENTS, events, &eventsSize) < 0) {
throw SocketError(
tools::ToString("failed setting ZMQ_EVENTS, reason: ", zmq_strerror(errno)));
throw SocketError(tools::ToString("failed setting ZMQ_EVENTS, reason: ", zmq_strerror(errno)));
}
}
@@ -524,8 +520,6 @@ class Socket final : public fair::mq::Socket
int fTimeout;
};
}
}
}
} // namespace fair::mq::shmem
#endif /* FAIR_MQ_SHMEM_SOCKET_H_ */

View File

@@ -19,21 +19,17 @@
#include <FairMQTransportFactory.h>
#include <fairmq/ProgOptions.h>
#include <FairMQLogger.h>
#include <fairmq/Tools.h>
#include <fairmq/tools/Strings.h>
#include <boost/version.hpp>
#include <zmq.h>
#include <memory> // unique_ptr
#include <memory> // unique_ptr, make_unique
#include <string>
#include <vector>
namespace fair
{
namespace mq
{
namespace shmem
namespace fair::mq::shmem
{
class TransportFactory final : public fair::mq::TransportFactory
@@ -86,7 +82,7 @@ class TransportFactory final : public fair::mq::TransportFactory
LOG(error) << "failed configuring context, reason: " << zmq_strerror(errno);
}
fManager = tools::make_unique<Manager>(fShmId, fDeviceId, segmentSize, config);
fManager = std::make_unique<Manager>(fShmId, fDeviceId, segmentSize, config);
} catch (boost::interprocess::interprocess_exception& e) {
LOG(error) << "Could not initialize shared memory transport: " << e.what();
throw std::runtime_error(tools::ToString("Could not initialize shared memory transport: ", e.what()));
@@ -98,52 +94,52 @@ class TransportFactory final : public fair::mq::TransportFactory
MessagePtr CreateMessage() override
{
return tools::make_unique<Message>(*fManager, this);
return std::make_unique<Message>(*fManager, this);
}
MessagePtr CreateMessage(Alignment alignment) override
{
return tools::make_unique<Message>(*fManager, alignment, this);
return std::make_unique<Message>(*fManager, alignment, this);
}
MessagePtr CreateMessage(const size_t size) override
{
return tools::make_unique<Message>(*fManager, size, this);
return std::make_unique<Message>(*fManager, size, this);
}
MessagePtr CreateMessage(const size_t size, Alignment alignment) override
{
return tools::make_unique<Message>(*fManager, size, alignment, this);
return std::make_unique<Message>(*fManager, size, alignment, this);
}
MessagePtr CreateMessage(void* data, const size_t size, fairmq_free_fn* ffn, void* hint = nullptr) override
{
return tools::make_unique<Message>(*fManager, data, size, ffn, hint, this);
return std::make_unique<Message>(*fManager, data, size, ffn, hint, this);
}
MessagePtr CreateMessage(UnmanagedRegionPtr& region, void* data, const size_t size, void* hint = 0) override
{
return tools::make_unique<Message>(*fManager, region, data, size, hint, this);
return std::make_unique<Message>(*fManager, region, data, size, hint, this);
}
SocketPtr CreateSocket(const std::string& type, const std::string& name) override
{
return tools::make_unique<Socket>(*fManager, type, name, GetId(), fZmqCtx, this);
return std::make_unique<Socket>(*fManager, type, name, GetId(), fZmqCtx, this);
}
PollerPtr CreatePoller(const std::vector<FairMQChannel>& channels) const override
{
return tools::make_unique<Poller>(channels);
return std::make_unique<Poller>(channels);
}
PollerPtr CreatePoller(const std::vector<FairMQChannel*>& channels) const override
{
return tools::make_unique<Poller>(channels);
return std::make_unique<Poller>(channels);
}
PollerPtr CreatePoller(const std::unordered_map<std::string, std::vector<FairMQChannel>>& channelsMap, const std::vector<std::string>& channelList) const override
{
return tools::make_unique<Poller>(channelsMap, channelList);
return std::make_unique<Poller>(channelsMap, channelList);
}
UnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, RegionCallback callback = nullptr, const std::string& path = "", int flags = 0) override
@@ -168,7 +164,7 @@ class TransportFactory final : public fair::mq::TransportFactory
UnmanagedRegionPtr CreateUnmanagedRegion(const size_t size, int64_t userFlags, RegionCallback callback, RegionBulkCallback bulkCallback, const std::string& path, int flags)
{
return tools::make_unique<UnmanagedRegion>(*fManager, size, userFlags, callback, bulkCallback, path, flags, this);
return std::make_unique<UnmanagedRegion>(*fManager, size, userFlags, callback, bulkCallback, path, flags, this);
}
void SubscribeToRegionEvents(RegionEventCallback callback) override { fManager->SubscribeToRegionEvents(callback); }
@@ -210,8 +206,6 @@ class TransportFactory final : public fair::mq::TransportFactory
std::unique_ptr<Manager> fManager;
};
} // namespace shmem
} // namespace mq
} // namespace fair
} // namespace fair::mq::shmem
#endif /* FAIR_MQ_SHMEM_TRANSPORTFACTORY_H_ */

View File

@@ -20,11 +20,7 @@
#include <cstddef> // size_t
#include <string>
namespace fair
{
namespace mq
{
namespace shmem
namespace fair::mq::shmem
{
class Message;
@@ -68,8 +64,6 @@ class UnmanagedRegion final : public fair::mq::UnmanagedRegion
uint16_t fRegionId;
};
}
}
}
} // namespace fair::mq::shmem
#endif /* FAIR_MQ_SHMEM_UNMANAGEDREGION_H_ */

View File

@@ -125,13 +125,12 @@ int main(int argc, char** argv)
}
cout << "Starting shared memory monitor for session: \"" << sessionName << "\" (shmId: " << shmId << ")..." << endl;
if (viewOnly && !interactive) {
cout << "running in non-interactive view-only mode, outputting with interval of " << intervalInMS << "ms. (change with --interval), press ctrl+C to exit." << endl;
}
Monitor monitor(shmId, selfDestruct, interactive, viewOnly, timeoutInMS, intervalInMS, runAsDaemon, cleanOnExit);
monitor.CatchSignals();
if (interactive || !viewOnly) {
monitor.CatchSignals();
}
monitor.Run();
} catch (Monitor::DaemonPresent& dp) {
return 0;

View File

@@ -9,45 +9,11 @@
#ifndef FAIR_MQ_TOOLS_CPPSTL_H
#define FAIR_MQ_TOOLS_CPPSTL_H
#include <functional>
#include <memory>
#include <type_traits>
namespace fair
{
namespace mq
{
namespace tools
namespace fair::mq::tools
{
// make_unique implementation, until C++14 is default
template<typename T, typename ...Args>
std::unique_ptr<T> make_unique(Args&& ...args)
{
return std::unique_ptr<T>(new T(std::forward<Args>(args)...));
}
// make_unique implementation (array variant), until C++14 is default
template<typename T>
std::unique_ptr<T> make_unique(std::size_t size)
{
return std::unique_ptr<T>(new typename std::remove_extent<T>::type[size]());
}
// provide an enum hasher to compensate std::hash not supporting enums in C++11
template<typename Enum>
struct HashEnum
{
auto operator()(const Enum& e) const noexcept
-> typename std::enable_if<std::is_enum<Enum>::value, std::size_t>::type
{
using _type = typename std::underlying_type<Enum>::type;
return std::hash<_type>{}(static_cast<_type>(e));
}
};
} /* namespace tools */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::tools
#endif /* FAIR_MQ_TOOLS_CPPSTL_H */

View File

@@ -11,9 +11,8 @@
#include "Strings.h"
namespace fair {
namespace mq {
namespace tools {
namespace fair::mq::tools
{
template<typename Tag, int Max>
struct InstanceLimiter
@@ -50,8 +49,6 @@ struct InstanceLimiter
template<typename Tag, int Max>
int InstanceLimiter<Tag, Max>::fCount(0);
} /* namespace tools */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::tools
#endif /* FAIR_MQ_TOOLS_INSTANCELIMIT_H */

View File

@@ -33,11 +33,7 @@
using namespace std;
namespace fair
{
namespace mq
{
namespace tools
namespace fair::mq::tools
{
// returns a map with network interface names as keys and their IP addresses as values
@@ -184,6 +180,4 @@ string getIpFromHostname(const string& hostname, boost::asio::io_service& ios)
}
}
} /* namespace tools */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::tools

View File

@@ -25,11 +25,7 @@ using io_service = class io_context;
} // namespace asio
} // namespace boost
namespace fair
{
namespace mq
{
namespace tools
namespace fair::mq::tools
{
struct DefaultRouteDetectionError : std::runtime_error { using std::runtime_error::runtime_error; };
@@ -47,8 +43,6 @@ std::string getIpFromHostname(const std::string& hostname);
std::string getIpFromHostname(const std::string& hostname, boost::asio::io_service& ios);
} /* namespace tools */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::tools
#endif /* FAIR_MQ_TOOLS_NETWORK_H */

View File

@@ -44,11 +44,7 @@ class LinePrinter
const string fPrefix;
};
namespace fair
{
namespace mq
{
namespace tools
namespace fair::mq::tools
{
/**
@@ -167,6 +163,4 @@ execute_result execute(const string& cmd, const string& prefix, const string& in
return result;
}
} /* namespace tools */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::tools

View File

@@ -11,11 +11,7 @@
#include <string>
namespace fair
{
namespace mq
{
namespace tools
namespace fair::mq::tools
{
/**
@@ -41,8 +37,6 @@ execute_result execute(const std::string& cmd,
const std::string& input = "",
int sig = -1);
} /* namespace tools */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::tools
#endif /* FAIR_MQ_TOOLS_PROCESS_H */

View File

@@ -16,11 +16,7 @@
#include <thread>
#include <chrono>
namespace fair
{
namespace mq
{
namespace tools
namespace fair::mq::tools
{
/**
@@ -136,8 +132,6 @@ class RateLimiter
int skip_check_count = 1;
};
} /* namespace tools */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::tools
#endif // FAIR_MQ_TOOLS_RATELIMIT_H

View File

@@ -8,9 +8,8 @@
#include "Semaphore.h"
namespace fair {
namespace mq {
namespace tools {
namespace fair::mq::tools
{
Semaphore::Semaphore()
: Semaphore(0)
@@ -68,6 +67,4 @@ auto SharedSemaphore::GetCount() const -> std::size_t
return fSemaphore->GetCount();
}
} /* namespace tools */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::tools

View File

@@ -15,9 +15,8 @@
#include <memory>
#include <mutex>
namespace fair {
namespace mq {
namespace tools {
namespace fair::mq::tools
{
/**
* @struct Semaphore Semaphore.h <fairmq/tools/Semaphore.h>
@@ -55,8 +54,6 @@ private:
std::shared_ptr<Semaphore> fSemaphore;
};
} /* namespace tools */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::tools
#endif /* FAIR_MQ_TOOLS_SEMAPHORE_H */

View File

@@ -15,11 +15,7 @@
#include <string>
#include <vector>
namespace fair
{
namespace mq
{
namespace tools
namespace fair::mq::tools
{
/// @brief concatenates a variable number of args with the << operator via a stringstream
@@ -43,8 +39,6 @@ inline auto ToStrVector(const int argc, char*const* argv, const bool dropProgram
}
}
} /* namespace tools */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::tools
#endif /* FAIR_MQ_TOOLS_STRINGS_H */

View File

@@ -19,11 +19,7 @@
using namespace std;
namespace fair
{
namespace mq
{
namespace tools
namespace fair::mq::tools
{
// generates UUID string
@@ -43,6 +39,4 @@ size_t UuidHash()
return uuid_hasher(u);
}
} /* namespace tools */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::tools

View File

@@ -11,11 +11,7 @@
#include <string>
namespace fair
{
namespace mq
{
namespace tools
namespace fair::mq::tools
{
// generates UUID string
@@ -24,8 +20,6 @@ std::string Uuid();
// generates UUID and returns its hash
std::size_t UuidHash();
} /* namespace tools */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::tools
#endif /* FAIR_MQ_TOOLS_UNIQUE_H */

View File

@@ -12,11 +12,7 @@
#include <ostream>
#include <tuple>
namespace fair
{
namespace mq
{
namespace tools
namespace fair::mq::tools
{
struct Version
@@ -32,8 +28,6 @@ struct Version
friend auto operator<<(std::ostream& os, const Version& v) -> std::ostream& { return os << v.fkMajor << "." << v.fkMinor << "." << v.fkPatch; }
};
} /* namespace tools */
} /* namespace mq */
} /* namespace fair */
} // namespace fair::mq::tools
#endif /* FAIR_MQ_TOOLS_VERSION_H */

Some files were not shown because too many files have changed in this diff Show More