Astral/kernel-src/io/net/localsocket.c

#include <kernel/sock.h>
#include <logging.h>
#include <kernel/alloc.h>
#include <kernel/interrupt.h>
#include <ringbuffer.h>
#include <kernel/vfs.h>
#include <kernel/file.h>
#include <time.h>

#define SOCKET_BUFFER (256 * 1024)
#define FILE_COUNT 256
#define BARRIER_SIZE 256

#define LOCALSOCK_CTRL_FILE 1

typedef struct {
	pid_t pid;
	uid_t uid;
	gid_t gid;
} localsock_cred_t;

struct localpair_t;
struct binding_t;
typedef struct {
	socket_t socket; // shutdown state is shared with pair
	ringbuffer_t ringbuffer; // shared with pair
	// client/connected server
	struct localpair_t *pair;
	// listening server
	struct binding_t *binding;
	ringbuffer_t backlog; // shared with binding
	bool listening; // shared with binding
	char *bindpath;
	ringbuffer_t fds; // shared with pair
	uintmax_t barriercurrent; // shared with pair
	uintmax_t barrierwrite; // shared with pair
	int filesremaining[BARRIER_SIZE]; // shared with pair
	int bytesremaining[BARRIER_SIZE]; // shared with pair
	time_t rcv_timeout_us;
	time_t snd_timeout_us;
	bool seqpacket;
	localsock_cred_t cred;
} localsocket_t;

// created by the client socket
// stays allocated until the last socket is closing
// in the case of a socket closing while on the backlog, it will remain allocated on the backlog until then
// connected whenever both arent NULL
// the mutex also protect shared data (such as the ringbuffer)
typedef struct localpair_t {
	localsocket_t *server; // from accept()
	localsocket_t *client; // from connect()
	mutex_t mutex;
} localpair_t;

// created on a bind
// binding stays valid as long as both the server socket and vnode are still alive
// keeps a reference to the vnode, but is only deleted from the vnode destructor
// so when the server socket destructor is called, it needs to unref the vnode
// to indicate the binding is no longer used. if the vnode refcount turns 0 then the binding
// will be destroyed.
// see localsock_destroy() (server side) and localsock_leavebinding() (vnode side)
typedef struct binding_t {
	localsocket_t *server;
	vnode_t *vnode;
	mutex_t mutex;
} binding_t;

static inline bool backlog_has_data(localsocket_t *server) {
	return RINGBUFFER_DATACOUNT(&server->backlog) >= sizeof(localpair_t *);
}

static inline bool backlog_has_space(localsocket_t *server) {
	return RINGBUFFER_FREESPACE(&server->backlog) >= sizeof(localpair_t *);
}

static inline void pushbacklog(localsocket_t *server, localpair_t *pair) {
	__assert(backlog_has_space(server));
	__assert(ringbuffer_write(&server->backlog, &pair, sizeof(pair)) == sizeof(pair));
}

static inline localpair_t *popbacklog(localsocket_t *server) {
	__assert(backlog_has_data(server));
	localpair_t *pair = NULL;
	__assert(ringbuffer_read(&server->backlog, &pair, sizeof(pair)) == sizeof(pair));
	return pair;
}

static bool is_barrier_used(localsocket_t *localsocket, uintmax_t barrier) {
	size_t index = barrier % BARRIER_SIZE;
	return localsocket->bytesremaining[index] || localsocket->filesremaining[index];
}

static bool has_barrier_space(localsocket_t *peer) {
	return !(is_barrier_used(peer, peer->barrierwrite) && peer->barrierwrite + 1 == peer->barriercurrent + BARRIER_SIZE);
}

static size_t control_file_count(sockctrl_t *ctrl, size_t len) {
	if (ctrl == NULL)
		return 0;

	size_t fdcount = 0;
	size_t ctrlcount = sock_countctrl(ctrl, len);
	for (size_t i = 0; i < ctrlcount; ++i) {
		if (ctrl->level == SOL_SOCKET && ctrl->type == LOCALSOCK_CTRL_FILE)
			fdcount += SOCK_CTRL_DATALEN(ctrl) / sizeof(int);

		ctrl = SOCK_CTRL_NEXT(ctrl);
	}

	return fdcount;
}

static bool control_space_available(localsocket_t *peer, size_t fdcount) {
	if (fdcount > RINGBUFFER_FREESPACE(&peer->fds) / sizeof(file_t *))
		return false;

	if ((fdcount || peer->seqpacket) && !has_barrier_space(peer))
		return false;

	return true;
}

static bool data_space_available(localsocket_t *localsocket, localsocket_t *peer, size_t count) {
	if (localsocket->seqpacket)
		return RINGBUFFER_FREESPACE(&peer->ringbuffer) >= count;

	return RINGBUFFER_DATACOUNT(&peer->ringbuffer) != SOCKET_BUFFER;
}

static void advance_barrier(localsocket_t *localsocket) {
	while (localsocket->barriercurrent != localsocket->barrierwrite && is_barrier_used(localsocket, localsocket->barriercurrent) == false)
		++localsocket->barriercurrent;
}

static void prepare_barrier(localsocket_t *localsocket) {
	if (is_barrier_used(localsocket, localsocket->barrierwrite)) {
		__assert(localsocket->barrierwrite + 1 != localsocket->barriercurrent + BARRIER_SIZE);
		++localsocket->barrierwrite;
	}
}

static void record_bytes(localsocket_t *localsocket, size_t datasent) {
	if (localsocket->seqpacket)
		prepare_barrier(localsocket);

	localsocket->bytesremaining[localsocket->barrierwrite % BARRIER_SIZE] += datasent;
}

