// spell-checker:disable

//! mmap module

pub(crate) use mmap::module_def;

#[pymodule]

mod mmap {

use crate::common::{

borrow::{BorrowedValue, BorrowedValueMut},

lock::{MapImmutable, PyMutex, PyMutexGuard},

};

use crate::vm::{

AsObject, FromArgs, Py, PyObject, PyObjectRef, PyPayload, PyRef, PyResult,

TryFromBorrowedObject, VirtualMachine, atomic_func,

builtins::{PyBytes, PyBytesRef, PyInt, PyIntRef, PyType, PyTypeRef},

byte::{bytes_from_object, value_from_object},

convert::ToPyException,

function::{ArgBytesLike, FuncArgs, OptionalArg},

protocol::{

BufferDescriptor, BufferMethods, PyBuffer, PyMappingMethods, PySequenceMethods,

},

sliceable::{SaturatedSlice, SequenceIndex, SequenceIndexOp},

types::{AsBuffer, AsMapping, AsSequence, Constructor, Representable},

};

use core::ops::{Deref, DerefMut};

use crossbeam_utils::atomic::AtomicCell;

use memmap2::{Mmap, MmapMut, MmapOptions};

use num_traits::Signed;

use std::io::{self, Write};

#[cfg(unix)]

use nix::{sys::stat::fstat, unistd};

#[cfg(unix)]

use rustpython_common::crt_fd;

#[cfg(windows)]

use rustpython_common::suppress_iph;

#[cfg(windows)]

use std::os::windows::io::{AsRawHandle, FromRawHandle, OwnedHandle, RawHandle};

#[cfg(windows)]

use windows_sys::Win32::{

Foundation::{

CloseHandle, DUPLICATE_SAME_ACCESS, DuplicateHandle, HANDLE, INVALID_HANDLE_VALUE,

},

Storage::FileSystem::{FILE_BEGIN, GetFileSize, SetEndOfFile, SetFilePointerEx},

System::Memory::{

CreateFileMappingW, FILE_MAP_COPY, FILE_MAP_READ, FILE_MAP_WRITE, FlushViewOfFile,

MapViewOfFile, PAGE_READONLY, PAGE_READWRITE, PAGE_WRITECOPY, UnmapViewOfFile,

},

System::Threading::GetCurrentProcess,

};

#[cfg(unix)]

fn validate_advice(vm: &VirtualMachine, advice: i32) -> PyResult<i32> {

match advice {

libc::MADV_NORMAL

| libc::MADV_RANDOM

| libc::MADV_SEQUENTIAL

| libc::MADV_WILLNEED

| libc::MADV_DONTNEED => Ok(advice),

#[cfg(any(

target_os = "linux",

target_os = "macos",

target_os = "ios",

target_os = "freebsd"

))]

libc::MADV_FREE => Ok(advice),

#[cfg(target_os = "linux")]

libc::MADV_DONTFORK

| libc::MADV_DOFORK

| libc::MADV_MERGEABLE

| libc::MADV_UNMERGEABLE

| libc::MADV_HUGEPAGE

| libc::MADV_NOHUGEPAGE

| libc::MADV_REMOVE

| libc::MADV_DONTDUMP

| libc::MADV_DODUMP

| libc::MADV_HWPOISON => Ok(advice),

#[cfg(target_os = "freebsd")]

libc::MADV_NOSYNC

| libc::MADV_AUTOSYNC

| libc::MADV_NOCORE

| libc::MADV_CORE

| libc::MADV_PROTECT => Ok(advice),

_ => Err(vm.new_value_error("Not a valid Advice value")),

}

}

#[repr(C)]

#[derive(PartialEq, Eq, Debug)]

enum AccessMode {

Default = 0,

Read = 1,

Write = 2,

Copy = 3,

}

impl<'a> TryFromBorrowedObject<'a> for AccessMode {

fn try_from_borrowed_object(vm: &VirtualMachine, obj: &'a PyObject) -> PyResult<Self> {

let i = u32::try_from_borrowed_object(vm, obj)?;

Ok(match i {

0 => Self::Default,

1 => Self::Read,

2 => Self::Write,

3 => Self::Copy,

_ => return Err(vm.new_value_error("Not a valid AccessMode value")),

})

}

}

