[deb_libnfs.git] / lib / socket.c

/*
   Copyright (C) 2010 by Ronnie Sahlberg <ronniesahlberg@gmail.com>

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU Lesser General Public License as published by
   the Free Software Foundation; either version 2.1 of the License, or
   (at your option) any later version.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU Lesser General Public License for more details.

   You should have received a copy of the GNU Lesser General Public License
   along with this program; if not, see <http://www.gnu.org/licenses/>.
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

#ifdef AROS
#include "aros_compat.h"
#endif

#ifdef WIN32
#include "win32_compat.h"
#endif

#ifdef HAVE_ARPA_INET_H
#include <arpa/inet.h>
#endif

#ifdef HAVE_POLL_H
#include <poll.h>
#endif

#ifdef HAVE_UNISTD_H
#include <unistd.h>
#endif

#ifdef HAVE_SYS_IOCTL_H
#include <sys/ioctl.h>
#endif

#ifdef HAVE_SYS_SOCKET_H
#include <sys/socket.h>
#endif

#ifdef HAVE_NETINET_TCP_H
#include <netinet/tcp.h>
#endif

#ifdef HAVE_NETDB_H
#include <netdb.h>
#endif

#ifdef HAVE_SYS_FILIO_H
#include <sys/filio.h>
#endif

#ifdef HAVE_SYS_SOCKIO_H
#include <sys/sockio.h>
#endif

#include <stdio.h>
#include <stdlib.h>
#include <assert.h>
#include <fcntl.h>
#include <string.h>
#include <errno.h>
#include <sys/types.h>
#include "libnfs-zdr.h"
#include "libnfs.h"
#include "libnfs-raw.h"
#include "libnfs-private.h"
#include "slist.h"

#ifdef WIN32
//has to be included after stdlib!!
#include "win32_errnowrapper.h"
#endif


static int rpc_reconnect_requeue(struct rpc_context *rpc);
static int rpc_connect_sockaddr_async(struct rpc_context *rpc, struct sockaddr_storage *s);

static void set_nonblocking(int fd)
{
	int v = 0;
#if defined(WIN32)
	long nonblocking=1;
	v = ioctl(fd, FIONBIO, &nonblocking);
#else
	v = fcntl(fd, F_GETFL, 0);
        fcntl(fd, F_SETFL, v | O_NONBLOCK);
#endif //FIXME
}

#ifdef HAVE_NETINET_TCP_H
int set_tcp_sockopt(int sockfd, int optname, int value)
{
	int level;

	#if defined(__FreeBSD__) || defined(__sun) || (defined(__APPLE__) && defined(__MACH__))
	struct protoent *buf;

	if ((buf = getprotobyname("tcp")) != NULL)
		level = buf->p_proto;
	else
		return -1;
	#else
		level = SOL_TCP;
	#endif

	return setsockopt(sockfd, level, optname, (char *)&value, sizeof(value));
}
#endif

int rpc_get_fd(struct rpc_context *rpc)
{
	assert(rpc->magic == RPC_CONTEXT_MAGIC);

	return rpc->fd;
}

int rpc_which_events(struct rpc_context *rpc)
{
	int events;

	assert(rpc->magic == RPC_CONTEXT_MAGIC);

	events = rpc->is_connected ? POLLIN : POLLOUT;

	if (rpc->is_udp != 0) {
		/* for udp sockets we only wait for pollin */
		return POLLIN;
	}

	if (rpc->outqueue) {
		events |= POLLOUT;
	}
	return events;
}

static int rpc_write_to_socket(struct rpc_context *rpc)
{
	int32_t count;

	assert(rpc->magic == RPC_CONTEXT_MAGIC);

	if (rpc->fd == -1) {
		rpc_set_error(rpc, "trying to write but not connected");
		return -1;
	}

	while (rpc->outqueue != NULL) {
		int64_t total;

		total = rpc->outqueue->outdata.size;

		count = send(rpc->fd, rpc->outqueue->outdata.data + rpc->outqueue->written, total - rpc->outqueue->written, 0);
		if (count == -1) {
			if (errno == EAGAIN || errno == EWOULDBLOCK) {
				return 0;
			}
			rpc_set_error(rpc, "Error when writing to socket :%s(%d)", strerror(errno), errno);
			return -1;
		}

		rpc->outqueue->written += count;
		if (rpc->outqueue->written == total) {
			struct rpc_pdu *pdu = rpc->outqueue;

	       	    	SLIST_REMOVE(&rpc->outqueue, pdu);
			SLIST_ADD_END(&rpc->waitpdu, pdu);
		}
	}
	return 0;
}

