<NME> server.c
<BEF> /*
 * Copyright 2013-2022 Fabian Groffen
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */


#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include <fcntl.h>
#include <pthread.h>
#include <poll.h>
#include <errno.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <netdb.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <sys/resource.h>
#include <assert.h>

#include "relay.h"
#include "queue.h"
#include "dispatcher.h"
#include "collector.h"
#include "server.h"

#ifdef HAVE_GZIP
#include <zlib.h>
#endif
#ifdef HAVE_LZ4
#include <lz4.h>
#include <lz4frame.h>
#endif
#ifdef HAVE_SNAPPY
#include <snappy-c.h>
#endif
#ifdef HAVE_SSL
#include <openssl/ssl.h>
#include <openssl/err.h>
#endif

typedef struct _z_strm {
	ssize_t (*strmwrite)(struct _z_strm *, const void *, size_t);
	int (*strmflush)(struct _z_strm *);
	int (*strmclose)(struct _z_strm *);
	const char *(*strmerror)(struct _z_strm *, int);     /* get last err str */
	struct _z_strm *nextstrm;                            /* set when chained */
#if defined(HAVE_LZ4) || defined(HAVE_SNAPPY) || defined(HAVE_GZIP)
	char obuf[METRIC_BUFSIZ];
	int obuflen;
#endif
#ifdef HAVE_SSL
	SSL_CTX *ctx;
#endif
	union {
#ifdef HAVE_GZIP
		z_streamp gz;
#endif
#ifdef HAVE_LZ4
		struct {
			void *cbuf;
			size_t cbuflen;
		} z;
#endif
#ifdef HAVE_SSL
		SSL *ssl;
#endif
		int sock;
	} hdl;
} z_strm;

struct _server {
	const char *ip;
	unsigned short port;
	char *instance;
	struct addrinfo *saddr;
	struct addrinfo *hint;
	char reresolve:1;
	int fd;
	z_strm *strm;
	queue *queue;
	size_t bsize;
	short iotimeout;
	unsigned int sockbufsize;
	unsigned char maxstalls:SERVER_STALL_BITS;
	const char **batch;
	con_type type;
	con_trnsp transport;
	char *mtlspemcert;
	char *mtlspemkey;
	con_proto ctype;
	pthread_t tid;
	struct _server **secondaries;
	size_t secondariescnt;
	char failover:1;
	char failure;       /* full byte for atomic access */
	char running;       /* full byte for atomic access */
	char keep_running;  /* full byte for atomic access */
	unsigned char stallseq;  /* full byte for atomic access */
	size_t metrics;
	size_t dropped;
	size_t stalls;
	size_t ticks;
	size_t prevmetrics;
	size_t prevdropped;
	size_t prevstalls;
	size_t prevticks;
};


/* connection specific writers and closers */

/* ordinary socket */
static inline ssize_t
sockwrite(z_strm *strm, const void *buf, size_t sze)
{
	return write(strm->hdl.sock, buf, sze);
}

static inline int
sockflush(z_strm *strm)
{
	/* noop, we don't use a stream in the normal case */
	(void)strm;
	return 0;
}

static inline int
sockclose(z_strm *strm)
{
	return close(strm->hdl.sock);
}

static inline const char *
sockerror(z_strm *strm, int rval)
{
	(void)strm;
	(void)rval;
	return strerror(errno);
}

#ifdef HAVE_GZIP
/* gzip wrapped socket */
static inline int gzipflush(z_strm *strm);

static inline ssize_t
gzipwrite(z_strm *strm, const void *buf, size_t sze)
{
	/* ensure we have space available */
	if (strm->obuflen + sze > METRIC_BUFSIZ)
		if (gzipflush(strm) != 0)
			return -1;

	/* append metric to buf */
	memcpy(strm->obuf + strm->obuflen, buf, sze);
	strm->obuflen += sze;

	return sze;
}

static inline int
gzipflush(z_strm *strm)
{
	int cret;
	char cbuf[METRIC_BUFSIZ];
	size_t cbuflen;
	char *cbufp;
	int oret;

	strm->hdl.gz->next_in = (Bytef *)strm->obuf;
	strm->hdl.gz->avail_in = strm->obuflen;
	strm->hdl.gz->next_out = (Bytef *)cbuf;
	strm->hdl.gz->avail_out = sizeof(cbuf);

	do {
		cret = deflate(strm->hdl.gz, Z_PARTIAL_FLUSH);
		if (cret == Z_OK && strm->hdl.gz->avail_out == 0) {
			/* too large output block, unlikely given the input, discard */
			/* TODO */
			break;
		}
		if (cret == Z_OK) {
			cbufp = cbuf;
			cbuflen = sizeof(cbuf) - strm->hdl.gz->avail_out;
			while (cbuflen > 0) {
				oret = strm->nextstrm->strmwrite(strm->nextstrm,
						cbufp, cbuflen);
				if (oret < 0)
					return -1;  /* failure is failure */

				/* update counters to possibly retry the remaining bit */
				cbufp += oret;
				cbuflen -= oret;
			}

			if (strm->hdl.gz->avail_in == 0)
				break;

			strm->hdl.gz->next_out = (Bytef *)cbuf;
			strm->hdl.gz->avail_out = sizeof(cbuf);
		} else {
			/* discard */
			break;
		}
	} while (1);
	/* flush whatever we wrote */
	strm->nextstrm->strmflush(strm->nextstrm);

	/* reset the write position, from this point it will always need to
	 * restart */
	strm->obuflen = 0;

	if (cret != Z_OK)
		return -1;  /* we must reset/free gzip */

	return 0;
}

static inline int
gzipclose(z_strm *strm)
{
	int ret = strm->nextstrm->strmclose(strm->nextstrm);
	deflateEnd(strm->hdl.gz);
	free(strm->hdl.gz);
	return ret;
}

static inline const char *
gziperror(z_strm *strm, int rval)
{
	return strm->nextstrm->strmerror(strm->nextstrm, rval);
}
#endif

#ifdef HAVE_LZ4
/* lz4 wrapped socket */

static inline int lzflush(z_strm *strm);

static inline ssize_t
lzwrite(z_strm *strm, const void *buf, size_t sze)
{
	size_t towrite = sze;

	/* use the same strategy as gzip: fill the buffer until space
	   runs out. we completely fill the output buffer before flushing */

	while (towrite > 0) {

		size_t avail = METRIC_BUFSIZ - strm->obuflen;
		size_t copysize = towrite > avail ? avail : towrite;

		/* copy into the output buffer as much as we can */

		if (copysize > 0) {
			memcpy(strm->obuf + strm->obuflen, buf, copysize);
			strm->obuflen += copysize;
			towrite -= copysize;
			buf += copysize;
		}

		/* if output buffer is full & still have bytes to write, flush now */

		if (strm->obuflen == METRIC_BUFSIZ && towrite > 0 && lzflush(strm) != 0) {
			logerr("Failed to flush LZ4 data to make space\n");
			return -1;
		}
	}

	return sze;
}

static inline int
lzflush(z_strm *strm)
{
	int oret;
	size_t ret;

	/* anything to do? */

	if (strm->obuflen == 0)
		return 0;

	/* the buffered data goes out as a single frame */

	ret = LZ4F_compressFrame(strm->hdl.z.cbuf, strm->hdl.z.cbuflen, strm->obuf, strm->obuflen, NULL);
	if (LZ4F_isError(ret)) {
		logerr("Failed to compress %d LZ4 bytes into a frame: %d\n",
		       strm->obuflen, (int)ret);
		return -1;
	}
		/* write and flush */

	if ((oret = strm->nextstrm->strmwrite(strm->nextstrm,
					      strm->hdl.z.cbuf, ret)) < 0) {
		logerr("Failed to write %lu bytes of compressed LZ4 data: %d\n",
		       ret, oret);
		return oret;
	}

	strm->nextstrm->strmflush(strm->nextstrm);

	/* the buffer is gone. reset the write position */

	strm->obuflen = 0;
	return 0;
}

static inline int
lzclose(z_strm *strm)
{
	lzflush(strm);
	free(strm->hdl.z.cbuf);
	return strm->nextstrm->strmclose(strm->nextstrm);
}

static inline const char *
lzerror(z_strm *strm, int rval)
{
	return strm->nextstrm->strmerror(strm->nextstrm, rval);
}
#endif

#ifdef HAVE_SNAPPY
/* snappy wrapped socket */
static inline int snappyflush(z_strm *strm);

static inline ssize_t
snappywrite(z_strm *strm, const void *buf, size_t sze)
{
	/* ensure we have space available */
	if (strm->obuflen + sze > METRIC_BUFSIZ)
		if (snappyflush(strm) != 0)
			return -1;

	/* append metric to buf */
	memcpy(strm->obuf + strm->obuflen, buf, sze);
	strm->obuflen += sze;

	return sze;
}

static inline int
snappyflush(z_strm *strm)
{
	int cret;
	char cbuf[METRIC_BUFSIZ];
	size_t cbuflen = sizeof(cbuf);
	char *cbufp = cbuf;
	int oret;

	cret = snappy_compress(strm->obuf, strm->obuflen, cbuf, &cbuflen);
	/* reset the write position, from this point it will always need to
	 * restart */
	strm->obuflen = 0;

	if (cret != SNAPPY_OK)
		return -1;  /* we must reset/free snappy */
	while (cbuflen > 0) {
		oret = strm->nextstrm->strmwrite(strm->nextstrm, cbufp, cbuflen);
		if (oret < 0)
			return -1;  /* failure is failure */
		strm->nextstrm->strmflush(strm->nextstrm);
		/* update counters to possibly retry the remaining bit */
		cbufp += oret;
		cbuflen -= oret;
	}

	return 0;
}

static inline int
snappyclose(z_strm *strm)
{
	int ret = strm->nextstrm->strmclose(strm->nextstrm);
	return ret;
}

static inline const char *
snappyerror(z_strm *strm, int rval)
{
	return strm->nextstrm->strmerror(strm->nextstrm, rval);
}
#endif

#ifdef HAVE_SSL
/* (Open|Libre)SSL wrapped socket */
static inline ssize_t
sslwrite(z_strm *strm, const void *buf, size_t sze)
{
	return (ssize_t)SSL_write(strm->hdl.ssl, buf, (int)sze);
}

static inline int
sslflush(z_strm *strm)
{
	/* noop */
	(void)strm;
	return 0;
}

static inline int
sslclose(z_strm *strm)
{
	int sock = SSL_get_fd(strm->hdl.ssl);
	SSL_free(strm->hdl.ssl);
	return close(sock);
}

static char _sslerror_buf[256];
static inline const char *
sslerror(z_strm *strm, int rval)
{
	int err = SSL_get_error(strm->hdl.ssl, rval);
	switch (err) {
		case SSL_ERROR_NONE:
			snprintf(_sslerror_buf, sizeof(_sslerror_buf),
					"%d: SSL_ERROR_NONE", err);
			break;
		case SSL_ERROR_ZERO_RETURN:
			snprintf(_sslerror_buf, sizeof(_sslerror_buf),
					"%d: TLS/SSL connection has been closed", err);
			break;
		case SSL_ERROR_WANT_READ:
		case SSL_ERROR_WANT_WRITE:
		case SSL_ERROR_WANT_CONNECT:
		case SSL_ERROR_WANT_ACCEPT:
			snprintf(_sslerror_buf, sizeof(_sslerror_buf),
					"%d: the read or write operation did not complete", err);
			break;
		case SSL_ERROR_WANT_X509_LOOKUP:
			snprintf(_sslerror_buf, sizeof(_sslerror_buf),
					"%d: call callback via SSL_CTX_set_client_cert_cb()", err);
			break;
#ifdef SSL_ERROR_WANT_ASYNC
		case SSL_ERROR_WANT_ASYNC:
			snprintf(_sslerror_buf, sizeof(_sslerror_buf),
					"%d: asynchronous engine is still processing data", err);
			break;
#endif
#ifdef SSL_ERROR_WANT_ASYNC_JOB
		case SSL_ERROR_WANT_ASYNC_JOB:
			snprintf(_sslerror_buf, sizeof(_sslerror_buf),
					"%d: no async jobs available in the pool", err);
			break;
#endif
		case SSL_ERROR_SYSCALL:
			snprintf(_sslerror_buf, sizeof(_sslerror_buf),
					"%d: I/O error: %s", err, strerror(errno));
			break;
		case SSL_ERROR_SSL:
			ERR_error_string_n(ERR_get_error(),
					_sslerror_buf, sizeof(_sslerror_buf));
			break;
	}
	return _sslerror_buf;
}
#endif

/**
 * Reads from the queue and sends items to the remote server.  This
 * function is designed to be a thread.  Data sending is attempted to be
 * batched, but sent one by one to reduce loss on sending failure.
 * A connection with the server is maintained for as long as there is
 * data to be written.  As soon as there is none, the connection is
 * dropped if a timeout of DISCONNECT_WAIT_TIME exceeds.
 */
static void *
server_queuereader(void *d)
{
	server *self = (server *)d;
	size_t len;
	ssize_t slen;
	const char **metric = self->batch;
	struct timeval start, stop;
	struct timeval timeout;
	queue *squeue;
	char idle = 0;
	size_t *secpos = NULL;
	unsigned char cnt;
	const char *p;

	*metric = NULL;

#define FAIL_WAIT_TIME   6  /* 6 * 250ms = 1.5s */
}

/**
 * Starts a previously created server using server_new().  Returns false
 * if starting a thread failed, after which the caller should
 * server_free() the given s pointer.
 */
char
server_start(server *s)
{
	if (pthread_create(&s->tid, NULL, &server_queuereader, s) != 0)
		return 0;
	
	return 1;
}

