async fn flow_write_task(
        flow_id: FlowId,
        flow_label: String,
        mut send_queue_reader: Box<dyn SendQueueReader + Send + Sync>,
        mut writer: Box<TlsWriteHalf>,
        metrics: DataPlaneMetrics,
        state: Weak<TransportImpl>,
    ) {
        let _updater = MetricsUpdater::new(metrics.clone(), true);
        let flow_tag = flow_id.flow_tag.to_string();
        loop {
            let loop_start_time = Instant::now();
            // If the TransportImpl has been deleted, abort.
            let state = match state.upgrade() {
                Some(transport) => transport,
                _ => return,
            };
            // Wait for the send requests
            let dequeued = send_queue_reader
                .dequeue(
                    DEQUEUE_BYTES,
                    Duration::from_millis(TRANSPORT_HEARTBEAT_SEND_INTERVAL_MS),
                )
                .await;

            let mut to_send = Vec::<u8>::new();
            if dequeued.is_empty() {
                // There is nothing to send, so issue a heartbeat message
                to_send.append(&mut Self::pack_header(None, false, true));
                state
                    .data_plane_metrics
                    .heart_beats_sent
                    .with_label_values(&[&flow_label, &flow_tag])
                    .inc();
            } else {
                for mut msg in dequeued {
                    to_send.append(&mut Self::pack_header(
                        Some(&msg.payload),
                        msg.sender_error,
                        false,
                    ));
                    to_send.append(&mut msg.payload.0);
                }
            }
            state
                .data_plane_metrics
                .write_task_overhead_time_msec
                .with_label_values(&[&flow_label, &flow_tag])
                .observe(loop_start_time.elapsed().as_millis() as f64);

            // Send the payload
            let start_time = Instant::now();
            if let Err(e) = writer.write_all(&to_send).await {
                warn!(
                    state.log,
                    "DataPlane::flow_write_task(): failed to write payload: flow: {:?}, {:?}",
                    flow_id,
                    e,
                );
                state.on_disconnect(flow_id).await;
                return;
            }
            // Flush the write
            if let Err(e) = writer.flush().await {
                warn!(
                    state.log,
                    "DataPlane::flow_write_task(): failed to flush: flow: {:?}, {:?}", flow_id, e,
                );
                state.on_disconnect(flow_id).await;
                return;
            }

            state
                .data_plane_metrics
                .socket_write_time_msec
                .with_label_values(&[&flow_label, &flow_tag])
                .observe(start_time.elapsed().as_millis() as f64);
            state
                .data_plane_metrics
                .socket_write_bytes
                .with_label_values(&[&flow_label, &flow_tag])
                .inc_by(to_send.len() as u64);
            state
                .data_plane_metrics
                .socket_write_size
                .with_label_values(&[&flow_label, &flow_tag])
                .observe(to_send.len() as f64);
        }
    }
/// Per-flow send task. Reads the requests from the send queue and writes to
/// the socket.
Per-flow send task. Reads the requests from the send queue and writes to
the socket.
Per - flow send task . Reads the requests from the send queue and writes to the socket .
async fn flow_write_task(
        flow_id: FlowId,
        flow_label: String,
        mut send_queue_reader: Box<dyn SendQueueReader + Send + Sync>,
        mut writer: Box<TlsWriteHalf>,
        metrics: DataPlaneMetrics,
        state: Weak<TransportImpl>,
    ) {
        let _updater = MetricsUpdater::new(metrics.clone(), true);
        let flow_tag = flow_id.flow_tag.to_string();
        loop {
            let loop_start_time = Instant::now();
            // If the TransportImpl has been deleted, abort.
            let state = match state.upgrade() {
                Some(transport) => transport,
                _ => return,
            };
            // Wait for the send requests
            let dequeued = send_queue_reader
                .dequeue(
                    DEQUEUE_BYTES,
                    Duration::from_millis(TRANSPORT_HEARTBEAT_SEND_INTERVAL_MS),
                )
                .await;
            let mut to_send = Vec::<u8>::new();
            if dequeued.is_empty() {
                // There is nothing to send, so issue a heartbeat message
                to_send.append(&mut Self::pack_header(None, false, true));
                state
                    .data_plane_metrics
                    .heart_beats_sent
                    .with_label_values(&[&flow_label, &flow_tag])
                    .inc();
            } else {
                for mut msg in dequeued {
                    to_send.append(&mut Self::pack_header(
                        Some(&msg.payload),
                        msg.sender_error,
                        false,
                    ));
                    to_send.append(&mut msg.payload.0);
                }
            }
            state
                .data_plane_metrics
                .write_task_overhead_time_msec
                .with_label_values(&[&flow_label, &flow_tag])
                .observe(loop_start_time.elapsed().as_millis() as f64);
            // Send the payload
            let start_time = Instant::now();
            if let Err(e) = writer.write_all(&to_send).await {
                warn!(
                    state.log,
                    "DataPlane::flow_write_task(): failed to write payload: flow: {:?}, {:?}",
                    flow_id,
                    e,
                );
                state.on_disconnect(flow_id).await;
                return;
            }
            // Flush the write
            if let Err(e) = writer.flush().await {
                warn!(
                    state.log,
                    "DataPlane::flow_write_task(): failed to flush: flow: {:?}, {:?}", flow_id, e,
                );
                state.on_disconnect(flow_id).await;
                return;
            }
            state
                .data_plane_metrics
                .socket_write_time_msec
                .with_label_values(&[&flow_label, &flow_tag])
                .observe(start_time.elapsed().as_millis() as f64);
            state
                .data_plane_metrics
                .socket_write_bytes
                .with_label_values(&[&flow_label, &flow_tag])
                .inc_by(to_send.len() as u64);
            state
                .data_plane_metrics
                .socket_write_size
                .with_label_values(&[&flow_label, &flow_tag])
                .observe(to_send.len() as f64);
        }
    }
