git.iko.soy/project-m36
on 3356ff7 help sign in
project-m36/src/lib/ProjectM36/Client.hs
1031 lines 54 kB history raw
1{-# LANGUAGE DeriveAnyClass, DeriveGeneric, ScopedTypeVariables, BangPatterns #-}
2{-|
3Module: ProjectM36.Client
5Client interface to local and remote Project:M36 databases. To get started, connect with 'connectProjectM36', then run some database changes with 'executeDatabaseContextExpr', and issue queries using 'executeRelationalExpr'.
6-}
7module ProjectM36.Client
8 (ConnectionInfo(..),
9 Connection(..),
10 Port,
11 Hostname,
12 DatabaseName,
13 ConnectionError(..),
14 connectProjectM36,
15 close,
16 closeRemote_,
17 executeRelationalExpr,
18 executeDatabaseContextExpr,
19 executeDatabaseContextIOExpr,
20 executeGraphExpr,
21 executeSchemaExpr,
22 executeTransGraphRelationalExpr,
23 commit,
24 rollback,
25 typeForRelationalExpr,
26 inclusionDependencies,
27 planForDatabaseContextExpr,
28 currentSchemaName,
29 SchemaName,
30 HeadName,
31 setCurrentSchemaName,
32 transactionGraphAsRelation,
33 relationVariablesAsRelation,
34 disconnectedTransactionIsDirty,
35 headName,
36 remoteDBLookupName,
37 defaultServerPort,
38 headTransactionId,
39 defaultDatabaseName,
40 defaultRemoteConnectionInfo,
41 defaultHeadName,
42 PersistenceStrategy(..),
43 RelationalExpr,
44 RelationalExprBase(..),
45 DatabaseContextExpr(..),
46 DatabaseContextIOExpr(..),
47 Attribute(..),
48 MergeStrategy(..),
49 attributesFromList,
50 createNodeId,
51 createSessionAtCommit,
52 createSessionAtHead,
53 closeSession,
54 addClientNode,
55 callTestTimeout_,
56 RelationCardinality(..),
57 TransactionGraphOperator(..),
58 ProjectM36.Client.autoMergeToHead,
59 transactionGraph_,
60 disconnectedTransaction_,
61 TransGraphRelationalExpr,
62 TransactionIdLookup(..),
63 TransactionIdHeadBacktrack(..),
64 NodeId(..),
65 Atom(..),
66 Session,
67 SessionId,
68 NotificationCallback,
69 emptyNotificationCallback,
70 EvaluatedNotification(..),
71 atomTypesAsRelation,
72 AttributeExpr,
73 inclusionDependencyForKey,
74 databaseContextExprForUniqueKey,
75 databaseContextExprForForeignKey,
76 createScriptedAtomFunction,
77 AttributeExprBase(..),
78 TypeConstructor(..),
79 TypeConstructorDef(..),
80 DataConstructorDef(..),
81 AttributeNames(..),
82 RelVarName,
83 IncDepName,
84 InclusionDependency(..),
85 AttributeName,
86 RequestTimeoutException(..),
87 RemoteProcessDiedException(..),
88 AtomType(..),
89 Atomable(..),
90 TupleExprBase(..),
91 AtomExprBase(..),
92 RestrictionPredicateExprBase(..),
93 ) where
94import ProjectM36.Base hiding (inclusionDependencies) --defined in this module as well
95import qualified ProjectM36.Base as B
96import ProjectM36.Error
97import ProjectM36.Atomable
98import ProjectM36.AtomFunction
99import ProjectM36.StaticOptimizer
100import ProjectM36.Key
101import qualified ProjectM36.IsomorphicSchema as Schema
102import Control.Monad.State
103import Control.Monad.Trans.Reader
104import qualified ProjectM36.RelationalExpression as RE
105import ProjectM36.DatabaseContext (basicDatabaseContext)
106import qualified ProjectM36.TransactionGraph as Graph
107import ProjectM36.TransactionGraph
108import qualified ProjectM36.Transaction as Trans
109import ProjectM36.TransactionGraph.Persist
110import ProjectM36.Attribute hiding (atomTypes)
111import ProjectM36.TransGraphRelationalExpression (TransGraphRelationalExpr, evalTransGraphRelationalExpr)
112import ProjectM36.Persist (DiskSync(..))
113import ProjectM36.FileLock
114import ProjectM36.Notifications
115import ProjectM36.Server.RemoteCallTypes
116import qualified ProjectM36.DisconnectedTransaction as Discon
117import ProjectM36.Relation (typesAsRelation)
118import ProjectM36.ScriptSession (initScriptSession, ScriptSession)
119import qualified ProjectM36.Relation as R
120import qualified ProjectM36.DatabaseContext as DBC
121import Control.Exception.Base
122import GHC.Conc.Sync
124import Network.Transport (Transport(closeTransport))
125import Network.Transport.TCP (createTransport, defaultTCPParameters, encodeEndPointAddress)
126import Control.Distributed.Process.Node (newLocalNode, initRemoteTable, runProcess, LocalNode, forkProcess, closeLocalNode)
127import Control.Distributed.Process.Extras.Internal.Types (whereisRemote)
128import Control.Distributed.Process.ManagedProcess.Client (call, safeCall)
129import Data.Either (isRight)
130import Data.UUID.V4 (nextRandom)
131import Data.Word
132import Control.Distributed.Process (ProcessId, Process, receiveWait, send, match, NodeId(..), reconnect)
133import Control.Exception (IOException, handle, AsyncException, throwIO, fromException, Exception)
134import Control.Concurrent.MVar
135import qualified Data.Map as M
136import Control.Distributed.Process.Serializable (Serializable)
137import qualified STMContainers.Map as STMMap
138import qualified STMContainers.Set as STMSet
139import qualified ProjectM36.Session as Sess
140import ProjectM36.Session
141import ProjectM36.Sessions
142import ListT
143import Data.Binary (Binary)
144import GHC.Generics (Generic)
145import Control.DeepSeq (force)
146import System.IO
147import Data.Time.Clock
149type Hostname = String
151type Port = Word16
153type DatabaseName = String
155-- | The type for notifications callbacks in the client. When a registered notification fires due to a changed relational expression evaluation, the server propagates the notifications to the clients in the form of the callback.
156type NotificationCallback = NotificationName -> EvaluatedNotification -> IO ()
158-- | The empty notification callback ignores all callbacks.
159emptyNotificationCallback :: NotificationCallback
160emptyNotificationCallback _ _ = pure ()
162type GhcPkgPath = String
164data RemoteProcessDiedException = RemoteProcessDiedException
165 deriving (Show, Eq)
167instance Exception RemoteProcessDiedException
169data RequestTimeoutException = RequestTimeoutException
170 deriving (Show, Eq)
172instance Exception RequestTimeoutException
174-- | Construct a 'ConnectionInfo' to describe how to make the 'Connection'. The database can be run within the current process or running remotely via distributed-process.
175data ConnectionInfo = InProcessConnectionInfo PersistenceStrategy NotificationCallback [GhcPkgPath]|
176 RemoteProcessConnectionInfo DatabaseName NodeId NotificationCallback
178type EvaluatedNotifications = M.Map NotificationName EvaluatedNotification
180-- | Used for callbacks from the server when monitored changes have been made.
181newtype NotificationMessage = NotificationMessage EvaluatedNotifications
182 deriving (Binary, Eq, Show, Generic)
184-- | When a notification is fired, the 'reportExpr' is evaluated in the commit's context, so that is returned along with the original notification.
185data EvaluatedNotification = EvaluatedNotification {
186 notification :: Notification,
187 reportRelation :: Either RelationalError Relation
188 }
189 deriving(Binary, Eq, Show, Generic)
192-- | Create a 'NodeId' for use in connecting to a remote server using distributed-process.
193createNodeId :: Hostname -> Port -> NodeId
194createNodeId host port = NodeId $ encodeEndPointAddress host (show port) 1
196-- | Use this for connecting to remote servers on the default port.
197defaultServerPort :: Port
198defaultServerPort = 6543
200-- | Use this for connecting to remote servers with the default database name.
201defaultDatabaseName :: DatabaseName
202defaultDatabaseName = "base"
204-- | Use this for connecting to remote servers with the default head name.
205defaultHeadName :: HeadName
206defaultHeadName = "master"
208-- | Create a connection configuration which connects to the localhost on the default server port and default server database name. The configured notification callback is set to ignore all events.
209defaultRemoteConnectionInfo :: ConnectionInfo
210defaultRemoteConnectionInfo = RemoteProcessConnectionInfo defaultDatabaseName (createNodeId "127.0.0.1" defaultServerPort) emptyNotificationCallback
212-- | The 'Connection' represents either local or remote access to a database. All operations flow through the connection.
213type ClientNodes = STMSet.Set ProcessId
215type TransactionGraphLockHandle = Handle
217-- internal structure specific to in-process connections
218data InProcessConnectionConf = InProcessConnectionConf {
219 ipPersistenceStrategy :: PersistenceStrategy,
220 ipClientNodes :: ClientNodes,
221 ipSessions :: Sessions,
222 ipTransactionGraph :: TVar TransactionGraph,
223 ipScriptSession :: Maybe ScriptSession,
224 ipLocalNode :: LocalNode,
225 ipTransport :: Transport, -- we hold onto this so that we can close it gracefully
226 ipLocks :: Maybe (TransactionGraphLockHandle, MVar LockFileHash) -- nothing when NoPersistence
227 }
229data RemoteProcessConnectionConf = RemoteProcessConnectionConf {
230 rLocalNode :: LocalNode,
231 rProcessId :: ProcessId, --remote processId
232 rTransport :: Transport --the TCP socket transport
233 }
235data Connection = InProcessConnection InProcessConnectionConf |
236 RemoteProcessConnection RemoteProcessConnectionConf
238-- | There are several reasons why a connection can fail.
239data ConnectionError = SetupDatabaseDirectoryError PersistenceError |
240 IOExceptionError IOException |
241 NoSuchDatabaseByNameError DatabaseName |
242 LoginError
243 deriving (Show, Eq, Generic)
245remoteDBLookupName :: DatabaseName -> String
246remoteDBLookupName = (++) "db-"
248createLocalNode :: IO (LocalNode, Transport)
249createLocalNode = do
250 eLocalTransport <- createTransport "127.0.0.1" "0" defaultTCPParameters
251 case eLocalTransport of
252 Left err -> error ("failed to create transport: " ++ show err)
253 Right localTransport -> do
254 localNode <- newLocalNode localTransport initRemoteTable
255 pure (localNode, localTransport)
257notificationListener :: NotificationCallback -> Process ()
258notificationListener callback = do
259 --pid <- getSelfPid
260 --liftIO $ putStrLn $ "LISTENER THREAD START " ++ show pid
261 _ <- forever $
262 receiveWait [
263 match (\(NotificationMessage eNots) ->
264 --say $ "NOTIFICATION: " ++ show eNots
265 -- when notifications are thrown, they are not adjusted for the current schema which could be problematic, but we don't have access to the current session here
266 liftIO $ mapM_ (uncurry callback) (M.toList eNots)
267 )
268 ]
269 --say "LISTENER THREAD EXIT"
270 pure ()
272startNotificationListener :: LocalNode -> NotificationCallback -> IO ProcessId
273startNotificationListener localNode callback = forkProcess localNode (notificationListener callback)
275createScriptSession :: [String] -> IO (Maybe ScriptSession)
276createScriptSession ghcPkgPaths = do
277 eScriptSession <- initScriptSession ghcPkgPaths
278 case eScriptSession of
279 Left err -> hPutStrLn stderr ("Failed to load scripting engine- scripting disabled: " ++ show err) >> pure Nothing --not a fatal error, but the scripting feature must be disabled
280 Right s -> pure (Just s)
282-- | To create a 'Connection' to a remote or local database, create a 'ConnectionInfo' and call 'connectProjectM36'.
283connectProjectM36 :: ConnectionInfo -> IO (Either ConnectionError Connection)
284--create a new in-memory database/transaction graph
285connectProjectM36 (InProcessConnectionInfo strat notificationCallback ghcPkgPaths) = do
286 freshId <- nextRandom
287 stamp <- getCurrentTime
288 let bootstrapContext = basicDatabaseContext
289 freshGraph = bootstrapTransactionGraph stamp freshId bootstrapContext
290 case strat of
291 --create date examples graph for now- probably should be empty context in the future
292 NoPersistence -> do
293 graphTvar <- newTVarIO freshGraph
294 clientNodes <- STMSet.newIO
295 sessions <- STMMap.newIO
296 (localNode, transport) <- createLocalNode
297 notificationPid <- startNotificationListener localNode notificationCallback
298 mScriptSession <- createScriptSession ghcPkgPaths
300 let conn = InProcessConnection InProcessConnectionConf {
301 ipPersistenceStrategy = strat,
302 ipClientNodes = clientNodes,
303 ipSessions = sessions,
304 ipTransactionGraph = graphTvar,
305 ipScriptSession = mScriptSession,
306 ipLocalNode = localNode,
307 ipTransport = transport,
308 ipLocks = Nothing}
309 addClientNode conn notificationPid
310 pure (Right conn)
311 MinimalPersistence dbdir -> connectPersistentProjectM36 strat NoDiskSync dbdir freshGraph notificationCallback ghcPkgPaths
312 CrashSafePersistence dbdir -> connectPersistentProjectM36 strat FsyncDiskSync dbdir freshGraph notificationCallback ghcPkgPaths
314connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId notificationCallback) = do
315 connStatus <- newEmptyMVar
316 (localNode, transport) <- createLocalNode
317 let dbName = remoteDBLookupName databaseName
318 --putStrLn $ "Connecting to " ++ show serverNodeId ++ " " ++ dbName
319 notificationListenerPid <- startNotificationListener localNode notificationCallback
320 runProcess localNode $ do
321 --even a failed connection retains an open socket!
322 mServerProcessId <- whereisRemote serverNodeId dbName
323 case mServerProcessId of
324 Nothing -> liftIO $ putMVar connStatus $ Left (NoSuchDatabaseByNameError databaseName)
325 Just serverProcessId -> do
326 loginConfirmation <- safeLogin (Login notificationListenerPid) serverProcessId
327 if not loginConfirmation then
328 liftIO $ putMVar connStatus (Left LoginError)
329 else
330 liftIO $ putMVar connStatus (Right $ RemoteProcessConnection RemoteProcessConnectionConf {rLocalNode = localNode, rProcessId = serverProcessId, rTransport = transport})
331 status <- takeMVar connStatus
332 pure status
334connectPersistentProjectM36 :: PersistenceStrategy ->
335 DiskSync ->
336 FilePath ->
337 TransactionGraph ->
338 NotificationCallback ->
339 [GhcPkgPath] ->
340 IO (Either ConnectionError Connection)
341connectPersistentProjectM36 strat sync dbdir freshGraph notificationCallback ghcPkgPaths = do
342 err <- setupDatabaseDir sync dbdir freshGraph
343 case err of
344 Left err' -> return $ Left (SetupDatabaseDirectoryError err')
345 Right (lockFileH, digest) -> do
346 mScriptSession <- createScriptSession ghcPkgPaths
347 graph <- transactionGraphLoad dbdir emptyTransactionGraph mScriptSession
348 case graph of
349 Left err' -> return $ Left (SetupDatabaseDirectoryError err')
350 Right graph' -> do
351 tvarGraph <- newTVarIO graph'
352 sessions <- STMMap.newIO
353 clientNodes <- STMSet.newIO
354 (localNode, transport) <- createLocalNode
355 lockMVar <- newMVar digest
356 let conn = InProcessConnection InProcessConnectionConf {
357 ipPersistenceStrategy = strat,
358 ipClientNodes = clientNodes,
359 ipSessions = sessions,
360 ipTransactionGraph = tvarGraph,
361 ipScriptSession = mScriptSession,
362 ipLocalNode = localNode,
363 ipTransport = transport,
364 ipLocks = Just (lockFileH, lockMVar)
365 }
367 notificationPid <- startNotificationListener localNode notificationCallback
368 addClientNode conn notificationPid
369 pure (Right conn)
371-- | Create a new session at the transaction id and return the session's Id.
372createSessionAtCommit :: Connection -> TransactionId -> IO (Either RelationalError SessionId)
373createSessionAtCommit conn@(InProcessConnection _) commitId = do
374 newSessionId <- nextRandom
375 atomically $ createSessionAtCommit_ commitId newSessionId conn
376createSessionAtCommit conn@(RemoteProcessConnection _) uuid = remoteCall conn (CreateSessionAtCommit uuid)
378createSessionAtCommit_ :: TransactionId -> SessionId -> Connection -> STM (Either RelationalError SessionId)
379createSessionAtCommit_ commitId newSessionId (InProcessConnection conf) = do
380 let sessions = ipSessions conf
381 graphTvar = ipTransactionGraph conf
382 graph <- readTVar graphTvar
383 case transactionForId commitId graph of
384 Left err -> pure (Left err)
385 Right transaction -> do
386 let freshDiscon = DisconnectedTransaction commitId (Trans.schemas transaction) False
387 keyDuplication <- STMMap.lookup newSessionId sessions
388 case keyDuplication of
389 Just _ -> pure $ Left (SessionIdInUseError newSessionId)
390 Nothing -> do
391 STMMap.insert (Session freshDiscon defaultSchemaName) newSessionId sessions
392 pure $ Right newSessionId
393createSessionAtCommit_ _ _ (RemoteProcessConnection _) = error "createSessionAtCommit_ called on remote connection"
395-- | Call 'createSessionAtHead' with a transaction graph's head's name to create a new session pinned to that head. This function returns a 'SessionId' which can be used in other function calls to reference the point in the transaction graph.
396createSessionAtHead :: Connection -> HeadName -> IO (Either RelationalError SessionId)
397createSessionAtHead conn@(InProcessConnection conf) headn = do
398 let graphTvar = ipTransactionGraph conf
399 newSessionId <- nextRandom
400 atomically $ do
401 graph <- readTVar graphTvar
402 case transactionForHead headn graph of
403 Nothing -> pure $ Left (NoSuchHeadNameError headn)
404 Just trans -> createSessionAtCommit_ (transactionId trans) newSessionId conn
405createSessionAtHead conn@(RemoteProcessConnection _) headn = remoteCall conn (CreateSessionAtHead headn)
407-- | Used internally for server connections to keep track of remote nodes for the purpose of sending notifications later.
408addClientNode :: Connection -> ProcessId -> IO ()
409addClientNode (RemoteProcessConnection _) _ = error "addClientNode called on remote connection"
410addClientNode (InProcessConnection conf) newProcessId = atomically (STMSet.insert newProcessId (ipClientNodes conf))
412-- | Discards a session, eliminating any uncommitted changes present in the session.
413closeSession :: SessionId -> Connection -> IO ()
414closeSession sessionId (InProcessConnection conf) =
415 atomically $ STMMap.delete sessionId (ipSessions conf)
416closeSession sessionId conn@(RemoteProcessConnection _) = remoteCall conn (CloseSession sessionId)
417-- | 'close' cleans up the database access connection and closes any relevant sockets.
418close :: Connection -> IO ()
419close (InProcessConnection conf) = do
420 atomically $ do
421 let sessions = ipSessions conf
422 STMMap.deleteAll sessions
423 pure ()
424 closeLocalNode (ipLocalNode conf)
425 closeTransport (ipTransport conf)
427close conn@(RemoteProcessConnection conf) = do
428 _ <- remoteCall conn Logout :: IO Bool
429 closeLocalNode (rLocalNode conf)
430 closeTransport (rTransport conf)
432--used only by the server EntryPoints
433closeRemote_ :: Connection -> IO ()
434closeRemote_ (InProcessConnection _) = error "invalid call of closeRemote_ on InProcessConnection"
435closeRemote_ (RemoteProcessConnection conf) = runProcess (rLocalNode conf) (reconnect (rProcessId conf))
437 --we need to actually close the localNode's connection to the remote
438--within the database server, we must catch and handle all exception lest they take down the database process- this handling might be different for other use-cases
439--exceptions should generally *NOT* be thrown from any Project:M36 code paths, but third-party code such as AtomFunction scripts could conceivably throw undefined, etc.
441excEither :: IO (Either RelationalError a) -> IO (Either RelationalError a)
442excEither = handle handler
443 where
444 handler exc | Just (_ :: AsyncException) <- fromException exc = throwIO exc
445 | otherwise = pure (Left (UnhandledExceptionError (show exc)))
447runProcessResult :: LocalNode -> Process a -> IO a
448runProcessResult localNode proc = do
449 ret <- newEmptyMVar
450 runProcess localNode $ do
451 val <- proc
452 liftIO $ putMVar ret val
453 takeMVar ret
455safeLogin :: Login -> ProcessId -> Process Bool
456safeLogin login procId = do
457 ret <- call procId login
458 case ret of
459 Left (_ :: ServerError) -> pure False
460 Right val -> pure val
462remoteCall :: (Serializable a, Serializable b) => Connection -> a -> IO b
463remoteCall (InProcessConnection _ ) _ = error "remoteCall called on local connection"
464remoteCall (RemoteProcessConnection conf) arg = runProcessResult localNode $ do
465 ret <- safeCall serverProcessId arg
466 case ret of
467 Left err -> error ("server died: " ++ show err)
468 Right ret' -> case ret' of
469 Left RequestTimeoutError -> liftIO (throwIO RequestTimeoutException)
470 Left (ProcessDiedError _) -> liftIO (throwIO RemoteProcessDiedException)
471 Right val -> pure val
472 where
473 localNode = rLocalNode conf
474 serverProcessId = rProcessId conf
476sessionForSessionId :: SessionId -> Sessions -> STM (Either RelationalError Session)
477sessionForSessionId sessionId sessions = do
478 maybeSession <- STMMap.lookup sessionId sessions
479 pure $ maybe (Left $ NoSuchSessionError sessionId) Right maybeSession
481schemaForSessionId :: Session -> STM (Either RelationalError Schema)
482schemaForSessionId session = do
483 let sname = schemaName session
484 if sname == defaultSchemaName then
485 pure (Right (Schema [])) -- the main schema includes no transformations (but neither do empty schemas :/ )
486 else
487 case M.lookup sname (subschemas session) of
488 Nothing -> pure (Left (SubschemaNameNotInUseError sname))
489 Just schema -> pure (Right schema)
491sessionAndSchema :: SessionId -> Sessions -> STM (Either RelationalError (Session, Schema))
492sessionAndSchema sessionId sessions = do
493 eSession <- sessionForSessionId sessionId sessions
494 case eSession of
495 Left err -> pure (Left err)
496 Right session -> do
497 eSchema <- schemaForSessionId session
498 case eSchema of
499 Left err -> pure (Left err)
500 Right schema -> pure (Right (session, schema))
502-- | Returns the name of the currently selected isomorphic schema.
503currentSchemaName :: SessionId -> Connection -> IO (Either RelationalError SchemaName)
504currentSchemaName sessionId (InProcessConnection conf) = atomically $ do
505 let sessions = ipSessions conf
506 eSession <- sessionForSessionId sessionId sessions
507 case eSession of
508 Left err -> pure (Left err)
509 Right session -> pure (Right (Sess.schemaName session))
510currentSchemaName sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveCurrentSchemaName sessionId)
512-- | Switch to the named isomorphic schema.
513setCurrentSchemaName :: SessionId -> Connection -> SchemaName -> IO (Either RelationalError ())
514setCurrentSchemaName sessionId (InProcessConnection conf) sname = atomically $ do
515 let sessions = ipSessions conf
516 eSession <- sessionForSessionId sessionId sessions
517 case eSession of
518 Left err -> pure (Left err)
519 Right session -> case Sess.setSchemaName sname session of
520 Left err -> pure (Left err)
521 Right newSession -> STMMap.insert newSession sessionId sessions >> pure (Right ())
522setCurrentSchemaName sessionId conn@(RemoteProcessConnection _) sname = remoteCall conn (ExecuteSetCurrentSchema sessionId sname)
524-- | Execute a relational expression in the context of the session and connection. Relational expressions are queries and therefore cannot alter the database.
525executeRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)
526executeRelationalExpr sessionId (InProcessConnection conf) expr = excEither $ atomically $ do
527 let sessions = ipSessions conf
528 eSession <- sessionAndSchema sessionId sessions
529 case eSession of
530 Left err -> pure $ Left err
531 Right (session, schema) -> do
532 let expr' = if schemaName session /= defaultSchemaName then
533 Schema.processRelationalExprInSchema schema expr
534 else
535 Right expr
536 case expr' of
537 Left err -> pure (Left err)
538 Right expr'' -> case runReader (RE.evalRelationalExpr expr'') (RE.mkRelationalExprState (Sess.concreteDatabaseContext session)) of
539 Left err -> pure (Left err)
540 Right rel -> pure (force (Right rel)) -- this is necessary so that any undefined/error exceptions are spit out here
541executeRelationalExpr sessionId conn@(RemoteProcessConnection _) relExpr = remoteCall conn (ExecuteRelationalExpr sessionId relExpr)
543-- | Execute a database context expression in the context of the session and connection. Database expressions modify the current session's disconnected transaction but cannot modify the transaction graph.
544executeDatabaseContextExpr :: SessionId -> Connection -> DatabaseContextExpr -> IO (Either RelationalError ())
545executeDatabaseContextExpr sessionId (InProcessConnection conf) expr = excEither $ atomically $ do
546 let sessions = ipSessions conf
547 eSession <- sessionAndSchema sessionId sessions
548 case eSession of
549 Left err -> pure (Left err)
550 Right (session, schema) -> do
551 let expr' = if schemaName session == defaultSchemaName then
552 Right expr
553 else
554 Schema.processDatabaseContextExprInSchema schema expr
555 case expr' of
556 Left err -> pure (Left err)
557 Right expr'' -> case runState (RE.evalDatabaseContextExpr expr'') (RE.freshDatabaseState (Sess.concreteDatabaseContext session)) of
558 (Just err,_) -> pure (Left err)
559 (Nothing, (_,_,False)) -> pure (Right ()) --optimization- if nothing was dirtied, nothing to do
560 (Nothing, (!context',_,True)) -> do
561 let newDiscon = DisconnectedTransaction (Sess.parentId session) newSchemas True
562 newSubschemas = Schema.processDatabaseContextExprSchemasUpdate (Sess.subschemas session) expr
563 newSchemas = Schemas context' newSubschemas
564 newSession = Session newDiscon (Sess.schemaName session)
565 STMMap.insert newSession sessionId sessions
566 pure (Right ())
567executeDatabaseContextExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (ExecuteDatabaseContextExpr sessionId dbExpr)
569-- | Similar to a git rebase, 'autoMergeToHead' atomically creates a temporary branch and merges it to the latest commit of the branch referred to by the 'HeadName' and commits the merge. This is useful to reduce incidents of 'TransactionIsNotAHeadError's but at the risk of merge errors (thus making it similar to rebasing).
570autoMergeToHead :: SessionId -> Connection -> MergeStrategy -> HeadName -> IO (Either RelationalError ())
571autoMergeToHead sessionId (InProcessConnection conf) strat headName' = do
572 let sessions = ipSessions conf
573 id1 <- nextRandom
574 id2 <- nextRandom
575 id3 <- nextRandom
576 stamp <- getCurrentTime
577 commitLock_ sessionId conf $ \graph -> do
578 eSession <- sessionForSessionId sessionId sessions
579 case eSession of
580 Left err -> pure (Left err)
581 Right session ->
582 case Graph.autoMergeToHead stamp (id1, id2, id3) (Sess.disconnectedTransaction session) headName' strat graph of
583 Left err -> pure (Left err)
584 Right (discon', graph') ->
585 pure (Right (discon', graph', [id1, id2, id3]))
586autoMergeToHead sessionId conn@(RemoteProcessConnection _) strat headName' = remoteCall conn (ExecuteAutoMergeToHead sessionId strat headName')
588-- | Execute a database context IO-monad-based expression for the given session and connection. `DatabaseContextIOExpr`s modify the DatabaseContext but cannot be purely implemented.
589--this is almost completely identical to executeDatabaseContextExpr above
590executeDatabaseContextIOExpr :: SessionId -> Connection -> DatabaseContextIOExpr -> IO (Either RelationalError ())
591executeDatabaseContextIOExpr sessionId (InProcessConnection conf) expr = excEither $ do
592 let sessions = ipSessions conf
593 scriptSession = ipScriptSession conf
594 eSession <- atomically $ sessionForSessionId sessionId sessions --potentially race condition due to interleaved IO?
595 case eSession of
596 Left err -> pure (Left err)
597 Right session -> do
598 res <- RE.evalDatabaseContextIOExpr scriptSession (Sess.concreteDatabaseContext session) expr
599 case res of
600 Left err -> pure (Left err)
601 Right context' -> do
602 let newDiscon = DisconnectedTransaction (Sess.parentId session) newSchemas True
603 newSchemas = Schemas context' (Sess.subschemas session)
604 newSession = Session newDiscon (Sess.schemaName session)
605 atomically $ STMMap.insert newSession sessionId sessions
606 pure (Right ())
607executeDatabaseContextIOExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (ExecuteDatabaseContextIOExpr sessionId dbExpr)
609{-
610executeGraphExprSTM_ :: TransactionId -> SessionId -> Session -> Sessions -> TransactionGraphOperator -> TransactionGraph -> TVar TransactionGraph -> STM (Either RelationalError (TransactionGraph, DisconnectedTransaction)
611executeGraphExprSTM_ freshId sessionId session sessions graphExpr graph graphTVar= do
612 case evalGraphOp freshId (Sess.disconnectedTransaction session) graph graphExpr of
613 Left err -> do
614 when updateGraphOnError (writeTVar graphTVar graph)
615 pure $ Left err
616 Right (discon', graph') -> do
617 writeTVar graphTVar graph'
618 let newSession = Session discon' (Sess.schemaName session)
619 STMMap.insert newSession sessionId sessions
620 pure $ Right graph'
621-}
623-- process notifications for commits
624executeCommitExprSTM_ :: DatabaseContext -> DatabaseContext -> ClientNodes -> STM (EvaluatedNotifications, ClientNodes)
625executeCommitExprSTM_ oldContext newContext nodes = do
626 let nots = notifications oldContext
627 fireNots = notificationChanges nots oldContext newContext
628 evaldNots = M.map mkEvaldNot fireNots
629 mkEvaldNot notif = EvaluatedNotification { notification = notif,
630 reportRelation = runReader (RE.evalRelationalExpr (reportExpr notif)) (RE.mkRelationalExprState oldContext) }
631 pure (evaldNots, nodes)
633-- | Execute a transaction graph expression in the context of the session and connection. Transaction graph operators modify the transaction graph state.
635-- OPTIMIZATION OPPORTUNITY: no locks are required to write new transaction data, only to update the transaction graph id file
636-- if writing data is re-entrant, we may be able to use unsafeIOtoSTM
637-- perhaps keep hash of data file instead of checking if our head was updated on every write
638executeGraphExpr :: SessionId -> Connection -> TransactionGraphOperator -> IO (Either RelationalError ())
639executeGraphExpr sessionId (InProcessConnection conf) graphExpr = excEither $ do
640 let sessions = ipSessions conf
641 freshId <- nextRandom
642 stamp <- getCurrentTime
643 commitLock_ sessionId conf $ \updatedGraph -> do
644 eSession <- sessionForSessionId sessionId sessions
645 case eSession of
646 Left err -> pure (Left err)
647 Right session -> do
648 let discon = Sess.disconnectedTransaction session
649 case evalGraphOp stamp freshId discon updatedGraph graphExpr of
650 Left err -> pure (Left err)
651 Right (discon', graph') -> do
652 --if freshId appears in the graph, then we need to pass it on
653 let transIds = [freshId | isRight (transactionForId freshId graph')]
654 pure (Right (discon', graph', transIds))
655{-
656executeGraphExpr sessionId (InProcessConnection conf) graphExpr = excEither $ do
657 let strat = ipPersistenceStrategy conf
658 clientNodes = ipClientNodes conf
659 sessions = ipSessions conf
660 graphTvar = ipTransactionGraph conf
661 mLockFileH = ipLocks conf
662 lockHandler body = case graphExpr of
663 Commit -> case mLockFileH of
664 Nothing -> body False
665 Just (lockFileH, lockMVar) -> let acquireLocks = do
666 lastWrittenDigest <- takeMVar lockMVar
667 lockFile lockFileH WriteLock
668 latestDigest <- readGraphTransactionIdDigest strat
669 pure (latestDigest /= lastWrittenDigest)
671 releaseLocks _ = do
672 --still holding the lock- get the latest digest
673 gDigest <- readGraphTransactionIdDigest strat
674 unlockFile lockFileH
675 putMVar lockMVar gDigest
676 in bracket acquireLocks releaseLocks body
677 _ -> body False
678 freshId <- nextRandom
679 lockHandler $ \dbWrittenByOtherProcess -> do
680 --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
681 --read transaction data and compare to existing graph
682 --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)
683 manip <- atomically $ do
684 eSession <- sessionForSessionId sessionId sessions
685 --handle graph update by other process
686 oldGraph <- readTVar graphTvar
687 case eSession of
688 Left err -> pure (Left err)
689 Right session -> do
690 let mScriptSession = ipScriptSession conf
691 dbdir = case strat of
692 MinimalPersistence x -> x
693 CrashSafePersistence x -> x
694 _ -> error "accessing dbdir on non-persisted connection"
695 eRefreshedGraph <- if dbWrittenByOtherProcess then
696 unsafeIOToSTM (transactionGraphLoad dbdir oldGraph mScriptSession)
697 else
698 pure (Right oldGraph)
699 case eRefreshedGraph of
700 Left err -> pure (Left (DatabaseLoadError err))
701 Right refreshedGraph -> do
702 --snip it
703 eGraph <- executeGraphExprSTM_ dbWrittenByOtherProcess freshId sessionId session sessions graphExpr refreshedGraph graphTvar
704 --snip it
705 case eGraph of
706 Left err -> pure (Left err)
707 Right newGraph -> do
708 --handle commit
709 if isCommit graphExpr then do
710 case transactionForId (Sess.parentId session) oldGraph of
711 Left err -> pure $ Left err
712 Right previousTrans -> do
713 (evaldNots, nodes) <- executeCommitExprSTM_ (Trans.concreteDatabaseContext previousTrans) (Sess.concreteDatabaseContext session) clientNodes
714 nodesToNotify <- toList (STMSet.stream nodes)
715 pure $ Right (evaldNots, nodesToNotify, newGraph)
716 else
717 pure $ Right (M.empty, [], newGraph)
718 case manip of
719 Left err -> return $ Just err
720 Right (notsToFire, nodesToNotify, newGraph) -> do
721 --update filesystem database, if necessary
722 processTransactionGraphPersistence strat newGraph
723 sendNotifications nodesToNotify (ipLocalNode conf) notsToFire
724 pure Nothing
725-}
726executeGraphExpr sessionId conn@(RemoteProcessConnection _) graphExpr = remoteCall conn (ExecuteGraphExpr sessionId graphExpr)
728-- | A trans-graph expression is a relational query executed against the entirety of a transaction graph.
729executeTransGraphRelationalExpr :: SessionId -> Connection -> TransGraphRelationalExpr -> IO (Either RelationalError Relation)
730executeTransGraphRelationalExpr _ (InProcessConnection conf) tgraphExpr = excEither . atomically $ do
731 let graphTvar = ipTransactionGraph conf
732 graph <- readTVar graphTvar
733 case evalTransGraphRelationalExpr tgraphExpr graph of
734 Left err -> pure (Left err)
735 Right relExpr -> case runReader (RE.evalRelationalExpr relExpr) (RE.mkRelationalExprState DBC.empty) of
736 Left err -> pure (Left err)
737 Right rel -> pure (force (Right rel))
738executeTransGraphRelationalExpr sessionId conn@(RemoteProcessConnection _) tgraphExpr = remoteCall conn (ExecuteTransGraphRelationalExpr sessionId tgraphExpr)
740-- | Schema expressions manipulate the isomorphic schemas for the current 'DatabaseContext'.
741executeSchemaExpr :: SessionId -> Connection -> Schema.SchemaExpr -> IO (Either RelationalError ())
742executeSchemaExpr sessionId (InProcessConnection conf) schemaExpr = atomically $ do
743 let sessions = ipSessions conf
744 eSession <- sessionAndSchema sessionId sessions
745 case eSession of
746 Left err -> pure (Left err)
747 Right (session, _) -> do
748 let subschemas' = subschemas session
749 case Schema.evalSchemaExpr schemaExpr (Sess.concreteDatabaseContext session) subschemas' of
750 Left err -> pure (Left err)
751 Right (newSubschemas, newContext) -> do
752 --hm- maybe we should start using lenses
753 let discon = Sess.disconnectedTransaction session
754 newSchemas = Schemas newContext newSubschemas
755 newSession = Session (DisconnectedTransaction (Discon.parentId discon) newSchemas True) (Sess.schemaName session)
756 STMMap.insert newSession sessionId sessions
757 pure (Right ())
758executeSchemaExpr sessionId conn@(RemoteProcessConnection _) schemaExpr = remoteCall conn (ExecuteSchemaExpr sessionId schemaExpr)
760-- | After modifying a 'DatabaseContext', 'commit' the transaction to the transaction graph at the head which the session is referencing. This will also trigger checks for any notifications which need to be propagated.
761commit :: SessionId -> Connection -> IO (Either RelationalError ())
762commit sessionId conn@(InProcessConnection _) = executeGraphExpr sessionId conn Commit
763commit sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteGraphExpr sessionId Commit)
765sendNotifications :: [ProcessId] -> LocalNode -> EvaluatedNotifications -> IO ()
766sendNotifications pids localNode nots = mapM_ sendNots pids
767 where
768 sendNots remoteClientPid =
769 unless (M.null nots) $ runProcess localNode $ send remoteClientPid (NotificationMessage nots)
771-- | Discard any changes made in the current 'Session' and 'DatabaseContext'. This resets the disconnected transaction to reference the original database context of the parent transaction and is a very cheap operation.
772rollback :: SessionId -> Connection -> IO (Either RelationalError ())
773rollback sessionId conn@(InProcessConnection _) = executeGraphExpr sessionId conn Rollback
774rollback sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteGraphExpr sessionId Rollback)
776-- | Write the transaction graph to disk. This function can be used to incrementally write new transactions to disk.
777processTransactionGraphPersistence :: PersistenceStrategy -> [TransactionId] -> TransactionGraph -> IO ()
778processTransactionGraphPersistence NoPersistence _ _ = pure ()
779processTransactionGraphPersistence (MinimalPersistence dbdir) transIds graph = transactionGraphPersist NoDiskSync dbdir transIds graph >> pure ()
780processTransactionGraphPersistence (CrashSafePersistence dbdir) transIds graph = transactionGraphPersist FsyncDiskSync dbdir transIds graph >> pure ()
782readGraphTransactionIdDigest :: PersistenceStrategy -> IO LockFileHash
783readGraphTransactionIdDigest NoPersistence = error "attempt to read digest from transaction log without persistence enabled"
784readGraphTransactionIdDigest (MinimalPersistence dbdir) = readGraphTransactionIdFileDigest dbdir
785readGraphTransactionIdDigest (CrashSafePersistence dbdir) = readGraphTransactionIdFileDigest dbdir
787-- | 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.
788typeForRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)
789typeForRelationalExpr sessionId conn@(InProcessConnection _) relExpr = atomically $ typeForRelationalExprSTM sessionId conn relExpr
790typeForRelationalExpr sessionId conn@(RemoteProcessConnection _) relExpr = remoteCall conn (ExecuteTypeForRelationalExpr sessionId relExpr)
792typeForRelationalExprSTM :: SessionId -> Connection -> RelationalExpr -> STM (Either RelationalError Relation)
793typeForRelationalExprSTM sessionId (InProcessConnection conf) relExpr = do
794 let sessions = ipSessions conf
795 eSession <- sessionAndSchema sessionId sessions
796 case eSession of
797 Left err -> pure $ Left err
798 Right (session, schema) -> do
799 let processed = if schemaName session == defaultSchemaName then
800 Right relExpr
801 else
802 Schema.processRelationalExprInSchema schema relExpr
803 case processed of
804 Left err -> pure (Left err)
805 Right relExpr' -> pure $ runReader (RE.typeForRelationalExpr relExpr') (RE.mkRelationalExprState (Sess.concreteDatabaseContext session))
807typeForRelationalExprSTM _ _ _ = error "typeForRelationalExprSTM called on non-local connection"
809-- | Return a 'Map' of the database's constraints at the context of the session and connection.
810inclusionDependencies :: SessionId -> Connection -> IO (Either RelationalError InclusionDependencies)
811inclusionDependencies sessionId (InProcessConnection conf) = do
812 let sessions = ipSessions conf
813 atomically $ do
814 eSession <- sessionAndSchema sessionId sessions
815 case eSession of
816 Left err -> pure $ Left err
817 Right (session, schema) -> do
818 let context = Sess.concreteDatabaseContext session
819 if schemaName session == defaultSchemaName then
820 pure $ Right (B.inclusionDependencies context)
821 else
822 pure (Schema.inclusionDependenciesInSchema schema (B.inclusionDependencies context))
824inclusionDependencies sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveInclusionDependencies sessionId)
826-- | Return an optimized database expression which is logically equivalent to the input database expression. This function can be used to determine which expression will actually be evaluated.
827planForDatabaseContextExpr :: SessionId -> Connection -> DatabaseContextExpr -> IO (Either RelationalError DatabaseContextExpr)
828planForDatabaseContextExpr sessionId (InProcessConnection conf) dbExpr = do
829 let sessions = ipSessions conf
830 atomically $ do
831 eSession <- sessionAndSchema sessionId sessions
832 case eSession of
833 Left err -> pure $ Left err
834 Right (session, _) -> if schemaName session == defaultSchemaName then
835 pure $ evalState (applyStaticDatabaseOptimization dbExpr) (RE.freshDatabaseState (Sess.concreteDatabaseContext session))
836 else -- don't show any optimization because the current optimization infrastructure relies on access to the base context- this probably underscores the need for each schema to have its own DatabaseContext, even if it is generated on-the-fly
837 pure (Right dbExpr)
839planForDatabaseContextExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (RetrievePlanForDatabaseContextExpr sessionId dbExpr)
841-- | Return a relation which represents the current state of the global transaction graph. The attributes are
842-- * current- boolean attribute representing whether or not the current session references this transaction
843-- * head- text attribute which is a non-empty 'HeadName' iff the transaction references a head.
844-- * id- id attribute of the transaction
845-- * parents- a relation-valued attribute which contains a relation of transaction ids which are parent transaction to the transaction
846transactionGraphAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
847transactionGraphAsRelation sessionId (InProcessConnection conf) = do
848 let sessions = ipSessions conf
849 tvar = ipTransactionGraph conf
850 atomically $ do
851 eSession <- sessionForSessionId sessionId sessions
852 case eSession of
853 Left err -> pure $ Left err
854 Right session -> do
855 graph <- readTVar tvar
856 pure $ graphAsRelation (Sess.disconnectedTransaction session) graph
858transactionGraphAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveTransactionGraph sessionId)
860-- | Returns the names and types of the relation variables in the current 'Session'.
861relationVariablesAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
862relationVariablesAsRelation sessionId (InProcessConnection conf) = do
863 let sessions = ipSessions conf
864 atomically $ do
865 eSession <- sessionAndSchema sessionId sessions
866 case eSession of
867 Left err -> pure (Left err)
868 Right (session, schema) -> do
869 let context = Sess.concreteDatabaseContext session
870 if Sess.schemaName session == defaultSchemaName then
871 pure $ R.relationVariablesAsRelation (relationVariables context)
872 else
873 case Schema.relationVariablesInSchema schema context of
874 Left err -> pure (Left err)
875 Right relvars -> pure $ R.relationVariablesAsRelation relvars
877relationVariablesAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveRelationVariableSummary sessionId)
879-- | Returns the transaction id for the connection's disconnected transaction committed parent transaction.
880headTransactionId :: SessionId -> Connection -> IO (Either RelationalError TransactionId)
881headTransactionId sessionId (InProcessConnection conf) = do
882 let sessions = ipSessions conf
883 atomically $ do
884 eSession <- sessionForSessionId sessionId sessions
885 case eSession of
886 Left err -> pure (Left err)
887 Right session -> pure $ Right (Sess.parentId session)
888headTransactionId sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveHeadTransactionId sessionId)
890headNameSTM_ :: SessionId -> Sessions -> TVar TransactionGraph -> STM (Either RelationalError HeadName)
891headNameSTM_ sessionId sessions graphTvar = do
892 graph <- readTVar graphTvar
893 eSession <- sessionForSessionId sessionId sessions
894 case eSession of
895 Left err -> pure (Left err)
896 Right session -> case transactionForId (Sess.parentId session) graph of
897 Left err -> pure (Left err)
898 Right parentTrans -> case headNameForTransaction parentTrans graph of
899 Nothing -> pure (Left UnknownHeadError)
900 Just headName' -> pure (Right headName')
902-- | Returns Just the name of the head of the current disconnected transaction or Nothing.
903headName :: SessionId -> Connection -> IO (Either RelationalError HeadName)
904headName sessionId (InProcessConnection conf) = do
905 let sessions = ipSessions conf
906 graphTvar = ipTransactionGraph conf
907 atomically (headNameSTM_ sessionId sessions graphTvar)
908headName sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteHeadName sessionId)
910-- | Returns a listing of all available atom types.
911atomTypesAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
912atomTypesAsRelation sessionId (InProcessConnection conf) = do
913 let sessions = ipSessions conf
914 atomically $ do
915 eSession <- sessionForSessionId sessionId sessions
916 case eSession of
917 Left err -> pure (Left err)
918 Right session ->
919 case typesAsRelation (typeConstructorMapping (Sess.concreteDatabaseContext session)) of
920 Left err -> pure (Left err)
921 Right rel -> pure (Right rel)
922atomTypesAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveAtomTypesAsRelation sessionId)
924disconnectedTransactionIsDirty :: SessionId -> Connection -> IO (Either RelationalError Bool)
925disconnectedTransactionIsDirty sessionId (InProcessConnection conf) = do
926 let sessions = ipSessions conf
927 atomically $ do
928 eSession <- sessionForSessionId sessionId sessions
929 case eSession of
930 Left err -> pure (Left err)
931 Right session ->
932 pure (Right (isDirty session))
933disconnectedTransactionIsDirty sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveSessionIsDirty sessionId)
935--used only for testing- we expect this to throw an exception
936callTestTimeout_ :: SessionId -> Connection -> IO Bool
937callTestTimeout_ _ (InProcessConnection _) = error "bad testing call"
938callTestTimeout_ sessionId conn@(RemoteProcessConnection _) = remoteCall conn (TestTimeout sessionId)
940--used in tests only
941transactionGraph_ :: Connection -> IO TransactionGraph
942transactionGraph_ (InProcessConnection conf) = atomically $ readTVar (ipTransactionGraph conf)
943transactionGraph_ _ = error "remote connection used"
945--used in tests only
946disconnectedTransaction_ :: SessionId -> Connection -> IO DisconnectedTransaction
947disconnectedTransaction_ sessionId (InProcessConnection conf) = do
948 let sessions = ipSessions conf
949 mSession <- atomically $ STMMap.lookup sessionId sessions
950 case mSession of
951 Nothing -> error "No such session"
952 Just (Sess.Session discon _) -> pure discon
953disconnectedTransaction_ _ _= error "remote connection used"
955-- wrap a graph evaluation in file locking
956commitLock_ :: SessionId ->
957 InProcessConnectionConf ->
958 (TransactionGraph ->
959 STM (Either RelationalError (DisconnectedTransaction, TransactionGraph, [TransactionId]))) ->
960 IO (Either RelationalError ())
961commitLock_ sessionId conf stmBlock = do
962 let sessions = ipSessions conf
963 strat = ipPersistenceStrategy conf
964 mScriptSession = ipScriptSession conf
965 graphTvar = ipTransactionGraph conf
966 clientNodes = ipClientNodes conf
967 mLockFileH = ipLocks conf
968 lockHandler body = case mLockFileH of
969 Nothing -> body False
970 Just (lockFileH, lockMVar) ->
971 let acquireLocks = do
972 lastWrittenDigest <- takeMVar lockMVar
973 lockFile lockFileH WriteLock
974 latestDigest <- readGraphTransactionIdDigest strat
975 pure (latestDigest /= lastWrittenDigest)
976 releaseLocks _ = do
977 --still holding the lock- get the latest digest
978 gDigest <- readGraphTransactionIdDigest strat
979 unlockFile lockFileH
980 putMVar lockMVar gDigest
981 in bracket acquireLocks releaseLocks body
982 manip <- lockHandler $ \dbWrittenByOtherProcess -> atomically $ do
983 eSession <- sessionForSessionId sessionId sessions
984 --handle graph update by other process
985 oldGraph <- readTVar graphTvar
986 case eSession of
987 Left err -> pure (Left err)
988 Right session -> do
989 let dbdir = case strat of
990 MinimalPersistence x -> x
991 CrashSafePersistence x -> x
992 _ -> error "accessing dbdir on non-persisted connection"
993 --this should also happen for non-commit expressions
994 eRefreshedGraph <- if dbWrittenByOtherProcess then
995 unsafeIOToSTM (transactionGraphLoad dbdir oldGraph mScriptSession)
996 else
997 pure (Right oldGraph)
998 case eRefreshedGraph of
999 Left err -> pure (Left (DatabaseLoadError err))
1000 Right refreshedGraph -> do
1001 eGraph <- stmBlock refreshedGraph
1002 case eGraph of
1003 Left err -> pure (Left err)
1004 Right (discon', graph', transactionIdsToPersist) -> do
1005 writeTVar graphTvar graph'
1006 let newSession = Session discon' (Sess.schemaName session)
1007 STMMap.insert newSession sessionId sessions
1008 case transactionForId (Sess.parentId session) oldGraph of
1009 Left err -> pure $ Left err
1010 Right previousTrans ->
1011 if not (Prelude.null transactionIdsToPersist) then do
1012 (evaldNots, nodes) <- executeCommitExprSTM_ (Trans.concreteDatabaseContext previousTrans) (Sess.concreteDatabaseContext session) clientNodes
1013 nodesToNotify <- toList (STMSet.stream nodes)
1014 pure $ Right (evaldNots, nodesToNotify, graph', transactionIdsToPersist)
1015 else pure (Right (M.empty, [], graph', []))
1017 --handle notification firing
1018 case manip of
1019 Left err -> pure (Left err)
1020 Right (notsToFire, nodesToNotify, newGraph, transactionIdsToPersist) -> do
1021 --update filesystem database, if necessary
1022 processTransactionGraphPersistence strat transactionIdsToPersist newGraph
1023 sendNotifications nodesToNotify (ipLocalNode conf) notsToFire
1024 pure (Right ())
1026writeDisconAndGraph_ :: TVar TransactionGraph -> SessionId -> Session -> Sessions -> DisconnectedTransaction -> TransactionGraph -> STM ()
1027writeDisconAndGraph_ graphTvar sessionId session sessions discon graph = do
1028 writeTVar graphTvar graph
1029 let newSession = Session discon (Sess.schemaName session)
1030 STMMap.insert newSession sessionId sessions