[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/>.
*/

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h>
#include <poll.h>
#include <string.h>
#include <errno.h>
#include <rpc/rpc.h>
#include <rpc/xdr.h>
#include <arpa/inet.h>
#include <sys/ioctl.h>
#include "libnfs.h"
#include "libnfs-raw.h"
#include "libnfs-private.h"
#include "slist.h"

static void set_nonblocking(int fd)
{
	unsigned v;
	v = fcntl(fd, F_GETFL, 0);
        fcntl(fd, F_SETFL, v | O_NONBLOCK);
}

int rpc_get_fd(struct rpc_context *rpc)
{
	return rpc->fd;
}

int rpc_which_events(struct rpc_context *rpc)
{
	int events = rpc->is_connected ? POLLIN : POLLOUT;

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

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

	if (rpc == NULL) {
		printf("trying to write to socket for NULL context\n");
		return -1;
	}
	if (rpc->fd == -1) {
		printf("trying to write but not connected\n");
		return -2;
	}

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

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

		count = write(rpc->fd, rpc->outqueue->outdata.data + rpc->outqueue->written, total - rpc->outqueue->written);
		if (count == -1) {
			if (errno == EAGAIN || errno == EWOULDBLOCK) {
				printf("socket would block, return from write to socket\n");
				return 0;
			}
			printf("Error when writing to socket :%s(%d)\n", strerror(errno), errno);
			return -3;
		}

		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;
	unsigned char *buf;
	ssize_t count;

	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 -2;
	}
	size = rpc->insize - rpc->inpos + available;
	buf = malloc(size);
	if (buf == NULL) {
		rpc_set_error(rpc, "Out of memory: failed to allocate %d bytes for input buffer. Closing socket.", size);
		return -3;
	}
	if (rpc->insize > rpc->inpos) {
		memcpy(buf, rpc->inbuf + rpc->inpos, rpc->insize - rpc->inpos);
		rpc->insize -= rpc->inpos;
		rpc->inpos   = 0;
	}

	count = read(rpc->fd, buf + rpc->insize, available);
	if (count == -1) {
		if (errno == EINTR) {
			free(buf);
			buf = NULL;
			return 0;
		}
		rpc_set_error(rpc, "Read from socket failed, errno:%d. Closing socket.", errno);
		free(buf);
		buf = NULL;
		return -4;
	}

	if (rpc->inbuf != NULL) {
		free(rpc->inbuf);
	}
	rpc->inbuf   = (char *)buf;
	rpc->insize += count;

	while (1) {
		if (rpc->insize - rpc->inpos < 4) {
			return 0;
		}
		count = rpc_get_pdu_size(rpc->inbuf + rpc->inpos);
		if (rpc->insize + rpc->inpos < count) {
			return 0;
		}
		if (rpc_process_pdu(rpc, rpc->inbuf + rpc->inpos, count) != 0) {
			rpc_set_error(rpc, "Invalid/garbage pdu received from server. Closing socket");
			return -5;
		}
		rpc->inpos += count;
		if (rpc->inpos == rpc->insize) {
			free(rpc->inbuf);
			rpc->inbuf = NULL;
			rpc->insize = 0;
			rpc->inpos = 0;
		}
	}
	return 0;
}


