U yhZ @sTUddlZddlZddlmZmZmZmZmZmZm Z m Z m Z m Z ddl mZddlmZmZmZmZmZmZddlmZe e ee efZeddddd ee eeeee eed d d ZGd ddZiZeedfed<eZejed<eeddddZde eeeefe eedddZ!e eej"efe edddZ#dS)N) AnyCallableDictIterableIteratorListOptionalSequenceTupleUnion) exposed_in)_C_library_opsautogradlibraryTensor)utilsz torch.library) device_typesschema)namefn mutates_argsrrreturncs&fdd}|dkr|S||S)a Wraps a function into custom operator. Reasons why you may want to create a custom op include: - Wrapping a third-party library or custom kernel to work with PyTorch subsystems like Autograd. - Preventing torch.compile/export/FX tracing from peeking inside your function. This API is used as a decorator around a function (please see examples). The provided function must have type hints; these are needed to interface with PyTorch's various subsystems. Args: name (str): A name for the custom op that looks like "{namespace}::{name}", e.g. "mylib::my_linear". The name is used as the op's stable identifier in PyTorch subsystems (e.g. torch.export, FX graphs). To avoid name collisions, please use your project name as the namespace; e.g. all custom ops in pytorch/fbgemm use "fbgemm" as the namespace. mutates_args (Iterable[str]): The names of args that the function mutates. This MUST be accurate, otherwise, the behavior is undefined. device_types (None | str | Sequence[str]): The device type(s) the function is valid for. If no device type is provided, then the function is used as the default implementation for all device types. Examples: "cpu", "cuda". schema (None | str): A schema string for the operator. If None (recommended) we'll infer a schema for the operator from its type annotations. We recommend letting us infer a schema unless you have a specific reason not to. Example: "(Tensor x, int y) -> (Tensor, Tensor)". .. note:: We recommend not passing in a ``schema`` arg and instead letting us infer it from the type annotations. It is error-prone to write your own schema. You may wish to provide your own schema if our interpretation of the type annotation is not what you want. For more info on how to write a schema string, see `here `_ Examples:: >>> import torch >>> from torch import Tensor >>> from torch.library import custom_op >>> import numpy as np >>> >>> @custom_op("mylib::numpy_sin", mutates_args=()) >>> def numpy_sin(x: Tensor) -> Tensor: >>> x_np = x.cpu().numpy() >>> y_np = np.sin(x_np) >>> return torch.from_numpy(y_np).to(device=x.device) >>> >>> x = torch.randn(3) >>> y = numpy_sin(x) >>> assert torch.allclose(y, x.sin()) >>> >>> # Example of a custom op that only works for one device type. >>> @custom_op("mylib::numpy_sin_cpu", mutates_args=(), device_types="cpu") >>> def numpy_sin_cpu(x: Tensor) -> Tensor: >>> x_np = x.numpy() >>> y_np = np.sin(x_np) >>> return torch.from_numpy(y_np) >>> >>> x = torch.randn(3) >>> y = numpy_sin_cpu(x) >>> assert torch.allclose(y, x.sin()) >>> >>> # Example of a custom op that mutates an input >>> @custom_op("mylib::numpy_sin_inplace", mutates_args={"x"}, device_types="cpu") >>> def numpy_sin_inplace(x: Tensor) -> None: >>> x_np = x.numpy() >>> np.sin(x_np, out=x_np) >>> >>> x = torch.randn(3) >>> expected = x.sin() >>> numpy_sin_inplace(x) >>> assert torch.allclose(x, expected) csddl}dkr*ddl}|jj|}n}d\}}t||||}dk rt}|jj j D]"}|j dk rb|j j rb| |jqb|tkrtddd|d|||S)Nr::z3Attempted to create a custom op with `mutates_args=z` and `schema=z*. The schema suggests that the op mutates z`which is different from what was provided to us in `mutates_args`. Please make these consistent.)torchZtorch._custom_op.implZ _custom_opimplZ infer_schemasplit CustomOpDefset _opoverload_schema arguments alias_infois_writeaddr ValueErrorregister_kernel)rr schema_str namespaceopnameresultexpectedargrrrrK/var/www/html/venv/lib/python3.8/site-packages/torch/_library/custom_ops.pyinnerqs$ zcustom_op..innerNr1)rrrrrr3r1r0r2 custom_opsWr4c@seZdZdZeeeeddddZeedddZedd d Z de e eed d d Z eedddZ ddee eddddZddddZddZdS)r aCustomOpDef is a wrapper around a function that turns it into a custom op. It has various methods for registering additional behavior for this custom op. You should not instantiate CustomOpDef directly; instead, use the :func:`torch.library.custom_op` API. N)r+rrrrcCsV||_||_||_||_i|_d|_d|_d|_t|j|j|_ | |t |j <dSN) _namespace_namer#_init_fn _backend_fns _abstract_fn_setup_context_fn _backward_fnget_library_allowing_overwrite_lib_register_to_dispatcherOPDEFS _qualname)selfr+rrrr1r1r2__init__szCustomOpDef.__init__)rcCs|jd|jS)Nr)r6r7rBr1r1r2rAszCustomOpDef._qualnamecCsd|jdS)Nz )rArDr1r1r2__repr__szCustomOpDef.__repr__)rrrcs"fdd}|dkr|S||S)aRegister an implementation for a device type for this operator. Some valid device_types are: "cpu", "cuda", "xla", "mps", "ipu", "xpu". This API may be used as a decorator. Args: fn (Callable): The function to register as the implementation for the given device types. device_types (str | Sequence[str]): The device device_types to register an impl to. Examples:: >>> # xdoctest: +REQUIRES(env:TORCH_DOCTEST_CUDA) >>> import torch >>> from torch import Tensor >>> from torch.library import custom_op >>> import numpy as np >>> >>> # Create a custom op that works on cpu >>> @custom_op("mylib::numpy_sin", mutates_args=(), device_types="cpu") >>> def numpy_sin(x: Tensor) -> Tensor: >>> x_np = x.numpy() >>> y_np = np.sin(x_np) >>> return torch.from_numpy(y_np) >>> >>> # Add implementations for the cuda device >>> @numpy_sin.register_kernel("cuda") >>> def _(x): >>> x_np = x.cpu().numpy() >>> y_np = np.sin(x_np) >>> return torch.from_numpy(y_np).to(device=x.device) >>> >>> x_cpu = torch.randn(3) >>> x_cuda = x_cpu.cuda() >>> assert torch.allclose(numpy_sin(x_cpu), x_cpu.sin()) >>> assert torch.allclose(numpy_sin(x_cuda), x_cuda.sin()) csdksttrg}nt}|D]Zjkrvfdd}dkr^jj|dnjj|t|j<q&|S)Nc sddt||D}j||}|}t|ts8|f}t|iD]V}t|}t||krj}t|}tdj d|d| |qB|S)NcSsh|]}t|qSr1)iduntyped_storage).0tensorr1r1r2 szSCustomOpDef.register_kernel..inner..backend_impl..zTensors returned from custom ops (1) must not be inputs to the custom op and (2) may not alias any inputs or other returns. Please clone the the offending output tensors (e.g. output.clone()) or refactor your code. Offending op: z (with implementation in )) iter_tensorsr9 isinstancetuplerFrGinspect getmodule RuntimeErrorr7r') argskwargsZstoragesr-Z tuple_resultrIkeyrmodule) device_typerBr1r2 backend_impls"     z@CustomOpDef.register_kernel..inner..backend_implZCompositeExplicitAutograd) rMstrlistr9r>rr7rZ_dispatch_key_for_device)rZdtypesrWrrB)rVr2r3s&  z*CustomOpDef.register_kernel..innerNr1)rBrrr3r1rZr2r)s)3zCustomOpDef.register_kernel)rrcCs ||_|S)a Register a FakeTensor implementation for this custom op. This is necessary to get the operator to work efficiently with torch.compile. The Fake impl (sometimes also known as a meta kernel or abstract impl) specifies the behavior of this operator on Tensors that carry no data. Given some input Tensors with certain properties (sizes/strides/storage_offset/device), it specifies what the properties of the output Tensors are. Please see :func:`torch.library.impl_abstract` for more details. Args: fn (Callable): The function to register as the FakeTensor implementation. Examples: >>> import torch >>> import numpy as np >>> from torch import Tensor >>> >>> # Example 1: an operator without data-dependent output shape >>> @torch.library.custom_op("mylib::linear", mutates_args=()) >>> def linear(x: Tensor, weight: Tensor, bias: Tensor) -> Tensor: >>> return (x @ weight.t()) + bias >>> >>> @linear.register_fake >>> def _(x, weight, bias): >>> assert x.dim() == 2 >>> assert weight.dim() == 2 >>> assert bias.dim() == 1 >>> assert x.shape[1] == weight.shape[1] >>> assert weight.shape[0] == bias.shape[0] >>> assert x.device == weight.device >>> return x.new_empty(x.size(0), weight.size(0)) >>> >>> x = torch.randn(2, 2) >>> weight = torch.randn(2, 2) >>> bias = torch.randn(2) >>> # xdoctest: +SKIP("Requires Python <= 3.11") >>> out = torch.compile(linear, fullgraph=True)(x, weight, bias) >>> # xdoctest: +SKIP("Requires Python <= 3.11") >>> assert torch.allclose(out, torch.nn.functional.linear(x, weight, bias)) >>> >>> # Example 2: an operator with data-dependent output shape >>> @torch.library.custom_op("mylib::nonzero", mutates_args=()) >>> def nonzero(x: Tensor) -> Tensor: >>> x_np = x.cpu().numpy() >>> res = np.stack(np.nonzero(x_np), axis=1) >>> return torch.tensor(res, device=x.device) >>> >>> @nonzero.register_fake >>> def _(x): >>> # Number of nonzero-elements is data-dependent. >>> # Since we cannot peek at the data in an abstract impl, >>> # we use the ctx object to construct a new symint that >>> # represents the data-dependent size. >>> ctx = torch.library.get_ctx() >>> nnz = ctx.new_dynamic_size() >>> shape = [nnz, x.dim()] >>> result = x.new_empty(shape, dtype=torch.int64) >>> return result >>> >>> x = torch.tensor([0, 1, 2, 0, 0, 1]) >>> # xdoctest: +SKIP("Requires Python <= 3.11") >>> out = torch.compile(nonzero, fullgraph=True)(x) >>> # xdoctest: +SKIP("Requires Python <= 3.11") >>> assert torch.allclose(out, x.nonzero()) )r:)rBrr1r1r2 register_fakesGzCustomOpDef.register_fake) setup_context)backwardr\rcCs:|jj}tj|s*td|d|d||_||_dS)adRegister a backward formula for this custom op. In order for an operator to work with autograd, you need to register a backward formula: 1. You must tell us how to compute gradients during the backward pass by providing us a "backward" function. 2. If you need any values from the forward to compute gradients, you can use `setup_context` to save values for backward. ``backward_fn`` runs during the backward pass. It accepts ``(ctx, *grads)``: - ``grads`` is one or more gradients. The number of gradients matches the number of outputs of the operator. The ``ctx`` object is `the same ctx object `_ used by :class:`torch.autograd.Function`. The semantics of ``backward_fn`` are the same as :meth:`torch.autograd.Function.backward`. ``setup_context(ctx, inputs, output)`` runs during the forward pass. Please save quantities needed for backward onto the ``ctx`` object via either :meth:`torch.autograd.function.FunctionCtx.save_for_backward` or assigning them as attributes of ``ctx``. If your custom op has kwarg-only arguments, we expect the signature of ``setup_context`` to be ``setup_context(ctx, inputs, keyword_only_inputs, output)``. Both ``setup_context_fn`` and ``backward_fn`` must be traceable. That is, they may not directly access :meth:`torch.Tensor.data_ptr` and they must not depend on or mutate global state. If you need a non-traceable backward, you can make it a separate custom_op that you call inside ``backward_fn``. Examples: >>> import torch >>> import numpy as np >>> from torch import Tensor >>> >>> @torch.library.custom_op("mylib::numpy_sin", mutates_args=()) >>> def numpy_sin(x: Tensor) -> Tensor: >>> x_np = x.cpu().numpy() >>> y_np = np.sin(x_np) >>> return torch.from_numpy(y_np).to(device=x.device) >>> >>> def setup_context(ctx, inputs, output) -> Tensor: >>> x, = inputs >>> ctx.save_for_backward(x) >>> >>> def backward(ctx, grad): >>> x, = ctx.saved_tensors >>> return grad * x.cos() >>> >>> numpy_sin.register_autograd(backward, setup_context=setup_context) >>> >>> x = torch.randn(3, requires_grad=True) >>> y = numpy_sin(x) >>> grad_x, = torch.autograd.grad(y, x, torch.ones_like(y)) >>> assert torch.allclose(grad_x, x.cos()) >>> >>> # Example with a keyword-only arg >>> @torch.library.custom_op("mylib::numpy_mul", mutates_args=()) >>> def numpy_mul(x: Tensor, *, val: float) -> Tensor: >>> x_np = x.cpu().numpy() >>> y_np = x_np * val >>> return torch.from_numpy(y_np).to(device=x.device) >>> >>> def setup_context(ctx, inputs, keyword_only_inputs, output) -> Tensor: >>> ctx.val = keyword_only_inputs["val"] >>> >>> def backward(ctx, grad): >>> return grad * ctx.val >>> >>> numpy_mul.register_autograd(backward, setup_context=setup_context) >>> >>> x = torch.randn(3, requires_grad=True) >>> y = numpy_mul(x, val=3.14) >>> grad_x, = torch.autograd.grad(y, x, torch.ones_like(y)) >>> assert torch.allclose(grad_x, torch.full_like(x, 3.14)) z=Cannot register autograd formula for non-functional operator z with schema zP. Please create a functional operator and register an autograd formula for that.N)r"r#rrZis_functional_schemarQr<r;)rBr]r\rr1r1r2register_autograd]sR zCustomOpDef.register_autogradcsj}jj}t|}t|r4td||j|tj j tj j gdt j j_fdd}|jj|ddt jj}|jj|ddd jjjrЇfd d }|jj|d dd dS) NzUcustom_op with kwarg-only Tensor args. Please make your tensors not kwarg-only. Got: )tagscsBjdkr6tjjrdStddjjdj||S)Nz&There was no fake impl registered for zM. This is necessary for torch.compile/export/fx tracing to work. Please use `z$.register_fake` to add an fake impl.)r:rrZcan_generate_trivial_fake_implr"rQr8__name__)rRrSrDr1r2 fake_impls z6CustomOpDef._register_to_dispatcher..fake_impl)Z _stacklevelZAutogradT)Z with_keysetc stj||D]`\}}|js q|jjs*qt|trBtj |qt|t t fr|D]}t|trTtj |qTqt (jj|t j@f||W5QRSQRXdSr5)rrZ zip_schemar%r&rMrrgraphZincrement_versionrNrYrZ!_AutoDispatchBelowADInplaceOrViewr"Z redispatchZ_after_ADInplaceOrView_keyset)ZkeysetrRrSr/valvrrBr1r2adinplaceorview_impls&   zACustomOpDef._register_to_dispatcher..adinplaceorview_implZADInplaceOrView)r>r7r#rZ parse_schemarZhas_kwarg_only_tensorsNotImplementedErrordefineTagZpt2_compliant_tagZneeds_fixed_stride_orderrZ lookup_oprAr"Z_register_fakerZmake_autograd_implrZ is_mutable)rBlibr*Z cpp_schemaraZ autograd_implrgr1rfr2r?s2    z#CustomOpDef._register_to_dispatchercOs |j||Sr5)r")rBrRrSr1r1r2__call__szCustomOpDef.__call__)N)r` __module__ __qualname____doc__rXrrCpropertyrArEdevice_types_trr)r[r^r?rlr1r1r1r2r s(  `O ]?r zlibrary.Library OPDEF_TO_LIBr@)r+rrcCs@|d|}|tkr(t|t|=t|d}|t|<|S)NrZFRAGMENT)rrZ_destroyrLibrary)r+rqualnamerkr1r1r2r=s  r=)rRrSallowed_nestingrc#sDfdd}|D]}||EdHq|D]}||EdHq,dS)Nc3sFt|tr|Vn0dkrBt|ttfrBtt|idEdHdS)Nrr)rMrrNrYrL)r/rur1r2check's ziter_tensors..check)values)rRrSrurwr/kwargr1rvr2rL$s   rL)oprcCsBt|tr|St|tjr |j}t|ts.t|tkr>t|SdSr5)rMr r OpOverloadr7rXAssertionErrorr@)rzr1r1r2_maybe_get_opdef3s  r})N)r)$rOweakreftypingrrrrrrrr r r Ztorch.utils._exposed_inr rrrrrrrrXrqr4r rr__annotations__WeakValueDictionaryr@r=intrLr{r}r1r1r1r2sJ0   v