Fix FI_EINTR (interrupted) corrupting internal state for no reason

lib/fabrics/include/mxl/fabrics.h

@@ -227,7 +227,8 @@ extern "C"
      * \param in_target A valid fabrics target
      * \param out_grainIndex The index of the grain that was written, if any.
      * \param in_timeoutMs How long should we wait for the grain (in milliseconds)
-     * \return The result code. MXL_ERR_NOT_READY if no grain was available before the timeout. \see mxlStatus
+     * \return The result code. MXL_ERR_NOT_READY if no grain was available before the timeout. Some providers return MXL_ERR_INTERRUPTED when the
+     * blocking read is interrupted by a POSIX signal. \see mxlStatus
      */
     MXL_EXPORT
     mxlStatus mxlFabricsTargetReadGrain(mxlFabricsTarget in_target, uint16_t in_timeoutMs, uint64_t* out_entryIndex);
@@ -346,7 +347,7 @@ extern "C"
      * \param in_initiator The initiator that should make progress.
      * \param in_timeoutMs The maximum time to wait for progress to be made (in milliseconds).
      * \return The result code. Returns MXL_ERR_NOT_READY if there is still progress to be made and not all operations have completed before the
-     * timeout.
+     * timeout. Some providers return MXL_ERR_INTERRUPTED if the operation is interrupted by the arrival of a POSIX signal.
      */
     MXL_EXPORT
     mxlStatus mxlFabricsInitiatorMakeProgressBlocking(mxlFabricsInitiator in_initiator, uint16_t in_timeoutMs);

lib/fabrics/ofi/src/fabrics.cpp

@@ -23,6 +23,7 @@
 #include "internal/Target.hpp"
 #include "internal/TargetInfo.hpp"
 #include "mxl/platform.h"
+#include "VariantUtils.hpp"
 
 namespace ofi = mxl::lib::fabrics::ofi;
 
