I am implementing a TCP server to pass messages.
From the tokio-yamux library I can share this tcp connection with different channels. This library only allows me to use the AsyncWriteExt and AsyncReadExt methods to write and read bytes. This is not a problem as it suits my needs.
The problem comes when I send a whole message and bytes are missing in it, so a global block happens in the application. The client sends X bytes, before being sent by tcp a header with the length of these is added. In this way the server can know firmly the bytes to be read.
For this reason, when I write bytes in this case 262144, only 262136 are written with the write_all or write method. Then as in the header it is indicated that in total there are 262144 it is blocked because there are still bytes to read.
I don't understand why this happens.
Little example: Client:
use futures::prelude::*;
use std::{error::Error, vec};
use tokio::{
io::{AsyncReadExt, AsyncWriteExt},
net::TcpStream,
};
use tokio_yamux::{config::Config, session::Session};
#[tokio::main]
pub async fn main() -> Result<(), Box<dyn Error>> {
let socket = TcpStream::connect("127.0.0.1:8080").await?;
println!("[client] connected to server: {:?}", socket.peer_addr()?);
println!("created stream");
let mut session = Session::new_client(socket, Config::default());
let ctrl = session.control();
let mut handles = Vec::new();
tokio::spawn(async move {
loop {
match session.next().await {
Some(Ok(_)) => (),
Some(Err(e)) => {
println!("{}", e);
break;
}
None => {
println!("closed");
break;
}
}
}
});
for _i in 0..1 {
let mut ctrl_clone = ctrl.clone();
handles.push(tokio::spawn(async move {
match ctrl_clone.open_stream().await {
Ok(mut stream) => {
// This sections represents Check User Operation
// Args -> operation_id = I selected a random i32
// -> queue_id = queue to write
// Writes to the server to identify the operation and get the queue
let operation_id = 0;
let queue_id = Some(2);
let data_to_send;
match queue_id {
Some(id) => {
data_to_send = vec![operation_id, id];
}
None => {
data_to_send = vec![operation_id];
}
}
let data_to_send: Vec<u8> = data_to_send
.clone()
.into_iter()
.flat_map(|x| i32::to_be_bytes(x))
.collect();
stream.write_all(&data_to_send).await.unwrap();
stream.flush().await.unwrap();
let sv_code: i32;
let mut buf = [1; 4];
// Reads from server to recieve an ACCEPTED MESSAGE
loop {
//stream.readable().await.unwrap();
match stream.read_exact(&mut buf).await {
Ok(0) => {}
Ok(n) => {
println!("Client: Reading Buffer: n_bytes {:?}", n);
sv_code = i32::from_be_bytes(buf);
break;
}
Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
//println!("Err: TCP -> SV (Write))");
continue;
}
Err(_e) => {
//return Err(e.into());
}
}
}
// Here I return a code to client just to Continue with the execution and start Sending messages
// Skipped
// This section represents SEND Operation
// Header
let data: [u8; 262140] = [1; 262140];
let mut vec_data = data.to_vec();
let len = data.len() as u32;
let len_slices = len.to_be_bytes();
for slice in len_slices {
vec_data.insert(0, slice);
}
println!("Total_len: {:?}", vec_data.len());
let n = stream.write_all(&vec_data).await.unwrap();
println!("N: {:?}", n);
stream.flush().await.unwrap();
println!("Fin write");
}
Err(e) => {
println!("{:?}", e);
}
}
}));
}
for handle in handles {
let _ = handle.await;
}
Ok(())
}
Server:
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
use std::error::Error;
use futures::prelude::*;
use tokio_yamux::{config::Config, session::Session};
// https://github.com/nervosnetwork/tentacle
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
let addr = "127.0.0.1:8080".to_string();
let listener = TcpListener::bind(&addr).await.unwrap();
while let Ok((socket, _)) = listener.accept().await {
println!("accepted a socket: {:?}", socket.peer_addr());
let mut session = Session::new_server(socket, Config::default());
tokio::spawn(async move {
while let Some(Ok(mut stream)) = session.next().await {
println!("Server accept a stream from client: id={}", stream.id());
tokio::spawn(async move {
let mut buffer = Vec::<u8>::with_capacity(8);
// Identify operation from Client
let op_code = stream.read_i32().await.unwrap();
println!("op_id: {:?}", op_code);
let queue_id = Some(stream.read_i32().await.unwrap());
println!("queue_id: {:?}", queue_id);
// Here I write to client -> Accepted
let mut sv_code: i32 = 0;
stream.write_all(&sv_code.to_be_bytes()).await.unwrap();
stream.flush().await.unwrap();
// Starting receiving messages
let mut total_bytes = 0;
let mut n_bytes_read = 0;
let mut chunk_id = 0;
let mut last_chunk = false;
let mut len_slices: [u8; 4] = [0; 4];
loop {
println!("n_bytes read: {:?}", n_bytes_read);
let mut capacity = 65535;
if n_bytes_read == 0 {
capacity = 65539;
}
let mut buffer = Vec::<u8>::with_capacity(capacity);
println!("Blocked?");
match stream.read_buf(&mut buffer).await {
Ok(0) => continue,
Ok(n) => {
println!("N: {:?}", n);
if n_bytes_read == 0 {
for i in 0..4 {
len_slices[i] = buffer.remove(0);
}
total_bytes = u32::from_le_bytes(len_slices);
total_bytes += 4;
}
buffer.truncate(n);
n_bytes_read += n;
if n_bytes_read == total_bytes.try_into().unwrap() {
last_chunk = true;
}
chunk_id += 1;
if last_chunk {
break;
}
}
Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
println!("Err: TCP -> SV (Write))");
continue;
}
Err(e) => {
break;
}
}
}
println!("Finished");
});
}
});
}
Ok(())
}
And the output:
Client:
[client] connected to server: 127.0.0.1:8080
created stream
Client: Reading Buffer: n_bytes 4
Total_len: 262144
Blocked here
Server:
accepted a socket: Ok(127.0.0.1:52430)
Server accept a stream from client: id=1
n_bytes read: 0
Blocked?
N: 65539
n_bytes read: 65539
Blocked?
N: 65535
n_bytes read: 131074
Blocked?
N: 65535
n_bytes read: 196609
Blocked?
N: 65527
n_bytes read: 262136
Blocked?
The error is in the Multiplexer Configuration. Here you can see: tokio_yamux::config::Config, the Config::default() allows a maximum window number of 262144 bytes. This was exceeded within this example, therefore the stream it created over the TcpStream reached the set limit. If a new configuration is created increasing the value of max_stream_window_size this error does not happen anymore.
Solution:
let config = Config {
accept_backlog: 256,
enable_keepalive: true,
keepalive_interval: Duration::from_secs(30),
connection_write_timeout: Duration::from_secs(10),
max_stream_count: 65535,
max_stream_window_size: 4294967295,
};
let mut session = Session::new_server(socket, config);