У меня есть скрипт (worker.py), который выводит небуферизованный вывод в виде...
1
2
3
.
.
.
n
где n — некоторое постоянное количество итераций, которое будет выполнять цикл в этом скрипте. В другом скрипте (service_controller.py) я запускаю несколько потоков, каждый из которых запускает подпроцесс, используя subprocess.Popen(stdout=subprocess.PIPE, ...); Теперь в моем основном потоке (service_controller.py) я хочу прочитать вывод подпроцесса worker.py каждого потока и использовать его для расчета оценки времени, оставшегося до завершения.
У меня работает вся логика, которая считывает стандартный вывод из worker.py и определяет последнее напечатанное число. Проблема в том, что я не могу понять, как это сделать неблокирующим способом. Если я прочитаю постоянный размер буфера, то каждое чтение будет заканчиваться ожиданием одних и тех же данных от каждого из рабочих. Я пробовал множество способов, включая использование fcntl, select + os.read и т. д. Какой здесь лучший вариант? Я могу опубликовать свой источник, если это необходимо, но я решил, что объяснение описывает проблему достаточно хорошо.
Спасибо за любую помощь здесь.
ИЗМЕНИТЬ
Добавление примера кода
У меня есть рабочий, который запускает подпроцесс.
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()
Я удалил журнал другого кода в надежде попытаться точно определить проблему.
