template<typename T>
        class interlocked_circle_buffer
        {
        protected:
            enum cell_status_enum 
            {
                cell_Free = 0,
                cell_Filling,
                cell_Full,
                cell_Reading,
                cell_Read
            };
            struct cell_struct    {
                volatile long status;
                T data;
                cell_struct(): status(cell_Free){}
            };
        public:
            interlocked_circle_buffer(size_t size) : 
                m_size(size),
                m_buffer(new cell_struct[size]),
                m_writing(0),
                m_reading(0),
                m_read(0)
                {}
            ~interlocked_circle_buffer(){delete [] m_buffer;}
            bool push(const T& value)
            {
                long write;
                bool searching = true;
                while (searching)
                {
                    write = m_writing;
                    long cellStatus = InterlockedCompareExchange(&m_buffer[write].status, cell_Filling, cell_Free);
                    switch (cellStatus)
                    {
                        case cell_Free:
                            searching = false;
                        case cell_Filling:
                            InterlockedCompareExchange(&m_writing, (write+1) % m_size, write);
                            break;
                        default:
                            return false;
                    }            
                }
                m_buffer[write].data = value; 
                m_buffer[write].status = cell_Full;
                return true;
            }
            bool pop(T * target)
            {
                long read;
                bool searching = true;
                while (searching)
                {
                    read = m_reading;
                    long cellStatus = InterlockedCompareExchange(&m_buffer[read].status, cell_Reading, cell_Full);
                    switch (cellStatus)
                    {
                    case cell_Full:
                        searching = false;
                    case cell_Reading:
                    case cell_Read:
                        InterlockedCompareExchange(&m_reading, (read+1) % m_size, read);
                        break;
                    default:
                        return false;
                    }
                }
                *target = m_buffer[read].data;
                m_buffer[read].status = cell_Read;
                if (InterlockedCompareExchange(&m_read, (read+1) % m_size, read) == read)
                {
                    while (InterlockedCompareExchange(&m_buffer[read].status, cell_Free, cell_Read) == cell_Read)
                    {
                        if (InterlockedCompareExchange(&m_read, (read+1) % m_size, read) != read)
                            break;
                        read = (read+1) % m_size;
                    }
                }
                return true;
            }
        protected: // member fields
            cell_struct * m_buffer;
            size_t m_size;
            volatile long m_writing, m_reading, m_read;
        };

Здравствуйте, Caracrist, Вы писали:

C>>>Вот тут лежит на с++.
C>>>http://www.rsdn.ru/forum/src/3725942.aspx

Автор: Caracrist
Дата: 05.03.10

C>>>Перевести на с# можно почти один к одному.

R>>За пару минут просмотра — 3 ошибки.
C>
C>Просвети

Не вопрос.

Допустим поток считывает m_writing в push(), но перед тем как сделать InterlockedCompareExchange() он прерывается. Пока он спит ситуация под его ногами полностью меняется — вся очередь заполняется/опустошается один или несколько раз. Теперь поток просыпается и имеем 2 неприятных сценария развития событий.
Первый — поток видит статус cell_Read, и возвращает false в то время как очередь не полная. Т.е. push() может спонтанно возвращать false.
Вторая — поток видит статус cell_Free но не в хвосте очереди, а в произвольном пустом месте. Это приведёт к тому, что элемент будет потреблён очень-очень не скоро и не в FIFO порядке и рабочие потоки будут спать пока он там лежит (думать, что очередь пустая), или ещё хуже приведёт к дедлоку, если это было критическое сообщение.

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

Функция pop() вообще выглядит слишком сложной, что бы быть корректной. В ней множество промежуточных состояний, а такие состояния обычно бывают достаточно проблематичны, т.к. между ними могут вклиниться другие потоки.
Практически все InterlockedCompareExchange() могут напороться на ABA. Допустим поток прерван перед "InterlockedCompareExchange(&m_reading, (read+1) % m_size, read)", очередь делает полный оборот, поток просыпается и смещает m_reading на 1, хотя этот элемент никто не потребил (его статус cell_Full). Думаю это может привести к дедлоку в push(), т.к. встретится элемент в cell_Full, который никто не собирается переводить в cell_Free. Такая же проблема может быть и в push().
А последняя часть pop() просто выглядит подозрительно.

Здравствуйте, remark, Вы писали:

