Exceptions: All of smocore likely now uses exceptions
This commit is contained in:
+4
-30
@@ -64,21 +64,8 @@ BodyViralPostingInvoker<void> Body::initializeCReq()
|
||||
<< '\n';
|
||||
}
|
||||
|
||||
sscl::MultiOperationResultSet attachResults = co_await
|
||||
device::DeviceManager::getInstance()
|
||||
.attachAllUnattachedDevicesFromCmdlineCReq();
|
||||
std::cout << "Mrntt: attached "
|
||||
<< attachResults.nSucceeded << " of " << attachResults.nTotal
|
||||
<< " sense devices." << "\n";
|
||||
|
||||
if (attachResults.nTotal > 0 && attachResults.nSucceeded == 0)
|
||||
{
|
||||
throw std::runtime_error(
|
||||
std::string(__func__)
|
||||
+ ": Failed to attach any of "
|
||||
+ std::to_string(attachResults.nTotal)
|
||||
+ " requested sense devices");
|
||||
}
|
||||
co_await device::DeviceManager::getInstance()
|
||||
.attachAllUnattachedDevicesFromCmdlineCReq();
|
||||
|
||||
co_return;
|
||||
}
|
||||
@@ -102,21 +89,8 @@ BodyViralPostingInvoker<void> Body::finalizeCReq()
|
||||
}
|
||||
|
||||
std::cout << "Mrntt: About to detach all sense devices." << "\n";
|
||||
sscl::MultiOperationResultSet detachResults = co_await
|
||||
device::DeviceManager::getInstance().detachAllAttachedDeviceRolesCReq();
|
||||
|
||||
if (detachResults.nFailed > 0)
|
||||
{
|
||||
std::cerr << "Mrntt: Failed to detach "
|
||||
<< detachResults.nFailed << " of " << detachResults.nTotal
|
||||
<< " sense devices." << "\n";
|
||||
}
|
||||
else
|
||||
{
|
||||
std::cout << "Mrntt: Successfully detached "
|
||||
<< detachResults.nSucceeded << " of " << detachResults.nTotal
|
||||
<< " sense devices." << "\n";
|
||||
}
|
||||
co_await device::DeviceManager::getInstance()
|
||||
.detachAllAttachedDeviceRolesCReq();
|
||||
|
||||
std::cout << "Mrntt: About to finalize all stim buff api libs." << "\n";
|
||||
co_await stim_buff::StimBuffApiManager::getInstance()
|
||||
|
||||
@@ -16,6 +16,7 @@
|
||||
#include <mind.h>
|
||||
#include <spinscale/co/postTarget.h>
|
||||
#include <spinscale/co/group.h>
|
||||
#include <spinscale/asynchronousLoop.h>
|
||||
|
||||
namespace smo {
|
||||
namespace device {
|
||||
@@ -43,6 +44,28 @@ std::shared_ptr<sscl::ComponentThread> threadForDeviceOp(
|
||||
return mind::globalMind->body.thread;
|
||||
}
|
||||
|
||||
void throwIfStimBuffAttachFailed(
|
||||
const stim_buff::StimBuffDeviceOpResult &result,
|
||||
const char *callerFn)
|
||||
{
|
||||
if (result.success) { return; }
|
||||
|
||||
throw std::runtime_error(
|
||||
std::string(callerFn) + ": attach failed for "
|
||||
+ result.deviceSpec->stringify());
|
||||
}
|
||||
|
||||
void throwIfStimBuffDetachFailed(
|
||||
const stim_buff::StimBuffDeviceOpResult &result,
|
||||
const char *callerFn)
|
||||
{
|
||||
if (result.success) { return; }
|
||||
|
||||
throw std::runtime_error(
|
||||
std::string(callerFn) + ": detach failed for "
|
||||
+ result.deviceSpec->stringify());
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
DeviceManager::~DeviceManager()
|
||||
@@ -60,38 +83,30 @@ const std::string DeviceManager::stringifyDeviceSpecs(void)
|
||||
return oss.str();
|
||||
}
|
||||
|
||||
mrntt::MrnttViralPostingInvoker<stim_buff::StimBuffDeviceOpResult>
|
||||
mrntt::MrnttViralPostingInvoker<void>
|
||||
DeviceManager::attachStimBuffDeviceCReq(
|
||||
const std::shared_ptr<DeviceAttachmentSpec>& spec)
|
||||
{
|
||||
assertMarionetteThread();
|
||||
|
||||
auto &sbam = stim_buff::StimBuffApiManager::getInstance();
|
||||
auto libOpt = sbam.getStimBuffApiLibByApiName(spec->stimBuffApi);
|
||||
if (!libOpt)
|
||||
{
|
||||
std::cerr << "attachStimBuffDeviceCReq: No library found for API '"
|
||||
<< spec->stimBuffApi << "'" << std::endl;
|
||||
co_return stim_buff::StimBuffDeviceOpResult{false, spec};
|
||||
}
|
||||
|
||||
auto &lib = *libOpt.value();
|
||||
auto &lib = sbam.getStimBuffApiLibByApiName(spec->stimBuffApi);
|
||||
|
||||
if (lib.isBeingDestroyed.load())
|
||||
{
|
||||
std::cerr << std::string(__func__) + ": Library is being destroyed"
|
||||
<< " for API '" << spec->stimBuffApi << "'. Bailing out.\n";
|
||||
co_return stim_buff::StimBuffDeviceOpResult{false, spec};
|
||||
throw std::runtime_error(
|
||||
std::string(__func__) + ": Library is being destroyed"
|
||||
+ " for API '" + spec->stimBuffApi + "'");
|
||||
}
|
||||
|
||||
if (!lib.stimBuffApiDesc.sal_mgmt_libOps.attachDeviceCReq)
|
||||
{
|
||||
std::cerr << std::string(__func__) + ": attachDeviceCReq() is NULL "
|
||||
"for library '" << lib.libraryPath << "'"
|
||||
<< std::endl;
|
||||
co_return stim_buff::StimBuffDeviceOpResult{false, spec};
|
||||
throw std::runtime_error(
|
||||
std::string(__func__) + ": attachDeviceCReq() is NULL "
|
||||
"for library '" + lib.libraryPath + "'");
|
||||
}
|
||||
|
||||
/* FIXME Locking here makes no sense. */
|
||||
sscl::co::CoQutex::ReleaseHandle sbamGuard =
|
||||
co_await sbam.s.lock.getAcquireInvocationAndSuspensionPolicy();
|
||||
sscl::co::CoQutex::ReleaseHandle libGuard =
|
||||
@@ -119,43 +134,38 @@ DeviceManager::attachStimBuffDeviceCReq(
|
||||
<< spec->deviceIdentifier << " to body thread" << "\n";
|
||||
}
|
||||
|
||||
co_return co_await lib.stimBuffApiDesc.sal_mgmt_libOps.attachDeviceCReq(
|
||||
sscl::co::ExplicitPostTarget{targetThread->getIoContext()},
|
||||
spec, targetThread);
|
||||
stim_buff::StimBuffDeviceOpResult result =
|
||||
co_await lib.stimBuffApiDesc.sal_mgmt_libOps.attachDeviceCReq(
|
||||
sscl::co::ExplicitPostTarget{targetThread->getIoContext()},
|
||||
spec, targetThread);
|
||||
throwIfStimBuffAttachFailed(result, __func__);
|
||||
co_return;
|
||||
}
|
||||
|
||||
mrntt::MrnttViralPostingInvoker<stim_buff::StimBuffDeviceOpResult>
|
||||
mrntt::MrnttViralPostingInvoker<void>
|
||||
DeviceManager::detachStimBuffDeviceCReq(
|
||||
const std::shared_ptr<DeviceAttachmentSpec>& spec)
|
||||
{
|
||||
assertMarionetteThread();
|
||||
|
||||
auto &sbam = stim_buff::StimBuffApiManager::getInstance();
|
||||
auto libOpt = sbam.getStimBuffApiLibByApiName(spec->stimBuffApi);
|
||||
if (!libOpt)
|
||||
{
|
||||
std::cerr << "detachStimBuffDeviceCReq: No library found for API '"
|
||||
<< spec->stimBuffApi << "'" << std::endl;
|
||||
co_return stim_buff::StimBuffDeviceOpResult{false, spec};
|
||||
}
|
||||
|
||||
auto &lib = *libOpt.value();
|
||||
auto &lib = sbam.getStimBuffApiLibByApiName(spec->stimBuffApi);
|
||||
|
||||
if (lib.isBeingDestroyed.load())
|
||||
{
|
||||
std::cerr << std::string(__func__) + ": Library is being destroyed"
|
||||
<< " for API '" << spec->stimBuffApi << "'. Bailing out.\n";
|
||||
co_return stim_buff::StimBuffDeviceOpResult{false, spec};
|
||||
throw std::runtime_error(
|
||||
std::string(__func__) + ": Library is being destroyed"
|
||||
+ " for API '" + spec->stimBuffApi + "'");
|
||||
}
|
||||
|
||||
if (!lib.stimBuffApiDesc.sal_mgmt_libOps.detachDeviceCReq)
|
||||
{
|
||||
std::cerr << std::string(__func__) + ": detachDeviceCReq() is NULL "
|
||||
"for library '" << lib.libraryPath << "'"
|
||||
<< std::endl;
|
||||
co_return stim_buff::StimBuffDeviceOpResult{false, spec};
|
||||
throw std::runtime_error(
|
||||
std::string(__func__) + ": detachDeviceCReq() is NULL "
|
||||
"for library '" + lib.libraryPath + "'");
|
||||
}
|
||||
|
||||
/* FIXME Locking here makes no sense. */
|
||||
sscl::co::CoQutex::ReleaseHandle sbamGuard =
|
||||
co_await sbam.s.lock.getAcquireInvocationAndSuspensionPolicy();
|
||||
sscl::co::CoQutex::ReleaseHandle libGuard =
|
||||
@@ -165,12 +175,15 @@ DeviceManager::detachStimBuffDeviceCReq(
|
||||
std::shared_ptr<sscl::ComponentThread> targetThread =
|
||||
threadForDeviceOp(*spec);
|
||||
|
||||
co_return co_await lib.stimBuffApiDesc.sal_mgmt_libOps.detachDeviceCReq(
|
||||
sscl::co::ExplicitPostTarget{targetThread->getIoContext()},
|
||||
spec);
|
||||
stim_buff::StimBuffDeviceOpResult result =
|
||||
co_await lib.stimBuffApiDesc.sal_mgmt_libOps.detachDeviceCReq(
|
||||
sscl::co::ExplicitPostTarget{targetThread->getIoContext()},
|
||||
spec);
|
||||
throwIfStimBuffDetachFailed(result, __func__);
|
||||
co_return;
|
||||
}
|
||||
|
||||
mrntt::MrnttViralPostingInvoker<DeviceManager::DeviceAttachmentIndResult>
|
||||
mrntt::MrnttViralPostingInvoker<std::shared_ptr<DeviceRole>>
|
||||
DeviceManager::newDeviceAttachmentSpecIndCReq(
|
||||
const DeviceAttachmentSpec &spec)
|
||||
{
|
||||
@@ -242,36 +255,29 @@ DeviceManager::newDeviceAttachmentSpecIndCReq(
|
||||
", deviceExists=" + std::to_string(deviceExists));
|
||||
}
|
||||
|
||||
// Already attached, return success
|
||||
co_return DeviceAttachmentIndResult{
|
||||
true, existingDeviceRole, specPtr};
|
||||
co_return existingDeviceRole;
|
||||
}
|
||||
|
||||
stim_buff::StimBuffDeviceOpResult attachResult =
|
||||
co_await attachStimBuffDeviceCReq(specPtr);
|
||||
dmGuard.release();
|
||||
|
||||
if (!attachResult.success)
|
||||
{
|
||||
std::cerr << __func__ << ": Attach failed for device spec "
|
||||
<< attachResult.deviceSpec->stringify() << std::endl;
|
||||
co_return DeviceAttachmentIndResult{
|
||||
false, nullptr, attachResult.deviceSpec};
|
||||
}
|
||||
/* FIXME:
|
||||
* We should add an unlocked flag to at/detachStimBuffDeviceCReq()
|
||||
* so we can call it with the devmgr lock held.
|
||||
*/
|
||||
co_await attachStimBuffDeviceCReq(specPtr);
|
||||
|
||||
try {
|
||||
// Create DeviceRole and add it to both DeviceManager's and Device's collections
|
||||
auto deviceRole = std::make_shared<DeviceRole>(*device, specPtr);
|
||||
device->deviceRoles.push_back(deviceRole);
|
||||
dm.s.rsrc.attachedDeviceRoles.push_back(deviceRole);
|
||||
sscl::co::CoQutex::ReleaseHandle dmGuardAfterAttach =
|
||||
co_await dm.s.lock.getAcquireInvocationAndSuspensionPolicy();
|
||||
(void)dmGuardAfterAttach;
|
||||
|
||||
co_return DeviceAttachmentIndResult{true, deviceRole, specPtr};
|
||||
} catch (const std::exception&) {
|
||||
// Attach failed, return error
|
||||
co_return DeviceAttachmentIndResult{false, nullptr, specPtr};
|
||||
}
|
||||
auto deviceRole = std::make_shared<DeviceRole>(*device, specPtr);
|
||||
device->deviceRoles.push_back(deviceRole);
|
||||
dm.s.rsrc.attachedDeviceRoles.push_back(deviceRole);
|
||||
|
||||
co_return deviceRole;
|
||||
}
|
||||
|
||||
mrntt::MrnttViralPostingInvoker<DeviceManager::DeviceAttachmentIndResult>
|
||||
mrntt::MrnttViralPostingInvoker<void>
|
||||
DeviceManager::removeDeviceAttachmentSpecCReq(
|
||||
const DeviceAttachmentSpec &spec)
|
||||
{
|
||||
@@ -294,95 +300,92 @@ DeviceManager::removeDeviceAttachmentSpecCReq(
|
||||
|
||||
if (!specPtr)
|
||||
{
|
||||
// Spec not found, return failure
|
||||
co_return DeviceAttachmentIndResult{false, nullptr, nullptr};
|
||||
throw std::runtime_error(
|
||||
std::string(__func__) + ": Device attachment spec not found: "
|
||||
+ spec.stringify());
|
||||
}
|
||||
|
||||
// Call detachStimBuffDeviceCReq first - only clean up metadata if this succeeds
|
||||
stim_buff::StimBuffDeviceOpResult detachResult =
|
||||
co_await detachStimBuffDeviceCReq(specPtr);
|
||||
dmGuard.release();
|
||||
|
||||
if (!detachResult.success)
|
||||
/* FIXME:
|
||||
* We should add an unlocked flag to at/detachStimBuffDeviceCReq()
|
||||
* so we can call it with the devmgr lock held.
|
||||
*/
|
||||
co_await detachStimBuffDeviceCReq(specPtr);
|
||||
|
||||
sscl::co::CoQutex::ReleaseHandle dmGuardAfterDetach =
|
||||
co_await dm.s.lock.getAcquireInvocationAndSuspensionPolicy();
|
||||
(void)dmGuardAfterDetach;
|
||||
|
||||
// Find the DeviceRole in attachedDeviceRoles
|
||||
auto deviceRoleIt = std::find_if(
|
||||
dm.s.rsrc.attachedDeviceRoles.begin(),
|
||||
dm.s.rsrc.attachedDeviceRoles.end(),
|
||||
[&specPtr](const std::shared_ptr<DeviceRole> &role) {
|
||||
return *role->deviceAttachmentSpec == *specPtr;
|
||||
}
|
||||
);
|
||||
|
||||
if (deviceRoleIt == dm.s.rsrc.attachedDeviceRoles.end())
|
||||
{
|
||||
// Detach failed, metadata remains intact
|
||||
co_return DeviceAttachmentIndResult{
|
||||
false, nullptr, detachResult.deviceSpec};
|
||||
throw std::runtime_error(
|
||||
std::string(__func__) + ": DeviceRole not found for spec (race "
|
||||
"condition)?: "
|
||||
+ specPtr->stringify() + ", deviceRoles="
|
||||
+ std::to_string(dm.s.rsrc.attachedDeviceRoles.size()));
|
||||
}
|
||||
|
||||
// Detach succeeded, now find and clean up metadata
|
||||
try {
|
||||
// Find the DeviceRole in attachedDeviceRoles
|
||||
auto deviceRoleIt = std::find_if(
|
||||
dm.s.rsrc.attachedDeviceRoles.begin(),
|
||||
dm.s.rsrc.attachedDeviceRoles.end(),
|
||||
[&specPtr](const std::shared_ptr<DeviceRole> &role) {
|
||||
return *role->deviceAttachmentSpec == *specPtr;
|
||||
}
|
||||
);
|
||||
auto deviceRole = *deviceRoleIt;
|
||||
auto& device = deviceRole->parentDevice;
|
||||
|
||||
if (deviceRoleIt == dm.s.rsrc.attachedDeviceRoles.end())
|
||||
{
|
||||
// DeviceRole not found, return failure
|
||||
co_return DeviceAttachmentIndResult{
|
||||
false, nullptr, detachResult.deviceSpec};
|
||||
}
|
||||
// Remove DeviceRole from DeviceManager's collection
|
||||
dm.s.rsrc.attachedDeviceRoles.erase(deviceRoleIt);
|
||||
|
||||
auto deviceRole = *deviceRoleIt;
|
||||
auto& device = deviceRole->parentDevice;
|
||||
|
||||
// Remove DeviceRole from DeviceManager's collection
|
||||
dm.s.rsrc.attachedDeviceRoles.erase(deviceRoleIt);
|
||||
|
||||
// Remove DeviceRole from Device's collection
|
||||
auto deviceRoleIt2 = std::find(
|
||||
device.deviceRoles.begin(),
|
||||
device.deviceRoles.end(),
|
||||
deviceRole);
|
||||
if (deviceRoleIt2 != device.deviceRoles.end())
|
||||
{
|
||||
device.deviceRoles.erase(deviceRoleIt2);
|
||||
}
|
||||
|
||||
// Remove DeviceAttachmentSpec from deviceAttachmentSpecs collection
|
||||
auto specIt = std::find_if(
|
||||
dm.s.rsrc.deviceAttachmentSpecs.begin(),
|
||||
dm.s.rsrc.deviceAttachmentSpecs.end(),
|
||||
[&specPtr](
|
||||
const std::shared_ptr<DeviceAttachmentSpec> &existingSpec)
|
||||
{
|
||||
return *existingSpec == *specPtr;
|
||||
}
|
||||
);
|
||||
|
||||
if (specIt != dm.s.rsrc.deviceAttachmentSpecs.end())
|
||||
{
|
||||
dm.s.rsrc.deviceAttachmentSpecs.erase(specIt);
|
||||
}
|
||||
|
||||
co_return DeviceAttachmentIndResult{
|
||||
true, deviceRole, detachResult.deviceSpec};
|
||||
} catch (const std::exception&) {
|
||||
// Cleanup failed, return error
|
||||
co_return DeviceAttachmentIndResult{
|
||||
false, nullptr, detachResult.deviceSpec};
|
||||
// Remove DeviceRole from Device's collection
|
||||
auto deviceRoleIt2 = std::find(
|
||||
device.deviceRoles.begin(),
|
||||
device.deviceRoles.end(),
|
||||
deviceRole);
|
||||
if (deviceRoleIt2 != device.deviceRoles.end())
|
||||
{
|
||||
device.deviceRoles.erase(deviceRoleIt2);
|
||||
}
|
||||
|
||||
// Remove DeviceAttachmentSpec from deviceAttachmentSpecs collection
|
||||
auto specIt = std::find_if(
|
||||
dm.s.rsrc.deviceAttachmentSpecs.begin(),
|
||||
dm.s.rsrc.deviceAttachmentSpecs.end(),
|
||||
[&specPtr](
|
||||
const std::shared_ptr<DeviceAttachmentSpec> &existingSpec)
|
||||
{
|
||||
return *existingSpec == *specPtr;
|
||||
}
|
||||
);
|
||||
|
||||
if (specIt != dm.s.rsrc.deviceAttachmentSpecs.end())
|
||||
{
|
||||
dm.s.rsrc.deviceAttachmentSpecs.erase(specIt);
|
||||
}
|
||||
|
||||
co_return;
|
||||
}
|
||||
|
||||
mrntt::MrnttViralPostingInvoker<sscl::MultiOperationResultSet>
|
||||
mrntt::MrnttViralPostingInvoker<sscl::MultiOperationResultSetWithException>
|
||||
DeviceManager::attachAllUnattachedDevicesFromCReq(
|
||||
const std::shared_ptr<std::vector<DeviceAttachmentSpec>> &specs)
|
||||
{
|
||||
assertMarionetteThread();
|
||||
|
||||
if (specs->empty()) {
|
||||
co_return sscl::MultiOperationResultSet{};
|
||||
const unsigned int nTotal = static_cast<unsigned int>(specs->size());
|
||||
if (nTotal == 0) {
|
||||
co_return sscl::MultiOperationResultSetWithException{};
|
||||
}
|
||||
|
||||
sscl::co::Group group;
|
||||
std::vector<
|
||||
mrntt::MrnttViralPostingInvoker<DeviceManager::DeviceAttachmentIndResult>>
|
||||
mrntt::MrnttViralPostingInvoker<std::shared_ptr<DeviceRole>>>
|
||||
invokers;
|
||||
invokers.reserve(specs->size());
|
||||
invokers.reserve(nTotal);
|
||||
|
||||
for (const auto &spec : *specs)
|
||||
{
|
||||
@@ -390,32 +393,32 @@ DeviceManager::attachAllUnattachedDevicesFromCReq(
|
||||
group.add(invokers.back());
|
||||
}
|
||||
|
||||
co_await group.getAwaitAllSettlementsInvoker();
|
||||
group.checkForAndReThrowGroupExceptions();
|
||||
const std::vector<sscl::co::Group::SettlementDescriptor> &settlements =
|
||||
co_await group.getAwaitAllSettlementsInvoker();
|
||||
|
||||
unsigned int nSucceeded = 0;
|
||||
unsigned int nFailed = 0;
|
||||
for (auto &invoker : invokers)
|
||||
using SettlementType = sscl::co::Group::SettlementDescriptor::TypeE;
|
||||
for (const auto &desc : settlements)
|
||||
{
|
||||
if (invoker.completedReturnValues().myReturnValue.success) {
|
||||
nSucceeded++;
|
||||
} else {
|
||||
if (desc.type == SettlementType::EXCEPTION_THROWN) {
|
||||
nFailed++;
|
||||
} else {
|
||||
nSucceeded++;
|
||||
}
|
||||
}
|
||||
|
||||
if (OptionParser::getOptions().verbose)
|
||||
{
|
||||
std::cout << __func__ << ": " << nSucceeded
|
||||
<< " devices attached, "
|
||||
<< nFailed << " devices failed\n";
|
||||
std::exception_ptr memberFailureException = nullptr;
|
||||
if (nFailed > 0) {
|
||||
memberFailureException = group.captureAggregatedGroupExceptions();
|
||||
}
|
||||
|
||||
co_return sscl::MultiOperationResultSet(
|
||||
static_cast<unsigned int>(specs->size()), nSucceeded, nFailed);
|
||||
co_return sscl::MultiOperationResultSetWithException(
|
||||
sscl::MultiOperationResultSet(nTotal, nSucceeded, nFailed),
|
||||
memberFailureException);
|
||||
}
|
||||
|
||||
mrntt::MrnttViralPostingInvoker<sscl::MultiOperationResultSet>
|
||||
mrntt::MrnttViralPostingInvoker<void>
|
||||
DeviceManager::attachAllUnattachedDevicesFromKnownListCReq()
|
||||
{
|
||||
assertMarionetteThread();
|
||||
@@ -447,39 +450,91 @@ DeviceManager::attachAllUnattachedDevicesFromKnownListCReq()
|
||||
|
||||
dmGuard.release();
|
||||
|
||||
co_return co_await attachAllUnattachedDevicesFromCReq(unattachedSpecs);
|
||||
const sscl::MultiOperationResultSetWithException batchResult =
|
||||
co_await attachAllUnattachedDevicesFromCReq(unattachedSpecs);
|
||||
|
||||
if (batchResult.results.nSucceeded > 0)
|
||||
{
|
||||
std::cout << "DeviceReattacher: Successfully reattached "
|
||||
<< batchResult.results.nSucceeded << " of "
|
||||
<< batchResult.results.nTotal
|
||||
<< " devices" << std::endl;
|
||||
}
|
||||
|
||||
if (batchResult.hasMemberFailure())
|
||||
{
|
||||
try {
|
||||
std::rethrow_exception(batchResult.memberFailureException);
|
||||
} catch (const std::exception &e) {
|
||||
std::cerr << __func__ << ": " << e.what() << std::endl;
|
||||
}
|
||||
}
|
||||
|
||||
co_return;
|
||||
}
|
||||
|
||||
mrntt::MrnttViralPostingInvoker<sscl::MultiOperationResultSet>
|
||||
mrntt::MrnttViralPostingInvoker<void>
|
||||
DeviceManager::attachAllUnattachedDevicesFromCmdlineCReq()
|
||||
{
|
||||
auto specs = std::make_shared<std::vector<DeviceAttachmentSpec>>(
|
||||
getInstance().s.rsrc.commandLineDASpecs);
|
||||
|
||||
co_return co_await attachAllUnattachedDevicesFromCReq(specs);
|
||||
const sscl::MultiOperationResultSetWithException batchResult =
|
||||
co_await attachAllUnattachedDevicesFromCReq(specs);
|
||||
|
||||
std::cout << "Mrntt: attached "
|
||||
<< batchResult.results.nSucceeded << " of "
|
||||
<< batchResult.results.nTotal
|
||||
<< " sense devices." << "\n";
|
||||
|
||||
if (batchResult.results.nTotal > 0
|
||||
&& batchResult.results.nSucceeded == 0)
|
||||
{
|
||||
std::string message =
|
||||
std::string(__func__)
|
||||
+ ": Startup policy requires at least one cmdline sense device "
|
||||
"to attach successfully; 0 of "
|
||||
+ std::to_string(batchResult.results.nTotal)
|
||||
+ " requested sense devices attached — aborting startup.";
|
||||
|
||||
if (batchResult.hasMemberFailure())
|
||||
{
|
||||
try {
|
||||
std::rethrow_exception(batchResult.memberFailureException);
|
||||
} catch (const std::exception &e) {
|
||||
message += "\n";
|
||||
message += e.what();
|
||||
} catch (...) {
|
||||
message += "\n<unknown exception type>";
|
||||
}
|
||||
}
|
||||
|
||||
throw std::runtime_error(message);
|
||||
}
|
||||
|
||||
co_return;
|
||||
}
|
||||
|
||||
mrntt::MrnttViralPostingInvoker<sscl::MultiOperationResultSet>
|
||||
mrntt::MrnttViralPostingInvoker<void>
|
||||
DeviceManager::detachAllAttachedDeviceRolesCReq()
|
||||
{
|
||||
assertMarionetteThread();
|
||||
|
||||
std::vector<std::shared_ptr<DeviceAttachmentSpec>> specsToDetach;
|
||||
specsToDetach.reserve(getInstance().s.rsrc.attachedDeviceRoles.size());
|
||||
for (const auto& deviceRole : getInstance().s.rsrc.attachedDeviceRoles)
|
||||
{
|
||||
for (const auto& deviceRole : getInstance().s.rsrc.attachedDeviceRoles) {
|
||||
specsToDetach.push_back(deviceRole->deviceAttachmentSpec);
|
||||
}
|
||||
|
||||
if (specsToDetach.empty()) {
|
||||
co_return sscl::MultiOperationResultSet{};
|
||||
sscl::AsynchronousLoop loop(
|
||||
static_cast<unsigned int>(specsToDetach.size()));
|
||||
if (loop.nTotalIsZero()) {
|
||||
co_return;
|
||||
}
|
||||
|
||||
sscl::co::Group group;
|
||||
std::vector<
|
||||
mrntt::MrnttViralPostingInvoker<stim_buff::StimBuffDeviceOpResult>>
|
||||
invokers;
|
||||
invokers.reserve(specsToDetach.size());
|
||||
std::vector<mrntt::MrnttViralPostingInvoker<void>> invokers;
|
||||
invokers.reserve(loop.nTotal);
|
||||
|
||||
for (const auto &spec : specsToDetach)
|
||||
{
|
||||
@@ -487,20 +542,39 @@ DeviceManager::detachAllAttachedDeviceRolesCReq()
|
||||
group.add(invokers.back());
|
||||
}
|
||||
|
||||
co_await group.getAwaitAllSettlementsInvoker();
|
||||
group.checkForAndReThrowGroupExceptions();
|
||||
const std::vector<sscl::co::Group::SettlementDescriptor> &settlements =
|
||||
co_await group.getAwaitAllSettlementsInvoker();
|
||||
|
||||
unsigned int nSucceeded = 0;
|
||||
unsigned int nFailed = 0;
|
||||
for (auto &invoker : invokers)
|
||||
using SettlementType = sscl::co::Group::SettlementDescriptor::TypeE;
|
||||
for (const auto &desc : settlements)
|
||||
{
|
||||
if (invoker.completedReturnValues().myReturnValue.success) {
|
||||
nSucceeded++;
|
||||
} else {
|
||||
nFailed++;
|
||||
loop.incrementSuccessOrFailureDueTo(
|
||||
desc.type != SettlementType::EXCEPTION_THROWN);
|
||||
}
|
||||
|
||||
if (loop.nFailed.load() > 0)
|
||||
{
|
||||
try {
|
||||
group.checkForAndReThrowGroupExceptions();
|
||||
} catch (const std::exception &e) {
|
||||
std::cerr << __func__ << ": " << e.what() << std::endl;
|
||||
}
|
||||
}
|
||||
|
||||
const unsigned int nSucceeded = loop.nSucceeded.load();
|
||||
const unsigned int nFailed = loop.nFailed.load();
|
||||
|
||||
if (nFailed > 0)
|
||||
{
|
||||
std::cerr << "Mrntt: Failed to detach "
|
||||
<< nFailed << " of " << loop.nTotal << " sense devices." << "\n";
|
||||
}
|
||||
else
|
||||
{
|
||||
std::cout << "Mrntt: Successfully detached "
|
||||
<< nSucceeded << " of " << loop.nTotal << " sense devices." << "\n";
|
||||
}
|
||||
|
||||
if (OptionParser::getOptions().verbose)
|
||||
{
|
||||
std::cout << __func__ << ": " << nSucceeded
|
||||
@@ -508,9 +582,7 @@ DeviceManager::detachAllAttachedDeviceRolesCReq()
|
||||
<< nFailed << " devices failed\n";
|
||||
}
|
||||
|
||||
co_return sscl::MultiOperationResultSet(
|
||||
static_cast<unsigned int>(specsToDetach.size()),
|
||||
nSucceeded, nFailed);
|
||||
co_return;
|
||||
}
|
||||
|
||||
void DeviceManager::initializeDeviceReattacher()
|
||||
|
||||
@@ -6,6 +6,7 @@
|
||||
#include <deviceManager/deviceReattacher.h>
|
||||
#include <deviceManager/deviceManager.h>
|
||||
#include <marionette/marionetteThread.h>
|
||||
#include <spinscale/co/nonViralCompletion.h>
|
||||
|
||||
namespace smo {
|
||||
namespace device {
|
||||
@@ -20,29 +21,25 @@ DeviceReattacher::DeviceReattacher(
|
||||
DeviceManager& parent, std::shared_ptr<sscl::ComponentThread> ioThread)
|
||||
: parent(parent), ioThread(ioThread), timer(ioThread->getIoContext())
|
||||
{
|
||||
/** EXPLANATION:
|
||||
* The thread on which DeviceReattacher runs is whichever thread executes
|
||||
* the io_context that owns deadline_timer. Timer async_wait handlers
|
||||
* (onTimeout, holdReattachCReq, reattachKnownListCReq) are dispatched on
|
||||
* that thread. ioThread selects that io_context here; start() only arms
|
||||
* the timer on it.
|
||||
*/
|
||||
}
|
||||
|
||||
mrntt::MrnttNonViralPostingInvoker DeviceReattacher::reattachKnownListCReq(
|
||||
[[maybe_unused]] sscl::co::ExplicitPostTarget postTarget,
|
||||
mrntt::MrnttNonViralNonPostingInvoker DeviceReattacher::reattachKnownListCReq(
|
||||
[[maybe_unused]] std::exception_ptr &exceptionPtr,
|
||||
[[maybe_unused]] std::function<void()> callback)
|
||||
{
|
||||
/** EXPLANATION:
|
||||
* DeviceManager attach APIs require the marionette thread; postTarget
|
||||
* selects where this coroutine runs (mrntt io_context). Completion still
|
||||
* posts back to the mrntt timer thread via callerIoContext.
|
||||
* Non-posting: invoked from holdReattachCReq on the timer callback thread
|
||||
* (see ctor). Nested DeviceManager attach APIs still post to MRNTT as
|
||||
* needed via their own viral posting invokers.
|
||||
*/
|
||||
(void)postTarget;
|
||||
|
||||
sscl::MultiOperationResultSet results = co_await
|
||||
parent.attachAllUnattachedDevicesFromKnownListCReq();
|
||||
if (results.nTotal > 0)
|
||||
{
|
||||
std::cout << "DeviceReattacher: Successfully reattached "
|
||||
<< results.nSucceeded << " of " << results.nTotal
|
||||
<< " devices" << std::endl;
|
||||
}
|
||||
|
||||
co_await parent.attachAllUnattachedDevicesFromKnownListCReq();
|
||||
co_return;
|
||||
}
|
||||
|
||||
@@ -92,10 +89,20 @@ void DeviceReattacher::holdReattachCReq()
|
||||
|
||||
reattachCReqInvoker.reset();
|
||||
reattachCReqInvoker.emplace(reattachKnownListCReq(
|
||||
sscl::co::ExplicitPostTarget{ioThread->getIoContext()},
|
||||
reattachLifetimeExceptionPtr,
|
||||
[this]()
|
||||
{
|
||||
sscl::co::NonViralCompletion nvc(reattachLifetimeExceptionPtr);
|
||||
if (nvc.hasException())
|
||||
{
|
||||
try {
|
||||
nvc.checkAndRethrowException();
|
||||
} catch (const std::exception &e) {
|
||||
std::cerr << "DeviceReattacher: " << e.what()
|
||||
<< std::endl;
|
||||
}
|
||||
}
|
||||
|
||||
sscl::SpinLock::Guard guard(deviceReattacherCanceler.s.lock);
|
||||
reattachOpInFlight = false;
|
||||
}));
|
||||
|
||||
@@ -13,8 +13,8 @@
|
||||
#include <deviceManager/deviceRole.h>
|
||||
#include <deviceManager/deviceReattacher.h>
|
||||
#include <marionette/marionetteThread.h>
|
||||
#include <spinscale/co/coQutex.h>
|
||||
#include <spinscale/multiOperationResultSet.h>
|
||||
#include <spinscale/co/coQutex.h>
|
||||
#include <spinscale/sharedResourceGroup.h>
|
||||
|
||||
namespace smo {
|
||||
@@ -25,13 +25,6 @@ class DeviceReattacher;
|
||||
class DeviceManager
|
||||
{
|
||||
public:
|
||||
struct DeviceAttachmentIndResult
|
||||
{
|
||||
bool success = false;
|
||||
std::shared_ptr<DeviceRole> deviceRole;
|
||||
std::shared_ptr<DeviceAttachmentSpec> deviceSpec;
|
||||
};
|
||||
|
||||
static DeviceManager& getInstance()
|
||||
{
|
||||
static DeviceManager instance;
|
||||
@@ -52,31 +45,27 @@ public:
|
||||
|
||||
static const std::string stringifyDeviceSpecs(void);
|
||||
|
||||
mrntt::MrnttViralPostingInvoker<DeviceAttachmentIndResult>
|
||||
mrntt::MrnttViralPostingInvoker<std::shared_ptr<DeviceRole>>
|
||||
newDeviceAttachmentSpecIndCReq(const DeviceAttachmentSpec &spec);
|
||||
|
||||
mrntt::MrnttViralPostingInvoker<DeviceAttachmentIndResult>
|
||||
mrntt::MrnttViralPostingInvoker<void>
|
||||
removeDeviceAttachmentSpecCReq(const DeviceAttachmentSpec &spec);
|
||||
|
||||
mrntt::MrnttViralPostingInvoker<stim_buff::StimBuffDeviceOpResult>
|
||||
mrntt::MrnttViralPostingInvoker<void>
|
||||
attachStimBuffDeviceCReq(
|
||||
const std::shared_ptr<DeviceAttachmentSpec>& spec);
|
||||
|
||||
mrntt::MrnttViralPostingInvoker<stim_buff::StimBuffDeviceOpResult>
|
||||
mrntt::MrnttViralPostingInvoker<void>
|
||||
detachStimBuffDeviceCReq(
|
||||
const std::shared_ptr<DeviceAttachmentSpec>& spec);
|
||||
|
||||
mrntt::MrnttViralPostingInvoker<sscl::MultiOperationResultSet>
|
||||
attachAllUnattachedDevicesFromCReq(
|
||||
const std::shared_ptr<std::vector<DeviceAttachmentSpec>> &specs);
|
||||
|
||||
mrntt::MrnttViralPostingInvoker<sscl::MultiOperationResultSet>
|
||||
mrntt::MrnttViralPostingInvoker<void>
|
||||
attachAllUnattachedDevicesFromKnownListCReq();
|
||||
|
||||
mrntt::MrnttViralPostingInvoker<sscl::MultiOperationResultSet>
|
||||
mrntt::MrnttViralPostingInvoker<void>
|
||||
attachAllUnattachedDevicesFromCmdlineCReq();
|
||||
|
||||
mrntt::MrnttViralPostingInvoker<sscl::MultiOperationResultSet>
|
||||
mrntt::MrnttViralPostingInvoker<void>
|
||||
detachAllAttachedDeviceRolesCReq();
|
||||
|
||||
private:
|
||||
@@ -88,6 +77,10 @@ private:
|
||||
DeviceManager(const DeviceManager&) = delete;
|
||||
DeviceManager& operator=(const DeviceManager&) = delete;
|
||||
|
||||
mrntt::MrnttViralPostingInvoker<sscl::MultiOperationResultSetWithException>
|
||||
attachAllUnattachedDevicesFromCReq(
|
||||
const std::shared_ptr<std::vector<DeviceAttachmentSpec>> &specs);
|
||||
|
||||
public:
|
||||
struct Resources
|
||||
{
|
||||
|
||||
@@ -9,7 +9,6 @@
|
||||
#include <optional>
|
||||
#include <boost/asio/deadline_timer.hpp>
|
||||
#include <marionette/marionetteThread.h>
|
||||
#include <spinscale/multiOperationResultSet.h>
|
||||
#include <spinscale/syncCancelerForAsyncWork.h>
|
||||
|
||||
namespace smo {
|
||||
@@ -37,17 +36,17 @@ private:
|
||||
void onTimeout(const boost::system::error_code& error);
|
||||
void holdReattachCReq();
|
||||
|
||||
mrntt::MrnttNonViralPostingInvoker reattachKnownListCReq(
|
||||
sscl::co::ExplicitPostTarget postTarget,
|
||||
mrntt::MrnttNonViralNonPostingInvoker reattachKnownListCReq(
|
||||
std::exception_ptr &exceptionPtr,
|
||||
std::function<void()> callback);
|
||||
|
||||
DeviceManager &parent;
|
||||
// io_context thread for timer and non-posting reattach shell (see ctor).
|
||||
std::shared_ptr<sscl::ComponentThread> ioThread;
|
||||
sscl::SyncCancelerForAsyncWork deviceReattacherCanceler;
|
||||
boost::asio::deadline_timer timer;
|
||||
std::exception_ptr reattachLifetimeExceptionPtr;
|
||||
std::optional<mrntt::MrnttNonViralPostingInvoker> reattachCReqInvoker;
|
||||
std::optional<mrntt::MrnttNonViralNonPostingInvoker> reattachCReqInvoker;
|
||||
bool reattachOpInFlight = false;
|
||||
std::chrono::steady_clock::time_point lastReattachReqTimestamp{};
|
||||
};
|
||||
|
||||
@@ -20,6 +20,9 @@ using MrnttPostingPromise =
|
||||
using MrnttNonViralPostingInvoker =
|
||||
sscl::co::NonViralPostingInvoker<MrnttPostingPromise>;
|
||||
|
||||
using MrnttNonViralNonPostingInvoker =
|
||||
sscl::co::NonViralNonPostingInvoker;
|
||||
|
||||
template <typename T>
|
||||
using MrnttViralPostingInvoker =
|
||||
sscl::co::ViralPostingInvoker<MrnttPostingPromise, T>;
|
||||
|
||||
@@ -41,8 +41,9 @@ public:
|
||||
|
||||
std::optional<std::shared_ptr<StimBuffApiLib>> getStimBuffApiLib(
|
||||
const std::string& libraryPath);
|
||||
std::optional<std::shared_ptr<StimBuffApiLib>> getStimBuffApiLibByApiName(
|
||||
std::optional<std::shared_ptr<StimBuffApiLib>> findStimBuffApiLibByApiName(
|
||||
const std::string& apiName);
|
||||
StimBuffApiLib &getStimBuffApiLibByApiName(const std::string& apiName);
|
||||
void unloadStimBuffApiLib(const std::string& libraryPath);
|
||||
|
||||
void loadAllStimBuffApiLibsFromOptions(
|
||||
|
||||
@@ -230,7 +230,7 @@ StimBuffApiManager::getStimBuffApiLib(const std::string& libraryPath)
|
||||
}
|
||||
|
||||
std::optional<std::shared_ptr<StimBuffApiLib>>
|
||||
StimBuffApiManager::getStimBuffApiLibByApiName(const std::string& apiName)
|
||||
StimBuffApiManager::findStimBuffApiLibByApiName(const std::string& apiName)
|
||||
{
|
||||
auto &libs = getInstance().s.rsrc.stimBuffApiLibs;
|
||||
auto it = std::find_if(libs.begin(), libs.end(),
|
||||
@@ -243,6 +243,19 @@ StimBuffApiManager::getStimBuffApiLibByApiName(const std::string& apiName)
|
||||
return std::nullopt;
|
||||
}
|
||||
|
||||
StimBuffApiLib &StimBuffApiManager::getStimBuffApiLibByApiName(
|
||||
const std::string& apiName)
|
||||
{
|
||||
auto libOpt = findStimBuffApiLibByApiName(apiName);
|
||||
if (!libOpt)
|
||||
{
|
||||
throw std::runtime_error(
|
||||
std::string(__func__) + ": No library for API '" + apiName + "'");
|
||||
}
|
||||
|
||||
return *libOpt.value();
|
||||
}
|
||||
|
||||
void StimBuffApiManager::unloadStimBuffApiLib(const std::string& libraryPath)
|
||||
{
|
||||
auto &libs = getInstance().s.rsrc.stimBuffApiLibs;
|
||||
|
||||
Reference in New Issue
Block a user