static int datapoll(localsocket_t *local, localsocket_t *peer, int events) {
	int revents = 0;

	// data to read?
	if ((events & POLLIN) && (RINGBUFFER_DATACOUNT(&local->ringbuffer) || (local->socket.shutdown & SOCKET_SHUTDOWN_READ) || (peer->socket.shutdown & SOCKET_SHUTDOWN_WRITE)))
		revents |= POLLIN;

	// XXX is this the correct behaviour?
	if ((events & POLLOUT) && ((local->socket.shutdown & SOCKET_SHUTDOWN_WRITE) || (peer->socket.shutdown & SOCKET_SHUTDOWN_READ)))
		revents |= POLLHUP;

	// space to write?
	else if ((events & POLLOUT) && RINGBUFFER_DATACOUNT(&peer->ringbuffer) != SOCKET_BUFFER)
		revents |= POLLOUT;

	return revents;
}

static int listeningpoll(localsocket_t *localsocket, int events) {
	int revents = 0;

	if (localsocket->socket.shutdown & SOCKET_SHUTDOWN_READ)
		return POLLHUP | (events & POLLIN);

	// can insert into backlog?
	if (backlog_has_space(localsocket))
		revents |= (events & POLLOUT) ? POLLOUT : 0;

	// can accept()?
	if (backlog_has_data(localsocket))
		revents |= (events & POLLIN) ? POLLIN : 0;

	return revents;
}

static int clientpoll(localsocket_t *localsocket, int events) {
	localsocket_t *server = localsocket->pair->server;

	// server disconnected
	if (server == NULL)
		return POLLHUP | ((events & POLLIN) ? POLLIN : 0);

	// the client has one more possible poll check:
	// it can still be connecting with the other end
	// so check if server has a backlog. if so, nothing can be done yet
	if (server->listening)
		return 0;

	return datapoll(localsocket, server, events);
}

static int serverpoll(localsocket_t *localsocket, int events) {
	localsocket_t *client = localsocket->pair->client;

	// client disconnected
	if (client == NULL)
		return POLLHUP | ((events & POLLIN) ? POLLIN : 0);

	return datapoll(localsocket, client, events);
}

// assumes all the needed locks for each case are being held
static int internalpoll(socket_t *socket, polldata_t *data, int events) {
	localsocket_t *localsocket = (localsocket_t *)socket;
	int revents = 0;
	// 4 possible cases:
	// socket is listening
	if (localsocket->listening)
		revents = listeningpoll(localsocket, events);

	// socket is client
	else if (localsocket->pair && localsocket->pair->client == localsocket)
		revents = clientpoll(localsocket, events);

	// socket is server
	else if (localsocket->pair && localsocket->pair->server == localsocket)
		revents = serverpoll(localsocket, events);

	// socket is not doing anything (closed/only bound)
	else
		revents = POLLHUP;

	if (revents == 0 && data)
		poll_add(&socket->pollheader, data, events);

	return revents;
}

int localsock_poll(socket_t *socket, polldata_t *data, int events) {
	localsocket_t *localsocket = (localsocket_t *)socket;

	MUTEX_ACQUIRE(&socket->mutex);
	if (localsocket->binding)
		MUTEX_ACQUIRE(&localsocket->binding->mutex);

	if (localsocket->pair)
		MUTEX_ACQUIRE(&localsocket->pair->mutex);

	int revents = internalpoll(socket, data, events);

	if (localsocket->pair)
		MUTEX_RELEASE(&localsocket->pair->mutex);

	if (localsocket->binding)
		MUTEX_RELEASE(&localsocket->binding->mutex);

	MUTEX_RELEASE(&socket->mutex);
	return revents;
}

#define LOCALSOCK_CTRL_FILE 1

static int sendctrl(socket_t *socket, sockctrl_t *ctrl, size_t len, size_t datasent) {
	localsocket_t *localsocket = (localsocket_t *)socket;
	localsocket_t *peer = localsocket->pair->client == localsocket ? localsocket->pair->server : localsocket->pair->client;
	if (peer == NULL)
		return 0;

	if (ctrl == NULL) {
		// if there is no control data, we just need to do byte accounting.
		record_bytes(peer, datasent);
		return 0;
	}

	size_t ctrlcount = sock_countctrl(ctrl, len);
	if (ctrlcount == 0) {
		// if there is no control data, we just need to do byte accounting.
		record_bytes(peer, datasent);
		return 0;
	}

	int error = 0;
	bool files_sent = false;

	for (size_t i = 0; i < ctrlcount; ++i) {
		if (ctrl->level != SOL_SOCKET)
			goto nextctrl;

		if (ctrl->type == LOCALSOCK_CTRL_FILE) {
			int *fds = (int *)ctrl->data;
			size_t fdcount = SOCK_CTRL_DATALEN(ctrl) / sizeof(int);
			if (fdcount == 0)
				goto nextctrl;

			// if theres no space in the ringbuffer for all fds, return ENOMEM
			// NOTE: I wasn't able to find what happens in this case in other implementations,
			// so this behaviour is likely wrong but suffices for now
			if (fdcount > RINGBUFFER_FREESPACE(&peer->fds) / sizeof(file_t *)) {
				error = ENOMEM;
				break;
	}

			file_t *files[fdcount];
			memset(files, 0, sizeof(files));

			for (size_t j = 0; j < fdcount; ++j) {
				files[j] = fd_get(fds[j]);
				if (files[j] == NULL) {
					error = EBADF;
					break;
				}
			}

			if (error) {
				// release all fds we acquired in case of error
				for (size_t j = 0; j < fdcount; ++j) {
					if (files[j] == NULL)
						break;

					fd_release(files[j]);
				}
				break;
			}

			// write it to the fd ring buffer
			__assert(ringbuffer_write(&peer->fds, files, fdcount * sizeof(file_t *)) == fdcount * sizeof(file_t *));

			// create a new barrier secion
			prepare_barrier(peer);

			peer->filesremaining[peer->barrierwrite % BARRIER_SIZE] = fdcount;
			peer->bytesremaining[peer->barrierwrite % BARRIER_SIZE] = datasent;

			files_sent = true;
		}

		nextctrl:
		ctrl = SOCK_CTRL_NEXT(ctrl);
	}

	// in case we have not sent any files, we still need to record the bytes
	if (error == 0 && !files_sent)
		record_bytes(peer, datasent);

	return error;
}

static int recvctrl(socket_t *socket, sockctrl_t *ctrl, size_t len, bool *truncated, size_t *receivedlen, uintmax_t flags) {
	localsocket_t *localsocket = (localsocket_t *)socket;
	int error = 0;
	*truncated = false;
	*receivedlen = 0;
	size_t current = localsocket->barriercurrent % BARRIER_SIZE;

	// ctlr can be null to discard any control data
	if (ctrl != NULL && len >= SOCK_CTRL_LEN(0)) {
		// files in the buffer
		size_t filecount = RINGBUFFER_DATACOUNT(&localsocket->fds) / sizeof(file_t *);
		size_t capacity = (len - SOCK_CTRL_ALIGN(sizeof(sockctrl_t))) / sizeof(int);

		// actual number that actually will be read will be the smallest of one of the following:
		// - number fds that fit in the data of the control
		// - files in the current section
		// - files actually in the ringbuffer
		size_t actualcount = min(min(capacity, filecount), localsocket->filesremaining[current]);
		if (actualcount == 0)
			goto discard;

		size_t donecount;
		size_t readcount = 0;
		int fds[actualcount];

		for (donecount = 0; donecount < actualcount; ++donecount) {
			// read the file pointers and insert them into the process fd table
			file_t *file;
			__assert(ringbuffer_read(&localsocket->fds, &file, sizeof(file_t *)) == sizeof(file_t *));
			++readcount;
			if (fd_insert(file, &fds[donecount], (flags & SOCKET_RECV_FLAGS_CLOEXEC_CTRL) ? O_CLOEXEC : 0)) {
				fd_release(file);
				break;
			}

			// fd_insert adds a reference, release ours here
			fd_release(file);
		}

		if (readcount != donecount)
			*truncated = true;

		if (donecount) {
			// prepare the ctrl that will be returned
			ctrl->length = donecount * sizeof(int) + sizeof(sockctrl_t);
			ctrl->level = SOL_SOCKET;
			ctrl->type = LOCALSOCK_CTRL_FILE;
			memcpy(ctrl->data, fds, donecount * sizeof(int));

			*receivedlen += ctrl->length;
		}

		localsocket->filesremaining[current] -= readcount;
	}

	discard:
	// discard the rest
	file_t *file = NULL;
	for (int i = 0; i < localsocket->filesremaining[current]; ++i) {
		ringbuffer_read(&localsocket->fds, &file, sizeof(file_t *));
		fd_release(file);
	}

	*truncated = *truncated || file != NULL;

	localsocket->filesremaining[current] = 0;
	advance_barrier(localsocket);

	return error;
}

static int localsock_send(socket_t *socket, sockdesc_t *sockdesc) {
	localsocket_t *localsocket = (localsocket_t *)socket;

	size_t fdcount = control_file_count(sockdesc->ctrl, sockdesc->ctrllen);
	if (fdcount > FILE_COUNT)
		return EINVAL;

	if (localsocket->seqpacket && sockdesc->count > SOCKET_BUFFER)
		return EMSGSIZE;

	MUTEX_ACQUIRE(&socket->mutex);

	int error;
	localpair_t *pair = localsocket->pair;
	// socket not connected
	if (pair == NULL) {
		error = ENOTCONN;
		goto leave;
	}

	MUTEX_ACQUIRE(&pair->mutex);
	localsocket_t *peer;

	// check/wait for space
	for (;;) {
		peer = pair->client == localsocket ? pair->server : pair->client;
		polldesc_t desc = {0};
		error = poll_initdesc(&desc, 1);
		if (error)
			goto leave;

		// if the client has closed this will return POLLHUP
		int revents = internalpoll(socket, &desc.data[0], POLLOUT);
		bool dataready = peer == NULL || data_space_available(localsocket, peer, sockdesc->count);
		bool ctrlready = peer == NULL || control_space_available(peer, fdcount);

		if ((revents && dataready && ctrlready) || (revents & POLLHUP)) {
			poll_leave(&desc);
			poll_destroydesc(&desc);
			break;
		}

		// add ourselves back to the loop to wait for more control space
		if (revents)
			poll_add(&socket->pollheader, &desc.data[0], POLLOUT);

		if (socket_nonblocking(socket, sockdesc->flags)) {
			error = EAGAIN;
			poll_leave(&desc);
			poll_destroydesc(&desc);
			goto leave;
		}

		MUTEX_RELEASE(&pair->mutex);
		MUTEX_RELEASE(&socket->mutex);

		error = poll_dowait(&desc, localsocket->snd_timeout_us);
		if (error == ETIMEDOUT)
			error = EAGAIN;

		poll_leave(&desc);
		poll_destroydesc(&desc);

		MUTEX_ACQUIRE(&socket->mutex);
		MUTEX_ACQUIRE(&pair->mutex);

		if (error)
			goto leave;
	}

	if (peer == NULL || (peer->socket.shutdown & SOCKET_SHUTDOWN_READ)) {
		if (current_thread()->proc && (sockdesc->flags & SOCKET_SEND_FLAGS_NOSIGNAL) == 0)
			signal_signalproc(current_thread()->proc, SIGPIPE);

		error = EPIPE;
		goto leave;
	}

	sockdesc->donecount = iovec_iterator_write_to_ringbuffer(sockdesc->iovec_iterator, &peer->ringbuffer, sockdesc->count);
	if (sockdesc->donecount == RINGBUFFER_USER_COPY_FAILED) {
		error = EFAULT;
		goto leave;
	}

	if (localsocket->seqpacket)
		__assert(sockdesc->donecount == sockdesc->count);

	__assert(sockdesc->donecount > 0);

	error = sendctrl(socket, sockdesc->ctrl, sockdesc->ctrllen, sockdesc->donecount);
	if (error) {
		ringbuffer_remove(&peer->ringbuffer, sockdesc->donecount);
		sockdesc->donecount = 0;
		goto leave;
	}

	// signal that there is data to read for any threads blocked on the peer socket
	poll_event(&peer->socket.pollheader, POLLIN);

	leave:
	if (pair)
		MUTEX_RELEASE(&pair->mutex);

	MUTEX_RELEASE(&socket->mutex);
	return error;
}


static int localsock_recv(socket_t *socket, sockdesc_t *sockdesc) {
	localsocket_t *localsocket = (localsocket_t *)socket;
	uintmax_t flags = sockdesc->flags;
	sockdesc->flags = 0;
	MUTEX_ACQUIRE(&socket->mutex);
	int error;

	localpair_t *pair = localsocket->pair;
	// socket not connected
	if (pair == NULL) {
		error = ENOTCONN;
		goto leave;
	}

	MUTEX_ACQUIRE(&pair->mutex);

	// check/wait for data
	for (;;) {
		polldesc_t desc = {0};
		error = poll_initdesc(&desc, 1);
		if (error)
			goto leave;

		// if the client has closed this will return POLLHUP
		int revents = internalpoll(socket, &desc.data[0], POLLIN);

		if (revents) {
			if ((revents & POLLHUP) == 0 && (flags & SOCKET_RECV_FLAGS_WAITALL) &&
				(flags & SOCKET_RECV_FLAGS_PEEK) == 0 && RINGBUFFER_DATACOUNT(&localsocket->ringbuffer) < sockdesc->count) {
				// wait for more data (WAITALL is set)
				poll_add(&socket->pollheader, &desc.data[0], POLLIN);
			} else {
				poll_leave(&desc);
				poll_destroydesc(&desc);
				break;
			}
		}

		if (socket_nonblocking(socket, flags)) {
			error = EAGAIN;
			poll_leave(&desc);
			poll_destroydesc(&desc);
			goto leave;
		}

		MUTEX_RELEASE(&pair->mutex);
		MUTEX_RELEASE(&socket->mutex);

		error = poll_dowait(&desc, localsocket->rcv_timeout_us);
		if (error == ETIMEDOUT)
			error = EAGAIN;

		poll_leave(&desc);
		poll_destroydesc(&desc);

		MUTEX_ACQUIRE(&socket->mutex);
		MUTEX_ACQUIRE(&pair->mutex);
		if (error)
			goto leave;
	}

	if ((flags & SOCKET_RECV_FLAGS_PEEK) == 0) {
		// receive control data
		bool truncated;
		error = recvctrl(socket, sockdesc->ctrl, sockdesc->ctrllen, &truncated, &sockdesc->ctrldone, flags);
		if (error)
			goto leave;

		if (truncated)
			sockdesc->flags |= SOCKET_RECV_FLAGS_CTRLTRUNCATED;
	}

	size_t recvcount = min(sockdesc->count, RINGBUFFER_DATACOUNT(&localsocket->ringbuffer));

	// check for a barrier from control data
	if (localsocket->bytesremaining[localsocket->barriercurrent % BARRIER_SIZE]) {
		recvcount = min(recvcount, localsocket->bytesremaining[localsocket->barriercurrent % BARRIER_SIZE]);

		if ((flags & SOCKET_RECV_FLAGS_PEEK) == 0) {
			// check if we should step forward from that barrier
			localsocket->bytesremaining[localsocket->barriercurrent % BARRIER_SIZE] -= recvcount;
			if (localsocket->bytesremaining[localsocket->barriercurrent % BARRIER_SIZE] == 0 &&
			    localsocket->bytesremaining[(localsocket->barriercurrent + 1) % BARRIER_SIZE])
				++localsocket->barriercurrent;
		}
	}

	advance_barrier(localsocket);

	if (flags & SOCKET_RECV_FLAGS_PEEK) {
		sockdesc->donecount = iovec_iterator_peek_from_ringbuffer(sockdesc->iovec_iterator, &localsocket->ringbuffer, 0, recvcount);
		if (sockdesc->donecount == RINGBUFFER_USER_COPY_FAILED) {
			error = EFAULT;
			goto leave;
		}
	} else {
		sockdesc->donecount = iovec_iterator_read_from_ringbuffer(sockdesc->iovec_iterator, &localsocket->ringbuffer, recvcount);
		if (sockdesc->donecount == RINGBUFFER_USER_COPY_FAILED) {
			error = EFAULT;
			goto leave;
		}

		localsocket_t *peer = pair->client == localsocket ? pair->server : pair->client;
		// signal that there is space to write for any threads blocked on this socket
		if (peer && sockdesc->donecount)
			poll_event(&peer->socket.pollheader, POLLOUT);
	}

	__assert(sockdesc->donecount == recvcount);

	leave:
	if (pair)
		MUTEX_RELEASE(&pair->mutex);

	MUTEX_RELEASE(&socket->mutex);
	return error;
}

static int localsock_accept(socket_t *_server, socket_t *_clientconnection, sockaddr_t *addr, uintmax_t flags) {
	localsocket_t *server = (localsocket_t *)_server;
	localsocket_t *clientconnection = (localsocket_t *)_clientconnection;
	addr->path[0] = '\0';

	MUTEX_ACQUIRE(&server->socket.mutex);
	binding_t *binding = server->binding;
	int error;

	if (binding == NULL) {
		MUTEX_RELEASE(&server->socket.mutex);
		return EINVAL;
	}

	MUTEX_ACQUIRE(&binding->mutex);

	if (server->listening == false) {
		error = EINVAL;
		goto leave;
	}

	localpair_t *pair = NULL;

	// check if there is anything in the backlog
	for (;;) {
		polldesc_t desc = {0};
		error = poll_initdesc(&desc, 1);
		if (error)
			goto leave;

		int revents = internalpoll(&server->socket, &desc.data[0], POLLIN);

		if (revents) {
			// there is something but we need to see if it is valid

			poll_leave(&desc);
			poll_destroydesc(&desc);

			if (!backlog_has_data(server)) {
				error = EINVAL;
				goto leave;
			}

			pair = popbacklog(server);
			MUTEX_ACQUIRE(&pair->mutex);
			if (pair->client == NULL) {
				// client closed the connection, delete the pair and keep trying
				free(pair);
				continue;
			}
			break;
		}

		if (socket_nonblocking(&server->socket, flags)) {
			error = EAGAIN;
			poll_leave(&desc);
			poll_destroydesc(&desc);
			goto leave;
		}

		MUTEX_RELEASE(&binding->mutex);
		MUTEX_RELEASE(&server->socket.mutex);

		error = poll_dowait(&desc, server->rcv_timeout_us);
		if (error == ETIMEDOUT)
			error = EAGAIN;

		poll_leave(&desc);
		poll_destroydesc(&desc);

		if (error)
			return error;

		MUTEX_ACQUIRE(&server->socket.mutex);
		MUTEX_ACQUIRE(&binding->mutex);
	}

	if (server->bindpath) {
		size_t length = strlen(server->bindpath) + 1;
		clientconnection->bindpath = alloc(length);
		if (clientconnection->bindpath == NULL)
			goto leave;

		memcpy(clientconnection->bindpath, server->bindpath, length);
	}

	// we have a valid pair and we have acquired the mutex already
	// configure the client connection. we don't lock it because it should only be
	// curretly accessible here anyways
	clientconnection->pair = pair;
	pair->server = clientconnection;

	if (current_thread()->proc)
		clientconnection->cred = server->cred;

	// signal to the client that they can start sending data now
	poll_event(&pair->client->socket.pollheader, POLLOUT);

	MUTEX_RELEASE(&pair->mutex);

	leave:
	MUTEX_RELEASE(&binding->mutex);
	MUTEX_RELEASE(&server->socket.mutex);
	return error;
}

static int localsock_connect(socket_t *socket, sockaddr_t *addr, uintmax_t flags, cred_t *cred) {
	localsocket_t *localsocket = (localsocket_t *)socket;
	int error;
	vnode_t *refnode = NULL;
	vnode_t *result = NULL;
	binding_t *binding = NULL;

	MUTEX_ACQUIRE(&socket->mutex);
	// listening/bound socket
	if (localsocket->listening || localsocket->binding) {
		error = EOPNOTSUPP;
		goto leave;
	}

	// connected socket
	if (localsocket->pair) {
		error = EISCONN;
		goto leave;
	}

	if (current_thread()->proc) {
		localsocket->cred.pid = current_thread()->proc->pid;
		localsocket->cred.uid = current_thread()->proc->cred.euid;
		localsocket->cred.gid = current_thread()->proc->cred.egid;
	}

	// get the vnode specified in addr
	refnode = addr->path[0] == '/' ? proc_get_root() : proc_get_cwd();
	error = vfs_lookup(&result, refnode, addr->path, NULL, 0);
	if (error)
		goto leave;

	// check if its a socket node
	if (result->type != V_TYPE_SOCKET) {
		error = ENOTSOCK;
		// locked by vfs_lookup
		VOP_UNLOCK(result);
		goto leave;
	}

	// and we have write access
	error = VOP_ACCESS(result, V_ACCESS_WRITE, cred);
	if (error) {
		// locked by vfs_lookup
		VOP_UNLOCK(result);
		goto leave;
	}

	// and if it has a valid binding
	binding = result->socketbinding;
	// locked by vfs_lookup
	VOP_UNLOCK(result);
	if (binding == NULL) {
		error = ECONNREFUSED;
		goto leave;
	}

	// and if the server has not closed the socket and is listening
	MUTEX_ACQUIRE(&binding->mutex);
	localsocket_t *server = binding->server;
	if (server == NULL || server->listening == false || (server->socket.shutdown & SOCKET_SHUTDOWN_READ)) {
		error = ECONNREFUSED;
		goto leave;
	}

	if (server->seqpacket != localsocket->seqpacket) {
		error = EPROTOTYPE;
		goto leave;
	}

	// we have a listening socket! we have to add ourselves to the backlog now.

	// check if there is space to add in the backlog
	for (;;) {
		polldesc_t desc = {0};
		error = poll_initdesc(&desc, 1);
		if (error)
			goto leave;

		int revents = internalpoll(&server->socket, &desc.data[0], POLLOUT);

		if (revents) {
			poll_leave(&desc);
			poll_destroydesc(&desc);
			if (revents & POLLHUP) {
				error = ECONNREFUSED;
				goto leave;
			}
			break;
		}

		if (socket_nonblocking(socket, flags)) {
			error = EAGAIN;
			poll_leave(&desc);
			poll_destroydesc(&desc);
			goto leave;
		}

		// to make sure the binding doesn't get destroyed once we stop locking it, reference the vnode
		vnode_t *vn = binding->vnode;
		VOP_HOLD(vn);
		MUTEX_RELEASE(&binding->mutex);

		error = poll_dowait(&desc, localsocket->snd_timeout_us);
		if (error == ETIMEDOUT)
			error = EAGAIN;

		poll_leave(&desc);
		poll_destroydesc(&desc);

		if (error) {
			VOP_RELEASE(vn);
			binding = NULL;
			goto leave_nobinding;
		}

		MUTEX_ACQUIRE(&binding->mutex);
		server = binding->server;
		if (server == NULL || server->listening == false || (server->socket.shutdown & SOCKET_SHUTDOWN_READ)) {
			// server closed while we waited!
			// release the vnode so it can get deleted and set the binding to NULL as we can't be sure on its state anymore.
			MUTEX_RELEASE(&binding->mutex);
			VOP_RELEASE(vn);
			binding = NULL;
			error = ECONNREFUSED;
			goto leave;
		}

		if (server->seqpacket != localsocket->seqpacket) {
			MUTEX_RELEASE(&binding->mutex);
			VOP_RELEASE(vn);
			binding = NULL;
			error = EPROTOTYPE;
			goto leave;
		}

		VOP_RELEASE(vn);
	}

	localpair_t *pair = alloc(sizeof(localpair_t));
	if (pair == NULL) {
		error = ENOMEM;
		goto leave;
	}

	pair->server = server;
	pair->client = localsocket;
	MUTEX_INIT(&pair->mutex);
	MUTEX_ACQUIRE(&pair->mutex);

	pushbacklog(server, pair);

	// signal to the server they can accept() now
	poll_event(&server->socket.pollheader, POLLIN);

	// we will not use any more binding related shared data, unlock it.
	// we still hold the lock over the pair, though. this will prevent both the server
	// being destroyed and accepting us before we're ready.
	// also NULL the binding to indicate that its state will be uncertain from now on.
	MUTEX_RELEASE(&binding->mutex);
	binding = NULL;

	localsocket->pair = pair;
	MUTEX_RELEASE(&pair->mutex);

	error = 0;

	leave:
	if (binding)
		MUTEX_RELEASE(&binding->mutex);

	leave_nobinding:
	if (result)
		VOP_RELEASE(result);

	if (refnode)
		VOP_RELEASE(refnode);

	MUTEX_RELEASE(&socket->mutex);
	return error;
}

static int localsock_listen(socket_t *socket, int backlogsize) {
	localsocket_t *localsocket = (localsocket_t *)socket;
	MUTEX_ACQUIRE(&socket->mutex);
	
	int error = 0;
	// already listening
	if (localsocket->listening)
		goto leave;

	// connected/unbound socket
	if (localsocket->pair || localsocket->binding == NULL) {
		error = EOPNOTSUPP;
		goto leave;
	}

	if (current_thread()->proc) {
		localsocket->cred.pid = current_thread()->proc->pid;
		localsocket->cred.uid = current_thread()->proc->cred.euid;
		localsocket->cred.gid = current_thread()->proc->cred.egid;
	}

	MUTEX_ACQUIRE(&localsocket->binding->mutex);

	error = ringbuffer_init(&localsocket->backlog, backlogsize * sizeof(localpair_t *));
	if (error) {
		MUTEX_RELEASE(&localsocket->binding->mutex);
		goto leave;
	}

	localsocket->listening = true;

	MUTEX_RELEASE(&localsocket->binding->mutex);

	leave:
	MUTEX_RELEASE(&socket->mutex);
	return error;
}

static void localsock_destroy(socket_t *socket) {
	localsocket_t *localsocket = (localsocket_t *)socket;

	// disconnect the socket (connected/connecting sockets)
	localpair_t *pair = localsocket->pair;
	if (localsocket->pair) {
		MUTEX_ACQUIRE(&pair->mutex);
		localsocket_t **peerp;
		localsocket_t **ourp;

		if (pair->server == localsocket) {
			peerp = &pair->client;
			ourp = &pair->server;
		} else if (pair->client == localsocket) {
			peerp = &pair->server;
			ourp = &pair->client;
		} else {
			__assert(!"freeing socket which is neither client nor server");
		}

		if (*peerp == NULL) {
			free(pair);
		} else {
			poll_event(&(*peerp)->socket.pollheader, POLLHUP);
			*ourp = NULL;
			MUTEX_RELEASE(&pair->mutex);
		}
	}

	// leave binding (bound socket)
	binding_t *binding = localsocket->binding;
	if (binding) {
		MUTEX_ACQUIRE(&binding->mutex);
		__assert(binding->vnode);
		binding->server = NULL;
		MUTEX_RELEASE(&binding->mutex);
		VOP_RELEASE(binding->vnode);
	}

	// destroy backlog (for listening sockets)
	if (localsocket->listening) {
		localpair_t *pair = NULL;
		while (ringbuffer_read(&localsocket->backlog, &pair, sizeof(pair)) == sizeof(pair)) {
			MUTEX_ACQUIRE(&pair->mutex);
			if (pair->client == NULL) {
				// client gave up
				free(pair);
			} else {
				// client still trying to connect
				pair->server = NULL;
				MUTEX_RELEASE(&pair->mutex);
			}
		}

		ringbuffer_destroy(&localsocket->backlog);
		localsocket->listening = false;
	}

	poll_event(&socket->pollheader, POLLHUP);

	// the peer only accesses the socket while the pair mutex is acquired
	// since its gone now, we don't have to worry about the other end accessing it

	if (localsocket->bindpath)
		free(localsocket->bindpath);

	// discard any files in-flight
	file_t *file = NULL;
	while (ringbuffer_read(&localsocket->fds, &file, sizeof(file_t *)))
		fd_release(file);

	ringbuffer_destroy(&localsocket->fds);
	ringbuffer_destroy(&localsocket->ringbuffer);
	free(socket);
}

static int localsock_bind(socket_t *socket, sockaddr_t *addr, cred_t *cred) {
	MUTEX_ACQUIRE(&socket->mutex);
	int error;
	char *path = NULL;
	vnode_t *refnode = NULL;

	binding_t *binding = alloc(sizeof(binding_t));
	if (binding == NULL)
		return ENOMEM;
	
	path = alloc(strlen(addr->path) + 1);
	if (path == NULL) {
		error = ENOMEM;
		goto cleanup;
	}

	strcpy(path, addr->path);

	refnode = *path == '/' ? proc_get_root() : proc_get_cwd();
	vattr_t attr = {
		.mode = current_thread()->proc ? UMASK(0777) : 0644,
		.gid = current_thread()->proc ? current_thread()->proc->cred.gid : 0,
		.uid = current_thread()->proc ? current_thread()->proc->cred.uid : 0
	};

	vnode_t *vnode;
	error = vfs_create(refnode, path, &attr, V_TYPE_SOCKET, &vnode);
	if (error)
		goto cleanup;

	localsocket_t *localsocket = (localsocket_t *)socket;
	binding->vnode = vnode;
	binding->server = localsocket;
	MUTEX_INIT(&binding->mutex);

	vnode->socketbinding = binding;
	// locked by vfs_create
	VOP_UNLOCK(vnode);
	localsocket->binding = binding;
	localsocket->bindpath = path;

	cleanup:
	if (error) {
		free(path);
		if (binding)
			free(binding);
	}

	if (refnode) {
		VOP_RELEASE(refnode);
	}

	MUTEX_RELEASE(&socket->mutex);
	return error;
}

static size_t localsocket_datacount(socket_t *socket) {
	localsocket_t *localsocket = (localsocket_t *)socket;
	size_t nbytes = 0;
	MUTEX_ACQUIRE(&socket->mutex);
	if (localsocket->pair == NULL) 
		goto cleanup;

	MUTEX_ACQUIRE(&localsocket->pair->mutex);
	nbytes = RINGBUFFER_DATACOUNT(&localsocket->ringbuffer);
	MUTEX_RELEASE(&localsocket->pair->mutex);

	cleanup:
	MUTEX_RELEASE(&socket->mutex);
	return nbytes;
}

// only called on vnode remove (vfs_inactive)
void localsock_leavebinding(vnode_t *vnode) {
	binding_t *binding = vnode->socketbinding;
	MUTEX_ACQUIRE(&binding->mutex);
	if (binding->server) {
		// server socket still exists, so don't free the structure and only remove the vnode pointer from it
		binding->vnode = NULL;
		MUTEX_RELEASE(&binding->mutex);
	} else {
		// server socket is gone, destroy the binding structure
		free(binding);
	}

	// not really needed but added just to make sure
	vnode->socketbinding = NULL;
}

int localsock_pair(socket_t **ret1, socket_t **ret2, int protocol) {
	socket_t *socket1 = socket_create(SOCKET_TYPE_LOCAL, protocol);
	if (socket1 == NULL)
		return ENOMEM;

	socket_t *socket2 = socket_create(SOCKET_TYPE_LOCAL, protocol);
	if (socket2 == NULL) {
		localsock_destroy(socket1);
		return ENOMEM;
	}

	localpair_t *pair = alloc(sizeof(localpair_t));
	if (pair == NULL) {
		localsock_destroy(socket1);
		localsock_destroy(socket2);
		return ENOMEM;
	}

	MUTEX_INIT(&pair->mutex);
	pair->server = (localsocket_t *)socket1;
	pair->client = (localsocket_t *)socket2;
	((localsocket_t *)socket1)->pair = pair;
	((localsocket_t *)socket2)->pair = pair;

	if (current_thread()->proc) {
		pair->server->cred.pid = current_thread()->proc->pid;
		pair->server->cred.uid = current_thread()->proc->cred.euid;
		pair->server->cred.gid = current_thread()->proc->cred.egid;
		pair->client->cred = pair->server->cred;
	}

	*ret1 = socket1;
	*ret2 = socket2;
	return 0;
}

int localsock_pair_seqpacket(socket_t **ret1, socket_t **ret2, int protocol) {
	int error = localsock_pair(ret1, ret2, protocol);
	if (error)
		return error;

	localsocket_t *localsocket1 = (localsocket_t *)*ret1;
	localsocket_t *localsocket2 = (localsocket_t *)*ret2;

	localsocket1->seqpacket = true;
	localsocket2->seqpacket = true;

	return 0;
}

static int localsock_getname(socket_t *socket, sockaddr_t *addr) {
	localsocket_t *localsocket = (localsocket_t *)socket;
	MUTEX_ACQUIRE(&socket->mutex);
	if (localsocket->bindpath)
		strcpy(addr->path, localsocket->bindpath);
	else
		addr->path[0] = '\0';

	MUTEX_RELEASE(&socket->mutex);
	return 0;
}

static int localsock_getpeername(socket_t *socket, sockaddr_t *addr) {
	localsocket_t *localsocket = (localsocket_t *)socket;
	MUTEX_ACQUIRE(&socket->mutex);
	int error = 0;

	localpair_t *pair = localsocket->pair;
	if (pair == NULL) {
		error = ENOTCONN;
		goto leave;
	}

	MUTEX_ACQUIRE(&pair->mutex);

	localsocket_t *peer = pair->client == localsocket ? pair->server : pair->client;
	if (peer == NULL) {
		error = ENOTCONN;
		goto leave_pair;
	}

	if (peer->bindpath)
		strcpy(addr->path, peer->bindpath);
	else
		addr->path[0] = '\0';

	leave_pair:
	MUTEX_RELEASE(&pair->mutex);

	leave:
	MUTEX_RELEASE(&socket->mutex);
	return error;
}

static int localsock_shutdown(socket_t *socket, int how) {
	localsocket_t *localsocket = (localsocket_t *)socket;
	MUTEX_ACQUIRE(&socket->mutex);
	int error;
	localpair_t *pair = localsocket->pair;

	if (localsocket->listening) {
		socket->shutdown |= how;
		if (how & SOCKET_SHUTDOWN_READ)
			poll_event(&socket->pollheader, POLLHUP | POLLIN);

		error = 0;
		goto leave;
	}

	// socket not connected
	if (pair == NULL) {
		error = ENOTCONN;
		goto leave;
	}

	MUTEX_ACQUIRE(&pair->mutex);

	localsocket_t *peer = localsocket->pair->client == localsocket ? localsocket->pair->server : localsocket->pair->client;

	socket->shutdown |= how;

	int poll_events = 0;
	int this_events = 0;

	// XXX is this the correct behaviour?
	if (how & SOCKET_SHUTDOWN_READ) {
		poll_events |= POLLHUP;
		this_events |= POLLIN;
	}

	if (how & SOCKET_SHUTDOWN_WRITE) {
		poll_events |= POLLIN;
		this_events |= POLLOUT;
	}

	if (how == SOCKET_SHUTDOWN_RW)
		this_events |= POLLHUP;

	if (peer)
		poll_event(&peer->socket.pollheader, poll_events);

	poll_event(&socket->pollheader, this_events);

	error = 0;

	leave:
	if (pair)
		MUTEX_RELEASE(&pair->mutex);

	MUTEX_RELEASE(&socket->mutex);
	return error;
}

