tough: migrate to async #687

phu-cinemo · 2023-09-28T07:39:39Z

Depend on tokio for all IO access and convert all related interfaces to async.
Replace std::io::Read with futures_core::stream::Stream for hopefully seamless migration once std::async_iter stabilized. Use async-compatible locks from tokio and coincidentally fix a locking issue in datastore.

This also affects the crates tough-kms, tough-ssm and tuftool.

Issue #, if available:
#213
#602
Description of changes:

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

webern

This is an amazing PR. I've gotten through 33 / 54 file 😅!

webern · 2023-10-04T22:38:34Z

tough/src/lib.rs

-{
-    root: R,
+pub struct RepositoryLoader<'a> {
+    root: &'a [u8],


I'm wondering if this is the right interface.

What I'm thinking about here is whether there should be a Root object (maybe one of the existing ones, need to looks at those types) that deserializes itself from a Transport.

pub struct RepositoryLoader { root: DeserializedRoot, ...

impl DeserializedRoot { pub async fn from_https(url: Url) -> Result<Self> {} pub async fn from_file(path: impl AsRef<Path>) -> Result<Self> {} pub async fn from_bytes(path: &[u8]) -> Result<Self> {} }

Basically a convenience to the user. Just a thought...

Since the root is supposed to be "shipped with your software", I figured having it in a file or in memory would be the most common case. Having Read/Stream-like access to a file doesn't make much sense as there are no async interfaces for that in serde_json. So it has to be completely in memory anyway.

You can read a file into memory with a single call already, so that would stay the same with your approach in terms of convenience.
If users keep the root in memory already (because it's linked in the binary or because they get it by means we don't support directly), that would require an extra step now.

But maybe from_https is a common thing to do?
Also, your proposal adds the slight benefit of an early deserialization error.

tough/src/lib.rs

webern · 2023-10-04T23:31:43Z

tough/src/datastore.rs

    /// TODO: [provide a thread safe interface](https://github.com/awslabs/tough/issues/602)
    ///


Looks to me like it's thread safe now! @cbgbt do you agree?

Suggested change

/// TODO: [provide a thread safe interface](https://github.com/awslabs/tough/issues/602)

///

webern · 2023-10-04T23:39:41Z

tough/src/datastore.rs

-    pub(crate) fn reader(&self, file: &str) -> Result<Option<impl Read>> {
-        let path = self.read().path().join(file);
-        match File::open(&path) {
+    pub(crate) async fn bytes(&self, file: &str) -> Result<Option<Vec<u8>>> {


Stating the obvious: returning the bytes here instead of a reader or stream solves our thread safety problems.

tough/src/http.rs

phu-cinemo · 2023-10-06T09:40:23Z

updated according to suggestions

webern

Again: amazing work.

I have gotten through all of the files. Now I need to test it in Bottlerocket and pubsys.

webern · 2023-10-06T17:30:31Z

tuftool/src/main.rs

+                Ok(entry) => {
+                    if entry.file_type().is_file() {
+                        let future = async move { process_target(entry.path()).await };
+                        Some(tokio::task::spawn(future).await.unwrap())


Why do you have an unwrap here? We generally try to avoid panicking and favor errors.
Maybe a failure to spawn is catastrophic, but maybe there's a way to produce a SpawnErrSnafu.

we'd only panic if the task happened to panic, that's why I opted for unwrap here.
but I'm sure I can turn that into an error, no problem.

webern · 2023-10-06T17:36:01Z

tuftool/src/main.rs

-        .collect()
+        }
+    })
+    .try_collect()


Note to myself: double check parallelization.

good instincts! it wasn't really parallel at all.. see my other comment

webern · 2023-10-06T17:49:03Z

tough/src/http.rs

 enum HttpResult {
-    /// We got a response with an HTTP code that indicates success.
-    Ok(reqwest::blocking::Response),
+    Ok(reqwest::Response),
+    Err(ErrorClass),
+}


Could be:

type HttpResult = std::result::Result<reqwest::Response, ErrorClass>;

The orphan rule would keep us from implementing external traits like From for it

webern · 2023-10-06T17:49:38Z

tough/src/http.rs

 /// Much of the complexity in the `fetch_with_retries` function is in deciphering the `Result`
 /// we get from `reqwest::Client::execute`. Using this enum we categorize the states of the
 /// `Result` into the categories that we need to understand.


This documentation is more applicable to enum ErrorClass now.

webern · 2023-10-06T17:53:13Z

tough/src/io.rs

-            self.digest.is_some(),
-            "DigestAdapter::read called after end of file"
-        );
+impl Stream for DigestAdapter {


I trust that your Stream implementations are correct as I have never implemented Stream myself.

webern · 2023-10-06T18:00:16Z

tough/src/io.rs

 use ring::digest::{Context, SHA256};
-use std::io::{self, Read};
+use std::{convert::TryInto, path::Path, task::Poll};
+use tokio::fs::metadata;


Suggested change

use tokio::fs::metadata;

use tokio::fs;

webern · 2023-10-06T18:00:47Z

tough/src/io.rs

-    }
+/// Async analogue of `std::path::Path::is_file`
+pub async fn is_file(path: impl AsRef<Path>) -> bool {
+    metadata(path).await.map(|m| m.is_file()).unwrap_or(false)


nit: slightly clearer that this is a tokio function.

Suggested change

metadata(path).await.map(|m| m.is_file()).unwrap_or(false)

fs::metadata(path).await.map(|m| m.is_file()).unwrap_or(false)

webern · 2023-10-06T18:00:56Z

tough/src/io.rs

-    }
+/// Async analogue of `std::path::Path::is_dir`
+pub async fn is_dir(path: impl AsRef<Path>) -> bool {
+    metadata(path).await.map(|m| m.is_dir()).unwrap_or(false)


Suggested change

metadata(path).await.map(|m| m.is_dir()).unwrap_or(false)

fs::metadata(path).await.map(|m| m.is_dir()).unwrap_or(false)

webern · 2023-10-06T18:05:00Z

tuftool/src/main.rs

-                    None
+    let indir = indir.as_ref().to_owned();
+
+    let (tx, rx) = tokio::sync::mpsc::channel(10);


Is 10 the max parallelization? Should it be num_cpu?

let (tx, rx) = tokio::sync::mpsc::channel(10);

It's just the buffer depth of the producer. WalkDir will run on a special thread reserved for blocking tasks. I chose some value >1 to avoid starvation of the consumers while this thread isn't scheduled.
tokio with rt-multi-thread enabled will automatically scale the number of threads according to the number of cores. As the consuming futures are spawned as independent tasks, they will evenly distribute across those threads.

Edit: Writing this I realized that I failed to spawn all tasks before awaiting them. This is fixed now.

webern · 2023-10-06T18:42:31Z

tough/src/lib.rs

 pub use crate::transport::{
    DefaultTransport, FilesystemTransport, Transport, TransportError, TransportErrorKind,
 };
 pub use crate::urlpath::SafeUrlPath;
+use async_recursion::async_recursion;
+pub use async_trait::async_trait;
+use bytes::Bytes;


This is in the interface:

Suggested change

use bytes::Bytes;

pub use bytes::Bytes;

webern · 2023-10-06T18:43:30Z

tough/src/lib.rs

 use crate::schema::{
    DelegatedRole, Delegations, Role, RoleType, Root, Signed, Snapshot, Timestamp,
 };
 pub use crate::target_name::TargetName;
+use crate::transport::IntoVec;


Can we implement some convenience so that download_target("sometarget").to_bytes() or to_vec() is possible?

I generalized the implementation of IntoVec to deal with any error type and made the trait public. It now can be used on streams returned by the public interface, e.g. read_target.
I hope that's what you had in mind!

webern · 2023-10-06T21:19:15Z

I have started the testing and integration process here bottlerocket-os/twoliter#98

phu-cinemo · 2023-10-09T08:17:23Z

I incorporated another round of suggestions

matze · 2023-10-26T12:36:05Z

@webern is there anything blocking from further review?

webern · 2023-10-26T15:55:59Z

@webern is there anything blocking from further review?

I have been out on vacation and will resume the work next week. The blocker is that we want to make sure Bottlerocket can be built, run and update using the new code. I have done about 75% of the work thus far.

webern · 2023-10-31T22:42:14Z

Update. I've run into a problem integrating this downstream in Bottlerocket's update system.

The LZ4 decompression library we use in updog takes a blocking Read object. I am holding a Stream of the target, which in this case is the very large bottlerocket-image.lz4 file.

I am not seeing an async-compatible (ideally stream compatible) LZ4 library at the moment.

The best I can think of is to write the LZ4 file to disk then open that file with std::fs to get a blocking Read that I can feed to the lz4 decompression lib. But this is going to be slower than what we have now (feeding the tough http transport Read straight to decompression). The file is also too large to store it in memory. Another options might be somehow spawning a runtime and blocking on each next of the Stream to convert to a Read?

@bcressey and @ecpullen 👀

https://github.com/bottlerocket-os/bottlerocket/blob/d74bdb9a3758b2d177921c9a26f8fbd4afedc100/sources/updater/updog/src/main.rs#L229

https://github.com/10XGenomics/lz4-rs/blob/418855f339fe229083f6f8cce77a61d8632ab4c6/src/decoder.rs#L23

webern · 2023-10-31T22:50:49Z

Update. I've run into a problem integrating this downstream in Bottlerocket's update system.

Hmmm, I wonder if I can create a Read object that somehow holds a handle to the tokio runtime and uses block_on.

webern · 2023-10-31T23:24:51Z

Update. I've run into a problem integrating this downstream in Bottlerocket's update system.

Hmmm, I wonder if I can create a Read object that somehow holds a handle to the tokio runtime and uses block_on.

Untested but this compiles

struct StreamRead<S>
where
    S: Stream<Item = std::result::Result<Bytes, tough::error::Error>> + Send + Unpin,
{
    runtime: tokio::runtime::Handle,
    stream: S,
}

impl<S> StreamRead<S>
where
    S: Stream<Item = std::result::Result<Bytes, tough::error::Error>> + Send + Unpin,
{
    fn new(stream: S) -> Self {
        let runtime = tokio::runtime::Handle::current();
        Self { runtime, stream }
    }
}

impl<S> Read for StreamRead<S>
where
    S: Stream<Item = std::result::Result<Bytes, tough::error::Error>> + Send + Unpin,
{
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        let maybe_next = self.runtime.block_on(self.stream.next());
        let next = match maybe_next {
            None => return Ok(0),
            Some(next) => next,
        };
        let bytes = next.map_err(|e| std::io::Error::new(ErrorKind::Other, e))?;
        let buffer_length = buf.len();
        let bytes_len = bytes.len();
        if bytes_len > buffer_length {
            // TODO - handle this painful case
            return Err(std::io::Error::new(ErrorKind::Other, "Bug"));
        }
        let bytes = bytes.iter().as_slice();
        // TODO - this seems stupid
        for i in 0..bytes_len {
            buf[i] = bytes[i];
        }
        Ok(bytes_len)
    }
}

phu-cinemo · 2023-11-02T11:30:42Z

@webern I think you can find the necessary adapters for this in tokio_util.
I didn't try it myself, but you should be able to get from Stream to AsyncRead using tokio_util::io::StreamReader. Then, you'd use tokio_util::io::SyncIoBridge to get a blocking Read.

webern

I have tested this in Bottlerocket and Bottlerocket's release tooling. Nice work!

@phu-cinemo Can you rebase to resolve the Cargo.toml conflict?

webern · 2023-11-02T22:22:13Z

Here is the corresponding Bottlerocket PR bottlerocket-os/bottlerocket#3566

Depend on tokio for all IO access and convert all related interfaces to async. Replace `std::io::Read` with `futures_core::stream::Stream` for hopefully seamless migration once `std::async_iter` stabilized. Use async-compatible locks from tokio and coincidentally fix a locking issue in datastore. This also affects the crates tough-kms, tough-ssm and tuftool.

phu-cinemo · 2023-11-03T08:17:24Z

@webern Thanks! I squashed the fixups and rebased.

zmrow

Nice work!

tough/Cargo.toml

tuftool/Cargo.toml

bcressey · 2023-11-03T18:17:01Z

tuftool/src/main.rs

 where
    T: serde::Serialize,
 {
    // Use `tempfile::NamedTempFile::persist` to perform an atomic file write.
    let parent = path.parent().context(error::PathParentSnafu { path })?;
-    let mut writer =
+    let file =


Does this need the same tokio::task::spawn_blocking handling since NamedTempFile can block?

I can do this after merging. #697

bcressey · 2023-11-03T22:05:04Z

tough/src/http.rs

 #[derive(Debug)]
-pub struct RetryRead {
+pub(crate) struct RetryStream {


Does this need to be pub(crate)? Not sure why RetryRead was pub before.

I will check this after merging #698

webern · 2023-11-03T22:47:50Z

@phu-cinemo thank you for this! Merging!

webern reviewed Oct 5, 2023

View reviewed changes

webern reviewed Oct 6, 2023

View reviewed changes

webern mentioned this pull request Oct 6, 2023

pubsys: switch to tough async bottlerocket-os/twoliter#98

Merged

5 tasks

webern approved these changes Nov 2, 2023

View reviewed changes

webern mentioned this pull request Nov 2, 2023

tough async bottlerocket-os/bottlerocket#3566

Merged

5 tasks

phu-cinemo force-pushed the async branch from aa7a3bc to c8361e8 Compare November 3, 2023 08:15

zmrow approved these changes Nov 3, 2023

View reviewed changes

webern approved these changes Nov 3, 2023

View reviewed changes

tough/Cargo.toml Show resolved Hide resolved

tuftool/Cargo.toml Show resolved Hide resolved

tuftool/Cargo.toml Show resolved Hide resolved

bcressey approved these changes Nov 3, 2023

View reviewed changes

This was referenced Nov 3, 2023

check for blocking tempfile usage in tuftool #697

Closed

can tough RetryRead be private? #698

Closed

make cargo dependencies less specific #699

Closed

webern merged commit f5fbea3 into awslabs:develop Nov 3, 2023
9 checks passed

This was referenced Nov 3, 2023

Switch to async / std::future::Future across codebase #213

Closed

Datastore: Read/write operations may not be thread safe #602

Closed

flavio mentioned this pull request Nov 10, 2023

cosign/tuf: use trustroot sigstore/sigstore-rs#305

Merged

		/// TODO: [provide a thread safe interface](https://github.com/awslabs/tough/issues/602)
		///

	metadata(path).await.map(\|m\| m.is_file()).unwrap_or(false)
	fs::metadata(path).await.map(\|m\| m.is_file()).unwrap_or(false)

	metadata(path).await.map(\|m\| m.is_dir()).unwrap_or(false)
	fs::metadata(path).await.map(\|m\| m.is_dir()).unwrap_or(false)

tough: migrate to async #687

tough: migrate to async #687

Conversation

phu-cinemo commented Sep 28, 2023

webern left a comment

Choose a reason for hiding this comment

webern Oct 4, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

phu-cinemo commented Oct 6, 2023

webern left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

phu-cinemo Oct 9, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

phu-cinemo Oct 9, 2023 • edited Loading

Choose a reason for hiding this comment

webern commented Oct 6, 2023

phu-cinemo commented Oct 9, 2023

matze commented Oct 26, 2023

webern commented Oct 26, 2023

webern commented Oct 31, 2023 • edited Loading

webern commented Oct 31, 2023

webern commented Oct 31, 2023

phu-cinemo commented Nov 2, 2023

webern left a comment

Choose a reason for hiding this comment

webern commented Nov 2, 2023

phu-cinemo commented Nov 3, 2023

zmrow left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

webern commented Nov 3, 2023

webern Oct 4, 2023 •

edited

Loading

phu-cinemo Oct 9, 2023 •

edited

Loading

phu-cinemo Oct 9, 2023 •

edited

Loading

webern commented Oct 31, 2023 •

edited

Loading