git.iko.soy/project-m36
on fd05d50 help sign in
project-m36/src/lib/ProjectM36/Client.hs
1057 lines 55 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 RelationalError(..),
87 RequestTimeoutException(..),
88 RemoteProcessDiedException(..),
89 AtomType(..),
90 Atomable(..),
91 TupleExprBase(..),
92 AtomExprBase(..),
93 RestrictionPredicateExprBase(..),
94 withTransaction
95 ) where
96import ProjectM36.Base hiding (inclusionDependencies) --defined in this module as well
97import qualified ProjectM36.Base as B
98import ProjectM36.Error
99import ProjectM36.Atomable
100import ProjectM36.AtomFunction
101import ProjectM36.StaticOptimizer
102import ProjectM36.Key
103import qualified ProjectM36.IsomorphicSchema as Schema
104import Control.Monad.State
105import Control.Monad.Trans.Reader
106import qualified ProjectM36.RelationalExpression as RE
107import ProjectM36.DatabaseContext (basicDatabaseContext)
108import qualified ProjectM36.TransactionGraph as Graph
109import ProjectM36.TransactionGraph
110import qualified ProjectM36.Transaction as Trans
111import ProjectM36.TransactionGraph.Persist
112import ProjectM36.Attribute hiding (atomTypes)
113import ProjectM36.TransGraphRelationalExpression (TransGraphRelationalExpr, evalTransGraphRelationalExpr)
114import ProjectM36.Persist (DiskSync(..))
115import ProjectM36.FileLock
116import ProjectM36.Notifications
117import ProjectM36.Server.RemoteCallTypes
118import qualified ProjectM36.DisconnectedTransaction as Discon
119import ProjectM36.Relation (typesAsRelation)
120import ProjectM36.ScriptSession (initScriptSession, ScriptSession)
121import qualified ProjectM36.Relation as R
122import qualified ProjectM36.DatabaseContext as DBC
123import Control.Exception.Base
124import GHC.Conc.Sync
126import Network.Transport (Transport(closeTransport))
127import Network.Transport.TCP (createTransport, defaultTCPParameters, encodeEndPointAddress)
128import Control.Distributed.Process.Node (newLocalNode, initRemoteTable, runProcess, LocalNode, forkProcess, closeLocalNode)
129import Control.Distributed.Process.Extras.Internal.Types (whereisRemote)
130import Control.Distributed.Process.ManagedProcess.Client (call, safeCall)
131import Data.Either (isRight)
132import Data.UUID.V4 (nextRandom)
133import Data.Word
134import Control.Distributed.Process (ProcessId, Process, receiveWait, send, match, NodeId(..), reconnect)
135import Control.Exception (IOException, handle, AsyncException, throwIO, fromException, Exception)
136import Control.Concurrent.MVar
137import qualified Data.Map as M
138import Control.Distributed.Process.Serializable (Serializable)
139import qualified STMContainers.Map as STMMap
140import qualified STMContainers.Set as STMSet
141import qualified ProjectM36.Session as Sess
142import ProjectM36.Session
143import ProjectM36.Sessions
144import ListT
145import Data.Binary (Binary)
146import GHC.Generics (Generic)
147import Control.DeepSeq (force)
148import System.IO
149import Data.Time.Clock
151type Hostname = String
153type Port = Word16
155type DatabaseName = String
157-- | 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.
158type NotificationCallback = NotificationName -> EvaluatedNotification -> IO ()
160-- | The empty notification callback ignores all callbacks.
161emptyNotificationCallback :: NotificationCallback
162emptyNotificationCallback _ _ = pure ()
164type GhcPkgPath = String
166data RemoteProcessDiedException = RemoteProcessDiedException
167 deriving (Show, Eq)
169instance Exception RemoteProcessDiedException
171data RequestTimeoutException = RequestTimeoutException
172 deriving (Show, Eq)
174instance Exception RequestTimeoutException
176-- | 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.
177data ConnectionInfo = InProcessConnectionInfo PersistenceStrategy NotificationCallback [GhcPkgPath]|
178 RemoteProcessConnectionInfo DatabaseName NodeId NotificationCallback
180type EvaluatedNotifications = M.Map NotificationName EvaluatedNotification
182-- | Used for callbacks from the server when monitored changes have been made.
183newtype NotificationMessage = NotificationMessage EvaluatedNotifications
184 deriving (Binary, Eq, Show, Generic)
186-- | When a notification is fired, the 'reportExpr' is evaluated in the commit's context, so that is returned along with the original notification.
187data EvaluatedNotification = EvaluatedNotification {
188 notification :: Notification,
189 reportRelation :: Either RelationalError Relation
190 }
191 deriving(Binary, Eq, Show, Generic)
194-- | Create a 'NodeId' for use in connecting to a remote server using distributed-process.
195createNodeId :: Hostname -> Port -> NodeId
196createNodeId host port = NodeId $ encodeEndPointAddress host (show port) 1
198-- | Use this for connecting to remote servers on the default port.
199defaultServerPort :: Port
200defaultServerPort = 6543
202-- | Use this for connecting to remote servers with the default database name.
203defaultDatabaseName :: DatabaseName
204defaultDatabaseName = "base"
206-- | Use this for connecting to remote servers with the default head name.
207defaultHeadName :: HeadName
208defaultHeadName = "master"
210-- | 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.
211defaultRemoteConnectionInfo :: ConnectionInfo
212defaultRemoteConnectionInfo = RemoteProcessConnectionInfo defaultDatabaseName (createNodeId "127.0.0.1" defaultServerPort) emptyNotificationCallback
214-- | The 'Connection' represents either local or remote access to a database. All operations flow through the connection.
215type ClientNodes = STMSet.Set ProcessId
217type TransactionGraphLockHandle = Handle
219-- internal structure specific to in-process connections
220data InProcessConnectionConf = InProcessConnectionConf {
221 ipPersistenceStrategy :: PersistenceStrategy,
222 ipClientNodes :: ClientNodes,
223 ipSessions :: Sessions,
224 ipTransactionGraph :: TVar TransactionGraph,
225 ipScriptSession :: Maybe ScriptSession,
226 ipLocalNode :: LocalNode,
227 ipTransport :: Transport, -- we hold onto this so that we can close it gracefully
228 ipLocks :: Maybe (TransactionGraphLockHandle, MVar LockFileHash) -- nothing when NoPersistence
229 }
231data RemoteProcessConnectionConf = RemoteProcessConnectionConf {
232 rLocalNode :: LocalNode,
233 rProcessId :: ProcessId, --remote processId
234 rTransport :: Transport --the TCP socket transport
235 }
237data Connection = InProcessConnection InProcessConnectionConf |
238 RemoteProcessConnection RemoteProcessConnectionConf
240-- | There are several reasons why a connection can fail.
241data ConnectionError = SetupDatabaseDirectoryError PersistenceError |
242 IOExceptionError IOException |
243 NoSuchDatabaseByNameError DatabaseName |
244 LoginError
245 deriving (Show, Eq, Generic)
247remoteDBLookupName :: DatabaseName -> String
248remoteDBLookupName = (++) "db-"
250createLocalNode :: IO (LocalNode, Transport)
251createLocalNode = do
252 eLocalTransport <- createTransport "127.0.0.1" "0" defaultTCPParameters
253 case eLocalTransport of
254 Left err -> error ("failed to create transport: " ++ show err)
255 Right localTransport -> do
256 localNode <- newLocalNode localTransport initRemoteTable
257 pure (localNode, localTransport)
259notificationListener :: NotificationCallback -> Process ()
260notificationListener callback = do
261 --pid <- getSelfPid
262 --liftIO $ putStrLn $ "LISTENER THREAD START " ++ show pid
263 _ <- forever $
264 receiveWait [
265 match (\(NotificationMessage eNots) ->
266 --say $ "NOTIFICATION: " ++ show eNots
267 -- 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
268 liftIO $ mapM_ (uncurry callback) (M.toList eNots)
269 )
270 ]
271 --say "LISTENER THREAD EXIT"
272 pure ()
274startNotificationListener :: LocalNode -> NotificationCallback -> IO ProcessId
275startNotificationListener localNode callback = forkProcess localNode (notificationListener callback)
277createScriptSession :: [String] -> IO (Maybe ScriptSession)
278createScriptSession ghcPkgPaths = do
279 eScriptSession <- initScriptSession ghcPkgPaths
280 case eScriptSession of
281 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
282 Right s -> pure (Just s)
284-- | To create a 'Connection' to a remote or local database, create a 'ConnectionInfo' and call 'connectProjectM36'.
285connectProjectM36 :: ConnectionInfo -> IO (Either ConnectionError Connection)
286--create a new in-memory database/transaction graph
287connectProjectM36 (InProcessConnectionInfo strat notificationCallback ghcPkgPaths) = do
288 freshId <- nextRandom
289 stamp <- getCurrentTime
290 let bootstrapContext = basicDatabaseContext
291 freshGraph = bootstrapTransactionGraph stamp freshId bootstrapContext
292 case strat of
293 --create date examples graph for now- probably should be empty context in the future
294 NoPersistence -> do
295 graphTvar <- newTVarIO freshGraph
296 clientNodes <- STMSet.newIO
297 sessions <- STMMap.newIO
298 (localNode, transport) <- createLocalNode
299 notificationPid <- startNotificationListener localNode notificationCallback
300 mScriptSession <- createScriptSession ghcPkgPaths
302 let conn = InProcessConnection InProcessConnectionConf {
303 ipPersistenceStrategy = strat,
304 ipClientNodes = clientNodes,
305 ipSessions = sessions,
306 ipTransactionGraph = graphTvar,
307 ipScriptSession = mScriptSession,
308 ipLocalNode = localNode,
309 ipTransport = transport,
310 ipLocks = Nothing}
311 addClientNode conn notificationPid
312 pure (Right conn)
313 MinimalPersistence dbdir -> connectPersistentProjectM36 strat NoDiskSync dbdir freshGraph notificationCallback ghcPkgPaths
314 CrashSafePersistence dbdir -> connectPersistentProjectM36 strat FsyncDiskSync dbdir freshGraph notificationCallback ghcPkgPaths
316connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId notificationCallback) = do
317 connStatus <- newEmptyMVar
318 (localNode, transport) <- createLocalNode
319 let dbName = remoteDBLookupName databaseName
320 --putStrLn $ "Connecting to " ++ show serverNodeId ++ " " ++ dbName
321 notificationListenerPid <- startNotificationListener localNode notificationCallback
322 runProcess localNode $ do
323 --even a failed connection retains an open socket!
324 mServerProcessId <- whereisRemote serverNodeId dbName
325 case mServerProcessId of
326 Nothing -> liftIO $ putMVar connStatus $ Left (NoSuchDatabaseByNameError databaseName)
327 Just serverProcessId -> do
328 loginConfirmation <- safeLogin (Login notificationListenerPid) serverProcessId
329 if not loginConfirmation then
330 liftIO $ putMVar connStatus (Left LoginError)
331 else
332 liftIO $ putMVar connStatus (Right $ RemoteProcessConnection RemoteProcessConnectionConf {rLocalNode = localNode, rProcessId = serverProcessId, rTransport = transport})
333 status <- takeMVar connStatus
334 pure status
336connectPersistentProjectM36 :: PersistenceStrategy ->
337 DiskSync ->
338 FilePath ->
339 TransactionGraph ->
340 NotificationCallback ->
341 [GhcPkgPath] ->
342 IO (Either ConnectionError Connection)
343connectPersistentProjectM36 strat sync dbdir freshGraph notificationCallback ghcPkgPaths = do
344 err <- setupDatabaseDir sync dbdir freshGraph
345 case err of
346 Left err' -> return $ Left (SetupDatabaseDirectoryError err')
347 Right (lockFileH, digest) -> do
348 mScriptSession <- createScriptSession ghcPkgPaths
349 graph <- transactionGraphLoad dbdir emptyTransactionGraph mScriptSession
350 case graph of
351 Left err' -> return $ Left (SetupDatabaseDirectoryError err')
352 Right graph' -> do
353 tvarGraph <- newTVarIO graph'
354 sessions <- STMMap.newIO
355 clientNodes <- STMSet.newIO
356 (localNode, transport) <- createLocalNode
357 lockMVar <- newMVar digest
358 let conn = InProcessConnection InProcessConnectionConf {
359 ipPersistenceStrategy = strat,
360 ipClientNodes = clientNodes,
361 ipSessions = sessions,
362 ipTransactionGraph = tvarGraph,
363 ipScriptSession = mScriptSession,
364 ipLocalNode = localNode,
365 ipTransport = transport,
366 ipLocks = Just (lockFileH, lockMVar)
367 }
369 notificationPid <- startNotificationListener localNode notificationCallback
370 addClientNode conn notificationPid
371 pure (Right conn)
373-- | Create a new session at the transaction id and return the session's Id.
374createSessionAtCommit :: Connection -> TransactionId -> IO (Either RelationalError SessionId)
375createSessionAtCommit conn@(InProcessConnection _) commitId = do
376 newSessionId <- nextRandom
377 atomically $ createSessionAtCommit_ commitId newSessionId conn
378createSessionAtCommit conn@(RemoteProcessConnection _) uuid = remoteCall conn (CreateSessionAtCommit uuid)
380createSessionAtCommit_ :: TransactionId -> SessionId -> Connection -> STM (Either RelationalError SessionId)
381createSessionAtCommit_ commitId newSessionId (InProcessConnection conf) = do
382 let sessions = ipSessions conf
383 graphTvar = ipTransactionGraph conf
384 graph <- readTVar graphTvar
385 case transactionForId commitId graph of
386 Left err -> pure (Left err)
387 Right transaction -> do
388 let freshDiscon = DisconnectedTransaction commitId (Trans.schemas transaction) False
389 keyDuplication <- STMMap.lookup newSessionId sessions
390 case keyDuplication of
391 Just _ -> pure $ Left (SessionIdInUseError newSessionId)
392 Nothing -> do
393 STMMap.insert (Session freshDiscon defaultSchemaName) newSessionId sessions
394 pure $ Right newSessionId
395createSessionAtCommit_ _ _ (RemoteProcessConnection _) = error "createSessionAtCommit_ called on remote connection"
397-- | 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.
398createSessionAtHead :: Connection -> HeadName -> IO (Either RelationalError SessionId)
399createSessionAtHead conn@(InProcessConnection conf) headn = do
400 let graphTvar = ipTransactionGraph conf
401 newSessionId <- nextRandom
402 atomically $ do
403 graph <- readTVar graphTvar
404 case transactionForHead headn graph of
405 Nothing -> pure $ Left (NoSuchHeadNameError headn)
406 Just trans -> createSessionAtCommit_ (transactionId trans) newSessionId conn
407createSessionAtHead conn@(RemoteProcessConnection _) headn = remoteCall conn (CreateSessionAtHead headn)
409-- | Used internally for server connections to keep track of remote nodes for the purpose of sending notifications later.
410addClientNode :: Connection -> ProcessId -> IO ()
411addClientNode (RemoteProcessConnection _) _ = error "addClientNode called on remote connection"
412addClientNode (InProcessConnection conf) newProcessId = atomically (STMSet.insert newProcessId (ipClientNodes conf))
414-- | Discards a session, eliminating any uncommitted changes present in the session.
415closeSession :: SessionId -> Connection -> IO ()
416closeSession sessionId (InProcessConnection conf) =
417 atomically $ STMMap.delete sessionId (ipSessions conf)
418closeSession sessionId conn@(RemoteProcessConnection _) = remoteCall conn (CloseSession sessionId)
419-- | 'close' cleans up the database access connection and closes any relevant sockets.
420close :: Connection -> IO ()
421close (InProcessConnection conf) = do
422 atomically $ do
423 let sessions = ipSessions conf
424 STMMap.deleteAll sessions
425 pure ()
426 closeLocalNode (ipLocalNode conf)
427 closeTransport (ipTransport conf)
429close conn@(RemoteProcessConnection conf) = do
430 _ <- remoteCall conn Logout :: IO Bool
431 closeLocalNode (rLocalNode conf)
432 closeTransport (rTransport conf)
434--used only by the server EntryPoints
435closeRemote_ :: Connection -> IO ()
436closeRemote_ (InProcessConnection _) = error "invalid call of closeRemote_ on InProcessConnection"
437closeRemote_ (RemoteProcessConnection conf) = runProcess (rLocalNode conf) (reconnect (rProcessId conf))
439 --we need to actually close the localNode's connection to the remote
440--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
441--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.
443excEither :: IO (Either RelationalError a) -> IO (Either RelationalError a)
444excEither = handle handler
445 where
446 handler exc | Just (_ :: AsyncException) <- fromException exc = throwIO exc
447 | otherwise = pure (Left (UnhandledExceptionError (show exc)))
449runProcessResult :: LocalNode -> Process a -> IO a
450runProcessResult localNode proc = do
451 ret <- newEmptyMVar
452 runProcess localNode $ do
453 val <- proc
454 liftIO $ putMVar ret val
455 takeMVar ret
457safeLogin :: Login -> ProcessId -> Process Bool
458safeLogin login procId = do
459 ret <- call procId login
460 case ret of
461 Left (_ :: ServerError) -> pure False
462 Right val -> pure val
464remoteCall :: (Serializable a, Serializable b) => Connection -> a -> IO b
465remoteCall (InProcessConnection _ ) _ = error "remoteCall called on local connection"
466remoteCall (RemoteProcessConnection conf) arg = runProcessResult localNode $ do
467 ret <- safeCall serverProcessId arg
468 case ret of
469 Left err -> error ("server died: " ++ show err)
470 Right ret' -> case ret' of
471 Left RequestTimeoutError -> liftIO (throwIO RequestTimeoutException)
472 Left (ProcessDiedError _) -> liftIO (throwIO RemoteProcessDiedException)
473 Right val -> pure val
474 where
475 localNode = rLocalNode conf
476 serverProcessId = rProcessId conf
478sessionForSessionId :: SessionId -> Sessions -> STM (Either RelationalError Session)
479sessionForSessionId sessionId sessions = do
480 maybeSession <- STMMap.lookup sessionId sessions
481 pure $ maybe (Left $ NoSuchSessionError sessionId) Right maybeSession
483schemaForSessionId :: Session -> STM (Either RelationalError Schema)
484schemaForSessionId session = do
485 let sname = schemaName session
486 if sname == defaultSchemaName then
487 pure (Right (Schema [])) -- the main schema includes no transformations (but neither do empty schemas :/ )
488 else
489 case M.lookup sname (subschemas session) of
490 Nothing -> pure (Left (SubschemaNameNotInUseError sname))
491 Just schema -> pure (Right schema)
493sessionAndSchema :: SessionId -> Sessions -> STM (Either RelationalError (Session, Schema))
494sessionAndSchema sessionId sessions = do
495 eSession <- sessionForSessionId sessionId sessions
496 case eSession of
497 Left err -> pure (Left err)
498 Right session -> do
499 eSchema <- schemaForSessionId session
500 case eSchema of
501 Left err -> pure (Left err)
502 Right schema -> pure (Right (session, schema))
504-- | Returns the name of the currently selected isomorphic schema.
505currentSchemaName :: SessionId -> Connection -> IO (Either RelationalError SchemaName)
506currentSchemaName sessionId (InProcessConnection conf) = atomically $ do
507 let sessions = ipSessions conf
508 eSession <- sessionForSessionId sessionId sessions
509 case eSession of
510 Left err -> pure (Left err)
511 Right session -> pure (Right (Sess.schemaName session))
512currentSchemaName sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveCurrentSchemaName sessionId)
514-- | Switch to the named isomorphic schema.
515setCurrentSchemaName :: SessionId -> Connection -> SchemaName -> IO (Either RelationalError ())
516setCurrentSchemaName sessionId (InProcessConnection conf) sname = atomically $ do
517 let sessions = ipSessions conf
518 eSession <- sessionForSessionId sessionId sessions
519 case eSession of
520 Left err -> pure (Left err)
521 Right session -> case Sess.setSchemaName sname session of
522 Left err -> pure (Left err)
523 Right newSession -> STMMap.insert newSession sessionId sessions >> pure (Right ())
524setCurrentSchemaName sessionId conn@(RemoteProcessConnection _) sname = remoteCall conn (ExecuteSetCurrentSchema sessionId sname)
526-- | Execute a relational expression in the context of the session and connection. Relational expressions are queries and therefore cannot alter the database.
527executeRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)
528executeRelationalExpr sessionId (InProcessConnection conf) expr = excEither $ atomically $ do
529 let sessions = ipSessions conf
530 eSession <- sessionAndSchema sessionId sessions
531 case eSession of
532 Left err -> pure $ Left err
533 Right (session, schema) -> do
534 let expr' = if schemaName session /= defaultSchemaName then
535 Schema.processRelationalExprInSchema schema expr
536 else
537 Right expr
538 case expr' of
539 Left err -> pure (Left err)
540 Right expr'' -> case runReader (RE.evalRelationalExpr expr'') (RE.mkRelationalExprState (Sess.concreteDatabaseContext session)) of
541 Left err -> pure (Left err)
542 Right rel -> pure (force (Right rel)) -- this is necessary so that any undefined/error exceptions are spit out here
543executeRelationalExpr sessionId conn@(RemoteProcessConnection _) relExpr = remoteCall conn (ExecuteRelationalExpr sessionId relExpr)
545-- | 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.
546executeDatabaseContextExpr :: SessionId -> Connection -> DatabaseContextExpr -> IO (Either RelationalError ())
547executeDatabaseContextExpr sessionId (InProcessConnection conf) expr = excEither $ atomically $ do
548 let sessions = ipSessions conf
549 eSession <- sessionAndSchema sessionId sessions
550 case eSession of
551 Left err -> pure (Left err)
552 Right (session, schema) -> do
553 let expr' = if schemaName session == defaultSchemaName then
554 Right expr
555 else
556 Schema.processDatabaseContextExprInSchema schema expr
557 case expr' of
558 Left err -> pure (Left err)
559 Right expr'' -> case runState (RE.evalDatabaseContextExpr expr'') (RE.freshDatabaseState (Sess.concreteDatabaseContext session)) of
560 (Just err,_) -> pure (Left err)
561 (Nothing, (_,_,False)) -> pure (Right ()) --optimization- if nothing was dirtied, nothing to do
562 (Nothing, (!context',_,True)) -> do
563 let newDiscon = DisconnectedTransaction (Sess.parentId session) newSchemas True
564 newSubschemas = Schema.processDatabaseContextExprSchemasUpdate (Sess.subschemas session) expr
565 newSchemas = Schemas context' newSubschemas
566 newSession = Session newDiscon (Sess.schemaName session)
567 STMMap.insert newSession sessionId sessions
568 pure (Right ())
569executeDatabaseContextExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (ExecuteDatabaseContextExpr sessionId dbExpr)
571-- | 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).
572autoMergeToHead :: SessionId -> Connection -> MergeStrategy -> HeadName -> IO (Either RelationalError ())
573autoMergeToHead sessionId (InProcessConnection conf) strat headName' = do
574 let sessions = ipSessions conf
575 id1 <- nextRandom
576 id2 <- nextRandom
577 id3 <- nextRandom
578 stamp <- getCurrentTime
579 commitLock_ sessionId conf $ \graph -> do
580 eSession <- sessionForSessionId sessionId sessions
581 case eSession of
582 Left err -> pure (Left err)
583 Right session ->
584 case Graph.autoMergeToHead stamp (id1, id2, id3) (Sess.disconnectedTransaction session) headName' strat graph of
585 Left err -> pure (Left err)
586 Right (discon', graph') ->
587 pure (Right (discon', graph', [id1, id2, id3]))
588autoMergeToHead sessionId conn@(RemoteProcessConnection _) strat headName' = remoteCall conn (ExecuteAutoMergeToHead sessionId strat headName')
590-- | Execute a database context IO-monad-based expression for the given session and connection. `DatabaseContextIOExpr`s modify the DatabaseContext but cannot be purely implemented.
591--this is almost completely identical to executeDatabaseContextExpr above
592executeDatabaseContextIOExpr :: SessionId -> Connection -> DatabaseContextIOExpr -> IO (Either RelationalError ())
593executeDatabaseContextIOExpr sessionId (InProcessConnection conf) expr = excEither $ do
594 let sessions = ipSessions conf
595 scriptSession = ipScriptSession conf
596 eSession <- atomically $ sessionForSessionId sessionId sessions --potentially race condition due to interleaved IO?
597 case eSession of
598 Left err -> pure (Left err)
599 Right session -> do
600 res <- RE.evalDatabaseContextIOExpr scriptSession (Sess.concreteDatabaseContext session) expr
601 case res of
602 Left err -> pure (Left err)
603 Right context' -> do
604 let newDiscon = DisconnectedTransaction (Sess.parentId session) newSchemas True
605 newSchemas = Schemas context' (Sess.subschemas session)
606 newSession = Session newDiscon (Sess.schemaName session)
607 atomically $ STMMap.insert newSession sessionId sessions
608 pure (Right ())
609executeDatabaseContextIOExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (ExecuteDatabaseContextIOExpr sessionId dbExpr)
611{-
612executeGraphExprSTM_ :: TransactionId -> SessionId -> Session -> Sessions -> TransactionGraphOperator -> TransactionGraph -> TVar TransactionGraph -> STM (Either RelationalError (TransactionGraph, DisconnectedTransaction)
613executeGraphExprSTM_ freshId sessionId session sessions graphExpr graph graphTVar= do
614 case evalGraphOp freshId (Sess.disconnectedTransaction session) graph graphExpr of
615 Left err -> do
616 when updateGraphOnError (writeTVar graphTVar graph)
617 pure $ Left err
618 Right (discon', graph') -> do
619 writeTVar graphTVar graph'
620 let newSession = Session discon' (Sess.schemaName session)
621 STMMap.insert newSession sessionId sessions
622 pure $ Right graph'
623-}
625-- process notifications for commits
626executeCommitExprSTM_ :: DatabaseContext -> DatabaseContext -> ClientNodes -> STM (EvaluatedNotifications, ClientNodes)
627executeCommitExprSTM_ oldContext newContext nodes = do
628 let nots = notifications oldContext
629 fireNots = notificationChanges nots oldContext newContext
630 evaldNots = M.map mkEvaldNot fireNots
631 mkEvaldNot notif = EvaluatedNotification { notification = notif,
632 reportRelation = runReader (RE.evalRelationalExpr (reportExpr notif)) (RE.mkRelationalExprState oldContext) }
633 pure (evaldNots, nodes)
635-- | Execute a transaction graph expression in the context of the session and connection. Transaction graph operators modify the transaction graph state.
637-- OPTIMIZATION OPPORTUNITY: no locks are required to write new transaction data, only to update the transaction graph id file
638-- if writing data is re-entrant, we may be able to use unsafeIOtoSTM
639-- perhaps keep hash of data file instead of checking if our head was updated on every write
640executeGraphExpr :: SessionId -> Connection -> TransactionGraphOperator -> IO (Either RelationalError ())
641executeGraphExpr sessionId (InProcessConnection conf) graphExpr = excEither $ do
642 let sessions = ipSessions conf
643 freshId <- nextRandom
644 stamp <- getCurrentTime
645 commitLock_ sessionId conf $ \updatedGraph -> do
646 eSession <- sessionForSessionId sessionId sessions
647 case eSession of
648 Left err -> pure (Left err)
649 Right session -> do
650 let discon = Sess.disconnectedTransaction session
651 case evalGraphOp stamp freshId discon updatedGraph graphExpr of
652 Left err -> pure (Left err)
653 Right (discon', graph') -> do
654 --if freshId appears in the graph, then we need to pass it on
655 let transIds = [freshId | isRight (transactionForId freshId graph')]
656 pure (Right (discon', graph', transIds))
657{-
658executeGraphExpr sessionId (InProcessConnection conf) graphExpr = excEither $ do
659 let strat = ipPersistenceStrategy conf
660 clientNodes = ipClientNodes conf
661 sessions = ipSessions conf
662 graphTvar = ipTransactionGraph conf
663 mLockFileH = ipLocks conf
664 lockHandler body = case graphExpr of
665 Commit -> case mLockFileH of
666 Nothing -> body False
667 Just (lockFileH, lockMVar) -> let acquireLocks = do
668 lastWrittenDigest <- takeMVar lockMVar
669 lockFile lockFileH WriteLock
670 latestDigest <- readGraphTransactionIdDigest strat
671 pure (latestDigest /= lastWrittenDigest)
673 releaseLocks _ = do
674 --still holding the lock- get the latest digest
675 gDigest <- readGraphTransactionIdDigest strat
676 unlockFile lockFileH
677 putMVar lockMVar gDigest
678 in bracket acquireLocks releaseLocks body
679 _ -> body False
680 freshId <- nextRandom
681 lockHandler $ \dbWrittenByOtherProcess -> do
682 --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
683 --read transaction data and compare to existing graph
684 --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)
685 manip <- atomically $ do
686 eSession <- sessionForSessionId sessionId sessions
687 --handle graph update by other process
688 oldGraph <- readTVar graphTvar
689 case eSession of
690 Left err -> pure (Left err)
691 Right session -> do
692 let mScriptSession = ipScriptSession conf
693 dbdir = case strat of
694 MinimalPersistence x -> x
695 CrashSafePersistence x -> x
696 _ -> error "accessing dbdir on non-persisted connection"
697 eRefreshedGraph <- if dbWrittenByOtherProcess then
698 unsafeIOToSTM (transactionGraphLoad dbdir oldGraph mScriptSession)
699 else
700 pure (Right oldGraph)
701 case eRefreshedGraph of
702 Left err -> pure (Left (DatabaseLoadError err))
703 Right refreshedGraph -> do
704 --snip it
705 eGraph <- executeGraphExprSTM_ dbWrittenByOtherProcess freshId sessionId session sessions graphExpr refreshedGraph graphTvar
706 --snip it
707 case eGraph of
708 Left err -> pure (Left err)
709 Right newGraph -> do
710 --handle commit
711 if isCommit graphExpr then do
712 case transactionForId (Sess.parentId session) oldGraph of
713 Left err -> pure $ Left err
714 Right previousTrans -> do
715 (evaldNots, nodes) <- executeCommitExprSTM_ (Trans.concreteDatabaseContext previousTrans) (Sess.concreteDatabaseContext session) clientNodes
716 nodesToNotify <- toList (STMSet.stream nodes)
717 pure $ Right (evaldNots, nodesToNotify, newGraph)
718 else
719 pure $ Right (M.empty, [], newGraph)
720 case manip of
721 Left err -> return $ Just err
722 Right (notsToFire, nodesToNotify, newGraph) -> do
723 --update filesystem database, if necessary
724 processTransactionGraphPersistence strat newGraph
725 sendNotifications nodesToNotify (ipLocalNode conf) notsToFire
726 pure Nothing
727-}
728executeGraphExpr sessionId conn@(RemoteProcessConnection _) graphExpr = remoteCall conn (ExecuteGraphExpr sessionId graphExpr)
730-- | A trans-graph expression is a relational query executed against the entirety of a transaction graph.
731executeTransGraphRelationalExpr :: SessionId -> Connection -> TransGraphRelationalExpr -> IO (Either RelationalError Relation)
732executeTransGraphRelationalExpr _ (InProcessConnection conf) tgraphExpr = excEither . atomically $ do
733 let graphTvar = ipTransactionGraph conf
734 graph <- readTVar graphTvar
735 case evalTransGraphRelationalExpr tgraphExpr graph of
736 Left err -> pure (Left err)
737 Right relExpr -> case runReader (RE.evalRelationalExpr relExpr) (RE.mkRelationalExprState DBC.empty) of
738 Left err -> pure (Left err)
739 Right rel -> pure (force (Right rel))
740executeTransGraphRelationalExpr sessionId conn@(RemoteProcessConnection _) tgraphExpr = remoteCall conn (ExecuteTransGraphRelationalExpr sessionId tgraphExpr)
742-- | Schema expressions manipulate the isomorphic schemas for the current 'DatabaseContext'.
743executeSchemaExpr :: SessionId -> Connection -> Schema.SchemaExpr -> IO (Either RelationalError ())
744executeSchemaExpr sessionId (InProcessConnection conf) schemaExpr = atomically $ do
745 let sessions = ipSessions conf
746 eSession <- sessionAndSchema sessionId sessions
747 case eSession of
748 Left err -> pure (Left err)
749 Right (session, _) -> do
750 let subschemas' = subschemas session
751 case Schema.evalSchemaExpr schemaExpr (Sess.concreteDatabaseContext session) subschemas' of
752 Left err -> pure (Left err)
753 Right (newSubschemas, newContext) -> do
754 --hm- maybe we should start using lenses
755 let discon = Sess.disconnectedTransaction session
756 newSchemas = Schemas newContext newSubschemas
757 newSession = Session (DisconnectedTransaction (Discon.parentId discon) newSchemas True) (Sess.schemaName session)
758 STMMap.insert newSession sessionId sessions
759 pure (Right ())
760executeSchemaExpr sessionId conn@(RemoteProcessConnection _) schemaExpr = remoteCall conn (ExecuteSchemaExpr sessionId schemaExpr)
762-- | 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.
763commit :: SessionId -> Connection -> IO (Either RelationalError ())
764commit sessionId conn@(InProcessConnection _) = executeGraphExpr sessionId conn Commit
765commit sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteGraphExpr sessionId Commit)
767sendNotifications :: [ProcessId] -> LocalNode -> EvaluatedNotifications -> IO ()
768sendNotifications pids localNode nots = mapM_ sendNots pids
769 where
770 sendNots remoteClientPid =
771 unless (M.null nots) $ runProcess localNode $ send remoteClientPid (NotificationMessage nots)
773-- | 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.
774rollback :: SessionId -> Connection -> IO (Either RelationalError ())
775rollback sessionId conn@(InProcessConnection _) = executeGraphExpr sessionId conn Rollback
776rollback sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteGraphExpr sessionId Rollback)
778-- | Write the transaction graph to disk. This function can be used to incrementally write new transactions to disk.
779processTransactionGraphPersistence :: PersistenceStrategy -> [TransactionId] -> TransactionGraph -> IO ()
780processTransactionGraphPersistence NoPersistence _ _ = pure ()
781processTransactionGraphPersistence (MinimalPersistence dbdir) transIds graph = transactionGraphPersist NoDiskSync dbdir transIds graph >> pure ()
782processTransactionGraphPersistence (CrashSafePersistence dbdir) transIds graph = transactionGraphPersist FsyncDiskSync dbdir transIds graph >> pure ()
784readGraphTransactionIdDigest :: PersistenceStrategy -> IO LockFileHash
785readGraphTransactionIdDigest NoPersistence = error "attempt to read digest from transaction log without persistence enabled"
786readGraphTransactionIdDigest (MinimalPersistence dbdir) = readGraphTransactionIdFileDigest dbdir
787readGraphTransactionIdDigest (CrashSafePersistence dbdir) = readGraphTransactionIdFileDigest dbdir
789-- | 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.
790typeForRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)
791typeForRelationalExpr sessionId conn@(InProcessConnection _) relExpr = atomically $ typeForRelationalExprSTM sessionId conn relExpr
792typeForRelationalExpr sessionId conn@(RemoteProcessConnection _) relExpr = remoteCall conn (ExecuteTypeForRelationalExpr sessionId relExpr)
794typeForRelationalExprSTM :: SessionId -> Connection -> RelationalExpr -> STM (Either RelationalError Relation)
795typeForRelationalExprSTM sessionId (InProcessConnection conf) relExpr = do
796 let sessions = ipSessions conf
797 eSession <- sessionAndSchema sessionId sessions
798 case eSession of
799 Left err -> pure $ Left err
800 Right (session, schema) -> do
801 let processed = if schemaName session == defaultSchemaName then
802 Right relExpr
803 else
804 Schema.processRelationalExprInSchema schema relExpr
805 case processed of
806 Left err -> pure (Left err)
807 Right relExpr' -> pure $ runReader (RE.typeForRelationalExpr relExpr') (RE.mkRelationalExprState (Sess.concreteDatabaseContext session))
809typeForRelationalExprSTM _ _ _ = error "typeForRelationalExprSTM called on non-local connection"
811-- | Return a 'Map' of the database's constraints at the context of the session and connection.
812inclusionDependencies :: SessionId -> Connection -> IO (Either RelationalError InclusionDependencies)
813inclusionDependencies sessionId (InProcessConnection conf) = do
814 let sessions = ipSessions conf
815 atomically $ do
816 eSession <- sessionAndSchema sessionId sessions
817 case eSession of
818 Left err -> pure $ Left err
819 Right (session, schema) -> do
820 let context = Sess.concreteDatabaseContext session
821 if schemaName session == defaultSchemaName then
822 pure $ Right (B.inclusionDependencies context)
823 else
824 pure (Schema.inclusionDependenciesInSchema schema (B.inclusionDependencies context))
826inclusionDependencies sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveInclusionDependencies sessionId)
828-- | 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.
829planForDatabaseContextExpr :: SessionId -> Connection -> DatabaseContextExpr -> IO (Either RelationalError DatabaseContextExpr)
830planForDatabaseContextExpr sessionId (InProcessConnection conf) dbExpr = do
831 let sessions = ipSessions conf
832 atomically $ do
833 eSession <- sessionAndSchema sessionId sessions
834 case eSession of
835 Left err -> pure $ Left err
836 Right (session, _) -> if schemaName session == defaultSchemaName then
837 pure $ evalState (applyStaticDatabaseOptimization dbExpr) (RE.freshDatabaseState (Sess.concreteDatabaseContext session))
838 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
839 pure (Right dbExpr)
841planForDatabaseContextExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (RetrievePlanForDatabaseContextExpr sessionId dbExpr)
843-- | Return a relation which represents the current state of the global transaction graph. The attributes are
844-- * current- boolean attribute representing whether or not the current session references this transaction
845-- * head- text attribute which is a non-empty 'HeadName' iff the transaction references a head.
846-- * id- id attribute of the transaction
847-- * parents- a relation-valued attribute which contains a relation of transaction ids which are parent transaction to the transaction
848transactionGraphAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
849transactionGraphAsRelation sessionId (InProcessConnection conf) = do
850 let sessions = ipSessions conf
851 tvar = ipTransactionGraph conf
852 atomically $ do
853 eSession <- sessionForSessionId sessionId sessions
854 case eSession of
855 Left err -> pure $ Left err
856 Right session -> do
857 graph <- readTVar tvar
858 pure $ graphAsRelation (Sess.disconnectedTransaction session) graph
860transactionGraphAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveTransactionGraph sessionId)
862-- | Returns the names and types of the relation variables in the current 'Session'.
863relationVariablesAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
864relationVariablesAsRelation sessionId (InProcessConnection conf) = do
865 let sessions = ipSessions conf
866 atomically $ do
867 eSession <- sessionAndSchema sessionId sessions
868 case eSession of
869 Left err -> pure (Left err)
870 Right (session, schema) -> do
871 let context = Sess.concreteDatabaseContext session
872 if Sess.schemaName session == defaultSchemaName then
873 pure $ R.relationVariablesAsRelation (relationVariables context)
874 else
875 case Schema.relationVariablesInSchema schema context of
876 Left err -> pure (Left err)
877 Right relvars -> pure $ R.relationVariablesAsRelation relvars
879relationVariablesAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveRelationVariableSummary sessionId)
881-- | Returns the transaction id for the connection's disconnected transaction committed parent transaction.
882headTransactionId :: SessionId -> Connection -> IO (Either RelationalError TransactionId)
883headTransactionId sessionId (InProcessConnection conf) = do
884 let sessions = ipSessions conf
885 atomically $ do
886 eSession <- sessionForSessionId sessionId sessions
887 case eSession of
888 Left err -> pure (Left err)
889 Right session -> pure $ Right (Sess.parentId session)
890headTransactionId sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveHeadTransactionId sessionId)
892headNameSTM_ :: SessionId -> Sessions -> TVar TransactionGraph -> STM (Either RelationalError HeadName)
893headNameSTM_ sessionId sessions graphTvar = do
894 graph <- readTVar graphTvar
895 eSession <- sessionForSessionId sessionId sessions
896 case eSession of
897 Left err -> pure (Left err)
898 Right session -> case transactionForId (Sess.parentId session) graph of
899 Left err -> pure (Left err)
900 Right parentTrans -> case headNameForTransaction parentTrans graph of
901 Nothing -> pure (Left UnknownHeadError)
902 Just headName' -> pure (Right headName')
904-- | Returns Just the name of the head of the current disconnected transaction or Nothing.
905headName :: SessionId -> Connection -> IO (Either RelationalError HeadName)
906headName sessionId (InProcessConnection conf) = do
907 let sessions = ipSessions conf
908 graphTvar = ipTransactionGraph conf
909 atomically (headNameSTM_ sessionId sessions graphTvar)
910headName sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteHeadName sessionId)
912-- | Returns a listing of all available atom types.
913atomTypesAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
914atomTypesAsRelation sessionId (InProcessConnection conf) = do
915 let sessions = ipSessions conf
916 atomically $ do
917 eSession <- sessionForSessionId sessionId sessions
918 case eSession of
919 Left err -> pure (Left err)
920 Right session ->
921 case typesAsRelation (typeConstructorMapping (Sess.concreteDatabaseContext session)) of
922 Left err -> pure (Left err)
923 Right rel -> pure (Right rel)
924atomTypesAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveAtomTypesAsRelation sessionId)
926disconnectedTransactionIsDirty :: SessionId -> Connection -> IO (Either RelationalError Bool)
927disconnectedTransactionIsDirty sessionId (InProcessConnection conf) = do
928 let sessions = ipSessions conf
929 atomically $ do
930 eSession <- sessionForSessionId sessionId sessions
931 case eSession of
932 Left err -> pure (Left err)
933 Right session ->
934 pure (Right (isDirty session))
935disconnectedTransactionIsDirty sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveSessionIsDirty sessionId)
937--used only for testing- we expect this to throw an exception
938callTestTimeout_ :: SessionId -> Connection -> IO Bool
939callTestTimeout_ _ (InProcessConnection _) = error "bad testing call"
940callTestTimeout_ sessionId conn@(RemoteProcessConnection _) = remoteCall conn (TestTimeout sessionId)
942--used in tests only
943transactionGraph_ :: Connection -> IO TransactionGraph
944transactionGraph_ (InProcessConnection conf) = atomically $ readTVar (ipTransactionGraph conf)
945transactionGraph_ _ = error "remote connection used"
947--used in tests only
948disconnectedTransaction_ :: SessionId -> Connection -> IO DisconnectedTransaction
949disconnectedTransaction_ sessionId (InProcessConnection conf) = do
950 let sessions = ipSessions conf
951 mSession <- atomically $ STMMap.lookup sessionId sessions
952 case mSession of
953 Nothing -> error "No such session"
954 Just (Sess.Session discon _) -> pure discon
955disconnectedTransaction_ _ _= error "remote connection used"
957-- wrap a graph evaluation in file locking
958commitLock_ :: SessionId ->
959 InProcessConnectionConf ->
960 (TransactionGraph ->
961 STM (Either RelationalError (DisconnectedTransaction, TransactionGraph, [TransactionId]))) ->
962 IO (Either RelationalError ())
963commitLock_ sessionId conf stmBlock = do
964 let sessions = ipSessions conf
965 strat = ipPersistenceStrategy conf
966 mScriptSession = ipScriptSession conf
967 graphTvar = ipTransactionGraph conf
968 clientNodes = ipClientNodes conf
969 mLockFileH = ipLocks conf
970 lockHandler body = case mLockFileH of
971 Nothing -> body False
972 Just (lockFileH, lockMVar) ->
973 let acquireLocks = do
974 lastWrittenDigest <- takeMVar lockMVar
975 lockFile lockFileH WriteLock
976 latestDigest <- readGraphTransactionIdDigest strat
977 pure (latestDigest /= lastWrittenDigest)
978 releaseLocks _ = do
979 --still holding the lock- get the latest digest
980 gDigest <- readGraphTransactionIdDigest strat
981 unlockFile lockFileH
982 putMVar lockMVar gDigest
983 in bracket acquireLocks releaseLocks body
984 manip <- lockHandler $ \dbWrittenByOtherProcess -> atomically $ do
985 eSession <- sessionForSessionId sessionId sessions
986 --handle graph update by other process
987 oldGraph <- readTVar graphTvar
988 case eSession of
989 Left err -> pure (Left err)
990 Right session -> do
991 let dbdir = case strat of
992 MinimalPersistence x -> x
993 CrashSafePersistence x -> x
994 _ -> error "accessing dbdir on non-persisted connection"
995 --this should also happen for non-commit expressions
996 eRefreshedGraph <- if dbWrittenByOtherProcess then
997 unsafeIOToSTM (transactionGraphLoad dbdir oldGraph mScriptSession)
998 else
999 pure (Right oldGraph)
1000 case eRefreshedGraph of
1001 Left err -> pure (Left (DatabaseLoadError err))
1002 Right refreshedGraph -> do
1003 eGraph <- stmBlock refreshedGraph
1004 case eGraph of
1005 Left err -> pure (Left err)
1006 Right (discon', graph', transactionIdsToPersist) -> do
1007 writeTVar graphTvar graph'
1008 let newSession = Session discon' (Sess.schemaName session)
1009 STMMap.insert newSession sessionId sessions
1010 case transactionForId (Sess.parentId session) oldGraph of
1011 Left err -> pure $ Left err
1012 Right previousTrans ->
1013 if not (Prelude.null transactionIdsToPersist) then do
1014 (evaldNots, nodes) <- executeCommitExprSTM_ (Trans.concreteDatabaseContext previousTrans) (Sess.concreteDatabaseContext session) clientNodes
1015 nodesToNotify <- toList (STMSet.stream nodes)
1016 pure $ Right (evaldNots, nodesToNotify, graph', transactionIdsToPersist)
1017 else pure (Right (M.empty, [], graph', []))
1019 --handle notification firing
1020 case manip of
1021 Left err -> pure (Left err)
1022 Right (notsToFire, nodesToNotify, newGraph, transactionIdsToPersist) -> do
1023 --update filesystem database, if necessary
1024 processTransactionGraphPersistence strat transactionIdsToPersist newGraph
1025 sendNotifications nodesToNotify (ipLocalNode conf) notsToFire
1026 pure (Right ())
1028writeDisconAndGraph_ :: TVar TransactionGraph -> SessionId -> Session -> Sessions -> DisconnectedTransaction -> TransactionGraph -> STM ()
1029writeDisconAndGraph_ graphTvar sessionId session sessions discon graph = do
1030 writeTVar graphTvar graph
1031 let newSession = Session discon (Sess.schemaName session)
1032 STMMap.insert newSession sessionId sessions
1035-- | Runs an IO monad, commits the result when the monad returns no errors, otherwise, rolls back the changes and the error.
1036withTransaction :: SessionId -> Connection -> IO (Either RelationalError a) -> IO (Either RelationalError ()) -> IO (Either RelationalError a)
1037withTransaction sessionId conn io successFunc = bracketOnError (pure ()) (const do_rollback) block
1038 where
1039 do_rollback = rollback sessionId conn
1040 block _ = do
1041 eErr <- io
1042 case eErr of
1043 Left err -> do
1044 _ <- do_rollback
1045 pure (Left err)
1046 Right val -> do
1047 eIsDirty <- disconnectedTransactionIsDirty sessionId conn
1048 case eIsDirty of
1049 Left err -> pure (Left err)
1050 Right dirty ->
1051 if dirty then do
1052 res <- successFunc
1053 case res of
1054 Left err -> pure (Left err)
1055 Right _ -> pure (Right val)
1056 else -- no updates executed, so don't create a commit
1057 pure (Right val)