static void bus_reset_queues(sd_bus *b) {
assert(b);
- /* NOTE: We _must_ decrement b->Xqueue_size before calling
- * sd_bus_message_unref() for _each_ message. Otherwise the
- * self-reference checks in sd_bus_unref() will fire for each message.
- * We would thus recurse into sd_bus_message_unref() and trigger the
- * assert(m->n_ref > 0) */
-
while (b->rqueue_size > 0)
sd_bus_message_unref(b->rqueue[--b->rqueue_size]);
else
return -EINVAL;
- if (r < 0)
+ if (r < 0) {
+ sd_bus_close(bus);
return r;
+ }
return bus_send_hello(bus);
}
int bus_set_address_system_remote(sd_bus *b, const char *host) {
_cleanup_free_ char *e = NULL;
+ char *m = NULL, *c = NULL;
assert(b);
assert(host);
- e = bus_address_escape(host);
- if (!e)
- return -ENOMEM;
+ /* Let's see if we shall enter some container */
+ m = strchr(host, ':');
+ if (m) {
+ m++;
+
+ /* Let's make sure this is not a port of some kind,
+ * and is a valid machine name. */
+ if (!in_charset(m, "0123456789") && machine_name_is_valid(m)) {
+ char *t;
+
+ /* Cut out the host part */
+ t = strndupa(host, m - host - 1);
+ e = bus_address_escape(t);
+ if (!e)
+ return -ENOMEM;
- b->address = strjoin("unixexec:path=ssh,argv1=-xT,argv2=", e, ",argv3=systemd-stdio-bridge", NULL);
+ c = strappenda(",argv4=--machine=", m);
+ }
+ }
+
+ if (!e) {
+ e = bus_address_escape(host);
+ if (!e)
+ return -ENOMEM;
+ }
+
+ b->address = strjoin("unixexec:path=ssh,argv1=-xT,argv2=", e, ",argv3=systemd-stdio-bridge", c, NULL);
if (!b->address)
return -ENOMEM;
if (!bus)
return NULL;
- /* TODO/FIXME: It's naive to think REFCNT_GET() is thread-safe in any
- * way but exclusive REFCNT_DEC(). The current logic _must_ lock around
- * REFCNT_GET() until REFCNT_DEC() or two threads might end up in
- * parallel in bus_reset_queues(). But locking would totally break the
- * recursion we introduce by bus_reset_queues()...
- * (Imagine one thread in sd_bus_message_unref() setting n_ref to 0 and
- * thus calling into sd_bus_unref(). If at the same time the real
- * thread calls sd_bus_unref(), both end up with "q == true" and will
- * call into bus_reset_queues().
- * If we require the main bus to be alive until all dispatch threads
- * are done, there is no need to do ref-counts at all. So in both ways,
- * the REFCNT thing is humbug.)
- *
- * On a second note: messages are *not* required to have ->bus set nor
- * does it have to be _this_ bus that they're assigned to. This whole
- * ref-cnt checking breaks apart if a message is not assigned to us.
- * (which is _very_ easy to trigger with the current API). */
-
- if (REFCNT_GET(bus->n_ref) == bus->rqueue_size + bus->wqueue_size + 1) {
- bool q = true;
-
- for (i = 0; i < bus->rqueue_size; i++)
- if (bus->rqueue[i]->n_ref > 1) {
- q = false;
- break;
- }
-
- if (q) {
- for (i = 0; i < bus->wqueue_size; i++)
- if (bus->wqueue[i]->n_ref > 1) {
- q = false;
- break;
- }
- }
-
- /* We are the only holders on the messages, and the
- * messages are the only holders on us, so let's drop
- * the messages and thus implicitly also kill our own
- * last references.
- * bus_reset_queues() decrements the queue-size before
- * calling into sd_bus_message_unref(). Thus, it
- * protects us from recursion. */
-
- if (q)
- bus_reset_queues(bus);
- }
-
i = REFCNT_DEC(bus->n_ref);
if (i > 0)
return NULL;
_cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
_cleanup_bus_message_unref_ sd_bus_message* m = NULL;
struct reply_callback *c;
+ sd_bus_slot *slot;
usec_t n;
int r;
hashmap_remove(bus->reply_callbacks, &c->cookie);
c->cookie = 0;
- bus->current_message = m;
- bus->current_slot = container_of(c, sd_bus_slot, reply_callback);
+ slot = container_of(c, sd_bus_slot, reply_callback);
bus->iteration_counter ++;
- r = c->callback(bus, m, bus->current_slot->userdata, &error_buffer);
- r = bus_maybe_reply_error(m, r, &error_buffer);
-
+ bus->current_message = m;
+ bus->current_slot = sd_bus_slot_ref(slot);
+ r = c->callback(bus, m, slot->userdata, &error_buffer);
+ bus->current_slot = sd_bus_slot_unref(slot);
bus->current_message = NULL;
- bus->current_slot = NULL;
- return r;
+ if (slot->floating) {
+ bus_slot_disconnect(slot);
+ sd_bus_slot_unref(slot);
+ }
+
+ return bus_maybe_reply_error(m, r, &error_buffer);
}
static int process_hello(sd_bus *bus, sd_bus_message *m) {
static int process_reply(sd_bus *bus, sd_bus_message *m) {
_cleanup_bus_message_unref_ sd_bus_message *synthetic_reply = NULL;
_cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
- sd_bus_slot *slot;
struct reply_callback *c;
+ sd_bus_slot *slot;
int r;
assert(bus);
return 0;
c->cookie = 0;
- slot = container_of(c, sd_bus_slot, reply_callback);
- if (c->timeout != 0) {
- prioq_remove(bus->reply_callbacks_prioq, c, &c->prioq_idx);
- c->timeout = 0;
- }
+ slot = container_of(c, sd_bus_slot, reply_callback);
if (m->n_fds > 0 && !(bus->hello_flags & KDBUS_HELLO_ACCEPT_FD)) {
&SD_BUS_ERROR_MAKE_CONST(SD_BUS_ERROR_INCONSISTENT_MESSAGE, "Reply message contained file descriptor"),
&synthetic_reply);
if (r < 0)
- goto finish;
+ return r;
r = bus_seal_synthetic_message(bus, synthetic_reply);
if (r < 0)
- goto finish;
+ return r;
m = synthetic_reply;
} else {
r = sd_bus_message_rewind(m, true);
if (r < 0)
- goto finish;
+ return r;
}
- bus->current_slot = slot;
- r = c->callback(bus, m, bus->current_slot->userdata, &error_buffer);
- bus->current_slot = NULL;
+ if (c->timeout != 0) {
+ prioq_remove(bus->reply_callbacks_prioq, c, &c->prioq_idx);
+ c->timeout = 0;
+ }
- r = bus_maybe_reply_error(m, r, &error_buffer);
+ bus->current_slot = sd_bus_slot_ref(slot);
+ r = c->callback(bus, m, slot->userdata, &error_buffer);
+ bus->current_slot = sd_bus_slot_unref(slot);
-finish:
if (slot->floating) {
bus_slot_disconnect(slot);
sd_bus_slot_unref(slot);
}
- return r;
+ return bus_maybe_reply_error(m, r, &error_buffer);
}
static int process_filter(sd_bus *bus, sd_bus_message *m) {
bus->filter_callbacks_modified = false;
LIST_FOREACH(callbacks, l, bus->filter_callbacks) {
+ sd_bus_slot *slot;
if (bus->filter_callbacks_modified)
break;
if (r < 0)
return r;
- bus->current_slot = container_of(l, sd_bus_slot, filter_callback);
- r = l->callback(bus, m, bus->current_slot->userdata, &error_buffer);
- bus->current_slot = NULL;
+ slot = container_of(l, sd_bus_slot, filter_callback);
+
+ bus->current_slot = sd_bus_slot_ref(slot);
+ r = l->callback(bus, m, slot->userdata, &error_buffer);
+ bus->current_slot = sd_bus_slot_unref(slot);
r = bus_maybe_reply_error(m, r, &error_buffer);
if (r != 0)
c = hashmap_first(bus->reply_callbacks);
if (c) {
_cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
+ sd_bus_slot *slot;
/* First, fail all outstanding method calls */
r = bus_message_new_synthetic_error(
hashmap_remove(bus->reply_callbacks, &c->cookie);
c->cookie = 0;
- bus->current_message = m;
- bus->current_slot = container_of(c, sd_bus_slot, reply_callback);
+ slot = container_of(c, sd_bus_slot, reply_callback);
bus->iteration_counter++;
- r = c->callback(bus, m, bus->current_slot->userdata, &error_buffer);
- r = bus_maybe_reply_error(m, r, &error_buffer);
-
- bus->current_slot = NULL;
+ bus->current_message = m;
+ bus->current_slot = sd_bus_slot_ref(slot);
+ r = c->callback(bus, m, slot->userdata, &error_buffer);
+ bus->current_slot = sd_bus_slot_unref(slot);
+ bus->current_message = NULL;
+
+ if (slot->floating) {
+ bus_slot_disconnect(slot);
+ sd_bus_slot_unref(slot);
+ }
- goto finish;
+ return bus_maybe_reply_error(m, r, &error_buffer);
}
/* Then, synthesize a Disconnected message */
struct pollfd p[2] = {};
int r, e, n;
struct timespec ts;
- usec_t m = (usec_t) -1;
+ usec_t m = USEC_INFINITY;
assert(bus);
struct bus_match_component *components = NULL;
unsigned n_components = 0;
- sd_bus_slot *s;
+ sd_bus_slot *s = NULL;
int r = 0;
assert_return(bus, -EINVAL);
assert(event);
sd_bus_flush(bus);
+ sd_bus_close(bus);
return 1;
}
~ianmdlvl / elogind.git / blobdiff