BlockQueue.h

// Aleth: Ethereum C++ client, tools and libraries.
// Copyright 2014-2019 Aleth Authors.
// Licensed under the GNU General Public License, Version 3.


#pragma once

#include <condition_variable>
#include <thread>
#include <deque>
#include <boost/thread.hpp>
#include <libdevcore/Common.h>
#include <libdevcore/Log.h>
#include <libethcore/Common.h>
#include <libdevcore/Guards.h>
#include <libethcore/BlockHeader.h>
#include "VerifiedBlock.h"

namespace dev
{

namespace eth
{

class BlockChain;

struct BlockQueueStatus
{
    size_t importing;
    size_t verified;
    size_t verifying;
    size_t unverified;
    size_t future;
    size_t unknown;
    size_t bad;
};

enum class QueueStatus
{
    Ready,
    Importing,
    UnknownParent,
    Bad,
    Unknown
};

std::ostream& operator<< (std::ostream& os, QueueStatus const& obj);

template<class T>
class SizedBlockQueue
{
public:
    std::size_t count() const { return m_queue.size(); }

    std::size_t size() const { return m_size; }
    
    bool isEmpty() const { return m_queue.empty(); }

    h256 nextHash() const { return m_queue.front().verified.info.sha3Uncles(); }

    T const& next() const { return m_queue.front(); }

    void clear()
    {
        m_queue.clear();
        m_size = 0;
    }

    void enqueue(T&& _t)
    {
        m_queue.emplace_back(std::move(_t));
        m_size += m_queue.back().blockData.size();
    }

    T dequeue()
    {
        T t;
        std::swap(t, m_queue.front());
        m_queue.pop_front();
        m_size -= t.blockData.size();

        return t;
    }

    std::vector<T> dequeueMultiple(std::size_t _n)
    {
        return removeRange(m_queue.begin(), m_queue.begin() + _n);
    }

    bool remove(h256 const& _hash)
    {
        std::vector<T> removed = removeIf(sha3UnclesEquals(_hash));
        return !removed.empty();
    }

    template<class Pred>
    std::vector<T> removeIf(Pred _pred)
    {
        auto const removedBegin = std::remove_if(m_queue.begin(), m_queue.end(), _pred);

        return removeRange(removedBegin, m_queue.end());
    }

    bool replace(h256 const& _hash, T&& _t)
    {
        auto const it = std::find_if(m_queue.begin(), m_queue.end(), sha3UnclesEquals(_hash));

        if (it == m_queue.end())
            return false;

        m_size -= it->blockData.size();
        m_size += _t.blockData.size();
        *it = std::move(_t);

        return true;
    }

private:
    static std::function<bool(T const&)> sha3UnclesEquals(h256 const& _hash)
    {
        return [&_hash](T const& _t) { return _t.verified.info.sha3Uncles() == _hash; };
    }

    std::vector<T> removeRange(typename std::deque<T>::iterator _begin, typename std::deque<T>::iterator _end)
    {
        std::vector<T> ret(std::make_move_iterator(_begin), std::make_move_iterator(_end));

        for (auto it = ret.begin(); it != ret.end(); ++it)
            m_size -= it->blockData.size();

        m_queue.erase(_begin, _end);
        return ret;
    }

    std::deque<T> m_queue;
    std::atomic<size_t> m_size = {0};   ///< Tracks total size in bytes
};

template<class KeyType>
class SizedBlockMap
{
public:
    std::size_t count() const { return m_map.size(); }

    std::size_t size() const { return m_size; }

    bool isEmpty() const { return m_map.empty(); }

    KeyType firstKey() const { return m_map.begin()->first; }

    void clear()
    {
        m_map.clear();
        m_size = 0;
    }

    void insert(KeyType const& _key, h256 const& _hash, bytes&& _blockData)
    {
        auto hashAndBlock = std::make_pair(_hash, std::move(_blockData));
        auto keyAndValue = std::make_pair(_key, std::move(hashAndBlock));
        m_map.insert(std::move(keyAndValue));
        m_size += _blockData.size();
    }

    // returns removed (hash, block data) pairs
    std::vector<std::pair<h256, bytes>> removeByKeyEqual(KeyType const& _key)
    {
        auto const equalRange = m_map.equal_range(_key);
        return removeRange(equalRange.first, equalRange.second);
    }

    // returns removed (hash, block data) pairs
    std::vector<std::pair<h256, bytes>> removeByKeyNotGreater(KeyType const& _key)
    {
        return removeRange(m_map.begin(), m_map.upper_bound(_key));
    }

private:
    // map of key to (hash, block data) pairs
    using BlockMultimap = std::multimap<KeyType, std::pair<h256, bytes>>;

    std::vector<std::pair<h256, bytes>> removeRange(typename BlockMultimap::iterator _begin, typename BlockMultimap::iterator _end)
    {
        std::vector<std::pair<h256, bytes>> removed;
        std::size_t removedSize = 0;
        for (auto it = _begin; it != _end; ++it)
        {
            removed.push_back(std::move(it->second));
            removedSize += removed.back().second.size();
        }

        m_size -= removedSize;
        m_map.erase(_begin, _end);

        return removed;
    }
        
    BlockMultimap m_map;
    std::atomic<size_t> m_size = {0};   ///< Tracks total size in bytes
};

/**
 * @brief A queue of blocks. Sits between network or other I/O and the BlockChain.
 * Sorts them ready for blockchain insertion (with the BlockChain::sync() method).
 * @threadsafe
 */
class BlockQueue: HasInvariants
{
public:
    BlockQueue();
    ~BlockQueue();

