Parallel Sorting With Multithread

First we must discuss what is thread and why we use it. It means small works and can work multiple with other threads and make paralel working.

The important point when using thread is decide to when it should stop. You can use detach or join for it stop.Detached thread has continued to run in the backround. Join is a thread waiting to be completed before starting another job.

In this project you are asked to implement a parallel program using multi-threading: You are given an array A of N unsorted integers , and you are to sort these numbers in parallel.


In this project you required to solve this problem with 8 threads. Each thread would be responsible for sorting exactly integers of the array as shown in following figure. For example, the first thread T1 would sort the integers in the second thread T2 would sort the integers in and so on. You would then have 8 sorted sub-arrays and you compute the overall sorted array.

Main Thread:

(1) Check command line arguments for validity. Exit in error.
(2) Allocate space for N integers
(3) Read the integers from the file and close the file
(4) Create 8 worker threads to sort the 8 sub-arrays using qsort.
(5) Start timer (use gettimeofday)
(6) Wait for the termination of the worker threads
(7) Stop the timer (use gettimeofday)
(8) Write the sorted integers to file “sorted.dat”.
(9) Print the execution time, and free up the allocated spaces

Parallel Sorting

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <time.h>
#include <sys/time.h>
#define NUM_THREADS 8


void *worker_thread1(int *);
void *worker_thread2(int *);
void *worker_thread3(int *);
void *worker_thread4(int *);
void *worker_thread5(int *);
void *worker_thread6(int *);
void *worker_thread7(int *);
void *worker_thread8(int *);
void merge(int , int , int[] , int[], int);
int *A; // N büyüklüğünde bir char array
int base;
int *C;//big array
int k;
int i1=0, i2=0, i3=0, i4=0, i5=0, i6=0, i7=0, i8=0;
pthread_t thread1, thread2, thread3, thread4, thread5, thread6, thread7, thread8;
int compare (const void * , const void * );
int compare (const void * a, const void * b)// compare method to sort of an array
{
  return ( *(int*)a - *(int*)b );
}
void* worker_thread1(int * p)// p:start  q= end of array
 {  //printf("**%d\n",*p);
   
int i=0;  
  qsort (&A[0], base/8, sizeof(int), compare); // make qsort the array
  
              //thread2 nin bitmesini bekle
                while(i2==0);
  pthread_join(thread2, NULL);

  merge(base/8,base/8,&A[0],&A[base/8],0);
  
  int g;
  for(g=0;g<2*base/8;g++)
  {
   A[g]=C[g]; // A da T1 ve T2 var
  }
   
              //thread3 nin bitmesini bekle
                while(i3==0);
  pthread_join(thread3, NULL);
  merge(base/4,base/4,&A[0],&A[(2*base)/8],0);
  int y;
  for(y=0;y<base/2;y++)
  {
   
   A[y]=C[y]; // Artk A da T1 T2 T3 T4 var
  }
  
               //thread5 nin bitmesini bekle
                while(i5==0);
  pthread_join(thread5, NULL);
  
  merge(base/2,base/2,&A[0],&A[(4*base)/8],0);// Artık Cde hepsi sıralı
  
  
  
                i1=1;
  return 0; 
}
void* worker_thread2(int * p)// p:start  q= end of array
 {  
    
  qsort (&A[base/8], base/8, sizeof(int), compare); // make qsort the array
                i2=1;
  return 0;
}
void* worker_thread3(int * p)// p:start  q= end of array
 {  
  int i=(2*base)/8;  
  qsort (p, base/8, sizeof(int), compare); // make qsort the array

         //thread4 nin bitmesini bekle
                while(i4==0);
  pthread_join(thread4, NULL);
  merge(base/8,base/8,&A[(2*base)/8],&A[(3*base)/8],i);
  int f;
  for(f=2*base/8;f<4*base/8;f++)
  {
   A[f]=C[f];// artk a da T3 ve T4 sıralı array var
   
  }
  i3=1;
  return 0;
  
}
void* worker_thread4(int * p)// p:start  q= end of array
 {  
    
  qsort (&A[(3*base)/8], base/8, sizeof(int), compare); // make qsort the array
                i4=1;
  return 0;
} 
void* worker_thread5(int * p)// p:start  q= end of array
 { 
  int i=(4*base)/8;  
  qsort (p, base/8, sizeof(int), compare); // make qsort the array

           //thread6 nin bitmesini bekle
                while(i6==0);
  pthread_join(thread6, NULL);
  merge(base/8,base/8,&A[(4*base)/8],&A[(5*base)/8],i);
  int h;
  for(h=(4*base)/8;h<(6*base)/8;h++)
  {
   A[h]=C[h]; // A da T5 ve T6 var
  }
  
  
               //thread7 nin bitmesini bekle
                while(i7==0);
  pthread_join(thread7, NULL);
  merge(base/4,base/4,&A[(4*base)/8],&A[(6*base)/8],(4*base)/8);
  
  for(h=4*base/8;h<base;h++)
  {
   A[h]=C[h]; // A da T5 T6 T7 T8 var
  }
  i5=1;
  return 0;
  
}
void* worker_thread6(int * p)// p:start  q= end of array
 {
    
  qsort (&A[(5*base)/8], base/8, sizeof(int), compare); // make qsort the array
                i6=1;
  return 0;
}
void* worker_thread7(int * p)// p:start  q= end of array
 {  
  int i=(6*base)/8;  
  qsort (&A[(6*base)/8], base/8, sizeof(int), compare); // make qsort the array


                //thread8 nin bitmesini bekle
                while(i8==0);
  pthread_join(thread8, NULL);
  merge(base/8,base/8,&A[(6*base)/8],&A[(7*base)/8],i);
  int h;
  for(h=(6*base)/8;h<base;h++) // burada 7 ve 8 i birlestircez
  {
   A[h]=C[h]; // burada T7 ve T8 var
  }
                i7=1;
  return 0;
  
}
void* worker_thread8(int * p)// p:start  q= end of array
 {  
    
  qsort (&A[(7*base)/8], base/8, sizeof(int), compare); // make qsort the array
                i8=1;
  return 0;
}   

