Commit 3e323b7d authored by Nika Layzell's avatar Nika Layzell
Browse files

Bug 1739727 - Part 2: Improve rust async support in moz_task, r=emilio 

This patch contains changes to moz_task to improve it's support for
async execution on multiple threads. Unlike the previous executor
implementation, this new implementation reduces the amount of unsafe
code substantially by depending on the async-task crate
(https://crates.io/crates/async-task) for the core task implementation.
This adds a few additional features:

 * Both local (no Send bound) and non-local (with Send bound) execution support,
 * Support for spawning on arbitrary nsIEventTargets or the background task pool,
 * Returned Task objects from runnables which may be .await-ed on or detach()-ed,
 * Support for spawning with the NS_DISPATCH_EVENT_MAY_BLOCK flag set,
 * Automatic use of NS_DISPATCH_AT_END when required,
 * Support for specifying the runnable priority for runnables.

There are also some correctness improvements, and exposed a better API
for dispatching normal runnable functions to background threads.

After these changes the TaskRunnable API should no longer be necessary.
It is re-implemented on top of the executor and kept in-place to avoid
rewriting all consumers.

Differential Revision: https://phabricator.services.mozilla.com/D130705
parent 19317ab2

intl/l10n/rust/gtest/test.rs

+5 −16
Original line number Diff line number Diff line
@@ -4,17 +4,10 @@ 


use l10nregistry_ffi::load::{load_async, load_sync};

use moz_task;

use std::{

    sync::atomic::{AtomicBool, Ordering::Relaxed},

    sync::Arc,

};



#[no_mangle]

pub extern "C" fn Rust_L10NLoadAsync(it_worked: *mut bool) {

    let done = Arc::new(AtomicBool::new(false));

    let done2 = done.clone();



    moz_task::spawn_current_thread(async move {

    let future = async move {

        match load_async("resource://gre/localization/en-US/toolkit/about/aboutAbout.ftl").await {

            Ok(res) => {

                assert_eq!(res.len(), 460);
@@ -25,16 +18,12 @@ pub extern "C" fn Rust_L10NLoadAsync(it_worked: *mut bool) { 
                    *it_worked = true;

                }

            }

            Err(err) => println!("{:?}", err),

            Err(err) => panic!("{:?}", err),

        }



        done.store(true, Relaxed);

    })

    .unwrap();

    };



    unsafe {

        moz_task::gtest_only::spin_event_loop_until(move || done2.load(Relaxed)).unwrap();

        *it_worked = true;

        moz_task::gtest_only::spin_event_loop_until("Rust_L10NLoadAsync", future).unwrap();

    }

}
@@ -50,6 +39,6 @@ pub extern "C" fn Rust_L10NLoadSync(it_worked: *mut bool) { 
                *it_worked = true;

            }

        }

        Err(err) => println!("{:?}", err),

        Err(err) => panic!("{:?}", err),

    }

}

intl/l10n/rust/l10nregistry-ffi/src/registry.rs

