Merge branch 'devel-alf-publish-sca'

Author: PascalBoeschoten

src/CommandLineUtilities/AliceLowlevelFrontend/AliceLowlevelFrontend.h

@@ -16,6 +16,7 @@
 #include <boost/algorithm/string/predicate.hpp>
 #include <boost/format.hpp>
 #include "AlfException.h"
+#include "Sca.h"
 
 namespace AliceO2 {
 namespace roc {
@@ -161,6 +162,31 @@ class PublishRpc : DimRpcInfoWrapper
     }
 };
 
+class PublishScaRpc : DimRpcInfoWrapper
+{
+  public:
+    PublishScaRpc(const std::string& serviceName)
+      : DimRpcInfoWrapper(serviceName)
+    {
+    }
+
+    void publish(std::string dnsName, double frequency, const std::vector<Sca::CommandData>& commandDataPairs)
+    {
+      std::ostringstream stream;
+      stream << dnsName << ';';
+      for (size_t i = 0; i < commandDataPairs.size(); ++i) {
+        stream << commandDataPairs[i].command << ',' << commandDataPairs[i].data;
+        if ((i + 1) < commandDataPairs.size()) {
+          stream << '\n';
+        }
+      }
+      stream << ';' << frequency;
+      printf("Publish SCA: %s\n", stream.str().c_str());
+      setString(stream.str());
+      getString();
+    }
+};
+
 class PublishStopRpc: DimRpcInfoWrapper
 {
   public:

src/CommandLineUtilities/AliceLowlevelFrontend/ProgramAliceLowlevelFrontendClient.cxx

@@ -69,10 +69,13 @@ class ProgramAliceLowlevelFrontendClient: public Program
       Alf::ScaGpioWriteRpc scaGpioWriteRpc(names.scaGpioWrite());
       Alf::ScaWriteSequence scaWriteSequence(names.scaWriteSequence());
       Alf::PublishRpc publishRpc(names.publishStartCommandRpc());
+      Alf::PublishScaRpc publishScaRpc(names.publishScaStartCommandRpc());
 
       publishRpc.publish("ALF/TEST/1", 1.0, {0x1fc});
       publishRpc.publish("ALF/TEST/2", 3.0, {0x100, 0x104, 0x108});
 
+      publishScaRpc.publish("ALF/TEST/SCA_1", 1.0, {{0x0, 0x1}, {0x10, 0x11}});
+
       for (int i = 0; i < 10; ++i) {
         cout << "SCA GPIO write '" << i << "'" << endl;
         cout << "  result: " << scaGpioWriteRpc.write(i) << endl;
@@ -135,6 +138,7 @@ class ProgramAliceLowlevelFrontendClient: public Program
       Alf::PublishStopRpc publishStopRpc(names.publishStopCommandRpc());
       publishStopRpc.stop("ALF/TEST/1");
       publishStopRpc.stop("ALF/TEST/2");
+      publishStopRpc.stop("ALF/TEST/SCA_1");
     }
 
     int mSerialNumber;

src/CommandLineUtilities/AliceLowlevelFrontend/ProgramAliceLowlevelFrontendServer.cxx

@@ -1,6 +1,17 @@
 /// \file ProgramAliceLowlevelFrontendServer.cxx
 /// \brief Utility that starts the ALICE Lowlevel Frontend (ALF) DIM server
 ///
+/// This file contains a bunch of classes that together form the server part of ALF.
+/// It is single-threaded, because an earlier multi-threaded implementation ran into strange locking issues with DIM's
+/// thread, and it was determined that it would cost less time to rewrite than to debug.
+///
+/// The idea is that the DIM thread calls the RPC handler functions of the ProgramAliceLowlevelFrontendServer class.
+/// These handlers then, depending on the RPC:
+/// a) Execute the request immediately (such as for register reads and writes)
+/// b) Put a corresponding command object in a lock-free thread-safe queue (such as for publish start/stop commands),
+///    the mCommandQueue. The main thread of ProgramAliceLowlevelFrontendServer periodically takes commands from this
+///    queue and handles them.
+///
 /// \author Pascal Boeschoten ([email protected])
 
 #include "Common/Program.h"
@@ -65,8 +76,19 @@ static std::vector<T> lexicalCastVector(const std::vector<U>& items)
 struct ServiceDescription
 {
     std::string dnsName;
-    std::vector<uintptr_t> addresses;
     std::chrono::milliseconds interval;
+
+    struct Register
+    {
+        std::vector<uintptr_t> addresses;
+    };
+
+    struct ScaSequence
+    {
+        std::vector<Sca::CommandData> commandDataPairs;
+    };
+
+    boost::variant<Register, ScaSequence> type;
 };
 
 /// Class that handles adding/removing publishers
@@ -101,7 +123,7 @@ class PublisherRegistry
       for (auto& kv: mServices) {
         auto& service = *kv.second;
         if (service.nextUpdate < now) {
-          service.updateRegisterValues(*mChannel.get());
+          service.updateValues(*mChannel.get());
           service.advanceUpdateTime();
           if (service.nextUpdate < next) {
             next = service.nextUpdate;
@@ -120,31 +142,67 @@ class PublisherRegistry
         {
           mDescription = description;
           nextUpdate = std::chrono::steady_clock::now();
-          registerValues.resize(mDescription.addresses.size());
+
+          Visitor::apply(mDescription.type,
+            [&](const ServiceDescription::Register& type){
+              registerValues.resize(type.addresses.size());
+
+              getInfoLogger() << "Starting publisher '" << mDescription.dnsName << "' with "
+                << type.addresses.size() << " address(es) at interval "
+                << mDescription.interval.count() << "ms" << endm;
+            },
+            [&](const ServiceDescription::ScaSequence& type){
+              registerValues.resize(type.commandDataPairs.size() * 2); // Two result 32-bit integers per pair
+
+              getInfoLogger() << "Starting SCA publisher '" << mDescription.dnsName << "' with "
+                << type.commandDataPairs.size() << " commands(s) at interval "
+                << mDescription.interval.count() << "ms" << endm;
+            }
+          );
 
           auto format = (b::format("I:%1%") % registerValues.size()).str();
           dimService = std::make_unique<DimService>(mDescription.dnsName.c_str(), format.c_str(), registerValues.data(),
             registerValues.size());
-
-          getInfoLogger() << "Starting publisher '" << mDescription.dnsName << "' with "
-            << mDescription.addresses.size() << " address(es) at interval "
-            << mDescription.interval.count() << "ms" << endm;
         }
 
         ~Service()
         {
           getInfoLogger() << "Stopping publisher '" << mDescription.dnsName << "'" << endm;
         }
 
-        void updateRegisterValues(RegisterReadWriteInterface& channel)
+        void updateValues(BarInterface& channel)
         {
           getInfoLogger() << "Updating '" << mDescription.dnsName << "':" << endm;
-          for (size_t i = 0; i < mDescription.addresses.size(); ++i) {
-            auto index = mDescription.addresses[i] / 4;
-            auto value = channel.readRegister(index);
-            getInfoLogger() << "  " << mDescription.addresses[i] << " = " << value << endm;
-            registerValues.at(i) = channel.readRegister(index);
-          }
+
+          Visitor::apply(mDescription.type,
+            [&](const ServiceDescription::Register& type){
+              for (size_t i = 0; i < type.addresses.size(); ++i) {
+                auto index = type.addresses[i] / 4;
+                auto value = channel.readRegister(index);
+                getInfoLogger() << "  " << type.addresses[i] << " = " << value << endm;
+                registerValues.at(i) = channel.readRegister(index);
+              }
+            },
+            [&](const ServiceDescription::ScaSequence& type){
+              std::fill(registerValues.begin(), registerValues.end(), 0); // Reset array in case of aborts
+              auto sca = Sca(channel, channel.getCardType());
+              for (size_t i = 0; i < type.commandDataPairs.size(); ++i) {
+                try {
+                  const auto& pair = type.commandDataPairs[i];
+                  sca.write(pair);
+                  auto result = sca.read();
+                  registerValues[i*2] = result.command;
+                  registerValues[i*2 + 1] = result.data;
+                } catch (const ScaException &e) {
+                  // If an SCA error occurs, we stop executing the sequence of commands and set the error value
+                  registerValues[i*2] = 0xffffffff;
+                  registerValues[i*2 + 1] = 0xffffffff;
+                  break;
+                }
+              }
+            }
+          );
+
           dimService->updateService();
         }
 
@@ -243,6 +301,8 @@ class ProgramAliceLowlevelFrontendServer: public AliceO2::Common::Program
         [&](auto parameter){return registerWrite(parameter, bar0);});
       auto serverPublishStart = makeServer(names.publishStartCommandRpc(),
         [&](auto parameter){return publishStartCommand(parameter, mCommandQueue);});
+      auto serverPublishScaStart = makeServer(names.publishScaStartCommandRpc(),
+        [&](auto parameter){return publishScaStartCommand(parameter, mCommandQueue);});
       auto serverPublishStop = makeServer(names.publishStopCommandRpc(),
         [&](auto parameter){return publishStopCommand(parameter, mCommandQueue);});
       auto serverScaRead = makeServer(names.scaRead(),
@@ -278,6 +338,8 @@ class ProgramAliceLowlevelFrontendServer: public AliceO2::Common::Program
         }
 
         publisherRegistry.loop();
+
+        // Dummy service. Temporary.
         mTemperature = double((std::rand() % 100) + 400) / 10.0;
         temperatureService.updateService(mTemperature);
       }
@@ -292,6 +354,16 @@ class ProgramAliceLowlevelFrontendServer: public AliceO2::Common::Program
       }
     }
 
