Очередь потоков

← →
gate A20 (2002-05-18 13:23) [0]

Господа мастера! Срочно!
Посоветуйте, как организовать очередь потоков на запись в файл (на локальной машине, естественно). Ситуация такая - есть некий файл, создаётся куча потоков (которые могут принадлежать разным процессам), которые желают писать что-то в файл. Причём писать они должны в порядке своего создания, FIFO, Кто-Первый-Встал-Того-И-Тапки или как там это ещё. Придумал следующее: (просьба их лажать или предлагать новое):
Есть маза делать всё типа как в примере Handshake в главе 9 у Рихтера, (шарится память, к файлу приматывается поток-сервак, именованое событие-реквест серваку и именованое событие-акцепт клиенту), но сразу после создания очередного потока-клиента (до запроса серваку) оный поток проверяет, "кто последний в очереди" - в пошареную память последний поток пишет свой хэндл. Возникает вопрос - потоки в очереди принадлежат разным процессах, и хэндл потока может таковым не являтся в другом процессе (хэндл - он же типа не реальный, а виртуальный адрес, и действует в рамках одного процесса).
Примечание: на поле шареной памяти с хэндлом можно повесить объект типа "Читай Кто Хочет Писай Только Один" (забыл как такая штука в дельфях называется).
зы: Тут люди присоветовали сокетами это делать - там вроде как очередь можно делать, но чё-то я думаю, что нафиг надо - сокеты на локальной машине. К тому же в виндах, по-моему, очередь в сокетах не организуешь - это в юниксе было можно.

← →
paul_shmakov (2002-05-18 18:42) [1]

вот примерная схема:

есть поток сервер и рабочие потоки.

рабочий поток при своем создании обращается к потоку-серверу и сообщает о своем желании записи в файл (скажем, вызывает RequestWriteOperation). тот ему в ответ выдает handle объекта событие (CreateEvent), т.е. для каждого рабочего потока свой объект-событие. если требуется передача в другой процесс, то поможет DuplicateHandle.
рабочий поток ждет установки события (WaitForSingleObject).

поток-сервер хранит список этих событий. каждое событие по очереди (FIFO) устанавливается (SetEvent). рабочий поток просыпается и выполняет запись в файл.

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

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

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

далее, поток-сервер повторяет процедуру для следующего события в списке, пробуждая следующий рабочий поток и т.д.

рабочие потоки могут несколько раз вызывать RequestWriteOperation, если нужно.

не забывайте только DuplicateHandle вызывать, когда нужно (т.е. когда идет передача хендла в другой процесс) и CloseHandle.

← →
gate A20 (2002-05-20 18:27) [2]

2paul_shmakov:
Большое спасибо за совет, но, по-моему, он не совсем мне подходит вот по каким причинам:

>рабочий поток при своем создании обращается к потоку-серверу
а как Вы себе это представляете? А если сервак _уже_ ждёт конца файловой операции какого-либо из клиентов? Он же недоступен в этот момент. А ставить на поток-сервак синхронизатор типа Free/Busy - имхо только усложонять себе жизнь.
>рабочий поток при своем создании обращается к потоку-серверу
Всё конечно хорошо, но если (винда обладает вытесняющей многозадачностью) в момент создания рабочего потока (т.е. клиента) в другом процессе с более высоким КЛАССОМ приоритета создаётся поток, то шедуллер управление передаст этому новому потоку. Оговорочка: первый поток не успел сделать реквест серваку. Это момент раз.
>которое тоже передается в рабочий поток при его создании (т.е. при вызове RequestWriteOperation).
то же, что и выше.
>то поможет DuplicateHandle - почему именно так? имхо, проще будет OpenEvent(...MyQueueEvent+N...), где N - индекс события в пресловутом массиве. Впрочем, это, наверное, непринципиально.
>рабочий поток просыпается
а если не проснётся? предположим, процесс, в рамках которого поток существовал, будет прибит (специфика задачи такова, что это очень может быть). Данные для реквеста потеряны. Наверное, следует CreateRemoteThread"ом создавать поток-клиент в адресном пр-ве того же процесса, где сервак? Это момент два.
И ещё непонятно, когда сервак начинает разруливать очередь событий? - это три.

