Python: неблокирующее чтение из стандартного потока подпроцесса - PullRequest
Новая
       31

Python: неблокирующее чтение из стандартного потока подпроцесса

0 голосов
/ 18 марта 2010

У меня есть скрипт (worker.py), который печатает небуферизованный вывод в виде ... 

1
2
3
.
.
.
n

где n - это некоторое постоянное число итераций, которое выполнит цикл в этом скрипте. В другом скрипте (service_controller.py) я запускаю несколько потоков, каждый из которых запускает подпроцесс с помощью subprocess.Popen (stdout = subprocess.PIPE, ...); Теперь в моем основном потоке (service_controller.py) я хочу прочитать вывод подпроцесса каждого потока worker.py и использовать его для вычисления оценки времени, оставшегося до завершения.

У меня есть вся логика, которая читает stdout из worker.py и определяет последнее напечатанное число. Проблема в том, что я не могу понять, как сделать это неблокирующим способом. Если я читаю постоянный размер буфера, то каждое чтение будет ждать одинаковых данных от каждого из работников. Я пробовал множество способов, включая использование fcntl, select + os.read и т. Д. Какой мой лучший вариант здесь? При необходимости я могу опубликовать свой источник, но я решил, что объяснение достаточно хорошо описывает проблему.

Спасибо за любую помощь здесь.

EDIT
Добавление примера кода

У меня есть работник, который запускает подпроцесс. 

class WorkerThread(threading.Thread):
    def __init__(self):
        self.completed = 0
        self.process = None
        self.lock = threading.RLock()
        threading.Thread.__init__(self)

    def run(self):
        cmd = ["/path/to/script", "arg1", "arg2"]
        self.process = subprocess.Popen(cmd, stdout=subprocess.PIPE, bufsize=1, shell=False)
        #flags = fcntl.fcntl(self.process.stdout, fcntl.F_GETFL)
        #fcntl.fcntl(self.process.stdout.fileno(), fcntl.F_SETFL, flags | os.O_NONBLOCK)

    def get_completed(self):
        self.lock.acquire();
        fd = select.select([self.process.stdout.fileno()], [], [], 5)[0]
        if fd:
            self.data += os.read(fd, 1)
            try:
                self.completed = int(self.data.split("\n")[-2])
            except IndexError:
                pass
        self.lock.release()
        return self.completed

У меня есть ThreadManager. 

class ThreadManager():
    def __init__(self):
        self.pool = []
        self.running = []
        self.lock = threading.Lock()

    def clean_pool(self, pool):
        for worker in [x for x in pool is not x.isAlive()]:
            worker.join()
            pool.remove(worker)
            del worker
        return pool

    def run(self, concurrent=5):
        while len(self.running) + len(self.pool) > 0:
            self.clean_pool(self.running)
            n = min(max(concurrent - len(self.running), 0), len(self.pool))
            if n > 0:
                for worker in self.pool[0:n]:
                    worker.start()
                self.running.extend(self.pool[0:n])
                del self.pool[0:n]
            time.sleep(.01)
         for worker in self.running + self.pool:
             worker.join()

и некоторый код для его запуска. 

threadManager = ThreadManager()
for i in xrange(0, 5):
    threadManager.pool.append(WorkerThread())
threadManager.run()

Я удалил журнал другого кода в надежде попытаться определить проблему.

Ответы [ 2 ]

2 голосов
/ 18 марта 2010

Вместо того, чтобы ваш service_controller был заблокирован из-за доступа ввода / вывода, только цикл потока должен читать свой собственный вывод контролируемого процесса.

тогда у вас может быть метод в поточном объекте, управляющий процессом, чтобы получить последний результат опроса.

конечно, не забудьте в этом случае использовать некоторый механизм блокировки для защиты буфера, который будет использоваться как потоком для его заполнения, так и методом, вызываемым контроллером для его получения.

1 голос
/ 21 сентября 2011

Ваш метод WorkerThread.run () запускает подпроцесс, а затем немедленно завершает работу. Функция Run () должна выполнить опрос и обновлять WorkerThread.completed до завершения подпроцесса.

...