在C中使用线程的递归函数

Question

我有下面的代码

    #include "stdio.h"
#include "stdlib.h"
#include <string.h>

#define MAXBINS 8


void swap_long(unsigned long int **x, unsigned long int **y){

  unsigned long int *tmp;
  tmp = x[0];
  x[0] = y[0];
  y[0] = tmp;

}

void swap(unsigned int **x, unsigned int **y){

  unsigned int *tmp;
  tmp = x[0];
  x[0] = y[0];
  y[0] = tmp;

}

void truncated_radix_sort(unsigned long int *morton_codes, 
              unsigned long int *sorted_morton_codes, 
              unsigned int *permutation_vector,
              unsigned int *index,
              int *level_record,
              int N, 
              int population_threshold,
              int sft, int lv){

  int BinSizes[MAXBINS] = {0};
  unsigned int *tmp_ptr;
  unsigned long int *tmp_code;

  level_record[0] = lv; // record the level of the node

  if(N<=population_threshold || sft < 0) { // Base case. The node is a leaf
    memcpy(permutation_vector, index, N*sizeof(unsigned int)); // Copy the pernutation vector
    memcpy(sorted_morton_codes, morton_codes, N*sizeof(unsigned long int)); // Copy the Morton codes 

    return;
  }
  else{

    // Find which child each point belongs to 
    int j = 0;
    for(j=0; j<N; j++){
      unsigned int ii = (morton_codes[j]>>sft) & 0x07;
      BinSizes[ii]++;
    }


    // scan prefix 
    int offset = 0, i = 0;
    for(i=0; i<MAXBINS; i++){
      int ss = BinSizes[i];
      BinSizes[i] = offset;
      offset += ss;
    }

    for(j=0; j<N; j++){
      unsigned int ii = (morton_codes[j]>>sft) & 0x07;
      permutation_vector[BinSizes[ii]] = index[j];
      sorted_morton_codes[BinSizes[ii]] = morton_codes[j];
      BinSizes[ii]++;
    }

    //swap the index pointers  
    swap(&index, &permutation_vector);

    //swap the code pointers 
    swap_long(&morton_codes, &sorted_morton_codes);

    /* Call the function recursively to split the lower levels */
    offset = 0; 
    for(i=0; i<MAXBINS; i++){

      int size = BinSizes[i] - offset;

      truncated_radix_sort(&morton_codes[offset], 
               &sorted_morton_codes[offset], 
               &permutation_vector[offset], 
               &index[offset], &level_record[offset], 
               size, 
               population_threshold,
               sft-3, lv+1);
      offset += size;  
    }


  } 
}

我试着把这个街区

int j = 0;
    for(j=0; j<N; j++){
      unsigned int ii = (morton_codes[j]>>sft) & 0x07;
      BinSizes[ii]++;
    }

将其替换为下面的

    int rc,j;
    pthread_t *thread = (pthread_t *)malloc(NTHREADS*sizeof(pthread_t));
    belong *belongs = (belong *)malloc(NTHREADS*sizeof(belong));
    pthread_mutex_init(&bin_mtx, NULL);
    for (j = 0; j < NTHREADS; j++){
        belongs[j].n = NTHREADS;
        belongs[j].N = N;
        belongs[j].tid = j;
        belongs[j].sft = sft;
        belongs[j].BinSizes = BinSizes;
        belongs[j].mcodes = morton_codes;
        rc = pthread_create(&thread[j], NULL, belong_wrapper, (void *)&belongs[j]);
    }

    for (j = 0; j < NTHREADS; j++){
        rc = pthread_join(thread[j], NULL);
    }

并在递归函数之外定义这些

typedef struct{
    int n, N, tid, sft;
    int *BinSizes;
    unsigned long int *mcodes;
}belong;

pthread_mutex_t bin_mtx;

void * belong_wrapper(void *arg){
    int n, N, tid, sft, j;
    int *BinSizes;
    unsigned int ii;
    unsigned long int *mcodes;
    n = ((belong *)arg)->n;
    N = ((belong *)arg)->N;
    tid = ((belong *)arg)->tid;
    sft = ((belong *)arg)->sft;
    BinSizes = ((belong *)arg)->BinSizes;
    mcodes = ((belong *)arg)->mcodes;
    for (j = tid; j<N; j+=n){
        ii = (mcodes[j] >> sft) & 0x07;
        pthread_mutex_lock(&bin_mtx);
        BinSizes[ii]++;
        pthread_mutex_unlock(&bin_mtx);
    }

}

不过，执行起来比串行的时间要长得多.为什么会发生这种情况？我该换什么？

Answer 1

由于您使用单个互斥锁来保护对BinSizes数组的更新，所以最终还是要对该数组执行所有更新:在任何给定的时间，只有一个线程可以调用BinSizes[ii]++。基本上，您仍然在按顺序执行您的函数，但是会产生创建和销毁线程的额外开销。

我可以为您想到几个选项(可能还有更多)：

1. 按照@Chris的建议做，并让每个线程更新BinSizes的一部分。根据用于计算ii的计算的属性，这可能是不可行的。
2. 创建多个互斥对象，表示BinSizes的不同分区。例如，如果BinSizes有10个元素，您可以为元素0-4创建一个互斥，为元素5-9创建另一个互斥，然后在线程中使用它们，如下所示： 若(ii < 5) { mtx_index = 0；} pthread_mutex_lock(&bin_mtxmtx_index)；BinSizesii++；pthread_mutex_unlock(&bin_mtxmtx_index)； 您可以将这个想法推广到任意大小的BinSizes和任何范围:可能您可以为每个数组元素设置一个不同的互斥对象。当然，如果有人试图同时锁定多个互斥对象，那么您将面临创建每一个互斥对象的开销，以及死锁的可能性。
3. 最后，您可以完全放弃并行化这个块的想法:正如其他用户所提到的那样，使用线程的方式会受到某种程度的收益递减的影响。除非您的BinSizes数组非常大，否则即使您“做得对”，也可能看不到并行化的巨大好处。

Answer 2

对于大多数问题，tl;dr添加线程并不是一个简单的解决方法。您的代码不具有令人难堪的可并行性，而且该代码几乎没有任何实际的并发性。

您可以为BinSizes上的每一个(廉价的)整数操作旋转互斥体。这将粉碎任何并行性，因为您的所有线程都是在此上序列化的。

您可以同时运行的少数指令( for循环和morton代码数组上的几个操作)比(Un)锁定互斥对象要便宜得多:即使使用原子增量(如果可用的话)，也比不同步部分要昂贵得多。

一个解决方法是为每个线程提供自己的输出数组，并在所有任务完成后将它们组合起来。

此外，每次调用都要创建和连接多个线程。与计算相比，创建线程相对较昂贵，因此通常建议创建一个长期存在的线程池，以扩展该成本。

即使这样做，您也需要根据有多少(免费)内核来调整线程数量。如果在递归函数中执行此操作，那么同时存在多少线程？创建比内核更多的线程来调度线程是没有意义的。

哦，你在漏掉记忆。