есть очередь. В нее пишут N-потоков и читают M-потоков.

Пишушие потоки постоянно вычитываю данные с устройства.
Если пишущий поток хочет записать в очередь, то хотелось бы приостанавливать читающие потоки и
писать, т.к. потеря данных критична. Но если просто приостановить поток,
то при добавлении в него нового эл-та у читающий станут невалидными итераторы (если например это std::deque).

Как реализовать для потока механизм транзакций или что-то в этом роде ?

зачем нам итерирование по очереди? вполне хватит push/pop-ов. В таком случае проблем с итераторами нету вообще. А еще можно посмотреть в сторону IO.Completision, если нас устраивает форточки-онли платформа, и критично быстродействие)

хорошо - уговорил 

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

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

CEvent event;

при чтении:

event.Lock(INFINITE);//вечно
...проводим операции чтения
event.Unlock();


при записи то же самое. Таким образом получаем только один поток, работающий с данными. Тебе наверное надо, чтобы несколько могли читать одновременно. Тогда вводим event на запись и переменную типа volatile LONG rdData. Event на запись с ручным сбросом.

При записи:

enentWr.SetEvent();
дожидаемся окончания операциё чтения:
while(rdData)Sleep(0);
eventWr.ResetEvent();

При чтении:
while(WaitForSingleObject(eventWr,0)!=WAIT_OBJECT_0)Sleep(0);
::InterLockedIcrement(&rdData)//функция может называться несколько иначе, но Interlocked! Она монопольно испльзует переменную
//поработали
::InterLockedDecrement(&rdData)//тоже может быть другая, но суть: вычитает 1 из rdData. Она обратна Increment, которая +1

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

Но это все наверное невозможно делать в Borland C++ 3.1. и как это осуществлять под *nix я тоже не знаю.  
С наступающим

Добавлено @ 10:32 
да, по моему я перепутал SetEvent и ResetEvent    

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

З.Ы. Похоже нужно с приоритетами потоков ковыряться 

Во первых, если читающий вдруг получил меньший приоритет уже начав читать данные, а пишущий также вдруг получив больший приоритет и как следствие процессорное время написал, после обратной смены приоритетов представь, что же прочитает читающий? Или это г..но для тебя не критично? Во-вторых, тебе известен механизм раздачи процессорного времени между потоками с разными приоритетами в Windows XP, Windows CE, Unix и т.д. Мне лично нет. Я только предполагаю. Я почти уверен, что я не одинок.
В - третьих, после того, что ты процитировал, я долго и упорно писал текст дальше, именно это решение позволяет снизить ожидание до минимума (позволяющего корректно работать), но тебе похоже было леннь почитать дальше. Если тебе не лень, то в дальнейшем ты поймешь, что даже этот (первый из двух приведенных вариантов) вполне хорош, потому что стопудово работает. Ну решать тебе. Могу лишь добавить, что через приоритеты ты никуда не прийдёшь  

извлечение объекта из очереди не такая уж длительная операция  



Приоритет потока влияет на количество процессорного времени отдаваемого на выполнение этого потока, но ни в коем случае не влияет на приоритет доступа к памяти. У всех потоков одного процесса он одинаковый

Это сообщение отредактировал(а) VectorMan - 25.12.2006, 15:57

takedo прав:
Если поток уже захватил ресурс (читает или пишет), то остальные не должны его получить, невзирая на приоритеты. Т.е. читать могут несколько, писать - один. Это классическая схема доступа, прекрасно описана у Рихтера: Single writer - multiple readers. Для удобства использования все это дело лучше завернуть в класс.
Стало быть, остается: а) сделать операцию доступа (чтения\записи) достаточно короткой (т.е. блокировка ставится  на чтение\запись одного элемента). б) за счет раздачи приоритетов можно повысить вероятность того, что писатель быстрее будет получать доступ, чем читатели.
Это все.

