dash/src/bls/bls_worker.cpp

// Copyright (c) 2018 The Dash Core developers
// Distributed under the MIT/X11 software license, see the accompanying
// file COPYING or http://www.opensource.org/licenses/mit-license.php.

#include "bls_worker.h"
#include "hash.h"
#include "serialize.h"

#include "util.h"

template <typename T>
bool VerifyVectorHelper(const std::vector<T>& vec, size_t start, size_t count)
{
    if (start == 0 && count == 0) {
        count = vec.size();
    }
    std::set<uint256> set;
    for (size_t i = start; i < start + count; i++) {
        if (!vec[i].IsValid())
            return false;
        // check duplicates
        if (!set.emplace(vec[i].GetHash()).second) {
            return false;
        }
    }
    return true;
}

// Creates a doneCallback and a future. The doneCallback simply finishes the future
template <typename T>
std::pair<std::function<void(const T&)>, std::future<T> > BuildFutureDoneCallback()
{
    auto p = std::make_shared<std::promise<T> >();
    std::function<void(const T&)> f = [p](const T& v) {
        p->set_value(v);
    };
    return std::make_pair(std::move(f), p->get_future());
}
template <typename T>
std::pair<std::function<void(T)>, std::future<T> > BuildFutureDoneCallback2()
{
    auto p = std::make_shared<std::promise<T> >();
    std::function<void(const T&)> f = [p](T v) {
        p->set_value(v);
    };
    return std::make_pair(std::move(f), p->get_future());
}


/////

CBLSWorker::CBLSWorker()
{
}

CBLSWorker::~CBLSWorker()
{
    Stop();
}

void CBLSWorker::Start()
{
    int workerCount = std::thread::hardware_concurrency() / 2;
    workerCount = std::max(std::min(1, workerCount), 4);
    workerPool.resize(workerCount);
    RenameThreadPool(workerPool, "dash-bls-worker");
}

void CBLSWorker::Stop()
{
    workerPool.clear_queue();
    workerPool.stop(true);
}

bool CBLSWorker::GenerateContributions(int quorumThreshold, const BLSIdVector& ids, BLSVerificationVectorPtr& vvecRet, BLSSecretKeyVector& skShares)
{
    BLSSecretKeyVectorPtr svec = std::make_shared<BLSSecretKeyVector>((size_t)quorumThreshold);
    vvecRet = std::make_shared<BLSVerificationVector>((size_t)quorumThreshold);
    skShares.resize(ids.size());

    for (int i = 0; i < quorumThreshold; i++) {
        (*svec)[i].MakeNewKey();
    }
    std::list<std::future<bool> > futures;
    size_t batchSize = 8;

    for (size_t i = 0; i < quorumThreshold; i += batchSize) {
        size_t start = i;
        size_t count = std::min(batchSize, quorumThreshold - start);
        auto f = [&, start, count](int threadId) {
            for (size_t j = start; j < start + count; j++) {
                (*vvecRet)[j] = (*svec)[j].GetPublicKey();
            }
            return true;
        };
        futures.emplace_back(workerPool.push(f));
    }

    for (size_t i = 0; i < ids.size(); i += batchSize) {
        size_t start = i;
        size_t count = std::min(batchSize, ids.size() - start);
        auto f = [&, start, count](int threadId) {
            for (size_t j = start; j < start + count; j++) {
                if (!skShares[j].SecretKeyShare(*svec, ids[j])) {
                    return false;
                }
            }
            return true;
        };
        futures.emplace_back(workerPool.push(f));
    }
    bool success = true;
    for (auto& f : futures) {
        if (!f.get()) {
            success = false;
        }
    }
    return success;
}

// aggregates a single vector of BLS objects in parallel
// the input vector is split into batches and each batch is aggregated in parallel
// when enough batches are finished to form a new batch, the new batch is queued for further parallel aggregation
// when no more batches can be created from finished batch results, the final aggregated is created and the doneCallback
// called.
// The Aggregator object needs to be created on the heap and it will delete itself after calling the doneCallback
// The input vector is not copied into the Aggregator but instead a vector of pointers to the original entries from the
// input vector is stored. This means that the input vector must stay alive for the whole lifetime of the Aggregator
template <typename T>
struct Aggregator {
    typedef T ElementType;

