Skip to content

Commit e44c3d0

Browse files
committed
Refactor chunk_codec
1 parent 0f1c050 commit e44c3d0

5 files changed

Lines changed: 206 additions & 116 deletions

File tree

.github/workflows/rust.yml

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ on:
88

99
jobs:
1010
clippy:
11-
name: Format & Clippy
11+
name: Cargo Format & Clippy & Test
1212
runs-on: ubuntu-latest
1313
steps:
1414
- name: Checkout
@@ -31,6 +31,7 @@ jobs:
3131
pushd $t
3232
cargo fmt --check
3333
cargo clippy -- -D warnings
34+
cargo test --no-fail-fast --all-features
3435
popd
3536
done
3637

pixie-server/src/udp.rs

Lines changed: 8 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -6,8 +6,8 @@ use crate::{
66
};
77
use anyhow::{bail, ensure, Context, Result};
88
use pixie_shared::{
9-
ChunkHash, HintPacket, RegistrationInfo, UdpRequest, ACTION_PORT, BODY_LEN, CHUNKS_PORT,
10-
HINT_PORT, PACKET_LEN,
9+
chunk_codec::Encoder, ChunkHash, HintPacket, RegistrationInfo, UdpRequest, ACTION_PORT,
10+
CHUNKS_PORT, HINT_PORT, PACKET_LEN,
1111
};
1212
use std::{
1313
collections::BTreeSet,
@@ -67,48 +67,24 @@ async fn broadcast_chunks(
6767
_ = state.cancel_token.cancelled() => break,
6868
};
6969

70-
let Some(data) = state
70+
let Some(cdata) = state
7171
.get_chunk_cdata(index)
7272
.with_context(|| format!("get chunk {}", hex::encode(index)))?
7373
else {
7474
log::warn!("Chunk {} not found", hex::encode(index));
7575
continue;
7676
};
7777

78-
let num_packets = data.len().div_ceil(BODY_LEN);
79-
write_buf[..32].clone_from_slice(&index);
80-
81-
let mut xor = [[0; BODY_LEN]; 32];
82-
8378
let hosts_cfg = &state.config.hosts;
8479
let chunks_addr = SocketAddrV4::new(ip, CHUNKS_PORT);
8580

86-
for index in 0..num_packets {
87-
write_buf[32..34].clone_from_slice(&(index as u16).to_le_bytes());
88-
let start = index * BODY_LEN;
89-
let len = BODY_LEN.min(data.len() - start);
90-
let body = &data[start..start + len];
91-
let group = index & 31;
92-
body.iter()
93-
.zip(xor[group].iter_mut())
94-
.for_each(|(a, b)| *b ^= a);
95-
write_buf[34..34 + len].clone_from_slice(body);
96-
81+
let mut encoder = Encoder::new(cdata);
82+
write_buf[..32].clone_from_slice(&index);
83+
while let Some(len) = encoder.next_packet(&mut write_buf[32..]) {
9784
time::sleep_until(wait_for).await;
9885

99-
let sent_len = socket.send_to(&write_buf[..34 + len], chunks_addr).await?;
100-
ensure!(sent_len == 34 + len, "Could not send packet");
101-
wait_for += 8 * (sent_len as u32) * Duration::from_secs(1) / hosts_cfg.broadcast_speed;
102-
}
103-
104-
for (index, body) in xor.iter().enumerate().take(num_packets) {
105-
write_buf[32..34].clone_from_slice(&(index as u16).wrapping_sub(32).to_le_bytes());
106-
let len = BODY_LEN;
107-
write_buf[34..34 + len].clone_from_slice(body);
108-
109-
time::sleep_until(wait_for).await;
110-
let sent_len = socket.send_to(&write_buf[..34 + len], chunks_addr).await?;
111-
ensure!(sent_len == 34 + len, "Could not send packet");
86+
let sent_len = socket.send_to(&write_buf[..32 + len], chunks_addr).await?;
87+
ensure!(sent_len == 32 + len, "Could not send packet");
11288
wait_for += 8 * (sent_len as u32) * Duration::from_secs(1) / hosts_cfg.broadcast_speed;
11389
}
11490
}

pixie-shared/src/chunk_codec.rs

Lines changed: 186 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,186 @@
1+
use crate::BODY_LEN;
2+
use alloc::{vec, vec::Vec};
3+
4+
pub struct Decoder {
5+
data: Vec<u8>,
6+
missing_first: Vec<bool>,
7+
missing_second: [u16; 32],
8+
missing_third: u16,
9+
}
10+
11+
impl Decoder {
12+
pub fn new(size: usize) -> Self {
13+
let num_packets = size.div_ceil(BODY_LEN);
14+
let data = vec![0; 32 * BODY_LEN + size];
15+
let missing_first = vec![true; 32 + num_packets];
16+
let missing_second: [u16; 32] = (0..32)
17+
.map(|i| ((num_packets + 31 - i) / 32) as u16)
18+
.collect::<Vec<_>>()
19+
.try_into()
20+
.unwrap();
21+
let missing_third = missing_second.iter().map(|&x| (x != 0) as u16).sum();
22+
Decoder {
23+
data,
24+
missing_first,
25+
missing_second,
26+
missing_third,
27+
}
28+
}
29+
30+
pub fn add_packet(&mut self, buf: &[u8]) {
31+
let index = u16::from_le_bytes(buf[..2].try_into().unwrap()) as usize;
32+
33+
let rot_index = (index as u16).wrapping_add(32) as usize;
34+
match &mut self.missing_first[rot_index] {
35+
false => return,
36+
x @ true => *x = false,
37+
}
38+
39+
let start = rot_index * BODY_LEN;
40+
self.data[start..start + buf.len() - 2].clone_from_slice(&buf[2..]);
41+
42+
let group = index & 31;
43+
match &mut self.missing_second[group] {
44+
0 => return,
45+
x @ 1 => *x = 0,
46+
x @ 2.. => {
47+
*x -= 1;
48+
return;
49+
}
50+
}
51+
52+
match &mut self.missing_third {
53+
0 => unreachable!(),
54+
x @ 1.. => *x -= 1,
55+
}
56+
}
57+
58+
pub fn finish(&mut self) -> Option<Vec<u8>> {
59+
if self.missing_third != 0 {
60+
return None;
61+
}
62+
63+
let mut xor = [[0; BODY_LEN]; 32];
64+
for packet in 0..self.missing_first.len() {
65+
if !self.missing_first[packet] {
66+
let group = packet & 31;
67+
self.data[BODY_LEN * packet..]
68+
.iter()
69+
.zip(xor[group].iter_mut())
70+
.for_each(|(a, b)| *b ^= a);
71+
}
72+
}
73+
for packet in 0..self.missing_first.len() {
74+
if self.missing_first[packet] {
75+
let group = packet & 31;
76+
self.data[BODY_LEN * packet..]
77+
.iter_mut()
78+
.zip(xor[group].iter())
79+
.for_each(|(a, b)| *a = *b);
80+
}
81+
}
82+
Some(self.data[32 * BODY_LEN..].to_vec())
83+
}
84+
}
85+
86+
pub struct Encoder {
87+
data: Vec<u8>,
88+
xor: Vec<[u8; BODY_LEN]>,
89+
idx: usize,
90+
}
91+
92+
impl Encoder {
93+
pub fn new(data: Vec<u8>) -> Self {
94+
let idx = 0;
95+
let xor = Vec::new();
96+
Encoder { data, xor, idx }
97+
}
98+
99+
pub fn next_packet(&mut self, out_buf: &mut [u8]) -> Option<usize> {
100+
let start = self.idx * BODY_LEN;
101+
if start < self.data.len() {
102+
let end = self.data.len().min(start + BODY_LEN);
103+
let body = &self.data[start..end];
104+
let len = body.len();
105+
out_buf[0..2].copy_from_slice(&(self.idx as u16).to_le_bytes());
106+
out_buf[2..2 + len].copy_from_slice(body);
107+
let group = self.idx & 31;
108+
if self.xor.len() <= group {
109+
self.xor.resize(group + 1, [0; BODY_LEN]);
110+
}
111+
body.iter()
112+
.zip(self.xor[group].iter_mut())
113+
.for_each(|(a, b)| *b ^= a);
114+
self.idx += 1;
115+
Some(2 + len)
116+
} else if let Some(buf) = self.xor.pop() {
117+
let group = self.xor.len();
118+
out_buf[0..2].copy_from_slice(&(group as u16).wrapping_sub(32).to_le_bytes());
119+
out_buf[2..2 + BODY_LEN].copy_from_slice(&buf);
120+
Some(2 + BODY_LEN)
121+
} else {
122+
None
123+
}
124+
}
125+
}
126+
127+
#[cfg(test)]
128+
mod tests {
129+
use super::*;
130+
use crate::PACKET_LEN;
131+
132+
fn test_chunk_skip_packet(chunk: &[u8]) {
133+
let mut encoder = Encoder::new(chunk.to_vec());
134+
let mut packets = Vec::new();
135+
let mut buf = [0u8; PACKET_LEN];
136+
while let Some(len) = encoder.next_packet(&mut buf) {
137+
packets.push(buf[..len].to_vec());
138+
}
139+
140+
for skip_idx in 0..packets.len() {
141+
let mut decoder = Decoder::new(chunk.len());
142+
for (idx, packet) in packets.iter().enumerate() {
143+
if idx != skip_idx {
144+
decoder.add_packet(packet);
145+
}
146+
}
147+
let decoded = decoder.finish().expect("Failed to decode chunk");
148+
assert_eq!(
149+
decoded, chunk,
150+
"Failed to decode chunk with skip index {}",
151+
skip_idx
152+
);
153+
}
154+
155+
let mut decoder = Decoder::new(chunk.len());
156+
for (idx, packet) in packets.iter().enumerate() {
157+
if !(idx.is_multiple_of(33) && idx / 33 < 32) {
158+
decoder.add_packet(packet);
159+
}
160+
}
161+
let decoded = decoder.finish().expect("Failed to decode chunk");
162+
assert_eq!(decoded, chunk, "Failed to decode chunk with multiple skips");
163+
}
164+
165+
#[test]
166+
fn test_small_chunk() {
167+
let mut chunk = vec![0u8; 20];
168+
let mut val = u64::MAX / 5;
169+
for x in &mut chunk {
170+
val = val.wrapping_mul(0x5DEECE66D).wrapping_add(0xB);
171+
*x = val.to_be_bytes()[0];
172+
}
173+
test_chunk_skip_packet(&chunk);
174+
}
175+
176+
#[test]
177+
fn test_big_chunk() {
178+
let mut chunk = vec![0u8; 200 << 10];
179+
let mut val = u64::MAX / 5;
180+
for x in &mut chunk {
181+
val = val.wrapping_mul(0x5DEECE66D).wrapping_add(0xB);
182+
*x = val.to_be_bytes()[0];
183+
}
184+
test_chunk_skip_packet(&chunk);
185+
}
186+
}

pixie-shared/src/lib.rs

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
extern crate alloc;
44

55
pub mod bijection;
6+
pub mod chunk_codec;
67
#[cfg(feature = "std")]
78
pub mod config;
89

0 commit comments

Comments
 (0)