image_framework_ymj/image_framework/thead/LibapiThread.cpp

#include <iostream>
#include <time.h>
#include <stdio.h>
#include <chrono>
#include "LibapiThread.h"
#include "LibapiQueue.h"
#include "LibapiQueues.h"
#include <iostream>

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()"<<std::endl;
	void* p = NULL;
	m_queue->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<mutex> 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::milliseconds>(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<void*>* LibapiThread::get_queue()
{
	return m_queue;
}