ncd: modules: add sys.start_process()

blog_channels.txt

@@ -135,3 +135,4 @@ ncd_backtrack 4
 ncd_socket 4
 ncd_depend_scope 4
 ncd_substr 4
+ncd_sys_start_process 4

generated/blog_channel_ncd_sys_start_process.h

@@ -0,0 +1,4 @@
+#ifdef BLOG_CURRENT_CHANNEL
+#undef BLOG_CURRENT_CHANNEL
+#endif
+#define BLOG_CURRENT_CHANNEL BLOG_CHANNEL_ncd_sys_start_process

generated/blog_channels_defines.h

@@ -135,4 +135,5 @@
 #define BLOG_CHANNEL_ncd_socket 134
 #define BLOG_CHANNEL_ncd_depend_scope 135
 #define BLOG_CHANNEL_ncd_substr 136
-#define BLOG_NUM_CHANNELS 137
+#define BLOG_CHANNEL_ncd_sys_start_process 137
+#define BLOG_NUM_CHANNELS 138

generated/blog_channels_list.h

@@ -135,3 +135,4 @@
 {"ncd_socket", 4},
 {"ncd_depend_scope", 4},
 {"ncd_substr", 4},
+{"ncd_sys_start_process", 4},

ncd/CMakeLists.txt

@@ -63,6 +63,7 @@ if (NOT EMSCRIPTEN)
         modules/net_ipv6_addr.c
         modules/net_ipv6_route.c
         modules/socket.c
