import atexit import re from pathlib import Path from typing import Any, Dict, Optional, Sequence from .ffi import ffi, load_mono from .types import Runtime, RuntimeInfo, StrOrPath from .util import optional_path_as_string, path_as_string __all__ = ["Mono"] _MONO: Any = None _ROOT_DOMAIN: Any = None class Mono(Runtime): def __init__( self, libmono: Optional[Path], *, domain: Optional[str] = None, debug: bool = False, jit_options: Optional[Sequence[str]] = None, config_file: Optional[Path] = None, global_config_file: Optional[Path] = None, assembly_dir: Optional[str] = None, config_dir: Optional[str] = None, set_signal_chaining: bool = False, trace_mask: Optional[str] = None, trace_level: Optional[str] = None, ): self._assemblies: Dict[Path, Any] = {} self._version: str = initialize( config_file=optional_path_as_string(config_file), debug=debug, jit_options=jit_options, global_config_file=optional_path_as_string(global_config_file), libmono=libmono, assembly_dir=assembly_dir, config_dir=config_dir, set_signal_chaining=set_signal_chaining, trace_mask=trace_mask, trace_level=trace_level, ) if domain is None: self._domain = _ROOT_DOMAIN else: raise NotImplementedError def _get_callable( self, assembly_path: StrOrPath, typename: str, function: str ) -> "MonoMethod": assembly_path = Path(assembly_path) assembly = self._assemblies.get(assembly_path) if not assembly: assembly = _MONO.mono_domain_assembly_open( self._domain, path_as_string(assembly_path).encode("utf8") ) _check_result(assembly, f"Unable to load assembly {assembly_path}") self._assemblies[assembly_path] = assembly image = _MONO.mono_assembly_get_image(assembly) _check_result(image, "Unable to load image from assembly") desc = MethodDesc(typename, function) method = desc.search(image) _check_result( method, f"Could not find method {typename}.{function} in assembly" ) return MonoMethod(method) def info(self) -> RuntimeInfo: return RuntimeInfo( kind="Mono", version=self._version, initialized=True, shutdown=_MONO is None, properties={}, ) def shutdown(self) -> None: # We don't implement non-root-domains, yet. When we do, it needs to be # released here. pass class MethodDesc: def __init__(self, typename: str, function: str): self._desc = f"{typename}:{function}" self._ptr = _MONO.mono_method_desc_new( self._desc.encode("utf8"), 1, # include_namespace ) def search(self, image: str): return _MONO.mono_method_desc_search_in_image(self._ptr, image) def __del__(self): if _MONO: _MONO.mono_method_desc_free(self._ptr) class MonoMethod: def __init__(self, ptr): self._ptr = ptr def __call__(self, ptr, size): exception = ffi.new("MonoObject**") params = ffi.new("void*[2]") # Keep these alive until the function is called by assigning them locally ptr_ptr = ffi.new("void**", ptr) size_ptr = ffi.new("int32_t*", size) params[0] = ptr_ptr params[1] = size_ptr res = _MONO.mono_runtime_invoke(self._ptr, ffi.NULL, params, exception) _check_result(res, "Failed to call method") unboxed = ffi.cast("int32_t*", _MONO.mono_object_unbox(res)) _check_result(unboxed, "Failed to convert result to int") return unboxed[0] def initialize( libmono: Optional[Path], debug: bool = False, jit_options: Optional[Sequence[str]] = None, config_file: Optional[str] = None, global_config_file: Optional[str] = None, assembly_dir: Optional[str] = None, config_dir: Optional[str] = None, set_signal_chaining: bool = False, trace_mask: Optional[str] = None, trace_level: Optional[str] = None, ) -> str: global _MONO, _ROOT_DOMAIN if _MONO is None: _MONO = load_mono(libmono) if trace_mask is not None: _MONO.mono_trace_set_mask_string(trace_mask.encode("utf8")) if trace_level is not None: _MONO.mono_trace_set_level_string(trace_level.encode("utf8")) if assembly_dir is not None and config_dir is not None: _MONO.mono_set_dirs(assembly_dir.encode("utf8"), config_dir.encode("utf8")) # Load in global config (i.e /etc/mono/config) global_encoded = global_config_file or ffi.NULL _MONO.mono_config_parse(global_encoded) # Even if we don't have a domain config file, we still need to set it # as something, see https://github.com/pythonnet/clr-loader/issues/8 if config_file is None: config_file = "" config_encoded = config_file.encode("utf8") if jit_options: options = [ffi.new("char[]", o.encode("utf8")) for o in jit_options] _MONO.mono_jit_parse_options(len(options), options) else: options = [] if set_signal_chaining: _MONO.mono_set_signal_chaining(True) if debug: _MONO.mono_debug_init(_MONO.MONO_DEBUG_FORMAT_MONO) _ROOT_DOMAIN = _MONO.mono_jit_init(b"clr_loader") _MONO.mono_domain_set_config(_ROOT_DOMAIN, b".", config_encoded) _check_result(_ROOT_DOMAIN, "Failed to initialize Mono") build = _MONO.mono_get_runtime_build_info() _check_result(build, "Failed to get Mono version") ver_str = ffi.string(build).decode("utf8") # e.g. '6.12.0.122 (tarball)' ver = re.match(r"^(?P\d+)\.(?P\d+)\.[\d.]+", ver_str) if ver is not None: major = int(ver.group("major")) minor = int(ver.group("minor")) if major < 6 or (major == 6 and minor < 12): import warnings warnings.warn( "Hosting Mono versions before v6.12 is known to be problematic. " "If the process crashes shortly after you see this message, try " "updating Mono to at least v6.12." ) atexit.register(_release) return ver_str def _release() -> None: global _MONO, _ROOT_DOMAIN if _ROOT_DOMAIN is not None and _MONO is not None: _MONO.mono_jit_cleanup(_ROOT_DOMAIN) _MONO = None _ROOT_DOMAIN = None def _check_result(res: Any, msg: str) -> None: if res == ffi.NULL or not res: raise RuntimeError(msg)