    size_t batchSize{16};
    std::shared_ptr<std::vector<const T*> > inputVec;

    bool parallel;
    ctpl::thread_pool& workerPool;

    std::mutex m;
    // items in the queue are all intermediate aggregation results of finished batches.
    // The intermediate results must be deleted by us again (which we do in SyncAggregateAndPushAggQueue)
    boost::lockfree::queue<T*> aggQueue;
    std::atomic<size_t> aggQueueSize{0};

    // keeps track of currently queued/in-progress batches. If it reaches 0, we are done
    std::atomic<size_t> waitCount{0};

    typedef std::function<void(const T& agg)> DoneCallback;
    DoneCallback doneCallback;

    // TP can either be a pointer or a reference
    template <typename TP>
    Aggregator(const std::vector<TP>& _inputVec,
               size_t start, size_t count,
               bool _parallel,
               ctpl::thread_pool& _workerPool,
               DoneCallback _doneCallback) :
            workerPool(_workerPool),
            parallel(_parallel),
            aggQueue(0),
            doneCallback(std::move(_doneCallback))
    {
        inputVec = std::make_shared<std::vector<const T*> >(count);
        for (size_t i = 0; i < count; i++) {
            (*inputVec)[i] = pointer(_inputVec[start + i]);
        }
    }

    const T* pointer(const T& v) { return &v; }
    const T* pointer(const T* v) { return v; }

    // Starts aggregation.
    // If parallel=true, then this will return fast, otherwise this will block until aggregation is done
    void Start()
    {
        size_t batchCount = (inputVec->size() + batchSize - 1) / batchSize;

        if (!parallel) {
            if (inputVec->size() == 1) {
                doneCallback(*(*inputVec)[0]);
            } else {
                doneCallback(SyncAggregate(*inputVec, 0, inputVec->size()));
            }
            delete this;
            return;
        }

        if (batchCount == 1) {
            // just a single batch of work, take a shortcut.
            PushWork([this](int threadId) {
                if (inputVec->size() == 1) {
                    doneCallback(*(*inputVec)[0]);
                } else {
                    doneCallback(SyncAggregate(*inputVec, 0, inputVec->size()));
                }
                delete this;
            });
            return;
        }

        // increment wait counter as otherwise the first finished async aggregation might signal that we're done
        IncWait();
        for (size_t i = 0; i < batchCount; i++) {
            size_t start = i * batchSize;
            size_t count = std::min(batchSize, inputVec->size() - start);
            AsyncAggregateAndPushAggQueue(inputVec, start, count, false);
        }
        // this will decrement the wait counter and in most cases NOT finish, as async work is still in progress
        CheckDone();
    }

    void IncWait()
    {
        ++waitCount;
    }

    void CheckDone()
    {
        if (--waitCount == 0) {
            Finish();
        }
    }

    void Finish()
    {
        // All async work is done, but we might have items in the aggQueue which are the results of the async
        // work. This is the case when these did not add up to a new batch. In this case, we have to aggregate
        // the items into the final result

        std::vector<T*> rem(aggQueueSize);
        for (size_t i = 0; i < rem.size(); i++) {
            T* p = nullptr;
            bool s = aggQueue.pop(p);
            assert(s);
            rem[i] = p;
        }

        T r;
        if (rem.size() == 1) {
            // just one intermediate result, which is actually the final result
            r = *rem[0];
        } else {
            // multiple intermediate results left which did not add up to a new batch. aggregate them now
            r = SyncAggregate(rem, 0, rem.size());
        }

        // all items which are left in the queue are intermediate results, so we must delete them
        for (size_t i = 0; i < rem.size(); i++) {
            delete rem[i];
        }
        doneCallback(r);

        delete this;
    }

    void AsyncAggregateAndPushAggQueue(std::shared_ptr<std::vector<const T*> >& vec, size_t start, size_t count, bool del)
    {
        IncWait();
        PushWork(std::bind(&Aggregator::SyncAggregateAndPushAggQueue, this, vec, start, count, del));
    }

    void SyncAggregateAndPushAggQueue(std::shared_ptr<std::vector<const T*> >& vec, size_t start, size_t count, bool del)
    {
        // aggregate vec and push the intermediate result onto the work queue
        PushAggQueue(SyncAggregate(*vec, start, count));
        if (del) {
            for (size_t i = 0; i < count; i++) {
                delete (*vec)[start + i];
            }
        }
        CheckDone();
    }

