[原創(chuàng)]發(fā)布一個(gè)分布式計(jì)算框架雛形

redor

目前這個(gè)任務(wù)調(diào)度只負(fù)責(zé)任務(wù)的數(shù)據(jù)ID或者數(shù)據(jù)ID打包，不負(fù)責(zé)存儲(chǔ)數(shù)據(jù)本身，需要單獨(dú)
的數(shù)據(jù)存儲(chǔ)，任務(wù)計(jì)算節(jié)點(diǎn)需要自行設(shè)計(jì)數(shù)據(jù)存儲(chǔ)部分，這個(gè)程序包會(huì)提供一個(gè)獲取任務(wù)
和提交任務(wù)的客戶端接口libmtask，mtask.h；具體的實(shí)現(xiàn)參考btask.c 這個(gè)就是一個(gè)
benchmark。任務(wù)調(diào)度服務(wù)啟動(dòng)/usr/sbin/qtaskd -d -c /etc/qtaskd.ini
然后通過(guò)瀏覽器打開(kāi)http://serverIP:2080/　頁(yè)面上可以添加任務(wù)，這些任務(wù)有編號(hào),會(huì)顯示任務(wù)目前的完成情況。
btask使用方法就是對(duì)照自己配置好的任務(wù)，任務(wù)計(jì)算節(jié)點(diǎn)可以有三種類(lèi)型：
1. 只提交任務(wù)，比如數(shù)據(jù)源頭，這個(gè)節(jié)點(diǎn)不獲取任務(wù)（./btask -h IP -p 2066 -m 0 -q 1 -d）；
2. 獲取任務(wù)計(jì)算完成以后提交一個(gè)完成通知，另外同時(shí)提交給下一個(gè)任務(wù)(./btask -h IP -p 2066 -m 1 -q 2 -d)；
3. 結(jié)束計(jì)算節(jié)點(diǎn)，只獲取任務(wù)完成之后提交完成通知，不需要把任務(wù)繼續(xù)提交給一個(gè)任務(wù)(./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. 
    

復(fù)制代碼

寫(xiě)了一個(gè)調(diào)用的例子，不包括數(shù)據(jù)存取部分，演示了ID包的任務(wù)提交，　任務(wù)獲取，　任務(wù)完成，　還在弄具體的數(shù)據(jù)計(jì)算系統(tǒng)，弄完給大家分享。

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. }
     

復(fù)制代碼

hellioncu

看不太明白{:3_184:}

redor

回復(fù) 2# hellioncu


    就是支持N個(gè)節(jié)點(diǎn)同時(shí)計(jì)算的一個(gè)任務(wù)調(diào)度服務(wù)，看截圖。

newmax123

如何編寫(xiě)任務(wù)  發(fā)布任務(wù)呢

redor

回復(fù) 4# newmax123


    任務(wù)部分屬于業(yè)務(wù)層的東西，　調(diào)度中心只負(fù)責(zé)存儲(chǔ)數(shù)據(jù)的ID，具體的數(shù)據(jù)存儲(chǔ)需要使用獨(dú)立的存儲(chǔ)服務(wù)，我正在弄一個(gè)基于mongodb的計(jì)算系統(tǒng)，弄完放出來(lái)給大家看。
任務(wù)的獲取和任務(wù)的finish可以直接調(diào)用libmtask　mtask.h里的函數(shù)　btask.c里有例子。