source: trunk/user/sort/sort.c @ 619

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

1) Fix a bug in KSH : after the "load" command,

the [ksh] prompt is now printed after completion
of the loaded application.

2) Fix a bug in vmm_handle_cow() : the copy-on-write

use now a hal_remote_memcpy() to replicate the page content.


  • Property svn:executable set to *
File size: 12.5 KB
Line 
1///////////////////////////////////////////////////////////////////////////////
2// File   :  sort.c
3// Date   :  November 2013
4// Author :  Cesar Fuguet Tortolero <cesar.fuguet-tortolero@lip6.fr>
5///////////////////////////////////////////////////////////////////////////////
6// This multi-threaded application implement a multi-stage sort application.
7// The various stages are separated by synchronisation barriers.
8// There is one thread per physical cores.
9// Computation is organised as a binary tree:
10// - All threads execute in parallel a buble sort on a sub-array during the
11//   the first stage of parallel sort,
12// - The number of participating threads is divided by 2 at each next stage,
13//   to make a merge sort, on two subsets of previous stage.
14//
15//       Number_of_stages = number of barriers = log2(Number_of_threads)
16//
17// Constraints :
18// - It supports up to 1024 cores: x_size, y_size, and ncores must be
19//   power of 2 (max 16*16 clusters / max 4 cores per cluster)
20// _ The array of values to be sorted (ARRAY_LENGTH) must be power of 2
21//   larger than the number of cores.
22///////////////////////////////////////////////////////////////////////////////
23
24#include <stdio.h>
25#include <stdlib.h>
26#include <unistd.h>
27#include <pthread.h>
28#include <almosmkh.h>
29#include <hal_macros.h>
30
31#define ARRAY_LENGTH        256        // number of values
32#define MAX_THREADS         1024       // 16 * 16 * 4
33#define USE_DQT_BARRIER     1
34#define DISPLAY_ARRAY       0
35#define INTERACTIVE_MODE    0
36
37/////////////////////////////////////////////////////////////
38// argument for the sort() function (one thread per core)
39/////////////////////////////////////////////////////////////
40
41typedef struct
42{
43    unsigned int threads;       // total number of threads
44    unsigned int thread_uid;    // thread user index (0 to threads -1)
45    unsigned int main_uid;      // main thread user index
46}
47args_t;
48
49//////////////////////////////////////////
50//      Global variables
51//////////////////////////////////////////
52
53int                 array0[ARRAY_LENGTH];    // values to sort
54int                 array1[ARRAY_LENGTH];   
55
56pthread_barrier_t   barrier;                 // synchronisation variables
57
58pthread_attr_t      attr[MAX_THREADS];       // thread attributes (one per thread)
59args_t              arg[MAX_THREADS];        // sort function arguments (one per thread)
60
61////////////////////////////////////
62static void bubbleSort( int * array,
63                        unsigned int length,
64                        unsigned int init_pos )
65{
66    unsigned int i;
67    unsigned int j;
68    int          aux;
69
70    for(i = 0; i < length; i++)
71    {
72        for(j = init_pos; j < (init_pos + length - i - 1); j++)
73        {
74            if(array[j] > array[j + 1])
75            {
76                aux          = array[j + 1];
77                array[j + 1] = array[j];
78                array[j]     = aux;
79            }
80        }
81    }
82}  // end bubbleSort()
83
84
85///////////////////////////////////
86static void merge( const int * src,
87                   int       * dst,
88                   int         length,
89                   int         init_pos_src_a,
90                   int         init_pos_src_b,
91                   int         init_pos_dst )
92{
93    int i;
94    int j;
95    int k;
96
97    i = 0;
98    j = 0;
99    k = init_pos_dst;
100
101    while((i < length) || (j < length))
102    {
103        if((i < length) && (j < length))
104        {
105            if(src[init_pos_src_a + i] < src[init_pos_src_b + j])
106            {
107                dst[k++] = src[init_pos_src_a + i];
108                i++;
109            }
110            else
111            {
112                dst[k++] = src[init_pos_src_b + j];
113                j++;
114            }
115        }
116        else if(i < length)
117        {
118            dst[k++] = src[init_pos_src_a + i];
119            i++;
120        }
121        else
122        {
123            dst[k++] = src[init_pos_src_b + j];
124            j++;
125        }
126    }
127}  // end merge()
128
129//////////////////////////////////////
130static void sort( const args_t * ptr )
131{
132    unsigned int       i;
133    unsigned long long cycle;
134    unsigned int       cxy;
135    unsigned int       lid;
136
137    int         * src_array  = NULL;
138    int         * dst_array  = NULL;
139
140    // get core coordinates an date
141    get_core( &cxy , &lid );
142    get_cycle( &cycle );
143
144    unsigned int  thread_uid = ptr->thread_uid;
145    unsigned int  threads    = ptr->threads;
146    unsigned int  main_uid   = ptr->main_uid;
147
148    unsigned int  items      = ARRAY_LENGTH / threads;
149    unsigned int  stages     = __builtin_ctz( threads ) + 1;
150
151    printf("\n[SORT] thread[%d] : start\n", thread_uid );
152
153    bubbleSort( array0, items, items * thread_uid );
154
155    printf("\n[SORT] thread[%d] : stage 0 completed\n", thread_uid );
156
157    /////////////////////////////////
158    pthread_barrier_wait( &barrier ); 
159    printf("\n[SORT] thread[%d] exit barrier 0\n", thread_uid );
160
161    // the number of threads contributing to sort
162    // is divided by 2 at each next stage
163    for ( i = 1 ; i < stages ; i++ )
164    {
165        pthread_barrier_wait( &barrier );
166
167        if( (thread_uid & ((1<<i)-1)) == 0 )
168        {
169            printf("\n[SORT] thread[%d] : stage %d start\n", thread_uid , i );
170
171            if((i % 2) == 1)               // odd stage
172            {
173                src_array = array0;
174                dst_array = array1;
175            }
176            else                           // even stage
177            {
178                src_array = array1;
179                dst_array = array0;
180            }
181
182            merge( src_array, 
183                   dst_array,
184                   items << i,
185                   items * thread_uid,
186                   items * (thread_uid + (1 << (i-1))),
187                   items * thread_uid );
188
189            printf("\n[SORT] thread[%d] : stage %d completed\n", thread_uid , i );
190        }
191
192        /////////////////////////////////
193        pthread_barrier_wait( &barrier );
194        printf("\n[SORT] thread[%d] exit barrier %d\n", thread_uid , i );
195
196    }
197
198    // all threads but the main thread exit
199    if( thread_uid != main_uid ) pthread_exit( NULL );
200
201} // end sort()
202
203
204/////////////////
205void main( void )
206{
207    int                    error;
208    unsigned int           x_size;             // number of rows
209    unsigned int           y_size;             // number of columns
210    unsigned int           ncores;             // number of cores per cluster
211    unsigned int           total_threads;      // total number of threads
212    unsigned int           thread_uid;         // user defined thread index
213    unsigned int           main_cxy;           // cluster identifier for main
214    unsigned int           main_x;             // X coordinate for main thread
215    unsigned int           main_y;             // Y coordinate for main thread
216    unsigned int           main_lid;           // core local index for main thread
217    unsigned int           main_uid;           // thread user index for main thread
218    unsigned int           x;                  // X coordinate for a thread
219    unsigned int           y;                  // Y coordinate for a thread
220    unsigned int           lid;                // core local index for a thread
221    unsigned int           n;                  // index in array to sort
222    unsigned long long     cycle;              // current date for log
223    pthread_t              trdid;              // kernel allocated thread index (unused)
224    pthread_barrierattr_t  barrier_attr;       // barrier attributes
225
226    // compute number of threads (one thread per core)
227    get_config( &x_size , &y_size , &ncores );
228    total_threads = x_size * y_size * ncores;
229
230    // get core coordinates and user index for the main thread
231    get_core( &main_cxy , & main_lid );
232    main_x   = HAL_X_FROM_CXY( main_cxy );
233    main_y   = HAL_Y_FROM_CXY( main_cxy );
234    main_uid = (((main_x * y_size) + main_y) * ncores) + main_lid; 
235
236    // checks number of threads
237    if ( (total_threads != 1)   && (total_threads != 2)   && (total_threads != 4)   && 
238         (total_threads != 8)   && (total_threads != 16 ) && (total_threads != 32)  && 
239         (total_threads != 64)  && (total_threads != 128) && (total_threads != 256) && 
240         (total_threads != 512) && (total_threads != 1024) )
241    {
242        printf("\n[SORT ERROR] number of cores must be power of 2\n");
243        exit( 0 );
244    }
245
246    // check array size
247    if ( ARRAY_LENGTH % total_threads) 
248    {
249        printf("\n[SORT ERROR] array size must be multiple of number of threads\n");
250        exit( 0 );
251    }
252
253    printf("\n\n[SORT] main starts on core[%x,%d] / %d threads / %d values / PID %x\n",
254    main_cxy, main_lid, total_threads, ARRAY_LENGTH, getpid() );
255
256    // initialize barrier
257    if( USE_DQT_BARRIER )
258    {
259        barrier_attr.x_size   = x_size; 
260        barrier_attr.y_size   = y_size;
261        barrier_attr.nthreads = ncores;
262        error = pthread_barrier_init( &barrier, &barrier_attr , total_threads );
263    }
264    else // use SIMPLE_BARRIER
265    {
266        error = pthread_barrier_init( &barrier, NULL , total_threads );
267    }
268
269    if( error )
270    {
271        printf("\n[SORT ERROR] cannot initialise barrier\n" );
272        exit( 0 );
273    }
274
275    printf("\n[SORT] main completes barrier init\n");
276
277    // Array to sort initialization
278    for ( n = 0 ; n < ARRAY_LENGTH ; n++ )
279    {
280        array0[n] = ARRAY_LENGTH - n - 1;
281    }
282
283#if DISPLAY_ARRAY
284printf("\n*** array before sort\n");
285for( n=0; n<ARRAY_LENGTH; n++) printf("array[%d] = %d\n", n , array0[n] );
286#endif
287
288    printf("\n[SORT] main completes array init\n");
289
290    // launch other threads to execute sort() function
291    // on cores other than the core running the main thread
292    for ( x=0 ; x<x_size ; x++ )
293    {
294        for ( y=0 ; y<y_size ; y++ )
295        {
296            for ( lid=0 ; lid<ncores ; lid++ )
297            {
298                thread_uid = (((x * y_size) + y) * ncores) + lid;
299
300                // set sort arguments for all threads
301                arg[thread_uid].threads      = total_threads;
302                arg[thread_uid].thread_uid   = thread_uid;
303                arg[thread_uid].main_uid     = main_uid;
304
305                // set thread attributes for all threads
306                attr[thread_uid].attributes = PT_ATTR_DETACH          |
307                                              PT_ATTR_CLUSTER_DEFINED | 
308                                              PT_ATTR_CORE_DEFINED;
309                attr[thread_uid].cxy        = HAL_CXY_FROM_XY( x , y );
310                attr[thread_uid].lid        = lid;
311
312                if( thread_uid != main_uid )
313                {
314                    if ( pthread_create( &trdid,              // not used because no join
315                                         &attr[thread_uid],   // thread attributes
316                                         &sort,               // entry function
317                                         &arg[thread_uid] ) ) // sort arguments
318                    {
319                        printf("\n[SORT ERROR] main cannot create thread %x \n", thread_uid );
320                        exit( 0 );
321                    }
322                    else
323                    {
324                        printf("\n[SORT] main created thread %x \n", thread_uid );
325                    }
326                }
327            }
328        }
329    }
330   
331    get_cycle( &cycle );
332    printf("\n[SORT] main completes threads create at cycle %d\n", (unsigned int)cycle );
333
334#if INTERACTIVE_MODE
335idbg();
336#endif
337   
338    // the main thread run also the sort() function
339    sort( &arg[main_uid] );
340
341#if INTERACTIVE_MODE
342idbg();
343#endif
344
345    // destroy barrier
346    pthread_barrier_destroy( &barrier );
347
348#if INTERACTIVE_MODE
349idbg();
350#endif
351
352    // Check result
353    int    success = 1;
354    int*   res_array = ( (total_threads ==   2) ||
355                         (total_threads ==   8) || 
356                         (total_threads ==  32) || 
357                         (total_threads == 128) || 
358                         (total_threads == 512) ) ? array1 : array0;
359   
360    for( n=0 ; n<(ARRAY_LENGTH-2) ; n++ )
361    {
362        if ( res_array[n] > res_array[n+1] )
363        {
364            printf("\n[SORT] array[%d] = %d > array[%d] = %d\n",
365            n , res_array[n] , n+1 , res_array[n+1] );
366            success = 0;
367            break;
368        }
369    }
370
371#if DISPLAY_ARRAY
372printf("\n*** array after sort\n");
373for( n=0; n<ARRAY_LENGTH; n++) printf("array[%d] = %d\n", n , res_array[n] );
374#endif
375
376    get_cycle( &cycle );
377
378    if ( success )
379    {
380        printf("\n[SORT] success at cycle %d\n", (unsigned int)cycle );
381    }
382    else
383    {
384        printf("\n[SORT] failure at cycle %d\n", (unsigned int)cycle );
385    }
386
387#if INTERACTIVE_MODE
388idbg();
389#endif
390
391    exit( 0 );
392
393}  // end main()
394
395
396/*
397vim: tabstop=4 : shiftwidth=4 : expandtab
398*/
Note: See TracBrowser for help on using the repository browser.