    void PushAggQueue(const T& v)
    {
        aggQueue.push(new T(v));

        if (++aggQueueSize >= batchSize) {
            // we've collected enough intermediate results to form a new batch.
            std::shared_ptr<std::vector<const T*> > newBatch;
            {
                std::unique_lock<std::mutex> l(m);
                if (aggQueueSize < batchSize) {
                    // some other worker thread grabbed this batch
                    return;
                }
                newBatch = std::make_shared<std::vector<const T*> >(batchSize);
                // collect items for new batch
                for (size_t i = 0; i < batchSize; i++) {
                    T* p = nullptr;
                    bool s = aggQueue.pop(p);
                    assert(s);
                    (*newBatch)[i] = p;
                }
                aggQueueSize -= batchSize;
            }

            // push new batch to work queue. del=true this time as these items are intermediate results and need to be deleted
            // after aggregation is done
            AsyncAggregateAndPushAggQueue(newBatch, 0, newBatch->size(), true);
        }
    }

    template <typename TP>
    T SyncAggregate(const std::vector<TP>& vec, size_t start, size_t count)
    {
        T result = *vec[start];
        for (size_t j = 1; j < count; j++) {
            result.AggregateInsecure(*vec[start + j]);
        }
        return result;
    }

    template <typename Callable>
    void PushWork(Callable&& f)
    {
        workerPool.push(f);
    }
};

// Aggregates multiple input vectors into a single output vector
// Inputs are in the following form:
//   [
//     [a1, b1, c1, d1],
//     [a2, b2, c2, d2],
//     [a3, b3, c3, d3],
//     [a4, b4, c4, d4],
//   ]
// The result is in the following form:
//   [ a1+a2+a3+a4, b1+b2+b3+b4, c1+c2+c3+c4, d1+d2+d3+d4]
// Same rules for the input vectors apply to the VectorAggregator as for the Aggregator (they must stay alive)
template <typename T>
struct VectorAggregator {
    typedef Aggregator<T> AggregatorType;
    typedef std::vector<T> VectorType;
    typedef std::shared_ptr<VectorType> VectorPtrType;
    typedef std::vector<VectorPtrType> VectorVectorType;
    typedef std::function<void(const VectorPtrType& agg)> DoneCallback;
    DoneCallback doneCallback;

    const VectorVectorType& vecs;
    size_t start;
    size_t count;
    bool parallel;
    ctpl::thread_pool& workerPool;

    std::atomic<size_t> doneCount;

    VectorPtrType result;
    size_t vecSize;

    VectorAggregator(const VectorVectorType& _vecs,
                     size_t _start, size_t _count,
                     bool _parallel, ctpl::thread_pool& _workerPool,
                     DoneCallback _doneCallback) :
            vecs(_vecs),
            parallel(_parallel),
            start(_start),
            count(_count),
            workerPool(_workerPool),
            doneCallback(std::move(_doneCallback))
    {
        assert(!vecs.empty());
        vecSize = vecs[0]->size();
        result = std::make_shared<VectorType>(vecSize);
        doneCount = 0;
    }

    void Start()
    {
        std::vector<AggregatorType*> aggregators;
        for (size_t i = 0; i < vecSize; i++) {
            std::vector<const T*> tmp(count);
            for (size_t j = 0; j < count; j++) {
                tmp[j] = &(*vecs[start + j])[i];
            }

            auto aggregator = new AggregatorType(std::move(tmp), 0, count, parallel, workerPool, std::bind(&VectorAggregator::CheckDone, this, std::placeholders::_1, i));
            // we can't directly start the aggregator here as it might be so fast that it deletes "this" while we are still in this loop
            aggregators.emplace_back(aggregator);
        }
        for (auto agg : aggregators) {
            agg->Start();
        }
    }

    void CheckDone(const T& agg, size_t idx)
    {
        (*result)[idx] = agg;
        if (++doneCount == vecSize) {
            doneCallback(result);
            delete this;
        }
    }
};

// See comment of AsyncVerifyContributionShares for a description on what this does
// Same rules as in Aggregator apply for the inputs
struct ContributionVerifier {
    struct BatchState {
        size_t start;
        size_t count;

