2
\$\begingroup\$

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 touch storage_ and bitflag_
  • require_lock_ is checked to see if we need to acquire the mutex
  • write_ 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 as write_ 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_;
    };
}
\$\endgroup\$

1 Answer 1

2
\$\begingroup\$

When you say this queue doesn't support multiple "writers", what do you mean? Do you mean multiple producers? Because both pushing (producing) and popping (consuming) from a queue are mutative ("writing") operations. And if you don't support the scenario of "one writer pushing items and one writer popping them", then you can't really call it a concurrent queue at all. So I'm going to assume that you meant to support "one producer, one consumer"... and then I'll show where the bug is with that scenario.


  • You only use use_count<T> with T = std::atomic_size_t, so it probably shouldn't be a class template at all (it should just be a plain old class).

  • If you keep it a template, the constructor should probably be a variadic template. Perfect forwarding and variadic templates go together like peanut butter and jelly. And the constructor should definitely be explicit, to disable unwanted implicit conversions.

    template <typename... Args>
    explicit use_count(Args&&... args) : data_(std::forward<Args>(args)...) { }
    
  • You declare a private copy-constructor and copy-assignment operator for use_count, but never define them. This C++03 idiom is obsolete as of C++11. The better way would be to not declare them at all (since having an atomic_size_t member will prevent the class from being copyable/movable anyway), or to define them as =delete.

  • Your fifo constructor should also be marked explicit.

  • bitflag_ should probably be a std::unique_ptr<std::atomic_size_t[]> instead of a raw pointer.

  • Your code would be more readable if you removed all the blank lines (except the ones between function definitions, I guess). For example, you wrote push() in 21 lines, 5 of which were blank and 2 of which were uncuddled {s. It could have been 14 lines.


Okay, here's the bug (I think; please let me know if I missed something). Suppose thread T1 is trying to resize the storage, and thread T2 is coming in as a new reader trying to pop.

T1: enters push
T1: increments concurrent_users_.data_ from 0 to 1
T1: calls conditional_lock, which returns without doing anything
T1: enters resize_storage with id == size_
T1: sets require_lock_ = true
T1: locks lock_
T1: tests concurrent_users_() > 1, which is false because concurrent_users_.data_ == 1
T1: therefore doesn't execute any iterations of the while loop
T1: a bunch of unsynchronized stuff culminating in delete [] bitflag_;

Meanwhile,

T2: enters pop
T2: enters pop_generic
T2: increments concurrent_users_.data_ from 1 to 2
T2: enters conditional_lock
T2: tests requires_lock_, which is true
T2: decrements concurrent_users_.data_ from 2 to 1
T2: acquires lock_ and releases it
T2: increments concurrent_users_.data_ from 1 to 2
T2: returns from conditional_lock
T2: a bunch of unsynchronized stuff culminating in unset_bitflag_

Let's interleave those so it's clear what's going on:

T1: enters push
T1: increments concurrent_users_.data_ from 0 to 1
T1: calls conditional_lock, which returns without doing anything
T1: enters resize_storage with id == size_
T1: sets require_lock_ = true

T2: enters pop
T2: enters pop_generic
T2: increments concurrent_users_.data_ from 1 to 2
T2: enters conditional_lock
T2: tests requires_lock_, which is true
T2: decrements concurrent_users_.data_ from 2 to 1
T2: acquires lock_ and releases it

T1: locks lock_
T1: tests concurrent_users_() > 1, which is false because concurrent_users_.data_ == 1
T1: therefore doesn't execute any iterations of the while loop
T1: a bunch of unsynchronized stuff culminating in delete [] bitflag_;

T2: increments concurrent_users_.data_ from 1 to 2
T2: returns from conditional_lock
T2: a bunch of unsynchronized stuff culminating in unset_bitflag_


So, you've got an unsynchronized data race on the memory pointed to by bitflag_. T1 could easily deallocate that memory before T2 looks at it, which means T2 could be reading garbage memory and/or stomping on memory that's already been reallocated and is now in use by another thread.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.