changes, bugfixes and refactoring

Author: Dantas198

pom.xml

@@ -17,6 +17,11 @@
                     <target>8</target>
                 </configuration>
             </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <version>2.19.1</version>
+            </plugin>
         </plugins>
     </build>
 

src/main/java/certifier/AbstractCertifier.java

@@ -17,6 +17,7 @@ public abstract class AbstractCertifier implements Certifier<Long>{
 
     final Timestamp<Long> currentCommitTs;
     final Timestamp<Long> lowWaterMark;
+    final Timestamp<Long> lastTombStone;
 
     final LinkedHashMap<Timestamp<Long>, RunningState> runningTransactions;
 
@@ -28,6 +29,7 @@ public AbstractCertifier(long timestep){
         lowWaterMark = new MonotonicTimestamp(-1);
         runningTransactions = new LinkedHashMap<>();
         history = new HashMap<>();
+        lastTombStone = new MonotonicTimestamp(timestep);
     }
 
     public AbstractCertifier(AbstractCertifier certifier){
@@ -36,12 +38,15 @@ public AbstractCertifier(AbstractCertifier certifier){
         lowWaterMark = certifier.lowWaterMark;
         runningTransactions = new LinkedHashMap<>(certifier.runningTransactions);
         history = new HashMap<>(certifier.history);
+        lastTombStone = certifier.lastTombStone;
     }
 
     public abstract CompletableFuture<Timestamp<Long>> start();
 
     public abstract long truncateStartTS(Timestamp<Long> startTimestamp);
 
+    public abstract long truncateForGC(Timestamp<Long> startTimestamp);
+
     public abstract Timestamp<Long> treatCommit(BitWriteSet newBws, Timestamp<Long> ts);
 
     public abstract void update(Timestamp<Long> commitTimestamp);
@@ -67,35 +72,37 @@ public Timestamp<Long> commit(BitWriteSet newBws, Timestamp<Long> ts) {
 
     @Override
     public void transactionStarted(Timestamp<Long> startTimestamp) {
-        Timestamp<Long> truncated = new MonotonicTimestamp(truncateStartTS(startTimestamp));
+        Timestamp<Long> truncated = new MonotonicTimestamp(truncateForGC(startTimestamp));
         LOG.info("transaction started truncated={}", truncated.toPrimitive());
         this.runningTransactions.putIfAbsent(truncated, new RunningState());
         this.runningTransactions.get(truncated).addTransaction();
     }
 
     @Override
-    public void transactionCommited(Timestamp<Long> startTimestamp){
-        Timestamp<Long> truncated = new MonotonicTimestamp(truncateStartTS(startTimestamp));
+    public void transactionEnded(Timestamp<Long> startTimestamp){
+        Timestamp<Long> truncated = new MonotonicTimestamp(truncateForGC(startTimestamp));
         LOG.info("transaction commited={}", truncated.toPrimitive());
         this.runningTransactions.get(truncated).removeTransaction();
     }
 
     @Override
-    public void setTombstone(Timestamp<Long> commitTimestamp, LocalDateTime value){
-        LOG.info("set Tombstone={}", commitTimestamp.toPrimitive());
-        this.runningTransactions.get(commitTimestamp).setTombstone(value);
+    public void setTombstone(LocalDateTime value){
+        LOG.info("set Tombstone={}", currentCommitTs.toPrimitive());
+        this.runningTransactions.get(currentCommitTs).setTombstone(value);
     }
 
     @Override
     public Timestamp<Long> getSafeToDeleteTimestamp(){
         Timestamp<Long> newLowWaterMark = lowWaterMark;
         for(Map.Entry<Timestamp<Long>, RunningState> entry : runningTransactions.entrySet()){
             RunningState runningState = entry.getValue();
-            if(!runningState.isCleared())
+            if(!runningState.isCleared()) {
                 return new MonotonicTimestamp(newLowWaterMark);
+            }
             else
                 newLowWaterMark = entry.getKey();
         }
+        LOG.info("Normal GC newLowWaterMark = {}", newLowWaterMark.toPrimitive());
         return new MonotonicTimestamp(newLowWaterMark);
     }
 
@@ -107,25 +114,22 @@ public void evictStoredWriteSets(Long newLowWaterMark){
             this.runningTransactions.remove(i);
         }
         this.lowWaterMark.setPrimitive(newLowWaterMark);
+        LOG.info("GC performed newLowWaterMark = {}", this.lowWaterMark.toPrimitive());
     }
 
 
-    @Override
-    public Timestamp<Long> forceEvictStoredWriteSets(LocalDateTime eventTime, long intervalSec){
-        ArrayList<Timestamp<Long>> removed = new ArrayList<>();
-        this.runningTransactions.forEach((k,v) -> {
-            LocalDateTime tombstone = v.getTombstone();
+    public Timestamp<Long> getForceDeleteTimestamp(LocalDateTime eventTime, long intervalSec){
+        Timestamp<Long> newLowWaterMark = lowWaterMark;
+        for(Map.Entry<Timestamp<Long>, RunningState> entry : runningTransactions.entrySet()){
+            LocalDateTime tombstone = entry.getValue().getTombstone();
             if(tombstone != null){
                 long interval = Duration.between(tombstone, eventTime).getSeconds();
                 if(interval > intervalSec)
-                    removed.add(k);
+                    newLowWaterMark = entry.getKey();
             }
-        });
-        removed.forEach(v -> {
-            this.runningTransactions.remove(v);
-            this.history.remove(v);
-        });
-        return removed.get(removed.size() - 1);
+        }
+        LOG.info("Force GC newLowWaterMark = {}", newLowWaterMark.toPrimitive());
+        return newLowWaterMark;
     }
 
     public Timestamp<Long> getCurrentCommitTs() {

src/main/java/certifier/Certifier.java

@@ -14,8 +14,8 @@ public interface Certifier<V> extends Serializable {
     Timestamp<V> getCurrentCommitTs();
     Timestamp<V> getSafeToDeleteTimestamp();
     void evictStoredWriteSets(V newLowWaterMark);
-    Timestamp<Long> forceEvictStoredWriteSets(LocalDateTime eventTime, long intervalSec);
-    void setTombstone(Timestamp<Long> commitTimestamp, LocalDateTime value);
-    void transactionCommited(Timestamp<Long> startTimestamp);
+    Timestamp<Long> getForceDeleteTimestamp(LocalDateTime eventTime, long intervalSec);
+    void setTombstone(LocalDateTime value);
+    void transactionEnded(Timestamp<Long> startTimestamp);
     void transactionStarted(Timestamp<Long> startTimestamp);
 }

src/main/java/certifier/CertifierImpl.java

@@ -5,6 +5,7 @@
 import transaction_manager.utils.BitWriteSet;
 
 import java.io.Serializable;
+import java.time.LocalDateTime;
 import java.util.concurrent.CompletableFuture;
 
 public class CertifierImpl extends AbstractCertifier {
@@ -31,6 +32,11 @@ public long truncateStartTS(Timestamp<Long> startTimestamp) {
         return startTimestamp.toPrimitive();
     }
 
+    @Override
+    public long truncateForGC(Timestamp<Long> startTimestamp) {
+        return startTimestamp.toPrimitive();
+    }
+
     @Override
     public Timestamp<Long> treatCommit(BitWriteSet newBws, Timestamp<Long> ts){
         if (isWritable(newBws, ts, currentCommitTs)) {

src/main/java/certifier/IntervalCertifierImpl.java

@@ -3,11 +3,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import transaction_manager.utils.BitWriteSet;
-
-import java.io.Serializable;
 import java.util.LinkedList;
-
-import java.util.Queue;
 import java.util.concurrent.CompletableFuture;
 
 
@@ -18,6 +14,7 @@ public class IntervalCertifierImpl extends AbstractCertifier {
     private final Timestamp<Long> provisionalCommitTs;
     private final LinkedList<CompletableFuture<Void>> startsOnWait;
 
+
     public IntervalCertifierImpl(long timestep) {
         super(timestep);
         currentStartTs = new MonotonicTimestamp(0);
@@ -49,7 +46,12 @@ public CompletableFuture<Timestamp<Long>> start() {
 
     @Override
     public long truncateStartTS(Timestamp<Long> startTimestamp) {
-        return startTimestamp.toPrimitive() / timestep * timestep + timestep;
+        return truncateForGC(startTimestamp) + timestep;
+    }
+
+    @Override
+    public long truncateForGC(Timestamp<Long> startTimestamp) {
+        return startTimestamp.toPrimitive() / timestep * timestep;
     }
 
     @Override

src/main/java/npvs/AbstractNPVS.java

@@ -3,6 +3,7 @@
 import certifier.MonotonicTimestamp;
 import certifier.Timestamp;
 import npvs.messaging.FlushMessage;
+import npvs.messaging.NPVSReply;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import transaction_manager.utils.ByteArrayWrapper;
@@ -15,17 +16,26 @@ public abstract class AbstractNPVS implements NPVS<Long>{
     private static final Logger LOG = LoggerFactory.getLogger(AbstractNPVS.class);
     private final HashSet<Timestamp<Long>> requestsMade;
     private final Timestamp<Long> currentCommitTs;
+    private Timestamp<Long> rebootTs;
 
     public AbstractNPVS(){
         this.requestsMade = new HashSet<>();
         this.currentCommitTs = new MonotonicTimestamp(0);
+        this.rebootTs = new MonotonicTimestamp(-1);
     }
 
-    public abstract void evictVersions(Timestamp<Long> lowWaterMark);
+    public abstract void evict(Timestamp<Long> lowWaterMark);
 
     public abstract CompletableFuture<Void> putImpl(Map<ByteArrayWrapper, byte[]> writeMap, Timestamp<Long> ts);
 
-    public abstract CompletableFuture<NPVSReply> get(ByteArrayWrapper key, Timestamp<Long> ts);
+    public abstract CompletableFuture<NPVSReply> getImpl(ByteArrayWrapper key, Timestamp<Long> ts);
+
+    public CompletableFuture<NPVSReply> get(ByteArrayWrapper key, Timestamp<Long> ts){
+        if (ts.isBefore(rebootTs))
+            return CompletableFuture.completedFuture(NPVSReply.FAIL());
+        else
+            return getImpl(key, ts);
+    }
 
     @Override
     public CompletableFuture<Void> put(FlushMessage flushMessage) {
@@ -41,4 +51,8 @@ public CompletableFuture<Void> put(FlushMessage flushMessage) {
         }
         return putImpl(flushMessage.getWriteMap(), flushMessage.getCurrentTimestamp());
     }
+
+    public void setRebootTs(Timestamp<Long> rebootTs) {
+        this.rebootTs = rebootTs;
+    }
 }

src/main/java/npvs/NPVS.java

@@ -2,6 +2,7 @@
 
 import certifier.Timestamp;
 import npvs.messaging.FlushMessage;
+import npvs.messaging.NPVSReply;
 import transaction_manager.utils.ByteArrayWrapper;
 
 import java.util.concurrent.CompletableFuture;
@@ -10,4 +11,5 @@ public interface NPVS<V> {
 
     CompletableFuture<Void> put(FlushMessage flushMessage);
     CompletableFuture<NPVSReply> get(ByteArrayWrapper key, Timestamp<V> ts);
+    void evict(Timestamp<Long> lowWaterMark);
 }

src/main/java/npvs/RaftMessagingService.java renamed to src/main/java/npvs/NPVSRaftClient.java

@@ -14,15 +14,16 @@
 import java.util.Random;
 import java.util.concurrent.TimeoutException;
 
-public class RaftMessagingService {
+import static transaction_manager.raft.sofa_jraft.rpc.RaftMessagingService.getResponseFromFollower;
+import static transaction_manager.raft.sofa_jraft.rpc.RaftMessagingService.refreshLeader;
+
+public class NPVSRaftClient {
 
     private final CliClientServiceImpl cliClientService;
-    private final String groupId;
     private PeerId leader;
 
 
-    public RaftMessagingService(String groupId, String confStr){
-        this.groupId = groupId;
+    public NPVSRaftClient(String groupId, String confStr){
         Configuration conf = new Configuration();
         if (!conf.parse(confStr)) {
             throw new IllegalArgumentException("Fail to parse conf:" + confStr);
@@ -31,33 +32,16 @@ public RaftMessagingService(String groupId, String confStr){
         cliClientService = new CliClientServiceImpl();
         cliClientService.init(new CliOptions());
         try {
-            refreshLeader();
+            leader = refreshLeader(cliClientService, groupId);
         } catch (TimeoutException | InterruptedException e) {
             e.printStackTrace();
         }
     }
 
-    // raft servers should be ready first
-    public void refreshLeader() throws TimeoutException, InterruptedException {
-        if (!RouteTable.getInstance().refreshLeader(cliClientService, groupId, 1000).isOk()) {
-            int waitMs = 2000;
-            Thread.sleep(waitMs);
-            System.out.println("Waiting for raft, retrying in " + waitMs/1000 + " seconds");
-            refreshLeader();
-        }
-        leader = RouteTable.getInstance().selectLeader(groupId);
-        System.out.println("Leader is " + leader);
-    }
-
     @SuppressWarnings("unchecked")
     public Timestamp<Long> getTimestamp() {
         try {
-            List<PeerId> pps = RouteTable.getInstance().getConfiguration("manager").getPeers();
-            pps.remove(leader);
-            int randomIndex = new Random().nextInt(pps.size());
-            PeerId pp = pps.get(randomIndex);
-
-            return ((ValueResponse<Timestamp<Long>>) cliClientService.getRpcClient().invokeSync(pp.getEndpoint(), new GetTimestamp(), 30000)).getValue();
+            return getResponseFromFollower(cliClientService, new GetTimestamp(), leader);
         } catch (InterruptedException | RemotingException e) {
             e.printStackTrace();
         }