Identify lack of scalability in gwas_linear_regression

[eric-czech]

It appears that this function does not scale well when run on a cluster.

Notes from my most recent attempt:

• The code I ran is here: https://github.com/related-sciences/ukb-gwas-pipeline-nealelab/blob/4f862e31b8093d25fdaa8da7f841b9be8583cda4/scripts/gwas.py#L268
• This works on a dataset with fewer variants (chr XY, which has ~8k variants compared to ~141k for chr 21)
• The operation utilizes only one or two workers in a cluster of 20 n1-highmem-8 instances:

CPU utilization across worker VMs

Status Page

• Drilling in on one of the workers that is running all the tasks, I see that the only not obviously parallelizable task it seems to be running is "solve-triangular":

Full Task List

---

The job ultimately failed with the error "ValueError: Could not find dependent ('transpose-e1c6cc7244771a105b73686cc88c4e43', 42, 21). Check worker logs".

Several of the workers show log messages like this:

distributed.worker - INFO - Dependent not found: ('rechunk-merge-66cac011d34e1c66cde96678a9e011b5', 0, 21) 0 . Asking scheduler

Perhaps this is what happens when one node unexpectedly becomes unreachable? I'm not sure.

I will run this again on a smaller dataset that didn't fail to get a performance report and task graph screenshot (which doesn't work on this data because the UI won't render so many nodes).

[eric-czech]

Notes from a more detailed performance report resulting from running this for a smaller dataset (that succeeds):

(renamed to .txt to avoid github attachment filter)
gwas-height-performance-report.html.txt

This zoomed-out view of the task stream in the report doesn't strike me as very healthy:

Task graph:

@mrocklin (cc: @ravwojdyla) do you have any suggestions on how to identify why the work isn't being distributed well on larger datasets for this workflow?

[mrocklin]

Thank you for producing the performance report. If you want to publish these in the future then you may also want to look into gist.github.com and https://raw.githack.com/ .

I've only looked very briefly at it, but the thing that stands out the most is the 760s transfer times leading up to sum tasks. At expected bandwidths, these would be 150GB payloads, which I'm assuming is higher than you're expecting. It's also odd in how synchronized these transfers are, they all end within a few seconds of each other.

[mrocklin]

ValueError: Could not find dependent ('transpose-e1c6cc7244771a105b73686cc88c4e43', 42, 21). Check worker logs

I'm curious, do worker logs report anything strange?

Also I'm curious, which version of distributed are you running? (adding this to the performance report here: dask/distributed#4249)

cc'ing @quasiben , who cares a bit about this space and has a lot of experience tracking down similar performance problems.

[mrocklin]

Also it looks like some of the workers are under memory pressure. I wonder if maybe they're struggling to accept a large-ish payload.

…

On Tue, Nov 17, 2020 at 11:13 AM Eric Czech ***@***.***> wrote: Notes from a more detailed performance report resulting from running this for a smaller dataset (that succeeds): (renamed to .txt to avoid github attachment filter) gwas-height-performance-report.html.txt <https://github.com/pystatgen/sgkit/files/5555640/gwas-height-performance-report.html.txt> This zoomed-out view of the task stream in the report doesn't strike me as very healthy: [image: Screen Shot 2020-11-17 at 2 07 57 PM] <https://user-images.githubusercontent.com/6130352/99435626-62c72080-28de-11eb-9627-560ead8131e8.png> Task graph: [image: Screen Shot 2020-11-17 at 1 54 52 PM] <https://user-images.githubusercontent.com/6130352/99435949-d406d380-28de-11eb-8097-2d3ba8e9bdaf.png> @mrocklin <https://github.com/mrocklin> (cc: @ravwojdyla <https://github.com/ravwojdyla>) do you have any suggestions on how to identify why the work isn't being distributed well on larger datasets for this workflow? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <https://github.com/pystatgen/sgkit/issues/390#issuecomment-729142413>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AACKZTEPOAG7V7AMMB4M3GTSQLDPHANCNFSM4TY6UJFA> .

[quasiben]

When I looked I saw some fairly lengthy disk-read/writes which I assume to be dask spilling. Which might correspond to the workers under memory pressure

[mrocklin]

Two minutes is a long time for a disk read/write. My initial guess is that there is some other external factor causing contention. That's just a wild guess though. Looking at the worker/thread ratio, I wonder if it would make more sense to have far more workers with fewer threads each. Perhaps try four threads per worker?

