fix "cannot resume dead coroutine" for servers
This commit is contained in:
parent
8489c36df7
commit
1bd3463488
@ -60,8 +60,37 @@ local ServerSocket = {__index={
|
|||||||
get_port=function(self) return network.__get_serverport(self.id) end,
|
get_port=function(self) return network.__get_serverport(self.id) end,
|
||||||
}}
|
}}
|
||||||
|
|
||||||
|
|
||||||
|
local _tcp_server_callbacks = {}
|
||||||
|
|
||||||
network.tcp_open = function (port, handler)
|
network.tcp_open = function (port, handler)
|
||||||
return setmetatable({id=network.__open(port, function(id)
|
local socket = setmetatable({id=network.__open(port)}, ServerSocket)
|
||||||
|
|
||||||
|
_tcp_server_callbacks[socket.id] = function(id)
|
||||||
handler(setmetatable({id=id}, Socket))
|
handler(setmetatable({id=id}, Socket))
|
||||||
end)}, ServerSocket)
|
end
|
||||||
|
return socket
|
||||||
|
end
|
||||||
|
|
||||||
|
network.__pull_events = function()
|
||||||
|
local cleaned = false
|
||||||
|
local connections = network.__pull_connections()
|
||||||
|
for i, pair in ipairs(connections) do
|
||||||
|
local sid, cid = unpack(pair)
|
||||||
|
local callback = _tcp_server_callbacks[sid]
|
||||||
|
|
||||||
|
if callback then
|
||||||
|
callback(cid)
|
||||||
|
end
|
||||||
|
|
||||||
|
-- remove dead servers
|
||||||
|
if not cleaned then
|
||||||
|
for sid, callback in pairs(_tcp_server_callbacks) do
|
||||||
|
if not network.__is_serveropen(sid) then
|
||||||
|
_tcp_server_callbacks[sid] = nil
|
||||||
|
end
|
||||||
|
end
|
||||||
|
cleaned = true
|
||||||
|
end
|
||||||
|
end
|
||||||
end
|
end
|
||||||
|
|||||||
@ -34,7 +34,10 @@ local function complete_app_lib(app)
|
|||||||
app.reconfig_packs = core.reconfig_packs
|
app.reconfig_packs = core.reconfig_packs
|
||||||
app.get_setting = core.get_setting
|
app.get_setting = core.get_setting
|
||||||
app.set_setting = core.set_setting
|
app.set_setting = core.set_setting
|
||||||
app.tick = coroutine.yield
|
app.tick = function()
|
||||||
|
coroutine.yield()
|
||||||
|
network.__pull_events()
|
||||||
|
end
|
||||||
app.get_version = core.get_version
|
app.get_version = core.get_version
|
||||||
app.get_setting_info = core.get_setting_info
|
app.get_setting_info = core.get_setting_info
|
||||||
app.load_content = function()
|
app.load_content = function()
|
||||||
|
|||||||
@ -44,11 +44,11 @@ public:
|
|||||||
};
|
};
|
||||||
|
|
||||||
struct CoreParameters {
|
struct CoreParameters {
|
||||||
bool headless = false;
|
bool headless = true;
|
||||||
bool testMode = false;
|
bool testMode = false;
|
||||||
std::filesystem::path resFolder = "res";
|
std::filesystem::path resFolder = "res";
|
||||||
std::filesystem::path userFolder = ".";
|
std::filesystem::path userFolder = ".";
|
||||||
std::filesystem::path scriptFile;
|
std::filesystem::path scriptFile = "res/content/remp/remp_server.lua";
|
||||||
std::filesystem::path projectFolder;
|
std::filesystem::path projectFolder;
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|||||||
@ -182,14 +182,17 @@ static int l_available(lua::State* L, network::Network& network) {
|
|||||||
return 0;
|
return 0;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
struct ConnectionEvent {
|
||||||
|
u64id_t server;
|
||||||
|
u64id_t client;
|
||||||
|
};
|
||||||
|
|
||||||
|
static std::vector<ConnectionEvent> clients_queue {};
|
||||||
|
|
||||||
static int l_open(lua::State* L, network::Network& network) {
|
static int l_open(lua::State* L, network::Network& network) {
|
||||||
int port = lua::tointeger(L, 1);
|
int port = lua::tointeger(L, 1);
|
||||||
lua::pushvalue(L, 2);
|
u64id_t id = network.openServer(port, [](u64id_t sid, u64id_t id) {
|
||||||
auto callback = lua::create_lambda_nothrow(L);
|
clients_queue.push_back({sid, id});
|
||||||
u64id_t id = network.openServer(port, [callback](u64id_t id) {
|
|
||||||
engine->postRunnable([=]() {
|
|
||||||
callback({id});
|
|
||||||
});
|
|
||||||
});
|
});
|
||||||
return lua::pushinteger(L, id);
|
return lua::pushinteger(L, id);
|
||||||
}
|
}
|
||||||
@ -250,6 +253,23 @@ static int l_get_total_download(lua::State* L, network::Network& network) {
|
|||||||
return lua::pushinteger(L, network.getTotalDownload());
|
return lua::pushinteger(L, network.getTotalDownload());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
static int l_pull_connections(lua::State* L, network::Network& network) {
|
||||||
|
lua::createtable(L, clients_queue.size(), 0);
|
||||||
|
for (size_t i = 0; i < clients_queue.size(); i++) {
|
||||||
|
lua::createtable(L, 2, 0);
|
||||||
|
|
||||||
|
lua::pushinteger(L, clients_queue[i].server);
|
||||||
|
lua::rawseti(L, 1);
|
||||||
|
|
||||||
|
lua::pushinteger(L, clients_queue[i].client);
|
||||||
|
lua::rawseti(L, 2);
|
||||||
|
|
||||||
|
lua::rawseti(L, i + 1);
|
||||||
|
}
|
||||||
|
clients_queue.clear();
|
||||||
|
return 1;
|
||||||
|
}
|
||||||
|
|
||||||
template <int(*func)(lua::State*, network::Network&)>
|
template <int(*func)(lua::State*, network::Network&)>
|
||||||
int wrap(lua_State* L) {
|
int wrap(lua_State* L) {
|
||||||
int result = 0;
|
int result = 0;
|
||||||
@ -267,13 +287,13 @@ int wrap(lua_State* L) {
|
|||||||
return result;
|
return result;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
const luaL_Reg networklib[] = {
|
const luaL_Reg networklib[] = {
|
||||||
{"get", wrap<l_get>},
|
{"get", wrap<l_get>},
|
||||||
{"get_binary", wrap<l_get_binary>},
|
{"get_binary", wrap<l_get_binary>},
|
||||||
{"post", wrap<l_post>},
|
{"post", wrap<l_post>},
|
||||||
{"get_total_upload", wrap<l_get_total_upload>},
|
{"get_total_upload", wrap<l_get_total_upload>},
|
||||||
{"get_total_download", wrap<l_get_total_download>},
|
{"get_total_download", wrap<l_get_total_download>},
|
||||||
|
{"__pull_connections", wrap<l_pull_connections>},
|
||||||
{"__open", wrap<l_open>},
|
{"__open", wrap<l_open>},
|
||||||
{"__closeserver", wrap<l_closeserver>},
|
{"__closeserver", wrap<l_closeserver>},
|
||||||
{"__connect", wrap<l_connect>},
|
{"__connect", wrap<l_connect>},
|
||||||
|
|||||||
@ -90,7 +90,7 @@ public:
|
|||||||
}
|
}
|
||||||
|
|
||||||
void update() override {
|
void update() override {
|
||||||
if (id == 0) {
|
if (!alive || id == 0) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
if (lua::requireglobal(L, "__vc_resume_coroutine")) {
|
if (lua::requireglobal(L, "__vc_resume_coroutine")) {
|
||||||
|
|||||||
@ -477,6 +477,7 @@ public:
|
|||||||
};
|
};
|
||||||
|
|
||||||
class SocketTcpSServer : public TcpServer {
|
class SocketTcpSServer : public TcpServer {
|
||||||
|
u64id_t id;
|
||||||
Network* network;
|
Network* network;
|
||||||
SOCKET descriptor;
|
SOCKET descriptor;
|
||||||
std::vector<u64id_t> clients;
|
std::vector<u64id_t> clients;
|
||||||
@ -485,14 +486,14 @@ class SocketTcpSServer : public TcpServer {
|
|||||||
std::unique_ptr<std::thread> thread = nullptr;
|
std::unique_ptr<std::thread> thread = nullptr;
|
||||||
int port;
|
int port;
|
||||||
public:
|
public:
|
||||||
SocketTcpSServer(Network* network, SOCKET descriptor, int port)
|
SocketTcpSServer(u64id_t id, Network* network, SOCKET descriptor, int port)
|
||||||
: network(network), descriptor(descriptor), port(port) {}
|
: id(id), network(network), descriptor(descriptor), port(port) {}
|
||||||
|
|
||||||
~SocketTcpSServer() {
|
~SocketTcpSServer() {
|
||||||
closeSocket();
|
closeSocket();
|
||||||
}
|
}
|
||||||
|
|
||||||
void startListen(consumer<u64id_t> handler) override {
|
void startListen(ConnectCallback handler) override {
|
||||||
thread = std::make_unique<std::thread>([this, handler]() {
|
thread = std::make_unique<std::thread>([this, handler]() {
|
||||||
while (open) {
|
while (open) {
|
||||||
logger.info() << "listening for connections";
|
logger.info() << "listening for connections";
|
||||||
@ -518,7 +519,7 @@ public:
|
|||||||
std::lock_guard lock(clientsMutex);
|
std::lock_guard lock(clientsMutex);
|
||||||
clients.push_back(id);
|
clients.push_back(id);
|
||||||
}
|
}
|
||||||
handler(id);
|
handler(this->id, id);
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
@ -558,7 +559,7 @@ public:
|
|||||||
}
|
}
|
||||||
|
|
||||||
static std::shared_ptr<SocketTcpSServer> openServer(
|
static std::shared_ptr<SocketTcpSServer> openServer(
|
||||||
Network* network, int port, consumer<u64id_t> handler
|
u64id_t id, Network* network, int port, ConnectCallback handler
|
||||||
) {
|
) {
|
||||||
SOCKET descriptor = socket(
|
SOCKET descriptor = socket(
|
||||||
AF_INET, SOCK_STREAM, 0
|
AF_INET, SOCK_STREAM, 0
|
||||||
@ -585,7 +586,7 @@ public:
|
|||||||
}
|
}
|
||||||
logger.info() << "opened server at port " << port;
|
logger.info() << "opened server at port " << port;
|
||||||
auto server =
|
auto server =
|
||||||
std::make_shared<SocketTcpSServer>(network, descriptor, port);
|
std::make_shared<SocketTcpSServer>(id, network, descriptor, port);
|
||||||
server->startListen(std::move(handler));
|
server->startListen(std::move(handler));
|
||||||
return server;
|
return server;
|
||||||
}
|
}
|
||||||
@ -645,9 +646,9 @@ u64id_t Network::connect(const std::string& address, int port, consumer<u64id_t>
|
|||||||
return id;
|
return id;
|
||||||
}
|
}
|
||||||
|
|
||||||
u64id_t Network::openServer(int port, consumer<u64id_t> handler) {
|
u64id_t Network::openServer(int port, ConnectCallback handler) {
|
||||||
u64id_t id = nextServer++;
|
u64id_t id = nextServer++;
|
||||||
auto server = SocketTcpSServer::openServer(this, port, handler);
|
auto server = SocketTcpSServer::openServer(id, this, port, handler);
|
||||||
servers[id] = std::move(server);
|
servers[id] = std::move(server);
|
||||||
return id;
|
return id;
|
||||||
}
|
}
|
||||||
|
|||||||
@ -12,6 +12,7 @@
|
|||||||
namespace network {
|
namespace network {
|
||||||
using OnResponse = std::function<void(std::vector<char>)>;
|
using OnResponse = std::function<void(std::vector<char>)>;
|
||||||
using OnReject = std::function<void(int)>;
|
using OnReject = std::function<void(int)>;
|
||||||
|
using ConnectCallback = std::function<void(u64id_t, u64id_t)>;
|
||||||
|
|
||||||
class Requests {
|
class Requests {
|
||||||
public:
|
public:
|
||||||
@ -64,7 +65,7 @@ namespace network {
|
|||||||
class TcpServer {
|
class TcpServer {
|
||||||
public:
|
public:
|
||||||
virtual ~TcpServer() {}
|
virtual ~TcpServer() {}
|
||||||
virtual void startListen(consumer<u64id_t> handler) = 0;
|
virtual void startListen(ConnectCallback handler) = 0;
|
||||||
virtual void close() = 0;
|
virtual void close() = 0;
|
||||||
virtual bool isOpen() = 0;
|
virtual bool isOpen() = 0;
|
||||||
virtual int getPort() const = 0;
|
virtual int getPort() const = 0;
|
||||||
@ -106,7 +107,7 @@ namespace network {
|
|||||||
|
|
||||||
u64id_t connect(const std::string& address, int port, consumer<u64id_t> callback);
|
u64id_t connect(const std::string& address, int port, consumer<u64id_t> callback);
|
||||||
|
|
||||||
u64id_t openServer(int port, consumer<u64id_t> handler);
|
u64id_t openServer(int port, ConnectCallback handler);
|
||||||
|
|
||||||
u64id_t addConnection(const std::shared_ptr<Connection>& connection);
|
u64id_t addConnection(const std::shared_ptr<Connection>& connection);
|
||||||
|
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user