DaskCluster: a general dask wrapper for mintpy

[Ovec8hkin]

Generalized use of Dask with creation of DaskCluster object that handles cluster setup, client connections, and worker submission and processing (#354). This should make integrating dask into other scripts much easier.

And update the Dask performance figure of runtime vs number of cores (6 in #347).

Reminders

• Pass Codacy code review (green)
• Pass testing with $MINTPY_HOME/test/test_smallbaselineApp.py
• Make sure that your code follows our style. Use the other functions/files as a basis.
• If modifying functionality, describe changes to function behavior and arguments in a comment below the function declaration.
• If adding new functionality, add a detailed description to the documentation and/or an example.

[Ovec8hkin]

@yunjunz @falkamelung I would have someone checkout this branch and test it on reasonably large dataset (AND LOOK AT THE RESULTS) to make sure it works as expected. I messed it up in a few places when I initially wrote it, which caused only part of the dataset to run correctly. Kilauea would be a good dataset to test on in my opinion.

[yunjunz]

From my testing with two datasets, the current memory usage is not quite as specified in the -r option. We need to re-check the following to make sure memorySize we input is the max memory the program is gonna use during the whole time period. This is important for job scheduling on HPC where control/check is more restrict.

• the calculation from memorySize to chunkSize
• for -w var a temporary float64 is used, this might bring up the memory (now fixed)
• in the end of the cluster job, there is a memory surge, need to find out why.

[yunjunz]

On my laptop, when I have 2 ifgram_inversion.py running, the following message came out. I guess it's fine. The "6 workers" is a little bit weird, will check again in more details tomorrow (it was a bug, now it's fixed).

------- start parallel processing using Dask -------
input Dask cluster type: local
initiate Dask cluster
/opt/local/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/distributed/node.py:244: UserWarning: Port 8787 is already in use.
Perhaps you already have a cluster running?
Hosting the HTTP server on port 57863 instead
  http_address["port"], self.http_server.port
scale the cluster to 6 workers

FYI, I was running ifgram_inversion.py inputs/ifgramStack.h5 -w var for the Fernandina dataset in one terminal and $MINTPY_HOME/test/test_smallbaselineApp.py in another terminal.

[yunjunz]

@Ovec8hkin could we remove the dask-worker-space folder after the running?

[Ovec8hkin]

@yunjunz these changes to cluster.py use quite a few non-pythonic anti patterns that I think we should avoid. Namely, there doesn't need to be empty returns in most of the functions, and most of the variables do not need to be written as self.parameter_name, especially those that are simply computed and then used once within their own function (ie. self.futures, self.start_time_sub) Permission to clean this up?

[Ovec8hkin]

On my laptop, when I have 2 ifgram_inversion.py running, the following message came out. I guess it's fine. The "6 workers" is a little bit weird, will check again in more details tomorrow (it was a bug, now it's fixed).

------- start parallel processing using Dask -------
input Dask cluster type: local
initiate Dask cluster
/opt/local/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/distributed/node.py:244: UserWarning: Port 8787 is already in use.
Perhaps you already have a cluster running?
Hosting the HTTP server on port 57863 instead
  http_address["port"], self.http_server.port
scale the cluster to 6 workers

FYI, I was running ifgram_inversion.py inputs/ifgramStack.h5 -w var for the Fernandina dataset in one terminal and $MINTPY_HOME/test/test_smallbaselineApp.py in another terminal.

This it probably due to how dask creates its worker Client object. Im not sure what happens if multiple client objects are running simultaneously. Its possible they are sharing workers. I would advise against running multiple ifgram_inversion processes simultaneously, as its very possible that workers submitted from one client, might finish and get collected by a different running client, which could mess up results. This would need to be tested more carefully.

[yunjunz]

Hi @Ovec8hkin, the Dask cluster port change message is fine as explained, so we can ignore it I think.

For the variables in the cluster.py, if you find variables used once and can be cleaned, yes please.

From my side, the last thing to check is the memory. With the default 4GB memory input, ifgram_inversion.py is actually using up to 10 GB in the process. We definitely want to avoid that. But this is not related to Dask, I guess we do that in another issue / pull request.

The rest looks all good to me!

[yunjunz]

Here is an overview of what got changed by this pull request:

Complexity increasing per file
==============================
- mintpy/ifgram_inversion.py  1
         

Complexity decreasing per file
==============================
+ mintpy/objects/cluster.py  -5
         

Clones removed
==============
+ mintpy/ifgram_inversion.py  -2
         

See the complete overview on Codacy

[yunjunz]

All the tests look good and normal. After so many rounds of refactoring, I think now ifgram_inversion.py is the most beautiful script in this repo!