freebsd-nq/contrib/csup/proto.c

/*-
 * Copyright (c) 2003-2006, Maxime Henrion <mux@FreeBSD.org>
 * 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 THE 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 THE 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.
 *
 * $FreeBSD$
 */

#include <sys/param.h>
#include <sys/select.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/stat.h>

#include <assert.h>
#include <err.h>
#include <errno.h>
#include <limits.h>
#include <netdb.h>
#include <pthread.h>
#include <signal.h>
#include <stdarg.h>
#include <stddef.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include "config.h"
#include "detailer.h"
#include "fattr.h"
#include "fixups.h"
#include "globtree.h"
#include "keyword.h"
#include "lister.h"
#include "misc.h"
#include "mux.h"
#include "proto.h"
#include "queue.h"
#include "stream.h"
#include "threads.h"
#include "updater.h"

struct killer {
	pthread_t thread;
	sigset_t sigset;
	struct mux *mux;
	int killedby;
};

static void		 killer_start(struct killer *, struct mux *);
static void		*killer_run(void *);
static void		 killer_stop(struct killer *);

static int		 proto_waitconnect(int);
static int		 proto_greet(struct config *);
static int		 proto_negproto(struct config *);
static int		 proto_login(struct config *);
static int		 proto_fileattr(struct config *);
static int		 proto_xchgcoll(struct config *);
static struct mux	*proto_mux(struct config *);

static int		 proto_escape(struct stream *, const char *);
static void		 proto_unescape(char *);

static int
proto_waitconnect(int s)
{
	fd_set readfd;
	socklen_t len;
	int error, rv, soerror;

	FD_ZERO(&readfd);
	FD_SET(s, &readfd);

	do {
		rv = select(s + 1, &readfd, NULL, NULL, NULL);
	} while (rv == -1 && errno == EINTR);
	if (rv == -1)
		return (-1);
	/* Check that the connection was really successful. */
	len = sizeof(soerror);
	error = getsockopt(s, SOL_SOCKET, SO_ERROR, &soerror, &len);
	if (error) {
		/* We have no choice but faking an error here. */
		errno = ECONNREFUSED;
		return (-1);
	}
	if (soerror) {
		errno = soerror;
		return (-1);
	}
	return (0);
}

/* Connect to the CVSup server. */
int
proto_connect(struct config *config, int family, uint16_t port)
{
	char addrbuf[NI_MAXHOST];
	/* Enough to hold sizeof("cvsup") or any port number. */
	char servname[8];
	struct addrinfo *res, *ai, hints;
	int error, opt, s;

	s = -1;
	if (port != 0)
		snprintf(servname, sizeof(servname), "%d", port);
	else {
		strncpy(servname, "cvsup", sizeof(servname) - 1);
		servname[sizeof(servname) - 1] = '\0';
	}
	memset(&hints, 0, sizeof(hints));
	hints.ai_family = family;
	hints.ai_socktype = SOCK_STREAM;
	error = getaddrinfo(config->host, servname, &hints, &res);
	/*
	 * Try with the hardcoded port number for OSes that don't
	 * have cvsup defined in the /etc/services file.
	 */
	if (error == EAI_SERVICE) {
		strncpy(servname, "5999", sizeof(servname) - 1);
		servname[sizeof(servname) - 1] = '\0';
		error = getaddrinfo(config->host, servname, &hints, &res);
	}
	if (error) {
		lprintf(0, "Name lookup failure for \"%s\": %s\n", config->host,
		    gai_strerror(error));
		return (STATUS_TRANSIENTFAILURE);
	}
	for (ai = res; ai != NULL; ai = ai->ai_next) {
		s = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol);
		if (s != -1) {
			error = 0;
			if (config->laddr != NULL) {
				opt = 1;
				(void)setsockopt(s, SOL_SOCKET, SO_REUSEADDR,
				    &opt, sizeof(opt));
				error = bind(s, config->laddr,
				    config->laddrlen);
			}
			if (!error) {
				error = connect(s, ai->ai_addr, ai->ai_addrlen);
				if (error && errno == EINTR)
					error = proto_waitconnect(s);
			}
			if (error)
				close(s);
		}
		(void)getnameinfo(ai->ai_addr, ai->ai_addrlen, addrbuf,
		    sizeof(addrbuf), NULL, 0, NI_NUMERICHOST);
		if (s == -1 || error) {
			lprintf(0, "Cannot connect to %s: %s\n", addrbuf,
			    strerror(errno));
			continue;
		}
		lprintf(1, "Connected to %s\n", addrbuf);
		freeaddrinfo(res);
		config->socket = s;
		return (STATUS_SUCCESS);
	}
	freeaddrinfo(res);
	return (STATUS_TRANSIENTFAILURE);
}

