Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add a stream serializer #270

4 changes: 4 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
@@ -1,6 +1,10 @@
build/
CMakeFiles/

spack-build*
spack-configure-args.txt
install-time-test-log.txt

.dir-locals.el

*.a
Expand Down
44 changes: 44 additions & 0 deletions src/checkpoint/checkpoint_api.h
Original file line number Diff line number Diff line change
Expand Up @@ -217,6 +217,50 @@ std::unique_ptr<T> deserializeFromFile(std::string const& file);
template <typename T>
void deserializeInPlaceFromFile(std::string const& file, T* buf);

/**
* \brief Serialize \c T to a stream
*
* Byte-serializes \c T to stream. Handling of any errors during writing
* to the stream will be handled by the stream itself, e.g. any exceptions
* or status bits to check will depend on stream type.
*
* \param[in] target the \c T to serialize
* \param[in] stream to serialize into, with tellp and write functions.
*/
template <typename T, typename StreamT>
void serializeToStream(T& target, StreamT& stream);

/**
* \brief De-serialize and reify \c T from a stream
*
* De-serializes an object recursively by first invoking the reconstruction
* strategy and then \c serialize functions/methods recursively to rebuild the
* state of the object as serialized. During reconstruction, based on trait
* detection, \c T will either be default constructed or reconstructed based on
* a user-defined reconstruct method.
*
* \param[in] stream the stream to read with bytes for \c T, with tellg and read functions
*
* \return unique pointer to the new object \c T
*/
template <typename T, typename StreamT>
std::unique_ptr<T> deserializeFromStream(StreamT& stream);

/**
* \brief De-serialize and reify \c T from a stream in place on an existing
* pointer to \c T
*
* De-serializes an object recursively by invoking the \c serialize
* functions/methods recursively to rebuild the state of the object as
* serialized.
*
* \param[in] stream the stream to read with bytes for \c T, with tellg and read functions
* \param[in] t a valid, constructed \c T to deserialize into
*/
template <typename T, typename StreamT>
void deserializeInPlaceFromStream(StreamT& stream, T* buf);


} /* end namespace checkpoint */

#endif /*INCLUDED_CHECKPOINT_CHECKPOINT_API_H*/
25 changes: 25 additions & 0 deletions src/checkpoint/checkpoint_api.impl.h
Original file line number Diff line number Diff line change
Expand Up @@ -123,6 +123,31 @@ void deserializeInPlaceFromFile(std::string const& file, T* t) {
);
}

template <typename T, typename StreamT>
void serializeToStream(T& target, StreamT& stream) {
auto len = getSize<T>(target);
dispatch::Standard::pack<T, StreamSerializer<StreamT, eSerializationMode::Packing>>(
target, len, eSerializationMode::Packing, stream
);
}

template <typename T, typename StreamT>
std::unique_ptr<T> deserializeFromStream(StreamT& stream) {
auto mem = dispatch::Standard::allocate<T>();
T* t_buf = dispatch::Standard::construct<T>(mem);
auto t = dispatch::Standard::unpack<T, StreamSerializer<StreamT, eSerializationMode::Unpacking>>(
t_buf, eSerializationMode::Unpacking, stream
);
return std::unique_ptr<T>(t);
}

template <typename T, typename StreamT>
void deserializeInPlaceFromStream(StreamT& stream, T* t) {
dispatch::Standard::unpack<T, StreamSerializer<StreamT, eSerializationMode::Unpacking>>(
t, eSerializationMode::Unpacking, stream
);
}

} /* end namespace checkpoint */

#endif /*INCLUDED_CHECKPOINT_CHECKPOINT_API_IMPL_H*/
5 changes: 4 additions & 1 deletion src/checkpoint/serializers/serializers_headers.h
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,7 @@
#include "checkpoint/serializers/sizer.h"
#include "checkpoint/serializers/packer.h"
#include "checkpoint/serializers/unpacker.h"
#include "checkpoint/serializers/stream_serializer.h"

#define checkpoint_serializer_variadic_args() \
checkpoint::Footprinter, \
Expand All @@ -58,6 +59,8 @@
checkpoint::PackerIO, \
checkpoint::Unpacker, \
checkpoint::UnpackerIO, \
checkpoint::Sizer \
checkpoint::Sizer, \
checkpoint::IStreamSerializer, \
checkpoint::OStreamSerializer \

