Отправка UDP по разным адресам через один сокет

← →
Добежал (2009-01-29 16:03) [0]

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

Как сделано сейчас: все общение с сокетом синхронизировано. То есть, поток входит в синхронизирующую секцию, отправляет данные, ждет и принимает данные (используется функция Select), выходит из критической секции. Соответственно, на время общения одного потока другие потоки ждут.

Что нужно сделать: распараллелить работу. Ведь UDP позволяет отправлять в одно и тоже время по разным адресам пакеты.

Мои соображения: с отправкой проблем нету, каждый из потоков вызывает SendTo с нужными ему параметрами - не вопрос.
А вот с приемом возникают проблемы... Допустим, поток ставит Select на нужный сокет, событие срабатывает... Но ведь это может быть пришел ответ для другого потока. Причем поскольку каждый поток ставит Select, то все определяется тем кто из потоков раньше сделает recv. Притом, что полученные данные могут предназначаться не тому потоку, который раньше всех успел сделать recv. Способ откидывается.

Я придумал только ввод некоего обслуживающего сокет потока. Он постоянно выгребает данные, пришедшие сокету и складывает их в какой-то синхронизированный буфер.
Сетевые потоки тогда работают так: посылают запрос с помощью SendTo, а потом начинают мониторить синхронизированный буфер... Вопрос только в том, как они его будут мониторить?!
Создавать на каждый такой запрос данных свой объект ядра, служащий синхронизатором для ответа именно данному потоку? Как-то нерационально на мой взгляд. Выжидать константное время и проверять в буфере появился ли ответ? Тогда не экономно, так как ответ может появится гораздо раньше, излишнее ожидание.

Есть еще и прочие проблемы - допустим, поток по таймауту не дождался ответа, "ушел". А ответ все таки пришел и был помещен обслуживающим потоком в синхронизированный буфер. Так и будет этот ответ висеть получается? Ведь ожидающий эту информацию поток уже ушел...

Как обычно все организовывается в данных ситуациях? Есть ли замечания по моей логике?

P.S. Дополнительный вопрос - является ли вызов методов SendTo и recv потокобезопасными для socket"а?

← →
Добежал (2009-01-29 18:26) [1]

Удалено модератором

← →
Сергей М. © (2009-01-29 19:26) [2]

За каким лешим все это ?

Каждый поток вправе создать своё собственное гнездо и работать с ним не задумываясь о других потоках и прочей синхрохрени !

Это что, откровение для тебя ?)

← →
Сергей М. © (2009-01-29 19:27) [3]

Удалено модератором

← →
Piter © (2009-01-29 21:48) [4]

Сергей М. © (29.01.09 19:26) [2]
Каждый поток вправе создать своё собственное гнездо

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

← →
Сергей М. © (2009-01-29 22:00) [5]

> Piter © (29.01.09 21:48) [4]

> будут открываться различные рандомные локальные UDP-порты

И чего ?
Что в том плохого ?

> по условиям задачи это неприемлимо

Условия, imho, кривые изначально.

← →
Сергей М. © (2009-01-29 22:02) [6]

> поток ставит Select на нужный сокет, событие срабатывает

Какое нахрен событие ?
Где в вопросе хоть что-либо напоминающее Winsock-события ?
В топку.

← →
Piter © (2009-01-29 22:25) [7]

Удалено модератором

← →
Сергей М. © (2009-01-29 22:40) [8]

Удалено модератором

← →
Вариант (2009-01-30 06:28) [9]

> Добежал (29.01.09 16:03)

Я решаю для себя такую задачу обычно разбив ее на сущности.
Определим логические сущности
1 Транспорт
- отвечает за передачу данных из буфера TX в транспортную среду. Доступ к буферу в случае потоков должен быть синхронизирован. Варианты исполнения буфера - Например TThreadList и о добавлении стуктуры данных -это оповещение событием или сообщением.
Или передача указателя на структуру данных через сообщения. Или что-то другое.....
- отвечает за прием за прием данных в буфер RX из транспортной среды и оповещение Декодера(парсера) о наличии данных в буфере. Варианты реализации примерно такие же как и у буфера TX.

2 Кодер
Задача подготовить структуры данных к передаче в транспортной среде и положить их в TX буфер, оповестить об этом Транспорт

3 Декодер - получив оповещение от Транспорта, что есть данные для приема - забрать их, декодировать по заданному вам алгоритму в ваши структуры данных. Передать структуру данных получателю (как вариант например Потоку работы с БД) Отсюда вытекает, что в случае множества получателей внутри процесса структура данных должна иметь адрес - кому она предназначена, что бы быть потом адресованной нужному получателю. Как это реализовать ?

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

