Removed all unsafe

benches/mandelbrot_pixbypix.rs

@@ -179,13 +179,17 @@ fn criterion_benchmark(c: &mut Criterion) {
             &format!("rayon {threads} worker threads"),
             &threads,
             |b, &threads| {
-                let pool = Rc::new(ThreadPoolBuilder::new().num_threads(threads as usize).build().unwrap());
-                b.iter(|| mandelbrot_rayon(1000, pool.clone())); 
+                let pool = Rc::new(
+                    ThreadPoolBuilder::new()
+                        .num_threads(threads as usize)
+                        .build()
+                        .unwrap(),
+                );
+                b.iter(|| mandelbrot_rayon(1000, pool.clone()));
             },
         );
     }
     group.finish();
-
 }
 
 criterion_group!(benches, criterion_benchmark);

benches/mandelbrot_rustspp.rs

@@ -95,7 +95,7 @@ fn mandelbrot_rustspp(size: usize, threads: usize) {
         pipeline.post(i as usize).unwrap();
     }
     let rendered_image = pipeline.collect();
-    
+
     let mut bytes = 0usize;
     for line in rendered_image {
         bytes += line.line_buffer.len()
@@ -132,7 +132,8 @@ async fn mandelbrot_tokio(size: usize, threads: usize) {
             receiver
         })
         .buffered(threads)
-        .for_each(async move |_rendered_line| {}).await;
+        .for_each(async move |_rendered_line| {})
+        .await;
 }
 
 fn mandelbrot_rayon(size: usize, thread_pool: Rc<rayon::ThreadPool>) -> Vec<ImageLine> {
@@ -158,7 +159,8 @@ async fn mandelbrot_tokio_unordered(size: usize, threads: usize) {
             receiver
         })
         .buffer_unordered(threads)
-        .for_each(async move |_rendered_line| {}).await;
+        .for_each(async move |_rendered_line| {})
+        .await;
 }
 
 fn mandelbrot_benches(c: &mut Criterion) {
@@ -186,7 +188,12 @@ fn mandelbrot_benches(c: &mut Criterion) {
             &format!("rayon {threads} worker threads"),
             &threads,
             |b, &threads| {
-                let pool = Rc::new(ThreadPoolBuilder::new().num_threads(threads as usize).build().unwrap());
+                let pool = Rc::new(
+                    ThreadPoolBuilder::new()
+                        .num_threads(threads as usize)
+                        .build()
+                        .unwrap(),
+                );
                 b.iter(|| mandelbrot_rayon(1000, pool.clone()));
             },
         );
@@ -195,19 +202,18 @@ fn mandelbrot_benches(c: &mut Criterion) {
             &format!("mandelbrot tokio ordered {threads} worker threads"),
             &threads,
             |b, &threads| {
-               b.iter(|| mandelbrot_tokio(1000, threads));
+                b.iter(|| mandelbrot_tokio(1000, threads));
             },
         );
 
         group.bench_with_input(
             &format!("mandelbrot tokio unordered {threads} worker threads"),
             &threads,
             |b, &threads| {
-               b.iter(|| mandelbrot_tokio_unordered(1000, threads));
+                b.iter(|| mandelbrot_tokio_unordered(1000, threads));
             },
         );
     }
-
 }
 //criterion_group!(benches, criterion_benchmark);
 criterion_group!(benches, mandelbrot_benches);

src/blocks/in_block.rs

@@ -77,7 +77,7 @@ impl<TInput, TCollected, TFactory> PipelineBlock<TInput, TCollected>
 }
 
 impl<
-        TInput: 'static,
+        TInput: 'static + Send,
         TCollected: 'static + Send,
         THandler: In<TInput, TCollected> + Send + 'static,
         TFactory: FnMut() -> THandler,

src/blocks/inout_block.rs

@@ -80,12 +80,12 @@ impl<
 }
 
 impl<
-        TInput: 'static,
-        TOutput: 'static,
-        TCollected: 'static,
-        TStage: InOut<TInput, TOutput> + Send + 'static,
-        TFactory: FnMut() -> TStage,
-        TNextStep: PipelineBlock<TOutput, TCollected> + Send + Sync + 'static,
+    TInput: 'static + Send,
+    TOutput: 'static,
+    TCollected: 'static,
+    TStage: InOut<TInput, TOutput> + Send + 'static,
+    TFactory: FnMut() -> TStage,
+    TNextStep: PipelineBlock<TOutput, TCollected> + Send + Sync + 'static,
     > InOutBlock<TInput, TOutput, TCollected, TStage, TFactory, TNextStep>
 {
     pub fn new(

src/main.rs

@@ -29,7 +29,6 @@ struct SyncNotNeeded {
     test: std::rc::Rc<i64>,
 }
 
-//unsafe impl Send for SyncNotNeeded {}
 /*
 fn send_sync_notneeded() {
     let pipeline = pipeline![

src/work_storage/blocking_ordered_set.rs

@@ -39,6 +39,3 @@ impl<T> BlockingOrderedSet<T> {
         }
     }
 }
-
-unsafe impl<T> Send for BlockingOrderedSet<T> {}
-unsafe impl<T> Sync for BlockingOrderedSet<T> {}

src/work_storage/blocking_queue.rs

@@ -59,6 +59,3 @@ impl<T> BlockingQueue<T> {
         popped.unwrap()
     }
 }
-
-unsafe impl<T> Send for BlockingQueue<T> {}
-unsafe impl<T> Sync for BlockingQueue<T> {}