object_store/
throttle.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18//! A throttling object store wrapper
19use parking_lot::Mutex;
20use std::ops::Range;
21use std::{convert::TryInto, sync::Arc};
22
23use crate::multipart::{MultipartStore, PartId};
24use crate::{CopyOptions, GetOptions, RenameOptions, UploadPart};
25use crate::{
26    GetResult, GetResultPayload, ListResult, MultipartId, MultipartUpload, ObjectMeta, ObjectStore,
27    PutMultipartOptions, PutOptions, PutPayload, PutResult, Result, path::Path,
28};
29use async_trait::async_trait;
30use bytes::Bytes;
31use futures_util::{FutureExt, StreamExt, stream::BoxStream};
32use std::time::Duration;
33
34/// Configuration settings for throttled store
35#[derive(Debug, Default, Clone, Copy)]
36pub struct ThrottleConfig {
37    /// Sleep duration for every call to [`delete`], or every element in [`delete_stream`].
38    ///
39    /// Sleeping is done before the underlying store is called and independently of the success of
40    /// the operation.
41    ///
42    /// [`delete`]: crate::ObjectStoreExt::delete
43    /// [`delete_stream`]: ThrottledStore::delete_stream
44    pub wait_delete_per_call: Duration,
45
46    /// Sleep duration for every byte received during [`get_opts`](ThrottledStore::get_opts).
47    ///
48    /// Sleeping is performed after the underlying store returned and only for successful gets. The
49    /// sleep duration is additive to [`wait_get_per_call`](Self::wait_get_per_call).
50    ///
51    /// Note that the per-byte sleep only happens as the user consumes the output bytes. Should
52    /// there be an intermediate failure (i.e. after partly consuming the output bytes), the
53    /// resulting sleep time will be partial as well.
54    pub wait_get_per_byte: Duration,
55
56    /// Sleep duration for every call to [`get_opts`](ThrottledStore::get_opts).
57    ///
58    /// Sleeping is done before the underlying store is called and independently of the success of
59    /// the operation. The sleep duration is additive to
60    /// [`wait_get_per_byte`](Self::wait_get_per_byte).
61    pub wait_get_per_call: Duration,
62
63    /// Sleep duration for every call to [`list`](ThrottledStore::list).
64    ///
65    /// Sleeping is done before the underlying store is called and independently of the success of
66    /// the operation. The sleep duration is additive to
67    /// [`wait_list_per_entry`](Self::wait_list_per_entry).
68    pub wait_list_per_call: Duration,
69
70    /// Sleep duration for every entry received during [`list`](ThrottledStore::list).
71    ///
72    /// Sleeping is performed after the underlying store returned and only for successful lists.
73    /// The sleep duration is additive to [`wait_list_per_call`](Self::wait_list_per_call).
74    ///
75    /// Note that the per-entry sleep only happens as the user consumes the output entries. Should
76    /// there be an intermediate failure (i.e. after partly consuming the output entries), the
77    /// resulting sleep time will be partial as well.
78    pub wait_list_per_entry: Duration,
79
80    /// Sleep duration for every call to
81    /// [`list_with_delimiter`](ThrottledStore::list_with_delimiter).
82    ///
83    /// Sleeping is done before the underlying store is called and independently of the success of
84    /// the operation. The sleep duration is additive to
85    /// [`wait_list_with_delimiter_per_entry`](Self::wait_list_with_delimiter_per_entry).
86    pub wait_list_with_delimiter_per_call: Duration,
87
88    /// Sleep duration for every entry received during
89    /// [`list_with_delimiter`](ThrottledStore::list_with_delimiter).
90    ///
91    /// Sleeping is performed after the underlying store returned and only for successful gets. The
92    /// sleep duration is additive to
93    /// [`wait_list_with_delimiter_per_call`](Self::wait_list_with_delimiter_per_call).
94    pub wait_list_with_delimiter_per_entry: Duration,
95
96    /// Sleep duration for every call to [`put_opts`](ThrottledStore::put_opts).
97    ///
98    /// Sleeping is done before the underlying store is called and independently of the success of
99    /// the operation.
100    pub wait_put_per_call: Duration,
101}
102
103/// Sleep only if non-zero duration
104async fn sleep(duration: Duration) {
105    if !duration.is_zero() {
106        tokio::time::sleep(duration).await
107    }
108}
109
110/// Store wrapper that wraps an inner store with some `sleep` calls.
111///
112/// This can be used for performance testing.
113///
114/// **Note that the behavior of the wrapper is deterministic and might not reflect real-world
115/// conditions!**
116#[derive(Debug)]
117pub struct ThrottledStore<T> {
118    inner: T,
119    config: Arc<Mutex<ThrottleConfig>>,
120}
121
122impl<T> ThrottledStore<T> {
123    /// Create new wrapper with zero waiting times.
124    pub fn new(inner: T, config: ThrottleConfig) -> Self {
125        Self {
126            inner,
127            config: Arc::new(Mutex::new(config)),
128        }
129    }
130
131    /// Mutate config.
132    pub fn config_mut<F>(&self, f: F)
133    where
134        F: Fn(&mut ThrottleConfig),
135    {
136        let mut guard = self.config.lock();
137        f(&mut guard)
138    }
139
140    /// Return copy of current config.
141    pub fn config(&self) -> ThrottleConfig {
142        *self.config.lock()
143    }
144}
145
146impl<T: ObjectStore> std::fmt::Display for ThrottledStore<T> {
147    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
148        write!(f, "ThrottledStore({})", self.inner)
149    }
150}
151
152#[async_trait]
153#[deny(clippy::missing_trait_methods)]
154impl<T: ObjectStore> ObjectStore for ThrottledStore<T> {
155    async fn put_opts(
156        &self,
157        location: &Path,
158        payload: PutPayload,
159        opts: PutOptions,
160    ) -> Result<PutResult> {
161        sleep(self.config().wait_put_per_call).await;
162        self.inner.put_opts(location, payload, opts).await
163    }
164
165    async fn put_multipart_opts(
166        &self,
167        location: &Path,
168        opts: PutMultipartOptions,
169    ) -> Result<Box<dyn MultipartUpload>> {
170        let upload = self.inner.put_multipart_opts(location, opts).await?;
171        Ok(Box::new(ThrottledUpload {
172            upload,
173            sleep: self.config().wait_put_per_call,
174        }))
175    }
176
177    async fn get_opts(&self, location: &Path, options: GetOptions) -> Result<GetResult> {
178        sleep(self.config().wait_get_per_call).await;
179
180        // need to copy to avoid moving / referencing `self`
181        let wait_get_per_byte = self.config().wait_get_per_byte;
182
183        let result = self.inner.get_opts(location, options).await?;
184        Ok(throttle_get(result, wait_get_per_byte))
185    }
186
187    async fn get_ranges(&self, location: &Path, ranges: &[Range<u64>]) -> Result<Vec<Bytes>> {
188        let config = self.config();
189
190        let total_bytes: u64 = ranges.iter().map(|range| range.end - range.start).sum();
191        let sleep_duration =
192            config.wait_get_per_call + config.wait_get_per_byte * total_bytes as u32;
193
194        sleep(sleep_duration).await;
195
196        self.inner.get_ranges(location, ranges).await
197    }
198
199    fn delete_stream(
200        &self,
201        locations: BoxStream<'static, Result<Path>>,
202    ) -> BoxStream<'static, Result<Path>> {
203        // We wait for a certain duration before each delete location.
204        // This may be suboptimal if the inner store implements batch deletes.
205        // But there is no way around unnecessary waits since we do not know
206        // how the inner store implements `delete_stream`.
207        let wait_delete_per_call = self.config().wait_delete_per_call;
208        let locations = throttle_stream(locations, move |_| wait_delete_per_call);
209        self.inner.delete_stream(locations)
210    }
211
212    fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, Result<ObjectMeta>> {
213        let stream = self.inner.list(prefix);
214        let config = Arc::clone(&self.config);
215        futures_util::stream::once(async move {
216            let config = *config.lock();
217            let wait_list_per_entry = config.wait_list_per_entry;
218            sleep(config.wait_list_per_call).await;
219            throttle_stream(stream, move |_| wait_list_per_entry)
220        })
221        .flatten()
222        .boxed()
223    }
224
225    fn list_with_offset(
226        &self,
227        prefix: Option<&Path>,
228        offset: &Path,
229    ) -> BoxStream<'static, Result<ObjectMeta>> {
230        let stream = self.inner.list_with_offset(prefix, offset);
231        let config = Arc::clone(&self.config);
232        futures_util::stream::once(async move {
233            let config = *config.lock();
234            let wait_list_per_entry = config.wait_list_per_entry;
235            sleep(config.wait_list_per_call).await;
236            throttle_stream(stream, move |_| wait_list_per_entry)
237        })
238        .flatten()
239        .boxed()
240    }
241
242    async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result<ListResult> {
243        sleep(self.config().wait_list_with_delimiter_per_call).await;
244
245        match self.inner.list_with_delimiter(prefix).await {
246            Ok(list_result) => {
247                let entries_len = usize_to_u32_saturate(list_result.objects.len());
248                sleep(self.config().wait_list_with_delimiter_per_entry * entries_len).await;
249                Ok(list_result)
250            }
251            Err(err) => Err(err),
252        }
253    }
254
255    async fn copy_opts(&self, from: &Path, to: &Path, options: CopyOptions) -> Result<()> {
256        sleep(self.config().wait_put_per_call).await;
257
258        self.inner.copy_opts(from, to, options).await
259    }
260
261    async fn rename_opts(&self, from: &Path, to: &Path, options: RenameOptions) -> Result<()> {
262        sleep(self.config().wait_put_per_call).await;
263
264        self.inner.rename_opts(from, to, options).await
265    }
266}
267
268/// Saturated `usize` to `u32` cast.
269fn usize_to_u32_saturate(x: usize) -> u32 {
270    x.try_into().unwrap_or(u32::MAX)
271}
272
273fn throttle_get(result: GetResult, wait_get_per_byte: Duration) -> GetResult {
274    #[allow(clippy::infallible_destructuring_match)]
275    let s = match result.payload {
276        GetResultPayload::Stream(s) => s,
277        #[cfg(all(feature = "fs", not(target_arch = "wasm32")))]
278        GetResultPayload::File(_, _) => unimplemented!(),
279    };
280
281    let stream = throttle_stream(s, move |bytes| {
282        let bytes_len: u32 = usize_to_u32_saturate(bytes.len());
283        wait_get_per_byte * bytes_len
284    });
285
286    GetResult {
287        payload: GetResultPayload::Stream(stream),
288        ..result
289    }
290}
291
292fn throttle_stream<T: Send + 'static, E: Send + 'static, F>(
293    stream: BoxStream<'_, Result<T, E>>,
294    delay: F,
295) -> BoxStream<'_, Result<T, E>>
296where
297    F: Fn(&T) -> Duration + Send + Sync + 'static,
298{
299    stream
300        .then(move |result| {
301            let delay = result.as_ref().ok().map(&delay).unwrap_or_default();
302            sleep(delay).then(|_| futures_util::future::ready(result))
303        })
304        .boxed()
305}
306
307#[async_trait]
308impl<T: MultipartStore> MultipartStore for ThrottledStore<T> {
309    async fn create_multipart(&self, path: &Path) -> Result<MultipartId> {
310        self.inner.create_multipart(path).await
311    }
312
313    async fn create_multipart_opts(
314        &self,
315        path: &Path,
316        opts: PutMultipartOptions,
317    ) -> Result<MultipartId> {
318        self.inner.create_multipart_opts(path, opts).await
319    }
320
321    async fn put_part(
322        &self,
323        path: &Path,
324        id: &MultipartId,
325        part_idx: usize,
326        data: PutPayload,
327    ) -> Result<PartId> {
328        sleep(self.config().wait_put_per_call).await;
329        self.inner.put_part(path, id, part_idx, data).await
330    }
331
332    async fn complete_multipart(
333        &self,
334        path: &Path,
335        id: &MultipartId,
336        parts: Vec<PartId>,
337    ) -> Result<PutResult> {
338        self.inner.complete_multipart(path, id, parts).await
339    }
340
341    async fn abort_multipart(&self, path: &Path, id: &MultipartId) -> Result<()> {
342        self.inner.abort_multipart(path, id).await
343    }
344}
345
346#[derive(Debug)]
347struct ThrottledUpload {
348    upload: Box<dyn MultipartUpload>,
349    sleep: Duration,
350}
351
352#[async_trait]
353impl MultipartUpload for ThrottledUpload {
354    fn put_part(&mut self, data: PutPayload) -> UploadPart {
355        let duration = self.sleep;
356        let put = self.upload.put_part(data);
357        Box::pin(async move {
358            sleep(duration).await;
359            put.await
360        })
361    }
362
363    async fn complete(&mut self) -> Result<PutResult> {
364        self.upload.complete().await
365    }
366
367    async fn abort(&mut self) -> Result<()> {
368        self.upload.abort().await
369    }
370}
371
372#[cfg(test)]
373mod tests {
374    use super::*;
375    #[cfg(target_os = "linux")]
376    use crate::GetResultPayload;
377    use crate::ObjectStoreExt;
378    use crate::{integration::*, memory::InMemory};
379    use futures_util::TryStreamExt;
380    use tokio::time::Duration;
381    use tokio::time::Instant;
382
383    const WAIT_TIME: Duration = Duration::from_millis(100);
384    const ZERO: Duration = Duration::from_millis(0); // Duration::default isn't constant
385
386    macro_rules! assert_bounds {
387        ($d:expr, $lower:expr) => {
388            assert_bounds!($d, $lower, $lower + 2);
389        };
390        ($d:expr, $lower:expr, $upper:expr) => {
391            let d = $d;
392            let lower = $lower * WAIT_TIME;
393            let upper = $upper * WAIT_TIME;
394            assert!(d >= lower, "{:?} must be >= than {:?}", d, lower);
395            assert!(d < upper, "{:?} must be < than {:?}", d, upper);
396        };
397    }
398
399    #[tokio::test]
400    async fn throttle_test() {
401        let inner = InMemory::new();
402        let store = ThrottledStore::new(inner, ThrottleConfig::default());
403
404        put_get_delete_list(&store).await;
405        list_uses_directories_correctly(&store).await;
406        list_with_delimiter(&store).await;
407        rename_and_copy(&store).await;
408        copy_if_not_exists(&store).await;
409        stream_get(&store).await;
410        multipart(&store, &store).await;
411        multipart_with_opts(&store, &store).await;
412        multipart_put_part_out_of_order(&store, &store).await;
413    }
414
415    #[tokio::test]
416    async fn delete_test() {
417        let inner = InMemory::new();
418        let store = ThrottledStore::new(inner, ThrottleConfig::default());
419
420        assert_bounds!(measure_delete(&store, None).await, 0);
421        assert_bounds!(measure_delete(&store, Some(0)).await, 0);
422        assert_bounds!(measure_delete(&store, Some(10)).await, 0);
423
424        store.config_mut(|cfg| cfg.wait_delete_per_call = WAIT_TIME);
425        assert_bounds!(measure_delete(&store, None).await, 1);
426        assert_bounds!(measure_delete(&store, Some(0)).await, 1);
427        assert_bounds!(measure_delete(&store, Some(10)).await, 1);
428    }
429
430    #[tokio::test]
431    // macos github runner is so slow it can't complete within WAIT_TIME*2
432    #[cfg(target_os = "linux")]
433    async fn delete_stream_test() {
434        let inner = InMemory::new();
435        let store = ThrottledStore::new(inner, ThrottleConfig::default());
436
437        assert_bounds!(measure_delete_stream(&store, 0).await, 0);
438        assert_bounds!(measure_delete_stream(&store, 10).await, 0);
439
440        store.config_mut(|cfg| cfg.wait_delete_per_call = WAIT_TIME);
441        assert_bounds!(measure_delete_stream(&store, 0).await, 0);
442        assert_bounds!(measure_delete_stream(&store, 10).await, 10);
443    }
444
445    #[tokio::test]
446    // macos github runner is so slow it can't complete within WAIT_TIME*2
447    #[cfg(target_os = "linux")]
448    async fn get_test() {
449        let inner = InMemory::new();
450        let store = ThrottledStore::new(inner, ThrottleConfig::default());
451
452        assert_bounds!(measure_get(&store, None).await, 0);
453        assert_bounds!(measure_get(&store, Some(0)).await, 0);
454        assert_bounds!(measure_get(&store, Some(10)).await, 0);
455
456        store.config_mut(|cfg| cfg.wait_get_per_call = WAIT_TIME);
457        assert_bounds!(measure_get(&store, None).await, 1);
458        assert_bounds!(measure_get(&store, Some(0)).await, 1);
459        assert_bounds!(measure_get(&store, Some(10)).await, 1);
460
461        store.config_mut(|cfg| {
462            cfg.wait_get_per_call = ZERO;
463            cfg.wait_get_per_byte = WAIT_TIME;
464        });
465        assert_bounds!(measure_get(&store, Some(2)).await, 2);
466
467        store.config_mut(|cfg| {
468            cfg.wait_get_per_call = WAIT_TIME;
469            cfg.wait_get_per_byte = WAIT_TIME;
470        });
471        assert_bounds!(measure_get(&store, Some(2)).await, 3);
472    }
473
474    #[tokio::test]
475    // macos github runner is so slow it can't complete within WAIT_TIME*2
476    #[cfg(target_os = "linux")]
477    async fn list_test() {
478        let inner = InMemory::new();
479        let store = ThrottledStore::new(inner, ThrottleConfig::default());
480
481        assert_bounds!(measure_list(&store, 0).await, 0);
482        assert_bounds!(measure_list(&store, 10).await, 0);
483
484        store.config_mut(|cfg| cfg.wait_list_per_call = WAIT_TIME);
485        assert_bounds!(measure_list(&store, 0).await, 1);
486        assert_bounds!(measure_list(&store, 10).await, 1);
487
488        store.config_mut(|cfg| {
489            cfg.wait_list_per_call = ZERO;
490            cfg.wait_list_per_entry = WAIT_TIME;
491        });
492        assert_bounds!(measure_list(&store, 2).await, 2);
493
494        store.config_mut(|cfg| {
495            cfg.wait_list_per_call = WAIT_TIME;
496            cfg.wait_list_per_entry = WAIT_TIME;
497        });
498        assert_bounds!(measure_list(&store, 2).await, 3);
499    }
500
501    #[tokio::test]
502    // macos github runner is so slow it can't complete within WAIT_TIME*2
503    #[cfg(target_os = "linux")]
504    async fn list_with_delimiter_test() {
505        let inner = InMemory::new();
506        let store = ThrottledStore::new(inner, ThrottleConfig::default());
507
508        assert_bounds!(measure_list_with_delimiter(&store, 0).await, 0);
509        assert_bounds!(measure_list_with_delimiter(&store, 10).await, 0);
510
511        store.config_mut(|cfg| cfg.wait_list_with_delimiter_per_call = WAIT_TIME);
512        assert_bounds!(measure_list_with_delimiter(&store, 0).await, 1);
513        assert_bounds!(measure_list_with_delimiter(&store, 10).await, 1);
514
515        store.config_mut(|cfg| {
516            cfg.wait_list_with_delimiter_per_call = ZERO;
517            cfg.wait_list_with_delimiter_per_entry = WAIT_TIME;
518        });
519        assert_bounds!(measure_list_with_delimiter(&store, 2).await, 2);
520
521        store.config_mut(|cfg| {
522            cfg.wait_list_with_delimiter_per_call = WAIT_TIME;
523            cfg.wait_list_with_delimiter_per_entry = WAIT_TIME;
524        });
525        assert_bounds!(measure_list_with_delimiter(&store, 2).await, 3);
526    }
527
528    #[tokio::test]
529    async fn put_test() {
530        let inner = InMemory::new();
531        let store = ThrottledStore::new(inner, ThrottleConfig::default());
532
533        assert_bounds!(measure_put(&store, 0).await, 0);
534        assert_bounds!(measure_put(&store, 10).await, 0);
535
536        store.config_mut(|cfg| cfg.wait_put_per_call = WAIT_TIME);
537        assert_bounds!(measure_put(&store, 0).await, 1);
538        assert_bounds!(measure_put(&store, 10).await, 1);
539
540        store.config_mut(|cfg| cfg.wait_put_per_call = ZERO);
541        assert_bounds!(measure_put(&store, 0).await, 0);
542    }
543
544    async fn place_test_object(store: &ThrottledStore<InMemory>, n_bytes: Option<usize>) -> Path {
545        let path = Path::from("foo");
546
547        if let Some(n_bytes) = n_bytes {
548            let data: Vec<_> = std::iter::repeat_n(1u8, n_bytes).collect();
549            store.put(&path, data.into()).await.unwrap();
550        } else {
551            // ensure object is absent
552            store.delete(&path).await.unwrap();
553        }
554
555        path
556    }
557
558    #[allow(dead_code)]
559    async fn place_test_objects(store: &ThrottledStore<InMemory>, n_entries: usize) -> Path {
560        let prefix = Path::from("foo");
561
562        // clean up store
563        let entries: Vec<_> = store.list(Some(&prefix)).try_collect().await.unwrap();
564
565        for entry in entries {
566            store.delete(&entry.location).await.unwrap();
567        }
568
569        // create new entries
570        for i in 0..n_entries {
571            let path = prefix.clone().join(i.to_string().as_str());
572            store.put(&path, "bar".into()).await.unwrap();
573        }
574
575        prefix
576    }
577
578    async fn measure_delete(store: &ThrottledStore<InMemory>, n_bytes: Option<usize>) -> Duration {
579        let path = place_test_object(store, n_bytes).await;
580
581        let t0 = Instant::now();
582        store.delete(&path).await.unwrap();
583
584        t0.elapsed()
585    }
586
587    #[allow(dead_code)]
588    async fn measure_delete_stream(store: &ThrottledStore<InMemory>, n_entries: usize) -> Duration {
589        let prefix = place_test_objects(store, n_entries).await;
590
591        // materialize the paths so that the throttle time for listing is not counted
592        let paths = store.list(Some(&prefix)).collect::<Vec<_>>().await;
593        let paths = futures_util::stream::iter(paths)
594            .map(|x| x.map(|m| m.location))
595            .boxed();
596
597        let t0 = Instant::now();
598        store
599            .delete_stream(paths)
600            .try_collect::<Vec<_>>()
601            .await
602            .unwrap();
603
604        t0.elapsed()
605    }
606
607    #[allow(dead_code)]
608    #[cfg(target_os = "linux")]
609    async fn measure_get(store: &ThrottledStore<InMemory>, n_bytes: Option<usize>) -> Duration {
610        let path = place_test_object(store, n_bytes).await;
611
612        let t0 = Instant::now();
613        let res = store.get(&path).await;
614        if n_bytes.is_some() {
615            // need to consume bytes to provoke sleep times
616            let s = match res.unwrap().payload {
617                GetResultPayload::Stream(s) => s,
618                GetResultPayload::File(_, _) => unimplemented!(),
619            };
620
621            s.map_ok(|b| bytes::BytesMut::from(&b[..]))
622                .try_concat()
623                .await
624                .unwrap();
625        } else {
626            assert!(res.is_err());
627        }
628
629        t0.elapsed()
630    }
631
632    #[allow(dead_code)]
633    async fn measure_list(store: &ThrottledStore<InMemory>, n_entries: usize) -> Duration {
634        let prefix = place_test_objects(store, n_entries).await;
635
636        let t0 = Instant::now();
637        store
638            .list(Some(&prefix))
639            .try_collect::<Vec<_>>()
640            .await
641            .unwrap();
642
643        t0.elapsed()
644    }
645
646    #[allow(dead_code)]
647    async fn measure_list_with_delimiter(
648        store: &ThrottledStore<InMemory>,
649        n_entries: usize,
650    ) -> Duration {
651        let prefix = place_test_objects(store, n_entries).await;
652
653        let t0 = Instant::now();
654        store.list_with_delimiter(Some(&prefix)).await.unwrap();
655
656        t0.elapsed()
657    }
658
659    async fn measure_put(store: &ThrottledStore<InMemory>, n_bytes: usize) -> Duration {
660        let data: Vec<_> = std::iter::repeat_n(1u8, n_bytes).collect();
661
662        let t0 = Instant::now();
663        store.put(&Path::from("foo"), data.into()).await.unwrap();
664
665        t0.elapsed()
666    }
667}
object_store/throttle.rs

object_store/
throttle.rs
