From 97ffcbc8b6b5762bb9a3269bf9c74f4244da45bc Mon Sep 17 00:00:00 2001 From: Marius Hanne Date: Wed, 11 Feb 2015 04:33:51 +0100 Subject: [PATCH 1/3] mempool integration --- Gemfile | 2 +- bin/bitcoin_node | 5 +++++ lib/bitcoin/node.rb | 16 +++++++++++++++- lib/bitcoin/node/command_handler.rb | 23 ++++++++++++++++++++--- lib/bitcoin/node/connection_handler.rb | 12 +++++++++--- spec/node/command_api_spec.rb | 13 ++++++------- 6 files changed, 56 insertions(+), 15 deletions(-) diff --git a/Gemfile b/Gemfile index 10cf221..22bf56d 100644 --- a/Gemfile +++ b/Gemfile @@ -7,7 +7,7 @@ gem 'bitcoin-ruby', # path: '../bitcoin-ruby' gem 'bitcoin-ruby-blockchain', - git: "git://github.com/mhanne/bitcoin-ruby-blockchain" + git: "git://github.com/mhanne/bitcoin-ruby-blockchain", branch: "mempool" # path: '../bitcoin-ruby-blockchain' gem 'bitcoin-ruby-wallet', diff --git a/bin/bitcoin_node b/bin/bitcoin_node index d4ff8c2..ff9efdc 100755 --- a/bin/bitcoin_node +++ b/bin/bitcoin_node @@ -56,6 +56,11 @@ optparse = OptionParser.new do |opts| options[:storage] = storage end + opts.on("-m", "--mempool [DB]", + "Use given mempool db (default: #{options[:mempool]})") do |mempool| + options[:mempool][:db] = mempool + end + opts.on("--skip-validation", "Skip validation of blocks and transactions") do options[:skip_validation] = true end diff --git a/lib/bitcoin/node.rb b/lib/bitcoin/node.rb index 2614075..6b19e2b 100644 --- a/lib/bitcoin/node.rb +++ b/lib/bitcoin/node.rb @@ -102,6 +102,12 @@ class Node skip_validation: false, check_blocks: 1000, connection_timeout: 10, + mempool: { + db: "sqlite:/", + max_age: 3600, + log_level: :debug, + }, + } def initialize config = {} @@ -127,7 +133,8 @@ def set_store skip_validation: @config[:skip_validation], index_nhash: @config[:index_nhash], index_p2sh_type: @config[:index_p2sh_type], - log_level: @config[:log][:storage] + log_level: @config[:log][:storage], + mempool: @config[:mempool], }) @store.log.level = @config[:log][:storage] @store.check_consistency(@config[:check_blocks]) @@ -301,6 +308,10 @@ def run end end + @store.mempool.subscribe(:accepted) do |data| + tx = Bitcoin::P::Tx.new(data[:payload].htb) + push_notification :tx, [tx, 0] + end end end @@ -440,6 +451,7 @@ def work_queue exit end else + @store.mempool.add(obj[1]) if @store.in_sync? drop = @unconfirmed.size - @config[:max][:unconfirmed] + 1 drop.times { @unconfirmed.shift } if drop > 0 unless @unconfirmed[obj[1].hash] @@ -467,7 +479,9 @@ def work_inv_queue return if @queue.size >= @config[:max][:queue] while inv = @inv_queue.shift next if !@store.in_sync? && inv[0] == :tx && @notifiers.empty? + @store.mempool.inv(inv[1]) if inv[0] == :tx && @store.in_sync? next if @queue.map{|i|i[1]}.map(&:hash).include?(inv[1]) + next if @store.in_sync? && @store.mempool.exists?(inv[1]) inv[2].send("send_getdata_#{inv[0]}", inv[1]) end end diff --git a/lib/bitcoin/node/command_handler.rb b/lib/bitcoin/node/command_handler.rb index 73db9f2..ff4e4d1 100644 --- a/lib/bitcoin/node/command_handler.rb +++ b/lib/bitcoin/node/command_handler.rb @@ -108,6 +108,15 @@ def handle_unmonitor request { id: id } end + [:accepted, :rejected, :doublespend, :seen, :confirmed].each do |type| + define_method("handle_monitor_mempool_#{type}") do |request, params| + monitor_id = @monitors.size + id = @node.store.mempool.subscribe(type) do |data| + respond(request, data) + end + end + end + # Handle +monitor block+ command; def handle_monitor_block request, params monitor_id = @monitors.size @@ -127,7 +136,6 @@ def respond_missed_blocks request, monitor_id end def respond_monitor_block request, block, height = nil - p :mbb height ||= block.height respond(request, { hash: block.hash, hex: block.to_payload.hth, height: height, depth: height }) end @@ -283,6 +291,12 @@ def handle_info peers: (blocks.inject{|a,b| a+=b; a } / blocks.size rescue '?' ), sync: @node.store.in_sync?, }, + mempool: { + total: @node.store.mempool.transactions.count, + accepted: @node.store.mempool.accepted.count, + rejected: @node.store.mempool.rejected.count, + doublespends: @node.store.mempool.doublespend.count, + }, addrs: { alive: @node.addrs.select{|a| a.alive?}.size, total: @node.addrs.size, @@ -491,6 +505,7 @@ def handle_relay_tx request, params = {} details: validator.error }) end + @node.store.mempool.add(tx) #@node.store.store_tx(tx) @node.relay_tx[tx.hash] = tx @node.relay_propagation[tx.hash] = 0 @@ -533,8 +548,10 @@ def handle_store_block params # { method: "store_tx", params: { hex: } } def handle_store_tx params tx = Bitcoin::P::Tx.new(params[:hex].htb) - @node.queue << [:tx, tx] - { queued: tx.hash } +# @node.queue << [:tx, tx] +# { queued: tx.hash } + @node.store.mempool.add(tx, params[:skip_validation]) + { added: tx.hash } end # # format node uptime diff --git a/lib/bitcoin/node/connection_handler.rb b/lib/bitcoin/node/connection_handler.rb index 0de9207..cc6702c 100644 --- a/lib/bitcoin/node/connection_handler.rb +++ b/lib/bitcoin/node/connection_handler.rb @@ -109,6 +109,9 @@ def complete_handshake @node.push_notification(:connection, info.merge(type: :connected)) @node.addrs << addr send_data P::Addr.pkt(@node.addr) if @node.config[:announce] + # send_data Bitcoin::P.pkt("mempool", "") + send_getblocks + end # error parsing a message, log as warning but otherwise ignore @@ -120,11 +123,14 @@ def on_error(type, data) # add to inv_queue, unlesss maximum is reached def on_inv_transaction(hash) log.debug { ">> inv transaction: #{hash.hth}" } + + # TODO: get relay propagation from mempool if @node.relay_propagation.keys.include?(hash.hth) @node.relay_propagation[hash.hth] += 1 end - return if @node.inv_queue.size >= @node.config[:max][:inv] - @node.queue_inv([:tx, hash, self]) + + @node.store.mempool.inv(hash) + send_getdata_tx(hash) unless @node.store.mempool.exists?(hash.hth) end # received +inv_block+ message for given +hash+. @@ -170,7 +176,7 @@ def on_addr(addr) # push tx to storage queue def on_tx(tx) log.debug { ">> tx: #{tx.hash} (#{tx.payload.size} bytes)" } - @node.queue.push([:tx, tx]) + @node.store.mempool.add(tx) end # received +block+ message for given +blk+. diff --git a/spec/node/command_api_spec.rb b/spec/node/command_api_spec.rb index 6ca77eb..7f39ad8 100644 --- a/spec/node/command_api_spec.rb +++ b/spec/node/command_api_spec.rb @@ -522,10 +522,9 @@ def store_block block r1 = send "monitor", channel: "tx" should_receive r1, id: 0 tx = @block.tx[0] - r2 = send "store_tx", hex: tx.to_payload.hth - should_receive r2, { "queued" => tx.hash } - + r2 = send "store_tx", hex: tx.to_payload.hth, skip_validation: true should_receive_tx(r1, tx, 0) + should_receive r2, { "added" => tx.hash } end it "should unmonitor tx" do @@ -537,7 +536,7 @@ def store_block block tx = @block.tx[0] r3 = send "store_tx", hex: tx.to_payload.hth - should_receive r3, { "queued" => tx.hash } + should_receive r3, { "added" => tx.hash } test_command("tslb") {|r| (0..TSLB_TIMEOUT).include?(r['tslb']).should == true } end @@ -610,9 +609,9 @@ def store_block block r1 = send "monitor", channel: "output" should_receive r1, id: 0 tx = @block.tx[0] - r2 = send "store_tx", hex: tx.to_payload.hth - should_receive r2, { "queued" => tx.hash } + r2 = send "store_tx", hex: tx.to_payload.hth, skip_validation: true should_receive_output(r1, tx, 0, 0) + should_receive r2, { "added" => tx.hash } end it "should unmonitor outputs" do @@ -621,7 +620,7 @@ def store_block block tx = @block.tx[0] r2 = send "store_tx", hex: tx.to_payload.hth - should_receive r2, { "queued" => tx.hash } + should_receive r2, { "added" => tx.hash } test_command("tslb") {|r| (0..TSLB_TIMEOUT).include?(r['tslb']).should == true } end From 460f8bccec8019d6ca48ac7dbf158d6cad6e7b77 Mon Sep 17 00:00:00 2001 From: Marius Hanne Date: Mon, 22 Jun 2015 04:41:49 +0200 Subject: [PATCH 2/3] command_handler: query mempool statistics --- lib/bitcoin/node/command_handler.rb | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/lib/bitcoin/node/command_handler.rb b/lib/bitcoin/node/command_handler.rb index ff4e4d1..cec6ff6 100644 --- a/lib/bitcoin/node/command_handler.rb +++ b/lib/bitcoin/node/command_handler.rb @@ -319,6 +319,17 @@ def handle_info Bitcoin.namecoin? ? {names: @node.store.db[:names].count}.merge(info) : info end + def handle_mempool_stats + mempool = @node.store.mempool + { + total: mempool.transactions.count, + accepted: mempool.accepted.count, + rejected: mempool.rejected.count, + doublespend: mempool.doublespend.count, + oldest: (Time.now - mempool.transactions.order(:created_at).first[:created_at]).to_i + } + end + def add_monitor params, channels @monitors << { params: params, channels: channels } @monitors.size - 1 From 2bd2b2a151c9fdf757c87f337255cfa29beccce3 Mon Sep 17 00:00:00 2001 From: Marius Hanne Date: Tue, 23 Jun 2015 20:17:17 +0200 Subject: [PATCH 3/3] run received tx / inv tx messages through queues again --- lib/bitcoin/node/connection_handler.rb | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/lib/bitcoin/node/connection_handler.rb b/lib/bitcoin/node/connection_handler.rb index cc6702c..fb7513b 100644 --- a/lib/bitcoin/node/connection_handler.rb +++ b/lib/bitcoin/node/connection_handler.rb @@ -129,8 +129,8 @@ def on_inv_transaction(hash) @node.relay_propagation[hash.hth] += 1 end - @node.store.mempool.inv(hash) - send_getdata_tx(hash) unless @node.store.mempool.exists?(hash.hth) + return if @node.inv_queue.size >= @node.config[:max][:inv] + @node.queue_inv([:tx, hash, self]) end # received +inv_block+ message for given +hash+. @@ -176,7 +176,7 @@ def on_addr(addr) # push tx to storage queue def on_tx(tx) log.debug { ">> tx: #{tx.hash} (#{tx.payload.size} bytes)" } - @node.store.mempool.add(tx) + @node.queue.push([:tx, tx]) end # received +block+ message for given +blk+.