Add primitive FIFO and hook it up to sys$pipe().

It's now possible to do this in bash:

cat kernel.map | fgrep List

This is very cool! :^)

AK/CircularQueue.h

@@ -5,32 +5,34 @@
 
 namespace AK {
 
-template<typename T, size_t capacity>
+template<typename T, size_t Capacity>
 class CircularQueue {
 public:
     CircularQueue()
     {
-        for (size_t i = 0; i < capacity; ++i)
+        for (size_t i = 0; i < Capacity; ++i)
             m_elements[i] = T();
     }
 
     bool isEmpty() const { return !m_size; }
     size_t size() const { return m_size; }
 
+    size_t capacity() const { return Capacity; }
+
     void dump() const
     {
-        kprintf("CircularQueue<%zu>:\n", capacity);
+        kprintf("CircularQueue<%zu>:\n", Capacity);
         kprintf(" size: %zu\n", m_size);
-        for (size_t i = 0; i < capacity; ++i) {
+        for (size_t i = 0; i < Capacity; ++i) {
             kprintf(" [%zu] %d %c\n", i, m_elements[i], i == m_head ? '*' : ' ');
         }
     }
 
     void enqueue(const T& t)
     {
-        m_elements[(m_head + m_size) % capacity] = t;
+        m_elements[(m_head + m_size) % Capacity] = t;
-        if (m_size == capacity)
+        if (m_size == Capacity)
-            m_head = (m_head + 1) % capacity;
+            m_head = (m_head + 1) % Capacity;
         else
             ++m_size;
     }
@@ -39,13 +41,13 @@ public:
     {
         ASSERT(!isEmpty());
         T value = m_elements[m_head];
-        m_head = (m_head + 1) % capacity;
+        m_head = (m_head + 1) % Capacity;
         --m_size;
         return value;
     }
 
 private:
-    T m_elements[capacity];
+    T m_elements[Capacity];
     size_t m_size { 0 };
     size_t m_head { 0 };
 };

Kernel/Console.cpp

@@ -43,7 +43,7 @@ ssize_t Console::write(const byte* data, size_t size)
         return 0;
     for (size_t i = 0; i < size; ++i)
         putChar(data[i]);
-    return 0;
+    return size;
 }
 
 void Console::putChar(char ch)

Kernel/FIFO.cpp

@@ -0,0 +1,87 @@
+#include "FIFO.h"
+#include <AK/StdLib.h>
+
+//#define FIFO_DEBUG
+
+RetainPtr<FIFO> FIFO::create()
+{
+    return adopt(*new FIFO);
+}
+
+FIFO::FIFO()
+{
+}
+
+void FIFO::open(Direction direction)
+{
+    if (direction == Reader) {
+        ++m_readers;
+#ifdef FIFO_DEBUG
+        kprintf("open reader (%u)\n", m_readers);
+#endif
+    } else if (direction == Writer) {
+        ++m_writers;
+#ifdef FIFO_DEBUG
+        kprintf("open writer (%u)\n", m_writers);
+#endif
+    }
+}
+
+void FIFO::close(Direction direction)
+{
+    if (direction == Reader) {
+#ifdef FIFO_DEBUG
+        kprintf("close reader (%u - 1)\n", m_readers);
+#endif
+        ASSERT(m_readers);
+        --m_readers;
+    } else if (direction == Writer) {
+#ifdef FIFO_DEBUG
+        kprintf("close writer (%u - 1)\n", m_writers);
+#endif
+        ASSERT(m_writers);
+        --m_writers;
+    }
+}
+
+bool FIFO::can_read() const
+{
+    return !m_queue.isEmpty() || !m_writers;
+}
+
+bool FIFO::can_write() const
+{
+#ifdef FIFO_DEBUG
+    dbgprintf("can_write? size(%u) < capacity(%u) || !readers(%u)\n", m_queue.size(), m_queue.capacity(), m_readers);
+#endif
+    return m_queue.size() < m_queue.capacity() || !m_readers;
+}
+
+ssize_t FIFO::read(byte* buffer, size_t size)
+{
+    if (!m_writers && m_queue.isEmpty())
+        return 0;
+#ifdef FIFO_DEBUG
+    dbgprintf("fifo: read(%u)\n",size);
+#endif
+    size_t nread = min(size, m_queue.size());
+    for (size_t i = 0; i < nread; ++i)
+        buffer[i] = m_queue.dequeue();
+#ifdef FIFO_DEBUG
+    dbgprintf("   -> read (%c) %u\n", buffer[0], nread);
+#endif
+    return nread;
+}
+
+ssize_t FIFO::write(const byte* buffer, size_t size)
+{
+    if (!m_readers)
+        return 0;
+#ifdef FIFO_DEBUG
+    dbgprintf("fifo: write(%p, %u)\n", buffer, size);
+#endif
+    size_t nwritten = min(size, m_queue.capacity() - m_queue.size());
+    for (size_t i = 0; i < nwritten; ++i)
+        m_queue.enqueue(buffer[i]);
+    return nwritten;
+}

Kernel/FIFO.h

@@ -0,0 +1,31 @@
+#pragma once
+
+#include <AK/CircularQueue.h>
+#include <AK/Retainable.h>
+#include <AK/RetainPtr.h>
+#include <VirtualFileSystem/UnixTypes.h>
+
+class FIFO : public Retainable<FIFO> {
+public:
+    enum Direction {
+        Neither, Reader, Writer
+    };
+
+    static RetainPtr<FIFO> create();
+
+    void open(Direction);
+    void close(Direction);
+
+    Unix::ssize_t write(const byte*, Unix::size_t);
+    Unix::ssize_t read(byte*, Unix::size_t);
+
+    bool can_read() const;
+    bool can_write() const;
+
+private:
+    FIFO();
+
+    unsigned m_writers { 0 };
+    unsigned m_readers { 0 };
+    CircularQueue<byte, 16> m_queue;
+};

Kernel/Makefile

@@ -19,6 +19,7 @@ KERNEL_OBJS = \
        RTC.o \
        TTY.o \
        VirtualConsole.o \
+       FIFO.o \
        Scheduler.o
 
 VFS_OBJS = \

Kernel/Process.cpp

@@ -16,6 +16,7 @@
 #include <LibC/signal_numbers.h>
 #include "Syscall.h"
 #include "Scheduler.h"
+#include "FIFO.h"
 
 //#define DEBUG_IO
 //#define TASK_DEBUG
@@ -982,7 +983,42 @@ ssize_t Process::sys$write(int fd, const void* data, size_t size)
     auto* descriptor = file_descriptor(fd);
     if (!descriptor)
         return -EBADF;
-    auto nwritten = descriptor->write((const byte*)data, size);
+    ssize_t nwritten = 0;
+    if (descriptor->isBlocking()) {
+        while (nwritten < (ssize_t)size) {
+#ifdef IO_DEBUG
+            dbgprintf("while %u < %u\n", nwritten, size);
+#endif
+            if (!descriptor->can_write()) {
+#ifdef IO_DEBUG
+                dbgprintf("block write on %d\n", fd);
+#endif
+                m_blocked_fd = fd;
+                block(BlockedWrite);
+                Scheduler::yield();
+            }
+            ssize_t rc = descriptor->write((const byte*)data + nwritten, size - nwritten);
+#ifdef IO_DEBUG
+            dbgprintf("   -> write returned %d\n", rc);
+#endif
+            if (rc < 0) {
+                // FIXME: Support returning partial nwritten with errno.
+                ASSERT(nwritten == 0);
+                return rc;
+            }
+            if (rc == 0)
+                break;
+            if (has_unmasked_pending_signals()) {
+                block(BlockedSignal);
+                Scheduler::yield();
+                if (nwritten == 0)
+                    return -EINTR;
+            }
+            nwritten += rc;
+        }
+    } else {
+        nwritten = descriptor->write((const byte*)data, size);
+    }
     if (has_unmasked_pending_signals()) {
         block(BlockedSignal);
         Scheduler::yield();
@@ -1174,10 +1210,34 @@ int Process::sys$open(const char* path, int options)
     return fd;
 }
 
+int Process::alloc_fd()
+{
+    int fd = -1;
+    for (int i = 0; i < (int)m_max_open_file_descriptors; ++i) {
+        if (!m_file_descriptors[i]) {
+            fd = i;
+            break;
+        }
+    }
+    return fd;
+}
+
 int Process::sys$pipe(int* pipefd)
 {
     VALIDATE_USER_WRITE(pipefd, sizeof(int) * 2);
-    ASSERT_NOT_REACHED();
+    if (number_of_open_file_descriptors() + 2 > max_open_file_descriptors())
+        return -EMFILE;
+    auto fifo = FIFO::create();
+
+    int reader_fd = alloc_fd();
+    m_file_descriptors[reader_fd] = FileDescriptor::create_pipe_reader(*fifo);
+    pipefd[0] = reader_fd;
+
+    int writer_fd = alloc_fd();
+    m_file_descriptors[writer_fd] = FileDescriptor::create_pipe_writer(*fifo);
+    pipefd[1] = writer_fd;
+
+    return 0;
 }
 
 int Process::sys$killpg(int pgrp, int signum)
@@ -1353,12 +1413,15 @@ void Process::unblock()
     m_state = Process::Runnable;
 }
 
