Compare commits

..

5 Commits
asio ... master

Author SHA1 Message Date
azw
7eb242f5b8 1 2023-06-09 06:57:46 +00:00
azw
571a29eebb 1 2023-06-05 09:50:17 +00:00
azw
70e076631c 1 2023-06-05 09:41:56 +00:00
azw
39b01b2794 1 2023-06-05 09:40:34 +00:00
azw
36f7099354 1 2023-06-05 09:36:21 +00:00
20 changed files with 120 additions and 243 deletions

View File

@ -1,16 +0,0 @@
const dgram = require('dgram');
const client = dgram.createSocket('udp4');
client.on('close', () => {
console.log('close');
});
client.on('error', () => {
console.log('error');
});
client.on('message', (msg, rinfo) => {
console.log(`接收到来自:${rinfo.address}:${rinfo.port} 的消息: ${msg}`);
});
client.send('hello', 17601, '108.137.177.19');

View File

@ -1,17 +0,0 @@
const dgram = require('dgram');
const server = dgram.createSocket('udp4');
server.on('message', (msg, rinfo) => {
console.log('rinfo.address = ' + rinfo.address);
console.log('rinfo.port = ' + rinfo.port);
console.log(msg.toString());
server.send(msg, rinfo.port, rinfo.address);
});
server.on('listening', () => {
console.log('address:' + server.address().address);
console.log('port:' + server.address().port);
});
server.bind(17601);

View File

@ -24,6 +24,7 @@ message SS_CMKcpHandshake
optional string account_id = 2; //id
optional string session_id = 3; //session id
optional string team_uuid = 4; //
optional int32 secret_key_place = 5; // 0() 1kcp底层协议头之后()
}
message SS_SMKcpHandshake

View File

