Skip to content
This repository has been archived by the owner on Aug 2, 2022. It is now read-only.

Saving and restoring undo stack at shutdown and startup #9772

Merged
merged 9 commits into from
Dec 11, 2020
Merged

Saving and restoring undo stack at shutdown and startup #9772

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
aadb7ed
Initial code for saving and restoring undo stack at shutdown and startup
linhuang-blockone Dec 10, 2020
f60ca30
Call detach before discarding session and fix a wrong fix
linhuang-blockone Dec 10, 2020
7a44b93
remove unnecessary set_datadir, use constructor to pass in datadir, a…
linhuang-blockone Dec 10, 2020
cfb3618
Add missing restoring deleted keys and fix packing shared_bytes
linhuang-blockone Dec 10, 2020
03c6b22
Use push and incoporate Tim's other comments
linhuang-blockone Dec 10, 2020
b193aa3
Use simplifed >> and << operators
linhuang-blockone Dec 10, 2020
932ff72
Move m_revision = rev; out of for loop
linhuang-blockone Dec 10, 2020
01d553b
Do not keep undo_stack.dat when error happens
linhuang-blockone Dec 11, 2020
361272f
Re-activate revision check asserts and remove related #warnings
linhuang-blockone Dec 11, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion libraries/chain/combined_database.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -232,7 +232,7 @@ namespace eosio { namespace chain {
auto rdb = std::shared_ptr<rocksdb::DB>{ p };
return std::make_unique<rocks_db_type>(eosio::session::make_session(std::move(rdb), 1024));
}() },
kv_undo_stack(std::make_unique<eosio::session::undo_stack<rocks_db_type>>(*kv_database)),
kv_undo_stack(std::make_unique<eosio::session::undo_stack<rocks_db_type>>(*kv_database, cfg.state_dir)),
kv_snapshot_batch_threashold(cfg.persistent_storage_mbytes_batch * 1024 * 1024) {}

void combined_database::check_backing_store_setting(bool clean_startup) {
Expand Down
3 changes: 2 additions & 1 deletion libraries/chain_kv/include/b1/session/session.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -259,7 +259,8 @@ class session {
It& first_not_deleted_in_iterator_cache_(It& it, const It& end) const;

private:
parent_variant_type m_parent{ nullptr };
//parent_variant_type m_parent{ nullptr };
TimothyBanks marked this conversation as resolved.
Show resolved Hide resolved
parent_variant_type m_parent{ static_cast<Parent*>(nullptr) };
cache_type m_cache;
};

Expand Down
6 changes: 6 additions & 0 deletions libraries/chain_kv/include/b1/session/shared_bytes.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -508,8 +508,14 @@ shared_bytes::shared_bytes_iterator<Iterator_traits>::operator[](size_t index) c
return m_buffer[index];
}

struct shared_bytes_reflect{
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So, instead of making this struct and reflecting it. What you should do is this then.

Instead of using the FC_REFLECT, because the internals of shared_bytes are not very usable through reflection.

Add two public methods to shared_bytes:

template <typename Stream>
inline Stream& operator<<(Stream& ds, const shared_bytes& b) {
   ds << b.size();
   ds.write(b.data(), b.size());
   return ds;
}

and

template <typename Stream>
inline Stream& operator>>(Stream& ds, shared_bytes& b) {
   std::size_t sz;
   ds >> sz;
   shared_bytes tmp = {sz};
   ds.read(tmp.data(), tmp.size());
   b = tmp;
   return ds;
}

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

shared_bytes already has an out stream overload and a method that can create a shared_bytes from a hex string (which is what is streamed out). You would just need to provide the operator for streaming in.

std::vector<char> data;
};

} // namespace eosio::session

FC_REFLECT(eosio::session::shared_bytes_reflect, (data))

