Minecon724/dncurrency
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages 1 Wiki Activity Actions

Merge branch 'merge_unchecked_gap_cache'

Browse source
This commit is contained in:
clemahieu 2017-06-20 15:58:18 -05:00
commit cb21c9721b
15 changed files with 159 additions and 552 deletions
Download patch file Download diff file Expand all files Collapse all files

1
CMakeLists.txt
View file

@ -212,7 +212,6 @@ if (RAIBLOCKS_TEST)
add_executable (core_test
rai/core_test/block.cpp
rai/core_test/block_store.cpp
rai/core_test/block_synchronization.cpp
rai/core_test/node.cpp
rai/core_test/conflicts.cpp
rai/core_test/daemon.cpp

32
rai/core_test/block_store.cpp
View file

@ -183,14 +183,14 @@ TEST (bootstrap, simple)
rai::send_block block1 (0, 1, 2, rai::keypair ().prv, 4, 5);
rai::transaction transaction (store.environment, nullptr, true);
auto block2 (store.unchecked_get (transaction, block1.previous ()));
ASSERT_EQ (nullptr, block2);
ASSERT_TRUE (block2.empty ());
store.unchecked_put (transaction, block1.previous (), block1);
auto block3 (store.unchecked_get (transaction, block1.previous ()));
ASSERT_NE (nullptr, block3);
ASSERT_EQ (block1, *block3);
store.unchecked_del (transaction, block1.previous ());
ASSERT_FALSE (block3.empty ());
ASSERT_EQ (block1, *block3 [0]);
store.unchecked_del (transaction, block1.previous (), block1);
auto block4 (store.unchecked_get (transaction, block1.previous ()));
ASSERT_EQ (nullptr, block4);
ASSERT_TRUE (block4.empty ());
}
TEST (checksum, simple)
@ -760,3 +760,25 @@ TEST (block_store, upgrade_v5_v6)
ASSERT_EQ (1, info.block_count);
}
TEST (block_store, upgrade_v6_v7)
{
auto path (rai::unique_path ());
{
bool init (false);
rai::block_store store (init, path);
ASSERT_FALSE (init);
rai::transaction transaction (store.environment, nullptr, true);
rai::genesis genesis;;
genesis.initialize (transaction, store);
store.version_put (transaction, 6);
rai::send_block send1 (0, 0, 0, rai::test_genesis_key.prv, rai::test_genesis_key.pub, 0);
store.unchecked_put (transaction, send1.hash (), send1);
ASSERT_NE (store.unchecked_end (), store.unchecked_begin (transaction));
}
bool init (false);
rai::block_store store (init, path);
ASSERT_FALSE (init);
rai::transaction transaction (store.environment, nullptr, false);
ASSERT_EQ (store.unchecked_end (), store.unchecked_begin (transaction));
}

278
rai/core_test/block_synchronization.cpp
View file

