git.iko.soy/rowtype-yoga/psfp
on 110e143 help sign in
rowtype-yoga/psfp/server/src/JobQueue.purs
182 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.Show.Generic (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 Data.UUID as UUID
16import Effect (Effect)
17import Effect.Aff (Aff, Fiber, delay, error, killFiber, launchAff, launchAff_, parallel, sequential)
18import Effect.Class (class MonadEffect, liftEffect)
19import Effect.Ref (Ref)
20import Effect.Ref as Ref
22newtype JobId = JobId UUID
24derive instance ntJobId ∷ Newtype JobId _
25derive instance genericJobId ∷ Generic JobId _
27instance showJobId ∷ Show JobId where
28 show = genericShow
30derive newtype instance jobIdEq ∷ Eq JobId
32newtype NewJob a = NewJob (JobId -> a -> Aff Unit)
34data PendingJob a
35 = PendingJob
36 { id ∷ JobId
37 , job ∷ a -> Aff Unit
38 , addedAt ∷ JSDate
39 }
41data Job a
42 = Pending (PendingJob a)
43 | Running (RunningJob a)
45getAddedAt ∷ ∀ a. Job a -> JSDate
46getAddedAt = case _ of
47 Pending (PendingJob { addedAt }) -> addedAt
48 Running (RunningJob { addedAt }) -> addedAt
50instance eqJob ∷ Eq (Job a) where
51 eq j1 j2 = getId j1 == getId j2
53instance ordJob ∷ Ord (Job a) where
54 compare j1 j2 = compare (getAddedAt j1) (getAddedAt j2)
56getId ∷ ∀ a. Job a -> JobId
57getId = case _ of
58 Pending (PendingJob { id }) -> id
59 Running (RunningJob { id }) -> id
61newtype RunningJob a = RunningJob
62 { id ∷ JobId
63 , addedAt ∷ JSDate
64 , startedAt ∷ JSDate
65 , fiber ∷ Fiber Unit
66 , resource ∷ a
67 }
69type QueueParams =
70 { maxSize ∷ Int
71 , timeout ∷ Milliseconds
72 }
74newtype ResourcePool a = ResourcePool (Array a)
76derive instance ntResourcePool ∷ Newtype (ResourcePool a) _
78data Queue a
79 = Queue QueueParams (Ref (ResourcePool a)) (Ref (Array (Job a)))
81mkQueue ∷ ∀ m a. MonadEffect m => QueueParams -> ResourcePool a -> m (Queue a)
82mkQueue params pool =
83 liftEffect do
84 poolRef <- Ref.new pool
85 jobsRef <- Ref.new []
86 pure $ Queue params poolRef jobsRef
88data EnqueueResult
89 = Enqueued JobId
90 | QueueFull
92derive instance eqEnqueueResult ∷ Eq EnqueueResult
93derive instance genericEnqueueResult ∷ Generic EnqueueResult _
95instance showEnqueueResult ∷ Show EnqueueResult where
96 show = genericShow
98startJob ∷ ∀ a. Eq a => Queue a -> a -> PendingJob a -> Effect Unit
99startJob q@(Queue { timeout } _ jobsRef) resource (PendingJob { id, job, addedAt }) = do
100 startedAt <- JSDate.now
101 iso <- JSDate.toISOString startedAt
102 -- log $ "Starting Job: " <> toString (unwrap id) <> " at: " <> iso
103 let remove = removeFromQueue id q
104 fiber <-
105 launchAff <<< sequential
106 $ parallel (job resource *> remove)
107 *> parallel (delay timeout *> remove)
108 let running = RunningJob { id, fiber, addedAt, startedAt, resource }
109 jobsRef
110 # Ref.modify_ (map \j -> if getId j == id then Running running else j)
112jobsInQueue ∷ ∀ a. Queue a -> Effect Int
113jobsInQueue (Queue _ _ jobsRef) = do
114 jobs <- Ref.read jobsRef
115 pure $ length jobs
117runningJobs ∷ ∀ a. Queue a -> Effect (Array (RunningJob a))
118runningJobs (Queue _ resourceRef jobsRef) = do
119 jobs <- Ref.read jobsRef
120 pure do
121 job <- jobs
122 case job of
123 Running rj -> [ rj ]
124 Pending _ -> []
126pendingJobs ∷ ∀ a. Queue a -> Effect (Array (PendingJob a))
127pendingJobs (Queue _ resourceRef jobsRef) = do
128 jobs <- Ref.read jobsRef
129 pool <- Ref.read resourceRef
130 pure do
131 job <- jobs
132 case job of
133 Running _ -> []
134 Pending pj -> [ pj ]
136startJobs ∷ ∀ a. Eq a => Queue a -> Effect Unit
137startJobs q@(Queue { timeout } poolRef jobsRef) = do
138 pool <- Ref.read poolRef
139 pending <- pendingJobs q
140 for_ (pending `zip` un ResourcePool pool) \(job /\ resource) -> do
141 Ref.modify_ (over ResourcePool $ delete resource) poolRef
142 startJob q resource job
144removeStaleJobs ∷ ∀ a. Eq a => Queue a -> Effect Unit
145removeStaleJobs q@(Queue { timeout } resourceRef jobsRef) = do
146 running <- runningJobs q
147 nowMillis <- JSDate.now <#> JSDate.getTime
148 for_ running
149 $ \(RunningJob { startedAt, id }) -> do
150 let
151 runningSince = (nowMillis - JSDate.getTime startedAt) # Milliseconds
152 timedOut = runningSince > timeout
153 when timedOut $ launchAff_ (removeFromQueue id q)
155removeFromQueue ∷ ∀ a. Eq a => JobId -> Queue a -> Aff Unit
156removeFromQueue id q@(Queue _ poolRef jobsRef) = do
157 maybeJob <- Ref.read jobsRef <#> find (\x -> (getId x) == id) # liftEffect
158 -- remove from queue
159 jobsRef # Ref.modify_ (filter (\x -> (getId x) /= id)) # liftEffect
160 case maybeJob of
161 Just (Running (RunningJob { fiber })) -> killFiber (error "Removed from queue") fiber
162 _ -> pure unit
163 -- free resource
164 case maybeJob of
165 Just (Running (RunningJob { resource })) -> do
166 poolRef # Ref.modify_ (over ResourcePool (_ `snoc` resource)) # liftEffect
167 _ -> pure unit
168 startJobs q # liftEffect
170--| Add to the end of the queue. If the queue is full the result will be `false`
171enqueue ∷ ∀ a. Eq a => NewJob a -> Queue a -> Effect EnqueueResult
172enqueue (NewJob newJob) q@(Queue { maxSize, timeout } _ jobsRef) = do
173 id <- genUUID <#> JobId
174 jobs <- Ref.read jobsRef
175 if length jobs >= maxSize then
176 pure QueueFull
177 else do
178 addedAt <- JSDate.now
179 let new = Pending (PendingJob { id, addedAt, job: newJob id })
180 jobsRef # Ref.modify_ (_ `snoc` new)
181 startJobs q
182 pure (Enqueued id)