server_common/a8
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

remove tcpsession2.*

Browse Source
This commit is contained in:
aozhiwei 2019-07-01 09:58:16 +08:00
parent bb10963a71
commit d0c032181b
7 changed files with 136 additions and 626 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
a8/basehttpsession.h
View File

@ -1,11 +1,7 @@
#ifndef A8_BASE_HTTPSESSION_H
#define A8_BASE_HTTPSESSION_H
#ifdef A8_TCP_SESSION2
#include <a8/tcpsession2.h>
#else
#include <a8/tcpsession.h>
#endif
namespace a8
{

6
a8/tcplistener.cc
View File

@ -10,11 +10,7 @@
#include <a8/a8.h>
#include <a8/tcplistener.h>
#ifdef A8_TCP_SESSION2
#include <a8/tcpsession2.h>
#else
#include <a8/tcpsession.h>
#endif
#include <a8/tcpsessionpool.h>
namespace a8
@ -288,9 +284,7 @@ namespace a8
#else
ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLRDHUP;
#endif
#ifdef A8_TCP_SESSION2
ev.events = EPOLLIN | EPOLLRDHUP;
#endif
ev.data.ptr = p;
int n = ::epoll_ctl(epoll_fd, EPOLL_CTL_ADD, sock, &ev);
assert(n == 0);

230
a8/tcpsession.cc
View File

@ -14,8 +14,6 @@
#include <a8/tcplistener.h>
#include <a8/perfmonitor.h>
#ifndef A8_TCP_SESSION2
static const int DEFAULT_MAX_PACKET_LEN = 1024 * 10;
static const int DEFAULT_MAX_RECV_BUFFERSIZE = 1024 * 64;
@ -30,7 +28,6 @@ namespace a8
TcpSession::~TcpSession()
{
Destory();
}
void TcpSession::SetMaxPacketLen(int max_packet_len)
@ -76,13 +73,20 @@ namespace a8
p->buff = (char*)malloc(bufflen);
memmove(p->buff, buff, bufflen);
p->bufflen = bufflen;
#ifdef NEW_NET
bool is_first_package = false;
send_buffer_mutex_.lock();
if (!work_node_ && !top_node_) {
work_node_ = p;
is_first_package = true;
#if 1
if (bot_node_) {
bot_node_->next = p;
bot_node_ = p;
} else {
top_node_ = p;
bot_node_ = p;
}
if (!sending_) {
NotifyEpollSend();
}
#else
if (sending_) {
if (bot_node_) {
bot_node_->next = p;
bot_node_ = p;
@ -90,27 +94,43 @@ namespace a8
top_node_ = p;
bot_node_ = p;
}
}
send_buffer_mutex_.unlock();
if (work_node_ && is_first_package) {
AsyncSend(is_first_package);
}
#else
send_buffer_mutex_.lock();
if(work_node_ == NULL && top_node_ == NULL){
work_node_ = p;
AsyncSend(true);
}else{
if (bot_node_){
bot_node_->next = p;
bot_node_ = p;
}else{
top_node_ = p;
bot_node_ = p;
NotifyEpollSend();
} else {
if (work_node_) {
abort();
}
if (top_node_ || bot_node_) {
abort();
}
int ret = DirectSend(p);
switch (ret) {
case 0:
case -2:
{
assert(ret == 0);
if (p->buff) {
free(p->buff);
}
free(p);
}
break;
case -1:
{
if (top_node_ || bot_node_) {
abort();
}
top_node_ = p;
bot_node_ = p;
NotifyEpollSend();
}
break;
default:
abort();
}
}
send_buffer_mutex_.unlock();
#endif
send_buffer_mutex_.unlock();
++master->send_node_num;
}
}
@ -136,15 +156,17 @@ namespace a8
void TcpSession::Reset()
{
ClearSendBuff();
ClearWorkBuff();
socket_ = -1;
remote_address = "";
remote_port = 0;
top_node_ = NULL;
bot_node_ = NULL;
work_node_ = NULL;
top_node_ = nullptr;
bot_node_ = nullptr;
work_node_ = nullptr;
socket_handle = 0;
recv_bufflen_ = 0;
is_activite = false;
sending_ = false;
}
void TcpSession::Destory()
@ -152,10 +174,11 @@ namespace a8
if (recv_buff_) {
recv_bufflen_ = 0;
free(recv_buff_);
recv_buff_ = NULL;
recv_buff_ = nullptr;
}
Close();
ClearSendBuff();
ClearWorkBuff();
}
void TcpSession::_ForceClose()
@ -247,7 +270,7 @@ namespace a8
if(socket_ == -1){
return;
}
#ifdef NEW_NET
++epoll_out_times;
if (!work_node_) {
send_buffer_mutex_.lock();
work_node_ = top_node_;
@ -256,20 +279,8 @@ namespace a8
send_buffer_mutex_.unlock();
}
if (work_node_) {
AsyncSend(false);
AsyncSend();
}
#else
send_buffer_mutex_.lock();
if (!work_node_) {
work_node_ = top_node_;
top_node_ = nullptr;
bot_node_ = nullptr;
}
if (work_node_) {
AsyncSend(false);
}
send_buffer_mutex_.unlock();
#endif
}
void TcpSession::Close()
@ -285,16 +296,13 @@ namespace a8
void TcpSession::ClearSendBuff()
{
a8::SendQueueNode* p_top_node_ = nullptr;
a8::SendQueueNode* p_work_node_ = nullptr;
send_buffer_mutex_.lock();
p_top_node_ = top_node_;
p_work_node_ = work_node_;
top_node_ = NULL;
bot_node_ = NULL;
work_node_ = NULL;
top_node_ = nullptr;
bot_node_ = nullptr;
send_buffer_mutex_.unlock();
a8::SendQueueNode *pdelnode = NULL;
a8::SendQueueNode *pdelnode = nullptr;
while (p_top_node_) {
pdelnode = p_top_node_;
p_top_node_ = p_top_node_->next;
@ -302,94 +310,104 @@ namespace a8
free(pdelnode->buff);
}
free(pdelnode);
--master->send_node_num;
}
while (p_work_node_) {
pdelnode = p_work_node_;
p_work_node_ = p_work_node_->next;
}
void TcpSession::ClearWorkBuff()
{
while (work_node_) {
a8::SendQueueNode *pdelnode = work_node_;
work_node_ = work_node_->next;
if (pdelnode->buff) {
free(pdelnode->buff);
}
free(pdelnode);
--master->send_node_num;
}
}
void TcpSession::AsyncSend(bool is_first_package)
void TcpSession::AsyncSend()
{
while (work_node_) {
#ifdef A8_PERF
a8::tick_t begin_tick = a8::XGetTickCount();
#endif
int sentbytes = ::send(socket_,
work_node_->buff + work_node_->sent_bytes,
work_node_->bufflen - work_node_->sent_bytes,
0);
#ifdef A8_PERF
a8::tick_t end_tick = a8::XGetTickCount();
if (end_tick - begin_tick > a8::PerfMonitor::Instance()->max_send_time) {
a8::PerfMonitor::Instance()->max_send_time = end_tick - begin_tick;
}
#endif
if (sentbytes <= 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
#ifdef A8_PERF
++a8::PerfMonitor::Instance()->send_eagain_times;
#endif
#ifdef NEW_NET
if (is_first_package) {
struct epoll_event ev;
ev.data.fd = socket_;
ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLRDHUP;
ev.data.ptr = this;
::epoll_ctl(epoll_fd, EPOLL_CTL_MOD, socket_, &ev);
return;
}
#endif
continue;
auto err_code = errno;
if (err_code == EAGAIN || err_code == EWOULDBLOCK) {
break;
} else {
Close();
break;
}
}
work_node_->sent_bytes += sentbytes;
if (work_node_->sent_bytes >= work_node_->bufflen) {
a8::SendQueueNode *pdelnode = work_node_;
#ifdef NEW_NET
#else
a8::SendQueueNode *nextnode = work_node_->next;
#endif
#ifdef NEW_NET
send_buffer_mutex_.lock();
work_node_ = work_node_->next;
--master->send_node_num;
++master->sent_node_num;
master->sent_bytes_num += pdelnode->sent_bytes;
if (!work_node_) {
work_node_ = top_node_;
top_node_ = nullptr;
bot_node_ = nullptr;
send_buffer_mutex_.lock();
if (top_node_) {
work_node_ = top_node_;
top_node_ = nullptr;
bot_node_ = nullptr;
}
if (!work_node_) {
sending_ = false;
struct epoll_event ev;
ev.data.fd = socket_;
ev.events = EPOLLIN | EPOLLRDHUP;
ev.data.ptr = this;
::epoll_ctl(epoll_fd, EPOLL_CTL_MOD, socket_, &ev);
}
send_buffer_mutex_.unlock();
}
#ifdef NEW_NET
if (!work_node_ && !is_first_package) {
struct epoll_event ev;
ev.data.fd = socket_;
ev.events = EPOLLIN | EPOLLET | EPOLLRDHUP;
ev.data.ptr = this;
::epoll_ctl(epoll_fd, EPOLL_CTL_MOD, socket_, &ev);
}
#endif
send_buffer_mutex_.unlock();
#else
work_node_ = work_node_->next; //!!!!要处理重入问题
#endif
if (pdelnode->buff) {
free(pdelnode->buff);
}
free(pdelnode);
#ifdef NEW_NET
#else
if (!nextnode) {
break;
}
}
}
void TcpSession::NotifyEpollSend()
{
sending_ = true;
struct epoll_event ev;
ev.data.fd = socket_;
ev.events = EPOLLIN | EPOLLOUT | EPOLLRDHUP;
ev.data.ptr = this;
::epoll_ctl(epoll_fd, EPOLL_CTL_MOD, socket_, &ev);
}
int TcpSession::DirectSend(a8::SendQueueNode* p)
{
++direct_send_times;
while (true) {
int sentbytes = ::send(socket_,
p->buff + p->sent_bytes,
p->bufflen - p->sent_bytes,
0);
if (sentbytes <= 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
return -1;
} else {
Close();
return -2;
}
} else {
p->sent_bytes += sentbytes;
if (p->sent_bytes >= p->bufflen) {
return 0;
}
#endif
}
}
}
}
#endif

