分布式队列和动态任务调度器的设计接口

1、6.线程池的实现共享资源计算的分解共享资源的设计思路1.线程分组竞争模式数据r尸内打嬲域I(kick)；山】|粧川1馄11耳:lnck3nckl)；:从产内存区i賊)则除栽駆:Unlcx1.(lociI)liKk.tlnrk?iIJnllKkWcVI；2.线程随机竞争模式3.条件同步模式4.批量私有化处理模式5.数据本地化相关解决思路的数据结构：串行队列，环行队列（缺点：空间不足时，重新扩大空间后需要复制数据，复制大量数据很费时间），2有既能将队列动态扩大，又不需要复制数据的队列结构吗？动态环行队列：多个环行队列组成的缘故动态环行队列。（用于管理内存碎片）上面这两种队列都是适用于单个线程内使用

2、的情形。如果有多个线程要操作队列，上述队列是不能提供数据竞争的安全的。要想让队列在多个线程操作的情形下取得安全，必须使用加锁、解锁等同步操作。3.支持同步的共享队列（1）要理解竞争资源（硬件、软件（变量，队列，数组，链表等）线程，资源竞争，信号量，互斥、同步，锁，原子操作共享队列的进队和出队操作仅仅是在串行队列的进队和出队基础上进行了锁保护，队列判空和满时也必须加锁，以避免线程同时操作串行队列内的数据结构。4偷取的共享队列（队列判空和满时不必须加锁）设置两个标志变量来表示队列为空和满，使用原子操作改变标志变量的值。在分布式队列和任务调度器的设计和实现中需要用到偷取的共享队列。5.队列池如何避免

3、共享队列资源的集中式锁竞争的现象？解决方案：队列池。将一个队列变成多个子队列，每个子队列带一把锁，多个共享子队列组成的复合数据结构称为队列池。队列池的操作有：单进单出操作、多进多出操作线程分组竞争模式，线程随机竞争模式分布式队列队列池的优点：缩减了队列空满判断时的同步操作缺点：无法减少进队和出队操作中的同步操作，解决方案7.分布式队列的设计和实现主要思想是，共享数据本地化，线程池。臥列池带有私有化操作的分布式队列7.多核下任务调度器静态任务图调度器动态任务调度Kr住务ill按非嵌套任务调度11meSumT(t2JW-ITm*1-TajT*2TinieSutn十T*jdwAvglimeTimeS

4、um+TWdwAvgTime任务分组任务分组调度rimcSunidwAvgTitne嵌套任务调度请大家自学C+模板设计方案1.设计并实现一个自动释放锁功能的类CScopedLock（已经有代码，大家会用就可以）接口如下：（1）设计并实现一个可以获取和释放锁功能的类CFastLock#defineFASTLOCK#defineFastLockInit(x)#defineFastLockAcquire(x)#defineFastLockTryAcquire(x)#defineFastLockRelease(x)#defineFastLockClose(x)classCFastLockprivate

5、:FASTLOCKm_Lock;public:CFastLock();CFastLock();CRITICAL_SECTIONInitializeCriticalSection(x)EnterCriticalSection(x)TryEnterCriticalSection(x)LeaveCriticalSection(x)DeleteCriticalSection(x)voidAcquire();intTryAcquire();voidRelease();voidRelease(UINTuType);/空的函数，主要是保持和CScopedLock类的兼容性;CFastLock:CFastLo

6、ck()FastLockInit(&m_Lock);CFastLock:CFastLock()FastLockClose(&m_Lock);voidCFastLock:Acquire()FastLockAcquire(&m_Lock);intCFastLock:TryAcquire()intnRet;nRet=FastLockTryAcquire(&m_Lock);returnnRet;voidCFastLock:Release()FastLockRelease(&m_Lock);2)在CFastLook的基础上，实现CScopedLocktemplateclassCScopedLockpri