И ещё раз большое спасибо за совет. BR, Gate A20

← →
paul_shmakov (2002-05-21 14:20) [3]

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

>рабочий поток при своем создании обращается к потоку-серверу
а как Вы себе это представляете? А если сервак _уже_ ждёт конца файловой операции какого-либо из клиентов? Он же недоступен в этот момент.

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

>рабочий поток при своем создании обращается к потоку-серверу
Всё конечно хорошо, но если (винда обладает вытесняющей многозадачностью) в момент создания рабочего потока (т.е. клиента) в другом процессе с более высоким КЛАССОМ приоритета создаётся поток, то шедуллер управление передаст этому новому потоку. Оговорочка: первый поток не успел сделать реквест серваку. Это момент раз.

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

>то поможет DuplicateHandle - почему именно так? имхо, проще будет OpenEvent(...MyQueueEvent+N...), где N - индекс события в пресловутом массиве. Впрочем, это, наверное, непринципиально.

абсолютно точно, как душа пожелает.

>рабочий поток просыпается
а если не проснётся? предположим, процесс, в рамках которого поток существовал, будет прибит (специфика задачи такова, что это очень может быть). Данные для реквеста потеряны. Наверное, следует CreateRemoteThread"ом создавать поток-клиент в адресном пр-ве того же процесса, где сервак? Это момент два.

проснется, никуда не денется. а если вы про то, что один из клиентов может повиснуть, и тем самым подвесить всех остальных, то можно поставить таймаут на файловую операцию (у WaitForSingleObject, которую вызывает сервер можно таймаут задать).

И ещё непонятно, когда сервак начинает разруливать очередь событий? - это три.

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

вообще, эту задачу можно кучей разных способов реализовать. например, самый простой и понятный:
1) сервер имеет скрытое окно, в это окно клиенты посылают сообщения с запросом на операцию с помощью SendMessage. чем удобно окно? оно уже имеет очередь сообщений (руками ее реализовывать не надо).
2) клиенты используют SendMessage. эта функция синхронная, т.е. поток клиента засыпает, пока сервер не обработает сообщение. т.о. клиенты все спят и ждут своей очереди.
3) сервер по очереди в цикле выборки сообщений обрабатывает запросы. как только сервер обрабатывает очередное сообщение с запросом и возвращает какое-нить значение, клиент просыпается и выполняет операцию работы с файлом.
4) проблема: серверу нужно узнать, когда очередной клиент завершит файловую операцию. только после этого он может обрабатывать следующее сообщение из очереди сообщений окна. для этого можно использовать, например, именованное событие (CreateEvent). сервер перед возвратом из обработчика сообщения сбрасывает это событие (ResetEvent), делает возврат из оконной процедуры и ждет установки события (WaitForSingleObject); клиент выполняет файловую операцию и устанавливает событие (SetEvent); сервер просыпается и обрабатывает следующее сообщение из очереди.

эта схема еще проще.

← →
Fellomena (2002-05-21 16:20) [4]

Я прошу прощения, но может проще так:
Есть глобальный стэк(FIFO) в который "складываем" handle всех создаваемых потоков в порядки их появления.
Первый появившийся поток ставит именованный мьютекс и пишет в файл, все остальные потоки ждут, пока мьютекс освободится.
Когда мьютекс освобождается - делаем проверку какой поток требует ресурс - если его дескриптор стоит последним в нашем стэке, то отдаём ресурс и устанавливаем мьютекс.
Поток отработал - удаляем его handle из стэка, освобождаем мьютекс и как раньше...

Вроде так можно...

← →
paul_shmakov (2002-05-21 16:40) [5]

2 Fellomena:
так именно про (примерно) такую схему и был первоначальный вопрос, обсуждаются всякие мелочи.
например:

Когда мьютекс освобождается - делаем проверку какой поток требует ресурс - если его дескриптор стоит последним в нашем стэке, то отдаём ресурс и устанавливаем мьютекс.

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

← →
Fellomena (2002-05-21 16:53) [6]

ok, sorry за невнятное прочтение предыдущих постов.