namespace std {

template <>
Expand Down
141 changes: 139 additions & 2 deletions libraries/chain_kv/include/b1/session/undo_stack.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,21 @@

#include <queue>

#include <fc/filesystem.hpp>
#include <fc/io/fstream.hpp>
#include <fc/io/datastream.hpp>
#include <fc/io/raw.hpp>
#include <fc/reflect/typename.hpp>
#include <fstream>
#include <b1/session/session.hpp>
#include <b1/session/session_variant.hpp>
#include <eosio/chain/exceptions.hpp>

namespace eosio::session {
const uint32_t undo_stack_magic_number = 0x30510ABC;
TimothyBanks marked this conversation as resolved.
Show resolved Hide resolved
const uint32_t undo_stack_min_supported_version = 1;
const uint32_t undo_stack_max_supported_version = 1;
constexpr auto undo_stack_filename = "undo_stack.dat";

/// \brief Represents a container of pending sessions to be committed.
template <typename Session>
Expand All @@ -19,7 +29,7 @@ class undo_stack {

/// \brief Constructor.
/// \param head The session that the changes are merged into when commit is called.
undo_stack(Session& head);
undo_stack(Session& head, const fc::path& datadir = {});
undo_stack(const undo_stack&) = delete;
undo_stack(undo_stack&&) = default;
~undo_stack();
Expand Down Expand Up @@ -66,18 +76,26 @@ class undo_stack {
/// \remarks This is the next session to be committed.
const_variant_type bottom() const;

void open();
void close();

private:
int64_t m_revision{ 0 };
Session* m_head;
std::deque<session_type> m_sessions; // Need a deque so pointers don't become invalidated. The session holds a
// pointer to the parent internally.
fc::path m_datadir;
};

template <typename Session>
undo_stack<Session>::undo_stack(Session& head) : m_head{ &head } {}
undo_stack<Session>::undo_stack(Session& head, const fc::path& datadir) : m_head{ &head }, m_datadir{ datadir } {
open();
}

template <typename Session>
undo_stack<Session>::~undo_stack() {
close();

for (auto& session : m_sessions) { session.undo(); }
TimothyBanks marked this conversation as resolved.
Show resolved Hide resolved
}

Expand Down Expand Up @@ -197,4 +215,123 @@ typename undo_stack<Session>::const_variant_type undo_stack<Session>::bottom() c
return { *m_head, nullptr };
}

template <typename Session>
void undo_stack<Session>::open() {
if (m_datadir.empty())
return;

if (!fc::is_directory(m_datadir))
fc::create_directories(m_datadir);

auto undo_stack_dat = m_datadir / undo_stack_filename;
if( fc::exists( undo_stack_dat ) ) {
try {
std::string content;
fc::read_file_contents( undo_stack_dat, content );

fc::datastream<const char*> ds( content.data(), content.size() );

// validate totem
uint32_t totem = 0;
fc::raw::unpack( ds, totem );
EOS_ASSERT( totem == undo_stack_magic_number, eosio::chain::chain_exception,
"Undo stack data file '${filename}' has unexpected magic number: ${actual_totem}. Expected ${expected_totem}",
("filename", undo_stack_dat.generic_string())
("actual_totem", totem)
("expected_totem", undo_stack_magic_number)
);

// validate version
uint32_t version = 0;
fc::raw::unpack( ds, version );
EOS_ASSERT( version >= undo_stack_min_supported_version && version <= undo_stack_max_supported_version,
eosio::chain::chain_exception,
"Unsupported version of Undo stack data file '${filename}'. "
"Undo stack data version is ${version} while code supports version(s) [${min},${max}]",
("filename", undo_stack_dat.generic_string())
("version", version)
("min", undo_stack_min_supported_version)
("max", undo_stack_max_supported_version)
);

int64_t rev; fc::raw::unpack( ds, rev );
revision(rev);
TimothyBanks marked this conversation as resolved.
Show resolved Hide resolved

size_t num_sessions; fc::raw::unpack( ds, num_sessions );
for( size_t i = 0; i < num_sessions; ++i ) {
session_type session;
TimothyBanks marked this conversation as resolved.
Show resolved Hide resolved
if (m_sessions.empty())
session.attach(*m_head);
else
session.attach(m_sessions.back());

size_t num_updated_keys; fc::raw::unpack( ds, num_updated_keys );
for( size_t j = 0; j < num_updated_keys; ++j ) {
shared_bytes_reflect raw_key; fc::raw::unpack( ds, raw_key );
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If you implement the above methods for shared_bytes, this becomes...

shared_bytes key;
ds >> key;

shared_bytes value;
ds >> value;

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the suggestion! This simplifies the code. Changed.

shared_bytes_reflect raw_value; fc::raw::unpack( ds, raw_value );

shared_bytes key(raw_key.data.data(), raw_key.data.size());
shared_bytes value(raw_value.data.data(), raw_value.data.size());
session.write(key, value);
}

size_t num_deleted_keys; fc::raw::unpack( ds, num_deleted_keys );
for( size_t j = 0; j < num_deleted_keys; ++j ) {
shared_bytes_reflect raw_key; fc::raw::unpack( ds, raw_key );

shared_bytes key(raw_key.data.data(), raw_key.data.size());
session.erase(key);
}

m_sessions.emplace_back(std::move(session));
}
} FC_CAPTURE_AND_RETHROW( (undo_stack_dat) )

fc::remove( undo_stack_dat );
}
}

template <typename Session>
void undo_stack<Session>::close() {
if (m_datadir.empty())
return;

auto undo_stack_dat = m_datadir / undo_stack_filename;

std::ofstream out( undo_stack_dat.generic_string().c_str(), std::ios::out | std::ios::binary | std::ofstream::trunc );
fc::raw::pack( out, undo_stack_magic_number );
fc::raw::pack( out, undo_stack_max_supported_version );

fc::raw::pack( out, revision() );
fc::raw::pack( out, size() ); // number of sessions

while ( !m_sessions.empty() ) {
auto& session = m_sessions.front();
auto updated_keys = session.updated_keys();
fc::raw::pack( out, updated_keys.size() ); // number of updated keys

for (const auto& key: updated_keys) {
auto value = session.read(key);

if ( value ) {
shared_bytes_reflect raw_key;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

with the datastream operators this would become...

out << key;
out << value;

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Changed.

raw_key.data.assign(key.data(), key.data() + key.size());
shared_bytes_reflect raw_value;
raw_value.data.assign(value->data(), value->data() + value->size());
fc::raw::pack( out, raw_key );
fc::raw::pack( out, raw_value );
}
}

auto deleted_keys = session.deleted_keys();
fc::raw::pack( out, deleted_keys.size() ); // number of deleted keys

for (const auto& key: deleted_keys) {
fc::raw::pack( out, key );
}

session.detach();
m_sessions.pop_front();
}
}
} // namespace eosio::session