connection.lua

local _NAME = ...

local obj = require 'obj'
local ffi = require 'ffi'

local log = require 'log'
local fiber = require 'fiber'
local errno = require 'errno'
local socket = require 'socket'

if not pcall(ffi.typeof,"struct iovec") then
	ffi.cdef[[
		struct iovec {
			void  *iov_base;
			size_t iov_len;
		};
	]]
end

ffi.cdef [[
	char *strerror(int errnum);
	ssize_t read(int fd, void *buf, size_t count);
	void *memcpy(void *dest, const void *src, size_t n);
	void *memmove(void *dest, const void *src, size_t n);
	ssize_t writev(int fd, const struct iovec *iov, int iovcnt);
	void *calloc(size_t nmemb, size_t size);
	void free(void *ptr);
]]

local C = ffi.C

local iovec = ffi.typeof('struct iovec') -- FIXME it is used once!
local IOVSZ = assert(ffi.sizeof(iovec))

local NOTCONNECTED = 0;
local CONNECTING   = 1;
local CONNECTED    = 2;
local RECONNECTING = 3;

---@class connection
---@field host string remote host
---@field port number remote port
---@field timeout number connection and request timeout
---@field private __id number identifier of object
---@field private _auto boolean
---@field private _reconnect number|false?
---@field private _gen number generation of connection
---@field state S2S connection state
---@field maxbuf number read buffer size
---@field rbuf ffi.cdata* read buffer
---@field avail number actual data length in buffer
---@field wsize number write buffer size
---@field wbuf ffi.cdata* write buffer
---@field wcur number current write buffer position
---@field wstash table stash for write buffer
---@field _flush fiber.channel flush channel
---@field connwait fiber.channel connection wait channel
---@field private ww fiber? writer worker fiber
---@field private rw fiber? read worker
local M = obj.class({},_NAME)

---@alias S2S
---| 0 NOTCONNECTED
---| 1 CONNECTING
---| 2 CONNECTED
---| 3 RECONNECTING
local S2S = {
	[NOTCONNECTED]  = 'NOTCONNECTED',
	[CONNECTING]    = 'CONNECTING',
	[CONNECTED]     = 'CONNECTED',
	[RECONNECTING]  = 'RECONNECTING',
	NOTCONNECTED = NOTCONNECTED,
	CONNECTED = CONNECTED,
	CONNECTING = CONNECTING,
	RECONNECTING = RECONNECTING,
}
M.S2S = S2S

local errno_is_transient = {
	[errno.EINPROGRESS] = true;
	[errno.EAGAIN] = true;
	[errno.EWOULDBLOCK] = true;
	[errno.EINTR] = true;
}

---@class connection.options
---@field timeout number? connection and request timeout
---@field autoconnect boolean? should we connect on init?
---@field reconnect number|false? reconnect timeout, if false then no reconnect
---@field maxbuf number? read buffer size, default is 2*1024*1024

--[[
options:
	timeout -
	autoconnect -
	reconnect -
	maxbuf -

internal fields:
	host - host connect to
	port - port connect to
	_reconnect - should we reconnect on connection reset?
	maxbuf - read buffer size. Although it is named 'maxbuf', buffer is not rellocatable.
	rbuf - read buffer
	avail - Actual data length in buffer
	state - connection stat, valid values are:
		NOTCONNECTED
		CONNECTED
		CONNECTING
		RECONNECTING
	_flush
	_auto
	_gen

]]
---@param host string remote host
---@param port number remote port
---@param opt connection.options?
function M:_init(host, port, opt)
	self.host = host;
	self.port = tonumber(port)
	opt = opt or {}

	self.timeout = tonumber(opt.timeout) or 1/3

	if opt.reconnect ~= nil then
		if opt.reconnect then
			self._reconnect = tonumber(opt.reconnect)
		else
			self._reconnect = false
		end
	else
		self._reconnect = 1/3
	end

	if opt.autoconnect ~= nil then
		self._auto = opt.autoconnect
	else
		self._auto = true
	end

	self.state = NOTCONNECTED
	self.maxbuf = opt.maxbuf or 2*1024*1024
	self.rbuf = ffi.cast('char *', ffi.C.calloc(1, self.maxbuf))

	do
		local firstcall = true
		ffi.gc(self.rbuf, function(ptr)
			if not firstcall then
				print("Called double __gc on:", tostring(ptr))
				return
			end
			firstcall = false
			ffi.C.free(ptr)
		end)
	end
	self.avail = 0ULL
	self._gen = 0

	self.wsize = 32
	self.wbuf = ffi.new('struct iovec[?]', self.wsize)

	self.wcur = 0
	self.wstash = {}

	self._flush = fiber.channel(0)

	self.connwait = fiber.channel(0)

	if self._auto then
		self:connect()
	end
	--[[ -- FIXME
	if opt.deadly or opt.deadly == nil then
		box.reload:register(self)
	end
	]]
