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
|
#include "utils.h"
//Receive from send and return the chonk
char *getChunkData(int mapperID) {
//Message
struct msgBuffer* message;
//Queue ID, not sure what it actually does
int mid;
//Queue Key
key_t key = 10;
mid = msgget(key, 0666);
if (mid < 0) {
perror("Cannot open queue.\n");
return NULL;
}
msgrcv(mid, (void *) &message, MSGSIZE, mapperID, 0);
if (strcmp("END", message -> msgText)) {
struct msgBuffer ACK = {mapperID, "ACK"};
msgsnd(mid, (void *) &ACK, MSGSIZE, 0);
}
return message -> msgText;
}
// sends chunks of size 1024 to the mappers in RR fashion
void sendChunkData(char *inputFile, int nMappers) {
}
// hash function to divide the list of word.txt files across reducers
//http://www.cse.yorku.ca/~oz/hash.html
int hashFunction(char* key, int reducers){
unsigned long hash = 0;
int c;
while ((c = *key++)!='\0')
hash = c + (hash << 6) + (hash << 16) - hash;
return (hash % reducers);
}
int getInterData(char *key, int reducerID) {
}
void shuffle(int nMappers, int nReducers) {
}
// check if the character is valid for a word
int validChar(char c){
return ((tolower(c) >= 'a') && tolower(c <='z')) ||
(c >= '0' && c <= '9');
}
char *getWord(char *chunk, int *i){
char *buffer = (char *)malloc(sizeof(char) * chunkSize);
memset(buffer, '\0', chunkSize);
int j = 0;
while((*i) < strlen(chunk)) {
// read a single word at a time from chunk
// printf("%d\n", i);
if (chunk[(*i)] == '\n' || chunk[(*i)] == ' ' || !validChar(chunk[(*i)]) || chunk[(*i)] == 0x0) {
buffer[j] = '\0';
if(strlen(buffer) > 0){
(*i)++;
return buffer;
}
j = 0;
(*i)++;
continue;
}
buffer[j] = chunk[(*i)];
j++;
(*i)++;
}
if(strlen(buffer) > 0)
return buffer;
return NULL;
}
void createOutputDir(){
mkdir("output", ACCESSPERMS);
mkdir("output/MapOut", ACCESSPERMS);
mkdir("output/ReduceOut", ACCESSPERMS);
}
char *createMapDir(int mapperID){
char *dirName = (char *) malloc(sizeof(char) * 100);
memset(dirName, '\0', 100);
sprintf(dirName, "output/MapOut/Map_%d", mapperID);
mkdir(dirName, ACCESSPERMS);
return dirName;
}
void removeOutputDir(){
pid_t pid = fork();
if(pid == 0){
char *argv[] = {"rm", "-rf", "output", NULL};
if (execvp(*argv, argv) < 0) {
printf("ERROR: exec failed\n");
exit(1);
}
exit(0);
} else{
wait(NULL);
}
}
void bookeepingCode(){
removeOutputDir();
sleep(1);
createOutputDir();
}
|
