Fix determining when the execution is really done in the presence of multiple readers

Author: gsakkis

pdal/StreamableExecutor.cpp

@@ -199,25 +199,18 @@ void PythonPointTable::reset()
         if (np)
         {
             m_arrays.push(m_curArray);
-            m_curArray = nullptr;
+            m_curArray = py_createArray();
+            m_producedCv.notify_one();
         }
-
-        bool done = numPoints() < m_limit;
-
-        // If we just pushed the last chunk, push a nullptr so that a reader knows.
-        if (done)
-            m_arrays.push(nullptr);
-        m_producedCv.notify_one();
-
-        if (done)
-            return;
-
         while (m_arrays.size() > m_prefetch)
             m_consumedCv.wait(l);
     }
+}
 
-    // Make a new array for data.
-    m_curArray = py_createArray();
+void PythonPointTable::done()
+{
+    m_arrays.push(nullptr);
+    m_producedCv.notify_one();
 }
 
 PyArrayObject *PythonPointTable::fetchArray()
@@ -267,6 +260,7 @@ PyArrayObject *StreamableExecutor::executeNext()
         m_thread.reset(new std::thread([this]()
         {
             m_manager.executeStream(m_table);
+            m_table.done();
         }));
     }
 

pdal/StreamableExecutor.hpp

@@ -59,6 +59,7 @@ class PythonPointTable : public StreamPointTable
     ~PythonPointTable();
 
     virtual void finalize();
+    void done();
     PyArrayObject *fetchArray();
 
 protected:

test/test_pipeline.py

@@ -464,7 +464,6 @@ def test_schema(self):
 
         assert ri.schema == r.schema
 
-    @pytest.mark.skip("segfaults")
     def test_merged_arrays(self):
         """Can we load data from a list of arrays to PDAL"""
         data = np.load(os.path.join(DATADIRECTORY, "test3d.npy"))
@@ -477,12 +476,11 @@ def test_merged_arrays(self):
             }
           ]
         }"""
-        arrays1 = list(pdal.Pipeline(filter_intensity, arrays, chunk_size=100))
-
         p = pdal.Pipeline(filter_intensity, arrays)
         p.execute()
-        arrays2 = p.arrays
-
-        assert len(arrays1) == len(arrays2)
-        for array1, array2 in zip(arrays1, arrays2):
-            np.testing.assert_array_equal(array1, array2)
+        non_streaming_array = np.concatenate(p.arrays)
+        for chunk_size in range(5, 100, 5):
+            streaming_arrays = list(pdal.Pipeline(filter_intensity, arrays,
+                                                  chunk_size=chunk_size))
+            np.testing.assert_array_equal(np.concatenate(streaming_arrays),
+                                          non_streaming_array)