@ -1,278 +0,0 @@
#include <gtest/gtest.h>
#include <rai/node/node.hpp>
#include <rai/node/testing.hpp>
static boost::log::sources::logger_mt test_log;
class test_synchronization : public rai::block_synchronization
{
public:
test_synchronization (rai::block_store & store_a) :
rai::block_synchronization (test_log),
store (store_a)
{
}
bool synchronized (MDB_txn * transaction_a, rai::block_hash const & hash_a) override
{
return store.block_exists (transaction_a, hash_a);
}
std::unique_ptr <rai::block> retrieve (MDB_txn * transaction_a, rai::block_hash const & hash_a) override
{
return store.unchecked_get (transaction_a, hash_a);
}
rai::sync_result target (MDB_txn * transaction_a, rai::block const & block_a) override
{
store.block_put (transaction_a, block_a.hash (), block_a);
return rai::sync_result::success;
}
rai::block_store & store;
};
TEST (pull_synchronization, empty)
{
bool init (false);
rai::block_store store (init, rai::unique_path ());
ASSERT_FALSE (init);
test_synchronization sync (store);
rai::transaction transaction (store.environment, nullptr, true);
ASSERT_EQ (rai::sync_result::success, sync.synchronize (transaction, 0));
ASSERT_EQ (0, store.block_count (transaction).sum ());
}
TEST (pull_synchronization, one)
{
bool init (false);
rai::block_store store (init, rai::unique_path ());
ASSERT_FALSE (init);
rai::open_block block1 (0, 1, 2, rai::keypair ().prv, 4, 5);
rai::send_block block2 (block1.hash (), 0, 1, rai::keypair ().prv, 3, 4);
rai::transaction transaction (store.environment, nullptr, true);
store.block_put (transaction, block1.hash (), block1);
store.unchecked_put (transaction, block2.hash (), block2);
test_synchronization sync (store);
ASSERT_EQ (rai::sync_result::success, sync.synchronize (transaction, block2.hash ()));
ASSERT_EQ (2, store.block_count (transaction).sum ());
ASSERT_NE (nullptr, store.block_get (transaction, block2.hash ()));
}
TEST (pull_synchronization, send_dependencies)
{
bool init (false);
rai::block_store store (init, rai::unique_path ());
ASSERT_FALSE (init);
rai::open_block block1 (0, 1, 2, rai::keypair ().prv, 4, 5);
rai::send_block block2 (block1.hash (), 0, 1, rai::keypair ().prv, 3, 4);
rai::send_block block3 (block2.hash (), 0, 1, rai::keypair ().prv, 3, 4);
rai::transaction transaction (store.environment, nullptr, true);
store.block_put (transaction, block1.hash (), block1);
store.unchecked_put (transaction, block2.hash (), block2);
store.unchecked_put (transaction, block3.hash (), block3);
test_synchronization sync (store);
ASSERT_EQ (rai::sync_result::success, sync.synchronize (transaction, block3.hash ()));
ASSERT_EQ (3, store.block_count (transaction).sum ());
ASSERT_NE (nullptr, store.block_get (transaction, block2.hash ()));
ASSERT_NE (nullptr, store.block_get (transaction, block3.hash ()));
}
TEST (pull_synchronization, change_dependencies)
{
bool init (false);
rai::block_store store (init, rai::unique_path ());
ASSERT_FALSE (init);
rai::open_block block1 (0, 1, 2, rai::keypair ().prv, 4, 5);
rai::send_block block2 (block1.hash (), 0, 1, rai::keypair ().prv, 3, 4);
rai::change_block block3 (block2.hash (), 0, rai::keypair ().prv, 2, 3);
rai::transaction transaction (store.environment, nullptr, true);
store.block_put (transaction, block1.hash (), block1);
store.unchecked_put (transaction, block2.hash (), block2);
store.unchecked_put (transaction, block3.hash (), block3);
test_synchronization sync (store);
ASSERT_EQ (rai::sync_result::success, sync.synchronize (transaction, block3.hash ()));
ASSERT_EQ (3, store.block_count (transaction).sum ());
ASSERT_NE (nullptr, store.block_get (transaction, block2.hash ()));
ASSERT_NE (nullptr, store.block_get (transaction, block3.hash ()));
}
TEST (pull_synchronization, open_dependencies)
{
bool init (false);
rai::block_store store (init, rai::unique_path ());
ASSERT_FALSE (init);
rai::open_block block1 (0, 1, 2, rai::keypair ().prv, 4, 5);
rai::send_block block2 (block1.hash (), 0, 1, rai::keypair ().prv, 3, 4);
rai::open_block block3 (block2.hash (), 1, 1, rai::keypair ().prv, 4, 5);
rai::transaction transaction (store.environment, nullptr, true);
store.block_put (transaction, block1.hash (), block1);
store.unchecked_put (transaction, block2.hash (), block2);
store.unchecked_put (transaction, block3.hash (), block3);
test_synchronization sync (store);
ASSERT_EQ (rai::sync_result::success, sync.synchronize (transaction, block3.hash ()));
ASSERT_EQ (3, store.block_count (transaction).sum ());
ASSERT_NE (nullptr, store.block_get (transaction, block2.hash ()));
ASSERT_NE (nullptr, store.block_get (transaction, block3.hash ()));
}
TEST (pull_synchronization, receive_dependencies)
{
bool init (false);
rai::block_store store (init, rai::unique_path ());
ASSERT_FALSE (init);
rai::open_block block1 (0, 1, 2, rai::keypair ().prv, 4, 5);
rai::send_block block2 (block1.hash (), 0, 1, rai::keypair ().prv, 3, 4);
rai::open_block block3 (block2.hash (), 1, 1, rai::keypair ().prv, 4, 5);
rai::send_block block4 (block2.hash (), 0, 1, rai::keypair ().prv, 3, 4);
rai::receive_block block5 (block3.hash (), block4.hash (), rai::keypair ().prv, 0, 0);
rai::transaction transaction (store.environment, nullptr, true);
store.block_put (transaction, block1.hash (), block1);
store.unchecked_put (transaction, block2.hash (), block2);
store.unchecked_put (transaction, block3.hash (), block3);
store.unchecked_put (transaction, block4.hash (), block4);
store.unchecked_put (transaction, block5.hash (), block5);
test_synchronization sync (store);
ASSERT_EQ (rai::sync_result::success, sync.synchronize (transaction, block5.hash ()));
ASSERT_EQ (3, store.block_count (transaction).sum ());
// Synchronize 2 per iteration in test mode
ASSERT_EQ (rai::sync_result::success, sync.synchronize (transaction, block5.hash ()));
ASSERT_EQ (5, store.block_count (transaction).sum ());
ASSERT_NE (nullptr, store.block_get (transaction, block2.hash ()));
ASSERT_NE (nullptr, store.block_get (transaction, block3.hash ()));
ASSERT_NE (nullptr, store.block_get (transaction, block4.hash ()));
ASSERT_NE (nullptr, store.block_get (transaction, block5.hash ()));
}
TEST (pull_synchronization, ladder_dependencies)
{
bool init (false);
rai::block_store store (init, rai::unique_path ());
ASSERT_FALSE (init);
rai::open_block block1 (0, 1, 2, rai::keypair ().prv, 4, 5);
rai::send_block block2 (block1.hash (), 0, 1, rai::keypair ().prv, 3, 4);
rai::open_block block3 (block2.hash (), 1, 1, rai::keypair ().prv, 4, 5);
rai::send_block block4 (block3.hash (), 0, 1, rai::keypair ().prv, 3, 4);
rai::receive_block block5 (block2.hash (), block4.hash (), rai::keypair ().prv, 0, 0);
rai::send_block block6 (block5.hash (), 0, 1, rai::keypair ().prv, 3, 4);
rai::receive_block block7 (block4.hash (), block6.hash (), rai::keypair ().prv, 0, 0);
rai::transaction transaction (store.environment, nullptr, true);
store.block_put (transaction, block1.hash (), block1);
store.unchecked_put (transaction, block2.hash (), block2);
store.unchecked_put (transaction, block3.hash (), block3);
store.unchecked_put (transaction, block4.hash (), block4);
store.unchecked_put (transaction, block5.hash (), block5);
store.unchecked_put (transaction, block6.hash (), block6);
store.unchecked_put (transaction, block7.hash (), block7);
test_synchronization sync (store);
ASSERT_EQ (rai::sync_result::success, sync.synchronize (transaction, block7.hash ()));
ASSERT_EQ (3, store.block_count (transaction).sum ());
// Synchronize 2 per iteration in test mode
ASSERT_EQ (rai::sync_result::success, sync.synchronize (transaction, block7.hash ()));
// Synchronize 2 per iteration in test mode
ASSERT_EQ (5, store.block_count (transaction).sum ());
ASSERT_EQ (rai::sync_result::success, sync.synchronize (transaction, block7.hash ()));
ASSERT_EQ (7, store.block_count (transaction).sum ());
ASSERT_NE (nullptr, store.block_get (transaction, block2.hash ()));
ASSERT_NE (nullptr, store.block_get (transaction, block3.hash ()));
ASSERT_NE (nullptr, store.block_get (transaction, block4.hash ()));
ASSERT_NE (nullptr, store.block_get (transaction, block5.hash ()));
ASSERT_NE (nullptr, store.block_get (transaction, block6.hash ()));
ASSERT_NE (nullptr, store.block_get (transaction, block7.hash ()));
}
/*TEST (push_synchronization, empty)
{
bool init (false);
rai::block_store store (init, rai::unique_path ());
ASSERT_FALSE (init);
std::vector <std::unique_ptr <rai::block>> blocks;
rai::push_synchronization sync (test_log, [&blocks] (MDB_txn * transaction_a, rai::block const & block_a)
{
blocks.push_back (block_a.clone ());
return rai::sync_result::success;
});
{
rai::transaction transaction (store.environment, nullptr, true);
ASSERT_EQ (rai::sync_result::error, sync.synchronize (transaction, 0));
}
ASSERT_EQ (0, blocks.size ());
}
TEST (push_synchronization, one)
{
bool init (false);
rai::block_store store (init, rai::unique_path ());
ASSERT_FALSE (init);
rai::open_block block1 (0, 1, 2, rai::keypair ().prv, 4, 5);
rai::send_block block2 (block1.hash (), 0, 1, rai::keypair ().prv, 3, 4);
std::vector <std::unique_ptr <rai::block>> blocks;
{
rai::transaction transaction (store.environment, nullptr, true);
store.block_put (transaction, block1.hash (), block1);
store.block_put (transaction, block2.hash (), block2);
}
rai::push_synchronization sync (test_log, [&blocks, &store] (MDB_txn * transaction_a, rai::block const & block_a)
{
store.block_put (transaction_a, block_a.hash (), block_a);
blocks.push_back (block_a.clone ());
return rai::sync_result::success;
}, store);
{
rai::transaction transaction (store.environment, nullptr, true);
store.unsynced_put (transaction, block2.hash ());
ASSERT_EQ (rai::sync_result::success, sync.synchronize (transaction, block2.hash ()));
}
ASSERT_EQ (1, blocks.size ());
ASSERT_EQ (block2, *blocks [0]);
}*/
// Make sure synchronize terminates even with forks
TEST (pull_synchronization, dependent_fork)
{
rai::system system0 (24000, 1);
auto & node0 (*system0.nodes [0]);
rai::keypair key0;
rai::keypair key1;
rai::transaction transaction (node0.store.environment, nullptr, true);
auto genesis (node0.ledger.latest (transaction, rai::test_genesis_key.pub));
rai::send_block send0 (genesis, key0.pub, rai::genesis_amount - 1 * rai::Gxrb_ratio, rai::test_genesis_key.prv, rai::test_genesis_key.pub, 0);
rai::send_block send1 (genesis, key0.pub, rai::genesis_amount - 2 * rai::Gxrb_ratio, rai::test_genesis_key.prv, rai::test_genesis_key.pub, 0);
rai::send_block send2 (send0.hash (), key0.pub, rai::genesis_amount - 3 * rai::Gxrb_ratio, rai::test_genesis_key.prv, rai::test_genesis_key.pub, 0);
rai::send_block send3 (send1.hash (), key0.pub, rai::genesis_amount - 4 * rai::Gxrb_ratio, rai::test_genesis_key.prv, rai::test_genesis_key.pub, 0);
ASSERT_EQ (rai::process_result::progress, node0.ledger.process (transaction, send0).code);
ASSERT_EQ (rai::process_result::progress, node0.ledger.process (transaction, send2).code);
node0.store.unchecked_put (transaction, send1.hash (), send1);
node0.store.unchecked_put (transaction, send3.hash (), send3);
rai::pull_synchronization sync (node0, nullptr);
ASSERT_EQ (rai::sync_result::fork, sync.synchronize (transaction, send3.hash ()));
// Voting will either discard this block or commit it. If it's discarded we don't want to attempt it again
ASSERT_EQ (nullptr, node0.store.unchecked_get (transaction, send1.hash ()));
// This block will either succeed, if its predecessor is comitted by voting, or will be a gap and will be discarded
ASSERT_NE (nullptr, node0.store.unchecked_get (transaction, send3.hash ()));
ASSERT_TRUE (node0.active.active (send1));
}
// After a synchronization with no pulls or pushes required, clear blocks out of unchecked
TEST (pull_synchronization, clear_blocks)
{
rai::system system0 (24000, 1);
rai::system system1 (24001, 1);
auto & node0 (*system0.nodes [0]);
auto & node1 (*system1.nodes [0]);
rai::send_block send0 (0, 0, rai::genesis_amount - 1 * rai::Gxrb_ratio, rai::test_genesis_key.prv, rai::test_genesis_key.pub, 0);
{
rai::transaction transaction (node1.store.environment, nullptr, true);
node1.store.unchecked_put (transaction, send0.hash (), send0);
}
node1.bootstrap_initiator.bootstrap (node0.network.endpoint ());
auto iterations (0);
auto done (false);
while (!done)
{
{
rai::transaction transaction (node1.store.environment, nullptr, false);
done = node1.store.unchecked_get (transaction, send0.hash ()) == nullptr;
}
++iterations;
ASSERT_GT (200, iterations);
system0.poll ();
system1.poll ();
}
}