+2 −2
Original line number Diff line number Diff line
@@ -440,7 +440,7 @@ pub unsafe extern "C" fn l10nregistry_generate_bundles( 
            // Immediately spawn the task which will handle the async calls, and use an `UnboundedSender`

            // to send callbacks for specific `next()` calls to it.

            let (sender, mut receiver) = unbounded::<NextRequest>();

            moz_task::spawn_current_thread(async move {

            moz_task::spawn_local("l10nregistry_generate_bundles", async move {

                use futures::StreamExt;

                while let Some(req) = receiver.next().await {

                    let result = match iter.next().await {
@@ -450,7 +450,7 @@ pub unsafe extern "C" fn l10nregistry_generate_bundles( 
                    (req.callback)(&req.promise, result);

                }

            })

            .expect("Failed to spawn a task");

            .detach();

            let iter = GeckoFluentBundleAsyncIteratorWrapper(sender);

            Box::into_raw(Box::new(iter))

        }

intl/l10n/rust/l10nregistry-ffi/src/source.rs

+2 −2
Original line number Diff line number Diff line
@@ -344,10 +344,10 @@ pub unsafe extern "C" fn l10nfilesource_fetch_file( 
        ResourceStatus::Loaded(res) => callback(promise, Some(&res)),

        res @ ResourceStatus::Loading(_) => {

            let strong_promise = RefPtr::new(promise);

            moz_task::spawn_current_thread(async move {

            moz_task::spawn_local("l10nfilesource_fetch_file", async move {

                callback(&strong_promise, res.await.as_ref().map(|r| &**r));

            })

            .expect("Failed to spawn future");

            .detach();

        }

    }

}

intl/l10n/rust/localization-ffi/src/lib.rs

+7 −7
Original line number Diff line number Diff line
@@ -49,7 +49,7 @@ pub fn convert_args_to_owned(args: &[L10nArg]) -> Option<FluentArgs<'static>> { 
    for arg in args {

        let val = match arg.value {

            FluentArgument::Double_(d) => FluentValue::from(d),

            // We need this to be owned because we pass the result into `spawn_current_thread`.

            // We need this to be owned because we pass the result into `spawn_local`.

            FluentArgument::String(s) => FluentValue::from(Cow::Owned(s.to_utf8().to_string())),

        };

        result.set(arg.id.to_string(), val);
@@ -290,7 +290,7 @@ impl LocalizationRc { 
        let id = nsCString::from(id);

        let strong_promise = RefPtr::new(promise);



        moz_task::spawn_current_thread(async move {

        moz_task::spawn_local("LocalizationRc::format_value", async move {

            let mut errors = vec![];

            let value = if let Some(value) = bundles

                .format_value(&id.to_utf8(), args.as_ref(), &mut errors)
@@ -309,7 +309,7 @@ impl LocalizationRc { 
                .collect();

            callback(&strong_promise, &value, &errors);

        })

        .expect("Failed to spawn future");

        .detach();

    }



    pub fn format_values(
@@ -324,7 +324,7 @@ impl LocalizationRc { 


        let strong_promise = RefPtr::new(promise);



        moz_task::spawn_current_thread(async move {

        moz_task::spawn_local("LocalizationRc::format_values", async move {

            let mut errors = vec![];

            let ret_val = bundles

                .format_values(&keys, &mut errors)
@@ -350,7 +350,7 @@ impl LocalizationRc { 


            callback(&strong_promise, &ret_val, &errors);

        })

        .expect("Failed to spawn future");

        .detach();

    }



    pub fn format_messages(
@@ -369,7 +369,7 @@ impl LocalizationRc { 


        let strong_promise = RefPtr::new(promise);



        moz_task::spawn_current_thread(async move {

        moz_task::spawn_local("LocalizationRc::format_messages", async move {

            let mut errors = vec![];

            let ret_val = bundles

                .format_messages(&keys, &mut errors)
@@ -399,7 +399,7 @@ impl LocalizationRc { 


            callback(&strong_promise, &ret_val, &errors);

        })

        .expect("Failed to spawn future");

        .detach();

    }

}

toolkit/components/kvstore/src/lib.rs

+3 −8
Original line number Diff line number Diff line
@@ -25,10 +25,7 @@ mod task; 
use atomic_refcell::AtomicRefCell;

use error::KeyValueError;

use libc::c_void;

use moz_task::{

    create_background_task_queue, dispatch_background_task_with_options, DispatchOptions,

    TaskRunnable,

};

use moz_task::{create_background_task_queue, DispatchOptions, TaskRunnable};

use nserror::{nsresult, NS_ERROR_FAILURE, NS_ERROR_NO_AGGREGATION, NS_OK};

use nsstring::{nsACString, nsCString};

use owned_value::{owned_to_variant, variant_to_owned};
@@ -126,10 +123,8 @@ impl KeyValueService { 
            nsCString::from(name),

        ));



        dispatch_background_task_with_options(

            RefPtr::new(TaskRunnable::new("KVService::GetOrCreate", task)?.coerce()),

            DispatchOptions::default().may_block(true),

        )

        TaskRunnable::new("KVService::GetOrCreate", task)?

            .dispatch_background_task_with_options(DispatchOptions::default().may_block(true))

    }

}