Asyncronous Bidirectional IO Redirection for a child process

Question

I am trying to figure out a generalized way for Asynchronous Bidirectional IO Redirection of a child process. Basically, I would like to spawn an interactive child process that waits for input and any output should be read back. I tried to experiment with python.subprocess by spawning a new python process. A base simplistic example tried to achieve is as follows

process = subprocess.Popen(['/usr/bin/python'],shell=False,stdin=subprocess.PIPE, stdout=subprocess.PIPE)
while True:
    output = process.stdout.readline()
    print output
    input = sys.stdin.readline()
    process.stdin.write(input)

and executing the above code snippet simply hangs without any output. I tried running with /usr/bash and /usr/bin/irb but the result is all the same. My guess is, buffered IO is simply not gelling well with IO redirection.

So my question is, is it feasible to read the output of a child process without flushing the buffer or quitting the subprocess?

The following post mentions IPC sockets but for that I would have to change the child process which may not be feasible. Is there any other way to achieve it?

Note*** My ultimate goal is to create a server REPL process which can interact with a remote web client. Though the example given is of Python, my ultimate goal is to wrap all available REPL by a generalized wrapper.

---

With the help of some of the suggestion in the answers I came up with the following

#!/usr/bin/python
import subprocess, os, select
proc = subprocess.Popen(['/usr/bin/python'],shell=False,stdin=subprocess.PIPE, stdout=subprocess.PIPE,stderr=subprocess.PIPE)
for i in xrange(0,5):
    inputready, outputready, exceptready = select.select([proc.stdout, proc.stderr],[proc.stdout, proc.stderr],[proc.stdout, proc.stderr],0)
    if not inputready: print "No Data",
    print inputready, outputready, exceptready
    for s in inputready: print s.fileno(),s.readline()
proc.terminate()
print "After Terminating"
for i in xrange(0,5):
    inputready, outputready, exceptready = select.select([proc.stdout, proc.stderr],[proc.stdout, proc.stderr],[proc.stdout, proc.stderr],0)
    if not inputready: print "No Data",
    print inputready, outputready, exceptready
    for s in inputready: print s.fileno(),s.readline() 

now, though the programs is not in deadlock but unfortunately there is no output. Running the above code I get

No Data [] [] []
No Data [] [] []
No Data [] [] []
No Data [] [] []
No Data [] [] []
After Terminating
No Data [] [] []
No Data [] [] []
No Data [] [] []
No Data [] [] []
No Data [] [] []

Just FYI, running python as

/usr/bin/python 2>&1|tee test.out

seems to be working just fine.

I also came up with a 'C' code. But the result is not different.

int kbhit() {
    struct timeval tv;
    fd_set fds;
    tv.tv_sec = tv.tv_usec = 0;
    FD_ZERO(&fds);
    FD_SET(STDIN_FILENO, &fds);
    select(STDIN_FILENO+1, &fds, NULL, NULL, &tv);
    return FD_ISSET(STDIN_FILENO, &fds);
}
void receive(char *str) {
    char ch;
    fprintf(stderr,"IN1\n");
    if(!kbhit()) return;
    fprintf(stderr,"IN2\n");
    fprintf(stderr,"%d\n",kbhit());
    for(;kbhit() && (ch=fgetc(stdin))!=EOF;) {
        fprintf(stderr,"%c,%d",ch,kbhit());
    }
    fprintf(stderr,"Done\n");
}
int main(){
    pid_t pid;
    int rv, pipeP2C[2],pipeC2P[2];  
    pipe(pipeP2C);
    pipe(pipeC2P);
    pid=fork();
    if(pid){
        dup2(pipeP2C[1],1); /* Replace stdout with out side of the pipe */
        close(pipeP2C[0]);  /* Close unused side of pipe (in side) */
        dup2(pipeC2P[0],0); /* Replace stdin with in side of the pipe */
        close(pipeC2P[1]);  /* Close unused side of pipe (out side) */
        setvbuf(stdout,(char*)NULL,_IONBF,0);   /* Set non-buffered output on stdout */
        sleep(2);
        receive("quit()\n");
        wait(&rv);              /* Wait for child process to end */
        fprintf(stderr,"Child exited with a %d value\n",rv);
    }
    else{
        dup2(pipeP2C[0],0); /* Replace stdin with the in side of the pipe */
        close(pipeP2C[1]);  /* Close unused side of pipe (out side) */
        dup2(pipeC2P[1],1); /* Replace stdout with the out side of the pipe */
        close(pipeC2P[0]);  /* Close unused side of pipe (out side) */
        setvbuf(stdout,(char*)NULL,_IONBF,0);   /* Set non-buffered output on stdout */
        close(2), dup2(1,2); /*Redirect stderr to stdout */
        if(execl("/usr/bin/python","/usr/bin/python",NULL) == -1){
            fprintf(stderr,"execl Error!");
            exit(1);
        }
    }
    return 0;
}

