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

git-svn-id: svn://db.shs.com.ru/pip@240 12ceb7fc-bf1f-11e4-8940-5bc7170c53b5

Browse Source
This commit is contained in:
Бычков Андрей
2016-08-24 13:15:09 +00:00
parent 77a48eb270
commit 537c2fcd37
4 changed files with 27 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
src/io/pipeer.cpp
View File

@@ -29,6 +29,7 @@
#define _PIPEER_BROADCAST_PORT 13361
#define _PIPEER_TRAFFIC_PORT_S 13400
#define _PIPEER_TRAFFIC_PORT_E 14000
#define _PIPEER_PING_TIMEOUT 5.0
PIPeer::PeerData::PeerData(const PIString & n): PIObject(n) {
@@ -184,6 +185,8 @@ void PIPeer::timerEvent(void * data, int delim) {
switch (delim) {
case 1: // every 1 s
syncPeers();
piMSleep(100);
pingNeighbours();
break;
case 5: // every 5 s
checkNetwork();
@@ -354,6 +357,7 @@ PIPeer::PeerInfo * PIPeer::quickestPeer(const PIString & to) {
bool PIPeer::send(const PIString & to, const void * data, int size) {
PIByteArray ba(data, size);
// piCoutObj << "send" << ba.size_s() << "bytes" << _PIPEER_MSG_SIZE;
if (ba.size_s() <= _PIPEER_MSG_SIZE) {
ba.insert(0, uchar(1));
return sendInternal(to, ba);
@@ -364,7 +368,6 @@ bool PIPeer::send(const PIString & to, const void * data, int size) {
if (!dp) return false;
return dp->_data->send(ba);
}
//piCout << "[PIPeer] send" << size << "bytes ok";
return true;
}
@@ -378,8 +381,11 @@ bool PIPeer::sendInternal(const PIString & to, const PIByteArray & data) {
}
PIByteArray ba;
ba << int(4) << self_info.name << to << int(0) << data;
//piCoutObj << "sendInternal to" << to << data.size_s() << int(data.front());
if (!sendToNeighbour(dp, ba)) return false;
// piCoutObj << "sendInternal to" << to << data.size_s() << int(data.front());
if (!sendToNeighbour(dp, ba)) {
//piCoutObj << "send error";
return false;
}
return true;
}
@@ -406,19 +412,18 @@ bool PIPeer::dataRead(uchar * readed, int size) {
PIString from, to;
ba >> type;
PIMutexLocker locker(eth_mutex);
//piCoutObj << "received data from" << from << "packet" << type;
if (type == 5) { // ping request
PIString addr;
PISystemTime time;
ba >> to >> from >> addr >> time;
//piCout << "ping request" << to << from << addr;
// piCout << "ping request" << to << from << addr;
PIMutexLocker plocker(peers_mutex);
if (from == self_info.name) { // send ping back
const PeerInfo * pi = getPeerByName(to);
if (pi) {
if (pi->isNeighbour()) {
sba << int(6) << to << from << addr << time;
//piCout << " ping from" << from << addr << ", send back to" << pi->name;
// piCout << " ping from" << from << addr << ", send back to" << pi->name;
send_mutex.lock();
piForeachC (PeerInfo::Address & a, pi->addresses) {
if (eth_send.send(a.address, sba))
@@ -434,7 +439,7 @@ bool PIPeer::dataRead(uchar * readed, int size) {
PIString addr;
PISystemTime time, ptime, ctime = PISystemTime::current(true);
ba >> to >> from >> addr >> time;
//piCout << "ping reply" << to << from << addr;
// piCout << "ping reply" << to << from << addr;
PIMutexLocker plocker(peers_mutex);
if (to == self_info.name) { // ping echo
piForeach (PeerInfo & p, peers) {
@@ -448,13 +453,14 @@ bool PIPeer::dataRead(uchar * readed, int size) {
a.wait_ping = false;
if (a.ping < 0) a.ping = ptime.toMilliseconds();
else a.ping = 0.6 * a.ping + 0.4 * ptime.toMilliseconds();
//piCout << " ping echo" << p.name << a.address << a.ping;
// piCout << " ping echo" << p.name << a.address << a.ping;
return true;
}
}
}
return true;
}
// piCoutObj << "received data from" << from << "packet" << type;
if (type != 4) return true;
diag_d.received(size);
ba >> from >> to >> cnt >> pba;
@@ -731,24 +737,25 @@ void PIPeer::sendPeerRemove(const PIString & peer) {
void PIPeer::pingNeighbours() {
PIMutexLocker ml(peers_mutex);
PIByteArray ba, sba;
ba << int(5) << self_info.name;
//piCout << "*** pingNeighbours" << peers.size() << "...";
// piCoutObj << "*** pingNeighbours" << peers.size() << "...";
piForeach (PeerInfo & p, peers) {
if (!p.isNeighbour()) continue;
//piCout << " ping neighbour" << p.name << p.ping();
send_mutex.lock();
piForeach (PeerInfo::Address & a, p.addresses) {
//piCout << " address" << a.address << a.wait_ping;
// piCout << " address" << a.address << a.wait_ping;
if (a.wait_ping) {
if ((PISystemTime::current(true) - a.last_ping).abs().toSeconds() <= 5.)
if ((PISystemTime::current(true) - a.last_ping).abs().toSeconds() <= _PIPEER_PING_TIMEOUT)
continue;
a.ping = -1.;
}
a.wait_ping = true;
sba = ba;
sba << p.name << a.address << PISystemTime::current(true);
//piCout << "ping" << p.name << a.address << a.last_ping;
// piCout << "ping" << p.name << a.address << a.last_ping;
if (eth_send.send(a.address, sba))
diag_s.sended(sba.size_s());
}
@@ -797,7 +804,6 @@ void PIPeer::syncPeers() {
cp._data->setDist(cp.dist + 1);
cp.was_update = false;
}
pingNeighbours();
if (change) buildMap();
self_info.cnt++;
self_info.time = PISystemTime::current();