Skip to content

Commit

Permalink
Remove currentSnapshot (#2273)
Browse files Browse the repository at this point in the history
  • Loading branch information
@erikzhang
erikzhang authored Jan 28, 2021
1 parent 7938b93 commit 234350d
Show file tree
Hide file tree
Showing 11 changed files with 139 additions and 176 deletions.
71 changes: 36 additions & 35 deletions src/neo/Ledger/Blockchain.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -65,14 +65,15 @@ public class RelayResult { public IInventory Inventory; public VerifyResult Resu
private readonly ConcurrentDictionary<UInt256, Block> block_cache = new ConcurrentDictionary<UInt256, Block>();
private readonly Dictionary<uint, UnverifiedBlocksList> block_cache_unverified = new Dictionary<uint, UnverifiedBlocksList>();
internal readonly RelayCache RelayCache = new RelayCache(100);
private SnapshotCache currentSnapshot;
private ImmutableHashSet<UInt160> extensibleWitnessWhiteList;

public IStore Store { get; }
/// <summary>
/// A readonly view of the blockchain store.
/// Note: It doesn't need to be disposed because the <see cref="ISnapshot"/> inside it is null.
/// </summary>
public DataCache View => new SnapshotCache(Store);
public MemoryPool MemPool { get; }
public uint Height => NativeContract.Ledger.CurrentIndex(currentSnapshot);
public UInt256 CurrentBlockHash => NativeContract.Ledger.CurrentHash(currentSnapshot);

private static Blockchain singleton;
public static Blockchain Singleton
Expand Down Expand Up @@ -110,14 +111,14 @@ public Blockchain(NeoSystem system, IStore store)
{
if (singleton != null)
throw new InvalidOperationException();
if (!NativeContract.Ledger.Initialized(View))
DataCache snapshot = View;
if (!NativeContract.Ledger.Initialized(snapshot))
{
Persist(GenesisBlock);
}
else
{
UpdateCurrentSnapshot();
MemPool.LoadPolicy(currentSnapshot);
UpdateExtensibleWitnessWhiteList(snapshot);
}
singleton = this;
}
Expand All @@ -136,14 +137,16 @@ public SnapshotCache GetSnapshot()

private void OnImport(IEnumerable<Block> blocks, bool verify)
{
uint currentHeight = NativeContract.Ledger.CurrentIndex(View);
foreach (Block block in blocks)
{
if (block.Index <= Height) continue;
if (block.Index != Height + 1)
if (block.Index <= currentHeight) continue;
if (block.Index != currentHeight + 1)
throw new InvalidOperationException();
if (verify && !block.Verify(currentSnapshot))
if (verify && !block.Verify(View))
throw new InvalidOperationException();
Persist(block);
++currentHeight;
}
Sender.Tell(new ImportCompleted());
}
Expand Down Expand Up @@ -182,15 +185,17 @@ private void OnFillMemoryPool(IEnumerable<Transaction> transactions)
// Invalidate all the transactions in the memory pool, to avoid any failures when adding new transactions.
MemPool.InvalidateAllTransactions();

DataCache snapshot = View;

// Add the transactions to the memory pool
foreach (var tx in transactions)
{
if (NativeContract.Ledger.ContainsTransaction(View, tx.Hash))
if (NativeContract.Ledger.ContainsTransaction(snapshot, tx.Hash))
continue;
// First remove the tx if it is unverified in the pool.
MemPool.TryRemoveUnVerified(tx.Hash, out _);
// Add to the memory pool
MemPool.TryAdd(tx, currentSnapshot);
MemPool.TryAdd(tx, snapshot);
}
// Transactions originally in the pool will automatically be reverified based on their priority.

Expand All @@ -214,43 +219,45 @@ private void OnInventory(IInventory inventory, bool relay = true)

private VerifyResult OnNewBlock(Block block)
{
if (block.Index <= Height)
DataCache snapshot = View;
uint currentHeight = NativeContract.Ledger.CurrentIndex(snapshot);
if (block.Index <= currentHeight)
return VerifyResult.AlreadyExists;
if (block.Index - 1 > Height)
if (block.Index - 1 > currentHeight)
{
AddUnverifiedBlockToCache(block);
return VerifyResult.UnableToVerify;
}
if (block.Index == Height + 1)
if (block.Index == currentHeight + 1)
{
if (!block.Verify(currentSnapshot))
if (!block.Verify(snapshot))
return VerifyResult.Invalid;
block_cache.TryAdd(block.Hash, block);
block_cache_unverified.Remove(block.Index);
Persist(block);
if (block_cache_unverified.TryGetValue(Height + 1, out var unverifiedBlocks))
if (block_cache_unverified.TryGetValue(block.Index + 1, out var unverifiedBlocks))
{
foreach (var unverifiedBlock in unverifiedBlocks.Blocks)
Self.Tell(unverifiedBlock, ActorRefs.NoSender);
block_cache_unverified.Remove(Height + 1);
block_cache_unverified.Remove(block.Index + 1);
}
// We can store the new block in block_cache and tell the new height to other nodes after Persist().
system.LocalNode.Tell(Message.Create(MessageCommand.Ping, PingPayload.Create(Singleton.Height)));
system.LocalNode.Tell(Message.Create(MessageCommand.Ping, PingPayload.Create(block.Index)));
}
return VerifyResult.Succeed;
}

private VerifyResult OnNewInventory(IInventory inventory)
{
if (!inventory.Verify(currentSnapshot)) return VerifyResult.Invalid;
if (!inventory.Verify(View)) return VerifyResult.Invalid;
RelayCache.Add(inventory);
return VerifyResult.Succeed;
}

private VerifyResult OnNewTransaction(Transaction transaction)
{
if (ContainsTransaction(transaction.Hash)) return VerifyResult.AlreadyExists;
return MemPool.TryAdd(transaction, currentSnapshot);
return MemPool.TryAdd(transaction, View);
}

private void OnPreverifyCompleted(PreverifyCompleted task)
Expand Down Expand Up @@ -284,7 +291,7 @@ protected override void OnReceive(object message)
OnPreverifyCompleted(task);
break;
case Idle _:
if (MemPool.ReVerifyTopUnverifiedTransactionsIfNeeded(MaxTxToReverifyPerIdle, currentSnapshot))
if (MemPool.ReVerifyTopUnverifiedTransactionsIfNeeded(MaxTxToReverifyPerIdle, View))
Self.Tell(Idle.Instance, ActorRefs.NoSender);
break;
}
Expand Down Expand Up @@ -361,19 +368,13 @@ private void Persist(Block block)
}
}
if (commitExceptions != null) throw new AggregateException(commitExceptions);
UpdateExtensibleWitnessWhiteList(snapshot);
MemPool.UpdatePoolForBlockPersisted(block, snapshot);
}
UpdateCurrentSnapshot();
block_cache.TryRemove(block.PrevHash, out _);
MemPool.UpdatePoolForBlockPersisted(block, currentSnapshot);
Context.System.EventStream.Publish(new PersistCompleted { Block = block });
}

