ctpdev: ctpproxy publish CTP scalers

DataFormats/Detectors/CTP/include/DataFormatsCTP/Configuration.h

@@ -100,6 +100,8 @@ class CTPConfiguration
   CTPInput* isInputInConfig(const std::string inpname);
   uint64_t getDecrtiptorInputsMask(const std::string& name) const;
   std::map<o2::detectors::DetID::ID, std::vector<CTPInput>> getDet2InputMap();
+  uint64_t getTriggerClassMask() const;
+  std::vector<int> getTriggerClassList() const;
 
  private:
   std::string mName;

DataFormats/Detectors/CTP/include/DataFormatsCTP/Digits.h

@@ -33,7 +33,7 @@ static constexpr uint32_t NGBT = 80;
 static constexpr std::uint32_t NumOfHBInTF = 256;
 typedef std::bitset<NGBT> gbtword80_t;
 //
-static constexpr std::uint32_t CTP_NINPUTS = 46;    /// Max number of CTP inputs for all levels
+static constexpr std::uint32_t CTP_NINPUTS = 48;    /// Max number of CTP inputs for all levels
 static constexpr std::uint32_t CTP_NCLASSES = 64;   /// Number of classes in hardware
 static constexpr std::uint32_t CTP_MAXTRIGINPPERDET = 5; /// Max number of LM/L0inputs per detector
 /// Positions of CTP Detector inputs in CTPInputMask: first=offset, second=mask
@@ -57,7 +57,7 @@ struct CTPDigit {
     return intRecord == d.intRecord && CTPInputMask == d.CTPInputMask && CTPClassMask == d.CTPClassMask;
   }
 
-  ClassDefNV(CTPDigit, 2);
+  ClassDefNV(CTPDigit, 3);
 };
 
 std::ostream& operator<<(std::ostream& os, const CTPDigit& d);

DataFormats/Detectors/CTP/include/DataFormatsCTP/Scalers.h

@@ -82,6 +82,8 @@ class CTPRunScalers
   int convertRawToO2();
   int checkConsistency(const CTPScalerO2& scal0, const CTPScalerO2& scal1) const;
   int checkConsistency(const CTPScalerRecordO2& rec0, const CTPScalerRecordO2& rec1) const;
+  //
+  int parseZMQScalers(std::string zmqscalers);
 
  private:
   // map from class index to overflow

DataFormats/Detectors/CTP/src/Configuration.cxx

@@ -326,3 +326,22 @@ std::map<o2::detectors::DetID::ID, std::vector<CTPInput>> CTPConfiguration::getD
   }
   return det2inp;
 }
+uint64_t CTPConfiguration::getTriggerClassMask() const
+{
+  uint64_t clsmask = 0;
+  for (auto const& cls : mCTPClasses) {
+    clsmask |= cls.classMask;
+  }
+  return clsmask;
+}
+std::vector<int> CTPConfiguration::getTriggerClassList() const
+{
+  uint64_t clsmask = getTriggerClassMask();
+  std::vector<int> classlist;
+  for (int i = 0; i < 64; i++) {
+    if ((1ull << i) & clsmask) {
+      classlist.push_back(i);
+    }
+  }
+  return classlist;
+}

DataFormats/Detectors/CTP/src/Scalers.cxx

@@ -361,3 +361,8 @@ int CTPRunScalers::updateOverflows(const CTPScalerRaw& scal0, const CTPScalerRaw
   //std::cout << std::endl;
   return 0;
 }
+int CTPRunScalers::parseZMQScalers(std::string zmqscalers)
+{
+  std::vector<std::string> tokens = o2::utils::Str::tokenize(zmqscalers, ' ');
+  return 0;
+}

Detectors/CTP/CMakeLists.txt

@@ -13,4 +13,5 @@ add_subdirectory(simulation)
 add_subdirectory(reconstruction)
 add_subdirectory(workflow)
 add_subdirectory(workflowIO)
+add_subdirectory(workflowScalers)
 add_subdirectory(macro)

Detectors/CTP/reconstruction/CMakeLists.txt

@@ -18,3 +18,4 @@ o2_add_library(CTPReconstruction
                                      O2::rANS
                                      Microsoft.GSL::GSL)
 
+

Detectors/CTP/workflowScalers/CMakeLists.txt

