TopicPayloadPool.hpp

// Copyright 2020 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// 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.

/**
 * @file TopicPayloadPool.hpp
 */

#ifndef RTPS_HISTORY_TOPICPAYLOADPOOL_HPP
#define RTPS_HISTORY_TOPICPAYLOADPOOL_HPP

#include <fastdds/rtps/common/SerializedPayload.h>
#include <fastdds/rtps/history/IPayloadPool.h>
#include <fastdds/rtps/attributes/ResourceManagement.hpp>
#include <fastdds/dds/log/Log.hpp>
#include <rtps/history/PoolConfig.h>
#include <rtps/history/ITopicPayloadPool.h>

#include <atomic>
#include <cstddef>
#include <memory>
#include <mutex>
#include <vector>
#include <cassert>

namespace eprosima {
namespace fastdds {
namespace rtps {

class TopicPayloadPool : public ITopicPayloadPool
{

public:

    TopicPayloadPool() = default;

    virtual ~TopicPayloadPool()
    {
        EPROSIMA_LOG_INFO(RTPS_UTILS, "PayloadPool destructor");

        for (PayloadNode* payload : all_payloads_)
        {
            delete payload;
        }
    }

    bool get_payload(
            uint32_t size,
            SerializedPayload_t& payload) override;

    bool get_payload(
            const SerializedPayload_t& data,
            SerializedPayload_t& payload) override;

    bool release_payload(
            SerializedPayload_t& payload) override;

    /**
     * @brief Ensures the pool has capacity to fullfill the requirements of a new history.
     *
     * @param [in]  config              The new history's pool requirements.
     * @param [in]  is_reader_history   True if the new history is for a reader. False otherwise.
     * @return Whether the operation was successful or not.
     *
     * @pre
     *   - Current pool is configured for the same memory policy as @c config.memory_policy.
     *
     * @post
     *   - If @c config.maximum_size is not zero
     *     - The maximum size of the pool is increased by @c config.maximum_size.
     *   - else
     *     - The maximum size of the pool is set to the largest representable value.
     *   - If the pool is configured for PREALLOCATED or PREALLOCATED WITH REALLOC memory policy:
     *     - The pool has at least as many elements allocated (including elements already in use)
     *       as the sum of the @c config.initial_size for all reserved writer histories
     *       plus the maximum of the @c config.initial_size for all reserved reader histories.
     */
    bool reserve_history(
            const PoolConfig& config,
            bool is_reader) override;

    /**
     * @brief Informs the pool that some history requirements are not longer active.
     *
     * The pool can release some resources that are not needed any longer.
     *
     * @param [in]  config              The old history's pool requirements, which are no longer active.
     * @param [in]  is_reader_history   True if the history was for a reader. False otherwise.
     * @return Whether the operation was successful or not.
     *
     * @pre
     *   - Current pool is configured for the same memory policy as @c config.memory_policy.
     *   - If all remaining histories were reserved with non zero @c config.maximum_size
     *      - The number of elements in use is less than
     *        the sum of the @c config.maximum_size for all remaining histories
     *
     * @post
     *   - If all remaining histories were reserved with non zero @c config.maximum_size
     *      - The maximum size of the pool is set to
     *        the sum of the @c config.maximum_size for all remaining histories
     *   - else
     *     - The maximum size of the pool remains the largest representable value.
     *   - If the number of allocated elements is greater than the new maximum size,
     *     the excess of elements are freed until the number of allocated elemets is equal to the new maximum.
     */
    bool release_history(
            const PoolConfig& config,
            bool is_reader) override;

    size_t payload_pool_allocated_size() const override
    {
        return all_payloads_.size();
    }

    size_t payload_pool_available_size() const override
    {
        return free_payloads_.size();
    }

    static std::unique_ptr<ITopicPayloadPool> get(
            const BasicPoolConfig& config);

protected:

    class PayloadNode
    {
    public:

        explicit PayloadNode(
                uint32_t size)
        {
            if (!size)
            {
                //! At least, we need this to allocate space for a NodeInfo.
                //! In order to be able to place-construct later
                buffer = (octet*)calloc(sizeof(NodeInfo), sizeof(octet));
            }
            else
            {
                buffer = (octet*)calloc(size + sizeof(NodeInfo) - 1, sizeof(octet));
            }

            if (buffer == nullptr)
            {
                throw std::bad_alloc();
            }

            // The atomic may need some initialization depending on the platform
            new (buffer) NodeInfo();
            data_size(size);
        }

        ~PayloadNode()
        {
            info().~NodeInfo();
            free(buffer);
        }

        bool resize (
                uint32_t size)
        {
            assert(size > data_size());

            octet* old_buffer = buffer;
            buffer = (octet*)realloc(buffer, size + data_offset);
            if (!buffer)
            {
                buffer = old_buffer;
                return false;
            }
            memset(buffer + data_offset + data_size(), 0, (size - data_size()) * sizeof(octet));
            data_size(size);
            return true;
        }

