utils.py 21.4 KB
Newer Older
1
import asyncio
2
import datetime
Woosuk Kwon's avatar
Woosuk Kwon committed
3
import enum
4
import gc
5
import os
6
import socket
7
import subprocess
8
import sys
9
10
import tempfile
import threading
Zhuohan Li's avatar
Zhuohan Li committed
11
import uuid
12
import warnings
13
from collections import defaultdict
14
from functools import lru_cache, partial, wraps
15
from platform import uname
16
from typing import (Any, AsyncIterator, Awaitable, Callable, Dict, Generic,
17
18
                    Hashable, List, Optional, OrderedDict, Tuple, TypeVar,
                    Union)
Zhuohan Li's avatar
Zhuohan Li committed
19

20
import numpy as np
21
import psutil
Zhuohan Li's avatar
Zhuohan Li committed
22
import torch
23

24
import vllm.envs as envs
25
from vllm import _custom_ops as ops
26
from vllm.logger import enable_trace_function_call, init_logger
27

28
T = TypeVar("T")
29
30
31
32
33
34
logger = init_logger(__name__)

STR_DTYPE_TO_TORCH_DTYPE = {
    "half": torch.half,
    "bfloat16": torch.bfloat16,
    "float": torch.float,
35
    "fp8": torch.uint8,
36
37
    "fp8_e4m3": torch.uint8,
    "fp8_e5m2": torch.uint8,
38
}
Zhuohan Li's avatar
Zhuohan Li committed
39

Woosuk Kwon's avatar
Woosuk Kwon committed
40
41
42
43
44
45
46
47
48
49
50

class Device(enum.Enum):
    GPU = enum.auto()
    CPU = enum.auto()


class Counter:

    def __init__(self, start: int = 0) -> None:
        self.counter = start

Woosuk Kwon's avatar
Woosuk Kwon committed
51
    def __next__(self) -> int:
52
        i = self.counter
Woosuk Kwon's avatar
Woosuk Kwon committed
53
        self.counter += 1
54
        return i
Woosuk Kwon's avatar
Woosuk Kwon committed
55
56
57

    def reset(self) -> None:
        self.counter = 0
Zhuohan Li's avatar
Zhuohan Li committed
58

59

60
class LRUCache(Generic[T]):
61
62

    def __init__(self, capacity: int):
63
        self.cache: OrderedDict[Hashable, T] = OrderedDict()
64
65
66
67
68
69
70
71
        self.capacity = capacity

    def __contains__(self, key: Hashable) -> bool:
        return key in self.cache

    def __len__(self) -> int:
        return len(self.cache)

72
    def __getitem__(self, key: Hashable) -> Optional[T]:
73
74
        return self.get(key)

75
    def __setitem__(self, key: Hashable, value: T) -> None:
76
77
78
79
80
81
82
83
        self.put(key, value)

    def __delitem__(self, key: Hashable) -> None:
        self.pop(key)

    def touch(self, key: Hashable) -> None:
        self.cache.move_to_end(key)

84
85
86
    def get(self,
            key: Hashable,
            default_value: Optional[T] = None) -> Optional[T]:
87
        if key in self.cache:
88
            value: Optional[T] = self.cache[key]
89
90
91
92
93
            self.cache.move_to_end(key)
        else:
            value = default_value
        return value

94
    def put(self, key: Hashable, value: T) -> None:
95
96
97
98
        self.cache[key] = value
        self.cache.move_to_end(key)
        self._remove_old_if_needed()

99
    def _on_remove(self, key: Hashable, value: Optional[T]):
100
101
102
103
104
105
106
107
108
109
110
111
        pass

    def remove_oldest(self):
        if not self.cache:
            return
        key, value = self.cache.popitem(last=False)
        self._on_remove(key, value)

    def _remove_old_if_needed(self) -> None:
        while len(self.cache) > self.capacity:
            self.remove_oldest()

112
113
114
    def pop(self,
            key: Hashable,
            default_value: Optional[T] = None) -> Optional[T]:
115
        run_on_remove = key in self.cache
116
        value: Optional[T] = self.cache.pop(key, default_value)