R>Здравствуйте, Caracrist, Вы писали:

C>>>>Вот тут лежит на с++.
C>>>>http://www.rsdn.ru/forum/src/3725942.aspx

Автор: Caracrist
Дата: 05.03.10

C>>>>Перевести на с# можно почти один к одному.

R>>>За пару минут просмотра — 3 ошибки.
C>>
C>>Просвети

R>Не вопрос.

R>

R>Возможно всё ещё хуже, чем ты думаешь

Спасибо, не хуже. Я уже понял, что в целом, вся концепция — крах.

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

        template<typename T>
        class interlocked_circle_buffer
        {
        protected:
            enum cell_status_enum 
            {
                cell_Free = 0,
                cell_Written,
                cell_Full,
                cell_Read
            };
            struct cell_struct    {
                volatile long status;
                T data;
                cell_struct(): status(cell_Free){}
            };
        public:
            interlocked_circle_buffer(size_t size) : 
                m_size(size),
                m_buffer(new cell_struct[size]),
                m_writing(-1),
                m_reading(-1),
                m_read(0),
                m_written(0),
                m_filled(0),
                m_filling(0)
                {}
            ~interlocked_circle_buffer(){delete [] m_buffer;}
            bool push(const T& value)
            {
                if (InterlockedIncrement(&m_filling) > m_size)
                {
                    InterlockedDecrement(&m_filling);
                    return false;
                }
                long write = InterlockedIncrement(&m_writing) % m_size;
                m_buffer[write].data = value; 
                m_buffer[write].status = cell_Written;
                finalize_write();
                return true;
            }
            bool pop(T * target)
            {
                if (InterlockedDecrement(&m_filled) < 0)
                {
                    InterlockedIncrement(&m_filled);
                    return false;
                }
                long read = InterlockedIncrement(&m_reading) % m_size;
                *target = m_buffer[read].data;
                m_buffer[read].status = cell_Read;
                finalize_read();
                return true;
            }
        private:
            void finalize_read()
            {
                while (true)
                {
                    if (long read = InterlockedExchange(&m_read, -1) != -1)
                    {
                        if (InterlockedCompareExchange(&m_buffer[read].status, cell_Free, cell_Read) == cell_Read)
                        {
                            m_read = (read + 1) % m_size;
                            InterlockedDecrement(&m_filling);
                            continue;
                        }
                        else 
                        {
                            m_read = read;
                            if (InterlockedCompareExchange(&m_buffer[read].status, cell_Read, cell_Read) == cell_Read)
                                continue;
                        }
                    }
                    break;
                }
                if (m_reading > MAXLONG / 2)
                    InterlockedOperation(&m_reading, x % m_size);
            }
            void finalize_write()
            {
                while (true)
                {
                    if (long write = InterlockedExchange(&m_written, -1) != -1)
                    {
                        if (InterlockedCompareExchange(&m_buffer[write].status, cell_Full, cell_Written) == cell_Written)
                        {
                            m_written = (write + 1) % m_size;
                            InterlockedIncrement(&m_filled);
                            continue;
                        }
                        else 
                        {
                            m_written = write;
                            if (InterlockedCompareExchange(&m_buffer[write].status, cell_Written, cell_Written) == cell_Written)
                                continue;
                        }
                    }
                    break;
                }
                if (m_writing > MAXLONG / 2)
                    InterlockedOperation(&m_writing, x % m_size);
            }
        protected: // member fields
            cell_struct * m_buffer;
            long m_size;
            volatile long m_writing, m_reading
                , m_written, m_read
                , m_filling, m_filled;

        };

Здравствуйте, Caracrist, Вы писали:

