diff --git a/man/rules/meson.build b/man/rules/meson.build index f08e26a1cf6..68aeb749392 100644 --- a/man/rules/meson.build +++ b/man/rules/meson.build @@ -974,7 +974,11 @@ manpages = [ 'HAVE_PAM'], ['sd_varlink_call_and_upgrade', '3', - ['sd_varlink_call_and_upgradeb', 'sd_varlink_call_and_upgradebo'], + ['sd_varlink_call_and_upgradeb', + 'sd_varlink_call_and_upgradebo', + 'sd_varlink_invoke_and_upgrade', + 'sd_varlink_invoke_and_upgradeb', + 'sd_varlink_invoke_and_upgradebo'], ''], ['sd_varlink_connect_address', '3', diff --git a/man/sd_varlink_call_and_upgrade.xml b/man/sd_varlink_call_and_upgrade.xml index 48eb9de93f7..710f39fe8e0 100644 --- a/man/sd_varlink_call_and_upgrade.xml +++ b/man/sd_varlink_call_and_upgrade.xml @@ -19,6 +19,9 @@ sd_varlink_call_and_upgrade sd_varlink_call_and_upgradeb sd_varlink_call_and_upgradebo + sd_varlink_invoke_and_upgrade + sd_varlink_invoke_and_upgradeb + sd_varlink_invoke_and_upgradebo Invoke a Varlink method and take over the connection for a raw protocol @@ -59,6 +62,27 @@ int *ret_output_fd … + + + int sd_varlink_invoke_and_upgrade + sd_varlink *link + const char *method + sd_json_variant *parameters + + + + int sd_varlink_invoke_and_upgradeb + sd_varlink *link + const char *method + … + + + + int sd_varlink_invoke_and_upgradebo + sd_varlink *link + const char *method + … + @@ -111,14 +135,36 @@ relationship mirrors that of sd_varlink_call3 and its sd_varlink_callb()/sd_varlink_callbo() variants. + + sd_varlink_invoke_and_upgrade() is the asynchronous counterpart of + sd_varlink_call_and_upgrade(), for connections attached to an event loop with + sd_varlink_attach_event3. + It enqueues the method call with the upgrade request and returns immediately. Once the reply arrives it + is passed to the reply callback installed with + sd_varlink_bind_reply3. + If the server replied with a Varlink error, the connection stays in the Varlink protocol and may be used + further. Otherwise the connection is upgraded on the next dispatch after the reply callback returned + (i.e. the next invocation of sd_varlink_process(), typically by the event loop): + it is detached from the event loop, disconnected from the Varlink + protocol, and the underlying file descriptors are passed to the upgrade callback installed with + sd_varlink_bind_upgrade3, + with the same semantics as the descriptors returned by sd_varlink_call_and_upgrade(). + The callback takes ownership of both descriptors. Unlike with regular method calls, no further method + calls may be enqueued on the connection while an upgrade request is pending, since the connection leaves + the Varlink protocol once the request has been answered. sd_varlink_invoke_and_upgradeb() and + sd_varlink_invoke_and_upgradebo() relate to sd_varlink_invoke_and_upgrade() + like the …b() and …bo() variants described above. Return Value - On success, these functions return a positive integer if the connection was upgraded, or zero if the - server returned a Varlink error and ret_error_id was set to it. On failure, they - return a negative errno-style error code. + On success, sd_varlink_call_and_upgrade() and its variants return a positive + integer if the connection was upgraded. If the server returns a Varlink error and + ret_error_id was provided by the caller, it is set to the error identifier and zero + is returned. Otherwise, the Varlink error is treated as a failure. + sd_varlink_invoke_and_upgrade() and its variants return zero once the request has + been enqueued. On failure, all functions return a negative errno-style error code. Errors @@ -138,6 +184,16 @@ The Varlink connection object is not connected. + + -EBUSY + + Another method call is pending on the connection. Note that while a request + enqueued with sd_varlink_invoke_and_upgrade() is pending, this error is also + returned by any other function that enqueues a method call on the connection, such as + sd_varlink_call3 + or sd_varlink_invoke(). + + -EPROTO @@ -161,6 +217,9 @@ sd_varlink_call_and_upgrade() was added in version 261. sd_varlink_call_and_upgradeb() and sd_varlink_call_and_upgradebo() were added in version 262. + sd_varlink_invoke_and_upgrade(), + sd_varlink_invoke_and_upgradeb() and + sd_varlink_invoke_and_upgradebo() were added in version 263. diff --git a/src/libsystemd/libsystemd.sym b/src/libsystemd/libsystemd.sym index 42228115a31..aea64f65673 100644 --- a/src/libsystemd/libsystemd.sym +++ b/src/libsystemd/libsystemd.sym @@ -1117,4 +1117,6 @@ LIBSYSTEMD_263 { global: sd_device_new_from_device_node_fd; sd_device_trigger_with_args; + sd_varlink_invoke_and_upgrade; + sd_varlink_invoke_and_upgradeb; } LIBSYSTEMD_262; diff --git a/src/libsystemd/sd-varlink/sd-varlink.c b/src/libsystemd/sd-varlink/sd-varlink.c index aa8192f861b..4a6426bf06e 100644 --- a/src/libsystemd/sd-varlink/sd-varlink.c +++ b/src/libsystemd/sd-varlink/sd-varlink.c @@ -51,6 +51,7 @@ static const char* const varlink_state_table[_VARLINK_STATE_MAX] = { [VARLINK_IDLE_CLIENT] = "idle-client", [VARLINK_AWAITING_REPLY] = "awaiting-reply", [VARLINK_AWAITING_REPLY_MORE] = "awaiting-reply-more", + [VARLINK_AWAITING_REPLY_UPGRADE] = "awaiting-reply-upgrade", [VARLINK_CALLING] = "calling", [VARLINK_CALLED] = "called", [VARLINK_COLLECTING] = "collecting", @@ -104,7 +105,7 @@ static JsonStreamPhase varlink_phase(void *userdata) { /* Client side reading a reply with the per-call deadline in force. */ if (IN_SET(v->state, - VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, + VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_AWAITING_REPLY_UPGRADE, VARLINK_CALLING, VARLINK_COLLECTING) && !v->current) return JSON_STREAM_PHASE_AWAITING_REPLY; @@ -736,7 +737,7 @@ static int varlink_write(sd_varlink *v) { static int varlink_read(sd_varlink *v) { assert(v); - if (!IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_CALLING, VARLINK_COLLECTING, VARLINK_IDLE_SERVER)) + if (!IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_AWAITING_REPLY_UPGRADE, VARLINK_CALLING, VARLINK_COLLECTING, VARLINK_IDLE_SERVER)) return 0; if (v->current) return 0; @@ -861,7 +862,7 @@ static int varlink_dispatch_reply(sd_varlink *v) { assert(v); - if (!IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_CALLING, VARLINK_COLLECTING)) + if (!IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_AWAITING_REPLY_UPGRADE, VARLINK_CALLING, VARLINK_COLLECTING)) return 0; if (!v->current) return 0; @@ -917,7 +918,9 @@ static int varlink_dispatch_reply(sd_varlink *v) { v->current_reply_flags = flags; - if (IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE)) { + if (IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_AWAITING_REPLY_UPGRADE)) { + bool upgrade = v->state == VARLINK_AWAITING_REPLY_UPGRADE; + varlink_set_state(v, VARLINK_PROCESSING_REPLY); if (v->reply_callback) { @@ -931,6 +934,24 @@ static int varlink_dispatch_reply(sd_varlink *v) { if (v->state == VARLINK_PROCESSING_REPLY) { assert(v->n_pending > 0); + if (upgrade) { + assert(v->n_pending == 1); + v->n_pending--; + + if (error) { + /* The server refused the upgrade, hence the connection stays in the + * Varlink protocol and can be used further. */ + json_stream_set_flags(&v->stream, JSON_STREAM_BOUNDED_READS, false); + varlink_set_state(v, VARLINK_IDLE_CLIENT); + } else + /* The server accepted the upgrade. The fds are handed to the upgrade + * callback on the next dispatch, see varlink_dispatch_upgrade(), so + * that we don't invoke two callbacks in one go. */ + varlink_set_state(v, VARLINK_UPGRADING); + + return 1; + } + if (!FLAGS_SET(flags, SD_VARLINK_REPLY_CONTINUES)) v->n_pending--; @@ -1505,6 +1526,10 @@ static int varlink_dispatch_upgrade(sd_varlink *v) { assert(v); + /* Completes a protocol upgrade, both for servers (after sd_varlink_respond_and_upgrade()) and + * clients (after the reply to sd_varlink_invoke_and_upgrade() arrived): hands the connection's fds + * to the upgrade callback, once everything has been written out. */ + if (v->state != VARLINK_UPGRADING) return 0; @@ -1963,6 +1988,64 @@ _public_ int sd_varlink_invokeb(sd_varlink *v, const char *method, ...) { return sd_varlink_invoke(v, method, parameters); } +_public_ int sd_varlink_invoke_and_upgrade(sd_varlink *v, const char *method, sd_json_variant *parameters) { + _cleanup_(sd_json_variant_unrefp) sd_json_variant *m = NULL; + int r; + + assert_return(v, -EINVAL); + assert_return(method, -EINVAL); + + if (v->state == VARLINK_DISCONNECTED) + return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected."); + + /* Unlike for regular method calls we do not allow pipelining here: once the reply arrives the + * connection leaves the Varlink protocol, hence no other calls may be pending. */ + if (v->state != VARLINK_IDLE_CLIENT) + return varlink_log_errno(v, SYNTHETIC_ERRNO(EBUSY), "Connection busy."); + + assert(v->n_pending == 0); /* n_pending can't be > 0 if we are in VARLINK_IDLE_CLIENT state */ + + r = sd_json_buildo( + &m, + SD_JSON_BUILD_PAIR_STRING("method", method), + JSON_BUILD_PAIR_VARIANT_NON_EMPTY("parameters", parameters), + SD_JSON_BUILD_PAIR_BOOLEAN("upgrade", true)); + if (r < 0) + return varlink_log_errno(v, r, "Failed to build json message: %m"); + + /* Make sure we don't consume any data of the upgraded protocol that might follow the reply */ + json_stream_set_flags(&v->stream, JSON_STREAM_BOUNDED_READS, true); + + r = varlink_enqueue(v, m); + if (r < 0) { + json_stream_set_flags(&v->stream, JSON_STREAM_BOUNDED_READS, false); + return varlink_log_errno(v, r, "Failed to enqueue json message: %m"); + } + + varlink_set_state(v, VARLINK_AWAITING_REPLY_UPGRADE); + v->n_pending++; + json_stream_mark_activity(&v->stream); + + return 0; +} + +_public_ int sd_varlink_invoke_and_upgradeb(sd_varlink *v, const char *method, ...) { + _cleanup_(sd_json_variant_unrefp) sd_json_variant *parameters = NULL; + va_list ap; + int r; + + assert_return(v, -EINVAL); + assert_return(method, -EINVAL); + + va_start(ap, method); + r = sd_json_buildv(¶meters, ap); + va_end(ap); + if (r < 0) + return varlink_log_errno(v, r, "Failed to build json message: %m"); + + return sd_varlink_invoke_and_upgrade(v, method, parameters); +} + _public_ int sd_varlink_observe(sd_varlink *v, const char *method, sd_json_variant *parameters) { _cleanup_(sd_json_variant_unrefp) sd_json_variant *m = NULL; int r; diff --git a/src/libsystemd/sd-varlink/test-varlink.c b/src/libsystemd/sd-varlink/test-varlink.c index 7d4a1b29065..96d1f51ffa4 100644 --- a/src/libsystemd/sd-varlink/test-varlink.c +++ b/src/libsystemd/sd-varlink/test-varlink.c @@ -1282,6 +1282,142 @@ TEST(upgrade) { ASSERT_OK(sd_future_result(f)); } +typedef struct AsyncUpgradeTest { + sd_event *event; + bool got_reply, got_upgrade, echo_done, got_error; + int input_fd, output_fd; +} AsyncUpgradeTest; + +static int async_upgrade_echo_fiber(void *arg) { + AsyncUpgradeTest *t = ASSERT_PTR(arg); + + /* Same exchange as in upgrade_client_fiber(), over the fds donated to the upgrade callback */ + static const char msg[] = "Hello!"; + ASSERT_OK(loop_write(t->output_fd, msg, strlen(msg))); + ASSERT_OK_ERRNO(shutdown(t->output_fd, SHUT_WR)); + + char buf[64] = {}; + ssize_t n = ASSERT_OK(loop_read(t->input_fd, buf, strlen(msg), /* do_poll= */ true)); + ASSERT_EQ((size_t) n, strlen(msg)); + ASSERT_STREQ(buf, "!olleH"); + + t->input_fd = safe_close(t->input_fd); + t->output_fd = safe_close(t->output_fd); + t->echo_done = true; + + ASSERT_OK(sd_event_exit(t->event, EXIT_SUCCESS)); + return 0; +} + +static int async_upgrade_on_reply(sd_varlink *link, sd_json_variant *parameters, const char *error_id, sd_varlink_reply_flags_t flags, void *userdata) { + AsyncUpgradeTest *t = ASSERT_PTR(userdata); + + ASSERT_FALSE(t->got_upgrade); /* the reply callback must run before the upgrade callback */ + ASSERT_NULL(error_id); + ASSERT_FALSE(FLAGS_SET(flags, SD_VARLINK_REPLY_ERROR)); + t->got_reply = true; + return 0; +} + +static int async_upgrade_on_upgrade(sd_varlink *link, int input_fd, int output_fd, void *userdata) { + AsyncUpgradeTest *t = ASSERT_PTR(userdata); + + ASSERT_TRUE(t->got_reply); + ASSERT_GE(input_fd, 0); + ASSERT_GE(output_fd, 0); + ASSERT_NE(input_fd, output_fd); + ASSERT_FALSE(sd_varlink_is_connected(link)); + + t->got_upgrade = true; + t->input_fd = input_fd; + t->output_fd = output_fd; + + /* The raw I/O blocks, hence do it in a fiber so that the server can make progress */ + _cleanup_(sd_future_unrefp) sd_future *f = NULL; + ASSERT_OK(sd_fiber_new(t->event, "async-upgrade-echo", async_upgrade_echo_fiber, t, &f)); + ASSERT_OK(sd_fiber_set_floating(f, true)); + return 0; +} + +static int async_upgrade_on_error_reply(sd_varlink *link, sd_json_variant *parameters, const char *error_id, sd_varlink_reply_flags_t flags, void *userdata) { + AsyncUpgradeTest *t = ASSERT_PTR(userdata); + + ASSERT_STREQ(error_id, SD_VARLINK_ERROR_METHOD_NOT_FOUND); + ASSERT_TRUE(FLAGS_SET(flags, SD_VARLINK_REPLY_ERROR)); + t->got_error = true; + return 0; +} + +static int async_upgrade_on_unexpected_upgrade(sd_varlink *link, int input_fd, int output_fd, void *userdata) { + safe_close(input_fd); + safe_close(output_fd); + assert_not_reached(); +} + +TEST(upgrade_async) { + _cleanup_(sd_varlink_server_unrefp) sd_varlink_server *s = NULL; + _cleanup_(rm_rf_physical_and_freep) char *tmpdir = NULL; + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + const char *sp; + + ASSERT_OK(mkdtemp_malloc("/tmp/varlink-test-XXXXXX", &tmpdir)); + sp = strjoina(tmpdir, "/socket"); + + ASSERT_OK(sd_event_new(&e)); + + ASSERT_OK(sd_varlink_server_new(&s, SD_VARLINK_SERVER_UPGRADABLE)); + ASSERT_OK(sd_varlink_server_set_description(s, "upgrade-server")); + ASSERT_OK(varlink_server_bind_fiber(s, "io.test.Upgrade", method_upgrade)); + ASSERT_OK(sd_varlink_server_listen_address(s, sp, 0600)); + ASSERT_OK(sd_varlink_server_attach_event(s, e, 0)); + + /* First a client whose upgrade request is answered with an error: the connection must stay usable */ + AsyncUpgradeTest t2 = { .event = e, .input_fd = -EBADF, .output_fd = -EBADF }; + _cleanup_(sd_varlink_unrefp) sd_varlink *c2 = NULL; + ASSERT_OK(sd_varlink_connect_address(&c2, sp)); + ASSERT_OK(sd_varlink_set_description(c2, "async-upgrade-error-client")); + ASSERT_OK(sd_varlink_attach_event(c2, e, 0)); + sd_varlink_set_userdata(c2, &t2); + ASSERT_OK(sd_varlink_bind_reply(c2, async_upgrade_on_error_reply)); + ASSERT_OK(sd_varlink_bind_upgrade(c2, async_upgrade_on_unexpected_upgrade)); + ASSERT_OK(sd_varlink_invoke_and_upgrade(c2, "io.test.NoSuchMethod", /* parameters= */ NULL)); + + while (!t2.got_error) + ASSERT_OK(sd_event_run(e, UINT64_MAX)); + + ASSERT_TRUE(sd_varlink_is_connected(c2)); + ASSERT_OK_POSITIVE(sd_varlink_is_idle(c2)); + + /* After the refused upgrade regular calls work again */ + t2.got_error = false; + ASSERT_OK(sd_varlink_invoke(c2, "io.test.NoSuchMethod", /* parameters= */ NULL)); + while (!t2.got_error) + ASSERT_OK(sd_event_run(e, UINT64_MAX)); + ASSERT_TRUE(sd_varlink_is_connected(c2)); + + /* Now a client driven entirely by the event loop whose upgrade is accepted */ + AsyncUpgradeTest t = { .event = e, .input_fd = -EBADF, .output_fd = -EBADF }; + _cleanup_(sd_varlink_unrefp) sd_varlink *c = NULL; + ASSERT_OK(sd_varlink_connect_address(&c, sp)); + ASSERT_OK(sd_varlink_set_description(c, "async-upgrade-client")); + ASSERT_OK(sd_varlink_attach_event(c, e, 0)); + sd_varlink_set_userdata(c, &t); + ASSERT_OK(sd_varlink_bind_reply(c, async_upgrade_on_reply)); + ASSERT_OK(sd_varlink_bind_upgrade(c, async_upgrade_on_upgrade)); + ASSERT_OK(sd_varlink_invoke_and_upgrade(c, "io.test.Upgrade", /* parameters= */ NULL)); + + /* No other calls may be pipelined behind an upgrade request */ + ASSERT_ERROR(sd_varlink_invoke(c, "io.test.Upgrade", /* parameters= */ NULL), EBUSY); + ASSERT_ERROR(sd_varlink_invoke_and_upgrade(c, "io.test.Upgrade", /* parameters= */ NULL), EBUSY); + + ASSERT_OK(sd_event_loop(e)); + + ASSERT_TRUE(t.got_reply); + ASSERT_TRUE(t.got_upgrade); + ASSERT_TRUE(t.echo_done); + ASSERT_FALSE(sd_varlink_is_connected(c)); +} + static int upgrade_pipelining_client_fiber(void *arg) { union sockaddr_union sa = {}; _cleanup_close_ int fd = -EBADF; diff --git a/src/libsystemd/sd-varlink/varlink-internal.h b/src/libsystemd/sd-varlink/varlink-internal.h index 5cad0f69878..170a7513c59 100644 --- a/src/libsystemd/sd-varlink/varlink-internal.h +++ b/src/libsystemd/sd-varlink/varlink-internal.h @@ -15,6 +15,7 @@ typedef enum VarlinkState { VARLINK_IDLE_CLIENT, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, + VARLINK_AWAITING_REPLY_UPGRADE, VARLINK_CALLING, VARLINK_CALLED, VARLINK_COLLECTING, @@ -55,6 +56,7 @@ typedef enum VarlinkState { VARLINK_IDLE_CLIENT, \ VARLINK_AWAITING_REPLY, \ VARLINK_AWAITING_REPLY_MORE, \ + VARLINK_AWAITING_REPLY_UPGRADE, \ VARLINK_CALLING, \ VARLINK_CALLED, \ VARLINK_COLLECTING, \ diff --git a/src/systemd/sd-varlink.h b/src/systemd/sd-varlink.h index e04b60d2cdb..2b87cce0ae9 100644 --- a/src/systemd/sd-varlink.h +++ b/src/systemd/sd-varlink.h @@ -162,6 +162,15 @@ int sd_varlink_invokeb(sd_varlink *v, const char *method, ...); #define sd_varlink_invokebo(v, method, ...) \ sd_varlink_invokeb((v), (method), SD_JSON_BUILD_OBJECT(__VA_ARGS__)) +/* Enqueue method call with upgrade request. The reply is delivered to the reply callback. If the server + * accepted the upgrade, the connection is disconnected from the Varlink protocol afterwards and its fds are + * handed to the upgrade callback, see sd_varlink_bind_upgrade(). The asynchronous counterpart of + * sd_varlink_call_and_upgrade(). */ +int sd_varlink_invoke_and_upgrade(sd_varlink *v, const char *method, sd_json_variant *parameters); +int sd_varlink_invoke_and_upgradeb(sd_varlink *v, const char *method, ...); +#define sd_varlink_invoke_and_upgradebo(v, method, ...) \ + sd_varlink_invoke_and_upgradeb((v), (method), SD_JSON_BUILD_OBJECT(__VA_ARGS__)) + /* Enqueue method call, expect a reply now, and possibly more later, which are all delivered to the reply callback */ int sd_varlink_observe(sd_varlink *v, const char *method, sd_json_variant *parameters); int sd_varlink_observeb(sd_varlink *v, const char *method, ...);