a011cd3 · project-m36 · git.iko.soy

git.iko.soy/project-m36

on master help sign in

WIP notification listener working

a011cd3AgentM11 years ago

ProjectM36/Base.hs+5

compare with master

unifiedsplit

@@ -279,8 +279,13 @@ data DatabaseExpr where

279 Insert :: RelVarName -> RelationalExpr -> DatabaseExpr

280 Delete :: RelVarName -> RestrictionPredicateExpr -> DatabaseExpr

281 Update :: RelVarName -> M.Map AttributeName Atom -> RestrictionPredicateExpr -> DatabaseExpr -- needs restriction support

282+

282 AddInclusionDependency :: IncDepName -> InclusionDependency -> DatabaseExpr

283 AddInclusionDependency :: IncDepName -> InclusionDependency -> DatabaseExpr

283 RemoveInclusionDependency :: IncDepName -> DatabaseExpr

284 RemoveInclusionDependency :: IncDepName -> DatabaseExpr

285+

286+ AddNotification :: NotificationName -> RelationalExpr -> RelationalExpr -> DatabaseExpr

287+ RemoveNotification :: NotificationName -> DatabaseExpr

288+

284 MultipleExpr :: [DatabaseExpr] -> DatabaseExpr

289 MultipleExpr :: [DatabaseExpr] -> DatabaseExpr

285 deriving (Show, Eq, Binary, Generic)

290 deriving (Show, Eq, Binary, Generic)

286

291

ProjectM36/Client.hs+54−48

@@ -55,7 +55,7 @@ import ProjectM36.Persist (DiskSync(..))

55import ProjectM36.Notifications

56import ProjectM36.Server.RemoteCallTypes

57import Network.Transport.TCP (createTransport, defaultTCPParameters, encodeEndPointAddress)

58−import Control.Distributed.Process.Node (newLocalNode, initRemoteTable, runProcess, LocalNode)

58+import Control.Distributed.Process.Node (newLocalNode, initRemoteTable, runProcess, LocalNode, forkProcess)

59import Control.Distributed.Process.Extras.Internal.Types (whereisRemote)

60import Control.Distributed.Process.ManagedProcess.Client (call, safeCall)

61import Control.Distributed.Process (NodeId(..), getSelfPid)

@@ -64,7 +64,7 @@ import Data.UUID (UUID)

64import Data.UUID.V4 (nextRandom)

65import Control.Concurrent.STM

66import Data.Word

67−import Control.Distributed.Process (ProcessId, Process, receiveWait, send, match)

67+import Control.Distributed.Process (ProcessId, Process, receiveWait, send, match, say)

68import Control.Exception (IOException)

69import Control.Concurrent.MVar

70import qualified Data.Map as M

@@ -77,6 +77,7 @@ import ProjectM36.Sessions

77import ListT

78import Data.Binary (Binary)

79import GHC.Generics (Generic)

80+--import Debug.Trace

81type Hostname = String

82type Hostname = String

@@ -142,17 +143,20 @@ commonLocalNode = do

142

143

143notificationListener :: Process ()

144notificationListener :: Process ()

144notificationListener = do

145notificationListener = do

146+ pid <- getSelfPid

147+ liftIO $ putStrLn $ "LISTENER THREAD START " ++ show pid

148+ _ <- forever $ do

145− _ <- receiveWait [

149+ receiveWait [

146− match (\(NotificationMessage eNots) -> undefined)

150+ match (\(NotificationMessage eNots) -> say $ "NOTIFICATION: " ++ show eNots)

147− ]

151+ ]

148− liftIO $ putStrLn $ "LISTENER THREAD EXIT"

152+ say "LISTENER THREAD EXIT"

149 pure ()

153 pure ()

150

154

151startNotificationListener :: IO (ProcessId)

155startNotificationListener :: IO (ProcessId)

152startNotificationListener = do

156startNotificationListener = do

153 eLocalNode <- commonLocalNode

157 eLocalNode <- commonLocalNode

154 case eLocalNode of

158 case eLocalNode of

155− Left err -> error "Failed to start local notification listener"

159+ Left err -> error ("Failed to start local notification listener: " ++ show err)

156 Right localNode -> forkProcess localNode notificationListener

