Add retry mechanism and dynamic model loading to SONIC framework

HeterogeneousCore/SonicCore/BuildFile.xml

@@ -2,6 +2,7 @@
 <use name="FWCore/Concurrency"/>
 <use name="FWCore/MessageLogger"/>
 <use name="FWCore/ParameterSet"/>
+<use name="FWCore/PluginManager"/>
 <use name="FWCore/Utilities"/>
 <export>
   <lib name="1"/>

HeterogeneousCore/SonicCore/README.md

@@ -42,13 +42,18 @@ The python configuration for the producer should include a dedicated `PSet` for
 process.MyProducer = cms.EDProducer("MyProducer",
     Client = cms.PSet(
         # necessary client options go here
-        mode = cms.string("Sync"),
-        allowedTries = cms.untracked.uint32(0),
+        mode = cms.string(""),
+        Retry = cms.VPSet(
+          cms.PSet(
+            retryType = cms.string('RetrySameServerAction'),
+            allowedTries = cms.untracked.uint32(0)
+          )
+        )
     )
 )
 ```
 These parameters can be prepopulated and validated by the client using `fillDescriptions()`.
-The `mode` and `allowedTries` parameters are always necessary (example values are shown here, but other values are also allowed).
+The `mode` and `Retry` parameters are always necessary (example values are shown here, but other values are also allowed).
 These parameters are described in the next section.
 
 In addition, there is a `SonicOneEDAnalyzer` class template for user analysis, e.g. to produce simple ROOT files.
@@ -99,6 +104,7 @@ The `SonicClient` has three available modes:
 * `PseudoAsync`: turns a synchronous, blocking call into an asynchronous, non-blocking call, by waiting for the result in a separate `std::thread`.
 
 `Async` is the most efficient, but can only be used if asynchronous, non-blocking calls are supported by the communication protocol in use.
+When a fallback CPU server is used, `Sync` mode is enforced to avoid contention, user's configuration of `mode` will be respected for other cases with `Async` mode as the default.
 
 In addition, as indicated, the input and output data types must be specified.
 (If both types are the same, only the input type needs to be specified.)
@@ -110,9 +116,9 @@ For the `Sync` and `PseudoAsync` modes, `finish()` should be called at the end o
 For the `Async` mode, `finish()` should be called inside the communication protocol callback function (implementations may vary).
 
 When `finish()` is called, the success or failure of the call should be conveyed.
-If a call fails, it can optionally be retried. This is only allowed if the call failure does not cause an exception.
+If a call fails without raising an exception, it can be retried through an ordered chain of retry actions rather than a single fixed number of tries. 
+The chain is configured per client through a `Retry` `VPSet` parameter, where each `PSet` specifies a `retryType` plus any action-specific parameters; VPSet order is try order. 
 Therefore, if retrying is desired, any exception should be converted to a `LogWarning` or `LogError` message by the client.
-A Python configuration parameter can be provided to enable retries with a specified maximum number of allowed tries.
 
 The client must also provide a static method `fillPSetDescription()` to populate its parameters in the `fillDescriptions()` for the producers that use the client:
 ```cpp
