thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-11-24 08:32:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

Some fixes

Browse Source
This commit is contained in:
alesapin 2021-01-25 13:19:02 +03:00
parent 173b6fefdf
commit 7ff04d7532
4 changed files with 16 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/Coordination/LoggerWrapper.h
View File

@ -11,7 +11,9 @@ class LoggerWrapper : public nuraft::logger
public:
LoggerWrapper(const std::string & name)
: log(&Poco::Logger::get(name))
{}
{
set_level(4);
}
void put_details(
int level,
@ -25,7 +27,7 @@ public:
void set_level(int level) override
{
level = std::max(6, std::min(1, level));
level = std::min(6, std::max(1, level));
log->setLevel(level);
}

4
src/Coordination/NuKeeperServer.cpp
View File

@ -39,7 +39,7 @@ NuraftError NuKeeperServer::startup()
params.election_timeout_lower_bound_ = 200;
params.election_timeout_upper_bound_ = 400;
params.reserved_log_items_ = 5;
params.snapshot_distance_ = 5;
params.snapshot_distance_ = 50;
params.client_req_timeout_ = 3000;
params.return_method_ = nuraft::raft_params::blocking;
@ -127,12 +127,10 @@ TestKeeperStorage::ResponsesForSessions NuKeeperServer::readZooKeeperResponses(n
TestKeeperStorage::ResponsesForSessions NuKeeperServer::putRequests(const TestKeeperStorage::RequestsForSessions & requests)
{
std::vector<nuraft::ptr<nuraft::buffer>> entries;
LOG_DEBUG(&Poco::Logger::get("DEBUG"), "REQUESTS SIZE {}", requests.size());
for (auto & [session_id, request] : requests)
{
ops_mapping[session_id][request->xid] = request->makeResponse();
entries.push_back(getZooKeeperLogEntry(session_id, request));
LOG_DEBUG(&Poco::Logger::get("DEBUG"), "ENTRY SIZE {}", entries.back()->size());
}
auto result = raft_instance->append_entries(entries);

4
src/Coordination/NuKeeperStateMachine.cpp
View File

@ -50,7 +50,7 @@ NuKeeperStateMachine::NuKeeperStateMachine()
nuraft::ptr<nuraft::buffer> NuKeeperStateMachine::commit(const size_t log_idx, nuraft::buffer & data)
{
LOG_DEBUG(log, "Commiting logidx {}", log_idx);
//LOG_DEBUG(log, "Commiting logidx {}", log_idx);
if (data.size() == sizeof(size_t))
{
LOG_DEBUG(log, "Session ID response {}", log_idx);
@ -72,9 +72,9 @@ nuraft::ptr<nuraft::buffer> NuKeeperStateMachine::commit(const size_t log_idx, n
std::lock_guard lock(storage_lock);
responses_for_sessions = storage.processRequest(request_for_session.request, request_for_session.session_id);
}
//LOG_DEBUG(log, "TOTAL RESPONSES {} FIRST XID {}", responses_for_sessions.size(), responses_for_sessions[0].response->xid);
last_committed_idx = log_idx;
//LOG_DEBUG(log, "TOTAL RESPONSES {} FIRST XID {} FOR LOG IDX {}", responses_for_sessions.size(), responses_for_sessions[0].response->xid, log_idx);
return writeResponses(responses_for_sessions);
}
}

15
src/Coordination/WriteBufferFromNuraftBuffer.cpp
View File

@ -14,15 +14,18 @@ void WriteBufferFromNuraftBuffer::nextImpl()
if (is_finished)
throw Exception("WriteBufferFromNuraftBuffer is finished", ErrorCodes::CANNOT_WRITE_AFTER_END_OF_BUFFER);
size_t old_size = buffer->size();
/// pos may not be equal to vector.data() + old_size, because WriteBuffer::next() can be used to flush data
size_t pos_offset = pos - reinterpret_cast<Position>(buffer->data_begin());
LOG_DEBUG(&Poco::Logger::get("DEBUG"), "BUFFER SIZE {}", old_size * size_multiplier);
nuraft::ptr<nuraft::buffer> new_buffer = nuraft::buffer::alloc(old_size * size_multiplier);
memcpy(new_buffer->data_begin(), buffer->data_begin(), buffer->size());
buffer = new_buffer;
size_t old_size = buffer->size();
if (pos_offset == old_size)
{
nuraft::ptr<nuraft::buffer> new_buffer = nuraft::buffer::alloc(old_size * size_multiplier);
memcpy(new_buffer->data_begin(), buffer->data_begin(), buffer->size());
buffer = new_buffer;
}
internal_buffer = Buffer(reinterpret_cast<Position>(buffer->data_begin() + pos_offset), reinterpret_cast<Position>(buffer->data_begin() + buffer->size()));
working_buffer = internal_buffer;
}
WriteBufferFromNuraftBuffer::WriteBufferFromNuraftBuffer()
@ -38,7 +41,7 @@ void WriteBufferFromNuraftBuffer::finalize()
return;
is_finished = true;
size_t real_size = position() - reinterpret_cast<Position>(buffer->data_begin());
size_t real_size = pos - reinterpret_cast<Position>(buffer->data_begin());
nuraft::ptr<nuraft::buffer> new_buffer = nuraft::buffer::alloc(real_size);
memcpy(new_buffer->data_begin(), buffer->data_begin(), real_size);
buffer = new_buffer;