#if defined(__linux__)
#	define _XOPEN_SOURCE
#	define _BSD_SOURCE
#endif
#if defined(__sun) || defined(__linux__)
#	include <alloca.h>
#	include <sys/sendfile.h>
#endif
#include <errno.h>
#include <fcntl.h>
#include <libgen.h>
#include <sys/param.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <strings.h>
#include <sysexits.h>
#include <sys/types.h> 
#include <sys/socket.h>
#include <sys/stat.h>
#include <netinet/in.h>
#include <unistd.h>
#include <netdb.h>
#include <arpa/inet.h>
#include <poll.h>
#include <signal.h>
#include <sys/time.h>
#include <bzlib.h>
#include <zlib.h>
#if !defined(INFTIM)
#	define	INFTIM	(-1)
#endif

#include <sybfront.h>
#include <sybdb.h>

#ifndef __GNUC__
#	define __unused
#endif

static DBPROCESS *sybconnect(char *appname, char *user, char *pw, char *server);
static void sybinit(void);
static int err_handler(DBPROCESS *dbp, int severity, int dberr, int oserr,
    char *dberrstr, char *oserrstr);
static int msg_handler(DBPROCESS *dbp, DBINT msgno, int msgstate, int severity,
    char *msgtext, char *srvname, char *procname, int line);
static int dumper(DBPROCESS *dbp, const char *db), loader(DBPROCESS *dbp, const char *db);
static int transfer(const char *from, DBPROCESS *one, const char *to, DBPROCESS *two,
    unsigned short low, unsigned short high);
static void sighandler(int sig);

union {
	FILE	*pf;
	BZFILE	*bz;
	gzFile	 gz;
} dump;
enum {
	PLAIN, BZIP, GZIP
} dumptype = PLAIN;
static int level = 0, seekable = 0;

static struct addrinfo *ai;
static char ip[63];
static unsigned short ports[32] = { 0 };
/* static unsigned short port = 0; */
static int	sock, debug = 0, Timeout = INFTIM;
static DBINT	awaiting = 0 /*, awaiting1 = 0 */; /* should be an array really */
static struct dumpinfo {
	const char	*db;
	DBPROCESS	*dbp;
} dumpinfo;

#define dprintf(format, ...)	if (debug) fprintf(stderr, "%s: " format "\n", \
	__func__, __VA_ARGS__)

static void
sybdrain(DBPROCESS *dbp)
{
	int	res;

	do {
		res = dbresults(dbp);
		dprintf("dbresults(%p) returned %d", dbp, res);
		if (res == FAIL)
			fprintf(stderr, "dbresults(%p) reported FAILure\n", dbp);
	} while (res == SUCCEED);
}

static volatile int	stopnow = 0;

static void
sighandler(int sig)
{
	if (sig == SIGALRM) {
		fprintf(stderr, "Got SIGALRM -- leaving now.\n");
		exit(EX_UNAVAILABLE);
	}
	if (stopnow++ > 3) {
		fprintf(stderr, "Ok, repeated stop requests -- leaving.\n");
		exit(EX_UNAVAILABLE);
	} else
		fputs("Got a signal, trying to exit cleanly\n", stderr);
	fflush(stderr);
}

static void
sybstatements(DBPROCESS *dbp, const char * const commands[], const char *db,
    const char *address, unsigned short *port, char *buf, size_t buflen)
{
	int	len;

	while (*commands) {
		len = snprintf(buf, buflen, *commands, db, address, *port) + 2; /* XXX parallel==1 */
		if (len >= (int)buflen) {
			fprintf(stderr, "%s: provided buffer is not big "
			    "enough\n", __func__);
			exit(EX_SOFTWARE);
		}
		dprintf("\t%s", buf);
		/* XXX For some reason, we must append spaces */
		buf[len - 2] = ' '; buf[len - 1] = '\0';
		dbcmd(dbp, buf);
		buf += len;
		buflen -= len;
		commands++;
	}
	awaiting = 3101; /* May get a notice about db being busy */
	if (dbsqlexec(dbp) == FAIL) {
		fprintf(stderr, "failed to send above statements to server\n");
		exit(EX_PROTOCOL);
	}

	dprintf("commands above sent to server %p", dbp);

	if (awaiting == 0)
		exit(EX_PROTOCOL);

	awaiting = 412402; /* syb_open() failure */
	len = dbresults(dbp);
	dprintf("dbresults(%p) returned %d", dbp, len);
#if 0
	dbpoll(dbp, -1, &dbp, &len);
	dprintf("dbpoll(%p) returned %d", dbp, len);
	len = dbresults(dbp);
	dprintf("dbresults(%p) returned %d", dbp, len);
#endif

	if (awaiting == 0)
		exit(EX_PROTOCOL);

}

