Re: [PATCH 4/4] dbus: Handle partial reads and writes

[Denis Kenzior <denkenz@gmail.com>]

[-- Attachment #1: Type: text/plain, Size: 7627 bytes --]

Hi Andrew,

On 05/04/2016 06:03 AM, Andrew Zaborowski wrote:
> Loop until we have received or sent all of the data requested.  We pass
> MSG_WAITALL to potentially reduce the number of iterations.  Make sure
> we free body and header in classic_recv_message when
> dbus_message_build() returns an error.
>
> Calling sendms and recvmsg multiple times makes it even more tricky to
> understand what will happen with the FDs passed as oob data when a
> partial read/write happens or when there are multiple messsages with FDs
> attached in the queue.  What this patch does should be equivalent to
> what systemd dbus code does.  It seems to assume that the FDs are
> attached to the last byte of the message so on a partial write, the FDs
> are not written and the full array can be passed to sendmsg() again.
> I'm not sure if this is documented anywhere but I've seen an explicit
> statement of this at ...

I think this is a bad assumption to make.  Browsing af_unix.c makes me 
think this is not the case.

> ---
>   ell/dbus.c | 179 +++++++++++++++++++++++++++++++++++++------------------------
>   1 file changed, 110 insertions(+), 69 deletions(-)
>
> diff --git a/ell/dbus.c b/ell/dbus.c
> index 35378b2..3ca7b87 100644
> --- a/ell/dbus.c
> +++ b/ell/dbus.c
> @@ -579,63 +579,126 @@ static void classic_free(struct l_dbus *dbus)
>   	l_free(classic);
>   }
>
> +static bool msg_op(int fd, bool send, struct iovec iov[],
> +			int **fds, uint32_t *num_fds)
> +{

Please don't do this, the code is so much easier to understand when the 
rx and tx paths are separate.

> +	int r;
> +	struct msghdr msg;
> +	struct cmsghdr *cmsg;
> +	uint8_t *fd_buf = NULL;
> +	int buf_fds;
> +	int iovlen = 2;
> +
> +	if (send)
> +		buf_fds = *num_fds;
> +	else {
> +		buf_fds = 16;
> +		*num_fds = 0;
> +	}
> +
> +	if (buf_fds) {
> +		fd_buf = alloca(CMSG_SPACE(buf_fds * sizeof(int)));
> +
> +		if (send) {
> +			msg.msg_control = fd_buf;
> +			msg.msg_controllen = CMSG_LEN(buf_fds * sizeof(int));
> +
> +			/* Set up fd_buf contents */
> +			cmsg = CMSG_FIRSTHDR(&msg);
> +			cmsg->cmsg_len = msg.msg_controllen;
> +			cmsg->cmsg_level = SOL_SOCKET;
> +			cmsg->cmsg_type = SCM_RIGHTS;
> +			memcpy(CMSG_DATA(cmsg), *fds, buf_fds * sizeof(int));
> +		}
> +	}
> +
> +	while (iovlen) {
> +		memset(&msg, 0, sizeof(msg));
> +		msg.msg_iov = iov;
> +		msg.msg_iovlen = iovlen;
> +		msg.msg_control = fd_buf;
> +		msg.msg_controllen = fd_buf ?
> +			CMSG_LEN(buf_fds * sizeof(int)) : 0;
> +
> +		if (send)
> +			r = sendmsg(fd, &msg, 0);
> +		else
> +			r = recvmsg(fd, &msg, MSG_CMSG_CLOEXEC | MSG_WAITALL);
> +
> +		if (r <= 0) {

Strictly speaking, you can't call sendmsg in a loop.  The socket is 
marked non-blocking, so EWOULDBLOCK or EAGAIN is a valid error return. 
This means that you need to exit the loop and re-enter once POLLOUT is 
received.

> +			if (r != -1 || errno != EINTR)
> +				return false;
> +
> +			continue;
> +		}
> +
> +		while (iovlen && r >= (int) iov[0].iov_len) {
> +			r -= iov[0].iov_len;
> +			iov++;
> +			iovlen--;
> +		}
> +
> +		if (iovlen) {
> +			iov[0].iov_base += r;
> +			iov[0].iov_len -= r;
> +		}
> +
> +		if (send)
> +			continue;
> +
> +		for (cmsg = CMSG_FIRSTHDR(&msg); cmsg;
> +				cmsg = CMSG_NXTHDR(&msg, cmsg)) {
> +			int cnt;
> +
> +			if (cmsg->cmsg_level != SOL_SOCKET ||
> +						cmsg->cmsg_type != SCM_RIGHTS)
> +				continue;
> +
> +			cnt = (cmsg->cmsg_len - CMSG_LEN(0)) / sizeof(int);
> +
> +			*fds = l_realloc(*fds, (*num_fds + cnt) * sizeof(int));
> +
> +			memcpy(*fds + *num_fds, CMSG_DATA(cmsg),
> +					cnt * sizeof(int));
> +
> +			*num_fds += cnt;
> +                }
> +	}
> +
> +	return true;
> +}
> +
>   static bool classic_send_message(struct l_dbus *dbus,
>   					struct l_dbus_message *message)
>   {
>   	int fd = l_io_get_fd(dbus->io);
> -	struct msghdr msg;
>   	struct iovec iov[2];
> -	ssize_t len;
>   	int *fds;
>   	uint32_t num_fds = 0;
> -	struct cmsghdr *cmsg;
>
>   	iov[0].iov_base = _dbus_message_get_header(message, &iov[0].iov_len);
>   	iov[1].iov_base = _dbus_message_get_body(message, &iov[1].iov_len);
>
> -	memset(&msg, 0, sizeof(msg));
> -	msg.msg_iov = iov;
> -	msg.msg_iovlen = 2;
> -
>   	if (dbus->support_unix_fd)
>   		fds = _dbus_message_get_fds(message, &num_fds);
>
> -	if (num_fds) {
> -		msg.msg_control = alloca(CMSG_SPACE(num_fds * sizeof(int)));
> -		msg.msg_controllen = CMSG_LEN(num_fds * sizeof(int));
> -
> -		cmsg = CMSG_FIRSTHDR(&msg);
> -		cmsg->cmsg_len = msg.msg_controllen;
> -		cmsg->cmsg_level = SOL_SOCKET;
> -		cmsg->cmsg_type = SCM_RIGHTS;
> -		memcpy(CMSG_DATA(cmsg), fds, num_fds * sizeof(int));
> -	}
> -
> -	len = sendmsg(fd, &msg, 0);
> -	if (len < 0)
> -		return false;
> -
> -	return true;
> +	return msg_op(fd, true, iov, &fds, &num_fds);
>   }
>
>   static struct l_dbus_message *classic_recv_message(struct l_dbus *dbus)
>   {
>   	int fd = l_io_get_fd(dbus->io);
>   	struct dbus_header hdr;
> -	struct msghdr msg;
>   	struct iovec iov[2];
> -	struct cmsghdr *cmsg;
>   	ssize_t len;
>   	void *header, *body;
>   	size_t header_size, body_size;
> -	union {
> -		uint8_t bytes[CMSG_SPACE(16 * sizeof(int))];
> -		struct cmsghdr align;
> -	} fd_buf;
>   	int *fds = NULL;
>   	uint32_t num_fds = 0;
> +	struct l_dbus_message *msg;
> +	unsigned int i;
>
> -	len = recv(fd, &hdr, DBUS_HEADER_SIZE, MSG_PEEK);
> +	len = recv(fd, &hdr, DBUS_HEADER_SIZE, MSG_PEEK | MSG_WAITALL);
>   	if (len != DBUS_HEADER_SIZE)
>   		return NULL;
>
> @@ -650,59 +713,37 @@ static struct l_dbus_message *classic_recv_message(struct l_dbus *dbus)
>   	iov[1].iov_base = body;
>   	iov[1].iov_len  = body_size;
>
> -	memset(&msg, 0, sizeof(msg));
> -	msg.msg_iov = iov;
> -	msg.msg_iovlen = 2;
> -	msg.msg_control = &fd_buf;
> -	msg.msg_controllen = sizeof(fd_buf);
> -
> -	len = recvmsg(fd, &msg, MSG_CMSG_CLOEXEC);
> -	if (len < 0)
> -		goto cmsg_fail;
> -
> -	for (cmsg = CMSG_FIRSTHDR(&msg); cmsg;
> -				cmsg = CMSG_NXTHDR(&msg, cmsg)) {
> -		unsigned int i;
> -
> -		if (cmsg->cmsg_level != SOL_SOCKET ||
> -					cmsg->cmsg_type != SCM_RIGHTS)
> -			continue;
> -
> -		num_fds = (cmsg->cmsg_len - CMSG_LEN(0)) / sizeof(int);
> -		fds = (void *) CMSG_DATA(cmsg);
> -
> -		/* Set FD_CLOEXEC on all file descriptors */
> -		for (i = 0; i < num_fds; i++) {
> -			long flags;
> -
> -			flags = fcntl(fds[i], F_GETFD, NULL);
> -			if (flags < 0)
> -				continue;
> -
> -			if (!(flags & FD_CLOEXEC))
> -				fcntl(fds[i], F_SETFD, flags | FD_CLOEXEC);
> -                }
> -	}
> +	if (!msg_op(fd, false, iov, &fds, &num_fds))
> +		goto fail;
>
>   	if (hdr.endian != DBUS_NATIVE_ENDIAN) {
>   		l_util_debug(dbus->debug_handler,
>   				dbus->debug_data, "Endianness incorrect");
> -		goto bad_msg;
> +		goto fail;
>   	}
>
>   	if (hdr.version != 1) {
>   		l_util_debug(dbus->debug_handler,
>   				dbus->debug_data, "Protocol version incorrect");
> -		goto bad_msg;
> +		goto fail;
>   	}
>
> -	return dbus_message_build(header, header_size, body, body_size,
> +	msg = dbus_message_build(header, header_size, body, body_size,
>   					fds, num_fds);
> +	if (!msg)
> +		goto fail;
>
> -bad_msg:
> -cmsg_fail:
> -	l_free(header);
> +	l_free(fds);
> +
> +	return msg;
> +
> +fail:
> +	for (i = 0; i < num_fds; i++)
> +		close(fds[i]);
> +
> +	l_free(fds);
>   	l_free(body);
> +	l_free(header);
>
>   	return NULL;
>   }
>

Regards,
-Denis