c - 多线程合并排序无法正确对数组进行排序

Question

以下代码应该使用多个线程对数组进行合并排序。这个想法是将数组分割成近似相同大小的子数组，并使用线程同时对它们进行排序。然后，在它们全部排序后(在 pthread_join 的 for 循环之后)，将它们合并在一起。但是，我的代码没有正确对数组进行排序。

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/time.h>
#include <pthread.h>
#include <unistd.h>
#include <assert.h>

typedef struct Arguments
{
    long *nums;
    int start;
    int end;
    long *result;
} SortThreadArgs;

/** The number of threads to be used for sorting. Default: 1 */
int thread_count = 2;
pthread_mutex_t merge_lock;

/**
 * Print the given array of longs, an element per line.
 */
void print_long_array(const long *array, int count)
{
    for (int i = 0; i < count; ++i)
    {
        printf("%ld\n", array[i]);
    }
}

/**
 * Merge two slices of nums into the corresponding portion of target.
 */
void merge(long nums[], int from, int mid, int to, long target[])
{
    int left = from;
    int right = mid;

    // acquire lock before accessing shared data
    pthread_mutex_lock(&merge_lock);

    int i = from;
    for (; i < to && left < mid && right < to; i++)
    {
        if (nums[left] <= nums[right])
        {
            target[i] = nums[left];
            left++;
        }
        else
        {
            target[i] = nums[right];
            right++;
        }
    }
    if (left < mid)
    {
        memmove(&target[i], &nums[left], (mid - left) * sizeof(long));
    }
    else if (right < to)
    {
        memmove(&target[i], &nums[right], (to - right) * sizeof(long));
    }

    // release lock after modifying shared data
    pthread_mutex_unlock(&merge_lock);
}

/**
 * Sort the given slice of nums into target.
 *
 * Warning: nums gets overwritten.
 */
void merge_sort_aux(long nums[], int from, int to, long target[])
{
    if (to - from <= 1)
    {
        return;
    }

    int mid = (from + to) / 2;
    merge_sort_aux(target, from, mid, nums);
    merge_sort_aux(target, mid, to, nums);
    merge(nums, from, mid, to, target);
}

void *sort_thread(void *arg)
{
    SortThreadArgs *args = (SortThreadArgs *)arg;
    long *nums = args->nums;
    int start = args->start;
    int end = args->end;
    long *result = args->result;

    merge_sort_aux(nums, start, end, result);

    pthread_exit(NULL);
}

/**
 * Sort the given array and return the sorted version.
 *
 * The result is malloc'd so it is the caller's responsibility to free it.
 *
 * Warning: The source array gets overwritten.
 */
long *merge_sort(long nums[], int count)
{
    // Check if thread_count is divisible by count
    int mod = count % thread_count;

    // Create a temp array that has the same elements as the given array
    long *result = calloc(count, sizeof(long));
    assert(result != NULL);
    memmove(result, nums, count * sizeof(long));

    pthread_mutex_init(&merge_lock, NULL);

    pthread_t threads[thread_count];
    for (int i = 0; i < thread_count; i++)
    {
        // Calculate the start and end index based on the thread_num
        // i.e. Splitting the array into approximately same-size sub arrays
        int start = i * (count / thread_count);
        int end;
        if (i == thread_count - 1 && mod != 0) {
            end = count;
        } else {
            end = (i + 1) * (count / thread_count);
        }
        SortThreadArgs args = {nums, start, end, result};
        pthread_create(&threads[i], NULL, sort_thread, &args);
    }

    for (int i = 0; i < thread_count; i++)
    {
        pthread_join(threads[i], NULL);
    }

    // perform final merge operation on all subarrays
    for (int i = 0; i < thread_count; i++)
    {
        int start = i * (count / thread_count);
        int mid = start;
        int end;
        if (i == thread_count - 1 && mod != 0) {
            end = count;
        } else {
            end = (i + 1) * (count / thread_count);
        }
        merge(nums, 0, mid, end, result);
    }

    pthread_mutex_destroy(&merge_lock);

    return result;
}