static int rpc_read_from_socket(struct rpc_context *rpc)
{
	int available;
	int size;
	int pdu_size;
	int32_t count;

	assert(rpc->magic == RPC_CONTEXT_MAGIC);

	if (ioctl(rpc->fd, FIONREAD, &available) != 0) {
		rpc_set_error(rpc, "Ioctl FIONREAD returned error : %d. Closing socket.", errno);
		return -1;
	}

	if (available == 0) {
		rpc_set_error(rpc, "Socket has been closed");
		return -1;
	}

	if (rpc->is_udp) {
		char *buf;
		socklen_t socklen = sizeof(rpc->udp_src);

		buf = malloc(available);
		if (buf == NULL) {
			rpc_set_error(rpc, "Failed to malloc buffer for recvfrom");
			return -1;
		}
		count = recvfrom(rpc->fd, buf, available, MSG_DONTWAIT, (struct sockaddr *)&rpc->udp_src, &socklen);
		if (count < 0) {
			rpc_set_error(rpc, "Failed recvfrom: %s", strerror(errno));
			free(buf);
		}
		if (rpc_process_pdu(rpc, buf, count) != 0) {
			rpc_set_error(rpc, "Invalid/garbage pdu received from server. Ignoring PDU");
			free(buf);
			return -1;
		}
		free(buf);
		return 0;
	}

	/* read record marker, 4 bytes at the beginning of every pdu */
	if (rpc->inbuf == NULL) {
		rpc->insize = 4;
		rpc->inbuf = malloc(rpc->insize);
		if (rpc->inbuf == NULL) {
			rpc_set_error(rpc, "Failed to allocate buffer for record marker, errno:%d. Closing socket.", errno);
			return -1;
		}
	}
	if (rpc->inpos < 4) {
		size = 4 - rpc->inpos;

		count = recv(rpc->fd, rpc->inbuf + rpc->inpos, size, 0);
		if (count == -1) {
			if (errno == EINTR) {
				return 0;
			}
			rpc_set_error(rpc, "Read from socket failed, errno:%d. Closing socket.", errno);
			return -1;
		}
		available  -= count;
		rpc->inpos += count;
	}

	if (available == 0) {
		return 0;
	}

	pdu_size = rpc_get_pdu_size(rpc->inbuf);
	if (rpc->insize < pdu_size) {
		unsigned char *buf;
		
		buf = malloc(pdu_size);
		if (buf == NULL) {
			rpc_set_error(rpc, "Failed to allocate buffer of %d bytes for pdu, errno:%d. Closing socket.", pdu_size, errno);
			return -1;
		}
		memcpy(buf, rpc->inbuf, rpc->insize);
		free(rpc->inbuf);
		rpc->inbuf  = buf;
		rpc->insize = rpc_get_pdu_size(rpc->inbuf);
	}

	size = available;
	if (size > rpc->insize - rpc->inpos) {
		size = rpc->insize - rpc->inpos;
	}

	count = recv(rpc->fd, rpc->inbuf + rpc->inpos, size, 0);
	if (count == -1) {
		if (errno == EINTR) {
			return 0;
		}
		rpc_set_error(rpc, "Read from socket failed, errno:%d. Closing socket.", errno);
		return -1;
	}
	available  -= count;
	rpc->inpos += count;

	if (rpc->inpos == rpc->insize) {
		char *buf = rpc->inbuf;

		rpc->inbuf  = NULL;
		rpc->insize = 0;
		rpc->inpos  = 0;

		if (rpc_process_pdu(rpc, buf, pdu_size) != 0) {
			rpc_set_error(rpc, "Invalid/garbage pdu received from server. Closing socket");
			return -1;
		}
		free(buf);
	}

	return 0;
}


int rpc_service(struct rpc_context *rpc, int revents)
{
	assert(rpc->magic == RPC_CONTEXT_MAGIC);

	if (revents & POLLERR) {
#ifdef WIN32
		char err = 0;
#else
		int err = 0;
#endif
		socklen_t err_size = sizeof(err);

		if (getsockopt(rpc->fd, SOL_SOCKET, SO_ERROR,
				(char *)&err, &err_size) != 0 || err != 0) {
			if (err == 0) {
				err = errno;
			}
			rpc_set_error(rpc, "rpc_service: socket error "
					       "%s(%d).",
					       strerror(err), err);
		} else {
			rpc_set_error(rpc, "rpc_service: POLLERR, "
						"Unknown socket error.");
		}
		if (rpc->connect_cb != NULL) {
			rpc->connect_cb(rpc, RPC_STATUS_ERROR, rpc->error_string, rpc->connect_data);
		}
		return -1;
	}
	if (revents & POLLHUP) {
		rpc_set_error(rpc, "Socket failed with POLLHUP");
		if (rpc->connect_cb != NULL) {
			rpc->connect_cb(rpc, RPC_STATUS_ERROR, rpc->error_string, rpc->connect_data);
		}
		return -1;
	}

	if (rpc->is_connected == 0 && rpc->fd != -1 && revents&POLLOUT) {
		int err = 0;
		socklen_t err_size = sizeof(err);

		if (getsockopt(rpc->fd, SOL_SOCKET, SO_ERROR,
				(char *)&err, &err_size) != 0 || err != 0) {
			if (err == 0) {
				err = errno;
			}
			rpc_set_error(rpc, "rpc_service: socket error "
				  	"%s(%d) while connecting.",
					strerror(err), err);
			if (rpc->connect_cb != NULL) {
				rpc->connect_cb(rpc, RPC_STATUS_ERROR,
					NULL, rpc->connect_data);
			}
			return -1;
		}

		rpc->is_connected = 1;
		if (rpc->connect_cb != NULL) {
			rpc->connect_cb(rpc, RPC_STATUS_SUCCESS, NULL, rpc->connect_data);
		}
		return 0;
	}

	if (revents & POLLIN) {
		if (rpc_read_from_socket(rpc) != 0) {
		  	rpc_reconnect_requeue(rpc);
			return 0;
		}
	}

	if (revents & POLLOUT && rpc->outqueue != NULL) {
		if (rpc_write_to_socket(rpc) != 0) {
			rpc_set_error(rpc, "write to socket failed");
			return -1;
		}
	}

	return 0;
}

void rpc_set_autoreconnect(struct rpc_context *rpc)
{
	assert(rpc->magic == RPC_CONTEXT_MAGIC);

	rpc->auto_reconnect = 1;
}

void rpc_unset_autoreconnect(struct rpc_context *rpc)
{
	assert(rpc->magic == RPC_CONTEXT_MAGIC);

	rpc->auto_reconnect = 0;
}

void rpc_set_tcp_syncnt(struct rpc_context *rpc, int v)
{
	assert(rpc->magic == RPC_CONTEXT_MAGIC);

	rpc->tcp_syncnt = v;
}

