MapReduce/Barrier.cpp at master · itay-rubin/MapReduce · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
#include "Barrier.h"
#include <cstdlib>
#include <cstdio>

/**
 * Constructor for Barrier
 * @param numThreads - total number of threads the job is working with
 */
Barrier::Barrier(int numThreads)
 : startMutex(PTHREAD_MUTEX_INITIALIZER)
 , startCv(PTHREAD_COND_INITIALIZER)
 , waitMutex(PTHREAD_MUTEX_INITIALIZER)
 , waitCv(PTHREAD_COND_INITIALIZER)
 , startCount(0)
 , waitCount(0)
 , numThreads(numThreads)
 , jobDone(false)
{ }


/**
 * Barrier destructor
 */
Barrier::~Barrier()
{
	if (pthread_mutex_destroy(&startMutex) != 0 || pthread_mutex_destroy(&waitMutex) != 0) {
		fprintf(stderr, "[[Barrier]] error on pthread_mutex_destroy");
		exit(1);
	}
	if (pthread_cond_destroy(&startCv) != 0 || pthread_cond_destroy(&waitCv) != 0){
		fprintf(stderr, "[[Barrier]] error on pthread_cond_destroy");
		exit(1);
	}
}


/**
* A barrier for waitForJob
* @param done - true iff called by the last thread after job is done
*/
void Barrier::waitBarrier(bool done)
{
    if (pthread_mutex_lock(&waitMutex) != 0){
        fprintf(stderr, "[[Barrier]] error on pthread_mutex_lock");
        exit(1);
    }
    if (!done && !jobDone) {
        if (pthread_cond_wait(&waitCv, &waitMutex) != 0){
            fprintf(stderr, "[[Barrier]] error on pthread_cond_wait");
            exit(1);
        }
    } else {
        if (pthread_cond_broadcast(&waitCv) != 0) {
            fprintf(stderr, "[[Barrier]] error on pthread_cond_broadcast");
            exit(1);
        }
        jobDone = true;
    }
    if (pthread_mutex_unlock(&waitMutex) != 0) {
        fprintf(stderr, "[[Barrier]] error on pthread_mutex_unlock");
        exit(1);
    }
}


/**
 * A barrier for 2 places: before start mapping and before start reducing
 */
void Barrier::startBarrier()
{
    if (pthread_mutex_lock(&startMutex) != 0){
        fprintf(stderr, "[[Barrier]] error on pthread_mutex_lock");
        exit(1);
    }
    if (++startCount < numThreads) {
        if (pthread_cond_wait(&startCv, &startMutex) != 0){
            fprintf(stderr, "[[Barrier]] error on pthread_cond_wait");
            exit(1);
        }
    } else {
        if (pthread_cond_broadcast(&startCv) != 0)
        {
            fprintf(stderr, "[[Barrier]] error on pthread_cond_broadcast");
            exit(1);
        }
        startCount = 0;
    }
    if (pthread_mutex_unlock(&startMutex) != 0) {
        fprintf(stderr, "[[Barrier]] error on pthread_mutex_unlock");
        exit(1);
    }
}