Send a request to the Data Plane, awaiting if the queue is full. This is the primary async API for the Control Plane. It: 1. Tries to push immediately. 2. If full, awaits the eventfd signal from the TPC core (space freed). 3. Retries the push.
(&mut self, req: Req)
| 82 | /// 2. If full, awaits the eventfd signal from the TPC core (space freed). |
| 83 | /// 3. Retries the push. |
| 84 | pub async fn send_request(&mut self, req: Req) -> Result<()> |
| 85 | where |
| 86 | Req: Clone, |
| 87 | { |
| 88 | // Fast path: try immediate push. |
| 89 | match self.inner.try_send_request(req.clone()) { |
| 90 | Ok(()) => return Ok(()), |
| 91 | Err(BridgeError::Full { .. }) => {} |
| 92 | Err(e) => return Err(e), |
| 93 | } |
| 94 | |
| 95 | // Slow path: wait for space. |
| 96 | loop { |
| 97 | // Wait for the eventfd to become readable (TPC core freed a slot). |
| 98 | let mut guard = |
| 99 | self.req_space_fd |
| 100 | .readable() |
| 101 | .await |
| 102 | .map_err(|_| BridgeError::Backpressure { |
| 103 | percent: 100, |
| 104 | threshold: 95, |
| 105 | })?; |
| 106 | |
| 107 | // Consume the eventfd signal. |
| 108 | let _ = self.inner.req_wake.producer_wake.try_read(); |
| 109 | guard.clear_ready(); |
| 110 | |
| 111 | // Retry push. |
| 112 | match self.inner.try_send_request(req.clone()) { |
| 113 | Ok(()) => return Ok(()), |
| 114 | Err(BridgeError::Full { .. }) => continue, // Spurious wake, retry. |
| 115 | Err(e) => return Err(e), |
| 116 | } |
| 117 | } |
| 118 | } |
| 119 | |
| 120 | /// Receive a response from the Data Plane, awaiting if none available. |
| 121 | pub async fn recv_response(&mut self) -> Result<Rsp> { |
no test coverage detected