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

Changing push to use new synchronization. Removing block_path class.

Browse source
This commit is contained in:
clemahieu 2015-01-05 20:40:11 -06:00
commit 469f062c8a
4 changed files with 14 additions and 245 deletions
Download patch file Download diff file Expand all files Collapse all files

1
CMakeLists.txt
View file

@ -48,7 +48,6 @@ add_library (core
add_executable (core_test
rai/core_test/block.cpp
rai/core_test/block_path.cpp
rai/core_test/block_store.cpp
rai/core_test/block_synchronization.cpp
rai/core_test/client.cpp

91
rai/core/core.cpp
View file

@ -3519,65 +3519,6 @@ std::unique_ptr <rai::block> rai::push_synchronization::retrieve (rai::block_has
return store.block_get (hash_a);
}
rai::block_path::block_path (std::vector <std::unique_ptr <rai::block>> & path_a, std::function <std::unique_ptr <rai::block> (rai::block_hash const &)> const & retrieve_a) :
path (path_a),
retrieve (retrieve_a)
{
}
void rai::block_path::send_block (rai::send_block const & block_a)
{
auto block (retrieve (block_a.hashables.previous));
if (block != nullptr)
{
path.push_back (std::move (block));
}
}
void rai::block_path::receive_block (rai::receive_block const & block_a)
{
rai::block_path path_l (path, retrieve);
path_l.generate (block_a.hashables.source);
auto block (retrieve (block_a.hashables.previous));
if (block != nullptr)
{
path.push_back (std::move (block));
}
}
void rai::block_path::open_block (rai::open_block const & block_a)
{
auto block (retrieve (block_a.hashables.source));
if (block != nullptr)
{
path.push_back (std::move (block));
}
}
void rai::block_path::change_block (rai::change_block const & block_a)
{
auto block (retrieve (block_a.hashables.previous));
if (block != nullptr)
{
path.push_back (std::move (block));
}
}
void rai::block_path::generate (rai::block_hash const & hash_a)
{
auto block (retrieve (hash_a));
if (block != nullptr)
{
path.push_back (std::move (block));
auto previous_size (0);
while (previous_size != path.size ())
{
previous_size = path.size ();
path.back ()->visit (*this);
}
}
}
void rai::bulk_pull_client::process_end ()
{
rai::pull_synchronization synchronization ([this] (rai::block const & block_a)
@ -4275,7 +4216,11 @@ void rai::frontier_req_client::completed_pushes ()
rai::bulk_push_client::bulk_push_client (std::shared_ptr <rai::frontier_req_client> const & connection_a) :
connection (connection_a),
current (connection->pushes.begin ()),
end (connection->pushes.end ())
end (connection->pushes.end ()),
synchronization ([this] (rai::block const & block_a)
{
push_block (block_a);
}, connection_a->connection->client->store)
{
}
@ -4313,24 +4258,14 @@ void rai::bulk_push_client::push ()
{
if (current != end)
{
path.clear ();
rai::block_path filler (path, [this] (rai::block_hash const & hash_a)
{
std::unique_ptr <rai::block> result;
auto block (connection->connection->client->store.block_get (hash_a));
if (block != nullptr)
{
result = std::move (block);
}
return result;
});
auto hash (current->first);
rai::frontier frontier;
auto error (connection->connection->client->store.latest_get (hash, frontier));
assert (!error);
++current;
filler.generate (frontier.hash);
push_block ();
assert (synchronization.blocks.empty ());
synchronization.blocks.push (frontier.hash);
synchronization.synchronize_one ();
}
else
{
@ -4353,23 +4288,21 @@ void rai::bulk_push_client::send_finished ()
});
}
void rai::bulk_push_client::push_block ()
void rai::bulk_push_client::push_block (rai::block const & block_a)
{
assert (!path.empty ());
auto buffer (std::make_shared <std::vector <uint8_t>> ());
{
rai::vectorstream stream (*buffer);
rai::serialize_block (stream, *path.back ());
rai::serialize_block (stream, block_a);
}
path.pop_back ();
auto this_l (shared_from_this ());
boost::asio::async_write (connection->connection->socket, boost::asio::buffer (buffer->data (), buffer->size ()), [this_l] (boost::system::error_code const & ec, size_t size_a)
{
if (!ec)
{
if (!this_l->path.empty ())
if (!this_l->synchronization.blocks.empty ())
{
this_l->push_block ();
this_l->synchronization.synchronize_one ();
}
else
{

16
rai/core/core.hpp
View file

@ -440,18 +440,6 @@ public:
bool synchronized (rai::block_hash const &) override;
std::unique_ptr <rai::block> retrieve (rai::block_hash const &) override;
};
class block_path : public rai::block_visitor
{
public:
block_path (std::vector <std::unique_ptr <rai::block>> &, std::function <std::unique_ptr <rai::block> (rai::block_hash const &)> const &);
void generate (rai::block_hash const &);
void send_block (rai::send_block const &);
void receive_block (rai::receive_block const &);
void open_block (rai::open_block const &);
void change_block (rai::change_block const &);
std::vector <std::unique_ptr <rai::block>> & path;
std::function <std::unique_ptr <rai::block> (rai::block_hash const &)> retrieve;
};
class bootstrap_client : public std::enable_shared_from_this <bootstrap_client>
{
public:
@ -504,12 +492,12 @@ public:
~bulk_push_client ();
void start ();
void push ();
void push_block ();
void push_block (rai::block const &);
void send_finished ();
std::shared_ptr <rai::frontier_req_client> connection;
std::unordered_map <rai::account, rai::block_hash>::iterator current;
std::unordered_map <rai::account, rai::block_hash>::iterator end;
std::vector <std::unique_ptr <rai::block>> path;
rai::push_synchronization synchronization;
};
class message_parser
{

151
rai/core_test/block_path.cpp
View file

@ -1,151 +0,0 @@
#include <gtest/gtest.h>
#include <rai/core/core.hpp>
TEST (block_path, zero)
{
std::vector <std::unique_ptr <rai::block>> path;
std::unordered_map <rai::block_hash, std::unique_ptr <rai::block>> blocks;
rai::block_path block_path (path, [&blocks] (rai::block_hash const & hash_a)
{
std::unique_ptr <rai::block> result;
auto existing (blocks.find (hash_a));
if (existing != blocks.end ())
{
result = std::move (existing->second);
blocks.erase (existing);
}
return result;
});
block_path.generate (0);
ASSERT_EQ (0, path.size ());
ASSERT_EQ (0, blocks.size ());
}
TEST (block_path, one)
{
std::vector <std::unique_ptr <rai::block>> path;
std::unordered_map <rai::block_hash, std::unique_ptr <rai::block>> blocks;
std::unique_ptr <rai::block> block1 (new rai::send_block);
auto hash1 (block1->hash ());
blocks [hash1] = block1->clone ();
rai::block_path block_path (path, [&blocks] (rai::block_hash const & hash_a)
{
std::unique_ptr <rai::block> result;
auto existing (blocks.find (hash_a));
if (existing != blocks.end ())
{
result = std::move (existing->second);
blocks.erase (existing);
}
return result;
});
block_path.generate (hash1);
ASSERT_EQ (1, path.size ());
ASSERT_EQ (0, blocks.size ());
ASSERT_EQ (*block1, *path [0]);
}
TEST (block_path, two)
{
std::vector <std::unique_ptr <rai::block>> path;
std::unordered_map <rai::block_hash, std::unique_ptr <rai::block>> blocks;
std::unique_ptr <rai::send_block> block1 (new rai::send_block);
auto hash1 (block1->hash ());
blocks [hash1] = block1->clone ();
std::unique_ptr <rai::send_block> block2 (new rai::send_block);
block2->hashables.previous = hash1;
auto hash2 (block2->hash ());
blocks [hash2] = block2->clone ();
rai::block_path block_path (path, [&blocks] (rai::block_hash const & hash_a)
{
std::unique_ptr <rai::block> result;
auto existing (blocks.find (hash_a));
if (existing != blocks.end ())
{
result = std::move (existing->second);
blocks.erase (existing);
}
return result;
});
block_path.generate (hash2);
ASSERT_EQ (2, path.size ());
ASSERT_EQ (0, blocks.size ());
ASSERT_EQ (*block2, *path [0]);
ASSERT_EQ (*block1, *path [1]);
}
TEST (block_path, receive_one)
{
std::vector <std::unique_ptr <rai::block>> path;
std::unordered_map <rai::block_hash, std::unique_ptr <rai::block>> blocks;
std::unique_ptr <rai::send_block> block1 (new rai::send_block);
block1->hashables.previous = 1;
auto hash1 (block1->hash ());
blocks [hash1] = block1->clone ();
std::unique_ptr <rai::send_block> block2 (new rai::send_block);
block2->hashables.previous = 2;
auto hash2 (block2->hash ());
blocks [hash2] = block2->clone ();
std::unique_ptr <rai::receive_block> block3 (new rai::receive_block);
block3->hashables.previous = hash1;
block3->hashables.source = hash2;
auto hash3 (block3->hash ());
blocks [hash3] = block3->clone ();
rai::block_path block_path (path, [&blocks] (rai::block_hash const & hash_a)
{
std::unique_ptr <rai::block> result;
auto existing (blocks.find (hash_a));
if (existing != blocks.end ())
{
result = std::move (existing->second);
blocks.erase (existing);
}
return result;
});
block_path.generate (hash3);
ASSERT_EQ (3, path.size ());
ASSERT_EQ (0, blocks.size ());
ASSERT_EQ (*block3, *path [0]);
ASSERT_EQ (*block2, *path [1]);
ASSERT_EQ (*block1, *path [2]);
}
TEST (block_path, receive_two)
{
std::vector <std::unique_ptr <rai::block>> path;
std::unordered_map <rai::block_hash, std::unique_ptr <rai::block>> blocks;
std::unique_ptr <rai::send_block> block1 (new rai::send_block);
block1->hashables.previous = 1;
auto hash1 (block1->hash ());
blocks [hash1] = block1->clone ();
std::unique_ptr <rai::send_block> block4 (new rai::send_block);
auto hash4 (block4->hash ());
blocks [hash4] = block4->clone ();
std::unique_ptr <rai::send_block> block2 (new rai::send_block);
block2->hashables.previous = hash4;
auto hash2 (block2->hash ());
blocks [hash2] = block2->clone ();
std::unique_ptr <rai::receive_block> block3 (new rai::receive_block);
block3->hashables.previous = hash1;
block3->hashables.source = hash2;
auto hash3 (block3->hash ());
blocks [hash3] = block3->clone ();
rai::block_path block_path (path, [&blocks] (rai::block_hash const & hash_a)
{
std::unique_ptr <rai::block> result;
auto existing (blocks.find (hash_a));
if (existing != blocks.end ())
{
result = std::move (existing->second);
blocks.erase (existing);
}
return result;
});
block_path.generate (hash3);
ASSERT_EQ (4, path.size ());
ASSERT_EQ (0, blocks.size ());
ASSERT_EQ (*block3, *path [0]);
ASSERT_EQ (*block2, *path [1]);
ASSERT_EQ (*block4, *path [2]);
ASSERT_EQ (*block1, *path [3]);
}