Неверное значение локальной переменной после pthread_join

Question

Я пытаюсь суммировать целочисленный массив из 1000 элементов (где каждый элемент равен 1) с библиотекой pthread, разбивая массив на сегменты размером 10. Таким образом, для этого используются 100 потоков. Результаты этой параллельной операции ожидаются (1000). Но что интересно, последовательная сумма, которую я вычислил перед созданием потоков, обнуляется после моего первого вызова pthread_join(). Не уверен, что мне что-то здесь не хватает. Может ли кто-нибудь обнаружить здесь ошибку?

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

#define SEGMENT_SIZE 10
#define NUM_THREADS 100
int *array = NULL;

void* segment_sum(void *args)
{
        int index = (int)args;
        int sum = 0;

        for (int i = index * SEGMENT_SIZE; i < (index + 1) * SEGMENT_SIZE; i++) {
                sum += array[i];
        }

        return (void *)sum;
}

int main()
{
        pthread_t thread[NUM_THREADS];
        int res = 0;
        int seq_res = 0;
        int par_res = 0;

        array = calloc(1, sizeof(int) * NUM_THREADS * SEGMENT_SIZE);
        for (int i = 0; i < NUM_THREADS * SEGMENT_SIZE; i++) {
                array[i] = 1;
                seq_res += 1;
        }


        for (int i = 0; i < NUM_THREADS; i++) {
                res = pthread_create(&thread[i], NULL, segment_sum, (void *)i);
                if (res != 0) {
                        printf("\nError creating new thread");
                }
        }

        printf("\nindex = %d", seq_res);                // the sequential sum here is 1000

        for (int i = 0; i < NUM_THREADS; i++) {
                int sum = 0;
                res = pthread_join(thread[i], (void **)&sum);
                if (res != 0) {
                        printf("\nError creating new thread");
                }
                printf("\nindex = %d", seq_res);        // Here it is becoming zero!!!

                par_res += sum;
        }

        printf("\nmultithreaded sum: %d single threaded sum: %d\n", par_res, seq_res);
}

Answer 1

При компиляции программы постарайтесь как можно больше исключить предупреждения, поскольку они часто указывают на непереносимое поведение или скрытые ошибки. Здесь компиляция указывает на следующее:

pte.c: In function 'segment_sum':
pte.c:11:21: warning: cast from pointer to integer of different size [-Wpointer-to-int-cast]
   11 |         int index = (int)args;
      |                     ^
pte.c:18:16: warning: cast to pointer from integer of different size [-Wint-to-pointer-cast]
   18 |         return (void *)sum;
      |                ^
pte.c: In function 'main':
pte.c:36:69: warning: cast to pointer from integer of different size [-Wint-to-pointer-cast]
   36 |                 res = pthread_create(&thread[i], NULL, segment_sum, (void *)i);
      |                                                                     ^

Параметр, передаваемый потокам, является преобразованием указателя в "int". Рекомендуется передавать адрес в виде int. Следовательно, вы можете определить контекст для каждого потока:

struct thd_ctx {
  pthread_t thread;
  int index;
  int sum;
};

pthread_join () передается адрес указателя, который получает адрес ячейки памяти, в которой поток сохранил свой результат. Поток должен возвращать адрес этой ячейки памяти, а не хранимое в ней значение. Более того, поток не должен возвращать адрес автоматической переменной (т.е. в своем стеке), поскольку он не указан. Результатом должен быть адрес глобальной переменной (или "чего-то", видимого из присоединяющегося потока), возвращенный либо напрямую, либо через pthread_exit(). В этом усовершенствовании программы мы используем адрес поля "сумма" в контексте потока:

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

#define SEGMENT_SIZE 10
#define NUM_THREADS 100
int *array = NULL;


struct thd_ctx {
  pthread_t thread;
  int index;
  int sum;
};


void *segment_sum(void *args)
{
  int i;
  struct thd_ctx *ctx = (struct thd_ctx *)args;

  ctx->sum = 0;

  for (i = ctx->index * SEGMENT_SIZE; i < (ctx->index + 1) * SEGMENT_SIZE; i++) {
    ctx->sum += array[i];
  }

  return (void *)&(ctx->sum);
}


int main(void)
{
  struct thd_ctx thd_ctx[NUM_THREADS];
  int res = 0;
  int seq_res = 0;
  int par_res = 0;
  int i;

  array = calloc(1, sizeof(int) * NUM_THREADS * SEGMENT_SIZE);
  if (!array) {
    fprintf(stderr, "calloc(): error %d\n", errno);
    return 1;
  }

  for (i = 0; i < NUM_THREADS * SEGMENT_SIZE; i++) {
    array[i] = 1;
    seq_res += 1;
  }

  for (i = 0; i < NUM_THREADS; i++) {
    thd_ctx[i].index = i;
    res = pthread_create(&(thd_ctx[i].thread), NULL, segment_sum, (void *)&(thd_ctx[i]));
    if (res != 0) {
      fprintf(stderr, "Error %d creating new thread#%d\n", res, i);
      free(array);
      return 1;
    }
  }

  printf("Index = %d\n", seq_res);   // the sequential sum here is 1000

  for (i = 0; i < NUM_THREADS; i++) {
    int *sum = 0;
    res = pthread_join(thd_ctx[i].thread, (void **)&(sum));
    if (res != 0) {
      printf("Error %d joining thread#%d", res, i);
      free(array);
      return 1;
    }

    par_res += *sum;

    printf("sum = %d\n", par_res);
  }

  printf("\nMultithreaded sum: %d single threaded sum: %d\n", par_res, seq_res);

  free(array);

  return 0;
}