Refactors process_group_tests.py

[allenwang28]

What does this PR do?

As part of #97, this PR refactors process_group_test:

• Renames up _test_pg to run_collectives and extending it to accept a given list of collectives by name.
• Breaks up ProcessGroupTest into three tests: GlooTest, NCCLTests and DummyTests:
  • GlooTest logically tests every test using gloo, NCCLTest with NCCL, etc.
  • This allows some niceties, like marking once that we want to skip all NCCL tests
• Adds shutdown() and garbage collection etc. to avoid extraneous messages & warnings like

Traceback (most recent call last):
  File "/home/allencwang/workspace/torchft/torchft/process_group.py", line 824, in _future_handler
    cmd = future_queue.get(timeout=timedelta(seconds=10.0))
  File "/home/allencwang/workspace/torchft/torchft/multiprocessing.py", line 45, in get
    raise RuntimeError(f"process is not alive {self._p.exitcode}")
RuntimeError: process is not alive -15
[rank0]:[W207 10:50:12.933128109 CudaIPCTypes.cpp:16] Producer process has been terminated before all shared CUDA tensors released. See Note [Sharing CUDA tensors]

Why is this needed?

As part of #102, I noticed that there were some mismatches between which collectives ran on which backends (matrix is here). Therefore this logical grouping of tests by backend allows us to define which collectives should be tested explicitly

[d4l3k]

this seems like a pretty big decrease in coverage?

[allenwang28]

Added functionality back

[d4l3k]

What's the intention behind pulling this out? I'm not really convinced that this makes it all that much cleaner

In some ways I think I'd prefer if we got rid of the arg generation and instead flatten this out with direct calls i.e.

if collective == "allreduce":
    work = pg.allreduce(...)

work.wait()
...

[allenwang28]

I agree, I've removed the arg generation and included it in place for run_collective

[d4l3k]

This seems really slow -- how fast does this run? Launching the subprocess is pretty slow so would actually prefer to run these all on the same PG

If you want prettier printing we can use subtests?

i.e.

for collective in collectives:
    with self.subTest(collective=collective):
        ...

[allenwang28]

Good callout, with parameterized it took ~36s, without it took ~16s. I've removed parameterized.