00c9dc7 · project-m36 · git.iko.soy

git.iko.soy/project-m36

3noch found that automerge was not writing all new transactions to disk

this was caused by the transaction graph persistence optimization which assumed writing only the latest heads was safe this has been replaced by commit infrastructure being explicit about which transactions it has added to the graph this is potentially a further optimization since it shouldn't need to check the filesystem for which heads have already been written resolves #128

commit00c9dc785ca9caf4eafd05a5ec4b64ea5f168914 authorAgentM <agentm@themactionfaction.com> dateFri 4 Aug 2017, 22:28 (9 years ago) parent75dd04c

7 files changed+67−27

compare with master

unifiedsplit

project-m36.cabal+1−1 src/lib/ProjectM36/Client.hs+21−15 src/lib/ProjectM36/TransactionGraph.hs+3−1 src/lib/ProjectM36/TransactionGraph/Merge.hs+5 src/lib/ProjectM36/TransactionGraph/Persist.hs+15−8 test/TransactionGraph/Automerge.hs+21−1 test/TransactionGraph/Persist.hs+1−1

project-m36.cabal+1−1

@@ -482,7 +482,7 @@ Test-Suite test-multiprocess-access

482482 Hs-Source-Dirs: ./src/bin, ., test/

483483 CPP-Options: -DPROJECTM36_VERSION="v0.1"

484484

485−Test-Suite test-automerge

485+Test-Suite test-transactiongraph-automerge

486486 Default-Language: Haskell2010

487487 type: exitcode-stdio-1.0

488488 main-is: test/TransactionGraph/Automerge.hs

src/lib/ProjectM36/Client.hs+21−15

@@ -128,6 +128,7 @@ import Control.Distributed.Process.Extras.Internal.Types (whereisRemote)

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

129129import Control.Distributed.Process (NodeId(..), reconnect)

130130

131+import Data.Either (isRight)

131132import Data.UUID.V4 (nextRandom)

132133import Data.Word

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

@@ -580,7 +581,8 @@ autoMergeToHead sessionId (InProcessConnection conf) strat headName' = do

580581 Right session -> do

581582 case Graph.autoMergeToHead (id1, id2, id3) (Sess.disconnectedTransaction session) headName' strat graph of

582583 Left err -> pure (Left err)

583− Right (discon', graph') -> pure (Right (discon', graph', True))

584+ Right (discon', graph') -> do

585+ pure (Right (discon', graph', [id1, id2, id3]))

584586autoMergeToHead sessionId conn@(RemoteProcessConnection _) strat headName' = remoteCall conn (ExecuteAutoMergeToHead sessionId strat headName')

585587

586588-- | Execute a database context IO-monad-based expression for the given session and connection. `DatabaseContextIOExpr`s modify the DatabaseContext but cannot be purely implemented.

@@ -645,7 +647,13 @@ executeGraphExpr sessionId (InProcessConnection conf) graphExpr = excEither $ do

645647 let discon = Sess.disconnectedTransaction session

646648 case evalGraphOp freshId discon updatedGraph graphExpr of

647649 Left err -> pure (Left err)

648− Right (discon', graph') -> pure (Right (discon', graph', isCommit graphExpr))

650+ Right (discon', graph') -> do

651+ --if freshId appears in the graph, then we need to pass it on

