Ec-860

benchmarks/src/test/scala/benches/SnapshotAssemblerBench.scala

@@ -7,12 +7,12 @@ import benches.SnapshotAssemblerBench.SnapshotAssemblerBenchState
 import encry.view.state.avlTree.utils.implicits.Instances._
 import benches.StateBenches.{StateBenchState, benchSettings}
 import benches.Utils.{getRandomTempDir, utxoFromBoxHolder}
+import encry.nvg.fast.sync.SnapshotProcessor
 import encry.settings.Settings
 import encry.storage.{RootNodesStorage, VersionalStorage}
 import encry.storage.VersionalStorage.{StorageKey, StorageValue, StorageVersion}
 import encry.storage.levelDb.versionalLevelDB.{LevelDbFactory, VLDBWrapper, VersionalLevelDBCompanion}
 import encry.utils.FileHelper
-import encry.view.fast.sync.SnapshotHolder
 import encry.view.state.UtxoState
 import encry.view.state.avlTree.AvlTree
 import org.encryfoundation.common.utils.TaggedTypes.Height

src/main/resources/TestNetSettings.conf

@@ -145,7 +145,7 @@ delivery-manager-dispatcher {
   throughput = 1
 }
 mempool-dispatcher {
-  mailbox-type = "encry.view.mempool.MemoryPool$MemoryPoolPriorityQueue"
+  mailbox-type = "encry.mpg.MemoryPool$MemoryPoolPriorityQueue"
   type = Dispatcher
   executor = "thread-pool-executor"
   thread-pool-executor.fixed-pool-size = 2

src/main/resources/application.conf

@@ -18,6 +18,8 @@ encry {
     miningDelay = 5s
     # Is CLI available
     useCli = true
+    #Debug regime
+    isTestMod = false
   }
   mempool {
     # The time during which the transaction is considered valid
@@ -151,7 +153,14 @@ akka.http {
 akka.http.routing {
   verbose-error-messages = on
 }
-
+modifiers-validator-router-dispatcher {
+  executor = "thread-pool-executor"
+  # allocate exactly 5 threads for this pool
+  thread-pool-executor {
+    core-pool-size-min = 5
+    core-pool-size-max = 5
+  }
+}
 mining-dispatcher {
   type = Dispatcher
   executor = "thread-pool-executor"
@@ -211,7 +220,7 @@ delivery-manager-dispatcher {
   throughput = 1
 }
 mempool-dispatcher {
-  mailbox-type = "encry.view.mempool.MemoryPool$MemoryPoolPriorityQueue"
+  mailbox-type = "encry.mpg.MemoryPool$MemoryPoolPriorityQueue"
   type = Dispatcher
   executor = "thread-pool-executor"
   thread-pool-executor.fixed-pool-size = 2

src/main/scala/encry/EncryApp.scala

@@ -4,14 +4,14 @@ import java.io.File
 import java.net.InetAddress
 import java.nio.file.Files
 import akka.actor.SupervisorStrategy.Restart
-import akka.actor.{ActorRef, ActorSystem, OneForOneStrategy, Props}
+import akka.actor.{ ActorRef, ActorSystem, OneForOneStrategy, Props }
 import akka.http.scaladsl.Http
 import akka.http.scaladsl.model.HttpResponse
 import akka.http.scaladsl.server.ExceptionHandler
 import akka.stream.ActorMaterializer
 import com.typesafe.scalalogging.StrictLogging
 import encry.api.http.routes._
-import encry.api.http.{ApiRoute, CompositeHttpService}
+import encry.api.http.{ ApiRoute, CompositeHttpService }
 import encry.settings.EncryAppSettings
 import encry.stats.Zombie
 import encry.utils.NetworkTimeProvider
@@ -20,7 +20,7 @@ import kamon.influxdb.InfluxDBReporter
 import kamon.system.SystemMetrics
 import org.encryfoundation.common.utils.Algos
 import scala.concurrent.duration._
-import scala.concurrent.{Await, ExecutionContextExecutor, Future}
+import scala.concurrent.{ Await, ExecutionContextExecutor, Future }
 import scala.io.Source
 import scala.language.postfixOps
 

src/main/scala/encry/Starter.scala

@@ -1,10 +1,11 @@
 package encry
 
 import java.net.InetSocketAddress
-import akka.actor.{ Actor, ActorRef }
+
+import akka.actor.{Actor, ActorRef}
 import akka.http.scaladsl.Http
 import cats.Functor
-import cats.data.{ NonEmptyChain, Validated }
+import cats.data.{NonEmptyChain, Validated}
 import cats.instances.future._
 import cats.instances.option._
 import cats.syntax.apply._
@@ -15,19 +16,20 @@ import encry.Starter.InitNodeResult
 import encry.api.http.DataHolderForApi
 import encry.api.http.DataHolderForApi.PassForStorage
 import encry.cli.ConsoleListener
-import encry.cli.ConsoleListener.{ prompt, StartListening }
+import encry.cli.ConsoleListener.{StartListening, prompt}
 import encry.local.miner.Miner
 import encry.local.miner.Miner.StartMining
-import encry.network.NodeViewSynchronizer
+import encry.mpg.IntermediaryMempool
+import encry.network.NetworkRouter
+import encry.nvg.IntermediaryNVH
 import encry.settings._
 import encry.stats.StatsSender
-import encry.utils.{ Mnemonic, NetworkTimeProvider }
-import encry.view.NodeViewHolder
-import encry.view.mempool.MemoryPool
+import encry.utils.{Mnemonic, NetworkTimeProvider}
 import encry.view.wallet.AccountManager
+
 import scala.concurrent.Future
 import scala.io.StdIn
-import scala.util.{ Failure, Success, Try }
+import scala.util.{Failure, Success, Try}
 
 class Starter(settings: EncryAppSettings,
               timeProvider: NetworkTimeProvider,
@@ -40,7 +42,7 @@ class Starter(settings: EncryAppSettings,
 
   var initHttpApiServer: Option[Future[Http.ServerBinding]] = none
 
-  val preview =
+  val preview: String =
     """
       |XXXXXX  XX      XX    XXXXX  XXXXXX   XX    XX
       |XX      XXXX    XX  XX       XX   XX   XXXXXX
@@ -404,31 +406,26 @@ class Starter(settings: EncryAppSettings,
         context.system
           .actorOf(StatsSender.props(influxSettings, newSettings.network, newSettings.constants), "statsSender")
       }
-      lazy val dataHolderForApi =
+      val dataHolderForApi =
         context.system.actorOf(DataHolderForApi.props(newSettings, timeProvider), "dataHolder")
-      lazy val miner: ActorRef =
-        context.system.actorOf(Miner.props(dataHolderForApi, influxRef, newSettings), "miner")
-      lazy val memoryPool: ActorRef = context.system.actorOf(
-        MemoryPool
-          .props(newSettings, timeProvider, miner, influxRef)
-          .withDispatcher("mempool-dispatcher")
-      )
-      val nodeViewHolder: ActorRef = context.system.actorOf(
-        NodeViewHolder
-          .props(memoryPool, influxRef, dataHolderForApi, newSettings)
-          .withDispatcher("nvh-dispatcher"),
-        "nodeViewHolder"
-      )
 
       if (nodePass.nonEmpty) dataHolderForApi ! PassForStorage(nodePass)
 
-      context.system.actorOf(
-        NodeViewSynchronizer
-          .props(influxRef, nodeViewHolder, newSettings, memoryPool, dataHolderForApi)
+      val networkRouter = context.system.actorOf(
+        NetworkRouter
+          .props(networkSettings, settings.blackList, dataHolderForApi)
           .withDispatcher("nvsh-dispatcher"),
-        "nodeViewSynchronizer"
+        "networkRouter"
       )
 
+      val memoryPool: ActorRef = context.system.actorOf(
+        IntermediaryMempool.props(newSettings, timeProvider, influxRef, networkRouter)
+      )
+      val nvhRouter: ActorRef = context.system.actorOf(
+        IntermediaryNVH.props(newSettings, networkRouter, timeProvider, influxRef, memoryPool, dataHolderForApi)
+      )
+      val miner: ActorRef =
+        context.system.actorOf(Miner.props(dataHolderForApi, memoryPool, nvhRouter, influxRef, newSettings, timeProvider), "miner")
       if (newSettings.node.mining) miner ! StartMining
       if (newSettings.node.useCli) {
         context.system
@@ -441,15 +438,17 @@ class Starter(settings: EncryAppSettings,
 }
 
 object Starter {
-  final case class InitNodeResult(mnemonic: String,
-                                  walletPassword: String,
-                                  offlineGeneration: Boolean,
-                                  fastSync: Boolean,
-                                  snapshotCreation: Boolean,
-                                  peers: List[InetSocketAddress],
-                                  connectWithOnlyKnownPeers: Boolean,
-                                  nodePass: String = "",
-                                  nodeName: String,
-                                  declaredAddr: Option[InetSocketAddress],
-                                  bindAddr: InetSocketAddress)
+  final case class InitNodeResult(
+    mnemonic: String,
+    walletPassword: String,
+    offlineGeneration: Boolean,
+    fastSync: Boolean,
+    snapshotCreation: Boolean,
+    peers: List[InetSocketAddress],
+    connectWithOnlyKnownPeers: Boolean,
+    nodePass: String = "",
+    nodeName: String,
+    declaredAddr: Option[InetSocketAddress],
+    bindAddr: InetSocketAddress
+  )
 }

src/main/scala/encry/api/http/DataHolderForApi.scala

@@ -1,6 +1,7 @@
 package encry.api.http
 
 import java.net.{InetAddress, InetSocketAddress}
+
 import akka.actor.{Actor, ActorRef, Props, Stash}
 import akka.pattern._
 import akka.util.Timeout
@@ -18,19 +19,21 @@ import encry.network.BlackList.BanReason.InvalidNetworkMessage
 import encry.network.BlackList.{BanReason, BanTime, BanType}
 import encry.network.ConnectedPeersCollection
 import encry.network.NodeViewSynchronizer.ReceivableMessages._
-import encry.network.PeerConnectionHandler.ConnectedPeer
+import encry.network.PeerConnectionHandler.{ConnectedPeer, ConnectionType}
 import encry.network.PeersKeeper.BanPeerFromAPI
+import encry.nvg.NodeViewHolder.{NodeViewChange, UpdateHistoryReader}
 import encry.settings.EncryAppSettings
 import encry.utils.{NetworkTime, NetworkTimeProvider}
-import encry.view.NodeViewHolder.ReceivableMessages.{CreateAccountManagerFromSeed, GetDataFromCurrentView}
-import encry.view.history.History
+import encry.view.NodeViewHolder.ReceivableMessages.GetDataFromCurrentView
+import encry.view.history.{History, HistoryReader}
 import encry.view.state.{UtxoState, UtxoStateReader}
-import encry.view.wallet.EncryWallet
+import encry.view.wallet.{EncryWallet, WalletReader}
 import org.encryfoundation.common.crypto.PrivateKey25519
 import org.encryfoundation.common.modifiers.history.{Block, Header}
 import org.encryfoundation.common.modifiers.state.box.Box.Amount
 import org.encryfoundation.common.utils.Algos
 import org.encryfoundation.common.utils.TaggedTypes.ModifierId
+
 import scala.concurrent.Future
 
 class DataHolderForApi(settings: EncryAppSettings, ntp: NetworkTimeProvider)
@@ -52,23 +55,24 @@ class DataHolderForApi(settings: EncryAppSettings, ntp: NetworkTimeProvider)
   val launchTimeFuture: Future[NetworkTime.Time] = ntp.time()
 
   def awaitNVHRef: Receive = {
-    case UpdatedHistory(history) =>
+    case history: HistoryReader =>
       unstashAll()
+      logger.info("Got updated history at nvh")
       context.become(workingCycle(nvhRef = sender(), history = Some(history)))
     case PassForStorage(_) =>
       stash()
     case GetNodePassHashAndSalt => sender() ! storage.verifyPassword
   }
 
   def workingCycle(nvhRef: ActorRef,
-                   blackList: Seq[(InetAddress, (BanReason, BanTime, BanType))] = Seq.empty,
-                   connectedPeers: Seq[ConnectedPeer] = Seq.empty,
-                   history: Option[History] = None,
+                   blackList: List[(InetAddress, (BanReason, BanTime, BanType))] = List.empty,
+                   connectedPeers: List[(InetSocketAddress, String, ConnectionType)] = List.empty,
+                   history: Option[HistoryReader] = None,
                    state: Option[UtxoStateReader] = None,
                    transactionsOnMinerActor: Int = 0,
                    minerStatus: MinerStatus = MinerStatus(isMining = false, None),
                    blockInfo: BlockAndHeaderInfo = BlockAndHeaderInfo(None, None),
-                   allPeers: Seq[InetSocketAddress] = Seq.empty,
+                   allPeers: List[InetSocketAddress] = List.empty,
                    connectedPeersCollection: ConnectedPeersCollection = ConnectedPeersCollection()): Receive = {
 
     case UpdatingTransactionsNumberForApi(qty) =>
@@ -119,7 +123,7 @@ class DataHolderForApi(settings: EncryAppSettings, ntp: NetworkTimeProvider)
                      connectedPeersCollection)
       )
 
-    case ChangedHistory(reader: History) =>
+    case reader: HistoryReader =>
       context.become(
         workingCycle(nvhRef,
                      blackList,
@@ -133,7 +137,7 @@ class DataHolderForApi(settings: EncryAppSettings, ntp: NetworkTimeProvider)
                      connectedPeersCollection)
       )
 
-    case ChangedState(reader: UtxoStateReader) =>
+    case reader: UtxoStateReader =>
       context.become(
         workingCycle(nvhRef,
                      blackList,
@@ -180,32 +184,32 @@ class DataHolderForApi(settings: EncryAppSettings, ntp: NetworkTimeProvider)
     case RemovePeerFromBanList(peer) => context.system.eventStream.publish(RemovePeerFromBlackList(peer))
 
     case GetViewPrintAddress =>
-      (nvhRef ? GetDataFromCurrentView[History, UtxoState, EncryWallet, String] { view =>
+      (nvhRef ? GetDataFromCurrentView[HistoryReader, UtxoStateReader, WalletReader, String] { view =>
         view.vault.publicKeys.foldLeft("") { (str, k) =>
           str + s"Pay2PubKeyAddress : ${k.address.address} , Pay2ContractHashAddress : ${k.address.p2ch.address}" + "\n"
         }
       }).pipeTo(sender)
 
     case GetViewCreateKey =>
-      (nvhRef ? GetDataFromCurrentView[History, UtxoState, EncryWallet, PrivateKey25519] { view =>
+      (nvhRef ? GetDataFromCurrentView[HistoryReader, UtxoStateReader, WalletReader, PrivateKey25519] { view =>
          view.vault.accountManagers.head.createAccount(None)
       }).pipeTo(sender)
 
     case GetViewPrintPubKeys =>
-      (nvhRef ? GetDataFromCurrentView[History, UtxoState, EncryWallet, List[String]] { view =>
+      (nvhRef ? GetDataFromCurrentView[HistoryReader, UtxoStateReader, WalletReader, List[String]] { view =>
         view.vault.publicKeys.foldLeft(List.empty[String])((str, k) => str :+ Algos.encode(k.pubKeyBytes))
       }).pipeTo(sender)
 
     case GetViewGetBalance =>
-      (nvhRef ? GetDataFromCurrentView[History, UtxoState, EncryWallet, Map[String, List[(String, Amount)]]] { view =>
+      (nvhRef ? GetDataFromCurrentView[HistoryReader, UtxoStateReader, WalletReader, Map[String, List[(String, Amount)]]] { view =>
         val balance: Map[String, List[(String, Amount)]] = view.vault.getBalances.map {
           case ((key, token), amount) => Map(key -> List((token, amount)))
         }.foldLeft(Map.empty[String, List[(String, Amount)]]) { case (el1, el2) => el1 |+| el2 }
         if (balance.isEmpty) Map.empty[String, List[(String, Amount)]] else balance
       }).pipeTo(sender)
 
     case GetViewPrintPrivKeys =>
-      (nvhRef ? GetDataFromCurrentView[History, UtxoState, EncryWallet, String] { view =>
+      (nvhRef ? GetDataFromCurrentView[HistoryReader, UtxoStateReader, WalletReader, String] { view =>
         view.vault.accountManagers.head.accounts.foldLeft("")((str, k) => str + Algos.encode(k.privKeyBytes) + "\n")
       }).pipeTo(sender)
 
@@ -227,18 +231,13 @@ class DataHolderForApi(settings: EncryAppSettings, ntp: NetworkTimeProvider)
     case GetBannedPeersHelper => sender() ! blackList
 
     case GetConnectedPeersHelper => sender() ! connectedPeers
-      .map(
-        peer =>
-              PeerInfoResponse(peer.socketAddress.toString,
-                               Some(peer.handshake.nodeName),
-                               Some(peer.direction.toString)))
-
+      .map(peer => PeerInfoResponse(peer._1.toString, peer._2, peer._3.toString))
 
     case GetLastHeaderIdAtHeightHelper(i) =>
       sender() ! history.toList.flatMap(_.headerIdsAtHeight(i).map(Algos.encode))
 
     case CreateAccountManagerFromSeedHelper(seed) =>
-      (nvhRef ? CreateAccountManagerFromSeed(seed)).mapTo[Either[String, EncryWallet]].pipeTo(sender())
+      //(nvhRef ? CreateAccountManagerFromSeed(seed)).mapTo[Either[String, EncryWallet]].pipeTo(sender())
 
     case GetAllInfoHelper =>
 
@@ -314,9 +313,9 @@ object DataHolderForApi { //scalastyle:ignore
 
   final case class UpdatingMinerStatus(minerStatus: MinerStatus) extends AnyVal
 
-  final case class UpdatingPeersInfo(allPeers: Seq[InetSocketAddress],
-                                     connectedPeers: Seq[ConnectedPeer],
-                                     blackList: Seq[(InetAddress, (BanReason, BanTime, BanType))])
+  final case class UpdatingPeersInfo(allPeers: List[InetSocketAddress],
+                                     connectedPeers: List[(InetSocketAddress, String, ConnectionType)],
+                                     blackList: List[(InetAddress, (BanReason, BanTime, BanType))])
 
   final case class BlockAndHeaderInfo(header: Option[Header], block: Option[Block])
 
@@ -330,7 +329,7 @@ object DataHolderForApi { //scalastyle:ignore
 
   final case class GetLastHeaderIdAtHeightHelper(i: Int)
 
-  final case class Readers(h: Option[History], s: Option[UtxoStateReader])
+  final case class Readers(h: Option[HistoryReader], s: Option[UtxoStateReader])
 
   final case class PassForStorage(password: String)