git.iko.soy/project-m36
on 18f736a help sign in
project-m36/src/lib/ProjectM36/Client.hs
1084 lines 57 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 ProjectM36.Client.typeConstructorMapping,
28 planForDatabaseContextExpr,
29 currentSchemaName,
30 SchemaName,
31 HeadName,
32 setCurrentSchemaName,
33 transactionGraphAsRelation,
34 relationVariablesAsRelation,
35 disconnectedTransactionIsDirty,
36 headName,
37 remoteDBLookupName,
38 defaultServerPort,
39 headTransactionId,
40 defaultDatabaseName,
41 defaultRemoteConnectionInfo,
42 defaultHeadName,
43 PersistenceStrategy(..),
44 RelationalExpr,
45 RelationalExprBase(..),
46 DatabaseContextExpr(..),
47 DatabaseContextIOExpr(..),
48 Attribute(..),
49 MergeStrategy(..),
50 attributesFromList,
51 createNodeId,
52 createSessionAtCommit,
53 createSessionAtHead,
54 closeSession,
55 addClientNode,
56 callTestTimeout_,
57 RelationCardinality(..),
58 TransactionGraphOperator(..),
59 ProjectM36.Client.autoMergeToHead,
60 transactionGraph_,
61 disconnectedTransaction_,
62 TransGraphRelationalExpr,
63 TransactionIdLookup(..),
64 TransactionIdHeadBacktrack(..),
65 NodeId(..),
66 Atom(..),
67 Session,
68 SessionId,
69 NotificationCallback,
70 emptyNotificationCallback,
71 EvaluatedNotification(..),
72 atomTypesAsRelation,
73 AttributeExpr,
74 inclusionDependencyForKey,
75 databaseContextExprForUniqueKey,
76 databaseContextExprForForeignKey,
77 createScriptedAtomFunction,
78 AttributeExprBase(..),
79 TypeConstructor(..),
80 TypeConstructorDef(..),
81 DataConstructorDef(..),
82 AttributeNamesBase(..),
83 RelVarName,
84 IncDepName,
85 InclusionDependency(..),
86 AttributeName,
87 RelationalError(..),
88 RequestTimeoutException(..),
89 RemoteProcessDiedException(..),
90 AtomType(..),
91 Atomable(..),
92 TupleExprBase(..),
93 AtomExprBase(..),
94 RestrictionPredicateExprBase(..),
95 withTransaction
96 ) where
97import ProjectM36.Base hiding (inclusionDependencies) --defined in this module as well
98import qualified ProjectM36.Base as B
99import ProjectM36.Error
100import ProjectM36.Atomable
101import ProjectM36.AtomFunction
102import ProjectM36.StaticOptimizer
103import ProjectM36.Key
104import qualified ProjectM36.IsomorphicSchema as Schema
105import Control.Monad.State
106import Control.Monad.Trans.Reader
107import qualified ProjectM36.RelationalExpression as RE
108import ProjectM36.DatabaseContext (basicDatabaseContext)
109import qualified ProjectM36.TransactionGraph as Graph
110import ProjectM36.TransactionGraph
111import qualified ProjectM36.Transaction as Trans
112import ProjectM36.TransactionGraph.Persist
113import ProjectM36.Attribute
114import ProjectM36.TransGraphRelationalExpression (TransGraphRelationalExpr, evalTransGraphRelationalExpr)
115import ProjectM36.Persist (DiskSync(..))
116import ProjectM36.FileLock
117import ProjectM36.Notifications
118import ProjectM36.Server.RemoteCallTypes
119import qualified ProjectM36.DisconnectedTransaction as Discon
120import ProjectM36.Relation (typesAsRelation)
121import ProjectM36.ScriptSession (initScriptSession, ScriptSession)
122import qualified ProjectM36.Relation as R
123import qualified ProjectM36.DatabaseContext as DBC
124import Control.Exception.Base
125import GHC.Conc.Sync
127import Network.Transport (Transport(closeTransport))
128import Network.Transport.TCP (createTransport, defaultTCPParameters, encodeEndPointAddress)
129import Control.Distributed.Process.Node (newLocalNode, initRemoteTable, runProcess, LocalNode, forkProcess, closeLocalNode)
130import Control.Distributed.Process.Extras.Internal.Types (whereisRemote)
131import Control.Distributed.Process.ManagedProcess.Client (call, safeCall)
132import Data.Either (isRight)
133import Data.UUID.V4 (nextRandom)
134import Data.Word
135import Control.Distributed.Process (ProcessId, Process, receiveWait, send, match, NodeId(..), reconnect)
136import Control.Exception (IOException, handle, AsyncException, throwIO, fromException, Exception)
137import Control.Concurrent.MVar
138import qualified Data.Map as M
139import Control.Distributed.Process.Serializable (Serializable)
140import qualified STMContainers.Map as STMMap
141import qualified STMContainers.Set as STMSet
142import qualified ProjectM36.Session as Sess
143import ProjectM36.Session
144import ProjectM36.Sessions
145import ListT
146import Data.Binary (Binary)
147import GHC.Generics (Generic)
148import Control.DeepSeq (force)
149import System.IO
150import Data.Time.Clock
152type Hostname = String
154type Port = Word16
156type DatabaseName = String
158-- | 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.
159type NotificationCallback = NotificationName -> EvaluatedNotification -> IO ()
161-- | The empty notification callback ignores all callbacks.
162emptyNotificationCallback :: NotificationCallback
163emptyNotificationCallback _ _ = pure ()
165type GhcPkgPath = String
167data RemoteProcessDiedException = RemoteProcessDiedException
168 deriving (Show, Eq)
170instance Exception RemoteProcessDiedException
172data RequestTimeoutException = RequestTimeoutException
173 deriving (Show, Eq)
175instance Exception RequestTimeoutException
177-- | 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.
178data ConnectionInfo = InProcessConnectionInfo PersistenceStrategy NotificationCallback [GhcPkgPath]|
179 RemoteProcessConnectionInfo DatabaseName NodeId NotificationCallback
181type EvaluatedNotifications = M.Map NotificationName EvaluatedNotification
183-- | Used for callbacks from the server when monitored changes have been made.
184newtype NotificationMessage = NotificationMessage EvaluatedNotifications
185 deriving (Binary, Eq, Show, Generic)
187-- | When a notification is fired, the 'reportOldExpr' is evaluated in the commit's pre-change context while the 'reportNewExpr' is evaluated in the post-change context and they are returned along with the original notification.
188data EvaluatedNotification = EvaluatedNotification {
189 notification :: Notification,
190 reportOldRelation :: Either RelationalError Relation,
191 reportNewRelation :: Either RelationalError Relation
192 }
193 deriving(Binary, Eq, Show, Generic)
196-- | Create a 'NodeId' for use in connecting to a remote server using distributed-process.
197createNodeId :: Hostname -> Port -> NodeId
198createNodeId host port = NodeId $ encodeEndPointAddress host (show port) 1
200-- | Use this for connecting to remote servers on the default port.
201defaultServerPort :: Port
202defaultServerPort = 6543
204-- | Use this for connecting to remote servers with the default database name.
205defaultDatabaseName :: DatabaseName
206defaultDatabaseName = "base"
208-- | Use this for connecting to remote servers with the default head name.
209defaultHeadName :: HeadName
210defaultHeadName = "master"
212-- | 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.
213defaultRemoteConnectionInfo :: ConnectionInfo
214defaultRemoteConnectionInfo = RemoteProcessConnectionInfo defaultDatabaseName (createNodeId "127.0.0.1" defaultServerPort) emptyNotificationCallback
216-- | The 'Connection' represents either local or remote access to a database. All operations flow through the connection.
217type ClientNodes = STMSet.Set ProcessId
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 (LockFile, 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)
420-- | 'close' cleans up the database access connection and closes any relevant sockets.
421close :: Connection -> IO ()
422close (InProcessConnection conf) = do
423 atomically $ do
424 let sessions = ipSessions conf
425 STMMap.deleteAll sessions
426 pure ()
427 closeLocalNode (ipLocalNode conf)
428 closeTransport (ipTransport conf)
429 let mLocks = ipLocks conf
430 case mLocks of
431 Nothing -> pure ()
432 Just (lockFileH, _) -> closeLockFile lockFileH
434close conn@(RemoteProcessConnection conf) = do
435 _ <- remoteCall conn Logout :: IO Bool
436 closeLocalNode (rLocalNode conf)
437 closeTransport (rTransport conf)
439--used only by the server EntryPoints
440closeRemote_ :: Connection -> IO ()
441closeRemote_ (InProcessConnection _) = error "invalid call of closeRemote_ on InProcessConnection"
442closeRemote_ (RemoteProcessConnection conf) = runProcess (rLocalNode conf) (reconnect (rProcessId conf))
444 --we need to actually close the localNode's connection to the remote
445--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
446--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.
448excEither :: IO (Either RelationalError a) -> IO (Either RelationalError a)
449excEither = handle handler
450 where
451 handler exc | Just (_ :: AsyncException) <- fromException exc = throwIO exc
452 | otherwise = pure (Left (UnhandledExceptionError (show exc)))
454runProcessResult :: LocalNode -> Process a -> IO a
455runProcessResult localNode proc = do
456 ret <- newEmptyMVar
457 runProcess localNode $ do
458 val <- proc
459 liftIO $ putMVar ret val
460 takeMVar ret
462safeLogin :: Login -> ProcessId -> Process Bool
463safeLogin login procId = do
464 ret <- call procId login
465 case ret of
466 Left (_ :: ServerError) -> pure False
467 Right val -> pure val
469remoteCall :: (Serializable a, Serializable b) => Connection -> a -> IO b
470remoteCall (InProcessConnection _ ) _ = error "remoteCall called on local connection"
471remoteCall (RemoteProcessConnection conf) arg = runProcessResult localNode $ do
472 ret <- safeCall serverProcessId arg
473 case ret of
474 Left err -> error ("server died: " ++ show err)
475 Right ret' -> case ret' of
476 Left RequestTimeoutError -> liftIO (throwIO RequestTimeoutException)
477 Left (ProcessDiedError _) -> liftIO (throwIO RemoteProcessDiedException)
478 Right val -> pure val
479 where
480 localNode = rLocalNode conf
481 serverProcessId = rProcessId conf
483sessionForSessionId :: SessionId -> Sessions -> STM (Either RelationalError Session)
484sessionForSessionId sessionId sessions = do
485 maybeSession <- STMMap.lookup sessionId sessions
486 pure $ maybe (Left $ NoSuchSessionError sessionId) Right maybeSession
488schemaForSessionId :: Session -> STM (Either RelationalError Schema)
489schemaForSessionId session = do
490 let sname = schemaName session
491 if sname == defaultSchemaName then
492 pure (Right (Schema [])) -- the main schema includes no transformations (but neither do empty schemas :/ )
493 else
494 case M.lookup sname (subschemas session) of
495 Nothing -> pure (Left (SubschemaNameNotInUseError sname))
496 Just schema -> pure (Right schema)
498sessionAndSchema :: SessionId -> Sessions -> STM (Either RelationalError (Session, Schema))
499sessionAndSchema sessionId sessions = do
500 eSession <- sessionForSessionId sessionId sessions
501 case eSession of
502 Left err -> pure (Left err)
503 Right session -> do
504 eSchema <- schemaForSessionId session
505 case eSchema of
506 Left err -> pure (Left err)
507 Right schema -> pure (Right (session, schema))
509-- | Returns the name of the currently selected isomorphic schema.
510currentSchemaName :: SessionId -> Connection -> IO (Either RelationalError SchemaName)
511currentSchemaName sessionId (InProcessConnection conf) = atomically $ do
512 let sessions = ipSessions conf
513 eSession <- sessionForSessionId sessionId sessions
514 case eSession of
515 Left err -> pure (Left err)
516 Right session -> pure (Right (Sess.schemaName session))
517currentSchemaName sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveCurrentSchemaName sessionId)
519-- | Switch to the named isomorphic schema.
520setCurrentSchemaName :: SessionId -> Connection -> SchemaName -> IO (Either RelationalError ())
521setCurrentSchemaName sessionId (InProcessConnection conf) sname = atomically $ do
522 let sessions = ipSessions conf
523 eSession <- sessionForSessionId sessionId sessions
524 case eSession of
525 Left err -> pure (Left err)
526 Right session -> case Sess.setSchemaName sname session of
527 Left err -> pure (Left err)
528 Right newSession -> STMMap.insert newSession sessionId sessions >> pure (Right ())
529setCurrentSchemaName sessionId conn@(RemoteProcessConnection _) sname = remoteCall conn (ExecuteSetCurrentSchema sessionId sname)
531-- | Execute a relational expression in the context of the session and connection. Relational expressions are queries and therefore cannot alter the database.
532executeRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)
533executeRelationalExpr sessionId (InProcessConnection conf) expr = excEither $ atomically $ do
534 let sessions = ipSessions conf
535 eSession <- sessionAndSchema sessionId sessions
536 case eSession of
537 Left err -> pure $ Left err
538 Right (session, schema) -> do
539 let expr' = if schemaName session /= defaultSchemaName then
540 Schema.processRelationalExprInSchema schema expr
541 else
542 Right expr
543 case expr' of
544 Left err -> pure (Left err)
545 Right expr'' -> case optimizeRelationalExpr (Sess.concreteDatabaseContext session) expr'' of
546 Left err -> pure (Left err)
547 Right optExpr -> do
548 let evald = runReader (RE.evalRelationalExpr optExpr) (RE.mkRelationalExprState (Sess.concreteDatabaseContext session))
549 case evald of
550 Left err -> pure (Left err)
551 Right rel -> pure (force (Right rel)) -- this is necessary so that any undefined/error exceptions are spit out here
552executeRelationalExpr sessionId conn@(RemoteProcessConnection _) relExpr = remoteCall conn (ExecuteRelationalExpr sessionId relExpr)
554-- | 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.
555executeDatabaseContextExpr :: SessionId -> Connection -> DatabaseContextExpr -> IO (Either RelationalError ())
556executeDatabaseContextExpr sessionId (InProcessConnection conf) expr = excEither $ atomically $ do
557 let sessions = ipSessions conf
558 eSession <- sessionAndSchema sessionId sessions
559 case eSession of
560 Left err -> pure (Left err)
561 Right (session, schema) -> do
562 let expr' = if schemaName session == defaultSchemaName then
563 Right expr
564 else
565 Schema.processDatabaseContextExprInSchema schema expr
566 case expr' of
567 Left err -> pure (Left err)
568 Right expr'' -> case runState (do
569 eOptExpr <- applyStaticDatabaseOptimization expr''
570 case eOptExpr of
571 Left err -> pure (Left err)
572 Right optExpr ->
573 RE.evalDatabaseContextExpr optExpr) (RE.freshDatabaseState (Sess.concreteDatabaseContext session)) of
574 (Left err,_) -> pure (Left err)
575 (Right (), (_,_,False)) -> pure (Right ()) --optimization- if nothing was dirtied, nothing to do
576 (Right (), (!context',_,True)) -> do
577 let newDiscon = DisconnectedTransaction (Sess.parentId session) newSchemas True
578 newSubschemas = Schema.processDatabaseContextExprSchemasUpdate (Sess.subschemas session) expr
579 newSchemas = Schemas context' newSubschemas
580 newSession = Session newDiscon (Sess.schemaName session)
581 STMMap.insert newSession sessionId sessions
582 pure (Right ())
583executeDatabaseContextExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (ExecuteDatabaseContextExpr sessionId dbExpr)
585-- | 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).
586autoMergeToHead :: SessionId -> Connection -> MergeStrategy -> HeadName -> IO (Either RelationalError ())
587autoMergeToHead sessionId (InProcessConnection conf) strat headName' = do
588 let sessions = ipSessions conf
589 id1 <- nextRandom
590 id2 <- nextRandom
591 id3 <- nextRandom
592 stamp <- getCurrentTime
593 commitLock_ sessionId conf $ \graph -> do
594 eSession <- sessionForSessionId sessionId sessions
595 case eSession of
596 Left err -> pure (Left err)
597 Right session ->
598 case Graph.autoMergeToHead stamp (id1, id2, id3) (Sess.disconnectedTransaction session) headName' strat graph of
599 Left err -> pure (Left err)
600 Right (discon', graph') ->
601 pure (Right (discon', graph', [id1, id2, id3]))
602autoMergeToHead sessionId conn@(RemoteProcessConnection _) strat headName' = remoteCall conn (ExecuteAutoMergeToHead sessionId strat headName')
604-- | Execute a database context IO-monad-based expression for the given session and connection. `DatabaseContextIOExpr`s modify the DatabaseContext but cannot be purely implemented.
605--this is almost completely identical to executeDatabaseContextExpr above
606executeDatabaseContextIOExpr :: SessionId -> Connection -> DatabaseContextIOExpr -> IO (Either RelationalError ())
607executeDatabaseContextIOExpr sessionId (InProcessConnection conf) expr = excEither $ do
608 let sessions = ipSessions conf
609 scriptSession = ipScriptSession conf
610 eSession <- atomically $ sessionForSessionId sessionId sessions --potentially race condition due to interleaved IO?
611 case eSession of
612 Left err -> pure (Left err)
613 Right session -> do
614 res <- RE.evalDatabaseContextIOExpr scriptSession (Sess.concreteDatabaseContext session) expr
615 case res of
616 Left err -> pure (Left err)
617 Right context' -> do
618 let newDiscon = DisconnectedTransaction (Sess.parentId session) newSchemas True
619 newSchemas = Schemas context' (Sess.subschemas session)
620 newSession = Session newDiscon (Sess.schemaName session)
621 atomically $ STMMap.insert newSession sessionId sessions
622 pure (Right ())
623executeDatabaseContextIOExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (ExecuteDatabaseContextIOExpr sessionId dbExpr)
625{-
626executeGraphExprSTM_ :: TransactionId -> SessionId -> Session -> Sessions -> TransactionGraphOperator -> TransactionGraph -> TVar TransactionGraph -> STM (Either RelationalError (TransactionGraph, DisconnectedTransaction)
627executeGraphExprSTM_ freshId sessionId session sessions graphExpr graph graphTVar= do
628 case evalGraphOp freshId (Sess.disconnectedTransaction session) graph graphExpr of
629 Left err -> do
630 when updateGraphOnError (writeTVar graphTVar graph)
631 pure $ Left err
632 Right (discon', graph') -> do
633 writeTVar graphTVar graph'
634 let newSession = Session discon' (Sess.schemaName session)
635 STMMap.insert newSession sessionId sessions
636 pure $ Right graph'
637-}
639-- process notifications for commits
640executeCommitExprSTM_ :: DatabaseContext -> DatabaseContext -> ClientNodes -> STM (EvaluatedNotifications, ClientNodes)
641executeCommitExprSTM_ oldContext newContext nodes = do
642 let nots = notifications oldContext
643 fireNots = notificationChanges nots oldContext newContext
644 evaldNots = M.map mkEvaldNot fireNots
645 evalInContext expr ctx = runReader (RE.evalRelationalExpr expr) (RE.mkRelationalExprState ctx)
646 mkEvaldNot notif = EvaluatedNotification { notification = notif,
647 reportOldRelation = evalInContext (reportOldExpr notif) oldContext,
648 reportNewRelation = evalInContext (reportNewExpr notif) newContext}
649 pure (evaldNots, nodes)
651-- | Execute a transaction graph expression in the context of the session and connection. Transaction graph operators modify the transaction graph state.
653-- OPTIMIZATION OPPORTUNITY: no locks are required to write new transaction data, only to update the transaction graph id file
654-- if writing data is re-entrant, we may be able to use unsafeIOtoSTM
655-- perhaps keep hash of data file instead of checking if our head was updated on every write
656executeGraphExpr :: SessionId -> Connection -> TransactionGraphOperator -> IO (Either RelationalError ())
657executeGraphExpr sessionId (InProcessConnection conf) graphExpr = excEither $ do
658 let sessions = ipSessions conf
659 freshId <- nextRandom
660 stamp <- getCurrentTime
661 commitLock_ sessionId conf $ \updatedGraph -> do
662 eSession <- sessionForSessionId sessionId sessions
663 case eSession of
664 Left err -> pure (Left err)
665 Right session -> do
666 let discon = Sess.disconnectedTransaction session
667 case evalGraphOp stamp freshId discon updatedGraph graphExpr of
668 Left err -> pure (Left err)
669 Right (discon', graph') -> do
670 --if freshId appears in the graph, then we need to pass it on
671 let transIds = [freshId | isRight (transactionForId freshId graph')]
672 pure (Right (discon', graph', transIds))
673{-
674executeGraphExpr sessionId (InProcessConnection conf) graphExpr = excEither $ do
675 let strat = ipPersistenceStrategy conf
676 clientNodes = ipClientNodes conf
677 sessions = ipSessions conf
678 graphTvar = ipTransactionGraph conf
679 mLockFileH = ipLocks conf
680 lockHandler body = case graphExpr of
681 Commit -> case mLockFileH of
682 Nothing -> body False
683 Just (lockFileH, lockMVar) -> let acquireLocks = do
684 lastWrittenDigest <- takeMVar lockMVar
685 lockFile lockFileH WriteLock
686 latestDigest <- readGraphTransactionIdDigest strat
687 pure (latestDigest /= lastWrittenDigest)
689 releaseLocks _ = do
690 --still holding the lock- get the latest digest
691 gDigest <- readGraphTransactionIdDigest strat
692 unlockFile lockFileH
693 putMVar lockMVar gDigest
694 in bracket acquireLocks releaseLocks body
695 _ -> body False
696 freshId <- nextRandom
697 lockHandler $ \dbWrittenByOtherProcess -> do
698 --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
699 --read transaction data and compare to existing graph
700 --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)
701 manip <- atomically $ do
702 eSession <- sessionForSessionId sessionId sessions
703 --handle graph update by other process
704 oldGraph <- readTVar graphTvar
705 case eSession of
706 Left err -> pure (Left err)
707 Right session -> do
708 let mScriptSession = ipScriptSession conf
709 dbdir = case strat of
710 MinimalPersistence x -> x
711 CrashSafePersistence x -> x
712 _ -> error "accessing dbdir on non-persisted connection"
713 eRefreshedGraph <- if dbWrittenByOtherProcess then
714 unsafeIOToSTM (transactionGraphLoad dbdir oldGraph mScriptSession)
715 else
716 pure (Right oldGraph)
717 case eRefreshedGraph of
718 Left err -> pure (Left (DatabaseLoadError err))
719 Right refreshedGraph -> do
720 --snip it
721 eGraph <- executeGraphExprSTM_ dbWrittenByOtherProcess freshId sessionId session sessions graphExpr refreshedGraph graphTvar
722 --snip it
723 case eGraph of
724 Left err -> pure (Left err)
725 Right newGraph -> do
726 --handle commit
727 if isCommit graphExpr then do
728 case transactionForId (Sess.parentId session) oldGraph of
729 Left err -> pure $ Left err
730 Right previousTrans -> do
731 (evaldNots, nodes) <- executeCommitExprSTM_ (Trans.concreteDatabaseContext previousTrans) (Sess.concreteDatabaseContext session) clientNodes
732 nodesToNotify <- toList (STMSet.stream nodes)
733 pure $ Right (evaldNots, nodesToNotify, newGraph)
734 else
735 pure $ Right (M.empty, [], newGraph)
736 case manip of
737 Left err -> return $ Just err
738 Right (notsToFire, nodesToNotify, newGraph) -> do
739 --update filesystem database, if necessary
740 processTransactionGraphPersistence strat newGraph
741 sendNotifications nodesToNotify (ipLocalNode conf) notsToFire
742 pure Nothing
743-}
744executeGraphExpr sessionId conn@(RemoteProcessConnection _) graphExpr = remoteCall conn (ExecuteGraphExpr sessionId graphExpr)
746-- | A trans-graph expression is a relational query executed against the entirety of a transaction graph.
747executeTransGraphRelationalExpr :: SessionId -> Connection -> TransGraphRelationalExpr -> IO (Either RelationalError Relation)
748executeTransGraphRelationalExpr _ (InProcessConnection conf) tgraphExpr = excEither . atomically $ do
749 let graphTvar = ipTransactionGraph conf
750 graph <- readTVar graphTvar
751 case evalTransGraphRelationalExpr tgraphExpr graph of
752 Left err -> pure (Left err)
753 Right relExpr -> case runReader (RE.evalRelationalExpr relExpr) (RE.mkRelationalExprState DBC.empty) of
754 Left err -> pure (Left err)
755 Right rel -> pure (force (Right rel))
756executeTransGraphRelationalExpr sessionId conn@(RemoteProcessConnection _) tgraphExpr = remoteCall conn (ExecuteTransGraphRelationalExpr sessionId tgraphExpr)
758-- | Schema expressions manipulate the isomorphic schemas for the current 'DatabaseContext'.
759executeSchemaExpr :: SessionId -> Connection -> Schema.SchemaExpr -> IO (Either RelationalError ())
760executeSchemaExpr sessionId (InProcessConnection conf) schemaExpr = atomically $ do
761 let sessions = ipSessions conf
762 eSession <- sessionAndSchema sessionId sessions
763 case eSession of
764 Left err -> pure (Left err)
765 Right (session, _) -> do
766 let subschemas' = subschemas session
767 case Schema.evalSchemaExpr schemaExpr (Sess.concreteDatabaseContext session) subschemas' of
768 Left err -> pure (Left err)
769 Right (newSubschemas, newContext) -> do
770 --hm- maybe we should start using lenses
771 let discon = Sess.disconnectedTransaction session
772 newSchemas = Schemas newContext newSubschemas
773 newSession = Session (DisconnectedTransaction (Discon.parentId discon) newSchemas True) (Sess.schemaName session)
774 STMMap.insert newSession sessionId sessions
775 pure (Right ())
776executeSchemaExpr sessionId conn@(RemoteProcessConnection _) schemaExpr = remoteCall conn (ExecuteSchemaExpr sessionId schemaExpr)
778-- | 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.
779commit :: SessionId -> Connection -> IO (Either RelationalError ())
780commit sessionId conn@(InProcessConnection _) = executeGraphExpr sessionId conn Commit
781commit sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteGraphExpr sessionId Commit)
783sendNotifications :: [ProcessId] -> LocalNode -> EvaluatedNotifications -> IO ()
784sendNotifications pids localNode nots = mapM_ sendNots pids
785 where
786 sendNots remoteClientPid =
787 unless (M.null nots) $ runProcess localNode $ send remoteClientPid (NotificationMessage nots)
789-- | 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.
790rollback :: SessionId -> Connection -> IO (Either RelationalError ())
791rollback sessionId conn@(InProcessConnection _) = executeGraphExpr sessionId conn Rollback
792rollback sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteGraphExpr sessionId Rollback)
794-- | Write the transaction graph to disk. This function can be used to incrementally write new transactions to disk.
795processTransactionGraphPersistence :: PersistenceStrategy -> [TransactionId] -> TransactionGraph -> IO ()
796processTransactionGraphPersistence NoPersistence _ _ = pure ()
797processTransactionGraphPersistence (MinimalPersistence dbdir) transIds graph = transactionGraphPersist NoDiskSync dbdir transIds graph >> pure ()
798processTransactionGraphPersistence (CrashSafePersistence dbdir) transIds graph = transactionGraphPersist FsyncDiskSync dbdir transIds graph >> pure ()
800readGraphTransactionIdDigest :: PersistenceStrategy -> IO LockFileHash
801readGraphTransactionIdDigest NoPersistence = error "attempt to read digest from transaction log without persistence enabled"
802readGraphTransactionIdDigest (MinimalPersistence dbdir) = readGraphTransactionIdFileDigest dbdir
803readGraphTransactionIdDigest (CrashSafePersistence dbdir) = readGraphTransactionIdFileDigest dbdir
805-- | 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.
806typeForRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)
807typeForRelationalExpr sessionId conn@(InProcessConnection _) relExpr = atomically $ typeForRelationalExprSTM sessionId conn relExpr
808typeForRelationalExpr sessionId conn@(RemoteProcessConnection _) relExpr = remoteCall conn (ExecuteTypeForRelationalExpr sessionId relExpr)
810typeForRelationalExprSTM :: SessionId -> Connection -> RelationalExpr -> STM (Either RelationalError Relation)
811typeForRelationalExprSTM sessionId (InProcessConnection conf) relExpr = do
812 let sessions = ipSessions conf
813 eSession <- sessionAndSchema sessionId sessions
814 case eSession of
815 Left err -> pure $ Left err
816 Right (session, schema) -> do
817 let processed = if schemaName session == defaultSchemaName then
818 Right relExpr
819 else
820 Schema.processRelationalExprInSchema schema relExpr
821 case processed of
822 Left err -> pure (Left err)
823 Right relExpr' -> pure $ runReader (RE.typeForRelationalExpr relExpr') (RE.mkRelationalExprState (Sess.concreteDatabaseContext session))
825typeForRelationalExprSTM _ _ _ = error "typeForRelationalExprSTM called on non-local connection"
827-- | Return a 'Map' of the database's constraints at the context of the session and connection.
828inclusionDependencies :: SessionId -> Connection -> IO (Either RelationalError InclusionDependencies)
829inclusionDependencies sessionId (InProcessConnection conf) = do
830 let sessions = ipSessions conf
831 atomically $ do
832 eSession <- sessionAndSchema sessionId sessions
833 case eSession of
834 Left err -> pure $ Left err
835 Right (session, schema) -> do
836 let context = Sess.concreteDatabaseContext session
837 if schemaName session == defaultSchemaName then
838 pure $ Right (B.inclusionDependencies context)
839 else
840 pure (Schema.inclusionDependenciesInSchema schema (B.inclusionDependencies context))
842inclusionDependencies sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveInclusionDependencies sessionId)
844typeConstructorMapping :: SessionId -> Connection -> IO (Either RelationalError TypeConstructorMapping)
845typeConstructorMapping sessionId (InProcessConnection conf) = do
846 let sessions = ipSessions conf
847 atomically $ do
848 eSession <- sessionAndSchema sessionId sessions
849 case eSession of
850 Left err -> pure $ Left err
851 Right (session, _) -> --warning, no schema support for typeconstructors
852 pure (Right (B.typeConstructorMapping (Sess.concreteDatabaseContext session)))
853typeConstructorMapping sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveTypeConstructorMapping sessionId)
855-- | 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.
856planForDatabaseContextExpr :: SessionId -> Connection -> DatabaseContextExpr -> IO (Either RelationalError DatabaseContextExpr)
857planForDatabaseContextExpr sessionId (InProcessConnection conf) dbExpr = do
858 let sessions = ipSessions conf
859 atomically $ do
860 eSession <- sessionAndSchema sessionId sessions
861 case eSession of
862 Left err -> pure $ Left err
863 Right (session, _) -> if schemaName session == defaultSchemaName then
864 pure $ evalState (applyStaticDatabaseOptimization dbExpr) (RE.freshDatabaseState (Sess.concreteDatabaseContext session))
865 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
866 pure (Right dbExpr)
868planForDatabaseContextExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (RetrievePlanForDatabaseContextExpr sessionId dbExpr)
870-- | Return a relation which represents the current state of the global transaction graph. The attributes are
871-- * current- boolean attribute representing whether or not the current session references this transaction
872-- * head- text attribute which is a non-empty 'HeadName' iff the transaction references a head.
873-- * id- id attribute of the transaction
874-- * parents- a relation-valued attribute which contains a relation of transaction ids which are parent transaction to the transaction
875transactionGraphAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
876transactionGraphAsRelation sessionId (InProcessConnection conf) = do
877 let sessions = ipSessions conf
878 tvar = ipTransactionGraph conf
879 atomically $ do
880 eSession <- sessionForSessionId sessionId sessions
881 case eSession of
882 Left err -> pure $ Left err
883 Right session -> do
884 graph <- readTVar tvar
885 pure $ graphAsRelation (Sess.disconnectedTransaction session) graph
887transactionGraphAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveTransactionGraph sessionId)
889-- | Returns the names and types of the relation variables in the current 'Session'.
890relationVariablesAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
891relationVariablesAsRelation sessionId (InProcessConnection conf) = do
892 let sessions = ipSessions conf
893 atomically $ do
894 eSession <- sessionAndSchema sessionId sessions
895 case eSession of
896 Left err -> pure (Left err)
897 Right (session, schema) -> do
898 let context = Sess.concreteDatabaseContext session
899 if Sess.schemaName session == defaultSchemaName then
900 pure $ R.relationVariablesAsRelation (relationVariables context)
901 else
902 case Schema.relationVariablesInSchema schema context of
903 Left err -> pure (Left err)
904 Right relvars -> pure $ R.relationVariablesAsRelation relvars
906relationVariablesAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveRelationVariableSummary sessionId)
908-- | Returns the transaction id for the connection's disconnected transaction committed parent transaction.
909headTransactionId :: SessionId -> Connection -> IO (Either RelationalError TransactionId)
910headTransactionId sessionId (InProcessConnection conf) = do
911 let sessions = ipSessions conf
912 atomically $ do
913 eSession <- sessionForSessionId sessionId sessions
914 case eSession of
915 Left err -> pure (Left err)
916 Right session -> pure $ Right (Sess.parentId session)
917headTransactionId sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveHeadTransactionId sessionId)
919headNameSTM_ :: SessionId -> Sessions -> TVar TransactionGraph -> STM (Either RelationalError HeadName)
920headNameSTM_ sessionId sessions graphTvar = do
921 graph <- readTVar graphTvar
922 eSession <- sessionForSessionId sessionId sessions
923 case eSession of
924 Left err -> pure (Left err)
925 Right session -> case transactionForId (Sess.parentId session) graph of
926 Left err -> pure (Left err)
927 Right parentTrans -> case headNameForTransaction parentTrans graph of
928 Nothing -> pure (Left UnknownHeadError)
929 Just headName' -> pure (Right headName')
931-- | Returns Just the name of the head of the current disconnected transaction or Nothing.
932headName :: SessionId -> Connection -> IO (Either RelationalError HeadName)
933headName sessionId (InProcessConnection conf) = do
934 let sessions = ipSessions conf
935 graphTvar = ipTransactionGraph conf
936 atomically (headNameSTM_ sessionId sessions graphTvar)
937headName sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteHeadName sessionId)
939-- | Returns a listing of all available atom types.
940atomTypesAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
941atomTypesAsRelation sessionId (InProcessConnection conf) = do
942 let sessions = ipSessions conf
943 atomically $ do
944 eSession <- sessionForSessionId sessionId sessions
945 case eSession of
946 Left err -> pure (Left err)
947 Right session ->
948 case typesAsRelation (B.typeConstructorMapping (Sess.concreteDatabaseContext session)) of
949 Left err -> pure (Left err)
950 Right rel -> pure (Right rel)
951atomTypesAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveAtomTypesAsRelation sessionId)
953disconnectedTransactionIsDirty :: SessionId -> Connection -> IO (Either RelationalError Bool)
954disconnectedTransactionIsDirty sessionId (InProcessConnection conf) = do
955 let sessions = ipSessions conf
956 atomically $ do
957 eSession <- sessionForSessionId sessionId sessions
958 case eSession of
959 Left err -> pure (Left err)
960 Right session ->
961 pure (Right (isDirty session))
962disconnectedTransactionIsDirty sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveSessionIsDirty sessionId)
964--used only for testing- we expect this to throw an exception
965callTestTimeout_ :: SessionId -> Connection -> IO Bool
966callTestTimeout_ _ (InProcessConnection _) = error "bad testing call"
967callTestTimeout_ sessionId conn@(RemoteProcessConnection _) = remoteCall conn (TestTimeout sessionId)
969--used in tests only
970transactionGraph_ :: Connection -> IO TransactionGraph
971transactionGraph_ (InProcessConnection conf) = atomically $ readTVar (ipTransactionGraph conf)
972transactionGraph_ _ = error "remote connection used"
974--used in tests only
975disconnectedTransaction_ :: SessionId -> Connection -> IO DisconnectedTransaction
976disconnectedTransaction_ sessionId (InProcessConnection conf) = do
977 let sessions = ipSessions conf
978 mSession <- atomically $ STMMap.lookup sessionId sessions
979 case mSession of
980 Nothing -> error "No such session"
981 Just (Sess.Session discon _) -> pure discon
982disconnectedTransaction_ _ _= error "remote connection used"
984-- wrap a graph evaluation in file locking
985commitLock_ :: SessionId ->
986 InProcessConnectionConf ->
987 (TransactionGraph ->
988 STM (Either RelationalError (DisconnectedTransaction, TransactionGraph, [TransactionId]))) ->
989 IO (Either RelationalError ())
990commitLock_ sessionId conf stmBlock = do
991 let sessions = ipSessions conf
992 strat = ipPersistenceStrategy conf
993 mScriptSession = ipScriptSession conf
994 graphTvar = ipTransactionGraph conf
995 clientNodes = ipClientNodes conf
996 mLockFileH = ipLocks conf
997 lockHandler body = case mLockFileH of
998 Nothing -> body False
999 Just (lockFileH, lockMVar) ->
1000 let acquireLocks = do
1001 lastWrittenDigest <- takeMVar lockMVar
1002 lockFile lockFileH WriteLock
1003 latestDigest <- readGraphTransactionIdDigest strat
1004 pure (latestDigest /= lastWrittenDigest)
1005 releaseLocks _ = do
1006 --still holding the lock- get the latest digest
1007 gDigest <- readGraphTransactionIdDigest strat
1008 unlockFile lockFileH
1009 putMVar lockMVar gDigest
1010 in bracket acquireLocks releaseLocks body
1011 manip <- lockHandler $ \dbWrittenByOtherProcess -> atomically $ do
1012 eSession <- sessionForSessionId sessionId sessions
1013 --handle graph update by other process
1014 oldGraph <- readTVar graphTvar
1015 case eSession of
1016 Left err -> pure (Left err)
1017 Right session -> do
1018 let dbdir = case strat of
1019 MinimalPersistence x -> x
1020 CrashSafePersistence x -> x
1021 _ -> error "accessing dbdir on non-persisted connection"
1022 --this should also happen for non-commit expressions
1023 eRefreshedGraph <- if dbWrittenByOtherProcess then
1024 unsafeIOToSTM (transactionGraphLoad dbdir oldGraph mScriptSession)
1025 else
1026 pure (Right oldGraph)
1027 case eRefreshedGraph of
1028 Left err -> pure (Left (DatabaseLoadError err))
1029 Right refreshedGraph -> do
1030 eGraph <- stmBlock refreshedGraph
1031 case eGraph of
1032 Left err -> pure (Left err)
1033 Right (discon', graph', transactionIdsToPersist) -> do
1034 writeTVar graphTvar graph'
1035 let newSession = Session discon' (Sess.schemaName session)
1036 STMMap.insert newSession sessionId sessions
1037 case transactionForId (Sess.parentId session) oldGraph of
1038 Left err -> pure $ Left err
1039 Right previousTrans ->
1040 if not (Prelude.null transactionIdsToPersist) then do
1041 (evaldNots, nodes) <- executeCommitExprSTM_ (Trans.concreteDatabaseContext previousTrans) (Sess.concreteDatabaseContext session) clientNodes
1042 nodesToNotify <- toList (STMSet.stream nodes)
1043 pure $ Right (evaldNots, nodesToNotify, graph', transactionIdsToPersist)
1044 else pure (Right (M.empty, [], graph', []))
1046 --handle notification firing
1047 case manip of
1048 Left err -> pure (Left err)
1049 Right (notsToFire, nodesToNotify, newGraph, transactionIdsToPersist) -> do
1050 --update filesystem database, if necessary
1051 processTransactionGraphPersistence strat transactionIdsToPersist newGraph
1052 sendNotifications nodesToNotify (ipLocalNode conf) notsToFire
1053 pure (Right ())
1055writeDisconAndGraph_ :: TVar TransactionGraph -> SessionId -> Session -> Sessions -> DisconnectedTransaction -> TransactionGraph -> STM ()
1056writeDisconAndGraph_ graphTvar sessionId session sessions discon graph = do
1057 writeTVar graphTvar graph
1058 let newSession = Session discon (Sess.schemaName session)
1059 STMMap.insert newSession sessionId sessions
1062-- | Runs an IO monad, commits the result when the monad returns no errors, otherwise, rolls back the changes and the error.
1063withTransaction :: SessionId -> Connection -> IO (Either RelationalError a) -> IO (Either RelationalError ()) -> IO (Either RelationalError a)
1064withTransaction sessionId conn io successFunc = bracketOnError (pure ()) (const do_rollback) block
1065 where
1066 do_rollback = rollback sessionId conn
1067 block _ = do
1068 eErr <- io
1069 case eErr of
1070 Left err -> do
1071 _ <- do_rollback
1072 pure (Left err)
1073 Right val -> do
1074 eIsDirty <- disconnectedTransactionIsDirty sessionId conn
1075 case eIsDirty of
1076 Left err -> pure (Left err)
1077 Right dirty ->
1078 if dirty then do
1079 res <- successFunc
1080 case res of
1081 Left err -> pure (Left err)
1082 Right _ -> pure (Right val)
1083 else -- no updates executed, so don't create a commit
1084 pure (Right val)