From e4f25a558d48629155486d3e8517740328823f8f Mon Sep 17 00:00:00 2001 From: Gioacchino Mazzurco Date: Sat, 13 Mar 2021 20:17:11 +0100 Subject: [PATCH] Implement pull request mechanism in RsGxsNetService 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. --- libretroshare/src/gxs/rsgxsnetservice.cc | 115 +++++++++++++++------ libretroshare/src/gxs/rsgxsnetservice.h | 22 +++- libretroshare/src/libretroshare.pro | 3 +- libretroshare/src/rsitems/itempriorities.h | 8 +- libretroshare/src/rsitems/rsitem.h | 37 +++++-- libretroshare/src/rsitems/rsnxsitems.h | 55 ++++++---- libretroshare/src/serialiser/rsserial.cc | 33 +++--- libretroshare/src/util/cxx23retrocompat.h | 34 ++++++ 8 files changed, 233 insertions(+), 74 deletions(-) create mode 100644 libretroshare/src/util/cxx23retrocompat.h diff --git a/libretroshare/src/gxs/rsgxsnetservice.cc b/libretroshare/src/gxs/rsgxsnetservice.cc index e664b580c..adea14205 100644 --- a/libretroshare/src/gxs/rsgxsnetservice.cc +++ b/libretroshare/src/gxs/rsgxsnetservice.cc @@ -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 > // // 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 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 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 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 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 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(item->PacketSubType())) + { + case RsNxsSubtype::PULL_REQUEST: + std::unique_ptr pullItem( + static_cast(item) ); + handlePullRequest(std::move(pullItem)); + continue; + } + // RsNxsItem needs dynamic_cast, since they have derived siblings. // RsNxsItem *ni = dynamic_cast(item) ; @@ -5075,6 +5086,46 @@ void RsGxsNetService::handleRecvPublishKeys(RsNxsGroupPublishKeyItem *item) } } +std::error_condition RsGxsNetService::requestPull(std::set 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 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(mServType) ); + item->PeerId(peerId); + generic_sendItem(item); + } + + return std::error_condition(); +} + +void RsGxsNetService::handlePullRequest( + std::unique_ptr item ) +{ + pullFromPeers(std::set{item->PeerId()}); +} + bool RsGxsNetService::getGroupServerUpdateTS(const RsGxsGroupId& gid,rstime_t& group_server_update_TS, rstime_t& msg_server_update_TS) { RS_STACK_MUTEX(mNxsMutex) ; diff --git a/libretroshare/src/gxs/rsgxsnetservice.h b/libretroshare/src/gxs/rsgxsnetservice.h index a4b448003..406531852 100644 --- a/libretroshare/src/gxs/rsgxsnetservice.h +++ b/libretroshare/src/gxs/rsgxsnetservice.h @@ -423,6 +423,8 @@ private: */ void handleRecvPublishKeys(RsNxsGroupPublishKeyItem*) ; + void handlePullRequest(std::unique_ptr item); + /** E: item handlers **/ @@ -459,7 +461,23 @@ private: void locked_pushMsgRespFromList(std::list& 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 peers = std::set()); + + /** + * @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 peers = std::set() ); + 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 si); RsItem *generic_recvItem(); private: diff --git a/libretroshare/src/libretroshare.pro b/libretroshare/src/libretroshare.pro index d42f68328..9358d8147 100644 --- a/libretroshare/src/libretroshare.pro +++ b/libretroshare/src/libretroshare.pro @@ -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 \ diff --git a/libretroshare/src/rsitems/itempriorities.h b/libretroshare/src/rsitems/itempriorities.h index a376a9d64..7c2b92cb6 100644 --- a/libretroshare/src/rsitems/itempriorities.h +++ b/libretroshare/src/rsitems/itempriorities.h @@ -3,7 +3,9 @@ * * * libretroshare: retroshare core library * * * - * Copyright 2011-2011 by Cyril Soler * + * Copyright (C) 2011-2018 Cyril Soler * + * Copyright (C) 2021 Gioacchino Mazzurco * + * Copyright (C) 2021 Asociación Civil Altermundi * * * * 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 +#include + +using RsItemPriority = uint8_t; // This file centralises QoS priorities for all transfer RsItems // diff --git a/libretroshare/src/rsitems/rsitem.h b/libretroshare/src/rsitems/rsitem.h index 3ef44f859..70844f840 100644 --- a/libretroshare/src/rsitems/rsitem.h +++ b/libretroshare/src/rsitems/rsitem.h @@ -3,7 +3,8 @@ * * * libretroshare: retroshare core library * * * - * Copyright (C) 2018 Gioacchino Mazzurco * + * Copyright (C) 2018-2021 Gioacchino Mazzurco * + * Copyright (C) 2021 Asociación Civil Altermundi * * * * 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 @@ -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; diff --git a/libretroshare/src/rsitems/rsnxsitems.h b/libretroshare/src/rsitems/rsnxsitems.h index 2b6731bfc..8be63f74c 100644 --- a/libretroshare/src/rsitems/rsnxsitems.h +++ b/libretroshare/src/rsitems/rsnxsitems.h @@ -3,7 +3,10 @@ * * * libretroshare: retroshare core library * * * - * Copyright 2012 Christopher Evi-Parker,Robert Fernie* + * Copyright (C) 2012 Christopher Evi-Parker * + * Copyright (C) 2012 Robert Fernie * + * Copyright (C) 2021 Gioacchino Mazzurco * + * Copyright (C) 2021 Asociación Civil Altermundi * * * * 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 . * * * *******************************************************************************/ -#ifndef RSNXSITEMS_H -#define RSNXSITEMS_H +#pragma once #include #include @@ -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(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 diff --git a/libretroshare/src/serialiser/rsserial.cc b/libretroshare/src/serialiser/rsserial.cc index 52123abf1..4d3cda811 100644 --- a/libretroshare/src/serialiser/rsserial.cc +++ b/libretroshare/src/serialiser/rsserial.cc @@ -3,7 +3,9 @@ * * * libretroshare: retroshare core library * * * - * Copyright 2007-2008 by Robert Fernie * + * Copyright (C) 2007-2008 Robert Fernie * + * Copyright (C) 2021 Gioacchino Mazzurco * + * Copyright (C) 2021 Asociación Civil Altermundi * * * * 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 +#include #include #include #include #include +#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( + (ver << 24) + (std::to_underlying(service) << 8) + subtype )), + _priority_level(prio) {} + uint16_t RsItem::PacketService() const { return (type >> 8) & 0xFFFF; diff --git a/libretroshare/src/util/cxx23retrocompat.h b/libretroshare/src/util/cxx23retrocompat.h new file mode 100644 index 000000000..cd96b7cbf --- /dev/null +++ b/libretroshare/src/util/cxx23retrocompat.h @@ -0,0 +1,34 @@ +/******************************************************************************* + * RetroShare C++23 backwards compatibility utilities * + * * + * libretroshare: retroshare core library * + * * + * Copyright (C) 2021 Gioacchino Mazzurco * + * Copyright (C) 2021 Asociación Civil Altermundi * + * * + * 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 . * + * * + *******************************************************************************/ +#pragma once + +#include + +#if ! defined(__cpp_lib_to_underlying) +namespace std +{ +template +constexpr underlying_type_t to_underlying(Enum e) noexcept +{ return static_cast>(e); } +} +#endif // ! defined(__cpp_lib_to_underlying)