-void Process::block(Process::State state)
+void Process::block(Process::State new_state)
 {
-    ASSERT(current->state() == Process::Running);
+    if (state() != Process::Running) {
+        kprintf("Process::block: %s(%u) block(%u/%s) with state=%u/%s\n", name().characters(), pid(), new_state, toString(new_state), state(), toString(state()));
+    }
+    ASSERT(state() == Process::Running);
     system.nblocked++;
     m_was_interrupted_while_blocked = false;
-    set_state(state);
+    set_state(new_state);
 }
 
 void block(Process::State state)

Kernel/Process.h

@@ -52,6 +52,7 @@ public:
         BlockedSleep,
         BlockedWait,
         BlockedRead,
+        BlockedWrite,
         BlockedSignal,
     };
 
@@ -226,6 +227,8 @@ private:
     int do_exec(const String& path, Vector<String>&& arguments, Vector<String>&& environment);
     void push_value_on_stack(dword);
 
+    int alloc_fd();
+
     PageDirectory* m_page_directory { nullptr };
 
     Process* m_prev { nullptr };
@@ -257,6 +260,7 @@ private:
     pid_t m_waitee { -1 };
     int m_waitee_status { 0 };
     int m_fdBlockedOnRead { -1 };
+    int m_blocked_fd { -1 };
     size_t m_max_open_file_descriptors { 16 };
     SignalActionData m_signal_action_data[32];
     dword m_pending_signals { 0 };
