Changeset: 1b1b3916ebd3 for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=1b1b3916ebd3
Modified Files:
gdk/gdk_logger.c
sql/include/sql_catalog.h
sql/server/sql_mvc.c
sql/storage/bat/bat_storage.c
sql/storage/store.c
Branch: unlock
Log Message:
we now allow for concurrent inserts
diffs (truncated from 633 to 300 lines):
diff --git a/gdk/gdk_logger.c b/gdk/gdk_logger.c
--- a/gdk/gdk_logger.c
+++ b/gdk/gdk_logger.c
@@ -338,21 +338,25 @@ log_read_updates(logger *lg, trans *tr,
tv = lg->buf;
assert(nr <= (lng) BUN_MAX);
- if (l->flag == LOG_UPDATE) {
+ if (!lg->flushing && l->flag == LOG_UPDATE) {
uid = COLnew(0, TYPE_oid, (BUN)nr, PERSISTENT);
if (uid == NULL) {
return LOG_ERR;
}
}
- r = COLnew(0, tpe, (BUN) nr, PERSISTENT);
- if (r == NULL) {
- BBPreclaim(uid);
- return LOG_ERR;
+ if (!lg->flushing) {
+ r = COLnew(0, tpe, (BUN) nr, PERSISTENT);
+ if (r == NULL) {
+ if (uid)
+ BBPreclaim(uid);
+ return LOG_ERR;
+ }
}
if (l->flag == LOG_UPDATE_BULK) {
if (!mnstr_readLng(lg->input_log, &offset)) {
- BBPreclaim(r);
+ if (r)
+ BBPreclaim(r);
return LOG_ERR;
}
for (; res == LOG_OK && nr > 0; nr--) {
@@ -2075,11 +2079,10 @@ log_delta(logger *lg, BAT *uid, BAT *uva
if (log_write_format(lg, &l) != GDK_SUCCEED ||
!mnstr_writeLng(lg->output_log, nr) ||
- mnstr_write(lg->output_log, &tpe, 1, 1) != 1) {
+ mnstr_write(lg->output_log, &tpe, 1, 1) != 1){
logger_unlock(lg);
return GDK_FAIL;
}
- /* TODO call efficient n-value writes */
for (p = 0; p < BUNlast(uid) && ok == GDK_SUCCEED; p++) {
const oid id = BUNtoid(uid, p);
diff --git a/sql/include/sql_catalog.h b/sql/include/sql_catalog.h
--- a/sql/include/sql_catalog.h
+++ b/sql/include/sql_catalog.h
@@ -195,6 +195,7 @@ typedef int sqlid;
typedef struct sql_base {
int wtime;
int rtime;
+ int atime; /* appends should not conflict */
int stime;
int allocated;
int flags;
@@ -242,6 +243,7 @@ typedef struct sql_trans {
int wstime; /* first write transaction time stamp */
int rtime; /* timestamp of latest read performed in
transaction*/
int wtime; /* timestamp of latest write performed in
transaction*/
+ int atime;
int schema_number; /* schema timestamp */
int schema_updates; /* set on schema changes */
int active; /* active transaction */
diff --git a/sql/server/sql_mvc.c b/sql/server/sql_mvc.c
--- a/sql/server/sql_mvc.c
+++ b/sql/server/sql_mvc.c
@@ -509,6 +509,8 @@ sql_trans_deref( sql_trans *tr )
if (t->base.rtime < p->base.rtime)
t->base.rtime = p->base.rtime;
+ if (t->base.atime < p->base.atime)
+ t->base.atime = p->base.atime;
if (t->base.wtime < p->base.wtime)
t->base.wtime = p->base.wtime;
t->po = p->po;
@@ -525,6 +527,8 @@ sql_trans_deref( sql_trans *tr )
if (c->base.rtime <
p->base.rtime)
c->base.rtime =
p->base.rtime;
+ if (c->base.atime <
p->base.atime)
+ c->base.atime =
p->base.atime;
if (c->base.wtime <
p->base.wtime)
c->base.wtime =
p->base.wtime;
c->po = p->po;
@@ -547,6 +551,8 @@ sql_trans_deref( sql_trans *tr )
if (i->base.rtime < p->base.rtime)
i->base.rtime = p->base.rtime;
+ if (i->base.atime < p->base.atime)
+ i->base.atime = p->base.atime;
if (i->base.wtime < p->base.wtime)
i->base.wtime = p->base.wtime;
i->po = p->po;
@@ -633,7 +639,7 @@ build up the hash (not copied in the tra
store_lock();
/* if there is nothing to commit reuse the current transaction */
- if (tr->wtime == 0) {
+ if (tr->wtime == 0 && tr->atime == 0) {
if (!chain)
sql_trans_end(m->session, 1);
m->type = Q_TRANS;
@@ -708,7 +714,7 @@ mvc_rollback(mvc *m, int chain, const ch
tr = m->session->tr;
while (!tr->name || strcmp(tr->name, name) != 0) {
/* make sure we do not reuse changed data */
- if (tr->wtime)
+ if (tr->wtime || tr->atime)
tr->status = 1;
tr = sql_trans_destroy(tr, true);
}
@@ -724,7 +730,7 @@ mvc_rollback(mvc *m, int chain, const ch
}
m->session-> tr = tr;
/* make sure we do not reuse changed data */
- if (tr->wtime)
+ if (tr->wtime || tr->atime)
tr->status = 1;
sql_trans_end(m->session, 0);
if (chain)
@@ -740,7 +746,7 @@ mvc_rollback(mvc *m, int chain, const ch
TRC_INFO(SQL_TRANS,
"Commit%s%s rolled back%s%s%.200s\n",
name ? " " : "", name ? name : "",
- tr->wtime == 0 ? " (no changes)" : "",
+ (tr->wtime == 0 || tr->atime) ? " (no changes)" : "",
m->query ? ", query: " : "",
m->query ? m->query : "");
return msg;
diff --git a/sql/storage/bat/bat_storage.c b/sql/storage/bat/bat_storage.c
--- a/sql/storage/bat/bat_storage.c
+++ b/sql/storage/bat/bat_storage.c
@@ -13,6 +13,7 @@
#include "algebra.h"
#include "gdk_atoms.h"
+static MT_Lock segs_lock = MT_LOCK_INITIALIZER("segs_lock");
static MT_Lock destroy_lock = MT_LOCK_INITIALIZER("destroy_lock");
storage *tobe_destroyed_dbat = NULL;
sql_delta *tobe_destroyed_delta = NULL;
@@ -363,7 +364,6 @@ cs_update_bat( column_storage *cs, BAT *
if (b == NULL)
return LOG_ERR;
if (BATcount(b)==0) { /* alter add column */
- assert(BATtdense(tids));
if (BATappend(b, updates, NULL, true) != GDK_SUCCEED) {
bat_destroy(b);
return LOG_ERR;
@@ -579,7 +579,7 @@ delta_append_bat( sql_delta *bat, size_t
}
bat_destroy(ui);
} else {
- assert (isNew(t) || isTempTable(t) || bat->cs.cleared);
+ //assert (isNew(t) || isTempTable(t) || bat->cs.cleared);
if (BATcount(b) < offset) { /* add space */
const void *tv = ATOMnilptr(b->ttype);
lng i, d = offset - BATcount(b);
@@ -624,7 +624,7 @@ delta_append_val( sql_delta *bat, size_t
return LOG_ERR;
}
} else {
- assert (isNew(t) || isTempTable(t) || bat->cs.cleared);
+ //assert (isNew(t) || isTempTable(t) || bat->cs.cleared);
if (BATcount(b) < offset) { /* add space */
const void *tv = ATOMnilptr(b->ttype);
lng i, d = offset - BATcount(b);
@@ -731,7 +731,9 @@ dup_dbat(storage *obat, storage *bat, in
bat->end = bat->segs->end;
} else {
bat->end = obat->end = obat->segs->end;
+ MT_lock_set(&segs_lock);
bat->segs = dup_segments(obat->segs);
+ MT_lock_unset(&segs_lock);
assert(bat->end <= bat->segs->end);
}
return dup_cs(&obat->cs, &bat->cs, TYPE_bit, is_new, temp);
@@ -767,11 +769,7 @@ append_col(sql_trans *tr, sql_column *c,
bat = c->data;
/* appends only write */
- bat->cs.wtime = c->base.wtime = c->t->base.wtime = c->t->s->base.wtime
= tr->wtime = tr->wstime;
- /* inserts are ordered with the current delta implementation */
- /* therefor mark appends as reads */
- assert(tr != gtrans);
- c->t->s->base.rtime = c->t->base.rtime = tr->stime;
+ bat->cs.wtime = c->base.atime = c->t->base.atime = c->t->s->base.atime
= tr->atime = tr->wstime;
if (tpe == TYPE_bat)
ok = delta_append_bat(bat, offset, i, c->t);
else
@@ -794,7 +792,7 @@ append_idx(sql_trans *tr, sql_idx * i, s
bat = i->data;
/* appends only write */
- bat->cs.wtime = i->base.wtime = i->t->base.wtime = i->t->s->base.wtime
= tr->wtime = tr->wstime;
+ bat->cs.wtime = i->base.atime = i->t->base.atime = i->t->s->base.atime
= tr->atime = tr->wstime;
if (tpe == TYPE_bat)
ok = delta_append_bat(bat, offset, ib, i->t);
else
@@ -887,12 +885,12 @@ table_claim_space(sql_trans *tr, sql_tab
node *n = t->columns.set->h;
sql_column *c = n->data;
- t->base.wtime = t->s->base.wtime = tr->wtime = tr->wstime;
- c->base.wtime = tr->wstime;
+ t->base.atime = t->s->base.atime = tr->atime = tr->wstime;
+ c->base.atime = tr->wstime;
for (n = t->columns.set->h; n; n = n->next) {
c = n->data;
- c->base.wtime = tr->wstime;
+ c->base.atime = tr->wstime;
if (bind_col_data(tr, c) == LOG_ERR)
return LOG_ERR;
@@ -903,7 +901,7 @@ table_claim_space(sql_trans *tr, sql_tab
for (n = t->idxs.set->h; n; n = n->next) {
sql_idx *ci = n->data;
- ci->base.wtime = tr->wstime;
+ ci->base.atime = tr->wstime;
if (isTable(ci->t) && idx_has_column(ci->type)) {
if (bind_idx_data(tr, ci) == LOG_ERR)
return LOG_ERR;
@@ -969,7 +967,7 @@ claim_tab(sql_trans *tr, sql_table *t, s
/* add updates */
BAT *ui, *uv;
- if (table_claim_space(tr, t, cnt) == LOG_ERR) {
+ if (0 && table_claim_space(tr, t, cnt) == LOG_ERR) {
store_unlock();
return LOG_ERR;
}
@@ -1007,7 +1005,7 @@ claim_tab(sql_trans *tr, sql_table *t, s
assert(isTempTable(t) || s->cs.cleared || ps->cs.bid == s->cs.bid);
/* inserts only write */
- s->cs.wtime = t->base.wtime = t->s->base.wtime = tr->wtime = tr->wstime;
+ s->cs.wtime = t->base.atime = t->s->base.atime = tr->atime = tr->wstime;
store_unlock();
return (size_t)slot;
}
@@ -1426,8 +1424,12 @@ destroy_segs(segment *s)
static void
destroy_segments(segments *s)
{
- if (sql_ref_dec(&s->r) > 0)
+ MT_lock_set(&segs_lock);
+ if (sql_ref_dec(&s->r) > 0) {
+ MT_lock_unset(&segs_lock);
return;
+ }
+ MT_lock_unset(&segs_lock);
destroy_segs(s->head);
_DELETE(s);
}
@@ -1691,7 +1693,7 @@ minmax( sql_trans *tr )
for(sn = tr->schemas.set->h; sn && ok == LOG_OK; sn = sn->next) {
sql_schema *s = sn->data;
- if (!s->base.wtime)
+ if (!s->base.wtime && !s->base.atime)
continue;
if (!isTempSchema(s) && s->tables.set) {
node *n;
@@ -1853,7 +1855,9 @@ tr_update_dbat( sql_trans *tr, storage *
{
if (fs->cs.cleared) {
destroy_segments(ts->segs);
+ MT_lock_set(&segs_lock);
ts->segs = dup_segments(fs->segs);
+ MT_lock_unset(&segs_lock);
ts->end = ts->segs->end;
assert(ts->segs->head);
} else {
@@ -1950,8 +1954,7 @@ update_table(sql_trans *tr, sql_table *f
sql_column *cc = n->data; // TODO: either stick to to/from
terminology or old/current terminology
sql_column *oc = m->data;
- if (ATOMIC_GET(&store_nr_active) == 1 || (cc->base.wtime &&
cc->base.allocated)) {
- assert(!cc->base.wtime || oc->base.wtime <
cc->base.wtime);
+ if (ATOMIC_GET(&store_nr_active) == 1 || ((cc->base.wtime ||
cc->base.atime) && cc->base.allocated)) {
if (ATOMIC_GET(&store_nr_active) > 1 && cc->data) { /*
move delta */
sql_delta *b = cc->data;
sql_column *oldc = NULL;
@@ -2013,6 +2016,8 @@ update_table(sql_trans *tr, sql_table *f
if (oc->base.rtime < cc->base.rtime)
oc->base.rtime = cc->base.rtime;
+ if (oc->base.atime < cc->base.atime)
+ oc->base.atime = cc->base.atime;
if (oc->base.wtime < cc->base.wtime)
oc->base.wtime = cc->base.wtime;
if (cc->data)
@@ -2033,7 +2038,7 @@ update_table(sql_trans *tr, sql_table *f
ci->base.allocated = 0;
continue;
_______________________________________________
checkin-list mailing list
[email protected]
https://www.monetdb.org/mailman/listinfo/checkin-list
