improvement of global router after first test phase. Fixed a number of bugs.

git-svn-id: http://svn.code.sf.net/p/retroshare/code/trunk@8141 b45a01b8-16f6-495d-af2f-9b41ad6348cc
This commit is contained in:
csoler 2015-04-18 19:59:27 +00:00
parent 1257f276d4
commit 057b2afa4e
13 changed files with 330 additions and 251 deletions

View File

@ -210,6 +210,7 @@ RsGRouterRoutingInfoItem *RsGRouterSerialiser::deserialise_RsGRouterRoutingInfoI
ok &= getRawUInt32(data, pktsize, &offset, &item->sending_attempts);
ok &= getRawUInt32(data, pktsize, &offset, &item->client_id);
ok &= item->item_hash.deserialise(data, pktsize, offset) ;
ok &= item->tunnel_hash.deserialise(data, pktsize, offset) ;
ok &= getRawUInt32(data, pktsize, &offset, &item->routing_flags) ;
@ -575,6 +576,7 @@ uint32_t RsGRouterRoutingInfoItem::serial_size() const
s += sizeof(GRouterServiceId) ; // service_id
s += tunnel_hash.serial_size() ;
s += item_hash.serial_size() ;
s += 4 ; // routing_flags
s += incoming_routes.TlvSize() ; // incoming_routes
@ -673,6 +675,7 @@ bool RsGRouterRoutingInfoItem::serialise(void *data,uint32_t& size) const
ok &= setRawUInt32(data, tlvsize, &offset, sending_attempts) ;
ok &= setRawUInt32(data, tlvsize, &offset, client_id) ;
ok &= item_hash.serialise(data, tlvsize, offset) ;
ok &= tunnel_hash.serialise(data, tlvsize, offset) ;
ok &= setRawUInt32(data, tlvsize, &offset, routing_flags) ;
@ -739,6 +742,7 @@ std::ostream& RsGRouterRoutingInfoItem::print(std::ostream& o, uint16_t)
o << " Sending attempts:"<< sending_attempts << std::endl ;
o << " destination key: "<< data_item->destination_key << std::endl ;
o << " Client id: "<< client_id << std::endl ;
o << " item hash: "<< item_hash << std::endl ;
o << " tunnel hash: "<< tunnel_hash << std::endl ;
o << " Data size: "<< data_item->data_size << std::endl ;
o << " Signed receipt: "<< (void*)receipt_item << std::endl ;

View File

@ -48,7 +48,7 @@ const uint8_t RS_PKT_SUBTYPE_GROUTER_MATRIX_CLUES = 0x80 ; // it
const uint8_t RS_PKT_SUBTYPE_GROUTER_FRIENDS_LIST = 0x82 ; // item to save friend lists
const uint8_t RS_PKT_SUBTYPE_GROUTER_ROUTING_INFO_deprecated = 0x87 ; // deprecated. Don't use.
const uint8_t RS_PKT_SUBTYPE_GROUTER_ROUTING_INFO_deprecated2 = 0x88 ; // item to save routing info
const uint8_t RS_PKT_SUBTYPE_GROUTER_ROUTING_INFO = 0x90 ; // deprecated. Don't use.
const uint8_t RS_PKT_SUBTYPE_GROUTER_ROUTING_INFO = 0x92 ; //
const uint8_t QOS_PRIORITY_RS_GROUTER = 4 ; // relevant for items that travel through friends

View File

@ -44,17 +44,23 @@ static const uint32_t RS_GROUTER_MIN_CONFIG_SAVE_PERIOD = 5 ; // at mos
static const float RS_GROUTER_BASE_WEIGHT_ROUTED_MSG = 1.0f ; // base contribution of routed message clue to routing matrix
static const float RS_GROUTER_BASE_WEIGHT_GXS_PACKET = 0.1f ; // base contribution of GXS message to routing matrix
static const uint32_t MAX_TUNNEL_WAIT_TIME = 60 ; // wait for 60 seconds at most for a tunnel response.
static const uint32_t MAX_TUNNEL_UNMANAGED_TIME = 600 ; // min time before retry tunnels for that msg.
static const uint32_t MAX_DELAY_BETWEEN_TWO_SEND = 120 ; // wait for 120 seconds before re-sending.
static const uint32_t TUNNEL_OK_WAIT_TIME = 2 ; // wait for 2 seconds after last tunnel ok, so that we have a complete set of tunnels.
static const uint32_t MAX_GROUTER_DATA_SIZE = 2*1024*1024 ; // 2MB size limit. This is of course arbitrary.
static const uint32_t MAX_RECEIPT_WAIT_TIME = 20 ; // wait for at most 20 secs for a receipt. If not, cancel.
static const uint32_t MAX_TRANSACTION_ACK_WAITING_TIME = 60 ; // wait for at most 60 secs for a ACK. If not restart the transaction.
static const uint32_t DIRECT_FRIEND_TRY_DELAY = 20 ; // wait for 20 secs if no friends available, then try tunnels.
static const uint32_t MAX_INACTIVE_DATA_PIPE_DELAY = 300 ; // clean inactive data pipes for more than 5 mins
static const time_t RS_GROUTER_DEBUG_OUTPUT_PERIOD = 10 ; // Output everything
static const time_t RS_GROUTER_AUTOWASH_PERIOD = 10 ; // Autowash every minute. Not a costly operation.
static const time_t RS_GROUTER_MATRIX_UPDATE_PERIOD = 1 *10 ; // Check for key advertising every 10 minutes
static const time_t RS_GROUTER_ROUTING_WAITING_TIME = 2 *60 ; // time between two trial of sending a given message
//atic 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 = 6*256 ; // 6 distance units. That is a lot.
static const uint32_t GROUTER_ITEM_MAX_CACHE_KEEP_TIME = 86400 ; // Cached items are kept for 24 hours at most.
static const uint32_t GROUTER_ITEM_MAX_CACHE_KEEP_TIME_DEAD= 3600 ; // DEAD Items are kept in cache for only 1 hour to favor re-exploring dead routes.
static const uint32_t RS_GROUTER_DATA_STATUS_UNKNOWN = 0x0000 ; // unknown. Unused.
static const uint32_t RS_GROUTER_DATA_STATUS_PENDING = 0x0001 ; // item is pending. Should be sent asap.
@ -69,7 +75,7 @@ static const uint32_t RS_GROUTER_SENDING_STATUS_FRIEND = 0x0002 ; // item wa
static const uint32_t RS_GROUTER_TUNNEL_STATUS_UNMANAGED = 0x0000 ; // no tunnel requested atm
static const uint32_t RS_GROUTER_TUNNEL_STATUS_PENDING = 0x0001 ; // tunnel requested to turtle
static const uint32_t RS_GROUTER_TUNNEL_STATUS_READY = 0x0002 ; // tunnel is ready but we're still waiting for various confirmations
static const uint32_t RS_GROUTER_TUNNEL_STATUS_CAN_SEND = 0x0003 ; // tunnel is ready and data can be sent
//static const uint32_t RS_GROUTER_TUNNEL_STATUS_CAN_SEND = 0x0003 ; // tunnel is ready and data can be sent
class FriendTrialRecord
{
@ -90,6 +96,7 @@ class GRouterRoutingInfo
public:
GRouterRoutingInfo()
{
data_transaction_TS = 0 ; // this is not serialised.
data_item = NULL ;
receipt_item = NULL ;
}
@ -111,6 +118,7 @@ public:
RsGRouterSignedReceiptItem *receipt_item ;
RsTlvPeerIdSet incoming_routes ;
Sha1CheckSum item_hash ; // used to check re-existance, consistency, etc.
// non serialised data
@ -119,5 +127,6 @@ public:
static const uint32_t ROUTING_FLAGS_ALLOW_TUNNELS = 0x0001;
static const uint32_t ROUTING_FLAGS_ALLOW_FRIENDS = 0x0002;
static const uint32_t ROUTING_FLAGS_IS_ORIGIN = 0x0004;
static const uint32_t ROUTING_FLAGS_IS_DESTINATION = 0x0008;
};

