CH-J ToolsSerververwaltung über SSH
Menü
Veröffentlichte Quellen

CH-J Tools

Durchsuchen Sie Verzeichnisse und Dateien einer bestimmten Anwendungsausgabe.

Quellen als ZIP herunterladen
CH-J / Server Manager 872 Dateien
core-20261030T220507Z-61e4558d889f Hinzugefügt: · Erstellt:
pipeline.rs Hinzugefügt: Datum nicht erfasst 10,0 KB · 254 Zeilen
Code Datei herunterladen
1//! Bounded transfer pipeline (port of the 0.1.0 `TransferManager._pump`): at most `PIPELINE`
2//! chunks in flight, the confirmed offset only advances over contiguous acknowledged chunks,
3//! stop requests stop issuing new chunks, and a stall without confirmed data fails the run.
5use crate::core::error::{CoreError, CoreResult};
6use futures::future::BoxFuture;
7use futures::stream::{FuturesUnordered, StreamExt};
8use std::collections::BTreeMap;
9use std::path::Path;
10use std::sync::Arc;
11use std::time::{Duration, Instant};
12use tokio::sync::watch;
14pub const CHUNK: usize = 256 * 1024;
15pub const PIPELINE: usize = 8;
16pub const STALL: Duration = Duration::from_secs(60);
18/// Local file with positioned, blocking-thread I/O.
19#[derive(Clone)]
20pub struct LocalFile(pub Arc<std::fs::File>);
22impl LocalFile {
23 pub fn open(path: &Path, flags: &str) -> std::io::Result<LocalFile> {
24 let mut options = std::fs::OpenOptions::new();
25 match flags {
26 "r" => {
27 options.read(true);
28 }
29 "r+" => {
30 options.read(true).write(true);
31 }
32 // "wx": exclusive create, private mode.
33 _ => {
34 options.read(true).write(true).create_new(true);
35 #[cfg(unix)]
36 {
37 use std::os::unix::fs::OpenOptionsExt;
38 options.mode(0o600);
39 }
40 }
41 }
42 options.open(path).map(|f| LocalFile(Arc::new(f)))
43 }
45 pub async fn read_at(&self, offset: u64, len: usize) -> std::io::Result<Vec<u8>> {
46 let file = self.0.clone();
47 tokio::task::spawn_blocking(move || {
48 let mut buffer = vec![0u8; len];
49 let mut total = 0;
50 while total < len {
51 let read = positioned_read(&file, &mut buffer[total..], offset + total as u64)?;
52 if read == 0 {
53 break;
54 }
55 total += read;
56 }
57 buffer.truncate(total);
58 Ok(buffer)
59 })
60 .await
61 .map_err(|_| std::io::Error::other("local read task failed"))?
62 }
64 pub async fn write_at(&self, offset: u64, data: Vec<u8>) -> std::io::Result<()> {
65 let file = self.0.clone();
66 tokio::task::spawn_blocking(move || {
67 let mut written = 0;
68 while written < data.len() {
69 let n = positioned_write(&file, &data[written..], offset + written as u64)?;
70 if n == 0 {
71 return Err(std::io::Error::new(std::io::ErrorKind::WriteZero, "write returned zero"));
72 }
73 written += n;
74 }
75 Ok(())
76 })
77 .await
78 .map_err(|_| std::io::Error::other("local write task failed"))?
79 }
81 pub async fn finish(&self, size: u64) -> std::io::Result<()> {
82 let file = self.0.clone();
83 tokio::task::spawn_blocking(move || {
84 file.set_len(size)?;
85 file.sync_all()
86 })
87 .await
88 .map_err(|_| std::io::Error::other("local sync task failed"))?
89 }
91 pub fn metadata(&self) -> std::io::Result<std::fs::Metadata> {
92 self.0.metadata()
93 }
96#[cfg(unix)]
97fn positioned_read(file: &std::fs::File, buffer: &mut [u8], offset: u64) -> std::io::Result<usize> {
98 std::os::unix::fs::FileExt::read_at(file, buffer, offset)
100#[cfg(windows)]
101fn positioned_read(file: &std::fs::File, buffer: &mut [u8], offset: u64) -> std::io::Result<usize> {
102 std::os::windows::fs::FileExt::seek_read(file, buffer, offset)
104#[cfg(unix)]
105fn positioned_write(file: &std::fs::File, data: &[u8], offset: u64) -> std::io::Result<usize> {
106 std::os::unix::fs::FileExt::write_at(file, data, offset)
108#[cfg(windows)]
109fn positioned_write(file: &std::fs::File, data: &[u8], offset: u64) -> std::io::Result<usize> {
110 std::os::windows::fs::FileExt::seek_write(file, data, offset)
113pub type Producer = Arc<dyn Fn(u64, usize) -> BoxFuture<'static, CoreResult<Vec<u8>>> + Send + Sync>;
114pub type Consumer = Arc<dyn Fn(Vec<u8>, u64) -> BoxFuture<'static, CoreResult<()>> + Send + Sync>;
116/// Why the pump ended without completing.
117#[derive(Debug)]
118pub enum PumpStop {
119 /// A stop request (pause/cancel/shutdown) was honoured; carries the kind.
120 Stopped(String),
121 Failed(CoreError),
124/// Copies `[start, size)` from `produce` to `consume`. `progress(confirmed)` is called after
125/// each contiguous advance. `stop` carries the requested stop kind (empty = keep running).
126pub async fn pump(start: u64, size: u64, produce: Producer, consume: Consumer, mut stop: watch::Receiver<String>, mut progress: impl FnMut(u64) + Send) -> Result<u64, PumpStop> {
127 let mut next = start;
128 let mut confirmed = start;
129 let mut completed: BTreeMap<u64, u64> = BTreeMap::new();
130 let mut inflight = FuturesUnordered::new();
131 let mut failure: Option<CoreError> = None;
132 let mut last_progress = Instant::now();
133 let grace = Duration::from_secs(5);
134 let mut stop_seen: Option<(String, Instant)> = None;
135 loop {
136 let stop_kind = stop.borrow().clone();
137 if !stop_kind.is_empty() && stop_seen.is_none() {
138 stop_seen = Some((stop_kind, Instant::now()));
139 }
140 while failure.is_none() && stop_seen.is_none() && inflight.len() < PIPELINE && next < size {
141 let position = next;
142 let length = ((size - position) as usize).min(CHUNK);
143 next += length as u64;
144 let (produce, consume) = (produce.clone(), consume.clone());
145 inflight.push(async move {
146 let buffer = produce(position, length).await?;
147 consume(buffer, position).await?;
148 Ok::<_, CoreError>((position, length as u64))
149 });
150 }
151 if inflight.is_empty() {
152 if let Some(error) = failure {
153 return Err(PumpStop::Failed(error));
154 }
155 if let Some((kind, _)) = stop_seen {
156 return Err(PumpStop::Stopped(kind));
157 }
158 if confirmed >= size {
159 return Ok(confirmed);
160 }
161 }
162 // Requests already on the wire normally settle quickly; a dead link is cut after the grace period.
163 if let Some((kind, since)) = &stop_seen {
164 // "shutdown" and a lost connection end the run at once.
165 if since.elapsed() >= if kind == "shutdown" || kind == "lost" { Duration::ZERO } else { grace } {
166 return Err(PumpStop::Stopped(kind.clone()));
167 }
168 }
169 if last_progress.elapsed() > STALL && failure.is_none() {
170 return Err(PumpStop::Failed(CoreError::new("TRANSFER_STALLED", "No data was confirmed for 60 seconds.")));
171 }
172 tokio::select! {
173 item = inflight.next(), if !inflight.is_empty() => {
174 match item {
175 Some(Ok((position, length))) => {
176 completed.insert(position, length);
177 let before = confirmed;
178 while let Some(length) = completed.remove(&confirmed) {
179 confirmed += length;
180 }
181 last_progress = Instant::now();
182 if confirmed > before {
183 progress(confirmed);
184 }
185 }
186 Some(Err(error)) => {
187 if failure.is_none() {
188 failure = Some(error);
189 }
190 }
191 None => {}
192 }
193 }
194 _ = stop.changed() => {}
195 _ = tokio::time::sleep(Duration::from_millis(500)) => {}
196 }
197 }
200/// Resume check: the last confirmed bytes of the partial copy must equal the source.
201pub const TAIL_CHECK: u64 = 1024 * 1024;
203pub async fn verify_tail(transferred: u64, read_local: BoxFuture<'_, CoreResult<Vec<u8>>>, read_remote: BoxFuture<'_, CoreResult<Vec<u8>>>, window: u64) -> CoreResult<()> {
204 if window == 0 || transferred == 0 {
205 return Ok(());
206 }
207 let (left, right) = futures::join!(read_local, read_remote);
208 let (left, right) = (left?, right?);
209 if left.len() as u64 != window || right.len() as u64 != window || left != right {
210 return Err(CoreError::new("FILE_RESUME_MISMATCH", "FILE_RESUME_MISMATCH: The partial data does not match the source; restart the transfer from the beginning."));
211 }
212 Ok(())
215#[cfg(test)]
216mod tests {
217 use super::*;
218 use parking_lot::Mutex;
220 #[tokio::test]
221 async fn pump_copies_in_order_and_honours_stop() {
222 let source: Arc<Vec<u8>> = Arc::new((0..(3 * CHUNK + 777)).map(|i| (i % 251) as u8).collect());
223 let target = Arc::new(Mutex::new(vec![0u8; source.len()]));
224 let src = source.clone();
225 let produce: Producer = Arc::new(move |pos, len| {
226 let src = src.clone();
227 Box::pin(async move {
228 // Out-of-order completion.
229 tokio::time::sleep(Duration::from_millis((pos / CHUNK as u64 % 3) * 5)).await;
230 Ok(src[pos as usize..pos as usize + len].to_vec())
231 })
232 });
233 let dst = target.clone();
234 let consume: Consumer = Arc::new(move |buf, pos| {
235 let dst = dst.clone();
236 Box::pin(async move {
237 dst.lock()[pos as usize..pos as usize + buf.len()].copy_from_slice(&buf);
238 Ok(())
239 })
240 });
241 let (_tx, rx) = watch::channel(String::new());
242 let mut seen = Vec::new();
243 let done = pump(0, source.len() as u64, produce.clone(), consume.clone(), rx, |c| seen.push(c)).await.unwrap();
244 assert_eq!(done, source.len() as u64);
245 assert_eq!(*target.lock(), *source);
246 assert!(seen.windows(2).all(|w| w[0] < w[1]));
247 let (tx, rx) = watch::channel(String::new());
248 tx.send("pause".into()).unwrap();
249 match pump(0, source.len() as u64, produce, consume, rx, |_| {}).await {
250 Err(PumpStop::Stopped(kind)) => assert_eq!(kind, "pause"),
251 other => panic!("{other:?}"),
252 }
253 }
SHA-256951cf8adf1feaee83dda56336092e7d2aca412c2dc2834f52cefe4f3db0b88f1

CH-J Proprietary Software License 1.14 · Die Quellen werden unter der CH-J Proprietary Software License 1.14 bereitgestellt. Ihre Verfügbarkeit ändert die Lizenzbedingungen nicht und gewährt keine zusätzlichen Rechte.

SHA-256 des Archivs77380d4805485ffa4027fbe7858b5836f197a4aa9d52358a43e2d12d018744df
Dateien