From 5dd5ea462c2405edd3504a52a3099d47fc7453d6 Mon Sep 17 00:00:00 2001
From: Thomas Schoebel-Theuer <tst@1und1.de>
Date: Mon, 7 Sep 2020 11:40:14 +0200
Subject: [PATCH] main: use the logger epoch

---
 kernel/sy_old/mars_main.c | 60 ++++++++++++++++++++++++++++++++++++++-
 1 file changed, 59 insertions(+), 1 deletion(-)

diff --git a/kernel/sy_old/mars_main.c b/kernel/sy_old/mars_main.c
index acc579b5..d549373d 100644
--- a/kernel/sy_old/mars_main.c
+++ b/kernel/sy_old/mars_main.c
@@ -852,6 +852,7 @@ struct mars_rotate {
 	struct mars_dent *next_log;
 	struct mars_dent *syncstatus_dent;
 	struct lamport_time sync_finish_stamp;
+	struct lamport_time prosumer_epoch;
 	struct client_brick *prosumer_brick;
 	struct client_brick *prosumer_new;
 	struct client_brick *prosumer_old;
@@ -4695,8 +4696,30 @@ int make_log_init(struct mars_dent *dent)
 			       parent_path,
 			       my_id());
 	mutex_unlock(&server_connect_lock);
-	if (!rot->trans_brick && trans_brick)
+	/* Recovery after a crash:
+	 * upon first instantiation, read the old persisted epoch stamp
+	 */
+	if (!rot->trans_brick && trans_brick) {
+		struct trans_logger_brick *tra = (void *)trans_brick;
+		const char *epoch_path;
+		const char *epoch_str;
+
+		epoch_path = path_make("%s/actual-%s/state-logger",
+				       parent_path,
+				       my_id());
+		epoch_str = ordered_readlink(epoch_path, NULL);
+		sscanf(epoch_str,
+#ifdef CURRENT_TIME
+		       "%lld.%ld",
+#else
+		       "%lld.%ld",
+#endif
+		       &tra->logger_epoch.tv_sec,
+		       &tra->logger_epoch.tv_nsec);
+		brick_string_free(epoch_path);
+		brick_string_free(epoch_str);
 		clear_vals(rot->msgs);
+	}
 	rot->trans_brick = (void*)trans_brick;
 	status = -ENOENT;
 	if (!trans_brick) {
@@ -5487,6 +5510,14 @@ int _start_trans(struct mars_rotate *rot)
 		MARS_ERR("initial connect failed\n");
 		goto done;
 	}
+	/* Reset the epoch stamp when necessary */
+	if (!trans_brick->logger_epoch.tv_sec ||
+	    (trans_brick->replay_mode &&
+	     (!rot->relevant_log->d_rest ||
+	      strcmp(rot->relevant_log->d_rest, my_id()) != 0))) {
+		MARS_DBG("reset logger_epoch '%s'\n", rot->parent_path);
+		get_lamport(NULL, &trans_brick->logger_epoch);
+	}
 
 	_change_trans(rot);
 
@@ -5562,6 +5593,30 @@ bool _is_secondary_fixing_safe(struct mars_rotate *rot)
 	return true;
 }
 
+static
+void __reset_all_epoch(struct mars_rotate *rot)
+{
+	const char *prosumer_epoch_path;
+	struct kstat old_epoch = {};
+
+	get_lamport(NULL, &rot->trans_brick->logger_epoch);
+
+	/* prevent conflicts with ongoing prosumer handover */
+	if (rot->old_inhibit_mask ||
+	    (rot->gate_brick && rot->gate_brick->inhibit_mask))
+		return;
+
+	prosumer_epoch_path = path_make("%s/actual-%s/prosumer-epoch",
+					rot->parent_path,
+					my_id());
+	/* race avoidance: only add small epsilon */
+	mars_stat(prosumer_epoch_path, &old_epoch, true);
+	if (old_epoch.mtime.tv_sec)
+		lamport_time_add_ns(&old_epoch.mtime, 1);
+	ordered_symlink("(none)", prosumer_epoch_path, &old_epoch.mtime);
+	brick_string_free(prosumer_epoch_path);
+}
+
 static
 int make_log_finalize(struct mars_dent *dent)
 {
@@ -5704,6 +5759,8 @@ int make_log_finalize(struct mars_dent *dent)
 				     trans_brick->replay_end_pos - trans_brick->replay_current_pos);
 			rot->replay_code = trans_brick->replay_code;
 			rot->log_is_really_damaged = true;
+			/* For safety, reset both epoch stamps */
+			__reset_all_epoch(rot);
 			/* Exception: set actual position for recovery */
 			_recover_versionlink(rot,
 					     rot->current_inf.inf_host,
@@ -6703,6 +6760,7 @@ int make_dev(struct mars_dent *dent)
 		goto done;
 	}
 	rot->if_brick->shutdown = rot->kill_device;
+	rot->if_brick->open_epoch = &rot->prosumer_epoch;
 	dev_brick->killme = true;
 	dev_brick->kill_ptr = (void**)&rot->if_brick;
 	dev_brick->rewire = true;