// main method

int main(int argc, char *argv[])
{   
 struct timeval start_time,end_time;
   base = (argc > 2) ? atoi(argv[1]) : -1; // convert to string to 
 if(base == -1){
 printf("Yanlis deger girdiniz!!");
 exit(0);}

 
  A = (int *)malloc(sizeof(int)*base); // allocate space for a 100 character string
  C = (int *)malloc(sizeof(int)*base); 
     if (A == NULL) {fputs ("Memory error",stderr); exit (2);}

 FILE *fp = fopen(argv[2], "rb");
 fread(A, sizeof(int), base, fp);
 
 fclose(fp);
 

 int r1=pthread_create(&thread1, NULL,(void *) &worker_thread1,(int *)&A[0]);
 int r2=pthread_create(&thread2, NULL,(void *) &worker_thread2,(int *)&A[base/8]);
 int r3=pthread_create(&thread3, NULL,(void *) &worker_thread3,(int *)&A[(2*base)/8]);
 int r4=pthread_create(&thread4, NULL,(void *) &worker_thread4,(int *)&A[(3*base)/8]);
 int r5=pthread_create(&thread5, NULL,(void *) &worker_thread5,(int *)&A[(4*base)/8]);
 int r6=pthread_create(&thread6, NULL,(void *) &worker_thread6,(int *)&A[(5*base)/8]);
 int r7=pthread_create(&thread7, NULL,(void *) &worker_thread7,(int *)&A[(6*base)/8]);
 int r8=pthread_create(&thread8, NULL,(void *) &worker_thread8,(int *)&A[(7*base)/8]);
 
 int x;
        gettimeofday(&start_time,NULL); // baslangıc zamanı
 
 //thread1 nin bitmesini bekle  
        while(i1==0);
 pthread_join( thread1, NULL);

 gettimeofday(&end_time,NULL); //bitis zamanı

 FILE *fx = fopen("sorted.dat", "wb"); //  sıralı diziyi dosyaya yazdır
 fwrite(&C[0], sizeof(int), base, fx);
 fclose(fx);

 //print execution time
 printf ("Execution time: %ld microseconds \n",((end_time.tv_sec *1000000 + end_time.tv_usec)-(start_time.tv_sec * 1000000 + start_time.tv_usec)));

 

 free(A);    // arraylari serbest bırak
 free(C);
return 0;
 

}
void merge(int m, int n, int A[], int B[],int ptr) {  // sıralı A ve B arrayini merge yapıp global C arrayine at

      int i, j;
      

      i = 0;

      j = 0;

   

      while (i < m && j < n) {

            if (A[i] <= B[j]) {

                  C[ptr] = A[i];
              

                  i++;
    ptr++;

            } else {

                  C[ptr] = B[j];
 
                  j++;
    ptr++;

            }

            

      }

      while(i<m && j>=n){
                  C[ptr] = A[i];
                  i++;
                  ptr++;
                  
      } 


      while(j<n && i>=m){
                  C[ptr] = B[j];
                  j++;
                  ptr++;
                 
      }
  
 


}