Peer connection

README.md

@@ -53,6 +53,27 @@ repository and submit a pull request back to develop.
     * reject
     * tx
     * version
+  * Message Serialisation
+    * addr
+    * block
+    * getaddr
+    * getblocks
+    * getdata
+    * getheaders
+    * headers
+    * inv
+    * notfound
+    * ping
+    * pong
+    * tx
+    * version
+  * Common Structure Serialisation
+    * varint/varint[]
+    * varstring/varstring[]
+    * inventory vector
+    * network address
+    * txin/txout/outpoint
+    * block header  
 * OTP Application / Full Node
   * Peer
     * Connection Pool/Acceptor and Handler
@@ -70,29 +91,9 @@ repository and submit a pull request back to develop.
     * Transaction Queues
   * Event Model
   * Logging Strategy
-  * Common Structure Serialisation
-    * varint/varint[]
-    * varstring/varstring[]
-    * inventory vector
-    * network address
-    * txin/txout/outpoint
-    * block header  
   * Message Serialisation
-    * addr
-    * alert
-    * block
-    * getaddr
-    * getblocks
-    * getdata
-    * getheaders
-    * headers
-    * inv
-    * notfound
-    * ping
-    * pong
     * reject
-    * tx
-    * version
+    * alert
 * OTP Application / Full Node
   * Server Layout and Deployment
   * Blockchain Bulk Storage and Index API

config/config.exs

@@ -21,4 +21,6 @@ use Mix.Config
 # Configuration from the imported file will override the ones defined
 # here (which is why it is important to import them last).
 #
-#     import_config "#{Mix.env}.exs"
+
+import_config "#{Mix.env}.exs"
+

config/dev.exs

@@ -0,0 +1,6 @@
+use Mix.Config
+
+config :bitcoin, :node, []
+
+config :exlager,
+  level: :info

config/test.exs

@@ -0,0 +1,9 @@
+use Mix.Config
+
+
+# TODO Logger.supresses unused variable warnings when it removes parts of the AST
+# Unfortunately Lager doesn't do that. Would be nice to find some workaround.
+
+config :exlager,
+  level: :error
+

lib/bitcoin.ex

@@ -1,2 +1,25 @@
 defmodule Bitcoin do
+  use Application
+
+  # See http://elixir-lang.org/docs/stable/elixir/Application.html
+  # for more information on OTP Applications
+  def start(_type, _args) do
+    import Supervisor.Spec
+
+    # Start node only if :bitcoin,:node config section is present
+    # TODO this is not great, because when using Bitcon-Ex as a lib,
+    # there must a way to overwrite our dev default (which is node enabled)
+    children = case Application.fetch_env(:bitcoin, :node) do
+      :error ->
+         []
+      {:ok, _node_config} ->
+         [ supervisor(Bitcoin.Node.Supervisor, []) ]
+    end
+
+    # See http://elixir-lang.org/docs/stable/elixir/Supervisor.html
+    # for other strategies and supported options
+    opts = [strategy: :one_for_one, name: Bitcoin.Supervisor]
+    Supervisor.start_link(children, opts)
+  end
 end
+

lib/bitcoin/node.ex

@@ -1,26 +1,73 @@
 defmodule Bitcoin.Node do
-  use Application
+  use GenServer
 
-  defmodule Subsystems do
-    use Supervisor
+  require Lager
 
-    def start_link do
-      Supervisor.start_link(__MODULE__, :ok, [])
-    end
+  @default_config %{
+    listen_ip: '0.0.0.0',
+    listen_port: 8333,
+    max_connections: 8,
+    user_agent: "/Bitcoin-Ex:0.0.0/",
+    data_directory: Path.expand("~/.bitcoin-ex"),
+    services: <<1, 0, 0, 0, 0, 0, 0, 0>> # TODO probably doesn't belong to config
+  }
+
+  @protocol_version 70002
+
+
+  # Interface
 
-    @peer_subsystem_name Bitcoin.Node.Peers
+  def start_link, do: GenServer.start(__MODULE__, nil, name: __MODULE__)
+  def version_fields,  do: GenServer.call(__MODULE__, :version_fields)
+  def config,  do: GenServer.call(__MODULE__, :config)
+  def nonce,  do: GenServer.call(__MODULE__, :nonce)
+  def height, do: 1
+
+  # Implementation
+
+  def init(_) do
+    self() |> send(:initialize)
+    {:ok, %{}}
+  end
 
