线程在 C pthreads 中不能同时工作

Question

所以我有一个快速排序算法，我想在两个不同的线程中运行，这个想法是让数组的 2 个独立的一半被一次排序（这不应该是分区的快速排序性质的问题）。

我的代码如下：

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>

void troca (int *v, int i, int j);
int partition (int *v, int ini, int fim);
void quicksort (int *v, int ini, int fim, int size);

typedef struct {
    int ini, mid, fim;
} thread_arg;

int size;
int *v;

void *thread_func(void* arg){
    thread_arg *a = arg;
    quicksort(v, a->ini, a->mid, a->fim);
    printf("done in\n");
    return NULL;
}

int main()
{
    // initializations
    scanf("%d", &size);
    v = malloc(size * sizeof(int));

    // read array
    for (int i = 0; i < size; ++i)
        scanf("%d", &v[i]);

    // arguments
    thread_arg argument1, argument2;
    int mid = partition(v, 0, size-1);

    argument1.ini = 0;
    argument1.mid = mid-1;
    argument1.fim = size;

    argument2.ini = mid;
    argument2.mid = size-1;
    argument2.fim = size;

    pthread_t thread1, thread2;    

    // thread and execution
    pthread_create(&thread1, NULL, thread_func, &argument1);
    printf("done out 1\n");
    pthread_create(&thread2, NULL, thread_func, &argument2);
    printf("done out 2\n");

    pthread_join(thread1, NULL);
    pthread_join(thread2, NULL);

    free(v);

    return 0;
}

void quicksort (int *v, int ini, int fim, int size){
    if (ini >= fim)
        return;
    int meio = partition(v, ini, fim);
    quicksort(v, ini, meio-1, size);
    quicksort(v, meio+1, fim, size);
}

int partition (int *v, int ini, int fim){
    int i = ini-1, j = ini;
    troca(v, (ini+fim)/2, fim);
    int pivo = v[fim];

    for (; j < fim; ++j)
    {
        if (v[j] <= pivo)
        {
            i++;
            troca(v, i, j);
        }
    }
    troca(v, i+1, fim);
    return i+1; //indice do pivo;
}


void troca (int *v, int i, int j){
    int aux = v[i];
    v[i] = v[j];
    v[j] = aux;
    return;
}

执行工作并完美排序，它确实生成了 2 个新的独立线程，每个线程对数组的一半进行排序。问题是，它不会同时进行。输入 100m 个随机数运行程序：

done out 1
done out 2
done in
done in

real    0m47,464s
user    0m50,686s
sys     0m0,452s

但是前 3 行出现大约需要 25 秒，最后一行大约需要 25 秒，这表明第二个线程正在等待第一个运行。

在htop 控制台中，似乎在某些时候两者同时运行（这是因为该程序运行速度比我的正常程序快一点）

最后，我知道同时处理这类数据是不安全的，但在这个排序示例中应该没问题。

Answer 1

您没有在线程之间公平地分配工作。要看到这一点，请按如下方式修改您的代码：

    // thread and execution
    pthread_create(&thread1, NULL, thread_func, &argument1);
    printf("done out 1 (%d)\n", argument1.mid - argument1.ini + 1);
    pthread_create(&thread2, NULL, thread_func, &argument2);
    printf("done out 2 (%d)\n", argument2.mid - argument2.ini + 1);

您会看到一个线程的工作量往往是另一个线程的两倍。

例如，这里有一些使用随机数据的运行：

完成 1 (66474145)
完成 2 (33525855)

完成 1 (21794872)
完成 2 (78205128)

完成 1 (67867800)
完成 2 (32132200)

如果您关心并发性，则永远不应该将您的工作分成少量非常大的块并将每个块分配给一个线程。相反，创建一个小任务队列，并让线程在完成时从队列中拉取任务。

Answer 2

线程正在同时运行（嗯，不一定是同时运行的，但可以感知是同时运行的）。您看到的 25 秒延迟是由于您的快速排序中的错误（或者可能是您在两个线程之间共享列表的方式）。从本质上讲，线程 2 比线程 1 分配了更多的工作，因此它需要更长的时间才能完成。线程 2 不是简单地“在”线程 1 之后执行，或者“等待”线程 1。

