Changeset: cbe8460880f7 for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=cbe8460880f7
Modified Files:
gdk/gdk_bbp.c
gdk/gdk_logger.c
gdk/gdk_logger_internals.h
sql/backends/monet5/sql_gencode.c
sql/storage/bat/bat_storage.c
sql/storage/store.c
Branch: unlock
Log Message:
add locks on gdk_logger
make sure we don't unload bats which have more rows than we committed.
diffs (truncated from 626 to 300 lines):
diff --git a/gdk/gdk_bbp.c b/gdk/gdk_bbp.c
--- a/gdk/gdk_bbp.c
+++ b/gdk/gdk_bbp.c
@@ -2469,6 +2469,14 @@ decref(bat i, bool logical, bool release
}
}
+ /* Make sure we do not unload bats which have more rows than marked
persistend */
+ if (b && BBP_lrefs(i) > 0 && DELTAdirty(b)) {
+ b->batDirtydesc = true;
+ b->theap.dirty = true;
+ if (b->tvheap)
+ b->tvheap->dirty = true;
+ }
+
/* we destroy transients asap and unload persistent bats only
* if they have been made cold or are not dirty */
if (BBP_refs(i) > 0 ||
@@ -2482,6 +2490,7 @@ decref(bat i, bool logical, bool release
assert((BBP_status(i) & BBPUNLOADING) == 0);
TRC_DEBUG(BAT_, "%s set to unloading BAT %d\n", func, i);
BBP_status_on(i, BBPUNLOADING, func);
+ assert(!b || BBP_lrefs(i) == 0 || !DELTAdirty(b));
swap = true;
}
@@ -2767,6 +2776,7 @@ BBPfree(BAT *b, const char *calledFrom)
}
/* clearing bits can be done without the lock */
TRC_DEBUG(BAT_, "turn off unloading %d\n", bid);
+ assert(!b || BBP_lrefs(bid) == 0 || !DELTAdirty(b));
BBP_status_off(bid, BBPUNLOADING, calledFrom);
BBP_unload_dec();
diff --git a/gdk/gdk_logger.c b/gdk/gdk_logger.c
--- a/gdk/gdk_logger.c
+++ b/gdk/gdk_logger.c
@@ -92,6 +92,18 @@ typedef enum {LOG_OK, LOG_EOF, LOG_ERR}
static gdk_return bm_commit(logger *lg);
static gdk_return tr_grow(trans *tr);
+static void
+logger_lock(logger *lg)
+{
+ MT_lock_set(&lg->lock);
+}
+
+static void
+logger_unlock(logger *lg)
+{
+ MT_lock_unset(&lg->lock);
+}
+
static char
find_type(logger *lg, int tpe)
{
@@ -153,6 +165,22 @@ log_find(BAT *b, BAT *d, int val)
return BUN_NONE;
}
+static log_bid
+internal_find_bat(logger *lg, log_id id)
+{
+ BATiter cni = bat_iterator(lg->catalog_id);
+ BUN p;
+
+ if (BAThash(lg->catalog_id) == GDK_SUCCEED) {
+ HASHloop_int(cni, cni.b->thash, p, &id) {
+ oid pos = p;
+ if (BUNfnd(lg->dcatalog, &pos) == BUN_NONE)
+ return *(log_bid *) Tloc(lg->catalog_bid, p);
+ }
+ }
+ return 0;
+}
+
static void
logbat_destroy(BAT *b)
{
@@ -210,7 +238,7 @@ log_read_clear(logger *lg, trans *tr, lo
static gdk_return
la_bat_clear(logger *lg, logaction *la)
{
- log_bid bid = logger_find_bat(lg, la->cid);
+ log_bid bid = internal_find_bat(lg, la->cid);
BAT *b;
if (lg->debug & 1)
@@ -426,7 +454,7 @@ la_bat_update_count(logger *lg, log_id i
static gdk_return
la_bat_updates(logger *lg, logaction *la)
{
- log_bid bid = logger_find_bat(lg, la->cid);
+ log_bid bid = internal_find_bat(lg, la->cid);
if (bid == 0)
return GDK_SUCCEED; /* ignore bats no longer in the catalog */
@@ -515,7 +543,7 @@ log_read_destroy(logger *lg, trans *tr,
static gdk_return
la_bat_destroy(logger *lg, logaction *la)
{
- log_bid bid = logger_find_bat(lg, la->cid);
+ log_bid bid = internal_find_bat(lg, la->cid);
if (bid && logger_del_bat(lg, bid) != GDK_SUCCEED)
return GDK_FAIL;
@@ -1703,6 +1731,7 @@ logger_new(int debug, const char *fn, co
.saved_id = getBBPlogno(), /* get saved log numer
from bbp */
.saved_tid = (int)getBBPtransid(), /* get saved
transaction id from bbp */
};
+ MT_lock_init(&lg->lock, fn);
/* probably open file and check version first, then call call old
logger code */
len = snprintf(filename, sizeof(filename), "%s%c%s%c", logdir, DIR_SEP,
fn, DIR_SEP);
@@ -1737,6 +1766,7 @@ void
logger_destroy(logger *lg)
{
if (lg->catalog_bid) {
+ logger_lock(lg);
BUN p, q;
BAT *b = lg->catalog_bid;
@@ -1758,6 +1788,7 @@ logger_destroy(logger *lg)
logbat_destroy(lg->dcatalog);
logbat_destroy(lg->catalog_cnt);
+ logger_unlock(lg);
}
GDKfree(lg->fn);
GDKfree(lg->dir);
@@ -1830,6 +1861,7 @@ logger_flush(logger *lg)
GDKfree(filename);
}
/* we read the full file because skipping is impossible with current
log format */
+ logger_lock(lg);
lg->flushing = 1;
log_return res = logger_read_transaction(lg);
lg->flushing = 0;
@@ -1843,9 +1875,12 @@ logger_flush(logger *lg)
}
/* remove old log file */
- if (res != LOG_ERR)
+ if (res != LOG_ERR) {
+ logger_unlock(lg);
return logger_cleanup(lg, lg->saved_id);
+ }
}
+ logger_unlock(lg);
return res == LOG_ERR ? GDK_FAIL : GDK_SUCCEED;
}
@@ -1861,68 +1896,21 @@ logger_changes(logger *lg)
int
logger_sequence(logger *lg, int seq, lng *id)
{
+ logger_lock(lg);
BUN p = log_find(lg->seqs_id, lg->dseqs, seq);
if (p != BUN_NONE) {
*id = *(lng *) Tloc(lg->seqs_val, p);
+ logger_unlock(lg);
return 1;
}
+ logger_unlock(lg);
return 0;
}
-/*
- * Changes made to the BAT descriptor should be stored in the log
- * files. Actually, we need to save the descriptor file, perhaps we
- * should simply introduce a versioning scheme.
- */
-gdk_return
-log_bat_persists(logger *lg, BAT *b, int id)
-{
- char ta = find_type(lg, b->ttype);
- logformat l;
-
- if (logger_add_bat(lg, b, id) != GDK_SUCCEED)
- return GDK_FAIL;
-
- l.flag = LOG_CREATE;
- l.id = id;
- if (!lg->inmemory && !LOG_DISABLED(lg)) {
- if (log_write_format(lg, &l) != GDK_SUCCEED ||
- mnstr_write(lg->output_log, &ta, 1, 1) != 1)
- return GDK_FAIL;
- }
- if (lg->debug & 1)
- fprintf(stderr, "#persists id (%d) bat (%d)\n", id,
b->batCacheid);
- if (lg->inmemory || LOG_DISABLED(lg))
- return GDK_SUCCEED;
- return log_bat(lg, b, id, 0, BATcount(b));
-}
-
-gdk_return
-log_bat_transient(logger *lg, int id)
-{
- log_bid bid = logger_find_bat(lg, id);
- logformat l;
-
- l.flag = LOG_DESTROY;
- l.id = id;
-
- if (lg->inmemory || LOG_DISABLED(lg))
- return GDK_SUCCEED;
-
- if (log_write_format(lg, &l) != GDK_SUCCEED) {
- TRC_CRITICAL(GDK, "write failed\n");
- return GDK_FAIL;
- }
- if (lg->debug & 1)
- fprintf(stderr, "#Logged destroyed bat (%d)\n", id);
-
- return logger_del_bat(lg, bid);
-}
-
static gdk_return
-_log_bat(logger *lg, BAT *b, log_id id, lng offset, lng cnt, int sliced)
+internal_log_bat(logger *lg, BAT *b, log_id id, lng offset, lng cnt, int
sliced)
{
char tpe = find_type(lg, b->ttype);
gdk_return ok = GDK_SUCCEED;
@@ -1979,23 +1967,94 @@ static gdk_return
return ok;
}
+/*
+ * Changes made to the BAT descriptor should be stored in the log
+ * files. Actually, we need to save the descriptor file, perhaps we
+ * should simply introduce a versioning scheme.
+ */
+gdk_return
+log_bat_persists(logger *lg, BAT *b, int id)
+{
+ logger_lock(lg);
+ char ta = find_type(lg, b->ttype);
+ logformat l;
+
+ if (logger_add_bat(lg, b, id) != GDK_SUCCEED) {
+ logger_unlock(lg);
+ return GDK_FAIL;
+ }
+
+ l.flag = LOG_CREATE;
+ l.id = id;
+ if (!lg->inmemory && !LOG_DISABLED(lg)) {
+ if (log_write_format(lg, &l) != GDK_SUCCEED ||
+ mnstr_write(lg->output_log, &ta, 1, 1) != 1) {
+ logger_unlock(lg);
+ return GDK_FAIL;
+ }
+ }
+ if (lg->debug & 1)
+ fprintf(stderr, "#persists id (%d) bat (%d)\n", id,
b->batCacheid);
+ if (lg->inmemory || LOG_DISABLED(lg)) {
+ logger_unlock(lg);
+ return GDK_SUCCEED;
+ }
+ gdk_return r = internal_log_bat(lg, b, id, 0, BATcount(b), 0);
+ logger_unlock(lg);
+ return r;
+}
+
+gdk_return
+log_bat_transient(logger *lg, int id)
+{
+ logger_lock(lg);
+ log_bid bid = internal_find_bat(lg, id);
+ logformat l;
+
+ l.flag = LOG_DESTROY;
+ l.id = id;
+
+ if (lg->inmemory || LOG_DISABLED(lg)) {
+ logger_unlock(lg);
+ return GDK_SUCCEED;
+ }
+
+ if (log_write_format(lg, &l) != GDK_SUCCEED) {
+ TRC_CRITICAL(GDK, "write failed\n");
+ logger_unlock(lg);
+ return GDK_FAIL;
+ }
+ if (lg->debug & 1)
+ fprintf(stderr, "#Logged destroyed bat (%d)\n", id);
+ gdk_return r = logger_del_bat(lg, bid);
+ logger_unlock(lg);
+ return r;
+}
+
gdk_return
log_bat(logger *lg, BAT *b, log_id id, lng offset, lng cnt)
{
- return _log_bat(lg, b, id, offset, cnt, 0);
+ logger_lock(lg);
+ gdk_return r = internal_log_bat(lg, b, id, offset, cnt, 0);
+ logger_unlock(lg);
_______________________________________________
checkin-list mailing list
[email protected]
https://www.monetdb.org/mailman/listinfo/checkin-list