@@ -238,13 +239,24 @@ mxlStatus mxlFabricsTargetReadGrainNonBlocking(mxlFabricsTarget in_target, uint6
     return ofi::try_run(
         [&]()
         {
-            auto res = ofi::TargetWrapper::fromAPI(in_target)->readGrain();
-            if (!res)
+            auto const result = ofi::TargetWrapper::fromAPI(in_target)->read();
+            if (!result)
             {
                 return MXL_ERR_NOT_READY;
             }
 
-            *out_grainIndex = res->grainIndex;
+            if (std::holds_alternative<ofi::Target::Interrupted>(*result))
+            {
+                return MXL_ERR_INTERRUPTED;
+            }
+
+            auto const grainResult = std::get_if<ofi::Target::GrainReadResult>(&*result);
+            if (grainResult == nullptr)
+            {
+                return MXL_ERR_INVALID_STATE;
+            }
+
+            *out_grainIndex = grainResult->grainIndex;
             return MXL_STATUS_OK;
         },
         "Failed to try for new grain");
@@ -261,13 +273,24 @@ mxlStatus mxlFabricsTargetReadGrain(mxlFabricsTarget in_target, uint16_t in_time
     return ofi::try_run(
         [&]()
         {
-            auto res = ofi::TargetWrapper::fromAPI(in_target)->readGrainBlocking(std::chrono::milliseconds(in_timeoutMs));
-            if (!res)
+            auto const result = ofi::TargetWrapper::fromAPI(in_target)->readBlocking(std::chrono::milliseconds(in_timeoutMs));
+            if (!result)
             {
                 return MXL_ERR_NOT_READY;
             }
 
-            *out_grainIndex = res->grainIndex;
+            if (std::holds_alternative<ofi::Target::Interrupted>(*result))
+            {
+                return MXL_ERR_INTERRUPTED;
+            }
+
+            auto const grainResult = std::get_if<ofi::Target::GrainReadResult>(&*result);
+            if (grainResult == nullptr)
+            {
+                return MXL_ERR_INVALID_STATE;
+            }
+
+            *out_grainIndex = grainResult->grainIndex;
             return MXL_STATUS_OK;
         },
         "Failed to wait for new grain");
@@ -284,14 +307,25 @@ mxlStatus mxlFabricsTargetReadSamplesNonBlocking(mxlFabricsTarget in_target, uin
     return ofi::try_run(
         [&]()
         {
-            auto res = ofi::TargetWrapper::fromAPI(in_target)->readSamples();
-            if (!res)
+            auto const result = ofi::TargetWrapper::fromAPI(in_target)->read();
+            if (!result)
             {
                 return MXL_ERR_NOT_READY;
             }
 
-            *out_headIndex = res->headIndex;
-            *out_count = res->count;
+            if (std::holds_alternative<ofi::Target::Interrupted>(*result))
+            {
+                return MXL_ERR_INTERRUPTED;
+            }
+
+            auto const sampleResult = std::get_if<ofi::Target::SampleReadResult>(&*result);
+            if (sampleResult == nullptr)
+            {
+                return MXL_ERR_INVALID_STATE;
+            }
+
+            *out_headIndex = sampleResult->headIndex;
+            *out_count = sampleResult->count;
             return MXL_STATUS_OK;
         },
         "Failed to try for new samples");
@@ -308,14 +342,25 @@ mxlStatus mxlFabricsTargetReadSamples(mxlFabricsTarget in_target, uint16_t in_ti
     return ofi::try_run(
         [&]()
         {
-            auto res = ofi::TargetWrapper::fromAPI(in_target)->readSamplesBlocking(std::chrono::milliseconds(in_timeoutMs));
-            if (!res)
+            auto const result = ofi::TargetWrapper::fromAPI(in_target)->readBlocking(std::chrono::milliseconds(in_timeoutMs));
+            if (!result)
             {
                 return MXL_ERR_NOT_READY;
             }
 
-            *out_headIndex = res->headIndex;
-            *out_count = res->count;
+            if (std::holds_alternative<ofi::Target::Interrupted>(*result))
+            {
+                return MXL_ERR_INTERRUPTED;
+            }
+
+            auto const sampleResult = std::get_if<ofi::Target::SampleReadResult>(&*result);
+            if (sampleResult == nullptr)
+            {
+                return MXL_ERR_INVALID_STATE;
+            }
+
+            *out_headIndex = sampleResult->headIndex;
+            *out_count = sampleResult->count;
             return MXL_STATUS_OK;
         },
         "Failed to wait for new samples");
@@ -462,12 +507,14 @@ mxlStatus mxlFabricsInitiatorMakeProgressNonBlocking(mxlFabricsInitiator in_init
     return ofi::try_run(
         [&]()
         {
-            if (ofi::InitiatorWrapper::fromAPI(in_initiator)->makeProgress())
-            {
-                return MXL_ERR_NOT_READY;
-            }
-
-            return MXL_STATUS_OK;
+            auto const result = ofi::InitiatorWrapper::fromAPI(in_initiator)->makeProgress();
+            return std::visit(
+                ofi::overloaded{
+                    [](ofi::Initiator::Ready) { return MXL_STATUS_OK; },
+                    [](ofi::Initiator::NotReady) { return MXL_ERR_NOT_READY; },
+                    [](ofi::Initiator::Interrupted) { return MXL_ERR_INTERRUPTED; },
+                },
+                result);
         },
         "Failed to make progress in the initiator");
 }
@@ -483,12 +530,14 @@ mxlStatus mxlFabricsInitiatorMakeProgressBlocking(mxlFabricsInitiator in_initiat
     return ofi::try_run(
         [&]()
         {
-            if (ofi::InitiatorWrapper::fromAPI(in_initiator)->makeProgressBlocking(std::chrono::milliseconds(in_timeoutMs)))
-            {
-                return MXL_ERR_NOT_READY;
-            }
-
-            return MXL_STATUS_OK;
+            auto const result = ofi::InitiatorWrapper::fromAPI(in_initiator)->makeProgressBlocking(std::chrono::milliseconds(in_timeoutMs));
+            return std::visit(
+                ofi::overloaded{
+                    [](ofi::Initiator::Ready) { return MXL_STATUS_OK; },
+                    [](ofi::Initiator::NotReady) { return MXL_ERR_NOT_READY; },
+                    [](ofi::Initiator::Interrupted) { return MXL_ERR_INTERRUPTED; },
+                },
+                result);
         },
         "Failed to make progress in the initiator");
 }

lib/fabrics/ofi/src/internal/Exception.cpp

@@ -33,6 +33,11 @@ namespace mxl::lib::fabrics::ofi
         return _fiErrno;
     }
 
+    bool FabricException::isInterrupted() const noexcept
+    {
+        return _fiErrno == -FI_EINTR;
+    }
+
     mxlStatus mxlStatusFromFiErrno(int fiErrno)
     {
         switch (fiErrno)

lib/fabrics/ofi/src/internal/Exception.hpp

@@ -138,6 +138,9 @@ namespace mxl::lib::fabrics::ofi
         [[nodiscard]]
         int fiErrno() const noexcept;
 
+        [[nodiscard]]
+        bool isInterrupted() const noexcept;
+
     private:
         int _fiErrno;
     };

lib/fabrics/ofi/src/internal/Initiator.cpp

@@ -89,7 +89,7 @@ namespace mxl::lib::fabrics::ofi
         _inner->transferSamples(headIndex, count);
     }
 
-    bool InitiatorWrapper::makeProgress()
+    Initiator::MakeProgressResult InitiatorWrapper::makeProgress()
     {
         if (!_inner)
         {
@@ -99,7 +99,7 @@ namespace mxl::lib::fabrics::ofi
         return _inner->makeProgress();
     }
 
-    bool InitiatorWrapper::makeProgressBlocking(std::chrono::steady_clock::duration timeout)
+    Initiator::MakeProgressResult InitiatorWrapper::makeProgressBlocking(std::chrono::steady_clock::duration timeout)
     {
         if (!_inner)
         {

lib/fabrics/ofi/src/internal/Initiator.hpp

@@ -15,6 +15,18 @@ namespace mxl::lib::fabrics::ofi
      */
     class Initiator
     {
+    public:
+        struct Ready
+        {};
+
+        struct NotReady
+        {};
+
+        struct Interrupted
+        {};
+
+        using MakeProgressResult = std::variant<Ready, NotReady, Interrupted>;
+
     public:
         virtual ~Initiator() = default;
 
@@ -69,13 +81,13 @@ namespace mxl::lib::fabrics::ofi
          *
          * This is the non-blocking version of the progress function.
          */
-        virtual bool makeProgress() = 0;
+        virtual MakeProgressResult makeProgress() = 0;
 
         /** \brief Attempts to progress execution, including connection management and data operations.
          *
          * This is the blocking version of the progress function.
          */
-        virtual bool makeProgressBlocking(std::chrono::steady_clock::duration) = 0;
+        virtual MakeProgressResult makeProgressBlocking(std::chrono::steady_clock::duration) = 0;
 
         /** \brief Shut down the initiator gracefully.
          *
@@ -141,11 +153,11 @@ namespace mxl::lib::fabrics::ofi
 
         /** \copydoc Initiator::makeProgress()
          */
-        bool makeProgress();
+        Initiator::MakeProgressResult makeProgress();
 
         /** \copydoc Initiator::makeProgressBlocking()
          */
-        bool makeProgressBlocking(std::chrono::steady_clock::duration);
+        Initiator::MakeProgressResult makeProgressBlocking(std::chrono::steady_clock::duration);
 
     private:
         std::unique_ptr<Initiator> _inner; /**< The underlying initiator implementation. */