Changeset: f309967b399d for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB/rev/f309967b399d
Modified Files:
gdk/gdk_logger.c
gdk/gdk_logger.h
gdk/gdk_logger_internals.h
sql/storage/bat/bat_logger.c
sql/storage/sql_storage.h
sql/storage/store.c
Branch: group-commit
Log Message:
Implement group-commit.
diffs (297 lines):
diff --git a/gdk/gdk_logger.c b/gdk/gdk_logger.c
--- a/gdk/gdk_logger.c
+++ b/gdk/gdk_logger.c
@@ -2093,6 +2093,12 @@ logger_new(int debug, const char *fn, co
fprintf(stderr, "#logger_new dir set to %s\n", lg->dir);
}
+ // flush variables
+ MT_lock_init(&lg->flush_lock, "flush_lock");
+ MT_lock_init(&lg->flush_queue_lock, "flush_queue_lock");
+ lg->flush_queue_begin = 0;
+ lg->flush_queue_length = 0;
+
if (logger_load(debug, fn, logdir, lg, filename) == GDK_SUCCEED) {
return lg;
}
@@ -2699,24 +2705,82 @@ log_tend(logger *lg)
return GDK_SUCCEED;
}
- if (log_write_format(lg, &l) != GDK_SUCCEED ||
- mnstr_flush(lg->output_log, MNSTR_FLUSH_DATA) ||
- (!(GDKdebug & NOSYNCMASK) && mnstr_fsync(lg->output_log)) ||
- new_logfile(lg) != GDK_SUCCEED) {
- TRC_CRITICAL(GDK, "write failed\n");
- return GDK_FAIL;
+ gdk_return write_format = log_write_format(lg, &l);
+ return write_format;
+}
+
+void
+add_tid_flush_queue(logger *lg, int tid) {
+ MT_lock_set(&lg->flush_queue_lock);
+ int end = (lg->flush_queue_begin + lg->flush_queue_length) %
FLUSH_QUEUE_SIZE;
+ lg->flush_queue[end] = tid;
+ lg->flush_queue_length++;
+ MT_lock_unset(&lg->flush_queue_lock);
+}
+
+void
+left_truncate_flush_queue(logger *lg, int limit) {
+ MT_lock_set(&lg->flush_queue_lock);
+ lg->flush_queue_begin = (lg->flush_queue_begin + limit) %
FLUSH_QUEUE_SIZE;
+ lg->flush_queue_length -= limit;
+ MT_lock_unset(&lg->flush_queue_lock);
+}
+
+int
+tid_in_flush_queue(logger *lg, int tid) {
+ for (int i = 0; i < lg->flush_queue_length; i++) {
+ int idx = (lg->flush_queue_begin + i) % FLUSH_QUEUE_SIZE;
+ if (lg->flush_queue[idx] == tid) {
+ return 1;
+ }
}
- return GDK_SUCCEED;
+ return 0;
+}
+
+int
+flush_queue_length(logger *lg) {
+ return lg->flush_queue_length;
}
gdk_return
-log_tdone(logger *lg, ulng commit_ts)
+log_tflush(logger *lg, int log_tid) {
+ add_tid_flush_queue(lg, log_tid);
+
+ MT_lock_set(&lg->flush_lock);
+ /* the transaction is not yet flushed */
+ if (tid_in_flush_queue(lg, log_tid)) {
+ /* number of transactions in the group commit */
+ int fqueue_length = flush_queue_length(lg);
+ /* flush + fsync */
+ if (mnstr_flush(lg->output_log, MNSTR_FLUSH_DATA) ||
+ (!(GDKdebug & NOSYNCMASK) &&
mnstr_fsync(lg->output_log)) ||
+ new_logfile(lg) != GDK_SUCCEED) {
+ /* flush failed */
+ MT_lock_unset(&lg->flush_lock);
+ return GDK_FAIL;
+ }
+ else {
+ /* flush succeeded */
+ left_truncate_flush_queue(lg, fqueue_length);
+ MT_lock_unset(&lg->flush_lock);
+ return GDK_SUCCEED;
+ }
+ }
+ /* the transaction was already flushed in a group commit, no need to do
anything */
+ else {
+ MT_lock_unset(&lg->flush_lock);
+ return GDK_SUCCEED;
+ }
+}
+
+gdk_return
+log_tdone(logger *lg, ulng commit_ts, int log_tid)
{
if (lg->debug & 1)
- fprintf(stderr, "#log_tdone %d\n", lg->tid);
+ fprintf(stderr, "#log_tdone %d\n", log_tid);
if (lg->current) {
- lg->current->last_tid = lg->tid;
+ lg->current->last_tid = log_tid;
lg->current->last_ts = commit_ts;
}
return GDK_SUCCEED;
@@ -2884,7 +2948,7 @@ logger_find_bat(logger *lg, log_id id)
gdk_return
-log_tstart(logger *lg, bool flushnow)
+log_tstart(logger *lg, bool flushnow, int *log_tid)
{
logformat l;
@@ -2906,6 +2970,7 @@ log_tstart(logger *lg, bool flushnow)
l.flag = LOG_START;
l.id = ++lg->tid;
+ *log_tid = lg->tid;
if (lg->debug & 1)
fprintf(stderr, "#log_tstart %d\n", lg->tid);
diff --git a/gdk/gdk_logger.h b/gdk/gdk_logger.h
--- a/gdk/gdk_logger.h
+++ b/gdk/gdk_logger.h
@@ -68,9 +68,16 @@ gdk_export gdk_return log_delta(logger *
/* mark end of batgroup insert or clear */
//gdk_export gdk_return log_batgroup_end(logger *lg, oid id);
-gdk_export gdk_return log_tstart(logger *lg, bool flush);
+gdk_export gdk_return log_tstart(logger *lg, bool flush, int *log_tid);
gdk_export gdk_return log_tend(logger *lg);
-gdk_export gdk_return log_tdone(logger *lg, ulng commit_ts);
+gdk_export gdk_return log_tdone(logger *lg, ulng commit_ts, int log_tid);
+
+/* flush functions */
+gdk_export gdk_return log_tflush(logger *lg, int log_tid); /* Flush the WAL to
disk using group commit */
+void add_tid_flush_queue(logger *lg, int tid); /* Add a transaction to the
flush queue */
+void left_truncate_flush_queue(logger *lg, int limit); /* Truncate the first
*limit* elements from the flush queue, i.e. mark those transactions as flushed
*/
+int tid_in_flush_queue(logger *lg, int tid); /* Check if a transaction id is
present in the flush queue, i.e. if the transaction is yet to be flushed */
+int flush_queue_length(logger *lg); /* Get the flush queue length, i.e. the
number of transactions in the group commit */
gdk_export gdk_return log_sequence(logger *lg, int seq, lng id);
gdk_export log_bid logger_find_bat(logger *lg, log_id id);
diff --git a/gdk/gdk_logger_internals.h b/gdk/gdk_logger_internals.h
--- a/gdk/gdk_logger_internals.h
+++ b/gdk/gdk_logger_internals.h
@@ -9,6 +9,8 @@
#ifndef _LOGGER_INTERNALS_H_
#define _LOGGER_INTERNALS_H_
+#define FLUSH_QUEUE_SIZE 2048 /* maximum size of the flush queue, i.e. maximum
number of transactions committing simultaneously */
+
typedef struct logged_range_t {
ulng id; /* log file id */
int first_tid; /* first */
@@ -63,6 +65,13 @@ struct logger {
void *buf;
size_t bufsize;
+
+ /* flush variables */
+ int flush_queue[FLUSH_QUEUE_SIZE]; /* circular array with the current
transactions' ids waiting to be flushed */
+ int flush_queue_begin; /* start index of the queue */
+ int flush_queue_length; /* length of the queue */
+ MT_Lock flush_queue_lock; /* to protect the queue against concurrent
reads and writes */
+ MT_Lock flush_lock; /* so only one transaction can flush to disk at any
given time */
};
struct old_logger {
diff --git a/sql/storage/bat/bat_logger.c b/sql/storage/bat/bat_logger.c
--- a/sql/storage/bat/bat_logger.c
+++ b/sql/storage/bat/bat_logger.c
@@ -3138,9 +3138,9 @@ bl_log_isnew(sqlstore *store)
}
static int
-bl_tstart(sqlstore *store, bool flush)
+bl_tstart(sqlstore *store, bool flush, int *log_tid)
{
- return log_tstart(store->logger, flush) == GDK_SUCCEED ? LOG_OK :
LOG_ERR;
+ return log_tstart(store->logger, flush, log_tid) == GDK_SUCCEED ?
LOG_OK : LOG_ERR;
}
static int
@@ -3150,9 +3150,15 @@ bl_tend(sqlstore *store)
}
static int
-bl_tdone(sqlstore *store, ulng commit_ts)
+bl_tflush(sqlstore *store, int log_tid)
{
- return log_tdone(store->logger, commit_ts) == GDK_SUCCEED ? LOG_OK :
LOG_ERR;
+ return log_tflush(store->logger, log_tid) == GDK_SUCCEED ? LOG_OK :
LOG_ERR;
+}
+
+static int
+bl_tdone(sqlstore *store, ulng commit_ts, int log_tid)
+{
+ return log_tdone(store->logger, commit_ts, log_tid) == GDK_SUCCEED ?
LOG_OK : LOG_ERR;
}
static int
@@ -3552,6 +3558,7 @@ bat_logger_init( logger_functions *lf )
lf->log_isnew = bl_log_isnew;
lf->log_tstart = bl_tstart;
lf->log_tend = bl_tend;
+ lf->log_tflush = bl_tflush;
lf->log_tdone = bl_tdone;
lf->log_sequence = bl_sequence;
lf->get_snapshot_files = bl_snapshot;
diff --git a/sql/storage/sql_storage.h b/sql/storage/sql_storage.h
--- a/sql/storage/sql_storage.h
+++ b/sql/storage/sql_storage.h
@@ -277,9 +277,10 @@ typedef int (*logger_changes_fptr)(struc
typedef int (*logger_get_sequence_fptr) (struct sqlstore *store, int seq, lng
*id);
typedef int (*log_isnew_fptr)(struct sqlstore *store);
-typedef int (*log_tstart_fptr) (struct sqlstore *store, bool flush);
+typedef int (*log_tstart_fptr) (struct sqlstore *store, bool flush, int
*log_tid);
typedef int (*log_tend_fptr) (struct sqlstore *store);
-typedef int (*log_tdone_fptr) (struct sqlstore *store, ulng commit_ts);
+typedef int (*log_tflush_fptr) (struct sqlstore *store, int log_tid);
+typedef int (*log_tdone_fptr) (struct sqlstore *store, ulng commit_ts, int
log_tid);
typedef lng (*log_save_id_fptr) (struct sqlstore *store);
typedef int (*log_sequence_fptr) (struct sqlstore *store, int seq, lng id);
@@ -312,6 +313,7 @@ typedef struct logger_functions {
log_isnew_fptr log_isnew;
log_tstart_fptr log_tstart;
log_tend_fptr log_tend;
+ log_tflush_fptr log_tflush;
log_tdone_fptr log_tdone;
log_save_id_fptr log_save_id;
log_sequence_fptr log_sequence;
diff --git a/sql/storage/store.c b/sql/storage/store.c
--- a/sql/storage/store.c
+++ b/sql/storage/store.c
@@ -3901,6 +3901,9 @@ sql_trans_commit(sql_trans *tr)
if (!list_empty(tr->changes)) {
int flush = 0;
ulng commit_ts = 0, oldest = 0;
+ /* stores the transaction id given at the log_tstart, to be
used by the tflush/tdone,
+ as we cannot rely on the current logger->tid due to the
parallel execution */
+ int log_tid;
MT_lock_set(&store->commit);
@@ -3928,7 +3931,7 @@ sql_trans_commit(sql_trans *tr)
flush = (tr->logchanges > min_changes &&
list_empty(store->changes));
if (flush)
MT_lock_set(&store->flush);
- ok = store->logger_api.log_tstart(store, flush);
+ ok = store->logger_api.log_tstart(store, flush,
&log_tid); /* wal start */
/* log */
for(node *n=tr->changes->h; n && ok == LOG_OK; n =
n->next) {
sql_change *c = n->data;
@@ -3951,11 +3954,16 @@ sql_trans_commit(sql_trans *tr)
ok = store->logger_api.log_sequence(store,
OBJ_SID, store->obj_id);
store->prev_oid = store->obj_id;
if (ok == LOG_OK && !flush)
- ok = store->logger_api.log_tend(store); /*
flush/sync */
+ ok = store->logger_api.log_tend(store); /* wal
end */
+ if (ok == LOG_OK && !flush) {
+ MT_lock_unset(&store->commit); /* release the
commit log when flushing to disk */
+ ok = store->logger_api.log_tflush(store,
log_tid); /* flush/sync */
+ MT_lock_set(&store->commit); /* acquire back
the commit lock */
+ }
store_lock(store);
commit_ts = tr->parent ? tr->parent->tid :
store_timestamp(store);
if (ok == LOG_OK && !flush)
/* mark as done */
- ok = store->logger_api.log_tdone(store,
commit_ts);
+ ok = store->logger_api.log_tdone(store,
commit_ts, log_tid);
} else {
store_lock(store);
commit_ts = tr->parent ? tr->parent->tid :
store_timestamp(store);
@@ -3981,9 +3989,13 @@ sql_trans_commit(sql_trans *tr)
/* when directly flushing: flush logger after changes got
applied */
if (flush) {
if (ok == LOG_OK) {
- ok = store->logger_api.log_tend(store); /*
flush/sync */
- if (ok == LOG_OK)
- ok = store->logger_api.log_tdone(store,
commit_ts); /* mark as done */
+ ok = store->logger_api.log_tend(store); /* wal
end */
+ if (ok == LOG_OK) {
+ ok =
store->logger_api.log_tflush(store, log_tid); /* flush/sync */
+ if (ok == LOG_OK) {
+ ok =
store->logger_api.log_tdone(store, commit_ts, log_tid); /* mark as done */
+ }
+ }
}
MT_lock_unset(&store->flush);
}
_______________________________________________
checkin-list mailing list -- [email protected]
To unsubscribe send an email to [email protected]