49
rai/core_test/gap_cache.cpp
View file

@ -6,8 +6,9 @@ TEST (gap_cache, add_new)
rai::system system (24000, 1);
rai::gap_cache cache (*system.nodes [0]);
rai::send_block block1 (0, 1, 2, rai::keypair ().prv, 4, 5);
cache.add (rai::send_block (block1), block1.previous ());
ASSERT_NE (cache.blocks.end (), cache.blocks.find (block1.previous ()));
rai::transaction transaction (system.nodes [0]->store.environment, nullptr, true);
cache.add (transaction, rai::send_block (block1), block1.previous ());
ASSERT_NE (system.nodes [0]->store.unchecked_end (), system.nodes [0]->store.unchecked_begin (transaction, block1.previous ()));
}
TEST (gap_cache, add_existing)
@ -15,16 +16,16 @@ TEST (gap_cache, add_existing)
rai::system system (24000, 1);
rai::gap_cache cache (*system.nodes [0]);
rai::send_block block1 (0, 1, 2, rai::keypair ().prv, 4, 5);
auto previous (block1.previous ());
cache.add (block1, previous);
auto existing1 (cache.blocks.find (previous));
ASSERT_NE (cache.blocks.end (), existing1);
rai::transaction transaction (system.nodes [0]->store.environment, nullptr, true);
cache.add (transaction, block1, block1.previous ());
auto existing1 (cache.blocks.get <1> ().find (block1.hash ()));
ASSERT_NE (cache.blocks.get <1> ().end (), existing1);
auto arrival (existing1->arrival);
while (arrival == std::chrono::system_clock::now ());
cache.add (block1, previous);
cache.add (transaction, block1, block1.previous ());
ASSERT_EQ (1, cache.blocks.size ());
auto existing2 (cache.blocks.find (previous));
ASSERT_NE (cache.blocks.end (), existing2);
auto existing2 (cache.blocks.get <1> ().find (block1.hash ()));
ASSERT_NE (cache.blocks.get <1> ().end (), existing2);
ASSERT_GT (existing2->arrival, arrival);
}
@ -33,18 +34,17 @@ TEST (gap_cache, comparison)
rai::system system (24000, 1);
rai::gap_cache cache (*system.nodes [0]);
rai::send_block block1 (1, 0, 2, rai::keypair ().prv, 4, 5);
auto previous1 (block1.previous ());
cache.add (rai::send_block (block1), previous1);
auto existing1 (cache.blocks.find (previous1));
ASSERT_NE (cache.blocks.end (), existing1);
rai::transaction transaction (system.nodes [0]->store.environment, nullptr, true);
cache.add (transaction, block1, block1.previous ());
auto existing1 (cache.blocks.get <1> ().find (block1.hash ()));
ASSERT_NE (cache.blocks.get <1> ().end (), existing1);
auto arrival (existing1->arrival);
while (std::chrono::system_clock::now () == arrival);
rai::send_block block3 (0, 42, 1, rai::keypair ().prv, 3, 4);
auto previous2 (block3.previous ());
cache.add (rai::send_block (block3), previous2);
cache.add (transaction, block3, block3.previous ());
ASSERT_EQ (2, cache.blocks.size ());
auto existing2 (cache.blocks.find (previous2));
ASSERT_NE (cache.blocks.end (), existing2);
auto existing2 (cache.blocks.get <1> ().find (block3.hash ()));
ASSERT_NE (cache.blocks.get <1> ().end (), existing2);
ASSERT_GT (existing2->arrival, arrival);
ASSERT_EQ (arrival, cache.blocks.get <1> ().begin ()->arrival);
}
@ -84,20 +84,11 @@ TEST (gap_cache, two_dependencies)
rai::send_block send2 (send1.hash (), key.pub, 0, rai::test_genesis_key.prv, rai::test_genesis_key.pub, system.work.generate (send1.hash ()));
rai::open_block open (send1.hash (), key.pub, key.pub, key.prv, key.pub, system.work.generate (key.pub));
ASSERT_EQ (0, system.nodes [0]->gap_cache.blocks.size ());
{
rai::transaction transaction (system.nodes [0]->store.environment, nullptr, true);
system.nodes [0]->process_receive_many (transaction, send2);
}
system.nodes [0]->process_receive_many (send2);
ASSERT_EQ (1, system.nodes [0]->gap_cache.blocks.size ());
{
rai::transaction transaction (system.nodes [0]->store.environment, nullptr, true);
system.nodes [0]->process_receive_many (transaction, open);
}
system.nodes [0]->process_receive_many (open);
ASSERT_EQ (2, system.nodes [0]->gap_cache.blocks.size ());
{
rai::transaction transaction (system.nodes [0]->store.environment, nullptr, true);
system.nodes [0]->process_receive_many (transaction, send1);
}
system.nodes [0]->process_receive_many (send1);
ASSERT_EQ (0, system.nodes [0]->gap_cache.blocks.size ());
rai::transaction transaction (system.nodes [0]->store.environment, nullptr, false);
ASSERT_TRUE (system.nodes [0]->store.block_exists (transaction, send1.hash ()));

