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