int main() {
    int count = 10;
    long array[10] = { 4, 8, 3, 10, 6, 7, 5, 1, 9, 2 };
    long *result = merge_sort(array, count);
    print_long_array(result, count);
}

我将数组拆分为子数组的方法是在 merge_sort 函数的第一个循环中。根据计数器 i，我计算 start 和 end 索引。例如，如果 thread_count = 2 且 count = 10，则 i = 0、start = 0、end = 5；对于 i = 1，开始 = 5，结束 = 10。

在数组{4, 8, 3, 10, 6, 7, 5, 1, 9, 2}中，打印输出为:

1
2
2
4
5
8
3
9
10
6

这真的很奇怪，因为有重复的 2。假设我的想法应该可以正常工作，但我真的不知道导致此问题的潜在错误是什么。如有任何帮助，我们将不胜感激!

当我尝试在 pthread_join 之后打印 for 循环中的每个子数组时，它显示子数组甚至没有排序。

---

更新

以下代码现在可以正确用于单线程合并排序:

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/time.h>
#include <pthread.h>
#include <unistd.h>

#include <assert.h>

/** The number of threads to be used for sorting. Default: 1 */
int thread_count = 1;

typedef struct Arguments
{
    long *nums;
    int from;
    int to;
    long *target;
} merge_sort_args;

/**
 * Print the given array of longs, an element per line.
 */
void print_long_array(const long *array, int count)
{
    for (int i = 0; i < count; ++i)
    {
        printf("%ld\n", array[i]);
    }
}

/**
 * Merge two slices of nums into the corresponding portion of target.
 */
void merge(long nums[], int from, int mid, int to, long target[])
{
    int left = from;
    int right = mid;

    int i = from;
    for (; i < to && left < mid && right < to; i++)
    {
        if (nums[left] <= nums[right])
        {
            target[i] = nums[left];
            left++;
        }
        else
        {
            target[i] = nums[right];
            right++;
        }
    }
    if (left < mid)
    {
        memcpy(&target[i], &nums[left], (mid - left) * sizeof(long));
    }
    else if (right < to)
    {
        memcpy(&target[i], &nums[right], (to - right) * sizeof(long));
    }
}

/**
 * Sort the given slice of nums into target.
 *
 * Warning: nums gets overwritten.
 */
void merge_sort_aux(long nums[], int from, int to, long target[])
{
    if (to - from <= 1)
    {
        return;
    }

    int mid = (from + to) / 2;
    merge_sort_aux(target, from, mid, nums);
    merge_sort_aux(target, mid, to, nums);
    merge(nums, from, mid, to, target);
}

void *sort_thread(void *arg) {
    merge_sort_args *msa = (merge_sort_args *)arg;

    merge_sort_aux(msa->nums, msa->from, msa->to, msa->target);
    pthread_exit(NULL);
}

/**
 * Sort the given array and return the sorted version.
 *
 * The result is malloc'd so it is the caller's responsibility to free it.
 *
 * Warning: The source array gets overwritten.
 */
long *merge_sort(long nums[], int count)
{
    long *result = calloc(count, sizeof(long));
    assert(result != NULL);

    memcpy(result, nums, count * sizeof(long));

    pthread_t threads[thread_count];
    merge_sort_args args[thread_count];
    // Split the array into sub arrays here
    for (int i = 0; i < thread_count; i++)
    {
        int start = i * (count / thread_count);
        int end = (i + 1) * (count / thread_count);
        args[i] = (merge_sort_args) {nums, start, end, result};
        pthread_create(&threads[i], NULL, sort_thread, &args[i]);
    }

    for (int i = 0; i < thread_count; i++)
    {
        pthread_join(threads[i], NULL);
    }

    for (int i = 1; i < thread_count; i++)
    {
        int start = i * (count / thread_count);
        int mid = start;
        int end = (i + 1) * (count / thread_count);
        merge(nums, 0, mid, end, result);
    }

    return result;
}

int main()
{
    long array[10] = { 10, 2, 5, 8, 1, 4, 7, 9, 3, 6 };
    int count = 10;
    long *result = merge_sort(array, count);
    print_long_array(result, count);
}

Answer 1

