Event supports cross-process and cross-device (IPC event)-pytorch-AtomGit

Mtest/allowlist_for_publicAPI.json+1-0

	@@ -585,6 +585,7 @@
585	"Union",	585	"Union",
586	"check_serializing_named_tensor",	586	"check_serializing_named_tensor",
587	"register_after_fork",	587	"register_after_fork",
		588	+ "reduce_event",
588	"reduce_tensor",	589	"reduce_tensor",
589	"reduce_storage"	590	"reduce_storage"
590	],	591	],

Atest/npu/test_ipc_event.py+217-0

@@ -0,0 +1,217 @@
	1	+import os
	2	+import gc
	3	+import unittest
	4	+import numpy as np
	5	+import torch.multiprocessing as mp
	6	+
	7	+import torch
	8	+import torch_npu
	9	+from torch_npu.testing.common_utils import SupportedDevices
	10	+from torch_npu.testing.testcase import TestCase, run_tests
	11	+from torch_npu.testing.common_distributed import skipIfUnsupportMultiNPU
	12	+
	13	+
	14	+def is_ipc_event_supported():
	15	+ try:
	16	+ ev = torch.npu.Event(enable_timing=False, interprocess=True)
	17	+ except RuntimeError as e:
	18	+ return False
	19	+ else:
	20	+ return True
	21	+
	22	+skip_ipc_event_case = not is_ipc_event_supported()
	23	+
	24	+
	25	+class Test_ipc_event(TestCase):
	26	+ @skipIfUnsupportMultiNPU(2)
	27	+ def test_d2d_copy1(self):
	28	+ a = torch.tensor(1.).to('npu:0')
	29	+ b = torch.tensor(1.).to('npu:1')
	30	+ b.copy_(a)
	31	+ self.assertEqual(a.cpu(), b.cpu())
	32	+
	33	+ @skipIfUnsupportMultiNPU(2)
	34	+ def test_d2d_copy2(self):
	35	+ a = torch.tensor(1.).to('npu:0')
	36	+ b = a.to('npu:1')
	37	+ self.assertEqual(a.cpu(), b.cpu())
	38	+
	39	+ @skipIfUnsupportMultiNPU(2)
	40	+ def test_d2d_copy3(self):
	41	+ a = torch.ones(2, 1024, 1024, 1024).to('npu:0')
	42	+ b = a.to('npu:1', non_blocking=True)
	43	+ self.assertEqual(a.cpu(), b.cpu())
	44	+
	45	+ @SupportedDevices(['Ascend910B'])
	46	+ def test_ipc_event_pickle(self):
	47	+ if skip_ipc_event_case:
	48	+ return
	49	+
	50	+ ev = torch.npu.Event(enable_timing=False, interprocess=True)
	51	+ ctx = mp.get_context("spawn")
	52	+ q = ctx.Queue()
	53	+ q.put(ev)
	54	+
	55	+ @staticmethod
	56	+ def _child_proc1(q):
	57	+ ev = q.get()
	58	+ assert ev.device.type == 'npu'
	59	+ assert ev.device.index == 0
	60	+ ev.wait()
	61	+ ev.synchronize()
	62	+
	63	+ @SupportedDevices(['Ascend910B'])
	64	+ def test_ipc_event_1(self):
	65	+ if skip_ipc_event_case:
	66	+ return
	67	+
	68	+ ctx = mp.get_context("spawn")
	69	+ q = ctx.Queue()
	70	+ p = ctx.Process(target=Test_ipc_event._child_proc1, args=(q,))
	71	+ p.start()
	72	+
	73	+ dev = torch.device("npu:0")
	74	+ with torch.npu.device(dev):
	75	+ stream = torch.npu.Stream()
	76	+ with torch.npu.stream(stream):
	77	+ ev = torch.npu.Event(enable_timing=False, interprocess=True)
	78	+ ev.record(stream)
	79	+
	80	+ q.put(ev)
	81	+ p.join()
	82	+
	83	+ @staticmethod
	84	+ def _child_proc2(q1, q2):
	85	+ dev = torch.device("npu:0")
	86	+ with torch.npu.device(dev):
	87	+ stream = torch.npu.Stream()
	88	+ with torch.npu.stream(stream):
	89	+ ev = q1.get()
	90	+ assert ev.device.type == 'npu'
	91	+ assert ev.device.index == 0
	92	+ ev.wait()
	93	+ ev.record(stream)
	94	+ q2.put('x')
	95	+ assert q1.get() == 'y'
	96	+
	97	+ @SupportedDevices(['Ascend910B'])
	98	+ def test_ipc_event_2(self):
	99	+ if skip_ipc_event_case:
	100	+ return
	101	+
	102	+ ctx = mp.get_context("spawn")
	103	+ q1 = ctx.Queue()
	104	+ q2 = ctx.Queue()
	105	+ p = ctx.Process(target=Test_ipc_event._child_proc2, args=(q1, q2))
	106	+ p.start()
	107	+
	108	+ dev = torch.device("npu:0")
	109	+ with torch.npu.device(dev):
	110	+ stream = torch.npu.Stream()
	111	+ with torch.npu.stream(stream):
	112	+ ev = torch.npu.Event(enable_timing=False, interprocess=True)
	113	+ ev.record(stream)
	114	+
	115	+ q1.put(ev)
	116	+ self.assertEqual(q2.get(), 'x')
	117	+ ev.wait()
	118	+ ev.synchronize()
	119	+ q1.put('y')
	120	+ p.join()
	121	+
	122	+ @skipIfUnsupportMultiNPU(2)
	123	+ @SupportedDevices(['Ascend910B'])
	124	+ def test_event_handle_multi_npu(self):
	125	+ if skip_ipc_event_case:
	126	+ return
	127	+
	128	+ d0 = torch.device("npu:0")
	129	+ d1 = torch.device("npu:1")
	130	+ with torch.npu.device(d0):
	131	+ e0 = torch.npu.Event(enable_timing=False, interprocess=True)
	132	+
	133	+ with torch.npu.device(d1):
	134	+ # create handle on different device from un-recorded event
	135	+ e0.ipc_handle()
	136	+
	137	+ with torch.npu.device(d0):
	138	+ e1 = torch.npu.Event(enable_timing=False, interprocess=True)
	139	+ stream = torch.npu.Stream()
	140	+ e1.record(stream)
	141	+
	142	+ with torch.npu.device(d1):
	143	+ # create handle on different device from recorded event
	144	+ e1.ipc_handle()
	145	+
	146	+ @staticmethod
	147	+ def _test_event_handle_importer_consumer(handle, p2c, c2p):
	148	+ e1 = torch.npu.Event.from_ipc_handle(0, handle)
	149	+ c2p.put(0) # notify parent child is ready
	150	+ p2c.get() # wait for record in parent
	151	+ e1.synchronize()
	152	+ c2p.put(1) # notify synchronization is done in child
	153	+ p2c.get() # wait for parent to finish before destructing child event
	154	+
	155	+ @SupportedDevices(['Ascend910B'])
	156	+ def test_event_handle_importer(self):
	157	+ if skip_ipc_event_case:
	158	+ return
	159	+
	160	+ e0 = torch.npu.Event(enable_timing=False, interprocess=True)
	161	+ self.assertTrue(e0.query())
	162	+
	163	+ ctx = mp.get_context("spawn")
	164	+ p2c = ctx.SimpleQueue()
	165	+ c2p = ctx.SimpleQueue()
	166	+ p = ctx.Process(
	167	+ target=Test_ipc_event._test_event_handle_importer_consumer,
	168	+ args=(e0.ipc_handle(), p2c, c2p),
	169	+ )
	170	+ p.start()
	171	+
	172	+ c2p.get() # wait for child to become ready
	173	+ e0.record()
	174	+ p2c.put(0) # notify child event is recorded
	175	+
	176	+ c2p.get() # wait for synchronization in child
	177	+ self.assertTrue(e0.query())
	178	+ p2c.put(1) # notify child that parent is done
	179	+ p.join()
	180	+
	181	+ @staticmethod
	182	+ def _test_event_handle_exporter_consumer(handle, p2c, c2p):
	183	+ stream = torch.npu.Stream()
	184	+ with torch.npu.stream(stream):
	185	+ e1 = torch.npu.Event.from_ipc_handle(torch.npu.current_device(), handle)
	186	+ e1.record()
	187	+ c2p.put(0)
	188	+ # wait for parent process finished synchronization before
	189	+ # destructing e1
	190	+ p2c.get()
	191	+
	192	+ @SupportedDevices(['Ascend910B'])
	193	+ def test_event_handle_exporter(self):
	194	+ if skip_ipc_event_case:
	195	+ return
	196	+
	197	+ e0 = torch.npu.Event(enable_timing=False, interprocess=True)
	198	+
	199	+ ctx = mp.get_context("spawn")
	200	+ p2c = ctx.SimpleQueue()
	201	+ c2p = ctx.SimpleQueue()
	202	+ p = ctx.Process(
	203	+ target=Test_ipc_event._test_event_handle_exporter_consumer,
	204	+ args=(e0.ipc_handle(), p2c, c2p),
	205	+ )
	206	+ p.start()
	207	+ # wait for event in child process is recorded
	208	+ c2p.get()
	209	+
	210	+ e0.synchronize()
	211	+ self.assertTrue(e0.query())
	212	+ p2c.put(0)
	213	+ p.join()
	214	+
	215	+
	216	+if __name__ == '__main__':
	217	+ run_tests()

