diff --git a/src/test/kotlin/kestrel/store/txn/LockManagerTest.kt b/src/test/kotlin/kestrel/store/txn/LockManagerTest.kt new file mode 100644 index 0000000..6cca4f1 --- /dev/null +++ b/src/test/kotlin/kestrel/store/txn/LockManagerTest.kt @@ -0,0 +1,117 @@ +package kestrel.store.txn + +import java.util.concurrent.TimeUnit +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith +import kotlin.test.assertTrue + +class LockManagerTest { + private fun waitUntil(timeoutMs: Long = 2000, predicate: () -> Boolean) { + val deadline = System.currentTimeMillis() + timeoutMs + while (!predicate()) { + if (System.currentTimeMillis() > deadline) throw AssertionError("condition never became true") + Thread.sleep(5) + } + } + + @Test + fun sharedLocksAreCompatibleAcrossTransactions() { + val locks = LockManager() + locks.acquireShared(1L, 100L) + locks.acquireShared(2L, 100L) + assertEquals(mapOf(1L to LockManager.LockMode.SHARED, 2L to LockManager.LockMode.SHARED), locks.holders(100L)) + } + + @Test + fun reacquiringTheSameOrWeakerModeIsANoop() { + val locks = LockManager() + locks.acquireExclusive(1L, 100L) + locks.acquireShared(1L, 100L) + assertEquals(mapOf(1L to LockManager.LockMode.EXCLUSIVE), locks.holders(100L)) + } + + @Test + fun upgradeFromSharedToExclusiveSucceedsWhenSoleHolder() { + val locks = LockManager() + locks.acquireShared(1L, 100L) + locks.acquireExclusive(1L, 100L) + assertEquals(mapOf(1L to LockManager.LockMode.EXCLUSIVE), locks.holders(100L)) + } + + @Test + fun exclusiveLockBlocksAnotherTransactionUntilReleased() { + val locks = LockManager() + locks.acquireExclusive(1L, 100L) + + val acquired = java.util.concurrent.atomic.AtomicBoolean(false) + val thread = Thread { + locks.acquireExclusive(2L, 100L) + acquired.set(true) + } + thread.start() + waitUntil { locks.waiters(100L).contains(2L) } + assertTrue(!acquired.get(), "txn 2 must not acquire while txn 1 holds the exclusive lock") + + locks.release(1L, 100L) + thread.join(2000) + assertTrue(acquired.get()) + assertEquals(mapOf(2L to LockManager.LockMode.EXCLUSIVE), locks.holders(100L)) + } + + @Test + fun releaseAllDropsEveryLockHeldByATransaction() { + val locks = LockManager() + locks.acquireExclusive(1L, 100L) + locks.acquireShared(1L, 200L) + locks.releaseAll(1L) + assertEquals(emptyMap(), locks.holders(100L)) + assertEquals(emptyMap(), locks.holders(200L)) + } + + @Test + fun circularWaitIsDetectedAndRejectedWithoutBlocking() { + val locks = LockManager() + locks.acquireExclusive(1L, 100L) // txn 1 holds resource A + locks.acquireExclusive(2L, 200L) // txn 2 holds resource B + + val blocked = java.util.concurrent.atomic.AtomicBoolean(false) + val thread = Thread { + locks.acquireExclusive(1L, 200L) // txn 1 wants B, held by txn 2: blocks + blocked.set(true) + } + thread.start() + waitUntil { locks.waiters(200L).contains(1L) } + + // txn 2 now wants A, held by txn 1, which is already waiting on txn 2: a cycle. + assertFailsWith { locks.acquireExclusive(2L, 100L) } + + // Unblock the background waiter so the test does not leak a thread. + locks.release(2L, 200L) + thread.join(2000) + assertTrue(blocked.get()) + } + + @Test + fun unrelatedWaitersDoNotFalselyTriggerDeadlock() { + val locks = LockManager() + locks.acquireExclusive(1L, 100L) + // txn 2 requests an unrelated resource txn 1 does not hold: no cycle, should not throw. + locks.acquireExclusive(2L, 200L) + } + + @Test + fun exclusiveWaiterEventuallyRunsToCompletionAfterHolderReleases() { + val locks = LockManager() + locks.acquireExclusive(1L, 100L) + val latch = java.util.concurrent.CountDownLatch(1) + val thread = Thread { + locks.acquireExclusive(2L, 100L) + latch.countDown() + } + thread.start() + waitUntil { locks.waiters(100L).contains(2L) } + locks.releaseAll(1L) + assertTrue(latch.await(2, TimeUnit.SECONDS)) + } +} diff --git a/src/test/kotlin/kestrel/store/txn/TransactionManagerTest.kt b/src/test/kotlin/kestrel/store/txn/TransactionManagerTest.kt new file mode 100644 index 0000000..bea82d2 --- /dev/null +++ b/src/test/kotlin/kestrel/store/txn/TransactionManagerTest.kt @@ -0,0 +1,88 @@ +package kestrel.store.txn + +import kestrel.store.page.DiskManager +import kestrel.store.page.Page +import java.io.File +import java.nio.file.Files +import kestrel.store.wal.WriteAheadLog +import kotlin.test.Test +import kotlin.test.assertContentEquals +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith + +class TransactionManagerTest { + private fun tempFile(prefix: String): File = Files.createTempFile(prefix, ".tmp").toFile().also { it.deleteOnExit() } + + private fun pageOf(fill: Int): ByteArray = ByteArray(Page.PAGE_SIZE) { fill.toByte() } + + private class Fixture { + val diskManager = DiskManager(Files.createTempFile("kestrel-txnmgr-data", ".db").toFile().also { it.deleteOnExit() }) + val wal = WriteAheadLog(Files.createTempFile("kestrel-txnmgr-wal", ".log").toFile().also { it.deleteOnExit() }) + val manager = TransactionManager(wal, diskManager) + } + + @Test + fun beginStartsATransactionInActiveStatus() { + val fixture = Fixture() + val txn = fixture.manager.begin() + assertEquals(TransactionManager.Status.ACTIVE, fixture.manager.statusOf(txn)) + } + + @Test + fun commitAppliesLoggedUpdateAndReleasesLocks() { + val fixture = Fixture() + val pageId = fixture.diskManager.allocatePage() + val before = pageOf(0) + fixture.diskManager.writePage(Page(pageId).also { System.arraycopy(before, 0, it.data, 0, before.size) }) + + val txn = fixture.manager.begin() + fixture.manager.acquireExclusive(txn, pageId) + val after = pageOf(3) + fixture.manager.logUpdate(txn, pageId, before, after) + fixture.diskManager.writePage(Page(pageId).also { System.arraycopy(after, 0, it.data, 0, after.size) }) + fixture.manager.commit(txn) + + assertEquals(TransactionManager.Status.COMMITTED, fixture.manager.statusOf(txn)) + assertContentEquals(after, fixture.diskManager.readPage(pageId).data) + + val otherTxn = fixture.manager.begin() + fixture.manager.acquireExclusive(otherTxn, pageId) // must not block: txn's locks were released on commit + } + + @Test + fun abortRollsBackLoggedUpdatesAndReleasesLocks() { + val fixture = Fixture() + val pageId = fixture.diskManager.allocatePage() + val before = pageOf(1) + fixture.diskManager.writePage(Page(pageId).also { System.arraycopy(before, 0, it.data, 0, before.size) }) + + val txn = fixture.manager.begin() + fixture.manager.acquireExclusive(txn, pageId) + val after = pageOf(4) + fixture.manager.logUpdate(txn, pageId, before, after) + fixture.diskManager.writePage(Page(pageId).also { System.arraycopy(after, 0, it.data, 0, after.size) }) + fixture.manager.abort(txn) + + assertEquals(TransactionManager.Status.ABORTED, fixture.manager.statusOf(txn)) + assertContentEquals(before, fixture.diskManager.readPage(pageId).data) + + val otherTxn = fixture.manager.begin() + fixture.manager.acquireExclusive(otherTxn, pageId) // must not block: aborted txn released its locks + } + + @Test + fun operationsOnAResolvedTransactionFail() { + val fixture = Fixture() + val txn = fixture.manager.begin() + fixture.manager.commit(txn) + assertFailsWith { fixture.manager.commit(txn) } + assertFailsWith { fixture.manager.abort(txn) } + assertFailsWith { fixture.manager.acquireExclusive(txn, 1L) } + } + + @Test + fun statusOfUnknownTransactionFails() { + val fixture = Fixture() + assertFailsWith { fixture.manager.statusOf(999L) } + } +}