Обработка SIGINT в Python: как защитить программу от Ctrl+C

import signal
import sys
import time

def signal_handler(sig, frame):
print('\nВы нажали Ctrl+C! Завершаю программу...')
sys.exit(0)

# Регистрируем обработчик сигнала
signal.signal(signal.SIGINT, signal_handler)

print('Нажмите Ctrl+C для выхода')

# Бесконечный цикл для демонстрации
while True:
time.sleep(1)

import signal
import time

def critical_operation():
# Блокируем SIGINT на время критической операции
mask = signal.SIG_BLOCK
signals = [signal.SIGINT]
with signal.pthread_sigmask(mask, signals):
print("Начало критической операции (SIGINT заблокирован)")
time.sleep(3) # Имитация длительной операции
print("Критическая операция завершена")

# Основная логика программы
try:
print("Программа запущена, нажмите Ctrl+C во время критической операции")
time.sleep(2)
critical_operation()
print("Программа продолжает работу")
while True:
time.sleep(1)
except KeyboardInterrupt:
print("\nПрограмма завершена пользователем")

import signal
import time

running = True

def signal_handler(sig, frame):
global running
print('\nПолучен сигнал на завершение, готовлюсь к выходу...')
running = False

signal.signal(signal.SIGINT, signal_handler)

print("Программа запущена. Нажмите Ctrl+C для корректного завершения")

# Основной цикл программы
while running:
# Выполнение задач программы
print("Работаю...")
time.sleep(1)

print("Программа корректно завершена")

import signal
import queue
import threading
import time

# Очередь для коммуникации между обработчиком сигнала и основным потоком
signal_queue = queue.Queue()

def signal_handler(sig, frame):
print("\nСигнал получен, планируем завершение...")
# Помещаем сигнал в очередь вместо немедленной обработки
signal_queue.put(sig)

# Регистрируем обработчик
signal.signal(signal.SIGINT, signal_handler)

def main_process():
try:
while True:
print("Выполняем основную работу...")

# Проверяем, есть ли сигнал в очереди (неблокирующий способ)
try:
sig = signal_queue.get_nowait()
print(f"Обрабатываем сигнал {sig} в основном потоке")
print("Выполняем корректное завершение...")
break
except queue.Empty:
# Если сигналов нет, продолжаем работу
pass

# Имитация полезной работы
time.sleep(1)

print("Программа завершена корректно")
except Exception as e:
print(f"Произошла ошибка: {e}")

main_process()

import signal
import time
import sys

interrupt_count = 0
max_interrupts = 3

def signal_handler(sig, frame):
global interrupt_count
interrupt_count += 1

if interrupt_count == 1:
print("\nНажмите Ctrl+C ещё раз для начала корректного завершения")
elif interrupt_count == 2:
print("\nИдёт корректное завершение... Нажмите Ctrl+C ещё раз для принудительного выхода")
# Здесь можно запустить процедуру корректного завершения
elif interrupt_count >= max_interrupts:
print("\nПринудительное завершение!")
sys.exit(1)

signal.signal(signal.SIGINT, signal_handler)

try:
print("Программа запущена. Нажмите Ctrl+C для взаимодействия")
while True:
time.sleep(0.5)
# Имитация сброса счетчика прерываний при длительном отсутствии Ctrl+C
if time.time() % 10 < 0.5 and interrupt_count > 0: 
print("Сброс счетчика прерываний")
interrupt_count = 0
except Exception as e:
print(f"Произошла ошибка: {e}")

import signal
import threading
import time
import sys

# Флаг для контроля завершения программы
shutting_down = False
# Флаг успешного завершения очистки
cleanup_completed = False

def cleanup_handler():
"""Симуляция длительного процесса очистки ресурсов"""
global cleanup_completed
print("Начинаю очистку ресурсов...")

# Имитация длительной операции
for i in range(5):
if shutting_down: # Проверка флага для ранней остановки
time.sleep(0.5)
print(f"Очистка: шаг {i+1}/5 завершен")

cleanup_completed = True
print("Очистка ресурсов завершена")

def force_exit():
"""Принудительное завершение, если очистка заняла слишком много времени"""
time.sleep(3) # Ждем 3 секунды
if not cleanup_completed:
print("\nТаймаут очистки ресурсов! Принудительное завершение.")
sys.exit(1)

def signal_handler(sig, frame):
global shutting_down

if shutting_down:
print("\nПолучен повторный сигнал прерывания. Принудительный выход.")
sys.exit(1)

print("\nПолучен сигнал прерывания. Начинаю корректное завершение.")
shutting_down = True

# Запускаем очистку ресурсов в отдельном потоке
cleanup_thread = threading.Thread(target=cleanup_handler)
cleanup_thread.start()

# Запускаем таймер для принудительного выхода
force_exit_thread = threading.Thread(target=force_exit)
force_exit_thread.daemon = True # Делаем поток демоном, чтобы он не блокировал выход
force_exit_thread.start()

# Ждем завершения очистки
cleanup_thread.join()

if cleanup_completed:
print("Программа успешно завершена")
sys.exit(0)

signal.signal(signal.SIGINT, signal_handler)

try:
print("Программа запущена. Нажмите Ctrl+C для завершения")
while not shutting_down:
# Основной цикл программы
time.sleep(0.5)
except Exception as e:
print(f"Произошла ошибка: {e}")
sys.exit(1)

import signal
import sys
import time
import contextlib

class SignalHandler:
def __init__(self):
self.interrupted = False
self.released = False
self.original_handler = None

def __enter__(self):
self.original_handler = signal.getsignal(signal.SIGINT)
signal.signal(signal.SIGINT, self.handler)
return self

def handler(self, sig, frame):
self.release()
self.interrupted = True
print("\nПрограмма получила сигнал на завершение")

def __exit__(self, type, value, traceback):
self.release()

def release(self):
if not self.released:
signal.signal(signal.SIGINT, self.original_handler)
self.released = True

@contextlib.contextmanager
def critical_section():
"""Защищенная от прерываний секция кода"""
original_handler = signal.getsignal(signal.SIGINT)
signal.signal(signal.SIGINT, signal.SIG_IGN)
try:
yield
finally:
signal.signal(signal.SIGINT, original_handler)

# Пример использования
try:
with SignalHandler() as h:
print("Программа запущена. Нажмите Ctrl+C для проверки обработки сигнала")

# Основной цикл
while not h.interrupted:
print("Выполняется обычная работа...")
time.sleep(1)

# Критическая секция, защищенная от прерываний
with critical_section():
print("Выполняется критическая операция (Ctrl+C заблокирован)...")
time.sleep(3)
print("Критическая операция завершена")

print("Выполняем контролируемое завершение...")
time.sleep(2)
print("Завершение выполнено успешно")

except Exception as e:
print(f"Произошла ошибка: {e}")

import signal
import sys
import time
import threading
import os
import atexit

class Application:
def __init__(self):
self.shutdown_event = threading.Event()
self.data_changed = False
self.data = []
self.data_file = "app_data.txt"
self.log_file = None
self.worker_threads = []

# Загрузка данных при запуске
self.load_data()

# Регистрируем обработчики завершения
signal.signal(signal.SIGINT, self.signal_handler)
atexit.register(self.cleanup)

def load_data(self):
"""Загрузка данных из файла"""
try:
if os.path.exists(self.data_file):
with open(self.data_file, 'r') as f:
self.data = [line.strip() for line in f.readlines()]
print(f"Загружено {len(self.data)} записей данных")
except Exception as e:
print(f"Ошибка при загрузке данных: {e}")

def save_data(self):
"""Сохранение данных в файл"""
if not self.data_changed:
print("Данные не изменялись, сохранение не требуется")
return

try:
with open(self.data_file, 'w') as f:
for item in self.data:
f.write(f"{item}\n")
print(f"Данные успешно сохранены в {self.data_file}")
self.data_changed = False
except Exception as e:
print(f"Ошибка при сохранении данных: {e}")

def start_logging(self):
"""Открытие лог-файла"""
try:
self.log_file = open("app.log", "a")
self.log("Приложение запущено")
except Exception as e:
print(f"Не удалось открыть лог-файл: {e}")

def log(self, message):
"""Запись в лог"""
if self.log_file and not self.log_file.closed:
timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
self.log_file.write(f"[{timestamp}] {message}\n")
self.log_file.flush()

def start_workers(self):
"""Запуск рабочих потоков"""
for i in range(3):
thread = threading.Thread(
target=self.worker_process, 
args=(i,), 
daemon=True
)
self.worker_threads.append(thread)
thread.start()
print(f"Запущен рабочий поток {i}")

