prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  Worker hangs when connects to PS.
  The worker will wait the grpc channel of PS is ready using `channel_ready_future`. We should set timeout and retry.
  ![image](https://user-images.githubusercontent.com/18071380/76311338-b4f58200-630b-11ea-927f-d7e87206f015.png)
  
  
  --- 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/worker/main.py`
  Content:
  ```
  1 import grpc
  2 
  3 from elasticdl.python.common import log_utils
  4 from elasticdl.python.common.args import parse_worker_args
  5 from elasticdl.python.common.grpc_utils import build_channel
  6 from elasticdl.python.worker.worker import Worker
  7 
  8 
  9 def main():
  10     args = parse_worker_args()
  11     logger = log_utils.get_logger(__name__)
  12     logger.info("Starting worker %d", args.worker_id)
  13     if args.master_addr is None:
  14         raise ValueError("master_addr is missing for worker")
  15 
  16     master_channel = build_channel(args.master_addr)
  17 
  18     ps_channels = []
  19     if args.ps_addrs:
  20         ps_addrs = args.ps_addrs.split(",")
  21 
  22         for addr in ps_addrs:
  23             # addr is in the form as "ps-pod-name.namespace.svc:port"
  24             channel = build_channel(addr)
  25 
  26             # Wait the channel is ready by a Future object.
  27             grpc.channel_ready_future(channel).result()
  28             logger.info(
  29                 "grpc channel %s to connect pod %s is ready"
  30                 % (addr, addr.split(".")[0])
  31             )
  32             ps_channels.append(channel)
  33 
  34     worker = Worker(
  35         args,
  36         channel=master_channel,
  37         ps_channels=ps_channels,
  38         set_parallelism=True,
  39     )
  40     worker.run()
  41 
  42 
  43 if __name__ == "__main__":
  44     main()
  45
  ```
  --- 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/worker/main.py b/elasticdl/python/worker/main.py
  --- a/elasticdl/python/worker/main.py
  +++ b/elasticdl/python/worker/main.py
  @@ -5,6 +5,9 @@
   from elasticdl.python.common.grpc_utils import build_channel
   from elasticdl.python.worker.worker import Worker
   
  +CONNECT_PS_MAX_RETRIES = 3
  +CONNECT_PS_TIMEOUT = 60
  +
   
   def main():
       args = parse_worker_args()
  @@ -23,13 +26,29 @@
               # addr is in the form as "ps-pod-name.namespace.svc:port"
               channel = build_channel(addr)
   
  -            # Wait the channel is ready by a Future object.
  -            grpc.channel_ready_future(channel).result()
  -            logger.info(
  -                "grpc channel %s to connect pod %s is ready"
  -                % (addr, addr.split(".")[0])
  -            )
  -            ps_channels.append(channel)
  +            succeeded = False
  +            for i in range(CONNECT_PS_MAX_RETRIES):
  +                try:
  +                    grpc.channel_ready_future(channel).result(
  +                        timeout=CONNECT_PS_TIMEOUT
  +                    )
  +                    logger.info(
  +                        "grpc channel %s to connect pod %s is ready"
  +                        % (addr, addr.split(".")[0])
  +                    )
  +                    ps_channels.append(channel)
  +                    succeeded = True
  +                    break
  +                except grpc.FutureTimeoutError:
  +                    logger.warning(
  +                        "Failed to connect pod %s with %d retry"
  +                        % (addr.split(".")[0], i)
  +                    )
  +            if not succeeded:
  +                raise TimeoutError(
  +                    "Time out to connect pod %s with 3 retries"
  +                    % addr.split(".")[0]
  +                )
   
       worker = Worker(
           args,
  
verification_info: {"golden_diff": "diff --git a/elasticdl/python/worker/main.py b/elasticdl/python/worker/main.py\n--- a/elasticdl/python/worker/main.py\n+++ b/elasticdl/python/worker/main.py\n@@ -5,6 +5,9 @@\n from elasticdl.python.common.grpc_utils import build_channel\n from elasticdl.python.worker.worker import Worker\n \n+CONNECT_PS_MAX_RETRIES = 3\n+CONNECT_PS_TIMEOUT = 60\n+\n \n def main():\n     args = parse_worker_args()\n@@ -23,13 +26,29 @@\n             # addr is in the form as \"ps-pod-name.namespace.svc:port\"\n             channel = build_channel(addr)\n \n-            # Wait the channel is ready by a Future object.\n-            grpc.channel_ready_future(channel).result()\n-            logger.info(\n-                \"grpc channel %s to connect pod %s is ready\"\n-                % (addr, addr.split(\".\")[0])\n-            )\n-            ps_channels.append(channel)\n+            succeeded = False\n+            for i in range(CONNECT_PS_MAX_RETRIES):\n+                try:\n+                    grpc.channel_ready_future(channel).result(\n+                        timeout=CONNECT_PS_TIMEOUT\n+                    )\n+                    logger.info(\n+                        \"grpc channel %s to connect pod %s is ready\"\n+                        % (addr, addr.split(\".\")[0])\n+                    )\n+                    ps_channels.append(channel)\n+                    succeeded = True\n+                    break\n+                except grpc.FutureTimeoutError:\n+                    logger.warning(\n+                        \"Failed to connect pod %s with %d retry\"\n+                        % (addr.split(\".\")[0], i)\n+                    )\n+            if not succeeded:\n+                raise TimeoutError(\n+                    \"Time out to connect pod %s with 3 retries\"\n+                    % addr.split(\".\")[0]\n+                )\n \n     worker = Worker(\n         args,\n", "issue": "Worker hangs when connects to PS.\nThe worker will wait the grpc channel of PS is ready using `channel_ready_future`. We should set timeout and retry.\r\n![image](https://user-images.githubusercontent.com/18071380/76311338-b4f58200-630b-11ea-927f-d7e87206f015.png)\r\n\n", "before_files": [{"content": "import grpc\n\nfrom elasticdl.python.common import log_utils\nfrom elasticdl.python.common.args import parse_worker_args\nfrom elasticdl.python.common.grpc_utils import build_channel\nfrom elasticdl.python.worker.worker import Worker\n\n\ndef main():\n    args = parse_worker_args()\n    logger = log_utils.get_logger(__name__)\n    logger.info(\"Starting worker %d\", args.worker_id)\n    if args.master_addr is None:\n        raise ValueError(\"master_addr is missing for worker\")\n\n    master_channel = build_channel(args.master_addr)\n\n    ps_channels = []\n    if args.ps_addrs:\n        ps_addrs = args.ps_addrs.split(\",\")\n\n        for addr in ps_addrs:\n            # addr is in the form as \"ps-pod-name.namespace.svc:port\"\n            channel = build_channel(addr)\n\n            # Wait the channel is ready by a Future object.\n            grpc.channel_ready_future(channel).result()\n            logger.info(\n                \"grpc channel %s to connect pod %s is ready\"\n                % (addr, addr.split(\".\")[0])\n            )\n            ps_channels.append(channel)\n\n    worker = Worker(\n        args,\n        channel=master_channel,\n        ps_channels=ps_channels,\n        set_parallelism=True,\n    )\n    worker.run()\n\n\nif __name__ == \"__main__\":\n    main()\n", "path": "elasticdl/python/worker/main.py"}], "after_files": [{"content": "import grpc\n\nfrom elasticdl.python.common import log_utils\nfrom elasticdl.python.common.args import parse_worker_args\nfrom elasticdl.python.common.grpc_utils import build_channel\nfrom elasticdl.python.worker.worker import Worker\n\nCONNECT_PS_MAX_RETRIES = 3\nCONNECT_PS_TIMEOUT = 60\n\n\ndef main():\n    args = parse_worker_args()\n    logger = log_utils.get_logger(__name__)\n    logger.info(\"Starting worker %d\", args.worker_id)\n    if args.master_addr is None:\n        raise ValueError(\"master_addr is missing for worker\")\n\n    master_channel = build_channel(args.master_addr)\n\n    ps_channels = []\n    if args.ps_addrs:\n        ps_addrs = args.ps_addrs.split(\",\")\n\n        for addr in ps_addrs:\n            # addr is in the form as \"ps-pod-name.namespace.svc:port\"\n            channel = build_channel(addr)\n\n            succeeded = False\n            for i in range(CONNECT_PS_MAX_RETRIES):\n                try:\n                    grpc.channel_ready_future(channel).result(\n                        timeout=CONNECT_PS_TIMEOUT\n                    )\n                    logger.info(\n                        \"grpc channel %s to connect pod %s is ready\"\n                        % (addr, addr.split(\".\")[0])\n                    )\n                    ps_channels.append(channel)\n                    succeeded = True\n                    break\n                except grpc.FutureTimeoutError:\n                    logger.warning(\n                        \"Failed to connect pod %s with %d retry\"\n                        % (addr.split(\".\")[0], i)\n                    )\n            if not succeeded:\n                raise TimeoutError(\n                    \"Time out to connect pod %s with 3 retries\"\n                    % addr.split(\".\")[0]\n                )\n\n    worker = Worker(\n        args,\n        channel=master_channel,\n        ps_channels=ps_channels,\n        set_parallelism=True,\n    )\n    worker.run()\n\n\nif __name__ == \"__main__\":\n    main()\n", "path": "elasticdl/python/worker/main.py"}]}