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

thread #169

Merged
peri4 merged 30 commits from thread into master 2022-11-10 16:16:08 +03:00
Conversation 0 Commits 30 Files Changed 36 +1028 -357
36 changed files with 1028 additions and 357 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()

57
libs/cloud/picloudclient.cpp
View File

@@ -33,28 +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;
softStopThreadedRead();
//eth.close();
//if (is_connected) disconnected();
close();
//piCoutObj << "~PICloudClient() closed";
internalDisconnect();
// stop(false);
is_deleted = true; is_deleted = true;
stopAndWait();
close();
internalDisconnect(); internalDisconnect();
//piCoutObj << "~PICloudClient() done"; //piCoutObj << "~PICloudClient() done" << this;
} }
@@ -69,6 +63,12 @@ void PICloudClient::setKeepConnection(bool on) {
} }
void PICloudClient::interrupt() {
cond_buff.notifyOne();
cond_connect.notifyOne();
}
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);
@@ -81,7 +81,7 @@ bool PICloudClient::openDevice() {
mutex_connect.unlock(); mutex_connect.unlock();
if (!conn_ok) { if (!conn_ok) {
mutex_connect.lock(); mutex_connect.lock();
eth.stop(); eth.stopAndWait();
eth.close(); eth.close();
mutex_connect.unlock(); mutex_connect.unlock();
} }
@@ -98,41 +98,44 @@ bool PICloudClient::closeDevice() {
if (is_connected) { if (is_connected) {
internalDisconnect(); internalDisconnect();
} }
eth.stop(); eth.stopAndWait();
eth.close(); eth.close();
return true; return true;
} }
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 ...";
reading_now = true;
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, [this](){return !buff.isEmpty() || !is_connected;});
if (is_connected) { if (is_connected) {
sz = piMini(max_size, buff.size()); if (buff.isEmpty()) {
sz = 0;
} else {
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;
reading_now = 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 || !is_connected) 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();
@@ -157,13 +160,18 @@ 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;
case PICloud::TCP::Data: case PICloud::TCP::Data:
if (is_connected) { if (is_connected) {
mutex_buff.lock(); mutex_buff.lock();
if (buff.size_s() > threadedReadBufferSize()) {
piCoutObj << "Error: buffer overflow, drop" << ba.size() << "bytes";
mutex_buff.unlock();
return;
}
buff.append(ba); buff.append(ba);
mutex_buff.unlock(); mutex_buff.unlock();
cond_buff.notifyOne(); cond_buff.notifyOne();
@@ -174,7 +182,6 @@ 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 if (buff.size_s() > threadedReadBufferSize()) piMSleep(100); // FIXME: sleep here is bad
//piCoutObj << "_readed done"; //piCoutObj << "_readed done";
} }

93
libs/cloud/picloudserver.cpp
View File

@@ -25,13 +25,22 @@ PICloudServer::PICloudServer(const PIString & path, PIIODevice::DeviceMode mode)
tcp.setRole(PICloud::TCP::Server); tcp.setRole(PICloud::TCP::Server);
tcp.setServerName(server_name); tcp.setServerName(server_name);
setName("cloud_server__" + server_name); setName("cloud_server__" + server_name);
is_deleted = false;
eth.setReopenEnabled(false);
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"; if (is_deleted) return;
eth.softStopThreadedRead(); //piCoutObj << "disconnected" << &eth;
for (auto c : clients_) {
delete c;
}
opened_ = false; opened_ = false;
ping_timer.stop(false); ping_timer.stop();
piMSleep(100); piMSleep(100);
}); });
CONNECTL(&ping_timer, tickEvent, [this] (void *, int){ CONNECTL(&ping_timer, tickEvent, [this] (void *, int){
@@ -41,8 +50,13 @@ PICloudServer::PICloudServer(const PIString & path, PIIODevice::DeviceMode mode)
PICloudServer::~PICloudServer() { PICloudServer::~PICloudServer() {
//piCoutObj << "~PICloudServer ..." << this;
is_deleted = true;
stop(); stop();
close(); close();
//piCout << "wait";
waitThreadedReadFinished();
//piCoutObj << "~PICloudServer done" << this;
} }
@@ -65,35 +79,32 @@ bool PICloudServer::openDevice() {
eth.startThreadedRead(); eth.startThreadedRead();
ping_timer.start(5000); ping_timer.start(5000);
return true; return true;
} } else {
ping_timer.stop(false); ping_timer.stop();
eth.close(); eth.close();
return false; return false;
}
} }
bool PICloudServer::closeDevice() { bool PICloudServer::closeDevice() {
eth.stop(); //piCoutObj << "closeDevice" << this;
ping_timer.stop(false); eth.stopAndWait();
clients_mutex.lock(); ping_timer.stop();
for (auto c : clients_) {
c->close();
c->stop();
}
clients_mutex.unlock();
eth.close(); eth.close();
for (auto c : clients_) for (auto c : clients_) {
delete c; delete c;
}
clients_.clear();
return true; return true;
} }
ssize_t PICloudServer::readDevice(void * read_to, ssize_t max_size) { ssize_t PICloudServer::readDevice(void * read_to, ssize_t max_size) {
if (is_deleted) return -1;
//piCoutObj << "readDevice"; //piCoutObj << "readDevice";
reading_now = true;
if (!opened_) openDevice(); if (!opened_) openDevice();
else piMSleep(eth.readTimeout()); //else piMSleep(eth.readTimeout());
reading_now = false;
return -1; return -1;
} }
@@ -104,12 +115,18 @@ ssize_t PICloudServer::writeDevice(const void * data, ssize_t max_size) {
} }
void PICloudServer::interrupt() {
eth.interrupt();
}
void PICloudServer::clientDisconnect(uint client_id) { void PICloudServer::clientDisconnect(uint client_id) {
tcp.sendDisconnected(client_id); tcp.sendDisconnected(client_id);
} }
int PICloudServer::sendData(const PIByteArray & data, uint client_id) { int PICloudServer::sendData(const PIByteArray & data, uint client_id) {
if (!opened_) return -1;
return tcp.sendData(data, client_id); return tcp.sendData(data, client_id);
} }
@@ -122,12 +139,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();
}
close(); close();
stop(); stopAndWait();
//piCoutObj << "~PICloudServer::Client done" << this;
} }
@@ -137,7 +152,7 @@ bool PICloudServer::Client::openDevice() {
bool PICloudServer::Client::closeDevice() { bool PICloudServer::Client::closeDevice() {
softStopThreadedRead(); //piCoutObj << "closeDevice" << this;
if (is_connected) { if (is_connected) {
server->clientDisconnect(client_id); server->clientDisconnect(client_id);
is_connected = false; is_connected = false;
@@ -151,33 +166,48 @@ 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, [this](){return !buff.isEmpty() || !is_connected;});
if (is_connected) { if (is_connected) {
if (buff.isEmpty()) {
sz = 0;
} else {
sz = piMini(max_size, buff.size()); sz = piMini(max_size, buff.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;
} }
ssize_t PICloudServer::Client::writeDevice(const void * data, ssize_t size) { ssize_t PICloudServer::Client::writeDevice(const void * data, ssize_t size) {
if (!is_connected) return -1;
return server->sendData(PIByteArray(data, size), client_id); return server->sendData(PIByteArray(data, size), client_id);
} }
void PICloudServer::Client::interrupt() {
cond_buff.notifyOne();
}
void PICloudServer::Client::pushBuffer(const PIByteArray & ba) { void PICloudServer::Client::pushBuffer(const PIByteArray & ba) {
if (!is_connected) return; if (!is_connected) return;
mutex_buff.lock(); mutex_buff.lock();
if (buff.size_s() > threadedReadBufferSize()) {
piCoutObj << "Error: buffer overflow, drop" << ba.size() << "bytes";
mutex_buff.unlock();
return;
}
buff.append(ba); buff.append(ba);
cond_buff.notifyOne(); cond_buff.notifyOne();
mutex_buff.unlock(); mutex_buff.unlock();
while (buff.size_s() > threadedReadBufferSize()) piMSleep(100); // FIXME: sleep here is bad
} }
void PICloudServer::_readed(PIByteArray & ba) { void PICloudServer::_readed(PIByteArray & ba) {
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);
if (hdr.second == tcp.role()) { if (hdr.second == tcp.role()) {
switch (hdr.first) { switch (hdr.first) {
@@ -189,8 +219,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;
@@ -206,8 +236,9 @@ void PICloudServer::_readed(PIByteArray & ba) {
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->stopAndWait();
oc->is_connected = false; oc->is_connected = false;
oc->close(); delete oc;
} }
} break; } break;
case PICloud::TCP::Data: { case PICloud::TCP::Data: {
@@ -215,7 +246,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;
@@ -226,11 +257,11 @@ 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();
while (it.hasNext()) { while (it.next()) {
it.next();
if (it.value() == c) { if (it.value() == c) {
index_clients.remove(it.key()); index_clients.remove(it.key());
break; break;

2
libs/console/piscreen.cpp
View File

@@ -588,7 +588,7 @@ void PIScreen::waitForFinish() {
void PIScreen::stop(bool clear) { void PIScreen::stop(bool clear) {
PIThread::stop(true); PIThread::stopAndWait();
if (clear) console.clearScreen(); if (clear) console.clearScreen();
#ifndef WINDOWS #ifndef WINDOWS
fflush(0); fflush(0);

2
libs/main/cloud/picloudclient.h
View File

@@ -43,6 +43,7 @@ public:
void setKeepConnection(bool on); void setKeepConnection(bool on);
bool isConnected() const {return is_connected;} bool isConnected() const {return is_connected;}
ssize_t bytesAvailable() const override {return buff.size();} ssize_t bytesAvailable() const override {return buff.size();}
void interrupt() override;
EVENT(connected); EVENT(connected);
EVENT(disconnected); EVENT(disconnected);
@@ -65,6 +66,7 @@ private:
PIConditionVariable cond_connect; PIConditionVariable cond_connect;
std::atomic_bool is_connected; std::atomic_bool is_connected;
std::atomic_bool is_deleted; std::atomic_bool is_deleted;
}; };
#endif // PICLOUDCLIENT_H #endif // PICLOUDCLIENT_H

3
libs/main/cloud/picloudserver.h
View File

@@ -51,6 +51,7 @@ public:
ssize_t writeDevice(const void * data, ssize_t size) override; ssize_t writeDevice(const void * data, ssize_t size) override;
DeviceInfoFlags deviceInfoFlags() const override {return PIIODevice::Reliable;} DeviceInfoFlags deviceInfoFlags() const override {return PIIODevice::Reliable;}
ssize_t bytesAvailable() const override {return buff.size();} ssize_t bytesAvailable() const override {return buff.size();}
void interrupt() override;
private: private:
void pushBuffer(const PIByteArray & ba); void pushBuffer(const PIByteArray & ba);
@@ -73,6 +74,7 @@ protected:
bool closeDevice() override; bool closeDevice() override;
ssize_t readDevice(void * read_to, ssize_t max_size) override; ssize_t readDevice(void * read_to, ssize_t max_size) override;
ssize_t writeDevice(const void * data, ssize_t max_size) override; ssize_t writeDevice(const void * data, ssize_t max_size) override;
void interrupt() override;
private: private:
EVENT_HANDLER1(void, _readed, PIByteArray &, ba); EVENT_HANDLER1(void, _readed, PIByteArray &, ba);
@@ -84,6 +86,7 @@ private:
PIMap<uint, Client *> index_clients; PIMap<uint, Client *> index_clients;
PITimer ping_timer; PITimer ping_timer;
mutable PIMutex clients_mutex; mutable PIMutex clients_mutex;
std::atomic_bool is_deleted;
}; };
#endif // PICLOUDSERVER_H #endif // PICLOUDSERVER_H

4
libs/main/core/pibase.h
View File

@@ -248,6 +248,10 @@
typedef long time_t; typedef long time_t;
#endif #endif
#ifdef POSIX_SIGNALS
# define PIP_INTERRUPT_SIGNAL SIGTERM
#endif
#ifdef LINUX #ifdef LINUX
# define environ __environ # define environ __environ
#endif #endif

1
libs/main/core/piincludes.h
View File

@@ -44,6 +44,7 @@ class PIInit;
#endif #endif
class PIChar; class PIChar;
class PICout; class PICout;
class PIWaitEvent;
struct lconv; struct lconv;

4
libs/main/core/piincludes_p.h
View File

@@ -22,6 +22,10 @@
#include "picout.h" #include "picout.h"
#ifdef WINDOWS #ifdef WINDOWS
# ifdef _WIN32_WINNT
# undef _WIN32_WINNT
# define _WIN32_WINNT 0x0600
# endif
# include <stdarg.h> # include <stdarg.h>
# include <windef.h> # include <windef.h>
# include <winbase.h> # include <winbase.h>

27
libs/main/core/piinit.cpp
View File

@@ -92,9 +92,20 @@ void __sighandler__(PISignals::Signal s) {
} }
#ifdef ANDROID #ifdef POSIX_SIGNALS
void android_thread_exit_handler(int sig) { void pipThreadSignalHandler(int sig) {
pthread_exit(0); //# ifdef ANDROID
// pthread_exit(0);
//# endif
}
void pipInitThreadSignals() {
struct sigaction actions;
memset(&actions, 0, sizeof(actions));
sigemptyset(&actions.sa_mask);
actions.sa_flags = 0;
actions.sa_handler = pipThreadSignalHandler;
if (sigaction(PIP_INTERRUPT_SIGNAL, &actions, 0) != 0)
piCout << "sigaction error:" << errorString();
} }
#endif #endif
@@ -166,14 +177,10 @@ PIInit::PIInit() {
setlocale(LC_ALL, ""); setlocale(LC_ALL, "");
setlocale(LC_NUMERIC, "C"); setlocale(LC_NUMERIC, "C");
# endif //HAS_LOCALE # endif //HAS_LOCALE
#else //ANDROID
struct sigaction actions;
memset(&actions, 0, sizeof(actions));
sigemptyset(&actions.sa_mask);
actions.sa_flags = 0;
actions.sa_handler = android_thread_exit_handler;
sigaction(SIGTERM, &actions, 0);
#endif //ANDROID #endif //ANDROID
#ifdef POSIX_SIGNALS
pipInitThreadSignals();
#endif
PRIVATE->delete_locs = false; PRIVATE->delete_locs = false;
__syslocname__ = __sysoemname__ = 0; __syslocname__ = __sysoemname__ = 0;
__utf8name__ = const_cast<char*>("UTF-8"); __utf8name__ = const_cast<char*>("UTF-8");

2
libs/main/core/piobject.cpp
View File

@@ -839,7 +839,7 @@ PIObject::Deleter::Deleter() {
PRIVATE->thread.setSlot([this](){ PRIVATE->thread.setSlot([this](){
PIVector<PIObject*> oq; PIVector<PIObject*> oq;
PRIVATE->thread.lock(); PRIVATE->thread.lock();
while(PRIVATE->obj_queue.isEmpty()) PRIVATE->cond_var.wait(PRIVATE->thread.mutex()); if (PRIVATE->obj_queue.isEmpty()) PRIVATE->cond_var.wait(PRIVATE->thread.mutex());
oq.swap(PRIVATE->obj_queue); oq.swap(PRIVATE->obj_queue);
PRIVATE->thread.unlock(); PRIVATE->thread.unlock();
for (PIObject * o : oq) deleteObject(o); for (PIObject * o : oq) deleteObject(o);

144
libs/main/core/piwaitevent_p.cpp Normal file
View File

@@ -0,0 +1,144 @@
/*
PIP - Platform Independent Primitives
Private PIP wait object
Ivan Pelipenko peri4ko@yandex.ru
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "piwaitevent_p.h"
#ifdef WINDOWS
//# ifdef _WIN32_WINNT
//# undef _WIN32_WINNT
//# define _WIN32_WINNT 0x0600
//# endif
# include <synchapi.h>
#else
# include <fcntl.h>
# include <sys/ioctl.h>
#endif
#include "pistring.h"
PIWaitEvent::~PIWaitEvent() {
destroy();
}
void PIWaitEvent::create() {
destroy();
#ifdef WINDOWS
event = CreateEventA(NULL, TRUE, FALSE, NULL);
if (!event) {
piCout << "Error with CreateEventA:" << errorString();
}
#else
for (int i = 0; i < 3; ++i) memset(&(fds[i]), 0, sizeof(fds[i]));
if (::pipe(pipe_fd) < 0) {
piCout << "Error with pipe:" << errorString();
} else {
fcntl(pipe_fd[ReadEnd], F_SETFL, O_NONBLOCK);
}
#endif
}
void PIWaitEvent::destroy() {
#ifdef WINDOWS
if (event) {
CloseHandle(event);
event = NULL;
}
#else
for (int i = 0; i < 2; ++i) {
if (pipe_fd[i] != 0) {
::close(pipe_fd[i]);
pipe_fd[i] = 0;
}
}
#endif
}
bool PIWaitEvent::wait(int fd, CheckRole role) {
if (!isCreate()) return false;
#ifdef WINDOWS
DWORD ret = WaitForSingleObjectEx(event, INFINITE, TRUE);
ResetEvent(event);
if (ret == WAIT_IO_COMPLETION || ret == WAIT_FAILED) return false;
#else
if (fd == -1) return false;
int nfds = piMaxi(pipe_fd[ReadEnd], fd) + 1;
int fd_index = role;
for (int i = 0; i < 3; ++i) FD_ZERO(&(fds[i]));
FD_SET(pipe_fd[ReadEnd], &(fds[CheckRead]));
FD_SET(fd, &(fds[CheckExeption]));
if (fd_index != CheckExeption) FD_SET(fd, &(fds[fd_index]));
::select(nfds, &(fds[CheckRead]), &(fds[CheckWrite]), &(fds[CheckExeption]), nullptr);
int buf = 0;
while (::read(pipe_fd[ReadEnd], &buf, sizeof(buf)) > 0);
if (FD_ISSET(fd, &(fds[CheckExeption]))) return false;
return FD_ISSET(fd, &(fds[fd_index]));
#endif
return true;
}
bool PIWaitEvent::sleep(int us) {
if (!isCreate()) return false;
#ifdef WINDOWS
DWORD ret = WaitForSingleObjectEx(event, us / 1000, TRUE);
ResetEvent(event);
return ret == WAIT_TIMEOUT;
#else
int nfds = pipe_fd[ReadEnd] + 1;
FD_ZERO(&(fds[CheckRead]));
FD_SET(pipe_fd[ReadEnd], &(fds[CheckRead]));
timeval timeout;
timeout.tv_sec = us / 1000000;
timeout.tv_usec = us % 1000000;
int ret = ::select(nfds, &(fds[CheckRead]), nullptr, nullptr, &timeout);
int buf = 0;
while (::read(pipe_fd[ReadEnd], &buf, sizeof(buf)) > 0);
return ret == 0;
#endif
}
void PIWaitEvent::interrupt() {
if (!isCreate()) return;
#ifdef WINDOWS
SetEvent(event);
#else
::write(pipe_fd[WriteEnd], "", 1);
#endif
}
bool PIWaitEvent::isCreate() const {
#ifdef WINDOWS
return event;
#else
return pipe_fd[ReadEnd] != 0;
#endif
}
void * PIWaitEvent::getEvent() const {
#ifdef WINDOWS
return event;
#else
return nullptr;
#endif
}

67
libs/main/core/piwaitevent_p.h Normal file
View File

@@ -0,0 +1,67 @@
/*
PIP - Platform Independent Primitives
Private PIP wait object
Ivan Pelipenko peri4ko@yandex.ru
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef PIWAITEVENT_P_H
#define PIWAITEVENT_P_H
#include "pibase.h"
#ifdef WINDOWS
# include <stdarg.h>
# include <windef.h>
# include <winbase.h>
#else
# include <unistd.h>
# include <sys/select.h>
#endif
class PIP_EXPORT PIWaitEvent {
public:
~PIWaitEvent();
enum CheckRole { // UNIX only
CheckRead,
CheckWrite,
CheckExeption
};
void create();
void destroy();
bool wait(int fd = -1, CheckRole role = CheckRead);
bool sleep(int us); // return if sleep done
void interrupt();
bool isCreate() const;
void * getEvent() const; // WINDOWS only
private:
#ifdef WINDOWS
void * event = nullptr;
#else
int pipe_fd[2] = {0, 0};
fd_set fds[3];
enum {
ReadEnd = 0,
WriteEnd = 1
};
#endif
};
#endif // PIWAITEVENT_P_H

20
libs/main/io_devices/pican.cpp
View File

@@ -18,7 +18,7 @@
*/ */
#include "pican.h" #include "pican.h"
#include "pipropertystorage.h" #include "pipropertystorage.h"
#include "piincludes_p.h" #include "piwaitevent_p.h"
#if !defined(WINDOWS) && !defined(MAC_OS) && !defined(MICRO_PIP) #if !defined(WINDOWS) && !defined(MAC_OS) && !defined(MICRO_PIP)
# define PIP_CAN # define PIP_CAN
#endif #endif
@@ -39,17 +39,24 @@
REGISTER_DEVICE(PICAN) REGISTER_DEVICE(PICAN)
PRIVATE_DEFINITION_START(PICAN)
PIWaitEvent event;
PRIVATE_DEFINITION_END(PICAN)
PICAN::PICAN(const PIString & path, PIIODevice::DeviceMode mode) : PIIODevice(path, mode) { PICAN::PICAN(const PIString & path, PIIODevice::DeviceMode mode) : PIIODevice(path, mode) {
setThreadedReadBufferSize(256); setThreadedReadBufferSize(256);
setPath(path); setPath(path);
can_id = 0; can_id = 0;
sock = 0; sock = 0;
PRIVATE->event.create();
} }
PICAN::~PICAN() { PICAN::~PICAN() {
stop(); stopAndWait();
close(); close();
PRIVATE->event.destroy();
} }
@@ -92,6 +99,7 @@ bool PICAN::openDevice() {
bool PICAN::closeDevice() { bool PICAN::closeDevice() {
#ifdef PIP_CAN #ifdef PIP_CAN
interrupt();
if (sock > 0) ::close(sock); if (sock > 0) ::close(sock);
#endif #endif
return true; return true;
@@ -103,9 +111,8 @@ ssize_t PICAN::readDevice(void * read_to, ssize_t max_size) {
//piCout << "PICAN read"; //piCout << "PICAN read";
can_frame frame; can_frame frame;
ssize_t ret = 0; ssize_t ret = 0;
reading_now = true; if (PRIVATE->event.wait(sock))
ret = ::read(sock, &frame, sizeof(can_frame)); ret = ::read(sock, &frame, sizeof(can_frame));
reading_now = false;
if (ret < 0) {/*piCoutObj << "Error while read CAN frame " << ret;*/ return -1;} if (ret < 0) {/*piCoutObj << "Error while read CAN frame " << ret;*/ return -1;}
//piCoutObj << "receive CAN frame Id =" << frame.can_id; //piCoutObj << "receive CAN frame Id =" << frame.can_id;
memcpy(read_to, frame.data, piMini(frame.can_dlc, max_size)); memcpy(read_to, frame.data, piMini(frame.can_dlc, max_size));
@@ -148,6 +155,11 @@ int PICAN::readedCANID() const {
} }
void PICAN::interrupt() {
PRIVATE->event.interrupt();
}
PIString PICAN::constructFullPathDevice() const { PIString PICAN::constructFullPathDevice() const {
PIString ret; PIString ret;
ret += path() + ":" + PIString::fromNumber(CANID(), 16); ret += path() + ":" + PIString::fromNumber(CANID(), 16);

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

@@ -39,6 +39,7 @@ public:
void setCANID(int id); void setCANID(int id);
int CANID() const; int CANID() const;
int readedCANID() const; int readedCANID() const;
void interrupt() override;
protected: protected:
bool openDevice() override; bool openDevice() override;
@@ -52,6 +53,7 @@ protected:
DeviceInfoFlags deviceInfoFlags() const override {return PIIODevice::Reliable;} DeviceInfoFlags deviceInfoFlags() const override {return PIIODevice::Reliable;}
private: private:
PRIVATE_DECLARATION(PIP_EXPORT)
int sock; int sock;
int can_id, readed_id; int can_id, readed_id;
}; };

310
libs/main/io_devices/piethernet.cpp
View File

@@ -67,6 +67,7 @@
# endif # endif
# endif # endif
#endif #endif
#include "piwaitevent_p.h"
#include <errno.h> #include <errno.h>
@@ -212,6 +213,7 @@ PRIVATE_DEFINITION_START(PIEthernet)
sockaddr_in addr_; sockaddr_in addr_;
sockaddr_in saddr_; sockaddr_in saddr_;
sockaddr_in raddr_; sockaddr_in raddr_;
PIWaitEvent event;
PRIVATE_DEFINITION_END(PIEthernet) PRIVATE_DEFINITION_END(PIEthernet)
@@ -240,15 +242,18 @@ PIEthernet::PIEthernet(int sock_, PIString ip_port): PIIODevice("", ReadWrite) {
setParameters(PIEthernet::ReuseAddress | PIEthernet::MulticastLoop); setParameters(PIEthernet::ReuseAddress | PIEthernet::MulticastLoop);
setType(TCP_Client, false); setType(TCP_Client, false);
setPath(ip_port); setPath(ip_port);
ethNonblocking(sock);
PRIVATE->event.create();
//piCoutObj << "new tcp client" << sock_; //piCoutObj << "new tcp client" << sock_;
} }
PIEthernet::~PIEthernet() { PIEthernet::~PIEthernet() {
//piCout << "~PIEthernet" << uint(this); //piCout << "~PIEthernet";
stop(); stopAndWait();
close(); close();
//piCoutObj << "~PIEthernet done"; PRIVATE->event.destroy();
//piCout << "~PIEthernet done";
} }
@@ -274,7 +279,9 @@ void PIEthernet::construct() {
bool PIEthernet::init() { bool PIEthernet::init() {
if (isOpened()) return true; if (isOpened()) return true;
//cout << "init " << type_ << endl; if (sock != -1) return true;
//piCout << "init " << type();
PRIVATE->event.destroy();
if (sock_s == sock) if (sock_s == sock)
sock_s = -1; sock_s = -1;
closeSocket(sock); closeSocket(sock);
@@ -289,6 +296,8 @@ bool PIEthernet::init() {
} }
PIFlags<Parameters> params = parameters(); PIFlags<Parameters> params = parameters();
sock = ::socket(AF_INET, st, pr); sock = ::socket(AF_INET, st, pr);
ethNonblocking(sock);
PRIVATE->event.create();
if (params[SeparateSockets]) if (params[SeparateSockets])
sock_s = ::socket(AF_INET, st, pr); sock_s = ::socket(AF_INET, st, pr);
else else
@@ -301,7 +310,7 @@ bool PIEthernet::init() {
if (params[PIEthernet::Broadcast]) ethSetsockoptBool(sock, SOL_SOCKET, SO_BROADCAST); if (params[PIEthernet::Broadcast]) ethSetsockoptBool(sock, SOL_SOCKET, SO_BROADCAST);
applyTimeouts(); applyTimeouts();
applyOptInt(IPPROTO_IP, IP_TTL, TTL()); applyOptInt(IPPROTO_IP, IP_TTL, TTL());
// piCoutObj << "inited" << path(); //piCoutObj << "inited" << path();
return true; return true;
} }
@@ -355,11 +364,12 @@ PIEthernet::Address PIEthernet::getBroadcast(const PIEthernet::Address & ip, con
bool PIEthernet::openDevice() { bool PIEthernet::openDevice() {
if (connected_) return true; if (connected_) return true;
//piCoutObj << "open";
init(); init();
if (sock == -1 || path().isEmpty()) return false; if (sock == -1 || path().isEmpty()) return false;
addr_r.set(path()); addr_r.set(path());
if (type() == TCP_Client) //if (type() == TCP_Client)
connecting_ = true; // connecting_ = true;
if (type() != UDP || mode() == PIIODevice::WriteOnly) if (type() != UDP || mode() == PIIODevice::WriteOnly)
return true; return true;
memset(&PRIVATE->addr_, 0, sizeof(PRIVATE->addr_)); memset(&PRIVATE->addr_, 0, sizeof(PRIVATE->addr_));
@@ -392,21 +402,26 @@ bool PIEthernet::openDevice() {
bool PIEthernet::closeDevice() { bool PIEthernet::closeDevice() {
//cout << "close\n"; //piCoutObj << "close";
if (server_thread_.isRunning()) { bool ned = connected_;
connected_ = connecting_ = false;
server_thread_.stop(); server_thread_.stop();
if (!server_thread_.waitForFinish(100)) PRIVATE->event.interrupt();
if (server_thread_.isRunning()) {
if (!server_thread_.waitForFinish(1000))
server_thread_.terminate(); server_thread_.terminate();
} }
PRIVATE->event.destroy();
if (sock_s == sock) if (sock_s == sock)
sock_s = -1; sock_s = -1;
closeSocket(sock); closeSocket(sock);
closeSocket(sock_s); closeSocket(sock_s);
while (!clients_.isEmpty()) while (!clients_.isEmpty())
delete clients_.back(); delete clients_.back();
bool ned = connected_; if (ned) {
connected_ = connecting_ = false; piCoutObj << "Disconnect on close";
if (ned) disconnected(false); disconnected(false);
}
return true; return true;
} }
@@ -556,6 +571,7 @@ bool PIEthernet::connect(bool threaded) {
connecting_ = true; connecting_ = true;
return true; return true;
} }
if (connected_) return false;
if (sock == -1) init(); if (sock == -1) init();
if (sock == -1) return false; if (sock == -1) return false;
memset(&PRIVATE->addr_, 0, sizeof(PRIVATE->addr_)); memset(&PRIVATE->addr_, 0, sizeof(PRIVATE->addr_));
@@ -566,13 +582,14 @@ bool PIEthernet::connect(bool threaded) {
#ifdef QNX #ifdef QNX
PRIVATE->addr_.sin_len = sizeof(PRIVATE->addr_); PRIVATE->addr_.sin_len = sizeof(PRIVATE->addr_);
#endif #endif
connected_ = (::connect(sock, (sockaddr * )&PRIVATE->addr_, sizeof(PRIVATE->addr_)) == 0); connecting_ = true;
connected_ = connectTCP();
connecting_ = false;
if (!connected_) { if (!connected_) {
piCoutObj << "Can`t connect to" << addr_r << "," << ethErrorString(); piCoutObj << "Can`t connect to" << addr_r << "," << ethErrorString();
} }
opened_ = connected_; opened_ = connected_;
if (connected_) { if (connected_) {
connecting_ = false;
connected(); connected();
} }
return connected_; return connected_;
@@ -618,7 +635,7 @@ bool PIEthernet::listen(bool threaded) {
return false; return false;
} }
opened_ = server_bounded = true; opened_ = server_bounded = true;
//piCoutObj << "listen on " << ip_ << ":" << port_; piCoutObj << "listen on" << path();
server_thread_.start(server_func); server_thread_.start(server_func);
return true; return true;
} }
@@ -675,40 +692,21 @@ bool PIEthernet::send(const PIEthernet::Address & addr, const PIByteArray & data
} }
void PIEthernet::interrupt() {
//piCout << "interrupt";
PRIVATE->event.interrupt();
}
ssize_t PIEthernet::readDevice(void * read_to, ssize_t max_size) { ssize_t PIEthernet::readDevice(void * read_to, ssize_t max_size) {
//piCout << "read" << sock; //piCout << "read" << sock;
if (sock == -1) init(); if (sock == -1) init();
if (sock == -1 || read_to == 0) return -1; if (sock == -1 || read_to == 0) return -1;
int rs = 0, s = 0, lerr = 0; int rs = 0, lerr = 0;
sockaddr_in client_addr; //piCoutObj << "read from " << path() << connecting_;
socklen_t slen = sizeof(client_addr);
// piCoutObj << "read from " << ip_ << ":" << port_;
switch (type()) { switch (type()) {
case TCP_SingleTCP:
reading_now = true;
::listen(sock, 64);
reading_now = false;
s = accept(sock, (sockaddr * )&client_addr, &slen);
if (s == -1) {
//piCoutObj << "Can`t accept new connection, " << ethErrorString();
piMinSleep();
return -1;
}
reading_now = true;
rs = ethRecv(s, read_to, max_size);
reading_now = false;
closeSocket(s);
return rs;
case TCP_Client: case TCP_Client:
if (connecting_) { if (connecting_) {
#ifdef ANDROID
/*if (sock_s == sock)
sock_s = -1;
closeSocket(sock);
closeSocket(sock_s);
init();
qDebug() << "init() in read thread";*/
#endif
addr_r.set(path()); addr_r.set(path());
memset(&PRIVATE->addr_, 0, sizeof(PRIVATE->addr_)); memset(&PRIVATE->addr_, 0, sizeof(PRIVATE->addr_));
PRIVATE->addr_.sin_port = htons(addr_r.port()); PRIVATE->addr_.sin_port = htons(addr_r.port());
@@ -717,11 +715,9 @@ ssize_t PIEthernet::readDevice(void * read_to, ssize_t max_size) {
#ifdef QNX #ifdef QNX
PRIVATE->addr_.sin_len = sizeof(PRIVATE->addr_); PRIVATE->addr_.sin_len = sizeof(PRIVATE->addr_);
#endif #endif
//piCout << "connect to " << path() << "..."; piCoutObj << "connect to " << path() << "...";
reading_now = true; connected_ = connectTCP();
connected_ = (::connect(sock, (sockaddr * )&(PRIVATE->addr_), sizeof(PRIVATE->addr_)) == 0); piCoutObj << "connect to " << path() << connected_;
reading_now = false;
//piCout << "connect to " << path() << connected_;
if (!connected_) if (!connected_)
piCoutObj << "Can`t connect to" << addr_r << "," << ethErrorString(); piCoutObj << "Can`t connect to" << addr_r << "," << ethErrorString();
opened_ = connected_; opened_ = connected_;
@@ -734,9 +730,25 @@ ssize_t PIEthernet::readDevice(void * read_to, ssize_t max_size) {
} }
if (!connected_) return -1; if (!connected_) return -1;
errorClear(); errorClear();
reading_now = true; #ifdef WINDOWS
{
long wr = waitForEvent(PRIVATE->event, FD_READ | FD_CLOSE);
switch (wr) {
case FD_READ:
//piCout << "fd_read ...";
rs = ethRecv(sock, read_to, max_size); rs = ethRecv(sock, read_to, max_size);
reading_now = false; break;
case FD_CLOSE:
//piCout << "fd_close ...";
rs = -1;
break;
default: break;
}
}
#else
if (PRIVATE->event.wait(sock))
rs = ethRecv(sock, read_to, max_size);
#endif
//piCoutObj << "readed" << rs; //piCoutObj << "readed" << rs;
if (rs <= 0) { if (rs <= 0) {
lerr = ethErrorCore(); lerr = ethErrorCore();
@@ -744,30 +756,47 @@ ssize_t PIEthernet::readDevice(void * read_to, ssize_t max_size) {
#ifdef WINDOWS #ifdef WINDOWS
if ((lerr == WSAEWOULDBLOCK || lerr == WSAETIMEDOUT) && !parameters()[DisonnectOnTimeout]) { if ((lerr == WSAEWOULDBLOCK || lerr == WSAETIMEDOUT) && !parameters()[DisonnectOnTimeout]) {
#elif defined(ANDROID) #elif defined(ANDROID)
if ((lerr == EWOULDBLOCK || lerr == EAGAIN || lerr == EINTR) && !parameters()[DisonnectOnTimeout]) { if ((lerr == EWOULDBLOCK || lerr == EAGAIN || lerr == EINTR) && !parameters()[DisonnectOnTimeout] && rs < 0) {
#else #else
if ((lerr == EWOULDBLOCK || lerr == EAGAIN) && !parameters()[DisonnectOnTimeout]) { if ((lerr == EWOULDBLOCK || lerr == EAGAIN) && !parameters()[DisonnectOnTimeout] && rs < 0) {
#endif #endif
//piCoutObj << errorString(); //piCoutObj << errorString();
return -1; return -1;
} }
if (connected_) { if (connected_.exchange(false)) {
opened_ = false;
piCoutObj << "Disconnect on read," << ethErrorString(); piCoutObj << "Disconnect on read," << ethErrorString();
opened_ = connected_ = false; closeSocket(sock);
init(); init();
disconnected(rs < 0); disconnected(rs < 0);
} }
if (parameters()[KeepConnection]) if (parameters()[KeepConnection]) {
connect(); connect();
}
//piCoutObj << "eth" << ip_ << "disconnected"; //piCoutObj << "eth" << ip_ << "disconnected";
} }
if (rs > 0) received(read_to, rs); if (rs > 0) received(read_to, rs);
return rs; return rs;
case UDP: case UDP: {
memset(&PRIVATE->raddr_, 0, sizeof(PRIVATE->raddr_)); memset(&PRIVATE->raddr_, 0, sizeof(PRIVATE->raddr_));
reading_now = true; //piCoutObj << "read from" << path() << "...";
#ifdef WINDOWS
long wr = waitForEvent(PRIVATE->event, FD_READ | FD_CLOSE);
switch (wr) {
case FD_READ:
//piCout << "fd_read ...";
rs = ethRecvfrom(sock, read_to, max_size, 0, (sockaddr*)&PRIVATE->raddr_); rs = ethRecvfrom(sock, read_to, max_size, 0, (sockaddr*)&PRIVATE->raddr_);
reading_now = false; break;
case FD_CLOSE:
//piCout << "fd_close ...";
rs = -1;
break;
default: break;
}
#else
rs = ethRecvfrom(sock, read_to, max_size, 0, (sockaddr*)&PRIVATE->raddr_);
#endif
//piCoutObj << "read from" << path() << rs << "bytes";
if (rs > 0) { if (rs > 0) {
addr_lr.set(uint(PRIVATE->raddr_.sin_addr.s_addr), ntohs(PRIVATE->raddr_.sin_port)); addr_lr.set(uint(PRIVATE->raddr_.sin_addr.s_addr), ntohs(PRIVATE->raddr_.sin_port));
//piCoutObj << "read from" << ip_r << ":" << port_r << rs << "bytes"; //piCoutObj << "read from" << ip_r << ":" << port_r << rs << "bytes";
@@ -775,6 +804,7 @@ ssize_t PIEthernet::readDevice(void * read_to, ssize_t max_size) {
} }
//else piCoutObj << "read returt" << rs << ", error" << ethErrorString(); //else piCoutObj << "read returt" << rs << ", error" << ethErrorString();
return rs; return rs;
}
default: break; default: break;
} }
return -1; return -1;
@@ -790,26 +820,6 @@ ssize_t PIEthernet::writeDevice(const void * data, ssize_t max_size) {
//piCoutObj << "sending to " << ip_s << ":" << port_s << " " << max_size << " bytes"; //piCoutObj << "sending to " << ip_s << ":" << port_s << " " << max_size << " bytes";
int ret = 0; int ret = 0;
switch (type()) { switch (type()) {
case TCP_SingleTCP:
memset(&PRIVATE->addr_, 0, sizeof(PRIVATE->addr_));
PRIVATE->addr_.sin_port = htons(addr_s.port());
PRIVATE->addr_.sin_addr.s_addr = addr_s.ip();
PRIVATE->addr_.sin_family = AF_INET;
#ifdef QNX
PRIVATE->addr_.sin_len = sizeof(PRIVATE->addr_);
#endif
//piCoutObj << "connect SingleTCP" << ip_s << ":" << port_s << "...";
if (::connect(sock, (sockaddr * )&PRIVATE->addr_, sizeof(PRIVATE->addr_)) != 0) {
//piCoutObj << "Can`t connect to " << ip_s << ":" << port_s << ", " << ethErrorString();
piMinSleep();
return -1;
}
//piCoutObj << "ok, write SingleTCP" << int(data) << max_size << "bytes ...";
ret = ::send(sock, (const char *)data, max_size, 0);
//piCoutObj << "ok, ret" << ret;
closeSocket(sock);
init();
return ret;
case UDP: case UDP:
PRIVATE->saddr_.sin_port = htons(addr_s.port()); PRIVATE->saddr_.sin_port = htons(addr_s.port());
PRIVATE->saddr_.sin_addr.s_addr = addr_s.ip(); PRIVATE->saddr_.sin_addr.s_addr = addr_s.ip();
@@ -823,7 +833,7 @@ ssize_t PIEthernet::writeDevice(const void * data, ssize_t max_size) {
#endif #endif
, (sockaddr * )&PRIVATE->saddr_, sizeof(PRIVATE->saddr_)); , (sockaddr * )&PRIVATE->saddr_, sizeof(PRIVATE->saddr_));
//piCout << "[PIEth] write to" << ip_s << ":" << port_s << "ok"; //piCout << "[PIEth] write to" << ip_s << ":" << port_s << "ok";
case TCP_Client: case TCP_Client: {
if (connecting_) { if (connecting_) {
memset(&PRIVATE->addr_, 0, sizeof(PRIVATE->addr_)); memset(&PRIVATE->addr_, 0, sizeof(PRIVATE->addr_));
addr_r.set(path()); addr_r.set(path());
@@ -834,7 +844,7 @@ ssize_t PIEthernet::writeDevice(const void * data, ssize_t max_size) {
PRIVATE->addr_.sin_len = sizeof(PRIVATE->addr_); PRIVATE->addr_.sin_len = sizeof(PRIVATE->addr_);
#endif #endif
//piCoutObj << "connect to " << ip << ":" << port_; //piCoutObj << "connect to " << ip << ":" << port_;
connected_ = (::connect(sock, (sockaddr * )&PRIVATE->addr_, sizeof(PRIVATE->addr_)) == 0); connected_ = connectTCP();
if (!connected_) if (!connected_)
piCoutObj << "Can`t connect to" << addr_r << "," << ethErrorString(); piCoutObj << "Can`t connect to" << addr_r << "," << ethErrorString();
opened_ = connected_; opened_ = connected_;
@@ -844,14 +854,46 @@ ssize_t PIEthernet::writeDevice(const void * data, ssize_t max_size) {
} }
} }
if (!connected_) return -1; if (!connected_) return -1;
ret = ::send(sock, (const char *)data, max_size, 0); auto disconnectFunc = [this](){
if (ret < 0) { if (connected_.exchange(false)) {
opened_ = false;
piCoutObj << "Disconnect on write," << ethErrorString(); piCoutObj << "Disconnect on write," << ethErrorString();
opened_ = connected_ = false; closeSocket(sock);
init(); init();
disconnected(true); disconnected(true);
} }
return ret; };
if (!isOptionSet(BlockingWrite)) {
ret = ::send(sock, (const char *)data, max_size, 0);
if (ret < 0) {
disconnectFunc();
return -1;
}
} else {
ssize_t remain_size = max_size;
const char * remain_data = (const char *)data;
while (remain_size > 0) {
int sr = ::send(sock, remain_data, remain_size, 0);
if (sr < 0) {
int err = ethErrorCore();
#ifdef WINDOWS
if (err == WSAEWOULDBLOCK) {
#else
if (err == EAGAIN || err == EWOULDBLOCK) {
#endif
piMinSleep();
//piCoutObj << "wait for write";
continue;
} else {
disconnectFunc();
return -1;
}
}
remain_data += sr;
remain_size -= sr;
}
}
return ret;}
default: break; default: break;
} }
return -1; return -1;
@@ -862,8 +904,7 @@ PIIODevice::DeviceInfoFlags PIEthernet::deviceInfoFlags() const {
switch (type()) { switch (type()) {
case UDP: return 0; case UDP: return 0;
case TCP_Client: case TCP_Client:
case TCP_Server: case TCP_Server: return Sequential | Reliable;
case TCP_SingleTCP: return Sequential | Reliable;
default: break; default: break;
} }
return 0; return 0;
@@ -890,7 +931,21 @@ void PIEthernet::server_func(void * eth) {
} }
sockaddr_in client_addr; sockaddr_in client_addr;
socklen_t slen = sizeof(client_addr); socklen_t slen = sizeof(client_addr);
#ifdef WINDOWS
long wr = ce->waitForEvent(ce->PRIVATEWB->event, FD_ACCEPT | FD_CLOSE);
if (wr != FD_ACCEPT) {
piMSleep(10);
return;
}
#else
if (!ce->PRIVATEWB->event.wait(ce->sock)) {
piMSleep(10);
return;
}
#endif
//piCout << "server" << "accept ...";
int s = accept(ce->sock, (sockaddr * )&client_addr, &slen); int s = accept(ce->sock, (sockaddr * )&client_addr, &slen);
//piCout << "server" << "accept done" << ethErrorString();
if (s == -1) { if (s == -1) {
int lerr = ethErrorCore(); int lerr = ethErrorCore();
#ifdef WINDOWS #ifdef WINDOWS
@@ -930,6 +985,49 @@ void PIEthernet::setType(Type t, bool reopen) {
} }
bool PIEthernet::connectTCP() {
::connect(sock, (sockaddr * )&(PRIVATE->addr_), sizeof(PRIVATE->addr_));
//piCout << errorString();
#ifdef WINDOWS
long wr = waitForEvent(PRIVATE->event, FD_CONNECT | FD_CLOSE);
switch (wr) {
case FD_CONNECT:
//piCout << "fd_connect ...";
return ethIsWriteable(sock);
default: break;
}
#else
if (PRIVATE->event.wait(sock, PIWaitEvent::CheckWrite)) {
if (ethIsWriteable(sock)) return true;
else {
closeSocket(sock);
init();
}
}
#endif
return false;
}
#ifdef WINDOWS
long PIEthernet::waitForEvent(PIWaitEvent & event, long mask) {
if (!event.isCreate() || sock < 0) return 0;
WSAEventSelect(sock, event.getEvent(), mask);
if (event.wait()) {
//DWORD wr = WSAWaitForMultipleEvents(1, &(PRIVATE->read_event), FALSE, WSA_INFINITE, TRUE);
//piCout << "wait result" << wr;
//if (wr == WSA_WAIT_EVENT_0) {
WSANETWORKEVENTS events;
memset(&events, 0, sizeof(events));
WSAEnumNetworkEvents(sock, event.getEvent(), &events);
//piCout << "wait result" << events.lNetworkEvents;
return events.lNetworkEvents;
}
return 0;
}
#endif
bool PIEthernet::configureDevice(const void * e_main, const void * e_parent) { bool PIEthernet::configureDevice(const void * e_main, const void * e_parent) {
PIConfig::Entry * em = (PIConfig::Entry * )e_main; PIConfig::Entry * em = (PIConfig::Entry * )e_main;
PIConfig::Entry * ep = (PIConfig::Entry * )e_parent; PIConfig::Entry * ep = (PIConfig::Entry * )e_parent;
@@ -1333,3 +1431,43 @@ int PIEthernet::ethSetsockoptBool(int sock, int level, int optname, bool value)
so = (value ? 1 : 0); so = (value ? 1 : 0);
return ethSetsockopt(sock, level, optname, &so, sizeof(so)); return ethSetsockopt(sock, level, optname, &so, sizeof(so));
} }
void PIEthernet::ethNonblocking(int sock) {
if (sock < 0) return;
#ifdef WINDOWS
u_long mode = 1;
ioctlsocket(sock, FIONBIO, &mode);
#else
fcntl(sock, F_SETFL, O_NONBLOCK);
#endif
}
bool PIEthernet::ethIsWriteable(int sock) {
/* fd_set fd_test;
FD_ZERO(&fd_test);
FD_SET(sock, &fd_test);
int fds = 0;
#ifndef WINDOWS
fds = sock + 1;
#endif
timeval timeout;
timeout.tv_sec = timeout.tv_usec = 0;
::select(fds, nullptr, &fd_test, nullptr, &timeout);
return FD_ISSET(sock, &fd_test);*/
#ifdef WINDOWS
fd_set fd_test;
FD_ZERO(&fd_test);
FD_SET(sock, &fd_test);
timeval timeout;
timeout.tv_sec = timeout.tv_usec = 0;
::select(0, nullptr, &fd_test, nullptr, &timeout);
return FD_ISSET(sock, &fd_test);
#else
int ret = 0;
socklen_t len = sizeof(ret);
getsockopt(sock, SOL_SOCKET, SO_ERROR, (char*)&ret, &len);
return ret == 0;
#endif
}

13
libs/main/io_devices/piethernet.h
View File

@@ -49,8 +49,7 @@ public:
enum Type { enum Type {
UDP /** UDP - User Datagram Protocol */ , UDP /** UDP - User Datagram Protocol */ ,
TCP_Client /** TCP client - allow connection to TCP server */ , TCP_Client /** TCP client - allow connection to TCP server */ ,
TCP_Server /** TCP server - receive connections from TCP clients */ , TCP_Server /** TCP server - receive connections from TCP clients */
TCP_SingleTCP /** TCP client single mode - connect & send & disconnect, on each packet */
}; };
//! \brief Parameters of %PIEthernet //! \brief Parameters of %PIEthernet
@@ -315,6 +314,8 @@ public:
bool canWrite() const override {return mode() & WriteOnly;} bool canWrite() const override {return mode() & WriteOnly;}
void interrupt() override;
int socket() const {return sock;} int socket() const {return sock;}
EVENT1(newConnection, PIEthernet * , client); EVENT1(newConnection, PIEthernet * , client);
@@ -487,7 +488,7 @@ protected:
PRIVATE_DECLARATION(PIP_EXPORT) PRIVATE_DECLARATION(PIP_EXPORT)
int sock, sock_s; int sock, sock_s;
bool connected_, connecting_, listen_threaded, server_bounded; std::atomic_bool connected_, connecting_, listen_threaded, server_bounded;
mutable Address addr_r, addr_s, addr_lr; mutable Address addr_r, addr_s, addr_lr;
Type eth_type; Type eth_type;
PIThread server_thread_; PIThread server_thread_;
@@ -500,6 +501,10 @@ private:
EVENT_HANDLER1(void, clientDeleted, PIObject *, o); EVENT_HANDLER1(void, clientDeleted, PIObject *, o);
static void server_func(void * eth); static void server_func(void * eth);
void setType(Type t, bool reopen = true); void setType(Type t, bool reopen = true);
bool connectTCP();
#ifdef WINDOWS
long waitForEvent(PIWaitEvent & event, long mask);
#endif
static int ethErrorCore(); static int ethErrorCore();
static PIString ethErrorString(); static PIString ethErrorString();
@@ -510,6 +515,8 @@ private:
static int ethSetsockopt(int sock, int level, int optname, const void * optval, int optlen); static int ethSetsockopt(int sock, int level, int optname, const void * optval, int optlen);
static int ethSetsockoptInt(int sock, int level, int optname, int value = 1); static int ethSetsockoptInt(int sock, int level, int optname, int value = 1);
static int ethSetsockoptBool(int sock, int level, int optname, bool value = true); static int ethSetsockoptBool(int sock, int level, int optname, bool value = true);
static void ethNonblocking(int sock);
static bool ethIsWriteable(int sock);
}; };

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

@@ -439,9 +439,7 @@ PIByteArray PIFile::get() {
ssize_t PIFile::readDevice(void * read_to, ssize_t max_size) { ssize_t PIFile::readDevice(void * read_to, ssize_t max_size) {
if (!canRead() || PRIVATE->fd == 0) return -1; if (!canRead() || PRIVATE->fd == 0) return -1;
reading_now = true;
ssize_t ret = fread(read_to, 1, max_size, PRIVATE->fd); ssize_t ret = fread(read_to, 1, max_size, PRIVATE->fd);
reading_now = false;
return ret; return ret;
} }

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

@@ -131,7 +131,8 @@ PIIODevice::PIIODevice(const PIString & path, PIIODevice::DeviceMode mode): PIOb
PIIODevice::~PIIODevice() { PIIODevice::~PIIODevice() {
stop(); destroying = true;
stopAndWait();
} }
@@ -213,11 +214,11 @@ void PIIODevice::stopThreadedRead() {
#ifdef MICRO_PIP #ifdef MICRO_PIP
read_thread.stop(); read_thread.stop();
#else #else
if (reading_now) {
read_thread.terminate();
reading_now = false;
} else {
read_thread.stop(); read_thread.stop();
if (!destroying) {
interrupt();
} else {
piCoutObj << "Error: Device is running after destructor!";
} }
#endif #endif
} }
@@ -269,6 +270,13 @@ void PIIODevice::stop() {
} }
void PIIODevice::stopAndWait(int timeout_ms) {
stop();
waitThreadedReadFinished(timeout_ms);
waitThreadedWriteFinished(timeout_ms);
}
ssize_t PIIODevice::read(void * read_to, ssize_t max_size) { ssize_t PIIODevice::read(void * read_to, ssize_t max_size) {
ssize_t ret = readDevice(read_to, max_size); ssize_t ret = readDevice(read_to, max_size);
return ret; return ret;
@@ -296,7 +304,6 @@ ssize_t PIIODevice::write(const void * data, ssize_t max_size) {
void PIIODevice::_init() { void PIIODevice::_init() {
reading_now = false;
setOptions(0); setOptions(0);
setReopenEnabled(true); setReopenEnabled(true);
setReopenTimeout(1000); setReopenTimeout(1000);
@@ -630,7 +637,3 @@ void PIIODevice::configureFromVariantDevice(const PIPropertyStorage & d) {
} }
void PIIODevice::softStopThreadedRead() {
read_thread.stop();
}

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

@@ -277,6 +277,14 @@ public:
//! \~russian Останавливает потоковое чтение и запись. //! \~russian Останавливает потоковое чтение и запись.
void stop(); void stop();
//! \~english Stop both threaded read and threaded write and wait for finish.
//! \~russian Останавливает потоковое чтение и запись и ожидает завершения.
void stopAndWait(int timeout_ms = -1);
//! \~english Interrupt blocking operation.
//! \~russian Прерывает блокирующую операцию.
virtual void interrupt() {}
//! \~english Read from device maximum "max_size" bytes to "read_to" //! \~english Read from device maximum "max_size" bytes to "read_to"
//! \~russian Читает из устройства не более "max_size" байт в "read_to" //! \~russian Читает из устройства не более "max_size" байт в "read_to"
@@ -382,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);
@@ -537,10 +543,6 @@ protected:
bool opened_ = false; bool opened_ = false;
void * ret_data_ = nullptr; void * ret_data_ = nullptr;
//! \~english Set this flag while blocking operations
//! \~russian Устанавливайте этот флаг во время блокирующих операций
std::atomic_bool reading_now;
private: private:
EVENT_HANDLER(void, read_func); EVENT_HANDLER(void, read_func);
EVENT_HANDLER(void, write_func); EVENT_HANDLER(void, write_func);
@@ -557,7 +559,7 @@ private:
PIQueue<PIPair<PIByteArray, ullong> > write_queue; PIQueue<PIPair<PIByteArray, ullong> > write_queue;
ullong tri = 0; ullong tri = 0;
uint threaded_read_buffer_size, reopen_timeout = 1000; uint threaded_read_buffer_size, reopen_timeout = 1000;
bool reopen_enabled = true; bool reopen_enabled = true, destroying = false;
static PIMutex nfp_mutex; static PIMutex nfp_mutex;
static PIMap<PIString, PIString> nfp_cache; static PIMap<PIString, PIString> nfp_cache;

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

@@ -192,7 +192,7 @@ PIPeer::~PIPeer() {
if (p._data) { if (p._data) {
p._data->dt_in.stop(); p._data->dt_in.stop();
p._data->dt_out.stop(); p._data->dt_out.stop();
p._data->t.stop(true); p._data->t.stopAndWait();
} }
destroyEths(); destroyEths();
piForeach (PIEthernet * i, eths_mcast) { piForeach (PIEthernet * i, eths_mcast) {

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

@@ -22,6 +22,7 @@
#include "piconfig.h" #include "piconfig.h"
#include "pidir.h" #include "pidir.h"
#include "pipropertystorage.h" #include "pipropertystorage.h"
#include "piwaitevent_p.h"
#include <errno.h> #include <errno.h>
#if defined(MICRO_PIP) #if defined(MICRO_PIP)
@@ -169,24 +170,22 @@ REGISTER_DEVICE(PISerial)
PRIVATE_DEFINITION_START(PISerial) PRIVATE_DEFINITION_START(PISerial)
PIWaitEvent event;
#ifdef WINDOWS #ifdef WINDOWS
PIWaitEvent event_write;
DCB desc, sdesc; DCB desc, sdesc;
void * hCom; HANDLE hCom = nullptr;
DWORD readed, mask; DWORD readed = 0, mask = 0;
OVERLAPPED overlap, overlap_write;
#else #else
termios desc, sdesc; termios desc, sdesc;
uint readed; uint readed = 0;
#endif #endif
PRIVATE_DEFINITION_END(PISerial) PRIVATE_DEFINITION_END(PISerial)
PISerial::DeviceInfo::DeviceInfo() {
vID = pID = 0;
}
PIString PISerial::DeviceInfo::id() const { PIString PISerial::DeviceInfo::id() const {
return PIString::fromNumber(vID, 16).toLowerCase().expandLeftTo(4, '0') + ":" + return PIString::fromNumber(vID, 16).toLowerCase().expandLeftTo(4, '0') + ":" +
PIString::fromNumber(pID, 16).toLowerCase().expandLeftTo(4, '0'); PIString::fromNumber(pID, 16).toLowerCase().expandLeftTo(4, '0');
@@ -209,19 +208,18 @@ PISerial::PISerial(const PIString & device_, PISerial::Speed speed_, PIFlags<PIS
PISerial::~PISerial() { PISerial::~PISerial() {
stop(); stopAndWait();
close(); close();
PRIVATE->event.destroy();
#ifdef WINDOWS
PRIVATE->event_write.destroy();
#endif
} }
void PISerial::construct() { void PISerial::construct() {
#ifdef WINDOWS
PRIVATE->hCom = 0;
#endif
fd = -1;
//setPriority(piHigh);
vtime = 10;
sending = false; sending = false;
//setPriority(piHigh);
setParameters(0); setParameters(0);
setSpeed(S115200); setSpeed(S115200);
setDataBitsCount(8); setDataBitsCount(8);
@@ -628,6 +626,15 @@ bool PISerial::send(const void * data, int size) {
} }
void PISerial::interrupt() {
//piCoutObj << "interrupt";
PRIVATE->event.interrupt();
#ifdef WINDOWS
PRIVATE->event_write.interrupt();
#endif
}
bool PISerial::openDevice() { bool PISerial::openDevice() {
PIString p = path(); PIString p = path();
//piCout << "ser open" << p; //piCout << "ser open" << p;
@@ -651,7 +658,7 @@ bool PISerial::openDevice() {
if (isReadable()) {ds |= GENERIC_READ; sm |= FILE_SHARE_READ;} if (isReadable()) {ds |= GENERIC_READ; sm |= FILE_SHARE_READ;}
if (isWriteable()) {ds |= GENERIC_WRITE; sm |= FILE_SHARE_WRITE;} if (isWriteable()) {ds |= GENERIC_WRITE; sm |= FILE_SHARE_WRITE;}
PIString wp = "//./" + p; PIString wp = "//./" + p;
PRIVATE->hCom = CreateFileA(wp.dataAscii(), ds, sm, 0, OPEN_EXISTING, FILE_ATTRIBUTE_SYSTEM, 0); PRIVATE->hCom = CreateFileA(wp.dataAscii(), ds, sm, 0, OPEN_EXISTING, FILE_ATTRIBUTE_SYSTEM | FILE_FLAG_OVERLAPPED, 0);
if (PRIVATE->hCom == INVALID_HANDLE_VALUE) { if (PRIVATE->hCom == INVALID_HANDLE_VALUE) {
piCoutObj << "Unable to open \"" << p << "\""; piCoutObj << "Unable to open \"" << p << "\"";
fd = -1; fd = -1;
@@ -675,6 +682,11 @@ bool PISerial::openDevice() {
//piCoutObj << "Initialized " << p; //piCoutObj << "Initialized " << p;
#endif #endif
applySettings(); applySettings();
PRIVATE->event.create();
#ifdef WINDOWS
PRIVATE->event_write.create();
#endif
return true; return true;
} }
@@ -696,6 +708,10 @@ bool PISerial::closeDevice() {
#endif #endif
fd = -1; fd = -1;
} }
PRIVATE->event.destroy();
#ifdef WINDOWS
PRIVATE->event_write.destroy();
#endif
return true; return true;
} }
@@ -796,15 +812,22 @@ ssize_t PISerial::readDevice(void * read_to, ssize_t max_size) {
#ifdef WINDOWS #ifdef WINDOWS
if (!canRead()) return -1; if (!canRead()) return -1;
if (sending) return -1; if (sending) return -1;
// piCoutObj << "com event ...";
//piCoutObj << "read ..." << PRIVATE->hCom; //piCoutObj << "read ..." << PRIVATE->hCom;
reading_now = true; memset(&(PRIVATE->overlap), 0, sizeof(PRIVATE->overlap));
ReadFile(PRIVATE->hCom, read_to, max_size, &PRIVATE->readed, 0); PRIVATE->overlap.hEvent = PRIVATE->event.getEvent();
reading_now = false; ReadFile(PRIVATE->hCom, read_to, max_size, NULL, &(PRIVATE->overlap));
PRIVATE->readed = 0;
if (PRIVATE->event.wait()) {
GetOverlappedResult(PRIVATE->hCom, &(PRIVATE->overlap), &(PRIVATE->readed), FALSE);
} else
return -1;
//piCoutObj << "read done" << PRIVATE->readed;
DWORD err = GetLastError(); DWORD err = GetLastError();
//piCout << err; if (err == ERROR_TIMEOUT && PRIVATE->readed == 0)
return 0;
if (err == ERROR_BAD_COMMAND || err == ERROR_ACCESS_DENIED) { if (err == ERROR_BAD_COMMAND || err == ERROR_ACCESS_DENIED) {
softStopThreadedRead(); piCoutObj << "read error" << (PRIVATE->readed) << errorString();
stop();
close(); close();
return 0; return 0;
} }
@@ -812,13 +835,12 @@ ssize_t PISerial::readDevice(void * read_to, ssize_t max_size) {
return PRIVATE->readed; return PRIVATE->readed;
#else #else
if (!canRead()) return -1; if (!canRead()) return -1;
reading_now = true; if (!PRIVATE->event.wait(fd)) return -1;
ssize_t ret = ::read(fd, read_to, max_size); ssize_t ret = ::read(fd, read_to, max_size);
reading_now = false;
if (ret < 0) { if (ret < 0) {
int err = errno; int err = errno;
if (err == EBADF || err == EFAULT || err == EINVAL || err == EIO) { if (err == EBADF || err == EFAULT || err == EINVAL || err == EIO) {
softStopThreadedRead(); stopThreadedRead();
close(); close();
return 0; return 0;
} }
@@ -834,12 +856,17 @@ ssize_t PISerial::writeDevice(const void * data, ssize_t max_size) {
return -1; return -1;
} }
#ifdef WINDOWS #ifdef WINDOWS
DWORD wrote; DWORD wrote(0);
// piCoutObj << "send ...";// << max_size;// << ": " << PIString((char*)data, max_size); //piCoutObj << "send ..." << max_size;// << ": " << PIString((char*)data, max_size);
sending = true; sending = true;
WriteFile(PRIVATE->hCom, data, max_size, &wrote, 0); memset(&(PRIVATE->overlap_write), 0, sizeof(PRIVATE->overlap_write));
PRIVATE->overlap_write.hEvent = PRIVATE->event_write.getEvent();
WriteFile(PRIVATE->hCom, data, max_size, NULL, &(PRIVATE->overlap_write));
if (PRIVATE->event_write.wait()) {
GetOverlappedResult(PRIVATE->hCom, &(PRIVATE->overlap_write), &wrote, FALSE);
}
sending = false; sending = false;
// piCoutObj << "send ok";// << wrote << " bytes in " << path(); //piCoutObj << "send ok" << wrote;// << " bytes in " << path();
#else #else
ssize_t wrote; ssize_t wrote;
wrote = ::write(fd, data, max_size); wrote = ::write(fd, data, max_size);
@@ -1150,7 +1177,7 @@ PIVector<PISerial::DeviceInfo> PISerial::availableDevicesInfo(bool test) {
if (test) { if (test) {
for (int i = 0; i < ret.size_s(); ++i) { for (int i = 0; i < ret.size_s(); ++i) {
#ifdef WINDOWS #ifdef WINDOWS
void * hComm = CreateFileA(ret[i].path.dataAscii(), GENERIC_READ, FILE_SHARE_READ, 0, OPEN_EXISTING, FILE_ATTRIBUTE_SYSTEM, 0); void * hComm = CreateFileA(ret[i].path.dataAscii(), GENERIC_READ, FILE_SHARE_READ, 0, OPEN_EXISTING, FILE_ATTRIBUTE_SYSTEM | FILE_FLAG_OVERLAPPED, 0);
if (hComm == INVALID_HANDLE_VALUE) { if (hComm == INVALID_HANDLE_VALUE) {
#else #else
int fd = ::open(ret[i].path.dataAscii(), O_NOCTTY | O_RDONLY); int fd = ::open(ret[i].path.dataAscii(), O_NOCTTY | O_RDONLY);

12
libs/main/io_devices/piserial.h
View File

@@ -90,19 +90,17 @@ public:
//! \~english Information about serial device //! \~english Information about serial device
//! \~russian Информация о последовательном устройстве //! \~russian Информация о последовательном устройстве
struct PIP_EXPORT DeviceInfo { struct PIP_EXPORT DeviceInfo {
DeviceInfo();
//! \~english Returns string representation of USB ID in format "xxxx:xxxx" (vID:pID) //! \~english Returns string representation of USB ID in format "xxxx:xxxx" (vID:pID)
//! \~russian Возвращает строковое представление USB ID в формате "xxxx:xxxx" (vID:pID) //! \~russian Возвращает строковое представление USB ID в формате "xxxx:xxxx" (vID:pID)
PIString id() const; PIString id() const;
//! \~english USB Vendor ID //! \~english USB Vendor ID
//! \~russian USB Vendor ID //! \~russian USB Vendor ID
uint vID; uint vID = 0;
//! \~english USB Product ID //! \~english USB Product ID
//! \~russian USB Product ID //! \~russian USB Product ID
uint pID; uint pID = 0;
//! \~english Path to device, e.g. "COM2" or "/dev/ttyUSB0" //! \~english Path to device, e.g. "COM2" or "/dev/ttyUSB0"
//! \~russian Путь к устройству, например "COM2" или "/dev/ttyUSB0" //! \~russian Путь к устройству, например "COM2" или "/dev/ttyUSB0"
@@ -240,6 +238,8 @@ public:
//! \~russian Пишет в порт байтовый массив "data". Возвращает если количество записанных байт = размер "data" //! \~russian Пишет в порт байтовый массив "data". Возвращает если количество записанных байт = размер "data"
bool send(const PIByteArray & data) {return send(data.data(), data.size_s());} bool send(const PIByteArray & data) {return send(data.data(), data.size_s());}
void interrupt() override;
//! \~english Returns all available speeds for serial devices //! \~english Returns all available speeds for serial devices
//! \~russian Возвращает все возможные скорости для устройств //! \~russian Возвращает все возможные скорости для устройств
static PIVector<int> availableSpeeds(); static PIVector<int> availableSpeeds();
@@ -310,8 +310,8 @@ protected:
bool closeDevice() override; bool closeDevice() override;
PRIVATE_DECLARATION(PIP_EXPORT) PRIVATE_DECLARATION(PIP_EXPORT)
int fd, vtime; int fd = -1, vtime = 10;
bool sending; std::atomic_bool sending;
PITimeMeasurer tm_; PITimeMeasurer tm_;
}; };

10
libs/main/io_utils/piconnection.cpp
View File

@@ -1114,6 +1114,9 @@ PIVector<PIIODevice * > PIConnection::DevicePool::boundedDevices(const PIConnect
PIConnection::DevicePool::DeviceData::~DeviceData() { PIConnection::DevicePool::DeviceData::~DeviceData() {
if (rthread) { if (rthread) {
rthread->stop();
if (dev) dev->interrupt();
if (!rthread->waitForFinish(1000))
rthread->terminate(); rthread->terminate();
delete rthread; delete rthread;
rthread = nullptr; rthread = nullptr;
@@ -1146,8 +1149,13 @@ void __DevicePool_threadReadDP(void * ddp) {
} }
if (dev->isClosed()) { if (dev->isClosed()) {
if (!dev->open()) { if (!dev->open()) {
piMSleep(dev->reopenTimeout()); PITimeMeasurer tm;
int timeout = dev->reopenTimeout();
while (tm.elapsed_m() < timeout) {
if (dd->rthread->isStopping())
return; return;
piMSleep(50);
}
} }
} }
PIByteArray ba; PIByteArray ba;

4
libs/main/piplatform.h
View File

@@ -100,4 +100,8 @@
#endif #endif
#if defined(LINUX) || defined(MAC_OS) || defined(ANDROID)
# define POSIX_SIGNALS
#endif
#endif // PIPLATFORM_H #endif // PIPLATFORM_H

34
libs/main/thread/pithread.cpp
View File

@@ -23,6 +23,9 @@
#ifndef MICRO_PIP #ifndef MICRO_PIP
# include "pisystemtests.h" # include "pisystemtests.h"
#endif #endif
#ifdef WINDOWS
# include <ioapiset.h>
#endif
#include <signal.h> #include <signal.h>
#if defined(WINDOWS) #if defined(WINDOWS)
# define __THREAD_FUNC_RET__ uint __stdcall # define __THREAD_FUNC_RET__ uint __stdcall
@@ -482,7 +485,7 @@ void __PIThreadCollection::stoppedAuto() {
auto_mutex.lock(); auto_mutex.lock();
auto_threads_.removeAll(t); auto_threads_.removeAll(t);
auto_mutex.unlock(); auto_mutex.unlock();
delete t; t->deleteLater();
} }
@@ -591,10 +594,35 @@ PIThread::~PIThread() {
} }
void PIThread::stop(bool wait) { void PIThread::stopAndWait(int timeout_ms) {
stop();
waitForFinish(timeout_ms);
}
#ifdef WINDOWS
NTAPI void winThreadAPC(ULONG_PTR) {
//piCout << "APC";
}
#endif
void PIThread::interrupt() {
if (PRIVATE->thread == 0) return;
piCout << "PIThread::interrupt";
#ifdef WINDOWS
CancelSynchronousIo(PRIVATE->thread);
QueueUserAPC(winThreadAPC, PRIVATE->thread, 0);
#else
# ifdef POSIX_SIGNALS
pthread_kill(PRIVATE->thread, PIP_INTERRUPT_SIGNAL);
# endif
#endif
}
void PIThread::stop() {
//PICout(PICoutManipulators::DefaultControls) << "thread" << this << "stop ..." << running_ << wait; //PICout(PICoutManipulators::DefaultControls) << "thread" << this << "stop ..." << running_ << wait;
terminating = true; terminating = true;
if (wait) waitForFinish();
} }

13
libs/main/thread/pithread.h
View File

@@ -107,10 +107,15 @@ public:
bool start(std::function<void()> func, int timer_delay) {ret_func = [func](void*){func();}; return start(timer_delay);} bool start(std::function<void()> func, int timer_delay) {ret_func = [func](void*){func();}; return start(timer_delay);}
EVENT_HANDLER0(bool, startOnce); EVENT_HANDLER0(bool, startOnce);
EVENT_HANDLER1(bool, startOnce, ThreadFunc, func) {ret_func = func; return startOnce();} EVENT_HANDLER1(bool, startOnce, ThreadFunc, func) {ret_func = func; return startOnce();}
EVENT_HANDLER0(void, stop) {stop(false);} EVENT_HANDLER0(void, stop);
EVENT_HANDLER1(void, stop, bool, wait);
EVENT_HANDLER0(void, terminate); EVENT_HANDLER0(void, terminate);
//! \~english Stop thread and wait for finish.
//! \~russian Останавливает поток и ожидает завершения.
void stopAndWait(int timeout_ms = -1);
void interrupt();
//! \~english Set common data passed to external function //! \~english Set common data passed to external function
//! \~russian Устанавливает данные, передаваемые в функцию потока //! \~russian Устанавливает данные, передаваемые в функцию потока
void setData(void * d) {data_ = d;} void setData(void * d) {data_ = d;}
@@ -195,7 +200,7 @@ public:
//! \~english Start thread without internal loop //! \~english Start thread without internal loop
//! \~russian Запускает поток без внутреннего цикла //! \~russian Запускает поток без внутреннего цикла
//! \fn void stop(bool wait = false) //! \fn void stop()
//! \brief //! \brief
//! \~english Stop thread //! \~english Stop thread
//! \~russian Останавливает поток //! \~russian Останавливает поток
@@ -203,7 +208,7 @@ public:
//! \fn void terminate() //! \fn void terminate()
//! \brief //! \brief
//! \~english Strongly stop thread //! \~english Strongly stop thread
//! \~russian Жестко останавливает поток //! \~russian Жёстко прерывает поток
//! \fn bool waitForStart(int timeout_msecs = -1) //! \fn bool waitForStart(int timeout_msecs = -1)
//! \brief //! \brief

4
libs/main/thread/pithreadpoolloop.cpp
View File

@@ -114,7 +114,7 @@ PIThreadPoolLoop::PIThreadPoolLoop(int thread_cnt) {
PIThreadPoolLoop::~PIThreadPoolLoop() { PIThreadPoolLoop::~PIThreadPoolLoop() {
for (auto * t: threads) { for (auto * t: threads) {
t->stop(false); t->stop();
if (!t->waitForFinish(100)) if (!t->waitForFinish(100))
t->terminate(); t->terminate();
delete t; delete t;
@@ -135,7 +135,7 @@ void PIThreadPoolLoop::start(int index_start, int index_count) {
while (1) { while (1) {
int cc = counter.fetch_add(1); int cc = counter.fetch_add(1);
if (cc >= end) { if (cc >= end) {
t->stop(false); t->stop();
return; return;
} }
func(cc); func(cc);

106
libs/main/thread/pitimer.cpp
View File

@@ -19,6 +19,7 @@
#include "pitimer.h" #include "pitimer.h"
#include "piincludes_p.h" #include "piincludes_p.h"
#include "piconditionvar.h"
#ifdef PIP_TIMER_RT #ifdef PIP_TIMER_RT
# include <csignal> # include <csignal>
#endif #endif
@@ -127,14 +128,14 @@ void _PITimerBase::setInterval(double i) {
interval_ = i; interval_ = i;
if (isRunning()) { if (isRunning()) {
//piCout << "change interval runtime"; //piCout << "change interval runtime";
stop(true); stop();
start(); start();
} }
} }
bool _PITimerBase::start(double interval_ms) { bool _PITimerBase::start(double interval_ms) {
if (isRunning()) stop(true); if (isRunning()) stop();
deferred_ = false; deferred_ = false;
setInterval(interval_ms); setInterval(interval_ms);
//piCout << "_PITimerBase::startTimer"<<interval_ms<<"..."; //piCout << "_PITimerBase::startTimer"<<interval_ms<<"...";
@@ -144,7 +145,7 @@ bool _PITimerBase::start(double interval_ms) {
void _PITimerBase::startDeferred(double interval_ms, PIDateTime start_datetime) { void _PITimerBase::startDeferred(double interval_ms, PIDateTime start_datetime) {
if (isRunning()) stop(true); if (isRunning()) stop();
deferred_ = true; deferred_ = true;
deferred_mode = true; deferred_mode = true;
deferred_datetime = start_datetime; deferred_datetime = start_datetime;
@@ -154,7 +155,7 @@ void _PITimerBase::startDeferred(double interval_ms, PIDateTime start_datetime)
void _PITimerBase::startDeferred(double interval_ms, double delay_ms) { void _PITimerBase::startDeferred(double interval_ms, double delay_ms) {
if (isRunning()) stop(true); if (isRunning()) stop();
deferred_ = true; deferred_ = true;
deferred_mode = false; deferred_mode = false;
deferred_delay = delay_ms; deferred_delay = delay_ms;
@@ -163,11 +164,11 @@ void _PITimerBase::startDeferred(double interval_ms, double delay_ms) {
} }
bool _PITimerBase::stop(bool wait) { bool _PITimerBase::stop() {
//piCout << GetCurrentThreadId() << "_PITimerBase::stop" << wait << isRunning(); //piCout << GetCurrentThreadId() << "_PITimerBase::stop" << wait << isRunning();
if (!isRunning()) return true; if (!isRunning()) return true;
//piCout << "_PITimerBase::stopTimer ..."; //piCout << "_PITimerBase::stopTimer ...";
running_ = !stopTimer(wait); running_ = !stopTimer();
return !running_; return !running_;
} }
@@ -177,19 +178,21 @@ bool _PITimerBase::stop(bool wait) {
class _PITimerImp_Thread: public _PITimerBase { class _PITimerImp_Thread: public _PITimerBase {
public: public:
_PITimerImp_Thread(); _PITimerImp_Thread();
virtual ~_PITimerImp_Thread(); virtual ~_PITimerImp_Thread() {stop();}
protected: protected:
void prepareStart(double interval_ms); void prepareStart(double interval_ms);
bool threadFunc(); // returns true if repeat is needed bool threadFunc(); // returns true if repeat is needed
int wait_dt, wait_dd, wait_tick; int wait_dt, wait_dd, wait_tick;
private: private:
virtual bool startTimer(double interval_ms); bool startTimer(double interval_ms) override;
virtual bool stopTimer(bool wait); bool stopTimer() override;
static void threadFuncS(void * d) {((_PITimerImp_Thread*)d)->threadFunc();} static void threadFuncS(void * d) {((_PITimerImp_Thread*)d)->threadFunc();}
void adjustTimes(); void adjustTimes();
bool smallWait(int ms);
PIThread thread_; PIThread thread_;
PISystemTime st_time, st_inc, st_wait, st_odt; PISystemTime st_time, st_inc, st_wait, st_odt;
PIConditionVariable event;
}; };
@@ -201,8 +204,8 @@ public:
virtual ~_PITimerImp_RT(); virtual ~_PITimerImp_RT();
protected: protected:
private: private:
virtual bool startTimer(double interval_ms); bool startTimer(double interval_ms) override;
virtual bool stopTimer(bool wait); bool stopTimer() override;
int ti; int ti;
_PITimerImp_RT_Private_ * priv; _PITimerImp_RT_Private_ * priv;
@@ -213,7 +216,7 @@ private:
class _PITimerImp_Pool: public _PITimerImp_Thread { class _PITimerImp_Pool: public _PITimerImp_Thread {
public: public:
_PITimerImp_Pool(); _PITimerImp_Pool();
virtual ~_PITimerImp_Pool() {stop(true);} virtual ~_PITimerImp_Pool() {stop();}
private: private:
class Pool: public PIThread { class Pool: public PIThread {
public: public:
@@ -226,8 +229,8 @@ private:
explicit Pool(); explicit Pool();
virtual ~Pool(); virtual ~Pool();
}; };
virtual bool startTimer(double interval_ms); bool startTimer(double interval_ms) override;
virtual bool stopTimer(bool wait); bool stopTimer() override;
}; };
@@ -235,19 +238,14 @@ private:
_PITimerImp_Thread::_PITimerImp_Thread() { _PITimerImp_Thread::_PITimerImp_Thread() {
thread_.setName("__S__PITimerImp_Thread::thread"); thread_.setName("__S__PITimerImp_Thread::thread");
wait_dt = 100; wait_dt = 1000;
wait_dd = 200; wait_dd = 2000;
wait_tick = 10; wait_tick = 1000;
//piCout << "_PITimerImp_Thread" << this << ", thread& =" << &thread_; //piCout << "_PITimerImp_Thread" << this << ", thread& =" << &thread_;
//piCout << "new _PITimerImp_Thread"; //piCout << "new _PITimerImp_Thread";
} }
_PITimerImp_Thread::~_PITimerImp_Thread() {
stop(true);
}
void _PITimerImp_Thread::prepareStart(double interval_ms) { void _PITimerImp_Thread::prepareStart(double interval_ms) {
if (interval_ms <= 0.) { if (interval_ms <= 0.) {
piCout << "Achtung! Start PITimer with interval <= 0!"; piCout << "Achtung! Start PITimer with interval <= 0!";
@@ -273,16 +271,14 @@ bool _PITimerImp_Thread::startTimer(double interval_ms) {
} }
bool _PITimerImp_Thread::stopTimer(bool wait) { bool _PITimerImp_Thread::stopTimer() {
#ifndef FREERTOS
thread_.stop(wait);
#else
thread_.stop(); thread_.stop();
if (wait) event.notifyAll();
if (!thread_.waitForFinish(10)) thread_.waitForFinish();
if (thread_.isRunning()) // if (wait)
thread_.terminate(); // if (!thread_.waitForFinish(10))
#endif // if (thread_.isRunning())
// thread_.terminate();
return true; return true;
} }
@@ -299,10 +295,10 @@ bool _PITimerImp_Thread::threadFunc() {
dwt = st_time - PISystemTime::current(true); dwt = st_time - PISystemTime::current(true);
if (wth > 0) { if (wth > 0) {
if (dwt.toMilliseconds() > wth + 1.) { if (dwt.toMilliseconds() > wth + 1.) {
piMSleep(wth); smallWait(wth);
return false; return false;
} else { } else {
dwt.sleep(); if (!smallWait(dwt.toMilliseconds())) return false;
deferred_ = false; deferred_ = false;
st_time = PISystemTime::current(true); st_time = PISystemTime::current(true);
} }
@@ -321,11 +317,11 @@ bool _PITimerImp_Thread::threadFunc() {
} }
if (wait_tick > 0) { if (wait_tick > 0) {
if (st_wait.toMilliseconds() > wait_tick + 1.) { if (st_wait.toMilliseconds() > wait_tick + 1.) {
piMSleep(wait_tick); smallWait(wait_tick);
return false; return false;
} else { } else {
//piCout << &thread_ << "sleep for" << st_wait; //piCout << &thread_ << "sleep for" << st_wait;
st_wait.sleep(); if (!smallWait(st_wait.toMilliseconds())) return false;
} }
} else { } else {
if (st_wait.toMilliseconds() > 0.1) if (st_wait.toMilliseconds() > 0.1)
@@ -366,6 +362,17 @@ void _PITimerImp_Thread::adjustTimes() {
} }
bool _PITimerImp_Thread::smallWait(int ms) {
if (thread_.isStopping()) return false;
if (ms > 0) {
thread_.mutex().lock();
event.waitFor(thread_.mutex(), ms);
thread_.mutex().unlock();
}
return !thread_.isStopping();
}
#ifdef PIP_TIMER_RT #ifdef PIP_TIMER_RT
@@ -392,7 +399,7 @@ _PITimerImp_RT::_PITimerImp_RT() {
_PITimerImp_RT::~_PITimerImp_RT() { _PITimerImp_RT::~_PITimerImp_RT() {
stop(true); stop();
delete priv; delete priv;
} }
@@ -425,7 +432,7 @@ bool _PITimerImp_RT::startTimer(double interval_ms) {
} }
bool _PITimerImp_RT::stopTimer(bool wait) { bool _PITimerImp_RT::stopTimer() {
if (ti < 0) return true; if (ti < 0) return true;
timer_delete(priv->tt); timer_delete(priv->tt);
ti = -1; ti = -1;
@@ -509,7 +516,7 @@ bool _PITimerImp_Pool::startTimer(double interval_ms) {
} }
bool _PITimerImp_Pool::stopTimer(bool wait) { bool _PITimerImp_Pool::stopTimer() {
Pool::instance()->remove(this); Pool::instance()->remove(this);
return true; return true;
} }
@@ -621,7 +628,7 @@ void PITimer::init() const {
void PITimer::destroy() { void PITimer::destroy() {
if (!imp) return; if (!imp) return;
//piCout << this << "destroy" << imp; //piCout << this << "destroy" << imp;
imp->stop(false); ///BUG: WTF FreeRTOS segfault on this! imp->stop();
delete imp; delete imp;
imp = 0; imp = 0;
} }
@@ -711,31 +718,14 @@ void PITimer::startDeferred(double interval_ms, PIDateTime start_datetime) {
bool PITimer::restart() { bool PITimer::restart() {
init(); init();
imp->stop(true); imp->stop();
return imp->start(); return imp->start();
} }
bool PITimer::stop() { bool PITimer::stop() {
return stop(true);
}
bool PITimer::stop(bool wait) {
init(); init();
//piCout << this << "stop" << imp << wait; //piCout << this << "stop" << imp << wait;
return imp->stop(wait); return imp->stop();
} }
bool PITimer::waitForFinish(int timeout_msecs) {
if (timeout_msecs < 0) {
while (isRunning())
piMinSleep();
return true;
}
PITimeMeasurer tm;
while (isRunning() && tm.elapsed_m() < timeout_msecs)
piMinSleep();
return tm.elapsed_m() < timeout_msecs;
}

7
libs/main/thread/pitimer.h
View File

@@ -53,7 +53,7 @@ public:
void startDeferred(PIDateTime start_datetime) {startDeferred(interval_, start_datetime);} void startDeferred(PIDateTime start_datetime) {startDeferred(interval_, start_datetime);}
void startDeferred(double interval_ms, PIDateTime start_datetime); void startDeferred(double interval_ms, PIDateTime start_datetime);
bool stop(bool wait); bool stop();
typedef void(*TickFunc)(PITimer*); typedef void(*TickFunc)(PITimer*);
TickFunc tfunc; TickFunc tfunc;
@@ -62,7 +62,7 @@ public:
protected: protected:
virtual bool startTimer(double interval_ms) = 0; virtual bool startTimer(double interval_ms) = 0;
virtual bool stopTimer(bool wait) = 0; virtual bool stopTimer() = 0;
double interval_, deferred_delay; double interval_, deferred_delay;
bool deferred_, deferred_mode; // mode: true - date, false - delay bool deferred_, deferred_mode; // mode: true - date, false - delay
@@ -161,9 +161,6 @@ public:
void startDeferred(double interval_ms, PIDateTime start_datetime); void startDeferred(double interval_ms, PIDateTime start_datetime);
EVENT_HANDLER0(bool, stop); EVENT_HANDLER0(bool, stop);
EVENT_HANDLER1(bool, stop, bool, wait);
bool waitForFinish() {return waitForFinish(-1);}
bool waitForFinish(int timeout_msecs);
//! \~english Set custom data //! \~english Set custom data
//! \~russian Установить данные, передаваемые в метод таймера //! \~russian Установить данные, передаваемые в метод таймера

243
main.cpp
View File

@@ -3,6 +3,7 @@
#include "pibytearray.h" #include "pibytearray.h"
#include "pimathbase.h" #include "pimathbase.h"
#include "pijson.h" #include "pijson.h"
#include "piwaitevent_p.h"
using namespace PICoutManipulators; using namespace PICoutManipulators;
@@ -12,43 +13,217 @@ typedef PIVector<PIVariant> PIVariantVector;
REGISTER_VARIANT(PIVariantMap); REGISTER_VARIANT(PIVariantMap);
REGISTER_VARIANT(PIVariantVector); REGISTER_VARIANT(PIVariantVector);
const char J[] = #ifdef WINDOWS
"[ \n" # include <namedpipeapi.h>
"{ \n" # include <handleapi.h>
" \"idFligth\":\"456123\", \n" # include <fileapi.h>
" \"FligthPath\": \"d:/orders/nicirt/BSK-52(BBR)/219031.001/EBN-NM-BBR.IMG\",\n" # include <processthreadsapi.h>
" \"FligthDataPath\": \"\", \n" typedef HANDLE pipe_type;
" \"Passport\": \n" #else
" { \n" # include <unistd.h>
" \"id\": \"\", \n" typedef int pipe_type;
" \"TypePlane\": \"\", \n" #endif
" \"FA_PLANEBORT\": \"Ka52-10\" \n"
" } \n"
" }, [1.1,2,3,4,{\"a\":null},{\"bool\":true,\"bool2\":false}] \n"
"] \n"
;
struct Pipe {
pipe_type fd_read = 0;
pipe_type fd_write = 0;
void create() {
#ifdef WINDOWS
CreatePipe(&fd_read, &fd_write, NULL, 0);
#else
pipe((int*)this);
#endif
}
void destoy() {
#ifdef WINDOWS
CloseHandle(fd_read);
CloseHandle(fd_write);
#else
close(fd_read);
close(fd_write);
#endif
}
int read(void * d, int s) {
#ifdef WINDOWS
DWORD ret(0);
ReadFile(fd_read, d, s, &ret, NULL);
return ret;
#else
return ::read(fd_read, d, s);
#endif
}
int write(void * d, int s) {
#ifdef WINDOWS
DWORD ret(0);
WriteFile(fd_write, d, s, &ret, NULL);
return ret;
#else
return ::write(fd_write, d, s);
#endif
}
};
constexpr int count = 4;
Pipe pipes[count];
class T: public PIThread {
public:
T(int index): PIThread() {ind = index; pipe = pipes[index];}
void run() {
PIByteArray data(1024);
piCout << "[T"<<ind<<"] start";
int readed = pipe.read(data.data(), data.size());
piCout << "[T"<<ind<<"] readed" << readed << errorString();
piCout << "[T"<<ind<<"] end";
}
int ind;
Pipe pipe;
};
PITimeMeasurer tm;
void phase(const char * msg) {
piCout << "";
piCout << piRound(tm.elapsed_s() * 10) / 10. << "s" << msg;
}
int main(int argc, char * argv[]) { int main(int argc, char * argv[]) {
struct { /*piCout << "main";
PIString name;
PIString surname; PITimer timer;
PIString email; timer.setSlot([](){
} persons[] = { static int cnt = 0;
{"Ivan", "Ivanov", "ivan@pip.ru"}, piCout << "tick" << ++cnt;
{"Igor", "Igorevich", "igor@pip.ru"}, });
{"Andrey", "Andreevich", "andrey@pip.ru"} timer.start(500);
};
PIJSON obj; piSleep(1.12);
PIJSON json; piCout << "end";
int index = 0; PITimeMeasurer tm;
for (const auto & p: persons) { timer.stop();
obj["index"] = index++; double tm_ms = tm.elapsed_m();
obj["name"] = p.name; piCout << "stop took" << tm_ms;*/
obj["surname"] = p.surname;
obj["email"] = p.email; PIWaitEvent event;
json << obj; event.create();
tm.reset();
PIThread::runOnce([&event](){
//piMSleep(100);
piCout << "interrupt";
//event.interrupt();
});
event.sleep(2010000);
double tm_ms = tm.elapsed_m();
piCout << "waited for" << tm_ms;
return 0;
/*for (int i = 0; i < count; ++i)
pipes[i].create();
PIVector<T*> threads;
piCout << "main start";
for (int i = 0; i < count; ++i) {
T * t = new T(i);
threads << t;
t->startOnce();
} }
piCout << json; piMSleep(100);
for (int i = 0; i < count; ++i) {
//pipes[i].write((void*)"string", 7);
piMSleep(500);
}
piCout << "main wait";
for (int i = 0; i < count; ++i) {
threads[i]->interrupt();
threads[i]->waitForFinish();
piCout << "main T" << i << "done";
}
piCout << "main end";
for (int i = 0; i < count; ++i) {
pipes[i].destoy();
delete threads[i];
}*/
//PIEthernet eth(PIEthernet::UDP), seth(PIEthernet::UDP);
//eth.setReadAddress("127.0.0.1", 50000);
//eth.startThreadedRead();
//piCout << eth.open();
PIByteArray req = PIByteArray::fromHex("205e011000000000ef");
PISerial ser;
ser.setSpeed(PISerial::S9600);
ser.setOption(PIIODevice::BlockingRead, false);
ser.setVTime(200);
ser.open("COM3");
CONNECTL(&ser, threadedReadEvent, ([](const uchar * data, ssize_t size){
piCout << "*ser readed" << size;
}));
PIThread thread;
thread.start([&](void*){
piCout << "[T] start";
PIByteArray data = ((PIIODevice*)&ser)->read(1024);
piCout << "[T] readed" << data.size();// << errorString();
piCout << "[T] end";
}, 200);
//ser.startThreadedRead();
piSleep(1);
ser.write(req);
phase("Send");
piSleep(2);
phase("End");
/*
PIEthernet eth(PIEthernet::TCP_Client), seth(PIEthernet::TCP_Server), * server_client = nullptr;
seth.listen("127.0.0.1", 50000, true);
//seth.startThreadedRead();
CONNECTL(&seth, newConnection, ([&server_client](PIEthernet * e){
server_client = e;
e->setName("TCP SC");
piCout << "newConn" << e;
CONNECTL(e, threadedReadEvent, ([](const uchar * data, ssize_t size){
piCout << "*TCP SC* readed" << size;
}));
CONNECTL(e, disconnected, ([](bool error){
piCout << "*TCP SC* disconnected" << error;
}));
e->startThreadedRead();
}));
eth.setName("TCP CC");
//eth.setParameter(PIEthernet::KeepConnection, false);
CONNECTL(&eth, connected, ([&eth](){
piCout << "*TCP CC* connected";
eth.send("byte", 5);
}));
CONNECTL(&eth, disconnected, ([](bool error){
piCout << "*TCP CC* disconnected" << error;
}));
piMSleep(500);
phase("Connect");
eth.connect("127.0.0.1", 50000);
eth.startThreadedRead();
piMSleep(500);
phase("Send 5");
piCout << "c-ing" << eth.isConnecting();
piCout << "c- ed" << eth.isConnected();
eth.send("byte", 5);
piMSleep(500);
phase("Send 6");
eth.send("bytes", 6);
piMSleep(500);
phase("Disconnect");
if (server_client)
server_client->close();
//eth.close();
piMSleep(500);
phase("END");
*/
return 0; return 0;
} }

13
main_picloud_test.cpp
View File

@@ -11,15 +11,15 @@ int main(int argc, char * argv[]) {
PICloudClient c("127.0.0.1:10101"); PICloudClient c("127.0.0.1:10101");
// c.setReopenEnabled(true); // c.setReopenEnabled(true);
PICloudServer s("127.0.0.1:10101"); PICloudServer s("127.0.0.1:10101");
PIVector<PICloudServer::Client *> clients; auto clients = new PIVector<PICloudServer::Client *>();
CONNECTL(&tm, tickEvent, ([&](void *, int){ CONNECTL(&tm, tickEvent, ([&](void *, int){
if (c.isConnected()) { if (c.isConnected()) {
PIString str = "ping"; PIString str = "ping";
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";
piCout << "[Server] send to" << cl << ":" << str; piCout << "[Server] send to" << cl << ":" << str;
@@ -40,7 +40,7 @@ int main(int argc, char * argv[]) {
CONNECTL(&c, disconnected, ([](){piCout << "disconnected";})); CONNECTL(&c, disconnected, ([](){piCout << "disconnected";}));
CONNECTL(&s, newConnection, ([&](PICloudServer::Client * cl){ CONNECTL(&s, newConnection, ([&](PICloudServer::Client * cl){
piCout << "[Server] new client:" << cl; piCout << "[Server] new client:" << cl;
clients << cl; clients->append(cl);
CONNECTL(cl, threadedReadEvent, ([cl, &rnd](const uchar * readed, ssize_t size){ CONNECTL(cl, threadedReadEvent, ([cl, &rnd](const uchar * readed, ssize_t size){
PIByteArray ba(readed, size); PIByteArray ba(readed, size);
PIString str = PIString(ba); PIString str = PIString(ba);
@@ -51,9 +51,10 @@ 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->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->close();
} }
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;

7
utils/cloud_dispatcher/dispatcherserver.cpp
View File

@@ -21,7 +21,7 @@ DispatcherServer::~DispatcherServer() {
void DispatcherServer::start() { void DispatcherServer::start() {
eth.listen(true); eth.listen(true);
timeout_timer.start(5000); timeout_timer.start(2000);
piCoutObj << "server started" << eth.readAddress(); piCoutObj << "server started" << eth.readAddress();
} }
@@ -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();
} }

4
utils/system_daemon/daemon.cpp
View File

@@ -24,7 +24,7 @@ Daemon::Remote::Remote(const PIString & n): PIThread() {
Daemon::Remote::~Remote() { Daemon::Remote::~Remote() {
shellClose(); shellClose();
ft.stop(); ft.stop();
stop(true); stopAndWait();
} }
@@ -41,7 +41,6 @@ void Daemon::Remote::shellClose() {
if (!term) return; if (!term) return;
piCoutObj << "shell close"; piCoutObj << "shell close";
term_timer.stop(); term_timer.stop();
term_timer.waitForFinish(1000);
term->destroy(); term->destroy();
delete term; delete term;
term = 0; term = 0;
@@ -331,6 +330,7 @@ Daemon::~Daemon() {
delete r; delete r;
} }
remotes.clear(); remotes.clear();
delete _self;
} }