async fn flow_write_task ( flow_id : FlowId , flow_label : String , mut send_queue_reader : Box < dyn SendQueueReader + Send + Sync > , mut writer : Box < TlsWriteHalf > , metrics : DataPlaneMetrics , state : Weak < TransportImpl > , ) { let _updater = MetricsUpdater :: new ( metrics . clone ( ) , true ) ; let flow_tag = flow_id . flow_tag . to_string ( ) ; loop { let loop_start_time = Instant :: now ( ) ; // If the TransportImpl has been deleted, abort. let state = match state . upgrade ( ) { Some ( transport ) => transport , _ => return , } ; // Wait for the send requests let dequeued = send_queue_reader . dequeue ( DEQUEUE_BYTES , Duration :: from_millis ( TRANSPORT_HEARTBEAT_SEND_INTERVAL_MS ) , ) . await ; let mut to_send = Vec :: < u8 > :: new ( ) ; if dequeued . is_empty ( ) { // There is nothing to send, so issue a heartbeat message to_send . append ( & mut Self :: pack_header ( None , false , true ) ) ; state . data_plane_metrics . heart_beats_sent . with_label_values ( & [ & flow_label , & flow_tag ] ) . inc ( ) ; } else { for mut msg in dequeued { to_send . append ( & mut Self :: pack_header ( Some ( & msg . payload ) , msg . sender_error , false , ) ) ; to_send . append ( & mut msg . payload . 0 ) ; } } state . data_plane_metrics . write_task_overhead_time_msec . with_label_values ( & [ & flow_label , & flow_tag ] ) . observe ( loop_start_time . elapsed ( ) . as_millis ( ) as f64 ) ; // Send the payload let start_time = Instant :: now ( ) ; if let Err ( e ) = writer . write_all ( & to_send ) . await { warn ! ( state log " " flow_id e ) ; state . on_disconnect ( flow_id ) . await ; return ; } // Flush the write if let Err ( e ) = writer . flush ( ) . await { warn ! ( state log " " flow_id e ) ; state . on_disconnect ( flow_id ) . await ; return ; } state . data_plane_metrics . socket_write_time_msec . with_label_values ( & [ & flow_label , & flow_tag ] ) . observe ( start_time . elapsed ( ) . as_millis ( ) as f64 ) ; state . data_plane_metrics . socket_write_bytes . with_label_values ( & [ & flow_label , & flow_tag ] ) . inc_by ( to_send . len ( ) as u64 ) ; state . data_plane_metrics . socket_write_size . with_label_values ( & [ & flow_label , & flow_tag ] ) . observe ( to_send . len ( ) as f64 ) ; } }
Per-flow send task.
Per - flow send task .
flow_id
FlowId
flow_label
String
send_queue_reader
Box<dyn SendQueueReader + Send + Sync>
writer
Box<TlsWriteHalf>
metrics
DataPlaneMetrics
state
Weak<TransportImpl>
flow_id
FlowId
flow_label
String
send_queue_reader
Box<dyn SendQueueReader + Send + Sync>
writer
Box<TlsWriteHalf>
metrics
DataPlaneMetrics
state
Weak<TransportImpl>