def worker_process(self, worker_id):
"""Имитация рабочего процесса"""
while not self.shutdown_event.is_set():
# Выполняем какую-то работу
new_data = f"Данные от потока {worker_id}: {time.time()}"
self.data.append(new_data)
self.data_changed = True
self.log(f"Поток {worker_id} добавил данные")

# Имитация работы
for _ in range(10):
if self.shutdown_event.is_set():
break
time.sleep(0.5)

def signal_handler(self, sig, frame):
"""Обработчик сигнала SIGINT"""
print("\nПолучен сигнал на завершение. Начинаю корректное завершение...")
self.shutdown()

def shutdown(self):
"""Запуск процедуры корректного завершения"""
if self.shutdown_event.is_set():
print("Завершение уже выполняется")
return

# Уведомляем потоки о необходимости завершения
self.shutdown_event.set()
self.log("Начато корректное завершение приложения")

# Ждем завершения всех рабочих потоков
print("Ожидание завершения рабочих потоков...")
for i, thread in enumerate(self.worker_threads):
print(f"Ожидание завершения потока {i}...")
thread.join(timeout=2)
if thread.is_alive():
print(f"Поток {i} не завершился вовремя")

# Завершаем работу приложения
self.cleanup()

print("Приложение успешно завершено")
sys.exit(0)

def cleanup(self):
"""Очистка ресурсов перед выходом"""
# Сохраняем данные
print("Сохранение данных...")
self.save_data()

# Закрываем лог-файл
if self.log_file and not self.log_file.closed:
self.log("Приложение завершено")
print("Закрытие лог-файла...")
self.log_file.close()

def run(self):
"""Основной метод запуска приложения"""
self.start_logging()
self.start_workers()

try:
print("Приложение работает. Нажмите Ctrl+C для завершения")
while not self.shutdown_event.is_set():
time.sleep(1)
except Exception as e:
print(f"Произошла ошибка: {e}")
self.shutdown()

# Запуск приложения
if __name__ == "__main__":
app = Application()
app.run()

import signal
import threading
import time

# Создаем событие для сигнализации о завершении
shutdown_event = threading.Event()

def signal_handler(sig, frame):
print("\nПолучен SIGINT, уведомляем все потоки о необходимости завершения...")
shutdown_event.set() # Устанавливаем флаг события

def worker(worker_id):
try:
while not shutdown_event.is_set():
print(f"Рабочий поток {worker_id} выполняет задачу...")
# Проверяем событие каждую секунду
# Это позволяет потоку реагировать на сигнал завершения
shutdown_event.wait(timeout=1)
print(f"Рабочий поток {worker_id} получил сигнал завершения и корректно останавливается")
except Exception as e:
print(f"Ошибка в потоке {worker_id}: {e}")

# Регистрируем обработчик сигнала в главном потоке
signal.signal(signal.SIGINT, signal_handler)

# Запускаем несколько рабочих потоков
threads = []
for i in range(3):
t = threading.Thread(target=worker, args=(i,))
t.start()
threads.append(t)

# Ожидаем завершения всех потоков
try:
while any(t.is_alive() for t in threads):
time.sleep(0.1)
print("Все потоки завершены")
except Exception as e:
print(f"Ошибка в главном потоке: {e}")

import signal
import threading
import queue
import time

# Очередь для коммуникации между потоками
command_queue = queue.Queue()

def signal_handler(sig, frame):
print("\nПолучен SIGINT, отправляем команды завершения всем потокам")
# Отправляем команду завершения в очередь для каждого потока
for _ in range(len(threads)):
command_queue.put("SHUTDOWN")

def worker(worker_id):
try:
while True:
# Выполняем работу
print(f"Поток {worker_id} работает...")

# Проверяем наличие команд в очереди (неблокирующий способ)
try:
command = command_queue.get(block=False)
if command == "SHUTDOWN":
print(f"Поток {worker_id} получил команду завершения")
break
command_queue.task_done()
except queue.Empty:
# Если команд нет, продолжаем работу
pass

time.sleep(1)

print(f"Поток {worker_id} завершается корректно")
except Exception as e:
print(f"Ошибка в потоке {worker_id}: {e}")

# Регистрируем обработчик
signal.signal(signal.SIGINT, signal_handler)

