Δημοσίευση πηγής

CH-J Tools

Περιηγηθείτε σε καταλόγους και αρχεία για μια συγκεκριμένη έκδοση εφαρμογών.

Κατεβάστε την πηγή ZIP

CH-J / Server Manager 1704 αρχεία

0.1.2-beta Προστέθηκε: 8 Νοε 2026, 2:53 π.μ. · Χτισμένο: 7 Νοε 2026, 7:32 μ.μ.

pipeline.rs Προστέθηκε: Ημερομηνία που δεν έχει καταγραφεί 10,0 KB · 254 γραμμές

Κωδικός Λήψη αρχείου

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 }

94}

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)

99}

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)

103}

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)

107}

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)

111}

112

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>;

115

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),

122}

123

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).

126

pub 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 }

198}

199

200/// Resume check: the last confirmed bytes of the partial copy must equal the source.

201pub const TAIL_CHECK: u64 = 1024 * 1024;

202

203

pub 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(())

213}

214

215#[cfg(test)]

216mod tests {

217 use super::*;

218 use parking_lot::Mutex;

219

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 }

254}

255

SHA-256

951cf8adf1feaee83dda56336092e7d2aca412c2dc2834f52cefe4f3db0b88f1

CH-J Proprietary Software License 1.14 · Η πηγή παρέχεται στο CH-J Proprietary Software License 1.14. Η διαθεσιμότητά του δεν αλλάζει τους όρους της άδειας χρήσης ούτε χορηγεί πρόσθετα δικαιώματα.

Archive SHA-256

ac8fe790f4ba60fc098f8c985fc2656cfcc6cdc21fb34b30e308a2b54e62258f