Блог

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

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

 SmartPtr ptr;

class Reader : public Thread {
    virtual void Run {
       for (;;) {
           SmartPtr local(ptr);
           // do smth   
       }
    }   
};

class Writer : public Thread {
    virtual void Run {
       for (;;) {
           SmartPtr newPtr(new Object);    
           ptr = newPtr;  
       }
    }
};

int main() {
    Pool* pool = SystemThreadPool();
    pool->Run(new Reader());
    pool->Run(new Writer());
    for (;;) // wait for crash :(
}
  

Когда я создаю локальную копию потока ptr , это означает, по крайней мере,

1. Считывает адрес.
2. Увеличить счетчик ссылок.

Я не могу выполнить эти две операции атомарно, и поэтому иногда мои читатели работают с удаленным объектом.

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

Для PowerPC я нашелhttp://drdobbs.com/184401888 , выглядит красиво, но использует инструкции, связанные с загрузкой, и инструкции с условным сохранением, которых у нас нет в x86.

Насколько я понимаю, указатели boost предоставляют такую функциональность только с использованием блокировок. Мне нужно решение без блокировки.

boost ::shared_ptr имеет atomic_store, который использует «безблокировочную» блокировку, которая должна быть достаточно быстрой для 99% возможных случаев.

     boost::shared_ptr<Object> ptr;
class Reader : public Thread {
    virtual void Run {
       for (;;) {
           boost::shared_ptr<Object> local(boost::atomic_load(amp;ptr));
           // do smth   
       }
    }   
};

class Writer : public Thread {
    virtual void Run {
       for (;;) {
           boost::shared_ptr<Object> newPtr(new Object);    
           boost::atomic_store(amp;ptr, newPtr);
       }
    }
};

int main() {
    Pool* pool = SystemThreadPool();
    pool->Run(new Reader());
    pool->Run(new Writer());
    for (;;)
}
  

Редактировать:

В ответ на комментарий ниже, реализация находится в «boost / shared_ptr.hpp»…

 template<class T> void atomic_store( shared_ptr<T> * p, shared_ptr<T> r )
{
    boost::detail::spinlock_pool<2>::scoped_lock lock( p );
    p->swap( r );
}

template<class T> shared_ptr<T> atomic_exchange( shared_ptr<T> * p, shared_ptr<T> r )
{
    boost::detail::spinlock amp; sp = boost::detail::spinlock_pool<2>::spinlock_for( p );

    sp.lock();
    p->swap( r );
    sp.unlock();

    return r; // return std::move( r )
}
  

С некоторыми хитростями вы должны быть в состоянии выполнить это, используя InterlockedCompareExchange128. Храните счетчик ссылок и указатель в массиве __int64 из 2 элементов. Если счетчик ссылок находится в массиве [0], а указатель в массиве [1], атомарное обновление будет выглядеть следующим образом:

 while(true)
{
    __int64 comparand[2];
    comparand[0] = refCount;
    comparand[1] = pointer;
    if(1 == InterlockedCompareExchange128(
        array,
        pointer,
        refCount   1,
        comparand))
    {
        // Pointer is ready for use. Exit the while loop.
    }
}
  

Если встроенная функция InterlockedCompareExchange128 недоступна для вашего компилятора, вы можете вместо этого использовать базовую инструкцию CMPXCHG16B, если вы не возражаете возиться с языком ассемблера.

Решение, предложенное RobH, не работает. У него та же проблема, что и в исходном вопросе: при доступе к объекту подсчета ссылок он, возможно, уже был удален.

Единственный способ, который я вижу для решения проблемы без глобальной блокировки (как в boost::atomic_store ) или условных инструкций чтения / записи, — это каким-то образом отсрочить уничтожение объекта (или общего объекта подсчета ссылок, если такая вещь используется). Итак, у зеннехоя есть хорошая идея, но его метод слишком небезопасен.

Я мог бы сделать это, сохранив копии всех указателей в потоке записи, чтобы автор мог контролировать уничтожение объектов:

 class Writer : public Thread {
    virtual void Run() {
        list<SmartPtr> ptrs; //list that holds all the old ptr values        

        for (;;) {
            SmartPtr newPtr(new Object);
            if(ptr)
                ptrs.push_back(ptr); //push previous pointer into the list
            ptr = newPtr;

            //Periodically go through the list and destroy objects that are not
            //referenced by other threads
            for(auto it=ptrs.begin(); it!=ptrs.end(); )
                if(it->refCount()==1)
                    it = ptrs.erase(it);
                else
                      it;
       }
    }
};
  

Однако все еще существуют требования к классу интеллектуального указателя. Это не работает с shared_ptr, поскольку операции чтения и записи не являются атомарными. Он почти работает с boost::intrusive_ptr. Назначение для intrusive_ptr реализовано следующим образом (псевдокод):

 //create temporary from rhs
tmp.ptr = rhs.ptr;
if(tmp.ptr)
    intrusive_ptr_add_ref(tmp.ptr);

//swap(tmp,lhs)
T* x = lhs.ptr;
lhs.ptr = tmp.ptr;
tmp.ptr = x;

//destroy temporary
if(tmp.ptr)
    intrusive_ptr_release(tmp.ptr);
  

Насколько я понимаю, единственное, чего здесь не хватает, — это ограничение памяти на уровне компилятора раньше lhs.ptr = tmp.ptr; . С добавлением этого, как чтение, так rhs и запись lhs будут потокобезопасными при строгих условиях: 1) архитектура x86 или x64 2) подсчет атомарных ссылок 3) rhs пересчет не должен равняться нулю во время назначения (гарантируется приведенным выше кодом записи) 4) запись только в один поток lhs (используя CAS, у вас может быть несколько записей).

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

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

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

в моем случае было бы достаточно 100 миллисекунд, я выбрал несколько секунд, чтобы быть в безопасности.