如何使用线程加速排序算法？

Question

我正在做一个任务，目标是通过创建多个线程来加快快速排序过程。然而，我想不出如何加快这个过程。我在开始时应用了允许的线程，但是这似乎只会减慢程序的运行速度？

基本上，目标是通过使用传统的递归快速排序来对简单的数组进行排序。但正如我之前所说的，当我使用clock()库对其性能进行计时时，似乎只会减慢它的速度。有什么建议吗？或者我还需要对这些线程做些什么呢？我将在这里上传我的完整源代码：

#include <pthread.h>
#include <stdio.h>
#include <sys/types.h>
#include <errno.h>
#include <unistd.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <string.h>
#include <stdlib.h>
#include <signal.h>
#include <math.h>
#include <sys/wait.h>
#include <assert.h>
#include <time.h>
static int maxThreads = 4;
#define SORT_THRESHOLD      40
//#includes==========
void *blin();
pthread_mutex_t mutex;
void *send(void *);

static int used = 0; 
static int reached = 0;
int doit = 0;
int threadNo = 0;

typedef struct _sortParams {
    char** array;
    int left;
    int right;
} SortParams;

static void insertSort(char** array, int left, int right) {
    int i, j;
    for (i = left + 1; i <= right; i++) {
        char* pivot = array[i];
        j = i - 1;
        while (j >= left && (strcmp(array[j],pivot) > 0)) {
            array[j + 1] = array[j];
            j--;
        }
        array[j + 1] = pivot;
    }
}





int going = 0;
int blins = 0;



void *send(void * p) {
    SortParams* params = (SortParams*) p;
    char** array = params->array;
    int left = params->left;
    int right = params->right;
    int i = left, j = right;
    
    if (j - i > SORT_THRESHOLD) {           
    /* if the sort range is substantial, use quick sort */

        int m = (i + j) >> 1;               /* pick pivot as median of         */
        char* temp, *pivot;                 /* first, last and middle elements */
        if (strcmp(array[i],array[m]) > 0) {
            temp = array[i]; array[i] = array[m]; array[m] = temp;
        }
        if (strcmp(array[m],array[j]) > 0) {
            temp = array[m]; array[m] = array[j]; array[j] = temp;
            if (strcmp(array[i],array[m]) > 0) {
                temp = array[i]; array[i] = array[m]; array[m] = temp;
            }
        }
        pivot = array[m];

        for (;;) {
            while (strcmp(array[i],pivot) < 0) i++; 
            /* move i down to first element greater than or equal to pivot */
            while (strcmp(array[j],pivot) > 0) j--; 
            /* move j up to first element less than or equal to pivot      */
            if (i < j) {
                char* temp = array[i];      /* if i and j have not passed each other */
                array[i++] = array[j];      /* swap their respective elements and    */
                array[j--] = temp;          /* advance both i and j                  */
            } else if (i == j) {
                i++; j--;
            } else break;                   /* if i > j, this partitioning is done  */
        }
        

        if (blins < 1) {
            blins++;
            SortParams first;  first.array = array; first.left = left; first.right = j;
            int ex;
            pthread_t thred[2];
            pthread_create(&thred[0], NULL, send, &first);
            pthread_join(thred[0], NULL);
            
        SortParams second; second.array = array; second.left = i; second.right = right;
        pthread_create(&thred[1], NULL, send, &second);
            pthread_join(thred[1], NULL); 
        
        } else {
        
        
        SortParams first;  first.array = array; first.left = left; first.right = j;
        send(&first);                  /* sort the left partition  */
        
        SortParams second; second.array = array; second.left = i; second.right = right;
        send(&second);                 /* sort the right partition */        
        
    }   
        
        

                
    } else insertSort(array,i,j);           /* for a small range use insert sort */
}

