ethminer/libethereum/TransactionQueue.h

/*
	This file is part of cpp-ethereum.

	cpp-ethereum is free software: you can redistribute it and/or modify
	it under the terms of the GNU General Public License as published by
	the Free Software Foundation, either version 3 of the License, or
	(at your option) any later version.

	cpp-ethereum is distributed in the hope that it will be useful,
	but WITHOUT ANY WARRANTY; without even the implied warranty of
	MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
	GNU General Public License for more details.

	You should have received a copy of the GNU General Public License
	along with cpp-ethereum.  If not, see <http://www.gnu.org/licenses/>.
*/
/** @file TransactionQueue.h
 * @author Gav Wood <i@gavwood.com>
 * @date 2014
 */

#pragma once

#include <functional>
#include <condition_variable>
#include <thread>
#include <deque>
#include <libdevcore/Common.h>
#include <libdevcore/Guards.h>
#include <libdevcore/Log.h>
#include <libethcore/Common.h>
#include "Transaction.h"

namespace dev
{
namespace eth
{

class BlockChain;

struct TransactionQueueChannel: public LogChannel { static const char* name(); static const int verbosity = 4; };
struct TransactionQueueTraceChannel: public LogChannel { static const char* name(); static const int verbosity = 7; };
#define ctxq dev::LogOutputStream<dev::eth::TransactionQueueTraceChannel, true>()

enum class IfDropped { Ignore, Retry };

/**
 * @brief A queue of Transactions, each stored as RLP.
 * Maintains a transaction queue sorted by nonce diff and gas price
 * @threadsafe
 */
class TransactionQueue
{
public:
	/// @brief TransactionQueue
	/// @param _limit Maximum number of pending transactions in the queue
	/// @param _futureLimit Maximum number of future nonce transactions
	TransactionQueue(unsigned _limit = 1024, unsigned _futureLimit = 1024);
	~TransactionQueue();
	void enqueue(RLP const& _data, h512 const& _nodeId);
	ImportResult import(bytes const& _tx, IfDropped _ik = IfDropped::Ignore) { return import(&_tx, _ik); }
	ImportResult import(Transaction const& _tx, IfDropped _ik = IfDropped::Ignore);

	void drop(h256 const& _txHash);

	unsigned waiting(Address const& _a) const;
	Transactions topTransactions(unsigned _limit) const;
	h256Hash knownTransactions() const;
	u256 maxNonce(Address const& _a) const;
	void setFuture(h256 const& _t);
	void dropGood(Transaction const& _t);

	void clear();
	template <class T> Handler<> onReady(T const& _t) { return m_onReady.add(_t); }
	template <class T> Handler<ImportResult, h256 const&, h512 const&> onImport(T const& _t) { return m_onImport.add(_t); }

private:
	struct VerifiedTransaction
	{
		VerifiedTransaction(Transaction const& _t): transaction(_t) {}
		VerifiedTransaction(VerifiedTransaction&& _t): transaction(std::move(_t.transaction)) {}

		VerifiedTransaction(VerifiedTransaction const&) = delete;
		VerifiedTransaction& operator=(VerifiedTransaction const&) = delete;

		Transaction transaction;
	};

	struct UnverifiedTransaction
	{
		UnverifiedTransaction() {}
		UnverifiedTransaction(bytesConstRef const& _t, h512 const& _nodeId): transaction(_t.toBytes()), nodeId(_nodeId) {}
		UnverifiedTransaction(UnverifiedTransaction&& _t): transaction(std::move(_t.transaction)) {}
		UnverifiedTransaction& operator=(UnverifiedTransaction&& _other) { transaction = std::move(_other.transaction); nodeId = std::move(_other.nodeId); return *this; }

		UnverifiedTransaction(UnverifiedTransaction const&) = delete;
		UnverifiedTransaction& operator=(UnverifiedTransaction const&) = delete;

		bytes transaction;
		h512 nodeId;
	};

	struct PriorityCompare
	{
		TransactionQueue& queue;
		bool operator()(VerifiedTransaction const& _first, VerifiedTransaction const& _second) const
		{
			u256 const& height1 = _first.transaction.nonce() - queue.m_currentByAddressAndNonce[_first.transaction.sender()].begin()->first;
			u256 const& height2 = _second.transaction.nonce() - queue.m_currentByAddressAndNonce[_second.transaction.sender()].begin()->first;
			return height1 < height2 || (height1 == height2 && _first.transaction.gasPrice() > _second.transaction.gasPrice());
		}
	};

	// Use a set with dynamic comparator for minmax priority queue. The comparator takes into account min account nonce. Updating it does not affect the order.
	using PriorityQueue = std::multiset<VerifiedTransaction, PriorityCompare>;

	ImportResult import(bytesConstRef _tx, IfDropped _ik = IfDropped::Ignore);
	ImportResult check_WITH_LOCK(h256 const& _h, IfDropped _ik);
	ImportResult manageImport_WITH_LOCK(h256 const& _h, Transaction const& _transaction);

	void insertCurrent_WITH_LOCK(std::pair<h256, Transaction> const& _p);
	void makeCurrent_WITH_LOCK(Transaction const& _t);
	bool remove_WITH_LOCK(h256 const& _txHash);
	u256 maxNonce_WITH_LOCK(Address const& _a) const;
	void verifierBody();

	mutable SharedMutex m_lock;													///< General lock.
	h256Hash m_known;															///< Hashes of transactions in both sets.

	std::unordered_map<h256, std::function<void(ImportResult)>> m_callbacks;	///< Called once.
	h256Hash m_dropped;															///< Transactions that have previously been dropped

	PriorityQueue m_current;
	std::unordered_map<h256, PriorityQueue::iterator> m_currentByHash;			///< Transaction hash to set ref
	std::unordered_map<Address, std::map<u256, PriorityQueue::iterator>> m_currentByAddressAndNonce; ///< Transactions grouped by account and nonce
	std::unordered_map<Address, std::map<u256, VerifiedTransaction>> m_future;	/// Future transactions

	Signal<> m_onReady;															///< Called when a subsequent call to import transactions will return a non-empty container. Be nice and exit fast.
	Signal<ImportResult, h256 const&, h512 const&> m_onImport;					///< Called for each import attempt. Arguments are result, transaction id an node id. Be nice and exit fast.
	unsigned m_limit;															///< Max number of pending transactions
	unsigned m_futureLimit;														///< Max number of future transactions
	unsigned m_futureSize = 0;													///< Current number of future transactions

	std::condition_variable m_queueReady;										///< Signaled when m_unverified has a new entry.
	std::vector<std::thread> m_verifiers;
	std::deque<UnverifiedTransaction> m_unverified;								///< Pending verification queue
	mutable Mutex x_queue;														///< Verification queue mutex
	bool m_aborting = false;													///< Exit condition for verifier.
};

}
}