По делу: можно сделать так:
при создании потока присваивать ему уникальный идентификатор в виде строки (второй стэк, идентичный по индексации первому).
И каждый раз в if OpenMutex(...) задавать имя для мьютекса, которое последнее во втором стэке, тогда доступ к ресурсу поимеет только тот поток, индекс(порядковый номер в стэке) будет соответствовать порядковому номеру идентификатора.

Пойдёт так?

← →
gate A20 (2002-05-21 19:00) [7]

А можно ли сделать так (маза родилась из-за лени):
В процессе-мониторесоздаётяс ФайлМэппинг(с хэндлом INVALID_HANDLE_VALUE) - шарная память короче. В том же процессе открывается нужный файл, полученный хэндл пишется в ФайлМэппинг.
В неком процессе возникло желание записать что-то в файл. В процессе - мониторе CreateRemoteThread"ом создаётся поток, причём то, что надо записать в файл, передаётся потоку как параметр. Этот поток начинает ждать освобождения хэндла файла (который в шарной памяти). Погодите закидывать меня дохлыми помидорами!
Как сказано у господина Рихтера, в случае, если несколько потоков с одинаковым (для этого и не только мы потоки в наш процесс пихнули) приоритетом ждут освождения некого объекта ядра, то (если ни один из этих потоков не приостановлен) управление получит дольше всех ждавший поток. Дальше у того же Рихтера сказано, что объект ядра файл (в смысле хэндл оного):
занят - если выдан запрос на ввод-вывод
свободен - завершено выполнение запроса на ввод-вывод
вот тепрь просьба лажать. спасибо.
2 paul_shmakov ©:
Вы предлагали юзать очередь сообщений - да, можно. Но не нужно - основному процессу программы вообще никчему всякие окна - он у меня сервис.

← →
paul_shmakov (2002-05-21 20:12) [8]

2 gate A20:
ok, пойдем по пунктам.

В процессе-мониторе создаётяс ФайлМэппинг(с хэндлом INVALID_HANDLE_VALUE) - шарная память короче. В том же процессе открывается нужный файл, полученный хэндл пишется в ФайлМэппинг.

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

В неком процессе возникло желание записать что-то в файл. В процессе - мониторе CreateRemoteThread"ом создаётся поток, причём то, что надо записать в файл, передаётся потоку как параметр.

1) CreateRemoteThread есть только в nt/2k/xp. 2) зачем? я не очень понимаю, для чего это. для того, чтобы только сервер осуществлял операции с файлом? если так, то OK, но не CreateRemoteThread; 3) "что надо записать в файл, передаётся потоку как параметр" - для этого нужно это "что надо" сначала записать в адресное пространство сервера, т.е. не так просто, хотя реализуемо.

Этот поток начинает ждать освобождения хэндла файла (который в шарной памяти).

хэндла он ждать не может, т.к. смотри выше (хэндл процессо-зависим).

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

это не так, точнее не для всех операционок.

А можно ли сделать так (маза родилась из-за лени)
лень здесь не помогла, это решение в реализации ничуть не проще. к тому же не совсем верное.

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

← →
gate A20 (2002-05-22 11:10) [9]

2paul_shmakov ©

зачем? зачем записывать хэндл открытого файла в шаренную память, если значение хэндла процессо-зависимо? смысла никакого.
Ну да. Хэндл - фактически виртуальный адрес объекта. Он естественно процессо-зависим. Ну так нигде кроме нашего процесса он и не используется. А в шарную память он пишется чтобы был виден всем потокам процесса - зачем каждый раз открывать файл? Где проблема?

CreateRemoteThread есть только в nt/2k/xp
Ничего другого я не поддерживаю.

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

хэндла он ждать не может, т.к. смотри выше
см. МСДН или Рихтера - может. http://www.compbooks.narod.ru/book_files/head9.htm#h9t7

это не так, точнее не для всех операционок
не знаю не знаю. Интересно, откуда у Вас такие сведения?

ну уж если окно не хочется, то нужно реализовать аналог этого решения на объектах синхронзации.
гм. а о чём же, по-Вашему, беседа идёт?

на самом деле глюк есть - когда очередная И/О операция будет закончена, освободятся все потоки. А надо один. т.е. надо ещё некое событие с автосбросом прилепить - чтобы только один поток релизился.

← →
paul_shmakov (2002-05-22 12:10) [10]

2 gate A20:

