3
\$\begingroup\$

I'm looking for a code review on a class that I've written. It was my first attempt at lockless programming and I'd love any feedback I can get.

#pragma once

#include <atomic>
#include <thread>
#include <memory>

namespace Pro {
    namespace Util {
        /*! Thread Safe Lock Free Pipe 
         */

        template<typename T>
        class alignas(64) Pipe {
        protected:
            size_t push_position_;
            char pad_p[64 - (sizeof(size_t))]; // Producer C-Line
            size_t pop_position_;
            char pad_c[64 - (sizeof(size_t))]; // Consumer C-Line
            T* queue_;
            size_t queue_offset_;
            size_t capacity_;
            char pad_sr[64 - sizeof(T*) - (sizeof(size_t) * 2)]; // Shared Read C-Line
            std::atomic<size_t> size_; 
            char pad_sw[64 - sizeof(std::atomic<size_t>)]; // Shared Write C-Line 

            inline bool DecrementSize(){
                if(size_ <= 0)
                    return false;
                size_.fetch_sub(1);
                return true;
            }

        public:
            //! If 0 is passed this than the pips will not be initialized allowing an array of pipes
            //! to be made on the stack without initializing them before required
            //! For situations where avoiding heap creation is required for memory alignment
            //! This approach requires that the Pipe is initialised before use  
            Pipe(const size_t size) {
                queue_ = nullptr;
                if(size != 0)
                    Initialize(size);
            }

            Pipe(const Pipe& rhs) = delete; 
            Pipe& operator=(const Pipe& rhs) = delete;

            Pipe(Pipe&& rhs){
                queue_ = rhs.queue_;
                capacity_ = rhs.capacity_;
                pop_position_ = rhs.pop_position_;
                queue_offset_ = rhs.queue_offset_;
                push_position_ = rhs.push_position_;
                size_ = rhs.size_.load();

                rhs.queue_ = nullptr;
                rhs.size_ = 0;
                rhs.pop_position_ = 0;
                rhs.queue_offset_ = 0;
                rhs.push_position_ = 0;
                rhs.capacity_ = 0; 
            }

            Pipe& operator=(Pipe&& rhs){
                queue_ = rhs.queue_;
                capacity_ = rhs.capacity_;
                queue_offset_ = rhs.queue_offset_;
                pop_position_ = rhs.pop_position_;
                push_position_ = rhs.push_position_;
                size_ = rhs.size_.load();

                rhs.queue_ = nullptr;
                rhs.size_ = 0;
                rhs.pop_position_ = 0;
                rhs.queue_offset_ = 0;
                rhs.push_position_ = 0;
                rhs.capacity_ = 0;
                return *this;
            }

            ~Pipe() {
                Destroy();
            }  

            //! Initializes the Pipe if created without initialisation
            //! Nullop if the data pointer is already populated
            inline void Initialize(const size_t size) {
                if (queue_ == nullptr) {
                    queue_ = reinterpret_cast<T*>(operator new(sizeof(T) * size + 64));
                    queue_offset_ = 64 - ((size_t)queue_ % 64);
                    queue_ = (T*)((char*)queue_ + queue_offset_);
                    capacity_ = size;
                    pop_position_ = push_position_ = size_ = 0;
                }
            }

            //! A destroyed Pipe may be reused after calling Initialize() again.
            //! There is no check inplace to determine if the Pipe is initialised for pop and pushes 
            inline void Destroy() {
                if (queue_ == nullptr)
                    return;
                while (Pop());
                queue_ = (T*)((char*)queue_ - queue_offset_);
                operator delete(queue_);
                queue_ = nullptr;
            }

            // Issue if pushing while poping with only one element
            inline bool Push(const T& obj) {
                if(capacity_ - size_ == 0)
                    return false;

                auto pos = push_position_++ % capacity_; 
                new(reinterpret_cast<T*>(queue_) + pos) T(obj);
                size_.fetch_add(1);
                return true;
            }

            //! Pushes a new objects into the queue
            inline bool Push(T&& obj) {
                if(capacity_ - size_ == 0)
                    return false;

                auto pos = push_position_++ % capacity_; 
                new(reinterpret_cast<T*>(queue_) + pos) T(std::move(obj));
                size_.fetch_add(1);
                return true;
            }


            //! Pushes a new objects into the queue, constructed with the provided arguments
            template<typename... Args>
            inline bool Emplace(Args... arguments) {
                if(capacity_ - size_  == 0)
                    return false;

                auto pos = push_position_++ % capacity_; 
                new(reinterpret_cast<T*>(queue_) + pos) T(arguments...);
                size_.fetch_add(1);
                return true;
            }

            //! Returns a copy to the next object being stored.
            //! Top doesn't wait for a resize to finish, due to its read only nature.
            inline const bool Top(const T& return_obj) const{
                if(size_ == 0)
                    return false;

                auto pos = pop_position_ % capacity_;
                return_obj = queue_[pos];
                return true;
            }

            //! Returns a copy to the next object being stored.
            inline bool Top(T& return_obj) {
                if(size_ == 0)
                    return false;

                auto pos = pop_position_ % capacity_;
                return_obj = queue_[pos];
                return true;
            }

            //! Removes the next object from the queue.
            inline bool Pop() {
                if(size_ == 0)
                    return false;

                // Get the next object
                auto pos = pop_position_++ % capacity_;
                // Deconstruct object
                (queue_ + pos)->~T();
                // Top doesn't increment, so do that here
                return DecrementSize();
            }

