compiles and everthing is Tokio

Author: Rickard Lundin

src/chunked/arrow_converter.rs

@@ -239,7 +239,7 @@ impl<'a> Converter<'a> for Slice2Arrow<'a> {
         (bytes_in, bytes_out, parse_duration, builder_write_duration)
     }
 
-    fn setup(&mut self, rt: runtime::Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
+    fn setup(&mut self, rt: &runtime::Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
         let o = output_factory(
             self.target.clone(),
             self.fixed_schema.clone(),
@@ -251,7 +251,7 @@ impl<'a> Converter<'a> for Slice2Arrow<'a> {
         o
     }
 
-    fn shutdown(&mut self,jh:JoinHandle<Result<Stats>>) {
+    fn shutdown(&mut self,rt: &runtime::Runtime,jh:JoinHandle<Result<Stats>>) {
         //        converter.shutdown();
         let schema = Schema::new(vec![Field::new(
             "id",
@@ -262,7 +262,8 @@ impl<'a> Converter<'a> for Slice2Arrow<'a> {
         let emptyrb = arrow::record_batch::RecordBatch::new_empty(Arc::new(schema));
         let c = self.consistent_counter.get();
         let _ = &self.masterbuilders.sender.clone().unwrap().send(c, emptyrb);
-
+        rt.spawn_blocking(|| async { });
+        
         //     jh.spawn_blocking(async move {threaded_writer.1.await;});
 //        threaded_writer.1.await.expect("The task being joined has panicked");;
 

src/chunked/mod.rs

@@ -40,6 +40,7 @@ use std::time::Duration;
 
 use self::residual_slicer::SLICER_IN_CHUNK_SIZE;
 use parquet::errors::{ParquetError, Result};
+use tokio::runtime;
 use crate::schema::FixedSchema;
 
 pub(crate) mod arrow_converter;
@@ -163,8 +164,8 @@ pub(crate) trait Converter<'a> {
 
     //    fn process(& mut self, slices: Vec< &'a[u8]>) -> usize;
     fn process(&mut self, slices: Vec<&'a [u8]>) -> (usize, usize, Duration, Duration);
-    fn setup(&mut self, rt: Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>);
-    fn shutdown(&mut self,jh:JoinHandle<Result<Stats>>);
+    fn setup(&mut self, rt: &Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>);
+    fn shutdown(&mut self,rt: &Runtime,jh:JoinHandle<Result<Stats>>);
 }
 
 pub trait ColumnBuilder {
@@ -179,6 +180,6 @@ pub trait RecordBatchOutput {
         schema: SchemaRef,
         fixed_schema: FixedSchema,
         outfile: PathBuf,
-        rt: tokio::runtime::Runtime
+        rt: &Runtime
     ) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>);
 }

src/chunked/recordbatch_output.rs

@@ -91,7 +91,7 @@ pub(crate) fn output_factory(
     fixed_schema: FixedSchema,
     schema: SchemaRef,
     _outfile: PathBuf,
-    rt:tokio::runtime::Runtime
+    rt:& tokio::runtime::Runtime
 ) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
     let mut pfo: Box<dyn RecordBatchOutput> =match target {
         Targets::Parquet => {
@@ -173,7 +173,7 @@ impl DeltaOut {
 
 impl RecordBatchOutput for DeltaOut {
 
-    fn setup(&mut self, schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf,rt:tokio::runtime::Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
+    fn setup(&mut self, schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf,rt:&Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
 
         let j:JoinHandle<Result<Stats>>=rt.spawn(Self::myDelta(schema,fixed_schema,outfile));
 
@@ -188,7 +188,7 @@ pub(crate) struct IcebergOut {
     pub(crate) sender: Option<Sender<RecordBatch>>,
 }
 impl RecordBatchOutput for IcebergOut {
-    fn setup(&mut self, schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf,rt: tokio::runtime::Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
+    fn setup(&mut self, schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf,rt: &Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
         todo!()
     }
 }
@@ -197,7 +197,7 @@ pub(crate) struct FlightOut {
 }
 
 impl RecordBatchOutput for FlightOut {
-    fn setup(&mut self, schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf,rt: tokio::runtime::Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
+    fn setup(&mut self, schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf,rt: &Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
         todo!()
     }
 }
@@ -207,7 +207,7 @@ pub(crate) struct ParquetFileOut {
     pub(crate) sender: Option<Sender<RecordBatch>>,
 }
 impl ParquetFileOut {
-    pub(crate) async fn myParquet(sender:Sender<RecordBatch>,mut receiver:  Receiver<RecordBatch>, schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf)->(Result<Stats>) {
+    pub(crate) async fn myParquet(mut receiver:  Receiver<RecordBatch>, schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf)->(Result<Stats>) {
 //        Self::deltasetup(fixed_schema).await.unwrap();
         let _out_file = fs::OpenOptions::new()
             .create(true)
@@ -219,7 +219,7 @@ impl ParquetFileOut {
 
         let mut writer: ArrowWriter<File> =
             ArrowWriter::try_new(_out_file, schema, Some(props.clone())).unwrap();
-        
+
             'outer: loop {
                 let mut message = receiver.recv();
 
@@ -251,14 +251,12 @@ impl ParquetFileOut {
     }
 }
 impl RecordBatchOutput for ParquetFileOut {
-    fn setup(&mut self, schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf,rt: tokio::runtime::Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
+    fn setup(&mut self, schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf,rt: &Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
         let (sender, mut receiver) = bounded::<RecordBatch>(100);
 
         self.sender = Some(sender.clone());
 
-        let j:JoinHandle<Result<Stats>>=rt.spawn(Self::myParquet(sender,receiver,schema,fixed_schema,outfile));
-
-//        let j:JoinHandle<Result<Stats>>=rt.spawn(Self::myDelta(schema,fixed_schema,outfile));
+        let j:JoinHandle<Result<Stats>>=rt.spawn(Self::myParquet(receiver,schema,fixed_schema,outfile));
 
         (self.sender.as_mut().cloned().unwrap(), j)
 
@@ -271,7 +269,7 @@ pub struct IpcFileOut {
 
 
 impl RecordBatchOutput for IpcFileOut {
-    fn setup(&mut self, schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf,rt: tokio::runtime::Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
+    fn setup(&mut self, schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf,rt: &Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
         todo!()
     }
 }

src/chunked/residual_slicer.rs

@@ -90,7 +90,7 @@ impl<'a> Slicer<'a> for ResidualSlicer<'a> {
             .build_global()
             .unwrap();
         let rt = Runtime::new().unwrap();
-        let threaded_writer = converter.setup(rt);
+        let threaded_writer = converter.setup(&rt);
 
         loop {
             let cr = ir.next().unwrap();
@@ -141,7 +141,7 @@ impl<'a> Slicer<'a> for ResidualSlicer<'a> {
         }
         info!("about to shudown converter...");
 
-        converter.shutdown(threaded_writer.1);
+        converter.shutdown(&rt,threaded_writer.1);
 //        let hhh=threaded_writer.1;
 //        rt.spawn(async {
 //            hhh.await.unwrap();            

src/chunked/self_converter.rs

@@ -39,6 +39,7 @@ use std::io::Write;
 use std::sync::mpsc::SyncSender;
 use tokio::task::JoinHandle;
 use std::time::Duration;
+use tokio::runtime;
 use tokio::runtime::Runtime;
 
 use super::{Converter, FnFindLastLineBreak};
@@ -72,11 +73,11 @@ impl<'a> Converter<'a> for SampleSliceAggregator<'a> {
         (bytes_processed, 0, duration, duration)
     }
 
-    fn setup(&mut self, rt: Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
+    fn setup(&mut self, rt: &Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
         todo!()
     }
 
-    fn shutdown(&mut self,jh:JoinHandle<Result<Stats>>) {
+    fn shutdown(&mut self,rt: &runtime::Runtime,jh:JoinHandle<Result<Stats>>) {
         todo!()
     }
 }