git.iko.soy/rowtype-yoga/psfp
on f86fac1 help sign in
rowtype-yoga/psfp/server/src/JobQueue.purs
181 lines 5.6 kB history raw
1module JobQueue where
3import Prelude
4import Data.Array (delete, filter, find, length, snoc, zip)
5import Data.Foldable (for_)
6import Data.Generic.Rep (class Generic)
7import Data.Generic.Rep.Show (genericShow)
8import Data.JSDate (JSDate)
9import Data.JSDate as JSDate
10import Data.Maybe (Maybe(..))
11import Data.Newtype (class Newtype, over, un)
12import Data.Time.Duration (Milliseconds(..))
13import Data.Tuple.Nested ((/\))
14import Data.UUID (UUID, genUUID)
15import Effect (Effect)
16import Effect.Aff (Aff, Fiber, delay, error, killFiber, launchAff, launchAff_, parallel, sequential)
17import Effect.Class (class MonadEffect, liftEffect)
18import Effect.Ref (Ref)
19import Effect.Ref as Ref
21newtype JobId = JobId UUID
23derive instance ntJobId ∷ Newtype JobId _
24derive instance genericJobId ∷ Generic JobId _
26instance showJobId ∷ Show JobId where
27 show = genericShow
29derive newtype instance jobIdEq ∷ Eq JobId
31newtype NewJob a = NewJob (JobId -> a -> Aff Unit)
33data PendingJob a
34 = PendingJob
35 { id ∷ JobId
36 , job ∷ a -> Aff Unit
37 , addedAt ∷ JSDate
38 }
40data Job a
41 = Pending (PendingJob a)
42 | Running (RunningJob a)
44getAddedAt ∷ ∀ a. Job a -> JSDate
45getAddedAt = case _ of
46 Pending (PendingJob { addedAt }) -> addedAt
47 Running (RunningJob { addedAt }) -> addedAt
49instance eqJob ∷ Eq (Job a) where
50 eq j1 j2 = getId j1 == getId j2
52instance ordJob ∷ Ord (Job a) where
53 compare j1 j2 = compare (getAddedAt j1) (getAddedAt j2)
55getId ∷ ∀ a. Job a -> JobId
56getId = case _ of
57 Pending (PendingJob { id }) -> id
58 Running (RunningJob { id }) -> id
60newtype RunningJob a = RunningJob
61 { id ∷ JobId
62 , addedAt ∷ JSDate
63 , startedAt ∷ JSDate
64 , fiber ∷ Fiber Unit
65 , resource ∷ a
66 }
68type QueueParams =
69 { maxSize ∷ Int
70 , timeout ∷ Milliseconds
71 }
73newtype ResourcePool a = ResourcePool (Array a)
75derive instance ntResourcePool ∷ Newtype (ResourcePool a) _
77data Queue a
78 = Queue QueueParams (Ref (ResourcePool a)) (Ref (Array (Job a)))
80mkQueue ∷ ∀ m a. MonadEffect m => QueueParams -> ResourcePool a -> m (Queue a)
81mkQueue params pool =
82 liftEffect do
83 poolRef <- Ref.new pool
84 jobsRef <- Ref.new []
85 pure $ Queue params poolRef jobsRef
87data EnqueueResult
88 = Enqueued JobId
89 | QueueFull
91derive instance eqEnqueueResult ∷ Eq EnqueueResult
92derive instance genericEnqueueResult ∷ Generic EnqueueResult _
94instance showEnqueueResult ∷ Show EnqueueResult where
95 show = genericShow
97startJob ∷ ∀ a. Eq a => Queue a -> a -> PendingJob a -> Effect Unit
98startJob q@(Queue { timeout } _ jobsRef) resource (PendingJob { id, job, addedAt }) = do
99 startedAt <- JSDate.now
100 iso <- JSDate.toISOString startedAt
101 -- log $ "Starting Job: " <> toString (unwrap id) <> " at: " <> iso
102 let remove = removeFromQueue id q
103 fiber <-
104 launchAff <<< sequential
105 $ parallel (job resource *> remove)
106 *> parallel (delay timeout *> remove)
107 let running = RunningJob { id, fiber, addedAt, startedAt, resource }
108 jobsRef
109 # Ref.modify_ (map \j -> if getId j == id then Running running else j)
111jobsInQueue ∷ ∀ a. Queue a -> Effect Int
112jobsInQueue (Queue _ _ jobsRef) = do
113 jobs <- Ref.read jobsRef
114 pure $ length jobs
116runningJobs ∷ ∀ a. Queue a -> Effect (Array (RunningJob a))
117runningJobs (Queue _ resourceRef jobsRef) = do
118 jobs <- Ref.read jobsRef
119 pure do
120 job <- jobs
121 case job of
122 Running rj -> [ rj ]
123 Pending _ -> []
125pendingJobs ∷ ∀ a. Queue a -> Effect (Array (PendingJob a))
126pendingJobs (Queue _ resourceRef jobsRef) = do
127 jobs <- Ref.read jobsRef
128 pool <- Ref.read resourceRef
129 pure do
130 job <- jobs
131 case job of
132 Running _ -> []
133 Pending pj -> [ pj ]
135startJobs ∷ ∀ a. Eq a => Queue a -> Effect Unit
136startJobs q@(Queue { timeout } poolRef jobsRef) = do
137 pool <- Ref.read poolRef
138 pending <- pendingJobs q
139 for_ (pending `zip` un ResourcePool pool) \(job /\ resource) -> do
140 Ref.modify_ (over ResourcePool $ delete resource) poolRef
141 startJob q resource job
143removeStaleJobs ∷ ∀ a. Eq a => Queue a -> Effect Unit
144removeStaleJobs q@(Queue { timeout } resourceRef jobsRef) = do
145 running <- runningJobs q
146 nowMillis <- JSDate.now <#> JSDate.getTime
147 for_ running
148 $ \(RunningJob { startedAt, id }) -> do
149 let
150 runningSince = (nowMillis - JSDate.getTime startedAt) # Milliseconds
151 timedOut = runningSince > timeout
152 when timedOut $ launchAff_ (removeFromQueue id q)
154removeFromQueue ∷ ∀ a. Eq a => JobId -> Queue a -> Aff Unit
155removeFromQueue id q@(Queue _ poolRef jobsRef) = do
156 maybeJob <- Ref.read jobsRef <#> find (\x -> (getId x) == id) # liftEffect
157 -- remove from queue
158 jobsRef # Ref.modify_ (filter (\x -> (getId x) /= id)) # liftEffect
159 case maybeJob of
160 Just (Running (RunningJob { fiber })) -> killFiber (error "Removed from queue") fiber
161 _ -> pure unit
162 -- free resource
163 case maybeJob of
164 Just (Running (RunningJob { resource })) -> do
165 poolRef # Ref.modify_ (over ResourcePool (_ `snoc` resource)) # liftEffect
166 _ -> pure unit
167 startJobs q # liftEffect
169--| Add to the end of the queue. If the queue is full the result will be `false`
170enqueue ∷ ∀ a. Eq a => NewJob a -> Queue a -> Effect EnqueueResult
171enqueue (NewJob newJob) q@(Queue { maxSize, timeout } _ jobsRef) = do
172 id <- genUUID <#> JobId
173 jobs <- Ref.read jobsRef
174 if length jobs >= maxSize then
175 pure QueueFull
176 else do
177 addedAt <- JSDate.now
178 let new = Pending (PendingJob { id, addedAt, job: newJob id })
179 jobsRef # Ref.modify_ (_ `snoc` new)
180 startJobs q
181 pure (Enqueued id)