1
/* Copyright (C) 2005 PrimeBase Technologies GmbH
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.
10
* This program is distributed in the hope that it will be useful,
11
* but WITHOUT ANY WARRANTY; without even the implied warranty of
12
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13
* GNU General Public License for more details.
15
* You should have received a copy of the GNU General Public License
16
* along with this program; if not, write to the Free Software
17
* Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
19
* 2005-01-24 Paul McCullagh
24
#include "xt_config.h"
34
#include "mysql_priv.h"
39
#include "filesys_xt.h"
40
#include "database_xt.h"
41
#include "memory_xt.h"
42
#include "strutil_xt.h"
43
#include "sortedlist_xt.h"
50
static void dl_wake_co_thread(XTDatabaseHPtr db);
53
* --------------------------------------------------------------------------------
57
xtBool XTDataSeqRead::sl_seq_init(struct XTDatabase *db, size_t buffer_size)
60
sl_buffer_size = buffer_size;
65
sl_buf_log_offset = 0;
67
sl_buffer = (xtWord1 *) xt_malloc_ns(buffer_size);
70
sl_rec_log_offset = 0;
74
return sl_buffer != NULL;
77
void XTDataSeqRead::sl_seq_exit()
80
xt_close_file_ns(sl_log_file);
84
xt_free_ns(sl_buffer);
89
XTOpenFilePtr XTDataSeqRead::sl_seq_open_file()
94
void XTDataSeqRead::sl_seq_pos(xtLogID *log_id, xtLogOffset *log_offset)
96
*log_id = sl_rec_log_id;
97
*log_offset = sl_rec_log_offset;
100
xtBool XTDataSeqRead::sl_seq_start(xtLogID log_id, xtLogOffset log_offset, xtBool missing_ok)
102
if (sl_rec_log_id != log_id) {
104
xt_close_file_ns(sl_log_file);
108
sl_rec_log_id = log_id;
109
sl_buf_log_offset = sl_rec_log_offset;
112
if (!sl_db->db_datalogs.dlc_open_log(&sl_log_file, log_id, missing_ok ? XT_FS_MISSING_OK : XT_FS_DEFAULT))
115
sl_log_eof = xt_seek_eof_file(NULL, sl_log_file);
117
sl_rec_log_offset = log_offset;
122
xtBool XTDataSeqRead::sl_rnd_read(xtLogOffset log_offset, size_t size, xtWord1 *buffer, size_t *data_read, struct XTThread *thread)
128
return xt_pread_file(sl_log_file, log_offset, size, 0, buffer, data_read, &thread->st_statistics.st_data, thread);
132
* Unlike the transaction log sequential reader, this function only returns
133
* the header of a record.
136
* This function now skips gaps. This should not be required, because in normal
137
* operation, no gaps should be created.
139
* However, if his happens there is a danger that a valid record after the
142
* So, if we find an invalid record, we scan through the log to find the next
143
* valid record. Note, that there is still a danger that will will find
144
* data that looks like a valid record, but is not.
146
* In this case, this "pseudo record" may cause the function to actually skip
149
* Note, any such malfunction will eventually cause the record to be lost forever
150
* after the garbage collector has run.
152
xtBool XTDataSeqRead::sl_seq_next(XTXactLogBufferDPtr *ret_entry, struct XTThread *thread)
154
XTXactLogBufferDPtr record;
159
xtBool reread_from_buffer;
161
xtLogOffset gap_start = 0;
163
/* Go to the next record (xseq_record_len must be initialized
164
* to 0 for this to work.
167
sl_rec_log_offset += sl_record_len;
170
if (sl_rec_log_offset < sl_buf_log_offset ||
171
sl_rec_log_offset >= sl_buf_log_offset + (xtLogOffset) sl_buffer_len) {
172
/* The current position is nowhere near the buffer, read data into the
175
tfer = sl_buffer_size;
176
if (!sl_rnd_read(sl_rec_log_offset, tfer, sl_buffer, &tfer, thread))
178
sl_buf_log_offset = sl_rec_log_offset;
179
sl_buffer_len = tfer;
181
/* Should we go to the next log? */
186
/* The start of the record is in the buffer: */
188
rec_offset = (size_t) (sl_rec_log_offset - sl_buf_log_offset);
189
max_rec_len = sl_buffer_len - rec_offset;
190
reread_from_buffer = FALSE;
193
/* Check the type of record: */
194
record = (XTXactLogBufferDPtr) (sl_buffer + rec_offset);
195
switch (record->xl.xl_status_1) {
196
case XT_LOG_ENT_HEADER:
197
if (sl_rec_log_offset != 0)
198
goto scan_to_next_record;
199
if (offsetof(XTXactLogHeaderDRec, xh_size_4) + 4 > max_rec_len) {
200
reread_from_buffer = TRUE;
203
len = XT_GET_DISK_4(record->xh.xh_size_4);
204
if (len > max_rec_len) {
205
reread_from_buffer = TRUE;
209
if (record->xh.xh_checksum_1 != XT_CHECKSUM_1(sl_rec_log_id))
211
if (XT_LOG_HEAD_MAGIC(record, len) != XT_LOG_FILE_MAGIC)
213
if (len > offsetof(XTXactLogHeaderDRec, xh_log_id_4) + 4) {
214
if (XT_GET_DISK_4(record->xh.xh_log_id_4) != sl_rec_log_id)
218
case XT_LOG_ENT_EXT_REC_OK:
219
case XT_LOG_ENT_EXT_REC_DEL:
221
xt_logf(XT_NS_CONTEXT, XT_LOG_WARNING, "Gap in data log %lu, start: %llu, size: %llu\n", (u_long) sl_rec_log_id, (u_llong) gap_start, (u_llong) (sl_rec_log_offset - gap_start));
224
len = offsetof(XTactExtRecEntryDRec, er_data);
225
if (len > max_rec_len) {
226
reread_from_buffer = TRUE;
229
size = XT_GET_DISK_4(record->er.er_data_size_4);
230
/* Verify the record as good as we can! */
232
goto scan_to_next_record;
233
if (sl_rec_log_offset + (xtLogOffset) offsetof(XTactExtRecEntryDRec, er_data) + size > sl_log_eof)
234
goto scan_to_next_record;
235
if (!XT_GET_DISK_4(record->er.er_tab_id_4))
236
goto scan_to_next_record;
237
if (!XT_GET_DISK_4(record->er.er_rec_id_4))
238
goto scan_to_next_record;
241
/* Note, we no longer assume EOF.
242
* Instead, we skip to the next value record. */
243
goto scan_to_next_record;
246
if (len <= max_rec_len) {
247
/* The record is completely in the buffer: */
248
sl_record_len = len+size;
254
/* The record is partially in the buffer. */
255
memmove(sl_buffer, sl_buffer + rec_offset, max_rec_len);
256
sl_buf_log_offset += rec_offset;
257
sl_buffer_len = max_rec_len;
259
/* Read the rest, as far as possible: */
260
tfer = sl_buffer_size - max_rec_len;
261
if (!sl_rnd_read(sl_buf_log_offset + max_rec_len, tfer, sl_buffer + max_rec_len, &tfer, thread))
263
sl_buffer_len += tfer;
265
if (sl_buffer_len < len)
266
/* A partial record is in the log, must be the end of the log: */
269
if (reread_from_buffer)
270
goto read_from_buffer;
272
/* The record is not completely in the buffer: */
274
*ret_entry = (XTXactLogBufferDPtr) sl_buffer;
279
gap_start = sl_rec_log_offset;
280
xt_logf(XT_NS_CONTEXT, XT_LOG_WARNING, "Gap found in data log %lu, starting at offset %llu\n", (u_long) sl_rec_log_id, (u_llong) gap_start);
288
xt_logf(XT_NS_CONTEXT, XT_LOG_WARNING, "Gap in data log %lu, start: %llu, size: %llu\n", (u_long) sl_rec_log_id, (u_llong) gap_start, (u_llong) (sl_rec_log_offset - gap_start));
295
void XTDataSeqRead::sl_seq_skip(size_t size)
297
sl_record_len += size;
300
void XTDataSeqRead::sl_seq_skip_to(off_t log_offset)
302
if (log_offset >= sl_rec_log_offset)
303
sl_record_len = (size_t) (log_offset - sl_rec_log_offset);
307
* --------------------------------------------------------------------------------
311
static xtBool dl_create_log_header(XTDataLogFilePtr data_log, XTOpenFilePtr of, XTThreadPtr thread)
313
XTXactLogHeaderDRec header;
315
/* The header was not completely written, so write a new one: */
316
memset(&header, 0, sizeof(XTXactLogHeaderDRec));
317
header.xh_status_1 = XT_LOG_ENT_HEADER;
318
header.xh_checksum_1 = XT_CHECKSUM_1(data_log->dlf_log_id);
319
XT_SET_DISK_4(header.xh_size_4, sizeof(XTXactLogHeaderDRec));
320
XT_SET_DISK_8(header.xh_free_space_8, 0);
321
XT_SET_DISK_8(header.xh_file_len_8, sizeof(XTXactLogHeaderDRec));
322
XT_SET_DISK_4(header.xh_log_id_4, data_log->dlf_log_id);
323
XT_SET_DISK_2(header.xh_version_2, XT_LOG_VERSION_NO);
324
XT_SET_DISK_4(header.xh_magic_4, XT_LOG_FILE_MAGIC);
325
if (!xt_pwrite_file(of, 0, sizeof(XTXactLogHeaderDRec), &header, &thread->st_statistics.st_data, thread))
327
if (!xt_flush_file(of, &thread->st_statistics.st_data, thread))
332
static xtBool dl_write_garbage_level(XTDataLogFilePtr data_log, XTOpenFilePtr of, xtBool flush, XTThreadPtr thread)
334
XTXactLogHeaderDRec header;
336
/* The header was not completely written, so write a new one: */
337
XT_SET_DISK_8(header.xh_free_space_8, data_log->dlf_garbage_count);
338
if (!xt_pwrite_file(of, offsetof(XTXactLogHeaderDRec, xh_free_space_8), 8, (xtWord1 *) &header.xh_free_space_8, &thread->st_statistics.st_data, thread))
340
if (flush && !xt_flush_file(of, &thread->st_statistics.st_data, thread))
347
* Extra garbage is the amount of space skipped during recovery of the data
348
* log file. We assume this space has not be counted as garbage,
349
* and add it to the garbage count.
351
* This may mean that our estimate of garbaged is higher than it should
352
* be, but that is better than the other way around.
354
* The fact is, there should not be any gaps in the data log files, so
355
* this is actually an exeption which should not occur.
357
static xtBool dl_write_log_header(XTDataLogFilePtr data_log, XTOpenFilePtr of, xtLogOffset extra_garbage, XTThreadPtr thread)
359
XTXactLogHeaderDRec header;
361
XT_SET_DISK_8(header.xh_file_len_8, data_log->dlf_log_eof);
364
data_log->dlf_garbage_count += extra_garbage;
365
if (data_log->dlf_garbage_count > data_log->dlf_log_eof)
366
data_log->dlf_garbage_count = data_log->dlf_log_eof;
367
XT_SET_DISK_8(header.xh_free_space_8, data_log->dlf_garbage_count);
368
if (!xt_pwrite_file(of, offsetof(XTXactLogHeaderDRec, xh_free_space_8), 16, (xtWord1 *) &header.xh_free_space_8, &thread->st_statistics.st_data, thread))
372
if (!xt_pwrite_file(of, offsetof(XTXactLogHeaderDRec, xh_file_len_8), 8, (xtWord1 *) &header.xh_file_len_8, &thread->st_statistics.st_data, thread))
375
if (!xt_flush_file(of, &thread->st_statistics.st_data, thread))
380
static void dl_free_seq_read(XTThreadPtr XT_UNUSED(self), XTDataSeqReadPtr seq_read)
382
seq_read->sl_seq_exit();
385
static void dl_recover_log(XTThreadPtr self, XTDatabaseHPtr db, XTDataLogFilePtr data_log)
387
XTDataSeqReadRec seq_read;
388
XTXactLogBufferDPtr record;
390
if (!seq_read.sl_seq_init(db, xt_db_log_buffer_size))
392
pushr_(dl_free_seq_read, &seq_read);
394
seq_read.sl_seq_start(data_log->dlf_log_id, 0, FALSE);
397
if (!seq_read.sl_seq_next(&record, self))
401
switch (record->xh.xh_status_1) {
402
case XT_LOG_ENT_HEADER:
403
data_log->dlf_garbage_count = XT_GET_DISK_8(record->xh.xh_free_space_8);
404
data_log->dlf_start_offset = XT_GET_DISK_8(record->xh.xh_comp_pos_8);
405
seq_read.sl_seq_skip_to((off_t) XT_GET_DISK_8(record->xh.xh_file_len_8));
410
ASSERT_NS(seq_read.sl_log_eof == seq_read.sl_rec_log_offset);
411
data_log->dlf_log_eof = seq_read.sl_rec_log_offset;
413
if (data_log->dlf_log_eof < (off_t) sizeof(XTXactLogHeaderDRec)) {
414
data_log->dlf_log_eof = sizeof(XTXactLogHeaderDRec);
415
if (!dl_create_log_header(data_log, seq_read.sl_log_file, self))
419
if (!dl_write_log_header(data_log, seq_read.sl_log_file, seq_read.sl_extra_garbage, self))
423
freer_(); // dl_free_seq_read(&seq_read)
427
* --------------------------------------------------------------------------------
428
* D A T A L O G C AC H E
431
void XTDataLogCache::dls_remove_log(XTDataLogFilePtr data_log)
433
xtLogID log_id = data_log->dlf_log_id;
435
switch (data_log->dlf_state) {
436
case XT_DL_HAS_SPACE:
437
xt_sl_delete(NULL, dlc_has_space, &log_id);
439
case XT_DL_TO_COMPACT:
440
xt_sl_delete(NULL, dlc_to_compact, &log_id);
442
case XT_DL_TO_DELETE:
443
xt_sl_delete(NULL, dlc_to_delete, &log_id);
446
xt_sl_delete(NULL, dlc_deleted, &log_id);
451
int XTDataLogCache::dls_get_log_state(XTDataLogFilePtr data_log)
453
if (data_log->dlf_to_much_garbage())
454
return XT_DL_TO_COMPACT;
455
if (data_log->dlf_space_avaliable() > 0)
456
return XT_DL_HAS_SPACE;
457
return XT_DL_READ_ONLY;
460
xtBool XTDataLogCache::dls_set_log_state(XTDataLogFilePtr data_log, int state)
462
xtLogID log_id = data_log->dlf_log_id;
464
xt_lock_mutex_ns(&dlc_lock);
465
if (state == XT_DL_MAY_COMPACT) {
466
if (data_log->dlf_state != XT_DL_UNKNOWN &&
467
data_log->dlf_state != XT_DL_HAS_SPACE &&
468
data_log->dlf_state != XT_DL_READ_ONLY)
470
state = XT_DL_TO_COMPACT;
472
if (state == XT_DL_UNKNOWN)
473
state = dls_get_log_state(data_log);
475
case XT_DL_HAS_SPACE:
476
if (data_log->dlf_state != XT_DL_HAS_SPACE) {
477
dls_remove_log(data_log);
478
if (!xt_sl_insert(NULL, dlc_has_space, &log_id, &log_id))
482
case XT_DL_TO_COMPACT:
483
#ifdef DEBUG_LOG_DELETE
484
printf("-- set to compact: %d\n", (int) log_id);
486
if (data_log->dlf_state != XT_DL_TO_COMPACT) {
487
dls_remove_log(data_log);
488
if (!xt_sl_insert(NULL, dlc_to_compact, &log_id, &log_id))
491
dl_wake_co_thread(dlc_db);
493
case XT_DL_COMPACTED:
494
#ifdef DEBUG_LOG_DELETE
495
printf("-- set compacted: %d\n", (int) log_id);
497
if (data_log->dlf_state != state)
498
dls_remove_log(data_log);
500
case XT_DL_TO_DELETE:
501
#ifdef DEBUG_LOG_DELETE
502
printf("-- set to delete log: %d\n", (int) log_id);
504
if (data_log->dlf_state != XT_DL_TO_DELETE) {
505
dls_remove_log(data_log);
506
if (!xt_sl_insert(NULL, dlc_to_delete, &log_id, &log_id))
511
#ifdef DEBUG_LOG_DELETE
512
printf("-- set DELETED log: %d\n", (int) log_id);
514
if (data_log->dlf_state != XT_DL_DELETED) {
515
dls_remove_log(data_log);
516
if (!xt_sl_insert(NULL, dlc_deleted, &log_id, &log_id))
521
if (data_log->dlf_state != state)
522
dls_remove_log(data_log);
525
data_log->dlf_state = state;
528
xt_unlock_mutex_ns(&dlc_lock);
532
xt_unlock_mutex_ns(&dlc_lock);
536
static int dl_cmp_log_id(XTThreadPtr XT_UNUSED(self), register const void *XT_UNUSED(thunk), register const void *a, register const void *b)
538
xtLogID log_id_a = *((xtLogID *) a);
539
xtLogID log_id_b = *((xtLogID *) b);
541
if (log_id_a == log_id_b)
543
if (log_id_a < log_id_b)
548
void XTDataLogCache::dlc_init(XTThreadPtr self, XTDatabaseHPtr db)
551
char log_dir[PATH_MAX];
554
XTDataLogFilePtr data_log= NULL;
556
memset(this, 0, sizeof(XTDataLogCacheRec));
559
xt_init_mutex_with_autoname(self, &dlc_lock);
560
xt_init_cond(self, &dlc_cond);
561
for (u_int i=0; i<XT_DL_NO_OF_SEGMENTS; i++) {
562
xt_init_mutex_with_autoname(self, &dlc_segment[i].dls_lock);
563
xt_init_cond(self, &dlc_segment[i].dls_cond);
565
dlc_has_space = xt_new_sortedlist(self, sizeof(xtLogID), 20, 10, dl_cmp_log_id, NULL, NULL, FALSE, FALSE);
566
dlc_to_compact = xt_new_sortedlist(self, sizeof(xtLogID), 20, 10, dl_cmp_log_id, NULL, NULL, FALSE, FALSE);
567
dlc_to_delete = xt_new_sortedlist(self, sizeof(xtLogID), 20, 10, dl_cmp_log_id, NULL, NULL, FALSE, FALSE);
568
dlc_deleted = xt_new_sortedlist(self, sizeof(xtLogID), 20, 10, dl_cmp_log_id, NULL, NULL, FALSE, FALSE);
569
xt_init_mutex_with_autoname(self, &dlc_mru_lock);
570
xt_init_mutex_with_autoname(self, &dlc_head_lock);
572
xt_strcpy(PATH_MAX, log_dir, dlc_db->db_main_path);
573
xt_add_data_dir(PATH_MAX, log_dir);
574
if (xt_fs_exists(log_dir)) {
575
pushsr_(od, xt_dir_close, xt_dir_open(self, log_dir, NULL));
576
while (xt_dir_next(self, od)) {
577
file = xt_dir_name(self, od);
578
if (xt_ends_with(file, ".xt")) {
579
if ((log_id = (xtLogID) xt_file_name_to_id(file))) {
580
if (!dlc_get_data_log(&data_log, log_id, TRUE, NULL))
582
dl_recover_log(self, db, data_log);
583
if (!dls_set_log_state(data_log, XT_DL_UNKNOWN))
598
void XTDataLogCache::dlc_exit(XTThreadPtr self)
600
XTDataLogFilePtr data_log, tmp_data_log;
601
XTOpenLogFilePtr open_log, tmp_open_log;
604
xt_free_sortedlist(self, dlc_has_space);
605
dlc_has_space = NULL;
607
if (dlc_to_compact) {
608
xt_free_sortedlist(self, dlc_to_compact);
609
dlc_to_compact = NULL;
612
xt_free_sortedlist(self, dlc_to_delete);
613
dlc_to_delete = NULL;
616
xt_free_sortedlist(self, dlc_deleted);
619
for (u_int i=0; i<XT_DL_NO_OF_SEGMENTS; i++) {
620
for (u_int j=0; j<XT_DL_SEG_HASH_TABLE_SIZE; j++) {
621
data_log = dlc_segment[i].dls_hash_table[j];
622
dlc_segment[i].dls_hash_table[j] = NULL;
624
if (data_log->dlf_log_file) {
625
xt_close_file_ns(data_log->dlf_log_file);
626
data_log->dlf_log_file = NULL;
629
open_log = data_log->dlf_free_list;
631
if (open_log->odl_log_file)
632
xt_close_file(self, open_log->odl_log_file);
633
tmp_open_log = open_log;
634
open_log = open_log->odl_next_free;
635
xt_free(self, tmp_open_log);
637
tmp_data_log = data_log;
638
data_log = data_log->dlf_next_hash;
640
xt_free(self, tmp_data_log);
643
xt_free_mutex(&dlc_segment[i].dls_lock);
644
xt_free_cond(&dlc_segment[i].dls_cond);
646
xt_free_mutex(&dlc_head_lock);
647
xt_free_mutex(&dlc_mru_lock);
648
xt_free_mutex(&dlc_lock);
649
xt_free_cond(&dlc_cond);
652
void XTDataLogCache::dlc_name(size_t size, char *path, xtLogID log_id)
656
sprintf(name, "dlog-%lu.xt", (u_long) log_id);
657
xt_strcpy(size, path, dlc_db->db_main_path);
658
xt_add_data_dir(size, path);
659
xt_add_dir_char(size, path);
660
xt_strcat(size, path, name);
663
xtBool XTDataLogCache::dlc_open_log(XTOpenFilePtr *fh, xtLogID log_id, int mode)
665
char log_path[PATH_MAX];
667
dlc_name(PATH_MAX, log_path, log_id);
668
return xt_open_file_ns(fh, log_path, XT_FT_STANDARD, mode, 16*1024*1024);
671
xtBool XTDataLogCache::dlc_unlock_log(XTDataLogFilePtr data_log)
673
if (data_log->dlf_log_file) {
674
xt_close_file_ns(data_log->dlf_log_file);
675
data_log->dlf_log_file = NULL;
678
return dls_set_log_state(data_log, XT_DL_UNKNOWN);
681
XTDataLogFilePtr XTDataLogCache::dlc_get_log_for_writing(off_t space_required, struct XTThread *thread)
683
xtLogID log_id, *log_id_ptr = NULL;
686
XTDataLogFilePtr data_log = NULL;
688
xt_lock_mutex_ns(&dlc_lock);
690
/* Look for an existing log with enough space: */
691
size = xt_sl_get_size(dlc_has_space);
692
for (idx=0; idx<size; idx++) {
693
log_id_ptr = (xtLogID *) xt_sl_item_at(dlc_has_space, idx);
694
if (!dlc_get_data_log(&data_log, *log_id_ptr, FALSE, NULL))
697
if (data_log->dlf_space_avaliable() >= space_required)
703
xt_sl_delete_item_at(NULL, dlc_has_space, idx);
711
if (!dlc_open_log(&data_log->dlf_log_file, *log_id_ptr, XT_FS_DEFAULT))
713
xt_sl_delete_item_at(NULL, dlc_has_space, idx);
716
/* Create a new log: */
717
log_id = dlc_next_log_id;
718
for (u_int i=0; i<XT_DL_MAX_LOG_ID; i++) {
720
if (log_id > XT_DL_MAX_LOG_ID)
722
if (!dlc_get_data_log(&data_log, log_id, FALSE, NULL))
727
dlc_next_log_id = log_id;
729
xt_register_ulxterr(XT_REG_CONTEXT, XT_ERR_LOG_MAX_EXCEEDED, (u_long) XT_DL_MAX_LOG_ID);
732
if (!dlc_get_data_log(&data_log, log_id, TRUE, NULL))
734
if (!dlc_open_log(&data_log->dlf_log_file, log_id, XT_FS_CREATE | XT_FS_MAKE_PATH))
736
data_log->dlf_log_eof = sizeof(XTXactLogHeaderDRec);
737
if (!dl_create_log_header(data_log, data_log->dlf_log_file, thread)) {
738
xt_close_file_ns(data_log->dlf_log_file);
741
/* By setting this late we ensure that the error
744
dlc_next_log_id = log_id;
746
data_log->dlf_state = XT_DL_EXCLUSIVE;
748
xt_unlock_mutex_ns(&dlc_lock);
752
xt_unlock_mutex_ns(&dlc_lock);
756
xtBool XTDataLogCache::dlc_get_data_log(XTDataLogFilePtr *lf, xtLogID log_id, xtBool create, XTDataLogSegPtr *ret_seg)
758
register XTDataLogSegPtr seg;
759
register u_int hash_idx;
760
register XTDataLogFilePtr data_log;
762
/* Which segment, and hash index: */
763
seg = &dlc_segment[log_id & XT_DL_SEGMENT_MASK];
764
hash_idx = (log_id >> XT_DL_SEGMENT_SHIFTS) % XT_DL_SEG_HASH_TABLE_SIZE;
766
/* Lock the segment: */
767
xt_lock_mutex_ns(&seg->dls_lock);
769
/* Find the log file on the hash list: */
770
data_log = seg->dls_hash_table[hash_idx];
772
if (data_log->dlf_log_id == log_id)
774
data_log = data_log->dlf_next_hash;
777
if (!data_log && create) {
778
/* Create a new log file structure: */
779
if (!(data_log = (XTDataLogFilePtr) xt_calloc_ns(sizeof(XTDataLogFileRec))))
781
data_log->dlf_log_id = log_id;
782
data_log->dlf_next_hash = seg->dls_hash_table[hash_idx];
783
seg->dls_hash_table[hash_idx] = data_log;
787
/* This gives the caller the lock: */
793
xt_unlock_mutex_ns(&seg->dls_lock);
798
xt_unlock_mutex_ns(&seg->dls_lock);
803
* If just_close is FALSE, then a log is being deleted.
804
* This means that that the log may still be in exclusive use by
805
* some thread. So we just close the log!
807
xtBool XTDataLogCache::dlc_remove_data_log(xtLogID log_id, xtBool just_close)
809
register XTDataLogSegPtr seg;
810
register u_int hash_idx;
811
register XTDataLogFilePtr data_log;
812
XTOpenLogFilePtr open_log, tmp_open_log;
814
/* Which segment, and hash index: */
815
seg = &dlc_segment[log_id & XT_DL_SEGMENT_MASK];
816
hash_idx = (log_id >> XT_DL_SEGMENT_SHIFTS) % XT_DL_SEG_HASH_TABLE_SIZE;
818
/* Lock the segment: */
820
xt_lock_mutex_ns(&seg->dls_lock);
822
/* Find the log file on the hash list: */
823
data_log = seg->dls_hash_table[hash_idx];
825
if (data_log->dlf_log_id == log_id)
827
data_log = data_log->dlf_next_hash;
831
xt_lock_mutex_ns(&dlc_mru_lock);
833
open_log = data_log->dlf_free_list;
835
if (open_log->odl_log_file)
836
xt_close_file_ns(open_log->odl_log_file);
838
/* Remove from MRU list: */
839
if (dlc_lru_open_log == open_log) {
840
dlc_lru_open_log = open_log->odl_mr_used;
841
ASSERT_NS(!open_log->odl_lr_used);
843
else if (open_log->odl_lr_used)
844
open_log->odl_lr_used->odl_mr_used = open_log->odl_mr_used;
845
if (dlc_mru_open_log == open_log) {
846
dlc_mru_open_log = open_log->odl_lr_used;
847
ASSERT_NS(!open_log->odl_mr_used);
849
else if (open_log->odl_mr_used)
850
open_log->odl_mr_used->odl_lr_used = open_log->odl_lr_used;
852
data_log->dlf_open_count--;
853
tmp_open_log = open_log;
854
open_log = open_log->odl_next_free;
855
xt_free_ns(tmp_open_log);
857
data_log->dlf_free_list = NULL;
859
xt_unlock_mutex_ns(&dlc_mru_lock);
861
if (data_log->dlf_open_count) {
862
if (!xt_timed_wait_cond_ns(&seg->dls_cond, &seg->dls_lock, 2000))
864
xt_unlock_mutex_ns(&seg->dls_lock);
868
/* Close the exclusive file if required: */
869
if (data_log->dlf_log_file) {
870
xt_close_file_ns(data_log->dlf_log_file);
871
data_log->dlf_log_file = NULL;
875
/* Remove the log from the hash list: */
876
XTDataLogFilePtr ptr, pptr = NULL;
878
ptr = seg->dls_hash_table[hash_idx];
883
ptr = ptr->dlf_next_hash;
886
if (ptr == data_log) {
888
pptr->dlf_next_hash = ptr->dlf_next_hash;
890
seg->dls_hash_table[hash_idx] = ptr->dlf_next_hash;
893
xt_free_ns(data_log);
897
xt_unlock_mutex_ns(&seg->dls_lock);
901
xt_unlock_mutex_ns(&seg->dls_lock);
905
xtBool XTDataLogCache::dlc_get_open_log(XTOpenLogFilePtr *ol, xtLogID log_id)
907
register XTDataLogSegPtr seg;
908
register u_int hash_idx;
909
register XTDataLogFilePtr data_log;
910
register XTOpenLogFilePtr open_log;
913
/* Which segment, and hash index: */
914
seg = &dlc_segment[log_id & XT_DL_SEGMENT_MASK];
915
hash_idx = (log_id >> XT_DL_SEGMENT_SHIFTS) % XT_DL_SEG_HASH_TABLE_SIZE;
917
/* Lock the segment: */
918
xt_lock_mutex_ns(&seg->dls_lock);
920
/* Find the log file on the hash list: */
921
data_log = seg->dls_hash_table[hash_idx];
923
if (data_log->dlf_log_id == log_id)
925
data_log = data_log->dlf_next_hash;
929
/* Create a new log file structure: */
930
dlc_name(PATH_MAX, path, log_id);
931
if (!xt_fs_exists(path)) {
932
xt_register_ixterr(XT_REG_CONTEXT, XT_ERR_DATA_LOG_NOT_FOUND, path);
935
if (!(data_log = (XTDataLogFilePtr) xt_calloc_ns(sizeof(XTDataLogFileRec))))
937
data_log->dlf_log_id = log_id;
938
data_log->dlf_next_hash = seg->dls_hash_table[hash_idx];
939
seg->dls_hash_table[hash_idx] = data_log;
942
if ((open_log = data_log->dlf_free_list)) {
943
/* Remove from the free list: */
944
if ((data_log->dlf_free_list = open_log->odl_next_free))
945
data_log->dlf_free_list->odl_prev_free = NULL;
947
/* This file has been most recently used: */
948
if (XT_TIME_DIFF(open_log->odl_ru_time, dlc_ru_now) > (XT_DL_LOG_POOL_SIZE >> 1)) {
949
/* Move to the front of the MRU list: */
950
xt_lock_mutex_ns(&dlc_mru_lock);
952
open_log->odl_ru_time = ++dlc_ru_now;
953
if (dlc_mru_open_log != open_log) {
954
/* Remove from the MRU list: */
955
if (dlc_lru_open_log == open_log) {
956
dlc_lru_open_log = open_log->odl_mr_used;
957
ASSERT_NS(!open_log->odl_lr_used);
959
else if (open_log->odl_lr_used)
960
open_log->odl_lr_used->odl_mr_used = open_log->odl_mr_used;
961
if (open_log->odl_mr_used)
962
open_log->odl_mr_used->odl_lr_used = open_log->odl_lr_used;
964
/* Make the file the most recently used: */
965
if ((open_log->odl_lr_used = dlc_mru_open_log))
966
dlc_mru_open_log->odl_mr_used = open_log;
967
open_log->odl_mr_used = NULL;
968
dlc_mru_open_log = open_log;
969
if (!dlc_lru_open_log)
970
dlc_lru_open_log = open_log;
972
xt_unlock_mutex_ns(&dlc_mru_lock);
976
/* Create a new open file: */
977
if (!(open_log = (XTOpenLogFilePtr) xt_calloc_ns(sizeof(XTOpenLogFileRec))))
979
dlc_name(PATH_MAX, path, log_id);
980
if (!xt_open_file_ns(&open_log->odl_log_file, path, XT_FT_STANDARD, XT_FS_DEFAULT, 16*1024*1204)) {
981
xt_free_ns(open_log);
984
open_log->olf_log_id = log_id;
985
open_log->odl_data_log = data_log;
986
data_log->dlf_open_count++;
988
/* Make the new open file the most recently used: */
989
xt_lock_mutex_ns(&dlc_mru_lock);
990
open_log->odl_ru_time = ++dlc_ru_now;
991
if ((open_log->odl_lr_used = dlc_mru_open_log))
992
dlc_mru_open_log->odl_mr_used = open_log;
993
open_log->odl_mr_used = NULL;
994
dlc_mru_open_log = open_log;
995
if (!dlc_lru_open_log)
996
dlc_lru_open_log = open_log;
998
xt_unlock_mutex_ns(&dlc_mru_lock);
1001
open_log->odl_in_use = TRUE;
1002
xt_unlock_mutex_ns(&seg->dls_lock);
1005
if (dlc_open_count > XT_DL_LOG_POOL_SIZE) {
1006
u_int target = XT_DL_LOG_POOL_SIZE / 4 * 3;
1007
xtLogID free_log_id;
1009
/* Remove some open files: */
1010
while (dlc_open_count > target) {
1011
XTOpenLogFilePtr to_free = dlc_lru_open_log;
1013
if (!to_free || to_free->odl_in_use)
1016
/* Dirty read the file ID: */
1017
free_log_id = to_free->olf_log_id;
1019
seg = &dlc_segment[free_log_id & XT_DL_SEGMENT_MASK];
1021
/* Lock the segment: */
1022
xt_lock_mutex_ns(&seg->dls_lock);
1024
/* Lock the MRU list: */
1025
xt_lock_mutex_ns(&dlc_mru_lock);
1027
/* Check if we have the same open file: */
1028
if (dlc_lru_open_log == to_free && !to_free->odl_in_use) {
1029
data_log = to_free->odl_data_log;
1031
/* Remove from the MRU list: */
1032
dlc_lru_open_log = to_free->odl_mr_used;
1033
ASSERT_NS(!to_free->odl_lr_used);
1035
if (dlc_mru_open_log == to_free) {
1036
dlc_mru_open_log = to_free->odl_lr_used;
1037
ASSERT_NS(!to_free->odl_mr_used);
1039
else if (to_free->odl_mr_used)
1040
to_free->odl_mr_used->odl_lr_used = to_free->odl_lr_used;
1042
/* Remove from the free list of the file: */
1043
if (data_log->dlf_free_list == to_free) {
1044
data_log->dlf_free_list = to_free->odl_next_free;
1045
ASSERT_NS(!to_free->odl_prev_free);
1047
else if (to_free->odl_prev_free)
1048
to_free->odl_prev_free->odl_next_free = to_free->odl_next_free;
1049
if (to_free->odl_next_free)
1050
to_free->odl_next_free->odl_prev_free = to_free->odl_prev_free;
1051
ASSERT_NS(data_log->dlf_open_count > 0);
1052
data_log->dlf_open_count--;
1058
xt_unlock_mutex_ns(&dlc_mru_lock);
1059
xt_unlock_mutex_ns(&seg->dls_lock);
1062
xt_close_file_ns(to_free->odl_log_file);
1063
xt_free_ns(to_free);
1071
xt_unlock_mutex_ns(&seg->dls_lock);
1075
void XTDataLogCache::dlc_release_open_log(XTOpenLogFilePtr open_log)
1077
register XTDataLogSegPtr seg;
1078
register XTDataLogFilePtr data_log = open_log->odl_data_log;
1080
/* Which segment, and hash index: */
1081
seg = &dlc_segment[open_log->olf_log_id & XT_DL_SEGMENT_MASK];
1083
xt_lock_mutex_ns(&seg->dls_lock);
1084
open_log->odl_next_free = data_log->dlf_free_list;
1085
open_log->odl_prev_free = NULL;
1086
if (data_log->dlf_free_list)
1087
data_log->dlf_free_list->odl_prev_free = open_log;
1088
data_log->dlf_free_list = open_log;
1089
open_log->odl_in_use = FALSE;
1091
/* Wakeup any exclusive lockers: */
1092
if (!xt_broadcast_cond_ns(&seg->dls_cond))
1093
xt_log_and_clear_exception_ns();
1095
xt_unlock_mutex_ns(&seg->dls_lock);
1099
* --------------------------------------------------------------------------------
1100
* D A T A L O G F I L E
1103
off_t XTDataLogFile::dlf_space_avaliable()
1105
if (dlf_log_eof < xt_db_data_log_threshold)
1106
return xt_db_data_log_threshold - dlf_log_eof;
1110
xtBool XTDataLogFile::dlf_to_much_garbage()
1114
return dlf_garbage_count * 100 / dlf_log_eof >= xt_db_garbage_threshold;
1118
* --------------------------------------------------------------------------------
1119
* D A T A L O G B U F F E R
1122
void XTDataLogBuffer::dlb_init(XTDatabaseHPtr db, size_t buffer_size)
1125
ASSERT_NS(!dlb_buffer_size);
1126
ASSERT_NS(!dlb_data_log);
1127
ASSERT_NS(!dlb_log_buffer);
1129
dlb_buffer_size = buffer_size;
1132
void XTDataLogBuffer::dlb_exit(XTThreadPtr self)
1134
dlb_close_log(self);
1135
if (dlb_log_buffer) {
1136
xt_free(self, dlb_log_buffer);
1137
dlb_log_buffer = NULL;
1140
dlb_buffer_offset = 0;
1141
dlb_buffer_size = 0;
1143
dlb_flush_required = FALSE;
1145
dlb_max_write_offset = 0;
1149
xtBool XTDataLogBuffer::dlb_close_log(XTThreadPtr thread)
1152
if (dlb_data_log->dlf_log_file) {
1153
if (!dl_write_log_header(dlb_data_log, dlb_data_log->dlf_log_file, 0, thread))
1157
/* Flush and commit the data in the old log: */
1158
if (!dlb_flush_log(TRUE, thread))
1161
if (!dlb_db->db_datalogs.dlc_unlock_log(dlb_data_log))
1163
dlb_data_log = NULL;
1168
/* When I use 'thread' instead of 'self', this means
1169
* that I will not throw an error.
1171
xtBool XTDataLogBuffer::dlb_get_log_offset(xtLogID *log_id, xtLogOffset *out_offset, size_t XT_UNUSED(req_size), struct XTThread *thread)
1173
/* Note, I am allowing a log to grow beyond the threshold.
1174
* The amount depends on the maximum extended record size.
1175
* If I don't some logs will never fill up, because of only having
1176
* a few more bytes available.
1178
if (!dlb_data_log || dlb_data_log->dlf_space_avaliable() == 0) {
1179
/* Release the old log: */
1180
if (!dlb_close_log(thread))
1183
if (!dlb_log_buffer) {
1184
if (!(dlb_log_buffer = (xtWord1 *) xt_malloc_ns(dlb_buffer_size)))
1188
/* I could use req_size instead of 1, but this would mean some logs
1189
* are never filled up.
1191
if (!(dlb_data_log = dlb_db->db_datalogs.dlc_get_log_for_writing(1, thread)))
1194
dlb_max_write_offset = dlb_data_log->dlf_log_eof;
1198
*log_id = dlb_data_log->dlf_log_id;
1199
*out_offset = dlb_data_log->dlf_log_eof;
1203
xtBool XTDataLogBuffer::dlb_flush_log(xtBool commit, XTThreadPtr thread)
1205
if (!dlb_data_log || !dlb_data_log->dlf_log_file)
1208
if (dlb_buffer_len) {
1209
if (!xt_pwrite_file(dlb_data_log->dlf_log_file, dlb_buffer_offset, dlb_buffer_len, dlb_log_buffer, &thread->st_statistics.st_data, thread))
1212
if (dlb_buffer_offset + (xtLogOffset) dlb_buffer_len > dlb_max_write_offset)
1213
dlb_max_write_offset = dlb_buffer_offset + (xtLogOffset) dlb_buffer_len;
1216
dlb_flush_required = TRUE;
1219
if (commit && dlb_flush_required) {
1221
/* This would normally be equal, however, in the case
1222
* where some other thread flushes the compactors
1223
* data log, the eof, can be greater than the
1226
* This occurs because the flush can come between the
1227
* dlb_get_log_offset() and dlb_write_thru_log() calls.
1229
ASSERT_NS(dlb_data_log->dlf_log_eof >= dlb_max_write_offset);
1231
if (!xt_flush_file(dlb_data_log->dlf_log_file, &thread->st_statistics.st_data, thread))
1233
dlb_flush_required = FALSE;
1238
//#define INJECT_ERROR
1244
xtBool XTDataLogBuffer::dlb_write_thru_log(xtLogID XT_NDEBUG_UNUSED(log_id), xtLogOffset log_offset, size_t size, xtWord1 *data, XTThreadPtr thread)
1246
ASSERT_NS(log_id == dlb_data_log->dlf_log_id);
1249
dlb_flush_log(FALSE, thread);
1253
if (inject_when > 1000 && inject_when < 1002)
1254
return xt_register_ferrno(XT_REG_CONTEXT, ESPIPE, xt_file_path(dlb_data_log->dlf_log_file));
1256
if (!xt_pwrite_file(dlb_data_log->dlf_log_file, log_offset, size, data, &thread->st_statistics.st_data, thread))
1258
/* Increment of dlb_data_log->dlf_log_eof was moved here from dlb_get_log_offset()
1259
* to ensure it is done after a successful update of the log, otherwise otherwise a
1260
* gap occurs in the log which cause eof to be detected in middle of the log
1262
dlb_data_log->dlf_log_eof += size;
1264
if (log_offset + (xtLogOffset) size > (xtLogOffset) dlb_max_write_offset)
1265
dlb_max_write_offset = log_offset + size;
1267
dlb_flush_required = TRUE;
1271
xtBool XTDataLogBuffer::dlb_append_log(xtLogID XT_NDEBUG_UNUSED(log_id), xtLogOffset log_offset, size_t size, xtWord1 *data, XTThreadPtr thread)
1273
ASSERT_NS(log_id == dlb_data_log->dlf_log_id);
1275
if (dlb_buffer_len) {
1276
/* Should be the case, we only write by appending: */
1277
ASSERT_NS(dlb_buffer_offset + (xtLogOffset) dlb_buffer_len == log_offset);
1278
/* Check if we are appending to the existing value in the buffer: */
1279
if (dlb_buffer_offset + (xtLogOffset) dlb_buffer_len == log_offset) {
1280
/* Can we just append: */
1281
if (dlb_buffer_size >= dlb_buffer_len + size) {
1282
memcpy(dlb_log_buffer + dlb_buffer_len, data, size);
1283
dlb_buffer_len += size;
1284
dlb_data_log->dlf_log_eof += size;
1288
if (dlb_flush_log(FALSE, thread) != OK)
1292
ASSERT_NS(dlb_buffer_len == 0);
1294
if (dlb_buffer_size >= size) {
1295
dlb_buffer_offset = log_offset;
1296
dlb_buffer_len = size;
1297
memcpy(dlb_log_buffer, data, size);
1298
dlb_data_log->dlf_log_eof += size;
1302
/* Write directly: */
1303
if (!xt_pwrite_file(dlb_data_log->dlf_log_file, log_offset, size, data, &thread->st_statistics.st_data, thread))
1306
if (log_offset + (xtLogOffset) size > (xtLogOffset) dlb_max_write_offset)
1307
dlb_max_write_offset = log_offset + size;
1309
dlb_flush_required = TRUE;
1310
dlb_data_log->dlf_log_eof += size;
1314
xtBool XTDataLogBuffer::dlb_read_log(xtLogID log_id, xtLogOffset log_offset, size_t size, xtWord1 *data, XTThreadPtr thread)
1317
XTOpenLogFilePtr open_log;
1319
if (dlb_data_log && log_id == dlb_data_log->dlf_log_id) {
1320
/* Reading from the write log, I can do this quicker: */
1321
if (dlb_buffer_len) {
1322
/* If it is in the buffer, then it is completely in the buffer. */
1323
if (log_offset >= dlb_buffer_offset) {
1324
if (log_offset + (xtLogOffset) size <= dlb_buffer_offset + (xtLogOffset) dlb_buffer_len) {
1325
memcpy(data, dlb_log_buffer + (log_offset - dlb_buffer_offset), size);
1328
/* Should not happen, reading past EOF: */
1330
memset(data, 0, size);
1333
/* In the write log, but not in the buffer,
1334
* must be completely not in the log,
1335
* because only whole records are written to the
1338
ASSERT_NS(log_offset + (xtLogOffset) size <= dlb_buffer_offset);
1340
return xt_pread_file(dlb_data_log->dlf_log_file, log_offset, size, size, data, NULL, &thread->st_statistics.st_data, thread);
1343
/* Read from some other log: */
1344
if (!dlb_db->db_datalogs.dlc_get_open_log(&open_log, log_id))
1347
if (!xt_pread_file(open_log->odl_log_file, log_offset, size, 0, data, &red_size, &thread->st_statistics.st_data, thread)) {
1348
dlb_db->db_datalogs.dlc_release_open_log(open_log);
1352
dlb_db->db_datalogs.dlc_release_open_log(open_log);
1354
if (red_size < size)
1355
memset(data + red_size, 0, size - red_size);
1361
* We assume that the given reference may not be valid.
1362
* Only valid references actually cause a delete.
1363
* Invalid references are logged, and ignored.
1365
* Note this routine does not lock the compactor.
1366
* This can lead to the some incorrect calculation is the
1367
* amount of garbage. But nothing serious I think.
1369
xtBool XTDataLogBuffer::dlb_delete_log(xtLogID log_id, xtLogOffset log_offset, size_t size, xtTableID tab_id, xtRecordID rec_id, XTThreadPtr thread)
1371
XTactExtRecEntryDRec record;
1372
xtWord1 status = XT_LOG_ENT_EXT_REC_DEL;
1373
XTOpenLogFilePtr open_log;
1374
xtBool to_much_garbage;
1375
XTDataLogFilePtr data_log;
1377
if (!dlb_read_log(log_id, log_offset, offsetof(XTactExtRecEntryDRec, er_data), (xtWord1 *) &record, thread))
1380
/* Already deleted: */
1381
if (record.er_status_1 == XT_LOG_ENT_EXT_REC_DEL)
1384
if (record.er_status_1 != XT_LOG_ENT_EXT_REC_OK ||
1385
size != XT_GET_DISK_4(record.er_data_size_4) ||
1386
tab_id != XT_GET_DISK_4(record.er_tab_id_4) ||
1387
rec_id != XT_GET_DISK_4(record.er_rec_id_4)) {
1388
xt_register_xterr(XT_REG_CONTEXT, XT_ERR_BAD_EXT_RECORD);
1392
if (dlb_data_log && log_id == dlb_data_log->dlf_log_id) {
1393
/* Writing to the write log, I can do this quicker: */
1394
if (dlb_buffer_len) {
1395
/* If it is in the buffer, then it is completely in the buffer. */
1396
if (log_offset >= dlb_buffer_offset) {
1397
if (log_offset + 1 <= dlb_buffer_offset + (xtLogOffset) dlb_buffer_len) {
1398
*(dlb_log_buffer + (log_offset - dlb_buffer_offset)) = XT_LOG_ENT_EXT_REC_DEL;
1399
goto inc_garbage_count;
1401
/* Should not happen, writing past EOF: */
1405
ASSERT_NS(log_offset + (xtLogOffset) size <= dlb_buffer_offset);
1408
if (!xt_pwrite_file(dlb_data_log->dlf_log_file, log_offset, 1, &status, &thread->st_statistics.st_data, thread))
1412
xt_lock_mutex_ns(&dlb_db->db_datalogs.dlc_head_lock);
1413
dlb_data_log->dlf_garbage_count += offsetof(XTactExtRecEntryDRec, er_data) + size;
1414
ASSERT_NS(dlb_data_log->dlf_garbage_count < dlb_data_log->dlf_log_eof);
1415
if (!dl_write_garbage_level(dlb_data_log, dlb_data_log->dlf_log_file, FALSE, thread)) {
1416
xt_unlock_mutex_ns(&dlb_db->db_datalogs.dlc_head_lock);
1419
dlb_flush_required = TRUE;
1420
xt_unlock_mutex_ns(&dlb_db->db_datalogs.dlc_head_lock);
1424
/* Write to some other log, open the log: */
1425
if (!dlb_db->db_datalogs.dlc_get_open_log(&open_log, log_id))
1428
/* Write the status byte: */
1429
if (!xt_pwrite_file(open_log->odl_log_file, log_offset, 1, &status, &thread->st_statistics.st_data, thread))
1432
data_log = open_log->odl_data_log;
1434
/* Adjust the garbage level in the header. */
1435
xt_lock_mutex_ns(&dlb_db->db_datalogs.dlc_head_lock);
1436
data_log->dlf_garbage_count += offsetof(XTactExtRecEntryDRec, er_data) + size;
1437
ASSERT_NS(data_log->dlf_garbage_count < data_log->dlf_log_eof);
1438
if (!dl_write_garbage_level(data_log, open_log->odl_log_file, FALSE, thread)) {
1439
xt_unlock_mutex_ns(&dlb_db->db_datalogs.dlc_head_lock);
1442
to_much_garbage = data_log->dlf_to_much_garbage();
1443
xt_unlock_mutex_ns(&dlb_db->db_datalogs.dlc_head_lock);
1445
if (to_much_garbage &&
1446
(data_log->dlf_state == XT_DL_HAS_SPACE || data_log->dlf_state == XT_DL_READ_ONLY)) {
1447
/* There is too much garbage, it may be compacted. */
1448
if (!dlb_db->db_datalogs.dls_set_log_state(data_log, XT_DL_MAY_COMPACT))
1452
/* Release the open log: */
1453
dlb_db->db_datalogs.dlc_release_open_log(open_log);
1458
dlb_db->db_datalogs.dlc_release_open_log(open_log);
1463
* Delete all the extended data belonging to a particular
1466
xtPublic void xt_dl_delete_ext_data(XTThreadPtr self, XTTableHPtr tab, xtBool XT_UNUSED(missing_ok), xtBool have_table_lock)
1469
xtRecordID page_rec_id, offs_rec_id;
1470
XTTabRecExtDPtr rec_buf;
1471
xtWord4 log_over_size;
1473
xtLogOffset log_offset;
1476
page_data = (xtWord1 *) xt_malloc(self, tab->tab_recs.tci_page_size);
1477
pushr_(xt_free, page_data);
1479
/* Scan the table, and remove all exended data... */
1480
if (!(ot = xt_open_table(tab))) {
1481
if (self->t_exception.e_xt_err == XT_SYSTEM_ERROR &&
1482
XT_FILE_NOT_FOUND(self->t_exception.e_sys_err))
1486
ot->ot_thread = self;
1488
/* {LOCK-EXT-REC} This lock is to stop the compactor changing records
1489
* while we are doing the delete.
1491
xt_lock_mutex_ns(&tab->tab_db->db_co_ext_lock);
1494
while (page_rec_id < tab->tab_rec_eof_id) {
1495
/* NOTE: There is a good reason for using xt_tc_read_page().
1496
* A deadlock can occur if using read, which can run out of
1497
* memory, which waits for the freeer, which may need to
1498
* open a table, which requires the db->db_tables lock,
1499
* which is owned by the this thread, when the function
1500
* is called from drop table.
1502
* xt_tc_read_page() should work because no more changes
1503
* should happen to the table while we are dropping it.
1505
if (!tab->tab_recs.xt_tc_read_page(ot->ot_rec_file, page_rec_id, page_data, self))
1508
for (offs_rec_id=0; offs_rec_id<tab->tab_recs.tci_rows_per_page && page_rec_id+offs_rec_id < tab->tab_rec_eof_id; offs_rec_id++) {
1509
rec_buf = (XTTabRecExtDPtr) (page_data + (offs_rec_id * tab->tab_recs.tci_rec_size));
1510
if (XT_REC_IS_EXT_DLOG(rec_buf->tr_rec_type_1)) {
1511
log_over_size = XT_GET_DISK_4(rec_buf->re_log_dat_siz_4);
1512
XT_GET_LOG_REF(log_id, log_offset, rec_buf);
1514
if (tab->tab_dic.dic_tab_flags & XT_TF_MEMORY_TABLE)
1515
xt_tab_free_ext_slot(tab, log_id, log_offset, log_over_size);
1517
if (!self->st_dlog_buf.dlb_delete_log(log_id, log_offset, log_over_size, tab->tab_id, page_rec_id+offs_rec_id, self)) {
1518
if (self->t_exception.e_xt_err != XT_ERR_BAD_EXT_RECORD &&
1519
self->t_exception.e_xt_err != XT_ERR_DATA_LOG_NOT_FOUND)
1520
xt_log_and_clear_exception(self);
1526
page_rec_id += tab->tab_recs.tci_rows_per_page;
1529
xt_unlock_mutex_ns(&tab->tab_db->db_co_ext_lock);
1531
xt_close_table(ot, TRUE, have_table_lock);
1533
freer_(); // xt_free(page_data)
1537
xt_unlock_mutex_ns(&tab->tab_db->db_co_ext_lock);
1539
xt_close_table(ot, TRUE, have_table_lock);
1544
* --------------------------------------------------------------------------------
1545
* GARBAGE COLLECTOR THREAD
1548
xtPublic void xt_dl_init_db(XTThreadPtr self, XTDatabaseHPtr db)
1550
xt_init_mutex_with_autoname(self, &db->db_co_ext_lock);
1551
xt_init_mutex_with_autoname(self, &db->db_co_dlog_lock);
1554
xtPublic void xt_dl_exit_db(XTThreadPtr self, XTDatabaseHPtr db)
1556
xt_stop_compactor(self, db); // Already done!
1557
db->db_co_thread = NULL;
1558
xt_free_mutex(&db->db_co_ext_lock);
1559
xt_free_mutex(&db->db_co_dlog_lock);
1562
xtPublic void xt_dl_set_to_delete(XTThreadPtr self, XTDatabaseHPtr db, xtLogID log_id)
1564
XTDataLogFilePtr data_log;
1566
if (!db->db_datalogs.dlc_get_data_log(&data_log, log_id, FALSE, NULL))
1569
if (!db->db_datalogs.dls_set_log_state(data_log, XT_DL_TO_DELETE))
1574
xtPublic void xt_dl_log_status(XTThreadPtr self, XTDatabaseHPtr db, XTStringBufferPtr strbuf)
1576
XTSortedListPtr list;
1577
XTDataLogFilePtr data_log;
1578
XTDataLogSegPtr seg;
1580
xtLogID *log_id_ptr;
1582
list = xt_new_sortedlist(self, sizeof(xtLogID), 20, 10, dl_cmp_log_id, NULL, NULL, FALSE, FALSE);
1583
pushr_(xt_free_sortedlist, list);
1585
for (u_int i=0; i<XT_DL_NO_OF_SEGMENTS; i++) {
1586
for (u_int j=0; j<XT_DL_SEG_HASH_TABLE_SIZE; j++) {
1587
seg = &db->db_datalogs.dlc_segment[i];
1588
data_log = seg->dls_hash_table[j];
1590
xt_sl_insert(self, list, &data_log->dlf_log_id, &data_log->dlf_log_id);
1591
data_log = data_log->dlf_next_hash;
1596
no_of_logs = xt_sl_get_size(list);
1597
for (u_int i=0; i<no_of_logs; i++) {
1598
log_id_ptr = (xtLogID *) xt_sl_item_at(list, i);
1599
if (!db->db_datalogs.dlc_get_data_log(&data_log, *log_id_ptr, FALSE, &seg))
1602
xt_sb_concat(self, strbuf, "d-log: ");
1603
xt_sb_concat_int8(self, strbuf, data_log->dlf_log_id);
1604
xt_sb_concat(self, strbuf, " status=");
1605
switch (data_log->dlf_state) {
1607
xt_sb_concat(self, strbuf, "?");
1609
case XT_DL_HAS_SPACE:
1610
xt_sb_concat(self, strbuf, "has-space ");
1612
case XT_DL_READ_ONLY:
1613
xt_sb_concat(self, strbuf, "read-only ");
1615
case XT_DL_TO_COMPACT:
1616
xt_sb_concat(self, strbuf, "to-compact");
1618
case XT_DL_COMPACTED:
1619
xt_sb_concat(self, strbuf, "compacted ");
1621
case XT_DL_TO_DELETE:
1622
xt_sb_concat(self, strbuf, "to-delete ");
1625
xt_sb_concat(self, strbuf, "deleted ");
1627
case XT_DL_EXCLUSIVE:
1628
xt_sb_concat(self, strbuf, "x-locked ");
1631
xt_sb_concat(self, strbuf, " eof=");
1632
xt_sb_concat_int8(self, strbuf, data_log->dlf_log_eof);
1633
xt_sb_concat(self, strbuf, " garbage=");
1634
xt_sb_concat_int8(self, strbuf, data_log->dlf_garbage_count);
1635
xt_sb_concat(self, strbuf, " g%=");
1636
if (data_log->dlf_log_eof)
1637
xt_sb_concat_int8(self, strbuf, data_log->dlf_garbage_count * 100 / data_log->dlf_log_eof);
1639
xt_sb_concat(self, strbuf, "100");
1640
xt_sb_concat(self, strbuf, " open=");
1641
xt_sb_concat_int8(self, strbuf, data_log->dlf_open_count);
1642
xt_sb_concat(self, strbuf, "\n");
1644
xt_unlock_mutex_ns(&seg->dls_lock);
1647
freer_(); // xt_free_sortedlist(list)
1650
xtPublic void xt_dl_delete_logs(XTThreadPtr self, XTDatabaseHPtr db)
1652
char path[PATH_MAX];
1657
xt_strcpy(PATH_MAX, path, db->db_main_path);
1658
xt_add_data_dir(PATH_MAX, path);
1659
if (!xt_fs_exists(path))
1661
pushsr_(od, xt_dir_close, xt_dir_open(self, path, NULL));
1662
while (xt_dir_next(self, od)) {
1663
file = xt_dir_name(self, od);
1664
if ((log_id = (xtLogID) xt_file_name_to_id(file))) {
1665
if (!db->db_datalogs.dlc_remove_data_log(log_id, TRUE))
1666
xt_log_and_clear_exception(self);
1668
if (xt_ends_with(file, ".xt")) {
1669
xt_add_dir_char(PATH_MAX, path);
1670
xt_strcat(PATH_MAX, path, file);
1671
xt_fs_delete(self, path);
1672
xt_remove_last_name_of_path(path);
1675
freer_(); // xt_dir_close(od)
1677
/* I no longer attach the condition: !db->db_multi_path
1678
* to removing this directory. This is because
1679
* the pbxt directory must now be removed explicitly
1680
* by drop database, or by delete all the PBXT
1683
if (!xt_fs_rmdir(NULL, path))
1684
xt_log_and_clear_exception(self);
1687
typedef struct XTCompactorState {
1688
XTSeqLogReadPtr cs_seqread;
1689
XTOpenTablePtr cs_ot;
1690
XTDataBufferRec cs_databuf;
1691
} XTCompactorStateRec, *XTCompactorStatePtr;
1693
static void dl_free_compactor_state(XTThreadPtr self, XTCompactorStatePtr cs)
1695
if (cs->cs_seqread) {
1696
cs->cs_seqread->sl_seq_exit();
1697
delete cs->cs_seqread;
1698
cs->cs_seqread = NULL;
1701
xt_db_return_table_to_pool(self, cs->cs_ot);
1704
xt_db_set_size(self, &cs->cs_databuf, 0);
1707
static XTOpenTablePtr dl_cs_get_open_table(XTThreadPtr self, XTCompactorStatePtr cs, xtTableID tab_id)
1710
if (cs->cs_ot->ot_table->tab_id == tab_id)
1713
xt_db_return_table_to_pool(self, cs->cs_ot);
1718
if (!(cs->cs_ot = xt_db_open_pool_table(self, self->st_database, tab_id, NULL, TRUE)))
1725
static void dl_co_wait(XTThreadPtr self, XTDatabaseHPtr db, u_int secs)
1727
xt_lock_mutex(self, &db->db_datalogs.dlc_lock);
1728
pushr_(xt_unlock_mutex, &db->db_datalogs.dlc_lock);
1730
xt_timed_wait_cond(self, &db->db_datalogs.dlc_cond, &db->db_datalogs.dlc_lock, secs * 1000);
1731
freer_(); // xt_unlock_mutex(&db->db_datalogs.dlc_lock)
1735
* Collect all the garbage in a file by moving all valid records
1736
* into some other data log and updating the handles.
1738
static xtBool dl_collect_garbage(XTThreadPtr self, XTDatabaseHPtr db, XTDataLogFilePtr data_log)
1740
XTXactLogBufferDPtr record;
1744
XTCompactorStateRec cs;
1747
XTTabRecExtDRec rec_buffer;
1750
xtLogOffset src_log_offset;
1751
xtLogID curr_log_id;
1752
xtLogOffset curr_log_offset;
1753
xtLogID dest_log_id = 0;
1754
xtLogOffset dest_log_offset = 0;
1755
off_t garbage_count = 0;
1757
memset(&cs, 0, sizeof(XTCompactorStateRec));
1759
if (!(cs.cs_seqread = new XTDataSeqRead()))
1760
xt_throw_errno(XT_CONTEXT, XT_ENOMEM);
1762
if (!cs.cs_seqread->sl_seq_init(db, xt_db_log_buffer_size)) {
1763
delete cs.cs_seqread;
1766
pushr_(dl_free_compactor_state, &cs);
1768
if (!cs.cs_seqread->sl_seq_start(data_log->dlf_log_id, data_log->dlf_start_offset, FALSE))
1773
/* Flush the destination log: */
1774
xt_lock_mutex(self, &db->db_co_dlog_lock);
1775
pushr_(xt_unlock_mutex, &db->db_co_dlog_lock);
1776
if (!self->st_dlog_buf.dlb_flush_log(TRUE, self))
1778
freer_(); // xt_unlock_mutex(&db->db_co_dlog_lock)
1780
/* Flush the transaction log. */
1781
if (!xt_xlog_flush_log(db, self))
1784
xt_lock_mutex_ns(&db->db_datalogs.dlc_head_lock);
1785
data_log->dlf_garbage_count += garbage_count;
1786
ASSERT(data_log->dlf_garbage_count < data_log->dlf_log_eof);
1787
if (!dl_write_garbage_level(data_log, cs.cs_seqread->sl_seq_open_file(), TRUE, self)) {
1788
xt_unlock_mutex_ns(&db->db_datalogs.dlc_head_lock);
1791
xt_unlock_mutex_ns(&db->db_datalogs.dlc_head_lock);
1793
freer_(); // dl_free_compactor_state(&cs)
1796
if (!cs.cs_seqread->sl_seq_next(&record, self))
1798
cs.cs_seqread->sl_seq_pos(&curr_log_id, &curr_log_offset);
1800
data_log->dlf_start_offset = curr_log_offset;
1803
switch (record->xh.xh_status_1) {
1804
case XT_LOG_ENT_EXT_REC_OK:
1805
size = XT_GET_DISK_4(record->er.er_data_size_4);
1806
tab_id = XT_GET_DISK_4(record->er.er_tab_id_4);
1807
rec_id = XT_GET_DISK_4(record->er.er_rec_id_4);
1809
if (!(ot = dl_cs_get_open_table(self, &cs, tab_id)))
1813
/* All this is required for a valid record address: */
1814
if (!rec_id || rec_id >= tab->tab_rec_eof_id)
1817
/* {LOCK-EXT-REC} It is important to prevent the compactor from modifying
1818
* a record that has been freed (and maybe allocated again).
1820
* Consider the following sequence:
1822
* 1. Compactor reads the record.
1823
* 2. The record is freed and reallocated.
1824
* 3. The compactor updates the record.
1826
* To prevent this, the compactor locks out the
1827
* sweeper using the db_co_ext_lock lock. The db_co_ext_lock lock
1828
* prevents a extended record from being moved and removed at the
1831
* The compactor also checks the status of the record before
1834
xt_lock_mutex(self, &db->db_co_ext_lock);
1835
pushr_(xt_unlock_mutex, &db->db_co_ext_lock);
1837
/* Read the record: */
1838
if (!xt_tab_get_rec_data(ot, rec_id, offsetof(XTTabRecExtDRec, re_data), (xtWord1 *) &rec_buffer)) {
1839
xt_log_and_clear_warning(self);
1840
freer_(); // xt_unlock_mutex(&db->db_co_ext_lockk)
1844
/* [(7)] REMOVE is followed by FREE:
1845
if (XT_REC_IS_REMOVED(rec_buffer.tr_rec_type_1) || !XT_REC_IS_EXT_DLOG(rec_buffer.tr_rec_type_1)) {
1847
if (!XT_REC_IS_EXT_DLOG(rec_buffer.tr_rec_type_1)) {
1848
freer_(); // xt_unlock_mutex(&db->db_co_ext_lock)
1852
XT_GET_LOG_REF(src_log_id, src_log_offset, &rec_buffer);
1853
src_size = (size_t) XT_GET_DISK_4(rec_buffer.re_log_dat_siz_4);
1855
/* Does the record agree with the current position: */
1856
if (curr_log_id != src_log_id ||
1857
curr_log_offset != src_log_offset ||
1859
freer_(); // xt_unlock_mutex(&db->db_co_ext_lock)
1863
size = offsetof(XTactExtRecEntryDRec, er_data) + size;
1865
/* Allocate space in a destination log: */
1866
xt_lock_mutex(self, &db->db_co_dlog_lock);
1867
pushr_(xt_unlock_mutex, &db->db_co_dlog_lock);
1868
if (!self->st_dlog_buf.dlb_get_log_offset(&dest_log_id, &dest_log_offset, size, self))
1870
freer_(); // xt_unlock_mutex(&db->db_co_dlog_lock)
1872
/* This record is referenced by the data: */
1873
xt_db_set_size(self, &cs.cs_databuf, size);
1874
if (!cs.cs_seqread->sl_rnd_read(src_log_offset, size, cs.cs_databuf.db_data, NULL, self))
1877
/* The problem with writing to the buffer here, is that other
1878
* threads want to read the data! */
1879
xt_lock_mutex(self, &db->db_co_dlog_lock);
1880
pushr_(xt_unlock_mutex, &db->db_co_dlog_lock);
1881
if (!self->st_dlog_buf.dlb_write_thru_log(dest_log_id, dest_log_offset, size, cs.cs_databuf.db_data, self))
1883
freer_(); // xt_unlock_mutex(&db->db_co_dlog_lock)
1885
/* Make sure we flush the compactor target log, before we
1886
* flush the transaction log!!
1887
* This is done here [(8)]
1890
XT_SET_LOG_REF(&rec_buffer, dest_log_id, dest_log_offset);
1892
if (!xt_tab_put_log_rec_data(ot, XT_LOG_ENT_REC_MOVED, 0, rec_id, 8, (xtWord1 *) &rec_buffer.re_log_id_2, &op_seq))
1894
tab->tab_co_op_seq = op_seq;
1896
/* Only records that were actually moved, count as garbage now!
1897
* This means, lost records, remain "lost" as far as the garbage
1898
* count is concerned!
1900
garbage_count += size;
1901
freer_(); // xt_unlock_mutex(&db->db_co_ext_lock)
1904
data_log->dlf_start_offset = curr_log_offset;
1907
/* Flush the distination log. */
1908
xt_lock_mutex(self, &db->db_co_dlog_lock);
1909
pushr_(xt_unlock_mutex, &db->db_co_dlog_lock);
1910
if (!self->st_dlog_buf.dlb_flush_log(TRUE, self))
1912
freer_(); // xt_unlock_mutex(&db->db_co_dlog_lock)
1914
/* Flush the transaction log. */
1915
if (!xt_xlog_flush_log(db, self))
1918
/* Save state in source log header. */
1919
xt_lock_mutex_ns(&db->db_datalogs.dlc_head_lock);
1920
data_log->dlf_garbage_count += garbage_count;
1921
ASSERT(data_log->dlf_garbage_count < data_log->dlf_log_eof);
1922
if (!dl_write_garbage_level(data_log, cs.cs_seqread->sl_seq_open_file(), TRUE, self)) {
1923
xt_unlock_mutex_ns(&db->db_datalogs.dlc_head_lock);
1926
xt_unlock_mutex_ns(&db->db_datalogs.dlc_head_lock);
1928
/* Wait for the writer to write all the changes.
1929
* Then we can start the delete process for the log:
1931
* Note, if we do not wait, then it could be some operations are held up,
1932
* by being out of sequence. This could cause the log to be deleted
1933
* before all the operations have been performed (which are on a table
1939
XTTableEntryPtr tab_ptr;
1943
freer_(); // dl_free_compactor_state(&cs)
1947
xt_ht_lock(self, db->db_tables);
1948
pushr_(xt_ht_unlock, db->db_tables);
1949
xt_enum_tables_init(&edx);
1950
while ((tab_ptr = xt_enum_tables_next(self, db, &edx))) {
1951
if (tab_ptr->te_table && tab_ptr->te_table->tab_co_op_seq > tab_ptr->te_table->tab_head_op_seq) {
1956
freer_(); // xt_ht_unlock(db->db_tables)
1961
/* Nobody will wake me, so check again shortly! */
1962
dl_co_wait(self, db, 1);
1965
db->db_datalogs.dls_set_log_state(data_log, XT_DL_COMPACTED);
1967
#ifdef DEBUG_LOG_DELETE
1968
printf("-- MARK FOR DELETE IN LOG: %d\n", (int) data_log->dlf_log_id);
1970
/* Log that this log should be deleted on the next checkpoint: */
1971
// transaction log...
1972
XTXactNewLogEntryDRec log_rec;
1973
log_rec.xl_status_1 = XT_LOG_ENT_DEL_LOG;
1974
log_rec.xl_checksum_1 = XT_CHECKSUM_1(data_log->dlf_log_id);
1975
XT_SET_DISK_4(log_rec.xl_log_id_4, data_log->dlf_log_id);
1976
if (!xt_xlog_log_data(self, sizeof(XTXactNewLogEntryDRec), (XTXactLogBufferDPtr) &log_rec, XT_XLOG_WRITE_AND_FLUSH)) {
1977
db->db_datalogs.dls_set_log_state(data_log, XT_DL_TO_COMPACT);
1981
freer_(); // dl_free_compactor_state(&cs)
1985
static void dl_co_not_busy(XTThreadPtr XT_UNUSED(self), XTDatabaseHPtr db)
1987
db->db_co_busy = FALSE;
1990
static void dl_co_main(XTThreadPtr self, xtBool once_off)
1992
XTDatabaseHPtr db = self->st_database;
1993
xtLogID *log_id_ptr, log_id;
1994
XTDataLogFilePtr data_log = NULL;
1996
xt_set_low_priority(self);
1998
while (!self->t_quit) {
1999
while (!self->t_quit) {
2000
xt_lock_mutex_ns(&db->db_datalogs.dlc_lock);
2001
if ((log_id_ptr = (xtLogID *) xt_sl_first_item(db->db_datalogs.dlc_to_compact))) {
2002
log_id = *log_id_ptr;
2006
xt_unlock_mutex_ns(&db->db_datalogs.dlc_lock);
2009
if (!db->db_datalogs.dlc_get_data_log(&data_log, log_id, FALSE, NULL))
2013
db->db_co_busy = TRUE;
2014
pushr_(dl_co_not_busy, db);
2015
dl_collect_garbage(self, db, data_log);
2016
freer_(); // dl_co_not_busy(db)
2019
xt_lock_mutex_ns(&db->db_datalogs.dlc_lock);
2020
xt_sl_delete(self, db->db_datalogs.dlc_to_compact, &log_id);
2021
xt_unlock_mutex_ns(&db->db_datalogs.dlc_lock);
2028
/* Wait for a signal that a data log can be collected: */
2029
dl_co_wait(self, db, 120);
2033
static void *dl_run_co_thread(XTThreadPtr self)
2035
XTDatabaseHPtr db = (XTDatabaseHPtr) self->t_data;
2039
if (!(mysql_thread = myxt_create_thread()))
2042
while (!self->t_quit) {
2045
* The garbage collector requires that the database
2046
* is in use because.
2048
xt_use_database(self, db, XT_FOR_COMPACTOR);
2050
/* {BACKGROUND-RELEASE-DB}
2051
* This action is both safe and required:
2053
* safe: releasing the database is safe because as
2054
* long as this thread is running the database
2055
* reference is valid, and this reference cannot
2056
* be the only one to the database because
2057
* otherwize this thread would not be running.
2059
* required: releasing the database is necessary
2060
* otherwise we cannot close the database
2061
* correctly because we only shutdown this
2062
* thread when the database is closed and we
2063
* only close the database when all references
2066
xt_heap_release(self, self->st_database);
2068
dl_co_main(self, FALSE);
2071
if (!(self->t_exception.e_xt_err == XT_SIGNAL_CAUGHT &&
2072
self->t_exception.e_sys_err == SIGTERM))
2073
xt_log_and_clear_exception(self);
2077
/* Avoid releasing the database (done above) */
2078
self->st_database = NULL;
2079
xt_unuse_database(self, self);
2081
/* After an exception, pause before trying again... */
2082
/* Number of seconds */
2088
while (!self->t_quit && count > 0) {
2095
* {MYSQL-THREAD-KILL}
2096
myxt_destroy_thread(mysql_thread, TRUE);
2101
static void dl_free_co_thread(XTThreadPtr self, void *data)
2103
XTDatabaseHPtr db = (XTDatabaseHPtr) data;
2105
if (db->db_co_thread) {
2106
xt_lock_mutex(self, &db->db_datalogs.dlc_lock);
2107
pushr_(xt_unlock_mutex, &db->db_datalogs.dlc_lock);
2108
db->db_co_thread = NULL;
2109
freer_(); // xt_unlock_mutex(&db->db_datalogs.dlc_lock)
2113
xtPublic void xt_start_compactor(XTThreadPtr self, XTDatabaseHPtr db)
2115
char name[PATH_MAX];
2117
sprintf(name, "GC-%s", xt_last_directory_of_path(db->db_main_path));
2118
xt_remove_dir_char(name);
2119
db->db_co_thread = xt_create_daemon(self, name);
2120
xt_set_thread_data(db->db_co_thread, db, dl_free_co_thread);
2121
xt_run_thread(self, db->db_co_thread, dl_run_co_thread);
2124
static void dl_wake_co_thread(XTDatabaseHPtr db)
2126
if (!xt_signal_cond(NULL, &db->db_datalogs.dlc_cond))
2127
xt_log_and_clear_exception_ns();
2130
xtPublic void xt_stop_compactor(XTThreadPtr self, XTDatabaseHPtr db)
2134
if (db->db_co_thread) {
2135
xt_lock_mutex(self, &db->db_datalogs.dlc_lock);
2136
pushr_(xt_unlock_mutex, &db->db_datalogs.dlc_lock);
2138
/* This pointer is safe as long as you have the transaction lock. */
2139
if ((thr_co = db->db_co_thread)) {
2140
xtThreadID tid = thr_co->t_id;
2142
/* Make sure the thread quits when woken up. */
2143
xt_terminate_thread(self, thr_co);
2145
dl_wake_co_thread(db);
2147
freer_(); // xt_unlock_mutex(&db->db_datalogs.dlc_lock)
2150
* This seems to kill the whole server sometimes!!
2151
* SIGTERM is going to a different thread??!
2152
xt_kill_thread(thread);
2154
xt_wait_for_thread_to_exit(tid, FALSE);
2156
/* PMC - This should not be necessary to set the signal here, but in the
2157
* debugger the handler is not called!!?
2158
thr_co->t_delayed_signal = SIGTERM;
2159
xt_kill_thread(thread);
2161
db->db_co_thread = NULL;
2164
freer_(); // xt_unlock_mutex(&db->db_datalogs.dlc_lock)
added
removed
plugin/pbxt/src/datalog_xt.cc
