@@ -20,7 +20,7 @@ module ProjectM36.Client
2020 typeForRelationalExpr,
2121 inclusionDependencies,
2222 planForDatabaseContextExpr,
23− processPersistence,
23+ processTransactionGraphPersistence,
2424 currentSchemaName,
2525 SchemaName,
2626 HeadName,
@@ -101,6 +101,7 @@ import ProjectM36.TransactionGraph.Persist
101101import ProjectM36.Attribute hiding (atomTypes)
102102import ProjectM36.TransGraphRelationalExpression (TransGraphRelationalExpr, evalTransGraphRelationalExpr)
103103import ProjectM36.Persist (DiskSync(..))
104+import ProjectM36.FileLock
104105import ProjectM36.Notifications
105106import ProjectM36.Server.RemoteCallTypes
106107import qualified ProjectM36.DisconnectedTransaction as Discon
@@ -108,6 +109,8 @@ import ProjectM36.Relation (typesAsRelation)
108109import ProjectM36.ScriptSession (initScriptSession, ScriptSession)
109110import qualified ProjectM36.Relation as R
110111import qualified ProjectM36.DatabaseContext as DBC
112+import Control.Exception.Base
113+import GHC.Conc.Sync
111114
112115import Network.Transport (Transport(closeTransport))
113116import Network.Transport.TCP (createTransport, defaultTCPParameters, encodeEndPointAddress)
@@ -117,7 +120,6 @@ import Control.Distributed.Process.ManagedProcess.Client (call, safeCall)
117120import Control.Distributed.Process (NodeId(..), reconnect)
118121
119122import Data.UUID.V4 (nextRandom)
120−import Control.Concurrent.STM
121123import Data.Word
122124import Control.Distributed.Process (ProcessId, Process, receiveWait, send, match)
123125import Control.Exception (IOException, handle, AsyncException, throwIO, fromException, Exception)
@@ -134,6 +136,7 @@ import Data.Binary (Binary)
134136import GHC.Generics (Generic)
135137import Control.DeepSeq (force)
136138import System.IO
139+import Crypto.Hash
137140
138141type Hostname = String
139142
@@ -195,6 +198,9 @@ defaultRemoteConnectionInfo = RemoteProcessConnectionInfo defaultDatabaseName (c
195198-- | The 'Connection' represents either local or remote access to a database. All operations flow through the connection.
196199type ClientNodes = STMSet.Set ProcessId
197200
201+type TransactionGraphLockHandle = Handle
202+type LockFileHash = Digest SHA256
203+
198204-- internal structure specific to in-process connections
199205data InProcessConnectionConf = InProcessConnectionConf {
200206 ipPersistenceStrategy :: PersistenceStrategy,
@@ -203,7 +209,8 @@ data InProcessConnectionConf = InProcessConnectionConf {
203209 ipTransactionGraph :: TVar TransactionGraph,
204210 ipScriptSession :: Maybe ScriptSession,
205211 ipLocalNode :: LocalNode,
206− ipTransport :: Transport -- we hold onto this so that we can close it gracefully
212+ ipTransport :: Transport, -- we hold onto this so that we can close it gracefully
213+ ipLocks :: Maybe (TransactionGraphLockHandle, MVar LockFileHash) -- nothing when NoPersistence
207214 }
208215
209216data RemoteProcessConnectionConf = RemoteProcessConnectionConf {
@@ -275,14 +282,16 @@ connectProjectM36 (InProcessConnectionInfo strat notificationCallback ghcPkgPath
275282 (localNode, transport) <- createLocalNode
276283 notificationPid <- startNotificationListener localNode notificationCallback
277284 mScriptSession <- createScriptSession ghcPkgPaths
285+
278286 let conn = InProcessConnection (InProcessConnectionConf {
279− ipPersistenceStrategy =strat,
287+ ipPersistenceStrategy = strat,
280288 ipClientNodes = clientNodes,
281289 ipSessions = sessions,
282290 ipTransactionGraph = graphTvar,
283291 ipScriptSession = mScriptSession,
284292 ipLocalNode = localNode,
285− ipTransport = transport })
293+ ipTransport = transport,
294+ ipLocks = Nothing})
286295 addClientNode conn notificationPid
287296 pure (Right conn)
288297 MinimalPersistence dbdir -> connectPersistentProjectM36 strat NoDiskSync dbdir freshGraph notificationCallback ghcPkgPaths
@@ -318,8 +327,8 @@ connectPersistentProjectM36 :: PersistenceStrategy ->
318327connectPersistentProjectM36 strat sync dbdir freshGraph notificationCallback ghcPkgPaths = do
319328 err <- setupDatabaseDir sync dbdir freshGraph
320329 case err of
321− Just err' -> return $ Left (SetupDatabaseDirectoryError err')
322− Nothing -> do
330+ Left err' -> return $ Left (SetupDatabaseDirectoryError err')
331+ Right (lockFileH, digest) -> do
323332 mScriptSession <- createScriptSession ghcPkgPaths
324333 graph <- transactionGraphLoad dbdir emptyTransactionGraph mScriptSession
325334 case graph of
@@ -329,6 +338,7 @@ connectPersistentProjectM36 strat sync dbdir freshGraph notificationCallback ghc
329338 sessions <- STMMap.newIO
330339 clientNodes <- STMSet.newIO
331340 (localNode, transport) <- createLocalNode
341+ lockMVar <- newMVar digest
332342 let conn = InProcessConnection (InProcessConnectionConf {
333343 ipPersistenceStrategy = strat,
334344 ipClientNodes = clientNodes,
@@ -336,7 +346,8 @@ connectPersistentProjectM36 strat sync dbdir freshGraph notificationCallback ghc
336346 ipTransactionGraph = tvarGraph,
337347 ipScriptSession = mScriptSession,
338348 ipLocalNode = localNode,
339− ipTransport = transport
349+ ipTransport = transport,
350+ ipLocks = Just (lockFileH, lockMVar)
340351 })
341352
342353 notificationPid <- startNotificationListener localNode notificationCallback
@@ -578,6 +589,7 @@ executeGraphExprSTM_ freshId sessionId sessions graphExpr graphTvar = do
578589 Right session -> case evalGraphOp freshId (Sess.disconnectedTransaction session) graph graphExpr of
579590 Left err -> pure $ Left err
580591 Right (discon', graph') -> do
592+
581593 writeTVar graphTvar graph'
582594 let newSession = Session discon' (Sess.schemaName session)
583595 STMMap.insert newSession sessionId sessions
@@ -591,53 +603,98 @@ executeCommitExprSTM_ oldContext newContext nodes = do
591603 evaldNots = M.map mkEvaldNot fireNots
592604 mkEvaldNot notif = EvaluatedNotification { notification = notif, reportRelation = evalState (RE.evalRelationalExpr (reportExpr notif)) (RE.mkRelationalExprState oldContext) }
593605 pure (evaldNots, nodes)
594−
595−
606+
596607-- | Execute a transaction graph expression in the context of the session and connection. Transaction graph operators modify the transaction graph state.
608+-- OPTIMIZATION OPPORTUNITY: no locks are required to write new transaction data, only to update the transaction graph id file
609+-- if writing data is re-entrant, we may be able to use unsafeIOtoSTM
610+-- perhaps keep hash of data file instead of checking if our head was updated on every write
597611executeGraphExpr :: SessionId -> Connection -> TransactionGraphOperator -> IO (Maybe RelationalError)
598612executeGraphExpr sessionId (InProcessConnection conf) graphExpr = excMaybe $ do
599613 let strat = ipPersistenceStrategy conf
600614 clientNodes = ipClientNodes conf
601615 sessions = ipSessions conf
602616 graphTvar = ipTransactionGraph conf
617+ mLockFileH = ipLocks conf
618+ lockHandler body = case graphExpr of
619+ Commit -> case mLockFileH of
620+ Nothing -> pure Nothing
621+ Just (lockFileH, lockMVar) -> let acquireLocks = do
622+ lastWrittenDigest <- takeMVar lockMVar
623+ lockFile lockFileH WriteLock
624+ latestDigest <- readGraphTransactionIdDigest strat
625+ pure (latestDigest /= lastWrittenDigest)
626+
627+ releaseLocks _ = do
628+ --still holding the lock- get the latest digest
629+ gDigest <- readGraphTransactionIdDigest strat
630+ unlockFile lockFileH
631+ putMVar lockMVar gDigest
632+ in bracket acquireLocks releaseLocks body
633+ _ -> body False
603634 freshId <- nextRandom
604− manip <- atomically $ do
635+ lockHandler $ \dbWrittenByOtherProcess -> do
636+ --if the database file has been updated since we wrote it last, load it before trying to sync our version done- this can result in TransactionNotAHeadErrors
637+ --read transaction data and compare to existing graph
638+ --in the future, we can detect if updated transaction graph can be safely merged (such as with a transaction on a separate head) (rebase-able commits should force the user to rebase from the client to confirm that the action makes sense)
639+ if dbWrittenByOtherProcess then do
640+ handleGraphFromDiskUpdate sessionId conf
641+ else do
642+ manip <- atomically $ do
643+ eSession <- sessionForSessionId sessionId sessions
644+ --handle graph update by other process
645+ oldGraph <- readTVar graphTvar
646+ case eSession of
647+ Left err -> pure (Left err)
648+ Right session -> do
649+ eGraph <- executeGraphExprSTM_ freshId sessionId sessions graphExpr graphTvar
650+ case eGraph of
651+ Left err -> pure (Left err)
652+ Right newGraph -> do
653+ --handle commmit
654+ if graphExpr == Commit then
655+ case transactionForId (Sess.parentId session) oldGraph of
656+ Left err -> pure $ Left err
657+ Right previousTrans -> do
658+ (evaldNots, nodes) <- executeCommitExprSTM_ (Trans.concreteDatabaseContext previousTrans) (Sess.concreteDatabaseContext session) clientNodes
659+ nodesToNotify <- toList (STMSet.stream nodes)
660+ pure $ Right (evaldNots, nodesToNotify, newGraph)
661+ else
662+ pure $ Right (M.empty, [], newGraph)
663+ case manip of
664+ Left err -> return $ Just err
665+ Right (notsToFire, nodesToNotify, newGraph) -> do
666+ --update filesystem database, if necessary
667+ processTransactionGraphPersistence strat newGraph
668+ sendNotifications nodesToNotify (ipLocalNode conf) notsToFire
669+ pure Nothing
670+executeGraphExpr sessionId conn@(RemoteProcessConnection _) graphExpr = remoteCall conn (ExecuteGraphExpr sessionId graphExpr)
671+
672+-- | Called when the transaction graph on disk has been updated by a secondary process. Reject the commit and update the graph
673+handleGraphFromDiskUpdate :: SessionId -> InProcessConnectionConf -> IO (Maybe RelationalError)
674+handleGraphFromDiskUpdate sessionId conf = do
675+ let graphTVar = ipTransactionGraph conf
676+ strat = ipPersistenceStrategy conf
677+ sessions = ipSessions conf
678+ mScriptSession = ipScriptSession conf
679+ dbdir = case strat of
680+ MinimalPersistence x -> x
681+ CrashSafePersistence x -> x
682+ _ -> error "handleGraphFromDiskUpdate on unpersisted database"
683+ atomically $ do
605684 eSession <- sessionForSessionId sessionId sessions
606− oldGraph <- readTVar graphTvar
607685 case eSession of
608− Left err -> pure (Left err)
686+ Left err -> pure (Just err)
609687 Right session -> do
610− eGraph <- executeGraphExprSTM_ freshId sessionId sessions graphExpr graphTvar
688+ graph <- readTVar graphTVar
689+ let parentTid = Discon.parentId (disconnectedTransaction session)
690+ --this should be re-entrant because it only reads from the disk to create immutable structures
691+ --BUG: here, we unsafely assume that the updated graph is based on the original- that might not be true- we should check that the graph is sound and consistence with the previous graph
692+ eGraph <- unsafeIOToSTM (transactionGraphLoad dbdir graph mScriptSession)
611693 case eGraph of
612− Left err -> pure (Left err)
613− Right newGraph -> do
614− if graphExpr == Commit then
615− case transactionForId (Sess.parentId session) oldGraph of
616− Left err -> pure $ Left err
617− Right previousTrans -> do
618− (evaldNots, nodes) <- executeCommitExprSTM_ (Trans.concreteDatabaseContext previousTrans) (Sess.concreteDatabaseContext session) clientNodes
619− nodesToNotify <- toList (STMSet.stream nodes)
620− pure $ Right (evaldNots, nodesToNotify, newGraph)
621− else
622− pure $ Right (M.empty, [], newGraph)
623− case manip of
624− Left err -> return $ Just err
625− Right (notsToFire, nodesToNotify, newGraph) -> do
626− --update filesystem database, if necessary
627− --this should really grab a lock at the beginning of the method to be threadsafe
628− processPersistence strat newGraph
629− sendNotifications nodesToNotify (ipLocalNode conf) notsToFire
630− return Nothing
631−executeGraphExpr sessionId conn@(RemoteProcessConnection _) graphExpr = remoteCall conn (ExecuteGraphExpr sessionId graphExpr)
632−
633−{-
634−commitSTM_ :: UUID -> SessionId -> Sessions -> TVar TransactionGraph -> STM (Maybe RelationalError)
635−commitSTM_ freshUUID sessionId sessions graph = do
636− mErr <- executeGraphExprSTM_ freshUUID sessionId sessions Commit graph
637− pure $ case mErr of
638− Left err -> Just err
639− Right _ -> Nothing
640−-}
694+ Left err -> pure (Just (DatabaseLoadError err))
695+ Right updatedGraph -> do
696+ writeTVar graphTVar updatedGraph
697+ pure (Just (TransactionIsNotAHeadError parentTid))
641698
642699-- | A trans-graph expression is a relational query executed against the entirety of a transaction graph.
643700executeTransGraphRelationalExpr :: SessionId -> Connection -> TransGraphRelationalExpr -> IO (Either RelationalError Relation)
@@ -686,10 +743,27 @@ rollback :: SessionId -> Connection -> IO (Maybe RelationalError)
686743rollback sessionId conn@(InProcessConnection _) = executeGraphExpr sessionId conn Rollback
687744rollback sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteGraphExpr sessionId Rollback)
688745
689−processPersistence :: PersistenceStrategy -> TransactionGraph -> IO ()
690−processPersistence NoPersistence _ = return ()
691−processPersistence (MinimalPersistence dbdir) graph = transactionGraphPersist NoDiskSync dbdir graph
692−processPersistence (CrashSafePersistence dbdir) graph = transactionGraphPersist FsyncDiskSync dbdir graph
746+processTransactionGraphPersistence :: PersistenceStrategy -> TransactionGraph -> IO ()
747+processTransactionGraphPersistence NoPersistence _ = pure ()
748+processTransactionGraphPersistence (MinimalPersistence dbdir) graph = transactionGraphPersist NoDiskSync dbdir graph >> pure ()
749+processTransactionGraphPersistence (CrashSafePersistence dbdir) graph = transactionGraphPersist FsyncDiskSync dbdir graph >> pure ()
750+
751+readGraphTransactionIdDigest :: PersistenceStrategy -> IO (Digest SHA256)
752+readGraphTransactionIdDigest NoPersistence = error "attempt to read digest from transaction log without persistence enabled"
753+readGraphTransactionIdDigest (MinimalPersistence dbdir) = readGraphTransactionIdFileDigest dbdir
754+readGraphTransactionIdDigest (CrashSafePersistence dbdir) = readGraphTransactionIdFileDigest dbdir
755+
756+{-
757+sameGraphTransactionIdDigest :: PersistenceStrategy -> Digest SHA256 -> IO (Bool)
758+sameGraphTransactionIdDigest NoPersistence _ = pure True
759+sameGraphTransactionIdDigest persist lastDigest = do
760+ let dbdir = case persist of
761+ MinimalPersistence x -> x
762+ CrashSafePersistence x -> x
763+ newDigest <- readGraphTransactionIdDigest persist
764+ pure (lastDigest == newDigest)
765+
766+-}
693767
694768-- | Return a relation whose type would match that of the relational expression if it were executed. This is useful for checking types and validating a relational expression's types.
695769typeForRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)