blob: 33ecf18530671ccd1fb6f86f235121755a36a90b (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
import logging
from rq import get_current_connection
from rq import Worker
logger = logging.getLogger(__name__)
def cleanup_ghosts(conn=None):
"""
RQ versions < 0.3.6 suffered from a race condition where workers, when
abruptly terminated, did not have a chance to clean up their worker
registration, leading to reports of ghosted workers in `rqinfo`. Since
0.3.6, new worker registrations automatically expire, and the worker will
make sure to refresh the registrations as long as it's alive.
This function will clean up any of such legacy ghosted workers.
"""
conn = conn if conn else get_current_connection()
for worker in Worker.all(connection=conn):
if conn.ttl(worker.key) == -1:
ttl = worker.worker_ttl
conn.expire(worker.key, ttl)
logger.info('Marked ghosted worker {0} to expire in {1} seconds.'.format(worker.name, ttl))
|