/* Greet the server. */
static int
proto_greet(struct config *config)
{
	char *line, *cmd, *msg, *swver;
	struct stream *s;

	s = config->server;
	line = stream_getln(s, NULL);
	cmd = proto_get_ascii(&line);
	if (cmd == NULL)
		goto bad;
	if (strcmp(cmd, "OK") == 0) {
		(void)proto_get_ascii(&line);	/* major number */
		(void)proto_get_ascii(&line);	/* minor number */
		swver = proto_get_ascii(&line);
	} else if (strcmp(cmd, "!") == 0) {
		msg = proto_get_rest(&line);
		if (msg == NULL)
			goto bad;
		lprintf(-1, "Rejected by server: %s\n", msg);
		return (STATUS_TRANSIENTFAILURE);
	} else
		goto bad;
	lprintf(2, "Server software version: %s\n",
	    swver != NULL ? swver : ".");
	return (STATUS_SUCCESS);
bad:
	lprintf(-1, "Invalid greeting from server\n");
	return (STATUS_FAILURE);
}

/* Negotiate protocol version with the server. */
static int
proto_negproto(struct config *config)
{
	struct stream *s;
	char *cmd, *line, *msg;
	int error, maj, min;

	s = config->server;
	proto_printf(s, "PROTO %d %d %s\n", PROTO_MAJ, PROTO_MIN, PROTO_SWVER);
	stream_flush(s);
	line = stream_getln(s, NULL);
	cmd = proto_get_ascii(&line);
	if (line == NULL)
		goto bad;
	if (strcmp(cmd, "!") == 0) {
		msg = proto_get_rest(&line);
		lprintf(-1, "Protocol negotiation failed: %s\n", msg);
		return (1);
	} else if (strcmp(cmd, "PROTO") != 0)
		goto bad;
	error = proto_get_int(&line, &maj, 10);
	if (!error)
		error = proto_get_int(&line, &min, 10);
	if (error)
		goto bad;
	if (maj != PROTO_MAJ || min != PROTO_MIN) {
		lprintf(-1, "Server protocol version %d.%d not supported "
		    "by client\n", maj, min);
		return (STATUS_FAILURE);
	}
	return (STATUS_SUCCESS);
bad:
	lprintf(-1, "Invalid PROTO command from server\n");
	return (STATUS_FAILURE);
}

static int
proto_login(struct config *config)
{
	struct stream *s;
	char host[MAXHOSTNAMELEN];
	char *line, *cmd, *realm, *challenge, *msg;

	s = config->server;
	gethostname(host, sizeof(host));
	host[sizeof(host) - 1] = '\0';
	proto_printf(s, "USER %s %s\n", getlogin(), host);
	stream_flush(s);
	line = stream_getln(s, NULL);
	cmd = proto_get_ascii(&line);
	realm = proto_get_ascii(&line);
	challenge = proto_get_ascii(&line);
	if (challenge == NULL || line != NULL)
		goto bad;
	if (strcmp(realm, ".") != 0 || strcmp(challenge, ".") != 0) {
		lprintf(-1, "Authentication required by the server and not "
		    "supported by client\n");
		return (STATUS_FAILURE);
	}
	proto_printf(s, "AUTHMD5 . . .\n");
	stream_flush(s);
	line = stream_getln(s, NULL);
	cmd = proto_get_ascii(&line);
	if (strcmp(cmd, "OK") == 0)
		return (STATUS_SUCCESS);
	if (strcmp(cmd, "!") == 0) {
		msg = proto_get_rest(&line);
		if (msg == NULL)
			goto bad;
		lprintf(-1, "Server error: %s\n", msg);
		return (STATUS_FAILURE);
	}
bad:
	lprintf(-1, "Invalid server reply to AUTHMD5\n");
	return (STATUS_FAILURE);
}

