Changes from the July DDS workshop

CMakeLists.txt

@@ -34,15 +34,30 @@ project(ALICEO2)
 Option(ALICEO2_MODULAR_BUILD "Modular build without environment variables" OFF)
 if(NOT ALICEO2_MODULAR_BUILD)
   IF(NOT DEFINED ENV{FAIRROOTPATH})
-    MESSAGE(FATAL_ERROR "You did not define the environment variable FAIRROOTPATH which is needed to find FairRoot. Please set this variable and execute cmake again.")
+    FIND_PROGRAM(FAIRROOT_CONFIG_EXECUTABLE NAMES fairroot-config 
+      )
+
+    IF(FAIRROOT_CONFIG_EXECUTABLE)
+      EXECUTE_PROCESS(COMMAND ${FAIRROOT_CONFIG_EXECUTABLE} --fairsoft_path
+                  OUTPUT_VARIABLE FAIRROOTPATH
+                 )
+    ENDIF(FAIRROOT_CONFIG_EXECUTABLE)
+    STRING(STRIP ${FAIRROOTPATH} FAIRROOTPATH)
+
+    IF(DEFINED FAIRROOTPATH)
+      SET(ENV{FAIRROOTPATH} ${FAIRROOTPATH})
+    ELSE(DEFINED FAIRROOTPATH)
+      MESSAGE(FATAL_ERROR "Can not find FairRoot. Either set variable FAIRROOTPATH or make sure the installation path is in variable PATH, and execute cmake again.")
+    ENDIF(DEFINED FAIRROOTPATH)
+  ELSE(NOT DEFINED ENV{FAIRROOTPATH})
+    SET(FAIRROOTPATH $ENV{FAIRROOTPATH})
   ENDIF(NOT DEFINED ENV{FAIRROOTPATH})
 
   IF(NOT DEFINED ENV{SIMPATH})
      MESSAGE(FATAL_ERROR "You did not define the environment variable SIMPATH which is nedded to find the external packages. Please set this variable and execute cmake again.")
   ENDIF(NOT DEFINED ENV{SIMPATH})
 
   SET(SIMPATH $ENV{SIMPATH})
-  SET(FAIRROOTPATH $ENV{FAIRROOTPATH})
 
   # where to look first for cmake modules, before ${CMAKE_ROOT}/Modules/
   # is checked
@@ -52,6 +67,7 @@ if(NOT ALICEO2_MODULAR_BUILD)
 
   Set(CheckSrcDir "${FAIRROOTPATH}/share/fairbase/cmake/checks")
 else(NOT ALICEO2_MODULAR_BUILD)
+  message(FATAL_ERROR "build option 'ALICEO2_MODULAR_BUILD' not supported yet")
   find_package(Boost REQUIRED)
 endif(NOT ALICEO2_MODULAR_BUILD)
 

Generators/CMakeLists.txt

@@ -2,6 +2,7 @@
 # the array .
 # The extension is already found.  Any number of sources could be listed here.
 
+if(PYTHIA8_INCLUDE_DIR)
 set(INCLUDE_DIRECTORIES
 ${CMAKE_SOURCE_DIR}/Generators
 )
@@ -38,3 +39,6 @@ set(LIBRARY_NAME O2Gen)
 set(DEPENDENCIES Base O2Data pythia8 Pythia6)
 
 GENERATE_LIBRARY()
+else(PYTHIA8_INCLUDE_DIR)
+  message(STATUS "module 'Generators' requires Pythia8 ... deactivated")
+endif(PYTHIA8_INCLUDE_DIR)

devices/aliceHLTwrapper/WrapperDevice.cxx

@@ -67,6 +67,10 @@ WrapperDevice::~WrapperDevice()
 }
 
 void WrapperDevice::Init()
+{
+}
+
+void WrapperDevice::InitTask()
 {
   /// inherited from FairMQDevice
 
@@ -310,14 +314,6 @@ void WrapperDevice::Pause()
   FairMQDevice::Pause();
 }
 
