split out Database.Queue from Database.Handle

Fsck can use the queue for efficiency since it is write-heavy, and only
reads a value before writing it. But, the queue is not suited to the Keys
database.

1 files changed, 104 insertions, 0 deletions

diff --git a/Database/Queue.hs b/Database/Queue.hs
new file mode 100644
index 000000000..149854757
--- /dev/null
+++ b/Database/Queue.hs
@@ -0,0 +1,104 @@
+{- Persistent sqlite database queues
+ -
+ - Copyright 2015 Joey Hess <id@joeyh.name>
+ -
+ - Licensed under the GNU GPL version 3 or higher.
+ -}
+
+{-# LANGUAGE BangPatterns #-}
+
+module Database.Queue (
+	DbQueue,
+	initDb,
+	openDbQueue,
+	queryDbQueue,
+	closeDbQueue,
+	QueueSize,
+	queueDb,
+) where
+
+import Utility.Monad
+import Database.Handle
+
+import Database.Persist.Sqlite
+import Control.Monad
+import Control.Concurrent
+import Data.Time.Clock
+
+{- A DbQueue wraps a DbHandle, adding a queue of writes to perform.
+ -
+ - This is efficient when there are frequent writes, but
+ - reads will not immediately have access to queued writes. -}
+data DbQueue = DQ DbHandle (MVar Queue)
+
+{- Opens the database queue, but does not perform any migrations. Only use
+ - if the database is known to exist and have the right tables; ie after
+ - running initDb. -}
+openDbQueue :: FilePath -> TableName -> IO DbQueue
+openDbQueue db tablename = DQ
+	<$> openDb db tablename
+	<*> (newMVar =<< emptyQueue)
+
+{- Must be called to ensure queued changes get written to the database. -}
+closeDbQueue :: DbQueue -> IO ()
+closeDbQueue h@(DQ hdl _) = do
+	flushDbQueue h
+	closeDb hdl
+
+{- Makes a queury using the DbQueue. This should not be used to make
+ - changes to the database!
+ -
+ - Queries will not return changes that have been recently queued,
+ - so use with care.
+ -}
+queryDbQueue :: DbQueue -> SqlPersistM a -> IO a
+queryDbQueue (DQ hdl _) = queryDb hdl
+
+{- A queue of actions to perform, with a count of the number of actions
+ - queued, and a last commit time. -}
+data Queue = Queue QueueSize LastCommitTime (SqlPersistM ())
+
+type QueueSize = Int
+
+type LastCommitTime = UTCTime
+
+emptyQueue :: IO Queue
+emptyQueue = do
+	now <- getCurrentTime
+	return $ Queue 0 now (return ())
+
+flushDbQueue :: DbQueue -> IO ()
+flushDbQueue (DQ hdl qvar) = do
+	Queue sz _ qa <- takeMVar qvar
+	when (sz > 0) $
+		commitDb hdl qa
+
+{- Queues a change to be made to the database. It will be queued
+ - to be committed later, unless the commitchecker action returns true,
+ - in which case any previously queued changes are also committed.
+ -
+ - Transactions built up by queueDb are sent to sqlite all at once.
+ - If sqlite fails due to another change being made concurrently by another
+ - process, the transaction is put back in the queue. This avoids
+ - the sqlite multiple writer problem.
+ -}
+queueDb 
+	:: DbQueue
+	-> (QueueSize -> LastCommitTime -> IO Bool) 
+	-> SqlPersistM ()
+	-> IO ()
+queueDb (DQ hdl qvar) commitchecker a = do
+	Queue sz lastcommittime qa <- takeMVar qvar
+	let !sz' = sz + 1
+	let qa' = qa >> a
+	let enqueue = putMVar qvar
+	ifM (commitchecker sz' lastcommittime)
+		( do
+			r <- commitDb' hdl qa'
+			case r of
+				Left _ -> enqueue $ Queue sz' lastcommittime qa'
+				Right _ -> do
+					now <- getCurrentTime
+					enqueue $ Queue 0 now (return ())
+		, enqueue $ Queue sz' lastcommittime qa'
+		)