117
118
119
120
121
122
123
124
125
126
        if run_on_remove:
            self._on_remove(key, value)
        return value

    def clear(self):
        while len(self.cache) > 0:
            self.remove_oldest()
        self.cache.clear()


127
128
129
130
def is_hip() -> bool:
    return torch.version.hip is not None


131
132
@lru_cache(maxsize=None)
def is_cpu() -> bool:
133
134
135
136
137
    from importlib.metadata import PackageNotFoundError, version
    try:
        return "cpu" in version("vllm")
    except PackageNotFoundError:
        return False
138
139


140
@lru_cache(maxsize=None)
141
142
143
144
145
146
147
148
def is_neuron() -> bool:
    try:
        import transformers_neuronx
    except ImportError:
        transformers_neuronx = None
    return transformers_neuronx is not None


149
150
151
152
153
154
155
156
157
@lru_cache(maxsize=None)
def is_tpu() -> bool:
    try:
        import libtpu
    except ImportError:
        libtpu = None
    return libtpu is not None


158
@lru_cache(maxsize=None)
159
160
def get_max_shared_memory_bytes(gpu: int = 0) -> int:
    """Returns the maximum shared memory per thread block in bytes."""
161
    max_shared_mem = (
162
        ops.get_max_shared_memory_per_block_device_attribute(gpu))
163
164
    # value 0 will cause MAX_SEQ_LEN become negative and test_attention.py
    # will fail
165
    assert max_shared_mem > 0, "max_shared_mem can not be zero"
166
167
168
    return int(max_shared_mem)


169
def get_cpu_memory() -> int:
170
    """Returns the total CPU memory of the node in bytes."""
171
    return psutil.virtual_memory().total
Zhuohan Li's avatar
Zhuohan Li committed
172
173
174
175


def random_uuid() -> str:
    return str(uuid.uuid4().hex)
176

177

178
179
180
181
182
183
184
185
@lru_cache(maxsize=None)
def get_vllm_instance_id():
    """
    If the environment variable VLLM_INSTANCE_ID is set, return it.
    Otherwise, return a random UUID.
    Instance id represents an instance of the VLLM. All processes in the same
    instance should have the same instance id.
    """
186
    return envs.VLLM_INSTANCE_ID or f"vllm-instance-{random_uuid()}"
187
188


189
@lru_cache(maxsize=None)
190
191
192
def in_wsl() -> bool:
    # Reference: https://github.com/microsoft/WSL/issues/4071
    return "microsoft" in " ".join(uname()).lower()
193
194


195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
def make_async(func: Callable[..., T]) -> Callable[..., Awaitable[T]]:
    """Take a blocking function, and run it on in an executor thread.

    This function prevents the blocking function from blocking the
    asyncio event loop.
    The code in this function needs to be thread safe.
    """

    def _async_wrapper(*args, **kwargs) -> asyncio.Future:
        loop = asyncio.get_event_loop()
        p_func = partial(func, *args, **kwargs)
        return loop.run_in_executor(executor=None, func=p_func)

    return _async_wrapper


211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
def merge_async_iterators(
        *iterators: AsyncIterator[T]) -> AsyncIterator[Tuple[int, T]]:
    """Merge multiple asynchronous iterators into a single iterator.

    This method handle the case where some iterators finish before others.
    When it yields, it yields a tuple (i, item) where i is the index of the
    iterator that yields the item.
    """
    queue: asyncio.Queue[Union[Tuple[int, T], Exception]] = asyncio.Queue()

    finished = [False] * len(iterators)

    async def producer(i: int, iterator: AsyncIterator[T]):
        try:
            async for item in iterator:
                await queue.put((i, item))
        except Exception as e:
            await queue.put(e)
        finished[i] = True

    _tasks = [
        asyncio.create_task(producer(i, iterator))
        for i, iterator in enumerate(iterators)
    ]

    async def consumer():
237
238
239
240
241
242
243
244
        try:
            while not all(finished) or not queue.empty():
                item = await queue.get()
                if isinstance(item, Exception):
                    raise item
                yield item
        except (Exception, asyncio.CancelledError) as e:
            for task in _tasks:
