@@ -25,6 +25,7 @@ o2_add_executable(digitizer-workflow
                           src/SimReaderSpec.cxx
                           src/SimpleDigitizerWorkflow.cxx
                           src/TPCDigitizerSpec.cxx
+                          src/TPCSectorDigitization.cxx
                           src/ZDCDigitizerSpec.cxx
                           src/TOFDigitizerSpec.cxx
                           $<$<BOOL:${ENABLE_UPGRADES}>:src/ITS3DigitizerSpec.cxx>
 
@@ -15,14 +15,14 @@
 #include <cassert>
 #include "Framework/RootSerializationSupport.h"
 #include "TPCDigitizerSpec.h"
+#include "TPCSectorDigitization.h"
 #include "Framework/ControlService.h"
 #include "Framework/ParallelContext.h"
 #include "Framework/ConfigParamRegistry.h"
 #include "Framework/DataProcessorSpec.h"
 #include "Framework/DataRefUtils.h"
 #include "Framework/Lifetime.h"
 #include "Framework/DeviceSpec.h"
-#include "DetectorsRaw/HBFUtils.h"
 #include "Headers/DataHeader.h"
 #include "TStopwatch.h"
 #include "Steer/HitProcessingManager.h" // for DigitizationContext
@@ -92,20 +92,6 @@ void writeToBranchHelper<o2::dataformats::MCTruthContainer<o2::MCCompLabel>>(TTr
   targetbr->DropBaskets("all");
 }
 
-std::string getBranchNameLeft(int sector)
-{
-  std::stringstream branchnamestreamleft;
-  branchnamestreamleft << "TPCHitsShiftedSector" << int(o2::tpc::Sector::getLeft(o2::tpc::Sector(sector)));
-  return branchnamestreamleft.str();
-}
-
-std::string getBranchNameRight(int sector)
-{
-  std::stringstream branchnamestreamright;
-  branchnamestreamright << "TPCHitsShiftedSector" << sector;
-  return branchnamestreamright.str();
-}
-
 using namespace o2::base;
 class TPCDPLDigitizerTask : public BaseDPLDigitizer
 {
@@ -178,23 +164,23 @@ class TPCDPLDigitizerTask : public BaseDPLDigitizer
     {
       std::stringstream brname;
       brname << "TPCDigit_" << mSector;
-      auto br = o2::base::getOrMakeBranch(*mInternalROOTFlushTTree, brname.str().c_str(), &mDigits);
+      auto br = o2::base::getOrMakeBranch(*mInternalROOTFlushTTree, brname.str().c_str(), &mFlushBuffers.digits);
       br->Fill();
       br->ResetAddress();
     }
     if (mWithMCTruth) {
       // labels
       std::stringstream brname;
       brname << "TPCDigitMCTruth_" << mSector;
-      auto br = o2::base::getOrMakeBranch(*mInternalROOTFlushTTree, brname.str().c_str(), &mLabels);
+      auto br = o2::base::getOrMakeBranch(*mInternalROOTFlushTTree, brname.str().c_str(), &mFlushBuffers.labels);
       br->Fill();
       br->ResetAddress();
     }
     {
       // common
       std::stringstream brname;
       brname << "TPCCommonMode_" << mSector;
-      auto br = o2::base::getOrMakeBranch(*mInternalROOTFlushTTree, brname.str().c_str(), &mCommonMode);
+      auto br = o2::base::getOrMakeBranch(*mInternalROOTFlushTTree, brname.str().c_str(), &mFlushBuffers.commonMode);
       br->Fill();
       br->ResetAddress();
     }
@@ -286,7 +272,6 @@ class TPCDPLDigitizerTask : public BaseDPLDigitizer
         }
         // TODO: make generic reset method?
         mFlushCounter = 0;
-        mDigitCounter = 0;
       }
     }
   }
@@ -391,88 +376,24 @@ class TPCDPLDigitizerTask : public BaseDPLDigitizer
       throw std::runtime_error("Digitizer can only work on single sectors");
     }
 
