Strata: one client per cloud store, its requests side by side, and counts 2ddd471651cd

tokio = { version = "1", default-features = false, features = ["rt-multi-thread", "rt"] }

    Some(shared(url, given).map(|s| Box::new(Shared(s)) as Box<dyn Store>))

/// One client per store and credentials, kept for the life of the process

/// (oxygenhub#234): a server opens a repository for every request, and a

/// new client each time meant a new runtime, new connections and signing

/// in to the provider again before the first byte.

type OpenStores = std::collections::HashMap<(String, Option<Credentials>), Arc<CloudStore>>;

fn shared(url: &str, given: Option<Credentials>) -> Result<Arc<CloudStore>> {

    static OPEN: Mutex<Option<OpenStores>> = Mutex::new(None);

    let key = (url.to_string(), given);

    let mut open = OPEN.lock().unwrap_or_else(|p| p.into_inner());

    let map = open.get_or_insert_with(Default::default);

    if let Some(s) = map.get(&key) {

        return Ok(s.clone());

    }

    let store = Arc::new(CloudStore::open_with(url, key.1.as_ref())?);

    map.insert(key, store.clone());

    Ok(store)

}

/// A shared [`CloudStore`] as a [`Store`].

struct Shared(Arc<CloudStore>);

impl Store for Shared {

    fn describe(&self) -> String {

        self.0.describe()

    }

    fn has(&self, key: &str) -> Result<bool> {

        self.0.has(key)

    }

    fn get(&self, key: &str) -> Result<Vec<u8>> {

        self.0.get(key)

    }

    fn get_range(&self, key: &str, range: Range<u64>) -> Result<Vec<u8>> {

        self.0.get_range(key, range)

    }

    fn put_new(&self, key: &str, bytes: &[u8]) -> Result<()> {

        self.0.put_new(key, bytes)

    }

    fn list(&self, prefix: &str) -> Result<Vec<Entry>> {

        self.0.list(prefix)

    }

    fn delete(&self, key: &str) -> Result<()> {

        self.0.delete(key)

    }

    fn read_ref(&self, key: &str) -> Result<Option<(Vec<u8>, Version)>> {

        self.0.read_ref(key)

    }

    fn write_ref(&self, key: &str, bytes: &[u8], expected: Option<&Version>) -> Result<Version> {

        self.0.write_ref(key, bytes, expected)

    }

}

static OPENS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

static CALLS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

static MICROS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

/// What the cloud stores have done since the process started: clients

/// opened, requests made and the time spent in them, for a server to show

/// (a hub's /health).

#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]

pub struct Stats {

    pub opens: u64,

    pub calls: u64,

    pub millis: u64,

}

pub fn stats() -> Stats {

    use std::sync::atomic::Ordering::Relaxed;

    Stats {

        opens: OPENS.load(Relaxed),

        calls: CALLS.load(Relaxed),

        millis: MICROS.load(Relaxed) / 1000,

    }

}

#[derive(Clone, PartialEq, Eq, Hash)]

    /// Always there until the store is dropped: two workers, so the

    /// requests of a shared store run side by side.

        OPENS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);

        let runtime = tokio::runtime::Builder::new_multi_thread()

            .worker_threads(2)

            .thread_name("oxy-strata")

        let started = std::time::Instant::now();

        let out = if tokio::runtime::Handle::try_current().is_ok() {

            std::thread::scope(|s| {

            })

        } else {

            runtime.block_on(f)

        };

        CALLS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);

        MICROS.fetch_add(

            started.elapsed().as_micros() as u64,

            std::sync::atomic::Ordering::Relaxed,

        );

        out

    /// One client per store and credentials (oxygenhub#234): opening the

    /// same again shares it, and only a new one counts as opened.

    #[test]

    fn a_store_opened_again_shares_its_client() {

        let creds = |k: &str| Credentials::Aws {

            access_key_id: k.into(),

            secret_access_key: "s".into(),

            session_token: None,

        };

        let url = "s3://oxy-shared-test/repos/a?region=eu-west-2";

        let before = super::stats().opens;

        let a = super::shared(url, Some(creds("one"))).unwrap();

        let b = super::shared(url, Some(creds("one"))).unwrap();

        assert!(std::sync::Arc::ptr_eq(&a, &b));

        let other_keys = super::shared(url, Some(creds("two"))).unwrap();

        assert!(

            !std::sync::Arc::ptr_eq(&a, &other_keys),

            "other credentials, another client"

        );

        assert!(

            super::stats().opens - before >= 2,

            "two clients opened (other tests may open more)"

        );

    }