Streaming I/O and support of named pipes (#570)

* added streaming i/o for scripting, and a byteutil.get_size function

* added i/o stream class, also added named pipes support on lua side via ffi

* added constant file.named_pipes_prefix

* added buffered and yield modes for io_stream

* added new time function for work with UTC - utc_time, utc_offset, local_time

* docs updated

* constant pid moved to os.pid

* now gmtime_s and localtime_s used only in windows
This commit is contained in:
Onran
2025-08-01 20:26:43 +03:00
committed by GitHub
parent cd2bc8fbf6
commit aae642a13e
16 changed files with 1303 additions and 1 deletions
@@ -0,0 +1,17 @@
local io_stream = require "core:io_stream"
local lib = {
read = file.__read_descriptor,
write = file.__write_descriptor,
flush = file.__flush_descriptor,
is_alive = file.__has_descriptor,
close = file.__close_descriptor
}
return function(path, mode)
return io_stream.new(
file.__open_descriptor(path, mode),
mode:find('b') ~= nil,
lib
)
end
@@ -0,0 +1,7 @@
local FFI = ffi
if FFI.os == "Windows" then
return require "core:internal/stream_providers/named_pipe_windows"
else
return require "core:internal/stream_providers/named_pipe_unix"
end
@@ -0,0 +1,21 @@
local forbiddenPaths = {
"/..\\", "\\../",
"/../", "\\..\\"
}
return function(path)
local corrected = true
if path:starts_with("../") or path:starts_with("..\\") then
corrected = false
else
for _, forbiddenPath in ipairs(forbiddenPaths) do
if path:find(forbiddenPath) then
corrected = false
break
end
end
end
if not corrected then error "special path \"../\" is not allowed in path to named pipe" end
end
@@ -0,0 +1,104 @@
local path_validate = require "core:internal/stream_providers/named_pipe_path_validate"
local io_stream = require "core:io_stream"
local FFI = ffi
FFI.cdef[[
int open(const char *pathname, int flags);
int close(int fd);
ssize_t read(int fd, void *buf, size_t count);
ssize_t write(int fd, const void *buf, size_t count);
int fcntl(int fd, int cmd, ...);
const char *strerror(int errnum);
]]
local C = FFI.C
local O_RDONLY = 0x0
local O_WRONLY = 0x1
local O_RDWR = 0x2
local O_NONBLOCK = 0x800
local F_GETFL = 3
local function getError()
local err = ffi.errno()
return ffi.string(C.strerror(err)).." ("..err..")"
end
local lib = {}
function lib.read(fd, len)
local buffer = FFI.new("uint8_t[?]", len)
local result = C.read(fd, buffer, len)
local out = Bytearray()
if result <= 0 then
return out
end
for i = 0, result - 1 do
out[i+1] = buffer[i]
end
return out
end
function lib.write(fd, bytearray)
local len = #bytearray
local buffer = FFI.new("uint8_t[?]", len)
for i = 1, len do
buffer[i-1] = bytearray[i]
end
if C.write(fd, buffer, len) == -1 then
error("failed to write to named pipe: "..getError())
end
end
function lib.flush(fd)
-- no flush on unix
end
function lib.is_alive(fd)
if fd == nil or fd < 0 then return false end
return C.fcntl(fd, F_GETFL) ~= -1
end
function lib.close(fd)
C.close(fd)
end
return function(path, mode)
path_validate(path)
path = "/tmp/"..path
local read = mode:find('r') ~= nil
local write = mode:find('w') ~= nil
local flags
if read and write then
flags = O_RDWR
elseif read then
flags = O_RDONLY
elseif write then
flags = O_WRONLY
else
error "mode must contain read or write flag"
end
flags = bit.bor(flags, O_NONBLOCK)
local fd = C.open(path, flags)
if fd == -1 then
error("failed to open named pipe: "..getError())
end
return io_stream.new(fd, mode:find('b') ~= nil, lib)
end
@@ -0,0 +1,144 @@
local path_validate = require "core:internal/stream_providers/named_pipe_path_validate"
local io_stream = require "core:io_stream"
local FFI = ffi
FFI.cdef[[
typedef void* HANDLE;
typedef uint32_t DWORD;
typedef int BOOL;
typedef void* LPVOID;
typedef const char* LPCSTR;
BOOL CloseHandle(HANDLE hObject);
DWORD GetFileType(HANDLE hFile);
BOOL ReadFile(HANDLE hFile, void* lpBuffer, DWORD nNumberOfBytesToRead,
DWORD* lpNumberOfBytesRead, void* lpOverlapped);
BOOL WriteFile(HANDLE hFile, const void* lpBuffer, DWORD nNumberOfBytesToWrite,
DWORD* lpNumberOfBytesWritten, void* lpOverlapped);
HANDLE CreateFileA(LPCSTR lpFileName, DWORD dwDesiredAccess, DWORD dwShareMode,
void* lpSecurityAttributes, DWORD dwCreationDisposition,
DWORD dwFlagsAndAttributes, HANDLE hTemplateFile);
BOOL PeekNamedPipe(
HANDLE hNamedPipe,
LPVOID lpBuffer,
DWORD nBufferSize,
DWORD* lpBytesRead,
DWORD* lpTotalBytesAvail,
DWORD* lpBytesLeftThisMessage
);
DWORD GetLastError(void);
BOOL FlushFileBuffers(HANDLE hFile);
]]
local C = FFI.C
local GENERIC_READ = 0x80000000
local GENERIC_WRITE = 0x40000000
local OPEN_EXISTING = 3
local FILE_ATTRIBUTE_NORMAL = 0x00000080
local FILE_TYPE_UNKNOWN = 0x0000
local INVALID_HANDLE_VALUE = FFI.cast("HANDLE", -1)
local lib = {}
local function is_data_available(handle)
local bytes_available = FFI.new("DWORD[1]")
local success = FFI.C.PeekNamedPipe(handle, nil, 0, nil, bytes_available, nil)
if success == 0 then
return -1
end
return bytes_available[0] > 0
end
function lib.read(handle, len)
local out = Bytearray()
local has_data, err = is_data_available(handle)
if not has_data then
return out
elseif hasData == -1 then
error("failed to read from named pipe: "..tostring(C.GetLastError()))
end
local buffer = FFI.new("uint8_t[?]", len)
local read = FFI.new("DWORD[1]")
local ok = C.ReadFile(handle, buffer, len, read, nil)
if ok == 0 or read[0] == 0 then
return out
end
for i = 0, read[0] - 1 do
out[i+1] = buffer[i]
end
return out
end
function lib.write(handle, bytearray)
local len = #bytearray
local buffer = FFI.new("uint8_t[?]", len)
for i = 1, len do
buffer[i-1] = bytearray[i]
end
local written = FFI.new("DWORD[1]")
if C.WriteFile(handle, buffer, len, written, nil) == 0 then
error("failed to write to named pipe: "..tostring(C.GetLastError()))
end
end
function lib.flush(handle)
C.FlushFileBuffers(handle)
end
function lib.is_alive(handle)
if handle == nil or handle == INVALID_HANDLE_VALUE then
return false
else
return C.GetFileType(handle) ~= FILE_TYPE_UNKNOWN
end
end
function lib.close(handle)
C.CloseHandle(handle)
end
return function(path, mode)
path_validate(path)
path = "\\\\.\\pipe\\"..path
local read = mode:find('r') ~= nil
local write = mode:find('w') ~= nil
local flags
if read and write then
flags = bit.bor(GENERIC_READ, GENERIC_WRITE)
elseif read then
flags = GENERIC_READ
elseif write then
flags = GENERIC_WRITE
else
error("mode must contain read or write flag")
end
local handle = C.CreateFileA(path, flags, 0, nil, OPEN_EXISTING,
FILE_ATTRIBUTE_NORMAL, nil)
if handle == INVALID_HANDLE_VALUE then
error("failed to open named pipe: "..tostring(C.GetLastError()))
end
return io_stream.new(handle, mode:find('b') ~= nil, lib)
end
+398
View File
@@ -0,0 +1,398 @@
local io_stream = { }
io_stream.__index = io_stream
local MAX_BUFFER_SIZE = 8192
local DEFAULT_MODE = "default"
local BUFFERED_MODE = "buffered"
local YIELD_MODE = "yield"
local ALL_MODES = {
DEFAULT_MODE,
BUFFERED_MODE,
YIELD_MODE
}
local FLUSH_MODE_ALL = "all"
local FLUSH_MODE_ONLY_BUFFER = "buffer"
local ALL_FLUSH_MODES = {
FLUSH_MODE_ALL,
FLUSH_MODE_ONLY_BUFFER
}
local CR = string.byte('\r')
local LF = string.byte('\n')
local function readFully(result, readFunc)
local isTable = type(result) == "table"
local buf
repeat
buf = readFunc(MAX_BUFFER_SIZE)
if isTable then
for i = 1, #buf do
result[#result + 1] = buf[i]
end
else result:append(buf) end
until #buf == 0
end
--[[
descriptor - descriptor of stream for provided I/O library
binaryMode - if enabled, most methods will expect bytes instead of strings
ioLib - I/O library. Should include the following functions:
read(descriptor: int, length: int) -> Bytearray
May return bytearray with a smaller size if bytes have not arrived yet or have run out
write(descriptor: int, data: Bytearray)
flush(descriptor: int)
is_alive(descriptor: int) -> bool
close(descriptor: int)
--]]
function io_stream.new(descriptor, binaryMode, ioLib, mode, flushMode)
mode = mode or DEFAULT_MODE
flushMode = flushMode or FLUSH_MODE_ALL
local self = setmetatable({}, io_stream)
self.descriptor = descriptor
self.binaryMode = binaryMode
self.maxBufferSize = MAX_BUFFER_SIZE
self.ioLib = ioLib
self:set_mode(mode)
self:set_flush_mode(flushMode)
return self
end
function io_stream:is_binary_mode()
return self.binaryMode
end
function io_stream:set_binary_mode(binaryMode)
self.binaryMode = binaryMode ~= nil
end
function io_stream:get_mode()
return self.mode
end
function io_stream:set_mode(mode)
if not table.has(ALL_MODES, mode) then
error("invalid stream mode: "..mode)
end
if self.mode == BUFFERED_MODE then
self.writeBuffer:clear()
self.readBuffer:clear()
end
if mode == BUFFERED_MODE and not self.writeBuffer then
self.writeBuffer = Bytearray()
self.readBuffer = Bytearray()
end
self.mode = mode
end
function io_stream:get_flush_mode()
return self.flushMode
end
function io_stream:set_flush_mode(flushMode)
if not table.has(ALL_FLUSH_MODES, flushMode) then
error("invalid flush mode: "..flushMode)
end
self.flushMode = flushMode
end
function io_stream:get_max_buffer_size()
return self.maxBufferSize
end
function io_stream:set_max_buffer_size(maxBufferSize)
self.maxBufferSize = maxBufferSize
end
function io_stream:available(length)
if self.mode == BUFFERED_MODE then
self:__update_read_buffer()
if not length then
return #self.readBuffer
else
return #self.readBuffer >= length
end
end
end
function io_stream:__update_read_buffer()
local readed = Bytearray()
readFully(readed, function(length) return self.ioLib.read(self.descriptor, length) end)
self.readBuffer:append(readed)
if #self.readBuffer > self.maxBufferSize then
error "buffer overflow"
end
end
function io_stream:__read(length)
if self.mode == YIELD_MODE then
local buffer = Bytearray()
while #buffer < length do
buffer:append(self.ioLib.read(self.descriptor, length - #buffer))
if #buffer < length then coroutine.yield() end
end
return buffer
elseif self.mode == BUFFERED_MODE then
self:__update_read_buffer()
if #self.readBuffer < length then
error "buffer underflow"
end
local copy
if #self.readBuffer == length then
copy = Bytearray()
copy:append(self.readBuffer)
self.readBuffer:clear()
else
copy = Bytearray()
for i = 1, length do
copy[i] = self.readBuffer[i]
end
self.readBuffer:remove(1, length)
end
return copy
elseif self.mode == DEFAULT_MODE then
return self.ioLib.read(self.descriptor, length)
end
end
function io_stream:__write(data)
if self.mode == BUFFERED_MODE then
self.writeBuffer:append(data)
if #self.writeBuffer > self.maxBufferSize then
error "buffer overflow"
end
elseif self.mode == DEFAULT_MODE or self.mode == YIELD_MODE then
return self.ioLib.write(self.descriptor, data)
end
end
function io_stream:read_fully(useTable)
if self.binaryMode then
local result = useTable and Bytearray() or { }
readFully(result, function() return self:__read(self.maxBufferSize) end)
else
if useTable then
local lines = { }
local line
repeat
line = self:read_line()
lines[#lines + 1] = line
until not line
return lines
else
local result = Bytearray()
readFully(result, function() return self:__read(self.maxBufferSize) end)
return utf8.tostring(result)
end
end
end
function io_stream:read_line()
local result = Bytearray()
local first = true
while true do
local char = self:__read(1)
if #char == 0 then
if first then return else break end
end
char = char[1]
if char == LF then break
elseif char == CR then
char = self:__read(1)
if char[1] == LF then break
else
result:append(CR)
result:append(char[1])
end
else result:append(char) end
first = false
end
return utf8.tostring(result)
end
function io_stream:write_line(str)
self:__write(utf8.tobytes(str .. LF))
end
function io_stream:read(arg, useTable)
local argType = type(arg)
if self.binaryMode then
local byteArr
if argType == "number" then
-- using 'arg' as length
byteArr = self:__read(arg)
if useTable == true then
local t = { }
for i = 1, #byteArr do
t[i] = byteArr[i]
end
return t
else
return byteArr
end
elseif argType == "string" then
return byteutil.unpack(
arg,
self:__read(byteutil.get_size(arg))
)
elseif argType == nil then
error(
"in binary mode the first argument must be a string data format"..
" for the library \"byteutil\" or the number of bytes to read"
)
else
error("unknown argument type: "..argType)
end
else
if not arg then
return self:read_line()
else
local linesCount = arg
local trimLastEmptyLines = useTable or true
if linesCount < 0 then error "count of lines to read must be positive" end
local result = { }
for i = 1, linesCount do
result[i] = self:read_line()
end
if trimLastEmptyLines then
local i = #result
while i >= 0 do
local length = utf8.length(result[i])
if length > 0 then break
else result[i] = nil end
i = i - 1
end
local i = 1
while #result > 0 do
local length = utf8.length(result[i])
if length > 0 then break
else table.remove(result, i) end
end
end
return result
end
end
end
function io_stream:write(arg, ...)
local argType = type(arg)
if self.binaryMode then
local byteArr
if argType ~= "string" then
-- using arg as bytes table/bytearray
if argType == "table" then
byteArr = Bytearray(arg)
else
byteArr = arg
end
else
byteArr = byteutil.pack(arg, ...)
end
self:__write(byteArr)
else
if argType == "string" then
self:write_line(arg)
elseif argType == "table" then
for i = 1, #arg do
self:write_line(arg[i])
end
else error("unknown argument type: "..argType) end
end
end
function io_stream:is_alive()
return self.ioLib.is_alive(self.descriptor)
end
function io_stream:is_closed()
return not self:is_alive()
end
function io_stream:close()
if self.mode == BUFFERED_MODE then
self.readBuffer:clear()
self.writeBuffer:clear()
end
return self.ioLib.close(self.descriptor)
end
function io_stream:flush()
if self.mode == BUFFERED_MODE and #self.writeBuffer > 0 then
self.ioLib.write(self.descriptor, self.writeBuffer)
self.writeBuffer:clear()
end
if self.flushMode ~= FLUSH_MODE_ONLY_BUFFER then self.ioLib.flush(self.descriptor) end
end
return io_stream
+22 -1
View File
@@ -317,10 +317,30 @@ entities.get_all = function(uids)
return stdcomp.get_all(uids)
end
end
local bytearray = require "core:internal/bytearray"
Bytearray = bytearray.FFIBytearray
Bytearray_as_string = bytearray.FFIBytearray_as_string
Bytearray_construct = function(...) return Bytearray(...) end
file.open = require "core:internal/stream_providers/file"
file.open_named_pipe = require "core:internal/stream_providers/named_pipe"
if ffi.os == "Windows" then
ffi.cdef[[
unsigned long GetCurrentProcessId();
]]
os.pid = ffi.C.GetCurrentProcessId()
else
ffi.cdef[[
int getpid(void);
]]
os.pid = ffi.C.getpid()
end
ffi = nil
math.randomseed(time.uptime() * 1536227939)
@@ -473,6 +493,7 @@ function __vc_on_world_quit()
_rules.clear()
gui_util:__reset_local()
stdcomp.__reset()
file.__close_all_descriptors()
end
local __vc_coroutines = {}
@@ -629,4 +650,4 @@ function dofile(path)
end
end
return _dofile(path)
end
end