/*
 * File attribute support negotiation.
 */
static int
proto_fileattr(struct config *config)
{
	fattr_support_t support;
	struct stream *s;
	char *line, *cmd;
	int error, i, n, attr;

	s = config->server;
	lprintf(2, "Negotiating file attribute support\n");
	proto_printf(s, "ATTR %d\n", FT_NUMBER);
	for (i = 0; i < FT_NUMBER; i++)
		proto_printf(s, "%x\n", fattr_supported(i));
	proto_printf(s, ".\n");
	stream_flush(s);
	line = stream_getln(s, NULL);
	if (line == NULL)
		goto bad;
	cmd = proto_get_ascii(&line);
	error = proto_get_int(&line, &n, 10);
	if (error || line != NULL || strcmp(cmd, "ATTR") != 0 || n > FT_NUMBER)
		goto bad;
	for (i = 0; i < n; i++) {
		line = stream_getln(s, NULL);
		if (line == NULL)
			goto bad;
		error = proto_get_int(&line, &attr, 16);
		if (error)
			goto bad;
		support[i] = fattr_supported(i) & attr;
	}
	for (i = n; i < FT_NUMBER; i++)
		support[i] = 0;
	line = stream_getln(s, NULL);
	if (line == NULL || strcmp(line, ".") != 0)
		goto bad;
	memcpy(config->fasupport, support, sizeof(config->fasupport));
	return (STATUS_SUCCESS);
bad:
	lprintf(-1, "Protocol error negotiating attribute support\n");
	return (STATUS_FAILURE);
}

/*
 * Exchange collection information.
 */
