code propre pour les E/S asynchrones

Question

Alors que les E/S asynchrones (descripteurs non bloquants avec select/poll/epoll/kqueue etc) ne sont pas les choses les plus documentées sur le web, il existe une poignée de bons exemples.

Toutefois, tous ces exemples, après avoir déterminé les poignées renvoyées par l'appel, ont simplement un stub 'do_some_io(fd)'. Ils n'expliquent pas vraiment comment mieux approcher les E/S asynchrones réelles dans une telle méthode.

Blocage des entrées-sorties est très simple et facile à lire le code. Non bloquant, l'IO asynchrone est, d'autre part, poilu et salissant.

Quelles sont les approches? Quels sont robustes et lisibles?

void do_some_io(int fd) { 
    switch(state) { 
    case STEP1: 
     ... async calls 
     if(io_would_block) 
      return; 
     state = STEP2; 
    case STEP2: 
     ... more async calls 
     if(io_would_block) 
      return; 
     state = STEP3; 
    case STEP3: 
     ... 
    } 
} 

ou peut-être (ab) en utilisant des GOTO calculés de GCC:

#define concatentate(x,y) x##y 
#define async_read_xx(var,bytes,line)  \ 
    concatentate(jmp,line):     \ 
    if(!do_async_read(bytes,&var)) {   \ 
     schedule(EPOLLIN);     \ 
     jmp_read = &&concatentate(jmp,line); \ 
     return;        \ 
} 

// macros for making async code read like sync code 
#define async_read(var,bytes) \ 
    async_read_xx(var,bytes,__LINE__) 

#define async_resume()   \ 
    if(jmp_read) {    \ 
     void* target = jmp_read; \ 
     jmp_read = NULL;   \ 
     goto *target;   \ 
    } 

void do_some_io() { 
    async_resume(); 
    async_read(something,sizeof(something)); 
    async_read(something_else,sizeof(something_else)); 
} 

Ou peut-être des exceptions C et une machine d'état, de sorte que les fonctions des travailleurs peut déclencher le bit abort/CV, ou peut-être un conduit de table machine d'état?

Ce n'est pas comment le faire fonctionner, c'est comment le rendre maintenable que je chasse!

Answer 1

Je suggère de jeter un oeil sur: http://www.kegel.com/c10k.html, en second lieu jeter un oeil sur les bibliothèques existantes comme libevent, Boost.Asio qui font déjà le travail et voir comment ils fonctionnent.

Le point est que l'approche peut être différente pour chaque type d'appel système:

• select est simple réacteur
• epoll ont déclenché à la fois par niveau ou qui nécessitent une approche différente
• IOCP est proactor exiger d'autres approche

Suggestion: utilisez une bonne bibliothèque existante comme Boost.Asio C++ ou libevent pour C.

EDIT: Voici comment gère ce ASIO