end

---Logs a message on the connection.
---@param l string log level
---@param msg string message to log
function M:log(l,msg,...)
	msg = tostring(msg)
	-- FIXME it's a bad pattern. We should always use format
	if string.match(msg,'%%') then
		msg = string.format(msg,...)
	else
		local m = { msg }
		for _,v in pairs({...}) do
			table.insert(m, tostring(v))
		end
		msg = table.concat(m,' ')
	end
	log.info( "[%s] {%s:%s} %s", l, self.host, self.port, msg )
end

function M:fdno()
	if self.s then
		return self.s:fd()
	else
		return -1
	end
end

---Returns a string representation of the connection.
---@return string
function M:_stringify()
	return string.format("cnn(%s:%s : %s:%s : %s)",self:fdno(),self.state,self.host,self.port,self.__id)
end

---Describes current connection.
---@return string
function M:desc()
	return tostring(self.host) .. ':' .. tostring(self.port) .. '/' .. self:fdno()
end

---Default callback for connection established.
function M:on_connected()
	self:log("D", "called default on_connected")
end

---Default callback for connection failed.
---@param e string|box.error error message
function M:on_disconnect(e)
	self:log("D", "called default on_disconnect: %s", e)
end

---Cleans up the connection.
---@param e integer errno code that triggered action
function M:_cleanup(e)
	self.state = NOTCONNECTED
	if self.ww then if self.ww ~= fiber.self() then pcall(fiber.cancel,self.ww) end self.ww = nil end
	if self.rw then if self.rw ~= fiber.self() then pcall(fiber.cancel,self.rw) end self.rw = nil end
	if self.s  then self.s:close() self.s = nil end

	self.wcur   = 0
	self.wstash = {}
	self.avail  = 0ULL

	self.lasterror = errno.strerror(e)

	while self.connwait:put(false, 0) do end
end

function M:destroy()
	self:_cleanup(0)
	local name = self.host..':'..self.port
	for k in pairs(self) do
		self[k] = nil
	end
	setmetatable(self,{
		__index = function(s,n)
			log.error("access to `"..n.."' on destroyed con "..name)
			fiber.cancel(fiber.self())
		end,
		__newindex = function(s,n)
			log.error("access to `"..n.."' on destroyed con"..name)
			fiber.cancel(fiber.self())
		end
	})
end

function M:close()
	self:_cleanup(0)
	self:log('I', self.host..':'..self.port..' closed')
end

function M:on_connect_failed(e)
	self:log('E','Connect failed:', errno.strerror(e))
	if self.state == CONNECTING then
		self:_cleanup(e)
		if self.state == NOTCONNECTED then
			if self._reconnect then
				self.state = RECONNECTING
				if self.on_connfail then
					fiber.create(function(self) fiber.name("net.cb") self:on_connfail(errno.strerror(e)) end,self)
				end
				fiber.sleep(self._reconnect)
				if self.state == RECONNECTING then
					self.state = NOTCONNECTED
					self:connect()
				end
			else
				-- self.state = NOTCONNECTED -- already
				fiber.create(function(self) fiber.name("net.cb") self:on_disconnect(errno.strerror(e)) end,self)
			end
		end
	end
end

---Callback is called when connection is about to reset
---@param e integer errno code
function M:on_connect_reset(e)
	self:log('W',"connection reset:",errno.strerror(e))
	if self.state == CONNECTED then
		-- TODO: stop all fibers
		self:_cleanup(0)

		if self._reconnect then
			self.state = NOTCONNECTED -- was RECONNECTING
			fiber.create(function(self) fiber.name("net.cb") self:on_disconnect(errno.strerror(e)) end,self)
			fiber.sleep(0)
			self:connect()
		else
			-- self.state = NOTCONNECTED -- already
			fiber.create(function(self) fiber.name("net.cb") self:on_disconnect(errno.strerror(e)) end,self)
		end
	end
end

