[FUSE] Start refactoring so we can handle multithread fuse

3 files changed, 239 insertions, 0 deletions

diff --git a/src/shared/protocol.h b/src/shared/protocol.h
new file mode 100644
index 0000000..41e9af2
--- /dev/null
+++ b/src/shared/protocol.h
@@ -0,0 +1,138 @@
+#ifndef _PROTOCOL_H_
+#define _PROTOCOL_H_
+
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <errno.h>
+#include "../types.h"
+#include "../serialize.h"
+
+#define FLAGS8_SERVER (1)
+
+#define REPLY_OK (0)
+#define REPLY_ERRNO (-1)
+#define REPLY_AGAIN (-2)
+#define REPLY_INTR (-3)
+#define REPLY_CLOSED (-4)
+#define REPLY_INCOMPLETE (-5)
+#define REPLY_WRONGMAGIC (-6)
+
+static inline int dnbd3_read_reply(int sock, dnbd3_reply_t *reply, bool wait)
+{
+	int ret = recv( sock, reply, sizeof(*reply), (wait ? MSG_WAITALL : MSG_DONTWAIT) | MSG_NOSIGNAL );
+	if ( ret == 0 ) return REPLY_CLOSED;
+	if ( ret < 0 ) {
+		if ( errno == EAGAIN || errno == EWOULDBLOCK ) return REPLY_AGAIN;
+		if ( errno == EINTR ) return REPLY_INTR;
+		return REPLY_ERRNO;
+	}
+	if ( !wait && ret != sizeof(*reply) ) ret += recv( sock, ((char*)reply) + ret, sizeof(*reply) - ret, MSG_WAITALL | MSG_NOSIGNAL );
+	if ( ret != sizeof(*reply) ) return REPLY_INCOMPLETE;
+	fixup_reply( *reply );
+	if ( reply->magic != dnbd3_packet_magic ) return REPLY_WRONGMAGIC;
+	return REPLY_OK;
+}
+
+static inline bool dnbd3_get_reply(int sock, dnbd3_reply_t *reply)
+{
+	return dnbd3_read_reply( sock, reply, true ) == REPLY_OK;
+}
+
+static inline bool dnbd3_select_image(int sock, const char *lower_name, uint16_t rid, uint8_t flags8)
+{
+	serialized_buffer_t serialized;
+	dnbd3_request_t request;
+	struct iovec iov[2];
+	serializer_reset_write( &serialized );
+	serializer_put_uint16( &serialized, PROTOCOL_VERSION );
+	serializer_put_string( &serialized, lower_name );
+	serializer_put_uint16( &serialized, rid );
+	serializer_put_uint8( &serialized, flags8 );
+	const ssize_t len = serializer_get_written_length( &serialized );
+	request.magic = dnbd3_packet_magic;
+	request.cmd = CMD_SELECT_IMAGE;
+	request.size = len;
+#ifdef _DEBUG
+	request.handle = 0;
+	request.offset = 0;
+#endif
+	fixup_request( request );
+	iov[0].iov_base = &request;
+	iov[0].iov_len = sizeof(request);
+	iov[1].iov_base = &serialized;
+	iov[1].iov_len = len;
+	return writev( sock, iov, 2 ) == len + (ssize_t)sizeof(request);
+}
+
+static inline bool dnbd3_get_block(int sock, uint64_t offset, uint32_t size, uint64_t handle)
+{
+	dnbd3_request_t request;
+	request.magic = dnbd3_packet_magic;
+	request.handle = handle;
+	request.cmd = CMD_GET_BLOCK;
+	request.offset = offset;
+	request.size = size;
+	fixup_request( request );
+	return send( sock, &request, sizeof(request), MSG_NOSIGNAL ) == sizeof(request);
+}
+
+static inline bool dnbd3_get_crc32(int sock, uint32_t *master, void *buffer, size_t *bufferLen)
+{
+	dnbd3_request_t request;
+	dnbd3_reply_t reply;
+	request.magic = dnbd3_packet_magic;
+	request.handle = 0;
+	request.cmd = CMD_GET_CRC32;
+	request.offset = 0;
+	request.size = 0;
+	fixup_request( request );
+	if ( send( sock, &request, sizeof(request), 0 ) != sizeof(request) ) return false;
+	if ( !dnbd3_get_reply( sock, &reply ) ) return false;
+	if ( reply.size == 0 ) {
+		*bufferLen = 0;
+		return true;
+	}
+	if ( reply.size < 4 ) return false;
+	reply.size -= 4;
+	if ( reply.cmd != CMD_GET_CRC32 || reply.size > *bufferLen ) return false;
+	*bufferLen = reply.size;
+	if ( recv( sock, master, sizeof(uint32_t), MSG_WAITALL | MSG_NOSIGNAL ) != sizeof(uint32_t) ) return false;
+	uint32_t done = 0;
+	while ( done < reply.size ) {
+		const ssize_t ret = recv( sock, (char*)buffer + done, reply.size - done, 0 );
+		if ( ret <= 0 ) return false;
+		done += ret;
+	}
+	return true;
+}
+
+/**
+ * Pass a full serialized_buffer_t and a socket fd. Parsed data will be returned in further arguments.
+ * Note that all strings will point into the passed buffer, so there's no need to free them.
+ * This function will also read the header for you, as this message can only occur during connection,
+ * where no unrequested messages could arrive inbetween.
+ */
+static inline bool dnbd3_select_image_reply(serialized_buffer_t *buffer, int sock, uint16_t *protocol_version, char **name, uint16_t *rid,
+        uint64_t *imageSize)
+{
+	dnbd3_reply_t reply;
+	if ( !dnbd3_get_reply( sock, &reply ) ) {
+		return false;
+	}
+	if ( reply.cmd != CMD_SELECT_IMAGE || reply.size < 3 || reply.size > MAX_PAYLOAD ) {
+		return false;
+	}
+// receive reply payload
+	if ( recv( sock, buffer, reply.size, MSG_WAITALL | MSG_NOSIGNAL ) != reply.size ) {
+		return false;
+	}
+// handle/check reply payload
+	serializer_reset_read( buffer, reply.size );
+	*protocol_version = serializer_get_uint16( buffer );
+	*name = serializer_get_string( buffer );
+	*rid = serializer_get_uint16( buffer );
+	*imageSize = serializer_get_uint64( buffer );
+	return true;
+}
+
+#endif
diff --git a/src/shared/signal.c b/src/shared/signal.c
new file mode 100644
index 0000000..a0697f8
--- /dev/null
+++ b/src/shared/signal.c
@@ -0,0 +1,52 @@
+#include "signal.h"
+#include <sys/eventfd.h>
+#include <poll.h>
+#include <inttypes.h>
+#include <errno.h>
+#include <unistd.h>
+
+int signal_new()
+{
+	return eventfd( 0, EFD_NONBLOCK );
+}
+
+int signal_newBlocking()
+{
+	return eventfd( 0, 0 );
+}
+
+int signal_call(int signalFd)
+{
+	if ( signalFd < 0 ) return 0;
+	static uint64_t one = 1;
+	return write( signalFd, &one, sizeof one ) == sizeof one;
+}
+
+int signal_wait(int signalFd, int timeoutMs)
+{
+	struct pollfd ps = {
+		.fd = signalFd,
+		.events = POLLIN
+	};
+	int ret = poll( &ps, 1, timeoutMs );
+	if ( ret == 0 ) return SIGNAL_TIMEOUT;
+	if ( ret == -1 ) return SIGNAL_ERROR;
+	if ( ps.revents & ( POLLERR | POLLNVAL ) ) return SIGNAL_ERROR;
+	return signal_clear( signalFd );
+}
+
+int signal_clear(int signalFd)
+{
+	uint64_t ret;
+	if ( read( signalFd, &ret, sizeof ret ) != sizeof ret ) {
+		if ( errno == EAGAIN ) return 0;
+		return SIGNAL_ERROR;
+	}
+	return (int)ret;
+}
+
+void signal_close(int signalFd)
+{
+	close( signalFd );
+}
+
diff --git a/src/shared/signal.h b/src/shared/signal.h
new file mode 100644
index 0000000..0e2f85f
--- /dev/null
+++ b/src/shared/signal.h
@@ -0,0 +1,49 @@
+#ifndef _SIGNAL_H_
+#define _SIGNAL_H_
+
+#define SIGNAL_OK (0)
+#define SIGNAL_TIMEOUT (-2)
+#define SIGNAL_ERROR (-1)
+
+/**
+ * Create a new signal fd (eventfd), nonblocking.
+ * @return >= 0 on success, which is the fd; < 0 on error
+ */
+int signal_new();
+
+/**
+ * Create a new signal fd (eventfd), blocking.
+ * @return >= 0 on success, which is the fd; < 0 on error
+ */
+int signal_newBlocking();
+
+/**
+ * Trigger the given signal, so a wait or clear call will succeed.
+ * @return SIGNAL_OK on success, SIGNAL_ERROR on error
+ */
+int signal_call(int signalFd);
+
+/**
+ * Wait for given signal, with an optional timeout.
+ * If timeout == 0, just poll once.
+ * If timeout < 0, wait forever.
+ * @return > 0 telling how many times the signal was called,
+ *    SIGNAL_TIMEOUT if the timeout was reached,
+ *    SIGNAL_ERROR if some error occured
+ */
+int signal_wait(int signalFd, int timeoutMs);
+
+/**
+ * Clears any pending signals on this signal fd.
+ * @return number of signals that were pending,
+ *    SIGNAL_ERROR if some error occured
+ */
+int signal_clear(int signalFd);
+
+/**
+ * Close the given signal.
+ */
+void signal_close(int signalFd);
+
+#endif
+