[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;
}

#ifndef TCP_SYNCNT
#define TCP_SYNCNT        7
#endif

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
d43a8953 PL	402	#ifndef TCP_SYNCNT
	403	#define TCP_SYNCNT 7
	404	#endif
	405
1744ef90 RS	406	static int rpc_connect_sockaddr_async(struct rpc_context rpc, struct sockaddr_storage s)
	407	{
	408	int socksize;
84004dbf	409
4a2b0876 RS	410	assert(rpc->magic == RPC_CONTEXT_MAGIC);
4a2b0876 RS	411
1744ef90	412	switch (s->ss_family) {
84004dbf	413	case AF_INET:
84004dbf	414	socksize = sizeof(struct sockaddr_in);
6874f61e	415	rpc->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
1c8b4547 PL	416	#ifdef HAVE_NETINET_TCP_H
	417	if (rpc->tcp_syncnt != RPC_PARAM_UNDEFINED) {
	418	set_tcp_sockopt(rpc->fd, TCP_SYNCNT, rpc->tcp_syncnt);
	419	}
	420	#endif
84004dbf	421	break;
1744ef90 RS	422	default:
	423	rpc_set_error(rpc, "Can not handle AF_FAMILY:%d", s->ss_family);
	424	return -1;
84004dbf RS	425	}
	426
	427	if (rpc->fd == -1) {
	428	rpc_set_error(rpc, "Failed to open socket");
1896d37b	429	return -1;
84004dbf RS	430	}
84004dbf RS	431
07fd0cbc RS	432	/* Some systems allow you to set capabilities on an executable
	433	* to allow the file to be executed with privilege to bind to
	434	* privileged system ports, even if the user is not root.
	435	*
	436	* Opportunistically try to bind the socket to a low numbered
	437	* system port in the hope that the user is either root or the
	438	* executable has the CAP_NET_BIND_SERVICE.
	439	*
	440	* As soon as we fail the bind() with EACCES we know we will never
	441	* be able to bind to a system port so we terminate the loop.
	442	*
	443	* On linux, use
	444	* sudo setcap 'cap_net_bind_service=+ep' /path/executable
	445	* to make the executable able to bind to a system port.
ac559609 RI	446	*
	447	* On Windows, there is no concept of privileged ports. Thus
	448	* binding will usually succeed.
07fd0cbc	449	*/
ac559609 RI	450	{
	451	struct sockaddr_in sin;
	452	static int portOfs = 0;
	453	const int firstPort = 512; /* >= 512 according to Sun docs */
	454	const int portCount = IPPORT_RESERVED - firstPort;
cb5b8be2	455	int startOfs, port, rc;
ac559609	456
cb5b8be2 RS	457	if (portOfs == 0) {
	458	portOfs = time(NULL) % 400;
	459	}
	460	startOfs = portOfs;
ac559609 RI	461	do {
	462	rc = -1;
	463	port = htons(firstPort + portOfs);
	464	portOfs = (portOfs + 1) % portCount;
	465
	466	/* skip well-known ports */
	467	if (!getservbyport(port, "tcp")) {
	468	memset(&sin, 0, sizeof(sin));
	469	sin.sin_port = port;
	470	sin.sin_family = AF_INET;
	471	sin.sin_addr.s_addr = 0;
	472
	473	rc = bind(rpc->fd, (struct sockaddr *)&sin, sizeof(struct sockaddr_in));
	474	#if !defined(WIN32)
	475	/* we got EACCES, so don't try again */
	476	if (rc != 0 && errno == EACCES)
	477	break;
	478	#endif
07fd0cbc	479	}
ac559609	480	} while (rc != 0 && portOfs != startOfs);
07fd0cbc	481	}
07fd0cbc	482
84004dbf	483	set_nonblocking(rpc->fd);
07fd0cbc	484
f96b24fa RI	485	if (connect(rpc->fd, (struct sockaddr *)s, socksize) != 0 && errno != EINPROGRESS) {
f96b24fa RI	486	rpc_set_error(rpc, "connect() to server failed. %s(%d)", strerror(errno), errno);
1896d37b	487	return -1;
84004dbf RS	488	}
	489
	490	return 0;
	491	}
	492