        BLSVerificationVectorPtr vvec;
        CBLSSecretKey skShare;

        // starts with 0 and is incremented if either vvec or skShare aggregation finishs. If it reaches 2, we know
        // that aggregation for this batch is fully done. We can then start verification.
        std::unique_ptr<std::atomic<int> > aggDone;

        // we can't directly update a vector<bool> in parallel
        // as vector<bool> is not thread safe (uses bitsets internally)
        // so we must use vector<char> temporarely and concatenate/convert
        // each batch result into a final vector<bool>
        std::vector<char> verifyResults;
    };

    CBLSId forId;
    const std::vector<BLSVerificationVectorPtr>& vvecs;
    const BLSSecretKeyVector& skShares;
    size_t batchSize;
    bool parallel;
    bool aggregated;

    ctpl::thread_pool& workerPool;

    size_t batchCount;
    size_t verifyCount;

    std::vector<BatchState> batchStates;
    std::atomic<size_t> verifyDoneCount{0};
    std::function<void(const std::vector<bool>&)> doneCallback;

    ContributionVerifier(const CBLSId& _forId, const std::vector<BLSVerificationVectorPtr>& _vvecs,
                         const BLSSecretKeyVector& _skShares, size_t _batchSize,
                         bool _parallel, bool _aggregated, ctpl::thread_pool& _workerPool,
                         std::function<void(const std::vector<bool>&)> _doneCallback) :
        forId(_forId),
        vvecs(_vvecs),
        skShares(_skShares),
        batchSize(_batchSize),
        parallel(_parallel),
        aggregated(_aggregated),
        workerPool(_workerPool),
        doneCallback(std::move(_doneCallback))
    {
    }

    void Start()
    {
        if (!aggregated) {
            // treat all inputs as one large batch
            batchSize = vvecs.size();
            batchCount = 1;
        } else {
            batchCount = (vvecs.size() + batchSize - 1) / batchSize;
        }
        verifyCount = vvecs.size();

        batchStates.resize(batchCount);
        for (size_t i = 0; i < batchCount; i++) {
            auto& batchState = batchStates[i];

            batchState.aggDone.reset(new std::atomic<int>(0));
            batchState.start = i * batchSize;
            batchState.count = std::min(batchSize, vvecs.size() - batchState.start);
            batchState.verifyResults.assign(batchState.count, 0);
        }

        if (aggregated) {
            size_t batchCount2 = batchCount; // 'this' might get deleted while we're still looping
            for (size_t i = 0; i < batchCount2; i++) {
                AsyncAggregate(i);
            }
        } else {
            // treat all inputs as a single batch and verify one-by-one
            AsyncVerifyBatchOneByOne(0);
        }
    }

    void Finish()
    {
        size_t batchIdx = 0;
        std::vector<bool> result(vvecs.size());
        for (size_t i = 0; i < vvecs.size(); i += batchSize) {
            auto& batchState = batchStates[batchIdx++];
            for (size_t j = 0; j < batchState.count; j++) {
                result[batchState.start + j] = batchState.verifyResults[j] != 0;
            }
        }
        doneCallback(result);
        delete this;
    }

    void AsyncAggregate(size_t batchIdx)
    {
        auto& batchState = batchStates[batchIdx];

        // aggregate vvecs and skShares of batch in parallel
        auto vvecAgg = new VectorAggregator<CBLSPublicKey>(vvecs, batchState.start, batchState.count, parallel, workerPool, std::bind(&ContributionVerifier::HandleAggVvecDone, this, batchIdx, std::placeholders::_1));
        auto skShareAgg = new Aggregator<CBLSSecretKey>(skShares, batchState.start, batchState.count, parallel, workerPool, std::bind(&ContributionVerifier::HandleAggSkShareDone, this, batchIdx, std::placeholders::_1));

        vvecAgg->Start();
        skShareAgg->Start();
    }

    void HandleAggVvecDone(size_t batchIdx, const BLSVerificationVectorPtr& vvec)
    {
        auto& batchState = batchStates[batchIdx];
        batchState.vvec = vvec;
        if (++(*batchState.aggDone) == 2) {
            HandleAggDone(batchIdx);
        }
    }
    void HandleAggSkShareDone(size_t batchIdx, const CBLSSecretKey& skShare)
    {
        auto& batchState = batchStates[batchIdx];
        batchState.skShare = skShare;
        if (++(*batchState.aggDone) == 2) {
            HandleAggDone(batchIdx);
        }
    }