18
a8/tcpsession.h
View File

@ -1,7 +1,6 @@
#ifndef A8_TCPSESSION_H
#define A8_TCPSESSION_H
#ifndef A8_TCPSESSION2_H
#define A8_TCPSESSION2_H
#ifndef A8_TCP_SESSION2
#include <mutex>
namespace a8
@ -32,8 +31,8 @@ namespace a8
virtual void OnError(int);
virtual void OnConnect();
virtual void OnDisConnect();
bool Alive();
virtual void Destory();
bool Alive();
protected:
virtual void SendBuff(const char* buff, unsigned int bufflen);
@ -52,15 +51,22 @@ namespace a8
private:
void ClearSendBuff();
void AsyncSend(bool is_first_package);
void ClearWorkBuff();
void AsyncSend();
void NotifyEpollSend();
int DirectSend(a8::SendQueueNode* node);
protected:
char *recv_buff_ = nullptr;
int recv_bufflen_ = 0;
int max_packet_len_ = 0;
volatile long long epoll_out_times = 0;
volatile long long epoll_in_times = 0;
volatile long long direct_send_times = 0;
private:
int socket_ = 0;
volatile bool sending_ = false;
a8::SendQueueNode* top_node_ = nullptr;
a8::SendQueueNode* bot_node_ = nullptr;
a8::SendQueueNode* work_node_ = nullptr;
@ -70,5 +76,5 @@ namespace a8
};
}
#endif
#endif

