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/ExecutorInterface.h>
#include <AVSCommon/Utils/Threading/TaskThread.h>
#include <AVSCommon/Utils/Power/PowerResource.h>

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

class Executor {
public:
    Executor() noexcept;

    Executor(const std::chrono::milliseconds& unused) noexcept;

    ~Executor() noexcept;

    bool execute(std::function<void()>&& function) noexcept;

    bool execute(const std::function<void()>& function) noexcept;

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

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

    void waitForSubmittedTasks() noexcept;

    void shutdown() noexcept;

    bool isShutdown() noexcept;

    operator std::shared_ptr<ExecutorInterface>() const noexcept;

private:
    // Friend declaration.
    friend class SharedExecutor;

    enum class QueuePosition {
        Front = 1,
        Back
    };

    bool execute(std::function<void()>&& function, QueuePosition queuePosition) noexcept;

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

    template <typename T>
    std::future<T> pushFunction(QueuePosition queuePosition, std::function<T()>&& function) noexcept;

    std::shared_ptr<class SharedExecutor> m_executor;
};

inline Executor::Executor(const std::chrono::milliseconds&) noexcept : Executor() {
}

template <typename Task, typename... Args>
auto Executor::submit(Task task, Args&&... args) noexcept -> std::future<decltype(task(args...))> {
    return pushTo(QueuePosition::Back, std::forward<Task>(task), std::forward<Args>(args)...);
}

template <typename Task, typename... Args>
auto Executor::submitToFront(Task task, Args&&... args) noexcept -> std::future<decltype(task(args...))> {
    return pushTo(QueuePosition::Front, std::forward<Task>(task), std::forward<Args>(args)...);
}

template <typename T>
inline static void forwardPromise(std::promise<T>& promise, std::future<T>& future) noexcept {
#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::promise<void>& promise, std::future<void>& future) noexcept {
#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>
inline auto Executor::pushTo(QueuePosition queuePosition, Task&& task, Args&&... args) noexcept
    -> std::future<decltype(task(args...))> {
    using ValueType = decltype(task(args...));
    // Remove arguments from the tasks type by binding the arguments to the task.
    std::function<ValueType()> fn{std::bind(std::forward<Task>(task), std::forward<Args>(args)...)};
    return pushFunction(queuePosition, std::move(fn));
}

template <typename T>
std::future<T> Executor::pushFunction(QueuePosition queuePosition, std::function<T()>&& function) noexcept {
    /*
     * Create a std::packaged_task with the correct return 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.
     */

    struct CallCtx {
        inline CallCtx(std::function<T()>&& function) : packagedTask{std::move(function)} {
        }

        std::packaged_task<T()> packagedTask;
        std::promise<T> cleanupPromise;
    };

    auto callCtx = std::make_shared<CallCtx>(std::move(function));

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

    // Create a promise/future that we will fulfill when we have cleaned up the task.
    auto cleanupFuture = callCtx->cleanupPromise.get_future();

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

    if (!execute(std::move(translated_task), queuePosition)) {
        return std::future<T>();
    }

    return cleanupFuture;
}

extern template std::future<void> Executor::pushFunction(
    QueuePosition queuePosition,
    std::function<void()>&& function) noexcept;
extern template std::future<bool> Executor::pushFunction(
    QueuePosition queuePosition,
    std::function<bool()>&& function) noexcept;
extern template std::future<std::string> Executor::pushFunction(
    QueuePosition queuePosition,
    std::function<std::string()>&& function) noexcept;

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

#endif  // ALEXA_CLIENT_SDK_AVSCOMMON_UTILS_INCLUDE_AVSCOMMON_UTILS_THREADING_EXECUTOR_H_