Mtest/npu/test_torch_npu.py+2-1

	@@ -266,7 +266,8 @@ class TorchNPUApiTestCase(TestCase):
266	self.assertEqual(s.query(), False)	266	self.assertEqual(s.query(), False)
267		267
268	def test_npu_event(self):	268	def test_npu_event(self):
269	- res = torch_npu.npu.Event(enable_timing=True, blocking=True, interprocess=True)	269	+ # The old CANN and HDK do not support IPC events, so the interprocess=True parameter is not specified here.
		270	+ res = torch_npu.npu.Event(enable_timing=True, blocking=True)
270	self.assertIsInstance(res, torch_npu.npu.Event)	271	self.assertIsInstance(res, torch_npu.npu.Event)
271		272
272	def test_npu_event_elapsed_time(self):	273	def test_npu_event_elapsed_time(self):

Mtest/torch_npu_schema.json+23-0

	@@ -935,6 +935,9 @@
935	"torch_npu.npu.Event": {	935	"torch_npu.npu.Event": {
936	"signature": "(enable_timing=False, blocking=False, interprocess=False)"	936	"signature": "(enable_timing=False, blocking=False, interprocess=False)"
937	},	937	},
		938	+ "torch_npu.npu.Event.from_ipc_handle": {
		939	+ "signature": "(device, handle)"
		940	+ },
938	"torch_npu.npu.Event.record": {	941	"torch_npu.npu.Event.record": {
939	"signature": "(self, stream=None)"	942	"signature": "(self, stream=None)"
940	},	943	},
	@@ -950,6 +953,9 @@
950	"torch_npu.npu.Event.synchronize": {	953	"torch_npu.npu.Event.synchronize": {
951	"signature": "(self)"	954	"signature": "(self)"
952	},	955	},
		956	+ "torch_npu.npu.Event.ipc_handle": {
		957	+ "signature": "(self)"
		958	+ },
953	"torch_npu.npu.ExternalEvent": {	959	"torch_npu.npu.ExternalEvent": {
954	"signature": "()"	960	"signature": "()"
955	},	961	},
	@@ -1553,6 +1559,9 @@
1553	"torch_npu.npu.streams.Event": {	1559	"torch_npu.npu.streams.Event": {
1554	"signature": "(enable_timing=False, blocking=False, interprocess=False)"	1560	"signature": "(enable_timing=False, blocking=False, interprocess=False)"
1555	},	1561	},
		1562	+ "torch_npu.npu.streams.Event.from_ipc_handle": {
		1563	+ "signature": "(device, handle)"
		1564	+ },
1556	"torch_npu.npu.streams.Event.record": {	1565	"torch_npu.npu.streams.Event.record": {
1557	"signature": "(self, stream=None)"	1566	"signature": "(self, stream=None)"
1558	},	1567	},
	@@ -1568,6 +1577,9 @@
1568	"torch_npu.npu.streams.Event.synchronize": {	1577	"torch_npu.npu.streams.Event.synchronize": {
1569	"signature": "(self)"	1578	"signature": "(self)"
1570	},	1579	},
		1580	+ "torch_npu.npu.streams.Event.ipc_handle": {
		1581	+ "signature": "(self)"
		1582	+ },
1571	"torch_npu.npu.streams.Stream": {	1583	"torch_npu.npu.streams.Stream": {
1572	"signature": "(device=None, priority=0, **kwargs)"	1584	"signature": "(device=None, priority=0, **kwargs)"
1573	},	1585	},
	@@ -2390,6 +2402,9 @@
2390	"torch_npu.distributed.all_gather_into_tensor_uneven": {	2402	"torch_npu.distributed.all_gather_into_tensor_uneven": {
2391	"signature": "(output, input, output_split_sizes=None, group=None, async_op=False)"	2403	"signature": "(output, input, output_split_sizes=None, group=None, async_op=False)"
2392	},	2404	},
		2405	+ "torch_npu.multiprocessing.reductions.rebuild_npu_event": {
		2406	+ "signature": "(device, handle)"
		2407	+ },
2393	"torch_npu.multiprocessing.reductions.rebuild_npu_tensor": {	2408	"torch_npu.multiprocessing.reductions.rebuild_npu_tensor": {
2394	"signature": "(tensor_cls, tensor_size, tensor_stride, tensor_offset, storage_cls, dtype, storage_device, storage_handle, storage_size_bytes, storage_offset_bytes, requires_grad, ref_counter_handle, ref_counter_offset, event_handle, event_sync_required)"	2409	"signature": "(tensor_cls, tensor_size, tensor_stride, tensor_offset, storage_cls, dtype, storage_device, storage_handle, storage_size_bytes, storage_offset_bytes, requires_grad, ref_counter_handle, ref_counter_offset, event_handle, event_sync_required)"
2395	},	2410	},
	@@ -2551,6 +2566,10 @@
2551	"signature": "(unsigned int flags)",	2566	"signature": "(unsigned int flags)",
2552	"file": "torch_npu/csrc/core/npu/NPUEvent.h"	2567	"file": "torch_npu/csrc/core/npu/NPUEvent.h"
2553	},	2568	},
		2569	+ "torch_c_func: c10_npu::NPUEvent::NPUEvent(c10::DeviceIndex device_index, const aclrtIpcEventHandle* handle)": {
		2570	+ "signature": "(c10::DeviceIndex device_index, const aclrtIpcEventHandle* handle)",
		2571	+ "file": "torch_npu/csrc/core/npu/NPUEvent.h"
		2572	+ },
2554	"torch_c_func: c10_npu::NPUEvent::~NPUEvent()": {	2573	"torch_c_func: c10_npu::NPUEvent::~NPUEvent()": {
2555	"signature": "()",	2574	"signature": "()",
2556	"file": "torch_npu/csrc/core/npu/NPUEvent.h"	2575	"file": "torch_npu/csrc/core/npu/NPUEvent.h"
	@@ -2607,6 +2626,10 @@
2607	"signature": "() -> void",	2626	"signature": "() -> void",
2608	"file": "torch_npu/csrc/core/npu/NPUEvent.h"	2627	"file": "torch_npu/csrc/core/npu/NPUEvent.h"
2609	},	2628	},
		2629	+ "torch_c_func: c10_npu::NPUEvent::ipc_handle": {
		2630	+ "signature": "(aclrtIpcEventHandle* handle) -> void",
		2631	+ "file": "torch_npu/csrc/core/npu/NPUEvent.h"
		2632	+ },