652+ let transIds = if isRight (transactionForId freshId graph') then

653+ [freshId]

654+ else

655+ []

656+ pure (Right (discon', graph', transIds))

649657{-

650658executeGraphExpr sessionId (InProcessConnection conf) graphExpr = excEither $ do

651659 let strat = ipPersistenceStrategy conf

@@ -768,10 +776,10 @@ rollback sessionId conn@(InProcessConnection _) = executeGraphExpr sessionId con

768776rollback sessionId conn@(RemoteProcessConnection _) = remoteCall conn (ExecuteGraphExpr sessionId Rollback)

769777

770778-- | Write the transaction graph to disk. This function can be used to incrementally write new transactions to disk.

771−processTransactionGraphPersistence :: PersistenceStrategy -> TransactionGraph -> IO ()

772−processTransactionGraphPersistence NoPersistence _ = pure ()

773−processTransactionGraphPersistence (MinimalPersistence dbdir) graph = transactionGraphPersist NoDiskSync dbdir graph >> pure ()

774−processTransactionGraphPersistence (CrashSafePersistence dbdir) graph = transactionGraphPersist FsyncDiskSync dbdir graph >> pure ()

779+processTransactionGraphPersistence :: PersistenceStrategy -> [TransactionId] -> TransactionGraph -> IO ()

780+processTransactionGraphPersistence NoPersistence _ _ = pure ()

781+processTransactionGraphPersistence (MinimalPersistence dbdir) transIds graph = transactionGraphPersist NoDiskSync dbdir transIds graph >> pure ()

782+processTransactionGraphPersistence (CrashSafePersistence dbdir) transIds graph = transactionGraphPersist FsyncDiskSync dbdir transIds graph >> pure ()

775783

776784readGraphTransactionIdDigest :: PersistenceStrategy -> IO (LockFileHash)

777785readGraphTransactionIdDigest NoPersistence = error "attempt to read digest from transaction log without persistence enabled"

@@ -947,13 +955,11 @@ disconnectedTransaction_ sessionId (InProcessConnection conf) = do

947955 Just (Sess.Session discon _) -> pure discon

948956disconnectedTransaction_ _ _= error "remote connection used"

949957

950−type ProcessCommit = Bool

951−

952958-- wrap a graph evaluation in file locking

953959commitLock_ :: SessionId ->

954960 InProcessConnectionConf ->

955961 (TransactionGraph ->

956− STM (Either RelationalError (DisconnectedTransaction, TransactionGraph, ProcessCommit))) ->

962+ STM (Either RelationalError (DisconnectedTransaction, TransactionGraph, [TransactionId]))) ->

957963 IO (Either RelationalError ())

958964commitLock_ sessionId conf stmBlock = do

959965 let sessions = ipSessions conf

@@ -998,25 +1004,25 @@ commitLock_ sessionId conf stmBlock = do

9981004 eGraph <- stmBlock refreshedGraph

9991005 case eGraph of

10001006 Left err -> pure (Left err)

1001− Right (discon', graph', processCommit) -> do

1007+ Right (discon', graph', transactionIdsToPersist) -> do

10021008 writeTVar graphTvar graph'

10031009 let newSession = Session discon' (Sess.schemaName session)

10041010 STMMap.insert newSession sessionId sessions

10051011 case transactionForId (Sess.parentId session) oldGraph of

10061012 Left err -> pure $ Left err

10071013 Right previousTrans -> do

1008− if processCommit then do

1014+ if not (Prelude.null transactionIdsToPersist) then do

10091015 (evaldNots, nodes) <- executeCommitExprSTM_ (Trans.concreteDatabaseContext previousTrans) (Sess.concreteDatabaseContext session) clientNodes

10101016 nodesToNotify <- toList (STMSet.stream nodes)

1011− pure $ Right (evaldNots, nodesToNotify, graph')

1012− else pure (Right (M.empty, [], graph'))

1017+ pure $ Right (evaldNots, nodesToNotify, graph', transactionIdsToPersist)

1018+ else pure (Right (M.empty, [], graph', []))

10131019

10141020 --handle notification firing

10151021 case manip of

10161022 Left err -> pure (Left err)

1017− Right (notsToFire, nodesToNotify, newGraph) -> do

1023+ Right (notsToFire, nodesToNotify, newGraph, transactionIdsToPersist) -> do

10181024 --update filesystem database, if necessary

1019− processTransactionGraphPersistence strat newGraph

1025+ processTransactionGraphPersistence strat transactionIdsToPersist newGraph

10201026 sendNotifications nodesToNotify (ipLocalNode conf) notsToFire

10211027 pure (Right ())

10221028{-

src/lib/ProjectM36/TransactionGraph.hs+3−1

@@ -440,12 +440,13 @@ createUnionMergeTransaction newId strategy graph (t1,t2) = do

440440 atomFuncs <- unionMergeAtomFunctions preference (atomFunctions contextA) (atomFunctions contextB)

441441 notifs <- unionMergeMaps preference (notifications contextA) (notifications contextB)

442442 types <- unionMergeTypeConstructorMapping preference (typeConstructorMapping contextA) (typeConstructorMapping contextB)

443+ dbcFuncs <- unionMergeDatabaseContextFunctions preference (dbcFunctions contextA) (dbcFunctions contextB)

443444 -- TODO: add merge of subschemas

444445 let newContext = DatabaseContext {

445446 inclusionDependencies = incDeps,

446447 relationVariables = relVars,

447448 atomFunctions = atomFuncs,

448− dbcFunctions = undefined,

449+ dbcFunctions = dbcFuncs,

449450 notifications = notifs,

450451 typeConstructorMapping = types

451452 }

@@ -507,6 +508,7 @@ autoMergeToHead (tempBranchTransId, tempCommitTransId, mergeTransId) discon merg

507508 let rel = runReader (evalRelationalExpr (RelationVariable "s" ())) (mkRelationalExprState $ D.concreteDatabaseContext discon'''')

508509 traceShowM rel

509510-}

511+

510512 pure (discon'''', graph'''')

511513

512514

src/lib/ProjectM36/TransactionGraph/Merge.hs+5

@@ -83,3 +83,8 @@ unionMergeTypeConstructorMapping prefer typesA typesB = do

8383 PreferNeither -> Left StrategyViolatesTypeConstructorMergeError

8484 ) [] (S.toList allFuncNames)

8585

86+unionMergeDatabaseContextFunctions :: MergePreference -> DatabaseContextFunctions -> DatabaseContextFunctions -> Either MergeError DatabaseContextFunctions

87+unionMergeDatabaseContextFunctions prefer funcsA funcsB = case prefer of

88+ PreferFirst -> pure $ HS.union funcsA funcsB

89+ PreferSecond -> pure $ HS.union funcsB funcsA

90+ PreferNeither -> pure $ HS.union funcsA funcsB

src/lib/ProjectM36/TransactionGraph/Persist.hs+15−8

@@ -70,7 +70,8 @@ bootstrapDatabaseDir :: DiskSync -> FilePath -> TransactionGraph -> IO (Handle,

7070bootstrapDatabaseDir sync dbdir bootstrapGraph = do

7171 createDirectory dbdir

7272 lockFileH <- openLockFile dbdir

73− digest <- bracket_ (lockFile lockFileH WriteLock) (unlockFile lockFileH) (transactionGraphPersist sync dbdir bootstrapGraph)

73+ let allTransIds = map transactionId (S.toList (transactionsForGraph bootstrapGraph))

74+ digest <- bracket_ (lockFile lockFileH WriteLock) (unlockFile lockFileH) (transactionGraphPersist sync dbdir allTransIds bootstrapGraph)

7475 pure (lockFileH, digest)

7576

7677openLockFile :: FilePath -> IO (Handle)

@@ -85,19 +86,25 @@ incrementally updates an existing database directory

8586-assume that all non-head transactions have already been written because this is an incremental (and concurrent!) write method

8687--store the head names with a symlink to the transaction under "heads"

8788-}

88−transactionGraphPersist :: DiskSync -> FilePath -> TransactionGraph -> IO LockFileHash

89−transactionGraphPersist sync destDirectory graph = do

90− transactionHeadTransactionsPersist sync destDirectory graph

89+transactionGraphPersist :: DiskSync -> FilePath -> [TransactionId] -> TransactionGraph -> IO LockFileHash

90+transactionGraphPersist sync destDirectory transIds graph = do

91+ transactionsPersist sync transIds destDirectory graph

9192 --write graph file

9293 newDigest <- writeGraphTransactionIdFile sync destDirectory graph

9394 --write heads file

9495 transactionGraphHeadsPersist sync destDirectory graph

9596 pure newDigest

9697

97−-- | The incremental writer which only writes from the set of heads. New heads must be written on every commit. Most heads will already be written on every commit.

98−transactionHeadTransactionsPersist :: DiskSync -> FilePath -> TransactionGraph -> IO ()

99−transactionHeadTransactionsPersist sync destDirectory graphIn = mapM_ (writeTransaction sync destDirectory) $ M.elems (transactionHeadsForGraph graphIn)

100−

98+-- | The incremental writer writes the transactions ids specified by the second argument.

99+

100+-- There was a bug here via #128 because automerge added multiple transactions to the graph but this function used to only write the head transactions from the graph. Automerge creates multiple transactions, so these are now passed in as the second argument.

101+transactionsPersist :: DiskSync -> [TransactionId] -> FilePath -> TransactionGraph -> IO ()

102+transactionsPersist sync transIds destDirectory graphIn = mapM_ writeTrans transIds

103+ where writeTrans tid = do

104+ case transactionForId tid graphIn of

105+ Left err -> error ("writeTransaction: " ++ show err)

106+ Right trans -> writeTransaction sync destDirectory trans

107+

101108{-

102109write graph heads to a file which can be atomically swapped

103110-}

test/TransactionGraph/Automerge.hs+21−1

@@ -6,6 +6,9 @@ import qualified Data.Set as S

66import TutorialD.Interpreter.TestBase

88import System.Exit

9+import System.IO.Temp

10+import System.FilePath

11+import Control.Exception.Base

912

1013main :: IO ()

1114main = do

@@ -14,7 +17,8 @@ main = do

1417

1518testList :: Test

1619testList = TestList [testAutomergeSuccess,

17− testAutomergeFailure]

20+ testAutomergeFailure,

21+ testAutomergeReconnect]

1822

1923checkEither :: IO (Either RelationalError a) -> IO a

2024checkEither io = do

@@ -67,5 +71,21 @@ testAutomergeFailure = TestCase $ do

6771

6872 assertEqual "merge failure" (Left (InclusionDependencyCheckError "s_pkey")) mergeRes

6973

74+--reported as #128

75+testAutomergeReconnect :: Test

76+testAutomergeReconnect = TestCase $ withSystemTempDirectory "m36testdb" $ \tempdir -> do

77+ let repro = do

78+ conn <- unsafeLeftCrash =<< connectProjectM36 (InProcessConnectionInfo (CrashSafePersistence (tempdir </> "test.db")) emptyNotificationCallback [])

79+ sess <- unsafeLeftCrash =<< createSessionAtHead conn "master"

80+ autoMergeToHead sess conn UnionMergeStrategy "master"

81+ -- commit sess conn

82+

83+ unsafeLeftCrash :: Show e => Either e a -> IO a

84+ unsafeLeftCrash = either (throwIO . userError . show) pure

85+ _ <- repro

86+ _ <- repro

87+ pure ()

88+

89+

7090

7191

test/TransactionGraph/Persist.hs+1−1

@@ -57,7 +57,7 @@ testDBSimplePersistence = TestCase $ withSystemTempDirectory "m36testdb" $ \temp

5757 Left err -> assertFailure (show err)

5858 Right (_, graph') -> do

5959 --persist the new graph

60− _ <- transactionGraphPersist NoDiskSync dbdir graph'

60+ _ <- transactionGraphPersist NoDiskSync dbdir [freshId'] graph'

6161 --reload the graph from the filesystem and confirm that the transaction is present

6262 graphErr <- transactionGraphLoad dbdir emptyTransactionGraph Nothing

6363 let mapEq graphArg = S.map transactionId (transactionsForGraph graphArg)