У меня есть обработчик сигналов, который выглядит хорошо работающим в любой части кода, кроме этого цикла:
while not interrupted and time_to_sleep > 0:
logging.info(f"Zzzz... interrupted={interrupted}, time_to_sleep={time_to_sleep}")
time_to_sleep_now = min(interval_to_check_signals, time_to_sleep)
time.sleep(time_to_sleep_now)
time_to_sleep = max(0, time_to_sleep - time_to_sleep_now)
Когда выполняется эта часть кода, я могу нажать CTRL
+ C
или CTRL
+ Z
, которые не работают. пример:
INFO:root:Zzzz... interrupted=False, time_to_sleep=55
INFO:root:Zzzz... interrupted=False, time_to_sleep=50
INFO:root:Zzzz... interrupted=False, time_to_sleep=45
^C^C^CINFO:root:Zzzz... interrupted=False, time_to_sleep=40
^CINFO:root:Zzzz... interrupted=False, time_to_sleep=35
^C^C^C^C^C^C^CINFO:root:Zzzz... interrupted=False, time_to_sleep=30
^ZINFO:root:Zzzz... interrupted=False, time_to_sleep=25
^Z^Z^Z^ZINFO:root:Zzzz... interrupted=False, time_to_sleep=20
INFO:root:Zzzz... interrupted=False, time_to_sleep=15
INFO:root:Zzzz... interrupted=False, time_to_sleep=10
Моя полная программа такая:
"""
xxx
"""
import logging
import os
import signal
import time
# Load Environment Variables
TIME_TO_SLEEP_BETWEEN_CHECK_QUEUE = int(os.getenv('TIME_TO_SLEEP_BETWEEN_CHECK_QUEUE', 5 * 60)) # time in seconds
logging.basicConfig(level=logging.INFO) # Use only for debug locally
interrupted = False
def signal_handler(signal: int, frame):
"""It's an Handler for when a system signal is made. (System Signal is like a CTRL+C command, or other exit order)
It'll flag a global variable that is to graceful exit.
For example, when kubernetes does not need some pod, it will send a signal to pod to close,
but because this service have a long sleep time, it's important to wake up sometimes only to check signals,
and die graceful.
:param signal: Signal code
:param frame:
:return:
"""
global interrupted
interrupted = True
logging.warning(f"Signal '{signal}' received, the program will stop")
def sleep_and_look_for_signals(time_to_sleep: int, interval_to_check_signals: int = 5) -> None:
"""Sleep some time, but while is sleeping, sometimes wakes up to check if is interrupted
:param time_to_sleep:
:param interval_to_check_signals:
:return:
"""
global interrupted
logging.info("Sleeping...")
while not interrupted and time_to_sleep > 0:
logging.info(f"Zzzz... interrupted={interrupted}, time_to_sleep={time_to_sleep}")
time_to_sleep_now = min(interval_to_check_signals, time_to_sleep)
time.sleep(time_to_sleep_now)
time_to_sleep = max(0, time_to_sleep - time_to_sleep_now)
logging.info("Wake up!")
def more_messages_in_queue_bla_bla_bla():
# This is an example
return False
if __name__ == "__main__":
# Set handler to signal
signal.signal(signal.SIGHUP, signal_handler) # CODE 1
signal.signal(signal.SIGINT, signal_handler) # CODE 2
signal.signal(signal.SIGTERM, signal_handler) # CODE 15
while True:
# Signal to exit handled was triggered?
if interrupted:
logging.warning("Graceful exit")
break
time_to_sleep_until_next_check = TIME_TO_SLEEP_BETWEEN_CHECK_QUEUE # Reset to default sleep time
# Do a lot of work here...
# Do a lot of work here...
# Do a lot of work here...
# Do a lot of work here...
if more_messages_in_queue_bla_bla_bla():
time_to_sleep_until_next_check = 0
# sleep
logging.info(f"Sleep {time_to_sleep_until_next_check} seconds...\n")
sleep_and_look_for_signals(time_to_sleep_until_next_check)