+        modules/sys_start_process.c
     )
     
     list(APPEND NCD_ADDITIONAL_LIBS

ncd/modules/modules.h

@@ -113,6 +113,7 @@ extern const struct NCDModuleGroup ncdmodule_reboot;
 extern const struct NCDModuleGroup ncdmodule_net_ipv6_addr;
 extern const struct NCDModuleGroup ncdmodule_net_ipv6_route;
 extern const struct NCDModuleGroup ncdmodule_socket;
+extern const struct NCDModuleGroup ncdmodule_sys_start_process;
 #endif
 
 static const struct NCDModuleGroup *ncd_modules[] = {
@@ -195,6 +196,7 @@ static const struct NCDModuleGroup *ncd_modules[] = {
     &ncdmodule_net_ipv6_addr,
     &ncdmodule_net_ipv6_route,
     &ncdmodule_socket,
+    &ncdmodule_sys_start_process,
 #endif
     NULL
 };

ncd/modules/sys_start_process.c

@@ -0,0 +1,1137 @@
+/**
+ * @file sys_start_process.c
+ * @author Ambroz Bizjak <ambrop7@gmail.com>
+ * 
+ * @section LICENSE
+ * 
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ * 1. Redistributions of source code must retain the above copyright
+ *    notice, this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright
+ *    notice, this list of conditions and the following disclaimer in the
+ *    documentation and/or other materials provided with the distribution.
+ * 3. Neither the name of the author nor the
+ *    names of its contributors may be used to endorse or promote products
+ *    derived from this software without specific prior written permission.
+ * 
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+ * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+ * DISCLAIMED. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY
+ * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+ * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+ * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+ * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+ * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ * 
+ * @section DESCRIPTION
+ * 
+ * Synopsis:
+ *   sys.start_process(list command, string mode)
+ * 
+ * Variables:
+ *   is_error - "true" if there was an error starting the process, "false" if the process
+ *              has been started successfully
+ * 
+ * Synopsis:
+ *   sys.start_process::wait()
+ * 
+ * Variables:
+ *   exit_status - the exit code if the process terminated normally, -1 if it terminated
+ *                 with a signal
+ * 
+ * Synopsis:
+ *   sys.start_process::terminate()
+ *   sys.start_process::kill()
+ * 
+ * Synopsis:
+ *   sys.start_process::read_pipe()
+ * 
+ * Description:
+ *   Creates a read interface to the process's standard output. Data is read using the
+ *   read() method on this object. Read errors are reported implicitly by this statement
+ *   going down and the 'is_error' variable changing to "true".
+ *   When read_pipe() is initialized for a process, it takes ownership of the read pipe
+ *   to the process. When read_pipe() is requested to terminate, it will close the pipe.
+ *   Attempting to initialize read_pipe() on a process which was not started with 'r'
+ *   in the mode argument, or where another read_pipe() object has already taken ownership
+ *   of the read pipe, will result in throwing an error to the interpreter.
+ * 
+ * Variables:
+ *   string is_error - "true" if there was a read error, "false" if not
+ * 
+ * Synopsis:
+ *   sys.start_process::read_pipe::read()
+ * 
+ * Description:
+ *   Reads some data. If a read error occurs, it is reported implicitly via the
+ *   read_pipe() object going down. If end of file is reached, this and any future read()
+ *   operations will indicate that via the 'not_eof' variable. It is guaranteed that after
+ *   EOF is reached, the read_pipe() object will not go down to report any errors.
+ *   WARNING: if a read() is requested to terminate before it has completed, the
+ *            read_pipe() will become unusable and any read() invocation after that will
+ *            throw an error to the interpreter.
+ * 
+ * Variables:
+ *   string (empty) - data that was read, or an empty string on EOF
+ *   string not_eof - "true" is EOF was not reached, "false" if it was
+ * 
+ * Synopsis:
+ *   sys.start_process::write_pipe()
+ * 
+ * Description:
+ *   Creates a write interface to the process's standard input. Data is written using the
+ *   write() method on this object. Write errors are reported implicitly by this statement
+ *   going down and the ''is_error variable changing to "true".
+ *   When write_pipe() is initialized for a process, it takes ownership of the write pipe
+ *   to the process. When write_pipe() is requested to terminate, it will close the pipe
+ *   (unless the close() has been used).
+ *   Attempting to initialize write_pipe() on a process which was not started with 'w'
+ *   in the mode argument, or where another write_pipe() object has already taken ownership
+ *   of the write pope, will result in throwing an error to the interpreter.
+ * 
+ * Variables:
+ *   string is_error - "true" if there was a write error, "false" if not
+ * 
+ * Synopsis:
+ *   sys.start_process::write_pipe::write(string data)
+ * 
+ * Description:
+ *   Writes the given data. If a write error occurs, it is reported implicitly via the
+ *   write_pipe() object going down.
+ *   WARNING: if a write() is requested to terminate before it has completed, the
+ *            write_pipe() will become unusable and any write() or close() invocation after
+ *            that will throw an error to the interpreter.
+ * 
+ * Synopsis:
+ *   sys.start_process::write_pipe::close(string data)
+ * 
+ * Description:
+ *   Closes the write pipe. This will make whatever is reading the other end of the pipe
+ *   encounter EOF after it has read any pending data. It is guaranteed that after the
+ *   pipe is closed, the write_pipe() object will not go down to report any errors.
+ *   After close() is performed, any further write() or close() calls are disallowed and
+ *   will throw errors to the interpreter.
+ */
+
+#include <stdlib.h>
+#include <string.h>
+#include <stdio.h>
+#include <inttypes.h>
+#include <limits.h>
+#include <unistd.h>
+
+#include <misc/offset.h>
+#include <structure/LinkedList0.h>
+#include <system/BProcess.h>
+#include <system/BConnection.h>
+#include <ncd/NCDModule.h>
+#include <ncd/extra/NCDBuf.h>
+#include <ncd/extra/value_utils.h>
+#include <ncd/extra/build_cmdline.h>
+
+#include <generated/blog_channel_ncd_sys_start_process.h>
+
+#define ModuleLog(i, ...) NCDModuleInst_Backend_Log((i), BLOG_CURRENT_CHANNEL, __VA_ARGS__)
+
+#define READ_BUF_SIZE 8192
+
+#define STATE_ERROR 1
+#define STATE_RUNNING 2
+#define STATE_TERMINATED 3
+#define STATE_DYING 4
+
+#define READER_STATE_RUNNING 1
+#define READER_STATE_EOF 2
+#define READER_STATE_ERROR 3
+#define READER_STATE_ABORTED 4
+
+#define WRITER_STATE_RUNNING 1
+#define WRITER_STATE_CLOSED 2
+#define WRITER_STATE_ERROR 3
+#define WRITER_STATE_ABORTED 4
+
+struct process_instance {
+    NCDModuleInst *i;
+    BProcess process;
+    LinkedList0 waits_list;
+    int read_fd;
+    int write_fd;
+    int exit_status;
+    int state;
+};
+
+struct wait_instance {
+    NCDModuleInst *i;
+    struct process_instance *pinst;
+    LinkedList0Node waits_list_node;
+    int exit_status;
+};
+
+struct read_pipe_instance {
+    NCDModuleInst *i;
+    int state;
+    int read_fd;
+    BConnection connection;
+    NCDBufStore store;
+    struct read_instance *read_inst;
+};
+
+struct read_instance {
+    NCDModuleInst *i;
+    struct read_pipe_instance *read_pipe_inst;
+    NCDBuf *buf;
+    size_t read_size;
+};
+
+struct write_pipe_instance {
+    NCDModuleInst *i;
+    int state;
+    int write_fd;
+    BConnection connection;
+    struct write_instance *write_inst;
+};
+
+struct write_instance {
+    NCDModuleInst *i;
+    struct write_pipe_instance *write_pipe_inst;
+    const char *data;
+    size_t length;
+};
+
+enum {STRING_IS_ERROR, STRING_EXIT_STATUS, STRING_NOT_EOF};
+
+static struct NCD_string_request strings[] = {
+    {"is_error"}, {"exit_status"}, {"not_eof"}, {NULL}
+};
+
+static int parse_mode (NCDModuleInst *i, NCDValRef mode_arg, int *out_read, int *out_write)
+{
+    if (!NCDVal_IsString(mode_arg)) {
+        ModuleLog(i, BLOG_ERROR, "mode argument must be a string");
+        return 0;
+    }
+    
+    const char *data = NCDVal_StringData(mode_arg);
+    size_t length = NCDVal_StringLength(mode_arg);
+    
+    *out_read = 0;
+    *out_write = 0;
+    
+    while (length > 0) {
+        if (*data == 'r') {
+            *out_read = 1;
+        }
+        else if (*data == 'w') {
+            *out_write = 1;
+        }
+        else {
+            ModuleLog(i, BLOG_ERROR, "invalid character in mode argument");
+            return 0;
+        }
+        data++;
+        length--;
+    }
+    
+    return 1;
+}
+
+static void instance_free (struct process_instance *o)
+{
+    // close write fd
+    if (o->write_fd != -1) {
+        close(o->write_fd);
+    }
+    
+    // close read fd
+    if (o->read_fd != -1) {
+        close(o->read_fd);
+    }
+    
+    NCDModuleInst_Backend_Dead(o->i);
+}
+
+static void process_handler (void *vo, int normally, uint8_t normally_exit_status)
+{
+    struct process_instance *o = vo;
+    ASSERT(o->state == STATE_RUNNING || o->state == STATE_DYING)
+    
+    ModuleLog(o->i, BLOG_INFO, "process terminated");
+    
+    // free process
+    BProcess_Free(&o->process);
+    
+    // remember exit code
+    o->exit_status = (!normally ? -1 : normally_exit_status);
+    
+    // finish waits
+    LinkedList0Node *ln;
+    while ((ln = LinkedList0_GetFirst(&o->waits_list))) {
+        struct wait_instance *winst = UPPER_OBJECT(ln, struct wait_instance, waits_list_node);
+        ASSERT(winst->pinst == o)
+        LinkedList0_Remove(&o->waits_list, &winst->waits_list_node);
+        winst->pinst = NULL;
+        winst->exit_status = o->exit_status;
+        NCDModuleInst_Backend_Up(winst->i);
+    }
+    
+    // if we have been requested to die, then die now
+    if (o->state == STATE_DYING) {
+        instance_free(o);
+        return;
+    }
+    
+    // set state
+    o->state = STATE_TERMINATED;
+}
+
+static void process_func_new (void *vo, NCDModuleInst *i, const struct NCDModuleInst_new_params *params)
+{
+    struct process_instance *o = vo;
+    o->i = i;
+    NCDModuleInst_Backend_PassMemToMethods(i);
+    
+    // check arguments
+    NCDValRef command_arg;
+    NCDValRef mode_arg;
+    if (!NCDVal_ListRead(params->args, 2, &command_arg, &mode_arg)) {
+        ModuleLog(i, BLOG_ERROR, "wrong arity");
+        goto fail0;
+    }
+    
+    // parse mode
+    int is_read;
+    int is_write;
+    if (!parse_mode(i, mode_arg, &is_read, &is_write)) {
+        goto fail0;
+    }
+    
+    // prepare for creating pipes
+    int fds[3];
+    int fds_map[2];
+    int num_fds = 0;
+    int read_fd = -1;
+    int write_fd = -1;
+    
+    // create read pipe
+    if (is_read) {
+        int pipefd[2];
+        if (pipe(pipefd) < 0) {
+            ModuleLog(i, BLOG_ERROR, "pipe failed");
+            goto error1;
+        }
+        read_fd = pipefd[0];
+        fds[num_fds] = pipefd[1];
+        fds_map[num_fds++] = STDOUT_FILENO;
+    }
+    
+    // create write pipe
+    if (is_write) {
+        int pipefd[2];
+        if (pipe(pipefd) < 0) {
+            ModuleLog(i, BLOG_ERROR, "pipe failed");
+            goto error1;
+        }
+        write_fd = pipefd[1];
+        fds[num_fds] = pipefd[0];
+        fds_map[num_fds++] = STDIN_FILENO;
+    }
+    
+    // terminate fds array
+    fds[num_fds] = -1;
+    
+    // build process parameters struct
+    struct BProcess_params p_params = {};
+    p_params.fds = fds;
+    p_params.fds_map = fds_map;
+    
+    // build command line
+    char *exec;
+    CmdLine cl;
+    if (!ncd_build_cmdline(i, BLOG_CURRENT_CHANNEL, command_arg, &exec, &cl)) {
+        goto error1;
+    }
+    
+    // start process
+    int res = BProcess_Init2(&o->process, i->params->iparams->manager, process_handler, o, exec, CmdLine_Get(&cl), p_params);
+    CmdLine_Free(&cl);
+    free(exec);
+    if (!res) {
+        ModuleLog(i, BLOG_ERROR, "BProcess_Init failed");
+        goto error1;
+    }
+    
+    // close child fds
+    while (num_fds-- > 0) {
+        close(fds[num_fds]);
+    }
+    
+    // init waits list
+    LinkedList0_Init(&o->waits_list);
+    
+    // remember our fds
+    o->read_fd = read_fd;
+    o->write_fd = write_fd;
+    
+    // set state
+    o->state = STATE_RUNNING;
+    
+    // go up
+    NCDModuleInst_Backend_Up(i);
+    return;
+    
+fail0:
+    NCDModuleInst_Backend_DeadError(i);
+    return;
+    
+error1:
+    if (write_fd != -1) {
+        close(write_fd);
+    }
+    if (read_fd != -1) {
+        close(read_fd);
+    }
+    while (num_fds-- > 0) {
+        close(fds[num_fds]);
+    }
+    
+    o->read_fd = -1;
+    o->write_fd = -1;
+    o->state = STATE_ERROR;
+    NCDModuleInst_Backend_Up(i);
+}
+
+static void process_func_die (void *vo)
+{
+    struct process_instance *o = vo;
+    ASSERT(o->state != STATE_DYING)
+    
+    // if process is not running, die immediately
+    if (o->state != STATE_RUNNING) {
+        instance_free(o);
+        return;
+    }
+    
+    ModuleLog(o->i, BLOG_INFO, "terminating process");
+    
+    // send termination signal
+    BProcess_Terminate(&o->process);
+    
+    // set state
+    o->state = STATE_DYING;
+}
+
+static int process_func_getvar (void *vo, NCD_string_id_t name, NCDValMem *mem, NCDValRef *out)
+{
+    struct process_instance *o = vo;
+    
+    if (name == strings[STRING_IS_ERROR].id) {
+        int is_error = (o->state == STATE_ERROR);
+        *out = ncd_make_boolean(mem, is_error, o->i->params->iparams->string_index);
+        return 1;
+    }
+    
+    return 0;
+}
+
+static void wait_func_new (void *vo, NCDModuleInst *i, const struct NCDModuleInst_new_params *params)
+{
+    struct wait_instance *o = vo;
+    o->i = i;
+    
+    if (!NCDVal_ListRead(params->args, 0)) {
+        ModuleLog(i, BLOG_ERROR, "wrong arity");
+        goto fail0;
+    }
+    
+    struct process_instance *pinst = params->method_user;
+    
+    if (pinst->state == STATE_ERROR) {
+        ModuleLog(i, BLOG_ERROR, "process did not start successfully");
+        goto fail0;
+    }
+    
+    if (pinst->state == STATE_TERMINATED) {
+        // not waiting, set no pinst
+        o->pinst = NULL;
+        
+        // remember exit code
+        o->exit_status = pinst->exit_status;
+        
+        // go up
+        NCDModuleInst_Backend_Up(i);
+    } else {
+        // waitint, set pinst
+        o->pinst = pinst;
+        
+        // insert to waits list
+        LinkedList0_Prepend(&pinst->waits_list, &o->waits_list_node);
+    }
+    
+    return;
+    
+fail0:
+    NCDModuleInst_Backend_DeadError(i);
+}
+
+static void wait_func_die (void *vo)
+{
+    struct wait_instance *o = vo;
+    
+    // remove from waits list
+    if (o->pinst) {
+        LinkedList0_Remove(&o->pinst->waits_list, &o->waits_list_node);
+    }
+    
+    NCDModuleInst_Backend_Dead(o->i);
+}
+
+static int wait_func_getvar (void *vo, NCD_string_id_t name, NCDValMem *mem, NCDValRef *out)
+{
+    struct wait_instance *o = vo;
+    ASSERT(!o->pinst)
+    
+    if (name == strings[STRING_EXIT_STATUS].id) {
+        if (o->exit_status == -1) {
+            *out = NCDVal_NewString(mem, "-1");
+        } else {
+            *out = ncd_make_uintmax(mem, o->exit_status);
+        }
+        return 1;
+    }
+    
+    return 0;
+}
+
+static void terminate_kill_new_common (void *vo, NCDModuleInst *i, const struct NCDModuleInst_new_params *params, int is_kill)
+{
+    if (!NCDVal_ListRead(params->args, 0)) {
+        ModuleLog(i, BLOG_ERROR, "wrong arity");
+        goto fail0;
+    }
+    
+    struct process_instance *pinst = params->method_user;
+    
+    if (pinst->state == STATE_ERROR) {
+        ModuleLog(i, BLOG_ERROR, "process did not start successfully");
+        goto fail0;
+    }
+    
+    if (pinst->state != STATE_TERMINATED) {
+        if (is_kill) {
+            BProcess_Kill(&pinst->process);
+        } else {
+            BProcess_Terminate(&pinst->process);
+        }
+    }
+    
+    NCDModuleInst_Backend_Up(i);
+    return;
+    
+fail0:
+    NCDModuleInst_Backend_DeadError(i);
+}
+
+static void terminate_func_new (void *vo, NCDModuleInst *i, const struct NCDModuleInst_new_params *params)
+{
+    terminate_kill_new_common(vo, i, params, 0);
+}
+
+static void kill_func_new (void *vo, NCDModuleInst *i, const struct NCDModuleInst_new_params *params)
+{
+    terminate_kill_new_common(vo, i, params, 1);
+}
+
+static void read_pipe_free_connection (struct read_pipe_instance *o)
+{
+    // disconnect read instance
+    if (o->read_inst) {
+        ASSERT(o->read_inst->read_pipe_inst == o)
+        o->read_inst->read_pipe_inst = NULL;
+    }
+    
+    // free store
+    NCDBufStore_Free(&o->store);
+    
+    // free connection read interface
+    BConnection_RecvAsync_Free(&o->connection);
+    
+    // free connection
+    BConnection_Free(&o->connection);
+    
+    // close fd
+    if (close(o->read_fd) < 0) {
+        ModuleLog(o->i, BLOG_ERROR, "close failed");
+    }
+}
+
+static void read_pipe_abort (struct read_pipe_instance *o)
+{
+    ASSERT(o->state == READER_STATE_RUNNING)
+    
+    // release connection resources
+    read_pipe_free_connection(o);
+    
+    // set state
+    o->state = READER_STATE_ABORTED;
+}
+
+static void read_pipe_connection_handler (void *vo, int event)
+{
+    struct read_pipe_instance *o = vo;
+    ASSERT(o->state == READER_STATE_RUNNING)
+    
+    if (event == BCONNECTION_EVENT_RECVCLOSED) {
+        // if we have read operation, make it finish with eof
+        if (o->read_inst) {
+            ASSERT(o->read_inst->read_pipe_inst == o)
+            o->read_inst->read_pipe_inst = NULL;
+            o->read_inst->read_size = 0;
+            NCDModuleInst_Backend_Up(o->read_inst->i);
+            o->read_inst = NULL;
+        }
+        
+        // free connection resources
+        read_pipe_free_connection(o);
+        
+        // set state closed
+        o->state = READER_STATE_EOF;
+        return;
+    }
+    
+    ModuleLog(o->i, BLOG_ERROR, "read pipe error");
+    
+    // free connection resources
+    read_pipe_free_connection(o);
+    
+    // set state error
+    o->state = READER_STATE_ERROR;
+    
+    // backtrack
+    NCDModuleInst_Backend_DownUp(o->i);
+}
+
+static void read_pipe_recv_handler_done (void *vo, int data_len)
+{
+    struct read_pipe_instance *o = vo;
+    ASSERT(o->state == READER_STATE_RUNNING)
+    ASSERT(o->read_inst)
+    ASSERT(o->read_inst->read_pipe_inst == o)
+    ASSERT(o->read_inst->buf)
+    ASSERT(data_len > 0)
+    ASSERT(data_len <= NCDBufStore_BufSize(&o->store))
+    
+    // finish read operation
+    o->read_inst->read_pipe_inst = NULL;
+    o->read_inst->read_size = data_len;
+    NCDModuleInst_Backend_Up(o->read_inst->i);
+    o->read_inst = NULL;
+}
+
+static void read_pipe_func_new (void *vo, NCDModuleInst *i, const struct NCDModuleInst_new_params *params)
+{
+    struct read_pipe_instance *o = vo;
+    o->i = i;
+    NCDModuleInst_Backend_PassMemToMethods(i);
+    
+    if (!NCDVal_ListRead(params->args, 0)) {
+        ModuleLog(i, BLOG_ERROR, "wrong arity");
+        goto fail0;
+    }
+    
+    struct process_instance *pinst = params->method_user;
+    
+    if (pinst->read_fd == -1) {
+        ModuleLog(i, BLOG_ERROR, "process did not start successfully, was not opened for reading or a read_pipe was already created");
+        goto fail0;
+    }
+    
+    // init connection
+    if (!BConnection_Init(&o->connection, BConnection_source_pipe(pinst->read_fd), i->params->iparams->reactor, o, read_pipe_connection_handler)) {
+        ModuleLog(i, BLOG_ERROR, "BConnection_Init failed");
+        goto fail0;
+    }
+    
+    // init connection read interface
+    BConnection_RecvAsync_Init(&o->connection);
+    
+    // set recv done callback
+    StreamRecvInterface_Receiver_Init(BConnection_RecvAsync_GetIf(&o->connection), read_pipe_recv_handler_done, o);
+    
+    // init store
+    NCDBufStore_Init(&o->store, READ_BUF_SIZE);
+    
+    // set variables
+    o->state = READER_STATE_RUNNING;
+    o->read_fd = pinst->read_fd;
+    o->read_inst = NULL;
+    
+    // steal read fd from process instance
+    pinst->read_fd = -1;
+    
+    // go up
+    NCDModuleInst_Backend_Up(i);
+    return;
+    
+fail0:
+    NCDModuleInst_Backend_DeadError(i);
+}
+
+static void read_pipe_func_die (void *vo)
+{
+    struct read_pipe_instance *o = vo;
+    
+    // free connection resources
+    if (o->state == READER_STATE_RUNNING) {
+        read_pipe_free_connection(o);
+    }
+    
+    NCDModuleInst_Backend_Dead(o->i);
+}
+
+static int read_pipe_func_getvar (void *vo, NCD_string_id_t name, NCDValMem *mem, NCDValRef *out)
+{
+    struct read_pipe_instance *o = vo;
+    
+    if (name == strings[STRING_IS_ERROR].id) {
+        int is_error = (o->state == READER_STATE_ERROR);
+        *out = ncd_make_boolean(mem, is_error, o->i->params->iparams->string_index);
+        return 1;
+    }
+    
+    return 0;
+}
+
+static void read_func_new (void *vo, NCDModuleInst *i, const struct NCDModuleInst_new_params *params)
+{
+    struct read_instance *o = vo;
+    o->i = i;
+    
+    if (!NCDVal_ListRead(params->args, 0)) {
+        ModuleLog(i, BLOG_ERROR, "wrong arity");
+        goto fail0;
+    }
+    
+    struct read_pipe_instance *read_pipe_inst = params->method_user;
+    
+    // check if a read error has already occured
+    if (read_pipe_inst->state == READER_STATE_ERROR) {
+        ModuleLog(i, BLOG_ERROR, "read error already occured");
+        goto fail0;
+    }
+    
+    // check if the read_pipe has been aborted
+    if (read_pipe_inst->state == READER_STATE_ABORTED) {
+        ModuleLog(i, BLOG_ERROR, "further reads are disallowed after a read has been aborted");
+        goto fail0;
+    }
+    
+    if (read_pipe_inst->state == READER_STATE_EOF) {
+        o->buf = NULL;
+        o->read_pipe_inst = NULL;
+        o->read_size = 0;
+        NCDModuleInst_Backend_Up(i);
+        return;
+    }
+    
+    // check if there's already a read in progress
+    if (read_pipe_inst->read_inst) {
+        ModuleLog(i, BLOG_ERROR, "read is already in progress");
+        goto fail0;
+    }
+    
+    // get buffer
+    o->buf = NCDBufStore_GetBuf(&read_pipe_inst->store);
+    if (!o->buf) {
+        ModuleLog(i, BLOG_ERROR, "NCDBufStore_GetBuf failed");
+        goto fail0;
+    }
+    
+    // set read_pipe
+    o->read_pipe_inst = read_pipe_inst;
+    
+    // register read in read_pipe
+    read_pipe_inst->read_inst = o;
+    
+    // receive
+    size_t buf_size = NCDBufStore_BufSize(&read_pipe_inst->store);
+    int to_read = (buf_size > INT_MAX ? INT_MAX : buf_size);
+    StreamRecvInterface_Receiver_Recv(BConnection_RecvAsync_GetIf(&read_pipe_inst->connection), (uint8_t *)NCDBuf_Data(o->buf), to_read);
+    return;
+    
+fail0:
+    NCDModuleInst_Backend_DeadError(i);
+}
+
+static void read_func_die (void *vo)
+{
+    struct read_instance *o = vo;
+    
+    // if we're receiving, abort read_pipe
+    if (o->read_pipe_inst) {
+        ASSERT(o->read_pipe_inst->state == READER_STATE_RUNNING)
+        ASSERT(o->read_pipe_inst->read_inst == o)
+        read_pipe_abort(o->read_pipe_inst);
+    }
+    
+    // release buffer
+    if (o->buf) {
+        NCDRefTarget_Deref(NCDBuf_RefTarget(o->buf));
+    }
+    
+    NCDModuleInst_Backend_Dead(o->i);
+}
+
+static int read_func_getvar (void *vo, NCD_string_id_t name, NCDValMem *mem, NCDValRef *out)
+{
+    struct read_instance *o = vo;
+    ASSERT(!o->read_pipe_inst)
+    
+    if (name == NCD_STRING_EMPTY) {
+        if (o->buf) {
+            *out = NCDVal_NewExternalString(mem, NCDBuf_Data(o->buf), o->read_size, NCDBuf_RefTarget(o->buf));
+        } else {
+            *out = NCDVal_NewIdString(mem, NCD_STRING_EMPTY, o->i->params->iparams->string_index);
+        }
+        return 1;
+    }
+    
+    if (name == strings[STRING_NOT_EOF].id) {
+        *out = ncd_make_boolean(mem, (o->read_size != 0), o->i->params->iparams->string_index);
+        return 1;
+    }
+    
+    return 0;
+}
+
+static void write_pipe_free_connection (struct write_pipe_instance *o)
+{
+    // disconnect write instance
+    if (o->write_inst) {
+        ASSERT(o->write_inst->write_pipe_inst == o)
+        o->write_inst->write_pipe_inst = NULL;
+    }
+    
+    // free connection send interface
+    BConnection_SendAsync_Free(&o->connection);
+    
+    // free connection
+    BConnection_Free(&o->connection);
+    
+    // close fd
+    if (close(o->write_fd) < 0) {
+        ModuleLog(o->i, BLOG_ERROR, "close failed");
+    }
+}
+
+static void write_pipe_abort (struct write_pipe_instance *o)
+{
+    ASSERT(o->state == WRITER_STATE_RUNNING)
+    
+    // release connection resources
+    write_pipe_free_connection(o);
+    
+    // set state
+    o->state = WRITER_STATE_ABORTED;
+}
+
+static void write_pipe_close (struct write_pipe_instance *o)
+{
+    ASSERT(o->state == WRITER_STATE_RUNNING)
+    
+    // release connection resources
+    write_pipe_free_connection(o);
+    
+    // set state
+    o->state = WRITER_STATE_CLOSED;
+}
+
+static void write_pipe_connection_handler (void *vo, int event)
+{
+    struct write_pipe_instance *o = vo;
+    ASSERT(o->state == WRITER_STATE_RUNNING)
+    
+    ModuleLog(o->i, BLOG_ERROR, "write pipe error");
+    
+    // free connection resources
+    write_pipe_free_connection(o);
+    
+    // set state error
+    o->state = WRITER_STATE_ERROR;
+    
+    // backtrack
+    NCDModuleInst_Backend_DownUp(o->i);
+}
+
+static void write_pipe_send_handler_done (void *vo, int data_len)
+{
+    struct write_pipe_instance *o = vo;
+    ASSERT(o->state == WRITER_STATE_RUNNING)
+    ASSERT(o->write_inst)
+    ASSERT(o->write_inst->write_pipe_inst == o)
+    ASSERT(data_len > 0)
+    ASSERT(data_len <= o->write_inst->length)
+    
+    // update write progress
+    o->write_inst->data += data_len;
+    o->write_inst->length -= data_len;
+    
+    // if there is more data, start another write operation
+    if (o->write_inst->length > 0) {
+        size_t to_send = (o->write_inst->length > INT_MAX ? INT_MAX : o->write_inst->length);
+        StreamPassInterface_Sender_Send(BConnection_SendAsync_GetIf(&o->connection), (uint8_t *)o->write_inst->data, to_send);
+        return;
+    }
+    
+    // finish write operation
+    o->write_inst->write_pipe_inst = NULL;
+    NCDModuleInst_Backend_Up(o->write_inst->i);
+    o->write_inst = NULL;
+}
+
+static void write_pipe_func_new (void *vo, NCDModuleInst *i, const struct NCDModuleInst_new_params *params)
+{
+    struct write_pipe_instance *o = vo;
+    o->i = i;
+    NCDModuleInst_Backend_PassMemToMethods(i);
+    
+    if (!NCDVal_ListRead(params->args, 0)) {
+        ModuleLog(i, BLOG_ERROR, "wrong arity");
+        goto fail0;
+    }
+    
+    struct process_instance *pinst = params->method_user;
+    
+    if (pinst->write_fd == -1) {
+        ModuleLog(i, BLOG_ERROR, "process did not start successfully, was not opened for writing or a write_pipe was already created");
+        goto fail0;
+    }
+    
+    // init connection
+    if (!BConnection_Init(&o->connection, BConnection_source_pipe(pinst->write_fd), i->params->iparams->reactor, o, write_pipe_connection_handler)) {
+        ModuleLog(i, BLOG_ERROR, "BConnection_Init failed");
+        goto fail0;
+    }
+    
+    // init connection send interface
+    BConnection_SendAsync_Init(&o->connection);
+    
+    // set send done callback
+    StreamPassInterface_Sender_Init(BConnection_SendAsync_GetIf(&o->connection), write_pipe_send_handler_done, o);
+    
+    // set variables
+    o->state = WRITER_STATE_RUNNING;
+    o->write_fd = pinst->write_fd;
+    o->write_inst = NULL;
+    
+    // steal write fd from process instance
+    pinst->write_fd = -1;
+    
+    // go up
+    NCDModuleInst_Backend_Up(i);
+    return;
+    
+fail0:
+    NCDModuleInst_Backend_DeadError(i);
+}
+
+static void write_pipe_func_die (void *vo)
+{
+    struct write_pipe_instance *o = vo;
+    
+    // free connection resources
+    if (o->state == WRITER_STATE_RUNNING) {
+        write_pipe_free_connection(o);
+    }
+    
+    NCDModuleInst_Backend_Dead(o->i);
+}
+
+static int write_pipe_func_getvar (void *vo, NCD_string_id_t name, NCDValMem *mem, NCDValRef *out)
+{
+    struct write_pipe_instance *o = vo;
+    
+    if (name == strings[STRING_IS_ERROR].id) {
+        int is_error = (o->state == WRITER_STATE_ERROR);
+        *out = ncd_make_boolean(mem, is_error, o->i->params->iparams->string_index);
+        return 1;
+    }
+    
+    return 0;
+}
+
+static void write_func_new (void *vo, NCDModuleInst *i, const struct NCDModuleInst_new_params *params)
+{
+    struct write_instance *o = vo;
+    o->i = i;
+    
+    NCDValRef data_arg;
+    if (!NCDVal_ListRead(params->args, 1, &data_arg)) {
+        ModuleLog(i, BLOG_ERROR, "wrong arity");
+        goto fail0;
+    }
+    if (!NCDVal_IsString(data_arg)) {
+        ModuleLog(i, BLOG_ERROR, "wrong type");
+        goto fail0;
+    }
+    
+    struct write_pipe_instance *write_pipe_inst = params->method_user;
+    
+    // check if a write error has already occured
+    if (write_pipe_inst->state == WRITER_STATE_ERROR) {
+        ModuleLog(i, BLOG_ERROR, "write() is disallowed after a write error has occured");
+        goto fail0;
+    }
+    
+    // check if the write_pipe has been aborted
+    if (write_pipe_inst->state == WRITER_STATE_ABORTED) {
+        ModuleLog(i, BLOG_ERROR, "write() is disallowed after a write() has been aborted");
+        goto fail0;
+    }
+    
+    // check if the write_pipe has been aborted
+    if (write_pipe_inst->state == WRITER_STATE_CLOSED) {
+        ModuleLog(i, BLOG_ERROR, "write() is disallowed after close() has been called");
+        goto fail0;
+    }
+    
+    // check if there's already a write in progress
+    if (write_pipe_inst->write_inst) {
+        ModuleLog(i, BLOG_ERROR, "write is already in progress");
+        goto fail0;
+    }
+    
+    // initialize write progress state
+    o->data = NCDVal_StringData(data_arg);
+    o->length = NCDVal_StringLength(data_arg);
+    
+    // if there's nothing to send, go up immediately
+    if (o->length == 0) {
+        o->write_pipe_inst = NULL;
+        NCDModuleInst_Backend_Up(i);
+        return;
+    }
+    
+    // set write_pipe
+    o->write_pipe_inst = write_pipe_inst;
+    
+    // register write in write_pipe
+    write_pipe_inst->write_inst = o;
+    
+    // start send operation
+    size_t to_send = (o->length > INT_MAX ? INT_MAX : o->length);
+    StreamPassInterface_Sender_Send(BConnection_SendAsync_GetIf(&write_pipe_inst->connection), (uint8_t *)o->data, to_send);
+    return;
+    
+fail0:
+    NCDModuleInst_Backend_DeadError(i);
+}
+
+static void write_func_die (void *vo)
+{
+    struct write_instance *o = vo;
+    
+    // if we're sending, abort write_pipe
+    if (o->write_pipe_inst) {
+        ASSERT(o->write_pipe_inst->state == WRITER_STATE_RUNNING)
+        ASSERT(o->write_pipe_inst->write_inst == o)
+        write_pipe_abort(o->write_pipe_inst);
+    }
+    
+    NCDModuleInst_Backend_Dead(o->i);
+}
+
+static void close_func_new (void *vo, NCDModuleInst *i, const struct NCDModuleInst_new_params *params)
+{
+    if (!NCDVal_ListRead(params->args, 0)) {
+        ModuleLog(i, BLOG_ERROR, "wrong arity");
+        goto fail0;
+    }
+    
+    struct write_pipe_instance *write_pipe_inst = params->method_user;
+    
+    // check if a write error has already occured
+    if (write_pipe_inst->state == WRITER_STATE_ERROR) {
+        ModuleLog(i, BLOG_ERROR, "close() is disallowed after a write error has occured");
+        goto fail0;
+    }
+    
+    // check if the write_pipe has been aborted
+    if (write_pipe_inst->state == WRITER_STATE_ABORTED) {
+        ModuleLog(i, BLOG_ERROR, "close() is disallowed after a write() has been aborted");
+        goto fail0;
+    }
+    
+    // check if the write_pipe has been closed
+    if (write_pipe_inst->state == WRITER_STATE_CLOSED) {
+        ModuleLog(i, BLOG_ERROR, "close() is disallowed after close() has been called");
+        goto fail0;
+    }
+    
+    // close
+    write_pipe_close(write_pipe_inst);
+    
+    // go up
+    NCDModuleInst_Backend_Up(i);
+    return;
+    
+fail0:
+    NCDModuleInst_Backend_DeadError(i);
+}
+
+static struct NCDModule modules[] = {
+    {
+        .type = "sys.start_process",
+        .func_new2 = process_func_new,
+        .func_die = process_func_die,
+        .func_getvar2 = process_func_getvar,
+        .alloc_size = sizeof(struct process_instance)
+    }, {
+        .type = "sys.start_process::wait",
+        .func_new2 = wait_func_new,
+        .func_die = wait_func_die,
+        .func_getvar2 = wait_func_getvar,
+        .alloc_size = sizeof(struct wait_instance)
+    }, {
+        .type = "sys.start_process::terminate",
+        .func_new2 = terminate_func_new,
+    }, {
+        .type = "sys.start_process::kill",
+        .func_new2 = kill_func_new,
+    }, {
+        .type = "sys.start_process::read_pipe",
+        .func_new2 = read_pipe_func_new,
+        .func_die = read_pipe_func_die,
+        .func_getvar2 = read_pipe_func_getvar,
+        .alloc_size = sizeof(struct read_pipe_instance)
+    }, {
+        .type = "sys.start_process::read_pipe::read",
+        .func_new2 = read_func_new,
+        .func_die = read_func_die,
+        .func_getvar2 = read_func_getvar,
+        .alloc_size = sizeof(struct read_instance)
+    }, {
+        .type = "sys.start_process::write_pipe",
+        .func_new2 = write_pipe_func_new,
+        .func_die = write_pipe_func_die,
+        .func_getvar2 = write_pipe_func_getvar,
+        .alloc_size = sizeof(struct write_pipe_instance)
+    }, {
+        .type = "sys.start_process::write_pipe::write",
+        .func_new2 = write_func_new,
+        .func_die = write_func_die,
+        .alloc_size = sizeof(struct write_instance)
+    }, {
+        .type = "sys.start_process::write_pipe::close",
+        .func_new2 = close_func_new
+    }, {
+        .type = NULL
+    }
+};
+
+const struct NCDModuleGroup ncdmodule_sys_start_process = {
+    .modules = modules,
+    .strings = strings
+};