245
246
247
248
249
                if sys.version_info >= (3, 9):
                    # msg parameter only supported in Python 3.9+
                    task.cancel(e)
                else:
                    task.cancel()
250
            raise e
251
252
253
254
255
        await asyncio.gather(*_tasks)

    return consumer()


256
def get_ip() -> str:
257
    host_ip = envs.VLLM_HOST_IP
258
259
260
261
262
    if host_ip:
        return host_ip

    # IP is not set, try to get it from the network interface

263
    # try ipv4
264
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
265
    try:
266
        s.connect(("8.8.8.8", 80))  # Doesn't need to be reachable
267
        return s.getsockname()[0]
268
269
270
271
272
    except Exception:
        pass

    # try ipv6
    try:
273
        s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
274
275
276
        # Google's public DNS server, see
        # https://developers.google.com/speed/public-dns/docs/using#addresses
        s.connect(("2001:4860:4860::8888", 80))  # Doesn't need to be reachable
277
        return s.getsockname()[0]
278
279
280
281
282
    except Exception:
        pass

    warnings.warn(
        "Failed to get the IP address, using 0.0.0.0 by default."
283
284
        "The value can be set by the environment variable"
        " VLLM_HOST_IP or HOST_IP.",
285
286
        stacklevel=2)
    return "0.0.0.0"
287
288


289
def get_distributed_init_method(ip: str, port: int) -> str:
290
291
292
    # Brackets are not permitted in ipv4 addresses,
    # see https://github.com/python/cpython/issues/103848
    return f"tcp://[{ip}]:{port}" if ":" in ip else f"tcp://{ip}:{port}"
293
294


295
def get_open_port() -> int:
296
297
    port = envs.VLLM_PORT
    if port is not None:
298
299
300
301
302
303
304
305
306
        while True:
            try:
                with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                    s.bind(("", port))
                    return port
            except OSError:
                port += 1  # Increment port number if already in use
                logger.info("Port %d is already in use, trying port %d",
                            port - 1, port)
307
308
309
310
311
312
313
314
315
316
    # try ipv4
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.bind(("", 0))
            return s.getsockname()[1]
    except OSError:
        # try ipv6
        with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s:
            s.bind(("", 0))
            return s.getsockname()[1]
317
318


319
320
def update_environment_variables(envs: Dict[str, str]):
    for k, v in envs.items():
321
        if k in os.environ and os.environ[k] != v:
322
323
324
            logger.warning(
                "Overwriting environment variable %s "
                "from '%s' to '%s'", k, os.environ[k], v)
325
        os.environ[k] = v
326
327


328
329
330
331
332
333
334
335
336
337
def chunk_list(lst, chunk_size):
    """Yield successive chunk_size chunks from lst."""
    return [lst[i:i + chunk_size] for i in range(0, len(lst), chunk_size)]


