async_llm_engine.py 37.1 KB
Newer Older
1
2
import asyncio
import time
Antoni Baum's avatar
Antoni Baum committed
3
from functools import partial
4
5
from typing import (AsyncIterator, Callable, Dict, Iterable, List, Optional,
                    Set, Tuple, Type, Union)
6

7
8
from transformers import PreTrainedTokenizer

9
import vllm.envs as envs
10
from vllm.config import DecodingConfig, ModelConfig
11
from vllm.core.scheduler import SchedulerOutputs
Woosuk Kwon's avatar
Woosuk Kwon committed
12
from vllm.engine.arg_utils import AsyncEngineArgs
13
from vllm.engine.async_timeout import asyncio_timeout
Woosuk Kwon's avatar
Woosuk Kwon committed
14
from vllm.engine.llm_engine import LLMEngine
15
from vllm.executor.ray_utils import initialize_ray_cluster, ray
16
from vllm.inputs import LLMInputs, PromptInputs
Woosuk Kwon's avatar
Woosuk Kwon committed
17
from vllm.logger import init_logger
18
from vllm.lora.request import LoRARequest
19
20
from vllm.outputs import EmbeddingRequestOutput, RequestOutput
from vllm.pooling_params import PoolingParams
Woosuk Kwon's avatar
Woosuk Kwon committed
21
from vllm.sampling_params import SamplingParams
22
from vllm.sequence import ExecuteModelRequest, SamplerOutput
yhu422's avatar
yhu422 committed
23
from vllm.usage.usage_lib import UsageContext
24
25

logger = init_logger(__name__)
26
ENGINE_ITERATION_TIMEOUT_S = envs.VLLM_ENGINE_ITERATION_TIMEOUT_S
27

Antoni Baum's avatar
Antoni Baum committed
28

29
30
31
32
class AsyncEngineDeadError(RuntimeError):
    pass


33
34
35
36
37
38
39
def _log_task_completion(task: asyncio.Task,
                         error_callback: Callable[[Exception], None]) -> None:
    """This function is only intended for the `engine.run_engine_loop()` task.

    In particular, that task runs a `while True` loop that can only exit if
    there is an exception.
    """
40
41

    exception = None
42
    try:
43
44
45
46
47
48
49
50
        return_value = task.result()
        raise AssertionError(
            f"The engine background task should never finish without an "
            f"exception. {return_value}")
    except asyncio.exceptions.CancelledError:
        # We assume that if the task is cancelled, we are gracefully shutting
        # down. This should only happen on program exit.
        logger.info("Engine is gracefully shutting down.")
51
52
53
54
55
    except Exception as e:
        exception = e
        logger.error("Engine background task failed", exc_info=e)
        error_callback(exception)
        raise AsyncEngineDeadError(
56
57
58
            "Task finished unexpectedly. This should never happen! "
            "Please open an issue on Github. See stack trace above for the"
            "actual cause.") from e
59
60


Antoni Baum's avatar
Antoni Baum committed
61
class AsyncStream:
62
63
    """A stream of RequestOutputs or EmbeddingRequestOutputs for a request
    that can be iterated over asynchronously."""
Antoni Baum's avatar
Antoni Baum committed
64
65
66

    def __init__(self, request_id: str) -> None:
        self.request_id = request_id
67
        self._queue: asyncio.Queue = asyncio.Queue()
Antoni Baum's avatar
Antoni Baum committed
68
69
        self._finished = False

70
71
    def put(self, item: Union[RequestOutput, EmbeddingRequestOutput,
                              Exception]) -> None:
Antoni Baum's avatar
Antoni Baum committed
72
73
74
75
76
        if self._finished:
            return
        self._queue.put_nowait(item)

    def finish(self) -> None:
77
        self._queue.put_nowait(StopAsyncIteration())
Antoni Baum's avatar
Antoni Baum committed
78
79
80
81
82
83
84
85
86
        self._finished = True

    @property
    def finished(self) -> bool:
        return self._finished

    def __aiter__(self):
        return self

87
    async def __anext__(self) -> Union[RequestOutput, EmbeddingRequestOutput]:
Antoni Baum's avatar
Antoni Baum committed
88
        result = await self._queue.get()
89
        if isinstance(result, Exception):
90
            raise result
Antoni Baum's avatar
Antoni Baum committed
91
92
93
        return result


94
95
96
97
98
99
100
101
class RequestTracker:
    """Synchronous abstraction for tracking requests."""

    def __init__(self) -> None:
        self._request_streams: Dict[str, AsyncStream] = {}
        self._finished_requests: asyncio.Queue[str] = asyncio.Queue()
        self._new_requests: asyncio.Queue[Tuple[AsyncStream,
                                                dict]] = asyncio.Queue()
102
        self.new_requests_event = asyncio.Event()
103
104
105
106

    def __contains__(self, item):
        return item in self._request_streams

