diff --git a/CMakeLists.txt b/CMakeLists.txt index 6fcefd79..365a54ed 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -2,8 +2,8 @@ cmake_minimum_required(VERSION 3.0) cmake_policy(SET CMP0017 NEW) # need include() with .cmake project(PIP) set(PIP_MAJOR 3) -set(PIP_MINOR 5) -set(PIP_REVISION 1) +set(PIP_MINOR 6) +set(PIP_REVISION 0) set(PIP_SUFFIX ) set(PIP_COMPANY SHS) set(PIP_DOMAIN org.SHS) diff --git a/libs/cloud/picloudserver.cpp b/libs/cloud/picloudserver.cpp index 9a66e290..61125f85 100644 --- a/libs/cloud/picloudserver.cpp +++ b/libs/cloud/picloudserver.cpp @@ -90,6 +90,7 @@ PIVector PICloudServer::clients() const { bool PICloudServer::openDevice() { //piCout << "PICloudServer open device" << path(); + if (is_deleted) return false; bool op = eth.connect(PIEthernet::Address::resolve(path()), false); if (op) { eth.startThreadedRead(); @@ -244,6 +245,7 @@ void PICloudServer::_readed(PIByteArray & ba) { Client * oc = index_clients.value(id, nullptr); clients_mutex.unlock(); if (oc) { + piCoutObj << "Warning: reject client with duplicated ID"; tcp.sendDisconnected(id); } else { Client * c = new Client(this, id); @@ -258,7 +260,7 @@ void PICloudServer::_readed(PIByteArray & ba) { } break; case PICloud::TCP::Disconnect: { uint id = tcp.parseDisconnect(ba); - //piCoutObj << "remove Client" << id; + //piCoutObj << "Close on logic"; clients_mutex.lock(); Client * oc = index_clients.take(id, nullptr); clients_.removeOne(oc); diff --git a/libs/main/core/pibase.h b/libs/main/core/pibase.h index b066e6cd..c8eff9d8 100644 --- a/libs/main/core/pibase.h +++ b/libs/main/core/pibase.h @@ -248,10 +248,6 @@ typedef long time_t; #endif -#ifdef POSIX_SIGNALS -# define PIP_INTERRUPT_SIGNAL SIGTERM -#endif - #ifdef LINUX # define environ __environ #endif diff --git a/libs/main/core/piinit.cpp b/libs/main/core/piinit.cpp index 3c768797..c0e70c84 100644 --- a/libs/main/core/piinit.cpp +++ b/libs/main/core/piinit.cpp @@ -92,24 +92,6 @@ void __sighandler__(PISignals::Signal s) { } -#ifdef POSIX_SIGNALS -void pipThreadSignalHandler(int sig) { -//# 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 - - PIInit::PIInit() { file_charset = 0; PISystemInfo * sinfo = PISystemInfo::instance(); @@ -178,9 +160,6 @@ PIInit::PIInit() { setlocale(LC_NUMERIC, "C"); # endif //HAS_LOCALE #endif //ANDROID -#ifdef POSIX_SIGNALS - pipInitThreadSignals(); -#endif PRIVATE->delete_locs = false; __syslocname__ = __sysoemname__ = 0; __utf8name__ = const_cast("UTF-8"); diff --git a/libs/main/core/piwaitevent_p.cpp b/libs/main/core/piwaitevent_p.cpp index 72d85aee..23534ec8 100644 --- a/libs/main/core/piwaitevent_p.cpp +++ b/libs/main/core/piwaitevent_p.cpp @@ -85,10 +85,12 @@ bool PIWaitEvent::wait(int fd, CheckRole role) { 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 sr = ::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; + //piCout << "wait result" << sr << FD_ISSET(fd, &(fds[CheckExeption])) << FD_ISSET(fd, &(fds[fd_index])); + if (sr == EBADF || sr == EINTR) return false; + if (FD_ISSET(fd, &(fds[CheckExeption]))) return true; return FD_ISSET(fd, &(fds[fd_index])); #endif return true; diff --git a/libs/main/io_devices/piethernet.cpp b/libs/main/io_devices/piethernet.cpp index 2ae8f8fd..ebd75e51 100644 --- a/libs/main/io_devices/piethernet.cpp +++ b/libs/main/io_devices/piethernet.cpp @@ -294,7 +294,6 @@ bool PIEthernet::init() { st = SOCK_STREAM; pr = IPPROTO_TCP; } - PIFlags params = parameters(); sock = ::socket(AF_INET, st, pr); ethNonblocking(sock); PRIVATE->event.create(); @@ -375,7 +374,6 @@ bool PIEthernet::openDevice() { memset(&PRIVATE->addr_, 0, sizeof(PRIVATE->addr_)); PRIVATE->addr_.sin_family = AF_INET; PRIVATE->addr_.sin_port = htons(addr_r.port()); - PIFlags params = parameters(); if (params[PIEthernet::Broadcast]) PRIVATE->addr_.sin_addr.s_addr = INADDR_ANY; else PRIVATE->addr_.sin_addr.s_addr = addr_r.ip(); #ifdef QNX @@ -419,7 +417,7 @@ bool PIEthernet::closeDevice() { while (!clients_.isEmpty()) delete clients_.back(); if (ned) { - piCoutObj << "Disconnect on close"; + //piCoutObj << "Disconnect on close"; disconnected(false); } return true; @@ -468,13 +466,6 @@ void PIEthernet::applyOptInt(int level, int opt, int val) { } -void PIEthernet::setParameter(PIEthernet::Parameters parameter, bool on) { - PIFlags cp = (PIFlags)(property("parameters").toInt()); - cp.setFlag(parameter, on); - setParameters(cp); -} - - bool PIEthernet::joinMulticastGroup(const PIString & group) { if (sock == -1) init(); if (sock == -1) return false; @@ -489,7 +480,6 @@ bool PIEthernet::joinMulticastGroup(const PIString & group) { if (!mcast_groups.contains(group)) mcast_groups << group; return true; } - PIFlags params = parameters(); addr_r.set(path()); #ifndef LWIP struct ip_mreqn mreq; @@ -536,7 +526,6 @@ bool PIEthernet::leaveMulticastGroup(const PIString & group) { piCoutObj << "Only UDP sockets can leave multicast groups"; return false; } - PIFlags params = parameters(); addr_r.set(path()); #ifndef LWIP struct ip_mreqn mreq; @@ -715,9 +704,9 @@ ssize_t PIEthernet::readDevice(void * read_to, ssize_t max_size) { #ifdef QNX PRIVATE->addr_.sin_len = sizeof(PRIVATE->addr_); #endif - piCoutObj << "connect to " << path() << "..."; + //piCoutObj << "connect to " << path() << "..."; connected_ = connectTCP(); - piCoutObj << "connect to " << path() << connected_; + //piCoutObj << "connect to " << path() << connected_; if (!connected_) piCoutObj << "Can`t connect to" << addr_r << "," << ethErrorString(); opened_.exchange(connected_); @@ -734,6 +723,7 @@ ssize_t PIEthernet::readDevice(void * read_to, ssize_t max_size) { { long wr = waitForEvent(PRIVATE->event, FD_READ | FD_CLOSE); switch (wr) { + case 0: return -1; case FD_READ: //piCout << "fd_read ..."; rs = ethRecv(sock, read_to, max_size); @@ -746,31 +736,47 @@ ssize_t PIEthernet::readDevice(void * read_to, ssize_t max_size) { } } #else - if (PRIVATE->event.wait(sock)) + if (PRIVATE->event.wait(sock)) { + errorClear(); rs = ethRecv(sock, read_to, max_size); + } #endif //piCoutObj << "readed" << rs; if (rs <= 0) { lerr = ethErrorCore(); - //piCoutObj << "readed error" << lerr << errorString().data() << parameters()[DisonnectOnTimeout]; + //piCoutObj << "readed" << rs << "error" << lerr; + + // async normal returns #ifdef WINDOWS - if ((lerr == WSAEWOULDBLOCK || lerr == WSAETIMEDOUT) && !parameters()[DisonnectOnTimeout]) { -#elif defined(ANDROID) - if ((lerr == EWOULDBLOCK || lerr == EAGAIN || lerr == EINTR) && !parameters()[DisonnectOnTimeout] && rs < 0) { + if (lerr == WSAEWOULDBLOCK) { #else - if ((lerr == EWOULDBLOCK || lerr == EAGAIN) && !parameters()[DisonnectOnTimeout] && rs < 0) { + if (lerr == EWOULDBLOCK || lerr == EAGAIN || lerr == EINTR) { #endif - //piCoutObj << errorString(); + //piCoutObj << "Ignore would_block" << lerr; return -1; } + + // if no disconnect on timeout + if (!params[DisonnectOnTimeout]) { +#ifdef WINDOWS + if (lerr == WSAETIMEDOUT) { +#else + if (lerr == ETIMEDOUT) { +#endif + //piCoutObj << "Ignore read timeout"; + return -1; + } + } + // disconnect here + //piCoutObj << "Disconnnected, check for event, connected =" << connected_; if (connected_.exchange(false)) { opened_ = false; - piCoutObj << "Disconnect on read," << ethErrorString(); + //piCoutObj << "Disconnect on read," << ethErrorString(); closeSocket(sock); init(); disconnected(rs < 0); } - if (parameters()[KeepConnection]) { + if (params[KeepConnection]) { connect(); } //piCoutObj << "eth" << ip_ << "disconnected"; @@ -857,7 +863,7 @@ ssize_t PIEthernet::writeDevice(const void * data, ssize_t max_size) { auto disconnectFunc = [this](){ if (connected_.exchange(false)) { opened_ = false; - piCoutObj << "Disconnect on write," << ethErrorString(); + //piCoutObj << "Disconnect on write," << ethErrorString(); closeSocket(sock); init(); disconnected(true); @@ -965,8 +971,8 @@ void PIEthernet::server_func(void * eth) { PIString ip = PIStringAscii(inet_ntoa(client_addr.sin_addr)); ip += ":" + PIString::fromNumber(htons(client_addr.sin_port)); PIEthernet * e = new PIEthernet(s, ip); + CONNECT1(void, PIObject *, e, deleted, ce, clientDeleted); ce->clients_mutex.lock(); - CONNECT1(void, PIObject *, e, deleted, ce, clientDeleted) ce->clients_ << e; ce->clients_mutex.unlock(); ce->newConnection(e); @@ -1012,15 +1018,17 @@ bool PIEthernet::connectTCP() { #ifdef WINDOWS long PIEthernet::waitForEvent(PIWaitEvent & event, long mask) { if (!event.isCreate() || sock < 0) return 0; - WSAEventSelect(sock, event.getEvent(), mask); + if (WSAEventSelect(sock, event.getEvent(), mask) == SOCKET_ERROR) { + if (ethErrorCore() == WSAEINPROGRESS) + return 0; + } 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; + //piCoutObj << "wait result" << events.lNetworkEvents; return events.lNetworkEvents; } return 0; diff --git a/libs/main/io_devices/piethernet.h b/libs/main/io_devices/piethernet.h index 06b2dbb9..d4b3c97b 100644 --- a/libs/main/io_devices/piethernet.h +++ b/libs/main/io_devices/piethernet.h @@ -200,16 +200,16 @@ public: //! Set parameters to "parameters_". You should to reopen %PIEthernet to apply them - void setParameters(PIFlags parameters_) {setProperty("parameters", (int)parameters_);} + void setParameters(PIFlags parameters_) {params = parameters_;} //! Set parameter "parameter" to state "on". You should to reopen %PIEthernet to apply this - void setParameter(PIEthernet::Parameters parameter, bool on = true); + void setParameter(PIEthernet::Parameters parameter, bool on = true) {params.setFlag(parameter, on);} //! Returns if parameter "parameter" is set - bool isParameterSet(PIEthernet::Parameters parameter) const {return ((PIFlags)(property("parameters").toInt()))[parameter];} + bool isParameterSet(PIEthernet::Parameters parameter) const {return params[parameter];} //! Returns parameters - PIFlags parameters() const {return (PIFlags)(property("parameters").toInt());} + PIFlags parameters() const {return params;} //! Returns %PIEthernet type Type type() const {return eth_type;} @@ -496,6 +496,7 @@ protected: PIVector clients_; PIQueue mcast_queue; PIStringList mcast_groups; + PIFlags params; private: EVENT_HANDLER1(void, clientDeleted, PIObject *, o); diff --git a/libs/main/thread/pithread.cpp b/libs/main/thread/pithread.cpp index ee5d388f..c217f54a 100644 --- a/libs/main/thread/pithread.cpp +++ b/libs/main/thread/pithread.cpp @@ -600,26 +600,6 @@ void PIThread::stopAndWait(int 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; terminating = true; diff --git a/libs/main/thread/pithread.h b/libs/main/thread/pithread.h index 948cbefd..74b2cda3 100644 --- a/libs/main/thread/pithread.h +++ b/libs/main/thread/pithread.h @@ -113,8 +113,6 @@ public: //! \~english Stop thread and wait for finish. //! \~russian Останавливает поток и ожидает завершения. void stopAndWait(int timeout_ms = -1); - - void interrupt(); //! \~english Set common data passed to external function //! \~russian Устанавливает данные, передаваемые в функцию потока diff --git a/main.cpp b/main.cpp index 2b673b70..e70a223e 100644 --- a/main.cpp +++ b/main.cpp @@ -103,7 +103,7 @@ int main(int argc, char * argv[]) { double tm_ms = tm.elapsed_m(); piCout << "stop took" << tm_ms;*/ - PIWaitEvent event; + /*PIWaitEvent event; event.create(); tm.reset(); PIThread::runOnce([&event](){ @@ -114,7 +114,7 @@ int main(int argc, char * argv[]) { event.sleep(2010000); double tm_ms = tm.elapsed_m(); piCout << "waited for" << tm_ms; - return 0; + return 0;*/ /*for (int i = 0; i < count; ++i) pipes[i].create(); @@ -148,7 +148,7 @@ int main(int argc, char * argv[]) { //eth.startThreadedRead(); //piCout << eth.open(); - PIByteArray req = PIByteArray::fromHex("205e011000000000ef"); + /*PIByteArray req = PIByteArray::fromHex("205e011000000000ef"); PISerial ser; ser.setSpeed(PISerial::S9600); ser.setOption(PIIODevice::BlockingRead, false); @@ -171,9 +171,9 @@ int main(int argc, char * argv[]) { phase("Send"); piSleep(2); - phase("End"); + phase("End");*/ + - /* PIEthernet eth(PIEthernet::TCP_Client), seth(PIEthernet::TCP_Server), * server_client = nullptr; seth.listen("127.0.0.1", 50000, true); @@ -224,6 +224,6 @@ int main(int argc, char * argv[]) { piMSleep(500); phase("END"); - */ + return 0; } diff --git a/utils/cloud_dispatcher/dispatcherclient.cpp b/utils/cloud_dispatcher/dispatcherclient.cpp index df6cd480..b14d6e66 100644 --- a/utils/cloud_dispatcher/dispatcherclient.cpp +++ b/utils/cloud_dispatcher/dispatcherclient.cpp @@ -3,6 +3,7 @@ DispatcherClient::DispatcherClient(PIEthernet * eth_, int id) : authorised(false), eth(eth_), streampacker(eth_), tcp(&streampacker), client_id(id) { + eth->setName(PIString::fromNumber(id)); CONNECTU(&streampacker, packetReceiveEvent, this, readed); CONNECTU(eth, disconnected, this, disconnected); piCoutObj << "client connected" << client_id << eth->sendAddress();