@@ -337,6 +341,7 @@ static inline const char* toString(Process::State state)
     case Process::BlockedSleep: return "Sleep";
     case Process::BlockedWait: return "Wait";
     case Process::BlockedRead: return "Read";
+    case Process::BlockedWrite: return "Write";
     case Process::BlockedSignal: return "Signal";
     case Process::BeingInspected: return "Inspect";
     }

Kernel/Scheduler.cpp

@@ -57,6 +57,13 @@ bool Scheduler::pick_next()
             return true;
         }
 
+        if (process.state() == Process::BlockedWrite) {
+            ASSERT(process.m_blocked_fd != -1);
+            if (process.m_file_descriptors[process.m_blocked_fd]->can_write())
+                process.unblock();
+            return true;
+        }
+
         if (process.state() == Process::Skip1SchedulerPass) {
             process.set_state(Process::Skip0SchedulerPasses);
             return true;

LibC/stdio.cpp

@@ -45,7 +45,6 @@ void __stdio_init()
 
 int setvbuf(FILE* stream, char* buf, int mode, size_t size)
 {
-    fprintf(stderr, "setvbuf(%p [fd=%d], %p, %d, %u)\n", stream, stream->fd, buf, mode, size);
     if (mode != _IONBF && mode != _IOLBF && mode != _IOFBF) {
         errno = EINVAL;
         return -1;

LibC/termcap.cpp

@@ -5,6 +5,8 @@
 #include <AK/HashMap.h>
 #include <AK/String.h>
 
+//#define TERMCAP_DEBUG
+
 extern "C" {
 
 char PC;
@@ -13,7 +15,9 @@ char* BC;
 
 int tgetent(char* bp, const char* name)
 {
+#ifdef TERMCAP_DEBUG
     fprintf(stderr, "tgetent: bp=%p, name='%s'\n", bp, name);
+#endif
     if (!strcmp(name, "ansi")) {
         PC = '\0';
         BC = const_cast<char*>("\033[D");
@@ -60,7 +64,9 @@ void ensure_caps()
 char* tgetstr(char* id, char** area)
 {
     ensure_caps();
-    fprintf(stderr, "tgetstr: id='%s', area=%p", id, area);
+#ifdef TERMCAP_DEBUG
+    fprintf(stderr, "tgetstr: id='%s'\n", id);
+#endif
     auto it = caps->find(id);
     if (it != caps->end()) {
         char* ret = *area;
@@ -74,7 +80,9 @@ char* tgetstr(char* id, char** area)
 
 int tgetflag(char* id)
 {
+#ifdef TERMCAP_DEBUG
     fprintf(stderr, "tgetflag: '%s'\n", id);
+#endif
     auto it = caps->find(id);
     if (it != caps->end())
         return 1;
@@ -83,11 +91,12 @@ int tgetflag(char* id)
 
 int tgetnum(char* id)
 {
+#ifdef TERMCAP_DEBUG
     fprintf(stderr, "tgetnum: '%s'\n", id);
+#endif
     auto it = caps->find(id);
-    if (it != caps->end()) {
+    if (it != caps->end())
         return atoi((*it).value);
-    }
     assert(false);
 }
 

Userland/fgrep.cpp

@@ -11,10 +11,10 @@ int main(int argc, char** argv)
     for (;;) {
         char buf[4096];
         fgets(buf, sizeof(buf), stdin);
-        if (feof(stdin))
-            return 0;
         if (strstr(buf, argv[1]))
             write(1, buf, strlen(buf));
+        if (feof(stdin))
+            return 0;
     }
     return 0;
 }

VirtualFileSystem/FileDescriptor.cpp

@@ -4,6 +4,7 @@
 #include <LibC/errno_numbers.h>
 #include "UnixTypes.h"
 #include <AK/BufferStream.h>
+#include "FIFO.h"
 
 #ifdef SERENITY
 #include "TTY.h"
@@ -14,6 +15,16 @@ RetainPtr<FileDescriptor> FileDescriptor::create(RetainPtr<VirtualFileSystem::No
     return adopt(*new FileDescriptor(move(vnode)));
 }
 
+RetainPtr<FileDescriptor> FileDescriptor::create_pipe_writer(FIFO& fifo)
+{
+    return adopt(*new FileDescriptor(fifo, FIFO::Writer));
+}
+
+RetainPtr<FileDescriptor> FileDescriptor::create_pipe_reader(FIFO& fifo)
+{
+    return adopt(*new FileDescriptor(fifo, FIFO::Reader));
+}
+
 FileDescriptor::FileDescriptor(RetainPtr<VirtualFileSystem::Node>&& vnode)
     : m_vnode(move(vnode))
 {
@@ -21,16 +32,27 @@ FileDescriptor::FileDescriptor(RetainPtr<VirtualFileSystem::Node>&& vnode)
 
 FileDescriptor::~FileDescriptor()
 {
+    if (m_fifo)
+        m_fifo->close(fifo_direction());
 }
 
 RetainPtr<FileDescriptor> FileDescriptor::clone()
 {
-    auto descriptor = FileDescriptor::create(m_vnode.copyRef());
+    RetainPtr<FileDescriptor> descriptor;
+    if (is_fifo()) {
+        descriptor = fifo_direction() == FIFO::Reader
+            ? FileDescriptor::create_pipe_reader(*m_fifo)
+            : FileDescriptor::create_pipe_writer(*m_fifo);
+    } else {
+        descriptor = FileDescriptor::create(m_vnode.copyRef());
+    }
     if (!descriptor)
         return nullptr;
     descriptor->m_currentOffset = m_currentOffset;
 #ifdef SERENITY
     descriptor->m_isBlocking = m_isBlocking;
+    descriptor->m_fd_flags = m_fd_flags;
+    descriptor->m_file_flags = m_file_flags;
 #endif
     return descriptor;
 }
@@ -46,6 +68,7 @@ bool additionWouldOverflow(Unix::off_t a, Unix::off_t b)
 
 int FileDescriptor::stat(Unix::stat* buffer)
 {
+    ASSERT(!is_fifo());
     if (!m_vnode)
         return -EBADF;
 
@@ -71,6 +94,7 @@ int FileDescriptor::stat(Unix::stat* buffer)
 
 Unix::off_t FileDescriptor::seek(Unix::off_t offset, int whence)
 {
+    ASSERT(!is_fifo());
     if (!m_vnode)
         return -EBADF;
 
@@ -112,6 +136,10 @@ Unix::off_t FileDescriptor::seek(Unix::off_t offset, int whence)
 
 Unix::ssize_t FileDescriptor::read(byte* buffer, Unix::size_t count)
 {
+    if (is_fifo()) {
+        ASSERT(fifo_direction() == FIFO::Reader);
+        return m_fifo->read(buffer, count);
+    }
     if (m_vnode->isCharacterDevice()) {
         // FIXME: What should happen to m_currentOffset?
         return m_vnode->characterDevice()->read(buffer, count);
@@ -123,6 +151,10 @@ Unix::ssize_t FileDescriptor::read(byte* buffer, Unix::size_t count)
 
 Unix::ssize_t FileDescriptor::write(const byte* data, Unix::size_t size)
 {
+    if (is_fifo()) {
+        ASSERT(fifo_direction() == FIFO::Writer);
+        return m_fifo->write(data, size);
+    }
     if (m_vnode->isCharacterDevice()) {
         // FIXME: What should happen to m_currentOffset?
         return m_vnode->characterDevice()->write(data, size);
@@ -132,8 +164,21 @@ Unix::ssize_t FileDescriptor::write(const byte* data, Unix::size_t size)
     return -1;
 }
 
+bool FileDescriptor::can_write()
+{
+    if (is_fifo()) {
+        ASSERT(fifo_direction() == FIFO::Writer);
+        return m_fifo->can_write();
+    }
+    return true;
+}
+
 bool FileDescriptor::hasDataAvailableForRead()
 {
+    if (is_fifo()) {
+        ASSERT(fifo_direction() == FIFO::Reader);
+        return m_fifo->can_read();
+    }
     if (m_vnode->isCharacterDevice())
         return m_vnode->characterDevice()->hasDataAvailableForRead();
     return true;
@@ -141,6 +186,8 @@ bool FileDescriptor::hasDataAvailableForRead()
 
 ByteBuffer FileDescriptor::readEntireFile()
 {
+    ASSERT(!is_fifo());
+
     if (m_vnode->isCharacterDevice()) {
         auto buffer = ByteBuffer::createUninitialized(1024);
         Unix::ssize_t nread = m_vnode->characterDevice()->read(buffer.pointer(), buffer.size());
@@ -153,6 +200,7 @@ ByteBuffer FileDescriptor::readEntireFile()
 
 bool FileDescriptor::isDirectory() const
 {
+    ASSERT(!is_fifo());
     return m_vnode->metadata().isDirectory();
 }
 
@@ -185,6 +233,8 @@ ssize_t FileDescriptor::get_dir_entries(byte* buffer, Unix::size_t size)
 #ifdef SERENITY
 bool FileDescriptor::isTTY() const
 {
+    if (is_fifo())
+        return false;
     if (auto* device = m_vnode->characterDevice())
         return device->isTTY();
     return false;
@@ -192,6 +242,8 @@ bool FileDescriptor::isTTY() const
 
 const TTY* FileDescriptor::tty() const
 {
+    if (is_fifo())
+        return nullptr;
     if (auto* device = m_vnode->characterDevice())
         return static_cast<const TTY*>(device);
     return nullptr;
@@ -199,6 +251,8 @@ const TTY* FileDescriptor::tty() const
 
 TTY* FileDescriptor::tty()
 {
+    if (is_fifo())
+        return nullptr;
     if (auto* device = m_vnode->characterDevice())
         return static_cast<TTY*>(device);
     return nullptr;
@@ -216,5 +270,18 @@ String FileDescriptor::absolute_path() const
     if (isTTY())
         return tty()->ttyName();
 #endif
+    if (is_fifo()) {
+        char buf[32];
+        ksprintf(buf, "fifo:%x", m_fifo.ptr());
+        return buf;
+    }
     return VirtualFileSystem::the().absolutePath(m_vnode->inode);
 }
+
+FileDescriptor::FileDescriptor(FIFO& fifo, FIFO::Direction direction)
+    : m_isBlocking(true)
+    , m_fifo(fifo)
+    , m_fifo_direction(direction)
+{
+    m_fifo->open(direction);
+}

VirtualFileSystem/FileDescriptor.h

@@ -2,7 +2,9 @@
 
 #include "VirtualFileSystem.h"
 #include "InodeMetadata.h"
+#include "FIFO.h"
 #include <AK/ByteBuffer.h>
+#include <AK/CircularQueue.h>
 #include <AK/Retainable.h>
 
 #ifdef SERENITY
@@ -12,6 +14,8 @@ class TTY;
 class FileDescriptor : public Retainable<FileDescriptor> {
 public:
     static RetainPtr<FileDescriptor> create(RetainPtr<VirtualFileSystem::Node>&&);
+    static RetainPtr<FileDescriptor> create_pipe_writer(FIFO&);
+    static RetainPtr<FileDescriptor> create_pipe_reader(FIFO&);
     ~FileDescriptor();
 
     RetainPtr<FileDescriptor> clone();
@@ -24,6 +28,7 @@ public:
     int stat(Unix::stat*);
 
     bool hasDataAvailableForRead();
+    bool can_write();
 
     ssize_t get_dir_entries(byte* buffer, Unix::size_t);
 
@@ -52,6 +57,9 @@ public:
 
     dword fd_flags() const { return m_fd_flags; }
     int set_fd_flags(dword flags) { m_fd_flags = flags; return 0; }
+
+    bool is_fifo() const { return m_fifo; }
+    FIFO::Direction fifo_direction() { return m_fifo_direction; }
 #endif
 
     ByteBuffer& generatorCache() { return m_generatorCache; }
@@ -59,6 +67,7 @@ public:
 private:
     friend class VirtualFileSystem;
     explicit FileDescriptor(RetainPtr<VirtualFileSystem::Node>&&);
+    FileDescriptor(FIFO&, FIFO::Direction);
 
     RetainPtr<VirtualFileSystem::Node> m_vnode;
 
@@ -70,6 +79,9 @@ private:
     bool m_isBlocking { true };
     dword m_fd_flags { 0 };
     dword m_file_flags { 0 };
+
+    RetainPtr<FIFO> m_fifo;
+    FIFO::Direction m_fifo_direction { FIFO::Neither };
 #endif
 };