#[cfg(unix)]

#[pyattr]

use libc::{

MADV_DONTNEED, MADV_NORMAL, MADV_RANDOM, MADV_SEQUENTIAL, MADV_WILLNEED, MAP_ANON,

MAP_ANONYMOUS, MAP_PRIVATE, MAP_SHARED, PROT_EXEC, PROT_READ, PROT_WRITE,

};

#[cfg(target_os = "macos")]

#[pyattr]

use libc::{MADV_FREE_REUSABLE, MADV_FREE_REUSE};

#[cfg(any(

target_os = "android",

target_os = "dragonfly",

target_os = "fuchsia",

target_os = "freebsd",

target_os = "linux",

target_os = "netbsd",

target_os = "openbsd",

target_vendor = "apple"

))]

#[pyattr]

use libc::MADV_FREE;

#[cfg(target_os = "linux")]

#[pyattr]

use libc::{

MADV_DODUMP, MADV_DOFORK, MADV_DONTDUMP, MADV_DONTFORK, MADV_HUGEPAGE, MADV_HWPOISON,

MADV_MERGEABLE, MADV_NOHUGEPAGE, MADV_REMOVE, MADV_UNMERGEABLE,

};

#[cfg(any(

target_os = "android",

all(

target_os = "linux",

any(

target_arch = "aarch64",

target_arch = "arm",

target_arch = "powerpc",

target_arch = "powerpc64",

target_arch = "s390x",

target_arch = "x86",

target_arch = "x86_64",

target_arch = "sparc64"

)

)

))]

#[pyattr]

use libc::MADV_SOFT_OFFLINE;

#[cfg(all(target_os = "linux", target_arch = "x86_64", target_env = "gnu"))]

#[pyattr]

use libc::{MAP_DENYWRITE, MAP_EXECUTABLE, MAP_POPULATE};

// MAP_STACK is available on Linux, OpenBSD, and NetBSD

#[cfg(any(target_os = "linux", target_os = "openbsd", target_os = "netbsd"))]

#[pyattr]

use libc::MAP_STACK;

// FreeBSD-specific MADV constants

#[cfg(target_os = "freebsd")]

#[pyattr]

use libc::{MADV_AUTOSYNC, MADV_CORE, MADV_NOCORE, MADV_NOSYNC, MADV_PROTECT};

#[pyattr]

const ACCESS_DEFAULT: u32 = AccessMode::Default as u32;

#[pyattr]

const ACCESS_READ: u32 = AccessMode::Read as u32;

#[pyattr]

const ACCESS_WRITE: u32 = AccessMode::Write as u32;

#[pyattr]

const ACCESS_COPY: u32 = AccessMode::Copy as u32;

#[cfg(not(target_arch = "wasm32"))]

#[pyattr(name = "PAGESIZE", once)]

fn page_size(_vm: &VirtualMachine) -> usize {

page_size::get()

}

#[cfg(not(target_arch = "wasm32"))]

#[pyattr(name = "ALLOCATIONGRANULARITY", once)]

fn granularity(_vm: &VirtualMachine) -> usize {

page_size::get_granularity()

}

#[pyattr(name = "error", once)]

fn error_type(vm: &VirtualMachine) -> PyTypeRef {

vm.ctx.exceptions.os_error.to_owned()

}

/// Named file mapping on Windows using raw Win32 APIs.

/// Supports tagname parameter for inter-process shared memory.

#[cfg(windows)]

struct NamedMmap {

map_handle: HANDLE,

view_ptr: *mut u8,

len: usize,

}

#[cfg(windows)]

// SAFETY: The memory mapping is managed by the OS and is safe to share

// across threads. Access is synchronized by PyMutex in PyMmap.

unsafe impl Send for NamedMmap {}

#[cfg(windows)]

unsafe impl Sync for NamedMmap {}

#[cfg(windows)]