View File

@ -257,14 +257,7 @@
#define GROUTER_DEBUG
/**********************/
static const uint32_t MAX_TUNNEL_WAIT_TIME = 60 ; // wait for 60 seconds at most for a tunnel response.
static const uint32_t MAX_TUNNEL_UNMANAGED_TIME = 600 ; // min time before retry tunnels for that msg.
static const uint32_t MAX_DELAY_BETWEEN_TWO_SEND = 120 ; // wait for 120 seconds before re-sending.
static const uint32_t TUNNEL_OK_WAIT_TIME = 3 ; // wait for 3 seconds after last tunnel ok, so that we have a complete set of tunnels.
static const uint32_t MAX_GROUTER_DATA_SIZE = 2*1024*1024 ; // 2MB size limit. This is of course arbitrary.
static const uint32_t MAX_RECEIPT_WAIT_TIME = 20 ; // wait for at most 20 secs for a receipt. If not, cancel.
static const uint32_t MAX_TRANSACTION_ACK_WAITING_TIME = 60 ; // wait for at most 60 secs for a ACK. If not restart the transaction.
static const uint32_t DIRECT_FRIEND_TRY_DELAY = 20 ; // wait for 20 secs if no friends available, then try tunnels.
const std::string p3GRouter::SERVICE_INFO_APP_NAME = "Global Router" ;
@ -408,7 +401,9 @@ bool p3GRouter::unregisterKey(const RsGxsId& key_id,const GRouterServiceId& sid)
if(it == _owned_key_ids.end())
{
#ifdef GROUTER_DEBUG
std::cerr << "p3GRouter::unregisterKey(): key " << key_id << " not found." << std::endl;
#endif
return false ;
}
@ -611,6 +606,8 @@ void GRouterTunnelInfo::addVirtualPeer(const TurtleVirtualPeerId& vpid)
}
RsGRouterAbstractMsgItem *GRouterDataInfo::addDataChunk(RsGRouterTransactionChunkItem *chunk)
{
last_activity_TS = time(NULL) ;
if(incoming_data_buffer == NULL)
{
if(chunk->chunk_start != 0)
@ -807,8 +804,11 @@ if(!_pending_messages.empty())
{
#ifdef GROUTER_DEBUG
grouter_debug() << " " << std::hex << it->first << std::dec
<< " data_status=" << it->second.data_status << ", tunnel_status=" << it->second.tunnel_status
<< " last tried: "<< now - it->second.last_tunnel_request_TS << " (secs ago)" << ", last sent: " << now - it->second.last_tunnel_sent_TS << " (secs ago) " ;
<< ", data_status=" << it->second.data_status << ", tunnel_status=" << it->second.tunnel_status
<< ", last received: "<< now - it->second.received_time_TS << " (secs ago)"
<< ", last TR: "<< now - it->second.last_tunnel_request_TS << " (secs ago)"
<< ", last tunnel sent: " << now - it->second.last_tunnel_sent_TS << " (secs ago) "
<< ", last friend sent: " << now - it->second.last_friend_sent_TS << " (secs ago) " ;
#endif
if(it->second.data_status == RS_GROUTER_DATA_STATUS_PENDING && (it->second.routing_flags & GRouterRoutingInfo::ROUTING_FLAGS_ALLOW_TUNNELS))
@ -859,9 +859,10 @@ if(!_pending_messages.empty())
std::cerr << " closing pending tunnels." << std::endl;
#endif
mTurtle->stopMonitoringTunnels(it->second.tunnel_hash) ;
it->second.routing_flags &= ~GRouterRoutingInfo::ROUTING_FLAGS_ALLOW_TUNNELS ;
//it->second.tunnel_status = RS_GROUTER_TUNNEL_STATUS_UNMANAGED ;
it->second.routing_flags &= ~GRouterRoutingInfo::ROUTING_FLAGS_ALLOW_TUNNELS ;
it->second.tunnel_status = RS_GROUTER_TUNNEL_STATUS_UNMANAGED ;
//it->second.data_status = RS_GROUTER_DATA_STATUS_PENDING ;
}
#ifdef GROUTER_DEBUG
@ -870,17 +871,17 @@ if(!_pending_messages.empty())
#endif
// also check that all tunnels are actually active, to remove any old dead tunnels
if(it->second.tunnel_status == RS_GROUTER_TUNNEL_STATUS_READY)
{
std::map<TurtleFileHash,GRouterTunnelInfo>::iterator it2 = _tunnels.find(it->second.tunnel_hash) ;
if(it2 == _tunnels.end() || it2->second.virtual_peers.empty()) ;
{
std::cerr << " re-setting tunnel status to PENDING, as no tunnels are actually present." << std::endl;
it->second.tunnel_status = RS_GROUTER_TUNNEL_STATUS_PENDING ;
}
}
//
// if(it->second.tunnel_status == RS_GROUTER_TUNNEL_STATUS_READY)
// {
// std::map<TurtleFileHash,GRouterTunnelInfo>::iterator it2 = _tunnels.find(it->second.tunnel_hash) ;
//
// if(it2 == _tunnels.end() || it2->second.virtual_peers.empty()) ;
// {
// std::cerr << " re-setting tunnel status to PENDING, as no tunnels are actually present." << std::endl;
// it->second.tunnel_status = RS_GROUTER_TUNNEL_STATUS_PENDING ;
// }
// }
}
#ifdef GROUTER_DEBUG
if(!priority_list.empty())
@ -895,7 +896,7 @@ if(!_pending_messages.empty())
for(uint32_t i=0;i<priority_list.size();++i)
{
#ifdef GROUTER_DEBUG
grouter_debug() << " askign tunnel management for msg=" << priority_list[i].first << " hash=" << priority_list[i].second->tunnel_hash << std::endl;
grouter_debug() << " asking tunnel management for msg=" << priority_list[i].second->data_item->routing_id << " hash=" << priority_list[i].second->tunnel_hash << std::endl;
#endif
mTurtle->monitorTunnels(priority_list[i].second->tunnel_hash,this,false) ;
@ -903,8 +904,6 @@ if(!_pending_messages.empty())
priority_list[i].second->tunnel_status = RS_GROUTER_TUNNEL_STATUS_PENDING ;
priority_list[i].second->last_tunnel_request_TS = now ;
}
}
void p3GRouter::routePendingObjects()
@ -949,12 +948,16 @@ void p3GRouter::routePendingObjects()
if(peers.empty())
{
#ifdef GROUTER_DEBUG
std::cerr << " no direct friends available" << std::endl;
#endif
if(it->second.received_time_TS + DIRECT_FRIEND_TRY_DELAY < now &&
!(it->second.routing_flags & GRouterRoutingInfo::ROUTING_FLAGS_ALLOW_TUNNELS))
{
#ifdef GROUTER_DEBUG
std::cerr << " enabling tunnels for this message." << std::endl;
#endif
it->second.routing_flags |= GRouterRoutingInfo::ROUTING_FLAGS_ALLOW_TUNNELS ;
}
continue ;
@ -985,7 +988,9 @@ void p3GRouter::routePendingObjects()
}
else if(it->second.data_status == RS_GROUTER_DATA_STATUS_ONGOING && now > MAX_TRANSACTION_ACK_WAITING_TIME + it->second.data_transaction_TS)
{
#ifdef GROUTER_DEBUG
std::cerr << " waited too long for this transation. Switching back to PENDING." << std::endl;
#endif
it->second.data_status = RS_GROUTER_DATA_STATUS_PENDING ;
}
@ -993,15 +998,19 @@ void p3GRouter::routePendingObjects()
{
// send the receipt through all incoming routes, as soon as it gets delivered.
#ifdef GROUTER_DEBUG
std::cerr << " receipt should be sent back. Trying all incoming routes..." << std::endl;
#endif
std::list<RsGRouterTransactionChunkItem*> chunks ;
sliceDataItem(it->second.receipt_item,chunks) ;
for(std::set<RsPeerId>::iterator it2=it->second.incoming_routes.ids.begin();it2!=it->second.incoming_routes.ids.end();)
if(mServiceControl->isPeerConnected(getServiceInfo().mServiceType,*it2))
if(mServiceControl->isPeerConnected(getServiceInfo().mServiceType,*it2) || mTurtle->isTurtlePeer(*it2))
{
#ifdef GROUTER_DEBUG
std::cerr << " sending receipt back to " << *it2 << " which is online." << std::endl;
#endif
for(std::list<RsGRouterTransactionChunkItem*>::const_iterator it3(chunks.begin());it3!=chunks.end();++it3)
locked_sendTransactionData(*it2,*(*it3) ) ;
@ -1062,7 +1071,9 @@ void p3GRouter::locked_collectAvailableFriends(const GRouterKeyId& gxs_id,std::l
uint32_t max_count = is_origin?3:1 ;
float probability_threshold = is_origin?0.0:0.5 ;
std::cerr << " position at origin: " << is_origin << " => mac_count=" << max_count << ", proba threashold=" << probability_threshold << std::endl;
#ifdef GROUTER_DEBUG
std::cerr << " position at origin: " << is_origin << " => mac_count=" << max_count << ", proba threshold=" << probability_threshold << std::endl;
#endif
std::vector<std::pair<float,RsPeerId> > mypairs ;
@ -1080,7 +1091,9 @@ void p3GRouter::locked_collectAvailableFriends(const GRouterKeyId& gxs_id,std::l
if( (*it).first >= probability_threshold )
{
friend_peers.push_back( (*it).second ), ++n ;
#ifdef GROUTER_DEBUG
std::cerr << " keeping " << (*it).second << std::endl;
#endif
if(!is_origin) // only collect one peer if we're not at origin.
break ;
@ -1175,14 +1188,13 @@ bool p3GRouter::locked_sendTransactionData(const RsPeerId& pid,const RsGRouterTr
void p3GRouter::autoWash()
{
bool items_deleted = false ;
time_t now = time(NULL) ;
std::map<GRouterMsgPropagationId,GRouterClientService *> failed_msgs ;
{
RS_STACK_MUTEX(grMtx) ;
time_t now = time(NULL) ;
for(std::map<GRouterMsgPropagationId, GRouterRoutingInfo>::iterator it=_pending_messages.begin();it!=_pending_messages.end();)
if( it->second.data_status == RS_GROUTER_DATA_STATUS_DONE
|| ((it->second.received_time_TS + GROUTER_ITEM_MAX_CACHE_KEEP_TIME < now)
@ -1225,21 +1237,42 @@ void p3GRouter::autoWash()
if(!mTurtle->isTurtlePeer(*it2))
{
vpids_to_remove.push_back(*it2) ;
#ifdef GROUTER_DEBUG
std::cerr << " " << *it2 << " is not an active tunnel for hash " << it->first << ". Removing virtual peer id." << std::endl;
#endif
}
for(std::list<TurtleVirtualPeerId>::const_iterator it2=vpids_to_remove.begin();it2!=vpids_to_remove.end();++it2)
it->second.removeVirtualPeer(*it2) ;
}
// Also clean incoming data pipes
for(std::map<RsPeerId,GRouterDataInfo>::iterator it(_incoming_data_pipes.begin());it!=_incoming_data_pipes.end();)
if(it->second.last_activity_TS + MAX_INACTIVE_DATA_PIPE_DELAY < now)
{
#ifdef GROUTER_DEBUG
std::cerr << " removing data pipe for peer " << it->first << " which is too old." << std::endl;
#endif
std::map<RsPeerId,GRouterDataInfo>::iterator ittmp = it ;
++ittmp ;
it->second.clear() ;
_incoming_data_pipes.erase(it) ;
it == ittmp ;
}
// look into pending items.
else
++it ;
}
// Look into pending items.
for(std::map<GRouterMsgPropagationId,GRouterClientService*>::const_iterator it(failed_msgs.begin());it!=failed_msgs.end();++it)
{
#ifdef GROUTER_DEBUG
std::cerr << " notifying client for message id " << std::hex << it->first << " state = FAILED" << std::endl;
#endif
it->second->notifyDataStatus(it->first ,GROUTER_CLIENT_SERVICE_DATA_STATUS_FAILED) ;
}
#warning should we also clean incoming data pipes?
if(items_deleted)
_changed = true ;
@ -1366,8 +1399,9 @@ void p3GRouter::handleIncomingReceiptItem(RsGRouterSignedReceiptItem *receipt_it
return ;
}
// check signature.
if(receipt_item->data_hash != RsDirUtil::sha1sum(it->second.data_item->data_bytes,it->second.data_item->data_size))
// check hash.
if(receipt_item->data_hash != it->second.item_hash)
{
std::cerr << " checking receipt hash : FAILED. Receipt is dropped." << std::endl;
return ;
@ -1376,6 +1410,7 @@ void p3GRouter::handleIncomingReceiptItem(RsGRouterSignedReceiptItem *receipt_it
else
std::cerr << " checking receipt hash : OK" << std::endl;
#endif
// check signature.
if(! verifySignedDataItem(receipt_item))
{
@ -1393,6 +1428,14 @@ void p3GRouter::handleIncomingReceiptItem(RsGRouterSignedReceiptItem *receipt_it
std::cerr << " message is at origin. Setting message transmission to DONE" << std::endl;
#endif
it->second.data_status = RS_GROUTER_DATA_STATUS_DONE;
if(locked_getClientAndServiceId(it->second.tunnel_hash,it->second.data_item->destination_key,client_service,service_id))
mid = it->first ;
else
{
mid = 0 ;
std::cerr << " ERROR: cannot retrieve service ID for message " << std::hex << it->first << std::dec << std::endl;
}
}
else
{
@ -1403,14 +1446,6 @@ void p3GRouter::handleIncomingReceiptItem(RsGRouterSignedReceiptItem *receipt_it
it->second.receipt_item = receipt_item->duplicate() ;
}
if(locked_getClientAndServiceId(it->second.tunnel_hash,it->second.data_item->destination_key,client_service,service_id))
mid = it->first ;
else
{
mid = 0 ;
std::cerr << " ERROR: cannot retrieve service ID for message " << std::hex << it->first << std::dec << std::endl;
}
changed = true ;
}
@ -1426,6 +1461,20 @@ void p3GRouter::handleIncomingReceiptItem(RsGRouterSignedReceiptItem *receipt_it
IndicateConfigChanged() ;
}
Sha1CheckSum p3GRouter::computeDataItemHash(RsGRouterGenericDataItem *data_item)
{
uint32_t total_size = data_item->signed_data_size() + data_item->signature.TlvSize() ;
RsTemporaryMemory mem(total_size) ;
uint32_t offset = 0 ;
data_item->serialise_signed_data(mem,total_size) ;
offset += data_item->signed_data_size() ;
data_item->signature.SetTlv(mem, total_size,&offset) ;
return RsDirUtil::sha1sum(mem,total_size) ;
}
void p3GRouter::handleIncomingDataItem(RsGRouterGenericDataItem *data_item)
{
#ifdef GROUTER_DEBUG
@ -1436,6 +1485,7 @@ void p3GRouter::handleIncomingDataItem(RsGRouterGenericDataItem *data_item)
GRouterClientService *client = NULL ;
GRouterServiceId service_id = data_item->service_id ;
RsGRouterSignedReceiptItem *receipt_item = NULL ;
bool item_is_for_us = false ;
// Find client and service ID from destination key.
@ -1456,45 +1506,11 @@ void p3GRouter::handleIncomingDataItem(RsGRouterGenericDataItem *data_item)
item_is_for_us = _owned_key_ids.find( makeTunnelHash(data_item->destination_key,service_id) ) != _owned_key_ids.end() ;
}
if(!item_is_for_us)
if(item_is_for_us)
{
RS_STACK_MUTEX(grMtx) ;
std::cerr << " item is not for us. Storing/forwarding." << std::endl;
// item is not for us. We store it in pending messages and will deal with it later
std::map<GRouterMsgPropagationId,GRouterRoutingInfo>::iterator it = _pending_messages.find(data_item->routing_id) ;
if(it == _pending_messages.end())
{
std::cerr << " item is new. Storing it and forwarding it." << std::endl;
GRouterRoutingInfo& info(_pending_messages[data_item->routing_id]) ;
info.data_item = data_item->duplicate() ;
info.receipt_item = NULL ;
info.data_status = RS_GROUTER_DATA_STATUS_PENDING ;
info.tunnel_status = RS_GROUTER_TUNNEL_STATUS_UNMANAGED ;
info.last_tunnel_sent_TS = 0 ;
info.last_friend_sent_TS = 0 ;
info.last_tunnel_request_TS = 0 ;
info.sending_attempts = 0 ;
info.routing_flags = GRouterRoutingInfo::ROUTING_FLAGS_ALLOW_FRIENDS ; // don't allow tunnels just yet
info.received_time_TS = time(NULL) ;
info.tunnel_hash = makeTunnelHash(data_item->destination_key,data_item->service_id) ;
}
std::cerr << " storing incoming route: from " << data_item->PeerId() << std::endl;
if(!mTurtle->isTurtlePeer(data_item->PeerId()))
_pending_messages[data_item->routing_id].incoming_routes.ids.insert(data_item->PeerId()) ;
return ;
}
// Item is for us.
// The item's signature must be checked, and the item needs to be decrypted.
#ifdef GROUTER_DEBUG
std::cerr << " item is for us. Checking signature." << std::endl;
#endif
if(verifySignedDataItem(data_item)) // we should get proper flags out of this
{
@ -1509,7 +1525,7 @@ void p3GRouter::handleIncomingDataItem(RsGRouterGenericDataItem *data_item)
// compute the hash before decryption.
Sha1CheckSum data_hash = RsDirUtil::sha1sum(data_item->data_bytes,data_item->data_size) ;
Sha1CheckSum data_hash = computeDataItemHash(data_item) ;
if(!decryptDataItem(data_item))
{
@ -1530,17 +1546,17 @@ void p3GRouter::handleIncomingDataItem(RsGRouterGenericDataItem *data_item)
// No we need to send a signed receipt to the sender.
RsGRouterSignedReceiptItem receipt_item ;
receipt_item.data_hash = data_hash ;
receipt_item.routing_id = data_item->routing_id ;
receipt_item.destination_key = data_item->signature.keyId ;
receipt_item.flags = 0 ;
receipt_item = new RsGRouterSignedReceiptItem;
receipt_item->data_hash = data_hash ;
receipt_item->routing_id = data_item->routing_id ;
receipt_item->destination_key = data_item->signature.keyId ;
receipt_item->flags = 0 ;
#ifdef GROUTER_DEBUG
std::cerr << " preparing signed receipt." << std::endl;
#endif
if(!signDataItem(&receipt_item,data_item->destination_key))
if(!signDataItem(receipt_item,data_item->destination_key))
{
std::cerr << " signing: FAILED. Receipt dropped. ERROR." << std::endl;
return ;
@ -1548,28 +1564,67 @@ void p3GRouter::handleIncomingDataItem(RsGRouterGenericDataItem *data_item)
#ifdef GROUTER_DEBUG
std::cerr << " signing: OK." << std::endl;
#endif
// Normally (proxy mode) we should store the signed receipt so that it can be sent back, and handle it
// in the routePendingObjects() method.
std::list<RsGRouterTransactionChunkItem*> chunks ;
sliceDataItem(&receipt_item,chunks) ;
bool ok = true ;
for(std::list<RsGRouterTransactionChunkItem*>::const_iterator it(chunks.begin());it!=chunks.end();++it)
{
ok = ok && locked_sendTransactionData(data_item->PeerId(),**it) ;
delete *it ;
}
#ifdef GROUTER_DEBUG
else
std::cerr << " item is not for us. Storing/forwarding." << std::endl;
#endif
if(ok)
// locally store the item, even if we're the destination
RS_STACK_MUTEX(grMtx) ;
// item is not for us. We store it in pending messages and will deal with it later
std::map<GRouterMsgPropagationId,GRouterRoutingInfo>::iterator it = _pending_messages.find(data_item->routing_id) ;
Sha1CheckSum item_hash = computeDataItemHash(data_item) ;
if(it == _pending_messages.end())
{
#ifdef GROUTER_DEBUG
std::cerr << " sent signed receipt to " << data_item->PeerId() << std::endl;
std::cerr << " item is new. Storing it and forwarding it." << std::endl;
#endif
GRouterRoutingInfo& info(_pending_messages[data_item->routing_id]) ;
info.data_item = data_item->duplicate() ;
info.receipt_item = receipt_item ; // inited before, or NULL.
info.tunnel_status = RS_GROUTER_TUNNEL_STATUS_UNMANAGED ;
info.last_tunnel_sent_TS = 0 ;
info.last_friend_sent_TS = 0 ;
info.item_hash = item_hash ;
info.last_tunnel_request_TS = 0 ;
info.sending_attempts = 0 ;
info.received_time_TS = time(NULL) ;
info.tunnel_hash = makeTunnelHash(data_item->destination_key,data_item->service_id) ;
if(item_is_for_us)
{
info.routing_flags = GRouterRoutingInfo::ROUTING_FLAGS_IS_DESTINATION | GRouterRoutingInfo::ROUTING_FLAGS_ALLOW_FRIENDS ;
info.data_status = RS_GROUTER_DATA_STATUS_DONE ;
}
else
std::cerr << " sending signed receipt to " << data_item->PeerId() << ": FAILED." << std::endl;
{
info.routing_flags = GRouterRoutingInfo::ROUTING_FLAGS_ALLOW_FRIENDS ; // don't allow tunnels just yet
info.data_status = RS_GROUTER_DATA_STATUS_PENDING ;
}
}
else if(item_hash != it->second.item_hash) // make sure that the received item is identical to the one that is being saved already.
{
std::cerr << " (EE) reveived an item with known ID but different data hash. Is that an attack?? Item will be dropped." << std::endl;
return ;
}
else
receipt_item = it->second.receipt_item ;
#ifdef GROUTER_DEBUG
std::cerr << " storing incoming route: from " << data_item->PeerId() << std::endl;
#endif
_pending_messages[data_item->routing_id].incoming_routes.ids.insert(data_item->PeerId()) ;
if(receipt_item != NULL)
_pending_messages[data_item->routing_id].data_status = RS_GROUTER_DATA_STATUS_RECEIPT_OK ;
}
bool p3GRouter::locked_getClientAndServiceId(const TurtleFileHash& hash, const RsGxsId& destination_key, GRouterClientService *& client, GRouterServiceId& service_id)
@ -1694,7 +1749,7 @@ bool p3GRouter::signDataItem(RsGRouterAbstractMsgItem *item,const RsGxsId& signi
std::cerr << " Getting key material..." << std::endl;
#endif
uint32_t data_size = item->signed_data_size() ;
uint8_t *data = (uint8_t*)malloc(data_size) ;
RsTemporaryMemory data(data_size) ;
if(data == NULL)
throw std::runtime_error("Cannot allocate memory for signing data.") ;
@ -1710,14 +1765,11 @@ bool p3GRouter::signDataItem(RsGRouterAbstractMsgItem *item,const RsGxsId& signi
#ifdef GROUTER_DEBUG
std::cerr << "Created signature for data hash: " << RsDirUtil::sha1sum(data,data_size) << " and key id=" << signing_id << std::endl;
#endif
free(data) ;
return true ;
}
catch(std::exception& e)
{
std::cerr << " signing failed. Error: " << e.what() << std::endl;
if(data != NULL)
free(data) ;
item->signature.TlvClear() ;
return false ;
@ -1800,9 +1852,10 @@ bool p3GRouter::sendData(const RsGxsId& destination,const GRouterServiceId& clie
if(data_size > MAX_GROUTER_DATA_SIZE)
{
std::cerr << "GRouter max size limit exceeded (size=" << data_size << ", max=" << MAX_GROUTER_DATA_SIZE << "). Please send a smaller object!" << std::endl;
return false ;
}
// Make sure we have a unique id (at least locally).
// Make sure we have a unique id (at least locally). There's little chances that an id of the same value is out there anyway.
//
{
RsStackMutex mtx(grMtx) ;
@ -1862,6 +1915,7 @@ bool p3GRouter::sendData(const RsGxsId& destination,const GRouterServiceId& clie
info.last_tunnel_sent_TS = 0 ;
info.last_friend_sent_TS = 0 ;
info.last_tunnel_request_TS = 0 ;
info.item_hash = computeDataItemHash(data_item) ;
info.sending_attempts = 0 ;
info.routing_flags = GRouterRoutingInfo::ROUTING_FLAGS_IS_ORIGIN | GRouterRoutingInfo::ROUTING_FLAGS_ALLOW_FRIENDS ;// don't allow tunnels just yet
info.received_time_TS = now ;
@ -1881,6 +1935,7 @@ bool p3GRouter::sendData(const RsGxsId& destination,const GRouterServiceId& clie
grouter_debug() << " recv time = " << info.received_time_TS << std::endl;
grouter_debug() << " client id = " << std::hex << data_item->service_id << std::dec << std::endl;
grouter_debug() << " tunnel hash = " << info.tunnel_hash << std::endl;
grouter_debug() << " item hash = " << info.item_hash << std::endl;
grouter_debug() << " routing flags = " << info.routing_flags << std::endl;
#endif
@ -1948,10 +2003,13 @@ bool p3GRouter::loadList(std::list<RsItem*>& items)
if(NULL != (itm1 = dynamic_cast<RsGRouterRoutingInfoItem*>(*it)))
{
// clean data state
itm1->tunnel_status = RS_GROUTER_TUNNEL_STATUS_UNMANAGED ;
_pending_messages[itm1->data_item->routing_id] = *itm1 ;
//_pending_messages[itm1->data_item->routing_id].data_item = itm1->data_item ; // avoids duplication.
itm1->data_item = NULL ; // prevents deletion.
itm1->receipt_item = NULL ; // prevents deletion.
}
delete *it ;
@ -2036,7 +2094,7 @@ bool p3GRouter::getRoutingCacheInfo(std::vector<GRouterRoutingCacheInfo>& infos)
GRouterRoutingCacheInfo cinfo ;
cinfo.mid = it->first ;
cinfo.local_origin.clear() ; // not used before we implement proxys
cinfo.local_origin = it->second.incoming_routes.ids ;
cinfo.destination = it->second.data_item->destination_key ;
cinfo.routing_time = it->second.received_time_TS ;
cinfo.last_tunnel_attempt_time = it->second.last_tunnel_request_TS ;
@ -2045,7 +2103,7 @@ bool p3GRouter::getRoutingCacheInfo(std::vector<GRouterRoutingCacheInfo>& infos)
cinfo.data_status = it->second.data_status ;
cinfo.tunnel_status = it->second.tunnel_status ;
cinfo.data_size = it->second.data_item->data_size ;
cinfo.data_hash = RsDirUtil::sha1sum(it->second.data_item->data_bytes,it->second.data_item->data_size) ;
cinfo.item_hash = it->second.item_hash;
infos.push_back(cinfo) ;
}
@ -2082,6 +2140,7 @@ void p3GRouter::debugDump()
for(std::map<GRouterMsgPropagationId, GRouterRoutingInfo>::iterator it(_pending_messages.begin());it!=_pending_messages.end();++it)
{
grouter_debug() << " Msg id : " << std::hex << it->first << std::dec ;
grouter_debug() << " data hash : " << it->second.item_hash ;
grouter_debug() << " Destination : " << it->second.data_item->destination_key ;
grouter_debug() << " Received : " << now - it->second.received_time_TS << " secs ago.";
grouter_debug() << " Last tunnel sent: " << now - it->second.last_tunnel_sent_TS << " secs ago.";

View File

@ -83,12 +83,14 @@ public:
incoming_data_buffer = NULL ;
}
void clear() { delete incoming_data_buffer ; incoming_data_buffer = NULL ;}
void clear() { if(incoming_data_buffer!=NULL) delete incoming_data_buffer ; incoming_data_buffer = NULL ;}
// These two methods handle the memory management of buffers for each virtual peers.
RsGRouterAbstractMsgItem *addDataChunk(RsGRouterTransactionChunkItem *chunk_item) ;
RsGRouterTransactionChunkItem *incoming_data_buffer ;
time_t last_activity_TS ;
};
class p3GRouter: public RsGRouter, public RsTurtleClientService, public p3Service, public p3Config
@ -220,6 +222,8 @@ private:
void handleLowLevelTransactionChunkItem(RsGRouterTransactionChunkItem *chunk_item);
void handleLowLevelTransactionAckItem(RsGRouterTransactionAcknItem*) ;
static Sha1CheckSum computeDataItemHash(RsGRouterGenericDataItem *data_item);
class nullstream: public std::ostream {};
std::ostream& grouter_debug() const

View File

@ -125,7 +125,7 @@ public:
// virtual bool getPublicKey(const RsGxsId &id, RsTlvSecurityKey &key) = 0;
virtual void getOwnIds(std::list<RsGxsId>& ids) = 0;
virtual bool getOwnIds(std::list<RsGxsId>& ids) = 0;
virtual bool isOwnId(const RsGxsId& key_id) = 0 ;
virtual void timeStampKey(const RsGxsId& key_id) = 0 ;

View File

@ -44,7 +44,7 @@ public:
struct GRouterRoutingCacheInfo
{
GRouterMsgPropagationId mid ;
RsPeerId local_origin;
std::set<RsPeerId> local_origin;
GRouterKeyId destination ;
time_t routing_time;
time_t last_tunnel_attempt_time;
@ -53,7 +53,7 @@ public:
uint32_t data_status ;
uint32_t tunnel_status ;
uint32_t data_size ;
Sha1CheckSum data_hash ;
Sha1CheckSum item_hash ;
};
struct GRouterPublishedKeyInfo

View File

@ -223,7 +223,9 @@ public:
//virtual bool getNickname(const RsGxsId &id, std::string &nickname) = 0;
virtual bool getIdDetails(const RsGxsId &id, RsIdentityDetails &details) = 0;
virtual void getOwnIds(std::list<RsGxsId> &ownIds) = 0;
// Fills up list of all own ids. Returns false if ids are not yet loaded.
virtual bool getOwnIds(std::list<RsGxsId> &ownIds) = 0;
//
virtual bool submitOpinion(uint32_t& token, const RsGxsId &id,

View File

@ -431,10 +431,15 @@ bool p3IdService::isOwnId(const RsGxsId& id)
return std::find(mOwnIds.begin(),mOwnIds.end(),id) != mOwnIds.end() ;
}
void p3IdService::getOwnIds(std::list<RsGxsId> &ownIds)
bool p3IdService::getOwnIds(std::list<RsGxsId> &ownIds)
{
RsStackMutex stack(mIdMtx); /********** STACK LOCKED MTX ******/
if(!mOwnIdsLoaded)
return false ;
ownIds = mOwnIds;
return true ;
}

View File

@ -271,7 +271,7 @@ virtual bool getRecognTagRequest(const RsGxsId &id, const std::string &comment,
/**************** RsGixs Implementation ***************/
virtual void getOwnIds(std::list<RsGxsId> &ownIds);
virtual bool getOwnIds(std::list<RsGxsId> &ownIds);
//virtual bool getPublicKey(const RsGxsId &id, RsTlvSecurityKey &key) ;
//virtual void networkRequestPublicKey(const RsGxsId& key_id,const std::list<RsPeerId>& peer_ids) ;

View File

@ -85,7 +85,8 @@ p3MsgService::p3MsgService(p3ServiceControl *sc, p3IdService *id_serv)
_serialiser = new RsMsgSerialiser();
addSerialType(_serialiser);
mDistantMessagingEnabled = true ;
mShouldEnableDistantMessaging = true ;
mDistantMessagingEnabled = false ;
/* Initialize standard tag types */
if(sc)
@ -453,7 +454,7 @@ bool p3MsgService::saveList(bool& cleanup, std::list<RsItem*>& itemList)
RsConfigKeyValueSet *vitem = new RsConfigKeyValueSet ;
RsTlvKeyValue kv;
kv.key = "DISTANT_MESSAGES_ENABLED" ;
kv.value = mDistantMessagingEnabled?"YES":"NO" ;
kv.value = mShouldEnableDistantMessaging?"YES":"NO" ;
vitem->tlvkvs.pairs.push_back(kv) ;
itemList.push_back(vitem) ;
@ -533,8 +534,6 @@ bool p3MsgService::loadList(std::list<RsItem*>& load)
std::map<uint32_t, RsPeerId> srcIdMsgMap;
std::map<uint32_t, RsPeerId>::iterator srcIt;
bool distant_messaging_set = false ;
// load items and calculate next unique msgId
for(it = load.begin(); it != load.end(); ++it)
{
@ -585,18 +584,10 @@ bool p3MsgService::loadList(std::list<RsItem*>& load)
#ifdef MSG_DEBUG
std::cerr << "Loaded config default nick name for distant chat: " << kit->value << std::endl ;
#endif
enableDistantMessaging(kit->value == "YES") ;
distant_messaging_set = true ;
mShouldEnableDistantMessaging = (kit->value == "YES") ;
}
}
if(mDistantMessagingEnabled || !distant_messaging_set)
{
#ifdef DEBUG_DISTANT_MSG
std::cerr << "No config value for distant messaging. Setting it to true." << std::endl;
#endif
enableDistantMessaging(true) ;
}
// sort items into lists
std::list<RsMsgItem*>::iterator msgIt;
@ -1694,20 +1685,32 @@ void p3MsgService::connectToGlobalRouter(p3GRouter *gr)
void p3MsgService::enableDistantMessaging(bool b)
{
// Normally this method should work on the basis of each GXS id. For now, we just blindly enable
// messaging for all GXS ids for which we have the private key.
// We use a temporary variable because the call to OwnIds() might fail.
bool cchanged ;
{
RsStackMutex stack(mMsgMtx); /********** STACK LOCKED MTX ******/
cchanged = (mDistantMessagingEnabled != b) ;
mDistantMessagingEnabled = b ;
mShouldEnableDistantMessaging = b ;
IndicateConfigChanged() ;
}
std::list<RsGxsId> own_id_list ;
mIdService->getOwnIds(own_id_list) ;
bool p3MsgService::distantMessagingEnabled()
{
RsStackMutex stack(mMsgMtx); /********** STACK LOCKED MTX ******/
return mShouldEnableDistantMessaging ;
}
void p3MsgService::manageDistantPeers()
{
// now possibly flush pending messages
if(mShouldEnableDistantMessaging == mDistantMessagingEnabled)
return ;
#ifdef DEBUG_DISTANT_MSG
std::cerr << "p3MsgService::manageDistantPeers()" << std::endl;
#endif
std::list<RsGxsId> own_id_list ;
if(mIdService->getOwnIds(own_id_list))
{
#ifdef DEBUG_DISTANT_MSG
for(std::list<RsGxsId>::const_iterator it(own_id_list.begin());it!=own_id_list.end();++it)
std::cerr << (b?"Enabling":"Disabling") << " distant messaging, with peer id = " << *it << std::endl;
@ -1715,29 +1718,15 @@ void p3MsgService::enableDistantMessaging(bool b)
for(std::list<RsGxsId>::const_iterator it(own_id_list.begin());it!=own_id_list.end();++it)
{
if(b)
if(mShouldEnableDistantMessaging)
mGRouter->registerKey(*it,GROUTER_CLIENT_ID_MESSAGES,"Messaging contact") ;
else
mGRouter->unregisterKey(*it,GROUTER_CLIENT_ID_MESSAGES) ;
}
if(cchanged)
IndicateConfigChanged() ;
}
bool p3MsgService::distantMessagingEnabled()
{
RsStackMutex stack(mMsgMtx); /********** STACK LOCKED MTX ******/
return mDistantMessagingEnabled ;
mDistantMessagingEnabled = mShouldEnableDistantMessaging ;
}
void p3MsgService::manageDistantPeers()
{
// now possibly flush pending messages
#ifdef DEBUG_DISTANT_MSG
std::cerr << "p3MsgService::manageDistantPeers()" << std::endl;
#endif
enableDistantMessaging(mDistantMessagingEnabled) ;
}
void p3MsgService::notifyDataStatus(const GRouterMsgPropagationId& id,uint32_t data_status)

View File

@ -202,6 +202,7 @@ class p3MsgService: public p3Service, public p3Config, public pqiServiceMonitor,
std::string config_dir;
bool mDistantMessagingEnabled ;
bool mShouldEnableDistantMessaging ;
};
#endif // MESSAGE_SERVICE_HEADER

View File

@ -186,8 +186,8 @@ void GlobalRouterStatisticsWidget::updateContent()
painter.setFont(monospace_f) ;
static const QString data_status_string[4] = { "UNKOWN","PENDING","SENT","RECEIVED" } ;
static const QString tunnel_status_string[3] = { "UNMANAGED", "REQUESTED","ACTIVE" } ;
static const QString data_status_string[6] = { "Unkown","Pending","Sent","Receipt OK","Ongoing","Done" } ;
static const QString tunnel_status_string[4] = { "Unmanaged", "Pending","Ready","Can send" } ;
time_t now = time(NULL) ;
std::map<QString, std::vector<QString> > tos ;
@ -196,10 +196,16 @@ void GlobalRouterStatisticsWidget::updateContent()
{
QString packet_string ;
packet_string += QString("Id=")+QString::number(cache_infos[i].mid,16) ;
//packet_string += tr(" By ")+QString::fromStdString(cache_infos[i].local_origin.toStdString()) ;
packet_string += tr(" By (") ;
for(std::set<RsPeerId>::const_iterator it(cache_infos[i].local_origin.begin());it!=cache_infos[i].local_origin.end();++it)
packet_string += QString::fromStdString((*it).toStdString()) + " - ";
packet_string += ")" ;
packet_string += tr(" Size: ")+QString::number(cache_infos[i].data_size) ;
packet_string += tr(" Data status: ")+data_status_string[cache_infos[i].data_status % 4] ;
packet_string += tr(" Tunnel status: ")+tunnel_status_string[cache_infos[i].tunnel_status % 3] ;
packet_string += tr(" Data status: ")+data_status_string[cache_infos[i].data_status % 6] ;
packet_string += tr(" Data hash: ")+QString::fromStdString(cache_infos[i].item_hash.toStdString()) ;
packet_string += tr(" Tunnel status: ")+tunnel_status_string[cache_infos[i].tunnel_status % 4] ;
packet_string += " " + tr("Received: %1 secs ago, Send: %2 secs ago, Tried: %3 secs ago")
.arg(now - cache_infos[i].routing_time)
.arg(now - cache_infos[i].last_sent_time)