3

我正在尝试并行化合并排序算法。我正在做的是为每个线程划分输入数组，然后合并线程结果。我试图合并结果的方式是这样的：

thread 0                     |   thread 1        |   thread 2         |   thread 3

sort(A0)                     |   sort(A1)        |   sort(A2)         | sort(A3)
merge(A0,A1)                 |                   |   merge(A2,A3)     | 
merge(A0A1, A2A3)            |                   |                    |

所以，在我的函数结束时，sortManager我调用了mergeThreadResults应该实现上述逻辑的函数。在其中我迭代对以合并相应的线程。然后，如果需要，我将最后一个项目合并到线程 0。它看起来像这样：

void mergeThreadResults(long myRank, int myLeft, int myRight, int size, int threads) {

    int nextThread;
    int iter = 2;
    while (iter <= threads) {
        int nextThread = (myRank+1*iter) < threads ? (myRank+1*iter) : threads;
        int nextThreadRight = nextThread * ((float)size / (float)threads) - 1;

        printf("Merging threads %ld to %d\n", myRank, nextThread);
        
        if (myRank % iter != 0) {
            break;
        }

        merge(sortingArray, myLeft, myRight, nextThreadRight);
        sleep(3); // <- sleep

        myRight = nextThreadRight;
        iter = iter * 2;
    }

     if (myRank == 0 && nextThread < threads-1) {
        int nextThreadRight = threads * ((float)size / (float)threads) - 1;
        merge(sortingArray, myLeft, myRight, nextThreadRight);
     }

}

它似乎按预期工作。问题是，我正在使用一个sleep函数来同步线程，这远不是最好的方法。所以我试图用pthread实现一个障碍。
在其中，我尝试计算该循环需要多少次迭代并将其传递为breakpoint. 当所有线程都在同一点时，我释放合并功能并在新的循环中再次等待。这是我尝试过的：

        pthread_mutex_lock(&mutex);
        counter++;
        int breakpoint = threads % 2 == 0 ? threads/iter : threads/iter+1;
        if(counter >= breakpoint ) {
            counter = 0;
            pthread_cond_broadcast(&cond_var);
        } else {
            while (pthread_cond_wait(&cond_var, &mutex) != 0);
        }
        pthread_mutex_unlock(&mutex);

但它没有按预期工作。在最后一个周期完全结束之前的一些merge触发器，给我留下了一个部分排序的数组。

这是我的测试代码的一个小例子：

#define _GNU_SOURCE

#include <stdio.h>
#include <stdlib.h>
#include <limits.h>
#include <string.h>
#include <time.h>

#include <pthread.h>
#include <unistd.h>

// Initialize global variables
int sortingArray[20] = {5,-4,3,-1,-2,3,1,2,-2,-1,-2,-1,-2,-3,4,1234,534,123,87,123};
int counter = 0;
pthread_mutex_t mutex;
pthread_cond_t cond_var;

struct ThreadTask {
    long rank;
    int size;
    int threads;
};

void merge(int arr[], int left, int mid, int right) {
    /* Merge arrays */

    int i, j, k;
    int n1 = mid - left + 1;
    int n2 = right - mid;

    // Alocate temp arrays
    int *L = malloc((n1 + 2) * sizeof(int));
    int *R = malloc((n2 + 2) * sizeof(int));
    if (L == NULL || R == NULL) {
        fprintf(stderr, "Fatal: failed to allocate memory fo temp arrays.");
        exit(EXIT_FAILURE);
    }

    // Populate temp arrays
    for (i = 1; i <= n1; i++) {
        L[i] = arr[left + i - 1];
    }
    for (j = 1; j <= n2; j++) {
        R[j] = arr[mid + j];
    }

    L[n1 + 1] = INT_MAX;
    R[n2 + 1] = INT_MAX;
    i = 1;
    j = 1;

    // Merge arrays
    for (k = left; k <= right; k++) {
        if (L[i] <= R[j]) {
            arr[k] = L[i];
            i++;
        } else {
            arr[k] = R[j];
            j++;
        }
    }

    free(L);
    free(R);
}


void mergeSort(int arr[], int left, int right) {
    /* Sort and then merge arrays */

    if (left < right) {
        int mid = left + (right - left) / 2;

        mergeSort(arr, left, mid);
        mergeSort(arr, mid + 1, right);

        merge(arr, left, mid, right);
    }
}


