periodic (empty) → 0.1.0.0
raw patch · 6 files changed
+434/−0 lines, 6 filesdep +basedep +cerealdep +hedissetup-changed
Dependencies added: base, cereal, hedis, hspec, periodic, text, time
Files
- LICENSE +13/−0
- Setup.hs +2/−0
- example/Main.hs +13/−0
- periodic.cabal +53/−0
- src/System/Periodic.hs +236/−0
- test/Spec.hs +117/−0
+ LICENSE view
@@ -0,0 +1,13 @@+Copyright (c) 2016 Position Development LLC++Permission to use, copy, modify, and/or distribute this software for any purpose+with or without fee is hereby granted, provided that the above copyright notice+and this permission notice appear in all copies.++THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH+REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND+FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT,+INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS+OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER+TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF+THIS SOFTWARE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ example/Main.hs view
@@ -0,0 +1,13 @@+{-# LANGUAGE OverloadedStrings #-}+import Control.Concurrent (forkIO, threadDelay)+import Control.Monad (forever)+import qualified Data.Text.IO as T+import qualified Database.Redis as R+import System.Periodic+main = do rconn <- R.connect R.defaultConnectInfo+ scheduler <- create (Name "default") rconn (CheckInterval (Seconds 1)) (LockTimeout (Seconds 1000)) T.putStrLn+ addTask scheduler "print-hello-job" (Every (Seconds 10)) (T.putStrLn "hello")+ addTask scheduler "print-bye-job" (Every (Seconds 1)) (T.putStrLn "bye")+ forkIO (run scheduler)+ forkIO (run scheduler)+ forever (threadDelay 1000000)
+ periodic.cabal view
@@ -0,0 +1,53 @@+name: periodic+version: 0.1.0.0+synopsis: A reliable at-least-once periodic job scheduler backed by redis.+description: Please see README.md+homepage: https://github.com/positiondev/periodic+license: ISC+license-file: LICENSE+author: Daniel Patterson+maintainer: workers@positiondev.com+copyright: 2016 Position Development LLC+category: Web+build-type: Simple+-- extra-source-files:+cabal-version: >=1.10++library+ hs-source-dirs: src+ exposed-modules: System.Periodic+ build-depends: base >= 4.7 && < 5+ , text+ , time+ , cereal+ , hedis+ default-language: Haskell2010++executable example+ hs-source-dirs: example+ main-is: Main.hs+ build-depends: base >= 4.7 && < 5+ , text+ , time+ , cereal+ , hedis+ , periodic+ default-language: Haskell2010+++test-suite periodic-test+ type: exitcode-stdio-1.0+ hs-source-dirs: test, src+ main-is: Spec.hs+ build-depends: base+ , text+ , time+ , hedis+ , hspec+ , cereal+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ default-language: Haskell2010++source-repository head+ type: git+ location: https://github.com/positiondev/periodic
+ src/System/Periodic.hs view
@@ -0,0 +1,236 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE PartialTypeSignatures #-}+{-# LANGUAGE ScopedTypeVariables #-}++{-|++This module contains a very basic periodic job processor backed by+Redis, suitable to be run by clusters of machines (jobs will be run at+least once per scheduled time, and will not run more than once unless+the (configurable) timeout is reached, at which point retry+occurs. There is also no guarantee that distribution of jobs across+machines will be fair). It depends upon Redis not losing data once it+has acknowledged it, and guaranteeing the atomicity that is specified+for commands like EVAL (ie, that if you do several things within an+EVAL, they will all happen or none will happen). Nothing has been+tested with Redis clusters (and it likely will not work).++An example use is the following (provided in repository), noting that+we create two threads to process jobs, and could run this entire+executable many times and the jobs would only run at the scheduled+time.++> {-# LANGUAGE OverloadedStrings #-}+> import Control.Monad (forever)+> import System.Periodic+> import Control.Concurrent (threadDelay, forkIO)+> import qualified Data.Text.IO as T+> import qualified Database.Redis as R+> main = do rconn <- R.connect R.defaultConnectInfo+> scheduler <- create "default" rconn (CheckInterval (Seconds 1)) (LockTimeout (Seconds 1000)) (T.putStrLn)+> addTask scheduler "print-hello-job" (Every (Seconds 100)) (T.putStrLn "hello")+> addTask scheduler "print-bye-job" (Every (Seconds 10)) (T.putStrLn "bye")+> forkIO (run scheduler)+> forkIO (run scheduler)+> forever (threadDelay 1000000)+++-}++module System.Periodic+ ( -- * Types+ Time(..)+ , Seconds(..)+ , Period(..)+ , Name(..)+ , CheckInterval(..)+ , LockTimeout(..)+ , Logger+ , Scheduler+ -- * Managing Schedulers+ , create+ , run+ , destroy+ -- * Scheduling Tasks+ , addTask+ ) where++import Control.Concurrent (forkIO, threadDelay)+import Control.Concurrent.MVar+import Control.Exception (SomeException, catch)+import Control.Monad (forever, join, when)+import Data.Monoid+import Data.Ratio+import Data.Serialize+import Data.Text (Text)+import qualified Data.Text as T+import qualified Data.Text.Encoding as T+import Data.Time.Calendar+import Data.Time.Clock+import Data.Time.Clock.POSIX+import qualified Database.Redis as R+++-- | Time from midnight UTC+newtype Time = Time DiffTime++newtype Seconds = Seconds Int++-- | When the job should run - either at a particular offset from+-- midnight UTC, or every N seconds.+data Period = Daily Time | Every Seconds++-- | The name of the scheduler. This, when combined with the task+-- name, should be unique on a given Redis server.+newtype Name = Name Text++-- | How frequently we should check if jobs need to get run. If all+-- your jobs are infrequent (daily, or occurring hours apart), setting+-- this to a high number (every minute, or more) decreases the number+-- of queries to Redis.+newtype CheckInterval = CheckInterval Seconds++-- | How long a job that has been started and not marked as finished+-- should take before we run it again. Note that if you put this+-- number above the period, we will never retry the jobs (but they+-- will still keep running every period).+newtype LockTimeout = LockTimeout Seconds++-- | A function where log messages are sent.+type Logger = Text -> IO ()++-- | Internal type representing the current tasks that are scheduled.+data Scheduler = Scheduler { schedulerName :: Name+ , schedulerRedisConn :: R.Connection+ , schedulerTasks :: MVar [(Text, Period, IO ())]+ , schedulerCheckInterval :: CheckInterval+ , schedulerLockTimout :: LockTimeout+ , schedulerLogger :: Logger+ }++-- | This function creates a new scheduler, which can then have tasks+-- scheduled in it. The tasks themselves are _not_ stored in Redis; it+-- is only used to ensure we run the tasks at the right time (and not+-- more than once across multiple machines). Note that this does not+-- actually run any of the tasks - you must fork threads that call+-- 'run'.+create :: Name+ -> R.Connection+ -> CheckInterval+ -> LockTimeout+ -> (Text -> IO ())+ -> IO Scheduler+create name rconn check lock logger =+ do mv <- newMVar []+ return $ Scheduler name rconn mv check lock logger++-- | Add a new task to a given scheduler, to be run at the specified+-- period. Note that the name, when combined with the scheduler name,+-- should be unique for the given redis database, or else only one of+-- the jobs with the same name will be run.+addTask :: Scheduler -> Text -> Period -> IO () -> IO ()+addTask scheduler name when act =+ do modifyMVar_ (schedulerTasks scheduler) (return . (:) (name, when, act))+ return ()++lockTimeout = 10 * 3600++-- | Start a scheduler thread. You must start at least one, but can+-- start multiple threads if you have many tasks that take a long time+-- to run.+run :: Scheduler -> IO ()+run (Scheduler (Name nm) rconn mv (CheckInterval (Seconds check)) lock logger) = forever $+ do now <- getCurrentTime+ tasks <- readMVar mv+ mapM_ (tryRunTask logger lock nm rconn now) tasks+ threadDelay (check * 1000000)++lastStartedKey pname name =+ T.encodeUtf8 $ pname <> "-" <> name <> "-last-started"+lockedKey pname name =+ T.encodeUtf8 $ pname <> "-" <> name <> "-running-at"++-- | This clears out redis of any mention of tasks related to the+-- scheduler. It isn't necessary to do in normal scenarios.+destroy :: Scheduler -> IO ()+destroy (Scheduler (Name nm) rconn mv _ _ _) =+ do tasks <- readMVar mv+ R.runRedis rconn $+ R.del (concatMap (\(tnm,_,_) ->+ [lastStartedKey nm tnm+ ,lockedKey nm tnm])+ tasks)+ return ()++collapseError :: Either a (Maybe b) -> Maybe b+collapseError (Left _) = Nothing+collapseError (Right v) = v++collapseNumberBoolFalse :: Either R.Reply (Maybe Integer) -> Bool+collapseNumberBoolFalse (Left e) = error (show e)+collapseNumberBoolFalse (Right (Just 1)) = True+collapseNumberBoolFalse (Right (Just 0)) = False+++parseUnixTime s = let Right (n, d) = decode s in posixSecondsToUTCTime $ fromRational $ n % d++renderUnixTime t = let r = toRational . utcTimeToPOSIXSeconds $ t in encode (numerator r, denominator r)++minutes5 = 60 * 5++shouldRun :: LockTimeout -> Period -> Maybe UTCTime -> Maybe UTCTime -> UTCTime -> Bool+shouldRun (LockTimeout (Seconds timeout)) _ _ (Just locked) now+ = diffUTCTime now locked > fromIntegral timeout+shouldRun _ (Daily (Time t)) Nothing _ now+ | now > (now { utctDayTime = t }) && utctDayTime now - t < minutes5+ = True+shouldRun _ (Daily (Time t)) (Just last) _ now+ | now > (now { utctDayTime = t }) && last < (now { utctDayTime = t })+ = True+shouldRun _ (Every (Seconds n)) Nothing _ now+ = True+shouldRun _ (Every (Seconds n)) (Just last) locked now+ = diffUTCTime now last > fromIntegral n+shouldRun lock period last locked now = False+++tryRunTask :: Logger -> LockTimeout -> Text -> R.Connection -> UTCTime -> (Text, Period, IO ()) -> IO ()+tryRunTask logger timeout pname rconn now (name, period, task) =+ do lastStarted <- fmap parseUnixTime . collapseError <$>+ R.runRedis rconn+ (R.get (lastStartedKey pname name))+ lockedAt <- fmap parseUnixTime . collapseError <$>+ R.runRedis rconn+ (R.get (lockedKey pname name))+ when (shouldRun timeout period lastStarted lockedAt now) $+ do gotLock <-+ collapseNumberBoolFalse <$> R.runRedis rconn+ (R.eval "local lock = redis.call('get', KEYS[1])\n\+ \local last = redis.call('get', KEYS[2])\n\+ \if ((lock == false and ARGV[1] == '0') or (lock == ARGV[1]))\+ \ and ((last == false and ARGV[2] == '0') or (last == ARGV[2])) then\n\+ \ redis.call('set', KEYS[1], ARGV[3])\n\+ \ redis.call('set', KEYS[2], ARGV[3])\n\+ \ return 1\n\+ \else\n\+ \ return 0\n\+ \end" [lockedKey pname name, lastStartedKey pname name]+ [maybe "0" renderUnixTime lockedAt+ ,maybe "0" renderUnixTime lastStarted+ ,renderUnixTime now])+ when gotLock $+ do x <- newEmptyMVar+ jt <- forkIO (catch (task >> putMVar x Nothing)+ (\(e::SomeException) ->+ putMVar x (Just e)))+ res <- takeMVar x+ case res of+ Just e -> logger $ T.concat ["periodic["+ ,pname+ ,"::"+ ,name+ ,"] error: "+ ,T.pack (show e)]+ Nothing -> return ()+ R.runRedis rconn $ R.del [lockedKey pname name]+ return ()
+ test/Spec.hs view
@@ -0,0 +1,117 @@+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+import Control.Concurrent (forkIO, killThread, threadDelay)+import Control.Concurrent.MVar (MVar, modifyMVarMasked_, newMVar,+ readMVar, takeMVar)+import Control.Monad (replicateM)+import qualified Data.Text as T+import Data.Time.Calendar+import Data.Time.Clock+import qualified Database.Redis as R+import System.Periodic++import Test.Hspec+++date :: Integer -> Int -> Int -> UTCTime+date y m d = UTCTime (fromGregorian y m d) 0++time :: UTCTime -> Int -> Int -> Int -> UTCTime+time t h m s = t { utctDayTime = fromIntegral $ h * 60 * 60 + m * 60 + s }++runNThreads :: Int -> Int -> Scheduler -> IO ()+runNThreads n delay sched = do threads <- replicateM n (forkIO $ run sched)+ threadDelay delay+ mapM_ killThread threads++createSch :: T.Text -> IO Scheduler+createSch n = createSch' n print++createSch' :: T.Text -> Logger -> IO Scheduler+createSch' n l = do rconn <- R.connect R.defaultConnectInfo+ create (Name n) rconn (CheckInterval (Seconds 1)) (LockTimeout (Seconds 1000)) l++main :: IO ()+main = hspec $+ do describe "shouldRun" $+ do it "should run if it's never been run, not locked and time just passed" $+ shouldRun (LockTimeout (Seconds 1000)) (Daily (Time 5)) Nothing Nothing (time (date 2016 5 5) 0 0 6)+ it "shouldn't run if it was locked recently" $+ not $ shouldRun (LockTimeout (Seconds 1000)) (Daily (Time 5)) (Just (time (date 2016 5 4) 23 45 0)) (Just (time (date 2016 5 4) 23 45 0)) (time (date 2016 5 5) 0 0 6)+ it "should run if it was locked a long time ago" $+ shouldRun (LockTimeout (Seconds 1000))+ (Daily (Time 5))+ (Just (date 2016 5 4))+ (Just (date 2016 5 4))+ (date 2016 5 5)+ describe "simple job" $+ do it "should run " $+ do mvar <- newMVar 0+ scheduler <- createSch "simple-1"+ addTask scheduler "job" (Every (Seconds 100)) (modifyMVarMasked_ mvar (return . (+1)))++ runNThreads 1 30000 scheduler++ destroy scheduler+ v <- takeMVar mvar+ 1 `shouldBe` v+ it "should only run once per scheduled time" $+ do mvar <- newMVar 0+ scheduler <- createSch "simple-2"+ addTask scheduler "job" (Every (Seconds 100)) (modifyMVarMasked_ mvar (return . (+1)))+ runNThreads 3 100000 scheduler+ destroy scheduler+ v <- takeMVar mvar+ v `shouldBe` 1+ it "should run at each time point" $+ do mvar <- newMVar 0+ scheduler <- createSch "simple-3"+ addTask scheduler "job" (Every (Seconds 2)) (modifyMVarMasked_ mvar (return . (+1)))+ runNThreads 3 4000000 scheduler+ destroy scheduler+ v <- takeMVar mvar+ -- NOTE(dbp 2016-05-26): Precise timing is hard+ -- without making the tests super slow.+ (v == 3 || v == 2) `shouldBe` True+ it "should run at scheduled time" $+ do mvar <- newMVar 0+ scheduler <- createSch "simple-4"+ seconds <- utctDayTime <$> getCurrentTime+ addTask scheduler "job" (Daily (Time seconds)) (modifyMVarMasked_ mvar (return . (+1)))+ runNThreads 1 4000000 scheduler+ destroy scheduler+ v <- takeMVar mvar+ v `shouldBe` 1+ it "should not run at an unscheduled time" $+ do mvar <- newMVar 0+ scheduler <- createSch "simple-4"+ now <- getCurrentTime+ let seconds = utctDayTime $ addUTCTime 3600 now+ addTask scheduler "job" (Daily (Time seconds)) (modifyMVarMasked_ mvar (return . (+1)))+ runNThreads 1 2000000 scheduler+ destroy scheduler+ v <- takeMVar mvar+ v `shouldBe` 0+ describe "error handling" $+ do it "should keep running jobs if one throws an exception" $+ do mvar <- newMVar 0+ scheduler <- createSch' "error-1" (const $ return ())+ addTask scheduler "job-error" (Every (Seconds 100)) (error "blowing up")+ thread <- forkIO $ run scheduler+ threadDelay 300000+ addTask scheduler "job-success" (Every (Seconds 100)) (modifyMVarMasked_ mvar (return . (+1)))+ threadDelay 2000000+ killThread thread+ destroy scheduler+ v <- takeMVar mvar+ 1 `shouldBe` v+ it "should send exception to logger" $+ do mvar <- newMVar []+ scheduler <- createSch' "error-1" (\t -> modifyMVarMasked_ mvar (return . (t:)))+ addTask scheduler "job-error" (Every (Seconds 100)) (error "blowing up")+ thread <- forkIO $ run scheduler+ threadDelay 500000+ killThread thread+ destroy scheduler+ v <- takeMVar mvar+ length v `shouldBe` 1