open source

2020-11-11 16:17:13 +08:00
parent 4d989f3ecb
commit bc4ca748de
2441 changed files with 623057 additions and 2 deletions
@@ -0,0 +1,420 @@
+/**
+@file			JCThreadPool.h
+@brief			
+@author			hugao
+@version		1.0
+@date			2016_5_12
+*/
+
+#ifndef __JCThreadPool_H__
+#define __JCThreadPool_H__
+
+#include <thread>
+#include <atomic>
+#include <vector>
+#include <list>
+#include "JCCondition.h"
+
+/** 
+ *  @brief JCThreadPoolShare
+          *包含若干线程
+          *一个数据列表 m_ShareData
+          *多个线程共同处理这个数据列表。 
+*/
+
+template<class _Tp>
+class JCThreadPoolShare
+{
+private:
+	JCThreadPoolShare(JCThreadPoolShare const&);
+	JCThreadPoolShare& operator=(JCThreadPoolShare const&);     
+
+protected:
+	typedef _Tp _DataType;
+
+	typedef std::list<_DataType>			__DataQueue;
+	typedef typename __DataQueue::iterator	__DataQueueIter;
+
+	struct __Thread_t
+	{
+		std::thread *m_pThread;
+		JCCondition *m_pCondition;
+
+		__DataQueue *m_pData;
+		bool m_bNeedFree;
+
+		explicit __Thread_t( std::thread *p_pThread, JCCondition *p_pCondition, __DataQueue *p_pData, bool p_bNeedFree )
+		{
+			m_pThread = p_pThread;
+			m_pCondition = p_pCondition;
+			m_pData = p_pData;
+			m_bNeedFree = p_bNeedFree;
+		}
+
+		~__Thread_t()
+		{
+			if( 0 != m_pThread )
+			{
+				delete m_pThread;
+				m_pThread = 0;
+			}
+
+			if( m_bNeedFree )
+			{
+				if( 0 != m_pCondition )
+				{
+					delete m_pCondition;
+					m_pCondition = 0;
+				}
+
+				if( 0 != m_pData )
+				{
+					m_pData->clear();
+					delete m_pData;
+					m_pData = 0;
+				}
+			}
+		}
+	};
+	typedef std::vector<__Thread_t *>			__ThreadQueue;
+	typedef typename __ThreadQueue::iterator	__ThreadQueueIter;
+
+	__ThreadQueue m_Threads;
+	std::recursive_mutex m_Mutex;
+
+	JCCondition m_ShareCondition;
+	__DataQueue m_ShareData;
+
+#ifdef WIN32
+    std::atomic_uint32_t m_bWantToStop;
+#else
+    std::atomic_uint m_bWantToStop;
+#endif
+
+	__Thread_t *__GetCurrentThreadContext()
+	{
+		__Thread_t *pCurThread = 0;
+
+		std::thread::id _CurID = std::this_thread::get_id();
+        
+        std::unique_lock<std::recursive_mutex> _lock(m_Mutex);
+
+		if( m_Threads.size() )
+		{
+			__Thread_t *p;
+			__ThreadQueueIter iter;
+			for(iter=m_Threads.begin();iter!=m_Threads.end();iter++)
+			{
+				p = *iter;
+				if( p->m_pThread->get_id() == _CurID )
+				{
+					pCurThread = p;
+					break;;
+				}
+			}
+		}
+
+		return pCurThread;
+	}
+
+	void __RemoveCurrentThread()
+	{
+		__Thread_t *pCurThread = 0;
+
+		std::thread::id _CurID = std::this_thread::get_id();
+
+		std::unique_lock<std::recursive_mutex> _lock(m_Mutex);//kuo
+
+		if( m_Threads.size() )
+		{
+			__Thread_t *p;
+			__ThreadQueueIter iter;
+			for(iter=m_Threads.begin();iter!=m_Threads.end();iter++)
+			{
+				p = *iter;
+				if( p->m_pThread->get_id() == _CurID )
+				{
+					m_Threads.erase( iter );
+					p->m_pThread->detach();
+					delete p;
+					break;
+				}
+			}
+		}
+	}
+
+	virtual int __AddThread( std::thread *p_pThread )
+	{
+		__Thread_t *p = new __Thread_t( p_pThread, &m_ShareCondition, &m_ShareData, false );
+
+		m_Threads.push_back( p );
+
+		return m_Threads.size()-1;
+	}
+
+	bool __SendData( _DataType p_Data, JCCondition *p_pCondition, __DataQueue *p_pData )
+	{
+		if( 0 != m_bWantToStop )
+		{
+			return false;
+		}
+
+		p_pCondition->Lock();
+		p_pData->push_back( p_Data );
+		p_pCondition->Unlock();
+
+		p_pCondition->NotifyOne();
+
+		return true;
+	}
+
+public:
+	JCThreadPoolShare()
+	{
+		m_bWantToStop = false;
+	}
+
+	~JCThreadPoolShare()
+	{
+		Stop();
+	}
+
+	size_t GetSize()
+	{
+		std::unique_lock<std::recursive_mutex> _lock(m_Mutex);
+
+		return m_Threads.size();
+	}
+
+	template<typename F>
+	int CreateThread( F p_pfnThreadFunc )
+	{
+		m_bWantToStop = 0;
+
+		std::unique_lock<std::recursive_mutex> _lock(m_Mutex);
+
+		std::thread *pThread = new std::thread(p_pfnThreadFunc);
+
+		if( 0 == pThread )
+			return -1;
+		else
+			return __AddThread( pThread );
+	}
+
+	void ThreadStopped()
+	{
+		if( 0 == m_bWantToStop )
+		{
+			__RemoveCurrentThread();
+		}
+	}
+
+	//取出当前线程的任务进行处理。
+	//根据运行时所在线程。
+	bool WaitData( _DataType *p_pData )
+	{
+		if( 0 == p_pData )
+			return false;
+
+		m_bWantToStop = 0;
+
+		__Thread_t *pContext = __GetCurrentThreadContext();
+		JCCondition *pCondition;
+		__DataQueue *pDataList;
+
+		if( 0 == pContext )
+		{
+			pCondition = &m_ShareCondition;
+			pDataList = &m_ShareData;
+		}
+		else
+		{
+			pCondition = pContext->m_pCondition;
+			pDataList = pContext->m_pData;
+		}
+
+		pCondition->Lock();
+		if( pDataList->size() )
+		{
+			*p_pData = pDataList->front();
+			pDataList->pop_front();
+			pCondition->Unlock();
+			return true;
+		}
+
+		for(;0==m_bWantToStop;)
+		{
+			pCondition->Wait();
+			if( 0 != m_bWantToStop )
+			{
+				pCondition->Unlock();
+				return false;
+			}
+
+			if( pDataList->size() )
+			{
+				*p_pData = pDataList->front();
+				pDataList->pop_front();
+				pCondition->Unlock();
+				return true;
+			}
+		}
+
+		pCondition->Unlock();
+
+		return false;
+	}
+
+	virtual bool SendData( _DataType p_Data )
+	{
+		return __SendData( p_Data, &m_ShareCondition, &m_ShareData );
+	}
+
+	void ClearSharedData(){
+		m_ShareCondition.Lock();
+		m_ShareData.clear();
+		m_ShareCondition.Unlock();
+	}
+
+	virtual bool SendDataToThread( _DataType p_Data, size_t n )
+	{
+		return SendData( p_Data );
+	}
+#if 0
+    virtual void Stop()
+    {
+        std::recursive_mutex::scoped_lock _lock(m_Mutex);
+
+		m_bWantToStop = 1;
+
+		if( m_Threads.size() )
+		{
+			__ThreadQueueIter iter;
+			__Thread_t *pThread;
+
+			for(iter=m_Threads.begin();iter!=m_Threads.end();iter++)
+			{
+				pThread = *iter;
+				//pThread->m_pCondition->Lock();
+				//pThread->m_pCondition->NotifyAll();
+				//pThread->m_pCondition->Unlock();
+
+                bool finished = false;
+                while(!finished)
+                {
+                    std::this_thread::sleep(std::posix_time::milliseconds(10));
+                    pThread->m_pCondition->Lock();
+			        pThread->m_pCondition->NotifyAll();
+			        pThread->m_pCondition->Unlock();
+                    finished = pThread->m_pThread->timed_join( std::posix_time::milliseconds(1*10));
+                }
+    			//pThread->m_pThread->join();
+				delete pThread;
+			}
+
+			m_Threads.clear();
+		}
+
+		m_ShareCondition.NotifyAll();
+    }
+#endif
+#if 1
+	virtual void Stop(){
+		m_Mutex.lock();
+		//std::recursive_mutex::scoped_lock _lock(m_Mutex);
+
+		m_bWantToStop = 1;
+
+		if( m_Threads.size() )
+		{
+			__ThreadQueueIter iter;
+			__Thread_t *pThread;
+
+			for(iter=m_Threads.begin();iter!=m_Threads.end();iter++)
+			{
+				pThread = *iter;
+				pThread->m_pCondition->Lock();
+				pThread->m_pCondition->NotifyAll();
+				pThread->m_pCondition->Unlock();
+				m_Mutex.unlock();//join需要等待线程结束，而线程的结束可能需要 m_Mutex
+				pThread->m_pThread->join();
+				m_Mutex.lock();
+			}
+			//两个线程都停了，再删除。否则waitdata的时候会遍历数组，访问里面的对象
+			for(iter=m_Threads.begin();iter!=m_Threads.end();iter++){
+				pThread = *iter;
+				delete pThread;
+			}
+
+			m_Threads.clear();
+		}
+
+		m_ShareCondition.NotifyAll();
+		m_Mutex.unlock();
+	}
+#endif
+};
+
+template<class _Tp>
+class JCThreadPoolSingle : public JCThreadPoolShare<_Tp>
+{
+    typedef JCThreadPoolShare<_Tp> __BaseClass;
+    
+protected:
+	/*
+		single 的话
+			每个线程使用独立的 JCCondition 和 数据队列。
+
+	*/
+	virtual int __AddThread( std::thread *p_pThread )
+	{
+		JCCondition *pCondition = new JCCondition;
+        typename __BaseClass::__DataQueue *pData = new typename __BaseClass::__DataQueue;
+
+        typename __BaseClass::__Thread_t *p = new typename __BaseClass::__Thread_t( p_pThread, pCondition, pData, true );
+
+		__BaseClass::m_Threads.push_back( p );
+
+		return __BaseClass::m_Threads.size()-1;
+	}
+
+public:
+	virtual bool SendData( typename __BaseClass::_DataType p_Data )
+	{
+		throw "JCThreadPoolSingle::SendData error";
+		return false;
+	}
+
+	virtual bool SendDataToThread( typename __BaseClass::_DataType p_Data, size_t n )
+	{
+		std::unique_lock<std::recursive_mutex> _lock(__BaseClass::m_Mutex);
+
+		if( n >= __BaseClass::m_Threads.size() )
+		{
+			return false;
+		}
+
+        typename __BaseClass::__Thread_t *p = __BaseClass::m_Threads[n];
+
+		return this->__SendData( p_Data, p->m_pCondition, p->m_pData );
+	}
+
+	virtual void ClearDataOfThread(int p_n){
+		printf("----ClearDataOfThread %d\n",p_n);
+		std::unique_lock<std::recursive_mutex> _lock(__BaseClass::m_Mutex);
+		if( p_n >= (int)__BaseClass::m_Threads.size() ){
+			return ;
+		}
+        typename __BaseClass::__Thread_t *p = __BaseClass::m_Threads[p_n];
+		p->m_pCondition->Lock();
+		printf("    datasize=%d\n", p->m_pData->size());
+		p->m_pData->clear();
+		p->m_pCondition->Unlock();
+	}
+};
+//------------------------------------------------------------------------------
+
+
+#endif //__JCThreadPool_H__
+
+//-----------------------------END FILE--------------------------------