gh-71448: Support extension in BytesIO.truncate

Doc/library/io.rst

@@ -786,6 +786,13 @@ than raw I/O does.
 
       .. versionadded:: 3.5
 
+   .. method:: truncate(size=None, /)
+
+      In :class:`BytesIO`, this is the same as :meth:`IOBase.truncate`.
+
+      .. versionchanged:: next
+         Now extends the underlying buffer as :meth:`IOBase.truncate` documents.
+
 .. class:: BufferedReader(raw, buffer_size=DEFAULT_BUFFER_SIZE)
 
    A buffered binary stream providing higher-level access to a readable, non

Lib/_pyio.py

@@ -1033,7 +1033,7 @@ def truncate(self, pos=None):
                     pos = pos_index()
                 if pos < 0:
                     raise ValueError("negative truncate position %r" % (pos,))
-            del self._buffer[pos:]
+            self._buffer.resize(pos)
         return pos
 
     def readable(self):

Lib/test/test_io/test_memoryio.py

@@ -584,6 +584,41 @@ def test_relative_seek(self):
         memio.seek(1, 1)
         self.assertEqual(memio.read(), buf[1:])
 
+    def test_truncate_extend(self):
+        # gh-71448: Extending zero-fills, within the allocation or past it,
+        # and never writes into a buffer shared with the caller.
+        buf = self.buftype("abcdef")
+        private = self.ioclass()
+        private.write(buf)
+        for memio in self.ioclass(buf), private:
+            memio.seek(7)
+            self.assertEqual(memio.truncate(3), 3)
+            self.assertEqual(memio.truncate(5), 5)
+            self.assertEqual(memio.getvalue(), b"abc\0\0")
+            self.assertEqual(memio.truncate(), 7)
+            self.assertEqual(memio.getvalue(), b"abc\0\0\0\0")
+            self.assertEqual(memio.tell(), 7)
+        self.assertEqual(buf, b"abcdef")
+
+    def test_truncate_extend_too_large(self):
+        for size in sys.maxsize, sys.maxsize - 1:
+            private = self.ioclass()
+            private.write(b"abc")
+            for memio in self.ioclass(b"abc"), private:
+                with self.subTest(size=size, shared=memio is not private):
+                    self.assertRaises((MemoryError, OverflowError),
+                                      memio.truncate, size)
+                    self.assertEqual(memio.getvalue(), b"abc")
+
+    @support.bigmemtest(size=support._2G, memuse=1, dry_run=False)
+    def test_truncate_extend_bigmem(self, size):
+        # Sizes above 2**31 - 1 must work where C long is 32-bit.
+        memio = self.ioclass(b"abc")
+        self.assertEqual(memio.truncate(size), size)
+        with memio.getbuffer() as buf:
+            self.assertEqual(len(buf), size)
+            self.assertEqual(buf[:4], b"abc\0")
+
     def test_issue141311(self):
         memio = self.ioclass()
         # Seek allows PY_SSIZE_T_MAX, read should handle that.
@@ -984,12 +1019,6 @@ def __index__(self):
             finally:
                 size.result.release()
 
-        # Sizes above 2**31 - 1 must work where C long is 32-bit.
-        if sys.maxsize > 2**31:
-            stream = self.ioclass(b'abc')
-            self.assertEqual(stream.truncate(2**31), 2**31)
-            self.assertEqual(stream.getvalue(), b'abc')
-
     def test_write_overflow(self):
         # gh-160781: overallocation at a position near sys.maxsize
         # must raise OverflowError, not SystemError.

Misc/NEWS.d/next/Library/2026-06-25-22-43-34.gh-issue-71448.ajvMxg.rst

@@ -0,0 +1,2 @@
+Update :class:`io.BytesIO` implementation of :meth:`io.IOBase.truncate` to
+support extension on truncate to larger size than allocated.

Modules/_io/bytesio.c

@@ -164,8 +164,7 @@ scan_eol_lock_held(bytesio *self, Py_ssize_t len)
 }
 
 /* Internal routine for detaching the shared buffer of BytesIO objects.
-   The caller should ensure that the 'size' argument is non-negative and
-   not lesser than self->string_size.  Returns 0 on success, -1 otherwise. */
+   Data past 'size' is dropped.  Returns 0 on success, -1 otherwise. */
 static int
 unshare_buffer_lock_held(bytesio *self, size_t size)
 {
@@ -174,22 +173,30 @@ unshare_buffer_lock_held(bytesio *self, size_t size)
     PyObject *new_buf;
     assert(SHARED_BUF(self));
     assert(FT_ATOMIC_LOAD_SSIZE_RELAXED(self->exports) == 0);
-    assert(size >= (size_t)self->string_size);
+    assert(size <= PY_SSIZE_T_MAX);
     new_buf = PyBytes_FromStringAndSize(NULL, size);
     if (new_buf == NULL)
         return -1;
     memcpy(PyBytes_AS_STRING(new_buf), PyBytes_AS_STRING(self->buf),
-           self->string_size);
+           Py_MIN(size, (size_t)self->string_size));
     Py_SETREF(self->buf, new_buf);
     clear_shared_buf(self);
     return 0;
 }
 
