#include #include #include #include #include #include #include #include #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; }