Skip to content

Commit 6b52898

Browse files
IronsDuIronsDu
authored
Merge pull request #77 from IronsDu/feature/highwater
Feature/highwater
2 parents c4c1d63 + f487007 commit 6b52898

5 files changed

Lines changed: 88 additions & 49 deletions

File tree

‎examples/BenchWebsocket.cpp‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -28,14 +28,14 @@ static void sendPacket(const HttpSession::Ptr& session, const char* data, size_t
2828
bw.writeINT8('"');
2929
bw.writeINT8('}');
3030

31-
auto frame = std::make_shared<std::string>();
31+
std::string frame;
3232
WebSocketFormat::wsFrameBuild(bw.getData(),
3333
bw.getPos(),
34-
*frame,
34+
frame,
3535
WebSocketFormat::WebSocketFrameType::TEXT_FRAME,
3636
true,
3737
true);
38-
session->send(frame);
38+
session->send(std::move(frame));
3939
}
4040

4141
int main(int argc, char **argv)

‎examples/BroadCastServer.cpp‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,7 @@ static size_t getClientNum()
4949
return clients.size();
5050
}
5151

52-
static void broadCastPacket(const std::shared_ptr<std::string>& packet)
52+
static void broadCastPacket(const brynet::net::SendableMsg::Ptr & packet)
5353
{
5454
auto packetLen = packet->size();
5555
RecvPacketNum++;
@@ -94,6 +94,12 @@ int main(int argc, char** argv)
9494
});
9595
});
9696

