seqan/align_parallel/wavefront_alignment_scheduler.h

// ==========================================================================
//                 SeqAn - The Library for Sequence Analysis
// ==========================================================================
// Copyright (c) 2006-2018, Knut Reinert, FU Berlin
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are met:
//
//     * Redistributions of source code must retain the above copyright
//       notice, this list of conditions and the following disclaimer.
//     * Redistributions in binary form must reproduce the above copyright
//       notice, this list of conditions and the following disclaimer in the
//       documentation and/or other materials provided with the distribution.
//     * Neither the name of Knut Reinert or the FU Berlin nor the names of
//       its contributors may be used to endorse or promote products derived
//       from this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
// AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
// ARE DISCLAIMED. IN NO EVENT SHALL KNUT REINERT OR THE FU BERLIN BE LIABLE
// FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
// DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
// SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
// CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
// LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
// OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH
// DAMAGE.
//
// ==========================================================================
// Author: Rene Rahn <rene.rahn@fu-berlin.de>
// ==========================================================================

#ifndef INCLUDE_SEQAN_ALIGN_PARALLEL_PARALLEL_ALIGNMENT_SCHEDULER_H_
#define INCLUDE_SEQAN_ALIGN_PARALLEL_PARALLEL_ALIGNMENT_SCHEDULER_H_

namespace seqan
{

// ============================================================================
// Forwards
// ============================================================================

// ============================================================================
// Tags, Classes, Enums
// ============================================================================

// Yet internal class. Might need some redesign to make it truly generic.
/*
 * @class WavefrontAlignmentScheduler
 * @headerfile <align_parallel.h>
 * @brief A generic scheduler allowing to execute callables with a ring buffer for the stored tasks.
 *
 * @signature class WavefrontAlignmentScheduler;
 *
 * This schedule is at the moment only used for the wave-front alignment execution but could be generalized later.
 * It stores all scheduled callables in a @link ConcurrentSuspendableQueue @endlink which can hold a user defined
 * number of callables at the same time. It then uses recycable ids to fill up the queue with waiting jobs.
 * If the queue is full and a thread tries to add a new job, it will be suspended, until resources are freed by
 * the scheduler.
 */
class WavefrontAlignmentScheduler
{
public:

    //-------------------------------------------------------------------------
    // Member Types.

    using TCallable         = std::function<void(uint16_t)>;
    using TAlignmentQueue   = ConcurrentQueue<TCallable, Suspendable<Limit>>;
    using TRecycleList      = std::list<uint16_t>;

    //-------------------------------------------------------------------------
    // Private Member Variables.

    WavefrontTaskScheduler  _taskScheduler;
    ThreadPool              _pool;
    TRecycleList            _recycableIds;
    TAlignmentQueue         _queue;
    bool                    _receivedEndSignal;

    std::mutex              _mutexRecycleId;
    unsigned                _numParallelAlignments;

    std::mutex                       _mutexPushException;
    std::vector<std::exception_ptr>  _exceptionPointers;

    std::atomic<bool>               _isValid{true};

    std::function<void()> job = [this] ()
    {
        while (true)
        {
            TCallable callable;
            if (!popFront(callable, _queue))
                break;  // End of thread => No writers and queue is empty.

            uint16_t id = -1;

            { // Receive id.
                std::lock_guard<std::mutex> lck(_mutexRecycleId);
                SEQAN_ASSERT_NOT(_recycableIds.empty());
                id = _recycableIds.front();
                _recycableIds.pop_front();
            }

            try
            {
                callable(id);  // invokes the alignment with assigned id.
            }
            catch (...)
            {  // Catch any exception thrown by callable. Store exception, and set *this invalid.
               // We still keep running until the queue is empty. The thread is cleaned either by,
               // explicit wait or by destruction of *this.
                _isValid.store(false, std::memory_order_release);
                {
                    std::lock_guard<std::mutex> lck(_mutexPushException);
                    _exceptionPointers.push_back(std::current_exception());
                }
            }

            // Check if task scheduler is still valid.
            // If not, something went wrong, and we should not continue adding new tasks.
            // So we propagate the invalid state to *this and break exceution chain.
            if (!isValid(_taskScheduler))
            {
                _isValid.store(false, std::memory_order_release);
            }

            { // recycle id, when done.
                std::lock_guard<std::mutex> lck(_mutexRecycleId);
                _recycableIds.push_back(id);
            }
        }
        unlockReading(_queue);  // Notify that this reader is finished.
        unlockWriting(_taskScheduler);  // Notify that this writer is finished.
    };

    //-------------------------------------------------------------------------
    // Constructors.

    // implicitly deleted default constructor.

    WavefrontAlignmentScheduler(size_t const numParallelAlignments, size_t const numParallelTasks) :
        _taskScheduler(numParallelTasks),
        _queue(numParallelAlignments),
        _receivedEndSignal(false),
        _numParallelAlignments(numParallelAlignments)
    {
        SEQAN_ASSERT_GT(numParallelAlignments, 0u);  // Bad if reader is 0.

        // Setup recycable ids.
        _recycableIds.resize(numParallelAlignments);
        std::iota(std::begin(_recycableIds), std::end(_recycableIds), 0);

        setReaderWriterCount(_queue, numParallelAlignments, 1);

        _exceptionPointers.resize(numParallelAlignments, nullptr);

        try
        { // Create the threads here, later we can try to make lazy thread creation.
            for (unsigned i = 0; i < numParallelAlignments; ++i)
            {
                spawn(_pool, job);
            }
        }
        catch (...)  // Make sure all the spawned threads are safely stopped before re-throwing the exception.
        {
            unlockWriting(_queue);
            waitForWriters(_taskScheduler);
            join(_pool);
            throw;
        }

        setWriterCount(_taskScheduler, numParallelAlignments);
        // Notify task scheduler, that everything was setup correctly.
        for (unsigned i = 0; i < numParallelAlignments; ++i)
        {
            lockWriting(_taskScheduler);
        }
        waitForWriters(_taskScheduler);  // Invoke task scheduler.
    }

