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

piclouddispatcher patch

Browse Source
This commit is contained in:
Бычков Андрей
2021-08-06 10:49:14 +03:00
parent 25def958a1
commit ca09b9d14b
6 changed files with 83 additions and 339 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
main.cpp
View File

@@ -1,42 +1,64 @@
#include "pip.h" #include "pip.h"
class A: public PIObject {
PIOBJECT(A)
public:
EVENT_HANDLER1(void, eh, PIByteArray, i) {piCout << "eh" << i.toHex();}
};
class B: public PIObject { int main(int argc, char * argv[]) {
PIOBJECT(B) PICLI cli(argc, argv);
public: PITimer tm;
EVENT2(eh, PIByteArray, i, PIString, str); cli.addArgument("connect", true);
}; cli.addArgument("name", true);
PICloudClient c("127.0.0.1:10101");
template <typename S, typename T, typename std::enable_if< // c.setReopenEnabled(true);
std::is_same<T, PIVector<S>>::value PICloudServer s("127.0.0.1:10101");
, int>::type = 0> PIVector<S> rrr(PIVector<T>) { PIVector<PICloudServer::Client *> clients;
piCout << std::is_same<T, PIVector<S>>::value; CONNECTL(&tm, tickEvent, ([&](void *, int){
return PIVector<S>(); if (c.isConnected()) {
} PIString str = "ping";
piCout << "[Client] send:" << str;
c.write(str.toByteArray());
int main() { }
A a; if (s.isRunning()) {
B b; for (auto cl : clients) {
CONNECTU_QUEUED(&b, eh, &a, eh, &a); if (cl->isOpened()) {
b.eh(PIByteArray::fromHex("102030"), "str"); PIString str = "ping_S";
a.maybeCallQueuedEvents(); piCout << "[Server] send to" << cl << ":" << str;
/*PIDeque<int> x; cl->write(str.toByteArray());
x.resize(16, [](size_t i) {return i+1;}); }
piCout << x; }
PIDeque<PIDeque<int>> m = x.reshape(2,8); }
piCout << m; }));
piCout << x.reshape(4,4,PIDeque<int>::byColumn); CONNECTL(&c, threadedReadEvent, ([&](uchar * readed, int size){
piCout << x.reshape(2,8); PIByteArray ba(readed, size);
piCout << x.reshape(2,8,PIDeque<int>::byColumn); PIString str = PIString(ba);
PIDeque<int> y; piCout << "[Client] data:" << str;
y = m.reshape<int>(); if (str == "ping_S") c.write(PIString("pong_S").toByteArray());
piCout << y; }));
piCout << m.reshape<int>(PIDeque<int>::byColumn);*/ CONNECTL(&s, newConnection, ([&](PICloudServer::Client * cl){
piCout << "[Server] new client:" << cl;
clients << cl;
CONNECTL(cl, threadedReadEvent, ([&c, &s, cl](uchar * readed, int size){
PIByteArray ba(readed, size);
PIString str = PIString(ba);
piCout << "[Server] data from" << cl << ":" << str;
if (str == "ping") cl->write(PIString("pong").toByteArray());
}));
CONNECTL(cl, closed, ([&clients, cl](){
cl->stop();
clients.removeAll(cl);
cl->deleteLater();
}));
cl->startThreadedRead();
}));
if (cli.hasArgument("name")) s.setServerName(cli.argumentValue("name"));
if (cli.hasArgument("connect")) {
c.setServerName(cli.argumentValue("connect"));
c.startThreadedRead();
} else {
s.startThreadedRead();
}
tm.start(1000);
PIKbdListener ls;
ls.enableExitCapture(PIKbdListener::F10);
ls.start();
WAIT_FOR_EXIT
return 0; return 0;
} }

294
main_.cpp
View File

@@ -1,294 +0,0 @@
/*
PIP - Platform Independent Primitives
Test program
Copyright (C) 2014 Ivan Pelipenko peri4ko@gmail.com
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU 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 General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
//#define PIP_DEBUG
#include "pip.h"
//#include "pivariant.h"
//#include "picodeparser.h"
//#include "pidir.h"
#include "piconnection.h"
//#include <QList>
PIMutex mutex;
PIDeque<int> deq;
void timerE(void * t, int d) {
//PIMutexLocker l(mutex);
/*if (d == 1)
deq.push_back(int(t));
else
deq.push_front(int(t));*/
//printf("%d\n", deq.size());
piCout << "tick" << (int)t << d;
}
class A: public PIObject {
PIOBJECT(A)
public:
A() {}
EVENT_HANDLER1(void, handlerA1, PIString, s) {piCout << "handlerA1 s" << s;}
EVENT_HANDLER1(void, handlerA1, float, f) {piCout << "handlerA1 f" << f;}
//uchar _[0x10];
};
class B: public PIObject {
PIOBJECT(B)
public:
B() {}
EVENT1(eventB1, float, f)
EVENT1(eventB2, PIString, s)
//uchar _[0x20];
};
class C: public A, public B {
public:
C() {}
};
int main (int argc, char * argv[]) {
PIFile f("test.txt");
f.clear();
f << "1234566789\n";
piCout << f.size();
f.close();
piCout << f.size();
/*CONNECTU(&b, eventB1, &a, handlerA1)
CONNECTU(&b, eventB2, &a, handlerA1)
a.dump();
b.dump();
b.eventB1(0.33);
b.eventB2("str");*/
//conf.dump();
/*PIObject * t_o = &timer, * a_o = &a, * b_o = &b;
CONNECTU(t_o, tickEvent, a_o, handlerT)
CONNECTU(b_o, eventB1, &a, handlerA1)
CONNECTU(&b, eventB2, a_o, hAAA)
a_o->dump("* ");
b_o->dump("# ");
//dumpApplication();
b.eventB(10, "str");
b.eventB1(10.5);
b.eventB2("test");*/
/*PIKbdListener kbd;
PITimer timer(timerE);
timer.start(100);
kbd.enableExitCapture();
WAIT_FOR_EXIT*/
/*
PIMathVectorT3d v(1., -2., 3.);
PIMathMatrixT33d m0;
m0.setCol(0, v.normalized());
m0.setCol(1, PIMathVectorT3d(v[1], -v[0], 0.).normalized());
m0.setCol(2, m0.col(1) * m0.col(0));
piCout << m0.determinant();
//m0.setRow(1, PIMathVectorT3d(1., 2., 3.));
//piCout << m0 << NewLine << NewLine << m0.;
//PIMathMatrixT33d m0_(-1., 2., 3., 4., -5., 6., 7., 8., -9.);
//PIMathMatrixT33d m1_(10., 2., 30., 4., 50., 6., 70., 8., 90.);
//piCout << m0_.determinant() << m1_.determinant();
*/
/*
PITimer::TimerImplementation ti_ = PITimer::ThreadRT;
PITimer t_(timerE, 0, ti_);
t_.setData((void*)0x10);
t_.start(100);
piSleep(1);
piCout << "end";
t_.stop();
piCout << "end 1";
t_.waitForFinish();
piCout << "end 2";
*/
/*
PIVector<PITimer*> timers;
int tc = PIString(argv[1]).toInt();
piCout << "create ...";
for (int i = 0; i < tc; ++i) {
PITimer * t = new PITimer(timerE, 0, PITimer::Thread);
t->setData((void*)i);
//t->dump();
timers << t;
if (i > 10) t->addDelimiter(i / 10);
t->start(100);
}
piCout << "create ok";
piSleep(1);
dumpApplication();
piCout << "delete ...";
piForeach (PITimer * t, timers)
t->stop();
piForeach (PITimer * t, timers)
delete t;
piCout << "delete ok";
*/
/*
PIConsole console(false);
PISystemMonitor mon;
console.addVariable("", &mon);
mon.startOnSelf();
console.enableExitCapture();
console.start();
console.waitForFinish();
return 0;
*/
/*PITimeMeasurer tm_;
piCout << PIIODevice::normalizeFullPath("ser:///dev/ttyS1");
piCout << tm_.elapsed_u();
tm_.reset();
piCout << PIIODevice::normalizeFullPath("ser:///dev/ttyS2");
piCout << tm_.elapsed_u();
tm_.reset();
piCout << PIIODevice::normalizeFullPath("ser:///dev/ttyS3");
piCout << tm_.elapsed_u();
tm_.reset();
piCout << PIIODevice::normalizeFullPath("ser:///dev/ttyS1");
piCout << tm_.elapsed_u();
tm_.reset();*/
//msleep(6000000);
/*A a_;
PIFile file("piiodevice.h", PIIODevice::ReadOnly);
PIByteArray header = PIString("PI").toByteArray();
PIByteArray footer = PIString("}").toByteArray();
PIPacketExtractor pe(&file);
//pe.setPacketData(header.data(), header.size_s(), 10);
pe.setSplitMode(PIPacketExtractor::Footer);
pe.setHeader(header);
pe.setFooter(footer);
pe.setPayloadSize(3);
pe.setThreadedReadBufferSize(40);
//pe.setBufferSize(256);
//pe.setPacketData(0, 0, 20);
CONNECT2(void, uchar * , int , &pe, packetReceived, &a_, per)
pe.startThreadedRead();
piMSleep(500);*/
/*TC tc;
piMSleep(2000);
//piCout << tc.diagnostic("file://piiodevice.h")->receiveBytesPerSec();
piCout << tc.makeConfig();*/
/*PITimer timer;
piForTimes (50) {
piUSleep(10);
piCout << PISystemTime::current() << timer.elapsed_u();
}
piCout << NewLine;
piForTimes (50) {
piUSleep(100);
piCout << PISystemTime::current(true) << timer.elapsed_u();
}*/
/*PIConsole console;
PIEthernet eth(PIEthernet::TCP_Client);
eth.connect("192.168.20:5006");
eth.startThreadedRead(readed);
console.enableExitCapture();
console.start();
console.waitForFinish();*/
/*PIBinaryLog log_;
GPS_Data gps;
log_.open("log_gps__2014_07_03__11_22_18", PIIODevice::ReadOnly);
while (!log_.isEnd()) {
log_.read(&gps, sizeof(gps));
printf("%f %f\n", gps.lat, gps.lng);
}*/
return 0;
/*tm = PISystemTime::current();
for (int i = 0; i < 10000000; ++i) {
ql.append(i*10);
ql.prepend(i*10 + 1);
}
piCout << (PISystemTime::current() - tm).toMicroseconds();
*/
//tm = PISystemTime::current();
/*for (int i = 0; i < 100000000; ++i) {
pl.append(i*10);
pl.prepend(i*10 + 1);
}*/
//PICodeParser cd_;
//cd.includeDirectory("../qpicalculator");
//cd_.parseFile("piincludes.h");
//piCout << (PISystemTime::current() - tm).toMilliseconds();
/*piCout << NewLine;
piForeachCA (i, pl)
piCout << i;
pl.remove(1, 2).prepend(111).prepend(222);
pl.remove(1, 1);
piCout << NewLine;
piForeachCA (i, pl)
piCout << i;*/
/*piCout << NewLine;
for (int i = 0; i < pl.size_s(); ++i)
piCout << pl[i];
*/
/*PIEthernet eth(PIEthernet::UDP);
eth.setReadAddress("192.168.0.30:4001");
eth.setSendAddress("192.168.0.50:4001");
eth.startThreadedRead(readed);
piCout << "Connected";
//eth.send(PIString("This is test string!\n").toByteArray());
FOREVER_WAIT*/
/*
if (argc < 2) return 0;
PICodeParser cd;
//cd.includeDirectory("../qpicalculator");
cd.parseFile(argv[1]);
piForeachC (PICodeParser::Enum & e, cd.enums)
piCout << e.name << e.members;
*/
//piCout << v.toType<float>();
//piCout << v.toType<float>().toType<PIString>();
//PIFile::remove("ki");
/*PIConfig conf("protocols_commod.conf");
piCout << conf.allTree();
conf.setValue("rmd.123", 456);*/
/*PITimer tm;
piCout << tm.debug() << tm.properties();
tm.setDebug(false);
piCout << tm.debug() << tm.properties();
tm.setDebug(true);
piCout << tm.debug() << tm.properties();*/
//PIObject * ser = (PIObject * )PIIODevice::createFromFullPath("file://OM2:38400:7");
//piCout << ser << NewLine << ser->properties();
}

2
utils/cloud_dispatcher/dispatcherclient.cpp
View File

@@ -17,7 +17,7 @@ void DispatcherClient::start() {
DispatcherClient::~DispatcherClient() { DispatcherClient::~DispatcherClient() {
// delete eth; delete eth;
} }

15
utils/cloud_dispatcher/dispatcherserver.cpp
View File

@@ -4,6 +4,7 @@
DispatcherServer::DispatcherServer(PIEthernet::Address addr) : eth(PIEthernet::TCP_Server) { DispatcherServer::DispatcherServer(PIEthernet::Address addr) : eth(PIEthernet::TCP_Server) {
client_gid = 0; client_gid = 0;
max_connections = 1000;
eth.setParameter(PIEthernet::ReuseAddress); eth.setParameter(PIEthernet::ReuseAddress);
eth.setReadAddress(addr); eth.setReadAddress(addr);
CONNECTU(&eth, newConnection, this, newConnection); CONNECTU(&eth, newConnection, this, newConnection);
@@ -123,11 +124,9 @@ const DispatcherClient * DispatcherServer::getConnection(int index) {
const DispatcherClient * DispatcherServer::getServer(int index) { const DispatcherClient * DispatcherServer::getServer(int index) {
const DispatcherClient * ret = nullptr; const DispatcherClient * ret = nullptr;
const DispatcherClient * tmp;
map_mutex.lock(); map_mutex.lock();
if (index >=0 && index < clients.size_s()) { if (index >=0 && index < clients.size_s()) {
tmp = clients[index]; if (index_c_servers.contains(clients[index])) ret = clients[index];
if (index_c_servers.contains(tmp)) ret = tmp;
} }
map_mutex.unlock(); map_mutex.unlock();
return ret; return ret;
@@ -149,6 +148,11 @@ PISet<const DispatcherClient *> DispatcherServer::getServers(PISet<int> ids) {
} }
void DispatcherServer::setMaxConnections(uint max_count) {
max_connections = max_count;
}
void DispatcherServer::disconnectClient(DispatcherClient *client) { void DispatcherServer::disconnectClient(DispatcherClient *client) {
if (!clients.contains(client)) { if (!clients.contains(client)) {
//piCoutObj << "INVALID client" << client; //piCoutObj << "INVALID client" << client;
@@ -183,6 +187,11 @@ void DispatcherServer::disconnectClient(DispatcherClient *client) {
void DispatcherServer::newConnection(PIEthernet *cl) { void DispatcherServer::newConnection(PIEthernet *cl) {
if (clients.size_s() >= max_connections) {
cl->close();
delete cl;
return;
}
DispatcherClient * client = new DispatcherClient(cl, client_gid++); DispatcherClient * client = new DispatcherClient(cl, client_gid++);
CONNECTU(client, disconnectEvent, this, disconnectClient); CONNECTU(client, disconnectEvent, this, disconnectClient);
CONNECTL(client, registerServer, [this](const PIByteArray & sname, DispatcherClient * c){ CONNECTL(client, registerServer, [this](const PIByteArray & sname, DispatcherClient * c){

3
utils/cloud_dispatcher/dispatcherserver.h
View File

@@ -19,6 +19,8 @@ public:
const DispatcherClient * getConnection(int index); const DispatcherClient * getConnection(int index);
const DispatcherClient * getServer(int index); const DispatcherClient * getServer(int index);
PISet<const DispatcherClient *> getServers(PISet<int> ids); PISet<const DispatcherClient *> getServers(PISet<int> ids);
void setMaxConnections(uint max_count);
uint maxConnections() const {return max_connections;}
EVENT_HANDLER0(void, picoutStatus); EVENT_HANDLER0(void, picoutStatus);
private: private:
@@ -35,6 +37,7 @@ private:
PITimer timeout_timer; PITimer timeout_timer;
PIMutex map_mutex; PIMutex map_mutex;
uint client_gid; uint client_gid;
uint max_connections;
}; };
#endif // DISPATCHERSERVER_H #endif // DISPATCHERSERVER_H

12
utils/cloud_dispatcher/main.cpp
View File

@@ -69,15 +69,18 @@ int main (int argc, char * argv[]) {
PIString conf_path = confDir(); PIString conf_path = confDir();
PIDir::make(conf_path); PIDir::make(conf_path);
conf_path += "/picloud.conf"; conf_path += "/picloud.conf";
uint max_connections = 1000;
if (!PIFile::isExists(conf_path)) { if (!PIFile::isExists(conf_path)) {
PIFile f(conf_path, PIIODevice::ReadWrite); PIFile f(conf_path, PIIODevice::ReadWrite);
f << "ip = " << addr.ipString() << "\n" f << "ip = " << addr.ipString() << "\n"
<< "port = " << addr.port() << "\n"; << "port = " << addr.port() << "\n"
<< "max_connections = " << max_connections << "\n";
} }
{ {
PIConfig conf(conf_path, PIIODevice::ReadOnly); PIConfig conf(conf_path, PIIODevice::ReadOnly);
addr.setIP(conf.getValue("ip", addr.ipString()).toString()); addr.setIP(conf.getValue("ip", addr.ipString()).toString());
addr.setPort(conf.getValue("port", addr.port()).toUShort()); addr.setPort(conf.getValue("port", addr.port()).toUShort());
max_connections = conf.getValue("max_connections", max_connections).toUInt();
} }
PITimer status_timer; PITimer status_timer;
@@ -86,6 +89,7 @@ int main (int argc, char * argv[]) {
if (cli.hasArgument("port")) if (cli.hasArgument("port"))
addr.setPort(cli.argumentValue("port").toInt()); addr.setPort(cli.argumentValue("port").toInt());
DispatcherServer server(addr); DispatcherServer server(addr);
server.setMaxConnections(max_connections);
if (cli.hasArgument("screen")) { if (cli.hasArgument("screen")) {
PISet<const DispatcherClient *> sel_servers; PISet<const DispatcherClient *> sel_servers;
PIScreen screen(false); PIScreen screen(false);