diff options
| -rw-r--r-- | include/uv_link_t.h | 38 | ||||
| -rw-r--r-- | src/uv_link_t.c | 80 |
2 files changed, 91 insertions, 27 deletions
diff --git a/include/uv_link_t.h b/include/uv_link_t.h index 5e77e34..34c7760 100644 --- a/include/uv_link_t.h +++ b/include/uv_link_t.h @@ -64,6 +64,7 @@ struct uv_link_methods_s { uv_link_read_cb saved_read_cb; \ \ /* Private, used for close */ \ + int close_depth; \ int close_waiting; \ uv_link_close_cb saved_close_cb; @@ -85,32 +86,23 @@ UV_EXTERN void uv_link_propagate_read_cb(uv_link_t* link, ssize_t nread, const uv_buf_t* buf); -static int uv_link_propagate_write(uv_link_t* link, uv_link_t* source, - const uv_buf_t bufs[], unsigned int nbufs, - uv_stream_t* send_handle, - uv_link_write_cb cb, void* arg) { - return link->methods->write(link, source, bufs, nbufs, send_handle, cb, arg); -} +int uv_link_propagate_write(uv_link_t* link, uv_link_t* source, + const uv_buf_t bufs[], unsigned int nbufs, + uv_stream_t* send_handle, + uv_link_write_cb cb, void* arg); -static int uv_link_propagate_shutdown(uv_link_t* link, - uv_link_t* source, - uv_link_shutdown_cb cb, - void* arg) { - return link->methods->shutdown(link, source, cb, arg); -} +int uv_link_propagate_shutdown(uv_link_t* link, + uv_link_t* source, + uv_link_shutdown_cb cb, + void* arg); void uv_link_propagate_close(uv_link_t* link, uv_link_t* source, uv_link_close_cb cb); /* Use this to invoke methods of `link` */ -static int uv_link_read_start(uv_link_t* link) { - return link->methods->read_start(link); -} - -static int uv_link_read_stop(uv_link_t* link) { - return link->methods->read_stop(link); -} +int uv_link_read_start(uv_link_t* link); +int uv_link_read_stop(uv_link_t* link); static int uv_link_write(uv_link_t* link, const uv_buf_t bufs[], unsigned int nbufs, uv_stream_t* send_handle, @@ -118,11 +110,9 @@ static int uv_link_write(uv_link_t* link, const uv_buf_t bufs[], return uv_link_propagate_write(link, link, bufs, nbufs, send_handle, cb, arg); } -static int uv_link_try_write(uv_link_t* link, - const uv_buf_t bufs[], - unsigned int nbufs) { - return link->methods->try_write(link, bufs, nbufs); -} +int uv_link_try_write(uv_link_t* link, + const uv_buf_t bufs[], + unsigned int nbufs); static int uv_link_shutdown(uv_link_t* link, uv_link_shutdown_cb cb, void* arg) { diff --git a/src/uv_link_t.c b/src/uv_link_t.c index 11bacb2..3ee91b9 100644 --- a/src/uv_link_t.c +++ b/src/uv_link_t.c @@ -3,6 +3,8 @@ #include "src/common.h" +static void uv_link_maybe_close(uv_link_t* link); + static void uv_link_def_alloc_cb(uv_link_t* link, size_t suggested_size, uv_buf_t* buf) { @@ -33,9 +35,48 @@ int uv_link_init(uv_link_t* link, uv_link_methods_t const* methods) { } -static void uv_link_close_join(uv_link_t* link) { - if (--link->close_waiting == 0) - return link->saved_close_cb(link); +#define CLOSE_WRAP(RES) \ + do { \ + int err; \ + link->close_depth++; \ + err = (RES); \ + if (--link->close_depth == 0) \ + uv_link_maybe_close(link); \ + return err; \ + } while (0) + + +int uv_link_propagate_write(uv_link_t* link, uv_link_t* source, + const uv_buf_t bufs[], unsigned int nbufs, + uv_stream_t* send_handle, + uv_link_write_cb cb, void* arg) { + CLOSE_WRAP(link->methods->write(link, source, bufs, nbufs, send_handle, cb, + arg)); +} + + +int uv_link_propagate_shutdown(uv_link_t* link, + uv_link_t* source, + uv_link_shutdown_cb cb, + void* arg) { + CLOSE_WRAP(link->methods->shutdown(link, source, cb, arg)); +} + + +int uv_link_read_start(uv_link_t* link) { + CLOSE_WRAP(link->methods->read_start(link)); +} + + +int uv_link_read_stop(uv_link_t* link) { + CLOSE_WRAP(link->methods->read_stop(link)); +} + + +int uv_link_try_write(uv_link_t* link, + const uv_buf_t bufs[], + unsigned int nbufs) { + CLOSE_WRAP(link->methods->try_write(link, bufs, nbufs)); } @@ -44,6 +85,25 @@ void uv_link_close(uv_link_t* link, uv_link_close_cb cb) { } +void uv_link_maybe_close(uv_link_t* link) { + uv_link_close_cb cb; + + if (link->saved_close_cb == NULL) + return; + + cb = link->saved_close_cb; + + link->saved_close_cb = NULL; + return uv_link_propagate_close(link, link, cb); +} + + +static void uv_link_close_join(uv_link_t* link) { + if (--link->close_waiting == 0) + return link->saved_close_cb(link); +} + + void uv_link_propagate_close(uv_link_t* link, uv_link_t* source, uv_link_close_cb cb) { uv_link_t* root; @@ -51,6 +111,14 @@ void uv_link_propagate_close(uv_link_t* link, uv_link_t* source, CHECK_EQ(link->child, NULL, "uv_link_t: attempt to close chained link"); + /* We are in an API call, wait for it to end before destroying everything */ + if (link->close_depth != 0) { + CHECK_EQ(link, source, "pending close_cb for non-leaf link"); + + link->saved_close_cb = cb; + return; + } + /* Find root */ count = 1; for (root = link; root->parent != NULL; root = root->parent) @@ -122,7 +190,10 @@ void uv_link_propagate_alloc_cb(uv_link_t* link, if (link->child != NULL) target = link->child; + link->close_depth++; link->alloc_cb(target, suggested_size, buf); + if (--link->close_depth == 0) + uv_link_maybe_close(link); } @@ -135,5 +206,8 @@ void uv_link_propagate_read_cb(uv_link_t* link, if (link->child != NULL) target = link->child; + link->close_depth++; link->read_cb(target, nread, buf); + if (--link->close_depth == 0) + uv_link_maybe_close(link); } |