Answer 1

There are different way to do this. You can, for example:

• use SysV message queues and poll with timeout on the queue for message to arrive
• create a pipe() for the child and a pipe() for the father both using the O_NONBLOCK flag and then select() on the file descriptors for data to arrive (to can even handle timeouts if no data arrives)
• use socket() AF_UNIX or AF_INET, set it non blocking and select() or epoll() for data to arrive
• mmap() MAP_SHARED memory segments and signal the other process when data is arrived, pay attention to the shared segment with a locking mechanism.

I wrote a sample in C with double pipes:

#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <unistd.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/stat.h>
#include <sys/select.h>
#include <fcntl.h>
#include <signal.h>

#define BUFLEN (6*1024)
#define EXECFILE "/usr/bin/python"

char *itoa(int n, char *s, int b) {
        static char digits[] = "0123456789abcdefghijklmnopqrstuvwxyz";
        int i=0, sign;

        if ((sign = n) < 0)
                n = -n;

        do {
                s[i++] = digits[n % b];
        } while ((n /= b) > 0);

        if (sign < 0)
                s[i++] = '-';
        s[i] = '\0';

        return s;
}

/*
int set_nonblock(int sockfd) { // set socket to non blocking
        int arg,i;

        if ((arg=fcntl(sockfd, F_GETFL, NULL)) < 0) {
                printf("error getting socket flag for fd %i: fcntl(..., F_GETFL): %i\n", sockfd, errno);
                return -1;
        }
        // set O_NONBLOCK flag
        arg |= O_NONBLOCK;
        if ((i=fcntl(sockfd, F_SETFL, arg)) < 0) {
                printf("error setting socket flag for fd %i: fcntl(..., F_SETFL): %i\n", sockfd, errno);
                return -1;
        }
        return i;
}

int set_block(int sockfd) { // set socket to blocking
        int arg,i;

        if ((arg=fcntl(sockfd, F_GETFL, NULL)) < 0) {
                printf("error getting socket flag for fd %i: fcntl(..., F_GETFL): %i\n", sockfd, errno);
                return -1;
        }
        // clean O_NONBLOCK flag
        arg &= (~O_NONBLOCK);
        if ((i=fcntl(sockfd, F_SETFL, arg)) < 0) {
                printf("error setting socket flag for fd %i: fcntl(..., F_SETFL): %i\n", sockfd, errno);
                return -1;
        }
        return i;
}
*/
int main() {
        FILE *input;
        char slice[BUFLEN];
        int status = 0;
        pid_t pid;
        int err;
        int newfd;
        // if you want you can pass arguments to the program to execute
        // char *const arguments[] = {EXECFILE, "-v", NULL};
        char *const arguments[] = {EXECFILE,  NULL};
        int father2child_pipefd[2];
        int child2father_pipefd[2];
        char *read_data = NULL;
        FILE *retclam;
        fd_set myset;
        int x=1;

        signal(SIGPIPE, SIG_IGN);
        newfd = dup(0);
        input = fdopen(newfd, "r");

        pipe(father2child_pipefd); // Father speaking to child
        pipe(child2father_pipefd); // Child speaking to father

        pid = fork();
        if (pid > 0) { // Father
                close(father2child_pipefd[0]);
                close(child2father_pipefd[1]);

                // Write to the pipe reading from stdin
                retclam = fdopen(child2father_pipefd[0], "r");


                // set the two fd non blocking
                //set_nonblock(0);
                //set_nonblock(child2father_pipefd[0]);
                //set_nonblock(fileno(retclam));

                while(x==1) {
                        // clear the file descriptor set
                        FD_ZERO(&myset);
                        // add the stdin to the set
                        FD_SET(fileno(input), &myset);
                        // add the child pipe to the set
                        FD_SET(fileno(retclam), &myset);

                        // here we wait for data to arrive from stdin or from the child pipe. The last argument is a timeout, if you like
                        err = select(fileno(retclam)+1, &myset, NULL, NULL, NULL);
                        switch(err) {
                        case -1:
                                // Problem with select(). The errno variable knows why
                                //exit(1);
                                x=0;
                                break;
                        case 0:
                                // timeout on select(). Data did not arrived in time, only valid if the last attribute of select() was specified
                                break;
                        default:
                                // data is ready to be read
                                bzero(slice, BUFLEN);
                                if (FD_ISSET(fileno(retclam), &myset)) { // data ready on the child
                                        //set_block(fileno(retclam));
                                        read_data = fgets(slice, BUFLEN, retclam); // read a line from the child (max BUFLEN bytes)
                                        //set_nonblock(fileno(retclam));
                                        if (read_data == NULL) {
                                                //exit(0);
                                                x=0;
                                                break;
                                        }
                                        // write data back to stdout
                                        write (1, slice, strlen(slice));
                                        if(feof(retclam)) {
                                                //exit(0);
                                                x=0;
                                                break;
                                        }
                                        break;
                                }
                                bzero(slice, BUFLEN);
                                if (FD_ISSET(fileno(input), &myset)) { // data ready on stdin
                                        //printf("father\n");
                                        //set_block(fileno(input));
                                        read_data = fgets(slice, BUFLEN, input); // read a line from stdin (max BUFLEN bytes)
                                        //set_nonblock(fileno(input));
                                        if (read_data == NULL) {
                                                //exit (0);
                                                close(father2child_pipefd[1]);
                                                waitpid(pid, &status, 0);
                                                //fclose(input);
                                                break;
                                        }
                                        // write data to the child
                                        write (father2child_pipefd[1], slice, strlen(slice));
                                        /*
                                        if(feof(input)) {
                                                exit(0);
                                        }*/
                                        break;
                                }
                        }
                }

                close(father2child_pipefd[1]);
                fclose(input);
                fsync(1);
                waitpid(pid, &status, 0);

                // child process terminated
                fclose (retclam);

                // Parse output data from child
                // write (1, "you can append somethind else on stdout if you like");
                if (WEXITSTATUS(status) == 0) {
                        exit (0); // child process exited successfully
                }
        }

        if (pid == 0) { // Child
                close (0); // stdin is not needed
                close (1); // stdout is not needed
                // Close the write side of this pipe
                close(father2child_pipefd[1]);
                // Close the read side of this pipe
                close(child2father_pipefd[0]);

                // Let's read on stdin, but this stdin is associated to the read pipe
                dup2(father2child_pipefd[0], 0);
                // Let's speak on stdout, but this stdout is associated to the write pipe
                dup2(child2father_pipefd[1], 1);

                // if you like you can put something back to the father before execve
                //write (child2father_pipefd[1], "something", 9);
                //fsync(child2father_pipefd[1]);
                err = execve(EXECFILE, arguments, NULL);

                // we'll never be here again after execve succeeded!! So we get here only if the execve() failed
                //fprintf(stderr, "Problem executing file %s: %i: %s\n", EXECFILE, err, strerror(errno));
                exit (1);
        }

        if (pid < 0) { // Error
                exit (1);
        }

        fclose(input);

        return 0;
}

