fix parallel processing for highly disordered cases (local data replay)

iravasen · iravasen · commit 37d117ae9d47 · 2025-10-21T13:35:26.000+02:00
diff --git a/Detectors/ITSMFT/ITS/workflow/include/ITSWorkflow/ThresholdCalibratorSpec.h b/Detectors/ITSMFT/ITS/workflow/include/ITSWorkflow/ThresholdCalibratorSpec.h
@@ -232,7 +232,9 @@ class ITSThresholdCalibrator : public Task
   short int mRunTypeUp = -1;
   short int mRunTypeRU[N_RU] = {0};
   short int mRunTypeRUCopy[N_RU] = {0};
-  short int mCdwCntRU[N_RU][N_ROW] = {{0}};
+  bool mFlagsRU[N_RU] = {0};
+  //short int mCdwCntRU[N_RU][N_ROW] = {{0}};
+  std::map<short int,std::map<short int, std::array<std::array<int, 500>, 500>>> mCdwCntRU; // RU --> row --> 2D hit map 
   short int mLoopVal[N_RU][N_ROW] = {{0}};
   bool mActiveLinks[N_RU][3] = {{false}};
   std::set<short int> mRuSet;
@@ -241,6 +243,7 @@ class ITSThresholdCalibrator : public Task
   short int mMin = -1, mMax = -1, mMin2 = 0, mMax2 = 0;
   short int mStep = 1, mStep2 = 1;
   short int mStrobeWindow = 5; // 5 means 5*25ns = 125 ns
+  short int mRowScan = 512; // number of scanned rows, used only to normalize % of success 
 
   // Get threshold method (fit == 1, derivative == 0, or hitcounting == 2)
   char mFitType = -1;
@@ -293,6 +296,7 @@ class ITSThresholdCalibrator : public Task
   short int manualStep = 1, manualStep2 = 1;
   std::string manualScanType;
   short int manualStrobeWindow = 5;
+  short int manualRowScan = 512; // used only to normalize % of success in thr/ithr/vcasn scans
 
   // for CRU_ITS data processing
   bool isCRUITS = false;
@@ -306,7 +310,7 @@ class ITSThresholdCalibrator : public Task
   int maxDumpS = -1;                   // maximum number of s-curves to be dumped, default -1 = dump all
   std::string chipDumpS = "";          // list of comma-separated O2 chipIDs to be dumped, default is empty = dump all
   int dumpCounterS[24120] = {0};       // count dumps for every chip
-  int countCdw[24120] = {0};           // count how many CDWs have been processed with the maximum charge injected: usefull for s-curve dump when hits do not arrive in order
+  bool isChipDB[24120] = {0};          // check whether a chip has been already added to DB entry
   TFile* fileDumpS;                    // file where to store the s-curves on disk
   std::vector<short int> chipDumpList; // vector of chips to dump
 
