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

Ответ 1

В общем случае безопасно использовать контейнеры STL таким образом. Вам нужно будет реализовать специальный метод, чтобы сделать ваш поток кода безопасным. Решение, которое вы выбрали, зависит от ваших потребностей. Я бы, вероятно, решил это, сохранив два списка: по одному в каждом потоке. И передача изменений через блокировку бесплатной очереди (упоминается в комментариях к этому вопросу). Вы также можете ограничить время жизни ваших объектов Info, обернув их в boost:: shared_ptr, например.

typedef boost::shared_ptr<Info> InfoReference; 
typedef std::list<InfoReference> InfoList;

enum CommandValue
{
    Insert,
    Delete
}

struct Command
{
    CommandValue operation;
    InfoReference reference;
}

typedef LockFreeQueue<Command> CommandQueue;

class Thread1
{
    Thread1(CommandQueue queue) : m_commands(queue) {}
    void run()
    {
        while (!finished)
        {
            //Process Items and use 
            // deleteInfo() or addInfo()
        };

    }

    void deleteInfo(InfoReference reference)
    {
        Command command;
        command.operation = Delete;
        command.reference = reference;
        m_commands.produce(command);
    }

    void addInfo(InfoReference reference)
    {
        Command command;
        command.operation = Insert;
        command.reference = reference;
        m_commands.produce(command);
    }
}

private:
    CommandQueue& m_commands;
    InfoList m_infoList;
}   

class Thread2
{
    Thread2(CommandQueue queue) : m_commands(queue) {}

    void run()
    {
        while(!finished)
        {
            processQueue();
            processList();
        }   
    }

    void processQueue()
    {
        Command command;
        while (m_commands.consume(command))
        {
            switch(command.operation)
            {
                case Insert:
                    m_infoList.push_back(command.reference);
                    break;
                case Delete:
                    m_infoList.remove(command.reference);
                    break;
            }
        }
    }

    void processList()
    {
        // Iterate over m_infoList
    }

private:
    CommandQueue& m_commands;
    InfoList m_infoList;
}   


void main()
{
CommandQueue commands;

Thread1 thread1(commands);
Thread2 thread2(commands);

thread1.start();
thread2.start();

waitforTermination();

}

Это не скомпилировано. Вам все равно нужно убедиться, что доступ к вашим объектам Info является потокобезопасным.

Ответ 2

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

Вам нужна "ограниченная блокирующая очередь". Не пишите сами, дизайн замка не является тривиальным. Трудно найти хорошие примеры, большинство из них - это .NET-код. Эта статья выглядит многообещающе.

Ответ 3

Я хотел бы знать, какова цель этого списка, тогда было бы легче ответить на вопрос.

Как сказал Хоар, как правило, плохая идея - попытаться обмениваться данными для обмена данными между двумя потоками, а вам следует обмениваться данными: сообщениями.

Если этот список моделирует очередь, например, вы можете просто использовать один из различных способов связи (например, сокеты) между двумя потоками. Потребитель/производитель - это стандартная и хорошо известная проблема.

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

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

Ответ 4

Чтобы ваш итератор не был признан недействительным, вы должны заблокировать весь цикл for. Теперь я думаю, что первый поток может испытывать трудности с обновлением списка. Я постараюсь дать ему возможность выполнять свою работу на каждой (или на каждой N-й итерации).

В псевдокоде, который будет выглядеть так:

mutex_lock();
for(...){
  doAction();
  mutex_unlock();
  thread_yield();  // give first thread a chance
  mutex_lock();
  if(iterator_invalidated_flag) // set by first thread
    reset_iterator();
}
mutex_unlock();

Ответ 5

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

Ответ 6

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

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

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

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

Вы работаете на С++, и вы упомянули о необходимости очереди с операцией pop-if-not-empty. Несколько лет назад я написал двухзарядную очередь, используя ACE Library concurrency, так как Boost thread library еще не была готова для использования в производстве, а шанс для стандартной библиотеки С++ включая такие объекты, была далекой мечтой. Переносить его на нечто более современное было бы легко.

Эта моя очередь - concurrent::two_lock_queue - позволяет получить доступ к главе очереди только через RAII. Это гарантирует, что приобретение замка для чтения головки всегда будет сопрягаться с выпуском замка. Потребитель конструирует a const_front (const доступ к элементу head), a front (неконстантный доступ к элементу головы) или объект renewable_front (неконстантный доступ к головам и преемникам) для представления эксклюзивного право доступа к элементу головы очереди. Такие "передние" объекты не могут быть скопированы.

Класс two_lock_queue также предлагает функцию pop_front(), которая ждет, пока, по крайней мере, один элемент не будет доступен для удаления, но в соответствии с std::queue и std::stack стиль не смешивать мутацию контейнера и копирование значений, pop_front() возвращает void.