    // Default constructor.
    WavefrontAlignmentScheduler() : WavefrontAlignmentScheduler(16, 8)
    {}

    // Copy & Move C'tor
    WavefrontAlignmentScheduler(WavefrontAlignmentScheduler const &) = delete;
    WavefrontAlignmentScheduler(WavefrontAlignmentScheduler &&)      = delete;

    ///-------------------------------------------------------------------------
    // Destructor.

    ~WavefrontAlignmentScheduler()
    {
        // Signal that no more alignments will be added.
        if (!_receivedEndSignal)
            unlockWriting(_queue);

        SEQAN_ASSERT(_queue.writerCount == 0);

        // Wait until all remaining threads are finished with their execution.
        join(_pool);

        // In destructor of thread pool we wait for the outstanding alignments to be finished
        // and then continue destruction of the remaining members and cleaning up the stack.
    }

    // ------------------------------------------------------------------------
    // Member Functions.

    // Copy & Move assignment
    WavefrontAlignmentScheduler& operator=(WavefrontAlignmentScheduler const &) = delete;
    WavefrontAlignmentScheduler& operator=(WavefrontAlignmentScheduler &&)      = delete;
};

// ============================================================================
// Metafunctions
// ============================================================================

template<>
struct SchedulerTraits<WavefrontAlignmentScheduler>
{
    using TTask = typename WavefrontAlignmentScheduler::TCallable;
};

// ============================================================================
// Functions
// ============================================================================

/*
 * @fn WavefrontAlignmentScheduler#isValid
 * @headerfile <align_parallel.h>
 * @brief Checks if scheduler is in a valid state. This means that no callable has terminated with an exception.
 */
inline bool
isValid(WavefrontAlignmentScheduler const & me)
{
    return me._isValid.load(std::memory_order_acquire);
}

/*
 * @fn WavefrontAlignmentScheduler#scheduleTask
 * @headerfile <align_parallel.h>
 * @brief Adds a new task to the scheduler. Suspends until resources become available.
 * @throws ExceptionType?
 */
// basic exception-safety guarantee.
// Throws if appendValue failed.
inline void
scheduleTask(WavefrontAlignmentScheduler & me,
             typename SchedulerTraits<WavefrontAlignmentScheduler>::TTask && callable)
{
    if (!isValid(me))
        throw std::runtime_error("Invalid alignment scheduler!");

    // Spins until there is enough space to add to the queue.
    if (!appendValue(me._queue, std::forward<decltype(callable)>(callable)))
        throw std::runtime_error("Invalid alignment scheduler 2!");
}

inline void
scheduleTask(WavefrontAlignmentScheduler & me,
             typename SchedulerTraits<WavefrontAlignmentScheduler>::TTask & callable)
{
    if (!isValid(me))
        throw std::runtime_error("Invalid alignment scheduler!");
    // Spins until there is enough space to add to the queue.
    if(!appendValue(me._queue, callable))
        throw std::runtime_error("Invalid alignment scheduler 2!");
}

/*
 * @fn WavefrontAlignmentScheduler#notify
 * @headerfile <align_parallel.h>
 * @brief Notify the scheduler that no more jobs will follow.
 */
inline void
notify(WavefrontAlignmentScheduler & me)
{
    unlockWriting(me._queue);
    me._receivedEndSignal = true;
}

/*
 * @fn WavefrontAlignmentScheduler#wait
 * @headerfile <align_parallel.h>
 * @brief Explicit barrier on the scheduler. Suspends until all scheduled jobs have been finsihed.
 *
 * Note, can dead lock if notify is never called.
 */
// Only possible if some other thread is signaling the end of it.
inline void
wait(WavefrontAlignmentScheduler & me)
{
    join(me._pool);
    wait(me._taskScheduler);
}

/*
 * @fn WavefrontAlignmentScheduler#wait2
 * @headerfile <align_parallel.h>
 * @brief Explicit barrier on the scheduler. Suspends until all scheduled jobs have been finsihed.
 *
 * Note, can dead lock if notify is never called.
 */
template <typename TNotifiable>
inline void
wait2(WavefrontAlignmentScheduler & me, TNotifiable & notifiable)
{
    join(me._pool);
    notify(notifiable);
    wait(me._taskScheduler);
}

/*
 * @fn WavefrontAlignmentScheduler#getExceptions
 * @headerfile <align_parallel.h>
 * @brief Returns vector of captured exceptions if any was thrown by the callable.
 *
 * Note, can dead lock if notify is never called.
 */
inline auto
getExceptions(WavefrontAlignmentScheduler & me)
{
    auto vec = me._exceptionPointers;
    auto innerExceptions = getExceptions(me._taskScheduler);
    std::copy(std::begin(innerExceptions), std::end(innerExceptions), std::back_inserter(vec));
    return vec;
}

/*
 * @fn WavefrontAlignmentScheduler#taskScheduler
 * @headerfile <align_parallel.h>
 * @brief Returns lvalue reference to the underlying task_scheduler.
 */
inline auto&
taskScheduler(WavefrontAlignmentScheduler & me)
{
    return me._taskScheduler;
}

}  // namespace seqan

#endif  // #ifndef INCLUDE_SEQAN_ALIGN_PARALLEL_PARALLEL_ALIGNMENT_SCHEDULER_H_