Skręcony klient sieciowy z pracownikami wieloprocesowymi?

Question

Tak, mam aplikację, która używa Twisted + Stomper jako klienta STOMP, który prowadzi prace do wieloprocesowości. Pracowników.

To wydaje się działać ok, gdy po prostu użyć skryptu Pythona to odpalić, który (uproszczone) wygląda mniej więcej tak:

# stompclient.py 

logging.config.fileConfig(config_path) 
logger = logging.getLogger(__name__) 

# Add observer to make Twisted log via python 
twisted.python.log.PythonLoggingObserver().start() 

# initialize the process pool. (child processes get forked off immediately) 
pool = multiprocessing.Pool(processes=processes) 

StompClientFactory.username = username 
StompClientFactory.password = password 
StompClientFactory.destination = destination 
reactor.connectTCP(host, port, StompClientFactory()) 
reactor.run() 

Jak to zostanie zapakowana do wdrożenia, myślałem, że chciałbym skorzystać skryptu twistd i uruchom go z pliku tac.

Oto mój bardzo podobne-wyglądający plik tac:

# stompclient.tac 

logging.config.fileConfig(config_path) 
logger = logging.getLogger(__name__) 

# Add observer to make Twisted log via python 
twisted.python.log.PythonLoggingObserver().start() 

# initialize the process pool. (child processes get forked off immediately) 
pool = multiprocessing.Pool(processes=processes) 

StompClientFactory.username = username 
StompClientFactory.password = password 
StompClientFactory.destination = destination 

application = service.Application('myapp') 

service = internet.TCPClient(host, port, StompClientFactory()) 
service.setServiceParent(application) 

Dla przykładu, ja zawalił lub zmienił kilka szczegółów; miejmy nadzieję, że nie stanowiły one sedna problemu. Na przykład moja aplikacja ma system wtyczek, pula jest inicjowana przez osobną metodę, a następnie praca jest delegowana do puli za pomocą metody pool.apply_async() przekazującej jedną z metod process() mojej wtyczki.

Po uruchomieniu skryptu (stompclient.py) wszystko działa zgodnie z oczekiwaniami.

Wydaje się również do pracy OK, jeśli biegnę skręt w trybie non-daemon (N):

twistd -noy stompclient.tac 

jednak robi nie pracę podczas uruchamiania w trybie demona:

twistd -oy stompclient.tac 

Aplikacja wydaje się uruchamiać OK, ale kiedy próbuje rozwidlić pracę, po prostu się zawiesza. Przez "zawiesza się", mam na myśli to, że wygląda na to, że proces potomny nigdy nie jest proszony o nic, a rodzic (który nazywa się pool.apply_async()) siedzi po prostu czekając na odpowiedź, aby powrócić.

Jestem pewien, że robię coś głupiego z Twisted + multiprocessing, ale naprawdę mam nadzieję, że ktoś może wyjaśnić moją wadę w moim podejściu.

Z góry dziękuję!

Answer 1

Ponieważ różnica między wywołaniem roboczym a nieumiejętnym wywołaniem jest tylko opcją "-n", wydaje się, że najprawdopodobniej problem jest spowodowany procesem demonizacji (którego "-n" nie zapobiega).

W systemie POSIX jednym z kroków związanych z demonizacją jest rozwidlenie i wyjście nadrzędne. To między innymi dlatego, że twój kod jest uruchamiany w innym procesie niż ten, w którym oceniano plik .tac. To również zmienia relację potomek/rodzic procesów, które zostały uruchomione w pliku .tac - jako pulę procesów wieloprocesowych.

Procesy w puli wieloprocesorowej rozpoczynają się od rodzica rozpoczętego procesu twistd. Jednakże, gdy proces ten kończy się jako część demonizacji, ich rodzic staje się procesem inicjalizacji systemu. Może to powodować pewne problemy, chociaż prawdopodobnie nie jest to problem, który opisałeś.Prawdopodobnie istnieją również inne szczegóły implementacji niskiego poziomu, które normalnie umożliwiają działanie modułu wieloprocesowego, ale które są przerywane przez proces demonizacji.

Na szczęście unikanie tej dziwnej interakcji powinno być proste. Funkcje API serwisów Twisted umożliwiają uruchomienie kodu po zakończeniu demonizacji. Jeśli korzystasz z tych interfejsów API, możesz opóźnić inicjalizację puli procesowej modułu wieloprocesowego do czasu demonizacji i, miejmy nadzieję, uniknąć problemu. Oto przykład, co to może wyglądać następująco:

from twisted.application.service import Service 

class MultiprocessingService(Service): 
    def startService(self): 
     self.pool = multiprocessing.Pool(processes=processes) 

MultiprocessingService().setServiceParent(application) 

Teraz, osobno, można również uruchomić w problemach związanych oczyścić procesów potomnych modułu wieloprocesorowe, albo ewentualnie problemy z procesów tworzonych ze skręconymi w proces tworzenia API Reactor.spawnProcess. Dzieje się tak dlatego, że częśd prawidłowego postępowania z procesami podrzędnymi ogólnie obejmuje obsługę sygnału SIGCHLD. Skręceni i wieloprocesorowi nie będą współpracować w tym zakresie, więc jeden z nich zostanie powiadomiony o wszystkich dzieciach wychodzących, a drugi nigdy nie zostanie powiadomiony. Jeśli w ogóle nie korzystasz z Twisted's API do tworzenia procesów potomnych, może to być w porządku dla ciebie - ale możesz chcieć sprawdzić, czy jakakolwiek obsługa sygnału, którą moduł wieloprocesowy próbuje zainstalować, faktycznie "wygrywa" i nie otrzymuje zastąpiony przez własny uchwyt Twisted.

Answer 2

Możliwym pomysł dla Ciebie ...

Podczas pracy w trybie demona twistd zamknie stdin, stdout i stderr. Czy coś, co Twoi klienci czytają lub piszą do nich?