static int
proto_xchgcoll(struct config *config)
{
	struct coll *coll;
	struct stream *s;
	struct globtree *diraccept, *dirrefuse;
	struct globtree *fileaccept, *filerefuse;
	char *line, *cmd, *collname, *pat;
	char *msg, *release, *ident, *rcskey, *prefix;
	size_t i, len;
	int error, flags, options;

	s = config->server;
	lprintf(2, "Exchanging collection information\n");
	STAILQ_FOREACH(coll, &config->colls, co_next) {
		proto_printf(s, "COLL %s %s %o %d\n", coll->co_name,
		    coll->co_release, coll->co_umask, coll->co_options);
		for (i = 0; i < pattlist_size(coll->co_accepts); i++) {
		    proto_printf(s, "ACC %s\n",
			pattlist_get(coll->co_accepts, i));
		}
		for (i = 0; i < pattlist_size(coll->co_refusals); i++) {
		    proto_printf(s, "REF %s\n",
			pattlist_get(coll->co_refusals, i));
		}
		proto_printf(s, ".\n");
	}
	proto_printf(s, ".\n");
	stream_flush(s);
	STAILQ_FOREACH(coll, &config->colls, co_next) {
		if (coll->co_options & CO_SKIP)
			continue;
		line = stream_getln(s, NULL);
		if (line == NULL)
			goto bad;
		cmd = proto_get_ascii(&line);
		collname = proto_get_ascii(&line);
		release = proto_get_ascii(&line);
		error = proto_get_int(&line, &options, 10);
		if (error || line != NULL)
			goto bad;
		if (strcmp(cmd, "COLL") != 0 ||
		    strcmp(collname, coll->co_name) != 0 ||
		    strcmp(release, coll->co_release) != 0)
			goto bad;
		coll->co_options =
		    (coll->co_options | (options & CO_SERVMAYSET)) &
		    ~(~options & CO_SERVMAYCLEAR);
		while ((line = stream_getln(s, NULL)) != NULL) {
		 	if (strcmp(line, ".") == 0)
				break;
			cmd = proto_get_ascii(&line);
			if (cmd == NULL)
				goto bad;
			if (strcmp(cmd, "!") == 0) {
				msg = proto_get_rest(&line);
				if (msg == NULL)
					goto bad;
				lprintf(-1, "Server message: %s\n", msg);
			} else if (strcmp(cmd, "PRFX") == 0) {
				prefix = proto_get_ascii(&line);
				if (prefix == NULL || line != NULL)
					goto bad;
				coll->co_cvsroot = xstrdup(prefix);
			} else if (strcmp(cmd, "KEYALIAS") == 0) {
				ident = proto_get_ascii(&line);
				rcskey = proto_get_ascii(&line);
				if (rcskey == NULL || line != NULL)
					goto bad;
				error = keyword_alias(coll->co_keyword, ident,
				    rcskey);
				if (error)
					goto bad;
			} else if (strcmp(cmd, "KEYON") == 0) {
				ident = proto_get_ascii(&line);
				if (ident == NULL || line != NULL)
					goto bad;
				error = keyword_enable(coll->co_keyword, ident);
				if (error)
					goto bad;
			} else if (strcmp(cmd, "KEYOFF") == 0) {
				ident = proto_get_ascii(&line);
				if (ident == NULL || line != NULL)
					goto bad;
				error = keyword_disable(coll->co_keyword,
				    ident);
				if (error)
					goto bad;
			}
		}
		if (line == NULL)
			goto bad;
		keyword_prepare(coll->co_keyword);

		diraccept = globtree_true();
		fileaccept = globtree_true();
		dirrefuse = globtree_false();
		filerefuse = globtree_false();

		if (pattlist_size(coll->co_accepts) > 0) {
			globtree_free(diraccept);
			globtree_free(fileaccept);
			diraccept = globtree_false();
			fileaccept = globtree_false();
			flags = FNM_PATHNAME | FNM_LEADING_DIR |
			    FNM_PREFIX_DIRS;
			for (i = 0; i < pattlist_size(coll->co_accepts); i++) {
				pat = pattlist_get(coll->co_accepts, i);
				diraccept = globtree_or(diraccept,
				    globtree_match(pat, flags));

				len = strlen(pat);
				if (coll->co_options & CO_CHECKOUTMODE &&
				    (len == 0 || pat[len - 1] != '*')) {
					/* We must modify the pattern so that it
					   refers to the RCS file, rather than
					   the checked-out file. */
					xasprintf(&pat, "%s,v", pat);
					fileaccept = globtree_or(fileaccept,
					    globtree_match(pat, flags));
					free(pat);
				} else {
					fileaccept = globtree_or(fileaccept,
					    globtree_match(pat, flags));
				}
			}
		}

		for (i = 0; i < pattlist_size(coll->co_refusals); i++) {
			pat = pattlist_get(coll->co_refusals, i);
			dirrefuse = globtree_or(dirrefuse,
			    globtree_match(pat, 0));
			len = strlen(pat);
			if (coll->co_options & CO_CHECKOUTMODE &&
			    (len == 0 || pat[len - 1] != '*')) {
				/* We must modify the pattern so that it refers
				   to the RCS file, rather than the checked-out
				   file. */
				xasprintf(&pat, "%s,v", pat);
				filerefuse = globtree_or(filerefuse,
				    globtree_match(pat, 0));
				free(pat);
			} else {
				filerefuse = globtree_or(filerefuse,
				    globtree_match(pat, 0));
			}
		}

		coll->co_dirfilter = globtree_and(diraccept,
		    globtree_not(dirrefuse));
		coll->co_filefilter = globtree_and(fileaccept,
		    globtree_not(filerefuse));

		/* At this point we don't need the pattern lists anymore. */
		pattlist_free(coll->co_accepts);
		pattlist_free(coll->co_refusals);
		coll->co_accepts = NULL;
		coll->co_refusals = NULL;

		/* Set up a mask of file attributes that we don't want to sync
		   with the server. */
		if (!(coll->co_options & CO_SETOWNER))
			coll->co_attrignore |= FA_OWNER | FA_GROUP;
		if (!(coll->co_options & CO_SETMODE))
			coll->co_attrignore |= FA_MODE;
		if (!(coll->co_options & CO_SETFLAGS))
			coll->co_attrignore |= FA_FLAGS;
	}
	return (STATUS_SUCCESS);
bad:
	lprintf(-1, "Protocol error during collection exchange\n");
	return (STATUS_FAILURE);
}