…

On Tue, Nov 17, 2020 at 12:39 PM Benjamin Zaitlen ***@***.***> wrote: When I looked I saw some fairly lengthy disk-read/writes which I assume to be dask spilling. Which might correspond to the workers under memory pressure [image: Screen Shot 2020-11-17 at 3 32 36 PM] <https://user-images.githubusercontent.com/1403768/99444378-20a3dc00-28ea-11eb-8890-fe2cf70240fd.png> — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <https://github.com/pystatgen/sgkit/issues/390#issuecomment-729193609>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AACKZTGNBX6YOZINCQP2AZDSQLNO3ANCNFSM4TY6UJFA> .

[eric-czech]

Thank you for producing the performance report. If you want to publish these in the future then you may also want to look into gist.github.com and https://raw.githack.com/ .

👍

I'm curious, do worker logs report anything strange?

I see messages like "Worker stream died during communication" so I'm sure a couple workers in the cluster had died. At the end I was down to 18 instead of 20. Full log from one worker is here jic.

When I looked I saw some fairly lengthy disk-read/writes which I assume to be dask spilling. Which might correspond to the workers under memory pressure

Thanks Ben, what do you make of the "transfer-sub" tasks (the long red bars)? Do you have any intuition for what's happening in those?

Memory pressure does seem to be part of the problem -- I rechunked my input to 1/16th of the original chunk size and the whole job has progressed further. Parallel utilization is still pretty disappointing across the whole cluster:

CPU utilization across workers after rechunking input to 1/16th of original

[eric-czech]

Looking at the worker/thread ratio, I wonder if it would make more sense to
have far more workers with fewer threads each. Perhaps try four threads
per worker?

Alright, I can try that. Hey @quasiben, how do you set the number of workers per VM in Cloud Provider?

[mrocklin]

Maybe you can choose smaller VMs?

…

On Tue, Nov 17, 2020, 1:05 PM Eric Czech ***@***.***> wrote: Looking at the worker/thread ratio, I wonder if it would make more sense to have far more workers with fewer threads each. Perhaps try four threads per worker? Alright, I can try that. Hey @quasiben <https://github.com/quasiben>, how do you set the number of workers per VM in Cloud Provider? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <https://github.com/pystatgen/sgkit/issues/390#issuecomment-729213000>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AACKZTDB4HWUPL3D2RSO7C3SQLQQJANCNFSM4TY6UJFA> .

[quasiben]

Alright, I can try that. Hey @quasiben, how do you set the number of workers per VM in Cloud Provider?

That's a good question. I don't think this is supported but shouldn't be too hard. You can control the number of threads with worker_options={"nthreads": 2 }. As @mrocklin suggest, this often is tuned with smaller VMs. Can I ask you to file an issue on dask-cloudprovider to support multiple workers per VM ?

[eric-czech]

Can I ask you to file an issue on dask-cloudprovider to support multiple workers per VM ?

You bet! dask/dask-cloudprovider#173

[eric-czech]

The larger dataset with smaller chunks did ultimately finish with no errors. Here are a couple readouts:

Performance report (26M): https://drive.google.com/file/d/1feWLKNrjQkslKDIZ7T39fPCNQQDBrQFs/view?usp=sharing

It doesn't seem like any of network, disk, or cpu are even close to being saturated so I assume there is some room for improvement.

As a very rough estimate, this task takes about 5 hrs on a single 64 vCPU VM and 3 hours on a cluster of 20 8 vCPU VMs (160 vCPUs). That would imply ~2 hours with perfect scaling so this should be an approximate ceiling for improvements.

Log: dask_gwas_chr21_log.txt

I will try again with smaller VMs and see if there are any major differences.

[eric-czech]

FYI @ravwojdyla and I have been talking a bit about some similar observations on a much simpler workflow in https://github.com/related-sciences/data-team/issues/38 (private). One conclusion there was that the individual objects on GCS are so small that API requests aren't efficient yet bigger chunk sizes would start to make the workloads fail. The GCS objects for a chunk are roughly 2MiB on disk but >100MB in memory. Another potential explanation for this behavior is that the GCS objects aren't being loaded asynchronously. We're still investigating both.

[mrocklin]

I don't know if you're aware of the work done by @martindurant on nicer async IO for remote storage and zarr, but he might be good to talk to.

It looks like you're still primarily blocked by a few oddly long transfers. They're less prominent now, which is good, but probably worth investigating further.