Is torch safe?

results: list[bytes] = pickle.loads(data)

state = pickle.loads(data)

bundle = pickle.loads(data)

return pickle.loads(guards_state)

entry = pickle.loads(pickled_content)

_CODE_STATE = pickle.loads(content)

code_state = pickle.loads(payload)

return dill.load(f)

dst = load(content, Loader=Loader)

compiled_fn, lambda: pickle.loads(pickled_content)

inputs_meta = pickle.load(f)

return pickle.load(read_pipe)

key = pickle.loads(key_bytes)

value = pickle.loads(value_bytes)

cache._cache = pickle.load(fp)

value = pickle.loads(value_bytes)

yield pickle.loads(content), content, True

yield pickle.loads(content), content, False

return pickle.loads(data)

args, kwargs = pickle.load(f)

state = pickle.load(f)

cache_entry = cast(CacheEntry, pickle.loads(pickled_value))

graph = pickle.load(f)

snapshot_dict = cast(_Snapshot, pickle.load(f))

return pickle.loads(data)

exec(custom_rearrange_callable_code, globals(), namespace)

return compile(  # pyrefly: ignore  # no-matching-overload
                model,
                fullgraph=fullgraph,
                dynamic=dynamic,
                backend=backend,
                mode=mode,
                options=opti …

result = eval(arg, frame_globals, eval_locals)

result = eval(cmd, frame_globals, eval_locals)

exec(cmd, frame_globals, eval_locals)

exec(func_source, namespace)

compile(f"mod = {module_str}", "<module repr>", "exec")

exec(func_str, {"op": fn}, out)

exec(pycode, globals_for_guard_fn, out)

tensor_id = id(eval(tensor_source, global_scope, scope))

fail_reason = eval(part, global_scope, scope)

eval(guard, guard_manager.global_scope, local_scope)

idx: compile(expr_str, f"<derived arg {idx}>", "eval")

syms[f"_derived_{idx}"] = eval(code, {"__builtins__": {}}, syms)

fn = eval(f"lambda {','.join(vars)}: ({','.join(rotated)})")

result = eval(
                    compile(tree, filename, "eval"),
                    {"__builtins__": {}},
                    {},
                )

compile(tree, filename, "eval"),

_input_associated_real_value = eval(self.source.name, scope)

code = compile(source, f"{__file__}:codegen({artifact_name})", "exec")

exec(code, globals_dict, local_dict)

value = eval(self._name_template, globals, locals)

value = eval(self._name_template.format(tmpvar), globals, locals)

result = eval(expr, {"__builtins__": {}}, namespace)

exec(code, namespace)

exec(tuning_code, scope)

subprocess.run(
            [
                "wget",
                "-O",
                magic_trace_cache,
                "-q",
                "https://github.com/janestreet/magic-trace/releases/download/v1.0.2/magic-trace", …

subprocess.run(["chmod", "+x", magic_trace_cache])

p = subprocess.Popen(args, stderr=subprocess.PIPE, encoding="utf-8")

with subprocess.Popen(
                [
                    "gprof2dot",
                    "-f",
                    "pstats",
                    "--node-label=total-time-percentage",
                    "--node-label=self-time-percenta …

subprocess.check_call(
                    ["dot", "-Tsvg", "-o", str(svg_path)],
                    stdin=gprof2dot_process.stdout,
                )

cuda_version_out = subprocess.check_output(["nvcc", "--version"])

gprof2dot = subprocess.Popen(
                [
                    "gprof2dot",
                    "-f",
                    "pstats",
                    "--node-label=total-time-percentage",
                    "--node-label=self-time-p …

dot = subprocess.Popen(
                ["dot", "-Tsvg", "-o", svg_file],
                stdin=gprof2dot.stdout,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
            )

subprocess.Popen(
            cmd,
            cwd=subdir,
            stdout=stdout,
            stderr=stderr,
            env=new_env,
        ) as p,

self.process = subprocess.Popen(
            cmd,
            env=env,
            pass_fds=(subproc_read_fd, subproc_write_fd),
        )

subprocess.run(
                                cmd, capture_output=True, text=True, check=True
                            )

cmd_output = subprocess.run(
                ("openssl", "sha512", filename), capture_output=True, text=True
            )

out = subprocess.check_output(
                            ["ldd", os.path.join(search, file)]
                        )

subprocess.check_call(
                        shlex.split(halide_cmd_gen.get_command_line())
                    )

subprocess.check_output(
                                cmd_parts, stderr=subprocess.STDOUT, env=os.environ
                            )

output = subprocess.check_output(
                            cmd_parts,
                            stderr=subprocess.STDOUT,
                            text=True,
                            env=os.environ,
                        )

result = subprocess.run(
            ["nvidia-smi", "topo", "-m"],
            capture_output=True,
            text=True,
            timeout=5,
        )

self.process = subprocess.Popen(
            cmd,
            env={
                **python_subprocess_env(),
                # Safeguard against creating a SubprocPool in the subprocess.
                "TORCH_WARM_POOL": "0", …

result = subprocess.run(run_cmd, env=env)

subprocess.check_output(
                _compiler_command(cpp_compiler, "--version"),
                stderr=subprocess.DEVNULL,
            )

subprocess.check_output(_compiler_command(cxx, "--version"))

subprocess.check_call(
                [
                    conda,
                    "create",
                    f"--prefix={prefix}",
                    "--channel=conda-forge",
                    "--quiet",
                    "-y" …

subprocess.check_output(
            _compiler_command(compiler, "/help"), stderr=subprocess.STDOUT
        )

vs_install_path = subprocess.check_output(
                cmd, text=True, encoding="utf-8"
            ).strip()

out = subprocess.check_output(
            _compiler_command(compiler, "-v"),
            stderr=subprocess.STDOUT,
            text=True,
        )

_WORKER_SERVER = _WorkerServer("::", worker_port)

with open("/proc/self/maps") as f:

with open(filename, "rb") as in_file:

with open(registry_path) as f:

with open(os.path.join(path, "entry"), "rb") as f:

with open(filepath, "rb") as file:

with open(path, "rb") as f:

content = importlib.resources.read_text(__package__, "schema.yaml")

thrift_content = importlib.resources.read_text(
            # pyrefly: ignore [bad-argument-type]
            __package__,
            "export_schema.thrift",
        )

with open(filename) as f:

with open(input_data_path, "rb") as f:

with open(path) as f:

with open(op_conf) as f:

with open(op_conf, mode) as op_conf_file:

with open(log_path, newline="") as file:

with open(fpath, "rb") as fp:

with open(fpath, "rb") as fp:

with open(local_cache_path) as local_cache_fp:

with open(module, "rb") as f:

with open(path, "rb") as f:

with open(os.path.join(subdir, path), "rb") as f:

with open(
                os.path.join(
                    os.path.dirname(os.path.dirname(__file__)),
                    "csrc",
                    "inductor",
                    "aoti_runtime",
                    "model.h", …

code = Path(script).read_text()

open(resource_path) as resource_file,

with open(error_path, encoding="utf-8") as fh:

with open(
                os.path.join(os.path.dirname(__file__), "aoti_runtime", "interface.cpp")
            ) as f:

with open(target_file, "w") as fd:

with open(minified_repro_path, "w") as fd:

with open(path, "wb") as f:

shutil.rmtree(self.path_prefix(), ignore_errors=True)

with open(tmp_path, "wb") as f:

with open(save_path, "w") as f:

with open(file_name, "w") as fd:

shutil.copyfile(file_name, repro_path)

with open(file_name, "w") as fd:

with open(file_name, "w") as fd:

shutil.copyfile(file_name, latest_repro)

with open(file_name, "w") as fd:

with open(file_name, "w") as fd:

shutil.copyfile(file_name, repro_path)

with open(os.path.join(debug_dir(), name), "w") as fd:

with open(filename, "wb") as f:

with open(os.path.join(root_dir, sys.argv[1] + ".py"), "w") as f:

shutil.rmtree(AOTAutogradCache._get_tmp_dir(), ignore_errors=True)

with open(
            f"{folder_name}/{current_name}/{current_name}_{type_name}_{graph_index}/{current_name}_{type_name}_{graph_index}.input",
            "wb",
        ) as f:

with open(fx_graph_file, "w") as f:

with open(svg_path, "w") as f:

with open(out, "w") as f:

os.unlink(tmp_path)

with open(op_conf, "w") as op_conf_file:

with open(log_path, "a") as f:

fetch("./model_info.json").then(function(response) {

import urllib.request

urllib.request.urlretrieve(
                "https://raw.githubusercontent.com/brendangregg/FlameGraph/master/flamegraph.pl",
                tmp_name,
            )

import requests

resp = requests.post(url, timeout=timeout)

import aiohttp

client_timeout = aiohttp.ClientTimeout(total=timeout)

async with aiohttp.ClientSession(timeout=client_timeout) as session:

import urllib3.exceptions  # type: ignore[import]

import urllib.request

path = urllib.request.url2pathname(full_path)

from urllib.request import Request, urlopen

with urlopen(f"https://github.com/{repo_owner}/{repo_name}/tree/main/"):

with urlopen(url) as r:

response = json.loads(_read_url(Request(url, headers=headers)))

response = json.loads(_read_url(Request(url, headers=headers)))

req = Request(url, headers={"User-Agent": "torch.hub"})

with urlopen(req) as u:

from urllib import request, error

with request.urlopen(url, timeout=15) as f1, open(path, 'wb' if binary else 'w') as f2: