PERF-#239: Improve task scheduler for MPI backend

[arunjose696]

What do these changes do?

Improve task scheduler for MPI backend, Replaced the round robin algorithm to submit tasks, with a getting a worker rank with maximum data share. Data is pushed to workers in the put method.
Data id and data sizes would be stored in object store.

• passes flake8 .
• passes black .
• signed commit with git commit -s
• Resolves [MPI] Improve task scheduler for MPI backend #239
• tests added and passing
• module layout described at docs/developer/architecture.rst is up-to-date

[YarShev]

Why do we need to put data into the object store of the master process? We want to push data to a worker and save the owner of the data into the object store of the main process. In order to achieve this we should probably rework push_data method.

[arunjose696]

Have added an additional function(push_data_directly_to_worker) to push data directly without storing in object store. Did not modify push_data method as the new method would need the actual data also as argument and it relied on object_store.

[YarShev]

As far as I understand, there is a case when we push the data to, for instance, rank 2 during ref = unidist.put(data) and then we push the data owner (the same worker with rank 2) to it itself during foo.remote(ref). We should probably avoid such cases, i.e., don't push the data owner if the worker is the data owner itself.

# simple example
import unidist

unidist.init()

o_ref = unidist.put(1)

@unidist.remote
def foo(obj):
    return obj

foo.remote(o_ref) # here we push the owner (rank 2) to the worker with rank 2

[YarShev]

Is it possible if we don't get into any of if branches? You can add raise an exception at the end of the function to check this when testing with Modin.

[arunjose696]

Here as the plan is to just collect data_ids, recursively so if the args are not data_ids, eg functions, python objects etc it is expected not to get into any if branches. Will add an exception if object store does not contain an owner.

[YarShev]

When is data_ids an empty list?

[arunjose696]

This would for the case where there are no arguments for the remote function. So the data_ids would be empty and none of workers will have data shares so submitting them using round robin.

eg.

@unidist.remotedef foo():
    return 1 
o = foo.remote()

[YarShev]

Put a comment describing this into the code.

[YarShev]

Why do we need this map?

[arunjose696]

This is no longer required, will remove this.

The original purpose of this map was to make sure to not send same python object to workers multiple times, as put was planned to be used in submit method for all arguments but this was later changed.