git.iko.soy/project-m36
on 4e1d107 help sign in
project-m36/ProjectM36/Client.hs
279 lines 13 kB history raw
1{-# LANGUAGE OverloadedStrings #-}
2module ProjectM36.Client
3 (ConnectionInfo(..),
4 Connection(..),
5 Port,
6 Hostname,
7 DatabaseName,
8 ConnectionError,
9 connectProjectM36,
10 close,
11 executeRelationalExpr,
12 executeDatabaseContextExpr,
13 executeGraphExpr,
14 commit,
15 rollback,
16 typeForRelationalExpr,
17 inclusionDependencies,
18 planForDatabaseContextExpr,
19 processPersistence,
20 transactionGraphAsRelation,
21 headName,
22 remoteDBLookupName,
23 defaultServerPort,
24 headTransactionUUID,
25 PersistenceStrategy(..),
26 RelationalExpr(..),
27 DatabaseExpr(..),
28 Attribute(..),
29 attributesFromList,
30 createNodeId,
31 TransactionGraphOperator(..),
32 Atomable,
33 NodeId(..),
34 Atom(..),
35 AtomType(..)) where
36import ProjectM36.Base hiding (inclusionDependencies) --defined in this module as well
37import qualified ProjectM36.Base as B
38import ProjectM36.Error
39import ProjectM36.StaticOptimizer
40import Control.Monad.State
41import ProjectM36.Transaction
42import qualified ProjectM36.RelationalExpression as RE
43import ProjectM36.TransactionGraph
44import ProjectM36.TransactionGraph.Persist
45import ProjectM36.Attribute
46import ProjectM36.Persist (DiskSync(..))
47import ProjectM36.Daemon.RemoteCallTypes
48import Network.Transport.TCP (createTransport, defaultTCPParameters, encodeEndPointAddress)
49import Control.Distributed.Process.Node (newLocalNode, initRemoteTable, runProcess, LocalNode)
50import Control.Distributed.Process.Extras.Internal.Types (whereisRemote)
51import Control.Distributed.Process.ManagedProcess.Client (call, safeCall)
52import Control.Distributed.Process (NodeId(..))
54import Data.UUID (UUID)
55import Data.UUID.V4 (nextRandom)
56import Control.Concurrent.STM
57import Data.Word
58import Control.Distributed.Process (ProcessId, Process)
59import Control.Exception (IOException)
60import Control.Concurrent.MVar
61import qualified Data.Map as M
62import Control.Distributed.Process.Serializable (Serializable)
63--import Control.Distributed.Process.Debug
65type Hostname = String
67type Port = Word16
69type DatabaseName = String
71data ConnectionInfo = InProcessConnectionInfo PersistenceStrategy |
72 RemoteProcessConnectionInfo DatabaseName NodeId
74createNodeId :: Hostname -> Port -> NodeId
75createNodeId host port = NodeId $ encodeEndPointAddress host (show port) 0
77defaultServerPort :: Port
78defaultServerPort = 6543
80data Connection = InProcessConnection PersistenceStrategy (TVar (DisconnectedTransaction, TransactionGraph)) |
81 RemoteProcessConnection LocalNode ProcessId -- perhaps should reference the local node as well
83data ConnectionError = SetupDatabaseDirectoryError PersistenceError |
84 IOExceptionError IOException |
85 NoSuchDatabaseByNameError DatabaseName |
86 LoginError
87 deriving (Show, Eq)
89remoteDBLookupName :: DatabaseName -> String
90remoteDBLookupName = (++) "db-"
92commonLocalNode :: IO (Either ConnectionError LocalNode)
93commonLocalNode = do
94 eLocalTransport <- createTransport "127.0.0.1" "0" defaultTCPParameters
95 case eLocalTransport of
96 Left err -> pure (Left $ IOExceptionError err)
97 Right localTransport -> newLocalNode localTransport initRemoteTable >>= pure . Right
99-- connect to an in-process database
100connectProjectM36 :: ConnectionInfo -> IO (Either ConnectionError Connection)
101--create a new in-memory database/transaction graph
102connectProjectM36 (InProcessConnectionInfo strat) = do
103 freshUUID <- nextRandom
104 let bootstrapContext = RE.basicDatabaseContext
105 freshDiscon = newDisconnectedTransaction freshUUID bootstrapContext
106 freshGraph = bootstrapTransactionGraph freshUUID bootstrapContext
107 case strat of
108 --create date examples graph for now- probably should be empty context in the future
109 NoPersistence -> do
110 tvar <- newTVarIO (freshDiscon, freshGraph)
111 return $ Right $ InProcessConnection strat tvar
112 MinimalPersistence dbdir -> connectPersistentProjectM36 strat NoDiskSync dbdir freshDiscon freshGraph
113 CrashSafePersistence dbdir -> connectPersistentProjectM36 strat FsyncDiskSync dbdir freshDiscon freshGraph
115connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId) = do
116 connStatus <- newEmptyMVar
117 eLocalNode <- commonLocalNode
118 let dbName = remoteDBLookupName databaseName
119 putStrLn $ show serverNodeId ++ " " ++ dbName
120 case eLocalNode of
121 Left err -> pure (Left err)
122 Right localNode -> do
123 runProcess localNode $ do
124 mServerProcessId <- whereisRemote serverNodeId dbName
125 case mServerProcessId of
126 Nothing -> liftIO $ putMVar connStatus $ Left (NoSuchDatabaseByNameError databaseName)
127 Just serverProcessId -> do
128 loginConfirmation <- call serverProcessId Login
129 if not loginConfirmation then
130 liftIO $ putMVar connStatus (Left LoginError)
131 else do
132 liftIO $ putMVar connStatus (Right $ RemoteProcessConnection localNode serverProcessId)
133 status <- takeMVar connStatus
134 pure status
136connectPersistentProjectM36 :: PersistenceStrategy ->
137 DiskSync ->
138 FilePath ->
139 DisconnectedTransaction ->
140 TransactionGraph ->
141 IO (Either ConnectionError Connection)
142connectPersistentProjectM36 strat sync dbdir freshDiscon freshGraph = do
143 err <- setupDatabaseDir sync dbdir freshGraph
144 case err of
145 Just err' -> return $ Left (SetupDatabaseDirectoryError err')
146 Nothing -> do
147 graph <- transactionGraphLoad dbdir emptyTransactionGraph
148 case graph of
149 Left err' -> return $ Left (SetupDatabaseDirectoryError err')
150 Right graph' -> do
151 tvar <- newTVarIO (freshDiscon, graph')
152 return $ Right $ InProcessConnection strat tvar
154close :: Connection -> IO ()
155close (InProcessConnection _ _) = pure ()
156close (RemoteProcessConnection localNode serverProcessId) = do
157 runProcessResult localNode $ do
158 call serverProcessId Logout
160runProcessResult :: LocalNode -> Process a -> IO a
161runProcessResult localNode proc = do
162 ret <- newEmptyMVar
163 runProcess localNode $ do
164 val <- proc
165 liftIO $ putMVar ret val
166 takeMVar ret
168remoteCall :: (Serializable a, Serializable b) => Connection -> a -> IO b
169remoteCall (InProcessConnection _ _) _ = error "remoteCall called on local connection"
170remoteCall (RemoteProcessConnection localNode serverProcessId) arg = runProcessResult localNode $ do
171 ret <- safeCall serverProcessId arg
172 case ret of
173 Left err -> error (show err)
174 Right ret' -> pure ret'
176executeRelationalExpr :: Connection -> RelationalExpr -> IO (Either RelationalError Relation)
177executeRelationalExpr (InProcessConnection _ tvar) expr = atomically $ do
178 ((DisconnectedTransaction _ context), _) <- readTVar tvar
179 return $ evalState (RE.evalRelationalExpr expr) context
180executeRelationalExpr conn@(RemoteProcessConnection _ _) relExpr = remoteCall conn (ExecuteRelationalExpr relExpr)
182executeDatabaseContextExpr :: Connection -> DatabaseExpr -> IO (Maybe RelationalError)
183executeDatabaseContextExpr (InProcessConnection _ tvar) expr = atomically $ do
184 ((DisconnectedTransaction parentUUID context), graph) <- readTVar tvar
185 case runState (RE.evalContextExpr expr) context of
186 (Just err,_) -> return $ Just err
187 (Nothing, context') -> do
188 let newDiscon = DisconnectedTransaction parentUUID context'
189 writeTVar tvar (newDiscon, graph)
190 return Nothing
191executeDatabaseContextExpr conn@(RemoteProcessConnection _ _) dbExpr = remoteCall conn (ExecuteDatabaseContextExpr dbExpr)
193executeGraphExpr :: Connection -> TransactionGraphOperator -> IO (Maybe RelationalError)
194executeGraphExpr (InProcessConnection strat tvar) graphExpr = do
195 freshUUID <- nextRandom
196 manip <- atomically $ do
197 (discon, graph) <- readTVar tvar
198 case evalGraphOp freshUUID discon graph graphExpr of
199 Left err -> return $ Left err
200 Right (discon', graph') -> do
201 writeTVar tvar (discon', graph')
202 return $ Right (discon', graph')
203 case manip of
204 Left err -> return $ Just err
205 Right (_, newGraph) -> do
206 --update filesystem database, if necessary
207 --this should really grab a lock at the beginning of the method to be threadsafe
208 processPersistence strat newGraph
209 return Nothing
210executeGraphExpr conn@(RemoteProcessConnection _ _) graphExpr = remoteCall conn (ExecuteGraphExpr graphExpr)
212commit :: Connection -> IO (Maybe RelationalError)
213commit conn@(InProcessConnection _ _) = executeGraphExpr conn Commit
214commit conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteGraphExpr Commit)
216rollback :: Connection -> IO (Maybe RelationalError)
217rollback conn@(InProcessConnection _ _) = executeGraphExpr conn Rollback
218rollback conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteGraphExpr Rollback)
220processPersistence :: PersistenceStrategy -> TransactionGraph -> IO ()
221processPersistence NoPersistence _ = return ()
222processPersistence (MinimalPersistence dbdir) graph = transactionGraphPersist NoDiskSync dbdir graph
223processPersistence (CrashSafePersistence dbdir) graph = transactionGraphPersist FsyncDiskSync dbdir graph
225typeForRelationalExpr :: Connection -> RelationalExpr -> IO (Either RelationalError Relation)
226typeForRelationalExpr conn@(InProcessConnection _ _) relExpr = atomically $ typeForRelationalExprSTM conn relExpr
227typeForRelationalExpr conn@(RemoteProcessConnection _ _) relExpr = remoteCall conn (ExecuteTypeForRelationalExpr relExpr)
229typeForRelationalExprSTM :: Connection -> RelationalExpr -> STM (Either RelationalError Relation)
230typeForRelationalExprSTM (InProcessConnection _ tvar) relExpr = do
231 (DisconnectedTransaction _ context, _) <- readTVar tvar
232 pure $ evalState (RE.typeForRelationalExpr relExpr) context
233typeForRelationalExprSTM _ _ = error "typeForRelationalExprSTM called on non-local connection"
235inclusionDependencies :: Connection -> IO (M.Map IncDepName InclusionDependency)
236inclusionDependencies (InProcessConnection _ tvar) = do
237 atomically $ do
238 ((DisconnectedTransaction _ context), _) <- readTVar tvar
239 pure (B.inclusionDependencies context)
240inclusionDependencies conn@(RemoteProcessConnection _ _) = remoteCall conn RetrieveInclusionDependencies
243planForDatabaseContextExpr :: Connection -> DatabaseExpr -> IO (Either RelationalError DatabaseExpr)
244planForDatabaseContextExpr (InProcessConnection _ tvar) dbExpr = do
245 atomically $ do
246 ((DisconnectedTransaction _ context), _) <- readTVar tvar
247 pure $ evalState (applyStaticDatabaseOptimization dbExpr) context
248planForDatabaseContextExpr conn@(RemoteProcessConnection _ _) dbExpr = remoteCall conn (RetrievePlanForDatabaseContextExpr dbExpr)
250transactionGraphAsRelation :: Connection -> IO (Either RelationalError Relation)
251transactionGraphAsRelation (InProcessConnection _ tvar) = do
252 atomically $ do
253 (discon, graph) <- readTVar tvar
254 pure $ graphAsRelation discon graph
255transactionGraphAsRelation conn@(RemoteProcessConnection _ _) = remoteCall conn RetrieveTransactionGraph
257-- | Returns the UUID for the connection's disconnected transaction committed parent transaction.
258headTransactionUUID :: Connection -> IO (UUID)
259headTransactionUUID (InProcessConnection _ tvar) = do
260 atomically $ do
261 (DisconnectedTransaction parentUUID _, _) <- readTVar tvar
262 pure parentUUID
263headTransactionUUID conn@(RemoteProcessConnection _ _) = remoteCall conn RetrieveHeadTransactionUUID
265-- | Returns Just the name of the head of the current disconnected transaction or Nothing.
266headName :: Connection -> IO (Maybe HeadName)
267headName (InProcessConnection _ tvar) = do
268 atomically $ do
269 (DisconnectedTransaction parentUUID _, graph) <- readTVar tvar
270 pure $ case transactionForUUID parentUUID graph of
271 Left _ -> Nothing
272 Right parentTrans -> headNameForTransaction parentTrans graph
273headName (RemoteProcessConnection localNode serverProcessId) = do
274 runProcessResult localNode $ do
275 ret <- call serverProcessId ExecuteHeadName
276 pure ret