/**
			}
			if (idle == 1)
				/* ensure blocks are pushed out as soon as we're idling,
				 * this allows compressors to benefit from a larger
				 * stream of data to gain better compression */
				self->strm->strmflush(self->strm);
			gettimeofday(&stop, NULL);
			__sync_add_and_fetch(&(self->ticks), timediff(start, stop));
			if (__sync_bool_compare_and_swap(&(self->keep_running), 0, 0))
				break;
			/* nothing to do, so slow down for a bit */
			usleep((200 + (rand() % 100)) * 1000);  /* 200ms - 300ms */
			/* if we are in failure mode, keep checking if we can
			 * connect, this avoids unnecessary queue moves */
			if (__sync_bool_compare_and_swap(&(self->failure), 0, 0))
				/* it makes no sense to try and do something, so skip */
				continue;
		} else if (self->secondariescnt > 0 &&
				(__sync_add_and_fetch(&(self->failure), 0) >= FAIL_WAIT_TIME ||
				 (!self->failover && LEN_CRITICAL(self->queue))))
		{
			size_t i;

			gettimeofday(&start, NULL);
			if (self->secondariescnt > 0) {
				if (secpos == NULL) {
					secpos = malloc(sizeof(size_t) * self->secondariescnt);
					if (secpos == NULL) {
						logerr("server: failed to allocate memory "
								"for secpos\n");
						gettimeofday(&stop, NULL);
						__sync_add_and_fetch(&(self->ticks),
								timediff(start, stop));
						continue;
					}
					for (i = 0; i < self->secondariescnt; i++)
						secpos[i] = i;
				}
				if (!self->failover) {
					/* randomise the failover list such that in the
					 * grand scheme of things we don't punish the first
					 * working server in the list to deal with all
					 * traffic meant for a now failing server */
					for (i = 0; i < self->secondariescnt; i++) {
						size_t n = rand() % (self->secondariescnt - i);
						if (n != i) {
							size_t t = secpos[n];
							secpos[n] = secpos[i];
							secpos[i] = t;
						}
					}
				}
			}

			/* offload data from our queue to our secondaries
			 * when doing so, observe the following:
			 * - avoid nodes that are in failure mode
			 * - avoid nodes which queues are >= critical_len
			 * when no nodes remain given the above
			 * - send to nodes which queue size < critical_len
			 * where there are no such nodes
			 * - do nothing (we will overflow, since we can't send
			 *   anywhere) */
			*metric = NULL;
			squeue = NULL;
			for (i = 0; i < self->secondariescnt; i++) {
				/* both conditions below make sure we skip ourself */
				if (__sync_add_and_fetch(
							&(self->secondaries[secpos[i]]->failure), 0))
					continue;
				squeue = self->secondaries[secpos[i]]->queue;
				if (!self->failover && LEN_CRITICAL(squeue)) {
					squeue = NULL;
					continue;
				}
				if (*metric == NULL) {
					/* send up to batch size of our queue to this queue */
					len = queue_dequeue_vector(
							self->batch, self->queue, self->bsize);
					self->batch[len] = NULL;
					metric = self->batch;
				}

				for (; *metric != NULL; metric++)
					if (!queue_putback(squeue, *metric))
						break;
				/* try to put back stuff that didn't fit */
				for (; *metric != NULL; metric++)
					if (!queue_putback(self->queue, *metric))
						break;
			}
			for (; *metric != NULL; metric++) {
				if (mode & MODE_DEBUG)
					logerr("dropping metric: %s", *metric);
				free((char *)*metric);
				__sync_add_and_fetch(&(self->dropped), 1);
			}
			gettimeofday(&stop, NULL);
			__sync_add_and_fetch(&(self->ticks),
					timediff(start, stop));
			if (squeue == NULL) {
				/* we couldn't do anything, take it easy for a bit */
				if (__sync_add_and_fetch(&(self->failure), 0) > 1) {
					/* This is a compound because I can't seem to figure
					 * out how to atomically just "set" a variable.
					 * It's not bad when in the middle there is a ++,
					 * all that counts is that afterwards its > 0. */
					__sync_and_and_fetch(&(self->failure), 0);
					__sync_add_and_fetch(&(self->failure), 1);
				}
				if (__sync_bool_compare_and_swap(&(self->keep_running), 0, 0))
					break;
				usleep((200 + (rand() % 100)) * 1000);  /* 200ms - 300ms */
			}
		} else if (__sync_add_and_fetch(&(self->failure), 0) > 0) {
			if (__sync_bool_compare_and_swap(&(self->keep_running), 0, 0))
				break;
			usleep((200 + (rand() % 100)) * 1000);  /* 200ms - 300ms */
			/* avoid overflowing */
			if (__sync_add_and_fetch(&(self->failure), 0) > FAIL_WAIT_TIME)
				__sync_sub_and_fetch(&(self->failure), 1);
		}

		/* at this point we've got work to do, if we're instructed to
		 * shut down, however, try to get everything out of the door
		 * (until we fail, see top of this loop) */

		gettimeofday(&start, NULL);

		/* try to connect */
		if (self->fd < 0) {
			if (self->reresolve) {  /* can only be CON_UDP/CON_TCP */
				struct addrinfo *saddr;
				char sport[8];

				/* re-lookup the address info, if it fails, stay with
				 * whatever we have such that resolution errors incurred
				 * after starting the relay won't make it fail */
				freeaddrinfo(self->saddr);
				snprintf(sport, sizeof(sport), "%u", self->port);
				if (getaddrinfo(self->ip, sport, self->hint, &saddr) == 0) {
					self->saddr = saddr;
				} else {
					if (__sync_fetch_and_add(&(self->failure), 1) == 0)
						logerr("failed to resolve %s:%u, server unavailable\n",
								self->ip, self->port);
					self->saddr = NULL;
					/* this will break out below */
				}
			}

			if (self->ctype == CON_PIPE) {
				int intconn[2];
				if (pipe(intconn) < 0) {
					if (__sync_fetch_and_add(&(self->failure), 1) == 0)
						logerr("failed to create pipe: %s\n", strerror(errno));
					continue;
				}
				dispatch_addconnection(intconn[0], NULL);
				self->fd = intconn[1];
			} else if (self->ctype == CON_FILE) {
				if ((self->fd = open(self->ip,
								O_WRONLY | O_APPEND | O_CREAT,
								S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH)) < 0)
				{
					if (__sync_fetch_and_add(&(self->failure), 1) == 0)
						logerr("failed to open file '%s': %s\n",
								self->ip, strerror(errno));
					continue;
				}
			} else if (self->ctype == CON_UDP) {
				struct addrinfo *walk;

				for (walk = self->saddr; walk != NULL; walk = walk->ai_next) {
					if ((self->fd = socket(walk->ai_family,
									walk->ai_socktype,
									walk->ai_protocol)) < 0)
					{
						if (walk->ai_next == NULL &&
								__sync_fetch_and_add(&(self->failure), 1) == 0)
							logerr("failed to create udp socket: %s\n",
									strerror(errno));
						continue;
					}
					if (connect(self->fd, walk->ai_addr, walk->ai_addrlen) < 0)
					{
						if (walk->ai_next == NULL &&
								__sync_fetch_and_add(&(self->failure), 1) == 0)
							logerr("failed to connect udp socket: %s\n",
									strerror(errno));
						close(self->fd);
						self->fd = -1;
						continue;
					}

					/* we made it up here, so this connection is usable */
					break;
				}
				/* if this didn't resolve to anything, treat as failure */
				if (self->saddr == NULL)
					__sync_add_and_fetch(&(self->failure), 1);
				/* if all addrinfos failed, try again later */
				if (self->fd < 0)
					continue;
			} else {  /* CON_TCP */
				int ret;
				int args;
				struct addrinfo *walk;

				for (walk = self->saddr; walk != NULL; walk = walk->ai_next) {
					if ((self->fd = socket(walk->ai_family,
									walk->ai_socktype,
									walk->ai_protocol)) < 0)
					{
						if (walk->ai_next == NULL &&
								__sync_fetch_and_add(&(self->failure), 1) == 0)
							logerr("failed to create socket: %s\n",
									strerror(errno));
						continue;
					}

					/* put socket in non-blocking mode such that we can
					 * poll() (time-out) on the connect() call */
					args = fcntl(self->fd, F_GETFL, NULL);
					if (fcntl(self->fd, F_SETFL, args | O_NONBLOCK) < 0) {
						logerr("failed to set socket non-blocking mode: %s\n",
								strerror(errno));
						close(self->fd);
						self->fd = -1;
						continue;
					}
					ret = connect(self->fd, walk->ai_addr, walk->ai_addrlen);

					if (ret < 0 && errno == EINPROGRESS) {
						/* wait for connection to succeed if the OS thinks
						 * it can succeed */
						struct pollfd ufds[1];
						ufds[0].fd = self->fd;
						ufds[0].events = POLLIN | POLLOUT;
						ret = poll(ufds, 1, self->iotimeout + (rand() % 100));
						if (ret == 0) {
							/* time limit expired */
							if (walk->ai_next == NULL &&
									__sync_fetch_and_add(
										&(self->failure), 1) == 0)
								logerr("failed to connect() to "
										"%s:%u: Operation timed out\n",
										self->ip, self->port);
							close(self->fd);
							self->fd = -1;
							continue;
						} else if (ret < 0) {
							/* some select error occurred */
							if (walk->ai_next &&
									__sync_fetch_and_add(
										&(self->failure), 1) == 0)
								logerr("failed to poll() for %s:%u: %s\n",
										self->ip, self->port, strerror(errno));
							close(self->fd);
							self->fd = -1;
							continue;
						} else {
							if (ufds[0].revents & POLLHUP) {
								if (walk->ai_next == NULL &&
										__sync_fetch_and_add(
											&(self->failure), 1) == 0)
									logerr("failed to connect() for %s:%u: "
											"Connection refused\n",
											self->ip, self->port);
								close(self->fd);
								self->fd = -1;
								continue;
							}
						}
					} else if (ret < 0) {
						if (walk->ai_next == NULL &&
								__sync_fetch_and_add(&(self->failure), 1) == 0)
						{
							logerr("failed to connect() to %s:%u: %s\n",
									self->ip, self->port, strerror(errno));
							dispatch_check_rlimit_and_warn();
						}
						close(self->fd);
						self->fd = -1;
						continue;
					}

					/* make socket blocking again */
					if (fcntl(self->fd, F_SETFL, args) < 0) {
						logerr("failed to remove socket non-blocking "
								"mode: %s\n", strerror(errno));
						close(self->fd);
						self->fd = -1;
						continue;
					}

					/* disable Nagle's algorithm, issue #208 */
					args = 1;
					if (setsockopt(self->fd, IPPROTO_TCP, TCP_NODELAY,
								&args, sizeof(args)) != 0)
						; /* ignore */

#ifdef TCP_USER_TIMEOUT
					/* break out of connections when no ACK is being
					 * received for +- 10 seconds instead of
					 * retransmitting for +- 15 minutes available on
					 * linux >= 2.6.37
					 * the 10 seconds is in line with the SO_SNDTIMEO
					 * set on the socket below */
					args = 10000 + (rand() % 300);
					if (setsockopt(self->fd, IPPROTO_TCP, TCP_USER_TIMEOUT,
								&args, sizeof(args)) != 0)
						; /* ignore */
#endif

					/* if we reached up here, we're good to go, so don't
					 * continue with the other addrinfos */
					break;
				}
				/* if this didn't resolve to anything, treat as failure */
				if (self->saddr == NULL)
					__sync_add_and_fetch(&(self->failure), 1);
				/* all available addrinfos failed on us */
				if (self->fd < 0)
					continue;
			}

			/* ensure we will break out of connections being stuck more
			 * quickly than the kernel would give up */
			timeout.tv_sec = 10;
			timeout.tv_usec = (rand() % 300) * 1000;
			if (setsockopt(self->fd, SOL_SOCKET, SO_SNDTIMEO,
						&timeout, sizeof(timeout)) != 0)
				; /* ignore */
			if (self->sockbufsize > 0)
				if (setsockopt(self->fd, SOL_SOCKET, SO_SNDBUF,
							&self->sockbufsize, sizeof(self->sockbufsize)) != 0)
					; /* ignore */
#ifdef SO_NOSIGPIPE
			if (self->ctype == CON_TCP || self->ctype == CON_UDP) {
				int enable = 1;
				if (setsockopt(self->fd, SOL_SOCKET, SO_NOSIGPIPE,
							(void *)&enable, sizeof(enable)) != 0)
					logout("warning: failed to ignore SIGPIPE on socket: %s\n",
							strerror(errno));
			}
#endif

#ifdef HAVE_GZIP
			if ((self->transport & 0xFFFF) == W_GZIP) {
				self->strm->hdl.gz = malloc(sizeof(z_stream));
				if (self->strm->hdl.gz != NULL) {
					self->strm->hdl.gz->zalloc = Z_NULL;
					self->strm->hdl.gz->zfree = Z_NULL;
					self->strm->hdl.gz->opaque = Z_NULL;
					self->strm->hdl.gz->next_in = Z_NULL;
				}
				if (self->strm->hdl.gz == NULL ||
						deflateInit2(self->strm->hdl.gz,
							Z_DEFAULT_COMPRESSION,
							Z_DEFLATED,
							15 + 16,
							8,
							Z_DEFAULT_STRATEGY) != Z_OK)
				{
					logerr("failed to open gzip stream: out of memory\n");
					close(self->fd);
					self->fd = -1;
					continue;
				}
				self->strm->obuflen = 0;
			}
#endif
#ifdef HAVE_LZ4
			if ((self->transport & 0xFFFF) == W_LZ4) {
				self->strm->obuflen = 0;

				/* get the maximum size that should ever be required and allocate for it */

				self->strm->hdl.z.cbuflen = LZ4F_compressFrameBound(sizeof(self->strm->obuf), NULL);
				if ((self->strm->hdl.z.cbuf = malloc(self->strm->hdl.z.cbuflen)) == NULL) {
					logerr("Failed to allocate %lu bytes for compressed LZ4 data\n", self->strm->hdl.z.cbuflen);
					close(self->fd);
					self->fd = -1;
					continue;
				}
			}
#endif
#ifdef HAVE_SNAPPY
			if ((self->transport & 0xFFFF) == W_SNAPPY) {
				self->strm->obuflen = 0;
			}
#endif

#ifdef HAVE_SSL
			if (self->transport & W_SSL) {
				int rv;
				z_strm *sstrm;

				if ((self->transport & 0xFFFF) == W_PLAIN) {
 					/* just SSL, nothing else */
					sstrm = self->strm;
				} else {
					sstrm = self->strm->nextstrm;
				}

				sstrm->hdl.ssl = SSL_new(sstrm->ctx);
				SSL_set_tlsext_host_name(sstrm->hdl.ssl, self->ip);
				if (SSL_set_fd(sstrm->hdl.ssl, self->fd) == 0) {
					logerr("failed to SSL_set_fd: %s\n",
							ERR_reason_error_string(ERR_get_error()));
					sstrm->strmclose(sstrm);
					self->fd = -1;
					continue;
				}
				if (self->transport & W_MTLS) {
					int ret;

					/* issue #444 */
					ret = SSL_use_certificate_file(sstrm->hdl.ssl,
												   self->mtlspemcert,
												   SSL_FILETYPE_PEM);
					if (ret == 1) {
						ret = SSL_use_PrivateKey_file(sstrm->hdl.ssl,
													  self->mtlspemkey,
													  SSL_FILETYPE_PEM);
					}

					if (ret != 1) {
						logerr("failed to enable mTLS: %s\n",
							   sslerror(sstrm, ret));
						sstrm->strmclose(sstrm);
						self->fd = -1;
						continue;
					}
				}
				if ((rv = SSL_connect(sstrm->hdl.ssl)) != 1) {
					logerr("failed to connect ssl stream: %s\n",
							sslerror(sstrm, rv));
					sstrm->strmclose(sstrm);
					self->fd = -1;
					continue;
				}
				if ((rv = SSL_get_verify_result(sstrm->hdl.ssl)) != X509_V_OK) {
					logerr("failed to verify ssl certificate: %s\n",
							sslerror(sstrm, rv));
					sstrm->strmclose(sstrm);
					self->fd = -1;
					continue;
				}
			} else
#endif
			{
				z_strm *pstrm;

				if (self->transport == W_PLAIN) { /* just plain socket */
					pstrm = self->strm;
				} else {
					pstrm = self->strm->nextstrm;
				}
				pstrm->hdl.sock = self->fd;
			}
		}

		/* send up to batch size */
		len = queue_dequeue_vector(self->batch, self->queue, self->bsize);
		self->batch[len] = NULL;
		metric = self->batch;

		if (len != 0 &&
				__sync_bool_compare_and_swap(&(self->keep_running), 0, 0))
		{
			/* be noisy during shutdown so we can track any slowing down
			 * servers, possibly preventing us to shut down */
			logerr("shutting down %s:%u: waiting for %zu metrics\n",
					self->ip, self->port, len + queue_len(self->queue));
		}

		if (len == 0 && __sync_add_and_fetch(&(self->failure), 0)) {
			/* if we don't have anything to send, we have at least a
			 * connection succeed, so assume the server is up again,
			 * this is in particular important for recovering this
			 * node by probes, to avoid starvation of this server since
			 * its queue is possibly being offloaded to secondaries */
			if (self->ctype != CON_UDP)
				logerr("server %s:%u: OK after probe\n", self->ip, self->port);
			__sync_and_and_fetch(&(self->failure), 0);
		}

		for (; *metric != NULL; metric++) {
			len = *(size_t *)(*metric);
			/* Write to the stream, this may not succeed completely due
			 * to flow control and whatnot, which the docs suggest need
			 * resuming to complete.  So, use a loop, but to avoid
			 * getting endlessly stuck on this, only try a limited
			 * number of times for a single metric. */
			for (cnt = 0, p = *metric + sizeof(size_t); cnt < 10; cnt++) {
				if ((slen = self->strm->strmwrite(self->strm, p, len)) != len) {
					if (slen >= 0) {
						p += slen;
						len -= slen;
					} else if (errno != EINTR) {
						break;
					}
					/* allow the remote to catch up */
					usleep((50 + (rand() % 150)) * 1000);  /* 50ms - 200ms */
				} else {
					break;
				}
			}
			if (slen != len) {
				/* not fully sent (after tries), or failure
				 * close connection regardless so we don't get
				 * synchonisation problems */
				if (self->ctype != CON_UDP &&
						__sync_fetch_and_add(&(self->failure), 1) == 0)
					logerr("failed to write() to %s:%u: %s\n",
							self->ip, self->port,
							(slen < 0 ?
							 self->strm->strmerror(self->strm, slen) :
							 "incomplete write"));
				self->strm->strmclose(self->strm);
				self->fd = -1;
				/* put back stuff we couldn't process */
				for (; *metric != NULL; metric++) {
					if (!queue_putback(self->queue, *metric)) {
						if (mode & MODE_DEBUG)
							logerr("server %s:%u: dropping metric: %s",
									self->ip, self->port,
									*metric + sizeof(size_t));
						free((char *)*metric);
						__sync_add_and_fetch(&(self->dropped), 1);
					}
				}
				break;
			} else if (!__sync_bool_compare_and_swap(&(self->failure), 0, 0)) {
				if (self->ctype != CON_UDP)
					logerr("server %s:%u: OK\n", self->ip, self->port);
				__sync_and_and_fetch(&(self->failure), 0);
			}
			free((char *)*metric);
			__sync_add_and_fetch(&(self->metrics), 1);
		}

		gettimeofday(&stop, NULL);
		__sync_add_and_fetch(&(self->ticks), timediff(start, stop));

		idle = 0;
	}
	__sync_and_and_fetch(&(self->running), 0);

	if (self->fd >= 0)
		self->strm->strmclose(self->strm);
	if (secpos != NULL)
		free(secpos);
	return NULL;
}