Answer 2

In the Python code you posted, you're not using the right streams:

inputready, outputready, exceptready = select.select(
    [proc.stdout, proc.stderr], # read list
    [proc.stdout, proc.stderr], # write list
    [proc.stdout, proc.stderr], # error list.
    0)                          # time out.

I haven't tried fixing it, but I bet reading and writing to the same set of streams is incorrect.

---

There are multiple things going wrong in your sample. The first is that the python executable that you launch as as a child process produces no output. The second is that there is a race condition since you can invoke select() 5 times in a row before the child process produces output, in which case you will kill the process before reading anything.

I fixed the three problems mentioned above (write list, starting a process that produces output and race condition). Try out this sample and see if it works for you:

#!/usr/bin/python
import subprocess, os, select, time

path = "/usr/bin/python"
proc = subprocess.Popen([path, "foo.py"], shell=False,
                        stdin=subprocess.PIPE,
                        stdout=subprocess.PIPE,
                        stderr=subprocess.PIPE)
for i in xrange(0,5):
    time.sleep(1)
    inputready, outputready, exceptready = select.select(
        [proc.stdout, proc.stderr], [proc.stdin,],
        [proc.stdout, proc.stderr, proc.stdin], 0)
    if not inputready:
        print "No Data",
    print inputready, outputready, exceptready
    for s in inputready:
        print s.fileno(),s.readline()