-    def init(:ok) do
-      children = [
-        supervisor(@peer_subsystem_name, [[name: @peer_subsystem_name]])
-      ]
+  def handle_info(:initialize, state) do
+    Lager.info "Node initialization"
 
-      supervise([], strategy: :one_for_one)
+    config = case Application.fetch_env(:bitcoin, :node) do
+      :error -> @default_config
+      {:ok, config} -> 
+        @default_config |> Map.merge(config |> Enum.into(%{}))
     end
 
+    File.mkdir_p(config.data_directory)
+
+    state = state|> Map.merge(%{
+      nonce: Bitcoin.Util.nonce64(),
+      config: config
+    })
+
+    {:noreply, state}
+  end
+
+  def handle_call(:config, _from, state), do: {:reply, state.config, state}
+  def handle_call(:nonce, _from, state), do: {:reply, state.nonce, state}
+
+  def handle_call(:version_fields, _from, state) do
+    fields = %{
+      height: height(),
+      nonce: state.nonce,
+      relay: true,
+      services: <<1, 0, 0, 0, 0, 0, 0, 0>>,
+      timestamp: timestamp(),
+      version: @protocol_version,
+      user_agent: state.config[:user_agent],
+    }
+    {:reply, fields, state}
   end
 
-  def start(_type, _args) do
-    Subsystems.start_link()
+
+  def timestamp do
+    {megas, s, _milis} = :os.timestamp
+    round(1.0e6*megas + s)
   end
-end
+end

lib/bitcoin/node/network.ex

@@ -0,0 +1,36 @@
+defmodule Bitcoin.Node.Network do
+
+  # TODO 
+  # def connected?
+  # def connections
+  # def connect (this should be called on node start if node is started with some addnode option)
+  # some kind of health indicator?
+  #
+  # This module is also probably where Node will be requesting to fetch misisng inv / headers etc.
+
+  alias Bitcoin.Node.Network
+
+  @default_modules [
+    # Addrs managager, keeps list of IPs to connect to
+    addr: Network.Addr,
+    # Peer connection handler, exchanges information with a single peer
+    peer: Network.Peer,
+    # Peers discovery - find IPs of peers to connect to if we have non in the database
+    discovery: Network.Discovery,
+    # Connection manager, accepts incoming connection, keeps track of all connected peers
+    connection_manager: Network.ConnectionManager
+  ]
+
+
+  def find_more_addrs do
+    modules[:discovery].begin_discovery()
+  end
+
+  def modules do
+    case Application.get_env(:bitcoin, :node, :modules) do
+       nil -> @default_modules
+      list -> @default_modules |> Keyword.merge(list)
+    end
+  end
+
+end

lib/bitcoin/node/network/addr.ex

@@ -0,0 +1,50 @@
+defmodule Bitcoin.Node.Network.Addr do
+  @moduledoc """
+    Keeps database of known network nodes.
+
+    Dummy version. Would be nice to switch to some dedicated struct from Protocol.NetworkAddress.
+    We may want to keep fields like last connection try times, last successful connection time,
+    maybe some score (e.g. higher for addrs from trusted seeds). Score could also help with blacklisting
+    nodes from which we detected abuse.
+  """
+  use GenServer
+
+  require Lager
+
+  alias Bitcoin.Protocol.Types.NetworkAddress
+
+  # Ignaring opts which contains modules list since we don't currently need it
+  def start_link(_opts \\ %{}), do: GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
+  def add(%NetworkAddress{} = addr), do: GenServer.cast(__MODULE__, {:add, addr})
+  def get, do: GenServer.call(__MODULE__, :get)
+  def count, do: GenServer.call(__MODULE__, :count)
+  def clear, do: GenServer.cast(__MODULE__, :clear)
+
+  def handle_cast({:add, %NetworkAddress{} = addr}, addrs) do
+    Lager.debug("adding new network address #{addr.address |> :inet.ntoa}")
+    existing = addrs[addr.address]
+
+    # If we already have this address, update timestamp if it's older
+    if (!existing || existing && existing.time < addr.time) && valid?(addr) do
+      {:noreply, addrs |> Map.put(addr.address, addr)}
+    else
+      {:noreply, addrs}
+    end
+  end
+
+  def handle_cast(:clear, _addrs) do
+    {:noreply, %{}}
+  end
+
+  def handle_call(:count, _from, addrs) do
+    {:reply, addrs |> Map.size, addrs}
+  end
+
+  def handle_call(:get, _from, addrs) when addrs == %{}, do: {:reply, nil, addrs}
+  def handle_call(:get, _from, addrs) do
+    {:reply, addrs |> Map.values |> Enum.random, addrs}
+  end
+
+  defp valid?(%NetworkAddress{} = na), do: na.time <= Bitcoin.Node.timestamp()
+
+end