419
a8/tcpsession2.cc
View File

@ -1,419 +0,0 @@
#include <memory.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <netinet/tcp.h>
#include <thread>
#include <mutex>
#include <a8/a8.h>
#ifdef A8_TCP_SESSION2
#include <a8/tcpsession2.h>
#else
#include <a8/tcpsession.h>
#endif
#include <a8/tcplistener.h>
#include <a8/perfmonitor.h>
#ifdef A8_TCP_SESSION2
static const int DEFAULT_MAX_PACKET_LEN = 1024 * 10;
static const int DEFAULT_MAX_RECV_BUFFERSIZE = 1024 * 64;
namespace a8
{
TcpSession::TcpSession()
{
INIT_LIST_HEAD(&session_entry);
max_packet_len_ = DEFAULT_MAX_PACKET_LEN;
}
TcpSession::~TcpSession()
{
}
void TcpSession::SetMaxPacketLen(int max_packet_len)
{
max_packet_len_ = std::max(max_packet_len, DEFAULT_MAX_PACKET_LEN);
}
int TcpSession::Socket()
{
return socket_;
}
void TcpSession::OnError(int)
{
}
void TcpSession::OnConnect()
{
}
void TcpSession::OnDisConnect()
{
}
void TcpSession::DecodePacket(char* buf, int& offset, unsigned int buflen)
{
DecodeUserPacket(buf, offset, buflen);
}
bool TcpSession::Alive()
{
return is_activite || time(nullptr) - create_time < 60 * 5;
}
void TcpSession::SendBuff(const char* buff, unsigned int bufflen)
{
if (socket_ == -1) {
return;
}
if (bufflen > 0) {
a8::SendQueueNode* p = (a8::SendQueueNode*)malloc(sizeof(a8::SendQueueNode));
memset(p, 0, sizeof(a8::SendQueueNode));
p->buff = (char*)malloc(bufflen);
memmove(p->buff, buff, bufflen);
p->bufflen = bufflen;
send_buffer_mutex_.lock();
#if 1
if (bot_node_) {
bot_node_->next = p;
bot_node_ = p;
} else {
top_node_ = p;
bot_node_ = p;
}
if (!sending_) {
NotifyEpollSend();
}
#else
if (sending_) {
if (bot_node_) {
bot_node_->next = p;
bot_node_ = p;
} else {
top_node_ = p;
bot_node_ = p;
}
NotifyEpollSend();
} else {
if (work_node_) {
abort();
}
if (top_node_ || bot_node_) {
abort();
}
int ret = DirectSend(p);
switch (ret) {
case 0:
case -2:
{
assert(ret == 0);
if (p->buff) {
free(p->buff);
}
free(p);
}
break;
case -1:
{
if (top_node_ || bot_node_) {
abort();
}
top_node_ = p;
bot_node_ = p;
NotifyEpollSend();
}
break;
default:
abort();
}
}
#endif
send_buffer_mutex_.unlock();
++master->send_node_num;
}
}
void TcpSession::SendText(const std::string& text)
{
SendBuff(text.c_str(), text.size());
}
void TcpSession::SetSocket(int sock)
{
socket_ = sock;
}
bool TcpSession::AllocRecvBuf()
{
if (!recv_buff_) {
recv_buff_ = (char *)malloc(max_packet_len_ + 1);
}
recv_bufflen_ = 0;
return recv_buff_ != nullptr;
}
void TcpSession::Reset()
{
ClearSendBuff();
ClearWorkBuff();
socket_ = -1;
remote_address = "";
remote_port = 0;
top_node_ = nullptr;
bot_node_ = nullptr;
work_node_ = nullptr;
socket_handle = 0;
recv_bufflen_ = 0;
is_activite = false;
sending_ = false;
}
void TcpSession::Destory()
{
if (recv_buff_) {
recv_bufflen_ = 0;
free(recv_buff_);
recv_buff_ = nullptr;
}
Close();
ClearSendBuff();
ClearWorkBuff();
}
void TcpSession::_ForceClose()
{
if (socket_ != -1) {
int oldsocket = socket_;
::close(socket_);
socket_ = -1;
struct epoll_event ev;
epoll_ctl(epoll_fd, EPOLL_CTL_DEL, oldsocket, &ev);
master->FreeClientWithNoLock(this);
}
}
void TcpSession::OnSocketRead(char* buf, unsigned int buflen)
{
unsigned int already_read_bytes = 0;
do {
if (already_read_bytes < buflen) {
int read_bytes = std::min(buflen - already_read_bytes,
(unsigned int)max_packet_len_ - recv_bufflen_);
if (read_bytes > 0) {
memmove(&recv_buff_[recv_bufflen_], buf + already_read_bytes, read_bytes);
recv_bufflen_ += read_bytes;
already_read_bytes += read_bytes;
}
}
int offset = 0;
int prev_offset = 0;
do {
prev_offset = offset;
DecodePacket(recv_buff_, offset, recv_bufflen_);
} while (prev_offset < offset && offset < recv_bufflen_);
if (offset > 0 && offset < recv_bufflen_){
memmove(recv_buff_, recv_buff_ + offset, recv_bufflen_ - offset);
}
recv_bufflen_ -= offset;
if (recv_bufflen_ >= max_packet_len_) {
//收到超长包
Close();
return;
}
} while (already_read_bytes < buflen);
}
void TcpSession::DoClientRecv()
{
if (socket_ == -1) {
return;
}
char recvbuf[DEFAULT_MAX_RECV_BUFFERSIZE];
while (true) {
#ifdef A8_PERF
a8::tick_t begin_tick = a8::XGetTickCount();
#endif
int ret = ::recv(socket_, recvbuf, DEFAULT_MAX_RECV_BUFFERSIZE, 0);
#ifdef A8_PERF
a8::tick_t end_tick = a8::XGetTickCount();
if (end_tick - begin_tick > a8::PerfMonitor::Instance()->max_recv_time) {
a8::PerfMonitor::Instance()->max_recv_time = end_tick - begin_tick;
}
#endif
if (ret < 0) {
if (errno != EAGAIN) {
Close();
return;
} else {
#ifdef A8_PERF
++a8::PerfMonitor::Instance()->recv_eagain_times;
#endif
}
break;
} else if (ret == 0) {
Close();
return;
} else {
OnSocketRead(recvbuf, ret);
if (ret < DEFAULT_MAX_RECV_BUFFERSIZE) {
break;
}
}
}
}
void TcpSession::DoClientSend()
{
if(socket_ == -1){
return;
}
++epoll_out_times;
if (!work_node_) {
send_buffer_mutex_.lock();
work_node_ = top_node_;
top_node_ = nullptr;
bot_node_ = nullptr;
send_buffer_mutex_.unlock();
}
if (work_node_) {
AsyncSend();
}
}
void TcpSession::Close()
{
if (socket_ != -1){
master->LockClients();
_ForceClose();
master->UnLockClients();
}
OnDisConnect();
}
void TcpSession::ClearSendBuff()
{
a8::SendQueueNode* p_top_node_ = nullptr;
send_buffer_mutex_.lock();
p_top_node_ = top_node_;
top_node_ = nullptr;
bot_node_ = nullptr;
send_buffer_mutex_.unlock();
a8::SendQueueNode *pdelnode = nullptr;
while (p_top_node_) {
pdelnode = p_top_node_;
p_top_node_ = p_top_node_->next;
if (pdelnode->buff) {
free(pdelnode->buff);
}
free(pdelnode);
--master->send_node_num;
}
}
void TcpSession::ClearWorkBuff()
{
while (work_node_) {
a8::SendQueueNode *pdelnode = work_node_;
work_node_ = work_node_->next;
if (pdelnode->buff) {
free(pdelnode->buff);
}
free(pdelnode);
--master->send_node_num;
}
}
void TcpSession::AsyncSend()
{
while (work_node_) {
int sentbytes = ::send(socket_,
work_node_->buff + work_node_->sent_bytes,
work_node_->bufflen - work_node_->sent_bytes,
0);
if (sentbytes <= 0) {
auto err_code = errno;
if (err_code == EAGAIN || err_code == EWOULDBLOCK) {
break;
} else {
Close();
break;
}
}
work_node_->sent_bytes += sentbytes;
if (work_node_->sent_bytes >= work_node_->bufflen) {
a8::SendQueueNode *pdelnode = work_node_;
work_node_ = work_node_->next;
--master->send_node_num;
++master->sent_node_num;
master->sent_bytes_num += pdelnode->sent_bytes;
if (!work_node_) {
send_buffer_mutex_.lock();
if (top_node_) {
work_node_ = top_node_;
top_node_ = nullptr;
bot_node_ = nullptr;
}
if (!work_node_) {
sending_ = false;
struct epoll_event ev;
ev.data.fd = socket_;
ev.events = EPOLLIN | EPOLLRDHUP;
ev.data.ptr = this;
::epoll_ctl(epoll_fd, EPOLL_CTL_MOD, socket_, &ev);
}
send_buffer_mutex_.unlock();
}
if (pdelnode->buff) {
free(pdelnode->buff);
}
free(pdelnode);
}
}
}
void TcpSession::NotifyEpollSend()
{
sending_ = true;
struct epoll_event ev;
ev.data.fd = socket_;
ev.events = EPOLLIN | EPOLLOUT | EPOLLRDHUP;
ev.data.ptr = this;
::epoll_ctl(epoll_fd, EPOLL_CTL_MOD, socket_, &ev);
}
int TcpSession::DirectSend(a8::SendQueueNode* p)
{
++direct_send_times;
while (true) {
int sentbytes = ::send(socket_,
p->buff + p->sent_bytes,
p->bufflen - p->sent_bytes,
0);
if (sentbytes <= 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
return -1;
} else {
Close();
return -2;
}
} else {
p->sent_bytes += sentbytes;
if (p->sent_bytes >= p->bufflen) {
return 0;
}
}
}
}
}
#endif