@@ -126,6 +132,12 @@ void MyClient::fillPSetDescription(edm::ParameterSetDescription& iDesc) {
 
 As indicated, the `fillBasePSetDescription()` function should always be applied to the `descClient` object,
 to ensure that it includes the necessary parameters.
-(Calling `fillBasePSetDescription(descClient, false)` will omit the `allowedTries` parameter, disabling retries.)
+`Retry` is a vector of `PSet` with the parameters `retryType` and `allowedTries`.
+`retryType` is the action type that inherents from `RetryActionBase`.
+`allowedTries` is a parameter consumed by `RetrySameServerAction`.
+
+The default `Retry` action is a single `RetryFallbackServerAction`.
+To disable `Retry`, leave the `Retry` `VPSet` empty.
+
 
 Example client code can be found in the `interface` and `src` directories of the other Sonic packages in this repository.

HeterogeneousCore/SonicCore/interface/RetryActionBase.h

@@ -0,0 +1,36 @@
+#ifndef HeterogeneousCore_SonicCore_RetryActionBase
+#define HeterogeneousCore_SonicCore_RetryActionBase
+
+#include "FWCore/PluginManager/interface/PluginFactory.h"
+#include "FWCore/ParameterSet/interface/ParameterSet.h"
+#include "HeterogeneousCore/SonicCore/interface/SonicClientBase.h"
+#include <memory>
+#include <string>
+
+// Base class for retry actions
+class RetryActionBase {
+public:
+  RetryActionBase(const edm::ParameterSet& conf, SonicClientBase* client);
+  virtual ~RetryActionBase() = default;
+
+  bool shouldRetry() const { return shouldRetry_; }  // Getter for shouldRetry_
+
+  virtual void retry() = 0;  // Pure virtual function for execution logic
+  virtual void start() = 0;  // Pure virtual function for execution logic for initialization
+
+protected:
+  void eval();                // interface for calling evaluate in client
+  void finish(bool success);  // interface for calling finish directly in client
+
+protected:
+  SonicClientBase* client_;
+  bool shouldRetry_;  // Flag to track if further retries should happen
+};
+
+// Define the factory for creating retry actions
+using RetryActionFactory =
+    edmplugin::PluginFactory<RetryActionBase*(const edm::ParameterSet&, SonicClientBase* client)>;
+
+#endif
+
+#define DEFINE_RETRY_ACTION(type) DEFINE_EDM_PLUGIN(RetryActionFactory, type, #type);

HeterogeneousCore/SonicCore/interface/SonicClientBase.h

@@ -9,12 +9,15 @@
 #include "HeterogeneousCore/SonicCore/interface/SonicDispatcherPseudoAsync.h"
 
 #include <string>
+#include <vector>
 #include <exception>
 #include <memory>
 #include <optional>
 
 enum class SonicMode { Sync = 1, Async = 2, PseudoAsync = 3 };
 
+class RetryActionBase;
+
 class SonicClientBase {
 public:
   //constructor
@@ -37,10 +40,11 @@ class SonicClientBase {
   virtual void reset() {}
 
   //provide base params
-  static void fillBasePSetDescription(edm::ParameterSetDescription& desc, bool allowRetry = true);
+  static void fillBasePSetDescription(edm::ParameterSetDescription& desc);
 
 protected:
   void setMode(SonicMode mode);
+  void setUserMode(const std::string& userMode);
 
   virtual void evaluate() = 0;
 
@@ -54,14 +58,25 @@ class SonicClientBase {
   SonicMode mode_;
   bool verbose_;
   std::unique_ptr<SonicDispatcher> dispatcher_;
-  unsigned allowedTries_, tries_;
+  unsigned totalTries_;
   std::optional<edm::WaitingTaskWithArenaHolder> holder_;
 
+  // Use a unique_ptr with a custom deleter to avoid incomplete type issues
+  struct RetryDeleter {
+    void operator()(RetryActionBase* ptr) const;
+  };
+
+  using RetryActionPtr = std::unique_ptr<RetryActionBase, RetryDeleter>;
+  std::vector<RetryActionPtr> retryActions_;
+
   //for logging/debugging
   std::string debugName_, clientName_, fullDebugName_;
+  //remember what user set at config time
+  std::string userMode_;
 
   friend class SonicDispatcher;
   friend class SonicDispatcherPseudoAsync;
+  friend class RetryActionBase;
 };
 
 #endif

HeterogeneousCore/SonicCore/plugins/BuildFile.xml

@@ -0,0 +1,6 @@
+<use name="FWCore/Framework"/>
+<use name="FWCore/PluginManager"/>
+<use name="FWCore/ParameterSet"/>
+<use name="HeterogeneousCore/SonicCore"/>
+<plugin file="*.cc" name="pluginHeterogeneousCoreSonicCore"/>
+

HeterogeneousCore/SonicCore/plugins/RetrySameServerAction.cc

@@ -0,0 +1,28 @@
+#include "HeterogeneousCore/SonicCore/interface/RetryActionBase.h"
+#include "HeterogeneousCore/SonicCore/interface/SonicClientBase.h"
+
+class RetrySameServerAction : public RetryActionBase {
+public:
+  RetrySameServerAction(const edm::ParameterSet& pset, SonicClientBase* client)
+      : RetryActionBase(pset, client), allowedTries_(pset.getUntrackedParameter<unsigned>("allowedTries", 0)) {}
+
+  void start() override { tries_ = 0; };
+
+protected:
+  void retry() override;
+
+private:
+  unsigned allowedTries_, tries_;
+};
+
+void RetrySameServerAction::retry() {
+  ++tries_;
+  //if max retries has not been exceeded, call evaluate again
+  if (tries_ >= allowedTries_) {
+    shouldRetry_ = false;  // Flip flag when max retries are reached
+    edm::LogInfo("RetrySameServerAction") << "Max retry attempts reached. No further retries.";
+  }
+  eval();
+}
+
+DEFINE_RETRY_ACTION(RetrySameServerAction)

HeterogeneousCore/SonicCore/src/RetryActionBase.cc

@@ -0,0 +1,15 @@
+#include "HeterogeneousCore/SonicCore/interface/RetryActionBase.h"
+
+// Constructor implementation
+RetryActionBase::RetryActionBase(const edm::ParameterSet& conf, SonicClientBase* client)
+    : client_(client), shouldRetry_(true) {
+  if (client_ == nullptr) {
+    throw cms::Exception("RetryActionBase") << "client pointer cannot be null";
+  }
+}
+
+void RetryActionBase::eval() { client_->evaluate(); }
+
+void RetryActionBase::finish(bool success) { client_->finish(success); }
+
+EDM_REGISTER_PLUGINFACTORY(RetryActionFactory, "RetryActionFactory");

HeterogeneousCore/SonicCore/src/SonicClientBase.cc

@@ -1,28 +1,50 @@
 #include "HeterogeneousCore/SonicCore/interface/SonicClientBase.h"
+#include "HeterogeneousCore/SonicCore/interface/RetryActionBase.h"
 #include "FWCore/Utilities/interface/Exception.h"
 #include "FWCore/ParameterSet/interface/allowedValues.h"
 
+// Custom deleter implementation
+void SonicClientBase::RetryDeleter::operator()(RetryActionBase* ptr) const { delete ptr; }
+
 SonicClientBase::SonicClientBase(const edm::ParameterSet& params,
                                  const std::string& debugName,
                                  const std::string& clientName)
-    : allowedTries_(params.getUntrackedParameter<unsigned>("allowedTries", 0)),
-      debugName_(debugName),
+    : debugName_(debugName),
       clientName_(clientName),
-      fullDebugName_(debugName_) {
+      fullDebugName_(debugName_),
+      userMode_(params.getParameter<std::string>("mode")) {
   if (!clientName_.empty())
     fullDebugName_ += ":" + clientName_;
 
-  std::string modeName(params.getParameter<std::string>("mode"));
-  if (modeName == "Sync")
+  const auto& retryPSetList = params.getParameter<std::vector<edm::ParameterSet>>("Retry");
+
+  for (const auto& retryPSet : retryPSetList) {
+    const std::string& actionType = retryPSet.getParameter<std::string>("retryType");
+
+    auto retryAction = RetryActionFactory::get()->create(actionType, retryPSet, this);
+    if (retryAction) {
+      //Convert to  RetryActionPtr Type from raw pointer of retryAction
+      retryActions_.emplace_back(RetryActionPtr(retryAction.release()));
+    } else {
+      throw cms::Exception("Configuration")
+          << "Unknown Retry type " << actionType << " for SonicClient: " << fullDebugName_;
+    }
+  }
+
+  setUserMode(userMode_);
+}
+void SonicClientBase::setUserMode(const std::string& userMode) {
+  if (userMode == "Sync")
     setMode(SonicMode::Sync);
-  else if (modeName == "Async")
+  else if (userMode == "Async")
     setMode(SonicMode::Async);
-  else if (modeName == "PseudoAsync")
+  else if (userMode == "PseudoAsync")
+    setMode(SonicMode::PseudoAsync);
+  else if (userMode.empty())
     setMode(SonicMode::PseudoAsync);
   else
-    throw cms::Exception("Configuration") << "Unknown mode for SonicClient: " << modeName;
+    throw cms::Exception("Configuration") << "Unknown mode for SonicClient: " << userMode;
 }
-
 void SonicClientBase::setMode(SonicMode mode) {
   if (dispatcher_ and mode_ == mode)
     return;
@@ -40,24 +62,33 @@ void SonicClientBase::start(edm::WaitingTaskWithArenaHolder holder) {
   holder_ = std::move(holder);
 }
 
-void SonicClientBase::start() { tries_ = 0; }
+void SonicClientBase::start() {
+  totalTries_ = 0;
+  // initialize all actions
+  for (auto& action : retryActions_) {
+    action->start();
+  }
+}
 
 void SonicClientBase::finish(bool success, std::exception_ptr eptr) {
   //retries are only allowed if no exception was raised
   if (!success and !eptr) {
-    ++tries_;
-    //if max retries has not been exceeded, call evaluate again
-    if (tries_ < allowedTries_) {
-      evaluate();
-      //avoid calling doneWaiting() twice
-      return;
-    }
-    //prepare an exception if exceeded
-    else {
-      edm::Exception ex(edm::errors::ExternalFailure);
-      ex << "SonicCallFailed: call failed after max " << tries_ << " tries";
-      eptr = make_exception_ptr(ex);
+    ++totalTries_;
+    edm::LogInfo("SonicClientBase") << "finish: failed after total tries of " << totalTries_;
+    for (const auto& action : retryActions_) {
+      if (action->shouldRetry()) {
+        edm::LogInfo("SonicClientBase") << "Calling retry()";
+        // retry() must trigger eval() or finish()
+        action->retry();
+        return;
+      }
     }
+    //prepare an exception if no more retry actions left
+    edm::LogInfo("SonicClientBase") << "SonicCallFailed: call failed, no retry actions available after " << totalTries_
+                                    << " tries.";
+    edm::Exception ex(edm::errors::ExternalFailure);
+    ex << "SonicCallFailed: call failed, no retry actions available after " << totalTries_ << " tries.";
+    eptr = make_exception_ptr(ex);
   }
   if (holder_) {
     holder_->doneWaiting(eptr);
@@ -70,11 +101,10 @@ void SonicClientBase::finish(bool success, std::exception_ptr eptr) {
     reset();
 }
 
-void SonicClientBase::fillBasePSetDescription(edm::ParameterSetDescription& desc, bool allowRetry) {
+void SonicClientBase::fillBasePSetDescription(edm::ParameterSetDescription& desc) {
   //restrict allowed values
-  desc.ifValue(edm::ParameterDescription<std::string>("mode", "PseudoAsync", true),
-               edm::allowedValues<std::string>("Sync", "Async", "PseudoAsync"));
-  if (allowRetry)
-    desc.addUntracked<unsigned>("allowedTries", 0);
+  desc.ifValue(edm::ParameterDescription<std::string>("mode", "", true),
+               edm::allowedValues<std::string>("Sync", "Async", "PseudoAsync", ""));
+  desc.add("sonicClientBase", desc);
   desc.addUntracked<bool>("verbose", false);
 }

HeterogeneousCore/SonicCore/test/DummyClient.h

@@ -36,7 +36,7 @@ class DummyClient : public SonicClient<int> {
     this->output_ = this->input_ * factor_;
 
     //simulate a failure
-    if (this->tries_ < fails_)
+    if (this->totalTries_ < fails_)
       this->finish(false);
     else
       this->finish(true);

HeterogeneousCore/SonicCore/test/sonicTestAna_cfg.py

@@ -16,16 +16,27 @@
         mode = cms.string("Sync"),
         factor = cms.int32(-1),
         wait = cms.int32(10),
-        allowedTries = cms.untracked.uint32(0),
         fails = cms.uint32(0),
+        Retry = cms.VPSet(
+          cms.PSet(
+            retryType = cms.string('RetrySameServerAction'),
+            allowedTries = cms.untracked.uint32(0),
+          )
+        )
     ),
 )
 
 process.dummySyncAnaRetry = process.dummySyncAna.clone(
     Client = dict(
         wait = 2,
-        allowedTries = 2,
         fails = 1,
+        Retry = cms.VPSet(
+          cms.PSet(
+            retryType = cms.string('RetrySameServerAction'),
+            allowedTries = cms.untracked.uint32(2),
+          )
+        )
+
     )
 )