Coverage for wasmtime/component/_linker.py: 98%
148 statements
« prev ^ index » next coverage.py v7.11.3, created at 2026-05-27 18:55 +0000
1import ctypes
2from ctypes import POINTER, byref, create_string_buffer, CFUNCTYPE, c_void_p
3from .. import Managed, Engine, WasmtimeError, Module, _ffi as ffi, Storelike, StoreContext
4from .._config import setter_property
5from ._instance import Instance
6from ._component import Component
7from ._resource_type import ResourceType
8from typing import Union, Tuple, Callable, Optional, List, Any
9from .._slab import Slab
10from ._enter import catch_exceptions
11from ._types import ValType, FuncType
13LinkerInstanceParent = Union['Linker', 'LinkerInstance']
15ResourceDtor = Callable[[StoreContext, int], None]
16RESOURCE_DTORS: Slab[ResourceDtor] = Slab()
18UserFunc = Callable[..., Optional[Any]]
19FUNCTIONS: Slab[UserFunc] = Slab()
22class Linker(Managed["ctypes._Pointer[ffi.wasmtime_component_linker_t]"]):
23 engine: Engine
24 locked: bool
26 def __init__(self, engine: Engine):
27 """
28 Creates a new linker ready to instantiate modules within the store
29 provided.
30 """
31 self._set_ptr(ffi.wasmtime_component_linker_new(engine.ptr()))
32 self.engine = engine
33 self.locked = False
35 def _delete(self, ptr: "ctypes._Pointer[ffi.wasmtime_component_linker_t]") -> None:
36 self._assert_not_locked()
37 ffi.wasmtime_component_linker_delete(ptr)
39 @setter_property
40 def allow_shadowing(self, allow: bool) -> None:
41 """
42 Configures whether definitions are allowed to shadow one another within
43 this linker
44 """
45 self._assert_not_locked()
46 ffi.wasmtime_component_linker_allow_shadowing(self.ptr(), allow)
48 def _assert_not_locked(self) -> None:
49 if self.locked:
50 raise WasmtimeError("cannot use linker while it's in use by other instances")
52 def root(self) -> "LinkerInstance":
53 """
54 Returns the root instance used to defined new items in this linker.
56 While this root instance is alive this linker cannot be used for any
57 other operations. Until the returned object is destroyed this linker is
58 considered "locked".
60 It's recommended to bind the return value in a `with` block to
61 automatically dispose of it when it's done.
62 """
63 self._assert_not_locked()
64 ptr = ffi.wasmtime_component_linker_root(self.ptr())
65 return LinkerInstance._from_ptr(ptr, self)
67 def add_wasip2(self) -> None:
68 """
69 Adds the WASIp2 API definitions, from Wasmtime, in this linker.
70 """
71 self._assert_not_locked()
72 err = ffi.wasmtime_component_linker_add_wasip2(self.ptr())
73 if err:
74 raise WasmtimeError._from_ptr(err)
76 def add_wasi_http(self) -> None:
77 """
78 Adds the WASI HTTP API definitions to this linker.
80 This adds `wasi:http/types` and `wasi:http/outgoing-handler`.
81 Requires WASIp2 to be added first via `add_wasip2()`.
82 """
83 self._assert_not_locked()
84 err = ffi.wasmtime_component_linker_add_wasi_http(self.ptr())
85 if err:
86 raise WasmtimeError._from_ptr(err)
88 def instantiate(self, store: Storelike, component: Component) -> Instance:
89 """
90 Instantiates the given component using this linker within the provided
91 store.
93 Returns the instantiated `Instance` on success.
94 """
95 self._assert_not_locked()
96 instance = ffi.wasmtime_component_instance_t()
97 err = ffi.wasmtime_component_linker_instantiate(
98 self.ptr(),
99 store._context(),
100 component.ptr(),
101 byref(instance))
102 if err:
103 raise WasmtimeError._from_ptr(err)
104 return Instance._from_raw(instance)
106 def define_unknown_imports_as_traps(self, component: Component) -> None:
107 """
108 Configures this linker to define any unknown imports of `component` as
109 traps which will error when invoked.
110 """
111 self._assert_not_locked()
112 err = ffi.wasmtime_component_linker_define_unknown_imports_as_traps(
113 self.ptr(),
114 component.ptr())
115 if err:
116 raise WasmtimeError._from_ptr(err)
119class LinkerInstance(Managed["ctypes._Pointer[ffi.wasmtime_component_linker_instance_t]"]):
120 parent: Union[LinkerInstanceParent, None]
121 locked: bool
123 def __init__(self) -> None:
124 raise WasmtimeError("Cannot directly construct a `LinkerInstance`")
126 def _delete(self, ptr: "ctypes._Pointer[ffi.wasmtime_component_linker_instance_t]") -> None:
127 self._assert_not_locked()
128 ffi.wasmtime_component_linker_instance_delete(ptr)
129 assert(self.parent is not None)
130 assert(self.parent.locked)
131 self.parent.locked = False
132 self.parent = None
134 def _assert_not_locked(self) -> None:
135 if self.locked:
136 raise WasmtimeError("cannot use linker instance while it's in use by other instances")
138 @classmethod
139 def _from_ptr(cls, ptr: "ctypes._Pointer[ffi.wasmtime_component_linker_instance_t]", parent: LinkerInstanceParent) -> "LinkerInstance":
140 if not isinstance(ptr, POINTER(ffi.wasmtime_component_linker_instance_t)):
141 raise TypeError("wrong pointer type")
142 ty: "LinkerInstance" = cls.__new__(cls)
143 ty._set_ptr(ptr)
144 ty.parent = parent
145 ty.locked = False
146 assert(not parent.locked)
147 parent.locked = True
148 return ty
151 def add_instance(self, name: str) -> "LinkerInstance":
152 """
153 Adds a new instance to this linker instance under the given name.
155 Returns a new `LinkerInstance` which can be used to define further
156 items.
158 While the returned instance is alive this linker instance cannot be
159 used for other operations. Until the returned object is destroyed this
160 linker instance is considered "locked".
162 It's recommended to bind the return value in a `with` block to
163 automatically dispose of it when it's done.
164 """
165 self._assert_not_locked()
166 name_bytes = name.encode('utf-8')
167 name_buf = create_string_buffer(name_bytes)
168 ptr = POINTER(ffi.wasmtime_component_linker_instance_t)()
169 err = ffi.wasmtime_component_linker_instance_add_instance(self.ptr(),
170 name_buf,
171 len(name_bytes),
172 byref(ptr))
173 if err:
174 raise WasmtimeError._from_ptr(err)
175 return LinkerInstance._from_ptr(ptr, self)
177 def add_module(self, name: str, module: Module) -> None:
178 """
179 Adds a new module to this linker instance under the given name.
180 """
181 self._assert_not_locked()
182 name_bytes = name.encode('utf-8')
183 name_buf = create_string_buffer(name_bytes)
184 err = ffi.wasmtime_component_linker_instance_add_module(self.ptr(),
185 name_buf,
186 len(name_bytes),
187 module.ptr())
188 if err:
189 raise WasmtimeError._from_ptr(err)
191 def add_resource(self, name: str, ty: ResourceType, dtor: ResourceDtor) -> None:
192 """
193 Defines a the resource type `ty` under `name`
195 When a value of this resource is destroyed then `dtor` is invoked.
196 """
197 self._assert_not_locked()
198 name_bytes = name.encode('utf-8')
199 name_buf = create_string_buffer(name_bytes)
200 idx = RESOURCE_DTORS.allocate(dtor)
201 err = ffi.wasmtime_component_linker_instance_add_resource(self.ptr(),
202 name_buf,
203 len(name_bytes),
204 ty.ptr(),
205 resource_dtor_impl,
206 idx,
207 resource_dtor_finalize)
208 if err:
209 raise WasmtimeError._from_ptr(err)
211 def add_func(self, name: str, func: UserFunc) -> None:
212 self._assert_not_locked()
213 name_bytes = name.encode('utf-8')
214 name_buf = create_string_buffer(name_bytes)
215 idx = FUNCTIONS.allocate(func)
216 err = ffi.wasmtime_component_linker_instance_add_func(self.ptr(),
217 name_buf,
218 len(name_bytes),
219 func_impl,
220 idx,
221 func_finalize)
222 if err:
223 raise WasmtimeError._from_ptr(err)
226@ffi.wasmtime_component_resource_destructor_t
227def resource_dtor_impl(idx, context, rep): # type: ignore
228 def run(store: StoreContext) -> None:
229 return RESOURCE_DTORS.get(idx or 0)(store, rep)
230 return catch_exceptions(context, run)
233@CFUNCTYPE(None, c_void_p)
234def resource_dtor_finalize(idx): # type: ignore
235 if RESOURCE_DTORS:
236 RESOURCE_DTORS.deallocate(idx or 0)
237 return None
240@ffi.wasmtime_component_func_callback_t
241def func_impl(idx, context, ty_raw, args, nargs, results, nresults): # type: ignore
242 def run(store: StoreContext) -> None:
243 with FuncType._from_ptr(ty_raw, owner=store) as fty:
244 func = FUNCTIONS.get(idx or 0)
245 param_tys = fty.params
246 result_ty = fty.result
247 assert(len(param_tys) == nargs)
248 if result_ty is None:
249 assert(nresults == 0)
250 else:
251 assert(nresults == 1)
253 pyargs = []
254 for (_, ty), i in zip(param_tys, range(nargs)):
255 val = ty.convert_from_c(args[i])
256 pyargs.append(val)
257 result = func(store, *pyargs)
259 if result_ty is None:
260 assert(result is None)
261 else:
262 result_ty.convert_to_c(store, result, results)
263 return catch_exceptions(context, run)
266@CFUNCTYPE(None, c_void_p)
267def func_finalize(idx): # type: ignore
268 if FUNCTIONS:
269 FUNCTIONS.deallocate(idx or 0)
270 return None