stormdl-segment 0.1.2

Segment manager with adaptive splitting algorithm
Documentation 
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
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137

use crate::splitter::{initial_segments, split_range};
use parking_lot::RwLock;
use std::sync::Arc;
use stormdl_core::{ByteRange, SegmentState, SegmentStatus};

pub struct SegmentManager {
    segments: Arc<RwLock<Vec<SegmentState>>>,
    total_size: u64,
    min_segment_size: u64,
    max_segments: usize,
}

impl SegmentManager {
    pub fn new(total_size: u64) -> Self {
        Self {
            segments: Arc::new(RwLock::new(Vec::new())),
            total_size,
            min_segment_size: 256 * 1024,
            max_segments: 32,
        }
    }

    pub fn with_config(total_size: u64, min_segment_size: u64, max_segments: usize) -> Self {
        Self {
            segments: Arc::new(RwLock::new(Vec::new())),
            total_size,
            min_segment_size,
            max_segments,
        }
    }

    pub fn with_segments(total_size: u64, num_segments: usize) -> Self {
        let manager = Self::new(total_size);
        let ranges = split_range(total_size, num_segments);
        let segments: Vec<SegmentState> = ranges
            .into_iter()
            .enumerate()
            .map(|(id, range)| SegmentState::new(id, range))
            .collect();
        *manager.segments.write() = segments;
        manager
    }

    pub fn initialize(&self) -> Vec<SegmentState> {
        let num_segments = initial_segments(self.total_size);
        let ranges = split_range(self.total_size, num_segments);

        let segments: Vec<SegmentState> = ranges
            .into_iter()
            .enumerate()
            .map(|(id, range)| SegmentState::new(id, range))
            .collect();

        *self.segments.write() = segments.clone();
        segments
    }

    pub fn get_segments(&self) -> Vec<SegmentState> {
        self.segments.read().clone()
    }

    pub fn update_segment(&self, id: usize, downloaded: u64, speed: f64) {
        let mut segments = self.segments.write();
        if let Some(segment) = segments.get_mut(id) {
            segment.downloaded = downloaded;
            segment.speed = speed;
        }
    }

    pub fn mark_complete(&self, id: usize) {
        let mut segments = self.segments.write();
        if let Some(segment) = segments.get_mut(id) {
            segment.status = SegmentStatus::Complete;
            segment.downloaded = segment.range.len();
        }
    }

    pub fn mark_error(&self, id: usize) {
        let mut segments = self.segments.write();
        if let Some(segment) = segments.get_mut(id) {
            segment.status = SegmentStatus::Error;
        }
    }

    pub fn split_segment(&self, id: usize) -> Option<SegmentState> {
        let mut segments = self.segments.write();

        if segments.len() >= self.max_segments {
            return None;
        }

        let segment = segments.get(id)?;
        let remaining = segment.remaining();

        if remaining < self.min_segment_size * 2 {
            return None;
        }

        let current_offset = segment.range.start + segment.downloaded;
        let split_point = current_offset + remaining / 2;

        let new_id = segments.len();
        let new_range = ByteRange::new(split_point, segment.range.end);

        segments.get_mut(id)?.range.end = split_point;

        let new_segment = SegmentState::new(new_id, new_range);
        segments.push(new_segment.clone());

        Some(new_segment)
    }

    pub fn total_downloaded(&self) -> u64 {
        self.segments.read().iter().map(|s| s.downloaded).sum()
    }

    pub fn all_complete(&self) -> bool {
        self.segments
            .read()
            .iter()
            .all(|s| s.status == SegmentStatus::Complete)
    }

    pub fn average_speed(&self) -> f64 {
        let segments = self.segments.read();
        let active: Vec<_> = segments
            .iter()
            .filter(|s| s.status == SegmentStatus::Active)
            .collect();

        if active.is_empty() {
            return 0.0;
        }

        active.iter().map(|s| s.speed).sum::<f64>() / active.len() as f64
    }
}