static int rpc_connect_sockaddr_async(struct rpc_context *rpc, struct sockaddr_storage *s)
{
	int socksize;

	assert(rpc->magic == RPC_CONTEXT_MAGIC);

	switch (s->ss_family) {
	case AF_INET:
		socksize = sizeof(struct sockaddr_in);
		rpc->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
#ifdef HAVE_NETINET_TCP_H
		if (rpc->tcp_syncnt != RPC_PARAM_UNDEFINED) {
			set_tcp_sockopt(rpc->fd, TCP_SYNCNT, rpc->tcp_syncnt);
		}
#endif
		break;
	default:
		rpc_set_error(rpc, "Can not handle AF_FAMILY:%d", s->ss_family);
		return -1;
	}

	if (rpc->fd == -1) {
		rpc_set_error(rpc, "Failed to open socket");
		return -1;
	}

	/* Some systems allow you to set capabilities on an executable
	 * to allow the file to be executed with privilege to bind to
	 * privileged system ports, even if the user is not root.
	 *
	 * Opportunistically try to bind the socket to a low numbered
	 * system port in the hope that the user is either root or the
	 * executable has the CAP_NET_BIND_SERVICE.
	 *
	 * As soon as we fail the bind() with EACCES we know we will never
	 * be able to bind to a system port so we terminate the loop.
	 *
	 * On linux, use
	 *    sudo setcap 'cap_net_bind_service=+ep' /path/executable
	 * to make the executable able to bind to a system port.
	 *
	 * On Windows, there is no concept of privileged ports. Thus
	 * binding will usually succeed.
	 */
	{
		struct sockaddr_in sin;
		static int portOfs = 0;
		const int firstPort = 512;	/* >= 512 according to Sun docs */
		const int portCount = IPPORT_RESERVED - firstPort;
		int startOfs, port, rc;

		if (portOfs == 0) {
			portOfs = time(NULL) % 400;
		}
		startOfs = portOfs;
		do {
			rc = -1;
			port = htons(firstPort + portOfs);
			portOfs = (portOfs + 1) % portCount;

			/* skip well-known ports */
			if (!getservbyport(port, "tcp")) {
				memset(&sin, 0, sizeof(sin));
				sin.sin_port        = port;
				sin.sin_family      = AF_INET;
				sin.sin_addr.s_addr = 0;

				rc = bind(rpc->fd, (struct sockaddr *)&sin, sizeof(struct sockaddr_in));
#if !defined(WIN32)
				/* we got EACCES, so don't try again */
				if (rc != 0 && errno == EACCES)
					break;
#endif
			}
		} while (rc != 0 && portOfs != startOfs);
	}

	set_nonblocking(rpc->fd);

	if (connect(rpc->fd, (struct sockaddr *)s, socksize) != 0 && errno != EINPROGRESS) {
		rpc_set_error(rpc, "connect() to server failed. %s(%d)", strerror(errno), errno);
		return -1;
	}		

	return 0;
}	    

int rpc_connect_async(struct rpc_context *rpc, const char *server, int port, rpc_cb cb, void *private_data)
{
	struct sockaddr_in *sin = (struct sockaddr_in *)&rpc->s;

	assert(rpc->magic == RPC_CONTEXT_MAGIC);

	if (rpc->fd != -1) {
		rpc_set_error(rpc, "Trying to connect while already connected");
		return -1;
	}

	if (rpc->is_udp != 0) {
		rpc_set_error(rpc, "Trying to connect on UDP socket");
		return -1;
	}

	rpc->auto_reconnect = 0;

	sin->sin_family = AF_INET;
	sin->sin_port   = htons(port);
	if (inet_pton(AF_INET, server, &sin->sin_addr) != 1) {
		rpc_set_error(rpc, "Not a valid server ip address");
		return -1;
	}


	switch (rpc->s.ss_family) {
	case AF_INET:
#ifdef HAVE_SOCKADDR_LEN
		sin->sin_len = sizeof(struct sockaddr_in);
#endif
		break;
	}

	rpc->connect_cb  = cb;
	rpc->connect_data = private_data;

	if (rpc_connect_sockaddr_async(rpc, &rpc->s) != 0) {
		return -1;
	}

	return 0;
}	    

int rpc_disconnect(struct rpc_context *rpc, char *error)
{
	assert(rpc->magic == RPC_CONTEXT_MAGIC);

	rpc_unset_autoreconnect(rpc);

	if (rpc->fd != -1) {
		close(rpc->fd);
	}
	rpc->fd  = -1;

	rpc->is_connected = 0;

	rpc_error_all_pdus(rpc, error);

	return 0;
}

static void reconnect_cb(struct rpc_context *rpc, int status, void *data _U_, void *private_data)
{
	assert(rpc->magic == RPC_CONTEXT_MAGIC);

	if (status != RPC_STATUS_SUCCESS) {
		rpc_error_all_pdus(rpc, "RPC ERROR: Failed to reconnect async");
		return;
	}

	rpc->is_connected = 1;
	rpc->connect_cb   = NULL;
}

/* disconnect but do not error all PDUs, just move pdus in-flight back to the outqueue and reconnect */
static int rpc_reconnect_requeue(struct rpc_context *rpc)
{
	struct rpc_pdu *pdu;

	assert(rpc->magic == RPC_CONTEXT_MAGIC);

	if (rpc->fd != -1) {
		close(rpc->fd);
	}
	rpc->fd  = -1;

	rpc->is_connected = 0;

	/* socket is closed so we will not get any replies to any commands
	 * in flight. Move them all over from the waitpdu queue back to the out queue
	 */
	for (pdu=rpc->waitpdu; pdu; pdu=pdu->next) {
		SLIST_REMOVE(&rpc->waitpdu, pdu);
		SLIST_ADD(&rpc->outqueue, pdu);
		/* we have to re-send the whole pdu again */
		pdu->written = 0;
	}

	if (rpc->auto_reconnect != 0) {
		rpc->connect_cb  = reconnect_cb;

		if (rpc_connect_sockaddr_async(rpc, &rpc->s) != 0) {
			rpc_error_all_pdus(rpc, "RPC ERROR: Failed to reconnect async");
			return -1;
		}
	}

	return 0;
}


int rpc_bind_udp(struct rpc_context *rpc, char *addr, int port)
{
	struct addrinfo *ai = NULL;
	char service[6];

	assert(rpc->magic == RPC_CONTEXT_MAGIC);

	if (rpc->is_udp == 0) {
		rpc_set_error(rpc, "Cant not bind UDP. Not UDP context");
		return -1;
	}

	sprintf(service, "%d", port);
	if (getaddrinfo(addr, service, NULL, &ai) != 0) {
		rpc_set_error(rpc, "Invalid address:%s. "
			"Can not resolv into IPv4/v6 structure.");
		return -1;
 	}

	switch(ai->ai_family) {
	case AF_INET:
		rpc->fd = socket(ai->ai_family, SOCK_DGRAM, 0);
		if (rpc->fd == -1) {
			rpc_set_error(rpc, "Failed to create UDP socket: %s", strerror(errno)); 
			freeaddrinfo(ai);
			return -1;
		}

		if (bind(rpc->fd, (struct sockaddr *)ai->ai_addr, sizeof(struct sockaddr_in)) != 0) {
			rpc_set_error(rpc, "Failed to bind to UDP socket: %s",strerror(errno)); 
			freeaddrinfo(ai);
			return -1;
		}
		break;
	default:
		rpc_set_error(rpc, "Can not handle UPD sockets of family %d yet", ai->ai_family);
		freeaddrinfo(ai);
		return -1;
	}

	freeaddrinfo(ai);

	return 0;
}

