changed to use async for all outpures.

Author: Rickard Lundin

src/chunked/arrow_converter.rs

@@ -64,9 +64,11 @@ use rayon::join;
 use std::sync::atomic::{AtomicUsize, Ordering, ATOMIC_USIZE_INIT};
 use std::sync::mpsc::{sync_channel, Receiver, RecvError, SyncSender};
 use std::thread;
-use std::thread::JoinHandle;
+use tokio::task::JoinHandle;
 use std::time::{Duration, Instant};
 use thread::spawn;
+use tokio::runtime;
+use tokio::runtime::Runtime;
 use crate::schema::FixedSchema;
 //use ordered_channel::Sender;
 //use crossbeam::channel::{Receiver, Sender};
@@ -237,18 +239,19 @@ impl<'a> Converter<'a> for Slice2Arrow<'a> {
         (bytes_in, bytes_out, parse_duration, builder_write_duration)
     }
 
-    fn setup(&mut self) -> (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(),
             self.masterbuilders.schema_factory(),
             self.masterbuilders.outfile.clone(),
+            rt
         );
         self.masterbuilders.sender = Some(o.0.clone());
         o
     }
-
-    fn shutdown(&mut self) {
+    
+    fn shutdown(&mut self,jh:JoinHandle<Result<Stats>>) {
         //        converter.shutdown();
         let schema = Schema::new(vec![Field::new(
             "id",
@@ -259,6 +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);
+        
+        
     }
 }
 

src/chunked/mod.rs

@@ -34,7 +34,8 @@ use ordered_channel::Sender;
 use std::fs;
 use std::path::PathBuf;
 use std::sync::mpsc::SyncSender;
-use std::thread::JoinHandle;
+use tokio::task::JoinHandle;
+use tokio::runtime::Runtime;
 use std::time::Duration;
 
 use self::residual_slicer::SLICER_IN_CHUNK_SIZE;
@@ -162,8 +163,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) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>);
-    fn shutdown(&mut self);
+    fn setup(&mut self, rt: Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>);
+    fn shutdown(&mut self,jh:JoinHandle<Result<Stats>>);
 }
 
 pub trait ColumnBuilder {
@@ -178,6 +179,6 @@ pub trait RecordBatchOutput {
         schema: SchemaRef,
         fixed_schema: FixedSchema,
         outfile: PathBuf,
-        
+        rt: tokio::runtime::Runtime
     ) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>);
 }

src/chunked/recordbatch_output.rs

@@ -24,7 +24,7 @@
 // File created: 2024-05-07
 // Last updated: 2024-05-25
 //
-
+use tokio::runtime::Runtime;
 use std::env::VarError;
 use crate::chunked::trimmer::{trimmer_factory, ColumnTrimmer};
 use arrow::array::{ArrayRef, BooleanBuilder, Int32Builder, Int64Builder, StringBuilder};
@@ -56,14 +56,15 @@ use arrow_ipc::writer::IpcWriteOptions;
 use arrow_ipc::CompressionType;
 use atomic_counter::{AtomicCounter, ConsistentCounter};
 use crossbeam::atomic::AtomicConsume;
-use libc::bsearch;
+use libc::{bsearch, send};
 use ordered_channel::Sender;
 use parquet::errors::{ParquetError, Result};
 use parquet::file::metadata::FileMetaData;
 use std::sync::atomic::{AtomicUsize, Ordering, ATOMIC_USIZE_INIT};
 use std::sync::mpsc::{sync_channel, Receiver, RecvError, SyncSender};
 use std::thread;
-use std::thread::JoinHandle;
+//use std::thread::JoinHandle;
+use tokio::task::JoinHandle;
 use std::time::{Duration, Instant};
 use thread::spawn;
 use Compression::SNAPPY;