static int
intraccept(int s)
{
	int	result;
#if 0
	struct pollfd	 pollfd;

	pollfd.fd = s;
	pollfd.events = POLLRDNORM;
	result = poll(&pollfd, 1, 10000 /* 10 seconds */);
	switch (result) {
	case 0:
		fprintf(stderr, "Timeout (%d seconds) awaiting connection\n",
		    10000/1000);
		exit(EX_PROTOCOL);
	case -1:
		perror("poll");
		exit(EX_PROTOCOL);
	case 1:
		dprintf("Incoming connection detected on %d", s);
		break;
	default:
		fprintf(stderr, "Unexpected value %d returned by poll\n",
		    result);
		exit(EX_SOFTWARE);
	}
#endif

	for (;;) {
		result = accept(s, NULL, 0);
		if (result >= 0)
			break;
		if (stopnow)
			exit(EX_UNAVAILABLE);
		if (errno == EINTR)
			continue;
		perror("accept");
		exit(EX_SOFTWARE);
	}

	dprintf("socket %d accepted", result);
	if (fcntl(result, F_SETFL, O_NONBLOCK) == -1) {
		perror("fcntl(..., O_NONBLOCK)");
		exit(EX_OSERR);
	}

	return result;
}

static void
close_dump(void)
{
	switch (dumptype) {
	case BZIP:
		BZ2_bzclose(dump.bz);
		break;
	case GZIP:
		gzclose(dump.gz);
		break;
	case PLAIN:
		fclose(dump.pf);
	}
}

