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
|
/*
* (C) Copyright 2005- ECMWF.
*
* This software is licensed under the terms of the Apache Licence Version 2.0
* which can be obtained at http://www.apache.org/licenses/LICENSE-2.0.
*
* In applying this licence, ECMWF does not waive the privileges and immunities granted to it by
* virtue of its status as an intergovernmental organisation nor does it submit to any jurisdiction.
*/
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include "grib_api_internal.h"
#define NUM_THREADS 8
#define FILES_PER_ITERATION 10
char* INPUT_FILE = NULL;
void do_stuff(void* data);
void* runner(void* ptr); /* the thread function */
int main(int argc, char** argv)
{
#if GRIB_PTHREADS
long i;
int thread_counter = 0;
pthread_t workers[NUM_THREADS];
INPUT_FILE = argv[1];
/* Create worker threads */
for (i = 0; i < NUM_THREADS; i++) {
/* Now we will create the thread passing it data as a paramater*/
pthread_create(&workers[thread_counter], NULL, runner, (void*)i);
thread_counter++;
}
/* Waiting for threads to complete */
for (i = 0; i < NUM_THREADS; i++) {
pthread_join(workers[i], NULL);
}
#elif GRIB_OMP_THREADS
long i;
INPUT_FILE = argv[1];
#pragma omp parallel for schedule(static) num_threads(NUM_THREADS)
for (i = 0; i < NUM_THREADS; i++) {
do_stuff((void*)i);
}
#else
fprintf(stderr, "This test requires either GRIB_PTHREADS or GRIB_OMP_THREADS to be set to a non-zero value\n");
return 1;
#endif
return 0;
}
void* runner(void* ptr)
{
do_stuff(ptr);
pthread_exit(0);
}
static int encode_file(const char* input_file, const char* output_file)
{
grib_handle* source_handle = NULL;
const void* buffer = NULL;
int err = 0;
FILE* in = fopen(input_file, "rb");
FILE* out = fopen(output_file, "wb");
ECCODES_ASSERT(in);
ECCODES_ASSERT(out);
while ((source_handle = grib_handle_new_from_file(0, in, &err)) != NULL) {
size_t size = 0, values_len = 0;
int i;
double* values = NULL;
long count;
double d, e;
grib_handle* clone_handle = grib_handle_clone(source_handle);
ECCODES_ASSERT(clone_handle);
GRIB_CHECK(grib_get_size(clone_handle, "values", &values_len), 0);
values = (double*)malloc(values_len * sizeof(double));
d = 10e-10;
e = d;
count = 1;
for (i = 0; i < values_len; i++) {
if (count > 1000) {
e *= 2;
count = 1;
}
values[i] = d;
d += e;
count++;
}
GRIB_CHECK(grib_set_long(clone_handle, "bitsPerValue", 16), 0);
/*GRIB_CHECK(grib_set_string(clone_handle, "packingType", "grid_ccsds", &str_len), 0);*/
/*GRIB_CHECK(grib_set_string(clone_handle, "packingType", "grid_simple", &str_len), 0);*/
GRIB_CHECK(grib_set_double_array(clone_handle, "values", values, values_len), 0);
/* get the coded message in a buffer */
GRIB_CHECK(grib_get_message(clone_handle, &buffer, &size), 0);
/* write the buffer to a file */
if (fwrite(buffer, 1, size, out) != size) {
perror(output_file);
return 1;
}
grib_handle_delete(clone_handle);
grib_handle_delete(source_handle);
free(values);
}
fclose(out);
fclose(in);
return 0;
}
void do_stuff(void* arg)
{
long number = (long)arg;
char output_file[50];
int i;
for (i = 0; i < FILES_PER_ITERATION; i++) {
snprintf(output_file, 50, "temp.grib_encode_pthreads_test.out_%d-%d.grib", (int)number, i);
encode_file(INPUT_FILE, output_file);
}
}
|
