We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Non-integer timeout for signal.receive causes malformed redis query
<!--
General questions should be asked on the mailing list ray-dev@googlegroups.com.
Questions about how to use Ray should be asked on
[StackOverflow](https://stackoverflow.com/questions/tagged/ray).

Before submitting an issue, please fill out the following form.
-->

### System information
- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: Ubuntu 18.04
- **Ray installed from (source or binary)**: source
- **Ray version**: 0.7.0.dev2
- **Python version**: 3.7.2

### Describe the problem
The timeout argument in `signal.receive()` does not work with non-integer values. For instance, calling `signal.receive(sources, timeout=0.01)` causes a `redis.exceptions.ResponseError: timeout is not an integer or out of range`.

This is because of L131 in `signal.py`, where the timeout is converted to ms and then converted to a string - redis expects an int, but if you pass a double to the method (like 0.1), the result of `str(1000 * timeout)` is  `100.0`. The correct string would have been `100`. A fix would be to change `str(1000 * timeout)` to `str(int(1000 * timeout))` and have checks to ensure timeout is not < 1000.

https://github.com/ray-project/ray/blob/d951eb740ffe22c385d75df62aa18da790706804/python/ray/experimental/signal.py#L129-L133

### Source code / logs
Traceback: 
```
Traceback (most recent call last):
  File "driver.py", line 21, in <module>
    signals = signal.receive(self.clients, timeout=0.01)
  File "/home/romilb/ray/python/ray/experimental/signal.py", line 141, in receive
    answers = ray.worker.global_worker.redis_client.execute_command(query)
  File "/home/romilb/anaconda3/lib/python3.7/site-packages/redis/client.py", line 668, in execute_command
    return self.parse_response(connection, command_name, **options)
  File "/home/romilb/anaconda3/lib/python3.7/site-packages/redis/client.py", line 680, in parse_response
    response = connection.read_response()
  File "/home/romilb/anaconda3/lib/python3.7/site-packages/redis/connection.py", line 629, in read_response
    raise response
redis.exceptions.ResponseError: timeout is not an integer or out of range
```
I also printed the corresponding redis query:

```
(pid=31330) XREAD BLOCK 10.0 STREAMS fc96f993802bb5a31b5868e69179fe76b62b7c3b 0
```


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `python/ray/experimental/signal.py`
Content:
```
1 from __future__ import absolute_import
2 from __future__ import division
3 from __future__ import print_function
4 
5 from collections import defaultdict
6 
7 import ray
8 import ray.cloudpickle as cloudpickle
9 
10 # This string should be identical to the name of the signal sent upon
11 # detecting that an actor died.
12 # This constant is also used in NodeManager::PublishActorStateTransition()
13 # in node_manager.cc
14 ACTOR_DIED_STR = "ACTOR_DIED_SIGNAL"
15 
16 
17 class Signal(object):
18     """Base class for Ray signals."""
19     pass
20 
21 
22 class ErrorSignal(Signal):
23     """Signal raised if an exception happens in a task or actor method."""
24 
25     def __init__(self, error):
26         self.error = error
27 
28 
29 class ActorDiedSignal(Signal):
30     """Signal raised if an exception happens in a task or actor method."""
31 
32     def __init__(self):
33         pass
34 
35 
36 def _get_task_id(source):
37     """Return the task id associated to the generic source of the signal.
38 
39     Args:
40         source: source of the signal, it can be either an object id returned
41             by a task, a task id, or an actor handle.
42 
43     Returns:
44         - If source is an object id, return id of task which creted object.
45         - If source is an actor handle, return id of actor's task creator.
46         - If source is a task id, return same task id.
47     """
48     if type(source) is ray.actor.ActorHandle:
49         return source._ray_actor_id
50     else:
51         if type(source) is ray.TaskID:
52             return source
53         else:
54             return ray._raylet.compute_task_id(source)
55 
56 
57 def send(signal):
58     """Send signal.
59 
60     The signal has a unique identifier that is computed from (1) the id
61     of the actor or task sending this signal (i.e., the actor or task calling
62     this function), and (2) an index that is incremented every time this
63     source sends a signal. This index starts from 1.
64 
65     Args:
66         signal: Signal to be sent.
67     """
68     if hasattr(ray.worker.global_worker, "actor_creation_task_id"):
69         source_key = ray.worker.global_worker.actor_id.hex()
70     else:
71         # No actors; this function must have been called from a task
72         source_key = ray.worker.global_worker.current_task_id.hex()
73 
74     encoded_signal = ray.utils.binary_to_hex(cloudpickle.dumps(signal))
75     ray.worker.global_worker.redis_client.execute_command(
76         "XADD " + source_key + " * signal " + encoded_signal)
77 
78 
79 def receive(sources, timeout=None):
80     """Get all outstanding signals from sources.
81 
82     A source can be either (1) an object ID returned by the task (we want
83     to receive signals from), or (2) an actor handle.
84 
85     When invoked by the same entity E (where E can be an actor, task or
86     driver), for each source S in sources, this function returns all signals
87     generated by S since the last receive() was invoked by E on S. If this is
88     the first call on S, this function returns all past signals generated by S
89     so far. Note that different actors, tasks or drivers that call receive()
90     on the same source S will get independent copies of the signals generated
91     by S.
92 
93     Args:
94         sources: List of sources from which the caller waits for signals.
95             A source is either an object ID returned by a task (in this case
96             the object ID is used to identify that task), or an actor handle.
97             If the user passes the IDs of multiple objects returned by the
98             same task, this function returns a copy of the signals generated
99             by that task for each object ID.
100         timeout: Maximum time (in seconds) this function waits to get a signal
101             from a source in sources. If None, the timeout is infinite.
102 
103     Returns:
104         A list of pairs (S, sig), where S is a source in the sources argument,
105             and sig is a signal generated by S since the last time receive()
106             was called on S. Thus, for each S in sources, the return list can
107             contain zero or multiple entries.
108     """
109 
110     # If None, initialize the timeout to a huge value (i.e., over 30,000 years
111     # in this case) to "approximate" infinity.
112     if timeout is None:
113         timeout = 10**12
114 
115     if timeout < 0:
116         raise ValueError("The 'timeout' argument cannot be less than 0.")
117 
118     if not hasattr(ray.worker.global_worker, "signal_counters"):
119         ray.worker.global_worker.signal_counters = defaultdict(lambda: b"0")
120 
121     signal_counters = ray.worker.global_worker.signal_counters
122 
123     # Map the ID of each source task to the source itself.
124     task_id_to_sources = defaultdict(lambda: [])
125     for s in sources:
126         task_id_to_sources[_get_task_id(s).hex()].append(s)
127 
128     # Construct the redis query.
129     query = "XREAD BLOCK "
130     # Multiply by 1000x since timeout is in sec and redis expects ms.
131     query += str(1000 * timeout)
132     query += " STREAMS "
133     query += " ".join([task_id for task_id in task_id_to_sources])
134     query += " "
135     query += " ".join([
136         ray.utils.decode(signal_counters[ray.utils.hex_to_binary(task_id)])
137         for task_id in task_id_to_sources
138     ])
139 
140     answers = ray.worker.global_worker.redis_client.execute_command(query)
141     if not answers:
142         return []
143 
144     results = []
145     # Decoding is a little bit involved. Iterate through all the answers:
146     for i, answer in enumerate(answers):
147         # Make sure the answer corresponds to a source, s, in sources.
148         task_id = ray.utils.decode(answer[0])
149         task_source_list = task_id_to_sources[task_id]
150         # The list of results for source s is stored in answer[1]
151         for r in answer[1]:
152             for s in task_source_list:
153                 if r[1][1].decode("ascii") == ACTOR_DIED_STR:
154                     results.append((s, ActorDiedSignal()))
155                 else:
156                     # Now it gets tricky: r[0] is the redis internal sequence
157                     # id
158                     signal_counters[ray.utils.hex_to_binary(task_id)] = r[0]
159                     # r[1] contains a list with elements (key, value), in our
160                     # case we only have one key "signal" and the value is the
161                     # signal.
162                     signal = cloudpickle.loads(
163                         ray.utils.hex_to_binary(r[1][1]))
164                     results.append((s, signal))
165 
166     return results
167 
168 
169 def forget(sources):
170     """Ignore all previous signals associated with each source S in sources.
171 
172     The index of the next expected signal from S is set to the index of
173     the last signal that S sent plus 1. This means that the next receive()
174     on S will only get the signals generated after this function was invoked.
175 
176     Args:
177         sources: list of sources whose past signals are forgotten.
178     """
179     # Just read all signals sent by all sources so far.
180     # This will results in ignoring these signals.
181     receive(sources, timeout=0)
182 
183 
184 def reset():
185     """
186     Reset the worker state associated with any signals that this worker
187     has received so far.
188 
189     If the worker calls receive() on a source next, it will get all the
190     signals generated by that source starting with index = 1.
191     """
192     if hasattr(ray.worker.global_worker, "signal_counters"):
193         ray.worker.global_worker.signal_counters = defaultdict(lambda: b"0")
194
```
--- 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()

```

diff --git a/python/ray/experimental/signal.py b/python/ray/experimental/signal.py
--- a/python/ray/experimental/signal.py
+++ b/python/ray/experimental/signal.py
@@ -2,6 +2,8 @@
 from __future__ import division
 from __future__ import print_function
 
+import logging
+
 from collections import defaultdict
 
 import ray
@@ -13,6 +15,8 @@
 # in node_manager.cc
 ACTOR_DIED_STR = "ACTOR_DIED_SIGNAL"
 
+logger = logging.getLogger(__name__)
+
 
 class Signal(object):
     """Base class for Ray signals."""
@@ -125,10 +129,16 @@
     for s in sources:
         task_id_to_sources[_get_task_id(s).hex()].append(s)
 
+    if timeout < 1e-3:
+        logger.warning("Timeout too small. Using 1ms minimum")
+        timeout = 1e-3
+
+    timeout_ms = int(1000 * timeout)
+
     # Construct the redis query.
     query = "XREAD BLOCK "
-    # Multiply by 1000x since timeout is in sec and redis expects ms.
-    query += str(1000 * timeout)
+    # redis expects ms.
+    query += str(timeout_ms)
     query += " STREAMS "
     query += " ".join([task_id for task_id in task_id_to_sources])
     query += " "

{"golden_diff": "diff --git a/python/ray/experimental/signal.py b/python/ray/experimental/signal.py\n--- a/python/ray/experimental/signal.py\n+++ b/python/ray/experimental/signal.py\n@@ -2,6 +2,8 @@\n from __future__ import division\n from __future__ import print_function\n \n+import logging\n+\n from collections import defaultdict\n \n import ray\n@@ -13,6 +15,8 @@\n # in node_manager.cc\n ACTOR_DIED_STR = \"ACTOR_DIED_SIGNAL\"\n \n+logger = logging.getLogger(__name__)\n+\n \n class Signal(object):\n     \"\"\"Base class for Ray signals.\"\"\"\n@@ -125,10 +129,16 @@\n     for s in sources:\n         task_id_to_sources[_get_task_id(s).hex()].append(s)\n \n+    if timeout < 1e-3:\n+        logger.warning(\"Timeout too small. Using 1ms minimum\")\n+        timeout = 1e-3\n+\n+    timeout_ms = int(1000 * timeout)\n+\n     # Construct the redis query.\n     query = \"XREAD BLOCK \"\n-    # Multiply by 1000x since timeout is in sec and redis expects ms.\n-    query += str(1000 * timeout)\n+    # redis expects ms.\n+    query += str(timeout_ms)\n     query += \" STREAMS \"\n     query += \" \".join([task_id for task_id in task_id_to_sources])\n     query += \" \"\n", "issue": "Non-integer timeout for signal.receive causes malformed redis query\n<!--\r\nGeneral questions should be asked on the mailing list ray-dev@googlegroups.com.\r\nQuestions about how to use Ray should be asked on\r\n[StackOverflow](https://stackoverflow.com/questions/tagged/ray).\r\n\r\nBefore submitting an issue, please fill out the following form.\r\n-->\r\n\r\n### System information\r\n- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: Ubuntu 18.04\r\n- **Ray installed from (source or binary)**: source\r\n- **Ray version**: 0.7.0.dev2\r\n- **Python version**: 3.7.2\r\n\r\n### Describe the problem\r\nThe timeout argument in `signal.receive()` does not work with non-integer values. For instance, calling `signal.receive(sources, timeout=0.01)` causes a `redis.exceptions.ResponseError: timeout is not an integer or out of range`.\r\n\r\nThis is because of L131 in `signal.py`, where the timeout is converted to ms and then converted to a string - redis expects an int, but if you pass a double to the method (like 0.1), the result of `str(1000 * timeout)` is  `100.0`. The correct string would have been `100`. A fix would be to change `str(1000 * timeout)` to `str(int(1000 * timeout))` and have checks to ensure timeout is not < 1000.\r\n\r\nhttps://github.com/ray-project/ray/blob/d951eb740ffe22c385d75df62aa18da790706804/python/ray/experimental/signal.py#L129-L133\r\n\r\n### Source code / logs\r\nTraceback: \r\n```\r\nTraceback (most recent call last):\r\n  File \"driver.py\", line 21, in <module>\r\n    signals = signal.receive(self.clients, timeout=0.01)\r\n  File \"/home/romilb/ray/python/ray/experimental/signal.py\", line 141, in receive\r\n    answers = ray.worker.global_worker.redis_client.execute_command(query)\r\n  File \"/home/romilb/anaconda3/lib/python3.7/site-packages/redis/client.py\", line 668, in execute_command\r\n    return self.parse_response(connection, command_name, **options)\r\n  File \"/home/romilb/anaconda3/lib/python3.7/site-packages/redis/client.py\", line 680, in parse_response\r\n    response = connection.read_response()\r\n  File \"/home/romilb/anaconda3/lib/python3.7/site-packages/redis/connection.py\", line 629, in read_response\r\n    raise response\r\nredis.exceptions.ResponseError: timeout is not an integer or out of range\r\n```\r\nI also printed the corresponding redis query:\r\n\r\n```\r\n(pid=31330) XREAD BLOCK 10.0 STREAMS fc96f993802bb5a31b5868e69179fe76b62b7c3b 0\r\n```\r\n\n", "before_files": [{"content": "from __future__ import absolute_import\nfrom __future__ import division\nfrom __future__ import print_function\n\nfrom collections import defaultdict\n\nimport ray\nimport ray.cloudpickle as cloudpickle\n\n# This string should be identical to the name of the signal sent upon\n# detecting that an actor died.\n# This constant is also used in NodeManager::PublishActorStateTransition()\n# in node_manager.cc\nACTOR_DIED_STR = \"ACTOR_DIED_SIGNAL\"\n\n\nclass Signal(object):\n    \"\"\"Base class for Ray signals.\"\"\"\n    pass\n\n\nclass ErrorSignal(Signal):\n    \"\"\"Signal raised if an exception happens in a task or actor method.\"\"\"\n\n    def __init__(self, error):\n        self.error = error\n\n\nclass ActorDiedSignal(Signal):\n    \"\"\"Signal raised if an exception happens in a task or actor method.\"\"\"\n\n    def __init__(self):\n        pass\n\n\ndef _get_task_id(source):\n    \"\"\"Return the task id associated to the generic source of the signal.\n\n    Args:\n        source: source of the signal, it can be either an object id returned\n            by a task, a task id, or an actor handle.\n\n    Returns:\n        - If source is an object id, return id of task which creted object.\n        - If source is an actor handle, return id of actor's task creator.\n        - If source is a task id, return same task id.\n    \"\"\"\n    if type(source) is ray.actor.ActorHandle:\n        return source._ray_actor_id\n    else:\n        if type(source) is ray.TaskID:\n            return source\n        else:\n            return ray._raylet.compute_task_id(source)\n\n\ndef send(signal):\n    \"\"\"Send signal.\n\n    The signal has a unique identifier that is computed from (1) the id\n    of the actor or task sending this signal (i.e., the actor or task calling\n    this function), and (2) an index that is incremented every time this\n    source sends a signal. This index starts from 1.\n\n    Args:\n        signal: Signal to be sent.\n    \"\"\"\n    if hasattr(ray.worker.global_worker, \"actor_creation_task_id\"):\n        source_key = ray.worker.global_worker.actor_id.hex()\n    else:\n        # No actors; this function must have been called from a task\n        source_key = ray.worker.global_worker.current_task_id.hex()\n\n    encoded_signal = ray.utils.binary_to_hex(cloudpickle.dumps(signal))\n    ray.worker.global_worker.redis_client.execute_command(\n        \"XADD \" + source_key + \" * signal \" + encoded_signal)\n\n\ndef receive(sources, timeout=None):\n    \"\"\"Get all outstanding signals from sources.\n\n    A source can be either (1) an object ID returned by the task (we want\n    to receive signals from), or (2) an actor handle.\n\n    When invoked by the same entity E (where E can be an actor, task or\n    driver), for each source S in sources, this function returns all signals\n    generated by S since the last receive() was invoked by E on S. If this is\n    the first call on S, this function returns all past signals generated by S\n    so far. Note that different actors, tasks or drivers that call receive()\n    on the same source S will get independent copies of the signals generated\n    by S.\n\n    Args:\n        sources: List of sources from which the caller waits for signals.\n            A source is either an object ID returned by a task (in this case\n            the object ID is used to identify that task), or an actor handle.\n            If the user passes the IDs of multiple objects returned by the\n            same task, this function returns a copy of the signals generated\n            by that task for each object ID.\n        timeout: Maximum time (in seconds) this function waits to get a signal\n            from a source in sources. If None, the timeout is infinite.\n\n    Returns:\n        A list of pairs (S, sig), where S is a source in the sources argument,\n            and sig is a signal generated by S since the last time receive()\n            was called on S. Thus, for each S in sources, the return list can\n            contain zero or multiple entries.\n    \"\"\"\n\n    # If None, initialize the timeout to a huge value (i.e., over 30,000 years\n    # in this case) to \"approximate\" infinity.\n    if timeout is None:\n        timeout = 10**12\n\n    if timeout < 0:\n        raise ValueError(\"The 'timeout' argument cannot be less than 0.\")\n\n    if not hasattr(ray.worker.global_worker, \"signal_counters\"):\n        ray.worker.global_worker.signal_counters = defaultdict(lambda: b\"0\")\n\n    signal_counters = ray.worker.global_worker.signal_counters\n\n    # Map the ID of each source task to the source itself.\n    task_id_to_sources = defaultdict(lambda: [])\n    for s in sources:\n        task_id_to_sources[_get_task_id(s).hex()].append(s)\n\n    # Construct the redis query.\n    query = \"XREAD BLOCK \"\n    # Multiply by 1000x since timeout is in sec and redis expects ms.\n    query += str(1000 * timeout)\n    query += \" STREAMS \"\n    query += \" \".join([task_id for task_id in task_id_to_sources])\n    query += \" \"\n    query += \" \".join([\n        ray.utils.decode(signal_counters[ray.utils.hex_to_binary(task_id)])\n        for task_id in task_id_to_sources\n    ])\n\n    answers = ray.worker.global_worker.redis_client.execute_command(query)\n    if not answers:\n        return []\n\n    results = []\n    # Decoding is a little bit involved. Iterate through all the answers:\n    for i, answer in enumerate(answers):\n        # Make sure the answer corresponds to a source, s, in sources.\n        task_id = ray.utils.decode(answer[0])\n        task_source_list = task_id_to_sources[task_id]\n        # The list of results for source s is stored in answer[1]\n        for r in answer[1]:\n            for s in task_source_list:\n                if r[1][1].decode(\"ascii\") == ACTOR_DIED_STR:\n                    results.append((s, ActorDiedSignal()))\n                else:\n                    # Now it gets tricky: r[0] is the redis internal sequence\n                    # id\n                    signal_counters[ray.utils.hex_to_binary(task_id)] = r[0]\n                    # r[1] contains a list with elements (key, value), in our\n                    # case we only have one key \"signal\" and the value is the\n                    # signal.\n                    signal = cloudpickle.loads(\n                        ray.utils.hex_to_binary(r[1][1]))\n                    results.append((s, signal))\n\n    return results\n\n\ndef forget(sources):\n    \"\"\"Ignore all previous signals associated with each source S in sources.\n\n    The index of the next expected signal from S is set to the index of\n    the last signal that S sent plus 1. This means that the next receive()\n    on S will only get the signals generated after this function was invoked.\n\n    Args:\n        sources: list of sources whose past signals are forgotten.\n    \"\"\"\n    # Just read all signals sent by all sources so far.\n    # This will results in ignoring these signals.\n    receive(sources, timeout=0)\n\n\ndef reset():\n    \"\"\"\n    Reset the worker state associated with any signals that this worker\n    has received so far.\n\n    If the worker calls receive() on a source next, it will get all the\n    signals generated by that source starting with index = 1.\n    \"\"\"\n    if hasattr(ray.worker.global_worker, \"signal_counters\"):\n        ray.worker.global_worker.signal_counters = defaultdict(lambda: b\"0\")\n", "path": "python/ray/experimental/signal.py"}], "after_files": [{"content": "from __future__ import absolute_import\nfrom __future__ import division\nfrom __future__ import print_function\n\nimport logging\n\nfrom collections import defaultdict\n\nimport ray\nimport ray.cloudpickle as cloudpickle\n\n# This string should be identical to the name of the signal sent upon\n# detecting that an actor died.\n# This constant is also used in NodeManager::PublishActorStateTransition()\n# in node_manager.cc\nACTOR_DIED_STR = \"ACTOR_DIED_SIGNAL\"\n\nlogger = logging.getLogger(__name__)\n\n\nclass Signal(object):\n    \"\"\"Base class for Ray signals.\"\"\"\n    pass\n\n\nclass ErrorSignal(Signal):\n    \"\"\"Signal raised if an exception happens in a task or actor method.\"\"\"\n\n    def __init__(self, error):\n        self.error = error\n\n\nclass ActorDiedSignal(Signal):\n    \"\"\"Signal raised if an exception happens in a task or actor method.\"\"\"\n\n    def __init__(self):\n        pass\n\n\ndef _get_task_id(source):\n    \"\"\"Return the task id associated to the generic source of the signal.\n\n    Args:\n        source: source of the signal, it can be either an object id returned\n            by a task, a task id, or an actor handle.\n\n    Returns:\n        - If source is an object id, return id of task which creted object.\n        - If source is an actor handle, return id of actor's task creator.\n        - If source is a task id, return same task id.\n    \"\"\"\n    if type(source) is ray.actor.ActorHandle:\n        return source._ray_actor_id\n    else:\n        if type(source) is ray.TaskID:\n            return source\n        else:\n            return ray._raylet.compute_task_id(source)\n\n\ndef send(signal):\n    \"\"\"Send signal.\n\n    The signal has a unique identifier that is computed from (1) the id\n    of the actor or task sending this signal (i.e., the actor or task calling\n    this function), and (2) an index that is incremented every time this\n    source sends a signal. This index starts from 1.\n\n    Args:\n        signal: Signal to be sent.\n    \"\"\"\n    if hasattr(ray.worker.global_worker, \"actor_creation_task_id\"):\n        source_key = ray.worker.global_worker.actor_id.hex()\n    else:\n        # No actors; this function must have been called from a task\n        source_key = ray.worker.global_worker.current_task_id.hex()\n\n    encoded_signal = ray.utils.binary_to_hex(cloudpickle.dumps(signal))\n    ray.worker.global_worker.redis_client.execute_command(\n        \"XADD \" + source_key + \" * signal \" + encoded_signal)\n\n\ndef receive(sources, timeout=None):\n    \"\"\"Get all outstanding signals from sources.\n\n    A source can be either (1) an object ID returned by the task (we want\n    to receive signals from), or (2) an actor handle.\n\n    When invoked by the same entity E (where E can be an actor, task or\n    driver), for each source S in sources, this function returns all signals\n    generated by S since the last receive() was invoked by E on S. If this is\n    the first call on S, this function returns all past signals generated by S\n    so far. Note that different actors, tasks or drivers that call receive()\n    on the same source S will get independent copies of the signals generated\n    by S.\n\n    Args:\n        sources: List of sources from which the caller waits for signals.\n            A source is either an object ID returned by a task (in this case\n            the object ID is used to identify that task), or an actor handle.\n            If the user passes the IDs of multiple objects returned by the\n            same task, this function returns a copy of the signals generated\n            by that task for each object ID.\n        timeout: Maximum time (in seconds) this function waits to get a signal\n            from a source in sources. If None, the timeout is infinite.\n\n    Returns:\n        A list of pairs (S, sig), where S is a source in the sources argument,\n            and sig is a signal generated by S since the last time receive()\n            was called on S. Thus, for each S in sources, the return list can\n            contain zero or multiple entries.\n    \"\"\"\n\n    # If None, initialize the timeout to a huge value (i.e., over 30,000 years\n    # in this case) to \"approximate\" infinity.\n    if timeout is None:\n        timeout = 10**12\n\n    if timeout < 0:\n        raise ValueError(\"The 'timeout' argument cannot be less than 0.\")\n\n    if not hasattr(ray.worker.global_worker, \"signal_counters\"):\n        ray.worker.global_worker.signal_counters = defaultdict(lambda: b\"0\")\n\n    signal_counters = ray.worker.global_worker.signal_counters\n\n    # Map the ID of each source task to the source itself.\n    task_id_to_sources = defaultdict(lambda: [])\n    for s in sources:\n        task_id_to_sources[_get_task_id(s).hex()].append(s)\n\n    if timeout < 1e-3:\n        logger.warning(\"Timeout too small. Using 1ms minimum\")\n        timeout = 1e-3\n\n    timeout_ms = int(1000 * timeout)\n\n    # Construct the redis query.\n    query = \"XREAD BLOCK \"\n    # redis expects ms.\n    query += str(timeout_ms)\n    query += \" STREAMS \"\n    query += \" \".join([task_id for task_id in task_id_to_sources])\n    query += \" \"\n    query += \" \".join([\n        ray.utils.decode(signal_counters[ray.utils.hex_to_binary(task_id)])\n        for task_id in task_id_to_sources\n    ])\n\n    answers = ray.worker.global_worker.redis_client.execute_command(query)\n    if not answers:\n        return []\n\n    results = []\n    # Decoding is a little bit involved. Iterate through all the answers:\n    for i, answer in enumerate(answers):\n        # Make sure the answer corresponds to a source, s, in sources.\n        task_id = ray.utils.decode(answer[0])\n        task_source_list = task_id_to_sources[task_id]\n        # The list of results for source s is stored in answer[1]\n        for r in answer[1]:\n            for s in task_source_list:\n                if r[1][1].decode(\"ascii\") == ACTOR_DIED_STR:\n                    results.append((s, ActorDiedSignal()))\n                else:\n                    # Now it gets tricky: r[0] is the redis internal sequence\n                    # id\n                    signal_counters[ray.utils.hex_to_binary(task_id)] = r[0]\n                    # r[1] contains a list with elements (key, value), in our\n                    # case we only have one key \"signal\" and the value is the\n                    # signal.\n                    signal = cloudpickle.loads(\n                        ray.utils.hex_to_binary(r[1][1]))\n                    results.append((s, signal))\n\n    return results\n\n\ndef forget(sources):\n    \"\"\"Ignore all previous signals associated with each source S in sources.\n\n    The index of the next expected signal from S is set to the index of\n    the last signal that S sent plus 1. This means that the next receive()\n    on S will only get the signals generated after this function was invoked.\n\n    Args:\n        sources: list of sources whose past signals are forgotten.\n    \"\"\"\n    # Just read all signals sent by all sources so far.\n    # This will results in ignoring these signals.\n    receive(sources, timeout=0)\n\n\ndef reset():\n    \"\"\"\n    Reset the worker state associated with any signals that this worker\n    has received so far.\n\n    If the worker calls receive() on a source next, it will get all the\n    signals generated by that source starting with index = 1.\n    \"\"\"\n    if hasattr(ray.worker.global_worker, \"signal_counters\"):\n        ray.worker.global_worker.signal_counters = defaultdict(lambda: b\"0\")\n", "path": "python/ray/experimental/signal.py"}]}