107
108
    def __len__(self) -> int:
        return len(self._request_streams)
109
110
111
112
113
114
115
116

    def propagate_exception(self,
                            exc: Exception,
                            request_id: Optional[str] = None) -> None:
        """Propagate an exception to request streams
        (all if request_id is None)."""
        if request_id is not None:
            self._request_streams[request_id].put(exc)
117
            self.abort_request(request_id)
118
        else:
119
            for rid, stream in self._request_streams.items():
120
                stream.put(exc)
121
                self.abort_request(rid)
122
123

    def process_request_output(self,
124
125
                               request_output: Union[RequestOutput,
                                                     EmbeddingRequestOutput],
126
127
128
129
130
131
132
133
                               *,
                               verbose: bool = False) -> None:
        """Process a request output from the engine."""
        request_id = request_output.request_id

        self._request_streams[request_id].put(request_output)
        if request_output.finished:
            if verbose:
134
                logger.info("Finished request %s.", request_id)
135
136
            self.abort_request(request_id)

137
138
139
140
141
142
143
144
    def process_exception(self,
                          request_id: str,
                          exception: Exception,
                          *,
                          verbose: bool = False) -> None:
        """Propagate an exception from the engine."""
        self._request_streams[request_id].put(exception)
        if verbose:
145
            logger.info("Finished request %s.", request_id)
146
147
        self.abort_request(request_id)

148
149
150
151
152
153
154
155
156
157
158
159
    def add_request(self, request_id: str,
                    **engine_add_request_kwargs) -> AsyncStream:
        """Add a request to be sent to the engine on the next background
        loop iteration."""
        if request_id in self._request_streams:
            raise KeyError(f"Request {request_id} already exists.")

        stream = AsyncStream(request_id)
        self._new_requests.put_nowait((stream, {
            "request_id": request_id,
            **engine_add_request_kwargs
        }))
160
161
162

        self.new_requests_event.set()

163
164
165
166
167
        return stream

    def abort_request(self, request_id: str, *, verbose: bool = False) -> None:
        """Abort a request during next background loop iteration."""
        if verbose:
168
            logger.info("Aborted request %s.", request_id)
169
170
171
172
173
174
175
176
177
178

        self._finished_requests.put_nowait(request_id)

        if request_id not in self._request_streams or self._request_streams[
                request_id].finished:
            # The request has already finished or been aborted.
            return

        self._request_streams[request_id].finish()

179
    def get_new_and_finished_requests(self) -> Tuple[List[Dict], Set[str]]:
180
181
        """Get the new requests and finished requests to be
        sent to the engine."""
182
        new_requests: List[Dict] = []
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
        finished_requests: Set[str] = set()

        while not self._finished_requests.empty():
            request_id = self._finished_requests.get_nowait()
            finished_requests.add(request_id)
            self._request_streams.pop(request_id, None)

        while not self._new_requests.empty():
            stream, new_request = self._new_requests.get_nowait()
            if stream.request_id in finished_requests:
                # The request has already been aborted.
                stream.finish()
                continue
            self._request_streams[stream.request_id] = stream
            new_requests.append(new_request)

        return new_requests, finished_requests
Antoni Baum's avatar
Antoni Baum committed
200

201
    async def wait_for_new_requests(self):
202
203
204
205
206
207
        if not self.has_new_requests():
            await self.new_requests_event.wait()
        self.new_requests_event.clear()

    def has_new_requests(self):
        return not self._new_requests.empty()
208

Antoni Baum's avatar
Antoni Baum committed
209
210
211
212

class _AsyncLLMEngine(LLMEngine):
    """Extension of LLMEngine to add async methods."""

213
    async def step_async(
214
215
        self, virtual_engine: int
    ) -> List[Union[RequestOutput, EmbeddingRequestOutput]]:
Antoni Baum's avatar
Antoni Baum committed
216
217
218
219
220
221
222
223
224
        """Performs one decoding iteration and returns newly generated results.
        The workers are ran asynchronously if possible.

        This function performs one decoding iteration of the engine. It first
        schedules the sequences to be executed in the next iteration and the
        token blocks to be swapped in/out/copy. Then, it executes the model
        and updates the scheduler with the model outputs. Finally, it decodes
        the sequences and returns the newly generated results.
        """
225
226
        seq_group_metadata_list, scheduler_outputs = self.scheduler[
            virtual_engine].schedule()
Mor Zusman's avatar
Mor Zusman committed
227
228
        finished_requests_ids = self.scheduler[
            virtual_engine].get_and_reset_finished_requests_ids()
Antoni Baum's avatar
Antoni Baum committed
229

230
231
        if not scheduler_outputs.is_empty():
            # Execute the model.