impl core::fmt::Debug for NamedMmap {

fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {

f.debug_struct("NamedMmap")

.field("map_handle", &self.map_handle)

.field("view_ptr", &self.view_ptr)

.field("len", &self.len)

.finish()

}

}

#[cfg(windows)]

impl Drop for NamedMmap {

fn drop(&mut self) {

unsafe {

if !self.view_ptr.is_null() {

UnmapViewOfFile(

windows_sys::Win32::System::Memory::MEMORY_MAPPED_VIEW_ADDRESS {

Value: self.view_ptr as *mut _,

},

);

}

if !self.map_handle.is_null() {

CloseHandle(self.map_handle);

}

}

}

}

#[derive(Debug)]

enum MmapObj {

Write(MmapMut),

Read(Mmap),

#[cfg(windows)]

Named(NamedMmap),

}

impl MmapObj {

fn as_slice(&self) -> &[u8] {

match self {

MmapObj::Read(mmap) => &mmap[..],

MmapObj::Write(mmap) => &mmap[..],

#[cfg(windows)]

MmapObj::Named(named) => unsafe {

core::slice::from_raw_parts(named.view_ptr, named.len)

},

}

}

}

#[pyattr]

#[pyclass(name = "mmap")]

#[derive(Debug, PyPayload)]

struct PyMmap {

closed: AtomicCell<bool>,

mmap: PyMutex<Option<MmapObj>>,

#[cfg(unix)]

fd: AtomicCell<i32>,

#[cfg(windows)]

handle: AtomicCell<isize>, // HANDLE is isize on Windows

offset: i64,

size: AtomicCell<usize>,

pos: AtomicCell<usize>, // relative to offset

exports: AtomicCell<usize>,

access: AccessMode,

}

impl PyMmap {

/// Close the underlying file handle/descriptor if open

fn close_handle(&self) {

#[cfg(unix)]

{

let fd = self.fd.swap(-1);

if fd >= 0 {

unsafe { libc::close(fd) };

}

}

#[cfg(windows)]

{

let handle = self.handle.swap(INVALID_HANDLE_VALUE as isize);

if handle != INVALID_HANDLE_VALUE as isize {

unsafe { CloseHandle(handle as HANDLE) };

}

}

}

}

impl Drop for PyMmap {

fn drop(&mut self) {

self.close_handle();

}

}

#[cfg(unix)]

#[derive(FromArgs)]

struct MmapNewArgs {

#[pyarg(any)]

fileno: i32,

#[pyarg(any)]

length: isize,

#[pyarg(any, default = libc::MAP_SHARED)]

flags: libc::c_int,

#[pyarg(any, default = libc::PROT_WRITE | libc::PROT_READ)]

prot: libc::c_int,

#[pyarg(any, default = AccessMode::Default)]

access: AccessMode,

#[pyarg(any, default = 0)]

offset: i64,

}

#[cfg(windows)]

#[derive(FromArgs)]

struct MmapNewArgs {

#[pyarg(any)]

fileno: i32,

#[pyarg(any)]

length: isize,

#[pyarg(any, default)]

tagname: Option<PyObjectRef>,

#[pyarg(any, default = AccessMode::Default)]

access: AccessMode,

#[pyarg(any, default = 0)]

offset: i64,

}

impl MmapNewArgs {

/// Validate mmap constructor arguments

fn validate_new_args(&self, vm: &VirtualMachine) -> PyResult<usize> {

if self.length < 0 {

return Err(vm.new_overflow_error("memory mapped length must be positive"));

}

if self.offset < 0 {

return Err(vm.new_overflow_error("memory mapped offset must be positive"));

}

Ok(self.length as usize)

}

}

#[derive(FromArgs)]

pub struct FlushOptions {

#[pyarg(positional, default)]

offset: Option<isize>,

#[pyarg(positional, default)]

size: Option<isize>,

}

impl FlushOptions {

fn values(self, len: usize) -> Option<(usize, usize)> {

let offset = match self.offset {

Some(o) if o < 0 => return None,

Some(o) => o as usize,

None => 0,

};

let size = match self.size {

Some(s) if s < 0 => return None,

Some(s) => s as usize,

None => len,

};

if len.checked_sub(offset)? < size {

return None;

}

Some((offset, size))

}

}

