ae/CPP_Core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修完bug

Browse Source
This commit is contained in:
种太阳的热带美短
2026-08-10 16:36:12 +08:00
parent 14675e7822
commit a742862f66
12 changed files with 522 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Psc_Cpp_Core/Base/RingBuffer.hpp
+12 -2
View File
@@ -6,6 +6,12 @@
#include <string>
#include <vector>
namespace Psc {
template <typename Size_T>
struct Ring_Buffer_State {
Size_T capacity{};
Size_T used{};
Size_T free{};
};
// -------------------- utils --------------------
template <typename Size_T>
bool is_pow2(Size_T x) {
@@ -53,10 +59,14 @@ public:
[[nodiscard]] bool full() const {
return ((tail + 1) & mask_) == head;
}
[[nodiscard]] Ring_Buffer_State<Size_T> state() {
std::lock_guard<MutexT> lock(mutex_);
return {capacity, used_space_unsafe(), free_space_unsafe()};
}
[[nodiscard]] std::string state_str() {
std::lock_guard<MutexT> lock(mutex_);
auto free_space = free_space_unsafe();
auto used_space = used_space_unsafe();
const auto free_space = free_space_unsafe();
const auto used_space = used_space_unsafe();
return VAR_STR_5(free_space, used_space, head, tail, capacity);
}
protected:
src/Psc_Cpp_Core/Serial/Serial_Coro.h
+23 -1
View File
@@ -8,37 +8,47 @@
#include <memory>
#include <string>
#include <string_view>
#include "Psc_Cpp_Core/Statistics/Statistics.h"
#include "Serial.h"
namespace Psc::serial {
class Serial_Coro : public Serial {
public:
Psc::Transport_Statistics statistics;
bool open() {
if (port_ && port_->is_open()) {
return true;
}
statistics.record_start_attempt();
port_ = std::make_unique<asio::serial_port>(io_context_);
asio::error_code ec;
port_->open(normalize_port_name(serial_name), ec);
if (ec) {
statistics.record_error(error_text(ec));
return false;
}
apply_options(ec);
if (ec) {
statistics.record_error(error_text(ec));
close();
return false;
}
serial::set_block(port_->native_handle(), false);
statistics.record_started();
statistics.set_receive_buffer(static_cast<std::size_t>(std::max(buffer_byte_size, 0)), 0);
return true;
}
void close() {
if (!port_) {
return;
}
const auto was_open = port_->is_open();
asio::error_code ec;
port_->cancel(ec);
port_->close(ec);
port_.reset();
io_context_.restart();
if (was_open)
statistics.record_stopped();
}
[[nodiscard]] asio::awaitable<void> tick_coro() {
tick();
@@ -75,13 +85,17 @@ private:
return "";
}
auto available = serial::get_available_bytes(port_->native_handle());
statistics.set_receive_buffer(static_cast<std::size_t>(std::max(buffer_byte_size, 0)), static_cast<std::size_t>(std::max(available, 0)));
if (available <= 0) {
return "";
}
auto data = serial::read_all(port_->native_handle());
if (data.size() > max_size) {
statistics.record_dropped(data.size() - max_size);
data.resize(max_size);
}
if (!data.empty())
statistics.record_received(data.size());
return data;
}
std::size_t write_impl(std::string_view data) {
@@ -91,7 +105,12 @@ private:
}
asio::error_code ec;
auto size = asio::write(*port_, asio::buffer(data), ec);
return ec ? 0 : size;
if (ec) {
statistics.record_error(error_text(ec));
return 0;
}
statistics.record_sent(size);
return size;
}
static std::string normalize_port_name(std::string_view port_name) {
std::string ret(port_name);
@@ -102,6 +121,9 @@ private:
#endif
return ret;
}
static std::string error_text(const asio::error_code& ec) {
return std::string(ec.category().name()) + ":" + std::to_string(ec.value()) + " " + Psc::platform_2_utf8(ec.message());
}
void apply_options(asio::error_code& ec) {
if (baud_rate != static_cast<Baud_Rate_Type>(-1)) {
port_->set_option(asio::serial_port_base::baud_rate(
src/Psc_Cpp_Core/Statistics/Statistics.h
+191
View File
@@ -1,7 +1,12 @@
#pragma once
#include <algorithm>
#include <array>
#include <atomic>
#include <cmath>
#include <cstdint>
#include <mutex>
#include <string>
#include <utility>
#include "../Base/global_include.h"
#include "global.h"
namespace Psc {
@@ -230,4 +235,190 @@ public:
total = 0;
}
};
class Throughput_Statistics : public Base_Statistics {
public:
Throughput_Statistics() : last_sample_ms_(get_current_ms()) {
}
void clear() {
std::lock_guard lock(mtx_);
bytes_total_ = 0;
events_total_ = 0;
window_bytes_ = 0;
window_events_ = 0;
bytes_per_second_.clear();
events_per_second_.clear();
bytes_per_event_.clear();
last_sample_ms_ = get_current_ms();
start_time = last_sample_ms_;
}
void update(std::size_t bytes) {
std::lock_guard lock(mtx_);
bytes_total_ += bytes;
++events_total_;
window_bytes_ += bytes;
++window_events_;
bytes_per_event_.update(static_cast<double>(bytes));
sample_locked(get_current_ms());
}
[[nodiscard]] JSON to_json() const override {
std::lock_guard lock(mtx_);
sample_locked(get_current_ms());
JSON ret = JSON::object();
ret.append({"bytes_total", bytes_total_});
ret.append({"events_total", events_total_});
ret.append({"bytes_per_second", bytes_per_second_.to_json()});
ret.append({"events_per_second", events_per_second_.to_json()});
ret.append({"bytes_per_event", bytes_per_event_.to_json()});
return ret;
}
private:
void sample_locked(std::uint64_t now) const {
constexpr std::uint64_t sample_period_ms = 250;
const auto elapsed_ms = now - last_sample_ms_;
if (elapsed_ms < sample_period_ms)
return;
const auto elapsed_seconds = static_cast<double>(elapsed_ms) / 1000.0;
bytes_per_second_.update(static_cast<double>(window_bytes_) / elapsed_seconds);
events_per_second_.update(static_cast<double>(window_events_) / elapsed_seconds);
window_bytes_ = 0;
window_events_ = 0;
last_sample_ms_ = now;
}
mutable std::mutex mtx_;
mutable std::uint64_t bytes_total_{};
mutable std::uint64_t events_total_{};
mutable std::uint64_t window_bytes_{};
mutable std::uint64_t window_events_{};
mutable std::uint64_t last_sample_ms_{};
mutable Value_Statistics bytes_per_second_;
mutable Value_Statistics events_per_second_;
mutable Value_Statistics bytes_per_event_;
};
class Transport_Statistics : public Base_Statistics {
public:
void record_received(std::size_t bytes) {
received_.update(bytes);
last_activity_unix_ms_.store(get_unix_ms(), std::memory_order_relaxed);
}
void record_sent(std::size_t bytes) {
sent_.update(bytes);
last_activity_unix_ms_.store(get_unix_ms(), std::memory_order_relaxed);
}
void record_dropped(std::size_t bytes) {
dropped_.update(bytes);
last_activity_unix_ms_.store(get_unix_ms(), std::memory_order_relaxed);
}
void record_start_attempt() {
std::lock_guard lock(mtx_);
++start_attempt_total_;
}
void record_started() {
std::lock_guard lock(mtx_);
++start_success_total_;
++active_sessions_;
max_active_sessions_ = std::max(max_active_sessions_, active_sessions_);
last_started_unix_ms_ = get_unix_ms();
if (active_sessions_ == 1)
active_since_ms_ = get_current_ms();
}
void record_stopped() {
std::lock_guard lock(mtx_);
if (active_sessions_ == 0)
return;
--active_sessions_;
++stop_total_;
last_stopped_unix_ms_ = get_unix_ms();
if (active_sessions_ == 0)
active_since_ms_ = 0;
}
void record_error(std::string error) {
std::lock_guard lock(mtx_);
++error_total_;
last_error_unix_ms_ = get_unix_ms();
last_error_ = std::move(error);
}
void set_send_buffer(std::size_t capacity, std::size_t used) {
std::lock_guard lock(mtx_);
send_buffer_capacity_ = capacity;
send_buffer_used_ = used;
send_buffer_high_water_ = std::max(send_buffer_high_water_, used);
}
void set_receive_buffer(std::size_t capacity, std::size_t used) {
std::lock_guard lock(mtx_);
receive_buffer_capacity_ = capacity;
receive_buffer_used_ = used;
receive_buffer_high_water_ = std::max(receive_buffer_high_water_, used);
}
[[nodiscard]] std::uint64_t last_activity_unix_ms() const {
return last_activity_unix_ms_.load(std::memory_order_relaxed);
}
[[nodiscard]] std::uint64_t active_sessions() const {
std::lock_guard lock(mtx_);
return active_sessions_;
}
[[nodiscard]] JSON to_json() const override {
JSON traffic = JSON::object();
traffic.append({"received", received_.to_json()});
traffic.append({"sent", sent_.to_json()});
traffic.append({"dropped", dropped_.to_json()});
std::lock_guard lock(mtx_);
JSON lifecycle = JSON::object();
lifecycle.append({"start_attempt_total", start_attempt_total_});
lifecycle.append({"start_success_total", start_success_total_});
lifecycle.append({"stop_total", stop_total_});
lifecycle.append({"error_total", error_total_});
lifecycle.append({"active_sessions", active_sessions_});
lifecycle.append({"max_active_sessions", max_active_sessions_});
lifecycle.append({"active_duration_ms", active_since_ms_ == 0 ? 0 : get_current_ms() - active_since_ms_});
lifecycle.append({"last_started_unix_ms", last_started_unix_ms_});
lifecycle.append({"last_stopped_unix_ms", last_stopped_unix_ms_});
lifecycle.append({"last_error_unix_ms", last_error_unix_ms_});
lifecycle.append({"last_activity_unix_ms", last_activity_unix_ms_.load(std::memory_order_relaxed)});
lifecycle.append({"last_error", last_error_});
JSON buffers = JSON::object();
buffers.append({"send", buffer_json(send_buffer_capacity_, send_buffer_used_, send_buffer_high_water_)});
buffers.append({"receive", buffer_json(receive_buffer_capacity_, receive_buffer_used_, receive_buffer_high_water_)});
JSON ret = JSON::object();
ret.append({"traffic", traffic});
ret.append({"lifecycle", lifecycle});
ret.append({"buffers", buffers});
return ret;
}
private:
[[nodiscard]] static std::uint64_t get_unix_ms() {
using namespace std::chrono;
return static_cast<std::uint64_t>(duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count());
}
[[nodiscard]] static JSON buffer_json(std::size_t capacity, std::size_t used, std::size_t high_water) {
JSON ret = JSON::object();
ret.append({"capacity_bytes", capacity});
ret.append({"used_bytes", used});
ret.append({"free_bytes", capacity > used ? capacity - used : 0});
ret.append({"utilization_percent", capacity == 0 ? 0.0 : static_cast<double>(used) * 100.0 / static_cast<double>(capacity)});
ret.append({"high_water_bytes", high_water});
return ret;
}
mutable std::mutex mtx_;
Throughput_Statistics received_;
Throughput_Statistics sent_;
Throughput_Statistics dropped_;
std::atomic<std::uint64_t> last_activity_unix_ms_{};
std::uint64_t start_attempt_total_{};
std::uint64_t start_success_total_{};
std::uint64_t stop_total_{};
std::uint64_t error_total_{};
std::uint64_t active_sessions_{};
std::uint64_t max_active_sessions_{};
std::uint64_t active_since_ms_{};
std::uint64_t last_started_unix_ms_{};
std::uint64_t last_stopped_unix_ms_{};
std::uint64_t last_error_unix_ms_{};
std::size_t send_buffer_capacity_{};
std::size_t send_buffer_used_{};
std::size_t send_buffer_high_water_{};
std::size_t receive_buffer_capacity_{};
std::size_t receive_buffer_used_{};
std::size_t receive_buffer_high_water_{};
std::string last_error_;
};
} // namespace Psc
src/Psc_Cpp_Core/socket/ASIO_Utils.h
+6 -2
View File
@@ -1,14 +1,15 @@
#pragma once
#include <string>
#include <system_error>
#include "Psc_Cpp_Core/Base/RingBuffer.hpp"
#include "Socket.h"
#include "global.h"
namespace Psc::asio_socket {
#ifndef PSC_ASIO_STREAM_RING_BUFFER
#define PSC_ASIO_STREAM_RING_BUFFER StreamRingBuffer_ST
#define PSC_ASIO_STREAM_RING_BUFFER StreamRingBuffer_MT
#endif
#ifndef PSC_ASIO_PACKET_RING_BUFFER
#define PSC_ASIO_PACKET_RING_BUFFER RingBuffer_ST
#define PSC_ASIO_PACKET_RING_BUFFER RingBuffer_MT
#endif
using ASIO_StreamRingBuffer = PSC_ASIO_STREAM_RING_BUFFER;
using ASIO_PacketRingBuffer = PSC_ASIO_PACKET_RING_BUFFER;
@@ -26,4 +27,7 @@ endpoint_to_sockaddr(const asio::ip::udp::endpoint& endpoint) {
inline bool would_block(const asio::error_code& ec) {
return ec == asio::error::would_block || ec == asio::error::try_again;
}
inline std::string error_text(const asio::error_code& ec) {
return std::string(ec.category().name()) + ":" + std::to_string(ec.value()) + " " + Psc::platform_2_utf8(ec.message());
}
} // namespace Psc::asio_socket
src/Psc_Cpp_Core/socket/TCP_Client.cpp
+33 -3
View File
@@ -2,6 +2,14 @@
#include <array>
#include <string_view>
namespace Psc::asio_socket {
namespace {
void update_buffer_statistics(TCP_Client& client, ASIO_StreamRingBuffer& send_buffer, ASIO_StreamRingBuffer& receive_buffer) {
const auto send = send_buffer.state();
const auto receive = receive_buffer.state();
client.statistics.set_send_buffer(send.capacity, send.used);
client.statistics.set_receive_buffer(receive.capacity, receive.used);
}
}
TCP_Client::~TCP_Client() {
close();
}
@@ -20,9 +28,11 @@ void TCP_Client::create() {
resolver = std::make_unique<asio::ip::tcp::resolver>(io_context);
socket = std::make_unique<asio::ip::tcp::socket>(io_context);
socket_fd = socket_id(socket.get());
update_buffer_statistics(*this, send_buffer, recv_buffer);
set_state(Reconnecting);
}
void TCP_Client::close() {
const bool was_connected = state == Connected;
connect_pending = false;
read_pending = false;
write_pending = false;
@@ -37,9 +47,12 @@ void TCP_Client::close() {
socket->close(ec);
}
socket_fd = static_cast<Socket_FD>(-1);
if (was_connected)
statistics.record_stopped();
set_state(Not_Created);
}
void TCP_Client::schedule_recreate() {
const bool was_connected = state == Connected;
connect_pending = false;
read_pending = false;
write_pending = false;
@@ -52,6 +65,8 @@ void TCP_Client::schedule_recreate() {
}
next_reconnect_tp = std::chrono::steady_clock::now() +
std::chrono::milliseconds(reconnect_backoff_ms);
if (was_connected)
statistics.record_stopped();
set_state(Reconnecting);
}
bool TCP_Client::start_connect() {
@@ -65,6 +80,7 @@ bool TCP_Client::start_connect() {
resolver = std::make_unique<asio::ip::tcp::resolver>(io_context);
socket = std::make_unique<asio::ip::tcp::socket>(io_context);
connect_pending = true;
statistics.record_start_attempt();
set_state(Connecting);
resolver->async_resolve(
dest_address.ip, std::to_string(dest_address.port),
@@ -73,6 +89,7 @@ bool TCP_Client::start_connect() {
if (state != Connecting || !socket)
return;
if (ec) {
statistics.record_error(error_text(ec));
schedule_recreate();
return;
}
@@ -83,11 +100,13 @@ bool TCP_Client::start_connect() {
if (state != Connecting || !socket)
return;
if (connect_ec) {
statistics.record_error(error_text(connect_ec));
schedule_recreate();
return;
}
socket_fd = socket_id(socket.get());
address = dest_address;
statistics.record_started();
set_state(Connected);
start_read();
start_write();
@@ -124,11 +143,15 @@ void TCP_Client::start_read() {
return;
if (ec == asio::error::eof ||
ec == asio::error::connection_reset || ec) {
statistics.record_error(error_text(ec));
schedule_recreate();
return;
}
recv_buffer.write_best_effort(recv_storage.data(),
n);
const auto written = recv_buffer.write_best_effort(recv_storage.data(), n);
statistics.record_received(n);
if (written < n)
statistics.record_dropped(n - written);
update_buffer_statistics(*this, send_buffer, recv_buffer);
start_read();
});
}
@@ -151,10 +174,13 @@ void TCP_Client::start_write() {
if (ec == asio::error::operation_aborted)
return;
if (ec) {
statistics.record_error(error_text(ec));
schedule_recreate();
return;
}
send_buffer.skip(sent);
statistics.record_sent(sent);
update_buffer_statistics(*this, send_buffer, recv_buffer);
if (sent != 0)
start_write();
});
@@ -168,12 +194,16 @@ std::string TCP_Client::read() {
break;
ret.append(buffer.data(), n);
}
update_buffer_statistics(*this, send_buffer, recv_buffer);
return ret;
}
void TCP_Client::send(std::string_view data) {
if (data.empty())
return;
send_buffer.write_best_effort(data.data(), data.size());
const auto written = send_buffer.write_best_effort(data.data(), data.size());
if (written < data.size())
statistics.record_dropped(data.size() - written);
update_buffer_statistics(*this, send_buffer, recv_buffer);
if (state == Connected)
start_write();
}
src/Psc_Cpp_Core/socket/TCP_Client.h
+2
View File
@@ -7,6 +7,7 @@
#include <string_view>
#include <vector>
#include "ASIO_Utils.h"
#include "Psc_Cpp_Core/Statistics/Statistics.h"
namespace Psc::asio_socket {
class TCP_Client : public Socket_Base {
public:
@@ -32,6 +33,7 @@ public:
};
void close() override;
std::atomic<State> state{Not_Created};
Transport_Statistics statistics;
void set_state(State s);
std::function<void(State old_state, State new_state)> on_state_change =
nullptr;
src/Psc_Cpp_Core/socket/TCP_Server.cpp
+91 -18
View File
@@ -2,6 +2,14 @@
#include <array>
#include <string_view>
namespace Psc::asio_socket {
namespace {
void update_buffer_statistics(const std::shared_ptr<TCP_Connect>& conn) {
const auto send = conn->send_buffer.state();
const auto receive = conn->recv_buffer.state();
conn->statistics.set_send_buffer(send.capacity, send.used);
conn->statistics.set_receive_buffer(receive.capacity, receive.used);
}
}
TCP_Server::~TCP_Server() {
close();
}
@@ -47,28 +55,34 @@ bool TCP_Server::listen(const Sockaddr_In& addr) {
create();
if (addr.ip.empty() || addr.port == 0) {
state = Not_Set_Field;
statistics.record_error("listen address is empty");
return false;
}
asio::error_code ec;
const auto address_value = asio::ip::make_address(addr.ip, ec);
if (ec) {
state = Not_bind;
statistics.record_error(error_text(ec));
return false;
}
asio::ip::tcp::endpoint endpoint(address_value,
static_cast<unsigned short>(addr.port));
acceptor->open(endpoint.protocol(), ec);
if (ec)
if (ec) {
statistics.record_error(error_text(ec));
return false;
}
acceptor->set_option(asio::socket_base::reuse_address(true), ec);
acceptor->bind(endpoint, ec);
if (ec) {
state = Not_bind;
statistics.record_error(error_text(ec));
return false;
}
acceptor->listen(asio::socket_base::max_listen_connections, ec);
if (ec) {
state = Not_Listen;
statistics.record_error(error_text(ec));
return false;
}
address = addr;
@@ -84,11 +98,14 @@ void TCP_Server::close() {
pending_accept_socket->close(ec);
}
pending_accept_socket.reset();
for (auto& [_, conn] : tcp_clients) {
for (const auto& conn : get_all_clients()) {
close_client(conn);
}
tcp_clients.clear();
accepted_clients.clear();
{
std::lock_guard lock(clients_mutex);
tcp_clients.clear();
accepted_clients.clear();
}
if (acceptor) {
asio::error_code ec;
acceptor->cancel(ec);
@@ -98,6 +115,7 @@ void TCP_Server::close() {
state = Not_Created;
}
std::shared_ptr<TCP_Connect> TCP_Server::accept() const {
std::lock_guard lock(clients_mutex);
while (!accepted_clients.empty()) {
auto conn = accepted_clients.front().lock();
accepted_clients.erase(accepted_clients.begin());
@@ -111,6 +129,7 @@ void TCP_Server::start_accept() {
return;
pending_accept_socket = std::make_shared<asio::ip::tcp::socket>(io_context);
accept_pending = true;
statistics.record_start_attempt();
acceptor->async_accept(
*pending_accept_socket, [this](const asio::error_code& ec) {
accept_pending = false;
@@ -137,11 +156,20 @@ void TCP_Server::start_accept() {
conn->info.fd = socket_id(conn->socket.get());
conn->info.sockaddr =
endpoint_to_sockaddr(conn->socket->remote_endpoint(option_ec));
tcp_clients[conn->info.fd] = conn;
accepted_clients.push_back(conn);
update_buffer_statistics(conn);
conn->statistics.record_started();
statistics.record_started();
{
std::lock_guard lock(clients_mutex);
tcp_clients[conn->info.fd] = conn;
accepted_clients.push_back(conn);
}
start_read(conn);
start_write(conn);
}
else if (ec) {
statistics.record_error(error_text(ec));
}
pending_accept_socket.reset();
start_accept();
});
@@ -159,12 +187,24 @@ void TCP_Server::start_read(const std::shared_ptr<TCP_Connect>& conn) {
return;
if (ec == asio::error::operation_aborted)
return;
if (ec == asio::error::eof || ec == asio::error::connection_reset ||
ec) {
if (ec == asio::error::eof) {
close_client(conn);
return;
}
conn->recv_buffer.write_best_effort(conn->recv_storage.data(), n);
if (ec) {
conn->statistics.record_error(error_text(ec));
statistics.record_error(error_text(ec));
close_client(conn);
return;
}
const auto written = conn->recv_buffer.write_best_effort(conn->recv_storage.data(), n);
conn->statistics.record_received(n);
statistics.record_received(n);
if (written < n) {
conn->statistics.record_dropped(n - written);
statistics.record_dropped(n - written);
}
update_buffer_statistics(conn);
start_read(conn);
});
}
@@ -188,20 +228,24 @@ void TCP_Server::start_write(const std::shared_ptr<TCP_Connect>& conn) {
if (ec == asio::error::operation_aborted)
return;
if (ec) {
conn->statistics.record_error(error_text(ec));
statistics.record_error(error_text(ec));
close_client(conn);
return;
}
conn->send_buffer.skip(sent);
conn->push_speed.update(static_cast<unsigned int>(sent));
conn->send_num.update(static_cast<double>(sent));
conn->statistics.record_sent(sent);
statistics.record_sent(sent);
update_buffer_statistics(conn);
if (sent != 0)
start_write(conn);
});
}
void TCP_Server::close_client(const std::shared_ptr<TCP_Connect>& conn) {
if (!conn || conn->closing)
if (!conn || conn->closing.exchange(true))
return;
conn->closing = true;
conn->statistics.record_stopped();
statistics.record_stopped();
if (conn->socket) {
asio::error_code ec;
conn->socket->cancel(ec);
@@ -210,6 +254,7 @@ void TCP_Server::close_client(const std::shared_ptr<TCP_Connect>& conn) {
}
}
void TCP_Server::cleanup_closed_clients() {
std::lock_guard lock(clients_mutex);
for (auto it = tcp_clients.begin(); it != tcp_clients.end();) {
if (!it->second || it->second->closing || !it->second->socket ||
!it->second->socket->is_open()) {
@@ -231,7 +276,7 @@ void TCP_Server::cleanup_closed_clients() {
}
void TCP_Server::flush_clients() {
start_accept();
for (auto& [_, conn] : tcp_clients) {
for (const auto& conn : get_all_clients()) {
start_read(conn);
start_write(conn);
}
@@ -239,20 +284,23 @@ void TCP_Server::flush_clients() {
void TCP_Server::write_to_all_clients(std::string_view data) {
if (data.empty())
return;
for (auto& [_, conn] : tcp_clients) {
for (const auto& conn : get_all_clients()) {
if (!conn || conn->closing)
continue;
auto written =
conn->send_buffer.write_best_effort(data.data(), data.size());
if (written < data.size())
conn->lose_speed.update(static_cast<unsigned int>(data.size() - written));
if (written < data.size()) {
conn->statistics.record_dropped(data.size() - written);
statistics.record_dropped(data.size() - written);
}
update_buffer_statistics(conn);
start_write(conn);
}
}
std::vector<TCP_Server::Read_Info> TCP_Server::read_from_all_clients() {
std::vector<Read_Info> ret;
std::array<char, 16 * 1024> buffer{};
for (auto& [_, conn] : tcp_clients) {
for (const auto& conn : get_all_clients()) {
if (!conn || conn->closing)
continue;
std::string data;
@@ -264,12 +312,14 @@ std::vector<TCP_Server::Read_Info> TCP_Server::read_from_all_clients() {
}
if (!data.empty())
ret.push_back({conn->info, std::move(data)});
update_buffer_statistics(conn);
}
cleanup_closed_clients();
return ret;
}
std::vector<Socket_FD> TCP_Server::client_fds() {
std::vector<Socket_FD> ret;
std::lock_guard lock(clients_mutex);
for (const auto& [fd, conn] : tcp_clients) {
if (conn && !conn->closing)
ret.push_back(fd);
@@ -278,12 +328,35 @@ std::vector<Socket_FD> TCP_Server::client_fds() {
}
std::vector<std::shared_ptr<TCP_Connect>> TCP_Server::get_all_clients() {
std::vector<std::shared_ptr<TCP_Connect>> ret;
std::lock_guard lock(clients_mutex);
for (const auto& [_, conn] : tcp_clients) {
if (conn && !conn->closing)
ret.push_back(conn);
}
return ret;
}
std::shared_ptr<TCP_Connect> TCP_Server::get_client(Socket_FD fd) {
std::lock_guard lock(clients_mutex);
const auto iterator = tcp_clients.find(fd);
return iterator == tcp_clients.end() || !iterator->second || iterator->second->closing ? nullptr : iterator->second;
}
Psc::JSON TCP_Server::get_statistics_json() {
std::size_t send_capacity = 0;
std::size_t send_used = 0;
std::size_t receive_capacity = 0;
std::size_t receive_used = 0;
for (const auto& conn : get_all_clients()) {
const auto send = conn->send_buffer.state();
const auto receive = conn->recv_buffer.state();
send_capacity += send.capacity;
send_used += send.used;
receive_capacity += receive.capacity;
receive_used += receive.used;
}
statistics.set_send_buffer(send_capacity, send_used);
statistics.set_receive_buffer(receive_capacity, receive_used);
return statistics.to_json();
}
std::string TCP_Server::to_string() {
return "TCP_Server:[" + (address ? address->to_string() : std::string{}) +
"]";
src/Psc_Cpp_Core/socket/TCP_Server.h
+11 -7
View File
@@ -1,7 +1,9 @@
#pragma once
#include <array>
#include <atomic>
#include <map>
#include <memory>
#include <mutex>
#include <string_view>
#include <vector>
#include "ASIO_Utils.h"
@@ -13,14 +15,12 @@ public:
std::shared_ptr<asio::ip::tcp::socket> socket;
ASIO_StreamRingBuffer send_buffer;
ASIO_StreamRingBuffer recv_buffer;
Speed_Statistics push_speed;
Speed_Statistics lose_speed;
Value_Statistics send_num;
Transport_Statistics statistics;
std::array<char, 16 * 1024> recv_storage{};
std::vector<char> send_storage;
bool read_pending = false;
bool write_pending = false;
bool closing = false;
std::atomic_bool read_pending = false;
std::atomic_bool write_pending = false;
std::atomic_bool closing = false;
[[nodiscard]] std::string to_string() const {
return info.to_string();
}
@@ -50,10 +50,13 @@ public:
Not_Listen,
Working
};
State state = Not_Created;
std::atomic<State> state{Not_Created};
Transport_Statistics statistics;
std::vector<Read_Info> read_from_all_clients();
std::vector<Socket_FD> client_fds();
std::vector<std::shared_ptr<TCP_Connect>> get_all_clients();
std::shared_ptr<TCP_Connect> get_client(Socket_FD fd);
Psc::JSON get_statistics_json();
void flush_clients();
[[nodiscard]] std::shared_ptr<TCP_Connect> accept() const;
protected:
@@ -68,6 +71,7 @@ protected:
bool accept_pending = false;
std::shared_ptr<asio::ip::tcp::socket> pending_accept_socket;
mutable std::vector<std::weak_ptr<TCP_Connect>> accepted_clients;
mutable std::mutex clients_mutex;
void start_accept();
void start_read(const std::shared_ptr<TCP_Connect>& conn);
void start_write(const std::shared_ptr<TCP_Connect>& conn);
src/Psc_Cpp_Core/socket/UDP_Client.cpp
+32 -3
View File
@@ -1,6 +1,14 @@
#include "UDP_Client.h"
#include <string_view>
namespace Psc::asio_socket {
namespace {
void update_buffer_statistics(Transport_Statistics& statistics, ASIO_PacketRingBuffer& send_buffer, ASIO_PacketRingBuffer& receive_buffer) {
const auto send = send_buffer.state();
const auto receive = receive_buffer.state();
statistics.set_send_buffer(send.capacity, send.used);
statistics.set_receive_buffer(receive.capacity, receive.used);
}
}
void UDP_Client::create() {
socket = std::make_unique<asio::ip::udp::socket>(io_context);
asio::error_code ec;
@@ -13,6 +21,7 @@ void UDP_Client::create() {
send_storage.assign(static_cast<size_t>(read_chunk_size), 0);
send_buffer.init(user_buffer_size);
recv_buffer.init(user_buffer_size);
update_buffer_statistics(statistics, send_buffer, recv_buffer);
state = Not_Set_Field;
}
bool UDP_Client::connect() {
@@ -22,14 +31,17 @@ bool UDP_Client::connect() {
create();
if (dest_address.ip.empty() || dest_address.port == 0) {
state = Not_Set_Field;
statistics.record_error("destination address is empty");
return false;
}
statistics.record_start_attempt();
asio::error_code ec;
endpoint =
asio::ip::udp::endpoint(asio::ip::make_address(dest_address.ip, ec),
static_cast<unsigned short>(dest_address.port));
if (ec) {
state = Not_Set_Field;
statistics.record_error(error_text(ec));
return false;
}
io_context.restart();
@@ -42,10 +54,12 @@ bool UDP_Client::connect() {
if (connect_ec) {
connected = false;
state = Not_Set_Field;
statistics.record_error(error_text(connect_ec));
return;
}
connected = true;
state = Working;
statistics.record_started();
start_read();
start_write();
});
@@ -57,6 +71,7 @@ std::string UDP_Client::read() {
if (!recv_buffer.read(ret.data(), out_len))
return {};
ret.resize(out_len);
update_buffer_statistics(statistics, send_buffer, recv_buffer);
return ret;
}
void UDP_Client::tick() {
@@ -78,7 +93,9 @@ void UDP_Client::tick() {
void UDP_Client::send(std::string_view data) {
if (data.empty())
return;
send_buffer.write(data.data(), data.size());
if (!send_buffer.write(data.data(), data.size()))
statistics.record_dropped(data.size());
update_buffer_statistics(statistics, send_buffer, recv_buffer);
if (!connected)
connect();
if (connected)
@@ -100,9 +117,14 @@ void UDP_Client::start_read() {
if (ec) {
connected = false;
state = Not_Set_Field;
statistics.record_error(error_text(ec));
statistics.record_stopped();
return;
}
recv_buffer.write(recv_storage.data(), n);
statistics.record_received(n);
if (!recv_buffer.write(recv_storage.data(), n))
statistics.record_dropped(n);
update_buffer_statistics(statistics, send_buffer, recv_buffer);
start_read();
});
}
@@ -117,7 +139,7 @@ void UDP_Client::start_write() {
send_storage.resize(out_len);
write_pending = true;
socket->async_send(asio::buffer(send_storage.data(), send_storage.size()),
[this](const asio::error_code& ec, std::size_t) {
[this](const asio::error_code& ec, std::size_t sent) {
write_pending = false;
if (!connected)
return;
@@ -126,15 +148,20 @@ void UDP_Client::start_write() {
if (ec) {
connected = false;
state = Not_Set_Field;
statistics.record_error(error_text(ec));
statistics.record_stopped();
return;
}
send_buffer.skip_one();
statistics.record_sent(sent);
update_buffer_statistics(statistics, send_buffer, recv_buffer);
send_storage.assign(static_cast<size_t>(read_chunk_size),
0);
start_write();
});
}
void UDP_Client::close() {
const bool was_connected = connected;
connected = false;
connect_pending = false;
read_pending = false;
@@ -145,6 +172,8 @@ void UDP_Client::close() {
socket->close(ec);
}
state = Not_Created;
if (was_connected)
statistics.record_stopped();
}
std::string UDP_Client::to_string() {
return "UDP_Client:[" + dest_address.to_string() + "]";
src/Psc_Cpp_Core/socket/UDP_Client.h
+4 -1
View File
@@ -1,9 +1,11 @@
#pragma once
#include <array>
#include <atomic>
#include <memory>
#include <string_view>
#include <vector>
#include "ASIO_Utils.h"
#include "Psc_Cpp_Core/Statistics/Statistics.h"
namespace Psc::asio_socket {
class UDP_Client {
public:
@@ -27,7 +29,8 @@ public:
Connecting,
Working,
};
State state = Not_Created;
std::atomic<State> state{Not_Created};
Transport_Statistics statistics;
private:
asio::io_context io_context;
std::unique_ptr<asio::ip::udp::socket> socket;
src/Psc_Cpp_Core/socket/UDP_Server.cpp
+102 -10
View File
@@ -2,6 +2,14 @@
#include <algorithm>
#include <string_view>
namespace Psc::asio_socket {
namespace {
void update_buffer_statistics(Transport_Statistics& statistics, ASIO_PacketRingBuffer& send_buffer, ASIO_PacketRingBuffer& receive_buffer) {
const auto send = send_buffer.state();
const auto receive = receive_buffer.state();
statistics.set_send_buffer(send.capacity, send.used);
statistics.set_receive_buffer(receive.capacity, receive.used);
}
}
void UDP_Server::create() {
socket = std::make_unique<asio::ip::udp::socket>(io_context);
socket_fd = socket_id(socket.get());
@@ -11,6 +19,7 @@ void UDP_Server::create() {
send_storage.assign(static_cast<size_t>(read_chunk_size), 0);
read_pending = false;
write_pending = false;
update_buffer_statistics(statistics, send_buffer, recv_buffer);
state = Not_Bind;
}
bool UDP_Server::bind() {
@@ -18,19 +27,24 @@ bool UDP_Server::bind() {
create();
if (bind_address.ip.empty() || bind_address.port == 0) {
state = Not_Set_Field;
statistics.record_error("bind address is empty");
return false;
}
statistics.record_start_attempt();
asio::error_code ec;
auto address_value = asio::ip::make_address(bind_address.ip, ec);
if (ec) {
state = Not_Bind;
statistics.record_error(error_text(ec));
return false;
}
asio::ip::udp::endpoint endpoint(
address_value, static_cast<unsigned short>(bind_address.port));
socket->open(endpoint.protocol(), ec);
if (ec)
if (ec) {
statistics.record_error(error_text(ec));
return false;
}
socket->set_option(asio::socket_base::reuse_address(true), ec);
socket->set_option(
asio::socket_base::receive_buffer_size(static_cast<int>(buffer_size)),
@@ -38,10 +52,12 @@ bool UDP_Server::bind() {
socket->bind(endpoint, ec);
if (ec) {
state = Not_Bind;
statistics.record_error(error_text(ec));
return false;
}
bound = true;
state = Working;
statistics.record_started();
start_read();
return true;
}
@@ -69,17 +85,35 @@ void UDP_Server::start_read() {
return;
if (ec == asio::error::operation_aborted)
return;
if (ec)
if (ec) {
statistics.record_error(error_text(ec));
return;
}
last_peer = endpoint_to_sockaddr(recv_endpoint);
has_last_peer = true;
if (std::find(clients.begin(), clients.end(), last_peer) ==
clients.end()) {
clients.push_back(last_peer);
std::shared_ptr<Transport_Statistics> peer_statistics;
{
std::lock_guard lock(clients_mutex);
if (std::find(clients.begin(), clients.end(), last_peer) == clients.end()) {
clients.push_back(last_peer);
peer_statistics = std::make_shared<Transport_Statistics>();
peer_statistics->record_started();
client_statistics.emplace(last_peer, peer_statistics);
}
else {
peer_statistics = client_statistics.at(last_peer);
}
}
statistics.record_received(n);
peer_statistics->record_received(n);
if (recv_buffer.write(recv_storage.data(), n)) {
recv_peers.push_back(last_peer);
}
else {
statistics.record_dropped(n);
peer_statistics->record_dropped(n);
}
update_buffer_statistics(statistics, send_buffer, recv_buffer);
start_read();
});
}
@@ -95,10 +129,18 @@ void UDP_Server::start_write() {
return;
send_storage.resize(out_len);
auto endpoint = send_endpoints.front();
const auto peer = endpoint_to_sockaddr(endpoint);
std::shared_ptr<Transport_Statistics> peer_statistics;
{
std::lock_guard lock(clients_mutex);
const auto iterator = client_statistics.find(peer);
if (iterator != client_statistics.end())
peer_statistics = iterator->second;
}
write_pending = true;
socket->async_send_to(
asio::buffer(send_storage.data(), send_storage.size()), endpoint,
[this](const asio::error_code& ec, std::size_t) {
[this, peer_statistics](const asio::error_code& ec, std::size_t sent) {
write_pending = false;
if (!bound)
return;
@@ -106,9 +148,18 @@ void UDP_Server::start_write() {
return;
if (!ec) {
send_buffer.skip_one();
statistics.record_sent(sent);
if (peer_statistics)
peer_statistics->record_sent(sent);
if (!send_endpoints.empty())
send_endpoints.erase(send_endpoints.begin());
}
else {
statistics.record_error(error_text(ec));
if (peer_statistics)
peer_statistics->record_error(error_text(ec));
}
update_buffer_statistics(statistics, send_buffer, recv_buffer);
send_storage.assign(static_cast<size_t>(read_chunk_size), 0);
start_write();
});
@@ -121,13 +172,20 @@ void UDP_Server::close() {
socket->cancel(ec);
socket->close(ec);
}
clients.clear();
{
std::lock_guard lock(clients_mutex);
for (const auto& [_, peer_statistics] : client_statistics)
peer_statistics->record_stopped();
clients.clear();
client_statistics.clear();
}
recv_peers.clear();
send_endpoints.clear();
bound = false;
has_last_peer = false;
socket_fd = static_cast<Socket_FD>(-1);
state = Not_Created;
statistics.record_stopped();
}
std::vector<UDP_Server::Read_Info> UDP_Server::read() {
std::vector<Read_Info> ret;
@@ -139,6 +197,7 @@ std::vector<UDP_Server::Read_Info> UDP_Server::read() {
ret.push_back({recv_peers.front(), std::string(data.data(), out_len)});
recv_peers.erase(recv_peers.begin());
}
update_buffer_statistics(statistics, send_buffer, recv_buffer);
return ret;
}
void UDP_Server::reply_last_peer(std::string_view data) {
@@ -154,16 +213,49 @@ void UDP_Server::send_to(std::string_view ip, uint16_t port,
return;
asio::error_code ec;
auto endpoint = asio::ip::udp::endpoint(asio::ip::make_address(ip, ec), port);
if (ec)
if (ec) {
statistics.record_error(error_text(ec));
return;
}
const auto peer = Sockaddr_In(ip, port);
std::shared_ptr<Transport_Statistics> peer_statistics;
{
std::lock_guard lock(clients_mutex);
const auto [iterator, inserted] = client_statistics.try_emplace(peer, std::make_shared<Transport_Statistics>());
peer_statistics = iterator->second;
if (inserted) {
clients.push_back(peer);
peer_statistics->record_started();
}
}
if (send_buffer.write(data.data(), data.size())) {
send_endpoints.push_back(endpoint);
}
else {
statistics.record_dropped(data.size());
peer_statistics->record_dropped(data.size());
}
update_buffer_statistics(statistics, send_buffer, recv_buffer);
start_write();
}
void UDP_Server::write_to_all_clients(std::string_view msg) {
for (const auto& client : clients) {
send_to(client.ip, static_cast<uint16_t>(client.port), msg);
for (const auto& client : get_all_clients()) {
send_to(client.address.ip, static_cast<uint16_t>(client.address.port), msg);
}
}
std::vector<UDP_Server::Client_Info> UDP_Server::get_all_clients() const {
std::vector<Client_Info> ret;
std::lock_guard lock(clients_mutex);
ret.reserve(clients.size());
for (const auto& client : clients)
ret.push_back({client, client_statistics.at(client)});
return ret;
}
std::optional<UDP_Server::Client_Info> UDP_Server::get_client(std::size_t index) const {
std::lock_guard lock(clients_mutex);
if (index >= clients.size())
return std::nullopt;
const auto& client = clients[index];
return Client_Info{client, client_statistics.at(client)};
}
} // namespace Psc::asio_socket
src/Psc_Cpp_Core/socket/UDP_Server.h
+15 -2
View File
@@ -1,9 +1,13 @@
#pragma once
#include <atomic>
#include <deque>
#include <map>
#include <memory>
#include <mutex>
#include <string_view>
#include <vector>
#include "ASIO_Utils.h"
#include "Psc_Cpp_Core/Statistics/Statistics.h"
namespace Psc::asio_socket {
class UDP_Server {
public:
@@ -34,9 +38,15 @@ public:
Not_Bind,
Working,
};
State state = Not_Created;
std::atomic<State> state{Not_Created};
struct Client_Info {
Sockaddr_In address;
std::shared_ptr<Transport_Statistics> statistics;
};
[[nodiscard]] std::vector<Client_Info> get_all_clients() const;
[[nodiscard]] std::optional<Client_Info> get_client(std::size_t index) const;
Transport_Statistics statistics;
Sockaddr_In bind_address;
std::vector<Sockaddr_In> clients;
Socket_FD socket_fd = static_cast<Socket_FD>(-1);
bool bound = false;
int read_chunk_size = 1024 * 1024;
@@ -55,6 +65,9 @@ private:
std::vector<char> send_storage;
bool read_pending = false;
bool write_pending = false;
mutable std::mutex clients_mutex;
std::vector<Sockaddr_In> clients;
std::map<Sockaddr_In, std::shared_ptr<Transport_Statistics>> client_statistics;
void start_read();
void start_write();
};