Добрый день. 
Уважаемые программисты, дайте совет, как лучше спроектировать такую вот систему...

Есть несколько (n) потоков. В каждом потоке происходит одно и то же, а именно... Примерно 1000 раз в секунду приходит сообщение определенного формата. Сообщения складываются в пакеты. Максимальное количество сообщений в пакете - 16. Сообщение содержит информацию о своем номере в пакете.
Есть еще один поток. Задача данного потока находить в каждом пакете всех потоков сообщения с одинаковыми номерами и производить с ними определенные действия. После чего сообщения уничтожаются. В каком-то из потоков (или в нескольких потоках) могут быть пропуски сообщений. Например в потоке k отутствует сообщение с номером 7. В таком случае никакие действия не производятся. Во всех остальных потоках сообщения с номером 7 уничтожаются. Т.е ищутся полные соответствия для всех потоков. Вполне возможно, что обработка будет выполняться дольше, чем прием, поэтому требуется хранить всю цепочку сообщений и периодически при достижении определенного размера цепочки, пропускать часть данных. 
Систему приема данных я решил сделать следующим образом...
Класс message
Его задача разделение сообщения на номер и данные.
Класс packet
Его задача формирование пакета, а именно списка объектов типа message.
Класс block
Его задача формирования блока данных для своего потока, а именно списка объектов типа packet

Как организовать систему обработки данных?
На рисунке я привел простой пример. В отдельном потоке есть объект некоторого класса. Вызвав его метод join, мы соединяем строки для определенного пакета, определенного сообщения каждого потока. Зеленая рамка показывает, что пакет с номером 1 найден и такие данные поступают на обработку. Красная линия показывает, что в потоке n был пропуск пакета 1, и поэтому данные остальных потоков для пакета 1 игнорируются (уничтожаются).

Спасибо

Присоединённый файл ( Кол-во скачиваний: 24 )
 pic.GIF 11,62 Kb

По потоками ты понимаешь stream или thread? Просто если thread, то твоя система смысла не имеет.

Правильно ли я понял:
1. собираем n пакетов по не более чем 16 сообщений из n источников
2. обрабатываем сообщения с одинаковыми номерами во всех пакетах, остальные удаляем.
3. повторяем п. 1

bsa, спасибо за ответ.



Да, все так и есть. Обработка должна вестись параллельно, закончеными пакетами. Т.е пакет заполняется сообщениями. Как только пакет наполнился, он добавляется в блок. Поток обработки следит за появлением в блоках новых пакетов. Если блок не пустой, то производится поиск сообщений с одинаковыми номерами, попутно их обрабатывая и удаляя непарные. В идеале в очереди (block) должно быть не более одного пакета.  

Что касается потоков, то для потока обработки данных (на рис. thread data processing) применяется TThread (C++ Builder), потоки данных - это TidUDPServer с включенным ThreadedEvent=true.
В любом случае объясните, пожалуйста, почему не имеет смысла.

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

OK. 
Количество потоков в примере я ограничик четырьмя. Все описание основывается на рисунке из первого поста 
Глобально переменные...


Далее на форму выложены компоненты TIdUDPServer в количестве равном количеству потоков, т.е в данном случае 4
Все обработчики OnUDPRead повешаны на следующий код...


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

Вот в том и вопрос как лучше организовать обрабатывающую структуру. Во первых, раз объекты глобальные (понимаю, что плохо, но не понимаю, как в моем случае сделать по другому). Соответственно требуется организовать совместный доступ из разных потоков. Предположим можно будет обойтись критическими секциями. При условии, что в блоке все пакеты также нумерованы, можно конечно вытаскивать из каждого блока по пакету. Затем в этих пакетах (из четырех блоков) искать сообщения с одинаковыми номерами. Но как-то через то место все получается.
В целом я описал структуру довольно подробно. Как бы вы спроектировали архитектуру такой программы. Смотрел в сторону паттернов проектирования. Ничего подходящего не нашел.

Это сообщение отредактировал(а) Бегемот - 3.9.2014, 15:18

А может быть организавать что-то типа таблицы (см. рис). Каждый поток будет иметь ней свой столбец. Номер сообщения - это номер строки. Таким образом приходит сообщение, например, с номером четыре из первого потока, в таблицу в соответствующую ячейку записывается указатель на данные, инкрементируется счетчик заполненых ячеек для строки 4 и вызывается метод проверки, заполнена ли строка полностью. Если заполнена, то имея указатели на все сообщения производим обработку. Удаляем из таблицы строку. В таблице будут копиться записи, где в каком-то (каких-то) из столбцов отсутствуют ссылки сообщения. Эти записи так же периодически чистятся. Плюс избавляемся от дополнительных прослоек "пакет" и "блок". Так же сам механизм обработки будет запускаться не в результате поиска, а в результате добавления в строку последнего значения. Как думаете?

Это сообщение отредактировал(а) Бегемот - 3.9.2014, 16:07

Присоединённый файл ( Кол-во скачиваний: 9 )
 pic2.GIF 12,01 Kb

Бегемот, этот вариант выглядит значительно лучше.

Мне в этом варианте не нравится вот что...
У меня будет один объект класса Table и к нему примерно 1000 раз в секунду каждый из потоков будет обращать. Как вы полагаете это будет работать?
Опять же объект глобальный. 
Как-то вот так...





Это сообщение отредактировал(а) Бегемот - 4.9.2014, 12:24

Бегемот, не обязательно делать глобальную таблицу. можно сделать только столбцы. В итоге обращений к каждому будет меньше, чем к таблице.
Как вариант, можно сделать доступ к ячейкам через свой мьютекс, тогда обращение к одной ячейке таблицы не будет мешать обращениям к другим.