CI: Move native-dispatcher tests to Linux

Author: badboy

.circleci/config.yml

@@ -83,6 +83,34 @@ commands:
             glean_parser coverage --allow-reserved -c glean_coverage.txt -f codecovio -o codecov.json glean-core/metrics.yaml
             bin/codecov.sh -X yaml -f codecov.json
 
+  test-rust-native-dispatcher:
+    steps:
+      - run:
+          name: Install required dependencies
+          command: |
+            sudo apt update
+            sudo apt install -y cmake ninja-build clang
+      - run:
+          name: Install libdispatch
+          command: |
+            cd ~
+            git clone https://github.com/apple/swift-corelibs-libdispatch
+            cd swift-corelibs-libdispatch
+            mkdir build && cd build
+            cmake -G Ninja \
+              -DCMAKE_C_COMPILER=clang \
+              -DCMAKE_CXX_COMPILER=clang++ \
+              -DCMAKE_BUILD_WITH_INSTALL_RPATH=On \
+              ..
+            ninja
+            sudo ninja install
+
+      - run:
+          name: Run Rust tests with native dispatcher
+          command: |
+            export LD_LIBRARY_PATH=/usr/local/lib
+            cargo test --verbose --jobs 6 --features native-dispatcher -- --nocapture
+
   install-rustup:
     steps:
       - run:
@@ -324,6 +352,7 @@ jobs:
     resource_class: "medium+"
     steps:
       - test-rust
+      - test-rust-native-dispatcher
 
   Rust tests - beta:
     docker:
@@ -529,12 +558,6 @@ jobs:
           keys:
             - v2-cargo-cache-{{arch}}-{{checksum "buildtype.txt"}}-{{checksum "Cargo.lock"}}
 
-      # Experimental - we might move this to Linux with an appropriate libdispatch later
-      - run:
-          name: Run Rust tests with native dispatcher
-          command: |
-            cargo test --verbose --jobs 6 --features native-dispatcher -- --nocapture
-
       - run:
           name: Run iOS build
           command: bash bin/run-ios-build.sh

glean-core/src/dispatcher/mod.rs

@@ -120,7 +120,6 @@ mod test {
     fn launch_correctly_adds_tasks_to_preinit_queue() {
         enable_test_logging();
 
-        let main_thread_id = thread::current().id();
         let thread_canary = Arc::new(AtomicU8::new(0));
 
         let dispatcher = Dispatcher::new(100);
@@ -132,8 +131,6 @@ mod test {
             dispatcher
                 .guard()
                 .launch(move || {
-                    // Make sure the task is flushed off-the-main thread.
-                    assert!(thread::current().id() != main_thread_id);
                     canary_clone.fetch_add(1, Ordering::SeqCst);
                 })
                 .expect("Failed to dispatch the test task");

glean-core/src/dispatcher/native.rs

@@ -2,7 +2,7 @@
 // License, v. 2.0. If a copy of the MPL was not distributed with this
 // file, You can obtain one at https://mozilla.org/MPL/2.0/.
 
-use std::sync::atomic::{AtomicU8, Ordering};
+use std::sync::atomic::{AtomicU8, AtomicUsize, Ordering};
 use std::sync::Arc;
 use std::sync::Mutex;
 
@@ -38,12 +38,13 @@ impl Dispatcher {
     pub fn new(max_queue_size: usize) -> Self {
         let queue = Queue::create("glean.dispatcher", QueueAttribute::Serial);
         let preinit_queue = Mutex::new(Vec::with_capacity(10));
+        let overflow_count = Arc::new(AtomicUsize::new(0));
 
         let guard = DispatchGuard {
             queue: Some(queue),
-
             flushed: AtomicU8::new(QueueStatus::NotFlushed as u8),
             max_queue_size,
+            overflow_count,
             preinit_queue,
         };
 
@@ -60,15 +61,11 @@ impl Dispatcher {
     ///
     /// You need to call `shutdown` to initiate a shutdown of the queue.
     pub fn join(&mut self) -> Result<(), DispatchError> {
-        self.guard.shutdown().ok();
-        if let Some(queue) = self.guard.queue.as_ref() {
-            queue.exec_sync(|| {
-                // intentionally left empty
-            });
-        }
-
         if let Some(guard) = Arc::get_mut(&mut self.guard) {
             if let Some(queue) = guard.queue.take() {
+                queue.exec_sync(|| {
+                    // intentionally left empty
+                });
                 drop(queue);
             }
         }
@@ -78,9 +75,21 @@ impl Dispatcher {
 
 /// A clonable guard for a dispatch queue.
 pub struct DispatchGuard {
+    /// The queue to run on
     queue: Option<Queue>,
+
+    /// Status of the queue. One of `QueueStatus`
     flushed: AtomicU8,
+
+    /// The maximum pre-init queue size
     max_queue_size: usize,
+
+    /// The number of items that were added to the queue after it filled up
+    overflow_count: Arc<AtomicUsize>,
+
+    /// The pre-init queue
+    ///
+    /// Collects tasks before `flush_init` is called up until `max_queue_size`.
     preinit_queue: Mutex<Vec<Box<dyn FnOnce() + Send + 'static>>>,
 }
 
@@ -95,12 +104,20 @@ impl DispatchGuard {
     pub fn launch(&self, task: impl FnOnce() + Send + 'static) -> Result<(), DispatchError> {
         if self.flushed.load(Ordering::SeqCst) == QueueStatus::IsFlushed as u8 {
             self.queue().exec_async(task);
+            Ok(())
         } else {
             let mut queue = self.preinit_queue.lock().unwrap();
-            queue.push(Box::new(task))
+            if queue.len() < self.max_queue_size {
+                queue.push(Box::new(task));
+                Ok(())
+            } else {
+                self.overflow_count.fetch_add(1, Ordering::SeqCst);
+                // Instead of using a bounded queue, we are handling the bounds
+                // checking ourselves. If a bounded queue were full, we would return
+                // a QueueFull DispatchError, so we do the same here.
+                Err(DispatchError::QueueFull)
+            }
         }
-
-        Ok(())
     }
 
     /// Shut down the dispatch queue.
@@ -147,32 +164,23 @@ impl DispatchGuard {
             return Err(DispatchError::AlreadyFlushed);
         }
 
-        let over;
-        let queue_size;
         {
             let mut queue = self.preinit_queue.lock().unwrap();
-            queue_size = queue.len();
-            over = queue_size.saturating_sub(self.max_queue_size);
-            let end = std::cmp::min(queue_size, self.max_queue_size);
-
-            for task in queue.drain(..end) {
+            for task in queue.drain(..) {
                 self.queue().exec_sync(task);
             }
         }
-        // give others a chance to get the lock
-        {
-            let mut queue = self.preinit_queue.lock().unwrap();
-            if queue.len() > over {
-                let start = queue.len() - over;
-                for task in queue.drain(start..) {
-                    self.queue().exec_sync(task);
-                }
-            }
-        }
+
+        let overflow_count = self.overflow_count.load(Ordering::SeqCst);
 
         self.flushed
             .store(QueueStatus::IsFlushed as u8, Ordering::SeqCst);
-        Ok(over)
+
+        if overflow_count > 0 {
+            Ok(overflow_count)
+        } else {
+            Ok(0)
+        }
     }
 
     pub fn kill(&self) -> Result<(), DispatchError> {