YGGverse/aquatic
1
0
Fork 0
mirror of https://github.com/YGGverse/aquatic.git synced 2026-04-02 02:35:31 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

bittorrent_udp: add RequestParseError; remove InvalidRequest; other fixes

Browse source
This commit is contained in:
Joakim Frostegård 2020-04-10 15:14:50 +02:00
parent f80646e3a8
commit fe85901021
5 changed files with 114 additions and 70 deletions
Download patch file Download diff file Expand all files Collapse all files

1
TODO.md
View file

@ -11,7 +11,6 @@
* quickcheck request byte conversion * quickcheck request byte conversion
* other test cases * other test cases
* Check if announce response to bytes code changed caused slowdown * Check if announce response to bytes code changed caused slowdown
* thiserror errors instead of InvalidRequest
## Not important ## Not important

27
aquatic/src/lib/network.rs
View file

@ -151,16 +151,27 @@ fn handle_readable_socket(
Ok(Request::Scrape(r)) => { Ok(Request::Scrape(r)) => {
scrape_requests.push((r, src)); scrape_requests.push((r, src));
}, },
Ok(Request::Invalid(r)) => {
let response = Response::Error(ErrorResponse {
transaction_id: r.transaction_id,
message: "Invalid request".to_string(),
});
responses.push((response, src));
},
Err(err) => { Err(err) => {
eprintln!("request_from_bytes error: {:?}", err); eprintln!("request_from_bytes error: {:?}", err);
if let Some(transaction_id) = err.transaction_id {
let opt_message = if err.error.is_some() {
Some("Parse error".to_string())
} else if let Some(message) = err.message {
Some(message)
} else {
None
};
if let Some(message) = opt_message {
let response = ErrorResponse {
transaction_id,
message,
};
responses.push((response.into(), src));
}
}
}, },
} }
}, },

14
aquatic_bench/src/bin/bench_handlers/main.rs
View file

@ -2,11 +2,11 @@
//! //!
//! Example summary output: //! Example summary output:
//! ``` //! ```
//! ## Average results over 50 rounds //! ## Average results over 20 rounds
//! //!
//! Connect handler: 2 530 072 requests/second, 395.38 ns/request //! Connect handler: 2 473 860 requests/second, 404.94 ns/request
//! Announce handler: 309 719 requests/second, 3229.87 ns/request //! Announce handler: 302 665 requests/second, 3306.17 ns/request
//! Scrape handler: 595 259 requests/second, 1680.01 ns/request //! Scrape handler: 745 598 requests/second, 1341.30 ns/request
//! ``` //! ```
use std::time::{Duration, Instant}; use std::time::{Duration, Instant};
@ -77,7 +77,7 @@ fn main(){
let mut buffer = [0u8; MAX_REQUEST_BYTES]; let mut buffer = [0u8; MAX_REQUEST_BYTES];
let mut cursor = Cursor::new(buffer.as_mut()); let mut cursor = Cursor::new(buffer.as_mut());
request_to_bytes(&mut cursor, Request::Connect(request)); request_to_bytes(&mut cursor, Request::Connect(request)).unwrap();
(buffer, src) (buffer, src)
}) })
@ -132,7 +132,7 @@ fn main(){
let mut buffer = [0u8; MAX_REQUEST_BYTES]; let mut buffer = [0u8; MAX_REQUEST_BYTES];
let mut cursor = Cursor::new(buffer.as_mut()); let mut cursor = Cursor::new(buffer.as_mut());
request_to_bytes(&mut cursor, Request::Announce(request)); request_to_bytes(&mut cursor, Request::Announce(request)).unwrap();
(buffer, src) (buffer, src)
}) })
@ -186,7 +186,7 @@ fn main(){
let mut buffer = [0u8; MAX_REQUEST_BYTES]; let mut buffer = [0u8; MAX_REQUEST_BYTES];
let mut cursor = Cursor::new(buffer.as_mut()); let mut cursor = Cursor::new(buffer.as_mut());
request_to_bytes(&mut cursor, Request::Scrape(request)); request_to_bytes(&mut cursor, Request::Scrape(request)).unwrap();
(buffer, src) (buffer, src)
}) })

135
bittorrent_udp/src/converters/requests.rs
View file