int main() {

    int count = 100000;
    char * array[count];  
    char * random[10] = {"asdfs", "wesasd", "asded", "aaddsdaa", "dsfs", "av", "bb", 
    "zz", "das", "efdxse"};
    int r = 0;
    for(int ni = 0; ni < count; ni++) {
        r = (rand() % 4);
        char string[100];
        strcpy(string, "");
        int b = (rand() % 50)+1;
        for (int bb = 0; bb < b; bb++) {
            r = (rand() % 4);
            if (r == 0) {
                strcat(string, "a");
            }
            if (r == 1) {
                strcat(string, "b");
            }
            if (r == 2) {
                strcat(string, "c");
            }
            if (r == 3) {
                strcat(string, "d");
            }
            if (r == 4) {
                strcat(string, "e");
            }
        }
        array[ni] = malloc(sizeof(string));
        strcpy(array[ni], string);      
    }
    
    clock_t t;
    
    t = clock();
    SortParams parameters; // declare structure
    parameters.array = array; parameters.left = 0; parameters.right = count - 1;
    //sleep(5);
    send(&parameters);
    
    t = clock() - t;
    
    double total = ((double)t)/CLOCKS_PER_SEC;
    
    printf("%f \n", total);

    char ** jink = parameters.array;
    
    
    for (int ni = 0; ni < count/10; ni++) {
        printf("%s \n", jink[ni]);
    }
    // */
    
    
    for (int ni = 0; ni < count; ni++) {
        free(array[ni]);
    } printf("%f \n", total);

    

    return 0;
}

您应该能够简单地复制/粘贴，它应该可以工作，但如您所见，我已经创建了2个线程，但它比没有线程慢。

Answer 1

让我用这个类比来解释主要问题。

你有一堆纸，每张纸上都有号码。该堆是无序的，您需要对其进行排序。

顺序

你把这堆东西一分为二。对它们进行排序，然后将有序的堆组合成一个大堆。你不能把所有的床单都放在桌子上，因为它很小。所以你一次只能拿到20张纸。当您需要对大于20页的文件进行排序时，您可以暂时将不适合放在桌子上的内容存储在盒子中。

并行

每次你把这堆东西分成两堆的时候，你就会叫快递服务。比方说，它在15分钟内到达，你将两堆邮件发送给住在城市另一边的两个朋友。你的朋友对堆进行排序(如果它们足够大，他们还会拆分它们，并将堆发送给他们的朋友，依此类推)，然后他们会使用快递服务将排序后的堆发送回给您。您需要等待所有的堆从您的朋友到达之前，您可以合并和排序。

你可以看到，如果你只有10张纸要排序(甚至1000张)，单独排序会快得多。协调的所有开销，将数据发送到城镇的另一端是非常大的，即使你的朋友(和他们的朋友)可以并行地做这项工作(即排序表)。

为了获得任何明显的加速，你需要你的堆足够大，以便从引入并行化中获得的收益超过了启动新线程和同步它们的工作的需要(延迟快递服务引入)。

此外，在您当前的实现中，您甚至不能并行工作。启动一个线程，然后立即加入它，这类似于将第一堆发送给你的朋友，然后等待第一个朋友的结果，然后再将第二堆发送给第二个朋友。如果你自己做这件事，对任何大小的文件来说都会快得多。

该怎么办呢？

首先，您需要在一些大型数据上进行测试，以查看收益。

其次，在你的系统中创建更多的线程是没有意义的，因为你有很多的执行器。在创建了这么多线程之后，它们应该只对已有的线程进行排序，而不会产生任何新线程。

你真的需要确保线程并行工作。这是唯一对您有利的地方，也是并行算法加速的唯一来源。所有其他因素都对你不利。

Answer 2

代码中最小的改进是并行地对两个部分进行排序，然后将两个线程连接起来：

    ...
    if (blins < 1) {
        blins++;
        SortParams first;  first.array = array; first.left = left; first.right = j;
        int ex;
        pthread_t thred[2];
        pthread_create(&thred[0], NULL, send, &first);   // start first thread

    SortParams second; second.array = array; second.left = i; second.right = right;
    pthread_create(&thred[1], NULL, send, &second);     // start second thread
    pthread_join(thred[0], NULL);                       // wait for first thread
        pthread_join(thred[1], NULL);                   // wait for second thread

    } else {
    ...

在我的测试中，收益在50%到30%之间。但是在递归处理中使用线程是勇敢的。相反，我会固定工作线程的数量(内核或硬件线程的数量-1?)将数组拆分为该数量的区块，并为每个区块分配一个线程。然后，您只需合并已排序的块