How to create non-blocking continuous reading from `stdin`?

Question

I have a single process, which has been created like this:

p = subprocess.Popen(args   = './myapp',
                     stdin  = subprocess.PIPE,
                     stdout = subprocess.PIPE,
                     universal_newlines=True)

Later on, I'm trying to write to p's stdin:

p.stdin.write('my message\n')

The myapp process has the following setup:

q = queue.Queue()
def get_input():
    for line in iter(sys.stdin.readline, ''):
        q.put(line)
    sys.stdin.close()

threading.Thread(name   = 'input-getter',
                 target = get_input).start()

And it is trying to read new lines continuously, like so:

try:
    print('input:', q.get_nowait())
except Empty:
    print('no input')

Unfortunately the subprocess never receives any of my messages. Of course, when I use:

p.communicate('my message\n')

the subprocess recieves the message, but as expected, the communicate method closes p's stdin, so there are no more communications going on.

Answer 1

p = subprocess.Popen(args   = './myapp',
                     stdin  = subprocess.PIPE,
                     stdout = subprocess.PIPE,
                     universal_newlines=True)

while p.poll() is None:
    data = p.stdout.readline()

This will create a non-blocking read of your process until the process exits. However, there are some cautions to be aware of here. For instance, if you would pipe stderr as well, but not read from it.. Then you will most likely fill a buffer or two and you will hang the program anyway. So always make sure you clear out any buffer I/O's when doing things manually.

A better alternative would be to use select.epoll() if possible, this is only available on unix systems but gives you a hell of a lot better performance and error handling :)

epoll = select.epoll()
epoll.register(p.stdout.fileno(), select.EPOLLHUP) # Use select.EPOLLIN for stdin.

for fileno, event in epoll.poll(1):
    if fileno == p.stdout.fileno():
        # ... Do something ...

NOTE: Remember that whenever a process expects input, it usually indicates this via stdout, so you'll still register STDOUT with select.epoll in order to check for "waiting for input". You can register select.EPOLLIN to check if input was given, but I hardly see the point because remember, that would what you choose to input to the process which you should already be aware is "happening".

Checking if the process expects input

You can use select.epoll to check if the process is awaiting input or not without blocking your application execution with the above example. But there are better alternatives.

Pexpect is one library that does it really well and works with SSH for instance.

It works a little bit different from subprocess but might be a good alternative.

Getting subprocess.popen to work with SSH