只有当数组长度是线程数的倍数时，计算切片边界的方法才有效。你应该使用这个:

long *merge_sort(long nums[], int count)
{
    long *result = calloc(count, sizeof(long));
    assert(result != NULL);

    memcpy(result, nums, count * sizeof(long));

    pthread_t threads[thread_count];
    merge_sort_args args[thread_count];
    // Split the array into sub arrays here
    for (long long i = 0; i < thread_count; i++) {
        int start = (i * count) / thread_count;
        int end = ((i + 1) * count) / thread_count;
        args[i] = (merge_sort_args) { nums, start, end, result };
        pthread_create(&threads[i], NULL, sort_thread, &args[i]);
    }

    for (int i = 0; i < thread_count; i++) {
        pthread_join(threads[i], NULL);
    }

    for (int i = 1; i < thread_count; i++) {
        int mid = args[i].from;
        int end = args[i].to;
        merge(nums, 0, mid, end, result);
    }

    return result;
}

还有另一个主要问题:您没有在 mergesort_aux 和 merge 函数中正确处理源数组和目标数组。由于无论如何都会修改源数组，因此您应该使用最后一个参数作为临时工作空间进行排序和合并。

这是该程序的修改版本:

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/time.h>
#include <pthread.h>
#include <unistd.h>
#include <assert.h>

/** The number of threads to be used for sorting. Default: 1 */
int thread_count = 3;

typedef struct Arguments {
    long *nums;
    int from;
    int to;
    long *temp;
} merge_sort_args;

/**
 * Print the given array of longs, an element per line.
 */
void print_long_array(const long *array, int count)
{
    for (int i = 0; i < count; ++i) {
        printf("%ld\n", array[i]);
    }
}

/**
 * Merge two slices of nums using the temp work space
 */
void merge(long nums[], int from, int mid, int to, long temp[])
{
    int i = from;
    int left = from;
    int right = mid;

    memcpy(&temp[from], &nums[from], sizeof(*nums) * (to - from));

    for (i = left = from, right = mid; left < mid && right < to; i++) {
        if (temp[left] <= temp[right]) {
            nums[i] = temp[left++];
        } else {
            nums[i] = temp[right++];
        }
    }
    if (left < mid) {
        memcpy(&nums[i], &temp[left], (mid - left) * sizeof(*nums));
    } else {
        memcpy(&nums[i], &temp[right], (to - right) * sizeof(*nums));
    }
}

/**
 * Sort the given slice of nums.
 */
void merge_sort_aux(long nums[], int from, int to, long temp[])
{
    if (to - from > 1) {
        int mid = from + (to - from) / 2;
        merge_sort_aux(nums, from, mid, nums);
        merge_sort_aux(nums, mid, to, temp);
        merge(nums, from, mid, to, temp);
    }
}

void *sort_thread(void *arg) {
    merge_sort_args *msa = (merge_sort_args *)arg;

    merge_sort_aux(msa->nums, msa->from, msa->to, msa->temp);
    pthread_exit(NULL);
}

/**
 * Sort the given array in place.
 */
void merge_sort(long nums[], int count)
{
    long *temp = calloc(count, sizeof(long));
    assert(temp != NULL);

    pthread_t threads[thread_count];
    merge_sort_args args[thread_count];
    // Split the array into sub arrays here
    for (long long i = 0; i < thread_count; i++) {
        int start = (i * count) / thread_count;
        int end = ((i + 1) * count) / thread_count;
        args[i] = (merge_sort_args) { nums, start, end, temp };
        pthread_create(&threads[i], NULL, sort_thread, &args[i]);
    }

    for (int i = 0; i < thread_count; i++) {
        pthread_join(threads[i], NULL);
    }

    for (int i = 1; i < thread_count; i++) {
        int mid = args[i].from;
        int end = args[i].to;
        merge(nums, 0, mid, end, temp);
    }
    free(temp);
}

int main(void)
{
    long array[10] = { 10, 2, 5, 8, 1, 4, 7, 9, 3, 6 };
    int count = sizeof(array) / sizeof(array[0]);
    merge_sort(array, count);
    print_long_array(array, count);
}