1 #[cfg(test)] 2 mod association_test; 3 4 mod association_internal; 5 mod association_stats; 6 7 use crate::chunk::chunk_abort::ChunkAbort; 8 use crate::chunk::chunk_cookie_ack::ChunkCookieAck; 9 use crate::chunk::chunk_cookie_echo::ChunkCookieEcho; 10 use crate::chunk::chunk_error::ChunkError; 11 use crate::chunk::chunk_forward_tsn::{ChunkForwardTsn, ChunkForwardTsnStream}; 12 use crate::chunk::chunk_heartbeat::ChunkHeartbeat; 13 use crate::chunk::chunk_heartbeat_ack::ChunkHeartbeatAck; 14 use crate::chunk::chunk_init::ChunkInit; 15 use crate::chunk::chunk_payload_data::{ChunkPayloadData, PayloadProtocolIdentifier}; 16 use crate::chunk::chunk_reconfig::ChunkReconfig; 17 use crate::chunk::chunk_selective_ack::ChunkSelectiveAck; 18 use crate::chunk::chunk_shutdown::ChunkShutdown; 19 use crate::chunk::chunk_shutdown_ack::ChunkShutdownAck; 20 use crate::chunk::chunk_shutdown_complete::ChunkShutdownComplete; 21 use crate::chunk::chunk_type::*; 22 use crate::chunk::Chunk; 23 use crate::error::{Error, Result}; 24 use crate::error_cause::*; 25 use crate::packet::Packet; 26 use crate::param::param_heartbeat_info::ParamHeartbeatInfo; 27 use crate::param::param_outgoing_reset_request::ParamOutgoingResetRequest; 28 use crate::param::param_reconfig_response::{ParamReconfigResponse, ReconfigResult}; 29 use crate::param::param_state_cookie::ParamStateCookie; 30 use crate::param::param_supported_extensions::ParamSupportedExtensions; 31 use crate::param::Param; 32 use crate::queue::control_queue::ControlQueue; 33 use crate::queue::payload_queue::PayloadQueue; 34 use crate::queue::pending_queue::PendingQueue; 35 use crate::stream::*; 36 use crate::timer::ack_timer::*; 37 use crate::timer::rtx_timer::*; 38 use crate::util::*; 39 40 use association_internal::*; 41 use association_stats::*; 42 43 use bytes::{Bytes, BytesMut}; 44 use rand::random; 45 use std::collections::{HashMap, VecDeque}; 46 use std::fmt; 47 use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU8, AtomicUsize, Ordering}; 48 use std::sync::Arc; 49 use std::time::SystemTime; 50 use tokio::sync::{broadcast, mpsc, Mutex, Semaphore}; 51 use util::Conn; 52 53 pub(crate) const RECEIVE_MTU: usize = 8192; 54 /// MTU for inbound packet (from DTLS) 55 pub(crate) const INITIAL_MTU: u32 = 1228; 56 /// initial MTU for outgoing packets (to DTLS) 57 pub(crate) const INITIAL_RECV_BUF_SIZE: u32 = 1024 * 1024; 58 pub(crate) const COMMON_HEADER_SIZE: u32 = 12; 59 pub(crate) const DATA_CHUNK_HEADER_SIZE: u32 = 16; 60 pub(crate) const DEFAULT_MAX_MESSAGE_SIZE: u32 = 65536; 61 62 /// other constants 63 pub(crate) const ACCEPT_CH_SIZE: usize = 16; 64 65 /// association state enums 66 #[derive(Debug, Copy, Clone, PartialEq)] 67 pub(crate) enum AssociationState { 68 Closed = 0, 69 CookieWait = 1, 70 CookieEchoed = 2, 71 Established = 3, 72 ShutdownAckSent = 4, 73 ShutdownPending = 5, 74 ShutdownReceived = 6, 75 ShutdownSent = 7, 76 } 77 78 impl From<u8> for AssociationState { from(v: u8) -> AssociationState79 fn from(v: u8) -> AssociationState { 80 match v { 81 1 => AssociationState::CookieWait, 82 2 => AssociationState::CookieEchoed, 83 3 => AssociationState::Established, 84 4 => AssociationState::ShutdownAckSent, 85 5 => AssociationState::ShutdownPending, 86 6 => AssociationState::ShutdownReceived, 87 7 => AssociationState::ShutdownSent, 88 _ => AssociationState::Closed, 89 } 90 } 91 } 92 93 impl fmt::Display for AssociationState { fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result94 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { 95 let s = match *self { 96 AssociationState::Closed => "Closed", 97 AssociationState::CookieWait => "CookieWait", 98 AssociationState::CookieEchoed => "CookieEchoed", 99 AssociationState::Established => "Established", 100 AssociationState::ShutdownPending => "ShutdownPending", 101 AssociationState::ShutdownSent => "ShutdownSent", 102 AssociationState::ShutdownReceived => "ShutdownReceived", 103 AssociationState::ShutdownAckSent => "ShutdownAckSent", 104 }; 105 write!(f, "{s}") 106 } 107 } 108 109 /// retransmission timer IDs 110 #[derive(Default, Debug, Copy, Clone, PartialEq)] 111 pub(crate) enum RtxTimerId { 112 #[default] 113 T1Init, 114 T1Cookie, 115 T2Shutdown, 116 T3RTX, 117 Reconfig, 118 } 119 120 impl fmt::Display for RtxTimerId { fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result121 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { 122 let s = match *self { 123 RtxTimerId::T1Init => "T1Init", 124 RtxTimerId::T1Cookie => "T1Cookie", 125 RtxTimerId::T2Shutdown => "T2Shutdown", 126 RtxTimerId::T3RTX => "T3RTX", 127 RtxTimerId::Reconfig => "Reconfig", 128 }; 129 write!(f, "{s}") 130 } 131 } 132 133 /// ack mode (for testing) 134 #[derive(Default, Debug, Copy, Clone, PartialEq)] 135 pub(crate) enum AckMode { 136 #[default] 137 Normal, 138 NoDelay, 139 AlwaysDelay, 140 } 141 142 impl fmt::Display for AckMode { fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result143 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { 144 let s = match *self { 145 AckMode::Normal => "Normal", 146 AckMode::NoDelay => "NoDelay", 147 AckMode::AlwaysDelay => "AlwaysDelay", 148 }; 149 write!(f, "{s}") 150 } 151 } 152 153 /// ack transmission state 154 #[derive(Default, Debug, Copy, Clone, PartialEq)] 155 pub(crate) enum AckState { 156 #[default] 157 Idle, // ack timer is off 158 Immediate, // will send ack immediately 159 Delay, // ack timer is on (ack is being delayed) 160 } 161 162 impl fmt::Display for AckState { fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result163 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { 164 let s = match *self { 165 AckState::Idle => "Idle", 166 AckState::Immediate => "Immediate", 167 AckState::Delay => "Delay", 168 }; 169 write!(f, "{s}") 170 } 171 } 172 173 /// Config collects the arguments to create_association construction into 174 /// a single structure 175 pub struct Config { 176 pub net_conn: Arc<dyn Conn + Send + Sync>, 177 pub max_receive_buffer_size: u32, 178 pub max_message_size: u32, 179 pub name: String, 180 } 181 182 ///Association represents an SCTP association 183 ///13.2. Parameters Necessary per Association (i.e., the TCB) 184 ///Peer : Tag value to be sent in every packet and is received 185 ///Verification: in the INIT or INIT ACK chunk. 186 ///Tag : 187 /// 188 ///My : Tag expected in every inbound packet and sent in the 189 ///Verification: INIT or INIT ACK chunk. 190 /// 191 ///Tag : 192 ///State : A state variable indicating what state the association 193 /// : is in, i.e., COOKIE-WAIT, COOKIE-ECHOED, ESTABLISHED, 194 /// : SHUTDOWN-PENDING, SHUTDOWN-SENT, SHUTDOWN-RECEIVED, 195 /// : SHUTDOWN-ACK-SENT. 196 /// 197 /// No Closed state is illustrated since if a 198 /// association is Closed its TCB SHOULD be removed. 199 pub struct Association { 200 name: String, 201 state: Arc<AtomicU8>, 202 max_message_size: Arc<AtomicU32>, 203 inflight_queue_length: Arc<AtomicUsize>, 204 will_send_shutdown: Arc<AtomicBool>, 205 awake_write_loop_ch: Arc<mpsc::Sender<()>>, 206 close_loop_ch_rx: Mutex<broadcast::Receiver<()>>, 207 accept_ch_rx: Mutex<mpsc::Receiver<Arc<Stream>>>, 208 net_conn: Arc<dyn Conn + Send + Sync>, 209 bytes_received: Arc<AtomicUsize>, 210 bytes_sent: Arc<AtomicUsize>, 211 212 pub(crate) association_internal: Arc<Mutex<AssociationInternal>>, 213 } 214 215 impl Association { 216 /// server accepts a SCTP stream over a conn server(config: Config) -> Result<Self>217 pub async fn server(config: Config) -> Result<Self> { 218 let (a, mut handshake_completed_ch_rx) = Association::new(config, false).await?; 219 220 if let Some(err_opt) = handshake_completed_ch_rx.recv().await { 221 if let Some(err) = err_opt { 222 Err(err) 223 } else { 224 Ok(a) 225 } 226 } else { 227 Err(Error::ErrAssociationHandshakeClosed) 228 } 229 } 230 231 /// Client opens a SCTP stream over a conn client(config: Config) -> Result<Self>232 pub async fn client(config: Config) -> Result<Self> { 233 let (a, mut handshake_completed_ch_rx) = Association::new(config, true).await?; 234 235 if let Some(err_opt) = handshake_completed_ch_rx.recv().await { 236 if let Some(err) = err_opt { 237 Err(err) 238 } else { 239 Ok(a) 240 } 241 } else { 242 Err(Error::ErrAssociationHandshakeClosed) 243 } 244 } 245 246 /// Shutdown initiates the shutdown sequence. The method blocks until the 247 /// shutdown sequence is completed and the connection is closed, or until the 248 /// passed context is done, in which case the context's error is returned. shutdown(&self) -> Result<()>249 pub async fn shutdown(&self) -> Result<()> { 250 log::debug!("[{}] closing association..", self.name); 251 252 let state = self.get_state(); 253 if state != AssociationState::Established { 254 return Err(Error::ErrShutdownNonEstablished); 255 } 256 257 // Attempt a graceful shutdown. 258 self.set_state(AssociationState::ShutdownPending); 259 260 if self.inflight_queue_length.load(Ordering::SeqCst) == 0 { 261 // No more outstanding, send shutdown. 262 self.will_send_shutdown.store(true, Ordering::SeqCst); 263 let _ = self.awake_write_loop_ch.try_send(()); 264 self.set_state(AssociationState::ShutdownSent); 265 } 266 267 { 268 let mut close_loop_ch_rx = self.close_loop_ch_rx.lock().await; 269 let _ = close_loop_ch_rx.recv().await; 270 } 271 272 Ok(()) 273 } 274 275 /// Close ends the SCTP Association and cleans up any state close(&self) -> Result<()>276 pub async fn close(&self) -> Result<()> { 277 log::debug!("[{}] closing association..", self.name); 278 279 let _ = self.net_conn.close().await; 280 281 let mut ai = self.association_internal.lock().await; 282 ai.close().await 283 } 284 new(config: Config, is_client: bool) -> Result<(Self, mpsc::Receiver<Option<Error>>)>285 async fn new(config: Config, is_client: bool) -> Result<(Self, mpsc::Receiver<Option<Error>>)> { 286 let net_conn = Arc::clone(&config.net_conn); 287 288 let (awake_write_loop_ch_tx, awake_write_loop_ch_rx) = mpsc::channel(1); 289 let (accept_ch_tx, accept_ch_rx) = mpsc::channel(ACCEPT_CH_SIZE); 290 let (handshake_completed_ch_tx, handshake_completed_ch_rx) = mpsc::channel(1); 291 let (close_loop_ch_tx, close_loop_ch_rx) = broadcast::channel(1); 292 let (close_loop_ch_rx1, close_loop_ch_rx2) = 293 (close_loop_ch_tx.subscribe(), close_loop_ch_tx.subscribe()); 294 let awake_write_loop_ch = Arc::new(awake_write_loop_ch_tx); 295 296 let ai = AssociationInternal::new( 297 config, 298 close_loop_ch_tx, 299 accept_ch_tx, 300 handshake_completed_ch_tx, 301 Arc::clone(&awake_write_loop_ch), 302 ); 303 304 let bytes_received = Arc::new(AtomicUsize::new(0)); 305 let bytes_sent = Arc::new(AtomicUsize::new(0)); 306 let name = ai.name.clone(); 307 let state = Arc::clone(&ai.state); 308 let max_message_size = Arc::clone(&ai.max_message_size); 309 let inflight_queue_length = Arc::clone(&ai.inflight_queue_length); 310 let will_send_shutdown = Arc::clone(&ai.will_send_shutdown); 311 312 let mut init = ChunkInit { 313 initial_tsn: ai.my_next_tsn, 314 num_outbound_streams: ai.my_max_num_outbound_streams, 315 num_inbound_streams: ai.my_max_num_inbound_streams, 316 initiate_tag: ai.my_verification_tag, 317 advertised_receiver_window_credit: ai.max_receive_buffer_size, 318 ..Default::default() 319 }; 320 init.set_supported_extensions(); 321 322 let name1 = name.clone(); 323 let name2 = name.clone(); 324 325 let bytes_received1 = Arc::clone(&bytes_received); 326 let bytes_sent2 = Arc::clone(&bytes_sent); 327 328 let net_conn1 = Arc::clone(&net_conn); 329 let net_conn2 = Arc::clone(&net_conn); 330 331 let association_internal = Arc::new(Mutex::new(ai)); 332 let association_internal1 = Arc::clone(&association_internal); 333 let association_internal2 = Arc::clone(&association_internal); 334 335 { 336 let association_internal3 = Arc::clone(&association_internal); 337 338 let mut ai = association_internal.lock().await; 339 ai.t1init = Some(RtxTimer::new( 340 Arc::downgrade(&association_internal3), 341 RtxTimerId::T1Init, 342 MAX_INIT_RETRANS, 343 )); 344 ai.t1cookie = Some(RtxTimer::new( 345 Arc::downgrade(&association_internal3), 346 RtxTimerId::T1Cookie, 347 MAX_INIT_RETRANS, 348 )); 349 ai.t2shutdown = Some(RtxTimer::new( 350 Arc::downgrade(&association_internal3), 351 RtxTimerId::T2Shutdown, 352 NO_MAX_RETRANS, 353 )); // retransmit forever 354 ai.t3rtx = Some(RtxTimer::new( 355 Arc::downgrade(&association_internal3), 356 RtxTimerId::T3RTX, 357 NO_MAX_RETRANS, 358 )); // retransmit forever 359 ai.treconfig = Some(RtxTimer::new( 360 Arc::downgrade(&association_internal3), 361 RtxTimerId::Reconfig, 362 NO_MAX_RETRANS, 363 )); // retransmit forever 364 ai.ack_timer = Some(AckTimer::new( 365 Arc::downgrade(&association_internal3), 366 ACK_INTERVAL, 367 )); 368 } 369 370 tokio::spawn(async move { 371 Association::read_loop( 372 name1, 373 bytes_received1, 374 net_conn1, 375 close_loop_ch_rx1, 376 association_internal1, 377 ) 378 .await; 379 }); 380 381 tokio::spawn(async move { 382 Association::write_loop( 383 name2, 384 bytes_sent2, 385 net_conn2, 386 close_loop_ch_rx2, 387 association_internal2, 388 awake_write_loop_ch_rx, 389 ) 390 .await; 391 }); 392 393 if is_client { 394 let mut ai = association_internal.lock().await; 395 ai.set_state(AssociationState::CookieWait); 396 ai.stored_init = Some(init); 397 ai.send_init()?; 398 let rto = ai.rto_mgr.get_rto(); 399 if let Some(t1init) = &ai.t1init { 400 t1init.start(rto).await; 401 } 402 } 403 404 Ok(( 405 Association { 406 name, 407 state, 408 max_message_size, 409 inflight_queue_length, 410 will_send_shutdown, 411 awake_write_loop_ch, 412 close_loop_ch_rx: Mutex::new(close_loop_ch_rx), 413 accept_ch_rx: Mutex::new(accept_ch_rx), 414 net_conn, 415 bytes_received, 416 bytes_sent, 417 association_internal, 418 }, 419 handshake_completed_ch_rx, 420 )) 421 } 422 read_loop( name: String, bytes_received: Arc<AtomicUsize>, net_conn: Arc<dyn Conn + Send + Sync>, mut close_loop_ch: broadcast::Receiver<()>, association_internal: Arc<Mutex<AssociationInternal>>, )423 async fn read_loop( 424 name: String, 425 bytes_received: Arc<AtomicUsize>, 426 net_conn: Arc<dyn Conn + Send + Sync>, 427 mut close_loop_ch: broadcast::Receiver<()>, 428 association_internal: Arc<Mutex<AssociationInternal>>, 429 ) { 430 log::debug!("[{}] read_loop entered", name); 431 432 let mut buffer = vec![0u8; RECEIVE_MTU]; 433 let mut done = false; 434 let mut n; 435 while !done { 436 tokio::select! { 437 _ = close_loop_ch.recv() => break, 438 result = net_conn.recv(&mut buffer) => { 439 match result { 440 Ok(m) => { 441 n=m; 442 } 443 Err(err) => { 444 log::warn!("[{}] failed to read packets on net_conn: {}", name, err); 445 break; 446 } 447 } 448 } 449 }; 450 451 // Make a buffer sized to what we read, then copy the data we 452 // read from the underlying transport. We do this because the 453 // user data is passed to the reassembly queue without 454 // copying. 455 log::debug!("[{}] recving {} bytes", name, n); 456 let inbound = Bytes::from(buffer[..n].to_vec()); 457 bytes_received.fetch_add(n, Ordering::SeqCst); 458 459 { 460 let mut ai = association_internal.lock().await; 461 if let Err(err) = ai.handle_inbound(&inbound).await { 462 log::warn!("[{}] failed to handle_inbound: {:?}", name, err); 463 done = true; 464 } 465 } 466 } 467 468 { 469 let mut ai = association_internal.lock().await; 470 if let Err(err) = ai.close().await { 471 log::warn!("[{}] failed to close association: {:?}", name, err); 472 } 473 } 474 475 log::debug!("[{}] read_loop exited", name); 476 } 477 write_loop( name: String, bytes_sent: Arc<AtomicUsize>, net_conn: Arc<dyn Conn + Send + Sync>, mut close_loop_ch: broadcast::Receiver<()>, association_internal: Arc<Mutex<AssociationInternal>>, mut awake_write_loop_ch: mpsc::Receiver<()>, )478 async fn write_loop( 479 name: String, 480 bytes_sent: Arc<AtomicUsize>, 481 net_conn: Arc<dyn Conn + Send + Sync>, 482 mut close_loop_ch: broadcast::Receiver<()>, 483 association_internal: Arc<Mutex<AssociationInternal>>, 484 mut awake_write_loop_ch: mpsc::Receiver<()>, 485 ) { 486 log::debug!("[{}] write_loop entered", name); 487 let done = Arc::new(AtomicBool::new(false)); 488 let name = Arc::new(name); 489 490 let limit = { 491 #[cfg(test)] 492 { 493 1 494 } 495 #[cfg(not(test))] 496 { 497 8 498 } 499 }; 500 501 let sem = Arc::new(Semaphore::new(limit)); 502 while !done.load(Ordering::Relaxed) { 503 //log::debug!("[{}] gather_outbound begin", name); 504 let (packets, continue_loop) = { 505 let mut ai = association_internal.lock().await; 506 ai.gather_outbound().await 507 }; 508 //log::debug!("[{}] gather_outbound done with {}", name, packets.len()); 509 510 // We schedule a new task here for a reason: 511 // If we don't tokio tends to run the write_loop and read_loop of one connection on the same OS thread 512 // This means that even though we release the lock above, the read_loop isn't able to take it, simply because it is not being scheduled by tokio 513 // Doing it this way, tokio schedules this to a new thread, this future is suspended, and the read_loop can make progress 514 let net_conn = Arc::clone(&net_conn); 515 let bytes_sent = Arc::clone(&bytes_sent); 516 let name2 = Arc::clone(&name); 517 let done2 = Arc::clone(&done); 518 let sem = Arc::clone(&sem); 519 sem.acquire().await.unwrap().forget(); 520 tokio::task::spawn(async move { 521 let mut buf = BytesMut::with_capacity(16 * 1024); 522 for raw in packets { 523 buf.clear(); 524 if let Err(err) = raw.marshal_to(&mut buf) { 525 log::warn!("[{}] failed to serialize a packet: {:?}", name2, err); 526 } else { 527 let raw = buf.as_ref(); 528 if let Err(err) = net_conn.send(raw.as_ref()).await { 529 log::warn!("[{}] failed to write packets on net_conn: {}", name2, err); 530 done2.store(true, Ordering::Relaxed) 531 } else { 532 bytes_sent.fetch_add(raw.len(), Ordering::SeqCst); 533 } 534 } 535 //log::debug!("[{}] sending {} bytes done", name, raw.len()); 536 } 537 sem.add_permits(1); 538 }); 539 540 if !continue_loop { 541 break; 542 } 543 544 //log::debug!("[{}] wait awake_write_loop_ch", name); 545 tokio::select! { 546 _ = awake_write_loop_ch.recv() =>{} 547 _ = close_loop_ch.recv() => { 548 done.store(true, Ordering::Relaxed); 549 } 550 }; 551 //log::debug!("[{}] wait awake_write_loop_ch done", name); 552 } 553 554 { 555 let mut ai = association_internal.lock().await; 556 if let Err(err) = ai.close().await { 557 log::warn!("[{}] failed to close association: {:?}", name, err); 558 } 559 } 560 561 log::debug!("[{}] write_loop exited", name); 562 } 563 564 /// bytes_sent returns the number of bytes sent bytes_sent(&self) -> usize565 pub fn bytes_sent(&self) -> usize { 566 self.bytes_sent.load(Ordering::SeqCst) 567 } 568 569 /// bytes_received returns the number of bytes received bytes_received(&self) -> usize570 pub fn bytes_received(&self) -> usize { 571 self.bytes_received.load(Ordering::SeqCst) 572 } 573 574 /// open_stream opens a stream open_stream( &self, stream_identifier: u16, default_payload_type: PayloadProtocolIdentifier, ) -> Result<Arc<Stream>>575 pub async fn open_stream( 576 &self, 577 stream_identifier: u16, 578 default_payload_type: PayloadProtocolIdentifier, 579 ) -> Result<Arc<Stream>> { 580 let mut ai = self.association_internal.lock().await; 581 ai.open_stream(stream_identifier, default_payload_type) 582 } 583 584 /// accept_stream accepts a stream accept_stream(&self) -> Option<Arc<Stream>>585 pub async fn accept_stream(&self) -> Option<Arc<Stream>> { 586 let mut accept_ch_rx = self.accept_ch_rx.lock().await; 587 accept_ch_rx.recv().await 588 } 589 590 /// max_message_size returns the maximum message size you can send. max_message_size(&self) -> u32591 pub fn max_message_size(&self) -> u32 { 592 self.max_message_size.load(Ordering::SeqCst) 593 } 594 595 /// set_max_message_size sets the maximum message size you can send. set_max_message_size(&self, max_message_size: u32)596 pub fn set_max_message_size(&self, max_message_size: u32) { 597 self.max_message_size 598 .store(max_message_size, Ordering::SeqCst); 599 } 600 601 /// set_state atomically sets the state of the Association. set_state(&self, new_state: AssociationState)602 fn set_state(&self, new_state: AssociationState) { 603 let old_state = AssociationState::from(self.state.swap(new_state as u8, Ordering::SeqCst)); 604 if new_state != old_state { 605 log::debug!( 606 "[{}] state change: '{}' => '{}'", 607 self.name, 608 old_state, 609 new_state, 610 ); 611 } 612 } 613 614 /// get_state atomically returns the state of the Association. get_state(&self) -> AssociationState615 fn get_state(&self) -> AssociationState { 616 self.state.load(Ordering::SeqCst).into() 617 } 618 } 619