close handles on process exit

This commit is contained in:
payonel 2018-03-11 22:18:49 -07:00
parent 7b966cfdf6
commit a28d16f5d9
7 changed files with 92 additions and 76 deletions

View File

@ -1,6 +1,7 @@
local process = require("process")
local unicode = require("unicode")
local event = require("event")
local thread = require("thread")
local event_mt = getmetatable(event.handlers)
-- WARNING this code does not use official kernel API and is likely to change
@ -42,33 +43,38 @@ local cols =
return count == 0 and "-" or tostring(count)
end},
{"THREADS", function(_,p)
-- threads are handles with mt.close
-- threads are handles with mt.close == thread.waitForAll
local count = 0
for _,h in pairs(p.data.handles or {}) do
for h in pairs(p.data.handles) do
local mt = getmetatable(h)
if mt and mt.__index and mt.__index.close then
count = count + #h
break -- there is only one thread handle manager
if mt and mt.__status then
count = count + 1
end
end
return count == 0 and "-" or tostring(count)
end},
{"PARENT", function(_,p)
for _,process_info in pairs(process.list) do
for _,handle in pairs(process_info.data.handles or {}) do
for handle in pairs(process_info.data.handles) do
local mt = getmetatable(handle)
if mt and mt.__index and mt.__index.close then
for _,ht in ipairs(handle) do
local ht_mt = getmetatable(ht)
if ht_mt.process == p then
return thread_id(nil,process_info)
end
end
break
if mt and mt.__status then
if mt.process == p then
return thread_id(nil, process_info)
end
end
end
return thread_id(nil,p.parent)
end
return thread_id(nil, p.parent)
end},
{"HANDLES", function(_, p)
local count = 0
for stream,closure in pairs(p.data.handles) do
cprint(string.format("%s %s", stream, closure))
if closure then
count = count + 1
end
end
return count == 0 and "-" or tostring(count)
end},
{"CMD", function(_,p) return p.command end},
}

View File

@ -1,4 +1,5 @@
local process = require("process")
local fs = require("filesystem")
--Initialize coroutine library--
local _coroutine = coroutine -- real coroutine backend
@ -62,8 +63,20 @@ process.list[init_thread] = {
data =
{
vars={},
handles={},
io={}, --init will populate this
coroutine_handler = _coroutine
},
instances = setmetatable({}, {__mode="v"})
}
-- intercept fs open
local fs_open = fs.open
fs.open = function(...)
local fs_open_result = table.pack(fs_open(...))
if fs_open_result[1] then
process.closeOnExit(fs_open_result[1])
end
return table.unpack(fs_open_result, 1, fs_open_result.n)
end

View File

@ -17,21 +17,32 @@ function buffer.new(mode, stream)
bufferWrite = "",
bufferSize = math.max(512, math.min(8 * 1024, computer.freeMemory() / 8)),
bufferMode = "full",
readTimeout = math.huge
readTimeout = math.huge,
}
mode = mode or "r"
for i = 1, unicode.len(mode) do
result.mode[unicode.sub(mode, i, i)] = true
end
-- when stream closes, result should close first
-- when result closes, stream should close after
-- when stream closes, it is removed from the proc
stream.close = setmetatable({close = stream.close,parent = result},{__call = buffer.close})
return setmetatable(result, metatable)
end
function buffer:close()
if self.mode.w or self.mode.a then
self:flush()
end
self.closed = true
-- self is either the buffer, or the stream.close callable
local meta = getmetatable(self)
if meta == metatable.__metatable then
return self.stream:close()
end
local parent = self.parent
if parent.mode.w or parent.mode.a then
parent:flush()
end
parent.closed = true
return self.close(parent.stream)
end
function buffer:flush()

View File

@ -78,7 +78,7 @@ end
-- redirects as built by buildCommentRedirects
function sh.internal.openCommandRedirects(redirects)
local data = process.info().data
local ios, handles = data.io, data.handles
local ios = data.io
for _,rjob in ipairs(redirects) do
local from_io, to_io, mode = table.unpack(rjob)
@ -93,7 +93,6 @@ function sh.internal.openCommandRedirects(redirects)
io.stderr:write("could not open '" .. to_io .. "': " .. reason .. "\n")
os.exit(1)
end
table.insert(handles, file)
ios[from_io] = file
end
end

View File

@ -134,16 +134,16 @@ function pipe.buildPipeChain(progs)
-- B needs to be a stack in case any thread in B calls read
pipe.createCoroutineStack(thread)
chain[i] = thread
local data = process.info(thread).data
local pio = data.io
local proc = process.info(thread)
local pio = proc.data.io
local piped_stream
if i < #progs then
local handle = setmetatable({redirect = {rawget(pio, 1)},buffer = ""}, {__index = pipe_stream})
process.closeOnExit(handle, proc)
piped_stream = buffer.new("rw", handle)
piped_stream:setvbuf("no", 1024)
pio[1] = piped_stream
table.insert(data.handles, piped_stream)
end
if prev_piped_stream then
@ -194,7 +194,7 @@ function pipe.popen(prog, mode, env)
local chain = {}
-- to simplify the code - shell.execute is run within a function to pass (prog, env)
-- if cmd_proc where to come second (mode=="w") then the pipe_proc would have to pass
-- if cmd_proc were to come second (mode=="w") then the pipe_proc would have to pass
-- the starting args. which is possible, just more complicated
local cmd_proc = process.load(function() return shell.execute(prog, env) end, nil, nil, prog)

