C - Как использовать оба aio_read() и aio_write()

Question

Я реализую игровой сервер, где мне нужно читать и писать. Поэтому я принимаю входящее соединение и начинаю читать с него, используя aio_read(), но когда мне нужно что-то отправить, я перестаю читать, используя aio_cancel(), а затем используйте aio_write(). В обратном вызове записи я возобновляю чтение. Итак, я все время читаю, но когда мне нужно что-то отправить - я пауза чтение.

Он работает на ~ 20% времени - в другом случае вызова aio_cancel() терпит неудачу с «Операция в настоящее время в процессе» - и я не могу отменить его (даже в пределах постоянного в то время как цикла). Таким образом, моя добавленная операция записи никогда не происходит.

Как хорошо использовать эти функции? Что я пропустил?

EDIT: Используется под Linux 2.6.35. Ubuntu 10 - 32 бит.

Пример кода:

void handle_read(union sigval sigev_value) { /* handle data or disconnection */ } 
void handle_write(union sigval sigev_value) { /* free writing buffer memory */ } 
void start() 
{ 
    const int acceptorSocket = socket(AF_INET, SOCK_STREAM, 0); 
    struct sockaddr_in addr; 
memset(&addr, 0, sizeof(struct sockaddr_in)); 
addr.sin_family = AF_INET; 
addr.sin_addr.s_addr = INADDR_ANY; 
addr.sin_port = htons(port); 
    bind(acceptorSocket, (struct sockaddr*)&addr, sizeof(struct sockaddr_in)); 

    listen(acceptorSocket, SOMAXCONN); 

    struct sockaddr_in address; 
socklen_t addressLen = sizeof(struct sockaddr_in); 

    for(;;) 
    { 
     const int incomingSocket = accept(acceptorSocket, (struct sockaddr*)&address, &addressLen); 
     if(incomingSocket == -1) 
     { /* handle error ... */} 
     else 
     { 
       //say socket to append outcoming messages at writing: 
       const int currentFlags = fcntl(incomingSocket, F_GETFL, 0); 
       if(currentFlags < 0) { /* handle error ... */ } 
       if(fcntl(incomingSocket, F_SETFL, currentFlags | O_APPEND) == -1) { /* handle another error ... */ } 

       //start reading: 
       struct aiocb* readingAiocb = new struct aiocb; 
       memset(readingAiocb, 0, sizeof(struct aiocb)); 
       readingAiocb->aio_nbytes = MY_SOME_BUFFER_SIZE; 
       readingAiocb->aio_fildes = socketDesc; 
       readingAiocb->aio_buf = mySomeReadBuffer; 
       readingAiocb->aio_sigevent.sigev_notify = SIGEV_THREAD; 
       readingAiocb->aio_sigevent.sigev_value.sival_ptr = (void*)mySomeData; 
       readingAiocb->aio_sigevent.sigev_notify_function = handle_read; 
       if(aio_read(readingAiocb) != 0) { /* handle error ... */ } 
      } 
    } 
} 

//called at any time from server side: 
send(void* data, const size_t dataLength) 
{ 
    //... some thread-safety precautions not needed here ... 

    const int cancellingResult = aio_cancel(socketDesc, readingAiocb); 
    if(cancellingResult != AIO_CANCELED) 
    { 
     //this one happens ~80% of the time - embracing previous call to permanent while cycle does not help: 
     if(cancellingResult == AIO_NOTCANCELED) 
     { 
      puts(strerror(aio_return(readingAiocb))); // "Operation now in progress" 
      /* don't know what to do... */ 
     } 
    } 
    //otherwise it's okay to send: 
    else 
    { 
     aio_write(...); 
    } 
} 

Answer 1

Если вы хотите иметь отдельные очереди AIO для чтения и записи, так что выданная позже запись может выполняться до чтения, выпущенного ранее, тогда вы можете использовать dup() для создания дубликата сокета и использовать его для выдачи прочтений а другой - для записи.