zss, приведенный мной пример надо дорабатывать, так как возможен одновременный доступ писателя и читателя к данным.
Когда? Вот когда:
1. eventWr сброшен
Читатель А зашел в while(WaitForSingleObject(eventWr,0)==WAIT_OBJECT_0)Sleep(0); и пошёл дальше. (здесь условие изменено, это правильно)
2. Писатель взвел в это же время eventWr.SetEvent(); и пошел смотреть значение переменной while(rdData)Sleep(0); которая равна 0 в этот момент.
3. После пункта 2 читатель А увеличил значение переменной rdData и пошел работать дальше. Тоже самое(работа с ресурсом) может делать в это время и писатель.!!!
Так что этот вариант не прокатывает, надо подумать ещё чуток    .

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

Понятно - спасибо

Есть еще проблемка, которая вылезла при отладке.
Я знаю максимальный размер буфера устройства (чтение происходит в асинхронном режиме через ReadFile).
Я при запуске приложения выделяю такой буфер. (std::vector<unsigned char>::resize()).
После этого я читаю в вектор данные(1-е копирование). Для того, чтоб положить их в очередь, я копирую в новый вектор
размером = размеру прочитанного блока (2-е копирование). После этого кладу в очередь. Метод push контейнера (например deque) вызывает
копирующий конструктор (3-е копирование). Если из очереди читающий поток беред данные, то вызывается еще копирующий конструктор.
К сожалению по-другому никак (это 4 копирование). К этому блоку добавляется заголовок (тоесть 5-е копирование в новый буфер в котором есть заголовок)

Не многовато ли копирований ? Размер буфера может достигать 5 Мб. В результате ООООчень накладно.
Можно конечно использовать boost::shared_ptr<void> но не совсем понятно как выделить нужный объем памяти.
Тоесть как спросить у файла сколько он может мне однать данных. Да и к томуже нужно хранить
размер этого блока. Впринципе сам std::vector можно считать smart_pointer-ом в каком-то смысле,
т.к. он хранит размер и внутди памить выделяет динамически. Но он не ведет счетчик ссылок и поэтому 
это приводит к излишним операзиям копирования.

Какие будут идеи ?

чего то так все запутанно.... А почему нельзя просто выделять буфер BYTE* p_buffer = new[BUFLEN];, а дальше работать по адресу этого буфера? Если уж совсем по адресу не получается(я не знаю как std::vector делает операцию = для всего массива), то у по крайней мере стоит посмотреть в сторону ::memcpy(,p_buffer ,sizeof(BYTE)*BUFLEN) - весьма быстрая операция. Но все таки лучше применить адрес! А ещё лучше воспользоваться классом CByteArray и его ссылкой во все твои конструкторы. Если читаешь структуры, то нет ничего лучше CArray<твой тип,твой тип>, и тоже по адресу или по ссылке работай. А больше никаких идей у меня нет. Ах да - есть! Зачем тебе асинхронное чтение из файла?

а если положу указатель в контейнер, то потом не узнаю размер блока. Поэтому использую вектор.
Выделять изначально BUFLEN расточительно. Устройство может отдать и 100 байт вместо 5 Мб (максимально)


я не работал с ним, но подозреваю, что это тоже самое



если данных нет, то устройство может не отпустить. Или я не прав ? (надо тестить)

Во первых, CByteArray - это суть тот же вектор, поэтому ты можешь работать и по ссылке или адресу на вектор. Во-вторых, ну не отпустит и что с того? У тебя ведь на этой операции всего один ОТДЕЛЬНЫЙ поток повиснет - это никому не помешает особо. А когда надо будет завершить программу, то вызовешь что-то типа CancelIO и поставишь галочку на выход, чтобы у тебя снова в чтение не было захода(я так понимаю читаешь ты постоянно в цике while(!галочка){ReadFile();...}).
Да у тебя ведь несколько читателей. Но при работе с ReadFile (я про работу с COM портом) можно настроиться так, что функция ReadFile не ждет данных, а считывает что есть, а если нет ничего, возвращает 0 и все 

Добавлено @ 15:28 
Да, я понял, что ты не с ком портом работаешь, просто я функцию реадфиле там использовал

я тогда не смогу поток прибить при остановке читальщика с устройства


Он отменит операцию I/O ?

это не потры, а дровина