Skip to content
This repository was archived by the owner on Oct 4, 2019. It is now read-only.

Commit a556538

Browse files
kotbegemotafalaleev
kotbegemot
authored and
afalaleev
committed
Merge pull request #500 from GolosChain/499-configurable-read-write-locks
Configurable read write locks. #499
2 parents a2e3455 + a3e0b5f commit a556538

13 files changed

Lines changed: 224 additions & 55 deletions

File tree

‎libraries/chain/database.cpp‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -101,7 +101,7 @@ namespace golos {
101101

102102
if (chainbase_flags & chainbase::database::read_write) {
103103
if (!find<dynamic_global_property_object>()) {
104-
with_write_lock([&]() {
104+
with_strong_write_lock([&]() {
105105
init_genesis(initial_supply);
106106
});
107107
}
@@ -111,7 +111,7 @@ namespace golos {
111111
auto log_head = _block_log.head();
112112

113113
// Rewind all undo state. This should return us to the state at the last irreversible block.
114-
with_write_lock([&]() {
114+
with_strong_write_lock([&]() {
115115
undo_all();
116116
FC_ASSERT(revision() ==
117117
head_block_num(), "Chainbase revision does not match head block num",
@@ -161,7 +161,7 @@ namespace golos {
161161
skip_validate_invariants |
162162
skip_block_log;
163163

164-
with_write_lock([&]() {
164+
with_strong_write_lock([&]() {
165165
auto itr = _block_log.read_block(0);
166166
auto last_block_num = _block_log.head()->block_num();
167167

@@ -628,7 +628,7 @@ namespace golos {
628628
//fc::time_point begin_time = fc::time_point::now();
629629

630630
bool result;
631-
with_write_lock([&]() {
631+
with_strong_write_lock([&]() {
632632
detail::with_skip_flags(*this, skip, [&]() {
633633
detail::without_pending_transactions(*this, std::move(_pending_tx), [&]() {
634634
try {
@@ -756,7 +756,7 @@ namespace golos {
756756
void database::push_transaction(const signed_transaction &trx, uint32_t skip) {
757757
try {
758758
FC_ASSERT(fc::raw::pack_size(trx) <= (get_dynamic_global_properties().maximum_block_size - 256));
759-
with_write_lock([&]() {
759+
with_weak_write_lock([&]() {
760760
detail::with_producing(*this, [&]() {
761761
detail::with_skip_flags(*this, skip, [&]() {
762762
_push_transaction(trx);
@@ -830,7 +830,7 @@ namespace golos {
830830

831831
signed_block pending_block;
832832

833-
with_write_lock([&]() { detail::with_skip_flags(*this, skip, [&]() {
833+
with_strong_write_lock([&]() { detail::with_skip_flags(*this, skip, [&]() {
834834
//
835835
// The following code throws away existing pending_tx_session and
836836
// rebuilds it by re-applying pending transactions.
@@ -2930,7 +2930,7 @@ namespace golos {
29302930

29312931

29322932
void database::validate_transaction(const signed_transaction &trx) {
2933-
database::with_write_lock([&]() {
2933+
database::with_weak_write_lock([&]() {
29342934
auto session = start_undo_session(true);
29352935
_apply_transaction(trx);
29362936
session.undo();

‎plugins/chain/include/golos/plugins/chain/plugin.hpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -41,7 +41,7 @@ namespace golos {
4141

4242
void plugin_shutdown() override;
4343

44-
bool accept_block(const protocol::signed_block &block, bool currently_syncing, uint32_t skip);
44+
bool accept_block(const protocol::signed_block &block, bool currently_syncing = false, uint32_t skip = 0);
4545

4646
void accept_transaction(const protocol::signed_transaction &trx);
4747

‎plugins/chain/plugin.cpp‎

Lines changed: 145 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
#include <iostream>
99
#include <golos/protocol/protocol.hpp>
1010
#include <golos/protocol/types.hpp>
11+
#include <future>
1112

1213
namespace golos {
1314
namespace plugins {
@@ -31,11 +32,34 @@ namespace chain {
3132

3233
uint32_t allow_future_time = 5;
3334

35+
uint64_t read_wait_micro;
36+
uint32_t max_read_wait_retries;
37+
38+
uint64_t write_wait_micro;
39+
uint32_t max_write_wait_retries;
40+
41+
golos::chain::database db;
42+
43+
bool single_write_thread = false;
44+
45+
plugin_impl() {
46+
// get default settings
47+
read_wait_micro = db.read_wait_micro();
48+
max_read_wait_retries = db.max_read_wait_retries();
49+
50+
write_wait_micro = db.write_wait_micro();
51+
max_write_wait_retries = db.max_write_wait_retries();
52+
}
53+
3454
// HELPERS
3555
golos::chain::database &database() {
3656
return db;
3757
}
3858

59+
boost::asio::io_service& io_service() {
60+
return appbase::app().get_io_service();
61+
}
62+
3963
constexpr const static char *plugin_name = "chain_api";
4064
static const std::string &name() {
4165
static std::string name = plugin_name;
@@ -45,9 +69,6 @@ namespace chain {
4569
void check_time_in_block(const protocol::signed_block &block);
4670
bool accept_block(const protocol::signed_block &block, bool currently_syncing, uint32_t skip);
4771
void accept_transaction(const protocol::signed_transaction &trx);
48-
49-
50-
golos::chain::database db;
5172
};
5273

5374
void plugin::plugin_impl::check_time_in_block(const protocol::signed_block &block) {
@@ -66,11 +87,40 @@ namespace chain {
6687

6788
check_time_in_block(block);
6889

69-
return db.push_block(block, skip);
90+
if (single_write_thread) {
91+
std::promise<bool> promise;
92+
auto result = promise.get_future();
93+
94+
io_service().post([&]{
95+
try {
96+
promise.set_value(db.push_block(block, skip));
97+
} catch(...) {
98+
promise.set_exception(std::current_exception());
99+
}
100+
});
101+
return result.get(); // if an exception was, it will be thrown
102+
} else {
103+
return db.push_block(block, skip);
104+
}
70105
}
71106

72107
void plugin::plugin_impl::accept_transaction(const protocol::signed_transaction &trx) {
73-
db.push_transaction(trx);
108+
if (single_write_thread) {
109+
std::promise<bool> promise;
110+
auto wait = promise.get_future();
111+
112+
io_service().post([&]{
113+
try {
114+
db.push_transaction(trx);
115+
promise.set_value(true);
116+
} catch(...) {
117+
promise.set_exception(std::current_exception());
118+
}
119+
});
120+
wait.get(); // if an exception was, it will be thrown
121+
} else {
122+
db.push_transaction(trx);
123+
}
74124
}
75125

76126
plugin::plugin() {
@@ -89,23 +139,73 @@ namespace chain {
89139

90140
void plugin::set_program_options(boost::program_options::options_description &cli,
91141
boost::program_options::options_description &cfg) {
92-
cfg.add_options()("shared-file-dir", boost::program_options::value<boost::filesystem::path>()->default_value("blockchain"),
93-
"the location of the chain shared memory files (absolute path or relative to application data dir)")(
94-
"shared-file-size", boost::program_options::value<std::string>()->default_value("54G"),
95-
"Size of the shared memory file. Default: 54G")("checkpoint,c",
96-
boost::program_options::value<std::vector<std::string>>()->composing(),
97-
"Pairs of [BLOCK_NUM,BLOCK_ID] that should be enforced as checkpoints.")(
98-
"flush-state-interval", boost::program_options::value<uint32_t>(),
99-
"flush shared memory changes to disk every N blocks");
100-
cli.add_options()("replay-blockchain", boost::program_options::bool_switch()->default_value(false),
101-
"clear chain database and replay all blocks")("resync-blockchain",
102-
boost::program_options::bool_switch()->default_value(
103-
false),
104-
"clear chain database and block log")(
105-
"check-locks", boost::program_options::bool_switch()->default_value(false),
106-
"Check correctness of chainbase locking")("validate-database-invariants",
107-
boost::program_options::bool_switch()->default_value(false),
108-
"Validate all supply invariants check out");
142+
cfg.add_options()
143+
(
144+
"shared-file-dir",
145+
boost::program_options::value<boost::filesystem::path>()->default_value("blockchain"),
146+
"the location of the chain shared memory files (absolute path or relative to application data dir)"
147+
)
148+
(
149+
"shared-file-size",
150+
boost::program_options::value<std::string>()->default_value("64G"),
151+
"Size of the shared memory file. Default: 54G"
152+
)
153+
(
154+
"checkpoint,c",
155+
boost::program_options::value<std::vector<std::string>>()->composing(),
156+
"Pairs of [BLOCK_NUM,BLOCK_ID] that should be enforced as checkpoints."
157+
)
158+
(
159+
"flush-state-interval",
160+
boost::program_options::value<uint32_t>(),
161+
"flush shared memory changes to disk every N blocks"
162+
)
163+
(
164+
"read-wait-micro",
165+
boost::program_options::value<uint64_t>(),
166+
"maximum microseconds for trying to get read lock"
167+
)
168+
(
169+
"max-read-wait-retries",
170+
boost::program_options::value<uint32_t>(),
171+
"maximum number of retries to get read lock"
172+
)
173+
(
174+
"write-wait-micro",
175+
boost::program_options::value<uint64_t>(),
176+
"maximum microseconds for trying to get write lock"
177+
)
178+
(
179+
"max-write-wait-retries",
180+
boost::program_options::value<uint32_t>(),
181+
"maximum number of retries to get write lock"
182+
)
183+
(
184+
"single-write-thread",
185+
boost::program_options::value<bool>()->default_value(false),
186+
"push blocks and transactions from one thread"
187+
);
188+
cli.add_options()
189+
(
190+
"replay-blockchain",
191+
boost::program_options::bool_switch()->default_value(false),
192+
"clear chain database and replay all blocks"
193+
)
194+
(
195+
"resync-blockchain",
196+
boost::program_options::bool_switch()->default_value(false),
197+
"clear chain database and block log"
198+
)
199+
(
200+
"check-locks",
201+
boost::program_options::bool_switch()->default_value(false),
202+
"Check correctness of chainbase locking"
203+
)
204+
(
205+
"validate-database-invariants",
206+
boost::program_options::bool_switch()->default_value(false),
207+
"Validate all supply invariants check out"
208+
);
109209
}
110210

111211
void plugin::plugin_initialize(const boost::program_options::variables_map &options) {
@@ -122,6 +222,24 @@ namespace chain {
122222
}
123223
}
124224

225+
if (options.count("read-wait-micro")) {
226+
my->read_wait_micro = options.at("read-wait-micro").as<uint64_t>();
227+
}
228+
229+
if (options.count("max-read-wait-retries")) {
230+
my->max_read_wait_retries = options.at("max-read-wait-retries").as<uint32_t>();
231+
}
232+
233+
if (options.count("write-wait-micro")) {
234+
my->write_wait_micro = options.at("write-wait-micro").as<uint64_t>();
235+
}
236+
237+
if (options.count("max-write-wait-retries")) {
238+
my->max_write_wait_retries = options.at("max-write-wait-retries").as<uint32_t>();
239+
}
240+
241+
my->single_write_thread = options.at("single-write-thread").as<bool>();
242+
125243
my->shared_memory_size = fc::parse_size(options.at("shared-file-size").as<std::string>());
126244

127245
my->replay = options.at("replay-blockchain").as<bool>();
@@ -156,6 +274,11 @@ namespace chain {
156274
my->db.add_checkpoints(my->loaded_checkpoints);
157275
my->db.set_require_locking(my->check_locks);
158276

277+
my->db.read_wait_micro(my->read_wait_micro);
278+
my->db.max_read_wait_retries(my->max_read_wait_retries);
279+
my->db.write_wait_micro(my->write_wait_micro);
280+
my->db.max_write_wait_retries(my->max_write_wait_retries);
281+
159282
if (my->replay) {
160283
ilog("Replaying blockchain on user request.");
161284
my->db.reindex(appbase::app().data_dir() / "blockchain", my->shared_memory_dir, my->shared_memory_size);

‎plugins/database_api/api.cpp‎

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -126,13 +126,15 @@ namespace golos {
126126
if (!_follow_api) {
127127
return 0;
128128
}
129-
msg_pack msg;
130-
msg.args = std::vector<fc::variant>({fc::variant(account), fc::variant(1)});
131-
auto reputations = _follow_api->get_account_reputations(msg);
132-
if (reputations.empty()) {
133-
return 0;
129+
130+
auto &rep_idx = database().get_index<follow::reputation_index>().indices().get<follow::by_account>();
131+
auto itr = rep_idx.find(account);
132+
133+
if (rep_idx.end() != itr) {
134+
return itr->reputation;
134135
}
135-
return reputations[0].reputation;
136+
137+
return 0;
136138
}
137139
138140
template<typename T>

‎plugins/network_broadcast_api/network_broadcast_api.cpp‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,7 @@ namespace golos {
4646
const auto max_block_age = args.args->at(1).as<uint32_t>();
4747
FC_ASSERT(!check_max_block_age(max_block_age));
4848
}
49-
pimpl->_chain.db().push_transaction(trx);
49+
pimpl->_chain.accept_transaction(trx);
5050
pimpl->_p2p.broadcast_transaction(trx);
5151

5252
return broadcast_transaction_return();
@@ -73,7 +73,7 @@ namespace golos {
7373
pimpl->_callback_expirations[trx.expiration].push_back(trx.id());
7474
}
7575

76-
pimpl->_chain.db().push_transaction(trx);
76+
pimpl->_chain.accept_transaction(trx);
7777
pimpl->_p2p.broadcast_transaction(trx);
7878
transfer.complete();
7979

@@ -84,7 +84,7 @@ namespace golos {
8484
const auto n_args = args.args->size();
8585
FC_ASSERT(n_args == 1, "Expected 1 argument, got 0");
8686
auto block = args.args->at(0).as<signed_block>();
87-
pimpl->_chain.db().push_block(block);
87+
pimpl->_chain.accept_block(block);
8888
pimpl->_p2p.broadcast_block(block);
8989
return broadcast_block_return();
9090
}
@@ -116,7 +116,7 @@ namespace golos {
116116
}
117117

118118

119-
pimpl->_chain.db().push_transaction(trx);
119+
pimpl->_chain.accept_transaction(trx);
120120
pimpl->_p2p.broadcast_transaction(trx);
121121
transfer.complete();
122122

‎plugins/p2p/p2p_plugin.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -161,7 +161,7 @@ namespace golos {
161161

162162
void p2p_plugin_impl::handle_transaction(const trx_message &trx_msg) {
163163
try {
164-
chain.db().push_transaction(trx_msg.trx);
164+
chain.accept_transaction(trx_msg.trx);
165165
} FC_CAPTURE_AND_RETHROW((trx_msg))
166166
}
167167

‎plugins/social_network/social_network.cpp‎

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -120,13 +120,15 @@ namespace golos {
120120
if (!follow_api_) {
121121
return 0;
122122
}
123-
msg_pack msg;
124-
msg.args = std::vector<fc::variant>({fc::variant(account), fc::variant(1)});
125-
auto reputations = follow_api_->get_account_reputations(msg);
126-
if (reputations.empty()) {
127-
return 0;
123+
124+
auto &rep_idx = database().get_index<follow::reputation_index>().indices().get<follow::by_account>();
125+
auto itr = rep_idx.find(account);
126+
127+
if (rep_idx.end() != itr) {
128+
return itr->reputation;
128129
}
129-
return reputations[0].reputation;
130+
131+
return 0;
130132
}
131133
132134
comment_object::id_type get_parent(const discussion_query &query) const {

0 commit comments

Comments
 (0)