#[derive(FromArgs, Clone)]

pub struct FindOptions {

#[pyarg(positional)]

sub: Vec<u8>,

#[pyarg(positional, default)]

start: Option<isize>,

#[pyarg(positional, default)]

end: Option<isize>,

}

#[cfg(all(unix, not(target_os = "redox")))]

#[derive(FromArgs)]

pub struct AdviseOptions {

#[pyarg(positional)]

option: libc::c_int,

#[pyarg(positional, default)]

start: Option<PyIntRef>,

#[pyarg(positional, default)]

length: Option<PyIntRef>,

}

#[cfg(all(unix, not(target_os = "redox")))]

impl AdviseOptions {

fn values(self, len: usize, vm: &VirtualMachine) -> PyResult<(libc::c_int, usize, usize)> {

let start = self

.start

.map(|s| {

s.try_to_primitive::<usize>(vm)

.ok()

.filter(|s| *s < len)

.ok_or_else(|| vm.new_value_error("madvise start out of bounds"))

})

.transpose()?

.unwrap_or(0);

let length = self

.length

.map(|s| {

s.try_to_primitive::<usize>(vm)

.map_err(|_| vm.new_value_error("madvise length invalid"))

})

.transpose()?

.unwrap_or(len);

if isize::MAX as usize - start < length {

return Err(vm.new_overflow_error("madvise length too large"));

}

let length = if start + length > len {

len - start

} else {

length

};

Ok((self.option, start, length))

}

}