160 Right localNode -> forkProcess localNode notificationListener

157

161

158-- | To create a 'Connection' to a remote or local database, create a connectionInfo and call 'connectProjectM36'.

162-- | To create a 'Connection' to a remote or local database, create a connectionInfo and call 'connectProjectM36'.

@@ -168,7 +172,7 @@ connectProjectM36 (InProcessConnectionInfo strat) = do

168 graphTvar <- newTVarIO freshGraph

172 graphTvar <- newTVarIO freshGraph

169 clientNodes <- STMSet.newIO

173 clientNodes <- STMSet.newIO

170 sessions <- STMMap.newIO

174 sessions <- STMMap.newIO

171− notProcId <- startNotificationListener

175+ --notificationPid <- startNotificationListener

172 return $ Right $ InProcessConnection strat clientNodes sessions graphTvar

176 return $ Right $ InProcessConnection strat clientNodes sessions graphTvar

173 MinimalPersistence dbdir -> connectPersistentProjectM36 strat NoDiskSync dbdir freshGraph

177 MinimalPersistence dbdir -> connectPersistentProjectM36 strat NoDiskSync dbdir freshGraph

174 CrashSafePersistence dbdir -> connectPersistentProjectM36 strat FsyncDiskSync dbdir freshGraph

178 CrashSafePersistence dbdir -> connectPersistentProjectM36 strat FsyncDiskSync dbdir freshGraph

@@ -176,6 +180,7 @@ connectProjectM36 (InProcessConnectionInfo strat) = do

176connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId) = do

180connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId) = do

177 connStatus <- newEmptyMVar

181 connStatus <- newEmptyMVar

178 eLocalNode <- commonLocalNode

182 eLocalNode <- commonLocalNode

183+ notificationListenerPid <- startNotificationListener

179 let dbName = remoteDBLookupName databaseName

184 let dbName = remoteDBLookupName databaseName

180 putStrLn $ show serverNodeId ++ " " ++ dbName

185 putStrLn $ show serverNodeId ++ " " ++ dbName

181 case eLocalNode of

186 case eLocalNode of

@@ -183,11 +188,10 @@ connectProjectM36 (RemoteProcessConnectionInfo databaseName serverNodeId) = do

183 Right localNode -> do

188 Right localNode -> do

184 runProcess localNode $ do

189 runProcess localNode $ do

185 mServerProcessId <- whereisRemote serverNodeId dbName

190 mServerProcessId <- whereisRemote serverNodeId dbName

186− selfPid <- getSelfPid

187 case mServerProcessId of

191 case mServerProcessId of

188 Nothing -> liftIO $ putMVar connStatus $ Left (NoSuchDatabaseByNameError databaseName)

192 Nothing -> liftIO $ putMVar connStatus $ Left (NoSuchDatabaseByNameError databaseName)

189 Just serverProcessId -> do

193 Just serverProcessId -> do

190− loginConfirmation <- call serverProcessId (Login selfPid)

194+ loginConfirmation <- call serverProcessId (Login notificationListenerPid)

191 if not loginConfirmation then

195 if not loginConfirmation then

192 liftIO $ putMVar connStatus (Left LoginError)

196 liftIO $ putMVar connStatus (Left LoginError)

193 else do

197 else do

@@ -315,9 +319,9 @@ executeDatabaseContextExpr sessionId conn@(RemoteProcessConnection _ _) dbExpr =

315

319

316executeGraphExprSTM_ :: UUID -> SessionId -> Sessions -> TransactionGraphOperator -> TVar TransactionGraph -> STM (Either RelationalError TransactionGraph)

320executeGraphExprSTM_ :: UUID -> SessionId -> Sessions -> TransactionGraphOperator -> TVar TransactionGraph -> STM (Either RelationalError TransactionGraph)

317executeGraphExprSTM_ freshUUID sessionId sessions graphExpr graphTvar = do

321executeGraphExprSTM_ freshUUID sessionId sessions graphExpr graphTvar = do

318− graph <- readTVar graphTvar

322+ graph <- readTVar graphTvar

319− eSession <- sessionForSessionId sessionId sessions

323+ eSession <- sessionForSessionId sessionId sessions

