pulsar-client-cpp/lib/ObjectPool.h

/**
 * Copyright 2016 Yahoo Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#ifndef LIB_OBJECTPOOL_H_
#define LIB_OBJECTPOOL_H_

#include <algorithm>
#include <boost/thread/locks.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/make_shared.hpp>
#include <boost/thread/tss.hpp>

namespace pulsar {

template<typename T, int MaxSize>
class Allocator {
    public:
    // Allocator must be stateless, so put everything in this static
    class Impl {
        public:
        // cheap lock to acquire
        static boost::mutex mutex_;

        // note: use std::forward_list<> when switching to C++11 mode
        struct Node {
            Node* next;
            explicit Node(Node* n) : next(n) {}
        };
        Node *head_;
        int pushSize_;

        struct GlobalPool {
            Node *node_;
            int nodeCount_;
            GlobalPool *next_;
            explicit GlobalPool(GlobalPool* n) : next_(n) {}
        };
        static struct GlobalPool *globalPool_;
        static int globalNodeCount_;

        Impl(const Impl&);
        void operator=(const Impl&);

        void* pop() {
            if (!head_) {
                // size = 0
                boost::lock_guard<boost::mutex> lock(mutex_);

                if (!globalPool_) {
                    return NULL;
                }

                GlobalPool *poolEntry = globalPool_;
                head_ = globalPool_->node_;
                pushSize_ += globalPool_->nodeCount_;
                globalNodeCount_ -= globalPool_->nodeCount_;
                globalPool_ = globalPool_->next_;
                delete poolEntry;

            }
            void* result = head_;
            if (result) {
                head_ = head_->next;
                pushSize_--;
            }
            return result;
        }

        bool push(void* p) {

            // Once thread specific entries reaches 10% of max size, push them to GlobalPool
            if (pushSize_ >= MaxSize*0.1) {

                bool deleteList = true;
                {
                    // Move the entries to global pool
                    boost::lock_guard<boost::mutex> lock(mutex_);

                    // If total node count reached max allowed cache limit,
                    // skip adding to global pool.
                    if ((globalNodeCount_ + pushSize_) <= MaxSize) {

                        deleteList = false;

                        globalPool_ = new GlobalPool(globalPool_);
                        globalPool_->node_ = head_;
                        globalPool_->nodeCount_ = pushSize_;
                        globalNodeCount_ += pushSize_;
                    }

                }
                if (deleteList) {
                    pushSize_ = 0;
                    deleteLinkedList(head_);
                }
                head_ = new(p) Node(0);
                pushSize_ = 1;
                return true;
            }

            head_ = new(p) Node(head_);
            pushSize_++;
            return true;
        }

        static void deleteLinkedList(Node* head) {
            Node* n = head;
            while (n) {
                void* p = n;
                n = n->next;
                ::operator delete(p);
            }
        }
    public:
        Impl() {
            pushSize_ = 0;
            head_ = 0;
        }

        ~Impl() {
            // No need for mutex for pop
            deleteLinkedList(head_);
        }

        void* allocate() {
            void* result = pop();
            if (!result) {
                result = ::operator new(std::max(sizeof(T), sizeof(Node)));
            }
            return result;
        }

        void deallocate(void* p) {
            if (!push(p)) {
                ::operator delete(p);
            }
        }
    };

    static boost::thread_specific_ptr<Impl> implPtr_;
    typedef size_t size_type;
    typedef T* pointer;
    typedef const void* const_pointer;

    Allocator() {
    }

    Allocator(const Allocator& /*other*/) {
    }

    template<typename Other, int OtherSize>
    Allocator(const Allocator<Other, OtherSize>& /*other*/) {
    }

    pointer allocate(size_type n, const void * /*hint*/ = 0) {
        Impl *impl = implPtr_.get();
        if (!impl) {
            implPtr_.reset(new Impl);
            impl = implPtr_.get();
        }
        void* p = (n == 1)
            ? impl->allocate()
            : operator new(n * sizeof(T));
        return static_cast<T*>(p);
    }

    void deallocate(pointer ptr, size_type n) {
        Impl *impl = implPtr_.get();
        if (!impl) {
            implPtr_.reset(new Impl);
            impl = implPtr_.get();
        }
        if (n == 1)
            impl->deallocate(ptr);
        else
            ::operator delete(ptr);
    }

    template<typename Other> struct rebind {
        typedef Allocator<Other, MaxSize> other;
    };
};

// typename Allocator<Type,MaxSize>::Impl is important else the compiler
// doesn't understand that it is a type
template <typename Type, int MaxSize>
boost::thread_specific_ptr<typename Allocator<Type,MaxSize>::Impl> Allocator<Type,MaxSize>::implPtr_;

template <typename Type, int MaxSize>
boost::mutex Allocator<Type,MaxSize>::Impl::mutex_;

template <typename Type, int MaxSize>
struct Allocator<Type,MaxSize>::Impl::GlobalPool *Allocator<Type,MaxSize>::Impl::globalPool_;

template <typename Type, int MaxSize>
int Allocator<Type,MaxSize>::Impl::globalNodeCount_;

template<typename Type, int MaxSize>
class ObjectPool {
    typedef boost::shared_ptr<Type> TypeSharedPtr;

    Allocator<Type, MaxSize> allocator_;

 public:
    ObjectPool() {

    }

    TypeSharedPtr create() {
        return boost::allocate_shared<Type>(allocator_);
    }

    ~ObjectPool() {

        struct Allocator<Type,MaxSize>::Impl::GlobalPool *poolEntry = Allocator<Type,MaxSize>::Impl::globalPool_;
        while(poolEntry) {
            Allocator<Type, MaxSize>::Impl::deleteLinkedList(poolEntry->node_);
            struct Allocator<Type,MaxSize>::Impl::GlobalPool *delEntry = poolEntry;
            poolEntry = poolEntry->next_;
            ::operator delete(delEntry);
        }
    }
 private:
    ObjectPool<Type, MaxSize>(const ObjectPool<Type, MaxSize>&);
    ObjectPool<Type, MaxSize>& operator=(const ObjectPool<Type, MaxSize>&);
};

}
#endif /* LIB_OBJECTPOOL_H_ */