        uint32_t data_size() const
        {
            return info().data_size;
        }

        static uint32_t data_size(
                octet* data)
        {
            return info(data).data_size;
        }

        void data_size(
                uint32_t size)
        {
            info().data_size = size;
        }

        uint32_t data_index() const
        {
            return info().data_index;
        }

        static uint32_t data_index(
                octet* data)
        {
            return info(data).data_index;
        }

        void data_index(
                uint32_t index)
        {
            info().data_index = index;
        }

        octet* data() const
        {
            return info().data;
        }

        void reference()
        {
            info().ref_counter.fetch_add(1, std::memory_order_relaxed);
        }

        bool dereference()
        {
            return (info().ref_counter.fetch_sub(1, std::memory_order_acq_rel) == 1);
        }

        static void reference(
                octet* data)
        {
            info(data).ref_counter.fetch_add(1, std::memory_order_relaxed);
        }

        static bool dereference(
                octet* data)
        {
            return (info(data).ref_counter.fetch_sub(1, std::memory_order_acq_rel) == 1);
        }

    private:

        struct NodeInfo
        {
            std::atomic<uint32_t> ref_counter{ 0 };
            uint32_t data_size = 0;
            uint32_t data_index = 0;
            octet data[1];
        };

        octet* buffer = nullptr;

        // Payload data comes after the metadata
        static constexpr size_t data_offset = offsetof(NodeInfo, data);

        NodeInfo& info() const
        {
            return *reinterpret_cast<NodeInfo*>(buffer);
        }

        static NodeInfo& info(
                octet* data)
        {
            return *reinterpret_cast<NodeInfo*>(data - data_offset);
        }

    };

    /**
     * Adds a new payload in the pool, but does not add it to the list of free payloads
     *
     * @param [IN] size  Minimum size required for the payload data
     * @return The node representing the newly allocated payload.
     *
     * @post
     *   - @c payload_pool_allocated_size() increases by one
     *   - @c payload_pool_available_size() does not change
     */
    virtual PayloadNode* allocate(
            uint32_t size);

    PayloadNode* do_allocate(
            uint32_t size);

    virtual void update_maximum_size(
            const PoolConfig& config,
            bool is_reserve);

    /**
     * Ensures the pool has capacity for at least @c num_payloads elements.
     *
     * @param [IN] min_num_payloads Minimum number of payloads reserved in the pool
     * @param [IN] size             Size to allocate for the payloads that need to be added to the pool
     *
     * @pre
     *   - @c min_num_payloads <= @c max_pool_size_
     * @post
     *   - @c payload_pool_allocated_size() >= @c min_num_payloads
     */
    virtual void reserve (
            uint32_t min_num_payloads,
            uint32_t size);

    /**
     * Ensures the pool has capacity for at most @c num_payloads elements.
     *
     * @param [IN] max_num_payloads Maximum number of payloads reserved in the pool
     *
     * @return @c true on success, @c false otherwise
     *
     * @post
     *   - On success, payload_pool_allocated_size() <= max_num_payloads
     *   - On failure, memory for some payloads may have been released, but payload_pool_allocated_size() > min_num_payloads
     */
    bool shrink (
            uint32_t max_num_payloads);

    /**
     * @brief Get a serialized payload for a new sample.
     *
     * If the payload is recycled from the pool, @c resizable controls whether it can
     * be reallocated to accomodate larger sizes.
     * If @c resizable is false and there is at least one free payload in the pool, that payload will
     * be returned even though it may not reach the requested size.
     *
     * If @c resizable is true and the reallocation fails, the operation returns false and
     * the payload is returned to the pool.
     *
     * @param [in]     size          Number of bytes required for the serialized payload
     * @param [in,out] cache_change  Cache change to assign the payload to
     * @param [in]     resizable     Whether payloads recycled from the pool are resizable to accomodate larger sizes
     *
     * @returns whether the operation succeeded or not
     *
     * @post
     *   On success:
     *     @li Field @c cache_change.payload_owner equals this
     *     @li Field @c serializedPayload.data points to a buffer of at least @c size bytes
     *     @li Field @c serializedPayload.max_size is greater than or equal to @c size
     */
    virtual bool do_get_payload(
            uint32_t size,
            SerializedPayload_t& payload,
            bool resizeable);

    virtual MemoryManagementPolicy_t memory_policy() const = 0;

    uint32_t max_pool_size_             = 0;  //< Maximum size of the pool
    uint32_t infinite_histories_count_  = 0;  //< Number of infinite histories reserved
    uint32_t finite_max_pool_size_      = 0;  //< Maximum size of the pool if no infinite histories were reserved

    std::vector<PayloadNode*> free_payloads_; //< Payloads that are free
    std::vector<PayloadNode*> all_payloads_;  //< All payloads

    std::mutex mutex_;

};


}  // namespace rtps
}  // namespace fastdds
}  // namespace eprosima

#endif  // RTPS_HISTORY_TOPICPAYLOADPOOL_HPP