@ -12,54 +12,87 @@ use super::common::*;
const PROTOCOL_IDENTIFIER: i64 = 4_497_486_125_440; const PROTOCOL_IDENTIFIER: i64 = 4_497_486_125_440;
#[derive(Debug)]
pub struct RequestParseError {
pub transaction_id: Option<TransactionId>,
pub message: Option<String>,
pub error: Option<io::Error>,
}
impl RequestParseError {
pub fn new(err: io::Error, transaction_id: i32) -> Self {
Self {
transaction_id: Some(TransactionId(transaction_id)),
message: None,
error: Some(err)
}
}
pub fn io(err: io::Error) -> Self {
Self {
transaction_id: None,
message: None,
error: Some(err)
}
}
pub fn text(transaction_id: i32, message: &str) -> Self {
Self {
transaction_id: Some(TransactionId(transaction_id)),
message: Some(message.to_string()),
error: None,
}
}
}
#[inline] #[inline]
pub fn request_to_bytes( pub fn request_to_bytes(
bytes: &mut impl Write, bytes: &mut impl Write,
request: Request request: Request
){ ) -> Result<(), io::Error> {
match request { match request {
Request::Connect(r) => { Request::Connect(r) => {
bytes.write_i64::<NetworkEndian>(PROTOCOL_IDENTIFIER).unwrap(); bytes.write_i64::<NetworkEndian>(PROTOCOL_IDENTIFIER)?;
bytes.write_i32::<NetworkEndian>(0).unwrap(); bytes.write_i32::<NetworkEndian>(0)?;
bytes.write_i32::<NetworkEndian>(r.transaction_id.0).unwrap(); bytes.write_i32::<NetworkEndian>(r.transaction_id.0)?;
}, },
Request::Announce(r) => { Request::Announce(r) => {
bytes.write_i64::<NetworkEndian>(r.connection_id.0).unwrap(); bytes.write_i64::<NetworkEndian>(r.connection_id.0)?;
bytes.write_i32::<NetworkEndian>(1).unwrap(); bytes.write_i32::<NetworkEndian>(1)?;
bytes.write_i32::<NetworkEndian>(r.transaction_id.0).unwrap(); bytes.write_i32::<NetworkEndian>(r.transaction_id.0)?;
bytes.write_all(&r.info_hash.0).unwrap(); bytes.write_all(&r.info_hash.0)?;
bytes.write_all(&r.peer_id.0).unwrap(); bytes.write_all(&r.peer_id.0)?;
bytes.write_i64::<NetworkEndian>(r.bytes_downloaded.0).unwrap(); bytes.write_i64::<NetworkEndian>(r.bytes_downloaded.0)?;
bytes.write_i64::<NetworkEndian>(r.bytes_left.0).unwrap(); bytes.write_i64::<NetworkEndian>(r.bytes_left.0)?;
bytes.write_i64::<NetworkEndian>(r.bytes_uploaded.0).unwrap(); bytes.write_i64::<NetworkEndian>(r.bytes_uploaded.0)?;
bytes.write_i32::<NetworkEndian>(event_to_i32(r.event)).unwrap(); bytes.write_i32::<NetworkEndian>(event_to_i32(r.event))?;
bytes.write_all(&r.ip_address.map_or( bytes.write_all(&r.ip_address.map_or(
[0; 4], [0; 4],
|ip| ip.octets() |ip| ip.octets()
)).unwrap(); ))?;
bytes.write_u32::<NetworkEndian>(r.key.0).unwrap(); bytes.write_u32::<NetworkEndian>(r.key.0)?;
bytes.write_i32::<NetworkEndian>(r.peers_wanted.0).unwrap(); bytes.write_i32::<NetworkEndian>(r.peers_wanted.0)?;
bytes.write_u16::<NetworkEndian>(r.port.0).unwrap(); bytes.write_u16::<NetworkEndian>(r.port.0)?;
}, },
Request::Scrape(r) => { Request::Scrape(r) => {
bytes.write_i64::<NetworkEndian>(r.connection_id.0).unwrap(); bytes.write_i64::<NetworkEndian>(r.connection_id.0)?;
bytes.write_i32::<NetworkEndian>(2).unwrap(); bytes.write_i32::<NetworkEndian>(2)?;
bytes.write_i32::<NetworkEndian>(r.transaction_id.0).unwrap(); bytes.write_i32::<NetworkEndian>(r.transaction_id.0)?;
for info_hash in r.info_hashes { for info_hash in r.info_hashes {
bytes.write_all(&info_hash.0).unwrap(); bytes.write_all(&info_hash.0)?;
} }
} }
_ => () // Invalid requests should never happen
} }
Ok(())
} }
@ -67,12 +100,15 @@ pub fn request_to_bytes(
pub fn request_from_bytes( pub fn request_from_bytes(
bytes: &[u8], bytes: &[u8],
max_scrape_torrents: u8, max_scrape_torrents: u8,
) -> Result<Request,io::Error> { ) -> Result<Request, RequestParseError> {
let mut cursor = Cursor::new(bytes); let mut cursor = Cursor::new(bytes);
let connection_id = cursor.read_i64::<NetworkEndian>()?; let connection_id = cursor.read_i64::<NetworkEndian>()
let action = cursor.read_i32::<NetworkEndian>()?; .map_err(RequestParseError::io)?;
let transaction_id = cursor.read_i32::<NetworkEndian>()?; let action = cursor.read_i32::<NetworkEndian>()
.map_err(RequestParseError::io)?;
let transaction_id = cursor.read_i32::<NetworkEndian>()
.map_err(RequestParseError::io)?;
match action { match action {
// Connect // Connect
@ -82,12 +118,10 @@ pub fn request_from_bytes(
transaction_id: TransactionId(transaction_id) transaction_id: TransactionId(transaction_id)
}).into()) }).into())
} else { } else {
Ok(Request::Invalid(InvalidRequest { Err(RequestParseError::text(
transaction_id: TransactionId(transaction_id), transaction_id,
message: "Protocol identifier missing"
"Please send protocol identifier in connect request" ))
.to_string()
}))
} }
}, },
@ -97,19 +131,29 @@ pub fn request_from_bytes(
let mut peer_id = [0; 20]; let mut peer_id = [0; 20];
let mut ip = [0; 4]; let mut ip = [0; 4];
cursor.read_exact(&mut info_hash)?; cursor.read_exact(&mut info_hash)
cursor.read_exact(&mut peer_id)?; .map_err(|err| RequestParseError::new(err, transaction_id))?;
cursor.read_exact(&mut peer_id)
.map_err(|err| RequestParseError::new(err, transaction_id))?;
let bytes_downloaded = cursor.read_i64::<NetworkEndian>()?; let bytes_downloaded = cursor.read_i64::<NetworkEndian>()
let bytes_left = cursor.read_i64::<NetworkEndian>()?; .map_err(|err| RequestParseError::new(err, transaction_id))?;
let bytes_uploaded = cursor.read_i64::<NetworkEndian>()?; let bytes_left = cursor.read_i64::<NetworkEndian>()
let event = cursor.read_i32::<NetworkEndian>()?; .map_err(|err| RequestParseError::new(err, transaction_id))?;
let bytes_uploaded = cursor.read_i64::<NetworkEndian>()
.map_err(|err| RequestParseError::new(err, transaction_id))?;
let event = cursor.read_i32::<NetworkEndian>()
.map_err(|err| RequestParseError::new(err, transaction_id))?;
cursor.read_exact(&mut ip)?; cursor.read_exact(&mut ip)
.map_err(|err| RequestParseError::new(err, transaction_id))?;
let key = cursor.read_u32::<NetworkEndian>()?; let key = cursor.read_u32::<NetworkEndian>()
let peers_wanted = cursor.read_i32::<NetworkEndian>()?; .map_err(|err| RequestParseError::new(err, transaction_id))?;
let port = cursor.read_u16::<NetworkEndian>()?; let peers_wanted = cursor.read_i32::<NetworkEndian>()
.map_err(|err| RequestParseError::new(err, transaction_id))?;
let port = cursor.read_u16::<NetworkEndian>()
.map_err(|err| RequestParseError::new(err, transaction_id))?;
let opt_ip = if ip == [0; 4] { let opt_ip = if ip == [0; 4] {
None None
@ -150,9 +194,6 @@ pub fn request_from_bytes(
}).into()) }).into())
} }
_ => Ok(Request::Invalid(InvalidRequest { _ => Err(RequestParseError::text(transaction_id, "Invalid action"))
transaction_id: TransactionId(transaction_id),
message: "Invalid action".to_string()
}))
} }
} }

7
bittorrent_udp/src/types/request.rs
View file

@ -39,19 +39,12 @@ pub struct ScrapeRequest {
pub info_hashes: Vec<InfoHash> pub info_hashes: Vec<InfoHash>
} }
/// This is used for returning specific errors from the parser
#[derive(PartialEq, Eq, Clone, Debug)]
pub struct InvalidRequest {
pub transaction_id: TransactionId,
pub message: String
}
#[derive(PartialEq, Eq, Clone, Debug)] #[derive(PartialEq, Eq, Clone, Debug)]
pub enum Request { pub enum Request {
Connect(ConnectRequest), Connect(ConnectRequest),
Announce(AnnounceRequest), Announce(AnnounceRequest),
Scrape(ScrapeRequest), Scrape(ScrapeRequest),
Invalid(InvalidRequest),
} }