планировщик типа cron

например так: если задач не 999999999

import Control.Concurrent.Async
import System.Process

addTask :: Int -> String -> IO ()
addTask d t = async $ threadDelay d >> runCommand t >> addTask d t

main = do
  addTask 1000 "echo \"foo\""
  ...

На питоне я сделал так (можете сделать по-своему): Задачи складываются в очередь «ожидания» с приоритетом, где приоритет есть unix timestamp когда оно должно запуститься. Более срочные задачи в очереди стоят первыми.

это умеет делать RTS причем весьма эффективно до тех пор пока задач не слишком много, но к тому времени когда ты в это упрёшься ты успеешь написать более аккуратное решение.

другой вариант ghc зависимый:

GHC.Event.registerTimeout :: EventManager -> Int -> TimeoutCallback -> IO TimeoutKey

Внутри честная priority-queue.

А если надо ограничиить кол-во одновременно выполняемых задач?

ммм.. например так:

main = go <$> newTChanIO
  where 
    go ch = do
       _ <- replicateM n $ forkIO worker
       addTask 1000 "fooo"    
     where
       addTask delay t = async $ do
          threadDelay delay
          atomically $ writeTChan ch (delay,t)
       worker = do
          (delay, t) <- atomicallty (readTChan ch)
          run t
          addTask delay t
       run = undefined

в принципе твой вариант на STM тоже хорошо ложится:

newtype TimedTask a = TT { unTT :: (UTCTime, a)}

instance Ord a where
  compare = compare `on` (fst . unTT)

-- подойдет любая очередь приоритетов
class Heap where
   deleteMin :: Heap a -> Heap a
   findMin :: Ord a => Heap a -> Maybe a
   insert  :: Ord a => a -> Heap a -> Heap a

worker first storage mdelay = do
  ctime <- getCurrentTime
  maybe (newTVarIO False) (registerDelay) mdelay
  ret <- atomically $ 
      (do 
         (tm,task) <- takeTMVar first
         if tm < ctime 
           then do
               Foldable.forM_ (putTMVar first) =<< findMin <$> readTVar storage  
               modifyTVar storage deleteMin 
               return $ Right task
            else return $ Left (tm `diffTime` ctime))
      `orElse` (readTVar delay >>= flip unless retry)        
   case ret of
     Right task -> do 
       run task 
       addAgain task
       worker first loop Nothing
     Left d -> worker fist storage (Just {- тут тип нужно привести -} d)

и создавай сколько нужно воркеров, добавление задачи будет так:

addTask first storage d t = do 
   c <- getCurrectTime
   let c' = c `addTime` d --на память не помню но в UTCTime все методы нужные есть
   atomically $ do
      f@(t, _) <- readTMVar first
      if t < c' 
        then modifyTVar storage (insert (d,t))
        else do
           modifyTVar storage (insert f)
           swapTMVar first (d,t)

я почти сплю так что с десяток очепяток тут есть, но смысл такой.

http://ocamlunix.forge.ocamlcore.org/processes.html

Будет проще, чем в хаскеле, потому что энергичность + императивность по желанию.

Нет, проще точно не будет.

как бы выглядел ... ML-подобных ЯП?

Не могу удержаться, чтобы не сказать «нечитаемо». Примеры в треде подтвердили это.

Приведи пример читаемого, пожалуйста.

Я вот подумал что вариант «async $ threadDelay d >> runCommand t >> addTask d t» мне нравится больше всего. Наверно, можно тупо добавить сюда семафор для ограничения конкурентности, не?

не-не-не. в этом варианте смысл в том, что ты загружаешь легкий поток на задачу, и тебе необходимо запустить по потоку на задачу, так что это останется.

Если ты хочешь ограничить кол-во конкурентно выполняющихся задач, то тебе нужна точка синхронизации, самое простое это использовать каналы (STM или IO), я сразу использовал STM.

в тред призываются решения задачи на лиспе:

а). любой способ без ограничения кол-ва воркеров

б). любой способ, но с ограничением кол-ва задач

в). очередь с аналогом priority queue и возможностью добавлять новые задачи

не-не-не. в этом варианте смысл в том, что ты загружаешь легкий поток на задачу, и тебе необходимо запустить по потоку на задачу, так что это останется.

Я понимаю :). Я просто хотел сказать что в фазе ожидания пусть они все работают параллельно. А вот когда пришло время что-то запускать пусть пытаются получить семафор. Или это не ложится на хаскель?

зачем семафор, когда есть каналы?

Но вообще можно:

http://hackage.haskell.org/packages/archive/SafeSemaphore/0.9.0/doc/html/Cont...

import Control.Concurrent.SSem
addTask sem d t = async $ do
  threadDelay d
  withSem sem (runProcess t)
  async $ addTask sem d t

А с каналами какой код будет? Я не понимаю...

планировщик типа cron (комментарий) - это с каналами.

В общем-то с семафором тоже просто получилось.

Всё, теперь понял, спасибо. Я голосую за вариант с семафором. Не знаю на сколько это «функционально», но выглядит лаконично и понятно.

не знаю, философией кода не сильно увлекаюсь. А решение с семафором и в правду оказалось самым простым и адекватным, но если потребуется расширение операций, например, процесс не должен задерживаться более, чем на T+dt, то нужно будет переключиться на STM семафоры.

процесс не должен задерживаться более, чем на T+dt

А как мы ему помешаем в этом? Прибьём его?

не в ту сторону. Сейчас выполняется максимум n процессов, и следующий готовый будет ждать пока не выполниться хотя бы один из них, при этом ждать он может сколь угодно долго. Вполне возможная проблема, то что мы не можем позволять ему ждать больше, чем какой-то лимит времени, в этом случае если лимит почти превышен, то мы забиваем на ограничение и запускаем его.