I'll redirect to another question+answer if this is what you're after (because SSH will spawn a stdin in a protected manner.

Python + SSH Password auth (no external libraries or public/private keys)?

Answer 2

I think you are maybe just not seeing the output of what is going on. Here's a complete example that seems to work on my box, unless I'm completely misunderstanding what you want. The main change I made is setting stdout for p to sys.stdout instead of subprocess.PIPE. Maybe I'm misunderstanding the thrust of your question and that bit is crucial...

Here's the full code and output:

In the sending (testing) process (I named it test_comms.py). I'm on Windows currently, so excuse the .bat:

import time
import subprocess
import sys

# Note I'm sending stdout to sys.stdout for observation purposes
p = subprocess.Popen(args = 'myapp.bat',
                     stdin  = subprocess.PIPE,
                     stdout = sys.stdout,
                     universal_newlines=True)

#Send 10 messages to the process's stdin, 1 second apart                    
for i in range(10):
    time.sleep(1)
    p.stdin.write('my message\n')

myapp.bat is trivially:

echo "In the bat cave (script)"
python myapp.py

myapp.py contains (using Queue rather than queue - current environment Python 2):

import Queue
from Queue import Empty
import threading
import sys
import time

def get_input():
    print("Started the listening thread")
    for line in iter(sys.stdin.readline, ''):
        print("line arrived to put on the queue\n")
        q.put(line)
    sys.stdin.close()

print("Hi, I'm here via popen")    
q = Queue.Queue()

threading.Thread(name   = 'input-getter',
                 target = get_input).start()

print("stdin listener Thread created and started")

# Read off the queue - note it's being filled asynchronously based on 
# When it receives messages.  I set the read interval below to 2 seconds
# to illustrate the queue filling and emptying.
while True:
    time.sleep(2)
    try:
        print('Queue size is',q.qsize())
        print('input:', q.get_nowait())
    except Empty:
        print('no input')

print("Past my end of code...")

Output:

D:\>comms_test.py

D:\>echo "In the bat cave (script)"
"In the bat cave (script)"

D:\>python myapp.py
Hi, I'm here via popen
Started the listening threadstdin listener Thread created and started

line arrived to put on the queue

line arrived to put on the queue

('Queue size is', 2)
('input:', 'my message\n')
line arrived to put on the queue

line arrived to put on the queue

('Queue size is', 3)
('input:', 'my message\n')
line arrived to put on the queue

line arrived to put on the queue

('Queue size is', 4)
('input:', 'my message\n')
line arrived to put on the queue

line arrived to put on the queue

('Queue size is', 5)
('input:', 'my message\n')
line arrived to put on the queue

line arrived to put on the queue


D:\>('Queue size is', 6)
('input:', 'my message\n')
('Queue size is', 5)
('input:', 'my message\n')
('Queue size is', 4)
('input:', 'my message\n')
('Queue size is', 3)
('input:', 'my message\n')
('Queue size is', 2)
('input:', 'my message\n')
('Queue size is', 1)
('input:', 'my message\n')
('Queue size is', 0)
no input
('Queue size is', 0)
no input
('Queue size is', 0)
no input

Answer 3

For everything working fine you have to flush output in main process (p.stdout) and subprocess (sys.stdout) .

communicate does both flush:

• it flush the p.stdin when closing it
• it wait the sys.stdout output to be flushed (just before exiting)

example of working main.py

import subprocess,time
import sys
p = subprocess.Popen(args   = ['python3', './myapp.py'],
                     stdin  = subprocess.PIPE,
                     stdout = subprocess.PIPE,
                     universal_newlines=True)

time.sleep(0.5)
p.stdin.write('my message\n')
p.stdin.flush()
#print("ici")
for i,l in  enumerate(iter(p.stdout.readline, ''),start=1):

    print("main:received:",i,repr(l))
    if i == 6:
        break
    print("mainprocess:send:other message n°{}".format(i))
    p.stdin.write("other message n°{}\n".format(i))
    p.stdin.flush()

print("main:waiting for subprocess")
p.stdin.close()    
p.wait()

example of myapp.py import queue,threading,sys,time,rpdb

q = queue.Queue()
def get_input():
    for line in iter(sys.stdin.readline, ''):
        q.put(line)
    sys.stdin.close()

threading.Thread(name   = 'input-getter',
                 target = get_input).start()
for i in range(6):
    try:
        l= q.get_nowait()
        print('myapp:input:', l,end="")
        sys.stdout.flush()

    except queue.Empty:
        print("myapp:no input")
        sys.stdout.flush()    
        time.sleep(1)

result:

main:received: 1 'myapp:no input\n'
mainprocess:send:other message n°1
main:received: 2 'myapp:input: my message\n'
mainprocess:send:other message n°2
main:received: 3 'myapp:input: other message n°1\n'
mainprocess:send:other message n°3
main:received: 4 'myapp:no input\n'
mainprocess:send:other message n°4
main:received: 5 'myapp:input: other message n°2\n'
mainprocess:send:other message n°5
main:received: 6 'myapp:input: other message n°3\n'
main:waiting for subprocess

Answer 4

Trying to investigate your program, I wrote my own "continually stream stuff to cat and catch what it returns" program. I didn't implement the subprocess side of it, but hopefully the structure is similar.

This line is very odd about your program...

for line in iter(sys.stdin.readline, ''):
    q.put(line)
sys.stdin.close()

That looks an awful lot like

for line in stdin:
    q.put(line)

Note that the loop is going to end when the pipe is closed and there's no need to re-close it afterwards.

If you need to continously asynchronously read stdin, you should be able to construct a reading thread near-identical to child_reader in the code below. Just replace child.stdout with stdin.

import subprocess
import threading
import random

# We may need to guard this?
child = subprocess.Popen('cat', stdout=subprocess.PIPE, stdin=subprocess.PIPE)

# Continuously print what the process outputs...
def print_child():
    for line in child.stdout:
        print(line)

child_reader = threading.Thread(target = print_child)
child_reader.start()

for i in range(10000):
    chars = 'ABC\n'
    child.stdin.write(random.choice(chars).encode())

# Send EOF.
# This kills the cat.
child.stdin.close()

# I don't think order matters here?
child.wait()
child_reader.join()

Answer 5

I've written a program that does... basically everything involving IO asynchronously. It reads input on a thread, it outputs on a thread, it creates a process, and it communicates with that process on a thread.

I am not sure exactly what your program needs to accomplish, but hopefully this code accomplishes it.

# Asynchronous cat program!

# Asynchronously read stdin
# Pump the results into a threadsafe queue
# Asynchronously feed the contents to cat
# Then catch the output from cat and print it
# Thread all the things

import subprocess
import threading
import queue
import sys

my_queue = queue.Queue()

# Input!
def input_method():
    for line in sys.stdin: # End on EOF
        if line == 'STOP\n': # Also end on STOP
            break
        my_queue.put(line)
input_thread = threading.Thread(target=input_method)
input_thread.start()

print ('Input thread started')


# Subprocess!
cat_process = subprocess.Popen('cat', stdout=subprocess.PIPE, stdin=subprocess.PIPE)

print ('cat process started')

queue_alive = True
# Continuously dump the queue into cat
def queue_dump_method():
    while queue_alive:
        try:
            line = my_queue.get(timeout=2)
            cat_process.stdin.write(line.encode())
            cat_process.stdin.flush() # For some reason, we have to manually flush
            my_queue.task_done() # Needed?
        except queue.Empty:
            pass
queue_dump_thread = threading.Thread(target = queue_dump_method)
queue_dump_thread.start()

print ('Queue dump thread started')

# Output!
def output_method():
    for line in cat_process.stdout:
        print(line)
output_thread = threading.Thread(target=output_method)
output_thread.start()

print ('Output thread started')


# input_thread will die when we type STOP
input_thread.join()
print ('Input thread joined')

# Now we wait for the queue to finish processing
my_queue.join()
print ('Queue empty')

queue_alive = False
queue_dump_thread.join()
print ("Queue dump thread joined")

# Send EOF to cat
cat_process.stdin.close()

# This kills the cat
cat_process.wait()
print ('cat process done')

# And make sure we're done outputting
output_thread.join()
print ('Output thread joined')