threadshare: Drop support for multi-threaded runtime as it is consistently slower

And switch to the new built-in spawning support of CurrentThread

gst-plugin-threadshare/Cargo.toml

@@ -15,7 +15,6 @@ tokio = { git = "https://github.com/tokio-rs/tokio" }
 tokio-reactor = { git = "https://github.com/tokio-rs/tokio" }
 tokio-executor = { git = "https://github.com/tokio-rs/tokio" }
 tokio-timer = { git = "https://github.com/tokio-rs/tokio" }
-tokio-threadpool = { git = "https://github.com/tokio-rs/tokio" }
 tokio-current-thread = { git = "https://github.com/tokio-rs/tokio" }
 futures = "0.1"
 lazy_static = "1.0"

gst-plugin-threadshare/examples/benchmark.rs

@@ -53,9 +53,8 @@ fn main() {
     assert_eq!(args.len(), 6);
     let n_streams: u16 = args[1].parse().unwrap();
     let source = &args[2];
-    let n_threads: i32 = args[3].parse().unwrap();
+    let n_groups: u32 = args[3].parse().unwrap();
-    let n_groups: u32 = args[4].parse().unwrap();
+    let wait: u32 = args[4].parse().unwrap();
-    let wait: u32 = args[5].parse().unwrap();
 
     let l = glib::MainLoop::new(None, false);
     let pipeline = gst::Pipeline::new(None);
@@ -90,7 +89,6 @@ fn main() {
                 source
                     .set_property("context", &format!("context-{}", (i as u32) % n_groups))
                     .unwrap();
-                source.set_property("context-threads", &n_threads).unwrap();
                 source.set_property("context-wait", &wait).unwrap();
 
                 source
@@ -104,7 +102,6 @@ fn main() {
                 source
                     .set_property("context", &format!("context-{}", (i as u32) % n_groups))
                     .unwrap();
-                source.set_property("context-threads", &n_threads).unwrap();
                 source.set_property("context-wait", &wait).unwrap();
 
                 source
@@ -132,7 +129,6 @@ fn main() {
                 source
                     .set_property("context", &format!("context-{}", (i as u32) % n_groups))
                     .unwrap();
-                source.set_property("context-threads", &n_threads).unwrap();
                 source.set_property("context-wait", &wait).unwrap();
 
                 source

gst-plugin-threadshare/src/appsrc.rs

@@ -24,7 +24,7 @@ use gobject_subclass::object::*;
 use gst_plugin::element::*;
 
 use std::sync::Mutex;
-use std::{u16, u32};
+use std::u32;
 
 use futures::future;
 use futures::sync::{mpsc, oneshot};
@@ -37,7 +37,6 @@ use rand;
 use iocontext::*;
 
 const DEFAULT_CONTEXT: &'static str = "";
-const DEFAULT_CONTEXT_THREADS: i32 = 0;
 const DEFAULT_CONTEXT_WAIT: u32 = 0;
 const DEFAULT_CAPS: Option<gst::Caps> = None;
 const DEFAULT_MAX_BUFFERS: u32 = 10;
@@ -46,7 +45,6 @@ const DEFAULT_DO_TIMESTAMP: bool = false;
 #[derive(Debug, Clone)]
 struct Settings {
     context: String,
-    context_threads: i32,
     context_wait: u32,
     caps: Option<gst::Caps>,
     max_buffers: u32,
@@ -57,7 +55,6 @@ impl Default for Settings {
     fn default() -> Self {
         Settings {
             context: DEFAULT_CONTEXT.into(),
-            context_threads: DEFAULT_CONTEXT_THREADS,
             context_wait: DEFAULT_CONTEXT_WAIT,
             caps: DEFAULT_CAPS,
             max_buffers: DEFAULT_MAX_BUFFERS,
@@ -66,7 +63,7 @@ impl Default for Settings {
     }
 }
 
-static PROPERTIES: [Property; 6] = [
+static PROPERTIES: [Property; 5] = [
     Property::String(
         "context",
         "Context",
@@ -74,14 +71,6 @@ static PROPERTIES: [Property; 6] = [
         Some(DEFAULT_CONTEXT),
         PropertyMutability::ReadWrite,
     ),
-    Property::Int(
-        "context-threads",
-        "Context Threads",
-        "Number of threads for the context thread-pool if we create it",
-        (-1, u16::MAX as i32),
-        DEFAULT_CONTEXT_THREADS,
-        PropertyMutability::ReadWrite,
-    ),
     Property::UInt(
         "context-wait",
         "Context Wait",
@@ -472,7 +461,6 @@ impl AppSrc {
 
         let io_context = IOContext::new(
             &settings.context,
-            settings.context_threads as isize,
             settings.context_wait,
         ).map_err(|err| {
             gst_error_msg!(
@@ -565,10 +553,6 @@ impl ObjectImpl<Element> for AppSrc {
                 let mut settings = self.settings.lock().unwrap();
                 settings.context = value.get().unwrap_or_else(|| "".into());
             }
-            Property::Int("context-threads", ..) => {
-                let mut settings = self.settings.lock().unwrap();
-                settings.context_threads = value.get().unwrap();
-            }
             Property::UInt("context-wait", ..) => {
                 let mut settings = self.settings.lock().unwrap();
                 settings.context_wait = value.get().unwrap();
@@ -597,10 +581,6 @@ impl ObjectImpl<Element> for AppSrc {
                 let mut settings = self.settings.lock().unwrap();
                 Ok(settings.context.to_value())
             }
-            Property::Int("context-threads", ..) => {
-                let mut settings = self.settings.lock().unwrap();
-                Ok(settings.context_threads.to_value())
-            }
             Property::UInt("context-wait", ..) => {
                 let mut settings = self.settings.lock().unwrap();
                 Ok(settings.context_wait.to_value())

gst-plugin-threadshare/src/iocontext.rs

@@ -18,7 +18,7 @@
 use std::collections::HashMap;
 use std::io;
 use std::mem;
-use std::sync::atomic;
+use std::sync::{atomic, mpsc};
 use std::sync::{Arc, Mutex, Weak};
 use std::thread;
 
@@ -26,14 +26,12 @@ use futures::future;
 use futures::stream::futures_unordered::FuturesUnordered;
 use futures::sync::oneshot;
 use futures::{Future, Stream};
-use tokio_threadpool;
 use tokio::reactor;
 use tokio_timer::timer;
+use tokio_current_thread;
 
 use gst;
 
-use either::Either;
-
 lazy_static! {
     static ref CONTEXTS: Mutex<HashMap<String, Weak<IOContextInner>>> = Mutex::new(HashMap::new());
     static ref CONTEXT_CAT: gst::DebugCategory = gst::DebugCategory::new(
@@ -50,49 +48,14 @@ const SHUTDOWN_NOW: usize = 1;
 struct IOContextRunner {
     name: String,
     shutdown: Arc<atomic::AtomicUsize>,
-    pending_futures: Option<Arc<Mutex<Vec<Box<Future<Item = (), Error = ()> + Send + 'static>>>>>,
 }
 
 impl IOContextRunner {
-    fn start_single_threaded(
+    fn start(
         name: &str,
         wait: u32,
         reactor: reactor::Reactor,
-    ) -> (IOContextExecutor, IOContextShutdown) {
+    ) -> (tokio_current_thread::Handle, IOContextShutdown) {
-        let handle = reactor.handle().clone();
-        let handle2 = reactor.handle().clone();
-        let shutdown = Arc::new(atomic::AtomicUsize::new(RUNNING));
-        let shutdown_clone = shutdown.clone();
-        let name_clone = name.into();
-
-        let pending_futures = Arc::new(Mutex::new(Vec::new()));
-        let pending_futures_clone = pending_futures.clone();
-
-        let mut runner = IOContextRunner {
-            shutdown: shutdown_clone,
-            name: name_clone,
-            pending_futures: Some(pending_futures),
-        };
-        let join = thread::spawn(move || {
-            runner.run(wait, reactor);
-        });
-
-        let executor = IOContextExecutor {
-            handle: handle,
-            pending_futures: pending_futures_clone,
-        };
-
-        let shutdown = IOContextShutdown {
-            name: name.into(),
-            shutdown: shutdown,
-            handle: handle2,
-            join: Some(join),
-        };
-
-        (executor, shutdown)
-    }
-
-    fn start(name: &str, wait: u32, reactor: reactor::Reactor) -> IOContextShutdown {
         let handle = reactor.handle().clone();
         let shutdown = Arc::new(atomic::AtomicUsize::new(RUNNING));
         let shutdown_clone = shutdown.clone();
@@ -101,85 +64,56 @@ impl IOContextRunner {
         let mut runner = IOContextRunner {
             shutdown: shutdown_clone,
             name: name_clone,
-            pending_futures: None,
         };
+
+        let (sender, receiver) = mpsc::channel();
+
         let join = thread::spawn(move || {
-            runner.run(wait, reactor);
+            runner.run(wait, reactor, sender);
         });
 
         let shutdown = IOContextShutdown {
             name: name.into(),
-            shutdown: shutdown,
+            shutdown,
-            handle: handle,
+            handle,
             join: Some(join),
         };
 
-        shutdown
+        let runtime_handle = receiver.recv().unwrap();
+
+        (runtime_handle, shutdown)
     }
 
-    fn run(&mut self, wait: u32, reactor: reactor::Reactor) {
+    fn run(&mut self, wait: u32, reactor: reactor::Reactor, sender: mpsc::Sender<tokio_current_thread::Handle>) {
         use std::time;
         let wait = time::Duration::from_millis(wait as u64);
 
         gst_debug!(CONTEXT_CAT, "Started reactor thread '{}'", self.name);
 
-        if let Some(ref pending_futures) = self.pending_futures {
+        let handle = reactor.handle();
-            use tokio_current_thread;
+        let mut enter = ::tokio_executor::enter().unwrap();
+        let timer = timer::Timer::new(reactor);
+        let timer_handle = timer.handle();
+        let mut current_thread = tokio_current_thread::CurrentThread::new_with_park(timer);
 
-            let handle = reactor.handle();
+        let _ = sender.send(current_thread.handle());
-            let mut enter = ::tokio_executor::enter().unwrap();
-            let timer = timer::Timer::new(reactor);
-            let timer_handle = timer.handle();
-            let mut current_thread = tokio_current_thread::CurrentThread::new_with_park(timer);
 
-            ::tokio_reactor::with_default(&handle, &mut enter, |mut enter| {
+        ::tokio_timer::with_default(&timer_handle, &mut enter, |mut enter| {
-                ::tokio_timer::with_default(&timer_handle, &mut enter, |enter| loop {
+            ::tokio_reactor::with_default(&handle, &mut enter, |enter| loop {
-                    let now = time::Instant::now();
-
-                    if self.shutdown.load(atomic::Ordering::SeqCst) > RUNNING {
-                        break;
-                    }
-
-                    {
-                        let mut pending_futures = pending_futures.lock().unwrap();
-                        while let Some(future) = pending_futures.pop() {
-                            current_thread.spawn(future);
-                        }
-                    }
-
-                    gst_trace!(CONTEXT_CAT, "Turning current thread '{}'", self.name);
-                    while current_thread
-                        .enter(enter)
-                        .turn(Some(time::Duration::from_millis(0)))
-                        .unwrap()
-                        .has_polled()
-                    {}
-                    gst_trace!(CONTEXT_CAT, "Turned current thread '{}'", self.name);
-
-                    let elapsed = now.elapsed();
-                    if elapsed < wait {
-                        gst_trace!(
-                            CONTEXT_CAT,
-                            "Waiting for {:?} before polling again",
-                            wait - elapsed
-                        );
-                        thread::sleep(wait - elapsed);
-                    }
-                })
-            });
-        } else {
-            let mut reactor = reactor;
-
-            loop {
                 let now = time::Instant::now();
 
                 if self.shutdown.load(atomic::Ordering::SeqCst) > RUNNING {
                     break;
                 }
 
-                gst_trace!(CONTEXT_CAT, "Turning reactor '{}'", self.name);
+                gst_trace!(CONTEXT_CAT, "Turning current thread '{}'", self.name);
-                reactor.turn(None).unwrap();
+                while current_thread
-                gst_trace!(CONTEXT_CAT, "Turned reactor '{}'", self.name);
+                    .enter(enter)
+                    .turn(Some(time::Duration::from_millis(0)))
+                    .unwrap()
+                    .has_polled()
+                {}
+                gst_trace!(CONTEXT_CAT, "Turned current thread '{}'", self.name);
 
                 let elapsed = now.elapsed();
                 if elapsed < wait {
@@ -190,8 +124,8 @@ impl IOContextRunner {
                     );
                     thread::sleep(wait - elapsed);
                 }
-            }
+            })
-        }
+        });
     }
 }
 
@@ -223,30 +157,13 @@ impl Drop for IOContextShutdown {
     }
 }
 
-struct IOContextExecutor {
-    handle: reactor::Handle,
-    pending_futures: Arc<Mutex<Vec<Box<Future<Item = (), Error = ()> + Send + 'static>>>>,
-}
-
-impl IOContextExecutor {
-    fn spawn<F>(&self, future: F)
-    where
-        F: Future<Item = (), Error = ()> + Send + 'static,
-    {
-        use tokio_executor::park::Unpark;
-
-        self.pending_futures.lock().unwrap().push(Box::new(future));
-        self.handle.unpark();
-    }
-}
-
 #[derive(Clone)]
 pub struct IOContext(Arc<IOContextInner>);
 
 struct IOContextInner {
     name: String,
-    pool: Either<tokio_threadpool::ThreadPool, IOContextExecutor>,
+    runtime_handle: Mutex<tokio_current_thread::Handle>,
-    handle: reactor::Handle,
+    reactor_handle: reactor::Handle,
     // Only used for dropping
     _shutdown: IOContextShutdown,
     pending_futures: Mutex<(
@@ -264,7 +181,7 @@ impl Drop for IOContextInner {
 }
 
 impl IOContext {
-    pub fn new(name: &str, n_threads: isize, wait: u32) -> Result<Self, io::Error> {
+    pub fn new(name: &str, wait: u32) -> Result<Self, io::Error> {
         let mut contexts = CONTEXTS.lock().unwrap();
         if let Some(context) = contexts.get(name) {
             if let Some(context) = context.upgrade() {
@@ -274,53 +191,14 @@ impl IOContext {
         }
 
         let reactor = reactor::Reactor::new()?;
+        let reactor_handle = reactor.handle().clone();
 
-        let handle = reactor.handle().clone();
+        let (runtime_handle, shutdown) = IOContextRunner::start(name, wait, reactor);
-        let (pool, shutdown) = if n_threads >= 0 {
-            let timers = Arc::new(Mutex::new(HashMap::<_, timer::Handle>::new()));
-            let t1 = timers.clone();
-
-            let handle = handle.clone();
-            let shutdown = IOContextRunner::start(name, wait, reactor);
-
-            // FIXME: The executor threads are not throttled at all, only the reactor
-            let mut pool_builder = tokio_threadpool::Builder::new();
-            pool_builder
-                .around_worker(move |w, enter| {
-                    let timer_handle = t1.lock().unwrap().get(w.id()).unwrap().clone();
-
-                    ::tokio_reactor::with_default(&handle, enter, |enter| {
-                        timer::with_default(&timer_handle, enter, |_| {
-                            w.run();
-                        });
-                    });
-                })
-                .custom_park(move |worker_id| {
-                    // Create a new timer
-                    let timer = timer::Timer::new(::tokio_threadpool::park::DefaultPark::new());
-
-                    timers
-                        .lock()
-                        .unwrap()
-                        .insert(worker_id.clone(), timer.handle());
-
-                    timer
-                });
-
-            if n_threads > 0 {
-                pool_builder.pool_size(n_threads as usize);
-            }
-            (Either::Left(pool_builder.build()), shutdown)
-        } else {
-            let (executor, shutdown) = IOContextRunner::start_single_threaded(name, wait, reactor);
-
-            (Either::Right(executor), shutdown)
-        };
 
         let context = Arc::new(IOContextInner {
             name: name.into(),
-            pool,
+            runtime_handle: Mutex::new(runtime_handle),
-            handle,
+            reactor_handle,
             _shutdown: shutdown,
             pending_futures: Mutex::new((0, HashMap::new())),
         });
@@ -334,14 +212,11 @@ impl IOContext {
     where
         F: Future<Item = (), Error = ()> + Send + 'static,
     {
-        match self.0.pool {
+        self.0.runtime_handle.lock().unwrap().spawn(future).unwrap();
-            Either::Left(ref pool) => pool.spawn(future),
-            Either::Right(ref pool) => pool.spawn(future),
-        }
     }
 
-    pub fn handle(&self) -> &reactor::Handle {
+    pub fn reactor_handle(&self) -> &reactor::Handle {
-        &self.0.handle
+        &self.0.reactor_handle
     }
 
     pub fn acquire_pending_future_id(&self) -> PendingFutureId {

gst-plugin-threadshare/src/lib.rs

@@ -29,10 +29,9 @@ extern crate gstreamer as gst;
 
 extern crate futures;
 extern crate tokio;
-extern crate tokio_current_thread;
 extern crate tokio_executor;
 extern crate tokio_reactor;
-extern crate tokio_threadpool;
+extern crate tokio_current_thread;
 extern crate tokio_timer;
 
 extern crate either;

gst-plugin-threadshare/src/proxy.rs

@@ -26,7 +26,7 @@ use gst_plugin::element::*;
 use std::collections::HashMap;
 use std::collections::VecDeque;
 use std::sync::{Arc, Mutex, Weak};
-use std::{u16, u32, u64};
+use std::{u32, u64};
 
 use futures;
 use futures::future;
@@ -49,7 +49,6 @@ const DEFAULT_MAX_SIZE_BUFFERS: u32 = 200;
 const DEFAULT_MAX_SIZE_BYTES: u32 = 1024 * 1024;
 const DEFAULT_MAX_SIZE_TIME: u64 = gst::SECOND_VAL;
 const DEFAULT_CONTEXT: &'static str = "";
-const DEFAULT_CONTEXT_THREADS: i32 = 0;
 const DEFAULT_CONTEXT_WAIT: u32 = 0;
 
 #[derive(Debug, Clone)]
@@ -71,7 +70,6 @@ struct SettingsSrc {
     max_size_bytes: u32,
     max_size_time: u64,
     context: String,
-    context_threads: i32,
     context_wait: u32,
     proxy_context: String,
 }
@@ -83,14 +81,13 @@ impl Default for SettingsSrc {
             max_size_bytes: DEFAULT_MAX_SIZE_BYTES,
             max_size_time: DEFAULT_MAX_SIZE_TIME,
             context: DEFAULT_CONTEXT.into(),
-            context_threads: DEFAULT_CONTEXT_THREADS,
             context_wait: DEFAULT_CONTEXT_WAIT,
             proxy_context: DEFAULT_PROXY_CONTEXT.into(),
         }
     }
 }
 
-static PROPERTIES_SRC: [Property; 7] = [
+static PROPERTIES_SRC: [Property; 6] = [
     Property::UInt(
         "max-size-buffers",
         "Max Size Buffers",
@@ -122,14 +119,6 @@ static PROPERTIES_SRC: [Property; 7] = [
         Some(DEFAULT_CONTEXT),
         PropertyMutability::ReadWrite,
     ),
-    Property::Int(
-        "context-threads",
-        "Context Threads",
-        "Number of threads for the context thread-pool if we create it",
-        (-1, u16::MAX as i32),
-        DEFAULT_CONTEXT_THREADS,
-        PropertyMutability::ReadWrite,
-    ),
     Property::UInt(
         "context-wait",
         "Context Wait",
@@ -1078,7 +1067,6 @@ impl ProxySrc {
 
         let io_context = IOContext::new(
             &settings.context,
-            settings.context_threads as isize,
             settings.context_wait,
         ).map_err(|err| {
             gst_error_msg!(
@@ -1260,10 +1248,6 @@ impl ObjectImpl<Element> for ProxySrc {
                 let mut settings = self.settings.lock().unwrap();
                 settings.context = value.get().unwrap_or_else(|| "".into());
             }
-            Property::Int("context-threads", ..) => {
-                let mut settings = self.settings.lock().unwrap();
-                settings.context_threads = value.get().unwrap();
-            }
             Property::UInt("context-wait", ..) => {
                 let mut settings = self.settings.lock().unwrap();
                 settings.context_wait = value.get().unwrap();
@@ -1296,10 +1280,6 @@ impl ObjectImpl<Element> for ProxySrc {
                 let mut settings = self.settings.lock().unwrap();
                 Ok(settings.context.to_value())
             }
-            Property::Int("context-threads", ..) => {
-                let mut settings = self.settings.lock().unwrap();
-                Ok(settings.context_threads.to_value())
-            }
             Property::UInt("context-wait", ..) => {
                 let mut settings = self.settings.lock().unwrap();
                 Ok(settings.context_wait.to_value())

gst-plugin-threadshare/src/queue.rs

@@ -25,7 +25,7 @@ use gst_plugin::element::*;
 
 use std::collections::VecDeque;
 use std::sync::Mutex;
-use std::{u16, u32, u64};
+use std::{u32, u64};
 
 use futures;
 use futures::future;
@@ -41,7 +41,6 @@ const DEFAULT_MAX_SIZE_BUFFERS: u32 = 200;
 const DEFAULT_MAX_SIZE_BYTES: u32 = 1024 * 1024;
 const DEFAULT_MAX_SIZE_TIME: u64 = gst::SECOND_VAL;
 const DEFAULT_CONTEXT: &'static str = "";
-const DEFAULT_CONTEXT_THREADS: i32 = 0;
 const DEFAULT_CONTEXT_WAIT: u32 = 0;
 
 #[derive(Debug, Clone)]
@@ -50,7 +49,6 @@ struct Settings {
     max_size_bytes: u32,
     max_size_time: u64,
     context: String,
-    context_threads: i32,
     context_wait: u32,
 }
 
@@ -61,13 +59,12 @@ impl Default for Settings {
             max_size_bytes: DEFAULT_MAX_SIZE_BYTES,
             max_size_time: DEFAULT_MAX_SIZE_TIME,
             context: DEFAULT_CONTEXT.into(),
-            context_threads: DEFAULT_CONTEXT_THREADS,
             context_wait: DEFAULT_CONTEXT_WAIT,
         }
     }
 }
 
-static PROPERTIES: [Property; 6] = [
+static PROPERTIES: [Property; 5] = [
     Property::UInt(
         "max-size-buffers",
         "Max Size Buffers",
@@ -99,14 +96,6 @@ static PROPERTIES: [Property; 6] = [
         Some(DEFAULT_CONTEXT),
         PropertyMutability::ReadWrite,
     ),
-    Property::Int(
-        "context-threads",
-        "Context Threads",
-        "Number of threads for the context thread-pool if we create it",
-        (-1, u16::MAX as i32),
-        DEFAULT_CONTEXT_THREADS,
-        PropertyMutability::ReadWrite,
-    ),
     Property::UInt(
         "context-wait",
         "Context Wait",
@@ -704,7 +693,6 @@ impl Queue {
 
         let io_context = IOContext::new(
             &settings.context,
-            settings.context_threads as isize,
             settings.context_wait,
         ).map_err(|err| {
             gst_error_msg!(
@@ -866,10 +854,6 @@ impl ObjectImpl<Element> for Queue {
                 let mut settings = self.settings.lock().unwrap();
                 settings.context = value.get().unwrap_or_else(|| "".into());
             }
-            Property::Int("context-threads", ..) => {
-                let mut settings = self.settings.lock().unwrap();
-                settings.context_threads = value.get().unwrap();
-            }
             Property::UInt("context-wait", ..) => {
                 let mut settings = self.settings.lock().unwrap();
                 settings.context_wait = value.get().unwrap();
@@ -898,10 +882,6 @@ impl ObjectImpl<Element> for Queue {
                 let mut settings = self.settings.lock().unwrap();
                 Ok(settings.context.to_value())
             }
-            Property::Int("context-threads", ..) => {
-                let mut settings = self.settings.lock().unwrap();
-                Ok(settings.context_threads.to_value())
-            }
             Property::UInt("context-wait", ..) => {
                 let mut settings = self.settings.lock().unwrap();
                 Ok(settings.context_wait.to_value())

gst-plugin-threadshare/src/tcpclientsrc.rs

@@ -46,7 +46,6 @@ const DEFAULT_PORT: u32 = 5000;
 const DEFAULT_CAPS: Option<gst::Caps> = None;
 const DEFAULT_CHUNK_SIZE: u32 = 4096;
 const DEFAULT_CONTEXT: &'static str = "";
-const DEFAULT_CONTEXT_THREADS: i32 = 0;
 const DEFAULT_CONTEXT_WAIT: u32 = 0;
 
 #[derive(Debug, Clone)]
@@ -56,7 +55,6 @@ struct Settings {
     caps: Option<gst::Caps>,
     chunk_size: u32,
     context: String,
-    context_threads: i32,
     context_wait: u32,
 }
 
@@ -68,13 +66,12 @@ impl Default for Settings {
             caps: DEFAULT_CAPS,
             chunk_size: DEFAULT_CHUNK_SIZE,
             context: DEFAULT_CONTEXT.into(),
-            context_threads: DEFAULT_CONTEXT_THREADS,
             context_wait: DEFAULT_CONTEXT_WAIT,
         }
     }
 }
 
-static PROPERTIES: [Property; 7] = [
+static PROPERTIES: [Property; 6] = [
     Property::String(
         "address",
         "Address",
@@ -112,14 +109,6 @@ static PROPERTIES: [Property; 7] = [
         Some(DEFAULT_CONTEXT),
         PropertyMutability::ReadWrite,
     ),
-    Property::Int(
-        "context-threads",
-        "Context Threads",
-        "Number of threads for the context thread-pool if we create it",
-        (-1, u16::MAX as i32),
-        DEFAULT_CONTEXT_THREADS,
-        PropertyMutability::ReadWrite,
-    ),
     Property::UInt(
         "context-wait",
         "Context Wait",
@@ -456,7 +445,6 @@ impl TcpClientSrc {
 
         let io_context = IOContext::new(
             &settings.context,
-            settings.context_threads as isize,
             settings.context_wait,
         ).map_err(|err| {
             gst_error_msg!(
@@ -646,10 +634,6 @@ impl ObjectImpl<Element> for TcpClientSrc {
                 let mut settings = self.settings.lock().unwrap();
                 settings.context = value.get().unwrap_or_else(|| "".into());
             }
-            Property::Int("context-threads", ..) => {
-                let mut settings = self.settings.lock().unwrap();
-                settings.context_threads = value.get().unwrap();
-            }
             Property::UInt("context-wait", ..) => {
                 let mut settings = self.settings.lock().unwrap();
                 settings.context_wait = value.get().unwrap();
@@ -682,10 +666,6 @@ impl ObjectImpl<Element> for TcpClientSrc {
                 let mut settings = self.settings.lock().unwrap();
                 Ok(settings.context.to_value())
             }
-            Property::Int("context-threads", ..) => {
-                let mut settings = self.settings.lock().unwrap();
-                Ok(settings.context_threads.to_value())
-            }
             Property::UInt("context-wait", ..) => {
                 let mut settings = self.settings.lock().unwrap();
                 Ok(settings.context_wait.to_value())

gst-plugin-threadshare/src/udpsrc.rs

@@ -47,7 +47,6 @@ const DEFAULT_REUSE: bool = true;
 const DEFAULT_CAPS: Option<gst::Caps> = None;
 const DEFAULT_MTU: u32 = 1500;
 const DEFAULT_CONTEXT: &'static str = "";
-const DEFAULT_CONTEXT_THREADS: i32 = 0;
 const DEFAULT_CONTEXT_WAIT: u32 = 0;
 
 #[derive(Debug, Clone)]
@@ -58,7 +57,6 @@ struct Settings {
     caps: Option<gst::Caps>,
     mtu: u32,
     context: String,
-    context_threads: i32,
     context_wait: u32,
 }
 
@@ -71,13 +69,12 @@ impl Default for Settings {
             caps: DEFAULT_CAPS,
             mtu: DEFAULT_MTU,
             context: DEFAULT_CONTEXT.into(),
-            context_threads: DEFAULT_CONTEXT_THREADS,
             context_wait: DEFAULT_CONTEXT_WAIT,
         }
     }
 }
 
-static PROPERTIES: [Property; 8] = [
+static PROPERTIES: [Property; 7] = [
     Property::String(
         "address",
         "Address",
@@ -122,14 +119,6 @@ static PROPERTIES: [Property; 8] = [
         Some(DEFAULT_CONTEXT),
         PropertyMutability::ReadWrite,
     ),
-    Property::Int(
-        "context-threads",
-        "Context Threads",
-        "Number of threads for the context thread-pool if we create it",
-        (-1, u16::MAX as i32),
-        DEFAULT_CONTEXT_THREADS,
-        PropertyMutability::ReadWrite,
-    ),
     Property::UInt(
         "context-wait",
         "Context Wait",
@@ -443,7 +432,6 @@ impl UdpSrc {
 
         let io_context = IOContext::new(
             &settings.context,
-            settings.context_threads as isize,
             settings.context_wait,
         ).map_err(|err| {
             gst_error_msg!(
@@ -534,7 +522,7 @@ impl UdpSrc {
             )
         })?;
 
-        let socket = net::UdpSocket::from_std(socket, io_context.handle()).map_err(|err| {
+        let socket = net::UdpSocket::from_std(socket, io_context.reactor_handle()).map_err(|err| {
             gst_error_msg!(
                 gst::ResourceError::OpenRead,
                 ["Failed to setup socket for tokio: {}", err]
@@ -717,10 +705,6 @@ impl ObjectImpl<Element> for UdpSrc {
                 let mut settings = self.settings.lock().unwrap();
                 settings.context = value.get().unwrap_or_else(|| "".into());
             }
-            Property::Int("context-threads", ..) => {
-                let mut settings = self.settings.lock().unwrap();
-                settings.context_threads = value.get().unwrap();
-            }
             Property::UInt("context-wait", ..) => {
                 let mut settings = self.settings.lock().unwrap();
                 settings.context_wait = value.get().unwrap();
@@ -757,10 +741,6 @@ impl ObjectImpl<Element> for UdpSrc {
                 let mut settings = self.settings.lock().unwrap();
                 Ok(settings.context.to_value())
             }
-            Property::Int("context-threads", ..) => {
-                let mut settings = self.settings.lock().unwrap();
-                Ok(settings.context_threads.to_value())
-            }
             Property::UInt("context-wait", ..) => {
                 let mut settings = self.settings.lock().unwrap();
                 Ok(settings.context_wait.to_value())

gst-plugin-threadshare/tests/appsrc.rs

@@ -55,7 +55,8 @@ fn init() {
     });
 }
 
-fn test_push(n_threads: i32) {
+#[test]
+fn test_push() {
     init();
 
     let pipeline = gst::Pipeline::new(None);
@@ -66,7 +67,6 @@ fn test_push(n_threads: i32) {
 
     let caps = gst::Caps::new_simple("foo/bar", &[]);
     appsrc.set_property("caps", &caps).unwrap();
-    appsrc.set_property("context-threads", &n_threads).unwrap();
     appsrc.set_property("do-timestamp", &true).unwrap();
 
     appsink.set_property("emit-signals", &true).unwrap();
@@ -141,13 +141,3 @@ fn test_push(n_threads: i32) {
 
     pipeline.set_state(gst::State::Null).into_result().unwrap();
 }
-
-#[test]
-fn test_push_single_threaded() {
-    test_push(-1);
-}
-
-#[test]
-fn test_push_multi_threaded() {
-    test_push(2);
-}

gst-plugin-threadshare/tests/proxy.rs

@@ -55,7 +55,8 @@ fn init() {
     });
 }
 
-fn test_push(n_threads: i32) {
+#[test]
+fn test_push() {
     init();
 
     let pipeline = gst::Pipeline::new(None);
@@ -72,13 +73,10 @@ fn test_push(n_threads: i32) {
 
     fakesrc.set_property("num-buffers", &3i32).unwrap();
     proxysink
-        .set_property("proxy-context", &format!("test-{}", n_threads))
+        .set_property("proxy-context", &"test")
         .unwrap();
     proxysrc
-        .set_property("proxy-context", &format!("test-{}", n_threads))
+        .set_property("proxy-context", &"test")
-        .unwrap();
-    proxysrc
-        .set_property("context-threads", &n_threads)
         .unwrap();
 
     appsink.set_property("emit-signals", &true).unwrap();
@@ -132,13 +130,3 @@ fn test_push(n_threads: i32) {
 
     pipeline.set_state(gst::State::Null).into_result().unwrap();
 }
-
-#[test]
-fn test_push_single_threaded() {
-    test_push(-1);
-}
-
-#[test]
-fn test_push_multi_threaded() {
-    test_push(2);
-}

gst-plugin-threadshare/tests/queue.rs

@@ -55,7 +55,8 @@ fn init() {
     });
 }
 
-fn test_push(n_threads: i32) {
+#[test]
+fn test_push() {
     init();
 
     let pipeline = gst::Pipeline::new(None);
@@ -68,7 +69,6 @@ fn test_push(n_threads: i32) {
     queue.link(&appsink).unwrap();
 
     fakesrc.set_property("num-buffers", &3i32).unwrap();
-    queue.set_property("context-threads", &n_threads).unwrap();
 
     appsink.set_property("emit-signals", &true).unwrap();
 
@@ -121,13 +121,3 @@ fn test_push(n_threads: i32) {
 
     pipeline.set_state(gst::State::Null).into_result().unwrap();
 }
-
-#[test]
-fn test_push_single_threaded() {
-    test_push(-1);
-}
-
-#[test]
-fn test_push_multi_threaded() {
-    test_push(2);
-}

gst-plugin-threadshare/tests/udpsrc.rs

@@ -56,7 +56,8 @@ fn init() {
     });
 }
 
-fn test_push(n_threads: i32) {
+#[test]
+fn test_push() {
     init();
 
     let pipeline = gst::Pipeline::new(None);
@@ -67,9 +68,8 @@ fn test_push(n_threads: i32) {
 
     let caps = gst::Caps::new_simple("foo/bar", &[]);
     udpsrc.set_property("caps", &caps).unwrap();
-    udpsrc.set_property("context-threads", &n_threads).unwrap();
     udpsrc
-        .set_property("port", &((5000 + n_threads) as u32))
+        .set_property("port", &(5000 as u32))
         .unwrap();
 
     appsink.set_property("emit-signals", &true).unwrap();
@@ -112,7 +112,7 @@ fn test_push(n_threads: i32) {
         let socket = net::UdpSocket::bind("0.0.0.0:0").unwrap();
 
         let ipaddr = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
-        let dest = SocketAddr::new(ipaddr, (5000 + n_threads) as u16);
+        let dest = SocketAddr::new(ipaddr, 5000 as u16);
 
         for _ in 0..3 {
             socket.send_to(&buffer, dest).unwrap();
@@ -144,13 +144,3 @@ fn test_push(n_threads: i32) {
 
     pipeline.set_state(gst::State::Null).into_result().unwrap();
 }
-
-#[test]
-fn test_push_single_threaded() {
-    test_push(-1);
-}
-
-#[test]
-fn test_push_multi_threaded() {
-    test_push(2);
-}