Jak poprawnie zamknąć procesy robocze Pythona RQ w sposób dynamiczny?

Question

Korzystając z , staramy się dynamicznie zarządzać procesami roboczymi. Używamy niestandardowy zbudowany skrypt pracownik, który (w uproszczonej formie) jest następujący:

from rq import Connection, Worker 

queues_to_listen_on = get_queues_to_listen_on() 

with Connection(connection = get_worker_connection()): 
    w = Worker(queues_to_listen_on) 
    w.work() 

Jesteśmy szczególnie zainteresowani wyłączenia pracowników. Głównym problemem, jaki mamy, jest to, jak wykonać pełne wdzięku zamknięcie pracownika, w sposób umożliwiający ukończenie bieżącej pracy przed zamknięciem. Obsługa sygnału request_stop(...) na odpowiednim obiekcie Worker wydaje się robić to, czego potrzebujemy, ale wydaje się, że nie ma żadnego sposobu (przynajmniej według mojej wiedzy) na jego emisję, chyba że poprzez naciśnięcie CTRL+C na proces roboczy uruchomiony w terminalu.

Jak ja to widzę, istnieją dwa prawdopodobne rozwiązania (nie może z pewnością być więcej) - w kolejności preferencji:

1. programowo przy użyciu biblioteki rq, wysyła sygnał do request_stop a tym samym wywołać wdzięku shutdown .
2. W jakiś sposób pobierz pid z prawidłowego procesu (nie wiesz, czy proces workhorse lub proces nasłuchujący pracownika) i używając innej metody, wyślij odpowiedni sygnał do tego procesu. Mamy na to kilka sposobów, ale prawdopodobnie wymagałoby to więcej pracy i wprowadzenia innych zmiennych do problemu, który wolałbym pominąć (na przykład, używając Fabric do uruchomienia zdalnego polecenia lub czegoś podobnego).

Jeśli istnieje lepszy sposób na rozwiązanie tego problemu lub inną alternatywę, która pozwoliłaby osiągnąć ten sam cel, byłbym wdzięczny za sugestie.

Answer 1

Opcja 1 jest zdecydowanie lepsza pod względem wzornictwa.

jednak do rozwiązania konkretnego problemu z konieczności korzystania CTRL + C aby zakończyć proces (nienawidzę tego też), można użyć następującą strategię dla swoich pracowników:

# WORKER_NAME.py 
import os 

PID = os.getpid() 

@atexit.register 
def clean_shut(): 
    print "Clean shut performed" 

    try: 
     os.unlink("WORKER_NAME.%d" % PID) 
    except: 
     pass 

# Worker main 
def main(): 
    f = open("WORKER_NAME.%d" % PID, "w") 
    f.write("Delete this to end WORKER_NAME gracefully") 
    f.close() 

    while os.path.exists("WORKER_NAME.%d" % PID): 
     # Worker working 

I w skrypcie master dostać PID pracownika jako @Borys zasugerował, wysłać żądanie zatrzymania ciepłego i os.unlink("path/to/WORKER_NAME.%d" % worker_PID), aby zabezpieczyć wdzięczne wyłączenie :)

Dotyczy to tylko pracowników z pętlą nieskończoną. Jeśli proces roboczy wywołuje rzeczy, które blokują nawet zwykłe, sekwencyjne jednorazowe zadanie, musisz przejść dalej do możliwej procedury blokowania, aby rozwiązać problem, na przykład zastosować jakąś strategię limitu czasu.