From a93bf3cf86e04de8d48b2e71e9715cf0a96534cb Mon Sep 17 00:00:00 2001 From: Lennart Poettering Date: Fri, 4 Sep 2026 07:41:43 +0200 Subject: [PATCH] sd-varlink: add sd_varlink_invoke_and_upgrade() Add the asynchronous counterpart of sd_varlink_call_and_upgrade(): the method call is enqueued with the upgrade request, the reply is delivered to the reply callback as usual, and if the server accepted the upgrade the connection's file descriptors are handed to the callback installed with sd_varlink_bind_upgrade(), mirroring how the server side works. If the server replied with an error the connection stays usable. A new client state VARLINK_AWAITING_REPLY_UPGRADE tracks the pending request; no other calls may be pipelined behind it. --- man/rules/meson.build | 6 +- man/sd_varlink_call_and_upgrade.xml | 65 ++++++++- src/libsystemd/libsystemd.sym | 2 + src/libsystemd/sd-varlink/sd-varlink.c | 91 ++++++++++++- src/libsystemd/sd-varlink/test-varlink.c | 136 +++++++++++++++++++ src/libsystemd/sd-varlink/varlink-internal.h | 2 + src/systemd/sd-varlink.h | 9 ++ 7 files changed, 303 insertions(+), 8 deletions(-) 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, ...);