text: # Copyright 2016, 2017 Intel Corporation
  #
  # 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.
  # ------------------------------------------------------------------------------
  
  # This file provides the 'core', ie. the Validator class
  import hashlib
  import logging
  import os
  import signal
  import time
  import threading
  
  from sawtooth_validator.concurrent.threadpool import \
      InstrumentedThreadPoolExecutor
  from sawtooth_validator.consensus.notifier import ConsensusNotifier
  from sawtooth_validator.consensus.proxy import ConsensusProxy
  from sawtooth_validator.consensus.proxy import ConsensusActivationObserver
  from sawtooth_validator.consensus.registry import ConsensusRegistry
  from sawtooth_validator.database.lmdb_nolock_database import LMDBNoLockDatabase
  from sawtooth_validator.database.native_lmdb import NativeLmdbDatabase
  from sawtooth_validator.journal.block_validator import \
      BlockValidationResultStore
  from sawtooth_validator.journal.block_sender import BroadcastBlockSender
  from sawtooth_validator.journal.block_store import BlockStore
  from sawtooth_validator.journal.block_manager import BlockManager
  from sawtooth_validator.journal.completer import Completer
  from sawtooth_validator.journal.responder import Responder
  from sawtooth_validator.journal.journal import Journal
  from sawtooth_validator.networking.dispatch import Dispatcher
  from sawtooth_validator.state.batch_tracker import BatchTracker
  from sawtooth_validator.state.merkle import MerkleDatabase
  from sawtooth_validator.state.settings_view import SettingsViewFactory
  from sawtooth_validator.state.settings_cache import SettingsObserver
  from sawtooth_validator.state.settings_cache import SettingsCache
  from sawtooth_validator.state.identity_view import IdentityViewFactory
  from sawtooth_validator.state.state_view import StateViewFactory
  from sawtooth_validator.gossip.permission_verifier import PermissionVerifier
  from sawtooth_validator.gossip.permission_verifier import IdentityCache
  from sawtooth_validator.gossip.identity_observer import IdentityObserver
  from sawtooth_validator.networking.interconnect import Interconnect
  from sawtooth_validator.gossip.gossip import Gossip
  
  from sawtooth_validator.server.events.broadcaster import EventBroadcaster
  
  from sawtooth_validator.journal.receipt_store import TransactionReceiptStore
  
  from sawtooth_validator.server import network_handlers
  from sawtooth_validator.server import component_handlers
  from sawtooth_validator.server import consensus_handlers
  
  LOGGER = logging.getLogger(__name__)
  
  
  class Validator:
      def __init__(self,
                   bind_network,
                   bind_component,
                   bind_consensus,
                   endpoint,
                   peering,
                   seeds_list,
                   peer_list,
                   data_dir,
                   config_dir,
                   identity_signer,
                   key_dir,
                   scheduler_type,
                   permissions,
                   minimum_peer_connectivity,
                   maximum_peer_connectivity,
                   state_pruning_block_depth,
                   fork_cache_keep_time,
                   network_public_key=None,
                   network_private_key=None,
                   roles=None,
                   component_thread_pool_workers=10,
                   network_thread_pool_workers=10,
                   signature_thread_pool_workers=3):
          """Constructs a validator instance.
  
          Args:
              bind_network (str): the network endpoint
              bind_component (str): the component endpoint
              endpoint (str): the zmq-style URI of this validator's
                  publically reachable endpoint
              peering (str): The type of peering approach. Either 'static'
                  or 'dynamic'. In 'static' mode, no attempted topology
                  buildout occurs -- the validator only attempts to initiate
                  peering connections with endpoints specified in the
                  peer_list. In 'dynamic' mode, the validator will first
                  attempt to initiate peering connections with endpoints
                  specified in the peer_list and then attempt to do a
                  topology buildout starting with peer lists obtained from
                  endpoints in the seeds_list. In either mode, the validator
                  will accept incoming peer requests up to max_peers.
              seeds_list (list of str): a list of addresses to connect
                  to in order to perform the initial topology buildout
              peer_list (list of str): a list of peer addresses
              data_dir (str): path to the data directory
              config_dir (str): path to the config directory
              identity_signer (Signer): cryptographic signer the validator uses
                  for signing
              key_dir (str): path to the data directory
              component_thread_pool_workers (int): number of workers in the
                  component thread pool; defaults to 10.
              network_thread_pool_workers (int): number of workers in the network
                  thread pool; defaults to 10.
              signature_thread_pool_workers (int): number of workers in the
                  signature thread pool; defaults to 3.
          """
          # -- Setup Global State Database and Factory -- #
          global_state_db_filename = os.path.join(
              data_dir, 'merkle-{}.lmdb'.format(bind_network[-2:]))
          LOGGER.debug(
              'global state database file is %s', global_state_db_filename)
          # Here we are creating the LMDB database, in total there are 3 lmdb databases, for state (merkle tree), transaction receipts, and blocks
          global_state_db = NativeLmdbDatabase(
              global_state_db_filename,
              indexes=MerkleDatabase.create_index_configuration())
          state_view_factory = StateViewFactory(global_state_db)
  
          # -- Setup Receipt Store -- #
          receipt_db_filename = os.path.join(
              data_dir, 'txn_receipts-{}.lmdb'.format(bind_network[-2:]))
          LOGGER.debug('txn receipt store file is %s', receipt_db_filename)
          receipt_db = LMDBNoLockDatabase(receipt_db_filename, 'c')
          receipt_store = TransactionReceiptStore(receipt_db)
  
          # -- Setup Block Store -- #
          block_db_filename = os.path.join(
              data_dir, 'block-{}.lmdb'.format(bind_network[-2:]))
          LOGGER.debug('block store file is %s', block_db_filename)
          block_db = NativeLmdbDatabase(
              block_db_filename,
              indexes=BlockStore.create_index_configuration())
          block_store = BlockStore(block_db)
          # The cache keep time for the journal's block cache must be greater
          # than the cache keep time used by the completer.
          base_keep_time = 1200
  
          # FUTURE: Take a look at this block manager, maybe block handling can be altered a bit using this
          block_manager = BlockManager()
          block_manager.add_commit_store(block_store)
  
          block_status_store = BlockValidationResultStore()
  
          # -- Setup Thread Pools -- #
          # FUTURE: Similar to these, create a threadpool for ServerRegisterReceiver
          component_thread_pool = InstrumentedThreadPoolExecutor(
              max_workers=component_thread_pool_workers,
              name='Component')
          network_thread_pool = InstrumentedThreadPoolExecutor(
              max_workers=network_thread_pool_workers,
              name='Network')
          client_thread_pool = InstrumentedThreadPoolExecutor(
              max_workers=5,
              name='Client')
          sig_pool = InstrumentedThreadPoolExecutor(
              max_workers=signature_thread_pool_workers,
              name='Signature')
  
          # -- Setup Dispatchers -- #
          component_dispatcher = Dispatcher()
          network_dispatcher = Dispatcher()
  
          # -- Setup Services -- #
          component_service = Interconnect(
              bind_component,
              component_dispatcher,
              secured=False,
              reap=False,
              max_incoming_connections=20,
              monitor=True,
              max_future_callback_workers=10)
  
          zmq_identity = hashlib.sha512(
              time.time().hex().encode()).hexdigest()[:23]
  
          secure = False
          if network_public_key is not None and network_private_key is not None:
              secure = True
  
          network_service = Interconnect(
              bind_network,
              dispatcher=network_dispatcher,
              zmq_identity=zmq_identity,
              secured=secure,
              server_public_key=network_public_key,
              server_private_key=network_private_key,
              reap=True,
              public_endpoint=endpoint,
              connection_timeout=120,
              max_incoming_connections=100,
              max_future_callback_workers=10,
              authorize=True,
              signer=identity_signer,
              roles=roles)
  
          # -- Setup Transaction Execution Platform -- #
          batch_tracker = BatchTracker(block_store.has_batch)
  
          settings_cache = SettingsCache(
              SettingsViewFactory(state_view_factory),
          )
  
          event_broadcaster = EventBroadcaster(
              component_service, block_store, receipt_store)
  
          # -- Consensus Engine -- #
          consensus_thread_pool = InstrumentedThreadPoolExecutor(
              max_workers=3,
              name='Consensus')
          consensus_dispatcher = Dispatcher()
          consensus_service = Interconnect(
              bind_consensus,
              consensus_dispatcher,
              secured=False,
              reap=False,
              max_incoming_connections=20,
              max_future_callback_workers=10)
  
          consensus_registry = ConsensusRegistry()
  
          consensus_notifier = ConsensusNotifier(
              consensus_service,
              consensus_registry,
              identity_signer.get_public_key().as_hex())
  
          consensus_activation_observer = ConsensusActivationObserver(
              consensus_registry,
              consensus_notifier,
              settings_view_factory=SettingsViewFactory(state_view_factory))
  
          # -- Setup P2P Networking -- #
          gossip = Gossip(
              network_service,
              settings_cache,
              lambda: block_store.chain_head,
              block_store.chain_head_state_root,
              consensus_notifier,
              endpoint=endpoint,
              peering_mode=peering,
              initial_seed_endpoints=seeds_list,
              initial_peer_endpoints=peer_list,
              minimum_peer_connectivity=minimum_peer_connectivity,
              maximum_peer_connectivity=maximum_peer_connectivity,
              topology_check_frequency=1
          )
  
          consensus_notifier.set_gossip(gossip)
  
          completer = Completer(
              block_manager=block_manager,
              transaction_committed=block_store.has_transaction,
              get_committed_batch_by_id=block_store.get_batch,
              get_committed_batch_by_txn_id=(
                  block_store.get_batch_by_transaction
              ),
              gossip=gossip,
              cache_keep_time=base_keep_time,
              cache_purge_frequency=30,
              requested_keep_time=300)
          self._completer = completer
  
          block_sender = BroadcastBlockSender(completer, gossip)
  
          identity_view_factory = IdentityViewFactory(
              StateViewFactory(global_state_db))
  
          id_cache = IdentityCache(identity_view_factory)
  
          # -- Setup Permissioning -- #
          permission_verifier = PermissionVerifier(
              permissions,
              block_store.chain_head_state_root,
              id_cache)
  
          identity_observer = IdentityObserver(
              to_update=id_cache.invalidate,
              forked=id_cache.forked)
  
          settings_observer = SettingsObserver(
              to_update=settings_cache.invalidate,
              forked=settings_cache.forked)
  
          # -- Setup Journal -- #
          journal = Journal(
              block_store=block_store,
              block_manager=block_manager,
              state_database=global_state_db,
              block_sender=block_sender,
              block_status_store=block_status_store,
              consensus_notifier=consensus_notifier,
              consensus_registry=consensus_registry,
              state_pruning_block_depth=state_pruning_block_depth,
              fork_cache_keep_time=fork_cache_keep_time,
              data_dir=data_dir,
              batch_observers=[batch_tracker],
              invalid_transaction_observers=[batch_tracker],
              observers=[
                  event_broadcaster,
                  receipt_store,
                  batch_tracker,
                  identity_observer,
                  settings_observer,
                  consensus_activation_observer
              ],
              key_dir=key_dir,
              genesis_observers=[receipt_store],
          )
  
          completer.set_get_chain_head(lambda: journal.chain_head)
  
          responder = Responder(completer)
  
          completer.set_on_block_received(journal.queue_block)
  
          # -- Register Message Handler -- #
          network_handlers.add(
              network_dispatcher, network_service, gossip, completer,
              responder, network_thread_pool, sig_pool,
              lambda block_id: block_id in block_manager, journal.has_batch,
              permission_verifier, consensus_notifier)
  
          component_handlers.add(
              component_dispatcher, gossip,
              completer, block_store, batch_tracker,
              global_state_db, self.get_chain_head_state_root_hash,
              receipt_store, event_broadcaster, permission_verifier,
              component_thread_pool, client_thread_pool,
              sig_pool, journal, identity_signer.get_public_key().as_hex())
  
          # -- Store Object References -- #
          self._component_dispatcher = component_dispatcher
          self._component_service = component_service
          self._component_thread_pool = component_thread_pool
  
          self._network_dispatcher = network_dispatcher
          self._network_service = network_service
          self._network_thread_pool = network_thread_pool
  
          consensus_proxy = ConsensusProxy(
              block_manager=block_manager,
              journal=journal,
              gossip=gossip,
              identity_signer=identity_signer,
              settings_view_factory=SettingsViewFactory(state_view_factory),
              state_view_factory=state_view_factory,
              consensus_registry=consensus_registry,
              consensus_notifier=consensus_notifier)
  
          consensus_handlers.add(
              consensus_dispatcher,
              consensus_thread_pool,
              consensus_proxy,
              consensus_notifier)
  
          self._block_status_store = block_status_store
  
          self._consensus_notifier = consensus_notifier
          self._consensus_dispatcher = consensus_dispatcher
          self._consensus_service = consensus_service
          self._consensus_thread_pool = consensus_thread_pool
          self._consensus_registry = consensus_registry
  
          self._client_thread_pool = client_thread_pool
          self._sig_pool = sig_pool
  
          self._gossip = gossip
  
          self._journal = journal
  
      def start(self):
          self._component_dispatcher.start()
          self._component_service.start()
          # will check if genesis block should be created
          self._journal.start()
  
          self._consensus_dispatcher.start()
          self._consensus_service.start()
          self._network_dispatcher.start()
          self._network_service.start()
  
          self._gossip.start()
  
          self._completer.set_on_batch_received(self._journal.submit_batch)
          signal_event = threading.Event()
  
          signal.signal(signal.SIGTERM,
                        lambda sig, fr: signal_event.set())
          # This is where the main thread will be during the bulk of the
          # validator's life.
          # The validator's main thread just sleeps for the rest time, and the signal_event is set when SIGTERM is received
          while not signal_event.is_set():
              signal_event.wait(timeout=20)
  
      def stop(self):
          self._gossip.stop()
          self._component_dispatcher.stop()
          self._network_dispatcher.stop()
          self._network_service.stop()
  
          self._component_service.stop()
  
          self._consensus_service.stop()
          self._consensus_dispatcher.stop()
  
          self._network_thread_pool.shutdown(wait=True)
          self._component_thread_pool.shutdown(wait=True)
          self._client_thread_pool.shutdown(wait=True)
          self._sig_pool.shutdown(wait=True)
  
          self._journal.stop()
  
          threads = threading.enumerate()
  
          # This will remove the MainThread, which will exit when we exit with
          # a sys.exit() or exit of main().
          threads.remove(threading.current_thread())
  
          while threads:
              if len(threads) < 4:
                  LOGGER.info(
                      "remaining threads: %s",
                      ", ".join(
                          ["{} ({})".format(x.name, x.__class__.__name__)
                           for x in threads]))
              for t in threads.copy():
                  if not t.is_alive():
                      t.join()
                      threads.remove(t)
                  if threads:
                      time.sleep(1)
  
          LOGGER.info("All threads have been stopped and joined")
  
      def get_chain_head_state_root_hash(self):
          return self._chain_controller.chain_head.state_root_hash
  
  
  def unwrap_if_not_none(blkw):
      if blkw:
          return blkw.block
      return None
  