-void WrapperDevice::Shutdown()
-{
-  /// inherited from FairMQDevice
-  int iResult=0;
-  // TODO: shutdown component and delete instance
-  FairMQDevice::Shutdown();
-}
-
 void WrapperDevice::SetProperty(const int key, const string& value)
 {
   /// inherited from FairMQDevice

devices/aliceHLTwrapper/WrapperDevice.h

@@ -48,12 +48,12 @@ class WrapperDevice : public FairMQDevice {
   /// inherited from FairMQDevice
   virtual void Init();
   /// inherited from FairMQDevice
+  virtual void InitTask();
+  /// inherited from FairMQDevice
   virtual void Run();
   /// inherited from FairMQDevice
   virtual void Pause();
   /// inherited from FairMQDevice
-  virtual void Shutdown();
-  /// inherited from FairMQDevice
   /// handle device specific properties and forward to FairMQDevice::SetProperty
   virtual void SetProperty(const int key, const std::string& value);
   /// inherited from FairMQDevice

devices/aliceHLTwrapper/aliceHLTWrapper.cxx

@@ -85,6 +85,21 @@ struct SocketProperties_t {
   {}
 };
 
+ostream& operator<<(ostream &out, const SocketProperties_t& me)
+{
+  out << "Socket configuration:"
+      << " type       = " << me.type
+      << " size       = " << me.size
+      << " method     = " << me.method
+      << " address    = " << me.address
+      << " ddsprop    = " << me.ddsprop
+      << " ddscount   = " << me.ddscount
+      << " ddsminport = " << me.ddsminport
+      << " ddsmaxport = " << me.ddsmaxport
+    ;
+  return out;
+}
+
 FairMQDevice* gDevice = NULL;
 static void s_signal_handler(int signal)
 {
@@ -382,7 +397,16 @@ int main(int argc, char** argv)
     if (pollingPeriod > 0) device.SetProperty(ALICE::HLT::WrapperDevice::PollingPeriod, pollingPeriod);
     if (skipProcessing) device.SetProperty(ALICE::HLT::WrapperDevice::SkipProcessing, skipProcessing);
     for (unsigned iInput = 0; iInput < numInputs; iInput++) {
-      FairMQChannel inputChannel(inputSockets[iInput].type.c_str(), inputSockets[iInput].method.c_str(), inputSockets[iInput].address.c_str());
+      std::cout << "input socket " << iInput << " " << inputSockets[iInput] << endl;
+      // if running in DDS mode, the address contains now the IP address of the host
+      // machine in order to check if a DDS property comes from the same machine.
+      // This will be obsolete as soon as DDS propagates properties only within
+      // collections. Then the address can be empty for connecting sockets and
+      // can be directly used in the creation of the channel. The next two lines are
+      // then obsolete
+      std::string address=inputSockets[iInput].address.c_str();
+      if (inputSockets[iInput].method.compare("connect")==0 && bUseDDS) address="";
+      FairMQChannel inputChannel(inputSockets[iInput].type.c_str(), inputSockets[iInput].method.c_str(), address);
       // set High-water-mark for the sockets. in ZMQ, depending on the socket type, some
       // have only send buffers (PUB, PUSH), some only receive buffers (SUB, PULL), and
       // some have both (DEALER, ROUTER, PAIR, REQ, REP)
@@ -392,7 +416,12 @@ int main(int argc, char** argv)
       device.fChannels["data-in"].push_back(inputChannel);
     }
     for (unsigned iOutput = 0; iOutput < numOutputs; iOutput++) {
-      FairMQChannel outputChannel(outputSockets[iOutput].type.c_str(), outputSockets[iOutput].method.c_str(), outputSockets[iOutput].address.c_str());
+      std::cout << "output socket " << iOutput << " " << outputSockets[iOutput] << endl;
+      // see comment above, next two lines obsolete if DDS implements property
+      // propagation within collections
+      std::string address=outputSockets[iOutput].address.c_str();
+      if (outputSockets[iOutput].method.compare("connect")==0 && bUseDDS) address="";
+      FairMQChannel outputChannel(outputSockets[iOutput].type.c_str(), outputSockets[iOutput].method.c_str(), address);
       // we set both snd and rcv to the same value for the moment, see above
       outputChannel.UpdateSndBufSize(outputSockets[iOutput].size);
       outputChannel.UpdateRcvBufSize(outputSockets[iOutput].size);
@@ -438,6 +467,7 @@ int main(int argc, char** argv)
         device.fChannels["data-out"].at(iOutput).UpdateAddress(outputSockets[iOutput].address.c_str());
       }
     }
+    cout << "finnished fetching properties from DDS, now waiting for end of state INIT_DEVICE ..." << endl;
     device.WaitForEndOfState("INIT_DEVICE");
 #endif
     device.ChangeState("INIT_TASK");

topologies/o2prototype_topology.xml

@@ -30,39 +30,36 @@ The following parameters need adjustment when extending the FLP-EPN configuratio
     <property id="CollectorInputAddress" />
 
     <declrequirement id="collectorhosts">
-        <hostPattern value="cn(48)\.internal"/>
+        <hostPattern value="cn(59)\.internal"/>
     </declrequirement>
 
     <declrequirement id="flphosts">
-        <!-- <hostPattern value="cn(00|01|02|03|04|05|06|07|08|09|10|11|12|13|14|15|16|17)\.internal"/> -->
-        <hostPattern value="cn(00)\.internal"/>
+        <!-- <hostPattern value="cn(00|01|03|04|05|07|09|10|12|13|14|15|17|18|19|20|24|25|27|28|29|30|32|35|48|49|50|51|52|53|54|55|56|57|58|59)\.internal"/> -->
+        <hostPattern value="cn(01|03)\.internal"/>
     </declrequirement>
 
     <declrequirement id="epnhosts">
         <!-- <hostPattern value="cn(18|19|20|21|22|23|24|25|27|28|30|31|32|33|34|35)\.internal"/> -->
-        <hostPattern value="cn(35)\.internal"/>
+        <hostPattern value="cn(10|13|14|15)\.internal"/>
     </declrequirement>
 
     <decltask id="dataPublisher">
-        <exe reachable="true">$ALICEO2_INSTALL_DIR/bin/aliceHLTWrapper ClusterPublisher 1 --dds --poll-period 100 --output type=push,size=5000,method=bind,address=dummy,property=DataPublisherOutputAddress,min-port=48000 --library libAliHLTUtil.so --component FilePublisher --run 167808 --parameter '-datafilelist /tmp/tpc-cluster-publisher.conf'</exe>
-        <requirement>flphosts</requirement>
+        <exe reachable="true">$ALICEO2_INSTALL_DIR/bin/aliceHLTWrapper ClusterPublisher_%collectionIndex%_%taskIndex% 1 --dds --poll-period 100 --output type=push,size=5000,method=bind,address=dummy,property=DataPublisherOutputAddress,min-port=48000 --library libAliHLTUtil.so --component FilePublisher --run 167808 --parameter '-datafilelist /home/richterm/workdir/dds-rundir/data-configuration/tpc-cluster-publisher_slice%collectionIndex%.conf'</exe>
         <properties>
             <id access="write">DataPublisherOutputAddress</id>
         </properties>
     </decltask>
 
     <decltask id="relay">
-        <exe reachable="true">$ALICEO2_INSTALL_DIR/bin/aliceHLTWrapper Relay 1 --dds --poll-period 100 --input type=pull,size=5000,method=connect,property=DataPublisherOutputAddress,count=1,global --output type=push,size=5000,method=connect,property=FLPSenderInputAddress,count=1 --library libAliHLTUtil.so --component BlockFilter --run 167808  --parameter '-loglevel=0x79'</exe>
-        <requirement>flphosts</requirement>
+        <exe reachable="true">$ALICEO2_INSTALL_DIR/bin/aliceHLTWrapper Relay_%collectionIndex%_%taskIndex% 1 --dds --poll-period 100 --input type=pull,size=5000,method=connect,property=DataPublisherOutputAddress,count=1,global --output type=push,size=5000,method=connect,property=FLPSenderInputAddress,count=1 --library libAliHLTUtil.so --component BlockFilter --run 167808  --parameter '-loglevel=0x79' -s 10000000</exe>
         <properties>
             <id access="read">DataPublisherOutputAddress</id>
             <id access="read">FLPSenderInputAddress</id>
         </properties>
     </decltask>
 
     <decltask id="flpSender">
-        <exe reachable="true">$ALICEO2_INSTALL_DIR/bin/flpSender_dds --id 0 --event-size 593750 --num-inputs 3 --num-outputs 1 --input-socket-type sub --input-buff-size 10 --input-method bind --input-rate-logging 0 --input-socket-type sub --input-buff-size 10 --input-method bind --input-rate-logging 0 --input-socket-type sub --input-buff-size 10 --input-method bind --input-rate-logging 0 --output-socket-type push --output-buff-size 10 --output-method connect --output-rate-logging 1</exe>
-        <requirement>flphosts</requirement>
+        <exe reachable="true">$ALICEO2_INSTALL_DIR/bin/flpSender_dds --id flpSender_%collectionIndex%_%taskIndex% --event-size 593750 --num-inputs 3 --num-outputs 4 --input-socket-type sub --input-buff-size 10 --input-method bind --input-rate-logging 0 --input-socket-type sub --input-buff-size 10 --input-method bind --input-rate-logging 0 --input-socket-type sub --input-buff-size 10 --input-method bind --input-rate-logging 0 --output-socket-type push --output-buff-size 10 --output-method connect --output-rate-logging 1</exe>
         <properties>
             <id access="write">FLPSenderInputAddress</id>
             <id access="write">FLPSenderHeartbeatInputAddress</id>
@@ -71,8 +68,7 @@ The following parameters need adjustment when extending the FLP-EPN configuratio
     </decltask>
 
     <decltask id="epnReceiver">
-        <exe reachable="true">$ALICEO2_INSTALL_DIR/bin/epnReceiver_dds --id EPN0 --num-outputs 2 --num-flps 1 --input-socket-type pull --input-buff-size 10 --input-method bind --input-rate-logging 1 --output-socket-type pub --output-buff-size 10 --output-method connect --output-rate-logging 0 --nextstep-socket-type push --nextstep-buff-size 10 --nextstep-method bind --nextstep-rate-logging 1</exe>
-        <requirement>epnhosts</requirement>
+        <exe reachable="true">$ALICEO2_INSTALL_DIR/bin/epnReceiver_dds --id EPN_%collectionIndex%_%taskIndex% --num-outputs 3 --num-flps 2 --input-socket-type pull --input-buff-size 10 --input-method bind --input-rate-logging 1 --output-socket-type pub --output-buff-size 10 --output-method connect --output-rate-logging 0 --nextstep-socket-type push --nextstep-buff-size 10 --nextstep-method bind --nextstep-rate-logging 1</exe>
         <properties>
             <id access="read">FLPSenderHeartbeatInputAddress</id>
             <id access="write">EPNReceiverInputAddress</id>
@@ -81,8 +77,7 @@ The following parameters need adjustment when extending the FLP-EPN configuratio
     </decltask>
 
     <decltask id="tracker">
-        <exe reachable="true">$ALICEO2_INSTALL_DIR/bin/aliceHLTWrapper Tracker 1 --dds --poll-period 100 --input type=pull,size=5000,method=connect,property=EPNReceiverOutputAddress,count=1 --output type=push,size=500,method=bind,property=TrackingOutputAddress,min-port=48000 --library libAliHLTTPC.so --component TPCCATracker --run 167808 --parameter '-GlobalTracking -allowGPU -GPUHelperThreads 4 -loglevel=0x7c'</exe>
-        <requirement>epnhosts</requirement>
+        <exe reachable="true">$ALICEO2_INSTALL_DIR/bin/aliceHLTWrapper Tracker_%collectionIndex%_%taskIndex% 1 --dds --poll-period 100 --input type=pull,size=5000,method=connect,property=EPNReceiverOutputAddress,count=1 --output type=push,size=500,method=bind,property=TrackingOutputAddress,min-port=48000 --library libAliHLTTPC.so --component TPCCATracker --run 167808 --parameter '-GlobalTracking -allowGPU -GPUHelperThreads 4 -loglevel=0x7c'</exe>
         <!-- <requirement></requirement> -->
         <properties>
             <id access="read">EPNReceiverOutputAddress</id>
@@ -91,8 +86,7 @@ The following parameters need adjustment when extending the FLP-EPN configuratio
     </decltask>
 
     <decltask id="merger">
-            <exe reachable="true">$ALICEO2_INSTALL_DIR/bin/aliceHLTWrapper GlobalMerger 1 --dds --poll-period 100 --input type=pull,size=5000,method=connect,property=TrackingOutputAddress,count=1 --output type=push,size=500,method=connect,property=CollectorInputAddress,count=1,global  --library libAliHLTTPC.so --component TPCCAGlobalMerger --run 167808 --parameter '-loglevel=0x7c'</exe>
-        <requirement>epnhosts</requirement>
+            <exe reachable="true">$ALICEO2_INSTALL_DIR/bin/aliceHLTWrapper GlobalMerger_%collectionIndex%_%taskIndex% 1 --dds --poll-period 100 --input type=pull,size=5000,method=connect,property=TrackingOutputAddress,count=1 --output type=push,size=500,method=connect,property=CollectorInputAddress,count=1,global  --library libAliHLTTPC.so --component TPCCAGlobalMerger --run 167808 --parameter '-loglevel=0x7c'</exe>
         <!-- <requirement></requirement> -->
         <properties>
             <id access="read">TrackingOutputAddress</id>
@@ -101,26 +95,38 @@ The following parameters need adjustment when extending the FLP-EPN configuratio
     </decltask>
 
     <decltask id="collector">
-            <exe reachable="true">$ALICEO2_INSTALL_DIR/bin/aliceHLTWrapper Collector 1 --dds --poll-period 100 --input type=pull,size=500,method=bind,property=CollectorInputAddress,min-port=48000 --library libAliHLTUtil.so --component FileWriter --run 167808 --parameter '-directory tracker-output -subdir -idfmt=%04d -specfmt=_%08x -blocknofmt= -loglevel=0x7c -write-all-blocks -publisher-conf tracker-output/datablocks.txt'</exe>
+            <exe reachable="true">$ALICEO2_INSTALL_DIR/bin/aliceHLTWrapper Collector 1 --dds --poll-period 100 --input type=pull,size=500,method=bind,property=CollectorInputAddress,min-port=48000 --library libAliHLTUtil.so --component BlockFilter --run 167808 --parameter '-origin SKIP' -s 1000000</exe>
         <requirement>collectorhosts</requirement>
         <properties>
             <id access="write">CollectorInputAddress</id>
     </properties>
     </decltask>
 
-
+    <declcollection id="flpcollection">
+        <requirement>flphosts</requirement>
+        <tasks>
+            <id>dataPublisher</id>
+            <id>relay</id>
+            <id>flpSender</id>
+        </tasks>
+    </declcollection>
+
+    <declcollection id="epncollection">
+        <requirement>epnhosts</requirement>
+        <tasks>
+            <id>epnReceiver</id>
+            <id>tracker</id>
+            <id>merger</id>
+        </tasks>
+    </declcollection>
 
     <main id="main">
         <task>collector</task>
-        <group id="groupFLP" n="1">
-            <task>dataPublisher</task>
-            <task>relay</task>
-            <task>flpSender</task>
+        <group id="groupFLP" n="2">
+            <collection>flpcollection</collection>
         </group>
-        <group id="groupEPN" n="1">
-            <task>epnReceiver</task>
-            <task>tracker</task>
-            <task>merger</task>
+        <group id="groupEPN" n="4">
+            <collection>epncollection</collection>
         </group>
     </main>