resizeable_lockfree_queue.hpp

// Copyright (c) 2020 - 2021 by Apex.AI Inc. All rights reserved.
//
// 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.
//
// SPDX-License-Identifier: Apache-2.0
 
#ifndef IOX_HOOFS_CONCURRENT_RESIZEABLE_LOCKFREE_QUEUE_HPP
#define IOX_HOOFS_CONCURRENT_RESIZEABLE_LOCKFREE_QUEUE_HPP
 
#include "iceoryx_hoofs/concurrent/lockfree_queue.hpp"
#include "iceoryx_hoofs/cxx/type_traits.hpp"
#include "iceoryx_hoofs/cxx/vector.hpp"
 
#include <atomic>
 
namespace iox
{
namespace concurrent
{
 
// Remark: We use protected inheritance to make the base class methods inaccessible for the user.
// We cannot use virtual functions since we need to use this class in shared memory.
// Some of the methods need to be rewritten specifically for this class, others simply redirect
// the call to the base class.
//
// Since supporting the resize (setCapacity) functionality has an impact on the runtime even
// if the feature is not used, we provide a queue wihout resize functionality in an additional
// base class that can be used separately.
template <typename ElementType, uint64_t MaxCapacity>
class ResizeableLockFreeQueue : protected LockFreeQueue<ElementType, MaxCapacity>
{
  private:
    using Base = LockFreeQueue<ElementType, MaxCapacity>;
 
  public:
    using element_t = ElementType;
    static constexpr uint64_t MAX_CAPACITY = MaxCapacity;
 
    ResizeableLockFreeQueue() noexcept = default;
    ~ResizeableLockFreeQueue() noexcept = default;
 
    // deleted for now, can be implemented later if needed
    // note: concurrent copying or moving in lockfree fashion is nontrivial
    ResizeableLockFreeQueue(const ResizeableLockFreeQueue&) = delete;
    ResizeableLockFreeQueue(ResizeableLockFreeQueue&&) = delete;
    ResizeableLockFreeQueue& operator=(const ResizeableLockFreeQueue&) = delete;
    ResizeableLockFreeQueue& operator=(ResizeableLockFreeQueue&&) = delete;
 
    ResizeableLockFreeQueue(const uint64_t initialCapacity) noexcept;

    static constexpr uint64_t maxCapacity() noexcept;
 
    using Base::empty;
    using Base::pop;
    using Base::size;
    using Base::tryPush;

    uint64_t capacity() const noexcept;

    iox::cxx::optional<ElementType> push(const ElementType& value) noexcept;

    iox::cxx::optional<ElementType> push(ElementType&& value) noexcept;
 
    // overloads to set the capacity
    // 1) The most general one allows providing a removeHandler to specify remove behavior.
    //    This could e.g. be to store them in a container.
    // 2) The second overload discards removed elements.


    template <typename Function,
              typename = typename std::enable_if<cxx::is_invocable<Function, ElementType>::value>::type>
    bool setCapacity(const uint64_t newCapacity, Function&& removeHandler) noexcept;

    bool setCapacity(const uint64_t newCapacity) noexcept;
 
  private:
    using BufferIndex = typename Base::BufferIndex;
    std::atomic<uint64_t> m_capacity{MaxCapacity};
    // must be operator= otherwise it is undefined, see https://en.cppreference.com/w/cpp/atomic/ATOMIC_FLAG_INIT
    std::atomic_flag m_resizeInProgress = ATOMIC_FLAG_INIT;
    iox::cxx::vector<BufferIndex, MaxCapacity> m_unusedIndices;

    uint64_t increaseCapacity(const uint64_t toIncrease) noexcept;

    template <typename Function>
    uint64_t decreaseCapacity(const uint64_t toDecrease, Function&& removeHandler) noexcept;

    bool tryGetUsedIndex(BufferIndex& index) noexcept;
 
    template <typename T>
    iox::cxx::optional<ElementType> pushImpl(T&& value) noexcept;
};
 
} // namespace concurrent
} // namespace iox
 
#include "iceoryx_hoofs/internal/concurrent/lockfree_queue/resizeable_lockfree_queue.inl"
 
#endif // IOX_HOOFS_CONCURRENT_RESIZEABLE_LOCKFREE_QUEUE_HPP