Comment implémenter une boucle d'événement de base ?

Question

Si vous avez travaillé avec des boîtes à outils graphiques, vous savez qu'il existe une boucle d'événements/une boucle principale qui doit être exécutée une fois que tout est terminé, et qui maintient l'application en vie et réagit aux différents événements. Par exemple, pour Qt, vous feriez ceci dans main() :

int main() {
    QApplication app(argc, argv);
    // init code
    return app.exec();
}

Dans ce cas, app.exec() est la boucle principale de l'application.

La façon évidente d'implémenter ce type de boucle serait :

void exec() {
    while (1) {
        process_events(); // create a thread for each new event (possibly?)
    }
}

Mais cela plafonne le CPU à 100% et est pratiquement inutile. Maintenant, comment puis-je implémenter une telle boucle d'événements qui soit réactive sans manger complètement le CPU ?

Les réponses sont appréciées en Python et/ou C++. Merci.

Note de bas de page : Pour des raisons d'apprentissage, je vais implémenter mes propres signaux/slots, et je les utiliserai pour générer des événements personnalisés (par ex. go_forward_event(steps) ). Mais si vous savez comment je peux utiliser les événements système manuellement, j'aimerais aussi le savoir.

Answer 1

Je me suis souvent posé la même question !

La boucle principale d'une interface graphique ressemble à ceci, en pseudo-code :

void App::exec() {
    for(;;) {
        vector<Waitable> waitables;
        waitables.push_back(m_networkSocket);
        waitables.push_back(m_xConnection);
        waitables.push_back(m_globalTimer);
        Waitable* whatHappened = System::waitOnAll(waitables);
        switch(whatHappened) {
            case &m_networkSocket: readAndDispatchNetworkEvent(); break;
            case &m_xConnection: readAndDispatchGuiEvent(); break;
            case &m_globalTimer: readAndDispatchTimerEvent(); break;
        }
    }
}

Qu'est-ce qu'un "Waitable" ? Eh bien, cela dépend du système. Sous UNIX, il s'agit d'un "descripteur de fichier" et "waitOnAll" est l'appel système ::select. Le soi-disant vector<Waitable> est un ::fd_set sur UNIX, et "whatHappened" est en fait interrogé par l'intermédiaire de FD_ISSET . Les waitable-handles réels sont acquis de diverses manières, par exemple m_xConnection peut être tiré de ::XConnectionNumber(). X11 fournit également une API portable de haut niveau pour cela -- ::XNextEvent() -- mais si vous l'utilisiez, vous ne seriez pas en mesure d'attendre plusieurs sources d'événements. simultanément .

Comment fonctionne le blocage ? "waitOnAll" est un appel système qui demande au système d'exploitation de mettre votre processus sur une "liste de sommeil". Cela signifie que vous ne recevez pas de temps CPU jusqu'à ce qu'un événement se produise sur l'un des waitables. Cela signifie donc que votre processus est inactif, consommant 0 % de CPU. Lorsqu'un événement se produit, votre processus y réagit brièvement, puis retourne à l'état d'inactivité. Les applications GUI passent presque tous leur temps d'inactivité.

Qu'arrive-t-il à tous les cycles du CPU pendant que vous dormez ? Ça dépend. Parfois, un autre processus en aura l'utilité. Si ce n'est pas le cas, votre système d'exploitation fera tourner le CPU en boucle, ou le mettra en mode basse consommation temporaire, etc.

N'hésitez pas à demander plus de détails !

Answer 2

Python :

Vous pouvez examiner l'implémentation de la Réacteur tordu qui est probablement la meilleure implémentation d'une boucle d'événement en python. Les réacteurs dans Twisted sont des implémentations d'une interface et vous pouvez spécifier un type de réacteur à exécuter : select, epoll, kqueue (tous basés sur une api c utilisant ces appels système), il y a aussi des réacteurs basés sur les boîtes à outils QT et GTK.

Une mise en œuvre simple consisterait à utiliser select :

#echo server that accepts multiple client connections without forking threads

import select
import socket
import sys

host = ''
port = 50000
backlog = 5
size = 1024
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind((host,port))
server.listen(backlog)
input = [server,sys.stdin]
running = 1

#the eventloop running
while running:
    inputready,outputready,exceptready = select.select(input,[],[])

    for s in inputready:

        if s == server:
            # handle the server socket
            client, address = server.accept()
            input.append(client)

        elif s == sys.stdin:
            # handle standard input
            junk = sys.stdin.readline()
            running = 0

        else:
            # handle all other sockets
            data = s.recv(size)
            if data:
                s.send(data)
            else:
                s.close()
                input.remove(s)
server.close()

Answer 3

En général, je fais ça avec une sorte de sémaphore de comptage :

1. Le sémaphore commence à zéro.
2. La boucle d'événement attend sur le sémaphore.
3. Un ou plusieurs événements arrivent, le sémaphore est incrémenté.
4. Le gestionnaire d'événement débloque et décrémente le sémaphore et traite l'événement.
5. Lorsque tous les événements sont traités, le sémaphore est remis à zéro et la boucle d'événements se bloque à nouveau.

