@@ -23,6 +23,7 @@
 #include <string>
 #include <fairmq/Channel.h>
 #include <mutex>
+#include <vector>
 #include <fairlogger/Logger.h>
 #include <fairmq/Parts.h>
 #include <fairmq/TransportFactory.h>
@@ -274,36 +275,53 @@ bool waitForControlInput(int workerID)
 // and send their hits to the merger, as the forked worker processes do otherwise.
 std::atomic<bool> gServerDone{false};
 std::atomic<int> gNextWorkerID{0};
-// the transport threads create their channels at the same moment; the shared transport factory is not made for that
-std::mutex gChannelSetupMutex;
 
-std::unique_ptr<fair::mq::Channel> makeWorkerChannel(char const* name, char const* type, std::shared_ptr<fair::mq::TransportFactory> const& factory, std::string const& address)
+// The channels of the transport threads, kept here so that they can be closed before the process exits
+struct MTChannels {
+  std::shared_ptr<fair::mq::TransportFactory> factory;
+  std::mutex setupMutex; // the threads create their channels at the same moment; the factory is not made for that
+  // one request channel for all threads, used by one thread at a time: with one per thread, only one thread got work
+  std::mutex requestMutex;
+  std::unique_ptr<fair::mq::Channel> request;
+  std::vector<std::unique_ptr<fair::mq::Channel>> data; // one per thread
+};
+MTChannels gMTChannels;
+
+fair::mq::Channel* makeWorkerChannel(std::unique_ptr<fair::mq::Channel>& owner, char const* name, char const* type, std::string const& address)
 {
-  std::lock_guard<std::mutex> lock(gChannelSetupMutex);
-  auto channel = std::make_unique<fair::mq::Channel>(name, type, factory);
-  if (!channel->Connect(address) || !channel->Validate()) {
+  std::lock_guard<std::mutex> lock(gMTChannels.setupMutex);
+  owner = std::make_unique<fair::mq::Channel>(name, type, gMTChannels.factory);
+  if (!owner->Connect(address) || !owner->Validate()) {
     LOG(fatal) << "Could not connect channel " << name << " of a transport thread to " << address;
   }
-  return channel;
+  return owner.get();
+}
+
+// Closes the channels of the transport threads; with linger -1 this waits until the merger has taken all queued hits
+void closeMTChannels()
+{
+  auto& hooks = o2::base::MTWorkerHooks::instance();
+  hooks.fetchChunk = nullptr;
+  hooks.dataChannel = nullptr;
+  gMTChannels.request.reset();
+  gMTChannels.data.clear();
+  gMTChannels.factory.reset();
 }
 
 // The Geant4 worker threads already exist when this is called; they use these hooks from the first real ProcessRun.
 void installMTWorkerHooks(std::string const& serveraddress, std::string const& mergeraddress)
 {
-  auto factory = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
+  gMTChannels.factory = fair::mq::TransportFactory::CreateTransportFactory("zeromq");
   auto& hooks = o2::base::MTWorkerHooks::instance();
-  hooks.fetchChunk = [factory, serveraddress]() -> std::unique_ptr<o2::data::PrimaryChunk> {
+  hooks.fetchChunk = [serveraddress]() -> std::unique_ptr<o2::data::PrimaryChunk> {
     thread_local int workerID = gNextWorkerID++;
-    // one request channel for all threads, used by one thread at a time
-    static std::mutex fetchMutex;
-    static std::unique_ptr<fair::mq::Channel> channel;
-    std::lock_guard<std::mutex> lock(fetchMutex);
-    if (!channel) {
-      channel = makeWorkerChannel("primary-get", "req", factory, serveraddress);
+    std::lock_guard<std::mutex> lock(gMTChannels.requestMutex);
+    if (!gMTChannels.request) {
+      makeWorkerChannel(gMTChannels.request, "primary-get", "req", serveraddress);
     }
     using Status = o2::devices::O2SimDevice::ChunkStatus;
     while (true) {
-      auto fetched = o2::devices::O2SimDevice::fetchChunk(workerID, *channel);
+      auto fetched = o2::devices::O2SimDevice::fetchChunk(workerID, *gMTChannels.request);
       if (fetched.status == Status::Work) {
         return std::move(fetched.chunk);
       }
@@ -314,12 +332,16 @@ void installMTWorkerHooks(std::string const& serveraddress, std::string const& m
       // Retry and EndOfWork: ask again, as a worker process does; the server answers as idle once done
     }
   };
-  hooks.dataChannel = [factory, mergeraddress]() -> fair::mq::Channel* {
-    thread_local std::unique_ptr<fair::mq::Channel> channel;
+  hooks.dataChannel = [mergeraddress]() -> fair::mq::Channel* {
+    thread_local fair::mq::Channel* channel = nullptr;
     if (!channel) {
-      channel = makeWorkerChannel("simdata", "push", factory, mergeraddress);
+      std::unique_ptr<fair::mq::Channel> owner;
+      channel = makeWorkerChannel(owner, "simdata", "push", mergeraddress);
+      channel->GetSocket().SetLinger(-1); // keep queued hits until the merger has taken them
+      std::lock_guard<std::mutex> lock(gMTChannels.setupMutex);
+      gMTChannels.data.push_back(std::move(owner));
     }
-    return channel.get();
+    return channel;
   };
 }
 
@@ -341,8 +363,7 @@ void installMTWorkerHooks(std::string const& serveraddress, std::string const& m
   }).detach();
   o2::simpubsub::publishMessage(pushchannel, o2::simpubsub::simStatusString("WORKER0", "STATUS", "SETUP COMPLETED"));
 
-  // a run ends when every thread has been told that there is no more work; ProcessRun is given a large
-  // but finite number of events and is repeated if the server still has work when it is exhausted
+  // a run ends once every thread was told there is no more work; it is repeated if the events ran out first
   constexpr int kEventBlock = 100000;
   auto vmc = TVirtualMC::GetMC();
   if (auto& beforeTransport = o2::base::MTWorkerHooks::instance().beforeTransport) {
@@ -368,7 +389,7 @@ void installMTWorkerHooks(std::string const& serveraddress, std::string const& m
     }
   }
   o2::SimSetup::shutdown();
-  sleep(10); // give the merger time to receive the last messages
+  closeMTChannels();
   _exit(0);
 }