Coverage for wasmtime/_sharedmemory.py: 80%

41 statements  

« prev     ^ index     » next       coverage.py v7.11.3, created at 2025-12-01 19:40 +0000

1from . import _ffi as ffi 

2import ctypes 

3from typing import Optional, Any 

4from wasmtime import MemoryType, WasmtimeError, Engine, Managed 

5from ._store import Storelike 

6 

7 

8 

9class SharedMemory(Managed["ctypes._Pointer[ffi.wasmtime_sharedmemory_t]"]): 

10 def __init__(self, engine: Engine, ty: MemoryType): 

11 """ 

12 Creates a new shared memory in `store` with the given `ty` 

13 """ 

14 

15 sharedmemory_ptr = ctypes.POINTER(ffi.wasmtime_sharedmemory_t)() 

16 error = ffi.wasmtime_sharedmemory_new(engine.ptr(), ty.ptr(), ctypes.byref(sharedmemory_ptr)) 

17 if error: 

18 raise WasmtimeError._from_ptr(error) 

19 self._set_ptr(sharedmemory_ptr) 

20 

21 def _delete(self, ptr: "ctypes._Pointer[ffi.wasmtime_sharedmemory_t]") -> None: 

22 ffi.wasmtime_sharedmemory_delete(ptr) 

23 

24 @classmethod 

25 def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasmtime_sharedmemory_t]") -> "SharedMemory": 

26 if not isinstance(ptr, ctypes.POINTER(ffi.wasmtime_sharedmemory_t)): 

27 raise TypeError("wrong shared memory pointer type provided to _from_ptr") 

28 

29 ty: "SharedMemory" = cls.__new__(cls) 

30 ty._set_ptr(ptr) 

31 return ty 

32 

33 def type(self) -> MemoryType: 

34 """ 

35 Gets the type of this memory as a `MemoryType` 

36 """ 

37 

38 ptr = ffi.wasmtime_sharedmemory_type(self.ptr()) 

39 return MemoryType._from_ptr(ptr, None) 

40 

41 def grow(self, delta: int) -> int: 

42 """ 

43 Grows this memory by the given number of pages 

44 """ 

45 

46 if delta < 0: 

47 raise WasmtimeError("cannot grow by negative amount") 

48 prev = ctypes.c_uint64(0) 

49 error = ffi.wasmtime_sharedmemory_grow(self.ptr(), delta, ctypes.byref(prev)) 

50 if error: 

51 raise WasmtimeError._from_ptr(error) 

52 return prev.value 

53 

54 def size(self) -> int: 

55 """ 

56 Returns the size, in WebAssembly pages, of this shared memory. 

57 """ 

58 

59 return ffi.wasmtime_sharedmemory_size(ctypes.byref(self.ptr())) 

60 

61 def data_ptr(self) -> "ctypes._Pointer[ctypes.c_ubyte]": 

62 """ 

63 Returns the raw pointer in memory where this wasm shared memory lives. 

64 

65 Remember that all accesses to wasm shared memory should be bounds-checked 

66 against the `data_len` method. 

67 """ 

68 return ffi.wasmtime_sharedmemory_data(self.ptr()) 

69 

70 def data_len(self) -> int: 

71 """ 

72 Returns the raw byte length of this memory. 

73 """ 

74 

75 return ffi.wasmtime_sharedmemory_data_size(self.ptr()) 

76 

77 def _as_extern(self) -> ffi.wasmtime_extern_t: 

78 union = ffi.wasmtime_extern_union(sharedmemory=ctypes.pointer(self.ptr())) 

79 return ffi.wasmtime_extern_t(ffi.WASMTIME_EXTERN_SHAREDMEMORY, union)