#endif /*INCLUDED_CHECKPOINT_SERIALIZERS_SERIALIZERS_HEADERS_H*/
126 changes: 126 additions & 0 deletions src/checkpoint/serializers/stream_serializer.h
Original file line number Diff line number Diff line change
@@ -0,0 +1,126 @@
/*
//@HEADER
// *****************************************************************************
//
// stream_serializer.h
// DARMA/checkpoint => Serialization Library
//
// Copyright 2019 National Technology & Engineering Solutions of Sandia, LLC
// (NTESS). Under the terms of Contract DE-NA0003525 with NTESS, the U.S.
// Government retains certain rights in this software.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are met:
//
// * Redistributions of source code must retain the above copyright notice,
// this list of conditions and the following disclaimer.
//
// * Redistributions in binary form must reproduce the above copyright notice,
// this list of conditions and the following disclaimer in the documentation
// and/or other materials provided with the distribution.
//
// * Neither the name of the copyright holder nor the names of its
// contributors may be used to endorse or promote products derived from this
// software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
// AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
// ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
// LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
// CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
// SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
// CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
// ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
// POSSIBILITY OF SUCH DAMAGE.
//
// Questions? Contact darma@sandia.gov
//
// *****************************************************************************
//@HEADER
*/

#if !defined INCLUDED_CHECKPOINT_SERIALIZERS_STREAM_SERIALIZER_H
#define INCLUDED_CHECKPOINT_SERIALIZERS_STREAM_SERIALIZER_H

#include "checkpoint/common.h"
#include "checkpoint/serializers/base_serializer.h"
#include <ostream>
#include <istream>

namespace checkpoint {

namespace {
template<Serializer::ModeType mode>
struct false_type : std::false_type {};

template <typename StreamT, Serializer::ModeType mode>
struct StreamHolder {
StreamHolder(StreamT& stream){
static_assert(false_type<mode>::value, "Unsupported serialization mode");
}
};

template <typename StreamT>
struct StreamHolder<StreamT, Serializer::ModeType::Packing> {
StreamT& stream;

StreamHolder(StreamT& stream) : stream(stream) {};

void copy(char* ptr, SerialSizeType len){
stream.write(ptr, len);
}

SerialSizeType position(){
return stream.tellp();
}
};

template <typename StreamT>
struct StreamHolder<StreamT, Serializer::ModeType::Unpacking> {
StreamT& stream;

StreamHolder(StreamT& stream) : stream(stream) {};

void copy(char* ptr, SerialSizeType len){
stream.read(ptr, len);
}

SerialSizeType position(){
return stream.tellg();
}
};
}

template <typename StreamT, Serializer::ModeType mode>
struct StreamSerializer : Serializer {
StreamSerializer(ModeType const& in_mode, StreamT& in_stream)
: Serializer(mode), stream(in_stream), stream_start_position(stream.position()) {
#ifdef DEBUG
assert(in_mode == mode);
#endif
}

StreamSerializer(SerialSizeType size, ModeType const& in_mode, StreamT& in_stream)
: StreamSerializer(in_mode, in_stream) {}

void contiguousBytes(void* ptr, SerialSizeType size, SerialSizeType num_elms) {
stream.copy(static_cast<char*>(ptr), size*num_elms);
}

SerialSizeType usedBufferSize() {
return stream.position() - stream_start_position;
}

private:
StreamHolder<StreamT, mode> stream;
const SerialSizeType stream_start_position;
};

using IStreamSerializer = StreamSerializer<std::istream, Serializer::ModeType::Unpacking>;
using OStreamSerializer = StreamSerializer<std::ostream, Serializer::ModeType::Packing>;

} /* end namespace checkpoint */

#endif /*INCLUDED_CHECKPOINT_SERIALIZERS_STREAM_SERIALIZER_H*/
6 changes: 5 additions & 1 deletion tests/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ macro(checkpoint_link_target target has_mpi)
target_include_directories(${target} PUBLIC ${PROJECT_TEST_UNIT_DIR})
target_include_directories(${target} PRIVATE ${GTEST_INCLUDE_DIRS})
target_link_libraries(${target} PRIVATE ${GTEST_BOTH_LIBRARIES})
target_link_libraries(${target} PRIVATE Threads::Threads)
target_link_libraries(${target} PUBLIC ${CHECKPOINT_LIBRARY})

if (${has_mpi})
Expand Down Expand Up @@ -57,8 +58,11 @@ if (checkpoint_tests_enabled)
${PROJECT_TEST_UNIT_DIR}/test_commons.h
${PROJECT_TEST_UNIT_DIR}/test_harness.h
)

#System's pthread library
find_package(Threads REQUIRED)

if(checkpoint_mpi_enabled)
if(checkpoint_mpi_enabled)
set(
TEST_HEADER_FILES ${TEST_HEADER_FILES}
${PROJECT_TEST_MPI_UNIT_DIR}/mpi-init.h
Expand Down
Loading