Как организовать работу с буферизацией?

всем привет, у меня сейчас архитектурная проблема и хочу обратиться за советом к более опытным товарищам. В общем, я все работа. над организацией связи через pipe и столкнулся со следующей проблемой. Отправляя текст через свою программу, я могу получить тот же текст, но в виде трех сообщений. Например я отправляю "Привет", а получаю "П" "риве" "т". А мне хотелось бы получиться сообщение целиком. Я пришел к выводу, что надо делать некий накопительный буфер, и потом из него сообщения извлекать. Так вот, как архитектурно грамотно организовать такой буфер? Клиент по сути создает экземпляр класса pipe и ожидает, когда придет команда умереть через пайп, сам экземпляр pipe при инициализации создает поток слушатель, к которому повешен callback. Далее приходят данные, попадают в callback и должна произойти сборка какая то. Появилась мысль написать класс, который будет буферизировать в себя данные и по запросу отдавать полученные команды. Так вот приходят мне данные, я их скидываю в буфер, для этого есть идея написать класс. Далее, в другом потоке я вызываю из этого класса команду извлечь текущие команды и получаю несколько команд, которые потом обрабатываю. Вопрос отчасти связан с тем, что чтобы работать с одним экземпляром класса, надо его делать глобальным, или как то передать в класс работы с пайпом. Для разбора сообщений планирую сделать некие маркеры начала и конца сообщения. Как бы сделали это вы?

 

В классе пайпа создать коллбек или виртуальную функцию, которую он будет вызывать, когда соберет в своем буфере команду целиком.

 

Либо сделать отдельный метод setCallback, куда передавать функтор, когда сообщенице целиком было принято:

Код (C++):

1.  
2. class IPipeHandler
3. {
4. //...
5.     virtual void handle(void *data, size_t length) = 0;
6. };
7.  
8. class Pipe
9. {
10. // ...
11.     IPipeHandler *pHandler;
12.     uint8_t *pBuffer; // буфер, куда собираем
13.     size_t nBytes; // Количество байт в буфере
14. // ...
15.     void setCallback(IPipeHandler *handler)
16.    {
17.         pHandler = handler;
18.    }
19.  
20.    void process()
21.    {
22.         //...
23.         // Этот участок кода срабатывает, когда собрали сообщение в буфере
24.         if (pHandler != NULL)
25.             pHandler->handle(pBuffer, nBytes);
26.         // ...
27.     }
28.     // ...
29. };
30.  

 

SadKo,
Да-да, сеньор-девелоперы 300к/сек. используют паттерн Наблюдатель, а не коллбеки всякие...

 

rmn, SadKo, спасибо, то есть вы считаете что лучше что бы сам класс пайпа и собирал данные? просто что ему тогда считать маркерами начала и конца сообщения?

 

drem1lin,
Ну, обычно в потоковых протоколах первым передают размер всего сообщения, а затем данные указанного размера. Считываем слово (двойное слово), например 0x1234, затем выделяем буфер и читаем в него 0x1234 байт (возможно фрагментами). Как считали, заполненный буфер передаем в коллбек и затем освобождаем.

 

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

 

Блин, курите мануалы по сетевым протоколам. Протокол обмена по пайпу ничем не отличается от протокола обмена по сети.

 

Но его размер фиксирован и ты также можешь собрать его по частям.

Код (C):