2610	"torch_c_func: at_npu::NPUGeneratorImpl::NPUGeneratorImpl": {	2633	"torch_c_func: at_npu::NPUGeneratorImpl::NPUGeneratorImpl": {
2611	"signature": "(c10::DeviceIndex device_index = -1)",	2634	"signature": "(c10::DeviceIndex device_index = -1)",
2612	"file": "torch_npu/csrc/aten/NPUGeneratorImpl.h"	2635	"file": "torch_npu/csrc/aten/NPUGeneratorImpl.h"

Mthird_party/acl/inc/acl/acl_rt.h+31-0

	@@ -27,6 +27,7 @@ extern "C" {
27	#define ACL_EVENT_CAPTURE_STREAM_PROGRESS 0x00000002U	27	#define ACL_EVENT_CAPTURE_STREAM_PROGRESS 0x00000002U
28	#define ACL_EVENT_TIME_LINE 0x00000008U	28	#define ACL_EVENT_TIME_LINE 0x00000008U
29	#define ACL_EVENT_EXTERNAL 0x00000020U	29	#define ACL_EVENT_EXTERNAL 0x00000020U
		30	+#define ACL_EVENT_IPC 0x00000040U
30		31
31	// for create stream	32	// for create stream
32	#define ACL_STREAM_FAST_LAUNCH 0x00000001U	33	#define ACL_STREAM_FAST_LAUNCH 0x00000001U
	@@ -65,6 +66,8 @@ extern "C" {
65		66
66	#define ACL_VALUE_WAIT_EQ 0x1	67	#define ACL_VALUE_WAIT_EQ 0x1
67		68
		69	+#define ACL_IPC_EVENT_HANDLE_SIZE 64U
		70	+
68	constexpr int32_t DEVICE_UTILIZATION_NOT_SUPPORT = -1;	71	constexpr int32_t DEVICE_UTILIZATION_NOT_SUPPORT = -1;
69		72
70	typedef enum aclrtRunMode {	73	typedef enum aclrtRunMode {
	@@ -446,6 +449,10 @@ typedef struct {
446	uint8_t rsv[16];	449	uint8_t rsv[16];
447	} aclrtMemcpyBatchAttr;	450	} aclrtMemcpyBatchAttr;
448		451
		452	+typedef struct aclrtIpcEventHandle {
		453	+ char reserved[ACL_IPC_EVENT_HANDLE_SIZE];
		454	+} aclrtIpcEventHandle;
		455	+
449	/**	456	/**
450	* @ingroup AscendCL	457	* @ingroup AscendCL
451	* @brief peek at last error by level	458	* @brief peek at last error by level
	@@ -975,6 +982,30 @@ ACL_FUNC_VISIBILITY aclError aclrtQueryEventStatus(aclrtEvent event, aclrtEventR
975	*/	982	*/
976	ACL_FUNC_VISIBILITY aclError aclrtQueryEventWaitStatus(aclrtEvent event, aclrtEventWaitStatus *status);	983	ACL_FUNC_VISIBILITY aclError aclrtQueryEventWaitStatus(aclrtEvent event, aclrtEventWaitStatus *status);
977		984
		985	+/**
		986	+ * @ingroup AscendCL
		987	+ * @brief get an interprocess handle for a previously allocated event.
		988	+ *
		989	+ * @param [in] event event allocated with ACL_EVENT_IPC flags
		990	+ * @param [out] handle handle for interprocess
		991	+ *
		992	+ * @retval ACL_SUCCESS The function is successfully executed.
		993	+ * @retval OtherValues Failure
		994	+ */
		995	+ACL_FUNC_VISIBILITY aclError aclrtIpcGetEventHandle(aclrtEvent event, aclrtIpcEventHandle *handle);
		996	+
		997	+/**
		998	+ * @ingroup AscendCL
		999	+ * @brief opens an interprocess event handle for user in the current process.
		1000	+ *
		1001	+ * @param [in] handle interprocess handle to open
		1002	+ * @param [out] event returns the imported event
		1003	+ *
		1004	+ * @retval ACL_SUCCESS The function is successfully executed.
		1005	+ * @retval OtherValues Failure
		1006	+ */
		1007	+ACL_FUNC_VISIBILITY aclError aclrtIpcOpenEventHandle(aclrtIpcEventHandle handle, aclrtEvent *event);
		1008	+
978	/**	1009	/**
979	* @ingroup AscendCL	1010	* @ingroup AscendCL
980	* @brief Block Host Running, wait event to be complete	1011	* @brief Block Host Running, wait event to be complete

Mthird_party/acl/libs/acl.cpp+2-0

	@@ -47,6 +47,8 @@ aclError aclrtRecordEvent(aclrtEvent event, aclrtStream stream){return 0;}
47	aclError aclrtStreamWaitEvent(aclrtStream stream, aclrtEvent event){return 0;}	47	aclError aclrtStreamWaitEvent(aclrtStream stream, aclrtEvent event){return 0;}
48	aclError aclrtSynchronizeEvent(aclrtEvent event){return 0;}	48	aclError aclrtSynchronizeEvent(aclrtEvent event){return 0;}
49	aclError aclrtEventElapsedTime(float *ms, aclrtEvent start, aclrtEvent end){return 0;}	49	aclError aclrtEventElapsedTime(float *ms, aclrtEvent start, aclrtEvent end){return 0;}
		50	+aclError aclrtIpcGetEventHandle(aclrtEvent event, aclrtIpcEventHandle *handle){return 0;}
		51	+aclError aclrtIpcOpenEventHandle(aclrtIpcEventHandle handle, aclrtEvent *event){return 0;}
50		52
51	// memory相关操作	53	// memory相关操作
52	aclError aclrtMalloc(void **devPtr, size_t size, aclrtMemMallocPolicy policy){return 0;}	54	aclError aclrtMalloc(void **devPtr, size_t size, aclrtMemMallocPolicy policy){return 0;}

Mtorch_npu/csrc/aten/common/CopyKernel.cpp+47-13

	@@ -2,8 +2,10 @@
2	#include <ATen/core/CachingHostAllocator.h>	2	#include <ATen/core/CachingHostAllocator.h>
3		3
4	#include "op_plugin/OpInterface.h"	4	#include "op_plugin/OpInterface.h"
		5	+#include "torch_npu/csrc/core/npu/interface/AclInterface.h"
5	#include "torch_npu/csrc/core/npu/NPUException.h"	6	#include "torch_npu/csrc/core/npu/NPUException.h"
6	#include "torch_npu/csrc/core/npu/NPUGuard.h"	7	#include "torch_npu/csrc/core/npu/NPUGuard.h"
		8	+#include "torch_npu/csrc/core/npu/NPUEvent.h"
7	#include "torch_npu/csrc/core/npu/NPUPeerToPeerAccess.h"	9	#include "torch_npu/csrc/core/npu/NPUPeerToPeerAccess.h"
8	#include "torch_npu/csrc/framework/utils/CalcuOpUtil.h"	10	#include "torch_npu/csrc/framework/utils/CalcuOpUtil.h"
9	#include "torch_npu/csrc/core/npu/register/OptionsManager.h"	11	#include "torch_npu/csrc/core/npu/register/OptionsManager.h"
	@@ -260,31 +262,63 @@ bool can_use_memcpy(at::Tensor& dst, const at::Tensor& src)
260		262
261	void copy_d2d(at::Tensor& self, const at::Tensor& src, bool non_blocking)	263	void copy_d2d(at::Tensor& self, const at::Tensor& src, bool non_blocking)
262	{	264	{
263	- c10_npu::NPUGuard guard(src.device());	265	+ c10::Device src_device = src.device();
		266	+ c10::Device dst_device = self.device();
		267	+
		268	+ c10_npu::NPUGuard guard(src_device);
		269	+
264	// p2p enable and synchronize self stream	270	// p2p enable and synchronize self stream
265	- auto self_device_idx = self.device().index();	271	+ auto dst_device_idx = dst_device.index();
266	- auto src_device_idx = src.device().index();	272	+ auto src_device_idx = src_device.index();
267	- if (self_device_idx != src_device_idx) {	273	+ c10_npu::NPUStream src_stream = c10_npu::getCurrentNPUStream(src_device_idx);
		274	+ if (dst_device_idx != src_device_idx) {
268	bool warning_flag = false;	275	bool warning_flag = false;
269	- NpuP2pCtrl::get_instance().get_p2p_access(src_device_idx, self_device_idx, warning_flag);	276	+ NpuP2pCtrl::get_instance().get_p2p_access(src_device_idx, dst_device_idx, warning_flag);
270	// In the same 'os', tensor can copy even if the enable fails	277	// In the same 'os', tensor can copy even if the enable fails
271	if (warning_flag) {	278	if (warning_flag) {
272	- ASCEND_LOGW("p2p enable from %d to %d is fails", src_device_idx, self_device_idx);	279	+ ASCEND_LOGW("p2p enable from %d to %d is fails", src_device_idx, dst_device_idx);
		280	+ }
		281	+
		282	+ // This is a cross-device copy on the src current stream and dst current
		283	+ // stream. We perform a two-way barrier between both devices' streams
		284	+ // before the copy. This ensures that any write-after-write and
		285	+ // write-after-read dependencies on the destination side are handled, so
		286	+ // that no one is operating on the dst memory when we perform the copy.
		287	+ // src waits on dst barrier (src already waits on src)
		288	+ if (c10_npu::acl::IsSupportIpcEvent()) {
		289	+ c10_npu::NPUEvent dst_ready(ACL_EVENT_IPC);
		290	+ guard.set_device(dst_device);
		291	+ dst_ready.record(c10_npu::getCurrentNPUStream(dst_device_idx));
		292	+ guard.set_device(src_device);
		293	+ dst_ready.block(src_stream);
		294	+ } else {
		295	+ guard.set_device(dst_device);
		296	+ c10_npu::NPUStream dst_stream = c10_npu::getCurrentNPUStream(dst_device_idx);
		297	+ NPU_CHECK_ERROR(c10_npu::acl::AclrtSynchronizeStreamWithTimeout(dst_stream));
		298	+ guard.set_device(src_device);
273	}	299	}
274	- guard.set_device(self.device());
275	- c10_npu::NPUStream dst_stream = c10_npu::getCurrentNPUStream(self_device_idx);
276	- NPU_CHECK_ERROR(c10_npu::acl::AclrtSynchronizeStreamWithTimeout(dst_stream));
277	- guard.set_device(src.device());
278	}	300	}
		301	+
279	if (self.dtype() != src.dtype()) {	302	if (self.dtype() != src.dtype()) {
280	custom_ops::npu_dtype_cast_(self, src); // npu_dtype_cast_ will call copy function.	303	custom_ops::npu_dtype_cast_(self, src); // npu_dtype_cast_ will call copy function.
281	return;	304	return;
282	}	305	}
283	copy_d2d_dtype(self, src, non_blocking);	306	copy_d2d_dtype(self, src, non_blocking);
		307	+
284	// synchronize src stream for different devices copy	308	// synchronize src stream for different devices copy
285	- if (self_device_idx != src_device_idx) {	309	+ if (dst_device_idx != src_device_idx) {
286	- c10_npu::NPUStream copy_stream = c10_npu::getCurrentNPUStream();	310	+ // dst waits on src barrier (dst already waits on dst). We cannot
287	- NPU_CHECK_ERROR(c10_npu::acl::AclrtSynchronizeStreamWithTimeout(copy_stream));	311	+ // operate on dst's copy until the copy is complete.
		312	+
		313	+ // Still on src_device, record stream event
		314	+ if (c10_npu::acl::IsSupportIpcEvent()) {
		315	+ c10_npu::NPUEvent src_ready(ACL_EVENT_IPC);
		316	+ src_ready.record(src_stream);
		317	+ guard.set_device(dst_device);
		318	+ src_ready.block(c10_npu::getCurrentNPUStream(dst_device_idx));
		319	+ } else {
		320	+ NPU_CHECK_ERROR(c10_npu::acl::AclrtSynchronizeStreamWithTimeout(src_stream));
		321	+ }
288	}	322	}
289	}	323	}
290		324

Mtorch_npu/csrc/core/npu/NPUEvent.cpp+54-4

	@@ -21,6 +21,25 @@ NPUEvent::NPUEvent()
21	flags_ = c10_npu::acl::IsExistCreateEventExWithFlag() ? ACL_EVENT_SYNC : ACL_EVENT_DEFAULT;	21	flags_ = c10_npu::acl::IsExistCreateEventExWithFlag() ? ACL_EVENT_SYNC : ACL_EVENT_DEFAULT;
22	}	22	}
23		23
		24	+NPUEvent::NPUEvent(
		25	+ c10::DeviceIndex device_index, const aclrtIpcEventHandle* handle) : device_index_(device_index)
		26	+{
		27	+ NPUGuard guard(device_index_);
		28	+ LazySetDevice(device_index_);
		29	+ NPU_CHECK_ERROR(acl::AclIpcOpenEventHandle(*handle, &event_));
		30	+ c10_npu::NPUEventManager::GetInstance().AddIpcEvent(event_);
		31	+ flags_ = ACL_EVENT_IPC;
		32	+ is_created_ = true;
		33	+
		34	+#ifndef BUILD_LIBTORCH
		35	+ const c10_npu::impl::PyCallbackTrigger *trigger = c10_npu::impl::NPUTrace::getTrace();
		36	+ if (C10_UNLIKELY(trigger)) {
		37	+ trigger->traceNpuEventOpenHandle(reinterpret_cast<uintptr_t>(handle),
		38	+ reinterpret_cast<uintptr_t>(event_));
		39	+ }
		40	+#endif
		41	+}
		42	+
24	NPUEvent::~NPUEvent()	43	NPUEvent::~NPUEvent()
25	{	44	{
26	try {	45	try {
	@@ -29,7 +48,9 @@ NPUEvent::~NPUEvent()
29	}	48	}
30	if (is_created_ && (c10_npu::NpuSysCtrl::GetInstance().GetInitFlag())) {	49	if (is_created_ && (c10_npu::NpuSysCtrl::GetInstance().GetInitFlag())) {
31	NPU_CHECK_ERROR(c10_npu::queue::LaunchLazyDestroyEventTask(event_, device_index_));	50	NPU_CHECK_ERROR(c10_npu::queue::LaunchLazyDestroyEventTask(event_, device_index_));
32	- if (!c10_npu::acl::IsExistCreateEventExWithFlag() \|\| c10_npu::option::OptionsManager::GetPerStreamQueue()) {	51	+ if (!c10_npu::acl::IsExistCreateEventExWithFlag() \|\|
		52	+ c10_npu::option::OptionsManager::GetPerStreamQueue() \|\|
		53	+ flags_ == ACL_EVENT_IPC) {
33	c10_npu::NPUEventManager::GetInstance().QueryAndDestroyEvent();	54	c10_npu::NPUEventManager::GetInstance().QueryAndDestroyEvent();
34	}	55	}
35	}	56	}
	@@ -105,8 +126,9 @@ void NPUEvent::block(const NPUStream& stream)
105	createEvent(stream.device_index());	126	createEvent(stream.device_index());
106	}	127	}
107	if (is_created_) {	128	if (is_created_) {
108	- // If multiple task queues are dequeued, it is necessary to ensure that the record is dequeued before wait	129	+ // If using multiple task queues or using IPC events across devices in a single process,
109	- while (c10_npu::option::OptionsManager::GetPerStreamQueue() &&	130	+ // it is necessary to ensure that the enqueued record is dequeued before wait.
		131	+ while ((c10_npu::option::OptionsManager::GetPerStreamQueue() \|\| flags_ == ACL_EVENT_IPC) &&
110	!c10_npu::NPUEventManager::GetInstance().IsEventRecorded(event_)) {	132	!c10_npu::NPUEventManager::GetInstance().IsEventRecorded(event_)) {
111	std::this_thread::sleep_for(std::chrono::microseconds(10)); // 10 us	133	std::this_thread::sleep_for(std::chrono::microseconds(10)); // 10 us
112	}	134	}
	@@ -199,6 +221,31 @@ void NPUEvent::reset(const NPUStream& stream) const
199	}	221	}
200	}	222	}
201		223
		224	+// Note: AclIpcGetEventHandle must be called on the same device as the event
		225	+void NPUEvent::ipc_handle(aclrtIpcEventHandle* handle)
		226	+{
		227	+ if (!is_created_) {
		228	+ // this NPUEvent object was initially constructed from flags but event_
		229	+ // is not created yet.
		230	+ createEvent(getCurrentNPUStream().device_index());
		231	+ }
		232	+
		233	+ // If using Event across processes, make sure that the enqueued record is dequeued before the wait of other processes.
		234	+ while (c10_npu::option::OptionsManager::GetTaskQueueEnable() != 0 && !c10_npu::NPUEventManager::GetInstance().IsEventRecorded(event_)) {
		235	+ std::this_thread::sleep_for(std::chrono::microseconds(10)); // 10 us
		236	+ }
		237	+
		238	+ NPUGuard guard(device_index_);
		239	+ NPU_CHECK_ERROR(acl::AclIpcGetEventHandle(event_, handle));
		240	+#ifndef BUILD_LIBTORCH
		241	+ const c10_npu::impl::PyCallbackTrigger *trigger = c10_npu::impl::NPUTrace::getTrace();
		242	+ if (C10_UNLIKELY(trigger)) {
		243	+ trigger->traceNpuEventGetHandle(reinterpret_cast<uintptr_t>(event_),
		244	+ reinterpret_cast<uintptr_t>(handle));
		245	+ }
		246	+#endif
		247	+}
		248	+
202	void NPUEvent::createEvent(c10::DeviceIndex device_index)	249	void NPUEvent::createEvent(c10::DeviceIndex device_index)
203	{	250	{
204	device_index_ = device_index;	251	device_index_ = device_index;
	@@ -212,12 +259,15 @@ void NPUEvent::createEvent(c10::DeviceIndex device_index)
212	trigger->traceNpuEventCreation(reinterpret_cast<uintptr_t>(event_));	259	trigger->traceNpuEventCreation(reinterpret_cast<uintptr_t>(event_));
213	}	260	}
214	#endif	261	#endif
		262	+ if (flags_ == ACL_EVENT_IPC) {
		263	+ c10_npu::NPUEventManager::GetInstance().AddIpcEvent(event_);
		264	+ }
215	is_created_ = true;	265	is_created_ = true;
216	}	266	}
217		267
218	void NPUEvent::moveHelper(NPUEvent&& other)	268	void NPUEvent::moveHelper(NPUEvent&& other)
219	{	269	{
220	- flags_ = c10_npu::acl::IsExistCreateEventExWithFlag() ? ACL_EVENT_SYNC : ACL_EVENT_DEFAULT;	270	+ std::swap(flags_, other.flags_);
221	std::swap(is_created_, other.is_created_);	271	std::swap(is_created_, other.is_created_);
222	std::swap(was_recorded_, other.was_recorded_);	272	std::swap(was_recorded_, other.was_recorded_);
223	std::swap(device_index_, other.device_index_);	273	std::swap(device_index_, other.device_index_);

Mtorch_npu/csrc/core/npu/NPUEvent.h+10-4

	@@ -9,13 +9,21 @@
9	namespace c10_npu {	9	namespace c10_npu {
10	/*	10	/*
11	* NPUEvents are movable not copyable wrappers around NPU's events.	11	* NPUEvents are movable not copyable wrappers around NPU's events.
12	-* NPUEvents are constructed lazily when first recorded.	12	+*
		13	+* NPUEvents are constructed lazily when first recorded unless it is
		14	+* reconstructed from a aclrtIpcEventHandle. The event has a device, and this
		15	+* device is acquired from the first recording stream. However, if reconstructed
		16	+* from a handle, the device should be explicitly specified; or if ipc_handle() is
		17	+* called before the event is ever recorded, it will use the current device.
		18	+* Later streams that record the event must match this device.
13	*/	19	*/
14	struct C10_NPU_API NPUEvent {	20	struct C10_NPU_API NPUEvent {
15	// Constructors	21	// Constructors
16	// Default value for `flags` is specified below	22	// Default value for `flags` is specified below
17	NPUEvent();	23	NPUEvent();
18	NPUEvent(unsigned int flags) : flags_(flags) {}	24	NPUEvent(unsigned int flags) : flags_(flags) {}
		25	+ NPUEvent(c10::DeviceIndex device_index, const aclrtIpcEventHandle* handle);
		26	+
19	~NPUEvent();	27	~NPUEvent();
20		28
21	NPUEvent(const NPUEvent&) = delete;	29	NPUEvent(const NPUEvent&) = delete;
	@@ -50,8 +58,7 @@ struct C10_NPU_API NPUEvent {
50	uint64_t recorded_time() const;	58	uint64_t recorded_time() const;
51	void synchronize() const;	59	void synchronize() const;
52	void reset(const NPUStream& stream) const;	60	void reset(const NPUStream& stream) const;
53	-	61	+ void ipc_handle(aclrtIpcEventHandle* handle);
54	- // npu do not support IpcEventHandle until now
55		62
56	private:	63	private:
57	unsigned int flags_;	64	unsigned int flags_;
	@@ -64,5 +71,4 @@ private:
64	void createEvent(c10::DeviceIndex device_index);	71	void createEvent(c10::DeviceIndex device_index);
65	void moveHelper(NPUEvent&& other);	72	void moveHelper(NPUEvent&& other);
66	};	73	};
67	-
68	} // namespace c10_npu	74	} // namespace c10_npu

Mtorch_npu/csrc/core/npu/NPUEventManager.cpp+27-2

	@@ -30,6 +30,7 @@ void NPUEventManager::run(aclrtEvent event)
30	return;	30	return;
31	}	31	}
32	ASCEND_LOGI("Event: aclrtDestroyEvent is successfully executed, event=%p", event);	32	ASCEND_LOGI("Event: aclrtDestroyEvent is successfully executed, event=%p", event);
		33	+ RemoveIpcEvent(event);
33	}	34	}
34		35
35	aclError NPUEventManager::QueryAndDestroyEvent()	36	aclError NPUEventManager::QueryAndDestroyEvent()
	@@ -37,7 +38,7 @@ aclError NPUEventManager::QueryAndDestroyEvent()
37	std::lock_guard<std::mutex> guard(event_queue_mutex_);	38	std::lock_guard<std::mutex> guard(event_queue_mutex_);
38	while (!npu_events_.empty()) {	39	while (!npu_events_.empty()) {
39	aclrtEvent event = npu_events_.front();	40	aclrtEvent event = npu_events_.front();
40	- if (c10_npu::option::OptionsManager::GetPerStreamQueue()) {	41	+ if (c10_npu::option::OptionsManager::GetPerStreamQueue() \|\| IsIpcEvent(event)) {
41	if (!c10_npu::NPUEventManager::GetInstance().IsEventRecorded(event) \|\| !c10_npu::NPUEventManager::GetInstance().IsEventWaited(event)) {	42	if (!c10_npu::NPUEventManager::GetInstance().IsEventRecorded(event) \|\| !c10_npu::NPUEventManager::GetInstance().IsEventWaited(event)) {
42	break;	43	break;
43	}	44	}
	@@ -66,7 +67,7 @@ aclError NPUEventManager::QueryAndDestroyEvent()
66		67
67	aclError NPUEventManager::LazyDestroy(aclrtEvent npu_event)	68	aclError NPUEventManager::LazyDestroy(aclrtEvent npu_event)
68	{	69	{
69	- if (c10_npu::acl::IsExistCreateEventExWithFlag() && !c10_npu::option::OptionsManager::GetPerStreamQueue()) {	70	+ if (c10_npu::acl::IsExistCreateEventExWithFlag() && !c10_npu::option::OptionsManager::GetPerStreamQueue() && !IsIpcEvent(npu_event)) {
70	int err = aclrtDestroyEvent(npu_event);	71	int err = aclrtDestroyEvent(npu_event);
71	if (err == ACL_ERROR_NONE) {	72	if (err == ACL_ERROR_NONE) {
72	ASCEND_LOGI("Event: aclrtDestroyEvent is successfully executed, event=%p", npu_event);	73	ASCEND_LOGI("Event: aclrtDestroyEvent is successfully executed, event=%p", npu_event);
	@@ -103,6 +104,8 @@ void NPUEventManager::ClearEvent()
103	}	104	}
104	npu_events_.pop_front();	105	npu_events_.pop_front();
105	}	106	}
		107	+
		108	+ ipc_events_.clear();
106	}	109	}
107	void NPUEventManager::IncreaseUnrecordedCount(aclrtEvent event)	110	void NPUEventManager::IncreaseUnrecordedCount(aclrtEvent event)
108	{	111	{
	@@ -190,4 +193,26 @@ void NPUEventManager::ClearUnrecordedCount()
190	event_unrecorded_count_.clear();	193	event_unrecorded_count_.clear();
191	}	194	}
192		195
		196	+void NPUEventManager::AddIpcEvent(aclrtEvent event)
		197	+{
		198	+ std::lock_guard<std::mutex> guard(ipc_event_mutex_);
		199	+ auto ret = ipc_events_.insert(event);
		200	+ TORCH_CHECK(ret.second,
		201	+ "Event: insert ipc event failed, event=", (void *) event, PTA_ERROR(ErrCode::INTERNAL));
		202	+}
		203	+
		204	+void NPUEventManager::RemoveIpcEvent(aclrtEvent event)
		205	+{
		206	+ std::lock_guard<std::mutex> guard(ipc_event_mutex_);
		207	+ auto it = ipc_events_.find(event);
		208	+ if (it != ipc_events_.end()) {
		209	+ ipc_events_.erase(it);
		210	+ }
		211	+}
		212	+
		213	+bool NPUEventManager::IsIpcEvent(aclrtEvent event)
		214	+{
		215	+ std::lock_guard<std::mutex> guard(ipc_event_mutex_);
		216	+ return ipc_events_.find(event) != ipc_events_.end();
		217	+}
193	} // namespace c10_npu	218	} // namespace c10_npu

Mtorch_npu/csrc/core/npu/NPUEventManager.h+7-0

	@@ -2,6 +2,7 @@
2		2
3	#include <deque>	3	#include <deque>
4	#include <mutex>	4	#include <mutex>
		5	+#include <unordered_set>
5	#include <c10/core/thread_pool.h>	6	#include <c10/core/thread_pool.h>
6	#include <c10/util/flat_hash_map.h>	7	#include <c10/util/flat_hash_map.h>
7	#include <third_party/acl/inc/acl/acl.h>	8	#include <third_party/acl/inc/acl/acl.h>
	@@ -25,10 +26,13 @@ public:
25	void DecreaseUnwaitedCount(aclrtEvent event);	26	void DecreaseUnwaitedCount(aclrtEvent event);
26	bool IsEventWaited(aclrtEvent event);	27	bool IsEventWaited(aclrtEvent event);
27	void ClearUnrecordedCount();	28	void ClearUnrecordedCount();
		29	+ void AddIpcEvent(aclrtEvent event);
28	~NPUEventManager() {}	30	~NPUEventManager() {}
29		31
30	private:	32	private:
31	void run(aclrtEvent event);	33	void run(aclrtEvent event);
		34	+ void RemoveIpcEvent(aclrtEvent event);
		35	+ bool IsIpcEvent(aclrtEvent event);
32		36
33	private:	37	private:
34	std::mutex event_queue_mutex_;	38	std::mutex event_queue_mutex_;
	@@ -40,6 +44,9 @@ private:
40	ska::flat_hash_map<aclrtEvent, int> event_unrecorded_count_;	44	ska::flat_hash_map<aclrtEvent, int> event_unrecorded_count_;
41	std::mutex event_unwaited_count_mutex_;	45	std::mutex event_unwaited_count_mutex_;
42	ska::flat_hash_map<aclrtEvent, int> event_unwaited_count_;	46	ska::flat_hash_map<aclrtEvent, int> event_unwaited_count_;
		47	+
		48	+ std::mutex ipc_event_mutex_;
		49	+ std::unordered_set<aclrtEvent> ipc_events_;
43	};	50	};
44		51
45	} // namespace c10_npu	52	} // namespace c10_npu

Mtorch_npu/csrc/core/npu/interface/AclInterface.cpp+57-0

	@@ -27,6 +27,8 @@ LOAD_FUNCTION(aclrtCreateEventWithFlag)
27	LOAD_FUNCTION(aclrtCreateEventExWithFlag)	27	LOAD_FUNCTION(aclrtCreateEventExWithFlag)
28	LOAD_FUNCTION(aclrtQueryEventWaitStatus)	28	LOAD_FUNCTION(aclrtQueryEventWaitStatus)
29	LOAD_FUNCTION(aclrtQueryEventStatus)	29	LOAD_FUNCTION(aclrtQueryEventStatus)
		30	+LOAD_FUNCTION(aclrtIpcGetEventHandle)
		31	+LOAD_FUNCTION(aclrtIpcOpenEventHandle)
30	LOAD_FUNCTION(aclprofCreateStepInfo)	32	LOAD_FUNCTION(aclprofCreateStepInfo)
31	LOAD_FUNCTION(aclprofGetStepTimestamp)	33	LOAD_FUNCTION(aclprofGetStepTimestamp)
32	LOAD_FUNCTION(aclprofDestroyStepInfo)	34	LOAD_FUNCTION(aclprofDestroyStepInfo)
	@@ -326,6 +328,61 @@ bool IsExistQueryEventRecordedStatus()
326	}	328	}
327	}	329	}
328		330
		331	+aclError AclIpcGetEventHandle(aclrtEvent event, aclrtIpcEventHandle *handle)
		332	+{
		333	+ typedef aclError (aclIpcGetEventHandle)(aclrtEvent event, aclrtIpcEventHandle handle);
		334	+ static aclIpcGetEventHandle func = nullptr;
		335	+ if (func == nullptr) {
		336	+ func = (aclIpcGetEventHandle)GET_FUNC(aclrtIpcGetEventHandle);
		337	+ }
		338	+ TORCH_CHECK(func, "Failed to find function ", "aclrtIpcGetEventHandle", PTA_ERROR(ErrCode::NOT_FOUND));
		339	+ return func(event, handle);
		340	+}
		341	+
		342	+aclError AclIpcOpenEventHandle(aclrtIpcEventHandle handle, aclrtEvent *event)
		343	+{
		344	+ typedef aclError (aclIpcOpenEventHandle)(aclrtIpcEventHandle handle, aclrtEvent event);
		345	+ static aclIpcOpenEventHandle func = nullptr;
		346	+ if (func == nullptr) {
		347	+ func = (aclIpcOpenEventHandle)GET_FUNC(aclrtIpcOpenEventHandle);
		348	+ }
		349	+ TORCH_CHECK(func, "Failed to find function ", "aclrtIpcOpenEventHandle", PTA_ERROR(ErrCode::NOT_FOUND));
		350	+ return func(handle, event);
		351	+}
		352	+
		353	+bool IsSupportIpcEvent()
		354	+{
		355	+ const static bool is_support = []() -> bool {
		356	+ if (!IsExistCreateEventExWithFlag()) {
		357	+ ASCEND_LOGD("IsSupportIpcEvent return false because aclrtCreateEventExWithFlag does not exist.");
		358	+ return false;
		359	+ }
		360	+
		361	+ auto func = GET_FUNC(aclrtIpcGetEventHandle);
		362	+ if (func == nullptr) {
		363	+ ASCEND_LOGD("IsSupportIpcEvent return false because aclrtIpcGetEventHandle does not exist.");
		364	+ return false;
		365	+ }
		366	+
		367	+ c10::DeviceIndex device_index = c10_npu::current_device();
		368	+ c10_npu::LazySetDevice(device_index);
		369	+
		370	+ aclrtEvent npu_event = nullptr;
		371	+ auto ret = c10_npu::acl::AclrtCreateEventWithFlag(&npu_event, ACL_EVENT_IPC);
		372	+ if (ret == ACL_ERROR_RT_FEATURE_NOT_SUPPORT) {
		373	+ ASCEND_LOGD("IsSupportIpcEvent return false because create event with flag ACL_EVENT_IPC failed.");
		374	+ return false;
		375	+ }
		376	+ NPU_CHECK_ERROR(ret);
		377	+ NPU_CHECK_ERROR(aclrtDestroyEvent(npu_event));
		378	+
		379	+ ASCEND_LOGD("IsSupportIpcEvent return true.");
		380	+ return true;
		381	+ }();
		382	+
		383	+ return is_support;
		384	+}
		385	+
