« back to all changes in this revision
Viewing changes to plugin/pbxt/src/thread_xt.cc
- browse files at revision 1455.3.1
- compare with another revision
- compare with revision 383.1.33
- download diff
- download tarball
- view history from revision 1455.3.1
- Committer: Vladimir Kolesnikov
- Date: 2010-04-08 16:11:29 UTC
- mto: (1457.1.1 bad-staging) (1455.3.13 drizzle-pbxt-6)
- mto: This revision was merged to the branch mainline in revision 1462.
- Revision ID: vladimir@primebase.org-20100408161129-x85uvbu9qavjck66
lp:drizzle + pbxt 1.1 + test results
- files added:
- plugin/pbxt
- plugin/pbxt/bin
- plugin/pbxt/src
- tests/r/pbxt
- files modified:
- drizzled/sql_table.cc
added
removed
plugin/pbxt/src/thread_xt.cc
1
/* Copyright (c) 2005 PrimeBase Technologies GmbH
2
*
3
* PrimeBase XT
4
*
5
* This program is free software; you can redistribute it and/or modify
6
* it under the terms of the GNU General Public License as published by
7
* the Free Software Foundation; either version 2 of the License, or
8
* (at your option) any later version.
9
*
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.
14
*
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., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
18
*
19
* 2005-01-03 Paul McCullagh
20
*
21
* H&G2JCtL
22
*/
23
24
#include "xt_config.h"
25
26
#ifdef DRIZZLED
27
#include <bitset>
28
#endif
29
30
#ifndef XT_WIN
31
#include <unistd.h>
32
#include <sys/time.h>
33
#include <sys/resource.h>
34
#endif
35
#include <time.h>
36
#include <stdarg.h>
37
#include <signal.h>
38
#include <stdlib.h>
39
#include <ctype.h>
40
#include <errno.h>
41
42
#include "xt_defs.h"
43
#include "strutil_xt.h"
44
#include "pthread_xt.h"
45
#include "thread_xt.h"
46
#include "memory_xt.h"
47
#include "sortedlist_xt.h"
48
#include "trace_xt.h"
49
#include "myxt_xt.h"
50
#include "database_xt.h"
51
52
#ifdef DEBUG
53
#define DEBUG_NO_ACTIVITY
54
#endif
55
56
void xt_db_exit_thread(XTThreadPtr self);
57
58
static void thr_accumulate_statistics(XTThreadPtr self);
59
60
#define XT_NO_THREAD_ID 0
61
#define XT_SYS_THREAD_ID 1
62
#define XT_MIN_THREAD_ID 2
63
64
/*
65
* -----------------------------------------------------------------------
66
* THREAD GLOBALS
67
*/
68
69
xtPublic u_int xt_thr_maximum_threads;
70
xtPublic u_int xt_thr_current_thread_count;
71
xtPublic u_int xt_thr_current_max_threads;
72
73
/* This structure is a double linked list of thread, with a wait
74
* condition on it.
75
*/
76
static XTLinkedListPtr thr_list;
77
78
/* This structure maps thread ID's to thread pointers. */
79
THR_ARRAY_LOCK_TYPE xt_thr_array_resize_lock;
80
xtPublic XTThreadDataRec *xt_thr_array;
81
static xt_mutex_type thr_array_lock;
82
83
/* Global accumulated statistics: */
84
static XTStatisticsRec thr_statistics;
85
86
#ifdef DEBUG
87
static void break_in_assertion(c_char *expr, c_char *func, c_char *file, u_int line)
88
{
89
printf("%s(%s:%d) %s\n", func, file, (int) line, expr);
90
}
91
#endif
92
93
/*
94
* -----------------------------------------------------------------------
95
* Lock task
96
*/
97
98
void XTLockTask::tk_init(struct XTThread *self)
99
{
100
xt_init_mutex_with_autoname(self, <_mutex);
101
}
102
103
void XTLockTask::tk_exit()
104
{
105
xt_free_mutex(<_mutex);
106
XTTask::tk_exit();
107
}
108
109
void XTLockTask::tk_lock()
110
{
111
xt_lock_mutex_ns(<_mutex);
112
}
113
114
void XTLockTask::tk_unlock()
115
{
116
xt_unlock_mutex_ns(<_mutex);
117
}
118
119
/*
120
* -----------------------------------------------------------------------
121
* Error logging
122
*/
123
124
static xt_mutex_type log_mutex;
125
static int log_level = 0;
126
static FILE *log_file = NULL;
127
static xtBool log_newline = TRUE;
128
129
xtPublic xtBool xt_init_logging(void)
130
{
131
int err;
132
133
log_file = stdout;
134
log_level = XT_LOG_TRACE;
135
err = xt_p_mutex_init_with_autoname(&log_mutex, NULL);
136
if (err) {
137
xt_log_errno(XT_NS_CONTEXT, err);
138
log_file = NULL;
139
log_level = 0;
140
return FALSE;
141
}
142
if (!xt_init_trace()) {
143
xt_exit_logging();
144
return FALSE;
145
}
146
return TRUE;
147
}
148
149
xtPublic void xt_exit_logging(void)
150
{
151
if (log_file) {
152
xt_free_mutex(&log_mutex);
153
log_file = NULL;
154
}
155
xt_exit_trace();
156
}
157
158
xtPublic void xt_get_now(char *buffer, size_t len)
159
{
160
time_t ticks;
161
struct tm ltime;
162
163
ticks = time(NULL);
164
if (ticks == (time_t) -1) {
165
#ifdef XT_WIN
166
printf(buffer, "** error %d getting time **", errno);
167
#else
168
snprintf(buffer, len, "** error %d getting time **", errno);
169
#endif
170
return;
171
}
172
localtime_r(&ticks, <ime);
173
strftime(buffer, len, "%y%m%d %H:%M:%S", <ime);
174
}
175
176
static void thr_log_newline(XTThreadPtr self, c_char *func, c_char *file, u_int line, int level)
177
{
178
c_char *level_str;
179
char time_str[200];
180
char thr_name[XT_THR_NAME_SIZE+3];
181
182
xt_get_now(time_str, 200);
183
if (self && *self->t_name) {
184
xt_strcpy(XT_THR_NAME_SIZE+3, thr_name, " ");
185
xt_strcat(XT_THR_NAME_SIZE+3, thr_name, self->t_name);
186
}
187
else
188
thr_name[0] = 0;
189
switch (level) {
190
case XT_LOG_FATAL: level_str = " [Fatal]"; break;
191
case XT_LOG_ERROR: level_str = " [Error]"; break;
192
case XT_LOG_WARNING: level_str = " [Warning]"; break;
193
case XT_LOG_INFO: level_str = " [Note]"; break;
194
case XT_LOG_TRACE: level_str = " [Trace]"; break;
195
default: level_str = " "; break;
196
}
197
if (func && *func && *func != '-') {
198
char func_name[XT_MAX_FUNC_NAME_SIZE];
199
200
xt_strcpy_term(XT_MAX_FUNC_NAME_SIZE, func_name, func, '(');
201
if (file && *file)
202
fprintf(log_file, "%s%s%s %s(%s:%d) ", time_str, level_str, thr_name, func_name, xt_last_name_of_path(file), line);
203
else
204
fprintf(log_file, "%s%s%s %s() ", time_str, level_str, thr_name, func_name);
205
}
206
else {
207
if (file && *file)
208
fprintf(log_file, "%s%s%s [%s:%d] ", time_str, level_str, thr_name, xt_last_name_of_path(file), line);
209
else
210
fprintf(log_file, "%s%s%s ", time_str, level_str, thr_name);
211
}
212
}
213
214
#ifdef XT_WIN
215
/* Windows uses printf()!! */
216
#define DEFAULT_LOG_BUFFER_SIZE 2000
217
#else
218
#ifdef DEBUG
219
#define DEFAULT_LOG_BUFFER_SIZE 10
220
#else
221
#define DEFAULT_LOG_BUFFER_SIZE 2000
222
#endif
223
#endif
224
225
void xt_log_flush(XTThreadPtr XT_UNUSED(self))
226
{
227
fflush(log_file);
228
}
229
230
/*
231
* Log the given formated string information to the log file.
232
* Before each new line, this function writes the
233
* log header, which includes the time, log level,
234
* and source file and line number (optional).
235
*/
236
static void thr_log_va(XTThreadPtr self, c_char *func, c_char *file, u_int line, int level, c_char *fmt, va_list ap)
237
{
238
char buffer[DEFAULT_LOG_BUFFER_SIZE];
239
char *log_string = NULL;
240
241
if (level > log_level)
242
return;
243
244
xt_lock_mutex_ns(&log_mutex);
245
246
#ifdef XT_WIN
247
vsprintf(buffer, fmt, ap);
248
log_string = buffer;
249
#else
250
#if !defined(va_copy) || defined(XT_SOLARIS)
251
int len;
252
253
len = vsnprintf(buffer, DEFAULT_LOG_BUFFER_SIZE-1, fmt, ap);
254
if (len > DEFAULT_LOG_BUFFER_SIZE-1)
255
len = DEFAULT_LOG_BUFFER_SIZE-1;
256
buffer[len] = 0;
257
log_string = buffer;
258
#else
259
/* Use the buffer, unless it is too small */
260
va_list ap2;
261
262
va_copy(ap2, ap);
263
if (vsnprintf(buffer, DEFAULT_LOG_BUFFER_SIZE, fmt, ap) >= DEFAULT_LOG_BUFFER_SIZE) {
264
if (vasprintf(&log_string, fmt, ap2) == -1)
265
log_string = NULL;
266
}
267
else
268
log_string = buffer;
269
#endif
270
#endif
271
272
if (log_string) {
273
char *str, *str_end, tmp_ch;
274
275
str = log_string;
276
while (*str) {
277
if (log_newline) {
278
thr_log_newline(self, func, file, line, level);
279
log_newline = FALSE;
280
}
281
str_end = strchr(str, '\n');
282
if (str_end) {
283
str_end++;
284
tmp_ch = *str_end;
285
*str_end = 0;
286
log_newline = TRUE;
287
}
288
else {
289
str_end = str + strlen(str);
290
tmp_ch = 0;
291
}
292
fprintf(log_file, "%s", str);
293
fflush(log_file);
294
*str_end = tmp_ch;
295
str = str_end;
296
}
297
298
if (log_string != buffer)
299
free(log_string);
300
}
301
302
xt_unlock_mutex_ns(&log_mutex);
303
}
304
305
xtPublic void xt_logf(XTThreadPtr self, c_char *func, c_char *file, u_int line, int level, c_char *fmt, ...)
306
{
307
va_list ap;
308
309
va_start(ap, fmt);
310
thr_log_va(self, func, file, line, level, fmt, ap);
311
va_end(ap);
312
}
313
314
xtPublic void xt_log(XTThreadPtr self, c_char *func, c_char *file, u_int line, int level, c_char *string)
315
{
316
xt_logf(self, func, file, line, level, "%s", string);
317
}
318
319
static int thr_log_error_va(XTThreadPtr self, c_char *func, c_char *file, u_int line, int level, int xt_err, int sys_err, c_char *fmt, va_list ap)
320
{
321
int default_level;
322
char xt_err_string[50];
323
324
*xt_err_string = 0;
325
switch (xt_err) {
326
case XT_ASSERTION_FAILURE:
327
strcpy(xt_err_string, "Assertion");
328
default_level = XT_LOG_FATAL;
329
break;
330
case XT_SYSTEM_ERROR:
331
strcpy(xt_err_string, "errno");
332
default_level = XT_LOG_ERROR;
333
break;
334
case XT_SIGNAL_CAUGHT:
335
strcpy(xt_err_string, "Signal");
336
default_level = XT_LOG_ERROR;
337
break;
338
default:
339
sprintf(xt_err_string, "%d", xt_err);
340
default_level = XT_LOG_ERROR;
341
break;
342
}
343
if (level == XT_LOG_DEFAULT)
344
level = default_level;
345
346
if (*xt_err_string) {
347
if (sys_err)
348
xt_logf(self, func, file, line, level, "%s (%d): ", xt_err_string, sys_err);
349
else
350
xt_logf(self, func, file, line, level, "%s: ", xt_err_string);
351
}
352
thr_log_va(self, func, file, line, level, fmt, ap);
353
xt_logf(self, func, file, line, level, "\n");
354
return level;
355
}
356
357
/* The function returns the actual log level used. */
358
xtPublic int xt_log_errorf(XTThreadPtr self, c_char *func, c_char *file, u_int line, int level, int xt_err, int sys_err, c_char *fmt, ...)
359
{
360
va_list ap;
361
362
va_start(ap, fmt);
363
level = thr_log_error_va(self, func, file, line, level, xt_err, sys_err, fmt, ap);
364
va_end(ap);
365
return level;
366
}
367
368
/* The function returns the actual log level used. */
369
xtPublic int xt_log_error(XTThreadPtr self, c_char *func, c_char *file, u_int line, int level, int xt_err, int sys_err, c_char *string)
370
{
371
return xt_log_errorf(self, func, file, line, level, xt_err, sys_err, "%s", string);
372
}
373
374
xtPublic void xt_log_exception(XTThreadPtr self, XTExceptionPtr e, int level)
375
{
376
ASSERT_NS(e->e_xt_err);
377
level = xt_log_error(
378
self,
379
e->e_func_name,
380
e->e_source_file,
381
e->e_source_line,
382
level,
383
e->e_xt_err,
384
e->e_sys_err,
385
e->e_err_msg);
386
/* Dump the catch trace: */
387
if (*e->e_catch_trace)
388
xt_logf(self, NULL, NULL, 0, level, "%s", e->e_catch_trace);
389
}
390
391
xtPublic void xt_log_and_clear_exception(XTThreadPtr self)
392
{
393
xt_log_exception(self, &self->t_exception, XT_LOG_DEFAULT);
394
xt_clear_exception(self);
395
}
396
397
xtPublic void xt_log_and_clear_exception_ns(void)
398
{
399
xt_log_and_clear_exception(xt_get_self());
400
}
401
402
xtPublic void xt_log_and_clear_warning(XTThreadPtr self)
403
{
404
xt_log_exception(self, &self->t_exception, XT_LOG_WARNING);
405
xt_clear_exception(self);
406
}
407
408
xtPublic void xt_log_and_clear_warning_ns(void)
409
{
410
xt_log_and_clear_warning(xt_get_self());
411
}
412
413
/*
414
* -----------------------------------------------------------------------
415
* Exceptions
416
*/
417
418
static void thr_add_catch_trace(XTExceptionPtr e, c_char *func, c_char *file, u_int line)
419
{
420
if (func && *func && *func != '-') {
421
xt_strcat_term(XT_CATCH_TRACE_SIZE, e->e_catch_trace, func, '(');
422
xt_strcat(XT_CATCH_TRACE_SIZE, e->e_catch_trace, "(");
423
}
424
if (file && *file) {
425
xt_strcat(XT_CATCH_TRACE_SIZE, e->e_catch_trace, xt_last_name_of_path(file));
426
if (line) {
427
char buffer[40];
428
429
sprintf(buffer, "%u", line);
430
xt_strcat(XT_CATCH_TRACE_SIZE, e->e_catch_trace, ":");
431
xt_strcat(XT_CATCH_TRACE_SIZE, e->e_catch_trace, buffer);
432
}
433
}
434
if (func && *func && *func != '-')
435
xt_strcat(XT_CATCH_TRACE_SIZE, e->e_catch_trace, ")");
436
xt_strcat(XT_CATCH_TRACE_SIZE, e->e_catch_trace, "\n");
437
}
438
439
static void thr_save_error_va(XTExceptionPtr e, XTThreadPtr self, xtBool throw_it, c_char *func, c_char *file, u_int line, int xt_err, int sys_err, c_char *fmt, va_list ap)
440
{
441
int i;
442
443
if (!e)
444
return;
445
446
e->e_xt_err = xt_err;
447
e->e_sys_err = sys_err;
448
vsnprintf(e->e_err_msg, XT_ERR_MSG_SIZE, fmt, ap);
449
450
/* Make the first character of the message upper case: */
451
/* This did not work for foreign languages! */
452
if (e->e_err_msg[0] >= 'a' && e->e_err_msg[0] <= 'z')
453
e->e_err_msg[0] = (char) toupper(e->e_err_msg[0]);
454
455
if (func && *func && *func != '-')
456
xt_strcpy_term(XT_MAX_FUNC_NAME_SIZE, e->e_func_name, func, '(');
457
else
458
*e->e_func_name = 0;
459
if (file && *file) {
460
xt_strcpy(XT_SOURCE_FILE_NAME_SIZE, e->e_source_file, xt_last_name_of_path(file));
461
e->e_source_line = line;
462
}
463
else {
464
*e->e_source_file = 0;
465
e->e_source_line = 0;
466
}
467
*e->e_catch_trace = 0;
468
469
if (!self)
470
return;
471
472
/* Create a stack trace for this exception: */
473
thr_add_catch_trace(e, func, file, line);
474
for (i=self->t_call_top-1; i>=0; i--)
475
thr_add_catch_trace(e, self->t_call_stack[i].cs_func, self->t_call_stack[i].cs_file, self->t_call_stack[i].cs_line);
476
477
if (throw_it)
478
xt_throw(self);
479
}
480
481
/*
482
* -----------------------------------------------------------------------
483
* THROWING EXCEPTIONS
484
*/
485
486
/* If we have to allocate resources and the hold them temporarily during which
487
* time an exception could occur, then these functions provide a holding
488
* place for the data, which will be freed in the case of an exception.
489
*
490
* Note: the free functions could themselves allocated resources.
491
* to make sure all things work out we only remove the resource from
492
* then stack when it is freed.
493
*/
494
static void thr_free_resources(XTThreadPtr self, XTResourcePtr top)
495
{
496
XTResourcePtr rp;
497
XTThreadFreeFunc free_func;
498
499
if (!top)
500
return;
501
while (self->t_res_top > top) {
502
/* Pop the top resource: */
503
rp = (XTResourcePtr) (((char *) self->t_res_top) - self->t_res_top->r_prev_size);
504
505
/* Free the resource: */
506
if (rp->r_free_func) {
507
free_func = rp->r_free_func;
508
rp->r_free_func = NULL;
509
free_func(self, rp->r_data);
510
}
511
512
self->t_res_top = rp;
513
}
514
}
515
516
xtPublic void xt_bug(XTThreadPtr XT_UNUSED(self))
517
{
518
static int *bug_ptr = NULL;
519
520
bug_ptr = NULL;
521
}
522
523
/*
524
* This function is called when an exception is caught.
525
* It restores the function call top and frees
526
* any resource allocated by lower levels.
527
*/
528
xtPublic void xt_caught(XTThreadPtr self)
529
{
530
/* Restore the call top: */
531
self->t_call_top = self->t_jmp_env[self->t_jmp_depth].jb_call_top;
532
533
/* Free the temporary data that would otherwize be lost
534
* This should do nothing, because we actually free things on throw
535
* (se below).
536
*/
537
thr_free_resources(self, self->t_jmp_env[self->t_jmp_depth].jb_res_top);
538
}
539
540
xtPublic void xt_enter_exception_handler(XTThreadPtr self, XTExceptionPtr e)
541
{
542
ASSERT_NS(self->t_exception.e_xt_err);
543
self->t_in_handler++;
544
*e = self->t_exception;
545
/* If freeing resources, generates an error, we will know: */
546
self->t_exception.e_xt_err = 0;
547
}
548
549
xtPublic void xt_exit_exception_handler(XTThreadPtr self, XTExceptionPtr e)
550
{
551
if (self->t_exception.e_xt_err)
552
xt_log_and_clear_exception(self);
553
/* Restore the exception: */
554
self->t_exception = *e;
555
self->t_in_handler--;
556
}
557
558
/* Throw an already registered error: */
559
xtPublic void xt_throw(XTThreadPtr self)
560
{
561
if (self) {
562
/* We should not throw an exception in an exception handler! */
563
ASSERT_NS(!self->t_in_handler);
564
565
/* We cannot actually handle the situation that an exception is
566
* thrown when freeing resources here!
567
*/
568
if (self->t_jmp_depth > 0 && self->t_jmp_depth <= XT_MAX_JMP) {
569
/* Save the exception, in case it is overwritten during free
570
* resources!
571
*/
572
XTException e;
573
574
/* As recommended by Barry: free the resources before the stack is invalid! */
575
xt_enter_exception_handler(self, &e);
576
thr_free_resources(self, self->t_jmp_env[self->t_jmp_depth-1].jb_res_top);
577
xt_exit_exception_handler(self, &e);
578
579
/* Then do the longjmp: */
580
if (!self->t_in_handler)
581
longjmp(self->t_jmp_env[self->t_jmp_depth-1].jb_buffer, 1);
582
}
583
}
584
585
/*
586
* We cannot throw an error, because it will not be caught.
587
* This means there is no try ... catch block above.
588
* In this case, we just return.
589
* The calling functions must handle errors...
590
xt_caught(self);
591
xt_log(XT_CONTEXT, XT_LOG_FATAL, "Uncaught exception\n");
592
xt_exit_thread(self, NULL);
593
*/
594
}
595
596
xtPublic void xt_throwf(XTThreadPtr self, c_char *func, c_char *file, u_int line, int xt_err, int sys_err, c_char *fmt, ...)
597
{
598
va_list ap;
599
XTThreadPtr thread = self ? self : xt_get_self();
600
601
va_start(ap, fmt);
602
thr_save_error_va(thread ? &thread->t_exception : NULL, thread, self ? TRUE : FALSE, func, file, line, xt_err, sys_err, fmt, ap);
603
va_end(ap);
604
}
605
606
xtPublic void xt_throw_error(XTThreadPtr self, c_char *func, c_char *file, u_int line, int xt_err, int sys_err, c_char *msg)
607
{
608
xt_throwf(self, func, file, line, xt_err, sys_err, "%s", msg);
609
}
610
611
#define XT_SYS_ERR_SIZE 300
612
613
#ifdef XT_WIN
614
static c_char *thr_get_sys_error(int err, char *err_msg)
615
#else
616
static c_char *thr_get_sys_error(int err, char *XT_UNUSED(err_msg))
617
#endif
618
{
619
#ifdef XT_WIN
620
char *ptr;
621
622
if (!FormatMessage(FORMAT_MESSAGE_FROM_SYSTEM, NULL,
623
err, 0, err_msg, XT_SYS_ERR_SIZE, NULL)) {
624
return strerror(err);
625
}
626
627
ptr = &err_msg[strlen(err_msg)];
628
while (ptr-1 > err_msg) {
629
if (*(ptr-1) != '\n' && *(ptr-1) != '\r' && *(ptr-1) != '.')
630
break;
631
ptr--;
632
}
633
*ptr = 0;
634
return err_msg;
635
#else
636
return strerror(err);
637
#endif
638
}
639
640
static c_char *thr_get_err_string(int xt_err)
641
{
642
c_char *str;
643
644
switch (xt_err) {
645
case XT_ERR_STACK_OVERFLOW: str = "Stack overflow"; break;
646
case XT_ERR_JUMP_OVERFLOW: str = "Jump overflow"; break;
647
case XT_ERR_TABLE_EXISTS: str = "Table `%s` already exists"; break;
648
case XT_ERR_NAME_TOO_LONG: str = "Name '%s' is too long"; break;
649
case XT_ERR_TABLE_NOT_FOUND: str = "Table `%s` not found"; break;
650
case XT_ERR_SESSION_NOT_FOUND: str = "Session %s not found"; break;
651
case XT_ERR_BAD_ADDRESS: str = "Incorrect address '%s'"; break;
652
case XT_ERR_UNKNOWN_SERVICE: str = "Unknown service '%s'"; break;
653
case XT_ERR_UNKNOWN_HOST: str = "Host '%s' not found"; break;
654
case XT_ERR_TOKEN_EXPECTED: str = "%s expected in place of %s"; break;
655
case XT_ERR_PROPERTY_REQUIRED: str = "Property '%s' required"; break;
656
case XT_ERR_DEADLOCK: str = "Deadlock, transaction aborted"; break;
657
case XT_ERR_CANNOT_CHANGE_DB: str = "Cannot change database while transaction is in progress"; break;
658
case XT_ERR_ILLEGAL_CHAR: str = "Illegal character: '%s'"; break;
659
case XT_ERR_UNTERMINATED_STRING:str = "Unterminated string: %s"; break;
660
case XT_ERR_SYNTAX: str = "Syntax error near %s"; break;
661
case XT_ERR_ILLEGAL_INSTRUCTION:str = "Illegal instruction"; break;
662
case XT_ERR_OUT_OF_BOUNDS: str = "Memory reference out of bounds"; break;
663
case XT_ERR_STACK_UNDERFLOW: str = "Stack underflow"; break;
664
case XT_ERR_TYPE_MISMATCH: str = "Type mismatch"; break;
665
case XT_ERR_ILLEGAL_TYPE: str = "Illegal type for operator"; break;
666
case XT_ERR_ID_TOO_LONG: str = "Identifier too long: %s"; break;
667
case XT_ERR_TYPE_OVERFLOW: str = "Type overflow: %s"; break;
668
case XT_ERR_TABLE_IN_USE: str = "Table `%s` in use"; break;
669
case XT_ERR_NO_DATABASE_IN_USE: str = "No database in use"; break;
670
case XT_ERR_CANNOT_RESOLVE_TYPE:str = "Cannot resolve type with ID: %s"; break;
671
case XT_ERR_BAD_INDEX_DESC: str = "Unsupported index description: %s"; break;
672
case XT_ERR_WRONG_NO_OF_VALUES: str = "Incorrect number of values"; break;
673
case XT_ERR_CANNOT_OUTPUT_VALUE:str = "Cannot output given type"; break;
674
case XT_ERR_COLUMN_NOT_FOUND: str = "Column `%s.%s` not found"; break;
675
case XT_ERR_NOT_IMPLEMENTED: str = "Not implemented"; break;
676
case XT_ERR_UNEXPECTED_EOS: str = "Connection unexpectedly lost"; break;
677
case XT_ERR_BAD_TOKEN: str = "Incorrect binary token"; break;
678
case XT_ERR_RES_STACK_OVERFLOW: str = "Internal error: resource stack overflow"; break;
679
case XT_ERR_BAD_INDEX_TYPE: str = "Unsupported index type: %s"; break;
680
case XT_ERR_INDEX_EXISTS: str = "Index '%s' already exists"; break;
681
case XT_ERR_INDEX_STRUC_EXISTS: str = "Index '%s' has an identical structure"; break;
682
case XT_ERR_INDEX_NOT_FOUND: str = "Index '%s' not found"; break;
683
case XT_ERR_INDEX_CORRUPT: str = "Cannot read index '%s'"; break;
684
case XT_ERR_TYPE_NOT_SUPPORTED: str = "Data type %s not supported"; break;
685
case XT_ERR_BAD_TABLE_VERSION: str = "Table `%s` version not supported, upgrade required"; break;
686
case XT_ERR_BAD_RECORD_FORMAT: str = "Record format unknown, either corrupted or upgrade required"; break;
687
case XT_ERR_BAD_EXT_RECORD: str = "Extended record part does not match reference"; break;
688
case XT_ERR_RECORD_CHANGED: str = "Record already updated, transaction aborted"; break;
689
case XT_ERR_XLOG_WAS_CORRUPTED: str = "Corrupted transaction log has been truncated"; break;
690
case XT_ERR_DUPLICATE_KEY: str = "Duplicate unique key"; break;
691
case XT_ERR_NO_DICTIONARY: str = "Table `%s` has not yet been opened by MySQL"; break;
692
case XT_ERR_TOO_MANY_TABLES: str = "Limit of %s tables per database exceeded"; break;
693
case XT_ERR_KEY_TOO_LARGE: str = "Index '%s' exceeds the key size limit of %s"; break;
694
case XT_ERR_MULTIPLE_DATABASES: str = "Multiple database in a single transaction is not permitted"; break;
695
case XT_ERR_NO_TRANSACTION: str = "Internal error: no transaction running"; break;
696
case XT_ERR_A_EXPECTED_NOT_B: str = "%s expected in place of %s"; break;
697
case XT_ERR_NO_MATCHING_INDEX: str = "Matching index required for '%s'"; break;
698
case XT_ERR_TABLE_LOCKED: str = "Table `%s` locked"; break;
699
case XT_ERR_NO_REFERENCED_ROW: str = "Constraint: `%s`"; break; // "Foreign key '%s', referenced row does not exist"
700
case XT_ERR_ROW_IS_REFERENCED: str = "Constraint: `%s`"; break; // "Foreign key '%s', has a referencing row"
701
case XT_ERR_BAD_DICTIONARY: str = "Internal dictionary does not match MySQL dictionary"; break;
702
case XT_ERR_LOADING_MYSQL_DIC: str = "Error loading %s.frm file, MySQL error: %s"; break;
703
case XT_ERR_COLUMN_IS_NOT_NULL: str = "Column `%s` is NOT NULL"; break;
704
case XT_ERR_INCORRECT_NO_OF_COLS: str = "Incorrect number of columns near %s"; break;
705
case XT_ERR_FK_ON_TEMP_TABLE: str = "Cannot create foreign key on temporary table"; break;
706
case XT_ERR_REF_TABLE_NOT_FOUND: str = "Referenced table `%s` not found"; break;
707
case XT_ERR_REF_TYPE_WRONG: str = "Incorrect data type on referenced column `%s`"; break;
708
case XT_ERR_DUPLICATE_FKEY: str = "Duplicate unique foreign key, contraint: %s"; break;
709
case XT_ERR_INDEX_FILE_TO_LARGE: str = "Index file has grown too large: %s"; break;
710
case XT_ERR_UPGRADE_TABLE: str = "Table `%s` must be upgraded from PBXT version %s"; break;
711
case XT_ERR_INDEX_NEW_VERSION: str = "Table `%s` index created by a newer version, upgrade required"; break;
712
case XT_ERR_LOCK_TIMEOUT: str = "Lock timeout on table `%s`"; break;
713
case XT_ERR_CONVERSION: str = "Error converting value for column `%s.%s`"; break;
714
case XT_ERR_NO_ROWS: str = "No matching row found in table `%s`"; break;
715
case XT_ERR_DATA_LOG_NOT_FOUND: str = "Data log not found: '%s'"; break;
716
case XT_ERR_LOG_MAX_EXCEEDED: str = "Maximum log count, %s, exceeded"; break;
717
case XT_ERR_MAX_ROW_COUNT: str = "Maximum row count reached"; break;
718
case XT_ERR_FILE_TOO_LONG: str = "File cannot be mapped, too large: '%s'"; break;
719
case XT_ERR_BAD_IND_BLOCK_SIZE: str = "Table `%s`, incorrect index block size: %s"; break;
720
case XT_ERR_INDEX_CORRUPTED: str = "Table `%s` index is corrupted, REPAIR TABLE required"; break;
721
case XT_ERR_NO_INDEX_CACHE: str = "Not enough index cache memory to handle concurrent updates"; break;
722
case XT_ERR_INDEX_LOG_CORRUPT: str = "Index log corrupt: '%s'"; break;
723
case XT_ERR_TOO_MANY_THREADS: str = "Too many threads: %s, increase pbxt_max_threads"; break;
724
case XT_ERR_TOO_MANY_WAITERS: str = "Too many waiting threads: %s"; break;
725
case XT_ERR_INDEX_OLD_VERSION: str = "Table `%s` index created by an older version, REPAIR TABLE required"; break;
726
case XT_ERR_PBXT_TABLE_EXISTS: str = "System table cannot be dropped because PBXT table still exists"; break;
727
case XT_ERR_SERVER_RUNNING: str = "A server is possibly already running"; break;
728
case XT_ERR_INDEX_MISSING: str = "Index file of table '%s' is missing"; break;
729
case XT_ERR_RECORD_DELETED: str = "Record was deleted"; break;
730
case XT_ERR_NEW_TYPE_OF_XLOG: str = "Transaction log %s, is using a newer format, upgrade required"; break;
731
case XT_ERR_NO_BEFORE_IMAGE: str = "Internal error: no before image"; break;
732
case XT_ERR_FK_REF_TEMP_TABLE: str = "Foreign key may not reference temporary table"; break;
733
case XT_ERR_FILE_OP_NOT_SUPP: str = "File operation not supported on this file: %s"; break;
734
case XT_ERR_INDEX_NOT_RECOVERED: str = "Index of table '%s' cannot be used because the table was not recovered"; break;
735
case XT_ERR_MYSQL_SHUTDOWN: str = "Cannot open table, MySQL has shutdown"; break;
736
case XT_ERR_MYSQL_NO_THREAD: str = "Cannot create thread, MySQL has shutdown"; break;
737
case XT_ERR_BUFFER_TOO_SMALL: str = "System backup buffer too small"; break;
738
case XT_ERR_BAD_BACKUP_FORMAT: str = "Unknown or corrupt backup format, restore aborted"; break;
739
case XT_ERR_PBXT_NOT_INSTALLED: str = "PBXT plugin is not installed"; break;
740
case XT_ERR_LOG_HEADER_CORRUPT: str = "Transaction log %s, file header corrupt"; break;
741
default: str = "Unknown XT error"; break;
742
}
743
return str;
744
}
745
746
xtPublic void xt_throw_i2xterr(XTThreadPtr self, c_char *func, c_char *file, u_int line, int xt_err, c_char *item, c_char *item2)
747
{
748
xt_throwf(self, func, file, line, xt_err, 0, thr_get_err_string(xt_err), item, item2);
749
}
750
751
xtPublic void xt_throw_ixterr(XTThreadPtr self, c_char *func, c_char *file, u_int line, int xt_err, c_char *item)
752
{
753
xt_throw_i2xterr(self, func, file, line, xt_err, item, NULL);
754
}
755
756
xtPublic void xt_throw_tabcolerr(XTThreadPtr self, c_char *func, c_char *file, u_int line, int xt_err, XTPathStrPtr tab_item, c_char *item2)
757
{
758
char buffer[XT_TABLE_NAME_BUF_SIZE];
759
760
xt_tab_make_table_name(tab_item, buffer, sizeof(buffer));
761
xt_throw_i2xterr(self, func, file, line, xt_err, buffer, item2);
762
}
763
764
xtPublic void xt_throw_taberr(XTThreadPtr self, c_char *func, c_char *file, u_int line, int xt_err, XTPathStrPtr tab_item)
765
{
766
char buffer[XT_TABLE_NAME_BUF_SIZE];
767
768
xt_tab_make_table_name(tab_item, buffer, sizeof(buffer));
769
xt_throw_ixterr(self, func, file, line, xt_err, buffer);
770
}
771
772
xtPublic void xt_throw_ulxterr(XTThreadPtr self, c_char *func, c_char *file, u_int line, int xt_err, u_long value)
773
{
774
char buffer[100];
775
776
sprintf(buffer, "%lu", value);
777
xt_throw_ixterr(self, func, file, line, xt_err, buffer);
778
}
779
780
xtPublic void xt_throw_sulxterr(XTThreadPtr self, c_char *func, c_char *file, u_int line, int xt_err, c_char *item, u_long value)
781
{
782
char buffer[100];
783
784
sprintf(buffer, "%lu", value);
785
xt_throw_i2xterr(self, func, file, line, xt_err, item, buffer);
786
}
787
788
xtPublic void xt_throw_xterr(XTThreadPtr self, c_char *func, c_char *file, u_int line, int xt_err)
789
{
790
xt_throw_ixterr(self, func, file, line, xt_err, NULL);
791
}
792
793
xtPublic void xt_throw_errno(XTThreadPtr self, c_char *func, c_char *file, u_int line, int err)
794
{
795
char err_msg[XT_SYS_ERR_SIZE];
796
797
xt_throw_error(self, func, file, line, XT_SYSTEM_ERROR, err, thr_get_sys_error(err, err_msg));
798
}
799
800
xtPublic void xt_throw_ferrno(XTThreadPtr self, c_char *func, c_char *file, u_int line, int err, c_char *path)
801
{
802
char err_msg[XT_SYS_ERR_SIZE];
803
804
xt_throwf(self, func, file, line, XT_SYSTEM_ERROR, err, "%s: '%s'", thr_get_sys_error(err, err_msg), path);
805
}
806
807
xtPublic void xt_throw_assertion(XTThreadPtr self, c_char *func, c_char *file, u_int line, c_char *str)
808
{
809
xt_throw_error(self, func, file, line, XT_ASSERTION_FAILURE, 0, str);
810
}
811
812
static void xt_log_assertion(XTThreadPtr self, c_char *func, c_char *file, u_int line, c_char *str)
813
{
814
xt_log_error(self, func, file, line, XT_LOG_DEFAULT, XT_ASSERTION_FAILURE, 0, str);
815
}
816
817
xtPublic void xt_throw_signal(XTThreadPtr self, c_char *func, c_char *file, u_int line, int sig)
818
{
819
#ifdef XT_WIN
820
char buffer[100];
821
822
sprintf(buffer, "Signal #%d", sig);
823
xt_throw_error(self, func, file, line, XT_SIGNAL_CAUGHT, sig, buffer);
824
#else
825
xt_throw_error(self, func, file, line, XT_SIGNAL_CAUGHT, sig, strsignal(sig));
826
#endif
827
}
828
829
/*
830
* -----------------------------------------------------------------------
831
* REGISTERING EXCEPTIONS
832
*/
833
834
xtPublic void xt_registerf(c_char *func, c_char *file, u_int line, int xt_err, int sys_err, c_char *fmt, ...)
835
{
836
va_list ap;
837
XTThreadPtr thread = xt_get_self();
838
839
va_start(ap, fmt);
840
thr_save_error_va(thread ? &thread->t_exception : NULL, thread, FALSE, func, file, line, xt_err, sys_err, fmt, ap);
841
va_end(ap);
842
}
843
844
xtPublic void xt_register_i2xterr(c_char *func, c_char *file, u_int line, int xt_err, c_char *item, c_char *item2)
845
{
846
xt_registerf(func, file, line, xt_err, 0, thr_get_err_string(xt_err), item, item2);
847
}
848
849
xtPublic void xt_register_ixterr(c_char *func, c_char *file, u_int line, int xt_err, c_char *item)
850
{
851
xt_register_i2xterr(func, file, line, xt_err, item, NULL);
852
}
853
854
xtPublic void xt_register_tabcolerr(c_char *func, c_char *file, u_int line, int xt_err, XTPathStrPtr tab_item, c_char *item2)
855
{
856
char buffer[XT_TABLE_NAME_BUF_SIZE];
857
858
xt_tab_make_table_name(tab_item, buffer, sizeof(buffer));
859
xt_strcat(sizeof(buffer), buffer, ".");
860
xt_strcat(sizeof(buffer), buffer, item2);
861
xt_register_ixterr(func, file, line, xt_err, buffer);
862
}
863
864
xtPublic void xt_register_taberr(c_char *func, c_char *file, u_int line, int xt_err, XTPathStrPtr tab_item)
865
{
866
char buffer[XT_TABLE_NAME_BUF_SIZE];
867
868
xt_tab_make_table_name(tab_item, buffer, sizeof(buffer));
869
xt_register_ixterr(func, file, line, xt_err, buffer);
870
}
871
872
xtPublic void xt_register_ulxterr(c_char *func, c_char *file, u_int line, int xt_err, u_long value)
873
{
874
char buffer[100];
875
876
sprintf(buffer, "%lu", value);
877
xt_register_ixterr(func, file, line, xt_err, buffer);
878
}
879
880
xtPublic xtBool xt_register_ferrno(c_char *func, c_char *file, u_int line, int err, c_char *path)
881
{
882
char err_msg[XT_SYS_ERR_SIZE];
883
884
xt_registerf(func, file, line, XT_SYSTEM_ERROR, err, "%s: '%s'", thr_get_sys_error(err, err_msg), path);
885
return FAILED;
886
}
887
888
xtPublic void xt_register_error(c_char *func, c_char *file, u_int line, int xt_err, int sys_err, c_char *msg)
889
{
890
xt_registerf(func, file, line, xt_err, sys_err, "%s", msg);
891
}
892
893
xtPublic xtBool xt_register_errno(c_char *func, c_char *file, u_int line, int err)
894
{
895
char err_msg[XT_SYS_ERR_SIZE];
896
897
xt_register_error(func, file, line, XT_SYSTEM_ERROR, err, thr_get_sys_error(err, err_msg));
898
return FAILED;
899
}
900
901
xtPublic void xt_register_xterr(c_char *func, c_char *file, u_int line, int xt_err)
902
{
903
xt_register_error(func, file, line, xt_err, 0, thr_get_err_string(xt_err));
904
}
905
906
/*
907
* -----------------------------------------------------------------------
908
* CREATING EXCEPTIONS
909
*/
910
911
xtPublic void xt_exceptionf(XTExceptionPtr e, XTThreadPtr self, c_char *func, c_char *file, u_int line, int xt_err, int sys_err, c_char *fmt, ...)
912
{
913
va_list ap;
914
915
va_start(ap, fmt);
916
thr_save_error_va(e, self, FALSE, func, file, line, xt_err, sys_err, fmt, ap);
917
va_end(ap);
918
}
919
920
xtPublic void xt_exception_error(XTExceptionPtr e, XTThreadPtr self, c_char *func, c_char *file, u_int line, int xt_err, int sys_err, c_char *msg)
921
{
922
xt_exceptionf(e, self, func, file, line, xt_err, sys_err, "%s", msg);
923
}
924
925
xtPublic xtBool xt_exception_errno(XTExceptionPtr e, XTThreadPtr self, c_char *func, c_char *file, u_int line, int err)
926
{
927
char err_msg[XT_SYS_ERR_SIZE];
928
929
xt_exception_error(e, self, func, file, line, XT_SYSTEM_ERROR, err, thr_get_sys_error(err, err_msg));
930
return FAILED;
931
}
932
933
xtPublic void xt_exception_xterr(XTExceptionPtr e, XTThreadPtr self, c_char *func, c_char *file, u_int line, int xt_err)
934
{
935
xt_exception_error(e, self, func, file, line, xt_err, 0, thr_get_err_string(xt_err));
936
}
937
938
/*
939
* -----------------------------------------------------------------------
940
* LOG ERRORS
941
*/
942
943
xtPublic void xt_log_errno(XTThreadPtr self, c_char *func, c_char *file, u_int line, int err)
944
{
945
XTExceptionRec e;
946
947
xt_exception_errno(&e, self, func, file, line, err);
948
xt_log_exception(self, &e, XT_LOG_DEFAULT);
949
}
950
951
/*
952
* -----------------------------------------------------------------------
953
* Assertions and failures (one breakpoints for all failures)
954
*/
955
//#define CRASH_ON_ASSERT
956
957
xtPublic xtBool xt_assert(XTThreadPtr self, c_char *expr, c_char *func, c_char *file, u_int line)
958
{
959
(void) self;
960
#ifdef DEBUG
961
//xt_set_fflush(TRUE);
962
//xt_dump_trace();
963
break_in_assertion(expr, func, file, line);
964
#ifdef CRASH_ON_ASSERT
965
abort();
966
#endif
967
#ifdef XT_WIN
968
FatalAppExit(0, "Assertion Failed!");
969
#endif
970
#else
971
xt_throw_assertion(self, func, file, line, expr);
972
#endif
973
return FALSE;
974
}
975
976
xtPublic xtBool xt_assume(XTThreadPtr self, c_char *expr, c_char *func, c_char *file, u_int line)
977
{
978
xt_log_assertion(self, func, file, line, expr);
979
return FALSE;
980
}
981
982
/*
983
* -----------------------------------------------------------------------
984
* Create and destroy threads
985
*/
986
987
typedef struct ThreadData {
988
xtBool td_started;
989
XTThreadPtr td_thr;
990
void *(*td_start_routine)(XTThreadPtr self);
991
} ThreadDataRec, *ThreadDataPtr;
992
993
#ifdef XT_WIN
994
pthread_key(void *, thr_key);
995
#else
996
static pthread_key_t thr_key;
997
#endif
998
999
#ifdef HANDLE_SIGNALS
1000
static void thr_ignore_signal(int sig)
1001
{
1002
#pragma unused(sig)
1003
}
1004
1005
static void thr_throw_signal(int sig)
1006
{
1007
XTThreadPtr self;
1008
1009
self = xt_get_self();
1010
1011
if (self->t_main) {
1012
/* The main thread will pass on a signal to all threads: */
1013
xt_signal_all_threads(self, sig);
1014
if (sig != SIGTERM) {
1015
if (self->t_disable_interrupts) {
1016
self->t_delayed_signal = sig;
1017
self->t_disable_interrupts = FALSE; /* Prevent infinite loop */
1018
}
1019
else {
1020
self->t_delayed_signal = 0;
1021
xt_throw_signal(self, "thr_throw_signal", NULL, 0, sig);
1022
}
1023
}
1024
}
1025
else {
1026
if (self->t_disable_interrupts) {
1027
self->t_delayed_signal = sig;
1028
self->t_disable_interrupts = FALSE; /* Prevent infinite loop */
1029
}
1030
else {
1031
self->t_delayed_signal = 0;
1032
xt_throw_signal(self, "thr_throw_signal", NULL, 0, sig);
1033
}
1034
}
1035
}
1036
1037
static xtBool thr_setup_signals(void)
1038
{
1039
struct sigaction action;
1040
1041
sigemptyset(&action.sa_mask);
1042
action.sa_flags = 0;
1043
action.sa_handler = thr_ignore_signal;
1044
1045
if (sigaction(SIGPIPE, &action, NULL) == -1)
1046
goto error_occurred;
1047
if (sigaction(SIGHUP, &action, NULL) == -1)
1048
goto error_occurred;
1049
1050
action.sa_handler = thr_throw_signal;
1051
1052
if (sigaction(SIGQUIT, &action, NULL) == -1)
1053
goto error_occurred;
1054
if (sigaction(SIGTERM, &action, NULL) == -1)
1055
goto error_occurred;
1056
#ifndef DEBUG
1057
if (sigaction(SIGILL, &action, NULL) == -1)
1058
goto error_occurred;
1059
if (sigaction(SIGBUS, &action, NULL) == -1)
1060
goto error_occurred;
1061
if (sigaction(SIGSEGV, &action, NULL) == -1)
1062
goto error_occurred;
1063
#endif
1064
return TRUE;
1065
1066
error_occurred:
1067
xt_log_errno(XT_NS_CONTEXT, errno);
1068
return FALSE;
1069
}
1070
#endif
1071
1072
typedef void *(*ThreadMainFunc)(XTThreadPtr self);
1073
1074
extern "C" void *xt_thread_main(void *data)
1075
{
1076
ThreadDataPtr td = (ThreadDataPtr) data;
1077
XTThreadPtr self = td->td_thr;
1078
ThreadMainFunc start_routine;
1079
void *return_data;
1080
1081
enter_();
1082
self->t_pthread = pthread_self();
1083
start_routine = td->td_start_routine;
1084
return_data = NULL;
1085
1086
#ifdef HANDLE_SIGNALS
1087
if (!thr_setup_signals())
1088
return NULL;
1089
#endif
1090
1091
try_(a) {
1092
if (!xt_set_key((pthread_key_t)thr_key, self, &self->t_exception))
1093
throw_();
1094
td->td_started = TRUE;
1095
return_data = (*start_routine)(self);
1096
}
1097
catch_(a) {
1098
xt_log_and_clear_exception(self);
1099
}
1100
cont_(a);
1101
1102
outer_();
1103
xt_free_thread(self);
1104
1105
/* {MYSQL-THREAD-KILL}
1106
* Clean up any remaining MySQL thread!
1107
*/
1108
myxt_delete_remaining_thread();
1109
return return_data;
1110
}
1111
1112
static void thr_free_data(XTThreadPtr self)
1113
{
1114
if (self->t_free_data) {
1115
(*self->t_free_data)(self, self->t_data);
1116
self->t_data = NULL;
1117
}
1118
}
1119
1120
xtPublic void xt_set_thread_data(XTThreadPtr self, void *data, XTThreadFreeFunc free_func)
1121
{
1122
thr_free_data(self);
1123
self->t_free_data = free_func;
1124
self->t_data = data;
1125
}
1126
1127
static void thr_exit(XTThreadPtr self)
1128
{
1129
/* Free the thread temporary data. */
1130
thr_free_resources(self, (XTResourcePtr) self->x.t_res_stack);
1131
xt_db_exit_thread(self);
1132
thr_free_data(self); /* Free custom user data. */
1133
1134
if (self->t_id > 0) {
1135
ASSERT(self->t_id < xt_thr_current_max_threads);
1136
xt_lock_mutex(self, &thr_array_lock);
1137
pushr_(xt_unlock_mutex, &thr_array_lock);
1138
thr_accumulate_statistics(self);
1139
// No read resize lock required if I have the array lock
1140
xt_thr_array[self->t_id].td_thread = NULL;
1141
xt_thr_current_thread_count--;
1142
if (self->t_id+1 == xt_thr_current_max_threads) {
1143
/* We can reduce the current maximum,
1144
* this makes operations that scan the array faster!
1145
*/
1146
u_int i;
1147
1148
i = self->t_id;
1149
for(;;) {
1150
if (xt_thr_array[i].td_thread)
1151
break;
1152
if (!i)
1153
break;
1154
i--;
1155
}
1156
xt_thr_current_max_threads = i+1;
1157
}
1158
freer_(); // xt_unlock_mutex(&thr_array_lock)
1159
}
1160
1161
xt_free_cond(&self->t_cond);
1162
xt_free_mutex(&self->t_lock);
1163
1164
self->st_tasks_todo.pl_exit();
1165
self->st_tasks_done.pl_exit();
1166
1167
self->st_thread_list_count = 0;
1168
self->st_thread_list_size = 0;
1169
if (self->st_thread_list) {
1170
xt_free_ns(self->st_thread_list);
1171
self->st_thread_list = NULL;
1172
}
1173
}
1174
1175
#ifdef DEBUG
1176
#define XT_THREAD_ARRAY_INC_SIZE 5
1177
#else
1178
#define XT_THREAD_ARRAY_INC_SIZE 100
1179
#endif
1180
1181
static xtBool thr_init_ns(XTThreadPtr new_thread)
1182
{
1183
new_thread->t_res_top = (XTResourcePtr) new_thread->x.t_res_stack;
1184
1185
new_thread->st_thread_list_count = 0;
1186
new_thread->st_thread_list_size = 0;
1187
new_thread->st_thread_list = NULL;
1188
1189
if (!new_thread->st_tasks_todo.pl_init_ns())
1190
goto failed;
1191
1192
if (!new_thread->st_tasks_done.pl_init_ns())
1193
goto failed;
1194
1195
xt_init_cond(NULL, &new_thread->t_cond);
1196
xt_init_mutex_with_autoname(NULL, &new_thread->t_lock);
1197
1198
xt_lock_mutex_ns(&thr_array_lock);
1199
1200
ASSERT_NS(xt_thr_current_thread_count <= xt_thr_current_max_threads);
1201
ASSERT_NS(xt_thr_current_max_threads <= xt_thr_maximum_threads);
1202
if (xt_thr_current_thread_count == xt_thr_maximum_threads) {
1203
XTThreadDataRec *tmp;
1204
1205
/* We can use a fixed thread ID because only one thread can do
1206
* this at any given time.
1207
*
1208
* We use XT_SYS_THREAD_ID to avoid the problem that we this function
1209
* allocates thread IDs, but needs a thread ID to aquire the
1210
* lock here!
1211
*/
1212
THR_ARRAY_WRITE_LOCK(&xt_thr_array_resize_lock, XT_SYS_THREAD_ID);
1213
if (!(tmp = (XTThreadDataRec *) realloc(xt_thr_array, (xt_thr_maximum_threads + XT_THREAD_ARRAY_INC_SIZE) * sizeof(XTThreadDataRec)))) {
1214
THR_ARRAY_UNLOCK(&xt_thr_array_resize_lock, XT_SYS_THREAD_ID);
1215
goto failed_2;
1216
}
1217
1218
xt_thr_array = tmp;
1219
xt_thr_maximum_threads += XT_THREAD_ARRAY_INC_SIZE;
1220
1221
for (u_int i=xt_thr_maximum_threads - XT_THREAD_ARRAY_INC_SIZE; i<xt_thr_maximum_threads; i++)
1222
xt_thr_array[i].td_thread = NULL;
1223
1224
for (u_int i=xt_thr_maximum_threads - XT_THREAD_ARRAY_INC_SIZE; i<xt_thr_maximum_threads; i++) {
1225
if (!(xt_thr_array[i].td_waiting = (XTWaitThreadPtr) xt_calloc_ns(sizeof(XTWaitThreadRec)))) {
1226
xt_thr_maximum_threads = i;
1227
THR_ARRAY_UNLOCK(&xt_thr_array_resize_lock, XT_SYS_THREAD_ID);
1228
goto failed_2;
1229
}
1230
xt_init_mutex_with_autoname(NULL, &xt_thr_array[i].td_waiting->wt_lock);
1231
xt_init_cond(NULL, &xt_thr_array[i].td_waiting->wt_cond);
1232
xt_spinlock_init_with_autoname(NULL, &xt_thr_array[i].td_waiting->wt_wait_list_lock);
1233
}
1234
1235
THR_ARRAY_UNLOCK(&xt_thr_array_resize_lock, XT_SYS_THREAD_ID);
1236
}
1237
1238
if (xt_thr_current_thread_count == xt_thr_current_max_threads) {
1239
new_thread->t_id = xt_thr_current_thread_count;
1240
xt_thr_array[new_thread->t_id].td_thread = new_thread;
1241
xt_thr_current_max_threads++;
1242
}
1243
else {
1244
/* There must be a free slot: */
1245
for (u_int i=0; i<xt_thr_current_max_threads; i++) {
1246
if (!xt_thr_array[i].td_thread) {
1247
new_thread->t_id = i;
1248
xt_thr_array[i].td_thread = new_thread;
1249
break;
1250
}
1251
}
1252
}
1253
xt_thr_current_thread_count++;
1254
1255
xt_unlock_mutex_ns(&thr_array_lock);
1256
1257
xt_db_init_thread_ns(new_thread);
1258
return OK;
1259
1260
failed_2:
1261
xt_unlock_mutex_ns(&thr_array_lock);
1262
1263
failed:
1264
thr_exit(new_thread);
1265
return FAILED;
1266
}
1267
1268
/*
1269
* The caller of this function automatically becomes the main thread.
1270
*/
1271
xtPublic XTThreadPtr xt_init_threading()
1272
{
1273
volatile XTThreadPtr self = NULL;
1274
XTExceptionRec e;
1275
int err;
1276
1277
#ifdef HANDLE_SIGNALS
1278
if (!thr_setup_signals())
1279
return NULL;
1280
#endif
1281
1282
xt_p_init_threading();
1283
1284
err = pthread_key_create(&thr_key, NULL);
1285
if (err) {
1286
xt_log_errno(XT_NS_CONTEXT, err);
1287
return NULL;
1288
}
1289
1290
if ((err = xt_p_mutex_init_with_autoname(&thr_array_lock, NULL))) {
1291
xt_log_errno(XT_NS_CONTEXT, err);
1292
goto failed;
1293
}
1294
1295
xt_thr_maximum_threads = 2;
1296
if (!(xt_thr_array = (XTThreadDataRec *) malloc(xt_thr_maximum_threads * sizeof(XTThreadDataRec)))) {
1297
xt_log_errno(XT_NS_CONTEXT, XT_ENOMEM);
1298
goto failed;
1299
}
1300
1301
xt_thr_array[XT_NO_THREAD_ID].td_thread = (XTThreadPtr) 1; // Dummy, not used
1302
xt_thr_array[XT_NO_THREAD_ID].td_waiting = NULL;
1303
xt_thr_array[XT_SYS_THREAD_ID].td_thread = (XTThreadPtr) 1; // System ID, used for resizing the thread array
1304
xt_thr_array[XT_SYS_THREAD_ID].td_waiting = NULL;
1305
xt_thr_current_thread_count = XT_MIN_THREAD_ID;
1306
xt_thr_current_max_threads = XT_MIN_THREAD_ID;
1307
1308
THR_ARRAY_INIT_LOCK(NULL, &xt_thr_array_resize_lock);
1309
1310
/* Create the main thread: */
1311
self = xt_create_thread("MainThread", TRUE, FALSE, &e);
1312
if (!self) {
1313
xt_log_exception(NULL, &e, XT_LOG_DEFAULT);
1314
goto failed;
1315
}
1316
1317
try_(a) {
1318
XTThreadPtr thread = self;
1319
thr_list = xt_new_linkedlist(thread, NULL, NULL, TRUE);
1320
1321
}
1322
catch_(a) {
1323
XTThreadPtr thread = self;
1324
xt_log_and_clear_exception(thread);
1325
xt_exit_threading(thread);
1326
}
1327
cont_(a);
1328
1329
return self;
1330
1331
failed:
1332
xt_exit_threading(NULL);
1333
return NULL;
1334
}
1335
1336
xtPublic void xt_exit_threading(XTThreadPtr self)
1337
{
1338
if (thr_list) {
1339
xt_free_linkedlist(self, thr_list);
1340
thr_list = NULL;
1341
}
1342
1343
/* This should be the main thread! */
1344
if (self) {
1345
ASSERT(self->t_main);
1346
xt_free_thread(self);
1347
}
1348
1349
if (xt_thr_array) {
1350
for (u_int i=XT_MIN_THREAD_ID; i<xt_thr_maximum_threads; i++) {
1351
xt_free_mutex(&xt_thr_array[i].td_waiting->wt_lock);
1352
xt_free_cond(&xt_thr_array[i].td_waiting->wt_cond);
1353
xt_spinlock_free(self, &xt_thr_array[i].td_waiting->wt_wait_list_lock);
1354
if (xt_thr_array[i].td_waiting->wt_wait_list)
1355
xt_free_ns(xt_thr_array[i].td_waiting->wt_wait_list);
1356
xt_free_ns(xt_thr_array[i].td_waiting);
1357
}
1358
1359
free(xt_thr_array);
1360
xt_thr_array = NULL;
1361
THR_ARRAY_FREE_LOCK(self, &xt_thr_array_resize_lock);
1362
xt_free_mutex(&thr_array_lock);
1363
}
1364
1365
xt_thr_current_thread_count = 0;
1366
xt_thr_current_max_threads = 0;
1367
1368
/* I no longer delete 'thr_key' because
1369
* functions that call xt_get_self() after this
1370
* point will get junk back if we delete
1371
* thr_key. In particular the XT_THREAD_LOCK_INFO
1372
* code fails
1373
if (thr_key) {
1374
pthread_key_delete(thr_key);
1375
thr_key = (pthread_key_t) 0;
1376
}
1377
*/
1378
}
1379
1380
xtPublic void xt_wait_for_all_threads(XTThreadPtr self)
1381
{
1382
if (thr_list)
1383
xt_ll_wait_till_empty(self, thr_list);
1384
}
1385
1386
/*
1387
* Call this function in a busy wait loop!
1388
* Use if for wait loops that are not
1389
* time critical.
1390
*/
1391
xtPublic void xt_busy_wait(void)
1392
{
1393
#ifdef XT_WIN
1394
Sleep(1);
1395
#else
1396
usleep(10);
1397
#endif
1398
}
1399
1400
xtPublic void xt_critical_wait(void)
1401
{
1402
/* NOTE: On Mac xt_busy_wait() works better than xt_yield()
1403
*/
1404
#if defined(XT_MAC) || defined(XT_WIN)
1405
xt_busy_wait();
1406
#else
1407
xt_yield();
1408
#endif
1409
}
1410
1411
1412
/*
1413
* Use this for loops that time critical.
1414
* Time critical means we need to get going
1415
* as soon as possible!
1416
*/
1417
xtPublic void xt_yield(void)
1418
{
1419
#ifdef XT_WIN
1420
Sleep(0);
1421
#elif defined(XT_MAC) || defined(XT_SOLARIS)
1422
usleep(0);
1423
#elif defined(XT_NETBSD)
1424
sched_yield();
1425
#else
1426
pthread_yield();
1427
#endif
1428
}
1429
1430
xtPublic void xt_sleep_milli_second(u_int t)
1431
{
1432
#ifdef XT_WIN
1433
Sleep(t);
1434
#else
1435
usleep(t * 1000);
1436
#endif
1437
}
1438
1439
xtPublic void xt_signal_all_threads(XTThreadPtr self, int sig)
1440
{
1441
XTLinkedItemPtr li;
1442
XTThreadPtr sig_thr;
1443
1444
xt_ll_lock(self, thr_list);
1445
try_(a) {
1446
li = thr_list->ll_items;
1447
while (li) {
1448
sig_thr = (XTThreadPtr) li;
1449
if (sig_thr != self)
1450
pthread_kill(sig_thr->t_pthread, sig);
1451
li = li->li_next;
1452
}
1453
}
1454
catch_(a) {
1455
xt_ll_unlock(self, thr_list);
1456
throw_();
1457
}
1458
cont_(a);
1459
xt_ll_unlock(self, thr_list);
1460
}
1461
1462
/*
1463
* Apply the given function to all threads except self!
1464
*/
1465
xtPublic void xt_do_to_all_threads(XTThreadPtr self, void (*do_func_ptr)(XTThreadPtr self, XTThreadPtr to_thr, void *thunk), void *thunk)
1466
{
1467
XTLinkedItemPtr li;
1468
XTThreadPtr to_thr;
1469
1470
xt_ll_lock(self, thr_list);
1471
pushr_(xt_ll_unlock, thr_list);
1472
1473
li = thr_list->ll_items;
1474
while (li) {
1475
to_thr = (XTThreadPtr) li;
1476
if (to_thr != self)
1477
(*do_func_ptr)(self, to_thr, thunk);
1478
li = li->li_next;
1479
}
1480
1481
freer_(); // xt_ll_unlock(thr_list)
1482
}
1483
1484
xtPublic XTThreadPtr xt_get_self(void)
1485
{
1486
XTThreadPtr self;
1487
1488
/* First check if the handler has the data: */
1489
if ((self = myxt_get_self()))
1490
return self;
1491
/* Then it must be a background process, and the
1492
* thread info is stored in the local key: */
1493
return (XTThreadPtr) xt_get_key((pthread_key_t)thr_key);
1494
}
1495
1496
xtPublic void xt_set_self(XTThreadPtr self)
1497
{
1498
xt_set_key((pthread_key_t)thr_key, self, NULL);
1499
}
1500
1501
xtPublic void xt_clear_exception(XTThreadPtr thread)
1502
{
1503
thread->t_exception.e_xt_err = 0;
1504
thread->t_exception.e_sys_err = 0;
1505
*thread->t_exception.e_err_msg = 0;
1506
*thread->t_exception.e_func_name = 0;
1507
*thread->t_exception.e_source_file = 0;
1508
thread->t_exception.e_source_line = 0;
1509
*thread->t_exception.e_catch_trace = 0;
1510
}
1511
1512
/*
1513
* Create a thread without requiring thread to do it (as in xt_create_daemon()).
1514
*
1515
* This function returns NULL on error.
1516
*/
1517
xtPublic XTThreadPtr xt_create_thread(c_char *name, xtBool main_thread, xtBool user_thread, XTExceptionPtr e)
1518
{
1519
volatile XTThreadPtr self;
1520
1521
self = (XTThreadPtr) xt_calloc_ns(sizeof(XTThreadRec));
1522
if (!self) {
1523
xt_exception_errno(e, XT_CONTEXT, ENOMEM);
1524
return NULL;
1525
}
1526
1527
if (!xt_set_key((pthread_key_t)thr_key, self, e)) {
1528
xt_free_ns(self);
1529
return NULL;
1530
}
1531
1532
xt_strcpy(XT_THR_NAME_SIZE, self->t_name, name);
1533
self->t_main = main_thread;
1534
self->t_daemon = FALSE;
1535
1536
if (!thr_init_ns(self)) {
1537
*e = self->t_exception;
1538
xt_set_key((pthread_key_t)thr_key, NULL, NULL);
1539
xt_free_ns(self);
1540
self = NULL;
1541
}
1542
1543
if (self && user_thread) {
1544
/* Add non-temporary threads to the thread list. */
1545
try_(b) {
1546
xt_ll_add(self, thr_list, &self->t_links, TRUE);
1547
}
1548
catch_(b) {
1549
*e = self->t_exception;
1550
xt_free_thread(self);
1551
self = NULL;
1552
}
1553
cont_(b);
1554
}
1555
1556
return self;
1557
}
1558
1559
/*
1560
* Create a daemon thread.
1561
*/
1562
xtPublic XTThreadPtr xt_create_daemon_ns(c_char *name)
1563
{
1564
XTThreadPtr new_thread;
1565
1566
/* NOTE: thr_key will be set when this thread start running. */
1567
if (!(new_thread = (XTThreadPtr) xt_calloc_ns(sizeof(XTThreadRec))))
1568
return NULL;
1569
1570
xt_strcpy(XT_THR_NAME_SIZE, new_thread->t_name, name);
1571
new_thread->t_main = FALSE;
1572
new_thread->t_daemon = TRUE;
1573
1574
if (!thr_init_ns(new_thread)) {
1575
xt_free_ns(new_thread);
1576
return NULL;
1577
}
1578
1579
return new_thread;
1580
}
1581
1582
xtPublic XTThreadPtr xt_create_daemon(XTThreadPtr self, c_char *name)
1583
{
1584
XTThreadPtr new_thread;
1585
1586
if (!(new_thread = xt_create_daemon_ns(name)))
1587
xt_throw(self);
1588
return new_thread;
1589
}
1590
1591
void xt_free_thread(XTThreadPtr self)
1592
{
1593
thr_exit(self);
1594
if (!self->t_daemon && thr_list)
1595
xt_ll_remove(self, thr_list, &self->t_links, TRUE);
1596
/* Note, if I move this before thr_exit() then self = xt_get_self(); will fail in
1597
* xt_close_file_ns() which is called by xt_unuse_database()!
1598
*/
1599
1600
/*
1601
* Do not clear the pthread's key value unless it is the same as the thread just released.
1602
* This can happen during shutdown when the engine is deregistered with the PBMS engine.
1603
*
1604
* What happens is that during deregistration the PBMS engine calls close to close all
1605
* PBXT resources on all MySQL THDs created by PBMS for it's own pthreads. So the 'self'
1606
* being freed is not the same 'self' associated with the PBXT 'thr_key'.
1607
*/
1608
if (thr_key && (self == ((XTThreadPtr) xt_get_key((pthread_key_t)thr_key)))) {
1609
xt_set_key((pthread_key_t)thr_key, NULL, NULL);
1610
}
1611
xt_free_ns(self);
1612
}
1613
1614
xtPublic xtBool xt_run_thread_ns(XTThreadPtr child, void *(*start_routine)(XTThreadPtr))
1615
{
1616
ThreadDataRec data;
1617
int err;
1618
pthread_t child_thread;
1619
1620
// 'data' can be on the stack because we are waiting for the thread to start
1621
// before exiting the function.
1622
data.td_started = FALSE;
1623
data.td_thr = child;
1624
data.td_start_routine = start_routine;
1625
#ifdef XT_WIN
1626
{
1627
pthread_attr_t attr = { 0, 0, 0 };
1628
1629
attr.priority = THREAD_PRIORITY_NORMAL;
1630
err = pthread_create(&child_thread, &attr, xt_thread_main, &data);
1631
}
1632
#else
1633
err = pthread_create(&child_thread, NULL, xt_thread_main, &data);
1634
#endif
1635
if (err) {
1636
xt_free_thread(child);
1637
return xt_register_errno(XT_REG_CONTEXT, err);
1638
}
1639
1640
while (!data.td_started) {
1641
/* Check that the self is still alive: */
1642
if (pthread_kill(child_thread, 0))
1643
break;
1644
xt_busy_wait();
1645
}
1646
1647
return OK;
1648
}
1649
1650
xtPublic void xt_run_thread(XTThreadPtr self, XTThreadPtr child, void *(*start_routine)(XTThreadPtr))
1651
{
1652
if (!xt_run_thread_ns(child, start_routine))
1653
xt_throw(self);
1654
}
1655
1656
xtPublic void xt_exit_thread(XTThreadPtr self, void *result)
1657
{
1658
xt_free_thread(self);
1659
pthread_exit(result);
1660
}
1661
1662
xtPublic void *xt_wait_for_thread_to_exit(xtThreadID tid, xtBool ignore_error)
1663
{
1664
int err;
1665
void *value_ptr = NULL;
1666
xtBool ok = FALSE;
1667
XTThreadPtr thread;
1668
pthread_t t1 = 0;
1669
1670
xt_lock_mutex_ns(&thr_array_lock);
1671
if (tid < xt_thr_maximum_threads) {
1672
// No resize lock required if I have the array lock.
1673
if ((thread = xt_thr_array[tid].td_thread)) {
1674
t1 = thread->t_pthread;
1675
ok = TRUE;
1676
}
1677
}
1678
xt_unlock_mutex_ns(&thr_array_lock);
1679
if (ok) {
1680
err = xt_p_join(t1, &value_ptr);
1681
if (err && !ignore_error)
1682
xt_log_errno(XT_NS_CONTEXT, err);
1683
}
1684
return value_ptr;
1685
}
1686
1687
/*
1688
* Kill the given thead, and wait for it to terminate.
1689
* This function just returns if the self is already dead.
1690
*/
1691
xtPublic void xt_kill_thread(pthread_t t1)
1692
{
1693
int err;
1694
void *value_ptr = NULL;
1695
1696
err = pthread_kill(t1, SIGTERM);
1697
if (err)
1698
return;
1699
err = xt_p_join(t1, &value_ptr);
1700
if (err)
1701
xt_log_errno(XT_NS_CONTEXT, err);
1702
}
1703
1704
/*
1705
* -----------------------------------------------------------------------
1706
* Read/write locking
1707
*/
1708
1709
#ifdef XT_THREAD_LOCK_INFO
1710
xtPublic xtBool xt_init_rwlock(XTThreadPtr self, xt_rwlock_type *rwlock, const char *name)
1711
#else
1712
xtPublic xtBool xt_init_rwlock(XTThreadPtr self, xt_rwlock_type *rwlock)
1713
#endif
1714
{
1715
int err;
1716
1717
#ifdef XT_THREAD_LOCK_INFO
1718
err = xt_p_rwlock_init_with_name(rwlock, NULL, name);
1719
#else
1720
err = xt_p_rwlock_init(rwlock, NULL);
1721
#endif
1722
1723
if (err) {
1724
xt_throw_errno(XT_CONTEXT, err);
1725
return FAILED;
1726
}
1727
return OK;
1728
}
1729
1730
xtPublic void xt_free_rwlock(xt_rwlock_type *rwlock)
1731
{
1732
int err;
1733
1734
for (;;) {
1735
err = xt_p_rwlock_destroy(rwlock);
1736
if (err != XT_EBUSY)
1737
break;
1738
xt_busy_wait();
1739
}
1740
/* PMC - xt_xn_exit_db() is called even when xt_xn_init_db() is not fully completed!
1741
* This generates a lot of log entries. But I have no desire to only call
1742
* free for those articles that I have init'ed!
1743
if (err)
1744
xt_log_errno(XT_NS_CONTEXT, err);
1745
*/
1746
}
1747
1748
xtPublic xt_rwlock_type *xt_slock_rwlock(XTThreadPtr self, xt_rwlock_type *rwlock)
1749
{
1750
int err;
1751
1752
for (;;) {
1753
err = xt_slock_rwlock_ns(rwlock);
1754
if (err != XT_EAGAIN)
1755
break;
1756
xt_busy_wait();
1757
}
1758
if (err) {
1759
xt_throw_errno(XT_CONTEXT, err);
1760
return NULL;
1761
}
1762
return rwlock;
1763
}
1764
1765
xtPublic xt_rwlock_type *xt_xlock_rwlock(XTThreadPtr self, xt_rwlock_type *rwlock)
1766
{
1767
int err;
1768
1769
for (;;) {
1770
err = xt_xlock_rwlock_ns(rwlock);
1771
if (err != XT_EAGAIN)
1772
break;
1773
xt_busy_wait();
1774
}
1775
1776
if (err) {
1777
xt_throw_errno(XT_CONTEXT, err);
1778
return NULL;
1779
}
1780
return rwlock;
1781
}
1782
1783
xtPublic void xt_unlock_rwlock(XTThreadPtr XT_UNUSED(self), xt_rwlock_type *rwlock)
1784
{
1785
int err;
1786
1787
err = xt_unlock_rwlock_ns(rwlock);
1788
if (err)
1789
xt_log_errno(XT_NS_CONTEXT, err);
1790
}
1791
1792
/*
1793
* -----------------------------------------------------------------------
1794
* Mutex locking
1795
*/
1796
1797
xtPublic xt_mutex_type *xt_new_mutex(XTThreadPtr self)
1798
{
1799
xt_mutex_type *mx;
1800
1801
if (!(mx = (xt_mutex_type *) xt_calloc(self, sizeof(xt_mutex_type))))
1802
return NULL;
1803
pushr_(xt_free, mx);
1804
if (!xt_init_mutex_with_autoname(self, mx)) {
1805
freer_();
1806
return NULL;
1807
}
1808
popr_();
1809
return mx;
1810
}
1811
1812
xtPublic void xt_delete_mutex(XTThreadPtr self, xt_mutex_type *mx)
1813
{
1814
if (mx) {
1815
xt_free_mutex(mx);
1816
xt_free(self, mx);
1817
}
1818
}
1819
1820
#ifdef XT_THREAD_LOCK_INFO
1821
xtPublic xtBool xt_init_mutex(XTThreadPtr self, xt_mutex_type *mx, const char *name)
1822
#else
1823
xtPublic xtBool xt_init_mutex(XTThreadPtr self, xt_mutex_type *mx)
1824
#endif
1825
{
1826
int err;
1827
1828
err = xt_p_mutex_init_with_name(mx, NULL, name);
1829
if (err) {
1830
xt_throw_errno(XT_CONTEXT, err);
1831
return FALSE;
1832
}
1833
return TRUE;
1834
}
1835
1836
void xt_free_mutex(xt_mutex_type *mx)
1837
{
1838
int err;
1839
1840
for (;;) {
1841
err = xt_p_mutex_destroy(mx);
1842
if (err != XT_EBUSY)
1843
break;
1844
xt_busy_wait();
1845
}
1846
/* PMC - xt_xn_exit_db() is called even when xt_xn_init_db() is not fully completed!
1847
if (err)
1848
xt_log_errno(XT_NS_CONTEXT, err);
1849
*/
1850
}
1851
1852
xtPublic xtBool xt_lock_mutex(XTThreadPtr self, xt_mutex_type *mx)
1853
{
1854
int err;
1855
1856
for (;;) {
1857
err = xt_lock_mutex_ns(mx);
1858
if (err != XT_EAGAIN)
1859
break;
1860
xt_busy_wait();
1861
}
1862
1863
if (err) {
1864
xt_throw_errno(XT_CONTEXT, err);
1865
return FALSE;
1866
}
1867
return TRUE;
1868
}
1869
1870
xtPublic void xt_unlock_mutex(XTThreadPtr self, xt_mutex_type *mx)
1871
{
1872
int err;
1873
1874
err = xt_unlock_mutex_ns(mx);
1875
if (err)
1876
xt_throw_errno(XT_CONTEXT, err);
1877
}
1878
1879
xtPublic xtBool xt_set_key(pthread_key_t key, const void *value, XTExceptionPtr e)
1880
{
1881
#ifdef XT_WIN
1882
my_pthread_setspecific_ptr(thr_key, (void *) value);
1883
#else
1884
int err;
1885
1886
err = pthread_setspecific(key, value);
1887
if (err) {
1888
if (e)
1889
xt_exception_errno(e, XT_NS_CONTEXT, err);
1890
return FALSE;
1891
}
1892
#endif
1893
return TRUE;
1894
}
1895
1896
xtPublic void *xt_get_key(pthread_key_t key)
1897
{
1898
#ifdef XT_WIN
1899
return my_pthread_getspecific_ptr(void *, thr_key);
1900
#else
1901
return pthread_getspecific(key);
1902
#endif
1903
}
1904
1905
xtPublic xt_cond_type *xt_new_cond(XTThreadPtr self)
1906
{
1907
xt_cond_type *cond;
1908
1909
if (!(cond = (xt_cond_type *) xt_calloc(self, sizeof(xt_cond_type))))
1910
return NULL;
1911
pushr_(xt_free, cond);
1912
if (!xt_init_cond(self, cond)) {
1913
freer_();
1914
return NULL;
1915
}
1916
popr_();
1917
return cond;
1918
}
1919
1920
xtPublic void xt_delete_cond(XTThreadPtr self, xt_cond_type *cond)
1921
{
1922
if (cond) {
1923
xt_free_cond(cond);
1924
xt_free(self, cond);
1925
}
1926
}
1927
1928
xtPublic xtBool xt_init_cond(XTThreadPtr self, xt_cond_type *cond)
1929
{
1930
int err;
1931
1932
err = pthread_cond_init(cond, NULL);
1933
if (err) {
1934
xt_throw_errno(XT_CONTEXT, err);
1935
return FALSE;
1936
}
1937
return TRUE;
1938
}
1939
1940
xtPublic void xt_free_cond(xt_cond_type *cond)
1941
{
1942
int err;
1943
1944
for (;;) {
1945
err = pthread_cond_destroy(cond);
1946
if (err != XT_EBUSY)
1947
break;
1948
xt_busy_wait();
1949
}
1950
/* PMC - xt_xn_exit_db() is called even when xt_xn_init_db() is not fully completed!
1951
if (err)
1952
xt_log_errno(XT_NS_CONTEXT, err);
1953
*/
1954
}
1955
1956
xtPublic xtBool xt_throw_delayed_signal(XTThreadPtr self, c_char *func, c_char *file, u_int line)
1957
{
1958
XTThreadPtr me = self ? self : xt_get_self();
1959
1960
if (me->t_delayed_signal) {
1961
int sig = me->t_delayed_signal;
1962
1963
me->t_delayed_signal = 0;
1964
xt_throw_signal(self, func, file, line, sig);
1965
return FAILED;
1966
}
1967
return OK;
1968
}
1969
1970
xtPublic xtBool xt_wait_cond(XTThreadPtr self, xt_cond_type *cond, xt_mutex_type *mutex)
1971
{
1972
int err;
1973
XTThreadPtr me = self ? self : xt_get_self();
1974
1975
/* PMC - In my tests, if I throw an exception from within the wait
1976
* the condition and the mutex remain locked.
1977
*/
1978
me->t_disable_interrupts = TRUE;
1979
err = xt_p_cond_wait(cond, mutex);
1980
me->t_disable_interrupts = FALSE;
1981
if (err) {
1982
xt_throw_errno(XT_CONTEXT, err);
1983
return FALSE;
1984
}
1985
if (me->t_delayed_signal) {
1986
xt_throw_delayed_signal(XT_CONTEXT);
1987
return FALSE;
1988
}
1989
return TRUE;
1990
}
1991
1992
xtPublic xtBool xt_suspend(XTThreadPtr thread)
1993
{
1994
xtBool ok;
1995
1996
// You can only suspend yourself.
1997
ASSERT_NS(pthread_equal(thread->t_pthread, pthread_self()));
1998
1999
xt_lock_mutex_ns(&thread->t_lock);
2000
ok = xt_wait_cond(NULL, &thread->t_cond, &thread->t_lock);
2001
xt_unlock_mutex_ns(&thread->t_lock);
2002
return ok;
2003
}
2004
2005
xtPublic xtBool xt_unsuspend(XTThreadPtr target)
2006
{
2007
return xt_broadcast_cond_ns(&target->t_cond);
2008
}
2009
2010
xtPublic void xt_lock_thread(XTThreadPtr thread)
2011
{
2012
xt_lock_mutex_ns(&thread->t_lock);
2013
}
2014
2015
xtPublic void xt_unlock_thread(XTThreadPtr thread)
2016
{
2017
xt_unlock_mutex_ns(&thread->t_lock);
2018
}
2019
2020
xtPublic xtBool xt_wait_thread(XTThreadPtr thread)
2021
{
2022
return xt_wait_cond(NULL, &thread->t_cond, &thread->t_lock);
2023
}
2024
2025
xtPublic xtBool xt_timed_wait_thread(XTThreadPtr thread, u_long milli_sec)
2026
{
2027
return xt_timed_wait_cond(NULL, &thread->t_cond, &thread->t_lock, milli_sec);
2028
}
2029
2030
xtPublic void xt_signal_thread(XTThreadPtr target)
2031
{
2032
xt_broadcast_cond_ns(&target->t_cond);
2033
}
2034
2035
xtPublic void xt_terminate_thread(XTThreadPtr XT_UNUSED(self), XTThreadPtr target)
2036
{
2037
target->t_quit = TRUE;
2038
target->t_delayed_signal = SIGTERM;
2039
}
2040
2041
xtPublic xtProcID xt_getpid()
2042
{
2043
#ifdef XT_WIN
2044
return GetCurrentProcessId();
2045
#else
2046
return getpid();
2047
#endif
2048
}
2049
2050
xtPublic xtBool xt_process_exists(xtProcID pid)
2051
{
2052
xtBool found;
2053
2054
#ifdef XT_WIN
2055
HANDLE h;
2056
DWORD code;
2057
2058
found = FALSE;
2059
h = OpenProcess(PROCESS_QUERY_INFORMATION, FALSE, pid);
2060
if (h) {
2061
if (GetExitCodeProcess(h, &code)) {
2062
if (code == STILL_ACTIVE)
2063
found = TRUE;
2064
}
2065
CloseHandle(h);
2066
}
2067
else {
2068
int err;
2069
2070
err = HRESULT_CODE(GetLastError());
2071
if (err != ERROR_INVALID_PARAMETER)
2072
found = TRUE;
2073
}
2074
#else
2075
found = TRUE;
2076
if (kill(pid, 0) == -1) {
2077
if (errno == ESRCH)
2078
found = FALSE;
2079
}
2080
#endif
2081
return found;
2082
}
2083
2084
xtPublic xtBool xt_timed_wait_cond(XTThreadPtr self, xt_cond_type *cond, xt_mutex_type *mutex, u_long milli_sec)
2085
{
2086
int err;
2087
struct timespec abstime;
2088
XTThreadPtr me = self ? self : xt_get_self();
2089
2090
#ifdef XT_WIN
2091
union ft64 now;
2092
2093
GetSystemTimeAsFileTime(&now.ft);
2094
2095
/* System time is measured in 100ns units.
2096
* This calculation will be reversed by the Windows implementation
2097
* of pthread_cond_timedwait(), in order to extract the
2098
* milli-second timeout!
2099
*/
2100
abstime.tv.i64 = now.i64 + (milli_sec * 10000);
2101
2102
abstime.max_timeout_msec = milli_sec;
2103
#else
2104
struct timeval now;
2105
u_llong micro_sec;
2106
2107
/* Get the current time in microseconds: */
2108
gettimeofday(&now, NULL);
2109
micro_sec = (u_llong) now.tv_sec * (u_llong) 1000000 + (u_llong) now.tv_usec;
2110
2111
/* Add the timeout which is in milli seconds */
2112
micro_sec += (u_llong) milli_sec * (u_llong) 1000;
2113
2114
/* Setup the end time, which is in nano-seconds. */
2115
abstime.tv_sec = (long) (micro_sec / 1000000); /* seconds */
2116
abstime.tv_nsec = (long) ((micro_sec % 1000000) * 1000); /* and nanoseconds */
2117
#endif
2118
2119
me->t_disable_interrupts = TRUE;
2120
err = xt_p_cond_timedwait(cond, mutex, &abstime);
2121
me->t_disable_interrupts = FALSE;
2122
if (err && err != ETIMEDOUT) {
2123
xt_throw_errno(XT_CONTEXT, err);
2124
return FALSE;
2125
}
2126
if (me->t_delayed_signal) {
2127
xt_throw_delayed_signal(XT_CONTEXT);
2128
return FALSE;
2129
}
2130
return TRUE;
2131
}
2132
2133
xtPublic xtBool xt_signal_cond(XTThreadPtr self, xt_cond_type *cond)
2134
{
2135
int err;
2136
2137
err = pthread_cond_signal(cond);
2138
if (err) {
2139
xt_throw_errno(XT_CONTEXT, err);
2140
return FAILED;
2141
}
2142
return OK;
2143
}
2144
2145
xtPublic void xt_broadcast_cond(XTThreadPtr self, xt_cond_type *cond)
2146
{
2147
int err;
2148
2149
err = pthread_cond_broadcast(cond);
2150
if (err)
2151
xt_throw_errno(XT_CONTEXT, err);
2152
}
2153
2154
xtPublic xtBool xt_broadcast_cond_ns(xt_cond_type *cond)
2155
{
2156
int err;
2157
2158
err = pthread_cond_broadcast(cond);
2159
if (err) {
2160
xt_register_errno(XT_REG_CONTEXT, err);
2161
return FAILED;
2162
}
2163
return OK;
2164
}
2165
2166
static int prof_setjmp_count = 0;
2167
2168
xtPublic int prof_setjmp(void)
2169
{
2170
prof_setjmp_count++;
2171
return 0;
2172
}
2173
2174
xtPublic void xt_set_low_priority(XTThreadPtr self)
2175
{
2176
int err = xt_p_set_low_priority(self->t_pthread);
2177
if (err) {
2178
self = NULL; /* Will cause logging, instead of throwing exception */
2179
xt_throw_errno(XT_CONTEXT, err);
2180
}
2181
}
2182
2183
xtPublic void xt_set_normal_priority(XTThreadPtr self)
2184
{
2185
int err = xt_p_set_normal_priority(self->t_pthread);
2186
if (err) {
2187
self = NULL; /* Will cause logging, instead of throwing exception */
2188
xt_throw_errno(XT_CONTEXT, err);
2189
}
2190
}
2191
2192
xtPublic void xt_set_high_priority(XTThreadPtr self)
2193
{
2194
int err = xt_p_set_high_priority(self->t_pthread);
2195
if (err) {
2196
self = NULL; /* Will cause logging, instead of throwing exception */
2197
xt_throw_errno(XT_CONTEXT, err);
2198
}
2199
}
2200
2201
xtPublic void xt_set_priority(XTThreadPtr self, int priority)
2202
{
2203
if (priority < XT_PRIORITY_NORMAL)
2204
xt_set_low_priority(self);
2205
else if (priority > XT_PRIORITY_NORMAL)
2206
xt_set_high_priority(self);
2207
else
2208
xt_set_normal_priority(self);
2209
}
2210
2211
/* ----------------------------------------------------------------------
2212
* THREAD WAIT LIST
2213
*/
2214
2215
/*
2216
* Add a thread the wakeup list of a thread.
2217
*/
2218
xtPublic xtBool xt_add_to_wakeup_list(xtThreadID waiting_id, xtThreadID wait_for_id)
2219
{
2220
XTWaitThreadPtr wt;
2221
2222
THR_ARRAY_READ_LOCK(&xt_thr_array_resize_lock, waiting_id);
2223
wt = xt_thr_array[wait_for_id].td_waiting;
2224
THR_ARRAY_UNLOCK(&xt_thr_array_resize_lock, waiting_id);
2225
xt_spinlock_lock(&wt->wt_wait_list_lock);
2226
if (wt->wt_wait_list_count == wt->wt_wait_list_size) {
2227
if (!xt_realloc_ns((void **) &wt->wt_wait_list, (wt->wt_wait_list_size+1) * sizeof(xtThreadID)))
2228
return FAILED;
2229
wt->wt_wait_list_size++;
2230
}
2231
for (u_int i=0; i<wt->wt_wait_list_count; i++) {
2232
if (wt->wt_wait_list[i] == waiting_id)
2233
goto done;
2234
}
2235
wt->wt_wait_list[wt->wt_wait_list_count] = waiting_id;
2236
wt->wt_wait_list_count++;
2237
done:
2238
xt_spinlock_unlock(&wt->wt_wait_list_lock);
2239
return OK;
2240
}
2241
2242
/*
2243
* Wakeup a single thread.
2244
*/
2245
xtPublic void xt_wakeup_thread(xtThreadID thd_id, XTThreadPtr thread)
2246
{
2247
XTWaitThreadPtr target_wt;
2248
2249
THR_ARRAY_READ_LOCK(&xt_thr_array_resize_lock, thread->t_id);
2250
target_wt = xt_thr_array[thd_id].td_waiting;
2251
THR_ARRAY_UNLOCK(&xt_thr_array_resize_lock, thread->t_id);
2252
xt_lock_mutex_ns(&target_wt->wt_lock);
2253
xt_broadcast_cond_ns(&target_wt->wt_cond);
2254
xt_unlock_mutex_ns(&target_wt->wt_lock);
2255
}
2256
2257
/*
2258
* Wakeup a list of threads (the list is local to the calling thread).
2259
*/
2260
xtPublic void xt_wakeup_thread_list(XTThreadPtr thread)
2261
{
2262
XTWaitThreadPtr target_wt;
2263
2264
for (u_int i=0; i<thread->st_thread_list_count; i++) {
2265
THR_ARRAY_READ_LOCK(&xt_thr_array_resize_lock, thread->t_id);
2266
target_wt = xt_thr_array[thread->st_thread_list[i]].td_waiting;
2267
THR_ARRAY_UNLOCK(&xt_thr_array_resize_lock, thread->t_id);
2268
xt_lock_mutex_ns(&target_wt->wt_lock);
2269
xt_broadcast_cond_ns(&target_wt->wt_cond);
2270
xt_unlock_mutex_ns(&target_wt->wt_lock);
2271
}
2272
thread->st_thread_list_count = 0;
2273
}
2274
2275
/*
2276
* Wakeup all threads waiting for this thread.
2277
*/
2278
xtPublic void xt_wakeup_waiting_threads(XTThreadPtr thread)
2279
{
2280
XTWaitThreadPtr wt;
2281
XTWaitThreadPtr target_wt;
2282
2283
THR_ARRAY_READ_LOCK(&xt_thr_array_resize_lock, thread->t_id);
2284
wt = xt_thr_array[thread->t_id].td_waiting;
2285
THR_ARRAY_UNLOCK(&xt_thr_array_resize_lock, thread->t_id);
2286
if (!wt->wt_wait_list_count)
2287
return;
2288
2289
xt_spinlock_lock(&wt->wt_wait_list_lock);
2290
if (thread->st_thread_list_size < wt->wt_wait_list_count) {
2291
if (!xt_realloc_ns((void **) &thread->st_thread_list, wt->wt_wait_list_count * sizeof(xtThreadID)))
2292
goto failed;
2293
thread->st_thread_list_size = wt->wt_wait_list_count;
2294
}
2295
memcpy(thread->st_thread_list, wt->wt_wait_list, wt->wt_wait_list_count * sizeof(xtThreadID));
2296
thread->st_thread_list_count = wt->wt_wait_list_count;
2297
wt->wt_wait_list_count = 0;
2298
xt_spinlock_unlock(&wt->wt_wait_list_lock);
2299
2300
xt_wakeup_thread_list(thread);
2301
return;
2302
2303
failed:
2304
for (u_int i=0; i<wt->wt_wait_list_count; i++) {
2305
THR_ARRAY_READ_LOCK(&xt_thr_array_resize_lock, thread->t_id);
2306
target_wt = xt_thr_array[wt->wt_wait_list[i]].td_waiting;
2307
THR_ARRAY_UNLOCK(&xt_thr_array_resize_lock, thread->t_id);
2308
xt_lock_mutex_ns(&target_wt->wt_lock);
2309
xt_broadcast_cond_ns(&target_wt->wt_cond);
2310
xt_unlock_mutex_ns(&target_wt->wt_lock);
2311
}
2312
wt->wt_wait_list_count = 0;
2313
xt_spinlock_unlock(&wt->wt_wait_list_lock);
2314
}
2315
2316
/*
2317
* -----------------------------------------------------------------------
2318
* STATISTICS
2319
*/
2320
2321
#ifdef DEBUG_NO_ACTIVITY
2322
static void debug_no_activity()
2323
{
2324
xt_logf(XT_NT_INFO, "No activity!\n");
2325
}
2326
2327
time_t last_call_time;
2328
u_llong last_commit_total;
2329
xtWord8 last_rec_flush_time = (xtWord8) -1;
2330
xtWord8 last_rec_write_time = (xtWord8) -1;
2331
xtWord8 last_ind_flush_time = (xtWord8) -1;
2332
xtWord8 last_ind_write_time = (xtWord8) -1;
2333
xtWord8 last_ilog_flush_time = (xtWord8) -1;
2334
xtWord8 last_ilog_write_time = (xtWord8) -1;
2335
xtWord8 last_xlog_flush_time = (xtWord8) -1;
2336
xtWord8 last_xlog_write_time = (xtWord8) -1;
2337
xtWord8 last_data_flush_time = (xtWord8) -1;
2338
xtWord8 last_data_write_time = (xtWord8) -1;
2339
#endif
2340
2341
xtPublic void xt_gather_statistics(XTStatisticsPtr stats)
2342
{
2343
XTThreadDataRec *thr_data;
2344
XTThreadPtr thr;
2345
xtWord8 s;
2346
2347
xt_lock_mutex_ns(&thr_array_lock);
2348
*stats = thr_statistics;
2349
// Ignore index 0, it is not used!
2350
// No read resize lock required if I have the array lock
2351
thr_data = &xt_thr_array[XT_MIN_THREAD_ID];
2352
for (u_int i=XT_MIN_THREAD_ID; i<xt_thr_current_max_threads; i++) {
2353
if ((thr = thr_data->td_thread)) {
2354
stats->st_commits += thr->st_statistics.st_commits;
2355
stats->st_rollbacks += thr->st_statistics.st_rollbacks;
2356
stats->st_stat_read += thr->st_statistics.st_stat_read;
2357
stats->st_stat_write += thr->st_statistics.st_stat_write;
2358
2359
XT_ADD_STATS(stats->st_rec, thr->st_statistics.st_rec);
2360
if ((s = thr->st_statistics.st_rec.ts_flush_start))
2361
stats->st_rec.ts_flush_time += xt_trace_clock() - s;
2362
#ifdef XT_TIME_DISK_WRITES
2363
if ((s = thr->st_statistics.st_rec.ts_write_start))
2364
stats->st_rec.ts_write_time += xt_trace_clock() - s;
2365
#endif
2366
#ifdef XT_TIME_DISK_READS
2367
if ((s = thr->st_statistics.st_rec.ts_read_start))
2368
stats->st_rec.ts_read_time += xt_trace_clock() - s;
2369
#endif
2370
stats->st_rec_cache_hit += thr->st_statistics.st_rec_cache_hit;
2371
stats->st_rec_cache_miss += thr->st_statistics.st_rec_cache_miss;
2372
stats->st_rec_cache_frees += thr->st_statistics.st_rec_cache_frees;
2373
2374
XT_ADD_STATS(stats->st_ind, thr->st_statistics.st_ind);
2375
if ((s = thr->st_statistics.st_ind.ts_flush_start))
2376
stats->st_ind.ts_flush_time += xt_trace_clock() - s;
2377
#ifdef XT_TIME_DISK_WRITES
2378
if ((s = thr->st_statistics.st_ind.ts_write_start))
2379
stats->st_ind.ts_write_time += xt_trace_clock() - s;
2380
#endif
2381
#ifdef XT_TIME_DISK_READS
2382
if ((s = thr->st_statistics.st_ind.ts_read_start))
2383
stats->st_ind.ts_read_time += xt_trace_clock() - s;
2384
#endif
2385
stats->st_ind_cache_hit += thr->st_statistics.st_ind_cache_hit;
2386
stats->st_ind_cache_miss += thr->st_statistics.st_ind_cache_miss;
2387
XT_ADD_STATS(stats->st_ilog, thr->st_statistics.st_ilog);
2388
if ((s = thr->st_statistics.st_ilog.ts_flush_start))
2389
stats->st_ilog.ts_flush_time += xt_trace_clock() - s;
2390
#ifdef XT_TIME_DISK_WRITES
2391
if ((s = thr->st_statistics.st_ilog.ts_write_start))
2392
stats->st_ilog.ts_write_time += xt_trace_clock() - s;
2393
#endif
2394
#ifdef XT_TIME_DISK_READS
2395
if ((s = thr->st_statistics.st_ilog.ts_read_start))
2396
stats->st_ilog.ts_read_time += xt_trace_clock() - s;
2397
#endif
2398
2399
XT_ADD_STATS(stats->st_xlog, thr->st_statistics.st_xlog);
2400
if ((s = thr->st_statistics.st_xlog.ts_flush_start))
2401
stats->st_xlog.ts_flush_time += xt_trace_clock() - s;
2402
#ifdef XT_TIME_DISK_WRITES
2403
if ((s = thr->st_statistics.st_xlog.ts_write_start))
2404
stats->st_xlog.ts_write_time += xt_trace_clock() - s;
2405
#endif
2406
#ifdef XT_TIME_DISK_READS
2407
if ((s = thr->st_statistics.st_xlog.ts_read_start))
2408
stats->st_xlog.ts_read_time += xt_trace_clock() - s;
2409
#endif
2410
stats->st_xlog_cache_hit += thr->st_statistics.st_xlog_cache_hit;
2411
stats->st_xlog_cache_miss += thr->st_statistics.st_xlog_cache_miss;
2412
2413
XT_ADD_STATS(stats->st_data, thr->st_statistics.st_data);
2414
if ((s = thr->st_statistics.st_data.ts_flush_start))
2415
stats->st_data.ts_flush_time += xt_trace_clock() - s;
2416
#ifdef XT_TIME_DISK_WRITES
2417
if ((s = thr->st_statistics.st_data.ts_write_start))
2418
stats->st_data.ts_write_time += xt_trace_clock() - s;
2419
#endif
2420
#ifdef XT_TIME_DISK_READS
2421
if ((s = thr->st_statistics.st_data.ts_read_start))
2422
stats->st_data.ts_read_time += xt_trace_clock() - s;
2423
#endif
2424
2425
stats->st_scan_index += thr->st_statistics.st_scan_index;
2426
stats->st_scan_table += thr->st_statistics.st_scan_table;
2427
stats->st_row_select += thr->st_statistics.st_row_select;
2428
stats->st_row_insert += thr->st_statistics.st_row_insert;
2429
stats->st_row_update += thr->st_statistics.st_row_update;
2430
stats->st_row_delete += thr->st_statistics.st_row_delete;
2431
2432
stats->st_wait_for_xact += thr->st_statistics.st_wait_for_xact;
2433
stats->st_retry_index_scan += thr->st_statistics.st_retry_index_scan;
2434
stats->st_reread_record_list += thr->st_statistics.st_reread_record_list;
2435
2436
stats->st_ind_cache_dirty += thr->st_statistics.st_ind_cache_dirty;
2437
}
2438
thr_data++;
2439
}
2440
xt_unlock_mutex_ns(&thr_array_lock);
2441
2442
#ifdef DEBUG_NO_ACTIVITY
2443
time_t now = time(NULL);
2444
2445
/* Make sure at least 1 second has gone by: */
2446
if (!last_call_time || now > last_call_time) {
2447
if (last_commit_total &&
2448
last_commit_total == stats->st_commits &&
2449
last_rec_flush_time == stats->st_rec.ts_flush_time &&
2450
last_ind_flush_time == stats->st_ind.ts_flush_time &&
2451
last_ilog_flush_time == stats->st_ilog.ts_flush_time &&
2452
last_xlog_flush_time == stats->st_xlog.ts_flush_time &&
2453
last_data_flush_time == stats->st_data.ts_flush_time &&
2454
last_rec_write_time == stats->st_rec.ts_write_time &&
2455
last_ind_write_time == stats->st_ind.ts_write_time &&
2456
last_ilog_write_time == stats->st_ilog.ts_write_time &&
2457
last_xlog_write_time == stats->st_xlog.ts_write_time &&
2458
last_data_write_time == stats->st_data.ts_write_time
2459
)
2460
debug_no_activity();
2461
2462
last_call_time = now;
2463
last_commit_total = stats->st_commits;
2464
2465
last_rec_flush_time = stats->st_rec.ts_flush_time;
2466
last_ind_flush_time = stats->st_ind.ts_flush_time;
2467
last_ilog_flush_time = stats->st_ilog.ts_flush_time;
2468
last_xlog_flush_time = stats->st_xlog.ts_flush_time;
2469
last_data_flush_time = stats->st_data.ts_flush_time;
2470
2471
last_rec_write_time = stats->st_rec.ts_write_time;
2472
last_ind_write_time = stats->st_ind.ts_write_time;
2473
last_ilog_write_time = stats->st_ilog.ts_write_time;
2474
last_xlog_write_time = stats->st_xlog.ts_write_time;
2475
last_data_write_time = stats->st_data.ts_write_time;
2476
}
2477
#endif
2478
}
2479
2480
xtPublic int xt_get_index_cache_dirty_perc()
2481
{
2482
XTThreadDataRec *thr;
2483
XTStatisticsRec stats;
2484
double v;
2485
2486
xt_lock_mutex_ns(&thr_array_lock);
2487
stats = thr_statistics;
2488
// Ignore index 0 and 1, it is not used!
2489
// No read resize lock required if I have the array lock
2490
thr = &xt_thr_array[XT_MIN_THREAD_ID];
2491
for (u_int i=XT_MIN_THREAD_ID; i<xt_thr_current_max_threads; i++) {
2492
if (thr->td_thread)
2493
stats.st_ind_cache_dirty += thr->td_thread->st_statistics.st_ind_cache_dirty;
2494
thr++;
2495
}
2496
xt_unlock_mutex_ns(&thr_array_lock);
2497
v = (double) stats.st_ind_cache_dirty * (double) XT_INDEX_PAGE_SIZE;
2498
return (int) (v / (double) xt_ind_get_size() * (double) 100);
2499
}
2500
2501
static void thr_accumulate_statistics(XTThreadPtr self)
2502
{
2503
thr_statistics.st_commits += self->st_statistics.st_commits;
2504
thr_statistics.st_rollbacks += self->st_statistics.st_rollbacks;
2505
thr_statistics.st_stat_read += self->st_statistics.st_stat_read;
2506
thr_statistics.st_stat_write += self->st_statistics.st_stat_write;
2507
2508
XT_ADD_STATS(thr_statistics.st_rec, self->st_statistics.st_rec);
2509
thr_statistics.st_rec_cache_hit += self->st_statistics.st_rec_cache_hit;
2510
thr_statistics.st_rec_cache_miss += self->st_statistics.st_rec_cache_miss;
2511
thr_statistics.st_rec_cache_frees += self->st_statistics.st_rec_cache_frees;
2512
2513
XT_ADD_STATS(thr_statistics.st_ind, self->st_statistics.st_ind);
2514
thr_statistics.st_ind_cache_hit += self->st_statistics.st_ind_cache_hit;
2515
thr_statistics.st_ind_cache_miss += self->st_statistics.st_ind_cache_miss;
2516
XT_ADD_STATS(thr_statistics.st_ilog, self->st_statistics.st_ilog);
2517
2518
XT_ADD_STATS(thr_statistics.st_xlog, self->st_statistics.st_xlog);
2519
thr_statistics.st_xlog_cache_hit += self->st_statistics.st_xlog_cache_hit;
2520
thr_statistics.st_xlog_cache_miss += self->st_statistics.st_xlog_cache_miss;
2521
2522
XT_ADD_STATS(thr_statistics.st_data, self->st_statistics.st_data);
2523
2524
thr_statistics.st_scan_index += self->st_statistics.st_scan_index;
2525
thr_statistics.st_scan_table += self->st_statistics.st_scan_table;
2526
thr_statistics.st_row_select += self->st_statistics.st_row_select;
2527
thr_statistics.st_row_insert += self->st_statistics.st_row_insert;
2528
thr_statistics.st_row_update += self->st_statistics.st_row_update;
2529
thr_statistics.st_row_delete += self->st_statistics.st_row_delete;
2530
2531
thr_statistics.st_wait_for_xact += self->st_statistics.st_wait_for_xact;
2532
thr_statistics.st_retry_index_scan += self->st_statistics.st_retry_index_scan;
2533
thr_statistics.st_reread_record_list += self->st_statistics.st_reread_record_list;
2534
2535
thr_statistics.st_ind_cache_dirty += self->st_statistics.st_ind_cache_dirty;
2536
}
2537
2538
xtPublic u_llong xt_get_statistic(XTStatisticsPtr stats, XTDatabaseHPtr db, u_int rec_id)
2539
{
2540
u_llong stat_value;
2541
2542
switch (rec_id) {
2543
case XT_STAT_TIME_CURRENT:
2544
stat_value = (u_llong) time(NULL);
2545
break;
2546
case XT_STAT_TIME_PASSED:
2547
stat_value = (u_llong) xt_trace_clock();
2548
break;
2549
case XT_STAT_COMMITS:
2550
stat_value = stats->st_commits;
2551
break;
2552
case XT_STAT_ROLLBACKS:
2553
stat_value = stats->st_rollbacks;
2554
break;
2555
case XT_STAT_STAT_READS:
2556
stat_value = stats->st_stat_read;
2557
break;
2558
case XT_STAT_STAT_WRITES:
2559
stat_value = stats->st_stat_write;
2560
break;
2561
2562
case XT_STAT_REC_BYTES_IN:
2563
stat_value = stats->st_rec.ts_read;
2564
break;
2565
case XT_STAT_REC_BYTES_OUT:
2566
stat_value = stats->st_rec.ts_write;
2567
break;
2568
case XT_STAT_REC_SYNC_COUNT:
2569
stat_value = stats->st_rec.ts_flush;
2570
break;
2571
case XT_STAT_REC_SYNC_TIME:
2572
stat_value = stats->st_rec.ts_flush_time;
2573
break;
2574
case XT_STAT_REC_CACHE_HIT:
2575
stat_value = stats->st_rec_cache_hit;
2576
break;
2577
case XT_STAT_REC_CACHE_MISS:
2578
stat_value = stats->st_rec_cache_miss;
2579
break;
2580
case XT_STAT_REC_CACHE_FREES:
2581
stat_value = stats->st_rec_cache_frees;
2582
break;
2583
case XT_STAT_REC_CACHE_USAGE:
2584
stat_value = (u_llong) xt_tc_get_usage();
2585
break;
2586
2587
case XT_STAT_IND_BYTES_IN:
2588
stat_value = stats->st_ind.ts_read;
2589
break;
2590
case XT_STAT_IND_BYTES_OUT:
2591
stat_value = stats->st_ind.ts_write;
2592
break;
2593
case XT_STAT_IND_SYNC_COUNT:
2594
stat_value = stats->st_ind.ts_flush;
2595
break;
2596
case XT_STAT_IND_SYNC_TIME:
2597
stat_value = stats->st_ind.ts_flush_time;
2598
break;
2599
case XT_STAT_IND_CACHE_HIT:
2600
stat_value = stats->st_ind_cache_hit;
2601
break;
2602
case XT_STAT_IND_CACHE_MISS:
2603
stat_value = stats->st_ind_cache_miss;
2604
break;
2605
case XT_STAT_IND_CACHE_USAGE:
2606
stat_value = (u_llong) xt_ind_get_usage();
2607
break;
2608
case XT_STAT_ILOG_BYTES_IN:
2609
stat_value = stats->st_ilog.ts_read;
2610
break;
2611
case XT_STAT_ILOG_BYTES_OUT:
2612
stat_value = stats->st_ilog.ts_write;
2613
break;
2614
case XT_STAT_ILOG_SYNC_COUNT:
2615
stat_value = stats->st_ilog.ts_flush;
2616
break;
2617
case XT_STAT_ILOG_SYNC_TIME:
2618
stat_value = stats->st_ilog.ts_flush_time;
2619
break;
2620
2621
case XT_STAT_XLOG_BYTES_IN:
2622
stat_value = stats->st_xlog.ts_read;
2623
break;
2624
case XT_STAT_XLOG_BYTES_OUT:
2625
stat_value = stats->st_xlog.ts_write;
2626
break;
2627
case XT_STAT_XLOG_SYNC_COUNT:
2628
stat_value = stats->st_xlog.ts_flush;
2629
break;
2630
case XT_STAT_XLOG_SYNC_TIME:
2631
stat_value = stats->st_xlog.ts_flush_time;
2632
break;
2633
case XT_STAT_XLOG_CACHE_HIT:
2634
stat_value = stats->st_xlog_cache_hit;
2635
break;
2636
case XT_STAT_XLOG_CACHE_MISS:
2637
stat_value = stats->st_xlog_cache_miss;
2638
break;
2639
case XT_STAT_XLOG_CACHE_USAGE:
2640
stat_value = (u_llong) xt_xlog_get_usage();
2641
break;
2642
2643
case XT_STAT_DATA_BYTES_IN:
2644
stat_value = stats->st_data.ts_read;
2645
break;
2646
case XT_STAT_DATA_BYTES_OUT:
2647
stat_value = stats->st_data.ts_write;
2648
break;
2649
case XT_STAT_DATA_SYNC_COUNT:
2650
stat_value = stats->st_data.ts_flush;
2651
break;
2652
case XT_STAT_DATA_SYNC_TIME:
2653
stat_value = stats->st_data.ts_flush_time;
2654
break;
2655
2656
case XT_STAT_BYTES_TO_CHKPNT:
2657
stat_value = db ? xt_bytes_since_last_checkpoint(db, db->db_xlog.xl_write_log_id, db->db_xlog.xl_write_log_offset) : 0;
2658
break;
2659
case XT_STAT_LOG_BYTES_TO_WRITE:
2660
stat_value = db ? db->db_xlog.xl_log_bytes_written - db->db_xlog.xl_log_bytes_read : 0;//db->db_xlog.xlog_bytes_to_write();
2661
break;
2662
case XT_STAT_BYTES_TO_SWEEP:
2663
/* This stat is potentially very expensive: */
2664
stat_value = db ? xt_xn_bytes_to_sweep(db, xt_get_self()) : 0;
2665
break;
2666
case XT_STAT_WAIT_FOR_XACT:
2667
stat_value = stats->st_wait_for_xact;
2668
break;
2669
case XT_STAT_XACT_TO_CLEAN:
2670
#ifdef XT_SWEEPER_SORT_XACTS
2671
stat_value = db ? db->db_xn_curr_id + 1 - db->db_sw_to_add + db->db_sw_list_size : 0;
2672
#else
2673
stat_value = db ? db->db_xn_curr_id + 1 - db->db_xn_to_clean_id : 0;
2674
#endif
2675
break;
2676
case XT_STAT_SWEEPER_WAITS:
2677
stat_value = db ? db->db_stat_sweep_waits : 0;
2678
break;
2679
2680
case XT_STAT_SCAN_INDEX:
2681
stat_value = stats->st_scan_index;
2682
break;
2683
case XT_STAT_SCAN_TABLE:
2684
stat_value = stats->st_scan_table;
2685
break;
2686
case XT_STAT_ROW_SELECT:
2687
stat_value = stats->st_row_select;
2688
break;
2689
case XT_STAT_ROW_INSERT:
2690
stat_value = stats->st_row_insert;
2691
break;
2692
case XT_STAT_ROW_UPDATE:
2693
stat_value = stats->st_row_update;
2694
break;
2695
case XT_STAT_ROW_DELETE:
2696
stat_value = stats->st_row_delete;
2697
break;
2698
2699
case XT_STAT_RETRY_INDEX_SCAN:
2700
stat_value = stats->st_retry_index_scan;
2701
break;
2702
case XT_STAT_REREAD_REC_LIST:
2703
stat_value = stats->st_reread_record_list;
2704
break;
2705
2706
case XT_STAT_IND_CACHE_DIRTY:
2707
ASSERT_NS(stats->st_ind_cache_dirty >= 0);
2708
if (stats->st_ind_cache_dirty < 0)
2709
stat_value = 0;
2710
else
2711
stat_value = stats->st_ind_cache_dirty;
2712
stat_value *= (u_llong) XT_INDEX_PAGE_SIZE;
2713
break;
2714
2715
#ifdef XT_TIME_DISK_WRITES
2716
case XT_STAT_REC_WRITE_TIME:
2717
stat_value = stats->st_rec.ts_write_time;
2718
break;
2719
case XT_STAT_IND_WRITE_TIME:
2720
stat_value = stats->st_ind.ts_write_time;
2721
break;
2722
case XT_STAT_ILOG_WRITE_TIME:
2723
stat_value = stats->st_ilog.ts_write_time;
2724
break;
2725
case XT_STAT_XLOG_WRITE_TIME:
2726
stat_value = stats->st_xlog.ts_write_time;
2727
break;
2728
case XT_STAT_DATA_WRITE_TIME:
2729
stat_value = stats->st_data.ts_write_time;
2730
break;
2731
#endif
2732
2733
#ifdef XT_TIME_DISK_READS
2734
case XT_STAT_REC_READ_TIME:
2735
stat_value = stats->st_rec.ts_read_time;
2736
break;
2737
case XT_STAT_IND_READ_TIME:
2738
stat_value = stats->st_ind.ts_read_time;
2739
break;
2740
case XT_STAT_LOG_READ_TIME:
2741
stat_value = stats->st_ilog.ts_read_time + stats->st_xlog.ts_read_time + stats->st_data.ts_read_time;
2742
break;
2743
#endif
2744
2745
default:
2746
stat_value = 0;
2747
break;
2748
}
2749
return stat_value;
2750
}
2751
Loggerhead is a web-based interface for Breezy
Version: 2.0.1