int rpc_set_udp_destination(struct rpc_context *rpc, char *addr, int port, int is_broadcast)
{
	struct addrinfo *ai = NULL;
	char service[6];

	assert(rpc->magic == RPC_CONTEXT_MAGIC);

	if (rpc->is_udp == 0) {
		rpc_set_error(rpc, "Can not set destination sockaddr. Not UDP context");
		return -1;
	}

	sprintf(service, "%d", port);
	if (getaddrinfo(addr, service, NULL, &ai) != 0) {
		rpc_set_error(rpc, "Invalid address:%s. "
			"Can not resolv into IPv4/v6 structure.");
		return -1;
 	}

	if (rpc->udp_dest) {
		free(rpc->udp_dest);
		rpc->udp_dest = NULL;
	}
	rpc->udp_dest = malloc(ai->ai_addrlen);
	if (rpc->udp_dest == NULL) {
		rpc_set_error(rpc, "Out of memory. Failed to allocate sockaddr structure");
		freeaddrinfo(ai);
		return -1;
	}
	memcpy(rpc->udp_dest, ai->ai_addr, ai->ai_addrlen);
	freeaddrinfo(ai);

	rpc->is_broadcast = is_broadcast;
	setsockopt(rpc->fd, SOL_SOCKET, SO_BROADCAST, (char *)&is_broadcast, sizeof(is_broadcast));

	return 0;
}

struct sockaddr *rpc_get_recv_sockaddr(struct rpc_context *rpc)
{
	assert(rpc->magic == RPC_CONTEXT_MAGIC);

	return (struct sockaddr *)&rpc->udp_src;
}

int rpc_queue_length(struct rpc_context *rpc)
{
	int i=0;
	struct rpc_pdu *pdu;

	assert(rpc->magic == RPC_CONTEXT_MAGIC);

	for(pdu = rpc->outqueue; pdu; pdu = pdu->next) {
		i++;
	}
	for(pdu = rpc->waitpdu; pdu; pdu = pdu->next) {
		i++;
	}
	return i;
}
Commit	Line	Data
84004dbf RS	1	/*
	2	Copyright (C) 2010 by Ronnie Sahlberg <ronniesahlberg@gmail.com>
	3
	4	This program is free software; you can redistribute it and/or modify
	5	it under the terms of the GNU Lesser General Public License as published by
	6	the Free Software Foundation; either version 2.1 of the License, or
	7	(at your option) any later version.
	8
	9	This program is distributed in the hope that it will be useful,
	10	but WITHOUT ANY WARRANTY; without even the implied warranty of
	11	MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	12	GNU Lesser General Public License for more details.
	13
	14	You should have received a copy of the GNU Lesser General Public License
	15	along with this program; if not, see <http://www.gnu.org/licenses/>.
	16	*/
00748f36 RS	17	#ifdef HAVE_CONFIG_H
	18	#include "config.h"
	19	#endif
	20
	21	#ifdef AROS
	22	#include "aros_compat.h"
	23	#endif
	24
a8a1b858 M	25	#ifdef WIN32
a8a1b858 M	26	#include "win32_compat.h"
c022471e RS	27	#endif
	28
	29	#ifdef HAVE_ARPA_INET_H
a8a1b858	30	#include <arpa/inet.h>
c022471e	31	#endif
84004dbf	32
00748f36 RS	33	#ifdef HAVE_POLL_H
00748f36 RS	34	#include <poll.h>
fc01d2a9	35	#endif
d7c6e9aa	36
00748f36 RS	37	#ifdef HAVE_UNISTD_H
00748f36 RS	38	#include <unistd.h>
d7c6e9aa RS	39	#endif
d7c6e9aa RS	40
00748f36 RS	41	#ifdef HAVE_SYS_IOCTL_H
00748f36 RS	42	#include <sys/ioctl.h>
d7c6e9aa RS	43	#endif
d7c6e9aa RS	44
7057e733 RS	45	#ifdef HAVE_SYS_SOCKET_H
	46	#include <sys/socket.h>
	47	#endif
	48
1c8b4547 PL	49	#ifdef HAVE_NETINET_TCP_H
	50	#include <netinet/tcp.h>
	51	#endif
	52
bff8fe46 RS	53	#ifdef HAVE_NETDB_H
	54	#include <netdb.h>
	55	#endif
	56
fc01d2a9 TN	57	#ifdef HAVE_SYS_FILIO_H
	58	#include <sys/filio.h>
	59	#endif
bff8fe46	60
647d2ea1 RS	61	#ifdef HAVE_SYS_SOCKIO_H
	62	#include <sys/sockio.h>
	63	#endif
bff8fe46 RS	64
	65	#include <stdio.h>
	66	#include <stdlib.h>
	67	#include <assert.h>
	68	#include <fcntl.h>
	69	#include <string.h>
	70	#include <errno.h>
485bc9b9	71	#include <sys/types.h>
763cd6e3	72	#include "libnfs-zdr.h"
84004dbf RS	73	#include "libnfs.h"
	74	#include "libnfs-raw.h"
	75	#include "libnfs-private.h"
	76	#include "slist.h"
	77
fcc42bfe M	78	#ifdef WIN32
	79	//has to be included after stdlib!!
	80	#include "win32_errnowrapper.h"
	81	#endif
	82
	83
1744ef90 RS	84	static int rpc_reconnect_requeue(struct rpc_context *rpc);
1744ef90 RS	85	static int rpc_connect_sockaddr_async(struct rpc_context rpc, struct sockaddr_storage s);
b077fdeb	86
84004dbf RS	87	static void set_nonblocking(int fd)
84004dbf RS	88	{
99c14c9b	89	int v = 0;
6874f61e	90	#if defined(WIN32)
99c14c9b	91	long nonblocking=1;
622489d3	92	v = ioctl(fd, FIONBIO, &nonblocking);
6874f61e	93	#else
84004dbf RS	94	v = fcntl(fd, F_GETFL, 0);
84004dbf RS	95	fcntl(fd, F_SETFL, v \| O_NONBLOCK);
a8a1b858	96	#endif //FIXME
84004dbf RS	97	}
84004dbf RS	98
1c8b4547 PL	99	#ifdef HAVE_NETINET_TCP_H
	100	int set_tcp_sockopt(int sockfd, int optname, int value)
	101	{
	102	int level;
	103
	104	#if defined(__FreeBSD__) \|\| defined(__sun) \|\| (defined(__APPLE__) && defined(__MACH__))
	105	struct protoent *buf;
	106
	107	if ((buf = getprotobyname("tcp")) != NULL)
	108	level = buf->p_proto;
	109	else
	110	return -1;
	111	#else
	112	level = SOL_TCP;
	113	#endif
	114
	115	return setsockopt(sockfd, level, optname, (char *)&value, sizeof(value));
	116	}
	117	#endif
	118
