Coverage for wasmtime/_memory.py: 97%

1from . import _ffi as ffi 

2import ctypes 

3import typing 

4from wasmtime import MemoryType, WasmtimeError 

5from ._store import Storelike 

6 

7 

8class Memory: 

9 _memory: ffi.wasmtime_memory_t 

10 

11 def __init__(self, store: Storelike, ty: MemoryType): 

12 """ 

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

14 """ 

15 

16 mem = ffi.wasmtime_memory_t() 

17 error = ffi.wasmtime_memory_new(store._context(), ty.ptr(), ctypes.byref(mem)) 

18 if error: 

19 raise WasmtimeError._from_ptr(error) 

20 self._memory = mem 

21 

22 @classmethod 

23 def _from_raw(cls, mem: ffi.wasmtime_memory_t) -> "Memory": 

24 ty: "Memory" = cls.__new__(cls) 

25 ty._memory = mem 

26 return ty 

27 

28 def type(self, store: Storelike) -> MemoryType: 

29 """ 

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

31 """ 

32 

33 ptr = ffi.wasmtime_memory_type(store._context(), ctypes.byref(self._memory)) 

34 return MemoryType._from_ptr(ptr, None) 

35 

36 def grow(self, store: Storelike, delta: int) -> int: 

37 """ 

38 Grows this memory by the given number of pages 

39 """ 

40 

41 if delta < 0: 

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

43 prev = ctypes.c_uint64(0) 

44 error = ffi.wasmtime_memory_grow(store._context(), ctypes.byref(self._memory), delta, ctypes.byref(prev)) 

45 if error: 

46 raise WasmtimeError._from_ptr(error) 

47 return prev.value 

48 

49 def size(self, store: Storelike) -> int: 

50 """ 

51 Returns the size, in WebAssembly pages, of this memory. 

52 """ 

53 

54 return ffi.wasmtime_memory_size(store._context(), ctypes.byref(self._memory)) 

55 

56 def data_ptr(self, store: Storelike) -> "ctypes._Pointer[ctypes.c_ubyte]": 

57 """ 

58 Returns the raw pointer in memory where this wasm memory lives. 

59 

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

61 against the `data_len` method. 

62 """ 

63 return ffi.wasmtime_memory_data(store._context(), ctypes.byref(self._memory)) 

64 

65 def get_buffer_ptr(self, store: Storelike, 

66 size: typing.Optional[int] = None, 

67 offset: int = 0) -> ctypes.Array: 

68 """ 

69 return raw pointer to buffer suitable for creating zero-copy writable NumPy Buffer Protocol 

70 this method is also used internally by `read()` and `write()` 

71 

72 np_mem = np.frombuffer(memory.get_buffer_ptr(store), dtype=np.uint8) 

73 np_mem[start:end] = A # write 

74 B = np_mem[start:end] # read 

75 """ 

76 if size is None: 

77 size = self.data_len(store) 

78 ptr_type = ctypes.c_ubyte * size 

79 return ptr_type.from_address(ctypes.addressof(self.data_ptr(store).contents) + offset) 

80 

81 def read( 

82 self, 

83 store: Storelike, 

84 start: typing.Optional[int] = 0, 

85 stop: typing.Optional[int] = None) -> bytearray: 

86 """ 

87 Reads this memory starting from `start` and up to `stop` 

88 and returns a copy of the contents as a `bytearray`. 

89 

90 The indexing behavior of this method is similar to `list[start:stop]` 

91 where negative starts can be used to read from the end, for example. 

92 """ 

93 size = self.data_len(store) 

94 key = slice(start, stop, None) 

95 start, stop, _ = key.indices(size) 

96 val_size = stop - start 

97 if val_size <= 0: 

98 # return bytearray of size zero 

99 return bytearray(0) 

100 src_ptr = self.get_buffer_ptr(store, val_size, start) 

101 return bytearray(src_ptr) 

102 

103 def write( 

104 self, 

105 store: Storelike, 

106 value: typing.Union[bytearray, bytes], 

107 start: typing.Optional[int] = None) -> int: 

108 """ 

109 write a bytearray value into a possibly large slice of memory 

110 negative start is allowed in a way similat to list slice mylist[-10:] 

111 if value is not bytearray it will be used to construct an intermediate bytearray (copyied twice) 

112 return number of bytes written 

113 raises IndexError when trying to write outside the memory range 

114 this happens when start offset is >= size or when end side of value is >= size 

115 """ 

116 size = self.data_len(store) 

117 key = slice(start, None) 

118 start = key.indices(size)[0] 

119 if start >= size: 

120 raise IndexError("index out of range") 

121 # value must be bytearray ex. cast bytes() to bytearray 

122 if not isinstance(value, bytearray): 

123 value = bytearray(value) 

124 val_size = len(value) 

125 if val_size == 0: 

126 return val_size 

127 # stop is exclusive 

128 stop = start + val_size 

129 if stop > size: 

130 raise IndexError("index out of range") 

131 ptr_type = ctypes.c_ubyte * val_size 

132 src_ptr = ptr_type.from_buffer(value) 

133 dst_ptr = self.get_buffer_ptr(store, val_size, start) 

134 ctypes.memmove(dst_ptr, src_ptr, val_size) 

135 return val_size 

136 

137 def data_len(self, store: Storelike) -> int: 

138 """ 

139 Returns the raw byte length of this memory. 

140 """ 

141 

142 return ffi.wasmtime_memory_data_size(store._context(), ctypes.byref(self._memory)) 

143 

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

145 union = ffi.wasmtime_extern_union(memory=self._memory) 

146 return ffi.wasmtime_extern_t(ffi.WASMTIME_EXTERN_MEMORY, union)