/**
 * Allocate a new (outbound) server.  Effectively this means a thread
 * that reads from the queue and sends this as good as it can to the ip
 * address and port associated.
 */
server *
server_new(
		const char *ip,
		unsigned short port,
		con_type type,
		con_trnsp transport,
		char *mtlspemcert,
		char *mtlspemkey,
		con_proto ctype,
		struct addrinfo *saddr,
		struct addrinfo *hint,
		size_t qsize,
		size_t bsize,
		int maxstalls,
		unsigned short iotimeout,
		unsigned int sockbufsize)
{
	server *ret;

	if ((ret = malloc(sizeof(server))) == NULL)
		return NULL;

	ret->type = type;
	ret->transport = transport;
	ret->mtlspemcert = mtlspemcert == NULL ? NULL : strdup(mtlspemcert);
	ret->mtlspemkey = mtlspemkey == NULL ? NULL : strdup(mtlspemkey);
	ret->ctype = ctype;
	ret->tid = 0;
	ret->secondaries = NULL;
	ret->secondariescnt = 0;
	ret->ip = strdup(ip);
	if (ret->ip == NULL) {
		free(ret);
		return NULL;
	}
	ret->port = port;
	ret->instance = NULL;
	ret->bsize = bsize;
	ret->iotimeout = iotimeout < 250 ? 600 : iotimeout;
	ret->sockbufsize = sockbufsize;
	ret->maxstalls = maxstalls;
	if ((ret->batch = malloc(sizeof(char *) * (bsize + 1))) == NULL) {
		free((char *)ret->ip);
		free(ret);
		return NULL;
	}
	ret->fd = -1;
	if ((ret->strm = malloc(sizeof(z_strm))) == NULL) {
		free((char *)ret->ip);
		free(ret->batch);
		free(ret);
		return NULL;
	}
	ret->strm->nextstrm = NULL;

	/* setup normal or SSL-wrapped socket first */
#ifdef HAVE_SSL
	if (transport & W_SSL) {
		/* create an auto-negotiate context */
		const SSL_METHOD *m = SSLv23_client_method();
		ret->strm->ctx = SSL_CTX_new(m);

		if (sslCA != NULL) {
			if (ret->strm->ctx == NULL ||
				SSL_CTX_load_verify_locations(ret->strm->ctx,
											   sslCAisdir ? NULL : sslCA,
											   sslCAisdir ? sslCA : NULL) == 0)
			{
				char *err = ERR_error_string(ERR_get_error(), NULL);
				logerr("failed to load SSL verify locations from %s for "
						"%s:%d: %s\n", sslCA, ret->ip, ret->port, err);
				free((char *)ret->ip);
				free(ret->batch);
				free(ret->strm);
				free(ret);
				return NULL;
			}
		}
		SSL_CTX_set_verify(ret->strm->ctx, SSL_VERIFY_PEER, NULL);

		ret->strm->strmwrite = &sslwrite;
		ret->strm->strmflush = &sslflush;
		ret->strm->strmclose = &sslclose;
		ret->strm->strmerror = &sslerror;
	} else
#endif
	{
		ret->strm->strmwrite = &sockwrite;
		ret->strm->strmflush = &sockflush;
		ret->strm->strmclose = &sockclose;
		ret->strm->strmerror = &sockerror;
	}
	
	/* now see if we have a compressor defined */
	if ((transport & 0xFFFF) == W_PLAIN) {
		/* catch noop */
	}
#ifdef HAVE_GZIP
	else if ((transport & 0xFFFF) == W_GZIP) {
		z_strm *gzstrm = malloc(sizeof(z_strm));
		if (gzstrm == NULL) {
			free((char *)ret->ip);
			free(ret->batch);
			free(ret->strm);
			free(ret);
			return NULL;
		}
		gzstrm->strmwrite = &gzipwrite;
		gzstrm->strmflush = &gzipflush;
		gzstrm->strmclose = &gzipclose;
		gzstrm->strmerror = &gziperror;
		gzstrm->nextstrm = ret->strm;
		ret->strm = gzstrm;
	}
#endif
#ifdef HAVE_LZ4
	else if ((transport & 0xFFFF) == W_LZ4) {
		z_strm *lzstrm = malloc(sizeof(z_strm));
		if (lzstrm == NULL) {
			free((char *)ret->ip);
			free(ret->batch);
			free(ret->strm);
			free(ret);
			return NULL;
		}
		lzstrm->strmwrite = &lzwrite;
		lzstrm->strmflush = &lzflush;
		lzstrm->strmclose = &lzclose;
		lzstrm->strmerror = &lzerror;
		lzstrm->nextstrm = ret->strm;
		ret->strm = lzstrm;
	}
#endif
#ifdef HAVE_SNAPPY
	else if ((transport & 0xFFFF) == W_SNAPPY) {
		z_strm *snpstrm = malloc(sizeof(z_strm));
		if (snpstrm == NULL) {
			free((char *)ret->ip);
			free(ret->batch);
			free(ret->strm);
			free(ret);
			return NULL;
		}
		snpstrm->strmwrite = &snappywrite;
		snpstrm->strmflush = &snappyflush;
		snpstrm->strmclose = &snappyclose;
		snpstrm->strmerror = &snappyerror;
		snpstrm->nextstrm = ret->strm;
		ret->strm = snpstrm;
	}
#endif

	ret->saddr = saddr;
	ret->reresolve = 0;
	ret->hint = NULL;
	if (hint != NULL) {
		ret->reresolve = 1;
		ret->hint = hint;
	}
	ret->queue = queue_new(qsize);
	if (ret->queue == NULL) {
		free(ret->batch);
		free((char *)ret->ip);
		free(ret);
		return NULL;
	}

	ret->failover = 0;
	ret->failure = 0;
	ret->running = 0;
	ret->keep_running = 1;
	ret->stallseq = 0;
	ret->metrics = 0;
	ret->dropped = 0;
	ret->stalls = 0;
	ret->ticks = 0;
	ret->prevmetrics = 0;
	ret->prevdropped = 0;
	ret->prevstalls = 0;
	ret->prevticks = 0;
	ret->tid = 0;

	return ret;
}

/**
 * Compare server s against the address info in saddr.  A server is
 * considered to be equal is it is of the same socket family, type and
 * protocol, and if the target address and port are the same.  When
 * saddr is NULL, a match against the given ip is attempted, e.g. for
 * file destinations.
 */
char
server_cmp(server *s, struct addrinfo *saddr, const char *ip)
{
	if ((saddr == NULL || s->saddr == NULL)) {
		if (strcmp(s->ip, ip) == 0)
			return 0;
	} else if (
			s->saddr->ai_family == saddr->ai_family &&
			s->saddr->ai_socktype == saddr->ai_socktype &&
			s->saddr->ai_protocol == saddr->ai_protocol
	   )
	{
		if (saddr->ai_family == AF_INET) {
			struct sockaddr_in *l = ((struct sockaddr_in *)s->saddr->ai_addr);
			struct sockaddr_in *r = ((struct sockaddr_in *)saddr->ai_addr);
			if (l->sin_port == r->sin_port &&
					l->sin_addr.s_addr == r->sin_addr.s_addr)
				return 0;
		} else if (saddr->ai_family == AF_INET6) {
			struct sockaddr_in6 *l =
				((struct sockaddr_in6 *)s->saddr->ai_addr);
			struct sockaddr_in6 *r =
				((struct sockaddr_in6 *)saddr->ai_addr);
			if (l->sin6_port == r->sin6_port &&
					memcmp(l->sin6_addr.s6_addr, r->sin6_addr.s6_addr,
						sizeof(l->sin6_addr.s6_addr)) == 0)
				return 0;
		}
	}

	return 1;  /* not equal */
}

/**
 * Starts a previously created server using server_new().  Returns
 * errno if starting a thread failed, after which the caller should
 * server_free() the given s pointer.
 */
char
server_start(server *s)
{
	return pthread_create(&s->tid, NULL, &server_queuereader, s);
}

/**
 * Adds a list of secondary servers to this server.  A secondary server
 * is a server which' queue will be checked when this server has nothing
 * to do.
 */
void
server_add_secondaries(server *self, server **secondaries, size_t count)
{
	self->secondaries = secondaries;
	self->secondariescnt = count;
}

/**
 * Flags this server as part of a failover cluster, which means the
 * secondaries are used only to offload on failure, not on queue stress.
 */
void
server_set_failover(server *self)
{
	self->failover = 1;
}

/**
 * Sets instance name only used for carbon_ch cluster type.
 */
void
server_set_instance(server *self, char *instance)
{
	if (self->instance != NULL)
		free(self->instance);
	self->instance = strdup(instance);
}

/**
 * Thin wrapper around the associated queue with the server object.
 * Returns true if the metric could be queued for sending, or the metric
 * was dropped because the associated server is down.  Returns false
 * otherwise (when a retry seems like it could succeed shortly).
 */
inline char
server_send(server *s, const char *d, char force)
{
	if (queue_free(s->queue) == 0) {
		char failure = __sync_add_and_fetch(&(s->failure), 0);
		if (!force && s->secondariescnt > 0) {
			size_t i;
			/* don't immediately drop if we know there are others that
			 * back us up */
			for (i = 0; i < s->secondariescnt; i++) {
				if (!__sync_add_and_fetch(&(s->secondaries[i]->failure), 0)) {
					failure = 0;
					break;
				}
			}
		}
		if (failure || force ||
				__sync_add_and_fetch(&(s->stallseq), 0) == s->maxstalls)
		{
			__sync_add_and_fetch(&(s->dropped), 1);
			/* excess event will be dropped by the enqueue below */
		} else {
			__sync_add_and_fetch(&(s->stallseq), 1);
			__sync_add_and_fetch(&(s->stalls), 1);
			return 0;
		}
	} else {
		__sync_and_and_fetch(&(s->stallseq), 0);
	}
	queue_enqueue(s->queue, d);

	return 1;
}

/**
 * Tells this server to finish sending pending items from its queue.
 */
void
server_shutdown(server *s)
{
	int i;
	size_t failures;
	size_t inqueue;

	/* this function should only be called on a running server */
	if (__sync_bool_compare_and_swap(&(s->keep_running), 0, 0))
		return;

	if (s->secondariescnt > 0) {
		/* if we have a working connection, or we still have stuff in
		 * our queue, wait for our secondaries, as they might need us,
		 * or we need them */
		do {
			failures = 0;
			inqueue = 0;
			for (i = 0; i < s->secondariescnt; i++) {
				if (__sync_add_and_fetch(&(s->secondaries[i]->failure), 0))
					failures++;
				if (__sync_add_and_fetch(&(s->secondaries[i]->running), 0))
					inqueue += queue_len(s->secondaries[i]->queue);
			}
			/* loop until we all failed, or nothing is in the queues */
		} while (failures != s->secondariescnt &&
				inqueue != 0 &&
				logout("any_of cluster pending %zu metrics "
					"(with %zu failed nodes)\n", inqueue, failures) >= -1 &&
				usleep((200 + (rand() % 100)) * 1000) <= 0);
		/* shut down entire cluster */
		for (i = 0; i < s->secondariescnt; i++)
			__sync_bool_compare_and_swap(
					&(s->secondaries[i]->keep_running), 1, 0);
		/* to pretend to be dead for above loop (just in case) */
		if (inqueue != 0)
			for (i = 0; i < s->secondariescnt; i++)
				__sync_add_and_fetch(&(s->secondaries[i]->failure), 1);
		/* wait for the secondaries to be stopped so we surely don't get
		 * invalid reads when server_free is called */
		for (i = 0; i < s->secondariescnt; i++) {
			while (__sync_add_and_fetch(&(s->secondaries[i]->running), 0))
				usleep((200 + (rand() % 100)) * 1000);
		}
	}

	__sync_bool_compare_and_swap(&(s->keep_running), 1, 0);
}

/**
 * Frees this server and associated resources.  This includes joining
 * the server thread.
 */
void
server_free(server *s) {
	int err;

	if (s->tid != 0 && (err = pthread_join(s->tid, NULL)) != 0)
		logerr("%s:%u: failed to join server thread: %s\n",
				s->ip, s->port, strerror(err));
	s->tid = 0;

	if (s->ctype == CON_TCP) {
		size_t qlen = queue_len(s->queue);
		if (qlen > 0)
			logerr("dropping %zu metrics for %s:%u\n",
					qlen, s->ip, s->port);
	}

	queue_destroy(s->queue);
#ifdef HAVE_SSL
	if (s->transport & W_SSL)
		SSL_CTX_free(s->strm->ctx);
#endif
	free(s->batch);
	if (s->instance)
		free(s->instance);
	if (s->saddr != NULL)
		freeaddrinfo(s->saddr);
	if (s->hint)
		free(s->hint);
	if (s->mtlspemcert)
		free(s->mtlspemcert);
	if (s->mtlspemkey)
		free(s->mtlspemkey);
	free((char *)s->ip);
	if (s->strm->nextstrm != NULL)
		free(s->strm->nextstrm);
	free(s->strm);
	s->ip = NULL;
	free(s);
}

/**
 * Swaps the queue between server l to r.  This is assumes both l and r
 * are not running.
 */
void
server_swap_queue(server *l, server *r)
{
	queue *t;

	assert(l->keep_running == 0 || l->tid == 0);
	assert(r->keep_running == 0 || r->tid == 0);

	t = l->queue;
	l->queue = r->queue;
	r->queue = t;
	
	/* swap associated statistics as well */
	l->metrics = r->metrics;
	l->dropped = r->dropped;
	l->stalls = r->stalls;
	l->ticks = r->ticks;
	l->prevmetrics = r->prevmetrics;
	l->prevdropped = r->prevdropped;
	l->prevstalls = r->prevstalls;
	l->prevticks = r->prevticks;
}

/**
 * Returns the ip address this server points to.
 */
inline const char *
server_ip(server *s)
{
	if (s == NULL)
		return NULL;
	return s->ip;
}

/**
 * Returns the port this server connects at.
 */
inline unsigned short
server_port(server *s)
{
	if (s == NULL)
		return 0;
	return s->port;
}

/**
 * Returns the instance associated with this server.
 */
inline char *
server_instance(server *s)
{
	return s->instance;
}

/**
 * Returns the connection protocol of this server.
 */
inline con_proto
server_ctype(server *s)
{
	if (s == NULL)
		return CON_PIPE;
	return s->ctype;
}

/**
 * Returns the connection transport of this server.
 */
inline con_trnsp
server_transport(server *s)
{
	if (s == NULL)
		return W_PLAIN;
	return s->transport;
}

/**
 * Returns the connection type of this server.
 */
inline con_type
server_type(server *s)
{
	if (s == NULL)
		return T_LINEMODE;
	return s->type;
}

/**
 * Returns whether the last action on this server caused a failure.
 */
inline char
server_failed(server *s)
{
	if (s == NULL)
		return 0;
	return __sync_add_and_fetch(&(s->failure), 0);
}

