SHS/pip
4
3
Fork 0
Code Issues 67 Pull Requests 1 Actions Releases Wiki Activity

some fixes for picloud, but still not working correctly

Browse Source
This commit is contained in:
Бычков Андрей
2022-11-08 14:43:52 +03:00
parent 36ff427e0d
commit 897f03f3d0
10 changed files with 60 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
CMakeLists.txt
View File

@@ -492,6 +492,8 @@ if (NOT CROSSTOOLS)
#target_link_libraries(pip_plugin pip) #target_link_libraries(pip_plugin pip)
add_executable(pip_test "main.cpp") add_executable(pip_test "main.cpp")
target_link_libraries(pip_test pip) target_link_libraries(pip_test pip)
add_executable(pip_cloud_test "main_picloud_test.cpp")
target_link_libraries(pip_cloud_test pip_cloud)
endif() endif()
else() else()

35
libs/cloud/picloudclient.cpp
View File

@@ -33,22 +33,22 @@ PICloudClient::PICloudClient(const PIString & path, PIIODevice::DeviceMode mode)
if (is_deleted) return; if (is_deleted) return;
bool need_disconn = is_connected; bool need_disconn = is_connected;
//piCoutObj << "eth disconnected"; //piCoutObj << "eth disconnected";
eth.softStopThreadedRead(); eth.stop();
opened_ = false; opened_ = false;
internalDisconnect(); internalDisconnect();
if (need_disconn) if (need_disconn) disconnected();
disconnected();
//piCoutObj << "eth disconnected done"; //piCoutObj << "eth disconnected done";
}); });
} }
PICloudClient::~PICloudClient() { PICloudClient::~PICloudClient() {
//piCoutObj << "~PICloudClient()"; piCoutObj << "~PICloudClient() ..." << this;
is_deleted = true;
stopAndWait(); stopAndWait();
close(); close();
is_deleted = true;
internalDisconnect(); internalDisconnect();
piCoutObj << "~PICloudClient() done" << this;
} }
@@ -70,14 +70,14 @@ void PICloudClient::interrupt() {
bool PICloudClient::openDevice() { bool PICloudClient::openDevice() {
//piCoutObj << "open";// << path(); piCoutObj << "open";// << path();
bool op = eth.connect(PIEthernet::Address::resolve(path()), false); bool op = eth.connect(PIEthernet::Address::resolve(path()), false);
if (op) { if (op) {
mutex_connect.lock(); mutex_connect.lock();
eth.startThreadedRead(); eth.startThreadedRead();
//piCoutObj << "connecting..."; piCoutObj << "connecting...";
bool conn_ok = cond_connect.waitFor(mutex_connect, (int)eth.readTimeout()); bool conn_ok = cond_connect.waitFor(mutex_connect, (int)eth.readTimeout());
//piCoutObj << "conn_ok" << conn_ok << is_connected; piCoutObj << "conn_ok" << conn_ok << is_connected;
mutex_connect.unlock(); mutex_connect.unlock();
if (!conn_ok) { if (!conn_ok) {
mutex_connect.lock(); mutex_connect.lock();
@@ -105,36 +105,37 @@ bool PICloudClient::closeDevice() {
ssize_t PICloudClient::readDevice(void * read_to, ssize_t max_size) { ssize_t PICloudClient::readDevice(void * read_to, ssize_t max_size) {
if (is_deleted) return -1; if (is_deleted || max_size <= 0) return -1;
//piCoutObj << "readDevice"; piCoutObj << "readDevice ...";
if (!is_connected && eth.isClosed()) openDevice(); if (!is_connected && eth.isClosed()) openDevice();
ssize_t sz = -1; ssize_t sz = -1;
mutex_buff.lock(); mutex_buff.lock();
cond_buff.wait(mutex_buff);
if (is_connected) { if (is_connected) {
if (buff.isEmpty()) { if (buff.isEmpty()) {
sz = 0; sz = 0;
} else { } else {
sz = piMini(max_size, buff.size()); sz = piMin<ssize_t>(max_size, buff.size_s());
memcpy(read_to, buff.data(), sz); memcpy(read_to, buff.data(), sz);
buff.remove(0, sz); buff.remove(0, sz);
} }
if (sz == 0) cond_buff.wait(mutex_buff);
} }
mutex_buff.unlock(); mutex_buff.unlock();
if (!is_connected) opened_ = false; if (!is_connected) opened_ = false;
//piCoutObj << "readDevice done" << sz; piCoutObj << "readDevice done" << sz;
return sz; return sz;
} }
ssize_t PICloudClient::writeDevice(const void * data, ssize_t size) { ssize_t PICloudClient::writeDevice(const void * data, ssize_t size) {
if (is_deleted) return -1; if (is_deleted) return -1;
// piCoutObj << "writeDevice"; piCoutObj << "writeDevice" << size;
return tcp.sendData(PIByteArray(data, size)); return tcp.sendData(PIByteArray(data, size));
} }
void PICloudClient::internalDisconnect() { void PICloudClient::internalDisconnect() {
piCoutObj << "internalDisconnect";
is_connected = false; is_connected = false;
cond_buff.notifyOne(); cond_buff.notifyOne();
cond_connect.notifyOne(); cond_connect.notifyOne();
@@ -146,7 +147,7 @@ void PICloudClient::internalDisconnect() {
void PICloudClient::_readed(PIByteArray & ba) { void PICloudClient::_readed(PIByteArray & ba) {
if (is_deleted) return; if (is_deleted) return;
PIPair<PICloud::TCP::Type, PICloud::TCP::Role> hdr = tcp.parseHeader(ba); PIPair<PICloud::TCP::Type, PICloud::TCP::Role> hdr = tcp.parseHeader(ba);
//piCoutObj << "_readed" << ba.size() << hdr.first << hdr.second; piCoutObj << "_readed" << ba.size() << hdr.first << hdr.second;
if (hdr.second == tcp.role()) { if (hdr.second == tcp.role()) {
switch (hdr.first) { switch (hdr.first) {
case PICloud::TCP::Connect: case PICloud::TCP::Connect:
@@ -159,7 +160,7 @@ void PICloudClient::_readed(PIByteArray & ba) {
} }
break; break;
case PICloud::TCP::Disconnect: case PICloud::TCP::Disconnect:
eth.softStopThreadedRead(); eth.stop();
opened_ = false; opened_ = false;
eth.close(); eth.close();
break; break;
@@ -177,5 +178,5 @@ void PICloudClient::_readed(PIByteArray & ba) {
//piCoutObj << "readed" << ba.toHex(); //piCoutObj << "readed" << ba.toHex();
} }
while (buff.size_s() > threadedReadBufferSize()) piMSleep(100); // FIXME: sleep here is bad while (buff.size_s() > threadedReadBufferSize()) piMSleep(100); // FIXME: sleep here is bad
//piCoutObj << "_readed done"; piCoutObj << "_readed done";
} }

42
libs/cloud/picloudserver.cpp
View File

@@ -26,11 +26,10 @@ PICloudServer::PICloudServer(const PIString & path, PIIODevice::DeviceMode mode)
tcp.setServerName(server_name); tcp.setServerName(server_name);
setName("cloud_server__" + server_name); setName("cloud_server__" + server_name);
CONNECT1(void, PIByteArray, &streampacker, packetReceiveEvent, this, _readed); CONNECT1(void, PIByteArray, &streampacker, packetReceiveEvent, this, _readed);
CONNECTL(&eth, connected, [this](){opened_ = true; piCoutObj << "connected"; tcp.sendStart();}); CONNECTL(&eth, connected, [this](){opened_ = true; piCoutObj << "connected" << &eth; tcp.sendStart();});
CONNECTL(&eth, disconnected, [this](bool){ CONNECTL(&eth, disconnected, [this](bool){
piCoutObj << "disconnected"; piCoutObj << "disconnected" << &eth;
eth.softStopThreadedRead(); eth.stop();
eth.interrupt();
opened_ = false; opened_ = false;
ping_timer.stop(false); ping_timer.stop(false);
piMSleep(100); piMSleep(100);
@@ -42,8 +41,10 @@ PICloudServer::PICloudServer(const PIString & path, PIIODevice::DeviceMode mode)
PICloudServer::~PICloudServer() { PICloudServer::~PICloudServer() {
piCoutObj << "~PICloudServer ..." << this;
stopAndWait(); stopAndWait();
close(); close();
piCoutObj << "~PICloudServer done" << this;
} }
@@ -60,7 +61,7 @@ PIVector<PICloudServer::Client *> PICloudServer::clients() const {
bool PICloudServer::openDevice() { bool PICloudServer::openDevice() {
//piCout << "PICloudServer open device" << path(); piCout << "PICloudServer open device" << path();
bool op = eth.connect(PIEthernet::Address::resolve(path()), false); bool op = eth.connect(PIEthernet::Address::resolve(path()), false);
if (op) { if (op) {
eth.startThreadedRead(); eth.startThreadedRead();
@@ -74,17 +75,19 @@ bool PICloudServer::openDevice() {
bool PICloudServer::closeDevice() { bool PICloudServer::closeDevice() {
eth.stop(); piCoutObj << "closeDevice" << this;
eth.stopAndWait();
ping_timer.stop(false); ping_timer.stop(false);
clients_mutex.lock(); clients_mutex.lock();
for (auto c : clients_) { for (auto c : clients_) {
c->stopAndWait();
c->close(); c->close();
c->stop();
} }
clients_mutex.unlock(); clients_mutex.unlock();
eth.close(); eth.close();
for (auto c : clients_) for (auto c : clients_) {
delete c; delete c;
}
return true; return true;
} }
@@ -126,12 +129,10 @@ PICloudServer::Client::Client(PICloudServer * srv, uint id) : server(srv), clien
PICloudServer::Client::~Client() { PICloudServer::Client::~Client() {
if (is_connected) { piCoutObj << "~PICloudServer::Client..." << this;
is_connected = false;
cond_buff.notifyOne();
}
stopAndWait();
close(); close();
stopAndWait();
piCoutObj << "~PICloudServer::Client done" << this;
} }
@@ -141,6 +142,7 @@ bool PICloudServer::Client::openDevice() {
bool PICloudServer::Client::closeDevice() { bool PICloudServer::Client::closeDevice() {
piCoutObj << "closeDevice" << this;
if (is_connected) { if (is_connected) {
server->clientDisconnect(client_id); server->clientDisconnect(client_id);
is_connected = false; is_connected = false;
@@ -154,7 +156,6 @@ ssize_t PICloudServer::Client::readDevice(void * read_to, ssize_t max_size) {
if (!is_connected) return -1; if (!is_connected) return -1;
ssize_t sz = -1; ssize_t sz = -1;
mutex_buff.lock(); mutex_buff.lock();
cond_buff.wait(mutex_buff);
if (is_connected) { if (is_connected) {
if (buff.isEmpty()) { if (buff.isEmpty()) {
sz = 0; sz = 0;
@@ -163,6 +164,7 @@ ssize_t PICloudServer::Client::readDevice(void * read_to, ssize_t max_size) {
memcpy(read_to, buff.data(), sz); memcpy(read_to, buff.data(), sz);
buff.remove(0, sz); buff.remove(0, sz);
} }
if (sz == 0) cond_buff.wait(mutex_buff);
} }
mutex_buff.unlock(); mutex_buff.unlock();
return sz; return sz;
@@ -201,8 +203,8 @@ void PICloudServer::_readed(PIByteArray & ba) {
if (oc) { if (oc) {
tcp.sendDisconnected(id); tcp.sendDisconnected(id);
} else { } else {
//piCoutObj << "new Client" << id;
Client * c = new Client(this, id); Client * c = new Client(this, id);
piCoutObj << "new Client" << id << c;
CONNECT1(void, PIObject *, c, deleted, this, clientDeleted); CONNECT1(void, PIObject *, c, deleted, this, clientDeleted);
clients_mutex.lock(); clients_mutex.lock();
clients_ << c; clients_ << c;
@@ -213,13 +215,14 @@ void PICloudServer::_readed(PIByteArray & ba) {
} break; } break;
case PICloud::TCP::Disconnect: { case PICloud::TCP::Disconnect: {
uint id = tcp.parseDisconnect(ba); uint id = tcp.parseDisconnect(ba);
//piCoutObj << "remove Client" << id; piCoutObj << "remove Client" << id;
clients_mutex.lock(); clients_mutex.lock();
Client * oc = index_clients.value(id, nullptr); Client * oc = index_clients.value(id, nullptr);
clients_mutex.unlock(); clients_mutex.unlock();
if (oc) { if (oc) {
oc->is_connected = false; //oc->is_connected = false;
oc->close(); //oc->close();
delete oc;
} }
} break; } break;
case PICloud::TCP::Data: { case PICloud::TCP::Data: {
@@ -227,7 +230,7 @@ void PICloudServer::_readed(PIByteArray & ba) {
clients_mutex.lock(); clients_mutex.lock();
Client * oc = index_clients.value(d.first, nullptr); Client * oc = index_clients.value(d.first, nullptr);
clients_mutex.unlock(); clients_mutex.unlock();
//piCoutObj << "data for" << d.first << d.second.toHex(); piCoutObj << "data for" << d.first << d.second.size();
if (oc && !d.second.isEmpty()) oc->pushBuffer(d.second); if (oc && !d.second.isEmpty()) oc->pushBuffer(d.second);
} break; } break;
default: break; default: break;
@@ -238,6 +241,7 @@ void PICloudServer::_readed(PIByteArray & ba) {
void PICloudServer::clientDeleted(PIObject * o) { void PICloudServer::clientDeleted(PIObject * o) {
PICloudServer::Client * c = (PICloudServer::Client*)o; PICloudServer::Client * c = (PICloudServer::Client*)o;
piCoutObj << "clientDeleted" << c;
clients_mutex.lock(); clients_mutex.lock();
clients_.removeOne(c); clients_.removeOne(c);
auto it = index_clients.makeIterator(); auto it = index_clients.makeIterator();

4
libs/main/io_devices/piiodevice.cpp
View File

@@ -637,7 +637,3 @@ void PIIODevice::configureFromVariantDevice(const PIPropertyStorage & d) {
} }
void PIIODevice::softStopThreadedRead() {
read_thread.stop();
}

2
libs/main/io_devices/piiodevice.h
View File

@@ -390,8 +390,6 @@ public:
//! \~russian Пишет в устройство блок памяти "mb" //! \~russian Пишет в устройство блок памяти "mb"
ssize_t write(const PIMemoryBlock & mb) {return write(mb.data(), mb.size());} ssize_t write(const PIMemoryBlock & mb) {return write(mb.data(), mb.size());}
void softStopThreadedRead();
EVENT_VHANDLER(void, flush) {;} EVENT_VHANDLER(void, flush) {;}
EVENT(opened); EVENT(opened);

2
libs/main/io_devices/piserial.cpp
View File

@@ -827,7 +827,7 @@ ssize_t PISerial::readDevice(void * read_to, ssize_t max_size) {
return 0; return 0;
if (err == ERROR_BAD_COMMAND || err == ERROR_ACCESS_DENIED) { if (err == ERROR_BAD_COMMAND || err == ERROR_ACCESS_DENIED) {
piCoutObj << "read error" << (PRIVATE->readed) << errorString(); piCoutObj << "read error" << (PRIVATE->readed) << errorString();
softStopThreadedRead(); stop();
close(); close();
return 0; return 0;
} }

6
main_picloud_test.cpp
View File

@@ -18,7 +18,7 @@ int main(int argc, char * argv[]) {
piCout << "[Client] send:" << str; piCout << "[Client] send:" << str;
c.write(str.toByteArray()); c.write(str.toByteArray());
} }
if (s.isRunning()) { if (s.isThreadedRead()) {
for (auto cl : clients) { for (auto cl : clients) {
if (cl->isOpened()) { if (cl->isOpened()) {
PIString str = "ping_S"; PIString str = "ping_S";
@@ -51,9 +51,11 @@ int main(int argc, char * argv[]) {
} }
})); }));
CONNECTL(cl, closed, ([&clients, cl](){ CONNECTL(cl, closed, ([&clients, cl](){
piCout << "[Server] client closed ..." << cl;
cl->stop(); cl->stop();
clients.removeAll(cl); clients.removeAll(cl);
cl->deleteLater(); piCout << "[Server] client closed ok" << cl;
//cl->deleteLater();
})); }));
cl->startThreadedRead(); cl->startThreadedRead();
})); }));

16
utils/cloud_dispatcher/dispatcherclient.cpp
View File

@@ -3,7 +3,6 @@
DispatcherClient::DispatcherClient(PIEthernet * eth_, int id) : authorised(false), eth(eth_), streampacker(eth_), tcp(&streampacker), client_id(id) { DispatcherClient::DispatcherClient(PIEthernet * eth_, int id) : authorised(false), eth(eth_), streampacker(eth_), tcp(&streampacker), client_id(id) {
CONNECTU(&disconnect_tm, tickEvent, eth, close);
CONNECTU(&streampacker, packetReceiveEvent, this, readed); CONNECTU(&streampacker, packetReceiveEvent, this, readed);
CONNECTU(eth, disconnected, this, disconnected); CONNECTU(eth, disconnected, this, disconnected);
piCoutObj << "client connected" << client_id << eth->sendAddress(); piCoutObj << "client connected" << client_id << eth->sendAddress();
@@ -12,7 +11,6 @@ DispatcherClient::DispatcherClient(PIEthernet * eth_, int id) : authorised(false
void DispatcherClient::start() { void DispatcherClient::start() {
eth->startThreadedRead(); eth->startThreadedRead();
//disconnect_tm.start(10000);
} }
@@ -27,18 +25,18 @@ PIString DispatcherClient::address() {
void DispatcherClient::close() { void DispatcherClient::close() {
eth->softStopThreadedRead(); eth->stopAndWait();
eth->interrupt();
} }
void DispatcherClient::sendConnected(uint client_id) { void DispatcherClient::sendConnected(uint client_id) {
//piCoutObj << "sendConnected"; piCoutObj << "sendConnected" << client_id;
tcp.sendConnected(client_id); tcp.sendConnected(client_id);
} }
void DispatcherClient::sendDisconnected(uint client_id) { void DispatcherClient::sendDisconnected(uint client_id) {
piCoutObj << "sendDisconnected" << client_id;
tcp.sendDisconnected(client_id); tcp.sendDisconnected(client_id);
} }
@@ -61,7 +59,7 @@ void DispatcherClient::authorise(bool ok) {
void DispatcherClient::disconnected(bool withError) { void DispatcherClient::disconnected(bool withError) {
//piCoutObj << "client disconnected" << eth->sendAddress(); piCoutObj << "client disconnected" << withError << eth->sendAddress();
disconnectEvent(this); disconnectEvent(this);
} }
@@ -70,16 +68,18 @@ void DispatcherClient::readed(PIByteArray & ba) {
PIPair<PICloud::TCP::Type, PICloud::TCP::Role> hdr = tcp.parseHeader(ba); PIPair<PICloud::TCP::Type, PICloud::TCP::Role> hdr = tcp.parseHeader(ba);
// piCoutObj << "readed" << hdr.first << hdr.second; // piCoutObj << "readed" << hdr.first << hdr.second;
if (hdr.first == PICloud::TCP::InvalidType) { if (hdr.first == PICloud::TCP::InvalidType) {
disconnected(true);
piCoutObj << "invalid message"; piCoutObj << "invalid message";
disconnected(true);
return; return;
} }
if (authorised) { if (authorised) {
if (hdr.second == tcp.role()) { if (hdr.second == tcp.role()) {
switch (hdr.first) { switch (hdr.first) {
case PICloud::TCP::Connect: case PICloud::TCP::Connect:
piCoutObj << "PICloud::TCP::Connect";
return; return;
case PICloud::TCP::Disconnect: case PICloud::TCP::Disconnect:
piCoutObj << "PICloud::TCP::Disconnect";
disconnected(false); disconnected(false);
return; return;
case PICloud::TCP::Data: case PICloud::TCP::Data:
@@ -113,9 +113,11 @@ void DispatcherClient::readed(PIByteArray & ba) {
return; return;
} }
case PICloud::TCP::Disconnect: case PICloud::TCP::Disconnect:
piCoutObj << "unauthorised PICloud::TCP::Disconnect";
disconnected(false); disconnected(false);
return; return;
default: default:
piCoutObj << "authorised invalid message";
disconnected(true); disconnected(true);
return; return;
} }

1
utils/cloud_dispatcher/dispatcherclient.h
View File

@@ -33,7 +33,6 @@ private:
EVENT_HANDLER1(void, readed, PIByteArray &, data); EVENT_HANDLER1(void, readed, PIByteArray &, data);
EVENT_HANDLER1(void, disconnected, bool, withError); EVENT_HANDLER1(void, disconnected, bool, withError);
PITimer disconnect_tm;
std::atomic_bool authorised; std::atomic_bool authorised;
PIEthernet * eth; PIEthernet * eth;
PIStreamPacker streampacker; PIStreamPacker streampacker;

5
utils/cloud_dispatcher/dispatcherserver.cpp
View File

@@ -181,7 +181,7 @@ void DispatcherServer::disconnectClient(DispatcherClient *client) {
//piCoutObj << "INVALID client" << client; //piCoutObj << "INVALID client" << client;
return; return;
} }
piCoutObj << "remove" << client->clientId(); piCoutObj << "remove ..." << client->clientId();
map_mutex.lock(); map_mutex.lock();
clients.removeAll(client); clients.removeAll(client);
rm_clients.removeAll(client); rm_clients.removeAll(client);
@@ -206,9 +206,10 @@ void DispatcherServer::disconnectClient(DispatcherClient *client) {
cc->removeClient(client); cc->removeClient(client);
index_c_clients.remove(client); index_c_clients.remove(client);
} }
client->close(); //client->close();
rmrf_clients << client; rmrf_clients << client;
map_mutex.unlock(); map_mutex.unlock();
piCoutObj << "remove done" << client->clientId();
} }