    void HandleVerifyDone(size_t batchIdx, size_t count)
    {
        size_t c = verifyDoneCount += count;
        if (c == verifyCount) {
            Finish();
        }
    }

    void HandleAggDone(size_t batchIdx)
    {
        auto& batchState = batchStates[batchIdx];

        if (batchState.vvec == nullptr || batchState.vvec->empty() || !batchState.skShare.IsValid()) {
            // something went wrong while aggregating and there is nothing we can do now except mark the whole batch as failed
            // this can only happen if inputs were invalid in some way
            batchState.verifyResults.assign(batchState.count, 0);
            HandleVerifyDone(batchIdx, batchState.count);
            return;
        }

        AsyncAggregatedVerifyBatch(batchIdx);
    }

    void AsyncAggregatedVerifyBatch(size_t batchIdx)
    {
        auto f = [this, batchIdx](int threadId) {
            auto& batchState = batchStates[batchIdx];
            bool result = Verify(batchState.vvec, batchState.skShare);
            if (result) {
                // whole batch is valid
                batchState.verifyResults.assign(batchState.count, 1);
                HandleVerifyDone(batchIdx, batchState.count);
            } else {
                // at least one entry in the batch is invalid, revert to per-contribution verification (but parallelized)
                AsyncVerifyBatchOneByOne(batchIdx);
            }
        };
        PushOrDoWork(std::move(f));
    }

    void AsyncVerifyBatchOneByOne(size_t batchIdx)
    {
        size_t count = batchStates[batchIdx].count;
        batchStates[batchIdx].verifyResults.assign(count, 0);
        for (size_t i = 0; i < count; i++) {
            auto f = [this, i, batchIdx](int threadId) {
                auto& batchState = batchStates[batchIdx];
                batchState.verifyResults[i] = Verify(vvecs[batchState.start + i], skShares[batchState.start + i]);
                HandleVerifyDone(batchIdx, 1);
            };
            PushOrDoWork(std::move(f));
        }
    }

    bool Verify(const BLSVerificationVectorPtr& vvec, const CBLSSecretKey& skShare)
    {
        CBLSPublicKey pk1;
        if (!pk1.PublicKeyShare(*vvec, forId)) {
            return false;
        }

        CBLSPublicKey pk2 = skShare.GetPublicKey();
        return pk1 == pk2;
    }

    template <typename Callable>
    void PushOrDoWork(Callable&& f)
    {
        if (parallel) {
            workerPool.push(std::move(f));
        } else {
            f(0);
        }
    }
};

void CBLSWorker::AsyncBuildQuorumVerificationVector(const std::vector<BLSVerificationVectorPtr>& vvecs,
                                                    size_t start, size_t count, bool parallel,
                                                    std::function<void(const BLSVerificationVectorPtr&)> doneCallback)
{
    if (start == 0 && count == 0) {
        count = vvecs.size();
    }
    if (vvecs.empty() || count == 0 || start > vvecs.size() || start + count > vvecs.size()) {
        doneCallback(nullptr);
        return;
    }
    if (!VerifyVerificationVectors(vvecs, start, count)) {
        doneCallback(nullptr);
        return;
    }

    auto agg = new VectorAggregator<CBLSPublicKey>(vvecs, start, count, parallel, workerPool, std::move(doneCallback));
    agg->Start();
}

std::future<BLSVerificationVectorPtr> CBLSWorker::AsyncBuildQuorumVerificationVector(const std::vector<BLSVerificationVectorPtr>& vvecs,
                                                                                     size_t start, size_t count, bool parallel)
{
    auto p = BuildFutureDoneCallback<BLSVerificationVectorPtr>();
    AsyncBuildQuorumVerificationVector(vvecs, start, count, parallel, std::move(p.first));
    return std::move(p.second);
}

BLSVerificationVectorPtr CBLSWorker::BuildQuorumVerificationVector(const std::vector<BLSVerificationVectorPtr>& vvecs,
                                                                   size_t start, size_t count, bool parallel)
{
    return AsyncBuildQuorumVerificationVector(vvecs, start, count, parallel).get();
}