protected override void PostStop()
{
base.PostStop();
currentSnapshot?.Dispose();
}

public static Props Props(NeoSystem system, IStore store)
{
return Akka.Actor.Props.Create(() => new Blockchain(system, store)).WithMailbox("blockchain-mailbox");
Expand All @@ -390,21 +391,21 @@ private void SendRelayResult(IInventory inventory, VerifyResult result)
Context.System.EventStream.Publish(rr);
}

private void UpdateCurrentSnapshot()
private void UpdateExtensibleWitnessWhiteList(DataCache snapshot)
{
Interlocked.Exchange(ref currentSnapshot, GetSnapshot())?.Dispose();
uint currentHeight = NativeContract.Ledger.CurrentIndex(snapshot);
var builder = ImmutableHashSet.CreateBuilder<UInt160>();
builder.Add(NativeContract.NEO.GetCommitteeAddress(currentSnapshot));
var validators = NativeContract.NEO.GetNextBlockValidators(currentSnapshot);
builder.Add(NativeContract.NEO.GetCommitteeAddress(snapshot));
var validators = NativeContract.NEO.GetNextBlockValidators(snapshot);
builder.Add(Contract.GetBFTAddress(validators));
builder.UnionWith(validators.Select(u => Contract.CreateSignatureRedeemScript(u).ToScriptHash()));
var oracles = NativeContract.RoleManagement.GetDesignatedByRole(currentSnapshot, Role.Oracle, Height);
var oracles = NativeContract.RoleManagement.GetDesignatedByRole(snapshot, Role.Oracle, currentHeight);
if (oracles.Length > 0)
{
builder.Add(Contract.GetBFTAddress(oracles));
builder.UnionWith(oracles.Select(u => Contract.CreateSignatureRedeemScript(u).ToScriptHash()));
}
var stateValidators = NativeContract.RoleManagement.GetDesignatedByRole(currentSnapshot, Role.StateValidator, Height);
var stateValidators = NativeContract.RoleManagement.GetDesignatedByRole(snapshot, Role.StateValidator, currentHeight);
if (stateValidators.Length > 0)
{
builder.Add(Contract.GetBFTAddress(stateValidators));
Expand Down
14 changes: 3 additions & 11 deletions src/neo/Ledger/MemoryPool.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -59,8 +59,6 @@ public class MemoryPool : IReadOnlyCollection<Transaction>
internal int SortedTxCount => _sortedTransactions.Count;
internal int UnverifiedSortedTxCount => _unverifiedSortedTransactions.Count;

private int _maxTxPerBlock;

/// <summary>
/// Total maximum capacity of transactions the pool can hold.
/// </summary>
Expand Down Expand Up @@ -103,11 +101,6 @@ public MemoryPool(NeoSystem system, int capacity)
Capacity = capacity;
}

internal void LoadPolicy(DataCache snapshot)
{
_maxTxPerBlock = (int)NativeContract.Policy.GetMaxTransactionsPerBlock(snapshot);
}

/// <summary>
/// Determine whether the pool is holding this transaction and has at some point verified it.
/// Note: The pool may not have verified it since the last block was persisted. To get only the
Expand Down Expand Up @@ -350,8 +343,6 @@ internal void InvalidateVerifiedTransactions()
// Note: this must only be called from a single thread (the Blockchain actor)
internal void UpdatePoolForBlockPersisted(Block block, DataCache snapshot)
{
LoadPolicy(snapshot);

_txRwLock.EnterWriteLock();
try
{
Expand All @@ -370,8 +361,8 @@ internal void UpdatePoolForBlockPersisted(Block block, DataCache snapshot)
_txRwLock.ExitWriteLock();
}

ReverifyTransactions(_sortedTransactions, _unverifiedSortedTransactions,
_maxTxPerBlock, MaxMillisecondsToReverifyTx, snapshot);
uint _maxTxPerBlock = NativeContract.Policy.GetMaxTransactionsPerBlock(snapshot);
ReverifyTransactions(_sortedTransactions, _unverifiedSortedTransactions, (int)_maxTxPerBlock, MaxMillisecondsToReverifyTx, snapshot);
}

internal void InvalidateAllTransactions()
Expand Down Expand Up @@ -468,6 +459,7 @@ internal bool ReVerifyTopUnverifiedTransactionsIfNeeded(int maxToVerify, DataCac
{
if (_unverifiedSortedTransactions.Count > 0)
{
uint _maxTxPerBlock = NativeContract.Policy.GetMaxTransactionsPerBlock(snapshot);
int verifyCount = _sortedTransactions.Count > _maxTxPerBlock ? 1 : maxToVerify;
ReverifyTransactions(_sortedTransactions, _unverifiedSortedTransactions,
verifyCount, MaxMillisecondsToReverifyTxPerIdle, snapshot);
Expand Down
58 changes: 29 additions & 29 deletions src/neo/Network/P2P/RemoteNode.ProtocolHandler.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -170,23 +170,22 @@ private void OnGetAddrMessageReceived()
/// <param name="payload">A GetBlocksPayload including start block Hash and number of blocks requested.</param>
private void OnGetBlocksMessageReceived(GetBlocksPayload payload)
{
// The default value of payload.Count is -1
int count = payload.Count < 0 || payload.Count > InvPayload.MaxHashesCount ? InvPayload.MaxHashesCount : payload.Count;
DataCache snapshot = Blockchain.Singleton.View;
UInt256 hash = payload.HashStart;
TrimmedBlock state = NativeContract.Ledger.GetTrimmedBlock(snapshot, hash);
if (state == null) return;
uint currentHeight = NativeContract.Ledger.CurrentIndex(snapshot);
List<UInt256> hashes = new List<UInt256>();
using (SnapshotCache snapshot = Blockchain.Singleton.GetSnapshot())
for (uint i = 1; i <= count; i++)
{
UInt256 hash = payload.HashStart;
// The default value of payload.Count is -1
int count = payload.Count < 0 || payload.Count > InvPayload.MaxHashesCount ? InvPayload.MaxHashesCount : payload.Count;
TrimmedBlock state = NativeContract.Ledger.GetTrimmedBlock(snapshot, hash);
if (state == null) return;
for (uint i = 1; i <= count; i++)
{
uint index = state.Index + i;
if (index > Blockchain.Singleton.Height)
break;
hash = NativeContract.Ledger.GetBlockHash(snapshot, index);
if (hash == null) break;
hashes.Add(hash);
}
uint index = state.Index + i;
if (index > currentHeight)
break;
hash = NativeContract.Ledger.GetBlockHash(snapshot, index);
if (hash == null) break;
hashes.Add(hash);
}
if (hashes.Count == 0) return;
EnqueueMessage(Message.Create(MessageCommand.Inv, InvPayload.Create(InventoryType.Block, hashes.ToArray())));
Expand Down Expand Up @@ -273,18 +272,15 @@ private void OnGetDataMessageReceived(InvPayload payload)
/// <param name="payload">A GetBlocksPayload including start block index and number of blocks' headers requested.</param>
private void OnGetHeadersMessageReceived(GetBlockByIndexPayload payload)
{
uint index = payload.IndexStart;
if (index > Blockchain.Singleton.Height) return;
DataCache snapshot = Blockchain.Singleton.View;
if (payload.IndexStart > NativeContract.Ledger.CurrentIndex(snapshot)) return;
List<Header> headers = new List<Header>();
using (SnapshotCache snapshot = Blockchain.Singleton.GetSnapshot())
uint count = payload.Count == -1 ? HeadersPayload.MaxHeadersCount : (uint)payload.Count;
for (uint i = 0; i < count; i++)
{
uint count = payload.Count == -1 ? HeadersPayload.MaxHeadersCount : (uint)payload.Count;
for (uint i = 0; i < count; i++)
{
var header = NativeContract.Ledger.GetHeader(snapshot, index + i);
if (header == null) break;
headers.Add(header);
}
var header = NativeContract.Ledger.GetHeader(snapshot, payload.IndexStart + i);
if (header == null) break;
headers.Add(header);
}
if (headers.Count == 0) return;
EnqueueMessage(Message.Create(MessageCommand.Headers, HeadersPayload.Create(headers.ToArray())));
Expand All @@ -295,7 +291,7 @@ private void OnInventoryReceived(IInventory inventory)
pendingKnownHashes.Remove(inventory.Hash);
if (inventory is Block block)
{
if (block.Index > Blockchain.Singleton.Height + InvPayload.MaxHashesCount) return;
if (block.Index > NativeContract.Ledger.CurrentIndex(Blockchain.Singleton.View) + InvPayload.MaxHashesCount) return;
UpdateLastBlockIndex(block.Index, false);
}
knownHashes.Add(inventory.Hash);
Expand All @@ -310,12 +306,16 @@ private void OnInvMessageReceived(InvPayload payload)
switch (payload.Type)
{
case InventoryType.Block:
using (SnapshotCache snapshot = Blockchain.Singleton.GetSnapshot())
{
DataCache snapshot = Blockchain.Singleton.View;
hashes = hashes.Where(p => !NativeContract.Ledger.ContainsBlock(snapshot, p)).ToArray();
}
break;
case InventoryType.TX:
using (SnapshotCache snapshot = Blockchain.Singleton.GetSnapshot())
{
DataCache snapshot = Blockchain.Singleton.View;
hashes = hashes.Where(p => !NativeContract.Ledger.ContainsTransaction(snapshot, p)).ToArray();
}
break;
}
if (hashes.Length == 0) return;
Expand All @@ -333,7 +333,7 @@ private void OnMemPoolMessageReceived()
private void OnPingMessageReceived(PingPayload payload)
{
UpdateLastBlockIndex(payload.LastBlockIndex, true);
EnqueueMessage(Message.Create(MessageCommand.Pong, PingPayload.Create(Blockchain.Singleton.Height, payload.Nonce)));
EnqueueMessage(Message.Create(MessageCommand.Pong, PingPayload.Create(NativeContract.Ledger.CurrentIndex(Blockchain.Singleton.View), payload.Nonce)));
}

private void OnPongMessageReceived(PingPayload payload)
Expand Down
3 changes: 2 additions & 1 deletion src/neo/Network/P2P/RemoteNode.cs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
using Neo.Ledger;
using Neo.Network.P2P.Capabilities;
using Neo.Network.P2P.Payloads;
using Neo.SmartContract.Native;
using System.Collections;
using System.Collections.Generic;
using System.Linq;
Expand Down Expand Up @@ -173,7 +174,7 @@ private void OnStartProtocol()
{
var capabilities = new List<NodeCapability>
{
new FullNodeCapability(Blockchain.Singleton.Height)
new FullNodeCapability(NativeContract.Ledger.CurrentIndex(Blockchain.Singleton.View))
};

if (LocalNode.Singleton.ListenerTcpPort > 0) capabilities.Add(new ServerCapability(NodeCapabilityType.TcpServer, (ushort)LocalNode.Singleton.ListenerTcpPort));
Expand Down
Loading

0 comments on commit 234350d

Please sign in to comment.