Autoriser Ctrl-C à interrompre une extension Python C

Il existe une autre façon de résoudre ce problème si vous ne voulez pas que votre extension C (ou ctypes DLL) soit liée à Python, comme dans le cas où vous souhaitez créer une bibliothèque C avec des liaisons dans plusieurs langues, vous devez autoriser votre Extension C à exécuter pendant de longues périodes, et vous pouvez modifier l'extension C :

Inclure l'en-tête du signal dans l'extension C.

#include <signal.h>

Créez un typedef de gestionnaire de signal dans l'extension C.

typedef void (*sighandler_t)(int);

Ajoutez des gestionnaires de signaux dans l'extension C qui effectueront les actions nécessaires pour interrompre tout code en cours d'exécution long (définir un indicateur d'arrêt, etc.) et enregistrer les gestionnaires de signaux Python existants.

sighandler_t old_sig_int_handler = signal(SIGINT, your_sig_handler);
sighandler_t old_sig_term_handler = signal(SIGTERM, your_sig_handler);

Restaurez les gestionnaires de signaux existants chaque fois que l'extension C revient. Cette étape garantit que les gestionnaires de signaux Python sont réappliqués.

signal(SIGINT, old_sig_int_handler);
signal(SIGTERM, old_sig_term_handler);

Si le code de longue durée est interrompu (drapeau, etc.), rendez le contrôle à Python avec un code de retour indiquant le numéro du signal.

return SIGINT;

En Python, envoie le signal reçu dans l'extension C.

import os
import signal

status = c_extension.run()

if status in [signal.SIGINT, signal.SIGTERM]:
    os.kill(os.getpid(), status)

Python effectuera l'action que vous attendez, comme lever un KeyboardInterrupt pour SIGINT.

Cependant, Ctrl-C ne semble pas avoir d'effet

Ctrl-C dans le shell envoie SIGINT au groupe de processus de premier plan. python à la réception du signal définit un drapeau en code C. Si votre extension C s'exécute dans le thread principal, aucun gestionnaire de signal Python ne sera exécuté (et donc vous ne verrez pas KeyboardInterrupt exception sur Ctrl-C ) sauf si vous appelez le PyErr_CheckSignals() qui vérifie l'indicateur (cela signifie qu'il ne devrait pas vous ralentir) et exécute les gestionnaires de signaux Python si nécessaire ou si votre simulation permet au code Python de s'exécuter (par exemple, si la simulation utilise des rappels Python). Voici un exemple de code d'un module d'extension pour CPython créé à l'aide de pybind11 suggéré par @Matt :

PYBIND11_MODULE(example, m)
{
    m.def("long running_func", []()
    {
        for (;;) {
            if (PyErr_CheckSignals() != 0)
                throw py::error_already_set();
            // Long running iteration
        }
    });
}

Si l'extension s'exécute dans un thread d'arrière-plan, il suffit de libérer GIL (pour permettre au code Python de s'exécuter dans le thread principal qui permet aux gestionnaires de signaux de s'exécuter). PyErr_CheckSignals() renvoie toujours 0 dans un fil d'arrière-plan.

Connexe :Cython, Python et KeybordInterrupt pris en compte

Python a un gestionnaire de signal installé sur SIGINT qui définit simplement un indicateur qui est vérifié par la boucle d'interprétation principale. Pour que ce gestionnaire fonctionne correctement, l'interpréteur Python doit exécuter du code Python.

Plusieurs options s'offrent à vous :

Utilisez Py_BEGIN_ALLOW_THREADS /Py_END_ALLOW_THREADS pour libérer le GIL autour de votre code d'extension C. Vous ne pouvez pas utiliser de fonctions Python lorsque vous ne détenez pas le GIL, mais le code Python (et d'autres codes C) peut s'exécuter en même temps que votre thread C (véritable multithreading). Un thread Python distinct peut s'exécuter parallèlement à l'extension C et intercepter les signaux Ctrl+C.
Configurez votre propre SIGINT gestionnaire et appelez le gestionnaire de signal d'origine (Python). Votre SIGINT le gestionnaire peut alors faire tout ce qu'il doit faire pour annuler le code d'extension C et rendre le contrôle à l'interpréteur Python.

Pas élégant, mais la seule approche que j'ai trouvée qui interrompt également les appels de bibliothèques externes en C++ et tue tous les processus enfants en cours d'exécution.

#include <csignal>
#include <pybind11/pybind11.h>

void catch_signals() {
  auto handler = [](int code) { throw std::runtime_error("SIGNAL " + std::to_string(code)); };
  signal(SIGINT, handler);
  signal(SIGTERM, handler);
  signal(SIGKILL, handler);
}

PYBIND11_MODULE(example, m)
{
    m.def("some_func", []()
    {
        catch_signals();
        // ...
    });
}

import sys

from example import some_func

try:
    some_func()
except RuntimeError as e:
    if "SIGNAL" in str(e):
        code = int(str(e).rsplit(" ", 1)[1])
        sys.exit(128 + code)
    raise