232
233
234
235
236
            execute_model_req = ExecuteModelRequest(
                seq_group_metadata_list=seq_group_metadata_list,
                blocks_to_swap_in=scheduler_outputs.blocks_to_swap_in,
                blocks_to_swap_out=scheduler_outputs.blocks_to_swap_out,
                blocks_to_copy=scheduler_outputs.blocks_to_copy,
237
                virtual_engine=virtual_engine,
238
239
                num_lookahead_slots=scheduler_outputs.num_lookahead_slots,
                running_queue_size=scheduler_outputs.running_queue_size,
Mor Zusman's avatar
Mor Zusman committed
240
                finished_requests_ids=finished_requests_ids)
241
            output = await self.model_executor.execute_model_async(
242
                execute_model_req)
243
244
        else:
            output = []
Antoni Baum's avatar
Antoni Baum committed
245

246
        request_outputs = self._process_model_outputs(
247
            output, scheduler_outputs.scheduled_seq_groups,
248
            scheduler_outputs.ignored_seq_groups, seq_group_metadata_list)
Antoni Baum's avatar
Antoni Baum committed
249

250
        # Log stats.
251
        self.do_log_stats(scheduler_outputs, output)
252

253
254
255
        # Tracing
        self.do_tracing(scheduler_outputs)

256
257
        return request_outputs

258
259
260
261
    async def stop_remote_worker_execution_loop_async(self) -> None:
        """Stop the remote worker execution loop."""
        await self.model_executor.stop_remote_worker_execution_loop_async()

262
    async def process_model_inputs_async(
263
        self,
264
265
        request_id: str,
        inputs: PromptInputs,
266
        lora_request: Optional[LoRARequest] = None,
267
268
269
270
271
272
273
274
275
    ) -> LLMInputs:
        if isinstance(inputs, str):
            inputs = {"prompt": inputs}

        if "prompt_token_ids" not in inputs:
            tokenizer = self.get_tokenizer_group("prompts must be None if "
                                                 "skip_tokenizer_init is True")

            prompt_token_ids = await tokenizer.encode_async(
276
                request_id=request_id,
277
                prompt=inputs["prompt"],
278
                lora_request=lora_request)
279
280
281
        else:
            prompt_token_ids = inputs["prompt_token_ids"]

282
283
284
285
286
        llm_inputs = LLMInputs(prompt_token_ids=prompt_token_ids,
                               prompt=inputs.get("prompt"),
                               multi_modal_data=inputs.get("multi_modal_data"))

        return self.input_processor(llm_inputs)
287
288
289
290

    async def add_request_async(
        self,
        request_id: str,
291
        inputs: PromptInputs,
292
        params: Union[SamplingParams, PoolingParams],
293
294
        arrival_time: Optional[float] = None,
        lora_request: Optional[LoRARequest] = None,
295
        trace_headers: Optional[Dict[str, str]] = None,
296
297
298
299
300
301
    ) -> None:
        if lora_request is not None and not self.lora_config:
            raise ValueError(f"Got lora_request {lora_request} but LoRA is "
                             "not enabled!")
        if arrival_time is None:
            arrival_time = time.time()
302
303
304
305
306

        processed_inputs = await self.process_model_inputs_async(
            request_id=request_id, inputs=inputs, lora_request=lora_request)

        self._add_processed_request(
307
            request_id=request_id,
308
309
310
311
            processed_inputs=processed_inputs,
            params=params,
            arrival_time=arrival_time,
            lora_request=lora_request,
312
            trace_headers=trace_headers,
313
        )
314

315
    async def check_health_async(self) -> None:
316
317
        if self.tokenizer:
            self.tokenizer.check_health()
318
        self.model_executor.check_health()
319

320

321
class AsyncLLMEngine:
322
    """An asynchronous wrapper for :class:`LLMEngine`.
323

324
325
326
327
328
    This class is used to wrap the :class:`LLMEngine` class to make it
    asynchronous. It uses asyncio to create a background loop that keeps
    processing incoming requests. The :class:`LLMEngine` is kicked by the
    generate method when there are requests in the waiting queue. The generate
    method yields the outputs from the :class:`LLMEngine` to the caller.
329
330
331
332
333

    Args:
        worker_use_ray: Whether to use Ray for model workers. Required for
            distributed execution. Should be the same as
            `parallel_config.worker_use_ray`.
Zhuohan Li's avatar
Zhuohan Li committed
334
        engine_use_ray: Whether to make LLMEngine a Ray actor. If so, the
335
336
            async frontend will be executed in a separate process as the
            model workers.
337
        log_requests: Whether to log the requests.
zspo's avatar
zspo committed
338
339
        max_log_len: Maximum number of prompt characters or prompt ID numbers
            being printed in log.
340
341
        start_engine_loop: If True, the background task to run the engine
            will be automatically started in the generate call.
342
343
        *args: Arguments for :class:`LLMEngine`.
        **kwargs: Arguments for :class:`LLMEngine`.
344
    """
345

Antoni Baum's avatar
Antoni Baum committed
346
347
    _engine_class: Type[_AsyncLLMEngine] = _AsyncLLMEngine