84004dbf RS	119	int rpc_get_fd(struct rpc_context *rpc)
84004dbf RS	120	{
4a2b0876 RS	121	assert(rpc->magic == RPC_CONTEXT_MAGIC);
4a2b0876 RS	122
84004dbf RS	123	return rpc->fd;
	124	}
	125
	126	int rpc_which_events(struct rpc_context *rpc)
	127	{
4a2b0876 RS	128	int events;
	129
	130	assert(rpc->magic == RPC_CONTEXT_MAGIC);
	131
	132	events = rpc->is_connected ? POLLIN : POLLOUT;
84004dbf	133
5911f3e8 RS	134	if (rpc->is_udp != 0) {
	135	/* for udp sockets we only wait for pollin */
	136	return POLLIN;
	137	}
	138
84004dbf RS	139	if (rpc->outqueue) {
	140	events \|= POLLOUT;
	141	}
	142	return events;
	143	}
	144
	145	static int rpc_write_to_socket(struct rpc_context *rpc)
	146	{
d14e2838	147	int32_t count;
84004dbf	148
4a2b0876 RS	149	assert(rpc->magic == RPC_CONTEXT_MAGIC);
4a2b0876 RS	150
84004dbf	151	if (rpc->fd == -1) {
1896d37b RS	152	rpc_set_error(rpc, "trying to write but not connected");
1896d37b RS	153	return -1;
84004dbf RS	154	}
	155
	156	while (rpc->outqueue != NULL) {
183451cf	157	int64_t total;
84004dbf RS	158
	159	total = rpc->outqueue->outdata.size;
	160
6874f61e	161	count = send(rpc->fd, rpc->outqueue->outdata.data + rpc->outqueue->written, total - rpc->outqueue->written, 0);
84004dbf RS	162	if (count == -1) {
84004dbf RS	163	if (errno == EAGAIN \|\| errno == EWOULDBLOCK) {
84004dbf RS	164	return 0;
84004dbf RS	165	}
1896d37b RS	166	rpc_set_error(rpc, "Error when writing to socket :%s(%d)", strerror(errno), errno);
1896d37b RS	167	return -1;
84004dbf RS	168	}
	169
	170	rpc->outqueue->written += count;
	171	if (rpc->outqueue->written == total) {
	172	struct rpc_pdu *pdu = rpc->outqueue;
	173
	174	SLIST_REMOVE(&rpc->outqueue, pdu);
	175	SLIST_ADD_END(&rpc->waitpdu, pdu);
	176	}
	177	}
	178	return 0;
	179	}
	180
	181	static int rpc_read_from_socket(struct rpc_context *rpc)
	182	{
	183	int available;
	184	int size;
cdb19ec1	185	int pdu_size;
d14e2838	186	int32_t count;
84004dbf	187
f3a75078	188	assert(rpc->magic == RPC_CONTEXT_MAGIC);
84004dbf RS	189
	190	if (ioctl(rpc->fd, FIONREAD, &available) != 0) {
	191	rpc_set_error(rpc, "Ioctl FIONREAD returned error : %d. Closing socket.", errno);
	192	return -1;
	193	}
a8a1b858	194
84004dbf RS	195	if (available == 0) {
84004dbf RS	196	rpc_set_error(rpc, "Socket has been closed");
1896d37b	197	return -1;
84004dbf	198	}
cdb19ec1	199
0268794f RS	200	if (rpc->is_udp) {
	201	char *buf;
	202	socklen_t socklen = sizeof(rpc->udp_src);
	203
	204	buf = malloc(available);
	205	if (buf == NULL) {
	206	rpc_set_error(rpc, "Failed to malloc buffer for recvfrom");
	207	return -1;
	208	}
	209	count = recvfrom(rpc->fd, buf, available, MSG_DONTWAIT, (struct sockaddr *)&rpc->udp_src, &socklen);
	210	if (count < 0) {
	211	rpc_set_error(rpc, "Failed recvfrom: %s", strerror(errno));
	212	free(buf);
	213	}
	214	if (rpc_process_pdu(rpc, buf, count) != 0) {
	215	rpc_set_error(rpc, "Invalid/garbage pdu received from server. Ignoring PDU");
	216	free(buf);
	217	return -1;
	218	}
	219	free(buf);
	220	return 0;
	221	}
	222
cdb19ec1 RS	223	/* read record marker, 4 bytes at the beginning of every pdu */
	224	if (rpc->inbuf == NULL) {
	225	rpc->insize = 4;
	226	rpc->inbuf = malloc(rpc->insize);
	227	if (rpc->inbuf == NULL) {
	228	rpc_set_error(rpc, "Failed to allocate buffer for record marker, errno:%d. Closing socket.", errno);
	229	return -1;
	230	}
	231	}
	232	if (rpc->inpos < 4) {
	233	size = 4 - rpc->inpos;
	234
6874f61e	235	count = recv(rpc->fd, rpc->inbuf + rpc->inpos, size, 0);
cdb19ec1 RS	236	if (count == -1) {
	237	if (errno == EINTR) {
	238	return 0;
	239	}
	240	rpc_set_error(rpc, "Read from socket failed, errno:%d. Closing socket.", errno);
	241	return -1;
	242	}
	243	available -= count;
	244	rpc->inpos += count;
	245	}
	246
	247	if (available == 0) {
	248	return 0;
	249	}
	250
	251	pdu_size = rpc_get_pdu_size(rpc->inbuf);
	252	if (rpc->insize < pdu_size) {
	253	unsigned char *buf;
	254
	255	buf = malloc(pdu_size);
	256	if (buf == NULL) {
	257	rpc_set_error(rpc, "Failed to allocate buffer of %d bytes for pdu, errno:%d. Closing socket.", pdu_size, errno);
	258	return -1;
	259	}
	260	memcpy(buf, rpc->inbuf, rpc->insize);
	261	free(rpc->inbuf);
	262	rpc->inbuf = buf;
	263	rpc->insize = rpc_get_pdu_size(rpc->inbuf);
84004dbf	264	}
cdb19ec1 RS	265
	266	size = available;
	267	if (size > rpc->insize - rpc->inpos) {
	268	size = rpc->insize - rpc->inpos;
84004dbf RS	269	}
84004dbf RS	270
6874f61e	271	count = recv(rpc->fd, rpc->inbuf + rpc->inpos, size, 0);
84004dbf RS	272	if (count == -1) {
84004dbf RS	273	if (errno == EINTR) {
84004dbf RS	274	return 0;
	275	}
	276	rpc_set_error(rpc, "Read from socket failed, errno:%d. Closing socket.", errno);
1896d37b	277	return -1;
84004dbf	278	}
cdb19ec1 RS	279	available -= count;
cdb19ec1 RS	280	rpc->inpos += count;
84004dbf	281
cdb19ec1	282	if (rpc->inpos == rpc->insize) {
dd97d43a RS	283	char *buf = rpc->inbuf;
	284
	285	rpc->inbuf = NULL;
	286	rpc->insize = 0;
	287	rpc->inpos = 0;
	288
	289	if (rpc_process_pdu(rpc, buf, pdu_size) != 0) {
84004dbf	290	rpc_set_error(rpc, "Invalid/garbage pdu received from server. Closing socket");
1896d37b	291	return -1;
84004dbf	292	}
751770fd	293	free(buf);
84004dbf	294	}
cdb19ec1	295
84004dbf RS	296	return 0;
	297	}
	298
	299
	300
	301	int rpc_service(struct rpc_context *rpc, int revents)
	302	{
4a2b0876 RS	303	assert(rpc->magic == RPC_CONTEXT_MAGIC);
4a2b0876 RS	304
84004dbf	305	if (revents & POLLERR) {
fcc42bfe	306	#ifdef WIN32
a8a1b858	307	char err = 0;
fcc42bfe M	308	#else
	309	int err = 0;
	310	#endif
912f7ad5 RS	311	socklen_t err_size = sizeof(err);
	312
	313	if (getsockopt(rpc->fd, SOL_SOCKET, SO_ERROR,
bb4e9ed6	314	(char *)&err, &err_size) != 0 \|\| err != 0) {
912f7ad5 RS	315	if (err == 0) {
	316	err = errno;
	317	}
	318	rpc_set_error(rpc, "rpc_service: socket error "
	319	"%s(%d).",
	320	strerror(err), err);
84004dbf	321	} else {
912f7ad5 RS	322	rpc_set_error(rpc, "rpc_service: POLLERR, "
912f7ad5 RS	323	"Unknown socket error.");
84004dbf	324	}
b990de23 RS	325	if (rpc->connect_cb != NULL) {
	326	rpc->connect_cb(rpc, RPC_STATUS_ERROR, rpc->error_string, rpc->connect_data);
	327	}
84004dbf RS	328	return -1;
	329	}
	330	if (revents & POLLHUP) {
84004dbf	331	rpc_set_error(rpc, "Socket failed with POLLHUP");
b990de23 RS	332	if (rpc->connect_cb != NULL) {
	333	rpc->connect_cb(rpc, RPC_STATUS_ERROR, rpc->error_string, rpc->connect_data);
	334	}
1896d37b	335	return -1;
84004dbf RS	336	}
	337
	338	if (rpc->is_connected == 0 && rpc->fd != -1 && revents&POLLOUT) {
912f7ad5 RS	339	int err = 0;
	340	socklen_t err_size = sizeof(err);
	341
	342	if (getsockopt(rpc->fd, SOL_SOCKET, SO_ERROR,
bb4e9ed6	343	(char *)&err, &err_size) != 0 \|\| err != 0) {
912f7ad5 RS	344	if (err == 0) {
	345	err = errno;
	346	}
	347	rpc_set_error(rpc, "rpc_service: socket error "
	348	"%s(%d) while connecting.",
	349	strerror(err), err);
b990de23 RS	350	if (rpc->connect_cb != NULL) {
b990de23 RS	351	rpc->connect_cb(rpc, RPC_STATUS_ERROR,
912f7ad5	352	NULL, rpc->connect_data);
b990de23	353	}
912f7ad5 RS	354	return -1;
	355	}
	356
84004dbf	357	rpc->is_connected = 1;
b990de23 RS	358	if (rpc->connect_cb != NULL) {
	359	rpc->connect_cb(rpc, RPC_STATUS_SUCCESS, NULL, rpc->connect_data);
	360	}
84004dbf RS	361	return 0;
	362	}
	363
b077fdeb RS	364	if (revents & POLLIN) {
b077fdeb RS	365	if (rpc_read_from_socket(rpc) != 0) {
1744ef90	366	rpc_reconnect_requeue(rpc);
b077fdeb	367	return 0;
84004dbf RS	368	}
	369	}
	370
b077fdeb RS	371	if (revents & POLLOUT && rpc->outqueue != NULL) {
	372	if (rpc_write_to_socket(rpc) != 0) {
	373	rpc_set_error(rpc, "write to socket failed");
1896d37b	374	return -1;
84004dbf RS	375	}
	376	}
	377
	378	return 0;
	379	}
	380