81
a8/tcpsession2.h
View File

@ -1,81 +0,0 @@
#ifndef A8_TCPSESSION2_H
#define A8_TCPSESSION2_H
#ifdef A8_TCP_SESSION2
#include <mutex>
namespace a8
{
struct SendQueueNode;
class TcpListener;
class TcpSession
{
public:
bool is_activite = false;
time_t create_time = 0;
a8::TcpListener* master = nullptr;
unsigned long saddr = 0;
std::string remote_address;
int remote_port = 0;
int socket_handle = -1;
int epoll_fd = 0;
list_head session_entry;
public:
TcpSession();
virtual ~TcpSession();
void SetMaxPacketLen(int max_packet_len);
int Socket();
virtual void OnError(int);
virtual void OnConnect();
virtual void OnDisConnect();
virtual void Destory();
bool Alive();
protected:
virtual void SendBuff(const char* buff, unsigned int bufflen);
void SendText(const std::string& text);
void SetSocket(int sock);
bool AllocRecvBuf();
virtual void Reset();
void _ForceClose();
virtual void OnSocketRead(char* buf, unsigned int buflen);
virtual void DecodePacket(char* buf, int& offset, unsigned int buflen);
virtual void DecodeUserPacket(char* buf, int& offset, unsigned int buflen) = 0;
void DoClientRecv();
void DoClientSend();
void Close();
private:
void ClearSendBuff();
void ClearWorkBuff();
void AsyncSend();
void NotifyEpollSend();
int DirectSend(a8::SendQueueNode* node);
protected:
char *recv_buff_ = nullptr;
int recv_bufflen_ = 0;
int max_packet_len_ = 0;
volatile long long epoll_out_times = 0;
volatile long long epoll_in_times = 0;
volatile long long direct_send_times = 0;
private:
int socket_ = 0;
volatile bool sending_ = false;
a8::SendQueueNode* top_node_ = nullptr;
a8::SendQueueNode* bot_node_ = nullptr;
a8::SendQueueNode* work_node_ = nullptr;
std::mutex send_buffer_mutex_;
friend class TcpListener;
friend class TcpListenerImpl;
};
}
#endif
#endif

4
a8/tcpsessionpool.cc
View File

@ -1,11 +1,7 @@
#include <time.h>
#include <a8/a8.h>
#ifdef A8_TCP_SESSION2
#include <a8/tcpsession2.h>
#else
#include <a8/tcpsession.h>
#endif
#include <a8/tcpsessionpool.h>
namespace a8