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
|
/* Copyright (c) 2007-2009, Stanford University
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
* * Neither the name of Stanford University nor the
* names of its contributors may be used to endorse or promote products
* derived from this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY STANFORD UNIVERSITY ``AS IS'' AND ANY
* EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
* WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
* DISCLAIMED. IN NO EVENT SHALL STANFORD UNIVERSITY BE LIABLE FOR ANY
* DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
* (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
* LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
* ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
* SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/
#ifndef MAP_REDUCE_H_
#define MAP_REDUCE_H_
#include <sys/types.h>
#include <stdbool.h>
/* Standard data types for the function arguments and results */
/* Argument to map function. This is specified by the splitter function.
* length - number of elements of data. The default splitter function gives
length in terms of the # of elements of unit_size bytes.
* data - data to process of a user defined type
*/
typedef struct
{
intptr_t length;
void *data;
} map_args_t;
/* Single element of result
* key - pointer to the key
* val - pointer to the value
*/
typedef struct
{
void *key;
void *val;
} keyval_t;
/* List of results
* length - number of key value pairs
* data - array of key value pairs
*/
typedef struct
{
int length;
keyval_t *data;
} final_data_t;
/* Scheduler function pointer type definitions */
/* Map function takes in map_args_t, as supplied by the splitter
* and emit_intermediate() should be called on any key value pairs
* in the intermediate result set.
*/
typedef void (*map_t)(map_args_t *);
struct iterator_t;
typedef struct iterator_t iterator_t;
int iter_next (iterator_t *itr, void **);
int iter_size (iterator_t *itr);
/* Reduce function takes in a key pointer, a list of value pointers, and a
* length of the list. emit() should be called on any key value pairs
* in the result set.
*/
typedef void (*reduce_t)(void *, iterator_t *itr);
/* Combiner function takes in an iterator for a particular key,
* and returns a reduced value. The operation should be identical to the
* reduce function, except that this function returns the reduced value
* directly. */
typedef void *(*combiner_t)(iterator_t *itr);
/* Splitter function takes in a pointer to the input data, an interger of
* the number of bytes requested, and an uninitialized pointer to a
* map_args_t pointer. The result is stored in map_args_t. The splitter
* should return 1 if the result is valid or 0 if there is no more data.
*/
typedef int (*splitter_t)(void *, int, map_args_t *);
/* Locator function takes in a pointer to map_args_t, and returns
* the memory address where this map task would be heavily accessing.
* The runtime would then try to dispatch the task to a thread that
* is nearby the physical memory that backs the address. */
typedef void* (*locator_t)(map_args_t *);
/* Partition function takes in the number of reduce tasks, a pointer to
* a key, and the lendth of the key in bytes. It assigns a key to a reduce task.
* The value returned is the # of the reduce task where the key will be processed.
* This value should be the same for all keys that are equal.
*/
typedef int (*partition_t)(int, void *, int);
/* key_cmp(key1, key2) returns:
* 0 if key1 == key2
* + if key1 > key2
* - if key1 < key2
*/
typedef int (*key_cmp_t)(const void *, const void*);
/* The arguments to operate the runtime. */
typedef struct
{
void * task_data; /* The data to run MapReduce on.
* If splitter is NULL, this should be an array. */
off_t data_size; /* Total # of bytes of data */
int unit_size; /* # of bytes for one element
* (if necessary, on average) */
map_t map; /* Map function pointer, must be user defined */
reduce_t reduce; /* If NULL, identity reduce function is used,
* which emits a keyval pair for each val. */
combiner_t combiner; /* If NULL, no combiner would be called. */
splitter_t splitter; /* If NULL, the array splitter is used.*/
locator_t locator; /* If NULL, no locality based optimization is
performed. */
key_cmp_t key_cmp; /* Key comparison function.
Must be user defined.*/
final_data_t *result; /* Pointer to output data.
* Must be allocated by user */
/*** Optional arguments must be zero if not used ***/
partition_t partition; /* Default partition function is a
* hash function */
/* Creates one emit queue for each reduce task,
* instead of per reduce thread. This improves
* time to emit if data is emitted in order,
* but can increase merge time. */
bool use_one_queue_per_task;
int L1_cache_size; /* Size of L1 cache in bytes */
int num_map_threads; /* # of threads to run map tasks on.
* Default is one per processor */
int num_reduce_threads; /* # of threads to run reduce tasks on.
* Default is one per processor */
int num_merge_threads; /* # of threads to run merge tasks on.
* Default is one per processor */
int num_procs; /* Maximum number of processors to use. */
int proc_offset; /* number of procs to skip for thread binding */
/* (useful if you have multiple MR's running
* and you don't want them binding to the same
* hardware thread). */
float key_match_factor; /* Magic number that describes the ratio of
* the input data size to the output data size.
* This is used as a hint. */
} map_reduce_args_t;
/* Runtime defined functions. */
/* MapReduce initialization function. Called once per process. */
int map_reduce_init ();
/* MapReduce finalization function. Called once per process. */
int map_reduce_finalize ();
/* The main MapReduce engine. This is the function called by the application.
* It is responsible for creating and scheduling all map and reduce tasks, and
* also organizes and maintains the data which is passed from application to
* map tasks, map tasks to reduce tasks, and reduce tasks back to the
* application. Results are stored in args->result. A return value less than zero
* represents an error. This function is not thread safe.
*/
int map_reduce (map_reduce_args_t * args);
/* This should be called from the map function. It stores a key with key_size
* bytes and a value in the intermediate queues for processing by the reduce
* task. The runtime will call partiton function to assign the key to a
* reduce task.
*/
void emit_intermediate(void *key, void *val, int key_size);
/* This should be called from the reduce function. It stores a key and a value
* in the reduce queue. This will be in the final result array.
*/
void emit(void *key, void *val);
/* This is the built in partition function which is a hash. It is global so
* the user defined partition function can call it.
*/
int default_partition(int reduce_tasks, void* key, int key_size);
#endif // MAP_REDUCE_H_
|
