Various changes

CMakeLists.txt

@@ -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 "  ")
 ################################################################################

fairmq/DeviceRunner.cxx

@@ -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);
             }

fairmq/FairMQChannel.h

@@ -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);

fairmq/FairMQDevice.cxx

@@ -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;
@@ -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

fairmq/FairMQDevice.h

@@ -54,38 +54,6 @@ 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 +96,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 +107,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 +118,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 +129,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 +417,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 +427,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 +439,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(); }
 

fairmq/FairMQMessage.h

@@ -39,7 +39,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;

fairmq/FairMQSocket.h

@@ -24,8 +24,9 @@ namespace fair
 namespace mq
 {
 
-enum class TransferResult : int
+enum class TransferCode : int
 {
+    success = 0,
     error = -1,
     timeout = -2,
     interrupted = -3

fairmq/devices/FairMQBenchmarkSampler.h

@@ -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;