320− case eSession of

324+ case eSession of

321 Left err -> pure $ Left err

325 Left err -> pure $ Left err

322 Right (Session discon) -> case evalGraphOp freshUUID discon graph graphExpr of

326 Right (Session discon) -> case evalGraphOp freshUUID discon graph graphExpr of

323 Left err -> pure $ Left err

327 Left err -> pure $ Left err

@@ -327,61 +331,63 @@ executeGraphExprSTM_ freshUUID sessionId sessions graphExpr graphTvar = do

327 STMMap.insert newSession sessionId sessions

331 STMMap.insert newSession sessionId sessions

328 pure $ Right graph'

332 pure $ Right graph'

329

333

330−

334+-- process notifications for commits

335+executeCommitExprSTM_ :: DatabaseContext -> DatabaseContext -> ClientNodes -> STM (EvaluatedNotifications, ClientNodes)

336+executeCommitExprSTM_ oldContext newContext nodes = do

337+ let nots = notifications oldContext

338+ fireNots = notificationChanges nots oldContext newContext

339+ evaldNots = M.map mkEvaldNot fireNots

340+ mkEvaldNot notif = EvaluatedNotification { notification = notif, reportRelation = evalState (RE.evalRelationalExpr (reportExpr notif)) oldContext }

341+ pure (evaldNots, nodes)

342+

343+

331-- | Execute a transaction graph expression in the context of the session and connection. Transaction graph operators modify the transaction graph state.

344-- | Execute a transaction graph expression in the context of the session and connection. Transaction graph operators modify the transaction graph state.

332executeGraphExpr :: SessionId -> Connection -> TransactionGraphOperator -> IO (Maybe RelationalError)

345executeGraphExpr :: SessionId -> Connection -> TransactionGraphOperator -> IO (Maybe RelationalError)

333−executeGraphExpr sessionId (InProcessConnection strat _ sessions graphTvar) graphExpr = do

346+executeGraphExpr sessionId (InProcessConnection strat clientNodes sessions graphTvar) graphExpr = do

334 freshUUID <- nextRandom

347 freshUUID <- nextRandom

348+ manip <- atomically $ do

349+ eSession <- sessionForSessionId sessionId sessions

350+ oldGraph <- readTVar graphTvar

351+ case eSession of

352+ Left err -> pure (Left err)

353+ Right (Session (DisconnectedTransaction parentUUID currentContext)) -> do

335− manip <- atomically $ executeGraphExprSTM_ freshUUID sessionId sessions graphExpr graphTvar

354+ eGraph <- executeGraphExprSTM_ freshUUID sessionId sessions graphExpr graphTvar

355+ case eGraph of

356+ Left err -> pure (Left err)

357+ Right newGraph -> do

358+ if graphExpr == Commit then

359+ case transactionForUUID parentUUID oldGraph of

360+ Left err -> pure $ Left err

361+ Right (Transaction _ _ previousContext) -> do

362+ (evaldNots, nodes) <- executeCommitExprSTM_ previousContext currentContext clientNodes

363+ nodesToNotify <- toList (STMSet.stream nodes)

364+ pure $ Right (evaldNots, nodesToNotify, newGraph)

365+ else

366+ pure $ Right (M.empty, [], newGraph)

336 case manip of

367 case manip of

337 Left err -> return $ Just err

368 Left err -> return $ Just err

338− Right newGraph -> do

369+ Right (notsToFire, nodesToNotify, newGraph) -> do

339 --update filesystem database, if necessary

370 --update filesystem database, if necessary

340 --this should really grab a lock at the beginning of the method to be threadsafe

371 --this should really grab a lock at the beginning of the method to be threadsafe

341 processPersistence strat newGraph

372 processPersistence strat newGraph

373+ sendNotifications nodesToNotify notsToFire

342 return Nothing

374 return Nothing

343executeGraphExpr sessionId conn@(RemoteProcessConnection _ _) graphExpr = remoteCall conn (ExecuteGraphExpr sessionId graphExpr)

375executeGraphExpr sessionId conn@(RemoteProcessConnection _ _) graphExpr = remoteCall conn (ExecuteGraphExpr sessionId graphExpr)

344

376