impl Constructor for PyMmap {

type Args = MmapNewArgs;

#[cfg(unix)]

fn py_new(_cls: &Py<PyType>, args: Self::Args, vm: &VirtualMachine) -> PyResult<Self> {

use libc::{MAP_PRIVATE, MAP_SHARED, PROT_READ, PROT_WRITE};

let mut map_size = args.validate_new_args(vm)?;

let MmapNewArgs {

fileno: fd,

flags,

prot,

access,

offset,

..

} = args;

if (access != AccessMode::Default)

&& ((flags != MAP_SHARED) || (prot != (PROT_WRITE | PROT_READ)))

{

return Err(vm.new_value_error("mmap can't specify both access and flags, prot."));

}

// TODO: memmap2 doesn't support mapping with prot and flags right now

let (_flags, _prot, access) = match access {

AccessMode::Read => (MAP_SHARED, PROT_READ, access),

AccessMode::Write => (MAP_SHARED, PROT_READ | PROT_WRITE, access),

AccessMode::Copy => (MAP_PRIVATE, PROT_READ | PROT_WRITE, access),

AccessMode::Default => {

let access = if (prot & PROT_READ) != 0 && (prot & PROT_WRITE) != 0 {

access

} else if (prot & PROT_WRITE) != 0 {

AccessMode::Write

} else {

AccessMode::Read

};

(flags, prot, access)

}

};

let fd = unsafe { crt_fd::Borrowed::try_borrow_raw(fd) };

// macOS: Issue #11277: fsync(2) is not enough on OS X - a special, OS X specific

// fcntl(2) is necessary to force DISKSYNC and get around mmap(2) bug

#[cfg(target_os = "macos")]

if let Ok(fd) = fd {

use std::os::fd::AsRawFd;

unsafe { libc::fcntl(fd.as_raw_fd(), libc::F_FULLFSYNC) };

}

if let Ok(fd) = fd {

let metadata = fstat(fd)

.map_err(|err| io::Error::from_raw_os_error(err as i32).to_pyexception(vm))?;

let file_len = metadata.st_size as i64;

if map_size == 0 {

if file_len == 0 {

return Err(vm.new_value_error("cannot mmap an empty file"));

}

if offset > file_len {

return Err(vm.new_value_error("mmap offset is greater than file size"));

}

map_size = (file_len - offset)

.try_into()

.map_err(|_| vm.new_value_error("mmap length is too large"))?;

} else if offset > file_len || file_len - offset < map_size as i64 {

return Err(vm.new_value_error("mmap length is greater than file size"));

}

}

let mut mmap_opt = MmapOptions::new();

let mmap_opt = mmap_opt.offset(offset as u64).len(map_size);

let (fd, mmap) = || -> std::io::Result<_> {

if let Ok(fd) = fd {

let new_fd: crt_fd::Owned = unistd::dup(fd)?.into();

let mmap = match access {

AccessMode::Default | AccessMode::Write => {

MmapObj::Write(unsafe { mmap_opt.map_mut(&new_fd) }?)

}

AccessMode::Read => MmapObj::Read(unsafe { mmap_opt.map(&new_fd) }?),

AccessMode::Copy => MmapObj::Write(unsafe { mmap_opt.map_copy(&new_fd) }?),

};

Ok((Some(new_fd), mmap))

} else {

let mmap = MmapObj::Write(mmap_opt.map_anon()?);

Ok((None, mmap))

}

}()

.map_err(|e| e.to_pyexception(vm))?;

Ok(Self {

closed: AtomicCell::new(false),

mmap: PyMutex::new(Some(mmap)),

fd: AtomicCell::new(fd.map_or(-1, |fd| fd.into_raw())),

offset,

size: AtomicCell::new(map_size),

pos: AtomicCell::new(0),

exports: AtomicCell::new(0),

access,

})

}

#[cfg(windows)]

fn py_new(_cls: &Py<PyType>, args: Self::Args, vm: &VirtualMachine) -> PyResult<Self> {

let mut map_size = args.validate_new_args(vm)?;

let MmapNewArgs {

fileno,

tagname,

access,

offset,

..

} = args;

// Parse tagname: None or a string

let tag_str: Option<String> = match tagname {

Some(ref obj) if !vm.is_none(obj) => {

let s = obj

.try_to_value::<String>(vm)

.map_err(|_| vm.new_type_error("tagname must be a string or None"))?;

if s.contains('\0') {

return Err(vm.new_value_error("tagname must not contain null characters"));

}

Some(s)

}

_ => None,

};

// Get file handle from fileno

// fileno -1 or 0 means anonymous mapping

let fh: Option<HANDLE> = if fileno != -1 && fileno != 0 {

// Convert CRT file descriptor to Windows HANDLE

// Use suppress_iph! to avoid crashes when the fd is invalid.

// This is critical because socket fds wrapped via _open_osfhandle

// may cause crashes in _get_osfhandle on Windows.

// See Python bug https://bugs.python.org/issue30114

let handle = unsafe { suppress_iph!(libc::get_osfhandle(fileno)) };

// Check for invalid handle value (-1 on Windows)

if handle == -1 || handle == INVALID_HANDLE_VALUE as isize {

return Err(vm.new_os_error(format!("Invalid file descriptor: {}", fileno)));

}

Some(handle as HANDLE)

} else {

None

};

// Get file size if we have a file handle and map_size is 0

let mut duplicated_handle: HANDLE = INVALID_HANDLE_VALUE;

if let Some(fh) = fh {

// Duplicate handle so Python code can close the original

let mut new_handle: HANDLE = INVALID_HANDLE_VALUE;

let result = unsafe {

DuplicateHandle(

GetCurrentProcess(),

fh,

GetCurrentProcess(),

&mut new_handle,

0,

0, // not inheritable

DUPLICATE_SAME_ACCESS,

)

};

if result == 0 {

return Err(io::Error::last_os_error().to_pyexception(vm));

}

duplicated_handle = new_handle;

// Get file size

let mut high: u32 = 0;

let low = unsafe { GetFileSize(fh, &mut high) };

if low == u32::MAX {

let err = io::Error::last_os_error();

if err.raw_os_error() != Some(0) {

unsafe { CloseHandle(duplicated_handle) };

return Err(err.to_pyexception(vm));

}

}

let file_len = ((high as i64) << 32) | (low as i64);

if map_size == 0 {

if file_len == 0 {

unsafe { CloseHandle(duplicated_handle) };

return Err(vm.new_value_error("cannot mmap an empty file"));

}

if offset >= file_len {

unsafe { CloseHandle(duplicated_handle) };

return Err(vm.new_value_error("mmap offset is greater than file size"));

}

if file_len - offset > isize::MAX as i64 {

unsafe { CloseHandle(duplicated_handle) };

return Err(vm.new_value_error("mmap length is too large"));

}

map_size = (file_len - offset) as usize;

} else {

// If map_size > file_len, extend the file (Windows behavior)

let required_size = offset.checked_add(map_size as i64).ok_or_else(|| {

unsafe { CloseHandle(duplicated_handle) };

vm.new_overflow_error("mmap size would cause file size overflow")

})?;

if required_size > file_len {

// Extend file using SetFilePointerEx + SetEndOfFile

let result = unsafe {

SetFilePointerEx(

duplicated_handle,

required_size,

core::ptr::null_mut(),

FILE_BEGIN,

)

};

if result == 0 {

let err = io::Error::last_os_error();

unsafe { CloseHandle(duplicated_handle) };

return Err(err.to_pyexception(vm));

}

let result = unsafe { SetEndOfFile(duplicated_handle) };

if result == 0 {

let err = io::Error::last_os_error();

unsafe { CloseHandle(duplicated_handle) };

return Err(err.to_pyexception(vm));

}

}

}

}

// When tagname is provided, use raw Win32 APIs for named shared memory

if let Some(ref tag) = tag_str {

let (fl_protect, desired_access) = match access {

AccessMode::Default | AccessMode::Write => (PAGE_READWRITE, FILE_MAP_WRITE),

AccessMode::Read => (PAGE_READONLY, FILE_MAP_READ),

AccessMode::Copy => (PAGE_WRITECOPY, FILE_MAP_COPY),

};

let fh = if let Some(fh) = fh {

// Close the duplicated handle - we'll use the original

// file handle for CreateFileMappingW

if duplicated_handle != INVALID_HANDLE_VALUE {

unsafe { CloseHandle(duplicated_handle) };

}

fh

} else {

INVALID_HANDLE_VALUE

};

let tag_wide: Vec<u16> = tag.encode_utf16().chain(core::iter::once(0)).collect();

let total_size = (offset as u64)

.checked_add(map_size as u64)

.ok_or_else(|| vm.new_overflow_error("mmap offset plus size would overflow"))?;

let size_hi = (total_size >> 32) as u32;

let size_lo = total_size as u32;

let map_handle = unsafe {

CreateFileMappingW(

fh,

core::ptr::null(),

fl_protect,

size_hi,

size_lo,

tag_wide.as_ptr(),

)

};

if map_handle.is_null() {

return Err(io::Error::last_os_error().to_pyexception(vm));

}

let off_hi = (offset as u64 >> 32) as u32;

let off_lo = offset as u32;

let view =

unsafe { MapViewOfFile(map_handle, desired_access, off_hi, off_lo, map_size) };

if view.Value.is_null() {

unsafe { CloseHandle(map_handle) };

return Err(io::Error::last_os_error().to_pyexception(vm));

}

let named = NamedMmap {

map_handle,

view_ptr: view.Value as *mut u8,

len: map_size,

};

return Ok(Self {

closed: AtomicCell::new(false),

mmap: PyMutex::new(Some(MmapObj::Named(named))),

handle: AtomicCell::new(INVALID_HANDLE_VALUE as isize),

offset,

size: AtomicCell::new(map_size),

pos: AtomicCell::new(0),

exports: AtomicCell::new(0),

access,

});

}

let mut mmap_opt = MmapOptions::new();

let mmap_opt = mmap_opt.offset(offset as u64).len(map_size);

let (handle, mmap) = if duplicated_handle != INVALID_HANDLE_VALUE {

// Safety: We just duplicated this handle and it's valid

let owned_handle =

unsafe { OwnedHandle::from_raw_handle(duplicated_handle as RawHandle) };

let mmap_result = match access {

AccessMode::Default | AccessMode::Write => {

unsafe { mmap_opt.map_mut(&owned_handle) }.map(MmapObj::Write)

}

AccessMode::Read => unsafe { mmap_opt.map(&owned_handle) }.map(MmapObj::Read),

AccessMode::Copy => {

unsafe { mmap_opt.map_copy(&owned_handle) }.map(MmapObj::Write)

}

};

let mmap = mmap_result.map_err(|e| e.to_pyexception(vm))?;

// Keep the handle alive

let raw = owned_handle.as_raw_handle() as isize;

core::mem::forget(owned_handle);

(raw, mmap)

} else {

// Anonymous mapping

let mmap = mmap_opt.map_anon().map_err(|e| e.to_pyexception(vm))?;

(INVALID_HANDLE_VALUE as isize, MmapObj::Write(mmap))

};

Ok(Self {

closed: AtomicCell::new(false),

mmap: PyMutex::new(Some(mmap)),

handle: AtomicCell::new(handle),

offset,

size: AtomicCell::new(map_size),

pos: AtomicCell::new(0),

exports: AtomicCell::new(0),

access,

})

}

}

