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

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？）将数组拆分为该数量的块，并且每个块有一个线程。然后你只合并排序的块