Catch any exception thrown by the StreamableExecutor thread and rethrow it from executeNext

Author: gsakkis

pdal/StreamableExecutor.cpp

@@ -188,13 +188,15 @@ StreamableExecutor::StreamableExecutor(std::string const& json,
                                        int prefetch)
     : PipelineExecutor(json, arrays, level)
     , m_table(chunkSize, prefetch)
+    , m_exc(nullptr)
 {
-    if (!m_manager.pipelineStreamable())
-        throw pdal_error("Pipeline is not streamable");
-
     m_thread.reset(new std::thread([this]()
     {
-        m_manager.executeStream(m_table);
+        try {
+            m_manager.executeStream(m_table);
+        } catch (...) {
+            m_exc = std::current_exception();
+        }
         m_table.done();
     }));
 }
@@ -220,6 +222,8 @@ PyArrayObject *StreamableExecutor::executeNext()
         arr = m_table.fetchArray();
         if (arr == nullptr)
             m_executed = true;
+        if (m_exc)
+            std::rethrow_exception(m_exc);
     }
     return arr;
 }

pdal/StreamableExecutor.hpp

@@ -91,6 +91,7 @@ class StreamableExecutor : public PipelineExecutor
 
     PythonPointTable m_table;
     std::unique_ptr<std::thread> m_thread;
+    std::exception_ptr m_exc;
 };
 
 } // namespace python

test/test_pipeline.py

@@ -436,9 +436,10 @@ class TestPipelineIterator:
     @pytest.mark.parametrize("filename", ["sort.json", "sort.py"])
     def test_non_streamable(self, filename):
         r = get_pipeline(filename)
+        assert not r.streamable
         with pytest.raises(RuntimeError) as info:
-            r.iterator()
-        assert "Pipeline is not streamable" in str(info.value)
+            next(r.iterator(chunk_size=100))
+        assert "Attempting to use stream mode" in str(info.value)
 
     @pytest.mark.parametrize("filename", ["range.json", "range.py"])
     def test_array(self, filename):