-    mDigitizer.setSector(sector);
-    mDigitizer.setSeedingTimeframe(dh->tfCounter);
-    mDigitizer.init();
-
-    auto& eventParts = context->getEventParts();
+    TStopwatch timer;
+    timer.Start();
 
-    auto flushDigitsAndLabels = [this, digitsAccum, &labelAccum, &commonModeAccum](bool finalFlush = false) {
+    auto onFlush = [this, digitsAccum, &labelAccum, &commonModeAccum]() {
       mFlushCounter++;
-      // flush previous buffer
-      mDigits.clear();
-      mLabels.clear();
-      mCommonMode.clear();
-      mDigitizer.flush(mDigits, mLabels, mCommonMode, finalFlush);
-      LOG(info) << "TPC: Flushed " << mDigits.size() << " digits, " << mLabels.getNElements() << " labels and " << mCommonMode.size() << " common mode entries";
-
       if (mInternalWriter) {
         // the natural place to write out this independent datachunk immediately ...
         writeToROOTFile();
       } else {
         // ... or to accumulate and later forward to next DPL proc
-        std::copy(mDigits.begin(), mDigits.end(), std::back_inserter(*digitsAccum));
+        std::copy(mFlushBuffers.digits.begin(), mFlushBuffers.digits.end(), std::back_inserter(*digitsAccum));
         if (mWithMCTruth) {
-          labelAccum.mergeAtBack(mLabels);
+          labelAccum.mergeAtBack(mFlushBuffers.labels);
         }
-        std::copy(mCommonMode.begin(), mCommonMode.end(), std::back_inserter(commonModeAccum));
+        std::copy(mFlushBuffers.commonMode.begin(), mFlushBuffers.commonMode.end(), std::back_inserter(commonModeAccum));
       }
-      mDigitCounter += mDigits.size();
     };
-
-    if (isContinuous) {
-      auto& hbfu = o2::raw::HBFUtils::Instance();
-      double time = hbfu.getFirstIRofTF(o2::InteractionRecord(0, hbfu.orbitFirstSampled)).bc2ns() / 1000.;
-      mDigitizer.setOutputDigitTimeOffset(time);
-      if (!irecords.empty()) {
-        mDigitizer.setStartTime(irecords[0].getTimeNS() / 1000.f);
-      }
-    }
-
-    TStopwatch timer;
-    timer.Start();
-
-    // loop over all composite collisions given from context
-    // (aka loop over all the interaction records)
-    for (int collID = 0; collID < irecords.size(); ++collID) {
-      const double eventTime = irecords[collID].getTimeNS() / 1000.f;
-      LOG(info) << "TPC: Event time " << eventTime << " us";
-      mDigitizer.setEventTime(eventTime);
-      if (!isContinuous) {
-        mDigitizer.setStartTime(eventTime);
-      }
-      size_t startSize = mDigitCounter; // digitsAccum->size();
-
-      // for each collision, loop over the constituents event and source IDs
-      // (background signal merging is basically taking place here)
-      for (auto& part : eventParts[collID]) {
-        const int eventID = part.entryID;
-        const int sourceID = part.sourceID;
-
-        // get the hits for this event and this source
-        std::vector<o2::tpc::HitGroup> hitsLeft;
-        std::vector<o2::tpc::HitGroup> hitsRight;
-        context->retrieveHits(mSimChains, getBranchNameLeft(sector).c_str(), part.sourceID, part.entryID, &hitsLeft);
-        context->retrieveHits(mSimChains, getBranchNameRight(sector).c_str(), part.sourceID, part.entryID, &hitsRight);
-        LOG(debug) << "TPC: Found " << hitsLeft.size() << " hit groups left and " << hitsRight.size() << " hit groups right in collision " << collID << " eventID " << part.entryID;
-
-        mDigitizer.process(hitsLeft, eventID, sourceID);
-        mDigitizer.process(hitsRight, eventID, sourceID);
-
-        flushDigitsAndLabels();
-
-        if (!isContinuous) {
-          eventAccum.emplace_back(startSize, mDigits.size());
-        }
-      }
-    }
-
-    // final flushing step; getting everything not yet written out
-    if (isContinuous) {
-      LOG(info) << "TPC: Final flush";
-      flushDigitsAndLabels(true);
-      eventAccum.emplace_back(0, mDigitCounter); // all digits are grouped to 1 super-event pseudo-triggered mode
-    }
+    eventAccum = digitizeSector(mDigitizer, *context, mSimChains, sector, dh->tfCounter, mFlushBuffers, onFlush);
 
     if (!mInternalWriter) {
       // send out to next stage
@@ -490,13 +411,10 @@ class TPCDPLDigitizerTask : public BaseDPLDigitizer
   o2::tpc::Digitizer mDigitizer;
   o2::tpc::VDriftHelper mTPCVDriftHelper{};
   std::vector<TChain*> mSimChains;
-  std::vector<o2::tpc::Digit> mDigits;
-  o2::dataformats::MCTruthContainer<o2::MCCompLabel> mLabels;
-  std::vector<o2::tpc::CommonMode> mCommonMode;
-  std::vector<int> mListOfSectors; //  a list of sectors treated by this task
+  SectorFlushBuffers mFlushBuffers; // output of the latest flush
+  std::vector<int> mListOfSectors;  //  a list of sectors treated by this task
   TFile* mInternalROOTFlushFile = nullptr;
   TTree* mInternalROOTFlushTTree = nullptr;
-  size_t mDigitCounter = 0;
   size_t mFlushCounter = 0;
   int mLaneId = 0; // the id of the current process within the parallel pipeline
   int mSector = 0;
 
@@ -0,0 +1,120 @@
+// 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.
+
+#include "TPCSectorDigitization.h"
+#include <sstream>
+#include <string>
+#include <TChain.h>
+#include "DetectorsRaw/HBFUtils.h"
+#include "Framework/Logger.h"
+#include "SimulationDataFormat/DigitizationContext.h"
+#include "TPCBase/Sector.h"
+#include "TPCSimulation/Point.h"
+#include "TPCSimulation/Digitizer.h"
+
+namespace o2::tpc
+{
+
+namespace
+{
+std::string getBranchNameLeft(int sector)
+{
+  std::stringstream branchnamestreamleft;
+  branchnamestreamleft << "TPCHitsShiftedSector" << int(o2::tpc::Sector::getLeft(o2::tpc::Sector(sector)));
+  return branchnamestreamleft.str();
+}
+
+std::string getBranchNameRight(int sector)
+{
+  std::stringstream branchnamestreamright;
+  branchnamestreamright << "TPCHitsShiftedSector" << sector;
+  return branchnamestreamright.str();
+}
+} // namespace
+
+std::vector<dataformats::RangeReference<int, int>> digitizeSector(Digitizer& digitizer, const steer::DigitizationContext& context,
+                                                                  const std::vector<TChain*>& simChains, int sector, ULong64_t timeframe,
+                                                                  SectorFlushBuffers& buffers, const std::function<void()>& onFlush)
+{
+  std::vector<dataformats::RangeReference<int, int>> events;
+  const bool isContinuous = digitizer.isContinuousReadout();
+  const auto& irecords = context.getEventRecords();
+  const auto& eventParts = context.getEventParts();
+
+  digitizer.setSector(sector);
+  digitizer.setSeedingTimeframe(timeframe);
+  digitizer.init();
+
+  size_t digitCounter = 0;
+  auto flush = [&](bool finalFlush = false) {
+    buffers.digits.clear();
+    buffers.labels.clear();
+    buffers.commonMode.clear();
+    digitizer.flush(buffers.digits, buffers.labels, buffers.commonMode, finalFlush);
+    LOG(info) << "TPC: Flushed " << buffers.digits.size() << " digits, " << buffers.labels.getNElements() << " labels and " << buffers.commonMode.size() << " common mode entries";
+    onFlush();
+    digitCounter += buffers.digits.size();
+  };
+
+  if (isContinuous) {
+    auto& hbfu = o2::raw::HBFUtils::Instance();
+    double time = hbfu.getFirstIRofTF(o2::InteractionRecord(0, hbfu.orbitFirstSampled)).bc2ns() / 1000.;
+    digitizer.setOutputDigitTimeOffset(time);
+    if (!irecords.empty()) {
+      digitizer.setStartTime(irecords[0].getTimeNS() / 1000.f);
+    }
+  }
+
+  // loop over all composite collisions given from context
+  // (aka loop over all the interaction records)
+  for (int collID = 0; collID < irecords.size(); ++collID) {
+    const double eventTime = irecords[collID].getTimeNS() / 1000.f;
+    LOG(info) << "TPC: Event time " << eventTime << " us";
+    digitizer.setEventTime(eventTime);
+    if (!isContinuous) {
+      digitizer.setStartTime(eventTime);
+    }
+    size_t startSize = digitCounter;
+
+    // for each collision, loop over the constituents event and source IDs
+    // (background signal merging is basically taking place here)
+    for (auto& part : eventParts[collID]) {
+      const int eventID = part.entryID;
+      const int sourceID = part.sourceID;
+
+      // get the hits for this event and this source
+      std::vector<o2::tpc::HitGroup> hitsLeft;
+      std::vector<o2::tpc::HitGroup> hitsRight;
+      context.retrieveHits(simChains, getBranchNameLeft(sector).c_str(), part.sourceID, part.entryID, &hitsLeft);
+      context.retrieveHits(simChains, getBranchNameRight(sector).c_str(), part.sourceID, part.entryID, &hitsRight);
+      LOG(debug) << "TPC: Found " << hitsLeft.size() << " hit groups left and " << hitsRight.size() << " hit groups right in collision " << collID << " eventID " << part.entryID;
+
+      digitizer.process(hitsLeft, eventID, sourceID);
+      digitizer.process(hitsRight, eventID, sourceID);
+
+      flush();
+
+      if (!isContinuous) {
+        events.emplace_back(startSize, buffers.digits.size());
+      }
+    }
+  }
+
+  // final flushing step; getting everything not yet written out
+  if (isContinuous) {
+    LOG(info) << "TPC: Final flush";
+    flush(true);
+    events.emplace_back(0, digitCounter); // all digits are grouped to 1 super-event pseudo-triggered mode
+  }
+  return events;
+}
+
+} // namespace o2::tpc
@@ -0,0 +1,55 @@
+// 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.
+
+#ifndef STEER_DIGITIZERWORKFLOW_TPCSECTORDIGITIZATION_H_
+#define STEER_DIGITIZERWORKFLOW_TPCSECTORDIGITIZATION_H_
+
+#include <functional>
+#include <vector>
+#include <RtypesCore.h>
+#include "CommonDataFormat/RangeReference.h"
+#include "DataFormatsTPC/Digit.h"
+#include "SimulationDataFormat/MCCompLabel.h"
+#include "SimulationDataFormat/MCTruthContainer.h"
+#include "TPCSimulation/CommonMode.h"
+
+class TChain;
+
+namespace o2
+{
+namespace steer
+{
+class DigitizationContext;
+}
+namespace tpc
+{
+class Digitizer;
+
+/// Output buffers of one flush of a sector
+struct SectorFlushBuffers {
+  std::vector<Digit> digits;
+  dataformats::MCTruthContainer<MCCompLabel> labels;
+  std::vector<CommonMode> commonMode;
+};
+
+/// Digitize all collisions of one sector of a timeframe.
+/// After each flush the new output is in buffers and onFlush is called on the calling thread.
+/// The function makes no DPL calls. Concurrent calls need one Digitizer with sector seeding and one set of chains
+/// per thread; Digitizer::init() is called here and is not thread-safe across Digitizer instances.
+/// \return the digit groups, one per collision part in triggered mode, one for the timeframe in continuous mode
+std::vector<dataformats::RangeReference<int, int>> digitizeSector(Digitizer& digitizer, const steer::DigitizationContext& context,
+                                                                  const std::vector<TChain*>& simChains, int sector, ULong64_t timeframe,
+                                                                  SectorFlushBuffers& buffers, const std::function<void()>& onFlush);
+
+} // namespace tpc
+} // namespace o2
+
+#endif