Нашел маленький oops ,

        template<typename T>
        class interlocked_circle_buffer
        {
        protected:
            enum cell_status_enum 
            {
                cell_Free = 0,
                cell_Written,
                cell_Full,
                cell_Read
            };
            struct cell_struct    {
                volatile long status;
                T data;
                cell_struct(): status(cell_Free){}
            };
        public:
            interlocked_circle_buffer(size_t size) : 
                m_size(size),
                m_buffer(new cell_struct[size]),
                m_writing(-1),
                m_reading(-1),
                m_read(0),
                m_written(0),
                m_filled(0),
                m_filling(0)
                {}
            ~interlocked_circle_buffer(){delete [] m_buffer;}
            bool push(const T& value)
            {
                if (InterlockedIncrement(&m_filling) > m_size)
                {
                    InterlockedDecrement(&m_filling);
                    return false;
                }
                long write = InterlockedIncrement(&m_writing) % m_size;
                m_buffer[write].data = value; 
                m_buffer[write].status = cell_Written;
                finalize_write();
                return true;
            }
            bool pop(T * target)
            {
                if (InterlockedDecrement(&m_filled) < 0)
                {
                    InterlockedIncrement(&m_filled);
                    return false;
                }
                long read = InterlockedIncrement(&m_reading) % m_size;
                *target = m_buffer[read].data;
                m_buffer[read].status = cell_Read;
                finalize_read();
                return true;
            }
        private:
            void finalize_read()
            {
                while (true)
                {
                    long read = InterlockedExchange(&m_read, -1);
                    if (read != -1)
                    //if (long read = InterlockedExchange(&m_read, -1) != -1) // oops 
                    {
                        if (InterlockedCompareExchange(&m_buffer[read].status, cell_Free, cell_Read) == cell_Read)
                        {
                            m_read = (read + 1) % m_size;
                            InterlockedDecrement(&m_filling);
                            continue;
                        }
                        else 
                        {
                            m_read = read;
                            if (InterlockedCompareExchange(&m_buffer[read].status, cell_Read, cell_Read) == cell_Read)
                                continue;
                        }
                    }
                    break;
                }
                if (m_reading > MAXLONG / 2)
                    InterlockedOperation(&m_reading, x % m_size);
            }
            void finalize_write()
            {
                while (true)
                {
                    long write = InterlockedExchange(&m_written, -1);
                    if (write != -1)
                    //if (long write = InterlockedExchange(&m_written, -1) != -1) // oops 
                    {
                        if (InterlockedCompareExchange(&m_buffer[write].status, cell_Full, cell_Written) == cell_Written)
                        {
                            m_written = (write + 1) % m_size;
                            InterlockedIncrement(&m_filled);
                            continue;
                        }
                        else 
                        {
                            m_written = write;
                            if (InterlockedCompareExchange(&m_buffer[write].status, cell_Written, cell_Written) == cell_Written)
                                continue;
                        }
                    }
                    break;
                }
                if (m_writing > MAXLONG / 2)
                    InterlockedOperation(&m_writing, x % m_size);
            }
        protected: // member fields
            cell_struct * m_buffer;
            long m_size;
            volatile long m_writing, m_reading
                , m_written, m_read
                , m_filling, m_filled;

        };

Потестировал всё это дело на скорость... Быстре просто лочить по тупому и всё.
Interlocked быстрее только если буфер маленький относительно количества потоков, а операция копирования дорогая...

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

Здравствуйте, Caracrist, Вы писали:

C>Потестировал всё это дело на скорость... Быстре просто лочить по тупому и всё.
C>Interlocked быстрее только если буфер маленький относительно количества потоков, а операция копирования дорогая...

C>Вывод: лочить простым мутексом и копировать указатели. Меньше шансов на баги и производительность супер.

Преимущество lock-free не в скорости лока, а в отсутствии ожидания потоков друг друга. Напимер код

EnterCriticalSection
a++;
LeaveCriticalSection

на x86 всего лишь в два раза медленнее, чем InterlockedIncrement, даже без spincount, но только если секция в момент захвата свободна. Если же потоки часто "пересекаются" на общем ресурсе, то lock-free даст лучшую производительность системы за счёт отсутствия ожидания потоками доступа к ресурсу, в том числе и за счёт уменьшения переключений контекстов потоков. Последнее, правда, можно несколько снивелировать, предваряя усыпление потока spin-lock'ом.

Здравствуйте, Jolly Roger, Вы писали:

JR>Здравствуйте, Caracrist, Вы писали:

C>>Потестировал всё это дело на скорость... Быстре просто лочить по тупому и всё.
C>>Interlocked быстрее только если буфер маленький относительно количества потоков, а операция копирования дорогая...

C>>Вывод: лочить простым мутексом и копировать указатели. Меньше шансов на баги и производительность супер.

JR>Преимущество lock-free не в скорости лока, а в отсутствии ожидания потоков друг друга. Напимер код

JR>

JR>EnterCriticalSection
JR>a++;
JR>LeaveCriticalSection
JR>

