rllib/agents/dqn/apex.py
def sample_from_replay_buffer_place_on_learner_queue_non_blocking(
        self, num_samples_collected: Dict[ActorHandle, int]
    ) -> None:
        """Get samples from the replay buffer and place them on the learner queue.

        Args:
            num_samples_collected: A mapping from ActorHandle (RolloutWorker) to
                number of samples returned by the remote worker. This is used to
                implement training intensity which is the concept of triggering a
                certain amount of training based on the number of samples that have
                been collected since the last time that training was triggered.

        """

        def wait_on_replay_actors(timeout: float) -> None:
            """Wait for the replay actors to finish sampling for timeout seconds.
            If the timeout is None, then block on the actors indefinitely.
            """
            replay_samples_ready: Dict[ActorHandle, T] = wait_asynchronous_requests(
                remote_requests_in_flight=self.remote_replay_requests_in_flight,
                ray_wait_timeout_s=timeout,
            )

            for replay_actor, sample_batches in replay_samples_ready.items():
                for sample_batch in sample_batches:
                    self.replay_sample_batches.append((replay_actor, sample_batch))

        num_samples_collected = sum(num_samples_collected.values())
        self.curr_num_samples_collected += num_samples_collected
        if self.curr_num_samples_collected >= self.config["train_batch_size"]:
            wait_on_replay_actors(None)
            training_intensity = int(self.config["training_intensity"] or 1)
            num_requests_to_launch = (
                self.curr_num_samples_collected / self.config["train_batch_size"]
            ) * training_intensity
            num_requests_to_launch = max(1, round(num_requests_to_launch))
            self.curr_num_samples_collected = 0
            for _ in range(num_requests_to_launch):
                rand_actor = random.choice(self.replay_actors)
                replay_samples_ready: Dict[
                    ActorHandle, T
                ] = asynchronous_parallel_requests(
                    remote_requests_in_flight=self.remote_replay_requests_in_flight,
                    actors=[rand_actor],
                    ray_wait_timeout_s=0.1,
                    max_remote_requests_in_flight_per_actor=num_requests_to_launch,
                    remote_fn=lambda actor: actor.replay(),
                )
            for replay_actor, sample_batches in replay_samples_ready.items():
                for sample_batch in sample_batches:
                    self.replay_sample_batches.append((replay_actor, sample_batch))

        wait_on_replay_actors(0.1)

        # add the sample batches to the learner queue
        while self.replay_sample_batches:
            try:
                item = self.replay_sample_batches[0]
                # the replay buffer returns none if it has not been filled to
                # the minimum threshold yet.
                if item:
                    self.learner_thread.inqueue.put(
                        self.replay_sample_batches[0], timeout=0.001
                    )
                    self.replay_sample_batches.pop(0)
            except queue.Full:
                break
Get samples from the replay buffer and place them on the learner queue.

        Args:
            num_samples_collected: A mapping from ActorHandle (RolloutWorker) to
                number of samples returned by the remote worker. This is used to
                implement training intensity which is the concept of triggering a
                certain amount of training based on the number of samples that have
                been collected since the last time that training was triggered.

        
Get samples from the replay buffer and place them on the learner queue.
Get samples from the replay buffer and place them on the learner queue .
def sample_from_replay_buffer_place_on_learner_queue_non_blocking(
        self, num_samples_collected: Dict[ActorHandle, int]
    ) -> None:
        def wait_on_replay_actors(timeout: float) -> None:
            replay_samples_ready: Dict[ActorHandle, T] = wait_asynchronous_requests(
                remote_requests_in_flight=self.remote_replay_requests_in_flight,
                ray_wait_timeout_s=timeout,
            )
            for replay_actor, sample_batches in replay_samples_ready.items():
                for sample_batch in sample_batches:
                    self.replay_sample_batches.append((replay_actor, sample_batch))
        num_samples_collected = sum(num_samples_collected.values())
        self.curr_num_samples_collected += num_samples_collected
        if self.curr_num_samples_collected >= self.config["train_batch_size"]:
            wait_on_replay_actors(None)
            training_intensity = int(self.config["training_intensity"] or 1)
            num_requests_to_launch = (
                self.curr_num_samples_collected / self.config["train_batch_size"]
            ) * training_intensity
            num_requests_to_launch = max(1, round(num_requests_to_launch))
            self.curr_num_samples_collected = 0
            for _ in range(num_requests_to_launch):
                rand_actor = random.choice(self.replay_actors)
                replay_samples_ready: Dict[
                    ActorHandle, T
                ] = asynchronous_parallel_requests(
                    remote_requests_in_flight=self.remote_replay_requests_in_flight,
                    actors=[rand_actor],
                    ray_wait_timeout_s=0.1,
                    max_remote_requests_in_flight_per_actor=num_requests_to_launch,
                    remote_fn=lambda actor: actor.replay(),
                )
            for replay_actor, sample_batches in replay_samples_ready.items():
                for sample_batch in sample_batches:
                    self.replay_sample_batches.append((replay_actor, sample_batch))
        wait_on_replay_actors(0.1)
        while self.replay_sample_batches:
            try:
                item = self.replay_sample_batches[0]
                if item:
                    self.learner_thread.inqueue.put(
                        self.replay_sample_batches[0], timeout=0.001
                    )
                    self.replay_sample_batches.pop(0)
            except queue.Full:
                break
