diff --git a/src/libsystemd/sd-netlink/netlink-internal.h b/src/libsystemd/sd-netlink/netlink-internal.h index 514f22511c..bca13bce57 100644 --- a/src/libsystemd/sd-netlink/netlink-internal.h +++ b/src/libsystemd/sd-netlink/netlink-internal.h @@ -7,6 +7,7 @@ #include "list.h" #include "netlink-types.h" +#include "ordered-set.h" #include "prioq.h" #include "time-util.h" @@ -72,11 +73,9 @@ struct sd_netlink { Hashmap *broadcast_group_refs; bool broadcast_group_dont_leave:1; /* until we can rely on 4.2 */ - sd_netlink_message **rqueue; - unsigned rqueue_size; - - sd_netlink_message **rqueue_partial; - unsigned rqueue_partial_size; + OrderedSet *rqueue; + Hashmap *rqueue_by_serial; + Hashmap *rqueue_partial_by_serial; struct nlmsghdr *rbuffer; @@ -148,8 +147,6 @@ void message_seal(sd_netlink_message *m); int netlink_open_family(sd_netlink **ret, int family); bool netlink_pid_changed(sd_netlink *nl); -int netlink_rqueue_make_room(sd_netlink *nl); -int netlink_rqueue_partial_make_room(sd_netlink *nl); int socket_bind(sd_netlink *nl); int socket_broadcast_group_ref(sd_netlink *nl, unsigned group); diff --git a/src/libsystemd/sd-netlink/netlink-socket.c b/src/libsystemd/sd-netlink/netlink-socket.c index 1da459c014..96162963a7 100644 --- a/src/libsystemd/sd-netlink/netlink-socket.c +++ b/src/libsystemd/sd-netlink/netlink-socket.c @@ -180,11 +180,12 @@ int socket_write_message(sd_netlink *nl, sd_netlink_message *m) { return k; } -static int socket_recv_message(int fd, struct iovec *iov, uint32_t *ret_mcast_group, bool peek) { +static int socket_recv_message(int fd, void *buf, size_t buf_size, uint32_t *ret_mcast_group, bool peek) { + struct iovec iov = IOVEC_MAKE(buf, buf_size); union sockaddr_union sender; CMSG_BUFFER_TYPE(CMSG_SPACE(sizeof(struct nl_pktinfo))) control; struct msghdr msg = { - .msg_iov = iov, + .msg_iov = &iov, .msg_iovlen = 1, .msg_name = &sender, .msg_namelen = sizeof(sender), @@ -194,14 +195,17 @@ static int socket_recv_message(int fd, struct iovec *iov, uint32_t *ret_mcast_gr ssize_t n; assert(fd >= 0); - assert(iov); + assert(peek || (buf && buf_size > 0)); n = recvmsg_safe(fd, &msg, MSG_TRUNC | (peek ? MSG_PEEK : 0)); if (n < 0) { if (n == -ENOBUFS) return log_debug_errno(n, "sd-netlink: kernel receive buffer overrun"); - if (ERRNO_IS_TRANSIENT(n)) + if (ERRNO_IS_TRANSIENT(n)) { + if (ret_mcast_group) + *ret_mcast_group = 0; return 0; + } return (int) n; } @@ -216,9 +220,14 @@ static int socket_recv_message(int fd, struct iovec *iov, uint32_t *ret_mcast_gr return (int) n; } + if (ret_mcast_group) + *ret_mcast_group = 0; return 0; } + if (!peek && (size_t) n > buf_size) /* message did not fit in read buffer */ + return -EIO; + if (ret_mcast_group) { struct nl_pktinfo *pi; @@ -232,151 +241,221 @@ static int socket_recv_message(int fd, struct iovec *iov, uint32_t *ret_mcast_gr return (int) n; } +DEFINE_PRIVATE_HASH_OPS_WITH_VALUE_DESTRUCTOR( + netlink_message_hash_ops, + void, trivial_hash_func, trivial_compare_func, + sd_netlink_message, sd_netlink_message_unref); + +static int netlink_queue_received_message(sd_netlink *nl, sd_netlink_message *m) { + uint32_t serial; + int r; + + assert(nl); + assert(m); + + if (ordered_set_size(nl->rqueue) >= NETLINK_RQUEUE_MAX) + return log_debug_errno(SYNTHETIC_ERRNO(ENOBUFS), + "sd-netlink: exhausted the read queue size (%d)", NETLINK_RQUEUE_MAX); + + r = ordered_set_ensure_put(&nl->rqueue, &netlink_message_hash_ops, m); + if (r < 0) + return r; + + sd_netlink_message_ref(m); + + if (sd_netlink_message_is_broadcast(m)) + return 0; + + serial = message_get_serial(m); + if (serial == 0) + return 0; + + if (sd_netlink_message_get_errno(m) < 0) { + _cleanup_(sd_netlink_message_unrefp) sd_netlink_message *old = NULL; + + old = hashmap_remove(nl->rqueue_by_serial, UINT32_TO_PTR(serial)); + if (old) + log_debug("sd-netlink: received error message with serial %"PRIu32", but another message with " + "the same serial is already stored in the read queue, replacing.", serial); + } + + r = hashmap_ensure_put(&nl->rqueue_by_serial, &netlink_message_hash_ops, UINT32_TO_PTR(serial), m); + if (r == -EEXIST) { + if (!sd_netlink_message_is_error(m)) + log_debug("sd-netlink: received message with serial %"PRIu32", but another message with " + "the same serial is already stored in the read queue, ignoring.", serial); + return 0; + } + if (r < 0) { + sd_netlink_message_unref(ordered_set_remove(nl->rqueue, m)); + return r; + } + + sd_netlink_message_ref(m); + return 0; +} + +static int netlink_queue_partially_received_message(sd_netlink *nl, sd_netlink_message *m) { + uint32_t serial; + int r; + + assert(nl); + assert(m); + assert(m->hdr->nlmsg_flags & NLM_F_MULTI); + + if (hashmap_size(nl->rqueue_partial_by_serial) >= NETLINK_RQUEUE_MAX) + return log_debug_errno(SYNTHETIC_ERRNO(ENOBUFS), + "sd-netlink: exhausted the partial read queue size (%d)", NETLINK_RQUEUE_MAX); + + serial = message_get_serial(m); + r = hashmap_ensure_put(&nl->rqueue_partial_by_serial, &netlink_message_hash_ops, UINT32_TO_PTR(serial), m); + if (r < 0) + return r; + + sd_netlink_message_ref(m); + return 0; +} + +static int parse_message_one(sd_netlink *nl, uint32_t group, const struct nlmsghdr *hdr, sd_netlink_message **ret) { + _cleanup_(sd_netlink_message_unrefp) sd_netlink_message *m = NULL; + size_t size; + int r; + + assert(nl); + assert(hdr); + assert(ret); + + /* not broadcast and not for us */ + if (group == 0 && hdr->nlmsg_pid != nl->sockaddr.nl.nl_pid) + goto finalize; + + /* silently drop noop messages */ + if (hdr->nlmsg_type == NLMSG_NOOP) + goto finalize; + + /* check that we support this message type */ + r = netlink_get_policy_set_and_header_size(nl, hdr->nlmsg_type, NULL, &size); + if (r == -EOPNOTSUPP) { + log_debug("sd-netlink: ignored message with unknown type: %i", hdr->nlmsg_type); + goto finalize; + } + if (r < 0) + return r; + + /* check that the size matches the message type */ + if (hdr->nlmsg_len < NLMSG_LENGTH(size)) { + log_debug("sd-netlink: message is shorter than expected, dropping."); + goto finalize; + } + + r = message_new_empty(nl, &m); + if (r < 0) + return r; + + m->multicast_group = group; + m->hdr = memdup(hdr, hdr->nlmsg_len); + if (!m->hdr) + return -ENOMEM; + + /* seal and parse the top-level message */ + r = sd_netlink_message_rewind(m, nl); + if (r < 0) + return r; + + *ret = TAKE_PTR(m); + return 1; + +finalize: + *ret = NULL; + return 0; +} + /* On success, the number of bytes received is returned and *ret points to the received message * which has a valid header and the correct size. * If nothing useful was received 0 is returned. * On failure, a negative error code is returned. */ int socket_read_message(sd_netlink *nl) { - _cleanup_(sd_netlink_message_unrefp) sd_netlink_message *first = NULL; - bool multi_part = false, done = false; - size_t len, allocated; - struct iovec iov = {}; - uint32_t group = 0; - unsigned i = 0; + bool done = false; + uint32_t group; + size_t len; int r; assert(nl); - assert(nl->rbuffer); /* read nothing, just get the pending message size */ - r = socket_recv_message(nl->fd, &iov, NULL, true); + r = socket_recv_message(nl->fd, NULL, 0, NULL, true); if (r <= 0) return r; - else - len = (size_t) r; + len = (size_t) r; /* make room for the pending message */ if (!greedy_realloc((void**) &nl->rbuffer, len, sizeof(uint8_t))) return -ENOMEM; - allocated = MALLOC_SIZEOF_SAFE(nl->rbuffer); - iov = IOVEC_MAKE(nl->rbuffer, allocated); - /* read the pending message */ - r = socket_recv_message(nl->fd, &iov, &group, false); + r = socket_recv_message(nl->fd, nl->rbuffer, MALLOC_SIZEOF_SAFE(nl->rbuffer), &group, false); if (r <= 0) return r; - else - len = (size_t) r; + len = (size_t) r; - if (len > allocated) - /* message did not fit in read buffer */ - return -EIO; - - if (NLMSG_OK(nl->rbuffer, len) && nl->rbuffer->nlmsg_flags & NLM_F_MULTI) { - multi_part = true; - - for (i = 0; i < nl->rqueue_partial_size; i++) - if (message_get_serial(nl->rqueue_partial[i]) == - nl->rbuffer->nlmsg_seq) { - first = nl->rqueue_partial[i]; - break; - } - } - - for (struct nlmsghdr *new_msg = nl->rbuffer; NLMSG_OK(new_msg, len) && !done; new_msg = NLMSG_NEXT(new_msg, len)) { - _cleanup_(sd_netlink_message_unrefp) sd_netlink_message *m = NULL; - size_t size; - - if (group == 0 && new_msg->nlmsg_pid != nl->sockaddr.nl.nl_pid) - /* not broadcast and not for us */ - continue; - - if (new_msg->nlmsg_type == NLMSG_NOOP) - /* silently drop noop messages */ - continue; - - if (new_msg->nlmsg_type == NLMSG_DONE) { - /* finished reading multi-part message */ - done = true; - - /* if first is not defined, put NLMSG_DONE into the receive queue. */ - if (first) - continue; - } - - /* check that we support this message type */ - r = netlink_get_policy_set_and_header_size(nl, new_msg->nlmsg_type, NULL, &size); - if (r < 0) { - if (r == -EOPNOTSUPP) - log_debug("sd-netlink: ignored message with unknown type: %i", - new_msg->nlmsg_type); - - continue; - } - - /* check that the size matches the message type */ - if (new_msg->nlmsg_len < NLMSG_LENGTH(size)) { - log_debug("sd-netlink: message is shorter than expected, dropping"); - continue; - } - - r = message_new_empty(nl, &m); - if (r < 0) - return r; - - m->multicast_group = group; - m->hdr = memdup(new_msg, new_msg->nlmsg_len); - if (!m->hdr) - return -ENOMEM; - - /* seal and parse the top-level message */ - r = sd_netlink_message_rewind(m, nl); - if (r < 0) - return r; - - /* push the message onto the multi-part message stack */ - if (first) - m->next = first; - first = TAKE_PTR(m); - } - - if (len > 0) - log_debug("sd-netlink: discarding %zu bytes of incoming message", len); - - if (!first) + if (!NLMSG_OK(nl->rbuffer, len)) { + log_debug("sd-netlink: received invalid message, discarding %zu bytes of incoming message", len); return 0; + } - if (!multi_part || done) { - /* we got a complete message, push it on the read queue */ - r = netlink_rqueue_make_room(nl); + for (struct nlmsghdr *hdr = nl->rbuffer; NLMSG_OK(hdr, len); hdr = NLMSG_NEXT(hdr, len)) { + _cleanup_(sd_netlink_message_unrefp) sd_netlink_message *m = NULL; + + r = parse_message_one(nl, group, hdr, &m); if (r < 0) return r; + if (r == 0) + continue; - nl->rqueue[nl->rqueue_size++] = TAKE_PTR(first); + if (hdr->nlmsg_flags & NLM_F_MULTI) { + if (hdr->nlmsg_type == NLMSG_DONE) { + _cleanup_(sd_netlink_message_unrefp) sd_netlink_message *existing = NULL; - if (multi_part && (i < nl->rqueue_partial_size)) { - /* remove the message form the partial read queue */ - memmove(nl->rqueue_partial + i, nl->rqueue_partial + i + 1, - sizeof(sd_netlink_message*) * (nl->rqueue_partial_size - i - 1)); - nl->rqueue_partial_size--; - } + /* finished reading multi-part message */ + existing = hashmap_remove(nl->rqueue_partial_by_serial, UINT32_TO_PTR(hdr->nlmsg_seq)); - return 1; - } else { - /* we only got a partial multi-part message, push it on the - partial read queue */ - if (i < nl->rqueue_partial_size) - nl->rqueue_partial[i] = TAKE_PTR(first); - else { - r = netlink_rqueue_partial_make_room(nl); + /* if we receive only NLMSG_DONE, put it into the receive queue. */ + r = netlink_queue_received_message(nl, existing ?: m); + if (r < 0) + return r; + + done = true; + } else { + sd_netlink_message *existing; + + existing = hashmap_get(nl->rqueue_partial_by_serial, UINT32_TO_PTR(hdr->nlmsg_seq)); + if (existing) { + /* This is the continuation of the previously read messages. + * Let's append this message at the end. */ + while (existing->next) + existing = existing->next; + existing->next = TAKE_PTR(m); + } else { + /* This is the first message. Put it into the queue for partially + * received messages. */ + r = netlink_queue_partially_received_message(nl, m); + if (r < 0) + return r; + } + } + + } else { + r = netlink_queue_received_message(nl, m); if (r < 0) return r; - nl->rqueue_partial[nl->rqueue_partial_size++] = TAKE_PTR(first); + done = true; } - - return 0; } + + if (len > 0) + log_debug("sd-netlink: discarding trailing %zu bytes of incoming message", len); + + return done; } diff --git a/src/libsystemd/sd-netlink/netlink-util.c b/src/libsystemd/sd-netlink/netlink-util.c index 12cdc99ff2..c6091542d2 100644 --- a/src/libsystemd/sd-netlink/netlink-util.c +++ b/src/libsystemd/sd-netlink/netlink-util.c @@ -673,6 +673,15 @@ int netlink_open_family(sd_netlink **ret, int family) { return 0; } +static bool serial_used(sd_netlink *nl, uint32_t serial) { + assert(nl); + + return + hashmap_contains(nl->reply_callbacks, UINT32_TO_PTR(serial)) || + hashmap_contains(nl->rqueue_by_serial, UINT32_TO_PTR(serial)) || + hashmap_contains(nl->rqueue_partial_by_serial, UINT32_TO_PTR(serial)); +} + void netlink_seal_message(sd_netlink *nl, sd_netlink_message *m) { uint32_t picked; @@ -689,7 +698,7 @@ void netlink_seal_message(sd_netlink *nl, sd_netlink_message *m) { such messages */ nl->serial = nl->serial == UINT32_MAX ? 1 : nl->serial + 1; - } while (hashmap_contains(nl->reply_callbacks, UINT32_TO_PTR(picked))); + } while (serial_used(nl, picked)); m->hdr->nlmsg_seq = picked; message_seal(m); diff --git a/src/libsystemd/sd-netlink/sd-netlink.c b/src/libsystemd/sd-netlink/sd-netlink.c index b99abae640..6eb7f659ae 100644 --- a/src/libsystemd/sd-netlink/sd-netlink.c +++ b/src/libsystemd/sd-netlink/sd-netlink.c @@ -61,10 +61,6 @@ static int netlink_new(sd_netlink **ret) { .serial = (uint32_t) (now(CLOCK_MONOTONIC) % UINT32_MAX) + 1, }; - /* We guarantee that the read buffer has at least space for a message header */ - if (!greedy_realloc((void**) &nl->rbuffer, sizeof(struct nlmsghdr), sizeof(uint8_t))) - return -ENOMEM; - *ret = TAKE_PTR(nl); return 0; } @@ -120,18 +116,12 @@ int sd_netlink_increase_rxbuf(sd_netlink *nl, size_t size) { static sd_netlink *netlink_free(sd_netlink *nl) { sd_netlink_slot *s; - unsigned i; assert(nl); - for (i = 0; i < nl->rqueue_size; i++) - sd_netlink_message_unref(nl->rqueue[i]); - free(nl->rqueue); - - for (i = 0; i < nl->rqueue_partial_size; i++) - sd_netlink_message_unref(nl->rqueue_partial[i]); - free(nl->rqueue_partial); - + ordered_set_free(nl->rqueue); + hashmap_free(nl->rqueue_by_serial); + hashmap_free(nl->rqueue_partial_by_serial); free(nl->rbuffer); while ((s = nl->slots)) { @@ -179,57 +169,27 @@ int sd_netlink_send( return 1; } -int netlink_rqueue_make_room(sd_netlink *nl) { - assert(nl); - - if (nl->rqueue_size >= NETLINK_RQUEUE_MAX) - return log_debug_errno(SYNTHETIC_ERRNO(ENOBUFS), - "sd-netlink: exhausted the read queue size (%d)", - NETLINK_RQUEUE_MAX); - - if (!GREEDY_REALLOC(nl->rqueue, nl->rqueue_size + 1)) - return -ENOMEM; - - return 0; -} - -int netlink_rqueue_partial_make_room(sd_netlink *nl) { - assert(nl); - - if (nl->rqueue_partial_size >= NETLINK_RQUEUE_MAX) - return log_debug_errno(SYNTHETIC_ERRNO(ENOBUFS), - "sd-netlink: exhausted the partial read queue size (%d)", - NETLINK_RQUEUE_MAX); - - if (!GREEDY_REALLOC(nl->rqueue_partial, nl->rqueue_partial_size + 1)) - return -ENOMEM; - - return 0; -} - -static int dispatch_rqueue(sd_netlink *nl, sd_netlink_message **message) { +static int dispatch_rqueue(sd_netlink *nl, sd_netlink_message **ret) { + sd_netlink_message *m; int r; assert(nl); - assert(message); + assert(ret); - if (nl->rqueue_size <= 0) { + if (ordered_set_size(nl->rqueue) <= 0) { /* Try to read a new message */ r = socket_read_message(nl); - if (r == -ENOBUFS) { /* FIXME: ignore buffer overruns for now */ + if (r == -ENOBUFS) /* FIXME: ignore buffer overruns for now */ log_debug_errno(r, "sd-netlink: Got ENOBUFS from netlink socket, ignoring."); - return 1; - } - if (r <= 0) + else if (r < 0) return r; } /* Dispatch a queued message */ - *message = nl->rqueue[0]; - nl->rqueue_size--; - memmove(nl->rqueue, nl->rqueue + 1, sizeof(sd_netlink_message*) * nl->rqueue_size); - - return 1; + m = ordered_set_steal_first(nl->rqueue); + sd_netlink_message_unref(hashmap_remove_value(nl->rqueue_by_serial, UINT32_TO_PTR(message_get_serial(m)), m)); + *ret = m; + return !!m; } static int process_timeout(sd_netlink *nl) { @@ -469,7 +429,7 @@ int sd_netlink_wait(sd_netlink *nl, uint64_t timeout_usec) { assert_return(nl, -EINVAL); assert_return(!netlink_pid_changed(nl), -ECHILD); - if (nl->rqueue_size > 0) + if (ordered_set_size(nl->rqueue) > 0) return 0; r = netlink_poll(nl, false, timeout_usec); @@ -570,39 +530,32 @@ int sd_netlink_read( timeout = calc_elapse(usec); for (;;) { + _cleanup_(sd_netlink_message_unrefp) sd_netlink_message *m = NULL; usec_t left; - for (unsigned i = 0; i < nl->rqueue_size; i++) { - _cleanup_(sd_netlink_message_unrefp) sd_netlink_message *incoming = NULL; - uint32_t received_serial; + m = hashmap_remove(nl->rqueue_by_serial, UINT32_TO_PTR(serial)); + if (m) { uint16_t type; - received_serial = message_get_serial(nl->rqueue[i]); - if (received_serial != serial) - continue; - - incoming = nl->rqueue[i]; - /* found a match, remove from rqueue and return it */ - memmove(nl->rqueue + i, nl->rqueue + i + 1, - sizeof(sd_netlink_message*) * (nl->rqueue_size - i - 1)); - nl->rqueue_size--; + sd_netlink_message_unref(ordered_set_remove(nl->rqueue, m)); - r = sd_netlink_message_get_errno(incoming); + r = sd_netlink_message_get_errno(m); if (r < 0) return r; - r = sd_netlink_message_get_type(incoming, &type); + r = sd_netlink_message_get_type(m, &type); if (r < 0) return r; if (type == NLMSG_DONE) { - *ret = NULL; + if (ret) + *ret = NULL; return 0; } if (ret) - *ret = TAKE_PTR(incoming); + *ret = TAKE_PTR(m); return 1; } @@ -656,7 +609,7 @@ int sd_netlink_get_events(sd_netlink *nl) { assert_return(nl, -EINVAL); assert_return(!netlink_pid_changed(nl), -ECHILD); - return nl->rqueue_size == 0 ? POLLIN : 0; + return ordered_set_size(nl->rqueue) == 0 ? POLLIN : 0; } int sd_netlink_get_timeout(sd_netlink *nl, uint64_t *timeout_usec) { @@ -666,7 +619,7 @@ int sd_netlink_get_timeout(sd_netlink *nl, uint64_t *timeout_usec) { assert_return(timeout_usec, -EINVAL); assert_return(!netlink_pid_changed(nl), -ECHILD); - if (nl->rqueue_size > 0) { + if (ordered_set_size(nl->rqueue) > 0) { *timeout_usec = 0; return 1; } @@ -678,7 +631,6 @@ int sd_netlink_get_timeout(sd_netlink *nl, uint64_t *timeout_usec) { } *timeout_usec = c->timeout; - return 1; }