refactor: pipeline event.

Author: eliasyaoyc

src/execution/executor.rs

@@ -0,0 +1,104 @@
+use std::{collections::VecDeque, sync::Arc};
+
+use anyhow::Result;
+
+use super::executor_graph::ExecutorGraph;
+use super::executor_task::ExecutionTask;
+use super::parallel::pipeline_builder::PipelineBuilder;
+use super::physical::PhysicalOperatorRef;
+use super::runtime::thread::Thread;
+use super::runtime::thread::ThreadJoinHandle;
+use super::runtime::ExecutionRuntime;
+use super::runtime::TrySpawn;
+
+pub struct ExecutionQueue(pub VecDeque<ExecutionTask>);
+
+impl ExecutionQueue {
+    pub fn push_task(&mut self, task: ExecutionTask) {
+        self.0.push_back(task);
+    }
+}
+
+pub struct ExecutionContext {
+    pub global_execution_queue: ExecutionQueue,
+}
+
+impl ExecutionContext {
+    pub fn get_global_execution_queue(&mut self) -> &mut ExecutionQueue {
+        &mut self.global_execution_queue
+    }
+}
+
+pub struct Executor {
+    pub context: ExecutionContext,
+    pub graph: ExecutorGraph,
+    pub threads_num: usize,
+    pub runtime: Arc<ExecutionRuntime>,
+}
+
+impl Executor {
+    pub async fn initialize(plan: PhysicalOperatorRef) -> Result<Executor> {
+        let mut builder = PipelineBuilder::new();
+        let meta_pipeline = builder.finalize(plan)?;
+
+        let graph = ExecutorGraph::from_pipeline(meta_pipeline)?;
+
+        Ok(Executor {
+            context: ExecutionContext {
+                global_execution_queue: ExecutionQueue(VecDeque::new()),
+            },
+            graph,
+            threads_num: 0,
+            runtime: Arc::new(ExecutionRuntime::with_default_worker_threads()?),
+        })
+    }
+
+    pub async fn execute(&mut self) -> Result<()> {
+        // Schedule the pipelines that do not have dependencies.
+        self.graph
+            .init_schedule_event_queue(&mut self.context.global_execution_queue)
+            .await;
+
+        // Pull execution task from `GlobalExecutionQueue`.
+        self.runtime.spawn(ExecutionTask::empty());
+        Ok(())
+    }
+
+    fn execute_threads(self: &Arc<Self>) -> Vec<ThreadJoinHandle<Result<()>>> {
+        let mut thread_join_handles = Vec::with_capacity(self.threads_num);
+        for thread_num in 0..self.threads_num {
+            let this = self.clone();
+            #[allow(unused_mut)]
+            let mut name = Some(format!("PipelineTask-{}", thread_num));
+
+            thread_join_handles.push(Thread::named_spawn(name, move || unsafe {
+                // let this_clone = this.clone();
+
+                // let try_result = catch_unwind(move || -> Result<()> {
+                //     match this_clone.execute_single_thread(thread_num) {
+                //         Ok(_) => Ok(()),
+                //         Err(cause) => Err(cause),
+                //     }
+                // });
+
+                // // finish the pipeline executor when has error or panic
+                // if let Err(cause) = try_result.flatten() {
+                //     // this.finish(Some(cause));
+                // }
+
+                Ok(())
+            }));
+        }
+        thread_join_handles
+    }
+
+    unsafe fn execute_single_thread(&self, thread_num: usize) -> Result<()> {
+        Ok(())
+    }
+
+    fn complete_pipeline(&self) {}
+
+    fn finish_task(&self) -> Result<()> {
+        Ok(())
+    }
+}

src/executor/executor_graph.rs src/execution/executor_graph.rssrc/executor/executor_graph.rs renamed to src/execution/executor_graph.rs

