Redo the buffer handling for input buffers and make sure we only read one PDU at a time

examples/nfsclient-sync.c

@@ -46,6 +46,8 @@ struct client {
 };
 
 
+char buf[2*1024*1024];
+
 int main(int argc _U_, char *argv[] _U_)
 {
 	struct nfs_context *nfs;
@@ -58,7 +60,6 @@ int main(int argc _U_, char *argv[] _U_)
 	client.server = SERVER;
 	client.export = EXPORT;
 	client.is_finished = 0;
-	char buf[16];
 	off_t offset;
 	struct statvfs svfs;
 	exports export, tmp;
@@ -102,6 +103,7 @@ int main(int argc _U_, char *argv[] _U_)
 		exit(10);
 	}
 
+#if 0
 	ret = nfs_read(nfs, nfsfh, 16, buf);
 	if (ret < 0) {
 		printf("Failed to pread(%s) %s\n", NFSFILE, nfs_get_error(nfs));
@@ -112,7 +114,8 @@ int main(int argc _U_, char *argv[] _U_)
 		printf("%02x ", buf[i]&0xff);
 	}
 	printf("\n");
-	ret = nfs_read(nfs, nfsfh, 16, buf);
+#endif
+	ret = nfs_read(nfs, nfsfh, sizeof(buf), buf);
 	if (ret < 0) {
 		printf("Failed to pread(%s) %s\n", NFSFILE, nfs_get_error(nfs));
 		exit(10);

include/libnfs-private.h

@@ -35,8 +35,8 @@ struct rpc_context {
        struct rpc_pdu *outqueue;
        struct rpc_pdu *waitpdu;
 
-       int insize;
        int inpos;
+       int insize;
        char *inbuf;
 };
 

lib/libnfs-sync.c

@@ -1074,7 +1074,6 @@ void mount_getexports_cb(struct rpc_context *mount_context _U_, int status, void
 	struct sync_cb_data *cb_data = private_data;
 	exports export = *(exports *)data;
 
-	printf("got exports back\n");
 	cb_data->is_finished = 1;
 	cb_data->status = status;
 	cb_data->return_data = NULL;

lib/libnfs.c

@@ -857,7 +857,7 @@ static void nfs_pread_mcb(struct rpc_context *rpc _U_, int status, void *command
 			rpc_set_error(nfs->rpc, "NFS: Read failed with %s(%d)", nfsstat3_to_str(res->status), nfsstat3_to_errno(res->status));
 			data->error = 1;
 		} else {
-			memcpy(&data->buffer[mdata->offset], res->READ3res_u.resok.data.data_val, res->READ3res_u.resok.count);
+			memcpy(&data->buffer[mdata->offset - data->start_offset], res->READ3res_u.resok.data.data_val, res->READ3res_u.resok.count);
 			if ((unsigned)data->max_offset < mdata->offset + res->READ3res_u.resok.count) {
 				data->max_offset = mdata->offset + res->READ3res_u.resok.count;
 			}
@@ -870,7 +870,6 @@ static void nfs_pread_mcb(struct rpc_context *rpc _U_, int status, void *command
 		return;
 	}
 
-
 	if (data->error != 0) {
 		data->cb(-EFAULT, nfs, command_data, data->private_data);
 		free_nfs_cb_data(data);
@@ -948,7 +947,6 @@ int nfs_pread_async(struct nfs_context *nfs, struct nfsfh *nfsfh, off_t offset,
 		mdata->data   = data;
 		mdata->offset = offset;
 		mdata->count  = readcount;
-
 		if (rpc_nfs_read_async(nfs->rpc, nfs_pread_mcb, &nfsfh->fh, offset, readcount, mdata) != 0) {
 			rpc_set_error(nfs->rpc, "RPC error: Failed to send READ call for %s", data->path);
 			data->cb(-ENOMEM, nfs, rpc_get_error(nfs->rpc), data->private_data);

lib/socket.c

@@ -94,7 +94,7 @@ static int rpc_read_from_socket(struct rpc_context *rpc)
 {
 	int available;
 	int size;
-	unsigned char *buf;
+	int pdu_size;
 	ssize_t count;
 
 	if (ioctl(rpc->fd, FIONREAD, &available) != 0) {
@@ -105,57 +105,77 @@ static int rpc_read_from_socket(struct rpc_context *rpc)
 		rpc_set_error(rpc, "Socket has been closed");
 		return -1;
 	}
-	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 -1;
+
+	/* 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->insize > rpc->inpos) {
-		memcpy(buf, rpc->inbuf + rpc->inpos, rpc->insize - rpc->inpos);
-		rpc->insize -= rpc->inpos;
-		rpc->inpos   = 0;
+	if (rpc->inpos < 4) {
+		size = 4 - rpc->inpos;
+
+		count = read(rpc->fd, rpc->inbuf + rpc->inpos, size);
+		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;
 	}
 
-	count = read(rpc->fd, buf + rpc->insize, available);
+	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 = read(rpc->fd, rpc->inbuf + rpc->inpos, size);
 	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 -1;
 	}
+	available  -= count;
+	rpc->inpos += count;
 
-	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) {
+	if (rpc->inpos == rpc->insize) {
+		if (rpc_process_pdu(rpc, rpc->inbuf, pdu_size) != 0) {
 			rpc_set_error(rpc, "Invalid/garbage pdu received from server. Closing socket");
 			return -1;
 		}
-		rpc->inpos += count;
-		if (rpc->inpos == rpc->insize) {
-			free(rpc->inbuf);
-			rpc->inbuf = NULL;
-			rpc->insize = 0;
-			rpc->inpos = 0;
-		}
+		free(rpc->inbuf);
+		rpc->inbuf  = NULL;
+		rpc->insize = 0;
+		rpc->inpos  = 0;
 	}
+
 	return 0;
 }