gt:         memoryPool = new MemoryPool();
context: /**
   * Copyright 2013 Google Inc.
   * Copyright 2014 Andreas Schildbach
   *
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package com.github.kryptohash.kryptohashj.core;
  
  import com.github.kryptohash.kryptohashj.crypto.DRMWorkaround;
  import com.github.kryptohash.kryptohashj.net.BlockingClientManager;
  import com.github.kryptohash.kryptohashj.net.ClientConnectionManager;
  import com.github.kryptohash.kryptohashj.net.FilterMerger;
  import com.github.kryptohash.kryptohashj.net.NioClientManager;
  import com.github.kryptohash.kryptohashj.net.discovery.PeerDiscovery;
  import com.github.kryptohash.kryptohashj.net.discovery.PeerDiscoveryException;
  import com.github.kryptohash.kryptohashj.net.discovery.TorDiscovery;
  import com.github.kryptohash.kryptohashj.script.Script;
  import com.github.kryptohash.kryptohashj.utils.ExponentialBackoff;
  import com.github.kryptohash.kryptohashj.utils.ListenerRegistration;
  import com.github.kryptohash.kryptohashj.utils.Threading;
  import com.google.common.annotations.VisibleForTesting;
  import com.google.common.base.Preconditions;
  import com.google.common.collect.ImmutableList;
  import com.google.common.collect.Lists;
  import com.google.common.net.InetAddresses;
  import com.google.common.primitives.Ints;
  import com.google.common.primitives.Longs;
  import com.google.common.util.concurrent.*;
  import com.subgraph.orchid.TorClient;
  import net.jcip.annotations.GuardedBy;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import javax.annotation.Nullable;
  import java.io.IOException;
  import java.net.Inet6Address;
  import java.net.InetAddress;
  import java.net.InetSocketAddress;
  import java.net.NoRouteToHostException;
  import java.net.Socket;
  import java.util.*;
  import java.util.concurrent.*;
  import java.util.concurrent.locks.ReentrantLock;
  
  import static com.google.common.base.Preconditions.checkArgument;
  import static com.google.common.base.Preconditions.checkNotNull;
  import static com.google.common.base.Preconditions.checkState;
  
  /**
   * <p>Runs a set of connections to the P2P network, brings up connections to replace disconnected nodes and manages
   * the interaction between them all. Most applications will want to use one of these.</p>
   * 
   * <p>PeerGroup tries to maintain a constant number of connections to a set of distinct peers.
   * Each peer runs a network listener in its own thread.  When a connection is lost, a new peer
   * will be tried after a delay as long as the number of connections less than the maximum.</p>
   * 
   * <p>Connections are made to addresses from a provided list.  When that list is exhausted,
   * we start again from the head of the list.</p>
   * 
   * <p>The PeerGroup can broadcast a transaction to the currently connected set of peers.  It can
   * also handle download of the blockchain from peers, restarting the process when peers die.</p>
   *
   * <p>PeerGroup implements the {@link Service} interface. This means before it will do anything,
   * you must call the {@link com.google.common.util.concurrent.Service#start()} method (which returns
   * a future) or {@link com.google.common.util.concurrent.Service#startAndWait()} method, which will block
   * until peer discovery is completed and some outbound connections have been initiated (it will return
   * before handshaking is done, however). You should call {@link com.google.common.util.concurrent.Service#stop()}
   * when finished. Note that not all methods of PeerGroup are safe to call from a UI thread as some may do
   * network IO, but starting and stopping the service should be fine.</p>
   */
  public class PeerGroup extends AbstractExecutionThreadService implements TransactionBroadcaster {
      private static final Logger log = LoggerFactory.getLogger(PeerGroup.class);
      private static final int DEFAULT_CONNECTIONS = 4;
      private static final int TOR_TIMEOUT_SECONDS = 60;
  
      protected final ReentrantLock lock = Threading.lock("peergroup");
  
      // Addresses to try to connect to, excluding active peers.
      @GuardedBy("lock") private final PriorityQueue<PeerAddress> inactives;
      @GuardedBy("lock") private final Map<PeerAddress, ExponentialBackoff> backoffMap;
  
      // Currently active peers. This is an ordered list rather than a set to make unit tests predictable.
      private final CopyOnWriteArrayList<Peer> peers;
      // Currently connecting peers.
      private final CopyOnWriteArrayList<Peer> pendingPeers;
      private final ClientConnectionManager channels;
      @Nullable private final TorClient torClient;
  
      // The peer that has been selected for the purposes of downloading announced data.
      @GuardedBy("lock") private Peer downloadPeer;
      // Callback for events related to chain download
      @Nullable @GuardedBy("lock") private PeerEventListener downloadListener;
      // Callbacks for events related to peer connection/disconnection
      private final CopyOnWriteArrayList<ListenerRegistration<PeerEventListener>> peerEventListeners;
      // Peer discovery sources, will be polled occasionally if there aren't enough inactives.
      private final CopyOnWriteArraySet<PeerDiscovery> peerDiscoverers;
      // The version message to use for new connections.
      @GuardedBy("lock") private VersionMessage versionMessage;
      // Switch for enabling download of pending transaction dependencies.
      @GuardedBy("lock") private boolean downloadTxDependencies;
      // A class that tracks recent transactions that have been broadcast across the network, counts how many
      // peers announced them and updates the transaction confidence data. It is passed to each Peer.
      private final MemoryPool memoryPool;
      // How many connections we want to have open at the current time. If we lose connections, we'll try opening more
      // until we reach this count.
      @GuardedBy("lock") private int maxConnections;
      // Minimum protocol version we will allow ourselves to connect to: require Bloom filtering.
      private volatile int vMinRequiredProtocolVersion = FilteredBlock.MIN_PROTOCOL_VERSION;
  
      // Runs a background thread that we use for scheduling pings to our peers, so we can measure their performance
      // and network latency. We ping peers every pingIntervalMsec milliseconds.
      private volatile Timer vPingTimer;
      /** How many milliseconds to wait after receiving a pong before sending another ping. */
      public static final long DEFAULT_PING_INTERVAL_MSEC = 2000;
      private long pingIntervalMsec = DEFAULT_PING_INTERVAL_MSEC;
  
      @GuardedBy("lock") private boolean useLocalhostPeerWhenPossible = true;
      @GuardedBy("lock") private boolean ipv6Unreachable = false;
  
      private final NetworkParameters params;
      private final AbstractBlockChain chain;
      @GuardedBy("lock") private long fastCatchupTimeSecs;
      private final CopyOnWriteArrayList<Wallet> wallets;
      private final CopyOnWriteArrayList<PeerFilterProvider> peerFilterProviders;
  
      // This event listener is added to every peer. It's here so when we announce transactions via an "inv", every
      // peer can fetch them.
      private final AbstractPeerEventListener peerListener = new AbstractPeerEventListener() {
          @Override
          public List<Message> getData(Peer peer, GetDataMessage m) {
              return handleGetData(m);
          }
  
          @Override
          public void onBlocksDownloaded(Peer peer, Block block, int blocksLeft) {
              final double rate = checkNotNull(chain).getFalsePositiveRate();
              final double target = bloomFilterMerger.getBloomFilterFPRate() * MAX_FP_RATE_INCREASE;
              if (rate > target) {
                  // TODO: Avoid hitting this path if the remote peer didn't acknowledge applying a new filter yet.
                  if (log.isDebugEnabled())
                      log.debug("Force update Bloom filter due to high false positive rate ({} vs {})", rate, target);
                  recalculateFastCatchupAndFilter(FilterRecalculateMode.FORCE_SEND_FOR_REFRESH);
              }
          }
      };
  
      private int minBroadcastConnections = 2;
      private final AbstractWalletEventListener walletEventListener = new AbstractWalletEventListener() {
          // Because calculation of the new filter takes place asynchronously, these flags deduplicate requests.
          @GuardedBy("this") private boolean sendIfChangedQueued, dontSendQueued;
  
          private Runnable bloomSendIfChanged = new Runnable() {
              @Override public void run() {
                  recalculateFastCatchupAndFilter(FilterRecalculateMode.SEND_IF_CHANGED);
                  synchronized (walletEventListener) {
                      sendIfChangedQueued = false;
                  }
              }
          };
          private Runnable bloomDontSend = new Runnable() {
              @Override public void run() {
                  recalculateFastCatchupAndFilter(FilterRecalculateMode.DONT_SEND);
                  synchronized (walletEventListener) {
                      dontSendQueued = false;
                  }
              }
          };
  
          private synchronized void queueRecalc(boolean andTransmit) {
              if (andTransmit) {
                  if (!sendIfChangedQueued) {
                      log.info("Queuing recalc of the Bloom filter due to new keys or scripts becoming available");
                      sendIfChangedQueued = true;
                      Uninterruptibles.putUninterruptibly(jobQueue, bloomSendIfChanged);
                  }
              } else {
                  if (!dontSendQueued) {
                      log.info("Queuing recalc of the Bloom filter due to observing a pay to pubkey output on a relevant tx");
                      dontSendQueued = true;
                      Uninterruptibles.putUninterruptibly(jobQueue, bloomDontSend);
                  }
              }
          }
  
          @Override public void onScriptsAdded(Wallet wallet, List<Script> scripts) {
              queueRecalc(true);
          }
  
          @Override public void onKeysAdded(List<Ed25519Key> keys) {
              queueRecalc(true);
          }
  
          @Override
          public void onCoinsReceived(Wallet wallet, Transaction tx, Coin prevBalance, Coin newBalance) {
              // We received a relevant transaction. We MAY need to recalculate and resend the Bloom filter, but only
              // if we have received a transaction that includes a relevant pay-to-pubkey output.
              //
              // The reason is that pay-to-pubkey outputs, when spent, will not repeat any data we can predict in their
              // inputs. So a remote peer will update the Bloom filter for us when such an output is seen matching the
              // existing filter, so that it includes the tx hash in which the pay-to-pubkey output was observed. Thus
              // the spending transaction will always match (due to the outpoint structure).
              //
              // Unfortunately, whilst this is required for correct sync of the chain in blocks, there are two edge cases.
              //
              // (1) If a wallet receives a relevant, confirmed p2pubkey output that was not broadcast across the network,
              // for example in a coinbase transaction, then the node that's serving us the chain will update its filter
              // but the rest will not. If another transaction then spends it, the other nodes won't match/relay it.
              //
              // (2) If we receive a p2pubkey output broadcast across the network, all currently connected nodes will see
              // it and update their filter themselves, but any newly connected nodes will receive the last filter we
              // calculated, which would not include this transaction.
              //
              // For this reason we check if the transaction contained any relevant pay to pubkeys and force a recalc
              // and possibly retransmit if so. The recalculation process will end up including the tx hash into the
              // filter. In case (1), we need to retransmit the filter to the connected peers. In case (2), we don't
              // and shouldn't, we should just recalculate and cache the new filter for next time.
              for (TransactionOutput output : tx.getOutputs()) {
                  if (output.getScriptPubKey().isSentToRawPubKey() && output.isMine(wallet)) {
                      if (tx.getConfidence().getConfidenceType() == TransactionConfidence.ConfidenceType.BUILDING)
                          queueRecalc(true);
                      else
                          queueRecalc(false);
                      return;
                  }
              }
          }
      };
  
      // Exponential backoff for peers starts at 1 second and maxes at 10 minutes.
      private ExponentialBackoff.Params peerBackoffParams = new ExponentialBackoff.Params(1000, 1.5f, 10 * 60 * 1000);
      // Tracks failures globally in case of a network failure.
      private ExponentialBackoff groupBackoff = new ExponentialBackoff(new ExponentialBackoff.Params(1000, 1.5f, 10 * 1000));
  
      // Things for the dedicated PeerGroup management thread to do.
      private LinkedBlockingQueue<Runnable> jobQueue = new LinkedBlockingQueue<Runnable>();
  
      // This is a synchronized set, so it locks on itself. We use it to prevent TransactionBroadcast objects from
      // being garbage collected if nothing in the apps code holds on to them transitively. See the discussion
      // in broadcastTransaction.
      private final Set<TransactionBroadcast> runningBroadcasts;
  
      private class PeerStartupListener extends AbstractPeerEventListener {
          @Override
          public void onPeerConnected(Peer peer, int peerCount) {
              handleNewPeer(peer);
          }
  
          @Override
          public void onPeerDisconnected(Peer peer, int peerCount) {
              // The channel will be automatically removed from channels.
              handlePeerDeath(peer, null);
          }
      }
  
      @VisibleForTesting
      PeerEventListener startupListener = new PeerStartupListener();
  
      /**
       * <p>A reasonable default for the bloom filter false positive rate on mainnet. FP rates are values between 0.0 and 1.0
       * where 1.0 is "all transactions" i.e. 100%.</p>
       *
       * <p>Users for which low data usage is of utmost concern, 0.0001 may be better, for users
       * to whom anonymity is of utmost concern, 0.001 (0.1%) should provide very good privacy.</p>
       */
      public static final double DEFAULT_BLOOM_FILTER_FP_RATE = 0.0005;
      /** Maximum increase in FP rate before forced refresh of the bloom filter */
      public static final double MAX_FP_RATE_INCREASE = 2.0f;
      // An object that calculates bloom filters given a list of filter providers, whilst tracking some state useful
      // for privacy purposes.
      private FilterMerger bloomFilterMerger;
  
      /** The default timeout between when a connection attempt begins and version message exchange completes */
      public static final int DEFAULT_CONNECT_TIMEOUT_MILLIS = 5000;
      private volatile int vConnectTimeoutMillis = DEFAULT_CONNECT_TIMEOUT_MILLIS;
  
      /**
       * Creates a PeerGroup with the given parameters. No chain is provided so this node will report its chain height
       * as zero to other peers. This constructor is useful if you just want to explore the network but aren't interested
       * in downloading block data.
       *
       * @param params Network parameters
       */
  
      public PeerGroup(NetworkParameters params) {
          this(params, null);
      }
  
      /**
       * Creates a PeerGroup for the given network and chain. Blocks will be passed to the chain as they are broadcast
       * and downloaded. This is probably the constructor you want to use.
       */
      public PeerGroup(NetworkParameters params, @Nullable AbstractBlockChain chain) {
          this(params, chain, new NioClientManager());
      }
  
      /**
       * <p>Creates a PeerGroup that accesses the network via the Tor network. The provided TorClient is used so you can
       * preconfigure it beforehand. It should not have been already started. You can just use "new TorClient()" if
       * you don't have any particular configuration requirements.</p>
       *
       * <p>Peer discovery is automatically configured to use DNS seeds resolved via a random selection of exit nodes.
       * If running on the Oracle JDK the unlimited strength jurisdiction checks will also be overridden,
       * as they no longer apply anyway and can cause startup failures due to the requirement for AES-256.</p>
       *
       * <p>The user does not need any additional software for this: it's all pure Java. As of April 2014 <b>this mode
       * is experimental</b>.</p>
       *
       * @throws java.util.concurrent.TimeoutException if Tor fails to start within 20 seconds.
       */
      public static PeerGroup newWithTor(NetworkParameters params, @Nullable AbstractBlockChain chain, TorClient torClient) throws TimeoutException {
          checkNotNull(torClient);
          DRMWorkaround.maybeDisableExportControls();
          BlockingClientManager manager = new BlockingClientManager(torClient.getSocketFactory());
          final int CONNECT_TIMEOUT_MSEC = TOR_TIMEOUT_SECONDS * 1000;
          manager.setConnectTimeoutMillis(CONNECT_TIMEOUT_MSEC);
          PeerGroup result = new PeerGroup(params, chain, manager, torClient);
          result.setConnectTimeoutMillis(CONNECT_TIMEOUT_MSEC);
          result.addPeerDiscovery(new TorDiscovery(params, torClient));
          return result;
      }
  
      /**
       * Creates a new PeerGroup allowing you to specify the {@link ClientConnectionManager} which is used to create new
       * connections and keep track of existing ones.
       */
      public PeerGroup(NetworkParameters params, @Nullable AbstractBlockChain chain, ClientConnectionManager connectionManager) {
          this(params, chain, connectionManager, null);
      }
  
      /**
       * Creates a new PeerGroup allowing you to specify the {@link ClientConnectionManager} which is used to create new
       * connections and keep track of existing ones.
       */
      private PeerGroup(NetworkParameters params, @Nullable AbstractBlockChain chain, ClientConnectionManager connectionManager, @Nullable TorClient torClient) {
          this.params = checkNotNull(params);
          this.chain = chain;
          fastCatchupTimeSecs = params.getGenesisBlock().getTimeSeconds();
          wallets = new CopyOnWriteArrayList<Wallet>();
          peerFilterProviders = new CopyOnWriteArrayList<PeerFilterProvider>();
          this.torClient = torClient;
  
          // This default sentinel value will be overridden by one of two actions:
          //   - adding a peer discovery source sets it to the default
          //   - using connectTo() will increment it by one
          maxConnections = 0;
  
          int height = chain == null ? 0 : chain.getBestChainHeight();
          versionMessage = new VersionMessage(params, (long)height);
          // We never request that the remote node wait for a bloom filter yet, as we have no wallets
          versionMessage.relayTxesBeforeFilter = true;
  
          downloadTxDependencies = true;
  
  