struct Job{

Input input;

Output output;

JobAdditional additional;

MonopolyAllocator<TRT::Tensor>::MonopolyDataPointer mono_tensor;

std::shared_ptr<std::promise<Output>> pro;

};

virtual ~InferController(){

stop();

}

void stop(){

run_ = false;

cond_.notify_all();

////////////////////////////////////////// cleanup jobs

{

std::unique_lock<std::mutex> l(jobs_lock_);

while(!jobs_.empty()){

auto& item = jobs_.front();

if(item.pro)

item.pro->set_value(Output());

jobs_.pop();

}

};

if(worker_){

worker_->join();

worker_.reset();

}

}

bool startup(const StartParam& param){

run_ = true;

std::promise<bool> pro; //* kp: promise and future: https://www.youtube.com/watch?v=XDZkyQVsbDY

start_param_ = param; // trtmodel file name and gpuid

worker_ = std::make_shared<std::thread>(&InferController::worker, this, std::ref(pro)); //* kp: shared_ptr (highly recommended)ref: https://www.geeksforgeeks.org/auto_ptr-unique_ptr-shared_ptr-weak_ptr-2/#:~:text=unique_ptr%20to%20another.-,shared_ptr,-A%20shared_ptr%20is

//* kp: std::thread Here we spawn a new thread that calls aref: https://www.cplusplus.com/reference/thread/thread/?kw=thread

// how to construct a thread with args //*ref: https://www.cplusplus.com/reference/thread/thread/thread/#:~:text=A%20pointer%20to%20function

return pro.get_future().get(); // the line 'result.set_value(true);' in yolo.cpp will be returned here.

}

virtual std::shared_future<Output> commit(const Input& input){

Job job;

job.pro = std::make_shared<std::promise<Output>>();

if(!preprocess(job, input)){

job.pro->set_value(Output());

return job.pro->get_future();

}

///////////////////////////////////////////////////////////

{

std::unique_lock<std::mutex> l(jobs_lock_); //* kp: unique_lock It unblocks when it goes out of the scope (curly bracket).

jobs_.push(job);

};

cond_.notify_one();

return job.pro->get_future();

}

virtual std::vector<std::shared_future<Output>> commits(const std::vector<Input>& inputs){

int batch_size = std::min((int)inputs.size(), this->tensor_allocator_->capacity());

std::vector<Job> jobs(inputs.size());

std::vector<std::shared_future<Output>> results(inputs.size()); // intputs.size: 56 batch_size = monopoly_size:16*2 = 32

int nepoch = (inputs.size() + batch_size - 1) / batch_size; // if the num of inputs are bigger than batch size, several epoch are needed.

for(int epoch = 0; epoch < nepoch; ++epoch){

int begin = epoch * batch_size;

int end = std::min((int)inputs.size(), begin + batch_size); // after we set the begin and end point for this batch

for(int i = begin; i < end; ++i){ // we for loop this batch in one epoch

Job& job = jobs[i];

job.pro = std::make_shared<std::promise<Output>>();

if(!preprocess(job, inputs[i])){

job.pro->set_value(Output());

}

results[i] = job.pro->get_future();

}

///////////////////////////////////////////////////////////

{

std::unique_lock<std::mutex> l(jobs_lock_);

for(int i = begin; i < end; ++i){

jobs_.emplace(std::move(jobs[i]));

};

}

cond_.notify_one(); //* kp: notify_one and wait is a pair.

}

return results;

}

virtual void worker(std::promise<bool>& result) = 0; //* kp: virtual function

virtual bool preprocess(Job& job, const Input& input) = 0;

virtual bool get_jobs_and_wait(std::vector<Job>& fetch_jobs, int max_size){

std::unique_lock<std::mutex> l(jobs_lock_);

cond_.wait(l, [&](){

return !run_ || !jobs_.empty();

}); //* kp : condition variable and lambda expression.

if(!run_) return false;

fetch_jobs.clear();

for(int i = 0; i < max_size && !jobs_.empty(); ++i){

fetch_jobs.emplace_back(std::move(jobs_.front()));

jobs_.pop();

}

return true;

}

virtual bool get_job_and_wait(Job& fetch_job){

std::unique_lock<std::mutex> l(jobs_lock_);

cond_.wait(l, [&](){ // Wait causes the current thread to block until the condition variable is notified.

return !run_ || !jobs_.empty(); // When the pred func returns trueref: https://en.cppreference.com/w/cpp/thread/condition_variable/wait

}); // Waiting will be over if return true.

if(!run_) return false;

fetch_job = std::move(jobs_.front());

jobs_.pop();

return true;

}

StartParam start_param_;

std::atomic<bool> run_;

std::mutex jobs_lock_;

std::queue<Job> jobs_;

std::shared_ptr<std::thread> worker_;

std::condition_variable cond_;

std::shared_ptr<MonopolyAllocator<TRT::Tensor>> tensor_allocator_;