emu sync uv

emu-sync-uv.cpp

#include "uv.h"
#include <iostream>
#include <memory>
#include <mutex>
#include <optional>
#include <semaphore>
#include <thread>
#include <unistd.h>
#include <vector>

class Task {
private:
  std::counting_semaphore<1> sig;
public:
  Task(): sig(0) {}

  ~Task() {}

  void Wait() {
    sig.acquire();
  }

  void Done() {
    sig.release();
  }
};

class Runner {
public:
  std::vector<std::shared_ptr<Task>> tasks_;
  std::mutex tasks_mutex_;
  
  uv_loop_t *worker_loop_;

  void Queue(std::shared_ptr<Task> task) {
    std::lock_guard<std::mutex> lock(tasks_mutex_);
    tasks_.emplace_back(task);
  }

  std::optional<std::shared_ptr<Task>> Dequeue() {
    std::lock_guard<std::mutex> lock(tasks_mutex_);

    if (tasks_.empty()) {
      return {};
    }

    auto firstTask = tasks_.front();
    tasks_.erase(tasks_.begin());

    return firstTask;
  }

  void StartAsyncWorker() {
    worker_loop_ = uv_loop_new();
    worker_loop_->data = this;

    std::thread worker_thread([this]() {
      uv_timer_t check_timer;

      uv_timer_init(worker_loop_, &check_timer);

      check_timer.data = this;

      uv_timer_start(&check_timer, [](uv_timer_t *check_timer) {

        auto runner = reinterpret_cast<Runner*>(check_timer->data);

        std::optional<std::shared_ptr<Task>> task;
        while((task = runner->Dequeue()).has_value()) {
          std::cout << "release one yo\n";
          auto t = task->get();
          t->Done();
        }
        
      }, 0, 5000);
      
      uv_run(worker_loop_, UV_RUN_DEFAULT);
      std::cout << "worker thread uv loop finished\n";
    });

    worker_thread.detach();
  }
};

int main() {
  uv_loop_t* loop = uv_default_loop();

  auto runner = new Runner();
  runner->StartAsyncWorker();

  uv_poll_t stdin_poll_handle;
  uv_poll_init(loop, &stdin_poll_handle, STDIN_FILENO);

  stdin_poll_handle.data = runner;

  uv_poll_start(&stdin_poll_handle, UV_READABLE, [](uv_poll_t *handle, int status, int events) {
    auto runner = reinterpret_cast<Runner*>(handle->data);
    if (status < 0) {
      std::cerr << "stdin error with err: " << uv_strerror(status) << std::endl;
      return;
    }

    if (events & UV_READABLE) {
      char buf[1024];
      ssize_t nread = 0;

      while ((nread = read(STDIN_FILENO, buf, sizeof(buf))) > 0) {
        std::cout << "Received " << nread << " bytes from stdin: " << buf;
        std::cout << "Creating new task and block this thread" << buf;

        std::shared_ptr<Task> t = std::make_shared<Task>();
        runner->Queue(t);
        t->Wait();
        std::cout << "DDDDDOOOOONNNNEEE\n";
      }
      std::cout << std::endl;
    }
  });


  uv_run(loop, UV_RUN_DEFAULT);
}