Add stop_token, Modify GPU inference, Support hybrid

opus24 · opus24 · commit 25847977d775 · 2024-09-19T18:02:42.000+09:00
diff --git a/examples/lpu_inference.py b/examples/lpu_inference.py
@@ -5,12 +5,13 @@
     "Hello, my name is"
 ]
 # Create a sampling params object.
-sampling_params = SamplingParams(temperature=0.8, top_p=0.95, top_k=1, min_tokens=30, max_tokens=30)
+sampling_params = SamplingParams(temperature=0.8, top_p=0.8, top_k=1, repetition_penalty=1.2, max_tokens=60)
 
 # Create an LLM.
 #llm = LLM(model="facebook/opt-1.3b", device="fpga", pipeline_parallel_size=2)
-llm = LLM(model="meta-llama/Meta-Llama-3-8B", device="fpga", tensor_parallel_size=1)
+#llm = LLM(model="meta-llama/Meta-Llama-3-8B", device="fpga", tensor_parallel_size=1)
 #llm = LLM(model="TinyLlama/TinyLlama-1.1B-Chat-v1.0", device="fpga", tensor_parallel_size=1)
+llm = LLM(model="TinyLlama/TinyLlama-1.1B-Chat-v1.0", device="fpga", num_lpu_devices=2, num_gpu_devices=1)
 
 # Generate texts from the prompts. The output is a list of RequestOutput objects
 # that contain the prompt, generated text, and other information.
diff --git a/examples/lpu_openai_completion_client.py b/examples/lpu_openai_completion_client.py
@@ -0,0 +1,31 @@
+from openai import OpenAI
+
+# Modify OpenAI's API key and API base to use vLLM's API server.
+openai_api_key = "EMPTY"
+openai_api_base = "http://localhost:8000/v1"
+
+client = OpenAI(
+    api_key=openai_api_key,
+    base_url=openai_api_base,
+)
+
+models = client.models.list()
+model = models.data[0].id
+
+# Completion API
+stream=True
+prompt="Hello, my name is"
+completion = client.completions.create(
+    model=model,
+    prompt=prompt,
+    stream=stream,
+    )
+
+print("Prompt:", prompt)
+print("Completion results:")
+if stream:
+    for c in completion:
+        print(c.choices[0].text, end="")
+    print()
+else:
+    print(completion)
diff --git a/examples/mini_testbench.sh b/examples/mini_testbench.sh
@@ -8,6 +8,7 @@ current_datetime=$(date "+%Y-%m-%d %H:%M:%S")
 echo "$current_datetime"
 echo "$current_datetime" >> ${log_sum}
 
+# LLMEngine Test
 for model_id in "${model_ids[@]}"; do
   for num_device in "${num_devices[@]}"; do
     #IFS='\' read -ra parts <<< "$model_id"
@@ -24,6 +25,7 @@ for model_id in "${model_ids[@]}"; do
   done
 done
 
+# LLMEngineAsync Test with vLLM serve
 for model_id in "${model_ids[@]}"; do
   for num_device in "${num_devices[@]}"; do
     model_name=$(echo "$model_id" | awk -F'/' '{print $NF}')
@@ -65,3 +67,44 @@ for model_id in "${model_ids[@]}"; do
 done
 
 
+
+# OpenAI API Test
+model_id=${model_ids[0]}
+num_device=${num_devices[0]}
+model_name=$(echo "$model_id" | awk -F'/' '{print $NF}')
+echo "*********************************"
+echo "**** Start serving_${model_name}_${num_device}"
+echo "*********************************"
+python -m vllm.entrypoints.api_server --model ${model_id} --device fpga --tensor-parallel-size ${num_device} &
+
+# Waiting for server
+while ! nc -z localhost "8000"; do  
+    echo "[Testbench] Waiting for server..."
+    sleep 3 
+done
+echo "[Testbench] The server is ready!"
+
+python lpu_openai_completion_client.py > log/openai_serve_${model_name}_${num_device}.txt
+
+# Waiting for process kill
+PID=$(jobs -p | tail -n 1)
+if [ -n "$PID" ]; then
+    kill -SIGINT "$PID"
+    while true; do
+        if ps -p "$PID" > /dev/null; then
+            echo "[Testbench] Kill the process..."
+            sleep 3
+        else
+            echo "[Testbench] Process (PID: $PID) is killed."
+            break
+        fi
+    done
+fi
+
+# Write log in text file
+echo "*********************************" >> ${log_sum}
+echo "The Result of log/openai_serve_${model_name}_${num_device}.txt" >> ${log_sum}
+tail -n 1 "log/openai_serve_${model_name}_${num_device}.txt" >> ${log_sum}
+echo "" >> ${log_sum}
+
+
diff --git a/examples/openai_test.sh b/examples/openai_test.sh
@@ -0,0 +1 @@
+python -m vllm.entrypoints.openai.api_server --model facebook/opt-1.3b --device fpga --tensor-parallel-size 2
diff --git a/examples/vllm_serve.sh b/examples/vllm_serve.sh
@@ -0,0 +1,4 @@
+
+#python -m vllm.entrypoints.api_server --model facebook/opt-1.3b --device fpga --tensor-parallel-size 2
+python -m vllm.entrypoints.api_server --model facebook/opt-1.3b --device fpga --num-gpu-devices 1 --num-lpu-devices 2
+#python -m vllm.entrypoints.api_server --model facebook/opt-1.3b --device fpga --num_gpu_devices 1 --num_lpu_devices 2
diff --git a/vllm/config.py b/vllm/config.py
@@ -1011,14 +1011,18 @@ def is_multi_step(self) -> bool:
 
 class DeviceConfig:
     device: Optional[torch.device]