> P.S. Дополнительный вопрос - является ли вызов методов SendTo
> и recv потокобезопасными для socket"а?

Если я правильно понял вопрос - можно ли вызывать их из разных потоков (разделить передачу и прием) или потока не создавашего сокет?
Для Windows (к сожалению не знаю как работают сокеты в этом случае в других системах, но подозреваю так же) -
да можно разделить прием от передачи по разным потокам. А создателем сокета может быть совсем другой поток.

← →
Добежал (2009-01-30 11:11) [10]

Вариант, здорово, конечно, но очевидно.

Протокол уже есть, определен не мной. Задача поставлена.

Я предложил свое решение проблемы - это сделать обслуживающий поток, который постоянно будет выгребать данные из сокета и складывать в общий буфер какой-то. Я обозначил проблему, которую вижу при этом - как сигнализировать исходному потоку о том, что данные получены? На каждый запрос создавать объект ядра? Как-то расточительно.
Ждать константное время и и проверять ответ - неэффективно в плане времени.

Или может есть другие стандартные решения данной проблемы.

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

← →
Сергей М. © (2009-01-30 13:36) [11]

> на время общения одного потока другие потоки ждут

А что им остается делать, если твой прикладной протокол изначально, как я понял, не подразумевает асинхронности ?

← →
Anatoly Podgoretsky © (2009-01-30 13:45) [12]

> Сергей М. (30.01.2009 13:36:11) [11]

> не подразумевает асинхронности

И потоковости.

← →
Добежал (2009-01-30 13:47) [13]

а причем здесь асинхронность протокола, если общение идет с разными клиентами?

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

← →
Вариант (2009-01-30 14:15) [14]

> Мне нужно конкретика по тому, как организовать прием данных.

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

← →
Сергей М. © (2009-01-30 14:38) [15]

> причем здесь асинхронность протокола

Притом что синхронный протокол подразумевает схему :

1. Отправка запроса
2. Ожидание ответа на предыдущий запрос
3. Прием/обработка ответа на предыдущий запрос
4. goto 1

Асинхронный же подразумевает иную схему:

1. Отправка одного или более маркированных запросов, если таковые имеются во вх.очереди.
2. Таймаут-ожидание ответа на любой из отправленных запросов, при таймауте goto 1
3. Прием и диспетчеризация ответа на конкретный ранее когда-либо отправленный запрос.
4. goto 1
2.

← →
Сергей М. © (2009-01-30 14:39) [16]

> 2.

Это очепятка

← →
Добежал (2009-01-30 15:26) [17]

> Притом что синхронный протокол подразумевает схему :

слушай, не нужно вырывать слово из контекста. Я не задавал вопрос "причем здесь асинхронность протокола", я задавал вопрос: "причем здесь асинхронность протокола, если общение идет с разными клиентами?".

Протокол HTTP точно также синхронный. Но почему же тогда Apache в одно время может общаться с тысячами клиентов, а?!

По старой привычке вместо того, чтобы помочь решить проблемы, ты продолжаешь докапываться к словам, причем чем дальше тем делая это все более неумело. Каждый поток приложения общается со СВОИМ устройством по СИНХРОННОМУ ПРОТОКОЛУ и никакой асинхронный протокол тут нафиг не нужно. Проблема одна - что пакеты от разных IP приходят на один порт и нужно синхронизировать чтение приходящих пакетов.

← →
Добежал (2009-01-30 15:29) [18]

> Создать событие на поток - это не расточительство

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

← →
Anatoly Podgoretsky © (2009-01-30 16:26) [19]

> Протокол HTTP точно также синхронный.

Неправда

← →
Anatoly Podgoretsky © (2009-01-30 16:27) [20]

В потоках конечно проще использовать синхронные методы, но ты то не умеешь работать с потоками, видимо реализовал по схеме Архангельского.

← →
Сергей М. © (2009-01-30 16:35) [21]

> почему же тогда Apache в одно время может общаться с тысячами
> клиентов, а?

Потому что это сервер, создающий для каждого клиента отдельное TCP-соединение.

А у тебя UDP, который НЕ подразумевает соединение !

Так что не надо сравнивать кислое с пресным, добежавший ты наш)

← →
Добежал (2009-01-30 16:48) [22]

> Потому что это сервер, создающий для каждого клиента отдельное
> TCP-соединение

и что дальше? А сам протокол верхнего уровня HTTP синхронный. И у меня протокол верхнего уровня синхронный. И ты пишешь, что это бред. Ну иди заяви, что протокол HTTP тоже бред.

Я не спрашиваю как мне идентифицировать устройства. Я спрашиваю как лучше синхронизировать выборку сообщений. Вопрос ИМЕННО В ЭТОМ.
Если ты хочешь сказать, что это невозможно сделать - то это неправда. Я уже предложил способ, как это можно сделать. И вопрос в том, насколько оптимален этот способ и можно ли сделать лучше.

