From 9efb7f413687f20d9efba9d6a504266a847415e9 Mon Sep 17 00:00:00 2001
From: csoler <retroshare.project@gmail.com>
Date: Sat, 12 Apr 2014 19:58:45 +0000
Subject: [PATCH] improved routing algorithm. Fixed several bugs.
git-svn-id: http://svn.code.sf.net/p/retroshare/code/trunk@7269 b45a01b8-16f6-495d-af2f-9b41ad6348cc
---
libretroshare/src/grouter/grouteritems.cc | 162 ++++++++--------
libretroshare/src/grouter/grouteritems.h | 50 +++--
libretroshare/src/grouter/groutermatrix.cc | 2 +-
libretroshare/src/grouter/groutertypes.h | 11 +-
libretroshare/src/grouter/p3grouter.cc | 204 ++++++++++++---------
libretroshare/src/grouter/p3grouter.h | 1 +
libretroshare/src/retroshare/rsgrouter.h | 3 +-
7 files changed, 231 insertions(+), 202 deletions(-)
diff --git a/libretroshare/src/grouter/grouteritems.cc b/libretroshare/src/grouter/grouteritems.cc
index fca710763..5430ae4bb 100644
--- a/libretroshare/src/grouter/grouteritems.cc
+++ b/libretroshare/src/grouter/grouteritems.cc
@@ -30,46 +30,46 @@ bool RsGRouterItem::serialise_header(void *data,uint32_t& pktsize,uint32_t& tlvs
}
/* serialise the data to the buffer */
-uint32_t RsGRouterPublishKeyItem::serial_size() const
-{
- uint32_t s = 8 ; // header
- s += POW_PAYLOAD_SIZE ; // proof of work bytes
- s += 4 ; // diffusion_id
- s += published_key.serial_size() ; // sha1 for published_key
- s += 4 ; // service id
- s += 4 ; // randomized distance
- s += GetTlvStringSize(description_string) ; // description
- s += fingerprint.serial_size() ; // fingerprint
-
- return s ;
-}
-bool RsGRouterPublishKeyItem::serialise(void *data, uint32_t& pktsize) const
-{
- uint32_t tlvsize,offset=0;
- bool ok = true;
-
- if(!serialise_header(data,pktsize,tlvsize,offset))
- return false ;
-
- memcpy(&((uint8_t*)data)[offset],pow_bytes,POW_PAYLOAD_SIZE) ;
- offset += 8 ;
-
- /* add mandatory parts first */
- ok &= setRawUInt32(data, tlvsize, &offset, diffusion_id);
- ok &= published_key.serialise(data, tlvsize, offset) ;
- ok &= setRawUInt32(data, tlvsize, &offset, service_id);
- ok &= setRawUFloat32(data, tlvsize, &offset, randomized_distance);
- ok &= SetTlvString(data, tlvsize, &offset, TLV_TYPE_STR_VALUE, description_string);
- ok &= fingerprint.serialise(data, tlvsize, offset) ;
-
- if (offset != tlvsize)
- {
- ok = false;
- std::cerr << "RsFileItemSerialiser::serialiseData() Size Error! " << std::endl;
- }
-
- return ok;
-}
+// uint32_t RsGRouterPublishKeyItem::serial_size() const
+// {
+// uint32_t s = 8 ; // header
+// s += POW_PAYLOAD_SIZE ; // proof of work bytes
+// s += 4 ; // diffusion_id
+// s += published_key.serial_size() ; // sha1 for published_key
+// s += 4 ; // service id
+// s += 4 ; // randomized distance
+// s += GetTlvStringSize(description_string) ; // description
+// s += fingerprint.serial_size() ; // fingerprint
+//
+// return s ;
+// }
+//bool RsGRouterPublishKeyItem::serialise(void *data, uint32_t& pktsize) const
+//{
+// uint32_t tlvsize,offset=0;
+// bool ok = true;
+//
+// if(!serialise_header(data,pktsize,tlvsize,offset))
+// return false ;
+//
+// memcpy(&((uint8_t*)data)[offset],pow_bytes,POW_PAYLOAD_SIZE) ;
+// offset += 8 ;
+//
+// /* add mandatory parts first */
+// ok &= setRawUInt32(data, tlvsize, &offset, diffusion_id);
+// ok &= published_key.serialise(data, tlvsize, offset) ;
+// ok &= setRawUInt32(data, tlvsize, &offset, service_id);
+// ok &= setRawUFloat32(data, tlvsize, &offset, randomized_distance);
+// ok &= SetTlvString(data, tlvsize, &offset, TLV_TYPE_STR_VALUE, description_string);
+// ok &= fingerprint.serialise(data, tlvsize, offset) ;
+//
+// if (offset != tlvsize)
+// {
+// ok = false;
+// std::cerr << "RsFileItemSerialiser::serialiseData() Size Error! " << std::endl;
+// }
+//
+// return ok;
+//}
/**********************************************************************************************/
/* PROOF OF WORK STUFF */
@@ -157,7 +157,7 @@ RsItem *RsGRouterSerialiser::deserialise(void *data, uint32_t *pktsize)
switch(getRsItemSubType(rstype))
{
- case RS_PKT_SUBTYPE_GROUTER_PUBLISH_KEY: return deserialise_RsGRouterPublishKeyItem(data, *pktsize);
+ //case RS_PKT_SUBTYPE_GROUTER_PUBLISH_KEY: return deserialise_RsGRouterPublishKeyItem(data, *pktsize);
case RS_PKT_SUBTYPE_GROUTER_DATA: return deserialise_RsGRouterGenericDataItem(data, *pktsize);
case RS_PKT_SUBTYPE_GROUTER_ACK: return deserialise_RsGRouterACKItem(data, *pktsize);
case RS_PKT_SUBTYPE_GROUTER_MATRIX_CLUES: return deserialise_RsGRouterMatrixCluesItem(data, *pktsize);
@@ -170,32 +170,32 @@ RsItem *RsGRouterSerialiser::deserialise(void *data, uint32_t *pktsize)
return NULL;
}
-RsGRouterPublishKeyItem *RsGRouterSerialiser::deserialise_RsGRouterPublishKeyItem(void *data, uint32_t pktsize) const
-{
- uint32_t offset = 8; // skip the header
- uint32_t rssize = getRsItemSize(data);
- bool ok = true ;
-
- RsGRouterPublishKeyItem *item = new RsGRouterPublishKeyItem() ;
-
- memcpy(&((uint8_t*)data)[offset],item->pow_bytes,RsGRouterProofOfWorkObject::POW_PAYLOAD_SIZE) ;
- offset += 8 ;
-
- ok &= getRawUInt32(data, pktsize, &offset, &item->diffusion_id); // file hash
- ok &= item->published_key.deserialise(data, pktsize, offset) ;
- ok &= getRawUInt32(data, pktsize, &offset, &item->service_id); // file hash
- ok &= getRawUFloat32(data, pktsize, &offset, item->randomized_distance); // file hash
- ok &= GetTlvString(data, pktsize, &offset, TLV_TYPE_STR_VALUE,item->description_string);
- ok &= item->fingerprint.deserialise(data,pktsize,offset) ;
-
- if (offset != rssize || !ok)
- {
- std::cerr << __PRETTY_FUNCTION__ << ": error while deserialising! Item will be dropped." << std::endl;
- return NULL ;
- }
-
- return item;
-}
+//RsGRouterPublishKeyItem *RsGRouterSerialiser::deserialise_RsGRouterPublishKeyItem(void *data, uint32_t pktsize) const
+//{
+// uint32_t offset = 8; // skip the header
+// uint32_t rssize = getRsItemSize(data);
+// bool ok = true ;
+//
+// RsGRouterPublishKeyItem *item = new RsGRouterPublishKeyItem() ;
+//
+// memcpy(&((uint8_t*)data)[offset],item->pow_bytes,RsGRouterProofOfWorkObject::POW_PAYLOAD_SIZE) ;
+// offset += 8 ;
+//
+// ok &= getRawUInt32(data, pktsize, &offset, &item->diffusion_id); // file hash
+// ok &= item->published_key.deserialise(data, pktsize, offset) ;
+// ok &= getRawUInt32(data, pktsize, &offset, &item->service_id); // file hash
+// ok &= getRawUFloat32(data, pktsize, &offset, item->randomized_distance); // file hash
+// ok &= GetTlvString(data, pktsize, &offset, TLV_TYPE_STR_VALUE,item->description_string);
+// ok &= item->fingerprint.deserialise(data,pktsize,offset) ;
+//
+// if (offset != rssize || !ok)
+// {
+// std::cerr << __PRETTY_FUNCTION__ << ": error while deserialising! Item will be dropped." << std::endl;
+// return NULL ;
+// }
+//
+// return item;
+//}
RsGRouterGenericDataItem *RsGRouterSerialiser::deserialise_RsGRouterGenericDataItem(void *data, uint32_t pktsize) const
{
@@ -560,20 +560,20 @@ bool RsGRouterRoutingInfoItem::serialise(void *data,uint32_t& size) const
// ------------------------------------- IO --------------------------------------- //
// -----------------------------------------------------------------------------------//
//
-std::ostream& RsGRouterPublishKeyItem::print(std::ostream& o, uint16_t)
-{
- o << "GRouterPublishKeyItem:" << std::endl ;
- o << " POW bytes : \""<< RsPgpId(pow_bytes).toStdString() << "\"" << std::endl ;
- o << " direct origin: \""<< PeerId() << "\"" << std::endl ;
- o << " Key: " << published_key.toStdString() << std::endl ;
- o << " Req. Id: " << std::hex << diffusion_id << std::dec << std::endl ;
- o << " Srv. Id: " << std::hex << service_id << std::dec << std::endl ;
- o << " Distance: " << randomized_distance << std::endl ;
- o << " Description: " << description_string << std::endl ;
- o << " Fingerprint: " << fingerprint.toStdString() << std::endl ;
-
- return o ;
-}
+//std::ostream& RsGRouterPublishKeyItem::print(std::ostream& o, uint16_t)
+//{
+// o << "GRouterPublishKeyItem:" << std::endl ;
+// o << " POW bytes : \""<< RsPgpId(pow_bytes).toStdString() << "\"" << std::endl ;
+// o << " direct origin: \""<< PeerId() << "\"" << std::endl ;
+// o << " Key: " << published_key.toStdString() << std::endl ;
+// o << " Req. Id: " << std::hex << diffusion_id << std::dec << std::endl ;
+// o << " Srv. Id: " << std::hex << service_id << std::dec << std::endl ;
+// o << " Distance: " << randomized_distance << std::endl ;
+// o << " Description: " << description_string << std::endl ;
+// o << " Fingerprint: " << fingerprint.toStdString() << std::endl ;
+//
+// return o ;
+//}
std::ostream& RsGRouterACKItem::print(std::ostream& o, uint16_t)
{
o << "RsGRouterACKItem:" << std::endl ;
diff --git a/libretroshare/src/grouter/grouteritems.h b/libretroshare/src/grouter/grouteritems.h
index 3e75b80f4..eea14f12f 100644
--- a/libretroshare/src/grouter/grouteritems.h
+++ b/libretroshare/src/grouter/grouteritems.h
@@ -44,12 +44,6 @@ const uint8_t QOS_PRIORITY_RS_GROUTER_PUBLISH_KEY = 3 ; // slow items. No nee
const uint8_t QOS_PRIORITY_RS_GROUTER_ACK = 3 ;
const uint8_t QOS_PRIORITY_RS_GROUTER_DATA = 3 ;
-const uint32_t RS_GROUTER_ACK_STATE_RECEIVED = 0x0001 ; // data was received, directly
-const uint32_t RS_GROUTER_ACK_STATE_RECEIVED_INDIRECTLY = 0x0002 ; // data was received indirectly
-const uint32_t RS_GROUTER_ACK_STATE_GIVEN_UP = 0x0003 ; // data was given up. No route.
-const uint32_t RS_GROUTER_ACK_STATE_NO_ROUTE = 0x0004 ; // data was given up. No route.
-const uint32_t RS_GROUTER_ACK_STATE_UNKNOWN = 0x0005 ; // unknown destination key
-const uint32_t RS_GROUTER_ACK_STATE_TOO_FAR = 0x0006 ; // dropped because of distance
/***********************************************************************************/
/* Basic GRouter Item Class */
@@ -107,27 +101,27 @@ class RsGRouterProofOfWorkObject
/* Specific packets */
/***********************************************************************************/
-class RsGRouterPublishKeyItem: public RsGRouterItem, public RsGRouterProofOfWorkObject
-{
- public:
- RsGRouterPublishKeyItem() : RsGRouterItem(RS_PKT_SUBTYPE_GROUTER_PUBLISH_KEY) { setPriorityLevel(QOS_PRIORITY_RS_GROUTER_PUBLISH_KEY) ; }
-
- virtual bool serialise(void *data,uint32_t& size) const ;
- virtual uint32_t serial_size() const ;
-
- virtual void clear() {}
- virtual std::ostream& print(std::ostream &out, uint16_t indent = 0) ;
-
- // packet data
- //
- GRouterKeyPropagationId diffusion_id ;
- GRouterKeyId published_key ;
- uint32_t service_id ;
- float randomized_distance ;
- std::string description_string ;
- PGPFingerprintType fingerprint ;
-
-};
+//class RsGRouterPublishKeyItem: public RsGRouterItem, public RsGRouterProofOfWorkObject
+//{
+// public:
+// RsGRouterPublishKeyItem() : RsGRouterItem(RS_PKT_SUBTYPE_GROUTER_PUBLISH_KEY) { setPriorityLevel(QOS_PRIORITY_RS_GROUTER_PUBLISH_KEY) ; }
+//
+// virtual bool serialise(void *data,uint32_t& size) const ;
+// virtual uint32_t serial_size() const ;
+//
+// virtual void clear() {}
+// virtual std::ostream& print(std::ostream &out, uint16_t indent = 0) ;
+//
+// // packet data
+// //
+// GRouterKeyPropagationId diffusion_id ;
+// GRouterKeyId published_key ;
+// uint32_t service_id ;
+// float randomized_distance ;
+// std::string description_string ;
+// PGPFingerprintType fingerprint ;
+//
+//};
class RsGRouterGenericDataItem: public RsGRouterItem, public RsGRouterNonCopyableObject
{
@@ -252,7 +246,7 @@ class RsGRouterSerialiser: public RsSerialType
virtual RsItem *deserialise (void *data, uint32_t *size) ;
private:
- RsGRouterPublishKeyItem *deserialise_RsGRouterPublishKeyItem(void *data,uint32_t size) const ;
+ //RsGRouterPublishKeyItem *deserialise_RsGRouterPublishKeyItem(void *data,uint32_t size) const ;
RsGRouterGenericDataItem *deserialise_RsGRouterGenericDataItem(void *data,uint32_t size) const ;
RsGRouterACKItem *deserialise_RsGRouterACKItem(void *data,uint32_t size) const ;
RsGRouterMatrixCluesItem *deserialise_RsGRouterMatrixCluesItem(void *data,uint32_t size) const ;
diff --git a/libretroshare/src/grouter/groutermatrix.cc b/libretroshare/src/grouter/groutermatrix.cc
index 6647428fb..47a462fd1 100644
--- a/libretroshare/src/grouter/groutermatrix.cc
+++ b/libretroshare/src/grouter/groutermatrix.cc
@@ -184,7 +184,7 @@ bool GRouterMatrix::computeRoutingProbabilities(const GRouterKeyId& key_id, cons
}
if(total > 0.0f)
- for(int i=0;i<friends.size();++i)
+ for(uint32_t i=0;i<friends.size();++i)
probas[i] /= total ;
return true ;
diff --git a/libretroshare/src/grouter/groutertypes.h b/libretroshare/src/grouter/groutertypes.h
index b93b7c101..8eeb0a0ea 100644
--- a/libretroshare/src/grouter/groutertypes.h
+++ b/libretroshare/src/grouter/groutertypes.h
@@ -29,13 +29,10 @@
#include <time.h>
#include <list>
#include "pgp/rscertificate.h"
+#include "retroshare/rsgrouter.h"
class RsGRouterGenericDataItem ;
-typedef uint32_t GRouterServiceId ;
-typedef uint32_t GRouterKeyPropagationId ;
-typedef uint64_t GRouterMsgPropagationId ;
-
static const uint32_t GROUTER_CLIENT_ID_MESSAGES = 0x1001 ;
static const uint32_t RS_GROUTER_MATRIX_MAX_HIT_ENTRIES = 5;
@@ -50,6 +47,7 @@ static const time_t RS_GROUTER_MATRIX_UPDATE_PERIOD = 1 *10 ; // Check
static const time_t RS_GROUTER_PUBLISH_KEY_TIME_INTERVAL = 2 *60 ; // Advertise each key once a day at most.
static const time_t RS_GROUTER_ROUTING_WAITING_TIME = 2 *60 ; // time between two trial of sending a given message
//static const time_t RS_GROUTER_ROUTING_WAITING_TIME = 3600 ; // time between two trial of sending a given message
+static const time_t RS_GROUTER_MEAN_EXPECTED_RTT = 30 ; // reference RTT time for a message.
static const uint32_t GROUTER_ITEM_DISTANCE_UNIT = 256 ; // One unit of distance between two peers
static const uint32_t GROUTER_ITEM_MAX_TRAVEL_DISTANCE = 16*256 ; // 16 distance units. That is a lot.
@@ -85,10 +83,11 @@ class GRouterRoutingInfo
public:
uint32_t status_flags ; // pending, waiting, etc.
RsPeerId origin ; // which friend sent us that item
- time_t received_time ; // time at which the item was received
- time_t last_activity ; // time at which the item was received
+ time_t received_time ; // time at which the item was originally received
+ time_t last_sent ; // last time the item was sent to friends
std::list<FriendTrialRecord> tried_friends ; // list of friends to which the item was sent ordered with time.
+ GRouterKeyId destination_key ; // ultimate destination for this key
RsGRouterGenericDataItem *data_item ;
};
diff --git a/libretroshare/src/grouter/p3grouter.cc b/libretroshare/src/grouter/p3grouter.cc
index 31fd01715..995353e4c 100644
--- a/libretroshare/src/grouter/p3grouter.cc
+++ b/libretroshare/src/grouter/p3grouter.cc
@@ -235,8 +235,10 @@ int p3GRouter::tick()
//
if(now > _last_matrix_update_time + RS_GROUTER_MATRIX_UPDATE_PERIOD)
{
+ RsStackMutex mtx(grMtx) ;
+
_last_matrix_update_time = now ;
- _routing_matrix.updateRoutingProbabilities() ;
+ _routing_matrix.updateRoutingProbabilities() ; // This should be locked.
}
#ifdef GROUTER_DEBUG
@@ -289,10 +291,10 @@ void p3GRouter::autoWash()
time_t now = time(NULL) ;
for(std::map<GRouterMsgPropagationId,GRouterRoutingInfo>::iterator it(_pending_messages.begin());it!=_pending_messages.end();)
- if(it->second.last_activity + GROUTER_ITEM_MAX_CACHE_KEEP_TIME < now)
+ if(it->second.received_time + GROUTER_ITEM_MAX_CACHE_KEEP_TIME < now) // is the item too old for cache
{
#ifdef GROUTER_DEBUG
- std::cerr << " Removing cache item " << std::hex << it->first << std::dec << " for key id " << it->second.data_item->destination_key << std::endl;
+ std::cerr << " Removing cache item " << std::hex << it->first << std::dec << std::endl;
#endif
delete it->second.data_item ;
std::map<GRouterMsgPropagationId,GRouterRoutingInfo>::iterator tmp(it) ;
@@ -300,6 +302,14 @@ void p3GRouter::autoWash()
_pending_messages.erase(it) ;
it = tmp ;
}
+ else if(it->second.data_item != NULL && it->second.status_flags == RS_GROUTER_ROUTING_STATE_SENT && it->second.last_sent+RS_GROUTER_ROUTING_WAITING_TIME < now)
+ {
+ it->second.status_flags = RS_GROUTER_ROUTING_STATE_PEND ;
+#ifdef GROUTER_DEBUG
+ std::cerr << " Scheduling the item " << std::hex << it->first << std::dec << " for sending again." << std::endl;
+#endif
+ ++it ;
+ }
else
++it ;
@@ -325,35 +335,56 @@ void p3GRouter::routePendingObjects()
mServiceControl->getPeersConnected(RS_SERVICE_TYPE_GROUTER,lst) ;
RsPeerId own_id( mServiceControl->getOwnId() );
- std::vector<RsPeerId> pids ;
- for(std::set<RsPeerId>::const_iterator it(lst.begin());it!=lst.end();++it)
- pids.push_back(*it) ;
-
+ // The policy is the following:
+ //
+ // - all pending messages should be handled. A msg is pending when it is waiting for routage.
+ // A pending message should always have a non NULL data item attached.
+ //
for(std::map<GRouterMsgPropagationId, GRouterRoutingInfo>::iterator it(_pending_messages.begin());it!=_pending_messages.end();)
- if((it->second.status_flags == RS_GROUTER_ROUTING_STATE_PEND) || (it->second.status_flags == RS_GROUTER_ROUTING_STATE_SENT && it->second.tried_friends.front().time_stamp+RS_GROUTER_ROUTING_WAITING_TIME < now))
+ if(it->second.status_flags == RS_GROUTER_ROUTING_STATE_PEND)
{
+ // make sure we have data to send.
+ //
+ if(it->second.data_item == NULL)
+ {
+ std::cerr << " (EE) Pending item has no data!!" << std::endl;
+ ++it ;
+ continue ;
+ }
#ifdef GROUTER_DEBUG
std::cerr << " Msg id: " << std::hex << it->first << std::dec << std::endl;
std::cerr << " Origin: " << it->second.origin.toStdString() << std::endl;
if(!it->second.tried_friends.empty())
{
std::cerr << " Last : " << it->second.tried_friends.front().friend_id.toStdString() << std::endl;
- std::cerr << " R Time: " << it->second.tried_friends.front().time_stamp << std::endl;
+ std::cerr << " S Time: " << it->second.tried_friends.front().time_stamp << std::endl;
}
- std::cerr << " Recvd : " << it->second.received_time << std::endl;
- std::cerr << " Last M: " << it->second.last_activity << std::endl;
+ std::cerr << " Recvd : " << now - it->second.received_time << " secs ago." << std::endl;
+ std::cerr << " Sent : " << now - it->second.last_sent << " secs ago." << std::endl;
std::cerr << " Flags : " << it->second.status_flags << std::endl;
std::cerr << " Dist : " << it->second.data_item->randomized_distance<< std::endl;
std::cerr << " Probabilities: " << std::endl;
#endif
+ std::vector<RsPeerId> pids ;
+ for(std::set<RsPeerId>::const_iterator its(lst.begin());its!=lst.end();++its)
+ if(*its != it->second.origin)
+ pids.push_back(*its) ;
+ if(pids.empty()) // no friends to send to!! Send back a give up signal.
+ {
+ sendACK(it->second.origin,it->first,RS_GROUTER_ACK_STATE_GVNP) ;
+ it->second.status_flags = RS_GROUTER_ROUTING_STATE_DEAD ;
+ delete it->second.data_item ;
+ it->second.data_item = NULL ;
+ ++it ;
+ continue ;
+ }
std::vector<float> probas ; // friends probabilities for online friend list.
RsPeerId routed_friend ; // friend chosen for the next hop
- bool should_remove = false ; // should we remove this from the map?
// Retrieve probabilities for this key. This call always succeeds. If no route is known, all probabilities become equal.
//
- _routing_matrix.computeRoutingProbabilities(it->second.data_item->destination_key, pids, probas) ;
+ _routing_matrix.computeRoutingProbabilities(it->second.destination_key, pids, probas) ;
// Compute the maximum branching factor.
@@ -386,7 +417,6 @@ void p3GRouter::routePendingObjects()
ftr.nb_friends = probas.size() ;
it->second.tried_friends.push_front(ftr) ;
- it->second.status_flags = RS_GROUTER_ROUTING_STATE_SENT ;
#ifdef GROUTER_DEBUG
std::cerr << " Routing probability: " << ftr.probability << std::endl;
@@ -400,28 +430,13 @@ void p3GRouter::routePendingObjects()
sendItem(new_item) ;
}
-
- if(should_remove)
- {
- // We remove from the map. That means the RsItem* has been transfered to somewhere else.
- //
-#ifdef GROUTER_DEBUG
- std::cerr << " Removing item from pending items" << std::endl;
-#endif
-
- std::map<GRouterMsgPropagationId, GRouterRoutingInfo>::iterator tmp(it) ;
- delete it->second.data_item ;
- ++tmp ;
- _pending_messages.erase(it) ;
- it = tmp ;
- }
- else
- ++it ;
+ it->second.status_flags = RS_GROUTER_ROUTING_STATE_SENT ;
+ it->second.last_sent = now ;
}
else
{
#ifdef GROUTER_DEBUG
- std::cerr << "Skipping " << std::hex << it->first << std::dec << ", dest=" << it->second.data_item->destination_key.toStdString() << ", state = " << it->second.status_flags ;
+ std::cerr << "Skipping " << std::hex << it->first << std::dec << ", state = " << it->second.status_flags ;
if(!it->second.tried_friends.empty())
std::cerr << ", stamp=" << it->second.tried_friends.front().time_stamp << " - " << it->second.tried_friends.front().friend_id.toStdString() << std::endl;
else
@@ -467,21 +482,23 @@ uint32_t p3GRouter::computeBranchingFactor(const std::vector<RsPeerId>& friends,
uint32_t dist_index = std::min( (uint32_t)(dist / (float)GROUTER_ITEM_DISTANCE_UNIT), MAX_DIST_INDEX-1) ;
return std::max(2, (int)(friends.size()*branching_factors[dist_index])) ;
+}
- //// Now temper the branching factor by how likely we are to already have a good guess from the probabilities:
- //// - if the largest probability is much larger than the second one
- //
- //std::vector<float> probs(probas) ;
- //std::sort(probs.begin(),probs.end()) ;
- //int n=0 ;
-
- //for(int i=probs.size()-1;i>=0;--i)
- // if(probs[i] > 0.5 * probs.back())
- // ++n ;
-
- //// send the final value
-
- //return std::max(1, std::min(n, (int)(friends.size()*branching_factors[dist_index]))) ;
+float p3GRouter::computeMatrixContribution(float base,uint32_t time_shift,float probability)
+{
+ // This function computes the contribution to the routing matrix for an ACK that was
+ // received. The different variables are:
+ // base : base contribution. 1.0 for directly received items, 0.5 for indirectly received items.
+ // time_shift : time in seconds between when the item was sent and when the item was ACKed. This is a clue of
+ // how far the destination is, and is used to favor fast routes.
+ // probability : probability with which the item was sent. This should be used for importance-sampling the resulting weight
+
+ if(probability == 0.0f)
+ {
+ std::cerr << "Probability is NULL !!!!! This should not happen." << std::endl;
+ return 0.0f ;
+ }
+ return base * exp(-float(time_shift) / float(RS_GROUTER_MEAN_EXPECTED_RTT)) / probability ;
}
class peer_comparison_function
@@ -604,7 +621,6 @@ std::set<uint32_t> p3GRouter::computeRoutingFriends_old(const std::vector<RsPeer
//
return res ;
}
-#endif
void p3GRouter::locked_forwardKey(const RsGRouterPublishKeyItem& item)
{
@@ -636,6 +652,8 @@ void p3GRouter::locked_forwardKey(const RsGRouterPublishKeyItem& item)
std::cerr << " Not forwarding to source id " << item.PeerId() << std::endl;
#endif
}
+#endif
+
bool p3GRouter::registerKey(const GRouterKeyId& key,const GRouterServiceId& client_id,const std::string& description)
{
RsStackMutex mtx(grMtx) ;
@@ -761,8 +779,11 @@ void p3GRouter::handleRecvACKItem(RsGRouterACKItem *item)
#endif
return ;
}
- uint32_t next_state ;
- uint32_t forward_state = RS_GROUTER_ACK_STATE_UNKNOWN ;
+ uint32_t next_state = it->second.status_flags;
+ uint32_t forward_state = RS_GROUTER_ACK_STATE_UNKN ;
+ bool update_routing_matrix = false ;
+
+ time_t now = time(NULL) ;
switch(item->state)
{
@@ -779,26 +800,11 @@ void p3GRouter::handleRecvACKItem(RsGRouterACKItem *item)
next_state = RS_GROUTER_ROUTING_STATE_ARVD ;
- {
-#warning UNFINISHED code.
+ update_routing_matrix = true ;
+ break ;
+
- // Now compute the weight for that particular item. See with what probabilities it was chosen.
- //
- // The real formula should be:
- // weight = w(ACK type) / probability
- //
- // ... where probability is the probability with whitch the item was sent in the first place.
- //
- // The time should also be set so that the routing clue has less importance.
- //
- float weight = (item->state == RS_GROUTER_ACK_STATE_RCVD)?1.0f : 0.5;
-#ifdef GROUTER_DEBUG
- std::cerr << " weight = " << weight << std::endl;
-#endif
- _routing_matrix.addRoutingClue(it->second.data_item->destination_key,item->PeerId(),weight) ;
- }
-
- case RS_GROUTER_ACK_STATE_GIVEN_UP: // route is bad. We forward back and update the routing matrix.
+ case RS_GROUTER_ACK_STATE_GVNP: // route is bad. We forward back and update the routing matrix.
break ;
}
@@ -812,16 +818,46 @@ void p3GRouter::handleRecvACKItem(RsGRouterACKItem *item)
// Just decrement the list of tried friends
//
+ bool found = false ;
+
for(std::list<FriendTrialRecord>::iterator it2(it->second.tried_friends.begin());it2!=it->second.tried_friends.end();++it2)
if( (*it2).friend_id == item->PeerId())
{
+ if(update_routing_matrix)
+ {
+ // Now compute the weight for that particular item. See with what probabilities it was chosen.
+ //
+ // The real formula should be:
+ // weight = w(ACK type) / probability
+ //
+ // ... where probability is the probability with whitch the item was sent in the first place.
+ //
+ // The time should also be set so that the routing clue has less importance.
+ //
+ float base = (item->state == RS_GROUTER_ACK_STATE_RCVD)?1.0f : 0.5 ;
+ uint32_t time_shift = now - (*it2).time_stamp ;
+ float probability = (*it2).probability;
+
+ float weight = computeMatrixContribution(base,time_shift,probability) ;
+#ifdef GROUTER_DEBUG
+ std::cerr << " base contrib = " << base << std::endl;
+ std::cerr << " time shift = " << time_shift << std::endl;
+ std::cerr << " sendind proba = " << probability << std::endl;
+ std::cerr << " ==> final weight : " << weight << std::endl;
+#endif
+ _routing_matrix.addRoutingClue(it->second.destination_key,item->PeerId(),weight) ;
+ }
#ifdef GROUTER_DEBUG
std::cerr << " Removing friend try for peer " << item->PeerId() << ". " << it->second.tried_friends.size() << " tries left." << std::endl;
#endif
it->second.tried_friends.erase(it2) ;
+ found = true ;
break ;
}
+ if(!found)
+ std::cerr << " (EE) friend try not found!! This should not happen. Needs debugging." << std::endl;
+
if(it->second.tried_friends.empty())
{
delete it->second.data_item ;
@@ -838,7 +874,6 @@ void p3GRouter::handleRecvACKItem(RsGRouterACKItem *item)
forward_state = RS_GROUTER_ACK_STATE_GVNP ;
}
}
- it->second.last_activity = time(NULL) ;
// Now send an ACK if necessary.
//
@@ -849,13 +884,14 @@ void p3GRouter::handleRecvACKItem(RsGRouterACKItem *item)
std::cerr << "ACK triage phase ended. Next state = " << statusString[next_state] << ", forwarded ack=" << ackString[forward_state] << std::endl;
#endif
- if(forward_state != RS_GROUTER_ACK_STATE_UNKNOWN && it->second.origin != mLinkMgr->getOwnId())
+ if(forward_state != RS_GROUTER_ACK_STATE_UNKN && it->second.origin != mLinkMgr->getOwnId())
{
#ifdef GROUTER_DEBUG
std::cerr << " forwarding ACK to origin: " << it->second.origin.toStdString() << std::endl;
#endif
sendACK(it->second.origin,item->mid,item->state) ;
}
+ it->second.status_flags = next_state ;
}
void p3GRouter::handleRecvDataItem(RsGRouterGenericDataItem *item)
@@ -929,9 +965,10 @@ void p3GRouter::handleRecvDataItem(RsGRouterGenericDataItem *item)
info.data_item = item->duplicate() ;
item_copy = info.data_item ;
- info.origin = RsPeerId(item->PeerId()) ;
+ info.origin = item->PeerId() ;
info.received_time = time(NULL) ;
- info.last_activity = info.received_time ;
+ info.last_sent = 0 ;
+ info.destination_key = item->destination_key ;
info.status_flags = RS_GROUTER_ROUTING_STATE_PEND ;
_pending_messages[item->routing_id] = info ;
@@ -981,7 +1018,6 @@ void p3GRouter::handleRecvDataItem(RsGRouterGenericDataItem *item)
if(returned_ack != RS_GROUTER_ACK_STATE_UNKN)
sendACK(item->PeerId(),item->routing_id,returned_ack) ;
- itr->second.last_activity = now ;
_changed = true ;
}
@@ -1003,10 +1039,11 @@ void p3GRouter::sendData(const GRouterKeyId& destination, RsGRouterGenericDataIt
info.data_item = item ;
info.status_flags = RS_GROUTER_ROUTING_STATE_PEND ;
- info.origin = RsPeerId(mLinkMgr->getOwnId()) ;
+ info.origin = mLinkMgr->getOwnId() ;
info.data_item->randomized_distance = 0 ;
- info.last_activity = now ;
+ info.last_sent = 0 ;
info.received_time = now ;
+ info.destination_key = destination ;
// Make sure we have a unique id (at least locally).
//
@@ -1019,8 +1056,8 @@ void p3GRouter::sendData(const GRouterKeyId& destination, RsGRouterGenericDataIt
#ifdef GROUTER_DEBUG
std::cerr << "p3GRouter::sendGRouterData(): pushing the followign item in the msg pending list:" << std::endl;
std::cerr << " data_item.size = " << info.data_item->data_size << std::endl;
- std::cerr << " data_item.byte = " << info.data_item->data_bytes << std::endl;
- std::cerr << " destination = " << info.data_item->destination_key << std::endl;
+ std::cerr << " data_item.byte = " << RsDirUtil::sha1sum(info.data_item->data_bytes,info.data_item->data_size) << std::endl;
+ std::cerr << " destination = " << info.destination_key << std::endl;
std::cerr << " status = " << info.status_flags << std::endl;
std::cerr << " distance = " << info.data_item->randomized_distance << std::endl;
std::cerr << " origin = " << info.origin.toStdString() << std::endl;
@@ -1150,10 +1187,10 @@ bool p3GRouter::getRoutingCacheInfo(std::vector<GRouterRoutingCacheInfo>& infos)
cinfo.mid = it->first ;
cinfo.local_origin = it->second.origin ;
- cinfo.destination = it->second.data_item->destination_key ;
+ cinfo.destination = it->second.destination_key ;
cinfo.time_stamp = it->second.received_time ;
cinfo.status = it->second.status_flags ;
- cinfo.data_size = it->second.data_item->data_size ;
+ cinfo.data_size = (it->second.data_item==NULL)?0:(it->second.data_item->data_size) ;
}
return true ;
}
@@ -1197,12 +1234,9 @@ void p3GRouter::debugDump()
for(std::map<GRouterMsgPropagationId, GRouterRoutingInfo>::iterator it(_pending_messages.begin());it!=_pending_messages.end();++it)
{
- std::cerr << " Msg id: " << std::hex << it->first << std::dec
- << " Local Origin: " << it->second.origin.toStdString() ;
- if(it->second.data_item != NULL)
- std::cerr << " Destination: " << it->second.data_item->destination_key ;
- if(!it->second.tried_friends.empty())
- std::cerr << " Time : " << now - it->second.tried_friends.front().time_stamp << " secs ago.";
+ std::cerr << " Msg id: " << std::hex << it->first << std::dec << " Local Origin: " << it->second.origin.toStdString() ;
+ std::cerr << " Destination: " << it->second.destination_key ;
+ std::cerr << " Time : " << now - it->second.last_sent << " secs ago.";
std::cerr << " Status: " << statusString[it->second.status_flags] << std::endl;
}
diff --git a/libretroshare/src/grouter/p3grouter.h b/libretroshare/src/grouter/p3grouter.h
index 7184ae926..cc5f9ec4e 100644
--- a/libretroshare/src/grouter/p3grouter.h
+++ b/libretroshare/src/grouter/p3grouter.h
@@ -156,6 +156,7 @@ class p3GRouter: public RsGRouter, public p3Service, public p3Config
//
static uint32_t computeBranchingFactor(const std::vector<RsPeerId>& friends,uint32_t dist) ;
static std::set<uint32_t> computeRoutingFriends(const std::vector<RsPeerId>& friends,const std::vector<float>& probas,uint32_t N) ;
+ static float computeMatrixContribution(float base,uint32_t time_shift,float probability) ;
uint32_t computeRandomDistanceIncrement(const RsPeerId& pid,const GRouterKeyId& destination_id) ;
diff --git a/libretroshare/src/retroshare/rsgrouter.h b/libretroshare/src/retroshare/rsgrouter.h
index 6ee9d20d3..fb349a55e 100644
--- a/libretroshare/src/retroshare/rsgrouter.h
+++ b/libretroshare/src/retroshare/rsgrouter.h
@@ -26,10 +26,11 @@
#pragma once
#include "util/rsdir.h"
-#include "grouter/groutertypes.h"
#include "retroshare/rsids.h"
typedef GRouterKeyIdType GRouterKeyId ; // we use SSLIds, so that it's easier in the GUI to mix up peer ids with grouter ids.
+typedef uint32_t GRouterServiceId ;
+typedef uint64_t GRouterMsgPropagationId ;
class GRouterClientService ;
class RsGRouterGenericDataItem ;