template <typename T>
void AsyncAggregateHelper(ctpl::thread_pool& workerPool,
                          const std::vector<T>& vec, size_t start, size_t count, bool parallel,
                          std::function<void(const T&)> doneCallback)
{
    if (start == 0 && count == 0) {
        count = vec.size();
    }
    if (vec.empty() || count == 0 || start > vec.size() || start + count > vec.size()) {
        doneCallback(T());
        return;
    }
    if (!VerifyVectorHelper(vec, start, count)) {
        doneCallback(T());
        return;
    }

    auto agg = new Aggregator<T>(vec, start, count, parallel, workerPool, std::move(doneCallback));
    agg->Start();
}

void CBLSWorker::AsyncAggregateSecretKeys(const BLSSecretKeyVector& secKeys,
                                          size_t start, size_t count, bool parallel,
                                          std::function<void(const CBLSSecretKey&)> doneCallback)
{
    AsyncAggregateHelper(workerPool, secKeys, start, count, parallel, doneCallback);
}

std::future<CBLSSecretKey> CBLSWorker::AsyncAggregateSecretKeys(const BLSSecretKeyVector& secKeys,
                                                                size_t start, size_t count, bool parallel)
{
    auto p = BuildFutureDoneCallback<CBLSSecretKey>();
    AsyncAggregateSecretKeys(secKeys, start, count, parallel, std::move(p.first));
    return std::move(p.second);
}

CBLSSecretKey CBLSWorker::AggregateSecretKeys(const BLSSecretKeyVector& secKeys,
                                              size_t start, size_t count, bool parallel)
{
    return AsyncAggregateSecretKeys(secKeys, start, count, parallel).get();
}

void CBLSWorker::AsyncAggregatePublicKeys(const BLSPublicKeyVector& pubKeys,
                                          size_t start, size_t count, bool parallel,
                                          std::function<void(const CBLSPublicKey&)> doneCallback)
{
    AsyncAggregateHelper(workerPool, pubKeys, start, count, parallel, doneCallback);
}

std::future<CBLSPublicKey> CBLSWorker::AsyncAggregatePublicKeys(const BLSPublicKeyVector& pubKeys,
                                                                size_t start, size_t count, bool parallel)
{
    auto p = BuildFutureDoneCallback<CBLSPublicKey>();
    AsyncAggregatePublicKeys(pubKeys, start, count, parallel, std::move(p.first));
    return std::move(p.second);
}

CBLSPublicKey CBLSWorker::AggregatePublicKeys(const BLSPublicKeyVector& pubKeys,
                                              size_t start, size_t count, bool parallel)
{
    return AsyncAggregatePublicKeys(pubKeys, start, count, parallel).get();
}

void CBLSWorker::AsyncAggregateSigs(const BLSSignatureVector& sigs,
                                    size_t start, size_t count, bool parallel,
                                    std::function<void(const CBLSSignature&)> doneCallback)
{
    AsyncAggregateHelper(workerPool, sigs, start, count, parallel, doneCallback);
}

std::future<CBLSSignature> CBLSWorker::AsyncAggregateSigs(const BLSSignatureVector& sigs,
                                                          size_t start, size_t count, bool parallel)
{
    auto p = BuildFutureDoneCallback<CBLSSignature>();
    AsyncAggregateSigs(sigs, start, count, parallel, std::move(p.first));
    return std::move(p.second);
}

CBLSSignature CBLSWorker::AggregateSigs(const BLSSignatureVector& sigs,
                                        size_t start, size_t count, bool parallel)
{
    return AsyncAggregateSigs(sigs, start, count, parallel).get();
}


CBLSPublicKey CBLSWorker::BuildPubKeyShare(const BLSVerificationVectorPtr& vvec, const CBLSId& id)
{
    CBLSPublicKey pkShare;
    pkShare.PublicKeyShare(*vvec, id);
    return pkShare;
}

void CBLSWorker::AsyncVerifyContributionShares(const CBLSId& forId, const std::vector<BLSVerificationVectorPtr>& vvecs, const BLSSecretKeyVector& skShares,
                                               bool parallel, bool aggregated, std::function<void(const std::vector<bool>&)> doneCallback)
{
    if (!forId.IsValid() || !VerifyVerificationVectors(vvecs)) {
        std::vector<bool> result;
        result.assign(vvecs.size(), false);
        doneCallback(result);
        return;
    }

    auto verifier = new ContributionVerifier(forId, vvecs, skShares, 8, parallel, aggregated, workerPool, std::move(doneCallback));
    verifier->Start();
}

