Chinaunix首页 | 论坛 | 博客

路漫漫vincent.blog.chinaunix.net

首页　| 　博文目录　| 　关于我

vincent-cws

  • 博客访问： 957107
  • 博文数量： 116
  • 博客积分： 3923
  • 博客等级： 中校
  • 技术积分： 1337
  • 用 户 组： 普通用户
  • 注册时间： 2009-04-23 01:22
文章分类

全部博文（116）

文章存档

2013年（1）

2012年（17）

2011年（69）

2009年（29）

我的朋友
最近访客
推荐博文
相关博文
工作队列实现

分类： WINDOWS

2011-11-15 22:34:58

http://vincent.blog.chinaunix.net
最近在阅读关于电池充电的驱动代码，芯片是PM8606和PM8607，大概的原理就是主控芯片通过I2C总线和PM860x芯片通信，然后linux驱动挂接一个IRQ响应PM860x的中断信息，然后在IRQ的相应代码里通过I2C总线去读PM860x的中断寄存器状态(3个），然后根据指定的不同位置位来区别不同的事件，根据不同的事件去抛出一个工作给内核的工作队列去执行相关任务，这任务的主要内容是根据不同的事件去跑一个有穷状态机的流程，状态机主要工作就是自己设定PM860x的一些电压，电流，温度的门槛，当充电到一定程序越过这些门槛就进入下一个状态，然后重新设定相关门槛。然后还有一个定时器驱动的工作也定时的抛出任务给工作队列，主要是读取当前PM860x一些电压，电流，温度信息，然后根据这些信息做安全保护的作用。

其中我对里面的工作队列实现比较有兴趣，就自己用线程池模拟多cpu的情况实现了一个，不过区别还是蛮大的，还有就是实现时，我调试了一个很久的问题，就是我仿照linux的内核调用去INITWORK(work,func)时，发现work是静态的，然后插入到队列，后来发现这样是有问题，因为当我连续插入相同多个静态work时，work的链表指针已经被修改了，引起问题，还有就是多线程的同步时，由于多个锁，经常在退出时，忘记把相关的其它锁也释放，引起死锁。。。反正就是多联系，多调试。。。

下面是我实现的原理图：



代码如下：
  1. /**
  2.  * file: main.c
  3.  * author: vincent.cws2008@gmail.com
  4.  * history:
  5.  * @2011-11-06: initial
  6.  * @2011-11-13: add the notifier to register
  7.  */

  9. #include "threadpool.h"
  10. #include "stdio.h"
  11. #include "workqueuex.h"
  12. #include <stdlib.h>
  13. #include <time.h>

  15. static lock_t lock;

  17. void fwork_1(struct work_s *pwork)
  18. {
  19.     /* need to deleted by it self */
  20.     __lockx(lock);
  21.     printf("function: %s[thread_id:%08x] running ...\n", "fwork_1", gettidx());
  22.     fflush(stdout);    
  23.     __unlockx(lock);
  24.     FREE_WORK(pwork);
  25. }

  27. void fwork_2(struct work_s *pwork)
  28. {
  29.     /* need to deleted by it self */
  30.     __lockx(lock);
  31.     printf("function: %s[thread_id:%08x] running ...\n", "fwork_2", gettidx());
  32.     fflush(stdout);    
  33.     __unlockx(lock);
  34.     FREE_WORK(pwork);
  35. }
  36. void fwork_3(struct work_s *pwork)
  37. {
  38.     /* need to deleted by it self */
  39.     __lockx(lock);
  40.     printf("function: %s[thread_id:%08x] running ...\n", "fwork_3", gettidx());
  41.     fflush(stdout);    
  42.     __unlockx(lock);
  43.     FREE_WORK(pwork);
  44. }
  45. void fwork_4(struct work_s *pwork)
  46. {
  47.     /* need to deleted by it self */
  48.     __lockx(lock);
  49.     printf("function: %s[thread_id:%08x] running ...\n", "fwork_4", gettidx());
  50.     fflush(stdout);    
  51.     __unlockx(lock);
  52.     FREE_WORK(pwork);
  53. }

  55. static fwork_t fworks[] = {
  56.     fwork_1,
  57.     fwork_2,
  58.     fwork_3,
  59.     fwork_4,
  60. };

  62. void main()
  63. {
  64.     int count = 100;
  65.     work_t *pnew=null;
  66.     tasks_sched_t *pts = get_tasks_sched();
  67.     __assert(pts);
  68.     __lockx_init(lock,0);
  69.     
  70.     srand ((unsigned)time(NULL));
  71.     
  72.     init_tasks_sched();

  74.     while (count--)
  75.     {
  76.         int n = rand()%ARRAY_SIZE(fworks);
  77.         NEW_WORK(pnew,fworks[n]);
  78.         schedule_work(pnew);
  79.     }
  80.     //while (1);
  81.     sleep(5000);
  82.     exit_tasks_sched();
  83.     __lockx_exit(lock,0);
  84. }

  1. /**
  2.  * file: notifierx.c
  3.  * author: vincent.cws2008@gmail.com
  4.  * history:
  5.  * initial @ 2011-11-13
  6.  */

  8. #include "notifierx.h"


  11. int notifierx_chain_register(struct notifierx_head *pnh,
  12.         struct notifierx_block *pnb)
  13. {
  14.     __assert(pnh&&pnb);
  15.     __lockx(pnh->cslock);
  16.     list_add(&pnb->list, &pnh->head);
  17.     __unlockx(pnh->cslock);    
  18.     return 0;
  19. }

  21. int notifierx_chain_all_unregister(struct notifierx_head *pnh)
  22. {
  23.     struct list_head *list_p;
  24.     notifierx_block_t* pnb_it=null;
  25.     notifierx_block_t* pnb_temp=null;
  26.     __assert(pnh);

  28.     __lockx(pnh->cslock);
  29.     list_for_each_safe(list_p, pnb_temp, &(pnh->head))
  30.     {
  31.         pnb_it = list_entry(list_p, notifierx_block_t, list);
  32.         list_del(&pnb_it->list);
  33.     }
  34.     
  35.     __unlockx(pnh->cslock);

  37.     return 0;
  38. }

  40. int notifierx_chain_unregister(struct notifierx_head *pnh,
  41.         struct notifierx_block *pnb)
  42. {
  43.     struct list_head *list_p;
  44.     struct notifierx_block *pnb_it;
  45.     __assert(pnh&&pnb);
  46.     
  47.     __lockx(pnh->cslock);
  48.     list_for_each(list_p, &(pnh->head))
  49.     {
  50.         pnb_it = list_entry(list_p, notifierx_block_t, list);
  51.         if (pnb_it->notifierx_call == pnb->notifierx_call)
  52.         {
  53.             list_del(&pnb_it->list);
  54.             __unlockx(pnh->cslock);
  55.             return 0;
  56.         }
  57.     }
  58.     __unlockx(pnh->cslock);    
  59.     return -1;
  60. }

  62. int notifierx_call_chain(struct notifierx_head *pnh, unsigned long nlen, void *pdata)
  63. {
  64.     struct list_head *list_p;
  65.     notifierx_block_t* pnb_it=null;

  67.     __assert(pnh);
  68.     
  69.     __lockx(pnh->cslock);
  70.     list_for_each(list_p, &(pnh->head))
  71.     {
  72.         pnb_it = list_entry(list_p, notifierx_block_t, list);
  73.         pnb_it->notifierx_call(pnb_it, nlen, pdata);
  74.     }    
  75.     __unlockx(pnh->cslock);

  77.     return 0;
  78. }

  1. /**
  2.  * file: notifierx.h
  3.  * author: vincent.cws2008@gmail.com
  4.  * history:
  5.  * initial @ 2011-11-13
  6.  */

  8. #ifndef _NOTIFIERX_H_
  9. #define _NOTIFIERX_H_

  11. #include "list.h"
  12. #include "platform.h"

  14. typedef struct notifierx_block {
  15.     int (*notifierx_call)(struct notifierx_block *pnb, unsigned long nlen, const void *pdata);
  16.     struct list_head list;
  17. }notifierx_block_t;

  19. struct notifierx_head {
  20.     lock_t cslock;
  21.     attr_t attr;
  22.     struct list_head head;
  23. };

  25. #define INIT_NOTIFIERX_HEAD(name) do {                \
  26.         __lockx_init((name).cslock,(name).attr);    \
  27.         INIT_LIST_HEAD(&(name).head);                \
  28.     } while (0)
  29.         

  31. #define EXIT_NOTIFIERX_HEAD(name) do {                    \
  32.         notifierx_chain_all_unregister(&(name));    \
  33.         __lockx_exit((name).cslock, (name).attr);        \
  34.     } while (0)        

  36. extern int notifierx_chain_register(struct notifierx_head *pnh, struct notifierx_block *pnb);
  37. extern int notifierx_chain_all_unregister(struct notifierx_head *pnh);
  38. extern int notifierx_chain_unregister(struct notifierx_head *pnh, struct notifierx_block *pnb);
  39. extern int notifierx_call_chain(struct notifierx_head *pnh, unsigned long nlen, void *pdata);

  41. #endif

  1. /**
  2.  * file: platform.h
  3.  * author: vincent.cws2008@gmail.com
  4.  * history:
  5.  * @2011-11-15: initial
  6.  */

  8. #ifndef _PLATFORM_H_
  9. #define _PLATFORM_H_

  11. #ifdef WIN32
  12. # include "windows.h"
  13. #else
  14. # include <pthread.h>
  15. #endif

  17. #ifndef null
  18. #define null (void*)0
  19. #endif

  21. #ifndef NULL
  22. #define NULL (void*)0
  23. #endif

  25. #ifndef __assert
  26. #define __assert(x)
  27. #endif


  30. #ifdef WIN32
  31. #define sleep Sleep
  32. #define gettidx() GetCurrentThreadId()
  33. #else
  34. #define gettidx() pthread_self()
  35. #endif

  37. #ifdef WIN32
  38. #    define thrd_handle    HANDLE
  39. #    define event_t        HANDLE
  40. #    define lock_t        CRITICAL_SECTION
  41. #    define attr_t        void*
  42. #else
  43. #    define thrd_handle    pthread_t
  44. #    define event_t        pthread_cond_t
  45. #    define lock_t        pthread_mutex_t
  46. #    define attr_t        pthread_mutexattr_t
  47. #endif

  49. #ifdef WIN32

  51. #ifndef hnull
  52. #define hnull INVALID_HANDLE_VALUE
  53. #endif
  54. #ifndef null
  55. #define null NULL
  56. #endif
  57. #define __lockx(lock)            EnterCriticalSection(&(lock))
  58. #define __unlockx(lock)            LeaveCriticalSection(&(lock))
  59. #define __lockx_init(lock,attr) InitializeCriticalSection(&(lock))
  60. #define __lockx_exit(lock,attr) DeleteCriticalSection(&(lock));

  62. #else

  64. #ifndef hnull
  65. #define hnull (void*)0
  66. #endif
  67. #ifndef null
  68. #define null (void*)0
  69. #endif
  70. #define __lockx(lock)    pthread_mutex_lock(&(lock))
  71. #define __unlockx(lock) pthread_mutex_unlock(&(lock))
  72. #define __lockx_init(lock,attr) \
  73.     pthread_mutexattr_init(&(attr)); \
  74.     pthread_mutexattr_settype(&(attr),PTHREAD_MUTEX_RECURSIVE); \
  75.     pthread_mutex_init(&(lock),&(attr)); \

  77. #define __lockx_exit(attr,attr) \
  78.     pthread_mutex_destroy(&(lock)); \
  79.     pthread_mutexattr_destroy(&(attr)); \
  80.     
  81. #endif


  84. #define ARRAY_SIZE(x) (sizeof(x) / sizeof((x)[0]))

  86. #endif

  1. /**
  2.  * file: threadpool.c
  3.  * author: vincent.cws2008@gmail.com
  4.  * history:
  5.  * @2011-11-06: initial
  6.  * @2011-11-13: add the notifier to register
  7.  */

  9. #include "threadpool.h"

  11. #ifdef WIN32
  12. static DWORD WINAPI thread_proc(LPVOID pvoid)
  13. #else
  14. static void *thread_proc(void *pvoid)
  15. #endif
  16. {
  17.     thrdpool_t* pthrdpool = thrdpool();
  18.     thrd_desc_t* pthrd_desc=(thrd_desc_t*)pvoid;

  20.     thrd_msg_t thrd_msg;

  22.     __assert(pthrd_desc&&pthrdpool);

  24.        INIT_THRD_MSG(thrd_msg, pthrd_desc->h);
  25.         
  26.     while (1)
  27.     {
  28. #        ifdef WIN32
  29.         if (WaitForSingleObject((HANDLE)(pthrd_desc->hevent_go), INFINITE) == WAIT_OBJECT_0)
  30. #        else
  31.         int ret=0;
  32.         __lockx(pthrd_desc->cslock);
  33.         ret = pthread_cond_wait(&pthrd_desc->hevent_go, &pthrd_desc->cslock);
  34.         __unlockx(pthrd_desc->cslock);
  35.         if (ret == 0)
  36. #        endif
  37.         {
  38.             call_chain(pthrdpool->thrdpool_chain,thrd_msg,stat_busy);
  39.             __lockx(pthrd_desc->cslock);
  40.             while(pthrd_desc->ref_cnts>0) {
  41.                 if (pthrd_desc->thrd_hook)
  42.                     pthrd_desc->thrd_hook(pthrd_desc->pthrd_data);
  43.                 pthrd_desc->ref_cnts--;
  44.             }
  45. #            ifdef WIN32
  46.                 RESET_EVENT(pthrd_desc->hevent_go);
  47. #            endif
  48.             if (pthrd_desc&&pthrdpool->mod)
  49.                 CLEAN_THRD_HOOK(*pthrd_desc);
  50.             __unlockx(pthrd_desc->cslock);

  52.         }
  53. #        ifdef WIN32
  54.         if (WaitForSingleObject(pthrd_desc->hevent_exit, 0) == WAIT_OBJECT_0)
  55.             break;
  56. #        endif
  57.         
  58.         call_chain(pthrdpool->thrdpool_chain,thrd_msg,stat_free);
  59.     } /* while(1) */
  60.     return 0;    
  61. }

  63. static thrd_handle __thrd_alloc(thrd_hook_t thrd_hook, void* pdata)
  64. {
  65.     struct list_head *list_p;
  66.     thrd_desc_t* pthrd_desc=null;
  67.     
  68.     thrdpool_t* pthrdpool = thrdpool();    
  69.     __assert(pthrdpool);
  70.     __lockx(pthrdpool->cslock);
  71.     list_for_each(list_p, &(pthrdpool->list))
  72.     {
  73.         pthrd_desc = list_entry(list_p, thrd_desc_t, list);
  74.         if (!pthrd_desc->thrd_hook)
  75.         {
  76.             __lockx(pthrd_desc->cslock);
  77.             SET_HANDLER_THRD_DESC(*pthrd_desc, thrd_hook, pdata);    
  78.             __unlockx(pthrd_desc->cslock);
  79.             __unlockx(pthrdpool->cslock);
  80.             return pthrd_desc->h;
  81.         }
  82.     }
  83.     __unlockx(pthrdpool->cslock);
  84.     return 0;
  85. }

  87. static int __has_free_thrds()
  88. {
  89.     struct list_head *list_p;
  90.     thrd_desc_t* pthrd_desc=null;
  91.     
  92.     thrdpool_t* pthrdpool = thrdpool();    
  93.     __assert(pthrdpool);
  94.     __lockx(pthrdpool->cslock);
  95.     list_for_each(list_p, &(pthrdpool->list))
  96.     {
  97.         pthrd_desc = list_entry(list_p, thrd_desc_t, list);
  98.         if (0==pthrd_desc->ref_cnts) {
  99.             __unlockx(pthrdpool->cslock);
  100.             return 1;
  101.         }
  102.     }    
  103.     __unlockx(pthrdpool->cslock);
  104.     return 0;
  105. }

  107. static void __thrd_free(thrd_handle h)
  108. {
  109.     struct list_head *list_p;
  110.     thrd_desc_t* pthrd_desc=null;
  111.     
  112.     thrdpool_t* pthrdpool = thrdpool();    
  113.     __assert(pthrdpool);
  114.     __lockx(pthrdpool->cslock);
  115.     list_for_each(list_p, &(pthrdpool->list))
  116.     {
  117.         pthrd_desc = list_entry(list_p, thrd_desc_t, list);
  118.         if (h == pthrd_desc->h) {
  119.             __lockx(pthrd_desc->cslock);
  120.             CLEAR_THRD_DESC(*pthrd_desc);    
  121.             __unlockx(pthrd_desc->cslock);
  122.         }
  123.     }    
  124.     __unlockx(pthrdpool->cslock);
  125. }
  126.     
  127. static int __wakeup(thrd_handle h)
  128. {
  129.     struct list_head *list_p;
  130.     thrd_desc_t* pthrd_desc=null;
  131.     
  132.     thrdpool_t* pthrdpool = thrdpool();    
  133.     __assert(pthrdpool);
  134.     __lockx(pthrdpool->cslock);
  135.     list_for_each(list_p, &(pthrdpool->list))
  136.     {
  137.         pthrd_desc = list_entry(list_p, thrd_desc_t, list);
  138.         if (h==pthrd_desc->h)
  139.         {
  140.             __lockx(pthrd_desc->cslock);
  141.             if (hnull != pthrd_desc->hevent_go)
  142.             {
  143.                 SET_EVENT(pthrd_desc->hevent_go);
  144.                 pthrd_desc->ref_cnts++;
  145.             }
  146.             __unlockx(pthrd_desc->cslock);
  147.             __unlockx(pthrdpool->cslock);
  148.             return 0;
  149.         }
  150.     }    
  151.     __unlockx(pthrdpool->cslock);
  152.     return -1;
  153. }
  154.     
  155. static int __thrd_add()
  156. {
  157.     thrdpool_t* pthrdpool = thrdpool();
  158.     thrd_desc_t* pthrd_desc=null;
  159.     __assert(pthrdpool);

  161.     pthrd_desc=(thrd_desc_t*)malloc(sizeof(thrd_desc_t));
  162.     __assert(pthrd_desc);
  163.     
  164.     INIT_THRD_DESC(*pthrd_desc);

  166.     __lockx_init(pthrd_desc->cslock,pthrd_desc->attr);

  168. #    ifdef WIN32    
  169.     pthrd_desc->h = CreateThread(
  170.         null, // default security attributes
  171.         0, // use default stack size
  172.         thread_proc,// thread function
  173.         pthrd_desc, // argument to thread function
  174.         0, // use default creation flags
  175.         0);
  176.     if(NULL == pthrd_desc->h)
  177.         goto __err;
  178. #    else
  179.     if(pthread_create(&pthrd_desc->h, null, thread_proc, null)!=0)
  180.         goto __err;
  181. #    endif
  182.     
  183.     __lockx(pthrdpool->cslock);
  184.     list_add(&pthrd_desc->list, &pthrdpool->list);
  185.     __unlockx(pthrdpool->cslock);
  186.     return 0;

  188. __err:
  189.     __lockx_exit(pthrd_desc->cslock, pthrd_desc->attr);
  190.     free(pthrd_desc);

  192.     return -1;
  193. }

  195. static int __thrd_del(thrd_handle h)
  196. {
  197.     struct list_head *list_p;
  198.     thrd_desc_t* pthrd_desc=null;
  199.     

  201.     thrdpool_t* pthrdpool = thrdpool();    
  202.     __assert(pthrdpool);
  203.     __lockx(pthrdpool->cslock);
  204.     list_for_each(list_p, &(pthrdpool->list))
  205.     {
  206.         pthrd_desc = list_entry(list_p, thrd_desc_t, list);
  207.         if (h==pthrd_desc->h) {
  208.             
  209. #            ifdef WIN32
  210.             SET_EVENT(pthrd_desc->hevent_exit);
  211.             SET_EVENT(pthrd_desc->hevent_go);
  212. #            else
  213.             SET_EVENT(pthrd_desc->hevent_go);
  214.             sleep(1);
  215.             pthread_cancel(pthrd_desc->h);
  216. #            endif
  217.             
  218.             /* wait the thread exit */
  219.             if (hnull != pthrd_desc->h)
  220.             {
  221. #                ifdef WIN32
  222.                 /* terminal the log server thread until 1 second over */
  223.                 unsigned long state = WaitForSingleObject (pthrd_desc->h, 2000);
  224.                 switch (state)
  225.                 {
  226.                 case WAIT_TIMEOUT:
  227.                     {
  228.                         unsigned long exit_code = 0;
  229.                         if (0 != TerminateThread(pthrd_desc->h, exit_code))
  230.                             ; /* terminate the server thread */
  231.                         else
  232.                             ; /* terminate the LogSrv thread failed */
  233.                         break;
  234.                     }
  235.                 case WAIT_ABANDONED: break; /* the Thread abandoned */
  236.                 case WAIT_OBJECT_0: break; /* exit the server thread */
  237.                 default: break;
  238.                 }
  239.                 CloseHandle(pthrd_desc->h);
  240. #                else
  241.                 int state = pthread_join(pthrd_desc->h, null);
  242. #                endif
  243.                 pthrd_desc->h=hnull;
  244.             }

  246.             __lockx_exit(pthrd_desc->cslock,pthrd_desc->attr);

  248. #            ifdef WIN32
  249.             EXIT_EVENT(pthrd_desc->hevent_exit);
  250. #            endif

  252.             EXIT_EVENT(pthrd_desc->hevent_go);

  254.             /* remove itself from the list */
  255.             list_del(&pthrd_desc->list);
  256.             /* free the memory from the malloc of the thread descript struct */
  257.             free(pthrd_desc);
  258.             
  259.             break;
  260.         } // if (h==pthrd_desc->h)
  261.     }    
  262.     __unlockx(pthrdpool->cslock);
  263.     return 0;
  264. }

  266. static void __display(const thrd_desc_t* pthrd_desc)
  267. {
  268.     __assert(pthrd_desc);
  269.     printf("handle:%08x, hook: %s\n",
  270.         pthrd_desc->h, pthrd_desc->thrd_hook ? "installed":"uninstalled");
  271. }
  272.     
  273. static int __query()
  274. {
  275.     struct list_head *list_p;
  276.     thrd_desc_t* pthrd_desc=null;
  277.     
  278.     thrdpool_t* pthrdpool = thrdpool();    
  279.     __assert(pthrdpool);
  280.     __lockx(pthrdpool->cslock);
  281.     list_for_each(list_p, &(pthrdpool->list))
  282.     {
  283.         pthrd_desc = list_entry(list_p, thrd_desc_t, list);
  284.         __display(pthrd_desc);
  285.     }    
  286.     printf("------------------------------\n");
  287.     __unlockx(pthrdpool->cslock);
  288.     return 0;
  289. }

  291. static int __reg_notifier(struct notifierx_block *pnb)
  292. {
  293.     thrdpool_t* pthrdpool = thrdpool();    
  294.     __assert(pthrdpool&&pnb);
  295.     return notifierx_chain_register(&pthrdpool->thrdpool_chain, pnb);
  296. }

  298. static int __unreg_notifier(struct notifierx_block *pnb)
  299. {
  300.     thrdpool_t* pthrdpool = thrdpool();    
  301.     __assert(pnb);
  302.     return notifierx_chain_unregister(&pthrdpool->thrdpool_chain, pnb);
  303. }

  305. thrdpool_t* thrdpool()
  306. {
  307.     static thrdpool_t s_thrdpool;
  308.     return &s_thrdpool;
  309. }

  311. int thrdpool_init(int nthrds, int mod)
  312. {
  313.     thrdpool_t* pthrdpool = thrdpool();
  314.     int ret=0, count=0;
  315.     __assert(pthrdpool);

  317.     INIT_NOTIFIERX_HEAD(pthrdpool->thrdpool_chain);
  318.     __lockx_init(pthrdpool->cslock,pthrdpool->attr);

  320.     pthrdpool->mod=mod;

  322.     pthrdpool->list.next=&pthrdpool->list;
  323.     pthrdpool->list.prev=&pthrdpool->list;

  325.     pthrdpool->ops.thrd_alloc    =__thrd_alloc;
  326.     pthrdpool->ops.thrd_free    =__thrd_free;
  327.     pthrdpool->ops.wakeup    =__wakeup;
  328.     pthrdpool->ops.thrd_add =__thrd_add;
  329.     pthrdpool->ops.thrd_del    =__thrd_del;
  330.     pthrdpool->ops.query    =__query;
  331.     pthrdpool->ops.reg_notifier = __reg_notifier;
  332.     pthrdpool->ops.unreg_notifier = __unreg_notifier;
  333.     pthrdpool->ops.has_free_thrds = __has_free_thrds;

  335.     while (count<nthrds) {
  336.         ret=__thrd_add();
  337.         __assert(-1!=ret);
  338.         count++;
  339.     }
  340.     return 0;
  341. }

  343. int thrdpool_exit()
  344. {
  345.     struct list_head *list_p;
  346.     thrd_desc_t* pthrd_desc=null;
  347.     thrd_desc_t* pthrd_temp=null;
  348.     thrdpool_t* pthrdpool = thrdpool();    
  349.     __assert(pthrdpool);

  351.     list_for_each_safe(list_p, pthrd_temp, &(pthrdpool->list))
  352.     {
  353.         pthrd_desc = list_entry(list_p, thrd_desc_t, list);
  354.         if (hnull!=pthrd_desc->h)
  355.             __thrd_del(pthrd_desc->h);
  356.     }
  357.     __lockx_exit(pthrdpool->cslock,pthrdpool->attr);
  358.     EXIT_NOTIFIERX_HEAD(pthrdpool->thrdpool_chain);

  360.     return 0;
  361. }

  1. /**
  2.  * file: threadpool.h
  3.  * author: vincent.cws2008@gmail.com
  4.  * history:
  5.  * @2011-11-06: initial
  6.  * @2011-11-13: add the notifier to register
  7.  */

  9. #ifndef _THREADPOOL_H_
  10. #define _THREADPOOL_H_

  12. #include "list.h"
  13. #include "notifierx.h"
  14. #include "platform.h"


  17. typedef void (*thrd_hook_t)(void *pdata);

  19. #define stat_busy    1
  20. #define stat_free    2

  22. typedef struct thrd_msg_s{
  23.     thrd_handle h;
  24.     int stat;
  25.     void* v;
  26. }thrd_msg_t;

  28. #define INIT_THRD_MSG(st,hdl) \
  29. {\
  30.     (st).h=(hdl); \
  31.     (st).stat=stat_free;\
  32. }
  33. #define SET_THRD_STAT(st,s) \
  34. {\
  35.     (st).stat=(s);\
  36. }
  37. #define ST_SIZE_AND_VAL(st) sizeof(st),&(st)

  39. #define call_chain(chain,st,s) \
  40. {    \
  41.     SET_THRD_STAT((st), (s)); \
  42.     notifierx_call_chain(&(chain), ST_SIZE_AND_VAL(st)); \
  43. }

  45. #define reg_thread_notifierx(tp,fn) {        \
  46.     static struct notifierx_block fn##_nb;    \
  47.     (fn##_nb).notifierx_call=(fn);            \
  48.     notifierx_chain_register(&(tp).thrdpool_chain, &(fn##_nb)); \
  49. }

  51. typedef struct thrd_desc_s{
  52.     thrd_handle h;
  53.     thrd_hook_t thrd_hook;
  54.     void *pthrd_data;
  55.     event_t hevent_go;
  56.     event_t hevent_exit;
  57.     struct list_head list;
  58.     unsigned int ref_cnts;
  59.     lock_t cslock;
  60.     attr_t attr;
  61. }thrd_desc_t;


  64. #ifdef WIN32
  65. #define INIT_EVENT(h) { h=CreateEvent(NULL, TRUE, FALSE, NULL); }
  66. #define EXIT_EVENT(h) { if (hnull != (h)) CloseHandle(h); }
  67. #define SET_EVENT(h) { if (hnull != (h)) SetEvent(h); }
  68. #define RESET_EVENT(h) {if (hnull != (h)) ResetEvent(h); }
  69. #else
  70. #define INIT_EVENT(h) { pthread_cond_init(&(h), null); }
  71. #define EXIT_EVENT(h) { pthread_cond_destroy(&(h); }
  72. #define SET_EVENT(h) { pthread_cond_signal(&(h)); }
  73. #endif

  75. #define INIT_THRD_DESC(td) \
  76. { \
  77.     (td).h = 0; \
  78.     (td).thrd_hook = (thrd_hook_t)0; \
  79.     (td).pthrd_data = 0; \
  80.     INIT_EVENT((td).hevent_go); \
  81.     INIT_EVENT((td).hevent_exit); \
  82.     (td).list.prev=&(td).list; \
  83.     (td).list.next=&(td).list; \
  84.     (td).ref_cnts=0; \
  85. }

  87. #define CLEAN_THRD_HOOK(td) \
  88. { \
  89.     (td).thrd_hook = (thrd_hook_t)0; \
  90. }

  92. #define CLEAR_THRD_DESC(td) \
  93. { \
  94.     (td).thrd_hook = (thrd_hook_t)0; \
  95.     (td).pthrd_data = 0; \
  96. }

  98. #define SET_HANDLER_THRD_DESC(td,h,p) \
  99. { \
  100.     (td).thrd_hook = h; \
  101.     (td).pthrd_data = p; \
  102. }

  104. #define mod_tp_normal 0 /*mod==0: hook need NOT auto reset; */
  105. #define mod_tp_reset 1 /*mod==1: hook need to reset when next invoked */

  107. typedef struct thrdpool_ops{
  108.     thrd_handle (*thrd_alloc)(thrd_hook_t thrd_hook, void* pdata);
  109.     void (*thrd_free)(thrd_handle h);

  111.     int (*wakeup)(thrd_handle h);
  112.     
  113.     int (*thrd_add)();
  114.     int (*thrd_del)(thrd_handle h);
  115.     
  116.     int (*query)();
  117.     
  118.     int (*has_free_thrds)();

  120.     int (*reg_notifier)(struct notifierx_block *pnb);
  121.     int (*unreg_notifier)(struct notifierx_block *pnb);    
  122. }thrdpool_ops_t;

  124. typedef struct thrdpool_s{
  125.     struct thrdpool_ops ops;
  126.     struct list_head list;
  127.     struct notifierx_head thrdpool_chain;
  128.     lock_t cslock;
  129.     attr_t attr;
  130.     int mod;    /*mod==0: hook need NOT auto reset;
  131.      mod==1: hook need to reset when next invoked */
  132. }thrdpool_t;

  134. extern thrdpool_t* thrdpool();

  136. extern int thrdpool_init(int nthrds, int mod);
  137. extern int thrdpool_exit();




  142. #endif

  1. /**
  2.  * file: workqueuex.c
  3.  * author: vincent.cws2008@gmail.com
  4.  * history:
  5.  * @2011-11-15: initial
  6.  */

  8. #include "workqueuex.h"
  9. #include "threadpool.h"

  11. static struct workqueue_s *keventd_wq;

  13. tasks_sched_t *get_tasks_sched()
  14. {
  15.     static tasks_sched_t s_tasks_sched;
  16.     return &s_tasks_sched;
  17. }

  19. static struct workqueue_s* __create_wq(const char *name)
  20. {
  21.     struct workqueue_s *pnew_wq;
  22.     tasks_sched_t *pts = get_tasks_sched();
  23.     __assert(pts);

  25.     pnew_wq=malloc(sizeof(*pnew_wq));
  26.     __assert(pnew_wq);

  28.     INIT_WORKQUEUE(*pnew_wq,name,*pts);
  29.     
  30.     return pnew_wq;
  31. }

  33. static void __destroy_wq(struct workqueue_s *pwq)
  34. {
  35.     if (pwq) {
  36.         list_del(&pwq->list);
  37.         __lockx_exit(pwq->cslock, pwq->attr);
  38.         free(pwq);
  39.     }
  40. }

  42. static int __work_empty()
  43. {
  44.     struct list_head *wqlist_p;
  45.     struct workqueue_s *pwq_it;
  46. //    struct work_s *pwork_it;
  47.     
  48.     tasks_sched_t *pts=get_tasks_sched();
  49.     __assert(pts);
  50.     
  51.     list_for_each(wqlist_p, &(pts->wq_head))
  52.     {
  53.         pwq_it = list_entry(wqlist_p, workqueue_t, list);
  54.         __lockx(pwq_it->cslock);
  55.         if(!list_empty(&pwq_it->head)){
  56.             __unlockx(pwq_it->cslock);
  57.             return 0;
  58.         }
  59.         __unlockx(pwq_it->cslock);
  60.     }
  61.     return 1;
  62. }

  64. static struct work_s* __get_work()
  65. {
  66.     struct list_head *wqlist_p, *worklist_p;
  67.     struct workqueue_s *pwq_it;
  68.     struct work_s *pwork_it;

  70.     tasks_sched_t *pts=get_tasks_sched();
  71.     __assert(pts);
  72.     
  73.     list_for_each(wqlist_p, &(pts->wq_head))
  74.     {
  75.         pwq_it = list_entry(wqlist_p, workqueue_t, list);
  76.         __lockx(pwq_it->cslock);

  78.         list_for_each(worklist_p, &(pwq_it->head))
  79.         {
  80.             pwork_it = list_entry(worklist_p, work_t, list);
  81.             /* printf("[self:%08x,prev:%08x,next:%08x]\n",
  82.                 pwork_it, pwork_it->list.prev, pwork_it->list.next); */
  83.             list_del(&pwork_it->list);
  84.             __unlockx(pwq_it->cslock);
  85.             return pwork_it;
  86.         }
  87.         __unlockx(pwq_it->cslock);
  88.     }
  89.     return null;
  90. }

  92. static void __run(void)
  93. {
  94.     thrd_handle h=0;
  95.     thrdpool_t *ptp=thrdpool();
  96.     __assert(ptp);

  98.     if (ptp->ops.has_free_thrds()&&!__work_empty())
  99.     {
  100.         struct work_s* pnew_work=__get_work();
  101.         if(pnew_work) {
  102.             h=ptp->ops.thrd_alloc(pnew_work->fwork, pnew_work);
  103.             if (-1!=h) ptp->ops.wakeup(h);
  104.         }
  105.     }
  106. }

  108. int queue_work(struct workqueue_s *pwq, struct work_s *pwork)
  109. {
  110.     __assert(pwq&&pwork);
  111.     
  112.     __lockx(pwq->cslock);        

  114.     list_add(&pwork->list, &pwq->head);

  116.     __unlockx(pwq->cslock);
  117.     return 0;
  118. }

  120. int schedule_work(struct work_s *pwork)
  121. {
  122.     __assert(pwork);
  123.     queue_work(keventd_wq, pwork);
  124.     __run();
  125.     return 0;
  126. }


  129. static int __workqueue_thrd_callback(struct notifierx_block *pnb,
  130.                                      unsigned long nlen, const void *pdata)
  131. {
  132.     thrd_msg_t *pmsg=(thrd_msg_t*)pdata;
  133.     __assert(pnb);
  134.     
  135.     if (!pmsg) return 0;
  136.     switch(pmsg->stat) {
  137.     case stat_busy:
  138.         break;
  139.     case stat_free:
  140.         if(!__work_empty())
  141.             __run();
  142.         break;
  143.     default:
  144.         break;
  145.     }
  146.     return 0;
  147. }


  150. void init_tasks_sched(void)
  151. {
  152.     tasks_sched_t *pts=get_tasks_sched();
  153.     thrdpool_t *ptp=thrdpool();
  154.     
  155.     thrdpool_init(threads_number,mod_tp_reset);

  157.     __assert(pts&&ptp);
  158.     pts->pthrd_pool = ptp;
  159.     INIT_LIST_HEAD(&pts->wq_head);
  160.     pts->run == __run;

  162.     reg_thread_notifierx(*ptp,__workqueue_thrd_callback);
  163.     keventd_wq = __create_wq("events");
  164.     
  165. }

  167. void exit_tasks_sched(void)
  168. {
  169.     struct list_head *wqlist_p;
  170.     struct workqueue_s *pwq_it, *pwq_temp;
  171.     
  172.     tasks_sched_t *pts = get_tasks_sched();
  173.     __assert(pts);
  174.     
  175.     list_for_each_safe(wqlist_p, pwq_temp, &(pts->wq_head))
  176.     {
  177.         pwq_it = list_entry(wqlist_p, workqueue_t, list);
  178.         __destroy_wq(pwq_it);
  179.     }
  180.     
  181.     thrdpool_exit();
  182. }

  1. /**
  2.  * file: workqueuex.h
  3.  * author: vincent.cws2008@gmail.com
  4.  * history:
  5.  * @2011-11-15: initial
  6.  */

  8. #ifndef _WORKQUEUE_H_
  9. #define _WORKQUEUE_H_

  11. #include "workqueuex.h"

  13. #include "list.h"

  15. #include "platform.h"

  17. #define threads_number 1

  19. extern struct workqueue_s;

  21. extern struct work_s;
  22. typedef void (*fwork_t)(struct work_s *pwork);

  24. typedef struct tasks_sched{
  25.     struct thrdpool_s *pthrd_pool;
  26.     void (*run)(void);
  27.     struct list_head wq_head;
  28. }tasks_sched_t;


  31. typedef struct workqueue_s {
  32.     struct list_head list;    /* list member of the workqueues */
  33.     struct list_head head;    /* head of the sub-list */
  34.     const char *name;
  35.     int nthread;            /* -1 for any thread, else thread_number=nthread%numbers */
  36.     lock_t cslock;
  37.     attr_t attr;
  38. }workqueue_t;

  40. #define INIT_WORKQUEUE(wq,pname,ts) { \
  41.     (name)=(pname); \
  42.     __lockx_init((wq).cslock,(wq).attr); \
  43.     INIT_LIST_HEAD(&(wq).head); \
  44.     list_add(&(wq).list, &(ts).wq_head); \
  45.     (wq).nthread = 0; \
  46. }

  48. typedef struct work_s {
  49. //    void *pdata;
  50.     fwork_t fwork;
  51.     struct list_head list;
  52. }work_t;

  54. #define INIT_WORK(w,f) { \
  55.     (w).fwork=(f); \
  56.     INIT_LIST_HEAD(&(w).list); \
  57. }

  59. #define DECLARE_WORK(w, f)    \
  60.     struct work_s w = __WORK_INITIALIZER(n, f)

  62. #define NEW_WORK(pnew,f) {\
  63.     pnew=(work_t*)malloc(sizeof(work_t)); \
  64.     __assert(pnew); \
  65.     INIT_WORK(*pnew, (f)); \
  66. }

  68. #define FREE_WORK(pnew) {if(pnew) free(pnew);}


  71. extern int queue_work(struct workqueue_s *pwq, struct work_s *pwork);
  72. extern int schedule_work(struct work_s *pwork);

  74. extern tasks_sched_t *get_tasks_sched();

  76. extern void init_tasks_sched(void);
  77. extern void exit_tasks_sched(void);

  79. #endif

附件如下：

 workqueue.rar  


---- 代码一点一点积累，问题一点一点解决，莫浮躁，切记！

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

上一篇：线程池实现

下一篇：[转] Linux内核的时钟中断机制

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

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

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

16024965号-6