Coverage for wasmtime/_memory.py: 97%

1from . import _ffi as ffi 

2from ctypes import * 

3import ctypes 

4import typing 

5from wasmtime import MemoryType, WasmtimeError 

6from ._store import Storelike 

7 

8 

9class Memory: 

10 _memory: ffi.wasmtime_memory_t 

11 

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

13 """ 

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

15 """ 

16 

17 mem = ffi.wasmtime_memory_t() 

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

19 if error: 

20 raise WasmtimeError._from_ptr(error) 

21 self._memory = mem 

22 

23 @classmethod 

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

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

26 ty._memory = mem 

27 return ty 

28 

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

30 """ 

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

32 """ 

33 

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

35 return MemoryType._from_ptr(ptr, None) 

36 

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

38 """ 

39 Grows this memory by the given number of pages 

40 """ 

41 

42 if delta < 0: 

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

44 prev = ffi.c_uint64(0) 

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

46 if error: 

47 raise WasmtimeError._from_ptr(error) 

48 return prev.value 

49 

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

51 """ 

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

53 """ 

54 

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

56 

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

58 """ 

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

60 

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

62 against the `data_len` method. 

63 """ 

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

65 

66 def get_buffer_ptr(self, store: Storelike, 

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

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

69 """ 

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

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

72 

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

74 np_mem[start:end] = A # write 

75 B = np_mem[start:end] # read 

76 """ 

77 if size is None: 

78 size = self.data_len(store) 

79 ptr_type = ctypes.c_ubyte * size 

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

81 

82 def read( 

83 self, 

84 store: Storelike, 

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

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

87 """ 

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

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

90 

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

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

93 """ 

94 size = self.data_len(store) 

95 key = slice(start, stop, None) 

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

97 val_size = stop - start 

98 if val_size <= 0: 

99 # return bytearray of size zero 

100 return bytearray(0) 

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

102 return bytearray(src_ptr) 

103 

104 def write( 

105 self, 

106 store: Storelike, 

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

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

109 """ 

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

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

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

113 return number of bytes written 

114 raises IndexError when trying to write outside the memory range 

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

116 """ 

117 size = self.data_len(store) 

118 key = slice(start, None) 

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

120 if start >= size: 

121 raise IndexError("index out of range") 

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

123 if not isinstance(value, bytearray): 

124 value = bytearray(value) 

125 val_size = len(value) 

126 if val_size == 0: 

127 return val_size 

128 # stop is exclusive 

129 stop = start + val_size 

130 if stop > size: 

131 raise IndexError("index out of range") 

132 ptr_type = ctypes.c_ubyte * val_size 

133 src_ptr = ptr_type.from_buffer(value) 

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

135 ctypes.memmove(dst_ptr, src_ptr, val_size) 

136 return val_size 

137 

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

139 """ 

140 Returns the raw byte length of this memory. 

141 """ 

142 

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

144 

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

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

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