refactor

asm2025 · Apr 8, 2024 · 44a938d · 44a938d
1 parent 0070d35
commit 44a938d
Show file tree

Hide file tree

Showing 8 changed files with 6,693 additions and 42 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -54,4 +54,4 @@ run = "cargo run"
 check = "cargo check"
 build = "cargo build"
 release = "cargo build --release"
-features = "cargo tree --features"
+features = "cargo tree --features" 
diff --git a/src/lib.rs b/src/lib.rs
@@ -15,3 +15,7 @@ pub mod web;
 pub use self::app::*;
 
 mod app;
+
+pub(crate) trait ThreadSafe: Send + Sync {}
+pub(crate) trait ThreadClonable: ThreadSafe + Clone {}
+pub(crate) trait ThreadStatic: ThreadClonable + 'static {}
diff --git a/src/threading/consumer.rs b/src/threading/consumer.rs
@@ -13,6 +13,7 @@ use tokio::{
 };
 
 use super::{cond::Mutcond, *};
+use crate::{ThreadClonable, ThreadStatic};
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct ConsumerOptions {
@@ -72,7 +73,7 @@ impl ConsumerOptions {
 }
 
 #[derive(Clone, Debug)]
-pub struct Consumer<T: Send + Sync + Clone + 'static> {
+pub struct Consumer<T: ThreadStatic> {
     options: ConsumerOptions,
     items: Arc<Mutex<LinkedList<T>>>,
     items_cond: Arc<Mutcond>,
@@ -87,7 +88,7 @@ pub struct Consumer<T: Send + Sync + Clone + 'static> {
     running: Arc<AtomicUsize>,
 }
 
-impl<T: Send + Sync + Clone> Consumer<T> {
+impl<T: ThreadClonable> Consumer<T> {
     pub fn new() -> Self {
         Consumer {
             options: Default::default(),
@@ -207,10 +208,7 @@ impl<T: Send + Sync + Clone> Consumer<T> {
         self.running.fetch_sub(1, Ordering::SeqCst);
     }
 
-    pub fn start<TD: TaskDelegation<Consumer<T>, T> + Send + Sync + Clone + 'static>(
-        &self,
-        delegate: &TD,
-    ) {
+    pub fn start<TD: TaskDelegation<Consumer<T>, T> + ThreadStatic>(&self, delegate: &TD) {
         if self.is_cancelled() {
             panic!("Queue is already cancelled.")
         }
@@ -331,9 +329,7 @@ impl<T: Send + Sync + Clone> Consumer<T> {
         }
     }
 
-    pub async fn start_async<
-        TD: AsyncTaskDelegation<Consumer<T>, T> + Send + Sync + Clone + 'static,
-    >(
+    pub async fn start_async<TD: AsyncTaskDelegation<Consumer<T>, T> + ThreadStatic>(
         &self,
         delegate: &TD,
     ) {
@@ -571,7 +567,7 @@ impl<T: Send + Sync + Clone> Consumer<T> {
     }
 }
 
-impl<T: Send + Sync + Clone> AwaitableConsumer for Consumer<T> {
+impl<T: ThreadClonable> AwaitableConsumer for Consumer<T> {
     fn is_cancelled(&self) -> bool {
         Consumer::is_cancelled(self)
     }

diff --git a/src/threading/injector_consumer.rs b/src/threading/injector_consumer.rs
@@ -13,6 +13,7 @@ use tokio::{
 };
 
 use super::{cond::Mutcond, *};
+use crate::{ThreadClonable, ThreadStatic};
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct InjectorWorkerOptions {
@@ -70,7 +71,7 @@ impl InjectorWorkerOptions {
 }
 
 #[derive(Debug, Clone)]
-pub struct InjectorWorker<T: Send + Sync + Clone + 'static> {
+pub struct InjectorWorker<T: ThreadStatic> {
     options: InjectorWorkerOptions,
     injector: Arc<Injector<T>>,
     stealers: Arc<Mutex<Vec<Stealer<T>>>>,
@@ -85,7 +86,7 @@ pub struct InjectorWorker<T: Send + Sync + Clone + 'static> {
     running: Arc<AtomicUsize>,
 }
 
-impl<T: Send + Sync + Clone> InjectorWorker<T> {
+impl<T: ThreadClonable> InjectorWorker<T> {
     pub fn new() -> Self {
         let options: InjectorWorkerOptions = Default::default();
         InjectorWorker {
@@ -206,10 +207,7 @@ impl<T: Send + Sync + Clone> InjectorWorker<T> {
         self.running.fetch_sub(1, Ordering::SeqCst);
     }
 
-    pub fn start<TD: TaskDelegation<InjectorWorker<T>, T> + Send + Sync + Clone + 'static>(
-        &self,
-        delegate: &TD,
-    ) {
+    pub fn start<TD: TaskDelegation<InjectorWorker<T>, T> + ThreadStatic>(&self, delegate: &TD) {
         if self.is_cancelled() {
             panic!("Queue is already cancelled.")
         }
@@ -317,9 +315,7 @@ impl<T: Send + Sync + Clone> InjectorWorker<T> {
         }
     }
 
-    pub async fn start_async<
-        TD: AsyncTaskDelegation<InjectorWorker<T>, T> + Send + Sync + Clone + 'static,
-    >(
+    pub async fn start_async<TD: AsyncTaskDelegation<InjectorWorker<T>, T> + ThreadStatic>(
         &self,
         delegate: &TD,
     ) {
@@ -523,7 +519,7 @@ impl<T: Send + Sync + Clone> InjectorWorker<T> {
     }
 }
 
-impl<T: Send + Sync + Clone> AwaitableConsumer for InjectorWorker<T> {
+impl<T: ThreadClonable> AwaitableConsumer for InjectorWorker<T> {
     fn is_cancelled(&self) -> bool {
         self.is_cancelled()
     }

diff --git a/src/threading/mod.rs b/src/threading/mod.rs
@@ -14,6 +14,7 @@ pub use self::parallel_consumer::*;
 pub use self::producer_consumer::*;
 
 use super::error;
+use crate::{ThreadClonable, ThreadStatic};
 
 mod cond;
 mod consumer;
@@ -74,20 +75,18 @@ impl fmt::Display for QueueBehavior {
     }
 }
 
-pub trait TaskDelegationBase<TD: Send + Clone + 'static, T: Send + Clone + 'static> {
+pub trait TaskDelegationBase<TD: ThreadStatic, T: ThreadStatic> {
     fn on_started(&self, td: &TD);
     fn on_completed(&self, td: &TD, item: &T, result: &TaskResult) -> bool;
     fn on_cancelled(&self, td: &TD);
     fn on_finished(&self, td: &TD);
 }
 
-pub trait TaskDelegation<TD: Send + Clone + 'static, T: Send + Clone + 'static>:
-    TaskDelegationBase<TD, T>
-{
+pub trait TaskDelegation<TD: ThreadStatic, T: ThreadStatic>: TaskDelegationBase<TD, T> {
     fn process(&self, td: &TD, item: &T) -> Result<TaskResult>;
 }
 
-pub trait AsyncTaskDelegation<TD: Send + Clone + 'static, T: Send + Clone + 'static>:
+pub trait AsyncTaskDelegation<TD: ThreadStatic, T: ThreadStatic>:
     TaskDelegationBase<TD, T>
 {
     fn process(&self, td: &TD, item: &T) -> impl Future<Output = Result<TaskResult>> + Send;

diff --git a/src/threading/parallel_consumer.rs b/src/threading/parallel_consumer.rs
@@ -112,7 +112,7 @@ impl ParallelOptions {
 }
 
 #[derive(Clone, Debug)]
-pub struct Parallel<T: Send + Sync + Clone + 'static> {
+pub struct Parallel<T: ThreadStatic> {
     options: ParallelOptions,
     started: Arc<Mutex<bool>>,
     finished: Arc<AtomicBool>,
@@ -124,7 +124,7 @@ pub struct Parallel<T: Send + Sync + Clone + 'static> {
     _marker: PhantomData<T>,
 }
 
-impl<T: Send + Sync + Clone> Parallel<T> {
+impl<T: ThreadClonable> Parallel<T> {
     pub fn new() -> Self {
         let options: ParallelOptions = Default::default();
         Parallel {
@@ -212,7 +212,7 @@ impl<T: Send + Sync + Clone> Parallel<T> {
 
     pub fn start<
         I: IntoParallelIterator<Item = T> + Len + Send + 'static,
-        TD: TaskDelegation<Parallel<T>, T> + Send + Sync + Clone + 'static,
+        TD: TaskDelegation<Parallel<T>, T> + ThreadStatic,
     >(
         &self,
         collection: I,
@@ -334,7 +334,7 @@ impl<T: Send + Sync + Clone> Parallel<T> {
     }
 }
 
-impl<T: Send + Sync + Clone> AwaitableConsumer for Parallel<T> {
+impl<T: ThreadClonable> AwaitableConsumer for Parallel<T> {
     fn is_cancelled(&self) -> bool {
         Parallel::is_cancelled(self)
     }

diff --git a/src/threading/producer_consumer.rs b/src/threading/producer_consumer.rs
@@ -72,12 +72,12 @@ impl ProducerConsumerOptions {
 }
 
 #[derive(Clone, Debug)]
-pub struct Producer<T: Send + Sync + Clone + 'static> {
+pub struct Producer<T: ThreadStatic> {
     pc: Arc<ProducerConsumer<T>>,
     sender: Arc<channel::Sender<T>>,
 }
 
-impl<T: Send + Sync + Clone> Producer<T> {
+impl<T: ThreadClonable> Producer<T> {
     fn new(pc: &ProducerConsumer<T>, sender: &channel::Sender<T>) -> Self {
         Producer {
             pc: Arc::new(pc.clone()),
@@ -105,7 +105,7 @@ impl<T: Send + Sync + Clone> Producer<T> {
 }
 
 #[derive(Clone, Debug)]
-pub struct ProducerConsumer<T: Send + Sync + Clone + 'static> {
+pub struct ProducerConsumer<T: ThreadStatic> {
     options: ProducerConsumerOptions,
     started: Arc<Mutex<bool>>,
     finished: Arc<AtomicBool>,
@@ -120,7 +120,7 @@ pub struct ProducerConsumer<T: Send + Sync + Clone + 'static> {
     receiver: channel::Receiver<T>,
 }
 
-impl<T: Send + Sync + Clone> ProducerConsumer<T> {
+impl<T: ThreadClonable> ProducerConsumer<T> {
     pub fn new() -> Self {
         let options: ProducerConsumerOptions = Default::default();
         let (sender, receiver) = channel::bounded::<T>(options.capacity);
@@ -255,10 +255,7 @@ impl<T: Send + Sync + Clone> ProducerConsumer<T> {
         Producer::new(self, &self.sender)
     }
 
-    pub fn start<TD: TaskDelegation<ProducerConsumer<T>, T> + Send + Sync + Clone + 'static>(
-        &self,
-        delegate: &TD,
-    ) {
+    pub fn start<TD: TaskDelegation<ProducerConsumer<T>, T> + ThreadStatic>(&self, delegate: &TD) {
         if self.is_cancelled() {
             panic!("Queue is already cancelled.")
         }
@@ -375,9 +372,7 @@ impl<T: Send + Sync + Clone> ProducerConsumer<T> {
         }
     }
 
-    pub async fn start_async<
-        TD: AsyncTaskDelegation<ProducerConsumer<T>, T> + Send + Sync + Clone + 'static,
-    >(
+    pub async fn start_async<TD: AsyncTaskDelegation<ProducerConsumer<T>, T> + ThreadStatic>(
         &self,
         delegate: &TD,
     ) {
@@ -538,7 +533,7 @@ impl<T: Send + Sync + Clone> ProducerConsumer<T> {
     }
 }
 
-impl<T: Send + Sync + Clone> AwaitableConsumer for ProducerConsumer<T> {
+impl<T: ThreadClonable> AwaitableConsumer for ProducerConsumer<T> {
     fn is_cancelled(&self) -> bool {
         ProducerConsumer::is_cancelled(self)
     }