Hello Neil, Tuesday, April 7, 2009, 6:13:29 PM, you wrote:

Calls to System.Cmd.system, i.e. running external console processes. It's a make system I'm writing, so virtually all the time is spent in calls to ghc etc.

To Bulat: I should have been clearer with the spec. The idea is that multiple calls to paralell_ can execute, and a function executing inside parallel_ can itself call parallel_. For this reason I need one top-level thread pool, which requires unsafePerformIO. If I create a thread pool every time, I end up with more threads than I want.

this is smth new to solve i propose to use concept similar to Capability of GHC RTS: we have one Capability provided by thread calling para and N-1 Capabilities provided by your thread pool. all that we need is to reuse current thread Capability as part of pool! para xs = do sem <- newQSem for xs $ \x -> do writeChan chan (x `finally` signalQSem sem) tid <- forkIO (executing commands from chan...) waitQSem sem killThread tid instead of killThread we really should send pseudo-job (like my Nothing value) that will led to self-killing of job that gets this signal this solution still may lead to a bit more or less than N threads executed at the same time. your turn! -- Best regards, Bulat mailto:Bulat.Ziganshin@gmail.com

Neil Mitchell wrote:

I guess the "nested calls to parallel_" bit is the part of the spec that makes everything much harder!

Thanks

Neil

Yes. Much more annoying. But the problem here is generic. To avoid it you must never allow all thread to block at once. The parallel_ function is such a job, so you solved this with the 'idempotent' trick. You solution works by blocking all but 1 thread. 1a) Some worker thread 1 executes parallel_ with some jobs 1b) These get submitted the work queue 'chan' 1c) worker thread 1 starts on those same jobs, ignoring the queue 1d) worker thread 1 reaches the job being processed by thread 2 1e) worker thread 1 blocks until the jobs is finished in modifyMVar 2a) Worker thread 2 grabs a job posted by thread 1, that calls parallel_ 2b) This batch of jobs gets submitted to the work queue 'chan' 2c) worker thread 2 starts on those same jobs, ignoring the queue 1d) worker thread 2 reaches the job being processed by thread 3 1e) worker thread 2 blocks until the jobs is finished in modifyMVar 3...4...5... And now only 1 thread is still working, and it has to work in series. I think I can fix this...

Hi Sebastian:

How about using unsafeInterleaveIO to get a lazy suspension of the result of each action, and then using par to spark off each of them? If that works you can reuse the existing task-parallel system of GHC to do the heavily lifting for you, instead of having to write your own.

par is likely to spark all the computations, and then switch between them - which will mean I've got more than N things running in parallel.

i think the only way to solve this problem is to create one more thread each time. let's see: on every call to para you need to alloc one thread to wait for jobs completion. so on each nested call to para you have minus one worker thread. finally you will eat them all!

so you need to make fork: one thread should serve jobs and another one wait for completion of this jobs bucket. and with killItself flag you will finish superfluous thread JIT

You are right, your previous solution was running at N-1 threads if the order was a little unlucky. I've attached a new version which I think gives you N threads always executing at full potential. It's basically your idea from the last post, with the main logic being: parallel_ (x1:xs) = do sem <- newQSem $ 1 - length xs forM_ xs $ \x -> writeChan queue (x >> signalQSem sem, False) x1 addWorker waitQSem sem writeChan queue (signalQSem sem, True) waitQSem sem Where the second flag being True = kill, as you suggested. I think I've got the semaphore logic right - anyone want to see if I missed something? With this new version running 1000000 items takes ~1 second, instead of ~10 seconds before, so an order of magnitude improvement, and greater fairness. Very nice, thanks for all the help! Thanks Neil

Hello Neil, Tuesday, April 7, 2009, 7:33:25 PM, you wrote:

parallel_ (x1:xs) = do sem <- newQSem $ 1 - length xs forM_ xs $ \x -> writeChan queue (x >> signalQSem sem, False) x1 addWorker waitQSem sem writeChan queue (signalQSem sem, True) waitQSem sem

Where the second flag being True = kill, as you suggested. I think I've got the semaphore logic right - anyone want to see if I missed something?

Neil, executing x1 directly in parallel_ is incorrect idea. you should have N worker threads, not N-1 threads plus one job executed in main thread. imagine that you have 1000 jobs and N=4. that you will got here is 3 threads each executed 333 jobs and 1 job executed by main thread so you still need to detach one more worker job and finish it just before we are ready to finish waiting for QSem and continue in main thread which is sole reason why we need killItself flag. in this code snipped this flag is completely useless, btw -- Best regards, Bulat mailto:Bulat.Ziganshin@gmail.com

Hi

par is likely to spark all the computations, and then switch between them - which will mean I've got more than N things running in parallel.

| par/GHC RTS limits amount of Haskell threads running simultaneously. | with a system call marked as safe, Capability will be freed while we | execute external program so nothing will be limited except for amount | of tasks *starting* (as opposite to running) simultaneously :))) Yeah, I misspoke - I want to avoid starting N things.

...

...

parallel_ (x1:xs) = do     sem <- newQSem $ 1 - length xs     forM_ xs $ \x ->         writeChan queue (x >> signalQSem sem, False)     x1     addWorker     waitQSem sem     writeChan queue (signalQSem sem, True)     waitQSem sem

...

Neil, executing x1 directly in parallel_ is incorrect idea.

It's a very slight optimisation, as it saves us queueing and dequeueing x1, since we know the worker we're about the spawn on the line below will grab x1 immediately.