/**
 * Returns the wall-clock time in microseconds (us) consumed sending metrics.
 */
inline size_t
server_get_ticks(server *s)
{
	if (s == NULL)
		return 0;
	return __sync_add_and_fetch(&(s->ticks), 0);
}

/**
 * Returns the wall-clock time in microseconds (us) consumed since last
 * call to this function.
 */
inline size_t
server_get_ticks_sub(server *s)
{
	size_t d;
	if (s == NULL)
		return 0;
	d = __sync_add_and_fetch(&(s->ticks), 0) - s->prevticks;
	s->prevticks += d;
	return d;
}

/**
 * Returns the number of metrics sent since start.
 */
inline size_t
server_get_metrics(server *s)
{
	if (s == NULL)
		return 0;
	return __sync_add_and_fetch(&(s->metrics), 0);
}

/**
 * Returns the number of metrics sent since last call to this function.
 */
inline size_t
server_get_metrics_sub(server *s)
{
	size_t d;
	if (s == NULL)
		return 0;
	d = __sync_add_and_fetch(&(s->metrics), 0) - s->prevmetrics;
	s->prevmetrics += d;
	return d;
}

/**
 * Returns the number of metrics dropped since start.
 */
inline size_t
server_get_dropped(server *s)
{
	if (s == NULL)
		return 0;
	return __sync_add_and_fetch(&(s->dropped), 0);
}

/**
 * Returns the number of metrics dropped since last call to this function.
 */
inline size_t
server_get_dropped_sub(server *s)
{
	size_t d;
	if (s == NULL)
		return 0;
	d = __sync_add_and_fetch(&(s->dropped), 0) - s->prevdropped;
	s->prevdropped += d;
	return d;
}

/**
 * Returns the number of stalls since start.  A stall happens when the
 * queue is full, but it appears as if it would be a good idea to wait
 * for a brief period and retry.
 */
inline size_t
server_get_stalls(server *s)
{
	if (s == NULL)
		return 0;
	return __sync_add_and_fetch(&(s->stalls), 0);
}

/**
 * Returns the number of stalls since last call to this function.
 */
inline size_t
server_get_stalls_sub(server *s)
{
	size_t d;
	if (s == NULL)
		return 0;
	d = __sync_add_and_fetch(&(s->stalls), 0) - s->prevstalls;
	s->prevstalls += d;
	return d;
}

/**
 * Returns the (approximate) number of metrics waiting to be sent.
 */
inline size_t
server_get_queue_len(server *s)
{
	if (s == NULL)
		return 0;
	return queue_len(s->queue);
}

/**
 * Returns the allocated size of the queue backing metrics waiting to be
 * sent.
 */
inline size_t
server_get_queue_size(server *s)
{
	if (s == NULL)
		return 0;
	return queue_size(s->queue);
}

<MSG> server_start: fix return

Allow errno to be retrieved for more informational error messages on
failure.  Use server_start explicitly rather than implicitly in
router_readconfig.

