Skip to content

M
mariadb
Commit 2657607c authored by Bradley C. Kuszmaul's avatar Bradley C. Kuszmaul
Browse files
Options

Logging improvements. Mostly fixes #27. Addresses #455, #27. 

git-svn-id: file:///svn/tokudb@2471 c7de825b-a66e-492c-adef-691d508d4ae1
parent cd58a9b9
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 1996 additions and 54 deletions
newbrt/Makefile
View file @ 2657607c
......@@ -151,7 +151,7 @@ brt.o: $(BRT_INTERNAL_H_INCLUDES) key.h log_header.h
fifo.o: fifo.h brttypes.h
memory.o: memory.h
primes.o: primes.h toku_assert.h
fifo-test: fifo.o memory.o toku_assert.o
fifo-test: fifo.o memory.o toku_assert.o ybt.o
brt-serialize.o: $(BRT_INTERNAL_H_INCLUDES) key.h wbuf.h rbuf.h
brt-bigtest: memory.o ybt.o brt.o pma.o cachetable.o key.o fifo.o brt-serialize.o
brt-bigtest.o: brt.h ../include/db.h
......
newbrt/brt-internal.h
View file @ 2657607c
......@@ -144,28 +144,6 @@ void toku_brtnode_free (BRTNODE *node);
#define DEADBEEF ((void*)0xDEADBEEFDEADBEEF)
#endif
/* tree command types */
enum brt_cmd_type {
BRT_NONE = 0,
BRT_INSERT = 1,
BRT_DELETE = 2,
BRT_DELETE_BOTH = 3,
};
/* tree commands */
struct brt_cmd {
enum brt_cmd_type type;
TXNID xid;
union {
/* insert or delete */
struct brt_cmd_insert_delete {
DBT *key;
DBT *val;
} id;
} u;
};
typedef struct brt_cmd BRT_CMD_S, *BRT_CMD;
struct brtenv {
CACHETABLE ct;
TOKULOGGER logger;
......
newbrt/brt2.c 0 → 100644
View file @ 2657607c
This diff is collapsed.
newbrt/brttypes.h
View file @ 2657607c
......@@ -9,6 +9,8 @@
#endif
#define _FILE_OFFSET_BITS 64
#include "../include/db.h"
typedef struct brt *BRT;
struct brt_header;
struct wbuf;
......@@ -68,4 +70,26 @@ typedef struct intpairarray {
typedef struct cachetable *CACHETABLE;
typedef struct cachefile *CACHEFILE;
/* tree command types */
enum brt_cmd_type {
BRT_NONE = 0,
BRT_INSERT = 1,
BRT_DELETE = 2,
BRT_DELETE_BOTH = 3,
};
/* tree commands */
struct brt_cmd {
enum brt_cmd_type type;
TXNID xid;
union {
/* insert or delete */
struct brt_cmd_insert_delete {
DBT *key;
DBT *val;
} id;
} u;
};
typedef struct brt_cmd BRT_CMD_S, *BRT_CMD;
#endif
newbrt/cachetable.c
View file @ 2657607c
......@@ -106,6 +106,7 @@ int toku_cachefile_of_filenum (CACHETABLE t, FILENUM filenum, CACHEFILE *cf, BRT
return ENOENT;
}
// If something goes wrong, close the fd. After this, the caller shouldn't close the fd, but instead should close the cachefile.
int toku_cachetable_openfd (CACHEFILE *cf, CACHETABLE t, int fd, BRT brt) {
int r;
CACHEFILE extant;
......@@ -114,7 +115,7 @@ int toku_cachetable_openfd (CACHEFILE *cf, CACHETABLE t, int fd, BRT brt) {
struct fileid fileid;
memset(&fileid, 0, sizeof(fileid));
r=fstat(fd, &statbuf);
if (r != 0) return errno;
if (r != 0) { r=errno; close(fd); }
fileid.st_dev = statbuf.st_dev;
fileid.st_ino = statbuf.st_ino;
for (extant = t->cachefiles; extant; extant=extant->next) {
......
newbrt/fifo.c
View file @ 2657607c
......@@ -5,6 +5,7 @@
#include "toku_assert.h"
#include "memory.h"
#include "fifo.h"
#include "ybt.h"
static void fifo_init(struct fifo *fifo) {
fifo->head = fifo->tail = 0;
......@@ -78,6 +79,10 @@ int toku_fifo_enq(FIFO fifo, const void *key, unsigned int keylen, const void *d
return 0;
}
int toku_fifo_enq_cmdstruct (FIFO fifo, const BRT_CMD cmd) {
return toku_fifo_enq(fifo, cmd->u.id.key->data, cmd->u.id.key->size, cmd->u.id.val->data, cmd->u.id.val->size, cmd->type, cmd->xid);
}
/* peek at the head (the oldest entry) of the fifo */
int toku_fifo_peek(FIFO fifo, bytevec *key, unsigned int *keylen, bytevec *data, unsigned int *datalen, int *type, TXNID *xid) {
struct fifo_entry *entry = fifo_peek(fifo);
......@@ -91,6 +96,22 @@ int toku_fifo_peek(FIFO fifo, bytevec *key, unsigned int *keylen, bytevec *data,
return 0;
}
// fill in the BRT_CMD, using the two DBTs for the DBT part.
int toku_fifo_peek_cmdstruct (FIFO fifo, BRT_CMD cmd, DBT*key, DBT*data) {
int type;
bytevec keyb,datab;
unsigned int keylen,datalen;
int r = toku_fifo_peek(fifo, &keyb, &keylen, &datab, &datalen, &type, &cmd->xid);
if (r!=0) return r;
cmd->type=type;
toku_fill_dbt(key, keyb, keylen);
toku_fill_dbt(data, datab, datalen);
cmd->u.id.key=key;
cmd->u.id.val=data;
return 0;
}
int toku_fifo_deq(FIFO fifo) {
struct fifo_entry *entry = fifo_deq(fifo);
if (entry == 0) return -1; // if entry is 0 then it was an empty fifo
......
newbrt/fifo.h
View file @ 2657607c
#ifndef FIFO_H
#define FIFO_H
#include "brttypes.h"
struct fifo_entry {
......@@ -19,8 +21,10 @@ typedef struct fifo *FIFO;
int toku_fifo_create(FIFO *);
void toku_fifo_free(FIFO *);
int toku_fifo_n_entries(FIFO);
int toku_fifo_enq_cmdstruct (FIFO fifo, const BRT_CMD cmd);
int toku_fifo_enq (FIFO, const void *key, ITEMLEN keylen, const void *data, ITEMLEN datalen, int type, TXNID xid);
int toku_fifo_peek (FIFO, bytevec *key, ITEMLEN *keylen, bytevec *data, ITEMLEN *datalen, int *type, TXNID *xid);
int toku_fifo_peek_cmdstruct (FIFO, BRT_CMD, DBT*, DBT*); // fill in the BRT_CMD, using the two DBTs for the DBT part.
int toku_fifo_deq(FIFO);
int toku_fifo_peek_deq (FIFO, bytevec *key, ITEMLEN *keylen, bytevec *data, ITEMLEN *datalen, int *type);
void toku_fifo_iterate (FIFO, void(*f)(bytevec key,ITEMLEN keylen,bytevec data,ITEMLEN datalen,int type, TXNID xid, void*), void*);
......@@ -38,3 +42,4 @@ void toku_fifo_iterate (FIFO, void(*f)(bytevec key,ITEMLEN keylen,bytevec data,I
} \
})
#endif
newbrt/log.c
View file @ 2657607c
......@@ -633,5 +633,7 @@ int toku_txnid2txn (TOKULOGGER logger, TXNID txnid, TOKUTXN *result) {
return 0;
}
}
return -1;
// If there is no txn, then we treat it as the null txn.
*result = 0;
return 0;
}
newbrt/logformat.c
View file @ 2657607c
......@@ -54,11 +54,13 @@ const struct logtype rollbacks[] = {
const struct logtype logtypes[] = {
{"commit", 'C', FA{{"TXNID", "txnid", 0},NULLFIELD}},
{"delete", 'D', FA{{"FILENUM", "filenum", 0},
#if 0
{"tl_delete", 'D', FA{{"FILENUM", "filenum", 0}, // tl logentries can be used, by themselves, to rebuild the whole DB from scratch.
{"DISKOFF", "diskoff", 0},
{"BYTESTRING", "key", 0},
{"BYTESTRING", "data", 0},
NULLFIELD}},
#endif
{"fcreate", 'F', FA{{"TXNID", "txnid", 0},
{"BYTESTRING", "fname", 0},
{"u_int32_t", "mode", "0%o"},
......
newbrt/pma.c
View file @ 2657607c
......@@ -717,23 +717,27 @@ int toku_pma_insert (PMA pma, DBT *k, DBT *v, TOKULOGGER logger, TXNID xid, FILE
*fingerprint += rand4fingerprint*toku_calccrc32_kvpair(k->data, k->size, v->data, v->size);
struct kv_pair *pair = pma->pairs[idx];
{
TOKUTXN txn;
int r;
if ((r=toku_txnid2txn(logger, xid, &txn))) return r;
const BYTESTRING key = { pair->keylen, toku_memdup(kv_pair_key_const(pair), pair->keylen) };
const BYTESTRING data = { pair->vallen, toku_memdup(kv_pair_val_const(pair), pair->vallen) };
if ((r = toku_logger_save_rollback_insertatleaf(txn, pma->filenum, key, data))) {
toku_free(key.data); toku_free(data.data);
return r;
if (logger) {
{
TOKUTXN txn;
int r;
if ((r=toku_txnid2txn(logger, xid, &txn))) return r;
if (txn) {
const BYTESTRING key = { pair->keylen, toku_memdup(kv_pair_key_const(pair), pair->keylen) };
const BYTESTRING data = { pair->vallen, toku_memdup(kv_pair_val_const(pair), pair->vallen) };
if ((r = toku_logger_save_rollback_insertatleaf(txn, pma->filenum, key, data))) {
toku_free(key.data); toku_free(data.data);
return r;
}
}
}
{
const BYTESTRING key = { pair->keylen, kv_pair_key(pair) };
const BYTESTRING data = { pair->vallen, kv_pair_val(pair) };
int r = toku_log_insertinleaf (logger, xid, pma->filenum, diskoff, idx, key, data);
if (r!=0) return r;
if (node_lsn) *node_lsn = toku_logger_last_lsn(logger);
}
}
{
const BYTESTRING key = { pair->keylen, kv_pair_key(pair) };
const BYTESTRING data = { pair->vallen, kv_pair_val(pair) };
int r = toku_log_insertinleaf (logger, xid, pma->filenum, diskoff, idx, key, data);
if (r!=0) return r;
if (logger && node_lsn) *node_lsn = toku_logger_last_lsn(logger);
}
return 0;
......@@ -773,11 +777,14 @@ static int pma_log_delete (PMA pma, const char *key, int keylen, const char *val
}
if (logger) {
TOKUTXN txn;
if (0!=toku_txnid2txn(logger, xid, &txn)) return -1;
const BYTESTRING deletedkey = { keylen, toku_memdup(key, keylen) };
const BYTESTRING deleteddata = { vallen, toku_memdup(val, vallen) };
int r=toku_logger_save_rollback_deleteatleaf(txn, pma->filenum, deletedkey, deleteddata);
if (r!=0) { toku_free(deletedkey.data); toku_free(deleteddata.data); return r;
int r=toku_txnid2txn(logger, xid, &txn);
if (r!=0) return r;
if (txn) {
const BYTESTRING deletedkey = { keylen, toku_memdup(key, keylen) };
const BYTESTRING deleteddata = { vallen, toku_memdup(val, vallen) };
r=toku_logger_save_rollback_deleteatleaf(txn, pma->filenum, deletedkey, deleteddata);
if (r!=0) { toku_free(deletedkey.data); toku_free(deleteddata.data); return r;
}
}
if (node_lsn) *node_lsn = toku_logger_last_lsn(logger);
}
......@@ -926,11 +933,13 @@ int toku_pma_insert_or_replace (PMA pma, DBT *k, DBT *v,
{
TOKUTXN txn;
if ((r=toku_txnid2txn(logger, xid, &txn))) return r;
const BYTESTRING key = { k->size, toku_memdup(k->data, k->size) };
const BYTESTRING data = { v->size, toku_memdup(v->data, v->size) };
if ((r = toku_logger_save_rollback_insertatleaf(txn, pma->filenum, key, data))) {
toku_free(key.data); toku_free(data.data);
return r;
if (txn) {
const BYTESTRING key = { k->size, toku_memdup(k->data, k->size) };
const BYTESTRING data = { v->size, toku_memdup(v->data, v->size) };
if ((r = toku_logger_save_rollback_insertatleaf(txn, pma->filenum, key, data))) {
toku_free(key.data); toku_free(data.data);
return r;
}
}
}
{
......
newbrt/roll.c
View file @ 2657607c
......@@ -96,8 +96,6 @@ void toku_recover_commit (LSN UU(lsn), TXNID UU(txnid)) {
#define ABORTIT { le=le; txn=txn; fprintf(stderr, "%s:%d (%s) not ready to go\n", __FILE__, __LINE__, __func__); abort(); }
void toku_recover_delete (LSN UU(lsn), FILENUM UU(filenum),DISKOFF UU(diskoff),BYTESTRING UU(key),BYTESTRING UU(data)) { assert(0); }
void toku_recover_fcreate (LSN UU(lsn), TXNID UU(txnid),BYTESTRING fname,u_int32_t mode) {
char *fixed_fname = fixup_fname(&fname);
int fd = creat(fixed_fname, mode);
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7