[c10d] Fix stuck after onCompletionhook exception was caught

[HOOLoLo]

Currently, the thread runhookloop will not clean the completedWorkList_ if the onCompletionHook raise an exception.
It leads to the main thread which called waitForPendingWorks will be stucked.

Test Case:

  def test_on_completion_hook_exception(self):
        pg = self._get_process_group()

        def hook(work_info: torch._C._distributed_c10d.WorkInfo):
            raise RuntimeError("hook error")

        pg._register_on_completion_hook(hook)
        tensor = torch.ones([2, 3]).cuda(self.rank) * self.rank
        pg.broadcast([tensor]).wait()

        # N.B.: destroy_process_group is necessary to wait for
        # all pending works to finish.
        c10d.destroy_process_group(pg)

cc @mrshenli @pritamdamania87 @zhaojuanmao @satgera @gqchen @aazzolini @osalpekar @jiayisuse @H-Huang @kwen2501 @awgu @penguinwu @fegin @XilunWu @wanchaol @fduwjj @wz337 @tianyu-l @wconstab @yf225 @chauhang @d4l3k

[pytorch-bot]

🔗 Helpful Links

🧪 See artifacts and rendered test results at hud.pytorch.org/pr/126666

• 📄 Preview Python docs built from this PR
• 📄 Preview C++ docs built from this PR
• ❓ Need help or want to give feedback on the CI? Visit the bot commands wiki or our office hours

Note: Links to docs will display an error until the docs builds have been completed.

✅ No Failures

As of commit c835056 with merge base 5fb11cd ():
💚 Looks good so far! There are no failures yet. 💚

This comment was automatically generated by Dr. CI and updates every 15 minutes.

[HOOLoLo]

Currently, the thread runhookloop will not clean the completedWorkList_ if the onCompletionHook raise an exception. It leads to the main thread which called waitForPendingWorks will be stucked.

Test Case:

  def test_on_completion_hook_exception(self):
        pg = self._get_process_group()

        def hook(work_info: torch._C._distributed_c10d.WorkInfo):
            raise RuntimeError("hook error")

        pg._register_on_completion_hook(hook)
        tensor = torch.ones([2, 3]).cuda(self.rank) * self.rank
        pg.broadcast([tensor]).wait()

        # N.B.: destroy_process_group is necessary to wait for
        # all pending works to finish.
        c10d.destroy_process_group(pg)

cc @mrshenli @pritamdamania87 @zhaojuanmao @satgera @gqchen @aazzolini @osalpekar @jiayisuse @H-Huang @kwen2501 @awgu @penguinwu @fegin @XilunWu @wanchaol @fduwjj @wz337 @tianyu-l @wconstab @yf225 @chauhang @d4l3k

This pr mv try catch into the for loop to deal with each completedwork, and always erase it whether it catch exception or not.
But I don't know why abort comm after a hook exception was caught.
What is going to do when a hook raise exception? kill the process? or just show an information

[wconstab]

@shuqiangzhang can you review?