Chinaunix首页 | 论坛 | 博客

ha0xFF

首页　| 　博文目录　| 　关于我

txawplh

  • 博客访问： 85631
  • 博文数量： 12
  • 博客积分： 0
  • 博客等级： 民兵
  • 技术积分： 200
  • 用 户 组： 普通用户
  • 注册时间： 2013-01-23 09:57
文章分类

全部博文（12）

文章存档

2015年（11）

2014年（1）

我的朋友
最近访客
推荐博文
相关博文
SSDB源码分析（5）：任务 和 任务执行

分类： 服务器与存储

2015-08-09 02:23:19

    之前讲到SSDB会注册任务处理函数，还会将客户端的请求封装为任务进行分发处理:
    1. 注册处理函数

点击(此处)折叠或打开


  2. // net/proc.h 
  3. #define DEF_PROC(f) int proc_##f(NetworkServer *net, Link *link, const Request &req, Response *resp)

  5. // serv.cpp
  6. #define REG_PROC(c, f) net->proc_map.set_proc(#c, f, proc_##c)

  8. void SSDBServer::reg_procs(NetworkServer *net){
  9.     REG_PROC(get, "rt");
  10.     REG_PROC(set, "wt");
  11.     REG_PROC(del, "wt");
  12.     REG_PROC(setx, "wt");
  13.     REG_PROC(setnx, "wt");
  14.     REG_PROC(getset, "wt");
  15.     REG_PROC(getbit, "rt");
  16.     REG_PROC(setbit, "wt");
  17.     REG_PROC(countbit, "rt");
  18.     REG_PROC(substr, "rt");
  19.     REG_PROC(getrange, "rt");
  20.     REG_PROC(strlen, "rt");
  21.     .......
  22. }
    2. 任务分发处理:
    

点击(此处)折叠或打开


  2. // server.cpp 
  3. if(cmd->flags & Command::FLAG_THREAD){
  4.             if(cmd->flags & Command::FLAG_WRITE){
  5.                 job->result = PROC_THREAD;
  6.                 // 写任务线程池处理
  7.                 writer->push(*job);
  8.             }else{
  9.                 job->result = PROC_THREAD;
  10.                 // 读任务线程池处理
  11.                 reader->push(*job);
  12.             }
  13.             return;
  14.         }

  16.         proc_t p = cmd->proc;
  17.         job->time_wait = 1000 * (millitime() - job->stime);
  18.         // 当前主线程处理
  19.         job->result = (*p)(this, job->link, *req, &resp);
  20.         job->time_proc = 1000 * (millitime() - job->stime) - job->time_wait;

下面将详细分析SSDB中的任务和任务执行

一. ProcJob && Command && ProcMap

    与任务相关的数据结构的定义都放在proc.h中，交给任务线程池处理的对象是Command类型的，而不是ProcJob类型，Command对象是根据ProcJob构造来的。它们两者的区别在于：
    ProcJob关心的是在哪个Link上需要进行任务处理，至于处理的具体任务是什么，处理函数是什么，ProcJob并不关心。
    Command 则是众多处理任务中的一个，它包含了处理函数是什么，具体任务类型是什么。在一个Link(连接)上可能会发生很多次任务处理，那么只会有一个ProcJob，但会包含很多Command。

    ProcMap则是处理函数的注册表，NetworkServer将根据名字查找注册函数，并获得任务类型

