Add cancel() method to measurement

Author: ayllon

SEFramework/SEFramework/Pipeline/Measurement.h

@@ -1,4 +1,5 @@
-/** Copyright © 2019-2022 Université de Genève, LMU Munich - Faculty of Physics, IAP-CNRS/Sorbonne Université
+/**
+ * Copyright © 2019-2022 Université de Genève, LMU Munich - Faculty of Physics, IAP-CNRS/Sorbonne Université
  *
  * This library is free software; you can redistribute it and/or modify it under
  * the terms of the GNU Lesser General Public License as published by the Free
@@ -39,6 +40,7 @@ class Measurement : public PipelineReceiver<SourceGroupInterface>, public Pipeli
   virtual void startThreads() = 0;
   virtual void stopThreads() = 0;
   virtual void synchronizeThreads() = 0;
+  virtual void cancel() = 0;
 };
 
 }

SEImplementation/SEImplementation/Measurement/MultithreadedMeasurement.h

@@ -46,6 +46,7 @@ class MultithreadedMeasurement : public Measurement {
       , m_group_counter(0)
       , m_input_done(false)
       , m_abort_raised(false)
+      , m_cancel(false)
       , m_semaphore(max_queue_size) {}
 
   ~MultithreadedMeasurement() override;
@@ -57,6 +58,10 @@ class MultithreadedMeasurement : public Measurement {
   void stopThreads() override;
   void synchronizeThreads() override;
 
+  void cancel() override {
+    m_cancel = true;
+  }
+
 private:
   using QueuePair = std::pair<int, std::unique_ptr<SourceGroupInterface>>;
   // We want O(1) for the *lowest* value (received order)
@@ -70,7 +75,7 @@ class MultithreadedMeasurement : public Measurement {
   std::unique_ptr<std::thread>        m_output_thread;
 
   int              m_group_counter;
-  std::atomic_bool m_input_done, m_abort_raised;
+  std::atomic_bool m_input_done, m_abort_raised, m_cancel;
 
   std::condition_variable                         m_new_output;
   OutputQueue                                     m_output_queue;

SEImplementation/src/lib/Measurement/MultithreadedMeasurement.cpp

@@ -80,6 +80,10 @@ void MultithreadedMeasurement::receiveSource(std::unique_ptr<SourceGroupInterfac
   // Put the new SourceGroup into the input queue
   auto order_number = m_group_counter;
   auto lambda       = [this, order_number, source_group = std::move(source_group)]() mutable {
+    // Flush the queue without measurements on cancel
+    if (m_cancel) {
+      return;
+    }
     // Trigger measurements
     for (const auto& source : *source_group) {
       m_source_to_row(source);
@@ -114,7 +118,7 @@ void MultithreadedMeasurement::outputThreadStatic(MultithreadedMeasurement* meas
 void MultithreadedMeasurement::outputThreadLoop() {
   int next_id = 0;
 
-  while (m_thread_pool->activeThreads() > 0) {
+  while (m_thread_pool->activeThreads() > 0 && !m_cancel) {
     std::unique_lock<std::mutex> output_lock(m_output_queue_mutex);
 
     // Wait for something in the output queue