diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,29 @@ +# Version 0.1 + +* CHANGED the type of `encodeAeson` and `encodeBinary`. + +* Added `hinput`, `houtput`, `encodeNS`, `decodeNS`, `encodeAeson'`, + `encodeBinary'`. + +* Added `Contravariant` `Rep` instance for `Encode` and `Input`. + +* Added `InputDefault`, `OutputDefault` and related instances. + +* Added `EncodeDefault` instances for `aeson`'s `Encoding and `Value`, + and for `binary`'s `Put`. + +* Export `BindingName` constructor. + + +# Version 0.0.2 + +* Added `FromJSON`, `ToJSON` instances for `Name`. + +* Improved type-parameter order in `zero`, `one`, `foldM` and similar. + +* Added manual transactional migrations suport. + + +# Version 0.0.1 + +* Initial version. diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,219 @@ +Copyright 2024, Renzo Carbonara. + +Except where otherwise explicitly mentioned, all of contents of the +"job" project are licensed under the Apache License, +Version 2.0 (the "License"); you may not use this file except in +compliance with the License. You may obtain a copy of the License at the +end of this file or at http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. + + +------------------------------------------------------------------------------- + + + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright [yyyy] [name of copyright owner] + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,18 @@ +# job + +Job queue thing in Haskell. + +* Copyright © Renzo Carbonara, 2024. + +* Licensing terms in `LICENSE` file (`Apache-2.0`). + + + +## Development + +1. Clone this repo and run `nix develop` in the top-level folder. + +2. Within that environment, you can use `cabal` as usual. + +3. If you don't have `nix` installed, then have `nix` installed. + diff --git a/job.cabal b/job.cabal new file mode 100644 --- /dev/null +++ b/job.cabal @@ -0,0 +1,65 @@ +cabal-version: 2.4 +name: job +version: 0.1 +license: Apache-2.0 +license-file: LICENSE +extra-source-files: README.md CHANGELOG.md +author: Renzo Carbonara +maintainer: renλren.zone +copyright: Renzo Carbonara, 2024 +category: Databases +build-type: Simple +synopsis: Job queue +description: Job queue +homepage: https://github.com/k0001/hs-job +bug-reports: https://github.com/k0001/hs-job/issues +tested-with: GHC == 9.8.1 + +common basic + default-language: GHC2021 + ghc-options: -O2 -Wall + build-depends: + base == 4.*, + default-extensions: + BlockArguments + DataKinds + DeriveAnyClass + DerivingStrategies + DuplicateRecordFields + LambdaCase + MultiWayIf + OverloadedRecordDot + RecordWildCards + RoleAnnotations + +library + import: basic + hs-source-dirs: lib + exposed-modules: + Job + Job.Memory + build-depends: + async, + base, + containers, + hashable, + mmzk-typeid, + resourcet, + resourcet-extra, + safe-exceptions, + stm, + time, + transformers, + +test-suite test + import: basic + type: exitcode-stdio-1.0 + main-is: Main.hs + hs-source-dirs: test + ghc-options: -O2 -Wall -threaded -rtsopts -with-rtsopts=-N + build-depends: + base, + job, + random, + resourcet, + time, diff --git a/lib/Job.hs b/lib/Job.hs new file mode 100644 --- /dev/null +++ b/lib/Job.hs @@ -0,0 +1,223 @@ +{-# LANGUAGE StrictData #-} +{-# LANGUAGE NoFieldSelectors #-} + +module Job + ( -- * Queue + Queue (..) + , push + , prune + + -- * Work + , Work (..) + , retry + , finish + + -- * Meta + , Meta (..) + + -- * Id + , Id + , unsafeIdFromUUID7 + , idFromUUID7 + , newId + + -- * Nice + , Nice (..) + , nice0 + ) where + +import Control.Monad +import Control.Monad.IO.Class +import Data.Acquire qualified as A +import Data.Hashable +import Data.Int +import Data.Time qualified as Time +import Data.UUID.V7 (UUID) +import Data.UUID.V7 qualified as UUID7 +import Data.Word +import GHC.Records +import GHC.Stack + +-------------------------------------------------------------------------------- + +-- | Nice value for a job. +-- +-- * The __lower__ the value, the less “nice” a job is to other jobs. Meaning +-- it will have priority over other jobs, and possibly be allocated more +-- resources. +-- +-- * The __higher__ the value, the “nicer” a job is to other jobs. Meaning it +-- will have less priority over other jobs, and possibly allow competing jobs +-- to take more resources. +-- +-- * Use 'nice0' (i.e,. @'Nice' 0@) as default 'Nice' value. +newtype Nice = Nice {int32 :: Int32} + deriving newtype (Eq, Ord, Show, Enum, Bounded) + +-- | @'nice0' = 'Nice' 0@ +nice0 :: Nice +nice0 = Nice 0 + +-------------------------------------------------------------------------------- + +-- | Unique identifier for the scheduled @job@ (and re-scheduled @job@, see +-- 'pull' and 'retry'). +newtype Id + = -- | Unsafe because there's no guarantee that the UUID is V7, + -- which this library expects. Use 'idFromUUID7' if possible. + UnsafeId UUID + deriving newtype (Eq, Ord, Show, Hashable) + +instance HasField "uuid7" Id UUID where + getField (UnsafeId u) = u + +unsafeIdFromUUID7 :: (HasCallStack) => UUID -> Id +unsafeIdFromUUID7 = maybe (error "unsafeIdFromUUID7") id . idFromUUID7 + +idFromUUID7 :: UUID -> Maybe Id +idFromUUID7 u = UnsafeId u <$ guard (UUID7.validate u) + +newId :: (MonadIO m) => m Id +newId = UnsafeId <$> UUID7.genUUID + +-------------------------------------------------------------------------------- + +-- | A @job@ together with its 'Queue' execution context details. +-- +-- As soon as you get your hands on a 'Work', which you do through 'pull', +-- start working on it right away. +data Work job = Work + { id :: Id + -- ^ Unique identifier for the scheduled @job@ (and re-scheduled @job@, see + -- 'pull' and 'retry'). + , job :: job + -- ^ The actual @job@ to be carried out. + , meta :: Meta + -- ^ 'Meta'data accompanying the @job@ being performed. + , retry :: Nice -> Time.UTCTime -> IO () + -- ^ Once this 'Work' is released, reschedule to be executed at the + -- specified 'Time.UTCTime' at the earliest. + -- + -- See the documentation for 'Queue'\'s 'pull'. + -- + -- @ + -- 'retry' _ _ '>>' 'retry' n t == 'retry' n t + -- 'finish' '>>' 'retry' n t == 'retry' n t + -- 'retry' n t '>>' 'finish' == 'finish' + -- @ + , finish :: IO () + -- ^ Once this 'Work' is released, remove it from the execution queue. + -- + -- See the documentation for 'Queue'\'s 'pull'. + -- + -- @ + -- 'finish' '>>' 'retry' n t == 'retry' n t + -- 'retry' n t '>>' 'finish' == 'finish' + -- 'finish' '>>' 'finish' == 'finish' + -- @ + } + deriving stock (Functor) + +-- | Like the 'retry' field in 'Work', except with a bit more +-- polymorphic type and intended to be used as a top-level function. +retry + :: forall job m + . (MonadIO m) + => Work job + -> Nice + -> Time.UTCTime + -> m () +retry Work{retry = f} n t = liftIO $ f n t + +-- | Like the 'finish' field in 'Work', except with a bit more +-- polymorphic type and intended to be used as a top-level function. +finish + :: forall job m + . (MonadIO m) + => Work job + -> m () +finish Work{finish = m} = liftIO m + +-- | A @job@ 'Queue'. +-- +-- * @job@s can be 'push'ed to the 'Queue' for eventual execution, 'pull'ed +-- from the 'Queue' for immediate execution, and the 'Queue' itself can be +-- pruned. +-- +-- * @"Job.Memory".'Job.Memory.queue'@ is an in-memory implementation that can +-- serve as reference. +-- +-- * Other backends are expected to provide a 'Queue' implementation. +data Queue job = Queue + { push :: Nice -> Time.UTCTime -> job -> IO Id + -- ^ Push new @job@ to the queue so to be executed after the specified + -- 'Time.UTCTime', which may be in the past. Throws if the 'Queue' has + -- already been released. + , pull :: A.Acquire (Maybe (Work job)) + -- ^ Pull some 'Work' from the queue + -- + -- * If the 'Queue' itself has been released, then returns 'Nothing' + -- right away. + -- + -- * Otherwise, blocks until 'Work' is available. + -- + -- * On 'A.ReleaseExceptionWith', the @job@ is automatically rescheduled + -- for re-execution after a few seconds. This behavior can be overriden + -- by using 'Work'\'s 'retry' or 'finish'. + -- + -- * On 'A.ReleaseNormal' or 'A.ReleaseEarly', the @job@ is automatically + -- removed from the 'Queue'. This behavior can be overriden + -- by using 'Work'\'s 'retry' or 'finish'. + , prune :: forall a. (Monoid a) => (Id -> Meta -> job -> (Bool, a)) -> IO a + -- ^ Prune @job@s from the 'Queue', keeping only those for which the given + -- function returns 'True' (like 'List.filter'). Allows collecting some + -- additional 'Monoid'al output. The given @job@s are in no particular + -- order. Throws if the 'Queue' has already been released. + -- __IMPORTANT:__ If you remove a @job@ that is currently active, + -- it might be 'push'ed back to the 'Queue' later if required by 'retry' + -- or a 'Work' exception. + } + +-- | Like the 'push' field in 'Queue', except with a bit more polymorphic type +-- and intended to be used as a top-level function. +push + :: forall job m + . (MonadIO m) + => Queue job + -> Nice + -> Time.UTCTime + -> job + -> m Id +push Queue{push = f} n t j = liftIO $ f n t j + +-- | Like the 'prune' field in 'Queue', except with a bit more polymorphic type +-- and intended to be used as a top-level function. +prune + :: forall job a m + . (Monoid a, MonadIO m) + => Queue job + -> (Id -> Meta -> job -> (Bool, a)) + -> m a +prune Queue{prune = f} g = liftIO $ f g + +-- | Wrapper for all the @job@-related data accessible through 'Queue'\'s +-- 'prune' function. +data Meta = Meta + { alive :: Maybe Time.UTCTime + -- ^ If 'Just', the @job@ is currently being 'Work'ed on, allegedly. The + -- last time the @job@ sent a heartbeat is attached. + , nice :: Nice + -- ^ 'Nice' value used while scheduling the @job@. + , wait :: Time.UTCTime + -- ^ Time until the 'Queue' is or was supposed to wait before considering + -- working on the @job@. + , try :: Word32 + -- ^ How many tries have been attempted already (excluding the current + -- one, in case 'alive' is 'Just'). + } + deriving + ( Eq + , Ord + -- ^ Order compatible with the one in 'prune'. + , Show + ) diff --git a/lib/Job/Memory.hs b/lib/Job/Memory.hs new file mode 100644 --- /dev/null +++ b/lib/Job/Memory.hs @@ -0,0 +1,323 @@ +{-# LANGUAGE StrictData #-} +{-# LANGUAGE NoFieldSelectors #-} + +module Job.Memory (queue) where + +import Control.Concurrent +import Control.Concurrent.Async qualified as As +import Control.Concurrent.STM +import Control.Exception.Safe qualified as Ex +import Control.Monad +import Control.Monad.IO.Class +import Control.Monad.Trans.Class (lift) +import Control.Monad.Trans.Maybe (MaybeT (..), runMaybeT) +import Control.Monad.Trans.Resource.Extra qualified as R +import Control.Monad.Trans.State.Strict (put, runStateT) +import Data.Acquire qualified as A +import Data.Fixed +import Data.Function +import Data.List qualified as List +import Data.Map.Strict (Map) +import Data.Map.Strict qualified as Map +import Data.Set (Set) +import Data.Set qualified as Set +import Data.Time qualified as Time +import Data.UUID.V7 (UUID) +import Data.UUID.V7 qualified as UUID7 +import GHC.IO.Exception +import GHC.Stack + +import Job + +-------------------------------------------------------------------------------- + +keepAliveBeat :: Micro +keepAliveBeat = 4 + +retryDelay :: Time.NominalDiffTime +retryDelay = 2 + +data Env job = Env + { jobs :: TVar (Map Id (Meta, job)) + , queued :: TVar (Set (Meta, Id)) + , worker :: TMVar (UUID, (Id, Meta, job) -> STM ()) + , active :: TVar Bool + } + +acqEnv :: A.Acquire (Env job) +acqEnv = do + jobs <- R.mkAcquire1 (newTVarIO mempty) \t -> do + atomically $ writeTVar t mempty + queued <- R.mkAcquire1 (newTVarIO mempty) \t -> do + atomically $ writeTVar t mempty + worker <- R.mkAcquire1 newEmptyTMVarIO \t -> + atomically $ void $ tryTakeTMVar t + active <- R.mkAcquire1 (newTVarIO True) \tv -> + atomically $ writeTVar tv False -- TODO wait until no workers running? + pure Env{..} + +ensureActive :: Env job -> STM () +ensureActive env = + readTVar env.active >>= \case + True -> pure () + False -> throwSTM $ resourceVanished "Job.Memory.queue" + +connect1 :: forall job. Env job -> Time.UTCTime -> STM (Maybe Id) +connect1 env now = + readTVar env.active >>= \case + False -> pure Nothing + True -> + nextJob >>= mapM \x@(i, _, _) -> do + (_, f) <- takeTMVar env.worker + f x >> pure i + where + nextJob :: STM (Maybe (Id, Meta, job)) + nextJob = do + q0 <- readTVar env.queued + forM (Set.minView q0) \((m0@Meta{..}, i), q1) -> do + writeTVar env.queued q1 + let m = Meta{alive = Just now, ..} + jobs0 <- readTVar env.jobs + (!jobs1, yj) <- flip runStateT Nothing do + Map.alterF + ( mapM \(m1, j) -> do + when (m0 /= m1) $ Ex.throwString "m0 /= m1" + put (Just j) + pure (m, j) + ) + i + jobs0 + case yj of + Just j -> writeTVar env.jobs jobs1 >> pure (i, m, j) + Nothing -> Ex.throwString "job in queue but not in map" + +connectMany :: forall job. Env job -> IO [Id] +connectMany env = fmap ($ []) (go id) + where + go :: ([Id] -> [Id]) -> IO ([Id] -> [Id]) + go f = do + -- We perform each connect1 in a separate transaction + -- so that we don't hog concurrent access to the Env. + now <- Time.getCurrentTime + atomically (connect1 env now) >>= \case + Just i -> go ((i :) . f) + Nothing -> pure f + +-- | An in-memory 'Queue'. +queue :: forall job. A.Acquire (Queue job) +queue = do + env <- acqEnv + void do + -- This background thread perform 'connectMany' every 60 seconds. + -- Not really necessary, just in case a timer died for some reason. + R.mkAcquire1 + ( As.async $ forever do + threadDelay 60_000_000 + void $ connectMany env + ) + As.uninterruptibleCancel + pure + Queue + { push = \nice wait job -> do + let m = Meta{nice, wait, alive = Nothing, try = 0} + i <- newId + atomically do + ensureActive env + modifyTVar' env.jobs $ Map.insert i (m, job) + modifyTVar' env.queued $ Set.insert (m, i) + void $ As.async do + threadDelayUTCTime wait + void $ connectMany env + pure i + , --------- + prune = \f -> + atomically do + ensureActive env + jobs0 <- readTVar env.jobs + let (js, qs, a) = + List.foldl' + ( \(!js0, !qs0, !al) (i, (m, j)) -> + case f i m j of + (False, ar) -> (js0, qs0, al <> ar) + (True, ar) -> + ( (i, (m, j)) : js0 + , maybe ((m, i) : qs0) (const qs0) m.alive + , al <> ar + ) + ) + mempty + $ List.sortOn (\(i, (m, _)) -> (m, i)) + $ Map.toList jobs0 + writeTVar env.jobs $! Map.fromList js + writeTVar env.queued $! Set.fromList qs + pure a + , --------- + pull = runMaybeT do + liftIO (readTVarIO env.active) >>= \case + False -> MaybeT $ pure Nothing + True -> pure () + tout :: TVar (Maybe (Maybe (Nice, Time.UTCTime))) <- + liftIO $ newTVarIO Nothing + (i :: Id, meta :: Meta, job :: job) <- MaybeT do + R.mkAcquireType1 + ( do + k0 <- UUID7.genUUID + Ex.bracketOnError + ( atomically do + readTVar env.active >>= \case + False -> pure Nothing + True -> do + tj <- newEmptyTMVar + -- Blocks until we become next worker + putTMVar env.worker (k0, putTMVar tj) + pure (Just tj) + ) + ( mapM_ \_ -> atomically do + tryTakeTMVar env.worker >>= \case + Just (k1, f) + | k0 /= k1 -> + putTMVar env.worker (k1, f) + _ -> pure () + ) + ( \case + Nothing -> pure Nothing + Just tj -> atomically do + readTVar env.active >>= \case + False -> pure Nothing + True -> + -- Blocks until we get job input + Just <$> takeTMVar tj + ) + ) + ( \yx rt -> forM_ yx \(i, m0, job) -> do + ym1 <- + readTVarIO tout >>= \case + -- no 'retry', and no 'finish', + -- and release with exception. + Nothing + | A.ReleaseExceptionWith _ <- rt -> do + now <- Time.getCurrentTime + let alive = Nothing + try = succ m0.try + nice = m0.nice + wait = Time.addUTCTime retryDelay now + pure $ Just Meta{..} + -- explicit 'retry'. + Just (Just (nice, wait)) -> do + let alive = Nothing + try = succ m0.try + pure $ Just Meta{..} + -- explicit 'finish', + -- or release without exception. + _ -> pure Nothing + case ym1 of + Nothing -> + atomically $ modifyTVar' env.jobs $ Map.delete i + Just !m1 -> do + atomically do + ensureActive env + modifyTVar' env.queued $ Set.insert (m1, i) + modifyTVar' env.jobs $ + Map.insert i (m1, job) + void $ As.async do + threadDelayUTCTime m1.wait + void $ connectMany env + ) + -- While working on this job, send heartbeats + void $ lift do + R.mkAcquire1 + ( As.async $ fix \again -> do + eactive <- Ex.tryAny do + threadDelayMicro keepAliveBeat + now <- Time.getCurrentTime + let m1 + | Meta{..} <- meta = + Meta{alive = Just now, ..} + atomically do + a <- readTVar env.active + when a do + modifyTVar' env.jobs $ + Map.insert i (m1, job) + pure a + case eactive of + Right False -> pure () + _ -> again + ) + As.uninterruptibleCancel + pure + Work + { id = i + , job + , meta + , retry = \n w -> + atomically $ writeTVar tout $ Just (Just (n, w)) + , finish = atomically $ writeTVar tout $ Just Nothing + } + } + +-------------------------------------------------------------------------------- + +-- | Like 'threadDelay', but waits until a specified 'Time.UTCTime'. +threadDelayUTCTime :: Time.UTCTime -> IO () +threadDelayUTCTime wait = do + t <- registerDelayUTCTime wait + atomically $ readTVar t >>= check + +-- | Like 'registerDelay', but waits until a specified 'Time.UTCTime'. +registerDelayUTCTime :: Time.UTCTime -> IO (TVar Bool) +registerDelayUTCTime wait = do + start <- Time.getCurrentTime + if wait <= start + then newTVarIO True + else do + tvtmp <- + registerDelayMicro $ + ceilingPicoToMicro $ + Time.nominalDiffTimeToSeconds $ + Time.diffUTCTime wait start + tvout <- newTVarIO False + void $ Ex.mask_ $ As.async do + atomically $ readTVar tvtmp >>= check + fix \again -> do + now <- Time.getCurrentTime + if wait <= now + then atomically $ writeTVar tvout True + else do + -- Only happens if there were leap seconds involved. + -- We just loop innefficiently for a bit. + threadDelay 10_000 -- 10 milliseconds + again + pure tvout + +ceilingPicoToMicro :: Pico -> Micro +ceilingPicoToMicro (MkFixed p) = MkFixed (ceiling (toRational p / 1_000_000)) + +-- | Like 'registerDelay', but not limited to @'maxBound' :: 'Int'@. +registerDelayMicro :: Micro -> IO (TVar Bool) +registerDelayMicro us + | us <= 0 = newTVarIO True + | otherwise = do + tv <- liftIO $ newTVarIO False + void $ Ex.mask_ $ As.async do + threadDelayMicro us + atomically $ writeTVar tv True + pure tv + +-- | Like 'threadDelay', but not limited to @'maxBound' :: 'Int'@. +threadDelayMicro :: Micro -> IO () +threadDelayMicro = fix \again us -> when (us > 0) do + let d = floor (min us stepMax * 1_000_000) + threadDelay d + again (us - stepMax) + where + stepMax :: Micro + stepMax = MkFixed (fromIntegral (maxBound :: Int)) + +-------------------------------------------------------------------------------- + +resourceVanished :: (HasCallStack) => String -> IOError +resourceVanished s = + (userError s) + { ioe_location = prettyCallStack (popCallStack callStack) + , ioe_type = ResourceVanished + } diff --git a/test/Main.hs b/test/Main.hs new file mode 100644 --- /dev/null +++ b/test/Main.hs @@ -0,0 +1,191 @@ +{-# OPTIONS_GHC -Wno-incomplete-patterns #-} + +module Main (main) where + +import Control.Concurrent +import Control.Exception qualified as Ex +import Control.Monad +import Data.Acquire qualified as A +import Data.Function +import Data.IORef +import Data.List qualified as List +import Data.Time qualified as Time +import Job +import Job.Memory qualified +import System.Random.Stateful qualified as R + +-------------------------------------------------------------------------------- + +infix 4 ==. +(==.) :: (Eq a, Show a) => a -> a -> IO () +(==.) l r = + if l == r + then mempty + else fail $ (showsPrec 11 l . showString " /= " . showsPrec 11 r) "" + +infix 4 >. +(>.) :: (Ord a, Show a) => a -> a -> IO () +(>.) l r = + if l > r + then mempty + else fail $ (showsPrec 11 l . showString " <= " . showsPrec 11 r) "" + +main :: IO () +main = A.withAcquire (Job.Memory.queue @String) \q -> do + t0 <- Time.getCurrentTime + + putStrLn "a" + [] <- prune q \i m j -> (False, [(i, m, j)]) + + putStrLn "b" + i0 <- push q nice0 t0 "j0" + [(x0i, x0m, x0j)] <- prune q \i m j -> (False, [(i, m, j)]) + x0i ==. i0 + x0j ==. "j0" + x0m.nice ==. nice0 + x0m.wait ==. t0 + x0m.try ==. 0 + x0m.alive ==. Nothing + [] <- prune q \i m j -> (False, [(i, m, j)]) + + putStrLn "c" + i1 <- push q nice0 t0 "j1" + [(x1i, x1m, x1j)] <- prune q \i m j -> (True, [(i, m, j)]) + x1i ==. i1 + x1j ==. "j1" + x1m.nice ==. nice0 + x1m.wait ==. t0 + x1m.try ==. 0 + x1m.alive ==. Nothing + + putStrLn "d" + [(x2i, x2m, x2j)] <- prune q \i m j -> (True, [(i, m, j)]) + x2i ==. i1 + x2j ==. "j1" + x2m.nice ==. nice0 + x2m.wait ==. t0 + x2m.try ==. 0 + x2m.alive ==. Nothing + + putStrLn "e" + t1 <- Time.getCurrentTime + A.withAcquire q.pull $ mapM_ \w -> do + putStrLn "e'" + w.id ==. i1 + w.job ==. "j1" + w.meta.nice ==. nice0 + w.meta.wait ==. t0 + w.meta.try ==. 0 + retry w (Nice 1) t1 + + putStrLn "f" + [(x3i, x3m, x3j)] <- prune q \i m j -> (True, [(i, m, j)]) + x3i ==. i1 + x3j ==. "j1" + x3m.nice ==. Nice 1 + x3m.wait ==. t1 + x3m.try ==. 1 + x3m.alive ==. Nothing + + putStrLn "g" + A.withAcquire q.pull $ mapM_ \w -> do + putStrLn "g'" + w.id ==. i1 + w.job ==. "j1" + w.meta.nice ==. Nice 1 + w.meta.wait ==. t1 + w.meta.try ==. 1 + retry w (Nice 1) t1 + + putStrLn "h" + [(x4i, x4m, x4j)] <- prune q \i m j -> (True, [(i, m, j)]) + x4i ==. i1 + x4j ==. "j1" + x4m.nice ==. Nice 1 + x4m.wait ==. t1 + x4m.try ==. 2 + x4m.alive ==. Nothing + + putStrLn "i" + A.withAcquire q.pull $ mapM_ \w -> do + putStrLn "i'" + w.id ==. i1 + w.job ==. "j1" + w.meta.nice ==. Nice 1 + w.meta.wait ==. t1 + w.meta.try ==. 2 + pure () + + putStrLn "j" + [] <- prune q \i m j -> (True, [(i, m, j)]) + t2 <- Time.addUTCTime 0.3 <$> Time.getCurrentTime + i2 <- push q nice0 t2 "j2" + [(x5i, x5m, x5j)] <- prune q \i m j -> (True, [(i, m, j)]) + x5i ==. i2 + x5j ==. "j2" + x5m.nice ==. nice0 + x5m.wait ==. t2 + x5m.try ==. 0 + x5m.alive ==. Nothing + + putStrLn "k" + Ex.handle (\case Err 1 -> pure ()) do + A.withAcquire q.pull $ mapM_ \w -> do + putStrLn "k'" + w.id ==. i2 + w.job ==. "j2" + w.meta.nice ==. nice0 + w.meta.wait ==. t2 + w.meta.try ==. 0 + void $ Ex.throwIO $ Err 1 + + putStrLn "l" + [(x6i, x6m, x6j)] <- prune q \i m j -> (True, [(i, m, j)]) + x6i ==. i2 + x6j ==. "j2" + x6m.nice ==. nice0 + x6m.wait >. t2 + x6m.try ==. 1 + x6m.alive ==. Nothing + + putStrLn "m" + i3 <- push q nice0 t0 "j3" + i4 <- push q (pred nice0) t1 "j4" + i5 <- push q nice0 t0 "j5" + jIds0 <- prune q \i _ _ -> (True, [i]) + jIds0 ==. [i4, i3, i5, i2] + + putStrLn "n" + forM_ (zip jIds0 (List.drop 1 (List.tails jIds0))) \(jId, tl) -> + A.withAcquire q.pull $ mapM_ \w -> do + w.id ==. jId + jIds1 <- prune q \i _ _ -> (True, [i]) + jIds1 ==. tl <> [jId] + [] <- prune q \i m j -> (True, [(i, m, j)]) + + putStrLn "o" + do + t3 <- Time.getCurrentTime + rwout <- newIORef ([] :: [String]) + let wins :: [String] = fmap show [0 .. 1000 :: Word] + forM_ wins \win -> do + wrd :: Word <- R.uniformM R.globalStdGen + let pct = toRational wrd / toRational (maxBound :: Word) + ndt = fromRational (pct * 0.5) + push q nice0 (Time.addUTCTime ndt t3) win + putStrLn "o'" + replicateM_ 4 $ forkIO $ fix \again -> do + y <- A.withAcquire q.pull $ mapM \w -> + atomicModifyIORef' rwout \xs -> (w.job : xs, ()) + forM_ y \_ -> again + threadDelay 3_100_000 + [] <- prune q \i m j -> (True, [(i, m, j)]) + wout <- readIORef rwout + length wout ==. length wins + List.sort wout ==. List.sort wins + + putStrLn "done" + +data Err = Err Int + deriving stock (Eq, Show) + deriving anyclass (Ex.Exception)