    void setChain(BlockChain const& _bc) { m_bc = &_bc; }

    /// Import a block into the queue.
    ImportResult import(bytesConstRef _block, bool _isOurs = false);

    /// Notes that time has moved on and some blocks that used to be "in the future" may no be valid.
    void tick();

    /// Grabs at most @a _max of the blocks that are ready, giving them in the correct order for insertion into the chain.
    /// Don't forget to call doneDrain() once you're done importing.
    void drain(std::vector<VerifiedBlock>& o_out, unsigned _max);

    /// Must be called after a drain() call. Notes that the drained blocks have been imported into the blockchain, so we can forget about them.
    /// @returns true iff there are additional blocks ready to be processed.
    bool doneDrain(h256s const& _knownBad = h256s());

    /// Notify the queue that the chain has changed and a new block has attained 'ready' status (i.e. is in the chain).
    void noteReady(h256 const& _b) { WriteGuard l(m_lock); noteReady_WITH_LOCK(_b); }

    /// Force a retry of all the blocks with unknown parents.
    void retryAllUnknown();

    /// Get information on the items queued.
    std::pair<unsigned, unsigned> items() const { ReadGuard l(m_lock); return std::make_pair(m_readySet.size(), m_unknownSet.size()); }

    /// Clear everything.
    void clear();

    /// Stop all activity, leaves the class in limbo, waiting for destruction
    void stop();

    /// Return first block with an unknown parent.
    h256 firstUnknown() const { ReadGuard l(m_lock); return m_unknownSet.size() ? *m_unknownSet.begin() : h256(); }

    /// Get some infomration on the current status.
    BlockQueueStatus status() const;

    /// Get some infomration on the given block's status regarding us.
    QueueStatus blockStatus(h256 const& _h) const;

    Handler<> onReady(std::function<void(void)> _t) { return m_onReady.add(_t); }
    Handler<> onBlocksDrained(std::function<void(void)> _t) { return m_onBlocksDrained.add(_t); }

    template <class T> void setOnBad(T const& _t) { m_onBad = _t; }

    bool knownFull() const;
    bool unknownFull() const;
    u256 difficulty() const;    // Total difficulty of queueud blocks
    bool isActive() const;

private:
    struct UnverifiedBlock
    {
        h256 hash;
        h256 parentHash;
        bytes blockData;
    };

    void noteReady_WITH_LOCK(h256 const& _b);

    bool invariants() const override;

    void verifierBody();
    void collectUnknownBad_WITH_BOTH_LOCKS(h256 const& _bad);
    void updateBad_WITH_LOCK(h256 const& _bad);
    void drainVerified_WITH_BOTH_LOCKS();

    std::size_t knownSize() const;
    std::size_t knownCount() const;
    std::size_t unknownSize() const;
    std::size_t unknownCount() const;

    BlockChain const* m_bc;                                             ///< The blockchain into which our imports go.

    mutable boost::shared_mutex m_lock;                                 ///< General lock for the sets, m_future and m_unknown.
    h256Hash m_drainingSet;                                             ///< All blocks being imported.
    h256Hash m_readySet;                                                ///< All blocks ready for chain import.
    h256Hash m_unknownSet;                                              ///< Set of all blocks whose parents are not ready/in-chain.
    SizedBlockMap<h256> m_unknown;                                      ///< For blocks that have an unknown parent; we map their parent hash to the block stuff, and insert once the block appears.
    h256Hash m_knownBad;                                                ///< Set of blocks that we know will never be valid.
    SizedBlockMap<time_t> m_future;                                     ///< Set of blocks that are not yet valid. Ordered by timestamp
    h256Hash m_futureSet;  ///< Set of all blocks that are not yet valid.
    Signal<> m_onReady;                                                 ///< Called when a subsequent call to import blocks will return a non-empty container. Be nice and exit fast.
    Signal<> m_onBlocksDrained;                                         ///< Called when blocks have been drained from the block queue. Be nice and exit fast.

    mutable Mutex m_verification;                                       ///< Mutex that allows writing to m_verified, m_verifying and m_unverified.
    std::condition_variable m_moreToVerify;                             ///< Signaled when m_unverified has a new entry.
    SizedBlockQueue<VerifiedBlock> m_verified;                              ///< List of blocks, in correct order, verified and ready for chain-import.
    SizedBlockQueue<VerifiedBlock> m_verifying;                             ///< List of blocks being verified; as long as the block component (bytes) is empty, it's not finished.
    SizedBlockQueue<UnverifiedBlock> m_unverified;                          ///< List of <block hash, parent hash, block data> in correct order, ready for verification.

    std::vector<std::thread> m_verifiers;                               ///< Threads who only verify.
    std::atomic<bool> m_deleting = {false};                             ///< Exit condition for verifiers.

    std::function<void(Exception&)> m_onBad;                            ///< Called if we have a block that doesn't verify.
    u256 m_difficulty;                                                  ///< Total difficulty of blocks in the queue
    u256 m_drainingDifficulty;                                          ///< Total difficulty of blocks in draining

    Logger m_logger{createLogger(VerbosityDebug, "bq")};
    Logger m_loggerDetail{createLogger(VerbosityTrace, "bq")};
};

boost::log::formatting_ostream& operator<<(
    boost::log::formatting_ostream& _out, BlockQueueStatus const& _s);
}
}