job (empty) → 0.1
raw patch · 7 files changed
+1068/−0 lines, 7 filesdep +asyncdep +basedep +containers
Dependencies added: async, base, containers, hashable, job, mmzk-typeid, random, resourcet, resourcet-extra, safe-exceptions, stm, time, transformers
Files
- CHANGELOG.md +29/−0
- LICENSE +219/−0
- README.md +18/−0
- job.cabal +65/−0
- lib/Job.hs +223/−0
- lib/Job/Memory.hs +323/−0
- test/Main.hs +191/−0
+ CHANGELOG.md view
@@ -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.
+ LICENSE view
@@ -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.
+ README.md view
@@ -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.+
+ job.cabal view
@@ -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,
+ lib/Job.hs view
@@ -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+ )
+ lib/Job/Memory.hs view
@@ -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+ }
+ test/Main.hs view
@@ -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)