diff --git a/Control/Distributed/CCTools/WorkQueue.hs b/Control/Distributed/CCTools/WorkQueue.hs new file mode 100644 --- /dev/null +++ b/Control/Distributed/CCTools/WorkQueue.hs @@ -0,0 +1,16 @@ +-- -*- indent-tabs-mode: nil -*- -- +module Control.Distributed.CCTools.WorkQueue + ( + + module Control.Distributed.CCTools.WorkQueue.WorkQueue + , module Control.Distributed.CCTools.WorkQueue.Task + , module Control.Distributed.CCTools.WorkQueue.Types + , module Control.Distributed.CCTools.WorkQueue.Debug + + ) + where + +import Control.Distributed.CCTools.WorkQueue.WorkQueue +import Control.Distributed.CCTools.WorkQueue.Task +import Control.Distributed.CCTools.WorkQueue.Types +import Control.Distributed.CCTools.WorkQueue.Debug diff --git a/Control/Distributed/CCTools/WorkQueue/Debug.hs b/Control/Distributed/CCTools/WorkQueue/Debug.hs new file mode 100644 --- /dev/null +++ b/Control/Distributed/CCTools/WorkQueue/Debug.hs @@ -0,0 +1,11 @@ +-- -*- indent-tabs-mode: nil -*- -- + +module Control.Distributed.CCTools.WorkQueue.Debug + ( + DebugFlag(..) + , setDebugFlags + ) + where + +import Control.Distributed.CCTools.WorkQueue.Internal.Debug +import Control.Distributed.CCTools.WorkQueue.Internal.Types diff --git a/Control/Distributed/CCTools/WorkQueue/Internal/Cast.hs b/Control/Distributed/CCTools/WorkQueue/Internal/Cast.hs new file mode 100644 --- /dev/null +++ b/Control/Distributed/CCTools/WorkQueue/Internal/Cast.hs @@ -0,0 +1,8 @@ +-- -*- indent-tabs-mode: nil -*- -- + +{-# LANGUAGE MultiParamTypeClasses #-} + +module Control.Distributed.CCTools.WorkQueue.Internal.Cast where + +class Castable a b where + cast :: a -> b diff --git a/Control/Distributed/CCTools/WorkQueue/Internal/CastInstances.hs b/Control/Distributed/CCTools/WorkQueue/Internal/CastInstances.hs new file mode 100644 --- /dev/null +++ b/Control/Distributed/CCTools/WorkQueue/Internal/CastInstances.hs @@ -0,0 +1,146 @@ +-- -*- indent-tabs-mode: nil -*- -- + +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE MultiParamTypeClasses #-} + +module Control.Distributed.CCTools.WorkQueue.Internal.CastInstances where + +import Bindings.CCTools.WorkQueue + +import Control.Distributed.CCTools.WorkQueue.Internal.Cast +import Control.Distributed.CCTools.WorkQueue.Internal.Types + +import Foreign.C.Types + +import Control.Lens ((^.)) + +-- * Haskell-provided + +-- ** 'Int' <-> 'CInt' +instance Castable Int CInt where cast = fromIntegral +instance Castable CInt Int where cast = fromIntegral + +-- ** 'Double' <-> 'CDouble' +instance Castable Double CDouble where cast = fromRational . toRational +instance Castable CDouble Double where cast = fromRational . toRational + + + +-- * For WorkQueue + +-- ** Utilities + +-- *** 'Castable' a b => 'Castable' ('NPlus' a) b +instance Castable a b => Castable (NPlus a) b where + cast (NP v) = cast v + +instance ( Castable a b + , Num a, Num b + , Ord a + , Show a) + => Castable a (NPlus b) + where + cast v | v > 0 = NP $ cast v + | otherwise = error $ "Got a non-positive value when constructing an NPlus: " ++ show v + +-- *** DataSize + +instance Castable (DataSize a) Integer where + cast = _unDS + +-- *** Time + +instance Castable (EpochTime a) Integer where cast = _unET +instance Castable (DiffTime a) Integer where cast = _unDT + + +-- *** File Locality + +instance Castable (Location a) FilePath where cast = _unL + + +-- ** 'MasterMode' + +instance Castable MasterMode CInt where + cast Standalone = c'WORK_QUEUE_MASTER_MODE_STANDALONE + cast Catalog = c'WORK_QUEUE_MASTER_MODE_CATALOG + +instance Castable CInt MasterMode where + cast v | v == c'WORK_QUEUE_MASTER_MODE_STANDALONE = Standalone + | v == c'WORK_QUEUE_MASTER_MODE_CATALOG = Catalog + | otherwise = error $ "Cannot cast to MasterMode: " ++ show v + + +-- ** 'Hunger' + +instance Castable CInt Hunger where + cast v | v <= 0 = Full + | otherwise = Hungry $ fromIntegral v + + +-- ** 'FastAbort' + +instance Castable FastAbort CDouble where + cast FastAbortOff = 0 + cast (FastAbort v) = cast v + +-- ** 'Timeout' + +instance Castable Timeout CInt where + cast Forever = -1 + cast (Seconds v) = fromIntegral v + +-- ** 'Cached' + +instance Castable Cached CInt where + cast c = if c + then c'WORK_QUEUE_CACHE + else c'WORK_QUEUE_NOCACHE + + +-- ** 'WorkerScheduleAlg' + +instance Castable WorkerScheduleAlg CInt where + cast FCFS = c'WORK_QUEUE_SCHEDULE_FCFS + cast FILES = c'WORK_QUEUE_SCHEDULE_FILES + cast TIME = c'WORK_QUEUE_SCHEDULE_TIME + cast RAND = c'WORK_QUEUE_SCHEDULE_RAND + +instance Castable CInt WorkerScheduleAlg where + cast v | v == c'WORK_QUEUE_SCHEDULE_FCFS = FCFS + | v == c'WORK_QUEUE_SCHEDULE_FILES = FILES + | v == c'WORK_QUEUE_SCHEDULE_TIME = TIME + | v == c'WORK_QUEUE_SCHEDULE_RAND = RAND + | otherwise = error $ "Cannot cast CInt to WorkerScheduleAlg: " ++ show v + +-- ** 'TaskOrdering' + +instance Castable TaskOrdering CInt where + cast LIFO = c'WORK_QUEUE_TASK_ORDER_LIFO + cast FIFO = c'WORK_QUEUE_TASK_ORDER_FIFO + +-- ** 'FileType' + +instance Castable FileType CInt where + cast InputFile = c'WORK_QUEUE_INPUT + cast OutputFile = c'WORK_QUEUE_OUTPUT + +-- ** TaskID + +instance Castable CInt TaskID where + cast = TID . fromIntegral + +instance Integral b => Castable TaskID b where + cast (TID v) = fromIntegral v + + +-- ** Port + +instance Castable (Maybe Port) CInt where + cast Nothing = c'WORK_QUEUE_RANDOM_PORT + cast (Just p) = cast $ p^.unP + +-- ** Command + +instance Castable Command String where + cast = _unC diff --git a/Control/Distributed/CCTools/WorkQueue/Internal/Debug.hs b/Control/Distributed/CCTools/WorkQueue/Internal/Debug.hs new file mode 100644 --- /dev/null +++ b/Control/Distributed/CCTools/WorkQueue/Internal/Debug.hs @@ -0,0 +1,18 @@ +-- -*- indent-tabs-mode: nil -*- -- + +module Control.Distributed.CCTools.WorkQueue.Internal.Debug where + +import Control.Distributed.CCTools.WorkQueue.Internal.Types (DebugFlag (..)) + +import Bindings.CCTools.Debug + +import Foreign.C.String (CString, newCString) + + +setDebugFlags :: [DebugFlag] -> IO () +setDebugFlags = mapM_ ( (c'cctools_debug_flags_set =<<) . cDebugFlag ) + where + cDebugFlag :: DebugFlag -> IO CString + cDebugFlag All = newCString "all" + cDebugFlag WorkQueue = newCString "wq" + diff --git a/Control/Distributed/CCTools/WorkQueue/Internal/Task.hs b/Control/Distributed/CCTools/WorkQueue/Internal/Task.hs new file mode 100644 --- /dev/null +++ b/Control/Distributed/CCTools/WorkQueue/Internal/Task.hs @@ -0,0 +1,175 @@ +-- -*- indent-tabs-mode: nil -*- -- + +-- | Creation and operations on 'Task's + +module Control.Distributed.CCTools.WorkQueue.Internal.Task where + +import Control.Distributed.CCTools.WorkQueue.Internal.Cast +import Control.Distributed.CCTools.WorkQueue.Internal.Types +import Control.Distributed.CCTools.WorkQueue.Internal.CastInstances () + +import Bindings.CCTools.WorkQueue + +import Foreign.C +import Foreign.Storable + +import Control.Applicative ((<$>)) +import Control.Lens +import System.IO.Unsafe +import System.Exit (ExitCode (..) ) + +import Data.ByteString (ByteString) +import qualified Data.ByteString as BS + + + +-- * Task creation/deletion +task :: Command -> IO Task +task c = t' + where + t' = T <$> t + t = c'work_queue_task_create =<< newCString (view unC c) + +delete :: Task -> IO () +delete t = c'work_queue_task_delete t' + where + t' = view unT t + +-- * File specifications + +specifyFile :: Task -> Location Local -> Location Remote -> FileType -> Cached -> IO () +specifyFile t l r k c = do + l' <- newCString $ l^.unL + r' <- newCString $ r^.unL + c'work_queue_task_specify_file (t^.unT) + l' + r' + (cast k) + (cast c) + + +specifyBuffer :: Task -> ByteString -> Location Remote -> Cached -> IO () +specifyBuffer t s r c = + BS.useAsCStringLen s $ \(d,l) -> do + r' <- newCString $ r^.unL + c'work_queue_task_specify_buffer (t^.unT) + d + (cast l) + r' + (cast c) + + +specifyFileCommand :: Task -> Location Remote -> Command -> FileType -> Cached -> IO () +specifyFileCommand t l e k c = do + path <- newCString $ l^.unL + cmd' <- newCString $ e^.unC + c'work_queue_task_specify_file_command (t^.unT) + path + cmd' + (cast k) + (cast c) + +specifyTag :: Task -> String -> IO () +specifyTag t s = c'work_queue_task_specify_tag (t^.unT) =<< newCString s + +specifyAlgorithm :: Task -> WorkerScheduleAlg -> IO () +specifyAlgorithm t a = c'work_queue_task_specify_algorithm (t^.unT) (cast a) + + +-- * Task properties + +withTask :: (a -> IO b) -> (C'work_queue_task -> a) -> Task -> IO b +withTask g f t = g =<< f <$> peek (t^.unT) + +unsafeWithTask :: (a -> IO b) -> (C'work_queue_task -> a) -> Task -> b +unsafeWithTask g f = unsafeDupablePerformIO . withTask g f + +tag :: Task -> String +tag = unsafeWithTask peekCString c'work_queue_task'tag + +command :: Task -> Command +command = unsafeWithTask (fmap C . peekCString) c'work_queue_task'command_line + +workerSelectionAlgorithm :: Task -> WorkerScheduleAlg +workerSelectionAlgorithm = unsafeWithTask (return . cast) c'work_queue_task'worker_selection_algorithm + +output :: Task -> ByteString +output = unsafeWithTask BS.packCString c'work_queue_task'output + +taskID :: Task -> TaskID +taskID = unsafeWithTask (return . TID . fromIntegral) c'work_queue_task'taskid + +returnStatus :: Task -> ExitCode +returnStatus = unsafeWithTask mkExitCode c'work_queue_task'return_status + where mkExitCode = return . exitcode . fromIntegral + exitcode c | c == 0 = ExitSuccess + | otherwise = ExitFailure c + +result :: Task -> Maybe () +result = unsafeWithTask mkMaybe c'work_queue_task'result + where mkMaybe = return . mkMaybe' . fromIntegral + mkMaybe' s | s == 0 = Just () + | otherwise = Nothing + +host :: Task -> Hostname +host = unsafeWithTask (fmap H . peekCString) c'work_queue_task'hostname + +taskTimeType :: Integral a => (a -> t) -> (C'work_queue_task -> a) -> Task -> t +taskTimeType c = unsafeWithTask (return . c) + +taskEpochTime :: Integral a => (C'work_queue_task -> a) -> Task -> EpochTime MicroSeconds +taskEpochTime f = taskTimeType ET (fromIntegral . f) + +taskTime :: Integral a => (C'work_queue_task -> a) -> Task -> DiffTime MicroSeconds +taskTime f = taskTimeType DT (fromIntegral . f) + +submitTime :: Task -> EpochTime MicroSeconds +submitTime = taskEpochTime c'work_queue_task'time_task_submit + +finishTime :: Task -> EpochTime MicroSeconds +finishTime = taskEpochTime c'work_queue_task'time_task_finish + +appDelayTime :: Task -> EpochTime MicroSeconds +appDelayTime = taskEpochTime c'work_queue_task'time_app_delay + +timeSendInputStart :: Task -> EpochTime MicroSeconds +timeSendInputStart = taskEpochTime c'work_queue_task'time_send_input_start + +timeSendInputFinish :: Task -> EpochTime MicroSeconds +timeSendInputFinish = taskEpochTime c'work_queue_task'time_send_input_finish + +sendTime :: Task -> DiffTime MicroSeconds +sendTime t = t1 `diffTime` t0 + where t0 = timeSendInputStart t + t1 = timeSendInputFinish t + +timeExecuteCmdStart :: Task -> EpochTime MicroSeconds +timeExecuteCmdStart = taskEpochTime c'work_queue_task'time_execute_cmd_start + +timeExecuteCmdFinish :: Task -> EpochTime MicroSeconds +timeExecuteCmdFinish = taskEpochTime c'work_queue_task'time_execute_cmd_finish + +timeExecuteCmd :: Task -> DiffTime MicroSeconds +timeExecuteCmd t = t1 `diffTime` t0 + where t0 = timeExecuteCmdStart t + t1 = timeExecuteCmdFinish t + +timeReceiveOutputStart :: Task -> EpochTime MicroSeconds +timeReceiveOutputStart = taskEpochTime c'work_queue_task'time_receive_output_start + +timeReceiveOutputFinish :: Task -> EpochTime MicroSeconds +timeReceiveOutputFinish = taskEpochTime c'work_queue_task'time_receive_output_finish + +timeReceiveOutput :: Task -> DiffTime MicroSeconds +timeReceiveOutput t = t1 `diffTime` t0 + where t0 = timeReceiveOutputStart t + t1 = timeReceiveOutputFinish t + +totalBytesTransferred :: Task -> DataSize Bytes +totalBytesTransferred = unsafeWithTask (return . DS . fromIntegral) c'work_queue_task'total_bytes_transferred + +totalTransferTime :: Task -> DiffTime MicroSeconds +totalTransferTime = taskTime c'work_queue_task'total_transfer_time + +totalCmdExecutionTime :: Task -> DiffTime MicroSeconds +totalCmdExecutionTime = taskTime c'work_queue_task'cmd_execution_time diff --git a/Control/Distributed/CCTools/WorkQueue/Internal/Types.hs b/Control/Distributed/CCTools/WorkQueue/Internal/Types.hs new file mode 100644 --- /dev/null +++ b/Control/Distributed/CCTools/WorkQueue/Internal/Types.hs @@ -0,0 +1,180 @@ +-- -*- indent-tabs-mode: nil -*- -- + +{-# LANGUAGE EmptyDataDecls #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE StandaloneDeriving #-} +{-# LANGUAGE TemplateHaskell #-} + +module Control.Distributed.CCTools.WorkQueue.Internal.Types where + +import Bindings.CCTools.WorkQueue + +import Foreign.Ptr +import Foreign.ForeignPtr + +import Control.Lens + +-- * Usefull types + +newtype NPlus a = NP { _unZP :: a } + deriving + (Enum, Num, Real, Eq, Ord, Show) + +deriving instance Integral a => Integral (NPlus a) + +nplus :: (Ord a, Num a, Show a) => a -> NPlus a +nplus v | v > 0 = NP v + | otherwise = error $ "Cannot create an NPlus with: " ++ show v + + + +-- ** Unit-typed data volume + +data Bytes +newtype DataSize a = DS { _unDS :: Integer } +makeLenses ''DataSize + +bytes :: Integral i => i -> DataSize Bytes +bytes = DS . fromIntegral + + + +-- ** Unit-typed time + +data Seconds +data MicroSeconds + +-- | Time since January 1, 1970 +newtype EpochTime a = ET { _unET :: Integer } + +-- | Difference between two 'EpochTime's +newtype DiffTime a = DT { _unDT :: Integer } + +makeLenses ''EpochTime +makeLenses ''DiffTime + +diffTime :: EpochTime a -> EpochTime a -> DiffTime a +diffTime t1 t0 = DT $ t1' - t0' + where t0' = view unET t0 + t1' = view unET t1 + +epoch :: Integral i => i -> EpochTime a +epoch = ET . fromIntegral + +epochSeconds :: Integral i => i -> EpochTime Seconds +epochSeconds = epoch + +epochMicroSeconds :: Integral i => i -> EpochTime MicroSeconds +epochMicroSeconds = epoch + + + + +-- ** Locality + +data Local +data Remote +newtype Location a = L { _unL :: FilePath } + +makeLenses ''Location + +remote :: FilePath -> Location Remote +remote = L + +local :: FilePath -> Location Local +local = L + + + +-- * WorkQueue-related + +-- ** Interface to WorkQueue + +newtype WorkQueue = WQ { _unWQ :: ForeignPtr C'work_queue } +newtype Task = T { _unT :: Ptr C'work_queue_task } +newtype Stats = S { _unS :: ForeignPtr C'work_queue_stats } + +makeLenses ''WorkQueue +makeLenses ''Task +makeLenses ''Stats + +data MasterMode = Standalone | Catalog + deriving + Show + +data Hunger = Full | Hungry (NPlus Int) + deriving + (Eq + , Show) + +data FastAbort = FastAbortOff | FastAbort (NPlus Double) + deriving + Show + +data Timeout = Forever | Seconds (NPlus Int) + deriving + Show + +type Cached = Bool + +data WorkerScheduleAlg = FCFS | FILES | TIME | RAND + deriving + (Read, Show) + +data TaskOrdering = LIFO | FIFO + deriving + (Read, Show) + +data FileType = InputFile | OutputFile + deriving + (Read, Show) + +-- ** Semantically-typed + +newtype Command = C { _unC :: String } +newtype TaskID = TID { _unTID :: Int } +newtype Hostname = H { _unH :: String } +newtype Port = P { _unP :: NPlus Int } + deriving + Show + +data QueueParams = QP { + _qport :: Maybe Port + , _name :: Maybe String + , _fastabort :: Maybe FastAbort + , _taskordering :: Maybe TaskOrdering + , _priority :: Maybe Int + , _mode :: Maybe MasterMode + , _logfile :: Maybe FilePath + , _scheduler :: Maybe WorkerScheduleAlg + } deriving Show + +makeLenses ''Command +makeLenses ''TaskID +makeLenses ''Port +makeLenses ''Hostname +makeLenses ''QueueParams + + +-- *** Smart Constructors + +-- | A 'Port' is a postive 'Int' +port :: Int -> Port +port = P . fromIntegral + +-- | Creates a 'Timeout' using 'Seconds' +seconds :: Integral i => i -> Timeout +seconds = Seconds . nplus . fromIntegral + +-- | A 'Command' to be executed remotely +cmd :: String -> Command +cmd = C + +-- | The 'Hostname' of a node +hostname :: String -> Hostname +hostname = H + + +-- * Enabling WorkQueue debugging + +data DebugFlag = All | WorkQueue diff --git a/Control/Distributed/CCTools/WorkQueue/Internal/WorkQueue.hs b/Control/Distributed/CCTools/WorkQueue/Internal/WorkQueue.hs new file mode 100644 --- /dev/null +++ b/Control/Distributed/CCTools/WorkQueue/Internal/WorkQueue.hs @@ -0,0 +1,235 @@ +-- -*- indent-tabs-mode: nil -*- -- + +-- | Creation and operations on a 'WorkQueue' + +module Control.Distributed.CCTools.WorkQueue.Internal.WorkQueue where + +import Control.Distributed.CCTools.WorkQueue.Internal.Cast +import Control.Distributed.CCTools.WorkQueue.Internal.Types +import Control.Distributed.CCTools.WorkQueue.Internal.CastInstances () + +import Bindings.CCTools.WorkQueue + +import Foreign.C +import Foreign.Ptr +import Foreign.ForeignPtr +import Foreign.Storable +import Foreign.Marshal.Utils (toBool) + +import Control.Applicative ((<$>)) +import Control.Lens +import Control.Monad ((>=>), void) +import System.IO.Unsafe +import Control.Monad.Loops + +withJust :: Maybe a -> (a -> IO ()) -> IO () +withJust m f = case m of + Just a -> f a + _ -> return () + +-- * WorkQueue creation/deletion + +-- | The default parameters for initializing a 'WorkQueue' +-- +-- > defaultQParams :: QueueParams +-- > defaultQParams = QP { +-- > _qport = Just . port $ fromIntegral c'WORK_QUEUE_DEFAULT_PORT +-- > , _name = Nothing +-- > , _fastabort = Nothing +-- > , _taskordering = Nothing +-- > , _priority = Nothing +-- > , _mode = Nothing +-- > , _logfile = Nothing +-- > , _scheduler = Nothing +-- > } + +defaultQParams :: QueueParams +defaultQParams = QP { + _qport = Just . port $ fromIntegral c'WORK_QUEUE_DEFAULT_PORT + , _name = Nothing + , _fastabort = Nothing + , _taskordering = Nothing + , _priority = Nothing + , _mode = Nothing + , _logfile = Nothing + , _scheduler = Nothing + } + +workqueue :: QueueParams -> IO WorkQueue +workqueue p = do + ptr <- c'work_queue_create (cast $ p^.qport) + q <- WQ <$> newForeignPtr p'work_queue_delete ptr + + withJust (p^.fastabort ) (setFastAbort q) + withJust (p^.taskordering) (setTaskOrdering q) + withJust (p^.priority ) (setPriority q) + withJust (p^.mode ) (setMasterMode q) + withJust (p^.name ) (setName q) + withJust (p^.logfile ) (setLog q) + withJust (p^.scheduler ) (setScheduleAlg q) + + return q + +setName :: WorkQueue -> String -> IO () +setName q n = withForeignPtr (q^.unWQ) $ \q' -> do + n' <- newCString n + c'work_queue_specify_name q' n' + +setFastAbort :: WorkQueue -> FastAbort -> IO () +setFastAbort q a = withForeignPtr (q^.unWQ) (void . flip c'work_queue_activate_fast_abort (cast a)) + +setTaskOrdering :: WorkQueue -> TaskOrdering -> IO () +setTaskOrdering q o = withForeignPtr (q^.unWQ) (`c'work_queue_specify_task_order` cast o) + +setPriority :: WorkQueue -> Int -> IO () +setPriority q p = withForeignPtr (q^.unWQ) (`c'work_queue_specify_priority` cast p) + +setMasterMode :: WorkQueue -> MasterMode -> IO () +setMasterMode q m = withForeignPtr (q^.unWQ) (`c'work_queue_specify_master_mode` cast m) + +setLog :: WorkQueue -> FilePath -> IO () +setLog q p = withForeignPtr (q^.unWQ) $ \q' -> + c'work_queue_specify_log q' =<< newCString p + +setScheduleAlg :: WorkQueue -> WorkerScheduleAlg -> IO () +setScheduleAlg q a = withForeignPtr (q^.unWQ) (`c'work_queue_specify_algorithm` cast a) + +shutdownWorkers :: WorkQueue -> NPlus Int -> IO (NPlus Int) +shutdownWorkers q n = withForeignPtr (q^.unWQ) + (fmap cast . flip c'work_queue_shut_down_workers (cast n)) + + +-- * Task submission/retrieval + +submit :: WorkQueue -> Task -> IO TaskID +submit q t = withForeignPtr (q^.unWQ) + (fmap cast . flip c'work_queue_submit (t^.unT)) + +wait :: WorkQueue -> Timeout -> IO (Maybe Task) +wait q t = withForeignPtr (q^.unWQ) $ \q' -> do + r <- c'work_queue_wait q' $ cast t + return $ if r == nullPtr + then Nothing + else Just $ T r + + +cancelTaskBy :: (Ptr C'work_queue -> a -> IO (Ptr C'work_queue_task)) + -> (b -> IO a) + -> WorkQueue -> b -> IO (Maybe Task) +cancelTaskBy c'cancel mkAttr q attribute = + withForeignPtr (q^.unWQ) $ \q' -> do + a <- mkAttr attribute + r <- c'cancel q' a + return $ if r == nullPtr + then Nothing + else Just $ T r + +cancelTaskID :: WorkQueue -> TaskID -> IO (Maybe Task) +cancelTaskID = cancelTaskBy c'work_queue_cancel_by_taskid (return . cast) + +cancelTaskTag :: WorkQueue -> String -> IO (Maybe Task) +cancelTaskTag = cancelTaskBy c'work_queue_cancel_by_tasktag newCString + +-- | Main pattern for processing tasks after submission +-- +-- Basically: +-- +-- > eventLoop q timeout update process = +-- > while (q is not empty) +-- > update q +-- > r <- wait q timeout +-- > when r is Just: process q r +eventLoop :: WorkQueue + -> Timeout + -> (WorkQueue -> IO ()) -- ^ "update": called before 'wait' + -> (WorkQueue -> Task -> IO ()) -- ^ "process" a returned 'Task' + -> IO () +eventLoop q timeout update process = + whileM_ (not <$> isEmpty q) $ do + update q + r <- wait q timeout + maybe (return ()) (process q) r + +-- * WorkQueue properties + +getHunger :: WorkQueue -> IO Hunger +getHunger q = cast <$> withForeignPtr (q^.unWQ) c'work_queue_hungry + +isFull :: WorkQueue -> IO Bool +isFull = fmap (Full ==) . getHunger + +isEmpty :: WorkQueue -> IO Bool +isEmpty q = toBool <$> withForeignPtr (q^.unWQ) c'work_queue_empty + +getPort :: WorkQueue -> IO Port +getPort q = P . cast <$> withForeignPtr (q^.unWQ) c'work_queue_port + +getName :: WorkQueue -> IO String +getName q = peekCString =<< withForeignPtr (q^.unWQ) c'work_queue_name + + +-- * WorkQueue Runtime Statistics + +getWorkerSummary :: WorkQueue -> IO String +getWorkerSummary q = peekCString =<< withForeignPtr (q^.unWQ) c'work_queue_get_worker_summary + +getStats :: WorkQueue -> IO Stats +getStats q = withForeignPtr (q^.unWQ) $ \q' -> do + fptr <- mallocForeignPtr + withForeignPtr fptr $ c'work_queue_get_stats q' + return $ S fptr + +withStats :: (a -> b) -> (C'work_queue_stats -> a) -> Stats -> IO b +withStats g f s = withForeignPtr (s^.unS) (peek >=> (return . g . f)) + +unsafeWithStats :: (a -> b) -> (C'work_queue_stats -> a) -> Stats -> b +unsafeWithStats g f s = unsafeDupablePerformIO $ withStats g f s + + +workersInit :: Stats -> Int +workersInit = unsafeWithStats fromIntegral c'work_queue_stats'workers_init + +workersReady :: Stats -> Int +workersReady = unsafeWithStats fromIntegral c'work_queue_stats'workers_ready + +workersBusy :: Stats -> Int +workersBusy = unsafeWithStats fromIntegral c'work_queue_stats'workers_busy + +workersCancelling :: Stats -> Int +workersCancelling = unsafeWithStats fromIntegral c'work_queue_stats'workers_cancelling + +tasksRunning :: Stats -> Int +tasksRunning = unsafeWithStats fromIntegral c'work_queue_stats'tasks_running + +tasksWaiting :: Stats -> Int +tasksWaiting = unsafeWithStats fromIntegral c'work_queue_stats'tasks_waiting + +tasksComplete :: Stats -> Int +tasksComplete = unsafeWithStats fromIntegral c'work_queue_stats'tasks_complete + +totalTasksDispatched :: Stats -> Int +totalTasksDispatched = unsafeWithStats fromIntegral c'work_queue_stats'total_tasks_dispatched + +totalTasksComplete :: Stats -> Int +totalTasksComplete = unsafeWithStats fromIntegral c'work_queue_stats'total_tasks_complete + +totalWorkersJoined :: Stats -> Int +totalWorkersJoined = unsafeWithStats fromIntegral c'work_queue_stats'total_workers_joined + +totalWorkersRemoved :: Stats -> Int +totalWorkersRemoved = unsafeWithStats fromIntegral c'work_queue_stats'total_workers_removed + +totalBytesSent :: Stats -> Integer +totalBytesSent = unsafeWithStats fromIntegral c'work_queue_stats'total_bytes_sent + +totalBytesReceived :: Stats -> Integer +totalBytesReceived = unsafeWithStats fromIntegral c'work_queue_stats'total_bytes_received + +startTime :: Stats -> EpochTime MicroSeconds +startTime = unsafeWithStats (ET . fromIntegral) c'work_queue_stats'start_time + +totalSendTime :: Stats -> DiffTime MicroSeconds +totalSendTime = unsafeWithStats (DT . fromIntegral) c'work_queue_stats'total_send_time + +totalReceiveTime :: Stats -> DiffTime MicroSeconds +totalReceiveTime = unsafeWithStats (DT . fromIntegral) c'work_queue_stats'total_receive_time diff --git a/Control/Distributed/CCTools/WorkQueue/Task.hs b/Control/Distributed/CCTools/WorkQueue/Task.hs new file mode 100644 --- /dev/null +++ b/Control/Distributed/CCTools/WorkQueue/Task.hs @@ -0,0 +1,20 @@ +-- -*- indent-tabs-mode: nil -*- -- + +module Control.Distributed.CCTools.WorkQueue.Task + ( + + -- * Task specification + task, specifyFile, specifyBuffer, specifyFileCommand, specifyTag, specifyAlgorithm, delete + + -- * Properties + , tag, command, workerSelectionAlgorithm, output, taskID, returnStatus, result, host + , submitTime, finishTime, appDelayTime, timeSendInputStart, timeSendInputFinish + , sendTime, timeExecuteCmdStart, timeExecuteCmdFinish, timeExecuteCmd + , timeReceiveOutputStart, timeReceiveOutputFinish, timeReceiveOutput + , totalBytesTransferred, totalTransferTime, totalCmdExecutionTime + + ) + where + + +import Control.Distributed.CCTools.WorkQueue.Internal.Task diff --git a/Control/Distributed/CCTools/WorkQueue/Types.hs b/Control/Distributed/CCTools/WorkQueue/Types.hs new file mode 100644 --- /dev/null +++ b/Control/Distributed/CCTools/WorkQueue/Types.hs @@ -0,0 +1,67 @@ +-- -*- indent-tabs-mode: nil -*- -- + +module Control.Distributed.CCTools.WorkQueue.Types + ( + + -- * Utility Types + + -- ** Positive numbers + NPlus, nplus + + -- ** Unit-typed values + + -- | These values are instances of 'Marshallable'. You can use + -- the 'marshal' function to extract the actual value + + -- *** Data + , Bytes, DataSize + , bytes + + -- *** Time + , Seconds, MicroSeconds, EpochTime, DiffTime + , epochSeconds, epochMicroSeconds + , diffTime + + + -- *** File Locality + , Local, Remote, Location + , local, remote + + + -- * WorkQueue API + + -- ** WorkQueue, Task, and runtime stats + , WorkQueue, Task, Stats + + -- ** WorkQueue Parameters + , QueueParams(..) + + -- ** WorkQueue properties + , MasterMode(..) + , Hunger(..) + , FastAbort(..) + , Timeout(..), seconds + , Cached + , WorkerScheduleAlg(..) + , TaskOrdering(..) + , FileType(..) + , Port, port + + -- ** Task properties + , Command, cmd + , TaskID + , Hostname, hostname + + -- ** Debugging + , DebugFlag(..) + + -- ** Conversions and marshalling + , Castable(..) + + ) + where + + +import Control.Distributed.CCTools.WorkQueue.Internal.Types +import Control.Distributed.CCTools.WorkQueue.Internal.Cast +import Control.Distributed.CCTools.WorkQueue.Internal.CastInstances () diff --git a/Control/Distributed/CCTools/WorkQueue/WorkQueue.hs b/Control/Distributed/CCTools/WorkQueue/WorkQueue.hs new file mode 100644 --- /dev/null +++ b/Control/Distributed/CCTools/WorkQueue/WorkQueue.hs @@ -0,0 +1,42 @@ +-- -*- indent-tabs-mode: nil -*- -- + +module Control.Distributed.CCTools.WorkQueue.WorkQueue + ( + + -- * WorkQueue specification + defaultQParams + , workqueue, setName, setFastAbort, setTaskOrdering, setPriority, setMasterMode, setScheduleAlg, shutdownWorkers + + -- * Logging + , setLog + + -- * Task Submission and Retrieval + , submit, eventLoop + , wait, cancelTaskID, cancelTaskTag + + -- * Properties + , getHunger, isFull, isEmpty, getPort, getName + + -- * Runtime Statistics + , getWorkerSummary + , getStats + + -- ** Workers + , workersInit, workersReady, workersBusy, workersCancelling + , totalWorkersJoined, totalWorkersRemoved + + -- ** Tasks + , tasksRunning, tasksWaiting, tasksComplete + , totalTasksDispatched, totalTasksComplete + + -- ** Data + , totalBytesSent, totalBytesReceived + + -- ** Times + , startTime, totalSendTime, totalReceiveTime + + ) + where + + +import Control.Distributed.CCTools.WorkQueue.Internal.WorkQueue diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,339 @@ + GNU GENERAL PUBLIC LICENSE + Version 2, June 1991 + + Copyright (C) 1989, 1991 Free Software Foundation, Inc., + 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + Everyone is permitted to copy and distribute verbatim copies + of this license document, but changing it is not allowed. + + Preamble + + The licenses for most software are designed to take away your +freedom to share and change it. By contrast, the GNU General Public +License is intended to guarantee your freedom to share and change free +software--to make sure the software is free for all its users. This +General Public License applies to most of the Free Software +Foundation's software and to any other program whose authors commit to +using it. (Some other Free Software Foundation software is covered by +the GNU Lesser General Public License instead.) You can apply it to +your programs, too. + + When we speak of free software, we are referring to freedom, not +price. Our General Public Licenses are designed to make sure that you +have the freedom to distribute copies of free software (and charge for +this service if you wish), that you receive source code or can get it +if you want it, that you can change the software or use pieces of it +in new free programs; and that you know you can do these things. + + To protect your rights, we need to make restrictions that forbid +anyone to deny you these rights or to ask you to surrender the rights. +These restrictions translate to certain responsibilities for you if you +distribute copies of the software, or if you modify it. + + For example, if you distribute copies of such a program, whether +gratis or for a fee, you must give the recipients all the rights that +you have. You must make sure that they, too, receive or can get the +source code. And you must show them these terms so they know their +rights. + + We protect your rights with two steps: (1) copyright the software, and +(2) offer you this license which gives you legal permission to copy, +distribute and/or modify the software. + + Also, for each author's protection and ours, we want to make certain +that everyone understands that there is no warranty for this free +software. If the software is modified by someone else and passed on, we +want its recipients to know that what they have is not the original, so +that any problems introduced by others will not reflect on the original +authors' reputations. + + Finally, any free program is threatened constantly by software +patents. We wish to avoid the danger that redistributors of a free +program will individually obtain patent licenses, in effect making the +program proprietary. To prevent this, we have made it clear that any +patent must be licensed for everyone's free use or not licensed at all. + + The precise terms and conditions for copying, distribution and +modification follow. + + GNU GENERAL PUBLIC LICENSE + TERMS AND CONDITIONS FOR COPYING, DISTRIBUTION AND MODIFICATION + + 0. This License applies to any program or other work which contains +a notice placed by the copyright holder saying it may be distributed +under the terms of this General Public License. The "Program", below, +refers to any such program or work, and a "work based on the Program" +means either the Program or any derivative work under copyright law: +that is to say, a work containing the Program or a portion of it, +either verbatim or with modifications and/or translated into another +language. (Hereinafter, translation is included without limitation in +the term "modification".) Each licensee is addressed as "you". + +Activities other than copying, distribution and modification are not +covered by this License; they are outside its scope. The act of +running the Program is not restricted, and the output from the Program +is covered only if its contents constitute a work based on the +Program (independent of having been made by running the Program). +Whether that is true depends on what the Program does. + + 1. You may copy and distribute verbatim copies of the Program's +source code as you receive it, in any medium, provided that you +conspicuously and appropriately publish on each copy an appropriate +copyright notice and disclaimer of warranty; keep intact all the +notices that refer to this License and to the absence of any warranty; +and give any other recipients of the Program a copy of this License +along with the Program. + +You may charge a fee for the physical act of transferring a copy, and +you may at your option offer warranty protection in exchange for a fee. + + 2. You may modify your copy or copies of the Program or any portion +of it, thus forming a work based on the Program, and copy and +distribute such modifications or work under the terms of Section 1 +above, provided that you also meet all of these conditions: + + a) You must cause the modified files to carry prominent notices + stating that you changed the files and the date of any change. + + b) You must cause any work that you distribute or publish, that in + whole or in part contains or is derived from the Program or any + part thereof, to be licensed as a whole at no charge to all third + parties under the terms of this License. + + c) If the modified program normally reads commands interactively + when run, you must cause it, when started running for such + interactive use in the most ordinary way, to print or display an + announcement including an appropriate copyright notice and a + notice that there is no warranty (or else, saying that you provide + a warranty) and that users may redistribute the program under + these conditions, and telling the user how to view a copy of this + License. (Exception: if the Program itself is interactive but + does not normally print such an announcement, your work based on + the Program is not required to print an announcement.) + +These requirements apply to the modified work as a whole. If +identifiable sections of that work are not derived from the Program, +and can be reasonably considered independent and separate works in +themselves, then this License, and its terms, do not apply to those +sections when you distribute them as separate works. But when you +distribute the same sections as part of a whole which is a work based +on the Program, the distribution of the whole must be on the terms of +this License, whose permissions for other licensees extend to the +entire whole, and thus to each and every part regardless of who wrote it. + +Thus, it is not the intent of this section to claim rights or contest +your rights to work written entirely by you; rather, the intent is to +exercise the right to control the distribution of derivative or +collective works based on the Program. + +In addition, mere aggregation of another work not based on the Program +with the Program (or with a work based on the Program) on a volume of +a storage or distribution medium does not bring the other work under +the scope of this License. + + 3. You may copy and distribute the Program (or a work based on it, +under Section 2) in object code or executable form under the terms of +Sections 1 and 2 above provided that you also do one of the following: + + a) Accompany it with the complete corresponding machine-readable + source code, which must be distributed under the terms of Sections + 1 and 2 above on a medium customarily used for software interchange; or, + + b) Accompany it with a written offer, valid for at least three + years, to give any third party, for a charge no more than your + cost of physically performing source distribution, a complete + machine-readable copy of the corresponding source code, to be + distributed under the terms of Sections 1 and 2 above on a medium + customarily used for software interchange; or, + + c) Accompany it with the information you received as to the offer + to distribute corresponding source code. (This alternative is + allowed only for noncommercial distribution and only if you + received the program in object code or executable form with such + an offer, in accord with Subsection b above.) + +The source code for a work means the preferred form of the work for +making modifications to it. For an executable work, complete source +code means all the source code for all modules it contains, plus any +associated interface definition files, plus the scripts used to +control compilation and installation of the executable. However, as a +special exception, the source code distributed need not include +anything that is normally distributed (in either source or binary +form) with the major components (compiler, kernel, and so on) of the +operating system on which the executable runs, unless that component +itself accompanies the executable. + +If distribution of executable or object code is made by offering +access to copy from a designated place, then offering equivalent +access to copy the source code from the same place counts as +distribution of the source code, even though third parties are not +compelled to copy the source along with the object code. + + 4. You may not copy, modify, sublicense, or distribute the Program +except as expressly provided under this License. Any attempt +otherwise to copy, modify, sublicense or distribute the Program is +void, and will automatically terminate your rights under this License. +However, parties who have received copies, or rights, from you under +this License will not have their licenses terminated so long as such +parties remain in full compliance. + + 5. You are not required to accept this License, since you have not +signed it. However, nothing else grants you permission to modify or +distribute the Program or its derivative works. These actions are +prohibited by law if you do not accept this License. Therefore, by +modifying or distributing the Program (or any work based on the +Program), you indicate your acceptance of this License to do so, and +all its terms and conditions for copying, distributing or modifying +the Program or works based on it. + + 6. Each time you redistribute the Program (or any work based on the +Program), the recipient automatically receives a license from the +original licensor to copy, distribute or modify the Program subject to +these terms and conditions. You may not impose any further +restrictions on the recipients' exercise of the rights granted herein. +You are not responsible for enforcing compliance by third parties to +this License. + + 7. If, as a consequence of a court judgment or allegation of patent +infringement or for any other reason (not limited to patent issues), +conditions are imposed on you (whether by court order, agreement or +otherwise) that contradict the conditions of this License, they do not +excuse you from the conditions of this License. If you cannot +distribute so as to satisfy simultaneously your obligations under this +License and any other pertinent obligations, then as a consequence you +may not distribute the Program at all. For example, if a patent +license would not permit royalty-free redistribution of the Program by +all those who receive copies directly or indirectly through you, then +the only way you could satisfy both it and this License would be to +refrain entirely from distribution of the Program. + +If any portion of this section is held invalid or unenforceable under +any particular circumstance, the balance of the section is intended to +apply and the section as a whole is intended to apply in other +circumstances. + +It is not the purpose of this section to induce you to infringe any +patents or other property right claims or to contest validity of any +such claims; this section has the sole purpose of protecting the +integrity of the free software distribution system, which is +implemented by public license practices. Many people have made +generous contributions to the wide range of software distributed +through that system in reliance on consistent application of that +system; it is up to the author/donor to decide if he or she is willing +to distribute software through any other system and a licensee cannot +impose that choice. + +This section is intended to make thoroughly clear what is believed to +be a consequence of the rest of this License. + + 8. If the distribution and/or use of the Program is restricted in +certain countries either by patents or by copyrighted interfaces, the +original copyright holder who places the Program under this License +may add an explicit geographical distribution limitation excluding +those countries, so that distribution is permitted only in or among +countries not thus excluded. In such case, this License incorporates +the limitation as if written in the body of this License. + + 9. The Free Software Foundation may publish revised and/or new versions +of the General Public License from time to time. Such new versions will +be similar in spirit to the present version, but may differ in detail to +address new problems or concerns. + +Each version is given a distinguishing version number. If the Program +specifies a version number of this License which applies to it and "any +later version", you have the option of following the terms and conditions +either of that version or of any later version published by the Free +Software Foundation. If the Program does not specify a version number of +this License, you may choose any version ever published by the Free Software +Foundation. + + 10. If you wish to incorporate parts of the Program into other free +programs whose distribution conditions are different, write to the author +to ask for permission. For software which is copyrighted by the Free +Software Foundation, write to the Free Software Foundation; we sometimes +make exceptions for this. Our decision will be guided by the two goals +of preserving the free status of all derivatives of our free software and +of promoting the sharing and reuse of software generally. + + NO WARRANTY + + 11. BECAUSE THE PROGRAM IS LICENSED FREE OF CHARGE, THERE IS NO WARRANTY +FOR THE PROGRAM, TO THE EXTENT PERMITTED BY APPLICABLE LAW. EXCEPT WHEN +OTHERWISE STATED IN WRITING THE COPYRIGHT HOLDERS AND/OR OTHER PARTIES +PROVIDE THE PROGRAM "AS IS" WITHOUT WARRANTY OF ANY KIND, EITHER EXPRESSED +OR IMPLIED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF +MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE. THE ENTIRE RISK AS +TO THE QUALITY AND PERFORMANCE OF THE PROGRAM IS WITH YOU. SHOULD THE +PROGRAM PROVE DEFECTIVE, YOU ASSUME THE COST OF ALL NECESSARY SERVICING, +REPAIR OR CORRECTION. + + 12. IN NO EVENT UNLESS REQUIRED BY APPLICABLE LAW OR AGREED TO IN WRITING +WILL ANY COPYRIGHT HOLDER, OR ANY OTHER PARTY WHO MAY MODIFY AND/OR +REDISTRIBUTE THE PROGRAM AS PERMITTED ABOVE, BE LIABLE TO YOU FOR DAMAGES, +INCLUDING ANY GENERAL, SPECIAL, INCIDENTAL OR CONSEQUENTIAL DAMAGES ARISING +OUT OF THE USE OR INABILITY TO USE THE PROGRAM (INCLUDING BUT NOT LIMITED +TO LOSS OF DATA OR DATA BEING RENDERED INACCURATE OR LOSSES SUSTAINED BY +YOU OR THIRD PARTIES OR A FAILURE OF THE PROGRAM TO OPERATE WITH ANY OTHER +PROGRAMS), EVEN IF SUCH HOLDER OR OTHER PARTY HAS BEEN ADVISED OF THE +POSSIBILITY OF SUCH DAMAGES. + + END OF TERMS AND CONDITIONS + + How to Apply These Terms to Your New Programs + + If you develop a new program, and you want it to be of the greatest +possible use to the public, the best way to achieve this is to make it +free software which everyone can redistribute and change under these terms. + + To do so, attach the following notices to the program. It is safest +to attach them to the start of each source file to most effectively +convey the exclusion of warranty; and each file should have at least +the "copyright" line and a pointer to where the full notice is found. + + + Copyright (C) + + This program is free software; you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation; either version 2 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License along + with this program; if not, write to the Free Software Foundation, Inc., + 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +Also add information on how to contact you by electronic and paper mail. + +If the program is interactive, make it output a short notice like this +when it starts in an interactive mode: + + Gnomovision version 69, Copyright (C) year name of author + Gnomovision comes with ABSOLUTELY NO WARRANTY; for details type `show w'. + This is free software, and you are welcome to redistribute it + under certain conditions; type `show c' for details. + +The hypothetical commands `show w' and `show c' should show the appropriate +parts of the General Public License. Of course, the commands you use may +be called something other than `show w' and `show c'; they could even be +mouse-clicks or menu items--whatever suits your program. + +You should also get your employer (if you work as a programmer) or your +school, if any, to sign a "copyright disclaimer" for the program, if +necessary. Here is a sample; alter the names: + + Yoyodyne, Inc., hereby disclaims all copyright interest in the program + `Gnomovision' (which makes passes at compilers) written by James Hacker. + + , 1 April 1989 + Ty Coon, President of Vice + +This General Public License does not permit incorporating your program into +proprietary programs. If your program is a subroutine library, you may +consider it more useful to permit linking proprietary applications with the +library. If this is what you want to do, use the GNU Lesser General +Public License instead of this License. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/cctools-workqueue.cabal b/cctools-workqueue.cabal new file mode 100644 --- /dev/null +++ b/cctools-workqueue.cabal @@ -0,0 +1,145 @@ +-- -*- indent-tabs-mode: nil -*- -- +-- Initial hs-cctools-workqueue.cabal generated by cabal init. For further +-- documentation, see http://haskell.org/cabal/users-guide/ + +name: cctools-workqueue +version: 3.6.1.0.1.0.0 + +synopsis: High-level interface to CCTools' WorkQueue library + +description: This package provides a high-level Haskell interface to the WorkQueue library: + + . + . + Work Queue is a framework for building large + master-worker applications that span many + computers including clusters, clouds, and + grids. Work Queue applications are written in C, + Perl, or Python using a simple API that allows + users to define tasks, submit them to the queue, + and wait for completion. Tasks are executed by a + standard worker process that can run on any + available machine. Each worker calls home to the + master process, arranges for data transfer, and + executes the tasks. Together, the master and + worker handle a wide variety of failures, + allowing for dynamically scalable and robust + applications. + . + <> + . + . + Example usage: + . + By default the master listens on 9123 on the localhost host. + Start a worker: + . + > $ work_queue_worker -d all localhost 9123 + . + Now we can execute the following master program: + . + > module Main where + > + > import Control.Distributed.CCTools.WorkQueue + > + > import Control.Applicative ((<$>)) + > import qualified Data.ByteString.Char8 as BS (pack, putStrLn) + > import Foreign.C.String (newCStringLen) + > import Control.Monad (forM_) + > + > + > + > mktask :: Show a => a -> IO Task + > mktask v = do + > let script = BS.pack . unlines $ [ + > "t=$(echo $RANDOM % 10 | bc)" + > , "sleep $t" + > , "echo " ++ show v + > ] + > t <- task $ cmd "bash script.sh" + > specifyBuffer t script (remote "script.sh") False + > specifyTag t $ show v + > return t + > + > printStats :: WorkQueue -> IO () + > printStats q = do + > s <- getStats q + > print $ map ($ s) [tasksRunning, tasksWaiting, tasksComplete] + > + > processResult :: WorkQueue -> Task -> IO () + > processResult q r = do + > putStrLn $ "Got: " ++ show (tag r) + > BS.putStrLn . output $ r + > delete r + > + > main = do + > setDebugFlags [All] + > q <- workqueue defaultQParams + > ts <- mapM mktask [1..10] + > forM_ ts (submit q) + > + > eventLoop q (seconds 1) printStats processResult + > + > putStrLn "Done!" + + + + +license: GPL-2 +license-file: LICENSE +author: Badi' Abdul-Wahid +maintainer: abdulwahidc@gmail.com +-- copyright: +category: Distributed Computing +build-type: Simple +cabal-version: >=1.8 + +homepage: http://bitbucket.org/badi/hs-cctools-workqueue + +source-repository head + type: git + location: git@bitbucket.org:badi/hs-cctools-workqueue.git + + + +flag warn + description: Turn on warnings + default: False + +library + exposed-modules: Control.Distributed.CCTools.WorkQueue.Task + , Control.Distributed.CCTools.WorkQueue.Types + , Control.Distributed.CCTools.WorkQueue.WorkQueue + , Control.Distributed.CCTools.WorkQueue.Debug + , Control.Distributed.CCTools.WorkQueue + + other-modules: Control.Distributed.CCTools.WorkQueue.Internal.Cast + , Control.Distributed.CCTools.WorkQueue.Internal.Types + , Control.Distributed.CCTools.WorkQueue.Internal.CastInstances + , Control.Distributed.CCTools.WorkQueue.Internal.Task + , Control.Distributed.CCTools.WorkQueue.Internal.WorkQueue + , Control.Distributed.CCTools.WorkQueue.Internal.Debug + + + if impl(ghc == 7.4.*) + build-depends: base ==4.5.* + + if impl(ghc == 7.6.*) + build-depends: base ==4.6.* + + build-depends: bindings-cctools == 3.6.1.0.1.0.0 + , lens + , unix + , bytestring + , monad-loops + + extra-libraries: dttools + + ghc-options: -O3 -funbox-strict-fields + + if flag(warn) + ghc-options: -W + -fwarn-duplicate-exports -fwarn-incomplete-patterns -fwarn-incomplete-uni-patterns + -fwarn-incomplete-record-updates -fwarn-missing-fields -fwarn-missing-signatures + -fwarn-name-shadowing -fwarn-tabs -fwarn-unused-imports -fwarn-unused-matches + -fwarn-warnings-deprecations