static int
loader(DBPROCESS *dbp, const char *db)
{
	char		 buf[65536*4];
	ssize_t		 size = 0;
	off_t		 offset = 0;
	int		 incoming = 10, result = 1, use_sf;
	int		 timeout = 1000;
#ifdef SF_NODISKIO
	off_t		 sent;
	int		 flags = 0;
#else
	size_t		 sent;
#endif
	struct pollfd	 pollfds[2];
	struct stat	 sb;
	int		 sybsock = DBIORDESC(dbp);
	const char	*commands[] = {
		"load database %s from 'pipe::P%s:%hu'",
		"online database %s",
		NULL
	};
	const char	*optional[] = {
		"use %s",
		"EXEC sp_post_xpload",
		NULL
	};

	if (dumptype == PLAIN) {
		if (fstat(fileno(dump.pf), &sb)) {
			perror("fstat");
			exit(EX_NOINPUT);
		}
		use_sf = seekable;
		dprintf("stdin is %sa seekable (%d), will %suse sendfile",
		    use_sf ? "" : "not ", use_sf, use_sf ? "" : "not ");
	} else
		use_sf = 0;

	sybstatements(dbp, commands, db, ip, ports, buf, sizeof buf);

	awaiting = 8009; /* failure */
	pollfds[0].fd = sybsock;
	pollfds[0].events = POLLRDNORM|POLLIN;
	pollfds[1].fd = sock;
	pollfds[1].events = POLLWRNORM|POLLOUT;

	for (;;) {
		int	err;

		err = poll(pollfds, 2, timeout);
		if (err == 0) {
			if (incoming) {
				dprintf("Will wait for %d more %d-second "
				    "periods for incoming connection",
				    incoming--, timeout/1000);
				continue;
			}
			fprintf(stderr, "Giving up on ever receiving "
			    "connection for loading\n");
			exit(EX_PROTOCOL);
		}
		if (err == -1 && errno != EINTR) {
			perror("poll");
			close(incoming);
			return 1;
		}
		if (stopnow)
			exit(EX_UNAVAILABLE);

		if (pollfds[0].revents) {
			dprintf("server had something to say (socket %.*sready)", 
			    pollfds[1].revents ? 0 : 4, "not ");
			dbresults(dbp);
			if (pollfds[1].fd == sock && awaiting == 0)
				exit(EX_PROTOCOL);
		}

		if (pollfds[1].revents) {
			if (pollfds[1].fd == sock) {
				incoming = intraccept(sock);
				pollfds[1].fd = incoming;
				dprintf("Accepted sock %d as %d. "
				    "Sending first chunk",
				    sock, incoming);
				timeout = Timeout;
				/*
				 * May get a notice about running sp_post_xpload
				 */
				awaiting = 3163;
			}
			if (use_sf) {
#ifdef SF_NODISKIO	/* BSD sendfile */
				flags = SF_NODISKIO - flags;
				size = sendfile(fileno(dump.pf), incoming, offset, 0,
				    NULL, &sent, flags);
				if (size > 0)
					dprintf("done sendfile-ing with %d. "
					    "sent %llu bytes",
					    flags, (unsigned long long)sent);
				if (size == 0 || errno == EAGAIN || errno == EBUSY) {
					offset += sent;
					if (offset == sb.st_size) {
						result = 0;
						goto out;
					}
					continue;
				}
#elif defined(__linux__)
				size = sendfile(fileno(dump.pf), incoming,
				    &offset, sb.st_size - offset);
				switch (errno) {
				case EAGAIN:
					dprintf("sent %ld bytes out, offset now %lld",
					    (long)size, (long long)offset);
					/* FALLTHROUGH */
				case 0:
					continue;
				}
#else	/* Linux/Solaris sendfile */
				struct sendfilevec sfv = {
					fileno(dump.pf), 0
				};
				sfv.sfv_off = offset;
				sfv.sfv_len = sb.st_size - offset;

				dprintf("sending %d bytes out, offset now %lld",
				    (int)(sb.st_size - offset), (unsigned long long)offset);
				signal(SIGPIPE, SIG_IGN);
				size = sendfilev(incoming, &sfv, 1, &sent);
				signal(SIGPIPE, sighandler);
				offset += sent;
				if (size >= 0) {
					dprintf("sent everything out, offset now %lld",
					    (long long)offset);
					goto out;
				}
				if (errno == EAGAIN) {
					dprintf("sent some bytes out, offset now %lld",
					    (long long)offset);
					continue;
				}
#endif
				perror("sendfile");
				result = 1;
				goto out;
			} else {
				if (offset == 0) {
					static int notfirst;

					/* fill the buffer again */
					switch (dumptype) {
					case PLAIN:
						size = fread(buf, 1,
						    sizeof buf, dump.pf);
						if (size == 0) {
							if (ferror(stdin)) {
								perror("fread");
								result = 1;
							}
							goto out;
						}
						break;
					case GZIP:
						size = gzread(dump.gz, buf,
						    sizeof buf);
						if (size == -1) {
							fprintf(stderr, "gzread: "
							    "%s\n", gzerror(dump.gz,
								&notfirst));
							goto out;
						}
						break;
					case BZIP:
						size = BZ2_bzread(dump.bz, buf,
						    sizeof buf);
						if (size == -1) {
							fprintf(stderr, "bzread: "
							    "%s\n",
							    BZ2_bzerror(dump.bz,
								&notfirst));
							goto out;
						}
						break;
					}
					if (!notfirst) {
						dprintf("Read %lld. The dump starts "
						    "with `%.4s'", (long long)size, buf);
						notfirst = 1;
					}
				}

				if (size == 0) {
					result = 0;
					goto out;
				}
				dprintf("sending %d bytes (starting at %d)",
				    (int)size, (int)offset);
				signal(SIGPIPE, SIG_IGN);
				sent = write(incoming, buf + offset, size);
				signal(SIGPIPE, sighandler);
				dprintf("sent %d bytes (starting at %d)",
				    (int)sent, (int)offset);
				if (sent == -1) {
					perror("sending data");
					result = 1;
					goto out;
				}
				size -= sent;
				offset = (offset + sent) % sizeof buf;
			}
		}
	}

out:
	close(incoming);
	close_dump();

	dprintf("ended %ssuccessfully, socket %d closed, "
	    "awaiting server's confirmation...", result ? "un" : "", incoming);
	sybdrain(dbp);
	if (awaiting == 0) { /* Got a notice to run sp_post_xpload */
		fprintf(stderr, "%s: Server suggested, we run `%s', obeying\n",
		    __func__, "sp_post_xpload");
		sybstatements(dbp, optional, db, ip, ports, buf, sizeof buf);
		sybdrain(dbp);
	}

	return result;
}

