Make sure to print an error if data is lost while terminating DPL proxy

matthiasrichter · matthiasrichter · commit 2e46494d7940 · 2020-06-15T15:42:29.000+02:00
Amending PR #3801. Checking for the requested change of device state before dispatching messages might silently drop data. Have to ensure that message are always dispatched, and if the sending does not succeed an error message is printed. Only after that the pending device state change is checked and can break the loop.
diff --git a/Framework/Core/src/ExternalFairMQDeviceProxy.cxx b/Framework/Core/src/ExternalFairMQDeviceProxy.cxx
@@ -69,7 +69,7 @@ void sendOnChannel(FairMQDevice& device, FairMQParts& messages, std::string cons
   // TODO: we might want to treat this error condition some levels higher up, but for
   // the moment its an appropriate solution. The important thing is not to drop
   // messages and to be informed about the congestion.
-  while (device.NewStatePending() == false && device.Send(messages, channel, index, timeout) < 0) {
+  while (device.Send(messages, channel, index, timeout) < 0) {
     if (timeout == 0) {
       timeout = 1;
     } else if (timeout < maxTimeout) {
@@ -83,6 +83,15 @@ void sendOnChannel(FairMQDevice& device, FairMQParts& messages, std::string cons
         timeout += 1;
       }
     }
+    if (device.NewStatePending()) {
+      LOG(ERROR) << "device state change is requested, dropping " << messages.Size() << " pending message(s)\n"
+                 << "on channel " << channel << "\n"
+                 << "ATTENTION: DATA IS LOST! Could not dispatch data to downstream consumer(s), check if\n"
+                 << "consumers have been terminated too early";
+      // make sure we disable the warning below
+      timeout = maxTimeout + 1;
+      break;
+    }
   }
   if (timeout > 0 && timeout <= maxTimeout) {
     LOG(WARNING) << "dispatching on channel " << channel << " was delayed by " << timeout << " ms";
