Simplify StreamableExecutor

Author: gsakkis

pdal/PyPipeline.hpp

@@ -55,7 +55,6 @@ class PDAL_DLL PipelineExecutor {
 
     point_count_t execute();
 
-    bool executed() const { return m_executed; }
     const PointViewSet& views() const;
     std::string getPipeline() const;
     std::string getMetadata() const;

pdal/StreamableExecutor.cpp

@@ -190,49 +190,37 @@ StreamableExecutor::StreamableExecutor(std::string const& json,
 {
     if (!m_manager.pipelineStreamable())
         throw pdal_error("Pipeline is not streamable");
-}
-
-StreamableExecutor::~StreamableExecutor()
-{
-    stop();
-}
 
-PyArrayObject *StreamableExecutor::executeNext()
-{
-    if (!m_thread)
+    m_thread.reset(new std::thread([this]()
     {
-        m_thread.reset(new std::thread([this]()
-        {
-            m_manager.executeStream(m_table);
-            m_table.done();
-        }));
-    }
-
-    // Blocks until something is ready.
-    PyArrayObject *arr = m_table.fetchArray();
-    if (arr == nullptr)
-        done();
-    return arr;
+        m_manager.executeStream(m_table);
+        m_table.done();
+    }));
 }
 
-void StreamableExecutor::stop()
+StreamableExecutor::~StreamableExecutor()
 {
-    if (m_thread)
+    if (!m_executed)
     {
         m_table.disable();
         while (PyArrayObject* arr = m_table.fetchArray())
             Py_XDECREF(arr);
-        done();
     }
-}
-
-void StreamableExecutor::done()
-{
     Py_BEGIN_ALLOW_THREADS
     m_thread->join();
     Py_END_ALLOW_THREADS
-    m_thread.reset();
-    m_executed = true;
+}
+
+PyArrayObject *StreamableExecutor::executeNext()
+{
+    PyArrayObject* arr = nullptr;
+    if (!m_executed)
+    {
+        arr = m_table.fetchArray();
+        if (arr == nullptr)
+            m_executed = true;
+    }
+    return arr;
 }
 
 } // namespace python

pdal/StreamableExecutor.hpp

@@ -84,12 +84,9 @@ class StreamableExecutor : public PipelineExecutor
                        int prefetch);
     ~StreamableExecutor();
 
-    std::string getSchema() const;
     PyArrayObject* executeNext();
-    void stop();
 
 private:
-    void done();
     ConstPointTableRef pointTable() const { return m_table; }
 
     PythonPointTable m_table;

pdal/libpdalpython.pyx

@@ -7,7 +7,6 @@ from types import SimpleNamespace
 cimport cython
 from cython.operator cimport dereference as deref
 from cpython.ref cimport Py_DECREF
-from libcpp cimport bool
 from libcpp.memory cimport make_shared, shared_ptr, unique_ptr
 from libcpp.set cimport set as cpp_set
 from libcpp.string cimport string
@@ -93,7 +92,6 @@ cdef extern from "PyPipeline.hpp" namespace "pdal::python":
     cdef cppclass PipelineExecutor:
         PipelineExecutor(string, vector[shared_ptr[Array]], int) except +
         int execute() except +
-        bool executed() except +
         string getPipeline() except +
         string getMetadata() except +
         string getSchema() except +
@@ -105,7 +103,6 @@ cdef extern from "StreamableExecutor.hpp" namespace "pdal::python":
     cdef cppclass StreamableExecutor(PipelineExecutor):
         StreamableExecutor(string, vector[shared_ptr[Array]], int, int, int) except +
         np.PyArrayObject* executeNext() except +
-        void stop() except +
 
 
 cdef extern from "PyArray.hpp" namespace "pdal::python":
@@ -222,25 +219,20 @@ cdef class Pipeline(PipelineResultsMixin):
 cdef class PipelineIterator(PipelineResultsMixin):
     cdef unique_ptr[StreamableExecutor] _executor
 
-    cdef set_executor(self, StreamableExecutor* executor):
-        self._executor.reset(executor)
-
     def __iter__(self):
         return self
 
     def __next__(self):
-        cdef StreamableExecutor* executor = self._executor.get()
-        if executor.executed():
-            raise StopIteration
-
-        arr_ptr = executor.executeNext()
+        arr_ptr = self._executor.get().executeNext()
         if arr_ptr is NULL:
-            executor.stop()
             raise StopIteration
 
         arr = <object> arr_ptr
         Py_DECREF(arr)
         return arr
 
+    cdef set_executor(self, StreamableExecutor* executor):
+        self._executor.reset(executor)
+
     cdef PipelineExecutor* _get_executor(self) except NULL:
         return self._executor.get()