static int
dumper(DBPROCESS *dbp, const char *db)
{
	char		 buf[65536*4];
	ssize_t		 size;
	int		 incoming = 10, timeout = 1000;
	struct pollfd	 pollfds[2];
	int		 sybsock = DBIORDESC(dbp);
	const char	*commands[] = {
		"EXEC sp_dboption '%s', 'single user', 'TRUE'",
		"use %s",
		" EXEC sp_flushstats ",
		" checkpoint ",
		" use master ",
		" dump database %s to 'pipe::P%s:%hu' ",
		" EXEC sp_dboption '%s', 'single user', 'FALSE' ",
		" use %s ",
		" checkpoint ",
		NULL
	};

	sybstatements(dbp, commands, db, ip, ports, buf, sizeof buf);

#if 0
	/* Await the message 602801, indicating success, up to 20 times */
	for (awaiting = 602801, incoming = 20; awaiting; incoming--)
		if (dbresults(dbp) == NO_MORE_RESULTS || !incoming)
			exit(EX_PROTOCOL);
#endif

	pollfds[0].fd = sock;
	pollfds[0].events = POLLRDNORM|POLLIN;
	pollfds[1].fd = sybsock;
	pollfds[1].events = POLLRDNORM|POLLIN;
	for (;;) {
		int	err;

		err = poll(pollfds, 2, timeout);
		if (err == 0) {
			if (incoming) {
				dprintf("Will wait for %d more %d-second "
				    "periods for incoming connection",
				    incoming--, timeout/1000);
				continue;
			}
			fprintf(stderr, "Giving up on ever receiving "
			    "connection for loading\n");
			exit(EX_PROTOCOL);
		}
		if (err == -1 && errno != EINTR) {
			perror("poll");
			close(incoming);
			return 1;
		}
		if (stopnow) {
			dprintf("Canceling operation on %p", dbp);
			close(incoming);
			dbcanquery(dbp);
			dbsettime(20); /* twenty seconds timeout */
			alarm(21); /* the above timeout seems faulty */
			sybdrain(dbp);
			exit(EX_UNAVAILABLE);
		}


		if (pollfds[1].revents)
			dbresults(dbp);

		if (pollfds[0].revents) {
			if (pollfds[0].fd == sock) {
				incoming = intraccept(sock);
				pollfds[0].fd = incoming;
				
				dprintf("Accepted sock %d as %d. "
				    "Reading first chunk",
				    sock, incoming);
				timeout = Timeout;
			}
			size = recv(incoming, buf, sizeof buf, 0);
			if (size == 0)
				break;
			dprintf("Received %zd bytes from server, outputting",
			    size);
			if (size > 0) switch(dumptype) {
			case GZIP:
				gzwrite(dump.gz, buf, size);
				continue;
			case BZIP:
				BZ2_bzwrite(dump.bz, buf, size);
				continue;
			case PLAIN:
				fwrite(buf, 1, size, dump.pf);
				continue;
			}
			else if (errno == EAGAIN)
				continue;
			else {
				perror("reading from socket");
				close(incoming);
				return 1;
			}
		}
	}

	close(incoming);
	close_dump();
	dprintf("ended, %d closed, awaiting server's confirmation...", incoming);
	sybdrain(dbp);

	return 0;
}

static int
err_handler(DBPROCESS *dbp, int severity, int dberr, int oserr,
    char *dberrstr, char *oserrstr)
{

	if (1 || dberr != SYBESMSG) {
		if (dberrstr == NULL)
			dberrstr = "(null)";
		if (oserrstr == NULL)
			oserrstr = "(null)";

		fprintf(stderr, "ERR(%p): %s (%d) %s (%d)\n", dbp,
		    dberrstr, dberr, oserrstr, oserr);
		fflush(stderr);
	}
	return (dbp == NULL) || (DBDEAD(dbp)) ? INT_EXIT : INT_CANCEL;
}

/*
 * Return the difference between the two timeval operands - in microseconds
 */
static long
timediff(const struct timeval to, const struct timeval from)
{
	return (to.tv_sec - from.tv_sec)*1000000 + to.tv_usec - from.tv_usec;
}

