From 06ebeefb102b223ff77dd47e16b64bd9575f7c34 Mon Sep 17 00:00:00 2001 From: Garrett D'Amore Date: Thu, 26 Dec 2024 10:20:33 -0800 Subject: aio: introduce NNG_ESTOPPED This error code results when an AIO is stopped permanently, as a result of nni_aio_close or nni_aio_stop. The associated AIO object cannot be used again. This discrimantes against a file being closed, or a temporary cancellation which might allow the aio to be reused. Consumers must check for this error status in their callbacks, and not resubmit an operation that failed with this error. Doing so, will result in an infinite loop of submit / errors. --- src/sp/protocol/pubsub0/pub_test.c | 102 ------------------------------------ src/sp/protocol/pubsub0/sub_test.c | 29 ++++++++-- src/sp/protocol/pubsub0/xsub_test.c | 18 +++++++ 3 files changed, 43 insertions(+), 106 deletions(-) (limited to 'src/sp/protocol/pubsub0') diff --git a/src/sp/protocol/pubsub0/pub_test.c b/src/sp/protocol/pubsub0/pub_test.c index a2a20dd7..8b3a5d78 100644 --- a/src/sp/protocol/pubsub0/pub_test.c +++ b/src/sp/protocol/pubsub0/pub_test.c @@ -160,103 +160,6 @@ test_pub_send_queued(void) NUTS_CLOSE(pub); NUTS_CLOSE(sub); } -static void -test_sub_recv_ctx_closed(void) -{ - nng_socket sub; - nng_ctx ctx; - nng_aio *aio; - NUTS_PASS(nng_sub0_open(&sub)); - NUTS_PASS(nng_ctx_open(&ctx, sub)); - NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL)); - nng_ctx_close(ctx); - nng_ctx_recv(ctx, aio); - nng_aio_wait(aio); - NUTS_FAIL(nng_aio_result(aio), NNG_ECLOSED); - nng_aio_free(aio); - NUTS_CLOSE(sub); -} - -static void -test_sub_ctx_recv_aio_stopped(void) -{ - nng_socket sub; - nng_ctx ctx; - nng_aio *aio; - - NUTS_PASS(nng_sub0_open(&sub)); - NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL)); - NUTS_PASS(nng_ctx_open(&ctx, sub)); - - nng_aio_stop(aio); - nng_ctx_recv(ctx, aio); - nng_aio_wait(aio); - NUTS_FAIL(nng_aio_result(aio), NNG_ECANCELED); - NUTS_PASS(nng_ctx_close(ctx)); - NUTS_CLOSE(sub); - nng_aio_free(aio); -} - -static void -test_sub_close_context_recv(void) -{ - nng_socket sub; - nng_ctx ctx; - nng_aio *aio; - - NUTS_PASS(nng_sub0_open(&sub)); - NUTS_PASS(nng_ctx_open(&ctx, sub)); - NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL)); - nng_aio_set_timeout(aio, 1000); - nng_ctx_recv(ctx, aio); - NUTS_PASS(nng_ctx_close(ctx)); - nng_aio_wait(aio); - NUTS_FAIL(nng_aio_result(aio), NNG_ECLOSED); - - NUTS_CLOSE(sub); - nng_aio_free(aio); -} - -static void -test_sub_ctx_recv_nonblock(void) -{ - nng_socket sub; - nng_ctx ctx; - nng_aio *aio; - - NUTS_PASS(nng_sub0_open(&sub)); - NUTS_PASS(nng_ctx_open(&ctx, sub)); - NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL)); - - nng_aio_set_timeout(aio, 0); // Instant timeout - nng_ctx_recv(ctx, aio); - - nng_aio_wait(aio); - NUTS_FAIL(nng_aio_result(aio), NNG_ETIMEDOUT); - NUTS_CLOSE(sub); - nng_aio_free(aio); -} - -static void -test_sub_ctx_recv_cancel(void) -{ - nng_socket sub; - nng_ctx ctx; - nng_aio *aio; - - NUTS_PASS(nng_sub0_open(&sub)); - NUTS_PASS(nng_ctx_open(&ctx, sub)); - NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL)); - - nng_aio_set_timeout(aio, 1000); - nng_ctx_recv(ctx, aio); - nng_aio_abort(aio, NNG_ECANCELED); - - nng_aio_wait(aio); - NUTS_FAIL(nng_aio_result(aio), NNG_ECANCELED); - NUTS_CLOSE(sub); - nng_aio_free(aio); -} static void test_pub_send_buf_option(void) @@ -308,11 +211,6 @@ NUTS_TESTS = { { "pub validate peer", test_pub_validate_peer }, { "pub send queued", test_pub_send_queued }, { "pub send no pipes", test_pub_send_no_pipes }, - { "sub recv ctx closed", test_sub_recv_ctx_closed }, - { "sub recv aio ctx stopped", test_sub_ctx_recv_aio_stopped }, - { "sub close context recv", test_sub_close_context_recv }, - { "sub context recv nonblock", test_sub_ctx_recv_nonblock }, - { "sub context recv cancel", test_sub_ctx_recv_cancel }, { "pub send buf option", test_pub_send_buf_option }, { "pub cooked", test_pub_cooked }, { NULL, NULL }, diff --git a/src/sp/protocol/pubsub0/sub_test.c b/src/sp/protocol/pubsub0/sub_test.c index 74a547f4..6899d47d 100644 --- a/src/sp/protocol/pubsub0/sub_test.c +++ b/src/sp/protocol/pubsub0/sub_test.c @@ -223,6 +223,26 @@ test_sub_ctx_recv_aio_stopped(void) nng_aio_stop(aio); nng_ctx_recv(ctx, aio); nng_aio_wait(aio); + NUTS_FAIL(nng_aio_result(aio), NNG_ESTOPPED); + NUTS_PASS(nng_ctx_close(ctx)); + NUTS_CLOSE(sub); + nng_aio_free(aio); +} + +static void +test_sub_ctx_recv_aio_canceled(void) +{ + nng_socket sub; + nng_ctx ctx; + nng_aio *aio; + + NUTS_PASS(nng_sub0_open(&sub)); + NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL)); + NUTS_PASS(nng_ctx_open(&ctx, sub)); + + nng_ctx_recv(ctx, aio); + nng_aio_cancel(aio); + nng_aio_wait(aio); NUTS_FAIL(nng_aio_result(aio), NNG_ECANCELED); NUTS_PASS(nng_ctx_close(ctx)); NUTS_CLOSE(sub); @@ -270,7 +290,7 @@ test_sub_ctx_recv_nonblock(void) } static void -test_sub_ctx_recv_cancel(void) +test_sub_ctx_recv_abort(void) { nng_socket sub; nng_ctx ctx; @@ -282,10 +302,10 @@ test_sub_ctx_recv_cancel(void) nng_aio_set_timeout(aio, 1000); nng_ctx_recv(ctx, aio); - nng_aio_abort(aio, NNG_ECANCELED); + nng_aio_abort(aio, NNG_EAMBIGUOUS); nng_aio_wait(aio); - NUTS_FAIL(nng_aio_result(aio), NNG_ECANCELED); + NUTS_FAIL(nng_aio_result(aio), NNG_EAMBIGUOUS); NUTS_CLOSE(sub); nng_aio_free(aio); } @@ -593,9 +613,10 @@ TEST_LIST = { { "sub recv late", test_sub_recv_late }, { "sub recv ctx closed", test_sub_recv_ctx_closed }, { "sub recv aio ctx stopped", test_sub_ctx_recv_aio_stopped }, + { "sub recv aio ctx canceled", test_sub_ctx_recv_aio_canceled }, { "sub close context recv", test_sub_close_context_recv }, { "sub context recv nonblock", test_sub_ctx_recv_nonblock }, - { "sub context recv cancel", test_sub_ctx_recv_cancel }, + { "sub context recv abort", test_sub_ctx_recv_abort }, { "sub recv buf option", test_sub_recv_buf_option }, { "sub subscribe option", test_sub_subscribe_option }, { "sub unsubscribe option", test_sub_unsubscribe_option }, diff --git a/src/sp/protocol/pubsub0/xsub_test.c b/src/sp/protocol/pubsub0/xsub_test.c index eb918e14..fd26467d 100644 --- a/src/sp/protocol/pubsub0/xsub_test.c +++ b/src/sp/protocol/pubsub0/xsub_test.c @@ -337,6 +337,23 @@ test_xsub_recv_aio_stopped(void) nng_aio_stop(aio); nng_recv_aio(sub, aio); nng_aio_wait(aio); + NUTS_FAIL(nng_aio_result(aio), NNG_ESTOPPED); + NUTS_CLOSE(sub); + nng_aio_free(aio); +} + +static void +test_xsub_recv_aio_canceled(void) +{ + nng_socket sub; + nng_aio *aio; + + NUTS_PASS(nng_sub0_open_raw(&sub)); + NUTS_PASS(nng_aio_alloc(&aio, NULL, NULL)); + + nng_recv_aio(sub, aio); + nng_aio_cancel(aio); + nng_aio_wait(aio); NUTS_FAIL(nng_aio_result(aio), NNG_ECANCELED); NUTS_CLOSE(sub); nng_aio_free(aio); @@ -358,6 +375,7 @@ TEST_LIST = { { "xsub no context", test_xsub_no_context }, { "xsub raw", test_xsub_raw }, { "xsub recv aio stopped", test_xsub_recv_aio_stopped }, + { "xsub recv aio canceled", test_xsub_recv_aio_canceled }, { "xsub close during recv ", test_xsub_close_during_recv }, { "xsub close during pipe recv", test_xsub_close_during_pipe_recv }, { NULL, NULL }, -- cgit v1.2.3-70-g09d2