std::future<std::vector<bool> > CBLSWorker::AsyncVerifyContributionShares(const CBLSId& forId, const std::vector<BLSVerificationVectorPtr>& vvecs, const BLSSecretKeyVector& skShares,
                                                                          bool parallel, bool aggregated)
{
    auto p = BuildFutureDoneCallback<std::vector<bool> >();
    AsyncVerifyContributionShares(forId, vvecs, skShares, parallel, aggregated, std::move(p.first));
    return std::move(p.second);
}

std::vector<bool> CBLSWorker::VerifyContributionShares(const CBLSId& forId, const std::vector<BLSVerificationVectorPtr>& vvecs, const BLSSecretKeyVector& skShares,
                                                       bool parallel, bool aggregated)
{
    return AsyncVerifyContributionShares(forId, vvecs, skShares, parallel, aggregated).get();
}

std::future<bool> CBLSWorker::AsyncVerifyContributionShare(const CBLSId& forId,
                                                           const BLSVerificationVectorPtr& vvec,
                                                           const CBLSSecretKey& skContribution)
{
    if (!forId.IsValid() || !VerifyVerificationVector(*vvec)) {
        auto p = BuildFutureDoneCallback<bool>();
        p.first(false);
        return std::move(p.second);
    }

    auto f = [this, &forId, &vvec, &skContribution](int threadId) {
        CBLSPublicKey pk1;
        if (!pk1.PublicKeyShare(*vvec, forId)) {
            return false;
        }

        CBLSPublicKey pk2 = skContribution.GetPublicKey();
        return pk1 == pk2;
    };
    return workerPool.push(f);
}

bool CBLSWorker::VerifyContributionShare(const CBLSId& forId, const BLSVerificationVectorPtr& vvec,
                                         const CBLSSecretKey& skContribution)
{
    CBLSPublicKey pk1;
    if (!pk1.PublicKeyShare(*vvec, forId)) {
        return false;
    }

    CBLSPublicKey pk2 = skContribution.GetPublicKey();
    return pk1 == pk2;
}

bool CBLSWorker::VerifyVerificationVector(const BLSVerificationVector& vvec, size_t start, size_t count)
{
    return VerifyVectorHelper(vvec, start, count);
}

bool CBLSWorker::VerifyVerificationVectors(const std::vector<BLSVerificationVectorPtr>& vvecs,
                                           size_t start, size_t count)
{
    if (start == 0 && count == 0) {
        count = vvecs.size();
    }

    std::set<uint256> set;
    for (size_t i = 0; i < count; i++) {
        auto& vvec = vvecs[start + i];
        if (vvec == nullptr) {
            return false;
        }
        if (vvec->size() != vvecs[start]->size()) {
            return false;
        }
        for (size_t j = 0; j < vvec->size(); j++) {
            if (!(*vvec)[j].IsValid()) {
                return false;
            }
            // check duplicates
            if (!set.emplace((*vvec)[j].GetHash()).second) {
                return false;
            }
        }
    }

    return true;
}

bool CBLSWorker::VerifySecretKeyVector(const BLSSecretKeyVector& secKeys, size_t start, size_t count)
{
    return VerifyVectorHelper(secKeys, start, count);
}

bool CBLSWorker::VerifySignatureVector(const BLSSignatureVector& sigs, size_t start, size_t count)
{
    return VerifyVectorHelper(sigs, start, count);
}

void CBLSWorker::AsyncSign(const CBLSSecretKey& secKey, const uint256& msgHash, CBLSWorker::SignDoneCallback doneCallback)
{
    workerPool.push([secKey, msgHash, doneCallback](int threadId) {
        doneCallback(secKey.Sign(msgHash));
    });
}

std::future<CBLSSignature> CBLSWorker::AsyncSign(const CBLSSecretKey& secKey, const uint256& msgHash)
{
    auto p = BuildFutureDoneCallback<CBLSSignature>();
    AsyncSign(secKey, msgHash, std::move(p.first));
    return std::move(p.second);
}

