mirror of
https://github.com/RetroShare/RetroShare.git
synced 2024-10-01 02:35:48 -04:00
progress in grouter code
git-svn-id: http://svn.code.sf.net/p/retroshare/code/trunk@6955 b45a01b8-16f6-495d-af2f-9b41ad6348cc
This commit is contained in:
parent
dbfe3f6c51
commit
fd994a546b
@ -67,5 +67,9 @@ class GRouterCache
|
||||
// Cache of which message is pending, waiting for an ACK, etc.
|
||||
//
|
||||
std::map<GRouterMessageId,GRouterCacheInfo> _cache_info ;
|
||||
|
||||
// debug stuff
|
||||
//
|
||||
void debugDump() ;
|
||||
};
|
||||
|
||||
|
@ -124,6 +124,19 @@ RsGRouterItem *RsGRouterSerialiser::deserialise_RsGRouterACKItem(void *data, uin
|
||||
return NULL ;
|
||||
}
|
||||
|
||||
RsGRouterGenericDataItem *RsGRouterGenericDataItem::duplicate() const
|
||||
{
|
||||
RsGRouterGenericDataItem *item = new RsGRouterGenericDataItem ;
|
||||
*item = *this ; // copies everything.
|
||||
|
||||
// then duplicate the memory chunk
|
||||
|
||||
item->data_bytes = (uint8_t*)malloc(data_size) ;
|
||||
memcpy(item->data_bytes,data_bytes,data_size) ;
|
||||
|
||||
return item ;
|
||||
}
|
||||
|
||||
uint32_t RsGRouterGenericDataItem::serial_size() const
|
||||
{
|
||||
std::cerr << "(II) " << __PRETTY_FUNCTION__ << " not yet implemented!" << std::endl;
|
||||
@ -167,7 +180,7 @@ std::ostream& RsGRouterACKItem::print(std::ostream& o, uint16_t)
|
||||
{
|
||||
o << "RsGRouterACKItem:" << std::endl ;
|
||||
o << " direct origin: \""<< PeerId() << "\"" << std::endl ;
|
||||
o << " Key: " << destination_key.toStdString() << std::endl ;
|
||||
o << " Mid: " << mid << std::endl ;
|
||||
o << " State: " << state << std::endl ;
|
||||
|
||||
return o ;
|
||||
|
@ -28,8 +28,6 @@
|
||||
#include "serialiser/rsserial.h"
|
||||
#include "rsgrouter.h"
|
||||
|
||||
typedef uint32_t GRouterKeyPropagationId ;
|
||||
|
||||
// To be put in serialiser/rsserviceids.h
|
||||
static const uint8_t RS_SERVICE_TYPE_GROUTER = 0x0016 ;
|
||||
|
||||
@ -37,13 +35,15 @@ const uint8_t RS_PKT_SUBTYPE_GROUTER_PUBLISH_KEY = 0x01 ; // used to publish a
|
||||
const uint8_t RS_PKT_SUBTYPE_GROUTER_DATA = 0x02 ; // used to send data to a destination
|
||||
const uint8_t RS_PKT_SUBTYPE_GROUTER_ACK = 0x03 ; // acknowledgement of data received
|
||||
|
||||
const uint8_t QOS_PRIORITY_RS_GROUTER_PUBLISH_KEY = 3 ;
|
||||
const uint8_t QOS_PRIORITY_RS_GROUTER_PUBLISH_KEY = 3 ; // slow items. No need to congest the network with this.
|
||||
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
|
||||
|
||||
/***********************************************************************************/
|
||||
/* Basic GRouter Item Class */
|
||||
@ -99,9 +99,13 @@ class RsGRouterGenericDataItem: public RsGRouterItem
|
||||
virtual void clear() {}
|
||||
virtual std::ostream& print(std::ostream &out, uint16_t indent = 0) ;
|
||||
|
||||
RsGRouterGenericDataItem *duplicate() const ;
|
||||
|
||||
// packet data
|
||||
//
|
||||
GRouterMsgPropagationId routing_id ;
|
||||
GRouterKeyId destination_key ;
|
||||
|
||||
uint32_t data_size ;
|
||||
uint8_t *data_bytes;
|
||||
};
|
||||
@ -119,7 +123,7 @@ class RsGRouterACKItem: public RsGRouterItem
|
||||
|
||||
// packet data
|
||||
//
|
||||
GRouterKeyId destination_key ; // key that was the destination for the current ACK
|
||||
GRouterMsgPropagationId mid ; // message id to which this ack is a response
|
||||
uint32_t state ; // packet was delivered, not delivered, bounced, etc
|
||||
};
|
||||
|
||||
|
@ -30,6 +30,7 @@
|
||||
|
||||
typedef uint32_t GRouterServiceId ;
|
||||
typedef uint32_t GRouterKeyPropagationId ;
|
||||
typedef uint32_t GRouterMsgPropagationId ;
|
||||
|
||||
static const uint32_t RS_GROUTER_MATRIX_MAX_HIT_ENTRIES = 5;
|
||||
static const uint32_t RS_GROUTER_MATRIX_MIN_TIME_BETWEEN_HITS = 60; // can be set to up to half the publish time interval. Prevents flooding routes.
|
||||
@ -41,4 +42,7 @@ static const time_t RS_GROUTER_AUTOWASH_PERIOD = 60 ; // Autowas
|
||||
static const time_t RS_GROUTER_PUBLISH_CAMPAIGN_PERIOD = 1 *60 ; // Check for key advertising every 10 minutes
|
||||
static const time_t RS_GROUTER_PUBLISH_KEY_TIME_INTERVAL = 2 *60 ; // Advertise each key once a day at most.
|
||||
|
||||
|
||||
static const uint32_t RS_GROUTER_ROUTING_STATE_UNKN = 0x0000 ; // unknown. Unused.
|
||||
static const uint32_t RS_GROUTER_ROUTING_STATE_PEND = 0x0001 ; // item is pending. Should be sent asap.
|
||||
static const uint32_t RS_GROUTER_ROUTING_STATE_SENT = 0x0002 ; // item is sent. Waiting for answer
|
||||
static const uint32_t RS_GROUTER_ROUTING_STATE_OWND = 0x0003 ; // item is origined here. Needed for retry policy.
|
||||
|
@ -31,6 +31,8 @@
|
||||
#include "grouteritems.h"
|
||||
#include "groutertypes.h"
|
||||
|
||||
static const uint32_t RS_GROUTER_ROUTING_WAITING_TIME = 3600 ; // time between two trial of sending a message
|
||||
|
||||
p3GRouter::p3GRouter(p3LinkMgr *lm)
|
||||
: p3Service(RS_SERVICE_TYPE_GROUTER), p3Config(CONFIG_TYPE_GROUTER), mLinkMgr(lm), grMtx("GRouter")
|
||||
{
|
||||
@ -58,7 +60,7 @@ int p3GRouter::tick()
|
||||
{
|
||||
// route pending objects
|
||||
//
|
||||
routeObjects() ;
|
||||
routePendingObjects() ;
|
||||
|
||||
last_autowash_time = now ;
|
||||
autoWash() ;
|
||||
@ -105,13 +107,161 @@ void p3GRouter::autoWash()
|
||||
// cleanup cache
|
||||
}
|
||||
|
||||
void p3GRouter::routeObjects()
|
||||
void p3GRouter::routePendingObjects()
|
||||
{
|
||||
// Go through list of published keys
|
||||
// broadcast a publishKeyItem for each of them.
|
||||
//
|
||||
// The routing rules are the following:
|
||||
//
|
||||
// Go through list of cached routing objects. For each object:
|
||||
// if(Last try is old)
|
||||
// put the object in pending list
|
||||
//
|
||||
// (This loop is costly (could handle lots of items), so it should be done less often.)
|
||||
//
|
||||
// Add peer to the list of tried routes with time stamp
|
||||
// Keep the list of tried friends
|
||||
//
|
||||
// Go through list of pendign objects. For each object:
|
||||
// Select one route direction
|
||||
// - according to current probabilities from the routing matrix
|
||||
// - according to list of previous attempts
|
||||
//
|
||||
// if(route found)
|
||||
// forward item, update state and time stamp
|
||||
// else
|
||||
// if(I am not the sender)
|
||||
// send back ACK(given up) // if I am the sender, I will keep trying.
|
||||
//
|
||||
// Item has received an ACK
|
||||
//
|
||||
// ACK: given up => change route
|
||||
// ACK: received => Remove item from routed items
|
||||
//
|
||||
// The list in _pending_messages is necessarily short and most of the time empty. Once
|
||||
// treated, objects are stored in _routing_cache, where they wait for an answer.
|
||||
|
||||
std::cerr << "p3GRouter::routeObjects() Unimplemented !!" << std::endl;
|
||||
time_t now = time(NULL) ;
|
||||
|
||||
std::cerr << "p3GRouter::routeObjects() triage phase:" << std::endl;
|
||||
std::cerr << "Cached Items : " << _pending_messages.size() << std::endl;
|
||||
|
||||
std::list<std::string> lst_tmp ;
|
||||
std::list<SSLIdType> lst ;
|
||||
mLinkMgr->getOnlineList(lst_tmp) ;
|
||||
SSLIdType own_id( mLinkMgr->getOwnId() );
|
||||
|
||||
for(std::list<std::string>::const_iterator it(lst_tmp.begin());it!=lst_tmp.end();++it)
|
||||
lst.push_back(SSLIdType(*it)) ;
|
||||
|
||||
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)
|
||||
{
|
||||
std::cerr << " Msg id: " << std::hex << it->first << std::dec << std::endl;
|
||||
std::cerr << " Origin: " << it->second.origin.toStdString() << std::endl;
|
||||
std::cerr << " Last : " << it->second.tried_friends.front().friend_id.toStdString() << std::endl;
|
||||
std::cerr << " Time : " << it->second.tried_friends.front().time_stamp << std::endl;
|
||||
std::cerr << " Flags : " << it->second.status_flags << std::endl;
|
||||
std::cerr << " Probabilities: " << std::endl;
|
||||
|
||||
std::map<SSLIdType,float> probas ; // friends probabilities for online friend list.
|
||||
SSLIdType routed_friend ; // friend chosen for the next hop
|
||||
float best_proba = 0.0f; // temp variable used to select the best proba
|
||||
bool should_remove = false ; // should we remove this from the map?
|
||||
|
||||
// retrieve probabilities for this key.
|
||||
//
|
||||
if(! _routing_matrix.computeRoutingProbabilities(it->second.data_item->destination_key, lst, probas))
|
||||
{
|
||||
// key does not exist in routing matrix => send back an ACK(unknown)
|
||||
|
||||
std::cerr << " [Cannot compute. Unknown destination key!!] " << std::endl;
|
||||
|
||||
if(it->second.origin != own_id)
|
||||
{
|
||||
std::cerr << " removing item and notifying the sender (" << it->second.origin.toStdString() << ")" << std::endl;
|
||||
|
||||
sendACK(it->second.origin,it->first,RS_GROUTER_ACK_STATE_UNKNOWN) ;
|
||||
|
||||
// remove item from cache
|
||||
//
|
||||
should_remove = true ;
|
||||
}
|
||||
std::cerr << " item is ours. Keeping it until a route is known." << std::endl;
|
||||
|
||||
// else, select a routing friend at random, or just wait? Wait is probably better.
|
||||
}
|
||||
|
||||
bool friend_found = false ;
|
||||
|
||||
for(std::map<SSLIdType,float>::const_iterator it2(probas.begin());it2!=probas.end();++it2)
|
||||
{
|
||||
std::cerr << " " << it2->first.toStdString() << " : " << it2->second << std::endl;
|
||||
|
||||
// select the peer with highest probability that hasn't yet been tried.
|
||||
|
||||
if(it2->second > best_proba && !(it2->first == it->second.tried_friends.front().friend_id))
|
||||
{
|
||||
routed_friend = it2->first ;
|
||||
best_proba = it2->second ;
|
||||
friend_found = true ;
|
||||
}
|
||||
}
|
||||
|
||||
std::cerr << " Best route: " << routed_friend.toStdString() << ", with probability " << best_proba << std::endl;
|
||||
|
||||
// now, send the item.
|
||||
|
||||
if(friend_found)
|
||||
{
|
||||
// make a deep copy of the item
|
||||
RsGRouterGenericDataItem *new_item = it->second.data_item->duplicate() ;
|
||||
|
||||
// update cache entry
|
||||
FriendTrialRecord ftr ;
|
||||
ftr.time_stamp = now ;
|
||||
ftr.friend_id = routed_friend ;
|
||||
it->second.tried_friends.push_front(ftr) ;
|
||||
it->second.status_flags |= RS_GROUTER_ROUTING_STATE_SENT ;
|
||||
|
||||
std::cerr << " Sending..." << std::endl;
|
||||
// send
|
||||
new_item->PeerId(routed_friend.toStdString()) ;
|
||||
sendItem(new_item) ;
|
||||
}
|
||||
else if(!(it->second.status_flags & RS_GROUTER_ROUTING_STATE_OWND) || std::find(lst.begin(),lst.end(),it->second.origin) != lst.end())
|
||||
{
|
||||
// There's no correct friend to send this item to. We keep it for a while. If it's too old,
|
||||
// we discard it. For now, the procedure is to send back an ACK.
|
||||
|
||||
std::cerr << " Item has no route candidate. It's too old. " << std::endl;
|
||||
std::cerr << " sending ACK(no route) to peer " << it->second.origin.toStdString() << std::endl;
|
||||
|
||||
sendACK(it->second.origin,it->first,RS_GROUTER_ACK_STATE_NO_ROUTE) ;
|
||||
|
||||
should_remove = true ;
|
||||
}
|
||||
|
||||
if(should_remove)
|
||||
{
|
||||
// We remove from the map. That means the RsItem* has been transfered to somewhere else.
|
||||
//
|
||||
std::cerr << " Removing item from pending items" << std::endl;
|
||||
|
||||
std::map<GRouterMsgPropagationId, GRouterRoutingInfo>::iterator tmp(it) ;
|
||||
delete it->second.data_item ;
|
||||
++tmp ;
|
||||
_pending_messages.erase(it) ;
|
||||
it = tmp ;
|
||||
}
|
||||
else
|
||||
++it ;
|
||||
}
|
||||
else
|
||||
std::cerr << "Skipping " << std::hex << it->first << std::dec << ", dest=" << it->second.data_item->destination_key.toStdString() << ", state = " << it->second.status_flags << ", stamp=" << it->second.tried_friends.front().time_stamp << " - " << it->second.tried_friends.front().friend_id.toStdString() << std::endl;
|
||||
}
|
||||
|
||||
void p3GRouter::publishKeys()
|
||||
{
|
||||
// Go through list of published keys
|
||||
@ -236,18 +386,53 @@ void p3GRouter::handleRecvPublishKeyItem(RsGRouterPublishKeyItem *item)
|
||||
|
||||
void p3GRouter::handleRecvACKItem(RsGRouterACKItem *item)
|
||||
{
|
||||
std::cerr << "Received data item item for key " << item->destination_key.toStdString() << std::endl;
|
||||
std::cerr << "Received ACK item, mid=" << (void*)item->mid << std::endl;
|
||||
|
||||
// find the item in the pendign list,
|
||||
// - if not found, drop.
|
||||
//
|
||||
// ...and act appropriately:
|
||||
// - item was
|
||||
// - if we're origin
|
||||
// notify the client service
|
||||
// else
|
||||
// remove item
|
||||
//
|
||||
std::map<GRouterMsgPropagationId,GRouterRoutingInfo>::iterator it(_pending_messages.find(item->mid)) ;
|
||||
|
||||
if(it == _pending_messages.end())
|
||||
{
|
||||
std::cerr << "p3GRouter::handleRecvACKItem(): cannot find entry for message id " << std::hex << item->mid << std::dec << ". Dropping it." << std::endl;
|
||||
return ;
|
||||
}
|
||||
switch(item->state)
|
||||
{
|
||||
case RS_GROUTER_ACK_STATE_RECEIVED_INDIRECTLY:
|
||||
case RS_GROUTER_ACK_STATE_RECEIVED: // received: - do we update the routing matrix?
|
||||
// - and forward back
|
||||
// Do nothing. It was indirectly received: fine. We don't need to notify the origin
|
||||
// otherwise lots of ACKs will flow to the origin.
|
||||
break ;
|
||||
case RS_GROUTER_ACK_STATE_RECEIVED:
|
||||
// Notify the origin. This is the main route and it was successful.
|
||||
|
||||
std::cerr << " forwarding ACK to origin: " << it->second.origin.toStdString() << std::endl;
|
||||
|
||||
sendACK(it->second.origin,item->mid,RS_GROUTER_ACK_STATE_RECEIVED) ;
|
||||
break ;
|
||||
|
||||
case RS_GROUTER_ACK_STATE_GIVEN_UP: // route is bad. We forward back and update the routing matrix.
|
||||
break ;
|
||||
}
|
||||
|
||||
if(it->second.status_flags & RS_GROUTER_ROUTING_STATE_OWND)
|
||||
{
|
||||
// find the client service and notify it.
|
||||
std::cerr << " We're owner: should notify client id" << std::endl;
|
||||
}
|
||||
|
||||
// Always remove the item.
|
||||
//
|
||||
delete it->second.data_item ;
|
||||
_pending_messages.erase(it) ;
|
||||
}
|
||||
|
||||
void p3GRouter::handleRecvDataItem(RsGRouterGenericDataItem *item)
|
||||
@ -268,6 +453,17 @@ void p3GRouter::sendData(const GRouterKeyId& destination, void *& item_data,uint
|
||||
std::cerr << "(WW) " << __PRETTY_FUNCTION__ << ": not implemented." << std::endl;
|
||||
}
|
||||
|
||||
void p3GRouter::sendACK(const SSLIdType& peer, GRouterMsgPropagationId mid, uint32_t ack_flags)
|
||||
{
|
||||
RsGRouterACKItem *item = new RsGRouterACKItem ;
|
||||
|
||||
item->state = ack_flags ;
|
||||
item->mid = mid ;
|
||||
item->PeerId(peer.toStdString()) ;
|
||||
|
||||
sendItem(item) ;
|
||||
}
|
||||
|
||||
bool p3GRouter::loadList(std::list<RsItem*>& items)
|
||||
{
|
||||
std::cerr << "(WW) " << __PRETTY_FUNCTION__ << ": not implemented." << std::endl;
|
||||
@ -276,6 +472,11 @@ bool p3GRouter::loadList(std::list<RsItem*>& items)
|
||||
bool p3GRouter::saveList(bool&,std::list<RsItem*>& items)
|
||||
{
|
||||
std::cerr << "(WW) " << __PRETTY_FUNCTION__ << ": not implemented." << std::endl;
|
||||
|
||||
// We save
|
||||
// - the routing clues
|
||||
// - the pending items
|
||||
|
||||
return false ;
|
||||
}
|
||||
|
||||
|
@ -53,6 +53,23 @@ class GRouterPublishedKeyInfo
|
||||
time_t validity_time ;
|
||||
};
|
||||
|
||||
struct FriendTrialRecord
|
||||
{
|
||||
SSLIdType friend_id ; // id of the friend
|
||||
time_t time_stamp ; // time of the last tried
|
||||
};
|
||||
|
||||
class GRouterRoutingInfo
|
||||
{
|
||||
public:
|
||||
RsGRouterGenericDataItem *data_item ;
|
||||
|
||||
uint32_t status_flags ; // pending, waiting, etc.
|
||||
std::list<FriendTrialRecord> tried_friends ; // list of friends to which the item was sent ordered with time.
|
||||
SSLIdType origin ; // which friend sent us that item
|
||||
time_t received_time ; // time at which the item was received
|
||||
};
|
||||
|
||||
class p3GRouter: public RsGRouter, public p3Service, public p3Config
|
||||
{
|
||||
public:
|
||||
@ -88,6 +105,11 @@ class p3GRouter: public RsGRouter, public p3Service, public p3Config
|
||||
//
|
||||
void sendData(const GRouterKeyId& destination, void *& item_data,uint32_t item_size) ;
|
||||
|
||||
// Sends an ACK to the origin of the msg. This is used to notify for
|
||||
// unfound route, or message correctly received, depending on the particular situation.
|
||||
//
|
||||
void sendACK(const SSLIdType& peer,GRouterMsgPropagationId mid, uint32_t flags) ;
|
||||
|
||||
//===================================================//
|
||||
// Interface with RsGRouter //
|
||||
//===================================================//
|
||||
@ -123,7 +145,7 @@ class p3GRouter: public RsGRouter, public p3Service, public p3Config
|
||||
|
||||
private:
|
||||
void autoWash() ;
|
||||
void routeObjects() ;
|
||||
void routePendingObjects() ;
|
||||
void handleIncoming() ;
|
||||
void publishKeys() ;
|
||||
void debugDump() ;
|
||||
@ -155,10 +177,11 @@ class p3GRouter: public RsGRouter, public p3Service, public p3Config
|
||||
GRouterMatrix _routing_matrix ;
|
||||
|
||||
// Stores the routing events.
|
||||
// - pending items
|
||||
// - ongoing requests, waiting for return ACK
|
||||
// - pending items
|
||||
// Both a stored in 2 different lists, to allow a more efficient handling.
|
||||
//
|
||||
GRouterCache _routing_cache ;
|
||||
std::map<GRouterMsgPropagationId, GRouterRoutingInfo> _pending_messages;// pending messages
|
||||
|
||||
// Stores the keys which identify the router's node. For each key, a structure holds:
|
||||
// - the client service
|
||||
|
Loading…
Reference in New Issue
Block a user