---Callback is called when data is available for reading.
---
---if method raises an error, then connection is reset.
---@param is_last boolean true if this is the last read, false if more data is expected
function M:on_read(is_last)
	self:log('D',"on_read (last:",is_last,") ",ffi.string(self.rbuf,self.ravail))
	self.avail = 0ULL
end

---Callback is called when socket has been connected.
---
---In this method we create two fibers: one for reading and one for writing.
function M:on_connect_io()
	local err = self.s:getsockopt('SOL_SOCKET', 'SO_ERROR');
	if err ~= 0 then
		-- OLD TODO: error handling
		self:on_connect_failed( err )
		return
	end
	self.state = CONNECTED;

	local weak = setmetatable({}, { __mode = "kv" })
	weak.self = self
	--print('----', weak.self.s)

	self.ww = fiber.create(function (weak, gen)
		fiber.name(string.format("net.ww[%s:%s#%d]", weak.self.host, weak.self.port, gen))
		local s = weak.self.s
		while weak.self and gen == weak.self._gen do
			if s:writable(1) then
				if not weak.self then break end
				if weak.self:_writev() then
					-- The write buffer is drained
					weak.self._flush:get(1)
				end
			end
		end
	end, weak, self._gen)

	self.rw = fiber.create(function (weak, gen)
		fiber.name(string.format("net.rw[%s:%s#%d]", weak.self.host, weak.self.port, gen))
		local s = weak.self.s
		local fd = s:fd()
		local oft = 0ULL
		local sz  = weak.self.maxbuf
		while weak.self and gen == self._gen do
			local self = weak.self
			local rd = C.read(fd, self.rbuf + oft, sz - oft)
			-- local rd = C.read(s.socket.fd, self.rbuf + oft, 1)
			if rd >= 0 then
				self.avail = self.avail + rd;
				local avail = self.avail

				local status, err = pcall(self.on_read, self, rd == 0)
				if not status then
					self:log('E', 'on_read raised an error: ', err)
					self:on_connect_reset(errno.EINVAL) -- errno.EINVAL = 22
				end

				local pkoft = avail - self.avail
				-- print("avail ",avail, " -> ", self.avail, " pkoft = ", pkoft)

				-- FIXME: Is it a good idea?
				if self.avail > 0 then
					if self.avail == self.maxbuf then
						self:on_connect_reset(errno.EMSGSIZE)
						return
					end
					oft = self.avail
					-- print("avail ",avail, " -> ", self.avail, " pkoft = ", pkoft)
					C.memmove(self.rbuf,self.rbuf + pkoft,oft)
				else
					if rd == 0 then
						self:on_connect_reset(errno.ECONNABORTED)
						return
					end
					oft = 0
				end
			elseif errno_is_transient[errno()] then
				s:readable()
			else
				-- print( errno.strerror( errno() ))
				self:on_connect_reset(s:errno())
				return
			end
		end
	end, weak, self._gen)

	while self.connwait:put(true, 0) do end
	fiber.create(function(self) fiber.name("net.cb") self:on_connected() end,self)
end

function M:wait_con(timeout)
	if self.state == CONNECTED then
		return true
	end
	-- FIXME move define default timeout in the start of the file
	if self.connwait:get(timeout or self.timeout or 10) then
		return
	else
		-- FIXME Should we use to kinds of error here? There are two cases: it
		-- can be a connection timeout or not called connect method.
		error('Connection timeoud')
	end
end

