Simplify SonicCore and redesign SonicTriton

HeterogeneousCore/SonicCore/BuildFile.xml

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

HeterogeneousCore/SonicCore/README.md

@@ -42,10 +42,14 @@ 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),
     )
 )
 ```
-These parameters can be prepopulated and validated by the client using `fillDescriptions` (see below).
+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).
+These parameters are described in the next section.
 
 An example producer can be found in the [test](./test) folder.
 
@@ -62,9 +66,9 @@ To implement a concrete client, the following skeleton can be used for the `.h`
 #define HeterogeneousCore_MyPackage_MyClient
 
 #include "FWCore/ParameterSet/interface/ParameterSet.h"
-#include "HeterogeneousCore/SonicCore/interface/SonicClient*.h"
+#include "HeterogeneousCore/SonicCore/interface/SonicClient.h"
 
-class MyClient : public SonicClient*<Input,Output> {
+class MyClient : public SonicClient<Input,Output> {
 public:
   MyClient(const edm::ParameterSet& params);
 
@@ -77,12 +81,12 @@ protected:
 #endif
 ```
 
-The generic `SonicClient*` should be replaced with one of the available modes:
-* `SonicClientSync`: synchronous call, blocks until the result is returned.
-* `SonicClientAsync`: asynchronous, non-blocking call.
-* `SonicClientPseudoAsync`: turns a synchronous, blocking call into an asynchronous, non-blocking call, by waiting for the result in a separate `std::thread`.
+The `SonicClient` has three available modes:
+* `Sync`: synchronous call, blocks until the result is returned.
+* `Async`: asynchronous, non-blocking call.
+* `PseudoAsync`: turns a synchronous, blocking call into an asynchronous, non-blocking call, by waiting for the result in a separate `std::thread`.
 
-`SonicClientAsync` is the most efficient, but can only be used if asynchronous, non-blocking calls are supported by the communication protocol in use.
+`Async` is the most efficient, but can only be used if asynchronous, non-blocking calls are supported by the communication protocol in use.
 
 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.)
@@ -96,19 +100,20 @@ For the `Async` mode, `finish()` should be called inside the communication proto
 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.
 Therefore, if retrying is desired, any exception should be converted to a `LogWarning` or `LogError` message by the client.
-To enable retries with a specified maximum number of allowed tries (possibly obtained from a Python configuration parameter), the client should implement the following:
-```cpp
-protected:
-  unsigned allowedTries() const override;
-```
+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:
+The client must also provide a static method `fillPSetDescription()` to populate its parameters in the `fillDescriptions()` for the producers that use the client:
 ```cpp
 void MyClient::fillPSetDescription(edm::ParameterSetDescription& iDesc) {
   edm::ParameterSetDescription descClient;
+  fillBasePSetDescription(descClient);
   //add parameters
   iDesc.add<edm::ParameterSetDescription>("Client",descClient);
 }
 ```
 
+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.)
+
 Example client code can be found in the `interface` and `src` directories of the other Sonic packages in this repository.

HeterogeneousCore/SonicCore/interface/SonicClient.h

@@ -0,0 +1,15 @@
+#ifndef HeterogeneousCore_SonicCore_SonicClient
+#define HeterogeneousCore_SonicCore_SonicClient
+
+#include "HeterogeneousCore/SonicCore/interface/SonicClientBase.h"
+#include "HeterogeneousCore/SonicCore/interface/SonicClientTypes.h"
+
+//convenience definition for multiple inheritance (base and types)
+template <typename InputT, typename OutputT = InputT>
+class SonicClient : public SonicClientBase, public SonicClientTypes<InputT, OutputT> {
+public:
+  //constructor
+  SonicClient(const edm::ParameterSet& params) : SonicClientBase(params), SonicClientTypes<InputT, OutputT>() {}
+};
+
+#endif

HeterogeneousCore/SonicCore/interface/SonicClientBase.h

@@ -1,78 +1,60 @@
 #ifndef HeterogeneousCore_SonicCore_SonicClientBase
 #define HeterogeneousCore_SonicCore_SonicClientBase
 
+#include "FWCore/ParameterSet/interface/ParameterSet.h"
+#include "FWCore/ParameterSet/interface/ParameterSetDescription.h"
 #include "FWCore/Concurrency/interface/WaitingTaskWithArenaHolder.h"
 #include "FWCore/MessageLogger/interface/MessageLogger.h"
+#include "HeterogeneousCore/SonicCore/interface/SonicDispatcher.h"
+#include "HeterogeneousCore/SonicCore/interface/SonicDispatcherPseudoAsync.h"
 
 #include <string>
 #include <chrono>
 #include <exception>
