prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  The first worker may crash in ALLREDUCE mode 
  When the worker is the only running worker pod, `_get_peer_set` will get an empty peer set.
  Then consensus_init_kwars will set "known_addr_list" as empty.
  This will cause an error in  ftlib.
  
  
  ```
  [2020-03-30 06:16:07,202] [WARNING] [communicator.py:102:_get_peer_set] svc_name is edl-allreduce-haitao-ftlib-consensus
  [2020-03-30 06:16:09,206] [WARNING] [communicator.py:108:_get_peer_set] peer_set is set()
  Setting Bind Address as 11.233.87.89
  log file: /tmp/memberlist.log
  [2020-03-30 06:16:21,713] [WARNING] [communicator.py:37:__init__] Retry building consensus...
  [2020-03-30 06:16:21,713] [WARNING] [communicator.py:102:_get_peer_set] svc_name is edl-allreduce-haitao-ftlib-consensus
  [2020-03-30 06:16:21,714] [WARNING] [communicator.py:108:_get_peer_set] peer_set is set()
  Traceback (most recent call last):
    File "/usr/lib/python3.6/runpy.py", line 193, in _run_module_as_main
      "__main__", mod_spec)
    File "/usr/lib/python3.6/runpy.py", line 85, in _run_code
      exec(code, run_globals)
    File "/elasticdl/elasticdl/python/worker/main.py", line 76, in <module>
      main()
    File "/elasticdl/elasticdl/python/worker/main.py", line 70, in main
      set_parallelism=True,
    File "/elasticdl/elasticdl/python/worker/worker.py", line 122, in __init__
      self._init_from_args(args)
    File "/elasticdl/elasticdl/python/worker/worker.py", line 159, in _init_from_args
      if self._distribution_strategy == DistributionStrategy.ALLREDUCE
    File "/elasticdl/elasticdl/python/collective_ops/communicator.py", line 39, in __init__
      known_addr_list=list(self._get_peer_set(service_name))
    File "/usr/local/lib/python3.6/dist-packages/ftlib/impl.py", line 137, in manual_join
      return self.consensus.manual_join(*args, **kwargs)
    File "/usr/local/lib/python3.6/dist-packages/ftlib/consensus/gossip/impl.py", line 85, in manual_join
      self.joined = self._join(known_addr_list, wait_time=wait_time)
    File "/usr/local/lib/python3.6/dist-packages/ftlib/consensus/gossip/impl.py", line 92, in _join
      assert addr_list_len >= 1
  AssertionError
  ```
  
  --- END ISSUE ---
  
  Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
  --- BEGIN FILES ---
  Path: `elasticdl/python/collective_ops/communicator.py`
  Content:
  ```
  1 import socket
  2 
  3 from elasticdl.python.common.constants import CollectiveCommunicatorStatus
  4 from elasticdl.python.common.log_utils import default_logger as logger
  5 
  6 try:
  7     from ftlib import BasicFTLib
  8     from ftlib.ftlib_status import FTAllReduceStatus
  9 
  10     _FTLIB_INSTALLED = True
  11 except ImportError:
  12     BasicFTLib = object
  13     FTAllReduceStatus = object
  14     _FTLIB_INSTALLED = False
  15 
  16 
  17 _SUPPORTED_ALLREDUCE_OPS = ["MEAN"]
  18 _FTLIB_UNINSTALLED_DEFAULT_STATUS_MESSAGE = (
  19     "FTLib is not installed. Default to succeeded for testing purposes"
  20 )
  21 
  22 
  23 class CollectiveCommunicator(object):
  24     def __init__(self, service_name=None):
  25         if _FTLIB_INSTALLED:
  26             self._ftlib = BasicFTLib(
  27                 consensus="gossip",
  28                 commlib="pytorch",
  29                 consensus_init_kwargs={
  30                     "known_addr_list": list(self._get_peer_set(service_name)),
  31                     "custom_bind_addr": socket.gethostbyname(
  32                         socket.gethostname()
  33                     ),
  34                 },
  35             )
  36             while not self._ftlib.consensus_joined():
  37                 logger.warning("Retry building consensus...")
  38                 self._ftlib.manual_join(
  39                     known_addr_list=list(self._get_peer_set(service_name))
  40                 )
  41         else:
  42             logger.warning(
  43                 "FTLib is not installed. The CollectiveCommunicator "
  44                 "may not work as expected"
  45             )
  46             self._ftlib = None
  47 
  48     def allreduce(self, data, op="MEAN"):
  49         if data is None:
  50             logger.error("Data is required for allreduce operation")
  51             return CollectiveCommunicatorStatus.FAILED, data
  52         if op not in _SUPPORTED_ALLREDUCE_OPS:
  53             logger.error(
  54                 "%s is not in list of supported allreduce operations: %s"
  55                 % (op, _SUPPORTED_ALLREDUCE_OPS)
  56             )
  57             return CollectiveCommunicatorStatus.FAILED, data
  58         if self._ftlib is not None:
  59             res = self._ftlib.wait_gradients_ready(data)
  60             if res == FTAllReduceStatus.SUCCESS:
  61                 return CollectiveCommunicatorStatus.SUCCEEDED, data
  62             else:
  63                 return CollectiveCommunicatorStatus.FAILED, data
  64         else:
  65             logger.warning(_FTLIB_UNINSTALLED_DEFAULT_STATUS_MESSAGE)
  66             return CollectiveCommunicatorStatus.SUCCEEDED, data
  67 
  68     def broadcast(self, data, src_rank):
  69         if self._ftlib is not None:
  70             res = self._ftlib.broadcast(data, src_rank)
  71             if res == FTAllReduceStatus.SUCCESS:
  72                 return CollectiveCommunicatorStatus.SUCCEEDED, data
  73             else:
  74                 return CollectiveCommunicatorStatus.FAILED, data
  75         else:
  76             logger.warning(_FTLIB_UNINSTALLED_DEFAULT_STATUS_MESSAGE)
  77             return CollectiveCommunicatorStatus.SUCCEEDED, data
  78 
  79     def barrier(self):
  80         if self._ftlib is not None:
  81             res = self._ftlib.barrier()
  82             if res == FTAllReduceStatus.SUCCESS:
  83                 return CollectiveCommunicatorStatus.SUCCEEDED
  84             else:
  85                 return CollectiveCommunicatorStatus.FAILED
  86         else:
  87             logger.warning(_FTLIB_UNINSTALLED_DEFAULT_STATUS_MESSAGE)
  88             return CollectiveCommunicatorStatus.SUCCEEDED
  89 
  90     def is_initialized(self):
  91         """This will be `False` under three occasions:
  92            * New workers report joining in
  93            * Collective-communication operations fail or time out
  94            * Liveness probe fails for existing workers
  95         """
  96         if self._ftlib is not None:
  97             return self._ftlib.initialized
  98         else:
  99             return True
  100 
  101     def _get_peer_set(self, svc_name):
  102         if svc_name is None:
  103             return None
  104         my_ip = socket.gethostbyname(socket.gethostname())
  105         temp_set = socket.getaddrinfo(svc_name, 0, proto=socket.IPPROTO_TCP)
  106         peer_set = {peer[-1][0] for peer in temp_set if peer[-1][0] != my_ip}
  107         return peer_set
  108
  ```
  --- END FILES ---
  
  Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.
  
  Here is an example:
  
  ```diff
  diff --git a/examples/server_async.py b/examples/server_async.py
  --- a/examples/server_async.py
  +++ b/examples/server_async.py
  @@ -313,4 +313,4 @@
   
   
   if __name__ == "__main__":
  -    asyncio.run(run_async_server("."), debug=True)
  +    asyncio.run(run_async_server(), debug=True)
  diff --git a/examples/server_sync.py b/examples/server_sync.py
  --- a/examples/server_sync.py
  +++ b/examples/server_sync.py
  @@ -313,5 +313,5 @@
   
   
   if __name__ == "__main__":
  -    server = run_sync_server(".")
  +    server = run_sync_server()
       server.shutdown()
  
  ```
  
