Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

⚡ fix: improving in boost asio #11

Merged
merged 1 commit into from
Sep 12, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
85 changes: 47 additions & 38 deletions core/manager/manager.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -35,12 +35,22 @@ void Manager::run() noexcept {
void Manager::read() noexcept {
auto self(shared_from_this());

boost::asio::async_read_until(socket_, boost::asio::dynamic_buffer(data_), "\r\n",
[this, self](boost::system::error_code ec, std::size_t length) {
socket_.async_read_some(
boost::asio::buffer(buffer_),
[this, self](const boost::system::error_code& ec, std::size_t length) {
if (ec) return;
data_.erase(0, data_.find_first_not_of("\r\n"));
if (!data_.empty()) invokeAction();
data_.clear();
data_.append(buffer_.data(), length);

std::size_t start = 0;
std::size_t pos;

while ((pos = data_.find("\r\n", start)) != std::string::npos) {
std::string line = data_.substr(start, pos - start);
invokeAction(line);
start = pos + 2;
}

data_.erase(0, start);
read();
}
);
Expand All @@ -51,50 +61,51 @@ void Manager::result(std::string value) noexcept {
[this](boost::system::error_code ec, std::size_t){});
}

void Manager::invokeAction() noexcept {
std::istringstream request(data_);
void Manager::invokeAction(const std::string& line) noexcept {
std::vector<std::string> args;
std::string command;

request >> command;
boost::algorithm::to_upper(command);

if (std::find(commands.all.begin(), commands.all.end(), command) == commands.all.end()) return;
const char* ptr = line.c_str();
const char* end = ptr + line.size();

char quote = '\0';

std::string value;
while (ptr != end && !std::isspace(*ptr)) {
command.push_back(std::toupper(*ptr));
++ptr;
}

while (request) {
char peekStream = request.peek();
if (commands.all.find(command) == commands.all.end()) return result("ERROR: incorrect command");

if (peekStream == '"' || peekStream == '\'') {
quote = request.get();
value.clear();
char c;
bool escape = false;
while (ptr != end && std::isspace(*ptr)) ++ptr;

while (request.get(c)) {
if (escape) {
value.push_back(c);
escape = false;
}
else if (c == '\\') escape = true;
else if (c == quote) break;
else value.push_back(c);
}
} else if (std::isspace(peekStream)) {
request.ignore();
if (!value.empty()) {
std::string value;
char quote = '\0';
bool escape = false;

while (ptr != end) {
char c = *ptr++;

if (quote) {
if (escape) {
value.push_back(c);
escape = false;
} else if (c == '\\') {
escape = true;
} else if (c == quote) {
quote = '\0';
args.push_back(std::move(value));
value.clear();
} else {
value.push_back(c);
}
} else {
request >> value;
if (request) {
} else if (c == '"' || c == '\'') {
quote = c;
} else if (std::isspace(c)) {
if (!value.empty()) {
args.push_back(std::move(value));
value.clear();
}
} else {
value.push_back(c);
}
}

Expand All @@ -110,8 +121,6 @@ void Manager::invokeAction() noexcept {
if (commands.set == command) return invokeSet(args);
if (commands.del == command) return invokeDel(args);
if (commands.keys == command) return invokeKeys(args);

return result("ERROR: incorrect command");
}

void Manager::invokeDel(std::vector<std::string> args) noexcept {
Expand Down
6 changes: 4 additions & 2 deletions core/manager/manager.h
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
#include <functional>
#include <charconv>
#include <system_error>
#include <unordered_set>
#include <unordered_map>
#include <boost/asio.hpp>
#include <boost/algorithm/string.hpp>
Expand All @@ -44,7 +45,7 @@
std::string auth = "AUTH";
std::string use = "USE";
std::string keys = "KEYS";
std::vector<std::string> all = {"SET", "DEL", "GET", "AUTH", "USE", "KEYS"};
std::unordered_set<std::string> all = {"AUTH", "USE", "GET", "SET", "DEL", "KEYS"};
};


Expand All @@ -58,7 +59,7 @@
private:
void read() noexcept;
void result(std::string value) noexcept;
void invokeAction() noexcept;
void invokeAction(const std::string& line) noexcept;
void invokeDel(std::vector<std::string> args) noexcept;
void invokeSet(std::vector<std::string> args);
void invokeGet(std::vector<std::string> args) noexcept;
Expand All @@ -74,5 +75,6 @@
std::function<void()> onDisconnect_;
boost::asio::ip::tcp::socket socket_;
ConfigConnect& ConfigConn_;
std::array<char, 1024> buffer_;
};
#endif
12 changes: 9 additions & 3 deletions core/socket/socket.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -22,10 +22,16 @@
#include "socket.h"

CoreSocket::CoreSocket(boost::asio::io_context& io_context, std::string ip, short port, Cache& cache_)
: acceptor_(io_context, boost::asio::ip::tcp::endpoint(boost::asio::ip::make_address(ip), port)),
cache_(cache_), client_(0), socket_(io_context) {
acceptor_.set_option(boost::asio::socket_base::reuse_address(true));
: acceptor_(io_context), cache_(cache_), client_(0), socket_(io_context) {
boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::make_address_v4(ip), port);
acceptor_.open(endpoint.protocol());

acceptor_.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
acceptor_.set_option(boost::asio::ip::tcp::no_delay(true));

acceptor_.bind(endpoint);
acceptor_.listen(boost::asio::socket_base::max_listen_connections);

accept();
ping();
}
Expand Down
10 changes: 9 additions & 1 deletion main.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,15 @@ int main(){
boost::asio::io_context io_context;
ConfigHosting ConfigHost = Config::getConfigHost();
CoreSocket server(io_context, ConfigHost.address, ConfigHost.port, cache_);
io_context.run();

std::vector<std::thread> threads;
for (std::size_t i = 0; i < std::thread::hardware_concurrency(); ++i) {
threads.emplace_back([&io_context]() { io_context.run(); });
}

for (auto& thread : threads) {
thread.join();
}
}
catch (std::exception& e) {
std::cerr << "Exception: " << e.what() << "\n";
Expand Down
Loading