<DFF> @@ -501,17 +501,14 @@ server_new(
 }
 
 /**
- * Starts a previously created server using server_new().  Returns false
- * if starting a thread failed, after which the caller should
+ * Starts a previously created server using server_new().  Returns
+ * errno if starting a thread failed, after which the caller should
  * server_free() the given s pointer.
  */
 char
 server_start(server *s)
 {
-	if (pthread_create(&s->tid, NULL, &server_queuereader, s) != 0)
-		return 0;
-	
-	return 1;
+	return pthread_create(&s->tid, NULL, &server_queuereader, s);
 }
 
 /**

{'commit': 'diff --git a/relay.c b/relay.c\nindex 1d3c787..2a73ecb 100644\n--- a/relay.c\n+++ b/relay.c\n@@ -671,7 +671,7 @@ main(int argc, char * const argv[])\n \t\tlogerr("failed to create internal submission queue, shutting down\\n");\n \t\tkeep_running = 0;\n \t}\n-\tif (server_start(internal_submission) == 0) {\n+\tif (server_start(internal_submission) != 0) {\n \t\tlogerr("failed to start submission queue thread, shutting down\\n");\n \t\tserver_free(internal_submission);\n \t\tkeep_running = 0;\n@@ -688,7 +688,14 @@ main(int argc, char * const argv[])\n \tlogout("starting statistics collector\\n");\n \tcollector_start(&workers[1], rtr, internal_submission, smode == CUM);\n \n-\tlogout("startup sequence complete\\n");\n+\tlogout("starting servers\\n");\n+\tif (router_start(rtr) != 0) {\n+\t\t/* router_start logerrs itself */\n+\t\tkeep_running = 0;\n+\t}\n+\n+\tif (keep_running == 0)\n+\t\tlogout("startup sequence complete\\n");\n \n \t/* workers do the work, just wait */\n \twhile (keep_running)\ndiff --git a/router.c b/router.c\nindex e42829c..7e26516 100644\n--- a/router.c\n+++ b/router.c\n@@ -810,7 +810,7 @@ router_readconfig(router *orig,\n \t\t\t\t\t\t\t\t*proto == \'u\' ? CON_UDP : CON_TCP,\n \t\t\t\t\t\t\t\twalk == (void *)1 ? NULL : walk,\n \t\t\t\t\t\t\t\tqueuesize, batchsize, iotimeout);\n-\t\t\t\t\t\tif (newserver == NULL || server_start(newserver) == 0) {\n+\t\t\t\t\t\tif (newserver == NULL) {\n \t\t\t\t\t\t\tlogerr("failed to add server %s:%d (%s) "\n \t\t\t\t\t\t\t\t\t"to cluster %s: %s\\n", ip, port, proto,\n \t\t\t\t\t\t\t\t\tname, strerror(errno));\ndiff --git a/server.c b/server.c\nindex 6c7fc9e..021d057 100644\n--- a/server.c\n+++ b/server.c\n@@ -501,17 +501,14 @@ server_new(\n }\n \n /**\n- * Starts a previously created server using server_new().  Returns false\n- * if starting a thread failed, after which the caller should\n+ * Starts a previously created server using server_new().  Returns\n+ * errno if starting a thread failed, after which the caller should\n  * server_free() the given s pointer.\n  */\n char\n server_start(server *s)\n {\n-\tif (pthread_create(&s->tid, NULL, &server_queuereader, s) != 0)\n-\t\treturn 0;\n-\t\n-\treturn 1;\n+\treturn pthread_create(&s->tid, NULL, &server_queuereader, s);\n }\n \n /**\n', 'commit_subject': 'server_start: fix return', 'repo_name': 'grobian/carbon-c-relay', 'committer': {'name': 'Fabian Groffen', 'email': '2c4829061e3ecacc1da315323749610b9ffaef65@gentoo.org', 'time_sec': '1461214180', 'tz_offset': '120', 'date': {'seconds': '1461214180'}}, 'commit_tree': '8fde003607be87db5572d2ed8866ff063e528128', 'commit_message': 'server_start: fix return\n\nAllow errno to be retrieved for more informational error messages on\nfailure.  Use server_start explicitly rather than implicitly in\nrouter_readconfig.\n', 'language_name': 'Shell', 'license': 'apache-2.0', 'hunks': '@@ -501,17 +501,14 @@ server_new(\n }\n \n /**\n- * Starts a previously created server using server_new().  Returns false\n- * if starting a thread failed, after which the caller should\n+ * Starts a previously created server using server_new().  Returns\n+ * errno if starting a thread failed, after which the caller should\n  * server_free() the given s pointer.\n  */\n char\n server_start(server *s)\n {\n-\tif (pthread_create(&s->tid, NULL, &server_queuereader, s) != 0)\n-\t\treturn 0;\n-\t\n-\treturn 1;\n+\treturn pthread_create(&s->tid, NULL, &server_queuereader, s);\n }\n \n /**\n', 'addition_count': 3, 'deletion_count': 6, 'src_file': 'a/server.c', 'tgt_file': 'b/server.c', 'file_extension': '.c', 'before_file': '/*\n * Copyright 2013-2022 Fabian Groffen\n *\n * Licensed under the Apache License, Version 2.0 (the "License");\n * you may not use this file except in compliance with the License.\n * You may obtain a copy of the License at\n *\n *     http://www.apache.org/licenses/LICENSE-2.0\n *\n * Unless required by applicable law or agreed to in writing, software\n * distributed under the License is distributed on an "AS IS" BASIS,\n * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n * See the License for the specific language governing permissions and\n * limitations under the License.\n */\n\n\n#include <stdio.h>\n#include <unistd.h>\n#include <string.h>\n#include <fcntl.h>\n#include <pthread.h>\n#include <poll.h>\n#include <errno.h>\n#include <sys/time.h>\n#include <sys/types.h>\n#include <sys/socket.h>\n#include <sys/stat.h>\n#include <netdb.h>\n#include <netinet/in.h>\n#include <netinet/tcp.h>\n#include <arpa/inet.h>\n#include <sys/resource.h>\n#include <assert.h>\n\n#include "relay.h"\n#include "queue.h"\n#include "dispatcher.h"\n#include "collector.h"\n#include "server.h"\n\n#ifdef HAVE_GZIP\n#include <zlib.h>\n#endif\n#ifdef HAVE_LZ4\n#include <lz4.h>\n#include <lz4frame.h>\n#endif\n#ifdef HAVE_SNAPPY\n#include <snappy-c.h>\n#endif\n#ifdef HAVE_SSL\n#include <openssl/ssl.h>\n#include <openssl/err.h>\n#endif\n\ntypedef struct _z_strm {\n\tssize_t (*strmwrite)(struct _z_strm *, const void *, size_t);\n\tint (*strmflush)(struct _z_strm *);\n\tint (*strmclose)(struct _z_strm *);\n\tconst char *(*strmerror)(struct _z_strm *, int);     /* get last err str */\n\tstruct _z_strm *nextstrm;                            /* set when chained */\n#if defined(HAVE_LZ4) || defined(HAVE_SNAPPY) || defined(HAVE_GZIP)\n\tchar obuf[METRIC_BUFSIZ];\n\tint obuflen;\n#endif\n#ifdef HAVE_SSL\n\tSSL_CTX *ctx;\n#endif\n\tunion {\n#ifdef HAVE_GZIP\n\t\tz_streamp gz;\n#endif\n#ifdef HAVE_LZ4\n\t\tstruct {\n\t\t\tvoid *cbuf;\n\t\t\tsize_t cbuflen;\n\t\t} z;\n#endif\n#ifdef HAVE_SSL\n\t\tSSL *ssl;\n#endif\n\t\tint sock;\n\t} hdl;\n} z_strm;\n\nstruct _server {\n\tconst char *ip;\n\tunsigned short port;\n\tchar *instance;\n\tstruct addrinfo *saddr;\n\tstruct addrinfo *hint;\n\tchar reresolve:1;\n\tint fd;\n\tz_strm *strm;\n\tqueue *queue;\n\tsize_t bsize;\n\tshort iotimeout;\n\tunsigned int sockbufsize;\n\tunsigned char maxstalls:SERVER_STALL_BITS;\n\tconst char **batch;\n\tcon_type type;\n\tcon_trnsp transport;\n\tchar *mtlspemcert;\n\tchar *mtlspemkey;\n\tcon_proto ctype;\n\tpthread_t tid;\n\tstruct _server **secondaries;\n\tsize_t secondariescnt;\n\tchar failover:1;\n\tchar failure;       /* full byte for atomic access */\n\tchar running;       /* full byte for atomic access */\n\tchar keep_running;  /* full byte for atomic access */\n\tunsigned char stallseq;  /* full byte for atomic access */\n\tsize_t metrics;\n\tsize_t dropped;\n\tsize_t stalls;\n\tsize_t ticks;\n\tsize_t prevmetrics;\n\tsize_t prevdropped;\n\tsize_t prevstalls;\n\tsize_t prevticks;\n};\n\n\n/* connection specific writers and closers */\n\n/* ordinary socket */\nstatic inline ssize_t\nsockwrite(z_strm *strm, const void *buf, size_t sze)\n{\n\treturn write(strm->hdl.sock, buf, sze);\n}\n\nstatic inline int\nsockflush(z_strm *strm)\n{\n\t/* noop, we don\'t use a stream in the normal case */\n\t(void)strm;\n\treturn 0;\n}\n\nstatic inline int\nsockclose(z_strm *strm)\n{\n\treturn close(strm->hdl.sock);\n}\n\nstatic inline const char *\nsockerror(z_strm *strm, int rval)\n{\n\t(void)strm;\n\t(void)rval;\n\treturn strerror(errno);\n}\n\n#ifdef HAVE_GZIP\n/* gzip wrapped socket */\nstatic inline int gzipflush(z_strm *strm);\n\nstatic inline ssize_t\ngzipwrite(z_strm *strm, const void *buf, size_t sze)\n{\n\t/* ensure we have space available */\n\tif (strm->obuflen + sze > METRIC_BUFSIZ)\n\t\tif (gzipflush(strm) != 0)\n\t\t\treturn -1;\n\n\t/* append metric to buf */\n\tmemcpy(strm->obuf + strm->obuflen, buf, sze);\n\tstrm->obuflen += sze;\n\n\treturn sze;\n}\n\nstatic inline int\ngzipflush(z_strm *strm)\n{\n\tint cret;\n\tchar cbuf[METRIC_BUFSIZ];\n\tsize_t cbuflen;\n\tchar *cbufp;\n\tint oret;\n\n\tstrm->hdl.gz->next_in = (Bytef *)strm->obuf;\n\tstrm->hdl.gz->avail_in = strm->obuflen;\n\tstrm->hdl.gz->next_out = (Bytef *)cbuf;\n\tstrm->hdl.gz->avail_out = sizeof(cbuf);\n\n\tdo {\n\t\tcret = deflate(strm->hdl.gz, Z_PARTIAL_FLUSH);\n\t\tif (cret == Z_OK && strm->hdl.gz->avail_out == 0) {\n\t\t\t/* too large output block, unlikely given the input, discard */\n\t\t\t/* TODO */\n\t\t\tbreak;\n\t\t}\n\t\tif (cret == Z_OK) {\n\t\t\tcbufp = cbuf;\n\t\t\tcbuflen = sizeof(cbuf) - strm->hdl.gz->avail_out;\n\t\t\twhile (cbuflen > 0) {\n\t\t\t\toret = strm->nextstrm->strmwrite(strm->nextstrm,\n\t\t\t\t\t\tcbufp, cbuflen);\n\t\t\t\tif (oret < 0)\n\t\t\t\t\treturn -1;  /* failure is failure */\n\n\t\t\t\t/* update counters to possibly retry the remaining bit */\n\t\t\t\tcbufp += oret;\n\t\t\t\tcbuflen -= oret;\n\t\t\t}\n\n\t\t\tif (strm->hdl.gz->avail_in == 0)\n\t\t\t\tbreak;\n\n\t\t\tstrm->hdl.gz->next_out = (Bytef *)cbuf;\n\t\t\tstrm->hdl.gz->avail_out = sizeof(cbuf);\n\t\t} else {\n\t\t\t/* discard */\n\t\t\tbreak;\n\t\t}\n\t} while (1);\n\t/* flush whatever we wrote */\n\tstrm->nextstrm->strmflush(strm->nextstrm);\n\n\t/* reset the write position, from this point it will always need to\n\t * restart */\n\tstrm->obuflen = 0;\n\n\tif (cret != Z_OK)\n\t\treturn -1;  /* we must reset/free gzip */\n\n\treturn 0;\n}\n\nstatic inline int\ngzipclose(z_strm *strm)\n{\n\tint ret = strm->nextstrm->strmclose(strm->nextstrm);\n\tdeflateEnd(strm->hdl.gz);\n\tfree(strm->hdl.gz);\n\treturn ret;\n}\n\nstatic inline const char *\ngziperror(z_strm *strm, int rval)\n{\n\treturn strm->nextstrm->strmerror(strm->nextstrm, rval);\n}\n#endif\n\n#ifdef HAVE_LZ4\n/* lz4 wrapped socket */\n\nstatic inline int lzflush(z_strm *strm);\n\nstatic inline ssize_t\nlzwrite(z_strm *strm, const void *buf, size_t sze)\n{\n\tsize_t towrite = sze;\n\n\t/* use the same strategy as gzip: fill the buffer until space\n\t   runs out. we completely fill the output buffer before flushing */\n\n\twhile (towrite > 0) {\n\n\t\tsize_t avail = METRIC_BUFSIZ - strm->obuflen;\n\t\tsize_t copysize = towrite > avail ? avail : towrite;\n\n\t\t/* copy into the output buffer as much as we can */\n\n\t\tif (copysize > 0) {\n\t\t\tmemcpy(strm->obuf + strm->obuflen, buf, copysize);\n\t\t\tstrm->obuflen += copysize;\n\t\t\ttowrite -= copysize;\n\t\t\tbuf += copysize;\n\t\t}\n\n\t\t/* if output buffer is full & still have bytes to write, flush now */\n\n\t\tif (strm->obuflen == METRIC_BUFSIZ && towrite > 0 && lzflush(strm) != 0) {\n\t\t\tlogerr("Failed to flush LZ4 data to make space\\n");\n\t\t\treturn -1;\n\t\t}\n\t}\n\n\treturn sze;\n}\n\nstatic inline int\nlzflush(z_strm *strm)\n{\n\tint oret;\n\tsize_t ret;\n\n\t/* anything to do? */\n\n\tif (strm->obuflen == 0)\n\t\treturn 0;\n\n\t/* the buffered data goes out as a single frame */\n\n\tret = LZ4F_compressFrame(strm->hdl.z.cbuf, strm->hdl.z.cbuflen, strm->obuf, strm->obuflen, NULL);\n\tif (LZ4F_isError(ret)) {\n\t\tlogerr("Failed to compress %d LZ4 bytes into a frame: %d\\n",\n\t\t       strm->obuflen, (int)ret);\n\t\treturn -1;\n\t}\n\t\t/* write and flush */\n\n\tif ((oret = strm->nextstrm->strmwrite(strm->nextstrm,\n\t\t\t\t\t      strm->hdl.z.cbuf, ret)) < 0) {\n\t\tlogerr("Failed to write %lu bytes of compressed LZ4 data: %d\\n",\n\t\t       ret, oret);\n\t\treturn oret;\n\t}\n\n\tstrm->nextstrm->strmflush(strm->nextstrm);\n\n\t/* the buffer is gone. reset the write position */\n\n\tstrm->obuflen = 0;\n\treturn 0;\n}\n\nstatic inline int\nlzclose(z_strm *strm)\n{\n\tlzflush(strm);\n\tfree(strm->hdl.z.cbuf);\n\treturn strm->nextstrm->strmclose(strm->nextstrm);\n}\n\nstatic inline const char *\nlzerror(z_strm *strm, int rval)\n{\n\treturn strm->nextstrm->strmerror(strm->nextstrm, rval);\n}\n#endif\n\n#ifdef HAVE_SNAPPY\n/* snappy wrapped socket */\nstatic inline int snappyflush(z_strm *strm);\n\nstatic inline ssize_t\nsnappywrite(z_strm *strm, const void *buf, size_t sze)\n{\n\t/* ensure we have space available */\n\tif (strm->obuflen + sze > METRIC_BUFSIZ)\n\t\tif (snappyflush(strm) != 0)\n\t\t\treturn -1;\n\n\t/* append metric to buf */\n\tmemcpy(strm->obuf + strm->obuflen, buf, sze);\n\tstrm->obuflen += sze;\n\n\treturn sze;\n}\n\nstatic inline int\nsnappyflush(z_strm *strm)\n{\n\tint cret;\n\tchar cbuf[METRIC_BUFSIZ];\n\tsize_t cbuflen = sizeof(cbuf);\n\tchar *cbufp = cbuf;\n\tint oret;\n\n\tcret = snappy_compress(strm->obuf, strm->obuflen, cbuf, &cbuflen);\n\t/* reset the write position, from this point it will always need to\n\t * restart */\n\tstrm->obuflen = 0;\n\n\tif (cret != SNAPPY_OK)\n\t\treturn -1;  /* we must reset/free snappy */\n\twhile (cbuflen > 0) {\n\t\toret = strm->nextstrm->strmwrite(strm->nextstrm, cbufp, cbuflen);\n\t\tif (oret < 0)\n\t\t\treturn -1;  /* failure is failure */\n\t\tstrm->nextstrm->strmflush(strm->nextstrm);\n\t\t/* update counters to possibly retry the remaining bit */\n\t\tcbufp += oret;\n\t\tcbuflen -= oret;\n\t}\n\n\treturn 0;\n}\n\nstatic inline int\nsnappyclose(z_strm *strm)\n{\n\tint ret = strm->nextstrm->strmclose(strm->nextstrm);\n\treturn ret;\n}\n\nstatic inline const char *\nsnappyerror(z_strm *strm, int rval)\n{\n\treturn strm->nextstrm->strmerror(strm->nextstrm, rval);\n}\n#endif\n\n#ifdef HAVE_SSL\n/* (Open|Libre)SSL wrapped socket */\nstatic inline ssize_t\nsslwrite(z_strm *strm, const void *buf, size_t sze)\n{\n\treturn (ssize_t)SSL_write(strm->hdl.ssl, buf, (int)sze);\n}\n\nstatic inline int\nsslflush(z_strm *strm)\n{\n\t/* noop */\n\t(void)strm;\n\treturn 0;\n}\n\nstatic inline int\nsslclose(z_strm *strm)\n{\n\tint sock = SSL_get_fd(strm->hdl.ssl);\n\tSSL_free(strm->hdl.ssl);\n\treturn close(sock);\n}\n\nstatic char _sslerror_buf[256];\nstatic inline const char *\nsslerror(z_strm *strm, int rval)\n{\n\tint err = SSL_get_error(strm->hdl.ssl, rval);\n\tswitch (err) {\n\t\tcase SSL_ERROR_NONE:\n\t\t\tsnprintf(_sslerror_buf, sizeof(_sslerror_buf),\n\t\t\t\t\t"%d: SSL_ERROR_NONE", err);\n\t\t\tbreak;\n\t\tcase SSL_ERROR_ZERO_RETURN:\n\t\t\tsnprintf(_sslerror_buf, sizeof(_sslerror_buf),\n\t\t\t\t\t"%d: TLS/SSL connection has been closed", err);\n\t\t\tbreak;\n\t\tcase SSL_ERROR_WANT_READ:\n\t\tcase SSL_ERROR_WANT_WRITE:\n\t\tcase SSL_ERROR_WANT_CONNECT:\n\t\tcase SSL_ERROR_WANT_ACCEPT:\n\t\t\tsnprintf(_sslerror_buf, sizeof(_sslerror_buf),\n\t\t\t\t\t"%d: the read or write operation did not complete", err);\n\t\t\tbreak;\n\t\tcase SSL_ERROR_WANT_X509_LOOKUP:\n\t\t\tsnprintf(_sslerror_buf, sizeof(_sslerror_buf),\n\t\t\t\t\t"%d: call callback via SSL_CTX_set_client_cert_cb()", err);\n\t\t\tbreak;\n#ifdef SSL_ERROR_WANT_ASYNC\n\t\tcase SSL_ERROR_WANT_ASYNC:\n\t\t\tsnprintf(_sslerror_buf, sizeof(_sslerror_buf),\n\t\t\t\t\t"%d: asynchronous engine is still processing data", err);\n\t\t\tbreak;\n#endif\n#ifdef SSL_ERROR_WANT_ASYNC_JOB\n\t\tcase SSL_ERROR_WANT_ASYNC_JOB:\n\t\t\tsnprintf(_sslerror_buf, sizeof(_sslerror_buf),\n\t\t\t\t\t"%d: no async jobs available in the pool", err);\n\t\t\tbreak;\n#endif\n\t\tcase SSL_ERROR_SYSCALL:\n\t\t\tsnprintf(_sslerror_buf, sizeof(_sslerror_buf),\n\t\t\t\t\t"%d: I/O error: %s", err, strerror(errno));\n\t\t\tbreak;\n\t\tcase SSL_ERROR_SSL:\n\t\t\tERR_error_string_n(ERR_get_error(),\n\t\t\t\t\t_sslerror_buf, sizeof(_sslerror_buf));\n\t\t\tbreak;\n\t}\n\treturn _sslerror_buf;\n}\n#endif\n\n/**\n * Reads from the queue and sends items to the remote server.  This\n * function is designed to be a thread.  Data sending is attempted to be\n * batched, but sent one by one to reduce loss on sending failure.\n * A connection with the server is maintained for as long as there is\n * data to be written.  As soon as there is none, the connection is\n * dropped if a timeout of DISCONNECT_WAIT_TIME exceeds.\n */\nstatic void *\nserver_queuereader(void *d)\n{\n\tserver *self = (server *)d;\n\tsize_t len;\n\tssize_t slen;\n\tconst char **metric = self->batch;\n\tstruct timeval start, stop;\n\tstruct timeval timeout;\n\tqueue *squeue;\n\tchar idle = 0;\n\tsize_t *secpos = NULL;\n\tunsigned char cnt;\n\tconst char *p;\n\n\t*metric = NULL;\n\n#define FAIL_WAIT_TIME   6  /* 6 * 250ms = 1.5s */\n}\n\n/**\n * Starts a previously created server using server_new().  Returns false\n * if starting a thread failed, after which the caller should\n * server_free() the given s pointer.\n */\nchar\nserver_start(server *s)\n{\n\tif (pthread_create(&s->tid, NULL, &server_queuereader, s) != 0)\n\t\treturn 0;\n\t\n\treturn 1;\n}\n\n/**\n\t\t\t}\n\t\t\tif (idle == 1)\n\t\t\t\t/* ensure blocks are pushed out as soon as we\'re idling,\n\t\t\t\t * this allows compressors to benefit from a larger\n\t\t\t\t * stream of data to gain better compression */\n\t\t\t\tself->strm->strmflush(self->strm);\n\t\t\tgettimeofday(&stop, NULL);\n\t\t\t__sync_add_and_fetch(&(self->ticks), timediff(start, stop));\n\t\t\tif (__sync_bool_compare_and_swap(&(self->keep_running), 0, 0))\n\t\t\t\tbreak;\n\t\t\t/* nothing to do, so slow down for a bit */\n\t\t\tusleep((200 + (rand() % 100)) * 1000);  /* 200ms - 300ms */\n\t\t\t/* if we are in failure mode, keep checking if we can\n\t\t\t * connect, this avoids unnecessary queue moves */\n\t\t\tif (__sync_bool_compare_and_swap(&(self->failure), 0, 0))\n\t\t\t\t/* it makes no sense to try and do something, so skip */\n\t\t\t\tcontinue;\n\t\t} else if (self->secondariescnt > 0 &&\n\t\t\t\t(__sync_add_and_fetch(&(self->failure), 0) >= FAIL_WAIT_TIME ||\n\t\t\t\t (!self->failover && LEN_CRITICAL(self->queue))))\n\t\t{\n\t\t\tsize_t i;\n\n\t\t\tgettimeofday(&start, NULL);\n\t\t\tif (self->secondariescnt > 0) {\n\t\t\t\tif (secpos == NULL) {\n\t\t\t\t\tsecpos = malloc(sizeof(size_t) * self->secondariescnt);\n\t\t\t\t\tif (secpos == NULL) {\n\t\t\t\t\t\tlogerr("server: failed to allocate memory "\n\t\t\t\t\t\t\t\t"for secpos\\n");\n\t\t\t\t\t\tgettimeofday(&stop, NULL);\n\t\t\t\t\t\t__sync_add_and_fetch(&(self->ticks),\n\t\t\t\t\t\t\t\ttimediff(start, stop));\n\t\t\t\t\t\tcontinue;\n\t\t\t\t\t}\n\t\t\t\t\tfor (i = 0; i < self->secondariescnt; i++)\n\t\t\t\t\t\tsecpos[i] = i;\n\t\t\t\t}\n\t\t\t\tif (!self->failover) {\n\t\t\t\t\t/* randomise the failover list such that in the\n\t\t\t\t\t * grand scheme of things we don\'t punish the first\n\t\t\t\t\t * working server in the list to deal with all\n\t\t\t\t\t * traffic meant for a now failing server */\n\t\t\t\t\tfor (i = 0; i < self->secondariescnt; i++) {\n\t\t\t\t\t\tsize_t n = rand() % (self->secondariescnt - i);\n\t\t\t\t\t\tif (n != i) {\n\t\t\t\t\t\t\tsize_t t = secpos[n];\n\t\t\t\t\t\t\tsecpos[n] = secpos[i];\n\t\t\t\t\t\t\tsecpos[i] = t;\n\t\t\t\t\t\t}\n\t\t\t\t\t}\n\t\t\t\t}\n\t\t\t}\n\n\t\t\t/* offload data from our queue to our secondaries\n\t\t\t * when doing so, observe the following:\n\t\t\t * - avoid nodes that are in failure mode\n\t\t\t * - avoid nodes which queues are >= critical_len\n\t\t\t * when no nodes remain given the above\n\t\t\t * - send to nodes which queue size < critical_len\n\t\t\t * where there are no such nodes\n\t\t\t * - do nothing (we will overflow, since we can\'t send\n\t\t\t *   anywhere) */\n\t\t\t*metric = NULL;\n\t\t\tsqueue = NULL;\n\t\t\tfor (i = 0; i < self->secondariescnt; i++) {\n\t\t\t\t/* both conditions below make sure we skip ourself */\n\t\t\t\tif (__sync_add_and_fetch(\n\t\t\t\t\t\t\t&(self->secondaries[secpos[i]]->failure), 0))\n\t\t\t\t\tcontinue;\n\t\t\t\tsqueue = self->secondaries[secpos[i]]->queue;\n\t\t\t\tif (!self->failover && LEN_CRITICAL(squeue)) {\n\t\t\t\t\tsqueue = NULL;\n\t\t\t\t\tcontinue;\n\t\t\t\t}\n\t\t\t\tif (*metric == NULL) {\n\t\t\t\t\t/* send up to batch size of our queue to this queue */\n\t\t\t\t\tlen = queue_dequeue_vector(\n\t\t\t\t\t\t\tself->batch, self->queue, self->bsize);\n\t\t\t\t\tself->batch[len] = NULL;\n\t\t\t\t\tmetric = self->batch;\n\t\t\t\t}\n\n\t\t\t\tfor (; *metric != NULL; metric++)\n\t\t\t\t\tif (!queue_putback(squeue, *metric))\n\t\t\t\t\t\tbreak;\n\t\t\t\t/* try to put back stuff that didn\'t fit */\n\t\t\t\tfor (; *metric != NULL; metric++)\n\t\t\t\t\tif (!queue_putback(self->queue, *metric))\n\t\t\t\t\t\tbreak;\n\t\t\t}\n\t\t\tfor (; *metric != NULL; metric++) {\n\t\t\t\tif (mode & MODE_DEBUG)\n\t\t\t\t\tlogerr("dropping metric: %s", *metric);\n\t\t\t\tfree((char *)*metric);\n\t\t\t\t__sync_add_and_fetch(&(self->dropped), 1);\n\t\t\t}\n\t\t\tgettimeofday(&stop, NULL);\n\t\t\t__sync_add_and_fetch(&(self->ticks),\n\t\t\t\t\ttimediff(start, stop));\n\t\t\tif (squeue == NULL) {\n\t\t\t\t/* we couldn\'t do anything, take it easy for a bit */\n\t\t\t\tif (__sync_add_and_fetch(&(self->failure), 0) > 1) {\n\t\t\t\t\t/* This is a compound because I can\'t seem to figure\n\t\t\t\t\t * out how to atomically just "set" a variable.\n\t\t\t\t\t * It\'s not bad when in the middle there is a ++,\n\t\t\t\t\t * all that counts is that afterwards its > 0. */\n\t\t\t\t\t__sync_and_and_fetch(&(self->failure), 0);\n\t\t\t\t\t__sync_add_and_fetch(&(self->failure), 1);\n\t\t\t\t}\n\t\t\t\tif (__sync_bool_compare_and_swap(&(self->keep_running), 0, 0))\n\t\t\t\t\tbreak;\n\t\t\t\tusleep((200 + (rand() % 100)) * 1000);  /* 200ms - 300ms */\n\t\t\t}\n\t\t} else if (__sync_add_and_fetch(&(self->failure), 0) > 0) {\n\t\t\tif (__sync_bool_compare_and_swap(&(self->keep_running), 0, 0))\n\t\t\t\tbreak;\n\t\t\tusleep((200 + (rand() % 100)) * 1000);  /* 200ms - 300ms */\n\t\t\t/* avoid overflowing */\n\t\t\tif (__sync_add_and_fetch(&(self->failure), 0) > FAIL_WAIT_TIME)\n\t\t\t\t__sync_sub_and_fetch(&(self->failure), 1);\n\t\t}\n\n\t\t/* at this point we\'ve got work to do, if we\'re instructed to\n\t\t * shut down, however, try to get everything out of the door\n\t\t * (until we fail, see top of this loop) */\n\n\t\tgettimeofday(&start, NULL);\n\n\t\t/* try to connect */\n\t\tif (self->fd < 0) {\n\t\t\tif (self->reresolve) {  /* can only be CON_UDP/CON_TCP */\n\t\t\t\tstruct addrinfo *saddr;\n\t\t\t\tchar sport[8];\n\n\t\t\t\t/* re-lookup the address info, if it fails, stay with\n\t\t\t\t * whatever we have such that resolution errors incurred\n\t\t\t\t * after starting the relay won\'t make it fail */\n\t\t\t\tfreeaddrinfo(self->saddr);\n\t\t\t\tsnprintf(sport, sizeof(sport), "%u", self->port);\n\t\t\t\tif (getaddrinfo(self->ip, sport, self->hint, &saddr) == 0) {\n\t\t\t\t\tself->saddr = saddr;\n\t\t\t\t} else {\n\t\t\t\t\tif (__sync_fetch_and_add(&(self->failure), 1) == 0)\n\t\t\t\t\t\tlogerr("failed to resolve %s:%u, server unavailable\\n",\n\t\t\t\t\t\t\t\tself->ip, self->port);\n\t\t\t\t\tself->saddr = NULL;\n\t\t\t\t\t/* this will break out below */\n\t\t\t\t}\n\t\t\t}\n\n\t\t\tif (self->ctype == CON_PIPE) {\n\t\t\t\tint intconn[2];\n\t\t\t\tif (pipe(intconn) < 0) {\n\t\t\t\t\tif (__sync_fetch_and_add(&(self->failure), 1) == 0)\n\t\t\t\t\t\tlogerr("failed to create pipe: %s\\n", strerror(errno));\n\t\t\t\t\tcontinue;\n\t\t\t\t}\n\t\t\t\tdispatch_addconnection(intconn[0], NULL);\n\t\t\t\tself->fd = intconn[1];\n\t\t\t} else if (self->ctype == CON_FILE) {\n\t\t\t\tif ((self->fd = open(self->ip,\n\t\t\t\t\t\t\t\tO_WRONLY | O_APPEND | O_CREAT,\n\t\t\t\t\t\t\t\tS_IRUSR | S_IWUSR | S_IRGRP | S_IROTH)) < 0)\n\t\t\t\t{\n\t\t\t\t\tif (__sync_fetch_and_add(&(self->failure), 1) == 0)\n\t\t\t\t\t\tlogerr("failed to open file \'%s\': %s\\n",\n\t\t\t\t\t\t\t\tself->ip, strerror(errno));\n\t\t\t\t\tcontinue;\n\t\t\t\t}\n\t\t\t} else if (self->ctype == CON_UDP) {\n\t\t\t\tstruct addrinfo *walk;\n\n\t\t\t\tfor (walk = self->saddr; walk != NULL; walk = walk->ai_next) {\n\t\t\t\t\tif ((self->fd = socket(walk->ai_family,\n\t\t\t\t\t\t\t\t\twalk->ai_socktype,\n\t\t\t\t\t\t\t\t\twalk->ai_protocol)) < 0)\n\t\t\t\t\t{\n\t\t\t\t\t\tif (walk->ai_next == NULL &&\n\t\t\t\t\t\t\t\t__sync_fetch_and_add(&(self->failure), 1) == 0)\n\t\t\t\t\t\t\tlogerr("failed to create udp socket: %s\\n",\n\t\t\t\t\t\t\t\t\tstrerror(errno));\n\t\t\t\t\t\tcontinue;\n\t\t\t\t\t}\n\t\t\t\t\tif (connect(self->fd, walk->ai_addr, walk->ai_addrlen) < 0)\n\t\t\t\t\t{\n\t\t\t\t\t\tif (walk->ai_next == NULL &&\n\t\t\t\t\t\t\t\t__sync_fetch_and_add(&(self->failure), 1) == 0)\n\t\t\t\t\t\t\tlogerr("failed to connect udp socket: %s\\n",\n\t\t\t\t\t\t\t\t\tstrerror(errno));\n\t\t\t\t\t\tclose(self->fd);\n\t\t\t\t\t\tself->fd = -1;\n\t\t\t\t\t\tcontinue;\n\t\t\t\t\t}\n\n\t\t\t\t\t/* we made it up here, so this connection is usable */\n\t\t\t\t\tbreak;\n\t\t\t\t}\n\t\t\t\t/* if this didn\'t resolve to anything, treat as failure */\n\t\t\t\tif (self->saddr == NULL)\n\t\t\t\t\t__sync_add_and_fetch(&(self->failure), 1);\n\t\t\t\t/* if all addrinfos failed, try again later */\n\t\t\t\tif (self->fd < 0)\n\t\t\t\t\tcontinue;\n\t\t\t} else {  /* CON_TCP */\n\t\t\t\tint ret;\n\t\t\t\tint args;\n\t\t\t\tstruct addrinfo *walk;\n\n\t\t\t\tfor (walk = self->saddr; walk != NULL; walk = walk->ai_next) {\n\t\t\t\t\tif ((self->fd = socket(walk->ai_family,\n\t\t\t\t\t\t\t\t\twalk->ai_socktype,\n\t\t\t\t\t\t\t\t\twalk->ai_protocol)) < 0)\n\t\t\t\t\t{\n\t\t\t\t\t\tif (walk->ai_next == NULL &&\n\t\t\t\t\t\t\t\t__sync_fetch_and_add(&(self->failure), 1) == 0)\n\t\t\t\t\t\t\tlogerr("failed to create socket: %s\\n",\n\t\t\t\t\t\t\t\t\tstrerror(errno));\n\t\t\t\t\t\tcontinue;\n\t\t\t\t\t}\n\n\t\t\t\t\t/* put socket in non-blocking mode such that we can\n\t\t\t\t\t * poll() (time-out) on the connect() call */\n\t\t\t\t\targs = fcntl(self->fd, F_GETFL, NULL);\n\t\t\t\t\tif (fcntl(self->fd, F_SETFL, args | O_NONBLOCK) < 0) {\n\t\t\t\t\t\tlogerr("failed to set socket non-blocking mode: %s\\n",\n\t\t\t\t\t\t\t\tstrerror(errno));\n\t\t\t\t\t\tclose(self->fd);\n\t\t\t\t\t\tself->fd = -1;\n\t\t\t\t\t\tcontinue;\n\t\t\t\t\t}\n\t\t\t\t\tret = connect(self->fd, walk->ai_addr, walk->ai_addrlen);\n\n\t\t\t\t\tif (ret < 0 && errno == EINPROGRESS) {\n\t\t\t\t\t\t/* wait for connection to succeed if the OS thinks\n\t\t\t\t\t\t * it can succeed */\n\t\t\t\t\t\tstruct pollfd ufds[1];\n\t\t\t\t\t\tufds[0].fd = self->fd;\n\t\t\t\t\t\tufds[0].events = POLLIN | POLLOUT;\n\t\t\t\t\t\tret = poll(ufds, 1, self->iotimeout + (rand() % 100));\n\t\t\t\t\t\tif (ret == 0) {\n\t\t\t\t\t\t\t/* time limit expired */\n\t\t\t\t\t\t\tif (walk->ai_next == NULL &&\n\t\t\t\t\t\t\t\t\t__sync_fetch_and_add(\n\t\t\t\t\t\t\t\t\t\t&(self->failure), 1) == 0)\n\t\t\t\t\t\t\t\tlogerr("failed to connect() to "\n\t\t\t\t\t\t\t\t\t\t"%s:%u: Operation timed out\\n",\n\t\t\t\t\t\t\t\t\t\tself->ip, self->port);\n\t\t\t\t\t\t\tclose(self->fd);\n\t\t\t\t\t\t\tself->fd = -1;\n\t\t\t\t\t\t\tcontinue;\n\t\t\t\t\t\t} else if (ret < 0) {\n\t\t\t\t\t\t\t/* some select error occurred */\n\t\t\t\t\t\t\tif (walk->ai_next &&\n\t\t\t\t\t\t\t\t\t__sync_fetch_and_add(\n\t\t\t\t\t\t\t\t\t\t&(self->failure), 1) == 0)\n\t\t\t\t\t\t\t\tlogerr("failed to poll() for %s:%u: %s\\n",\n\t\t\t\t\t\t\t\t\t\tself->ip, self->port, strerror(errno));\n\t\t\t\t\t\t\tclose(self->fd);\n\t\t\t\t\t\t\tself->fd = -1;\n\t\t\t\t\t\t\tcontinue;\n\t\t\t\t\t\t} else {\n\t\t\t\t\t\t\tif (ufds[0].revents & POLLHUP) {\n\t\t\t\t\t\t\t\tif (walk->ai_next == NULL &&\n\t\t\t\t\t\t\t\t\t\t__sync_fetch_and_add(\n\t\t\t\t\t\t\t\t\t\t\t&(self->failure), 1) == 0)\n\t\t\t\t\t\t\t\t\tlogerr("failed to connect() for %s:%u: "\n\t\t\t\t\t\t\t\t\t\t\t"Connection refused\\n",\n\t\t\t\t\t\t\t\t\t\t\tself->ip, self->port);\n\t\t\t\t\t\t\t\tclose(self->fd);\n\t\t\t\t\t\t\t\tself->fd = -1;\n\t\t\t\t\t\t\t\tcontinue;\n\t\t\t\t\t\t\t}\n\t\t\t\t\t\t}\n\t\t\t\t\t} else if (ret < 0) {\n\t\t\t\t\t\tif (walk->ai_next == NULL &&\n\t\t\t\t\t\t\t\t__sync_fetch_and_add(&(self->failure), 1) == 0)\n\t\t\t\t\t\t{\n\t\t\t\t\t\t\tlogerr("failed to connect() to %s:%u: %s\\n",\n\t\t\t\t\t\t\t\t\tself->ip, self->port, strerror(errno));\n\t\t\t\t\t\t\tdispatch_check_rlimit_and_warn();\n\t\t\t\t\t\t}\n\t\t\t\t\t\tclose(self->fd);\n\t\t\t\t\t\tself->fd = -1;\n\t\t\t\t\t\tcontinue;\n\t\t\t\t\t}\n\n\t\t\t\t\t/* make socket blocking again */\n\t\t\t\t\tif (fcntl(self->fd, F_SETFL, args) < 0) {\n\t\t\t\t\t\tlogerr("failed to remove socket non-blocking "\n\t\t\t\t\t\t\t\t"mode: %s\\n", strerror(errno));\n\t\t\t\t\t\tclose(self->fd);\n\t\t\t\t\t\tself->fd = -1;\n\t\t\t\t\t\tcontinue;\n\t\t\t\t\t}\n\n\t\t\t\t\t/* disable Nagle\'s algorithm, issue #208 */\n\t\t\t\t\targs = 1;\n\t\t\t\t\tif (setsockopt(self->fd, IPPROTO_TCP, TCP_NODELAY,\n\t\t\t\t\t\t\t\t&args, sizeof(args)) != 0)\n\t\t\t\t\t\t; /* ignore */\n\n#ifdef TCP_USER_TIMEOUT\n\t\t\t\t\t/* break out of connections when no ACK is being\n\t\t\t\t\t * received for +- 10 seconds instead of\n\t\t\t\t\t * retransmitting for +- 15 minutes available on\n\t\t\t\t\t * linux >= 2.6.37\n\t\t\t\t\t * the 10 seconds is in line with the SO_SNDTIMEO\n\t\t\t\t\t * set on the socket below */\n\t\t\t\t\targs = 10000 + (rand() % 300);\n\t\t\t\t\tif (setsockopt(self->fd, IPPROTO_TCP, TCP_USER_TIMEOUT,\n\t\t\t\t\t\t\t\t&args, sizeof(args)) != 0)\n\t\t\t\t\t\t; /* ignore */\n#endif\n\n\t\t\t\t\t/* if we reached up here, we\'re good to go, so don\'t\n\t\t\t\t\t * continue with the other addrinfos */\n\t\t\t\t\tbreak;\n\t\t\t\t}\n\t\t\t\t/* if this didn\'t resolve to anything, treat as failure */\n\t\t\t\tif (self->saddr == NULL)\n\t\t\t\t\t__sync_add_and_fetch(&(self->failure), 1);\n\t\t\t\t/* all available addrinfos failed on us */\n\t\t\t\tif (self->fd < 0)\n\t\t\t\t\tcontinue;\n\t\t\t}\n\n\t\t\t/* ensure we will break out of connections being stuck more\n\t\t\t * quickly than the kernel would give up */\n\t\t\ttimeout.tv_sec = 10;\n\t\t\ttimeout.tv_usec = (rand() % 300) * 1000;\n\t\t\tif (setsockopt(self->fd, SOL_SOCKET, SO_SNDTIMEO,\n\t\t\t\t\t\t&timeout, sizeof(timeout)) != 0)\n\t\t\t\t; /* ignore */\n\t\t\tif (self->sockbufsize > 0)\n\t\t\t\tif (setsockopt(self->fd, SOL_SOCKET, SO_SNDBUF,\n\t\t\t\t\t\t\t&self->sockbufsize, sizeof(self->sockbufsize)) != 0)\n\t\t\t\t\t; /* ignore */\n#ifdef SO_NOSIGPIPE\n\t\t\tif (self->ctype == CON_TCP || self->ctype == CON_UDP) {\n\t\t\t\tint enable = 1;\n\t\t\t\tif (setsockopt(self->fd, SOL_SOCKET, SO_NOSIGPIPE,\n\t\t\t\t\t\t\t(void *)&enable, sizeof(enable)) != 0)\n\t\t\t\t\tlogout("warning: failed to ignore SIGPIPE on socket: %s\\n",\n\t\t\t\t\t\t\tstrerror(errno));\n\t\t\t}\n#endif\n\n#ifdef HAVE_GZIP\n\t\t\tif ((self->transport & 0xFFFF) == W_GZIP) {\n\t\t\t\tself->strm->hdl.gz = malloc(sizeof(z_stream));\n\t\t\t\tif (self->strm->hdl.gz != NULL) {\n\t\t\t\t\tself->strm->hdl.gz->zalloc = Z_NULL;\n\t\t\t\t\tself->strm->hdl.gz->zfree = Z_NULL;\n\t\t\t\t\tself->strm->hdl.gz->opaque = Z_NULL;\n\t\t\t\t\tself->strm->hdl.gz->next_in = Z_NULL;\n\t\t\t\t}\n\t\t\t\tif (self->strm->hdl.gz == NULL ||\n\t\t\t\t\t\tdeflateInit2(self->strm->hdl.gz,\n\t\t\t\t\t\t\tZ_DEFAULT_COMPRESSION,\n\t\t\t\t\t\t\tZ_DEFLATED,\n\t\t\t\t\t\t\t15 + 16,\n\t\t\t\t\t\t\t8,\n\t\t\t\t\t\t\tZ_DEFAULT_STRATEGY) != Z_OK)\n\t\t\t\t{\n\t\t\t\t\tlogerr("failed to open gzip stream: out of memory\\n");\n\t\t\t\t\tclose(self->fd);\n\t\t\t\t\tself->fd = -1;\n\t\t\t\t\tcontinue;\n\t\t\t\t}\n\t\t\t\tself->strm->obuflen = 0;\n\t\t\t}\n#endif\n#ifdef HAVE_LZ4\n\t\t\tif ((self->transport & 0xFFFF) == W_LZ4) {\n\t\t\t\tself->strm->obuflen = 0;\n\n\t\t\t\t/* get the maximum size that should ever be required and allocate for it */\n\n\t\t\t\tself->strm->hdl.z.cbuflen = LZ4F_compressFrameBound(sizeof(self->strm->obuf), NULL);\n\t\t\t\tif ((self->strm->hdl.z.cbuf = malloc(self->strm->hdl.z.cbuflen)) == NULL) {\n\t\t\t\t\tlogerr("Failed to allocate %lu bytes for compressed LZ4 data\\n", self->strm->hdl.z.cbuflen);\n\t\t\t\t\tclose(self->fd);\n\t\t\t\t\tself->fd = -1;\n\t\t\t\t\tcontinue;\n\t\t\t\t}\n\t\t\t}\n#endif\n#ifdef HAVE_SNAPPY\n\t\t\tif ((self->transport & 0xFFFF) == W_SNAPPY) {\n\t\t\t\tself->strm->obuflen = 0;\n\t\t\t}\n#endif\n\n#ifdef HAVE_SSL\n\t\t\tif (self->transport & W_SSL) {\n\t\t\t\tint rv;\n\t\t\t\tz_strm *sstrm;\n\n\t\t\t\tif ((self->transport & 0xFFFF) == W_PLAIN) {\n \t\t\t\t\t/* just SSL, nothing else */\n\t\t\t\t\tsstrm = self->strm;\n\t\t\t\t} else {\n\t\t\t\t\tsstrm = self->strm->nextstrm;\n\t\t\t\t}\n\n\t\t\t\tsstrm->hdl.ssl = SSL_new(sstrm->ctx);\n\t\t\t\tSSL_set_tlsext_host_name(sstrm->hdl.ssl, self->ip);\n\t\t\t\tif (SSL_set_fd(sstrm->hdl.ssl, self->fd) == 0) {\n\t\t\t\t\tlogerr("failed to SSL_set_fd: %s\\n",\n\t\t\t\t\t\t\tERR_reason_error_string(ERR_get_error()));\n\t\t\t\t\tsstrm->strmclose(sstrm);\n\t\t\t\t\tself->fd = -1;\n\t\t\t\t\tcontinue;\n\t\t\t\t}\n\t\t\t\tif (self->transport & W_MTLS) {\n\t\t\t\t\tint ret;\n\n\t\t\t\t\t/* issue #444 */\n\t\t\t\t\tret = SSL_use_certificate_file(sstrm->hdl.ssl,\n\t\t\t\t\t\t\t\t\t\t\t\t   self->mtlspemcert,\n\t\t\t\t\t\t\t\t\t\t\t\t   SSL_FILETYPE_PEM);\n\t\t\t\t\tif (ret == 1) {\n\t\t\t\t\t\tret = SSL_use_PrivateKey_file(sstrm->hdl.ssl,\n\t\t\t\t\t\t\t\t\t\t\t\t\t  self->mtlspemkey,\n\t\t\t\t\t\t\t\t\t\t\t\t\t  SSL_FILETYPE_PEM);\n\t\t\t\t\t}\n\n\t\t\t\t\tif (ret != 1) {\n\t\t\t\t\t\tlogerr("failed to enable mTLS: %s\\n",\n\t\t\t\t\t\t\t   sslerror(sstrm, ret));\n\t\t\t\t\t\tsstrm->strmclose(sstrm);\n\t\t\t\t\t\tself->fd = -1;\n\t\t\t\t\t\tcontinue;\n\t\t\t\t\t}\n\t\t\t\t}\n\t\t\t\tif ((rv = SSL_connect(sstrm->hdl.ssl)) != 1) {\n\t\t\t\t\tlogerr("failed to connect ssl stream: %s\\n",\n\t\t\t\t\t\t\tsslerror(sstrm, rv));\n\t\t\t\t\tsstrm->strmclose(sstrm);\n\t\t\t\t\tself->fd = -1;\n\t\t\t\t\tcontinue;\n\t\t\t\t}\n\t\t\t\tif ((rv = SSL_get_verify_result(sstrm->hdl.ssl)) != X509_V_OK) {\n\t\t\t\t\tlogerr("failed to verify ssl certificate: %s\\n",\n\t\t\t\t\t\t\tsslerror(sstrm, rv));\n\t\t\t\t\tsstrm->strmclose(sstrm);\n\t\t\t\t\tself->fd = -1;\n\t\t\t\t\tcontinue;\n\t\t\t\t}\n\t\t\t} else\n#endif\n\t\t\t{\n\t\t\t\tz_strm *pstrm;\n\n\t\t\t\tif (self->transport == W_PLAIN) { /* just plain socket */\n\t\t\t\t\tpstrm = self->strm;\n\t\t\t\t} else {\n\t\t\t\t\tpstrm = self->strm->nextstrm;\n\t\t\t\t}\n\t\t\t\tpstrm->hdl.sock = self->fd;\n\t\t\t}\n\t\t}\n\n\t\t/* send up to batch size */\n\t\tlen = queue_dequeue_vector(self->batch, self->queue, self->bsize);\n\t\tself->batch[len] = NULL;\n\t\tmetric = self->batch;\n\n\t\tif (len != 0 &&\n\t\t\t\t__sync_bool_compare_and_swap(&(self->keep_running), 0, 0))\n\t\t{\n\t\t\t/* be noisy during shutdown so we can track any slowing down\n\t\t\t * servers, possibly preventing us to shut down */\n\t\t\tlogerr("shutting down %s:%u: waiting for %zu metrics\\n",\n\t\t\t\t\tself->ip, self->port, len + queue_len(self->queue));\n\t\t}\n\n\t\tif (len == 0 && __sync_add_and_fetch(&(self->failure), 0)) {\n\t\t\t/* if we don\'t have anything to send, we have at least a\n\t\t\t * connection succeed, so assume the server is up again,\n\t\t\t * this is in particular important for recovering this\n\t\t\t * node by probes, to avoid starvation of this server since\n\t\t\t * its queue is possibly being offloaded to secondaries */\n\t\t\tif (self->ctype != CON_UDP)\n\t\t\t\tlogerr("server %s:%u: OK after probe\\n", self->ip, self->port);\n\t\t\t__sync_and_and_fetch(&(self->failure), 0);\n\t\t}\n\n\t\tfor (; *metric != NULL; metric++) {\n\t\t\tlen = *(size_t *)(*metric);\n\t\t\t/* Write to the stream, this may not succeed completely due\n\t\t\t * to flow control and whatnot, which the docs suggest need\n\t\t\t * resuming to complete.  So, use a loop, but to avoid\n\t\t\t * getting endlessly stuck on this, only try a limited\n\t\t\t * number of times for a single metric. */\n\t\t\tfor (cnt = 0, p = *metric + sizeof(size_t); cnt < 10; cnt++) {\n\t\t\t\tif ((slen = self->strm->strmwrite(self->strm, p, len)) != len) {\n\t\t\t\t\tif (slen >= 0) {\n\t\t\t\t\t\tp += slen;\n\t\t\t\t\t\tlen -= slen;\n\t\t\t\t\t} else if (errno != EINTR) {\n\t\t\t\t\t\tbreak;\n\t\t\t\t\t}\n\t\t\t\t\t/* allow the remote to catch up */\n\t\t\t\t\tusleep((50 + (rand() % 150)) * 1000);  /* 50ms - 200ms */\n\t\t\t\t} else {\n\t\t\t\t\tbreak;\n\t\t\t\t}\n\t\t\t}\n\t\t\tif (slen != len) {\n\t\t\t\t/* not fully sent (after tries), or failure\n\t\t\t\t * close connection regardless so we don\'t get\n\t\t\t\t * synchonisation problems */\n\t\t\t\tif (self->ctype != CON_UDP &&\n\t\t\t\t\t\t__sync_fetch_and_add(&(self->failure), 1) == 0)\n\t\t\t\t\tlogerr("failed to write() to %s:%u: %s\\n",\n\t\t\t\t\t\t\tself->ip, self->port,\n\t\t\t\t\t\t\t(slen < 0 ?\n\t\t\t\t\t\t\t self->strm->strmerror(self->strm, slen) :\n\t\t\t\t\t\t\t "incomplete write"));\n\t\t\t\tself->strm->strmclose(self->strm);\n\t\t\t\tself->fd = -1;\n\t\t\t\t/* put back stuff we couldn\'t process */\n\t\t\t\tfor (; *metric != NULL; metric++) {\n\t\t\t\t\tif (!queue_putback(self->queue, *metric)) {\n\t\t\t\t\t\tif (mode & MODE_DEBUG)\n\t\t\t\t\t\t\tlogerr("server %s:%u: dropping metric: %s",\n\t\t\t\t\t\t\t\t\tself->ip, self->port,\n\t\t\t\t\t\t\t\t\t*metric + sizeof(size_t));\n\t\t\t\t\t\tfree((char *)*metric);\n\t\t\t\t\t\t__sync_add_and_fetch(&(self->dropped), 1);\n\t\t\t\t\t}\n\t\t\t\t}\n\t\t\t\tbreak;\n\t\t\t} else if (!__sync_bool_compare_and_swap(&(self->failure), 0, 0)) {\n\t\t\t\tif (self->ctype != CON_UDP)\n\t\t\t\t\tlogerr("server %s:%u: OK\\n", self->ip, self->port);\n\t\t\t\t__sync_and_and_fetch(&(self->failure), 0);\n\t\t\t}\n\t\t\tfree((char *)*metric);\n\t\t\t__sync_add_and_fetch(&(self->metrics), 1);\n\t\t}\n\n\t\tgettimeofday(&stop, NULL);\n\t\t__sync_add_and_fetch(&(self->ticks), timediff(start, stop));\n\n\t\tidle = 0;\n\t}\n\t__sync_and_and_fetch(&(self->running), 0);\n\n\tif (self->fd >= 0)\n\t\tself->strm->strmclose(self->strm);\n\tif (secpos != NULL)\n\t\tfree(secpos);\n\treturn NULL;\n}\n\n/**\n * Allocate a new (outbound) server.  Effectively this means a thread\n * that reads from the queue and sends this as good as it can to the ip\n * address and port associated.\n */\nserver *\nserver_new(\n\t\tconst char *ip,\n\t\tunsigned short port,\n\t\tcon_type type,\n\t\tcon_trnsp transport,\n\t\tchar *mtlspemcert,\n\t\tchar *mtlspemkey,\n\t\tcon_proto ctype,\n\t\tstruct addrinfo *saddr,\n\t\tstruct addrinfo *hint,\n\t\tsize_t qsize,\n\t\tsize_t bsize,\n\t\tint maxstalls,\n\t\tunsigned short iotimeout,\n\t\tunsigned int sockbufsize)\n{\n\tserver *ret;\n\n\tif ((ret = malloc(sizeof(server))) == NULL)\n\t\treturn NULL;\n\n\tret->type = type;\n\tret->transport = transport;\n\tret->mtlspemcert = mtlspemcert == NULL ? NULL : strdup(mtlspemcert);\n\tret->mtlspemkey = mtlspemkey == NULL ? NULL : strdup(mtlspemkey);\n\tret->ctype = ctype;\n\tret->tid = 0;\n\tret->secondaries = NULL;\n\tret->secondariescnt = 0;\n\tret->ip = strdup(ip);\n\tif (ret->ip == NULL) {\n\t\tfree(ret);\n\t\treturn NULL;\n\t}\n\tret->port = port;\n\tret->instance = NULL;\n\tret->bsize = bsize;\n\tret->iotimeout = iotimeout < 250 ? 600 : iotimeout;\n\tret->sockbufsize = sockbufsize;\n\tret->maxstalls = maxstalls;\n\tif ((ret->batch = malloc(sizeof(char *) * (bsize + 1))) == NULL) {\n\t\tfree((char *)ret->ip);\n\t\tfree(ret);\n\t\treturn NULL;\n\t}\n\tret->fd = -1;\n\tif ((ret->strm = malloc(sizeof(z_strm))) == NULL) {\n\t\tfree((char *)ret->ip);\n\t\tfree(ret->batch);\n\t\tfree(ret);\n\t\treturn NULL;\n\t}\n\tret->strm->nextstrm = NULL;\n\n\t/* setup normal or SSL-wrapped socket first */\n#ifdef HAVE_SSL\n\tif (transport & W_SSL) {\n\t\t/* create an auto-negotiate context */\n\t\tconst SSL_METHOD *m = SSLv23_client_method();\n\t\tret->strm->ctx = SSL_CTX_new(m);\n\n\t\tif (sslCA != NULL) {\n\t\t\tif (ret->strm->ctx == NULL ||\n\t\t\t\tSSL_CTX_load_verify_locations(ret->strm->ctx,\n\t\t\t\t\t\t\t\t\t\t\t   sslCAisdir ? NULL : sslCA,\n\t\t\t\t\t\t\t\t\t\t\t   sslCAisdir ? sslCA : NULL) == 0)\n\t\t\t{\n\t\t\t\tchar *err = ERR_error_string(ERR_get_error(), NULL);\n\t\t\t\tlogerr("failed to load SSL verify locations from %s for "\n\t\t\t\t\t\t"%s:%d: %s\\n", sslCA, ret->ip, ret->port, err);\n\t\t\t\tfree((char *)ret->ip);\n\t\t\t\tfree(ret->batch);\n\t\t\t\tfree(ret->strm);\n\t\t\t\tfree(ret);\n\t\t\t\treturn NULL;\n\t\t\t}\n\t\t}\n\t\tSSL_CTX_set_verify(ret->strm->ctx, SSL_VERIFY_PEER, NULL);\n\n\t\tret->strm->strmwrite = &sslwrite;\n\t\tret->strm->strmflush = &sslflush;\n\t\tret->strm->strmclose = &sslclose;\n\t\tret->strm->strmerror = &sslerror;\n\t} else\n#endif\n\t{\n\t\tret->strm->strmwrite = &sockwrite;\n\t\tret->strm->strmflush = &sockflush;\n\t\tret->strm->strmclose = &sockclose;\n\t\tret->strm->strmerror = &sockerror;\n\t}\n\t\n\t/* now see if we have a compressor defined */\n\tif ((transport & 0xFFFF) == W_PLAIN) {\n\t\t/* catch noop */\n\t}\n#ifdef HAVE_GZIP\n\telse if ((transport & 0xFFFF) == W_GZIP) {\n\t\tz_strm *gzstrm = malloc(sizeof(z_strm));\n\t\tif (gzstrm == NULL) {\n\t\t\tfree((char *)ret->ip);\n\t\t\tfree(ret->batch);\n\t\t\tfree(ret->strm);\n\t\t\tfree(ret);\n\t\t\treturn NULL;\n\t\t}\n\t\tgzstrm->strmwrite = &gzipwrite;\n\t\tgzstrm->strmflush = &gzipflush;\n\t\tgzstrm->strmclose = &gzipclose;\n\t\tgzstrm->strmerror = &gziperror;\n\t\tgzstrm->nextstrm = ret->strm;\n\t\tret->strm = gzstrm;\n\t}\n#endif\n#ifdef HAVE_LZ4\n\telse if ((transport & 0xFFFF) == W_LZ4) {\n\t\tz_strm *lzstrm = malloc(sizeof(z_strm));\n\t\tif (lzstrm == NULL) {\n\t\t\tfree((char *)ret->ip);\n\t\t\tfree(ret->batch);\n\t\t\tfree(ret->strm);\n\t\t\tfree(ret);\n\t\t\treturn NULL;\n\t\t}\n\t\tlzstrm->strmwrite = &lzwrite;\n\t\tlzstrm->strmflush = &lzflush;\n\t\tlzstrm->strmclose = &lzclose;\n\t\tlzstrm->strmerror = &lzerror;\n\t\tlzstrm->nextstrm = ret->strm;\n\t\tret->strm = lzstrm;\n\t}\n#endif\n#ifdef HAVE_SNAPPY\n\telse if ((transport & 0xFFFF) == W_SNAPPY) {\n\t\tz_strm *snpstrm = malloc(sizeof(z_strm));\n\t\tif (snpstrm == NULL) {\n\t\t\tfree((char *)ret->ip);\n\t\t\tfree(ret->batch);\n\t\t\tfree(ret->strm);\n\t\t\tfree(ret);\n\t\t\treturn NULL;\n\t\t}\n\t\tsnpstrm->strmwrite = &snappywrite;\n\t\tsnpstrm->strmflush = &snappyflush;\n\t\tsnpstrm->strmclose = &snappyclose;\n\t\tsnpstrm->strmerror = &snappyerror;\n\t\tsnpstrm->nextstrm = ret->strm;\n\t\tret->strm = snpstrm;\n\t}\n#endif\n\n\tret->saddr = saddr;\n\tret->reresolve = 0;\n\tret->hint = NULL;\n\tif (hint != NULL) {\n\t\tret->reresolve = 1;\n\t\tret->hint = hint;\n\t}\n\tret->queue = queue_new(qsize);\n\tif (ret->queue == NULL) {\n\t\tfree(ret->batch);\n\t\tfree((char *)ret->ip);\n\t\tfree(ret);\n\t\treturn NULL;\n\t}\n\n\tret->failover = 0;\n\tret->failure = 0;\n\tret->running = 0;\n\tret->keep_running = 1;\n\tret->stallseq = 0;\n\tret->metrics = 0;\n\tret->dropped = 0;\n\tret->stalls = 0;\n\tret->ticks = 0;\n\tret->prevmetrics = 0;\n\tret->prevdropped = 0;\n\tret->prevstalls = 0;\n\tret->prevticks = 0;\n\tret->tid = 0;\n\n\treturn ret;\n}\n\n/**\n * Compare server s against the address info in saddr.  A server is\n * considered to be equal is it is of the same socket family, type and\n * protocol, and if the target address and port are the same.  When\n * saddr is NULL, a match against the given ip is attempted, e.g. for\n * file destinations.\n */\nchar\nserver_cmp(server *s, struct addrinfo *saddr, const char *ip)\n{\n\tif ((saddr == NULL || s->saddr == NULL)) {\n\t\tif (strcmp(s->ip, ip) == 0)\n\t\t\treturn 0;\n\t} else if (\n\t\t\ts->saddr->ai_family == saddr->ai_family &&\n\t\t\ts->saddr->ai_socktype == saddr->ai_socktype &&\n\t\t\ts->saddr->ai_protocol == saddr->ai_protocol\n\t   )\n\t{\n\t\tif (saddr->ai_family == AF_INET) {\n\t\t\tstruct sockaddr_in *l = ((struct sockaddr_in *)s->saddr->ai_addr);\n\t\t\tstruct sockaddr_in *r = ((struct sockaddr_in *)saddr->ai_addr);\n\t\t\tif (l->sin_port == r->sin_port &&\n\t\t\t\t\tl->sin_addr.s_addr == r->sin_addr.s_addr)\n\t\t\t\treturn 0;\n\t\t} else if (saddr->ai_family == AF_INET6) {\n\t\t\tstruct sockaddr_in6 *l =\n\t\t\t\t((struct sockaddr_in6 *)s->saddr->ai_addr);\n\t\t\tstruct sockaddr_in6 *r =\n\t\t\t\t((struct sockaddr_in6 *)saddr->ai_addr);\n\t\t\tif (l->sin6_port == r->sin6_port &&\n\t\t\t\t\tmemcmp(l->sin6_addr.s6_addr, r->sin6_addr.s6_addr,\n\t\t\t\t\t\tsizeof(l->sin6_addr.s6_addr)) == 0)\n\t\t\t\treturn 0;\n\t\t}\n\t}\n\n\treturn 1;  /* not equal */\n}\n\n/**\n * Starts a previously created server using server_new().  Returns\n * errno if starting a thread failed, after which the caller should\n * server_free() the given s pointer.\n */\nchar\nserver_start(server *s)\n{\n\treturn pthread_create(&s->tid, NULL, &server_queuereader, s);\n}\n\n/**\n * Adds a list of secondary servers to this server.  A secondary server\n * is a server which\' queue will be checked when this server has nothing\n * to do.\n */\nvoid\nserver_add_secondaries(server *self, server **secondaries, size_t count)\n{\n\tself->secondaries = secondaries;\n\tself->secondariescnt = count;\n}\n\n/**\n * Flags this server as part of a failover cluster, which means the\n * secondaries are used only to offload on failure, not on queue stress.\n */\nvoid\nserver_set_failover(server *self)\n{\n\tself->failover = 1;\n}\n\n/**\n * Sets instance name only used for carbon_ch cluster type.\n */\nvoid\nserver_set_instance(server *self, char *instance)\n{\n\tif (self->instance != NULL)\n\t\tfree(self->instance);\n\tself->instance = strdup(instance);\n}\n\n/**\n * Thin wrapper around the associated queue with the server object.\n * Returns true if the metric could be queued for sending, or the metric\n * was dropped because the associated server is down.  Returns false\n * otherwise (when a retry seems like it could succeed shortly).\n */\ninline char\nserver_send(server *s, const char *d, char force)\n{\n\tif (queue_free(s->queue) == 0) {\n\t\tchar failure = __sync_add_and_fetch(&(s->failure), 0);\n\t\tif (!force && s->secondariescnt > 0) {\n\t\t\tsize_t i;\n\t\t\t/* don\'t immediately drop if we know there are others that\n\t\t\t * back us up */\n\t\t\tfor (i = 0; i < s->secondariescnt; i++) {\n\t\t\t\tif (!__sync_add_and_fetch(&(s->secondaries[i]->failure), 0)) {\n\t\t\t\t\tfailure = 0;\n\t\t\t\t\tbreak;\n\t\t\t\t}\n\t\t\t}\n\t\t}\n\t\tif (failure || force ||\n\t\t\t\t__sync_add_and_fetch(&(s->stallseq), 0) == s->maxstalls)\n\t\t{\n\t\t\t__sync_add_and_fetch(&(s->dropped), 1);\n\t\t\t/* excess event will be dropped by the enqueue below */\n\t\t} else {\n\t\t\t__sync_add_and_fetch(&(s->stallseq), 1);\n\t\t\t__sync_add_and_fetch(&(s->stalls), 1);\n\t\t\treturn 0;\n\t\t}\n\t} else {\n\t\t__sync_and_and_fetch(&(s->stallseq), 0);\n\t}\n\tqueue_enqueue(s->queue, d);\n\n\treturn 1;\n}\n\n/**\n * Tells this server to finish sending pending items from its queue.\n */\nvoid\nserver_shutdown(server *s)\n{\n\tint i;\n\tsize_t failures;\n\tsize_t inqueue;\n\n\t/* this function should only be called on a running server */\n\tif (__sync_bool_compare_and_swap(&(s->keep_running), 0, 0))\n\t\treturn;\n\n\tif (s->secondariescnt > 0) {\n\t\t/* if we have a working connection, or we still have stuff in\n\t\t * our queue, wait for our secondaries, as they might need us,\n\t\t * or we need them */\n\t\tdo {\n\t\t\tfailures = 0;\n\t\t\tinqueue = 0;\n\t\t\tfor (i = 0; i < s->secondariescnt; i++) {\n\t\t\t\tif (__sync_add_and_fetch(&(s->secondaries[i]->failure), 0))\n\t\t\t\t\tfailures++;\n\t\t\t\tif (__sync_add_and_fetch(&(s->secondaries[i]->running), 0))\n\t\t\t\t\tinqueue += queue_len(s->secondaries[i]->queue);\n\t\t\t}\n\t\t\t/* loop until we all failed, or nothing is in the queues */\n\t\t} while (failures != s->secondariescnt &&\n\t\t\t\tinqueue != 0 &&\n\t\t\t\tlogout("any_of cluster pending %zu metrics "\n\t\t\t\t\t"(with %zu failed nodes)\\n", inqueue, failures) >= -1 &&\n\t\t\t\tusleep((200 + (rand() % 100)) * 1000) <= 0);\n\t\t/* shut down entire cluster */\n\t\tfor (i = 0; i < s->secondariescnt; i++)\n\t\t\t__sync_bool_compare_and_swap(\n\t\t\t\t\t&(s->secondaries[i]->keep_running), 1, 0);\n\t\t/* to pretend to be dead for above loop (just in case) */\n\t\tif (inqueue != 0)\n\t\t\tfor (i = 0; i < s->secondariescnt; i++)\n\t\t\t\t__sync_add_and_fetch(&(s->secondaries[i]->failure), 1);\n\t\t/* wait for the secondaries to be stopped so we surely don\'t get\n\t\t * invalid reads when server_free is called */\n\t\tfor (i = 0; i < s->secondariescnt; i++) {\n\t\t\twhile (__sync_add_and_fetch(&(s->secondaries[i]->running), 0))\n\t\t\t\tusleep((200 + (rand() % 100)) * 1000);\n\t\t}\n\t}\n\n\t__sync_bool_compare_and_swap(&(s->keep_running), 1, 0);\n}\n\n/**\n * Frees this server and associated resources.  This includes joining\n * the server thread.\n */\nvoid\nserver_free(server *s) {\n\tint err;\n\n\tif (s->tid != 0 && (err = pthread_join(s->tid, NULL)) != 0)\n\t\tlogerr("%s:%u: failed to join server thread: %s\\n",\n\t\t\t\ts->ip, s->port, strerror(err));\n\ts->tid = 0;\n\n\tif (s->ctype == CON_TCP) {\n\t\tsize_t qlen = queue_len(s->queue);\n\t\tif (qlen > 0)\n\t\t\tlogerr("dropping %zu metrics for %s:%u\\n",\n\t\t\t\t\tqlen, s->ip, s->port);\n\t}\n\n\tqueue_destroy(s->queue);\n#ifdef HAVE_SSL\n\tif (s->transport & W_SSL)\n\t\tSSL_CTX_free(s->strm->ctx);\n#endif\n\tfree(s->batch);\n\tif (s->instance)\n\t\tfree(s->instance);\n\tif (s->saddr != NULL)\n\t\tfreeaddrinfo(s->saddr);\n\tif (s->hint)\n\t\tfree(s->hint);\n\tif (s->mtlspemcert)\n\t\tfree(s->mtlspemcert);\n\tif (s->mtlspemkey)\n\t\tfree(s->mtlspemkey);\n\tfree((char *)s->ip);\n\tif (s->strm->nextstrm != NULL)\n\t\tfree(s->strm->nextstrm);\n\tfree(s->strm);\n\ts->ip = NULL;\n\tfree(s);\n}\n\n/**\n * Swaps the queue between server l to r.  This is assumes both l and r\n * are not running.\n */\nvoid\nserver_swap_queue(server *l, server *r)\n{\n\tqueue *t;\n\n\tassert(l->keep_running == 0 || l->tid == 0);\n\tassert(r->keep_running == 0 || r->tid == 0);\n\n\tt = l->queue;\n\tl->queue = r->queue;\n\tr->queue = t;\n\t\n\t/* swap associated statistics as well */\n\tl->metrics = r->metrics;\n\tl->dropped = r->dropped;\n\tl->stalls = r->stalls;\n\tl->ticks = r->ticks;\n\tl->prevmetrics = r->prevmetrics;\n\tl->prevdropped = r->prevdropped;\n\tl->prevstalls = r->prevstalls;\n\tl->prevticks = r->prevticks;\n}\n\n/**\n * Returns the ip address this server points to.\n */\ninline const char *\nserver_ip(server *s)\n{\n\tif (s == NULL)\n\t\treturn NULL;\n\treturn s->ip;\n}\n\n/**\n * Returns the port this server connects at.\n */\ninline unsigned short\nserver_port(server *s)\n{\n\tif (s == NULL)\n\t\treturn 0;\n\treturn s->port;\n}\n\n/**\n * Returns the instance associated with this server.\n */\ninline char *\nserver_instance(server *s)\n{\n\treturn s->instance;\n}\n\n/**\n * Returns the connection protocol of this server.\n */\ninline con_proto\nserver_ctype(server *s)\n{\n\tif (s == NULL)\n\t\treturn CON_PIPE;\n\treturn s->ctype;\n}\n\n/**\n * Returns the connection transport of this server.\n */\ninline con_trnsp\nserver_transport(server *s)\n{\n\tif (s == NULL)\n\t\treturn W_PLAIN;\n\treturn s->transport;\n}\n\n/**\n * Returns the connection type of this server.\n */\ninline con_type\nserver_type(server *s)\n{\n\tif (s == NULL)\n\t\treturn T_LINEMODE;\n\treturn s->type;\n}\n\n/**\n * Returns whether the last action on this server caused a failure.\n */\ninline char\nserver_failed(server *s)\n{\n\tif (s == NULL)\n\t\treturn 0;\n\treturn __sync_add_and_fetch(&(s->failure), 0);\n}\n\n/**\n * Returns the wall-clock time in microseconds (us) consumed sending metrics.\n */\ninline size_t\nserver_get_ticks(server *s)\n{\n\tif (s == NULL)\n\t\treturn 0;\n\treturn __sync_add_and_fetch(&(s->ticks), 0);\n}\n\n/**\n * Returns the wall-clock time in microseconds (us) consumed since last\n * call to this function.\n */\ninline size_t\nserver_get_ticks_sub(server *s)\n{\n\tsize_t d;\n\tif (s == NULL)\n\t\treturn 0;\n\td = __sync_add_and_fetch(&(s->ticks), 0) - s->prevticks;\n\ts->prevticks += d;\n\treturn d;\n}\n\n/**\n * Returns the number of metrics sent since start.\n */\ninline size_t\nserver_get_metrics(server *s)\n{\n\tif (s == NULL)\n\t\treturn 0;\n\treturn __sync_add_and_fetch(&(s->metrics), 0);\n}\n\n/**\n * Returns the number of metrics sent since last call to this function.\n */\ninline size_t\nserver_get_metrics_sub(server *s)\n{\n\tsize_t d;\n\tif (s == NULL)\n\t\treturn 0;\n\td = __sync_add_and_fetch(&(s->metrics), 0) - s->prevmetrics;\n\ts->prevmetrics += d;\n\treturn d;\n}\n\n/**\n * Returns the number of metrics dropped since start.\n */\ninline size_t\nserver_get_dropped(server *s)\n{\n\tif (s == NULL)\n\t\treturn 0;\n\treturn __sync_add_and_fetch(&(s->dropped), 0);\n}\n\n/**\n * Returns the number of metrics dropped since last call to this function.\n */\ninline size_t\nserver_get_dropped_sub(server *s)\n{\n\tsize_t d;\n\tif (s == NULL)\n\t\treturn 0;\n\td = __sync_add_and_fetch(&(s->dropped), 0) - s->prevdropped;\n\ts->prevdropped += d;\n\treturn d;\n}\n\n/**\n * Returns the number of stalls since start.  A stall happens when the\n * queue is full, but it appears as if it would be a good idea to wait\n * for a brief period and retry.\n */\ninline size_t\nserver_get_stalls(server *s)\n{\n\tif (s == NULL)\n\t\treturn 0;\n\treturn __sync_add_and_fetch(&(s->stalls), 0);\n}\n\n/**\n * Returns the number of stalls since last call to this function.\n */\ninline size_t\nserver_get_stalls_sub(server *s)\n{\n\tsize_t d;\n\tif (s == NULL)\n\t\treturn 0;\n\td = __sync_add_and_fetch(&(s->stalls), 0) - s->prevstalls;\n\ts->prevstalls += d;\n\treturn d;\n}\n\n/**\n * Returns the (approximate) number of metrics waiting to be sent.\n */\ninline size_t\nserver_get_queue_len(server *s)\n{\n\tif (s == NULL)\n\t\treturn 0;\n\treturn queue_len(s->queue);\n}\n\n/**\n * Returns the allocated size of the queue backing metrics waiting to be\n * sent.\n */\ninline size_t\nserver_get_queue_size(server *s)\n{\n\tif (s == NULL)\n\t\treturn 0;\n\treturn queue_size(s->queue);\n}\n', 'message': 'server_start: fix return\n\nAllow errno to be retrieved for more informational error messages on\nfailure.  Use server_start explicitly rather than implicitly in\nrouter_readconfig.\n', 'author_name': 'Fabian Groffen', 'lang': 'c'}