@@ -88,6 +89,7 @@ pub(crate) fn output_factory(
     fixed_schema: FixedSchema,
     schema: SchemaRef,
     _outfile: PathBuf,
+    rt:tokio::runtime::Runtime
 ) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
     let mut pfo: Box<dyn RecordBatchOutput> =match target {
         Targets::Parquet => {
@@ -114,7 +116,7 @@ pub(crate) fn output_factory(
 
     };
 
-    pfo.setup(schema,fixed_schema, _outfile)
+    pfo.setup(schema,fixed_schema, _outfile,rt)
 
 }
 
@@ -158,23 +160,22 @@ impl DeltaOut {
 
     todo!()
     }
-    async fn myDelta(schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf) {
-        let dout = Self::deltasetup(fixed_schema);
+    async fn myDelta(schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf)->(Result<Stats>) {
+        Self::deltasetup(fixed_schema).await.unwrap();
+        todo!()
+        
     }
 
 }
 
 
 impl RecordBatchOutput for DeltaOut {
 
-    fn setup(&mut self, schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
-        let runtime = Builder::new_multi_thread()
-            .worker_threads(1)
-            .enable_all()
-            .build()
-            .unwrap();
+    fn setup(&mut self, schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf,rt:tokio::runtime::Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
 
-        runtime.spawn(Self::myDelta(schema,fixed_schema,outfile));
+        let j:JoinHandle<Result<Stats>>=rt.spawn(Self::myDelta(schema,fixed_schema,outfile));
+        
+        
 
 //        let dout = Self::deltasetup(fixed_schema);
 
@@ -185,7 +186,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) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
+    fn setup(&mut self, schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf,rt: tokio::runtime::Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
         todo!()
     }
 }
@@ -194,7 +195,7 @@ pub(crate) struct FlightOut {
 }
 
 impl RecordBatchOutput for FlightOut {
-    fn setup(&mut self, schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
+    fn setup(&mut self, schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf,rt: tokio::runtime::Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
         todo!()
     }
 }
@@ -203,8 +204,9 @@ impl RecordBatchOutput for FlightOut {
 pub(crate) struct ParquetFileOut {
     pub(crate) sender: Option<Sender<RecordBatch>>,
 }
-impl RecordBatchOutput for ParquetFileOut {
-    fn setup(&mut self, schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
+impl ParquetFileOut {
+    pub(crate) async fn myParquet(schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf,rt:tokio::runtime::Runtime)->(Result<Stats>) {
+//        Self::deltasetup(fixed_schema).await.unwrap();
         let _out_file = fs::OpenOptions::new()
             .create(true)
             .append(true)
@@ -218,7 +220,8 @@ impl RecordBatchOutput for ParquetFileOut {
 
         let (sender, mut receiver) = bounded::<RecordBatch>(100);
 
-        let t: JoinHandle<Result<Stats>> = thread::spawn(move || {
+        self.sender = Some(sender.clone());
+
             'outer: loop {
                 let mut message = receiver.recv();
 
@@ -245,70 +248,29 @@ impl RecordBatchOutput for ParquetFileOut {
                 parse_duration: Default::default(),
                 builder_write_duration: Default::default(),
             })
-        });
-        self.sender = Some(sender.clone());
-        (sender, t)
-    }
-}
 
-pub struct IpcFileOut {
-    pub(crate) sender: Option<Sender<RecordBatch>>,
+
+    }
 }
+impl RecordBatchOutput for ParquetFileOut {
+    fn setup(&mut self, schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf,rt: tokio::runtime::Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
 
+        let j:JoinHandle<Result<Stats>>=rt.spawn(Self::myParquet(schema,fixed_schema,outfile,rt));
 
-impl RecordBatchOutput for IpcFileOut {
-    fn setup(&mut self, schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
-        let _out_file = fs::OpenOptions::new()
-            .create(true)
-            .append(true)
-            .open(outfile)
-            .expect("aaa");
+//        let j:JoinHandle<Result<Stats>>=rt.spawn(Self::myDelta(schema,fixed_schema,outfile));
 
-        let p = IpcWriteOptions::try_with_compression(
-            Default::default(),
-            Some(CompressionType::LZ4_FRAME),
-        );
+        (self.sender.as_mut().cloned().unwrap(), j)
 
-        let mut writer = arrow_ipc::writer::FileWriter::try_new_with_options(
-            _out_file,
-            &schema,
-            Default::default(),
-        )
-        .expect("TODO: panic message");
+    }
+}
 
-        let props = WriterProperties::builder().set_compression(SNAPPY).build();
+pub struct IpcFileOut {
+    pub(crate) sender: Option<Sender<RecordBatch>>,
+}
 
-        let (sender, mut receiver) = bounded::<RecordBatch>(1000);
 
-        let t: JoinHandle<Result<Stats>> = thread::spawn(move || {
-            'outer: loop {
-                let mut message = receiver.recv();
-
-                match message {
-                    Ok(rb) => {
-                        writer.write(&rb).expect("Error Writing batch");
-                        if (rb.num_rows() == 0) {
-                            break 'outer;
-                        }
-                    }
-                    Err(e) => {
-                        info!("got RecvError in channel , break to outer");
-                        break 'outer;
-                    }
-                }
-            }
-            info!("closing the writer for parquet");
-            writer.finish();
-            Ok(Stats {
-                bytes_in: 0,
-                bytes_out: 0,
-                num_rows: 0,
-                read_duration: Default::default(),
-                parse_duration: Default::default(),
-                builder_write_duration: Default::default(),
-            })
-        });
-        self.sender = Some(sender.clone());
-        (sender, t)
+impl RecordBatchOutput for IpcFileOut {
+    fn setup(&mut self, schema: SchemaRef,fixed_schema: FixedSchema, outfile: PathBuf,rt: tokio::runtime::Runtime) -> (Sender<RecordBatch>, JoinHandle<Result<Stats>>) {
+        todo!()
     }
 }

src/chunked/residual_slicer.rs

@@ -37,6 +37,7 @@ use std::io::{BufReader, Read};
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
+use tokio::runtime::Runtime;
 
 use super::{ChunkAndResidue, Converter, FnFindLastLineBreak, IterRevolver, Slicer, Stats};
 
@@ -88,8 +89,8 @@ impl<'a> Slicer<'a> for ResidualSlicer<'a> {
             .stack_size(((SLICER_IN_CHUNK_SIZE as f32) * 2f32) as usize)
             .build_global()
             .unwrap();
-
-        let threaded_writer = converter.setup();
+        let rt = Runtime::new().unwrap();
+        let threaded_writer = converter.setup(rt);
 
         loop {
             let cr = ir.next().unwrap();
@@ -140,8 +141,8 @@ impl<'a> Slicer<'a> for ResidualSlicer<'a> {
         }
         info!("about to shudown converter...");
 
-        converter.shutdown();
-        threaded_writer.1.join();
+        converter.shutdown(threaded_writer.1);
+//        threaded_writer.1.await.expect("The task being joined has panicked");;
         info!("converter has been shutdown");
 
         info!(

src/chunked/self_converter.rs

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