@ -20,9 +20,9 @@ endif()
set(CMAKE_CXX_FLAGS_RELEASE "-O3 -Wall -g -std=gnu++1z -DGAME_ID=${GAME_ID} -DNDEBUG")
if (${ASAN})
set(CMAKE_CXX_FLAGS_DEBUG "-Wall -g -std=gnu++1z -DGAME_ID=${GAME_ID} -DDEBUG -fsanitize=address -fno-omit-frame-pointer -DUSE_ASIO=1")
set(CMAKE_CXX_FLAGS_DEBUG "-Wall -g -std=gnu++1z -DGAME_ID=${GAME_ID} -DDEBUG -fsanitize=address -fno-omit-frame-pointer")
else()
set(CMAKE_CXX_FLAGS_DEBUG "-Wall -g -std=gnu++1z -DGAME_ID=${GAME_ID} -DDEBUG -DUSE_ASIO=1")
set(CMAKE_CXX_FLAGS_DEBUG "-Wall -g -std=gnu++1z -DGAME_ID=${GAME_ID} -DDEBUG")
endif()
include_directories(

View File

@ -25,7 +25,6 @@
#include "upstreammgr.h"
#include "master.h"
#include "mastermgr.h"
#include "iomgr.h"
#include "longsessionmgr.h"
@ -128,9 +127,6 @@ bool App::Init(int argc, char* argv[])
f8::Timer::Instance()->Init();
JsonDataMgr::Instance()->Init();
uuid_->SetMachineId((node_id_ - 1) * MAX_NODE_ID + instance_id_);
#ifdef USE_ASIO
IoMgr::Instance()->Init();
#endif
DownStreamMgr::Instance()->Init();
MasterMgr::Instance()->Init();
UpStreamMgr::Instance()->Init();
@ -193,9 +189,6 @@ void App::UnInit()
MasterMgr::Instance()->UnInit();
UpStreamMgr::Instance()->UnInit();
DownStreamMgr::Instance()->UnInit();
#ifdef USE_ASIO
IoMgr::Instance()->UnInit();
#endif
JsonDataMgr::Instance()->UnInit();
f8::Timer::Instance()->UnInit();
f8::MsgQueue::Instance()->UnInit();

View File

@ -23,13 +23,6 @@ enum InnerMesssage_e
IM_HandleRedirect
};
enum IoContext_e
{
IC_Master,
IC_UpStream,
IC_Max
};
//网络处理对象
enum NetHandler_e
{

View File

@ -1,41 +0,0 @@
#include "precompile.h"
#include "iomgr.h"
#ifdef USE_ASIO
void IoMgr::Init()
{
for (int i = 0; i < IC_Max; ++i) {
io_contexts_.push_back(std::vector<std::shared_ptr<asio::io_context>>());
auto& contexts = io_contexts_.at(i);
contexts.push_back(std::make_shared<asio::io_context>());
new std::thread(&IoMgr::WorkerThreadProc, this, contexts.at(0));
}
}
void IoMgr::UnInit()
{
}
std::shared_ptr<asio::io_context> IoMgr::GetIoContext(int type)
{
if (type >= 0 && type < io_contexts_.size()) {
auto& contexts = io_contexts_.at(type);
if (!contexts.empty()) {
return contexts.at(rand() % contexts.size());
}
}
return nullptr;
}
void IoMgr::WorkerThreadProc(std::shared_ptr<asio::io_context> io_context)
{
while (true) {
io_context->run();
io_context->reset();
}
}
#endif

View File

@ -1,28 +0,0 @@
#pragma once
#ifdef USE_ASIO
#include <asio.hpp>
class IoMgr : public a8::Singleton<IoMgr>
{
private:
IoMgr() {};
friend class a8::Singleton<IoMgr>;
public:
void Init();
void UnInit();
std::shared_ptr<asio::io_context> GetIoContext(int type);
private:
void WorkerThreadProc(std::shared_ptr<asio::io_context> io_context);
private:
std::vector<std::vector<std::shared_ptr<asio::io_context>>> io_contexts_;
};
#endif

View File

@ -1,6 +1,7 @@
#include "precompile.h"
#include <memory.h>
#include <string.h>
#include <f8/netmsghandler.h>
#include <f8/udplog.h>
@ -33,23 +34,22 @@ KcpSession::~KcpSession()
}
void KcpSession::Init(int socket_handle)
void KcpSession::Init(int socket_handle, int secret_key_place)
{
socket_handle_ = socket_handle;
secret_key_ = App::Instance()->NewUuid();
kcp_ = ikcp_create(socket_handle_, (void*)this);
#if 1
const KcpConf& kcp_conf = JsonDataMgr::Instance()->GetKcpConf();
ikcp_wndsize(kcp_, kcp_conf.sndwnd, kcp_conf.rcvwnd);
ikcp_nodelay(kcp_, kcp_conf.nodelay, kcp_conf.interval, kcp_conf.resend, kcp_conf.nc);
kcp_->rx_minrto = kcp_conf.rx_minrto;
kcp_->fastresend = kcp_conf.fastresend;
#else
ikcp_wndsize(kcp_, 128, 128);
ikcp_nodelay(kcp_, 2, 10, 2, 1);
kcp_->rx_minrto = 10;
kcp_->fastresend = 1;
#endif
secret_key_place_ = secret_key_place;
if (secret_key_place_ > 0) {
secret_key_place_ = 1;
} else if (secret_key_place_ < 0) {
secret_key_place_ = 0;
}
kcp_->output = UdpOutput;
init_tick_ = a8::XGetTickCount();
}
@ -75,8 +75,21 @@ void KcpSession::SendClientMsg(char* buf, int buf_len)
void KcpSession::OnRecvPacket(a8::UdpPacket* pkt)
{
const int IKCP_OVERHEAD = 24;
remote_addr_ = pkt->remote_addr;
ikcp_input(kcp_, pkt->buf, pkt->buf_len);
if (GetSecretKeyPlace()) {
char* buf = (char*)malloc(pkt->buf_len - GetSecretKeyLen());
int buflen = pkt->buf_len - GetSecretKeyLen();
memmove(buf, pkt->buf, IKCP_OVERHEAD);
if (pkt->buf_len > IKCP_OVERHEAD + GetSecretKeyLen()) {
memmove(buf + IKCP_OVERHEAD, pkt->buf + IKCP_OVERHEAD + GetSecretKeyLen(), buflen - IKCP_OVERHEAD);
}
ikcp_input(kcp_, buf, buflen);
free(buf);
} else {
ikcp_input(kcp_, pkt->buf, pkt->buf_len);
}
}
void KcpSession::UpdateInput()
@ -91,8 +104,20 @@ void KcpSession::UpdateInput()
void KcpSession::DecodeUserPacket(char* buf, int& offset, unsigned int buflen)
{
//packagelen + msgid + magiccode + msgbody
//2 + 2 + 4+ xx + \0 + xx
if (GetSecretKeyPlace()) {
DecodeUserPacketNew(buf, offset, buflen);
} else {
DecodeUserPacketOld(buf, offset, buflen);
}
}
int KcpSession::GetSecretKeyPlace()
{
return secret_key_place_;
}
void KcpSession::DecodeUserPacketOld(char* buf, int& offset, unsigned int buflen)
{
bool warning = false;
while (buflen - offset >= sizeof(f8::PackHead) + GetSecretKeyLen()) {
long long secret_key = KcpSession::ReadSecretKey(&buf[offset], buflen);
@ -124,6 +149,37 @@ void KcpSession::DecodeUserPacket(char* buf, int& offset, unsigned int buflen)
}
if (warning) {
f8::UdpLog::Instance()->Warning("收到kcp client非法数据包", {});
f8::UdpLog::Instance()->Warning("收到kcp client非法数据包1", {});
}
}
void KcpSession::DecodeUserPacketNew(char* buf, int& offset, unsigned int buflen)
{
bool warning = false;
while (buflen - offset >= sizeof(f8::PackHead)) {
f8::PackHead* p = (f8::PackHead*)&buf[offset];
if (p->magic_code == f8::MAGIC_CODE) {
if (buflen - offset < sizeof(f8::PackHead) + p->packlen) {
break;
}
App::Instance()->AddSocketMsg(SF_Client,
socket_handle_,
0,
//saddr,
p->msgid,
p->seqid,
&buf[offset + sizeof(f8::PackHead)],
p->packlen,
ST_Udp);
offset += sizeof(f8::PackHead) + p->packlen;
} else {
warning = true;
offset++;
continue;
}
}
if (warning) {
f8::UdpLog::Instance()->Warning("收到kcp client非法数据包2", {});
}
}