diff --git a/Detectors/ITSMFT/ITS/workflow/src/ThresholdCalibratorSpec.cxx b/Detectors/ITSMFT/ITS/workflow/src/ThresholdCalibratorSpec.cxx
@@ -149,6 +149,12 @@ void ITSThresholdCalibrator::init(InitContext& ic)
   // Parameters to operate in manual mode (when run type is not recognized automatically)
   isManualMode = ic.options().get<bool>("manual-mode");
   if (isManualMode) {
+    try {
+      manualRowScan = ic.options().get<short int>("manual-rowscan");
+    } catch (std::exception const& e) {
+      throw std::runtime_error("Number of scanned rows not found, mandatory in manual mode");
+    }
+    
     try {
       manualMin = ic.options().get<short int>("manual-min");
     } catch (std::exception const& e) {
@@ -746,39 +752,40 @@ void ITSThresholdCalibrator::extractThresholdRow(const short int& chipID, const
 
   } else { // threshold, vcasn, ithr, vresetd_2d
 
-    short int iRU = getRUID(chipID);
+    for (int scan_i = 0; scan_i < ((mScanType == 'r') ? N_RANGE : N_RANGE2); scan_i++) {
 #ifdef WITH_OPENMP
     omp_set_num_threads(mNThreads);
 #pragma omp parallel for schedule(dynamic)
 #endif
-    // Loop over all columns (pixels) in the row
-    for (short int col_i = 0; col_i < this->N_COL; col_i++) {
-      // Do the threshold fit
-      float thresh = 0., noise = 0.;
-      bool success = false;
-      int spoints = 0;
-      int scan_i = mScanType == 'r' ? (mLoopVal[iRU][row] - mMin) / mStep : 0;
-      if (isDumpS) { // already protected for multi-thread in the init
-        mFitHist->SetName(Form("scurve_chip%d_row%d_col%d_scani%d", chipID, row, col_i, scan_i));
-      }
+      // Loop over all columns (pixels) in the row
+      for (short int col_i = 0; col_i < this->N_COL; col_i++) {
+        // Do the threshold fit
+        float thresh = 0., noise = 0.;
+        bool success = false;
+        int spoints = 0;
 
-      success = this->findThreshold(chipID, mPixelHits[chipID][row][col_i],
-                                    this->mX, mScanType == 'r' ? N_RANGE2 : N_RANGE, thresh, noise, spoints, scan_i);
+        if (isDumpS) { // already protected for multi-thread in the init
+          mFitHist->SetName(Form("scurve_chip%d_row%d_col%d_scani%d", chipID, row, col_i, scan_i));
+        }
 
-      vChipid[col_i] = chipID;
-      vRow[col_i] = row;
-      vThreshold[col_i] = (mScanType == 'T' || mScanType == 'r') ? (short int)(thresh * 10.) : (short int)(thresh);
-      vNoise[col_i] = (float)(noise * 10.); // always factor 10 also for ITHR/VCASN to not have all zeros
-      vSuccess[col_i] = success;
-      vPoints[col_i] = spoints > 0 ? (unsigned char)(spoints) : 0;
+        success = this->findThreshold(chipID, mPixelHits[chipID][row][col_i],
+                                      this->mX, mScanType == 'r' ? N_RANGE2 : N_RANGE, thresh, noise, spoints, scan_i);
 
+        vChipid[col_i] = chipID;
+        vRow[col_i] = row;
+        vThreshold[col_i] = (mScanType == 'T' || mScanType == 'r') ? (short int)(thresh * 10.) : (short int)(thresh);
+        vNoise[col_i] = (float)(noise * 10.); // always factor 10 also for ITHR/VCASN to not have all zeros
+        vSuccess[col_i] = success;
+        vPoints[col_i] = spoints > 0 ? (unsigned char)(spoints) : 0;
+
+        if (mScanType == 'r') {
+          vMixData[col_i] = (scan_i * mStep) + mMin;
+        }
+      }
       if (mScanType == 'r') {
-        vMixData[col_i] = mLoopVal[iRU][row];
+        this->saveThreshold(); // save before moving to the next vresetd
       }
     }
-    if (mScanType == 'r') {
-      this->saveThreshold(); // save before moving to the next vresetd
-    }
 
     // Fill the ScTree tree
     if (mScanType == 'T' || mScanType == 'V' || mScanType == 'I') { // TODO: store also for other scans?
@@ -794,6 +801,9 @@ void ITSThresholdCalibrator::extractThresholdRow(const short int& chipID, const
 
   // Saves threshold information to internal memory
   if (mScanType != 'P' && mScanType != 'p' && mScanType != 't' && mScanType != 'R' && mScanType != 'r') {
+    if(mVerboseOutput) {
+      LOG(info)<<"Saving data of ChipID: "<<chipID<<" Row: "<<row;
+    }
     this->saveThreshold();
   }
 }
@@ -934,6 +944,7 @@ void ITSThresholdCalibrator::setRunType(const short int& runtype)
     this->mMax = 50;
     this->N_RANGE = 51;
     this->mCheckExactRow = true;
+    mRowScan = 512;
 
   } else if (runtype == THR_SCAN_SHORT || runtype == THR_SCAN_SHORT_100HZ ||
              runtype == THR_SCAN_SHORT_200HZ || runtype == THR_SCAN_SHORT_33 || runtype == THR_SCAN_SHORT_2_10HZ || runtype == THR_SCAN_SHORT_150INJ) {
@@ -951,6 +962,12 @@ void ITSThresholdCalibrator::setRunType(const short int& runtype)
         nInjScaled = nInj / 3;
       }
     }
+    mRowScan = 11;
+    if (runtype == THR_SCAN_SHORT_33) {
+      mRowScan = 33;
+    } else if(runtype == THR_SCAN_SHORT_2_10HZ){
+      mRowScan = 2;
+    } 
   } else if (runtype == VCASN150 || runtype == VCASN100 || runtype == VCASN100_100HZ || runtype == VCASN130 || runtype == VCASNBB) {
     // VCASN tuning for different target thresholds
     // Store average VCASN for each chip into CCDB
@@ -962,6 +979,7 @@ void ITSThresholdCalibrator::setRunType(const short int& runtype)
     this->mMax = inMaxVcasn; // 80 is the default
     this->N_RANGE = mMax - mMin + 1;
     this->mCheckExactRow = true;
+    mRowScan = 4; 
 
   } else if (runtype == ITHR150 || runtype == ITHR100 || runtype == ITHR100_100HZ || runtype == ITHR130) {
     // ITHR tuning  -- average ITHR per chip
@@ -973,6 +991,7 @@ void ITSThresholdCalibrator::setRunType(const short int& runtype)
     this->mMax = inMaxIthr; // 100 is the default
     this->N_RANGE = mMax - mMin + 1;
     this->mCheckExactRow = true;
+    mRowScan = 4; 
 
   } else if (runtype == DIGITAL_SCAN || runtype == DIGITAL_SCAN_100HZ || runtype == DIGITAL_SCAN_NOMASK) {
     // Digital scan -- only storing one value per chip, no fit needed
@@ -1078,6 +1097,7 @@ void ITSThresholdCalibrator::setRunType(const short int& runtype)
       if (scaleNinj) {
         nInjScaled = nInj / 3;
       }
+      mRowScan = manualRowScan;
     } else {
       throw runtype;
     }
@@ -1367,15 +1387,14 @@ void ITSThresholdCalibrator::run(ProcessingContext& pc)
         row = !mCdwVersion ? (short int)(calib.calibUserField & 0xffff) : (short int)(calib.calibUserField & 0x1ff);
         // cw counter
         cwcnt = (short int)(calib.calibCounter);
-        // count the last N injections
-        short int checkVal = (mScanType == 'I') ? mMin : mMax;
-        if ((mScanType != 'r' && mScanType != 'p' && mScanType != 't' && loopval == checkVal) ||
-            (mScanType == 'r' && realcharge == mMax2) ||
-            (mScanType == 'p' && realcharge == mMin2) ||
-            (mScanType == 't' && loopval == checkVal && realcharge == mMax2)) {
-          mCdwCntRU[iRU][row]++;
-          mLoopVal[iRU][row] = loopval; // keep loop val (relevant for VRESET2D and TOT_1ROW scan only)
-        }
+
+        // count injections
+	short int loopPoint = (loopval - this->mMin) / mStep;
+	short int chgPoint = (realcharge - this->mMin2) / mStep2;
+	auto &arr = mCdwCntRU[iRU][row];
+	arr[chgPoint][loopPoint]++;
+
+
         if (this->mVerboseOutput) {
           LOG(info) << "RU: " << iRU << " CDWcounter: " << cwcnt << " row: " << row << " Loopval: " << loopval << " realcharge: " << realcharge << " confDBv: " << mCdwVersion;
           LOG(info) << "NDIGITS: " << digits.size();
@@ -1470,12 +1489,7 @@ void ITSThresholdCalibrator::run(ProcessingContext& pc)
     if (ruIndex < 0) {
       continue;
     }
-    short int nL = 0;
-    for (int iL = 0; iL < 3; iL++) {
-      if (mActiveLinks[ruIndex][iL]) {
-        nL++; // count active links
-      }
-    }
+    short int nL = ruIndex > 47 ? 2 : 3; // total number of links per RU 
     std::vector<short int> chipEnabled = getChipListFromRu(ruIndex, mActiveLinks[ruIndex]); // chip boundaries
     // Fill the chipDone info string
     if (mRunTypeRUCopy[ruIndex] == nInjScaled * nL) {
@@ -1488,14 +1502,27 @@ void ITSThresholdCalibrator::run(ProcessingContext& pc)
       mRunTypeRUCopy[ruIndex] = 0; // reset here is safer (the other counter is reset in finalize)
     }
     // Check if scan of a row is finished: only for specific scans!
-    bool passCondition = (mCdwCntRU[ruIndex][row] >= nInjScaled * nL);
-    if (mScanType == 'p' || mScanType == 't') {
-      passCondition = passCondition && (mLoopVal[ruIndex][row] == mMax);
-      if (mVerboseOutput) {
-        LOG(info) << "PassCondition: " << passCondition << " - (mCdwCntRU,mLoopVal) of RU" << ruIndex << " row " << row << " = (" << mCdwCntRU[ruIndex][row] << ", " << mLoopVal[ruIndex][row] << ")";
+    bool passCondition = true; 
+    for(int j1=0; j1 < N_RANGE2; j1++) {
+      for(int j2=0; j2 < N_RANGE; j2++) {
+	if(mScanType == 't') { // ToT scan is done in specific ranges depending on charge (see ITSComm)
+	  if ((j1 == 0 && j2<((600-mMin)/mStep)) || (j2>=((600-mMin)/mStep) && j2<=((800-mMin)/mStep)) || (j1==1 && j2>((800-mMin)/mStep))) {
+	    if(mCdwCntRU[ruIndex][row][j1][j2] < nInjScaled * nL) {
+	      passCondition = false;
+	      break;
+	    }
+	  } 
+        } else if(mCdwCntRU[ruIndex][row][j1][j2] < nInjScaled * nL) {
+          passCondition = false;
+	  break;
+	}
       }
-    } else if (mVerboseOutput) {
-      LOG(info) << "PassCondition: " << passCondition << " - mCdwCntRU of RU" << ruIndex << " row " << row << " = " << mCdwCntRU[ruIndex][row];
+      if(!passCondition){
+        break;
+      }
+    }
+    if (mVerboseOutput) {
+      LOG(info) << "PassCondition: " << passCondition << " - mCdwCntRU of RU" << ruIndex << " row " << row << " = " << mCdwCntRU[ruIndex][row][0][0] << "(Links: "<<mActiveLinks[ruIndex][0]<<", "<<mActiveLinks[ruIndex][1]<<","<<mActiveLinks[ruIndex][2]<<")";
     }
 
     if (mScanType != 'D' && mScanType != 'A' && mScanType != 'P' && mScanType != 'R' && passCondition) {
@@ -1509,32 +1536,35 @@ void ITSThresholdCalibrator::run(ProcessingContext& pc)
           if (mPixelHits.count(chipID)) {
             if (mPixelHits[chipID].count(row)) { // make sure the row exists
               extractAndUpdate(chipID, row);
-              if (mScanType != 'p' && (mScanType != 'r' || mLoopVal[ruIndex][row] == mMax)) { // do not erase for scantype = p because in finalize() we have calculate2Dparams
+              if (mScanType != 'p') { // do not erase for scantype = p because in finalize() we have calculate2Dparams
                 mPixelHits[chipID].erase(row);
               }
             }
           }
         }
       }
-      mCdwCntRU[ruIndex][row] = 0; // reset
+      mCdwCntRU[ruIndex].erase(row); // row is gone
+    }
+
+    if(mRunTypeRU[ruIndex] >= nInjScaled * nL && passCondition) {
+      mFlagsRU[ruIndex] = true;
+      finalize();
+      LOG(info) << "Shipping all outputs to aggregator (before endOfStream arrival!)";
+      pc.outputs().snapshot(Output{"ITS", "TSTR", (unsigned int)mChipModSel}, this->mTuning);
+      pc.outputs().snapshot(Output{"ITS", "PIXTYP", (unsigned int)mChipModSel}, this->mPixStat);
+      pc.outputs().snapshot(Output{"ITS", "RUNT", (unsigned int)mChipModSel}, this->mRunType);
+      pc.outputs().snapshot(Output{"ITS", "SCANT", (unsigned int)mChipModSel}, this->mScanType);
+      pc.outputs().snapshot(Output{"ITS", "FITT", (unsigned int)mChipModSel}, this->mFitType);
+      pc.outputs().snapshot(Output{"ITS", "CONFDBV", (unsigned int)mChipModSel}, this->mConfDBv);
+      pc.outputs().snapshot(Output{"ITS", "QCSTR", (unsigned int)mChipModSel}, this->mChipDoneQc);
+      // reset the DCSconfigObject_t before next ship out
+      mTuning.clear();
+      mPixStat.clear();
+      mChipDoneQc.clear();
     }
   } // for (ROFs)
 
-  if (!(this->mRunTypeUp)) {
-    finalize();
-    LOG(info) << "Shipping all outputs to aggregator (before endOfStream arrival!)";
-    pc.outputs().snapshot(Output{"ITS", "TSTR", (unsigned int)mChipModSel}, this->mTuning);
-    pc.outputs().snapshot(Output{"ITS", "PIXTYP", (unsigned int)mChipModSel}, this->mPixStat);
-    pc.outputs().snapshot(Output{"ITS", "RUNT", (unsigned int)mChipModSel}, this->mRunType);
-    pc.outputs().snapshot(Output{"ITS", "SCANT", (unsigned int)mChipModSel}, this->mScanType);
-    pc.outputs().snapshot(Output{"ITS", "FITT", (unsigned int)mChipModSel}, this->mFitType);
-    pc.outputs().snapshot(Output{"ITS", "CONFDBV", (unsigned int)mChipModSel}, this->mConfDBv);
-    pc.outputs().snapshot(Output{"ITS", "QCSTR", (unsigned int)mChipModSel}, this->mChipDoneQc);
-    // reset the DCSconfigObject_t before next ship out
-    mTuning.clear();
-    mPixStat.clear();
-    mChipDoneQc.clear();
-  } else if (pc.transitionState() == TransitionHandlingState::Requested) {
+  if (pc.transitionState() == TransitionHandlingState::Requested) {
     LOG(info) << "Run stop requested during the scan, sending output to aggregator and then stopping to process new data";
     mRunStopRequested = true;
     finalize();                                                                             // calculating average thresholds based on what's collected up to this moment
@@ -1769,13 +1799,13 @@ void ITSThresholdCalibrator::finalize()
     if (mScanType == 'I') {
       // Only ITHR scan: assign default ITHR = 50 if chip has no avg ITHR
       for (auto& iRU : mRuSet) {
-        if (mRunTypeRU[iRU] >= nInjScaled * getNumberOfActiveLinks(mActiveLinks[iRU]) || mRunStopRequested) {
+        if (mFlagsRU[iRU] || mRunStopRequested) {
           std::vector<short int> chipList = getChipListFromRu(iRU, mActiveLinks[iRU]);
           for (size_t i = 0; i < chipList.size(); i++) {
             if ((chipList[i] % mChipModBase) != mChipModSel) {
               continue;
             }
-            if (!mThresholds.count(chipList[i])) {
+            if (!mThresholds.count(chipList[i]) && !isChipDB[chipList[i]]) {
               if (mVerboseOutput) {
                 LOG(info) << "Setting ITHR = 50 for chip " << chipList[i];
               }
@@ -1790,7 +1820,7 @@ void ITSThresholdCalibrator::finalize()
     auto it = this->mThresholds.cbegin();
     while (it != this->mThresholds.cend()) {
       short int iRU = getRUID(it->first);
-      if (!isCRUITS && (mRunTypeRU[iRU] < nInjScaled * getNumberOfActiveLinks(mActiveLinks[iRU]) && !mRunStopRequested)) {
+      if (!isCRUITS && (!mFlagsRU[iRU]) && !mRunStopRequested) {
         ++it;
         continue;
       }
@@ -1804,8 +1834,8 @@ void ITSThresholdCalibrator::finalize()
       }
       if (mVerboseOutput) {
         LOG(info) << "Average or mpv " << name << " of chip " << it->first << " = " << outVal << " e-";
-      }
-      float status = ((float)it->second[4] / (float)(it->second[4] + it->second[5])) * 100.; // percentage of successful threshold extractions
+      } 
+      float status = ((float)it->second[4] / (float)(mRowScan * N_COL)) * 100.; // percentage of successful threshold extractions
       if (status < mPercentageCut && (mScanType == 'I' || mScanType == 'V')) {
         if (mScanType == 'I') { // default ITHR if percentage of success < mPercentageCut
           outVal = 50.;
@@ -1822,6 +1852,7 @@ void ITSThresholdCalibrator::finalize()
       }
       std::vector<float> data = {outVal, rmsT, avgN, rmsN, status};
       this->addDatabaseEntry(it->first, name, data, false);
+      isChipDB[it->first] = true;
       it = this->mThresholds.erase(it);
     }
   } else if (this->mScanType == 'D' || this->mScanType == 'A') {
@@ -1831,7 +1862,7 @@ void ITSThresholdCalibrator::finalize()
     auto itchip = this->mPixelHits.cbegin();
     while (itchip != this->mPixelHits.cend()) { // loop over chips collected
       short int iRU = getRUID(itchip->first);
-      if (!isCRUITS && (mRunTypeRU[iRU] < nInjScaled * getNumberOfActiveLinks(mActiveLinks[iRU]) && !mRunStopRequested)) {
+      if (!isCRUITS && !mFlagsRU[iRU] && !mRunStopRequested) {
         ++itchip;
         continue;
       }
@@ -1845,7 +1876,7 @@ void ITSThresholdCalibrator::finalize()
       if (this->mVerboseOutput) {
         LOG(info) << "Chip " << itchip->first << " hits extracted";
       }
-      ++itchip;
+      itchip = mPixelHits.erase(itchip);
     }
 
     auto it = this->mNoisyPixID.cbegin();
@@ -1886,7 +1917,7 @@ void ITSThresholdCalibrator::finalize()
     auto itchip = this->mPixelHits.cbegin();
     while (itchip != mPixelHits.cend()) {
       int iRU = getRUID(itchip->first);
-      if (!mRunStopRequested && mRunTypeRU[iRU] < nInjScaled * getNumberOfActiveLinks(mActiveLinks[iRU])) {
+      if (!mRunStopRequested && !mFlagsRU[iRU]) {
         ++itchip;
         continue;
       }
@@ -1909,15 +1940,11 @@ void ITSThresholdCalibrator::finalize()
       if (this->mVerboseOutput) {
         LOG(info) << "Chip " << itchip->first << " hits extracted";
       }
-      ++itchip;
+      itchip = mPixelHits.erase(itchip);
     }
     // reset RU counters so that the chips which are done will not appear again in the DCSConfigObject
   }
 
-  for (auto& ru : thisRUs) {
-    mRunTypeRU[ru] = 0; // reset
-  }
-
   return;
 }
 
@@ -1996,6 +2023,7 @@ DataProcessorSpec getITSThresholdCalibratorSpec(const ITSCalibInpConf& inpConf)
             {"manual-step2", VariantType::Int, 1, {"Step2 value: defines the steps between manual-min2 and manual-max2. Default is 1. Use only in manual mode"}},
             {"manual-scantype", VariantType::String, "T", {"scan type, can be D, T, I, V, P, p: use only in manual mode"}},
             {"manual-strobewindow", VariantType::Int, 5, {"strobe duration in clock cycles, default is 5 = 125 ns: use only in manual mode"}},
+	    {"manual-rowscan", VariantType::Int, 512, {"Number of ALPIDE rows scanned in the run: use only in manual mode"}},
             {"save-tree", VariantType::Bool, false, {"Flag to save ROOT tree on disk: use only in manual mode"}},
             {"scale-ninj", VariantType::Bool, false, {"Flag to activate the scale of the number of injects to be used to count hits from specific MEBs: use only in manual mode and in combination with --meb-select"}},
             {"enable-mpv", VariantType::Bool, false, {"Flag to enable calculation of most-probable value in vcasn/ithr scans"}},
