diff --git a/Blockchain Cache Global.go b/Blockchain Cache Global.go new file mode 100644 index 0000000..b2f8492 --- /dev/null +++ b/Blockchain Cache Global.go @@ -0,0 +1,139 @@ +/* +File Name: Blockchain Cache Global.go +Copyright: 2021 Peernet s.r.o. +Author: Peter Kleissner +*/ + +package core + +import ( + "github.com/PeernetOfficial/core/blockchain" + "github.com/PeernetOfficial/core/protocol" + "github.com/enfipy/locker" +) + +// The blockchain cache stores blockchains. +type BlockchainCache struct { + BlockchainDirectory string // The directory for storing blockchains in a key-value store. + MaxBlockSize uint64 // Max block size to accept. + MaxBlockCount uint64 // Max block count to cache per peer. + LimitTotalRecords uint64 // Max count of blocks and header in total to keep across all blockchains. 0 = unlimited. Max Records * Max Block Size = Size Limit. + ReadOnly bool // Whether the cache is read only. + + store *blockchain.MultiStore + peerLock *locker.Locker +} + +func initBlockchainCache(BlockchainDirectory string, MaxBlockSize, MaxBlockCount, LimitTotalRecords uint64) (cache *BlockchainCache) { + if BlockchainDirectory == "" { + return nil + } + + var err error + + cache = &BlockchainCache{ + BlockchainDirectory: BlockchainDirectory, + MaxBlockSize: MaxBlockSize, + MaxBlockCount: MaxBlockCount, + LimitTotalRecords: LimitTotalRecords, + } + + cache.store, err = blockchain.InitMultiStore(BlockchainDirectory) + if err != nil { + Filters.LogError("initBlockchainCache", "initializing database '%s': %s", BlockchainDirectory, err.Error()) + return nil + } + + cache.peerLock = locker.Initialize() + + if LimitTotalRecords > 0 && cache.store.Database.Count() >= LimitTotalRecords { + cache.ReadOnly = true + } + + return cache +} + +// SeenBlockchainVersion shall be called with information about another peer's blockchain. +// If the reported version number is newer, all existing blocks are immediately deleted. +func (cache *BlockchainCache) SeenBlockchainVersion(peer *PeerInfo) { + cache.peerLock.Lock(string(peer.PublicKey.SerializeCompressed())) + defer cache.peerLock.Unlock(string(peer.PublicKey.SerializeCompressed())) + + // intermediate function to download and process blocks + downloadAndProcessBlocks := func(peer *PeerInfo, header *blockchain.MultiBlockchainHeader, offset, limit uint64) { + oldCount := len(header.ListBlocks) + + if limit > cache.MaxBlockCount { + limit = cache.MaxBlockCount + } + + peer.BlockDownload(peer.PublicKey, cache.MaxBlockCount, cache.MaxBlockSize, []protocol.BlockRange{{Offset: offset, Limit: limit}}, func(data []byte, targetBlock protocol.BlockRange, blockSize uint64, availability uint8) { + if availability != protocol.GetBlockStatusAvailable { + return + } + cache.store.WriteBlock(peer.PublicKey, peer.BlockchainVersion, targetBlock.Offset, data) + header.ListBlocks = append(header.ListBlocks, targetBlock.Offset) + }) + + // only update the blockchain header if it changed + if oldCount != len(header.ListBlocks) { + cache.store.WriteBlockchainHeader(peer.PublicKey, header) + } + } + + // get the old header + header, status, err := cache.store.AssessBlockchainHeader(peer.PublicKey, peer.BlockchainVersion, uint64(peer.BlockchainHeight)) + if err != nil { + return + } + + switch status { + case blockchain.MultiStatusEqual: + return + + case blockchain.MultiStatusInvalidRemote: + cache.store.DeleteBlockchain(peer.PublicKey, header) + + case blockchain.MultiStatusHeaderNA: + if header, err = cache.store.NewBlockchainHeader(peer.PublicKey, peer.BlockchainVersion, uint64(peer.BlockchainHeight)); err != nil { + return + } + + downloadAndProcessBlocks(peer, header, 0, uint64(peer.BlockchainHeight)) + + case blockchain.MultiStatusNewVersion: + // delete existing data first, then create it new + cache.store.DeleteBlockchain(peer.PublicKey, header) + + if header, err = cache.store.NewBlockchainHeader(peer.PublicKey, peer.BlockchainVersion, uint64(peer.BlockchainHeight)); err != nil { + return + } + + downloadAndProcessBlocks(peer, header, 0, uint64(peer.BlockchainHeight)) + + case blockchain.MultiStatusNewBlocks: + offset := header.Height + limit := uint64(peer.BlockchainHeight) - header.Height + + header.Height = uint64(peer.BlockchainHeight) + + downloadAndProcessBlocks(peer, header, offset, limit) + + } + + if cache.LimitTotalRecords > 0 { + cache.ReadOnly = cache.store.Database.Count() >= cache.LimitTotalRecords + } +} + +// remoteBlockchainUpdate shall be called to indicate a potential update of the remotes blockchain. +// It will use the blockchain version and height to update the data lake as appropriate. +// This function is called in the Go routine of the packet worker and therefore must not stall. +func (peer *PeerInfo) remoteBlockchainUpdate() { + if currentBackend.GlobalBlockchainCache == nil || currentBackend.GlobalBlockchainCache.ReadOnly || peer.BlockchainVersion == 0 && peer.BlockchainHeight == 0 { + return + } + + // TODO: This entire function should be instead a non-blocking message via a buffer channel. + go currentBackend.GlobalBlockchainCache.SeenBlockchainVersion(peer) +} diff --git a/Config Default.yaml b/Config Default.yaml index 15f0283..9d151d1 100644 --- a/Config Default.yaml +++ b/Config Default.yaml @@ -1,7 +1,7 @@ # Locations of important files and folders LogFile: "data/log.txt" # Log file. It contains informational and error messages. BlockchainMain: "data/blockchain main/" # Blockchain main stores the end-users blockchain data. It contains meta data of shared files, profile data, and social interactions. -BlockchainGlobal: "data/blockchain global/" # Blockchain global stores blockchain data from global users. +BlockchainGlobal: "data/blockchain global/" # Blockchain global caches blockchain data from global users. Empty to disable. WarehouseMain: "data/warehouse main/" # Warehouse main stores the actual data of files shared by the end-user. # Listen defines all IP:Port combinations to listen on. If empty, it will listen on all IPs automatically on available ports. @@ -35,3 +35,8 @@ LocalFirewall: false # Indicates that a local firewall may drop unsolicited i # PortForward specifies an external port that was manually forwarded by the user. All listening IPs must have that same port number forwarded! # If this setting is invalid, it will prohibit other peers from connecting. If set, it automatically disables UPnP. PortForward: 0 # Default not set. + +# Global blockchain cache limits +CacheMaxBlockSize: 4096 # Max block size to accept in bytes. +CacheMaxBlockCount: 256 # Max block count to cache per peer. +LimitTotalRecords: 0 # Record count limit. 0 = unlimited. Max Records * Max Block Size = Size Limit. diff --git a/Config.go b/Config.go index 02fb0c7..c4bc8fd 100644 --- a/Config.go +++ b/Config.go @@ -23,7 +23,7 @@ var config struct { // Locations of important files and folders LogFile string `yaml:"LogFile"` // Log file. It contains informational and error messages. BlockchainMain string `yaml:"BlockchainMain"` // Blockchain main stores the end-users blockchain data. It contains meta data of shared files, profile data, and social interactions. - BlockchainGlobal string `yaml:"BlockchainGlobal"` // Blockchain global stores blockchain data from global users. + BlockchainGlobal string `yaml:"BlockchainGlobal"` // Blockchain global caches blockchain data from global users. Empty to disable. WarehouseMain string `yaml:"WarehouseMain"` // Warehouse main stores the actual data of files shared by the end-user. // Listen settings @@ -45,6 +45,11 @@ var config struct { // PortForward specifies an external port that was manually forwarded by the user. All listening IPs must have that same port number forwarded! // If this setting is invalid, it will prohibit other peers from connecting. If set, it automatically disables UPnP. PortForward uint16 `yaml:"PortForward"` + + // Global blockchain cache limits + CacheMaxBlockSize uint64 `yaml:"CacheMaxBlockSize"` // Max block size to accept in bytes. + CacheMaxBlockCount uint64 `yaml:"CacheMaxBlockCount"` // Max block count to cache per peer. + LimitTotalRecords uint64 `yaml:"LimitTotalRecords"` // Record count limit. 0 = unlimited. Max Records * Max Block Size = Size Limit. } // peerSeed is a singl peer entry from the config's seed list diff --git a/Network.go b/Network.go index acf2a63..5f77d86 100644 --- a/Network.go +++ b/Network.go @@ -176,12 +176,18 @@ func (nets *Networks) packetWorker() { peer.UserAgent = announce.UserAgent } peer.Features = announce.Features + + isBlockchainUpdate := peer.BlockchainHeight != announce.BlockchainHeight || peer.BlockchainVersion != announce.BlockchainVersion peer.BlockchainHeight = announce.BlockchainHeight peer.BlockchainVersion = announce.BlockchainVersion Filters.MessageIn(peer, raw, announce) peer.cmdAnouncement(announce, connection) + + if isBlockchainUpdate { + peer.remoteBlockchainUpdate() + } } case protocol.CommandResponse: // Response @@ -205,12 +211,18 @@ func (nets *Networks) packetWorker() { peer.UserAgent = response.UserAgent } peer.Features = response.Features + + isBlockchainUpdate := peer.BlockchainHeight != response.BlockchainHeight || peer.BlockchainVersion != response.BlockchainVersion peer.BlockchainHeight = response.BlockchainHeight peer.BlockchainVersion = response.BlockchainVersion Filters.MessageIn(peer, raw, response) peer.cmdResponse(response, connection) + + if isBlockchainUpdate { + peer.remoteBlockchainUpdate() + } } case protocol.CommandLocalDiscovery: // Local discovery, sent via IPv4 broadcast and IPv6 multicast @@ -219,12 +231,18 @@ func (nets *Networks) packetWorker() { peer.UserAgent = announce.UserAgent } peer.Features = announce.Features + + isBlockchainUpdate := peer.BlockchainHeight != announce.BlockchainHeight || peer.BlockchainVersion != announce.BlockchainVersion peer.BlockchainHeight = announce.BlockchainHeight peer.BlockchainVersion = announce.BlockchainVersion Filters.MessageIn(peer, raw, announce) peer.cmdLocalDiscovery(announce, connection) + + if isBlockchainUpdate { + peer.remoteBlockchainUpdate() + } } case protocol.CommandPing: // Ping diff --git a/Peernet.go b/Peernet.go index 31d38be..00647fe 100644 --- a/Peernet.go +++ b/Peernet.go @@ -10,7 +10,7 @@ var userAgent = "Peernet Core/0.1" // must be overwritten by the caller // Init initializes the client. The config must be loaded first! // The User Agent must be provided in the form "Application Name/1.0". -func Init(UserAgent string) { +func Init(UserAgent string) (backend *Backend) { if userAgent = UserAgent; userAgent == "" { return } @@ -26,6 +26,13 @@ func Init(UserAgent string) { initBroadcastIPv4() initStore() initNetwork() + + backend = &Backend{} + backend.GlobalBlockchainCache = initBlockchainCache(config.BlockchainGlobal, config.CacheMaxBlockSize, config.CacheMaxBlockCount, config.LimitTotalRecords) + + currentBackend = backend + + return backend } // Connect starts bootstrapping and local peer discovery. @@ -38,3 +45,12 @@ func Connect() { go networks.startUPnP() go autoBucketRefresh() } + +// The Backend represents an instance of a Peernet client to be used by a frontend. +// Global variables and init functions are to be merged. +type Backend struct { + GlobalBlockchainCache *BlockchainCache // stores other peers blockchains +} + +// This variable is to be replaced later by pointers in structures. +var currentBackend *Backend diff --git a/go.mod b/go.mod index 6818578..59f1764 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,7 @@ go 1.16 require ( github.com/akrylysov/pogreb v0.10.1 + github.com/enfipy/locker v1.1.0 github.com/google/uuid v1.3.0 github.com/gorilla/mux v1.8.0 github.com/gorilla/websocket v1.4.2 diff --git a/go.sum b/go.sum index 6c088bb..cd5c0b6 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,7 @@ github.com/akrylysov/pogreb v0.10.1 h1:FqlR8VR7uCbJdfUob916tPM+idpKgeESDXOA1K0DK4w= github.com/akrylysov/pogreb v0.10.1/go.mod h1:pNs6QmpQ1UlTJKDezuRWmaqkgUE2TuU0YTWyqJZ7+lI= +github.com/enfipy/locker v1.1.0 h1:2zVJ0ky7cS1Vjs0x6OQWFiT2dSEiHrI5/O2KCz1fgGc= +github.com/enfipy/locker v1.1.0/go.mod h1:uuj+dvWHECshK8rkHcw+ZOb9SLo16yc0Em/JGUqRqko= github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I= github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI=