[原创]发布一个分布式计算框架雏形

redor

目前这个任务调度只负责任务的数据ID或者数据ID打包，不负责存储数据本身，需要单独
的数据存储，任务计算节点需要自行设计数据存储部分，这个程序包会提供一个获取任务
和提交任务的客户端接口libmtask，mtask.h；具体的实现参考btask.c 这个就是一个
benchmark。任务调度服务启动/usr/sbin/qtaskd -d -c /etc/qtaskd.ini
然后通过浏览器打开http://serverIP:2080/　页面上可以添加任务，这些任务有编号,会显示任务目前的完成情况。
btask使用方法就是对照自己配置好的任务，任务计算节点可以有三种类型：
1. 只提交任务，比如数据源头，这个节点不获取任务（./btask -h IP -p 2066 -m 0 -q 1 -d）；
2. 获取任务计算完成以后提交一个完成通知，另外同时提交给下一个任务(./btask -h IP -p 2066 -m 1 -q 2 -d)；
3. 结束计算节点，只获取任务完成之后提交完成通知，不需要把任务继续提交给一个任务(./btask -h IP -p 2066 -m 2 -q 0 -d)。　



下载：
SRPM打包：http://libibase.googlecode.com/files/srpms-20110603183000.tar.gz
RPM打包：　http://libibase.googlecode.com/files/rpms-20110603183000.tar.gz
源码:　http://libibase.googlecode.com/files/qmtask-0.0.5.tar.gz


客户端API

1. 
   
2. /* set message task */
   
3. int mtask_set(MTASK *mtask, char *ip, int port, int mtaskid, int qtaskid);
   
4. /* connect to qtask
   
5. * -1 mtask is NULL
   
6. * -2 socket() failed
   
7. * -3 connect() failed
   
8. * */
   
9. int mtask_connect(MTASK *mtask);
   
10. 
    
11. /* get new task count
    
12. * -1 mtask is NULL
    
13. * -2 mtask->fd <= 0 and mtask_connect failed
    
14. * -3 write() task header failed
    
15. * -4 write() task list[] failed
    
16. * -5 read task header failed
    
17. * -6 malloc failed
    
18. * -7 read task list[] failed
    
19. * ret >= 0 mhead.packetid
    
20. * */
    
21. int mtask_commit(MTASK *mtask, int flag, char *packet, int packet_len);
    
22. 
    
23. /* push packet
    
24. * return value
    
25. * -1 mtask is NULL
    
26. * -2 connection is bad
    
27. * -3 Invalid packet data
    
28. * -4 write() header failed
    
29. * -5 write() packet failed
    
30. * -6 recv() header failed
    
31. * */
    
32. int mtask_push(MTASK *mtask, int flag, char *packet, int packet_len);
    
33. 
    
34. /* pop packet
    
35. * return value
    
36. * -1 mtask is NULL
    
37. * -2 connection is bad
    
38. * -3 Invalid commitid
    
39. * -4 write() header failed
    
40. * -5 recv() header failed
    
41. * -6 malloc() for packet failed
    
42. * -7 recv() packet failed
    
43. */
    
44. int mtask_pop(MTASK *mtask);
    
45. 
    
46. /* over packet
    
47. * return value
    
48. * -1 mtask is NULL
    
49. * -2 connection is bad
    
50. * -3 Invalid packetid and commitid
    
51. * -4 write() header failed
    
52. * -5 recv() header failed
    
53. * */
    
54. int mtask_finish(MTASK *mtask, int flag);
    
55. 
    
56. /* close message task */
    
57. void mtask_close(MTASK *mtask);
    
58. 
    

复制代码

写了一个调用的例子，不包括数据存取部分，演示了ID包的任务提交，　任务获取，　任务完成，　还在弄具体的数据计算系统，弄完给大家分享。

1. 
   
2. #include <stdio.h>
   
3. #include <stdlib.h>
   
4. #include <string.h>
   
5. #include <unistd.h>
   
6. #include <errno.h>
   
7. #include <sys/socket.h>
   
8. #include <netinet/in.h>
   
9. #include <arpa/inet.h>
   
10. #include <sys/types.h>
    
11. #include "mtask.h"
    
12. int main(int argc, char **argv)
    