97+
session->setHighWaterCallback([]()
98+
{
99+
std::cout << "high water" << std::endl;
100+
},
101+
1024*1024*100);
102+
97103
session->setDataCallback([mainLoop](brynet::base::BasePacketReader& reader) {
98104
while (true)
99105
{
@@ -109,7 +115,7 @@ int main(int argc, char** argv)
109115
break;
110116
}
111117

112-
auto packet = std::make_shared<std::string>(buffer, packetLen);
118+
auto packet = brynet::net::MakeStringMsg(buffer, packetLen);
113119
mainLoop->runAsyncFunctor([packet]()
114120
{
115121
broadCastPacket(packet);

‎examples/HttpServer.cpp‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -53,14 +53,14 @@ int main(int argc, char **argv)
5353
std::cout << "frame enter of type:" << int(opcode) << std::endl;
5454
std::cout << "payload is:" << payload << std::endl;
5555
// echo frame
56-
auto frame = std::make_shared<std::string>();
56+
std::string frame;
5757
WebSocketFormat::wsFrameBuild(payload.c_str(),
5858
payload.size(),
59-
*frame,
59+
frame,
6060
WebSocketFormat::WebSocketFrameType::TEXT_FRAME,
6161
true,
6262
false);
63-
httpSession->send(frame);
63+
httpSession->send(std::move(frame));
6464
};
6565

6666
wrapper::HttpListenerBuilder listenBuilder;

‎include/brynet/net/TcpConnection.hpp‎

Lines changed: 73 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,42 @@ namespace brynet { namespace net {
4747
virtual size_t size() = 0;
4848
};
4949

50+
class StringSendMsg : public SendableMsg
51+
{
52+
public:
53+
explicit StringSendMsg(const char* buffer, size_t len)
54+
:
55+
mMsg(buffer, len)
56+
{}
57+
58+
explicit StringSendMsg(std::string buffer)
59+
:
60+
mMsg(std::move(buffer))
61+
{}
62+
63+
const void* data() override
64+
{
65+
return static_cast<const void*>(mMsg.data());
66+
}
67+
size_t size() override
68+
{
69+
return mMsg.size();
70+
}
71+
72+
private:
73+
std::string mMsg;
74+
};
75+
76+
static SendableMsg::Ptr MakeStringMsg(const char* buffer, size_t len)
77+
{
78+
return std::make_shared<StringSendMsg>(buffer, len);
79+
}
80+
81+
static SendableMsg::Ptr MakeStringMsg(std::string buffer)
82+
{
83+
return std::make_shared<StringSendMsg>(std::move(buffer));
84+
}
85+
5086
class TcpConnection : public Channel,
5187
public brynet::base::NonCopyable,
5288
public std::enable_shared_from_this<TcpConnection>
@@ -58,6 +94,7 @@ namespace brynet { namespace net {
5894
using DataCallback = std::function<void(brynet::base::BasePacketReader&)>;
5995
using DisconnectedCallback = std::function<void(Ptr)>;
6096
using PacketSendedCallback = std::function<void(void)>;
97+
using HighWaterCallback = std::function<void(void)>;
6198

6299
public:
63100
Ptr static Create(TcpSocket::Ptr socket,
@@ -135,12 +172,24 @@ namespace brynet { namespace net {
135172
auto sharedThis = shared_from_this();
136173
mEventLoop->runAsyncFunctor([sharedThis, msg, callback]() mutable
137174
{
175+
if(sharedThis->mAlreadyClose)
176+
{
177+
return;
178+
}
179+
138180
const auto len = msg->size();
181+
sharedThis->mSendingMsgSize += len;
139182
sharedThis->mSendList.emplace_back(PendingPacket{
140183
std::move(msg),
141184
len,
142185
std::move(callback) });
143186
sharedThis->runAfterFlush();
187+
188+
if(sharedThis->mSendingMsgSize > sharedThis->mHighWaterSize &&
189+
sharedThis->mHighWaterCallback != nullptr)
190+
{
191+
sharedThis->mHighWaterCallback();
192+
}
144193
});
145194
}
146195

@@ -149,49 +198,14 @@ namespace brynet { namespace net {
149198
size_t len,
150199
PacketSendedCallback&& callback = nullptr)
151200
{
152-
send(std::make_shared<std::string>(buffer, len), std::move(callback));
201+
send(MakeStringMsg(buffer, len), std::move(callback));
153202
}
154203

155204
void send(
156-
const std::shared_ptr<std::string>& packet,
205+
std::string buffer,
157206
PacketSendedCallback&& callback = nullptr)
158207
{
159-
class StringSendMsg : public SendableMsg
160-
{
161-
public:
162-
explicit StringSendMsg(std::shared_ptr<std::string>&& msg)
163-
:
164-
mMsg(std::move(msg))
165-
{}
166-
explicit StringSendMsg(const std::shared_ptr<std::string>& msg)
167-
:
168-
mMsg(msg)
169-
{}
170-
171-
const void* data() override
172-
{
173-
return static_cast<const void*>(mMsg->data());
174-
}
175-
176-
size_t size() override
177-
{
178-
return mMsg->size();
179-
}
180-
181-
private:
182-
std::shared_ptr<std::string> mMsg;
183-
};
184-
185-
auto sharedThis = shared_from_this();
186-
mEventLoop->runAsyncFunctor([sharedThis, packet, callback]() mutable
187-
{
188-
const auto len = packet->size();
189-
sharedThis->mSendList.emplace_back(PendingPacket{
190-
std::make_shared<StringSendMsg>(std::move(packet)),
191-
len,
192-
std::move(callback) });
193-
sharedThis->runAfterFlush();
194-
});
208+
send(MakeStringMsg(std::move(buffer)), std::move(callback));
195209
}
196210

197211
// setDataCallback(std::function<void(brynet::base::BasePacketReader&)>）
@@ -234,6 +248,16 @@ namespace brynet { namespace net {
234248
});
235249
}
236250

251+
void setHighWaterCallback(HighWaterCallback cb, size_t size)
252+
{
253+
auto sharedThis = shared_from_this();
254+
mEventLoop->runAsyncFunctor([=]() mutable
255+
{
256+
sharedThis->mHighWaterCallback = std::move(cb);
257+
sharedThis->mHighWaterSize = size;
258+
});
259+
}
260+
237261
void postShrinkReceiveBuffer()
238262
{
239263
auto sharedThis = shared_from_this();
@@ -288,7 +312,9 @@ namespace brynet { namespace net {
288312
mEventLoop(std::move(eventLoop)),
289313
mAlreadyClose(false),
290314
mMaxRecvBufferSize(maxRecvBufferSize),
291-
mEnterCallback(std::move(enterCallback))
315+
mSendingMsgSize(0),
316+
mEnterCallback(std::move(enterCallback)),
317+
mHighWaterSize(0)
292318
{
293319
mRecvData = false;
294320
mCheckTime = std::chrono::steady_clock::duration::zero();
@@ -850,6 +876,7 @@ namespace brynet { namespace net {
850876
{
851877
(packet.mCompleteCallback)();
852878
}
879+
mSendingMsgSize -= packet.data->size();
853880
it = mSendList.erase(it);
854881
}
855882

@@ -928,6 +955,7 @@ namespace brynet { namespace net {
928955
{
929956
b.mCompleteCallback();
930957
}
958+
mSendingMsgSize -= b.data->size();
931959
it = mSendList.erase(it);
932960
}
933961

@@ -980,9 +1008,11 @@ namespace brynet { namespace net {
9801008
});
9811009

9821010
mCanWrite = false;
983-
mDisConnectCallback = nullptr;
9841011
mDataCallback = nullptr;
1012+
mDisConnectCallback = nullptr;
1013+
mHighWaterCallback = nullptr;
9851014
mRecvBuffer = nullptr;
1015+
mSendList.clear();
9861016
}
9871017

9881018
void procCloseInLoop()
@@ -1212,10 +1242,13 @@ namespace brynet { namespace net {
12121242

12131243
using PacketListType = std::deque<PendingPacket>;
12141244
PacketListType mSendList;
1245+
size_t mSendingMsgSize;
12151246

12161247
EnterCallback mEnterCallback;
12171248
DataCallback mDataCallback;
12181249
DisconnectedCallback mDisConnectCallback;
1250+
HighWaterCallback mHighWaterCallback;
1251+
size_t mHighWaterSize;
12191252

12201253
bool mIsPostFlush;
12211254

‎include/brynet/net/http/HttpService.hpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,7 @@ namespace brynet { namespace net { namespace http {
3232
void send(PacketType&& packet,
3333
TcpConnection::PacketSendedCallback&& callback = nullptr)
3434
{
35-
mSession->send(std::forward<std::shared_ptr<std::string>>(packet),
35+
mSession->send(std::forward<PacketType&&>(packet),
3636
std::move(callback));
3737
}
3838
void send(const char* packet,

0 commit comments

Comments
 (0)