7、vate:T*m_pLock;UINTm_uType;public:/构造函数为CFastLock,CSpinLock而使用CScopedLock(T&l):m_uType(0)m_pLock=&l;m_pLock-Acquire();/下面的构造函数主要为读写锁CRWLock而使用CScopedLock(T&l,UINTuType):m_uType(uType)m_pLock=&l;m_pLock-Acquire(uType);CScopedLock()if(m_pLock!=NULL)if(!m_uType)m_pLock-Release();elsem_pLock-Release(m_u

8、Type);串行环行队列接口(T表示队列中存放的数据类型)templateclassCLocalQueuePRIVATE:UINTm_uHead;/队列头部位置UINTm_uTail;/队列尾部位置UINTm_uMaxSize;/队列中的最大数据数量T*m_pData;/存放数据的数组public:CLocalQueue()m_pData=NULL;CLocalQueue(UINTuMaxSize);virtualCLocalQueue()if(m_pData!=NULL)deletem_pData;voidCreate(UINTuMaxSize);intEnQueue(T&Data);int

9、DeQueue(T&Data);boolIsEmpty();boolIsFull();支持同步的共享队列(T表示队列中的数据类型，TQueue是队列类型templateclassCSharedQueueprivate:TQueuem_queue;intm_nQueueSize;CFastLockm_lock;public:CSharedQueue();CSharedQueue(intnQueueSize);virtualCSharedQueue();voidCreate(intnQueueSize);intEnQueue(T&Data);intDeQueue(T&Data);4.偷取的共享队列

10、()#includeFastLock.h#includeCScopedLock.h#defineDEFAULT_STEALQUEUE_SIZE4templateclassCStealQueuePRIVATE:UINTUINTUINTT*LONGvolatileLONGvolatileCFastLockpublic:typedefpublic:m_uHead;/队列头部位置m_uTail;/队列尾部位置m_uMaxSize;/队列中的最大数据数量m_pData;/存放数据的数组m_lFullFlag;/队列为满的标志m_lEmptyFlag;/队列为空的标志m_Lock;/锁CStealQueu

11、eSubQueue;CStealQueue()m_pData=NULL;m_lFullFlag=0;m_lEmptyFlag=1;CStealQueue(UINTuMaxSize);CStealQueue(UINTuQueueCount,UINTuMaxSize);/为兼容性而设计的构造函数/主要为方便分布式队列使用virtualCStealQueue()if(m_pData!=NULL)deletem_pData;voidCreate(UINTuMaxSize);intEnQueue(T&Data);intDeQueue(T&Data);intPush(TData)returnEnQueue

12、(Data);intPop(T&Data)returnDeQueue(Data);intIsFull();intIsEmpty();voidLock();voidUnlock();intEnQueueNoLock(T&Data);intDeQueueNoLock(T&Data);UINTGetCountNoLock();/获取队列中的数据个数SubQueue*GetSubQueue()returnthis;5.队列池templateclassCQueuePoolPRIVATE:SharedQueue*m_pQueue;intm_nQueueCount;intm_nQueueSize;CFast

13、Randomm_Random;public:typedefSharedQueueSubQueue;public:CQueuePool();CQueuePool(intnQueueCount,intnQueueSize);virtualCQueuePool();voidCreate(intnQueueCount,intnQueueSize);intEnQueue(T&Data);intEnQueue(T&Data,intnIndex);intDeQueue(T&Data);intDeQueue(T&Data,intnIndex);intPush(T&Data);intPop(T&Data);in

14、tGetQueueCount()returnm_nQueueCount;SubQueue*GetSubQueue(intnIndex)return&(m_pQueuenIndex);6.线程池structTHREADPOOL_STARTFUNC_PARMETERSDWORDdwTlsIndex;/本地存储索引intnId;/线程编号THREADFUNCStartFunc;/线程入口函数void*pArgs;/线程入口函数参数;#defineMAX_SINGLE_THREAD_COUNT16classCThreadPoolprivate:DWORDTHREADFUNCvoid*m_dwTlsIn

15、dex;m_StartFunc;m_pArgs;/m_StartFunc的参数intHANDLETHREADPOOL_m_nThreadCount;/线程组内的线程数量m_pThreadArray;/针对线程组STARTFUNC_PARMETERS*m_pPara;/针/对线程组intm_nSingleThreadCount;/独立创建的单个线程的数量HANDLEm_SingleThreadMAX_SINGLE_THREAD_COUNT;/用来保存独立创建的单个线程的数组/保存独立创建的单个线程的入口函数参数THREADPOOL_STARTFUNC_PARMETERSm_SingleParaM

16、AX_SINGLE_THREAD_COUNT;public:CThreadPool();CThreadPool(THREADFUNCStartFunc,void*pArgs,intnThreadCount);virtualCThreadPool();voidCreateThreadPool(THREADFUNCStartFunc,void*pArgs,intnThreadCount=0);voidCreateThread(THREADFUNCStartFunc,void*pArgs);intGetThreadId();intGetThreadCount();voidExecAndWait();

17、voidExecThread(intnIndex);voidWaitAllThread();intThreadPool_GetThreadId(void*pArgs);/获取线程的编号7.分布式队列/获取线程的编号，此编号对应该线程对应的本地队列编号#include#includeCapiGlobal.h#includeFastLock.h#includeCScopedLock.h#includeCapiGlobal.h获取线程Id回调函数定义typedefint(*GetThreadIdFunc)(void*pArg);templateclassCDistributedQueueprivat

18、e:LocalQueue*SharedQueue*m_ppLocalQueue;m_pSharedQueue;/本地队列数组/共享队列池或共享队列intm_nLocalQueueSize;intm_nSharedQueueSize;intm_nLocalQueueCount;intm_nSharedQueueCount;DWORDm_dwTlsIndex;/线程本地存储索引LONGvolatilem_lThreadIdIndex;/线程编号最大值GetThreadIdFuncm_GetThreadIdFunc;/获取线程编号回调函数指针，如果由外面/的线程池提供编号时，需要传入回调函数void

19、*m_pThreadIdFuncArg;/获取线程编号回调函数的参数CFastLockm_LocalQueueResizeLock;/专为下面的ResizeLocalQueue函数使用voidResizeLocalQueue();/将m_ppLocalQueue数组的大小扩大一倍public:CDistributedQueue()m_GetThreadIdFunc=NULL;m_pThreadIdFuncArg=NULL;m_lThreadIdIndex=0;/*分布式队列的创建函数paramintnLocalQueueSize-本地子队列的大小paramintnLocalQueueCount

20、-本地队列的个数(数组的大小)为0表示和CPU核数相等的个数paramintnSharedQueueSize-共享子队列的大小paramintnSharedQueueCount-共享子队列的个数为0表示和CPU核数相等的个数returnvoid-无*/voidCreate(intnLocalQueueSize,intnLocalQueueCount,intnSharedQueueSize,intnSharedQueueCount);/*分布式队列的创建函数paramintnLocalQueueSize-本地子队列的大小paramintnLocalQueueCount-本地队列的个数(数组的大小

21、)paramintnSharedQueueSize-共享子队列的大小paramintnSharedQueueCount-共享子队列的个数paramGetThreadIdFuncGetThreadId-获取线程Id回调函数paramvoid*pThreadIdFuncArg-GetThreadId回调函数的参数returnvoid-无*/voidCreate(intnLocalQueueSize,intnLocalQueueCount,intnSharedQueueSize,intnSharedQueueCount,GetThreadIdFuncGetThreadId,void*pThreadI

22、dFuncArg);virtualCDistributedQueue();/*分布式队列的获取线程Id函数如果m_GetThreadIdFunc回调函数不为空，则使用它获取Id否则根据分布式队列内部的编号机制获取线程IdreturnLONG-返回线程的编号*/LONGThreadIdGet();/*分布式队列的进队操作函数这里假定了本地队列可以无限进队进队策略按以下优先级进行：1、本地队列空时进入本地队列，2、共享队列未满时进入共享队列3、共享队列满时进入本地队列paramT&Data-要进队的数据returnvoid-无*/voidEnQueue(T&Data);intDeQueue(T&D

23、ata);/*分布式队列的本地队列进队函数将数据进入到当前线程的本地队列中paramT&Data-要进队的数据returnvoid-无*/voidPushToLocalQueue(T&Data);/*分布式队列的指定序号本地队列进队函数这是一个为特殊需求而设计的函数使用这个函数要特别小心，必须保证不会发生数据竞争问题paramT&Data-要进队的数据paramintnIndex-本地队列的序号returnvoid-无*/voidPushToLocalQueue(T&Data,intnIndex);intPopFromLocalQueue(T&Data);intPushToSharedQueue(T&Data);intPopFromSharedQueue(T&Data);SharedQueue*GetSharedQueue()returnm_pSharedQueue;/*分布式队列的共享队列批量私有化函数paramintnSharedQueueIndex-共享子队列序号returnint-成功返回CAPI_SUCCESS,失败返回CAPI_FAILED.*/intPrivatizeSharedQueue(intnSharedQueueIndex);8.动态任务调度器的设计接口templateclassCTaskSche