Add generic type to IOLooper

Change-Id: I41579bd7d194dc9c577c6470c32dc3c15647b02c
Signed-off-by: kexuan.yang <kexuan.yang@gmail.com>

Author: yangkx1024

src/core/buffer.rs

@@ -13,7 +13,7 @@ include!(concat!(env!("OUT_DIR"), "/protos/mod.rs"));
 #[repr(transparent)]
 pub struct Buffer(KV);
 
-pub trait Encoder: Send {
+pub trait Encoder: Send + Sync {
     fn encode_to_bytes(&self, raw_buffer: &Buffer, position: u32) -> Result<Vec<u8>>;
 }
 
@@ -22,7 +22,7 @@ pub struct DecodeResult {
     pub len: u32,
 }
 
-pub trait Decoder: Send {
+pub trait Decoder: Send + Sync {
     fn decode_bytes(&self, data: &[u8], position: u32) -> Result<DecodeResult>;
 }
 

src/core/io_looper.rs

@@ -1,5 +1,6 @@
 use crate::Error::{IOError, LockError};
 use crate::Result;
+use std::any::Any;
 use std::collections::VecDeque;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::mpsc::{channel, Receiver, Sender};
@@ -9,47 +10,47 @@ use std::thread::JoinHandle;
 
 const LOG_TAG: &str = "MMKV:IO";
 
-type Job = Box<dyn FnOnce(&mut dyn std::any::Any) + Send + 'static>;
+type Job = Box<dyn FnOnce(&mut dyn Any) + Send + 'static>;
 
 enum Signal {
     Normal,
     Kill(Job),
 }
 
-pub trait Callback: Send {}
+pub trait Callback: Send + Any {}
 
-pub struct IOLooper {
+pub struct IOLooper<T> {
     sender: Option<Sender<Signal>>,
     executor: Executor,
+    _marker: std::marker::PhantomData<T>,
 }
 
 struct Executor {
     queue: Arc<Mutex<VecDeque<Job>>>,
     join_handle: Option<JoinHandle<()>>,
 }
 
-impl IOLooper {
-    pub fn new<T>(callback: T) -> Self
-    where
-        T: Callback + 'static,
-    {
+impl<T: Callback + 'static> IOLooper<T> {
+    pub fn new(callback: T) -> Self {
         let (sender, receiver) = channel::<Signal>();
         let executor = Executor::new(receiver, callback);
         IOLooper {
             sender: Some(sender),
             executor,
+            _marker: std::marker::PhantomData,
         }
     }
 
-    pub fn post_and_kill<F>(&mut self, task: F)
-    where
-        F: FnOnce(&mut dyn std::any::Any) + Send + 'static,
-    {
+    pub fn post_and_kill<F: FnOnce(&mut T) + Send + 'static>(&mut self, task: F) {
+        let job: Job = Box::new(|callback| {
+            let callback = callback.downcast_mut::<T>().unwrap();
+            task(callback)
+        });
         self.executor.queue.lock().unwrap().clear();
         self.sender
             .as_ref()
             .unwrap()
-            .send(Signal::Kill(Box::new(task)))
+            .send(Signal::Kill(job))
             .unwrap();
         drop(self.sender.take());
         if let Some(handle) = self.executor.join_handle.take() {
@@ -58,14 +59,15 @@ impl IOLooper {
         }
     }
 
-    pub fn post<F>(&self, task: F) -> Result<()>
-    where
-        F: FnOnce(&mut dyn std::any::Any) + Send + 'static,
-    {
+    pub fn post<F: FnOnce(&mut T) + Send + 'static>(&self, task: F) -> Result<()> {
+        let job: Job = Box::new(|callback| {
+            let callback = callback.downcast_mut::<T>().unwrap();
+            task(callback)
+        });
         self.executor
             .queue
             .lock()
-            .map(|mut queue| queue.push_back(Box::new(task)))
+            .map(|mut queue| queue.push_back(job))
             .map_err(|e| LockError(e.to_string()))?;
 
         self.sender
@@ -91,7 +93,7 @@ impl IOLooper {
     }
 }
 
-impl Drop for IOLooper {
+impl<T> Drop for IOLooper<T> {
     fn drop(&mut self) {
         drop(self.sender.take());
 
@@ -103,10 +105,7 @@ impl Drop for IOLooper {
 }
 
 impl Executor {
-    pub fn new<T>(receiver: Receiver<Signal>, mut callback: T) -> Self
-    where
-        T: Callback + std::any::Any + 'static,
-    {
+    pub fn new<T: Callback + 'static>(receiver: Receiver<Signal>, mut callback: T) -> Self {
         let queue: Arc<Mutex<VecDeque<Job>>> = Arc::new(Mutex::new(VecDeque::with_capacity(100)));
         let queue_clone = Arc::clone(&queue);
         let handle = thread::spawn(move || loop {
@@ -164,31 +163,20 @@ mod tests {
         io_looper
             .post(|callback| {
                 thread::sleep(Duration::from_millis(100));
-                callback
-                    .downcast_ref::<SimpleCallback>()
-                    .unwrap()
-                    .print("first job")
+                callback.print("first job")
             })
             .expect("failed to execute job");
         io_looper
             .post(|callback| {
                 thread::sleep(Duration::from_millis(100));
-                callback
-                    .downcast_ref::<SimpleCallback>()
-                    .unwrap()
-                    .print("second job")
+                callback.print("second job")
             })
             .expect("failed to execute job");
         assert!(io_looper.sender.is_some());
         assert_eq!(io_looper.executor.queue.lock().unwrap().len(), 2);
         assert!(io_looper.executor.join_handle.is_some());
         thread::sleep(Duration::from_millis(50));
-        io_looper.post_and_kill(|callback| {
-            callback
-                .downcast_ref::<SimpleCallback>()
-                .unwrap()
-                .print("last job")
-        });
+        io_looper.post_and_kill(|callback| callback.print("last job"));
         assert!(io_looper.sender.is_none());
         assert!(io_looper.executor.queue.lock().unwrap().is_empty());
         assert!(io_looper.executor.join_handle.is_none());

src/core/memory_map.rs

@@ -76,6 +76,8 @@ impl DerefMut for RawMmap {
 
 unsafe impl Send for RawMmap {}
 
+unsafe impl Sync for RawMmap {}
+
 #[derive(Debug)]
 #[repr(transparent)]
 pub struct MemoryMap(RawMmap);

src/core/mmkv_impl.rs

@@ -19,7 +19,7 @@ const LOG_TAG: &str = "MMKV:Core";
 pub struct MmkvImpl {
     kv_map: HashMap<String, Buffer>,
     is_valid: bool,
-    io_looper: Option<IOLooper>,
+    io_looper: Option<IOLooper<IOWriter>>,
     #[cfg(feature = "encryption")]
     encryptor: Encryptor,
 }
@@ -77,12 +77,10 @@ impl MmkvImpl {
         }
         let result = self.kv_map.insert(key.to_string(), raw_buffer.clone());
         let duplicated = result.is_some();
-        self.io_looper.as_ref().unwrap().post(move |callback| {
-            callback
-                .downcast_mut::<IOWriter>()
-                .unwrap()
-                .write(raw_buffer, duplicated)
-        })
+        self.io_looper
+            .as_ref()
+            .unwrap()
+            .post(move |writer| writer.write(raw_buffer, duplicated))
     }
 
     pub fn get(&self, key: &str) -> Result<&Buffer> {
@@ -104,12 +102,10 @@ impl MmkvImpl {
             return Ok(());
         }
         let buffer = Buffer::deleted_buffer(key);
-        self.io_looper.as_ref().unwrap().post(move |callback| {
-            callback
-                .downcast_mut::<IOWriter>()
-                .unwrap()
-                .write(buffer, true)
-        })
+        self.io_looper
+            .as_ref()
+            .unwrap()
+            .post(move |writer| writer.write(buffer, true))
     }
 
     pub fn clear_data(&mut self) {
@@ -121,8 +117,8 @@ impl MmkvImpl {
         self.kv_map.clear();
         #[cfg(feature = "encryption")]
         let meta_file = self.encryptor.meta_file_path.clone();
-        self.io_looper.as_mut().unwrap().post_and_kill(|callback| {
-            callback.downcast_mut::<IOWriter>().unwrap().remove_file();
+        self.io_looper.as_mut().unwrap().post_and_kill(|writer| {
+            writer.remove_file();
             #[cfg(feature = "encryption")]
             let _ = fs::remove_file(meta_file);
             info!(LOG_TAG, "data cleared");

src/ffi/mod.rs

@@ -57,6 +57,8 @@ pub struct NativeLogger {
 
 unsafe impl Send for NativeLogger {}
 
+unsafe impl Sync for NativeLogger {}
+
 impl Drop for NativeLogger {
     fn drop(&mut self) {
         verbose!(LOG_TAG, "release {:?}", self);

src/log/logger.rs

@@ -8,7 +8,7 @@ use std::{process, thread};
 
 use crate::log::{LogLevel, Logger};
 struct LogWrapper {
-    io_looper: IOLooper,
+    io_looper: IOLooper<LogWriter>,
 }
 
 struct LogWriter {
@@ -54,8 +54,7 @@ impl LogWrapper {
         let log_str = format!("[{}] {}", tag, content);
         // write log in io thread
         self.io_looper
-            .post(move |callback| {
-                let writer = callback.downcast_ref::<LogWriter>().unwrap();
+            .post(move |writer| {
                 writer.write(level, time, pid, tid, log_str);
             })
             .unwrap();
@@ -89,8 +88,7 @@ pub fn set_logger(log_impl: Option<Box<dyn Logger>>) {
     // Here move the log_impl to io thread
     LOG_WRAPPER
         .io_looper
-        .post(|callback| {
-            let writer = callback.downcast_mut::<LogWriter>().unwrap();
+        .post(|writer| {
             drop(writer.inner_logger.take());
             writer.inner_logger = log_impl;
         })

src/log/mod.rs

@@ -7,7 +7,7 @@ See [MMKV::set_logger](crate::MMKV::set_logger)
 
 Logger should be [`Send`], cause it will be moved into io thread
  */
-pub trait Logger: Debug + Send {
+pub trait Logger: Debug + Send + Sync {
     fn verbose(&self, log_str: String);
     fn info(&self, log_str: String);
     fn debug(&self, log_str: String);