@@ -30,6 +30,9 @@ module ProjectM36.Client
3030 Attribute(..),
3131 attributesFromList,
3232 createNodeId,
33+ createSessionAtCommit,
34+ createSessionAtHead,
35+ closeSession,
3336 TransactionGraphOperator(..),
3437 Atomable,
3538 NodeId(..),
@@ -40,7 +43,6 @@ import qualified ProjectM36.Base as B
4043import ProjectM36.Error
4144import ProjectM36.StaticOptimizer
4245import Control.Monad.State
43−import ProjectM36.Transaction
4446import qualified ProjectM36.RelationalExpression as RE
4547import ProjectM36.TransactionGraph
4648import ProjectM36.TransactionGraph.Persist
@@ -63,6 +65,9 @@ import Control.Concurrent.MVar
6365import qualified Data.Map as M
6466import Control.Distributed.Process.Serializable (Serializable)
6567--import Control.Distributed.Process.Debug
68+import qualified STMContainers.Map as STMMap
69+import ProjectM36.Session
70+import ListT
6671
6772type Hostname = String
6873
@@ -84,9 +89,12 @@ defaultDatabaseName = "base"
8489
8590defaultRemoteConnectionInfo :: ConnectionInfo
8691defaultRemoteConnectionInfo = RemoteProcessConnectionInfo defaultDatabaseName (createNodeId "127.0.0.1" defaultServerPort)
92+
93+type Sessions = STMMap.Map SessionId Session
94+
8795
88−data Connection = InProcessConnection PersistenceStrategy (TVar (DisconnectedTransaction, TransactionGraph)) |
89− RemoteProcessConnection LocalNode ProcessId -- perhaps should reference the local node as well
96+data Connection = InProcessConnection PersistenceStrategy Sessions (TVar TransactionGraph) |
97+ RemoteProcessConnection LocalNode ProcessId
9098
9199data ConnectionError = SetupDatabaseDirectoryError PersistenceError |
92100 IOExceptionError IOException |
@@ -109,16 +117,16 @@ connectProjectM36 :: ConnectionInfo -> IO (Either ConnectionError Connection)
109117--create a new in-memory database/transaction graph
110118connectProjectM36 (InProcessConnectionInfo strat) = do
111119 freshUUID <- nextRandom
112− let bootstrapContext = RE.basicDatabaseContext
113− freshDiscon = newDisconnectedTransaction freshUUID bootstrapContext
120+ let bootstrapContext = RE.basicDatabaseContext
114121 freshGraph = bootstrapTransactionGraph freshUUID bootstrapContext
115122 case strat of
116123 --create date examples graph for now- probably should be empty context in the future
117124 NoPersistence -> do
118− tvar <- newTVarIO (freshDiscon, freshGraph)
119− return $ Right $ InProcessConnection strat tvar
120− MinimalPersistence dbdir -> connectPersistentProjectM36 strat NoDiskSync dbdir freshDiscon freshGraph
121− CrashSafePersistence dbdir -> connectPersistentProjectM36 strat FsyncDiskSync dbdir freshDiscon freshGraph
125+ graphTvar <- newTVarIO freshGraph
126+ sessions <- STMMap.newIO
127+ return $ Right $ InProcessConnection strat sessions graphTvar
128+ MinimalPersistence dbdir -> connectPersistentProjectM36 strat NoDiskSync dbdir freshGraph
129+ CrashSafePersistence dbdir -> connectPersistentProjectM36 strat FsyncDiskSync dbdir freshGraph
122130
123131connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId) = do
124132 connStatus <- newEmptyMVar
@@ -140,14 +148,13 @@ connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId) = do
140148 liftIO $ putMVar connStatus (Right $ RemoteProcessConnection localNode serverProcessId)
141149 status <- takeMVar connStatus
142150 pure status
143−
151+
144152connectPersistentProjectM36 :: PersistenceStrategy ->
145153 DiskSync ->
146154 FilePath ->
147− DisconnectedTransaction ->
148155 TransactionGraph ->
149156 IO (Either ConnectionError Connection)
150−connectPersistentProjectM36 strat sync dbdir freshDiscon freshGraph = do
157+connectPersistentProjectM36 strat sync dbdir freshGraph = do
151158 err <- setupDatabaseDir sync dbdir freshGraph
152159 case err of
153160 Just err' -> return $ Left (SetupDatabaseDirectoryError err')
@@ -156,11 +163,54 @@ connectPersistentProjectM36 strat sync dbdir freshDiscon freshGraph = do
156163 case graph of
157164 Left err' -> return $ Left (SetupDatabaseDirectoryError err')
158165 Right graph' -> do
159− tvar <- newTVarIO (freshDiscon, graph')
160− return $ Right $ InProcessConnection strat tvar
166+ tvarGraph <- newTVarIO graph'
167+ sessions <- STMMap.newIO
168+ return $ Right $ InProcessConnection strat sessions tvarGraph
169+
170+-- | Create a new session at the transaction UUID and return the session's Id.
171+createSessionAtCommit :: UUID -> Connection -> IO (Either RelationalError SessionId)
172+createSessionAtCommit commitUUID conn@(InProcessConnection _ _ _) = do
173+ newSessionId <- nextRandom
174+ atomically $ do
175+ createSessionAtCommit_ commitUUID newSessionId conn
176+createSessionAtCommit uuid conn@(RemoteProcessConnection _ _) = remoteCall conn (CreateSessionAtCommit uuid)
177+
178+createSessionAtCommit_ :: UUID -> SessionId -> Connection -> STM (Either RelationalError SessionId)
179+createSessionAtCommit_ commitUUID newSessionId (InProcessConnection _ sessions graphTvar) = do
180+ graph <- readTVar graphTvar
181+ case transactionForUUID commitUUID graph of
182+ Left err -> pure (Left err)
183+ Right _ -> do
184+ let freshDiscon = DisconnectedTransaction commitUUID RE.basicDatabaseContext
185+ keyDuplication <- STMMap.lookup newSessionId sessions
186+ case keyDuplication of
187+ Just _ -> pure $ Left (SessionIdInUse newSessionId)
188+ Nothing -> do
189+ STMMap.insert (Session freshDiscon) newSessionId sessions
190+ pure $ Right newSessionId
191+createSessionAtCommit_ _ _ (RemoteProcessConnection _ _) = error "createSessionAtCommit_ called on remote connection"
192+
193+createSessionAtHead :: HeadName -> Connection -> IO (Either RelationalError SessionId)
194+createSessionAtHead headn conn@(InProcessConnection _ _ tvar) = do
195+ newSessionId <- nextRandom
196+ atomically $ do
197+ graph <- readTVar tvar
198+ case transactionForHead headn graph of
199+ Nothing -> pure $ Left (NoSuchHeadNameError headn)
200+ Just trans -> createSessionAtCommit_ (transactionUUID trans) newSessionId conn
201+createSessionAtHead headn conn@(RemoteProcessConnection _ _) = remoteCall conn (CreateSessionAtHead headn)
202+
203+-- | Discards a session, eliminating any uncommitted changes present in the session.
204+closeSession :: SessionId -> Connection -> IO ()
205+closeSession sessionId (InProcessConnection _ sessions _) = do
206+ atomically $ do
207+ STMMap.delete sessionId sessions
208+closeSession sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (CloseSession sessionId)
161209
162210close :: Connection -> IO ()
163−close (InProcessConnection _ _) = pure ()
211+close (InProcessConnection _ sessions _) = atomically $ do
212+ traverse_ (\(k,_) -> STMMap.delete k sessions) (STMMap.stream sessions)
213+ pure ()
164214close (RemoteProcessConnection localNode serverProcessId) = do
165215 runProcessResult localNode $ do
166216 call serverProcessId Logout
@@ -174,114 +224,145 @@ runProcessResult localNode proc = do
174224 takeMVar ret
175225
176226remoteCall :: (Serializable a, Serializable b) => Connection -> a -> IO b
177−remoteCall (InProcessConnection _ _) _ = error "remoteCall called on local connection"
227+remoteCall (InProcessConnection _ _ _) _ = error "remoteCall called on local connection"
178228remoteCall (RemoteProcessConnection localNode serverProcessId) arg = runProcessResult localNode $ do
179229 ret <- safeCall serverProcessId arg
180230 case ret of
181231 Left err -> error (show err)
182232 Right ret' -> pure ret'
183233
184−executeRelationalExpr :: Connection -> RelationalExpr -> IO (Either RelationalError Relation)
185−executeRelationalExpr (InProcessConnection _ tvar) expr = atomically $ do
186− ((DisconnectedTransaction _ context), _) <- readTVar tvar
187− return $ evalState (RE.evalRelationalExpr expr) context
188−executeRelationalExpr conn@(RemoteProcessConnection _ _) relExpr = remoteCall conn (ExecuteRelationalExpr relExpr)
234+sessionForSessionId :: SessionId -> Sessions -> STM (Either RelationalError Session)
235+sessionForSessionId sessionId sessions = do
236+ maybeSession <- STMMap.lookup sessionId sessions
237+ pure $ maybe (Left $ NoSuchSession sessionId) Right maybeSession
238+
239+executeRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)
240+executeRelationalExpr sessionId (InProcessConnection _ sessions _) expr = atomically $ do
241+ eSession <- sessionForSessionId sessionId sessions
242+ case eSession of
243+ Left err -> pure $ Left err
244+ Right (Session (DisconnectedTransaction _ context)) -> pure $ evalState (RE.evalRelationalExpr expr) context
245+executeRelationalExpr sessionId conn@(RemoteProcessConnection _ _) relExpr = remoteCall conn (ExecuteRelationalExpr sessionId relExpr)
189246
190−executeDatabaseContextExpr :: Connection -> DatabaseExpr -> IO (Maybe RelationalError)
191−executeDatabaseContextExpr (InProcessConnection _ tvar) expr = atomically $ do
192− ((DisconnectedTransaction parentUUID context), graph) <- readTVar tvar
193− case runState (RE.evalContextExpr expr) context of
194− (Just err,_) -> return $ Just err
195− (Nothing, context') -> do
196− let newDiscon = DisconnectedTransaction parentUUID context'
197− writeTVar tvar (newDiscon, graph)
247+executeDatabaseContextExpr :: SessionId -> Connection -> DatabaseExpr -> IO (Maybe RelationalError)
248+executeDatabaseContextExpr sessionId (InProcessConnection _ sessions _) expr = atomically $ do
249+ eSession <- sessionForSessionId sessionId sessions
250+ case eSession of
251+ Left err -> pure $ Just err
252+ Right session -> case runState (RE.evalContextExpr expr) (sessionContext session) of
253+ (Just err,_) -> return $ Just err
254+ (Nothing, context') -> do
255+ let newDiscon = DisconnectedTransaction (sessionParentUUID session) context'
256+ newSession = Session newDiscon
257+ STMMap.insert newSession sessionId sessions
198258 return Nothing
199−executeDatabaseContextExpr conn@(RemoteProcessConnection _ _) dbExpr = remoteCall conn (ExecuteDatabaseContextExpr dbExpr)
259+executeDatabaseContextExpr sessionId conn@(RemoteProcessConnection _ _) dbExpr = remoteCall conn (ExecuteDatabaseContextExpr sessionId dbExpr)
200260
201−executeGraphExpr :: Connection -> TransactionGraphOperator -> IO (Maybe RelationalError)
202−executeGraphExpr (InProcessConnection strat tvar) graphExpr = do
261+executeGraphExpr :: SessionId -> Connection -> TransactionGraphOperator -> IO (Maybe RelationalError)
262+executeGraphExpr sessionId (InProcessConnection strat sessions graphTvar) graphExpr = do
203263 freshUUID <- nextRandom
204264 manip <- atomically $ do
205− (discon, graph) <- readTVar tvar
206− case evalGraphOp freshUUID discon graph graphExpr of
207− Left err -> return $ Left err
208− Right (discon', graph') -> do
209− writeTVar tvar (discon', graph')
210− return $ Right (discon', graph')
265+ graph <- readTVar graphTvar
266+ eSession <- sessionForSessionId sessionId sessions
267+ case eSession of
268+ Left err -> pure $ Left err
269+ Right (Session discon) -> case evalGraphOp freshUUID discon graph graphExpr of
270+ Left err -> pure $ Left err
271+ Right (discon', graph') -> do
272+ writeTVar graphTvar graph'
273+ let newSession = Session discon'
274+ STMMap.insert newSession sessionId sessions
275+ pure $ Right graph'
211276 case manip of
212277 Left err -> return $ Just err
213− Right (_, newGraph) -> do
278+ Right newGraph -> do
214279 --update filesystem database, if necessary
215280 --this should really grab a lock at the beginning of the method to be threadsafe
216281 processPersistence strat newGraph
217282 return Nothing
218−executeGraphExpr conn@(RemoteProcessConnection _ _) graphExpr = remoteCall conn (ExecuteGraphExpr graphExpr)
283+executeGraphExpr sessionId conn@(RemoteProcessConnection _ _) graphExpr = remoteCall conn (ExecuteGraphExpr sessionId graphExpr)
219284
220−commit :: Connection -> IO (Maybe RelationalError)
221−commit conn@(InProcessConnection _ _) = executeGraphExpr conn Commit
222−commit conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteGraphExpr Commit)
285+commit :: SessionId -> Connection -> IO (Maybe RelationalError)
286+commit sessionId conn@(InProcessConnection _ _ _) = executeGraphExpr sessionId conn Commit
287+commit sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteGraphExpr sessionId Commit)
223288
224−rollback :: Connection -> IO (Maybe RelationalError)
225−rollback conn@(InProcessConnection _ _) = executeGraphExpr conn Rollback
226−rollback conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteGraphExpr Rollback)
289+rollback :: SessionId -> Connection -> IO (Maybe RelationalError)
290+rollback sessionId conn@(InProcessConnection _ _ _) = executeGraphExpr sessionId conn Rollback
291+rollback sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteGraphExpr sessionId Rollback)
227292
228293processPersistence :: PersistenceStrategy -> TransactionGraph -> IO ()
229294processPersistence NoPersistence _ = return ()
230295processPersistence (MinimalPersistence dbdir) graph = transactionGraphPersist NoDiskSync dbdir graph
231296processPersistence (CrashSafePersistence dbdir) graph = transactionGraphPersist FsyncDiskSync dbdir graph
232297
233−typeForRelationalExpr :: Connection -> RelationalExpr -> IO (Either RelationalError Relation)
234−typeForRelationalExpr conn@(InProcessConnection _ _) relExpr = atomically $ typeForRelationalExprSTM conn relExpr
235−typeForRelationalExpr conn@(RemoteProcessConnection _ _) relExpr = remoteCall conn (ExecuteTypeForRelationalExpr relExpr)
298+typeForRelationalExpr :: SessionId -> Connection -> RelationalExpr -> IO (Either RelationalError Relation)
299+typeForRelationalExpr sessionId conn@(InProcessConnection _ _ _) relExpr = atomically $ typeForRelationalExprSTM sessionId conn relExpr
300+typeForRelationalExpr sessionId conn@(RemoteProcessConnection _ _) relExpr = remoteCall conn (ExecuteTypeForRelationalExpr sessionId relExpr)
236301
237−typeForRelationalExprSTM :: Connection -> RelationalExpr -> STM (Either RelationalError Relation)
238−typeForRelationalExprSTM (InProcessConnection _ tvar) relExpr = do
239− (DisconnectedTransaction _ context, _) <- readTVar tvar
240− pure $ evalState (RE.typeForRelationalExpr relExpr) context
241−typeForRelationalExprSTM _ _ = error "typeForRelationalExprSTM called on non-local connection"
302+typeForRelationalExprSTM :: SessionId -> Connection -> RelationalExpr -> STM (Either RelationalError Relation)
303+typeForRelationalExprSTM sessionId (InProcessConnection _ sessions _) relExpr = do
304+ eSession <- sessionForSessionId sessionId sessions
305+ case eSession of
306+ Left err -> pure $ Left err
307+ Right session -> pure $ evalState (RE.typeForRelationalExpr relExpr) (sessionContext session)
308+
309+typeForRelationalExprSTM _ _ _ = error "typeForRelationalExprSTM called on non-local connection"
242310
243−inclusionDependencies :: Connection -> IO (M.Map IncDepName InclusionDependency)
244−inclusionDependencies (InProcessConnection _ tvar) = do
311+inclusionDependencies :: SessionId -> Connection -> IO (Either RelationalError (M.Map IncDepName InclusionDependency))
312+inclusionDependencies sessionId (InProcessConnection _ sessions _) = do
245313 atomically $ do
246− ((DisconnectedTransaction _ context), _) <- readTVar tvar
247− pure (B.inclusionDependencies context)
248−inclusionDependencies conn@(RemoteProcessConnection _ _) = remoteCall conn RetrieveInclusionDependencies
314+ eSession <- sessionForSessionId sessionId sessions
315+ case eSession of
316+ Left err -> pure $ Left err
317+ Right session -> pure $ Right (B.inclusionDependencies (sessionContext session))
318+
319+inclusionDependencies sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveInclusionDependencies sessionId)
249320
250321
251−planForDatabaseContextExpr :: Connection -> DatabaseExpr -> IO (Either RelationalError DatabaseExpr)
252−planForDatabaseContextExpr (InProcessConnection _ tvar) dbExpr = do
322+planForDatabaseContextExpr :: SessionId -> Connection -> DatabaseExpr -> IO (Either RelationalError DatabaseExpr)
323+planForDatabaseContextExpr sessionId (InProcessConnection _ sessions _) dbExpr = do
253324 atomically $ do
254− ((DisconnectedTransaction _ context), _) <- readTVar tvar
255− pure $ evalState (applyStaticDatabaseOptimization dbExpr) context
256−planForDatabaseContextExpr conn@(RemoteProcessConnection _ _) dbExpr = remoteCall conn (RetrievePlanForDatabaseContextExpr dbExpr)
325+ eSession <- sessionForSessionId sessionId sessions
326+ case eSession of
327+ Left err -> pure $ Left err
328+ Right session -> pure $ evalState (applyStaticDatabaseOptimization dbExpr) (sessionContext session)
329+planForDatabaseContextExpr sessionId conn@(RemoteProcessConnection _ _) dbExpr = remoteCall conn (RetrievePlanForDatabaseContextExpr sessionId dbExpr)
257330
258−transactionGraphAsRelation :: Connection -> IO (Either RelationalError Relation)
259−transactionGraphAsRelation (InProcessConnection _ tvar) = do
331+transactionGraphAsRelation :: SessionId -> Connection -> IO (Either RelationalError Relation)
332+transactionGraphAsRelation sessionId (InProcessConnection _ sessions tvar) = do
260333 atomically $ do
261− (discon, graph) <- readTVar tvar
262− pure $ graphAsRelation discon graph
263−transactionGraphAsRelation conn@(RemoteProcessConnection _ _) = remoteCall conn RetrieveTransactionGraph
334+ eSession <- sessionForSessionId sessionId sessions
335+ case eSession of
336+ Left err -> pure $ Left err
337+ Right (Session discon) -> do
338+ graph <- readTVar tvar
339+ pure $ graphAsRelation discon graph
340+
341+transactionGraphAsRelation sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveTransactionGraph sessionId)
264342
265343-- | Returns the UUID for the connection's disconnected transaction committed parent transaction.
266−headTransactionUUID :: Connection -> IO (UUID)
267−headTransactionUUID (InProcessConnection _ tvar) = do
344+headTransactionUUID :: SessionId -> Connection -> IO (Maybe UUID)
345+headTransactionUUID sessionId (InProcessConnection _ sessions _) = do
268346 atomically $ do
269− (DisconnectedTransaction parentUUID _, _) <- readTVar tvar
270− pure parentUUID
271−headTransactionUUID conn@(RemoteProcessConnection _ _) = remoteCall conn RetrieveHeadTransactionUUID
347+ eSession <- sessionForSessionId sessionId sessions
348+ case eSession of
349+ Left _ -> pure Nothing
350+ Right session -> pure $ Just (sessionParentUUID session)
351+headTransactionUUID sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (RetrieveHeadTransactionUUID sessionId)
272352
273353-- | Returns Just the name of the head of the current disconnected transaction or Nothing.
274−headName :: Connection -> IO (Maybe HeadName)
275−headName (InProcessConnection _ tvar) = do
354+headName :: SessionId -> Connection -> IO (Maybe HeadName)
355+headName sessionId (InProcessConnection _ sessions graphTvar) = do
276356 atomically $ do
277− (DisconnectedTransaction parentUUID _, graph) <- readTVar tvar
278− pure $ case transactionForUUID parentUUID graph of
279− Left _ -> Nothing
280− Right parentTrans -> headNameForTransaction parentTrans graph
281−headName (RemoteProcessConnection localNode serverProcessId) = do
282− runProcessResult localNode $ do
283− ret <- call serverProcessId ExecuteHeadName
284− pure ret
357+ graph <- readTVar graphTvar
358+ eSession <- sessionForSessionId sessionId sessions
359+ case eSession of
360+ Left _ -> pure $ Nothing
361+ Right session -> pure $ case transactionForUUID (sessionParentUUID session) graph of
362+ Left _ -> Nothing
363+ Right parentTrans -> headNameForTransaction parentTrans graph
364+headName sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteHeadName sessionId)
365+
285366
286367
287368