static BUFFER_METHODS: BufferMethods = BufferMethods {

obj_bytes: |buffer| buffer.obj_as::<PyMmap>().as_bytes(),

obj_bytes_mut: |buffer| buffer.obj_as::<PyMmap>().as_bytes_mut(),

release: |buffer| {

buffer.obj_as::<PyMmap>().exports.fetch_sub(1);

},

retain: |buffer| {

buffer.obj_as::<PyMmap>().exports.fetch_add(1);

},

};

impl AsBuffer for PyMmap {

fn as_buffer(zelf: &Py<Self>, _vm: &VirtualMachine) -> PyResult<PyBuffer> {

let readonly = matches!(zelf.access, AccessMode::Read);

let buf = PyBuffer::new(

zelf.to_owned().into(),

BufferDescriptor::simple(zelf.__len__(), readonly),

&BUFFER_METHODS,

);

Ok(buf)

}

}

impl AsMapping for PyMmap {

fn as_mapping() -> &'static PyMappingMethods {

static AS_MAPPING: PyMappingMethods = PyMappingMethods {

length: atomic_func!(

|mapping, _vm| Ok(PyMmap::mapping_downcast(mapping).__len__())

),

subscript: atomic_func!(|mapping, needle, vm| {

PyMmap::mapping_downcast(mapping).getitem_inner(needle, vm)

}),

ass_subscript: atomic_func!(|mapping, needle, value, vm| {

let zelf = PyMmap::mapping_downcast(mapping);

if let Some(value) = value {

PyMmap::setitem_inner(zelf, needle, value, vm)

} else {

Err(vm

.new_type_error("mmap object doesn't support item deletion".to_owned()))

}

}),

};

&AS_MAPPING

}

}