348
349
350
351
352
    def __init__(self,
                 worker_use_ray: bool,
                 engine_use_ray: bool,
                 *args,
                 log_requests: bool = True,
353
                 max_log_len: Optional[int] = None,
354
                 start_engine_loop: bool = True,
355
                 **kwargs) -> None:
356
        self.worker_use_ray = worker_use_ray
Zhuohan Li's avatar
Zhuohan Li committed
357
        self.engine_use_ray = engine_use_ray
358
        self.log_requests = log_requests
359
        self.max_log_len = max_log_len
Antoni Baum's avatar
Antoni Baum committed
360
361
        self.engine = self._init_engine(*args, **kwargs)

362
        self.background_loop: Optional[asyncio.Future] = None
363
364
365
        # We need to keep a reference to unshielded
        # task as well to prevent it from being garbage
        # collected
366
        self._background_loop_unshielded: Optional[asyncio.Task] = None
367
        self.start_engine_loop = start_engine_loop
368
        self._errored_with: Optional[BaseException] = None
Antoni Baum's avatar
Antoni Baum committed
369

370
371
372
        # Lazy initialized fields
        self._request_tracker: RequestTracker

373
    @classmethod
yhu422's avatar
yhu422 committed
374
375
376
377
378
379
    def from_engine_args(
        cls,
        engine_args: AsyncEngineArgs,
        start_engine_loop: bool = True,
        usage_context: UsageContext = UsageContext.ENGINE_CONTEXT,
    ) -> "AsyncLLMEngine":
380
381
        """Creates an async LLM engine from the engine arguments."""
        # Create the engine configs.
382
        engine_config = engine_args.create_engine_config()
383
384
        distributed_executor_backend = (
            engine_config.parallel_config.distributed_executor_backend)
385

386
        if engine_config.device_config.device_type == "neuron":
387
388
            from vllm.executor.neuron_executor import NeuronExecutorAsync
            executor_class = NeuronExecutorAsync
389
390
391
        elif engine_config.device_config.device_type == "tpu":
            from vllm.executor.tpu_executor import TPUExecutorAsync
            executor_class = TPUExecutorAsync
392
        elif engine_config.device_config.device_type == "cpu":
393
394
            assert distributed_executor_backend is None, (
                "Distributed execution is not supported with the CPU backend.")
395
396
            from vllm.executor.cpu_executor import CPUExecutorAsync
            executor_class = CPUExecutorAsync
397
398
399
400
401
402
        elif engine_config.device_config.device_type == "openvino":
            assert distributed_executor_backend is None, (
                "Distributed execution is not supported with "
                "the OpenVINO backend.")
            from vllm.executor.openvino_executor import OpenVINOExecutorAsync
            executor_class = OpenVINOExecutorAsync
403
404
405
406
407
408
409
410
411
412
413
        elif engine_config.device_config.device_type == "xpu":
            if distributed_executor_backend is None:
                from vllm.executor.xpu_executor import XPUExecutorAsync
                executor_class = XPUExecutorAsync
            elif distributed_executor_backend == "ray":
                initialize_ray_cluster(engine_config.parallel_config)
                from vllm.executor.ray_xpu_executor import RayXPUExecutorAsync
                executor_class = RayXPUExecutorAsync
            else:
                raise RuntimeError(
                    "Not supported distributed execution model on XPU device.")
414
        elif distributed_executor_backend == "ray":
415
            initialize_ray_cluster(engine_config.parallel_config)
416
417
            from vllm.executor.ray_gpu_executor import RayGPUExecutorAsync
            executor_class = RayGPUExecutorAsync
418
419
420
421
        elif distributed_executor_backend == "mp":
            from vllm.executor.multiproc_gpu_executor import (
                MultiprocessingGPUExecutorAsync)
            executor_class = MultiprocessingGPUExecutorAsync
422
423
424
425
        else:
            from vllm.executor.gpu_executor import GPUExecutorAsync
            executor_class = GPUExecutorAsync
        # Create the async LLM engine.
yhu422's avatar
yhu422 committed
426
        engine = cls(
427
            distributed_executor_backend == "ray",
yhu422's avatar
yhu422 committed
428
            engine_args.engine_use_ray,
429
430
            **engine_config.to_dict(),
            executor_class=executor_class,
yhu422's avatar
yhu422 committed
431
432
433
434
435
436
            log_requests=not engine_args.disable_log_requests,
            log_stats=not engine_args.disable_log_stats,
            max_log_len=engine_args.max_log_len,
            start_engine_loop=start_engine_loop,
            usage_context=usage_context,
        )
437
438
        return engine

439
440
    @property
    def is_running(self) -> bool:
441
        return (self.background_loop is not None
442
                and self._background_loop_unshielded is not None
443
444
445
446
                and not self._background_loop_unshielded.done())

    @property
    def is_stopped(self) -> bool:
