source: trunk/user/sort/sort.c

Last change on this file was 676, checked in by alain, 6 years ago

Introduce chat application to test the named pipes.
  • Property svn:executable set to *
File size: 14.0 KB
Line 
1/*
2 * sort.c - Parallel sort
3 *
4 * Author Cesar Fuguet Tortolero (2013)
5 * Alain Greiner (2019)
6 *
7 * Copyright (c) UPMC Sorbonne Universites
8 *
9 * This is free software; you can redistribute it and/or modify it
10 * under the terms of the GNU General Public License as published by
11 * the Free Software Foundation; version 2.0 of the License.
12 *
13 * It is distributed in the hope that it will be useful, but
14 * WITHOUT ANY WARRANTY; without even the implied warranty of
15 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
16 * General Public License for more details.
17 *
18 * You should have received a copy of the GNU General Public License
19 * along with ALMOS-MKH; if not, write to the Free Software Foundation,
20 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
21 */
22
23///////////////////////////////////////////////////////////////////////////////
24// This multi-threaded application implement a multi-stage sort.
25// It has been writen by Cesar Fuget Tortolero in 2013.
26// It has been ported on ALMOS-MKH by Alain Greiner in 2019.
27//
28// There is one thread per physical cores.
29// Computation is organised as a binary tree:
30// - All threads execute in parallel a buble sort on a sub-array during the
31// the first stage of parallel sort,
32// - The number of participating threads is divided by 2 at each next stage,
33// to make a merge sort, on two subsets of previous stage.
34//
35// Number_of_stages = number of barriers = log2(Number_of_threads)
36//
37// The various stages are separated by synchronisation barriers, and the
38// main thread uses the join syscall to check that all threads completed
39// before printing the computation time (sequencial & parallel).
40// These results can be - optionnaly - registered in an instrumentation file.
41//
42// Constraints :
43// - It supports up to 1024 cores: x_size, y_size, and ncores must be
44// power of 2 (max 16*16 clusters / max 4 cores per cluster)
45// _ The array of values to be sorted (ARRAY_LENGTH) must be power of 2
46// larger than the number of cores.
47///////////////////////////////////////////////////////////////////////////////
48
49#include <stdio.h>
50#include <stdlib.h>
51#include <unistd.h>
52#include <pthread.h>
53#include <almosmkh.h>
54#include <hal_macros.h>
55
56#define ARRAY_LENGTH 64 // number of items
57#define MAX_THREADS 1024 // 16 * 16 * 4
58
59#define X_MAX 16 // max number of clusters in a row
60#define Y_MAX 16 // max number of clusters in a column
61#define CORES_MAX 4 // max number of cores in a cluster
62#define CLUSTERS_MAX X_MAX * Y_MAX
63
64#define USE_DQT_BARRIER 1 // use DQT barrier if non zero
65#define DISPLAY_ARRAY 0 // display items values before and after
66#define DEBUG_MAIN 0 // trace main function
67#define DEBUG_SORT 0 // trace sort function
68#define CHECK_RESULT 0 // for debug
69#define INSTRUMENTATION 1 // register computation times on file
70
71////////////////////////////////////////////////////////////////////////////////////
72// Sort specific global variables
73////////////////////////////////////////////////////////////////////////////////////
74
75int array0[ARRAY_LENGTH]; // values to sort
76int array1[ARRAY_LENGTH];
77
78unsigned int threads; // total number of working threads
79
80pthread_barrier_t barrier; // synchronisation variables
81
82/////////////////////////////////////////////////////////////////////////////////////
83// Global variables required by parallel_pthread_create()
84/////////////////////////////////////////////////////////////////////////////////////
85
86
87////////////////////////////////////
88static void bubbleSort( int * array,
89 unsigned int length,
90 unsigned int init_pos )
91{
92 unsigned int i;
93 unsigned int j;
94 int aux;
95
96 for(i = 0; i < length; i++)
97 {
98 for(j = init_pos; j < (init_pos + length - i - 1); j++)
99 {
100 if(array[j] > array[j + 1])
101 {
102 aux = array[j + 1];
103 array[j + 1] = array[j];
104 array[j] = aux;
105 }
106 }
107 }
108} // end bubbleSort()
109
110
111///////////////////////////////////
112static void merge( const int * src, // source array
113 int * dst, // destination array
114 int length, // number of items in a subset
115 int init_pos_src_a, // index first item in src subset A
116 int init_pos_src_b, // index first item in src subset B
117 int init_pos_dst ) // index first item in destination
118{
119 int i;
120 int j;
121 int k;
122
123 i = 0;
124 j = 0;
125 k = init_pos_dst;
126
127 while((i < length) || (j < length))
128 {
129 if((i < length) && (j < length))
130 {
131 if(src[init_pos_src_a + i] < src[init_pos_src_b + j])
132 {
133 dst[k++] = src[init_pos_src_a + i];
134 i++;
135 }
136 else
137 {
138 dst[k++] = src[init_pos_src_b + j];
139 j++;
140 }
141 }
142 else if(i < length)
143 {
144 dst[k++] = src[init_pos_src_a + i];
145 i++;
146 }
147 else
148 {
149 dst[k++] = src[init_pos_src_b + j];
150 j++;
151 }
152 }
153} // end merge()
154
155///////////////////////////////
156void * sort( void * arguments )
157{
158 unsigned int i;
159 int * src_array = NULL;
160 int * dst_array = NULL;
161
162 // get arguments
163 pthread_parallel_work_args_t * ptr = (pthread_parallel_work_args_t *)arguments;
164
165 unsigned int tid = ptr->tid;
166 pthread_barrier_t * parent_barrier = ptr->barrier;
167
168 unsigned int items = ARRAY_LENGTH / threads;
169 unsigned int stages = __builtin_ctz( threads ) + 1;
170
171#if DEBUG_SORT
172printf("\n[sort] start : ptr %x / tid %d / threads %d / parent_barrier %x\n",
173ptr, tid, threads, parent_barrier );
174#endif
175
176 bubbleSort( array0, items, items * tid );
177
178#if DEBUG_SORT
179printf("\n[sort] thread[%d] : stage 0 completed\n", tid );
180#endif
181
182 /////////////////////////////////
183 pthread_barrier_wait( &barrier );
184
185#if DEBUG_SORT
186printf("\n[sort] thread[%d] exit barrier 0\n", tid );
187#endif
188
189 // the number of threads contributing to sort is divided by 2
190 // and the number of items is multiplied by 2 at each next stage
191 for ( i = 1 ; i < stages ; i++ )
192 {
193 if((i % 2) == 1) // odd stage
194 {
195 src_array = array0;
196 dst_array = array1;
197 }
198 else // even stage
199 {
200 src_array = array1;
201 dst_array = array0;
202 }
203
204 if( (tid & ((1<<i)-1)) == 0 )
205 {
206
207#if DEBUG_SORT
208printf("\n[sort] thread[%d] : stage %d start\n", tid , i );
209#endif
210 merge( src_array,
211 dst_array,
212 items << (i-1),
213 items * tid,
214 items * (tid + (1 << (i-1))),
215 items * tid );
216
217#if DEBUG_SORT
218printf("\n[sort] thread[%d] : stage %d completed\n", tid , i );
219#endif
220 }
221
222 /////////////////////////////////
223 pthread_barrier_wait( &barrier );
224
225#if DEBUG_SORT
226printf("\n[sort] thread[%d] exit barrier %d\n", tid , i );
227#endif
228
229 } // en for stages
230
231 // sort thread signal completion to pthtread_parallel_create()
232 pthread_barrier_wait( parent_barrier );
233
234#if DEBUG_SORT
235printf("\n[sort] thread[%d] exit\n", tid );
236#endif
237
238 // sort thread exit
239 pthread_exit( NULL );
240
241 return NULL;
242
243} // end sort()
244
245
246////////////////
247int main( void )
248{
249 int error;
250 unsigned int x_size; // number of rows
251 unsigned int y_size; // number of columns
252 unsigned int ncores; // number of cores per cluster
253 pthread_barrierattr_t barrier_attr; // barrier attributes (used for DQT)
254 unsigned int n; // index in array to sort
255
256 unsigned long long start_cycle;
257 unsigned long long seq_end_cycle;
258 unsigned long long para_end_cycle;
259
260 /////////////////////////
261 get_cycle( &start_cycle );
262
263 // compute number of working threads (one thread per core)
264 hard_config_t config;
265 get_config( &config );
266 x_size = config.x_size;
267 y_size = config.y_size;
268 ncores = config.ncores;
269 threads = x_size * y_size * ncores;
270
271 // compute covering DQT size an level
272 unsigned int z = (x_size > y_size) ? x_size : y_size;
273 unsigned int root_level = (z == 1) ? 0 : (z == 2) ? 1 : (z == 4) ? 2 : (z == 8) ? 3 : 4;
274
275 // checks number of threads
276 if ( (threads != 1) && (threads != 2) && (threads != 4) &&
277 (threads != 8) && (threads != 16 ) && (threads != 32) &&
278 (threads != 64) && (threads != 128) && (threads != 256) &&
279 (threads != 512) && (threads != 1024) )
280 {
281 printf("\n[sort] ERROR : number of cores must be power of 2\n");
282 exit( 0 );
283 }
284
285 // check array size
286 if ( ARRAY_LENGTH % threads)
287 {
288 printf("\n[sort] ERROR : array size must be multiple of number of threads\n");
289 exit( 0 );
290 }
291
292 printf("\n[sort] main starts / %d threads / %d items / pid %x / cycle %d\n",
293 threads, ARRAY_LENGTH, getpid(), (unsigned int)start_cycle );
294
295 // initialize barrier
296 if( USE_DQT_BARRIER )
297 {
298 barrier_attr.x_size = x_size;
299 barrier_attr.y_size = y_size;
300 barrier_attr.nthreads = ncores;
301 error = pthread_barrier_init( &barrier, &barrier_attr , threads );
302 }
303 else // use SIMPLE_BARRIER
304 {
305 error = pthread_barrier_init( &barrier, NULL , threads );
306 }
307
308 if( error )
309 {
310 printf("\n[sort] ERROR : cannot initialise barrier\n" );
311 exit( 0 );
312 }
313
314#if DEBUG_MAIN
315if( USE_DQT_BARRIER ) printf("\n[sort] main completes DQT barrier init\n");
316else printf("\n[sort] main completes simple barrier init\n");
317#endif
318
319 // Array to sort initialization
320 for ( n = 0 ; n < ARRAY_LENGTH ; n++ )
321 {
322 array0[n] = ARRAY_LENGTH - n - 1;
323 }
324
325#if DISPLAY_ARRAY
326 printf("\n*** array before sort\n");
327 for( n=0; n<ARRAY_LENGTH; n++) printf("array[%d] = %d\n", n , array0[n] );
328#endif
329
330#if DEBUG_MAIN
331printf("\n[sort] main completes array init\n");
332#endif
333
334 ///////////////////////////
335 get_cycle( &seq_end_cycle );
336
337#if DEBUG_MAIN
338printf("\n[sort] main completes sequencial init at cycle %d\n",
339(unsigned int)seq_end_cycle );
340#endif
341
342 // create and execute the working threads
343 if( pthread_parallel_create( root_level,
344 &sort ) )
345 {
346 printf("\n[sort] ERROR : cannot create threads\n");
347 exit( 0 );
348 }
349
350 ////////////////////////////
351 get_cycle( &para_end_cycle );
352
353#if DEBUG_main
354printf("\n[sort] main completes parallel sort at cycle %d\n",
355(unsigned int)para_end_cycle );
356#endif
357
358 // destroy barrier
359 pthread_barrier_destroy( &barrier );
360
361#if DISPLAY_ARRAY
362 printf("\n*** array after merge %d\n", i );
363 for( n=0; n<ARRAY_LENGTH; n++) printf("array[%d] = %d\n", n , dst_array[n] );
364#endif
365
366#if CHECK_RESULT
367 int success = 1;
368 int * res_array = ( (threads == 2) ||
369 (threads == 8) ||
370 (threads == 32) ||
371 (threads == 128) ||
372 (threads == 512) ) ? array1 : array0;
373
374 for( n=0 ; n<(ARRAY_LENGTH-2) ; n++ )
375 {
376 if ( res_array[n] > res_array[n+1] )
377 {
378 printf("\n[sort] array[%d] = %d > array[%d] = %d\n",
379 n , res_array[n] , n+1 , res_array[n+1] );
380 success = 0;
381 break;
382 }
383 }
384
385 if ( success ) printf("\n[sort] success\n");
386 else printf("\n[sort] failure\n");
387#endif
388
389#if INSTRUMENTATION
390 char name[64];
391 char path[128];
392 unsigned long long instru_cycle;
393
394 // build file name
395 if( USE_DQT_BARRIER )
396 snprintf( name , 64 , "p_sort_dqt_%d_%d_%d", ARRAY_LENGTH, x_size * y_size, ncores );
397 else
398 snprintf( name , 64 , "p_sort_smp_%d_%d_%d", ARRAY_LENGTH, x_size * y_size, ncores );
399
400 // build file pathname
401 snprintf( path , 128 , "home/%s" , name );
402
403 // compute results
404 unsigned int sequencial = (unsigned int)(seq_end_cycle - start_cycle);
405 unsigned int parallel = (unsigned int)(para_end_cycle - seq_end_cycle);
406
407 // display results on process terminal
408 printf("\n----- %s -----\n"
409 " - sequencial : %d cycles\n"
410 " - parallel : %d cycles\n",
411 name, sequencial, parallel );
412
413 // open file
414 get_cycle( &instru_cycle );
415 FILE * stream = fopen( path , NULL );
416
417 if( stream == NULL )
418 {
419 printf("\n[sort] ERROR : cannot open instrumentation file <%s>\n", path );
420 exit(0);
421 }
422
423 printf("\n[sort] file <%s> open at cycle %d\n", path, (unsigned int)instru_cycle );
424
425#if IDBG
426idbg();
427#endif
428
429 // register results to file
430 get_cycle( &instru_cycle );
431 int ret = fprintf( stream , "\n----- %s -----\n"
432 " - sequencial : %d cycles\n"
433 " - parallel : %d cycles\n", name, sequencial, parallel );
434 if( ret < 0 )
435 {
436 printf("\n[sort] ERROR : cannot write to instrumentation file <%s>\n", path );
437 exit(0);
438 }
439
440 printf("\n[sort] file <%s> written at cycle %d\n", path, (unsigned int)instru_cycle );
441
442#if IDBG
443idbg();
444#endif
445
446 // close instrumentation file
447 get_cycle( &instru_cycle );
448 ret = fclose( stream );
449
450 if( ret )
451 {
452 printf("\n[sort] ERROR : cannot close instrumentation file <%s>\n", path );
453 exit(0);
454 }
455
456 printf("\n[sort] file <%s> closed at cycle %d\n", path, (unsigned int)instru_cycle );
457
458#endif
459
460 exit( 0 );
461
462 return 0;
463
464} // end main()
465
466/*
467vim: tabstop=4 : shiftwidth=4 : expandtab
468*/
Note: See TracBrowser for help on using the repository browser.