377+{-

345commitSTM_ :: UUID -> SessionId -> Sessions -> TVar TransactionGraph -> STM (Maybe RelationalError)

378commitSTM_ :: UUID -> SessionId -> Sessions -> TVar TransactionGraph -> STM (Maybe RelationalError)

346commitSTM_ freshUUID sessionId sessions graph = do

379commitSTM_ freshUUID sessionId sessions graph = do

347 mErr <- executeGraphExprSTM_ freshUUID sessionId sessions Commit graph

380 mErr <- executeGraphExprSTM_ freshUUID sessionId sessions Commit graph

348 pure $ case mErr of

381 pure $ case mErr of

349 Left err -> Just err

382 Left err -> Just err

350 Right _ -> Nothing

383 Right _ -> Nothing

384+-}

351

385

352-- | After modifying a session, '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.

386-- | After modifying a session, '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.

353commit :: SessionId -> Connection -> IO (Maybe RelationalError)

387commit :: SessionId -> Connection -> IO (Maybe RelationalError)

354−commit sessionId (InProcessConnection _ clientNodes sessions graphTvar) = do

388+commit sessionId conn@(InProcessConnection _ _ _ _) = executeGraphExpr sessionId conn Commit

355− freshUUID <- nextRandom

356− mNots <- atomically $ do

357− graph <- readTVar graphTvar

358− -- take the notifications of the current context and scan them for changes in the previous -> current contexts

359− eSession <- sessionForSessionId sessionId sessions

360− case eSession of

361− Left err -> pure (Left err)

362− Right (Session (DisconnectedTransaction parentUUID currentContext)) -> do

363− --grab previous context

364− case transactionForUUID parentUUID graph of

365− Left err -> pure $ Left err

366− Right (Transaction _ _ previousContext) -> do

367− let nots = notifications currentContext

368− fireNots = notificationChanges nots previousContext currentContext

369− evaldNots = M.map mkEvaldNot fireNots

370− mkEvaldNot notif = EvaluatedNotification { notification = notif, reportRelation = evalState (RE.evalRelationalExpr (reportExpr notif)) currentContext }

371− mErr <- commitSTM_ freshUUID sessionId sessions graphTvar

372− case mErr of

373− Just err -> pure (Left err)

374− Nothing -> do

375− notifyNodes <- toList (STMSet.stream clientNodes)

376− pure (Right (evaldNots, notifyNodes))

377− case mNots of

378− Left err -> pure (Just err)

379− Right (notsToFire, nodesToNotify) -> do

380− --trigger the notifications on all clients

381− sendNotifications nodesToNotify notsToFire

382− pure Nothing

383commit sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteGraphExpr sessionId Commit)

389commit sessionId conn@(RemoteProcessConnection _ _) = remoteCall conn (ExecuteGraphExpr sessionId Commit)

384−

390+

385sendNotifications :: [ProcessId] -> EvaluatedNotifications -> IO ()

391sendNotifications :: [ProcessId] -> EvaluatedNotifications -> IO ()

386sendNotifications pids nots = mapM_ sendNots pids

392sendNotifications pids nots = mapM_ sendNots pids

387 where

393 where

@@ -389,7 +395,7 @@ sendNotifications pids nots = mapM_ sendNots pids

389 eLocalNode <- commonLocalNode

395 eLocalNode <- commonLocalNode

390 case eLocalNode of

396 case eLocalNode of

391 Left err -> error ("Failed to get local node: " ++ show err)

397 Left err -> error ("Failed to get local node: " ++ show err)

392− Right localNode -> runProcess localNode $ send remoteClientPid (NotificationMessage nots)

398+ Right localNode -> when (not (M.null nots)) $ runProcess localNode $ send remoteClientPid (NotificationMessage nots)

393

399

394-- | Discard any changes made in the current session. This resets the disconnected transaction to reference the original database context of the parent transaction and is a very cheap operation.

400-- | Discard any changes made in the current session. This resets the disconnected transaction to reference the original database context of the parent transaction and is a very cheap operation.

395rollback :: SessionId -> Connection -> IO (Maybe RelationalError)

401rollback :: SessionId -> Connection -> IO (Maybe RelationalError)

ProjectM36/Error.hs+2

