Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
Args:
seed (Path): Generated jsfunfuzz file (acts as the seed for binaryen)
Returns:
bool: Returns True on successful wasm-opt execution, False otherwise
"""
assert platform.system() == "Linux"
assert seed.is_file()
seed_wrapper_output = seed.resolve().with_suffix(".wrapper")
seed_wasm_output = seed.resolve().with_suffix(".wasm")
sleep_time = 2
t_lock = threading.Lock()
with fasteners.try_lock(t_lock) as gotten:
while True:
if gotten:
try:
# Wrapping this in str() seems necessary for Python 3.7.x and lower.
# See Python issue 31961
subprocess.run([str(ensure_binaryen(BINARYEN_URL, BINARYEN_VERSION)),
str(seed),
"--translate-to-fuzz",
"--disable-simd",
"--output", str(seed_wasm_output),
f"--emit-js-wrapper={seed_wrapper_output}"], check=True)
except (subprocess.CalledProcessError, OSError):
print("wasm-opt aborted with a CalledProcessError or OSError. Trying again after 1 minute...")
sleep(60)
# Wrapping this in str() seems necessary for Python 3.7.x and lower.
# See Python issue 31961
def run(self, timeout=None):
"""Runs the engine (or die trying).
:param timeout: timeout to wait for any atoms to complete (this timeout
will be used during the waiting period that occurs when
unfinished atoms are being waited on).
"""
with fasteners.try_lock(self._lock) as was_locked:
if not was_locked:
raise exc.ExecutionFailure("Engine currently locked, please"
" try again later")
for _state in self.run_iter(timeout=timeout):
pass
Path: Path of the extracted wasm-opt binary
"""
shell_cache = sm_compile_helpers.ensure_cache_dir(Path.home())
binaryen_path = shell_cache / f"binaryen-version_{version}"
wasmopt_path = binaryen_path / ("wasm-opt" + (".exe" if platform.system() == "Windows" else ""))
if wasmopt_path.is_file():
return wasmopt_path
# binaryen archives get extracted to a different name on Windows
binaryen_path = shell_cache / (f"binaryen-version_{version}" +
("-x86_64-windows" if platform.system() == "Windows" else ""))
wasmopt_path = binaryen_path / ("wasm-opt" + (".exe" if platform.system() == "Windows" else ""))
sleep_time = 2
t_lock = threading.Lock()
with fasteners.try_lock(t_lock) as gotten:
while not wasmopt_path.is_file():
if gotten:
with requests.get(url, allow_redirects=True, stream=True) as binaryen_gzip_request:
try:
with tarfile.open(fileobj=io.BytesIO(binaryen_gzip_request.content), mode="r:gz") as f:
f.extractall(str(shell_cache.resolve()))
except OSError:
print("binaryen tarfile threw an OSError")
break
sleep(sleep_time)
sleep_time *= 2
if platform.system() == "Windows":
assert binaryen_path.is_dir()
new_win_binaryen_path = shell_cache / f"binaryen-version_{version}"
binaryen_path.rename(new_win_binaryen_path)
def wrapper(path, *args, **kwargs):
lockfile_name = os.path.join(path, LOCKFILE_NAME)
with fasteners.try_lock(fasteners.InterProcessLock(lockfile_name)) as locked:
if not locked:
raise RuntimeError("Failed to lock cache %r." % path)
return wrapped(path, *args, **kwargs)