Извиняюсь за многословность...

При обмене данными посредством nio и службы select, данные из входного потока поступают порциями, которые могут не соотвествовать размерам передаваемого байтового массива (сериализованного объекта в произвольном формате(бинарный, xml, и т.д.)). Очевидно, получаемые куски нужно ЯВНО копировать в буферы накопители для каждого живого ключа и както в будущем заканчивать это накопление. Стандартный алгоритм, где используется один небольшой перезаписываемый буфер, не подходит по вышеписанной причине.
ЯВНОЕ копирование означает, что придется отказаться от оптимизации чтения из системного буфера, при котором реального копирования не происходит. Иначе мы не сможем слепить целевой массив.

Вопрос первый: как лучше всего реализовать данный алгоритм? Может быть есть стандартные готовые решения?

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

Вопрос: Как "сказать" сокету, каналу или селектору сохранить канал и закрыть его если на нем не будет активности начиная с текущего момента и до некоторого таймаута. Да так сказать, чтобы он не заблокировался на этот срок. Я пробовал найти свойство у сокета, канала, в котором зафиксировано время его открытия, но не нашел. Это решило бы проблему. Наследовать класс и добавить свойство не получится, ибо объект входящего соединения создается системой, а не приложением:

Socket              s  = myServerSocket.accept();
SocketChannel c  = s.getChannel();

Подведу итог.
Я предполагаю, что для каждого канала нужны, как минимум, еще два свойства
1) Время открытия
2) Буфер накопитель

также я предполагаю, что придется парсить входящие порции на предмет наличия в них управляющих команд, реализующих некий оригинальный протокол приема массивов.
ТОгда зачем всё это, может быть эффективнее использовать стандартные протоколы типа RMI, CORBA

Можно ли всё это реализовать по другому, как-нибудь поизящнее?

Начнем с конца.



NIO труднее программировать, но работать ( возможно ) будет быстрее ( неблокирующий ввод/вывод одним потоком, прямое размещение буферов в памяти ).

Стандартные протоколы, по идее, проще программировать. Они для того и создавались, чтобы сразу, легко и быстро, не приходя в сознание, на профессиональном уровне и т.д. и т.п. Ну а то, что они чуть менее эффективны, так железо - дешево, а оплата труда - дорога, и т.д. и т.п.

Так что, ответа нет.

Теперь по сути.



Они могут поступать по одному байту! Могут поступать с паузами (и отсылаться они могут точно также) . Поэтому при чтении из буфера сообщение должно где-то накапливаться.  Приемная сторона должна знать длину сообщения. "Иначе мы не сможем слепить целевой массив".



Если через соединение передается неизвестное количество сообщений ( характерно для постоянных соединений), то пауза в данных не обязательно является сигналом к закрытию соединения. Закрывать соединение надо, если клиент явно прислал команду "конец связи" или пауза превысила таймаут. В последнем случае сервер убивает соединение (в целях сбережения своих ресурсов) без выяснения причин (врядли это всегда возможно) паузы, которых может быть много (обрыв на линии связи, отключили электричество у клиента, ...) .



Нужен внешний поток - монитор, который периодически вычисляет текущую паузу у каждого соединения и сравнивает ее с таймаутом. И принимает решение.

Мне кажется, изящнее ничего пока нет.




Это сообщение отредактировал(а) COVD - 9.2.2007, 17:41

Согласен.

Значит вердикт такой.
СокетСервер в своем потоке принимает сообщения и без обработки передает их соседнему потоку-Монитору, работающему в режиме событийного цикла, который ведет буферизацию, учет времени и состояния готовности для каждого канала. Быстро передав сообщение, СокетСервер читает текущее состояние (команду) текущего канала из переменной в Мониторе. Если состояние="ОТВЕТ ГОТОВ", то СокетСервер берет от Монитора байтовый массив с "ОТВЕТОМ" и передает его в поток вывода, если "ТАЙМАУТ" или "ЗАКРЫТЬ", то закрывает канал, если другая команда, то выполняет ее, если нет команд то ничего не делает и переходит к другому каналу.

