#include #include #include #include #include #include #define SOCKET_BUFFER (256 * 1024) struct localpair_t; struct binding_t; typedef struct { socket_t socket; ringbuffer_t ringbuffer; // shared with pair // client/connected server struct localpair_t *pair; // listening server struct binding_t *binding; struct localpair_t **backlog; // shared with binding int backlogsize; // shared with binding uintmax_t backlogcurrentwrite; // shared with binding uintmax_t backlogcurrentread; // shared with binding char *bindpath; } 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 void pushbacklog(localsocket_t *server, localpair_t *pair) { __assert(server->backlogcurrentwrite < server->backlogcurrentread + server->backlogsize); server->backlog[server->backlogcurrentwrite++] = pair; } static inline localpair_t *popbacklog(localsocket_t *server) { __assert(server->backlogcurrentread < server->backlogcurrentwrite); localpair_t *pair = server->backlog[server->backlogcurrentread]; server->backlog[server->backlogcurrentread++] = NULL; return pair; } static int datapoll(localsocket_t *local, localsocket_t *peer, int events) { int revents = 0; // data to read? if ((events & POLLIN) && RINGBUFFER_DATACOUNT(&local->ringbuffer)) revents |= POLLIN; // space to write? if ((events & POLLOUT) && RINGBUFFER_DATACOUNT(&peer->ringbuffer) != SOCKET_BUFFER) revents |= POLLOUT; return revents; } static int listeningpoll(localsocket_t *localsocket, int events) { int revents = 0; // can insert into backlog? if (localsocket->backlogcurrentwrite < localsocket->backlogcurrentread + localsocket->backlogsize) revents |= (events & POLLOUT) ? POLLOUT : 0; // can accept()? if (localsocket->backlogcurrentwrite != localsocket->backlogcurrentread) 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; // 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->backlog) 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; 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->backlog) 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, false); if (localsocket->binding) MUTEX_ACQUIRE(&localsocket->binding->mutex, false); if (localsocket->pair) MUTEX_ACQUIRE(&localsocket->pair->mutex, false); 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; } static int localsock_send(socket_t *socket, sockaddr_t *addr, void *buffer, size_t count, uintmax_t flags, size_t *sendcount) { localsocket_t *localsocket = (localsocket_t *)socket; MUTEX_ACQUIRE(&socket->mutex, false); int error; localpair_t *pair = localsocket->pair; // socket not connected if (pair == NULL) { error = ENOTCONN; goto leave; } MUTEX_ACQUIRE(&pair->mutex, false); // check/wait for space 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], POLLOUT); if (revents) { poll_leave(&desc); poll_destroydesc(&desc); break; } if (flags & V_FFLAGS_NONBLOCKING) { error = EAGAIN; poll_leave(&desc); poll_destroydesc(&desc); goto leave; } MUTEX_RELEASE(&pair->mutex); error = poll_dowait(&desc, 0); poll_leave(&desc); poll_destroydesc(&desc); MUTEX_ACQUIRE(&pair->mutex, false); if (error) goto leave; } localsocket_t *peer = pair->client == localsocket ? pair->server : pair->client; if (peer == NULL) { if (_cpu()->thread->proc) signal_signalproc(_cpu()->thread->proc, SIGPIPE); error = EPIPE; goto leave; } *sendcount = ringbuffer_write(&peer->ringbuffer, buffer, count); __assert(*sendcount > 0); // 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, sockaddr_t *addr, void *buffer, size_t count, uintmax_t flags, size_t *recvcount) { localsocket_t *localsocket = (localsocket_t *)socket; MUTEX_ACQUIRE(&socket->mutex, false); int error; localpair_t *pair = localsocket->pair; // socket not connected if (pair == NULL) { error = ENOTCONN; goto leave; } MUTEX_ACQUIRE(&pair->mutex, false); // 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) { poll_leave(&desc); poll_destroydesc(&desc); break; } if (flags & V_FFLAGS_NONBLOCKING) { error = EAGAIN; poll_leave(&desc); poll_destroydesc(&desc); goto leave; } MUTEX_RELEASE(&pair->mutex); error = poll_dowait(&desc, 0); poll_leave(&desc); poll_destroydesc(&desc); MUTEX_ACQUIRE(&pair->mutex, false); if (error) goto leave; } if (flags & SOCKET_RECV_FLAGS_PEEK) { *recvcount = ringbuffer_peek(&localsocket->ringbuffer, buffer, 0, count); } else { *recvcount = ringbuffer_read(&localsocket->ringbuffer, buffer, count); 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 && *recvcount) poll_event(&peer->socket.pollheader, POLLOUT); } 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; MUTEX_ACQUIRE(&server->socket.mutex, false); binding_t *binding = server->binding; int error; if (binding == NULL) { error = EINVAL; goto leave; } MUTEX_ACQUIRE(&server->binding->mutex, false); if (server->backlog == NULL) { 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); pair = popbacklog(server); MUTEX_ACQUIRE(&pair->mutex, false); if (pair->client == NULL) { // client closed the connection, delete the pair and keep trying free(pair); continue; } break; } if (flags & V_FFLAGS_NONBLOCKING) { error = EAGAIN; poll_leave(&desc); poll_destroydesc(&desc); goto leave; } MUTEX_RELEASE(&binding->mutex); error = poll_dowait(&desc, 0); poll_leave(&desc); poll_destroydesc(&desc); if (error) goto leave; MUTEX_ACQUIRE(&binding->mutex, false); } // 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; // signal to the client that they can start sending data now poll_event(&pair->client->socket.pollheader, POLLOUT); MUTEX_RELEASE(&pair->mutex); leave: if (binding) MUTEX_RELEASE(&binding->mutex); MUTEX_RELEASE(&server->socket.mutex); return error; } static int localsock_connect(socket_t *socket, sockaddr_t *addr, uintmax_t flags) { localsocket_t *localsocket = (localsocket_t *)socket; int error; vnode_t *refnode = NULL; vnode_t *result = NULL; binding_t *binding = NULL; MUTEX_ACQUIRE(&socket->mutex, false); // listening/bound socket if (localsocket->backlog || localsocket->binding) { error = EOPNOTSUPP; goto leave; } // connected socket if (localsocket->pair) { error = EISCONN; goto leave; } // get the vnode specified in addr refnode = addr->path[0] == '/' ? sched_getroot() : sched_getcwd(); 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; goto leave; } // and if it has a valid binding binding = result->socketbinding; if (binding == NULL) { error = ECONNREFUSED; goto leave; } // and if the server has not closed the socket and is listening MUTEX_ACQUIRE(&binding->mutex, false); localsocket_t *server = binding->server; if (server == NULL || server->backlog == NULL) { error = ECONNREFUSED; 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); break; } if (flags & V_FFLAGS_NONBLOCKING) { 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 VOP_HOLD(binding->vnode); MUTEX_RELEASE(&binding->mutex); error = poll_dowait(&desc, 0); poll_leave(&desc); poll_destroydesc(&desc); if (error) goto leave; MUTEX_ACQUIRE(&binding->mutex, false); if (binding->server == NULL) { // 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. VOP_RELEASE(binding->vnode); binding = NULL; error = ECONNREFUSED; goto leave; } VOP_RELEASE(binding->vnode); } 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, false); 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; // if nonblocking, return success. the userspace/whatever will // poll() the client to know when they can send data or if the server gave up and closed if (flags & V_FFLAGS_NONBLOCKING) { error = 0; goto leave; } // otherwise, we will just do the same thing as userspace (but here) polldesc_t desc = {0}; error = poll_initdesc(&desc, 1); if (error) goto leave; int revents = internalpoll(socket, &desc.data[0], POLLOUT); __assert(revents == 0); // release and wait. once we return, we will be connected. MUTEX_RELEASE(&pair->mutex); error = poll_dowait(&desc, 0); poll_leave(&desc); poll_destroydesc(&desc); if (error) goto leave; leave: if (binding) MUTEX_RELEASE(&binding->mutex); 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, false); int error = 0; // already listening if (localsocket->backlog) goto leave; // connected/unbound socket if (localsocket->pair || localsocket->binding == NULL) { error = EOPNOTSUPP; goto leave; } MUTEX_ACQUIRE(&localsocket->binding->mutex, false); localsocket->backlog = alloc(backlogsize * sizeof(localpair_t *)); if (localsocket->backlog == NULL) { MUTEX_RELEASE(&localsocket->binding->mutex); error = ENOMEM; goto leave; } localsocket->backlogsize = backlogsize; 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, false); 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 { *ourp = NULL; MUTEX_RELEASE(&pair->mutex); } } // leave binding (bound socket) binding_t *binding = localsocket->binding; if (binding) { MUTEX_ACQUIRE(&binding->mutex, false); __assert(binding->vnode); binding->server = NULL; MUTEX_RELEASE(&binding->mutex); VOP_RELEASE(binding->vnode); } // destroy backlog (for listening sockets) if (localsocket->backlog) { for (int i = 0; i < localsocket->backlogsize; ++i) { localpair_t *pair = localsocket->backlog[i]; if (pair == NULL) continue; MUTEX_ACQUIRE(&pair->mutex, false); if (pair->client == NULL) { // client gave up free(pair); } else { // client still trying to connect pair->server = NULL; MUTEX_RELEASE(&pair->mutex); } } free(localsocket->backlog); localsocket->backlog = NULL; } 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); ringbuffer_destroy(&localsocket->ringbuffer); free(socket); } static int localsock_bind(socket_t *socket, sockaddr_t *addr) { MUTEX_ACQUIRE(&socket->mutex, false); 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 == '/' ? sched_getroot() : sched_getcwd(); vattr_t attr = { .mode = _cpu()->thread->proc ? UMASK(0777) : 0644, .gid = _cpu()->thread->proc ? _cpu()->thread->proc->cred.gid : 0, .uid = _cpu()->thread->proc ? _cpu()->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); VOP_LOCK(vnode); vnode->socketbinding = binding; 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, false); if (localsocket->pair == NULL) goto cleanup; MUTEX_ACQUIRE(&localsocket->pair->mutex, false); 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, false); 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) { socket_t *socket1 = socket_create(SOCKET_TYPE_LOCAL); if (socket1 == NULL) return ENOMEM; socket_t *socket2 = socket_create(SOCKET_TYPE_LOCAL); 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; *ret1 = socket1; *ret2 = socket2; return 0; } static int localsock_getname(socket_t *socket, sockaddr_t *addr) { localsocket_t *localsocket = (localsocket_t *)socket; MUTEX_ACQUIRE(&socket->mutex, false); 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) { addr->path[0] = '\0'; return 0; } 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 }; socket_t *localsock_createsocket() { // 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; } socket->socket.ops = &socketops; return (socket_t *)socket; }