Add common handling for userdata drop/borrow semantics everywhere

Author: bossmc

capi/examples/client.c

@@ -184,7 +184,7 @@ int main(int argc, char *argv[]) {
     hyper_clientconn_options_exec(opts, exec);
 
     hyper_task *handshake = hyper_clientconn_handshake(io, opts);
-    hyper_task_set_userdata(handshake, (void *)EXAMPLE_HANDSHAKE);
+    hyper_task_set_userdata(handshake, (void *)EXAMPLE_HANDSHAKE, NULL);
 
     // Let's wait for the handshake to finish...
     hyper_executor_push(exec, handshake);
@@ -230,7 +230,7 @@ int main(int argc, char *argv[]) {
 
                 // Send it!
                 hyper_task *send = hyper_clientconn_send(client, req);
-                hyper_task_set_userdata(send, (void *)EXAMPLE_SEND);
+                hyper_task_set_userdata(send, (void *)EXAMPLE_SEND, NULL);
                 printf("sending ...\n");
                 hyper_executor_push(exec, send);
 
@@ -261,8 +261,8 @@ int main(int argc, char *argv[]) {
                 printf("\n");
 
                 hyper_body *resp_body = hyper_response_body(resp);
-                hyper_task *foreach = hyper_body_foreach(resp_body, print_each_chunk, NULL);
-                hyper_task_set_userdata(foreach, (void *)EXAMPLE_RESP_BODY);
+                hyper_task *foreach = hyper_body_foreach(resp_body, print_each_chunk, NULL, NULL);
+                hyper_task_set_userdata(foreach, (void *)EXAMPLE_RESP_BODY, NULL);
                 hyper_executor_push(exec, foreach);
 
                 // No longer need the response

capi/examples/upload.c

@@ -224,7 +224,7 @@ int main(int argc, char *argv[]) {
     hyper_clientconn_options_exec(opts, exec);
 
     hyper_task *handshake = hyper_clientconn_handshake(io, opts);
-    hyper_task_set_userdata(handshake, (void *)EXAMPLE_HANDSHAKE);
+    hyper_task_set_userdata(handshake, (void *)EXAMPLE_HANDSHAKE, NULL);
 
     // Let's wait for the handshake to finish...
     hyper_executor_push(exec, handshake);
@@ -274,17 +274,17 @@ int main(int argc, char *argv[]) {
                 // the body is sent immediately. This will just print if any
                 // informational headers are received.
                 printf("    with expect-continue ...\n");
-                hyper_request_on_informational(req, print_informational, NULL);
+                hyper_request_on_informational(req, print_informational, NULL, NULL);
 
                 // Prepare the req body
                 hyper_body *body = hyper_body_new();
-                hyper_body_set_userdata(body, &upload);
+                hyper_body_set_userdata(body, &upload, NULL);
                 hyper_body_set_data_func(body, poll_req_upload);
                 hyper_request_set_body(req, body);
 
                 // Send it!
                 hyper_task *send = hyper_clientconn_send(client, req);
-                hyper_task_set_userdata(send, (void *)EXAMPLE_SEND);
+                hyper_task_set_userdata(send, (void *)EXAMPLE_SEND, NULL);
                 printf("sending ...\n");
                 hyper_executor_push(exec, send);
 
@@ -315,7 +315,7 @@ int main(int argc, char *argv[]) {
 
                 // Set us up to peel data from the body a chunk at a time
                 hyper_task *body_data = hyper_body_data(resp_body);
-                hyper_task_set_userdata(body_data, (void *)EXAMPLE_RESP_BODY);
+                hyper_task_set_userdata(body_data, (void *)EXAMPLE_RESP_BODY, NULL);
                 hyper_executor_push(exec, body_data);
 
                 // No longer need the response
@@ -335,7 +335,7 @@ int main(int argc, char *argv[]) {
                     hyper_task_free(task);
 
                     hyper_task *body_data = hyper_body_data(resp_body);
-                    hyper_task_set_userdata(body_data, (void *)EXAMPLE_RESP_BODY);
+                    hyper_task_set_userdata(body_data, (void *)EXAMPLE_RESP_BODY, NULL);
                     hyper_executor_push(exec, body_data);
 
                     break;

capi/include/hyper.h

@@ -233,6 +233,20 @@ typedef struct hyper_waker hyper_waker;
 
 typedef int (*hyper_body_foreach_callback)(void*, const struct hyper_buf*);
 
+/*
+ Many hyper entites can be given userdata to allow user callbacks to corellate work together.
+ Since much of hyper is asychronous it's often useful to treat these userdata objects as "owned"
+ by the hyper entity (and hence to be cleaned up when that entity is dropped).
+
+ To acheive this a `hyepr_userdata_drop` callback is passed by calling code alongside the
+ userdata to register a cleanup function.
+
+ This function may be provided as NULL if the calling code wants to manage memory lifetimes
+ itself, in which case the hyper object will logically consider the userdata "borrowed" until
+ the hyper entity is dropped.
+ */
+typedef void (*hyper_userdata_drop)(void*);
+
 typedef int (*hyper_body_data_callback)(void*, struct hyper_context*, struct hyper_buf**);
 
 /*
@@ -252,20 +266,10 @@ typedef int (*hyper_body_data_callback)(void*, struct hyper_context*, struct hyp
  */
 typedef void (*hyper_service_callback)(void*, struct hyper_request*, struct hyper_response_channel*);
 
-/*
- Since a hyper_service owns the userdata passed to it the calling code can register a cleanup
- function to be called when the service is dropped.  This function may be omitted/a no-op if
- the calling code wants to manage memory lifetimes itself (e.g. if the userdata is a static
- global)
- */
-typedef void (*hyper_service_userdata_drop_callback)(void*);
-
 typedef void (*hyper_request_on_informational_callback)(void*, struct hyper_response*);
 
 typedef int (*hyper_headers_foreach_callback)(void*, const uint8_t*, size_t, const uint8_t*, size_t);
 
-typedef void (*hyper_io_userdata_drop_callback)(void*);
-
 typedef size_t (*hyper_io_read_callback)(void*, struct hyper_context*, uint8_t*, size_t);
 
 typedef size_t (*hyper_io_write_callback)(void*, struct hyper_context*, const uint8_t*, size_t);
@@ -319,12 +323,13 @@ struct hyper_task *hyper_body_data(struct hyper_body *body);
  */
 struct hyper_task *hyper_body_foreach(struct hyper_body *body,
                                       hyper_body_foreach_callback func,
-                                      void *userdata);
+                                      void *userdata,
+                                      hyper_userdata_drop drop);
 
 /*
  Set userdata on this body, which will be passed to callback functions.
  */
-void hyper_body_set_userdata(struct hyper_body *body, void *userdata);
+void hyper_body_set_userdata(struct hyper_body *body, void *userdata, hyper_userdata_drop drop);
 
 /*
  Set the data callback for this body.
@@ -659,12 +664,13 @@ struct hyper_service *hyper_service_new(hyper_service_callback service_fn);
 
  The service takes ownership of the userdata and will call the `drop_userdata` callback when
  the service task is complete.  If the `drop_userdata` callback is `NULL` then the service
- will instead forget the userdata when the associated task is completed and the calling code
- is responsible for cleaning up the userdata through some other mechanism.
+ will instead borrow the userdata and forget it when the associated task is completed and
+ thus the calling code is responsible for cleaning up the userdata through some other
+ mechanism.
  */
 void hyper_service_set_userdata(struct hyper_service *service,
                                 void *userdata,
-                                hyper_service_userdata_drop_callback drop_func);
+                                hyper_userdata_drop drop);
 
 /*
  Frees a hyper_service object if no longer needed
@@ -896,7 +902,8 @@ struct hyper_body *hyper_request_body(struct hyper_request *req);
  */
 enum hyper_code hyper_request_on_informational(struct hyper_request *req,
                                                hyper_request_on_informational_callback callback,
-                                               void *data);
+                                               void *data,
+                                               hyper_userdata_drop drop);
 
 /*
  Construct a new HTTP 200 Ok response
@@ -1047,9 +1054,7 @@ void hyper_io_free(struct hyper_io *io);
  `hyper_io` is destroyed (either explicitely by `hyper_io_free` or
  implicitely by an associated hyper task completing).
  */
-void hyper_io_set_userdata(struct hyper_io *io,
-                           void *data,
-                           hyper_io_userdata_drop_callback drop_func);
+void hyper_io_set_userdata(struct hyper_io *io, void *data, hyper_userdata_drop drop_func);
 
 /*
  Get the user data pointer for this IO value.
@@ -1161,7 +1166,7 @@ enum hyper_task_return_type hyper_task_type(struct hyper_task *task);
  This value will be passed to task callbacks, and can be checked later
  with `hyper_task_userdata`.
  */
-void hyper_task_set_userdata(struct hyper_task *task, void *userdata);
+void hyper_task_set_userdata(struct hyper_task *task, void *userdata, hyper_userdata_drop drop);
 
 /*
  Retrieve the userdata that has been set via `hyper_task_set_userdata`.

src/ffi/body.rs

@@ -7,7 +7,8 @@ use http_body_util::BodyExt as _;
 use libc::{c_int, size_t};
 
 use super::task::{hyper_context, hyper_task, hyper_task_return_type, AsTaskType};
-use super::{UserDataPointer, HYPER_ITER_CONTINUE};
+use super::HYPER_ITER_CONTINUE;
+use super::userdata::{Userdata, hyper_userdata_drop};
 use crate::body::{Bytes, Frame, Incoming as IncomingBody};
 
 /// A streaming HTTP body.
@@ -18,7 +19,7 @@ pub struct hyper_buf(pub(super) Bytes);
 
 pub(crate) struct UserBody {
     data_func: hyper_body_data_callback,
-    userdata: *mut c_void,
+    userdata: Userdata,
 }
 
 // ===== Body =====
@@ -88,15 +89,15 @@ ffi_fn! {
     /// chunks as they are received, or `HYPER_ITER_BREAK` to cancel.
     ///
     /// This will consume the `hyper_body *`, you shouldn't use it anymore or free it.
-    fn hyper_body_foreach(body: *mut hyper_body, func: hyper_body_foreach_callback, userdata: *mut c_void) -> *mut hyper_task {
+    fn hyper_body_foreach(body: *mut hyper_body, func: hyper_body_foreach_callback, userdata: *mut c_void, drop: hyper_userdata_drop) -> *mut hyper_task {
         let mut body = non_null!(Box::from_raw(body) ?= ptr::null_mut());
-        let userdata = UserDataPointer(userdata);
+        let userdata = Userdata::new(userdata, drop);
 
         Box::into_raw(hyper_task::boxed(async move {
             while let Some(item) = body.0.frame().await {
                 let frame = item?;
                 if let Ok(chunk) = frame.into_data() {
-                    if HYPER_ITER_CONTINUE != func(userdata.0, &hyper_buf(chunk)) {
+                    if HYPER_ITER_CONTINUE != func(userdata.as_ptr(), &hyper_buf(chunk)) {
                         return Err(crate::Error::new_user_aborted_by_callback());
                     }
                 }
@@ -108,9 +109,9 @@ ffi_fn! {
 
 ffi_fn! {
     /// Set userdata on this body, which will be passed to callback functions.
-    fn hyper_body_set_userdata(body: *mut hyper_body, userdata: *mut c_void) {
+    fn hyper_body_set_userdata(body: *mut hyper_body, userdata: *mut c_void, drop: hyper_userdata_drop) {
         let b = non_null!(&mut *body ?= ());
-        b.0.as_ffi_mut().userdata = userdata;
+        b.0.as_ffi_mut().userdata = Userdata::new(userdata, drop);
     }
 }
 
@@ -146,7 +147,7 @@ impl UserBody {
     pub(crate) fn new() -> UserBody {
         UserBody {
             data_func: data_noop,
-            userdata: std::ptr::null_mut(),
+            userdata: Userdata::default(),
         }
     }
 
@@ -155,7 +156,7 @@ impl UserBody {
         cx: &mut Context<'_>,
     ) -> Poll<Option<crate::Result<Frame<Bytes>>>> {
         let mut out = std::ptr::null_mut();
-        match (self.data_func)(self.userdata, hyper_context::wrap(cx), &mut out) {
+        match (self.data_func)(self.userdata.as_ptr(), hyper_context::wrap(cx), &mut out) {
             super::task::HYPER_POLL_READY => {
                 if out.is_null() {
                     Poll::Ready(None)

src/ffi/http_types.rs

@@ -5,7 +5,8 @@ use std::ffi::c_void;
 use super::body::hyper_body;
 use super::error::hyper_code;
 use super::task::{hyper_task_return_type, AsTaskType};
-use super::{UserDataPointer, HYPER_ITER_CONTINUE};
+use super::HYPER_ITER_CONTINUE;
+use super::userdata::{Userdata, hyper_userdata_drop};
 use crate::body::Incoming as IncomingBody;
 use crate::ext::{HeaderCaseMap, OriginalHeaderOrder, ReasonPhrase};
 use crate::header::{HeaderName, HeaderValue};
@@ -28,7 +29,7 @@ pub struct hyper_headers {
 
 pub(crate) struct OnInformational {
     func: hyper_request_on_informational_callback,
-    data: UserDataPointer,
+    userdata: Userdata,
 }
 
 type hyper_request_on_informational_callback = extern "C" fn(*mut c_void, *mut hyper_response);
@@ -337,10 +338,10 @@ ffi_fn! {
     /// NOTE: The `hyper_response *` is just borrowed data, and will not
     /// be valid after the callback finishes. You must copy any data you wish
     /// to persist.
-    fn hyper_request_on_informational(req: *mut hyper_request, callback: hyper_request_on_informational_callback, data: *mut c_void) -> hyper_code {
+    fn hyper_request_on_informational(req: *mut hyper_request, callback: hyper_request_on_informational_callback, data: *mut c_void, drop: hyper_userdata_drop) -> hyper_code {
         let ext = OnInformational {
             func: callback,
-            data: UserDataPointer(data),
+            userdata: Userdata::new(data, drop),
         };
         let req = non_null!(&mut *req ?= hyper_code::HYPERE_INVALID_ARG);
         req.0.extensions_mut().insert(ext);
@@ -724,7 +725,7 @@ unsafe fn raw_name_value(
 impl OnInformational {
     pub(crate) fn call(&mut self, resp: Response<IncomingBody>) {
         let mut resp = hyper_response::from(resp);
-        (self.func)(self.data.0, &mut resp);
+        (self.func)(self.userdata.as_ptr(), &mut resp);
     }
 }
 

src/ffi/io.rs

@@ -6,6 +6,7 @@ use libc::size_t;
 use tokio::io::{AsyncRead, AsyncWrite};
 
 use super::task::hyper_context;
+use super::userdata::Userdata;
 
 /// Sentinel value to return from a read or write callback that the operation
 /// is pending.
@@ -18,14 +19,12 @@ type hyper_io_read_callback =
     extern "C" fn(*mut c_void, *mut hyper_context<'_>, *mut u8, size_t) -> size_t;
 type hyper_io_write_callback =
     extern "C" fn(*mut c_void, *mut hyper_context<'_>, *const u8, size_t) -> size_t;
-type hyper_io_userdata_drop_callback = extern "C" fn(*mut c_void);
 
 /// An IO object used to represent a socket or similar concept.
 pub struct hyper_io {
     read: hyper_io_read_callback,
     write: hyper_io_write_callback,
-    userdata: *mut c_void,
-    userdata_drop: Option<hyper_io_userdata_drop_callback>,
+    userdata: Userdata,
 }
 
 ffi_fn! {
@@ -37,8 +36,7 @@ ffi_fn! {
         Box::into_raw(Box::new(hyper_io {
             read: read_noop,
             write: write_noop,
-            userdata: std::ptr::null_mut(),
-            userdata_drop: None,
+            userdata: Userdata::default(),
         }))
     } ?= std::ptr::null_mut()
 }
@@ -64,15 +62,10 @@ ffi_fn! {
     fn hyper_io_set_userdata(
         io: *mut hyper_io,
         data: *mut c_void,
-        drop_func: hyper_io_userdata_drop_callback,
+        drop_func: super::userdata::hyper_userdata_drop,
     ) {
         let io = non_null!(&mut *io? = ());
-        io.userdata = data;
-        io.userdata_drop = if (drop_func as *const c_void).is_null() {
-            None
-        } else {
-            Some(drop_func)
-        }
+        io.userdata = Userdata::new(data, drop_func);
     }
 }
 
@@ -83,7 +76,7 @@ ffi_fn! {
     ///
     /// Returns NULL if no userdata has been set.
     fn hyper_io_get_userdata(io: *mut hyper_io) -> *mut c_void {
-        non_null!(&mut *io ?= std::ptr::null_mut()).userdata
+        non_null!(&mut *io ?= std::ptr::null_mut()).userdata.as_ptr()
     }
 }
 
@@ -148,14 +141,6 @@ extern "C" fn write_noop(
     0
 }
 
-impl Drop for hyper_io {
-    fn drop(&mut self) {
-        if let Some(drop_fn) = self.userdata_drop {
-            drop_fn(self.userdata)
-        }
-    }
-}
-
 impl AsyncRead for hyper_io {
     fn poll_read(
         self: Pin<&mut Self>,
@@ -165,7 +150,7 @@ impl AsyncRead for hyper_io {
         let buf_ptr = unsafe { buf.unfilled_mut() }.as_mut_ptr() as *mut u8;
         let buf_len = buf.remaining();
 
-        match (self.read)(self.userdata, hyper_context::wrap(cx), buf_ptr, buf_len) {
+        match (self.read)(self.userdata.as_ptr(), hyper_context::wrap(cx), buf_ptr, buf_len) {
             HYPER_IO_PENDING => Poll::Pending,
             HYPER_IO_ERROR => Poll::Ready(Err(std::io::Error::new(
                 std::io::ErrorKind::Other,
@@ -191,7 +176,7 @@ impl AsyncWrite for hyper_io {
         let buf_ptr = buf.as_ptr();
         let buf_len = buf.len();
 
-        match (self.write)(self.userdata, hyper_context::wrap(cx), buf_ptr, buf_len) {
+        match (self.write)(self.userdata.as_ptr(), hyper_context::wrap(cx), buf_ptr, buf_len) {
             HYPER_IO_PENDING => Poll::Pending,
             HYPER_IO_ERROR => Poll::Ready(Err(std::io::Error::new(
                 std::io::ErrorKind::Other,