* Copyright (c) 2023 Huawei Device Co., Ltd.
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License 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 PREFERENCES_FRAMEWORKS_EXECUTOR_H
#define PREFERENCES_FRAMEWORKS_EXECUTOR_H
#include <condition_variable>
#include <mutex>
#include <queue>
#include <thread>
#include "preferences_thread.h"
#include "priority_queue.h"
namespace OHOS {
namespace NativePreferences {
class Executor : public std::enable_shared_from_this<Executor> {
public:
using TaskId = uint64_t;
using Task = std::function<void()>;
using Duration = std::chrono::steady_clock::duration;
using Time = std::chrono::steady_clock::time_point;
static constexpr Time INVALID_TIME = std::chrono::time_point<std::chrono::steady_clock, std::chrono::seconds>();
static constexpr Duration INVALID_INTERVAL = std::chrono::milliseconds(0);
static constexpr uint64_t UNLIMITED_TIMES = std::numeric_limits<uint64_t>::max();
static constexpr Duration INVALID_DELAY = std::chrono::seconds(0);
static constexpr TaskId INVALID_TASK_ID = static_cast<uint64_t>(0l);
enum Status {
RUNNING,
IS_STOPPING,
STOPPED
};
struct InnerTask {
std::function<void()> exec = []() {};
Duration interval = INVALID_INTERVAL;
uint64_t times = UNLIMITED_TIMES;
TaskId taskId = INVALID_TASK_ID;
InnerTask() = default;
bool Valid() const
{
return taskId != INVALID_TASK_ID;
}
};
Executor(const std::string &name)
: thread_([this, name] {
auto realName = std::string("task_queue_") + name;
PthreadSetNameNp(realName);
Run();
self_ = nullptr;
})
{
thread_.detach();
}
Executor()
: thread_([this] {
PthreadSetNameNp("Executor");
Run();
self_ = nullptr;
})
{
thread_.detach();
}
void Bind(PriorityQueue<InnerTask, Time, TaskId> *queue, std::function<bool(std::shared_ptr<Executor>)> idle,
std::function<bool(std::shared_ptr<Executor>, bool)> release)
{
std::unique_lock<decltype(mutex_)> lock(mutex_);
self_ = shared_from_this();
waits_ = queue;
idle_ = std::move(idle);
release_ = std::move(release);
condition_.notify_one();
}
void Stop(bool wait = false) noexcept
{
std::unique_lock<decltype(mutex_)> lock(mutex_);
running_ = IS_STOPPING;
condition_.notify_one();
cond_.wait(lock, [this, wait]() { return !wait || running_ == STOPPED; });
}
private:
static constexpr Duration TIME_OUT = std::chrono::seconds(2);
void Run()
{
std::unique_lock<decltype(mutex_)> lock(mutex_);
do {
do {
condition_.wait(lock, [this] {
return running_ == IS_STOPPING || waits_ != nullptr;
});
while (running_ == RUNNING && waits_ != nullptr && waits_->Size() > 0) {
auto currentTask = waits_->Pop();
lock.unlock();
currentTask.exec();
lock.lock();
waits_->Finish(currentTask.taskId);
}
if (!idle_(self_) && running_ == RUNNING) {
continue;
}
waits_ = nullptr;
} while (running_ == RUNNING &&
condition_.wait_until(lock, std::chrono::steady_clock::now() + TIME_OUT, [this]() {
return waits_ != nullptr;
}));
} while (!release_(self_, running_ == IS_STOPPING));
running_ = STOPPED;
cond_.notify_all();
}
Status running_ = RUNNING;
std::mutex mutex_;
std::condition_variable condition_;
std::condition_variable cond_;
std::shared_ptr<Executor> self_;
PriorityQueue<InnerTask, Time, TaskId> *waits_ = nullptr;
std::function<bool(std::shared_ptr<Executor>)> idle_;
std::function<bool(std::shared_ptr<Executor>, bool)> release_;
std::thread thread_;
};
}
}
#endif