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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161 | // Copyright 2021 Jeisson Hidalgo <jeisson.hidalgo@ucr.ac.cr> CC-BY 4.0
#include <assert.h>
#include <inttypes.h>
#include <pthread.h>
#include <semaphore.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <unistd.h>
// thread_shared_data_t
typedef struct shared_data {
sem_t* can_greet;
uint64_t thread_count;
} shared_data_t;
// thread_private_data_t
typedef struct private_data {
uint64_t thread_number; // rank
shared_data_t* shared_data;
} private_data_t;
/**
* @brief ...
*/
void* greet(void* data);
int create_threads(shared_data_t* shared_data);
// procedure main(argc, argv[])
int main(int argc, char* argv[]) {
int error = EXIT_SUCCESS;
// create thread_count as result of converting argv[1] to integer
// thread_count := integer(argv[1])
uint64_t thread_count = sysconf(_SC_NPROCESSORS_ONLN);
if (argc == 2) {
if (sscanf(argv[1], "%" SCNu64, &thread_count) == 1) {
} else {
fprintf(stderr, "Error: invalid thread count\n");
return 11;
}
}
shared_data_t* shared_data = (shared_data_t*)calloc(1, sizeof(shared_data_t));
if (shared_data) {
shared_data->can_greet = (sem_t*) calloc(thread_count, sizeof(sem_t));
shared_data->thread_count = thread_count;
for (uint64_t thread_number = 0; thread_number < shared_data->thread_count
; ++thread_number) {
// can_greet[thread_number] := create_semaphore(not thread_number)
error = sem_init(&shared_data->can_greet[thread_number], /*pshared*/ 0
, /*value*/ !thread_number);
}
if (shared_data->can_greet) {
struct timespec start_time, finish_time;
clock_gettime(CLOCK_MONOTONIC, &start_time);
error = create_threads(shared_data);
clock_gettime(CLOCK_MONOTONIC, &finish_time);
double elapsed_time = finish_time.tv_sec - start_time.tv_sec +
(finish_time.tv_nsec - start_time.tv_nsec) * 1e-9;
printf("Execution time: %.9lfs\n", elapsed_time);
free(shared_data->can_greet);
} else {
fprintf(stderr, "Error: could not allocate semaphores\n");
error = 13;
}
free(shared_data);
} else {
fprintf(stderr, "Error: could not allocate shared data\n");
error = 12;
}
return error;
} // end procedure
int create_threads(shared_data_t* shared_data) {
int error = EXIT_SUCCESS;
// for thread_number := 0 to thread_count do
pthread_t* threads = (pthread_t*)
malloc(shared_data->thread_count * sizeof(pthread_t));
private_data_t* private_data = (private_data_t*)
calloc(shared_data->thread_count, sizeof(private_data_t));
if (threads && private_data) {
for (uint64_t thread_number = 0; thread_number < shared_data->thread_count
; ++thread_number) {
if (error == EXIT_SUCCESS) {
private_data[thread_number].thread_number = thread_number;
private_data[thread_number].shared_data = shared_data;
// create_thread(greet, thread_number)
error = pthread_create(&threads[thread_number], /*attr*/ NULL, greet
, /*arg*/ &private_data[thread_number]);
if (error == EXIT_SUCCESS) {
} else {
fprintf(stderr, "Error: could not create secondary thread\n");
error = 21;
break;
}
} else {
fprintf(stderr, "Error: could not init semaphore\n");
error = 22;
break;
}
}
// print "Hello from main thread"
printf("Hello from main thread\n");
for (uint64_t thread_number = 0; thread_number < shared_data->thread_count
; ++thread_number) {
pthread_join(threads[thread_number], /*value_ptr*/ NULL);
sem_destroy(&shared_data->can_greet[thread_number]);
}
free(private_data);
free(threads);
} else {
fprintf(stderr, "Error: could not allocate %" PRIu64 " threads\n"
, shared_data->thread_count);
error = 23;
}
return error;
}
// procedure greet:
void* greet(void* data) {
assert(data);
private_data_t* private_data = (private_data_t*) data;
shared_data_t* shared_data = private_data->shared_data;
// Wait until it is my turn
// wait(can_greet[thread_number])
int error = sem_wait(&shared_data->can_greet[private_data->thread_number]);
if (error) {
fprintf(stderr, "error: could not wait for semaphore\n");
}
// print "Hello from secondary thread"
printf("Hello from secondary thread %" PRIu64 " of %" PRIu64 "\n"
, private_data->thread_number, shared_data->thread_count);
// Allow subsequent thread to do the task
// signal(can_greet[(thread_number + 1) mod thread_count])
const uint64_t next_thread = (private_data->thread_number + 1)
% shared_data->thread_count;
error = sem_post(&shared_data->can_greet[next_thread]);
if (error) {
fprintf(stderr, "error: could not increment semaphore\n");
}
return NULL;
} // end procedure
|