Convert devd's client socket to type SOCK_SEQPACKET.

This change consists of two merges from projects/zfsd/head along with the
addition of an ATF test case for the new functionality.

sbin/devd/tests/Makefile
sbin/devd/tests/client_test.c
	Add ATF test cases for reading events from both devd socket types.

r266519:
sbin/devd/devd.8
sbin/devd/devd.cc
	Create a new socket, of type SOCK_SEQPACKET, for communicating with
	clients. SOCK_SEQPACKET sockets preserve record boundaries,
	simplying code in the client. The old SOCK_STREAM socket is retained
	for backwards-compatibility with existing clients.

r269993:
sbin/devd/devd.8
	Fix grammar bug.

CR:		https://reviews.freebsd.org/rS266519
MFC after:	5 days
Sponsored by:	Spectra Logic

5 files changed, 276 insertions, 38 deletions

diff --git a/sbin/devd/Makefile b/sbin/devd/Makefile
index c53f094..518e5e2 100644
--- a/sbin/devd/Makefile
+++ b/sbin/devd/Makefile
@@ -1,5 +1,7 @@
 # $FreeBSD$
 
+.include <src.opts.mk>
+
 PROG_CXX=devd
 SRCS=	devd.cc token.l parse.y y.tab.h
 MAN=	devd.8 devd.conf.5
@@ -16,4 +18,8 @@ CFLAGS+=-I. -I${.CURDIR}
 
 CLEANFILES= y.output
 
+.if ${MK_TESTS} != "no"
+SUBDIR+=    tests
+.endif
+
 .include <bsd.prog.mk>
diff --git a/sbin/devd/devd.8 b/sbin/devd/devd.8
index fa34df2..12a92d9 100644
--- a/sbin/devd/devd.8
+++ b/sbin/devd/devd.8
@@ -25,7 +25,7 @@
 .\"
 .\" $FreeBSD$
 .\"
-.Dd January 30, 2013
+.Dd August 14, 2014
 .Dt DEVD 8
 .Os
 .Sh NAME
@@ -55,9 +55,7 @@ If option
 .Fl f
 is specified more than once, the last file specified is used.
 .It Fl l Ar num
-Limit concurrent
-.Pa /var/run/devd.pipe
-connections to
+Limit concurrent socket connections to
 .Ar num .
 The default connection limit is 10.
 .It Fl n
@@ -130,22 +128,27 @@ wish to hook into the
 system without modifying the user's other
 config files.
 .Pp
-All messages that
+Since
+.Xr devctl 4
+allows only one active reader,
 .Nm
-receives are forwarded to the
+multiplexes it, forwarding all events to any number of connected clients.
+Clients connect by opening the SOCK_SEQPACKET
 .Ux
 domain socket at
-.Pa /var/run/devd.pipe .
+.Pa /var/run/devd.seqpacket.pipe .
 .Sh FILES
-.Bl -tag -width ".Pa /var/run/devd.pipe" -compact
+.Bl -tag -width ".Pa /var/run/devd.seqpacket.pipe" -compact
 .It Pa /etc/devd.conf
 The default
 .Nm
 configuration file.
-.It Pa /var/run/devd.pipe
+.It Pa /var/run/devd.seqpacket.pipe
 The socket used by
 .Nm
 to communicate with its clients.
+.It Pa /var/run/devd.pipe
+A deprecated socket retained for use with old clients.
 .El
 .Sh SEE ALSO
 .Xr devctl 4 ,
diff --git a/sbin/devd/devd.cc b/sbin/devd/devd.cc
index ce2a4f3..c770204 100644
--- a/sbin/devd/devd.cc
+++ b/sbin/devd/devd.cc
@@ -100,7 +100,8 @@ __FBSDID("$FreeBSD$");
 #include "devd.h"		/* C compatible definitions */
 #include "devd.hh"		/* C++ class definitions */
 
-#define PIPE "/var/run/devd.pipe"
+#define STREAMPIPE "/var/run/devd.pipe"
+#define SEQPACKETPIPE "/var/run/devd.seqpacket.pipe"
 #define CF "/etc/devd.conf"
 #define SYSCTL "hw.bus.devctl_queue"
 
@@ -119,6 +120,11 @@ __FBSDID("$FreeBSD$");
 
 using namespace std;
 
+typedef struct client {
+	int fd;
+	int socktype;
+} client_t;
+
 extern FILE *yyin;
 extern int lineno;
 
@@ -822,12 +828,12 @@ process_event(char *buffer)
 }
 
 int