В сопутствующем файле есть тип concurrent::unconditional_pop, который позволяет обеспечить через RAII элемент заголовка очереди будет вытолкнут после выхода из текущей области.

Сопутствующий файл error.hh определяет исключения, возникающие из-за использования функции two_lock_queue::interrupt(), используемой для разблокирования потоков, ожидающих доступа к главе очереди.

Взгляните на код и сообщите мне, если вам нужно больше объяснений относительно того, как его использовать.

Ответ 7

Лучший способ сделать это - использовать контейнер, который внутренне синхронизирован. TBB и Microsoft concurrent_queue делают это. Энтони Уильямс также имеет хорошую реализацию в своем блоге здесь

Ответ 8

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

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

Жизнь была бы проще, если бы вы использовали очередь вместо списка, и попросите своего потребителя использовать синхронизированный вызов queue<Info>::pop_front() вместо итераторов, которые могут быть признаны недействительными за вашей спиной. Если вашему потребителю действительно нужно собирать куски Info за раз, используйте переменную состояния , которая сделает ваш потребительский блок до queue.size() >= minimum.

Библиотека Boost имеет приятную переносимую переменную состояния (которая работает даже со старыми версиями Windows), а также обычный материал библиотеки потоков.

Для очереди производителей-потребителей, использующей (старомодную) блокировку, проверьте BlockingQueue шаблонный класс библиотека ZThreads. Я сам не использовал ZThreads, беспокоясь о нехватке последних обновлений и потому, что он, похоже, не был широко использован. Тем не менее, я использовал его в качестве вдохновения для создания собственной потоковой очереди производителей-потребителей (прежде чем я узнал о lock-free очередях и TBB).

Блокированная библиотека очереди/стека, похоже, находится в очереди проверки Boost. Будем надеяться, что в ближайшем будущем мы увидим новый Boost.Lockfree!:)

Если есть интерес, я могу написать пример блокирующей очереди, в которой используется std:: queue и блокировка потока Boost.

ИЗМЕНИТЬ

В блоге, на который ссылается ответ Рика, уже есть пример блокирующей очереди, в котором используются std:: queue и Boost condvars. Если вашему потребителю нужно сожрать куски, вы можете расширить пример следующим образом:

void wait_for_data(size_t how_many)
    {
        boost::mutex::scoped_lock lock(the_mutex);
        while(the_queue.size() < how_many)
        {
            the_condition_variable.wait(lock);
        }
    }

Вы также можете настроить его, чтобы разрешить тайм-ауты и отмену.

Вы упомянули, что скорость была проблемой. Если ваш Info - тяжеловес, вам следует рассмотреть возможность передачи их shared_ptr. Вы также можете попытаться сделать свой фиксированный размер Info и использовать пул памяти (который может быть намного быстрее, чем куча).

Ответ 9

Если вы используете С++ 0x, вы можете внутренне синхронизировать итерацию списка таким образом:

Предполагая, что класс имеет шаблонный список с именем objects_, а boost:: mutex с именем mutex _

Метод toAll - это метод-член обертки списка

 void toAll(std::function<void (T*)> lambda)
 {
 boost::mutex::scoped_lock(this->mutex_);
 for(auto it = this->objects_.begin(); it != this->objects_.end(); it++)
 {
      T* object = it->second;
      if(object != nullptr)
      {
                lambda(object);
           }
      }
 }

Призвание:

synchronizedList1->toAll(
      [&](T* object)->void // Or the class that your list holds
      {
           for(auto it = this->knownEntities->begin(); it != this->knownEntities->end(); it++)
           {
                // Do something
           }
      }
 );

Ответ 10

Вы должны использовать некоторую библиотеку потоков. Если вы используете Intel TBB, вы можете использовать concurrent_vector или concurrent_queue. См. this.

Ответ 11

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

Я бы серьезно подумал Интерфейс передачи сообщений (MPI) вместо того, чтобы катить свое многопоточное решение. Есть доступная реализация С++ - OpenMPI, Boost.MPI, Microsoft MPI и т.д. и т.д.

Ответ 12

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

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

Что-то по строкам:

while (processItems) {
  Info item;
  lock(mutex);
  if (!infoList.empty()) {
     item = infoList.front();
     infoList.pop_front();
  }
  unlock(mutex);
  DoAction(item);
  delayALittle();
}

И функция вставки все равно должна выглядеть так:

lock(mutex);
infoList.push_back(item);
unlock(mutex);

Если очередь не будет массивной, у меня возникнет соблазн использовать что-то вроде std::vector<Info> или даже std::vector<boost::shared_ptr<Info> >, чтобы свести к минимуму копирование объектов Info (при условии, что это несколько дороже для копирования по сравнению для boost:: shared_ptr. Обычно операции над вектором имеют тенденцию быть немного быстрее, чем в списке, особенно если объекты, хранящиеся в векторе, являются маленькими и дешевыми для копирования.