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