« back to all changes in this revision
Viewing changes to client/drizzleslap.cc
- browse files at revision 1897
- compare with another revision
- compare with revision 2040.6.5
- download diff
- download tarball
- view history from revision 1897
- Committer: Brian Aker
- Date: 2010-11-01 03:30:01 UTC
- Revision ID: brian@tangent.org-20101101033001-vycyi7sky3jc8uo9
Boost fixup for slap
- files modified:
- client/drizzleslap.cc
added
removed
client/drizzleslap.cc
93
93
#include <string>
94
94
#include <iostream>
95
95
#include <fstream>
96
#include <pthread.h>
97
96
#include <drizzled/configmake.h>
98
97
/* Added this for string translation. */
99
98
#include <drizzled/gettext.h>
99
100
#include <boost/thread.hpp>
101
#include <boost/thread/mutex.hpp>
102
#include <boost/thread/condition_variable.hpp>
100
103
#include <boost/program_options.hpp>
101
104
102
105
#define SLAP_VERSION "1.5"
115
118
116
119
/* Global Thread counter */
117
120
uint32_t thread_counter;
118
pthread_mutex_t counter_mutex;
119
pthread_cond_t count_threshhold;
120
uint32_t master_wakeup;
121
pthread_mutex_t sleeper_mutex;
122
pthread_cond_t sleep_threshhold;
121
boost::mutex counter_mutex;
122
boost::condition_variable_any count_threshhold;
123
bool master_wakeup;
124
boost::mutex sleeper_mutex;
125
boost::condition_variable_any sleep_threshhold;
123
126
124
127
/* Global Thread timer */
125
128
static bool timer_alarm= false;
126
pthread_mutex_t timer_alarm_mutex;
127
pthread_cond_t timer_alarm_threshold;
129
boost::mutex timer_alarm_mutex;
130
boost::condition_variable_any timer_alarm_threshold;
128
131
129
132
std::vector < std::string > primary_keys;
130
133
225
228
OptionString *engine_stmt, Stats *sptr);
226
229
static int run_scheduler(Stats *sptr, Statement **stmts, uint32_t concur,
227
230
uint64_t limit);
228
extern "C" pthread_handler_t run_task(void *p);
229
extern "C" pthread_handler_t timer_thread(void *p);
230
231
void statement_cleanup(Statement *stmt);
231
232
void option_cleanup(OptionString *stmt);
232
233
void concurrency_loop(drizzle_con_st *con, uint32_t current, OptionString *eptr);
264
265
user_supplied_query.append(delimiter);
265
266
}
266
267
}
268
269
270
static void run_task(ThreadContext *ctx)
271
{
272
uint64_t counter= 0, queries;
273
uint64_t detach_counter;
274
unsigned int commit_counter;
275
drizzle_con_st con;
276
drizzle_result_st result;
277
drizzle_row_t row;
278
Statement *ptr;
279
280
sleeper_mutex.lock();
281
while (master_wakeup)
282
{
283
sleep_threshhold.wait(sleeper_mutex);
284
}
285
sleeper_mutex.unlock();
286
287
slap_connect(&con, true);
288
289
if (verbose >= 3)
290
printf("connected!\n");
291
queries= 0;
292
293
commit_counter= 0;
294
if (commit_rate)
295
run_query(&con, NULL, "SET AUTOCOMMIT=0", strlen("SET AUTOCOMMIT=0"));
296
297
limit_not_met:
298
for (ptr= ctx->getStmt(), detach_counter= 0;
299
ptr && ptr->getLength();
300
ptr= ptr->getNext(), detach_counter++)
301
{
302
if (!opt_only_print && detach_rate && !(detach_counter % detach_rate))
303
{
304
slap_close(&con);
305
slap_connect(&con, true);
306
}
307
308
/*
309
We have to execute differently based on query type. This should become a function.
310
*/
311
if ((ptr->getType() == UPDATE_TYPE_REQUIRES_PREFIX) ||
312
(ptr->getType() == SELECT_TYPE_REQUIRES_PREFIX))
313
{
314
int length;
315
unsigned int key_val;
316
char buffer[HUGE_STRING_LENGTH];
317
318
/*
319
This should only happen if some sort of new engine was
320
implemented that didn't properly handle UPDATEs.
321
322
Just in case someone runs this under an experimental engine we don't
323
want a crash so the if() is placed here.
324
*/
325
assert(primary_keys.size());
326
if (primary_keys.size())
327
{
328
key_val= (unsigned int)(random() % primary_keys.size());
329
const char *key;
330
key= primary_keys[key_val].c_str();
331
332
assert(key);
333
334
length= snprintf(buffer, HUGE_STRING_LENGTH, "%.*s '%s'",
335
(int)ptr->getLength(), ptr->getString(), key);
336
337
if (run_query(&con, &result, buffer, length))
338
{
339
fprintf(stderr,"%s: Cannot run query %.*s ERROR : %s\n",
340
internal::my_progname, (uint32_t)length, buffer, drizzle_con_error(&con));
341
exit(1);
342
}
343
}
344
}
345
else
346
{
347
if (run_query(&con, &result, ptr->getString(), ptr->getLength()))
348
{
349
fprintf(stderr,"%s: Cannot run query %.*s ERROR : %s\n",
350
internal::my_progname, (uint32_t)ptr->getLength(), ptr->getString(), drizzle_con_error(&con));
351
exit(1);
352
}
353
}
354
355
if (!opt_only_print)
356
{
357
while ((row = drizzle_row_next(&result)))
358
counter++;
359
drizzle_result_free(&result);
360
}
361
queries++;
362
363
if (commit_rate && (++commit_counter == commit_rate))
364
{
365
commit_counter= 0;
366
run_query(&con, NULL, "COMMIT", strlen("COMMIT"));
367
}
368
369
/* If the timer is set, and the alarm is not active then end */
370
if (opt_timer_length && timer_alarm == false)
371
goto end;
372
373
/* If limit has been reached, and we are not in a timer_alarm just end */
374
if (ctx->getLimit() && queries == ctx->getLimit() && timer_alarm == false)
375
goto end;
376
}
377
378
if (opt_timer_length && timer_alarm == true)
379
goto limit_not_met;
380
381
if (ctx->getLimit() && queries < ctx->getLimit())
382
goto limit_not_met;
383
384
385
end:
386
if (commit_rate)
387
run_query(&con, NULL, "COMMIT", strlen("COMMIT"));
388
389
slap_close(&con);
390
391
{
392
boost::mutex::scoped_lock scopedLock(counter_mutex);
393
thread_counter--;
394
count_threshhold.notify_one();
395
}
396
397
delete ctx;
398
}
399
267
400
/**
268
401
* commandline_options is the set of all options that can only be called via the command line.
269
402
525
658
526
659
slap_connect(&con, false);
527
660
528
pthread_mutex_init(&counter_mutex, NULL);
529
pthread_cond_init(&count_threshhold, NULL);
530
pthread_mutex_init(&sleeper_mutex, NULL);
531
pthread_cond_init(&sleep_threshhold, NULL);
532
pthread_mutex_init(&timer_alarm_mutex, NULL);
533
pthread_cond_init(&timer_alarm_threshold, NULL);
534
535
536
661
/* Main iterations loop */
537
662
burnin:
538
663
eptr= engine_options;
566
691
if (opt_burnin)
567
692
goto burnin;
568
693
569
pthread_mutex_destroy(&counter_mutex);
570
pthread_cond_destroy(&count_threshhold);
571
pthread_mutex_destroy(&sleeper_mutex);
572
pthread_cond_destroy(&sleep_threshhold);
573
pthread_mutex_destroy(&timer_alarm_mutex);
574
pthread_cond_destroy(&timer_alarm_threshold);
575
576
694
slap_close(&con);
577
695
578
696
/* now free all the strings we created */
1799
1917
return(0);
1800
1918
}
1801
1919
1920
1921
static void timer_thread()
1922
{
1923
/*
1924
We lock around the initial call in case were we in a loop. This
1925
also keeps the value properly syncronized across call threads.
1926
*/
1927
sleeper_mutex.lock();
1928
while (master_wakeup)
1929
{
1930
sleep_threshhold.wait(sleeper_mutex);
1931
}
1932
sleeper_mutex.unlock();
1933
1934
{
1935
boost::mutex::scoped_lock scopedLock(timer_alarm_mutex);
1936
1937
boost::xtime xt;
1938
xtime_get(&xt, boost::TIME_UTC);
1939
xt.sec += opt_timer_length;
1940
1941
(void)timer_alarm_threshold.timed_wait(scopedLock, xt);
1942
}
1943
1944
{
1945
boost::mutex::scoped_lock scopedLock(timer_alarm_mutex);
1946
timer_alarm= false;
1947
}
1948
}
1949
1802
1950
static int
1803
1951
run_scheduler(Stats *sptr, Statement **stmts, uint32_t concur, uint64_t limit)
1804
1952
{
1805
uint32_t y;
1806
1953
unsigned int real_concurrency;
1807
1954
struct timeval start_time, end_time;
1808
OptionString *sql_type;
1809
pthread_t mainthread; /* Thread descriptor */
1810
pthread_attr_t attr; /* Thread attributes */
1811
1812
1813
pthread_attr_init(&attr);
1814
pthread_attr_setdetachstate(&attr,
1815
PTHREAD_CREATE_DETACHED);
1816
1817
pthread_mutex_lock(&counter_mutex);
1818
thread_counter= 0;
1819
1820
pthread_mutex_lock(&sleeper_mutex);
1821
master_wakeup= 1;
1822
pthread_mutex_unlock(&sleeper_mutex);
1823
1824
real_concurrency= 0;
1825
1826
for (y= 0, sql_type= query_options;
1827
y < query_statements_count;
1828
y++, sql_type= sql_type->getNext())
1829
{
1830
unsigned int options_loop= 1;
1831
1832
if (sql_type->getOption())
1833
{
1834
options_loop= strtol(sql_type->getOption(),
1835
(char **)NULL, 10);
1836
options_loop= options_loop ? options_loop : 1;
1837
}
1838
1839
while (options_loop--)
1840
{
1841
for (uint32_t x= 0; x < concur; x++)
1842
{
1843
ThreadContext *con;
1844
con= new ThreadContext;
1845
if (con == NULL)
1846
{
1847
fprintf(stderr, "Memory Allocation error in scheduler\n");
1848
exit(1);
1849
}
1850
con->setStmt(stmts[y]);
1851
con->setLimit(limit);
1852
1853
real_concurrency++;
1854
/* now you create the thread */
1855
if (pthread_create(&mainthread, &attr, run_task,
1856
(void *)con) != 0)
1857
{
1858
fprintf(stderr,"%s: Could not create thread\n", internal::my_progname);
1859
exit(1);
1860
}
1861
thread_counter++;
1862
}
1863
}
1864
}
1865
1866
/*
1867
The timer_thread belongs to all threads so it too obeys the wakeup
1868
call that run tasks obey.
1869
*/
1870
if (opt_timer_length)
1871
{
1872
pthread_mutex_lock(&timer_alarm_mutex);
1873
timer_alarm= true;
1874
pthread_mutex_unlock(&timer_alarm_mutex);
1875
1876
if (pthread_create(&mainthread, &attr, timer_thread,
1877
(void *)&opt_timer_length) != 0)
1878
{
1879
fprintf(stderr,"%s: Could not create timer thread\n", internal::my_progname);
1880
exit(1);
1881
}
1882
}
1883
1884
pthread_mutex_unlock(&counter_mutex);
1885
pthread_attr_destroy(&attr);
1886
1887
pthread_mutex_lock(&sleeper_mutex);
1888
master_wakeup= 0;
1889
pthread_mutex_unlock(&sleeper_mutex);
1890
pthread_cond_broadcast(&sleep_threshhold);
1955
1956
{
1957
boost::mutex::scoped_lock lock(counter_mutex);
1958
1959
OptionString *sql_type;
1960
thread_counter= 0;
1961
1962
{
1963
boost::mutex::scoped_lock scopedWakeup(sleeper_mutex);
1964
master_wakeup= true;
1965
}
1966
1967
real_concurrency= 0;
1968
1969
uint32_t y;
1970
for (y= 0, sql_type= query_options;
1971
y < query_statements_count;
1972
y++, sql_type= sql_type->getNext())
1973
{
1974
unsigned int options_loop= 1;
1975
1976
if (sql_type->getOption())
1977
{
1978
options_loop= strtol(sql_type->getOption(),
1979
(char **)NULL, 10);
1980
options_loop= options_loop ? options_loop : 1;
1981
}
1982
1983
while (options_loop--)
1984
{
1985
for (uint32_t x= 0; x < concur; x++)
1986
{
1987
ThreadContext *con;
1988
con= new ThreadContext;
1989
if (con == NULL)
1990
{
1991
fprintf(stderr, "Memory Allocation error in scheduler\n");
1992
exit(1);
1993
}
1994
con->setStmt(stmts[y]);
1995
con->setLimit(limit);
1996
1997
real_concurrency++;
1998
1999
/* now you create the thread */
2000
boost::thread new_thread(boost::bind(&run_task, con));
2001
thread_counter++;
2002
}
2003
}
2004
}
2005
2006
/*
2007
The timer_thread belongs to all threads so it too obeys the wakeup
2008
call that run tasks obey.
2009
*/
2010
if (opt_timer_length)
2011
{
2012
{
2013
boost::mutex::scoped_lock alarmLock(timer_alarm_mutex);
2014
timer_alarm= true;
2015
}
2016
2017
boost::thread new_thread(&timer_thread);
2018
}
2019
}
2020
2021
{
2022
boost::mutex::scoped_lock scopedSleeper(sleeper_mutex);
2023
master_wakeup= false;
2024
}
2025
sleep_threshhold.notify_all();
1891
2026
1892
2027
gettimeofday(&start_time, NULL);
1893
2028
1894
2029
/*
1895
2030
We loop until we know that all children have cleaned up.
1896
2031
*/
1897
pthread_mutex_lock(&counter_mutex);
1898
while (thread_counter)
1899
2032
{
1900
struct timespec abstime;
2033
boost::mutex::scoped_lock scopedLock(counter_mutex);
2034
while (thread_counter)
2035
{
2036
boost::xtime xt;
2037
xtime_get(&xt, boost::TIME_UTC);
2038
xt.sec += 3;
1901
2039
1902
set_timespec(abstime, 3);
1903
pthread_cond_timedwait(&count_threshhold, &counter_mutex, &abstime);
2040
count_threshhold.timed_wait(scopedLock, xt);
2041
}
1904
2042
}
1905
pthread_mutex_unlock(&counter_mutex);
1906
2043
1907
2044
gettimeofday(&end_time, NULL);
1908
2045
1909
1910
2046
sptr->setTiming(timedif(end_time, start_time));
1911
2047
sptr->setUsers(concur);
1912
2048
sptr->setRealUsers(real_concurrency);
1915
2051
return(0);
1916
2052
}
1917
2053
1918
1919
pthread_handler_t timer_thread(void *p)
1920
{
1921
uint32_t *timer_length= (uint32_t *)p;
1922
struct timespec abstime;
1923
1924
1925
/*
1926
We lock around the initial call in case were we in a loop. This
1927
also keeps the value properly syncronized across call threads.
1928
*/
1929
pthread_mutex_lock(&sleeper_mutex);
1930
while (master_wakeup)
1931
{
1932
pthread_cond_wait(&sleep_threshhold, &sleeper_mutex);
1933
}
1934
pthread_mutex_unlock(&sleeper_mutex);
1935
1936
set_timespec(abstime, *timer_length);
1937
1938
pthread_mutex_lock(&timer_alarm_mutex);
1939
pthread_cond_timedwait(&timer_alarm_threshold, &timer_alarm_mutex, &abstime);
1940
pthread_mutex_unlock(&timer_alarm_mutex);
1941
1942
pthread_mutex_lock(&timer_alarm_mutex);
1943
timer_alarm= false;
1944
pthread_mutex_unlock(&timer_alarm_mutex);
1945
1946
return(0);
1947
}
1948
1949
pthread_handler_t run_task(void *p)
1950
{
1951
uint64_t counter= 0, queries;
1952
uint64_t detach_counter;
1953
unsigned int commit_counter;
1954
drizzle_con_st con;
1955
drizzle_result_st result;
1956
drizzle_row_t row;
1957
Statement *ptr;
1958
ThreadContext *ctx= (ThreadContext *)p;
1959
1960
pthread_mutex_lock(&sleeper_mutex);
1961
while (master_wakeup)
1962
{
1963
pthread_cond_wait(&sleep_threshhold, &sleeper_mutex);
1964
}
1965
pthread_mutex_unlock(&sleeper_mutex);
1966
1967
slap_connect(&con, true);
1968
1969
if (verbose >= 3)
1970
printf("connected!\n");
1971
queries= 0;
1972
1973
commit_counter= 0;
1974
if (commit_rate)
1975
run_query(&con, NULL, "SET AUTOCOMMIT=0", strlen("SET AUTOCOMMIT=0"));
1976
1977
limit_not_met:
1978
for (ptr= ctx->getStmt(), detach_counter= 0;
1979
ptr && ptr->getLength();
1980
ptr= ptr->getNext(), detach_counter++)
1981
{
1982
if (!opt_only_print && detach_rate && !(detach_counter % detach_rate))
1983
{
1984
slap_close(&con);
1985
slap_connect(&con, true);
1986
}
1987
1988
/*
1989
We have to execute differently based on query type. This should become a function.
1990
*/
1991
if ((ptr->getType() == UPDATE_TYPE_REQUIRES_PREFIX) ||
1992
(ptr->getType() == SELECT_TYPE_REQUIRES_PREFIX))
1993
{
1994
int length;
1995
unsigned int key_val;
1996
char buffer[HUGE_STRING_LENGTH];
1997
1998
/*
1999
This should only happen if some sort of new engine was
2000
implemented that didn't properly handle UPDATEs.
2001
2002
Just in case someone runs this under an experimental engine we don't
2003
want a crash so the if() is placed here.
2004
*/
2005
assert(primary_keys.size());
2006
if (primary_keys.size())
2007
{
2008
key_val= (unsigned int)(random() % primary_keys.size());
2009
const char *key;
2010
key= primary_keys[key_val].c_str();
2011
2012
assert(key);
2013
2014
length= snprintf(buffer, HUGE_STRING_LENGTH, "%.*s '%s'",
2015
(int)ptr->getLength(), ptr->getString(), key);
2016
2017
if (run_query(&con, &result, buffer, length))
2018
{
2019
fprintf(stderr,"%s: Cannot run query %.*s ERROR : %s\n",
2020
internal::my_progname, (uint32_t)length, buffer, drizzle_con_error(&con));
2021
exit(1);
2022
}
2023
}
2024
}
2025
else
2026
{
2027
if (run_query(&con, &result, ptr->getString(), ptr->getLength()))
2028
{
2029
fprintf(stderr,"%s: Cannot run query %.*s ERROR : %s\n",
2030
internal::my_progname, (uint32_t)ptr->getLength(), ptr->getString(), drizzle_con_error(&con));
2031
exit(1);
2032
}
2033
}
2034
2035
if (!opt_only_print)
2036
{
2037
while ((row = drizzle_row_next(&result)))
2038
counter++;
2039
drizzle_result_free(&result);
2040
}
2041
queries++;
2042
2043
if (commit_rate && (++commit_counter == commit_rate))
2044
{
2045
commit_counter= 0;
2046
run_query(&con, NULL, "COMMIT", strlen("COMMIT"));
2047
}
2048
2049
/* If the timer is set, and the alarm is not active then end */
2050
if (opt_timer_length && timer_alarm == false)
2051
goto end;
2052
2053
/* If limit has been reached, and we are not in a timer_alarm just end */
2054
if (ctx->getLimit() && queries == ctx->getLimit() && timer_alarm == false)
2055
goto end;
2056
}
2057
2058
if (opt_timer_length && timer_alarm == true)
2059
goto limit_not_met;
2060
2061
if (ctx->getLimit() && queries < ctx->getLimit())
2062
goto limit_not_met;
2063
2064
2065
end:
2066
if (commit_rate)
2067
run_query(&con, NULL, "COMMIT", strlen("COMMIT"));
2068
2069
slap_close(&con);
2070
2071
pthread_mutex_lock(&counter_mutex);
2072
thread_counter--;
2073
pthread_cond_signal(&count_threshhold);
2074
pthread_mutex_unlock(&counter_mutex);
2075
2076
delete ctx;
2077
2078
return(0);
2079
}
2080
2081
2054
/*
2082
2055
Parse records from comma seperated string. : is a reserved character and is used for options
2083
2056
on variables.
2420
2393
}
2421
2394
}
2422
2395
2423
void
2424
slap_close(drizzle_con_st *con)
2396
void slap_close(drizzle_con_st *con)
2425
2397
{
2426
2398
if (opt_only_print)
2427
2399
return;
2429
2401
drizzle_free(drizzle_con_drizzle(con));
2430
2402
}
2431
2403
2432
void
2433
slap_connect(drizzle_con_st *con, bool connect_to_schema)
2404
void slap_connect(drizzle_con_st *con, bool connect_to_schema)
2434
2405
{
2435
2406
/* Connect to server */
2436
2407
static uint32_t connection_retry_sleep= 100000; /* Microseconds */
2471
2442
ret, drizzle_con_error(con));
2472
2443
exit(1);
2473
2444
}
2474
2475
return;
2476
2445
}
2477
2446
2478
2447
void
Loggerhead is a web-based interface for Breezy
Version: 2.0.1