为了证明这一点，我在quicksort 中添加了一个unsigned long* 参数，并让它在每次调用时递增所述指针引用的值（基本上是计算每个线程调用quicksort 的次数），然后我运行它具有 10M（不是 100M）随机值。线程 1 的计数最终为 3,851,991，线程 2 的计数最终为 9,693,697。当然，由于列表生成的随机性，两个计数之间可能会有一些微小的变化。但差异几乎是 3 倍，这比您从轻微的随机变化所预期的要重要得多。

我建议您尝试quicksort 的不同实现（已知可以工作）。我还建议对数据共享更加小心（确保两个线程永远不会访问彼此的数据，尤其是在没有同步的情况下）以获得更准确的时间测量；您最不想做的事情是让一个线程对另一个线程的数据进行排序或取消排序。

Answer 3

线程创建不公平：线程#1 在线程#2 之前创建。此外，当线程#1 被创建时，它可能会运行并抢占主线程，主线程可能会等待它返回 CPU 以创建和启动线程#2。但是，在默认 SCHED_OTHER 策略下运行的线程具有不可预测的行为。

添加一些可预测性：

• 使线程在创建后同时启动。使用屏障同时触发所有线程的“执行”。参照。 pthread_barrier_init()

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>

void troca (int *v, int i, int j);
int partition (int *v, int ini, int fim);
void quicksort (int *v, int ini, int fim, int size);



pthread_barrier_t barrier;

typedef struct {
    int id;
    int ini, mid, fim;
} thread_arg;

int size;
int *v;

void *thread_func(void* arg){
    thread_arg *a = arg;
    pthread_barrier_wait(&barrier);
    quicksort(v, a->ini, a->mid, a->fim);
    printf("done in %d\n", a->id);
    return NULL;
}

int main()
{
    // initializations
    scanf("%d", &size);
    v = malloc(size * sizeof(int));

    // read array
    for (int i = 0; i < size; ++i)
        scanf("%d", &v[i]);

    // arguments
    thread_arg argument1, argument2;
    int mid = partition(v, 0, size-1);

    argument1.id = 1;
    argument1.ini = 0;
    argument1.mid = mid-1;
    argument1.fim = size;

    argument2.id = 2;
    argument2.ini = mid;
    argument2.mid = size-1;
    argument2.fim = size;

    pthread_t thread1, thread2;    

    pthread_barrier_init(&barrier, NULL, 3);

    // thread and execution
    pthread_create(&thread1, NULL, thread_func, &argument1);
    printf("done out 1\n");
    pthread_create(&thread2, NULL, thread_func, &argument2);
    printf("done out 2\n");

    // Start the threads
    pthread_barrier_wait(&barrier);

    pthread_join(thread1, NULL);
    pthread_join(thread2, NULL);

    free(v);
    pthread_barrier_destroy(&barrier);

    return 0;
}

void quicksort (int *v, int ini, int fim, int size){
    if (ini >= fim)
        return;
    int meio = partition(v, ini, fim);
    quicksort(v, ini, meio-1, size);
    quicksort(v, meio+1, fim, size);
}

int partition (int *v, int ini, int fim){
    int i = ini-1, j = ini;
    troca(v, (ini+fim)/2, fim);
    int pivo = v[fim];

    for (; j < fim; ++j)
    {
        if (v[j] <= pivo)
        {
            i++;
            troca(v, i, j);
        }
    }
    troca(v, i+1, fim);
    return i+1; //indice do pivo;
}


void troca (int *v, int i, int j){
    int aux = v[i];
    v[i] = v[j];
    v[j] = aux;
    return;
}

• 使线程在不同的 CPU 内核上运行以增加并行度。参照。 pthread_attr_setaffinity_np()
• 使线程实时。参照。 pthread_attr_setschedpolicy() 和 pthread_attr_setschedparam()