MapReduce/worker.cpp at master · Jiaran/MapReduce · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
#include "worker.h"

extern NodeInfo FILE_SYSTEM;
class Update : public TimerProcess{
private:
  int fdtoMaster;
  int workerID;
public:
  Update(int fd , int id){
    fdtoMaster = fd;
    workerID=id;
  }
  virtual void invoke(){
    char buf[20];
    sprintf(buf,"UPDATE*%d\n",workerID);
    Rio_writep(fdtoMaster, buf, strlen(buf));

  }

};

CWorker:: CWorker(const char * port,const char * ip,NodeInfo& master ):Node(port,ip){
  this->master = master;
}

void CWorker:: run(){
  Node::run();
  join();
  while(true){
    std::string s;
    std::cin>>s;
    if(s=="q"){
      break;
    }

  }
  Close(masterfd);
}

void CWorker:: join(){
  int fd = connectTo(master);
  char buf[MAXLINE];
  sprintf(buf,"%s","JOIN\n");
  Rio_writep(fd, buf, strlen(buf));

  sendNodeInfo(fd);

  int numBytes=Rio_readp(fd, buf, MAXLINE);
  buf[numBytes]=0;
  workerID=atoi(buf);
  printf("buf is ::::%s,,,workerID is %d\n",buf,workerID);
  masterfd = fd;
  timerProcess = new Update( masterfd,workerID);
  timerStart();

}

void CWorker:: processRequest(rio_t & client, int clientfd){
  char buf[MAXLINE];
  int numBytes;
  char * saveptr;
  char * cmd;
  numBytes = Rio_readlineb(&client, buf, MAXLINE);
  cmd = strtok_r(buf, " \r\n",&saveptr);

  if(cmd==NULL){
    Close(clientfd);
    return;
  }
  else if(strcmp(cmd,"MAP")==0){
    cmd = strtok_r(NULL, " \r\n",&saveptr);
    int jobID = atoi(cmd);
    printf("MAP it is , job ID %s\n",cmd);

    //handle map work;//////////////////////////
    myMapFunction.goMap(clientfd, client,jobID);
  }
  else if(strcmp(cmd,"REDUCE")==0){

    cmd = strtok_r(NULL, " \r\n",&saveptr);
    printf("REDUCE , job ID %s\n",cmd);
    int jobID = atoi(cmd);
    /*
    while(numBytes = Rio_readlineb(&client, buf, MAXLINE)>0 ){
      printf("buf is %s\n",buf);
    }
    */
    //handle reduce work/////////////////////
    myReduceFunction.getInfo(clientfd, client, jobID);

    complete(jobID);

  }
  else if(strcmp(cmd,"DONE")==0){
    //clear Map and Reduce;///////////////////////
    myMapFunction.printMap();
    myMapFunction.clearMap();
    myReduceFunction.clearReduce();
    printf("done\n");
  }
  else if(strcmp(cmd,"MODIFY")==0){
    //MODIFY JOBID\n IP*PORT
    cmd = strtok_r(NULL, " \r\n",&saveptr);
    int jobID =atoi(cmd);
    NodeInfo n= readNodeInfo(client);
    /////////////////////////////////
    myReduceFunction.changeInfo(jobID, n);
    printf("MODIFY jobID is %d, ip %s, port %s\n",jobID, n.IP, n.port);
  }
  else if(strcmp(cmd,"GETMAP")==0){
    cmd = strtok_r(NULL, " \r\n",&saveptr);
    int jobID =atoi(cmd);
    //call get map
    myMapFunction.getMap(clientfd,jobID);

  }

  Close(clientfd);

}


void CWorker:: complete(int jobID){
  int fd = connectTo(master);
  char buf[MAXLINE];

  sprintf(buf,"COMPLETE\n%d*%d\n",jobID,workerID);
  Rio_writep(fd, buf, strlen(buf));
  Close(fd);


}