void CBLSWorker::AsyncVerifySig(const CBLSSignature& sig, const CBLSPublicKey& pubKey, const uint256& msgHash,
                                CBLSWorker::SigVerifyDoneCallback doneCallback, CancelCond cancelCond)
{
    if (!sig.IsValid() || !pubKey.IsValid()) {
        doneCallback(false);
        return;
    }

    std::unique_lock<std::mutex> l(sigVerifyMutex);

    bool foundDuplicate = false;
    for (auto& s : sigVerifyQueue) {
        if (s.msgHash == msgHash) {
            foundDuplicate = true;
            break;
        }
    }

    if (foundDuplicate) {
        // batched/aggregated verification does not allow duplicate hashes, so we push what we currently have and start
        // with a fresh batch
        PushSigVerifyBatch();
    }

    sigVerifyQueue.emplace_back(std::move(doneCallback), std::move(cancelCond), sig, pubKey, msgHash);
    if (sigVerifyBatchesInProgress == 0 || sigVerifyQueue.size() >= SIG_VERIFY_BATCH_SIZE) {
        PushSigVerifyBatch();
    }
}

std::future<bool> CBLSWorker::AsyncVerifySig(const CBLSSignature& sig, const CBLSPublicKey& pubKey, const uint256& msgHash, CancelCond cancelCond)
{
    auto p = BuildFutureDoneCallback2<bool>();
    AsyncVerifySig(sig, pubKey, msgHash, std::move(p.first), cancelCond);
    return std::move(p.second);
}

bool CBLSWorker::IsAsyncVerifyInProgress()
{
    std::unique_lock<std::mutex> l(sigVerifyMutex);
    return sigVerifyBatchesInProgress != 0;
}

// sigVerifyMutex must be held while calling
void CBLSWorker::PushSigVerifyBatch()
{
    auto f = [this](int threadId, std::shared_ptr<std::vector<SigVerifyJob> > _jobs) {
        auto& jobs = *_jobs;
        if (jobs.size() == 1) {
            auto& job = jobs[0];
            if (!job.cancelCond()) {
                bool valid = job.sig.VerifyInsecure(job.pubKey, job.msgHash);
                job.doneCallback(valid);
            }
            std::unique_lock<std::mutex> l(sigVerifyMutex);
            sigVerifyBatchesInProgress--;
            if (!sigVerifyQueue.empty()) {
                PushSigVerifyBatch();
            }
            return;
        }

        CBLSSignature aggSig;
        std::vector<size_t> indexes;
        std::vector<CBLSPublicKey> pubKeys;
        std::vector<uint256> msgHashes;
        indexes.reserve(jobs.size());
        pubKeys.reserve(jobs.size());
        msgHashes.reserve(jobs.size());
        for (size_t i = 0; i < jobs.size(); i++) {
            auto& job = jobs[i];
            if (job.cancelCond()) {
                continue;
            }
            if (pubKeys.empty()) {
                aggSig = job.sig;
            } else {
                aggSig.AggregateInsecure(job.sig);
            }
            indexes.emplace_back(i);
            pubKeys.emplace_back(job.pubKey);
            msgHashes.emplace_back(job.msgHash);
        }

        if (!pubKeys.empty()) {
            bool allValid = aggSig.VerifyInsecureAggregated(pubKeys, msgHashes);
            if (allValid) {
                for (size_t i = 0; i < pubKeys.size(); i++) {
                    jobs[indexes[i]].doneCallback(true);
                }
            } else {
                // one or more sigs were not valid, revert to per-sig verification
                // TODO this could be improved if we would cache pairing results in some way as the previous aggregated verification already calculated all the pairings for the hashes
                for (size_t i = 0; i < pubKeys.size(); i++) {
                    auto& job = jobs[indexes[i]];
                    bool valid = job.sig.VerifyInsecure(job.pubKey, job.msgHash);
                    job.doneCallback(valid);
                }
            }
        }

        std::unique_lock<std::mutex> l(sigVerifyMutex);
        sigVerifyBatchesInProgress--;
        if (!sigVerifyQueue.empty()) {
            PushSigVerifyBatch();
        }
    };

    auto batch = std::make_shared<std::vector<SigVerifyJob> >(std::move(sigVerifyQueue));
    sigVerifyQueue.reserve(SIG_VERIFY_BATCH_SIZE);

    sigVerifyBatchesInProgress++;
    workerPool.push(f, batch);
}