+    /// Try to add a command to the queue
+    static bool tryAddToQueue(const CommandVariant& command, CommandQueue& queue)
+    {
+      if (!queue.write(command)) {
+        getInfoLogger() << "  command queue was full!" << endm;
+        return false;
+      }
+      return true;
+    }
+
     /// RPC handler for register reads
     static std::string registerRead(const std::string& parameter, ChannelSharedPtr channel)
     {
@@ -332,39 +404,61 @@ class ProgramAliceLowlevelFrontendServer: public AliceO2::Common::Program
     }
 
     /// RPC handler for publish commands
-    static std::string publishStartCommand(const std::string& parameter, CommandQueue& commandQueue)
+    static std::string publishStartCommand(const std::string& parameter, CommandQueue& queue)
     {
       getInfoLogger() << "Received publish command: '" << parameter << "'" << endm;
       auto params = split(parameter, ";");
 
       ServiceDescription description;
-      description.addresses = lexicalCastVector<uintptr_t >(split(params.at(1), ","));
+      description.type = ServiceDescription::Register{lexicalCastVector<uintptr_t >(split(params.at(1), ","))};
       description.dnsName = params.at(0);
       description.interval = std::chrono::milliseconds(int64_t(b::lexical_cast<double>(params.at(2)) * 1000.0));
 
-      if (!commandQueue.write(CommandPublishStart{std::move(description)})) {
-        getInfoLogger() << "  command queue was full!" << endm;
+      tryAddToQueue(CommandPublishStart{description}, queue);
+      return "";
+    }
+
+    /// RPC handler for SCA publish commands
+    static std::string publishScaStartCommand(const std::string& parameter, CommandQueue& queue)
+    {
+      getInfoLogger() << "Received SCA publish command: '" << parameter << "'" << endm;
+
+      auto params = split(parameter, ";");
+
+      // Convert command-data pair string sequence to binary format
+      ServiceDescription::ScaSequence sca;
+      auto commandDataPairStrings = split(params.at(1), "\n");
+      sca.commandDataPairs.resize(commandDataPairStrings.size());
+      for (size_t i = 0; i < commandDataPairStrings.size(); ++i) {
+        sca.commandDataPairs[i] = stringToScaCommandDataPair(commandDataPairStrings[i]);
       }
+
+      ServiceDescription description;
+      description.type = sca;
+      description.dnsName = params.at(0);
+      description.interval = std::chrono::milliseconds(int64_t(b::lexical_cast<double>(params.at(2)) * 1000.0));
+
+      tryAddToQueue(CommandPublishStart{description}, queue);
       return "";
     }
 
     /// RPC handler for publish stop commands
-    static std::string publishStopCommand(const std::string& parameter, CommandQueue& commandQueue)
+    static std::string publishStopCommand(const std::string& parameter, CommandQueue& queue)
     {
       getInfoLogger() << "Received stop command: '" << parameter << "'" << endm;
-      if (!commandQueue.write(CommandPublishStop{parameter})) {
-        getInfoLogger() << "  command queue was full!" << endm;
-      }
+      tryAddToQueue(CommandPublishStop{parameter}, queue);
       return "";
     }
 
+    /// RPC handler for SCA read commands
     static std::string scaRead(const std::string&, ChannelSharedPtr bar2)
     {
       getInfoLogger() << "SCA_READ" << endm;
       auto result = Sca(*bar2, bar2->getCardType()).read();
       return (b::format("0x%x,0x%x") % result.command % result.data).str();
     }
 
+    /// RPC handler for SCA write commands
     static std::string scaWrite(const std::string& parameter, ChannelSharedPtr bar2)
     {
       getInfoLogger() << "SCA_WRITE: '" << parameter << "'" << endm;
@@ -375,13 +469,15 @@ class ProgramAliceLowlevelFrontendServer: public AliceO2::Common::Program
       return "";
     }
 
+    /// RPC handler for SCA GPIO read commands
     static std::string scaGpioRead(const std::string&, ChannelSharedPtr bar2)
     {
       getInfoLogger() << "SCA_GPIO_READ" << endm;
       auto result = Sca(*bar2, bar2->getCardType()).gpioRead();
       return (b::format("0x%x") % result.data).str();
     }
 
+    /// RPC handler for SCA GPIO write commands
     static std::string scaGpioWrite(const std::string& parameter, ChannelSharedPtr bar2)
     {
       getInfoLogger() << "SCA_GPIO_WRITE: '" << parameter << "'" << endm;
@@ -390,6 +486,7 @@ class ProgramAliceLowlevelFrontendServer: public AliceO2::Common::Program
       return "";
     }
 
+    /// RPC handler for SCA blob write commands (sequence of commands)
     static std::string scaBlobWrite(const std::string& parameter, ChannelSharedPtr bar2)
     {
       getInfoLogger() << "SCA_BLOB_WRITE size=" << parameter.size() << " bytes" << endm;
@@ -401,26 +498,18 @@ class ProgramAliceLowlevelFrontendServer: public AliceO2::Common::Program
       std::stringstream resultBuffer;
       auto sca = Sca(*bar2, bar2->getCardType());
 
-      for (std::string token : tokenizer(parameter, sep)) {
+      for (const std::string& token : tokenizer(parameter, sep)) {
         // Walk through the tokens, these should be the pairs (or comments).
         if (token.find('#') == 0) {
           // We have a comment, skip this token
           continue;
         } else {
-          // The pairs are comma-separated, so we split them.
-          std::vector<std::string> pair = split(token, ",");
-          if (pair.size() != 2) {
-            BOOST_THROW_EXCEPTION(
-              AlfException() << ErrorInfo::Message("SCA command-data pair not formatted correctly"));
-          }
-          auto command = convertHexString(pair[0]);
-          auto data = convertHexString(pair[1]);
-
+          auto commandData = stringToScaCommandDataPair(token);
           try {
-            sca.write(command, data);
+            sca.write(commandData);
             auto result = sca.read();
-            getInfoLogger() << (b::format("cmd=0x%x data=0x%x result=0x%x") % command % data % result.data).str()
-              << endm;
+            getInfoLogger() << (b::format("cmd=0x%x data=0x%x result=0x%x") % commandData.command % commandData.data %
+              result.data).str() << endm;
             resultBuffer << std::hex << result.data << '\n';
           } catch (const ScaException &e) {
             // If an SCA error occurs, we stop executing the sequence of commands and return the results as far as we got
@@ -434,6 +523,20 @@ class ProgramAliceLowlevelFrontendServer: public AliceO2::Common::Program
       return resultBuffer.str();
     }
 
+    static Sca::CommandData stringToScaCommandDataPair(const std::string& string)
+    {
+      // The pairs are comma-separated, so we split them.
+      std::vector<std::string> pair = split(string, ",");
+      if (pair.size() != 2) {
+        BOOST_THROW_EXCEPTION(
+          AlfException() << ErrorInfo::Message("SCA command-data pair not formatted correctly"));
+      }
+      Sca::CommandData commandData;
+      commandData.command = convertHexString(pair[0]);
+      commandData.data = convertHexString(pair[1]);
+      return commandData;
+    };
+
     int mSerialNumber = 0;
     CommandQueue mCommandQueue;
     double mTemperature = 40;