(self)
| 209 | for buf in b0 + b1: np.testing.assert_equal(expected[buf], np.frombuffer(buf.as_memoryview(), np.int32)) |
| 210 | |
| 211 | def test_graph_offset_bufs(self): |
| 212 | self.skip_if_not_multigraph() |
| 213 | d0 = Device.DEFAULT |
| 214 | if not hasattr(Device[d0].allocator, "_offset"): self.skipTest("device does not support _offset") |
| 215 | |
| 216 | b0 = make_buffer(d0, fill=True) |
| 217 | b1 = make_view(b0, 0, b0.size) |
| 218 | b2 = make_view(b0, 0, b0.size) |
| 219 | c: dict[Buffer,UOp] = {} |
| 220 | |
| 221 | calls = [ |
| 222 | UOp(Ops.COPY).call(get_buf_uop(b0,c), get_buf_uop(b2,c), metadata=()), |
| 223 | get_ast(d0, 2).call(get_buf_uop(b1,c), get_buf_uop(b0,c), get_buf_uop(b2,c), metadata=()), |
| 224 | ] |
| 225 | |
| 226 | zero_bufs([b0]) |
| 227 | run_schedule(calls) |
| 228 | expected = np.frombuffer(b0.as_memoryview(), np.int32).copy() |
| 229 | |
| 230 | for _ in range(RUN_CNT): |
| 231 | zero_bufs([b0]) |
| 232 | make_graph(Device[d0].graph, calls)([], {}) |
| 233 | np.testing.assert_equal(expected, np.frombuffer(b0.as_memoryview(), np.int32)) |
| 234 | |
| 235 | def test_partial_write_preserves_write_dep(self): |
| 236 | self.skip_if_not_multigraph() |
nothing calls this directly
no test coverage detected