impl AsSequence for PyMmap {

fn as_sequence() -> &'static PySequenceMethods {

use rustpython_common::lock::LazyLock;

static AS_SEQUENCE: LazyLock<PySequenceMethods> = LazyLock::new(|| PySequenceMethods {

length: atomic_func!(|seq, _vm| Ok(PyMmap::sequence_downcast(seq).__len__())),

item: atomic_func!(|seq, i, vm| {

let zelf = PyMmap::sequence_downcast(seq);

zelf.getitem_by_index(i, vm)

}),

ass_item: atomic_func!(|seq, i, value, vm| {

let zelf = PyMmap::sequence_downcast(seq);

if let Some(value) = value {

PyMmap::setitem_by_index(zelf, i, value, vm)

} else {

Err(vm

.new_type_error("mmap object doesn't support item deletion".to_owned()))

}

}),

..PySequenceMethods::NOT_IMPLEMENTED

});

&AS_SEQUENCE

}

}

#[pyclass(

with(Constructor, AsMapping, AsSequence, AsBuffer, Representable),

flags(BASETYPE, HAS_WEAKREF)

)]

impl PyMmap {

fn as_bytes_mut(&self) -> BorrowedValueMut<'_, [u8]> {

PyMutexGuard::map(self.mmap.lock(), |m| {

match m.as_mut().expect("mmap closed or invalid") {

MmapObj::Read(_) => panic!("mmap can't modify a readonly memory map."),

MmapObj::Write(mmap) => &mut mmap[..],

#[cfg(windows)]

MmapObj::Named(named) => unsafe {

core::slice::from_raw_parts_mut(named.view_ptr, named.len)

},

}

})

.into()

}