447
448
        return self.errored or (self.background_loop is not None and
                                self._background_loop_unshielded is not None
449
450
451
452
453
454
455
456
457
458
459
460
                                and self._background_loop_unshielded.done())

    @property
    def errored(self) -> bool:
        return self._errored_with is not None

    def set_errored(self, exc: Exception) -> None:
        self._errored_with = exc

    def _error_callback(self, exc: Exception) -> None:
        self.set_errored(exc)
        self._request_tracker.propagate_exception(exc)
461

462
463
    async def get_tokenizer(self) -> "PreTrainedTokenizer":
        if self.engine_use_ray:
464
            return await self.engine.get_tokenizer.remote()  # type: ignore
465
466
        else:
            return self.engine.get_tokenizer()
467

468
    def start_background_loop(self) -> None:
Antoni Baum's avatar
Antoni Baum committed
469
        """Start the background loop."""
470
471
472
        if self.errored:
            raise AsyncEngineDeadError(
                "Background loop has errored already.") from self._errored_with
473
        if self.is_running:
Antoni Baum's avatar
Antoni Baum committed
474
            raise RuntimeError("Background loop is already running.")
475
476
        # Initialize the RequestTracker here so it uses the right event loop.
        self._request_tracker = RequestTracker()
477
478
479
480

        self._background_loop_unshielded = asyncio.get_event_loop(
        ).create_task(self.run_engine_loop())
        self._background_loop_unshielded.add_done_callback(
481
            partial(_log_task_completion, error_callback=self._error_callback))
482
        self.background_loop = asyncio.shield(self._background_loop_unshielded)
Antoni Baum's avatar
Antoni Baum committed
483
484
485

    def _init_engine(self, *args,
                     **kwargs) -> Union[_AsyncLLMEngine, "ray.ObjectRef"]:
Zhuohan Li's avatar
Zhuohan Li committed
486
        if not self.engine_use_ray:
Antoni Baum's avatar
Antoni Baum committed
487
            engine_class = self._engine_class
488
        elif self.worker_use_ray:
Antoni Baum's avatar
Antoni Baum committed
489
            engine_class = ray.remote(num_cpus=0)(self._engine_class).remote
490
        else:
Woosuk Kwon's avatar
Woosuk Kwon committed
491
492
            # FIXME(woosuk): This is a bit hacky. Be careful when changing the
            # order of the arguments.
493
494
            cache_config = kwargs["cache_config"]
            parallel_config = kwargs["parallel_config"]
495
496
            if (parallel_config.tensor_parallel_size == 1
                    and parallel_config.pipeline_parallel_size == 1):
Woosuk Kwon's avatar
Woosuk Kwon committed
497
498
499
500
501
                num_gpus = cache_config.gpu_memory_utilization
            else:
                num_gpus = 1
            engine_class = ray.remote(num_gpus=num_gpus)(
                self._engine_class).remote
Antoni Baum's avatar
Antoni Baum committed
502
503
        return engine_class(*args, **kwargs)

504
    async def engine_step(self, virtual_engine: int) -> bool:
505
506
507
        """Kick the engine to process the waiting requests.

        Returns True if there are in-progress requests."""
508
509

        new_requests, finished_requests = (
510
            self._request_tracker.get_new_and_finished_requests())
511
512
513
514

        for new_request in new_requests:
            # Add the request into the vLLM engine's waiting queue.
            # TODO: Maybe add add_request_batch to reduce Ray overhead
515
516
            try:
                if self.engine_use_ray:
517
518
                    await self.engine.add_request.remote(  # type: ignore
                        **new_request)
519
520
521
522
523
524
525
526
527
                else:
                    await self.engine.add_request_async(**new_request)
            except ValueError as e:
                # TODO: use a vLLM specific error for failed validation
                self._request_tracker.process_exception(
                    new_request["request_id"],
                    e,
                    verbose=self.log_requests,
                )
528
529
530
531

        if finished_requests:
            await self._engine_abort(finished_requests)

Zhuohan Li's avatar
Zhuohan Li committed
532
        if self.engine_use_ray:
533
            request_outputs = await self.engine.step.remote()  # type: ignore
534
        else:
535
            request_outputs = await self.engine.step_async(virtual_engine)
536

Antoni Baum's avatar
Antoni Baum committed
537
        # Put the outputs into the corresponding streams.
538
        for request_output in request_outputs:
539
            self._request_tracker.process_request_output(
540
                request_output, verbose=self.log_requests)
Antoni Baum's avatar
Antoni Baum committed
541

542
543
        return len(request_outputs) > 0

Antoni Baum's avatar
Antoni Baum committed
544
545
    async def _engine_abort(self, request_ids: Iterable[str]):
        if self.engine_use_ray:
546
            await self.engine.abort_request.remote(request_ids)  # type: ignore
Antoni Baum's avatar
Antoni Baum committed
547
548
549
550
        else:
            self.engine.abort_request(request_ids)

    async def run_engine_loop(self):
