def receive(sources, timeout=None):
    """Get all outstanding signals from sources.

    A source can be either (1) an object ID returned by the task (we want
    to receive signals from), or (2) an actor handle.

    When invoked by the same entity E (where E can be an actor, task or
    driver), for each source S in sources, this function returns all signals
    generated by S since the last receive() was invoked by E on S. If this is
    the first call on S, this function returns all past signals generated by S
    so far. Note that different actors, tasks or drivers that call receive()
    on the same source S will get independent copies of the signals generated
    by S.

    Args:
        sources: List of sources from which the caller waits for signals.
            A source is either an object ID returned by a task (in this case
            the object ID is used to identify that task), or an actor handle.
            If the user passes the IDs of multiple objects returned by the
            same task, this function returns a copy of the signals generated
            by that task for each object ID.
        timeout: Maximum time (in seconds) this function waits to get a signal
            from a source in sources. If None, the timeout is infinite.

    Returns:
        A list of pairs (S, sig), where S is a source in the sources argument,
            and sig is a signal generated by S since the last time receive()
            was called on S. Thus, for each S in sources, the return list can
            contain zero or multiple entries.
    """

    # If None, initialize the timeout to a huge value (i.e., over 30,000 years
    # in this case) to "approximate" infinity.
    if timeout is None:
        timeout = 10**12

    if timeout < 0:
        raise ValueError("The 'timeout' argument cannot be less than 0.")

    if not hasattr(ray.worker.global_worker, "signal_counters"):
        ray.worker.global_worker.signal_counters = defaultdict(lambda: b"0")

    signal_counters = ray.worker.global_worker.signal_counters

    # Map the ID of each source task to the source itself.
    task_id_to_sources = defaultdict(lambda: [])
    for s in sources:
        task_id_to_sources[_get_task_id(s).hex()].append(s)

    # Construct the redis query.
    query = "XREAD BLOCK "
    # Multiply by 1000x since timeout is in sec and redis expects ms.
    query += str(1000 * timeout)
    query += " STREAMS "
    query += " ".join([task_id for task_id in task_id_to_sources])
    query += " "
    query += " ".join([
        ray.utils.decode(signal_counters[ray.utils.hex_to_binary(task_id)])
        for task_id in task_id_to_sources
    ])

    answers = ray.worker.global_worker.redis_client.execute_command(query)
    if not answers:
        return []

    results = []
    # Decoding is a little bit involved. Iterate through all the answers:
    for i, answer in enumerate(answers):
        # Make sure the answer corresponds to a source, s, in sources.
        task_id = ray.utils.decode(answer[0])
        task_source_list = task_id_to_sources[task_id]
        # The list of results for source s is stored in answer[1]
        for r in answer[1]:
            for s in task_source_list:
                if r[1][1].decode("ascii") == ACTOR_DIED_STR:
                    results.append((s, ActorDiedSignal()))
                else:
                    # Now it gets tricky: r[0] is the redis internal sequence
                    # id
                    signal_counters[ray.utils.hex_to_binary(task_id)] = r[0]
                    # r[1] contains a list with elements (key, value), in our
                    # case we only have one key "signal" and the value is the
                    # signal.
                    signal = cloudpickle.loads(
                        ray.utils.hex_to_binary(r[1][1]))
                    results.append((s, signal))

    return results
def receive ( sources , timeout = None ) : if timeout is None : timeout = 10 ** 12 if timeout < 0 : raise ValueError ( "The 'timeout' argument cannot be less than 0." ) if not hasattr ( ray . worker . global_worker , "signal_counters" ) : ray . worker . global_worker . signal_counters = defaultdict ( lambda : b"0" ) signal_counters = ray . worker . global_worker . signal_counters task_id_to_sources = defaultdict ( lambda : [ ] ) for s in sources : task_id_to_sources [ _get_task_id ( s ) . hex ( ) ] . append ( s ) query = "XREAD BLOCK " query += str ( 1000 * timeout ) query += " STREAMS " query += " " . join ( [ task_id for task_id in task_id_to_sources ] ) query += " " query += " " . join ( [ ray . utils . decode ( signal_counters [ ray . utils . hex_to_binary ( task_id ) ] ) for task_id in task_id_to_sources ] ) answers = ray . worker . global_worker . redis_client . execute_command ( query ) if not answers : return [ ] results = [ ] for i , answer in enumerate ( answers ) : task_id = ray . utils . decode ( answer [ 0 ] ) task_source_list = task_id_to_sources [ task_id ] for r in answer [ 1 ] : for s in task_source_list : if r [ 1 ] [ 1 ] . decode ( "ascii" ) == ACTOR_DIED_STR : results . append ( ( s , ActorDiedSignal ( ) ) ) else : signal_counters [ ray . utils . hex_to_binary ( task_id ) ] = r [ 0 ] signal = cloudpickle . loads ( ray . utils . hex_to_binary ( r [ 1 ] [ 1 ] ) ) results . append ( ( s , signal ) ) return results
Get all outstanding signals from sources.

    A source can be either (1) an object ID returned by the task (we want
    to receive signals from), or (2) an actor handle.

    When invoked by the same entity E (where E can be an actor, task or
    driver), for each source S in sources, this function returns all signals
    generated by S since the last receive() was invoked by E on S. If this is
    the first call on S, this function returns all past signals generated by S
    so far. Note that different actors, tasks or drivers that call receive()
    on the same source S will get independent copies of the signals generated
    by S.

    Args:
        sources: List of sources from which the caller waits for signals.
            A source is either an object ID returned by a task (in this case
            the object ID is used to identify that task), or an actor handle.
            If the user passes the IDs of multiple objects returned by the
            same task, this function returns a copy of the signals generated
            by that task for each object ID.
        timeout: Maximum time (in seconds) this function waits to get a signal
            from a source in sources. If None, the timeout is infinite.

    Returns:
        A list of pairs (S, sig), where S is a source in the sources argument,
            and sig is a signal generated by S since the last time receive()
            was called on S. Thus, for each S in sources, the return list can
            contain zero or multiple entries.