>зачем? зачем записывать хэндл открытого файла в шаренную >память, если значение хэндла процессо-зависимо? смысла никакого.
Ну да. Хэндл - фактически виртуальный адрес объекта. Он естественно процессо-зависим. Ну так нигде кроме нашего процесса он и не используется. А в шарную память он пишется чтобы был виден всем потокам процесса - зачем каждый раз открывать файл? Где проблема?

вот так заява! (то, что жирным шрифтом выделено). исходная задача вами формулировалась для различных процессов!! (см. gate A20 (18.05.02 13:23)). поэтому тут столько и городилось! ну а если все в пределах одного процесса, то конечно все можно сильно упростить. все сложности были преимущественно как раз из-за требования обращений из разных процессов.

>CreateRemoteThread есть только в nt/2k/xp
Ничего другого я не поддерживаю.

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

>зачем? я не очень понимаю, для чего это
чтобы у всех потоков-клиентов был приоритет одинаковый

CreateRemoteThread не делает всем, созданным ей, потокам одинаковый приоритет! эта функция всего лишь позволяет создать поток в чужом процессе. хотя, возможно, я понимаю, о чем вы говорите. когда поток-клиент создается в клиентском процессе, а не в серверном, и у клиентского процесса класс приоритета выше/ниже, чем у серверного, то клиентский поток будет иметь базовый приоритет, который действительно будет отличаться от приоритета такого же потока, но созданного в процессе-сервере. об это речь?
если да, то здесь нет никаких проблем (если, конечно, а условиях задачи нет чего-нибудь ну очень специфичного). какая разница какой приоритет у рабочих потоков? они все равно все становятся в общую очередь, и обрабатываются по очереди. порядок обработки (выборки из очереди) не зависит от приоритета потока! будь один из них хоть idle, а другой realtime - все равно это не повлияет на очередность (в той реализации, о которой я раньше писал) - если первый в очередь встал, то первым и выйдет.

>хэндла он ждать не может, т.к. смотри выше
см. МСДН или Рихтера - может. http://www.compbooks.narod.ru/book_files/head9.htm#h9t7

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

>это не так, точнее не для всех операционок
не знаю не знаю. Интересно, откуда у Вас такие сведения?

проверьте :) вы меня так безапеляционно отправили читать МСДН и Рихтера, что позвольте я процитирую хотя бы второй из этих источников:
Тут возникает интересный вопрос. Если несколько потоков ждет один объект ядра, какой из них пробудится при освобождении этого объекта? Официально Microsoft отвечает на этот вопрос так: «Алгоритм действует честно" Что это за алгоритм, Micro soft не говорит, потому что нс хочст связывать себя обязательствами всегда придер живаться именно этого алгоритма. Она утверждает лишь одно- если объект ожидает ся несколькими потоками, то всякий раз, когда этот объект переходит в свободное состояние, каждый из них получает шанс на пробуждение. Таким образом, приоритет потока не имеет значения- поток с самым высоким приоритетом не обязательно первым захватит объект. Не получает преимущества и поток, который ждал дольше всех. Есть даже вероятность, что какой-то поток сумеет повторно захватить объект. Конечно, это было бы нечестно по отношению к другим потокам, и алгоритм пытается не допустить этого. Но никаких гарантий нет. На самом деле этот алгоритм просто использует популярную схему "первым во шел — первым вышел" (FIFO). B принципе, объект захватывается потоком, ждавшим дольше всех. Но в системе могут произойти какие-то события, которые повлияют на окончательное решение, и ил-за этого алгоритм становится менее предсказуемым. Вот почему Microsoft и не хочет говорить, как именно он работает. Одно из таких событий — приостановка какого-либо потока. Если поток ждет объект и вдруг приоста навливается, система просто забывает, что он ждал этот объект. А причина в том, что нет смысла планировать приостановленный поток. Когда он в конце концов возоб новляется, система считает, что он только что начал ждать данный объект.
вроде ясно сказано, что механизм недетермирован, иногда работает FIFO, а иногда и не работает.

на самом деле глюк есть - когда очередная И/О операция будет закончена, освободятся все потоки. А надо один. т.е. надо ещё некое событие с автосбросом прилепить - чтобы только один поток релизился.

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

Очередь потоков Найти похожие ветки