@@ -0,0 +1,6 @@
+o2_add_executable(
+                  proxy
+                  COMPONENT_NAME ctp
+                  SOURCES src/ctp-proxy.cxx
+                  PUBLIC_LINK_LIBRARIES O2::DCStestWorkflow
+                  O2::DataFormatsCTP)

Detectors/CTP/workflowScalers/py/pub_server.py

@@ -0,0 +1,24 @@
+import zmq
+import random
+import sys
+import time
+
+port = "500901"
+if len(sys.argv) > 1:
+    port =  sys.argv[1]
+    int(port)
+
+context = zmq.Context()
+socket = context.socket(zmq.PUB)
+socket.bind("tcp://*:%s" % port)
+while True:
+    topic = random.randrange(0,2**32)
+    #messagedata = random.randrange(1,215) - 80
+    header = "CTP"
+    messagedata = str(topic)+"1 2 3 4 5"
+    print("Sending:",header, messagedata)
+    data = str(messagedata).encode()
+    header = str(header).encode()
+    socket.send(memoryview(header),zmq.SNDMORE)
+    socket.send(memoryview(data))
+    time.sleep(1)

Detectors/CTP/workflowScalers/py/sub_client.py

@@ -0,0 +1,37 @@
+import sys
+import zmq
+
+port = "500901"
+if len(sys.argv) > 1:
+    port =  sys.argv[1]
+    int(port)
+
+if len(sys.argv) > 2:
+    port1 =  sys.argv[2]
+    int(port1)
+
+# Socket to talk to server
+context = zmq.Context()
+socket = context.socket(zmq.SUB)
+
+print("Collecting updates from CTP server, port:",port)
+socket.connect ("tcp://localhost:%s" % port)
+
+if len(sys.argv) > 2:
+    socket.connect ("tcp://localhost:%s" % port1)
+# Subscribe to zipcode, default is NYC, 10001
+topicfilter = "CTP"
+socket.setsockopt_string(zmq.SUBSCRIBE, topicfilter)
+
+# Process 5 updates
+total_value = 0
+while(1):
+    string = socket.recv_multipart()
+    #string = stringr.decode()
+    print("string:",string)
+    #topic, messagedata = string.split()
+    #total_value += int(messagedata)
+    #print(topic, messagedata)
+
+print("Average messagedata value for topic '%s' was %dF" % (topicfilter, total_value / update_nbr))
+

Detectors/CTP/workflowScalers/src/ctp-proxy.cxx

