packed_data.hpp

 
#pragma once
 
#include <rapidsmpf/memory/content_description.hpp>
#include <rapidsmpf/memory/packed_data.hpp>
#include <rapidsmpf/streaming/core/message.hpp>
 
namespace rapidsmpf::streaming {
 
inline ContentDescription get_content_description(PackedData const& obj) {
    return ContentDescription{
        {{obj.data->mem_type(), obj.data->size}}, ContentDescription::Spillable::YES
    };
}
 
Message to_message(std::uint64_t sequence_number, std::unique_ptr<PackedData> chunk) {
    auto cd = get_content_description(*chunk);
    return Message{
        sequence_number,
        std::move(chunk),
        cd,
        [](Message const& msg, MemoryReservation& reservation) -> Message {
            auto const& self = msg.get<PackedData>();
            auto chunk = std::make_unique<PackedData>(self.copy(reservation));
            auto cd = get_content_description(*chunk);
            return Message{msg.sequence_number(), std::move(chunk), cd, msg.copy_cb()};
        }
    };
}
 
}  // namespace rapidsmpf::streaming