Si vous ne voulez pas être aussi compliqué, vous pouvez simplement ajouter un appel à sleep() dans votre boucle while avec un temps de sommeil trivialement petit. Ainsi, le thread de traitement des messages cédera son temps CPU à d'autres threads. Le CPU ne sera plus bloqué à 100%, mais c'est toujours un gaspillage.

Answer 4

J'utiliserais une bibliothèque de messagerie simple et légère appelée ZeroMQ ( http://www.zeromq.org/ ). Il s'agit d'une bibliothèque open source (LGPL). Il s'agit d'une très petite bibliothèque ; sur mon serveur, l'ensemble du projet se compile en 60 secondes environ.

ZeroMQ simplifiera énormément votre code orienté événements, ET c'est aussi LA solution la plus efficace en termes de performances. La communication entre les threads en utilisant ZeroMQ est beaucoup plus rapide (en termes de vitesse) que l'utilisation de sémaphores ou de sockets UNIX locaux. ZeroMQ est également une solution 100% portable, alors que toutes les autres solutions lient votre code à un système d'exploitation spécifique.

Answer 5

Voici une boucle d'événement C++. A la création de l'objet EventLoop il crée un thread qui exécute continuellement toute tâche qui lui est confiée. S'il n'y a pas de tâches disponibles, le thread principal se met en veille jusqu'à ce qu'une tâche soit ajoutée.