def sample_from_replay_buffer_place_on_learner_queue_non_blocking ( self , num_samples_collected : Dict [ ActorHandle , int ] ) -> None : def wait_on_replay_actors ( timeout : float ) -> None : """Wait for the replay actors to finish sampling for timeout seconds.
            If the timeout is None, then block on the actors indefinitely.
            """ replay_samples_ready : Dict [ ActorHandle , T ] = wait_asynchronous_requests ( remote_requests_in_flight = self . remote_replay_requests_in_flight , ray_wait_timeout_s = timeout , ) for replay_actor , sample_batches in replay_samples_ready . items ( ) : for sample_batch in sample_batches : self . replay_sample_batches . append ( ( replay_actor , sample_batch ) ) num_samples_collected = sum ( num_samples_collected . values ( ) ) self . curr_num_samples_collected += num_samples_collected if self . curr_num_samples_collected >= self . config [ "train_batch_size" ] : wait_on_replay_actors ( None ) training_intensity = int ( self . config [ "training_intensity" ] or 1 ) num_requests_to_launch = ( self . curr_num_samples_collected / self . config [ "train_batch_size" ] ) * training_intensity num_requests_to_launch = max ( 1 , round ( num_requests_to_launch ) ) self . curr_num_samples_collected = 0 for _ in range ( num_requests_to_launch ) : rand_actor = random . choice ( self . replay_actors ) replay_samples_ready : Dict [ ActorHandle , T ] = asynchronous_parallel_requests ( remote_requests_in_flight = self . remote_replay_requests_in_flight , actors = [ rand_actor ] , ray_wait_timeout_s = 0.1 , max_remote_requests_in_flight_per_actor = num_requests_to_launch , remote_fn = lambda actor : actor . replay ( ) , ) for replay_actor , sample_batches in replay_samples_ready . items ( ) : for sample_batch in sample_batches : self . replay_sample_batches . append ( ( replay_actor , sample_batch ) ) wait_on_replay_actors ( 0.1 ) while self . replay_sample_batches : try : item = self . replay_sample_batches [ 0 ] if item : self . learner_thread . inqueue . put ( self . replay_sample_batches [ 0 ] , timeout = 0.001 ) self . replay_sample_batches . pop ( 0 ) except queue . Full : break
Get samples from the replay buffer and place them on the learner queue.
Get samples from the replay buffer and place them on the learner queue .
"""Get samples from the replay buffer and place them on the learner queue.

        Args:
            num_samples_collected: A mapping from ActorHandle (RolloutWorker) to
                number of samples returned by the remote worker. This is used to
                implement training intensity which is the concept of triggering a
                certain amount of training based on the number of samples that have
                been collected since the last time that training was triggered.

        """ """Wait for the replay actors to finish sampling for timeout seconds.
            If the timeout is None, then block on the actors indefinitely.
            """ # add the sample batches to the learner queue # the replay buffer returns none if it has not been filled to # the minimum threshold yet.
self
num_samples_collected
Dict[ActorHandle, int]
A mapping from ActorHandle (RolloutWorker) to
number of samples returned by the remote worker. This is used to
implement training intensity which is the concept of triggering a
certain amount of training based on the number of samples that have
been collected since the last time that training was triggered.
A mapping from ActorHandle ( RolloutWorker ) to number of samples returned by the remote worker . This is used to implement training intensity which is the concept of triggering a certain amount of training based on the number of samples that have been collected since the last time that training was triggered .