int rpc_service(struct rpc_context *rpc, int revents)
{
	if (revents & POLLERR) {
		int err = 0;
		socklen_t err_size = sizeof(err);

		if (getsockopt(rpc->fd, SOL_SOCKET, SO_ERROR,
				&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.");
		}
		rpc->connect_cb(rpc, RPC_STATUS_ERROR, rpc->error_string, rpc->connect_data);
		return -1;
	}
	if (revents & POLLHUP) {
		printf("rpc_service: POLLHUP, socket error\n");
		rpc_set_error(rpc, "Socket failed with POLLHUP");
		rpc->connect_cb(rpc, RPC_STATUS_ERROR, rpc->error_string, rpc->connect_data);
		return -2;
	}

	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,
				&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);
			rpc->connect_cb(rpc, RPC_STATUS_ERROR,
					NULL, rpc->connect_data);
			return -1;
		}

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

	if (revents & POLLOUT && rpc->outqueue != NULL) {
		if (rpc_write_to_socket(rpc) != 0) {
			printf("write to socket failed\n");
			return -3;
		}
	}

	if (revents & POLLIN) {
		if (rpc_read_from_socket(rpc) != 0) {
			rpc_disconnect(rpc, rpc_get_error(rpc));
			return -4;
		}
	}

	return 0;
}


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

	if (rpc->fd != -1) {
		rpc_set_error(rpc, "Trying to connect while already connected");
		printf("%s\n", rpc->error_string);
		return -1;
	}

	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");
		printf("%s\n", rpc->error_string);
		return -2;
	}

	switch (s.ss_family) {
	case AF_INET:
		socksize = sizeof(struct sockaddr_in);
#ifdef HAVE_SOCK_SIN_LEN
		sin->sin_len = socksize;
#endif
		rpc->fd = socket(AF_INET, SOCK_STREAM, 0);
		break;
	}

	if (rpc->fd == -1) {
		rpc_set_error(rpc, "Failed to open socket");
		printf("%s\n", rpc->error_string);
		return -3;
	}

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

	set_nonblocking(rpc->fd);

	if (connect(rpc->fd, (struct sockaddr *)&s, socksize) != 0 && errno != EINPROGRESS) {
		rpc_set_error(rpc, "connect() to server failed");
		printf("%s\n", rpc->error_string);
		return -4;
	}		

	return 0;
}	    

int rpc_disconnect(struct rpc_context *rpc, char *error)
{
	if (rpc->fd != -1) {
		close(rpc->fd);
	}
	rpc->fd  = -1;

	rpc->is_connected = 0;

	rpc_error_all_pdus(rpc, error);

	return 0;
}
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	*/
	17
	18	#include <stdio.h>
98f5fee8	19	#include <stdlib.h>
84004dbf RS	20	#include <unistd.h>
	21	#include <fcntl.h>
	22	#include <poll.h>
	23	#include <string.h>
	24	#include <errno.h>