View File

@ -11,7 +11,7 @@ public:
KcpSession();
~KcpSession();
void Init(int socket_handle);
void Init(int socket_handle, int secret_key_place);
void UnInit();
void Update(long long tick);
@ -19,6 +19,7 @@ public:
int GetSocketHandle() { return socket_handle_; }
long long GetSecretKey() { return secret_key_; }
void* GetSecretKeyDataPtr() { return &secret_key_; }
int GetSecretKeyPlace();
void SendClientMsg(char* buf, int buf_len);
virtual void OnRecvPacket(a8::UdpPacket* pkt) override;
@ -31,11 +32,14 @@ public:
protected:
virtual void DecodeUserPacket(char* buf, int& offset, unsigned int buflen) override;
void DecodeUserPacketOld(char* buf, int& offset, unsigned int buflen);
void DecodeUserPacketNew(char* buf, int& offset, unsigned int buflen);
private:
void UpdateInput();
private:
int secret_key_place_ = 0;
int socket_handle_ = 0;
long long secret_key_ = 0;
ikcpcb* kcp_ = nullptr;

View File

@ -5,10 +5,13 @@
#include "longsession.h"
#include "kcpsession.h"
#include "ss_msgid.pb.h"
#include "ss_proto.pb.h"
void LongSession::Init(f8::MsgHdr& hdr, const ss::SS_CMKcpHandshake& msg)
{
kcp_session_ = std::make_shared<KcpSession>();
kcp_session_->Init(hdr.socket_handle);
kcp_session_->Init(hdr.socket_handle, msg.secret_key_place());
}
void LongSession::UnInit()
@ -25,3 +28,8 @@ void LongSession::UpdatePing()
{
last_ping_tick_ = a8::XGetTickCount();
}
int LongSession::GetSecretKeyPlace()
{
return kcp_session_->GetSecretKeyPlace();
}

View File

@ -17,6 +17,7 @@ class LongSession
std::shared_ptr<KcpSession> GetKcpSession() { return kcp_session_; }
void UpdatePing();
int GetSecretKeyPlace();
private:
long long last_ping_tick_ = 0;