点击(此处)折叠或打开

  1. #define PROC_OK            0
  2. #define PROC_ERROR        -1
  3. #define PROC_THREAD 1
  4. #define PROC_BACKEND    100

  6. // 通过宏定义处理函数
  7. #define DEF_PROC(f) int proc_##f(NetworkServer *net, Link *link, const Request &req, Response *resp)

  9. typedef std::vector<Bytes> Request;
  10. // 处理函数的类型
  11. typedef int (*proc_t)(NetworkServer *net, Link *link, const Request &req, Response *resp);

  13. // Command类
  14. // FLAG_READ    : 读任务
  15. // FLAG_WRITE   : 写任务
  16. // FLAG_BACKEND : 需要新建线程背地执行的特殊任务（例如主从同步，日志处理）
  17. // FLAG_THREAD  : 需要放到任务线程池中执行的任务
  18. struct Command{
  19.     static const int FLAG_READ        = (1 << 0);
  20.     static const int FLAG_WRITE        = (1 << 1);
  21.     static const int FLAG_BACKEND    = (1 << 2);
  22.     static const int FLAG_THREAD    = (1 << 3);

  24.     std::string name;
  25.     int flags;
  26.     proc_t proc;
  27.     // 统计相关的变量
  28.     uint64_t calls;
  29.     double time_wait;
  30.     double time_proc;
  31.     
  32.     Command(){
  33.         flags = 0;
  34.         proc = NULL;
  35.         calls = 0;
  36.         time_wait = 0;
  37.         time_proc = 0;
  38.     }
  39. };


  42. // ProcJob类
  43. struct ProcJob{
  44.     int result;
  45.     NetworkServer *serv;
  46.     // 任务涉及的连接
  47.     Link *link;
  48.     Command *cmd;
  49.     double stime;
  50.     double time_wait;
  51.     double time_proc;
  52.     
  53.     ProcJob(){
  54.         result = 0;
  55.         serv = NULL;
  56.         link = NULL;
  57.         cmd = NULL;
  58.         stime = 0;
  59.         time_wait = 0;
  60.         time_proc = 0;
  61.     }
  62. };


  65. struct BytesEqual{
  66.     bool operator()(const Bytes &s1, const Bytes &s2) const {
  67.         return (bool)(s1.compare(s2) == 0);
  68.     }
  69. };
  70. struct BytesHash{
  71.     size_t operator()(const Bytes &s1) const {
  72.         unsigned long __h = 0;
  73.         const char *p = s1.data();
  74.         for (int i=0 ; i<s1.size(); i++)
  75.             __h = 5*__h + p[i];
  76.         return size_t(__h);
  77.     }
  78. };


  81. #define GCC_VERSION (__GNUC__ * 100 + __GNUC_MINOR__)
  82. #if GCC_VERSION >= 403
  83.     #include <tr1/unordered_map>
  84.     typedef std::tr1::unordered_map<Bytes, Command *, BytesHash, BytesEqual> proc_map_t;
  85. #else
  86.     #ifdef NEW_MAC
  87.         #include <unordered_map>
  88.         typedef std::unordered_map<Bytes, Command *, BytesHash, BytesEqual> proc_map_t;
  89.     #else
  90.         #include <ext/hash_map>
  91.         typedef __gnu_cxx::hash_map<Bytes, Command *, BytesHash, BytesEqual> proc_map_t;
  92.     #endif
  93. #endif

  95. // 将所有的处理函数都注册到ProcMap中
  96. class ProcMap
  97. {
  98. private:
  99.     proc_map_t proc_map;

  101. public:
  102.     ProcMap();
  103.     ~ProcMap();
  104.     void set_proc(const std::string &cmd, const char *sflags, proc_t proc);
  105.     void set_proc(const std::string &cmd, proc_t proc);
  106.     Command* get_proc(const Bytes &str);
  107.     
  108.     proc_map_t::iterator begin(){
  109.         return proc_map.begin();
  110.     }
  111.     proc_map_t::iterator end(){
  112.         return proc_map.end();
  113.     }
  114. };


二 . 任务执行： WorkerPool && ProcWoker

    在NetworkServer解析出客户端的请求后，会根据需要将任务添加到任务线程池中处理，处理完成之后，再从线程池的结果队列中获取已完成的任务
    

点击(此处)折叠或打开

  1. if(cmd->flags & Command::FLAG_THREAD){
  2.             if(cmd->flags & Command::FLAG_WRITE){
  3.                 job->result = PROC_THREAD;
  4.                 writer->push(*job);
  5.             }else{
  6.                 job->result = PROC_THREAD;
  7.                 reader->push(*job);
  8.             }
  9.             return;
  10.         }

    任务线程池(reader, writer) 类型为WorkerPool，它的定义在thread.h中：

点击(此处)折叠或打开

  1. template<class W, class JOB>
  2. class WorkerPool{
  3.     public:
  4.         class Worker{
  5.             public:
  6.                 Worker(){};
  7.                 Worker(const std::string &name);
  8.                 virtual ~Worker(){}
  9.                 int id;
  10.                 virtual void init(){}
  11.                 virtual void destroy(){}
  12.                 virtual int proc(JOB *job) = 0;
  13.             private:
  14.             protected:
  15.                 std::string name;
  16.         };
  17.     private:
  18.         std::string name;

  20.         // 工作者池需要处理的JOB
  21.         Queue<JOB> jobs;
  22.         // 处理完成后的结果
  23.         SelectableQueue<JOB> results;

  25.         int num_workers;
  26.         std::vector<pthread_t> tids;
  27.         bool started;

  29.         struct run_arg{
  30.             int id;
  31.             WorkerPool *tp;
  32.         };
  33.         static void* _run_worker(void *arg);
  34.     public:
  35.         WorkerPool(const char *name="");
  36.         ~WorkerPool();

  38.         int fd(){
  39.             return results.fd();
  40.         }
  41.         
  42.         int start(int num_workers);
  43.         int stop();
  44.         
  45.         int push(JOB job);
  46.         int pop(JOB *job);
  47. };
   
    线程池启动：
    