13. {
    
14.     int i = 0, x = 0, mid = 0, qid = 0, port = 0, flag = 0,
    
15.         isdaemon = 0, isout = 0, len = 0, packetid = 0;
    
16.     char *ip = NULL, *packet = NULL, block[MTASK_PACKET_MAX * sizeof(int64_t)], ch = 0;
    
17.     int64_t old = 0, *list = NULL;
    
18.     MTASK mtask = {0};
    
19.     pid_t pid = 0;
    
20. 
    
21.     /* get configure file */
    
22.     while((ch = getopt(argc, argv, "h:p:m:q:d")) != -1)
    
23.     {
    
24.         switch(ch)
    
25.         {
    
26.             case 'h':
    
27.                 ip = optarg;
    
28.                 break;
    
29.             case 'p':
    
30.                 port = atoi(optarg);
    
31.                 break;
    
32.             case 'm':
    
33.                 mid = atoi(optarg);
    
34.                 break;
    
35.             case 'q':
    
36.                 qid = atoi(optarg);
    
37.                 break;
    
38.             case 'o':
    
39.                 isout = 1;
    
40.                 break;
    
41.             case 'd':
    
42.                 isdaemon = 1;
    
43.                 break;
    
44.             default:
    
45.                 break;
    
46.             }
    
47.     }
    
48.     if(ip == NULL || port <= 0 || (qid == 0 && mid == 0))
    
49.     {
    
50.         fprintf(stderr, "Usage:%s -h host -p port -m commitid -q queueid -o output -d working as daemon\n", argv[0]);
    
51.         _exit(-1);
    
52.     }
    
53.     /* daemon */
    
54.     if(isdaemon)
    
55.     {
    
56.         pid = fork();
    
57.         switch (pid) {
    
58.             case -1:
    
59.                 perror("fork()");
    
60.                 exit(EXIT_FAILURE);
    
61.                 break;
    
62.             case 0: /* child process */
    
63.                 if(setsid() == -1)
    
64.                     exit(EXIT_FAILURE);
    
65.                 break;
    
66.             default:/* parent */
    
67.                 _exit(EXIT_SUCCESS);
    
68.                 break;
    
69.         }
    
70.     }
    
71.     if(mtask_set(&mtask, ip, port, mid, qid) == 0
    
72.             && mtask_connect(&mtask) == 0)
    
73.     {
    
74.         list = (int64_t *)block;
    
75.         if(mid <= 0 && qid > 0)
    
76.         {
    
77.             do
    
78.             {
    
79.                 old = random();
    
80.                 flag = 0;
    
81.                 if((old%33) == 0) flag = MTASK_TO_QHEAD;
    
82.                 x = 0;
    
83.                 while(x < MTASK_PACKET_MAX)
    
84.                 {
    
85.                     list[x++] = (int64_t)random();
    
86.                 }
    
87.                 if((packetid = mtask_push(&mtask, flag, block,
    
88.                                 sizeof(int64_t) * MTASK_PACKET_MAX)) >= 0)
    
89.                 {
    
90.                     if(isout)fprintf(stdout, "1:{%d:{packetid:%d}}\n", i, packetid);
    
91.                     ++i;
    
92.                 }
    
93.                 else
    
94.                 {
    
95.                     sleep(1);
    
96.                 }
    
97.             }while(1);
    
98.         }
    
99.         else
    
100.         {
     
101.             /*
     
102.             packetid = 0;
     
103.             do
     
104.             {
     
105.                 old = random();
     
106.                 flag = 0;
     
107.                 if((old%33) == 0) flag = MTASK_TO_QHEAD;
     
108.                 if((packetid = mtask_commit(&mtask, flag, NULL, 0)) > 0)
     
109.                 {
     
110.                     packet = mtask.packet;
     
111.                     len = mtask.length;
     
112.                     if(isout)fprintf(stdout, "2:{%d:{packetid:%d length:%d}}\n", i, packetid, len);
     
113.                     ++i;
     
114.                 }
     
115.                 else
     
116.                 {
     
117.                     packet = NULL;
     
118.                     len = 0;
     
119.                     sleep(1);
     
120.                 }
     
121.             }while(1);
     
122.             */
     
123.             packetid = 0;
     
124.             do
     
125.             {
     
126.                 old = random();
     
127.                 flag = 0;
     
128.                 if((old%33) == 0) flag = MTASK_TO_QHEAD;
     
129.                 if((packetid = mtask_pop(&mtask)) > 0)
     
130.                 {
     
131.                     packet = mtask.packet;
     
132.                     len = mtask.length;
     
133.                     if(isout)fprintf(stdout, "2:{%d:{packetid:%d length:%d}}\n", i, packetid, len);
     
134.                     ++i;
     
135.                     mtask_finish(&mtask, flag);
     
136.                 }
     
137.                 else
     
138.                 {
     
139.                     packet = NULL;
     
140.                     len = 0;
     
141.                     sleep(1);
     
142.                 }
     
143.             }while(1);
     
144.         }
     
145.         mtask_close(&mtask);
     
146.     }
     
147.     return 0;
     
148. }
     

复制代码

hellioncu

看不太明白{:3_184:}

redor

回复 2# hellioncu


    就是支持N个节点同时计算的一个任务调度服务，看截图。

newmax123

如何编写任务  发布任务呢

redor

回复 4# newmax123


    任务部分属于业务层的东西，　调度中心只负责存储数据的ID，具体的数据存储需要使用独立的存储服务，我正在弄一个基于mongodb的计算系统，弄完放出来给大家看。
任务的获取和任务的finish可以直接调用libmtask　mtask.h里的函数　btask.c里有例子。