並列分散C班

4.9 教科書:p171

ソース概要

ppb_queue.h

ppb_pipeline_isort.c

ppb_queue.h

#ifndef _PPB_QUEUE_H
#define _PPB_QUEUE_H

#ifndef MAX_QUEUE_NUM
#define MAX_QUEUE_NUM 10
#endif /* MAX_QUEUE_NUM */

/* キューの設定の構造体 */
typedef struct _queue {
    int values[MAX_QUEUE_NUM];  /* キューのバッファ領域 (環状バッファ) */
    int remain;        /* キュー中のアイテム数 */
    int rp, wp;                 /* バッファ中の現在の読出し点と書込み点 */
    pthread_mutex_t mutex;      /* キュー操作用の mutex */
    pthread_cond_t not_full;    /* キューが満杯かを見張る条件変数 */
    pthread_cond_t not_empty;   /* キューが空かを見張るための条件変数 */
} queue_t;

/* キューに数値vを追加する 
(キューが満杯の場合、空きが出来るまで待つ) */
void enqueue(queue_t* q, int v) {
    pthread_mutex_lock(&q->mutex);
    while (q->remain == MAX_QUEUE_NUM)
        pthread_cond_wait(&q->not_full, &q->mutex);
    q->values[q->wp] = v;
    q->wp++; q->remain++;
    if (q->wp == MAX_QUEUE_NUM) q->wp = 0;
    pthread_cond_signal(&q->not_empty);
    pthread_mutex_unlock(&q->mutex);
}
back_main

/* キューから数値vを取り出す 
(キューが空の場合、値が追加されるまで待つ) */
void dequeue(queue_t* q, int* v) {
pthread_mutex_lock(&q->mutex);
    while (q->remain == 0) 
    pthread_cond_wait(&q->not_empty, &q->mutex);
        *v = q->values[q->rp];
    q->rp++; q->remain--;
    if (q->rp == MAX_QUEUE_NUM) q->rp = 0;
    pthread_cond_signal(&q->not_full);
    pthread_mutex_unlock(&q->mutex);
}
back_main

#endif /* _PPB_QUEUE_H */

ppb_pipeline_isort.c

#include "ppb_queue.h"

#define VALUES_NUM 100          /* ソートする数の個数 */
#define MAX_QUEUE_NUM  10       /* スレッド間のキューで保持する最大アイテム数 */


typedef struct _thread_arg {
    int value;
    int id;
} thread_arg_t;

thread_arg_t targ[VALUES_NUM];
int numbers[VALUES_NUM];        /* ソートする数を格納 */
queue_t numq[VALUES_NUM+1];     /* スレッド間の値受け渡し用のキュー */

int main()
{
    int i = 0;
    pthread_t master;
    pthread_t handle[VALUES_NUM];

    /* ソートする数列の初期化
     * rand 関数は 0〜32767までの値を表示する。
     * 
     *
     * */
    for (i = 0; i < VALUES_NUM; i++) 
        numbers[i] = rand() % (VALUES_NUM * 10);

    /* 各スレッド間に置かれるキューの初期化 */
    for (i = 0; i < VALUES_NUM + 1; i++) {
        numq[i].rp = numq[i].wp = numq[i].remain = 0;
        pthread_mutex_init(&numq[i].mutex, NULL);
        pthread_cond_init(&numq[i].not_full, NULL);
        pthread_cond_init(&numq[i].not_empty, NULL);
    }

    /* 初期数列の表示 並べたあとも同じ配列に格納する */
    print_numbers("before", numbers);

    /* マスタースレッドの作成 */
    pthread_create(&master, NULL, (void *)master_func, NULL);

    /* 子スレッドの作成 */
    for (i = 0; i < VALUES_NUM; i++) {
        targ[i].value = INT_MAX;/* INT_MAX:int 型の最大値 */
        targ[i].id = i;
        pthread_create(&handle[i], NULL, (void *)thread_func (void*)&targ[i]); 
    }
    /* 最後のスレッドの処理が終わるのを待つ */
    pthread_join(handle[VALUES_NUM - 1], NULL);

    /* ソート結果の表示 */
    print_numbers("after", numbers);

    return 0;
}

/* 指定された数列の値を表示 */
void  print_numbers(char *message, int *array) {
    int i;
    printf("-- %s --\n", message);
    for (i = 0; i < VALUES_NUM; i++) 
        printf("%d ", array[i]);
    printf("\n");
}
back_main

void master_func(void *arg) {
    int i, v;
    /* ソートする数を順に最初のスレッドのキューに追加 */
    for (i = 0; i < VALUES_NUM; i++) 
        enqueue(&numq[0], numbers[i]);
    /* 最後のキューからの出力を読み出す(値は捨てるだけ) */
    for (i = 0; i < VALUES_NUM; i++) 
        dequeue(&numq[VALUES_NUM], &v);
}
back_main
void  thread_func(void *arg)
{
    /*
     * value, id の受け取り。id はスレッドの番号。
     * 最終的に格納する配列の要素番号を表す。
     */
    thread_arg_t *targ = (thread_arg_t*)arg;
    int i, v;
    for (i = 0; i < VALUES_NUM; i++) {
        /* ソートする数の個数分だけキューから値を読む */
        dequeue(&numq[targ->id], &v);
        /* 自分の保持する値と比べて、大きい方を次のスレッドのキューに渡す */
        if (v > targ->value) {
            enqueue(&numq[targ->id+1], v);
        } else {
            enqueue(&numq[targ->id+1], targ->value);
            targ->value = v;
        }
    }
    numbers[targ->id] = targ->value;
}
back_main
       

実行結果