Skip to content

Commit 1a3a57c

Browse files
kzhdevkzhdev
committed
Shutdown server 10 sec after last client unregistered; Stop client worker thread after closing websocket
1 parent fadea7f commit 1a3a57c

File tree

3 files changed

+67
-52
lines changed

3 files changed

+67
-52
lines changed

‎zorro_websocket_proxy/src/main.cpp‎

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -38,33 +38,30 @@ void sigHandler(int sig) {
3838

3939
/**
4040
* Usage:
41-
* ZorroWebsocketsProxy.exe [-p <port>] [-s <ws_queue_size>] [-l <logging_level>]
41+
* ZorroWebsocketsProxy.exe [-s <server_queue_size>] [-l <logging_level>]
4242
*
4343
* Arguments:
44+
* -s [optional]: Specify server to client queue size in Byte. Default to 16777216 Bytes.
4445
* -l [optional]: Specify logging level. By default, logging will be disabled in release build.
45-
* Valid Logging level: OFF, ERROR, WARNING, INFO, DEBUG
46-
*
46+
* Valid Logging level: OFF, ERROR, WARNING, INFO, DEBUG*
4747
*/
4848
int main(int argc, char* argv[])
4949
{
50-
uint32_t port = 55000;
51-
uint32_t ws_queue_size = 1 << 24; // 16MB
50+
uint32_t server_queue_size = 1 << 24; // 16MB
5251
#ifdef _DEBUG
5352
int32_t level = L_ERR | L_WARN | L_NOTICE | L_USER;
5453
#else
5554
int32_t level = L_ERR | L_WARN | L_USER;
5655
#endif
5756

5857
for (int i = 1; i < argc - 1; ++i) {
59-
if ((strcmp(argv[i], "-g") == 0) || (strcmp(argv[i], "-G") == 0)) {
60-
port = atoi(argv[++i]);
61-
}
62-
else if (strcmp(argv[1], "-l") == 0 || (strcmp(argv[i], "-L") == 0)) {
58+
if (_stricmp(argv[i], "-l") == 0) {
6359
std::string l = argv[++i];
6460
std::transform(l.begin(), l.end(), l.begin(), std::tolower);
6561
if (l == "off") {
6662
level = L_OFF;
67-
} else if (l == "error") {
63+
}
64+
else if (l == "error") {
6865
level = L_ERR;
6966
}
7067
else if (l == "warning") {
@@ -80,15 +77,18 @@ int main(int argc, char* argv[])
8077
level = L_ERR | L_WARN | L_NOTICE | L_USER | L_INFO | L_DEBUG;
8178
}
8279
}
80+
else if (_stricmp(argv[i], "-s") == 0) {
81+
server_queue_size = atoi(argv[++i]);
82+
}
8383
}
8484

8585
signal(SIGINT, sigHandler);
8686

8787
try {
8888
std::cout << "ZorroWebsocketProxy Started." << std::endl;
89-
std::unique_ptr<ZorroWebsocketProxy> proxy = std::make_unique<ZorroWebsocketProxy>(ws_queue_size);
89+
std::unique_ptr<ZorroWebsocketProxy> proxy = std::make_unique<ZorroWebsocketProxy>(server_queue_size, level);
9090
the_proxy = proxy.get();
91-
proxy->run(argv[0], level);
91+
proxy->run();
9292
}
9393
catch (...) {}
9494
std::cout << "ZorroWebsocketProxy Exit." << std::endl;

‎zorro_websocket_proxy/src/zorro_websocket_proxy.cpp‎

Lines changed: 52 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -55,15 +55,40 @@ namespace {
5555
fflush(logfile.file);
5656
}
5757
}
58+
59+
std::string GetExePath()
60+
{
61+
#ifdef _WIN32
62+
wchar_t result[MAX_PATH] = { 0 };
63+
GetModuleFileNameW(NULL, result, MAX_PATH);
64+
std::wstring wsPath(result);
65+
std::string path(wsPath.begin(), wsPath.end());
66+
auto pos = path.rfind("\\");
67+
#else
68+
char result[PATH_MAX];
69+
ssize_t count = readlink("/proc/self/exe", result, PATH_MAX);
70+
std::string path(result, (count > 0) ? count : 0);
71+
auto pos = path.rfind("/");
72+
#endif
73+
return path.substr(0, pos);
74+
}
5875
}
5976

60-
ZorroWebsocketProxy::ZorroWebsocketProxy(uint32_t ws_queue_size)
77+
ZorroWebsocketProxy::ZorroWebsocketProxy(uint32_t server_queue_size, int32_t logLevel)
6178
: client_queue_(1 << 16, CLIENT_TO_SERVER_QUEUE)
62-
, server_queue_(ws_queue_size, SERVER_TO_CLIENT_QUEUE)
79+
, server_queue_(server_queue_size, SERVER_TO_CLIENT_QUEUE)
6380
, client_index_(client_queue_.initial_reading_index())
64-
, pid_(GetCurrentProcessId())
81+
, pid_(GetCurrentProcessId())
82+
, exec_path_(GetExePath())
6583
, closed_sockets_(256) {
66-
84+
85+
if (!logLevel || !logfile.file) {
86+
lws_set_log_level(LLL_ERR, nullptr);
87+
}
88+
else {
89+
lws_set_log_level(logLevel, log);
90+
}
91+
6792
hMapFile_ = CreateFileMapping(
6893
INVALID_HANDLE_VALUE, // use paging file
6994
NULL, // default security
@@ -115,20 +140,7 @@ ZorroWebsocketProxy::~ZorroWebsocketProxy() {
115140
lwsl_user("WebosocketsProsy destroyed\n");
116141
}
117142

118-
void ZorroWebsocketProxy::run(const char* executable_path, int32_t logLevel) {
119-
if (!logLevel || !logfile.file) {
120-
lws_set_log_level(LLL_ERR, nullptr);
121-
}
122-
else {
123-
lws_set_log_level(logLevel, log);
124-
}
125-
126-
std::string path(executable_path);
127-
auto pos = path.rfind("zorro_websocket_proxy.exe");
128-
if (pos != std::string::npos) {
129-
exec_path_ = path.substr(0, pos - 1);
130-
}
131-
143+
void ZorroWebsocketProxy::run() {
132144
if (!own_shm_) {
133145
auto owner = owner_pid_->load(std::memory_order_relaxed);
134146
if (owner) {
@@ -164,7 +176,7 @@ void ZorroWebsocketProxy::run(const char* executable_path, int32_t logLevel) {
164176
std::this_thread::yield();
165177
}
166178

167-
if (shutdown_time_ && (get_timestamp() - shutdown_time_) >= 15000) {
179+
if (shutdown_time_ && (get_timestamp() - shutdown_time_) >= 10000) {
168180
run_.store(false, std::memory_order_release);
169181
break;
170182
}
@@ -216,7 +228,6 @@ void ZorroWebsocketProxy::handleClientRegistration(Message& msg) {
216228
auto reg = reinterpret_cast<RegisterMessage*>(msg.data);
217229
lwsl_user("Register client %d connected, name: %s\n", msg.pid, reg->name);
218230
reg->server_pid = pid_;
219-
clientSeen_ = true;
220231
shutdown_time_ = 0;
221232

222233
auto it = clients_.find(msg.pid);
@@ -250,6 +261,7 @@ void ZorroWebsocketProxy::unregisterClient(uint32_t pid) {
250261

251262
if (clients_.empty()) {
252263
lwsl_user("Last client disconnected. Shuting down...\n");
264+
shutdown_time_ = get_timestamp();
253265
}
254266
}
255267
}
@@ -275,27 +287,31 @@ void ZorroWebsocketProxy::openWs(Message& msg) {
275287
lwsl_user("Websocket %s already opened. id=%d, new=%d\n", req->url, id, req->new_connection);
276288
msg.status.store(Message::Status::SUCCESS, std::memory_order_release);
277289
return;
278-
}
290+
}
279291
}
280292

281-
lwsl_user("Opening ws %s\n", req->url);
282-
req->new_connection = true;
283-
auto websocket = std::make_shared<Websocket>(this, pid_ * 10000 + (++websocket_id_), req->url);
284-
auto b = websocket->open(msg.pid);
285-
if (b) {
286-
req->id = websocket->id();
287-
websocket->clients().emplace(msg.pid);
288-
websocketsByUrl_.emplace(req->url, websocket);
289-
websocketsById_.emplace(websocket->id(), std::move(websocket));
290-
}
291-
msg.status.store(b ? Message::Status::SUCCESS : Message::Status::FAILED, std::memory_order_release);
293+
openNewWs(msg, req);
292294
}
293295
else {
294296
snprintf(req->err, 256, "Client %d not found", msg.pid);
295297
msg.status.store(Message::Status::FAILED, std::memory_order_release);
296298
}
297299
}
298300

301+
void ZorroWebsocketProxy::openNewWs(Message& msg, WsOpen* req) {
302+
lwsl_user("Opening ws %s\n", req->url);
303+
req->new_connection = true;
304+
auto websocket = std::make_shared<Websocket>(this, pid_ * 10000 + (++websocket_id_), req->url);
305+
auto b = websocket->open(msg.pid);
306+
if (b) {
307+
req->id = websocket->id();
308+
websocket->clients().emplace(msg.pid);
309+
websocketsByUrl_.emplace(req->url, websocket);
310+
websocketsById_.emplace(websocket->id(), std::move(websocket));
311+
}
312+
msg.status.store(b ? Message::Status::SUCCESS : Message::Status::FAILED, std::memory_order_release);
313+
}
314+
299315
void ZorroWebsocketProxy::closeWs(Message& msg) {
300316
auto req = reinterpret_cast<WsClose*>(msg.data);
301317
auto client = getClient(msg.pid);
@@ -316,6 +332,7 @@ void ZorroWebsocketProxy::closeWs(uint32_t id, DWORD pid) {
316332
lwsl_user("WS client %d removed from ws id=%d\n", pid, id);
317333
if (websocket->clients_.empty()) {
318334
lwsl_user("Close ws %s\n", it->second->url().c_str());
335+
it->second->status_ = Websocket::Status::DISCONNECTING;
319336
it->second->stop();
320337
websocketsByUrl_.erase(it->second->url());
321338
websocketsById_.erase(it);
@@ -363,10 +380,6 @@ void ZorroWebsocketProxy::sendMessage(uint64_t index, uint32_t size, uint64_t no
363380

364381
bool ZorroWebsocketProxy::checkHeartbeats() {
365382
if (clients_.empty()) {
366-
if (clientSeen_ && !isProcessRunning(L"Zorro.exe") && !shutdown_time_) {
367-
lwsl_user("No Zorro running. Shuting down...\n");
368-
shutdown_time_ = get_timestamp();
369-
}
370383
return false;
371384
}
372385
auto now = get_timestamp();
@@ -381,6 +394,7 @@ bool ZorroWebsocketProxy::checkHeartbeats() {
381394

382395
if (clients_.empty()) {
383396
lwsl_user("Last client disconnected. Stop heartbeating...\n");
397+
shutdown_time_ = get_timestamp();
384398
}
385399
}
386400
else {
@@ -428,7 +442,6 @@ void ZorroWebsocketProxy::onWsOpened(uint32_t id, DWORD initiator) {
428442
open->initiator = initiator;
429443
open->new_connection = true;
430444
sendMessage(index, size);
431-
lwsl_user("send ws opened to client\n");
432445
}
433446

434447
void ZorroWebsocketProxy::onWsClosed(uint32_t id) {
@@ -441,6 +454,8 @@ void ZorroWebsocketProxy::onWsClosed(uint32_t id) {
441454
auto idx = closed_sockets_.reserve();
442455
(*closed_sockets_[idx]) = id;
443456
closed_sockets_.publish(idx);
457+
458+
lwsl_user("ws %d closed\n", id);
444459
}
445460

446461
void ZorroWebsocketProxy::onWsError(uint32_t id, const char* err, size_t len) {

‎zorro_websocket_proxy/src/zorro_websocket_proxy.h‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,6 @@ namespace websocket {
2626
HANDLE hMapFile_ = nullptr;
2727
LPVOID lpvMem_ = nullptr;
2828
bool own_shm_ = false;
29-
bool clientSeen_ = false;
3029
std::atomic<DWORD>* owner_pid_ = nullptr;
3130
std::string exec_path_;
3231

@@ -41,10 +40,10 @@ namespace websocket {
4140
uint64_t closed_sockets_index_ = 0;
4241

4342
public:
44-
ZorroWebsocketProxy(uint32_t ws_queue_size);
43+
ZorroWebsocketProxy(uint32_t server_queue_size, int32_t logLevel);
4544
~ZorroWebsocketProxy();
4645

47-
void run(const char* executable_path, int32_t logLevel);
46+
void run();
4847
void stop() noexcept { run_.store(false, std::memory_order_release); }
4948

5049
private:
@@ -55,6 +54,7 @@ namespace websocket {
5554
void unregisterClient(uint32_t pid);
5655
void handleClientHeartbeat(Message& msg);
5756
void openWs(Message& msg);
57+
void openNewWs(Message& msg, WsOpen* req);
5858
void closeWs(Message& msg);
5959
void closeWs(uint32_t id, DWORD pid);
6060
void sendWsRequest(Message& msg);

0 commit comments

Comments
 (0)