Монитор определяет длину сообщения, накапливает массивы до определенной длины или таймаута, меняет переменную состояния, определяет тип данных и по готовности передает их новому потоку Исполнителю. Потоки Исполнители открываются по необходимости и закрываются по готовности. В потоках Исполнителях вызываются тяжелые методы сериализации/десериализации, парсинга, запросы к базе данных, вычисления.

Итого от 2 до 2+X потоков.

Вроде бы не сложно.



Это сообщение отредактировал(а) JavaCraft - 10.2.2007, 13:41

Не принижайте  задачу  .  Это - сложно. 

Я под потоком - монитором подразумевал поток, занимаюшийся исключительно "отстрелом" мертвых соединений, у которых пауза в активности превысила таймаут. И ничем более. А определять момент окончание приема сообщения может основной поток (ввод/вывод, который с селектором работает). Он же может и отправлять принятое сообщение на обработку потокам - Исполнителям.

Действительно, на приемной стороне надо 2 потока (ввод/вывод + монитор/"санитар") + N потоков - исполнителей (worker threads), которые обычно организуются в виде пула.


Это сообщение отредактировал(а) COVD - 10.2.2007, 17:22

В общем что-то такое сваял. И вот в чем вопрос!
Сервер принимает пакеты sc.read(buffer) и передает их Монитору, который их склеивает в буферах накопителях.
Монитор готовит ответы(довольно длинные) и помещает их в буфера.
Когда сервер видит, что ответ для данного канала готов, он отправляет его клиенту через тот же самый сокет, методом sc.write(buffer).
Очевидно, если буфер большой, сервер не перейдет к обработке другого канала, пока не передаст буфер по этому каналу. Это потенциально, может перечеркнуть все выгоды асинхронного обмена.

В хелпе Eclipse я обнаружил после
sc.write(buffer);
вызов метода (сжатие путем сдвига текущей позиции в начало)
buffer.compact(); и пояснение "Если буфер передан не полностью.."

Как так не полностью... Как метод write решает когда прекратить передачу?
Как установить критерий прерывания, если меня не устраивают критерии по умолчанию?
Например, уменьшить размер порции или увеличить. И надо ли это делать?

и зачем вообще для этого compact(), который копирует данные? Вместо этого можно просто оставить состояние position буфера как есть. Тогда при следующей попытке write должен продолжить с текущей позиции. Или я не прав?

All, что скажете, по моим вопросам?

1. Поскольку обмен асинхронный, то ByteBuffer -ов надо два - один для read, другой для write.

2. 

buffer - фиксированной длины и обычно allocateDirect. "Ответ" в общем случае может быть длиннее буфера. Следовательно,  данные могут передаваться по частям. 

"..сервер не перейдет к обработке.., пока не передаст буфер по этому каналу" - неверно для неблокирующего режима. Операция sc.write(buffer) отошлет столько байт из buffer, сколько сможет и вернет управление. Возможно и ноль байт. После этой операции обычно делают compact(), чтобы удалить отосланные байты и освободить пространство для следующих данных. 

Предположим, что "ответ" готов и его надо отослать. Для этого "ответ" копируется в buffer. Вовсе не обязательно, что в buffer есть место для полного "ответа". Копируется по мере освобождения места в буфере.

Таким образом один поток заполняет буфер данными, а другой поток (который с селектором работает) его опустошает (sc.write(buffer) + buffer.compact()).

Между прочим, при чтении все то же самое, но наоборот.


Это сообщение отредактировал(а) COVD - 13.2.2007, 22:23

Обмен не полностью асинхронный. Если его разрешить полностью асинхронным, то для выполнения запросов, содержащихся в Целых сообщениях, потребуется открывать всё больше и больше потоков-Исполнителей. Предыдущие потоки-Исполнители не будут успевать завершаться. В результате производительность упадет до предела. Такой сервер легко забить запросами и "повесить на собственом галстуке".

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