# Создаем и запускаем потоки
threads = []
for i in range(3):
t = threading.Thread(target=worker, args=(i,))
t.daemon = True # Делаем поток демоном для автоматического завершения при выходе из программы
t.start()
threads.append(t)

# Главный цикл программы
try:
while any(t.is_alive() for t in threads):
time.sleep(0.1)
print("Все потоки завершены")
except Exception as e:
print(f"Ошибка в главном потоке: {e}")

import signal
import time
import concurrent.futures
import sys

# Флаг для отслеживания состояния завершения
is_shutting_down = False

def signal_handler(sig, frame):
global is_shutting_down
if is_shutting_down:
print("\nПринудительное завершение")
sys.exit(1)

print("\nПолучен SIGINT, начинаю корректное завершение...")
is_shutting_down = True

def task(task_id):
"""Задача, которая может быть прервана"""
for i in range(10):
# Проверяем глобальный флаг завершения
if is_shutting_down:
print(f"Задача {task_id} прервана на шаге {i}")
return f"Задача {task_id} завершена досрочно"

# Имитация работы
print(f"Задача {task_id}: шаг {i}")
time.sleep(1)

return f"Задача {task_id} завершена полностью"

# Регистрируем обработчик сигнала
signal.signal(signal.SIGINT, signal_handler)

# Используем пул потоков
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
# Запускаем задачи
future_to_task = {executor.submit(task, i): i for i in range(5)}

try:
# Получаем результаты по мере завершения задач
for future in concurrent.futures.as_completed(future_to_task):
task_id = future_to_task[future]
try:
result = future.result()
print(f"Результат: {result}")
except Exception as e:
print(f"Задача {task_id} вызвала исключение: {e}")

except KeyboardInterrupt:
# Этот блок может не выполниться, так как KeyboardInterrupt
# обрабатывается нашим обработчиком сигнала
pass

finally:
if is_shutting_down:
print("Отмена всех незавершенных задач...")
# Отменяем все незавершенные задачи
for future in future_to_task:
if not future.done():
future.cancel()

print("Ожидание завершения выполняющихся задач...")
# Даем немного времени для завершения текущих задач
executor.shutdown(wait=True, cancel_futures=True)

print("Программа завершена")

import signal
import threading
import time
import os

def thread_with_blocked_signals():
# Блокируем SIGINT в этом потоке
signal.pthread_sigmask(signal.SIG_BLOCK, [signal.SIGINT])
print(f"Поток {threading.current_thread().name} заблокировал SIGINT")

# Этот поток не будет прерван сигналом SIGINT
try:
while True:
print(f"Поток с блокировкой сигналов работает...")
time.sleep(1)
except Exception as e:
print(f"Исключение в защищенном потоке: {e}")

def thread_with_signals():
print(f"Поток {threading.current_thread().name} получает сигналы")

# Этот поток не будет получать сигналы напрямую,
# но может проверять флаги, установленные главным потоком
try:
while not shutdown_requested:
print(f"Поток с сигналами работает...")
time.sleep(1)
print("Поток с сигналами завершается по запросу")
except Exception as e:
print(f"Исключение в незащищенном потоке: {e}")

def signal_handler(sig, frame):
global shutdown_requested
print(f"\nПолучен сигнал {sig} в потоке {threading.current_thread().name}")
shutdown_requested = True
print("Запрошено завершение")

# Глобальный флаг для сигнализации о завершении
shutdown_requested = False

# Регистрируем обработчик в главном потоке
signal.signal(signal.SIGINT, signal_handler)

# Запускаем потоки
t1 = threading.Thread(target=thread_with_blocked_signals, daemon=True)
t2 = threading.Thread(target=thread_with_signals, daemon=True)

t1.start()
t2.start()

print(f"Главный поток (PID={os.getpid()}) ждет. Нажмите Ctrl+C для проверки обработки сигналов")
try:
while not shutdown_requested:
time.sleep(0.1)

print("Главный поток получил запрос на завершение")
print("Ожидание завершения потоков (или принудительное завершение через 5 секунд)")

# Устанавливаем таймаут для завершения потоков
t_end = time.time() + 5
while time.time() < t_end and (t1.is_alive() or t2.is_alive()):
time.sleep(0.1)

print("Программа завершена")
except Exception as e:
print(f"Исключение в главном потоке: {e}")