mirror of
https://github.com/MariaDB/server.git
synced 2025-01-22 14:54:20 +01:00
1222 lines
41 KiB
C
1222 lines
41 KiB
C
/* -*- mode: C; c-basic-offset: 4 -*- */
|
|
#ident "$Id$"
|
|
#ident "Copyright (c) 2007, 2008, 2009 Tokutek Inc. All rights reserved."
|
|
#ident "The technology is licensed by the Massachusetts Institute of Technology, Rutgers State University of New Jersey, and the Research Foundation of State University of New York at Stony Brook under United States of America Serial No. 11/760379 and to the patents and/or patent applications resulting from it."
|
|
|
|
#include "includes.h"
|
|
|
|
static toku_pthread_mutex_t logger_mutex = TOKU_PTHREAD_MUTEX_INITIALIZER;
|
|
void toku_logger_lock_init(void) {
|
|
int r = toku_pthread_mutex_init(&logger_mutex, NULL);
|
|
assert(r == 0);
|
|
}
|
|
|
|
void toku_logger_lock_destroy(void) {
|
|
int r = toku_pthread_mutex_destroy(&logger_mutex);
|
|
assert(r == 0);
|
|
}
|
|
|
|
void* toku_malloc_in_rollback(TOKUTXN txn, size_t size) {
|
|
return malloc_in_memarena(txn->rollentry_arena, size);
|
|
}
|
|
|
|
void *toku_memdup_in_rollback(TOKUTXN txn, const void *v, size_t len) {
|
|
void *r=toku_malloc_in_rollback(txn, len);
|
|
memcpy(r,v,len);
|
|
return r;
|
|
}
|
|
|
|
char *toku_strdup_in_rollback(TOKUTXN txn, const char *s) {
|
|
return toku_memdup_in_rollback(txn, s, strlen(s)+1);
|
|
}
|
|
|
|
static int
|
|
toku_logger_fsync_null(int fd __attribute__((__unused__))) {
|
|
return 0;
|
|
}
|
|
|
|
int toku_logger_find_next_unused_log_file(const char *directory, long long *result) {
|
|
DIR *d=opendir(directory);
|
|
long long max=-1; *result = max;
|
|
struct dirent *de;
|
|
if (d==0) return errno;
|
|
while ((de=readdir(d))) {
|
|
if (de==0) return errno;
|
|
long long thisl;
|
|
int r = sscanf(de->d_name, "log%lld.tokulog", &thisl);
|
|
if (r==1 && thisl>max) max=thisl;
|
|
}
|
|
*result=max+1;
|
|
int r = closedir(d);
|
|
return r;
|
|
}
|
|
|
|
static int
|
|
logfilenamecompare (const void *ap, const void *bp) {
|
|
char *a=*(char**)ap;
|
|
char *b=*(char**)bp;
|
|
return strcmp(a,b);
|
|
}
|
|
|
|
int toku_logger_find_logfiles (const char *directory, char ***resultp) {
|
|
int result_limit=2;
|
|
int n_results=0;
|
|
char **MALLOC_N(result_limit, result);
|
|
struct dirent *de;
|
|
DIR *d=opendir(directory);
|
|
if (d==0) {
|
|
toku_free(result);
|
|
return errno;
|
|
}
|
|
int dirnamelen = strlen(directory);
|
|
while ((de=readdir(d))) {
|
|
long long thisl;
|
|
int r = sscanf(de->d_name, "log%lld.tokulog", &thisl);
|
|
if (r!=1) continue; // Skip over non-log files.
|
|
if (n_results+1>=result_limit) {
|
|
result_limit*=2;
|
|
result = toku_realloc(result, result_limit*sizeof(*result));
|
|
}
|
|
int fnamelen = dirnamelen + strlen(de->d_name) + 2; // One for the slash and one for the trailing NUL.
|
|
char *fname = toku_malloc(fnamelen);
|
|
snprintf(fname, fnamelen, "%s/%s", directory, de->d_name);
|
|
result[n_results++] = fname;
|
|
}
|
|
// Return them in increasing order.
|
|
qsort(result, n_results, sizeof(result[0]), logfilenamecompare);
|
|
*resultp = result;
|
|
result[n_results]=0; // make a trailing null
|
|
return d ? closedir(d) : 0;
|
|
}
|
|
|
|
int toku_logger_create (TOKULOGGER *resultp) {
|
|
int r;
|
|
TAGMALLOC(TOKULOGGER, result);
|
|
if (result==0) return errno;
|
|
result->is_open=0;
|
|
result->is_panicked=0;
|
|
result->write_log_files = 1;
|
|
result->lg_max = 100<<20; // 100MB default
|
|
result->head = result->tail = 0;
|
|
result->lsn = result->written_lsn = result->fsynced_lsn = (LSN){0};
|
|
list_init(&result->live_txns);
|
|
result->n_in_buf=0;
|
|
result->n_in_file=0;
|
|
result->directory=0;
|
|
result->checkpoint_lsn=(LSN){0};
|
|
result->write_block_size = BRT_DEFAULT_NODE_SIZE; // default logging size is the same as the default brt block size
|
|
*resultp=result;
|
|
r = ml_init(&result->input_lock); if (r!=0) goto died0;
|
|
r = ml_init(&result->output_lock); if (r!=0) goto died1;
|
|
return 0;
|
|
|
|
died1:
|
|
ml_destroy(&result->input_lock);
|
|
died0:
|
|
toku_free(result);
|
|
return r;
|
|
}
|
|
|
|
void toku_logger_set_cachetable (TOKULOGGER tl, CACHETABLE ct) {
|
|
tl->ct = ct;
|
|
}
|
|
|
|
void toku_logger_write_log_files (TOKULOGGER tl, int write_log_files) {
|
|
tl->write_log_files = write_log_files;
|
|
}
|
|
|
|
static int (*toku_os_fsync_function)(int)=fsync;
|
|
|
|
static const int log_format_version=0;
|
|
|
|
// Write something out. Keep trying even if partial writes occur.
|
|
// On error: Return negative with errno set.
|
|
// On success return nbytes.
|
|
static int write_it (int fd, const void *bufv, int nbytes) {
|
|
int org_nbytes=nbytes;
|
|
const char *buf=bufv;
|
|
while (nbytes>0) {
|
|
int r = write(fd, buf, nbytes);
|
|
if (r<0 || errno!=EAGAIN) return r;
|
|
buf+=r;
|
|
nbytes-=r;
|
|
}
|
|
return org_nbytes;
|
|
}
|
|
|
|
static int open_logfile (TOKULOGGER logger) {
|
|
int r;
|
|
int fnamelen = strlen(logger->directory)+50;
|
|
char fname[fnamelen];
|
|
snprintf(fname, fnamelen, "%s/log%012lld.tokulog", logger->directory, logger->next_log_file_number);
|
|
if (logger->write_log_files) {
|
|
logger->fd = open(fname, O_CREAT+O_WRONLY+O_TRUNC+O_EXCL+O_BINARY, S_IRWXU); if (logger->fd==-1) return errno;
|
|
} else {
|
|
logger->fd = open(DEV_NULL_FILE, O_WRONLY+O_BINARY);
|
|
// printf("%s: %s %d\n", __FUNCTION__, DEV_NULL_FILE, logger->fd); fflush(stdout);
|
|
if (logger->fd==-1) return errno;
|
|
}
|
|
logger->next_log_file_number++;
|
|
r = write_it(logger->fd, "tokulogg", 8); if (r!=8) return errno;
|
|
int version_l = toku_htonl(log_format_version); //version MUST be in network byte order regardless of disk order
|
|
r = write_it(logger->fd, &version_l, 4); if (r!=4) return errno;
|
|
logger->fsynced_lsn = logger->written_lsn;
|
|
logger->n_in_file = 12;
|
|
return 0;
|
|
}
|
|
|
|
static int close_and_open_logfile (TOKULOGGER logger) {
|
|
int r;
|
|
r = toku_os_fsync_function(logger->fd); if (r!=0) return errno;
|
|
r = close(logger->fd); if (r!=0) return errno;
|
|
return open_logfile(logger);
|
|
}
|
|
|
|
int toku_logger_open (const char *directory, TOKULOGGER logger) {
|
|
if (logger->is_open) return EINVAL;
|
|
if (logger->is_panicked) return EINVAL;
|
|
int r;
|
|
long long nexti;
|
|
r = toku_logger_find_next_unused_log_file(directory, &nexti);
|
|
if (r!=0) return r;
|
|
logger->directory = toku_strdup(directory);
|
|
if (logger->directory==0) return errno;
|
|
logger->next_log_file_number = nexti;
|
|
open_logfile(logger);
|
|
|
|
logger->lsn.lsn = 0; // WRONG!!! This should actually be calculated by looking at the log file.
|
|
logger->written_lsn.lsn = 0;
|
|
logger->fsynced_lsn.lsn = 0;
|
|
|
|
logger->is_open = 1;
|
|
if (!logger->write_log_files)
|
|
toku_set_func_fsync(toku_logger_fsync_null);
|
|
|
|
return 0;
|
|
|
|
}
|
|
|
|
void toku_logger_panic (TOKULOGGER logger, int err) {
|
|
logger->panic_errno=err;
|
|
logger->is_panicked=1;
|
|
}
|
|
int toku_logger_panicked(TOKULOGGER logger) {
|
|
if (logger==0) return 0;
|
|
return logger->is_panicked;
|
|
}
|
|
int toku_logger_is_open(TOKULOGGER logger) {
|
|
if (logger==0) return 0;
|
|
return logger->is_open;
|
|
}
|
|
|
|
int toku_logger_set_lg_max(TOKULOGGER logger, u_int32_t lg_max) {
|
|
if (logger==0) return EINVAL; // no logger
|
|
if (logger->is_panicked) return EINVAL;
|
|
if (logger->is_open) return EINVAL;
|
|
if (lg_max>(1<<30)) return EINVAL; // too big
|
|
logger->lg_max = lg_max;
|
|
return 0;
|
|
}
|
|
int toku_logger_get_lg_max(TOKULOGGER logger, u_int32_t *lg_maxp) {
|
|
if (logger==0) return EINVAL; // no logger
|
|
if (logger->is_panicked) return EINVAL;
|
|
*lg_maxp = logger->lg_max;
|
|
return 0;
|
|
|
|
}
|
|
|
|
int toku_logger_set_lg_bsize(TOKULOGGER logger, u_int32_t bsize) {
|
|
if (logger==0) return EINVAL; // no logger
|
|
if (logger->is_panicked) return EINVAL;
|
|
if (logger->is_open) return EINVAL;
|
|
if (bsize<=0 || bsize>(1<<30)) return EINVAL;
|
|
logger->write_block_size = bsize;
|
|
return 0;
|
|
}
|
|
|
|
// Enter holding both locks
|
|
// Exit holding only the output_lock
|
|
static int do_write (TOKULOGGER logger, int do_fsync) {
|
|
int r;
|
|
struct logbytes *list = logger->head;
|
|
logger->head=logger->tail=0;
|
|
logger->n_in_buf=0;
|
|
r=ml_unlock(&logger->input_lock); if (r!=0) goto panic;
|
|
logger->n_in_buf=0;
|
|
while (list) {
|
|
if (logger->n_in_file + list->nbytes <= logger->lg_max) {
|
|
if (logger->n_in_buf + list->nbytes <= LOGGER_BUF_SIZE) {
|
|
memcpy(logger->buf+logger->n_in_buf, list->bytes, list->nbytes);
|
|
logger->n_in_buf+=list->nbytes;
|
|
logger->n_in_file+=list->nbytes;
|
|
logger->written_lsn = list->lsn;
|
|
struct logbytes *next=list->next;
|
|
toku_free(list);
|
|
list=next;
|
|
} else {
|
|
// it doesn't fit in the buffer, but it does fit in the file. So flush the buffer
|
|
r=write_it(logger->fd, logger->buf, logger->n_in_buf);
|
|
if (r!=logger->n_in_buf) { r=errno; goto panic; }
|
|
logger->n_in_buf=0;
|
|
// Special case for a log entry that's too big to fit in the buffer.
|
|
if (list->nbytes > LOGGER_BUF_SIZE) {
|
|
r=write_it(logger->fd, list->bytes, list->nbytes);
|
|
if (r!=list->nbytes) { r=errno; goto panic; }
|
|
logger->n_in_file+=list->nbytes;
|
|
logger->written_lsn = list->lsn;
|
|
struct logbytes *next=list->next;
|
|
toku_free(list);
|
|
list=next;
|
|
}
|
|
}
|
|
} else {
|
|
// The new item doesn't fit in the file, so write the buffer, reopen the file, and try again
|
|
r=write_it(logger->fd, logger->buf, logger->n_in_buf);
|
|
logger->n_in_buf=0;
|
|
r=close_and_open_logfile(logger); if (r!=0) goto panic;
|
|
}
|
|
}
|
|
r=write_it(logger->fd, logger->buf, logger->n_in_buf);
|
|
if (r!=logger->n_in_buf) { r=errno; goto panic; }
|
|
logger->n_in_buf=0;
|
|
if (do_fsync) {
|
|
r = toku_os_fsync_function(logger->fd);
|
|
logger->fsynced_lsn = logger->written_lsn;
|
|
}
|
|
return 0;
|
|
panic:
|
|
toku_logger_panic(logger, r);
|
|
return r;
|
|
}
|
|
|
|
// enter holding input_lock
|
|
// exit holding no locks
|
|
int toku_logger_log_bytes (TOKULOGGER logger, struct logbytes *bytes, int do_fsync) {
|
|
int r;
|
|
if (logger->is_panicked) return EINVAL;
|
|
logger->n_in_buf += bytes->nbytes;
|
|
if (logger->tail) {
|
|
logger->tail->next=bytes;
|
|
} else {
|
|
logger->head = bytes;
|
|
}
|
|
logger->tail = bytes;
|
|
bytes->next = 0;
|
|
if (logger->n_in_buf >= LOGGER_BUF_SIZE || do_fsync) {
|
|
// We must flush it
|
|
r=ml_unlock(&logger->input_lock); if (r!=0) goto panic;
|
|
r=ml_lock(&logger->output_lock); if (r!=0) goto panic;
|
|
if (logger->written_lsn.lsn < bytes->lsn.lsn) {
|
|
// We found that our record has not yet been written, so we must write it, and everything else
|
|
r=ml_lock(&logger->input_lock); if (r!=0) goto panic;
|
|
r=do_write(logger, do_fsync); if (r!=0) goto panic;
|
|
} else {
|
|
/* Our LSN has been written. We have the output lock */
|
|
if (do_fsync && logger->fsynced_lsn.lsn > bytes->lsn.lsn) {
|
|
/* But we need to fsync it. */
|
|
r = toku_os_fsync_function(logger->fd);
|
|
logger->fsynced_lsn = logger->written_lsn;
|
|
}
|
|
}
|
|
r=ml_unlock(&logger->output_lock); if (r!=0) goto panic;
|
|
} else {
|
|
r=ml_unlock(&logger->input_lock); if (r!=0) goto panic;
|
|
}
|
|
return 0;
|
|
panic:
|
|
toku_logger_panic(logger, r);
|
|
return r;
|
|
}
|
|
|
|
// No locks held on entry
|
|
// No locks held on exit.
|
|
// No locks are needed, since you cannot legally close the log concurrently with doing anything else.
|
|
// But grab the locks just to be careful, including one to prevent access
|
|
// between unlocking and destroying.
|
|
int toku_logger_close(TOKULOGGER *loggerp) {
|
|
TOKULOGGER logger = *loggerp;
|
|
if (logger->is_panicked) return EINVAL;
|
|
int r = 0;
|
|
int locked_logger = 0;
|
|
if (!logger->is_open) goto is_closed;
|
|
r = ml_lock(&logger->output_lock); if (r!=0) goto panic;
|
|
r = ml_lock(&logger->input_lock); if (r!=0) goto panic;
|
|
r = toku_pthread_mutex_lock(&logger_mutex); if (r!=0) goto panic;
|
|
locked_logger = 1;
|
|
r = do_write(logger, 1); if (r!=0) goto panic; //Releases the input lock
|
|
if (logger->fd!=-1) {
|
|
r = close(logger->fd); if (r!=0) { r=errno; goto panic; }
|
|
}
|
|
logger->fd=-1;
|
|
|
|
r = ml_unlock(&logger->output_lock); if (r!=0) goto panic;
|
|
is_closed:
|
|
r = ml_destroy(&logger->output_lock); if (r!=0) goto panic;
|
|
r = ml_destroy(&logger->input_lock); if (r!=0) goto panic;
|
|
logger->is_panicked=1; // Just in case this might help.
|
|
if (logger->directory) toku_free(logger->directory);
|
|
toku_free(logger);
|
|
*loggerp=0;
|
|
if (locked_logger) {
|
|
r = toku_pthread_mutex_unlock(&logger_mutex); if (r!=0) goto panic;
|
|
}
|
|
return r;
|
|
panic:
|
|
toku_logger_panic(logger, r);
|
|
return r;
|
|
}
|
|
|
|
// Entry: Holds no locks
|
|
// Exit: Holds no locks
|
|
// This is the exported fsync used by ydb.c
|
|
int toku_logger_fsync (TOKULOGGER logger) {
|
|
int r;
|
|
if (logger->is_panicked) return EINVAL;
|
|
r = ml_lock(&logger->output_lock); if (r!=0) goto panic;
|
|
r = ml_lock(&logger->input_lock); if (r!=0) goto panic;
|
|
r = do_write(logger, 1);
|
|
r = ml_unlock(&logger->output_lock); if (r!=0) goto panic;
|
|
return 0;
|
|
panic:
|
|
toku_logger_panic(logger, r);
|
|
return r;
|
|
}
|
|
|
|
// wbuf points into logbytes
|
|
int toku_logger_finish (TOKULOGGER logger, struct logbytes *logbytes, struct wbuf *wbuf, int do_fsync) {
|
|
if (logger->is_panicked) return EINVAL;
|
|
u_int32_t checksum = x1764_finish(&wbuf->checksum);
|
|
wbuf_int(wbuf, checksum);
|
|
wbuf_int(wbuf, 4+wbuf->ndone);
|
|
logbytes->nbytes=wbuf->ndone;
|
|
return toku_logger_log_bytes(logger, logbytes, do_fsync);
|
|
}
|
|
|
|
static void note_txn_closing (TOKUTXN txn);
|
|
|
|
void toku_logger_txn_close (TOKUTXN txn) {
|
|
memarena_close(&txn->rollentry_arena);
|
|
if (txn->rollentry_filename!=0) {
|
|
int r = close(txn->rollentry_fd);
|
|
assert(r==0);
|
|
r = unlink(txn->rollentry_filename);
|
|
assert(r==0);
|
|
toku_free(txn->rollentry_filename);
|
|
}
|
|
|
|
list_remove(&txn->live_txns_link);
|
|
note_txn_closing(txn);
|
|
toku_free(txn);
|
|
return;
|
|
}
|
|
|
|
int toku_commit_rollback_item (TOKUTXN txn, struct roll_entry *item, YIELDF yield, void*yieldv) {
|
|
int r=0;
|
|
rolltype_dispatch_assign(item, toku_commit_, r, txn, yield, yieldv);
|
|
return r;
|
|
}
|
|
|
|
int toku_abort_rollback_item (TOKUTXN txn, struct roll_entry *item, YIELDF yield, void*yieldv) {
|
|
int r=0;
|
|
rolltype_dispatch_assign(item, toku_rollback_, r, txn, yield, yieldv);
|
|
if (r!=0) return r;
|
|
return 0;
|
|
}
|
|
|
|
static int note_brt_used_in_parent_txn(OMTVALUE brtv, u_int32_t UU(index), void*parentv) {
|
|
return toku_txn_note_brt(parentv, brtv);
|
|
}
|
|
|
|
// Doesn't close the txn, just performs the commit operations.
|
|
int toku_logger_commit (TOKUTXN txn, int nosync, YIELDF yield, void*yieldv) {
|
|
// printf("%s:%d committing\n", __FILE__, __LINE__);
|
|
// panic handled in log_commit
|
|
int r = toku_log_commit(txn->logger, (LSN*)0, (txn->parent==0) && !nosync, txn->txnid64); // exits holding neither of the tokulogger locks.
|
|
if (r==0) {
|
|
if (txn->parent!=0) {
|
|
// First we must put a rollinclude entry into the parent if we have a rollentry file.
|
|
if (txn->rollentry_filename) {
|
|
int len = strlen(txn->rollentry_filename);
|
|
// Don't have to strdup the rollentry_filename because
|
|
// we take ownership of it.
|
|
BYTESTRING fname = {len, toku_strdup_in_rollback(txn, txn->rollentry_filename)};
|
|
r = toku_logger_save_rollback_rollinclude(txn->parent, fname);
|
|
if (r!=0) return r;
|
|
r = close(txn->rollentry_fd);
|
|
if (r!=0) {
|
|
// We have to do the unlink ourselves, and then
|
|
// set txn->rollentry_filename=0 so that the cleanup
|
|
// won't try to close the fd again.
|
|
unlink(txn->rollentry_filename);
|
|
toku_free(txn->rollentry_filename);
|
|
txn->rollentry_filename = 0;
|
|
return r;
|
|
}
|
|
// Stop the cleanup from closing and unlinking the file.
|
|
toku_free(txn->rollentry_filename);
|
|
txn->rollentry_filename = 0;
|
|
}
|
|
// Append the list to the front of the parent.
|
|
if (txn->oldest_logentry) {
|
|
// There are some entries, so link them in.
|
|
txn->oldest_logentry->prev = txn->parent->newest_logentry;
|
|
if (txn->parent->newest_logentry) {
|
|
txn->parent->newest_logentry->next = txn->oldest_logentry;
|
|
} else {
|
|
txn->parent->oldest_logentry = txn->oldest_logentry;
|
|
}
|
|
txn->parent->newest_logentry = txn->newest_logentry;
|
|
txn->parent->rollentry_resident_bytecount += txn->rollentry_resident_bytecount;
|
|
txn->parent->rollentry_raw_count += txn->rollentry_raw_count;
|
|
txn->rollentry_resident_bytecount = 0;
|
|
}
|
|
if (txn->parent->oldest_logentry==0) {
|
|
txn->parent->oldest_logentry = txn->oldest_logentry;
|
|
}
|
|
txn->newest_logentry = txn->oldest_logentry = 0;
|
|
// Put all the memarena data into the parent.
|
|
if (memarena_total_size_in_use(txn->rollentry_arena) > 0) {
|
|
// If there are no bytes to move, then just leave things alone, and let the memory be reclaimed on txn is closed.
|
|
memarena_move_buffers(txn->parent->rollentry_arena, txn->rollentry_arena);
|
|
}
|
|
|
|
// Note the open brts, the omts must be merged
|
|
r = toku_omt_iterate(txn->open_brts, note_brt_used_in_parent_txn, txn->parent);
|
|
assert(r==0);
|
|
|
|
r = toku_maybe_spill_rollbacks(txn->parent);
|
|
assert(r==0);
|
|
|
|
} else {
|
|
// do the commit calls and free everything
|
|
// we do the commit calls in reverse order too.
|
|
{
|
|
struct roll_entry *item;
|
|
//printf("%s:%d abort\n", __FILE__, __LINE__);
|
|
int count=0;
|
|
while ((item=txn->newest_logentry)) {
|
|
txn->newest_logentry = item->prev;
|
|
r = toku_commit_rollback_item(txn, item, yield, yieldv);
|
|
if (r!=0) return r;
|
|
count++;
|
|
if (count%2 == 0) yield(NULL, yieldv);
|
|
}
|
|
}
|
|
|
|
// Read stuff out of the file and execute it.
|
|
if (txn->rollentry_filename) {
|
|
r = toku_commit_fileentries(txn->rollentry_fd, txn, yield, yieldv);
|
|
}
|
|
}
|
|
}
|
|
return r;
|
|
}
|
|
|
|
#if 0
|
|
int toku_logger_log_checkpoint (TOKULOGGER logger) {
|
|
if (logger->is_panicked) return EINVAL;
|
|
int r = toku_cachetable_checkpoint(logger->ct);
|
|
if (r!=0) return r;
|
|
logger->checkpoint_lsns[1]=logger->checkpoint_lsns[0];
|
|
logger->checkpoint_lsns[0]=logger->lsn;
|
|
return toku_log_checkpoint(logger, (LSN*)0, 1);
|
|
}
|
|
#endif
|
|
|
|
int toku_logger_txn_begin (TOKUTXN parent_tokutxn, TOKUTXN *tokutxn, TOKULOGGER logger) {
|
|
if (logger->is_panicked) return EINVAL;
|
|
TAGMALLOC(TOKUTXN, result);
|
|
if (result==0) return errno;
|
|
int r =toku_log_xbegin(logger, &result->first_lsn, 0, parent_tokutxn ? parent_tokutxn->txnid64 : 0);
|
|
if (r!=0) { toku_logger_panic(logger, r); toku_free(result); return r; }
|
|
if ((r=toku_omt_create(&result->open_brts))!=0) {
|
|
toku_logger_panic(logger, r);
|
|
toku_free(result);
|
|
return r;
|
|
}
|
|
result->txnid64 = result->first_lsn.lsn;
|
|
result->logger = logger;
|
|
result->parent = parent_tokutxn;
|
|
result->oldest_logentry = result->newest_logentry = 0;
|
|
|
|
result->rollentry_arena = memarena_create();
|
|
|
|
list_push(&logger->live_txns, &result->live_txns_link);
|
|
result->rollentry_resident_bytecount=0;
|
|
result->rollentry_raw_count = 0;
|
|
result->rollentry_filename = 0;
|
|
result->rollentry_fd = -1;
|
|
result->rollentry_filesize = 0;
|
|
*tokutxn = result;
|
|
return 0;
|
|
}
|
|
|
|
int toku_logger_log_fcreate (TOKUTXN txn, const char *fname, FILENUM filenum, int mode) {
|
|
if (txn==0) return 0;
|
|
if (txn->logger->is_panicked) return EINVAL;
|
|
BYTESTRING bs = { .len=strlen(fname), .data = toku_strdup_in_rollback(txn, fname) };
|
|
int r = toku_log_fcreate (txn->logger, (LSN*)0, 0, toku_txn_get_txnid(txn), filenum, bs, mode);
|
|
if (r!=0) return r;
|
|
r = toku_logger_save_rollback_fcreate(txn, toku_txn_get_txnid(txn), filenum, bs);
|
|
return r;
|
|
}
|
|
|
|
/* fopen isn't really an action. It's just for bookkeeping. We need to know the filename that goes with a filenum. */
|
|
int toku_logger_log_fopen (TOKUTXN txn, const char * fname, FILENUM filenum) {
|
|
if (txn==0) return 0;
|
|
if (txn->logger->is_panicked) return EINVAL;
|
|
BYTESTRING bs;
|
|
bs.len = strlen(fname);
|
|
bs.data = (char*)fname;
|
|
return toku_log_fopen (txn->logger, (LSN*)0, 0, toku_txn_get_txnid(txn), bs, filenum);
|
|
}
|
|
|
|
static int
|
|
toku_fread_u_int8_t_nocrclen (FILE *f, u_int8_t *v) {
|
|
int vi=fgetc(f);
|
|
if (vi==EOF) return -1;
|
|
u_int8_t vc=(u_int8_t)vi;
|
|
*v = vc;
|
|
return 0;
|
|
}
|
|
|
|
int toku_fread_u_int8_t (FILE *f, u_int8_t *v, struct x1764 *mm, u_int32_t *len) {
|
|
int vi=fgetc(f);
|
|
if (vi==EOF) return -1;
|
|
u_int8_t vc=(u_int8_t)vi;
|
|
x1764_add(mm, &vc, 1);
|
|
(*len)++;
|
|
*v = vc;
|
|
return 0;
|
|
}
|
|
|
|
int toku_fread_u_int32_t_nocrclen (FILE *f, u_int32_t *v) {
|
|
u_int32_t result;
|
|
u_int8_t *cp = (u_int8_t*)&result;
|
|
int r;
|
|
r = toku_fread_u_int8_t_nocrclen (f, cp+0); if (r!=0) return r;
|
|
r = toku_fread_u_int8_t_nocrclen (f, cp+1); if (r!=0) return r;
|
|
r = toku_fread_u_int8_t_nocrclen (f, cp+2); if (r!=0) return r;
|
|
r = toku_fread_u_int8_t_nocrclen (f, cp+3); if (r!=0) return r;
|
|
*v = toku_dtoh32(result);
|
|
|
|
return 0;
|
|
}
|
|
int toku_fread_u_int32_t (FILE *f, u_int32_t *v, struct x1764 *checksum, u_int32_t *len) {
|
|
u_int32_t result;
|
|
u_int8_t *cp = (u_int8_t*)&result;
|
|
int r;
|
|
r = toku_fread_u_int8_t (f, cp+0, checksum, len); if(r!=0) return r;
|
|
r = toku_fread_u_int8_t (f, cp+1, checksum, len); if(r!=0) return r;
|
|
r = toku_fread_u_int8_t (f, cp+2, checksum, len); if(r!=0) return r;
|
|
r = toku_fread_u_int8_t (f, cp+3, checksum, len); if(r!=0) return r;
|
|
*v = toku_dtoh32(result);
|
|
return 0;
|
|
}
|
|
|
|
static int
|
|
toku_fread_int32_t (FILE *f, int32_t *v, struct x1764 *checksum, u_int32_t *len) {
|
|
u_int32_t uv;
|
|
int r = toku_fread_u_int32_t(f, &uv, checksum, len);
|
|
int32_t rv = uv;
|
|
if (r==0) *v=rv;
|
|
return r;
|
|
}
|
|
|
|
static int
|
|
toku_fread_u_int64_t (FILE *f, u_int64_t *v, struct x1764 *checksum, u_int32_t *len) {
|
|
u_int32_t v1,v2;
|
|
int r;
|
|
r=toku_fread_u_int32_t(f, &v1, checksum, len); if (r!=0) return r;
|
|
r=toku_fread_u_int32_t(f, &v2, checksum, len); if (r!=0) return r;
|
|
*v = (((u_int64_t)v1)<<32 ) | ((u_int64_t)v2);
|
|
return 0;
|
|
}
|
|
int toku_fread_LSN (FILE *f, LSN *lsn, struct x1764 *checksum, u_int32_t *len) {
|
|
return toku_fread_u_int64_t (f, &lsn->lsn, checksum, len);
|
|
}
|
|
int toku_fread_FILENUM (FILE *f, FILENUM *filenum, struct x1764 *checksum, u_int32_t *len) {
|
|
return toku_fread_u_int32_t (f, &filenum->fileid, checksum, len);
|
|
}
|
|
int toku_fread_DISKOFF (FILE *f, DISKOFF *diskoff, struct x1764 *checksum, u_int32_t *len) {
|
|
int r = toku_fread_u_int64_t (f, (u_int64_t*)diskoff, checksum, len); // sign conversion will be OK.
|
|
return r;
|
|
}
|
|
int toku_fread_BLOCKNUM (FILE *f, BLOCKNUM *blocknum, struct x1764 *checksum, u_int32_t *len) {
|
|
u_int64_t b = 0;
|
|
int r = toku_fread_u_int64_t (f, &b, checksum, len); // sign conversion will be OK.
|
|
blocknum->b=b;
|
|
return r;
|
|
}
|
|
int toku_fread_TXNID (FILE *f, TXNID *txnid, struct x1764 *checksum, u_int32_t *len) {
|
|
return toku_fread_u_int64_t (f, txnid, checksum, len);
|
|
}
|
|
|
|
// fills in the bs with malloced data.
|
|
int toku_fread_BYTESTRING (FILE *f, BYTESTRING *bs, struct x1764 *checksum, u_int32_t *len) {
|
|
int r=toku_fread_u_int32_t(f, (u_int32_t*)&bs->len, checksum, len);
|
|
if (r!=0) return r;
|
|
bs->data = toku_malloc(bs->len);
|
|
u_int32_t i;
|
|
for (i=0; i<bs->len; i++) {
|
|
r=toku_fread_u_int8_t(f, (u_int8_t*)&bs->data[i], checksum, len);
|
|
if (r!=0) {
|
|
toku_free(bs->data);
|
|
bs->data=0;
|
|
return r;
|
|
}
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
int toku_fread_LOGGEDBRTHEADER (FILE *f, LOGGEDBRTHEADER *v, struct x1764 *checksum, u_int32_t *len) {
|
|
int r;
|
|
r = toku_fread_u_int32_t(f, &v->size, checksum, len); if (r!=0) return r;
|
|
r = toku_fread_u_int32_t(f, &v->flags, checksum, len); if (r!=0) return r;
|
|
r = toku_fread_u_int32_t(f, &v->nodesize, checksum, len); if (r!=0) return r;
|
|
r = toku_fread_BLOCKNUM (f, &v->free_blocks, checksum, len); if (r!=0) return r;
|
|
r = toku_fread_BLOCKNUM (f, &v->unused_blocks, checksum, len); if (r!=0) return r;
|
|
r = toku_fread_BLOCKNUM (f, &v->root, checksum, len); if (r!=0) return r;
|
|
r = toku_fread_BLOCKNUM (f, &v->btt_size, checksum, len); if (r!=0) return r;
|
|
r = toku_fread_DISKOFF (f, &v->btt_diskoff, checksum, len); if (r!=0) return r;
|
|
XMALLOC_N(v->btt_size.b, v->btt_pairs);
|
|
int64_t i;
|
|
for (i=0; i<v->btt_size.b; i++) {
|
|
r = toku_fread_DISKOFF(f, &v->btt_pairs[i].off, checksum, len);
|
|
if (r!=0) { toku_free(v->btt_pairs); return r; }
|
|
r = toku_fread_int32_t (f, &v->btt_pairs[i].size, checksum, len);
|
|
if (r!=0) { toku_free(v->btt_pairs); return r; }
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
int toku_fread_INTPAIRARRAY (FILE *f, INTPAIRARRAY *v, struct x1764 *checksum, u_int32_t *len) {
|
|
int r;
|
|
u_int32_t i;
|
|
r = toku_fread_u_int32_t(f, &v->size, checksum, len); if (r!=0) return r;
|
|
MALLOC_N(v->size, v->array);
|
|
if (v->array==0) return errno;
|
|
for (i=0; i<v->size; i++) {
|
|
r = toku_fread_u_int32_t(f, &v->array[i].a, checksum, len); if (r!=0) return r;
|
|
r = toku_fread_u_int32_t(f, &v->array[i].b, checksum, len); if (r!=0) return r;
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
int toku_logprint_LSN (FILE *outf, FILE *inf, const char *fieldname, struct x1764 *checksum, u_int32_t *len, const char *format __attribute__((__unused__))) {
|
|
LSN v;
|
|
int r = toku_fread_LSN(inf, &v, checksum, len);
|
|
if (r!=0) return r;
|
|
fprintf(outf, " %s=%" PRIu64, fieldname, v.lsn);
|
|
return 0;
|
|
}
|
|
int toku_logprint_TXNID (FILE *outf, FILE *inf, const char *fieldname, struct x1764 *checksum, u_int32_t *len, const char *format __attribute__((__unused__))) {
|
|
TXNID v;
|
|
int r = toku_fread_TXNID(inf, &v, checksum, len);
|
|
if (r!=0) return r;
|
|
fprintf(outf, " %s=%" PRIu64, fieldname, v);
|
|
return 0;
|
|
}
|
|
|
|
int toku_logprint_u_int8_t (FILE *outf, FILE *inf, const char *fieldname, struct x1764 *checksum, u_int32_t *len, const char *format) {
|
|
u_int8_t v;
|
|
int r = toku_fread_u_int8_t(inf, &v, checksum, len);
|
|
if (r!=0) return r;
|
|
fprintf(outf, " %s=%d", fieldname, v);
|
|
if (format) fprintf(outf, format, v);
|
|
else if (v=='\'') fprintf(outf, "('\'')");
|
|
else if (isprint(v)) fprintf(outf, "('%c')", v);
|
|
else {}/*nothing*/
|
|
return 0;
|
|
|
|
}
|
|
|
|
int toku_logprint_u_int32_t (FILE *outf, FILE *inf, const char *fieldname, struct x1764 *checksum, u_int32_t *len, const char *format) {
|
|
u_int32_t v;
|
|
int r = toku_fread_u_int32_t(inf, &v, checksum, len);
|
|
if (r!=0) return r;
|
|
fprintf(outf, " %s=", fieldname);
|
|
fprintf(outf, format ? format : "%d", v);
|
|
return 0;
|
|
|
|
}
|
|
|
|
void toku_print_BYTESTRING (FILE *outf, u_int32_t len, char *data) {
|
|
fprintf(outf, "{len=%u data=\"", len);
|
|
u_int32_t i;
|
|
for (i=0; i<len; i++) {
|
|
switch (data[i]) {
|
|
case '"': fprintf(outf, "\\\""); break;
|
|
case '\\': fprintf(outf, "\\\\"); break;
|
|
case '\n': fprintf(outf, "\\n"); break;
|
|
default:
|
|
if (isprint(data[i])) fprintf(outf, "%c", data[i]);
|
|
else fprintf(outf, "\\%03o", (unsigned char)(data[i]));
|
|
}
|
|
}
|
|
fprintf(outf, "\"}");
|
|
|
|
}
|
|
|
|
int toku_logprint_BYTESTRING (FILE *outf, FILE *inf, const char *fieldname, struct x1764 *checksum, u_int32_t *len, const char *format __attribute__((__unused__))) {
|
|
BYTESTRING bs;
|
|
int r = toku_fread_BYTESTRING(inf, &bs, checksum, len);
|
|
if (r!=0) return r;
|
|
fprintf(outf, " %s=", fieldname);
|
|
toku_print_BYTESTRING(outf, bs.len, bs.data);
|
|
toku_free(bs.data);
|
|
return 0;
|
|
}
|
|
int toku_logprint_FILENUM (FILE *outf, FILE *inf, const char *fieldname, struct x1764 *checksum, u_int32_t *len, const char *format) {
|
|
return toku_logprint_u_int32_t(outf, inf, fieldname, checksum, len, format);
|
|
|
|
}
|
|
int toku_logprint_DISKOFF (FILE *outf, FILE *inf, const char *fieldname, struct x1764 *checksum, u_int32_t *len, const char *format __attribute__((__unused__))) {
|
|
DISKOFF v;
|
|
int r = toku_fread_DISKOFF(inf, &v, checksum, len);
|
|
if (r!=0) return r;
|
|
fprintf(outf, " %s=%" PRId64, fieldname, v);
|
|
return 0;
|
|
}
|
|
int toku_logprint_BLOCKNUM (FILE *outf, FILE *inf, const char *fieldname, struct x1764 *checksum, u_int32_t *len, const char *format __attribute__((__unused__))) {
|
|
BLOCKNUM v;
|
|
int r = toku_fread_BLOCKNUM(inf, &v, checksum, len);
|
|
if (r!=0) return r;
|
|
fprintf(outf, " %s=%"PRId64, fieldname, v.b);
|
|
return 0;
|
|
}
|
|
|
|
int toku_logprint_LOGGEDBRTHEADER (FILE *outf, FILE *inf, const char *fieldname, struct x1764 *checksum, u_int32_t *len, const char *format __attribute__((__unused__))) {
|
|
LOGGEDBRTHEADER v;
|
|
int r = toku_fread_LOGGEDBRTHEADER(inf, &v, checksum, len);
|
|
if (r!=0) return r;
|
|
fprintf(outf, " %s={size=%u flags=%u nodesize=%u free_blocks=%" PRId64 " unused_memory=%" PRId64, fieldname, v.size, v.flags, v.nodesize, v.free_blocks.b, v.unused_blocks.b);
|
|
fprintf(outf, " btt_size=%" PRId64 " btt_diskoff=%" PRId64 " btt_pairs={", v.btt_size.b, v.btt_diskoff) ;
|
|
int64_t i;
|
|
for (i=0; i<v.btt_size.b; i++) {
|
|
if (i>0) printf(" ");
|
|
fprintf(outf, "%" PRId64 ",%d", v.btt_pairs[i].off, v.btt_pairs[i].size);
|
|
}
|
|
fprintf(outf, "}");
|
|
return 0;
|
|
|
|
}
|
|
|
|
int toku_logprint_INTPAIRARRAY (FILE *outf, FILE *inf, const char *fieldname, struct x1764 *checksum, u_int32_t *len, const char *format __attribute__((__unused__))) {
|
|
INTPAIRARRAY v;
|
|
u_int32_t i;
|
|
int r = toku_fread_INTPAIRARRAY(inf, &v, checksum, len);
|
|
if (r!=0) return r;
|
|
fprintf(outf, " %s={size=%u array={", fieldname, v.size);
|
|
for (i=0; i<v.size; i++) {
|
|
if (i!=0) fprintf(outf, " ");
|
|
fprintf(outf, "{%u %u}", v.array[i].a, v.array[i].b);
|
|
}
|
|
toku_free(v.array);
|
|
return 0;
|
|
}
|
|
|
|
int toku_read_and_print_logmagic (FILE *f, u_int32_t *versionp) {
|
|
{
|
|
char magic[8];
|
|
int r=fread(magic, 1, 8, f);
|
|
if (r!=8) {
|
|
return DB_BADFORMAT;
|
|
}
|
|
if (memcmp(magic, "tokulogg", 8)!=0) {
|
|
return DB_BADFORMAT;
|
|
}
|
|
}
|
|
{
|
|
int version;
|
|
int r=fread(&version, 1, 4, f);
|
|
if (r!=4) {
|
|
return DB_BADFORMAT;
|
|
}
|
|
//printf("tokulog v.%d\n", toku_dtoh32(version));
|
|
//version MUST be in network order regardless of disk order
|
|
*versionp=toku_ntohl(version);
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
TXNID toku_txn_get_txnid (TOKUTXN txn) {
|
|
if (txn==0) return 0;
|
|
else return txn->txnid64;
|
|
}
|
|
|
|
LSN toku_txn_get_last_lsn (TOKUTXN txn) {
|
|
if (txn==0) return (LSN){0};
|
|
return txn->last_lsn;
|
|
}
|
|
LSN toku_logger_last_lsn(TOKULOGGER logger) {
|
|
LSN result=logger->lsn;
|
|
result.lsn--;
|
|
return result;
|
|
}
|
|
|
|
TOKULOGGER toku_txn_logger (TOKUTXN txn) {
|
|
return txn ? txn->logger : 0;
|
|
}
|
|
|
|
#if 0
|
|
static int
|
|
toku_abort_logentry_commit (struct logtype_commit *le __attribute__((__unused__)), TOKUTXN txn) {
|
|
toku_logger_panic(txn->logger, EINVAL);
|
|
return EINVAL;
|
|
}
|
|
#endif
|
|
|
|
// Doesn't close the txn, just performs the abort operations.
|
|
int toku_logger_abort(TOKUTXN txn, YIELDF yield, void*yieldv) {
|
|
//printf("%s:%d aborting\n", __FILE__, __LINE__);
|
|
// Must undo everything. Must undo it all in reverse order.
|
|
// Build the reverse list
|
|
//printf("%s:%d abort\n", __FILE__, __LINE__);
|
|
{
|
|
int r = toku_log_xabort(txn->logger, (LSN*)0, 0, txn->txnid64);
|
|
if (r!=0) return r;
|
|
}
|
|
{
|
|
struct roll_entry *item;
|
|
int count=0;
|
|
while ((item=txn->newest_logentry)) {
|
|
txn->newest_logentry = item->prev;
|
|
int r = toku_abort_rollback_item(txn, item, yield, yieldv);
|
|
if (r!=0) return r;
|
|
count++;
|
|
if (count%2 == 0) yield(NULL, yieldv);
|
|
}
|
|
}
|
|
list_remove(&txn->live_txns_link);
|
|
// Read stuff out of the file and roll it back.
|
|
if (txn->rollentry_filename) {
|
|
int r = toku_rollback_fileentries(txn->rollentry_fd, txn, yield, yieldv);
|
|
assert(r==0);
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
int toku_txnid2txn (TOKULOGGER logger, TXNID txnid, TOKUTXN *result) {
|
|
if (logger==0) return -1;
|
|
struct list *l;
|
|
for (l = list_head(&logger->live_txns); l != &logger->live_txns; l = l->next) {
|
|
TOKUTXN txn = list_struct(l, struct tokutxn, live_txns_link);
|
|
assert(txn->tag==TYP_TOKUTXN);
|
|
if (txn->txnid64==txnid) {
|
|
*result = txn;
|
|
return 0;
|
|
}
|
|
}
|
|
// If there is no txn, then we treat it as the null txn.
|
|
*result = 0;
|
|
return 0;
|
|
}
|
|
|
|
int toku_set_func_fsync (int (*fsync_function)(int)) {
|
|
toku_os_fsync_function = fsync_function;
|
|
return 0;
|
|
}
|
|
|
|
// Find the earliest LSN in a log
|
|
static int peek_at_log (TOKULOGGER logger, char* filename, LSN *first_lsn) {
|
|
logger=logger;
|
|
int fd = open(filename, O_RDONLY+O_BINARY);
|
|
if (fd<0) {
|
|
if (logger->write_log_files) printf("couldn't open: %s\n", strerror(errno));
|
|
return errno;
|
|
}
|
|
enum { SKIP = 12+1+4 }; // read the 12 byte header, the first cmd, and the first len
|
|
unsigned char header[SKIP+8];
|
|
int r = read(fd, header, SKIP+8);
|
|
if (r!=SKIP+8) return 0; // cannot determine that it's archivable, so we'll assume no. If a later-log is archivable is then this one will be too.
|
|
|
|
u_int64_t lsn;
|
|
{
|
|
struct rbuf rb;
|
|
rb.buf = header+SKIP;
|
|
rb.size = 8;
|
|
rb.ndone = 0;
|
|
lsn = rbuf_ulonglong(&rb);
|
|
}
|
|
|
|
r=close(fd);
|
|
if (r!=0) { return 0; }
|
|
|
|
first_lsn->lsn=lsn;
|
|
return 0;
|
|
}
|
|
|
|
// Return a malloc'd array of malloc'd strings which are the filenames that can be archived.
|
|
int toku_logger_log_archive (TOKULOGGER logger, char ***logs_p, int flags) {
|
|
if (flags!=0) return EINVAL; // don't know what to do.
|
|
int all_n_logs;
|
|
int i;
|
|
char **all_logs;
|
|
int r = toku_logger_find_logfiles (logger->directory, &all_logs);
|
|
if (r!=0) return r;
|
|
|
|
for (i=0; all_logs[i]; i++);
|
|
all_n_logs=i;
|
|
// get them into increasing order
|
|
qsort(all_logs, all_n_logs, sizeof(all_logs[0]), logfilenamecompare);
|
|
|
|
LSN oldest_live_txn_lsn={1LL<<63};
|
|
{
|
|
struct list *l;
|
|
for (l=list_head(&logger->live_txns); l!=&logger->live_txns; l=l->next) {
|
|
TOKUTXN txn = list_struct(l, struct tokutxn, live_txns_link);
|
|
if (oldest_live_txn_lsn.lsn>txn->txnid64) {
|
|
oldest_live_txn_lsn.lsn=txn->txnid64;
|
|
}
|
|
}
|
|
}
|
|
//printf("%s:%d Oldest txn is %lld\n", __FILE__, __LINE__, (long long)oldest_live_txn_lsn.lsn);
|
|
|
|
// Now starting at the last one, look for archivable ones.
|
|
// Count the total number of bytes, because we have to return a single big array. (That's the BDB interface. Bleah...)
|
|
LSN earliest_lsn_seen={(unsigned long long)(-1LL)};
|
|
r = peek_at_log(logger, all_logs[all_n_logs-1], &earliest_lsn_seen); // try to find the lsn that's in the most recent log
|
|
if ((earliest_lsn_seen.lsn <= logger->checkpoint_lsn.lsn)&&
|
|
(earliest_lsn_seen.lsn <= oldest_live_txn_lsn.lsn)) {
|
|
i=all_n_logs-1;
|
|
} else {
|
|
for (i=all_n_logs-2; i>=0; i--) { // start at all_n_logs-2 because we never archive the most recent log
|
|
r = peek_at_log(logger, all_logs[i], &earliest_lsn_seen);
|
|
if (r!=0) continue; // In case of error, just keep going
|
|
|
|
//printf("%s:%d file=%s firstlsn=%lld checkpoint_lsns={%lld %lld}\n", __FILE__, __LINE__, all_logs[i], (long long)earliest_lsn_seen.lsn, (long long)logger->checkpoint_lsns[0].lsn, (long long)logger->checkpoint_lsns[1].lsn);
|
|
if ((earliest_lsn_seen.lsn <= logger->checkpoint_lsn.lsn)&&
|
|
(earliest_lsn_seen.lsn <= oldest_live_txn_lsn.lsn)) {
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
// all log files up to, but but not including, i can be archived.
|
|
int n_to_archive=i;
|
|
int count_bytes=0;
|
|
for (i=0; i<n_to_archive; i++) {
|
|
count_bytes+=1+strlen(all_logs[i]);
|
|
}
|
|
char **result;
|
|
if (i==0) {
|
|
result=0;
|
|
} else {
|
|
result = toku_malloc((1+n_to_archive)*sizeof(*result) + count_bytes);
|
|
char *base = (char*)(result+1+n_to_archive);
|
|
for (i=0; i<n_to_archive; i++) {
|
|
int len=1+strlen(all_logs[i]);
|
|
result[i]=base;
|
|
memcpy(base, all_logs[i], len);
|
|
base+=len;
|
|
}
|
|
result[n_to_archive]=0;
|
|
}
|
|
for (i=0; all_logs[i]; i++) {
|
|
toku_free(all_logs[i]);
|
|
}
|
|
toku_free(all_logs);
|
|
*logs_p = result;
|
|
return 0;
|
|
}
|
|
|
|
int toku_maybe_spill_rollbacks (TOKUTXN txn) {
|
|
// Previously:
|
|
// if (txn->rollentry_resident_bytecount>txn->logger->write_block_size) {
|
|
// But now we use t
|
|
if (memarena_total_memory_size(txn->rollentry_arena) > txn->logger->write_block_size) {
|
|
struct roll_entry *item;
|
|
ssize_t bufsize = txn->rollentry_resident_bytecount;
|
|
char *MALLOC_N(bufsize, buf);
|
|
if (bufsize==0) return errno;
|
|
struct wbuf w;
|
|
wbuf_init(&w, buf, bufsize);
|
|
while ((item=txn->oldest_logentry)) {
|
|
assert(item->prev==0);
|
|
u_int32_t rollback_fsize = toku_logger_rollback_fsize(item);
|
|
txn->rollentry_resident_bytecount -= rollback_fsize;
|
|
txn->oldest_logentry = item->next;
|
|
if (item->next) { item->next->prev=0; }
|
|
toku_logger_rollback_wbufwrite(&w, item);
|
|
}
|
|
assert(txn->rollentry_resident_bytecount==0);
|
|
assert((ssize_t)w.ndone==bufsize);
|
|
txn->oldest_logentry = txn->newest_logentry = 0;
|
|
if (txn->rollentry_fd<0) {
|
|
const char filenamepart[] = "/__rolltmp.";
|
|
int fnamelen = strlen(txn->logger->directory)+sizeof(filenamepart)+16;
|
|
assert(txn->rollentry_filename==0);
|
|
txn->rollentry_filename = toku_malloc(fnamelen);
|
|
if (txn->rollentry_filename==0) return errno;
|
|
snprintf(txn->rollentry_filename, fnamelen, "%s%s%.16"PRIx64, txn->logger->directory, filenamepart, txn->txnid64);
|
|
txn->rollentry_fd = open(txn->rollentry_filename, O_CREAT+O_RDWR+O_EXCL+O_BINARY, 0600);
|
|
if (txn->rollentry_fd==-1) return errno;
|
|
}
|
|
uLongf compressed_len = compressBound(w.ndone);
|
|
char *MALLOC_N(compressed_len, compressed_buf);
|
|
{
|
|
int r = compress2((Bytef*)compressed_buf, &compressed_len,
|
|
(Bytef*)buf, w.ndone,
|
|
1);
|
|
assert(r==Z_OK);
|
|
}
|
|
{
|
|
u_int32_t v = toku_htod32(compressed_len);
|
|
ssize_t r = write_it(txn->rollentry_fd, &v, sizeof(v)); assert(r==sizeof(v));
|
|
}
|
|
{
|
|
ssize_t r = write_it(txn->rollentry_fd, compressed_buf, compressed_len);
|
|
if (r<0) return r;
|
|
assert(r==(ssize_t)compressed_len);
|
|
}
|
|
{
|
|
u_int32_t v = toku_htod32(w.ndone);
|
|
ssize_t r = write_it(txn->rollentry_fd, &v, sizeof(v)); assert(r==sizeof(v));
|
|
}
|
|
{
|
|
u_int32_t v = toku_htod32(compressed_len);
|
|
ssize_t r = write_it(txn->rollentry_fd, &v, sizeof(v)); assert(r==sizeof(v));
|
|
}
|
|
toku_free(compressed_buf);
|
|
txn->rollentry_filesize+=w.ndone;
|
|
toku_free(buf);
|
|
|
|
// Cleanup the rollback memory
|
|
memarena_clear(txn->rollentry_arena);
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
int toku_read_rollback_backwards(BREAD br, struct roll_entry **item, MEMARENA ma) {
|
|
u_int32_t nbytes_n; ssize_t sr;
|
|
if ((sr=bread_backwards(br, &nbytes_n, 4))!=4) { assert(sr<0); return errno; }
|
|
u_int32_t n_bytes=toku_dtoh32(nbytes_n);
|
|
unsigned char *buf = malloc_in_memarena(ma, n_bytes);
|
|
if (buf==0) return errno;
|
|
if ((sr=bread_backwards(br, buf, n_bytes-4))!=(ssize_t)n_bytes-4) { assert(sr<0); return errno; }
|
|
int r = toku_parse_rollback(buf, n_bytes, item, ma);
|
|
if (r!=0) return r;
|
|
return 0;
|
|
}
|
|
|
|
static int find_xid (OMTVALUE v, void *txnv) {
|
|
TOKUTXN txn = v;
|
|
TOKUTXN txnfind = txnv;
|
|
if (txn->txnid64<txnfind->txnid64) return -1;
|
|
if (txn->txnid64>txnfind->txnid64) return +1;
|
|
return 0;
|
|
}
|
|
|
|
static int find_filenum (OMTVALUE v, void *brtv) {
|
|
BRT brt = v;
|
|
BRT brtfind = brtv;
|
|
FILENUM fnum = toku_cachefile_filenum(brt ->cf);
|
|
FILENUM fnumfind = toku_cachefile_filenum(brtfind->cf);
|
|
if (fnum.fileid<fnumfind.fileid) return -1;
|
|
if (fnum.fileid>fnumfind.fileid) return +1;
|
|
return 0;
|
|
}
|
|
|
|
|
|
int toku_txn_note_brt (TOKUTXN txn, BRT brt) {
|
|
OMTVALUE txnv;
|
|
u_int32_t index;
|
|
int r = toku_omt_find_zero(brt->txns, find_xid, txn, &txnv, &index, NULL);
|
|
if (r==0) {
|
|
// It's already there.
|
|
assert((TOKUTXN)txnv==txn);
|
|
return 0;
|
|
}
|
|
// Otherwise it's not there.
|
|
r = toku_omt_insert_at(brt->txns, txn, index);
|
|
assert(r==0);
|
|
r = toku_omt_insert(txn->open_brts, brt, find_filenum, brt, 0);
|
|
assert(r==0);
|
|
return 0;
|
|
}
|
|
|
|
static int remove_brt (OMTVALUE txnv, u_int32_t UU(idx), void *brtv) {
|
|
TOKUTXN txn = txnv;
|
|
BRT brt = brtv;
|
|
OMTVALUE brtv_again=NULL;
|
|
u_int32_t index;
|
|
int r = toku_omt_find_zero(txn->open_brts, find_filenum, brt, &brtv_again, &index, NULL);
|
|
assert(r==0);
|
|
assert((void*)brtv_again==brtv);
|
|
r = toku_omt_delete_at(txn->open_brts, index);
|
|
assert(r==0);
|
|
return 0;
|
|
}
|
|
|
|
int toku_txn_note_close_brt (BRT brt) {
|
|
assert(toku_omt_size(brt->txns)==0);
|
|
int r = toku_omt_iterate(brt->txns, remove_brt, brt);
|
|
assert(r==0);
|
|
return 0;
|
|
}
|
|
|
|
static int remove_txn (OMTVALUE brtv, u_int32_t UU(idx), void *txnv) {
|
|
BRT brt = brtv;
|
|
TOKUTXN txn = txnv;
|
|
OMTVALUE txnv_again=NULL;
|
|
u_int32_t index;
|
|
int r = toku_omt_find_zero(brt->txns, find_xid, txn, &txnv_again, &index, NULL);
|
|
assert(r==0);
|
|
assert((void*)txnv_again==txnv);
|
|
r = toku_omt_delete_at(brt->txns, index);
|
|
assert(r==0);
|
|
if (toku_omt_size(brt->txns)==0 && brt->was_closed) {
|
|
//Close immediately.
|
|
assert(brt->close_db);
|
|
r = brt->close_db(brt->db, brt->close_flags);
|
|
}
|
|
return r;
|
|
}
|
|
|
|
// for every BRT in txn, remove it.
|
|
static void note_txn_closing (TOKUTXN txn) {
|
|
toku_omt_iterate(txn->open_brts, remove_txn, txn);
|
|
toku_omt_destroy(&txn->open_brts);
|
|
}
|
|
|
|
int toku_txn_find_by_xid (BRT brt, TXNID xid, TOKUTXN *txnptr) {
|
|
struct tokutxn fake_txn; fake_txn.txnid64 = xid;
|
|
u_int32_t index;
|
|
OMTVALUE txnv;
|
|
int r = toku_omt_find_zero(brt->txns, find_xid, &fake_txn, &txnv, &index, NULL);
|
|
if (r == 0) *txnptr = txnv;
|
|
return r;
|
|
}
|
|
|
|
// Return the number of bytes that went into the rollback data structure (the uncompressed count if there is compression)
|
|
int
|
|
toku_logger_txn_rolltmp_raw_count(TOKUTXN txn, u_int64_t *raw_count)
|
|
{
|
|
*raw_count = txn->rollentry_raw_count;
|
|
return 0;
|
|
}
|
|
|
|
int
|
|
toku_logger_iterate_over_live_txns (TOKULOGGER logger, int (*f)(TOKULOGGER, TOKUTXN, void*), void *v)
|
|
{
|
|
struct list *l;
|
|
for (l = list_head(&logger->live_txns); l != &logger->live_txns; l = l->next) {
|
|
TOKUTXN txn = list_struct(l, struct tokutxn, live_txns_link);
|
|
int r = f(logger, txn, v);
|
|
if (r!=0) return r;
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
TOKUTXN
|
|
toku_logger_txn_parent (TOKUTXN txn)
|
|
{
|
|
return txn->parent;
|
|
}
|
|
|
|
void
|
|
toku_logger_note_checkpoint(TOKULOGGER logger, LSN lsn)
|
|
{
|
|
logger->checkpoint_lsn = lsn;
|
|
}
|