Skip to content

Commit dc76936

Browse files
committed
Refactor chunk_codec
1 parent 0f1c050 commit dc76936

11 files changed

Lines changed: 295 additions & 137 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/Cargo.lock

Lines changed: 1 addition & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

pixie-server/src/ping.rs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ use crate::{
55
state::{State, UnitSelector},
66
};
77
use anyhow::{bail, Result};
8-
use pixie_shared::{PACKET_LEN, PING_PORT};
8+
use pixie_shared::{PING_PORT, UDP_BODY_LEN};
99
use std::{
1010
net::{IpAddr, Ipv4Addr},
1111
sync::Arc,
@@ -17,7 +17,7 @@ pub async fn main(state: Arc<State>) -> Result<()> {
1717
let socket = UdpSocket::bind((Ipv4Addr::UNSPECIFIED, PING_PORT)).await?;
1818
log::info!("Listening on {}", socket.local_addr()?);
1919

20-
let mut buf = [0; PACKET_LEN];
20+
let mut buf = [0; UDP_BODY_LEN];
2121
loop {
2222
let (len, peer_addr) = tokio::select! {
2323
x = socket.recv_from(&mut buf) => x?,

pixie-server/src/udp.rs

Lines changed: 10 additions & 34 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, UDP_BODY_LEN,
1111
};
1212
use std::{
1313
collections::BTreeSet,
@@ -28,7 +28,7 @@ async fn broadcast_chunks(
2828
mut rx: Receiver<ChunkHash>,
2929
) -> Result<()> {
3030
let mut queue = BTreeSet::<ChunkHash>::new();
31-
let mut write_buf = [0; PACKET_LEN];
31+
let mut write_buf = [0; UDP_BODY_LEN];
3232
let mut wait_for = Instant::now();
3333
let mut index = [0; 32];
3434

@@ -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
}
@@ -190,7 +166,7 @@ async fn broadcast_hint(state: &State, socket: &UdpSocket, ip: Ipv4Addr) -> Resu
190166
}
191167

192168
async fn handle_requests(state: &State, socket: &UdpSocket, tx: Sender<[u8; 32]>) -> Result<()> {
193-
let mut buf = [0; PACKET_LEN];
169+
let mut buf = [0; UDP_BODY_LEN];
194170
loop {
195171
let (len, addr) = tokio::select! {
196172
x = socket.recv_from(&mut buf) => x?,

pixie-shared/Cargo.toml

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ blake3 = { version = "1", default-features = false }
88
ipnet = { version = "2.9.0", optional = true, features = ["serde"] }
99
macaddr = { version = "1.0.1", optional = true, features = ["serde"] }
1010
serde = { version = "1.0.193", default-features = false, features = ["derive", "alloc"] }
11+
thiserror = { version = "2.0.12", default-features = false }
1112

1213
[features]
1314
std = ["ipnet", "macaddr", "serde/std"]

pixie-shared/src/chunk_codec.rs

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

pixie-shared/src/lib.rs

Lines changed: 2 additions & 3 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

@@ -110,9 +111,7 @@ pub struct HintPacket {
110111
}
111112

112113
/// The maximum number of bytes in a udp packet with mtu 1500
113-
pub const PACKET_LEN: usize = 1472;
114-
pub const HEADER_LEN: usize = 34;
115-
pub const BODY_LEN: usize = PACKET_LEN - HEADER_LEN;
114+
pub const UDP_BODY_LEN: usize = 1472;
116115

117116
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
118117
#[serde(rename_all = "lowercase")]

0 commit comments

Comments
 (0)