JR>на x86 всего лишь в два раза медленнее, чем InterlockedIncrement, даже без spincount, но только если секция в момент захвата свободна. Если же потоки часто "пересекаются" на общем ресурсе, то lock-free даст лучшую производительность системы за счёт отсутствия ожидания потоками доступа к ресурсу, в том числе и за счёт уменьшения переключений контекстов потоков. Последнее, правда, можно несколько снивелировать, предваряя усыпление потока spin-lock'ом.

Довёл до ума (если кто хочет посмотреть на это чудо, пишите )
После тестов с конечной версией, Interlocked быстрее в целом, в начале не значительно, но чем больше потоков тем больше разница. При 150+ потоках уже на порядки быстрее. Однако, есть ещё одно неприятное свойство: Interlocked вешает всю систему(win7 i7х64 4 x hyperthreading). Процессоры занимаются инвалидацией вместо выполнений других команд. Окна почти по пикселям рисуются.

Здравствуйте, Caracrist, Вы писали:

А что такое некий загадочный InterlockedOperation()?

Здравствуйте, remark, Вы писали:

R>Здравствуйте, Caracrist, Вы писали:

R>А что такое некий загадочный InterlockedOperation()?

R>

#ifndef InterlockedOperationEx
#define InterlockedOperationEx(targetPtr, operationExpr, finalizeOperation) do \
{ \
    long x(*(targetPtr)); \
    if (InterlockedCompareExchange(targetPtr,long(operationExpr),x) == x) finalizeOperation;\
} \
while(true)
#endif    

#ifndef InterlockedOperationReturn
#define InterlockedOperationReturn(targetPtr, operationExpr) \
    InterlockedOperationEx(targetPtr, operationExpr, return x)
#endif

#ifndef InterlockedOperation
#define InterlockedOperation(targetPtr, operationExpr) \
    InterlockedOperationEx(targetPtr, operationExpr, break)
#endif

Здравствуйте, Caracrist, Вы писали:

C>Пока что проблем не вижу, кроме теоретической возможности, что один поток застрянет в finalize_, однако это легко решается добавлением ограничения на максимум итераций (например: в один полный круг), однако вероятность этого настолько не велика, что я лично считаю, что такой счётчик это впустую потраченное процессорное время.

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

Засчитывается как 3:1

Здравствуйте, remark, Вы писали:

По-моему, если выполняющий Push низкоприоритетный поток заснёт сразу после
long write = InterlockedIncrement(&m_writing) % m_size;
но до выполнения
m_buffer[write].status = cell_Written;
то чтение из очереди окажется заблокированным до его пробуждения. Остальные писатели пройдут мимо InterlockedIncrement(&m_filled);, и Pop'ы будут всё время видеть m_filled равным нулю.

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

Здравствуйте, Jolly Roger, Вы писали:

JR>По-моему, если выполняющий Push низкоприоритетный поток заснёт сразу после
JR> long write = InterlockedIncrement(&m_writing) % m_size;
JR>но до выполнения
JR>m_buffer[write].status = cell_Written;
JR>то чтение из очереди окажется заблокированным до его пробуждения. Остальные писатели пройдут мимо InterlockedIncrement(&m_filled);, и Pop'ы будут всё время видеть m_filled равным нулю.

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

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

отсутсвует даже базовая гарантия в pop'е

Здравствуйте, Sni4ok, Вы писали:

S>отсутсвует даже базовая гарантия в pop'е

Она там есть.

П.С. я тоже так утверждать умею.
П.П.С. см. http://www.rsdn.ru/forum/src/3726883.1.aspx

Автор: Caracrist
Дата: 06.03.10

Здравствуйте, Caracrist, Вы писали:

S>>отсутсвует даже базовая гарантия в pop'е

C>Она там есть.

C>П.С. я тоже так утверждать умею.
C>П.П.С. см. http://www.rsdn.ru/forum/src/3726883.1.aspx

Автор: Caracrist
Дата: 06.03.10

    long read = InterlockedIncrement(&m_reading) % m_size;
    *target = m_buffer[read].data; //если тут произойдёт исключение, то колл. читателей будет неверно.
    m_buffer[read].status = cell_Read;
    finalize_read();

Здравствуйте, remark, Вы писали:


