Я пытаюсь использовать сельдерей в качестве канала управления для витой программы. My Twisted application - это уровень абстракции, который обеспечивает стандартный интерфейс для различных локально запущенных процессов (через ProcessProtocol). Я бы хотел использовать Celery для управления этим дистанционно - AMQP кажется идеальным методом управления многими Twisted приложениями из центрального местоположения, и я хотел бы воспользоваться функциями на основе задач Celery, например. задачи, подзадачи и т.д.
Это не работает так, как я планировал, и я надеюсь, что кто-то может помочь мне указать мне в правильном направлении, чтобы это работало.
Поведение, которое я пытаюсь достичь, когда я запускаю свой script, это:
- Начните слегка модифицированный сельдерей (см. ниже)
- Ожидание задач Сельдерея
- Когда получена задача "начать процесс", создайте ProcessProtocol
- Когда выполняются другие задачи, запустите функцию в Twisted-протоколе и верните результат, используя команду Отложенные
"слегка модифицированный сельдерей" celeryd с небольшой модификацией, которая позволяет задачам обращаться к реактору Twisted через self.app.twisted, и порожденный процесс через self.app.process. Чтобы все было в порядке, я использую интегрирование Cycle "solo" для пула процессов, которое не разворачивает новый процесс для рабочих.
Моя проблема возникает, когда я пытаюсь использовать задачу Celery для инициализации ProcessProtocol (т.е. запускающего внешний процесс). Процесс запускается правильно, но дочерний процесс ProcessProtocol childDataReceived никогда не вызывается. Я думаю, что это связано с тем, что дескрипторы файлов не наследуются/устанавливаются правильно.
Ниже приведен пример кода примера, основанного на примере "wc" в документации ProcessProtocol. Он включает в себя две задачи Celery - один для запуска процесса wc, а другой - подсчет слов в некотором тексте (с использованием ранее начатого процесса wc).
Этот пример довольно надуман, но если я смогу выполнить эту работу, он послужит хорошей отправной точкой для реализации моих ProcessProtocols, которые являются длительными процессами, которые будут реагировать на команды, написанные на stdin.
Я тестирую это, сначала запуская демона Celery:
python2.6 mycelery.py -l info -P solo
Затем в другом окне выполняется script, который отправляет две задачи:
python2.6 command_test.py
Ожидаемое поведение command_test.py выполняется для двух команд: один запускает процесс wc, а другой отправляет некоторый текст в CountWordsTask. Что на самом деле происходит:
- StartProcTask запускает процесс и получает "процесс запущен" в качестве ответа через Deffered
- CountWordsTask никогда не получает результат, потому что childDataReceived никогда не вызывается
Может кто-нибудь пролить свет на это или предложить некоторые советы о том, как лучше использовать Сельдерей в качестве канала управления для Twisted ProcessProtocols?
Было бы лучше написать реалистичную версию ProcessPool для Celery? Является ли мой метод вызова WorkerCommand.execute_from_commandline через reactor.callLater правильным подходом для обеспечения того, чтобы все произошло в Twisted thread?
Я читал об AMPoule, который, я думаю, мог бы предоставить некоторые из этих функций, но хотел бы придерживаться Сельдерея, если это возможно, поскольку я использую его в других частях моего приложения.
Любая помощь или помощь будут с благодарностью оценены!
myceleryd.py
from functools import partial
from celery.app import App
from celery.bin.celeryd import WorkerCommand
from twisted.internet import reactor
class MyCeleryApp(App):
def __init__(self, twisted, *args, **kwargs):
self.twisted = twisted
super(MyCeleryApp, self).__init__(*args, **kwargs)
def main():
get_my_app = partial(MyCeleryApp, reactor)
worker = WorkerCommand(get_app=get_my_app)
reactor.callLater(1, worker.execute_from_commandline)
reactor.run()
if __name__ == '__main__':
main()
protocol.py
from twisted.internet import protocol
from twisted.internet.defer import Deferred
class WCProcessProtocol(protocol.ProcessProtocol):
def __init__(self, text):
self.text = text
self._waiting = {} # Dict to contain deferreds, keyed by command name
def connectionMade(self):
if 'startup' in self._waiting:
self._waiting['startup'].callback('process started')
def outReceived(self, data):
fieldLength = len(data) / 3
lines = int(data[:fieldLength])
words = int(data[fieldLength:fieldLength*2])
chars = int(data[fieldLength*2:])
self.transport.loseConnection()
self.receiveCounts(lines, words, chars)
if 'countWords' in self._waiting:
self._waiting['countWords'].callback(words)
def processExited(self, status):
print 'exiting'
def receiveCounts(self, lines, words, chars):
print >> sys.stderr, 'Received counts from wc.'
print >> sys.stderr, 'Lines:', lines
print >> sys.stderr, 'Words:', words
print >> sys.stderr, 'Characters:', chars
def countWords(self, text):
self._waiting['countWords'] = Deferred()
self.transport.write(text)
return self._waiting['countWords']
tasks.py
from celery.task import Task
from protocol import WCProcessProtocol
from twisted.internet.defer import Deferred
from twisted.internet import reactor
class StartProcTask(Task):
def run(self):
self.app.proc = WCProcessProtocol('testing')
self.app.proc._waiting['startup'] = Deferred()
self.app.twisted.spawnProcess(self.app.proc,
'wc',
['wc'],
usePTY=True)
return self.app.proc._waiting['startup']
class CountWordsTask(Task):
def run(self):
return self.app.proc.countWords('test test')