View File

@ -133,30 +133,31 @@ void LongSessionMgr::ProcUdpPacket(a8::UdpPacket* pkt)
});
return;
}
session->GetKcpSession()->OnRecvPacket(pkt);
#if 0
if (pkt->buf_len > IKCP_OVERHEAD + KcpSession::GetSecretKeyLen()) {
long long secret_key = KcpSession::ReadSecretKey(pkt->buf + IKCP_OVERHEAD, pkt->buf_len);
if (secret_key == session->GetKcpSession()->GetSecretKey()) {
session->GetKcpSession()->OnRecvPacket(pkt);
if (session->GetSecretKeyPlace()) {
if (pkt->buf_len >= IKCP_OVERHEAD + KcpSession::GetSecretKeyLen()) {
long long secret_key = KcpSession::ReadSecretKey(pkt->buf + IKCP_OVERHEAD, pkt->buf_len);
if (secret_key == session->GetKcpSession()->GetSecretKey()) {
session->GetKcpSession()->OnRecvPacket(pkt);
} else {
f8::UdpLog::Instance()->Warning("ProcUdpPacket host:%s port:%d socket_handle%d secret_key:%d secret_error",
{
inet_ntoa(pkt->remote_addr.sin_addr),
pkt->remote_addr.sin_port,
socket_handle,
secret_key
});
}
} else {
f8::UdpLog::Instance()->Warning("ProcUdpPacket host:%s port:%d socket_handle%d secret_key:%d secret_error",
f8::UdpLog::Instance()->Warning("ProcUdpPacket host:%s port:%d buflen:%d bufflen_error",
{
inet_ntoa(pkt->remote_addr.sin_addr),
pkt->remote_addr.sin_port,
socket_handle,
secret_key
pkt->buf_len
});
}
} else {
f8::UdpLog::Instance()->Warning("ProcUdpPacket host:%s port:%d buflen:%d bufflen_error",
{
inet_ntoa(pkt->remote_addr.sin_addr),
pkt->remote_addr.sin_port,
pkt->buf_len
});
session->GetKcpSession()->OnRecvPacket(pkt);
}
#endif
}
void LongSessionMgr::DelSession(int socket_handle)

View File

@ -2,11 +2,7 @@
#include <string.h>
#ifdef USE_ASIO
#include <a8/asiotcpclient.h>
#else
#include <a8/tcpclient.h>
#endif
#include <f8/udplog.h>
#include <f8/timer.h>
@ -16,7 +12,6 @@
#include "master.h"
#include "app.h"
#include "iomgr.h"
const int PACK_MAX = 1024 * 64;
@ -29,15 +24,9 @@ void Master::Init(int instance_id, const std::string& remote_ip, int remote_port
recv_bufflen_ = 0;
last_pong_tick = a8::XGetTickCount();
recv_buff_ = (char*) malloc(PACK_MAX * 2);
#ifdef USE_ASIO
auto io_context = IoMgr::Instance()->GetIoContext(IC_Master);
if (!io_context) {
abort();
}
tcp_client_ = std::make_shared<a8::AsioTcpClient>(*io_context, remote_ip, remote_port);
#else
tcp_client_ = std::make_shared<a8::TcpClient>(remote_ip, remote_port);
#endif
tcp_client_ = std::make_shared<a8::TcpClient>();
tcp_client_->remote_address = remote_ip;
tcp_client_->remote_port = remote_port;
tcp_client_->on_error = std::bind(&Master::on_error, this, std::placeholders::_1, std::placeholders::_2);
tcp_client_->on_connect = std::bind(&Master::on_connect, this, std::placeholders::_1);
tcp_client_->on_disconnect = std::bind(&Master::on_disconnect, this, std::placeholders::_1);
@ -83,45 +72,29 @@ bool Master::Connected()
return tcp_client_->Connected();
}
#ifdef USE_ASIO
void Master::on_error(a8::AsioTcpClient* sender, int errorId)
#else
void Master::on_error(a8::TcpClient* sender, int errorId)
#endif
{
f8::UdpLog::Instance()->Error("Master errorid=%d remote_ip:%s remote_port:%d",
{
errorId,
sender->GetRemoteAddress(),
sender->GetRemotePort()
sender->remote_address,
sender->remote_port
});
}
#ifdef USE_ASIO
void Master::on_connect(a8::AsioTcpClient* sender)
#else
void Master::on_connect(a8::TcpClient* sender)
#endif
{
recv_bufflen_ = 0;
f8::UdpLog::Instance()->Info("masterserver connected", {});
}
#ifdef USE_ASIO
void Master::on_disconnect(a8::AsioTcpClient* sender)
#else
void Master::on_disconnect(a8::TcpClient* sender)
#endif
{
recv_bufflen_ = 0;
f8::UdpLog::Instance()->Info("masterserver %d disconnected after 10s later reconnect", {instance_id});
}
#ifdef USE_ASIO
void Master::on_socketread(a8::AsioTcpClient* sender, char* buf, unsigned int len)
#else
void Master::on_socketread(a8::TcpClient* sender, char* buf, unsigned int len)
#endif
{
#if 0
++App::Instance()->perf.read_count;

View File

@ -5,7 +5,6 @@
namespace a8
{
class TcpClient;
class AsioTcpClient;
}
class Master
@ -32,28 +31,17 @@ class Master
}
private:
#ifdef USE_ASIO
void on_error(a8::AsioTcpClient* sender, int errorId);
void on_connect(a8::AsioTcpClient* sender);
void on_disconnect(a8::AsioTcpClient* sender);
void on_socketread(a8::AsioTcpClient* sender, char* buf, unsigned int len);
#else
void on_error(a8::TcpClient* sender, int errorId);
void on_connect(a8::TcpClient* sender);
void on_disconnect(a8::TcpClient* sender);
void on_socketread(a8::TcpClient* sender, char* buf, unsigned int len);
#endif
void CheckAlive();
private:
char *recv_buff_ = nullptr;
unsigned int recv_bufflen_ = 0;
#ifdef USE_ASIO
std::shared_ptr<a8::AsioTcpClient> tcp_client_;
#else
std::shared_ptr<a8::TcpClient> tcp_client_;
#endif
f8::TimerWp timer_wp_;
f8::Attacher attacher_;
};

View File

@ -2,11 +2,7 @@
#include <string.h>
#ifdef USE_ASIO
#include <a8/asiotcpclient.h>
#else
#include <a8/tcpclient.h>
#endif
#include <f8/udplog.h>
#include <f8/timer.h>
@ -17,7 +13,6 @@
#include "upstream.h"
#include "app.h"
#include "iomgr.h"
const int PACK_MAX = 1024 * 64 * 2;
@ -33,15 +28,9 @@ void UpStream::Init(int instance_id, const std::string& remote_ip, int remote_po
recv_bufflen_ = 0;
last_pong_tick = a8::XGetTickCount();
recv_buff_ = (char*) malloc(PACK_MAX * 2);
#ifdef USE_ASIO
auto io_context = IoMgr::Instance()->GetIoContext(IC_UpStream);
if (!io_context) {
abort();
}
tcp_client_ = std::make_shared<a8::AsioTcpClient>(*io_context, remote_ip, remote_port);
#else
tcp_client_ = std::make_shared<a8::TcpClient>(remote_ip, remote_port);
#endif
tcp_client_ = std::make_shared<a8::TcpClient>();
tcp_client_->remote_address = remote_ip;
tcp_client_->remote_port = remote_port;
tcp_client_->on_error = std::bind(&UpStream::on_error, this, std::placeholders::_1, std::placeholders::_2);
tcp_client_->on_connect = std::bind(&UpStream::on_connect, this, std::placeholders::_1);
tcp_client_->on_disconnect = std::bind(&UpStream::on_disconnect, this, std::placeholders::_1);
@ -159,31 +148,23 @@ void UpStream::ForwardClientMsgEx(f8::MsgHdr* hdr)
}
}
#ifdef USE_ASIO
void UpStream::on_error(a8::AsioTcpClient* sender, int errorId)
#else
void UpStream::on_error(a8::TcpClient* sender, int errorId)
#endif
{
f8::UdpLog::Instance()->Error("target server errorid=%d remote_ip:%s remote_port:%d",
{
errorId,
sender->GetRemoteAddress(),
sender->GetRemotePort()
sender->remote_address,
sender->remote_port
});
}
#ifdef USE_ASIO
void UpStream::on_connect(a8::AsioTcpClient* sender)
#else
void UpStream::on_connect(a8::TcpClient* sender)
#endif
{
recv_bufflen_ = 0;
f8::UdpLog::Instance()->Info("target server connected remote_ip:%s remote_port:%d",
{
sender->GetRemoteAddress(),
sender->GetRemotePort()
sender->remote_address,
sender->remote_port
});
f8::MsgQueue::Instance()->PostMsg
(
@ -197,19 +178,15 @@ void UpStream::on_connect(a8::TcpClient* sender)
);
}
#ifdef USE_ASIO
void UpStream::on_disconnect(a8::AsioTcpClient* sender)
#else
void UpStream::on_disconnect(a8::TcpClient* sender)
#endif
{
recv_bufflen_ = 0;
f8::UdpLog::Instance()->Info("target server %d disconnected after 10s later reconnect "
"remote_ip:%s remote_port:%d",
{
instance_id,
sender->GetRemoteAddress(),
sender->GetRemotePort()
sender->remote_address,
sender->remote_port
});
f8::MsgQueue::Instance()->PostMsg
(
@ -223,11 +200,7 @@ void UpStream::on_disconnect(a8::TcpClient* sender)
);
}
#ifdef USE_ASIO
void UpStream::on_socketread(a8::AsioTcpClient* sender, char* buf, unsigned int len)
#else
void UpStream::on_socketread(a8::TcpClient* sender, char* buf, unsigned int len)
#endif
{
#if 0
++App::Instance()->perf.read_count;

View File

@ -5,7 +5,6 @@
namespace a8
{
class TcpClient;
class AsioTcpClient;
}
struct UpStreamMsgNode
@ -27,7 +26,6 @@ class UpStream
a8::tick_t last_pong_tick = 0;
public:
void Init(int instance_id, const std::string& remote_ip, int remote_port);
void UnInit();
@ -56,17 +54,10 @@ class UpStream
void ForwardClientMsgEx(f8::MsgHdr* hdr);
private:
#ifdef USE_ASIO
void on_error(a8::AsioTcpClient* sender, int errorId);
void on_connect(a8::AsioTcpClient* sender);
void on_disconnect(a8::AsioTcpClient* sender);
void on_socketread(a8::AsioTcpClient* sender, char* buf, unsigned int len);
#else
void on_error(a8::TcpClient* sender, int errorId);
void on_connect(a8::TcpClient* sender);
void on_disconnect(a8::TcpClient* sender);
void on_socketread(a8::TcpClient* sender, char* buf, unsigned int len);
#endif
void CheckAlive();
void AddStockMsg(unsigned short socket_handle, int msgid, ::google::protobuf::Message* msg,
@ -75,11 +66,7 @@ class UpStream
private:
char *recv_buff_ = nullptr;
unsigned int recv_bufflen_ = 0;
#ifdef USE_ASIO
std::shared_ptr<a8::AsioTcpClient> tcp_client_;
#else
std::shared_ptr<a8::TcpClient> tcp_client_;
#endif
f8::TimerWp timer_wp_;
f8::Attacher attacher_;

View File

@ -55,6 +55,7 @@ std::weak_ptr<UpStream> UpStreamMgr::RecreateUpStream(const std::string& host, i
int instance_id = curr_id_;
std::string remote_ip = host;
int remote_port = port;
std::shared_ptr<UpStream> conn = std::make_shared<UpStream>();
conn->Init(instance_id, remote_ip, remote_port);
id_hash_[conn->instance_id] = conn;

2
third_party/a8 vendored

@ -1 +1 @@
Subproject commit 466448796dff77a16190519c25879cfe0bc534ec
Subproject commit 1e577389c8a2870db9ddbf18577bfca24def049b

2
third_party/f8 vendored

@ -1 +1 @@
Subproject commit 8a49aa65c8576cabe1672be55fc903a6b17e2c37
Subproject commit 243bbe515ef4a01089f9a6cf608c93d4097018de