1744ef90 RS	493	int rpc_connect_async(struct rpc_context rpc, const char server, int port, rpc_cb cb, void *private_data)
	494	{
	495	struct sockaddr_in sin = (struct sockaddr_in )&rpc->s;
	496
4a2b0876 RS	497	assert(rpc->magic == RPC_CONTEXT_MAGIC);
4a2b0876 RS	498
1744ef90 RS	499	if (rpc->fd != -1) {
	500	rpc_set_error(rpc, "Trying to connect while already connected");
	501	return -1;
	502	}
	503
	504	if (rpc->is_udp != 0) {
	505	rpc_set_error(rpc, "Trying to connect on UDP socket");
	506	return -1;
	507	}
	508
	509	rpc->auto_reconnect = 0;
	510
	511	sin->sin_family = AF_INET;
	512	sin->sin_port = htons(port);
	513	if (inet_pton(AF_INET, server, &sin->sin_addr) != 1) {
	514	rpc_set_error(rpc, "Not a valid server ip address");
	515	return -1;
	516	}
	517
	518
	519	switch (rpc->s.ss_family) {
	520	case AF_INET:
	521	#ifdef HAVE_SOCKADDR_LEN
	522	sin->sin_len = sizeof(struct sockaddr_in);
	523	#endif
	524	break;
	525	}
	526
	527	rpc->connect_cb = cb;
	528	rpc->connect_data = private_data;
	529
	530	if (rpc_connect_sockaddr_async(rpc, &rpc->s) != 0) {
	531	return -1;
	532	}
	533
	534	return 0;
	535	}
	536
84004dbf RS	537	int rpc_disconnect(struct rpc_context rpc, char error)
84004dbf RS	538	{
4a2b0876 RS	539	assert(rpc->magic == RPC_CONTEXT_MAGIC);
4a2b0876 RS	540
1744ef90 RS	541	rpc_unset_autoreconnect(rpc);
1744ef90 RS	542
84004dbf RS	543	if (rpc->fd != -1) {
	544	close(rpc->fd);
	545	}
	546	rpc->fd = -1;
	547
	548	rpc->is_connected = 0;
	549
	550	rpc_error_all_pdus(rpc, error);
	551
	552	return 0;
	553	}
485bc9b9	554
1744ef90 RS	555	static void reconnect_cb(struct rpc_context rpc, int status, void data _U_, void *private_data)
1744ef90 RS	556	{
4a2b0876 RS	557	assert(rpc->magic == RPC_CONTEXT_MAGIC);
4a2b0876 RS	558
1744ef90 RS	559	if (status != RPC_STATUS_SUCCESS) {
	560	rpc_error_all_pdus(rpc, "RPC ERROR: Failed to reconnect async");
	561	return;
	562	}
	563
	564	rpc->is_connected = 1;
b990de23	565	rpc->connect_cb = NULL;
1744ef90 RS	566	}
	567
	568	/* disconnect but do not error all PDUs, just move pdus in-flight back to the outqueue and reconnect */
	569	static int rpc_reconnect_requeue(struct rpc_context *rpc)
b077fdeb RS	570	{
	571	struct rpc_pdu *pdu;
	572
4a2b0876 RS	573	assert(rpc->magic == RPC_CONTEXT_MAGIC);
4a2b0876 RS	574
b077fdeb RS	575	if (rpc->fd != -1) {
	576	close(rpc->fd);
	577	}
	578	rpc->fd = -1;
	579
	580	rpc->is_connected = 0;
	581
	582	/* socket is closed so we will not get any replies to any commands
	583	* in flight. Move them all over from the waitpdu queue back to the out queue
	584	*/
	585	for (pdu=rpc->waitpdu; pdu; pdu=pdu->next) {
	586	SLIST_REMOVE(&rpc->waitpdu, pdu);
	587	SLIST_ADD(&rpc->outqueue, pdu);
1744ef90 RS	588	/* we have to re-send the whole pdu again */
	589	pdu->written = 0;
	590	}
	591
	592	if (rpc->auto_reconnect != 0) {
	593	rpc->connect_cb = reconnect_cb;
	594
	595	if (rpc_connect_sockaddr_async(rpc, &rpc->s) != 0) {
	596	rpc_error_all_pdus(rpc, "RPC ERROR: Failed to reconnect async");
	597	return -1;
	598	}
b077fdeb RS	599	}
	600
	601	return 0;
	602	}
	603
