1
/* -*- mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; -*-
2
* vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
4
* Copyright (C) 2009 Sun Microsystems, Inc.
6
* This program is free software; you can redistribute it and/or modify
7
* it under the terms of the GNU General Public License as published by
8
* the Free Software Foundation; version 2 of the License.
10
* This program is distributed in the hope that it will be useful,
11
* but WITHOUT ANY WARRANTY; without even the implied warranty of
12
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13
* GNU General Public License for more details.
15
* You should have received a copy of the GNU General Public License
16
* along with this program; if not, write to the Free Software
17
* Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
21
Just a test application for threads.
29
#include <sys/types.h>
31
#include <sys/types.h>
36
#include <string.h> /* Pull in memset() */
45
#define snprintf _snprintf
48
#include <boost/scoped_ptr.hpp>
52
#define DEFAULT_CONCURRENCY 10
53
#define DEFAULT_INITIAL_LOAD 10000
54
#define DEFAULT_EXECUTE_SECONDS 120
55
#define TEST_FILENAME "concurrency_test.az"
57
#define HUGE_STRING_LENGTH 8192
59
/* Global Thread counter */
60
unsigned int thread_counter;
61
pthread_mutex_t counter_mutex;
62
pthread_cond_t count_threshhold;
63
unsigned int master_wakeup;
64
pthread_mutex_t sleeper_mutex;
65
pthread_cond_t sleep_threshhold;
66
static bool timer_alarm= false;
67
pthread_mutex_t timer_alarm_mutex;
68
pthread_cond_t timer_alarm_threshold;
70
pthread_mutex_t row_lock;
74
void *run_concurrent_task(void *p);
75
void *timer_thread(void *p);
77
void scheduler(az_method use_aio);
78
void create_data_file(azio_stream *write_handler, uint64_t rows);
79
unsigned int write_row(azio_stream *s);
81
typedef struct thread_context_st thread_context_st;
82
struct thread_context_st {
83
unsigned int how_often_to_write;
89
/* Use this for string generation */
90
static const char ALPHANUMERICS[]=
91
"0123456789ABCDEFGHIJKLMNOPQRSTWXYZabcdefghijklmnopqrstuvwxyz";
93
#define ALPHANUMERICS_SIZE (sizeof(ALPHANUMERICS)-1)
95
static void get_random_string(char *buffer, size_t size)
97
char *buffer_ptr= buffer;
100
*buffer_ptr++= ALPHANUMERICS[random() % ALPHANUMERICS_SIZE];
101
*buffer_ptr++= ALPHANUMERICS[random() % ALPHANUMERICS_SIZE];
104
int main(int argc, char *argv[])
108
drizzled::internal::my_init();
117
pthread_mutex_init(&counter_mutex, NULL);
118
pthread_cond_init(&count_threshhold, NULL);
119
pthread_mutex_init(&sleeper_mutex, NULL);
120
pthread_cond_init(&sleep_threshhold, NULL);
121
pthread_mutex_init(&timer_alarm_mutex, NULL);
122
pthread_cond_init(&timer_alarm_threshold, NULL);
123
pthread_mutex_init(&row_lock, NULL);
125
for (method= AZ_METHOD_BLOCK; method < AZ_METHOD_MAX; method++)
126
scheduler((az_method)method);
128
(void)pthread_mutex_destroy(&counter_mutex);
129
(void)pthread_cond_destroy(&count_threshhold);
130
(void)pthread_mutex_destroy(&sleeper_mutex);
131
(void)pthread_cond_destroy(&sleep_threshhold);
132
pthread_mutex_destroy(&timer_alarm_mutex);
133
pthread_cond_destroy(&timer_alarm_threshold);
134
pthread_mutex_destroy(&row_lock);
139
void scheduler(az_method use_aio)
143
boost::scoped_ptr<azio_stream> writer_handle_ap(new azio_stream);
144
azio_stream &writer_handle= *writer_handle_ap.get();
145
thread_context_st *context;
146
pthread_t mainthread; /* Thread descriptor */
147
pthread_attr_t attr; /* Thread attributes */
149
pthread_attr_init(&attr);
150
pthread_attr_setdetachstate(&attr,
151
PTHREAD_CREATE_DETACHED);
153
pthread_mutex_lock(&counter_mutex);
156
create_data_file(&writer_handle, DEFAULT_INITIAL_LOAD);
158
pthread_mutex_lock(&sleeper_mutex);
160
pthread_mutex_unlock(&sleeper_mutex);
162
context= (thread_context_st *)malloc(sizeof(thread_context_st) * DEFAULT_CONCURRENCY);
163
memset(context, 0, sizeof(thread_context_st) * DEFAULT_CONCURRENCY);
167
fprintf(stderr, "Could not allocate memory for context\n");
171
for (x= 0; x < DEFAULT_CONCURRENCY; x++)
174
context[x].how_often_to_write= random()%1000;
175
context[x].writer= &writer_handle;
176
context[x].counter= 0;
177
context[x].use_aio= use_aio;
179
/* now you create the thread */
180
if (pthread_create(&mainthread, &attr, run_concurrent_task,
181
(void *)context) != 0)
183
fprintf(stderr,"Could not create thread\n");
189
if (DEFAULT_EXECUTE_SECONDS)
191
time_t opt_timer_length= DEFAULT_EXECUTE_SECONDS;
192
pthread_mutex_lock(&timer_alarm_mutex);
194
pthread_mutex_unlock(&timer_alarm_mutex);
196
if (pthread_create(&mainthread, &attr, timer_thread,
197
(void *)&opt_timer_length) != 0)
199
fprintf(stderr,"%s: Could not create timer thread\n", drizzled::internal::my_progname);
204
pthread_mutex_unlock(&counter_mutex);
205
pthread_attr_destroy(&attr);
207
pthread_mutex_lock(&sleeper_mutex);
209
pthread_mutex_unlock(&sleeper_mutex);
210
pthread_cond_broadcast(&sleep_threshhold);
213
We loop until we know that all children have cleaned up.
215
pthread_mutex_lock(&counter_mutex);
216
while (thread_counter)
218
struct timespec abstime;
220
memset(&abstime, 0, sizeof(struct timespec));
223
pthread_cond_timedwait(&count_threshhold, &counter_mutex, &abstime);
225
pthread_mutex_unlock(&counter_mutex);
227
for (total= x= 0; x < DEFAULT_CONCURRENCY; x++)
228
total+= context[x].counter;
231
azclose(&writer_handle);
233
printf("Read %"PRIu64" rows\n", total);
236
void *timer_thread(void *p)
238
time_t *timer_length= (time_t *)p;
239
struct timespec abstime;
242
We lock around the initial call in case were we in a loop. This
243
also keeps the value properly syncronized across call threads.
245
pthread_mutex_lock(&sleeper_mutex);
246
while (master_wakeup)
248
pthread_cond_wait(&sleep_threshhold, &sleeper_mutex);
250
pthread_mutex_unlock(&sleeper_mutex);
252
set_timespec(abstime, *timer_length);
254
pthread_mutex_lock(&timer_alarm_mutex);
255
pthread_cond_timedwait(&timer_alarm_threshold, &timer_alarm_mutex, &abstime);
256
pthread_mutex_unlock(&timer_alarm_mutex);
258
pthread_mutex_lock(&timer_alarm_mutex);
260
pthread_mutex_unlock(&timer_alarm_mutex);
265
void *run_concurrent_task(void *p)
267
thread_context_st *context= (thread_context_st *)p;
271
boost::scoped_ptr<azio_stream> reader_handle_ap(new azio_stream);
272
azio_stream &reader_handle= *reader_handle_ap.get();
274
if (!(ret= azopen(&reader_handle, TEST_FILENAME, O_RDONLY,
277
printf("Could not open test file\n");
281
pthread_mutex_lock(&sleeper_mutex);
282
while (master_wakeup)
284
pthread_cond_wait(&sleep_threshhold, &sleeper_mutex);
286
pthread_mutex_unlock(&sleeper_mutex);
292
azread_init(&reader_handle);
293
while ((ret= azread_row(&reader_handle, &error)))
296
if (count % context->how_often_to_write)
298
write_row(context->writer);
301
/* If the timer is set, and the alarm is not active then end */
302
if (timer_alarm == false)
306
pthread_mutex_lock(&counter_mutex);
308
pthread_cond_signal(&count_threshhold);
309
pthread_mutex_unlock(&counter_mutex);
310
azclose(&reader_handle);
315
void create_data_file(azio_stream *write_handler, uint64_t rows)
320
if (!(ret= azopen(write_handler, TEST_FILENAME, O_CREAT|O_RDWR|O_TRUNC,
323
printf("Could not create test file\n");
327
for (x= 0; x < rows; x++)
328
write_row(write_handler);
330
azflush(write_handler, Z_SYNC_FLUSH);
333
unsigned int write_row(azio_stream *s)
336
char buffer[HUGE_STRING_LENGTH];
338
length= random() % HUGE_STRING_LENGTH;
340
/* Avoid zero length strings */
343
get_random_string(buffer, length);
344
pthread_mutex_lock(&row_lock);
345
azwrite_row(s, buffer, length);
346
pthread_mutex_unlock(&row_lock);
added
removed
plugin/archive/concurrency_test.cc