R>Мне кажется, что с фиксированным буфером вообще не получится сделать lock-free очередь, т.к. надо атомарно записать данные и сдвинуть указатель. lock-free очередь может быть только на основе связанного списка.

Мне кажется, такая очередь имеет смысл только если сами сохраняемые данные являются элементом списка. Если под каждое сохранение придётся выделять новый элемент, то мы окажемся зависимыми от менеджера памяти, либо придётся создавать ещё один пул свободных элементов, или даже несколько — свой на каждый поток. Не окажется ли ограниченного размера очередь, защищённая банальной комбинацией spin-lock'а с мьютексом более эффективной, как Вы думаете?

R>Кстати возможностью отслеживания смерти потоков под мьютексом пользуются очень редко, т.к. что бы это имело смысл алгоритм внутри критической секции должен быть полностью lock-free, т.е. теоретически lock-free.

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

Здравствуйте, Sni4ok, Вы писали:

S>Здравствуйте, Caracrist, Вы писали:

S>>>отсутсвует даже базовая гарантия в pop'е

C>>Она там есть.

C>>П.С. я тоже так утверждать умею.
C>>П.П.С. см. http://www.rsdn.ru/forum/src/3726883.1.aspx

Автор: Caracrist
Дата: 06.03.10

S>

S>    long read = InterlockedIncrement(&m_reading) % m_size;
S>    *target = m_buffer[read].data; //если тут произойдёт исключение, то колл. читателей будет неверно.
S>    m_buffer[read].status = cell_Read;
S>    finalize_read();
S>

Ура!
Наконец кто-то обратил внимание на это
Это легко решаемая проблема, весь необходимый finalize переносится в деструктор спецыального класса, добавляется ещё один флажок: cell_Broken
и всё работает как часы. однако такой код становится неудобо читаемым, и тут его анализировать(в виртуальной машине класса мозг) будет сложнее. Поэтому перед выкладкой я убрал весь код обработки ошибок.

Здравствуйте, Caracrist, Вы писали:


C>Наконец кто-то обратил внимание на это
C>Поэтому перед выкладкой я убрал весь код обработки ошибок.

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

Здравствуйте, Jolly Roger, Вы писали:

R>>Мне кажется, что с фиксированным буфером вообще не получится сделать lock-free очередь, т.к. надо атомарно записать данные и сдвинуть указатель. lock-free очередь может быть только на основе связанного списка.

JR>Мне кажется, такая очередь имеет смысл только если сами сохраняемые данные являются элементом списка. Если под каждое сохранение придётся выделять новый элемент, то мы окажемся зависимыми от менеджера памяти, либо придётся создавать ещё один пул свободных элементов, или даже несколько — свой на каждый поток. Не окажется ли ограниченного размера очередь, защищённая банальной комбинацией spin-lock'а с мьютексом более эффективной, как Вы думаете?


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

Теперь по поводу эффективности. Ну это сложный вопрос, зависит от многих факторов. Вообще, если мы говорим о производительности, то нужно использовать только интрузивные очереди, и тут никакого оверхеда обычно нет, т.к. сами данные-то нам всё равно скорее всего выделять динамически, ну и встроим туда указатель next. По поводу не интрузивных очередей — ну тут, да, надо делать по-поточные буферы сообщений — выделять и освобождать из таких буферов можно практически бесплатно. Есть и другие варианты, например, смотри вот эту очередь:
http://software.intel.com/en-us/articles/single-producer-single-consumer-queue
Там я сделал встроенный в очередь буфер узлов, фактически там 2 очереди — одна передаёт полезную нагрузку от производителя у потребителю, вторая — передаёт свободные узлы в обратном направлении. Передача осуществляется пачками, поэтому оверхед практически нулевой.

А по поводу очереди, защищенной спин-мьютексом — нет, не эффективнее.
Тут:
http://groups.google.com/group/lock-free/browse_frm/thread/c8e3201da4a6a300
как это сделать для интрузивной очереди на основе связанного списка.
Добавление в очередь — wait-free, один _InterlockedExchange(). Извлечение из очереди — lock-free, один _InterlockedCompareExchange64().
Скоро я тут покажу как это сделать для очереди на основе фиксированного буфера — один _InterlockedCompareExchange() на операцию. Спин-мьютекс не может быть быстрее, т.к. он подразумевает как минимум один _InterlockedCompareExchange().