#include #include #include #include #include "LibapiThread.h" #include "LibapiQueue.h" #include "LibapiQueues.h" #include using namespace std; static int id_max = -1; LibapiThread::LibapiThread(int id, std::string name, BaseRunnable* runnable) : m_pthread(nullptr), m_pauseFlag(false), m_stopFlag(false), m_state(Stoped), m_id(id), m_name(name), m_prunnable(runnable) { if (m_id == -1) { m_id = id_max + 1; id_max++; } m_queue = Queues::create_queue(m_id, m_name); } LibapiThread::~LibapiThread() { Queues::delete_queue(this->m_queue); stop(); } LibapiThread::State LibapiThread::state() const { return m_state; } void LibapiThread::start() { if (m_pthread == nullptr) { std::cout << this->get_name() << " thread start! " << std::endl; //spdlog::info(this->get_name() + "thread start!"); m_pthread = new thread(&LibapiThread::run, this); m_pauseFlag = false; m_stopFlag = false; m_state = Running; } } void LibapiThread::stop() { if (m_pthread != nullptr) { m_pauseFlag = false; m_stopFlag = true; m_condition.notify_one(); // Notify one waiting thread, if there is one. m_pthread->join(); // wait for thread finished delete m_pthread; m_pthread = nullptr; m_state = Stoped; } } void LibapiThread::join() { if (m_pthread != nullptr) { m_pthread->join(); // wait for thread finished } } void LibapiThread::pause() { if (m_pthread != nullptr) { m_pauseFlag = true; m_state = Paused; } } void LibapiThread::resume() { if (m_pthread != nullptr) { m_pauseFlag = false; m_condition.notify_all(); m_state = Running; } } bool LibapiThread::push(void* pMsg) { return m_queue->push(pMsg); } void* LibapiThread::pop() { std::cout<<"LibapiThread::pop()"<pop(p); return p; } void LibapiThread::notify() { m_condition.notify_one(); } void LibapiThread::run() { //cout << "enter thread:" << this_thread::get_id() << endl; cout << "enter thread:" << get_id() << endl; //spdlog::info("enter thread:%d", get_id()); while (!m_stopFlag) { if (/*!m_queue->empty() && */m_prunnable!=NULL) { void* pMsg = NULL; m_queue->pop(pMsg); m_prunnable->OnProcess(this, pMsg); } if (m_pauseFlag) { unique_lock locker(m_mutex); while (m_pauseFlag) { m_condition.wait(locker); // Unlock m_mutex and wait to be notified } locker.unlock(); } } m_pauseFlag = false; m_stopFlag = false; //cout << "exit thread:" << this_thread::get_id() << endl; cout << "exit thread:" << get_id() << endl; //spdlog::info("exit thread:%d", get_id()); } int LibapiThread::get_id() { return this->m_id; } void LibapiThread::set_id(int& id) { this->m_id = id; } std::string LibapiThread::get_name() { return this->m_name; } void LibapiThread::set_name(std::string& name) { this->m_name = name; } // 阻塞式的SLEEP void LibapiThread::delay_microseconds(int microseconds) { #ifndef _WIN32 struct timespec req, rem; req.tv_sec = microseconds / 1000000L; req.tv_nsec = (microseconds % 1000000L) * 1000L; while ((nanosleep(&req, &rem) == -1) && (errno == EINTR)) { req.tv_sec = rem.tv_sec; req.tv_nsec = rem.tv_nsec; } #endif auto start = std::chrono::system_clock::now(); while (true) { auto duration = std::chrono::duration_cast(std::chrono::system_clock::now() - start).count(); if (duration > microseconds) { //LOGGING_ERROR("timeout occurred,timeout %d ms", timeout_ms); break; } //std::this_thread::sleep_for(std::chrono::nanoseconds(1000)); } } // 消耗cpu资源,定时器很准,最高精度1ms void LibapiThread::delay_second(double second) { clock_t start_time; start_time = clock(); for (; (clock() - start_time) < second * CLOCKS_PER_SEC;); } LibapiQueue* LibapiThread::get_queue() { return m_queue; }