Merge pull request #578 from sdroege/thread-pool

Add glib::ThreadPool bindings

Author: sdroege

src/lib.rs

@@ -192,6 +192,9 @@ mod main_context_futures;
 mod source_futures;
 pub use source_futures::*;
 
+mod thread_pool;
+pub use thread_pool::ThreadPool;
+
 // Actual thread IDs can be reused by the OS once the old thread finished.
 // This works around it by using our own counter for threads.
 //

src/thread_pool.rs

@@ -0,0 +1,206 @@
+// Copyright 2020, The Gtk-rs Project Developers.
+// See the COPYRIGHT file at the top-level directory of this distribution.
+// Licensed under the MIT license, see the LICENSE file or <http://opensource.org/licenses/MIT>
+
+use glib_sys;
+use translate::*;
+
+use futures_channel::oneshot;
+use std::future::Future;
+use std::ptr;
+
+#[derive(Debug)]
+pub struct ThreadPool(ptr::NonNull<glib_sys::GThreadPool>);
+
+unsafe impl Send for ThreadPool {}
+unsafe impl Sync for ThreadPool {}
+
+impl ThreadPool {
+    pub fn new_shared(max_threads: Option<u32>) -> Result<Self, ::Error> {
+        unsafe {
+            let mut err = ptr::null_mut();
+            let pool = glib_sys::g_thread_pool_new(
+                Some(spawn_func),
+                ptr::null_mut(),
+                max_threads.map(|v| v as i32).unwrap_or(-1),
+                glib_sys::GFALSE,
+                &mut err,
+            );
+            if pool.is_null() {
+                Err(from_glib_full(err))
+            } else {
+                Ok(ThreadPool(ptr::NonNull::new_unchecked(pool)))
+            }
+        }
+    }
+
+    pub fn new_exclusive(max_threads: u32) -> Result<Self, ::Error> {
+        unsafe {
+            let mut err = ptr::null_mut();
+            let pool = glib_sys::g_thread_pool_new(
+                Some(spawn_func),
+                ptr::null_mut(),
+                max_threads as i32,
+                glib_sys::GTRUE,
+                &mut err,
+            );
+            if pool.is_null() {
+                Err(from_glib_full(err))
+            } else {
+                Ok(ThreadPool(ptr::NonNull::new_unchecked(pool)))
+            }
+        }
+    }
+
+    pub fn push<F: FnOnce() + Send + 'static>(&self, func: F) -> Result<(), ::Error> {
+        unsafe {
+            let func: Box<dyn FnOnce() + Send + 'static> = Box::new(func);
+            let func = Box::new(func);
+            let mut err = ptr::null_mut();
+
+            let func = Box::into_raw(func);
+            let ret: bool = from_glib(glib_sys::g_thread_pool_push(
+                self.0.as_ptr(),
+                func as *mut _,
+                &mut err,
+            ));
+            if ret {
+                Ok(())
+            } else {
+                let _ = Box::from_raw(func);
+                Err(from_glib_full(err))
+            }
+        }
+    }
+
+    pub fn push_future<T: Send + 'static, F: FnOnce() -> T + Send + 'static>(
+        &self,
+        func: F,
+    ) -> Result<impl Future<Output = T>, ::Error> {
+        use futures_util::future::FutureExt;
+
+        let (sender, receiver) = oneshot::channel();
+
+        self.push(move || {
+            let _ = sender.send(func());
+        })?;
+
+        Ok(receiver.map(|res| res.expect("Dropped before executing")))
+    }
+
+    pub fn set_max_threads(&self, max_threads: Option<u32>) -> Result<(), ::Error> {
+        unsafe {
+            let mut err = ptr::null_mut();
+            let ret: bool = from_glib(glib_sys::g_thread_pool_set_max_threads(
+                self.0.as_ptr(),
+                max_threads.map(|v| v as i32).unwrap_or(-1),
+                &mut err,
+            ));
+            if ret {
+                Ok(())
+            } else {
+                Err(from_glib_full(err))
+            }
+        }
+    }
+
+    pub fn get_max_threads(&self) -> Option<u32> {
+        unsafe {
+            let max_threads = glib_sys::g_thread_pool_get_max_threads(self.0.as_ptr());
+            if max_threads == -1 {
+                None
+            } else {
+                Some(max_threads as u32)
+            }
+        }
+    }
+
+    pub fn get_num_threads(&self) -> u32 {
+        unsafe { glib_sys::g_thread_pool_get_num_threads(self.0.as_ptr()) }
+    }
+
+    pub fn get_unprocessed(&self) -> u32 {
+        unsafe { glib_sys::g_thread_pool_unprocessed(self.0.as_ptr()) }
+    }
+
+    pub fn set_max_unused_threads(max_threads: Option<u32>) {
+        unsafe {
+            glib_sys::g_thread_pool_set_max_unused_threads(
+                max_threads.map(|v| v as i32).unwrap_or(-1),
+            )
+        }
+    }
+
+    pub fn get_max_unused_threads() -> Option<u32> {
+        unsafe {
+            let max_unused_threads = glib_sys::g_thread_pool_get_max_unused_threads();
+            if max_unused_threads == -1 {
+                None
+            } else {
+                Some(max_unused_threads as u32)
+            }
+        }
+    }
+
+    pub fn get_num_unused_threads() -> u32 {
+        unsafe { glib_sys::g_thread_pool_get_num_unused_threads() }
+    }
+
+    pub fn stop_unused_threads() {
+        unsafe {
+            glib_sys::g_thread_pool_stop_unused_threads();
+        }
+    }
+
+    pub fn set_max_idle_time(max_idle_time: u32) {
+        unsafe { glib_sys::g_thread_pool_set_max_idle_time(max_idle_time) }
+    }
+
+    pub fn get_max_idle_time() -> u32 {
+        unsafe { glib_sys::g_thread_pool_get_max_idle_time() }
+    }
+}
+
+impl Drop for ThreadPool {
+    fn drop(&mut self) {
+        unsafe {
+            glib_sys::g_thread_pool_free(self.0.as_ptr(), glib_sys::GFALSE, glib_sys::GTRUE);
+        }
+    }
+}
+
+unsafe extern "C" fn spawn_func(func: glib_sys::gpointer, _data: glib_sys::gpointer) {
+    let func: Box<Box<dyn FnOnce()>> = Box::from_raw(func as *mut _);
+    func()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_push() {
+        use std::sync::mpsc;
+
+        let p = ThreadPool::new_exclusive(1).unwrap();
+        let (sender, receiver) = mpsc::channel();
+
+        p.push(move || {
+            sender.send(true).unwrap();
+        })
+        .unwrap();
+
+        assert_eq!(receiver.recv(), Ok(true));
+    }
+
+    #[test]
+    fn test_push_future() {
+        let c = ::MainContext::new();
+        let p = ThreadPool::new_shared(None).unwrap();
+
+        let fut = p.push_future(|| true).unwrap();
+
+        let res = c.block_on(fut);
+        assert!(res);
+    }
+}