Changeset: 9a60cf2b1475 for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB/rev/9a60cf2b1475
Modified Files:
gdk/gdk_logger.c
gdk/gdk_logger_internals.h
Branch: pax-log
Log Message:
Introduce log_tcommit.
diffs (190 lines):
diff --git a/gdk/gdk_logger.c b/gdk/gdk_logger.c
--- a/gdk/gdk_logger.c
+++ b/gdk/gdk_logger.c
@@ -2537,10 +2537,9 @@ internal_log_bat(logger *lg, BAT *b, log
gdk_return ok = GDK_SUCCEED;
logformat l;
BUN p;
- lng nr;
+ lng nr = cnt;
l.flag = LOG_UPDATE_BULK;
l.id = id;
- nr = cnt;
if (LOG_DISABLED(lg) || !nr) {
/* logging is switched off */
@@ -2839,7 +2838,6 @@ new_logfile(logger *lg)
{
assert(!LOG_DISABLED(lg));
-
const lng log_large = (GDKdebug & FORCEMITOMASK)?LOG_MINI:LOG_LARGE;
gdk_return result = GDK_SUCCEED;
@@ -2850,18 +2848,19 @@ new_logfile(logger *lg)
return GDK_FAIL;
}
if (( p > log_large || (lg->end*1024) > log_large )) {
- if (ATOMIC_GET(&lg->refcount) == 1) {
+ if (ATOMIC_GET(&lg->refcount) == 0) {
lg->id++;
log_close_output(lg);
result = log_open_output(lg);
lg->request_rotation = false;
}
else {
- // Delegate wal rotation to next writer or last flusher.
+ // Delegate wal rotation to next writer or (TODO) last
flusher.
lg->request_rotation = true;
}
}
MT_lock_unset(&lg->rotation_lock);
+
return result;
}
@@ -2883,6 +2882,7 @@ log_tend(logger *lg)
if ((result = log_write_format(lg, &l)) != GDK_SUCCEED)
(void) ATOMIC_DEC(&lg->refcount);
+
return result;
}
@@ -2898,12 +2898,14 @@ log_tdone(logger *lg, ulng commit_ts)
return GDK_SUCCEED;
}
-static gdk_return
+gdk_return
log_tcommit(logger *lg, ulng commit_ts)
{
if (lg->debug & 1)
fprintf(stderr, "#log_tcommit " LLFMT "\n", commit_ts);
+ // TODO: check commit queue number
+
lg->end++;
if (LOG_DISABLED(lg)) {
return GDK_SUCCEED;
@@ -2912,29 +2914,21 @@ log_tcommit(logger *lg, ulng commit_ts)
gdk_return result;
logformat l;
l.flag = LOG_COMMIT;
- l.id = 0; // No purpose for now
+ l.id = 0; // number of transactions to be committed;
if ((result = log_write_format(lg, &l)) != GDK_SUCCEED) {
(void) ATOMIC_DEC(&lg->refcount);
return result;
}
- if (mnstr_flush(lg->output_log, MNSTR_FLUSH_DATA) ||
- (!(GDKdebug & NOSYNCMASK) &&
mnstr_fsync(lg->output_log)) ||
- new_logfile(lg) != GDK_SUCCEED) {
-
- }
-
-
-
return GDK_SUCCEED;
}
-static int
+static unsigned int
request_number_flush_queue(logger *lg) {
// Semaphore protects ring buffer structure in queue against overflowing
static unsigned int _number = 0;
- int result;
+ unsigned int result;
MT_sema_down(&lg->flush_queue_semaphore);
MT_lock_set(&lg->flush_queue_lock);
result = ++_number;
@@ -2984,7 +2978,7 @@ log_tflush(logger* lg, ulng log_file_id,
if (lg->flushnow) {
lg->flushnow = 0;
- log_tdone(lg, commit_ts);
+ if (!commit_ts) log_tdone(lg, commit_ts); // TODO: check if
return log_commit(lg);
}
@@ -2992,7 +2986,11 @@ log_tflush(logger* lg, ulng log_file_id,
return GDK_SUCCEED;
}
- if (log_file_id == lg->id) { // TODO: this check might be a data race
+ (void) ATOMIC_DEC(&lg->refcount);
+
+ MT_lock_set(&lg->rotation_lock);
+ if (log_file_id == lg->id) { // TODO: introduce lg->flushed and get rid
of log_file_id in signature
+ MT_lock_unset(&lg->rotation_lock);
unsigned int number = request_number_flush_queue(lg);
MT_lock_set(&lg->flush_lock);
@@ -3004,9 +3002,8 @@ log_tflush(logger* lg, ulng log_file_id,
if (mnstr_flush(lg->output_log, MNSTR_FLUSH_DATA) ||
(!(GDKdebug & NOSYNCMASK) &&
mnstr_fsync(lg->output_log)) ||
new_logfile(lg) != GDK_SUCCEED) {
- /* first flush failed */
+ // TODO: a failed flush/sync/rotate should be a
very fatal error.
MT_lock_unset(&lg->flush_lock);
- (void) ATOMIC_DEC(&lg->refcount);
return GDK_FAIL;
}
else {
@@ -3015,15 +3012,17 @@ log_tflush(logger* lg, ulng log_file_id,
}
}
/* else the transaction was already flushed in a group commit.
- * No need to do anything */
+ * another transaction successfully flushed my message. no need
to do anything. */
}
- /* else the log file has already rotated and hence my wal messages are
already flushed.
- * No need to do anything */
-
-
- log_tdone(lg, commit_ts);
- (void) ATOMIC_DEC(&lg->refcount);
+ else
+ /* else the log file has already rotated and hence my wal
messages are already flushed.
+ * no need to do anything */
+ MT_lock_unset(&lg->rotation_lock);
+
+
+ if (!commit_ts) log_tdone(lg, commit_ts);
MT_lock_unset(&lg->flush_lock);
+ // TODO: request number for commit message queue.
return GDK_SUCCEED;
}
@@ -3197,7 +3196,7 @@ log_tstart(logger *lg, bool flushnow, ul
{
MT_lock_set(&lg->rotation_lock);
log_lock(lg);
- if (flushnow || lg->request_rotation) {
+ if (flushnow || (lg->request_rotation && ATOMIC_GET(&lg->refcount) ==
0) ) {
lg->id++;
log_close_output(lg);
/* start new file */
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
@@ -42,7 +42,7 @@ struct logger {
void *funcdata;
stream *output_log;
stream *input_log; /* current stream to flush */
- lng end; /* end of pre-allocated blocks for faster
f(data)sync */
+ lng end; /* end of pre-allocated blocks for faster
f(data)sync */ // TODO: only incremen when actual files writes occur.
ATOMIC_TYPE refcount; /* Number of active writers and flushers in the
logger */ // TODO check refcount in c->log and c->end
MT_Lock rotation_lock;
@@ -54,7 +54,7 @@ struct logger {
BAT *catalog_lid; /* last tid, after which it gets
released/destroyed */
BAT *dcatalog; /* deleted from catalog table */
BUN cnt; /* number of persistent bats, incremented on
log flushing */
- BUN deleted; /* number of destroyed persistent bats, needed
for catalog vacuum */
+ BUN deleted; /* number of destroyed persistent bats, needekd
for catalog vacuum */
BAT *seqs_id; /* int id column */
BAT *seqs_val; /* lng value column */
_______________________________________________
checkin-list mailing list -- [email protected]
To unsubscribe send an email to [email protected]
