(call:UOp, sink:UOp)
| 209 | ]) |
| 210 | |
| 211 | def _validate(call:UOp, sink:UOp) -> UOp: |
| 212 | params = get_call_arg_uops(call) |
| 213 | shadows = tuple(UOp.new_buffer(("CPU",)*len(p.device) if isinstance(p.device, tuple) else "CPU", prod(p.max_shape), p.dtype.base) for p in params) |
| 214 | copies = tuple(p.copy_to_device(s.device).call(s, p) for s, p in zip(shadows, params)) |
| 215 | return UOp(Ops.LINEAR, src=copies + (call, UOp(Ops.CUSTOM_FUNCTION, dtypes.void, src=(sink,), arg="validate").call(*shadows, *params))) |
| 216 | pm_validate = PatternMatcher([(UPat(Ops.CALL, src=(UPat(Ops.SINK, name="sink"),), name="call", allow_any_len=True), _validate)]) + pm_flatten_linear |
| 217 | |
| 218 | # ctx is beam value |
nothing calls this directly
no test coverage detected
searching dependent graphs…