DiskLogger extends LifeCycleThread { public LSN switchLogFile(boolean incrementViewId) throws IOException { if (!hasLock()) { throw new IllegalStateException("the lock is held by another thread or the logger is not locked."); } LSN lastSyncedLSN = null; if (incrementViewId) { int view = currentViewId.getAndIncrement(); long seq = nextLogSequenceNo.getAndSet(1L) - 1L; assert (seq != 0) : "Checkpoint after checkpoint is not allowed!"; lastSyncedLSN = new LSN(view, seq); } else { lastSyncedLSN = new LSN(currentViewId.get(), nextLogSequenceNo.get() - 1L); } dropLogFile(); loadLogFile(); return lastSyncedLSN; } DiskLogger(String logfileDir, LSN initLSN, SyncMode syncMode, int pseudoSyncWait, int maxQ); void dropLogFile(); void loadLogFile(LSN initLSN); long getLogFileSize(); synchronized void append(LogEntry entry); void lock(); boolean hasLock(); void unlock(); LSN switchLogFile(boolean incrementViewId); @Override synchronized void start(); void run(); void shutdown(); synchronized void shutdown(boolean graceful); @Deprecated void destroy(); LSN getLatestLSN(); Object getRuntimeState(String property); Map<String, Object> getRuntimeState(); static final int MAX_ENTRIES_PER_BLOCK; }
@Test public void testSwitchLogFile() throws Exception { final AtomicInteger count = new AtomicInteger(0); SyncListener sl = new SyncListener() { public void synced(LSN lsn) { synchronized (count) { count.incrementAndGet(); count.notifyAll(); } } public void failed(Exception ex) { synchronized (count) { count.incrementAndGet(); count.notifyAll(); } } }; for (int i = 0; i < 100; i++) { String pl = "Entry " + (i + 1); ReusableBuffer plb = ReusableBuffer.wrap(pl.getBytes()); l.append(new LogEntry(plb, sl, LogEntry.PAYLOAD_TYPE_INSERT)); } synchronized (count) { if (count.get() < 100) count.wait(1000); } try { l.lock(); l.switchLogFile(false); } finally { l.unlock(); } for (int i = 0; i < 100; i++) { String pl = "Entry " + (i + 100 + 1); ReusableBuffer plb = ReusableBuffer.wrap(pl.getBytes()); l.append(new LogEntry(plb, sl, LogEntry.PAYLOAD_TYPE_INSERT)); } synchronized (count) { if (count.get() < 200) count.wait(1000); } }