diff options
Diffstat (limited to 'source/Provers.hs')
| -rw-r--r-- | source/Provers.hs | 642 |
1 files changed, 444 insertions, 198 deletions
diff --git a/source/Provers.hs b/source/Provers.hs index 7f85c29..941360c 100644 --- a/source/Provers.hs +++ b/source/Provers.hs @@ -32,11 +32,6 @@ module Provers , preparedVerificationBytes , preparedVerificationByteCount , preparedVerificationRequestId - , PreparedProverTask - , prepareProverTask - , preparedProverLogicalTask - , preparedProverRequest - , preparedProverTptpTask , PreparedTypedProverTask , prepareTypedProverTask , preparedTypedProverLogicalProblem @@ -45,24 +40,57 @@ module Provers , AcceptedVampireRun , acceptedVampireRequest , provedVampireRun - , runProver - , runPreparedProver - , runPreparedProverWithObserver , runPreparedTypedProver , runPreparedTypedProverWithObserver - , runVampireProcess + , EffectiveJobs + , effectiveJobs + , effectiveJobsValue + , JobsSelection(..) + , selectEffectiveJobs + , WorkPosition + , workPosition + , workPositionModuleOrdinal + , workPositionLocalRequestOrdinal + , VampireExecutor + , VampireExecutorObservation(..) + , withVampireExecutor + , runPreparedTypedProverWithExecutor + , vampireExecutorObservation ) where import Base import Checking.Authority qualified as Authority import Checking.Backend.Problem import Checking.Backend.Tptp -import Encoding -import Report.Location -import Syntax.Internal (Directness(..), Formula, Task(..), isIndirect) -import Control.Exception (IOException, displayException) +import Control.Concurrent.STM + ( TBQueue + , TMVar + , TVar + , atomically + , check + , modifyTVar' + , newEmptyTMVarIO + , newTBQueueIO + , newTVarIO + , orElse + , putTMVar + , readTBQueue + , readTVar + , readTVarIO + , takeTMVar + , writeTBQueue + , writeTVar + ) +import Control.Exception + ( AsyncException + , IOException + , SomeException + , displayException + , fromException + ) import Control.Exception qualified as Exception +import Control.Monad (forever, replicateM, unless) import Control.Monad.Logger import Data.ByteString qualified as ByteString import Data.IORef @@ -76,6 +104,7 @@ import Data.Set qualified as Set import Data.Text qualified as Text import Data.Text.Encoding qualified as TextEncoding import Data.Time +import Numeric.Natural (Natural) import System.Exit (ExitCode(..)) import System.Posix.Signals (sigKILL, signalProcessGroup) import System.Posix.Types (ProcessGroupID) @@ -92,7 +121,13 @@ import System.Timeout qualified as Timeout import Text.Megaparsec import Text.Megaparsec.Char qualified as Char import TextBuilder -import UnliftIO.Async (concurrently) +import UnliftIO.Async + ( async + , cancel + , concurrently + , waitCatch + , waitCatchSTM + ) data Vampire = Vampire { vampireExecutable :: FilePath @@ -119,9 +154,92 @@ vampireArguments Vampire{..} = , "--mode", "casc" , "--time_limit", toSeconds vampireTimeLimit , "--memory_limit", toMegabytes vampireMemoryLimit - , "--cores", "2" + , "--cores", "1" ] +-- | A strictly positive invocation-local worker bound. +newtype EffectiveJobs = EffectiveJobs Int + deriving (Show, Eq, Ord) + +effectiveJobs :: Int -> Maybe EffectiveJobs +effectiveJobs amount + | amount > 0 = Just (EffectiveJobs amount) + | otherwise = Nothing + +effectiveJobsValue :: EffectiveJobs -> Int +effectiveJobsValue (EffectiveJobs amount) = amount + +-- | The effective worker policy selected for one invocation. Processor +-- discovery is operational evidence only and enters no durable identity. +data JobsSelection = JobsSelection + { jobsSelectionDetectedProcessors :: !(Maybe Int) + , jobsSelectionEffectiveJobs :: !EffectiveJobs + , jobsSelectionWasOverridden :: !Bool + } + deriving (Show, Eq) + +-- | Select the worker bound, with processor discovery injected for focused +-- testing. Asynchronous cancellation is never mistaken for failed discovery. +selectEffectiveJobs + :: Maybe EffectiveJobs + -> IO Int + -> IO JobsSelection +selectEffectiveJobs override detectProcessors = + case override of + Just selected -> + pure + JobsSelection + { jobsSelectionDetectedProcessors = Nothing + , jobsSelectionEffectiveJobs = selected + , jobsSelectionWasOverridden = True + } + Nothing -> do + detectedResult <- Exception.try detectProcessors + case detectedResult of + Left detectionFailure + | Just asynchronous <- + (fromException detectionFailure + :: Maybe AsyncException) -> + Exception.throwIO asynchronous + | otherwise -> + fallback + Right detected + | detected > 0 -> + pure + JobsSelection + { jobsSelectionDetectedProcessors = + Just detected + , jobsSelectionEffectiveJobs = + EffectiveJobs (max 1 (detected - 1)) + , jobsSelectionWasOverridden = False + } + | otherwise -> + fallback + where + fallback = + pure + JobsSelection + { jobsSelectionDetectedProcessors = Nothing + , jobsSelectionEffectiveJobs = EffectiveJobs 1 + , jobsSelectionWasOverridden = False + } + +-- | Stable runtime diagnostic position. It is deliberately separate from +-- request, validation, cache, and mathematical identities. +data WorkPosition = WorkPosition !Natural !Natural + deriving (Show, Eq, Ord) + +workPosition :: Natural -> Natural -> WorkPosition +workPosition = WorkPosition + +workPositionModuleOrdinal :: WorkPosition -> Natural +workPositionModuleOrdinal (WorkPosition moduleOrdinal _localOrdinal) = + moduleOrdinal + +workPositionLocalRequestOrdinal :: WorkPosition -> Natural +workPositionLocalRequestOrdinal (WorkPosition _moduleOrdinal localOrdinal) = + localOrdinal + toSeconds :: TimeLimit -> String toSeconds (Seconds secs) = show secs @@ -292,39 +410,6 @@ preparedVerificationRequestId request = IndirectTask -> Authority.PreparedRequestIndirect) (preparedVerificationBytes request) -data PreparedProverTask = PreparedProverTask - !Task - !PreparedTptpTask - !PreparedVerificationRequest - -prepareProverTask :: Task -> PreparedProverTask -prepareProverTask task = - let preparedTptp = prepareTptpTask task - preparedRequest = - prepareVerificationRequest task preparedTptp - in - PreparedProverTask - task - preparedTptp - preparedRequest - -preparedProverLogicalTask :: PreparedProverTask -> Task -preparedProverLogicalTask - (PreparedProverTask task _preparedTptp _preparedRequest) = - task - -preparedProverRequest - :: PreparedProverTask - -> PreparedVerificationRequest -preparedProverRequest - (PreparedProverTask _task _preparedTptp preparedRequest) = - preparedRequest - -preparedProverTptpTask :: PreparedProverTask -> PreparedTptpTask -preparedProverTptpTask - (PreparedProverTask _task preparedTptp _preparedRequest) = - preparedTptp - data PreparedTypedProverTask ref local origin global = PreparedTypedProverTask !(TypedProblem ref local origin global) @@ -506,72 +591,6 @@ captureChunkSize :: Int captureChunkSize = 32 * 1024 -runProver - :: (MonadIO io, MonadLogger io) - => Vampire - -> Task - -> io - ( Location - , Formula - , Either ProverProcessError ProverAnswer - ) -runProver vampireCommand task = - runPreparedProver vampireCommand (prepareProverTask task) - -runPreparedProver - :: (MonadIO io, MonadLogger io) - => Vampire - -> PreparedProverTask - -> io - ( Location - , Formula - , Either ProverProcessError ProverAnswer - ) -runPreparedProver - vampireCommand = - runPreparedProverWithObserver - (\_request -> pure ()) - vampireCommand - -runPreparedProverWithObserver - :: (MonadIO io, MonadLogger io) - => (PreparedVerificationRequest -> IO ()) - -> Vampire - -> PreparedProverTask - -> io - ( Location - , Formula - , Either ProverProcessError ProverAnswer - ) -runPreparedProverWithObserver - observer - vampireCommand - (PreparedProverTask task preparedTask preparedRequest) = do - startTime <- liftIO getCurrentTime - transcriptResult <- - liftIO - (runPreparedVampireProcessWithObserver - observer - vampireCommand - preparedRequest) - let answer = - classifyVampireAnswer - task - vampireCommand - preparedRequest - <$> transcriptResult - endTime <- liftIO getCurrentTime - let duration = timeDifferenceToText startTime endTime - - logInfoN - ( duration - <> " " - <> preparedTptpConjectureText preparedTask - <> taskProvenance task - ) - - pure (taskLocation task, taskConjecture task, answer) - runPreparedTypedProver :: (MonadIO io, MonadLogger io) => Vampire @@ -630,52 +649,279 @@ runPreparedTypedProverWithObserver <> "]") pure answer -taskProvenance :: Task -> Text -taskProvenance task = - " [" <> directness <> " at " - <> Text.pack (show (locationToText (taskLocation task))) - <> "]" - where - directness = case taskDirectness task of - Direct -> "direct" - Indirect _ -> "indirect" +-- | Invocation-local bounded owner of Vampire subprocesses. The finite queue +-- carries only immutable prepared bytes plus runtime diagnostic position; +-- checker builders and typed tasks never cross this boundary. +data VampireExecutor = VampireExecutor + { executorQueue :: !(TBQueue ExecutorJob) + , executorClosed :: !(TVar Bool) + , executorCommand :: !Vampire + , executorRequestObserver + :: !(WorkPosition -> PreparedVerificationRequest -> IO ()) + , executorRuntime :: !(TVar VampireExecutorRuntime) + } -runVampireProcess - :: Vampire - -> Task - -> IO (Either ProverProcessError CompletedTranscript) -runVampireProcess - vampireCommand - task = - let preparedTask = prepareTptpTask task - in runPreparedVampireProcess - vampireCommand - (prepareVerificationRequest task preparedTask) - -prepareVerificationRequest - :: Task - -> PreparedTptpTask - -> PreparedVerificationRequest -prepareVerificationRequest task preparedTask = - PreparedVerificationRequest - { preparedVerificationDialect = VerificationFof - , preparedVerificationMode = vampireTaskMode task - , preparedVerificationInput = - TextEncoding.encodeUtf8 - (preparedTptpTextNewline preparedTask) - , preparedVerificationText = - preparedTptpText preparedTask +data ExecutorJob = ExecutorJob + { executorJobPosition :: !WorkPosition + , executorJobRequest :: !PreparedVerificationRequest + , executorJobCancelled :: !(TVar Bool) + , executorJobCompletion + :: !(TMVar + (Either + SomeException + (Either ProverProcessError ProverAnswer))) + } + +data VampireExecutorRuntime = VampireExecutorRuntime + { runtimeSubmittedCount :: !Int + , runtimeRunCount :: !Int + , runtimeLiveCount :: !Int + , runtimeMaximumLiveCount :: !Int + , runtimeFirstStartNanoseconds :: !(Maybe Word64) + , runtimeExecutionNanoseconds :: !Word64 + } + +data VampireExecutorObservation = VampireExecutorObservation + { vampireExecutorSubmittedCount :: !Int + , vampireExecutorRunCount :: !Int + , vampireExecutorMaximumLiveCount :: !Int + , vampireExecutorFirstStartNanoseconds :: !(Maybe Word64) + , vampireExecutorExecutionNanoseconds :: !Word64 + } + deriving (Show, Eq) + +data VampireExecutorClosed = VampireExecutorClosed + deriving (Show) + +instance Exception.Exception VampireExecutorClosed + +data RequestObserverFailure = RequestObserverFailure SomeException + +instance Show RequestObserverFailure where + show (RequestObserverFailure observerError) = + "request observer failed: " <> displayException observerError + +instance Exception.Exception RequestObserverFailure + +initialVampireExecutorRuntime :: VampireExecutorRuntime +initialVampireExecutorRuntime = + VampireExecutorRuntime + { runtimeSubmittedCount = 0 + , runtimeRunCount = 0 + , runtimeLiveCount = 0 + , runtimeMaximumLiveCount = 0 + , runtimeFirstStartNanoseconds = Nothing + , runtimeExecutionNanoseconds = 0 } -runPreparedVampireProcess - :: Vampire +-- | Bracket exactly the selected number of workers. Cancelling the bracket +-- cancels each worker; a worker owning a subprocess in turn terminates and +-- reaps that process group through the ordinary supervisor boundary. +withVampireExecutor + :: EffectiveJobs + -> Vampire + -> (WorkPosition -> PreparedVerificationRequest -> IO ()) + -> (VampireExecutor -> IO value) + -> IO value +withVampireExecutor selected command observer action = + Exception.bracket acquire release (action . fst) + where + workerCount = effectiveJobsValue selected + + acquire = do + queue <- newTBQueueIO + (fromIntegral workerCount * 2) + closed <- newTVarIO False + runtime <- newTVarIO initialVampireExecutorRuntime + let executor = + VampireExecutor + { executorQueue = queue + , executorClosed = closed + , executorCommand = command + , executorRequestObserver = observer + , executorRuntime = runtime + } + workers <- replicateM workerCount (async (executorWorker executor)) + pure (executor, workers) + + release (executor, workers) = do + atomically (writeTVar (executorClosed executor) True) + traverse_ cancel workers + traverse_ waitCatch workers + +runPreparedTypedProverWithExecutor + :: (MonadIO io, MonadLogger io) + => VampireExecutor + -> WorkPosition + -> PreparedTypedProverTask ref local origin global + -> io (Either ProverProcessError ProverAnswer) +runPreparedTypedProverWithExecutor + executor + position + (PreparedTypedProverTask + _problem + prepared + preparedRequest) = do + startTime <- liftIO getCurrentTime + answer <- liftIO (submitVampireRequest executor position preparedRequest) + endTime <- liftIO getCurrentTime + let dialect = + case preparedTypedTptpRoute prepared of + RouteFof -> "FOF" + RouteTh0 -> "TH0" + logInfoN + (timeDifferenceToText startTime endTime + <> " " + <> preparedTypedTptpConjectureText prepared + <> " [typed " + <> dialect + <> "]") + pure answer + +vampireExecutorObservation + :: VampireExecutor + -> IO VampireExecutorObservation +vampireExecutorObservation executor = do + runtime <- readTVarIO (executorRuntime executor) + pure + VampireExecutorObservation + { vampireExecutorSubmittedCount = runtimeSubmittedCount runtime + , vampireExecutorRunCount = runtimeRunCount runtime + , vampireExecutorMaximumLiveCount = + runtimeMaximumLiveCount runtime + , vampireExecutorFirstStartNanoseconds = + runtimeFirstStartNanoseconds runtime + , vampireExecutorExecutionNanoseconds = + runtimeExecutionNanoseconds runtime + } + +submitVampireRequest + :: VampireExecutor + -> WorkPosition -> PreparedVerificationRequest - -> IO (Either ProverProcessError CompletedTranscript) -runPreparedVampireProcess - vampireCommand = - runPreparedVampireProcessWithObserver - (\_request -> pure ()) - vampireCommand + -> IO (Either ProverProcessError ProverAnswer) +submitVampireRequest executor position request = + Exception.mask \restore -> do + -- Force the compact queue payload before handing it to a worker. + _ <- Exception.evaluate (preparedVerificationByteCount request) + _ <- Exception.evaluate + (Text.length (preparedVerificationText request)) + cancelled <- newTVarIO False + completion <- newEmptyTMVarIO + let job = + ExecutorJob + { executorJobPosition = position + , executorJobRequest = request + , executorJobCancelled = cancelled + , executorJobCompletion = completion + } + cancelJob = atomically (writeTVar cancelled True) + enqueue = atomically do + closed <- readTVar (executorClosed executor) + if closed + then pure False + else do + writeTBQueue (executorQueue executor) job + modifyTVar' + (executorRuntime executor) + (\runtime -> + runtime + { runtimeSubmittedCount = + runtimeSubmittedCount runtime + 1 + }) + pure True + accepted <- restore enqueue `Exception.onException` cancelJob + unless accepted (Exception.throwIO VampireExecutorClosed) + outcome <- + restore (atomically (takeTMVar completion)) + `Exception.onException` cancelJob + either Exception.throwIO pure outcome + +executorWorker :: VampireExecutor -> IO () +executorWorker executor = + forever do + job <- atomically (readTBQueue (executorQueue executor)) + cancelled <- readTVarIO (executorJobCancelled job) + unless cancelled (executeJob executor job) + +data JobWait value + = JobFinished !(Either SomeException value) + | JobCancelled + +executeJob :: VampireExecutor -> ExecutorJob -> IO () +executeJob executor job = + Exception.mask \restore -> do + running <- async + (restore + (observeExecutorRun executor + (runPreparedVerificationRequest + (executorRequestObserver executor + (executorJobPosition job)) + (executorCommand executor) + (executorJobRequest job)))) + waited <- restore + (atomically + ( (JobFinished <$> waitCatchSTM running) + `orElse` + (do + cancelled <- readTVar + (executorJobCancelled job) + check cancelled + pure JobCancelled) + )) + `Exception.onException` + (cancel running >> void (waitCatch running)) + case waited of + JobFinished result -> + atomically + (putTMVar (executorJobCompletion job) result) + JobCancelled -> do + cancel running + void (waitCatch running) + +observeExecutorRun :: VampireExecutor -> IO value -> IO value +observeExecutorRun executor action = do + started <- getMonotonicTimeNSec + atomically + (modifyTVar' + (executorRuntime executor) + (\runtime -> + let live = runtimeLiveCount runtime + 1 + in runtime + { runtimeRunCount = runtimeRunCount runtime + 1 + , runtimeLiveCount = live + , runtimeMaximumLiveCount = + max live (runtimeMaximumLiveCount runtime) + , runtimeFirstStartNanoseconds = + runtimeFirstStartNanoseconds runtime <|> Just started + })) + action `Exception.finally` do + finished <- getMonotonicTimeNSec + atomically + (modifyTVar' + (executorRuntime executor) + (\runtime -> + runtime + { runtimeLiveCount = runtimeLiveCount runtime - 1 + , runtimeExecutionNanoseconds = + runtimeExecutionNanoseconds runtime + + (finished - started) + })) + +runPreparedVerificationRequest + :: (PreparedVerificationRequest -> IO ()) + -> Vampire + -> PreparedVerificationRequest + -> IO (Either ProverProcessError ProverAnswer) +runPreparedVerificationRequest observer command request = do + transcriptResult <- + runPreparedVampireProcessWithObserver observer command request + pure + (classifyPreparedVampireAnswer + "typed obligation" + command + request + <$> transcriptResult) runPreparedVampireProcessWithObserver :: (PreparedVerificationRequest -> IO ()) @@ -706,10 +952,12 @@ runPreparedVampireProcessWithObserver pure (fromIntegral pid) Nothing -> impossible - "runVampireProcess: missing process-group id" + "Vampire supervisor: missing process-group id" restore (do - observer preparedRequest + runRequestObserver + observer + preparedRequest superviseVampireProcess executable (vampireTimeLimit vampireCommand) @@ -725,27 +973,50 @@ runPreparedVampireProcessWithObserver processHandle _ -> impossible - "runVampireProcess: expected CreatePipe handles") + "Vampire supervisor: expected CreatePipe handles") :: IO (Either - IOException + SomeException (Either ProverProcessError CompletedTranscript)) case processResult of Right result -> pure result - Left err -> do - started <- readIORef callbackStarted - pure - (Left - (if started - then - ProverLifecycleFailed - executable - (exceptionText err) - else - ProverLaunchFailed - executable - (exceptionText err))) + Left err + | Just (RequestObserverFailure observerFailure) <- + fromException err -> + Exception.throwIO observerFailure + | Just asynchronous <- + (fromException err :: Maybe AsyncException) -> + Exception.throwIO asynchronous + | Just ioFailure <- + (fromException err :: Maybe IOException) -> do + started <- readIORef callbackStarted + pure + (Left + (if started + then + ProverLifecycleFailed + executable + (exceptionText ioFailure) + else + ProverLaunchFailed + executable + (exceptionText ioFailure))) + | otherwise -> + Exception.throwIO err + +runRequestObserver + :: (PreparedVerificationRequest -> IO ()) + -> PreparedVerificationRequest + -> IO () +runRequestObserver observer request = + observer request `Exception.catch` \observerError -> + case fromException observerError :: Maybe AsyncException of + Just asynchronous -> + Exception.throwIO asynchronous + Nothing -> + Exception.throwIO + (RequestObserverFailure observerError) superviseVampireProcess :: FilePath @@ -1033,25 +1304,6 @@ exceptionText :: IOException -> Text exceptionText = Text.pack . displayException -classifyVampireAnswer - :: Task - -> Vampire - -> PreparedVerificationRequest - -> CompletedTranscript - -> ProverAnswer -classifyVampireAnswer - task - vampireCommand - preparedRequest - transcript = - classifyPreparedVampireAnswer - (Text.pack - (show - (taskConjectureLabel task))) - vampireCommand - preparedRequest - transcript - classifyPreparedVampireAnswer :: Text -> Vampire @@ -1152,12 +1404,6 @@ renderProtocolError protocolError CompleteCapturedStreams{..} = , completedStderr ] -vampireTaskMode :: Task -> VampireTaskMode -vampireTaskMode task = - if isIndirect task - then IndirectTask - else DirectTask - vampireStatusFromText :: Text -> VampireStatus vampireStatusFromText = \case "Theorem" -> StatusTheorem |