def cdiv(a: int, b: int) -> int:
    """Ceiling division."""
    return -(a // -b)


338
def _generate_random_fp8(
339
340
341
342
343
344
345
    tensor: torch.tensor,
    low: float,
    high: float,
) -> None:
    # NOTE(zhaoyang): Due to NaN and Inf representation for fp8 data type,
    # it may occur Inf or NaN if we directly use torch.randint
    # to generate random data for fp8 data.
346
    # For example, s.11111.00 in fp8e5m2 format represents Inf.
347
348
349
350
    #     | E4M3        | E5M2
    #-----|-------------|-------------------
    # Inf | N/A         | s.11111.00
    # NaN | s.1111.111  | s.11111.{01,10,11}
351
    from vllm import _custom_ops as ops
352
353
    tensor_tmp = torch.empty_like(tensor, dtype=torch.float16)
    tensor_tmp.uniform_(low, high)
354
    ops.convert_fp8(tensor, tensor_tmp)
355
356
357
    del tensor_tmp


358
359
360
def get_kv_cache_torch_dtype(
        cache_dtype: Optional[Union[str, torch.dtype]],
        model_dtype: Optional[Union[str, torch.dtype]] = None) -> torch.dtype:
361
362
363
364
365
366
367
368
369
370
    if isinstance(cache_dtype, str):
        if cache_dtype == "auto":
            if isinstance(model_dtype, str):
                torch_dtype = STR_DTYPE_TO_TORCH_DTYPE[model_dtype]
            elif isinstance(model_dtype, torch.dtype):
                torch_dtype = model_dtype
            else:
                raise ValueError(f"Invalid model dtype: {model_dtype}")
        elif cache_dtype in ["half", "bfloat16", "float"]:
            torch_dtype = STR_DTYPE_TO_TORCH_DTYPE[cache_dtype]
371
        elif cache_dtype == "fp8":
372
373
374
375
376
377
378
            torch_dtype = torch.uint8
        else:
            raise ValueError(f"Invalid kv cache dtype: {cache_dtype}")
    elif isinstance(cache_dtype, torch.dtype):
        torch_dtype = cache_dtype
    else:
        raise ValueError(f"Invalid kv cache dtype: {cache_dtype}")
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
    return torch_dtype


def create_kv_caches_with_random_flash(
    num_blocks: int,
    block_size: int,
    num_layers: int,
    num_heads: int,
    head_size: int,
    cache_dtype: Optional[Union[str, torch.dtype]],
    model_dtype: Optional[Union[str, torch.dtype]] = None,
    seed: int = 0,
    device: Optional[str] = "cuda",
) -> Tuple[List[torch.Tensor], List[torch.Tensor]]:
    assert cache_dtype != "fp8"
    torch.random.manual_seed(seed)
    if torch.cuda.is_available():
        torch.cuda.manual_seed(seed)

    torch_dtype = get_kv_cache_torch_dtype(cache_dtype, model_dtype)
    key_value_cache_shape = (num_blocks, 2, block_size, num_heads, head_size)
    scale = head_size**-0.5
    key_caches, value_caches = [], []
    for _ in range(num_layers):
        key_value_cache = torch.empty(size=key_value_cache_shape,
                                      dtype=torch_dtype,
                                      device=device)
        key_value_cache.uniform_(-scale, scale)
        key_caches.append(key_value_cache[:, 0])
        value_caches.append(key_value_cache[:, 1])
    return key_caches, value_caches


def create_kv_caches_with_random(
    num_blocks: int,
    block_size: int,
    num_layers: int,
    num_heads: int,
    head_size: int,
    cache_dtype: Optional[Union[str, torch.dtype]],
    model_dtype: Optional[Union[str, torch.dtype]] = None,
    seed: int = 0,
    device: Optional[str] = "cuda",
) -> Tuple[List[torch.Tensor], List[torch.Tensor]]:
    torch.random.manual_seed(seed)
    if torch.cuda.is_available():
        torch.cuda.manual_seed(seed)

    torch_dtype = get_kv_cache_torch_dtype(cache_dtype, model_dtype)
428
429
430
431
432
433
434
435
436

    scale = head_size**-0.5
    x = 16 // torch.tensor([], dtype=torch_dtype).element_size()
    key_cache_shape = (num_blocks, num_heads, head_size // x, block_size, x)
    key_caches = []
    for _ in range(num_layers):
        key_cache = torch.empty(size=key_cache_shape,
                                dtype=torch_dtype,
                                device=device)
437
        if cache_dtype in ["auto", "half", "bfloat16", "float"]:
438
            key_cache.uniform_(-scale, scale)
439
440
        elif cache_dtype == 'fp8':
            _generate_random_fp8(key_cache, -scale, scale)
441
442
443
        else:
            raise ValueError(
                f"Does not support key cache of type {cache_dtype}")
444
445
446
447
448
449
450
451
        key_caches.append(key_cache)

    value_cache_shape = (num_blocks, num_heads, head_size, block_size)
    value_caches = []
    for _ in range(num_layers):
        value_cache = torch.empty(size=value_cache_shape,
                                  dtype=torch_dtype,
                                  device=device)
452
        if cache_dtype in ["auto", "half", "bfloat16", "float"]:
453
            value_cache.uniform_(-scale, scale)
454
455
        elif cache_dtype == 'fp8':
            _generate_random_fp8(value_cache, -scale, scale)
456
457
458
        else:
            raise ValueError(
                f"Does not support value cache of type {cache_dtype}")
459
460
        value_caches.append(value_cache)
    return key_caches, value_caches
461
462


463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
@lru_cache
def print_warning_once(msg: str) -> None:
    logger.warning(msg)


@lru_cache(maxsize=None)
def is_pin_memory_available() -> bool:

    if in_wsl():
        # Pinning memory in WSL is not supported.
        # https://docs.nvidia.com/cuda/wsl-user-guide/index.html#known-limitations-for-linux-cuda-applications
        print_warning_once("Using 'pin_memory=False' as WSL is detected. "
                           "This may slow down the performance.")
        return False
    elif is_neuron():
        print_warning_once("Pin memory is not supported on Neuron.")
        return False
480
481
    elif is_cpu():
        return False
482
483
484
485
    return True


class CudaMemoryProfiler:
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506

    def __init__(self, device=None):
        self.device = device

    def current_memory_usage(self) -> float:
        # Return the memory usage in bytes.
        torch.cuda.reset_peak_memory_stats(self.device)
        mem = torch.cuda.max_memory_allocated(self.device)
        return mem

    def __enter__(self):
        self.initial_memory = self.current_memory_usage()
        # This allows us to call methods of the context manager if needed
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.final_memory = self.current_memory_usage()
        self.consumed_memory = self.final_memory - self.initial_memory

        # Force garbage collection
        gc.collect()
507
508


509
def str_to_int_tuple(s: str) -> Tuple[int, ...]:
510
511
512
513
514
515
516
517
518
    """Convert a string to a tuple of integers."""
    try:
        return tuple(map(int, s.split(",")))
    except ValueError as e:
        raise ValueError(
            "String must be a series of integers separated by commas "
            f"(e.g., 1, 2, 3). Given input: {s}") from e


519
520
521
522
523
524
525
526
527
528
529
530
def make_tensor_with_pad(
    x: List[List[int]],
    max_len: int,
    pad: int,
    dtype: torch.dtype,
    device: Optional[Union[str, torch.device]],
) -> torch.Tensor:
    """Make a padded tensor of a 2D inputs.

    The padding is applied to the end of each inner list until it reaches
    `max_len`.
    """
531
532
533
534
    padded_x = np.zeros([len(x), max_len], dtype=np.int32) + pad
    for ind, blocktb in enumerate(x):
        assert len(blocktb) <= max_len
        padded_x[ind, :len(blocktb)] = blocktb
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
    return torch.tensor(padded_x, dtype=dtype, device=device)


def async_tensor_h2d(
    data: list,
    dtype: torch.dtype,
    target_device: Union[str, torch.device],
    pin_memory: bool,
) -> torch.Tensor:
    """Asynchronously create a tensor and copy it from host to device."""
    t = torch.tensor(data, dtype=dtype, pin_memory=pin_memory, device="cpu")
    return t.to(device=target_device, non_blocking=True)


def maybe_expand_dim(tensor: torch.Tensor,
                     target_dims: int,
                     size: int = 1) -> torch.Tensor:
    """Expand the tensor to the target_dims."""
    if tensor.ndim < target_dims:
        tensor = tensor.view(-1, *([size] * (target_dims - tensor.ndim)))
    return tensor
556
557


558
559
560
561
562
def get_dtype_size(dtype: torch.dtype) -> int:
    """Get the size of the data type in bytes."""
    return torch.tensor([], dtype=dtype).element_size()


563
564
def merge_dicts(dict1: Dict[Any, List[Any]],
                dict2: Dict[Any, List[Any]]) -> Dict[Any, List[Any]]:
565
    """Merge 2 dicts that have key -> List of items.
566

567
568
569
570
571
572
573
574
575
576
577
    When a key conflicts, the values in dict1 is prioritized.
    """
    merged_dict = defaultdict(list)

    for key, value in dict1.items():
        merged_dict[key].extend(value)

    for key, value in dict2.items():
        merged_dict[key].extend(value)

    return dict(merged_dict)
578
579
580
581
582
583
584
585


def init_cached_hf_modules():
    """
    Lazy initialization of the Hugging Face modules.
    """
    from transformers.dynamic_module_utils import init_hf_modules
    init_hf_modules()
586
587


588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
@lru_cache(maxsize=None)
def find_library(lib_name: str) -> str:
    """
    Find the library file in the system.
    `lib_name` is full filename, with both prefix and suffix.
    This function resolves `lib_name` to the full path of the library.
    """
    # Adapted from https://github.com/openai/triton/blob/main/third_party/nvidia/backend/driver.py#L19 # noqa
    # According to https://en.wikipedia.org/wiki/Filesystem_Hierarchy_Standard
    # `/sbin/ldconfig` should exist in all Linux systems.
    # `/sbin/ldconfig` searches the library in the system
    libs = subprocess.check_output(["/sbin/ldconfig", "-p"]).decode()
    # each line looks like the following:
    # libcuda.so.1 (libc6,x86-64) => /lib/x86_64-linux-gnu/libcuda.so.1
    locs = [line.split()[-1] for line in libs.splitlines() if lib_name in line]
    # `LD_LIBRARY_PATH` searches the library in the user-defined paths
604
    env_ld_library_path = envs.LD_LIBRARY_PATH
605
606
607
608
609
610
611
612
613
614
615
    if not locs and env_ld_library_path:
        locs = [
            os.path.join(dir, lib_name)
            for dir in env_ld_library_path.split(":")
            if os.path.exists(os.path.join(dir, lib_name))
        ]
    if not locs:
        raise ValueError(f"Cannot find {lib_name} in the system.")
    return locs[0]


616
def find_nccl_library():
617
618
619
620
621
622
    """
    We either use the library file specified by the `VLLM_NCCL_SO_PATH`
    environment variable, or we find the library file brought by PyTorch.
    After importing `torch`, `libnccl.so.2` or `librccl.so.1` can be
    found by `ctypes` automatically.
    """
623
    so_file = envs.VLLM_NCCL_SO_PATH
624
625
626
627

    # manually load the nccl library
    if so_file:
        logger.info(
628
629
            "Found nccl from environment variable VLLM_NCCL_SO_PATH=%s",
            so_file)
630
631
    else:
        if torch.version.cuda is not None:
632
            so_file = "libnccl.so.2"
633
        elif torch.version.hip is not None:
634
            so_file = "librccl.so.1"
635
636
        else:
            raise ValueError("NCCL only supports CUDA and ROCm backends.")
637
        logger.info("Found nccl from library %s", so_file)
638
    return so_file
639
640
641
642
643
644
645


def enable_trace_function_call_for_thread() -> None:
    """Set up function tracing for the current thread,
    if enabled via the VLLM_TRACE_FUNCTION environment variable
    """

646
    if envs.VLLM_TRACE_FUNCTION:
647
648
649
650
651
652
653
654
        tmp_dir = tempfile.gettempdir()
        filename = (f"VLLM_TRACE_FUNCTION_for_process_{os.getpid()}"
                    f"_thread_{threading.get_ident()}_"
                    f"at_{datetime.datetime.now()}.log").replace(" ", "_")
        log_path = os.path.join(tmp_dir, "vllm", get_vllm_instance_id(),
                                filename)
        os.makedirs(os.path.dirname(log_path), exist_ok=True)
        enable_trace_function_call(log_path)
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695


def identity(value: T) -> T:
    return value


F = TypeVar('F', bound=Callable[..., Any])


def deprecate_kwargs(
        *kws: str,
        is_deprecated: Union[bool, Callable[[], bool]] = True,
        additional_message: Optional[str] = None) -> Callable[[F], F]:
    deprecated_kws = set(kws)

    if not callable(is_deprecated):
        is_deprecated = partial(identity, is_deprecated)

    def wrapper(fn: F) -> F:

        @wraps(fn)
        def inner(*args, **kwargs):
            if is_deprecated():
                deprecated_kwargs = kwargs.keys() & deprecated_kws
                if deprecated_kwargs:
                    msg = (
                        f"The keyword arguments {deprecated_kwargs} are "
                        "deprecated and will be removed in a future update.")
                    if additional_message is not None:
                        msg += f" {additional_message}"

                    warnings.warn(
                        DeprecationWarning(msg),
                        stacklevel=3,  # The inner function takes up one level
                    )

            return fn(*args, **kwargs)

        return inner  # type: ignore

    return wrapper