void mergeThreadResults(long myRank, int myLeft, int myRight, int size, int threads) {

    int nextThread;
    int iter = 2;
    while (iter <= threads) {
        int nextThread = (myRank+1*iter) < threads ? (myRank+1*iter) : threads;
        int nextThreadRight = nextThread * ((float)size / (float)threads) - 1;

        printf("Merging threads %ld to %d\n", myRank, nextThread);
        
        if (myRank % iter != 0) {
            break;
        }

        // barrier
        pthread_mutex_lock(&mutex);
        counter++;
        int breakpoint = threads % 2 == 0 ? threads/iter : threads/iter+1;
        if(counter >= breakpoint ) {
            counter = 0;
            pthread_cond_broadcast(&cond_var);
        } else {
            while (pthread_cond_wait(&cond_var, &mutex) != 0);
        }
        pthread_mutex_unlock(&mutex);

        merge(sortingArray, myLeft, myRight, nextThreadRight);
        sleep(2); // <- sleep

        myRight = nextThreadRight;
        iter = iter * 2;
    }

     if (myRank == 0 && nextThread < threads-1) {
        int nextThreadRight = threads * ((float)size / (float)threads) - 1;
        merge(sortingArray, myLeft, myRight, nextThreadRight);
     }

}

void *sortManager(void *threadInfo) {
    /* Manage mergeSort between threads */

    struct ThreadTask *currentTask = threadInfo;

    // Get task arguments
    long rank = currentTask->rank;
    int left= rank * ((float)currentTask->size / (float)currentTask->threads);
    int right = (rank + 1) * ((float)currentTask->size / (float)currentTask->threads) - 1;
    int mid = left + (right - left) / 2;

    // Execute merge for task division
    if (left < right) {
        mergeSort(sortingArray, left, mid);
        mergeSort(sortingArray, mid + 1, right);
        merge(sortingArray, left, mid, right);
    }

    // Merge thread results
    if (rank % 2 == 0)  {
        mergeThreadResults(rank, left, right, currentTask->size, currentTask->threads);
    }

    return 0;
}


struct ThreadTask *threadCreator(int size, int threads, pthread_t *thread_handles, struct ThreadTask *tasksHolder) {
    /* Create threads with each task info */

    struct ThreadTask *threadTask;

    for (long thread = 0; thread < threads; thread++){
        threadTask = &tasksHolder[thread];
        threadTask->rank = thread;
        threadTask->size = size;
        threadTask->threads = threads;

        pthread_create(&thread_handles[thread], NULL, sortManager, (void*) threadTask);
    }

    return tasksHolder;
}


void printArray(int arr[], int size) {
    /* Print array */

    for (int arrayIndex = 0; arrayIndex < size; arrayIndex++)
        printf("%d ", arr[arrayIndex]);
    printf("\n");
}


int main(int argc, char *argv[]) {

    // Initialize arguments
    int arraySize = 20;
    int totalThreads = 16;

    
    // Display input
    printf("\nInput array:\n");
    printArray(sortingArray, arraySize);
    

    // Initialize threads
    pthread_t *thread_handles;
    thread_handles = malloc(totalThreads * sizeof(pthread_t));

    // Create threads
    struct ThreadTask threadTasksHolder[totalThreads];
    *threadTasksHolder = *threadCreator(arraySize, totalThreads, thread_handles, threadTasksHolder);
    
    // Execute merge sort in each thread
    for (long thread = 0; thread < totalThreads; thread++) {
        pthread_join(thread_handles[thread], NULL);
    }
    free(thread_handles);
    

    // Display output
    printf("\nSorted array:\n");
    printArray(sortingArray, arraySize);
    
    return 0;
}
4

2 回答 2

1

正如@John Bollinger 所说，您的方法看起来非常困难，解决方案也同样复杂。但是如果你想实现一个屏障，我建议你把它放在mergein之后mergeThreadResults。这样，您可以等待在该周期中工作的所有线程完成，然后再移动到下一个。

要创建它，您需要在每次迭代中传递一个新的屏障。因为在每个周期执行合并的线程数都会减少。所以开始声明一些全球性的障碍：

int mergeCycleFlag = 0;
pthread_mutex_t mutex;
pthread_barrier_t *mergeBarrier;

该标志用于为每次迭代创建一个屏障，每个循环我们将需要多个 mergeBarrier。不要忘记在您的main函数中初始化它，您将使用迭代次数：mergeBarrier = realloc(mergeBarrier, howManyIterations);

然后我们可以像这样创建一个屏障：

        pthread_mutex_lock(&mutex);
        if (mergeCycleFlag != iter) { 
            mergeCycleFlag = iter;
            int mergesInLoop = threads % iter== 0 ? threads/iter: threads/iter+1;
            pthread_barrier_init(&mergeBarrier[iter], NULL, mergesInLoop);
        }
        pthread_mutex_unlock(&mutex);

        ... MERGE ...

        // Wait everyone finnish merging
        pthread_barrier_wait (&mergeBarrier[iter]);

请注意，我使用 alock创建屏障，因为我们不希望两个线程同时在这里搞乱。如果没有为此设置障碍，iter我们将创建一个具有现在应该工作的线程数的障碍。另外，我已经更改了您的breakpoint语句以适合计算我们期望执行的线程数merge。

经过一些调整，这就是你的mergeThreadResults样子：

