Fix cluster loss on ClusterFinderMT shutdown

collect() cached queue emptiness across sweeps, so a worker's last
write could land after the final sweep but before shutdown was
observed, stranding that frame's clusters. Read the stop flags before
sweeping instead: once workers are joined, a sweep that moves nothing
proves the queues are drained.

Found via ClusterFinderCUDA.ipynb's drift plot, which jumped by one
frame's worth of clusters at the very end.

include/aare/ClusterFinderMT.hpp

@@ -135,26 +135,46 @@ class ClusterFinderMT {
      * @brief Collect all the clusters from the output queues and write them to
      * the sink
      */
+    // void collect() {
+    //     bool empty = true;
+    //     Backoff backoff;
+    //     while (!m_stop_requested || !empty || !m_processing_threads_stopped)
+    //     {
+    //         bool moved_any = false;
+    //         for (auto &queue : m_output_queues) {
+    //             while (auto *front = queue->frontPtr()) {
+    //                 while (!m_sink.write(std::move(*front))) {
+    //                     backoff.pause();
+    //                 }
+    //                 queue->popFront();
+    //                 moved_any = true;
+    //             }
+    //         }
+    //         empty = !moved_any;
+    //         if (moved_any) {
+    //             backoff.reset();
+    //         } else {
+    //             backoff.pause();
+    //         }
+    //     }
+    // }
     void collect() {
-        bool empty = true;
         Backoff backoff;
-        while (!m_stop_requested || !empty || !m_processing_threads_stopped) {
+        for (;;) {
+            const bool done = m_stop_requested && m_processing_threads_stopped;
             bool moved_any = false;
             for (auto &queue : m_output_queues) {
                 while (auto *front = queue->frontPtr()) {
-                    while (!m_sink.write(std::move(*front))) {
+                    while (!m_sink.write(std::move(*front)))
                         backoff.pause();
-                    }
                     queue->popFront();
                     moved_any = true;
                 }
             }
-            empty = !moved_any;
-            if (moved_any) {
-                backoff.reset();
-            } else {
-                backoff.pause();
-            }
+            if (done &&
+                !moved_any) // swept after observing done, so this is real
+                break;
+            moved_any ? backoff.reset() : backoff.pause();
         }
     }