98f5fee8	25	#include <rpc/rpc.h>
84004dbf RS	26	#include <rpc/xdr.h>
	27	#include <arpa/inet.h>
	28	#include <sys/ioctl.h>
	29	#include "libnfs.h"
	30	#include "libnfs-raw.h"
	31	#include "libnfs-private.h"
	32	#include "slist.h"
	33
	34	static void set_nonblocking(int fd)
	35	{
	36	unsigned v;
	37	v = fcntl(fd, F_GETFL, 0);
	38	fcntl(fd, F_SETFL, v \| O_NONBLOCK);
	39	}
	40
	41	int rpc_get_fd(struct rpc_context *rpc)
	42	{
	43	return rpc->fd;
	44	}
	45
	46	int rpc_which_events(struct rpc_context *rpc)
	47	{
912f7ad5	48	int events = rpc->is_connected ? POLLIN : POLLOUT;
84004dbf RS	49
	50	if (rpc->outqueue) {
	51	events \|= POLLOUT;
	52	}
	53	return events;
	54	}
	55
	56	static int rpc_write_to_socket(struct rpc_context *rpc)
	57	{
	58	ssize_t count;
	59
	60	if (rpc == NULL) {
	61	printf("trying to write to socket for NULL context\n");
	62	return -1;
	63	}
	64	if (rpc->fd == -1) {
	65	printf("trying to write but not connected\n");
	66	return -2;
	67	}
	68
	69	while (rpc->outqueue != NULL) {
	70	ssize_t total;
	71
	72	total = rpc->outqueue->outdata.size;
	73
	74	count = write(rpc->fd, rpc->outqueue->outdata.data + rpc->outqueue->written, total - rpc->outqueue->written);
	75	if (count == -1) {
	76	if (errno == EAGAIN \|\| errno == EWOULDBLOCK) {
	77	printf("socket would block, return from write to socket\n");
	78	return 0;
	79	}
	80	printf("Error when writing to socket :%s(%d)\n", strerror(errno), errno);
	81	return -3;
	82	}
	83
	84	rpc->outqueue->written += count;
	85	if (rpc->outqueue->written == total) {
	86	struct rpc_pdu *pdu = rpc->outqueue;
	87
	88	SLIST_REMOVE(&rpc->outqueue, pdu);
	89	SLIST_ADD_END(&rpc->waitpdu, pdu);
	90	}
	91	}
	92	return 0;
	93	}
	94
	95	static int rpc_read_from_socket(struct rpc_context *rpc)
	96	{
	97	int available;
	98	int size;
	99	unsigned char *buf;
	100	ssize_t count;
	101
	102	if (ioctl(rpc->fd, FIONREAD, &available) != 0) {
	103	rpc_set_error(rpc, "Ioctl FIONREAD returned error : %d. Closing socket.", errno);
	104	return -1;
	105	}
	106	if (available == 0) {
	107	rpc_set_error(rpc, "Socket has been closed");
	108	return -2;
	109	}
	110	size = rpc->insize - rpc->inpos + available;
	111	buf = malloc(size);
	112	if (buf == NULL) {
113	rpc_set_error(rpc, "Out of memory: failed to allocate %d bytes for input buffer. Closing socket.", size);
114	return -3;
115	}
116	if (rpc->insize > rpc->inpos) {
117	memcpy(buf, rpc->inbuf + rpc->inpos, rpc->insize - rpc->inpos);
118	rpc->insize -= rpc->inpos;
119	rpc->inpos = 0;
120	}
121
122	count = read(rpc->fd, buf + rpc->insize, available);
123	if (count == -1) {
124	if (errno == EINTR) {
125	free(buf);
126	buf = NULL;
127	return 0;
128	}
129	rpc_set_error(rpc, "Read from socket failed, errno:%d. Closing socket.", errno);
130	free(buf);
131	buf = NULL;
132	return -4;
133	}
134
135	if (rpc->inbuf != NULL) {
136	free(rpc->inbuf);
137	}
138	rpc->inbuf = (char *)buf;
139	rpc->insize += count;
140
141	while (1) {
142	if (rpc->insize - rpc->inpos < 4) {
143	return 0;
144	}
145	count = rpc_get_pdu_size(rpc->inbuf + rpc->inpos);
146	if (rpc->insize + rpc->inpos < count) {
147	return 0;
148	}
149	if (rpc_process_pdu(rpc, rpc->inbuf + rpc->inpos, count) != 0) {
150	rpc_set_error(rpc, "Invalid/garbage pdu received from server. Closing socket");
151	return -5;
152	}
153	rpc->inpos += count;
154	if (rpc->inpos == rpc->insize) {
155	free(rpc->inbuf);
156	rpc->inbuf = NULL;
157	rpc->insize = 0;
158	rpc->inpos = 0;
159	}
160	}
161	return 0;
162	}
163
164
165
166	int rpc_service(struct rpc_context *rpc, int revents)
167	{
168	if (revents & POLLERR) {
912f7ad5 RS	169	int err = 0;
	170	socklen_t err_size = sizeof(err);
	171
	172	if (getsockopt(rpc->fd, SOL_SOCKET, SO_ERROR,
	173	&err, &err_size) != 0 \|\| err != 0) {
	174	if (err == 0) {
	175	err = errno;
	176	}
	177	rpc_set_error(rpc, "rpc_service: socket error "
	178	"%s(%d).",
	179	strerror(err), err);
84004dbf	180	} else {
912f7ad5 RS	181	rpc_set_error(rpc, "rpc_service: POLLERR, "
912f7ad5 RS	182	"Unknown socket error.");
84004dbf RS	183	}
	184	rpc->connect_cb(rpc, RPC_STATUS_ERROR, rpc->error_string, rpc->connect_data);
	185	return -1;
	186	}
	187	if (revents & POLLHUP) {
	188	printf("rpc_service: POLLHUP, socket error\n");
	189	rpc_set_error(rpc, "Socket failed with POLLHUP");
	190	rpc->connect_cb(rpc, RPC_STATUS_ERROR, rpc->error_string, rpc->connect_data);
	191	return -2;
	192	}
	193
	194	if (rpc->is_connected == 0 && rpc->fd != -1 && revents&POLLOUT) {
912f7ad5 RS	195	int err = 0;
	196	socklen_t err_size = sizeof(err);
	197
	198	if (getsockopt(rpc->fd, SOL_SOCKET, SO_ERROR,
	199	&err, &err_size) != 0 \|\| err != 0) {
	200	if (err == 0) {
	201	err = errno;
	202	}
	203	rpc_set_error(rpc, "rpc_service: socket error "
	204	"%s(%d) while connecting.",
	205	strerror(err), err);
	206	rpc->connect_cb(rpc, RPC_STATUS_ERROR,
	207	NULL, rpc->connect_data);
	208	return -1;
	209	}
	210
84004dbf RS	211	rpc->is_connected = 1;
	212	rpc->connect_cb(rpc, RPC_STATUS_SUCCESS, NULL, rpc->connect_data);
	213	return 0;
	214	}
	215
	216	if (revents & POLLOUT && rpc->outqueue != NULL) {
	217	if (rpc_write_to_socket(rpc) != 0) {
	218	printf("write to socket failed\n");
	219	return -3;
	220	}
	221	}
	222
	223	if (revents & POLLIN) {
	224	if (rpc_read_from_socket(rpc) != 0) {
	225	rpc_disconnect(rpc, rpc_get_error(rpc));
	226	return -4;
	227	}
	228	}
	229
	230	return 0;
	231	}
	232
	233
	234	int rpc_connect_async(struct rpc_context rpc, const char server, int port, rpc_cb cb, void *private_data)
	235	{
	236	struct sockaddr_storage s;
	237	struct sockaddr_in sin = (struct sockaddr_in )&s;
	238	int socksize;
	239
	240	if (rpc->fd != -1) {
	241	rpc_set_error(rpc, "Trying to connect while already connected");
	242	printf("%s\n", rpc->error_string);
	243	return -1;
	244	}
	245
	246	sin->sin_family = AF_INET;
	247	sin->sin_port = htons(port);
	248	if (inet_pton(AF_INET, server, &sin->sin_addr) != 1) {
	249	rpc_set_error(rpc, "Not a valid server ip address");
	250	printf("%s\n", rpc->error_string);
	251	return -2;
	252	}
	253
	254	switch (s.ss_family) {
	255	case AF_INET:
84004dbf	256	socksize = sizeof(struct sockaddr_in);
ea214e45 RS	257	#ifdef HAVE_SOCK_SIN_LEN
	258	sin->sin_len = socksize;
	259	#endif
	260	rpc->fd = socket(AF_INET, SOCK_STREAM, 0);
84004dbf RS	261	break;
	262	}
	263
	264	if (rpc->fd == -1) {
	265	rpc_set_error(rpc, "Failed to open socket");
	266	printf("%s\n", rpc->error_string);
	267	return -3;
	268	}
	269
	270	rpc->connect_cb = cb;
	271	rpc->connect_data = private_data;
	272
	273	set_nonblocking(rpc->fd);
	274
	275	if (connect(rpc->fd, (struct sockaddr *)&s, socksize) != 0 && errno != EINPROGRESS) {
	276	rpc_set_error(rpc, "connect() to server failed");
	277	printf("%s\n", rpc->error_string);
	278	return -4;
	279	}
	280
	281	return 0;
	282	}
	283
	284	int rpc_disconnect(struct rpc_context rpc, char error)
	285	{
	286	if (rpc->fd != -1) {
	287	close(rpc->fd);
	288	}
	289	rpc->fd = -1;
	290
	291	rpc->is_connected = 0;
	292
	293	rpc_error_all_pdus(rpc, error);
	294
	295	return 0;
	296	}