From 45a6c965986d484bd1d7d7d1978dbf0d82baec69 Mon Sep 17 00:00:00 2001 From: Lennart Poettering Date: Wed, 7 Aug 2019 12:44:13 +0200 Subject: [PATCH 1/5] varlink: fix support for more/continues method calls --- src/shared/varlink.c | 99 ++++++++++++++++++++++++++++++++++++-------- src/shared/varlink.h | 4 ++ 2 files changed, 85 insertions(+), 18 deletions(-) diff --git a/src/shared/varlink.c b/src/shared/varlink.c index 99343167f6..7a566762fa 100644 --- a/src/shared/varlink.c +++ b/src/shared/varlink.c @@ -29,6 +29,7 @@ typedef enum VarlinkState { /* Client side states */ VARLINK_IDLE_CLIENT, VARLINK_AWAITING_REPLY, + VARLINK_AWAITING_REPLY_MORE, VARLINK_CALLING, VARLINK_CALLED, VARLINK_PROCESSING_REPLY, @@ -39,7 +40,6 @@ typedef enum VarlinkState { VARLINK_PROCESSING_METHOD_MORE, VARLINK_PROCESSING_METHOD_ONEWAY, VARLINK_PROCESSED_METHOD, - VARLINK_PROCESSED_METHOD_MORE, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE, @@ -63,6 +63,7 @@ typedef enum VarlinkState { IN_SET(state, \ VARLINK_IDLE_CLIENT, \ VARLINK_AWAITING_REPLY, \ + VARLINK_AWAITING_REPLY_MORE, \ VARLINK_CALLING, \ VARLINK_CALLED, \ VARLINK_PROCESSING_REPLY, \ @@ -71,7 +72,6 @@ typedef enum VarlinkState { VARLINK_PROCESSING_METHOD_MORE, \ VARLINK_PROCESSING_METHOD_ONEWAY, \ VARLINK_PROCESSED_METHOD, \ - VARLINK_PROCESSED_METHOD_MORE, \ VARLINK_PENDING_METHOD, \ VARLINK_PENDING_METHOD_MORE) @@ -185,6 +185,7 @@ struct VarlinkServer { 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_PROCESSING_REPLY] = "processing-reply", @@ -193,7 +194,6 @@ static const char* const varlink_state_table[_VARLINK_STATE_MAX] = { [VARLINK_PROCESSING_METHOD_MORE] = "processing-method-more", [VARLINK_PROCESSING_METHOD_ONEWAY] = "processing-method-oneway", [VARLINK_PROCESSED_METHOD] = "processed-method", - [VARLINK_PROCESSED_METHOD_MORE] = "processed-method-more", [VARLINK_PENDING_METHOD] = "pending-method", [VARLINK_PENDING_METHOD_MORE] = "pending-method-more", [VARLINK_PENDING_DISCONNECT] = "pending-disconnect", @@ -405,7 +405,7 @@ static int varlink_test_disconnect(Varlink *v) { goto disconnect; /* If we are waiting for incoming data but the read side is shut down, disconnect. */ - if (IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_CALLING, VARLINK_IDLE_SERVER) && v->read_disconnected) + if (IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_CALLING, VARLINK_IDLE_SERVER) && v->read_disconnected) goto disconnect; /* Similar, if are a client that hasn't written anything yet but the write side is dead, also @@ -478,7 +478,7 @@ static int varlink_read(Varlink *v) { assert(v); - if (!IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_CALLING, VARLINK_IDLE_SERVER)) + if (!IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_CALLING, VARLINK_IDLE_SERVER)) return 0; if (v->connecting) /* read() on a socket while we are in connect() will fail with EINVAL, hence exit early here */ return 0; @@ -596,7 +596,7 @@ static int varlink_parse_message(Varlink *v) { static int varlink_test_timeout(Varlink *v) { assert(v); - if (!IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_CALLING)) + if (!IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_CALLING)) return 0; if (v->timeout == USEC_INFINITY) return 0; @@ -673,7 +673,7 @@ static int varlink_dispatch_reply(Varlink *v) { assert(v); - if (!IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_CALLING)) + if (!IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_CALLING)) return 0; if (!v->current) return 0; @@ -715,6 +715,11 @@ static int varlink_dispatch_reply(Varlink *v) { goto invalid; } + /* Replies with 'continue' set are only OK if we set 'more' when the method call was initiated */ + if (v->state != VARLINK_AWAITING_REPLY_MORE && FLAGS_SET(flags, VARLINK_REPLY_CONTINUES)) + goto invalid; + + /* An error is final */ if (error && FLAGS_SET(flags, VARLINK_REPLY_CONTINUES)) goto invalid; @@ -722,7 +727,7 @@ static int varlink_dispatch_reply(Varlink *v) { if (r < 0) goto invalid; - if (v->state == VARLINK_AWAITING_REPLY) { + if (IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE)) { varlink_set_state(v, VARLINK_PROCESSING_REPLY); if (v->reply_callback) { @@ -734,17 +739,18 @@ static int varlink_dispatch_reply(Varlink *v) { v->current = json_variant_unref(v->current); if (v->state == VARLINK_PROCESSING_REPLY) { - assert(v->n_pending > 0); - v->n_pending--; - varlink_set_state(v, v->n_pending == 0 ? VARLINK_IDLE_CLIENT : VARLINK_AWAITING_REPLY); + assert(v->n_pending > 0); + + if (!FLAGS_SET(flags, VARLINK_REPLY_CONTINUES)) + v->n_pending--; + + varlink_set_state(v, + FLAGS_SET(flags, VARLINK_REPLY_CONTINUES) ? VARLINK_AWAITING_REPLY_MORE : + v->n_pending == 0 ? VARLINK_IDLE_CLIENT : VARLINK_AWAITING_REPLY); } } else { assert(v->state == VARLINK_CALLING); - - if (FLAGS_SET(flags, VARLINK_REPLY_CONTINUES)) - goto invalid; - varlink_set_state(v, VARLINK_CALLED); } @@ -878,7 +884,6 @@ static int varlink_dispatch_method(Varlink *v) { varlink_set_state(v, VARLINK_PENDING_METHOD); break; - case VARLINK_PROCESSED_METHOD_MORE: /* One reply for a "more" message was sent, more to come */ case VARLINK_PROCESSING_METHOD_MORE: /* No reply for a "more" message was sent, more to come */ varlink_set_state(v, VARLINK_PENDING_METHOD_MORE); break; @@ -1073,7 +1078,7 @@ int varlink_get_events(Varlink *v) { return EPOLLOUT; if (!v->read_disconnected && - IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_CALLING, VARLINK_IDLE_SERVER) && + IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_CALLING, VARLINK_IDLE_SERVER) && !v->current && v->input_buffer_unscanned <= 0) ret |= EPOLLIN; @@ -1091,7 +1096,7 @@ int varlink_get_timeout(Varlink *v, usec_t *ret) { if (v->state == VARLINK_DISCONNECTED) return -ENOTCONN; - if (IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_CALLING) && + if (IN_SET(v->state, VARLINK_AWAITING_REPLY, VARLINK_AWAITING_REPLY_MORE, VARLINK_CALLING) && v->timeout != USEC_INFINITY) { if (ret) *ret = usec_add(v->timestamp, v->timeout); @@ -1259,6 +1264,8 @@ int varlink_send(Varlink *v, const char *method, JsonVariant *parameters) { if (v->state == VARLINK_DISCONNECTED) return -ENOTCONN; + + /* We allow enqueuing multiple method calls at once! */ if (!IN_SET(v->state, VARLINK_IDLE_CLIENT, VARLINK_AWAITING_REPLY)) return -EBUSY; @@ -1308,6 +1315,8 @@ int varlink_invoke(Varlink *v, const char *method, JsonVariant *parameters) { if (v->state == VARLINK_DISCONNECTED) return -ENOTCONN; + + /* We allow enqueing multiple method calls at once! */ if (!IN_SET(v->state, VARLINK_IDLE_CLIENT, VARLINK_AWAITING_REPLY)) return -EBUSY; @@ -1349,6 +1358,60 @@ int varlink_invokeb(Varlink *v, const char *method, ...) { return varlink_invoke(v, method, parameters); } +int varlink_observe(Varlink *v, const char *method, JsonVariant *parameters) { + _cleanup_(json_variant_unrefp) JsonVariant *m = NULL; + int r; + + assert_return(v, -EINVAL); + assert_return(method, -EINVAL); + + if (v->state == VARLINK_DISCONNECTED) + return -ENOTCONN; + /* Note that we don't allow enqueuing multiple method calls when we are in more/continues mode! We + * thus insist on an idle client here. */ + if (v->state != VARLINK_IDLE_CLIENT) + return -EBUSY; + + r = varlink_sanitize_parameters(¶meters); + if (r < 0) + return r; + + r = json_build(&m, JSON_BUILD_OBJECT( + JSON_BUILD_PAIR("method", JSON_BUILD_STRING(method)), + JSON_BUILD_PAIR("parameters", JSON_BUILD_VARIANT(parameters)), + JSON_BUILD_PAIR("more", JSON_BUILD_BOOLEAN(true)))); + if (r < 0) + return r; + + r = varlink_enqueue_json(v, m); + if (r < 0) + return r; + + + varlink_set_state(v, VARLINK_AWAITING_REPLY_MORE); + v->n_pending++; + v->timestamp = now(CLOCK_MONOTONIC); + + return 0; +} + +int varlink_observeb(Varlink *v, const char *method, ...) { + _cleanup_(json_variant_unrefp) JsonVariant *parameters = NULL; + va_list ap; + int r; + + assert_return(v, -EINVAL); + + va_start(ap, method); + r = json_buildv(¶meters, ap); + va_end(ap); + + if (r < 0) + return r; + + return varlink_observe(v, method, parameters); +} + int varlink_call( Varlink *v, const char *method, diff --git a/src/shared/varlink.h b/src/shared/varlink.h index d96fa93619..0208802010 100644 --- a/src/shared/varlink.h +++ b/src/shared/varlink.h @@ -86,6 +86,10 @@ int varlink_callb(Varlink *v, const char *method, JsonVariant **ret_parameters, int varlink_invoke(Varlink *v, const char *method, JsonVariant *parameters); int varlink_invokeb(Varlink *v, const char *method, ...); +/* Enqueue method call, expect a reply now, and possibly more later, which are all delivered to the reply callback */ +int varlink_observe(Varlink *v, const char *method, JsonVariant *parameters); +int varlink_observeb(Varlink *v, const char *method, ...); + /* Enqueue a final reply */ int varlink_reply(Varlink *v, JsonVariant *parameters); int varlink_replyb(Varlink *v, ...); From a0c41de277234b57bbcd6a315c9fcc5ec64e9f7c Mon Sep 17 00:00:00 2001 From: Lennart Poettering Date: Thu, 15 Aug 2019 09:34:05 +0200 Subject: [PATCH 2/5] varlink: move connection fds > fd2 We want to use this code in NSS modules, and we never know the execution environment we are run in there, hence let's move our fds up to ensure we won't step into dangerous fd territory. This is similar to how we already do it in sd-bus for client connection fds. --- src/shared/varlink.c | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/shared/varlink.c b/src/shared/varlink.c index 7a566762fa..a23525b0a4 100644 --- a/src/shared/varlink.c +++ b/src/shared/varlink.c @@ -287,6 +287,8 @@ int varlink_connect_address(Varlink **ret, const char *address) { if (v->fd < 0) return -errno; + v->fd = fd_move_above_stdio(v->fd); + if (connect(v->fd, &sockaddr.sa, SOCKADDR_UN_LEN(sockaddr.un)) < 0) { if (!IN_SET(errno, EAGAIN, EINPROGRESS)) return -errno; @@ -2220,6 +2222,8 @@ int varlink_server_listen_address(VarlinkServer *s, const char *address, mode_t if (fd < 0) return -errno; + fd = fd_move_above_stdio(fd); + (void) sockaddr_un_unlink(&sockaddr.un); RUN_WITH_UMASK(~m & 0777) From 88a36d369015674e3f9eb20f0fc4d33beedd91bb Mon Sep 17 00:00:00 2001 From: Lennart Poettering Date: Tue, 20 Aug 2019 14:07:09 +0200 Subject: [PATCH 3/5] varlink: port varlink code over to use getdtablesize() for sizing number of concurrent connections Use the official glibc API for determining this parameter. In most other cases in our tree it's better to go directly for RLIMIT_NOFILE since it's semantically what we want, but for this case it appears more appropriate to use the friendlier, shorter, explicit API. --- src/shared/varlink.c | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/src/shared/varlink.c b/src/shared/varlink.c index a23525b0a4..ac8930302a 100644 --- a/src/shared/varlink.c +++ b/src/shared/varlink.c @@ -2406,17 +2406,18 @@ int varlink_server_bind_connect(VarlinkServer *s, VarlinkConnect callback) { } unsigned varlink_server_connections_max(VarlinkServer *s) { - struct rlimit rl; + int dts; /* If a server is specified, return the setting for that server, otherwise the default value */ if (s) return s->connections_max; - assert_se(getrlimit(RLIMIT_NOFILE, &rl) >= 0); + dts = getdtablesize(); + assert_se(dts > 0); /* Make sure we never use up more than ¾th of RLIMIT_NOFILE for IPC */ - if (VARLINK_DEFAULT_CONNECTIONS_MAX > rl.rlim_cur / 4 * 3) - return rl.rlim_cur / 4 * 3; + if (VARLINK_DEFAULT_CONNECTIONS_MAX > (unsigned) dts / 4 * 3) + return dts / 4 * 3; return VARLINK_DEFAULT_CONNECTIONS_MAX; } From c4fe0cbc17ced6ea6af16af14dc102f1763594fd Mon Sep 17 00:00:00 2001 From: Lennart Poettering Date: Thu, 21 Nov 2019 12:14:44 +0100 Subject: [PATCH 4/5] varlink: drop too much whitespace --- src/shared/varlink.c | 3 --- 1 file changed, 3 deletions(-) diff --git a/src/shared/varlink.c b/src/shared/varlink.c index ac8930302a..55098a1212 100644 --- a/src/shared/varlink.c +++ b/src/shared/varlink.c @@ -343,11 +343,8 @@ static void varlink_detach_event_sources(Varlink *v) { assert(v); v->io_event_source = sd_event_source_disable_unref(v->io_event_source); - v->time_event_source = sd_event_source_disable_unref(v->time_event_source); - v->quit_event_source = sd_event_source_disable_unref(v->quit_event_source); - v->defer_event_source = sd_event_source_disable_unref(v->defer_event_source); } From f1194f5d592019e26a682695e1040612ec1d97b7 Mon Sep 17 00:00:00 2001 From: Lennart Poettering Date: Thu, 21 Nov 2019 12:14:58 +0100 Subject: [PATCH 5/5] varlink: fix enablement of varlink timeout event source --- src/shared/varlink.c | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/shared/varlink.c b/src/shared/varlink.c index 55098a1212..ee4fb9e843 100644 --- a/src/shared/varlink.c +++ b/src/shared/varlink.c @@ -1831,6 +1831,7 @@ static int prepare_callback(sd_event_source *s, void *userdata) { Varlink *v = userdata; int r, e; usec_t until; + bool have_timeout; assert(s); assert(v); @@ -1846,13 +1847,15 @@ static int prepare_callback(sd_event_source *s, void *userdata) { r = varlink_get_timeout(v, &until); if (r < 0) return r; - if (r > 0) { + have_timeout = r > 0; + + if (have_timeout) { r = sd_event_source_set_time(v->time_event_source, until); if (r < 0) return r; } - r = sd_event_source_set_enabled(v->time_event_source, r > 0 ? SD_EVENT_ON : SD_EVENT_OFF); + r = sd_event_source_set_enabled(v->time_event_source, have_timeout ? SD_EVENT_ON : SD_EVENT_OFF); if (r < 0) return r;