diy数据库（七）--线程控制块、消息、线程池

        每一个数据库实例都有一个线程池，这个线程池是可以动态增长的。这个线程池管理者这个数据库实例中的所有线程。

EDUSTADUS _stadus;//线程状态

每一个线程都处于一个有限状态机中，线程一共有五种状态，当线程收到消息时，会在五种状态之间转换。

EDUID  id;//线程的edu id

        bool       _isForced ;//是否关闭线程,一般由线程池来设置为true
        bool       _isDisconnected ;//是否断开连接，在代理线程接收到客户端的disconnect命令后就会将本属性设置为true
        ossQueue<pmdEDUEvent> _queue ;//消息队列，每个edu都有一个消息队列

2）线程的入口函数
typedef int (*pmdEntryPoint) ( pmdEDUCB *, void * ) ;//线程入口函数的类型，函数指针。

pmdEntryPoint getEntryFuncByType ( EDU_TYPES type ) ;//通过线程类型得到对应的入口函数

int pmdAgentEntryPoint ( pmdEDUCB *cb, void *arg ) ;//代理类型的函数指针
int pmdTcpListenerEntryPoint ( pmdEDUCB *cb, void *arg ) ;//监听类型的函数指针
int pmdEDUEntryPoint ( EDU_TYPES type, pmdEDUCB *cb, void *arg ) ;//线程的入口函数，实际上在函数体里面会根据type参数来调用对应的线程入口函数

三、线程间通信用的消息类

#ifndef PMDEDUEVENT_HPP__
#define PMDEDUEVENT_HPP__

#include "core.hpp"
enum pmdEDUEventTypes//消息类型
{
   PMD_EDU_EVENT_NONE = 0,//空的消息
   PMD_EDU_EVENT_TERM,     // 结束消息
   PMD_EDU_EVENT_MSG,
   PMD_EDU_EVENT_RESUME   // 唤醒消息
} ;

class pmdEDUEvent
{
public :
   pmdEDUEvent () :
   _eventType(PMD_EDU_EVENT_NONE),
   _release(false),
   _Data(NULL)
   {
   }

   pmdEDUEvent ( pmdEDUEventTypes type ) :
   _eventType(type),
   _release(false),
   _Data(NULL)
   {
   }

   pmdEDUEvent ( pmdEDUEventTypes type, bool release, void *data ) :
   _eventType(type),
   _release(release),
   _Data(data)
   {
   }

   void reset ()//清空
   {
      _eventType = PMD_EDU_EVENT_NONE ;
      _release = false ;
      _Data = NULL ;
   }

   pmdEDUEventTypes _eventType ;//事件类型
   bool             _release ;//释放
   void            *_Data ;//数据
} ;

#endif

四、线程池

#ifndef PMDEDUMGR_HPP__
#define PMDEDUMGR_HPP__

#include "core.hpp"
#include "pmdEDU.hpp"
#include "ossLatch.hpp"
#include "ossUtil.hpp"

#define EDU_SYSTEM     0x01//系统级edu
#define EDU_USER       0x02//用户级edu
#define EDU_ALL        ( EDU_SYSTEM | EDU_USER )


class pmdEDUMgr//线程池
{
private :
   std::map<EDUID, pmdEDUCB*> _runQueue ;//运行的线程队列
   std::map<EDUID, pmdEDUCB*> _idleQueue ;//空闲的线程队列
   std::map<unsigned int, EDUID> _tid_eduid_map ;//线程池中所有线程id和其edu的映射

   ossSLatch _mutex ;//对线程队列的的访问时必须用到的读写锁
   EDUID _EDUID ;//new下一个edu时分配的id
   std::map<unsigned int, EDUID> _mapSystemEDUS ;//系统edu的类型和eduid之间的映射
   bool _isQuiesced ;//标识数据库是否被挂起，当数据库正在挂起时线程池不接受请求，挂起数据库是为了使DBA可以对数据库进行一些特殊的操作
   bool _isDestroyed ;//线程池是否被销毁
public :
   pmdEDUMgr () :
   _EDUID(1),
   _isQuiesced(false),
   _isDestroyed(false)
   {
   }

   ~pmdEDUMgr ()
   {
      reset () ;
   }
   void reset ()
   {
      _destroyAll () ;//删除线程池中的所有线程
   }

