From a598c0b421b48bb50213c47b742fede04b5cd92d Mon Sep 17 00:00:00 2001
From: Christopher Hotchkiss <christopher.hotchkiss@gmail.com>
Date: Fri, 8 Oct 2021 11:32:58 -0400
Subject: [PATCH] Pushing in progress

---
 Cargo.toml                                |  1 +
 src/engine/io/block_layer/lock_manager.rs | 61 ++++++++++++++++++++++-
 2 files changed, 60 insertions(+), 2 deletions(-)

diff --git a/Cargo.toml b/Cargo.toml
index 9ce994c..717ff65 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -46,6 +46,7 @@ bytes = "1"
 futures = "0.3"
 hex-literal = "0.3.1"
 lru = "0.6.6"
+moka = { version = "0.6", features = ["future"] }
 nom = "7"
 log = "0.4"
 simplelog = "^0.10.0"
diff --git a/src/engine/io/block_layer/lock_manager.rs b/src/engine/io/block_layer/lock_manager.rs
index 8a879dc..46d4301 100644
--- a/src/engine/io/block_layer/lock_manager.rs
+++ b/src/engine/io/block_layer/lock_manager.rs
@@ -1,3 +1,60 @@
-pub struct LockManager;
+use std::sync::Arc;
 
-impl LockManager {}
+use super::file_manager::{FileManager, FileManagerError};
+use crate::engine::io::page_formats::{PageId, PageOffset};
+use moka::future::Cache;
+use thiserror::Error;
+use tokio::sync::RwLock;
+
+/// The LockManager is used for cooperative access to pages in the system.
+///
+/// Before accessing the I/O layer you must get a read or write lock on
+/// the page you need to access. Only AFTER you have the lock you should
+/// ask for the page.
+///
+/// TODO: Find a way that I can do this in a type enforcing way.
+
+#[derive(Clone)]
+pub struct LockManager {
+    file_manager: Arc<FileManager>,
+    locks: Cache<(PageId, PageOffset), Arc<RwLock<()>>>,
+}
+
+impl LockManager {
+    pub fn new(file_manager: Arc<FileManager>) -> LockManager {
+        LockManager {
+            file_manager,
+            locks: Cache::new(1000),
+        }
+    }
+
+    pub async fn get_offset(&self, page_id: PageId) -> Result<PageOffset, LockManagerError> {
+        Ok(self.file_manager.get_offset(&page_id).await?)
+    }
+
+    pub async fn get_offset_non_zero(
+        &self,
+        page_id: PageId,
+    ) -> Result<PageOffset, LockManagerError> {
+        let mut offset = PageOffset(0);
+        while offset == PageOffset(0) {
+            offset = self.file_manager.get_offset(&page_id).await?;
+        }
+        Ok(offset)
+    }
+
+    pub async fn get_lock(&self, page_id: PageId, offset: PageOffset) -> Arc<RwLock<()>> {
+        self.locks
+            .get_or_insert_with(
+                (page_id, offset),
+                async move { Arc::new(RwLock::const_new(())) },
+            )
+            .await
+    }
+}
+
+#[derive(Debug, Error)]
+pub enum LockManagerError {
+    #[error(transparent)]
+    FileManagerError(#[from] FileManagerError),
+}