diff --git a/.gitignore b/.gitignore index 589c9fb..60a41df 100755 --- a/.gitignore +++ b/.gitignore @@ -4,8 +4,9 @@ project/target/ project/plugins/ .idea/ *.log -neodb/ .ensime .ensime_cache/ src/main/resources/application.conf -.history \ No newline at end of file +.history +blockchain +ensime.sbt \ No newline at end of file diff --git a/api/ensime.sbt b/api/ensime.sbt new file mode 100644 index 0000000..02ef3dc --- /dev/null +++ b/api/ensime.sbt @@ -0,0 +1 @@ +ensimeScalaVersion in ThisBuild := "2.11.8" diff --git a/api/project/build.scala b/api/project/build.scala index a21241b..4ea3764 100644 --- a/api/project/build.scala +++ b/api/project/build.scala @@ -11,7 +11,7 @@ import com.earldouglas.xwp.ContainerPlugin.autoImport._ object BgeapiBuild extends Build { val Organization = "net.bitcoinprivacy" val Name = "bgeapi" - val Version = "3.3" + val Version = "3.3-SNAPSHOT" val ScalaVersion = "2.11.8" val ScalatraVersion = "2.4.0" @@ -24,6 +24,7 @@ object BgeapiBuild extends Build { version := Version, scalaVersion := ScalaVersion, resolvers += Classpaths.typesafeReleases, + resolvers += "Local Maven Repository" at "file:///root/.m2/repository", resolvers += "Scalaz Bintray Repo" at "http://dl.bintray.com/scalaz/releases", libraryDependencies ++= Seq( "net.bitcoinprivacy" %% "bge" % Version, diff --git a/api/project/plugins.sbt b/api/project/plugins.sbt index 25caeb5..f853f72 100644 --- a/api/project/plugins.sbt +++ b/api/project/plugins.sbt @@ -1,6 +1,6 @@ -addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.3") +addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.6") -addSbtPlugin("com.typesafe.sbt" % "sbt-native-packager" % "1.2.0") +addSbtPlugin("com.typesafe.sbt" % "sbt-native-packager" % "1.3.3") addSbtPlugin("com.mojolly.scalate" % "xsbt-scalate-generator" % "0.5.0") diff --git a/api/src/main/scala/api/MyScalatraServlet.scala b/api/src/main/scala/api/MyScalatraServlet.scala index 42a8e9f..af2f9de 100644 --- a/api/src/main/scala/api/MyScalatraServlet.scala +++ b/api/src/main/scala/api/MyScalatraServlet.scala @@ -9,12 +9,41 @@ import org.json4s.{DefaultFormats, Formats} import org.scalatra.json._ import net.bitcoinprivacy.bge.models._ import org.bitcoinj.core.{Address => BitcoinJAddress} -import org.bitcoinj.params.MainNetParams +import org.bitcoinj.params._ import org.scalatra.CorsSupport +import com.typesafe.config.ConfigFactory class MyScalatraServlet extends BgeapiStack with db.BitcoinDB with JacksonJsonSupport with CorsSupport { - // Sets up automatic case class to JSON output serialization, required by + lazy val conf = ConfigFactory.load() + lazy val network = conf.getString("network") + + lazy val networkParams = network match { + case "main" => + MainNetParams.get + case "regtest" => + RegTestParams.get + case "testnet" => + TestNet3Params.get + case _ => + throw new Exception("Unknow params for network"+network) + + } + + case class AddressPrefixes(p2pAddress: String, normalAddress: String) + lazy val prefix = network match { + case "main" => + AddressPrefixes("05", "00") + case "regtest" => + AddressPrefixes("C4", "6F") + case "testnet" => + AddressPrefixes("C4", "6F") + case _ => + throw new Exception("Unknow params for network"+network) + + } + + // Sets up automatic case class to JSON output serialization, required by // the JValueResult trait. protected implicit lazy val jsonFormats: Formats = DefaultFormats @@ -144,13 +173,13 @@ class MyScalatraServlet extends BgeapiStack with db.BitcoinDB with JacksonJsonSu def hexAddress(stringAddress: String): String = { val arrayAddress = stringAddress.split(",") if (arrayAddress.length == 1) { - val address = new BitcoinJAddress(MainNetParams.get, stringAddress) - (if(address.isP2SHAddress) "05" else "00")+valueOf(address.getHash160) + val address = new BitcoinJAddress(networkParams, stringAddress) + (if(address.isP2SHAddress) prefix.p2pAddress else prefix.normalAddress)+valueOf(address.getHash160) } else{ "0" + arrayAddress.length + (for (i <- 0 until arrayAddress.length) - yield valueOf(new BitcoinJAddress(MainNetParams.get, arrayAddress(i)).getHash160) ).mkString("") + yield valueOf(new BitcoinJAddress(networkParams, arrayAddress(i)).getHash160) ).mkString("") } } diff --git a/api/src/main/scala/api/models/Address.scala b/api/src/main/scala/api/models/Address.scala index e52f651..f925887 100644 --- a/api/src/main/scala/api/models/Address.scala +++ b/api/src/main/scala/api/models/Address.scala @@ -2,15 +2,28 @@ package net.bitcoinprivacy.bge.models import db._ import org.bitcoinj.core.{Address => Add} -import org.bitcoinj.params.MainNetParams +import org.bitcoinj.params._ import scala.slick.driver.PostgresDriver.simple._ - +import com.typesafe.config.ConfigFactory case class Address(address: String, balance: Long) case class AddressesSummary(count: Int, sum: Long) object Address extends db.BitcoinDB { - + lazy val conf = ConfigFactory.load() + lazy val network = conf.getString("network") + + lazy val params = network match { + case "main" => + MainNetParams.get + case "regtest" => + RegTestParams.get + case "testnet" => + TestNet3Params.get + case _ => + throw new Exception(s"Unknow params for network $network") + } + def getWallet(hash: Array[Byte], from: Int, until: Int) = DB withSession { implicit session => val repOpt = addresses.filter(_.hash === hash).map(_.representant).firstOption @@ -80,20 +93,21 @@ object Address extends db.BitcoinDB { } - def hashToAddress(hash: Array[Byte]): String = hash.length match { + def hashToAddress(hash: Array[Byte]): String = try {hash.length match { - case 20 => new Add(MainNetParams.get,0,hash).toString + case 20 => new Add(params,params.getAddressHeader,hash).toString - case 21 => new Add(MainNetParams.get,hash.head.toInt,hash.tail).toString + case 21 => new Add(params,hash.head.toInt,hash.tail).toString case 0 => "No decodable address found" case x if (x%20==1) => - + try { (for (i <- 1 to hash.length-20 by 20) yield hashToAddress(hash.slice(i,i+20)) ).mkString(",") - + } catch { case e:Exception => hash.toString } case _ => hash.length + " undefined" - } + }} catch { case _ => "Bitcoinj failed decoding address" } + } diff --git a/api/src/main/scala/api/models/Stats.scala b/api/src/main/scala/api/models/Stats.scala index b4ae064..be8e2cb 100644 --- a/api/src/main/scala/api/models/Stats.scala +++ b/api/src/main/scala/api/models/Stats.scala @@ -1,19 +1,23 @@ package net.bitcoinprivacy.bge.models import scala.slick.driver.PostgresDriver.simple._ +import com.typesafe.config.ConfigFactory - -case class Stats(block_height:Int, total_bitcoins_in_addresses:Long, total_transactions:Long, total_addresses:Long, total_closures:Long, total_addresses_with_balance:Long, total_closures_with_balance:Long, total_addresses_no_dust:Long, total_closures_no_dust:Long, gini_closure:Double,gini_address:Double, tstamp:Long) +case class Stats(block_height:Int, total_bitcoins_in_addresses:Long, total_transactions:Long, total_addresses:Long, total_closures:Long, total_addresses_with_balance:Long, total_closures_with_balance:Long, total_addresses_no_dust:Long, total_closures_no_dust:Long, gini_closure:Double,gini_address:Double, tstamp:Long, network:String, dust: Int) object Stats extends db.BitcoinDB -{ +{ + lazy val conf = ConfigFactory.load() + lazy val dustLimit = conf.getInt("dustLimit") + lazy val network= conf.getString("network") + def getStats = DB withSession { implicit session => val o = stats.sortBy(_.block_height.desc).firstOption for (p <- o) - yield Stats(p._1,p._2,p._3,p._4,p._5,p._6,p._7,p._8,p._9,p._10,p._11,p._12) + yield Stats(p._1,p._2,p._3,p._4,p._5,p._6,p._7,p._8,p._9,p._10,p._11,p._12, network, dustLimit) } def getAllStats = @@ -22,7 +26,7 @@ object Stats extends db.BitcoinDB val p1 = stats.sortBy(_.block_height.desc).run.toList p1 map { - p => Stats(p._1,p._2,p._3,p._4,p._5,p._6,p._7,p._8,p._9,p._10,p._11,p._12) + p => Stats(p._1,p._2,p._3,p._4,p._5,p._6,p._7,p._8,p._9,p._10,p._11,p._12, network, dustLimit) } } } diff --git a/build.sbt b/build.sbt index 057388e..e4ccbea 100644 --- a/build.sbt +++ b/build.sbt @@ -2,25 +2,25 @@ organization := "net.bitcoinprivacy" name := "bge" -version := "3.3" +version := "3.3-SNAPSHOT" -scalaVersion := "2.11.8" +scalaVersion := "2.12.4" maintainer := "Bitcoinprivacy " // additional libraries libraryDependencies ++= Seq( - "org.bitcoinj" % "bitcoinj-core" % "0.13.6", + "org.bitcoinj" % "bitcoinj-core" % "0.15-SNAPSHOT", "org.xerial.snappy"%"snappy-java"%"1.1.2.4", "org.iq80.leveldb"%"leveldb"%"0.7", "org.fusesource.leveldbjni"%"leveldbjni-all"%"1.8", - "org.postgresql" % "postgresql" % "9.4-1200-jdbc41", + "org.postgresql" % "postgresql" % "42.2.2", "com.typesafe.slick" %% "slick" % "2.1.0", "com.typesafe" % "config" % "1.2.1", - "org.scalacheck" %% "scalacheck" % "1.12.4" % "test", - "org.scalatest" %% "scalatest" % "2.1.5" % "test", + "org.scalacheck" %% "scalacheck" % "1.13.5" % "test", + "org.scalatest" %% "scalatest" % "3.2.0-SNAP10" % "test", "org.deephacks.lmdbjni" % "lmdbjni" % "0.4.6", - "com.typesafe.scala-logging" %% "scala-logging" % "3.4.0", + "com.typesafe.scala-logging" %% "scala-logging" % "3.5.0", "ch.qos.logback" % "logback-classic" % "1.1.7", //////////////////////////////////////////////////////////////////// // select one of the following depending on your architecture @@ -31,10 +31,15 @@ libraryDependencies ++= Seq( ) enablePlugins(JavaAppPackaging) +enablePlugins(DockerPlugin) + +dockerEntrypoint := Seq("bin/bge", "-Dlogback.configurationFile=/conf/logback.xml", "-Dconfig.file=/conf/bge.conf", "--", "start") +dockerUsername := Some("jorgemartinezpizarro") +daemonUser in Docker := "root" libraryDependencies ~= { _.map(_.exclude("org.slf4j", "slf4j-simple")) } -resolvers += "Local Maven Repository" at "file:///"+Path.userHome.absolutePath+"/.m2/repository" +resolvers += "Local Maven Repository" at "file:///root/.m2/repository" resolvers += "Fakod Snapshots" at "https://raw.github.com/FaKod/fakod-mvn-repo/master/snapshots" @@ -57,6 +62,7 @@ case x if Assembly.isConfigFile(x) => MergeStrategy.rename case PathList("META-INF", xs @ _*) => (xs map {_.toLowerCase}) match { + case ("manifest.mf" :: Nil) | ("index.list" :: Nil) | ("dependencies" :: Nil) => MergeStrategy.discard case ps @ (x :: xs) if ps.last.endsWith(".sf") || ps.last.endsWith(".dsa") => @@ -91,3 +97,5 @@ javaOptions in run += "-Xmx16G" javaOptions in run += "-Xms1G" fork := true + +testOptions in Test += Tests.Argument("-oD") diff --git a/project/build.properties b/project/build.properties index 59e7c05..66fe511 100644 --- a/project/build.properties +++ b/project/build.properties @@ -1 +1 @@ -sbt.version=0.13.11 \ No newline at end of file +sbt.version=1.1.0 \ No newline at end of file diff --git a/project/plugins.sbt b/project/plugins.sbt index 796830d..c2374bd 100644 --- a/project/plugins.sbt +++ b/project/plugins.sbt @@ -1,3 +1,3 @@ -addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.3") +addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.6") -addSbtPlugin("com.typesafe.sbt" % "sbt-native-packager" % "1.2.0") +addSbtPlugin("com.typesafe.sbt" % "sbt-native-packager" % "1.3.4") diff --git a/src/main/resources/reference.conf b/src/main/resources/reference.conf index d0ceb0a..1819993 100755 --- a/src/main/resources/reference.conf +++ b/src/main/resources/reference.conf @@ -11,11 +11,16 @@ levelDBFile = ${dataDir}"level.db" lockFile = ${dataDir}"lock" dustLimit=546 databaseName="postgres" -username = "root" -password = "trivial" -host = "localhost" +username = "user" +password = "pass" +host = "localhost:1234" jdbcOptions = "?useServerPrepStmts=false&rewriteBatchedStatements=true&maxWait=-1" mdb.path = ${dataDir}"mdb/" mdb.transactionSize = 100000 - -logger.scala.slick=ERROR \ No newline at end of file +network=main # accepts main, testnet, regtest +logger.scala.slick=ERROR +resumeBlockSize=1000000 # depending on the available ram +bitcoin.ip=localhost +bitcoin.maxPopulate=0 # 0 all,m number block until stop first populate. To rest partial blockchains without running all! +balanceUpdateLimit=50000 # for balance changes beyond this size, create new balance tables +checkUTXOsSize=100000 \ No newline at end of file diff --git a/src/main/scala/bge/Explorer.scala b/src/main/scala/bge/Explorer.scala index 8445268..27739d7 100755 --- a/src/main/scala/bge/Explorer.scala +++ b/src/main/scala/bge/Explorer.scala @@ -1,7 +1,7 @@ import actions._ import util._ import sys.process._ -import collection.mutable.Map + /** * Created with IntelliJ IDEA. @@ -13,20 +13,26 @@ import collection.mutable.Map object Explorer extends App with db.BitcoinDB { args.toList match{ case "start"::rest => - + if (!statsDone || rest.headOption == Some("--force")) populate + Seq("touch",lockFile).! - iterateResume + + iterateResume(false) case "populate"::rest => populate + case "closure"::rest => + + case "resume"::rest => Seq("touch",lockFile).! - iterateResume + + iterateResume(rest.headOption==Some("--newstats")) case "stop"::rest => @@ -45,9 +51,15 @@ object Explorer extends App with db.BitcoinDB { start [--force]: populate if necessary or --force, then resume stop: ask bge gently to stop at the end of the current iteration. Note that thís can take a few days if you are in the populate phase. populate: create the database movements with movements and closures. Deletes any existing data. - resume: update the database generated by populate with new incoming data. Works only if populate is already done. + resume [--newstats]: update the database generated by populate with new incoming data. Works only if populate is already done. The newstats flag forces a regeneration of balance, richlist and stats tables. info - """) + b""") + } + + def closure = { + new PopulateClosure((1 until blockCount).toVector) + createAddressIndexes + populateStats } def getInfo = { @@ -75,9 +87,9 @@ object Explorer extends App with db.BitcoinDB { lazy val outputMap: UTXOs = new UTXOs (table) // (txhash,index) -> (address,value,blockIn) val values = for ( (_,(_,value,_)) <- outputMap.view) yield value //makes it a lazy collection - val tuple = values.grouped(100000).foldLeft((0,0L)){ + val tuple = values.grouped(checkUTXOsSize).foldLeft((0,0L)){ case ((count,sum),group) => - log.info(count + " elements read at ") + //log.info(count + " elements read at ") val seq = group.toSeq (count+seq.size,sum+seq.sum) } @@ -98,63 +110,98 @@ object Explorer extends App with db.BitcoinDB { initializeStatsTables insertStatistics - + + if (!peerGroup.isRunning) startBitcoinJ + PopulateBlockReader - + createIndexes new PopulateClosure(PopulateBlockReader.processedBlocks) - createAddressIndexes + createAddressIndexes + syncAddressesWithUTXOs populateStats -// testValues - } def resume = { + val read = new ResumeBlockReader - val closure = new ResumeClosure(read.processedBlocks) - log.info("making new stats") - resumeStats(read.changedAddresses, closure.changedReps, closure.addedAds, closure.addedReps) + val closure = new ResumeClosure(read.processedBlocks, read.changedAddresses.toMap) + + // FIXME + // That check is neccessary due to a bug in updateBalances + // Delete checkBalances condition when it is fixed + val checkBalances = getSumBalance == getSumWalletsBalance + + if (!checkBalances) + log.info("Error during update of balances or stats .... creating balances and stats") + + if (read.changedAddresses.size < balanceUpdateLimit && checkBalances) { + val (adsAndBalances,repsAndChanges, repsAndBalances) = + resumeStats(read.changedAddresses.toMap, closure.touchedReps, closure.changedReps, closure.addedAds, closure.addedClosures) + Some( + (adsAndBalances,repsAndChanges, read.changedAddresses.toMap, repsAndBalances, closure.touchedReps, closure.changedReps, closure.addedAds, closure.addedClosures) + ) + } + else { + populateStats + None + } } - def iterateResume = { - // Seq("bitcoind","-daemon").run - - if (!peerGroup.isRunning) startBitcoinJ - - // if there are more stats than blocks we could delete it - for (block <- getWrongBlock){ - - rollBack(block) - populateStats - assert(getWrongBlock == None, "The database is inconsistent. See logs for details.") + def rollBackToLastStatIfNecessary: Boolean = { + var wrongBlockOption = getWrongBlock + var failed = false + while (wrongBlockOption != None) { + rollBack(wrongBlockOption.get) + failed = true + wrongBlockOption = getWrongBlock } + failed + } + + def iterateResume(newStats: Boolean) = { + + if (!peerGroup.isRunning) + startBitcoinJ + + log.info("Checking for wrong blocks") - while (new java.io.File(lockFile).exists) - { - if (blockCount > chain.getBestChainHeight) - { - log.info("waiting for new blocks") - chain.getHeightFuture(blockCount).get //wait until the chain overtakes our DB + val rollbacked = rollBackToLastStatIfNecessary + + if (rollbacked || newStats ) + populateStats + + while (new java.io.File(lockFile).exists) { + if (blockCount > chain.getBestChainHeight) { + log.info("Waiting for new blocks") + // wait until the chain overtakes our DB + waitForBitcoinJBlock(blockCount) } + resume + } - log.info("process stopped") + log.info("Lock file deleted. BGE shut down correctly!") } def getWrongBlock: Option[Int] = { - + + val lch = lastCompletedHeight + val bc = blockCount + if (bc-1 > lastCompletedHeight){ + log.error("Incomplete process, missing stats and maybe more data. Rollback required.") + return Some(bc-1) + } + val (count,amount) = sumUTXOs val (countDB, amountDB) = countUTXOs - val bc = blockCount val expected = totalExpectedSatoshi(bc) - val utxosMaxHeight = getUtxosMaxHeight - - val lch = lastCompletedHeight + val utxosMaxHeight = getUtxosMaxHeight val sameCount = count == countDB val sameValue = amount == amountDB val rightValue = amount <= expected - val rightBlock = (blockCount - 1 == lch) && utxosMaxHeight == lch + val rightBlock = (bc - 1 == lch) && utxosMaxHeight == lch if (!rightValue) log.error("we have " + ((amount-expected)/100000000.0) + " too many bitcoins") if (!sameCount) log.error("we lost utxos") @@ -164,30 +211,36 @@ object Explorer extends App with db.BitcoinDB { if (sameCount && sameValue && rightValue && rightBlock) None else if (utxosMaxHeight > bc -1) Some(utxosMaxHeight) else if (bc - 1 > lch) Some(bc-1) - else throw new Exception("This should not have happened. See logs for details.") + else Some(lch) + } - def resumeStats(changedAddresses: Map[Hash,Long], changedReps: Map[Hash,Set[Hash]], addedAds: Int, addedReps: Int) = { - - log.info(changedAddresses.size + " addresses changed balance") + def resumeStats(changedAddresses: Map[Hash,Long], touchedReps: Map[Hash,Set[Hash]], changedReps: Map[Hash,Set[Hash]], addedAds: Int, addedClosures: Int): + (Map[Hash, Long],Map[Hash, Long],Map[Hash, Long]) = { - if (changedAddresses.size < 38749 ) - { - updateBalanceTables(changedAddresses, changedReps) - insertRichestAddresses - insertRichestClosures - updateStatistics(changedReps,addedAds, addedReps) - } - else populateStats - + val (adsAndBalances,repsAndChanges, repsAndBalances) = updateBalanceTables(changedAddresses, touchedReps, changedReps) + + createRichestLists + + updateStatistics(addedAds, addedClosures) + + (adsAndBalances,repsAndChanges, repsAndBalances) } - def populateStats = { - createBalanceTables + def createRichestLists = { + var startTime = System.currentTimeMillis insertRichestAddresses insertRichestClosures - insertStatistics + log.info("Richest list created in " + (System.currentTimeMillis - startTime) / 1000 + "s") } + def populateStats = { + createBalanceTables + createRichestLists + insertStatistics + } + def waitForBitcoinJBlock(b: Int) = { + chain.getHeightFuture(b).get + } } diff --git a/src/main/scala/bge/actions/PopulateBlockReader.scala b/src/main/scala/bge/actions/PopulateBlockReader.scala index a3945b3..f70a349 100644 --- a/src/main/scala/bge/actions/PopulateBlockReader.scala +++ b/src/main/scala/bge/actions/PopulateBlockReader.scala @@ -5,7 +5,7 @@ import core._ import util._ import Hash._ -object PopulateBlockReader extends FastBlockReader with BitcoinDRawFileBlockSource { +object PopulateBlockReader extends FastBlockReader with PeerSource { // txhash -> ((index -> (address,value)),blockIn) // need to override this here to get the specialized type override lazy val table: LmdbMap = LmdbMap.create("utxos") diff --git a/src/main/scala/bge/actions/PopulateClosure.scala b/src/main/scala/bge/actions/PopulateClosure.scala index 9754b58..7126f23 100644 --- a/src/main/scala/bge/actions/PopulateClosure.scala +++ b/src/main/scala/bge/actions/PopulateClosure.scala @@ -11,19 +11,16 @@ class PopulateClosure(blockHeights: Vector[Int]) extends AddressClosure(blockHei lazy val table = LmdbMap.create("closures") override lazy val unionFindTable = new ClosureMap(table) - def saveTree(tree: DisjointSets[Hash]): Int = + def saveTree: Int = { + val tree = generatedTree val timeStart = System.currentTimeMillis var queries: Vector[(Array[Byte], Array[Byte])] = Vector() val totalElements = tree.elements.size var counter = 0 var counterTotal = 0 - - log.info("Saving tree to database...") var counterFinal = 0 - // tree.elements.keys.foldLeft(tree){(t,value) => - // val (parentOption, newTree) = tree.find(value) - // for (parent <- parentOption ) + table.commit for (key <- tree.elements.keys) @@ -42,12 +39,12 @@ class PopulateClosure(blockHeights: Vector[Int]) extends AddressClosure(blockHei } if (counterFinal % 1000000 == 0) { counterFinal = 0 - log.info("Saved until element %s in %s s, %s µs per element" format (counterTotal, (System.currentTimeMillis - timeStart)/1000, (System.currentTimeMillis - timeStart)*1000/(counterTotal+1))) + log.info("Saved to DB until element %s in %s s, %s µs per element" format (counterTotal, (System.currentTimeMillis - timeStart)/1000, (System.currentTimeMillis - timeStart)*1000/(counterTotal+1))) } // newTree } - log.info("Saved until element %s in %s s, %s µs per element" format (counterTotal, (System.currentTimeMillis - timeStart)/1000, (System.currentTimeMillis - timeStart)*1000/(counterTotal+1))) + log.info("Saved to DB until element %s in %s s, %s µs per element" format (counterTotal, (System.currentTimeMillis - timeStart)/1000, (System.currentTimeMillis - timeStart)*1000/(counterTotal+1))) saveElementsToDatabase(queries, counter) diff --git a/src/main/scala/bge/actions/ResumeBlockReader.scala b/src/main/scala/bge/actions/ResumeBlockReader.scala index e676a4e..a1febe1 100644 --- a/src/main/scala/bge/actions/ResumeBlockReader.scala +++ b/src/main/scala/bge/actions/ResumeBlockReader.scala @@ -92,7 +92,7 @@ class ResumeBlockReader extends FastBlockReader with PeerSource { } override def post = { - log.info("finishing ...") + //log.info("finishing ...") super.post table.close } @@ -108,7 +108,6 @@ class ResumeBlockReader extends FastBlockReader with PeerSource { } def saveUTXOs = { - log.info("Inserting UTXOs into SQL database ...") def vectorUTXOConverter[A, B, C](v: Map[(Hash, A), (Hash, B, C)]) = v map { case ((a, b), (c, d, e)) => (hashToArray(a), hashToArray(c), b, d, e) @@ -120,7 +119,5 @@ class ResumeBlockReader extends FastBlockReader with PeerSource { newUtxos.clear - log.info("Data inserted") - } } diff --git a/src/main/scala/bge/actions/ResumeClosure.scala b/src/main/scala/bge/actions/ResumeClosure.scala index c4799bc..b9b3cb3 100644 --- a/src/main/scala/bge/actions/ResumeClosure.scala +++ b/src/main/scala/bge/actions/ResumeClosure.scala @@ -2,114 +2,62 @@ package actions import core._ import db._ -import scala.slick.driver.PostgresDriver.simple._ import util._ -import util.Hash._ -import collection.mutable.Map +import Hash._ -class ResumeClosure(blockHeights: Vector[Int]) extends AddressClosure(blockHeights: Vector[Int]) with BitcoinDB{ +class ResumeClosure(blockHeights: Vector[Int], changedAds: Map[Hash,Long]) extends AddressClosure(blockHeights: Vector[Int]) with BitcoinDB { lazy val table = LmdbMap.open("closures") override lazy val unionFindTable = new ClosureMap(table) - lazy val changedReps = Map[Hash,Set[Hash]]() - - - override def insertInputsIntoTree(addressList: Iterable[Hash], tree: DisjointSets[Hash]): DisjointSets[Hash] = - { - val (pairList, tree1) = addressList.foldLeft((List[(Hash,Option[Hash])](),tree)){ - case ((l,t),a) => - val (repOpt,treex) = t.find(a) - (l:+(a,repOpt),treex) - } - - val addedTree = addressList.foldLeft(tree1)((t:DisjointSets[Hash],a:Hash) => t.add(a)) - val tree2 = addedTree.union(addressList) - - // update the postgres DB - - // must be at least 2 addresses - assert(addressList.size >= 2, "union of trivial closure") - val (representantOpt,result) = tree2 find (addressList.head) - val newRep = representantOpt.get - - var addRepFlag = 1 - - def recordChangedRep(newRep: Hash, oldRep: Hash) = { - // a changed to b, then b changed to c - // 1 we insert (b, a) in the map - // 2 we receive (c, b) cause b is already new, replace it with (c, {a,b}) - val oldReps = changedReps.getOrElse(newRep,Set()) - if (changedReps.contains(oldRep)) { - changedReps += (newRep -> (changedReps(oldRep)+ oldRep)) - changedReps -= oldRep - } - else changedReps += (newRep -> (oldReps + oldRep) ) + lazy val changedAddresses = changedAds.keys.toSet - Hash.zero(0) + //lazy val oldReps = getAddressReps(changedAddresses) // only read from addresses once is dangerous ... + lazy val oldReps = changedAddresses map (add => (add, getRepresentant(add))) toMap + + lazy val (finalTree, touchedReps) = // a map of new reps to sets of their addresses that have been touched + changedAddresses.foldLeft((generatedTree,Map[Hash,Set[Hash]]())){ + case ((tree,tR),address) => + val (repOpt, ntree) = tree.find(address) + (ntree, repOpt match { + case None => tR + (address -> Set(address)) + case Some (r) => tR + (r -> (tR.getOrElse(r, Set())+address)) + }) } - - - DB withSession { implicit session => - - for ((address, oldRepOpt) <- pairList) - oldRepOpt match - { - case None => - insertAddress(address,newRep) - - case Some(oldRep) if (oldRep != newRep) => - // if representant new, update everything that had the old one - val updateQuery = for(p <- addresses if p.representant === hashToArray(oldRep)) yield p.representant - updateQuery.update(newRep) // TODO: compile query - recordChangedRep(newRep, oldRep) - case _ => addRepFlag = 0 // one of the elements had this rep before => don't count a new one - } - } - addedReps += addRepFlag - - result - - - } - + // this includes addresses that are not yet in the tree! otherwise, the below would be easier + // streamOfUnionLists.flatten.flatten.toSet.groupBy(generatedTree.onlyFind) - override def saveTree(tree: DisjointSets[Hash]): Int = // don't replace the postgres DB - { - val no = tree.elements.size - startTableSize // return the number of new elements in the union-find structure - table.close // but don't forget to flush - saveAddresses - no - } - - lazy val addressBuffer: Map[Hash, Hash] = Map() - - def insertAddress(s: (Hash, Hash)) = - { - addressBuffer += s + // a map of new reps to the old reps of these addresses, or the addresses themselves as their new rep if they weren't here before + lazy val changedReps = for { (rep,ads) <- touchedReps } + yield (rep, ads.map(a => oldReps.getOrElse(a,a))) - if (addressBuffer.size >= populateTransactionSize) - saveAddresses + lazy val toSave = for {(rep,ads) <- touchedReps.toSeq + ad <- ads + if ! oldReps.contains(ad) } + yield (ad,rep) - def saveAddresses = { + lazy val addedAds = toSave.size - log.info("Inserting Addresses into SQL database ...") + lazy val addedReps = touchedReps.count { // the number of reps that were not reps before + case (rep,_) => ! oldReps.exists{case (ad,oldR) => rep==oldR} + } - val convertedVector = addressBuffer map (p => (hashToArray(p._1), hashToArray(p._2))) + lazy val deletedReps = changedReps.values.flatten.toSet.filter(p => !changedReps.contains(p) && oldReps.values.toSet.contains(p)).size - try{ - DB withSession (addresses.insertAll(convertedVector.toSeq:_*)(_)) - } - catch { - case e: java.sql.BatchUpdateException => throw e.getNextException - } + lazy val addedClosures = addedReps - deletedReps - addedAds += addressBuffer.size + override def saveTree: Int = // don't replace the postgres DB + { + val no = finalTree.elements.size - startTableSize // return the number of new elements in the union-find structure - addressBuffer.clear + updateAddressDB(changedReps) + saveAddresses(toSave) - log.info("Data inserted") + table.close // but don't forget to flush + no } + } diff --git a/src/main/scala/bge/core/AddressClosure.scala b/src/main/scala/bge/core/AddressClosure.scala index a1a7845..82e14b1 100644 --- a/src/main/scala/bge/core/AddressClosure.scala +++ b/src/main/scala/bge/core/AddressClosure.scala @@ -15,45 +15,57 @@ import java.lang.System abstract class AddressClosure(blockHeights: Vector[Int]) extends db.BitcoinDB { lazy val unionFindTable: Map[Hash,(Int,Hash)] = Map.empty - var addedAds = 0 - var addedReps = 0 - def saveTree(tree: DisjointSets[Hash]): Int + // def unions(blocks: Vector[Int]): Iterable[Seq[Hash]] = { + // val txAndAddressList = txListQuery(blocks) + // val addressesPerTxMap = txAndAddressList.groupBy(p=>Hash(p._1)) + // val hashList = addressesPerTxMap.values map (_ map (p=>Hash(p._2))) + // hashList // filter (_.length > 1) + // } - def generateTree: DisjointSets[Hash] = + // lazy val streamOfUnionLists: Stream[Iterable[Seq[Hash]]] = (0 until blockHeights.length by closureReadSize).toStream map startIndexToUnions + + // def startIndexToUnions(startIndex: Int): Iterable[Seq[Hash]] = { + // val blocks = blockHeights.slice(startIndex,startIndex+closureReadSize) + // val blockNo = blocks.head + // log.info("Closure working on " + blocks.length + " blocks from " + blockNo) + // unions(blocks) + // } + + lazy val generatedTree: DisjointSets[Hash] = { - // TODO: Same problem as with BlockReader, there we used a function "pre" to initialize the values. - addedAds = 0; - addedReps = 0; - def addBlocks(startIndex: Int, tree: DisjointSets[Hash]): DisjointSets[Hash] = { + def addBlocks(tree: DisjointSets[Hash], startIndex: Int): DisjointSets[Hash] = { val blocks = blockHeights.slice(startIndex,startIndex+closureReadSize) - for (blockNo <- blocks.headOption) - log.info("reading " + blocks.length + " blocks from " + blockNo) + val blockNo = blocks.head val txAndAddressList = txListQuery(blocks) val addressesPerTxMap = txAndAddressList.groupBy(p=>Hash(p._1)) val hashList = addressesPerTxMap.values map (_ map (p=>Hash(p._2))) - val nontrivials = hashList filter (_.length > 1) - - log.info("folding and merging " + nontrivials.size) - nontrivials.foldLeft (tree) ((t,l) => insertInputsIntoTree(l,t)) + // val nonTrivials = hashList filter (_.length > 1) // premature optimization? + val result = hashList.foldLeft (tree) ((t,l) => insertInputsIntoTree(l,t)) + log.info("Closured " + blocks.length + " blocks from " + blockNo) + result } - val result = (0 until blockHeights.length by closureReadSize).foldRight(new DisjointSets[Hash](unionFindTable))(addBlocks) - log.info("finished generation") - result + (0 until blockHeights.length by closureReadSize).foldLeft(new DisjointSets[Hash](unionFindTable))(addBlocks) } + def saveTree: Int + + // lazy val generatedTree: DisjointSets[Hash] = + // streamOfUnionLists.foldRight(new DisjointSets[Hash](unionFindTable)){ + // case (unions,tree) => + // unions.foldRight(tree)(insertInputsIntoTree) + // } + def insertInputsIntoTree(addresses: Iterable[Hash], tree: DisjointSets[Hash]): DisjointSets[Hash] = { val addedTree = addresses.foldLeft(tree)((t:DisjointSets[Hash],a:Hash) => t add a) addedTree.union(addresses) } - log.info("applying closure ") val timeStart = System.currentTimeMillis val startTableSize = unionFindTable.size - val countSave = saveTree(generateTree) - + val countSave = saveTree val totalTime = System.currentTimeMillis - timeStart log.info("Total of %s addresses added to closures in %s s, %s µs per address" format diff --git a/src/main/scala/bge/core/BitcoinDRawFileBlockSource.scala b/src/main/scala/bge/core/BitcoinDRawFileBlockSource.scala deleted file mode 100644 index 281c103..0000000 --- a/src/main/scala/bge/core/BitcoinDRawFileBlockSource.scala +++ /dev/null @@ -1,27 +0,0 @@ -package core - -import org.bitcoinj.core._ -import util._ - -import scala.collection.convert.WrapAsScala._ - -// In java that should be implements libs.BlockSource -trait BitcoinDRawFileBlockSource extends BlockSource -{ - override def blockSource: Iterator[(Block,Int)] = { - - startBitcoinJ - log.info("starting") - val almostCurrentChain = getCurrentLongestChainFromBlockCount - val blockMap = almostCurrentChain toMap - - for { - block <- asScalaIterator(loader) - hash = block.getHash - if blockMap.contains(hash) - } - yield - (block, blockMap(hash)) - } - -} diff --git a/src/main/scala/bge/core/BlockReader.scala b/src/main/scala/bge/core/BlockReader.scala index a7c23e0..b05963f 100644 --- a/src/main/scala/bge/core/BlockReader.scala +++ b/src/main/scala/bge/core/BlockReader.scala @@ -44,27 +44,22 @@ trait BlockReader extends BlockSource { } def process: Unit = { - for ((block, height) <- filteredBlockSource) - { - for (transaction <- transactionsInBlock(block)) { - saveTransaction(transaction, height) - transactionCounter +=1 + for ((block, height) <- filteredBlockSource) { + for (transaction <- transactionsInBlock(block)) { + saveTransaction(transaction, height) + transactionCounter +=1 + } + val blockHash = Hash(block.getHash.getBytes) + finishBlock(blockHash, block.getTransactions.size,getTxValue(block),block.getTimeSeconds,height) } - val blockHash = Hash(block.getHash.getBytes) - finishBlock(blockHash, block.getTransactions.size,getTxValue(block),block.getTimeSeconds,height) - } } def blockFilter(b: Block) = { val blockHash = Hash(b.getHash.getBytes) - !(savedBlockSet contains blockHash) + val exists = (savedBlockSet contains blockHash) + !exists } - // def withoutDuplicates(b: Block, t: Transaction): Boolean = - // !(Hash(b.getHash.getBytes) == Hash("00000000000a4d0a398161ffc163c503763b1f4360639393e0e4c8e300e0caec") && - // Hash(t.getHash.getBytes) == Hash("d5d27987d2a3dfc724e359870c6644b40e497bdc0589a033220fe15429d88599")) && the other transaction - // this isn't necessary anymore, because we simply update the LMDB with the same tx. So technically, we are missing those two duplicate tx in our utxo DB - lazy val filteredBlockSource = blockSource withFilter (p=>blockFilter(p._1)) diff --git a/src/main/scala/bge/core/BlockSource.scala b/src/main/scala/bge/core/BlockSource.scala index a42c738..6c323b1 100644 --- a/src/main/scala/bge/core/BlockSource.scala +++ b/src/main/scala/bge/core/BlockSource.scala @@ -23,7 +23,6 @@ trait BlockSource extends db.BitcoinDB { previousElement }._1 - val hashes = blockList map (_._1) assert(hashes.length == hashes.distinct.length, "duplicate block hashes in blockChain (bitcoinJ/levelDB problems?)") diff --git a/src/main/scala/bge/core/FastBlockReader.scala b/src/main/scala/bge/core/FastBlockReader.scala index 691af23..dafc45b 100644 --- a/src/main/scala/bge/core/FastBlockReader.scala +++ b/src/main/scala/bge/core/FastBlockReader.scala @@ -42,7 +42,7 @@ trait FastBlockReader extends BlockReader { for (output <- outputsInTransaction(trans)) { val addressOption: Option[Hash] = getAddressFromOutput(output: TransactionOutput) match { case Some(value) => Some(Hash(value)) - case None => None + case None => None } val value = output.getValue.value @@ -73,7 +73,7 @@ trait FastBlockReader extends BlockReader { def finishBlock(b: Hash, txs: Int, btcs: Long, tstamp: Long, height:Int) = { processedBlocks :+= height insertBlock(b, height, txs, btcs, tstamp) - log.info("Saved block " + height + " consisting of " + txs + " txs") + log.info(s"Saved block $height consisting of $txs txs") } def pre = { @@ -92,17 +92,18 @@ trait FastBlockReader extends BlockReader { def saveUnmatchedInputs: Unit = { - assert(outOfOrderInputMap.size == 0, "unmatched Inputs: " + outOfOrderInputMap.toList) - //for (((outpointTransactionHash, outpointIndex), transactionHash) <- outOfOrderInputMap) - // insertInsertIntoList(Some(transactionHash), Some(outpointTransactionHash), None, Some(outpointIndex), None, None) + val unmatchedCount = outOfOrderInputMap.size + for (input <- outOfOrderInputMap){ + log.error(input.toString) + } + assert(unmatchedCount == 0, unmatchedCount + " unmatched Inputs found") } def saveDataToDB: Unit = { val amount = vectorBlocks.length + vectorMovements.length - log.info("Saving blocks/movements (" + amount + ") into database ...") - val convertedVectorBlocks = vectorBlocks map { case (a,b,c,d,e) => (a.array.toArray,b,c,d,e) } + DB.withSession(blockDB.insertAll(convertedVectorBlocks:_*)(_)) def ohc(e:Option[Hash]):Array[Byte] = e.getOrElse(Hash.zero(0)).array.toArray @@ -118,11 +119,8 @@ trait FastBlockReader extends BlockReader { throw e.getNextException } - vectorMovements = Vector() vectorBlocks = Vector() - - //log.info("Data inserted") } // block @@ -177,6 +175,4 @@ trait FastBlockReader extends BlockReader { def removeUTXO(outpointTransactionHash: util.Hash, outpointIndex: Int): UTXOs = { outputMap -= (outpointTransactionHash -> outpointIndex) } - - } diff --git a/src/main/scala/bge/core/PeerSource.scala b/src/main/scala/bge/core/PeerSource.scala index e24c358..c6c126b 100644 --- a/src/main/scala/bge/core/PeerSource.scala +++ b/src/main/scala/bge/core/PeerSource.scala @@ -4,20 +4,20 @@ package core import util._ trait PeerSource extends BlockSource { - - lazy val truncated = getCurrentLongestChainFromBlockCount take 100 // take 100 so the changes don't get too big for memory - + + lazy val truncated = getCurrentLongestChainFromBlockCount take resumeBlockSize + override def blockSource = { val peer = peerGroup.getConnectedPeers().get(0); for ((_,end) <- truncated.lastOption) - log.info("reading blocks from " + blockCount + " to " + end) + log.info("Reading blocks from " + blockCount + " to " + end) for ((blockHash,no) <- truncated.toIterator) yield { val future = peer.getBlock(blockHash) - log.info("Waiting for node to send us the requested block: " + blockHash) +// log.info("Waiting for node to send us the requested block: " + blockHash) val block = future.get() - log.info("Block received") +// log.info("Block received") (block,no) } diff --git a/src/main/scala/db/BitcoinDB.scala b/src/main/scala/db/BitcoinDB.scala index e3e070d..6a64c5a 100644 --- a/src/main/scala/db/BitcoinDB.scala +++ b/src/main/scala/db/BitcoinDB.scala @@ -2,11 +2,10 @@ package db import slick.driver.PostgresDriver.simple._ -import slick.jdbc.{StaticQuery => Q} -import slick.jdbc.meta.MTable - +import scala.slick.jdbc.{StaticQuery => Q} +import scala.slick.jdbc.meta.MTable import util._ -import collection.mutable.Map +import Hash._ trait BitcoinDB { def blockDB = TableQuery[Blocks] @@ -45,7 +44,12 @@ trait BitcoinDB { } def blockCount: Int = DB withSession { implicit session => - blockDB.length.run + try { + blockDB.length.run + } + catch { case e: Exception => + 0 + } } def existsOutput(transactionHash: Hash, index: Int): Boolean = @@ -59,7 +63,7 @@ trait BitcoinDB { def getLastBlock = DB withSession { implicit session => blockDB.sortBy(_.block_height.desc).map(a => (a.hash, a.block_height)).take(1).run.head } def txListQuery(blocks: Seq[Int]) = { - val emptyArray = Hash.zero(0).array.toArray + val emptyArray = hashToArray(zero(0)) DB withSession { implicit session => val q = for (q <- movements.filter(_.height_out inSet blocks).filter(_.address =!= emptyArray)) yield (q.spent_in_transaction_hash, q.address) @@ -100,17 +104,17 @@ trait BitcoinDB { } } + def createBalanceTables = { var clock = System.currentTimeMillis - DB withSession { implicit session => - log.info("Creating balances") + DB withTransaction { implicit session => + deleteIfExists(balances, closureBalances) balances.ddl.create closureBalances.ddl.create // '\x' is not an address Q.updateNA("""insert into balances select address, sum(value) as balance from utxo where address != '\x' group by address;""").execute - (Q.u + "create index addresses_balance on balances(address)").execute (Q.u + "create index balance on balances(balance)").execute Q.updateNA(""" @@ -118,12 +122,18 @@ trait BitcoinDB { select address, sum(balance) from ((select a.representant as address, b.balance as balance from balances b, addresses a where b.address = a.hash and b.address != '\x') UNION ALL - (select a.address as address, a.balance as balance from balances a left outer join addresses b on a.address = b.hash where a.address != '\x' and b.representant is null)) table2 - group by address; - """).execute + (select a.address as address, a.balance as balance from balances a left outer join addresses b on a.address = b.hash where a.address != '\x' and b.representant is null)) table2 group by address; + """).execute // TODO: this second part can be left out when the addresses table contains all addresses + + (Q.u + "create index balance_2 on closure_balances(balance)").execute - (Q.u + "create index addresses_balance_2 on closure_balances(address)").execute - (Q.u + "create index balance_2 on closure_balances(balance)").execute + Q.updateNA("""delete from balances where balance = 0;""").execute // premature optimization + Q.updateNA("""delete from closure_balances where balance = 0;""").execute + + val b3 = balances.map(_.balance).sum.run.getOrElse(0) + val b4 = closureBalances.map(_.balance).sum.run.getOrElse(0) + + assert(b3 == b4, s"""Wrong balances after fresh create!""") log.info("Balances created in %s s" format (System.currentTimeMillis - clock) / 1000) } @@ -137,90 +147,128 @@ trait BitcoinDB { (values.size, values.sum) } } + + def syncAddressesWithUTXOs = DB withSession { implicit session => + // Add addresses that have not yet been spent + for { + (u,i) <- utxo.map(_.address).run.toVector zipWithIndex } { + try { + if (i%1000 == 0) + log.debug(s"Inserting $i ...") + addresses.insert((u,u)) + } + catch { + case e: Throwable => + } + } + } - def updateBalanceTables(changedAddresses: Map[Hash, Long], changedReps: Map[Hash, Set[Hash]]) = { - var clock = System.currentTimeMillis - log.info("Updating balances ...") - currentStat.total_bitcoins_in_addresses += changedAddresses.map { _._2 }.sum + def getSumBalance: Long = DB withSession { implicit session => + balances.map(_.balance).sum.run.getOrElse(0) + } - DB withSession { implicit session => + def getSumWalletsBalance: Long = DB withSession { implicit session => + closureBalances.map(_.balance).sum.run.getOrElse(0) + } - def updateAdsBalancesTable[A <: Table[(Array[Byte], Long)] with AddressField](adsAndBalances: Map[Hash, Long], balances: TableQuery[A]): Unit = { - for { - (address, balance) <- adsAndBalances - addressArray = Hash.hashToArray(address) - } - if (balance != 0L) - balances.insertOrUpdate(addressArray, balance) - else - balances.filter(_.address === addressArray).delete - } - // Hash.zero(0) is not an address - session.withTransaction { - val adsAndBalances = for ((address, change) <- changedAddresses - Hash.zero(0)) - yield (address, - balances.filter(_.address === Hash.hashToArray(address)). - map(_.balance).firstOption.getOrElse(0L) + change) - - updateAdsBalancesTable(adsAndBalances, balances) - - val table = LmdbMap.open("closures") - val unionFindTable = new ClosureMap(table) - val closures = new DisjointSets[Hash](unionFindTable) - - val repsAndChanges: collection.mutable.Map[Hash, Long] = collection.mutable.Map() - - for ((address, change) <- changedAddresses - Hash.zero(0) ) { - val rep = closures.find(address)._1.getOrElse(address) - val newBalance = repsAndChanges.getOrElse(rep, 0L) + change - repsAndChanges += (rep -> newBalance) - } - - // don't forget the new reps that had no balance change! - for ((rep,_) <- changedReps) - repsAndChanges += (rep -> repsAndChanges.getOrElse(rep, 0L)) - - val repsAndBalances: Map[Hash, Long] = - for { - (rep, change) <- repsAndChanges - oldReps = (changedReps.getOrElse(rep, Set()) + rep).map(Hash.hashToArray(_)) - } yield (rep, - closureBalances.filter(_.address inSetBind oldReps). - map(_.balance).sum.run.getOrElse(0L) + change) - - updateAdsBalancesTable(repsAndBalances, closureBalances) - - // delete all oldReps that have been unified into new ones - val toDelete = changedReps.values.fold(Set())((a, b) => a ++ b).map(Hash.hashToArray(_)) + def getUTXOSBalance(): Long = DB withSession { implicit session => + utxo.map(_.value).sum.run.getOrElse(0L) + } + + def saveBalances(adsAndBalances: Map[Hash, Long], repsAndBalances: Map[Hash, Long], oldReps: Iterable[Hash]): Unit = { + DB withTransaction { implicit session => + // delete merged wallets + val toDelete = oldReps.map(hashToArray) closureBalances.filter(_.address inSet toDelete).delete + for { + (balances, table) <- Set((adsAndBalances, balances), (repsAndBalances, closureBalances)) + (address, balance) <- balances + } { + if (balance > 0L) // premature optimization + table.insertOrUpdate(hashToArray(address), balance) + else if (balance == 0) + table.filter(_.address === hashToArray(address)).delete + } + } + } - table.close + def getRepresentant(a: Hash): Hash = DB withSession { implicit session => + addresses.filter(_.hash === hashToArray(a)).map(_.representant).run.map(Hash(_)).headOption.getOrElse(a) + } + + def updateAddressDB(changedReps: Map[Hash, Set[Hash]]) = DB withTransaction { implicit session => + for {(rep,oldReps) <- changedReps + if oldReps-rep != Set() // (premature?) optimization + } + { + val updateQuery = for(p <- addresses if p.representant inSet (oldReps-rep).map(hashToArray _)) yield p.representant + updateQuery.update(hashToArray(rep)) // TODO: compile query + } + } + + + def saveAddresses(news: Seq[(Hash, Hash)]) = { + + try{ + DB withSession { + val converted = news.map(p => (hashToArray(p._1),hashToArray(p._2))) + addresses.insertAll(converted:_*)(_)} + } + catch { + case e: java.sql.BatchUpdateException => throw e.getNextException + } + + } + + def updateBalanceTables(changedAddresses: Map[Hash, Long], touchedReps: Map[Hash, Set[Hash]], changedReps: Map[Hash, Set[Hash]]): + (Map[Hash, Long], Map[Hash, Long],Map[Hash, Long]) = { + val clock = System.currentTimeMillis - log.info("%s balances updated in %s s, %s µs per address " - format - (adsAndBalances.size, (System.currentTimeMillis - clock) / 1000, (System.currentTimeMillis - clock) * 1000 / (adsAndBalances.size + 1))) + val adsAndBalances: Map[Hash, Long] = + for ((address, change) <- changedAddresses - zero(0)) + yield (address, getBalance(address) + change) + + lazy val repsAndChanges = for ((rep,ads) <- touchedReps) + yield (rep,ads.toSeq.map(changedAddresses(_)).sum) // the toSeq is necessary, because otherwise equal values get swallowed in the set + + lazy val oldReps = changedReps.values.flatten + lazy val newReps = repsAndChanges.keys + // get old closure balances from DB with a single query + lazy val oldClosureBalances = getClosureBalances(oldReps++newReps) + + def oldSum(s: Set[Hash]): Long = s.toSeq.map(a => oldClosureBalances.getOrElse(a, getBalance(a))).sum + + def total(rep: Hash) = oldSum(changedReps.getOrElse(rep, Set())) + + val repsAndBalances = for ((rep,change) <- repsAndChanges) + yield (rep, total(rep)+change) + + saveBalances(adsAndBalances, repsAndBalances, oldReps) + + log.info("Balances updated in %s s, %s addresses, %s µs per address " + format + ((System.currentTimeMillis - clock) / 1000, adsAndBalances.size, (System.currentTimeMillis - clock) * 1000 / (adsAndBalances.size + 1))) + + (adsAndBalances,repsAndChanges, repsAndBalances) - - } - } } def insertRichestClosures = { - log.info("Calculating richest closure list...") - var startTime = System.currentTimeMillis + DB withSession { implicit session => + val bh = blockCount - 1 val topClosures = closureBalances.sortBy(_.balance.desc).take(richlistSize).run.toVector val topClosuresWithBh = for ((rep, bal) <- topClosures) yield (bh, rep, bal) + richestClosures.insertAll(topClosuresWithBh: _*) - log.info("RichestList calculated in " + (System.currentTimeMillis - startTime) / 1000 + "s") + } } def insertRichestAddresses = { - log.info("Calculating richest address list...") - var startTime = System.currentTimeMillis + DB withSession { implicit session => Q.updateNA(""" insert @@ -236,22 +284,22 @@ trait BitcoinDB { balance desc limit """ + richlistSize + """ ;""").execute - log.info("RichestList calculated in " + (System.currentTimeMillis - startTime) / 1000 + "s") + } } + def insertStatistics = { + val startTime = System.currentTimeMillis val (nonDustAddresses, addressGini) = getGini(balances) val (nonDustClosures, closureGini) = getGini(closureBalances) - log.info("Calculating stats...") - - val startTime = System.currentTimeMillis DB withSession { implicit session => - val query = """ - insert - into stats select + + List("""delete from stats where block_height = (select coalesce(max(block_height),0) from blocks);""", + """ + insert into stats select (select coalesce(max(block_height),0) from blocks), (select coalesce(sum(balance)/100000000,0) from balances), (select coalesce(sum(txs),0) from blocks), @@ -263,37 +311,18 @@ trait BitcoinDB { """ + nonDustClosures + """, """ + closureGini + """, """ + addressGini + """, - """ + (System.currentTimeMillis / 1000).toString + """;""" - - (Q.u + query).execute - log.info("Stats calculated in " + (System.currentTimeMillis - startTime) / 1000 + "s"); + """ + (System.currentTimeMillis / 1000).toString + """;""").foreach( q => (Q.u + q).execute ) + log.info("Stat created in " + (System.currentTimeMillis - startTime) / 1000 + "s"); } } - def updateStatistics(changedReps: Map[Hash, Set[Hash]], addedAds: Int, addedReps: Int) = { + def updateStatistics(addedAds: Int, addedClosures: Int) = { - log.info("Updating stats") val time = System.currentTimeMillis val (nonDustAddresses, addressGini) = getGini(balances) val (nonDustClosures, closureGini) = getGini(closureBalances) - - /* block_height - total_bitcoins_in_addresses XXX - total_transactions YYY - total_addresses XXX - total_closures XXX - total_addresses_with_balance YYY - total_closures_with_balance YYY - total_addresses_no_dust YYY - total_closures_no_dust YYY - gini_address - gini_closure - tstamp */ - - // Right now with update statistics we update the values with XXX in a faster way than reading all the database. We can improve it moving the rest (YYY) to a better position, or even using psql triggers. - // A first approach could be to modify the values direct in ResumeBlockReader, ResumeClosure (balances can be updated whenever a utxo is added or removed) - // Update should only add the ginis and call saveStat + DB withSession { implicit session => val stat = currentStat stat.total_addresses_with_balance = balances.length.run @@ -307,15 +336,14 @@ trait BitcoinDB { stat.total_transactions = blockDB.map(_.txs).filter(_ > 0).sum.run.getOrElse(0).toLong stat.total_bitcoins_in_addresses = balances.map(_.balance).sum.run.getOrElse(0L) / 100000000 stat.total_addresses += addedAds - stat.total_closures += addedReps - changedReps.values.map(_.size).sum + stat.total_closures += addedClosures saveStat(stat) - log.info("Updated in " + (System.currentTimeMillis - time) / 1000 + " seconds") + + log.info("Stat updated in " + (System.currentTimeMillis - time) / 1000 + " seconds") } } def getGini[A <: Table[_] with BalanceField](balanceTable: TableQuery[A]): (Long, Double) = { - log.info("calculating Gini: " + balanceTable) - val time = System.currentTimeMillis val balanceVector = DB withSession { implicit session => balanceTable.map(_.balance).filter(_ > dustLimit).sorted.run.toVector @@ -328,7 +356,7 @@ trait BitcoinDB { val summe = balances.sum val mainSum = balances.zipWithIndex.map(p => p._1 * (p._2 + 1.0) / n).sum val gini: Double = if (n == 0) 0.0 else 2.0 * mainSum / (summe) - (n + 1.0) / n - log.info("gini calculated in " + (System.currentTimeMillis - time) / 1000 + "s") + (n, gini) } @@ -361,15 +389,10 @@ trait BitcoinDB { DB withSession { implicit session => - // MOVEMENTS - - // get outputs from address for ( query <- List( "create index address on movements (address);", """create unique index tx_idx on movements (transaction_hash, "index");""", - // "create index spent_in_transaction_hash2 on movements (spent_in_transaction_hash, address);", - // not needed for closure "create index spent_in on movements(spent_in_transaction_hash)", // for api/Movements "create index height_in on movements (height_in);", // for closure "create index height_out_in on movements (height_out, height_in);", @@ -382,8 +405,7 @@ trait BitcoinDB { Q.updateNA(query).execute log.info("Finished" + query) } - - } + } log.info("Indexes created in %s s" format (System.currentTimeMillis - time) / 1000) @@ -394,10 +416,39 @@ trait BitcoinDB { } def saveStat(stat: Stat) = DB withSession { implicit session => + stats.filter(_.block_height === stat.block_height).delete stats.insert(Stat.unapply(stat).get) } - case class Stat(var block_height: Int, var total_bitcoins_in_addresses: Long, var total_transactions: Long, var total_addresses: Long, var total_closures: Long, var total_addresses_with_balance: Long, var total_closures_with_balance: Long, var total_addresses_no_dust: Long, var total_closures_no_dust: Long, var gini_closure: Double, var gini_address: Double, var tstamp: Long) + case class Stat(var block_height: Int, + var total_bitcoins_in_addresses: Long, + var total_transactions: Long, + var total_addresses: Long, + var total_closures: Long, + var total_addresses_with_balance: Long, + var total_closures_with_balance: Long, + var total_addresses_no_dust: Long, + var total_closures_no_dust: Long, + var gini_closure: Double, + var gini_address: Double, + var tstamp: Long) { + + def toMap() = getCCParams(this) + + override def equals(that: Any): Boolean = { + that match { + case that: Stat => + this.toMap()-"tstamp" == that.toMap()-"tstamp" + case _ => + false + } + } + + override def toString(): String = { + getCCParams(this).toString() + } + + } def lastCompletedHeight: Int = DB withSession { implicit session => stats.map(_.block_height).max.run.getOrElse(0) @@ -405,6 +456,8 @@ trait BitcoinDB { def getUtxosMaxHeight = DB withSession { implicit session => utxo.map(_.block_height).max.run.getOrElse(0) } + // FIXME: rollback do not reconstruct stats, it just delete them and create it again. + // It could be added but it does not hurt to create it again (60 minutes at the moment) def rollBack(blockHeight: Int) = DB withSession { implicit session => log.info("rolling back block " + blockHeight) @@ -412,25 +465,79 @@ trait BitcoinDB { stats.filter(_.block_height === blockHeight).delete richestAddresses.filter(_.block_height === blockHeight).delete richestClosures.filter(_.block_height === blockHeight).delete + val table = LmdbMap.open("utxos") val utxoTable = new UTXOs(table) - val utxoQuery = utxo.filter(_.block_height === blockHeight) for ((tx, idx) <- utxoQuery.map(p => (p.transaction_hash, p.index)).run) utxoTable -= Hash(tx) -> idx utxoQuery.delete - val movementQuery = movements.filter(_.height_out === blockHeight) - val utxoRows = movementQuery.filter(_.height_in =!= blockHeight).map(p => (p.transaction_hash, p.address, p.index, p.value, p.height_in)).run + val movementsToDelete = movements.filter(x => x.height_out === blockHeight || x.height_in === blockHeight) + val utxoRows = movements + .filter(x => x.height_in === blockHeight || x.height_out===blockHeight) + .map(p => (p.transaction_hash, p.address, p.index, p.value, p.height_in)) + .run + for ((tx, ad, idx, v, h) <- utxoRows) utxoTable += ((Hash(tx) -> idx) -> (Hash(ad), v, h)) - utxo.insertAll(utxoRows: _*) - movementQuery.delete + utxo.insertAll(utxoRows: _*) + movementsToDelete.delete blockDB.filter(_.block_height === blockHeight).delete - table.close + } + + def getAddressReps(a: Iterable[Hash]): Map[Hash, Hash] = DB withSession {implicit session => + addresses.filter(_.hash inSetBind (a.map(hashToArray _)) ).map(p => (p.hash,p.representant)).run.map(p=>(Hash(p._1),Hash(p._2))).toMap + } + + def getClosureBalances(a: Iterable[Hash]): Map[Hash, Long] = DB withSession { implicit session => + closureBalances.filter(_.address inSetBind a.map(hashToArray(_))).map(a => (a.address, a.balance)).run.map(p=> (Hash(p._1), p._2)).toMap + } + + def getWalletBalances(a: Set[Hash]): Long = DB withSession { implicit session => + closureBalances.filter(_.address inSet a.map(hashToArray(_))).map(_.balance).sum.run.getOrElse(0L) + } + + def getBalance(a: Hash): Long = DB withSession { implicit session => + balances.filter(_.address === hashToArray(a)).map(_.balance).firstOption.getOrElse(0L) + } + + def getAllBalances(): Map[Hash, Long] = DB withSession { implicit session => + balances.sortBy(_.address).map(p=>(p.address, p.balance)).run.map(p=>(Hash(p._1),p._2)).toMap + } + + def getClosures(): Map[Hash, Hash] = DB withSession { implicit session => + addresses.sortBy(_.hash).map(p=>(p.hash, p.representant)).run.map(p=>(Hash(p._1),Hash(p._2))).toMap + } + + def getEmptyUTXOs(): Iterable[(Hash)] = DB withSession { implicit session => + utxo.filter(_.value === 0L).map(_.address).run.map(Hash(_)) + } + + def getAllUTXOs(): Iterable[(Hash, Int)] = DB withSession { implicit session => + utxo.map(p =>(p.address, p.block_height)).run.map(p => (Hash(p._1), p._2)) + } + + def countClosures(): Int = DB withSession { implicit session => + addresses.groupBy(_.representant).map(p => p._1).length.run + } + + def countAddresses(): Int = DB withSession { implicit session => + addresses.length.run + } + + def getWallet(a: Hash): Set[Hash] = DB withSession { implicit session => + addresses.filter(_.representant === hashToArray(a)).map(_.hash).run.map(Hash(_)).toSet + } + + def getAllWalletBalances(): Map[Hash, Long] = DB withSession { implicit session => + closureBalances.sortBy(_.address).map(p=>(p.address, p.balance)).run.map(p=>(Hash(p._1),p._2)).toMap + } + def deleteLastStats = DB withSession { implicit session => + stats.filter(_.block_height === blockCount - 1).delete } } diff --git a/src/main/scala/util/DisjointSets.scala b/src/main/scala/util/DisjointSets.scala index 94823a4..0746ca2 100644 --- a/src/main/scala/util/DisjointSets.scala +++ b/src/main/scala/util/DisjointSets.scala @@ -13,7 +13,7 @@ class DisjointSets[A](val elements: Map[A,(Int,A)] = Map[A,(Int,A)]()) extends D } def asSet: Set[Set[A]] = { - val setOfMaps = elements.groupBy(_._2._2).values.toSet + val setOfMaps = elements.groupBy(p => onlyFind(p._2._2)).values.toSet val setOfSets = setOfMaps map (_.keys.toSet) setOfSets } diff --git a/src/main/scala/util/LmdbMap.scala b/src/main/scala/util/LmdbMap.scala index 7c7bd75..203700a 100644 --- a/src/main/scala/util/LmdbMap.scala +++ b/src/main/scala/util/LmdbMap.scala @@ -1,6 +1,6 @@ package util -import scala.collection.convert.WrapAsScala._ +import scala.collection.JavaConverters.asScalaIterator import org.fusesource.lmdbjni._ import org.fusesource.lmdbjni.Constants._ import scala.collection.mutable.Map @@ -128,7 +128,6 @@ class LmdbMap(val name: String = java.util.UUID.randomUUID.toString) for (kv <- cache) db.put(tx, kv._1, kv._2) tx.commit - log.info("commit took " + (System.currentTimeMillis - t) + " ms") cache.clear } diff --git a/src/main/scala/util/package.scala b/src/main/scala/util/package.scala index 286361c..cf1b1ef 100755 --- a/src/main/scala/util/package.scala +++ b/src/main/scala/util/package.scala @@ -5,9 +5,8 @@ import com.typesafe.config.ConfigFactory import java.net.InetAddress import org.bitcoinj.core._ -import org.bitcoinj.params.MainNetParams +import org.bitcoinj.params._ import org.bitcoinj.store._ -import org.bitcoinj.utils.BlockFileLoader import scala.collection.mutable.ArrayBuffer import com.typesafe.scalalogging._ import org.slf4j.LoggerFactory; @@ -15,7 +14,7 @@ import org.slf4j.LoggerFactory; package object util { lazy val conf = ConfigFactory.load() - + lazy val networkMode = conf.getString("network") lazy val closureTransactionSize = conf.getInt("closureTransactionSize") lazy val closureReadSize = conf.getInt("closureReadSize") lazy val populateTransactionSize = conf.getInt("populateTransactionSize") @@ -24,29 +23,40 @@ package object util lazy val dustLimit = conf.getLong("dustLimit") lazy val dataDir = conf.getString( "dataDir") lazy val richlistSize = conf.getInt("richlistSize") - + lazy val resumeBlockSize = conf.getInt("resumeBlockSize") + lazy val balanceUpdateLimit = conf.getInt("balanceUpdateLimit") lazy val blockStoreFile = new java.io.File(conf.getString("levelDBFile")) lazy val lockFile = conf.getString("lockFile") + lazy val checkUTXOsSize = conf.getInt("checkUTXOsSize") + lazy val internetAddress = conf.getString("bitcoin.ip") match { + case "localhost" => + InetAddress.getLocalHost() + case e: String => + InetAddress.getByName(e) + } + lazy val maxPopulate = conf.getInt("bitcoin.maxPopulate") - def params = MainNetParams.get + def params = + if (networkMode == "main") + MainNetParams.get + else if (networkMode == "testnet") + TestNet3Params.get + else + RegTestParams.get val log = Logger(LoggerFactory.getLogger("bge")) - - lazy val blockStore = new LevelDBBlockStore(new Context(params), blockStoreFile) //, factory) lazy val chain = new BlockChain(params, blockStore) lazy val peerGroup = new PeerGroup(params, chain) - lazy val loader = { - new BlockFileLoader(params,BlockFileLoader.getReferenceClientBlockFileList) - } - lazy val addr = new PeerAddress(InetAddress.getLocalHost(), params.getPort()); - + lazy val addr = new PeerAddress(params, internetAddress, params.getPort()); + def startBitcoinJ: Unit = { log.info("starting peergroup") peerGroup.start peerGroup.addAddress(addr) - peerGroup.waitForPeers(1).get(); + peerGroup.waitForPeers(1).get() + peerGroup.getDownloadPeer.setDownloadParameters(Long.MaxValue, false) // only download headers peerGroup.downloadBlockChain } @@ -78,6 +88,11 @@ package object util } } - -} + // convert case class to map. + def getCCParams(cc: AnyRef) = + (Map[String, Any]() /: cc.getClass.getDeclaredFields) {(a, f) => + f.setAccessible(true) + a + (f.getName -> f.get(cc)) + } +} diff --git a/src/main/webapp/WEB-INF/templates/layouts/default.jade b/src/main/webapp/WEB-INF/templates/layouts/default.jade deleted file mode 100644 index 96d7bd2..0000000 --- a/src/main/webapp/WEB-INF/templates/layouts/default.jade +++ /dev/null @@ -1,12 +0,0 @@ --@ val title: String --@ val headline: String = title --@ val body: String - -!!! -html - head - title= title - body - #content - h1= headline - != body diff --git a/src/main/webapp/WEB-INF/templates/views/hello-scalate.jade b/src/main/webapp/WEB-INF/templates/views/hello-scalate.jade deleted file mode 100644 index 62ef9d6..0000000 --- a/src/main/webapp/WEB-INF/templates/views/hello-scalate.jade +++ /dev/null @@ -1,4 +0,0 @@ -- attributes("title") = "Scalatra: a tiny, Sinatra-like web framework for Scala" -- attributes("headline") = "Welcome to Scalatra" - -p= "Hello, Scalate!" \ No newline at end of file diff --git a/src/main/webapp/WEB-INF/web.xml b/src/main/webapp/WEB-INF/web.xml deleted file mode 100644 index d8e6377..0000000 --- a/src/main/webapp/WEB-INF/web.xml +++ /dev/null @@ -1,15 +0,0 @@ - - - - - - org.scalatra.servlet.ScalatraListener - - diff --git a/src/test/docker/bitcoin/bitcoin.conf b/src/test/docker/bitcoin/bitcoin.conf new file mode 100644 index 0000000..e5010ba --- /dev/null +++ b/src/test/docker/bitcoin/bitcoin.conf @@ -0,0 +1,3 @@ +rpcuser='username' +rpcpassword='password' +regtest=1 diff --git a/src/test/docker/create.sh b/src/test/docker/create.sh new file mode 100755 index 0000000..66ebc68 --- /dev/null +++ b/src/test/docker/create.sh @@ -0,0 +1,5 @@ +#!/usr/bin/env bash + +CMD="docker exec --user bitcoin docker_bitcoin_1 bitcoin-cli -regtest -rpcuser=foo -rpcpassword=bar -rpcport=18333 " +ADD1=$(${CMD} getnewaddress) +X=$(${CMD} sendtoaddress ${ADD1} $1) diff --git a/src/test/docker/docker-compose.yml b/src/test/docker/docker-compose.yml new file mode 100644 index 0000000..b95f2bf --- /dev/null +++ b/src/test/docker/docker-compose.yml @@ -0,0 +1,48 @@ +version: "3" +services: + bitcoin: + image: ruimarinho/bitcoin-core:0.16.0 + networks: + bge_network: + ipv4_address: 172.25.1.2 + ports: + - "18444:18444" + command: + -printtoconsole + -regtest=1 + -rest + -rpcallowip=173.249.18.12 + -rpcpassword=bar + -rpcport=18333 + -rpcuser=foo + -server + volumes: + - "$DIR/files/bitcoin:/home/bitcoin/.bitcoin" + - "$DIR/docker/bitcoin/bitcoin.conf:/etc/bitcoin/bitcoin.conf" + postgres: + image: postgres:10.1 + networks: + bge_network: + ipv4_address: 172.25.1.3 + expose: + - 5433 + ports: + - "5433:5433" + environment: + POSTGRES_USER: root + POSTGRES_PASSWORD: trivial + POSTGRES_DB: postgres2 + volumes: + - "$DIR/files/psql:/var/lib/postgresql/data" + - "$DIR/docker/postgres/postgres.conf:/etc/postgresql/postgresql.conf" + command: + -c + config_file=/etc/postgresql/postgresql.conf +networks: + bge_network: + driver: bridge + ipam: + driver: default + config: + - + subnet: 172.25.1.0/24 diff --git a/src/test/docker/postgres/postgres.conf b/src/test/docker/postgres/postgres.conf new file mode 100644 index 0000000..2be8474 --- /dev/null +++ b/src/test/docker/postgres/postgres.conf @@ -0,0 +1,531 @@ +# ----------------------------- +# PostgreSQL configuration file +# ----------------------------- +# +# This file consists of lines of the form: +# +# name = value +# +# (The "=" is optional.) Whitespace may be used. Comments are introduced with +# "#" anywhere on a line. The complete list of parameter names and allowed +# values can be found in the PostgreSQL documentation. +# +# The commented-out settings shown in this file represent the default values. +# Re-commenting a setting is NOT sufficient to revert it to the default value; +# you need to reload the server. +# +# This file is read on server startup and when the server receives a SIGHUP +# signal. If you edit the file on a running system, you have to SIGHUP the +# server for the changes to take effect, or use "pg_ctl reload". Some +# parameters, which are marked below, require a server shutdown and restart to +# take effect. +# +# Any parameter can also be given as a command-line option to the server, e.g., +# "postgres -c log_connections=on". Some parameters can be changed at run time +# with the "SET" SQL command. +# +# Memory units: kB = kilobytes Time units: ms = milliseconds +# MB = megabytes s = seconds +# GB = gigabytes min = minutes +# h = hours +# d = days + + +#------------------------------------------------------------------------------ +# FILE LOCATIONS +#------------------------------------------------------------------------------ + +# The default values of these variables are driven from the -D command-line +# option or PGDATA environment variable, represented here as ConfigDir. + +#data_directory = 'ConfigDir' # use data in another directory + # (change requires restart) +#hba_file = 'ConfigDir/pg_hba.conf' # host-based authentication file + # (change requires restart) +#ident_file = 'ConfigDir/pg_ident.conf' # ident configuration file + # (change requires restart) + +# If external_pid_file is not explicitly set, no extra PID file is written. +#external_pid_file = '(none)' # write an extra PID file + # (change requires restart) + + +#------------------------------------------------------------------------------ +# CONNECTIONS AND AUTHENTICATION +#------------------------------------------------------------------------------ + +# - Connection Settings - + +#listen_addresses = 'localhost' # what IP address(es) to listen on; + # comma-separated list of addresses; + # defaults to 'localhost', '*' = all + # (change requires restart) +#port = 5432 # (change requires restart) +#max_connections = 100 # (change requires restart) +# Note: Increasing max_connections costs ~400 bytes of shared memory per +# connection slot, plus lock space (see max_locks_per_transaction). +#superuser_reserved_connections = 3 # (change requires restart) +#unix_socket_directory = '' # (change requires restart) +#unix_socket_group = '' # (change requires restart) +#unix_socket_permissions = 0777 # begin with 0 to use octal notation + # (change requires restart) +#bonjour = off # advertise server via Bonjour + # (change requires restart) +#bonjour_name = '' # defaults to the computer name + # (change requires restart) + +# - Security and Authentication - + +#authentication_timeout = 1min # 1s-600s +#ssl = off # (change requires restart) +#ssl_ciphers = 'ALL:!ADH:!LOW:!EXP:!MD5:@STRENGTH' # allowed SSL ciphers + # (change requires restart) +#ssl_renegotiation_limit = 512MB # amount of data between renegotiations +#password_encryption = on +#db_user_namespace = off + +# Kerberos and GSSAPI +#krb_server_keyfile = '' +#krb_srvname = 'postgres' # (Kerberos only) +#krb_caseins_users = off + +# - TCP Keepalives - +# see "man 7 tcp" for details + +#tcp_keepalives_idle = 0 # TCP_KEEPIDLE, in seconds; + # 0 selects the system default +#tcp_keepalives_interval = 0 # TCP_KEEPINTVL, in seconds; + # 0 selects the system default +#tcp_keepalives_count = 0 # TCP_KEEPCNT; + # 0 selects the system default + + +#------------------------------------------------------------------------------ +# RESOURCE USAGE (except WAL) +#------------------------------------------------------------------------------ + +# - Memory - + +#shared_buffers = 32MB # min 128kB + # (change requires restart) +#temp_buffers = 8MB # min 800kB +#max_prepared_transactions = 0 # zero disables the feature + # (change requires restart) +# Note: Increasing max_prepared_transactions costs ~600 bytes of shared memory +# per transaction slot, plus lock space (see max_locks_per_transaction). +# It is not advisable to set max_prepared_transactions nonzero unless you +# actively intend to use prepared transactions. +#work_mem = 1MB # min 64kB +#maintenance_work_mem = 16MB # min 1MB +#max_stack_depth = 2MB # min 100kB + +# - Kernel Resource Usage - + +#max_files_per_process = 1000 # min 25 + # (change requires restart) +#shared_preload_libraries = '' # (change requires restart) + +# - Cost-Based Vacuum Delay - + +#vacuum_cost_delay = 0ms # 0-100 milliseconds +#vacuum_cost_page_hit = 1 # 0-10000 credits +#vacuum_cost_page_miss = 10 # 0-10000 credits +#vacuum_cost_page_dirty = 20 # 0-10000 credits +#vacuum_cost_limit = 200 # 1-10000 credits + +# - Background Writer - + +#bgwriter_delay = 200ms # 10-10000ms between rounds +#bgwriter_lru_maxpages = 100 # 0-1000 max buffers written/round +#bgwriter_lru_multiplier = 2.0 # 0-10.0 multipler on buffers scanned/round + +# - Asynchronous Behavior - + +#effective_io_concurrency = 1 # 1-1000. 0 disables prefetching + + +#------------------------------------------------------------------------------ +# WRITE AHEAD LOG +#------------------------------------------------------------------------------ + +# - Settings - + +#wal_level = minimal # minimal, archive, or hot_standby + # (change requires restart) +#fsync = on # turns forced synchronization on or off +#synchronous_commit = on # immediate fsync at commit +#wal_sync_method = fsync # the default is the first option + # supported by the operating system: + # open_datasync + # fdatasync (default on Linux) + # fsync + # fsync_writethrough + # open_sync +#full_page_writes = on # recover from partial page writes +#wal_buffers = 64kB # min 32kB + # (change requires restart) +#wal_writer_delay = 200ms # 1-10000 milliseconds + +#commit_delay = 0 # range 0-100000, in microseconds +#commit_siblings = 5 # range 1-1000 + +# - Checkpoints - + +#checkpoint_segments = 3 # in logfile segments, min 1, 16MB each +#checkpoint_timeout = 5min # range 30s-1h +#checkpoint_completion_target = 0.5 # checkpoint target duration, 0.0 - 1.0 +#checkpoint_warning = 30s # 0 disables + +# - Archiving - + +#archive_mode = off # allows archiving to be done + # (change requires restart) +#archive_command = '' # command to use to archive a logfile segment +#archive_timeout = 0 # force a logfile segment switch after this + # number of seconds; 0 disables + +# - Streaming Replication - + +#max_wal_senders = 0 # max number of walsender processes + # (change requires restart) +#wal_sender_delay = 200ms # walsender cycle time, 1-10000 milliseconds +#wal_keep_segments = 0 # in logfile segments, 16MB each; 0 disables +#vacuum_defer_cleanup_age = 0 # number of xacts by which cleanup is delayed + +# - Standby Servers - + +#hot_standby = off # "on" allows queries during recovery + # (change requires restart) +#max_standby_archive_delay = 30s # max delay before canceling queries + # when reading WAL from archive; + # -1 allows indefinite delay +#max_standby_streaming_delay = 30s # max delay before canceling queries + # when reading streaming WAL; + # -1 allows indefinite delay + + +#------------------------------------------------------------------------------ +# QUERY TUNING +#------------------------------------------------------------------------------ + +# - Planner Method Configuration - + +#enable_bitmapscan = on +#enable_hashagg = on +#enable_hashjoin = on +#enable_indexscan = on +#enable_material = on +#enable_mergejoin = on +#enable_nestloop = on +#enable_seqscan = on +#enable_sort = on +#enable_tidscan = on + +# - Planner Cost Constants - + +#seq_page_cost = 1.0 # measured on an arbitrary scale +#random_page_cost = 4.0 # same scale as above +#cpu_tuple_cost = 0.01 # same scale as above +#cpu_index_tuple_cost = 0.005 # same scale as above +#cpu_operator_cost = 0.0025 # same scale as above +#effective_cache_size = 128MB + +# - Genetic Query Optimizer - + +#geqo = on +#geqo_threshold = 12 +#geqo_effort = 5 # range 1-10 +#geqo_pool_size = 0 # selects default based on effort +#geqo_generations = 0 # selects default based on effort +#geqo_selection_bias = 2.0 # range 1.5-2.0 +#geqo_seed = 0.0 # range 0.0-1.0 + +# - Other Planner Options - + +#default_statistics_target = 100 # range 1-10000 +#constraint_exclusion = partition # on, off, or partition +#cursor_tuple_fraction = 0.1 # range 0.0-1.0 +#from_collapse_limit = 8 +#join_collapse_limit = 8 # 1 disables collapsing of explicit + # JOIN clauses + + +#------------------------------------------------------------------------------ +# ERROR REPORTING AND LOGGING +#------------------------------------------------------------------------------ + +# - Where to Log - + +#log_destination = 'stderr' # Valid values are combinations of + # stderr, csvlog, syslog, and eventlog, + # depending on platform. csvlog + # requires logging_collector to be on. + +# This is used when logging to stderr: +#logging_collector = off # Enable capturing of stderr and csvlog + # into log files. Required to be on for + # csvlogs. + # (change requires restart) + +# These are only used if logging_collector is on: +#log_directory = 'pg_log' # directory where log files are written, + # can be absolute or relative to PGDATA +#log_filename = 'postgresql-%Y-%m-%d_%H%M%S.log' # log file name pattern, + # can include strftime() escapes +#log_truncate_on_rotation = off # If on, an existing log file of the + # same name as the new log file will be + # truncated rather than appended to. + # But such truncation only occurs on + # time-driven rotation, not on restarts + # or size-driven rotation. Default is + # off, meaning append to existing files + # in all cases. +#log_rotation_age = 1d # Automatic rotation of logfiles will + # happen after that time. 0 disables. +#log_rotation_size = 10MB # Automatic rotation of logfiles will + # happen after that much log output. + # 0 disables. + +# These are relevant when logging to syslog: +#syslog_facility = 'LOCAL0' +#syslog_ident = 'postgres' + +#silent_mode = off # Run server silently. + # DO NOT USE without syslog or + # logging_collector + # (change requires restart) + + +# - When to Log - + +#client_min_messages = notice # values in order of decreasing detail: + # debug5 + # debug4 + # debug3 + # debug2 + # debug1 + # log + # notice + # warning + # error + +#log_min_messages = warning # values in order of decreasing detail: + # debug5 + # debug4 + # debug3 + # debug2 + # debug1 + # info + # notice + # warning + # error + # log + # fatal + # panic + +#log_min_error_statement = error # values in order of decreasing detail: + # debug5 + # debug4 + # debug3 + # debug2 + # debug1 + # info + # notice + # warning + # error + # log + # fatal + # panic (effectively off) + +#log_min_duration_statement = -1 # -1 is disabled, 0 logs all statements + # and their durations, > 0 logs only + # statements running at least this number + # of milliseconds + + +# - What to Log - + +#debug_print_parse = off +#debug_print_rewritten = off +#debug_print_plan = off +#debug_pretty_print = on +#log_checkpoints = off +#log_connections = off +#log_disconnections = off +#log_duration = off +#log_error_verbosity = default # terse, default, or verbose messages +#log_hostname = off +#log_line_prefix = '' # special values: + # %a = application name + # %u = user name + # %d = database name + # %r = remote host and port + # %h = remote host + # %p = process ID + # %t = timestamp without milliseconds + # %m = timestamp with milliseconds + # %i = command tag + # %e = SQL state + # %c = session ID + # %l = session line number + # %s = session start timestamp + # %v = virtual transaction ID + # %x = transaction ID (0 if none) + # %q = stop here in non-session + # processes + # %% = '%' + # e.g. '<%u%%%d> ' +#log_lock_waits = off # log lock waits >= deadlock_timeout +#log_statement = 'none' # none, ddl, mod, all +#log_temp_files = -1 # log temporary files equal or larger + # than the specified size in kilobytes; + # -1 disables, 0 logs all temp files +#log_timezone = unknown # actually, defaults to TZ environment + # setting + + +#------------------------------------------------------------------------------ +# RUNTIME STATISTICS +#------------------------------------------------------------------------------ + +# - Query/Index Statistics Collector - + +#track_activities = on +#track_counts = on +#track_functions = none # none, pl, all +#track_activity_query_size = 1024 # (change requires restart) +#update_process_title = on +#stats_temp_directory = 'pg_stat_tmp' + + +# - Statistics Monitoring - + +#log_parser_stats = off +#log_planner_stats = off +#log_executor_stats = off +#log_statement_stats = off + + +#------------------------------------------------------------------------------ +# AUTOVACUUM PARAMETERS +#------------------------------------------------------------------------------ + +#autovacuum = on # Enable autovacuum subprocess? 'on' + # requires track_counts to also be on. +#log_autovacuum_min_duration = -1 # -1 disables, 0 logs all actions and + # their durations, > 0 logs only + # actions running at least this number + # of milliseconds. +#autovacuum_max_workers = 3 # max number of autovacuum subprocesses + # (change requires restart) +#autovacuum_naptime = 1min # time between autovacuum runs +#autovacuum_vacuum_threshold = 50 # min number of row updates before + # vacuum +#autovacuum_analyze_threshold = 50 # min number of row updates before + # analyze +#autovacuum_vacuum_scale_factor = 0.2 # fraction of table size before vacuum +#autovacuum_analyze_scale_factor = 0.1 # fraction of table size before analyze +#autovacuum_freeze_max_age = 200000000 # maximum XID age before forced vacuum + # (change requires restart) +#autovacuum_vacuum_cost_delay = 20ms # default vacuum cost delay for + # autovacuum, in milliseconds; + # -1 means use vacuum_cost_delay +#autovacuum_vacuum_cost_limit = -1 # default vacuum cost limit for + # autovacuum, -1 means use + # vacuum_cost_limit + + +#------------------------------------------------------------------------------ +# CLIENT CONNECTION DEFAULTS +#------------------------------------------------------------------------------ + +# - Statement Behavior - + +#search_path = '"$user",public' # schema names +#default_tablespace = '' # a tablespace name, '' uses the default +#temp_tablespaces = '' # a list of tablespace names, '' uses + # only default tablespace +#check_function_bodies = on +#default_transaction_isolation = 'read committed' +#default_transaction_read_only = off +#session_replication_role = 'origin' +#statement_timeout = 0 # in milliseconds, 0 is disabled +#vacuum_freeze_min_age = 50000000 +#vacuum_freeze_table_age = 150000000 +#bytea_output = 'hex' # hex, escape +#xmlbinary = 'base64' +#xmloption = 'content' + +# - Locale and Formatting - + +#datestyle = 'iso, mdy' +#intervalstyle = 'postgres' +#timezone = unknown # actually, defaults to TZ environment + # setting +#timezone_abbreviations = 'Default' # Select the set of available time zone + # abbreviations. Currently, there are + # Default + # Australia + # India + # You can create your own file in + # share/timezonesets/. +#extra_float_digits = 0 # min -15, max 3 +#client_encoding = sql_ascii # actually, defaults to database + # encoding + +# These settings are initialized by initdb, but they can be changed. +#lc_messages = 'C' # locale for system error message + # strings +#lc_monetary = 'C' # locale for monetary formatting +#lc_numeric = 'C' # locale for number formatting +#lc_time = 'C' # locale for time formatting + +# default configuration for text search +#default_text_search_config = 'pg_catalog.simple' + +# - Other Defaults - + +#dynamic_library_path = '$libdir' +#local_preload_libraries = '' + + +#------------------------------------------------------------------------------ +# LOCK MANAGEMENT +#------------------------------------------------------------------------------ + +#deadlock_timeout = 1s +#max_locks_per_transaction = 64 # min 10 + # (change requires restart) +# Note: Each lock table slot uses ~270 bytes of shared memory, and there are +# max_locks_per_transaction * (max_connections + max_prepared_transactions) +# lock table slots. + + +#------------------------------------------------------------------------------ +# VERSION/PLATFORM COMPATIBILITY +#------------------------------------------------------------------------------ + +# - Previous PostgreSQL Versions - + +#array_nulls = on +#backslash_quote = safe_encoding # on, off, or safe_encoding +#default_with_oids = off +#escape_string_warning = on +#lo_compat_privileges = off +#sql_inheritance = on +#standard_conforming_strings = off +#synchronize_seqscans = on + +# - Other Platforms and Clients - + +#transform_null_equals = off + + +#------------------------------------------------------------------------------ +# CUSTOMIZED OPTIONS +#------------------------------------------------------------------------------ + +#custom_variable_classes = '' # list of custom variable class names + +listen_addresses = '*' +port = 5433 diff --git a/src/test/docker/start.sh b/src/test/docker/start.sh new file mode 100755 index 0000000..b80a59f --- /dev/null +++ b/src/test/docker/start.sh @@ -0,0 +1,31 @@ +#!/bin/bash +export VERSION=SNAPSHOT +export REPOSITORY=jorgemartinezpizarro +export DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )/.." && pwd )" + +#mkdir -p "${DIR}/files/psql" +mkdir -p "${DIR}/files/bitcoin" +rm "${DIR}/files/psql/" -rf +rm "${DIR}/files/bitcoin/" -rf +rm "${DIR}/files/blockchain/" -rf +cd "${DIR}/docker" + +docker-compose kill +docker-compose up -d + +echo "Waiting for docker to be ready" + +for i in $(seq 1 1000) +do + docker exec docker_postgres_1 pg_isready -p 5433 > /dev/null 2> /dev/null + code1=$? + docker exec --user bitcoin docker_bitcoin_1 bitcoin-cli -regtest -rpcuser=foo -rpcpassword=bar -rpcport=18333 getnetworkinfo > /dev/null 2> /dev/null + code2=$? + if [ $code1 -eq 0 ] && [ $code2 -eq 0 ] + then + echo "Finished loading" + exit 0 + fi +done +echo "Timeout" +exit 1 diff --git a/src/test/docker/stop.sh b/src/test/docker/stop.sh new file mode 100755 index 0000000..def34f8 --- /dev/null +++ b/src/test/docker/stop.sh @@ -0,0 +1,12 @@ +#!/bin/bash +export VERSION=SNAPSHOT +export REPOSITORY=jorgemartinezpizarro +export DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )/.." && pwd )" +echo $DIR + +mkdir -p "${DIR}/data/psql" +mkdir -p "${DIR}/data/bitcoin" +rm "${DIR}/data/bitcoin/" -rf +rm "${DIR}/data/blockchain/" -rf +cd "${DIR}/docker" +docker-compose kill diff --git a/src/test/docker/total.sh b/src/test/docker/total.sh new file mode 100755 index 0000000..300bc68 --- /dev/null +++ b/src/test/docker/total.sh @@ -0,0 +1,4 @@ +#!/usr/bin/env bash +CMD="docker exec --user bitcoin docker_bitcoin_1 bitcoin-cli -regtest -rpcuser=foo -rpcpassword=bar -rpcport=18333 " +X=$(${CMD} getbalance) +echo $X diff --git a/src/test/files/.gitignore b/src/test/files/.gitignore new file mode 100644 index 0000000..5e7d273 --- /dev/null +++ b/src/test/files/.gitignore @@ -0,0 +1,4 @@ +# Ignore everything in this directory +* +# Except this file +!.gitignore diff --git a/src/test/resources/reference.conf b/src/test/resources/reference.conf new file mode 100644 index 0000000..5d3bf4d --- /dev/null +++ b/src/test/resources/reference.conf @@ -0,0 +1,22 @@ +populateTransactionSize=100000 // movemements +closureReadSize=1000 // blocks +closureTransactionSize=50000 +balanceTransactionSize=50000 +richlistSize=1000 +dataDir = "/root/Bitcoin-Graph-Explorer/src/test/files/blockchain/" +levelDBFile = ${dataDir}"level.db" +lockFile = ${dataDir}"lock" +dustLimit=100000 +databaseName="postgres2" +username = "root" +password = "trivial" +host="localhost:5433" +jdbcOptions = "?useServerPrepStmts=false&rewriteBatchedStatements=true&maxWait=-1" +mdb.path = ${dataDir}"mdb/" +mdb.transactionSize = 100000 +network=regtest # accepts main, testnet, regtest +resumeBlockSize=10000000 # depending on the available ram +bitcoin.ip=localhost +bitcoin.maxPopulate=0 # 0 all,m number block until stop first populate. To rest partial blockchains without running all! +balanceUpdateLimit=70000 +checkUTXOsSize=1000000 diff --git a/src/test/scala/DisjointSetChecks.scala b/src/test/scala/DisjointSetChecks.scala index 56a1a78..09f00aa 100644 --- a/src/test/scala/DisjointSetChecks.scala +++ b/src/test/scala/DisjointSetChecks.scala @@ -13,12 +13,12 @@ object DSSpecification extends Properties("DisjointSets") { def setOfPairsGen = for { s <- arbitrary[Set[Int]] - part <- Gen.choose(1,s.size) + part <- Gen.choose(0,s.size) e <- s } yield (part, e) def setOfPairsToDisjointSet(pairs: Set[(Int, Int)]) : Set[Set[Int]] = - pairs.groupBy(_._1).values.toSet map ((p:Set[(Int, Int)]) => (p map (q => q._2))) + pairs.groupBy(_._1).values.toSet map ((p:Set[(Int, Int)]) => (p map (_._2))) property("Sets of nonempty disjoint Sets can be transformed into union/find structures and back") = forAll(setOfPairsGen) { (pairs: Set[(Int,Int)]) => diff --git a/src/test/scala/ExplorerIntegrationTests.scala b/src/test/scala/ExplorerIntegrationTests.scala new file mode 100755 index 0000000..65b267e --- /dev/null +++ b/src/test/scala/ExplorerIntegrationTests.scala @@ -0,0 +1,143 @@ +import TestExplorer._ +import org.scalatest._ + +//////////////////////////////////////////////////////////////////////////////////////////////// +// +// The functions "safeRollback", "safePopulate" and "safeResume" are just a wrapper for the original +// Explorer functions, but adding an exhaustive analysis of all the generated data (in case of +// updateStatistis - updateBalances). +// +// Definitions can be found in TestExplorer. +// +// After each operation, several data will be checked: +// +// UPDATE +// +// consistency of repsAndBalances +// consistency of repsAndChanges +// consistency of addedReps +// consistency of addedAdds +// consistency of changedAddresses +// consistency of adsAndBalances +// consistency of changedReps +// total balances of wallets against addresses and utxos +// total_closures and total_addresses are the same as with createStatistics +// +// CREATE // ROLLBACK +// +// total balances of wallets against addresses and utxos +// +// NOTE: atm there is a bug leading the balances to be different than the "closureBalances". +// In order to fix the code we need first a test to cover this issue. I could not yet +// reproduce the error. +// +////////////////////////////////////////////////////////////////////////////////////////////// + +class ExplorerIntegrationTests extends FlatSpec with Matchers with CancelGloballyAfterFailure{ + + "docker" should "start bitcoin and postgres" in { + startDocker should be (0) + } + + it should "create 103 blocks and 5 txs" in { + addBlocks(102) should be (0) + pay(8) should be (0) + pay(7) should be (0) + pay(6) should be (0) + pay(5) should be (0) + pay(4) should be (0) + addBlocks(1) should be (0) + } + + it should "read 0 blocks from postgres" in { + bgeCount should be (0) + } + + it should "read 104 blocks from bitcoin" in { + bitcoinCount should be (104) + } + + it should "fail sending 5000 bitcoins" in { + pay(5000) should be (6) + } + + "explorer" should "populate 104 blocks" in { + safePopulate should be (None) + } + + it should "resume a block with 5 txs" in { + pay(19) + pay(17) + pay(8) + pay(14) + pay(100) + addBlocks(1) should be (0) + safeResume should be (None) + } + + it should "resume a block with 1 tx" in { + addTxs(1) should be (0) + addBlocks(1) should be (0) + safeResume should be (None) + } + + it should "resume a block with 7 txs" in { + addTxs(7) should be (0) + addBlocks(1) should be (0) + safeResume should be (None) + } + + it should "rollback once and resume it again" in { + val init = stat + safeRollback() should be (None) + safeResume should be (None) + init should be (stat) + } + + it should "resume 2 mores block with 6 and 0 txs" in { + addTxs(6) should be (0) + addBlocks(2) should be (0) + safeResume should be (None) + } + + it should "rollback again and resume it again" in { + val init = stat + safeRollback() should be (None) + safeResume should be (None) + init should be (stat) + } + + List((50, 2), (1, 90 + )).foreach{ case (blocks: Int, txs: Int) => { + + it should s"resume $blocks blocks with $txs txs each" in { + + (1 to blocks).foreach(i => { + addTxs(txs) should be (0) + addBlocks(1) should be (0) + }) + + safeResume should be (None) + + } + + }} + + it should "rollback 8 blocks and resume it again" in { + val init = stat + safeRollback() should be (None) + safeResume should be (None) + stat should be (init) + } + + it should "resume a block with 8 txs" in { + addTxs(8) should be (0) + addBlocks(1) should be (0) + safeResume should be (None) + } + + it should s"finish" in { + println(s"Created postgres DB with $bgeCount blocks, $totalAddresses addresses and $totalClosures wallets") + } + +} diff --git a/src/test/scala/RawReaderPropertySuite.scala b/src/test/scala/RawReaderPropertySuite.scala deleted file mode 100755 index 0a72f60..0000000 --- a/src/test/scala/RawReaderPropertySuite.scala +++ /dev/null @@ -1,21 +0,0 @@ -import org.scalatest.prop._ -import org.scalatest._ -import core._ -import org.scalacheck.Gen._ -import actions._ - -class RawReaderPropertySuite extends PropSpec with ShouldMatchers with PropertyChecks { - /*databaseFile = "blockchain/test.db" - - property("populater.end should be the minimum of given end and number of blocks available") - { - forAll (choose(0,280000), minSuccessful(1)) { (n:Int) => - { - val populater = new BlocksReader(List(n.toString,"init")) - populater.start should be(0) - populater.end should be (n) // TODO: This is not true. We need the number of blocks available and this is also at least a problem when 0 is given - } - - } - } */ -} \ No newline at end of file diff --git a/src/test/scala/TestExplorer.scala b/src/test/scala/TestExplorer.scala new file mode 100644 index 0000000..e33793a --- /dev/null +++ b/src/test/scala/TestExplorer.scala @@ -0,0 +1,181 @@ +import util.Hash +import Explorer._ +import sys.process._ +import scala.util.Random +import org.scalatest._ + +object TestExplorer extends db.BitcoinDB { + + // CONSTANTS + + val BITCOIN = "docker exec --user bitcoin docker_bitcoin_1 bitcoin-cli -regtest -rpcuser=foo -rpcpassword=bar -rpcport=18333 " + + val SCRIPTS = "/root/Bitcoin-Graph-Explorer/src/test/docker/" + + val RUNS = 5 + + // SOME HELPERS + + private def s: Map[Hash, Long] = Map() + + private def x:Map[Hash, Set[Hash]] = Map() + + private def nr(b: Long): String = ((b/10000)/10000.0).toString + + private def sx = "\n " + + private def sp = " " + + private def pri(a: Hash) = a.toString.drop(2).take(4) + + private def str1(n: Map[Hash, Long]): String = "(" + nr(n.map(_._2).sum) + ")" + sx + n.toSeq.sortBy(p=>pri(p._1)).map(m=>pri(m._1)+": "+(nr(m._2))).mkString(sx) + + private def str2(n: Map[Hash, Hash]): String = "(" + n.size + ")" + sx + n.toSeq.sortBy(p=>pri(p._1)).map(m=>pri(m._1) +" => "+(pri(m._2))).mkString(sx) + + private def str3(n: Map[Hash, Set[Hash]]): String = "(" + n.size + ")" + sx + n.toSeq.sortBy(p=>pri(p._1)).map(m=>pri(m._1) + " => "+ m._2.map((pri(_))).mkString(" - ")).mkString(sx) + + private def str4(n: Map[Hash, String]): String = "(" + n.size + ")" + sx + n.toSeq.sortBy(p=>pri(p._1)).map(m=>pri(m._1)+": "+m._2).mkString(sx) + + // BITCOIN-CLI functions + + def gen(i:Int): Int = (BITCOIN + "generate" + " " + i.toString) ! ProcessLogger(_ => ()) + + def addBlocks(n: Int): Int = (1 to n).map(_ => gen(1)).sum + + def startDocker = (SCRIPTS + "start.sh") ! ProcessLogger(_ => ()) + + def stopDocker = (SCRIPTS) + "stop.sh" ! ProcessLogger(_ => ()) + + def pay(n: Long) = (SCRIPTS + "create.sh" + " " + n.toString) ! ProcessLogger(_ => ()) + + def addTxs(n: Int) = (1 to n).map(_ => pay(Random.nextInt(49)+50)).sum + + def bitcoinCount: Int = ((BITCOIN + "getblockcount") !!).trim.toInt+1 + + def bitcoinTotal: Double = ((BITCOIN + "getbalance") !!).trim.toDouble * 100000000 + + def bgeTotal: Long = getSumBalance + + // EXPLORER functions + + def bgeCount = blockCount + + def totalAddresses: Int = countAddresses + + def totalClosures: Int = countClosures + + def stat = currentStat + + def safeRollback() = { + + deleteLastStats + // rollback do not reconstruct stats, it just delete them and create it again + rollBackToLastStatIfNecessary + populateStats + assertBalancesCreatedCorrectly + } + + def safeResume = { + + resume match { + case Some((a,b,c,d,e,y,f,x)) => + assertBalancesUpdatedCorrectly(a,b,c,d,e,y,f,x) + case None => + assertBalancesCreatedCorrectly + } + } + + def safePopulate = { + populate + assertBalancesCreatedCorrectly + } + + def assertBalancesCreatedCorrectly = + assertBalancesUpdatedCorrectly(s,s,s,s,x,x,0,0) + + def assertBalancesUpdatedCorrectly( + adsAndBalances: Map[Hash, Long], + repsAndChanges: Map[Hash, Long], + changedAddresses: Map[Hash, Long], + repsAndBalances: Map[Hash, Long], + touchedReps: Map[Hash, Set[Hash]], + changedReps: Map[Hash, Set[Hash]], + addedAdds: Int, + addedReps: Int): Option[String] = { + + // values to check + lazy val ca = changedAddresses - Hash.zero(0) + lazy val s1 = ca.map(_._2).sum + lazy val s2 = repsAndChanges.map(_._2).sum + lazy val b3 = getSumBalance + lazy val b4 = getSumWalletsBalance + lazy val ut = getUTXOSBalance() + lazy val x1 = getAllWalletBalances//.filter(_._2>0) + lazy val x2 = getAllBalances.groupBy(p => getRepresentant(p._1)).map(p => (p._1, p._2.map(_._2).sum))//.filter(_._2>0) + lazy val allClosures = getClosures.groupBy(_._2).map(p => (p._1, p._2.keySet)) + lazy val emptyUtxos = getEmptyUTXOs() + + // strings to print + lazy val wrongClosures = for {k <- x1.keys; if None == x2.get(k) || x2(k) != x1(k)} yield (k, x1(k)) + lazy val wrongWalletTableString = /*s""" + ${sx}TOUCHED REPS${str3(touchedReps)} + ${sx}WRONG reps${str1(wrongClosures.toMap)} + ${sx}CHANGED ADDs${str1(ca)} + ${sx}CHANGED REPRESENTANTS ${str3(changedReps)} + ${sx}REPS AND BALANCES${str1(repsAndBalances)} + ${sx}COUNT WALLETS CREATE, UPDATE ${countClosures()} ${currentStat.total_closures} + ${sx}WALLETS ${str3(allClosures)} + ${sx}EMPTY UTXOS ${sx}(${emptyUtxos.size})${sx}${emptyUtxos.map(pri(_))} + */ + s""" + ${sx}BALANCES ADD - REP - UTXOs${sx}${nr(b3)} ${nr(b4)} ${nr(ut)} + ${sx}Stats ${sx}${currentStat} + ${sx}UTXOs ${sx}${getAllUTXOs.filter(_._2 == currentStat.block_height).map(p=>pri(p._1)+": "+p._2)} +""" + + lazy val countsDiffBitcoinToBge = bitcoinTotal - b3 + + // test updateStatistics && test updateBalances + lazy val errorOption = + if (false && emptyUtxos.size > 0) + Some(s"___EMPTY UTXOs___$wrongWalletTableString") + else if (b3 != b4 || !wrongClosures.isEmpty) + Some(s"___WRONG_WALLET_TABLE___$wrongWalletTableString") + else if (x1.exists(r => r._1 != getRepresentant(r._1))) + Some(s"___WRONG_CLOSURE_IN_TABLE___$wrongWalletTableString") + else if (s1 != s2) + Some(s"___WRONG_CHANGED_VALUES___$wrongWalletTableString") + else if (ut != b3) + Some(s"___WRONG_ADDRESS_TABLE ___$wrongWalletTableString") + else if (countAddresses != currentStat.total_addresses) + Some(s"___WRONG_TOTAL_ADDRESSES___$wrongWalletTableString") + else if (countClosures != currentStat.total_closures) + Some(s"___WRONG_TOTAL_WALLETS___$wrongWalletTableString") + else + None + + errorOption + } +} + +object CancelGloballyAfterFailure { + @volatile var cancelRemaining = false +} + +trait CancelGloballyAfterFailure extends TestSuiteMixin { this: TestSuite => + import CancelGloballyAfterFailure._ + + abstract override def withFixture(test: NoArgTest): Outcome = { + if (cancelRemaining) + Canceled("Canceled by CancelGloballyAfterFailure because a test failed previously") + else + super.withFixture(test) match { + case failed: Failed => + cancelRemaining = true + failed + case outcome => outcome + } + } + + final def newInstance: Suite with OneInstancePerTest = throw new UnsupportedOperationException +}