golden_diff: diff --git a/elasticdl/python/collective_ops/communicator.py b/elasticdl/python/collective_ops/communicator.py
  --- a/elasticdl/python/collective_ops/communicator.py
  +++ b/elasticdl/python/collective_ops/communicator.py
  @@ -23,17 +23,18 @@
   class CollectiveCommunicator(object):
       def __init__(self, service_name=None):
           if _FTLIB_INSTALLED:
  +            peer_list = list(self._get_peer_set(service_name))
               self._ftlib = BasicFTLib(
                   consensus="gossip",
                   commlib="pytorch",
                   consensus_init_kwargs={
  -                    "known_addr_list": list(self._get_peer_set(service_name)),
  +                    "known_addr_list": peer_list,
                       "custom_bind_addr": socket.gethostbyname(
                           socket.gethostname()
                       ),
                   },
               )
  -            while not self._ftlib.consensus_joined():
  +            while peer_list and not self._ftlib.consensus_joined():
                   logger.warning("Retry building consensus...")
                   self._ftlib.manual_join(
                       known_addr_list=list(self._get_peer_set(service_name))
  
verification_info: {"golden_diff": "diff --git a/elasticdl/python/collective_ops/communicator.py b/elasticdl/python/collective_ops/communicator.py\n--- a/elasticdl/python/collective_ops/communicator.py\n+++ b/elasticdl/python/collective_ops/communicator.py\n@@ -23,17 +23,18 @@\n class CollectiveCommunicator(object):\n     def __init__(self, service_name=None):\n         if _FTLIB_INSTALLED:\n+            peer_list = list(self._get_peer_set(service_name))\n             self._ftlib = BasicFTLib(\n                 consensus=\"gossip\",\n                 commlib=\"pytorch\",\n                 consensus_init_kwargs={\n-                    \"known_addr_list\": list(self._get_peer_set(service_name)),\n+                    \"known_addr_list\": peer_list,\n                     \"custom_bind_addr\": socket.gethostbyname(\n                         socket.gethostname()\n                     ),\n                 },\n             )\n-            while not self._ftlib.consensus_joined():\n+            while peer_list and not self._ftlib.consensus_joined():\n                 logger.warning(\"Retry building consensus...\")\n                 self._ftlib.manual_join(\n                     known_addr_list=list(self._get_peer_set(service_name))\n", "issue": "The first worker may crash in ALLREDUCE mode \nWhen the worker is the only running worker pod, `_get_peer_set` will get an empty peer set.\r\nThen consensus_init_kwars will set \"known_addr_list\" as empty.\r\nThis will cause an error in  ftlib.\r\n\r\n\r\n```\r\n[2020-03-30 06:16:07,202] [WARNING] [communicator.py:102:_get_peer_set] svc_name is edl-allreduce-haitao-ftlib-consensus\r\n[2020-03-30 06:16:09,206] [WARNING] [communicator.py:108:_get_peer_set] peer_set is set()\r\nSetting Bind Address as 11.233.87.89\r\nlog file: /tmp/memberlist.log\r\n[2020-03-30 06:16:21,713] [WARNING] [communicator.py:37:__init__] Retry building consensus...\r\n[2020-03-30 06:16:21,713] [WARNING] [communicator.py:102:_get_peer_set] svc_name is edl-allreduce-haitao-ftlib-consensus\r\n[2020-03-30 06:16:21,714] [WARNING] [communicator.py:108:_get_peer_set] peer_set is set()\r\nTraceback (most recent call last):\r\n  File \"/usr/lib/python3.6/runpy.py\", line 193, in _run_module_as_main\r\n    \"__main__\", mod_spec)\r\n  File \"/usr/lib/python3.6/runpy.py\", line 85, in _run_code\r\n    exec(code, run_globals)\r\n  File \"/elasticdl/elasticdl/python/worker/main.py\", line 76, in <module>\r\n    main()\r\n  File \"/elasticdl/elasticdl/python/worker/main.py\", line 70, in main\r\n    set_parallelism=True,\r\n  File \"/elasticdl/elasticdl/python/worker/worker.py\", line 122, in __init__\r\n    self._init_from_args(args)\r\n  File \"/elasticdl/elasticdl/python/worker/worker.py\", line 159, in _init_from_args\r\n    if self._distribution_strategy == DistributionStrategy.ALLREDUCE\r\n  File \"/elasticdl/elasticdl/python/collective_ops/communicator.py\", line 39, in __init__\r\n    known_addr_list=list(self._get_peer_set(service_name))\r\n  File \"/usr/local/lib/python3.6/dist-packages/ftlib/impl.py\", line 137, in manual_join\r\n    return self.consensus.manual_join(*args, **kwargs)\r\n  File \"/usr/local/lib/python3.6/dist-packages/ftlib/consensus/gossip/impl.py\", line 85, in manual_join\r\n    self.joined = self._join(known_addr_list, wait_time=wait_time)\r\n  File \"/usr/local/lib/python3.6/dist-packages/ftlib/consensus/gossip/impl.py\", line 92, in _join\r\n    assert addr_list_len >= 1\r\nAssertionError\r\n```\n", "before_files": [{"content": "import socket\n\nfrom elasticdl.python.common.constants import CollectiveCommunicatorStatus\nfrom elasticdl.python.common.log_utils import default_logger as logger\n\ntry:\n    from ftlib import BasicFTLib\n    from ftlib.ftlib_status import FTAllReduceStatus\n\n    _FTLIB_INSTALLED = True\nexcept ImportError:\n    BasicFTLib = object\n    FTAllReduceStatus = object\n    _FTLIB_INSTALLED = False\n\n\n_SUPPORTED_ALLREDUCE_OPS = [\"MEAN\"]\n_FTLIB_UNINSTALLED_DEFAULT_STATUS_MESSAGE = (\n    \"FTLib is not installed. Default to succeeded for testing purposes\"\n)\n\n\nclass CollectiveCommunicator(object):\n    def __init__(self, service_name=None):\n        if _FTLIB_INSTALLED:\n            self._ftlib = BasicFTLib(\n                consensus=\"gossip\",\n                commlib=\"pytorch\",\n                consensus_init_kwargs={\n                    \"known_addr_list\": list(self._get_peer_set(service_name)),\n                    \"custom_bind_addr\": socket.gethostbyname(\n                        socket.gethostname()\n                    ),\n                },\n            )\n            while not self._ftlib.consensus_joined():\n                logger.warning(\"Retry building consensus...\")\n                self._ftlib.manual_join(\n                    known_addr_list=list(self._get_peer_set(service_name))\n                )\n        else:\n            logger.warning(\n                \"FTLib is not installed. The CollectiveCommunicator \"\n                \"may not work as expected\"\n            )\n            self._ftlib = None\n\n    def allreduce(self, data, op=\"MEAN\"):\n        if data is None:\n            logger.error(\"Data is required for allreduce operation\")\n            return CollectiveCommunicatorStatus.FAILED, data\n        if op not in _SUPPORTED_ALLREDUCE_OPS:\n            logger.error(\n                \"%s is not in list of supported allreduce operations: %s\"\n                % (op, _SUPPORTED_ALLREDUCE_OPS)\n            )\n            return CollectiveCommunicatorStatus.FAILED, data\n        if self._ftlib is not None:\n            res = self._ftlib.wait_gradients_ready(data)\n            if res == FTAllReduceStatus.SUCCESS:\n                return CollectiveCommunicatorStatus.SUCCEEDED, data\n            else:\n                return CollectiveCommunicatorStatus.FAILED, data\n        else:\n            logger.warning(_FTLIB_UNINSTALLED_DEFAULT_STATUS_MESSAGE)\n            return CollectiveCommunicatorStatus.SUCCEEDED, data\n\n    def broadcast(self, data, src_rank):\n        if self._ftlib is not None:\n            res = self._ftlib.broadcast(data, src_rank)\n            if res == FTAllReduceStatus.SUCCESS:\n                return CollectiveCommunicatorStatus.SUCCEEDED, data\n            else:\n                return CollectiveCommunicatorStatus.FAILED, data\n        else:\n            logger.warning(_FTLIB_UNINSTALLED_DEFAULT_STATUS_MESSAGE)\n            return CollectiveCommunicatorStatus.SUCCEEDED, data\n\n    def barrier(self):\n        if self._ftlib is not None:\n            res = self._ftlib.barrier()\n            if res == FTAllReduceStatus.SUCCESS:\n                return CollectiveCommunicatorStatus.SUCCEEDED\n            else:\n                return CollectiveCommunicatorStatus.FAILED\n        else:\n            logger.warning(_FTLIB_UNINSTALLED_DEFAULT_STATUS_MESSAGE)\n            return CollectiveCommunicatorStatus.SUCCEEDED\n\n    def is_initialized(self):\n        \"\"\"This will be `False` under three occasions:\n           * New workers report joining in\n           * Collective-communication operations fail or time out\n           * Liveness probe fails for existing workers\n        \"\"\"\n        if self._ftlib is not None:\n            return self._ftlib.initialized\n        else:\n            return True\n\n    def _get_peer_set(self, svc_name):\n        if svc_name is None:\n            return None\n        my_ip = socket.gethostbyname(socket.gethostname())\n        temp_set = socket.getaddrinfo(svc_name, 0, proto=socket.IPPROTO_TCP)\n        peer_set = {peer[-1][0] for peer in temp_set if peer[-1][0] != my_ip}\n        return peer_set\n", "path": "elasticdl/python/collective_ops/communicator.py"}], "after_files": [{"content": "import socket\n\nfrom elasticdl.python.common.constants import CollectiveCommunicatorStatus\nfrom elasticdl.python.common.log_utils import default_logger as logger\n\ntry:\n    from ftlib import BasicFTLib\n    from ftlib.ftlib_status import FTAllReduceStatus\n\n    _FTLIB_INSTALLED = True\nexcept ImportError:\n    BasicFTLib = object\n    FTAllReduceStatus = object\n    _FTLIB_INSTALLED = False\n\n\n_SUPPORTED_ALLREDUCE_OPS = [\"MEAN\"]\n_FTLIB_UNINSTALLED_DEFAULT_STATUS_MESSAGE = (\n    \"FTLib is not installed. Default to succeeded for testing purposes\"\n)\n\n\nclass CollectiveCommunicator(object):\n    def __init__(self, service_name=None):\n        if _FTLIB_INSTALLED:\n            peer_list = list(self._get_peer_set(service_name))\n            self._ftlib = BasicFTLib(\n                consensus=\"gossip\",\n                commlib=\"pytorch\",\n                consensus_init_kwargs={\n                    \"known_addr_list\": peer_list,\n                    \"custom_bind_addr\": socket.gethostbyname(\n                        socket.gethostname()\n                    ),\n                },\n            )\n            while peer_list and not self._ftlib.consensus_joined():\n                logger.warning(\"Retry building consensus...\")\n                self._ftlib.manual_join(\n                    known_addr_list=list(self._get_peer_set(service_name))\n                )\n        else:\n            logger.warning(\n                \"FTLib is not installed. The CollectiveCommunicator \"\n                \"may not work as expected\"\n            )\n            self._ftlib = None\n\n    def allreduce(self, data, op=\"MEAN\"):\n        if data is None:\n            logger.error(\"Data is required for allreduce operation\")\n            return CollectiveCommunicatorStatus.FAILED, data\n        if op not in _SUPPORTED_ALLREDUCE_OPS:\n            logger.error(\n                \"%s is not in list of supported allreduce operations: %s\"\n                % (op, _SUPPORTED_ALLREDUCE_OPS)\n            )\n            return CollectiveCommunicatorStatus.FAILED, data\n        if self._ftlib is not None:\n            res = self._ftlib.wait_gradients_ready(data)\n            if res == FTAllReduceStatus.SUCCESS:\n                return CollectiveCommunicatorStatus.SUCCEEDED, data\n            else:\n                return CollectiveCommunicatorStatus.FAILED, data\n        else:\n            logger.warning(_FTLIB_UNINSTALLED_DEFAULT_STATUS_MESSAGE)\n            return CollectiveCommunicatorStatus.SUCCEEDED, data\n\n    def broadcast(self, data, src_rank):\n        if self._ftlib is not None:\n            res = self._ftlib.broadcast(data, src_rank)\n            if res == FTAllReduceStatus.SUCCESS:\n                return CollectiveCommunicatorStatus.SUCCEEDED, data\n            else:\n                return CollectiveCommunicatorStatus.FAILED, data\n        else:\n            logger.warning(_FTLIB_UNINSTALLED_DEFAULT_STATUS_MESSAGE)\n            return CollectiveCommunicatorStatus.SUCCEEDED, data\n\n    def barrier(self):\n        if self._ftlib is not None:\n            res = self._ftlib.barrier()\n            if res == FTAllReduceStatus.SUCCESS:\n                return CollectiveCommunicatorStatus.SUCCEEDED\n            else:\n                return CollectiveCommunicatorStatus.FAILED\n        else:\n            logger.warning(_FTLIB_UNINSTALLED_DEFAULT_STATUS_MESSAGE)\n            return CollectiveCommunicatorStatus.SUCCEEDED\n\n    def is_initialized(self):\n        \"\"\"This will be `False` under three occasions:\n           * New workers report joining in\n           * Collective-communication operations fail or time out\n           * Liveness probe fails for existing workers\n        \"\"\"\n        if self._ftlib is not None:\n            return self._ftlib.initialized\n        else:\n            return True\n\n    def _get_peer_set(self, svc_name):\n        if svc_name is None:\n            return None\n        my_ip = socket.gethostbyname(socket.gethostname())\n        temp_set = socket.getaddrinfo(svc_name, 0, proto=socket.IPPROTO_TCP)\n        peer_set = {peer[-1][0] for peer in temp_set if peer[-1][0] != my_ip}\n        return peer_set\n", "path": "elasticdl/python/collective_ops/communicator.py"}]}