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