@@ -7,6 +7,7 @@ use petgraph::{
 };
 
 use super::{
+    executor::ExecutionQueue,
     parallel::{
         meta_pipeline::MetaPipeline,
         pipeline::Pipeline,
@@ -16,28 +17,6 @@ use super::{
 };
 use anyhow::Result;
 
-enum State {
-    Idle,
-    Processing,
-    Finished,
-}
-
-pub struct PipelineNode {
-    state: std::sync::Mutex<State>,
-    pipeline: Pipeline,
-}
-
-impl PipelineNode {
-    pub fn create(pipeline: &Pipeline) -> Arc<PipelineNode> {
-        Arc::new(PipelineNode {
-            state: std::sync::Mutex::new(State::Idle),
-            pipeline: pipeline.clone(),
-        })
-    }
-}
-
-type StateLockGuard = ExecutorGraph;
-
 pub struct ExecutorGraph {
     graph: StableGraph<Arc<PipelineEvent>, ()>,
 }
@@ -219,12 +198,15 @@ impl ExecutorGraph {
         Ok(())
     }
 
-    pub fn init_schedule_queue(&self, threads_num: usize) -> Result<()> {
+    pub async fn init_schedule_event_queue(&self, global_execution_queue: &mut ExecutionQueue) {
         // Schedule source node.
         for node_idx in self.graph.externals(Direction::Outgoing) {
             let node = self.get_node_by_index(node_idx);
+            // todo: fix deref pointer.
+            let pipeline_event =
+                unsafe { &mut *(node.as_ref() as *const PipelineEvent as *mut PipelineEvent) };
+            pipeline_event.schedule(global_execution_queue).await;
         }
-        Ok(())
     }
 
     fn get_node_by_index(&self, index: NodeIndex) -> Arc<PipelineEvent> {

src/execution/executor_task.rs

@@ -0,0 +1,97 @@
+use std::{
+    pin::Pin,
+    sync::Arc,
+    task::{Context, Poll},
+};
+
+use anyhow::anyhow;
+use anyhow::Result;
+use futures::{future::BoxFuture, Future};
+
+pub enum TaskExecutionMode {
+    Complete,
+    Partial,
+}
+
+pub enum TaskExecutionResult {
+    Finished,
+    NotFinished,
+    Error,
+    Blocked,
+}
+
+#[async_trait::async_trait]
+pub trait Task {
+    /// The name of task.
+    fn name(&self) -> String;
+
+    /// Execute the task in the specified execution mode
+    /// * If mode is Complete, Execute should always finish processing and
+    ///   return Finished
+    /// * If mode is Partial, Execute can return not_finished, in which case
+    ///   Execute will be called again
+    /// * In case of an error, error is returned
+    /// * In case the task has interrupted, blocked is returned.
+    async fn execute(&mut self, mode: TaskExecutionMode) -> Result<TaskExecutionResult>;
+}
+
+pub struct ExecutionTask {
+    inner: BoxFuture<'static, Result<()>>,
+}
+
+impl ExecutionTask {
+    // pub fn create<Inner>(inner: Inner) -> ExecutionTask
+    // where
+    //     Inner: Future<Output = Result<()>> + Send + 'static,
+    // {
+    //     ExecutionTask {
+    //         inner: inner.boxed(),
+    //     }
+    // }
+
+    pub fn create(inner: Arc<dyn Task>) -> ExecutionTask {
+        todo!()
+    }
+
+    pub fn empty() -> ExecutionTask {
+        todo!()
+    }
+}
+
+impl Future for ExecutionTask {
+    type Output = ();
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let inner = self.inner.as_mut();
+        let try_result =
+            std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || -> Poll<Result<()>> {
+                inner.poll(cx)
+            }));
+
+        match try_result {
+            Ok(Poll::Pending) => Poll::Pending,
+            Ok(Poll::Ready(_res)) => {
+                // self.queue.completed_async_task(
+                //     self.workers_condvar.clone(),
+                //     CompletedAsyncTask::create(self.processor_id, self.worker_id, res),
+                // );
+                Poll::Ready(())
+            }
+            Err(cause) => {
+                let _res: Result<()> = match cause.downcast_ref::<&'static str>() {
+                    Some(msg) => Err(anyhow!(msg.to_string())),
+                    None => match cause.downcast_ref::<String>() {
+                        Some(msg) => Err(anyhow!(msg.to_string())),
+                        None => Err(anyhow!("unknown panic message".to_string())),
+                    },
+                };
+
+                // self.queue.completed_async_task(
+                //     self.workers_condvar.clone(),
+                //     CompletedAsyncTask::create(self.processor_id, self.worker_id, res),
+                // );
+                Poll::Ready(())
+            }
+        }
+    }
+}

src/execution/mod.rs

@@ -0,0 +1,6 @@
+pub mod executor;
+pub mod executor_graph;
+pub mod executor_task;
+pub mod parallel;
+pub mod physical;
+pub mod runtime;

src/executor/parallel/meta_pipeline.rs src/execution/parallel/meta_pipeline.rssrc/executor/parallel/meta_pipeline.rs renamed to src/execution/parallel/meta_pipeline.rs

@@ -1,12 +1,9 @@
 use std::collections::HashMap;
 
-use crate::executor::{
-    physical::{PhysicalOperator, PhysicalOperatorRef},
-    Executor,
-};
+use crate::execution::executor::Executor;
+use crate::execution::physical::{PhysicalOperator, PhysicalOperatorRef};
 
 use super::pipeline::Pipeline;
-use anyhow::anyhow;
 use anyhow::Result;
 
 pub type PipelineIx = usize;

src/execution/parallel/mod.rs

@@ -0,0 +1,8 @@
+pub mod meta_pipeline;
+pub mod pipeline;
+pub mod pipeline_builder;
+pub mod pipeline_complete_event;
+pub mod pipeline_event;
+pub mod pipeline_finish_event;
+pub mod pipeline_initialize_event;
+pub mod pipeline_running_event;