static struct mux *
proto_mux(struct config *config)
{
	struct mux *m;
	struct stream *s, *wr;
	struct chan *chan0, *chan1;
	int id;

	s = config->server;
	lprintf(2, "Establishing multiplexed-mode data connection\n");
	proto_printf(s, "MUX\n");
	stream_flush(s);
	m = mux_open(config->socket, &chan0);
	if (m == NULL) {
		lprintf(-1, "Cannot open the multiplexer\n");
		return (NULL);
	}
	id = chan_listen(m);
	if (id == -1) {
		lprintf(-1, "ChannelMux.Listen failed: %s\n", strerror(errno));
		mux_close(m);
		return (NULL);
	}
	wr = stream_open(chan0, NULL, (stream_writefn_t *)chan_write, NULL);
	proto_printf(wr, "CHAN %d\n", id);
	stream_close(wr);
	chan1 = chan_accept(m, id);
	if (chan1 == NULL) {
		lprintf(-1, "ChannelMux.Accept failed: %s\n", strerror(errno));
		mux_close(m);
		return (NULL);
	}
	config->chan0 = chan0;
	config->chan1 = chan1;
	return (m);
}

/*
 * Initializes the connection to the CVSup server, that is handle
 * the protocol negotiation, logging in, exchanging file attributes
 * support and collections information, and finally run the update
 * session.
 */
int
proto_run(struct config *config)
{
	struct thread_args lister_args;
	struct thread_args detailer_args;
	struct thread_args updater_args;
	struct thread_args *args;
	struct killer killer;
	struct threads *workers;
	struct mux *m;
	int i, status;

	/*
	 * We pass NULL for the close() function because we'll reuse
	 * the socket after the stream is closed.
	 */
	config->server = stream_open_fd(config->socket, stream_read_fd,
	    stream_write_fd, NULL);
	status = proto_greet(config);
	if (status == STATUS_SUCCESS)
		status = proto_negproto(config);
	if (status == STATUS_SUCCESS)
		status = proto_login(config);
	if (status == STATUS_SUCCESS)
		status = proto_fileattr(config);
	if (status == STATUS_SUCCESS)
		status = proto_xchgcoll(config);
	if (status != STATUS_SUCCESS)
		return (status);

	/* Multi-threaded action starts here. */
	m = proto_mux(config);
	if (m == NULL)
		return (STATUS_FAILURE);

	stream_close(config->server);
	config->server = NULL;
	config->fixups = fixups_new();
	killer_start(&killer, m);

	/* Start the worker threads. */
	workers = threads_new();
	args = &lister_args;
	args->config = config;
	args->status = -1;
	args->errmsg = NULL;
	args->rd = NULL;
	args->wr = stream_open(config->chan0,
	    NULL, (stream_writefn_t *)chan_write, NULL);
	threads_create(workers, lister, args);

	args = &detailer_args;
	args->config = config;
	args->status = -1;
	args->errmsg = NULL;
	args->rd = stream_open(config->chan0,
	    (stream_readfn_t *)chan_read, NULL, NULL);
	args->wr = stream_open(config->chan1,
	    NULL, (stream_writefn_t *)chan_write, NULL);
	threads_create(workers, detailer, args);

	args = &updater_args;
	args->config = config;
	args->status = -1;
	args->errmsg = NULL;
	args->rd = stream_open(config->chan1,
	    (stream_readfn_t *)chan_read, NULL, NULL);
	args->wr = NULL;
	threads_create(workers, updater, args);

	lprintf(2, "Running\n");
	/* Wait for all the worker threads to finish. */
	status = STATUS_SUCCESS;
	for (i = 0; i < 3; i++) {
		args = threads_wait(workers);
		if (args->rd != NULL)
			stream_close(args->rd);
		if (args->wr != NULL)
			stream_close(args->wr);
		if (args->status != STATUS_SUCCESS) {
			assert(args->errmsg != NULL);
			if (status == STATUS_SUCCESS) {
				status = args->status;
				/* Shutdown the multiplexer to wake up all
				   the other threads. */
				mux_shutdown(m, args->errmsg, status);
			}
			free(args->errmsg);
		}
	}
	threads_free(workers);
	if (status == STATUS_SUCCESS) {
		lprintf(2, "Shutting down connection to server\n");
		chan_close(config->chan0);
		chan_close(config->chan1);
		chan_wait(config->chan0);
		chan_wait(config->chan1);
		mux_shutdown(m, NULL, STATUS_SUCCESS);
	}
	killer_stop(&killer);
	fixups_free(config->fixups);
	status = mux_close(m);
	if (status == STATUS_SUCCESS) {
		lprintf(1, "Finished successfully\n");
	} else if (status == STATUS_INTERRUPTED) {
		lprintf(-1, "Interrupted\n");
		if (killer.killedby != -1)
			kill(getpid(), killer.killedby);
	}
	return (status);
}