551
552
553
554
555
556
        if self.engine_use_ray:
            pipeline_parallel_size = 1  # type: ignore
        else:
            pipeline_parallel_size = \
                self.engine.parallel_config.pipeline_parallel_size
        has_requests_in_progress = [False] * pipeline_parallel_size
Antoni Baum's avatar
Antoni Baum committed
557
        while True:
558
            if not any(has_requests_in_progress):
559
                logger.debug("Waiting for new requests...")
560
561
562
563
564
565
566
567
568
569
570
571
                # Stop the execute model loop in parallel workers until there
                # are more requests to process. This avoids waiting
                # indefinitely in torch.distributed ops which may otherwise
                # timeout, and unblocks the RPC thread in the workers so that
                # they can process any other queued control plane messages,
                # such as add/remove lora adapters.
                if self.engine_use_ray:
                    await (self.engine.stop_remote_worker_execution_loop.
                           remote()  # type: ignore
                           )
                else:
                    await self.engine.stop_remote_worker_execution_loop_async()
572
                await self._request_tracker.wait_for_new_requests()
573
                logger.debug("Got new requests!")
574
575
576
577
578
                requests_in_progress = [
                    asyncio.create_task(self.engine_step(ve))
                    for ve in range(pipeline_parallel_size)
                ]
                has_requests_in_progress = [True] * pipeline_parallel_size
579
580
581
582

            # Abort if iteration takes too long due to unrecoverable errors
            # (eg. NCCL timeouts).
            try:
583
                async with asyncio_timeout(ENGINE_ITERATION_TIMEOUT_S):
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
                    done, _ = await asyncio.wait(
                        requests_in_progress,
                        return_when=asyncio.FIRST_COMPLETED)
                    for _ in range(pipeline_parallel_size):
                        await asyncio.sleep(0)
                for task in done:
                    result = task.result()
                    virtual_engine = requests_in_progress.index(task)
                    if self.engine_use_ray:
                        has_unfinished_requests = (
                            await (self.engine.
                                   has_unfinished_requests_for_virtual_engine.
                                   remote(  # type: ignore
                                       virtual_engine)))
                    else:
                        has_unfinished_requests = (
                            self.engine.
                            has_unfinished_requests_for_virtual_engine(
                                virtual_engine))
                    if result or has_unfinished_requests:
                        requests_in_progress[virtual_engine] = (
                            asyncio.create_task(
                                self.engine_step(virtual_engine)))
                        has_requests_in_progress[virtual_engine] = True
                    else:
                        has_requests_in_progress[virtual_engine] = False
610
611
612
613
614
            except asyncio.TimeoutError as exc:
                logger.error(
                    "Engine iteration timed out. This should never happen!")
                self.set_errored(exc)
                raise
Antoni Baum's avatar
Antoni Baum committed
615
616
617
618
619
            await asyncio.sleep(0)

    async def add_request(
        self,
        request_id: str,
620
        inputs: PromptInputs,
621
        params: Union[SamplingParams, PoolingParams],
Antoni Baum's avatar
Antoni Baum committed
622
        arrival_time: Optional[float] = None,
623
        lora_request: Optional[LoRARequest] = None,
624
        trace_headers: Optional[Dict[str, str]] = None,
Antoni Baum's avatar
Antoni Baum committed
625
626
    ) -> AsyncStream:
        if self.log_requests:
627
628
629
630
631
632
633
634
635
            if isinstance(inputs, str):
                shortened_prompt = inputs
                shortened_token_ids = None
            else:
                shortened_prompt = inputs.get("prompt")
                shortened_token_ids = inputs.get("prompt_token_ids")

            max_log_len = self.max_log_len
            if max_log_len is not None:
636
                if shortened_prompt is not None:
637
                    shortened_prompt = shortened_prompt[:max_log_len]
638
                if shortened_token_ids is not None:
639
640
                    shortened_token_ids = shortened_token_ids[:max_log_len]

641
642
            logger.info(
                "Received request %s: prompt: %r, "
643
644
645
                "params: %s, prompt_token_ids: %s, "
                "lora_request: %s.", request_id, shortened_prompt, params,
                shortened_token_ids, lora_request)
Antoni Baum's avatar
Antoni Baum committed
646

647
        if not self.is_running:
648
649
650
651
652
653
654
655
            if self.start_engine_loop:
                self.start_background_loop()
            else:
                raise AsyncEngineDeadError(
                    "Background loop is not running. If it was running, "
                    "inspect the output to find the stacktrace of the "
                    "error that caused the background loop to stop "
                    "(AsyncEngineDeadError).")
Antoni Baum's avatar
Antoni Baum committed
656

657
658
        if arrival_time is None:
            arrival_time = time.time()
659

660
        stream = self._request_tracker.add_request(
661
            request_id,
662
            inputs=inputs,
663
            params=params,
664
            arrival_time=arrival_time,
665
            lora_request=lora_request,
666
            trace_headers=trace_headers,
667
        )