proc.terminate()
print "After Terminating"

for i in xrange(0,5):
    inputready, outputready, exceptready = select.select(
        [proc.stdout, proc.stderr], [proc.stdin,],
        [proc.stdout, proc.stderr, proc.stdin], 0)
    if not inputready:
        print "No Data",
    print inputready, outputready, exceptready
    for s in inputready:
        print s.fileno(),s.readline()

The foo.py file I used contained this:

#!/usr/bin/python
print "Hello, world!"

---

The following version (mostly removed redundant output to make results easier to read):

#!/usr/bin/python
import subprocess, os, select, time

path = "/usr/bin/python"
proc = subprocess.Popen([path, "foo.py"], shell=False,
                        stdin=subprocess.PIPE,
                        stdout=subprocess.PIPE,
                        stderr=subprocess.PIPE)
for i in xrange(0,5):
    time.sleep(1)
    inputready, outputready, exceptready = select.select(
        [proc.stdout, proc.stderr], [proc.stdin,],
        [proc.stdout, proc.stderr, proc.stdin], 0)
    for s in inputready:
        line = s.readline()
        if line:
            print s.fileno(), line

proc.terminate()
print "After Terminating"

for i in xrange(0,5):
    time.sleep(1)
    inputready, outputready, exceptready = select.select(
        [proc.stdout, proc.stderr], [proc.stdin,],
        [proc.stdout, proc.stderr, proc.stdin], 0)
    for s in inputready:
        line = s.readline()
        if line:
            print s.fileno(), line

Gives the following output:

5 Hello, world!

After Terminating

Note that for some reason, using the timeout parameter in select.select() did not produce the expected results on my system, and I resorted to using time.sleep() instead.

---

Just FYI, running python as

/usr/bin/python 2>&1|tee test.out

seems to be working just fine.

You cannot get this effect because this example still gives the python interpreter a controlling tty. Without the controlling tty, the python interpreter does not print the Python version and does not display the >>> prompt.

A close example would be something like the following. You can replace the /dev/null with a file containing commands to send to the interpreter.

/usr/bin/python </dev/null 2>&1|tee test.out

If you redirect anything other than the controlling tty (keyboard) as the standard input to the process, you will get no output from the python interpreter. This is why your code appears not to work.

Answer 3

I use 2-way io in bash like this:

mkfifo hotleg
mkfifo coldleg

program <coldleg |tee hotleg &

while read LINE; do
 case $LINE in
  *)call_a_function $LINE;;
 esac
done <hotleg |tee coldleg &

(note that you can just ">" instead of tee, but you may want to see the output at first)

Answer 4

Your guess that buffered I/O is to blame is most likely correct. The way you wrote your loop, the read will block until it fills the required buffer, and you won't be able to process any input until it returns. This can easily cause a deadlock.

Popen.communicate deals with this by making a thread to work with each pipe, and by making sure it has all the data to be written to stdin, so that the actual write cannot be delayed while the file object waits for a buffer to fill or for the file object to be flushed/closed. I think you could make a solution involving threads work if you needed to, but that's not really asynchronous and probably not the easiest solution.

You can get around python's buffering by not using the file objects provided by Popen to access the pipes, and instead grabbing their fd's using the fileno() method. You can then use the fd's with os.read, os.write, and select.select. The os.read and os.write functions will do no buffering, but they will block until at least one byte can be read/written. You need to make sure the pipe is readable/writeable before calling them. The simplest way to do this is to use select.select() to wait for all the pipes you want to read/write, and make a single read or write call to every pipe that's ready when select() returns. You should be able to find examples of select loops if you search (they'll probably be using sockets instead of pipes, but the principle is the same). (Also, never do a read or write without checking first that it won't block, or you can end up with cases where you cause a deadlock with the child process. You have to be ready to read data even when you haven't yet written everything you want.)

Answer 5

If you need to control a Python interpreter session, you're probably better off with

• embedding Python into your program (plain evals if it's in Python itself), or
• using RPC facilities like rpyc like PyScripter does.

Btw in the latter case, the server can be run anywhere and PyScripter already has a working server module (client module is in Pascal, will need to translate).