Это сообщение отредактировал(а) JavaCraft - 14.2.2007, 01:07

Не совсем точно. Это системный буфер фиксированной длины, а мой buffer может быть любым. Я разобрался с этой проблемой. Прерывание, точнее Исключение, генерируется службой select  в момент заполнения системного буфера и блокировки канала в процессе записи из пользовательского буфера. Служба мониторит эти исключения и передает управление на другие каналы. Поэтому данные и передаются по частям.


Собственно "Ответ" и "Буфер" у меня синонимы, т.к. я имею в виду пользовательский буфер. В любой литературе, примерах и у меня под buffer понимается пользовательский буфер, а не системный. Описаний системного буфера я не встречал. Знаю только что он существует где-то глубоко в потрохах nio, а как он называется там внутри, кто его знает!


Так-то оно так, но ткните меня носом, где в литературе сказано, что чтение и запись выполняется в один и тот-же системный буфер(СБ). У чтения и записи разная логика работы. Запись в СБ выполняется до тех пор пока в нем есть место, после чего блокируется и СБ "выталкивается" в сеть.  При автоматическом режиме выталкивания, СБ не выталкивается в сеть пока не будет заполнен до конца. 
Если в нем находятся пакеты для чтения, зачем их выталкивать в сеть???!!!

Запись извне в буфер (OP_READ) для Чтения выполняется до тех пор пока в буфере есть место, после чего блокируется. Нигде не упоминается, что чтение не может быть выполнено при нулевом заполнении буфера, а также что оно автоматически инициируется при его заполнении. Этого нет. Буфер просто блокируется и вызывается исключение. Нигде не упоминается, что записанный(OP_WRITE) пакет не нужно читать селектом, и не упоминается как такие пакеты распознавать. Характеристики OP_READ и OP_WRITE являются постоянными параметрами зарегистрированного канала, а не пакетов.

Из всего этого можно сделать вывод, что существуют два системных буфера - для чтения СБ1() и для записи СБ2, скрытые от нас под методами read() и write(). Соотвественно, когда и как мы их будем заполнять или опустошать, в одном потоке или в двух, не принципиально. Ведь это делается независимо! Или я что-то напутал??? Поправте плиз!

Наверное поэтому в примерах в литературе не объясняется почему в одном цикле Select выполняется последовательное read, а потом после обработки, write для одного канала в одном потоке. А надо было бы объяснить!
Ведь при одном потоке у "писателя" просто не будет шансов поиметь свободное место в системном буфере и вставить туда хотя бы 1 байт (OP_WRITE), его постоянно будут заполнять поступающие извне пакеты (OP_READ), которые поступили ПОСЛЕ read(), но ДО write(). Это было бы именно так, если бы был один буфер, но это не так, не случайно в литературе нигде об этом нюансе не предупреждается. Этого просто не происходит.

Теперь о потоках. Два потока - приема и передачи, нужны, если нужна непрерывная трансляция в обе стороны. В моей задаче это не нужно. Поэтому, синхронность "макро-фаз" чтения и записи всё же сохраню, т.к. нельзя ответить на еще не полученный полностью запрос, как неразумно позволить задавать новые запросы(вопросы) не получив ответа на предыдущие. Иначе клиент засыплет сервер вопросами, ответы на которые его не особо интересуют. В других задачах, возможно, полностью асинхронный режим более рационален.

И всё равно не понял зачем использовать compact(), когда лучше обойтись без него. После каждой фазы буфер-накопитель очищается полностью, а в процессе записи достаточно счетчика текущей позиции.
Например, если "Ответ" весит 1Мб, а средний сегмент равен к примеру 128 байт, то данный мегабайт, постепенно уменьшаясь, будет перезаписан compact() примерно 8192 раза, в то время, как можно было бы обойтись без перезаписывания. А если "Ответ" больше 1М, к примеру 2D/3D матрица, как результат численного моделирования физических процессов может достигать, например, 10M? И это вполне законные запросы, которые нельзя отбросить как "слишком большие" - не поймут заказчики.