1.  
2. BOOL ReadFileEx (HANDLE handle, LPVOID buffer, DWORD size)
3. {
4.     DWORD bytesRead;
5.  
6.     while (size != 0)
7.     {
8.         if (!ReadFile (handle, buffer, size, &bytesRead, NULL) || bytesRead == 0)
9.             return FALSE;
10.      
11.         buffer = (LPBYTE)buffer + bytesRead;
12.         size -= bytesRead;
13.     }
14.  
15.     return TRUE;
16. }
17.  
18. LPVOID ReadCommand (HANDLE handle, DWORD* size)
19. {
20.     DWORD dataSize;
21.     LPVOID data;
22.  
23.     if (ReadFileEx (handle, &dataSize, sizeof(DWORD))
24.     {
25.         data = malloc (dataSize);
26.         if (data)
27.         {
28.             if (ReadFileEx (handle, data, dataSize))
29.             {
30.                 *size = dataSize;
31.                 return data;
32.             }
33.          
34.             free (data);
35.         }
36.     }
37.  
38.     return NULL;
39. }
40.  

 

rmn, мы с вами не много не понимаем друг друга. У меня сообщение не известной длины, и сама длина этого сообщения может быть больше 255 байт, а как вы предлагаете, я вместо длинны получу только первый байт длинны, я выше писал, что может придти только первый байт вначале. И такое поведение не позволит мне вычитать корректно все остальное. Поэтому хочу сделать буферизацию

 

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

 

HTML:

1.  
2. если допустимо, то самое простое разделять сообщения - символом перевода строки
3. если нет - то элементы разметки (вплоть до xml или json).
4. Т.к при обычной работе с пайпом следующие символы не будут прочитаны, пока не
5. прочитаны все отправленные предыдущие, можно использовать только один
6. достаточно большой буфер.
7. Мною бы было сделано так: каждая итерация чтения пишет с начала буфера, не
8. более чем его длина. Считанное разбирается, и исходя из прочитанного,
9. считыватель переключает своё состояние (ДКА).
10. Например:
11. есть команды
12. exec <name> <a> <b>
13. kill <a>
14. И соответствующий им простой ДКА (на рис. - неполный):
15.      
16.       e     x     e     c
17. -> q0 -> q1 -> q2 -> q3 -> q4*
18.       |
19.       | k   i     l     l
20.       -->q5 -> q6 -> q7 -> q8*
21.        
22. Состояние в начале q0.
23. В буфер читается первые 8 символов. Если успешно прошли в состояние q4, то
24. пытаемся считать аргументы a, b. Не вышло - возврат в q0.
25. Полная диаграмма переходов:
26.    | e | x | c | k | i | l
27. ---+---+---+---+---+---+----
28. q0 |q1 |   |   |q5 |   |  
29. ---+---+---+---+---+---+----
30. q1 |   |q2 |   |   |   |
31. ---+---+---+---+---+---+----
32. q2 |q3 |   |   |   |   |
33. ---+---+---+---+---+---+----
34. q3 |   |   |q4*|   |   |
35. ---+---+---+---+---+---+----
36. q4*|   |   |   |   |   |
37. ---+---+---+---+---+---+----
38. q5 |   |   |   |   |q6 |
39. ---+---+---+---+---+---+----
40. q6 |   |   |   |   |   |q7
41. ---+---+---+---+---+---+----
42. q7 |   |   |   |   |   |q8*
43. ---+---+---+---+---+---+----
44. q8*|   |   |   |   |   |
45. ---+---+---+---+---+---+----
46. ; Где пустым - возврат в начальное q0 (не подставлено в таблицу)
47. И это же в виде регулярки: (exec) + (kill).  Можно и не посимвольно, и с
48. помощью strcmp в простых случаях.
49. При достижении состояния, определяющего команду (напр., q4), создаяется новый объект внутри
50. класса-считывателя из пайпа, имеющий идентификатор пришедшей команды, и
51. начинается заполнение ожидаемых считываемых аргументов. В случае успешного
52. считывания пакет команд, упакованный в объект, кладется в очередь или список, и
53. ожидает новых команд из пайпа или запроса на считывание команд.
54. ЗЫ: насколько мне изменяет память, в никсах нет пайпов, там локальные сокеты
55.  

 

Есть в никсах и пайпы, и сокеты, и SHM.

 

я протрассировал, и получил, что максимум он считает длину 255 и после этого он будет читать пока буфер не закончится, а он закончится на 255 и получается что слишком длинные сообщения не примутся

 

drem1lin,
Ты потрассировал неправильно, попробуй еще раз. Этот код может принимать сообщения длиной от 1 до 2³²-1 байт (перед данными идет дворд размера).

Функция ReadFileEx() не возвращает управление, пока не прочитает ровно size байт или не произойдет ошибка чтения. Так что первым вызовом она прочитает из хендла дворд длины, а вторым - данные указанной длины целиком. Даже если системная ReadFile() будет при каждом вызове возвращать только один байт.

 

Извиняюсь, вот тут реально ошибся, тогда вы конечно правы.

 

#define PIPE_MAX_BUFFER_SIZE 65536 - это в манах для МС край
создаете ивент и делайте в отдельном потоке - выделяете хип и всё, ждете завершения потока по WaitForSingleObject, закрываете хэндл буфер полный, FlushFileBuffers желательно тоже