1744ef90	381	void rpc_set_autoreconnect(struct rpc_context *rpc)
84004dbf	382	{
4a2b0876 RS	383	assert(rpc->magic == RPC_CONTEXT_MAGIC);
4a2b0876 RS	384
1744ef90 RS	385	rpc->auto_reconnect = 1;
1744ef90 RS	386	}
84004dbf	387
1744ef90 RS	388	void rpc_unset_autoreconnect(struct rpc_context *rpc)
1744ef90 RS	389	{
4a2b0876 RS	390	assert(rpc->magic == RPC_CONTEXT_MAGIC);
4a2b0876 RS	391
1744ef90 RS	392	rpc->auto_reconnect = 0;
1744ef90 RS	393	}
070287e5	394
1c8b4547 PL	395	void rpc_set_tcp_syncnt(struct rpc_context *rpc, int v)
	396	{
	397	assert(rpc->magic == RPC_CONTEXT_MAGIC);
	398
	399	rpc->tcp_syncnt = v;
	400	}
	401
1744ef90 RS	402	static int rpc_connect_sockaddr_async(struct rpc_context rpc, struct sockaddr_storage s)
	403	{
	404	int socksize;
84004dbf	405
4a2b0876 RS	406	assert(rpc->magic == RPC_CONTEXT_MAGIC);
4a2b0876 RS	407
1744ef90	408	switch (s->ss_family) {
84004dbf	409	case AF_INET:
84004dbf	410	socksize = sizeof(struct sockaddr_in);
6874f61e	411	rpc->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
1c8b4547 PL	412	#ifdef HAVE_NETINET_TCP_H
	413	if (rpc->tcp_syncnt != RPC_PARAM_UNDEFINED) {
	414	set_tcp_sockopt(rpc->fd, TCP_SYNCNT, rpc->tcp_syncnt);
	415	}
	416	#endif
84004dbf	417	break;
1744ef90 RS	418	default:
	419	rpc_set_error(rpc, "Can not handle AF_FAMILY:%d", s->ss_family);
	420	return -1;
84004dbf RS	421	}
	422
	423	if (rpc->fd == -1) {
	424	rpc_set_error(rpc, "Failed to open socket");
1896d37b	425	return -1;
84004dbf RS	426	}
84004dbf RS	427
07fd0cbc RS	428	/* Some systems allow you to set capabilities on an executable
	429	* to allow the file to be executed with privilege to bind to
	430	* privileged system ports, even if the user is not root.
	431	*
	432	* Opportunistically try to bind the socket to a low numbered
	433	* system port in the hope that the user is either root or the
	434	* executable has the CAP_NET_BIND_SERVICE.
	435	*
	436	* As soon as we fail the bind() with EACCES we know we will never
	437	* be able to bind to a system port so we terminate the loop.
	438	*
	439	* On linux, use
	440	* sudo setcap 'cap_net_bind_service=+ep' /path/executable
	441	* to make the executable able to bind to a system port.
ac559609 RI	442	*
	443	* On Windows, there is no concept of privileged ports. Thus
	444	* binding will usually succeed.
07fd0cbc	445	*/
ac559609 RI	446	{
	447	struct sockaddr_in sin;
	448	static int portOfs = 0;
	449	const int firstPort = 512; /* >= 512 according to Sun docs */
	450	const int portCount = IPPORT_RESERVED - firstPort;
cb5b8be2	451	int startOfs, port, rc;
ac559609	452
cb5b8be2 RS	453	if (portOfs == 0) {
	454	portOfs = time(NULL) % 400;
	455	}
	456	startOfs = portOfs;
ac559609 RI	457	do {
	458	rc = -1;
	459	port = htons(firstPort + portOfs);
	460	portOfs = (portOfs + 1) % portCount;
	461
	462	/* skip well-known ports */
	463	if (!getservbyport(port, "tcp")) {
	464	memset(&sin, 0, sizeof(sin));
	465	sin.sin_port = port;
	466	sin.sin_family = AF_INET;
	467	sin.sin_addr.s_addr = 0;
	468
	469	rc = bind(rpc->fd, (struct sockaddr *)&sin, sizeof(struct sockaddr_in));
	470	#if !defined(WIN32)
	471	/* we got EACCES, so don't try again */
	472	if (rc != 0 && errno == EACCES)
	473	break;
	474	#endif
07fd0cbc	475	}
ac559609	476	} while (rc != 0 && portOfs != startOfs);
07fd0cbc	477	}
07fd0cbc	478
84004dbf	479	set_nonblocking(rpc->fd);
07fd0cbc	480
f96b24fa RI	481	if (connect(rpc->fd, (struct sockaddr *)s, socksize) != 0 && errno != EINPROGRESS) {
f96b24fa RI	482	rpc_set_error(rpc, "connect() to server failed. %s(%d)", strerror(errno), errno);
1896d37b	483	return -1;
84004dbf RS	484	}
	485
	486	return 0;
	487	}
	488