@@ -39,6 +39,8 @@ data RelationalError = NoSuchAttributeNamesError (S.Set AttributeName)

39 | AtomTypeCountError [AtomType] [AtomType]

40 | AtomFunctionTypeError AtomFunctionName Int AtomType AtomType --arg number

41 | RelationValuedAttributesNotSupportedError [AttributeName]

42+ | NotificationNameInUseError NotificationName

43+ | NotificationNameNotInUseError NotificationName

42 | ImportError T.Text -- really? This should be broken out into some other error type- this has nothing to do with relational algebra

44 | ImportError T.Text -- really? This should be broken out into some other error type- this has nothing to do with relational algebra

43 | ExportError T.Text

45 | ExportError T.Text

44 deriving (Show,Eq,Generic,Binary,Typeable)

46 deriving (Show,Eq,Generic,Binary,Typeable)

ProjectM36/Notifications.hs+7−2

@@ -3,10 +3,15 @@ import ProjectM36.Base

3import ProjectM36.RelationalExpression

4import Control.Monad.State

5import qualified Data.Map as M

6+import Data.Either (isRight)

7-- | Returns the notifications which should be triggered based on the transition from the first 'DatabaseContext' to the second 'DatabaseContext'.

8-- | Returns the notifications which should be triggered based on the transition from the first 'DatabaseContext' to the second 'DatabaseContext'.

8notificationChanges :: Notifications -> DatabaseContext -> DatabaseContext -> Notifications

9notificationChanges :: Notifications -> DatabaseContext -> DatabaseContext -> Notifications

9notificationChanges nots context1 context2 = M.filter notificationFilter nots

10notificationChanges nots context1 context2 = M.filter notificationFilter nots

10 where

11 where

11− notificationFilter (Notification chExpr _) = evalChangeExpr chExpr context1 /= evalChangeExpr chExpr context2

12+ notificationFilter (Notification chExpr _) = let oldChangeEval = evalChangeExpr chExpr context1

13+ newChangeEval = evalChangeExpr chExpr context2 in

14+ oldChangeEval /= newChangeEval && isRight oldChangeEval

12− evalChangeExpr chExpr = evalState (evalRelationalExpr chExpr)

15+ evalChangeExpr chExpr = evalState (evalRelationalExpr chExpr)

16+

17+

ProjectM36/RelationalExpression.hs+23−3

@@ -303,6 +303,29 @@ evalContextExpr (RemoveInclusionDependency depName) = do

303 else do

304 put $ currContext {inclusionDependencies = newDeps }

305 return Nothing

306+

307+-- | Add a notification which will send the resultExpr when triggerExpr changes between commits.

308+evalContextExpr (AddNotification notName triggerExpr resultExpr) = do

309+ currentContext <- get

310+ let nots = notifications currentContext

311+ if M.member notName nots then

312+ return $ Just (NotificationNameInUseError notName)

313+ else do

314+ let newNotifications = M.insert notName newNotification nots

315+ newNotification = Notification { changeExpr = triggerExpr,

316+ reportExpr = resultExpr }

317+ put $ currentContext { notifications = newNotifications }

318+ return Nothing

319+

320+evalContextExpr (RemoveNotification notName) = do

321+ currentContext <- get

322+ let nots = notifications currentContext

323+ if M.notMember notName nots then

324+ return $ Just (NotificationNameNotInUseError notName)

325+ else do

326+ let newNotifications = M.delete notName nots

327+ put $ currentContext { notifications = newNotifications }

328+ return Nothing

306

329

307evalContextExpr (MultipleExpr exprs) = do

330evalContextExpr (MultipleExpr exprs) = do

308 --the multiple expressions must pass the same context around- not the old unmodified context

331 --the multiple expressions must pass the same context around- not the old unmodified context

@@ -312,9 +335,6 @@ evalContextExpr (MultipleExpr exprs) = do

312 [] -> return $ Nothing

335 [] -> return $ Nothing

313 err:_ -> return $ Just err

336 err:_ -> return $ Just err

314

337

315−-- restrict relvar to get affected tuples, update tuples, delete restriction from relvar, relvar = relvar union updated tuples

316−--evalRelVarExpr (Update relVarName updateMap) = do