---Connects to the remote host and port.
---@return boolean true if connected, false if not
function M:connect()
	assert(type(self) == 'table',"object required")

	if self.state ~= NOTCONNECTED then
		return (self.state == CONNECTED)
	end

	-- connect timeout
	assert(not self.s, "Already have socket")

	self.state = CONNECTING
	self._gen = self._gen + 1

	local weak = setmetatable({}, { __mode = "kv" })
	weak.self = self

	fiber.create(function(weak)
		-- We don't need to check self because fiber is runned without yielding
		local ai = socket.getaddrinfo( weak.self.host, weak.self.port, weak.self.timeout, {
			['type'] = 'SOCK_STREAM',
		} )

		-- But after getaddrinfo we do need to check the link
		if not weak.self then return end

		if ai and #ai > 0 then
			--print(dumper(ai))
		else
			weak.self:on_connect_failed( errno() == 0 and errno.ENXIO or errno() )
			return
		end

		local ainfo = ai[1]
		local s = socket( ainfo.family, ainfo.type, ainfo.protocol )
		if not s then
			weak.self:on_connect_failed( errno() )
			return
		end

		s:nonblock(true)
		s:linger(1,0)

		while true do
			-- FIXME sysconnect should be not yielding, but we have to dowble check
			-- for traps from tnt team
			if s:sysconnect( ainfo.host, ainfo.port ) then
				weak.self.s = s
				--print("immediate connected")
				weak.self:on_connect_io()
				return
			else
				if s:errno() == errno.EINPROGRESS
				or s:errno() == errno.EALREADY
				or s:errno() == errno.EWOULDBLOCK
				then
					weak.self.s = s

					local wr = s:writable(weak.self.timeout)

					if not weak.self then s:close() return end

					if wr then
						weak.self:on_connect_io()
					else
						weak.self:on_connect_failed( errno.ETIMEDOUT )
					end

					return
				elseif s:errno() == errno.EINTR then
					-- again
					if not weak.self then s:close() return end
				else
					weak.self:on_connect_failed( s:errno() )
					return
				end
			end
		end

	end, weak)
end

function M:_wbuf_realloc()
	local old = self.wbuf
	local osz = self.wsize
	self.wsize = osz * 2
	self.wbuf = ffi.new('struct iovec[?]', self.wsize)
	C.memcpy(self.wbuf, old, self.wcur * ffi.sizeof(self.wbuf[0]))
end

---Pushes data to the write buffer.
---
---Does not write into socket, you must call :flush() to write data.
---@param buf string|ffi.cdata* data to write
---@param len number? length of data to write, if nil then uses #buf
function M:push_write( buf, len )
	if self.state ~= CONNECTED then
		error("Not connected")
	end
	if self.wcur == self.wsize - 1 then
		self:_wbuf_realloc()
	end

	local ffibuf
	if type(buf) == 'cdata' then
		ffibuf = ffi.cast('char *',buf)
	else
		ffibuf = ffi.cast('char *',buf)
	end

	-- print("push_write ",#buf, "; wcur = ",self.wcur, "; wstash = ", #self.wstash, "; buf = ",ffibuf)

	self.wbuf[self.wcur].iov_base = ffibuf
	self.wbuf[self.wcur].iov_len = len or #buf
	table.insert(self.wstash,buf)
	self.wcur = self.wcur + 1
end

---Writes data to the socket
---@return boolean is_drained true if write buffer was left empty, false if not
function M:_writev()
	if self.wcur == 0 then
		return true
	end
	--- should return true if no more tries should be done
	local wr = C.writev(self.s:fd(), self.wbuf, self.wcur)

	if wr > 0 then
		local len = 0
		for i = 0,self.wcur-1 do
			len = len + self.wbuf[i].iov_len
			local cptr = table.remove(self.wstash,1)
			if len == wr then
				if i == self.wcur - 1 then
					self.wcur = 0
					return true
				else
					self.wcur = self.wcur - (i+1)
					-- print("(1) new wcur = ",self.wcur, ' new wstash = ', #self.wstash)
					C.memmove( self.wbuf, self.wbuf[i+1], self.wcur * IOVSZ )
					return false
				end
			elseif len > wr then
				-- print("len ",len," > ", wr, " wcur = ", self.wcur, " i= ",i)
				local left = len - wr
				local offset = self.wbuf[i].iov_len - left

				table.insert(self.wstash,1,cptr)

				self.wcur = self.wcur - i -- wcur - (i+1) + 1
				-- print("(2) new wcur = ",self.wcur, ' new wstash = ', #self.wstash)
				C.memmove( self.wbuf, self.wbuf[i], self.wcur * IOVSZ )

				self.wbuf[0].iov_base = ffi.cast('char *',ffi.cast('char *',self.wbuf[0].iov_base) + offset)
				self.wbuf[0].iov_len = left
				break -- for
			end
		end
	elseif errno_is_transient[ ffi.errno() ] then
		-- print(errno.strerror( ffi.errno() ))
		-- iowait
	else
		print(errno.strerror( ffi.errno() ))
		self:on_connect_reset( ffi.errno() )
		return true
	end
	-- iowait ?
	return false

end

---Flushes the write buffer to the socket.
function M:flush()
	assert(type(self) == 'table', "object required")

	if self.state ~= CONNECTED then
		error("Not connected")
	end

	self._flush:put(true, 0)
end

return M