// called with socket locked
static int localsock_setopt(socket_t *socket, int layer, int optname, void *buffer, socklen_t len, cred_t *cred) {
	localsocket_t *localsocket = (localsocket_t *)socket;
	int error = 0;

	if (layer == SOL_SOCKET) {
		switch (optname) {
			case SO_RCVTIMEO:
			case SO_SNDTIMEO: {
				if (len != sizeof(timeval_t))
					return EINVAL;

				timeval_t tv;
				error = USERCOPY_POSSIBLY_FROM_USER(&tv, buffer, sizeof(tv));
				if (error)
					return error;

				time_t timeout_us;
				error = timeval_to_us(&tv, &timeout_us);
				if (error)
					return error;

				if (optname == SO_RCVTIMEO)
					localsocket->rcv_timeout_us = timeout_us;
				else
					localsocket->snd_timeout_us = timeout_us;
				break;
			}
			case SO_SNDLOWAT:
			case SO_OOBINLINE:
			case SO_KEEPALIVE:
			case SO_SNDBUF:
			case SO_RCVBUF:
			case SO_REUSEADDR:
			case SO_NO_CHECK:
				break;
			default:
				error = ENOPROTOOPT;
		}
	}

	if (error == ENOPROTOOPT) {
		printf("localsock_setopt: unknown %lu %lu called\n", layer, optname);
	}

	return error;
}

static int localsock_getopt(socket_t *socket, int layer, int optname, void *unsafe_buffer, socklen_t *unsafe_len, cred_t *cred) {
	localsocket_t *localsocket = (localsocket_t *)socket;
	socklen_t len;
	int error = USERCOPY_POSSIBLY_FROM_USER(&len, unsafe_len, sizeof(len));
	if (error)
		return error;

	if (layer == SOL_SOCKET) {
		switch (optname) {
			case SO_PEERCRED: {
				if (len != sizeof(localsock_cred_t))
					return EINVAL;

				localpair_t *pair = localsocket->pair;
				// socket not connected
				if (pair == NULL)
					return ENOTCONN;

				MUTEX_ACQUIRE(&pair->mutex);
				localsocket_t *peer = pair->client == localsocket ? pair->server : pair->client;
				if (peer == NULL) {
					MUTEX_RELEASE(&pair->mutex);
					return ENOTCONN;
				}

				localsock_cred_t cred = peer->cred;

				MUTEX_RELEASE(&pair->mutex);

				int error = USERCOPY_POSSIBLY_TO_USER(unsafe_buffer, &cred, sizeof(cred));
				return error;
			}
			case SO_SNDLOWAT: {
				int value = 1;
				return USERCOPY_POSSIBLY_TO_USER(unsafe_buffer, &value, len);
			}
			case SO_TYPE: {
				if (len != 4)
					return EINVAL;

				int type = localsocket->seqpacket ? SOCK_SEQPACKET : SOCK_STREAM;
				return USERCOPY_POSSIBLY_TO_USER(unsafe_buffer, &type, len);
			}
			case SO_OOBINLINE:
			case SO_KEEPALIVE:
			case SO_REUSEADDR:
			case SO_NO_CHECK:
				return USERCOPY_POSSIBLY_MEMSET_TO_USER(unsafe_buffer, 0, len);
			case SO_SNDBUF:
			case SO_RCVBUF: {
				if (len != 4)
					return EINVAL;

				int value = SOCKET_BUFFER;
				return USERCOPY_POSSIBLY_TO_USER(unsafe_buffer, &value, 4);
			}
			case SO_RCVTIMEO:
			case SO_SNDTIMEO: {
				if (len != sizeof(timeval_t))
					return EINVAL;

				timeval_t tv = timeval_from_us(optname == SO_RCVTIMEO ? localsocket->rcv_timeout_us : localsocket->snd_timeout_us);
				return USERCOPY_POSSIBLY_TO_USER(unsafe_buffer, &tv, sizeof(tv));
			}
		}
	}

	printf("localsock_getopt unknown %lu %lu\n", layer, optname);
	return ENOPROTOOPT;
}

static socketops_t socketops = {
	.bind = localsock_bind,
	.listen = localsock_listen,
	.connect = localsock_connect,
	.accept = localsock_accept,
	.send = localsock_send,
	.recv = localsock_recv,
	.destroy = localsock_destroy,
	.poll = localsock_poll,
	.datacount = localsocket_datacount,
	.getname = localsock_getname,
	.getpeername = localsock_getpeername,
	.shutdown = localsock_shutdown,
	.getopt = localsock_getopt,
	.setopt = localsock_setopt
};

socket_t *localsock_createsocket(int protocol) {
	// XXX possibly move this to a slab?
	localsocket_t *socket = alloc(sizeof(localsocket_t));
	if (socket == NULL)
		return NULL;

	// TODO would waste resources on listening sockets
	if (ringbuffer_init(&socket->ringbuffer, SOCKET_BUFFER)) {
		free(socket);
		return NULL;
	}

	if (ringbuffer_init(&socket->fds, FILE_COUNT * sizeof(file_t *))) {
		ringbuffer_destroy(&socket->ringbuffer);
		free(socket);
		return NULL;
	}

	socket->socket.ops = &socketops;

	return (socket_t *)socket;
}

socket_t *localsock_create_seqpacket_socket(int protocol) {
	localsocket_t *localsocket = (localsocket_t *)localsock_createsocket(protocol);
	if (localsocket == NULL)
		return NULL;

	localsocket->seqpacket = true;
	return &localsocket->socket;
}