1744ef90 RS	489	int rpc_connect_async(struct rpc_context rpc, const char server, int port, rpc_cb cb, void *private_data)
	490	{
	491	struct sockaddr_in sin = (struct sockaddr_in )&rpc->s;
	492
4a2b0876 RS	493	assert(rpc->magic == RPC_CONTEXT_MAGIC);
4a2b0876 RS	494
1744ef90 RS	495	if (rpc->fd != -1) {
	496	rpc_set_error(rpc, "Trying to connect while already connected");
	497	return -1;
	498	}
	499
	500	if (rpc->is_udp != 0) {
	501	rpc_set_error(rpc, "Trying to connect on UDP socket");
	502	return -1;
	503	}
	504
	505	rpc->auto_reconnect = 0;
	506
	507	sin->sin_family = AF_INET;
	508	sin->sin_port = htons(port);
	509	if (inet_pton(AF_INET, server, &sin->sin_addr) != 1) {
	510	rpc_set_error(rpc, "Not a valid server ip address");
	511	return -1;
	512	}
	513
	514
	515	switch (rpc->s.ss_family) {
	516	case AF_INET:
	517	#ifdef HAVE_SOCKADDR_LEN
	518	sin->sin_len = sizeof(struct sockaddr_in);
	519	#endif
	520	break;
	521	}
	522
	523	rpc->connect_cb = cb;
	524	rpc->connect_data = private_data;
	525
	526	if (rpc_connect_sockaddr_async(rpc, &rpc->s) != 0) {
	527	return -1;
	528	}
	529
	530	return 0;
	531	}
	532
84004dbf RS	533	int rpc_disconnect(struct rpc_context rpc, char error)
84004dbf RS	534	{
4a2b0876 RS	535	assert(rpc->magic == RPC_CONTEXT_MAGIC);
4a2b0876 RS	536
1744ef90 RS	537	rpc_unset_autoreconnect(rpc);
1744ef90 RS	538
84004dbf RS	539	if (rpc->fd != -1) {
	540	close(rpc->fd);
	541	}
	542	rpc->fd = -1;
	543
	544	rpc->is_connected = 0;
	545
	546	rpc_error_all_pdus(rpc, error);
	547
	548	return 0;
	549	}
