#ifdef UNSAFE_BUFFERS_BUILD
#pragma allow_unsafe_libc_calls
#endif
#ifndef MEDIA_BASE_CODEC_WORKER_IMPL_H_
#define MEDIA_BASE_CODEC_WORKER_IMPL_H_
#include <cstring>
#include "base/functional/bind.h"
#include "base/synchronization/lock.h"
#include "base/threading/thread.h"
#include "base/threading/thread_restrictions.h"
namespace media {
template <class WorkerInterface,
class WorkerImpl,
class Worker,
class WorkerStatus,
WorkerStatus StatusNotOk,
WorkerStatus StatusOk,
WorkerStatus StatusWork>
class CodecWorkerImpl {
public:
static constexpr WorkerInterface GetCodecWorkerInterface() {
return WorkerInterface{.init = Init,
.reset = Reset,
.sync = Sync,
.launch = Launch,
.execute = Execute,
.end = End};
}
private:
CodecWorkerImpl()
: thread_("CodecWorker"),
event_(base::WaitableEvent::ResetPolicy::AUTOMATIC) {
thread_.Start();
}
static CodecWorkerImpl* GetImpl(Worker* const worker) {
return reinterpret_cast<CodecWorkerImpl*>(worker->impl_);
}
static void Init(Worker* const worker) {
memset(worker, 0, sizeof(*worker));
worker->status_ = StatusNotOk;
worker->impl_ = nullptr;
}
static void Execute(Worker* const worker) {
if (worker->hook) {
worker->had_error |= !worker->hook(worker->data1, worker->data2);
}
}
void ChangeStateImpl(Worker* const worker, WorkerStatus new_status) {
bool deplete_work = false;
{
base::AutoLock lock(mutex_);
deplete_work = worker->status_ == StatusWork;
if (!deplete_work) {
worker->status_ = new_status;
}
}
if (deplete_work) {
thread_.task_runner()->PostTask(
FROM_HERE,
base::BindOnce([](base::WaitableEvent* event) { event->Signal(); },
base::Unretained(&event_)));
base::ScopedAllowBaseSyncPrimitives allow_wait;
event_.Wait();
base::AutoLock lock(mutex_);
worker->status_ = new_status;
}
if (new_status == StatusWork) {
thread_.task_runner()->PostTask(
FROM_HERE,
base::BindOnce(&CodecWorkerImpl::ExecuteOnTaskRunner,
base::Unretained(this), base::Unretained(worker)));
}
}
void ExecuteOnTaskRunner(Worker* worker) {
base::AutoLock lock(mutex_);
CHECK_EQ(worker->status_, StatusWork);
Execute(worker);
worker->status_ = StatusOk;
}
static void ChangeState(Worker* const worker, WorkerStatus new_status) {
if (!worker->impl_) {
return;
}
CodecWorkerImpl* impl = GetImpl(worker);
impl->ChangeStateImpl(worker, new_status);
}
static int Sync(Worker* const worker) {
ChangeState(worker, StatusOk);
return !worker->had_error;
}
static int Reset(Worker* const worker) {
if (worker->status_ < StatusOk) {
worker->had_error = false;
worker->impl_ = reinterpret_cast<WorkerImpl*>(new CodecWorkerImpl());
worker->status_ = StatusOk;
return 1;
}
int ok = Sync(worker);
worker->had_error = false;
CHECK(!ok || worker->status_ == StatusOk);
return ok;
}
static void Launch(Worker* const worker) { ChangeState(worker, StatusWork); }
static void End(Worker* const worker) {
if (worker->impl_) {
ChangeState(worker, StatusNotOk);
CodecWorkerImpl* impl = GetImpl(worker);
base::ScopedAllowBaseSyncPrimitives allow_wait;
delete impl;
worker->impl_ = nullptr;
}
}
base::Lock mutex_;
base::Thread thread_;
base::WaitableEvent event_;
};
}
#endif