git.iko.soy/project-m36
on 41d8517 help sign in
project-m36/src/lib/ProjectM36/Client.hs
1091 lines 58 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 takeMVar connStatus
335connectPersistentProjectM36 :: PersistenceStrategy ->
336 DiskSync ->
337 FilePath ->
338 TransactionGraph ->
339 NotificationCallback ->
340 [GhcPkgPath] ->
341 IO (Either ConnectionError Connection)
342connectPersistentProjectM36 strat sync dbdir freshGraph notificationCallback ghcPkgPaths = do
343 err <- setupDatabaseDir sync dbdir freshGraph
344 case err of
345 Left err' -> return $ Left (SetupDatabaseDirectoryError err')
346 Right (lockFileH, digest) -> do
347 mScriptSession <- createScriptSession ghcPkgPaths
348 graph <- transactionGraphLoad dbdir emptyTransactionGraph mScriptSession
349 case graph of
350 Left err' -> return $ Left (SetupDatabaseDirectoryError err')
351 Right graph' -> do
352 tvarGraph <- newTVarIO graph'
353 sessions <- STMMap.newIO
354 clientNodes <- STMSet.newIO
355 (localNode, transport) <- createLocalNode
356 lockMVar <- newMVar digest
357 let conn = InProcessConnection InProcessConnectionConf {
358 ipPersistenceStrategy = strat,
359 ipClientNodes = clientNodes,
360 ipSessions = sessions,
361 ipTransactionGraph = tvarGraph,
362 ipScriptSession = mScriptSession,
363 ipLocalNode = localNode,
364 ipTransport = transport,
365 ipLocks = Just (lockFileH, lockMVar)
366 }
368 notificationPid <- startNotificationListener localNode notificationCallback
369 addClientNode conn notificationPid
370 pure (Right conn)
372-- | Create a new session at the transaction id and return the session's Id.
373createSessionAtCommit :: Connection -> TransactionId -> IO (Either RelationalError SessionId)
374createSessionAtCommit conn@(InProcessConnection _) commitId = do
375 newSessionId <- nextRandom
376 atomically $ createSessionAtCommit_ commitId newSessionId conn
377createSessionAtCommit conn@(RemoteProcessConnection _) uuid = remoteCall conn (CreateSessionAtCommit uuid)
379createSessionAtCommit_ :: TransactionId -> SessionId -> Connection -> STM (Either RelationalError SessionId)
380createSessionAtCommit_ commitId newSessionId (InProcessConnection conf) = do
381 let sessions = ipSessions conf
382 graphTvar = ipTransactionGraph conf
383 graph <- readTVar graphTvar
384 case transactionForId commitId graph of
385 Left err -> pure (Left err)
386 Right transaction -> do
387 let freshDiscon = DisconnectedTransaction commitId (Trans.schemas transaction) False
388 keyDuplication <- STMMap.lookup newSessionId sessions
389 case keyDuplication of
390 Just _ -> pure $ Left (SessionIdInUseError newSessionId)
391 Nothing -> do
392 STMMap.insert (Session freshDiscon defaultSchemaName) newSessionId sessions
393 pure $ Right newSessionId
394createSessionAtCommit_ _ _ (RemoteProcessConnection _) = error "createSessionAtCommit_ called on remote connection"
396-- | 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.
397createSessionAtHead :: Connection -> HeadName -> IO (Either RelationalError SessionId)
398createSessionAtHead conn@(InProcessConnection conf) headn = do
399 let graphTvar = ipTransactionGraph conf
400 newSessionId <- nextRandom
401 atomically $ do
402 graph <- readTVar graphTvar
403 case transactionForHead headn graph of
404 Nothing -> pure $ Left (NoSuchHeadNameError headn)
405 Just trans -> createSessionAtCommit_ (transactionId trans) newSessionId conn
406createSessionAtHead conn@(RemoteProcessConnection _) headn = remoteCall conn (CreateSessionAtHead headn)
408-- | Used internally for server connections to keep track of remote nodes for the purpose of sending notifications later.
409addClientNode :: Connection -> ProcessId -> IO ()
410addClientNode (RemoteProcessConnection _) _ = error "addClientNode called on remote connection"
411addClientNode (InProcessConnection conf) newProcessId = atomically (STMSet.insert newProcessId (ipClientNodes conf))
413-- | Discards a session, eliminating any uncommitted changes present in the session.
414closeSession :: SessionId -> Connection -> IO ()
415closeSession sessionId (InProcessConnection conf) =
416 atomically $ STMMap.delete sessionId (ipSessions conf)
417closeSession 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)
428 let mLocks = ipLocks conf
429 case mLocks of
430 Nothing -> pure ()
431 Just (lockFileH, _) -> closeLockFile lockFileH
433close conn@(RemoteProcessConnection conf) = do
434 _ <- remoteCall conn Logout :: IO Bool
435 closeLocalNode (rLocalNode conf)
436 closeTransport (rTransport conf)
438--used only by the server EntryPoints
439closeRemote_ :: Connection -> IO ()
440closeRemote_ (InProcessConnection _) = error "invalid call of closeRemote_ on InProcessConnection"
441closeRemote_ (RemoteProcessConnection conf) = runProcess (rLocalNode conf) (reconnect (rProcessId conf))
443 --we need to actually close the localNode's connection to the remote
444--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
445--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.
447excEither :: IO (Either RelationalError a) -> IO (Either RelationalError a)
448excEither = handle handler
449 where
450 handler exc | Just (_ :: AsyncException) <- fromException exc = throwIO exc
451 | otherwise = pure (Left (UnhandledExceptionError (show exc)))
453runProcessResult :: LocalNode -> Process a -> IO a
454runProcessResult localNode proc = do
455 ret <- newEmptyMVar
456 runProcess localNode $ do
457 val <- proc
458 liftIO $ putMVar ret val
459 takeMVar ret
461safeLogin :: Login -> ProcessId -> Process Bool
462safeLogin login procId = do
463 ret <- call procId login
464 case ret of
465 Left (_ :: ServerError) -> pure False
466 Right val -> pure val
468remoteCall :: (Serializable a, Serializable b) => Connection -> a -> IO b
469remoteCall (InProcessConnection _ ) _ = error "remoteCall called on local connection"
470remoteCall (RemoteProcessConnection conf) arg = runProcessResult localNode $ do
471 ret <- safeCall serverProcessId arg
472 case ret of
473 Left err -> error ("server died: " ++ show err)
474 Right ret' -> case ret' of
475 Left RequestTimeoutError -> liftIO (throwIO RequestTimeoutException)
476 Left (ProcessDiedError _) -> liftIO (throwIO RemoteProcessDiedException)
477 Right val -> pure val
478 where
479 localNode = rLocalNode conf
480 serverProcessId = rProcessId conf
482sessionForSessionId :: SessionId -> Sessions -> STM (Either RelationalError Session)
483sessionForSessionId sessionId sessions =
484 maybe (Left $ NoSuchSessionError sessionId) Right <$> STMMap.lookup sessionId sessions
486schemaForSessionId :: Session -> STM (Either RelationalError Schema)
487schemaForSessionId session = do
488 let sname = schemaName session
489 if sname == defaultSchemaName then
490 pure (Right (Schema [])) -- the main schema includes no transformations (but neither do empty schemas :/ )
491 else
492 case M.lookup sname (subschemas session) of
493 Nothing -> pure (Left (SubschemaNameNotInUseError sname))
494 Just schema -> pure (Right schema)
496sessionAndSchema :: SessionId -> Sessions -> STM (Either RelationalError (Session, Schema))
497sessionAndSchema sessionId sessions = do
498 eSession <- sessionForSessionId sessionId sessions
499 case eSession of
500 Left err -> pure (Left err)
501 Right session -> do
502 eSchema <- schemaForSessionId session
503 case eSchema of
504 Left err -> pure (Left err)
505 Right schema -> pure (Right (session, schema))
507-- | Returns the name of the currently selected isomorphic schema.
508currentSchemaName :: SessionId -> Connection -> IO (Either RelationalError SchemaName)
509currentSchemaName sessionId (InProcessConnection conf) = atomically $ do
510 let sessions = ipSessions conf
511 eSession <- sessionForSessionId sessionId sessions
512 case eSession of
513 Left err -> pure (Left err)
514 Right session -> pure (Right (Sess.schemaName session))
515currentSchemaName sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveCurrentSchemaName sessionId)
517-- | Switch to the named isomorphic schema.
518setCurrentSchemaName :: SessionId -> Connection -> SchemaName -> IO (Either RelationalError ())
519setCurrentSchemaName sessionId (InProcessConnection conf) sname = atomically $ do
520 let sessions = ipSessions conf
521 eSession <- sessionForSessionId sessionId sessions
522 case eSession of
523 Left err -> pure (Left err)
524 Right session -> case Sess.setSchemaName sname session of
525 Left err -> pure (Left err)
526 Right newSession -> STMMap.insert newSession sessionId sessions >> pure (Right ())
527setCurrentSchemaName sessionId conn@(RemoteProcessConnection _) sname = remoteCall conn (ExecuteSetCurrentSchema sessionId sname)
529-- | Execute a relational expression in the context of the session and connection. Relational expressions are queries and therefore cannot alter the database.
530executeRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)
531executeRelationalExpr sessionId (InProcessConnection conf) expr = excEither $ atomically $ do
532 let sessions = ipSessions conf
533 eSession <- sessionAndSchema sessionId sessions
534 case eSession of
535 Left err -> pure $ Left err
536 Right (session, schema) -> do
537 let expr' = if schemaName session /= defaultSchemaName then
538 Schema.processRelationalExprInSchema schema expr
539 else
540 Right expr
541 case expr' of
542 Left err -> pure (Left err)
543 Right expr'' -> case optimizeRelationalExpr (Sess.concreteDatabaseContext session) expr'' of
544 Left err -> pure (Left err)
545 Right optExpr -> do
546 let evald = runReader (RE.evalRelationalExpr optExpr) (RE.mkRelationalExprState (Sess.concreteDatabaseContext session))
547 case evald of
548 Left err -> pure (Left err)
549 Right rel -> pure (force (Right rel)) -- this is necessary so that any undefined/error exceptions are spit out here
550executeRelationalExpr sessionId conn@(RemoteProcessConnection _) relExpr = remoteCall conn (ExecuteRelationalExpr sessionId relExpr)
552-- | 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.
553executeDatabaseContextExpr :: SessionId -> Connection -> DatabaseContextExpr -> IO (Either RelationalError ())
554executeDatabaseContextExpr sessionId (InProcessConnection conf) expr = excEither $ atomically $ do
555 let sessions = ipSessions conf
556 eSession <- sessionAndSchema sessionId sessions
557 case eSession of
558 Left err -> pure (Left err)
559 Right (session, schema) -> do
560 let expr' = if schemaName session == defaultSchemaName then
561 Right expr
562 else
563 Schema.processDatabaseContextExprInSchema schema expr
564 case expr' of
565 Left err -> pure (Left err)
566 Right expr'' -> case runState (do
567 eOptExpr <- applyStaticDatabaseOptimization expr''
568 case eOptExpr of
569 Left err -> pure (Left err)
570 Right optExpr ->
571 RE.evalDatabaseContextExpr optExpr) (RE.freshDatabaseState (Sess.concreteDatabaseContext session)) of
572 (Left err,_) -> pure (Left err)
573 (Right (), (_,_,False)) -> pure (Right ()) --optimization- if nothing was dirtied, nothing to do
574 (Right (), (!context',_,True)) -> do
575 let newDiscon = DisconnectedTransaction (Sess.parentId session) newSchemas True
576 newSubschemas = Schema.processDatabaseContextExprSchemasUpdate (Sess.subschemas session) expr
577 newSchemas = Schemas context' newSubschemas
578 newSession = Session newDiscon (Sess.schemaName session)
579 STMMap.insert newSession sessionId sessions
580 pure (Right ())
581executeDatabaseContextExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (ExecuteDatabaseContextExpr sessionId dbExpr)
583-- | 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). Alternatively, as an optimization, if a simple commit is possible (meaning that the head has not changed), then a fast-forward commit takes place instead.
584autoMergeToHead :: SessionId -> Connection -> MergeStrategy -> HeadName -> IO (Either RelationalError ())
585autoMergeToHead sessionId (InProcessConnection conf) strat headName' = do
586 let sessions = ipSessions conf
587 id1 <- nextRandom
588 id2 <- nextRandom
589 id3 <- nextRandom
590 stamp <- getCurrentTime
591 commitLock_ sessionId conf $ \graph -> do
592 eSession <- sessionForSessionId sessionId sessions
593 case eSession of
594 Left err -> pure (Left err)
595 Right session ->
596 case Graph.transactionForHead headName' graph of
597 Nothing -> pure (Left (NoSuchHeadNameError headName'))
598 Just headTrans -> do
599 --attempt fast-forward commit, if possible
600 let graphInfo = if Sess.parentId session == transactionId headTrans then do
601 ret <- Graph.evalGraphOp stamp id1 (Sess.disconnectedTransaction session) graph Commit
602 pure (ret, [id1])
603 else do
604 ret <- Graph.autoMergeToHead stamp (id1, id2, id3) (Sess.disconnectedTransaction session) headName' strat graph
605 pure (ret, [id1,id2,id3])
606 case graphInfo of
607 Left err -> pure (Left err)
608 Right ((discon', graph'), transactionIdsAdded) ->
609 pure (Right (discon', graph', transactionIdsAdded))
610autoMergeToHead sessionId conn@(RemoteProcessConnection _) strat headName' = remoteCall conn (ExecuteAutoMergeToHead sessionId strat headName')
612-- | Execute a database context IO-monad-based expression for the given session and connection. `DatabaseContextIOExpr`s modify the DatabaseContext but cannot be purely implemented.
613--this is almost completely identical to executeDatabaseContextExpr above
614executeDatabaseContextIOExpr :: SessionId -> Connection -> DatabaseContextIOExpr -> IO (Either RelationalError ())
615executeDatabaseContextIOExpr sessionId (InProcessConnection conf) expr = excEither $ do
616 let sessions = ipSessions conf
617 scriptSession = ipScriptSession conf
618 eSession <- atomically $ sessionForSessionId sessionId sessions --potentially race condition due to interleaved IO?
619 case eSession of
620 Left err -> pure (Left err)
621 Right session -> do
622 res <- RE.evalDatabaseContextIOExpr scriptSession (Sess.concreteDatabaseContext session) expr
623 case res of
624 Left err -> pure (Left err)
625 Right context' -> do
626 let newDiscon = DisconnectedTransaction (Sess.parentId session) newSchemas True
627 newSchemas = Schemas context' (Sess.subschemas session)
628 newSession = Session newDiscon (Sess.schemaName session)
629 atomically $ STMMap.insert newSession sessionId sessions
630 pure (Right ())
631executeDatabaseContextIOExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (ExecuteDatabaseContextIOExpr sessionId dbExpr)
633{-
634executeGraphExprSTM_ :: TransactionId -> SessionId -> Session -> Sessions -> TransactionGraphOperator -> TransactionGraph -> TVar TransactionGraph -> STM (Either RelationalError (TransactionGraph, DisconnectedTransaction)
635executeGraphExprSTM_ freshId sessionId session sessions graphExpr graph graphTVar= do
636 case evalGraphOp freshId (Sess.disconnectedTransaction session) graph graphExpr of
637 Left err -> do
638 when updateGraphOnError (writeTVar graphTVar graph)
639 pure $ Left err
640 Right (discon', graph') -> do
641 writeTVar graphTVar graph'
642 let newSession = Session discon' (Sess.schemaName session)
643 STMMap.insert newSession sessionId sessions
644 pure $ Right graph'
645-}
647-- process notifications for commits
648executeCommitExprSTM_ :: DatabaseContext -> DatabaseContext -> ClientNodes -> STM (EvaluatedNotifications, ClientNodes)
649executeCommitExprSTM_ oldContext newContext nodes = do
650 let nots = notifications oldContext
651 fireNots = notificationChanges nots oldContext newContext
652 evaldNots = M.map mkEvaldNot fireNots
653 evalInContext expr ctx = runReader (RE.evalRelationalExpr expr) (RE.mkRelationalExprState ctx)
654 mkEvaldNot notif = EvaluatedNotification { notification = notif,
655 reportOldRelation = evalInContext (reportOldExpr notif) oldContext,
656 reportNewRelation = evalInContext (reportNewExpr notif) newContext}
657 pure (evaldNots, nodes)
659-- | Execute a transaction graph expression in the context of the session and connection. Transaction graph operators modify the transaction graph state.
661-- OPTIMIZATION OPPORTUNITY: no locks are required to write new transaction data, only to update the transaction graph id file
662-- if writing data is re-entrant, we may be able to use unsafeIOtoSTM
663-- perhaps keep hash of data file instead of checking if our head was updated on every write
664executeGraphExpr :: SessionId -> Connection -> TransactionGraphOperator -> IO (Either RelationalError ())
665executeGraphExpr sessionId (InProcessConnection conf) graphExpr = excEither $ do
666 let sessions = ipSessions conf
667 freshId <- nextRandom
668 stamp <- getCurrentTime
669 commitLock_ sessionId conf $ \updatedGraph -> do
670 eSession <- sessionForSessionId sessionId sessions
671 case eSession of
672 Left err -> pure (Left err)
673 Right session -> do
674 let discon = Sess.disconnectedTransaction session
675 case evalGraphOp stamp freshId discon updatedGraph graphExpr of
676 Left err -> pure (Left err)
677 Right (discon', graph') -> do
678 --if freshId appears in the graph, then we need to pass it on
679 let transIds = [freshId | isRight (transactionForId freshId graph')]
680 pure (Right (discon', graph', transIds))
681{-
682executeGraphExpr sessionId (InProcessConnection conf) graphExpr = excEither $ do
683 let strat = ipPersistenceStrategy conf
684 clientNodes = ipClientNodes conf
685 sessions = ipSessions conf
686 graphTvar = ipTransactionGraph conf
687 mLockFileH = ipLocks conf
688 lockHandler body = case graphExpr of
689 Commit -> case mLockFileH of
690 Nothing -> body False
691 Just (lockFileH, lockMVar) -> let acquireLocks = do
692 lastWrittenDigest <- takeMVar lockMVar
693 lockFile lockFileH WriteLock
694 latestDigest <- readGraphTransactionIdDigest strat
695 pure (latestDigest /= lastWrittenDigest)
697 releaseLocks _ = do
698 --still holding the lock- get the latest digest
699 gDigest <- readGraphTransactionIdDigest strat
700 unlockFile lockFileH
701 putMVar lockMVar gDigest
702 in bracket acquireLocks releaseLocks body
703 _ -> body False
704 freshId <- nextRandom
705 lockHandler $ \dbWrittenByOtherProcess -> do
706 --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
707 --read transaction data and compare to existing graph
708 --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)
709 manip <- atomically $ do
710 eSession <- sessionForSessionId sessionId sessions
711 --handle graph update by other process
712 oldGraph <- readTVar graphTvar
713 case eSession of
714 Left err -> pure (Left err)
715 Right session -> do
716 let mScriptSession = ipScriptSession conf
717 dbdir = case strat of
718 MinimalPersistence x -> x
719 CrashSafePersistence x -> x
720 _ -> error "accessing dbdir on non-persisted connection"
721 eRefreshedGraph <- if dbWrittenByOtherProcess then
722 unsafeIOToSTM (transactionGraphLoad dbdir oldGraph mScriptSession)
723 else
724 pure (Right oldGraph)
725 case eRefreshedGraph of
726 Left err -> pure (Left (DatabaseLoadError err))
727 Right refreshedGraph -> do
728 --snip it
729 eGraph <- executeGraphExprSTM_ dbWrittenByOtherProcess freshId sessionId session sessions graphExpr refreshedGraph graphTvar
730 --snip it
731 case eGraph of
732 Left err -> pure (Left err)
733 Right newGraph -> do
734 --handle commit
735 if isCommit graphExpr then do
736 case transactionForId (Sess.parentId session) oldGraph of
737 Left err -> pure $ Left err
738 Right previousTrans -> do
739 (evaldNots, nodes) <- executeCommitExprSTM_ (Trans.concreteDatabaseContext previousTrans) (Sess.concreteDatabaseContext session) clientNodes
740 nodesToNotify <- toList (STMSet.stream nodes)
741 pure $ Right (evaldNots, nodesToNotify, newGraph)
742 else
743 pure $ Right (M.empty, [], newGraph)
744 case manip of
745 Left err -> return $ Just err
746 Right (notsToFire, nodesToNotify, newGraph) -> do
747 --update filesystem database, if necessary
748 processTransactionGraphPersistence strat newGraph
749 sendNotifications nodesToNotify (ipLocalNode conf) notsToFire
750 pure Nothing
751-}
752executeGraphExpr sessionId conn@(RemoteProcessConnection _) graphExpr = remoteCall conn (ExecuteGraphExpr sessionId graphExpr)
754-- | A trans-graph expression is a relational query executed against the entirety of a transaction graph.
755executeTransGraphRelationalExpr :: SessionId -> Connection -> TransGraphRelationalExpr -> IO (Either RelationalError Relation)
756executeTransGraphRelationalExpr _ (InProcessConnection conf) tgraphExpr = excEither . atomically $ do
757 let graphTvar = ipTransactionGraph conf
758 graph <- readTVar graphTvar
759 case evalTransGraphRelationalExpr tgraphExpr graph of
760 Left err -> pure (Left err)
761 Right relExpr -> case runReader (RE.evalRelationalExpr relExpr) (RE.mkRelationalExprState DBC.empty) of
762 Left err -> pure (Left err)
763 Right rel -> pure (force (Right rel))
764executeTransGraphRelationalExpr sessionId conn@(RemoteProcessConnection _) tgraphExpr = remoteCall conn (ExecuteTransGraphRelationalExpr sessionId tgraphExpr)
766-- | Schema expressions manipulate the isomorphic schemas for the current 'DatabaseContext'.
767executeSchemaExpr :: SessionId -> Connection -> Schema.SchemaExpr -> IO (Either RelationalError ())
768executeSchemaExpr sessionId (InProcessConnection conf) schemaExpr = atomically $ do
769 let sessions = ipSessions conf
770 eSession <- sessionAndSchema sessionId sessions
771 case eSession of
772 Left err -> pure (Left err)
773 Right (session, _) -> do
774 let subschemas' = subschemas session
775 case Schema.evalSchemaExpr schemaExpr (Sess.concreteDatabaseContext session) subschemas' of
776 Left err -> pure (Left err)
777 Right (newSubschemas, newContext) -> do
778 --hm- maybe we should start using lenses
779 let discon = Sess.disconnectedTransaction session
780 newSchemas = Schemas newContext newSubschemas
781 newSession = Session (DisconnectedTransaction (Discon.parentId discon) newSchemas True) (Sess.schemaName session)
782 STMMap.insert newSession sessionId sessions
783 pure (Right ())
784executeSchemaExpr sessionId conn@(RemoteProcessConnection _) schemaExpr = remoteCall conn (ExecuteSchemaExpr sessionId schemaExpr)
786-- | 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.
787commit :: SessionId -> Connection -> IO (Either RelationalError ())
788commit sessionId conn@(InProcessConnection _) = executeGraphExpr sessionId conn Commit
789commit sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteGraphExpr sessionId Commit)
791sendNotifications :: [ProcessId] -> LocalNode -> EvaluatedNotifications -> IO ()
792sendNotifications pids localNode nots = mapM_ sendNots pids
793 where
794 sendNots remoteClientPid =
795 unless (M.null nots) $ runProcess localNode $ send remoteClientPid (NotificationMessage nots)
797-- | 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.
798rollback :: SessionId -> Connection -> IO (Either RelationalError ())
799rollback sessionId conn@(InProcessConnection _) = executeGraphExpr sessionId conn Rollback
800rollback sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteGraphExpr sessionId Rollback)
802-- | Write the transaction graph to disk. This function can be used to incrementally write new transactions to disk.
803processTransactionGraphPersistence :: PersistenceStrategy -> [TransactionId] -> TransactionGraph -> IO ()
804processTransactionGraphPersistence NoPersistence _ _ = pure ()
805processTransactionGraphPersistence (MinimalPersistence dbdir) transIds graph = transactionGraphPersist NoDiskSync dbdir transIds graph >> pure ()
806processTransactionGraphPersistence (CrashSafePersistence dbdir) transIds graph = transactionGraphPersist FsyncDiskSync dbdir transIds graph >> pure ()
808readGraphTransactionIdDigest :: PersistenceStrategy -> IO LockFileHash
809readGraphTransactionIdDigest NoPersistence = error "attempt to read digest from transaction log without persistence enabled"
810readGraphTransactionIdDigest (MinimalPersistence dbdir) = readGraphTransactionIdFileDigest dbdir
811readGraphTransactionIdDigest (CrashSafePersistence dbdir) = readGraphTransactionIdFileDigest dbdir
813-- | 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.
814typeForRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)
815typeForRelationalExpr sessionId conn@(InProcessConnection _) relExpr = atomically $ typeForRelationalExprSTM sessionId conn relExpr
816typeForRelationalExpr sessionId conn@(RemoteProcessConnection _) relExpr = remoteCall conn (ExecuteTypeForRelationalExpr sessionId relExpr)
818typeForRelationalExprSTM :: SessionId -> Connection -> RelationalExpr -> STM (Either RelationalError Relation)
819typeForRelationalExprSTM sessionId (InProcessConnection conf) relExpr = do
820 let sessions = ipSessions conf
821 eSession <- sessionAndSchema sessionId sessions
822 case eSession of
823 Left err -> pure $ Left err
824 Right (session, schema) -> do
825 let processed = if schemaName session == defaultSchemaName then
826 Right relExpr
827 else
828 Schema.processRelationalExprInSchema schema relExpr
829 case processed of
830 Left err -> pure (Left err)
831 Right relExpr' -> pure $ runReader (RE.typeForRelationalExpr relExpr') (RE.mkRelationalExprState (Sess.concreteDatabaseContext session))
833typeForRelationalExprSTM _ _ _ = error "typeForRelationalExprSTM called on non-local connection"
835-- | Return a 'Map' of the database's constraints at the context of the session and connection.
836inclusionDependencies :: SessionId -> Connection -> IO (Either RelationalError InclusionDependencies)
837inclusionDependencies sessionId (InProcessConnection conf) = do
838 let sessions = ipSessions conf
839 atomically $ do
840 eSession <- sessionAndSchema sessionId sessions
841 case eSession of
842 Left err -> pure $ Left err
843 Right (session, schema) -> do
844 let context = Sess.concreteDatabaseContext session
845 if schemaName session == defaultSchemaName then
846 pure $ Right (B.inclusionDependencies context)
847 else
848 pure (Schema.inclusionDependenciesInSchema schema (B.inclusionDependencies context))
850inclusionDependencies sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveInclusionDependencies sessionId)
852typeConstructorMapping :: SessionId -> Connection -> IO (Either RelationalError TypeConstructorMapping)
853typeConstructorMapping sessionId (InProcessConnection conf) = do
854 let sessions = ipSessions conf
855 atomically $ do
856 eSession <- sessionAndSchema sessionId sessions
857 case eSession of
858 Left err -> pure $ Left err
859 Right (session, _) -> --warning, no schema support for typeconstructors
860 pure (Right (B.typeConstructorMapping (Sess.concreteDatabaseContext session)))
861typeConstructorMapping sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveTypeConstructorMapping sessionId)
863-- | 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.
864planForDatabaseContextExpr :: SessionId -> Connection -> DatabaseContextExpr -> IO (Either RelationalError DatabaseContextExpr)
865planForDatabaseContextExpr sessionId (InProcessConnection conf) dbExpr = do
866 let sessions = ipSessions conf
867 atomically $ do
868 eSession <- sessionAndSchema sessionId sessions
869 case eSession of
870 Left err -> pure $ Left err
871 Right (session, _) -> if schemaName session == defaultSchemaName then
872 pure $ evalState (applyStaticDatabaseOptimization dbExpr) (RE.freshDatabaseState (Sess.concreteDatabaseContext session))
873 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
874 pure (Right dbExpr)
876planForDatabaseContextExpr sessionId conn@(RemoteProcessConnection _) dbExpr = remoteCall conn (RetrievePlanForDatabaseContextExpr sessionId dbExpr)
878-- | Return a relation which represents the current state of the global transaction graph. The attributes are
879-- * current- boolean attribute representing whether or not the current session references this transaction
880-- * head- text attribute which is a non-empty 'HeadName' iff the transaction references a head.
881-- * id- id attribute of the transaction
882-- * parents- a relation-valued attribute which contains a relation of transaction ids which are parent transaction to the transaction
883transactionGraphAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
884transactionGraphAsRelation sessionId (InProcessConnection conf) = do
885 let sessions = ipSessions conf
886 tvar = ipTransactionGraph conf
887 atomically $ do
888 eSession <- sessionForSessionId sessionId sessions
889 case eSession of
890 Left err -> pure $ Left err
891 Right session ->
892 graphAsRelation (Sess.disconnectedTransaction session) <$> readTVar tvar
894transactionGraphAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveTransactionGraph sessionId)
896-- | Returns the names and types of the relation variables in the current 'Session'.
897relationVariablesAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
898relationVariablesAsRelation sessionId (InProcessConnection conf) = do
899 let sessions = ipSessions conf
900 atomically $ do
901 eSession <- sessionAndSchema sessionId sessions
902 case eSession of
903 Left err -> pure (Left err)
904 Right (session, schema) -> do
905 let context = Sess.concreteDatabaseContext session
906 if Sess.schemaName session == defaultSchemaName then
907 pure $ R.relationVariablesAsRelation (relationVariables context)
908 else
909 case Schema.relationVariablesInSchema schema context of
910 Left err -> pure (Left err)
911 Right relvars -> pure $ R.relationVariablesAsRelation relvars
913relationVariablesAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveRelationVariableSummary sessionId)
915-- | Returns the transaction id for the connection's disconnected transaction committed parent transaction.
916headTransactionId :: SessionId -> Connection -> IO (Either RelationalError TransactionId)
917headTransactionId sessionId (InProcessConnection conf) = do
918 let sessions = ipSessions conf
919 atomically $ do
920 eSession <- sessionForSessionId sessionId sessions
921 case eSession of
922 Left err -> pure (Left err)
923 Right session -> pure $ Right (Sess.parentId session)
924headTransactionId sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveHeadTransactionId sessionId)
926headNameSTM_ :: SessionId -> Sessions -> TVar TransactionGraph -> STM (Either RelationalError HeadName)
927headNameSTM_ sessionId sessions graphTvar = do
928 graph <- readTVar graphTvar
929 eSession <- sessionForSessionId sessionId sessions
930 case eSession of
931 Left err -> pure (Left err)
932 Right session -> case transactionForId (Sess.parentId session) graph of
933 Left err -> pure (Left err)
934 Right parentTrans -> case headNameForTransaction parentTrans graph of
935 Nothing -> pure (Left UnknownHeadError)
936 Just headName' -> pure (Right headName')
938-- | Returns Just the name of the head of the current disconnected transaction or Nothing.
939headName :: SessionId -> Connection -> IO (Either RelationalError HeadName)
940headName sessionId (InProcessConnection conf) = do
941 let sessions = ipSessions conf
942 graphTvar = ipTransactionGraph conf
943 atomically (headNameSTM_ sessionId sessions graphTvar)
944headName sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteHeadName sessionId)
946-- | Returns a listing of all available atom types.
947atomTypesAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
948atomTypesAsRelation sessionId (InProcessConnection conf) = do
949 let sessions = ipSessions conf
950 atomically $ do
951 eSession <- sessionForSessionId sessionId sessions
952 case eSession of
953 Left err -> pure (Left err)
954 Right session ->
955 case typesAsRelation (B.typeConstructorMapping (Sess.concreteDatabaseContext session)) of
956 Left err -> pure (Left err)
957 Right rel -> pure (Right rel)
958atomTypesAsRelation sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveAtomTypesAsRelation sessionId)
960disconnectedTransactionIsDirty :: SessionId -> Connection -> IO (Either RelationalError Bool)
961disconnectedTransactionIsDirty sessionId (InProcessConnection conf) = do
962 let sessions = ipSessions conf
963 atomically $ do
964 eSession <- sessionForSessionId sessionId sessions
965 case eSession of
966 Left err -> pure (Left err)
967 Right session ->
968 pure (Right (isDirty session))
969disconnectedTransactionIsDirty sessionId conn@(RemoteProcessConnection _) = remoteCall conn (RetrieveSessionIsDirty sessionId)
971--used only for testing- we expect this to throw an exception
972callTestTimeout_ :: SessionId -> Connection -> IO Bool
973callTestTimeout_ _ (InProcessConnection _) = error "bad testing call"
974callTestTimeout_ sessionId conn@(RemoteProcessConnection _) = remoteCall conn (TestTimeout sessionId)
976--used in tests only
977transactionGraph_ :: Connection -> IO TransactionGraph
978transactionGraph_ (InProcessConnection conf) = atomically $ readTVar (ipTransactionGraph conf)
979transactionGraph_ _ = error "remote connection used"
981--used in tests only
982disconnectedTransaction_ :: SessionId -> Connection -> IO DisconnectedTransaction
983disconnectedTransaction_ sessionId (InProcessConnection conf) = do
984 let sessions = ipSessions conf
985 mSession <- atomically $ STMMap.lookup sessionId sessions
986 case mSession of
987 Nothing -> error "No such session"
988 Just (Sess.Session discon _) -> pure discon
989disconnectedTransaction_ _ _= error "remote connection used"
991-- wrap a graph evaluation in file locking
992commitLock_ :: SessionId ->
993 InProcessConnectionConf ->
994 (TransactionGraph ->
995 STM (Either RelationalError (DisconnectedTransaction, TransactionGraph, [TransactionId]))) ->
996 IO (Either RelationalError ())
997commitLock_ sessionId conf stmBlock = do
998 let sessions = ipSessions conf
999 strat = ipPersistenceStrategy conf
1000 mScriptSession = ipScriptSession conf
1001 graphTvar = ipTransactionGraph conf
1002 clientNodes = ipClientNodes conf
1003 mLockFileH = ipLocks conf
1004 lockHandler body = case mLockFileH of
1005 Nothing -> body False
1006 Just (lockFileH, lockMVar) ->
1007 let acquireLocks = do
1008 lastWrittenDigest <- takeMVar lockMVar
1009 lockFile lockFileH WriteLock
1010 latestDigest <- readGraphTransactionIdDigest strat
1011 pure (latestDigest /= lastWrittenDigest)
1012 releaseLocks _ = do
1013 --still holding the lock- get the latest digest
1014 gDigest <- readGraphTransactionIdDigest strat
1015 unlockFile lockFileH
1016 putMVar lockMVar gDigest
1017 in bracket acquireLocks releaseLocks body
1018 manip <- lockHandler $ \dbWrittenByOtherProcess -> atomically $ do
1019 eSession <- sessionForSessionId sessionId sessions
1020 --handle graph update by other process
1021 oldGraph <- readTVar graphTvar
1022 case eSession of
1023 Left err -> pure (Left err)
1024 Right session -> do
1025 let dbdir = case strat of
1026 MinimalPersistence x -> x
1027 CrashSafePersistence x -> x
1028 _ -> error "accessing dbdir on non-persisted connection"
1029 --this should also happen for non-commit expressions
1030 eRefreshedGraph <- if dbWrittenByOtherProcess then
1031 unsafeIOToSTM (transactionGraphLoad dbdir oldGraph mScriptSession)
1032 else
1033 pure (Right oldGraph)
1034 case eRefreshedGraph of
1035 Left err -> pure (Left (DatabaseLoadError err))
1036 Right refreshedGraph -> do
1037 eGraph <- stmBlock refreshedGraph
1038 case eGraph of
1039 Left err -> pure (Left err)
1040 Right (discon', graph', transactionIdsToPersist) -> do
1041 writeTVar graphTvar graph'
1042 let newSession = Session discon' (Sess.schemaName session)
1043 STMMap.insert newSession sessionId sessions
1044 case transactionForId (Sess.parentId session) oldGraph of
1045 Left err -> pure $ Left err
1046 Right previousTrans ->
1047 if not (Prelude.null transactionIdsToPersist) then do
1048 (evaldNots, nodes) <- executeCommitExprSTM_ (Trans.concreteDatabaseContext previousTrans) (Sess.concreteDatabaseContext session) clientNodes
1049 nodesToNotify <- toList (STMSet.stream nodes)
1050 pure $ Right (evaldNots, nodesToNotify, graph', transactionIdsToPersist)
1051 else pure (Right (M.empty, [], graph', []))
1053 --handle notification firing
1054 case manip of
1055 Left err -> pure (Left err)
1056 Right (notsToFire, nodesToNotify, newGraph, transactionIdsToPersist) -> do
1057 --update filesystem database, if necessary
1058 processTransactionGraphPersistence strat transactionIdsToPersist newGraph
1059 sendNotifications nodesToNotify (ipLocalNode conf) notsToFire
1060 pure (Right ())
1062writeDisconAndGraph_ :: TVar TransactionGraph -> SessionId -> Session -> Sessions -> DisconnectedTransaction -> TransactionGraph -> STM ()
1063writeDisconAndGraph_ graphTvar sessionId session sessions discon graph = do
1064 writeTVar graphTvar graph
1065 let newSession = Session discon (Sess.schemaName session)
1066 STMMap.insert newSession sessionId sessions
1069-- | Runs an IO monad, commits the result when the monad returns no errors, otherwise, rolls back the changes and the error.
1070withTransaction :: SessionId -> Connection -> IO (Either RelationalError a) -> IO (Either RelationalError ()) -> IO (Either RelationalError a)
1071withTransaction sessionId conn io successFunc = bracketOnError (pure ()) (const do_rollback) block
1072 where
1073 do_rollback = rollback sessionId conn
1074 block _ = do
1075 eErr <- io
1076 case eErr of
1077 Left err -> do
1078 _ <- do_rollback
1079 pure (Left err)
1080 Right val -> do
1081 eIsDirty <- disconnectedTransactionIsDirty sessionId conn
1082 case eIsDirty of
1083 Left err -> pure (Left err)
1084 Right dirty ->
1085 if dirty then do
1086 res <- successFunc
1087 case res of
1088 Left err -> pure (Left err)
1089 Right _ -> pure (Right val)
1090 else -- no updates executed, so don't create a commit
1091 pure (Right val)