-create_socket(const char *name)
+create_socket(const char *name, int socktype)
 {
 	int fd, slen;
 	struct sockaddr_un sun;
 
-	if ((fd = socket(PF_LOCAL, SOCK_STREAM, 0)) < 0)
+	if ((fd = socket(PF_LOCAL, socktype, 0)) < 0)
 		err(1, "socket");
 	bzero(&sun, sizeof(sun));
 	sun.sun_family = AF_UNIX;
@@ -846,12 +852,13 @@ create_socket(const char *name)
 
 unsigned int max_clients = 10;	/* Default, can be overriden on cmdline. */
 unsigned int num_clients;
-list<int> clients;
+
+list<client_t> clients;
 
 void
 notify_clients(const char *data, int len)
 {
-	list<int>::iterator i;
+	list<client_t>::iterator i;
 
 	/*
 	 * Deliver the data to all clients.  Throw clients overboard at the
@@ -861,11 +868,17 @@ notify_clients(const char *data, int len)
 	 * kernel memory or tie up the limited number of available connections).
 	 */
 	for (i = clients.begin(); i != clients.end(); ) {
-		if (write(*i, data, len) != len) {
+		int flags;
+		if (i->socktype == SOCK_SEQPACKET)
+			flags = MSG_EOR;
+		else
+			flags = 0;
+
+		if (send(i->fd, data, len, flags) != len) {
 			--num_clients;
-			close(*i);
+			close(i->fd);
 			i = clients.erase(i);
-			devdlog(LOG_WARNING, "notify_clients: write() failed; "
+			devdlog(LOG_WARNING, "notify_clients: send() failed; "
 			    "dropping unresponsive client\n");
 		} else
 			++i;
@@ -877,7 +890,7 @@ check_clients(void)
 {
 	int s;
 	struct pollfd pfd;
-	list<int>::iterator i;
+	list<client_t>::iterator i;
 
 	/*
 	 * Check all existing clients to see if any of them have disappeared.
@@ -888,12 +901,12 @@ check_clients(void)
 	 */
 	pfd.events = 0;
 	for (i = clients.begin(); i != clients.end(); ) {
-		pfd.fd = *i;
+		pfd.fd = i->fd;
 		s = poll(&pfd, 1, 0);
 		if ((s < 0 && s != EINTR ) ||
 		    (s > 0 && (pfd.revents & POLLHUP))) {
 			--num_clients;
-			close(*i);
+			close(i->fd);
 			i = clients.erase(i);
 			devdlog(LOG_NOTICE, "check_clients:  "
 			    "dropping disconnected client\n");
@@ -903,9 +916,9 @@ check_clients(void)
 }
 
 void
-new_client(int fd)
+new_client(int fd, int socktype)
 {
-	int s;
+	client_t s;
 	int sndbuf_size;
 
 	/*
@@ -914,13 +927,14 @@ new_client(int fd)
 	 * by sending large buffers full of data we'll never read.
 	 */
 	check_clients();
-	s = accept(fd, NULL, NULL);
-	if (s != -1) {
+	s.socktype = socktype;
+	s.fd = accept(fd, NULL, NULL);
+	if (s.fd != -1) {
 		sndbuf_size = CLIENT_BUFSIZE;
-		if (setsockopt(s, SOL_SOCKET, SO_SNDBUF, &sndbuf_size,
+		if (setsockopt(s.fd, SOL_SOCKET, SO_SNDBUF, &sndbuf_size,
 		    sizeof(sndbuf_size)))
 			err(1, "setsockopt");
-		shutdown(s, SHUT_RD);
+		shutdown(s.fd, SHUT_RD);
 		clients.push_back(s);
 		++num_clients;
 	} else
@@ -934,7 +948,7 @@ event_loop(void)
 	int fd;
 	char buffer[DEVCTL_MAXBUF];
 	int once = 0;
-	int server_fd, max_fd;
+	int stream_fd, seqpacket_fd, max_fd;
 	int accepting;
 	timeval tv;
 	fd_set fds;
@@ -942,9 +956,10 @@ event_loop(void)
 	fd = open(PATH_DEVCTL, O_RDONLY | O_CLOEXEC);
 	if (fd == -1)
 		err(1, "Can't open devctl device %s", PATH_DEVCTL);
-	server_fd = create_socket(PIPE);
+	stream_fd = create_socket(STREAMPIPE, SOCK_STREAM);
+	seqpacket_fd = create_socket(SEQPACKETPIPE, SOCK_SEQPACKET);
 	accepting = 1;
-	max_fd = max(fd, server_fd) + 1;
+	max_fd = max(fd, max(stream_fd, seqpacket_fd)) + 1;
 	while (!romeo_must_die) {
 		if (!once && !no_daemon && !daemonize_quick) {
 			// Check to see if we have any events pending.
@@ -965,24 +980,28 @@ event_loop(void)
 		}
 		/*
 		 * When we've already got the max number of clients, stop
-		 * accepting new connections (don't put server_fd in the set),
-		 * shrink the accept() queue to reject connections quickly, and
-		 * poll the existing clients more often, so that we notice more
-		 * quickly when any of them disappear to free up client slots.
+		 * accepting new connections (don't put the listening sockets in
+		 * the set), shrink the accept() queue to reject connections
+		 * quickly, and poll the existing clients more often, so that we
+		 * notice more quickly when any of them disappear to free up
+		 * client slots.
 		 */
 		FD_ZERO(&fds);
 		FD_SET(fd, &fds);
 		if (num_clients < max_clients) {
 			if (!accepting) {
-				listen(server_fd, max_clients);
+				listen(stream_fd, max_clients);
+				listen(seqpacket_fd, max_clients);
 				accepting = 1;
 			}
-			FD_SET(server_fd, &fds);
+			FD_SET(stream_fd, &fds);
+			FD_SET(seqpacket_fd, &fds);
 			tv.tv_sec = 60;
 			tv.tv_usec = 0;
 		} else {
 			if (accepting) {
-				listen(server_fd, 0);
+				listen(stream_fd, 0);
+				listen(seqpacket_fd, 0);
 				accepting = 0;
 			}
 			tv.tv_sec = 2;
@@ -1022,8 +1041,14 @@ event_loop(void)
 				break;
 			}
 		}
-		if (FD_ISSET(server_fd, &fds))
-			new_client(server_fd);
+		if (FD_ISSET(stream_fd, &fds))
+			new_client(stream_fd, SOCK_STREAM);
+		/*
+		 * Aside from the socket type, both sockets use the same
+		 * protocol, so we can process clients the same way.
+		 */
+		if (FD_ISSET(seqpacket_fd, &fds))
+			new_client(seqpacket_fd, SOCK_SEQPACKET);
 	}
 	close(fd);
 }
diff --git a/sbin/devd/tests/Makefile b/sbin/devd/tests/Makefile
new file mode 100644
index 0000000..ee679ce
--- /dev/null
+++ b/sbin/devd/tests/Makefile
@@ -0,0 +1,12 @@
+# $FreeBSD$
+
+TESTSDIR=	${TESTSBASE}/sbin/devd
+
+ATF_TESTS_C=	client_test
+TEST_METADATA.client_test=	required_programs="devd"
+TEST_METADATA.client_test+=	required_user="root"
+TEST_METADATA.client_test+=	timeout=15
+
+WARNS?=	5
+
+.include <bsd.test.mk>
diff --git a/sbin/devd/tests/client_test.c b/sbin/devd/tests/client_test.c
new file mode 100644
index 0000000..bc0068c
--- /dev/null
+++ b/sbin/devd/tests/client_test.c
@@ -0,0 +1,192 @@
+/*-
+ * Copyright (c) 2014 Spectra Logic Corporation. All rights reserved.
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ * 1. Redistributions of source code must retain the above copyright
+ *    notice, this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright
+ *    notice, this list of conditions and the following disclaimer in the
+ *    documentation and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY AUTHOR AND CONTRIBUTORS ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED.  IN NO EVENT SHALL AUTHOR OR CONTRIBUTORS BE LIABLE
+ * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
+ * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
+ * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+ * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
+ * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+
+#include <sys/cdefs.h>
+__FBSDID("$FreeBSD$");
+
+#include <stdbool.h>
+#include <stdio.h>
+
+#include <sys/param.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+
+#include <atf-c.h>
+/* Helper functions*/
+
+/*
+ * Create two devd events.  The easiest way I know of, that requires no special
+ * hardware, is to create md(4) devices.
+ */
+static void
+create_two_events(void)
+{
+	FILE *create_stdout;
+	FILE *destroy_stdout;
+	char mdname[80];
+	char destroy_cmd[80];
+	char *error;
+
+	create_stdout = popen("mdconfig -a -s 64 -t null", "r");
+	ATF_REQUIRE(create_stdout != NULL);
+	error = fgets(mdname, sizeof(mdname), create_stdout);
+	ATF_REQUIRE(error != NULL);
+	/* We only expect one line of output */
+	ATF_REQUIRE_EQ(0, pclose(create_stdout));
+
+	snprintf(destroy_cmd, nitems(destroy_cmd), "mdconfig -d -u %s", mdname);
+	destroy_stdout = popen(destroy_cmd, "r");
+	/* We expect no output */
+	ATF_REQUIRE_EQ(0, pclose(destroy_stdout));
+}
+
+/*
+ * Test Cases
+ */
+
+/*
+ * Open a client connection to devd, create some events, and test that they can
+ * be read _whole_ and _one_at_a_time_ from the socket
+ */
+ATF_TC_WITHOUT_HEAD(seqpacket);
+ATF_TC_BODY(seqpacket, tc)
+{
+	int s;
+	int error;
+	struct sockaddr_un devd_addr;
+	bool got_create_event = false;
+	bool got_destroy_event = false;
+	const char create_pat[] =
+		"!system=DEVFS subsystem=CDEV type=CREATE cdev=md";
+	const char destroy_pat[] =
+		"!system=DEVFS subsystem=CDEV type=DESTROY cdev=md";
+
+	memset(&devd_addr, 0, sizeof(devd_addr));
+	devd_addr.sun_family = PF_LOCAL;
+	strlcpy(devd_addr.sun_path, "/var/run/devd.seqpacket.pipe",
+			sizeof(devd_addr.sun_path));
+
+	s = socket(PF_LOCAL, SOCK_SEQPACKET, 0);
+	ATF_REQUIRE(s >= 0);
+	error = connect(s, (struct sockaddr*)&devd_addr, SUN_LEN(&devd_addr));
+	ATF_REQUIRE_EQ(0, error);
+
+	create_two_events();
+
+	/*
+	 * Loop until both events are detected on _different_ reads
+	 * There may be extra events due to unrelated system activity
+	 * If we never get both events, then the test will timeout.
+	 */
+	while (!(got_create_event && got_destroy_event)) {
+		int cmp;
+		ssize_t len;
+		char event[1024];
+
+		len = recv(s, event, sizeof(event), MSG_WAITALL);
+		ATF_REQUIRE(len != -1);
+		/* NULL terminate the result */
+		event[len] = '\0';
+		printf("%s", event);
+		cmp = strncmp(event, create_pat, sizeof(create_pat) - 1);
+		if (cmp == 0)
+			got_create_event = true;
+
+		cmp = strncmp(event, destroy_pat, sizeof(destroy_pat) - 1);
+		if (cmp == 0)
+			got_destroy_event = true;
+	}
+}
+
+/*
+ * Open a client connection to devd using the stream socket, create some
+ * events, and test that they can be read in any number of reads.
+ */
+ATF_TC_WITHOUT_HEAD(stream);
+ATF_TC_BODY(stream, tc)
+{
+	int s;
+	int error;
+	struct sockaddr_un devd_addr;
+	bool got_create_event = false;
+	bool got_destroy_event = false;
+	const char create_pat[] =
+		"!system=DEVFS subsystem=CDEV type=CREATE cdev=md";
+	const char destroy_pat[] =
+		"!system=DEVFS subsystem=CDEV type=DESTROY cdev=md";
+	ssize_t len = 0;
+
+	memset(&devd_addr, 0, sizeof(devd_addr));
+	devd_addr.sun_family = PF_LOCAL;
+	strlcpy(devd_addr.sun_path, "/var/run/devd.pipe",
+			sizeof(devd_addr.sun_path));
+
+	s = socket(PF_LOCAL, SOCK_STREAM, 0);
+	ATF_REQUIRE(s >= 0);
+	error = connect(s, (struct sockaddr*)&devd_addr, SUN_LEN(&devd_addr));
+	ATF_REQUIRE_EQ(0, error);
+
+	create_two_events();
+
+	/*
+	 * Loop until both events are detected on _different_ reads
+	 * There may be extra events due to unrelated system activity
+	 * If we never get both events, then the test will timeout.
+	 */
+	while (!(got_create_event && got_destroy_event)) {
+		char event[1024];
+		ssize_t newlen;
+		char *create_pos, *destroy_pos;
+
+		newlen = read(s, &event[len], sizeof(event) - len);
+		ATF_REQUIRE(newlen != -1);
+		len += newlen;
+		/* NULL terminate the result */
+		event[newlen] = '\0';
+		printf("%s", event);
+
+		create_pos = strstr(event, create_pat);
+		if (create_pos != NULL);
+			got_create_event = true;
+
+		destroy_pos = strstr(event, destroy_pat);
+		if (destroy_pos != NULL);
+			got_destroy_event = true;
+
+	}
+}
+
+/*
+ * Main.
+ */
+
+ATF_TP_ADD_TCS(tp)
+{
+	ATF_TP_ADD_TC(tp, seqpacket);
+	ATF_TP_ADD_TC(tp, stream);
+
+	return (atf_no_error());
+}
+