   unsigned int size ()//得到所有线程的数量
   {
      unsigned int num = 0 ;
      _mutex.get_shared () ;//共享锁，因为是读
      num = ( unsigned int ) _runQueue.size() +
            ( unsigned int ) _idleQueue.size () ;
      _mutex.release_shared () ;
      return num ;
   }

   unsigned int sizeRun ()//得到运行队列的长度
   {
      unsigned int num = 0 ;
      _mutex.get_shared () ;
      num = ( unsigned int ) _runQueue.size () ;
      _mutex.release_shared () ;
      return num ;
   }

   unsigned int sizeIdle ()
   {
      unsigned int num = 0 ;
      _mutex.get_shared () ;
      num = ( unsigned int ) _idleQueue.size () ;
      _mutex.release_shared () ;
      return num ;
   }

   unsigned int sizeSystem ()//系统edu的数量
   {
      unsigned int num = 0 ;
      _mutex.get_shared () ;
      num = _mapSystemEDUS.size() ;
      _mutex.release_shared () ;
      return num ;
   }

   EDUID getSystemEDU ( EDU_TYPES edu )//由系统edu类型得到其对应的eduid
   {
      EDUID eduID = PMD_INVALID_EDUID;
      _mutex.get_shared () ;
      std::map<unsigned int, EDUID>::iterator it = _mapSystemEDUS.find( edu ) ;
      if ( it != _mapSystemEDUS.end() )
      {
         eduID = it->second  ;
      }
      _mutex.release_shared () ;
      return eduID ;
   }

   bool isSystemEDU ( EDUID eduID )
   {
      bool isSys = false ;
      _mutex.get_shared () ;
      isSys = _isSystemEDU ( eduID ) ;
      _mutex.release_shared () ;
      return isSys ;
   }

   void regSystemEDU ( EDU_TYPES edu, EDUID eduid )//注册一个系统edu
   {
      _mutex.get() ;
      _mapSystemEDUS[ edu ] = eduid ;
      _mutex.release () ;
   }

   bool isQuiesced ()
   {
      return _isQuiesced ;
   }
   void setQuiesced ( bool b )
   {
      _isQuiesced = b ;
   }
   bool isDestroyed ()
   {
      return _isDestroyed ;
   }
   static bool isPoolable ( EDU_TYPES type )//线程edu的类型对应的线程是否能回收到线程池，目前只有agent线程的edu可以被回收，因为正常情况下监听线程是不会退出的
   {
      return ( EDU_TYPE_AGENT == type ) ;
   }

private :
   int _createNewEDU ( EDU_TYPES type, void *arg, EDUID *eduid ) ;//没有空闲edu时，创建新的edu
   int _destroyAll () ;//销毁所有的线程
   int _forceEDUs ( int property = EDU_ALL ) ;//干掉某种类型的edu
   unsigned int _getEDUCount ( int property = EDU_ALL ) ;
   void _setDestroyed ( bool b )
   {
      _isDestroyed = b ;
   }
   bool _isSystemEDU ( EDUID eduID )
   {
      std::map<unsigned int, EDUID>::iterator it = _mapSystemEDUS.begin() ;
      while ( it != _mapSystemEDUS.end() )
      {
         if ( eduID == it->second )
         {
            return true ;
         }
         ++it ;
      }
      return false ;
   }

   int _destroyEDU ( EDUID eduID ) ;//销毁一个特定的edu

   int _deactivateEDU ( EDUID eduID ) ;//把一个edu放回线程池
public :
   /*
    * EDU Status Transition Table
    * C: CREATING
    * R: RUNNING
    * W: WAITING
    * I: IDLE
    * D: DESTROY
    * 函数
    * c: createNewEDU
    * a: activateEDU
    * d: destroyEDU
    * w: waitEDU
    * t: deactivateEDU
    *   C   R   W   I   D  <--- from
    * C c
    * R -   -   a   a   -  <--- Creating/Idle/Waiting status can move to Running status
    * W w   w   -   -   -  <--- Running status move to Waiting
    * I -   -   t   -   -  <--- Creating/Waiting status move to Idle
    * D d   -   d   d   -  <--- Creating / Waiting / Idle can be destroyed
    * ^ To
    */