            //! Returns and removes the next object from the queue.
            //! if the queue is empty, false is returned and return_obj isn't modified
            inline bool TopPop(T& return_obj) {
                if(size_ == 0)
                    return false;

                // Get the next object
                size_t pos = pop_position_++ % capacity_;
                // Create a copy before updating size, to prevent push from overwriting out object
                return_obj = *(queue_ + pos);
                return DecrementSize();
            }

            inline bool Empty() const { return size_ == 0; }

            //! Returns the count of objects being stored
            inline size_t size() const { return size_; }

            //! Returns the capacity of the queue before a resize is required.
            inline size_t capacity() const { return capacity_; }

        };
    }
}
\$\endgroup\$
3
\$\begingroup\$

Move Operators:

Much easier to write in terms of swap. Then you only need to write it once (in the swap method).

Also Best to mark move constructor and move assignment as noexcept (assuming they are). If you place them in containers it allows them to be moved otherwise containers will attempt to use copy semantics so that they get strong exception guarantees.

        Pipe(Pipe&& rhs)            noexcept
            : queue_(nullptr;
            , size_(0)
            , pop_position_(0)
            , queue_offset_(0)
            , push_position_(0)
            , capacity_(0)
        {
            rhs.swap(*this);
        }

        Pipe& operator=(Pipe&& rhs) noexcept
        {
            rhs.swap(*this);
        }

        void swap(Pipe& other)      noexcept
        {
            using std::swap;
            swap(queue_, other.queue_);
            swap(size_, other.size_);
            swap(pop_position_, other.pop_position_);
            swap(queue_offset_, other.queue_offset_);
            swap(push_position_, other.push_position_);
            swap(capacity_, other.capacity_);
        }

You can then write your own swap very easily.

  void swap(Pipe& lhs, Pipe& rhs) {
      lhs.swap(rhs);
  }

Variable declaration

Variable declaration order is important.
Objects will always be constructed in the order they are declared inside the class.

\$\endgroup\$
2
\$\begingroup\$

Include guards

It's worth mentioning that #pragma once isn't C++ standard, and likely won't be for a long while. A better alternative would be to use include guards, like this:

#ifndef LOCKLESS_SCSP_QUEUE_H_
#define LOCKLESS_SCSP_QUEUE_H_

// Original code goes here.

#endif

If you want to use both, you can always do something like this as well:

#ifndef LOCKLESS_SCSP_QUEUE_H_
#define LOCKLESS_SCSP_QUEUE_H_
#pragma once    

// Original code goes here.

#endif

Using include guards ensure that code using #pragma once will work properly on C++ compilers that don't support #pragma once.

It's worth noting that you should choose unique names for your include guards, and make them nice and long.


Using auto

I'm not a fan of using auto, so this is a little opinion based, but to quote a recent chat conversation with user @Mast:

@EthanBierlein The advantage of C++ in my opinion is you always know what you get.

There's no magic involved.

No assumptions.

No implicit type casting.

You all throw that overboard when you start auto-ing.

If I wanted an auto I'd write Python.

This is just my personal opinion on auto though, so it's really up to you whether you want to use auto or not.

If you want more info on this subject matter, you can take a look at the following links:


Nitpicks

  • I find this section of code particularly hard to read:

    size_t push_position_;
    char pad_p[64 - (sizeof(size_t))]; // Producer C-Line
    size_t pop_position_;
    char pad_c[64 - (sizeof(size_t))]; // Consumer C-Line
    T* queue_;
    size_t queue_offset_;
    size_t capacity_;
    char pad_sr[64 - sizeof(T*) - (sizeof(size_t) * 2)]; // Shared Read C-Line
    std::atomic<size_t> size_; 
    char pad_sw[64 - sizeof(std::atomic<size_t>)]; // Shared Write C-Line
    

    I'd re-organize it by variable type, and add some blank lines, to help with readability, like this:

    size_t push_position_;
    size_t pop_position_;
    size_t queue_offset_;
    size_t capacity_;
    
    char pad_p[64 - (sizeof(size_t))]; // Producer C-Line
    char pad_c[64 - (sizeof(size_t))]; // Consumer C-Line
    char pad_sw[64 - sizeof(std::atomic<size_t>)]; // Shared Write C-Line
    char pad_sr[64 - sizeof(T*) - (sizeof(size_t) * 2)]; // Shared Read C-Line
    
    std::atomic<size_t> size_; 
    T* queue_;
    
  • Why is the function DecrementSize called DecrementSize? I don't see it doing any decrementing of any kind. This should be changed.
  • Finally, just as a good habit, you should use curly braces whenever they're needed. It reduces the chance of a nasty error, like the famous Apple SSL bug. This means that something like this:

    if(size_ == 0)
        return false;
    

    Should become something like this:

    if(size_ == 0) {
        return false;
    }
    
\$\endgroup\$
  • \$\begingroup\$ The order of the member declarations isn't innocent. Looks like an attempts at cache-line aligning some of them. \$\endgroup\$ – Mat Sep 5 '15 at 5:49
  • \$\begingroup\$ Inside the function DecrementSize there's a function call to fetch_sub which will decrement an atomic variable. Also the formatting of the variables isn't avoidable since their placement ensures at least 64byte of separation which is required to stop False Sharing of cache between two CPU cores. I know it makes it harder to read, but I don't think I have a better option. I do agree that I've overused auto here, I usually only use it in for each loops and where the type is obvious. \$\endgroup\$ – Joshua Waring Sep 5 '15 at 5:56
  • \$\begingroup\$ You might also want to add this article on auto to your list. Also, don't think about it like a var in JavaScript or similar. auto doesn't mean a generic "variant" type, but rather, it is supposed to avoid having to repeat type declarations, making the code DRYer. But of course, like any other language feature, it can be overused and abused. \$\endgroup\$ – glampert Sep 5 '15 at 18:28

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

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