mirrors/dragonfly
1
0
Fork 0
mirror of https://github.com/dragonflydb/dragonfly.git synced 2025-05-11 10:25:47 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

fix(server): Fix connection bug (#695)

Browse source
This commit is contained in:
Vladislav 2023-01-16 19:53:48 +03:00 • committed by GitHub
parent 57e3ec9a81
commit daf5473ce1
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 11 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

18
src/facade/dragonfly_connection.cc
View file

@ -111,17 +111,18 @@ struct Connection::Request {
using MonitorMessage = std::string; using MonitorMessage = std::string;
struct PipelineMsg { struct PipelineMsg {
absl::InlinedVector<MutableSlice, 6> args;
// I do not use mi_heap_t explicitly but mi_stl_allocator at the end does the same job // I do not use mi_heap_t explicitly but mi_stl_allocator at the end does the same job
// of using the thread's heap. // of using the thread's heap.
// The capacity is chosen so that we allocate a fully utilized (256 bytes) block. // The capacity is chosen so that we allocate a fully utilized (256 bytes) block.
using StorageType = absl::InlinedVector<char, kReqStorageSize, mi_stl_allocator<char>>; using StorageType = absl::InlinedVector<char, kReqStorageSize, mi_stl_allocator<char>>;
StorageType storage;
PipelineMsg(size_t nargs, size_t capacity) : args(nargs), storage(capacity) { PipelineMsg(size_t nargs, size_t capacity) : args(nargs), storage(capacity) {
} }
void Reset(size_t nargs, size_t capacity);
absl::InlinedVector<MutableSlice, 6> args;
StorageType storage;
}; };
static constexpr size_t kSizeOfPipelineMsg = sizeof(PipelineMsg); static constexpr size_t kSizeOfPipelineMsg = sizeof(PipelineMsg);
@ -207,15 +208,18 @@ void Connection::RequestDeleter::operator()(Request* req) const {
void Connection::Request::Emplace(const RespVec& args, size_t capacity) { void Connection::Request::Emplace(const RespVec& args, size_t capacity) {
PipelineMsg* msg = get_if<PipelineMsg>(&payload); PipelineMsg* msg = get_if<PipelineMsg>(&payload);
if (msg) { if (msg) {
if (msg->storage.size() < capacity) { msg->Reset(args.size(), capacity);
msg->storage.resize(capacity);
}
} else { } else {
payload = PipelineMsg{args.size(), capacity}; payload = PipelineMsg{args.size(), capacity};
} }
SetArgs(args); SetArgs(args);
} }
void Connection::Request::PipelineMsg::Reset(size_t nargs, size_t capacity) {
storage.resize(capacity);
args.resize(nargs);
}
Connection::Connection(Protocol protocol, util::HttpListenerBase* http_listener, SSL_CTX* ctx, Connection::Connection(Protocol protocol, util::HttpListenerBase* http_listener, SSL_CTX* ctx,
ServiceInterface* service) ServiceInterface* service)
: io_buf_(kMinReadSize), http_listener_(http_listener), ctx_(ctx), service_(service) { : io_buf_(kMinReadSize), http_listener_(http_listener), ctx_(ctx), service_(service) {