485bc9b9 RS	604
	605	int rpc_bind_udp(struct rpc_context rpc, char addr, int port)
	606	{
	607	struct addrinfo *ai = NULL;
	608	char service[6];
	609
4a2b0876 RS	610	assert(rpc->magic == RPC_CONTEXT_MAGIC);
4a2b0876 RS	611
485bc9b9 RS	612	if (rpc->is_udp == 0) {
	613	rpc_set_error(rpc, "Cant not bind UDP. Not UDP context");
	614	return -1;
	615	}
	616
6874f61e	617	sprintf(service, "%d", port);
485bc9b9 RS	618	if (getaddrinfo(addr, service, NULL, &ai) != 0) {
	619	rpc_set_error(rpc, "Invalid address:%s. "
	620	"Can not resolv into IPv4/v6 structure.");
	621	return -1;
	622	}
	623
	624	switch(ai->ai_family) {
	625	case AF_INET:
	626	rpc->fd = socket(ai->ai_family, SOCK_DGRAM, 0);
	627	if (rpc->fd == -1) {
	628	rpc_set_error(rpc, "Failed to create UDP socket: %s", strerror(errno));
	629	freeaddrinfo(ai);
	630	return -1;
	631	}
	632
	633	if (bind(rpc->fd, (struct sockaddr *)ai->ai_addr, sizeof(struct sockaddr_in)) != 0) {
	634	rpc_set_error(rpc, "Failed to bind to UDP socket: %s",strerror(errno));
	635	freeaddrinfo(ai);
	636	return -1;
	637	}
	638	break;
	639	default:
	640	rpc_set_error(rpc, "Can not handle UPD sockets of family %d yet", ai->ai_family);
	641	freeaddrinfo(ai);
	642	return -1;
	643	}
	644
	645	freeaddrinfo(ai);
	646
	647	return 0;
	648	}
	649
5bf60dc6 RS	650	int rpc_set_udp_destination(struct rpc_context rpc, char addr, int port, int is_broadcast)
	651	{
	652	struct addrinfo *ai = NULL;
	653	char service[6];
	654
4a2b0876 RS	655	assert(rpc->magic == RPC_CONTEXT_MAGIC);
4a2b0876 RS	656
5bf60dc6 RS	657	if (rpc->is_udp == 0) {
	658	rpc_set_error(rpc, "Can not set destination sockaddr. Not UDP context");
	659	return -1;
	660	}
	661
6874f61e	662	sprintf(service, "%d", port);
5bf60dc6 RS	663	if (getaddrinfo(addr, service, NULL, &ai) != 0) {
	664	rpc_set_error(rpc, "Invalid address:%s. "
	665	"Can not resolv into IPv4/v6 structure.");
	666	return -1;
	667	}
	668
	669	if (rpc->udp_dest) {
	670	free(rpc->udp_dest);
	671	rpc->udp_dest = NULL;
	672	}
	673	rpc->udp_dest = malloc(ai->ai_addrlen);
	674	if (rpc->udp_dest == NULL) {
	675	rpc_set_error(rpc, "Out of memory. Failed to allocate sockaddr structure");
be7f5933	676	freeaddrinfo(ai);
5bf60dc6 RS	677	return -1;
	678	}
	679	memcpy(rpc->udp_dest, ai->ai_addr, ai->ai_addrlen);
	680	freeaddrinfo(ai);
	681
	682	rpc->is_broadcast = is_broadcast;
bb4e9ed6	683	setsockopt(rpc->fd, SOL_SOCKET, SO_BROADCAST, (char *)&is_broadcast, sizeof(is_broadcast));
5bf60dc6 RS	684
	685	return 0;
	686	}
c481da67 RS	687
	688	struct sockaddr rpc_get_recv_sockaddr(struct rpc_context rpc)
	689	{
4a2b0876 RS	690	assert(rpc->magic == RPC_CONTEXT_MAGIC);
4a2b0876 RS	691
c481da67 RS	692	return (struct sockaddr *)&rpc->udp_src;
c481da67 RS	693	}
83aa785d RS	694
	695	int rpc_queue_length(struct rpc_context *rpc)
	696	{
	697	int i=0;
	698	struct rpc_pdu *pdu;
	699
4a2b0876 RS	700	assert(rpc->magic == RPC_CONTEXT_MAGIC);
4a2b0876 RS	701
83aa785d RS	702	for(pdu = rpc->outqueue; pdu; pdu = pdu->next) {
	703	i++;
	704	}
	705	for(pdu = rpc->waitpdu; pdu; pdu = pdu->next) {
	706	i++;
	707	}
	708	return i;
	709	}