We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[Bug]: convolve with distributed kernel on multiple GPUs
### What happened?

convolve does not work if the kernel is distributed when more than one GPU is available.

### Code snippet triggering the error

```python
import heat as ht

dis_signal = ht.arange(0, 16, split=0, device='gpu', dtype=ht.int)
dis_kernel_odd = ht.ones(3, split=0, dtype=ht.int, device='gpu')
conv = ht.convolve(dis_signal, dis_kernel_odd, mode='full')
```


### Error message or erroneous outcome

```shell
$ CUDA_VISIBLE_DEVICES=0,1,2,3 srun --ntasks=2 -l python test.py 
1:Traceback (most recent call last):
1:   File ".../test.py", line 7, in <module>
1:     conv = ht.convolve(dis_signal, dis_kernel_odd, mode='full')
1:   File ".../heat-venv_2023/lib/python3.10/site-packages/heat/core/signal.py", line 161, in convolve
1:     local_signal_filtered = fc.conv1d(signal, t_v1)
1: RuntimeError: Expected all tensors to be on the same device, but found at least two devices, cuda:1 and cuda:0! (when checking argument for argument weight in method wrapper__cudnn_convolution)
```


### Version

main (development branch)

### Python version

3.10

### PyTorch version

1.12

### MPI version

```shell
OpenMPI 4.1.4
```


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `heat/core/signal.py`
Content:
```
1 """Provides a collection of signal-processing operations"""
2 
3 import torch
4 import numpy as np
5 
6 from .communication import MPI
7 from .dndarray import DNDarray
8 from .types import promote_types
9 from .manipulations import pad, flip
10 from .factories import array, zeros
11 import torch.nn.functional as fc
12 
13 __all__ = ["convolve"]
14 
15 
16 def convolve(a: DNDarray, v: DNDarray, mode: str = "full") -> DNDarray:
17     """
18     Returns the discrete, linear convolution of two one-dimensional `DNDarray`s or scalars.
19 
20     Parameters
21     ----------
22     a : DNDarray or scalar
23         One-dimensional signal `DNDarray` of shape (N,) or scalar.
24     v : DNDarray or scalar
25         One-dimensional filter weight `DNDarray` of shape (M,) or scalar.
26     mode : str
27         Can be 'full', 'valid', or 'same'. Default is 'full'.
28         'full':
29           Returns the convolution at
30           each point of overlap, with an output shape of (N+M-1,). At
31           the end-points of the convolution, the signals do not overlap
32           completely, and boundary effects may be seen.
33         'same':
34           Mode 'same' returns output  of length 'N'. Boundary
35           effects are still visible. This mode is not supported for
36           even-sized filter weights
37         'valid':
38           Mode 'valid' returns output of length 'N-M+1'. The
39           convolution product is only given for points where the signals
40           overlap completely. Values outside the signal boundary have no
41           effect.
42 
43     Examples
44     --------
45     Note how the convolution operator flips the second array
46     before "sliding" the two across one another:
47 
48     >>> a = ht.ones(10)
49     >>> v = ht.arange(3).astype(ht.float)
50     >>> ht.convolve(a, v, mode='full')
51     DNDarray([0., 1., 3., 3., 3., 3., 2.])
52     >>> ht.convolve(a, v, mode='same')
53     DNDarray([1., 3., 3., 3., 3.])
54     >>> ht.convolve(a, v, mode='valid')
55     DNDarray([3., 3., 3.])
56     >>> a = ht.ones(10, split = 0)
57     >>> v = ht.arange(3, split = 0).astype(ht.float)
58     >>> ht.convolve(a, v, mode='valid')
59     DNDarray([3., 3., 3., 3., 3., 3., 3., 3.])
60 
61     [0/3] DNDarray([3., 3., 3.])
62     [1/3] DNDarray([3., 3., 3.])
63     [2/3] DNDarray([3., 3.])
64     >>> a = ht.ones(10, split = 0)
65     >>> v = ht.arange(3, split = 0)
66     >>> ht.convolve(a, v)
67     DNDarray([0., 1., 3., 3., 3., 3., 3., 3., 3., 3., 3., 2.], dtype=ht.float32, device=cpu:0, split=0)
68 
69     [0/3] DNDarray([0., 1., 3., 3.])
70     [1/3] DNDarray([3., 3., 3., 3.])
71     [2/3] DNDarray([3., 3., 3., 2.])
72     """
73     if np.isscalar(a):
74         a = array([a])
75     if np.isscalar(v):
76         v = array([v])
77     if not isinstance(a, DNDarray):
78         try:
79             a = array(a)
80         except TypeError:
81             raise TypeError("non-supported type for signal: {}".format(type(a)))
82     if not isinstance(v, DNDarray):
83         try:
84             v = array(v)
85         except TypeError:
86             raise TypeError("non-supported type for filter: {}".format(type(v)))
87     promoted_type = promote_types(a.dtype, v.dtype)
88     a = a.astype(promoted_type)
89     v = v.astype(promoted_type)
90 
91     if len(a.shape) != 1 or len(v.shape) != 1:
92         raise ValueError("Only 1-dimensional input DNDarrays are allowed")
93     if mode == "same" and v.shape[0] % 2 == 0:
94         raise ValueError("Mode 'same' cannot be used with even-sized kernel")
95     if not v.is_balanced():
96         raise ValueError("Only balanced kernel weights are allowed")
97 
98     if v.shape[0] > a.shape[0]:
99         a, v = v, a
100 
101     # compute halo size
102     halo_size = torch.max(v.lshape_map[:, 0]).item() // 2
103 
104     # pad DNDarray with zeros according to mode
105     if mode == "full":
106         pad_size = v.shape[0] - 1
107         gshape = v.shape[0] + a.shape[0] - 1
108     elif mode == "same":
109         pad_size = v.shape[0] // 2
110         gshape = a.shape[0]
111     elif mode == "valid":
112         pad_size = 0
113         gshape = a.shape[0] - v.shape[0] + 1
114     else:
115         raise ValueError("Supported modes are 'full', 'valid', 'same', got {}".format(mode))
116 
117     a = pad(a, pad_size, "constant", 0)
118 
119     if a.is_distributed():
120         if (v.lshape_map[:, 0] > a.lshape_map[:, 0]).any():
121             raise ValueError(
122                 "Local chunk of filter weight is larger than the local chunks of signal"
123             )
124         # fetch halos and store them in a.halo_next/a.halo_prev
125         a.get_halo(halo_size)
126         # apply halos to local array
127         signal = a.array_with_halos
128     else:
129         signal = a.larray
130 
131     # flip filter for convolution as Pytorch conv1d computes correlations
132     v = flip(v, [0])
133     if v.larray.shape != v.lshape_map[0]:
134         # pads weights if input kernel is uneven
135         target = torch.zeros(v.lshape_map[0][0], dtype=v.larray.dtype, device=v.larray.device)
136         pad_size = v.lshape_map[0][0] - v.larray.shape[0]
137         target[pad_size:] = v.larray
138         weight = target
139     else:
140         weight = v.larray
141 
142     t_v = weight  # stores temporary weight
143 
144     # make signal and filter weight 3D for Pytorch conv1d function
145     signal = signal.reshape(1, 1, signal.shape[0])
146     weight = weight.reshape(1, 1, weight.shape[0])
147 
148     # cast to float if on GPU
149     if signal.is_cuda:
150         float_type = promote_types(signal.dtype, torch.float32).torch_type()
151         signal = signal.to(float_type)
152         weight = weight.to(float_type)
153         t_v = t_v.to(float_type)
154 
155     if v.is_distributed():
156         size = v.comm.size
157 
158         for r in range(size):
159             rec_v = v.comm.bcast(t_v, root=r)
160             t_v1 = rec_v.reshape(1, 1, rec_v.shape[0])
161             local_signal_filtered = fc.conv1d(signal, t_v1)
162             # unpack 3D result into 1D
163             local_signal_filtered = local_signal_filtered[0, 0, :]
164 
165             if a.comm.rank != 0 and v.lshape_map[0][0] % 2 == 0:
166                 local_signal_filtered = local_signal_filtered[1:]
167 
168             # accumulate filtered signal on the fly
169             global_signal_filtered = array(
170                 local_signal_filtered, is_split=0, device=a.device, comm=a.comm
171             )
172             if r == 0:
173                 # initialize signal_filtered, starting point of slice
174                 signal_filtered = zeros(
175                     gshape, dtype=a.dtype, split=a.split, device=a.device, comm=a.comm
176                 )
177                 start_idx = 0
178 
179             # accumulate relevant slice of filtered signal
180             # note, this is a binary operation between unevenly distributed dndarrays and will require communication, check out _operations.__binary_op()
181             signal_filtered += global_signal_filtered[start_idx : start_idx + gshape]
182             if r != size - 1:
183                 start_idx += v.lshape_map[r + 1][0].item()
184         return signal_filtered
185 
186     else:
187         # apply torch convolution operator
188         signal_filtered = fc.conv1d(signal, weight)
189 
190         # unpack 3D result into 1D
191         signal_filtered = signal_filtered[0, 0, :]
192 
193         # if kernel shape along split axis is even we need to get rid of duplicated values
194         if a.comm.rank != 0 and v.shape[0] % 2 == 0:
195             signal_filtered = signal_filtered[1:]
196 
197         return DNDarray(
198             signal_filtered.contiguous(),
199             (gshape,),
200             signal_filtered.dtype,
201             a.split,
202             a.device,
203             a.comm,
204             balanced=False,
205         ).astype(a.dtype.torch_type())
206
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/heat/core/signal.py b/heat/core/signal.py
--- a/heat/core/signal.py
+++ b/heat/core/signal.py
@@ -156,7 +156,8 @@
         size = v.comm.size
 
         for r in range(size):
-            rec_v = v.comm.bcast(t_v, root=r)
+            rec_v = t_v.clone()
+            v.comm.Bcast(rec_v, root=r)
             t_v1 = rec_v.reshape(1, 1, rec_v.shape[0])
             local_signal_filtered = fc.conv1d(signal, t_v1)
             # unpack 3D result into 1D

{"golden_diff": "diff --git a/heat/core/signal.py b/heat/core/signal.py\n--- a/heat/core/signal.py\n+++ b/heat/core/signal.py\n@@ -156,7 +156,8 @@\n         size = v.comm.size\n \n         for r in range(size):\n-            rec_v = v.comm.bcast(t_v, root=r)\n+            rec_v = t_v.clone()\n+            v.comm.Bcast(rec_v, root=r)\n             t_v1 = rec_v.reshape(1, 1, rec_v.shape[0])\n             local_signal_filtered = fc.conv1d(signal, t_v1)\n             # unpack 3D result into 1D\n", "issue": "[Bug]: convolve with distributed kernel on multiple GPUs\n### What happened?\r\n\r\nconvolve does not work if the kernel is distributed when more than one GPU is available.\r\n\r\n### Code snippet triggering the error\r\n\r\n```python\r\nimport heat as ht\r\n\r\ndis_signal = ht.arange(0, 16, split=0, device='gpu', dtype=ht.int)\r\ndis_kernel_odd = ht.ones(3, split=0, dtype=ht.int, device='gpu')\r\nconv = ht.convolve(dis_signal, dis_kernel_odd, mode='full')\r\n```\r\n\r\n\r\n### Error message or erroneous outcome\r\n\r\n```shell\r\n$ CUDA_VISIBLE_DEVICES=0,1,2,3 srun --ntasks=2 -l python test.py \r\n1:Traceback (most recent call last):\r\n1:   File \".../test.py\", line 7, in <module>\r\n1:     conv = ht.convolve(dis_signal, dis_kernel_odd, mode='full')\r\n1:   File \".../heat-venv_2023/lib/python3.10/site-packages/heat/core/signal.py\", line 161, in convolve\r\n1:     local_signal_filtered = fc.conv1d(signal, t_v1)\r\n1: RuntimeError: Expected all tensors to be on the same device, but found at least two devices, cuda:1 and cuda:0! (when checking argument for argument weight in method wrapper__cudnn_convolution)\r\n```\r\n\r\n\r\n### Version\r\n\r\nmain (development branch)\r\n\r\n### Python version\r\n\r\n3.10\r\n\r\n### PyTorch version\r\n\r\n1.12\r\n\r\n### MPI version\r\n\r\n```shell\r\nOpenMPI 4.1.4\r\n```\r\n\n", "before_files": [{"content": "\"\"\"Provides a collection of signal-processing operations\"\"\"\n\nimport torch\nimport numpy as np\n\nfrom .communication import MPI\nfrom .dndarray import DNDarray\nfrom .types import promote_types\nfrom .manipulations import pad, flip\nfrom .factories import array, zeros\nimport torch.nn.functional as fc\n\n__all__ = [\"convolve\"]\n\n\ndef convolve(a: DNDarray, v: DNDarray, mode: str = \"full\") -> DNDarray:\n    \"\"\"\n    Returns the discrete, linear convolution of two one-dimensional `DNDarray`s or scalars.\n\n    Parameters\n    ----------\n    a : DNDarray or scalar\n        One-dimensional signal `DNDarray` of shape (N,) or scalar.\n    v : DNDarray or scalar\n        One-dimensional filter weight `DNDarray` of shape (M,) or scalar.\n    mode : str\n        Can be 'full', 'valid', or 'same'. Default is 'full'.\n        'full':\n          Returns the convolution at\n          each point of overlap, with an output shape of (N+M-1,). At\n          the end-points of the convolution, the signals do not overlap\n          completely, and boundary effects may be seen.\n        'same':\n          Mode 'same' returns output  of length 'N'. Boundary\n          effects are still visible. This mode is not supported for\n          even-sized filter weights\n        'valid':\n          Mode 'valid' returns output of length 'N-M+1'. The\n          convolution product is only given for points where the signals\n          overlap completely. Values outside the signal boundary have no\n          effect.\n\n    Examples\n    --------\n    Note how the convolution operator flips the second array\n    before \"sliding\" the two across one another:\n\n    >>> a = ht.ones(10)\n    >>> v = ht.arange(3).astype(ht.float)\n    >>> ht.convolve(a, v, mode='full')\n    DNDarray([0., 1., 3., 3., 3., 3., 2.])\n    >>> ht.convolve(a, v, mode='same')\n    DNDarray([1., 3., 3., 3., 3.])\n    >>> ht.convolve(a, v, mode='valid')\n    DNDarray([3., 3., 3.])\n    >>> a = ht.ones(10, split = 0)\n    >>> v = ht.arange(3, split = 0).astype(ht.float)\n    >>> ht.convolve(a, v, mode='valid')\n    DNDarray([3., 3., 3., 3., 3., 3., 3., 3.])\n\n    [0/3] DNDarray([3., 3., 3.])\n    [1/3] DNDarray([3., 3., 3.])\n    [2/3] DNDarray([3., 3.])\n    >>> a = ht.ones(10, split = 0)\n    >>> v = ht.arange(3, split = 0)\n    >>> ht.convolve(a, v)\n    DNDarray([0., 1., 3., 3., 3., 3., 3., 3., 3., 3., 3., 2.], dtype=ht.float32, device=cpu:0, split=0)\n\n    [0/3] DNDarray([0., 1., 3., 3.])\n    [1/3] DNDarray([3., 3., 3., 3.])\n    [2/3] DNDarray([3., 3., 3., 2.])\n    \"\"\"\n    if np.isscalar(a):\n        a = array([a])\n    if np.isscalar(v):\n        v = array([v])\n    if not isinstance(a, DNDarray):\n        try:\n            a = array(a)\n        except TypeError:\n            raise TypeError(\"non-supported type for signal: {}\".format(type(a)))\n    if not isinstance(v, DNDarray):\n        try:\n            v = array(v)\n        except TypeError:\n            raise TypeError(\"non-supported type for filter: {}\".format(type(v)))\n    promoted_type = promote_types(a.dtype, v.dtype)\n    a = a.astype(promoted_type)\n    v = v.astype(promoted_type)\n\n    if len(a.shape) != 1 or len(v.shape) != 1:\n        raise ValueError(\"Only 1-dimensional input DNDarrays are allowed\")\n    if mode == \"same\" and v.shape[0] % 2 == 0:\n        raise ValueError(\"Mode 'same' cannot be used with even-sized kernel\")\n    if not v.is_balanced():\n        raise ValueError(\"Only balanced kernel weights are allowed\")\n\n    if v.shape[0] > a.shape[0]:\n        a, v = v, a\n\n    # compute halo size\n    halo_size = torch.max(v.lshape_map[:, 0]).item() // 2\n\n    # pad DNDarray with zeros according to mode\n    if mode == \"full\":\n        pad_size = v.shape[0] - 1\n        gshape = v.shape[0] + a.shape[0] - 1\n    elif mode == \"same\":\n        pad_size = v.shape[0] // 2\n        gshape = a.shape[0]\n    elif mode == \"valid\":\n        pad_size = 0\n        gshape = a.shape[0] - v.shape[0] + 1\n    else:\n        raise ValueError(\"Supported modes are 'full', 'valid', 'same', got {}\".format(mode))\n\n    a = pad(a, pad_size, \"constant\", 0)\n\n    if a.is_distributed():\n        if (v.lshape_map[:, 0] > a.lshape_map[:, 0]).any():\n            raise ValueError(\n                \"Local chunk of filter weight is larger than the local chunks of signal\"\n            )\n        # fetch halos and store them in a.halo_next/a.halo_prev\n        a.get_halo(halo_size)\n        # apply halos to local array\n        signal = a.array_with_halos\n    else:\n        signal = a.larray\n\n    # flip filter for convolution as Pytorch conv1d computes correlations\n    v = flip(v, [0])\n    if v.larray.shape != v.lshape_map[0]:\n        # pads weights if input kernel is uneven\n        target = torch.zeros(v.lshape_map[0][0], dtype=v.larray.dtype, device=v.larray.device)\n        pad_size = v.lshape_map[0][0] - v.larray.shape[0]\n        target[pad_size:] = v.larray\n        weight = target\n    else:\n        weight = v.larray\n\n    t_v = weight  # stores temporary weight\n\n    # make signal and filter weight 3D for Pytorch conv1d function\n    signal = signal.reshape(1, 1, signal.shape[0])\n    weight = weight.reshape(1, 1, weight.shape[0])\n\n    # cast to float if on GPU\n    if signal.is_cuda:\n        float_type = promote_types(signal.dtype, torch.float32).torch_type()\n        signal = signal.to(float_type)\n        weight = weight.to(float_type)\n        t_v = t_v.to(float_type)\n\n    if v.is_distributed():\n        size = v.comm.size\n\n        for r in range(size):\n            rec_v = v.comm.bcast(t_v, root=r)\n            t_v1 = rec_v.reshape(1, 1, rec_v.shape[0])\n            local_signal_filtered = fc.conv1d(signal, t_v1)\n            # unpack 3D result into 1D\n            local_signal_filtered = local_signal_filtered[0, 0, :]\n\n            if a.comm.rank != 0 and v.lshape_map[0][0] % 2 == 0:\n                local_signal_filtered = local_signal_filtered[1:]\n\n            # accumulate filtered signal on the fly\n            global_signal_filtered = array(\n                local_signal_filtered, is_split=0, device=a.device, comm=a.comm\n            )\n            if r == 0:\n                # initialize signal_filtered, starting point of slice\n                signal_filtered = zeros(\n                    gshape, dtype=a.dtype, split=a.split, device=a.device, comm=a.comm\n                )\n                start_idx = 0\n\n            # accumulate relevant slice of filtered signal\n            # note, this is a binary operation between unevenly distributed dndarrays and will require communication, check out _operations.__binary_op()\n            signal_filtered += global_signal_filtered[start_idx : start_idx + gshape]\n            if r != size - 1:\n                start_idx += v.lshape_map[r + 1][0].item()\n        return signal_filtered\n\n    else:\n        # apply torch convolution operator\n        signal_filtered = fc.conv1d(signal, weight)\n\n        # unpack 3D result into 1D\n        signal_filtered = signal_filtered[0, 0, :]\n\n        # if kernel shape along split axis is even we need to get rid of duplicated values\n        if a.comm.rank != 0 and v.shape[0] % 2 == 0:\n            signal_filtered = signal_filtered[1:]\n\n        return DNDarray(\n            signal_filtered.contiguous(),\n            (gshape,),\n            signal_filtered.dtype,\n            a.split,\n            a.device,\n            a.comm,\n            balanced=False,\n        ).astype(a.dtype.torch_type())\n", "path": "heat/core/signal.py"}], "after_files": [{"content": "\"\"\"Provides a collection of signal-processing operations\"\"\"\n\nimport torch\nimport numpy as np\n\nfrom .communication import MPI\nfrom .dndarray import DNDarray\nfrom .types import promote_types\nfrom .manipulations import pad, flip\nfrom .factories import array, zeros\nimport torch.nn.functional as fc\n\n__all__ = [\"convolve\"]\n\n\ndef convolve(a: DNDarray, v: DNDarray, mode: str = \"full\") -> DNDarray:\n    \"\"\"\n    Returns the discrete, linear convolution of two one-dimensional `DNDarray`s or scalars.\n\n    Parameters\n    ----------\n    a : DNDarray or scalar\n        One-dimensional signal `DNDarray` of shape (N,) or scalar.\n    v : DNDarray or scalar\n        One-dimensional filter weight `DNDarray` of shape (M,) or scalar.\n    mode : str\n        Can be 'full', 'valid', or 'same'. Default is 'full'.\n        'full':\n          Returns the convolution at\n          each point of overlap, with an output shape of (N+M-1,). At\n          the end-points of the convolution, the signals do not overlap\n          completely, and boundary effects may be seen.\n        'same':\n          Mode 'same' returns output  of length 'N'. Boundary\n          effects are still visible. This mode is not supported for\n          even-sized filter weights\n        'valid':\n          Mode 'valid' returns output of length 'N-M+1'. The\n          convolution product is only given for points where the signals\n          overlap completely. Values outside the signal boundary have no\n          effect.\n\n    Examples\n    --------\n    Note how the convolution operator flips the second array\n    before \"sliding\" the two across one another:\n\n    >>> a = ht.ones(10)\n    >>> v = ht.arange(3).astype(ht.float)\n    >>> ht.convolve(a, v, mode='full')\n    DNDarray([0., 1., 3., 3., 3., 3., 2.])\n    >>> ht.convolve(a, v, mode='same')\n    DNDarray([1., 3., 3., 3., 3.])\n    >>> ht.convolve(a, v, mode='valid')\n    DNDarray([3., 3., 3.])\n    >>> a = ht.ones(10, split = 0)\n    >>> v = ht.arange(3, split = 0).astype(ht.float)\n    >>> ht.convolve(a, v, mode='valid')\n    DNDarray([3., 3., 3., 3., 3., 3., 3., 3.])\n\n    [0/3] DNDarray([3., 3., 3.])\n    [1/3] DNDarray([3., 3., 3.])\n    [2/3] DNDarray([3., 3.])\n    >>> a = ht.ones(10, split = 0)\n    >>> v = ht.arange(3, split = 0)\n    >>> ht.convolve(a, v)\n    DNDarray([0., 1., 3., 3., 3., 3., 3., 3., 3., 3., 3., 2.], dtype=ht.float32, device=cpu:0, split=0)\n\n    [0/3] DNDarray([0., 1., 3., 3.])\n    [1/3] DNDarray([3., 3., 3., 3.])\n    [2/3] DNDarray([3., 3., 3., 2.])\n    \"\"\"\n    if np.isscalar(a):\n        a = array([a])\n    if np.isscalar(v):\n        v = array([v])\n    if not isinstance(a, DNDarray):\n        try:\n            a = array(a)\n        except TypeError:\n            raise TypeError(\"non-supported type for signal: {}\".format(type(a)))\n    if not isinstance(v, DNDarray):\n        try:\n            v = array(v)\n        except TypeError:\n            raise TypeError(\"non-supported type for filter: {}\".format(type(v)))\n    promoted_type = promote_types(a.dtype, v.dtype)\n    a = a.astype(promoted_type)\n    v = v.astype(promoted_type)\n\n    if len(a.shape) != 1 or len(v.shape) != 1:\n        raise ValueError(\"Only 1-dimensional input DNDarrays are allowed\")\n    if mode == \"same\" and v.shape[0] % 2 == 0:\n        raise ValueError(\"Mode 'same' cannot be used with even-sized kernel\")\n    if not v.is_balanced():\n        raise ValueError(\"Only balanced kernel weights are allowed\")\n\n    if v.shape[0] > a.shape[0]:\n        a, v = v, a\n\n    # compute halo size\n    halo_size = torch.max(v.lshape_map[:, 0]).item() // 2\n\n    # pad DNDarray with zeros according to mode\n    if mode == \"full\":\n        pad_size = v.shape[0] - 1\n        gshape = v.shape[0] + a.shape[0] - 1\n    elif mode == \"same\":\n        pad_size = v.shape[0] // 2\n        gshape = a.shape[0]\n    elif mode == \"valid\":\n        pad_size = 0\n        gshape = a.shape[0] - v.shape[0] + 1\n    else:\n        raise ValueError(\"Supported modes are 'full', 'valid', 'same', got {}\".format(mode))\n\n    a = pad(a, pad_size, \"constant\", 0)\n\n    if a.is_distributed():\n        if (v.lshape_map[:, 0] > a.lshape_map[:, 0]).any():\n            raise ValueError(\n                \"Local chunk of filter weight is larger than the local chunks of signal\"\n            )\n        # fetch halos and store them in a.halo_next/a.halo_prev\n        a.get_halo(halo_size)\n        # apply halos to local array\n        signal = a.array_with_halos\n    else:\n        signal = a.larray\n\n    # flip filter for convolution as Pytorch conv1d computes correlations\n    v = flip(v, [0])\n    if v.larray.shape != v.lshape_map[0]:\n        # pads weights if input kernel is uneven\n        target = torch.zeros(v.lshape_map[0][0], dtype=v.larray.dtype, device=v.larray.device)\n        pad_size = v.lshape_map[0][0] - v.larray.shape[0]\n        target[pad_size:] = v.larray\n        weight = target\n    else:\n        weight = v.larray\n\n    t_v = weight  # stores temporary weight\n\n    # make signal and filter weight 3D for Pytorch conv1d function\n    signal = signal.reshape(1, 1, signal.shape[0])\n    weight = weight.reshape(1, 1, weight.shape[0])\n\n    # cast to float if on GPU\n    if signal.is_cuda:\n        float_type = promote_types(signal.dtype, torch.float32).torch_type()\n        signal = signal.to(float_type)\n        weight = weight.to(float_type)\n        t_v = t_v.to(float_type)\n\n    if v.is_distributed():\n        size = v.comm.size\n\n        for r in range(size):\n            rec_v = t_v.clone()\n            v.comm.Bcast(rec_v, root=r)\n            t_v1 = rec_v.reshape(1, 1, rec_v.shape[0])\n            local_signal_filtered = fc.conv1d(signal, t_v1)\n            # unpack 3D result into 1D\n            local_signal_filtered = local_signal_filtered[0, 0, :]\n\n            if a.comm.rank != 0 and v.lshape_map[0][0] % 2 == 0:\n                local_signal_filtered = local_signal_filtered[1:]\n\n            # accumulate filtered signal on the fly\n            global_signal_filtered = array(\n                local_signal_filtered, is_split=0, device=a.device, comm=a.comm\n            )\n            if r == 0:\n                # initialize signal_filtered, starting point of slice\n                signal_filtered = zeros(\n                    gshape, dtype=a.dtype, split=a.split, device=a.device, comm=a.comm\n                )\n                start_idx = 0\n\n            # accumulate relevant slice of filtered signal\n            # note, this is a binary operation between unevenly distributed dndarrays and will require communication, check out _operations.__binary_op()\n            signal_filtered += global_signal_filtered[start_idx : start_idx + gshape]\n            if r != size - 1:\n                start_idx += v.lshape_map[r + 1][0].item()\n        return signal_filtered\n\n    else:\n        # apply torch convolution operator\n        signal_filtered = fc.conv1d(signal, weight)\n\n        # unpack 3D result into 1D\n        signal_filtered = signal_filtered[0, 0, :]\n\n        # if kernel shape along split axis is even we need to get rid of duplicated values\n        if a.comm.rank != 0 and v.shape[0] % 2 == 0:\n            signal_filtered = signal_filtered[1:]\n\n        return DNDarray(\n            signal_filtered.contiguous(),\n            (gshape,),\n            signal_filtered.dtype,\n            a.split,\n            a.device,\n            a.comm,\n            balanced=False,\n        ).astype(a.dtype.torch_type())\n", "path": "heat/core/signal.py"}]}