Это сообщение отредактировал(а) JavaCraft - 14.2.2007, 16:34

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

Добавлено @ 16:29 

Я буфером называл обьект ByteBuffer, который участвует в операциях read/write. Вы назначаете его размер и он обычно постоянен, в том смысле, что не меняется в зависимости от длины передаваемого сообщения ("Ответа") . А размер ответа может быть любой.



Я не знаю про "системный" буфер и ничего про него не говорил. Программисту доступен буфер - обьект класса ByteBuffer, который он сам создает и этого достаточно.

В неблокирующем режиме write копирует байты из этого ByteBuffer (столько, сколько сможет в данный момент) на отправку (по вашей терминологии - в "системный" буфер). И все. По какой причине не скопировано или сколько скопировано - не наша забота. 

Важно то, что если N байтов скопировано, то position увеличится на N. А скопированные байты так и останутся на своем месте в ByteBuffer .  Но мы имеем полное право их удалить и для этого делаем compact().

Это сообщение отредактировал(а) COVD - 14.2.2007, 16:52

Я отредактировал свой пост.

У меня копирование Из "Ответа" в "ByteBuffer" исключено, поэтому это одно и тоже. И у меня ByteBuffer переменной длины, в зависимости от длины Ответа, содержащегося в нем. Длина устанавливается один раз, в момент определения длины сообщения Запроса или Ответа. ByteBuffer передается в read() и write(), без копирования, без обнуления позиции и без компактификации. В общем, как я и говорил ранее, у меня Буфер и Ответ это одно и тоже. Я думаю мы поняли, кто, что имеет в виду.

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

Добавлено @ 17:09 


Раз у вас буфер одноразовый, то и чистить его, действительно, не надо. Поэтому у вас и возникает постоянно вопрос блокировки. Мне кажется это плохо вписывается в идею неблокирующего IO.  

Ну так понятно по какой причине. "Системный буфер" заполнился и это вызвало исключение.
Это не я придумал "Системный буфер", может он по другому называется, но он реализован в nio или еще глубже. Вопрос в том, является он парой буферов(массивов) - чтения и записи, или пакеты чтения и записи перемешаны в одном буфере(массиве). Я думаю, что имеет место первый случай.

Добавлено @ 17:17 




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

Да нет, вроде. Не должно быть исключения. Просто канал не будет готов к записи и не будет селектирован селектором. Со временем "системный буфер" рассосется и селектор селектирует каналы для write.

Добавлено @ 17:17 






Я имел в виду что каждого канала есть два постоянных ByteBuffer - один для read, другой для write. А у вас одноразовые буферы. Вы их создаете под каждое сообщение и выбрасываете. Отсюда и путаница 

Это сообщение отредактировал(а) COVD - 14.2.2007, 17:31

Селектор сообщает нам какой канал "не пуст" и какой канал "пуст", что трактуется как готовность к "чтению" из непустого буфера и "Записи" в пустой, соотвественно. Согласитесь, что одновременно канал не может быть и "пустым" и "не пустым", что говорит о двух буферах. Это так, в дополнение к предыдущим мыслям.

Эти Условия не конкурируют между собой и не противоречат друг другу:

(key.readyOps() & SelectionKey.OP_READ)==SelectionKey.OP_READ
(key.readyOps() & SelectionKey.OP_WRITE)==SelectionKey.OP_WRITE

Биты готовности к чтению и к записи устанавливаются и маскируются независимо.

А метод key.readyOps() вовсе не говорит о том что имеется в виду:
1) Зарегистрированный "интерес" канала, то в чем канал "заинтересован"
2) или Текущая готовность, которая может быть не равна зарегистрированному "интересу".

Остается только гадать.

Это сообщение отредактировал(а) JavaCraft - 14.2.2007, 17:51