<commit_before># Copyright (c) 2016, <NAME>
#
# All rights reserved.
#
# See the file "LICENCE" for information about the copyright
# and warranty status of this software.

'''Mempool handling.'''

import asyncio
import itertools
import time
from collections import defaultdict

from electrumx.lib.hash import hash_to_hex_str, hex_str_to_hash
from electrumx.lib.util import class_logger
from electrumx.server.db import UTXO


class MemPool(object):
    '''Representation of the daemon's mempool.

    Updated regularly in caught-up state.  Goal is to enable efficient
    response to the value() and transactions() calls.

    To that end we maintain the following maps:

       tx_hash -> (txin_pairs, txout_pairs, tx_fee, tx_size)
       hashX   -> set of all tx hashes in which the hashX appears

    A pair is a (hashX, value) tuple.  tx hashes are hex strings.
    '''

    def __init__(self, coin, tasks, daemon, notifications, lookup_utxos):
        self.logger = class_logger(__name__, self.__class__.__name__)
        self.coin = coin
        self.lookup_utxos = lookup_utxos
        self.tasks = tasks
        self.daemon = daemon
        self.notifications = notifications
        self.txs = {}
        self.hashXs = defaultdict(set)  # None can be a key
        self.fee_histogram = defaultdict(int)
        self.cached_compact_histogram = []
        self.histogram_time = 0

    async def _log_stats(self):
        while True:
            self.logger.info(f'{len(self.txs):,d} txs '
                             f'touching {len(self.hashXs):,d} addresses')
            await asyncio.sleep(120)

    async def _synchronize_forever(self):
        while True:
            await asyncio.sleep(5)
            await self._synchronize(False)

    async def _refresh_hashes(self):
        '''Return a (hash set, height) pair when we're sure which height they
        are for.'''
        while True:
            height = self.daemon.cached_height()
            hashes = await self.daemon.mempool_hashes()
            if height == await self.daemon.height():
                return set(hashes), height

    async def _synchronize(self, first_time):
        '''Asynchronously maintain mempool status with daemon.

        Processes the mempool each time the mempool refresh event is
        signalled.
        '''
        unprocessed = {}
        unfetched = set()
        touched = set()
        txs = self.txs
        next_refresh = 0
        fetch_size = 800
        process_some = self._async_process_some(fetch_size // 2)

        while True:
            now = time.time()
            # If processing a large mempool, a block being found might
            # shrink our work considerably, so refresh our view every 20s
            if now > next_refresh:
                hashes, height = await self._refresh_hashes()
                self._resync_hashes(hashes, unprocessed, unfetched, touched)
                next_refresh = time.time() + 20

            # Log progress of initial sync
            todo = len(unfetched) + len(unprocessed)
            if first_time:
                pct = (len(txs) - todo) * 100 // len(txs) if txs else 0
                self.logger.info(f'catchup {pct:d}% complete '
                                 f'({todo:,d} txs left)')
            if not todo:
                break

            # FIXME: parallelize
            if unfetched:
                count = min(len(unfetched), fetch_size)
                hex_hashes = [unfetched.pop() for n in range(count)]
                unprocessed.update(await self._fetch_raw_txs(hex_hashes))
            if unprocessed:
                await process_some(unprocessed, touched)

        await self.notifications.on_mempool(touched, height)

    def _resync_hashes(self, hashes, unprocessed, unfetched, touched):
        '''Re-sync self.txs with the list of hashes in the daemon's mempool.

        Additionally, remove gone hashes from unprocessed and
        unfetched.  Add new ones to unprocessed.
        '''
        txs = self.txs
        hashXs = self.hashXs
        fee_hist = self.fee_histogram
        gone = set(txs).difference(hashes)
        for hex_hash in gone:
            unfetched.discard(hex_hash)
            unprocessed.pop(hex_hash, None)
            item = txs.pop(hex_hash)
            if item:
                txin_pairs, txout_pairs, tx_fee, tx_size = item
                fee_rate = tx_fee // tx_size
                fee_hist[fee_rate] -= tx_size
                if fee_hist[fee_rate] == 0:
                    fee_hist.pop(fee_rate)
                tx_hashXs = set(hashX for hashX, value in txin_pairs)
                tx_hashXs.update(hashX for hashX, value in txout_pairs)
                for hashX in tx_hashXs:
                    hashXs[hashX].remove(hex_hash)
                    if not hashXs[hashX]:
                        del hashXs[hashX]
                touched.update(tx_hashXs)

        new = hashes.difference(txs)
        unfetched.update(new)
        for hex_hash in new:
            txs[hex_hash] = None

    def _async_process_some(self, limit):
        pending = []
        txs = self.txs

        async def process(unprocessed, touched):
            nonlocal pending

            raw_txs = {}

            while unprocessed and len(raw_txs) < limit:
                hex_hash, raw_tx = unprocessed.popitem()
                raw_txs[hex_hash] = raw_tx

            if unprocessed:
                deferred = []
            else:
                deferred = pending
                pending = []

            deferred = await self._process_raw_txs(raw_txs, deferred, touched)
            pending.extend(deferred)

        return process

    async def _fetch_raw_txs(self, hex_hashes):
        '''Fetch a list of mempool transactions.'''
        raw_txs = await self.daemon.getrawtransactions(hex_hashes)

        # Skip hashes the daemon has dropped.  Either they were
        # evicted or they got in a block.
        return {hh: raw for hh, raw in zip(hex_hashes, raw_txs) if raw}

    async def _process_raw_txs(self, raw_tx_map, pending, touched):
        '''Process the dictionary of raw transactions and return a dictionary
        of updates to apply to self.txs.
        '''
        script_hashX = self.coin.hashX_from_script
        deserializer = self.coin.DESERIALIZER

        # Deserialize each tx and put it in a pending list
        for tx_hash, raw_tx in raw_tx_map.items():
            tx, tx_size = deserializer(raw_tx).read_tx_and_vsize()

            # Convert the tx outputs into (hashX, value) pairs
            txout_pairs = [(script_hashX(txout.pk_script), txout.value)
                           for txout in tx.outputs]

            # Convert the tx inputs to ([prev_hex_hash, prev_idx) pairs
            txin_pairs = [(hash_to_hex_str(txin.prev_hash), txin.prev_idx)
                          for txin in tx.inputs]

            pending.append((tx_hash, txin_pairs, txout_pairs, tx_size))

        # The transaction inputs can be from other mempool<commit_msg>Deserialize mempool txs in a thread<commit_after># Copyright (c) 2016, <NAME>
#
# All rights reserved.
#
# See the file "LICENCE" for information about the copyright
# and warranty status of this software.

'''Mempool handling.'''

import asyncio
import itertools
import time
from collections import defaultdict

from electrumx.lib.hash import hash_to_hex_str, hex_str_to_hash
from electrumx.lib.util import class_logger
from electrumx.server.db import UTXO


class MemPool(object):
    '''Representation of the daemon's mempool.

    Updated regularly in caught-up state.  Goal is to enable efficient
    response to the value() and transactions() calls.

    To that end we maintain the following maps:

       tx_hash -> (txin_pairs, txout_pairs, tx_fee, tx_size)
       hashX   -> set of all tx hashes in which the hashX appears

    A pair is a (hashX, value) tuple.  tx hashes are hex strings.
    '''

    def __init__(self, coin, tasks, daemon, notifications, lookup_utxos):
        self.logger = class_logger(__name__, self.__class__.__name__)
        self.coin = coin
        self.lookup_utxos = lookup_utxos
        self.tasks = tasks
        self.daemon = daemon
        self.notifications = notifications
        self.txs = {}
        self.hashXs = defaultdict(set)  # None can be a key
        self.fee_histogram = defaultdict(int)
        self.cached_compact_histogram = []
        self.histogram_time = 0

    async def _log_stats(self):
        while True:
            self.logger.info(f'{len(self.txs):,d} txs '
                             f'touching {len(self.hashXs):,d} addresses')
            await asyncio.sleep(120)

    async def _synchronize_forever(self):
        while True:
            await asyncio.sleep(5)
            await self._synchronize(False)

    async def _refresh_hashes(self):
        '''Return a (hash set, height) pair when we're sure which height they
        are for.'''
        while True:
            height = self.daemon.cached_height()
            hashes = await self.daemon.mempool_hashes()
            if height == await self.daemon.height():
                return set(hashes), height

    async def _synchronize(self, first_time):
        '''Asynchronously maintain mempool status with daemon.

        Processes the mempool each time the mempool refresh event is
        signalled.
        '''
        unprocessed = {}
        unfetched = set()
        touched = set()
        txs = self.txs
        next_refresh = 0
        fetch_size = 800
        process_some = self._async_process_some(fetch_size // 2)

        while True:
            now = time.time()
            # If processing a large mempool, a block being found might
            # shrink our work considerably, so refresh our view every 20s
            if now > next_refresh:
                hashes, height = await self._refresh_hashes()
                self._resync_hashes(hashes, unprocessed, unfetched, touched)
                next_refresh = time.time() + 20

            # Log progress of initial sync
            todo = len(unfetched) + len(unprocessed)
            if first_time:
                pct = (len(txs) - todo) * 100 // len(txs) if txs else 0
                self.logger.info(f'catchup {pct:d}% complete '
                                 f'({todo:,d} txs left)')
            if not todo:
                break

            # FIXME: parallelize
            if unfetched:
                count = min(len(unfetched), fetch_size)
                hex_hashes = [unfetched.pop() for n in range(count)]
                unprocessed.update(await self._fetch_raw_txs(hex_hashes))
            if unprocessed:
                await process_some(unprocessed, touched)

        await self.notifications.on_mempool(touched, height)

    def _resync_hashes(self, hashes, unprocessed, unfetched, touched):
        '''Re-sync self.txs with the list of hashes in the daemon's mempool.

        Additionally, remove gone hashes from unprocessed and
        unfetched.  Add new ones to unprocessed.
        '''
        txs = self.txs
        hashXs = self.hashXs
        fee_hist = self.fee_histogram
        gone = set(txs).difference(hashes)
        for hex_hash in gone:
            unfetched.discard(hex_hash)
            unprocessed.pop(hex_hash, None)
            item = txs.pop(hex_hash)
            if item:
                txin_pairs, txout_pairs, tx_fee, tx_size = item
                fee_rate = tx_fee // tx_size
                fee_hist[fee_rate] -= tx_size
                if fee_hist[fee_rate] == 0:
                    fee_hist.pop(fee_rate)
                tx_hashXs = set(hashX for hashX, value in txin_pairs)
                tx_hashXs.update(hashX for hashX, value in txout_pairs)
                for hashX in tx_hashXs:
                    hashXs[hashX].remove(hex_hash)
                    if not hashXs[hashX]:
                        del hashXs[hashX]
                touched.update(tx_hashXs)

        new = hashes.difference(txs)
        unfetched.update(new)
        for hex_hash in new:
            txs[hex_hash] = None

    def _async_process_some(self, limit):
        pending = []
        txs = self.txs

        async def process(unprocessed, touched):
            nonlocal pending

            raw_txs = {}

            while unprocessed and len(raw_txs) < limit:
                hex_hash, raw_tx = unprocessed.popitem()
                raw_txs[hex_hash] = raw_tx

            if unprocessed:
                deferred = []
            else:
                deferred = pending
                pending = []

            deferred = await self._process_raw_txs(raw_txs, deferred, touched)
            pending.extend(deferred)

        return process

    async def _fetch_raw_txs(self, hex_hashes):
        '''Fetch a list of mempool transactions.'''
        raw_txs = await self.daemon.getrawtransactions(hex_hashes)

        # Skip hashes the daemon has dropped.  Either they were
        # evicted or they got in a block.
        return {hh: raw for hh, raw in zip(hex_hashes, raw_txs) if raw}

    async def _process_raw_txs(self, raw_tx_map, pending, touched):
        '''Process the dictionary of raw transactions and return a dictionary
        of updates to apply to self.txs.
        '''
        def deserialize_txs():
            script_hashX = self.coin.hashX_from_script
            deserializer = self.coin.DESERIALIZER

            # Deserialize each tx and put it in a pending list
            for tx_hash, raw_tx in raw_tx_map.items():
                tx, tx_size = deserializer(raw_tx).read_tx_and_vsize()

                # Convert the tx outputs into (hashX, value) pairs
                txout_pairs = [(script_hashX(txout.pk_script), txout.value)
                               for txout in tx.outputs]

                # Convert the tx inputs to ([prev_hex_hash, prev_idx) pairs
                txin_pairs = [(hash_to_hex_str(txin.prev_hash), txin.prev_idx)
                              for txin in tx.inputs]

                pending.append((tx_hash, txin_pairs, txout_pairs, tx_size))

        # Do this potentially slow operation in a thread so as not to
        # block
        await self.tasks.run_in_thread(deserialize_txs)

        # The transaction inputs can be from other mempool
{'fraction_non_alphanumeric': 0.07833099579242636, 'fraction_numerical': 0.0032258064516129032, 'mean_word_length': 1.9319490131578947, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5871150', 'n_tokens_mistral': 4322, 'n_tokens_neox': 4108, 'n_words': 1177}