/*
 * Write a string into the stream, escaping characters as needed.
 * Characters escaped:
 *
 * SPACE	-> "\_"
 * TAB		->  "\t"
 * NEWLINE	-> "\n"
 * CR		-> "\r"
 * \		-> "\\"
 */
static int
proto_escape(struct stream *wr, const char *s)
{
	size_t len;
	ssize_t n;
	char c;

	/* Handle characters that need escaping. */
	do {
		len = strcspn(s, " \t\r\n\\");
		n = stream_write(wr, s, len);
		if (n == -1)
			return (-1);
		c = s[len];
		switch (c) {
		case ' ':
			n = stream_write(wr, "\\_", 2);
			break;
		case '\t':
			n = stream_write(wr, "\\t", 2);
			break;
		case '\r':
			n = stream_write(wr, "\\r", 2);
			break;
		case '\n':
			n = stream_write(wr, "\\n", 2);
			break;
		case '\\':
			n = stream_write(wr, "\\\\", 2);
			break;
		}
		if (n == -1)
			return (-1);
		s += len + 1;
	} while (c != '\0');
	return (0);
}

/*
 * A simple printf() implementation specifically tailored for csup.
 * List of the supported formats:
 *
 * %c		Print a char.
 * %d or %i	Print an int as decimal.
 * %x		Print an int as hexadecimal.
 * %o		Print an int as octal.
 * %t		Print a time_t as decimal.
 * %s		Print a char * escaping some characters as needed.
 * %S		Print a char * without escaping.
 * %f		Print an encoded struct fattr *.
 * %F		Print an encoded struct fattr *, specifying the supported
 * 		attributes.
 */
int
proto_printf(struct stream *wr, const char *format, ...)
{
	fattr_support_t *support;
	long long longval;
	struct fattr *fa;
	const char *fmt;
	va_list ap;
	char *cp, *s, *attr;
	ssize_t n;
	int rv, val, ignore;
	char c;

	n = 0;
	rv = 0;
	fmt = format;
	va_start(ap, format);
	while ((cp = strchr(fmt, '%')) != NULL) {
		if (cp > fmt) {
			n = stream_write(wr, fmt, cp - fmt);
			if (n == -1)
				return (-1);
		}
		if (*++cp == '\0')
			goto done;
		switch (*cp) {
		case 'c':
			c = va_arg(ap, int);
			rv = stream_printf(wr, "%c", c);
			break;
		case 'd':
		case 'i':
			val = va_arg(ap, int);
			rv = stream_printf(wr, "%d", val);
			break;
		case 'x':
			val = va_arg(ap, int);
			rv = stream_printf(wr, "%x", val);
			break;
		case 'o':
			val = va_arg(ap, int);
			rv = stream_printf(wr, "%o", val);
			break;
		case 'S':
			s = va_arg(ap, char *);
			assert(s != NULL);
			rv = stream_printf(wr, "%s", s);
			break;
		case 's':
			s = va_arg(ap, char *);
			assert(s != NULL);
			rv = proto_escape(wr, s);
			break;
		case 't':
			longval = (long long)va_arg(ap, time_t);
			rv = stream_printf(wr, "%lld", longval);
			break;
		case 'f':
			fa = va_arg(ap, struct fattr *);
			attr = fattr_encode(fa, NULL, 0);
			rv = proto_escape(wr, attr);
			free(attr);
			break;
		case 'F':
			fa = va_arg(ap, struct fattr *);
			support = va_arg(ap, fattr_support_t *);
			ignore = va_arg(ap, int);
			attr = fattr_encode(fa, *support, ignore);
			rv = proto_escape(wr, attr);
			free(attr);
			break;
		case '%':
			n = stream_write(wr, "%", 1);
			if (n == -1)
				return (-1);
			break;
		}
		if (rv == -1)
			return (-1);
		fmt = cp + 1;
	}
	if (*fmt != '\0') {
		rv = stream_printf(wr, "%s", fmt);
		if (rv == -1)
			return (-1);
	}
done:
	va_end(ap);
	return (0);
}