← →
Anatoly Podgoretsky © (2009-01-30 16:51) [23]

> Добежал (30.01.2009 16:48:22) [22]

Метод Synchronize

← →
Сергей М. © (2009-01-30 17:03) [24]

> событие придется создавать не на поток, а на каждый запрос

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

Транспортный поток создает объект-очередь передачи, доступный всем прикл.потокам.

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

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

Прил.потоки ждут каждый своего сигнала, после чего выгребают свои очереди и сбрасывают сингал.

Все !

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

← →
Slym © (2009-01-30 17:27) [25]

Как обычно вода в ступе...
Не ветка помойка...
По сабжу...
Решение задачи возможно только при следующих условиях - уникальности и точной идентификации входящих пакетов!
Т.е. если отдельная задача тебует общения с устройством с уникальной парой RemoteIP:RemotePORT!

потоки передают сообщение, адресат (IP:PORT) и объект для ответа (колбек функция, объект синхронизации (евент)) транспортному потоку

send(msg:string;ip:string;port:word;callback:procedure)

травспортный поток заносит в свою внутреннюю таблицу ключ IP:PORT и соответствующий ему объект от
IP:port=callback
и отправляет сообщение...
при приходе ответа в таблице по ключу RemoteIP:RemotePort ищется калбек и при нахождении вызывается с принятым сообщением

← →
Slym © (2009-01-30 17:28) [26]

по типу NAT таблицы и преобразования пакетов

← →
Добежал (2009-01-30 17:34) [27]

Удалено модератором

← →
Добежал (2009-01-30 17:54) [28]

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

> Решение задачи возможно только при следующих условиях -
> уникальности и точной идентификации входящих пакетов!

такое условие есть.

> потоки передают сообщение, адресат (IP:PORT) и объект для
> ответа (колбек функция, объект синхронизации (евент)) транспортному
> потоку

вот тут начинаются проблемы. Система плагиновая DLL, поэтому потоков от сторонних плагинов может быть сколько угодно, может быть они вызовут транспорт один раз, а может будут вызывать постоянно. Соответственно, или на каждый запрос каждый раз создавать объект синхронизации и потом его удалять, или на каждый ID потока создавать свой объект синхронизации, а при каждом запросе искать его в таблице - вопрос тогда, когда и кем финализировать объекты синхронизации.
И нельзя ли сделать как-то оптимальнее.

Удалено модератором

Добежал (30.01.09 17:54) [28]
можно сразу в блокирующий метод обрамить код:

ClientTable - поля (Event,IP,Port,Msg); TRec=record Event,IP,Port,Msg end; function Transport.Send(const Msg,IP:string;Port:word;TimeOut:longint):string; begin result:=""; Event:=CreateEvent; try ClientTable.Add(Event,IP,Port); Socket.send(Msg,IP,Port); if WaitForSingleObject(Event,TimeOut)=OK then result:=ClientTable.GetMsg(Event); finally ClientTable.Remove(Event); CloseObject(Event); end; CloseHandle(Event); end; procedure OnSocket.Recv(RemoteIP,RemotePort,Data); var Record:TRec; begin if ClientTable.FindRecord(RemoteIP,RemotePort,var Record) then begin Record.Msg:=Data; SetEvent(Record.Event); endl end; procedure Thread.Execute begin while not Terminated do begin s:=Transport.send("Hello!","192.168.1.1",666,TimeOut); if length(s)<>0 then Writeln(s); end; end;
Доступ к ClientTable и Socket должен быть заключен в критическую секцию

можно не постоянно креатить евенты а держать пул евентов

← →
Добежал (2009-02-02 10:32) [33]

> можно не постоянно креатить евенты а держать пул евентов

ну да, наверное так и сделаю. Спасибо!

← →
Добежал (2009-02-02 13:44) [34]

Удалено модератором

Добежал (02.02.09 10:32) [33]
сделай "чтоб работало", оптимизировать потом будешь...
время создания евента ничтожно по сравнению с временем убитым на грамотную организацию пула :)

Slym © (02.02.09 19:13) [35]
сделай "чтоб работало", оптимизировать потом будешь...
время создания евента ничтожно по сравнению с временем убитым на грамотную организацию пула :)

Да я в общем переживаю не за время создания event"а, а за WaitForSingleObject. Если эвент занят, то поток уходит в режим ядра для засыпания, на это тратится немало тактов. С другой стороны, с современными процессорами это и не замечишь.

Просто хочется сделать нормально и забыть о транспорте.

Удалено модератором

Отправка UDP по разным адресам через один сокет Найти похожие ветки