frugalos 1.2.0

Frugal Object Storage
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

use futures::future::{ok, Either, Loop};
use futures::Future;
use rustracing::tag::Tag;
use rustracing_jaeger::Span;
use slog::Logger;
use std::cmp::min;

use crate::client::FrugalosClient;

#[allow(clippy::too_many_arguments)]
pub fn put_many_objects<E>(
    client: FrugalosClient,
    parent: Span,
    logger: Logger,
    bucket_id: String,
    object_id_prefix: String,
    object_start_index: usize,
    object_count: usize,
    concurrency: usize,
    content: Vec<u8>,
) -> impl Future<Item = (), Error = E> {
    futures::future::loop_fn(
        (
            0,
            parent,
            logger,
            client,
            bucket_id,
            object_id_prefix,
            content,
        ),
        move |(index, parent, logger, client, bucket_id, object_id_prefix, content)| {
            if index >= object_count {
                return Either::A(ok(Loop::Break(())));
            }
            if index % 1000 == 0 {
                info!(
                    logger,
                    "Put done: {} / {} (bucket_id = {}, prefix = {}, index = {})",
                    index,
                    object_count,
                    bucket_id,
                    object_id_prefix,
                    object_start_index + index,
                )
            }
            let mut futures = vec![];
            let this_time = min(object_count - index, concurrency);
            for i in 0..this_time {
                let span = parent.child("put_many_objects.put", |span| {
                    span.tag(Tag::new("index", (object_start_index + index + i) as i64))
                        .start()
                });

                let object_id = format!("{}{}", object_id_prefix, object_start_index + index + i);
                let logger = logger.clone();
                let bucket_id = bucket_id.clone();
                let future = client
                    .request(bucket_id.clone())
                    .span(&span)
                    .put(object_id.clone(), content.clone())
                    .then(move |result| {
                        if let Err(e) = track!(result) {
                            warn!(
                                logger,
                                "Cannot put object (bucket={:?}, object={:?}): {}",
                                bucket_id,
                                object_id,
                                e,
                            );
                        }
                        ok(())
                    });
                futures.push(future);
            }
            let future = futures::future::join_all(futures).map(move |_| {
                Loop::Continue((
                    index + this_time,
                    parent,
                    logger,
                    client,
                    bucket_id,
                    object_id_prefix,
                    content,
                ))
            });
            Either::B(future)
        },
    )
}