Tout d'abord, nous avons besoin d'une file d'attente sécurisée par des threads qui permet à plusieurs producteurs et à au moins un consommateur unique (l' EventLoop fil). Le site EventLoop L'objet contrôle les consommateurs et les producteurs. Avec une petite modification, il est possible d'ajouter plusieurs consommateurs (fils d'exécution), au lieu d'un seul fil.

#include <stdio.h>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <iostream>
#include <set>
#include <functional>

#if defined( WIN32 )
    #include <windows.h>
#endif

class EventLoopNoElements : public std::runtime_error
{
public:
    EventLoopNoElements(const char* error)
        : std::runtime_error(error)
    {
    }
};

template <typename Type>
struct EventLoopCompare {
    typedef std::tuple<std::chrono::time_point<std::chrono::system_clock>, Type> TimePoint;

    bool operator()(const typename EventLoopCompare<Type>::TimePoint left, const typename EventLoopCompare<Type>::TimePoint right) {
        return std::get<0>(left) < std::get<0>(right);
    }
};

/**
 * You can enqueue any thing with this event loop. Just use lambda functions, future and promises!
 * With lambda `event.enqueue( 1000, [myvar, myfoo](){ myvar.something(myfoo); } )`
 * With futures we can get values from the event loop:
 * ```
 * std::promise<int> accumulate_promise;
 * event.enqueue( 2000, [&accumulate_promise](){ accumulate_promise.set_value(10); } );
 * std::future<int> accumulate_future = accumulate_promise.get_future();
 * accumulate_future.wait(); // It is not necessary to call wait, except for syncing the output.
 * std::cout << "result=" << std::flush << accumulate_future.get() << std::endl;
 * ```
 * It is just not a nice ideia to add something which hang the whole event loop queue.
 */
template <class Type>
struct EventLoop {
    typedef std::multiset<
        typename EventLoopCompare<Type>::TimePoint,
        EventLoopCompare<Type>
    > EventLoopQueue;

    bool _shutdown;
    bool _free_shutdown;

    std::mutex _mutex;
    std::condition_variable _condition_variable;
    EventLoopQueue _queue;
    std::thread _runner;

    // free_shutdown - if true, run all events on the queue before exiting
    EventLoop(bool free_shutdown)
        : _shutdown(false),
        _free_shutdown(free_shutdown),
        _runner( &EventLoop<Type>::_event_loop, this )
    {
    }

    virtual ~EventLoop() {
        std::unique_lock<std::mutex> dequeuelock(_mutex);
        _shutdown = true;
        _condition_variable.notify_all();
        dequeuelock.unlock();

        if (_runner.joinable()) {
            _runner.join();
        }
    }

    // Mutex and condition variables are not movable and there is no need for smart pointers yet
    EventLoop(const EventLoop&) = delete;
    EventLoop& operator =(const EventLoop&) = delete;
    EventLoop(const EventLoop&&) = delete;
    EventLoop& operator =(const EventLoop&&) = delete;

    // To allow multiple threads to consume data, just add a mutex here and create multiple threads on the constructor
    void _event_loop() {
        while ( true ) {
            try {
                Type call = dequeue();
                call();
            }
            catch (EventLoopNoElements&) {
                return;
            }
            catch (std::exception& error) {
                std::cerr << "Unexpected exception on EventLoop dequeue running: '" << error.what() << "'" << std::endl;
            }
            catch (...) {
                std::cerr << "Unexpected exception on EventLoop dequeue running." << std::endl;
            }
        }
        std::cerr << "The main EventLoop dequeue stopped running unexpectedly!" << std::endl;
    }

    // Add an element to the queue
    void enqueue(int timeout, Type element) {
        std::chrono::time_point<std::chrono::system_clock> timenow = std::chrono::system_clock::now();
        std::chrono::time_point<std::chrono::system_clock> newtime = timenow + std::chrono::milliseconds(timeout);

        std::unique_lock<std::mutex> dequeuelock(_mutex);
        _queue.insert(std::make_tuple(newtime, element));
        _condition_variable.notify_one();
    }

    // Blocks until getting the first-element or throw EventLoopNoElements if it is shutting down
    // Throws EventLoopNoElements when it is shutting down and there are not more elements
    Type dequeue() {
        typename EventLoopQueue::iterator queuebegin;
        typename EventLoopQueue::iterator queueend;
        std::chrono::time_point<std::chrono::system_clock> sleeptime;

        // _mutex prevents multiple consumers from getting the same item or from missing the wake up
        std::unique_lock<std::mutex> dequeuelock(_mutex);
        do {
            queuebegin = _queue.begin();
            queueend = _queue.end();

            if ( queuebegin == queueend ) {
                if ( _shutdown ) {
                    throw EventLoopNoElements( "There are no more elements on the queue because it already shutdown." );
                }
                _condition_variable.wait( dequeuelock );
            }
            else {
                if ( _shutdown ) {
                    if (_free_shutdown) {
                        break;
                    }
                    else {
                        throw EventLoopNoElements( "The queue is shutting down." );
                    }
                }
                std::chrono::time_point<std::chrono::system_clock> timenow = std::chrono::system_clock::now();
                sleeptime = std::get<0>( *queuebegin );
                if ( sleeptime <= timenow ) {
                    break;
                }
                _condition_variable.wait_until( dequeuelock, sleeptime );
            }
        } while ( true );

        Type firstelement = std::get<1>( *queuebegin );
        _queue.erase( queuebegin );
        dequeuelock.unlock();
        return firstelement;
    }
};

Utilitaire permettant d'imprimer l'horodatage actuel :

std::string getTime() {
    char buffer[20];
#if defined( WIN32 )
    SYSTEMTIME wlocaltime;
    GetLocalTime(&wlocaltime);
    ::snprintf(buffer, sizeof buffer, "%02d:%02d:%02d.%03d ", wlocaltime.wHour, wlocaltime.wMinute, wlocaltime.wSecond, wlocaltime.wMilliseconds);
#else
    std::chrono::time_point< std::chrono::system_clock > now = std::chrono::system_clock::now();
    auto duration = now.time_since_epoch();
    auto hours = std::chrono::duration_cast< std::chrono::hours >( duration );
    duration -= hours;
    auto minutes = std::chrono::duration_cast< std::chrono::minutes >( duration );
    duration -= minutes;
    auto seconds = std::chrono::duration_cast< std::chrono::seconds >( duration );
    duration -= seconds;
    auto milliseconds = std::chrono::duration_cast< std::chrono::milliseconds >( duration );
    duration -= milliseconds;
    time_t theTime = time( NULL );
    struct tm* aTime = localtime( &theTime );
    ::snprintf(buffer, sizeof buffer, "%02d:%02d:%02d.%03ld ", aTime->tm_hour, aTime->tm_min, aTime->tm_sec, milliseconds.count());
#endif
    return buffer;
}

Exemple de programme utilisant ces derniers :

// g++ -o test -Wall -Wextra -ggdb -g3 -pthread test.cpp && gdb --args ./test
// valgrind --leak-check=full --show-leak-kinds=all --track-origins=yes --verbose ./test
// procdump -accepteula -ma -e -f "" -x c:\ myexe.exe
int main(int argc, char* argv[]) {
    std::cerr << getTime().c_str() << "Creating EventLoop" << std::endl;
    EventLoop<std::function<void()>>* eventloop = new EventLoop<std::function<void()>>(true);

    std::cerr << getTime().c_str() << "Adding event element" << std::endl;
    eventloop->enqueue( 3000, []{ std::cerr << getTime().c_str() << "Running task 3" << std::endl; } );
    eventloop->enqueue( 1000, []{ std::cerr << getTime().c_str() << "Running task 1" << std::endl; } );
    eventloop->enqueue( 2000, []{ std::cerr << getTime().c_str() << "Running task 2" << std::endl; } );

    std::this_thread::sleep_for( std::chrono::milliseconds(5000) );
    delete eventloop;
    std::cerr << getTime().c_str() << "Exiting after 10 seconds..." << std::endl;
    return 0;
}

Exemple de test de sortie :

02:08:28.960 Creating EventLoop
02:08:28.960 Adding event element
02:08:29.960 Running task 1
02:08:30.961 Running task 2
02:08:31.961 Running task 3
02:08:33.961 Exiting after 10 seconds...