Hi, I'm trying to understand the rationale for your proposed amount of
splitting and more precisely why that one is THE one.
If I put labels on your example numbers in one of your previous post:
nbrOfElements <- 97
nbrOfWorkers <- 5
With these, there are two extremes in how you can split up the
processing in chunks such that all workers are utilized:
(A) Each worker, called multiple times, processes one element each time:
> nbrOfElements <- 97
> nbrOfWorkers <- 5
> nbrOfChunks <- nbrOfElements
> sapply(parallel:::splitList(1:nbrOfElements, nbrOfChunks), length)
[1] 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1
[30] 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1
[59] 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1
[88] 1 1 1 1 1 1 1 1 1 1
(B) Each worker, called once, processes multiple element:
> nbrOfElements <- 97
> nbrOfWorkers <- 5
> nbrOfChunks <- nbrOfWorkers
> sapply(parallel:::splitList(1:nbrOfElements, nbrOfChunks), length)
[1] 20 19 19 19 20
I understand that neither of these two extremes may be the best when
it comes to orchestration overhead and load balancing. Instead, the
best might be somewhere in-between, e.g.
(C) Each worker, called multiple times, processing multiple elements:
> nbrOfElements <- 97
> nbrOfWorkers <- 5
> nbrOfChunks <- nbrOfElements / nbrOfWorkers
> sapply(parallel:::splitList(1:nbrOfElements, nbrOfChunks), length)
[1] 5 5 5 5 4 5 5 5 5 5 4 5 5 5 5 4 5 5 5 5
However, there are multiple alternatives between the two extremes, e.g.
> nbrOfChunks <- scale * nbrOfElements / nbrOfWorkers
So, is there a reason why you argue for scale = 1.0 to be the optimal?
FYI, In future.apply::future_lapply(X, FUN, ...) there is a
'future.scheduling' scale factor(*) argument where default
future.scheduling = 1 corresponds to (B) and future.scheduling = +Inf
to (A). Using future.scheduling = 4 achieves the amount of
load-balancing you propose in (C). (*) Different definition from the
above 'scale'. (Disclaimer: I'm the author)
/Henrik
On Mon, Feb 19, 2018 at 10:21 AM, Christian Krause
<christian.kra...@idiv.de> wrote:
> Dear R-Devel List,
>
> I have installed R 3.4.3 with the patch applied on our cluster and ran a
> *real-world* job of one of our users to confirm that the patch works to my
> satisfaction. Here are the results.
>
> The original was a series of jobs, all essentially doing the same stuff using
> bootstrapped data, so for the original there is more data and I show the
> arithmetic mean with standard deviation. The confirmation with the patched R
> was only a single instance of that series of jobs.
>
> ## Job Efficiency
>
> The job efficiency is defined as (this is what the `qacct-efficiency` tool
> below does):
>
> ```
> efficiency = cputime / cores / wallclocktime * 100%
> ```
>
> In simpler words: how well did the job utilize its CPU cores. It shows the
> percentage of time the job was actually doing stuff, as opposed to the
> difference:
>
> ```
> wasted = 100% - efficiency
> ```
>
> ... which, essentially, tells us how much of the resources were wasted, i.e.
> CPU cores just idling, without being used by anyone. We care a lot about that
> because, for our scientific computing cluster, wasted resources is like
> burning money.
>
> ### original
>
> This is the entire series from our job accounting database, filteres the
> successful jobs, calculates efficiency and then shows the average and
> standard deviation of the efficiency:
>
> ```
> $ qacct -j 4433299 | qacct-success | qacct-efficiency | meansd
> n=945 ∅ 61.7276 ± 7.78719
> ```
>
> This is the entire series from our job accounting database, filteres the
> successful jobs, calculates efficiency and does sort of a histogram-like
> binning before calculation of mean and standard deviation (to get a more
> detailed impression of the distribution when standard deviation of the
> previous command is comparatively high):
>
> ```
> $ qacct -j 4433299 | qacct-success | qacct-efficiency | meansd-bin -w 10 |
> sort -gk1 | column -t
> 10 - 20 -> n=3 ∅ 19.21666666666667 ± 0.9112811494447459
> 20 - 30 -> n=6 ∅ 26.418333333333333 ± 2.665996374091058
> 30 - 40 -> n=12 ∅ 35.11583333333334 ± 2.8575783082671196
> 40 - 50 -> n=14 ∅ 45.35285714285715 ± 2.98623361591005
> 50 - 60 -> n=344 ∅ 57.114593023255814 ± 2.1922005551774415
> 60 - 70 -> n=453 ∅ 64.29536423841049 ± 2.8334788433963856
> 70 - 80 -> n=108 ∅ 72.95592592592598 ± 2.5219474143639276
> 80 - 90 -> n=5 ∅ 81.526 ± 1.2802265424525452
> ```
>
> I have attached an example graph from our monitoring system of a single
> instance in my previous mail. There you can see that the load balancing does
> not actually work, i.e. same as `parLapply`. This reflects in the job
> efficiency.
>
> ### patch applied
>
> This is the single instance I used to confirm that the patch works:
>
> ```
> $ qacct -j 4562202 | qacct-efficiency
> 97.36
> ```
>
> The graph from our monitoring system is attached. As you can see, the load
> balancing works to a satisfying degree and the efficiency is well above 90%
> which was what I had hoped for :-)
>
> ## Additional Notes
>
> The list used in this jobs `parLapplyLB` is 5812 elements long. With the
> `splitList`-chunking from the patch, you'll get 208 lists of about 28
> elements (208 chunks of size 28). The job ran on 28 CPU cores and had a
> wallclock time of 120351.590 seconds, i.e. 33.43 hours. Thus, the function we
> apply to our list takes about 580 seconds per list element, i.e. about 10
> minutes. I suppose, for that runtime, we would get even better load balancing
> if we would reduce the chunk size even further, maybe even down to 1, thus
> getting our efficiency even closer to 100%.
>
> Of course, for really short-running functions, a higher chunk size may be
> more efficient because of the overhead. In our case, the overhead is
> negligible and that is why the low chunk size works really well. In contrast,
> for smallish lists with short-running functions, you might not even need load
> balancing and `parLapply` suffices. It only becomes an issue, when the
> runtime of the function is high and / or varying.
>
> In our case, the entire runtime of the entire series of jobs was:
>
> ```
> $ qacct -j 4433299 | awk '$1 == "wallclock" { sum += $2 } END { print sum,
> "seconds" }'
> 4.72439e+09 seconds
> ```
>
> Thats about 150 years on a single core or 7.5 years on a 20 core server! Our
> user was constantly using about 500 cores, so this took about 110 days. If
> you compare this to my 97% efficiency example, the jobs could have been
> finished in 75 days instead ;-)
>
> ## Upcoming Patch
>
> If this patch gets applied to the R code base (and I hope it will :-)) my
> colleague and I will submit another patch that adds the chunk size as an
> optional parameter to all off the load balancing functions. With that
> parameter, users of these functions *can* decide for themselves which chunk
> size they prefer for their code. As mentioned before, the most efficient
> chunk size depends on the used functions runtime, which is the only thing R
> does not know and users really should be allowed to specify explicitly. The
> default of this new optional parameter would be the one we used here and this
> would make that upcoming patch fully source-compatible.
>
> Best Regards
>
> On 02/12/2018 08:08 PM, Christian Krause wrote:
>> Dear R-Devel List,
>>
>> **TL;DR:** The function **parLapplyLB** of the parallel package has
>> [reportedly][1] (see also attached RRD output) not
>> been doing its job, i.e. not actually balancing the load. My colleague Dirk
>> Sarpe and I found the cause of the problem
>> and we also have a patch to fix it (attached). A similar fix has also been
>> provided [here][2].
>>
>> [1]:
>> https://stackoverflow.com/questions/38230831/why-does-parlapplylb-not-actually-balance-load
>> [2]: https://bugs.r-project.org/bugzilla3/show_bug.cgi?id=16792
>>
>>
>> ## The Call Chain
>>
>> First, we traced the relevant R function calls through the code, beginning
>> with `parLapplyLB`:
>>
>> 1. **parLapplyLB:** clusterApply.R:177, calls **splitList**, then
>> **clusterApplyLB**
>> 2. **splitList:** clusterApply.R:157
>> 3. **clusterApplyLB:** clusterApply.R:87, calls **dynamicClusterApply**
>> 4. **dynamicClusterApply:** clusterApply.R:39
>>
>>
>> ## splitList
>>
>> We used both our whiteboard and an R session to manually *run* a few
>> examples. We were using lists of 100 elements and 5
>> workers. First, lets take a look at **splitList**:
>>
>> ```r
>>> sapply(parallel:::splitList(1:100, 5), length)
>> [1] 20 20 20 20 20
>>
>>> sapply(parallel:::splitList(1:97, 5), length)
>> [1] 20 19 19 19 20
>>
>>> sapply(parallel:::splitList(1:97, 20), length)
>> [1] 5 5 5 5 4 5 5 5 5 5 4 5 5 5 5 4 5 5 5 5
>> ```
>>
>> As we can see in the examples, the work is distributed as equally as
>> possible.
>>
>>
>> ## dynamicClusterApply
>>
>> **dynamicClusterApply** works this way (simplified):
>>
>> 1. it first gives a chunk to each worker
>> 2. once a worker comes back with the result, it is given the next chunk
>>
>> **This is the important part:** As long as there are **more** chunks than
>> workers, there will be load balancing. If
>> there are fewer chunks than workers, each worker will get **at most one
>> chunk** and there is **no** load balancing.
>>
>>
>> ## parLapplyLB
>>
>> This is how **parLapplyLB** splits the input list (with a bit of
>> refactoring, for readability):
>>
>> ```r
>> parLapplyLB <- function(cl = NULL, X, fun, ...)
>> {
>> cl <- defaultCluster(cl)
>>
>> chunks <- splitList(X, length(cl))
>>
>> do.call(c,
>> clusterApplyLB(cl, x = chunks, fun = lapply, fun, ...),
>> quote = TRUE)
>> }
>> ```
>>
>> For our examples, the chunks have these sizes:
>>
>> ```r
>>> sapply(parallel:::splitList(1:100, 5), length)
>> [1] 20 20 20 20 20
>> ```
>>
>> There we have it: 5 chunks. 5 workers. With this work distribution, there
>> can't possibly be any load balancing, because
>> each worker is given a single chunk and then it stops working because there
>> are no more chunks.
>>
>> Instead, **parLapplyLB** should look like this (patch is attached):
>>
>> ```r
>> parLapplyLB <- function(cl = NULL, X, fun, ...)
>> {
>> cl <- defaultCluster(cl)
>>
>> chunkSize <- max(length(cl), ceiling(length(X) / length(cl)))
>>
>> chunks <- splitList(X, chunkSize)
>>
>> do.call(c,
>> clusterApplyLB(cl, x = chunks, fun = lapply, fun, ...),
>> quote = TRUE)
>> }
>> ```
>>
>> Examples with a cluster of 5 workers:
>>
>> ```r
>> # length(cl) < length(X)
>>> sapply(parallel:::splitList(1:100, ceiling(100 / 5)), length)
>> [1] 5 5 5 5 5 5 5 5 5 5 5 5 5 5 5 5 5 5 5 5
>>
>> # length(cl) >= length(X)
>>> sapply(parallel:::splitList(1:4, 4), length)
>> [1] 1 1 1 1
>> # one worker idles here, but we can't do better than that
>> ```
>>
>> With this patch, the number of chunks is larger than the number of workers,
>> if possible at all, and then load balancing
>> should work.
>>
>> Best Regards
>>
>>
>>
>> ______________________________________________
>> R-devel@r-project.org mailing list
>> https://stat.ethz.ch/mailman/listinfo/r-devel
>>
>
> --
> Christian Krause
>
> Scientific Computing Administration and Support
>
> ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Email: christian.kra...@idiv.de
>
> Office: BioCity Leipzig 5e, Room 3.201.3
>
> Phone: +49 341 97 33144
>
> ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> German Centre for Integrative Biodiversity Research (iDiv) Halle-Jena-Leipzig
>
> Deutscher Platz 5e
>
> 04103 Leipzig
>
> Germany
>
> ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> iDiv is a research centre of the DFG – Deutsche Forschungsgemeinschaft
>
> iDiv ist eine zentrale Einrichtung der Universität Leipzig im Sinne des § 92
> Abs. 1 SächsHSFG und wird zusammen mit der Martin-Luther-Universität
> Halle-Wittenberg und der Friedrich-Schiller-Universität Jena betrieben sowie
> in Kooperation mit dem Helmholtz-Zentrum für Umweltforschung GmbH – UFZ.
> Beteiligte Kooperationspartner sind die folgenden außeruniversitären
> Forschungseinrichtungen: das Helmholtz-Zentrum für Umweltforschung GmbH -
> UFZ, das Max-Planck-Institut für Biogeochemie (MPI BGC), das
> Max-Planck-Institut für chemische Ökologie (MPI CE), das Max-Planck-Institut
> für evolutionäre Anthropologie (MPI EVA), das Leibniz-Institut Deutsche
> Sammlung von Mikroorganismen und Zellkulturen (DSMZ), das Leibniz-Institut
> für Pflanzenbiochemie (IPB), das Leibniz-Institut für Pflanzengenetik und
> Kulturpflanzenforschung (IPK) und das Leibniz-Institut Senckenberg Museum für
> Naturkunde Görlitz (SMNG). USt-IdNr. DE 141510383
>
>
> ______________________________________________
> R-devel@r-project.org mailing list
> https://stat.ethz.ch/mailman/listinfo/r-devel
>
______________________________________________
R-devel@r-project.org mailing list
https://stat.ethz.ch/mailman/listinfo/r-devel
