andrey
49dc3cefe7
While the peer does not drain its receive buffer, every forwarded packet would re-stall the reader thread for up to writeTimeout(), so a queued Disconnect frame could wait behind the whole backlog. Remember a stalled target (send slower than 2 s) and drop data for a short cooldown instead of stalling; keep dispatcher writes bounded at 3 s. Control frames (Connected/Disconnected) are unaffected.
115 lines
2.8 KiB
C++
115 lines
2.8 KiB
C++
#include "cloudserver.h"
|
|
|
|
CloudServer::CloudServer(DispatcherClient * c, const PIByteArray & sname): server(c) {
|
|
setName(sname.toHex());
|
|
server_uuid = sname;
|
|
connects << CONNECTL(c, dataReadedServer, ([this](uint id, PIByteArray & ba) {
|
|
// keep the lookup and the send under the lock: the
|
|
// sub-client may be deleted by the drainer concurrently
|
|
mutex_clients.lock();
|
|
last_ping.reset();
|
|
DispatcherClient * cl = index_clients.value(id, nullptr);
|
|
if (cl) {
|
|
// a healthy forward is fast: a slow one means
|
|
// the peer is not draining, remember it and
|
|
// drop further data instead of stalling
|
|
PITimeMeasurer tm;
|
|
cl->sendData(ba);
|
|
if (tm.elapsed_s() > 2.0) {
|
|
cl->markWriteStalled();
|
|
piCoutObj << "write stalled to client" << id;
|
|
}
|
|
}
|
|
mutex_clients.unlock();
|
|
}));
|
|
connects << CONNECTL(c, pingReceived, [this]() {
|
|
mutex_clients.lock();
|
|
last_ping.reset();
|
|
mutex_clients.unlock();
|
|
});
|
|
{
|
|
PIMutexLocker locker(mutex_clients);
|
|
last_ping.reset();
|
|
}
|
|
}
|
|
|
|
|
|
CloudServer::~CloudServer() {}
|
|
|
|
|
|
PIByteArray CloudServer::serverUUID() const {
|
|
return server_uuid;
|
|
}
|
|
|
|
|
|
void CloudServer::addClient(DispatcherClient * c) {
|
|
mutex_clients.lock();
|
|
last_ping.reset();
|
|
clients << c;
|
|
uint cid = c->clientId();
|
|
index_clients.insert(cid, c);
|
|
mutex_clients.unlock();
|
|
c->sendConnected(1);
|
|
server->sendConnected(cid);
|
|
CONNECTL(c, dataReaded, ([this, cid](PIByteArray & ba) {
|
|
// piCoutObj << c->clientId() << "dataReaded";
|
|
PITimeMeasurer tm;
|
|
server->sendDataToClient(ba, cid);
|
|
if (tm.elapsed_s() > 2.0) {
|
|
server->markWriteStalled();
|
|
piCoutObj << "write stalled to server" << cid;
|
|
}
|
|
}));
|
|
}
|
|
|
|
|
|
void CloudServer::removeClient(DispatcherClient * c) {
|
|
mutex_clients.lock();
|
|
last_ping.reset();
|
|
clients.removeOne(c);
|
|
index_clients.remove(c->clientId());
|
|
mutex_clients.unlock();
|
|
server->sendDisconnected(c->clientId());
|
|
}
|
|
|
|
|
|
PIVector<DispatcherClient *> CloudServer::getClients() {
|
|
PIMutexLocker locker(mutex_clients);
|
|
PIVector<DispatcherClient *> cl = clients;
|
|
return cl;
|
|
}
|
|
|
|
|
|
double CloudServer::lastPing() {
|
|
PIMutexLocker locker(mutex_clients);
|
|
return last_ping.elapsed_s();
|
|
}
|
|
|
|
|
|
void CloudServer::close() {
|
|
server->close();
|
|
}
|
|
|
|
|
|
void CloudServer::stop() {
|
|
// copy the list under the lock: addClient()/removeClient() modify it
|
|
// concurrently from other threads
|
|
PIVector<DispatcherClient *> cl;
|
|
{
|
|
PIMutexLocker locker(mutex_clients);
|
|
cl = clients;
|
|
}
|
|
for (auto c: cl)
|
|
c->close();
|
|
server->stop();
|
|
}
|
|
|
|
|
|
void CloudServer::printStatus() {
|
|
PIMutexLocker locker(mutex_clients);
|
|
piCout << " " << "Clients for" << server->address() << server_uuid.toHex() << ":";
|
|
for (auto c: clients) {
|
|
piCout << " " << c->address() << c->clientId();
|
|
}
|
|
}
|