void mergeThreadResults(long myRank, int myLeft, int myRight, int size, int threads) {
    
    int nextThread, nextThreadRight;
    int groupSize = 2;

    while (groupSize <= threads) {
        if (myRank % groupSize != 0) { // Release threads that no long perform merges
            break;
        }

        nextThread = (myRank+1*groupSize) < threads ? (myRank+1*groupSize) : threads;
        nextThreadRight = nextThread * ((float)size / (float)threads) - 1;
 
        printf("Merging threads %ld to %d\n", myRank, nextThread-1);

        // Init barrier with number of threads you will wait merging 
        pthread_mutex_lock(&mutex);  // Just one thread can set the barrier
        if (mergeCycleFlag != groupSize) { 
            mergeCycleFlag = groupSize;
            int mergesInLoop = threads % groupSize == 0 ? threads/groupSize : threads/groupSize+1; // Calculate threads working in this step
            pthread_barrier_init(&mergeBarrier[groupSize], NULL, mergesInLoop);  // set barrier
        }
        pthread_mutex_unlock(&mutex);

        // Merge thread group with neighbour group
        merge(sortingArray, myLeft, myRight, nextThreadRight);

        // Wait everyone finnish merging
        pthread_barrier_wait (&mergeBarrier[groupSize]);

        myRight = nextThreadRight;
        groupSize = groupSize * 2;
    }

    // Merge thread 0
    if (myRank == 0 && nextThread < threads-1) {
        nextThreadRight = threads * ((float)size / (float)threads) - 1;
        merge(sortingArray, myLeft, myRight, nextThreadRight);
    }
}

最后，要使此解决方案起作用，您需要每个线程在合并结果之前都已完成其工作。因此，您需要在joinin之后调用它，或者在调用onmain之前使用所有线程实现另一个屏障。mergeThreadResultssortManager

此外，更好的方法是让线程只等待它们将合并的其他线程。就像，线程 0 只等待 1。然后等待 2……等等。

于 2021-06-15T17:37:20.473 回答
1

我正在尝试并行化合并排序算法。我正在做的是为每个线程划分输入数组，然后合并线程结果。

好的，但你的方法是不必要的困难。在合并过程的每一步，您都希望一半的线程等待另一半完成，而一个线程等待另一个完成的最自然的方法是使用pthread_join(). 如果您希望所有线程在同步后继续进行更多工作，那将是不同的，但在这种情况下，那些不负责任何更多合并的人根本没有什么可做的。

这是我尝试过的：

        pthread_mutex_lock(&mutex);
        counter++;
        int breakpoint = threads % 2 == 0 ? threads/iter : threads/iter+1;
        if(counter >= breakpoint ) {
            counter = 0;
            pthread_cond_broadcast(&cond_var);
        } else {
            while (pthread_cond_wait(&cond_var, &mutex) != 0);
        }
        pthread_mutex_unlock(&mutex);

这样做有几个问题，但最大的问题是障碍是工作的错误工具。在障碍物达到顶峰后，所有被阻止的线程都会继续进行。您希望一半线程继续进行，执行合并，但其他线程（应该）没有更多工作要做。你的计算breakpoint假设下半场不会回到障碍，他们确实不应该这样做。如果你坚持使用屏障，那么没有合并执行的线程应该在通过屏障后终止。

此外，从 2 开始是不正确iter的。如果使用屏障方法，则每次迭代中所有iter活动的线程必须在任何继续之前到达屏障，但如果从 2 开始，那么在第一次迭代中，只有一半的线程必须在通过之前到达障碍物。

此外，您的 CV 使用是非惯用的并且容易出现问题。没有任何记录在案的失败原因pthread_cond_wait()可以通过尝试再次等待来解决，因此您可能需要在出错时终止程序。还要注意pthread_mutex_lock(),pthread_mutex_unlock()和pthread_cond_broadcast()all 也可能失败。

另一方面，CV 容易受到（非常罕见的）虚假唤醒的影响，因此从等待成功返回后，您需要在继续之前再次检查条件，并可能再次等待。更像这样的东西：

        if (pthread_mutex_lock(&mutex) != 0) {
            perror("pthread_mutex_lock");
            abort();
        }
        counter++;
        int breakpoint = threads % 2 == 0 ? threads/iter : threads/iter+1;
        if(counter >= breakpoint ) {
            counter = 0;
            if (pthread_cond_broadcast(&cond_var) != 0) {
                perror("pthread_cond_broadcast");
                abort();
            }
        } else {
            do {
                if (pthread_cond_wait(&cond_var, &mutex) != 0) {
                    perror("pthread_cond_wait");
                    abort();
                }
            } while (counter < breakpoint);
        }
        if (pthread_mutex_unlock(&mutex) != 0) {
            perror("pthread_mutex_unlock");
            abort();
        }

        // some threads must terminate at this point
于 2021-06-15T01:58:45.910 回答