lib/bitcoin/node/network/connection_manager.ex

@@ -0,0 +1,94 @@
+defmodule Bitcoin.Node.Network.ConnectionManager do
+
+  # Reagent connection handler
+  defmodule ReagentHandler do
+    use Reagent
+
+    def handle(%Reagent.Connection{socket: socket}) do
+      {:ok, pid} = Bitcoin.Node.Network.Peer.start(socket)
+      # Potential issue:
+      # If the connection gets closed after Peer.start but before switching the controlling process
+      # then probably Peer will never receive _:tcp_closed. Not sure if we need to care because
+      # it should just timout then
+      socket |> :gen_tcp.controlling_process(pid)
+      socket |> :inet.setopts(active: true)
+      :ok
+    end
+  end
+
+  use GenServer
+
+  require Lager
+
+  alias Bitcoin.Protocol.Types.NetworkAddress
+
+  def start_link(%{modules: _modules} = opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__)
+  def connect(ip, port), do: GenServer.cast(__MODULE__, {:connect, ip, port})
+  def register_peer(), do: GenServer.call(__MODULE__, :register_peer)
+
+  def init(opts) do
+    state = %{
+      modules: opts.modules,
+      config: Bitcoin.Node.config(),
+      peers: []
+    }
+
+    {:ok, _pid} = Reagent.start_link(ReagentHandler, port: state.config[:listen_port])
+    self() |> send(:periodical_connectivity_check)
+    {:ok, state}
+  end
+
+  def handle_info(:periodical_connectivity_check, state) do
+    self() |> send(:check_connectivity)
+    self() |> Process.send_after(:periodical_connectivity_check, 10_000)
+    {:noreply, state}
+  end
+
+  def handle_info(:check_connectivity, state) do
+    num_conn = length(state.peers)
+    max_conn = state.config[:max_connections]
+    Lager.info("[CM] #{num_conn} peers connected")
+
+    # TODO we want to differentiate between outbound_max_connections and max_connections
+    # E.g. bitcoin-core behavior is that it won't have more than 8 outbound connections
+    # regardless of the max-connections setting.
+    # ALso, there's no hard limit on max_connections currently, Reagent limit should be
+    # dynamic plus we can go over limit if some peer connection is already in progress
+    # and we add another one
+    if num_conn < max_conn do
+      (0..(max_conn - num_conn)) |> Enum.each(fn _ ->
+        state |> add_peer()
+      end)
+    end
+
+    {:noreply, state}
+  end
+
+  def handle_info({:DOWN, _ref, :process, peer, _reason}, state) do
+    Lager.info("[CM] unregistered peer #{peer |> inspect}")
+    self() |> send(:check_connectivity)
+    {:noreply, state |> Map.put(:peers, state.peers |> List.delete(peer))}
+  end
+
+  def handle_call(:register_peer, {peer, _ref}, state) do
+    Lager.info("[CM] registered peer #{peer |> inspect}")
+    state = state |> Map.put(:peers, [peer | state.peers])
+    Process.monitor(peer)
+    {:reply, :ok, state}
+  end
+
+  def handle_cast({:connect, ip, port}, state) do
+    Bitcoin.Node.Network.Peer.start(ip, port)
+    {:noreply, state}
+  end
+
+  def add_peer(%{modules: modules}) do
+    case modules[:addr].get do
+      %NetworkAddress{address: ip, port: port} ->
+        connect(ip, port)
+      nil ->
+        Bitcoin.Node.Network.find_more_addrs()
+    end
+  end
+
+end