git.iko.soy/rowtype-yoga/psfp
on 3d6eb2f help sign in
rowtype-yoga/psfp/server/src/JobQueue.purs
186 lines 5.6 kB history raw
1module JobQueue where
3import Prelude
5import Data.Array (delete, filter, find, length, snoc, zip)
6import Data.Foldable (for_)
7import Data.Generic.Rep (class Generic)
8import Data.Generic.Rep.Show (genericShow)
9import Data.JSDate (JSDate)
10import Data.JSDate as JSDate
11import Data.Maybe (Maybe(..))
12import Data.Newtype (class Newtype, over, un)
13import Data.Time.Duration (Milliseconds(..))
14import Data.Tuple.Nested ((/\))
15import Data.UUID (UUID, genUUID)
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
23 = JobId UUID
25derive instance ntJobId :: Newtype JobId _
26derive instance genericJobId :: Generic JobId _
27instance showJobId :: Show JobId where
28 show = genericShow
30derive newtype instance jobIdEq :: Eq JobId
31newtype NewJob a
32 = 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
62 = RunningJob
63 { id ∷ JobId
64 , addedAt ∷ JSDate
65 , startedAt ∷ JSDate
66 , fiber ∷ Fiber Unit
67 , resource ∷ a
68 }
70type QueueParams
71 = { maxSize ∷ Int
72 , timeout ∷ Milliseconds
73 }
75newtype ResourcePool a
76 = ResourcePool (Array a)
78derive instance ntResourcePool :: Newtype (ResourcePool a) _
79data Queue a
80 = Queue QueueParams (Ref (ResourcePool a)) (Ref (Array (Job a)))
82mkQueue ∷ ∀ m a. MonadEffect m => QueueParams -> ResourcePool a -> m (Queue a)
83mkQueue params pool =
84 liftEffect do
85 poolRef <- Ref.new pool
86 jobsRef <- Ref.new []
87 pure $ Queue params poolRef jobsRef
89data EnqueueResult
90 = Enqueued JobId
91 | QueueFull
93derive instance eqEnqueueResult :: Eq EnqueueResult
94derive 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
104 remove = removeFromQueue id q
105 fiber <-
106 launchAff <<< sequential
107 $ parallel (job resource *> remove)
108 *> parallel (delay timeout *> remove)
109 let
110 running = RunningJob { id, fiber, addedAt, startedAt, resource }
111 jobsRef
112 # Ref.modify_ (map \j -> if getId j == id then Running running else j)
114jobsInQueue ∷ ∀ a. Queue a -> Effect Int
115jobsInQueue (Queue _ _ jobsRef) = do
116 jobs <- Ref.read jobsRef
117 pure $ length jobs
119runningJobs ∷ ∀ a. Queue a -> Effect (Array (RunningJob a))
120runningJobs (Queue _ resourceRef jobsRef) = do
121 jobs <- Ref.read jobsRef
122 pure do
123 job <- jobs
124 case job of
125 Running rj -> [ rj ]
126 Pending _ -> []
128pendingJobs ∷ ∀ a. Queue a -> Effect (Array (PendingJob a))
129pendingJobs (Queue _ resourceRef jobsRef) = do
130 jobs <- Ref.read jobsRef
131 pool <- Ref.read resourceRef
132 pure do
133 job <- jobs
134 case job of
135 Running _ -> []
136 Pending pj -> [ pj ]
138startJobs ∷ ∀ a. Eq a => Queue a -> Effect Unit
139startJobs q@(Queue { timeout } poolRef jobsRef) = do
140 pool <- Ref.read poolRef
141 pending <- pendingJobs q
142 for_ (pending `zip` un ResourcePool pool) \(job /\ resource) -> do
143 Ref.modify_ (over ResourcePool $ delete resource) poolRef
144 startJob q resource job
146removeStaleJobs ∷ ∀ a. Eq a => Queue a -> Effect Unit
147removeStaleJobs q@(Queue { timeout } resourceRef jobsRef) = do
148 running <- runningJobs q
149 nowMillis <- JSDate.now <#> JSDate.getTime
150 for_ running
151 $ \(RunningJob { startedAt, id }) -> do
152 let
153 runningSince = (nowMillis - JSDate.getTime startedAt) # Milliseconds
155 timedOut = runningSince > timeout
156 when timedOut $ launchAff_ (removeFromQueue id q)
158removeFromQueue ∷ ∀ a. Eq a => JobId -> Queue a -> Aff Unit
159removeFromQueue id q@(Queue _ poolRef jobsRef) = do
160 maybeJob <- Ref.read jobsRef <#> find (\x -> (getId x) == id) # liftEffect
161 -- remove from queue
162 jobsRef # Ref.modify_ (filter (\x -> (getId x) /= id)) # liftEffect
163 case maybeJob of
164 Just (Running (RunningJob { fiber })) -> killFiber (error "Removed from queue") fiber
165 _ -> pure unit
166 -- free resource
167 case maybeJob of
168 Just (Running (RunningJob { resource })) -> do
169 poolRef # Ref.modify_ (over ResourcePool (_ `snoc` resource)) # liftEffect
170 _ -> pure unit
171 startJobs q # liftEffect
173--| Add to the end of the queue. If the queue is full the result will be `false`
174enqueue ∷ ∀ a. Eq a => NewJob a -> Queue a -> Effect EnqueueResult
175enqueue (NewJob newJob) q@(Queue { maxSize, timeout } _ jobsRef) = do
176 id <- genUUID <#> JobId
177 jobs <- Ref.read jobsRef
178 if length jobs >= maxSize then
179 pure QueueFull
180 else do
181 addedAt <- JSDate.now
182 let
183 new = Pending (PendingJob { id, addedAt, job: newJob id })
184 jobsRef # Ref.modify_ (_ `snoc` new)
185 startJobs q
186 pure (Enqueued id)