@@ -0,0 +1,133 @@
+// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
+// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
+// All rights not expressly granted are reserved.
+//
+// This software is distributed under the terms of the GNU General Public
+// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
+// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
+// All rights not expressly granted are reserved.
+//
+// This software is distributed under the terms of the GNU General Public
+// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+// example to run:
+// o2-ctp-proxy --ctp-proxy '--channel-config "name=ctp-proxy,type=sub,method=connect,address=tcp://127.0.0.1:5556,rateLogging=1,transport=zeromq"'
+
+#include "Framework/WorkflowSpec.h"
+#include "Framework/DataProcessorSpec.h"
+#include "Framework/DataSpecUtils.h"
+#include "Framework/ControlService.h"
+#include "Framework/Logger.h"
+#include "Framework/Lifetime.h"
+#include "Framework/ConfigParamSpec.h"
+#include "Framework/ExternalFairMQDeviceProxy.h"
+#include "DetectorsCommonDataFormats/DetID.h"
+#include "Headers/DataHeaderHelpers.h"
+#include <fairmq/FairMQDevice.h>
+#include "CommonUtils/StringUtils.h"
+#include <vector>
+#include <string>
+
+using namespace o2::framework;
+using DetID = o2::detectors::DetID;
+InjectorFunction dcs2dpl()
+{
+
+  auto timesliceId = std::make_shared<size_t>(0);
+
+  return [timesliceId](FairMQDevice& device, FairMQParts& parts, ChannelRetriever channelRetriever) {
+    // make sure just 2 messages received
+    if (parts.Size() != 2) {
+      LOG(error) << "received " << parts.Size() << " instead of 2 expected";
+      return;
+    }
+    std::string messageHeader{static_cast<const char*>(parts.At(0)->GetData()), parts.At(0)->GetSize()};
+    size_t dataSize = parts.At(1)->GetSize();
+    std::string messageData{static_cast<const char*>(parts.At(1)->GetData()), parts.At(1)->GetSize()};
+    LOG(info) << "received message " << messageHeader << " of size " << dataSize << " Payload:" << messageData;
+    o2::header::DataHeader hdrF("CTP_COUNTERS", o2::header::gDataOriginCTP, 0);
+    OutputSpec outsp{hdrF.dataOrigin, hdrF.dataDescription, hdrF.subSpecification};
+    auto channel = channelRetriever(outsp, *timesliceId);
+    if (channel.empty()) {
+      LOG(error) << "No output channel found for OutputSpec " << outsp;
+      return;
+    }
+
+    hdrF.tfCounter = *timesliceId; // this also
+    hdrF.payloadSerializationMethod = o2::header::gSerializationMethodNone;
+    hdrF.splitPayloadParts = 1;
+    hdrF.splitPayloadIndex = 0;
+    hdrF.payloadSize = parts.At(1)->GetSize();
+    hdrF.firstTForbit = 0; // this should be irrelevant for Counters ? Orbit is in payload
+
+    auto fmqFactory = device.GetChannel(channel).Transport();
+
+    o2::header::Stack headerStackF{hdrF, DataProcessingHeader{*timesliceId, 1}};
+    auto hdMessageF = fmqFactory->CreateMessage(headerStackF.size(), fair::mq::Alignment{64});
+    auto plMessageF = fmqFactory->CreateMessage(hdrF.payloadSize, fair::mq::Alignment{64});
+    memcpy(hdMessageF->GetData(), headerStackF.data(), headerStackF.size());
+    memcpy(plMessageF->GetData(), parts.At(1)->GetData(), hdrF.payloadSize);
+
+    FairMQParts outParts;
+    outParts.AddPart(std::move(hdMessageF));
+    outParts.AddPart(std::move(plMessageF));
+    sendOnChannel(device, outParts, channel);
+    // LOG(info) << "Sent CTP counters DPL message";
+  };
+}
+
+// we need to add workflow options before including Framework/runDataProcessing
+void customize(std::vector<ConfigParamSpec>& workflowOptions)
+{
+  workflowOptions.push_back(ConfigParamSpec{"subscribe-to", VariantType::String, "type=sub,method=connect,address=tcp://188.184.30.57:5556,rateLogging=10,transport=zeromq", {"channel subscribe to"}});
+}
+
+#include "Framework/runDataProcessing.h"
+
+WorkflowSpec defineDataProcessing(ConfigContext const& config)
+{
+  LOG(info) << "Defining data processing";
+  auto setChanName = [](const std::string& chan, const std::string& name) {
+    size_t n = 0;
+    if (std::string(chan).find("name=") != std::string::npos) {
+      n = std::string(chan).find(",");
+      if (n == std::string::npos) {
+        throw std::runtime_error(fmt::format("wrongly formatted channel: {}", chan));
+      }
+      n++;
+    }
+    LOG(info) << "===>inside:" << name << " " << chan;
+    return o2::utils::Str::concat_string("name=", name, ",", chan.substr(n, chan.size()));
+  };
+  const std::string devName = "ctp-proxy";
+  auto chan = config.options().get<std::string>("subscribe-to");
+  if (chan.empty()) {
+    throw std::runtime_error("input channel is not provided");
+  }
+  chan = setChanName(chan, devName);
+  LOG(info) << "name:" << devName << " chan:" << chan;
+  LOG(info) << "Channels setup: " << chan;
+  Outputs ctpCountersOutputs;
+  ctpCountersOutputs.emplace_back("CTP", "CTP_COUNTERS", 0, Lifetime::Timeframe);
+  LOG(info) << "===> Proxy to be set";
+  DataProcessorSpec ctpProxy = specifyExternalFairMQDeviceProxy(
+    devName.c_str(),
+    std::move(ctpCountersOutputs),
+    // this is just default, can be overriden by --ctp-config-proxy '--channel-config..'
+    chan.c_str(),
+    dcs2dpl());
+  LOG(info) << "===> Proxy done";
+  WorkflowSpec workflow;
+  workflow.emplace_back(ctpProxy);
+  return workflow;
+}