feature (proc): implemented basic pipes and whole process pausing

inc/proc/pipe.h

@@ -0,0 +1,30 @@
+// This file is part of noxos and licensed under the MIT open source license
+
+#ifndef NOXOS_PIPE_H
+#define NOXOS_PIPE_H
+
+#ifndef NOX_PROCESS
+typedef struct process_T process_T;
+#endif
+
+#include "utils/stream.h"
+#include "utils/bitmap.h"
+
+#define PIPE_MAX_SENDERS        8
+#define PIPE_STD_STREAM_SIZE    0x1000
+
+typedef struct {
+    stream_T*   stream;
+    process_T*  receiver;
+    process_T*  senders         [PIPE_MAX_SENDERS];
+    bitmap_T    senders_bitmap;
+} pipe_T;
+
+void        pipe_init           (pipe_T* pipe, process_T* receiver);
+void        pipe_destruct       (pipe_T* pipe);
+bool        pipe_add_sender     (pipe_T* pipe, process_T* sender);
+void        pipe_remove_sender  (pipe_T* pipe, process_T* sender);
+uint64_t    pipe_write          (pipe_T* pipe, void* buffer_in, uint64_t n);
+uint64_t    pipe_read           (pipe_T* pipe, void* buffer_out, uint64_t n);
+
+#endif //NOXOS_PIPE_H

inc/proc/process.h

@@ -7,6 +7,7 @@
 #include "utils/string.h"
 #include "utils/bitmap.h"
 #include "proc/file_descriptor.h"
+#include "proc/pipe.h"
 #include "mm/page_map.h"
 #include "drivers/elf/elf.h"
 
@@ -20,6 +21,13 @@ typedef enum {
     PROCESS_KERNEL
 } processes_standard_E;
 