   int activateEDU ( EDUID eduID ) ;//将waiting和idle线程激活成running线程
   int waitEDU ( EDUID eduID ) ;//使edu从执行状态转换成等待状态
   int startEDU ( EDU_TYPES type, void* arg, EDUID *eduid ) ;//可以创建一个edu，并创建相应的线程。如果创建的是代理，而且线程池中有空闲线程则从线程池里面去拿，并把相应的EDU从空闲队列放到运行队列
   int postEDUPost ( EDUID eduID, pmdEDUEventTypes type,
                     bool release = false, void *pData = NULL ) ;//向一个指定线程的队列发送一个消息，将激活因为等待消息队列中的消息而阻塞的线程 
   int waitEDUPost ( EDUID eduID, pmdEDUEvent& event,
                     long long millsecond ) ;//等待指定edu的事件  
   int returnEDU ( EDUID eduID, bool force, bool* destroyed ) ;//把edu销毁或者回池
   int forceUserEDU ( EDUID eduID ) ;
   pmdEDUCB *getEDU ( unsigned int tid ) ;//通过线程id得到edu
   pmdEDUCB *getEDU () ;//得到当前的edu
   pmdEDUCB *getEDUByID ( EDUID eduID ) ;
   void setEDU ( unsigned int tid, EDUID eduid ) ;

} ;

#endif

分析 

1、EDU 的状态迁移图
    * C: CREATING
    * R: RUNNING
    * W: WAITING
    * I: IDLE
    * D: DESTROY
    * 函数
    * c: createNewEDU
    * a: activateEDU
    * d: destroyEDU
    * w: waitEDU
    * t: deactivateEDU
    *   C   R   W   I   D  <--- from
    * C c
    * R -   -   a   a   -  <---Idle/Waiting status can move to Running status
    * W w   w   -   -   -  <--- Creating/Running status move to Waiting
    * I t   -   t   -   -  <--- Waiting status move to Idle
    * D d   -   d   d   -  <--- Creating / Waiting / Idle can be destroyed

2、CREATING、 RUNNING、 WAITING这三种状态的线程的EDU存放在运行线程队列中，IDLE状态的线程放在空闲线程队列中，DESTROY线程的EDU不在任何线程队列中（因为当一个线程的状态为销毁状态时，他的线程管理块已经被线程池从两个线程队列中删除了），当线程实体检测到自己的状态为DESTROY时，该线程实体就会从线程执行函数返回，从而真正意义上结束线程

五、总结：

       1、本文主要分析了Diydb的线程模块，实际上就是一个比较完善的线程池模型

       2、线程池主要由以下几个类型实现：

             线程控制块：主要是对线程实体的封装。每个线程控制块实体与一个线程实体对应，其包含了这个线程的状态、类型、对应的线程池以及该线程的消息队列等。

            消             息：消息实体用来在线程间传递信息，通过这些消息可以传递普通的数据，也可以控制接收到消息的线程的状态

            线程管理 类：整个Diydb中只有一个线程管理类的实例，她是线程池的核心，主要管理着两个队列，一个是运行线程队列，一个是空闲线程队列（空闲线程队列中只可能是代理线程）。她控制着所有线程的状态，从而来维护线程池。

      3、本线程池中的线程分为系统线程（这里只有监听线程）和用户线程（这里只有代理线程），每个系统线程类型对应的只有一个线程实体，而每个用户线程类型的线程对应的是多个线程实体。

      4、只有代理线程的状态可能是空闲线程，所以从某种意义上来讲这个线程池实际上只对代理线程的效率有益，尽管系统线程的EDU也在运行线程队列中，而且其状态也受线程管理类的控制。

      5、线程池中的空闲线程不是系统启动时一次性创建完毕的，是当创建的代理线程处理完用户请求后（即用户断开连接），返回给空闲线程队列的（当然前提条件是当前空闲线程的个数小于maxpool，即空闲线程上限）。

      6、线程池的大小是由一个参数（maxpool）来控制的，当线程池中空闲线程的个数等于maxpool时，运行完毕的线程就不允许回池。

      7、当线程池中的空闲线程都用完以后，如果有新的用户请求过来，就会创建一个新的代理线程去处理这个用户的任务。所以，这个线程池中的线程数是可以动态变化的。

diy数据库（七）--线程控制块、消息、线程池

标签：