Coverage for wasmtime/_sharedmemory.py: 81%

42 statements  

« prev     ^ index     » next       coverage.py v7.6.12, created at 2025-02-20 16:25 +0000

1from . import _ffi as ffi 

2from ctypes import * 

3import ctypes 

4from typing import Optional, Any 

5from wasmtime import MemoryType, WasmtimeError, Engine, Managed 

6from ._store import Storelike 

7 

8 

9 

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

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

12 """ 

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

14 """ 

15 

16 sharedmemory_ptr = POINTER(ffi.wasmtime_sharedmemory_t)() 

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

18 if error: 

19 raise WasmtimeError._from_ptr(error) 

20 self._set_ptr(sharedmemory_ptr) 

21 

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

23 ffi.wasmtime_sharedmemory_delete(ptr) 

24 

25 @classmethod 

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

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

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

29 

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

31 ty._set_ptr(ptr) 

32 return ty 

33 

34 def type(self) -> MemoryType: 

35 """ 

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

37 """ 

38 

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

40 return MemoryType._from_ptr(ptr, None) 

41 

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

43 """ 

44 Grows this memory by the given number of pages 

45 """ 

46 

47 if delta < 0: 

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

49 prev = ffi.c_uint64(0) 

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

51 if error: 

52 raise WasmtimeError._from_ptr(error) 

53 return prev.value 

54 

55 def size(self) -> int: 

56 """ 

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

58 """ 

59 

60 return ffi.wasmtime_sharedmemory_size(byref(self.ptr())) 

61 

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

63 """ 

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

65 

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

67 against the `data_len` method. 

68 """ 

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

70 

71 def data_len(self) -> int: 

72 """ 

73 Returns the raw byte length of this memory. 

74 """ 

75 

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

77 

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

79 union = ffi.wasmtime_extern_union(sharedmemory=pointer(self.ptr())) 

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