Index send_all memoryviews by byte count, not element count

src/trio/_highlevel_socket.py

@@ -117,6 +117,11 @@ async def send_all(self, data: bytes | bytearray | memoryview) -> None:
                             raise trio.ClosedResourceError("socket was already closed")
                         await trio.lowlevel.checkpoint()
                         return
+                    # send() returns a byte count; index memoryviews in bytes too.
+                    # Contiguous multi-byte / multi-dim views must be cast first
+                    # (#3543). Leave empty views alone (cast can fail on strided).
+                    if data.contiguous:
+                        data = data.cast("B")
                     total_sent = 0
                     while total_sent < len(data):
                         with data[total_sent:] as remaining:

src/trio/_tests/test_unix_pipes.py

@@ -286,3 +286,32 @@ async def test_bizarro_OSError_from_receive() -> None:
 @skip_if_fbsd_pipes_broken
 async def test_pipe_fully() -> None:
     await check_one_way_stream(make_pipe, make_clogged_pipe)
+
+
+async def test_send_all_memoryview_itemsize_gt_1() -> None:
+    """Partial writes must advance by byte count, not element count (#3543)."""
+    import os
+
+    import trio
+
+    payload = bytes(range(256)) * 64  # 16 KiB
+    read_fd, write_fd = os.pipe()
+    received = bytearray()
+    async with (
+        trio.lowlevel.FdStream(write_fd) as sender,
+        trio.lowlevel.FdStream(
+            read_fd,
+        ) as receiver,
+    ):
+
+        async def send() -> None:
+            with memoryview(payload).cast("H") as data:
+                await sender.send_all(data)
+            await sender.aclose()
+
+        async with trio.open_nursery() as nursery:
+            nursery.start_soon(send)
+            async for chunk in receiver:
+                received.extend(chunk)
+
+    assert received == payload

src/trio/_unix_pipes.py

@@ -134,9 +134,14 @@ async def send_all(self, data: bytes | bytearray | memoryview) -> None:
             if self._fd_holder.closed:
                 raise trio.ClosedResourceError("file was already closed")
             await trio.lowlevel.checkpoint()
-            length = len(data)
             # adapted from the SocketStream code
             with memoryview(data) as view:
+                # os.write returns a byte count; index memoryviews in bytes too.
+                # Contiguous multi-byte / multi-dim views must be cast first
+                # (#3543). Leave empty views alone (cast can fail on strided).
+                if view.contiguous and view:
+                    view = view.cast("B")
+                length = len(view)
                 sent = 0
                 while sent < length:
                     with view[sent:] as remaining: