Combination of QueueManagers

5 files changed, 103 insertions, 25 deletions

diff --git a/server/src/Thermoprint/Server.hs b/server/src/Thermoprint/Server.hs
index 7767c12..8061b20 100644
--- a/server/src/Thermoprint/Server.hs
+++ b/server/src/Thermoprint/Server.hs
@@ -12,6 +12,7 @@ module Thermoprint.Server
        , module Data.Default.Class
        , module Servant.Server.Internal.Enter
        , module Thermoprint.Server.Printer
+       , module Thermoprint.Server.Queue
        ) where
 
 import Data.Default.Class
@@ -49,6 +50,7 @@ import Thermoprint.API (thermoprintAPI, PrinterId)
 
 import Thermoprint.Server.Database
 import Thermoprint.Server.Printer
+import Thermoprint.Server.Queue
 import qualified Thermoprint.Server.API as API (thermoprintServer)
 import Thermoprint.Server.API hiding (thermoprintServer)
 
diff --git a/server/src/Thermoprint/Server/Queue.hs b/server/src/Thermoprint/Server/Queue.hs
index 68dc7a9..69295bb 100644
--- a/server/src/Thermoprint/Server/Queue.hs
+++ b/server/src/Thermoprint/Server/Queue.hs
@@ -1,4 +1,4 @@
-{-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE FlexibleInstances, FlexibleContexts #-}
 {-# LANGUAGE ViewPatterns #-}
 {-# LANGUAGE RecordWildCards #-}
 {-# LANGUAGE DeriveGeneric, DeriveAnyClass #-}
@@ -7,7 +7,9 @@
 module Thermoprint.Server.Queue
        ( Queue(..), QueueEntry(..)
        , HasQueue(..)
-       , QueueManager(..), runQM
+       , QueueManager, QueueManagerM, runQM
+       , intersection, idQM
+       , union, nullQM
        ) where
 
 import           Thermoprint.API (PrintingError(..), Printout)
@@ -15,11 +17,14 @@ import qualified Thermoprint.API as API (JobStatus(..))
 
 import           Thermoprint.Server.Database
 
-import           Data.Sequence (Seq, ViewL(..), viewl)
+import           Data.Sequence (Seq, ViewL(..), viewl, (|>), (<|))
 import qualified Data.Sequence as Seq
+import           Data.Set (Set)
+import qualified Data.Set as Set
 
 import           Data.Time
-import           Data.Time.Clock
+import           Data.ExtendedReal
+import           Data.Fixed
 
 import           Control.DeepSeq (NFData)
 import           Data.Typeable (Typeable)
@@ -31,10 +36,14 @@ import           Control.Monad.State
 
 import           Data.Default.Class
 
+import           Control.Monad
 import           Control.Monad.Morph
 import           Control.Monad.Trans.Compose
+import           Data.Foldable
+import           Data.Function
 
 import           Data.Monoid
+import           Data.Ord
 
 -- | Zipper for 'Seq QueueEntry' with additional support for 'PrintingError' in the section after point
 data Queue = Queue
@@ -44,17 +53,6 @@ data Queue = Queue
   }
   deriving (Typeable, Generic, NFData)
 
-toSeq :: Queue -> Seq (Bool, QueueEntry, Maybe PrintingError)
-toSeq Queue{..} = fmap (\x -> (False, x, Nothing)) pending <> maybe Seq.empty (\c -> Seq.singleton (True, c, Nothing)) current <> fmap (\(x, p) -> (False, x, p)) history
-
-fromSeq :: Seq (Bool, QueueEntry, Maybe PrintingError) -> Queue
-fromSeq s = Queue pending' current' history'
-  where
-    (fmap (\(_, x, _) -> x) -> pending', pending'') = Seq.breakl (\(c, _, _) -> c) s
-    (current', history') = case viewl pending'' of
-      EmptyL -> (Nothing, Seq.empty)
-      (_, a, _) :< as -> (Just a, fmap (\(_, x, p) -> (x, p)) as)
-
 class HasQueue a where
   extractQueue :: a -> TVar Queue
 
@@ -72,10 +70,59 @@ data QueueEntry = QueueEntry
   { jobId :: JobId
   , created :: UTCTime
   }
-  deriving (Typeable, Generic, NFData)
+  deriving (Typeable, Generic, NFData, Eq, Ord)
+
+data QueueItem = Pending Int QueueEntry | Current QueueEntry | History Int QueueEntry (Maybe PrintingError)
+
+instance Eq QueueItem where
+  (Pending i a  ) == (Pending j b  ) = i == j && a == b
+  (Current   a  ) == (Current   b  ) =           a == b
+  (History i a _) == (History j b _) = i == j && a == b
+  _ == _ = False
+
+instance Ord QueueItem where
+  (Pending i a  ) `compare` (Pending j b  ) = compare i j <> compare a b
+  (Current   a  ) `compare` (Current   b  ) = compare a b
+  (History i a _) `compare` (History j b _) = compare i j <> compare a b
+  (Pending _ _  ) `compare` _               = LT
+  (Current   _  ) `compare` (Pending _ _  ) = GT
+  (Current   _  ) `compare` _               = LT
+  (History _ _ _) `compare` _               = GT
+
+newtype PlainQueueItem = Plain { unPlain :: QueueItem }
+
+instance Eq PlainQueueItem where
+  (unPlain -> Pending _ a  ) == (unPlain -> Pending _ b  ) = a == b
+  (unPlain -> Current   a  ) == (unPlain -> Current   b  ) = a == b
+  (unPlain -> History _ a _) == (unPlain -> History _ b _) = a == b
+  _ == _ = False
+
+instance Ord PlainQueueItem where
+  (unPlain -> Pending _ a  ) <= (unPlain -> Pending _ b  ) = a <= b
+  (unPlain -> Current   a  ) <= (unPlain -> Current   b  ) = a <= b
+  (unPlain -> History _ a _) <= (unPlain -> History _ b _) = a <= b
+  (unPlain -> Current _    ) <= (unPlain -> Pending _ _  ) = False
+  (unPlain -> History _ _ _) <= (unPlain -> Pending _ _  ) = False
+  (unPlain -> History _ _ _) <= (unPlain -> Current _    ) = False
+  (unPlain -> Pending _ _  ) <= _                          = True
+  (unPlain -> Current _    ) <= _                          = True
+
+fromZipper :: Queue -> Set QueueItem
+fromZipper Queue{..} = Set.fromList . toList $ mconcat [ Seq.mapWithIndex Pending pending
+                                                       , maybe Seq.empty (Seq.singleton . Current) current
+                                                       , Seq.mapWithIndex (\i (a, e) -> History i a e) history
+                                                       ]
+
+toZipper :: Set QueueItem -> Queue
+toZipper = Set.foldr' insert def
+  where
+    insert (Pending _ a) q@(Queue{..}) = q { pending = pending |> a }
+    insert (Current a) q = q { current = Just a }
+    insert (History _ a e) q@(Queue{..}) = q { history = history |> (a, e) }
 
 -- | A queue manager periodically modifies a 'Queue', e.g. for cleanup of old jobs
-type QueueManager t = ComposeT (StateT Queue) t STM DiffTime
+type QueueManager t = QueueManagerM t (Extended Micro)
+type QueueManagerM t = ComposeT (StateT Queue) t STM
 
 runQM :: ( HasQueue q
          , MFunctor t
@@ -84,7 +131,36 @@ runQM :: ( HasQueue q
          , Monad (t STM)
          ) => QueueManager t -> q -> t IO ()
 -- ^ Periodically modify a 'Queue'
-runQM qm (extractQueue -> q) = forever $ liftIO . threadDelay . toMicro =<< qm'
+runQM qm (extractQueue -> q) = sleep =<< qm'
   where
     qm' = hoist atomically $ (\(a, s) -> lift (writeTVar q s) >> return a) =<< runStateT (getComposeT qm) =<< lift (readTVar q)
-    toMicro = (`div` 10^6) . fromEnum
+    sleep (abs -> delay)
+      | (Finite d) <- delay = liftIO (threadDelay $ fromEnum d) >> runQM qm q
+      | otherwise = return ()
+
+intersection :: (Foldable f, MonadState Queue (QueueManagerM t)) => f (QueueManager t) -> QueueManager t
+-- ^ Combine two 'QueueManager's keeping only 'QueueEntry's both managers decide to keep
+intersection = foldr' (qmCombine Set.intersection) idQM
+
+idQM :: Monad (QueueManagerM t) => QueueManager t
+-- ^ Identity of 'intersect'
+idQM = return PosInf
+
+union :: (Foldable f, MonadState Queue (QueueManagerM t)) => f (QueueManager t) -> QueueManager t
+-- ^ Combine two 'QueueManager's keeping all 'QueueEntry's either of the managers decides to keep
+union = foldr' (qmCombine Set.union) nullQM
+
+nullQM :: MonadState Queue (QueueManagerM t) => QueueManager t
+-- ^ Identity of 'union'
+nullQM = put def >> return PosInf
+
+qmCombine :: MonadState Queue (QueueManagerM t)
+             => (Set PlainQueueItem -> Set PlainQueueItem -> Set PlainQueueItem)
+             -> (QueueManager t -> QueueManager t -> QueueManager t)
+qmCombine setCombine a b = do
+  (d1, s1) <- local a
+  (d2, s2) <- local b
+  put . toZipper . Set.map unPlain $ on setCombine (Set.map Plain . fromZipper) s1 s2
+  return $ min d1 d2
+  where
+    local x = ((,) <$> get <*> ((,) <$> x <*> get)) >>= (\(oldS, r) -> r <$ put oldS)
diff --git a/server/thermoprint-server.cabal b/server/thermoprint-server.cabal
index 2eda3b5..124fd11 100644
--- a/server/thermoprint-server.cabal
+++ b/server/thermoprint-server.cabal
@@ -48,6 +48,7 @@ library
                      , wai >=3.0.4 && <4
                      , warp >=3.1.9 && <4
                      , mmorph >=1.0.5 && <2
+                     , extended-reals >=0.2.1 && <1
   hs-source-dirs:      src
   default-language:    Haskell2010
 
diff --git a/server/thermoprint-server.nix b/server/thermoprint-server.nix
index 1cbccd2..41f7198 100644
--- a/server/thermoprint-server.nix
+++ b/server/thermoprint-server.nix
@@ -1,6 +1,6 @@
 { mkDerivation, base, conduit, containers, data-default-class
-, deepseq, dyre, either, exceptions, mmorph, monad-control
-, monad-logger, mtl, persistent, persistent-sqlite
+, deepseq, dyre, either, exceptions, extended-reals, mmorph
+, monad-control, monad-logger, mtl, persistent, persistent-sqlite
 , persistent-template, resourcet, servant-server, stdenv, stm, text
 , thermoprint-spec, time, transformers, wai, warp
 }:
@@ -12,8 +12,8 @@ mkDerivation {
   isExecutable = true;
   libraryHaskellDepends = [
     base conduit containers data-default-class deepseq dyre either
-    exceptions mmorph monad-control monad-logger mtl persistent
-    persistent-template resourcet servant-server stm text
+    exceptions extended-reals mmorph monad-control monad-logger mtl
+    persistent persistent-template resourcet servant-server stm text
     thermoprint-spec time transformers wai warp
   ];
   executableHaskellDepends = [
diff --git a/shell.nix b/shell.nix
index e047ae2..747ab7b 100644
--- a/shell.nix
+++ b/shell.nix
@@ -6,8 +6,7 @@ let
   packages = ps: with ps; [
     cabal-install hlint cabal2nix
     hspec QuickCheck quickcheck-instances
-    aeson-pretty
-  ] ++ (builtins.attrValues (import ./default.nix {}));
+  ] ++ (builtins.attrValues (import ./default.nix { inherit pkgs; }));
   ghc = pkgs.haskell.packages.${compilerName}.ghcWithPackages packages;
 in
 pkgs.stdenv.mkDerivation rec {