SymbX/systemd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sd-varlink: add async server-side upgrade API

Browse Source
This commit is contained in:
Lennart Poettering
2026-07-12 22:44:03 +02:00
parent f194afa164
commit 746d10d106
7 changed files with 490 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
man/rules/meson.build
+8
View File
@@ -977,6 +977,14 @@ manpages = [
''],
['sd_varlink_is_connected', '3', ['sd_varlink_is_idle'], ''],
['sd_varlink_push_fd', '3', ['sd_varlink_push_dup_fd'], ''],
['sd_varlink_reply_and_upgrade',
'3',
['sd_varlink_bind_upgrade',
'sd_varlink_respond_and_upgrade',
'sd_varlink_respond_and_upgradeb',
'sd_varlink_respond_and_upgradebo',
'sd_varlink_upgrade_t'],
''],
['sd_varlink_send', '3', ['sd_varlink_sendb', 'sd_varlink_sendbo'], ''],
['sd_varlink_server_listen_address',
'3',
man/sd_varlink_reply_and_upgrade.xml
+227
View File
@@ -0,0 +1,227 @@
<?xml version='1.0'?>
<!DOCTYPE refentry PUBLIC "-//OASIS//DTD DocBook XML V4.5//EN"
"http://www.oasis-open.org/docbook/xml/4.5/docbookx.dtd">
<!-- SPDX-License-Identifier: LGPL-2.1-or-later -->
<refentry id="sd_varlink_reply_and_upgrade" xmlns:xi="http://www.w3.org/2001/XInclude">
<refentryinfo>
<title>sd_varlink_reply_and_upgrade</title>
<productname>systemd</productname>
</refentryinfo>
<refmeta>
<refentrytitle>sd_varlink_reply_and_upgrade</refentrytitle>
<manvolnum>3</manvolnum>
</refmeta>
<refnamediv>
<refname>sd_varlink_reply_and_upgrade</refname>
<refname>sd_varlink_respond_and_upgrade</refname>
<refname>sd_varlink_respond_and_upgradeb</refname>
<refname>sd_varlink_respond_and_upgradebo</refname>
<refname>sd_varlink_bind_upgrade</refname>
<refname>sd_varlink_upgrade_t</refname>
<refpurpose>Reply to a Varlink method call and take over the connection for a raw protocol</refpurpose>
</refnamediv>
<refsynopsisdiv>
<funcsynopsis>
<funcsynopsisinfo>#include &lt;systemd/sd-varlink.h&gt;</funcsynopsisinfo>
<funcsynopsisinfo>typedef int (*<function>sd_varlink_upgrade_t</function>)(sd_varlink *<parameter>link</parameter>, int <parameter>input_fd</parameter>, int <parameter>output_fd</parameter>, void *<parameter>userdata</parameter>);</funcsynopsisinfo>
<funcprototype>
<funcdef>int <function>sd_varlink_reply_and_upgrade</function></funcdef>
<paramdef>sd_varlink *<parameter>link</parameter></paramdef>
<paramdef>sd_json_variant *<parameter>parameters</parameter></paramdef>
<paramdef>int *<parameter>ret_input_fd</parameter></paramdef>
<paramdef>int *<parameter>ret_output_fd</parameter></paramdef>
</funcprototype>
<funcprototype>
<funcdef>int <function>sd_varlink_respond_and_upgrade</function></funcdef>
<paramdef>sd_varlink *<parameter>link</parameter></paramdef>
<paramdef>sd_json_variant *<parameter>parameters</parameter></paramdef>
</funcprototype>
<funcprototype>
<funcdef>int <function>sd_varlink_respond_and_upgradeb</function></funcdef>
<paramdef>sd_varlink *<parameter>link</parameter></paramdef>
<paramdef>…</paramdef>
</funcprototype>
<funcprototype>
<funcdef>int <function>sd_varlink_respond_and_upgradebo</function></funcdef>
<paramdef>sd_varlink *<parameter>link</parameter></paramdef>
<paramdef>…</paramdef>
</funcprototype>
<funcprototype>
<funcdef>int <function>sd_varlink_bind_upgrade</function></funcdef>
<paramdef>sd_varlink *<parameter>link</parameter></paramdef>
<paramdef>sd_varlink_upgrade_t <parameter>callback</parameter></paramdef>
</funcprototype>
</funcsynopsis>
</refsynopsisdiv>
<refsect1>
<title>Description</title>
<para>These functions implement the server side of a Varlink <emphasis>protocol upgrade</emphasis>. A
Varlink client may request an upgrade when invoking a method (see
<citerefentry><refentrytitle>sd_varlink_call</refentrytitle><manvolnum>3</manvolnum></citerefentry> and
the <constant>SD_VARLINK_METHOD_UPGRADE</constant> flag passed to method callbacks). If a method is
invoked this way, the server may send one final reply and then abandon the Varlink protocol on the
connection entirely, handing the underlying socket or pipe file descriptors back to the caller so that an
arbitrary, non-Varlink ("raw") protocol may be spoken over them instead. This is useful for methods that
negotiate a switch to a streaming or binary protocol, in a manner similar to the
<literal>Upgrade:</literal> mechanism of HTTP.</para>
<para><function>sd_varlink_reply_and_upgrade()</function> sends a final reply to the currently processed
method call and then takes over the connection's file descriptors <emphasis>synchronously</emphasis>. It
takes the Varlink connection object, a JSON object with the reply parameters (which may be
<constant>NULL</constant>), and two output pointers for the file descriptors. The <parameter>parameters</parameter>
object is validated against the method's declared reply signature, if known. The reply is flushed to the
connection before the file descriptors are returned, so that the client has received the full reply before
the caller starts speaking the upgraded protocol. On success the connection is detached from its event
loop (if any) and disconnected, and ownership of the file descriptors is transferred to the caller, which
is responsible for eventually closing them with
<citerefentry project='man-pages'><refentrytitle>close</refentrytitle><manvolnum>2</manvolnum></citerefentry>.</para>
<para>The returned file descriptors are switched to blocking mode. For bidirectional sockets a single
underlying file descriptor carries both directions; in this case <parameter>ret_input_fd</parameter> and
<parameter>ret_output_fd</parameter> are returned as two independent (duplicated) descriptors referring
to the same socket, so they may be closed separately. For transports backed by a pair of pipes the two
descriptors differ and refer to the distinct read and write ends. Either
<parameter>ret_input_fd</parameter> or <parameter>ret_output_fd</parameter> (but not both) may be
<constant>NULL</constant>; in that case the unwanted direction is shut down (if they refer to the same
socket, see <citerefentry
project='man-pages'><refentrytitle>shutdown</refentrytitle><manvolnum>2</manvolnum></citerefentry>) or
closed (if separate file descriptors) rather than returned.</para>
<para>Note that <function>sd_varlink_reply_and_upgrade()</function> blocks synchronously until the reply
has been flushed to the socket. A misbehaving or adversarial client that stops reading could stall the
caller. It should hence not be used in servers that multiplex many Varlink connections in a single event
loop, as one stalled connection would block progress on all others. Use
<function>sd_varlink_respond_and_upgrade()</function> in that case instead.</para>
<para><function>sd_varlink_respond_and_upgrade()</function> is the <emphasis>asynchronous</emphasis>
counterpart of <function>sd_varlink_reply_and_upgrade()</function>. It enqueues the final reply and
returns immediately, without blocking and without returning the file descriptors directly. The reply is
subsequently flushed by the event loop during regular
<citerefentry><refentrytitle>sd_varlink_process</refentrytitle><manvolnum>3</manvolnum></citerefentry>
invocations. Once the reply has been fully flushed, the connection's file descriptors are handed to the
upgrade callback previously registered with <function>sd_varlink_bind_upgrade()</function>, and the
Varlink connection is disconnected. This variant is the right choice for servers that handle many
connections concurrently, as it never blocks on a single client.</para>
<para><function>sd_varlink_respond_and_upgradeb()</function> is similar to
<function>sd_varlink_respond_and_upgrade()</function>, but instead of expecting a fully constructed
<type>sd_json_variant</type> object carrying the reply parameters, this object is constructed on-the-fly
from the variadic argument list, in a style identical to
<citerefentry><refentrytitle>sd_json_build</refentrytitle><manvolnum>3</manvolnum></citerefentry>.
<function>sd_varlink_respond_and_upgradebo()</function> is identical to
<function>sd_varlink_respond_and_upgradeb()</function>, but an enclosing JSON object is added implicitly,
so that the argument list is expected to consist of object field pairs only, in a style identical to
<citerefentry><refentrytitle>sd_json_buildo</refentrytitle><manvolnum>3</manvolnum></citerefentry>.</para>
<para><function>sd_varlink_bind_upgrade()</function> registers the upgrade
<parameter>callback</parameter> that is invoked once a reply enqueued via
<function>sd_varlink_respond_and_upgrade()</function> (or one of its variants) has been flushed to the
connection. The callback is of type <type>sd_varlink_upgrade_t</type> and receives the Varlink connection
object, the input and output file descriptors of the (now upgraded) connection, and the userdata pointer
associated with the connection (see
<citerefentry><refentrytitle>sd_varlink_set_userdata</refentrytitle><manvolnum>3</manvolnum></citerefentry>).
Ownership of the two file descriptors is transferred to the callback (regardless if the callback later
fails or not). The descriptor semantics (blocking mode, duplication for bidirectional sockets, distinct
ends for pipe pairs) match those described for <function>sd_varlink_reply_and_upgrade()</function>
above. Passing <constant>NULL</constant> as <parameter>callback</parameter> removes a previously
registered callback.</para>
</refsect1>
<refsect1>
<title>Return Value</title>
<para>On success, <function>sd_varlink_reply_and_upgrade()</function>,
<function>sd_varlink_respond_and_upgrade()</function>,
<function>sd_varlink_respond_and_upgradeb()</function> and
<function>sd_varlink_respond_and_upgradebo()</function> return a positive integer.
<function>sd_varlink_bind_upgrade()</function> returns a non-negative integer on success. On failure, they
all return a negative errno-style error code.</para>
<refsect2>
<title>Errors</title>
<para>Returned errors may indicate the following problems:</para>
<variablelist>
<varlistentry>
<term><constant>-EINVAL</constant></term>
<listitem><para>An argument is invalid.</para></listitem>
</varlistentry>
<varlistentry>
<term><constant>-ENOTCONN</constant></term>
<listitem><para>The Varlink connection object is not connected.</para></listitem>
</varlistentry>
<varlistentry>
<term><constant>-EBUSY</constant></term>
<listitem><para>The connection is not currently processing a method call, so there is no reply to
send. When returned by <function>sd_varlink_bind_upgrade()</function>, a different upgrade callback
was already registered on the connection.</para></listitem>
</varlistentry>
<varlistentry>
<term><constant>-EPROTO</constant></term>
<listitem><para>The currently processed method call did not request a protocol upgrade, hence the
connection may not be taken over.</para></listitem>
</varlistentry>
<varlistentry>
<term><constant>-EBADMSG</constant></term>
<listitem><para>The client sent unexpected data before the upgrade completed, i.e. it pipelined
raw protocol data before receiving the upgrade reply.</para></listitem>
</varlistentry>
<varlistentry>
<term><constant>-ENOMEM</constant></term>
<listitem><para>Memory allocation failed.</para></listitem>
</varlistentry>
</variablelist>
</refsect2>
</refsect1>
<xi:include href="libsystemd-pkgconfig.xml" />
<refsect1>
<title>History</title>
<para><function>sd_varlink_reply_and_upgrade()</function> was added in version 261.</para>
<para><function>sd_varlink_respond_and_upgrade()</function>,
<function>sd_varlink_respond_and_upgradeb()</function>,
<function>sd_varlink_respond_and_upgradebo()</function> and
<function>sd_varlink_bind_upgrade()</function> were added in version 262.</para>
</refsect1>
<refsect1>
<title>See Also</title>
<para><simplelist type="inline">
<member><citerefentry><refentrytitle>systemd</refentrytitle><manvolnum>1</manvolnum></citerefentry></member>
<member><citerefentry><refentrytitle>sd-varlink</refentrytitle><manvolnum>3</manvolnum></citerefentry></member>
<member><citerefentry><refentrytitle>sd_varlink_reply</refentrytitle><manvolnum>3</manvolnum></citerefentry></member>
<member><citerefentry><refentrytitle>sd_varlink_call</refentrytitle><manvolnum>3</manvolnum></citerefentry></member>
<member><citerefentry><refentrytitle>sd_varlink_process</refentrytitle><manvolnum>3</manvolnum></citerefentry></member>
<member><citerefentry><refentrytitle>sd_varlink_set_userdata</refentrytitle><manvolnum>3</manvolnum></citerefentry></member>
<member><citerefentry><refentrytitle>sd_json_build</refentrytitle><manvolnum>3</manvolnum></citerefentry></member>
</simplelist></para>
</refsect1>
</refentry>
src/libsystemd/libsystemd.sym
+7
View File
@@ -1104,3 +1104,10 @@ global:
sd_varlink_reply_and_upgrade;
sd_varlink_set_sentinel;
} LIBSYSTEMD_260;
LIBSYSTEMD_262 {
global:
sd_varlink_bind_upgrade;
sd_varlink_respond_and_upgrade;
sd_varlink_respond_and_upgradeb;
} LIBSYSTEMD_261;
src/libsystemd/sd-varlink/sd-varlink.c
+111 -2
View File
@@ -62,9 +62,11 @@ static const char* const varlink_state_table[_VARLINK_STATE_MAX] = {
[VARLINK_PROCESSING_METHOD_ONEWAY] = "processing-method-oneway",
[VARLINK_PROCESSING_METHOD_UPGRADE] = "processing-method-upgrade",
[VARLINK_PROCESSED_METHOD] = "processed-method",
[VARLINK_PROCESSED_METHOD_UPGRADE] = "processed-method-upgrade",
[VARLINK_PENDING_METHOD] = "pending-method",
[VARLINK_PENDING_METHOD_MORE] = "pending-method-more",
[VARLINK_PENDING_METHOD_UPGRADE] = "pending-method-upgrade",
[VARLINK_UPGRADING] = "upgrading",
[VARLINK_PENDING_DISCONNECT] = "pending-disconnect",
[VARLINK_PENDING_TIMEOUT] = "pending-timeout",
[VARLINK_PROCESSING_DISCONNECT] = "processing-disconnect",
@@ -1407,6 +1409,11 @@ static int varlink_dispatch_method(sd_varlink *v) {
varlink_set_state(v, VARLINK_PENDING_METHOD_UPGRADE);
break;
case VARLINK_PROCESSED_METHOD_UPGRADE: /* Upgrade requested, let's now wait till everything is flushed */
varlink_clear_current(v);
varlink_set_state(v, VARLINK_UPGRADING);
break;
case VARLINK_DISCONNECTED: /* Handler called sd_varlink_close() on us, which is fine */
break;
@@ -1474,6 +1481,41 @@ static int varlink_handle_upgrade_fds(sd_varlink *v, int *ret_input_fd, int *ret
return 0;
}
static int varlink_dispatch_upgrade(sd_varlink *v) {
int r;
assert(v);
if (v->state != VARLINK_UPGRADING)
return 0;
r = json_stream_has_buffered_output(&v->stream);
if (r < 0)
return r;
if (r > 0)
return 0;
/* Detach from the event loop before stealing the fds */
sd_varlink_detach_event(v);
_cleanup_close_ int input_fd = -EBADF, output_fd = -EBADF;
r = varlink_handle_upgrade_fds(v, &input_fd, &output_fd);
sd_varlink_close(v);
if (r < 0)
return r;
/* Fail if by this time the caller still hasn't set an upgrade callback. */
if (!v->upgrade_callback)
return -ENOPKG;
/* No matter what, we donate the fds to callback */
r = v->upgrade_callback(v, TAKE_FD(input_fd), TAKE_FD(output_fd), v->userdata);
if (r < 0)
return r;
return 1;
}
_public_ int sd_varlink_process(sd_varlink *v) {
int r;
@@ -1490,6 +1532,12 @@ _public_ int sd_varlink_process(sd_varlink *v) {
if (r != 0)
goto finish;
r = varlink_dispatch_upgrade(v);
if (r < 0)
varlink_log_errno(v, r, "Upgrade dispatch failed: %m");
if (r != 0)
goto finish;
r = varlink_dispatch_reply(v);
if (r < 0)
varlink_log_errno(v, r, "Reply dispatch failed: %m");
@@ -2457,11 +2505,12 @@ _public_ int sd_varlink_replyb(sd_varlink *v, ...) {
return sd_varlink_reply(v, parameters);
}
_public_ int sd_varlink_reply_and_upgrade(sd_varlink *v, sd_json_variant *parameters, int *ret_input_fd, int *ret_output_fd) {
_public_ int sd_varlink_respond_and_upgrade(
sd_varlink *v,
sd_json_variant *parameters) {
int r;
assert_return(v, -EINVAL);
assert_return(ret_input_fd || ret_output_fd, -EINVAL);
if (v->state == VARLINK_DISCONNECTED)
return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected.");
@@ -2508,6 +2557,55 @@ _public_ int sd_varlink_reply_and_upgrade(sd_varlink *v, sd_json_variant *parame
if (r < 0)
return varlink_log_errno(v, r, "Failed to enqueue json message: %m");
switch (v->state) {
case VARLINK_PENDING_METHOD_UPGRADE:
varlink_clear_current(v);
varlink_set_state(v, VARLINK_UPGRADING);
break;
case VARLINK_PROCESSING_METHOD_UPGRADE:
varlink_set_state(v, VARLINK_PROCESSED_METHOD_UPGRADE);
break;
default:
assert_not_reached();
}
return 1;
}
_public_ int sd_varlink_respond_and_upgradeb(sd_varlink *v, ...) {
_cleanup_(sd_json_variant_unrefp) sd_json_variant *parameters = NULL;
va_list ap;
int r;
assert_return(v, -EINVAL);
va_start(ap, v);
r = sd_json_buildv(&parameters, ap);
va_end(ap);
if (r < 0)
return r;
return sd_varlink_respond_and_upgrade(v, parameters);
}
_public_ int sd_varlink_reply_and_upgrade(
sd_varlink *v,
sd_json_variant *parameters,
int *ret_input_fd,
int *ret_output_fd) {
int r;
assert_return(v, -EINVAL);
assert_return(ret_input_fd || ret_output_fd, -EINVAL);
r = sd_varlink_respond_and_upgrade(v, parameters);
if (r < 0)
return r;
/* Flush the reply to the socket before stealing the fds. The reply must be fully written
* before the caller starts speaking the upgraded protocol. */
r = json_stream_flush(&v->stream);
@@ -2797,6 +2895,17 @@ _public_ int sd_varlink_bind_reply(sd_varlink *v, sd_varlink_reply_t reply) {
return 0;
}
_public_ int sd_varlink_bind_upgrade(sd_varlink *v, sd_varlink_upgrade_t upgrade) {
assert_return(v, -EINVAL);
if (upgrade && v->upgrade_callback && upgrade != v->upgrade_callback)
return varlink_log_errno(v, SYNTHETIC_ERRNO(EBUSY), "A different callback was already set.");
v->upgrade_callback = upgrade;
return 0;
}
_public_ void* sd_varlink_set_userdata(sd_varlink *v, void *userdata) {
void *old;
src/libsystemd/sd-varlink/test-varlink.c
+121
View File
@@ -1177,6 +1177,127 @@ TEST(upgrade_pipelining) {
ASSERT_OK(sd_future_result(f));
}
/* Stashes the fds handed to the asynchronous upgrade callback so the raw-I/O fiber can pick them up.
* A single connection is upgraded per test run, so globals are sufficient. */
static int respond_upgrade_input_fd = -EBADF;
static int respond_upgrade_output_fd = -EBADF;
static int respond_upgrade_io_fiber(void *arg) {
_cleanup_close_ int input_fd = TAKE_FD(respond_upgrade_input_fd), output_fd = TAKE_FD(respond_upgrade_output_fd);
/* Same reverse-echo raw protocol as method_upgrade(), but reached through the asynchronous
* upgrade callback instead of a synchronous fiber method: read until the client shuts down its
* write side, reverse the bytes, and write them back. */
char buf[64] = {};
ssize_t n = ASSERT_OK(loop_read(input_fd, buf, sizeof(buf) - 1, /* do_poll= */ true));
ASSERT_GT(n, 0);
for (ssize_t i = 0; i < n / 2; i++)
SWAP_TWO(buf[i], buf[n - 1 - i]);
ASSERT_OK(loop_write(output_fd, buf, n));
return 0;
}
static int respond_upgrade_callback(sd_varlink *link, int input_fd, int output_fd, void *userdata) {
sd_event *e = ASSERT_PTR(userdata);
/* The upgrade callback is invoked from sd_varlink_process() in the main event loop (not a
* fiber), so it must not block on raw I/O. Hand the connection's fds to a floating fiber that
* speaks the raw protocol, allowing the client fiber to make progress concurrently. Ownership of
* the fds is transferred to us regardless of what happens next, so stash them before spawning. */
respond_upgrade_input_fd = input_fd;
respond_upgrade_output_fd = output_fd;
ASSERT_OK(sd_fiber_new(e, "respond-upgrade-io", respond_upgrade_io_fiber, /* userdata= */ NULL, /* destroy= */ NULL, /* ret= */ NULL));
return 0;
}
static int method_respond_upgrade(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) {
int r;
ASSERT_TRUE(FLAGS_SET(flags, SD_VARLINK_METHOD_UPGRADE));
/* Register the async upgrade callback, then enqueue the final reply. Unlike
* sd_varlink_reply_and_upgrade(), sd_varlink_respond_and_upgradebo() returns immediately without
* blocking: the reply is flushed by the event loop and the fds are subsequently handed to
* respond_upgrade_callback(). This is why the method may be bound as a regular (non-fiber) method. */
ASSERT_OK(sd_varlink_bind_upgrade(link, respond_upgrade_callback));
r = sd_varlink_respond_and_upgradebo(
link,
SD_JSON_BUILD_PAIR_STRING("greeting", "aloha"));
if (r < 0)
return r;
return 0;
}
static int respond_upgrade_client_fiber(void *arg) {
_cleanup_(sd_varlink_flush_close_unrefp) sd_varlink *c = NULL;
_cleanup_close_ int input_fd = -EBADF, output_fd = -EBADF;
sd_json_variant *o = NULL;
const char *error_id = NULL;
ASSERT_OK(sd_varlink_connect_address(&c, arg));
ASSERT_OK(sd_varlink_set_description(c, "respond-upgrade-client"));
ASSERT_OK(sd_varlink_call_and_upgrade(c, "io.test.RespondUpgrade", /* parameters= */ NULL, &o, &error_id, &input_fd, &output_fd));
ASSERT_NULL(error_id);
/* The reply parameters enqueued asynchronously by the server must have arrived intact. */
ASSERT_STREQ(sd_json_variant_string(sd_json_variant_by_key(o, "greeting")), "aloha");
ASSERT_GE(input_fd, 0);
ASSERT_GE(output_fd, 0);
ASSERT_NE(input_fd, output_fd); /* library dups for bidirectional sockets */
/* Send a test string, shut down write side so the server sees EOF, then read the reversed reply */
static const char msg[] = "Hello!";
ASSERT_OK(loop_write(output_fd, msg, strlen(msg)));
ASSERT_OK_ERRNO(shutdown(output_fd, SHUT_WR));
char buf[64] = {};
ssize_t n = ASSERT_OK(loop_read(input_fd, buf, strlen(msg), /* do_poll= */ true));
ASSERT_EQ((size_t) n, strlen(msg));
ASSERT_STREQ(buf, "!olleH");
ASSERT_OK(sd_event_exit(sd_fiber_get_event(), EXIT_SUCCESS));
return 0;
}
TEST(respond_upgrade) {
_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;
_cleanup_(sd_future_unrefp) sd_future *f = 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|SD_VARLINK_SERVER_INHERIT_USERDATA));
ASSERT_OK(sd_varlink_server_set_description(s, "respond-upgrade-server"));
/* The event loop is propagated to the async upgrade callback via the connection userdata (inherited
* from the server) so the callback can spawn the raw-I/O fiber. */
sd_varlink_server_set_userdata(s, e);
/* Unlike TEST(upgrade), this is a *regular* (non-fiber) method: it drives the asynchronous
* sd_varlink_respond_and_upgradebo() API, which never blocks the event loop on a single client. */
ASSERT_OK(sd_varlink_server_bind_method(s, "io.test.RespondUpgrade", method_respond_upgrade));
ASSERT_OK(sd_varlink_server_listen_address(s, sp, 0600));
ASSERT_OK(sd_varlink_server_attach_event(s, e, 0));
ASSERT_OK(sd_fiber_new(e, "respond-upgrade-client", respond_upgrade_client_fiber, (void*) sp, /* destroy= */ NULL, &f));
ASSERT_OK(sd_event_loop(e));
ASSERT_OK(sd_future_result(f));
}
typedef struct ExecDirServer {
sd_varlink_server *server;
const char *name;
src/libsystemd/sd-varlink/varlink-internal.h
+6 -1
View File
@@ -28,9 +28,11 @@ typedef enum VarlinkState {
VARLINK_PROCESSING_METHOD_ONEWAY,
VARLINK_PROCESSING_METHOD_UPGRADE,
VARLINK_PROCESSED_METHOD,
VARLINK_PROCESSED_METHOD_UPGRADE,
VARLINK_PENDING_METHOD,
VARLINK_PENDING_METHOD_MORE,
VARLINK_PENDING_METHOD_UPGRADE,
VARLINK_UPGRADING,
/* Common states (only during shutdown) */
VARLINK_PENDING_DISCONNECT,
@@ -64,9 +66,11 @@ typedef enum VarlinkState {
VARLINK_PROCESSING_METHOD_ONEWAY, \
VARLINK_PROCESSING_METHOD_UPGRADE, \
VARLINK_PROCESSED_METHOD, \
VARLINK_PROCESSED_METHOD_UPGRADE, \
VARLINK_PENDING_METHOD, \
VARLINK_PENDING_METHOD_MORE, \
VARLINK_PENDING_METHOD_UPGRADE)
VARLINK_PENDING_METHOD_UPGRADE, \
VARLINK_UPGRADING)
/* Tests whether we are expected to generate a method call reply, i.e. are processing a method call, except
* one with the ONEWAY flag set. */
@@ -92,6 +96,7 @@ typedef struct sd_varlink {
unsigned n_pending;
sd_varlink_reply_t reply_callback;
sd_varlink_upgrade_t upgrade_callback;
sd_json_variant *current;
sd_json_variant *current_collected;
src/systemd/sd-varlink.h
+10
View File
@@ -84,6 +84,7 @@ __extension__ typedef enum _SD_ENUM_TYPE_S64(sd_varlink_invocation_flags_t) {
typedef int (*sd_varlink_method_t)(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata);
typedef int (*sd_varlink_reply_t)(sd_varlink *link, sd_json_variant *parameters, const char *error_id, sd_varlink_reply_flags_t flags, void *userdata);
typedef int (*sd_varlink_upgrade_t)(sd_varlink *link, int input_fd, int output_fd, void *userdata);
typedef int (*sd_varlink_connect_t)(sd_varlink_server *server, sd_varlink *link, void *userdata);
typedef void (*sd_varlink_disconnect_t)(sd_varlink_server *server, sd_varlink *link, void *userdata);
@@ -182,6 +183,12 @@ int sd_varlink_replyb(sd_varlink *v, ...);
* connections. */
int sd_varlink_reply_and_upgrade(sd_varlink *v, sd_json_variant *parameters, int *ret_input_fd, int *ret_output_fd);
/* Like sd_varlink_reply_and_upgrade() but asynchronous */
int sd_varlink_respond_and_upgrade(sd_varlink *v, sd_json_variant *parameters);
int sd_varlink_respond_and_upgradeb(sd_varlink *v, ...);
#define sd_varlink_respond_and_upgradebo(v, ...) \
sd_varlink_respond_and_upgradeb((v), SD_JSON_BUILD_OBJECT(__VA_ARGS__))
/* Enqueue a (final) error */
int sd_varlink_error(sd_varlink *v, const char *error_id, sd_json_variant *parameters);
int sd_varlink_errorb(sd_varlink *v, const char *error_id, ...);
@@ -224,6 +231,9 @@ int sd_varlink_set_allow_fd_passing_output(sd_varlink *v, int b);
/* Bind a disconnect, reply or timeout callback */
int sd_varlink_bind_reply(sd_varlink *v, sd_varlink_reply_t reply);
/* Bind an upgrade callback (server side) */
int sd_varlink_bind_upgrade(sd_varlink *v, sd_varlink_upgrade_t upgrade);
void* sd_varlink_set_userdata(sd_varlink *v, void *userdata);
void* sd_varlink_get_userdata(sd_varlink *v);