Antoni Baum's avatar
Antoni Baum committed
668
669

        return stream
670

671
    async def generate(
672
        self,
673
        inputs: PromptInputs,
674
675
        sampling_params: SamplingParams,
        request_id: str,
676
        lora_request: Optional[LoRARequest] = None,
677
        trace_headers: Optional[Dict[str, str]] = None,
678
    ) -> AsyncIterator[RequestOutput]:
679
680
681
        """Generate outputs for a request.

        Generate outputs for a request. This method is a coroutine. It adds the
682
683
        request into the waiting queue of the LLMEngine and streams the outputs
        from the LLMEngine to the caller.
684
685

        Args:
686
687
688
            inputs: The inputs to the LLM. See
                :class:`~vllm.inputs.PromptInputs`
                for more details about the format of each input.
689
690
            sampling_params: The sampling parameters of the request.
            request_id: The unique id of the request.
691
            lora_request: LoRA request to use for generation, if any.
692
            trace_headers: OpenTelemetry trace headers.
693
694

        Yields:
695
696
            The output `RequestOutput` objects from the LLMEngine
            for the request.
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739

        Details:
            - If the engine is not running, start the background loop,
              which iteratively invokes
              :meth:`~vllm.engine.async_llm_engine.AsyncLLMEngine.engine_step`
              to process the waiting requests.
            - Add the request to the engine's `RequestTracker`.
              On the next background loop, this request will be sent to
              the underlying engine.
              Also, a corresponding `AsyncStream` will be created.
            - Wait for the request outputs from `AsyncStream` and yield them.

        Example:
            >>> # Please refer to entrypoints/api_server.py for
            >>> # the complete example.
            >>>
            >>> # initialize the engine and the example input
            >>> engine = AsyncLLMEngine.from_engine_args(engine_args)
            >>> example_input = {
            >>>     "prompt": "What is LLM?",
            >>>     "stream": False, # assume the non-streaming case
            >>>     "temperature": 0.0,
            >>>     "request_id": 0,
            >>> }
            >>>
            >>> # start the generation
            >>> results_generator = engine.generate(
            >>>    example_input["prompt"],
            >>>    SamplingParams(temperature=example_input["temperature"]),
            >>>    example_input["request_id"])
            >>>
            >>> # get the results
            >>> final_output = None
            >>> async for request_output in results_generator:
            >>>     if await request.is_disconnected():
            >>>         # Abort the request if the client disconnects.
            >>>         await engine.abort(request_id)
            >>>         # Return or raise an error
            >>>         ...
            >>>     final_output = request_output
            >>>
            >>> # Process and return the final output
            >>> ...
740
        """
741
        async for output in self._process_request(
742
                request_id,
743
                inputs,
744
                sampling_params,
745
                lora_request=lora_request,
746
                trace_headers=trace_headers,
747
        ):
748
            yield LLMEngine.validate_output(output, RequestOutput)
749
750
751

    async def encode(
        self,
752
        inputs: PromptInputs,
753
754
755
        pooling_params: PoolingParams,
        request_id: str,
        lora_request: Optional[LoRARequest] = None,
756
        trace_headers: Optional[Dict[str, str]] = None,
757
758
759
760
761
762
763
764
    ) -> AsyncIterator[EmbeddingRequestOutput]:
        """Generate outputs for a request from an embedding model.

        Generate outputs for a request. This method is a coroutine. It adds the
        request into the waiting queue of the LLMEngine and streams the outputs
        from the LLMEngine to the caller.

        Args:
765
766
767
            inputs: The inputs to the LLM. See
                :class:`~vllm.inputs.PromptInputs`
                for more details about the format of each input.
768
769
770
            pooling_params: The pooling parameters of the request.
            request_id: The unique id of the request.
            lora_request: LoRA request to use for generation, if any.
771
            trace_headers: OpenTelemetry trace headers.
772
773

        Yields:
774
            The output `EmbeddingRequestOutput` objects from the LLMEngine
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
            for the request.

        Details:
            - If the engine is not running, start the background loop,
              which iteratively invokes
              :meth:`~vllm.engine.async_llm_engine.AsyncLLMEngine.engine_step`
              to process the waiting requests.
            - Add the request to the engine's `RequestTracker`.
              On the next background loop, this request will be sent to
              the underlying engine.
              Also, a corresponding `AsyncStream` will be created.
            - Wait for the request outputs from `AsyncStream` and yield them.

        Example:
            >>> # Please refer to entrypoints/api_server.py for
            >>> # the complete example.
            >>>
            >>> # initialize the engine and the example input
            >>> engine = AsyncLLMEngine.from_engine_args(engine_args)
            >>> example_input = {
            >>>     "input": "What is LLM?",
            >>>     "request_id": 0,
            >>> }
            >>>
            >>> # start the generation
            >>> results_generator = engine.encode(
            >>>    example_input["input"],
            >>>    PoolingParams(),
            >>>    example_input["request_id"])
            >>>
            >>> # get the results
            >>> final_output = None
            >>> async for request_output in results_generator:
            >>>     if await request.is_disconnected():
            >>>         # Abort the request if the client disconnects.
            >>>         await engine.abort(request_id)
            >>>         # Return or raise an error
            >>>         ...
            >>>     final_output = request_output
            >>>
            >>> # Process and return the final output
            >>> ...
        """