485bc9b9	550
1744ef90 RS	551	static void reconnect_cb(struct rpc_context rpc, int status, void data _U_, void *private_data)
1744ef90 RS	552	{
4a2b0876 RS	553	assert(rpc->magic == RPC_CONTEXT_MAGIC);
4a2b0876 RS	554
1744ef90 RS	555	if (status != RPC_STATUS_SUCCESS) {
	556	rpc_error_all_pdus(rpc, "RPC ERROR: Failed to reconnect async");
	557	return;
	558	}
	559
	560	rpc->is_connected = 1;
b990de23	561	rpc->connect_cb = NULL;
1744ef90 RS	562	}
	563
	564	/* disconnect but do not error all PDUs, just move pdus in-flight back to the outqueue and reconnect */
	565	static int rpc_reconnect_requeue(struct rpc_context *rpc)
b077fdeb RS	566	{
	567	struct rpc_pdu *pdu;
	568
4a2b0876 RS	569	assert(rpc->magic == RPC_CONTEXT_MAGIC);
4a2b0876 RS	570
b077fdeb RS	571	if (rpc->fd != -1) {
	572	close(rpc->fd);
	573	}
	574	rpc->fd = -1;
	575
	576	rpc->is_connected = 0;
	577
	578	/* socket is closed so we will not get any replies to any commands
	579	* in flight. Move them all over from the waitpdu queue back to the out queue
	580	*/
	581	for (pdu=rpc->waitpdu; pdu; pdu=pdu->next) {
	582	SLIST_REMOVE(&rpc->waitpdu, pdu);
	583	SLIST_ADD(&rpc->outqueue, pdu);
1744ef90 RS	584	/* we have to re-send the whole pdu again */
	585	pdu->written = 0;
	586	}
	587
	588	if (rpc->auto_reconnect != 0) {
	589	rpc->connect_cb = reconnect_cb;
	590
	591	if (rpc_connect_sockaddr_async(rpc, &rpc->s) != 0) {
	592	rpc_error_all_pdus(rpc, "RPC ERROR: Failed to reconnect async");
	593	return -1;
	594	}
b077fdeb RS	595	}
	596
	597	return 0;
	598	}
	599
485bc9b9 RS	600
	601	int rpc_bind_udp(struct rpc_context rpc, char addr, int port)
	602	{
	603	struct addrinfo *ai = NULL;
	604	char service[6];
	605
4a2b0876 RS	606	assert(rpc->magic == RPC_CONTEXT_MAGIC);
4a2b0876 RS	607
485bc9b9 RS	608	if (rpc->is_udp == 0) {
	609	rpc_set_error(rpc, "Cant not bind UDP. Not UDP context");
	610	return -1;
	611	}
	612
6874f61e	613	sprintf(service, "%d", port);
485bc9b9 RS	614	if (getaddrinfo(addr, service, NULL, &ai) != 0) {
	615	rpc_set_error(rpc, "Invalid address:%s. "
	616	"Can not resolv into IPv4/v6 structure.");
	617	return -1;
	618	}
	619
	620	switch(ai->ai_family) {
	621	case AF_INET:
	622	rpc->fd = socket(ai->ai_family, SOCK_DGRAM, 0);
	623	if (rpc->fd == -1) {
	624	rpc_set_error(rpc, "Failed to create UDP socket: %s", strerror(errno));
	625	freeaddrinfo(ai);
	626	return -1;
	627	}
	628
	629	if (bind(rpc->fd, (struct sockaddr *)ai->ai_addr, sizeof(struct sockaddr_in)) != 0) {
	630	rpc_set_error(rpc, "Failed to bind to UDP socket: %s",strerror(errno));
	631	freeaddrinfo(ai);
	632	return -1;
	633	}
	634	break;
	635	default:
	636	rpc_set_error(rpc, "Can not handle UPD sockets of family %d yet", ai->ai_family);
	637	freeaddrinfo(ai);
	638	return -1;
	639	}
	640
	641	freeaddrinfo(ai);
	642
	643	return 0;
	644	}
	645
5bf60dc6 RS	646	int rpc_set_udp_destination(struct rpc_context rpc, char addr, int port, int is_broadcast)
	647	{
	648	struct addrinfo *ai = NULL;
	649	char service[6];
	650
4a2b0876 RS	651	assert(rpc->magic == RPC_CONTEXT_MAGIC);
4a2b0876 RS	652
5bf60dc6 RS	653	if (rpc->is_udp == 0) {
	654	rpc_set_error(rpc, "Can not set destination sockaddr. Not UDP context");
	655	return -1;
	656	}
	657
6874f61e	658	sprintf(service, "%d", port);
5bf60dc6 RS	659	if (getaddrinfo(addr, service, NULL, &ai) != 0) {
	660	rpc_set_error(rpc, "Invalid address:%s. "
	661	"Can not resolv into IPv4/v6 structure.");
	662	return -1;
	663	}
	664
	665	if (rpc->udp_dest) {
	666	free(rpc->udp_dest);
	667	rpc->udp_dest = NULL;
	668	}
	669	rpc->udp_dest = malloc(ai->ai_addrlen);
	670	if (rpc->udp_dest == NULL) {
	671	rpc_set_error(rpc, "Out of memory. Failed to allocate sockaddr structure");
be7f5933	672	freeaddrinfo(ai);
5bf60dc6 RS	673	return -1;
	674	}
	675	memcpy(rpc->udp_dest, ai->ai_addr, ai->ai_addrlen);
	676	freeaddrinfo(ai);
	677
	678	rpc->is_broadcast = is_broadcast;
bb4e9ed6	679	setsockopt(rpc->fd, SOL_SOCKET, SO_BROADCAST, (char *)&is_broadcast, sizeof(is_broadcast));
5bf60dc6 RS	680
	681	return 0;
	682	}
c481da67 RS	683
	684	struct sockaddr rpc_get_recv_sockaddr(struct rpc_context rpc)
	685	{
4a2b0876 RS	686	assert(rpc->magic == RPC_CONTEXT_MAGIC);
4a2b0876 RS	687
c481da67 RS	688	return (struct sockaddr *)&rpc->udp_src;
c481da67 RS	689	}
83aa785d RS	690
	691	int rpc_queue_length(struct rpc_context *rpc)
	692	{
	693	int i=0;
	694	struct rpc_pdu *pdu;
	695
4a2b0876 RS	696	assert(rpc->magic == RPC_CONTEXT_MAGIC);
4a2b0876 RS	697
83aa785d RS	698	for(pdu = rpc->outqueue; pdu; pdu = pdu->next) {
	699	i++;
	700	}
	701	for(pdu = rpc->waitpdu; pdu; pdu = pdu->next) {
	702	i++;
	703	}
	704	return i;
	705	}