点击(此处)折叠或打开

  1. template<class W, class JOB>
  2. void* WorkerPool<W, JOB>::_run_worker(void *arg){
  3.     struct run_arg *p = (struct run_arg*)arg;
  4.     int id = p->id;
  5.     WorkerPool *tp = p->tp;
  6.     delete p;

  8.     W w(tp->name);
  9.     Worker *worker = (Worker *)&w;
  10.     worker->id = id;
  11.     worker->init();
  12.     while(1){
  13.         JOB job;
  14.         // 获取任务
  15.         if(tp->jobs.pop(&job) == -1){
  16.             fprintf(stderr, "jobs.pop error\n");
  17.             ::exit(0);
  18.             break;
  19.         }
  20.         // 处理任务
  21.         worker->proc(&job);
  22.         // 写入结果
  23.         if(tp->results.push(job) == -1){
  24.             fprintf(stderr, "results.push error\n");
  25.             ::exit(0);
  26.             break;
  27.         }
  28.     }
  29.     worker->destroy();
  30.     return (void *)NULL;
  31. }

  33. template<class W, class JOB>
  34. int WorkerPool<W, JOB>::start(int num_workers){
  35.     this->num_workers = num_workers;
  36.     if(started){
  37.         return 0;
  38.     }
  39.     int err;
  40.     pthread_t tid;
  41.     for(int i=0; i<num_workers; i++){
  42.         struct run_arg *arg = new run_arg();
  43.         arg->id = i;
  44.         arg->tp = this;
  45.         // 创建任务线程
  46.         err = pthread_create(&tid, NULL, &WorkerPool::_run_worker, arg);
  47.         if(err != 0){
  48.             fprintf(stderr, "can't create thread: %s\n", strerror(err));
  49.         }else{
  50.             tids.push_back(tid);
  51.         }
  52.     }
  53.     started = true;
  54.     return 0;
  55. }


   
    Worker类抽象了具体的执行任务过程：
        step1 . 根据ProcJob获取对应的Command，执行其中的任务处理函数
        step2 . 将任务结果放到Link的输出缓冲区中

点击(此处)折叠或打开

  1. int ProcWorker::proc(ProcJob *job){
  2.     const Request *req = job->link->last_recv();
  3.     Response resp;
  4.     
  5.     proc_t p = job->cmd->proc;
  6.     // 任务等待的时长
  7.     job->time_wait = 1000 * (millitime() - job->stime);
  8.     // 根据传入的处理函数处理请求，得到结果
  9.     job->result = (*p)(job->serv, job->link, *req, &resp);
  10.     // 任务处理的时长
  11.     job->time_proc = 1000 * (millitime() - job->stime) - job->time_wait;
  12.     // 任务结果放到Link输出缓冲区中
  13.     if(job->link->send(resp.resp) == -1){
  14.         job->result = PROC_ERROR;
  15.     }else{
  16.         log_debug("w:%.3f,p:%.3f, req: %s, resp: %s",
  17.             job->time_wait, job->time_proc,
  18.             serialize_req(*req).c_str(),
  19.             serialize_req(resp.resp).c_str());
  20.     }
  21.     return 0;
  22. }


    线程池停止：
    

点击(此处)折叠或打开

  1. template<class W, class JOB>
  2. int WorkerPool<W, JOB>::stop(){
  3.     // TODO: notify works quit and wait
  4.     for(int i=0; i<tids.size(); i++){
  5. #ifdef OS_ANDROID
  6. #else
  7.         pthread_cancel(tids[i]);
  8. #endif
  9.     }
  10.     return 0;
  11. }