+typedef enum {
+    PROCESS_BLOCK_WAIT_PIPE_OUT,
+    PROCESS_BLOCK_WAIT_PIPE_IN,
+
+    PROCESS_BLOCK_ENUM_SIZE
+} process_blockers_E;
+
 typedef struct process_T process_T;
 struct process_T {
     char                        name        [128];
@@ -31,6 +39,12 @@ struct process_T {
     elf_executable_T*           executable;
     file_descriptor_array_T*    fd_array;
 
+    pipe_T*                     stdout;
+    pipe_T                      stdin;
+    pipe_T*                     stderr;
+
+    bitmap_T                    waiting;
+
     uint32_t                num_threads;
     void*                       threads;
     bitmap_T                    thread_ids;
@@ -45,6 +59,10 @@ void        process_kernel_spawn    (elf_executable_T* executable);
 pid_t       process_spawn           (pid_t parent, string_t name, elf_executable_T* executable, void* buffer);
 int32_t     process_get_thread_id   (process_T* process);
 void        process_clear_thread_id (process_T* process, uint32_t id);
+void        process_pause_pid       (pid_t pid);
+void        process_pause           (process_T* process);
+void        process_start_pid       (pid_t pid);
+void        process_start           (process_T* process);
 void        process_kill_pid        (pid_t pid);
 void        process_kill            (process_T* process);
 

inc/proc/scheduler.h

@@ -42,6 +42,8 @@ void            scheduler_start_thread              (thread_T* thread);
 void            scheduler_kill_thread               (thread_T* thread);
 
 pid_t           scheduler_register_process          (process_T* process);
+void            scheduler_pause_process             (process_T* process);
+void            scheduler_start_process             (process_T* process);
 void            scheduler_kill_process              (process_T* process);
 process_T*      scheduler_get_process               (pid_t pid);
 

src/platform/syscall.c

@@ -37,11 +37,34 @@ void syscall_handle_nx_fread(cpu_state_T* state) {
     uint8_t*          mem    = (uint8_t*)state->rdx;          // arg3
     uint64_t          n      = state->rcx;                    // arg4
 
-    vfs_node_T* node = file_descriptor_resolve(scheduler_get_current_process()->fd_array, fd);
+    process_T* process    = scheduler_get_current_process();
+    uint64_t   read_bytes = 0;
 
-    vfs_file_read(node, offset, n, mem);
+    switch (fd) {
+        case FILE_DESCRIPTOR_INVALID:
+        case FILE_DESCRIPTOR_STDOUT:
+        case FILE_DESCRIPTOR_STDERR: {
+            state->rax = STATUS_RESOURCE_NOT_AVAILABLE;
+            break;
+        }
 
-    state->rax = STATUS_SUCCESS;
+        case FILE_DESCRIPTOR_STDIN: {
+            read_bytes = pipe_read(&process->stdin, mem, n);
+            break;
+        }
+
+        default: {
+            vfs_node_T* node = file_descriptor_resolve(scheduler_get_current_process()->fd_array, fd);
+
+            // TODO: set read_bytes
+
+            vfs_file_read(node, offset, n, mem);
+
+            state->rax = STATUS_SUCCESS;
+            break;
+        }
+    }
+    // TODO: return read_bytes to process
 }
 
 void syscall_handle_nx_fwrite(cpu_state_T* state) {
@@ -50,15 +73,31 @@ void syscall_handle_nx_fwrite(cpu_state_T* state) {
     uint8_t*          mem    = (uint8_t*)state->rdx;          // arg3
     uint64_t          n      = state->rcx;                    // arg4
 
+    process_T* process       = scheduler_get_current_process();
+    uint64_t   written_bytes = 0;
+
     switch (fd) {
         case FILE_DESCRIPTOR_INVALID:
-        case FILE_DESCRIPTOR_STDIN:
+        case FILE_DESCRIPTOR_STDIN: {
-        case FILE_DESCRIPTOR_STDERR: {
             state->rax = STATUS_RESOURCE_NOT_AVAILABLE;
             break;
         }
 
+        case FILE_DESCRIPTOR_STDERR: {
+            if (process->stderr != NULL) {
+                written_bytes = pipe_write(process->stderr, mem, n);
+            }
+
+            log(LOG_DEBUG, "Syscall (nx_fwrite) to stderr: %s", mem);
+            state->rax = STATUS_SUCCESS;
+            break;
+        }
+
         case FILE_DESCRIPTOR_STDOUT: {
+            if (process->stdout != NULL) {
+                written_bytes = pipe_write(process->stdout, mem, n);
+            }
+
             log(LOG_DEBUG, "Syscall (nx_fwrite) to stdout: %s", mem);
             state->rax = STATUS_SUCCESS;
             break;
@@ -67,11 +106,15 @@ void syscall_handle_nx_fwrite(cpu_state_T* state) {
         default: {
             vfs_node_T* node = file_descriptor_resolve(scheduler_get_current_process()->fd_array, fd);
 
+            // TODO: set written_bytes
+
             vfs_file_write(node, offset, n, mem);
             state->rax = STATUS_SUCCESS;
             break;
         }
     }
+
+    // TODO: return written_bytes to process
 }
 
 void syscall_handle_nx_fdelete(cpu_state_T* state) {

src/proc/pipe.c

@@ -0,0 +1,77 @@
+// This file is part of noxos and licensed under the MIT open source license
+
+#include "proc/pipe.h"
+#include "proc/process.h"
+
+void pipe_init(pipe_T* pipe, process_T* receiver) {
+    pipe->stream         = stream_alloc(PIPE_STD_STREAM_SIZE);
+    pipe->receiver       = receiver;
+    pipe->senders_bitmap = bitmap_init(PIPE_MAX_SENDERS);
+}
+
+void pipe_destruct(pipe_T* pipe) {
+    bitmap_destruct(&pipe->senders_bitmap);
+    stream_destruct(pipe->stream);
+}
+
+
+bool pipe_add_sender(pipe_T* pipe, process_T* sender) {
+    for (int i = 0; i < PIPE_MAX_SENDERS; i++) {
+        if (bitmap_get(&pipe->senders_bitmap, i)) continue;
+
+        pipe->senders[i] = sender;
+        bitmap_set(&pipe->senders_bitmap, i, true);
+        return true;
+    }
+
+    return false;
+}
+
+void pipe_remove_sender(pipe_T* pipe, process_T* sender) {
+    for (int i = 0; i < PIPE_MAX_SENDERS; i++) {
+        if (!bitmap_get(&pipe->senders_bitmap, i)) continue;
+        if (pipe->senders[i] != sender) continue;
+
+        pipe->senders[i] = NULL;
+        bitmap_set(&pipe->senders_bitmap, i, false);
+    }
+}
+
+uint64_t pipe_write(pipe_T* pipe, void* buffer_in, uint64_t n) {
+    uint64_t written_bytes = stream_write(pipe->stream, buffer_in, n);
+
+    if (bitmap_get(&pipe->receiver->waiting, PROCESS_BLOCK_WAIT_PIPE_IN)) {
+        bitmap_set(&pipe->receiver->waiting, PROCESS_BLOCK_WAIT_PIPE_IN, false);
+        process_start(pipe->receiver);
+    }
+
+    if (written_bytes < n) {
+        for (int i = 0; i < PIPE_MAX_SENDERS; i++) {
+            if (!bitmap_get(&pipe->senders_bitmap, i)) continue;
+
+            bitmap_set(&pipe->senders[i]->waiting, PROCESS_BLOCK_WAIT_PIPE_OUT, true);
+            process_pause(pipe->senders[i]);
+        }
+    }
+
+    return written_bytes;
+}
+
+uint64_t pipe_read(pipe_T* pipe, void* buffer_out, uint64_t n) {
+    uint64_t read_bytes = stream_read(pipe->stream, buffer_out, n);
+
+    for (int i = 0; i < PIPE_MAX_SENDERS; i++) {
+        if (!bitmap_get(&pipe->senders_bitmap, i)) continue;
+        if (!bitmap_get(&pipe->senders[i]->waiting, PROCESS_BLOCK_WAIT_PIPE_OUT)) break;
+
+        bitmap_set(&pipe->senders[i]->waiting, PROCESS_BLOCK_WAIT_PIPE_OUT, false);
+        process_start(pipe->senders[i]);
+    }
+
+    if (read_bytes < n) {
+        bitmap_set(&pipe->receiver->waiting, PROCESS_BLOCK_WAIT_PIPE_IN, true);
+        process_pause(pipe->receiver);
+    }
+
+    return read_bytes;
+}

src/proc/process.c

@@ -13,7 +13,11 @@ void process_kernel_spawn(elf_executable_T* executable) {
     process->parent      = NULL;
     process->executable  = executable;
     process->fd_array    = file_descriptor_array_alloc();
+    process->stdout      = NULL;
+    process->stderr      = NULL;
+    process->waiting     = bitmap_init(PROCESS_BLOCK_ENUM_SIZE);
 
+    pipe_init(&process->stdin, process);
     memory_copy("kernel", process->name, 7);
 
     process->page_map = g_kernel_page_map;
@@ -25,13 +29,18 @@ void process_kernel_spawn(elf_executable_T* executable) {
 
 pid_t process_spawn(pid_t parent, string_t name, elf_executable_T* executable, void* buffer) {
     process_T* process = memory_allocate(sizeof(process_T));
+
     process->num_threads = 0;
     process->threads     = NULL;
     process->thread_ids  = bitmap_init(MAX_THREADS_PER_PROCESS);
     process->parent      = scheduler_get_process(parent);
     process->executable  = executable;
     process->fd_array    = file_descriptor_array_alloc();
+    process->stdout      = NULL;
+    process->stderr      = NULL;
+    process->waiting     = bitmap_init(PROCESS_BLOCK_ENUM_SIZE);
 
+    pipe_init(&process->stdin, process);
     memory_copy(name, process->name, MIN(string_length(name), 127));
 
     process->page_map = page_map_create();
@@ -56,6 +65,22 @@ void process_clear_thread_id(process_T* process, uint32_t id) {
     bitmap_set(&process->thread_ids, id, false);
 }
 
+void process_pause_pid(pid_t pid) {
+    process_pause(scheduler_get_process(pid));
+}
+
+void process_pause(process_T* process) {
+    scheduler_pause_process(process);
+}
+
+void process_start_pid(pid_t pid) {
+    process_start(scheduler_get_process(pid));
+}
+
+void process_start(process_T* process) {
+    scheduler_start_process(process);
+}
+
 void process_kill_pid(pid_t pid) {
     process_kill(scheduler_get_process(pid));
 }
@@ -63,6 +88,7 @@ void process_kill_pid(pid_t pid) {
 void process_kill(process_T* process) {
     scheduler_kill_process(process);
 
+    pipe_destruct(&process->stdin);
     file_descriptor_array_destruct(process->fd_array);
     elf_executable_destruct(process->executable);
     page_map_destruct(process->page_map);

src/proc/scheduler.c

@@ -203,6 +203,24 @@ pid_t scheduler_register_process(process_T* process) {
     return PROCESS_NONE;
 }
 
+void scheduler_pause_process(process_T* process) {
+    thread_T* thread = process->threads;
+    while (thread != NULL) {
+        thread_T* next = thread->local_next;
+        thread_pause(thread);
+        thread = next;
+    }
+}
+
+void scheduler_start_process(process_T* process) {
+    thread_T* thread = process->threads;
+    while (thread != NULL) {
+        thread_T* next = thread->local_next;
+        thread_start(thread);
+        thread = next;
+    }
+}
+
 void scheduler_kill_process(process_T* process) {
     scheduler_processes_chunk_T* chunk = process->chunk;