Git-Mirrors/RetroShare
1
0
Fork 0
mirror of https://github.com/RetroShare/RetroShare.git synced 2024-12-24 23:19:29 -05:00
Code Issues Actions Packages Projects Releases Wiki Activity

Implement pull request mechanism in RsGxsNetService

Browse Source 
This could be used to request the online peers to pull updates from us
  ASAP, as an exaple when a group is created a pull request can be
  emitted too so the online peers pull the groups from us ASAP instead
  of waiting for the usual 60 seconds. A mechanism like this is
  especially useful on mobile phones where the internet connection is
  usually turned on only in a few moments (as an example while the user
  is interacting with the app).
Cleanup a few old corners in the code keeping retro-compatibility and
  make the code more welcoming to new developers.
Put a bunch of dead code under #ifdef.
This commit is contained in:
Gioacchino Mazzurco 2021-03-13 20:17:11 +01:00
parent 53dfa67d78
commit e4f25a558d
No known key found for this signature in database
GPG Key ID: A1FBCA3872E87051
8 changed files with 233 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

115
libretroshare/src/gxs/rsgxsnetservice.cc
View File

@ -35,7 +35,7 @@
// |
// +----------- sharePublishKeys()
// |
// +----------- syncWithPeers()
// +----------- pullFromPeers()
// | |
// | +--if AutoSync--- send global UpdateTS of each peer to itself => the peer knows the last
// | | time current peer has received an updated from himself
@ -127,14 +127,14 @@
// (Set at server side to be mGrpServerUpdateItem->grpUpdateTS)
//
// Only updated in processCompletedIncomingTransaction() from Grp list transaction.
// Used in syncWithPeers() sending in RsNxsSyncGrp once to all peers: peer will send data if
// Used in pullFromPeers() sending in RsNxsSyncGrp once to all peers: peer will send data if
// has something new. All time comparisons are in the friends' clock time.
//
// mClientMsgUpdateMap: map< RsPeerId, map<grpId,TimeStamp > >
//
// Last msg list modification time sent by that peer Id
// Updated in processCompletedIncomingTransaction() from Grp list trans.
// Used in syncWithPeers() sending in RsNxsSyncGrp once to all peers.
// Used in pullFromPeers() sending in RsNxsSyncGrp once to all peers.
// Set at server to be mServerMsgUpdateMap[grpId]->msgUpdateTS
//
// mGrpServerUpdateItem: TimeStamp Last group local modification timestamp over all groups
@ -150,7 +150,7 @@
//
// tick() tick()
// | |
// +---- SyncWithPeers +-- recvNxsItemQueue()
// +---- pullFromPeers +-- recvNxsItemQueue()
// | |
// +---------------- Send global UpdateTS of each peer to itself => the peer knows +---------> +------ handleRecvSyncGroup( RsNxsSyncGrp*)
// | the last msg sent (stored in mClientGrpUpdateMap[peer_id]), | | - parse all subscribed groups. For each, send a RsNxsSyncGrpItem with publish TS
@ -457,7 +457,7 @@ int RsGxsNetService::tick()
if((elapsed) < now)
{
syncWithPeers();
pullFromPeers();
syncGrpStatistics();
checkDistantSyncState();
@ -570,39 +570,39 @@ RsGxsGroupId RsGxsNetService::hashGrpId(const RsGxsGroupId& gid,const RsPeerId&
return RsGxsGroupId( RsDirUtil::sha1sum(tmpmem,SIZE).toByteArray() );
}
void RsGxsNetService::syncWithPeers()
void RsGxsNetService::pullFromPeers(std::set<RsPeerId> peers)
{
#ifdef NXS_NET_DEBUG_0
GXSNETDEBUG___ << "RsGxsNetService::syncWithPeers() this=" << (void*)this << ". serviceInfo=" << mServiceInfo << std::endl;
RS_DBG("this=", (void*)this, ". serviceInfo=", mServiceInfo);
#endif
static RsNxsSerialiser ser(mServType) ; // this is used to estimate bandwidth.
RS_STACK_MUTEX(mNxsMutex) ;
std::set<RsPeerId> peers;
mNetMgr->getOnlineList(mServiceInfo.mServiceType, peers);
if(mAllowDistSync && mGxsNetTunnel != NULL)
/* If specific peers are passed as paramether ask only to them */
if(peers.empty())
{
// Grab all online virtual peers of distant tunnels for the current service.
mNetMgr->getOnlineList(mServiceInfo.mServiceType, peers);
std::list<RsGxsNetTunnelVirtualPeerId> vpids ;
mGxsNetTunnel->getVirtualPeers(vpids);
if(mAllowDistSync && mGxsNetTunnel != nullptr)
{
/* Grab all online virtual peers of distant tunnels for the current
* service. */
for(auto it(vpids.begin());it!=vpids.end();++it)
peers.insert(RsPeerId(*it)) ;
std::list<RsGxsNetTunnelVirtualPeerId> vpids ;
mGxsNetTunnel->getVirtualPeers(vpids);
for(auto it(vpids.begin());it!=vpids.end();++it)
peers.insert(RsPeerId(*it)) ;
}
}
if (peers.empty()) {
// nothing to do
return;
}
// Still empty? Then nothing to do
if (peers.empty()) return;
RS_STACK_MUTEX(mNxsMutex);
// for now just grps
for(auto sit = peers.begin(); sit != peers.end(); ++sit)
{
const RsPeerId peerId = *sit;
ClientGrpMap::const_iterator cit = mClientGrpUpdateMap.find(peerId);
@ -624,8 +624,7 @@ void RsGxsNetService::syncWithPeers()
generic_sendItem(grp);
}
if(!mAllowMsgSync)
return ;
if(!mAllowMsgSync) return;
#ifndef GXS_DISABLE_SYNC_MSGS
@ -746,7 +745,7 @@ void RsGxsNetService::syncWithPeers()
#endif
}
void RsGxsNetService::generic_sendItem(RsNxsItem *si)
void RsGxsNetService::generic_sendItem(rs_owner_ptr<RsItem> si)
{
// check if the item is to be sent to a distant peer or not
@ -1718,13 +1717,25 @@ RsItem *RsGxsNetService::generic_recvItem()
void RsGxsNetService::recvNxsItemQueue()
{
RsItem *item ;
RsItem* item;
while(NULL != (item=generic_recvItem()))
{
while(nullptr != (item=generic_recvItem()))
{
#ifdef NXS_NET_DEBUG_1
GXSNETDEBUG_P_(item->PeerId()) << "Received RsGxsNetService Item:" << (void*)item << " type=" << std::hex << item->PacketId() << std::dec << std::endl ;
RS_DBG( "Received RsGxsNetService Item: ", (void*)item, " type=",
item->PacketId() );
#endif
/* Handle pull request and other new items here to not mess with all the
* old nested code and items hell */
switch(static_cast<RsNxsSubtype>(item->PacketSubType()))
{
case RsNxsSubtype::PULL_REQUEST:
std::unique_ptr<RsNxsPullRequestItem> pullItem(
static_cast<RsNxsPullRequestItem*>(item) );
handlePullRequest(std::move(pullItem));
continue;
}
// RsNxsItem needs dynamic_cast, since they have derived siblings.
//
RsNxsItem *ni = dynamic_cast<RsNxsItem*>(item) ;
@ -5075,6 +5086,46 @@ void RsGxsNetService::handleRecvPublishKeys(RsNxsGroupPublishKeyItem *item)
}
}
std::error_condition RsGxsNetService::requestPull(std::set<RsPeerId> peers)
{
/* If specific peers are passed as paramether ask only to them */
if(peers.empty())
{
mNetMgr->getOnlineList(mServiceInfo.mServiceType, peers);
if(mAllowDistSync && mGxsNetTunnel != nullptr)
{
/* Grab all online virtual peers of distant tunnels for the current
* service. */
std::list<RsGxsNetTunnelVirtualPeerId> vpids ;
mGxsNetTunnel->getVirtualPeers(vpids);
for(auto it(vpids.begin());it!=vpids.end();++it)
peers.insert(RsPeerId(*it)) ;
}
}
// Still empty? Reports there are no available peers
if (peers.empty()) return std::errc::network_down;
for(auto& peerId: std::as_const(peers))
{
auto item = new RsNxsPullRequestItem(
static_cast<RsServiceType>(mServType) );
item->PeerId(peerId);
generic_sendItem(item);
}
return std::error_condition();
}
void RsGxsNetService::handlePullRequest(
std::unique_ptr<RsNxsPullRequestItem> item )
{
pullFromPeers(std::set<RsPeerId>{item->PeerId()});
}
bool RsGxsNetService::getGroupServerUpdateTS(const RsGxsGroupId& gid,rstime_t& group_server_update_TS, rstime_t& msg_server_update_TS)
{
RS_STACK_MUTEX(mNxsMutex) ;

22
libretroshare/src/gxs/rsgxsnetservice.h
View File

@ -423,6 +423,8 @@ private:
*/
void handleRecvPublishKeys(RsNxsGroupPublishKeyItem*) ;
void handlePullRequest(std::unique_ptr<RsNxsPullRequestItem> item);
/** E: item handlers **/
@ -459,7 +461,23 @@ private:
void locked_pushMsgRespFromList(std::list<RsNxsItem*>& itemL, const RsPeerId& sslId, const RsGxsGroupId &grp_id, const uint32_t& transN);
void checkDistantSyncState();
void syncWithPeers();
/**
* @brief Pull new stuff from peers
* @param peers peers to pull from, if empty all available peers are pulled
*/
void pullFromPeers(std::set<RsPeerId> peers = std::set<RsPeerId>());
/**
* @brief request online peers to pull updates from our node ASAP
* @param peers peers to which request pull from, if empty all available
* peers are requested to pull
* @return success or error details
* TODO: should this be exposed via RsNetworkExchangeService?
*/
std::error_condition requestPull(
std::set<RsPeerId> peers = std::set<RsPeerId>() );
void syncGrpStatistics();
void addGroupItemToList(NxsTransaction*& tr,
const RsGxsGroupId& grpId, uint32_t& transN,
@ -559,7 +577,7 @@ private:
void cleanRejectedMessages();
void processObserverNotifications();
void generic_sendItem(RsNxsItem *si);
void generic_sendItem(rs_owner_ptr<RsItem> si);
RsItem *generic_recvItem();
private:

3
libretroshare/src/libretroshare.pro
View File

@ -507,7 +507,8 @@ HEADERS += util/folderiterator.h \
util/cxx11retrocompat.h \
util/cxx14retrocompat.h \
util/cxx17retrocompat.h \
util/rsurl.h
util/cxx23retrocompat.h \
util/rsurl.h
SOURCES += ft/ftchunkmap.cc \
ft/ftcontroller.cc \

8
libretroshare/src/rsitems/itempriorities.h
View File

@ -3,7 +3,9 @@
* *
* libretroshare: retroshare core library *
* *
* Copyright 2011-2011 by Cyril Soler <csoler@users.sourceforge.net> *
* Copyright (C) 2011-2018 Cyril Soler <csoler@users.sourceforge.net> *
* Copyright (C) 2021 Gioacchino Mazzurco <gio@eigenlab.org> *
* Copyright (C) 2021 Asociación Civil Altermundi <info@altermundi.net> *
* *
* This program is free software: you can redistribute it and/or modify *
* it under the terms of the GNU Lesser General Public License as *
@ -21,7 +23,9 @@
*******************************************************************************/
#pragma once
#include <stdint.h>
#include <cstdint>
using RsItemPriority = uint8_t;
// This file centralises QoS priorities for all transfer RsItems
//

37
libretroshare/src/rsitems/rsitem.h
View File

@ -3,7 +3,8 @@
* *
* libretroshare: retroshare core library *
* *
* Copyright (C) 2018 Gioacchino Mazzurco <gio@eigenlab.org> *
* Copyright (C) 2018-2021 Gioacchino Mazzurco <gio@eigenlab.org> *
* Copyright (C) 2021 Asociación Civil Altermundi <info@altermundi.net> *
* *
* This program is free software: you can redistribute it and/or modify *
* it under the terms of the GNU Lesser General Public License as *
@ -28,6 +29,9 @@
#include "serialiser/rsserializer.h"
#include "serialiser/rsserializable.h"
#include "util/stacktrace.h"
#include "rsitems/itempriorities.h"
#include "rsitems/rsserviceids.h"
#include <typeinfo>
@ -42,8 +46,13 @@ struct RsItem : RsMemoryManagement::SmallObject, RsSerializable
virtual ~RsItem();
/// TODO: Do this make sense with the new serialization system?
virtual void clear() = 0;
/** TODO: Does the existence of this method make sense with the new
* serialization system? **/
virtual void clear()
{
RS_ERR("Called without being overridden, report to developers");
print_stacktrace();
}
/// @deprecated use << ostream operator instead
RS_DEPRECATED_FOR("<< ostream operator")
@ -70,14 +79,21 @@ struct RsItem : RsMemoryManagement::SmallObject, RsSerializable
uint8_t PacketType();
uint8_t PacketSubType() const;
/** For Service Packets, @deprecated use the costructor with priority
* paramether instead */
RS_DEPRECATED RsItem(uint8_t ver, uint16_t service, uint8_t subtype);
/// For Service Packets
RsItem(uint8_t ver, uint16_t service, uint8_t subtype);
RsItem( uint8_t ver, RsServiceType service, uint8_t subtype,
RsItemPriority prio );
uint16_t PacketService() const; /* combined Packet class/type (mid 16bits) */
void setPacketService(uint16_t service);
inline uint8_t priority_level() const { return _priority_level ;}
inline void setPriorityLevel(uint8_t l) { _priority_level = l ;}
#ifdef RS_DEAD_CODE
/*
* TODO: This default implementation should be removed and childs structs
* implement ::serial_process(...) as soon as all the codebase is ported to
@ -90,11 +106,12 @@ struct RsItem : RsMemoryManagement::SmallObject, RsSerializable
"overriding Class is: ", typeid(*this).name() );
print_stacktrace();
}
#endif //def RS_DEAD_CODE
protected:
uint32_t type;
RsPeerId peerId;
uint8_t _priority_level;
RsItemPriority _priority_level;
};
/// TODO: Do this make sense with the new serialization system?
@ -108,9 +125,17 @@ public:
uint32_t getRawLength() { return len; }
void * getRawData() { return data; }
virtual void clear() {}
// virtual void clear() override {}
virtual std::ostream &print(std::ostream &out, uint16_t indent = 0);
virtual void serial_process(RsGenericSerializer::SerializeJob,
RsGenericSerializer::SerializeContext&) override
{
RS_ERR( "called by an item using new serialization system ",
typeid(*this).name() );
print_stacktrace();
}
private:
void *data;
uint32_t len;

55
libretroshare/src/rsitems/rsnxsitems.h
View File

@ -3,7 +3,10 @@
* *
* libretroshare: retroshare core library *
* *
* Copyright 2012 Christopher Evi-Parker,Robert Fernie<retroshare@lunamutt.com>*
* Copyright (C) 2012 Christopher Evi-Parker *
* Copyright (C) 2012 Robert Fernie <retroshare@lunamutt.com> *
* Copyright (C) 2021 Gioacchino Mazzurco <gio@altermundi.net> *
* Copyright (C) 2021 Asociación Civil Altermundi <info@altermundi.net> *
* *
* This program is free software: you can redistribute it and/or modify *
* it under the terms of the GNU Lesser General Public License as *
@ -19,8 +22,7 @@
* along with this program. If not, see <https://www.gnu.org/licenses/>. *
* *
*******************************************************************************/
#ifndef RSNXSITEMS_H
#define RSNXSITEMS_H
#pragma once
#include <map>
#include <openssl/ssl.h>
@ -35,6 +37,7 @@
// These items have "flag type" numbers, but this is not used.
// TODO: refactor as C++11 enum class
const uint8_t RS_PKT_SUBTYPE_NXS_SYNC_GRP_REQ_ITEM = 0x01;
const uint8_t RS_PKT_SUBTYPE_NXS_SYNC_GRP_ITEM = 0x02;
const uint8_t RS_PKT_SUBTYPE_NXS_SYNC_GRP_STATS_ITEM = 0x03;
@ -47,14 +50,19 @@ const uint8_t RS_PKT_SUBTYPE_NXS_MSG_ITEM = 0x20;
const uint8_t RS_PKT_SUBTYPE_NXS_TRANSAC_ITEM = 0x40;
const uint8_t RS_PKT_SUBTYPE_NXS_GRP_PUBLISH_KEY_ITEM = 0x80;
// possibility create second service to deal with this functionality
enum class RsNxsSubtype : uint8_t
{
PULL_REQUEST = 0x90 /// @see RsNxsPullRequestItem
};
#ifdef RS_DEAD_CODE
// possibility create second service to deal with this functionality
const uint8_t RS_PKT_SUBTYPE_EXT_SEARCH_GRP = 0x0001;
const uint8_t RS_PKT_SUBTYPE_EXT_SEARCH_MSG = 0x0002;
const uint8_t RS_PKT_SUBTYPE_EXT_DELETE_GRP = 0x0004;
const uint8_t RS_PKT_SUBTYPE_EXT_DELETE_MSG = 0x0008;
const uint8_t RS_PKT_SUBTYPE_EXT_SEARCH_REQ = 0x0010;
#endif // def RS_DEAD_CODE
/*!
* Base class for Network exchange service
@ -65,17 +73,14 @@ const uint8_t RS_PKT_SUBTYPE_EXT_SEARCH_REQ = 0x0010;
*/
class RsNxsItem : public RsItem
{
public:
RsNxsItem(uint16_t servtype, uint8_t subtype) : RsItem(RS_PKT_VERSION_SERVICE, servtype, subtype), transactionNumber(0)
{
setPriorityLevel(QOS_PRIORITY_RS_GXS_NET);
return;
}
virtual ~RsNxsItem(){}
virtual void clear() = 0;
RsNxsItem(uint16_t servtype, uint8_t subtype):
RsItem(RS_PKT_VERSION_SERVICE, servtype, subtype), transactionNumber(0)
{ setPriorityLevel(QOS_PRIORITY_RS_GXS_NET); }
uint32_t transactionNumber; // set to zero if this is not a transaction item
virtual ~RsNxsItem() = default;
uint32_t transactionNumber; // set to zero if this is not a transaction item
};
@ -362,6 +367,22 @@ public:
};
/*!
* Used to request to a peer pull updates from us ASAP without waiting GXS sync
* timer */
struct RsNxsPullRequestItem: RsItem
{
explicit RsNxsPullRequestItem(RsServiceType servtype):
RsItem( RS_PKT_VERSION_SERVICE,
servtype,
static_cast<uint8_t>(RsNxsSubtype::PULL_REQUEST),
QOS_PRIORITY_RS_GXS_NET ) {}
/// @see RsSerializable
void serial_process( RsGenericSerializer::SerializeJob,
RsGenericSerializer::SerializeContext& ) override {}
};
/*!
* Used to respond to a RsGrpMsgsReq
@ -401,6 +422,7 @@ struct RsNxsMsg : RsNxsItem
RsGxsMsgMetaData* metaData;
};
#ifdef RS_DEAD_CODE
/*!
* Used to request a search of user data
*/
@ -422,7 +444,7 @@ public:
RsTlvBinaryData serviceSearchItem; // service aware of item class
uint32_t expiration; // expiration date
};
#endif //def RS_DEAD_CODE
#ifdef UNUSED_CODE
@ -511,6 +533,3 @@ public:
protected:
const uint16_t SERVICE_TYPE;
};
#endif // RSNXSITEMS_H

33
libretroshare/src/serialiser/rsserial.cc
View File

@ -3,7 +3,9 @@
* *
* libretroshare: retroshare core library *
* *
* Copyright 2007-2008 by Robert Fernie <retroshare@lunamutt.com> *
* Copyright (C) 2007-2008 Robert Fernie <retroshare@lunamutt.com> *
* Copyright (C) 2021 Gioacchino Mazzurco <gio@eigenlab.org> *
* Copyright (C) 2021 Asociación Civil Altermundi <info@altermundi.net> *
* *
* This program is free software: you can redistribute it and/or modify *
* it under the terms of the GNU Lesser General Public License as *
@ -20,21 +22,20 @@
* *
*******************************************************************************/
#include "serialiser/rsbaseserial.h"
#include "util/rsthreads.h"
#include "util/rsstring.h"
#include "util/rsprint.h"
#include "rsitems/rsitem.h"
#include "rsitems/itempriorities.h"
#include <math.h>
#include <cmath>
#include <map>
#include <vector>
#include <iostream>
#include <typeinfo>
#include "serialiser/rsbaseserial.h"
#include "util/cxx23retrocompat.h"
#include "util/rsthreads.h"
#include "util/rsstring.h"
#include "util/rsprint.h"
#include "rsitems/rsitem.h"
#include "rsitems/itempriorities.h"
/***
* #define RSSERIAL_DEBUG 1
@ -166,11 +167,17 @@ uint8_t RsItem::PacketSubType() const
/* For Service Packets */
RsItem::RsItem(uint8_t ver, uint16_t service, uint8_t subtype)
{
_priority_level = QOS_PRIORITY_UNKNOWN ; // This value triggers PQIInterface to complain about undefined priorities
// This value triggers PQIInterface to complain about undefined priorities
_priority_level = QOS_PRIORITY_UNKNOWN;
type = (ver << 24) + (service << 8) + subtype;
return;
}
RsItem::RsItem( uint8_t ver, RsServiceType service, uint8_t subtype,
RsItemPriority prio ):
type(static_cast<uint32_t>(
(ver << 24) + (std::to_underlying(service) << 8) + subtype )),
_priority_level(prio) {}
uint16_t RsItem::PacketService() const
{
return (type >> 8) & 0xFFFF;

34
libretroshare/src/util/cxx23retrocompat.h Normal file
View File

@ -0,0 +1,34 @@
/*******************************************************************************
* RetroShare C++23 backwards compatibility utilities *
* *
* libretroshare: retroshare core library *
* *
* Copyright (C) 2021 Gioacchino Mazzurco <gio@eigenlab.org> *
* Copyright (C) 2021 Asociación Civil Altermundi <info@altermundi.net> *
* *
* This program is free software: you can redistribute it and/or modify *
* it under the terms of the GNU Lesser General Public License as *
* published by the Free Software Foundation, either version 3 of the *
* License, or (at your option) any later version. *
* *
* This program is distributed in the hope that it will be useful, *
* but WITHOUT ANY WARRANTY; without even the implied warranty of *
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the *
* GNU Lesser General Public License for more details. *
* *
* You should have received a copy of the GNU Lesser General Public License *
* along with this program. If not, see <https://www.gnu.org/licenses/>. *
* *
*******************************************************************************/
#pragma once
#include <utility>
#if ! defined(__cpp_lib_to_underlying)
namespace std
{
template <class Enum>
constexpr underlying_type_t<Enum> to_underlying(Enum e) noexcept
{ return static_cast<std::underlying_type_t<Enum>>(e); }
}
#endif // ! defined(__cpp_lib_to_underlying)