static short parallel = 1;
static int
msg_handler(DBPROCESS *dbp, DBINT msgno, int msgstate, int severity,
    char *msgtext, char *srvname, char *procname, int line)
{
	int	msglen, portn = 0;
	static	unsigned long lastkb;
	static	struct timeval	lasttime, starttime;
	unsigned long	kb = 0;
	struct timeval	curtime;
	const char	*commandsdump[] = {
		"dump database %s to 'pipe::P%s:%hu'",
		"EXEC sp_dboption '%s', 'single user', 'FALSE'",
		"use %s"
		"checkpoint",
		NULL
	};

	if (starttime.tv_sec == 0) {
		gettimeofday(&lasttime, NULL);
		starttime = lasttime;
	}

	msglen = strlen(msgtext);

	/* The last char is often new-line, so we output one less */
	if (msgtext[msglen-1] == '\n')
		msglen--;
	if (debug)
		fprintf(stderr, "%s: (%d, %d) %.*s (%sawaited)", srvname, (int)msgno,
		    msgstate, msglen, msgtext,
		    awaiting ? awaiting == msgno ? "" : "not " : "nothing ");
	else
		fprintf(stderr, "%s: %.*s", srvname, msglen, msgtext);

	if (awaiting && msgno == awaiting) {
		dprintf("Caught the expected message #%d", (int)msgno);
		awaiting = 0;
	}

	if (msgno == 412501 && dbgetuserdata(dbp) &&
	    (msgtext = strstr(msgtext, "AWAITING ON"))) {
		if (sscanf(msgtext + sizeof("AWAITING ON"),
		    "%s %hu", ip, ports + portn) == 2) {
			char	buf[1023];
			dprintf("Got the ip and the port number %d: %s:%hu",
			    portn, ip, ports[portn]);
			if (++portn == parallel)
				sybstatements(dumpinfo.dbp, commandsdump, dumpinfo.db,
				    ip, ports, buf, sizeof buf);
			} else
				dprintf("Could not get ip:port from %s", msgtext);
	}

	/* Provide througput statistics: */
	if (msgno == 405801) {
		/* Find the third colon in the message: */
		msgtext = index(msgtext + 1, ':');
		msgtext = index(msgtext + 1, ':');
		msgtext = index(msgtext + 1, ':');
		/* Parse the number of dumped kilobytes out of the remainder: */
		sscanf(msgtext + 1, "%lu kilobytes ", &kb);
		gettimeofday(&curtime, NULL);
		fprintf(stderr, " (%.2fMb/s average, %.2fMb/s recent.)\n",
		    kb*1000000./timediff(curtime, starttime),
		    (kb - lastkb)*1000000./timediff(curtime, lasttime));
		lastkb = kb;
		lasttime = curtime;
	} else
		fputs("\n", stderr);

	return 0;
}

static void
sybinit(void)
{

	if (dbinit() == FAIL)
		exit(EX_SOFTWARE);

	dberrhandle(err_handler);
	dbmsghandle(msg_handler);
}

static DBPROCESS *
sybconnect(appname, user, pw, server)
	/* const */ char *appname, *user, *pw, *server;
{
	LOGINREC	*login;
	DBPROCESS	*dbp;

	login = dblogin();
	DBSETLUSER(login, user);
	DBSETLPWD(login, pw);
	DBSETLAPP(login, appname);

	dprintf("Connecting to %s as %s", server, user);
	dbp = dbopen(login, server);
	if (dbp == NULL) {
		fprintf(stderr, "Connecting to %s as %s failed\n",
		    server ? server : "null", user ? user : "null");
		exit(EX_DATAERR);
	}
	dprintf("dbopen succeeded (%p) for %s@%s", dbp, user ? user : "null",
	    server ? server : "null");
	dbloginfree(login);
	return dbp;
}

