backend/curl handle pause related errors (#46)

* backend/curl handle pause related errors

* backend/curl wake from loop when unpause fails

Author: bdbai

backends/curl/src/async.rs

@@ -84,7 +84,7 @@ impl AsyncResponse for CurlAsyncResponse {
         let mut buf = vec![];
         while let Some(()) = this
             .handle
-            .poll_bytes_async(|data| {
+            .wait_for_bytes(|data| {
                 if let Some(max_response_buffer_size) = this.max_response_buffer_size {
                     if buf.len() + data.len() > max_response_buffer_size as usize {
                         return Err(NyquestError::ResponseTooLarge);

backends/curl/src/async/handler.rs

@@ -32,16 +32,15 @@ impl EasyCallback for AsyncHandler {
             // ... signals an error condition to the library and returns CURLE_WRITE_ERROR.
             return Ok(0);
         };
-        {
-            let mut state = inner.ctx.state.lock().unwrap();
-            let state = &mut state.state;
+        let mut state = inner.ctx.state.lock().unwrap();
+        let state = &mut state.state;
+        inner.ctx.waker.wake();
+        if state.data_available() {
+            Err(WriteError::Pause)
+        } else {
             state.write_data(data);
+            Ok(data.len())
         }
-        unsafe {
-            inner.pause.pause_recv();
-        }
-        inner.ctx.waker.wake();
-        Ok(data.len())
     }
 
     fn header(&mut self, data: &[u8]) -> bool {
@@ -54,7 +53,9 @@ impl EasyCallback for AsyncHandler {
             let state = &mut state.state;
             if state.push_header_data(data) {
                 unsafe {
-                    inner.pause.pause_recv();
+                    if inner.pause.pause_recv().is_err() {
+                        return false;
+                    }
                 }
             }
         }

backends/curl/src/async/loop.rs

@@ -19,6 +19,7 @@ pub(super) struct RequestHandle {
     id: usize,
     shared_state: Arc<SharedRequestStates>,
     manager: LoopManagerShared,
+    is_recv_unpause_sent: bool,
 }
 
 enum LoopTask {
@@ -79,7 +80,7 @@ impl RequestHandle {
         .await
     }
 
-    pub(super) async fn poll_bytes_async<T>(
+    pub(super) async fn wait_for_bytes<T>(
         &mut self,
         cb: impl FnOnce(&mut Vec<u8>) -> nyquest_interface::Result<T>,
     ) -> nyquest_interface::Result<Option<T>> {
@@ -100,16 +101,20 @@ impl RequestHandle {
         cb: impl FnOnce(&mut Vec<u8>) -> nyquest_interface::Result<T>,
     ) -> Poll<nyquest_interface::Result<Option<T>>> {
         let mut state = self.shared_state.state.lock().unwrap();
-        if !state.state.response_buffer.is_empty() {
+        if state.state.data_available() {
             let res = cb(&mut state.state.response_buffer);
+            self.is_recv_unpause_sent = false;
             return Poll::Ready(res.map(Some));
         }
         if let RequestResult::Done { res, .. } = std::mem::take(&mut state.result) {
+            self.is_recv_unpause_sent = false;
             return Poll::Ready(res.map(|_| None));
         };
-        self.manager
-            .dispatch_task(LoopTask::UnpauseRecvHandle(self.id))
-            .ok();
+        if !std::mem::replace(&mut self.is_recv_unpause_sent, true) {
+            self.manager
+                .dispatch_task(LoopTask::UnpauseRecvHandle(self.id))
+                .ok();
+        }
         self.shared_state.waker.register(cx.waker());
         Poll::Pending
     }
@@ -244,6 +249,7 @@ impl LoopManager {
                         id: res,
                         shared_state,
                         manager: inner,
+                        is_recv_unpause_sent: false,
                     })
                 }
                 Err(res) => res,
@@ -333,7 +339,6 @@ fn run_loop(multl_waker_tx: oneshot::Sender<LoopManagerShared>) {
     {
         return;
     }
-    // TODO: store ctx in Easy2Handle
     let mut tasks = Default::default();
     let mut last_call = false;
     loop {
@@ -408,20 +413,32 @@ fn run_loop(multl_waker_tx: oneshot::Sender<LoopManagerShared>) {
                     };
                 }
                 LoopTask::UnpauseRecvHandle(id) => {
-                    if let Some(easy) = multi.lookup(id) {
-                        // Ignore the error. Also see
-                        // https://github.com/sagebind/isahc/blob/9d1edd475231ad5cfd5842d939db1382dc3a88f5/src/agent/mod.rs#L432
-                        easy.as_raw_easy_mut().unpause_recv().ok();
+                    let Some(mut handle) = multi.lookup(id) else {
+                        continue;
+                    };
+                    let ctx = handle.as_callback_mut().ctx.clone();
+                    if let Err(e) =
+                        handle.with_error_message(|e| e.as_raw_easy_mut().unpause_recv())
+                    {
+                        let res = Err(e.into());
+                        ctx.state.lock().unwrap().result = RequestResult::Done { res, id };
+                        ctx.waker.wake();
                     }
                 }
                 LoopTask::UnpauseSendHandle(id) => {
-                    if let Some(mut easy) = multi.lookup(id) {
+                    let Some(mut handle) = multi.lookup(id) else {
+                        continue;
+                    };
+                    let ctx = handle.as_callback_mut().ctx.clone();
+                    if let Err(e) = handle.with_error_message(|mut e| {
                         // curl seems buggy with unsized upload multipart streams (i.e. chunked)
                         // Pausing and unpausing again seems to make it work
-                        easy.as_mut().as_raw_easy_mut().pause_send().ok();
-                        // Ignore the error. Also see
-                        // https://github.com/sagebind/isahc/blob/9d1edd475231ad5cfd5842d939db1382dc3a88f5/src/agent/mod.rs#L432
-                        easy.as_raw_easy_mut().unpause_send().ok();
+                        e.as_mut().as_raw_easy_mut().pause_send()?;
+                        e.as_raw_easy_mut().unpause_send()
+                    }) {
+                        let res = Err(e.into());
+                        ctx.state.lock().unwrap().result = RequestResult::Done { res, id };
+                        ctx.waker.wake();
                     }
                 }
                 LoopTask::DropHandle(id) => {

backends/curl/src/async/pause.rs

@@ -1,5 +1,7 @@
 use curl_sys::{CURLPAUSE_RECV, CURLPAUSE_SEND};
 
+use crate::curl_ng::{CurlCodeContext, WithCurlCodeContext};
+
 pub const _CURLPAUSE_ALL: i32 = CURLPAUSE_RECV | CURLPAUSE_SEND;
 
 #[derive(Clone, Copy)]
@@ -14,8 +16,9 @@ impl EasyPause {
     /// The caller must ensure:
     /// 1. The handle is a valid CURL handle.
     /// 2. The handle is either within the same thread or we are in a callback.
-    pub(super) unsafe fn pause_recv(&self) {
-        curl_sys::curl_easy_pause(self.0, CURLPAUSE_RECV);
+    pub(super) unsafe fn pause_recv(&self) -> Result<(), CurlCodeContext> {
+        curl_sys::curl_easy_pause(self.0, CURLPAUSE_RECV)
+            .with_easy_context("curl_easy_pause recv in callback")
     }
 }
 

backends/curl/src/state.rs

@@ -42,4 +42,9 @@ impl RequestState {
         // TODO: handle max response buffer size
         self.response_buffer.extend_from_slice(data);
     }
+
+    #[cfg(feature = "async")]
+    pub(crate) fn data_available(&self) -> bool {
+        !self.response_buffer.is_empty()
+    }
 }