You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: async_cache.rs
    path of file: ./repos/turbo/crates/turborepo-cache/src
    
    the code of the file until where you have to start completion: use std::sync::{atomic::AtomicU8, Arc};

use futures::{stream::FuturesUnordered, StreamExt};
use tokio::sync::{mpsc, Semaphore};
use tracing::{warn, Instrument, Level};
use turbopath::{AbsoluteSystemPath, AbsoluteSystemPathBuf, AnchoredSystemPathBuf};
use turborepo_analytics::AnalyticsSender;
use turborepo_api_client::{APIAuth, APIClient};

use crate::{multiplexer::CacheMultiplexer, CacheError, CacheHitMetadata, CacheOpts};

const WARNING_CUTOFF: u8 = 4;

#[derive(Clone)]
pub struct AsyncCache {
    real_cache: Arc<CacheMultiplexer>,
    writer_sender: mpsc::Sender<WorkerRequest>,
}

enum WorkerRequest {
    WriteRequest {
        anchor: AbsoluteSystemPathBuf,
        key: String,
        duration: u64,
        files: Vec<AnchoredSystemPathBuf>,
    },
    Flush(tokio::sync::oneshot::Sender<()>),
    Shutdown(tokio::sync::oneshot::Sender<()>),
}

impl AsyncCache {
    pub fn new(
        opts: &CacheOpts,
        repo_root: &AbsoluteSystemPath,
        api_client: APIClient,
        api_auth: Option<APIAuth>,
        analytics_recorder: Option<AnalyticsSender>,
    ) -> Result<AsyncCache, CacheError> {
        let max_workers = opts.workers.try_into().expect("usize is smaller than u32");
        let real_cache = Arc::new(CacheMultiplexer::new(
            opts,
            repo_root,
            api_client,
            api_auth,
            analytics_recorder,
        )?);
        let (writer_sender, mut write_consumer) = mpsc::channel(1);

        // start a task to manage workers
        let worker_real_cache = real_cache.clone();
        tokio::spawn(async move {
     
    
    async fn round_trip_test_with_both_caches(test_case: &TestCase, port: u16) -> Result<()> {
        let repo_root = tempdir()?;
        let repo_root_path = AbsoluteSystemPathBuf::try_from(repo_root.path())?;
        test_case.initialize(&repo_root_path)?;

        let hash = format!("{}-both", test_case.hash);

        let opts = CacheOpts {
            override_dir: None,
            remote_cache_read_only: false,
            skip_remote: false,
            skip_filesystem: false,
            workers: 10,
            remote_cache_opts: Some(RemoteCacheOpts {
                unused_team_id: Some("my-team".to_string()),
                signature: false,
            }),
        };

        let api_client = APIClient::new(format!("http://localhost:{}", port), 200, "2.0.0", true)?;
        let api_auth = Some(APIAuth {
            team_id: Some("my-team-id".to_string()),
            token: "my-token".to_string(),
            team_slug: None,
        });
        let async_cache = AsyncCache::new(&opts, &repo_root_path, api_client, api_auth, None)?;

        // Ensure that the cache is empty
        let response = async_cache.exists(&hash).await;

        assert_matches!(response, Ok(None));

        // Add test case
        async_cache
            .put(
                repo_root_path.clone(),
                hash.clone(),
                test_case
                    .files
                    .iter()
                    .map(|f| f.path().to_owned())
                    .collect(),
                test_case.duration,
            )
            .await
            .unwrap();

        // Wait for async cache to process
        async_cache.wait().await.unwrap();

        let fs_cache_path = repo_root_path.join_components(&[
            "node_modules",
            ".cache",
            "turbo",
            &format!("{}.tar.zst", hash),
        ]);

        // Confirm that fs cache file exists
        assert!(fs_cache_path.exists());

        let response = async_cache.exists(&hash).await?;

        // Confirm that we fetch from local cache first.
        assert_eq!(
            response,
            Some(CacheHitMetadata {
                source: CacheSource::Local,
                time_saved: test_case.duration
            })
        );

        // Remove fs cache file
        fs_cache_path.remove_file()?;

        let response = async_cache.exists(&hash).await?;

        // Confirm that we still can fetch from remote cache
        assert_eq!(
            response,
            Some(CacheHitMetadata {
                source: CacheSource::Remote,
                time_saved: test_case.duration
            })
        );

        async_cache.shutdown().await.unwrap();
        assert!(
            async_cache.shutdown().await.is_err(),
            "second shutdown should error"
        );

        Ok(())
    }