#include #include #include #include #include #include #include #include #include #include #include #include #define BUFFER_SIZE 16 * PAGE_SIZE #define PIPE_ATOMIC_SIZE 4096 static scache_t *nodecache; static uintmax_t currentinode; static vops_t vnops; #define INTERNAL_LOCK(v) MUTEX_ACQUIRE(&(v)->lock) #define INTERNAL_UNLOCK(v) MUTEX_RELEASE(&(v)->lock) int pipefs_close(vnode_t *node, int flags, cred_t *cred) { INTERNAL_LOCK(node); pipenode_t *pipenode = (pipenode_t *)node; if (flags & V_FFLAGS_READ) { --pipenode->readers; } if (flags & V_FFLAGS_WRITE) { --pipenode->writers; } if (pipenode->readers == 0) poll_event(&pipenode->pollheader, POLLERR); if (pipenode->writers == 0) poll_event(&pipenode->pollheader, POLLHUP); INTERNAL_UNLOCK(node); return 0; } int pipefs_open(vnode_t **node, int flags, cred_t *cred) { if ((flags & (V_FFLAGS_WRITE | V_FFLAGS_READ)) == 0) return EINVAL; INTERNAL_LOCK(*node); pipenode_t *pipenode = (pipenode_t *)(*node); if (flags & V_FFLAGS_READ) { ++pipenode->readers; } if (flags & V_FFLAGS_WRITE) { ++pipenode->writers; } int error = 0; if (pipenode->writers > 0 && pipenode->readers > 0) goto leave; if (flags & V_FFLAGS_NONBLOCKING) { if (flags & V_FFLAGS_READ) { goto leave; } else if (flags & V_FFLAGS_WRITE) { error = ENXIO; goto leave; } } // from this point on, flags has only one of V_FFLAGS_READ or V_FFLAGS_WRITE // and there are either no writers or no readers. eventlistener_t listener; EVENT_INITLISTENER(&listener); if (flags & V_FFLAGS_READ) { EVENT_ATTACH(&listener, &pipenode->writeopenevent); } else { EVENT_ATTACH(&listener, &pipenode->readopenevent); } INTERNAL_UNLOCK(*node); error = EVENT_WAIT(&listener, 0); INTERNAL_LOCK(*node); EVENT_DETACHALL(&listener); leave: if (error) { INTERNAL_UNLOCK(*node); pipefs_close(*node, flags, cred); } else { if (flags & V_FFLAGS_READ) EVENT_SIGNAL(&pipenode->readopenevent); else EVENT_SIGNAL(&pipenode->writeopenevent); INTERNAL_UNLOCK(*node); } return error; } static int internalpoll(vnode_t *node, polldata_t *data, int events) { pipenode_t *pipenode = (pipenode_t *)node; int revents = 0; if (events & POLLIN) { events |= POLLHUP; if (pipenode->writers == 0) revents |= POLLHUP; else if (RINGBUFFER_DATACOUNT(&pipenode->data) > 0) revents |= POLLIN; } if (events & POLLOUT) { events |= POLLERR; if (pipenode->readers == 0) revents |= POLLERR; // poll will only return POLLOUT if an atomic write can be done without blocking // this is undocumented in POSIX but many unices implement it like this else if (RINGBUFFER_DATACOUNT(&pipenode->data) < BUFFER_SIZE - PIPE_ATOMIC_SIZE) revents |= POLLOUT; } if (revents == 0 && data) poll_add(&pipenode->pollheader, data, events); return revents; } int pipefs_read(vnode_t *node, iovec_iterator_t *iovec_iterator, size_t size, uintmax_t offset, int flags, size_t *readc, cred_t *cred) { pipenode_t *pipenode = (pipenode_t *)node; INTERNAL_LOCK(node); int error = 0; *readc = 0; // wait until there is data to read or return if nonblock while (1) { // check for available data or if the write end has been closed int revents = internalpoll(node, NULL, POLLIN); if (revents) break; if (flags & V_FFLAGS_NONBLOCKING) { error = EAGAIN; goto leave; } // do the wait polldesc_t desc = {0}; error = poll_initdesc(&desc, 1); if (error) goto leave; revents = internalpoll(node, &desc.data[0], POLLIN); INTERNAL_UNLOCK(node); error = poll_dowait(&desc, 0); poll_leave(&desc); poll_destroydesc(&desc); if (error) return error; INTERNAL_LOCK(node); } *readc = iovec_iterator_read_from_ringbuffer(iovec_iterator, &pipenode->data, size); if (*readc == RINGBUFFER_USER_COPY_FAILED) error = EFAULT; // signal that there is space to write for any threads blocked on this pipe if (RINGBUFFER_DATACOUNT(&pipenode->data) < BUFFER_SIZE - PIPE_ATOMIC_SIZE) poll_event(&pipenode->pollheader, POLLOUT); leave: INTERNAL_UNLOCK(node); return error; } static int writetopipe(pipenode_t *pipenode, iovec_iterator_t *iovec_iterator, size_t size, size_t *writec) { if (pipenode->readers == 0) { if (current_thread()->proc) signal_signalproc(current_thread()->proc, SIGPIPE); return EPIPE; } *writec = iovec_iterator_write_to_ringbuffer(iovec_iterator, &pipenode->data, size); if (*writec == RINGBUFFER_USER_COPY_FAILED) return EFAULT; __assert(*writec); poll_event(&pipenode->pollheader, POLLIN); return 0; } int pipefs_write(vnode_t *node, iovec_iterator_t *iovec_iterator, size_t size, uintmax_t offset, int flags, size_t *writec, cred_t *cred) { pipenode_t *pipenode = (pipenode_t *)node; INTERNAL_LOCK(node); int error = 0; *writec = 0; // there are 4 cases: // 1. blocking and n <= PIPE_ATOMIC_SIZE // atomic write, will only return once ALL the bytes can be written. // 2. nonblocking and n <= PIPE_ATOMIC_SIZE // if there is no room to write everything, fail with EAGAIN // 3. blocking and > PIPE_ATOMIC_SIZE // will only return when everything is written or a signal is caught, but is not guaranteed to be atomic. // 4. nonblocking and > PIPE_ATOMIC_SIZE // if pipe is full, fail with EAGAIN. otherwise, do a partial write. bool atomic = size <= PIPE_ATOMIC_SIZE; bool nonblock = flags & V_FFLAGS_NONBLOCKING; size_t case3written = 0; while (1) { // check for space or if the read end has been closed int revents = internalpoll(node, NULL, POLLOUT); if (revents & POLLERR) break; size_t freebytes = RINGBUFFER_FREESPACE(&pipenode->data); // case 1 if (nonblock == false && atomic && freebytes >= size) break; // case 2 if (nonblock && atomic && freebytes < size) { // if not enough space error = EAGAIN; goto leave; } else if (nonblock && atomic) { // write break; } // case 3 if (nonblock == false && atomic == false && freebytes) { size_t tmp; size_t writecount = min(freebytes, size - case3written); error = writetopipe(pipenode, iovec_iterator, writecount, &tmp); if (error) goto leave; case3written += writecount; __assert(tmp == writecount); if (case3written == size) { *writec = size; goto leave; } } if (nonblock && atomic == false && freebytes != 0) { // case 4 non not full pipe break; } else if (nonblock && atomic == false) { // case 4 full pipe error = EAGAIN; goto leave; } // do the wait polldesc_t desc = {0}; error = poll_initdesc(&desc, 1); if (error) goto leave; revents = internalpoll(node, &desc.data[0], POLLOUT); INTERNAL_UNLOCK(node); error = poll_dowait(&desc, 0); poll_leave(&desc); poll_destroydesc(&desc); if (error) return error; INTERNAL_LOCK(node); } error = writetopipe(pipenode, iovec_iterator, size, writec); leave: INTERNAL_UNLOCK(node); return error; } int pipefs_poll(vnode_t *node, polldata_t *data, int events) { int revents = 0; INTERNAL_LOCK(node); revents = internalpoll(node, data, events); INTERNAL_UNLOCK(node); return revents; } int pipefs_getattr(vnode_t *node, vattr_t *attr, cred_t *cred) { pipenode_t *pipenode = (pipenode_t *)node; INTERNAL_LOCK(node); *attr = pipenode->attr; attr->type = node->type; INTERNAL_UNLOCK(node); return 0; } int pipefs_setattr(vnode_t *node, vattr_t *attr, int which, cred_t *cred) { pipenode_t *pipenode = (pipenode_t *)node; INTERNAL_LOCK(node); if (which & V_ATTR_GID) pipenode->attr.gid = attr->gid; if (which & V_ATTR_UID) pipenode->attr.uid = attr->uid; if (which & V_ATTR_MODE) pipenode->attr.mode = attr->mode; if (which & V_ATTR_ATIME) pipenode->attr.atime = attr->atime; if (which & V_ATTR_MTIME) pipenode->attr.mtime = attr->mtime; if (which & V_ATTR_CTIME) pipenode->attr.ctime = attr->ctime; INTERNAL_UNLOCK(node); return 0; } int pipefs_inactive(vnode_t *node) { pipenode_t *pipenode = (pipenode_t *)node; INTERNAL_LOCK(node); ringbuffer_destroy(&pipenode->data); // TODO write this properly using cache ctor and dtor memset(node, 0, sizeof(pipenode_t)); VOP_INIT(node, &vnops, 0, V_TYPE_FIFO, NULL); pipenode->attr.inode = ++currentinode; POLL_INITHEADER(&pipenode->pollheader); EVENT_INITHEADER(&pipenode->writeopenevent); EVENT_INITHEADER(&pipenode->readopenevent); slab_free(nodecache, node); return 0; } // the mutex vnode is handled internally // (as we can sleep when waiting for data/space) // this does mean that operations are not atomic between a VOP_LOCK and VOP_UNLOCK static int pipefs_lock(vnode_t *vnode) { return 0; } static int pipefs_unlock(vnode_t *vnode) { return 0; } static int pipefs_advlock(vnode_t *node, int op, advlock_t *advlock) { return vfs_advlock(node, op, advlock); } static int pipefs_ioctl(vnode_t *node, unsigned long request, void *arg, int *result, cred_t *cred) { pipenode_t *pipenode = (pipenode_t *)node; int error = ENOTTY; INTERNAL_LOCK(node); switch (request) { case FIONREAD: { int i = RINGBUFFER_DATACOUNT(&pipenode->data); error = USERCOPY_POSSIBLY_TO_USER(arg, &i, sizeof(int)); break; } } INTERNAL_UNLOCK(node); return error; } static int pipefs_enodev() { return ENODEV; } static vops_t vnops = { .create = pipefs_enodev, .open = pipefs_open, .close = pipefs_close, .getattr = pipefs_getattr, .setattr = pipefs_setattr, .lookup = pipefs_enodev, .poll = pipefs_poll, .read = pipefs_read, .write = pipefs_write, .access = pipefs_enodev, .unlink = pipefs_enodev, .link = pipefs_enodev, .symlink = pipefs_enodev, .readlink = pipefs_enodev, .inactive = pipefs_inactive, .mmap = pipefs_enodev, .munmap = pipefs_enodev, .getdents = pipefs_enodev, .resize = pipefs_enodev, .rename = pipefs_enodev, .putpage = pipefs_enodev, .getpage = pipefs_enodev, .sync = pipefs_enodev, .advlock = pipefs_advlock, .ioctl = pipefs_ioctl, .lock = pipefs_lock, .unlock = pipefs_unlock }; static bool ctor(scache_t *cache, void *obj) { pipenode_t *node = obj; memset(node, 0, sizeof(pipenode_t)); VOP_INIT(&node->vnode, &vnops, 0, V_TYPE_FIFO, NULL); node->attr.inode = ++currentinode; POLL_INITHEADER(&node->pollheader); EVENT_INITHEADER(&node->writeopenevent); EVENT_INITHEADER(&node->readopenevent); return true; } void pipefs_init() { nodecache = slab_newcache(sizeof(pipenode_t), 0, ctor, NULL); __assert(nodecache); } INIT_ROUTINE_DEFINE(pipefs, INIT_ROUTINE_FLAGS_NONE, pipefs_init, bsp_early); int pipefs_newpipe(vnode_t **nodep) { vnode_t *vnode = slab_allocate(nodecache); pipenode_t *node = (pipenode_t *)vnode; if (node == NULL) return ENOMEM; int e = ringbuffer_init(&node->data, BUFFER_SIZE); if (e) { slab_free(nodecache, vnode); return e; } *nodep = vnode; return 0; } // assumes node is locked upon calling int pipefs_getbinding(vnode_t *node, vnode_t **pipep) { __assert(node->type == V_TYPE_FIFO); if (node->fifobinding) { *pipep = node->fifobinding; VOP_HOLD(*pipep); return 0; } int error = pipefs_newpipe(pipep); if (error) return error; VOP_HOLD(*pipep); node->fifobinding = *pipep; return error; } void pipefs_leavebinding(vnode_t *pipenode) { VOP_RELEASE(pipenode); }