+#include <memory>
+
+enum class SonicMode { Sync = 1, Async = 2, PseudoAsync = 3 };
 
 class SonicClientBase {
 public:
   //constructor
-  SonicClientBase() : tries_(0) {}
+  SonicClientBase(const edm::ParameterSet& params);
 
   //destructor
   virtual ~SonicClientBase() = default;
 
-  void setDebugName(const std::string& debugName) {
-    debugName_ = debugName;
-    fullDebugName_ = debugName_;
-    if (!clientName_.empty())
-      fullDebugName_ += ":" + clientName_;
-  }
+  void setDebugName(const std::string& debugName);
   const std::string& debugName() const { return debugName_; }
   const std::string& clientName() const { return clientName_; }
 
   //main operation
-  virtual void dispatch(edm::WaitingTaskWithArenaHolder holder) = 0;
+  virtual void dispatch(edm::WaitingTaskWithArenaHolder holder) { dispatcher_->dispatch(std::move(holder)); }
+
+  //helper: does nothing by default
+  virtual void reset() {}
+
+  //provide base params
+  static void fillBasePSetDescription(edm::ParameterSetDescription& desc, bool allowRetry = true);
 
 protected:
   virtual void evaluate() = 0;
 
-  //this should be overridden by clients that allow retries
-  virtual unsigned allowedTries() const { return 0; }
-
-  void setStartTime() {
-    tries_ = 0;
-    if (debugName_.empty())
-      return;
-    t0_ = std::chrono::high_resolution_clock::now();
-  }
-
-  void finish(bool success, std::exception_ptr eptr = std::exception_ptr{}) {
-    //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 {
-        cms::Exception ex("SonicCallFailed");
-        ex << "call failed after max " << tries_ << " tries";
-        eptr = make_exception_ptr(ex);
-      }
-    }
-    if (!debugName_.empty()) {
-      auto t1 = std::chrono::high_resolution_clock::now();
-      edm::LogInfo(fullDebugName_) << "Client time: "
-                                   << std::chrono::duration_cast<std::chrono::microseconds>(t1 - t0_).count();
-    }
-    holder_.doneWaiting(eptr);
-  }
+  void start(edm::WaitingTaskWithArenaHolder holder);
+
+  void finish(bool success, std::exception_ptr eptr = std::exception_ptr{});
 
   //members
-  unsigned tries_;
+  SonicMode mode_;
+  std::unique_ptr<SonicDispatcher> dispatcher_;
+  unsigned allowedTries_, tries_;
   edm::WaitingTaskWithArenaHolder holder_;
 
   //for logging/debugging
   std::string clientName_, debugName_, fullDebugName_;
   std::chrono::time_point<std::chrono::high_resolution_clock> t0_;
+
+  friend class SonicDispatcher;
+  friend class SonicDispatcherPseudoAsync;
 };
 
 #endif

HeterogeneousCore/SonicCore/interface/SonicDispatcher.h

@@ -0,0 +1,23 @@
+#ifndef HeterogeneousCore_SonicCore_SonicDispatcher
+#define HeterogeneousCore_SonicCore_SonicDispatcher
+
+#include "FWCore/Concurrency/interface/WaitingTaskWithArenaHolder.h"
+
+class SonicClientBase;
+
+class SonicDispatcher {
+public:
+  //constructor
+  SonicDispatcher(SonicClientBase* client) : client_(client) {}
+
+  //destructor
+  virtual ~SonicDispatcher() = default;
+
+  //main operation
+  virtual void dispatch(edm::WaitingTaskWithArenaHolder holder);
+
+protected:
+  SonicClientBase* client_;
+};
+
+#endif

HeterogeneousCore/SonicCore/interface/SonicDispatcherPseudoAsync.h

@@ -0,0 +1,38 @@
+#ifndef HeterogeneousCore_SonicCore_SonicDispatcherPseudoAsync
+#define HeterogeneousCore_SonicCore_SonicDispatcherPseudoAsync
+
+#include "HeterogeneousCore/SonicCore/interface/SonicDispatcher.h"
+
+#include <memory>
+#include <condition_variable>
+#include <mutex>
+#include <thread>
+#include <atomic>
+#include <exception>
+
+class SonicClientBase;
+
+//pretend to be async + non-blocking by waiting for blocking calls to return in separate std::thread
+class SonicDispatcherPseudoAsync : public SonicDispatcher {
+public:
+  //constructor
+  SonicDispatcherPseudoAsync(SonicClientBase* client);
+
+  //destructor
+  ~SonicDispatcherPseudoAsync() override;
+
+  //main operation
+  void dispatch(edm::WaitingTaskWithArenaHolder holder) override;
+
+private:
+  void waitForNext();
+
+  //members
+  bool hasCall_;
+  std::mutex mutex_;
+  std::condition_variable cond_;
+  std::atomic<bool> stop_;
+  std::unique_ptr<std::thread> thread_;
+};
+
+#endif