317−

318--run verification on all constraints

338--run verification on all constraints

319checkConstraints :: DatabaseContext -> Maybe RelationalError

339checkConstraints :: DatabaseContext -> Maybe RelationalError

320checkConstraints context = case failures of

340checkConstraints context = case failures of

ProjectM36/Server.hs+1−1

@@ -29,7 +29,7 @@ serverDefinition = defaultProcess {

29 handleCall (\conn (RetrievePlanForDatabaseContextExpr sessionId dbExpr) -> handleRetrievePlanForDatabaseContextExpr sessionId conn dbExpr),

30 handleCall (\conn (RetrieveHeadTransactionUUID sessionId) -> handleRetrieveHeadTransactionUUID sessionId conn),

31 handleCall (\conn (RetrieveTransactionGraph sessionId) -> handleRetrieveTransactionGraph sessionId conn),

32− handleCall (\conn Login -> handleLogin conn),

32+ handleCall (\conn (Login procId) -> handleLogin conn procId),

33 handleCall (\conn (CreateSessionAtHead headn) -> handleCreateSessionAtHead headn conn),

34 handleCall (\conn (CreateSessionAtCommit commitId) -> handleCreateSessionAtCommit commitId conn),

35 handleCall (\conn (CloseSession sessionId) -> handleCloseSession sessionId conn)

ProjectM36/Server/EntryPoints.hs+2−2

@@ -2,7 +2,7 @@ module ProjectM36.Server.EntryPoints where

2import ProjectM36.Base hiding (inclusionDependencies)

3import ProjectM36.Client

4import ProjectM36.Error

5−import Control.Distributed.Process (Process)

5+import Control.Distributed.Process (Process, ProcessId)

6import Control.Distributed.Process.ManagedProcess (ProcessReply)

7import Control.Distributed.Process.ManagedProcess.Server (reply)

8import Control.Monad.IO.Class (liftIO)

@@ -26,7 +26,7 @@ handleExecuteHeadName sessionId conn = do

27handleLogin :: Connection -> ProcessId -> Process (ProcessReply Bool Connection)

28handleLogin conn newClientProcessId = do

29− liftIO (addClientNode newClientProcessId)

29+ liftIO (addClientNode conn newClientProcessId)

30 reply True conn

32handleExecuteGraphExpr :: SessionId -> Connection -> TransactionGraphOperator -> Process (ProcessReply (Maybe RelationalError) Connection)

TutorialD/Interpreter/DatabaseExpr.hs+17

@@ -24,6 +24,8 @@ databaseExprP = insertP

24 <|> defineP

25 <|> undefineP

26 <|> assignP

27+ <|> addNotificationP

28+ <|> removeNotificationP

28assignP :: Parser DatabaseExpr

30assignP :: Parser DatabaseExpr

29assignP = do

31assignP = do

@@ -107,6 +109,20 @@ attributeAssignmentP = do

107 reservedOp ":="

109 reservedOp ":="

108 atom <- stringAtomP <|> intAtomP

110 atom <- stringAtomP <|> intAtomP

109 return $ (attrName, atom)

111 return $ (attrName, atom)

112+

113+addNotificationP :: Parser DatabaseExpr

114+addNotificationP = do

115+ reserved "notify"

116+ notName <- identifier

117+ triggerExpr <- relExprP

118+ resultExpr <- relExprP

119+ return $ AddNotification (T.pack notName) triggerExpr resultExpr

120+

121+removeNotificationP :: Parser DatabaseExpr

122+removeNotificationP = do

123+ reserved "unnotify"

124+ notName <- identifier

125+ return $ RemoveNotification (T.pack notName)

110

126

111databaseExprOpP :: Parser DatabaseExpr

127databaseExprOpP :: Parser DatabaseExpr

112databaseExprOpP = multipleDatabaseExprP

128databaseExprOpP = multipleDatabaseExprP

@@ -131,3 +147,4 @@ interpretNO context tutdstring = case parseString tutdstring of

131 Left err -> (Just err, context)

147 Left err -> (Just err, context)

132 Right parsed -> runState (evalContextExpr parsed) context

148 Right parsed -> runState (evalContextExpr parsed) context

133-}

149-}

150+