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
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
|
#include "mapper.h"
// combined value list corresponding to a word <1,1,1,1....>
valueList *createNewValueListNode(char *value){
valueList *newNode = (valueList *)malloc (sizeof(valueList));
strcpy(newNode -> value, value);
newNode -> next = NULL;
return newNode;
}
// insert new count to value list
valueList *insertNewValueToList(valueList *root, char *count){
valueList *tempNode = root;
if(root == NULL)
return createNewValueListNode(count);
while(tempNode -> next != NULL)
tempNode = tempNode -> next;
tempNode -> next = createNewValueListNode(count);
return root;
}
// free value list
void freeValueList(valueList *root) {
if(root == NULL) return;
valueList *tempNode = NULL;
while (root != NULL){
tempNode = root;
root = root -> next;
free(tempNode);
}
}
// create <word, value list>
intermediateDS *createNewInterDSNode(char *word, char *count){
intermediateDS *newNode = (intermediateDS *)malloc (sizeof(intermediateDS));
strcpy(newNode -> key, word);
newNode -> value = NULL;
newNode -> value = insertNewValueToList(newNode -> value, count);
newNode -> next = NULL;
return newNode;
}
// insert <word, value> to intermediate DS
intermediateDS *insertPairToInterDS(intermediateDS *root, char *word, char *count){
intermediateDS *tempNode = root;
if(root == NULL)
return createNewInterDSNode(word, count);
while(tempNode -> next != NULL) {
if(strcmp(tempNode -> key, word) == 0){
tempNode -> value = insertNewValueToList(tempNode -> value, count);
return root;
}
tempNode = tempNode -> next;
}
if(strcmp(tempNode -> key, word) == 0){
tempNode -> value = insertNewValueToList(tempNode -> value, count);
} else {
tempNode -> next = createNewInterDSNode(word, count);
}
return root;
}
// free intermediate DS
void freeInterDS(intermediateDS *root) {
if(root == NULL) return;
intermediateDS *tempNode = NULL;
while (root != NULL){
tempNode = root;
root = root -> next;
freeValueList(tempNode -> value);
free(tempNode);
}
}
// emit the <key, value> into intermediate DS
void emit(char *key, char *value) {
// printf("To be emitted: %s, %s\n", (char *)key, value);
interDS = insertPairToInterDS(interDS, key, value);
}
// map function
void map(char *chunkData){
char *cData = chunkData;
char *buffer;
int i = 0;
while((buffer = getWord(cData, &i)) != NULL){
emit(buffer, "1");
free(buffer);
buffer = NULL;
}
}
// generate the file name for word.txt
char *generateWordFileName(char *word){
char fileName[MAXKEYSZ];
memset(fileName, '\0', MAXKEYSZ);
strcpy(fileName, word);
strcat(fileName, ".txt");
fileName[strlen(fileName)] = '\0';
char *wordFileName = (char*)malloc(sizeof(char) * (2 * MAXKEYSZ));
memset(wordFileName, '\0', (2 * MAXVALUESZ));
strcpy(wordFileName, mapOutDir);
strcat(wordFileName, "/");
strcat(wordFileName, fileName);
wordFileName[strlen(wordFileName)] = '\0';
return wordFileName;
}
// write intermediate data to separate word.txt files
// Each file will have only one line <word 1 1 1 1 1 ...>
void writeIntermediateDS() {
intermediateDS *travNode = interDS;
while(travNode != NULL) {
// create file word.txt
// content : word 1 1 1 1 1 1.....
char *wordFileName = generateWordFileName(travNode -> key);
int fd = open(wordFileName, O_CREAT | O_WRONLY, 0777);
if (fd < 0){
printf("ERROR: Cannot open the file %s\n", wordFileName);
exit(0);
}
// word write
int ret = write(fd, travNode -> key, strlen(travNode -> key));
if(ret < 0){
printf("ERROR: Cannot write to file %s\n", wordFileName);
exit(0);
}
// append space
ret = write(fd, " ", 1);
if(ret < 0){
printf("ERROR: Cannot write to file %s\n", wordFileName);
exit(0);
}
// append value list
valueList *tNode = travNode -> value;
while(tNode -> next != NULL){
ret = write(fd, tNode -> value, strlen(tNode -> value)); // space after the word
if(ret < 0){
printf("ERROR: Cannot write to file %s\n", wordFileName);
exit(0);
}
// append space
ret = write(fd, " ", 1);
if(ret < 0){
printf("ERROR: Cannot write to file %s\n", wordFileName);
exit(0);
}
tNode = tNode -> next;
}
// last value
ret = write(fd, tNode -> value, strlen(tNode -> value)); // space after the word
if(ret < 0){
printf("ERROR: Cannot write to file %s\n", wordFileName);
exit(0);
}
// append newline
ret = write(fd, "\n", 1);
if(ret < 0){
printf("ERROR: Cannot write to file %s\n", wordFileName);
exit(0);
}
//close file
close(fd);
// go to next word in intermediate DS
travNode = travNode -> next;
}
}
int main(int argc, char *argv[]) {
if (argc < 2) {
printf("Less number of arguments.\n");
printf("./mapper mapperID\n");
exit(0);
}
// initializing global variables
mapperID = strtol(argv[1], NULL, 10);
interDS = NULL;
//create folder specifically for this mapper in output/MapOut
mapOutDir = createMapDir(mapperID);
int count = 0;
while(1) {
char chunkData[chunkSize + 1];
memset(chunkData, '\0', chunkSize + 1);
char *retChunk = getChunkData(mapperID);
if(retChunk == NULL) {
break;
}
count++;
strcpy(chunkData, retChunk);
free(retChunk);
map(chunkData);
}
//student code
writeIntermediateDS();
freeInterDS(interDS);
return 0;
}
|