-/* Internal routine for changing the size of the buffer of BytesIO objects.
-   The caller should ensure that the 'size' argument is non-negative.  Returns
-   0 on success, -1 otherwise. */
+/* Reallocate the buffer of a BytesIO object to exactly 'alloc' bytes, keeping
+   the data and unsharing it if needed.  Returns 0 on success, -1 otherwise. */
 static int
-resize_buffer_lock_held(bytesio *self, size_t size)
+realloc_buffer_lock_held(bytesio *self, size_t alloc)
+{
+    return SHARED_BUF(self) ? unshare_buffer_lock_held(self, alloc)
+                            : resize_unshared_buffer_lock_held(self, alloc);
+}
+
+/* Make the buffer of a BytesIO object private and at least 'size' bytes long,
+   keeping the first string_size bytes.  Returns 0 on success, -1 otherwise. */
+static int
+prepare_write_lock_held(bytesio *self, size_t size)
 {
     _Py_CRITICAL_SECTION_ASSERT_OBJECT_LOCKED(self);
 
@@ -205,13 +212,12 @@ resize_buffer_lock_held(bytesio *self, size_t size)
     if (size > PY_SSIZE_T_MAX)
         goto overflow;
 
-    if (size < alloc / 2) {
-        /* Major downsize; resize down to exact size. */
-        alloc = size + 1;
-    }
-    else if (size < alloc) {
-        /* Within allocated size; quick exit */
-        return 0;
+    if (size <= alloc) {
+        /* Within allocated size; quick exit unless shared */
+        if (!SHARED_BUF(self)) {
+            return 0;
+        }
+        alloc = Py_MAX(size, (size_t)self->string_size);
     }
     else if (size <= alloc * 1.125) {
         /* Moderate upsize; overallocate similar to list_resize() */
@@ -225,16 +231,7 @@ resize_buffer_lock_held(bytesio *self, size_t size)
         goto overflow;
     }
 
-    if (SHARED_BUF(self)) {
-        if (unshare_buffer_lock_held(self, alloc) < 0)
-            return -1;
-    }
-    else {
-        if (resize_unshared_buffer_lock_held(self, alloc) < 0)
-            return -1;
-    }
-
-    return 0;
+    return realloc_buffer_lock_held(self, alloc);
 
   overflow:
     PyErr_SetString(PyExc_OverflowError,
@@ -269,17 +266,9 @@ write_bytes_lock_held(bytesio *self, PyObject *b)
 
     assert(self->pos >= 0);
     size_t endpos = (size_t)self->pos + len;
-    if (endpos > (size_t)PyBytes_GET_SIZE(self->buf)) {
-        if (resize_buffer_lock_held(self, endpos) < 0) {
-            len = -1;
-            goto done;
-        }
-    }
-    else if (SHARED_BUF(self)) {
-        if (unshare_buffer_lock_held(self, Py_MAX(endpos, (size_t)self->string_size)) < 0) {
-            len = -1;
-            goto done;
-        }
+    if (prepare_write_lock_held(self, endpos) < 0) {
+        len = -1;
+        goto done;
     }
     assert(_PyBytes_IsMutable(self->buf));
 
@@ -433,13 +422,8 @@ _io_BytesIO_getvalue_impl(bytesio *self)
                                          self->string_size);
 
     if (self->string_size != PyBytes_GET_SIZE(self->buf)) {
-        if (SHARED_BUF(self)) {
-            if (unshare_buffer_lock_held(self, self->string_size) < 0)
-                return NULL;
-        }
-        else {
-            if (resize_unshared_buffer_lock_held(self, self->string_size) < 0)
-                return NULL;
+        if (realloc_buffer_lock_held(self, self->string_size) < 0) {
+            return NULL;
         }
     }
     set_shared_buf(self);
@@ -772,13 +756,22 @@ _io_BytesIO_truncate_impl(bytesio *self, PyObject *size)
     }
 
     if (new_size < self->string_size) {
-        Py_ssize_t old_string_size = self->string_size;
-        self->string_size = new_size;
-        if (resize_buffer_lock_held(self, new_size) < 0) {
-            self->string_size = old_string_size;
+        if (new_size < PyBytes_GET_SIZE(self->buf) / 2) {
+            /* Major downsize; resize down to exact size. */
+            if (realloc_buffer_lock_held(self, new_size + 1) < 0) {
+                return NULL;
+            }
+        }
+    }
+    else if (new_size > self->string_size) {
+        if (prepare_write_lock_held(self, new_size) < 0) {
             return NULL;
         }
+        /* Fill new space with zeros */
+        memset(PyBytes_AS_STRING(self->buf) + self->string_size, '\0',
+               new_size - self->string_size);
     }
+    self->string_size = new_size;
 
     return PyLong_FromSsize_t(new_size);
 }
@@ -1284,9 +1277,10 @@ bytesiobuf_getbuffer_lock_held(PyObject *op, Py_buffer *view, int flags)
     if (check_closed(b)) {
         return -1;
     }
-    if (FT_ATOMIC_LOAD_SSIZE_RELAXED(b->exports) == 0 && SHARED_BUF(b)) {
-        if (unshare_buffer_lock_held(b, b->string_size) < 0)
+    if (FT_ATOMIC_LOAD_SSIZE_RELAXED(b->exports) == 0) {
+        if (prepare_write_lock_held(b, b->string_size) < 0) {
             return -1;
+        }
     }
 
     /* cannot fail if view != NULL and readonly == 0 */