static void
getsock(unsigned short low, unsigned short high)
{
	int		err;
	unsigned short	port;

	sock = socket(ai->ai_family, SOCK_STREAM, 0);
	if (sock < 0) {
		perror("socket");
		exit(EX_TEMPFAIL);
	}
	for (port = low; port <= high; port++) {
		switch (ai->ai_family) {
		case AF_INET:
			((struct sockaddr_in *)ai->ai_addr)->sin_port = htons(port);
			break;
		case AF_INET6:
			((struct sockaddr_in6 *)ai->ai_addr)->sin6_port = htons(port);
			break;
		}
		if (bind(sock, (struct sockaddr *)ai->ai_addr, ai->ai_addrlen) == 0)
			break;
	}

	if (port > high) {
		fprintf(stderr, "Could not bind to any port in the "
		    "%hu-%hu range.\n", low, high);
		exit(EX_UNAVAILABLE);
	}

	err = getnameinfo((struct sockaddr *)ai->ai_addr, ai->ai_addrlen,
	    ip, sizeof ip, NULL, 0, NI_NUMERICHOST);
	if (err) {
		fprintf(stderr, "getnameinfo failed: %s", gai_strerror(err));
		exit(EX_SOFTWARE);
	}

	if (listen(sock, 1)) {
		perror("listen");
		exit(EX_TEMPFAIL);
	}

	ports[0] = port; /* parallel == 1 XXX */
	dprintf("Bound to port %hu", port);

}

static int
transfer(const char *from, DBPROCESS *one, const char *to, DBPROCESS *two,
    unsigned short low, unsigned short high)
{
	DBPROCESS	*other = NULL;
#if 1
	struct pollfd	 pollfds[2];
#endif
	char	buf[1023];
	int	result = 0, sybres, twospoke = 0;
	const char	*commandsload[] = {
		"load database %s from 'pipe::S:%s:%hu'",
		"online database %s",
		NULL
	};
	const char	*postload[] = {
		"use %s",
		"EXEC sp_post_xpload",
		NULL
	};
	const char	*predump[] = {
		"EXEC sp_dboption '%s', 'single user', 'TRUE'",
		"EXEC sp_flushstats",
		"use %s",
		"checkpoint ",
		"use master",
		NULL
	};
	/*
	 * First we tell the loading database to start waiting
	 */
	/*
	 * This makes sure, the statement constructed will contain
	 * the correct low port:
	 */
	sprintf(ip, "%hu", low);
	sybstatements(one, predump, from, NULL, 0, buf, sizeof buf);
	sybdrain(one);
	sybstatements(two, commandsload, to, ip, &high, buf, sizeof buf);

	dumpinfo.db = from;
	dumpinfo.dbp = one;
	dbsetuserdata(two, (void *)1);
	dbresults(two); /* one more line of results expected */

	if (!ports[0] /* port and ip must be initialized by msg_handler */) {
		fprintf(stderr, "port remains uninitialized\n");
		exit(EX_PROTOCOL);
	}

	dprintf("Transfer initiated (to %s:%hu), dbpolling", ip, ports[0]);
	awaiting = 412402; /* Make sure, the source had no problems */
	dprintf("Checking %p for message #%d", one, (int)awaiting);
	dbresults(one);
	dbresults(one);
	dbresults(one);
	if (awaiting == 0) {
		dprintf("Dumper (%p) reported difficulties, exiting", one);
		exit(EX_UNAVAILABLE);
	}
#if 1
	pollfds[0].fd = DBIORDESC(one);
	pollfds[1].fd = DBIORDESC(two);
	pollfds[0].events = pollfds[1].events = POLLRDNORM;
	for (;;) {
		sybres = poll(pollfds, 2, Timeout);
		switch (sybres) {
		case -1:
			perror("poll");
			continue;
		default:
			dprintf("poll returned %d, examining", sybres);
		}
		if (stopnow) {
			dbcanquery(one);
			dbcanquery(two);
			dbsettime(20); /* twenty seconds timeout */
			alarm(41); /* the above timeout seems faulty */
			sybdrain(one);
			sybdrain(two);
			exit(EX_UNAVAILABLE);
		}
		if (pollfds[0].revents) {
			dprintf("poll reported readiness of %p", one);
			sybres = dbresults(one);
			dprintf("dbresults(%p) returned %d", one, sybres);
			if (sybres != SUCCEED) {
				other = two;
				goto out;
			}
		}

		if (pollfds[1].revents) {
			dprintf("poll reported readiness of %p", two);
			twospoke = 1;
			awaiting = 304201; /* LOAD is complete */
			sybres = dbresults(two);
			dprintf("dbresults(%p) returned %d", two, sybres);
			if (awaiting == 0) {
				other = two;
				goto out;
			}
			if (sybres != SUCCEED) {
				other = one;
				goto out;
			}

		}
	}
out:
	awaiting = 3163; /* May get a notice about running sp_post_xpload */
	dprintf("Out of the poll loop, draining %p", other);
	if (other == two && !twospoke) {
		dprintf("Actually, %p never spoke, skipping it", two);
	} else
		sybdrain(other);

#else
	while (dbpoll(NULL, -1, &other, &sybres) != FAIL) {
		if (other == NULL)
			continue;
		dprintf("dbpoll returned %d for %p", sybres, other);
		if (dbresults(other) == NO_MORE_RESULTS) {
			if (other == one) {
				if (!twospoke)
					break;
			} else
				twospoke = 1;
			sybdrain(other == one ? two : one);
		}
	}
	if (other == NULL)
		dprintf("Other is %p. Strange (sybres is %d)", other, sybres);
#endif

	if (awaiting == 0) { /* Got a notice to run sp_post_xpload */
		fprintf(stderr, "%s: Server suggested, we run `%s', obeying\n",
		    __func__, "sp_post_xpload");
		sybstatements(two, postload, to, ip, ports, buf, sizeof buf);
		sybdrain(two);
	}

	return result;
}

int
main(int argc, char *argv[])
{
	char		*U = "sa", *P = NULL, *A = NULL, *S = NULL, *me = NULL;
	char		*U2 = "sa", *P2 = NULL, *S2 = NULL;
	const char	*db, *db2, *fname = NULL;
	int		(*processor)(DBPROCESS *, const char *db) = NULL;
	int		 opt;
	void		(*sigint)(int) = sighandler;
	unsigned short	 high=65000, low = 1025; /* port numbers range */
	DBPROCESS	*one, *two;

	A = basename(argv[0]);
	while ((opt = getopt(argc, argv, "n:T:f:z:j:dDLS:s:U:u:P:p:c:C:M:t")) != -1) {
		switch (opt) {
		case 'd':
			debug = 1;
			break;
		case 'f':
			fname = optarg;
			break;
		case 'z': case 'j':
			level = strtol(optarg, NULL, 0);
			if (level < 1 || level > 9) {
				fprintf(stderr, "%s: specified level must be "
				    "an integer from 1 to 9", optarg);
				exit(EX_USAGE);
			}
			if (opt == 'z')
				dumptype = GZIP;
			else
				dumptype = BZIP;
		case 'c':
			low = strtol(optarg, NULL, 0);
			if (low == 0 || low > high) {
				fprintf(stderr, "%s port should be an integer "
				    "%s or at %hd\n", "low", "under", high);
				exit(EX_USAGE);
			}
			break;
		case 'C':
			high = strtol(optarg, NULL, 0);
			if (high == 0 || low > high) {
				fprintf(stderr, "%s: %s port should be an integer "
				    "%s or at %hd\n", argv[0], "high", "over", low);
				exit(EX_USAGE);
			}
			break;
		case 'L':
			processor = loader;
			break;
		case 'D':
			processor = dumper;
			break;
		case 'M':
			me = optarg;
			break;
		case 'n':
			parallel = strtol(optarg, NULL, 0);
			if (parallel > 0 && parallel < 33)
				break;
			fprintf(stderr, "%s: -n switch expects a natural "
			    "number less or equal 32, you provided `%s'\n",
			    argv[0], optarg);
			exit(EX_USAGE);
		case 'P':
			P = optarg;
			break;
		case 'p':
			P2 = optarg;
			break;
		case 'S':
			S = optarg;
			dprintf("the primary DB-server is %s (%p)", S, S);
			break;
		case 's':
			S2 = optarg;
			break;
		case 'T':
			Timeout = atoi(optarg);
			dprintf("translating %s into %d seconds timeout",
			    optarg, Timeout);
			dbsettime(Timeout); /* twenty seconds timeout */
			Timeout *= 1000; /* translate into milliseconds */
			break;
		case 't':
			sigint = SIG_IGN;
			break;
		case 'U':
			U = optarg;
			break;
		case 'u':
			U2 = optarg;
			break;
		case 'h':
		default:
			fputs("No help for the wicked. Not yet...\n", stderr);
			exit(0);
		}
	
	}

	if (argc <= optind) {
		fprintf(stderr, "Database must be specified\n");
		exit(EX_USAGE);
	}

	/* Use handler to say goodbye to servers in case of a popular signal */
	signal(SIGINT, sigint);
	signal(SIGQUIT, sighandler);
	signal(SIGPIPE, sighandler);
	signal(SIGTERM, sighandler);
	signal(SIGALRM, sighandler);
	sybinit();
	atexit(dbexit);
	one = sybconnect(A, U, P, S);
	db = argv[optind];

	if (argc == optind + 1 || S2 == NULL) {
		if (S2) {
			fprintf(stderr, "The second database not specified\n");
			exit(EX_USAGE);
		}
		/*
		 * Either dump or a load of one database are specified...
		 */

		/* If the action is not specified, guess it from the program's name.*/
		if (processor == NULL)
			processor = strstr(A, "load") == NULL ? dumper : loader;

		/* If the filename is given (-f), open it, otherwise use stdin/out */
		if (processor == dumper) {
			if (fname == NULL)
				dump.pf = stdout;
			else {
				dump.pf = fopen(fname, "wb");
				if (dump.pf == NULL) {
					perror(fname);
					return EX_IOERR;
				}
			}
		} else {
			if (fname == NULL)
				dump.pf = stdin;
			else {
				dump.pf = fopen(fname, "rb");
				if (dump.pf == NULL) {
					perror(fname);
					return EX_NOINPUT;
				}
			}
		}

		if (level) {
			char mode[3] = { 'w' };
			if (processor != dumper) {
				if (dumptype == GZIP)
					goto gzip;
				else
					goto bzip;
			}
			mode[1] = level;
			if (dumptype == GZIP) {
				dump.gz = gzdopen(fileno(dump.pf), mode);
			} else  {
				dump.bz = BZ2_bzdopen(fileno(dump.pf), mode);
			}
			if (dump.pf == NULL) {
				perror("can not initialize compressor");
				exit(EX_IOERR);
			}
		} else if (processor == loader) {
			int	magic = 0;

			magic = getc(dump.pf);
			dprintf("Input's first byte is %x.", magic);
			magic = ungetc(magic, dump.pf);
			if (lseek(fileno(dump.pf), 0, SEEK_SET)) {
				if (errno == ESPIPE)
					seekable = 0;
				else {
					perror("lseek on input");
					exit(EX_NOINPUT);
				}
			} else
				seekable = 1;
			switch (magic) {
			case EOF:
				perror(fname ? fname : "stdin");
				exit(EX_NOINPUT);
			bzip:
			case 'B':
				dumptype = BZIP;
				dprintf("%s-ing input.", "Bunzip2");
				dump.bz = BZ2_bzdopen(fileno(dump.pf), "rb");
				break;
			gzip:
			case '\037':
				dumptype = GZIP;
				dump.gz = gzdopen(fileno(dump.pf), "rb");
				dprintf("%s-ing input.", "Gunzip");
				break;
			case 'V': /* Sybase dumps begin with "VOL" */
				dumptype = PLAIN;
				break;
			default:
				fprintf(stderr, "Unexpected first character "
				    "in the input.\n");
				exit(EX_NOINPUT);
			}
			if (dump.pf == NULL) {
				perror("can not initialize compressor");
				exit(EX_NOINPUT);
			}
			if (magic && !seekable && dumptype != PLAIN) {
				fprintf(stderr, "Automatic checking for "
				    "compression is currently broken for"
				    "non-seekable stdin.\nPlease, pipe it"
				    " through `%scat' or specify an -%c"
				    " option explicitly\n",
				    dumptype == GZIP ? "gz" : "bz",
				    dumptype == GZIP ? 'z' : 'j');
				exit(EX_USAGE);
			}
		}

		if (me == NULL) {
			me = alloca(MAXHOSTNAMELEN);
			if (gethostname(me, MAXHOSTNAMELEN)) {
				perror("Use -M switch: gethostname");
				exit(EX_UNAVAILABLE);
			}
		}
		dprintf("Will use name %s as my address.", me);
		opt = getaddrinfo(me, NULL, NULL, &ai);
		if (opt) {
			fprintf(stderr, "%s: can't figure `%s' out: %s\n", A, me,
			    gai_strerror(opt));
			exit(EX_NOHOST);
		}

		getsock(low, high);
		return processor(one, db);
	}

	/*
	 * Ok, direct transfer is requested.
	 * Parse the second set of command-line options
	 */

	db2 = argv[optind + 1];
	two = sybconnect(A, U2, P2, S2);

	return transfer(db, one, db2, two, low, high);
}