forget this. but it still a bit suboptimal: after everything was finished, we schedule one more empty job and wait while some worker thread will pick up it. it will go into Chan after all jobs scheduled at the time our jobs was executed so that we are doing here is eventually don't do any internal activity while we have all N external programs running

instead, my solution packed this flag together with last job so once last job is finished we are immediately returned from parallel_ so other internal activity may go on

There is no guarantee that the last job finishes last. If the first job takes longer than the last job we'll be one thread short while waiting on the first job. It's a shame, since removing that additional writeChan isn't particularly useful. Thanks Neil

Hi

...

...

...

parallel_ (x1:xs) = do     sem <- newQSem $ 1 - length xs     forM_ xs $ \x ->         writeChan queue (x >> signalQSem sem, False)     x1     addWorker     waitQSem sem     writeChan queue (signalQSem sem, True)     waitQSem sem

...

...

Neil, executing x1 directly in parallel_ is incorrect idea.

...

forget this. but it still a bit suboptimal...

i think i realized why you use this schema. my solution may lead to N-1 worker threads in the system if last job is too small - after its execution we finish one thread and have just N-1 working threads until parallel_ will be finished

but problem i mentioned in previous letter may also take place although it looks like less important. we may solve both problems by allowing worker thread to actively select its death time: it should die only at the moment when *last* job in bucket was finished - this guarantees us exactly N worker threads at any time. so:

parallel_ (x1:xs) = do    sem <- newQSem $ - length xs    jobsLast <- newMVar (length xs)    addWorker    forM_ (x1:xs) $ \x -> do        writeChan queue $ do           x           signalQSem sem           modifyMVar jobsLast $ \jobs -> do               return (jobs-1, jobs==0)

Yes, this saves us adding a kill job to the queue, but requires an extra MVar. I guess which one of these is to be preferred depends on performance measures. I've just found that QSem _ISN'T_ valid below 0, so the above code won't actually work with QSem as it stands. I've reported this on ghc-users@. I've also spotted that QSem creates plenty of MVar's itself, so the logic of moving to QSem instead of MVar isn't really valid (although the new approach is nicer, so that's good). Thanks Neil

Hi Bulat,

...

Yes, this saves us adding a kill job to the queue, but requires an extra MVar. I guess which one of these is to be preferred depends on performance measures.

i think that main problem with your last variant is that kill job added to the queue much later than real jobs. imagine the following scenario:

1. para used to add 100 jobs to the queue 2. while these jobs are executing, para in another thread adds another 100 jobs to the queue

in my variant, first para exits just when its own jobs are completed. in your variant, exit job is added to the queue only after second batch of 100 jobs so we will wait until all these jobs are executed

Ah, you are indeed right. I forgot about the property of timeliness - you want parallel_ to return as soon as it can. Once you're using an MVar to do the killing, you might as well use the same MVar to wait on, and avoid the semaphores, and avoid the bug with negative semaphores: http://hackage.haskell.org/trac/ghc/ticket/3159 I've attached a revised implementation. With my benchmark it gives a stack overflow: n = 1000000 main = do r <- newIORef (0 :: Int) let incRef = atomicModifyIORef r (\a -> let a2 = a + 1 in a2 `seq` (a2,a)) parallel_ $ replicate n $ incRef v <- readIORef r parallelStop print v However, calling with that many items in a parallel queue is highly unlikely, so I'll not worry (if anyone knows why it stack overflows, I am curious). This parallel_ implementation has all the properties I want, so I'm very happy with it. As a side note, it's necessary to add parallelStop, to kill all the threads - or you get thread blocked exceptions being raised. You create a thread pool waiting on a queue as a CAF, but after the last call to parallel_ finishes GHC realises the queue is unreachable. This means that all the threads are blocked indefinitely, so GHC raises exceptions. To solve this we call parallelStop to get all the threads to die nicely. Thanks Neil

Hi Bulat,

...

I've attached a revised implementation. With my benchmark it gives a stack overflow:

let incRef = atomicModifyIORef r (\a -> (a,a))

That was the problem, I now no longer get a stack overflow.

...

As a side note, it's necessary to add parallelStop, to kill all the threads - or you get thread blocked exceptions being raised.

alternatively, you can catch this exception in addWorker

I could, but it wouldn't be that easy to tell if the reason for the deadlock was the program finishing a real deadlock occurring in the system. I'd rather not catch exceptions where possible. Thanks Neil

Hi Gleb,

...

I've just found that QSem _ISN'T_ valid below 0, so the above code won't actually work with QSem as it stands.

I may be missing something, but I guess this code will work if 'newQSem (-n)' is replaced with 'newQSemN n', every 'signalQSem sem' with 'signalQSemN sem 1' and 'waitQSem sem' with 'waitQSemN sem n'.

Yep, that seems valid. However, the new solution requires an MVar to store the number of things we're waiting for anyway, so we might as well use that to also wake up the main thread. Semaphores aren't more efficient than MVar's, they are in fact implemented in terms of MVar's, so there is no benefit. Thanks Neil

Hi Bulat,

btw, if all that you need is to limit amount of simultaneous System.Cmd.system calls, you may go from opposite side: wrap this call into semaphore:

sem = unsafePerformIO$ newQSem numCapabilities

mysystem = bracket_ (waitQSem sem) (signalQSem sem) . system

and implement para as simple thread population:

para = mapM_ forkIO

My main motivation is to limit the number of system calls, but it's also useful from a user point of view if the system is doing a handful of things at a time - it makes it easier to track what's going on. I might try that tomorrow and see if it makes a difference to the performance. While the majority of computation is in system calls, quite a few of the threads open files etc, and having them all run in parallel would end up with way too many open handles etc. Thanks Neil