(self)
| 165 | for i, buf in enumerate(b): np.testing.assert_equal(expected[i], np.frombuffer(buf.as_memoryview(), np.int32)) |
| 166 | |
| 167 | def test_read_write_several_graphs(self): |
| 168 | d0 = Device.DEFAULT |
| 169 | b = [make_buffer(d0, fill=True) for _ in range(8)] |
| 170 | c: dict[Buffer,UOp] = {} |
| 171 | |
| 172 | calls1 = [get_ast(d0, 2).call(get_buf_uop(b[3],c), get_buf_uop(b[1],c), get_buf_uop(b[2],c), metadata=())] |
| 173 | calls2 = [get_ast(d0, 2).call(get_buf_uop(b[4],c), get_buf_uop(b[1],c), get_buf_uop(b[3],c), metadata=())] |
| 174 | calls3 = [get_ast(d0, 2).call(get_buf_uop(b[5],c), get_buf_uop(b[4],c), get_buf_uop(b[2],c), metadata=())] |
| 175 | |
| 176 | out = [b[3], b[4], b[5]] |
| 177 | zero_bufs(out) |
| 178 | run_schedule(calls1 + calls2 + calls3) |
| 179 | expected = [np.frombuffer(x.as_memoryview(), np.int32).copy() for x in b] |
| 180 | |
| 181 | for _ in range(RUN_CNT): |
| 182 | zero_bufs(out) |
| 183 | make_graph(Device[d0].graph, calls1)([], {}) |
| 184 | make_graph(Device[d0].graph, calls2)([], {}) |
| 185 | make_graph(Device[d0].graph, calls3)([], {}) |
| 186 | for i, buf in enumerate(b): np.testing.assert_equal(expected[i], np.frombuffer(buf.as_memoryview(), np.int32)) |
| 187 | |
| 188 | @needs_second_gpu |
| 189 | def test_copies_2_devs(self): |
nothing calls this directly
no test coverage detected