diff --git a/lib/fabrics/include/mxl/fabrics.h b/lib/fabrics/include/mxl/fabrics.h index 47a84f5fa..cc18fbe63 100644 --- a/lib/fabrics/include/mxl/fabrics.h +++ b/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); diff --git a/lib/fabrics/ofi/src/fabrics.cpp b/lib/fabrics/ofi/src/fabrics.cpp index b9ea9cd25..25b522691 100644 --- a/lib/fabrics/ofi/src/fabrics.cpp +++ b/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(*result)) + { + return MXL_ERR_INTERRUPTED; + } + + auto const grainResult = std::get_if(&*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(*result)) + { + return MXL_ERR_INTERRUPTED; + } + + auto const grainResult = std::get_if(&*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(*result)) + { + return MXL_ERR_INTERRUPTED; + } + + auto const sampleResult = std::get_if(&*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(*result)) + { + return MXL_ERR_INTERRUPTED; + } + + auto const sampleResult = std::get_if(&*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"); } diff --git a/lib/fabrics/ofi/src/internal/Exception.cpp b/lib/fabrics/ofi/src/internal/Exception.cpp index 9b7ad72c1..91bd66b46 100644 --- a/lib/fabrics/ofi/src/internal/Exception.cpp +++ b/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) diff --git a/lib/fabrics/ofi/src/internal/Exception.hpp b/lib/fabrics/ofi/src/internal/Exception.hpp index 8944f47f9..32c9fb37e 100644 --- a/lib/fabrics/ofi/src/internal/Exception.hpp +++ b/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; }; diff --git a/lib/fabrics/ofi/src/internal/Initiator.cpp b/lib/fabrics/ofi/src/internal/Initiator.cpp index 80a692104..a01405be4 100644 --- a/lib/fabrics/ofi/src/internal/Initiator.cpp +++ b/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) { diff --git a/lib/fabrics/ofi/src/internal/Initiator.hpp b/lib/fabrics/ofi/src/internal/Initiator.hpp index 027ae3e9f..4b3621531 100644 --- a/lib/fabrics/ofi/src/internal/Initiator.hpp +++ b/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; + 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 _inner; /**< The underlying initiator implementation. */ diff --git a/lib/fabrics/ofi/src/internal/RCInitiator.cpp b/lib/fabrics/ofi/src/internal/RCInitiator.cpp index 00b3c297f..69586990e 100644 --- a/lib/fabrics/ofi/src/internal/RCInitiator.cpp +++ b/lib/fabrics/ofi/src/internal/RCInitiator.cpp @@ -358,18 +358,18 @@ namespace mxl::lib::fabrics::ofi } } - bool RCInitiator::hasPendingWork() const noexcept + Initiator::MakeProgressResult RCInitiator::afterProgressResult() const noexcept { // Check if any of the targets have pending work. for (auto& [_, target] : _targets) { if (target.hasPendingWork()) { - return true; + return Initiator::NotReady{}; } } - return false; + return Initiator::Ready{}; } bool RCInitiator::hasTarget() const noexcept @@ -392,7 +392,7 @@ namespace mxl::lib::fabrics::ofi std::erase_if(_targets, [](auto const& item) { return item.second.canEvict(); }); } - void RCInitiator::blockOnCQ(std::chrono::steady_clock::duration timeout) + Initiator::MakeProgressResult RCInitiator::blockOnCQ(std::chrono::steady_clock::duration timeout) { // A zero timeout would cause the queue to block indefinetly, which // is not our documented behaviour. @@ -400,30 +400,45 @@ namespace mxl::lib::fabrics::ofi { // So just behave exactly like the non-blocking variant. makeProgress(); - return; + return afterProgressResult(); } for (;;) { - auto completion = _cq->readBlocking(timeout); - if (!completion) + try { - // No completion available, if we were flushing any endpoint, transition their state to done. - for (auto& [_, target] : _targets) + auto completion = _cq->readBlocking(timeout); + if (!completion) { - target.terminate(); + // No completion available, if we were flushing any endpoint, transition their state to done. + for (auto& [_, target] : _targets) + { + target.terminate(); + } + + return afterProgressResult(); } - return; - } - // Find the endpoint that this completion was generated from - auto ep = _targets.find(Endpoint::idFromToken(completion->token())); - if (ep == _targets.end()) - { - MXL_WARN("Received completion for an unknown endpoint"); + // Find the endpoint that this completion was generated from + auto ep = _targets.find(Endpoint::idFromToken(completion->token())); + if (ep == _targets.end()) + { + MXL_WARN("Received completion for an unknown endpoint"); + } + + ep->second.consume(*completion); + + return afterProgressResult(); } + catch (FabricException const& ex) + { + if (ex.isInterrupted()) + { + return Initiator::Interrupted{}; + } - return ep->second.consume(*completion); + throw; + } } } @@ -478,7 +493,7 @@ namespace mxl::lib::fabrics::ofi } } - bool RCInitiator::makeProgress() + Initiator::MakeProgressResult RCInitiator::makeProgress() { if (!hasTarget()) { @@ -495,18 +510,17 @@ namespace mxl::lib::fabrics::ofi // Evict any peers that are dead and no longer will make progress. evictDeadEndpoints(); - return hasPendingWork(); + return afterProgressResult(); } - bool RCInitiator::makeProgressBlocking(std::chrono::steady_clock::duration timeout) + Initiator::MakeProgressResult RCInitiator::makeProgressBlocking(std::chrono::steady_clock::duration timeout) { // If the timeout is less than our maintainance interval, just check all the queues once, execute all maintainance tasks once // and block on the completion queue for the rest of the time. if (timeout < EQPollInterval) { makeProgress(); - blockOnCQ(timeout); - return hasPendingWork(); + return blockOnCQ(timeout); } auto deadline = std::chrono::steady_clock::now() + timeout; @@ -514,9 +528,9 @@ namespace mxl::lib::fabrics::ofi for (;;) { // Poll all queues, execute all maintainance actions - if (!makeProgress()) + if (std::holds_alternative(makeProgress())) { - return false; + return Initiator::Ready{}; } // Calculate the remaining time until the user wants the blocking function to return. If there is no time left @@ -524,14 +538,18 @@ namespace mxl::lib::fabrics::ofi auto timeUntilDeadline = std::chrono::duration_cast(deadline - std::chrono::steady_clock::now()); if (timeUntilDeadline <= decltype(timeUntilDeadline){0}) { - return hasPendingWork(); + return afterProgressResult(); } // Block on the completion queue until a completion arrives, or the interval timeout occurs. - blockOnCQ(std::min(EQPollInterval, timeUntilDeadline)); - } + auto const res = blockOnCQ(std::min(EQPollInterval, timeUntilDeadline)); + if (std::holds_alternative(res)) + { + continue; + } - return hasPendingWork(); + return res; + } } void RCInitiator::shutdown() diff --git a/lib/fabrics/ofi/src/internal/RCInitiator.hpp b/lib/fabrics/ofi/src/internal/RCInitiator.hpp index 4566d1fd2..3def2987c 100644 --- a/lib/fabrics/ofi/src/internal/RCInitiator.hpp +++ b/lib/fabrics/ofi/src/internal/RCInitiator.hpp @@ -195,11 +195,11 @@ namespace mxl::lib::fabrics::ofi /** \copydoc Initiator::makeProgress() */ - virtual bool makeProgress() final; + virtual Initiator::MakeProgressResult makeProgress() final; /** \copydoc Initiator::makeProgressBlocking() */ - virtual bool makeProgressBlocking(std::chrono::steady_clock::duration) final; + virtual Initiator::MakeProgressResult makeProgressBlocking(std::chrono::steady_clock::duration) final; virtual void shutdown() final; @@ -207,7 +207,7 @@ namespace mxl::lib::fabrics::ofi /** \brief Returns true if any of the endpoints contained in this initiator have pending work. */ [[nodiscard]] - bool hasPendingWork() const noexcept; + Initiator::MakeProgressResult afterProgressResult() const noexcept; /** \brief Returns true if the initiator has at least 1 target added no matter what the state is. */ @@ -229,7 +229,8 @@ namespace mxl::lib::fabrics::ofi /** \brief Block on the completion queue with a timeout. */ - void blockOnCQ(std::chrono::steady_clock::duration timeout); + [[nodiscard]] + Initiator::MakeProgressResult blockOnCQ(std::chrono::steady_clock::duration timeout); /** \brief Poll the completion queue and process the events until the queue is empty. */ diff --git a/lib/fabrics/ofi/src/internal/RCTarget.cpp b/lib/fabrics/ofi/src/internal/RCTarget.cpp index 1c8defbb2..c89076ed3 100644 --- a/lib/fabrics/ofi/src/internal/RCTarget.cpp +++ b/lib/fabrics/ofi/src/internal/RCTarget.cpp @@ -63,60 +63,14 @@ namespace mxl::lib::fabrics::ofi , _state(WaitForConnectionRequest{std::move(pep)}) {} - std::optional RCTarget::readGrain() + std::optional RCTarget::read() { - if (!_proto->canReadGrains()) - { - throw Exception::unsupportedOperation("The current protocol does not support reading grains."); - } - - if (auto res = readNext(std::chrono::steady_clock::duration::zero()); res) - { - return std::get(*res); - } - return std::nullopt; - } - - std::optional RCTarget::readGrainBlocking(std::chrono::steady_clock::duration timeout) - { - if (!_proto->canReadGrains()) - { - throw Exception::unsupportedOperation("The current protocol does not support reading grains."); - } - - if (auto res = readNext(timeout); res) - { - return std::get(*res); - } - return std::nullopt; + return readNext(std::chrono::steady_clock::duration::zero()); } - std::optional RCTarget::readSamples() + std::optional RCTarget::readBlocking(std::chrono::steady_clock::duration timeout) { - if (!_proto->canReadSamples()) - { - throw Exception::unsupportedOperation("The current protocol does not support reading samples."); - } - - if (auto res = readNext(std::chrono::steady_clock::duration::zero()); res) - { - return std::get(*res); - } - return std::nullopt; - } - - std::optional RCTarget::readSamplesBlocking(std::chrono::steady_clock::duration timeout) - { - if (!_proto->canReadSamples()) - { - throw Exception::unsupportedOperation("The current protocol does not support reading samples."); - } - - if (auto res = readNext(timeout); res) - { - return std::get(*res); - } - return std::nullopt; + return readNext(timeout); } void RCTarget::shutdown() @@ -131,67 +85,106 @@ namespace mxl::lib::fabrics::ofi overloaded{[](std::monostate) -> State { throw Exception::invalidState("Target is in an invalid state an can no longer make progress"); }, [&](WaitForConnectionRequest state) -> State { - auto event = readEventQueue(*state.pep.eventQueue(), timeout); - - // Check if the entry is available and is a connection request - if (event && event->isConnReq()) + try { - auto remoteAddr = FabricAddress::fromDestination(event->connReq().info()); - MXL_INFO("Accept connection from: {}", remoteAddr.toString()); + auto event = readEventQueue(*state.pep.eventQueue(), timeout); + + // Check if the entry is available and is a connection request + if (event && event->isConnReq()) + { + auto remoteAddr = FabricAddress::fromDestination(event->connReq().info()); + MXL_INFO("Accept connection from: {}", remoteAddr.toString()); - auto cqAttr = CompletionQueue::Attributes::defaults(); - cqAttr.size = _setupOptions.cqDepth.value_or(CompletionQueue::Attributes::DEFAULT_SIZE); - auto cq = CompletionQueue::open(_domain, cqAttr); - auto endpoint = Endpoint::create(_domain, state.pep.id(), event->connReq().info()); - endpoint.bind(cq, FI_RECV); + auto cqAttr = CompletionQueue::Attributes::defaults(); + cqAttr.size = _setupOptions.cqDepth.value_or(CompletionQueue::Attributes::DEFAULT_SIZE); + auto cq = CompletionQueue::open(_domain, cqAttr); + auto endpoint = Endpoint::create(_domain, state.pep.id(), event->connReq().info()); + endpoint.bind(cq, FI_RECV); - auto eq = EventQueue::open(_domain->fabric(), EventQueue::Attributes::defaults()); - endpoint.bind(eq); + auto eq = EventQueue::open(_domain->fabric(), EventQueue::Attributes::defaults()); + endpoint.bind(eq); - // we are now ready to accept the connection - endpoint.accept(); - MXL_DEBUG("Accepted the connection waiting for connected event notification."); + // we are now ready to accept the connection + endpoint.accept(); + MXL_DEBUG("Accepted the connection waiting for connected event notification."); - // Return the new state as the variant type - return RCTarget::WaitForConnection{std::move(endpoint)}; + // Return the new state as the variant type + return RCTarget::WaitForConnection{std::move(endpoint)}; + } + + return WaitForConnectionRequest{.pep = std::move(state.pep)}; } + catch (FabricException const& ex) + { + if (ex.isInterrupted()) + { + result = std::make_optional(); + return WaitForConnectionRequest{.pep = std::move(state.pep)}; + } - return WaitForConnectionRequest{.pep = std::move(state.pep)}; + throw; + } }, [&](WaitForConnection state) -> State { - auto event = readEventQueue(*state.ep.eventQueue(), timeout); - - if (event && event->isConnected()) + try { - MXL_INFO("Received connected event notification, now connected."); + auto event = readEventQueue(*state.ep.eventQueue(), timeout); + + if (event && event->isConnected()) + { + MXL_INFO("Received connected event notification, now connected."); - // We have a connected event, so we can transition to the connected state - auto connected = Connected{.ep = std::move(state.ep)}; + // We have a connected event, so we can transition to the connected state + auto connected = Connected{.ep = std::move(state.ep)}; - // The endpoint is now ready, initialize the protocol. - _proto->start(connected.ep); + // The endpoint is now ready, initialize the protocol. + _proto->start(connected.ep); - return connected; + return connected; + } + + return WaitForConnection{std::move(state.ep)}; } + catch (FabricException const& ex) + { + if (ex.isInterrupted()) + { + result = std::make_optional(); + return WaitForConnection{.ep = std::move(state.ep)}; + } - return WaitForConnection{std::move(state.ep)}; + throw; + } }, [&](RCTarget::Connected state) -> State { - auto [completion, event] = readEndpointQueues(state.ep, timeout); - if (event && event.value().isShutdown()) + try { - MXL_INFO("Remote endpoint has shutdown the connection. Transitioning to listening to new connection."); - return WaitForConnectionRequest{.pep = makeListener(state.ep.domain()->fabric())}; + auto [completion, event] = readEndpointQueues(state.ep, timeout); + if (event && event.value().isShutdown()) + { + MXL_INFO("Remote endpoint has shutdown the connection. Transitioning to listening to new connection."); + return WaitForConnectionRequest{.pep = makeListener(state.ep.domain()->fabric())}; + } + + if (completion) + { + result = _proto->read(state.ep, *completion); + } + + return Connected{.ep = std::move(state.ep)}; } - - if (completion) + catch (FabricException const& ex) { - result = _proto->read(state.ep, *completion); - } + if (ex.isInterrupted()) + { + result = std::make_optional(); + return Connected{.ep = std::move(state.ep)}; + } - return Connected{.ep = std::move(state.ep)}; + throw; + } }}, std::move(_state)); diff --git a/lib/fabrics/ofi/src/internal/RCTarget.hpp b/lib/fabrics/ofi/src/internal/RCTarget.hpp index 583be7950..85559e2a3 100644 --- a/lib/fabrics/ofi/src/internal/RCTarget.hpp +++ b/lib/fabrics/ofi/src/internal/RCTarget.hpp @@ -33,21 +33,8 @@ namespace mxl::lib::fabrics::ofi static std::pair, std::unique_ptr> setup(mxlFabricsTargetConfig const& config, FabricInfoView info, TargetSetupOptions const& options = {}); - /** \copydoc Target::readGrain() - */ - virtual std::optional readGrain() final; - - /** \copydoc Target::readGrainBlocking() - */ - virtual std::optional readGrainBlocking(std::chrono::steady_clock::duration timeout) final; - - /** \copydoc Target::readSamples() - */ - virtual std::optional readSamples() final; - - /** copydoc Target::readSamplesBlocking() - */ - virtual std::optional readSamplesBlocking(std::chrono::steady_clock::duration timeout) final; + virtual std::optional read(); + virtual std::optional readBlocking(std::chrono::steady_clock::duration timeout) final; /** \brief Shut down the target. */ diff --git a/lib/fabrics/ofi/src/internal/RDMInitiator.cpp b/lib/fabrics/ofi/src/internal/RDMInitiator.cpp index 07d8ca86b..15a44b6d0 100644 --- a/lib/fabrics/ofi/src/internal/RDMInitiator.cpp +++ b/lib/fabrics/ofi/src/internal/RDMInitiator.cpp @@ -218,15 +218,15 @@ namespace mxl::lib::fabrics::ofi } // makeProgress - bool RDMInitiator::makeProgress() + Initiator::MakeProgressResult RDMInitiator::makeProgress() { activateIdleEndpoints(); pollCQ(); - return hasPendingWork(); + return afterProgressResult(); } // makeProgressBlocking - bool RDMInitiator::makeProgressBlocking(std::chrono::steady_clock::duration timeout) + Initiator::MakeProgressResult RDMInitiator::makeProgressBlocking(std::chrono::steady_clock::duration timeout) { auto now = std::chrono::steady_clock::now(); activateIdleEndpoints(); @@ -235,14 +235,25 @@ namespace mxl::lib::fabrics::ofi auto remaining = timeout - elapsed; if (remaining.count() >= 0) { - blockOnCQ(remaining); + try + { + blockOnCQ(remaining); + } + catch (FabricException const& ex) + { + if (ex.isInterrupted()) + { + return Initiator::Interrupted{}; + } + throw; + } } else { pollCQ(); } - return hasPendingWork(); + return afterProgressResult(); } RDMInitiatorTarget& RDMInitiator::findRemoteByEndpoint(Endpoint::Id id) @@ -267,17 +278,17 @@ namespace mxl::lib::fabrics::ofi return it->second; } - bool RDMInitiator::hasPendingWork() const noexcept + Initiator::MakeProgressResult RDMInitiator::afterProgressResult() const noexcept { for (auto const& [_, remote] : _targets) { if (remote.hasPendingWork()) { - return true; + return Initiator::NotReady{}; } } - return false; + return Initiator::Ready{}; } void RDMInitiator::blockOnCQ(std::chrono::steady_clock::duration timeout) diff --git a/lib/fabrics/ofi/src/internal/RDMInitiator.hpp b/lib/fabrics/ofi/src/internal/RDMInitiator.hpp index ebaf5b580..fe6264862 100644 --- a/lib/fabrics/ofi/src/internal/RDMInitiator.hpp +++ b/lib/fabrics/ofi/src/internal/RDMInitiator.hpp @@ -137,11 +137,11 @@ namespace mxl::lib::fabrics::ofi /** \copydoc Initiator::makeProgress() */ - virtual bool makeProgress() final; + virtual Initiator::MakeProgressResult makeProgress() final; /** \copydoc Initiator::makeProgressBlocking() */ - virtual bool makeProgressBlocking(std::chrono::steady_clock::duration timeout) final; + virtual Initiator::MakeProgressResult makeProgressBlocking(std::chrono::steady_clock::duration timeout) final; private: /** \brief Construct a new RDMInitiator object. @@ -167,7 +167,7 @@ namespace mxl::lib::fabrics::ofi /** \brief Returns true if any of the endpoints contained in this initiator have pending work. */ [[nodiscard]] - bool hasPendingWork() const noexcept; + Initiator::MakeProgressResult afterProgressResult() const noexcept; /** \brief Block on the completion queue with a timeout. */ diff --git a/lib/fabrics/ofi/src/internal/RDMTarget.cpp b/lib/fabrics/ofi/src/internal/RDMTarget.cpp index 6597a1fd2..f66cc8f27 100644 --- a/lib/fabrics/ofi/src/internal/RDMTarget.cpp +++ b/lib/fabrics/ofi/src/internal/RDMTarget.cpp @@ -85,60 +85,14 @@ namespace mxl::lib::fabrics::ofi , _protocol(std::move(proto)) {} - std::optional RDMTarget::readGrain() + std::optional RDMTarget::read() { - if (!_protocol->canReadGrains()) - { - throw Exception::unsupportedOperation("The current protocol does not support reading grains."); - } - - if (auto res = readNext({}); res) - { - return std::get(*res); - } - return std::nullopt; + return readNext({}); } - std::optional RDMTarget::readGrainBlocking(std::chrono::steady_clock::duration timeout) + std::optional RDMTarget::readBlocking(std::chrono::steady_clock::duration timeout) { - if (!_protocol->canReadGrains()) - { - throw Exception::unsupportedOperation("The current protocol does not support reading grains."); - } - - if (auto res = readNext(timeout); res) - { - return std::get(*res); - } - return std::nullopt; - } - - std::optional RDMTarget::readSamples() - { - if (!_protocol->canReadSamples()) - { - throw Exception::unsupportedOperation("The current protocol does not support reading samples."); - } - - if (auto res = readNext({}); res) - { - return std::get(*res); - } - return std::nullopt; - } - - std::optional RDMTarget::readSamplesBlocking(std::chrono::steady_clock::duration timeout) - { - if (!_protocol->canReadSamples()) - { - throw Exception::unsupportedOperation("The current protocol does not support reading samples."); - } - - if (auto res = readNext(timeout); res) - { - return std::get(*res); - } - return std::nullopt; + return readNext(timeout); } void RDMTarget::shutdown() @@ -147,10 +101,22 @@ namespace mxl::lib::fabrics::ofi template std::optional RDMTarget::readNext(std::chrono::steady_clock::duration timeout) { - auto completion = readCompletionQueue(*_ep.completionQueue(), timeout); - if (completion) + try + { + auto completion = readCompletionQueue(*_ep.completionQueue(), timeout); + if (completion) + { + return _protocol->read(_ep, *completion); + } + } + catch (FabricException const& ex) { - return _protocol->read(_ep, *completion); + if (ex.isInterrupted()) + { + return std::make_optional(); + } + + throw; } return {}; diff --git a/lib/fabrics/ofi/src/internal/RDMTarget.hpp b/lib/fabrics/ofi/src/internal/RDMTarget.hpp index 183412cbd..20ba6ca02 100644 --- a/lib/fabrics/ofi/src/internal/RDMTarget.hpp +++ b/lib/fabrics/ofi/src/internal/RDMTarget.hpp @@ -29,19 +29,11 @@ namespace mxl::lib::fabrics::ofi /** \copydoc Target::read() */ - virtual std::optional readGrain() final; + virtual std::optional read() final; /** \copydoc Target::readSamples() */ - virtual std::optional readSamples() final; - - /** \copydoc Target::readBlocking() - */ - virtual std::optional readGrainBlocking(std::chrono::steady_clock::duration timeout) final; - - /** \copydoc Target::readSamplesBlocking() - */ - virtual std::optional readSamplesBlocking(std::chrono::steady_clock::duration timeout) final; + virtual std::optional readBlocking(std::chrono::steady_clock::duration timeout) final; /** \copydoc Target::shutdown() */ diff --git a/lib/fabrics/ofi/src/internal/Target.cpp b/lib/fabrics/ofi/src/internal/Target.cpp index 432c95e23..31b26420b 100644 --- a/lib/fabrics/ofi/src/internal/Target.cpp +++ b/lib/fabrics/ofi/src/internal/Target.cpp @@ -36,44 +36,24 @@ namespace mxl::lib::fabrics::ofi return reinterpret_cast(this); } - std::optional TargetWrapper::readGrain() + std::optional TargetWrapper::read() { if (!_inner) { throw Exception::invalidState("Target is not set up."); } - return _inner->readGrain(); + return _inner->read(); } - std::optional TargetWrapper::readGrainBlocking(std::chrono::steady_clock::duration timeout) + std::optional TargetWrapper::readBlocking(std::chrono::steady_clock::duration timeout) { if (!_inner) { throw Exception::invalidState("Target is not set up."); } - return _inner->readGrainBlocking(timeout); - } - - std::optional TargetWrapper::readSamples() - { - if (!_inner) - { - throw Exception::invalidState("Target is not set up."); - } - - return _inner->readSamples(); - } - - std::optional TargetWrapper::readSamplesBlocking(std::chrono::steady_clock::duration timeout) - { - if (!_inner) - { - throw Exception::invalidState("Target is not set up."); - } - - return _inner->readSamplesBlocking(timeout); + return _inner->readBlocking(timeout); } template diff --git a/lib/fabrics/ofi/src/internal/Target.hpp b/lib/fabrics/ofi/src/internal/Target.hpp index 58411a205..8dabe6ec7 100644 --- a/lib/fabrics/ofi/src/internal/Target.hpp +++ b/lib/fabrics/ofi/src/internal/Target.hpp @@ -47,7 +47,10 @@ namespace mxl::lib::fabrics::ofi std::size_t count; }; - using ReadResult = std::variant; + struct Interrupted + {}; + + using ReadResult = std::variant; public: virtual ~Target() = default; @@ -57,26 +60,13 @@ namespace mxl::lib::fabrics::ofi * A non-blocking operation that also drives the connection forward. Continuous invocation of this function is necessary for connection * establishment and ongoing progress. */ - virtual std::optional readGrain() = 0; + virtual std::optional read() = 0; /** \brief Determine if new data can be consumed. * * A blocking version of readGrain. see readGrain(). */ - virtual std::optional readGrainBlocking(std::chrono::steady_clock::duration timeout) = 0; - - /** \brief Determine if new data can be consumed. - * - * A non-blocking operation that also drives the connection forward. Continuous invocation of this function is necessary for connection - * establishment and ongoing progress. - */ - virtual std::optional readSamples() = 0; - - /** \brief Determine if new data can be consumed. - * - * A blocking version of readSamples. see readSamples(). - */ - virtual std::optional readSamplesBlocking(std::chrono::steady_clock::duration timeout) = 0; + virtual std::optional readBlocking(std::chrono::steady_clock::duration timeout) = 0; /** \brief Shut down the target gracefully. * Initiates a graceful shutdown of the target and blocks until the shutdown is complete. @@ -125,19 +115,11 @@ namespace mxl::lib::fabrics::ofi /** \copydoc Target::readGrain() */ - std::optional readGrain(); + std::optional read(); /** \copydoc Target::readGrainBlocking(std::chrono::steady_clock::duration) */ - std::optional readGrainBlocking(std::chrono::steady_clock::duration timeout); - - /** \copydoc Target::readSamples() - */ - std::optional readSamples(); - - /** \copydoc Target::readSamplesBlocking(std::chrono::steady_clock::duration) - */ - std::optional readSamplesBlocking(std::chrono::steady_clock::duration timeout); + std::optional readBlocking(std::chrono::steady_clock::duration timeout); /** \brief Set up the target with the specified configuration. * diff --git a/tools/mxl-fabrics-demo/demo.cpp b/tools/mxl-fabrics-demo/demo.cpp index 03226d349..06c7fec64 100644 --- a/tools/mxl-fabrics-demo/demo.cpp +++ b/tools/mxl-fabrics-demo/demo.cpp @@ -359,7 +359,7 @@ class AppInitator status = makeProgress(std::chrono::milliseconds(250)); if (status == MXL_ERR_INTERRUPTED) { - return MXL_STATUS_OK; + return status; } if (status != MXL_ERR_NOT_READY && status != MXL_STATUS_OK) @@ -468,7 +468,7 @@ class AppInitator status = makeProgress(std::chrono::milliseconds(10)); if (status == MXL_ERR_INTERRUPTED) { - return MXL_STATUS_OK; + continue; } if (status != MXL_ERR_NOT_READY && status != MXL_STATUS_OK) @@ -561,7 +561,7 @@ class AppInitator status = makeProgress(std::chrono::milliseconds(10)); if (status == MXL_ERR_INTERRUPTED) { - return MXL_STATUS_OK; + continue; } if (status != MXL_ERR_NOT_READY && status != MXL_STATUS_OK) @@ -812,7 +812,7 @@ class AppTarget } else if (status == MXL_ERR_INTERRUPTED) { - return MXL_STATUS_OK; + continue; } else if (status != MXL_STATUS_OK) { @@ -872,7 +872,7 @@ class AppTarget } else if (status == MXL_ERR_INTERRUPTED) { - return MXL_STATUS_OK; + continue; } else if (status != MXL_STATUS_OK) {