class connection { 
    boost::asio:ip::tcp::socket socket_; 
public: 
    void run() 
    { 
     // for variable length chunks 
     async_read_until(socket_,resizable_buffer,'\n', 
       boost::bind(&run::on_line_recieved,this,errorplacehplder); 
     // or constant length chunks 
     async_read(socket_,buffer(some_buf,buf_size), 
       boost::bind(&run::on_line_recieved,this,errorplacehplder); 
    } 
    void on_line_recieved(error e) 
    { 
     // handle it 
     run(); 
    } 

}; 

Parce que fonctionne ASIO comme proactor il vous avertit lorsque l'opération est terminée et gère EWOULDBLOCK en interne.

Si vous mot comme réacteur vous pouvez simuler ce comportement:

class conn { 
    // Application logic 

    void run() { 
     read_chunk(&conn::on_chunk_read,size); 
    } 
    void on_chunk_read() { 
     /* do something;*/ 
    } 

    // Proactor wrappers 

    void read_chunk(void (conn::*callback),int size, int start_point=0) { 
     read(socket,buffer+start,size) 
     if(complete) 
      (this->*callback() 
     else { 
      this -> tmp_size-=size-read; 
      this -> tmp_start=start+read; 
      this -> tmp_callback=callback 
      your_event_library_register_op_on_readable(callback,socket,this); 
     } 
    } 
    void callback() 
    { 
     read_chunk(tmp_callback,tmp_size,tmp_start); 
    } 
} 

Quelque chose comme ça.

Answer 2

Les machines d'état sont une bonne approche. C'est un peu de complexité à l'avant qui vous sauvera des maux de tête dans le futur, où le futur commence vraiment, très bientôt. ;-)

Une autre méthode consiste à utiliser des threads et à bloquer les E/S sur un seul fd dans chaque thread. Le compromis ici est que vous faites E/S simple, mais peut introduire complexité en synchronisation.

Answer 3

Vous voulez découpler "io" du traitement, à quel point le code que vous lisez deviendra très lisible.Fondamentalement, vous avez:

    int read_io_event(...) { /* triggers when we get a read event from epoll/poll/whatever */ 

    /* read data from "fd" into a vstr/buffer/whatever */ 

    if (/* read failed */) /* return failure code to event callback */ ; 

    if (/* "message" received */) return process_io_event(); 

    if (/* we've read "too much" */) /* return failure code to event callback */ ; 

    return /* keep going code for event callback */ ; 
    } 


    int process_io_event(...) { 
     /* this is where you process the HTTP request/whatever */ 
    } 

... alors le code réel est en cas de processus, et même si vous avez plusieurs demandes réponses, il est assez facile à lire, vous faites juste « retour read_io_event() » après avoir réglé un état ou tout .

Answer 4

Un grand motif de conception "coroutine" existe pour résoudre ce problème.

C'est le meilleur des deux mondes: du code ordonné, exactement comme le flux synchrone io et de grandes performances sans changement de contexte, comme l'indique async io. Coroutine regarde à l'intérieur comme un thread synchrone odinaire, avec un seul pointeur d'instruction. Mais de nombreuses coroutines peuvent fonctionner dans un seul thread OS (ce que l'on appelle le "multitâche coopératif").

Exemple Code coroutine:

void do_some_io() { 
    blocking_read(something,sizeof(something)); 
    blocking_read(something_else,sizeof(something_else)); 
    blocking_write(something,sizeof(something)); 
} 

On dirait que le code synchrone, mais dans le flux de contrôle de fait d'utiliser une autre façon, comme ceci:

void do_some_io() { 
    // return control to network io scheduler, to handle another coroutine 
    blocking_read(something,sizeof(something)); 
    // when "something" is read, scheduler fill given buffer and resume this coroutine 

    // return control to network io scheduler, to handle another coroutine 
    CoroSleep(1000); 
    // scheduler create async timer and when it fires, scheduler pass control to this coroutine 
    ... 
    // and so on 

donc un seul contrôle de programmateur fileté plusieurs coroutines avec défini par l'utilisateur code et range les appels synchrones à io.

C++ coroutines exemple de mise en œuvre est « boost.coroutine » (en fait pas partie de boost :) http://www.crystalclearsoftware.com/soc/coroutine/ Cette bibliothèque implémente entièrement mécanique coroutine et peut utiliser boost.asio comme planificateur et async io couche.

Answer 5

Vous devez avoir une boucle principale qui fournit async_schedule(), async_foreach(), async_tick() etc. Ces fonctions placent les entrées dans une liste globale de méthodes qui seront exécutées lors de l'appel suivant à async_tick(). Ensuite, vous pouvez écrire du code beaucoup plus ordonné et n'incluant aucune instruction switch.

Vous pouvez simplement écrire:

async_schedule(callback, arg, timeout); 

Ou:

async_wait(condition, callback, arg, timeout); 

Ensuite, votre état peut même être réglé dans un autre thread (à condition que vous prenez soin de la sécurité de fil lors de l'accès à cette variable).

J'ai implémenté un framework async en C pour mon projet embarqué car je voulais avoir un multitâche non préemptif et async est parfait pour faire de nombreuses tâches en faisant un peu de travail à chaque itération de la boucle principale.

Le code est ici: https://github.com/mkschreder/fortmax-blocks/blob/master/common/kernel/async.c