+    num_gpu_devices: int
+    num_lpu_devices: int
 
-    def __init__(self, device: str = "auto") -> None:
+    def __init__(self, device: str = "auto", num_gpu_devices: int = 0, num_lpu_devices: int = 1) -> None:
         if device == "auto":
             # Automated device type detection
             if is_neuron():
                 self.device_type = "neuron"
             elif is_openvino():
                 self.device_type = "openvino"
+            elif current_platform.is_lpu():
+                self.device_type = "fpga"
             elif current_platform.is_tpu():
                 self.device_type = "tpu"
             elif is_cpu():
@@ -1042,6 +1046,8 @@ def __init__(self, device: str = "auto") -> None:
             # Set device with device type
             self.device = torch.device(self.device_type)
 
+        self.num_gpu_devices=num_gpu_devices
+        self.num_lpu_devices=num_lpu_devices
 
 class SpeculativeConfig:
     """Configuration for speculative decoding.
diff --git a/vllm/core/scheduler.py b/vllm/core/scheduler.py
@@ -7,7 +7,7 @@
 from typing import (Callable, Deque, Dict, Iterable, List, Optional, Set,
                     Tuple, Union)
 
-from vllm.config import CacheConfig, LoRAConfig, SchedulerConfig
+from vllm.config import CacheConfig, LoRAConfig, SchedulerConfig, DeviceConfig
 from vllm.core.interfaces import AllocStatus, BlockSpaceManager
 from vllm.logger import init_logger, print_logger
 from vllm.lora.request import LoRARequest
@@ -301,6 +301,7 @@ def __init__(
         scheduler_config: SchedulerConfig,
         cache_config: CacheConfig,
         lora_config: Optional[LoRAConfig],
+        device_config: Optional[DeviceConfig],
         pipeline_parallel_size: int = 1,
         output_proc_callback: Optional[Callable] = None,
     ) -> None:
@@ -310,6 +311,10 @@ def __init__(
         # simple and NOT fair. It can lead to starvation of some
         # LoRAs. This should be improved in the future.
         self.lora_config = lora_config
+        # NOTE(hyunjun): Currently, LPU vLLM backend needs to reduce scheduler dependency
+        # _can_append_slots, _append_slots
+        # Temporally, we change resource management flow with device config
+        self.device_config = device_config
 
         version = "v1"
         if self.scheduler_config.use_v2_block_manager:
@@ -576,63 +581,119 @@ def _schedule_running(
                 assert self.output_proc_callback is not None
                 self.output_proc_callback()
                 self.running = tmp
-
-            while not True: #TODO #self._can_append_slots(seq_group):
-                budget.subtract_num_batched_tokens(seq_group.request_id,
-                                                   num_running_tokens)
-                num_running_seqs = seq_group.get_max_num_running_seqs()
-                budget.subtract_num_seqs(seq_group.request_id,
-                                         num_running_seqs)
-
-                if (curr_loras is not None and seq_group.lora_int_id > 0
-                        and seq_group.lora_int_id in curr_loras):
-                    curr_loras.remove(seq_group.lora_int_id)
-
-                if running_queue:
-                    # Preempt the lowest-priority sequence groups.
-                    victim_seq_group = running_queue.pop()
-                    preempted_mode = self._preempt(victim_seq_group,
-                                                   blocks_to_swap_out)
-                    if preempted_mode == PreemptionMode.RECOMPUTE:
-                        preempted.append(victim_seq_group)
-                    else:
-                        swapped_out.append(victim_seq_group)
-                else:
-                    # No other sequence groups can be preempted.
-                    # Preempt the current sequence group.
-                    preempted_mode = self._preempt(seq_group,
-                                                   blocks_to_swap_out)
-                    if preempted_mode == PreemptionMode.RECOMPUTE:
-                        preempted.append(seq_group)
-                    else:
-                        swapped_out.append(seq_group)
-                    break
+            if self.device_config.device_type == "fpga":
+              while not True: #self._can_append_slots(seq_group):
+                  budget.subtract_num_batched_tokens(seq_group.request_id,
+                                                     num_running_tokens)
+                  num_running_seqs = seq_group.get_max_num_running_seqs()
+                  budget.subtract_num_seqs(seq_group.request_id,
+                                           num_running_seqs)
+
+                  if (curr_loras is not None and seq_group.lora_int_id > 0
+                          and seq_group.lora_int_id in curr_loras):
+                      curr_loras.remove(seq_group.lora_int_id)
+
+                  if running_queue:
+                      # Preempt the lowest-priority sequence groups.
+                      victim_seq_group = running_queue.pop()
+                      preempted_mode = self._preempt(victim_seq_group,
+                                                     blocks_to_swap_out)
+                      if preempted_mode == PreemptionMode.RECOMPUTE:
+                          preempted.append(victim_seq_group)
+                      else:
+                          swapped_out.append(victim_seq_group)
+                  else:
+                      # No other sequence groups can be preempted.
+                      # Preempt the current sequence group.
+                      preempted_mode = self._preempt(seq_group,
+                                                     blocks_to_swap_out)
+                      if preempted_mode == PreemptionMode.RECOMPUTE:
+                          preempted.append(seq_group)
+                      else:
+                          swapped_out.append(seq_group)
+                      break
+              else:
+                  is_prefill = seq_group.is_prefill()
+                  scheduled_seq_group: ScheduledSequenceGroup = \
+                      self._scheduled_seq_group_cache[self.cache_id].get_object()
+                  scheduled_seq_group.seq_group = seq_group
+                  if is_prefill:
+                      scheduled_seq_group.token_chunk_size = num_running_tokens
+                      prefill_seq_groups.append(scheduled_seq_group)
+                      ret.prefill_seq_groups_list.append(seq_group)
+                  else:
+                      scheduled_seq_group.token_chunk_size = 1
+                      decode_seq_groups.append(scheduled_seq_group)
+                      ret.decode_seq_groups_list.append(seq_group)
+
+                  budget.add_num_batched_tokens(seq_group.request_id,
+                                                num_running_tokens)
+                  # OPTIMIZATION:  Note that get_max_num_running_seqs is
+                  # expensive. For the default scheduling chase where
+                  # enable_chunking is False, num_seqs are updated before running
+                  # this method, so we don't have to update it again here.
+                  if enable_chunking:
+                      num_running_seqs = seq_group.get_max_num_running_seqs()
+                      budget.add_num_seqs(seq_group.request_id, num_running_seqs)
+                  if curr_loras is not None and seq_group.lora_int_id > 0:
+                      curr_loras.add(seq_group.lora_int_id)
             else:
-                #self._append_slots(seq_group, blocks_to_copy)
-                is_prefill = seq_group.is_prefill()
-                scheduled_seq_group: ScheduledSequenceGroup = \
-                    self._scheduled_seq_group_cache[self.cache_id].get_object()
-                scheduled_seq_group.seq_group = seq_group
-                if is_prefill:
-                    scheduled_seq_group.token_chunk_size = num_running_tokens
-                    prefill_seq_groups.append(scheduled_seq_group)
-                    ret.prefill_seq_groups_list.append(seq_group)
-                else:
-                    scheduled_seq_group.token_chunk_size = 1
-                    decode_seq_groups.append(scheduled_seq_group)
-                    ret.decode_seq_groups_list.append(seq_group)
-
-                budget.add_num_batched_tokens(seq_group.request_id,
-                                              num_running_tokens)
-                # OPTIMIZATION:  Note that get_max_num_running_seqs is
-                # expensive. For the default scheduling chase where
-                # enable_chunking is False, num_seqs are updated before running
-                # this method, so we don't have to update it again here.
-                if enable_chunking:
-                    num_running_seqs = seq_group.get_max_num_running_seqs()
-                    budget.add_num_seqs(seq_group.request_id, num_running_seqs)
-                if curr_loras is not None and seq_group.lora_int_id > 0:
-                    curr_loras.add(seq_group.lora_int_id)
+              while not self._can_append_slots(seq_group):
+                  budget.subtract_num_batched_tokens(seq_group.request_id,
+                                                     num_running_tokens)
+                  num_running_seqs = seq_group.get_max_num_running_seqs()
+                  budget.subtract_num_seqs(seq_group.request_id,
+                                           num_running_seqs)
+
+                  if (curr_loras is not None and seq_group.lora_int_id > 0
+                          and seq_group.lora_int_id in curr_loras):
+                      curr_loras.remove(seq_group.lora_int_id)
+
+                  if running_queue:
+                      # Preempt the lowest-priority sequence groups.
+                      victim_seq_group = running_queue.pop()
+                      preempted_mode = self._preempt(victim_seq_group,
+                                                     blocks_to_swap_out)
+                      if preempted_mode == PreemptionMode.RECOMPUTE:
+                          preempted.append(victim_seq_group)
+                      else:
+                          swapped_out.append(victim_seq_group)
+                  else:
+                      # No other sequence groups can be preempted.
+                      # Preempt the current sequence group.
+                      preempted_mode = self._preempt(seq_group,
+                                                     blocks_to_swap_out)
+                      if preempted_mode == PreemptionMode.RECOMPUTE:
+                          preempted.append(seq_group)
+                      else:
+                          swapped_out.append(seq_group)
+                      break
+              else:
+                  self._append_slots(seq_group, blocks_to_copy)
+                  is_prefill = seq_group.is_prefill()
+                  scheduled_seq_group: ScheduledSequenceGroup = \
+                      self._scheduled_seq_group_cache[self.cache_id].get_object()
+                  scheduled_seq_group.seq_group = seq_group
+                  if is_prefill:
+                      scheduled_seq_group.token_chunk_size = num_running_tokens
+                      prefill_seq_groups.append(scheduled_seq_group)
+                      ret.prefill_seq_groups_list.append(seq_group)
+                  else:
+                      scheduled_seq_group.token_chunk_size = 1
+                      decode_seq_groups.append(scheduled_seq_group)
+                      ret.decode_seq_groups_list.append(seq_group)
+
+                  budget.add_num_batched_tokens(seq_group.request_id,
+                                                num_running_tokens)
+                  # OPTIMIZATION:  Note that get_max_num_running_seqs is
+                  # expensive. For the default scheduling chase where
+                  # enable_chunking is False, num_seqs are updated before running
+                  # this method, so we don't have to update it again here.
+                  if enable_chunking:
+                      num_running_seqs = seq_group.get_max_num_running_seqs()
+                      budget.add_num_seqs(seq_group.request_id, num_running_seqs)
+                  if curr_loras is not None and seq_group.lora_int_id > 0:
+                      curr_loras.add(seq_group.lora_int_id)
 
         self._scheduler_running_outputs_cache[self.next_cache_id].reset()
         self._scheduled_seq_group_cache[self.next_cache_id].reset()
diff --git a/vllm/engine/arg_utils.py b/vllm/engine/arg_utils.py
@@ -14,7 +14,7 @@
                          PromptAdapterConfig, SchedulerConfig,
                          SpeculativeConfig, TokenizerPoolConfig)
 from vllm.executor.executor_base import ExecutorBase
-from vllm.logger import init_logger
+from vllm.logger import init_logger, print_logger
 from vllm.model_executor.layers.quantization import QUANTIZATION_METHODS
 from vllm.utils import FlexibleArgumentParser
 
@@ -149,6 +149,10 @@ class EngineArgs:
     collect_detailed_traces: Optional[str] = None
     disable_async_output_proc: bool = False
 
+    #NOTE(hyunjun): custom option for hybrid
+    num_gpu_devices: int = 0
+    num_lpu_devices: int = 1
+
     def __post_init__(self):
         if self.tokenizer is None:
             self.tokenizer = self.model
@@ -741,6 +745,16 @@ def add_cli_args(parser: FlexibleArgumentParser) -> FlexibleArgumentParser:
             default=EngineArgs.disable_async_output_proc,
             help="Disable async output processing. This may result in "
             "lower performance.")
+        parser.add_argument(
+            '--num-gpu-devices',
+            type=int,
+            default=0,
+            help='the number of gpu devices for hybrid system')
+        parser.add_argument(
+            '--num-lpu-devices',
+            type=int,
+            default=1,
+            help='the number of lpu devices for hybrid system')
         return parser
 
     @classmethod
@@ -775,8 +789,7 @@ def create_engine_config(self) -> EngineConfig:
         assert self.cpu_offload_gb >= 0, (
             "CPU offload space must be non-negative"
             f", but got {self.cpu_offload_gb}")
-
-        device_config = DeviceConfig(device=self.device)
+        device_config = DeviceConfig(device=self.device, num_gpu_devices=self.num_gpu_devices, num_lpu_devices=self.num_lpu_devices)
         model_config = ModelConfig(
             model=self.model,
             tokenizer=self.tokenizer,
diff --git a/vllm/engine/llm_engine.py b/vllm/engine/llm_engine.py
diff --git a/vllm/executor/lpu_executor.py b/vllm/executor/lpu_executor.py
diff --git a/vllm/worker/lpu_model_runner.py b/vllm/worker/lpu_model_runner.py
diff --git a/vllm/worker/lpu_worker.py b/vllm/worker/lpu_worker.py

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1 @@`
	`1`	`+python -m vllm.entrypoints.openai.api_server --model facebook/opt-1.3b --device fpga --tensor-parallel-size 2`