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
|
#define _POSIX_C_SOURCE 199309
#include <time.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#define DEFAULT_SIZE 10000000
#define DEFAULT_THREADS 1
void multithreaded_mergesort(int A[], int size, int num_threads);
typedef struct {
int* A;
int left;
int right;
} mergesort_arg;
typedef struct {
int* A;
int left;
int m;
int right;
} mergeing_arg;
void* mergesort_wrapper(const mergesort_arg* arg);
void mergesort(int A[], int left, int right);
void* merge_wrap(mergeing_arg* arg);
int merge( int A[] , int left , int m , int right);
int ms_diff(struct timespec start, struct timespec stop);
void show_usage(const char* exe);
int main(int argc, char* argv[]) {
int size = DEFAULT_SIZE;
int num_threads = DEFAULT_THREADS;
// Process command-line arguments.
if (argc == 3) {
char *end;
size = strtol(argv[1], &end, 10);
if (*end != 0) {
show_usage(argv[0]);
return EXIT_FAILURE;
}
num_threads = strtol(argv[2], &end, 10);
if (*end != 0) {
show_usage(argv[0]);
return EXIT_FAILURE;
}
if (size < 1 || num_threads < 1) {
show_usage(argv[0]);
return EXIT_FAILURE;
}
} else if (argc != 1) {
show_usage(argv[0]);
return EXIT_FAILURE;
}
// Seed the random number generator.
srand(time(NULL));
// Allocate and populate an array.
int* A = (int*)malloc(size * sizeof(int));
if (A == NULL) {
printf("ERROR: Unable to allocate array.\n");
return EXIT_FAILURE;
}
for (int i = 0; i < size; i++)
A[i] = rand();
// Start tracking execution time.
struct timespec start;
clock_gettime(CLOCK_REALTIME, &start);
multithreaded_mergesort(A, size, num_threads);
// Stop tracking execution time.
struct timespec stop;
clock_gettime(CLOCK_REALTIME, &stop);
printf("%d ms elapsed.\n", ms_diff(start, stop));
free(A);
return EXIT_SUCCESS;
}
void multithreaded_mergesort(int A[], int size, int num_threads) {
// Conceptually partition the array and use separate threads to
// independently sort each section. The thread that invoked this
// function will wait for all other threads to finish.
// Create function arguments for all threads.
mergesort_arg ms_args[num_threads];
for (int i = 0; i < num_threads; i++) {
ms_args[i].A = A;
ms_args[i].left = i * size / num_threads;
ms_args[i].right = (i + 1) * size / num_threads - 1;
}
// Create and start each thread.
pthread_t threads[num_threads];
for (int i = 0; i < num_threads; i++)
if (pthread_create(&threads[i], NULL, mergesort_wrapper, &ms_args[i]) != 0) {
printf("ERROR: Unable to create new thread.\n");
exit(EXIT_FAILURE);
}
// Wait for all threads to finish.
for (int i = 0; i < num_threads; i++)
pthread_join(threads[i], NULL);
// Perform "cleanup" work by merging remaining sections.
mergeing_arg m_args[num_threads / 2];
for (int values_left = num_threads; values_left > 1; values_left = (values_left + 1) / 2) {
// Set function arguments.
for (int i = 0; i < values_left / 2; i++) {
m_args[i].m = (((i + 1) * size / values_left - 1) + (i * size / values_left) ) / 2;
m_args[i].left = i * size / values_left;
m_args[i].right = (i + 1) * size / values_left - 1;
}
// Create and start threads.
for (int i = 0; i < values_left / 2; i++)
if (pthread_create(&threads[i], NULL, merge_wrap, &m_args[i]) != 0) {
printf("ERROR: Unable to create new thread.\n");
exit(EXIT_FAILURE);
}
// Wait for all threads to finish.
for (int i = 0; i < values_left / 2; i++)
pthread_join(threads[i], NULL);
}
}
void* mergesort_wrapper(const mergesort_arg* arg) {
fprintf(stderr, "Sorting section %d-%d.\n", arg->left, arg->right);
mergesort(arg->A, arg->left, arg->right);
return NULL;
}
void* merge_wrap(mergeing_arg* arg) {
merge(arg->A, arg->left, arg->m, arg->right);
return NULL;
}
// Indices left and right are inclusive.
void mergesort(int A[], int left, int right) {
if( left < right){
int m = (right + left )/2;
mergesort(A , left , m );
mergesort(A, m+1 , right);
merge( A, left, m, right);
}
}
int merge( int A[] , int left , int m , int right){
//int* A = (int*)malloc(size * sizeof(int));
int* b = (int*)malloc( ((right - left) + 1) * sizeof(int));
int i = left;
int j = m + 1;
int k = 0;
while( i <= m && j <= right ){
if(A[i] <= A[j])
b[k++] = A[i++];
else
b[k++] = A[j++];
}
while ( i <= m )
b[k++] = A[i++];
while( j <= right)
b[k++] = A[j++];
k--;
while( k >= 0 ){
A[left+ k] = b[k];
k--;
}
free(b);
}
int ms_diff(struct timespec start, struct timespec stop) {
return 1000 * (stop.tv_sec - start.tv_sec) + (stop.tv_nsec - start.tv_nsec) / 1000000;
}
void show_usage(const char* exe) {
printf("Usage: %s [<array-size> <num-threads>]\n", exe);
printf(" If omitted, <array-size> defaults to %d.\n", DEFAULT_SIZE);
printf(" If omitted, <num-threads> defaults to %d.\n", DEFAULT_THREADS);
}
|