View File

@ -125,7 +125,11 @@ function process.internal.close(thread, result)
checkArg(1,thread,"thread")
local pdata = process.info(thread).data
pdata.result = result
for _,v in pairs(pdata.handles) do
local handles = {}
for s,_ in pairs(pdata.handles) do
table.insert(handles, s)
end
for _,v in ipairs(handles) do
pcall(v.close, v)
end
process.list[thread] = nil
@ -146,6 +150,20 @@ function process.internal.continue(co, ...)
return table.unpack(result, 2, result.n)
end
function process.closeOnExit(stream, proc)
local handles = (proc or process.info()).data.handles
if not handles[stream] then
handles[stream] = stream.close
stream.close = function(...)
local close = handles[stream]
handles[stream] = nil
if close then
return close(...)
end
end
end
end
function process.running(level) -- kept for backwards compat, prefer process.info
local info = process.info(level)
if info then

View File

@ -60,22 +60,6 @@ function thread.waitForAll(threads, timeout)
end
local box_thread = {}
local box_thread_list = {close = thread.waitForAll}
local function get_process_threads(proc, bCreate)
local handles = proc.data.handles
for _,next_handle in ipairs(handles) do
local handle_mt = getmetatable(next_handle)
if handle_mt and handle_mt.__index == box_thread_list then
return next_handle
end
end
if bCreate then
local btm = setmetatable({}, {__index = box_thread_list})
table.insert(handles, btm)
return btm
end
end
function box_thread:resume()
local mt = getmetatable(self)
@ -120,31 +104,25 @@ function box_thread:detach()
end
function box_thread:attach(parent)
checkArg(1, parent, "thread", "number", "nil")
local mt = assert(getmetatable(self), "thread panic: no metadata")
local proc = process.info(parent)
local mt = assert(getmetatable(self), "thread panic: no metadata")
if not proc then return nil, "thread failed to attach, process not found" end
if mt.attached == proc then return self end -- already attached
-- remove from old parent
local waiting_handler
if mt.attached then
local prev_threads = assert(get_process_threads(mt.attached), "thread panic: no thread handle")
for index,t_in_list in ipairs(prev_threads) do
if t_in_list == self then
table.remove(prev_threads, index)
break
end
end
-- registration happens on the attached proc, unregister before reparenting
waiting_handler = mt.unregister()
mt.attached.data.handles[self] = nil
end
-- registration happens on the attached proc, unregister before reparenting
local waiting_handler = mt.unregister()
-- fix close
self.close = self.join
-- attach to parent or the current process
mt.attached = proc
-- this process may not have a box_thread list
local threads = get_process_threads(proc, true)
table.insert(threads, self)
process.closeOnExit(self, proc)
-- register on the new parent
if waiting_handler then -- event-waiting
@ -159,9 +137,9 @@ function thread.current()
local thread_root
while proc do
if thread_root then
for _,bt in ipairs(get_process_threads(proc) or {}) do
if bt.pco.root == thread_root then
return bt
for handle in pairs(proc.data.handles) do
if handle.pco and handle.pco.root == thread_root then
return handle
end
end
else
@ -174,9 +152,8 @@ end
function thread.create(fp, ...)
checkArg(1, fp, "function")
local t = {}
local mt = {__status="suspended",__index=box_thread}
setmetatable(t, mt)
local t = setmetatable({}, mt)
t.pco = pipe.createCoroutineStack(function(...)
mt.__status = "running"
local fp_co = t.pco.create(fp)
@ -246,7 +223,7 @@ function thread.create(fp, ...)
mt.id = nil
mt.reg = nil
-- before just removing a handler, make sure it is still ours
if id and mt.attached and mt.attached.data.handlers[id] == reg then
if id and mt.attached.data.handlers[id] == reg then
mt.attached.data.handlers[id] = nil
return reg
end
@ -285,19 +262,13 @@ function thread.create(fp, ...)
end
function mt.close()
if t:status() == "dead" then
return
end
local threads = get_process_threads(mt.attached)
for index,t_in_list in ipairs(threads) do
if t_in_list == t then
table.remove(threads, index)
break
end
end
local old_status = t:status()
mt.__status = "dead"
mt.attached.data.handles[t] = nil
if old_status ~= "dead" then
event.push("thread_exit")
end
end
t:attach() -- the current process
mt.private_resume(...) -- threads start out running
@ -321,8 +292,6 @@ do
end
assert(init_thread, "thread library panic: no init thread")
handlers_mt.threaded = true
-- handles might be optimized out for memory
root_data.handles = root_data.handles or {}
-- if we don't separate root handlers from thread handlers we see double dispatch
-- because the thread calls dispatch on pull as well
root_data.handlers = {} -- root handlers