Skip to content

Commit 1ffc865

Browse files
committed
Refactor chunk_codec
1 parent 0f1c050 commit 1ffc865

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