fn as_bytes(&self) -> BorrowedValue<'_, [u8]> {

PyMutexGuard::map_immutable(self.mmap.lock(), |m| {

m.as_ref().expect("mmap closed or invalid").as_slice()

})

.into()

}

fn __len__(&self) -> usize {

self.size.load()

}

#[inline]

fn pos(&self) -> usize {

self.pos.load()

}

#[inline]

fn advance_pos(&self, step: usize) {

self.pos.store(self.pos() + step);

}

#[inline]

fn try_writable<R>(

&self,

vm: &VirtualMachine,

f: impl FnOnce(&mut [u8]) -> R,

) -> PyResult<R> {

if matches!(self.access, AccessMode::Read) {

return Err(vm.new_type_error("mmap can't modify a readonly memory map."));

}

match self.check_valid(vm)?.deref_mut().as_mut().unwrap() {

MmapObj::Write(mmap) => Ok(f(&mut mmap[..])),

#[cfg(windows)]

MmapObj::Named(named) => Ok(f(unsafe {

core::slice::from_raw_parts_mut(named.view_ptr, named.len)

})),

_ => unreachable!("already checked"),

}

}

fn check_valid(&self, vm: &VirtualMachine) -> PyResult<PyMutexGuard<'_, Option<MmapObj>>> {

let m = self.mmap.lock();

if m.is_none() {

return Err(vm.new_value_error("mmap closed or invalid"));

}

Ok(m)

}

/// TODO: impl resize

#[allow(dead_code)]

fn check_resizeable(&self, vm: &VirtualMachine) -> PyResult<()> {

if self.exports.load() > 0 {

return Err(vm.new_buffer_error("mmap can't resize with extant buffers exported."));

}

if self.access == AccessMode::Write || self.access == AccessMode::Default {

return Ok(());

}

Err(vm.new_type_error("mmap can't resize a readonly or copy-on-write memory map."))

}

#[pygetset]

fn closed(&self) -> bool {

self.closed.load()

}

#[pymethod]

fn close(&self, vm: &VirtualMachine) -> PyResult<()> {

if self.closed() {

return Ok(());

}

if self.exports.load() > 0 {

return Err(vm.new_buffer_error("cannot close exported pointers exist."));

}

let mut mmap = self.mmap.lock();

self.closed.store(true);

*mmap = None;

self.close_handle();

Ok(())

}

fn get_find_range(&self, options: FindOptions) -> (usize, usize) {

let size = self.__len__();

let start = options

.start

.map(|start| start.saturated_at(size))

.unwrap_or_else(|| self.pos());

let end = options

.end

.map(|end| end.saturated_at(size))

.unwrap_or(size);

(start, end)

}

#[pymethod]

fn find(&self, options: FindOptions, vm: &VirtualMachine) -> PyResult<PyInt> {

let (start, end) = self.get_find_range(options.clone());

let sub = &options.sub;

// returns start position for empty string

if sub.is_empty() {

return Ok(PyInt::from(start as isize));

}

let mmap = self.check_valid(vm)?;

let buf = &mmap.as_ref().unwrap().as_slice()[start..end];

let pos = buf.windows(sub.len()).position(|window| window == sub);

Ok(pos.map_or(PyInt::from(-1isize), |i| PyInt::from(start + i)))

}

#[pymethod]

fn rfind(&self, options: FindOptions, vm: &VirtualMachine) -> PyResult<PyInt> {

let (start, end) = self.get_find_range(options.clone());

let sub = &options.sub;

// returns start position for empty string

if sub.is_empty() {

return Ok(PyInt::from(start as isize));

}

let mmap = self.check_valid(vm)?;

let buf = &mmap.as_ref().unwrap().as_slice()[start..end];