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

sd-varlink: async upgrade support (#42974)

Browse Source 
With 2c6f9af8e5 sd-varlink gained protocol
upgrade support, however only in a synchronous fashion. This adds
asynchronous protocol upgrade for the server side, thus enabling
multiplexing daemons (i.e. those that handle multiple connections from
the same event loop) to support protocol upgrades too.

I plan to use this in the upcoming "ptybroker" component that allows
acquiring a pty through varlink.

/cc @mvo5
This commit is contained in:
Lennart Poettering
2026-07-13 14:29:06 +02:00
committed by GitHub
parent 9a09141cae 953d21846a
commit 755efbe410
10 changed files with 826 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
man/rules/meson.build
+12
View File
@@ -968,6 +968,10 @@ manpages = [
'sd_uid_get_sessions',
'sd_uid_is_on_seat'],
'HAVE_PAM'],
['sd_varlink_call_and_upgrade',
'3',
['sd_varlink_call_and_upgradeb', 'sd_varlink_call_and_upgradebo'],
''],
['sd_varlink_connect_address',
'3',
['sd_varlink_connect_exec',
@@ -977,6 +981,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_call_and_upgrade.xml
+177
View File
@@ -0,0 +1,177 @@
<?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_call_and_upgrade" xmlns:xi="http://www.w3.org/2001/XInclude">
<refentryinfo>
<title>sd_varlink_call_and_upgrade</title>
<productname>systemd</productname>
</refentryinfo>
<refmeta>
<refentrytitle>sd_varlink_call_and_upgrade</refentrytitle>
<manvolnum>3</manvolnum>
</refmeta>
<refnamediv>
<refname>sd_varlink_call_and_upgrade</refname>
<refname>sd_varlink_call_and_upgradeb</refname>
<refname>sd_varlink_call_and_upgradebo</refname>
<refpurpose>Invoke a Varlink method and take over the connection for a raw protocol</refpurpose>
</refnamediv>
<refsynopsisdiv>
<funcsynopsis>
<funcsynopsisinfo>#include &lt;systemd/sd-varlink.h&gt;</funcsynopsisinfo>
<funcprototype>
<funcdef>int <function>sd_varlink_call_and_upgrade</function></funcdef>
<paramdef>sd_varlink *<parameter>link</parameter></paramdef>
<paramdef>const char *<parameter>method</parameter></paramdef>
<paramdef>sd_json_variant *<parameter>parameters</parameter></paramdef>
<paramdef>sd_json_variant **<parameter>ret_parameters</parameter></paramdef>
<paramdef>const char **<parameter>ret_error_id</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_call_and_upgradeb</function></funcdef>
<paramdef>sd_varlink *<parameter>link</parameter></paramdef>
<paramdef>const char *<parameter>method</parameter></paramdef>
<paramdef>sd_json_variant **<parameter>ret_parameters</parameter></paramdef>
<paramdef>const char **<parameter>ret_error_id</parameter></paramdef>
<paramdef>int *<parameter>ret_input_fd</parameter></paramdef>
<paramdef>int *<parameter>ret_output_fd</parameter></paramdef>
<paramdef>…</paramdef>
</funcprototype>
<funcprototype>
<funcdef>int <function>sd_varlink_call_and_upgradebo</function></funcdef>
<paramdef>sd_varlink *<parameter>link</parameter></paramdef>
<paramdef>const char *<parameter>method</parameter></paramdef>
<paramdef>sd_json_variant **<parameter>ret_parameters</parameter></paramdef>
<paramdef>const char **<parameter>ret_error_id</parameter></paramdef>
<paramdef>int *<parameter>ret_input_fd</parameter></paramdef>
<paramdef>int *<parameter>ret_output_fd</parameter></paramdef>
<paramdef>…</paramdef>
</funcprototype>
</funcsynopsis>
</refsynopsisdiv>
<refsect1>
<title>Description</title>
<para>These functions implement the client side of a Varlink <emphasis>protocol upgrade</emphasis>. They
are the counterpart of the server-side
<citerefentry><refentrytitle>sd_varlink_reply_and_upgrade</refentrytitle><manvolnum>3</manvolnum></citerefentry>
family. If a method call is issued this way, the connection may switch, after the reply, from the Varlink
protocol to an arbitrary, non-Varlink ("raw") protocol spoken over the underlying socket or pipe file
descriptors, in a manner similar to the <literal>Upgrade:</literal> mechanism of HTTP.</para>
<para><function>sd_varlink_call_and_upgrade()</function> synchronously invokes the method
<parameter>method</parameter> on the connection <parameter>link</parameter>, requesting a protocol upgrade,
and waits for the reply. It takes the Varlink connection object, the method name, a JSON object with the
method call parameters (which may be <constant>NULL</constant>), two output pointers for the reply
parameters and a possible error identifier, and two output pointers for the file descriptors. The reply
parameters returned in <parameter>ret_parameters</parameter> and the error identifier returned in
<parameter>ret_error_id</parameter> are borrowed references that remain valid only until the connection is
closed or unreffed. Either of the two may be passed as <constant>NULL</constant> if the information is not
needed.</para>
<para>If the server replies with a regular reply (rather than a Varlink error) the connection is upgraded:
it is disconnected from the Varlink protocol, and ownership of the underlying file descriptors is
transferred to the caller via <parameter>ret_input_fd</parameter> and <parameter>ret_output_fd</parameter>,
which is responsible for eventually closing them with
<citerefentry project='man-pages'><refentrytitle>close</refentrytitle><manvolnum>2</manvolnum></citerefentry>.
If the server replies with a Varlink error instead, the connection is <emphasis>not</emphasis> upgraded and
no file descriptors are returned; the error identifier is reported in <parameter>ret_error_id</parameter>
(if non-<constant>NULL</constant>), otherwise a negative errno-style error derived from it is returned.</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. At least one of
<parameter>ret_input_fd</parameter> or <parameter>ret_output_fd</parameter> must be
non-<constant>NULL</constant>.</para>
<para><function>sd_varlink_call_and_upgradeb()</function> is similar to
<function>sd_varlink_call_and_upgrade()</function>, but instead of expecting a fully constructed
<type>sd_json_variant</type> object carrying the method call 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_call_and_upgradebo()</function> is identical to
<function>sd_varlink_call_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>. This
relationship mirrors that of
<citerefentry><refentrytitle>sd_varlink_call</refentrytitle><manvolnum>3</manvolnum></citerefentry> and its
<function>sd_varlink_callb()</function>/<function>sd_varlink_callbo()</function> variants.</para>
</refsect1>
<refsect1>
<title>Return Value</title>
<para>On success, these functions return a positive integer if the connection was upgraded, or zero if the
server returned a Varlink error and <parameter>ret_error_id</parameter> was set to it. On failure, they
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>-EPROTO</constant></term>
<listitem><para>A protocol error occurred, for example the server sent unexpected raw protocol data
before the upgrade completed.</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_call_and_upgrade()</function> was added in version 261.</para>
<para><function>sd_varlink_call_and_upgradeb()</function> and
<function>sd_varlink_call_and_upgradebo()</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_call</refentrytitle><manvolnum>3</manvolnum></citerefentry></member>
<member><citerefentry><refentrytitle>sd_varlink_reply_and_upgrade</refentrytitle><manvolnum>3</manvolnum></citerefentry></member>
<member><citerefentry><refentrytitle>sd_json_build</refentrytitle><manvolnum>3</manvolnum></citerefentry></member>
</simplelist></para>
</refsect1>
</refentry>
man/sd_varlink_reply_and_upgrade.xml
+228
View File
@@ -0,0 +1,228 @@
<?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_call_and_upgrade</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
+8
View File
@@ -1104,3 +1104,11 @@ global:
sd_varlink_reply_and_upgrade;
sd_varlink_set_sentinel;
} LIBSYSTEMD_260;
LIBSYSTEMD_262 {
global:
sd_varlink_bind_upgrade;
sd_varlink_call_and_upgradeb;
sd_varlink_respond_and_upgrade;
sd_varlink_respond_and_upgradeb;
} LIBSYSTEMD_261;
src/libsystemd/sd-json/json-stream.c
+16 -3
View File
@@ -844,16 +844,29 @@ fail:
return r;
}
int json_stream_has_buffered_output(const JsonStream *s) {
assert(s);
if (s->output_buffer_size == 0 && !s->output_queue)
return false;
if (FLAGS_SET(s->flags, JSON_STREAM_WRITE_DISCONNECTED))
return -ECONNRESET;
return true;
}
int json_stream_flush(JsonStream *s) {
int ret = 0, r;
assert(s);
for (;;) {
if (s->output_buffer_size == 0 && !s->output_queue)
r = json_stream_has_buffered_output(s);
if (r < 0)
return r;
if (r == 0)
break;
if (FLAGS_SET(s->flags, JSON_STREAM_WRITE_DISCONNECTED))
return -ECONNRESET;
r = json_stream_write(s);
if (r < 0)
src/libsystemd/sd-json/json-stream.h
+1
View File
@@ -206,6 +206,7 @@ int json_stream_parse(JsonStream *s, sd_json_variant **ret);
/* Status accessors used by the consumer's state machine. */
bool json_stream_has_buffered_input(const JsonStream *s);
int json_stream_has_buffered_output(const JsonStream *s);
/* Compute the poll events the consumer should wait for. The stream queries the consumer's
* phase via the registered get_phase callback. In JSON_STREAM_PHASE_READING the stream asks
src/libsystemd/sd-varlink/sd-varlink.c
+238 -99
View File
@@ -48,27 +48,31 @@
#define VARLINK_COLLECT_MAX 1024U
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_CALLING] = "calling",
[VARLINK_CALLED] = "called",
[VARLINK_COLLECTING] = "collecting",
[VARLINK_COLLECTING_REPLY] = "collecting-reply",
[VARLINK_PROCESSING_REPLY] = "processing-reply",
[VARLINK_IDLE_SERVER] = "idle-server",
[VARLINK_PROCESSING_METHOD] = "processing-method",
[VARLINK_PROCESSING_METHOD_MORE] = "processing-method-more",
[VARLINK_PROCESSING_METHOD_ONEWAY] = "processing-method-oneway",
[VARLINK_PROCESSED_METHOD] = "processed-method",
[VARLINK_PENDING_METHOD] = "pending-method",
[VARLINK_PENDING_METHOD_MORE] = "pending-method-more",
[VARLINK_PENDING_DISCONNECT] = "pending-disconnect",
[VARLINK_PENDING_TIMEOUT] = "pending-timeout",
[VARLINK_PROCESSING_DISCONNECT] = "processing-disconnect",
[VARLINK_PROCESSING_TIMEOUT] = "processing-timeout",
[VARLINK_PROCESSING_FAILURE] = "processing-failure",
[VARLINK_DISCONNECTED] = "disconnected",
[VARLINK_IDLE_CLIENT] = "idle-client",
[VARLINK_AWAITING_REPLY] = "awaiting-reply",
[VARLINK_AWAITING_REPLY_MORE] = "awaiting-reply-more",
[VARLINK_CALLING] = "calling",
[VARLINK_CALLED] = "called",
[VARLINK_COLLECTING] = "collecting",
[VARLINK_COLLECTING_REPLY] = "collecting-reply",
[VARLINK_PROCESSING_REPLY] = "processing-reply",
[VARLINK_IDLE_SERVER] = "idle-server",
[VARLINK_PROCESSING_METHOD] = "processing-method",
[VARLINK_PROCESSING_METHOD_MORE] = "processing-method-more",
[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",
[VARLINK_PROCESSING_TIMEOUT] = "processing-timeout",
[VARLINK_PROCESSING_FAILURE] = "processing-failure",
[VARLINK_DISCONNECTED] = "disconnected",
};
DEFINE_PRIVATE_STRING_TABLE_LOOKUP_TO_STRING(varlink_state, VarlinkState);
@@ -112,7 +116,7 @@ static JsonStreamPhase varlink_phase(void *userdata) {
if (v->state == VARLINK_IDLE_CLIENT)
return JSON_STREAM_PHASE_IDLE_CLIENT;
if (IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE))
if (IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE, VARLINK_PENDING_METHOD_UPGRADE))
return JSON_STREAM_PHASE_PENDING_OUTPUT;
return JSON_STREAM_PHASE_OTHER;
@@ -1096,7 +1100,7 @@ static int varlink_fiber_entry(void *userdata) {
* VARLINK_PROCESSING_METHOD{,_MORE} to VARLINK_PENDING_METHOD{,_MORE}, so that's what we match
* here to decide whether the call still needs a reply. Any other state (e.g. IDLE_SERVER after
* the callback replied, or DISCONNECTED after sd_varlink_close()) means no fixup is needed. */
if (!IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE))
if (!IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE, VARLINK_PENDING_METHOD_UPGRADE))
return r;
if (r < 0) {
@@ -1118,7 +1122,7 @@ static int varlink_fiber_entry(void *userdata) {
r = 0;
/* If we didn't manage to enqueue a response, then fail the connection completely. */
if (r < 0 && IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE))
if (r < 0 && IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE, VARLINK_PENDING_METHOD_UPGRADE))
goto fail;
return r;
@@ -1264,20 +1268,20 @@ static int varlink_dispatch_method(sd_varlink *v) {
if (r < 0)
goto fail;
varlink_set_state(v, (flags & SD_VARLINK_METHOD_MORE) ? VARLINK_PROCESSING_METHOD_MORE :
(flags & SD_VARLINK_METHOD_ONEWAY) ? VARLINK_PROCESSING_METHOD_ONEWAY :
VARLINK_PROCESSING_METHOD);
varlink_set_state(v, (flags & SD_VARLINK_METHOD_MORE) ? VARLINK_PROCESSING_METHOD_MORE :
(flags & SD_VARLINK_METHOD_ONEWAY) ? VARLINK_PROCESSING_METHOD_ONEWAY :
(flags & SD_VARLINK_METHOD_UPGRADE) ? VARLINK_PROCESSING_METHOD_UPGRADE :
VARLINK_PROCESSING_METHOD);
assert(v->server);
/* Reset the per-call upgrade marker on every dispatch — a previous method's
* UPGRADE flag must not bleed into this one. The transport-level bounded reads
* stay active for SD_VARLINK_SERVER_UPGRADABLE servers regardless. */
v->protocol_upgrade = FLAGS_SET(flags, SD_VARLINK_METHOD_UPGRADE);
json_stream_set_flags(
&v->stream,
JSON_STREAM_BOUNDED_READS,
v->protocol_upgrade || FLAGS_SET(v->server->flags, SD_VARLINK_SERVER_UPGRADABLE));
FLAGS_SET(flags, SD_VARLINK_METHOD_UPGRADE) || FLAGS_SET(v->server->flags, SD_VARLINK_SERVER_UPGRADABLE));
/* First consult user supplied method implementations */
bool is_fiber = false;
@@ -1401,6 +1405,15 @@ static int varlink_dispatch_method(sd_varlink *v) {
varlink_set_state(v, VARLINK_PENDING_METHOD_MORE);
break;
case VARLINK_PROCESSING_METHOD_UPGRADE: /* Method call wasn't replied to, will be replied to later */
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;
@@ -1421,6 +1434,88 @@ fail:
return r;
}
static int varlink_handle_upgrade_fds(sd_varlink *v, int *ret_input_fd, int *ret_output_fd) {
int r;
assert(v);
assert(ret_input_fd || ret_output_fd);
/* Ensure no post-upgrade data was consumed into our input buffer (we ensure this via MSG_PEEK or
* byte-to-byte) and refuse the upgrade rather than silently losing the data. */
if (json_stream_has_buffered_input(&v->stream))
return varlink_log_errno(v, SYNTHETIC_ERRNO(EPROTO),
"Unexpected buffered data during protocol upgrade, refusing.");
_cleanup_close_ int input_fd = TAKE_FD(v->stream.input_fd),
output_fd = TAKE_FD(v->stream.output_fd);
/* Pass the connection fds to the caller, it owns them now. Reset to blocking mode
* since callers of the upgraded protocol will generally expect normal blocking
* semantics. For bidirectional sockets (input_fd == output_fd), dup the fd so that
* callers always get two independent fds they can close separately. */
if (input_fd == output_fd) {
output_fd = fcntl(input_fd, F_DUPFD_CLOEXEC, 3);
if (output_fd < 0)
return varlink_log_errno(v, errno, "Failed to dup upgraded connection fd: %m");
} else {
r = fd_nonblock(output_fd, false);
if (r < 0)
return varlink_log_errno(v, r, "Failed to set output fd to blocking mode: %m");
}
r = fd_nonblock(input_fd, false);
if (r < 0)
return varlink_log_errno(v, r, "Failed to set input fd to blocking mode: %m");
/* Hand out requested fds, shut down unwanted directions. */
if (ret_input_fd)
*ret_input_fd = TAKE_FD(input_fd);
else
(void) shutdown(input_fd, SHUT_RD);
if (ret_output_fd)
*ret_output_fd = TAKE_FD(output_fd);
else
(void) shutdown(output_fd, SHUT_WR);
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;
@@ -1437,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");
@@ -1516,7 +1617,7 @@ _public_ int sd_varlink_dispatch_again(sd_varlink *v) {
if (v->state == VARLINK_DISCONNECTED)
return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected.");
if (!IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE))
if (!IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE, VARLINK_PENDING_METHOD_UPGRADE))
return varlink_log_errno(v, SYNTHETIC_ERRNO(EBUSY), "Connection has no pending method.");
varlink_set_state(v, VARLINK_IDLE_SERVER);
@@ -2005,53 +2106,6 @@ _public_ int sd_varlink_call(
return sd_varlink_call_full(v, method, parameters, ret_parameters, ret_error_id, NULL);
}
static int varlink_handle_upgrade_fds(sd_varlink *v, int *ret_input_fd, int *ret_output_fd) {
int r;
assert(v);
assert(ret_input_fd || ret_output_fd);
/* Ensure no post-upgrade data was consumed into our input buffer (we ensure this via MSG_PEEK or
* byte-to-byte) and refuse the upgrade rather than silently losing the data. */
if (json_stream_has_buffered_input(&v->stream))
return varlink_log_errno(v, SYNTHETIC_ERRNO(EPROTO),
"Unexpected buffered data during protocol upgrade, refusing.");
_cleanup_close_ int input_fd = TAKE_FD(v->stream.input_fd),
output_fd = TAKE_FD(v->stream.output_fd);
/* Pass the connection fds to the caller, it owns them now. Reset to blocking mode
* since callers of the upgraded protocol will generally expect normal blocking
* semantics. For bidirectional sockets (input_fd == output_fd), dup the fd so that
* callers always get two independent fds they can close separately. */
if (input_fd == output_fd) {
output_fd = fcntl(input_fd, F_DUPFD_CLOEXEC, 3);
if (output_fd < 0)
return varlink_log_errno(v, errno, "Failed to dup upgraded connection fd: %m");
} else {
r = fd_nonblock(output_fd, false);
if (r < 0)
return varlink_log_errno(v, r, "Failed to set output fd to blocking mode: %m");
}
r = fd_nonblock(input_fd, false);
if (r < 0)
return varlink_log_errno(v, r, "Failed to set input fd to blocking mode: %m");
/* Hand out requested fds, shut down unwanted directions. */
if (ret_input_fd)
*ret_input_fd = TAKE_FD(input_fd);
else
(void) shutdown(input_fd, SHUT_RD);
if (ret_output_fd)
*ret_output_fd = TAKE_FD(output_fd);
else
(void) shutdown(output_fd, SHUT_WR);
return 0;
}
_public_ int sd_varlink_call_and_upgrade(
sd_varlink *v,
const char *method,
@@ -2076,11 +2130,9 @@ _public_ int sd_varlink_call_and_upgrade(
if (r < 0)
return varlink_log_errno(v, r, "Failed to build json message: %m");
v->protocol_upgrade = true;
json_stream_set_flags(&v->stream, JSON_STREAM_BOUNDED_READS, true);
r = varlink_call_internal(v, m);
if (r < 0) {
v->protocol_upgrade = false;
json_stream_set_flags(&v->stream, JSON_STREAM_BOUNDED_READS, false);
return r;
}
@@ -2125,13 +2177,37 @@ _public_ int sd_varlink_call_and_upgrade(
return 1;
finish:
v->protocol_upgrade = false;
json_stream_set_flags(&v->stream, JSON_STREAM_BOUNDED_READS, false);
assert(v->n_pending == 1);
v->n_pending--;
return r;
}
_public_ int sd_varlink_call_and_upgradeb(
sd_varlink *v,
const char *method,
sd_json_variant **ret_parameters,
const char **ret_error_id,
int *ret_input_fd,
int *ret_output_fd,
...) {
_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, ret_output_fd);
r = sd_json_buildv(&parameters, ap);
va_end(ap);
if (r < 0)
return varlink_log_errno(v, r, "Failed to build json message: %m");
return sd_varlink_call_and_upgrade(v, method, parameters, ret_parameters, ret_error_id, ret_input_fd, ret_output_fd);
}
_public_ int sd_varlink_callb_ap(
sd_varlink *v,
const char *method,
@@ -2360,8 +2436,8 @@ static int varlink_reply_internal(sd_varlink *v, sd_json_variant *parameters, bo
return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected.");
if (!IN_SET(v->state,
VARLINK_PROCESSING_METHOD, VARLINK_PROCESSING_METHOD_MORE,
VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE))
VARLINK_PROCESSING_METHOD, VARLINK_PROCESSING_METHOD_MORE, VARLINK_PROCESSING_METHOD_UPGRADE,
VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE, VARLINK_PENDING_METHOD_UPGRADE))
return varlink_log_errno(v, SYNTHETIC_ERRNO(EBUSY), "Connection busy.");
bool more = IN_SET(v->state, VARLINK_PROCESSING_METHOD_MORE, VARLINK_PENDING_METHOD_MORE);
@@ -2413,7 +2489,7 @@ static int varlink_reply_internal(sd_varlink *v, sd_json_variant *parameters, bo
if (r < 0)
return varlink_log_errno(v, r, "Failed to enqueue json message: %m");
if (IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE)) {
if (IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE, VARLINK_PENDING_METHOD_UPGRADE)) {
/* We just replied to a method call that was let hanging for a while (i.e. we were outside of
* the varlink_dispatch_method() stack frame), which means with this reply we are ready to
* process further messages. */
@@ -2454,30 +2530,33 @@ _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.");
/* Return -EBUSY if we are not processing methods at all */
if (!IN_SET(v->state,
VARLINK_PROCESSING_METHOD,
VARLINK_PENDING_METHOD))
VARLINK_PROCESSING_METHOD, VARLINK_PROCESSING_METHOD_MORE, VARLINK_PROCESSING_METHOD_ONEWAY, VARLINK_PROCESSING_METHOD_UPGRADE,
VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE, VARLINK_PENDING_METHOD_UPGRADE))
return varlink_log_errno(v, SYNTHETIC_ERRNO(EBUSY), "Connection busy.");
/* Verify the client actually requested a protocol upgrade */
if (!v->protocol_upgrade)
return varlink_log_errno(v, SYNTHETIC_ERRNO(EPROTO),
"Method call did not request a protocol upgrade.");
/* Return -EPROTO if we are processing methods, but an upgrade was not requested. */
if (!IN_SET(v->state,
VARLINK_PROCESSING_METHOD_UPGRADE,
VARLINK_PENDING_METHOD_UPGRADE))
return varlink_log_errno(v, SYNTHETIC_ERRNO(EPROTO), "Method call did not request a protocol upgrade.");
/* Ensure we did not buffer any data beyond the upgrade request. Check this before sending the
* reply so that we can return a normal error (the framework will send an error reply to the
* client). In normal operation this cannot happen because the client waits for our reply before
* sending raw data, and we set protocol_upgrade=true in dispatch to limit subsequent reads to
* single bytes. But a misbehaving client could pipeline data early. */
/* Ensure we did not buffer any data beyond the upgrade request. Check this before sending the reply
* so that we can return a normal error (the framework will send an error reply to the client). In
* normal operation this cannot happen because the client waits for our reply before sending raw
* data, and we disabled unbounded reads in dispatch to limit subsequent reads to single bytes. But a
* misbehaving client could pipeline data early. */
if (json_stream_has_buffered_input(&v->stream))
return varlink_log_errno(v, SYNTHETIC_ERRNO(EBADMSG),
"Unexpected buffered data from client during protocol upgrade.");
@@ -2503,6 +2582,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);
@@ -2548,8 +2676,8 @@ _public_ int sd_varlink_error(sd_varlink *v, const char *error_id, sd_json_varia
if (v->state == VARLINK_DISCONNECTED)
return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected.");
if (!IN_SET(v->state,
VARLINK_PROCESSING_METHOD, VARLINK_PROCESSING_METHOD_MORE,
VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE))
VARLINK_PROCESSING_METHOD, VARLINK_PROCESSING_METHOD_MORE, VARLINK_PROCESSING_METHOD_UPGRADE,
VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE, VARLINK_PENDING_METHOD_UPGRADE))
return varlink_log_errno(v, SYNTHETIC_ERRNO(EBUSY), "Connection busy.");
if (v->previous) {
@@ -2601,7 +2729,7 @@ _public_ int sd_varlink_error(sd_varlink *v, const char *error_id, sd_json_varia
if (r < 0)
return varlink_log_errno(v, r, "Failed to enqueue json message: %m");
if (IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE)) {
if (IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE, VARLINK_PENDING_METHOD_UPGRADE)) {
varlink_clear_current(v);
varlink_set_state(v, VARLINK_IDLE_SERVER);
} else
@@ -2792,6 +2920,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
+12 -9
View File
@@ -26,9 +26,13 @@ typedef enum VarlinkState {
VARLINK_PROCESSING_METHOD,
VARLINK_PROCESSING_METHOD_MORE,
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,
@@ -60,16 +64,21 @@ typedef enum VarlinkState {
VARLINK_PROCESSING_METHOD, \
VARLINK_PROCESSING_METHOD_MORE, \
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_MORE, \
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. */
#define VARLINK_STATE_WANTS_REPLY(state) \
IN_SET(state, \
VARLINK_PROCESSING_METHOD, \
VARLINK_PROCESSING_METHOD_MORE)
VARLINK_PROCESSING_METHOD_MORE, \
VARLINK_PROCESSING_METHOD_UPGRADE)
typedef struct sd_varlink {
unsigned n_ref;
@@ -86,14 +95,8 @@ typedef struct sd_varlink {
unsigned n_pending;
/* Per-call protocol-upgrade marker: set when the *current* method call carries the
* SD_VARLINK_METHOD_UPGRADE flag. Validated by sd_varlink_reply_and_upgrade() to
* ensure the caller's contract is honored. The transport-layer "stop reading at the
* next message boundary" behavior is governed independently by the JsonStream's
* bounded_reads flag. */
bool protocol_upgrade;
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
+13
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);
@@ -144,6 +145,9 @@ int sd_varlink_callb(sd_varlink *v, const char *method, sd_json_variant **ret_pa
* Returns > 0 if the connection was upgraded, 0 if a Varlink error occurred (and ret_error_id was set),
* or < 0 on local failure. */
int sd_varlink_call_and_upgrade(sd_varlink *v, const char *method, sd_json_variant *parameters, sd_json_variant **ret_parameters, const char **ret_error_id, int *ret_input_fd, int *ret_output_fd);
int sd_varlink_call_and_upgradeb(sd_varlink *v, const char *method, sd_json_variant **ret_parameters, const char **ret_error_id, int *ret_input_fd, int *ret_output_fd, ...);
#define sd_varlink_call_and_upgradebo(v, method, ret_parameters, ret_error_id, ret_input_fd, ret_output_fd, ...) \
sd_varlink_call_and_upgradeb((v), (method), (ret_parameters), (ret_error_id), (ret_input_fd), (ret_output_fd), SD_JSON_BUILD_OBJECT(__VA_ARGS__))
/* Send method call and begin collecting all 'more' replies into an array, finishing when a final reply is sent */
int sd_varlink_collect_full(sd_varlink *v, const char *method, sd_json_variant *parameters, sd_json_variant **ret_parameters, const char **ret_error_id, sd_varlink_reply_flags_t *ret_flags);
@@ -182,6 +186,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 +234,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);