Асинхронный ввод-вывод с именованными трубами WinAPI

Question

Хорошо, я задал несколько вопросов о разных аспектах попытки выполнить то, что я хочу сделать. На этот раз у меня большие проблемы, только чтение из именованного канала. Я думаю, что собрал достаточно информации, чтобы завершить проект, над которым я работаю, если я смогу правильно настроить его. Я включу весь соответствующий код ниже, но моя миссия такова: читайте вывод (непрерывно) из программы, которую я сделал , а не пишите и публикуйте ее в WinAPI. Поэтому моя проблема заключается в том, что я только что переключился с анонимных каналов на именованные каналы, и у меня возникают проблемы с их правильной настройкой, поэтому я могу получить информацию. У меня есть настройка рамки, основанная на примере из MSDN.

#define WAIT_TIME 2 // 2s 
#define INSTANCES 4 // Number of threads 
#define CONNECT_STATE 0 
#define READ_STATE 1 
#define WRITE_STATE 2 
#define WORLDRD 0 
#define WORLDWR 1 
#define WORLDINRD 2 
#define WORLDINWR 3 
#define BUFSIZE 0x1000 // Buffer size 4096 (in bytes) 
#define PIPE_TIMEOUT 0x1388 // Timeout 5000 (in ms) 

void Arc_Redirect::createProcesses() 
{ 
TCHAR programName[]=TEXT("EXEC_PROGRAM.exe"); 
PROCESS_INFORMATION pi; 
STARTUPINFO si; 
BOOL bSuccess = FALSE; 

ZeroMemory(hEvents,(sizeof(hEvents)*INSTANCES)); 
ZeroMemory(outStd,(sizeof(PIPE_HANDLES)*INSTANCES)); 

// Prep pipes 
for(int i=0;i<INSTANCES;i++) 
{ 
    hEvents[i] = ::CreateEvent(NULL, TRUE, FALSE, NULL); 

    if(hEvents[i] == NULL) 
     throw "Could not init program!"; 

    outStd[i].o1.hEvent = hEvents[i]; 

    outStd[i].hPipeInst = ::CreateNamedPipe(
     TEXT("\\\\.\\pipe\\arcworld"), PIPE_ACCESS_DUPLEX | FILE_FLAG_OVERLAPPED, PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT, 
     PIPE_UNLIMITED_INSTANCES, BUFSIZE*sizeof(TCHAR), BUFSIZE*sizeof(TCHAR), PIPE_TIMEOUT, NULL); 

    if(outStd[i].hPipeInst == INVALID_HANDLE_VALUE) 
     throw "Could not init program!"; 

    outStd[i].pendingIO = getState(outStd[i].hPipeInst,&outStd[i].o1); 

    outStd[i].dwState = outStd[i].pendingIO ? 
     CONNECT_STATE : READ_STATE; 
} 

// Set stuff up 
ZeroMemory(&pi, sizeof(PROCESS_INFORMATION)); 
ZeroMemory(&si, sizeof(STARTUPINFO)); 
si.cb = sizeof(STARTUPINFO); 
si.hStdError = outStd[WORLDRD].hPipeInst; 
si.hStdOutput = outStd[WORLDRD].hPipeInst; 
si.hStdInput = outStd[WORLDINWR].hPipeInst; 
si.dwFlags |= STARTF_USESHOWWINDOW|STARTF_USESTDHANDLES|FILE_FLAG_OVERLAPPED; 

    // Start our process with the si info 
CreateProcess(programName,NULL,NULL,NULL,FALSE,0,NULL,NULL,&si,&pi); 
} 

BOOL Arc_Redirect::getState(HANDLE hPipe, LPOVERLAPPED lpo) 
{ 
    BOOL connected, pendingIO = FALSE; 

    // Overlap connection for this pipe 
    connected = ::ConnectNamedPipe(hPipe,lpo); 

    if(connected) 
     throw "ConnectNamedPipe(); failed!"; 

    switch(GetLastError()) 
    { 
     case ERROR_IO_PENDING: 
      pendingIO = TRUE; 
     break; 
     case ERROR_PIPE_CONNECTED: 
      if(SetEvent(lpo->hEvent)) 
       break; 
     default: 
      throw "ConnectNamedPipe(); failed!"; 
     break; 
    } 

    return pendingIO; 
} 

outStd [СЛУЧАИ] определяется как PIPE_HANDLES (пользовательские структуры), которая находится ниже

typedef struct 
{ 
    HANDLE hPipeInst; 
    OVERLAPPED o1; 
    TCHAR chReq[BUFSIZE]; 
    TCHAR chReply[BUFSIZE]; 
    DWORD dwRead; 
    DWORD dwWritten; 
    DWORD dwState; 
    DWORD cbRet; 
    BOOL pendingIO; 
} PIPE_HANDLES, *LPSTDPIPE; 

Теперь отсюда, где я получаю немного потерял. Я не знаю, куда идти. Я попытался использовать цикл в примере MSDN, но он не работал должным образом для того, что я ищу. Мне нужно взять конец считывания и получить информацию (опять же, непрерывно), а также открыть конец записи, когда мне может понадобиться записать ее. У кого-нибудь есть идеи? Я пытаюсь сделать ReadFile(), как я бы сделал с анонимным каналом, но он, похоже, не работает одинаково.

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

Answer 1

У вас должна быть две структуры OVERLAPPED, одна для чтения и одна для записи. Также вам нужен один дескриптор события для каждого канала, когда вы хотите закрыть канал, и еще одно событие, когда вы хотите прервать все (и закрыть приложение). У вас может быть один объект WaitForMultipleObjects для каждой операции, в которой задействованы все труды, или отдельное чтение из двух потоков с одним WFMO. Я бы пошел только с одним потоком, потому что закрытие трубы тогда проще (в противном случае вам нужно иметь некоторый подсчет ссылок на дескриптор трубы и закрыть его только тогда, когда количество ссылок падает до нуля).

Когда вы получаете одно событие, обработайте его и попробуйте WFMO с 0 секундами на всех ручках, которые были в массиве после того, который вы только что обработали. Таким образом, ни одна труба не станет голодной. Когда 0 секунд WFMO истекает, повторите нормальный WFMO с начала.

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

Answer 2

В мире Linux одна программа может записывать именованный канал через вызовы write/fwrite, а другая программа может читать ее через read/fread().

Полный путь именованного канала должен использоваться в операциях чтения/записи.

Answer 3

Вы пробовали пропустить PIPE_NOWAIT вместо PIPE_WAIT в CreateNamedPipe позвонить? Это позволит ReadFile и WriteFile быть неблокирующими.

В качестве альтернативы, вы пробовали использовать async IO? Вы передаете флаг FILE_FLAG_OVERLAPPED, так что это должно сработать. Если вы пробовали, с какими проблемами вы столкнулись?