class_docstr: REST policy server that acts as an offline data source.
  
  This launches a multi-threaded server that listens on the specified host
  and port to serve policy requests and forward experiences to RLlib. For
  high performance experience collection, it implements InputReader.
  
  For an example, run `examples/serving/cartpole_server.py` along
  with `examples/serving/cartpole_client.py --inference-mode=local|remote`.
  
  WARNING: This class is not meant to be publicly exposed. Anyone that can
  communicate with this server can execute arbitary code on the machine. Use
  this with caution, in isolated environments, and at your own risk.
  
  .. testcode::
      :skipif: True
  
      import gymnasium as gym
      from ray.rllib.algorithms.ppo import PPOConfig
      from ray.rllib.env.policy_client import PolicyClient
      from ray.rllib.env.policy_server_input import PolicyServerInput
      addr, port = ...
      config = (
          PPOConfig()
          .environment("CartPole-v1")
          .offline_data(
              input_=lambda ioctx: PolicyServerInput(ioctx, addr, port)
          )
          # Run just 1 server (in the Algorithm's WorkerSet).
          .rollouts(num_rollout_workers=0)
      )
      algo = config.build()
      while True:
          algo.train()
      client = PolicyClient(
          "localhost:9900", inference_mode="local")
      eps_id = client.start_episode()
      env = gym.make("CartPole-v1")
      obs, info = env.reset()
      action = client.get_action(eps_id, obs)
      _, reward, _, _, _ = env.step(action)
      client.log_returns(eps_id, reward)
      client.log_returns(eps_id, reward)
      algo.stop()
class_docstr_tok: ["REST","policy","server","that","acts","as","an","offline","data","source",".","This","launches","a","multi-threaded","server","that","listens","on","the","specified","host","and","port","to","serve","policy","requests","and","forward","experiences","to","RLlib",".","For","high","performance","experience","collection",",","it","implements","InputReader",".","For","an","example",",","run","`","examples\/serving\/cartpole_server.py","`","along","with","`","examples\/serving\/cartpole_client.py","--","inference-mode=local|remote","`",".","WARNING",":","This","class","is","not","meant","to","be","publicly","exposed",".","Anyone","that","can","communicate","with","this","server","can","execute","arbitary","code","on","the","machine",".","Use","this","with","caution",",","in","isolated","environments",",","and","at","your","own","risk",".","..","testcode",":",":",":","skipif",":","True","import","gymnasium","as","gym","from","ray.rllib.algorithms.ppo","import","PPOConfig","from","ray.rllib.env.policy_client","import","PolicyClient","from","ray.rllib.env.policy_server_input","import","PolicyServerInput","addr",",","port","=","...","config","=","(","PPOConfig","(",")",".environment","(","``","CartPole-v1","''",")",".offline_data","(","input_=lambda","ioctx",":","PolicyServerInput","(","ioctx",",","addr",",","port",")",")","#","Run","just","1","server","(","in","the","Algorithm","'s","WorkerSet",")",".",".rollouts","(","num_rollout_workers=0",")",")","algo","=","config.build","(",")","while","True",":","algo.train","(",")","client","=","PolicyClient","(","``","localhost:9900","''",",","inference_mode=","''","local","''",")","eps_id","=","client.start_episode","(",")","env","=","gym.make","(","``","CartPole-v1","''",")","obs",",","info","=","env.reset","(",")","action","=","client.get_action","(","eps_id",",","obs",")","_",",","reward",",","_",",","_",",","_","=","env.step","(","action",")","client.log_returns","(","eps_id",",","reward",")","client.log_returns","(","eps_id",",","reward",")","algo.stop","(",")"]
code: def next(self):
      # Blocking wait until there is something in the deque.
      while len(self.samples_queue) == 0:
          time.sleep(0.1)
      # Utilize last items first in order to remain as closely as possible
      # to operating on-policy.
      return self.samples_queue.pop()
  
imports: from collections import deque
  from http.server import HTTPServer, SimpleHTTPRequestHandler
  import logging
  import queue
  from socketserver import ThreadingMixIn
  import threading
  import time
  import traceback
  from typing import List
  import ray.cloudpickle
  from ray.rllib.env.policy_client import _create_embedded_rollout_worker, Commands
  from ray.rllib.offline.input_reader import InputReader
  from ray.rllib.offline.io_context import IOContext
  from ray.rllib.policy.sample_batch import SampleBatch
  from ray.rllib.utils.annotations import override, PublicAPI
  from ray.rllib.evaluation.metrics import RolloutMetrics
  from ray.rllib.evaluation.sampler import SamplerInput
  from ray.rllib.utils.typing import SampleBatchType
label_desc: Use image node_id 2 for calling the PolicyServerInput obj's underlying member method code with example usage: obj.next() and returns: self