1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
|
local log = require "mason-core.log"
local _ = require "mason-core.functional"
local path = require "mason-core.path"
local fs = require "mason-core.fs"
local a = require "mason-core.async"
local Result = require "mason-core.result"
local InstallContext = require "mason-core.installer.context"
local settings = require "mason.settings"
local linker = require "mason-core.installer.linker"
local control = require "mason-core.async.control"
local Semaphore = control.Semaphore
local sem = Semaphore.new(settings.current.max_concurrent_installers)
local M = {}
---@async
local function create_prefix_dirs()
for _, p in ipairs { path.install_prefix(), path.bin_prefix(), path.package_prefix(), path.package_build_prefix() } do
if not fs.async.dir_exists(p) then
fs.async.mkdirp(p)
end
end
end
---@async
---@param context InstallContext
local function write_receipt(context)
log.fmt_debug("Writing receipt for %s", context.package)
context.receipt
:with_name(context.package.name)
:with_schema_version("1.0")
:with_completion_time(vim.loop.gettimeofday())
local receipt_path = path.concat { context.cwd:get(), "mason-receipt.json" }
local install_receipt = context.receipt:build()
fs.async.write_file(receipt_path, vim.json.encode(install_receipt))
end
local CONTEXT_REQUEST = {}
---@return InstallContext
function M.context()
return coroutine.yield(CONTEXT_REQUEST)
end
---@async
---@param context InstallContext
function M.prepare_installer(context)
create_prefix_dirs()
local package_build_prefix = path.package_build_prefix(context.package.name)
if fs.async.dir_exists(package_build_prefix) then
fs.async.rmrf(package_build_prefix)
end
fs.async.mkdirp(package_build_prefix)
context.cwd:set(package_build_prefix)
end
---@async
---@param context InstallContext
---@param installer async fun(context: InstallContext)
function M.run_installer(context, installer)
local thread = coroutine.create(function(...)
-- We wrap the installer with a function to allow it to be a spy instance (in which case it's not a function, but a metatable - coroutine.create expects functions only)
return installer(...)
end)
local step
local ret_val
step = function(...)
local ok, result = coroutine.resume(thread, ...)
if not ok then
error(result, 0)
elseif result == CONTEXT_REQUEST then
step(context)
elseif coroutine.status(thread) == "suspended" then
-- yield to parent coroutine
step(coroutine.yield(result))
else
ret_val = result
end
end
context.receipt:with_start_time(vim.loop.gettimeofday())
M.prepare_installer(context)
step(context)
return ret_val
end
---@async
---@param handle InstallHandle
---@param opts InstallContextOpts
function M.execute(handle, opts)
if handle:is_active() or handle:is_closed() then
log.fmt_debug("Received active or closed handle %s", handle)
return Result.failure "Invalid handle state."
end
handle:queued()
local permit = sem:acquire()
if handle:is_closed() then
permit:forget()
log.fmt_trace("Installation was aborted %s", handle)
return Result.failure "Installation was aborted."
end
log.fmt_trace("Activating handle %s", handle)
handle:active()
local pkg = handle.package
local context = InstallContext.new(handle, opts)
if opts.debug then
local append_log = a.scope(function(chunk)
context.fs:append_file("mason-debug.log", chunk)
end)
handle:on("stdout", append_log)
handle:on("stderr", append_log)
end
log.fmt_info("Executing installer for %s", pkg)
return Result.run_catching(function()
-- 1. run installer
a.wait(function(resolve, reject)
local cancel_thread = a.run(M.run_installer, function(success, result)
if success then
resolve(result)
else
reject(result)
end
end, context, pkg.spec.install)
handle:once("terminate", function()
handle:once("closed", function()
reject "Installation was aborted."
end)
cancel_thread()
end)
end)
-- 2. promote temporary installation dir
context:promote_cwd()
-- 3. link package
linker.link(context)
-- 4. write receipt
write_receipt(context)
end)
:on_success(function()
permit:forget()
handle:close()
log.fmt_info("Installation succeeded for %s", pkg)
end)
:on_failure(function(failure)
permit:forget()
log.fmt_error("Installation failed for %s error=%s", pkg, failure)
context.stdio_sink.stderr(tostring(failure))
context.stdio_sink.stderr "\n"
if not opts.debug then
-- clean up installation dir
pcall(function()
fs.async.rmrf(context.cwd:get())
end)
else
context.stdio_sink.stdout(
("[debug] Installation directory retained at %q.\n"):format(context.cwd:get())
)
end
-- unlink linked executables (in the rare occasion an error occurs after linking)
linker.unlink(context.package, context.receipt.links)
if not handle:is_closed() and not handle.is_terminated then
handle:close()
end
end)
end
---Runs the provided async functions concurrently and returns their result, once all are resolved.
---This is really just a wrapper around a.wait_all() that makes sure to patch the coroutine context before creating the
---new async execution contexts.
---@async
---@param suspend_fns async fun(ctx: InstallContext)[]
function M.run_concurrently(suspend_fns)
local context = M.context()
return a.wait_all(_.map(function(suspend_fn)
return _.partial(M.run_installer, context, suspend_fn)
end, suspend_fns))
end
return M
|