19
rai/core_test/network.cpp
View file

@ -867,25 +867,6 @@ TEST (network, ipv6_bind_send_ipv4)
});
}
TEST (bootstrap_processor, unchecked_only)
{
rai::system system (24000, 1);
rai::node_init init1;
auto node1 (std::make_shared <rai::node> (init1, system.service, 24001, rai::unique_path (), system.alarm, system.logging, system.work));
rai::send_block block1 (node1->latest (rai::test_genesis_key.pub), rai::test_genesis_key.pub, 0, rai::test_genesis_key.prv, rai::test_genesis_key.pub, 0);
node1->store.unchecked_put (rai::transaction (node1->store.environment, nullptr, true), block1.hash (), block1);
node1->bootstrap_initiator.bootstrap (system.nodes [0]->network.endpoint ());
ASSERT_EQ (block1.previous (), node1->latest (rai::test_genesis_key.pub));
auto iterations (0);
while (node1->latest (rai::test_genesis_key.pub) == block1.previous ())
{
system.poll ();
++iterations;
ASSERT_LT (iterations, 200);
}
node1->stop ();
}
TEST (network, endpoint_bad_fd)
{
rai::system system (24000, 1);

2
rai/core_test/node.cpp
View file

@ -872,7 +872,7 @@ TEST (node, fork_bootstrap_flip)
rai::send_block send1 (latest, key1.pub, rai::genesis_amount - rai::Gxrb_ratio, rai::test_genesis_key.prv, rai::test_genesis_key.pub, system0.work.generate (latest));
rai::keypair key2;
rai::send_block send2 (latest, key2.pub, rai::genesis_amount - rai::Gxrb_ratio, rai::test_genesis_key.prv, rai::test_genesis_key.pub, system0.work.generate (latest));
// Insert but don't rebroadcast, simulating well-established blocks
// Insert but don't rebroadcast, simulating settled blocks
node1.process_receive_many (send1);
node2.process_receive_many (send2);
{

2
rai/core_test/rpc.cpp
View file

@ -1331,7 +1331,7 @@ TEST (rpc, version)
ASSERT_EQ (200, response1.status);
ASSERT_EQ ("1", response1.json.get <std::string> ("rpc_version"));
ASSERT_EQ (200, response1.status);
ASSERT_EQ ("6", response1.json.get <std::string> ("store_version"));
ASSERT_EQ ("7", response1.json.get <std::string> ("store_version"));
ASSERT_EQ (boost::str (boost::format ("RaiBlocks %1%.%2%.%3%") % RAIBLOCKS_VERSION_MAJOR % RAIBLOCKS_VERSION_MINOR % RAIBLOCKS_VERSION_PATCH), response1.json.get <std::string> ("node_vendor"));
auto & headers (response1.resp.fields);
auto access_control (std::find_if (headers.begin (), headers.end (), [] (decltype (*headers.begin ()) & header_a) { return boost::iequals (header_a.first, "Access-Control-Allow-Origin"); }));

115
rai/node/bootstrap.cpp
View file

@ -120,71 +120,6 @@ rai::sync_result rai::block_synchronization::synchronize (MDB_txn * transaction_
return result;
}
rai::pull_synchronization::pull_synchronization (rai::node & node_a, std::shared_ptr <rai::bootstrap_attempt> attempt_a) :
block_synchronization (node_a.log),
node (node_a),
attempt (attempt_a)
{
}
std::unique_ptr <rai::block> rai::pull_synchronization::retrieve (MDB_txn * transaction_a, rai::block_hash const & hash_a)
{
return node.store.unchecked_get (transaction_a, hash_a);
}
rai::sync_result rai::pull_synchronization::target (MDB_txn * transaction_a, rai::block const & block_a)
{
auto result (rai::sync_result::error);
node.process_receive_many (transaction_a, block_a, [this, transaction_a, &result] (rai::process_return result_a, rai::block const & block_a)
{
this->node.store.unchecked_del (transaction_a, block_a.hash ());
switch (result_a.code)
{
case rai::process_result::progress:
case rai::process_result::old:
result = rai::sync_result::success;
break;
case rai::process_result::fork:
{
result = rai::sync_result::fork;
auto node_l (this->node.shared ());
auto block (node_l->ledger.forked_block (transaction_a, block_a));
auto attempt_l (attempt);
node_l->active.start (transaction_a, *block, [node_l, attempt_l] (rai::block & block_a)
{
node_l->process_confirmed (block_a);
// Resume synchronizing after fork resolution
assert (node_l->bootstrap_initiator.in_progress ());
node_l->process_unchecked (attempt_l);
});
this->node.network.broadcast_confirm_req (block_a);
this->node.network.broadcast_confirm_req (*block);
BOOST_LOG (log) << boost::str (boost::format ("Fork received in bootstrap between: %1% and %2% root %3%") % block_a.hash ().to_string () % block->hash ().to_string () % block_a.root ().to_string ());
break;
}
case rai::process_result::gap_previous:
case rai::process_result::gap_source:
result = rai::sync_result::error;
if (this->node.config.logging.bulk_pull_logging ())
{
// Any activity while bootstrapping can cause gaps so these aren't as noteworthy
BOOST_LOG (log) << boost::str (boost::format ("Gap received in bootstrap for block: %1%") % block_a.hash ().to_string ());
}
break;
default:
result = rai::sync_result::error;
BOOST_LOG (log) << boost::str (boost::format ("Error inserting block in bootstrap: %1%") % block_a.hash ().to_string ());
break;
}
});
return result;
}
bool rai::pull_synchronization::synchronized (MDB_txn * transaction_a, rai::block_hash const & hash_a)
{
return node.store.block_exists (transaction_a, hash_a);
}
rai::push_synchronization::push_synchronization (rai::node & node_a, std::function <rai::sync_result (MDB_txn *, rai::block const &)> const & target_a) :
block_synchronization (node_a.log),
target_m (target_a),
@ -224,19 +159,11 @@ endpoint (endpoint_a)
rai::bootstrap_client::~bootstrap_client ()
{
if (node->config.logging.network_logging ())
{
BOOST_LOG (node->log) << boost::str (boost::format ("Exiting bootstrap client to %1%") % endpoint);
}
attempt->connection_ending (this);
}
void rai::bootstrap_client::run ()
{
if (node->config.logging.network_logging ())
{
BOOST_LOG (node->log) << boost::str (boost::format ("Initiating bootstrap connection to %1%") % endpoint);
}
auto this_l (shared_from_this ());
socket.async_connect (endpoint, [this_l] (boost::system::error_code const & ec)
{
@ -250,7 +177,16 @@ void rai::bootstrap_client::run ()
{
if (this_l->node->config.logging.network_logging ())
{
BOOST_LOG (this_l->node->log) << boost::str (boost::format ("Error initiating bootstrap connection to %2%: %1%") % ec.message () % this_l->endpoint);
switch (ec.value())
{
default:
BOOST_LOG (this_l->node->log) << boost::str (boost::format ("Error initiating bootstrap connection to %2%: %1%") % ec.message () % this_l->endpoint);
break;
case boost::system::errc::connection_refused:
case boost::system::errc::operation_canceled:
case boost::system::errc::timed_out:
break;
}
}
}
});
@ -262,7 +198,6 @@ void rai::bootstrap_client::run ()
{
if (!this_l->connected)
{
BOOST_LOG (this_l->node->log) << boost::str (boost::format ("Bootstrap disconnecting from: %1% because because of connection timeout") % this_l->endpoint);
this_l->socket.close ();
}
}
@ -739,12 +674,34 @@ void rai::bootstrap_pull_cache::flush (size_t minimum_a)
{
while (!blocks_l.empty ())
{
rai::transaction transaction (attempt.node->store.environment, nullptr, true);
auto count (0);
while (!blocks_l.empty () && count < rai::blocks_per_transaction)
{
auto & front (blocks_l.front ());
attempt.node->store.unchecked_put (transaction, front->hash(), *front);
attempt.node->process_receive_many (*front, [this] (MDB_txn * transaction_a, rai::process_return result_a, rai::block const & block_a)
{
switch (result_a.code)
{
case rai::process_result::progress:
case rai::process_result::old:
break;
case rai::process_result::fork:
{
auto node_l (attempt.node);
auto block (node_l->ledger.forked_block (transaction_a, block_a));
node_l->active.start (transaction_a, *block, [node_l] (rai::block & block_a)
{
node_l->process_confirmed (block_a);
});
attempt.node->network.broadcast_confirm_req (block_a);
attempt.node->network.broadcast_confirm_req (*block);
BOOST_LOG (attempt.node->log) << boost::str (boost::format ("Fork received in bootstrap between: %1% and %2% root %3%") % block_a.hash ().to_string () % block->hash ().to_string () % block_a.root ().to_string ());
break;
}
default:
break;
}
});
blocks_l.pop_front ();
++count;
}
@ -776,6 +733,7 @@ state (rai::attempt_state::starting)
rai::bootstrap_attempt::~bootstrap_attempt ()
{
cache.flush (0);
node->bootstrap_initiator.notify_listeners ();
BOOST_LOG (node->log) << "Exiting bootstrap attempt";
}
@ -910,11 +868,11 @@ void rai::bootstrap_attempt::completed_pull (std::shared_ptr <rai::bootstrap_cli
}
client_a->pull_client.pull = rai::pull_info ();
}
cache.flush (cache.block_count);
if (repool)
{
pool_connection (client_a);
}
cache.flush (cache.block_count);
}
void rai::bootstrap_attempt::completed_pulls (std::shared_ptr <rai::bootstrap_client> client_a)
@ -922,7 +880,6 @@ void rai::bootstrap_attempt::completed_pulls (std::shared_ptr <rai::bootstrap_cl
BOOST_LOG (node->log) << "Completed pulls";
assert (node->bootstrap_initiator.in_progress ());
cache.flush (0);
node->process_unchecked (shared_from_this ());
auto pushes (std::make_shared <rai::bulk_push_client> (client_a));
pushes->start ();
}

10
rai/node/bootstrap.hpp
View file

@ -37,16 +37,6 @@ public:
boost::log::sources::logger_mt & log;
std::deque <rai::block_hash> blocks;
};
class pull_synchronization : public rai::block_synchronization
{
public:
pull_synchronization (rai::node &, std::shared_ptr <rai::bootstrap_attempt>);
bool synchronized (MDB_txn *, rai::block_hash const &) override;
std::unique_ptr <rai::block> retrieve (MDB_txn *, rai::block_hash const &) override;
rai::sync_result target (MDB_txn *, rai::block const &) override;
rai::node & node;
std::shared_ptr <rai::bootstrap_attempt> attempt;
};
class push_synchronization : public rai::block_synchronization
{
public:

132
rai/node/node.cpp
View file

@ -1243,50 +1243,35 @@ node (node_a)
{
}
void rai::gap_cache::add (rai::block const & block_a, rai::block_hash needed_a)
void rai::gap_cache::add (MDB_txn * transaction_a, rai::block const & block_a, rai::block_hash const & hash_a)
{
auto hash (block_a.hash ());
std::lock_guard <std::mutex> lock (mutex);
auto existing (blocks.get <2>().find (hash));
if (existing != blocks.get <2> ().end ())
auto existing (blocks.get <1>().find (hash));
if (existing != blocks.get <1> ().end ())
{
blocks.get <2> ().modify (existing, [&block_a] (rai::gap_information & info)
blocks.get <1> ().modify (existing, [&block_a] (rai::gap_information & info)
{
info.arrival = std::chrono::system_clock::now ();
});
}
else
{
blocks.insert ({std::chrono::system_clock::now (), needed_a, hash, std::unique_ptr <rai::votes> (new rai::votes (block_a)), block_a.clone ()});
node.store.unchecked_put (transaction_a, hash_a, block_a);
blocks.insert ({std::chrono::system_clock::now (), hash, std::unique_ptr <rai::votes> (new rai::votes (block_a))});
if (blocks.size () > max)
{
blocks.get <1> ().erase (blocks.get <1> ().begin ());
blocks.get <0> ().erase (blocks.get <0> ().begin ());
}
}
}
std::vector <std::unique_ptr <rai::block>> rai::gap_cache::get (rai::block_hash const & hash_a)
{
purge_old ();
std::lock_guard <std::mutex> lock (mutex);
std::vector <std::unique_ptr <rai::block>> result;
for (auto i (blocks.find (hash_a)), n (blocks.end ()); i != n && i->required == hash_a; ++i)
{
blocks.modify (i, [&result] (rai::gap_information & info)
{
result.push_back (std::move (info.block));
});
}
blocks.erase (hash_a);
return result;
}
void rai::gap_cache::vote (rai::vote const & vote_a)
{
std::lock_guard <std::mutex> lock (mutex);
auto hash (vote_a.block->hash ());
auto existing (blocks.get <2> ().find (hash));
if (existing != blocks.get <2> ().end ())
auto existing (blocks.get <1> ().find (hash));
if (existing != blocks.get <1> ().end ())
{
existing->votes->vote (vote_a);
rai::transaction transaction (node.store.environment, nullptr, false);
@ -1306,10 +1291,6 @@ void rai::gap_cache::vote (rai::vote const & vote_a)
}
node_l->bootstrap_initiator.bootstrap ();
}
else
{
BOOST_LOG (node_l->log) << boost::str (boost::format ("Block: %1% was inserted while voting") % hash.to_string ());
}
});
}
}
@ -1366,16 +1347,15 @@ void rai::node::process_receive_republish (std::unique_ptr <rai::block> incoming
{
std::vector <std::tuple <rai::process_return, std::unique_ptr <rai::block>>> completed;
{
rai::transaction transaction (store.environment, nullptr, true);
assert (incoming != nullptr);
process_receive_many (transaction, *incoming, [this, &completed, &transaction] (rai::process_return result_a, rai::block const & block_a)
process_receive_many (*incoming, [this, &completed] (MDB_txn * transaction_a, rai::process_return result_a, rai::block const & block_a)
{
switch (result_a.code)
{
case rai::process_result::progress:
{
auto node_l (shared_from_this ());
active.start (transaction, block_a, [node_l] (rai::block & block_a)
active.start (transaction_a, block_a, [node_l] (rai::block & block_a)
{
node_l->process_confirmed (block_a);
});
@ -1395,26 +1375,41 @@ void rai::node::process_receive_republish (std::unique_ptr <rai::block> incoming
}
}
void rai::node::process_receive_many (rai::block const & block_a, std::function <void (rai::process_return, rai::block const &)> completed_a)
{
rai::transaction transaction (store.environment, nullptr, true);
process_receive_many (transaction, block_a, completed_a);
}
void rai::node::process_receive_many (MDB_txn * transaction_a, rai::block const & block_a, std::function <void (rai::process_return, rai::block const &)> completed_a)
void rai::node::process_receive_many (rai::block const & block_a, std::function <void (MDB_txn *, rai::process_return, rai::block const &)> completed_a)
{
std::vector <std::unique_ptr <rai::block>> blocks;
blocks.push_back (block_a.clone ());
while (!blocks.empty ())
{
auto block (std::move (blocks.back ()));
blocks.pop_back ();
auto hash (block->hash ());
auto process_result (process_receive_one (transaction_a, *block));
completed_a (process_result, *block);
auto cached (gap_cache.get (hash));
blocks.resize (blocks.size () + cached.size ());
std::move (cached.begin (), cached.end (), blocks.end () - cached.size ());
rai::transaction transaction (store.environment, nullptr, true);
auto count (0);
while (!blocks.empty () && count < rai::blocks_per_transaction)
{
auto block (std::move (blocks.back ()));
blocks.pop_back ();
auto hash (block->hash ());
auto process_result (process_receive_one (transaction, *block));
completed_a (transaction, process_result, *block);
switch (process_result.code)
{
case rai::process_result::progress:
case rai::process_result::old:
{
auto cached (store.unchecked_get (transaction, hash));
for (auto i (cached.begin ()), n (cached.end ()); i != n; ++i)
{
store.unchecked_del (transaction, hash, **i);
blocks.push_back (std::move (*i));
}
std::lock_guard <std::mutex> lock (gap_cache.mutex);
gap_cache.blocks.get <1> ().erase (hash);
break;
}
default:
break;
}
++count;
}
}
}
@ -1440,8 +1435,7 @@ rai::process_return rai::node::process_receive_one (MDB_txn * transaction_a, rai
{
BOOST_LOG (log) << boost::str (boost::format ("Gap previous for: %1%") % block_a.hash ().to_string ());
}
auto previous (block_a.previous ());
gap_cache.add (block_a, previous);
gap_cache.add (transaction_a, block_a, block_a.previous ());
break;
}
case rai::process_result::gap_source:
@ -1450,8 +1444,7 @@ rai::process_return rai::node::process_receive_one (MDB_txn * transaction_a, rai
{
BOOST_LOG (log) << boost::str (boost::format ("Gap source for: %1%") % block_a.hash ().to_string ());
}
auto source (block_a.source ());
gap_cache.add (block_a, source);
gap_cache.add (transaction_a, block_a, block_a.source ());
break;
}
case rai::process_result::old:
@ -2116,45 +2109,6 @@ public:
};
}
void rai::node::process_unchecked (std::shared_ptr <rai::bootstrap_attempt> attempt_a)
{
auto block_count (0);
assert (attempt_a == nullptr || bootstrap_initiator.in_progress ());
static std::atomic_flag unchecked_in_progress = ATOMIC_FLAG_INIT;
if (!unchecked_in_progress.test_and_set ())
{
BOOST_LOG (log) << "Starting to process unchecked blocks";
rai::pull_synchronization synchronization (*this, attempt_a);
auto done (false);
while (!done)
{
rai::transaction transaction (store.environment, nullptr, true);
auto next (store.unchecked_begin (transaction));
if (next != store.unchecked_end ())
{
auto block (rai::block_hash (next->first));
if (config.logging.bulk_pull_logging () && block_count % 64 == 0)
{
BOOST_LOG (log) << boost::str (boost::format ("Committing block: %1% and dependencies") % block.to_string ());
}
++block_count;
auto error (synchronization.synchronize (transaction, block));
if (error == rai::sync_result::fork)
{
done = true;
}
}
else
{
done = true;
}
}
BOOST_LOG (log) << "Completed processing unchecked blocks";
unchecked_in_progress.clear ();
wallets.search_pending_all ();
}
}
void rai::node::process_confirmed (rai::block const & confirmed_a)
{
confirmed_visitor visitor (*this);

10
rai/node/node.hpp
View file

@ -123,17 +123,14 @@ class gap_information
{
public:
std::chrono::system_clock::time_point arrival;
rai::block_hash required;
rai::block_hash hash;
std::unique_ptr <rai::votes> votes;
std::unique_ptr <rai::block> block;
};
class gap_cache
{
public:
gap_cache (rai::node &);
void add (rai::block const &, rai::block_hash);
std::vector <std::unique_ptr <rai::block>> get (rai::block_hash const &);
void add (MDB_txn *, rai::block const &, rai::block_hash const &);
void vote (rai::vote const &);
rai::uint128_t bootstrap_threshold (MDB_txn *);
void purge_old ();
@ -142,7 +139,6 @@ public:
rai::gap_information,
boost::multi_index::indexed_by
<
boost::multi_index::hashed_non_unique <boost::multi_index::member <gap_information, rai::block_hash, &gap_information::required>>,
boost::multi_index::ordered_non_unique <boost::multi_index::member <gap_information, std::chrono::system_clock::time_point, &gap_information::arrival>>,
boost::multi_index::hashed_unique <boost::multi_index::member <gap_information, rai::block_hash, &gap_information::hash>>
>
@ -417,12 +413,10 @@ public:
void stop ();
std::shared_ptr <rai::node> shared ();
int store_version ();
void process_unchecked (std::shared_ptr <rai::bootstrap_attempt>);
void process_confirmed (rai::block const &);
void process_message (rai::message &, rai::endpoint const &);
void process_receive_republish (std::unique_ptr <rai::block>);
void process_receive_many (rai::block const &, std::function <void (rai::process_return, rai::block const &)> = [] (rai::process_return, rai::block const &) {});
void process_receive_many (MDB_txn *, rai::block const &, std::function <void (rai::process_return, rai::block const &)> = [] (rai::process_return, rai::block const &) {});
void process_receive_many (rai::block const &, std::function <void (MDB_txn *, rai::process_return, rai::block const &)> = [] (MDB_txn *, rai::process_return, rai::block const &) {});
rai::process_return process_receive_one (MDB_txn *, rai::block const &);
rai::process_return process (rai::block const &);
void keepalive_preconfigured (std::vector <std::string> const &);

4
rai/qt/qt.cpp
View file

@ -631,7 +631,7 @@ std::string rai_qt::status::text ()
}
result += ", Block: ";
if (unchecked != 0)
if (unchecked != 0 && wallet.wallet_m->node.bootstrap_initiator.in_progress ())
{
count_string += " (" + std::to_string (unchecked) + ")";
}
@ -1355,7 +1355,7 @@ void rai_qt::advanced_actions::refresh_count ()
auto size (wallet.wallet_m->node.store.block_count (transaction));
auto unchecked (wallet.wallet_m->node.store.unchecked_count (transaction));
auto count_string (std::to_string (size.sum ()));
if (unchecked != 0)
if (unchecked != 0 && wallet.wallet_m->node.bootstrap_initiator.in_progress ())
{
count_string += " (" + std::to_string (unchecked) + ")";
}

39
rai/secure.cpp
View file

@ -1542,7 +1542,7 @@ checksum (0)
error_a |= mdb_dbi_open (transaction, "change", MDB_CREATE, &change_blocks) != 0;
error_a |= mdb_dbi_open (transaction, "pending", MDB_CREATE, &pending) != 0;
error_a |= mdb_dbi_open (transaction, "representation", MDB_CREATE, &representation) != 0;
error_a |= mdb_dbi_open (transaction, "unchecked", MDB_CREATE, &unchecked) != 0;
error_a |= mdb_dbi_open (transaction, "unchecked", MDB_CREATE | MDB_DUPSORT, &unchecked) != 0;
error_a |= mdb_dbi_open (transaction, "unsynced", MDB_CREATE, &unsynced) != 0;
error_a |= mdb_dbi_open (transaction, "checksum", MDB_CREATE, &checksum) != 0;
error_a |= mdb_dbi_open (transaction, "sequence", MDB_CREATE, &sequence) != 0;
@ -1597,6 +1597,8 @@ void rai::block_store::do_upgrades (MDB_txn * transaction_a)
case 5:
upgrade_v5_to_v6 (transaction_a);
case 6:
upgrade_v6_to_v7 (transaction_a);
case 7:
break;
default:
assert (false);
@ -1766,6 +1768,12 @@ void rai::block_store::upgrade_v5_to_v6 (MDB_txn * transaction_a)
}
}
void rai::block_store::upgrade_v6_to_v7 (MDB_txn * transaction_a)
{
version_put (transaction_a, 7);
mdb_drop (transaction_a, unchecked, 0);
}
void rai::block_store::clear (MDB_dbi db_a)
{
rai::transaction transaction (environment, nullptr, true);
@ -2326,24 +2334,25 @@ void rai::block_store::unchecked_put (MDB_txn * transaction_a, rai::block_hash c
assert (status == 0);
}
std::unique_ptr <rai::block> rai::block_store::unchecked_get (MDB_txn * transaction_a, rai::block_hash const & hash_a)
std::vector <std::unique_ptr <rai::block>> rai::block_store::unchecked_get (MDB_txn * transaction_a, rai::block_hash const & hash_a)
{
MDB_val value;
auto status (mdb_get (transaction_a, unchecked, hash_a.val (), &value));
assert (status == 0 || status == MDB_NOTFOUND);
std::unique_ptr <rai::block> result;
if (status == 0)
{
rai::bufferstream stream (reinterpret_cast <uint8_t const *> (value.mv_data), value.mv_size);
result = rai::deserialize_block (stream);
assert (result != nullptr);
}
return result;
std::vector <std::unique_ptr <rai::block>> result;
for (auto i (unchecked_begin (transaction_a, hash_a)), n (unchecked_end ()); i != n && rai::block_hash (i->first) == hash_a; ++i)
{
rai::bufferstream stream (reinterpret_cast <uint8_t const *> (i->second.mv_data), i->second.mv_size);
result.push_back (rai::deserialize_block (stream));
}
return result;
}
void rai::block_store::unchecked_del (MDB_txn * transaction_a, rai::block_hash const & hash_a)
void rai::block_store::unchecked_del (MDB_txn * transaction_a, rai::block_hash const & hash_a, rai::block const & block_a)
{
auto status (mdb_del (transaction_a, unchecked, hash_a.val (), nullptr));
std::vector <uint8_t> vector;
{
rai::vectorstream stream (vector);
rai::serialize_block (stream, block_a);
}
auto status (mdb_del (transaction_a, unchecked, hash_a.val (), rai::mdb_val (vector.size (), vector.data ())));
assert (status == 0 || status == MDB_NOTFOUND);
}

5
rai/secure.hpp
View file

@ -365,8 +365,8 @@ public:
void unchecked_clear (MDB_txn *);
void unchecked_put (MDB_txn *, rai::block_hash const &, rai::block const &);
std::unique_ptr <rai::block> unchecked_get (MDB_txn *, rai::block_hash const &);
void unchecked_del (MDB_txn *, rai::block_hash const &);
std::vector <std::unique_ptr <rai::block>> unchecked_get (MDB_txn *, rai::block_hash const &);
void unchecked_del (MDB_txn *, rai::block_hash const &, rai::block const &);
rai::store_iterator unchecked_begin (MDB_txn *);
rai::store_iterator unchecked_begin (MDB_txn *, rai::block_hash const &);
rai::store_iterator unchecked_end ();
@ -394,6 +394,7 @@ public:
void upgrade_v3_to_v4 (MDB_txn *);
void upgrade_v4_to_v5 (MDB_txn *);
void upgrade_v5_to_v6 (MDB_txn *);
void upgrade_v6_to_v7 (MDB_txn *);
void clear (MDB_dbi);

13
rai/slow_test/node.cpp
View file

@ -230,19 +230,6 @@ TEST (node, fork_storm)
ASSERT_TRUE (true);
}
TEST (gap_cache, limit)
{
rai::system system (24000, 1);
rai::gap_cache cache (*system.nodes [0]);
for (auto i (0); i < cache.max * 2; ++i)
{
rai::send_block block1 (i, 0, 1, rai::keypair ().prv, 3, 4);
auto previous (block1.previous ());
cache.add (rai::send_block (block1), previous);
}
ASSERT_EQ (cache.max, cache.blocks.size ());
}
namespace
{
size_t heard_count (std::vector <uint8_t> const & nodes)