Executor.h

Go to the documentation of this file.

/*
 * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License").
 * You may not use this file except in compliance with the License.
 * A copy of the License is located at
 *
 *     http://aws.amazon.com/apache2.0/
 *
 * or in the "license" file accompanying this file. This file is distributed
 * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
 * express or implied. See the License for the specific language governing
 * permissions and limitations under the License.
 */

#ifndef ALEXA_CLIENT_SDK_AVSCOMMON_UTILS_INCLUDE_AVSCOMMON_UTILS_THREADING_EXECUTOR_H_
#define ALEXA_CLIENT_SDK_AVSCOMMON_UTILS_INCLUDE_AVSCOMMON_UTILS_THREADING_EXECUTOR_H_

#include <atomic>
#include <condition_variable>
#include <chrono>
#include <deque>
#include <functional>
#include <future>
#include <memory>
#include <mutex>
#include <utility>

#include "AVSCommon/Utils/Threading/TaskThread.h"
#include "AVSCommon/Utils/Power/PowerResource.h"

namespace alexaClientSDK {
namespace avsCommon {
namespace utils {
namespace threading {

class Executor {
public:
    Executor(const std::chrono::milliseconds& delayExit = std::chrono::milliseconds(1000));

    ~Executor();

    template <typename Task, typename... Args>
    auto submit(Task task, Args&&... args) -> std::future<decltype(task(args...))>;

    template <typename Task, typename... Args>
    auto submitToFront(Task task, Args&&... args) -> std::future<decltype(task(args...))>;

    void waitForSubmittedTasks();

    void shutdown();

    bool isShutdown();

private:
    using Queue = std::deque<std::function<void()>>;

    bool runNext();

    bool hasNext();

    std::function<void()> pop();

    template <typename Task, typename... Args>
    auto pushTo(bool front, Task task, Args&&... args) -> std::future<decltype(task(args...))>;

    Queue m_queue;

    bool m_threadRunning;

    std::chrono::milliseconds m_timeout;

    std::mutex m_queueMutex;

    std::atomic_bool m_shutdown;

    std::shared_ptr<power::PowerResource> m_powerResource;

    std::condition_variable m_delayedCondition;

    const uint64_t m_id;

    TaskThread m_taskThread;
};

template <typename Task, typename... Args>
auto Executor::submit(Task task, Args&&... args) -> std::future<decltype(task(args...))> {
    bool front = false;
    return pushTo(front, std::forward<Task>(task), std::forward<Args>(args)...);
}

template <typename Task, typename... Args>
auto Executor::submitToFront(Task task, Args&&... args) -> std::future<decltype(task(args...))> {
    bool front = true;
    return pushTo(front, std::forward<Task>(task), std::forward<Args>(args)...);
}

template <typename T>
inline static void forwardPromise(std::shared_ptr<std::promise<T>> promise, std::future<T>* future) {
#if __cpp_exceptions || defined(__EXCEPTIONS)
    try {
#endif
        promise->set_value(future->get());
#if __cpp_exceptions || defined(__EXCEPTIONS)
    } catch (...) {
        promise->set_exception(std::current_exception());
    }
#endif
}

template <>
inline void forwardPromise<void>(std::shared_ptr<std::promise<void>> promise, std::future<void>* future) {
#if __cpp_exceptions || defined(__EXCEPTIONS)
    try {
#endif
        future->get();
        promise->set_value();
#if __cpp_exceptions || defined(__EXCEPTIONS)
    } catch (...) {
        promise->set_exception(std::current_exception());
    }
#endif
}

template <typename Task, typename... Args>
auto Executor::pushTo(bool front, Task task, Args&&... args) -> std::future<decltype(task(args...))> {
    // Remove arguments from the tasks type by binding the arguments to the task.
    auto boundTask = std::bind(std::forward<Task>(task), std::forward<Args>(args)...);

    /*
     * Create a std::packaged_task with the correct return type. The decltype only returns the return value of the
     * boundTask. The following parentheses make it a function call with the boundTask return type. The package task
     * will then return a future of the correct type.
     *
     * Note: A std::packaged_task fulfills its future *during* the call to operator().  If the user of a
     * std::packaged_task hands it off to another thread to execute, and then waits on the future, they will be able to
     * retrieve the return value from the task and know that the task has executed, but they do not know exactly when
     * the task object has been deleted.  This distinction can be significant if the packaged task is holding onto
     * resources that need to be freed (through a std::shared_ptr for example).  If the user needs to wait for those
     * resources to be freed they have no way of knowing how long to wait.  The translated_task lambda below is a
     * workaround for this limitation.  It executes the packaged task, then disposes of it before passing the task's
     * return value back to the future that the user is waiting on.
     */
    using PackagedTaskType = std::packaged_task<decltype(boundTask())()>;
    auto packaged_task = std::make_shared<PackagedTaskType>(boundTask);

    // Create a promise/future that we will fulfill when we have cleaned up the task.
    auto cleanupPromise = std::make_shared<std::promise<decltype(task(args...))>>();
    auto cleanupFuture = cleanupPromise->get_future();

    // Remove the return type from the task by wrapping it in a lambda with no return value.
    auto translated_task = [packaged_task, cleanupPromise]() mutable {
        // Execute the task.
        packaged_task->operator()();
        // Note the future for the task's result.
        auto taskFuture = packaged_task->get_future();
        // Clean up the task.
        packaged_task.reset();
        // Forward the task's result to our cleanup promise/future.
        forwardPromise(cleanupPromise, &taskFuture);
    };

    // Release our local reference to packaged task so that the only remaining reference is inside the lambda.
    packaged_task.reset();

    {
        bool restart = false;
        std::lock_guard<std::mutex> queueLock{m_queueMutex};
        if (!m_shutdown) {
            restart = !m_threadRunning;
            if (m_powerResource) {
                m_powerResource->acquire();
            }
            m_queue.emplace(front ? m_queue.begin() : m_queue.end(), std::move(translated_task));
        } else {
            using FutureType = decltype(task(args...));
            return std::future<FutureType>();
        }

        if (restart) {
            // Restart task thread.
            m_taskThread.start(std::bind(&Executor::runNext, this));
            m_threadRunning = true;
        }
    }

    m_delayedCondition.notify_one();
    return cleanupFuture;
}

}  // namespace threading
}  // namespace utils
}  // namespace avsCommon
}  // namespace alexaClientSDK

#endif  // ALEXA_CLIENT_SDK_AVSCOMMON_UTILS_INCLUDE_AVSCOMMON_UTILS_THREADING_EXECUTOR_H_