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

PIPeer data transfers fixnsfer

Browse Source 
git-svn-id: svn://db.shs.com.ru/pip@47 12ceb7fc-bf1f-11e4-8940-5bc7170c53b5
This commit is contained in:
Пелипенко Иван
2015-03-31 08:06:02 +00:00
parent c87f73207d
commit 0701b15913
6 changed files with 88 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/io/pibasetransfer.cpp
View File

@@ -38,7 +38,7 @@ void PIBaseTransfer::stopReceive() {
} }
void PIBaseTransfer::received(PIByteArray& data) { void PIBaseTransfer::received(PIByteArray data) {
packet_header_size = sizeof(PacketHeader) + customHeader().size(); packet_header_size = sizeof(PacketHeader) + customHeader().size();
// piCoutObj << "receive" << data.size(); // piCoutObj << "receive" << data.size();
if (data.size() < sizeof(PacketHeader)) return; if (data.size() < sizeof(PacketHeader)) return;

2
src/io/pibasetransfer.h
View File

@@ -53,7 +53,7 @@ public:
EVENT(sendStarted) EVENT(sendStarted)
EVENT1(sendFinished, bool, ok) EVENT1(sendFinished, bool, ok)
EVENT1(sendRequest, PIByteArray &, data) EVENT1(sendRequest, PIByteArray &, data)
EVENT_HANDLER1(void, received, PIByteArray &, data); EVENT_HANDLER1(void, received, PIByteArray, data);
protected: protected:
uint packet_header_size, part_header_size; uint packet_header_size, part_header_size;

129
src/io/pipeer.cpp
View File

@@ -31,26 +31,40 @@
#define _PIPEER_TRAFFIC_PORT_E 14000 #define _PIPEER_TRAFFIC_PORT_E 14000
PIPeer::PeerData::~PeerData() { PIPeer::PeerData::PeerData(const PIString & n): PIObject(n) {
//if (t) piCout << "delete" << t; dt_in.setPacketSize(_PIPEER_MSG_SIZE);
if (t) delete t; dt_out.setPacketSize(_PIPEER_MSG_SIZE);
if (dt_in) delete dt_in; CONNECTU(&dt_in, sendRequest, this, dtSendRequestIn)
if (dt_out) delete dt_out; CONNECTU(&dt_out, sendRequest, this, dtSendRequestOut)
t = 0; CONNECTU(&dt_in, receiveFinished, this, dtReceiveFinishedIn)
dt_in = dt_out = 0; CONNECTU(&dt_out, receiveFinished, this, dtReceiveFinishedOut)
CONNECTU(&t, started, this, dtThread)
} }
void PIPeer::PeerData::initDT(const PIString & name) { void PIPeer::PeerData::dtThread() {
if (t == 0) t = new PIThread(); piCoutObj << "send DT ...";
if (dt_in == 0) dt_in = new PIDataTransfer(); dt_out.send(data);
if (dt_out == 0) dt_out = new PIDataTransfer(); piCoutObj << "send DT done";
//piCout << "new" << t; }
dt_in->setPacketSize(_PIPEER_MSG_SIZE);
dt_out->setPacketSize(_PIPEER_MSG_SIZE);
t->setName(name); bool PIPeer::PeerData::send(const PIByteArray & d) {
dt_in->setName(name); if (t.isRunning()) return false;
dt_out->setName(name); data = d;
t.startOnce();
return true;
}
void PIPeer::PeerData::receivedPacket(uchar type, const PIByteArray & d) {
PIDataTransfer * dt = 0;
if (type == 3)
dt = &dt_in;
if (type == 2)
dt = &dt_out;
piCoutObj << "DT received" << int(type) << d.size_s();
if (dt) dt->received(d);
} }
@@ -74,6 +88,17 @@ int PIPeer::PeerInfo::ping() const {
} }
void PIPeer::PeerInfo::init() {
if (_data == 0) _data = new PeerData(name);
}
void PIPeer::PeerInfo::destroy() {
if (_data) delete _data;
_data = 0;
}
PIString PIPeer::PeerInfo::fastestAddress() const { PIString PIPeer::PeerInfo::fastestAddress() const {
double mp = -1.; double mp = -1.;
PIString ret; PIString ret;
@@ -136,43 +161,6 @@ void PIPeer::timerEvent(void * data, int delim) {
} }
void PIPeer::dtSendRequest(uchar type, PIByteArray & data) {
PIObject * e = emitter();
//piCoutObj << "send request" << type << data.size_s();
if (!e) return;
data.push_front(uchar(2));
sendInternal(e->name(), data);
}
void PIPeer::dtReceiveFinished(bool ok) {
if (!ok) return;
PIDataTransfer * e = (PIDataTransfer*)emitter();
if (!e) return;
dataReceived(e->name(), e->data());
dataReceivedEvent(e->name(), e->data());
}
void PIPeer::dtThread() {
PIObject * e = emitter();
//piCoutObj << "send thread";
if (!e) return;
peers_mutex.lock();
PeerInfo * dp = quickestPeer(e->name());
if (dp == 0) {
peers_mutex.unlock();
return;
}
PIDataTransfer * dt = dp->_data.dt_out;
PIByteArray sd(dp->_data.data);
peers_mutex.unlock();
//piCoutObj << "send DT ...";
dt->send(sd);
//piCoutObj << "send DT done";
}
void PIPeer::initEths(PIStringList al) { void PIPeer::initEths(PIStringList al) {
PIEthernet * ce; PIEthernet * ce;
PIEthernet::InterfaceList il = PIEthernet::interfaces(); PIEthernet::InterfaceList il = PIEthernet::interfaces();
@@ -306,10 +294,7 @@ bool PIPeer::send(const PIString & to, const void * data, int size) {
PIMutexLocker mlocker(peers_mutex); PIMutexLocker mlocker(peers_mutex);
PeerInfo * dp = quickestPeer(to); PeerInfo * dp = quickestPeer(to);
if (!dp) return false; if (!dp) return false;
if (dp->_data.t->isRunning()) return false; return dp->_data->send(ba);
dp->_data.data = ba;
//piCoutObj << "start thread ...";
dp->_data.t->startOnce();
} }
//piCout << "[PIPeer] send" << size << "bytes ok"; //piCout << "[PIPeer] send" << size << "bytes ok";
return true; return true;
@@ -331,6 +316,12 @@ bool PIPeer::sendInternal(const PIString & to, const PIByteArray & data) {
} }
void PIPeer::dtReceived(const PIString & from, const PIByteArray & data) {
dataReceived(from, data);
dataReceivedEvent(from, data);
}
bool PIPeer::dataRead(uchar * readed, int size) { bool PIPeer::dataRead(uchar * readed, int size) {
if (size < 16) return true; if (size < 16) return true;
PIByteArray ba(readed, size), sba, pba; PIByteArray ba(readed, size), sba, pba;
@@ -406,14 +397,9 @@ bool PIPeer::dataRead(uchar * readed, int size) {
return true; return true;
} }
if (pt == 2 || pt == 3) { if (pt == 2 || pt == 3) {
PIDataTransfer * dt = 0;
if (pt == 3)
dt = fp->_data.dt_in;
if (pt == 2)
dt = fp->_data.dt_out;
peers_mutex.unlock(); peers_mutex.unlock();
//piCoutObj << "DT received" << pba.size_s(); if (fp->_data)
dt->received(pba); fp->_data->receivedPacket(pt, pba);
return true; return true;
} }
peers_mutex.unlock(); peers_mutex.unlock();
@@ -624,19 +610,17 @@ void PIPeer::removeNeighbour(const PIString & name) {
void PIPeer::addPeer(const PIPeer::PeerInfo & pd) { void PIPeer::addPeer(const PIPeer::PeerInfo & pd) {
peers << pd; peers << pd;
PeerData & p(peers.back()._data); PeerInfo & p(peers.back());
p.initDT(pd.name); p.init();
CONNECTU(p.dt_in, sendRequest, this, dtSendRequestIn) CONNECTU(p._data, sendRequest, this, sendInternal)
CONNECTU(p.dt_out, sendRequest, this, dtSendRequestOut) CONNECTU(p._data, received, this, dtReceived)
CONNECTU(p.dt_in, receiveFinished, this, dtReceiveFinished)
CONNECTU(p.dt_out, receiveFinished, this, dtReceiveFinished)
CONNECTU(p.t, started, this, dtThread)
} }
bool PIPeer::removePeer(const PIString & name) { bool PIPeer::removePeer(const PIString & name) {
for (int i = 0; i < peers.size_s(); ++i) for (int i = 0; i < peers.size_s(); ++i)
if (peers[i].name == name) { if (peers[i].name == name) {
peers[i].destroy();
peers.remove(i); peers.remove(i);
return true; return true;
} }
@@ -694,6 +678,7 @@ void PIPeer::syncPeers() {
if (cp.sync > 3) { if (cp.sync > 3) {
pn = cp.name; pn = cp.name;
//piCoutObj << "sync: remove " << pn; //piCoutObj << "sync: remove " << pn;
cp.destroy();
addToRemoved(cp); addToRemoved(cp);
peers.remove(i); peers.remove(i);
sendPeerRemove(pn); sendPeerRemove(pn);

40
src/io/pipeer.h
View File

@@ -31,15 +31,23 @@ class PIP_EXPORT PIPeer: public PIObject
{ {
PIOBJECT_SUBCLASS(PIPeer, PIObject) PIOBJECT_SUBCLASS(PIPeer, PIObject)
private: private:
struct PeerData {
explicit PeerData() {dt_in = dt_out = 0; t = 0;} class PeerData: public PIObject {
PeerData(const PeerData & ) {dt_in = dt_out = 0; t = 0;} PIOBJECT_SUBCLASS(PeerData, PIObject)
PeerData & operator =(const PeerData & ) {dt_in = dt_out = 0; t = 0; return *this;} public:
~PeerData(); PeerData(const PIString & n);
void initDT(const PIString & name); EVENT_HANDLER1(void, dtSendRequestIn, PIByteArray &, data) {data.push_front(uchar(2)); sendRequest(name(), data);}
EVENT_HANDLER1(void, dtSendRequestOut, PIByteArray &, data) {data.push_front(uchar(3)); sendRequest(name(), data);}
EVENT_HANDLER1(void, dtReceiveFinishedIn, bool, ok) {if (ok) received(name(), dt_in.data());}
EVENT_HANDLER1(void, dtReceiveFinishedOut, bool, ok) {if (ok) received(name(), dt_out.data());}
EVENT_HANDLER(void, dtThread);
EVENT2(received, const PIString &, from, const PIByteArray &, data);
EVENT2(sendRequest, const PIString &, to, const PIByteArray &, data);
bool send(const PIByteArray & d);
void receivedPacket(uchar type, const PIByteArray & d);
PIByteArray data; PIByteArray data;
PIThread * t; PIThread t;
PIDataTransfer * dt_in, * dt_out; PIDataTransfer dt_in, dt_out;
}; };
public: public:
@@ -51,7 +59,8 @@ public:
friend PIByteArray & operator <<(PIByteArray & s, const PIPeer::PeerInfo & v); friend PIByteArray & operator <<(PIByteArray & s, const PIPeer::PeerInfo & v);
friend PIByteArray & operator >>(PIByteArray & s, PIPeer::PeerInfo & v); friend PIByteArray & operator >>(PIByteArray & s, PIPeer::PeerInfo & v);
public: public:
PeerInfo() {dist = sync = cnt = 0; trace = -1; was_update = false;} PeerInfo() {dist = sync = cnt = 0; trace = -1; was_update = false; _data = 0;}
~PeerInfo() {}
struct Address { struct Address {
Address(const PIString & a = PIString(), const PIString & m = "255.255.255.0"); Address(const PIString & a = PIString(), const PIString & m = "255.255.255.0");
@@ -78,13 +87,14 @@ public:
void addNeighbours(const PIStringList & l) {piForeachC (PIString & n, l) if (!neighbours.contains(n)) neighbours << n;} void addNeighbours(const PIStringList & l) {piForeachC (PIString & n, l) if (!neighbours.contains(n)) neighbours << n;}
void removeNeighbour(const PIString & n) {neighbours.removeAll(n);} void removeNeighbour(const PIString & n) {neighbours.removeAll(n);}
void resetPing() {for (int i = 0; i < addresses.size_s(); ++i) addresses[i].ping = -1;} void resetPing() {for (int i = 0; i < addresses.size_s(); ++i) addresses[i].ping = -1;}
void init();
void destroy();
PIString nearest_address; PIString nearest_address;
int sync, cnt, trace; int sync, cnt, trace;
bool was_update; bool was_update;
PISystemTime time; PISystemTime time;
PeerData _data; PeerData * _data;
}; };
@@ -134,18 +144,14 @@ protected:
private: private:
EVENT_HANDLER2(void, timerEvent, void * , data, int, delim); EVENT_HANDLER2(void, timerEvent, void * , data, int, delim);
EVENT_HANDLER1(void, dtSendRequestIn, PIByteArray &, data) {dtSendRequest(2, data);} EVENT_HANDLER2(bool, sendInternal, const PIString &, to, const PIByteArray &, data);
EVENT_HANDLER1(void, dtSendRequestOut, PIByteArray &, data) {dtSendRequest(3, data);} EVENT_HANDLER2(void, dtReceived, const PIString &, from, const PIByteArray &, data);
EVENT_HANDLER1(void, dtReceiveFinished, bool, ok);
EVENT_HANDLER(void, dtThread);
bool hasPeer(const PIString & name) {piForeachC (PeerInfo & i, peers) if (i.name == name) return true; return false;} bool hasPeer(const PIString & name) {piForeachC (PeerInfo & i, peers) if (i.name == name) return true; return false;}
bool removePeer(const PIString & name); bool removePeer(const PIString & name);
void removeNeighbour(const PIString & name); void removeNeighbour(const PIString & name);
void addPeer(const PeerInfo & pd); void addPeer(const PeerInfo & pd);
bool sendInternal(const PIString & to, const PIByteArray & data);
void dtSendRequest(uchar type, PIByteArray & data);
void sendPeerInfo(const PeerInfo & info); void sendPeerInfo(const PeerInfo & info);
void sendPeerRemove(const PIString & peer); void sendPeerRemove(const PIString & peer);
void sendSelfInfo() {sendPeerInfo(self_info);} void sendSelfInfo() {sendPeerInfo(self_info);}

8
utils/system_daemon/daemon.cpp
View File

@@ -277,12 +277,12 @@ void Daemon::dataReceived(const PIString & from, const PIByteArray & data) {
r = remotes.value(from); r = remotes.value(from);
if (!r) break; if (!r) break;
ba >> dir; ba >> dir;
r->dir.cd(dir); r->dir_my.cd(dir);
{ {
PIVector<PIFile::FileInfo> fil = r->dir.entries(); PIVector<PIFile::FileInfo> fil = r->dir_my.entries();
piForeach (PIFile::FileInfo & f, fil) piForeach (PIFile::FileInfo & f, fil)
f.path = f.name(); f.path = f.name();
rba << int(ReplyChangeDir) << r->dir.absolutePath() << fil; rba << int(ReplyChangeDir) << r->dir_my.absolutePath() << fil;
} }
break; break;
case ReplyHostInfo: case ReplyHostInfo:
@@ -296,7 +296,7 @@ void Daemon::dataReceived(const PIString & from, const PIByteArray & data) {
{ {
PIVector<PIFile::FileInfo> fil; PIVector<PIFile::FileInfo> fil;
ba >> dir >> fil; ba >> dir >> fil;
r->dir.setDir(dir); r->dir_remote.setDir(dir);
fm.setRemoteDir(dir); fm.setRemoteDir(dir);
fm.setRemoteContent(fil); fm.setRemoteContent(fil);
fm.remoteRestoreDir(); fm.remoteRestoreDir();

4
utils/system_daemon/daemon.h
View File

@@ -66,9 +66,9 @@ private:
class Remote: public PIThread { class Remote: public PIThread {
public: public:
Remote(const PIString & n = PIString()) {dt.setName(n); ft.setName(n); dir = PIDir::current();} Remote(const PIString & n = PIString()) {dt.setName(n); ft.setName(n); dir_my = PIDir::current();}
void sendData(const PIByteArray & d) {_d = d; startOnce();} void sendData(const PIByteArray & d) {_d = d; startOnce();}
PIDir dir; PIDir dir_my, dir_remote;
PIDataTransfer dt; PIDataTransfer dt;
PIFileTransfer ft; PIFileTransfer ft;
PIByteArray _d; PIByteArray _d;