Тем не менее, я вторые рекомендации избегать AIO полностью и просто использовать цикл событий epoll() -driven с неблокирующими сокетами. Было показано, что этот метод масштабируется до большого числа клиентов - если вы получаете высокий уровень использования ЦП, проецируете его и выясняете, где это происходит, потому что есть вероятность, что это не ваш цикл событий, это преступник.

Answer 2

Прежде всего, считают демпинг AIO. Есть много других способов делать асинхронные операции ввода-вывода, которые не являются braindead (да, aio - breaindead). Множество альтернатив; если вы используете Linux, вы можете использовать libaio (io_submit и друзей). aio(7) упоминает это.

Назад к вопросу.
Я не использовал aio в течение длительного времени, но вот что я помню. aio_read и aio_write оба поставлены запросы (aiocb) в некоторой очереди. Они немедленно возвращаются, даже если запросы будут завершены через некоторое время. Вполне возможно поставить в очередь несколько запросов, не заботясь о том, что случилось с более ранними. Итак, в двух словах: прекратить отмену запросов на чтение и продолжать добавлять их.

/* populate read_aiocb */ 
rc = aio_read(&read_aiocb); 

/* time passes ... */ 
/* populate write_aiocb */ 
rc = aio_write(&write_aiocb) 

Позже вы свободны ждать используя aio_suspend, опрос с помощью aio_error, ждать сигналов и т.д.

я вижу, вы упоминаете epoll в свой комментарий. Вы должны обязательно пойти за libaio.

Answer 3

Не должно быть оснований останавливать или отменять запрос на чтение или запись в формате aio только потому, что вам нужно сделать другое чтение или запись. Если бы это было так, это бы повредило всю точку асинхронного чтения и записи, поскольку основная цель - позволить вам настроить операцию чтения или записи, а затем двигаться дальше. Поскольку несколько запросов могут быть поставлены в очередь, было бы намного лучше настроить пару асинхронных пулов чтения/записи, где вы можете захватить набор предварительно инициализированных структур aiocb из «доступного» пула, которые были настроены для асинхронных операций всякий раз, когда вам нужно их, а затем вернуть их в другой «завершенный» пул, когда они будут сделаны, и вы сможете получить доступ к буферам, на которые они указывают. Хотя они находятся в середине асинхронного чтения или записи, они будут находиться в «занятом» пуле и не будут затронуты. Таким образом вам не придется постоянно создавать структуры aiocb на куче динамически каждый раз, когда вам нужно сделать операцию чтения или записи, хотя это нормально делать ... это просто не очень эффективно, если вы никогда не планируете переходить через определенный предел, или планируют иметь только определенное количество запросов «в полете».

Кстати, обратите внимание на пару различных асинхронных запросов в полете, которые ваш асинхронный обработчик чтения/записи может быть прерван другим событием чтения/записи. Таким образом, вы действительно не хотите делать много дел со своим обработчиком. В описанном выше сценарии, который я описал, ваш обработчик в основном переместил бы структуру aiocb, которая активировала обработчик сигнала из одного из пулов в следующий на перечисленных «доступных» -> «занятых» -> «завершенных» этапах. Ваш основной код после чтения из буфера, на который указывают структуры aiocb в «готовом» пуле, затем переместит структуру обратно в «доступный» пул.

Answer 4

Если я не ошибаюсь, POSIX AIO (то есть aio_read(), aio_write() и т. Д.) Гарантированно работает только на дескрипторы файла поиска. Из aio_read(): страница руководства

The data is read starting at the absolute file offset aiocbp->aio_offset, regardless of the 
    current file position. After this request, the value of the current file position is unspeci‐ 
    fied. 

Для устройств, которые не имеют ассоциированную позицию файла, такие как сетевые розетки, AFAICS, POSIX AIO не определен. Возможно, это срабатывает для вашей текущей настройки, но это кажется более случайным, чем по дизайну.

Кроме того, в Linux POSIX AIO реализован в glibc с помощью потоков пользовательского пространства.

То есть, где возможно использование неблокирующий IO и epoll(). Однако epoll() не работает для файловых дескрипторов, доступных для поиска, таких как обычные файлы (то же самое относится и к классическому select()/poll()); в этом случае POSIX AIO является альтернативой для развертывания собственного пула потоков.