818
        async for output in self._process_request(
819
                request_id,
820
                inputs,
821
                pooling_params,
822
                lora_request=lora_request,
823
                trace_headers=trace_headers,
824
        ):
825
            yield LLMEngine.validate_output(output, EmbeddingRequestOutput)
826

827
    async def _process_request(
828
829
        self,
        request_id: str,
830
        inputs: PromptInputs,
831
        params: Union[SamplingParams, PoolingParams],
832
        *,
833
        lora_request: Optional[LoRARequest] = None,
834
        trace_headers: Optional[Dict[str, str]] = None,
835
836
837
838
839
840
841
    ) -> AsyncIterator[Union[RequestOutput, EmbeddingRequestOutput]]:
        """Common logic to process requests with SamplingParams or
        PoolingParams."""
        arrival_time = time.time()

        stream = await self.add_request(
            request_id,
842
            inputs,
843
844
845
            params,
            arrival_time=arrival_time,
            lora_request=lora_request,
846
            trace_headers=trace_headers,
847
        )
848

849
        try:
Antoni Baum's avatar
Antoni Baum committed
850
851
            async for request_output in stream:
                yield request_output
852
        except (Exception, asyncio.CancelledError) as e:
Antoni Baum's avatar
Antoni Baum committed
853
854
            self._abort(request_id)
            raise e
855

Antoni Baum's avatar
Antoni Baum committed
856
857
    async def abort(self, request_id: str) -> None:
        """Abort a request.
858

Antoni Baum's avatar
Antoni Baum committed
859
860
        Abort a submitted request. If the request is finished or not found,
        this method will be a no-op.
861

Antoni Baum's avatar
Antoni Baum committed
862
863
864
        Args:
            request_id: The unique id of the request.
        """
865
866
867
868
869
870
871
        if not self.is_running:
            raise AsyncEngineDeadError(
                "Background loop is not running. If it was running, "
                "inspect the output to find the stacktrace of the "
                "error that caused the background loop to stop "
                "(AsyncEngineDeadError).")

Antoni Baum's avatar
Antoni Baum committed
872
        return self._abort(request_id)
873

Antoni Baum's avatar
Antoni Baum committed
874
    def _abort(self, request_id: str) -> None:
875
876
877
878
879
880
881
882
        """Abort a request.

        Abort a submitted request. If the request is finished or not found,
        this method will be a no-op.

        Args:
            request_id: The unique id of the request.
        """
883
884
        self._request_tracker.abort_request(request_id,
                                            verbose=self.log_requests)
885

886
887
888
    async def get_model_config(self) -> ModelConfig:
        """Get the model configuration of the vLLM engine."""
        if self.engine_use_ray:
889
            return await self.engine.get_model_config.remote()  # type: ignore
890
891
892
        else:
            return self.engine.get_model_config()

893
894
895
896
897
898
899
900
    async def get_decoding_config(self) -> DecodingConfig:
        """Get the decoding configuration of the vLLM engine."""
        if self.engine_use_ray:
            return await self.engine.get_decoding_config.remote(  # type: ignore
            )
        else:
            return self.engine.get_decoding_config()

901
902
903
904
    async def do_log_stats(
            self,
            scheduler_outputs: Optional[SchedulerOutputs] = None,
            model_output: Optional[List[SamplerOutput]] = None) -> None:
905
        if self.engine_use_ray:
906
907
            await self.engine.do_log_stats.remote(  # type: ignore
                scheduler_outputs, model_output)
908
909
        else:
            self.engine.do_log_stats()
910

911
    async def check_health(self) -> None:
912
913
914
915
916
917
918
919
        """Raises an error if engine is unhealthy."""
        t = time.perf_counter()
        logger.debug("Starting health check...")
        if self.is_stopped:
            raise AsyncEngineDeadError("Background loop is stopped.")

        if self.engine_use_ray:
            try:
920
                await self.engine.check_health.remote()  # type: ignore
921
922
923
924
            except ray.exceptions.RayActorError as e:
                raise RuntimeError("Engine is dead.") from e
        else:
            await self.engine.check_health_async()
925
        logger.debug("Health check took %fs", time.perf_counter() - t)
926
927
928
929
930
931
932

    async def is_tracing_enabled(self) -> bool:
        if self.engine_use_ray:
            return await self.engine.is_tracing_enabled.remote(  # type: ignore
            )
        else:
            return self.engine.is_tracing_enabled()