線程安全位陣列？

Question

編輯問題：是否有可能線程安全地訪問位數組？我的下面的實現似乎需要互斥鎖來抵制並行化的目的。

我的任務是使用pthread創建一個雙生成器的並行實現。我決定使用Eratosthenes的篩子，並劃分標記已知素數因子的工作。我stag which線索得到哪些因素。

例如，如果有4個線程： 螺紋1馬克倍數3，11，19，27 ...螺紋 2馬克倍數5，圖13，21，29 ...螺紋 2馬克倍數7， 15,23,31 ... 線程兩個標記倍數9,17,25,33 ...

我跳過了偶數倍數以及偶數基數。我使用了一個bitarray，所以我運行它到INT_MAX。我遇到的問題是最大值爲1000萬，結果大約有5個數字，這與一個已知文件相比有多少誤差。結果一直下降到大約10000的最大值，其中它改變了1個數字。下面的任何內容都是無錯誤的。

起初我並不認爲需要進程間的通信。當我看到結果時，我添加了一個pthread barrier來讓所有線程在每組倍數之後跟上。這沒有任何改變。 圍繞mark（）函數添加一個互斥鎖可以做到這一點，但會減慢一切。

這是我的代碼。希望有人看到明顯的東西。

#include <pthread.h> 
#include <stdio.h> 
#include <sys/times.h> 
#include <stdlib.h> 
#include <unistd.h> 
#include <math.h> 
#include <string.h> 
#include <limits.h> 
#include <getopt.h> 

#define WORDSIZE 32 

struct t_data{ 
    int *ba; 
    unsigned int val; 
    int num_threads; 
    int thread_id; 
}; 

pthread_mutex_t mutex_mark; 

void mark(int *ba, unsigned int k) 
{ 
    ba[k/32] |= 1 << (k%32); 
} 

void mark(int *ba, unsigned int k) 
{ 
    pthread_mutex_lock(&mutex_mark); 
    ba[k/32] |= 1 << (k%32); 
    pthread_mutex_unlock(&mutex_mark); 
} 

void initBa(int **ba, unsigned int val) 
{ 
    *ba = calloc((val/WORDSIZE)+1, sizeof(int)); 
} 

void getPrimes(int *ba, unsigned int val) 
{ 
    int i, p; 
    p = -1; 

    for(i = 3; i<=val; i+=2){ 
      if(!isMarked(ba, i)){ 
        if(++p == 8){ 
         printf(" \n"); 
         p = 0; 
        } 
        printf("%9d", i); 
      } 
    } 
    printf("\n"); 
} 

void markTwins(int *ba, unsigned int val) 
{ 
    int i; 
    for(i=3; i<=val; i+=2){ 
     if(!isMarked(ba, i)){ 
      if(isMarked(ba, i+2)){ 
       mark(ba, i); 
      } 

     } 
    } 
} 

void *setPrimes(void *arg) 
{ 
    int *ba, thread_id, num_threads, status; 
    unsigned int val, i, p, start; 
    struct t_data *data = (struct t_data*)arg; 
    ba = data->ba; 
    thread_id = data->thread_id; 
    num_threads = data->num_threads; 
    val = data->val; 

    start = (2*(thread_id+2))-1; // stagger threads 

    i=3; 
    for(i=3; i<=sqrt(val); i+=2){ 
     if(!isMarked(ba, i)){ 
      p=start; 
      while(i*p <= val){ 
        mark(ba, (i*p)); 
       p += (2*num_threads); 
      }  
     }  
    } 
    return 0; 
} 

void usage(char *filename) 
{ 
    printf("Usage: \t%s [option] [arg]\n", filename); 
    printf("\t-q generate #'s internally only\n"); 
    printf("\t-m [size] maximum size twin prime to calculate\n"); 
    printf("\t-c [threads] number of threads\n"); 
    printf("Defaults:\n\toutput results\n\tsize = INT_MAX\n\tthreads = 1\n"); 
} 

int main(int argc, char **argv) 
{ 
    int *ba, i, num_threads, opt, output; 
    unsigned int val; 

    output = 1; 
    num_threads = 1; 
    val = INT_MAX; 

    while ((opt = getopt(argc, argv, "qm:c:")) != -1){ 
     switch (opt){ 
      case 'q': output = 0; 
       break; 
      case 'm': val = atoi(optarg); 
       break; 
      case 'c': num_threads = atoi(optarg); 
       break; 
      default: 
       usage(argv[0]); 
       exit(EXIT_FAILURE); 
     } 
    } 

    struct t_data data[num_threads];  
    pthread_t thread[num_threads]; 
    pthread_attr_t attr; 

    pthread_mutex_init(&mutex_mark, NULL); 

    initBa(&ba, val); 

    pthread_attr_init(&attr); 
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);  

    for(i=0; i < num_threads; i++){ 
     data[i].ba = ba; 
     data[i].thread_id = i; 
     data[i].num_threads = num_threads; 
     data[i].val = val; 
     if(0 != pthread_create(&thread[i], 
           &attr, 
           setPrimes, 
           (void*)&data[i])){ 
      perror("Cannot create thread"); 
      exit(EXIT_FAILURE); 
     } 
    } 

    for(i = 0; i < num_threads; i++){ 
     pthread_join(thread[i], NULL); 
    } 

    markTwins(ba, val); 
    if(output) 
     getPrimes(ba, val); 

    free(ba); 
    return 0; 
} 

編輯：我擺脫了障礙，並添加了一個mutex_lock標記功能。現在輸出是準確的，但是現在不止一個線程減慢了速度。任何關於加速的建議？

Answer 1

您當前正在執行的標記是正確的，但鎖定非常粗糙 - 整個數組只有一個鎖。這意味着你的線程不斷爭奪該鎖。

提高性能的一種方法是使鎖定細粒度：每個「標記」的操作只需要陣的一個整數獨佔訪問，所以你可以爲每個數組項互斥：

struct bitarray 
{ 
    int *bits; 
    pthread_mutex_t *locks; 
}; 

struct t_data 
{ 
    struct bitarray ba; 
    unsigned int val; 
    int num_threads; 
    int thread_id; 
}; 

void initBa(struct bitarray *ba, unsigned int val) 
{ 
    const size_t array_size = val/WORDSIZE + 1; 
    size_t i; 

    ba->bits = calloc(array_size, sizeof ba->bits[0]); 
    ba->locks = calloc(array_size, sizeof ba->locks[0]); 

    for (i = 0; i < array_size; i++) 
    { 
     pthread_mutex_init(&ba->locks[i], NULL); 
    } 
} 

void mark(struct bitarray ba, unsigned int k) 
{ 
    const unsigned int entry = k/32; 

    pthread_mutex_lock(&ba.locks[entry]); 
    ba.bits[entry] |= 1 << (k%32); 
    pthread_mutex_unlock(&ba.locks[entry]); 
} 

請注意，您的算法有競爭條件：考慮例如num_threads = 4，所以線程0從3開始，線程1從5開始，線程2從7開始。線程2可以完全執行，標記每個倍數爲7，然後再從15開始，之前線程0或線程1有機會將15標記爲3或5的倍數。線程2然後將做無用的工作，markin g每15的倍數。

---

另一種選擇，如果你的編譯器支持英特爾的風格原子內建命令，就是用這些代替鎖：

void mark(int *ba, unsigned int k) 
{ 
    __sync_or_and_fetch(&ba[k/32], 1U << k % 32); 
} 

Answer 2

你的mark()函數不是線程安全的 - 如果兩個線程嘗試在同一個int位置內設置位，可能會用另一個線程剛剛設置的位0覆蓋。