16

The problem: I have a poorly designed Fortran program (I cannot change it, I'm stuck with it) which takes text input from stdin and other input files, and writes text output results to stdout and other output files. The size of input and out is quite large, and I would like to avoid writing to the hard drive (slow operation). I have written a function that iterates over the lines of the several input files, and I also have parsers for multiple output. I don't really know if the program first read all the input and then starts to output, or starts outputting while reading the input.

The goal: To have a function that feeds the external program with what it wants, and parses the output as it comes from the program, without writing data to text files on the hard drive.

Research: The naive way using files is:

from subprocess import PIPE, Popen

def execute_simple(cmd, stdin_iter, stdout_parser, input_files, output_files):

    for filename, file_iter in input_files.iteritems():
        with open(filename ,'w') as f:
            for line in file_iter:
                f.write(line + '\n')


    p_sub = Popen(
        shlex.split(cmd),
        stdin = PIPE,
        stdout = open('stdout.txt', 'w'),
        stderr = open('stderr.txt', 'w'),
        bufsize=1
    )
    for line in stdin_iter:
        p_sub.stdin.write(line + '\n')

    p_sub.stdin.close()
    p_sub.wait()

    data = {}
    for filename, parse_func in output_files.iteritems():
        # The stdout.txt and stderr.txt is included here
        with open(filename,'r') as f:
            data[filename] = parse_func(
                    iter(f.readline, b'')
            )
    return data

I have tried to and the subprocess module to execute the external program together. The additional input/output files are handled with named pipes and multiprocessing. I want to feed stdin with an iterator (which returns the lines for input), save the stderr in a list, and parse the stdout as it comes from the external program. The input and output can be quite large, so using communicate is not feasible.

I have a parser on the format:

def parser(iterator):
    for line in iterator:
        # Do something
        if condition:
            break
    some_other_function(iterator)
    return data

I looked at this solution using select to choose the appropriate stream, however I don't know how to make it work with my stdout parser and how to feed the stdin.

I also look the asyncio module, but as I can see I will have the same problem with the parsing of stout.

18
  • If you start the Fortran program before you start any threads, you can start feeding the Fortran program data from a separate thread, and collect its output from the main thread. Alternatively, you could start an additional Python program, and pass it the Fortran's stdin for its stdout, and once again, your main program simply handles the Fortran program's stdout. Commented Jul 27, 2015 at 13:57
  • I think you're mixing things slightly. Processing something in time (on arrival) is not the same as asynchronous IO. For asynchronous IO on Unixes you can use asyncore module. It uses select, although it is not exactly ment for files. Why do you replace stdout and stderr with files instead of doing stdout.read() or stdout.readlines()? You can specify number of bytes and so you won't have to wait until it finishes. If you are worried about hard drive, mount a directory in RAM memory and write there.
    – Dalen
    Commented Jul 27, 2015 at 14:44
  • I mean, select is not ment for asynchronous file IO, just sockets. But you can certainly check whether some file is ready for wanted action.
    – Dalen
    Commented Jul 27, 2015 at 14:49
  • @Dalen: yeah I think an option like twisted could be an option.
    – crlb
    Commented Jul 27, 2015 at 14:59
  • 1
    As far as I am aware, twisted doesn't do files or terminals. So It cannot help.
    – Dalen
    Commented Jul 27, 2015 at 17:49

2 Answers 2

7
+50

You should use named pipes for all input and output to the Fortran program to avoid writing to disk. Then, in your consumer, you can use threads to read from each of the program's output sources and add the information to a Queue for in-order processing.

To model this, I created a python app daemon.py that reads from standard input and returns the square root until EOF. It logs all input to a log file specified as a command-line argument and prints the square root to stdout and all errors to stderr. I think it simulates your program (of course the number of output files is only one, but it can be scaled). You can view the source code for this test application here. Note the explicit call to stdout.flush(). By default, the standard output is print buffered, which means that this is output at the end and messages will not arrive in order. I hope your Fortran application does not buffer its output. I believe that my sample application will probably not run on Windows, due to a Unix-only use of select, which shouldn't matter in your case.

I have my consumer application which starts the daemon application as a subprocess, with stdin, stdout and stderr redirected to subprocess.PIPEs. each of these pipes is given to a different thread, one to give input, and three to handle the log file, errors and standard output respectively. They all add their messages to a shared Queue which your main thread reads from and sends to your parser.

This is my consumer's code:

import os, random, time
import subprocess
import threading
import Queue
import atexit

def setup():
    # make a named pipe for every file the program should write
    logfilepipe='logpipe'
    os.mkfifo(logfilepipe)

def cleanup():
    # put your named pipes here to get cleaned up
    logfilepipe='logpipe'
    os.remove(logfilepipe)

# run our cleanup code no matter what - avoid leaving pipes laying around
# even if we terminate early with Ctrl-C
atexit.register(cleanup)

# My example iterator that supplies input for the program. You already have an iterator 
# so don't worry about this. It just returns a random input from the sample_data list
# until the maximum number of iterations is reached.
class MyIter():
    sample_data=[0,1,2,4,9,-100,16,25,100,-8,'seven',10000,144,8,47,91,2.4,'^',56,18,77,94]
    def __init__(self, numiterations=1000):
        self.numiterations=numiterations
        self.current = 0

    def __iter__(self):
        return self

    def next(self):
        self.current += 1
        if self.current > self.numiterations:
            raise StopIteration
        else:
            return random.choice(self.__class__.sample_data)

# Your parse_func function - I just print it out with a [tag] showing its source.
def parse_func(source,line):
    print "[%s] %s" % (source,line)

# Generic function for sending standard input to the problem.
# p - a process handle returned by subprocess
def input_func(p, queue):
    # run the command with output redirected
    for line in MyIter(30): # Limit for testing purposes
        time.sleep(0.1) # sleep a tiny bit
        p.stdin.write(str(line)+'\n')
        queue.put(('INPUT', line))
    p.stdin.close()
    p.wait()

    # Once our process has ended, tell the main thread to quit
    queue.put(('QUIT', True))

# Generic function for reading output from the program. source can either be a
# named pipe identified by a string, or subprocess.PIPE for stdout and stderr.
def read_output(source, queue, tag=None):
    print "Starting to read output for %r" % source
    if isinstance(source,str):
        # Is a file or named pipe, so open it
        source=open(source, 'r') # open file with string name
    line = source.readline()
    # enqueue and read lines until EOF
    while line != '':
        queue.put((tag, line.rstrip()))
        line = source.readline()

if __name__=='__main__':
    cmd='daemon.py'

    # set up our FIFOs instead of using files - put file names into setup() and cleanup()
    setup()

    logfilepipe='logpipe'

    # Message queue for handling all output, whether it's stdout, stderr, or a file output by our command
    lq = Queue.Queue()

    # open the subprocess for command
    print "Running command."
    p = subprocess.Popen(['/path/to/'+cmd,logfilepipe],
                                    stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

    # Start threads to handle the input and output
    threading.Thread(target=input_func, args=(p, lq)).start()
    threading.Thread(target=read_output, args=(p.stdout, lq, 'OUTPUT')).start()
    threading.Thread(target=read_output, args=(p.stderr, lq, 'ERRORS')).start()

    # open a thread to read any other output files (e.g. log file) as named pipes
    threading.Thread(target=read_output, args=(logfilepipe, lq, 'LOG')).start()

    # Now combine the results from our threads to do what you want
    run=True
    while(run):
        (tag, line) = lq.get()
        if tag == 'QUIT':
            run=False
        else:
            parse_func(tag, line)

My iterator returns a random input value (some of which are junk to cause errors). Yours should be a drop-in replacement. The program will run until the end of its input and then wait for the subprocess to complete before enqueueing a QUIT message to your main thread. My parse_func is obviously super simple, simply printing out the message and its source, but you should be able to work with something. The function to read from an output source is designed to work with both PIPEs and strings - don't open the pipes on your main thread because they will block until input is available. So for file readers (e.g. reading log files), it's better to have the child thread open the file and block. However, we spawn the subprocess on the main thread so we can pass the handles for stdin, stdout and stderr to their respective child threads.

Based partially on this Python implementation of multitail.

3
  • Thank you for the response. There are some issues I have encountered: 1, This script does not end with a signal, it just hangs. 2, why do you have to have time.sleep(0.1) in MyIter? . 3, The queue gets alternating number of lines between executions.
    – crlb
    Commented Oct 14, 2015 at 9:51
  • Thanks for your response. I had wondered if you found it useful. As far as point 1, you can check for SIGINT using the signal module or by catching a KeyboardInterrupt (see this answer). You could check it in the while(run): loop, and maybe also implement some thread cleanup. For point 2 - I put the sleep in there to slow down the output so it was readable and you could see what was going on, but it's not necessary. I'll add a comment to that effect.
    – Aaron D
    Commented Oct 14, 2015 at 12:17
  • I think the approach is the right way to go, even though I have not yet succeed in making it work in my application. I will try out the catching the signal.
    – crlb
    Commented Oct 14, 2015 at 12:37
0

It is very important that the Fortran program calls flush at the end of every job (can be also moe frequently), if you are waiting for end of results before you send a new job.
The command depends on the compiler, e.g. GNU fortran CALL FLUSH(unitnumber) or it can be simulated by closing the outpud and open again for append.

You can also easily write a few empty lines with many whitespace characters at the end that it fill the buffer size and you get a new block of data. 5000 whitespace characters is probably good enough, but not too much that it will block the Fortran side of pipe. If you read these empty lines just after sending the new job, you even don't need a non blocking read. The last line of the job can be easily recognized in numeric applications. If you would write a "chat" application you need something what other people wrote.

4
  • this is worth considering, however this doesn't address the whole problem. The solution of the feeding input and parsing output in an arbitrarily manner from the fortran program is more like a "chat" application.
    – crlb
    Commented Aug 7, 2015 at 8:32
  • 1
    You can actually work with a program that does any number of things. The OP says he's stuck with a particular Fortran program and cannot change it; the assumption should be that (a) he knows what he's talking about; and (b) it can be made to work since the reason he's stuck with it is that it was, in fact, working... Commented Aug 7, 2015 at 11:33
  • @PatrickMaupin: understand the problem. I remember a program that worked correctly in the terminal but if it recognized that its stdout is a pipe, it didn't flushed acceptable. One solution would be to create and contol a virtual terminal programmatically (complicated) other solution was to force flush by terminating the program by closing its input and start it again frequenty. (perfect for simple programs, unuseable for nontrivial programs) Then it is really interesting to think about the remote side, even I forbade it initially.
    – hynekcer
    Commented Aug 8, 2015 at 1:21
  • The Python pty module works well under linux, and the program is unlikely to care about the rest of the files besides stdin/stdout, so named pipes will likely work fine for those, unless the Fortran program is doing seeks. Commented Aug 8, 2015 at 1:36

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.