/*
 * Unescape the string, see proto_escape().
 */
static void
proto_unescape(char *s)
{
	char *cp, *cp2;

	cp = s;
	while ((cp = strchr(cp, '\\')) != NULL) {
		switch (cp[1]) {
		case '_':
			*cp = ' ';
			break;
		case 't':
			*cp = '\t';
			break;
		case 'r':
			*cp = '\r';
			break;
		case 'n':
			*cp = '\n';
			break;
		case '\\':
			*cp = '\\';
			break;
		default:
			*cp = *(cp + 1);
		}
		cp2 = ++cp;
		while (*cp2 != '\0') {
			*cp2 = *(cp2 + 1);
			cp2++;
		}
	}
}

/*
 * Get an ascii token in the string.
 */
char *
proto_get_ascii(char **s)
{
	char *ret;

	ret = strsep(s, " ");
	if (ret == NULL)
		return (NULL);
	/* Make sure we disallow 0-length fields. */
	if (*ret == '\0') {
		*s = NULL;
		return (NULL);
	}
	proto_unescape(ret);
	return (ret);
}

/*
 * Get the rest of the string.
 */
char *
proto_get_rest(char **s)
{
	char *ret;

	if (s == NULL)
		return (NULL);
	ret = *s;
	proto_unescape(ret);
	*s = NULL;
	return (ret);
}

/*
 * Get an int token.
 */
int
proto_get_int(char **s, int *val, int base)
{
	char *cp, *end;
	long longval;

	cp = proto_get_ascii(s);
	if (cp == NULL)
		return (-1);
	errno = 0;
	longval = strtol(cp, &end, base);
	if (errno || *end != '\0')
		return (-1);
	if (longval > INT_MAX || longval < INT_MIN) {
		errno = ERANGE;
		return (-1);
	}
	*val = longval;
	return (0);
}

/*
 * Get a time_t token.
 *
 * Ideally, we would use an intmax_t and strtoimax() here, but strtoll()
 * is more portable and 64bits should be enough for a timestamp.
 */
int
proto_get_time(char **s, time_t *val)
{
	long long tmp;
	char *cp, *end;

	cp = proto_get_ascii(s);
	if (cp == NULL)
		return (-1);
	errno = 0;
	tmp = strtoll(cp, &end, 10);
	if (errno || *end != '\0')
		return (-1);
	*val = (time_t)tmp;
	return (0);
}

/* Start the killer thread.  It is used to protect against some signals
   during the multi-threaded run so that we can gracefully fail.  */
static void
killer_start(struct killer *k, struct mux *m)
{
	int error;

	k->mux = m;
	k->killedby = -1;
	sigemptyset(&k->sigset);
	sigaddset(&k->sigset, SIGINT);
	sigaddset(&k->sigset, SIGHUP);
	sigaddset(&k->sigset, SIGTERM);
	sigaddset(&k->sigset, SIGPIPE);
	pthread_sigmask(SIG_BLOCK, &k->sigset, NULL);
	error = pthread_create(&k->thread, NULL, killer_run, k);
	if (error)
		err(1, "pthread_create");
}

/* The main loop of the killer thread. */
static void *
killer_run(void *arg)
{
	struct killer *k;
	int error, sig, old;

	k = arg;
again:
	error = sigwait(&k->sigset, &sig);
	assert(!error);
	if (sig == SIGINT || sig == SIGHUP || sig == SIGTERM) {
		if (k->killedby == -1) {
			k->killedby = sig;
			/* Ensure we don't get canceled during the shutdown. */
			pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &old);
			mux_shutdown(k->mux, "Cleaning up ...",
			    STATUS_INTERRUPTED);
			pthread_setcancelstate(old, NULL);
		}
	}
	goto again;
}

/* Stop the killer thread. */
static void
killer_stop(struct killer *k)
{
	void *val;
	int error;

	error = pthread_cancel(k->thread);
	assert(!error);
	pthread_join(k->thread, &val);
	assert(val == PTHREAD_CANCELED);
	pthread_sigmask(SIG_UNBLOCK, &k->sigset, NULL);
}