329	aclError AclProfilingInit(const char *profilerResultPath, size_t length) {	386	aclError AclProfilingInit(const char *profilerResultPath, size_t length) {
330	typedef aclError (AclProfInitFunc) (const char , size_t);	387	typedef aclError (AclProfInitFunc) (const char , size_t);
331	static AclProfInitFunc func = nullptr;	388	static AclProfInitFunc func = nullptr;

Mtorch_npu/csrc/core/npu/interface/AclInterface.h+7-0

	@@ -12,6 +12,7 @@
12	struct aclrtMemUsageInfo;	12	struct aclrtMemUsageInfo;
13	struct aclOpExecutor;	13	struct aclOpExecutor;
14	struct aclrtUuid;	14	struct aclrtUuid;
		15	+struct aclrtIpcEventHandle;
15	namespace c10_npu {	16	namespace c10_npu {
16	namespace acl {	17	namespace acl {
17	enum aclrtEventWaitStatus {	18	enum aclrtEventWaitStatus {
	@@ -123,6 +124,12 @@ bool IsExistQueryEventRecordedStatus();
123	*/	124	*/
124	aclError AclQueryEventRecordedStatus(aclrtEvent event, aclrtEventRecordedStatus *status);	125	aclError AclQueryEventRecordedStatus(aclrtEvent event, aclrtEventRecordedStatus *status);
125		126
		127	+aclError AclIpcGetEventHandle(aclrtEvent event, aclrtIpcEventHandle *handle);
		128	+
		129	+aclError AclIpcOpenEventHandle(aclrtIpcEventHandle handle, aclrtEvent *event);
		130	+
		131	+bool IsSupportIpcEvent();
		132	+
126	aclError AclProfilingInit(const char *profilerResultPath, size_t length);	133	aclError AclProfilingInit(const char *profilerResultPath, size_t length);
127	aclError AclProfilingStart(const aclprofConfig *profilerConfig);	134	aclError AclProfilingStart(const aclprofConfig *profilerConfig);
128	aclError AclProfilingStop(const aclprofConfig *profilerConfig);	135	aclError AclProfilingStop(const aclprofConfig *profilerConfig);

Mtorch_npu/csrc/ipc/NPUIPCTypes.cpp+23-10

	@@ -5,6 +5,7 @@
5	#include <random>	5	#include <random>
6	#include <string>	6	#include <string>
7	#include "torch_npu/csrc/core/npu/NPUGuard.h"	7	#include "torch_npu/csrc/core/npu/NPUGuard.h"
		8	+#include "torch_npu/csrc/core/npu/interface/AsyncTaskQueueInterface.h"
8	#include "torch_npu/csrc/ipc/NPUIPCTypes.h"	9	#include "torch_npu/csrc/ipc/NPUIPCTypes.h"
9		10
10	#include "third_party/acl/inc/acl/acl_base.h"	11	#include "third_party/acl/inc/acl/acl_base.h"
	@@ -167,13 +168,31 @@ NpuIPCSentData::NpuIPCSentData(
167	counter_ptr_(counter_ptr),	168	counter_ptr_(counter_ptr),
168	device_(device)	169	device_(device)
169	{	170	{
170	- if (npu_ipc_global_entities.sync_events_used_.load() <	171	+ // NPU have the unofficial limit on the number of recorded blocking
171	- NPU_IPC_MAXIMUM_EVENTS_TO_USE) {	172	+ // interprocess events, to prevent using of all events, we are switching to
172	- // NPU does not suppurt event_sync in IPC now.	173	+ // StreamSync before limit reached.
		174	+ //
		175	+ // ```python
		176	+ // import torch
		177	+ // a = [ torch.npu.Event(
		178	+ // enable_timing=False, blocking=True, interprocess=True) for i in
		179	+ // range(30000) ]
		180	+ // [i.record() for i in a]
		181	+ // ```
		182	+ //
		183	+ if (c10_npu::acl::IsSupportIpcEvent() &&
		184	+ npu_ipc_global_entities.sync_events_used_.load() < NPU_IPC_MAXIMUM_EVENTS_TO_USE) {
		185	+ // More efficient would be to create event inside of main thread (at
		186	+ // the moment of the queue.put). The reason this is more efficient is
		187	+ // because the main thread may have queued extra work on the stream, which
		188	+ // this event will consequently wait for (uselessly).
		189	+ npu_ipc_global_entities.sync_events_used_++;
		190	+ event_ = c10_npu::NPUEvent(ACL_EVENT_IPC);
		191	+ event_.record(c10_npu::getCurrentNPUStream(device.index()));
		192	+ event_sync_required_ = true;
173	} else {	193	} else {
174	auto stream = c10_npu::getCurrentNPUStream(device.index());	194	auto stream = c10_npu::getCurrentNPUStream(device.index());
175	c10_npu::stream_synchronize(stream);	195	c10_npu::stream_synchronize(stream);
176	- event_ = nullptr;
177	event_sync_required_ = false;	196	event_sync_required_ = false;
178	}	197	}
179	}	198	}
	@@ -181,12 +200,6 @@ NpuIPCSentData::NpuIPCSentData(
181	NpuIPCSentData::~NpuIPCSentData()	200	NpuIPCSentData::~NpuIPCSentData()
182	{	201	{
183	ReturnRefCounter(handle_, offset_);	202	ReturnRefCounter(handle_, offset_);
184	- try {
185	- if (event_sync_required_) {
186	- // NPU does not suppurt event_sync in IPC now.
187	- }
188	- } catch (...) { /* No throw */
189	- }
190	}	203	}
191		204
192	uint64_t NpuIPCSentData::counter_value()	205	uint64_t NpuIPCSentData::counter_value()

Mtorch_npu/csrc/ipc/NPUIPCTypes.h+3-2

	@@ -5,6 +5,7 @@
5	#include "torch_npu/csrc/core/npu/NPUFunctions.h"	5	#include "torch_npu/csrc/core/npu/NPUFunctions.h"
6	#include "torch_npu/csrc/core/npu/NPUStream.h"	6	#include "torch_npu/csrc/core/npu/NPUStream.h"
7	#include "torch_npu/csrc/core/npu/NPUCachingAllocator.h"	7	#include "torch_npu/csrc/core/npu/NPUCachingAllocator.h"
		8	+#include "torch_npu/csrc/core/npu/NPUEvent.h"
8		9
9	namespace torch_npu {	10	namespace torch_npu {
10	namespace ipc {	11	namespace ipc {
	@@ -23,7 +24,7 @@ struct NpuIPCSentData final {
23	uint64_t offset_;	24	uint64_t offset_;
24	uint64_t* counter_ptr_; // Reference counter shared memory block	25	uint64_t* counter_ptr_; // Reference counter shared memory block
25	at::DataPtr original_ptr_; // Original mem allocation	26	at::DataPtr original_ptr_; // Original mem allocation
26	- char* event_; // Sync event	27	+ c10_npu::NPUEvent event_; // Sync event
27	bool event_sync_required_;	28	bool event_sync_required_;
28	at::Device device_;	29	at::Device device_;
29		30
	@@ -57,7 +58,7 @@ namespace {
57		58
58	inline constexpr int64_t NPU_IPC_REF_COUNTER_FILE_SIZE = 10000;	59	inline constexpr int64_t NPU_IPC_REF_COUNTER_FILE_SIZE = 10000;
59	inline constexpr int64_t NPU_IPC_WARN_AFTER_X_BLOCKS_IN_LIMBO = 1000;	60	inline constexpr int64_t NPU_IPC_WARN_AFTER_X_BLOCKS_IN_LIMBO = 1000;
60	-inline constexpr int64_t NPU_IPC_MAXIMUM_EVENTS_TO_USE = 0;	61	+inline constexpr int64_t NPU_IPC_MAXIMUM_EVENTS_TO_USE = 1000;
61		62
62	// All to be deleted data blocks with non zero reference counter goes there	63	// All to be deleted data blocks with non zero reference counter goes there
63	struct NpuIPCSentDataLimbo final {	64	struct NpuIPCSentDataLimbo final {

Mtorch_npu/csrc/ipc/StorageSharing.cpp+20-8

	@@ -12,9 +12,11 @@
12		12
13	#include "torch_npu/csrc/core/NPUStorageImpl.h"	13	#include "torch_npu/csrc/core/NPUStorageImpl.h"
14	#include "torch_npu/csrc/core/NPUBridge.h"	14	#include "torch_npu/csrc/core/NPUBridge.h"
		15	+#include "torch_npu/csrc/core/npu/interface/AclInterface.h"
15	#include "torch_npu/csrc/core/npu/NPUCachingAllocator.h"	16	#include "torch_npu/csrc/core/npu/NPUCachingAllocator.h"
16	#include "torch_npu/csrc/core/npu/NPUGuard.h"	17	#include "torch_npu/csrc/core/npu/NPUGuard.h"
17	#include "torch_npu/csrc/core/NPUStorageImpl.h"	18	#include "torch_npu/csrc/core/NPUStorageImpl.h"
		19	+#include "torch_npu/csrc/core/npu/interface/AsyncTaskQueueInterface.h"
18	#include "torch_npu/csrc/framework/FormatHelper.h"	20	#include "torch_npu/csrc/framework/FormatHelper.h"
19		21
20	#include "torch_npu/csrc/ipc/NPUIPCTypes.h"	22	#include "torch_npu/csrc/ipc/NPUIPCTypes.h"
	@@ -76,15 +78,15 @@ static PyObject* THNPStorage_shareNpu(PyObject* self, PyObject* args)
76	_ref_counter = PyBytes_FromString((sent_data->handle()).c_str());	78	_ref_counter = PyBytes_FromString((sent_data->handle()).c_str());
77	_ref_counter_offset = THPUtils_packUInt64(sent_data->offset());	79	_ref_counter_offset = THPUtils_packUInt64(sent_data->offset());
78		80
79	- // NOLINTNEXTLINE(cppcoreguidelines-init-variables)	81	+ aclrtIpcEventHandle ipc_event_handle{};
80	- aclrtNotify ipc_event_handle;
81		82
82	if (sent_data->event_sync_required_) {	83	if (sent_data->event_sync_required_) {
83	- // NPU does not suppurt event_sync in IPC now.	84	+ sent_data->event_.ipc_handle(&ipc_event_handle);
84	}	85	}
85		86
86	_event_handle = PyBytes_FromStringAndSize(	87	_event_handle = PyBytes_FromStringAndSize(
87	- (char*)&ipc_event_handle, sizeof(aclrtNotify));	88	+ reinterpret_cast<const char*>(&ipc_event_handle), ACL_IPC_EVENT_HANDLE_SIZE);
		89	+
88	_event_sync_required = PyBool_FromLong(sent_data->event_sync_required_);	90	_event_sync_required = PyBool_FromLong(sent_data->event_sync_required_);
89	}	91	}
90		92
	@@ -148,13 +150,13 @@ static PyObject* THNPStorage_releaseIPCCounter(PyObject* _unused, PyObject* args
148	END_HANDLE_TH_ERRORS	150	END_HANDLE_TH_ERRORS
149	}	151	}
150		152
151	-static std::string THNPStorage_bytesAsHandleString(PyObject* handle)	153	+static std::string THNPStorage_bytesAsHandleString(PyObject* handle, ssize_t expected_size)
152	{	154	{
153	HANDLE_TH_ERRORS	155	HANDLE_TH_ERRORS
154	char* buffer = nullptr;	156	char* buffer = nullptr;
155	Py_ssize_t handle_size = 0;	157	Py_ssize_t handle_size = 0;
156	if (PyBytes_AsStringAndSize(handle, &buffer, &handle_size) == -1) {	158	if (PyBytes_AsStringAndSize(handle, &buffer, &handle_size) == -1) {
157	- TORCH_CHECK(handle_size == ACL_IPC_HANDLE_SIZE, "incorrect handle", PTA_ERROR(ErrCode::PARAM));	159	+ TORCH_CHECK(handle_size == expected_size, "incorrect handle", PTA_ERROR(ErrCode::PARAM));
158	}	160	}
159	return std::string(buffer, handle_size);	161	return std::string(buffer, handle_size);
160	END_HANDLE_TH_ERRORS_RET("")	162	END_HANDLE_TH_ERRORS_RET("")
	@@ -197,10 +199,20 @@ static PyObject* THNPStorage_newSharedNpu(PyObject* _unused, PyObject* args)
197	c10_npu::LazySetDevice(device);	199	c10_npu::LazySetDevice(device);
198		200
199	if (PyObject_IsTrue(_event_sync_required)) {	201	if (PyObject_IsTrue(_event_sync_required)) {
200	- // TO BE DONE	202	+ // Ensure that producer prepared all tensor's data
		203	+ std::string s_ipc_event_handle =
		204	+ THNPStorage_bytesAsHandleString(_event_handle, ACL_IPC_EVENT_HANDLE_SIZE);
		205	+ if (s_ipc_event_handle.empty()) {
		206	+ return nullptr;
		207	+ }
		208	+
		209	+ auto ipc_event_handle = reinterpret_cast<const aclrtIpcEventHandle*>(
		210	+ s_ipc_event_handle.c_str());
		211	+ c10_npu::NPUEvent npu_event(device, ipc_event_handle);
		212	+ npu_event.block(c10_npu::getCurrentNPUStream(device));
201	}	213	}
202		214
203	- std::string s_handle = THNPStorage_bytesAsHandleString(_handle);	215	+ std::string s_handle = THNPStorage_bytesAsHandleString(_handle, ACL_IPC_HANDLE_SIZE);
204	if (s_handle.empty()) {	216	if (s_handle.empty()) {
205	return nullptr;	217	return nullptr;
206	}	218	}

Mtorch_npu/csrc/npu/Event.cpp+81-0

	@@ -27,6 +27,18 @@ static PyObject* THNPEvent_pynew(PyTypeObject type, PyObject args, PyObject *k
27	return nullptr;	27	return nullptr;
28	}	28	}
29		29
		30	+ TORCH_CHECK(
		31	+ !interprocess \|\| c10_npu::acl::IsSupportIpcEvent(),
		32	+ "Parameter interprocess is not supported, please upgrade the HDK(driver) or CANN package.",
		33	+ PTA_ERROR(ErrCode::NOT_SUPPORT));
		34	+
		35	+ // Runtime requires that the ACL_EVENT_IPC cannot be used together with other flags.
		36	+ // If this restriction is removed in the future, this check can be removed.
		37	+ TORCH_CHECK(
		38	+ !interprocess \|\| (!enable_timing && !external),
		39	+ "Parameter interprocess cannot be specified together with other parameters.",
		40	+ PTA_ERROR(ErrCode::NOT_SUPPORT));
		41	+
30	THPObjectPtr ptr(type->tp_alloc(type, 0));	42	THPObjectPtr ptr(type->tp_alloc(type, 0));
31	if (!ptr) {	43	if (!ptr) {
32	return nullptr;	44	return nullptr;
	@@ -40,6 +52,9 @@ static PyObject* THNPEvent_pynew(PyTypeObject type, PyObject args, PyObject *k
40	} else {	52	} else {
41	flags = enable_timing ? ACL_EVENT_TIME_LINE : ACL_EVENT_DEFAULT;	53	flags = enable_timing ? ACL_EVENT_TIME_LINE : ACL_EVENT_DEFAULT;
42	}	54	}
		55	+ if (interprocess) {
		56	+ flags = ACL_EVENT_IPC;
		57	+ }
43	if (external) {	58	if (external) {
44	flags = ACL_EVENT_EXTERNAL;	59	flags = ACL_EVENT_EXTERNAL;
45	}	60	}
	@@ -49,6 +64,57 @@ static PyObject* THNPEvent_pynew(PyTypeObject type, PyObject args, PyObject *k
49	END_HANDLE_TH_ERRORS	64	END_HANDLE_TH_ERRORS
50	}	65	}
51		66
		67	+static PyObject* THNPEvent_from_ipc_handle(
		68	+ PyObject* _type,
		69	+ PyObject* args,
		70	+ PyObject* kwargs)
		71	+{
		72	+ HANDLE_TH_ERRORS
		73	+ TORCH_CHECK(
		74	+ c10_npu::acl::IsSupportIpcEvent(),
		75	+ "The from_ipc_handle method is not supported, please upgrade the HDK(driver) or CANN package.",
		76	+ PTA_ERROR(ErrCode::NOT_SUPPORT));
		77	+
		78	+ auto type = (PyTypeObject*)_type;
		79	+
		80	+ static torch::PythonArgParser parser({
		81	+ "from_ipc_handle(Device device, std::string ipc_handle)",
		82	+ });
		83	+ constexpr int kArgCount = 2;
		84	+ torch::ParsedArgs<kArgCount> parsed_args;
		85	+ auto r = parser.parse(args, kwargs, parsed_args);
		86	+
		87	+ at::Device device = r.device(0);
		88	+ std::string handle_string = r.string(1);
		89	+
		90	+ TORCH_CHECK(
		91	+ handle_string.size() == sizeof(aclrtIpcEventHandle),
		92	+ "aclrtIpcEventHandle expects byte-like object of size ",
		93	+ sizeof(aclrtIpcEventHandle),
		94	+ ", but got ",
		95	+ handle_string.size(),
		96	+ PTA_ERROR(ErrCode::PARAM));
		97	+ TORCH_CHECK(
		98	+ device.type() == c10::DeviceType::PrivateUse1,
		99	+ "Event can only be created on "
		100	+ "PrivateUse1 devices, but got device type ",
		101	+ device.type(),
		102	+ PTA_ERROR(ErrCode::PARAM))
		103	+
		104	+ THPObjectPtr ptr(type->tp_alloc(type, 0));
		105	+ if (!ptr) {
		106	+ return nullptr;
		107	+ }
		108	+ THNPEvent* self = (THNPEvent*)ptr.get();
		109	+
		110	+ aclrtIpcEventHandle handle{};
		111	+ std::memcpy(&handle, handle_string.c_str(), handle_string.size());
		112	+ new (&self->npu_event) c10_npu::NPUEvent(device.index(), &handle);
		113	+
		114	+ return (PyObject*)ptr.release();
		115	+ END_HANDLE_TH_ERRORS
		116	+}
		117	+
52	static void THNPEvent_dealloc(THNPEvent *self)	118	static void THNPEvent_dealloc(THNPEvent *self)
53	{	119	{
54	self->npu_event.~NPUEvent();	120	self->npu_event.~NPUEvent();
	@@ -139,6 +205,16 @@ static PyObject* THNPEvent_reset(THNPEvent self, THNPStream stream)
139	END_HANDLE_TH_ERRORS	205	END_HANDLE_TH_ERRORS
140	}	206	}
141		207
		208	+static PyObject* THNPEvent_ipc_handle(PyObject* _self, PyObject* noargs)
		209	+{
		210	+ HANDLE_TH_ERRORS
		211	+ auto self = (THNPEvent*)_self;
		212	+ aclrtIpcEventHandle handle{};
		213	+ self->npu_event.ipc_handle(&handle);
		214	+ return PyBytes_FromStringAndSize((const char*)&handle, sizeof(handle));
		215	+ END_HANDLE_TH_ERRORS
		216	+}
		217	+
142	static struct PyGetSetDef THNPEvent_properties[] = {	218	static struct PyGetSetDef THNPEvent_properties[] = {
143	{"device", (getter)THNPEvent_get_device, nullptr, nullptr, nullptr},	219	{"device", (getter)THNPEvent_get_device, nullptr, nullptr, nullptr},
144	{"npu_event", (getter)THNPEvent_get_npu_event, nullptr, nullptr, nullptr},	220	{"npu_event", (getter)THNPEvent_get_npu_event, nullptr, nullptr, nullptr},
	@@ -146,6 +222,10 @@ static struct PyGetSetDef THNPEvent_properties[] = {
146	};	222	};
147		223
148	static PyMethodDef THNPEvent_methods[] = {	224	static PyMethodDef THNPEvent_methods[] = {
		225	+ {(char*)"from_ipc_handle",
		226	+ (PyCFunction)THNPEvent_from_ipc_handle,
		227	+ METH_CLASS \| METH_VARARGS \| METH_KEYWORDS,
		228	+ nullptr},
149	{(char*)"record", (PyCFunction)THNPEvent_record, METH_O, nullptr},	229	{(char*)"record", (PyCFunction)THNPEvent_record, METH_O, nullptr},
150	{(char*)"wait", (PyCFunction)THNPEvent_wait, METH_O, nullptr},	230	{(char*)"wait", (PyCFunction)THNPEvent_wait, METH_O, nullptr},
151	{(char*)"query", (PyCFunction)THNPEvent_query, METH_NOARGS, nullptr},	231	{(char*)"query", (PyCFunction)THNPEvent_query, METH_NOARGS, nullptr},
	@@ -153,6 +233,7 @@ static PyMethodDef THNPEvent_methods[] = {
153	{(char*)"recorded_time", (PyCFunction)THNPEvent_recorded_time, METH_NOARGS, nullptr},	233	{(char*)"recorded_time", (PyCFunction)THNPEvent_recorded_time, METH_NOARGS, nullptr},
154	{(char*)"synchronize", (PyCFunction)THNPEvent_synchronize, METH_NOARGS, nullptr},	234	{(char*)"synchronize", (PyCFunction)THNPEvent_synchronize, METH_NOARGS, nullptr},
155	{(char*)"reset", (PyCFunction)THNPEvent_reset, METH_O, nullptr},	235	{(char*)"reset", (PyCFunction)THNPEvent_reset, METH_O, nullptr},
		236	+ {(char*)"ipc_handle", THNPEvent_ipc_handle, METH_NOARGS, nullptr},
156	{nullptr}	237	{nullptr}
157	};	238	};
158		239

Mtorch_npu/csrc/sanitizer/PyCallbackTrigger.h+12-0

	@@ -60,6 +60,18 @@ struct PyCallbackTrigger {
60	CONCRETE_TRACE_NPU("NPUEventWaitCallbacks", event, stream);	60	CONCRETE_TRACE_NPU("NPUEventWaitCallbacks", event, stream);
61	}	61	}
62	}	62	}
		63	+ void traceNpuEventGetHandle(uintptr_t event, uintptr_t handle) const
		64	+ {
		65	+ if (sanitizer_mode == SanitizerMode::STREAM) {
		66	+ CONCRETE_TRACE_NPU("NPUEventGetHandleCallbacks", event, handle);
		67	+ }
		68	+ }
		69	+ void traceNpuEventOpenHandle(uintptr_t handle, uintptr_t event) const
		70	+ {
		71	+ if (sanitizer_mode == SanitizerMode::STREAM) {
		72	+ CONCRETE_TRACE_NPU("NPUEventOpenHandleCallbacks", handle, event);
		73	+ }
		74	+ }
63	void traceNpuMemoryAllocation(uintptr_t ptr) const	75	void traceNpuMemoryAllocation(uintptr_t ptr) const
64	{	76	{
65	if (sanitizer_mode == SanitizerMode::STREAM) {	77	if (sanitizer_mode == SanitizerMode::STREAM) {

Mtorch_npu/multiprocessing/reductions.py+13-1

	@@ -1,4 +1,4 @@
1	-__all__ = ["rebuild_npu_tensor"]	1	+__all__ = ["rebuild_npu_event", "rebuild_npu_tensor"]
2		2
3	import multiprocessing	3	import multiprocessing
4	import torch	4	import torch
	@@ -17,6 +17,15 @@ from torch.multiprocessing.reductions import (
17	import torch_npu	17	import torch_npu
18		18
19		19
		20	+def rebuild_npu_event(device, handle):
		21	+ return torch.npu.Event.from_ipc_handle(device, handle)
		22	+
		23	+
		24	+def _npu_reduce_event(event):
		25	+ handle = event.ipc_handle()
		26	+ return (rebuild_npu_event, (event.device, handle))
		27	+
		28	+
20	def rebuild_npu_tensor(	29	def rebuild_npu_tensor(
21	tensor_cls,	30	tensor_cls,
22	tensor_size,	31	tensor_size,
	@@ -186,6 +195,9 @@ def _npu_reduce_storage(storage):
186		195
187		196
188	def _add_reductions_methods():	197	def _add_reductions_methods():
		198	+ multiprocessing.reduction.register(torch.npu.Event, _npu_reduce_event)
		199	+
		200	+ torch.multiprocessing.reductions.reduce_event = _npu_reduce_event
189	torch.multiprocessing.reductions.reduce_tensor = _npu_reduce_tensor	201	torch.multiprocessing.reductions.reduce_tensor = _npu_reduce_tensor
190	torch.multiprocessing.reductions.reduce_storage = _npu_reduce_storage	202	torch.multiprocessing.reductions.reduce_storage = _npu_reduce_storage
191		203

Mtorch_npu/npu/streams.py+12-0

	@@ -135,6 +135,11 @@ class Event(torch_npu._C._NPUEventBase):
135	return super(Event, cls).__new__(cls, enable_timing=enable_timing, blocking=blocking,	135	return super(Event, cls).__new__(cls, enable_timing=enable_timing, blocking=blocking,
136	interprocess=interprocess, graph_external=False)	136	interprocess=interprocess, graph_external=False)
137		137
		138	+ @classmethod
		139	+ def from_ipc_handle(cls, device, handle):
		140	+ r"""Reconstruct an event from an IPC handle on the given device."""
		141	+ return super().from_ipc_handle(device, handle)
		142	+
138	def record(self, stream=None):	143	def record(self, stream=None):
139	r"""Records the event in a given stream.	144	r"""Records the event in a given stream.
140		145
	@@ -186,6 +191,13 @@ class Event(torch_npu._C._NPUEventBase):
186	"""	191	"""
187	super(Event, self).synchronize()	192	super(Event, self).synchronize()
188		193
		194	+ def ipc_handle(self):
		195	+ r"""Return an IPC handle of this event.
		196	+
		197	+ If not recorded yet, the event will use the current device.
		198	+ """
		199	+ return super().ipc_handle()
		200	+
189	@property	201	@property
190	def _as_parameter_(self):	202	def _as_parameter_(self):
191	return ctypes.c_void_p(self.npu_event)	203	return ctypes.c_void_p(self.npu_event)