注意： WorkerPool中的任务队列( jobs ) 和结果队列 ( results )使用了不同的数据结构，这与它们的应用场景有关：
    Queue : Queue内部仅仅使用pthread_mutex和pthread_cond_t来同步读写操作，当Queue为空时，读者(任务线程)会阻塞在条件变量上等待新任务的到达。
    SelectableQueue ：SelectableQueue的写入操作是工作者线程在完成任务之后写入结果，读取操作是网络处理主线程 (NetworkServer::serv )读取结果发送给客户端，因此在SelectableQueue为空时，不能让网络处理主线程阻塞在SelectableQueue上。因此在工作者线程写入时，会使用pipe通知主线程来读取，主线程读取的时候，仍然需要pthread_mutex同步。

    Queue的push和pop操作：

点击(此处)折叠或打开

  1. template <class T>
  2. int Queue<T>::push(const T item){
  3.     if(pthread_mutex_lock(&mutex) != 0){
  4.         return -1;
  5.     }
  6.     {
  7.         items.push(item);
  8.     }
  9.     pthread_mutex_unlock(&mutex);
  10.     pthread_cond_signal(&cond);
  11.     return 1;
  12. }

  14. template <class T>
  15. int Queue<T>::pop(T *data){
  16.     if(pthread_mutex_lock(&mutex) != 0){
  17.         return -1;
  18.     }
  19.     {
  20.         // 必须放在循环中, 因为 pthread_cond_wait 可能抢不到锁而被其它处理了
  21.         while(items.empty()){
  22.             //fprintf(stderr, "%d wait\n", pthread_self());
  23.             if(pthread_cond_wait(&cond, &mutex) != 0){
  24.                 //fprintf(stderr, "%s %d -1!\n", __FILE__, __LINE__);
  25.                 return -1;
  26.             }
  27.             //fprintf(stderr, "%d wait 2\n", pthread_self());
  28.         }
  29.         *data = items.front();
  30.         //fprintf(stderr, "%d job: %d\n", pthread_self(), (int)*data);
  31.         items.pop();
  32.     }
  33.     if(pthread_mutex_unlock(&mutex) != 0){
  34.         //fprintf(stderr, "error!\n");
  35.         return -1;
  36.     }
  37.         //fprintf(stderr, "%d wait end 2, job: %d\n", pthread_self(), (int)*data);
  38.     return 1;
  39. }


SelectableQueue的push和pop操作：

点击(此处)折叠或打开

  1. template <class T>
  2. int SelectableQueue<T>::push(const T item){
  3.     if(pthread_mutex_lock(&mutex) != 0){
  4.         return -1;
  5.     }
  6.     {
  7.         items.push(item);
  8.     }
  9.     // 写入管道通知主线程读取
  10.     if(::write(fds[1], "1", 1) == -1){
  11.         fprintf(stderr, "write fds error\n");
  12.         exit(0);
  13.     }
  14.     pthread_mutex_unlock(&mutex);
  15.     return 1;
  16. }

  18. template <class T>
  19. int SelectableQueue<T>::pop(T *data){
  20.     int n, ret = 1;
  21.     char buf[1];

  23.     while(1){
  24.         // 读取管道数据
  25.         n = ::read(fds[0], buf, 1);
  26.         if(n < 0){
  27.             if(errno == EINTR){
  28.                 continue;
  29.             }else{
  30.                 return -1;
  31.             }
  32.         }else if(n == 0){
  33.             ret = -1;
  34.         }else{
  35.             // phtead_mutex同步读取操作
  36.             if(pthread_mutex_lock(&mutex) != 0){
  37.                 return -1;
  38.             }
  39.             {
  40.                 if(items.empty()){
  41.                     fprintf(stderr, "%s %d error!\n", __FILE__, __LINE__);
  42.                     pthread_mutex_unlock(&mutex);
  43.                     return -1;
  44.                 }
  45.                 *data = items.front();
  46.                 items.pop();
  47.             }
  48.             pthread_mutex_unlock(&mutex);
  49.         }
  50.         break;
  51.     }
  52.     return ret;
  53. }

    


阅读(1880) | 评论(0) | 转发(0) |
0

上一篇：正则表达式 - 贪婪与非贪婪（惰性）

下一篇：SSDB源码分析（6）：在线备份功能 BackendDump

给主人留下些什么吧！~~
关于我们 | 关于IT168 | 联系方式 | 广告合作 | 法律声明 | 免费注册

Copyright 2001-2010 ChinaUnix.net All Rights Reserved 北京皓辰网域网络信息技术有限公司. 版权所有

感谢所有关心和支持过ChinaUnix的朋友们

16024965号-6