Warning: the code below is NOT multi read/write!
It'l work fine as long as you don't have multiple writes OR don't require resize. But with multiple writers, the code can enter a deadlock state. I'm revising the code and if I do manage to work around these issues, I'll post them in a linked question.
After, analysing what went wrong in lock-free job queue without size restriction (multiple read/write) I've come up with another solution, here's what it does:
require_lock_ is used to signal when we need to leave the lock-free setup and resize our buffers, so technically we don't have a lock-free setup**
lock_ this is the lock which is only used when we need to resize the buffers, so if you create your queue large enough, it is lock-free until all jobs are exhausted
concurrent_users_ keeps track of the number of users accessing the following members
read_, write_, size_ these are used to keep track of the number of jobs in the queue
storage_ this is a vector which allocates the space in which the jobs are stored. If this needs adjusting, a lock is used
*bitflag_ this is an array of bits, used to indicate which positions inside lookup_ are taken. This is kept in sync with storage_
Whenever a job is added, the following happens:
concurrent_users_is incremented to indicate we're about to touchstorage_andbitflag_require_lock_is checked to see if we need to acquire the mutexwrite_is increased to retrieve a unique index to which we will write our job- if this id is out of bounds, we try to resize our storage which calls for a mutex synchronization
- we store the job
- we mark the index in our array of bits, indicating this job is ready to use
When jobs are removed from the queue:
concurrent_users_is again incremented, note that this is automatically decreased when the guard goes out of scope- we check for the
require_lock_ read_is increased, just aswrite_was increased in the push function- if the received id is out of bounds, we return read_ to it's previous value and try a cleanup, since our job supply is exhausted
- we retrieve the job from our storage and remove the
bitflag_indicating the job is ready to use
As stated, this is not a lock-free setup, so technically it is not a solution for the initial question. However, it only locks when the storage is full, so if you really want to prevent mutex locking, you have to allocate enough room for the jobs you anticipate.
At this point, the code is not exception safe, I intent to change it so it guarantees a strong exception safety.
Also, if you never allow the job queue to become completely empty, you will end up with a lot of memory used for 'old' jobs and/or eventually will reach the maximum value of the write_ variable, at which point the code no longer accepts jobs until the queue is completely emptied.
The code should also be changed as to allow for a custom allocator, but these points I considered to be easy to implement once the lock-free setup was done.
I would love to hear your comments. I think it's rather nice, even though I was not able to make it completely lock-free and this solution feels a tiny bit like cheating.
- Is it really thread safe?
- Apart from the points I mentioned, what functionality could be improved/added?
fifo.h:
#pragma once
#include <atomic>
#include <memory>
#include <vector>
#include <mutex>
#include <thread>
#include <algorithm>
namespace lock_free
{
/**
* this class is used so we're able to use the RAII mechanism for locking
*/
template < typename T >
class use_count
{
public:
template < typename V >
use_count( V &&v ) :
data_( std::forward< V >( v ) ) { }
const T& operator()() const { return data_; }
void lock() { ++data_; }
void unlock() { --data_; }
private:
use_count( const use_count& );
use_count& operator = ( const use_count& );
T data_;
};
/**
* This is a lock free fifo, which can be used for multi-producer, multi-consumer
* type job queue
*/
template < typename Value >
class fifo
{
public:
typedef Value value_type;
fifo( size_t size = 1024 ) :
require_lock_( false ),
lock_(),
concurrent_users_( 0 ),
read_( 0 ),
write_( 0 ),
size_( size ),
storage_( size ),
bitflag_( new std::atomic_size_t[ std::max( size_t( 1 ), size / bits_per_section() ) ] )
{
fill_bitflags( 0 );
}
~fifo()
{
clear();
delete [] bitflag_;
}
/**
* pushes an item into the job queue, may throw if allocation fails
* leaving the queue unchanged
*/
void push( const value_type &value )
{
std::lock_guard< use_count< std::atomic_size_t > > lock( concurrent_users_ );
conditional_lock();
if ( write_ == std::numeric_limits< size_t >::max() )
{
throw std::logic_error( "fifo full, remove some jobs before adding new ones" );
}
const size_t id = write_++;
if ( id >= size_ )
{
resize_storage( id );
}
storage_[ id ] = value;
set_bitflag_( id, mask_for_id( id ) );
}
/**
* retrieves an item from the job queue.
* if no item was available, func is untouched and pop returns false
*/
bool pop( value_type &func )
{
auto assign = [ & ]( value_type &dst, value_type &src )
{
std::swap( dst, src );
};
return pop_generic( func, assign );
}
/**
* clears the job queue, storing all pending jobs in the supplied container.
* the container is also returned for convenience
*/
template < typename T >
T& pop_all( T &unfinished )
{
value_type tmp;
while ( pop( tmp ) )
{
unfinished.push_back( tmp );
}
return unfinished;
}
/**
* clears the job queue.
*/
void clear()
{
auto del = []( value_type&, value_type& ) {};
value_type tmp;
while ( pop_generic( tmp, del ) )
{
// empty
}
}
/**
* returns true if there are no pending jobs
*/
bool empty() const
{
return read_ == write_;
}
private:
fifo( const fifo& );
fifo& operator = ( const fifo& );
static constexpr size_t bits_per_section()
{
return sizeof( size_t ) * 8;
}
template < typename Assign >
bool pop_generic( value_type &value, Assign assign )
{
std::lock_guard< use_count< std::atomic_size_t > > lock( concurrent_users_ );
conditional_lock();
const size_t id = read_++;
if ( id >= write_ )
{
--read_;
try_cleanup();
return false;
}
const size_t mask = mask_for_id( id );
while ( !unset_bitflag_( id, mask ) )
{
std::this_thread::yield();
}
assign( value, storage_[ id ] );
return true;
}
void try_cleanup()
{
if ( !write_ || read_ != write_ || require_lock_ )
{
// early exit, avoids needless locking
return;
}
bool expected( false );
if ( require_lock_.compare_exchange_strong( expected, true ) )
{
std::lock_guard< std::mutex > guard( lock_ );
while ( concurrent_users_() > 1 )
{
std::this_thread::yield();
}
write_ = 0;
read_ = 0;
fill_bitflags( 0 );
require_lock_ = false;
}
}
void resize_storage( size_t id )
{
while ( size_ <= id )
{
if ( id == size_ )
{
require_lock_ = true;
std::lock_guard< std::mutex > guard( lock_ );
while ( concurrent_users_() > 1 )
{
std::this_thread::yield();
}
const size_t bitflag_size = size_ / bits_per_section();
storage_.resize( std::max( size_t( 1 ), size_ * 2 ) );
std::atomic_size_t *newbitflag = new std::atomic_size_t[ std::max( size_t( 1 ), bitflag_size * 2 ) ];
std::atomic_size_t *start = newbitflag;
const std::atomic_size_t *end = start + bitflag_size;
const std::atomic_size_t *src = bitflag_;
while ( start != end )
{
(start++)->store( *src++ );
}
end = newbitflag + bitflag_size * 2;
while ( start != end )
{
(start++)->store( 0 );
}
delete [] bitflag_;
bitflag_ = newbitflag;
size_ = storage_.size();
require_lock_ = false;
}
else
{
conditional_lock();
}
}
}
static size_t mask_for_id( size_t id )
{
const size_t offset = id / bits_per_section();
id -= offset * bits_per_section();
return size_t( 1 ) << id;
}
void set_bitflag_( size_t id, size_t mask )
{
bitflag_[ id / bits_per_section() ].fetch_or( mask );
}
bool unset_bitflag_( size_t id, size_t mask )
{
const size_t old = bitflag_[ id / bits_per_section() ].fetch_and( ~mask );
return ( old & mask ) == mask;
}
void conditional_lock()
{
if ( require_lock_ )
{
concurrent_users_.unlock();
lock_.lock();
lock_.unlock();
concurrent_users_.lock();
}
}
void fill_bitflags( size_t value )
{
std::atomic_size_t *start = bitflag_;
const std::atomic_size_t *end = start + size_ / bits_per_section();
while ( start != end )
{
(start++)->store( value );
}
}
std::atomic_bool require_lock_;
std::mutex lock_;
use_count< std::atomic_size_t > concurrent_users_;
std::atomic_size_t read_, write_, size_;
std::vector< value_type > storage_;
std::atomic_size_t *bitflag_;
};
}
