аккуратный код для асинхронного ввода-вывода

Question

Хотя асинхронный ввод-вывод (неблокирующие дескрипторы с select / poll / epoll / kqueue и т. Д.) - не самая документированная вещь в Интернете, есть несколько хороших примеров.

Однако во всех этих примерах, определив дескрипторы, возвращаемые при вызове, просто есть заглушка 'do_some_io(fd)'. Они действительно не объясняют, как наилучшим образом приблизиться к фактическому асинхронному вводу-выводу в таком методе.

Блокировка ввода-вывода очень удобна и проста для чтения кода. Неблокирующее асинхронное IO, с другой стороны, волосатое и грязное.

Какие есть подходы? Что является надежным и читаемым?

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:
       ...
  }
}

или, возможно, (ab), используя вычисленные goto 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));
}

Или, возможно, исключения C ++ и конечный автомат, чтобы рабочие функции могли запускать бит отмены / возобновления, или, возможно, конечный автомат, управляемый таблицей?

Дело не в том, чтобы заставить его работать, а в том, чтобы сделать его обслуживаемым, что я гоняюсь!

Answer 1

Я предлагаю взглянуть на: http://www.kegel.com/c10k.html, секунду, посмотрите на существующие библиотеки, такие как libevent, Boost.Asio, которые уже выполняют свою работу, и посмотрите, как они работают.

Дело в том, что подход может отличаться для каждого типа системного вызова:

• выберите простой реактор
• epoll имеет интерфейс, запускаемый по фронту или по уровню, который требует другого подхода
• iocp - это проактор, требующий другого подхода

Предложение: используйте хорошую существующую библиотеку, такую ​​как Boost.Asio для C ++ или libevent для C.

РЕДАКТИРОВАТЬ: Вот как 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();
   }

};

Поскольку ASIO работает как proactor, он уведомляет вас о завершении операции и обрабатывает EWOULDBLOCK внутри.

Если вы называете себя реактором, вы можете смоделировать это поведение:

 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);
    }
 }

Нечто подобное.

Answer 2

Конечные автоматы - один хороший подход. Немного сложнее, что избавит вас от головной боли в будущем, когда будущее начинается действительно, очень скоро. ; -)

Другой метод - использовать потоки и блокировать ввод / вывод по одному файлу в каждом потоке. Компромисс здесь в том, что вы делаете ввод-вывод простым, но может привнести сложность в синхронизацию.

Answer 3

Для решения этой проблемы существует великолепный шаблон проектирования "сопрограмма".

Это лучшее из обоих миров: аккуратный код, точно так же, как синхронный поток ввода-вывода и отличная производительность без переключения контекста, как дает асинхронный ввод-вывод. Сопрограмма выглядит внутри как обычный синхронный поток с указателем одной инструкции. Но многие сопрограммы могут работать в одном потоке ОС (так называемая «совместная многозадачность»).

Пример кода сопрограммы:

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

Выглядит как синхронный код, но на самом деле поток управления использует другой способ, например:

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 

Таким образом, однопоточный планировщик управляет многими сопрограммами с помощью пользовательского кода и аккуратных синхронных вызовов io.

Пример реализации сопрограмм C ++ - "boost.coroutine" (на самом деле не является частью boost :) http://www.crystalclearsoftware.com/soc/coroutine/ Эта библиотека полностью реализует механику сопрограмм и может использовать boost.asio в качестве планировщика и асинхронного слоя.

Answer 4

Вам необходим основной цикл, который обеспечивает async_schedule (), async_foreach (), async_tick () и т. Д. Эти функции, в свою очередь, помещают записи в глобальный список методов, которые будут выполняться при следующем вызове async_tick (). Затем вы можете написать код, который будет намного более аккуратным и не будет содержать операторов switch.

Вы можете просто написать:

async_schedule(callback, arg, timeout); 

Или:

async_wait(condition, callback, arg, timeout); 

Тогда ваше условие может быть даже установлено в другом потоке (при условии, что вы обращаетесь к безопасности потока при обращении к этой переменной).

Я реализовал асинхронную среду в C для моего встроенного проекта, потому что я хотел иметь многозадачность без вытеснения, и async идеально подходит для выполнения многих задач, выполняя небольшую работу во время каждой итерации основного цикла.

Код здесь: https://github.com/mkschreder/fortmax-blocks/blob/master/common/kernel/async.c

Answer 5

Вы хотите отделить «io» от обработки, после чего код, который вы читаете, станет очень читабельным. В основном у вас есть:

    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 */
    }

... тогда реальный код находится в процессе обработки, и даже если у вас есть несколько ответов на запросы, он довольно читабелен, вы просто делаете «return read_io_event ()» после установки состояния или чего-то еще.