mirror of
https://github.com/cloudwu/skynet.git
synced 2026-07-24 20:23:06 +00:00
Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1a169ec413 | ||
|
|
009e15556d | ||
|
|
1170acc792 | ||
|
|
37cb126812 | ||
|
|
aa65dac9ed | ||
|
|
672e218311 | ||
|
|
f7783d3881 | ||
|
|
b65757d8bf |
11
Makefile
11
Makefile
@@ -5,7 +5,7 @@ CSERVICE_PATH ?= cservice
|
|||||||
|
|
||||||
SKYNET_BUILD_PATH ?= .
|
SKYNET_BUILD_PATH ?= .
|
||||||
|
|
||||||
CFLAGS = -g -O2 -Wall -I$(LUA_INC) $(MYCFLAGS)
|
CFLAGS = -g -O2 -Wall -I$(LUA_INC) $(MYCFLAGS) $(SKYNET_DEFINES)
|
||||||
|
|
||||||
# lua
|
# lua
|
||||||
|
|
||||||
@@ -22,8 +22,11 @@ JEMALLOC_STATICLIB := 3rd/jemalloc/lib/libjemalloc_pic.a
|
|||||||
JEMALLOC_INC := 3rd/jemalloc/include/jemalloc
|
JEMALLOC_INC := 3rd/jemalloc/include/jemalloc
|
||||||
|
|
||||||
all : jemalloc
|
all : jemalloc
|
||||||
|
|
||||||
.PHONY : jemalloc
|
.PHONY : jemalloc
|
||||||
|
|
||||||
|
MALLOC_STATICLIB := $(JEMALLOC_STATICLIB)
|
||||||
|
|
||||||
$(JEMALLOC_STATICLIB) : 3rd/jemalloc/Makefile
|
$(JEMALLOC_STATICLIB) : 3rd/jemalloc/Makefile
|
||||||
cd 3rd/jemalloc && $(MAKE) CC=$(CC)
|
cd 3rd/jemalloc && $(MAKE) CC=$(CC)
|
||||||
|
|
||||||
@@ -33,7 +36,7 @@ $(JEMALLOC_STATICLIB) : 3rd/jemalloc/Makefile
|
|||||||
3rd/jemalloc/Makefile : | 3rd/jemalloc/autogen.sh
|
3rd/jemalloc/Makefile : | 3rd/jemalloc/autogen.sh
|
||||||
cd 3rd/jemalloc && ./autogen.sh --with-jemalloc-prefix=je_ --disable-valgrind
|
cd 3rd/jemalloc && ./autogen.sh --with-jemalloc-prefix=je_ --disable-valgrind
|
||||||
|
|
||||||
jemalloc : $(JEMALLOC_STATICLIB)
|
jemalloc : $(MALLOC_STATICLIB)
|
||||||
|
|
||||||
# skynet
|
# skynet
|
||||||
|
|
||||||
@@ -50,8 +53,8 @@ all : \
|
|||||||
$(foreach v, $(CSERVICE), $(CSERVICE_PATH)/$(v).so) \
|
$(foreach v, $(CSERVICE), $(CSERVICE_PATH)/$(v).so) \
|
||||||
$(foreach v, $(LUA_CLIB), $(LUA_CLIB_PATH)/$(v).so)
|
$(foreach v, $(LUA_CLIB), $(LUA_CLIB_PATH)/$(v).so)
|
||||||
|
|
||||||
$(SKYNET_BUILD_PATH)/skynet : $(foreach v, $(SKYNET_SRC), skynet-src/$(v)) $(LUA_LIB) $(JEMALLOC_STATICLIB)
|
$(SKYNET_BUILD_PATH)/skynet : $(foreach v, $(SKYNET_SRC), skynet-src/$(v)) $(LUA_LIB) $(MALLOC_STATICLIB)
|
||||||
$(CC) $(CFLAGS) -o $@ $^ -Iskynet-src -I$(JEMALLOC_INC) $(LDFLAGS) $(EXPORT) $(LIBS)
|
$(CC) $(CFLAGS) -o $@ $^ -Iskynet-src -I$(JEMALLOC_INC) $(LDFLAGS) $(EXPORT) $(SKYNET_LIBS)
|
||||||
|
|
||||||
$(LUA_CLIB_PATH) :
|
$(LUA_CLIB_PATH) :
|
||||||
mkdir $(LUA_CLIB_PATH)
|
mkdir $(LUA_CLIB_PATH)
|
||||||
|
|||||||
@@ -39,9 +39,7 @@ local client_meta = {
|
|||||||
|
|
||||||
return "[mongo client : " .. self.host .. port_string .."]"
|
return "[mongo client : " .. self.host .. port_string .."]"
|
||||||
end,
|
end,
|
||||||
__gc = function(self)
|
-- DO NOT need disconnect, because channel will shutdown during gc
|
||||||
self:disconnect()
|
|
||||||
end
|
|
||||||
}
|
}
|
||||||
|
|
||||||
local mongo_db = {}
|
local mongo_db = {}
|
||||||
|
|||||||
@@ -9,9 +9,7 @@ local redis = {}
|
|||||||
local command = {}
|
local command = {}
|
||||||
local meta = {
|
local meta = {
|
||||||
__index = command,
|
__index = command,
|
||||||
__gc = function(self)
|
-- DO NOT close channel in __gc
|
||||||
self[1]:close()
|
|
||||||
end,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
---------- redis response
|
---------- redis response
|
||||||
|
|||||||
@@ -30,6 +30,11 @@ local function suspend(s)
|
|||||||
assert(not s.co)
|
assert(not s.co)
|
||||||
s.co = coroutine.running()
|
s.co = coroutine.running()
|
||||||
skynet.wait()
|
skynet.wait()
|
||||||
|
-- wakeup closing corouting every time suspend,
|
||||||
|
-- because socket.close() will wait last socket buffer operation before clear the buffer.
|
||||||
|
if s.closing then
|
||||||
|
skynet.wakeup(s.closing)
|
||||||
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
-- read skynet_socket.h for these macro
|
-- read skynet_socket.h for these macro
|
||||||
@@ -37,7 +42,7 @@ end
|
|||||||
socket_message[1] = function(id, size, data)
|
socket_message[1] = function(id, size, data)
|
||||||
local s = socket_pool[id]
|
local s = socket_pool[id]
|
||||||
if s == nil then
|
if s == nil then
|
||||||
print("socket: drop package from " .. id)
|
skynet.error("socket: drop package from " .. id)
|
||||||
driver.drop(data, size)
|
driver.drop(data, size)
|
||||||
return
|
return
|
||||||
end
|
end
|
||||||
@@ -95,13 +100,14 @@ end
|
|||||||
socket_message[5] = function(id)
|
socket_message[5] = function(id)
|
||||||
local s = socket_pool[id]
|
local s = socket_pool[id]
|
||||||
if s == nil then
|
if s == nil then
|
||||||
print("socket: error on unknown", id)
|
skynet.error("socket: error on unknown", id)
|
||||||
return
|
return
|
||||||
end
|
end
|
||||||
if s.connected then
|
if s.connected then
|
||||||
print("socket: error on", id)
|
skynet.error("socket: error on", id)
|
||||||
end
|
end
|
||||||
s.connected = false
|
s.connected = false
|
||||||
|
|
||||||
wakeup(s)
|
wakeup(s)
|
||||||
end
|
end
|
||||||
|
|
||||||
@@ -175,7 +181,8 @@ function socket.close(id)
|
|||||||
-- notice: call socket.close in __gc should be carefully,
|
-- notice: call socket.close in __gc should be carefully,
|
||||||
-- because skynet.wait never return in __gc, so driver.clear may not be called
|
-- because skynet.wait never return in __gc, so driver.clear may not be called
|
||||||
if s.co then
|
if s.co then
|
||||||
-- reading this socket on another coroutine
|
-- reading this socket on another coroutine, so don't shutdown (clear the buffer) immediatel
|
||||||
|
-- wait reading coroutine read the buffer.
|
||||||
assert(not s.closing)
|
assert(not s.closing)
|
||||||
s.closing = coroutine.running()
|
s.closing = coroutine.running()
|
||||||
skynet.wait()
|
skynet.wait()
|
||||||
@@ -189,13 +196,6 @@ function socket.close(id)
|
|||||||
socket_pool[id] = nil
|
socket_pool[id] = nil
|
||||||
end
|
end
|
||||||
|
|
||||||
local function close_socket(s)
|
|
||||||
if s.closing then
|
|
||||||
skynet.wakeup(s.closing)
|
|
||||||
end
|
|
||||||
return driver.readall(s.buffer, buffer_pool)
|
|
||||||
end
|
|
||||||
|
|
||||||
function socket.read(id, sz)
|
function socket.read(id, sz)
|
||||||
local s = socket_pool[id]
|
local s = socket_pool[id]
|
||||||
assert(s)
|
assert(s)
|
||||||
@@ -204,7 +204,7 @@ function socket.read(id, sz)
|
|||||||
return ret
|
return ret
|
||||||
end
|
end
|
||||||
if not s.connected then
|
if not s.connected then
|
||||||
return false, close_socket(s)
|
return false, driver.readall(s.buffer, buffer_pool)
|
||||||
end
|
end
|
||||||
|
|
||||||
assert(not s.read_required)
|
assert(not s.read_required)
|
||||||
@@ -214,7 +214,7 @@ function socket.read(id, sz)
|
|||||||
if ret then
|
if ret then
|
||||||
return ret
|
return ret
|
||||||
else
|
else
|
||||||
return false, close_socket(s)
|
return false, driver.readall(s.buffer, buffer_pool)
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
@@ -222,14 +222,14 @@ function socket.readall(id)
|
|||||||
local s = socket_pool[id]
|
local s = socket_pool[id]
|
||||||
assert(s)
|
assert(s)
|
||||||
if not s.connected then
|
if not s.connected then
|
||||||
local r = close_socket(s)
|
local r = driver.readall(s.buffer, buffer_pool)
|
||||||
return r ~= "" and r
|
return r ~= "" and r
|
||||||
end
|
end
|
||||||
assert(not s.read_required)
|
assert(not s.read_required)
|
||||||
s.read_required = true
|
s.read_required = true
|
||||||
suspend(s)
|
suspend(s)
|
||||||
assert(s.connected == false)
|
assert(s.connected == false)
|
||||||
return close_socket(s)
|
return driver.readall(s.buffer, buffer_pool)
|
||||||
end
|
end
|
||||||
|
|
||||||
function socket.readline(id, sep)
|
function socket.readline(id, sep)
|
||||||
@@ -241,7 +241,7 @@ function socket.readline(id, sep)
|
|||||||
return ret
|
return ret
|
||||||
end
|
end
|
||||||
if not s.connected then
|
if not s.connected then
|
||||||
return false, close_socket(s)
|
return false, driver.readall(s.buffer, buffer_pool)
|
||||||
end
|
end
|
||||||
assert(not s.read_required)
|
assert(not s.read_required)
|
||||||
s.read_required = sep
|
s.read_required = sep
|
||||||
@@ -249,7 +249,7 @@ function socket.readline(id, sep)
|
|||||||
if s.connected then
|
if s.connected then
|
||||||
return driver.readline(s.buffer, buffer_pool, sep)
|
return driver.readline(s.buffer, buffer_pool, sep)
|
||||||
else
|
else
|
||||||
return false, close_socket(s)
|
return false, driver.readall(s.buffer, buffer_pool)
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
@@ -261,9 +261,6 @@ function socket.block(id)
|
|||||||
assert(not s.read_required)
|
assert(not s.read_required)
|
||||||
s.read_required = 0
|
s.read_required = 0
|
||||||
suspend(s)
|
suspend(s)
|
||||||
if not s.connected and s.closing then
|
|
||||||
skynet.wakeup(s.closing)
|
|
||||||
end
|
|
||||||
return s.connected
|
return s.connected
|
||||||
end
|
end
|
||||||
|
|
||||||
|
|||||||
@@ -27,8 +27,8 @@ function socket_channel.channel(desc)
|
|||||||
__host = assert(desc.host),
|
__host = assert(desc.host),
|
||||||
__port = assert(desc.port),
|
__port = assert(desc.port),
|
||||||
__auth = desc.auth,
|
__auth = desc.auth,
|
||||||
__response = desc.response,
|
__response = desc.response, -- It's for session mode
|
||||||
__request = {}, -- request seq { response func or session }
|
__request = {}, -- request seq { response func or session } -- It's for order mode
|
||||||
__thread = {}, -- coroutine seq or session->coroutine map
|
__thread = {}, -- coroutine seq or session->coroutine map
|
||||||
__result = {}, -- response result { coroutine -> result }
|
__result = {}, -- response result { coroutine -> result }
|
||||||
__result_data = {},
|
__result_data = {},
|
||||||
@@ -51,72 +51,101 @@ local function close_channel_socket(self)
|
|||||||
end
|
end
|
||||||
|
|
||||||
local function wakeup_all(self, errmsg)
|
local function wakeup_all(self, errmsg)
|
||||||
for i = 1, #self.__thread do
|
if self.__response then
|
||||||
local co = self.__thread[i]
|
for k,co in pairs(self.__thread) do
|
||||||
self.__thread[i] = nil
|
self.__thread[k] = nil
|
||||||
self.__result[co] = socket_error
|
self.__result[co] = socket_error
|
||||||
self.__result_data[co] = errmsg
|
self.__result_data[co] = errmsg
|
||||||
skynet.wakeup(co)
|
skynet.wakeup(co)
|
||||||
|
end
|
||||||
|
else
|
||||||
|
for i = 1, #self.__request do
|
||||||
|
self.__request[i] = nil
|
||||||
|
end
|
||||||
|
for i = 1, #self.__thread do
|
||||||
|
local co = self.__thread[i]
|
||||||
|
self.__thread[i] = nil
|
||||||
|
self.__result[co] = socket_error
|
||||||
|
self.__result_data[co] = errmsg
|
||||||
|
skynet.wakeup(co)
|
||||||
|
end
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
local function dispatch_response(self)
|
|
||||||
|
|
||||||
|
local function dispatch_by_session(self)
|
||||||
local response = self.__response
|
local response = self.__response
|
||||||
if response then
|
-- response() return session
|
||||||
-- response() return session
|
while self.__sock do
|
||||||
while self.__sock do
|
local ok , session, result_ok, result_data = pcall(response, self.__sock)
|
||||||
local ok , session, result_ok, result_data = pcall(response, self.__sock)
|
if ok and session then
|
||||||
if ok and session then
|
local co = self.__thread[session]
|
||||||
local co = self.__thread[session]
|
self.__thread[session] = nil
|
||||||
self.__thread[session] = nil
|
if co then
|
||||||
if co then
|
self.__result[co] = result_ok
|
||||||
self.__result[co] = result_ok
|
self.__result_data[co] = result_data
|
||||||
self.__result_data[co] = result_data
|
skynet.wakeup(co)
|
||||||
skynet.wakeup(co)
|
else
|
||||||
else
|
skynet.error("socket: unknown session :", session)
|
||||||
print("socket: unknown session :", session)
|
end
|
||||||
end
|
else
|
||||||
|
close_channel_socket(self)
|
||||||
|
local errormsg
|
||||||
|
if session ~= socket_error then
|
||||||
|
errormsg = session
|
||||||
|
end
|
||||||
|
wakeup_all(self, errormsg)
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
local function pop_response(self)
|
||||||
|
return table.remove(self.__request, 1), table.remove(self.__thread, 1)
|
||||||
|
end
|
||||||
|
|
||||||
|
local function push_response(self, response, co)
|
||||||
|
if self.__response then
|
||||||
|
-- response is session
|
||||||
|
self.__thread[response] = co
|
||||||
|
else
|
||||||
|
-- response is a function, push it to __request
|
||||||
|
table.insert(self.__request, response)
|
||||||
|
table.insert(self.__thread, co)
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
local function dispatch_by_order(self)
|
||||||
|
while self.__sock do
|
||||||
|
local func, co = pop_response(self)
|
||||||
|
if func == nil then
|
||||||
|
if not socket.block(self.__sock[1]) then
|
||||||
|
close_channel_socket(self)
|
||||||
|
wakeup_all(self)
|
||||||
|
end
|
||||||
|
else
|
||||||
|
local ok, result_ok, result_data = pcall(func, self.__sock)
|
||||||
|
if ok then
|
||||||
|
self.__result[co] = result_ok
|
||||||
|
self.__result_data[co] = result_data
|
||||||
|
skynet.wakeup(co)
|
||||||
else
|
else
|
||||||
close_channel_socket(self)
|
close_channel_socket(self)
|
||||||
local errormsg
|
local errmsg
|
||||||
if session ~= socket_error then
|
if result ~= socket_error then
|
||||||
errormsg = session
|
errmsg = result_ok
|
||||||
end
|
|
||||||
for k,co in pairs(self.__thread) do
|
|
||||||
-- throw error (errormsg)
|
|
||||||
self.__thread[k] = nil
|
|
||||||
self.__result[co] = socket_error
|
|
||||||
self.__result_data[co] = errormsg
|
|
||||||
skynet.wakeup(co)
|
|
||||||
end
|
end
|
||||||
|
wakeup_all(self, errmsg)
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
local function dispatch_function(self)
|
||||||
|
if self.__response then
|
||||||
|
return dispatch_by_session
|
||||||
else
|
else
|
||||||
-- pop response function from __request
|
return dispatch_by_order
|
||||||
while self.__sock do
|
|
||||||
local func = table.remove(self.__request, 1)
|
|
||||||
if func == nil then
|
|
||||||
if not socket.block(self.__sock[1]) then
|
|
||||||
close_channel_socket(self)
|
|
||||||
wakeup_all(self)
|
|
||||||
end
|
|
||||||
else
|
|
||||||
local ok, result_ok, result_data = pcall(func, self.__sock)
|
|
||||||
if ok then
|
|
||||||
local co = table.remove(self.__thread, 1)
|
|
||||||
self.__result[co] = result_ok
|
|
||||||
self.__result_data[co] = result_data
|
|
||||||
skynet.wakeup(co)
|
|
||||||
else
|
|
||||||
close_channel_socket(self)
|
|
||||||
local errmsg
|
|
||||||
if result ~= socket_error then
|
|
||||||
errmsg = result_ok
|
|
||||||
end
|
|
||||||
wakeup_all(self, errmsg)
|
|
||||||
end
|
|
||||||
end
|
|
||||||
end
|
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
@@ -128,14 +157,14 @@ local function connect_once(self)
|
|||||||
end
|
end
|
||||||
self.__authcoroutine = coroutine.running()
|
self.__authcoroutine = coroutine.running()
|
||||||
self.__sock = setmetatable( {fd} , channel_socket_meta )
|
self.__sock = setmetatable( {fd} , channel_socket_meta )
|
||||||
skynet.fork(dispatch_response, self)
|
skynet.fork(dispatch_function(self), self)
|
||||||
|
|
||||||
if self.__auth then
|
if self.__auth then
|
||||||
local ok , message = pcall(self.__auth, self)
|
local ok , message = pcall(self.__auth, self)
|
||||||
if not ok then
|
if not ok then
|
||||||
close_channel_socket(self)
|
close_channel_socket(self)
|
||||||
if message ~= socket_error then
|
if message ~= socket_error then
|
||||||
print("socket: auth failed", message)
|
skynet.error("socket: auth failed", message)
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
self.__authcoroutine = false
|
self.__authcoroutine = false
|
||||||
@@ -150,12 +179,15 @@ local function try_connect(self , once)
|
|||||||
local t = 100
|
local t = 100
|
||||||
while not self.__closed do
|
while not self.__closed do
|
||||||
if connect_once(self) then
|
if connect_once(self) then
|
||||||
|
if not once then
|
||||||
|
skynet.error("socket: connect to", self.__host, self.__port)
|
||||||
|
end
|
||||||
return
|
return
|
||||||
elseif once then
|
elseif once then
|
||||||
error(string.format("Connect to %s:%d failed", self.__host, self.__port))
|
error(string.format("Connect to %s:%d failed", self.__host, self.__port))
|
||||||
end
|
end
|
||||||
if t > 1000 then
|
if t > 1000 then
|
||||||
print("socket: try to reconnect", self.__host, self.__port)
|
skynet.error("socket: try to reconnect", self.__host, self.__port)
|
||||||
skynet.sleep(t)
|
skynet.sleep(t)
|
||||||
t = 0
|
t = 0
|
||||||
else
|
else
|
||||||
@@ -179,6 +211,7 @@ local function block_connect(self, once)
|
|||||||
if self.__closed then
|
if self.__closed then
|
||||||
return false
|
return false
|
||||||
end
|
end
|
||||||
|
|
||||||
if #self.__connecting > 0 then
|
if #self.__connecting > 0 then
|
||||||
-- connecting in other coroutine
|
-- connecting in other coroutine
|
||||||
local co = coroutine.running()
|
local co = coroutine.running()
|
||||||
@@ -187,7 +220,6 @@ local function block_connect(self, once)
|
|||||||
-- check connection again
|
-- check connection again
|
||||||
return block_connect(self, once)
|
return block_connect(self, once)
|
||||||
end
|
end
|
||||||
|
|
||||||
self.__connecting[1] = true
|
self.__connecting[1] = true
|
||||||
try_connect(self, once)
|
try_connect(self, once)
|
||||||
self.__connecting[1] = nil
|
self.__connecting[1] = nil
|
||||||
@@ -209,10 +241,30 @@ function channel:connect(once)
|
|||||||
return block_connect(self, once)
|
return block_connect(self, once)
|
||||||
end
|
end
|
||||||
|
|
||||||
|
local function wait_for_response(self, response)
|
||||||
|
local co = coroutine.running()
|
||||||
|
push_response(self, response, co)
|
||||||
|
skynet.wait()
|
||||||
|
|
||||||
|
local result = self.__result[co]
|
||||||
|
self.__result[co] = nil
|
||||||
|
local result_data = self.__result_data[co]
|
||||||
|
self.__result_data[co] = nil
|
||||||
|
|
||||||
|
if result == socket_error then
|
||||||
|
error(socket_error)
|
||||||
|
else
|
||||||
|
assert(result, result_data)
|
||||||
|
return result_data
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
function channel:request(request, response)
|
function channel:request(request, response)
|
||||||
assert(block_connect(self))
|
assert(block_connect(self))
|
||||||
|
|
||||||
if not socket.write(self.__sock[1], request) then
|
if not socket.write(self.__sock[1], request) then
|
||||||
|
close_channel_socket(self)
|
||||||
|
wakeup_all(self)
|
||||||
error(socket_error)
|
error(socket_error)
|
||||||
end
|
end
|
||||||
|
|
||||||
@@ -221,59 +273,13 @@ function channel:request(request, response)
|
|||||||
return
|
return
|
||||||
end
|
end
|
||||||
|
|
||||||
local co = coroutine.running()
|
return wait_for_response(self, response)
|
||||||
|
|
||||||
if self.__response then
|
|
||||||
-- response is session
|
|
||||||
self.__thread[response] = co
|
|
||||||
else
|
|
||||||
-- response is a function, push it to __request
|
|
||||||
table.insert(self.__request, response)
|
|
||||||
table.insert(self.__thread, co)
|
|
||||||
end
|
|
||||||
skynet.wait()
|
|
||||||
|
|
||||||
local result = self.__result[co]
|
|
||||||
self.__result[co] = nil
|
|
||||||
local result_data = self.__result_data[co]
|
|
||||||
self.__result_data[co] = nil
|
|
||||||
|
|
||||||
if result == socket_error then
|
|
||||||
if result_data then
|
|
||||||
print("socket: dispatch", request, result_data)
|
|
||||||
end
|
|
||||||
error(socket_error)
|
|
||||||
else
|
|
||||||
assert(result, result_data)
|
|
||||||
return result_data
|
|
||||||
end
|
|
||||||
end
|
end
|
||||||
|
|
||||||
function channel:response(response)
|
function channel:response(response)
|
||||||
assert(block_connect(self))
|
assert(block_connect(self))
|
||||||
|
|
||||||
assert(type(response) == "function")
|
return wait_for_response(self, response)
|
||||||
|
|
||||||
local co = coroutine.running()
|
|
||||||
table.insert(self.__request, response)
|
|
||||||
table.insert(self.__thread, co)
|
|
||||||
|
|
||||||
skynet.wait()
|
|
||||||
|
|
||||||
local result = self.__result[co]
|
|
||||||
self.__result[co] = nil
|
|
||||||
local result_data = self.__result_data[co]
|
|
||||||
self.__result_data[co] = nil
|
|
||||||
|
|
||||||
if result == socket_error then
|
|
||||||
if result_data then
|
|
||||||
print("socket: dispatch", request, result_data)
|
|
||||||
end
|
|
||||||
error(socket_error)
|
|
||||||
else
|
|
||||||
assert(result, result_data)
|
|
||||||
return result_data
|
|
||||||
end
|
|
||||||
end
|
end
|
||||||
|
|
||||||
function channel:close()
|
function channel:close()
|
||||||
|
|||||||
21
platform.mk
21
platform.mk
@@ -5,30 +5,37 @@ CC ?= gcc
|
|||||||
|
|
||||||
.PHONY : none $(PLATS) clean all cleanall
|
.PHONY : none $(PLATS) clean all cleanall
|
||||||
|
|
||||||
ifneq ($(PLAT), none)
|
#ifneq ($(PLAT), none)
|
||||||
|
|
||||||
.PHONY : default
|
.PHONY : default
|
||||||
|
|
||||||
default :
|
default :
|
||||||
$(MAKE) $(PLAT)
|
$(MAKE) $(PLAT)
|
||||||
|
|
||||||
endif
|
#endif
|
||||||
|
|
||||||
none :
|
none :
|
||||||
@echo "Please do 'make PLATFORM' where PLATFORM is one of these:"
|
@echo "Please do 'make PLATFORM' where PLATFORM is one of these:"
|
||||||
@echo " $(PLATS)"
|
@echo " $(PLATS)"
|
||||||
|
|
||||||
LIBS := -lpthread -lm
|
SKYNET_LIBS := -lpthread -lm
|
||||||
SHARED := -fPIC --shared
|
SHARED := -fPIC --shared
|
||||||
EXPORT := -Wl,-E
|
EXPORT := -Wl,-E
|
||||||
|
|
||||||
$(PLATS) : all
|
|
||||||
|
|
||||||
linux : PLAT = linux
|
linux : PLAT = linux
|
||||||
macosx : PLAT = macosx
|
macosx : PLAT = macosx
|
||||||
freebsd : PLAT = freebsd
|
freebsd : PLAT = freebsd
|
||||||
|
|
||||||
macosx linux : LIBS += -ldl
|
|
||||||
macosx : SHARED := -fPIC -dynamiclib -Wl,-undefined,dynamic_lookup
|
macosx : SHARED := -fPIC -dynamiclib -Wl,-undefined,dynamic_lookup
|
||||||
macosx : EXPORT :=
|
macosx : EXPORT :=
|
||||||
linux freebsd : LIBS += -lrt
|
macosx linux : SKYNET_LIBS += -ldl
|
||||||
|
linux freebsd : SKYNET_LIBS += -lrt
|
||||||
|
|
||||||
|
# Turn off jemalloc and malloc hook on macosx
|
||||||
|
|
||||||
|
macosx : MALLOC_STATICLIB :=
|
||||||
|
macosx : SKYNET_DEFINES :=-DNOUSE_JEMALLOC
|
||||||
|
|
||||||
|
|
||||||
|
linux macosx freebsd :
|
||||||
|
$(MAKE) all PLAT=$@ SKYNET_LIBS="$(SKYNET_LIBS)" SHARED="$(SHARED)" EXPORT="$(EXPORT)" MALLOC_STATICLIB="$(MALLOC_STATICLIB)" SKYNET_DEFINES="$(SKYNET_DEFINES)"
|
||||||
|
|||||||
@@ -1,18 +1,16 @@
|
|||||||
#include <stdio.h>
|
#include <stdio.h>
|
||||||
#include <string.h>
|
#include <string.h>
|
||||||
#include <assert.h>
|
#include <assert.h>
|
||||||
|
#include <stdlib.h>
|
||||||
|
|
||||||
#include "malloc_hook.h"
|
#include "malloc_hook.h"
|
||||||
#include "jemalloc.h"
|
|
||||||
|
|
||||||
#include "skynet.h"
|
#include "skynet.h"
|
||||||
|
|
||||||
|
|
||||||
static size_t _used_memory = 0;
|
static size_t _used_memory = 0;
|
||||||
static size_t _memory_block = 0;
|
static size_t _memory_block = 0;
|
||||||
typedef struct _mem_data {
|
typedef struct _mem_data {
|
||||||
uint32_t handle;
|
uint32_t handle;
|
||||||
ssize_t allocated;
|
ssize_t allocated;
|
||||||
} mem_data;
|
} mem_data;
|
||||||
|
|
||||||
#define SLOT_SIZE 0x10000
|
#define SLOT_SIZE 0x10000
|
||||||
@@ -20,13 +18,18 @@ typedef struct _mem_data {
|
|||||||
|
|
||||||
static mem_data mem_stats[SLOT_SIZE];
|
static mem_data mem_stats[SLOT_SIZE];
|
||||||
|
|
||||||
|
|
||||||
|
#ifndef NOUSE_JEMALLOC
|
||||||
|
|
||||||
|
#include "jemalloc.h"
|
||||||
|
|
||||||
static ssize_t*
|
static ssize_t*
|
||||||
get_allocated_field(uint32_t handle) {
|
get_allocated_field(uint32_t handle) {
|
||||||
int h = (int)(handle & (SLOT_SIZE - 1));
|
int h = (int)(handle & (SLOT_SIZE - 1));
|
||||||
mem_data *data = &mem_stats[h];
|
mem_data *data = &mem_stats[h];
|
||||||
uint32_t old_handle = data->handle;
|
uint32_t old_handle = data->handle;
|
||||||
ssize_t old_alloc = data->allocated;
|
ssize_t old_alloc = data->allocated;
|
||||||
if(old_handle == 0 || old_alloc <= 0) {
|
if(old_handle == 0 || old_alloc <= 0) {
|
||||||
// data->allocated may less than zero, because it may not count at start.
|
// data->allocated may less than zero, because it may not count at start.
|
||||||
if(!__sync_bool_compare_and_swap(&data->handle, old_handle, handle)) {
|
if(!__sync_bool_compare_and_swap(&data->handle, old_handle, handle)) {
|
||||||
return 0;
|
return 0;
|
||||||
@@ -34,150 +37,174 @@ get_allocated_field(uint32_t handle) {
|
|||||||
if (old_alloc < 0) {
|
if (old_alloc < 0) {
|
||||||
__sync_bool_compare_and_swap(&data->allocated, old_alloc, 0);
|
__sync_bool_compare_and_swap(&data->allocated, old_alloc, 0);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if(data->handle != handle) {
|
if(data->handle != handle) {
|
||||||
return 0;
|
return 0;
|
||||||
}
|
}
|
||||||
return &data->allocated;
|
return &data->allocated;
|
||||||
}
|
}
|
||||||
|
|
||||||
inline static void
|
inline static void
|
||||||
update_xmalloc_stat_alloc(uint32_t handle, size_t __n) {
|
update_xmalloc_stat_alloc(uint32_t handle, size_t __n) {
|
||||||
__sync_add_and_fetch(&_used_memory, __n);
|
__sync_add_and_fetch(&_used_memory, __n);
|
||||||
__sync_add_and_fetch(&_memory_block, 1);
|
__sync_add_and_fetch(&_memory_block, 1);
|
||||||
ssize_t* allocated = get_allocated_field(handle);
|
ssize_t* allocated = get_allocated_field(handle);
|
||||||
if(allocated) {
|
if(allocated) {
|
||||||
__sync_add_and_fetch(allocated, __n);
|
__sync_add_and_fetch(allocated, __n);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
inline static void
|
inline static void
|
||||||
update_xmalloc_stat_free(uint32_t handle, size_t __n) {
|
update_xmalloc_stat_free(uint32_t handle, size_t __n) {
|
||||||
__sync_sub_and_fetch(&_used_memory, __n);
|
__sync_sub_and_fetch(&_used_memory, __n);
|
||||||
__sync_sub_and_fetch(&_memory_block, 1);
|
__sync_sub_and_fetch(&_memory_block, 1);
|
||||||
ssize_t* allocated = get_allocated_field(handle);
|
ssize_t* allocated = get_allocated_field(handle);
|
||||||
if(allocated) {
|
if(allocated) {
|
||||||
__sync_sub_and_fetch(allocated, __n);
|
__sync_sub_and_fetch(allocated, __n);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
inline static void*
|
inline static void*
|
||||||
fill_prefix(char* ptr) {
|
fill_prefix(char* ptr) {
|
||||||
uint32_t handle = skynet_current_handle();
|
uint32_t handle = skynet_current_handle();
|
||||||
size_t size = je_malloc_usable_size(ptr);
|
size_t size = je_malloc_usable_size(ptr);
|
||||||
uint32_t *p = (uint32_t *)(ptr + size - sizeof(uint32_t));
|
uint32_t *p = (uint32_t *)(ptr + size - sizeof(uint32_t));
|
||||||
memcpy(p, &handle, sizeof(handle));
|
memcpy(p, &handle, sizeof(handle));
|
||||||
|
|
||||||
update_xmalloc_stat_alloc(handle, size);
|
update_xmalloc_stat_alloc(handle, size);
|
||||||
return ptr;
|
return ptr;
|
||||||
}
|
}
|
||||||
|
|
||||||
inline static void*
|
inline static void*
|
||||||
clean_prefix(char* ptr) {
|
clean_prefix(char* ptr) {
|
||||||
size_t size = je_malloc_usable_size(ptr);
|
size_t size = je_malloc_usable_size(ptr);
|
||||||
uint32_t *p = (uint32_t *)(ptr + size - sizeof(uint32_t));
|
uint32_t *p = (uint32_t *)(ptr + size - sizeof(uint32_t));
|
||||||
uint32_t handle;
|
uint32_t handle;
|
||||||
memcpy(&handle, p, sizeof(handle));
|
memcpy(&handle, p, sizeof(handle));
|
||||||
update_xmalloc_stat_free(handle, size);
|
update_xmalloc_stat_free(handle, size);
|
||||||
return ptr;
|
return ptr;
|
||||||
}
|
}
|
||||||
|
|
||||||
static void malloc_oom(size_t size) {
|
static void malloc_oom(size_t size) {
|
||||||
fprintf(stderr, "xmalloc: Out of memory trying to allocate %zu bytes\n",
|
fprintf(stderr, "xmalloc: Out of memory trying to allocate %zu bytes\n",
|
||||||
size);
|
size);
|
||||||
fflush(stderr);
|
fflush(stderr);
|
||||||
abort();
|
abort();
|
||||||
}
|
}
|
||||||
|
|
||||||
size_t
|
void
|
||||||
malloc_used_memory(void) {
|
memory_info_dump(void) {
|
||||||
return _used_memory;
|
je_malloc_stats_print(0,0,0);
|
||||||
}
|
}
|
||||||
|
|
||||||
size_t
|
size_t
|
||||||
malloc_memory_block(void) {
|
mallctl_int64(const char* name, size_t* newval) {
|
||||||
return _memory_block;
|
size_t v = 0;
|
||||||
|
size_t len = sizeof(v);
|
||||||
|
if(newval) {
|
||||||
|
je_mallctl(name, &v, &len, newval, sizeof(size_t));
|
||||||
|
} else {
|
||||||
|
je_mallctl(name, &v, &len, NULL, 0);
|
||||||
|
}
|
||||||
|
// printf("name: %s, value: %zd\n", name, v);
|
||||||
|
return v;
|
||||||
}
|
}
|
||||||
|
|
||||||
void memory_info_dump(void) {
|
int
|
||||||
je_malloc_stats_print(0,0,0);
|
mallctl_opt(const char* name, int* newval) {
|
||||||
|
int v = 0;
|
||||||
|
size_t len = sizeof(v);
|
||||||
|
if(newval) {
|
||||||
|
int ret = je_mallctl(name, &v, &len, newval, sizeof(int));
|
||||||
|
if(ret == 0) {
|
||||||
|
printf("set new value(%d) for (%s) succeed\n", *newval, name);
|
||||||
|
} else {
|
||||||
|
printf("set new value(%d) for (%s) failed: error -> %d\n", *newval, name, ret);
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
je_mallctl(name, &v, &len, NULL, 0);
|
||||||
|
}
|
||||||
|
|
||||||
|
return v;
|
||||||
}
|
}
|
||||||
|
|
||||||
size_t mallctl_int64(const char* name, size_t* newval) {
|
// hook : malloc, realloc, free, calloc
|
||||||
size_t v = 0;
|
|
||||||
size_t len = sizeof(v);
|
|
||||||
if(newval) {
|
|
||||||
je_mallctl(name, &v, &len, newval, sizeof(size_t));
|
|
||||||
} else {
|
|
||||||
je_mallctl(name, &v, &len, NULL, 0);
|
|
||||||
}
|
|
||||||
// printf("name: %s, value: %zd\n", name, v);
|
|
||||||
return v;
|
|
||||||
}
|
|
||||||
|
|
||||||
int mallctl_opt(const char* name, int* newval) {
|
|
||||||
int v = 0;
|
|
||||||
size_t len = sizeof(v);
|
|
||||||
if(newval) {
|
|
||||||
int ret = je_mallctl(name, &v, &len, newval, sizeof(int));
|
|
||||||
if(ret == 0) {
|
|
||||||
printf("set new value(%d) for (%s) succeed\n", *newval, name);
|
|
||||||
} else {
|
|
||||||
printf("set new value(%d) for (%s) failed: error -> %d\n", *newval, name, ret);
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
je_mallctl(name, &v, &len, NULL, 0);
|
|
||||||
}
|
|
||||||
|
|
||||||
return v;
|
|
||||||
}
|
|
||||||
|
|
||||||
void
|
|
||||||
dump_c_mem() {
|
|
||||||
int i;
|
|
||||||
size_t total = 0;
|
|
||||||
printf("dump all service mem:\n");
|
|
||||||
for(i=0; i<SLOT_SIZE; i++) {
|
|
||||||
mem_data* data = &mem_stats[i];
|
|
||||||
if(data->handle != 0 && data->allocated != 0) {
|
|
||||||
total += data->allocated;
|
|
||||||
printf("0x%x -> %zdkb\n", data->handle, data->allocated >> 10);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
printf("+total: %zdkb\n",total >> 10);
|
|
||||||
}
|
|
||||||
|
|
||||||
// hook : malloc, realloc, memalign, free, calloc
|
|
||||||
|
|
||||||
void *
|
void *
|
||||||
skynet_malloc(size_t size) {
|
skynet_malloc(size_t size) {
|
||||||
void* ptr = je_malloc(size + PREFIX_SIZE);
|
void* ptr = je_malloc(size + PREFIX_SIZE);
|
||||||
if(!ptr) malloc_oom(size);
|
if(!ptr) malloc_oom(size);
|
||||||
return fill_prefix(ptr);
|
return fill_prefix(ptr);
|
||||||
}
|
}
|
||||||
|
|
||||||
void *
|
void *
|
||||||
skynet_realloc(void *ptr, size_t size) {
|
skynet_realloc(void *ptr, size_t size) {
|
||||||
if (ptr == NULL) return skynet_malloc(size);
|
if (ptr == NULL) return skynet_malloc(size);
|
||||||
|
|
||||||
void* rawptr = clean_prefix(ptr);
|
void* rawptr = clean_prefix(ptr);
|
||||||
void *newptr = je_realloc(rawptr, size+PREFIX_SIZE);
|
void *newptr = je_realloc(rawptr, size+PREFIX_SIZE);
|
||||||
if(!newptr) malloc_oom(size);
|
if(!newptr) malloc_oom(size);
|
||||||
return fill_prefix(newptr);
|
return fill_prefix(newptr);
|
||||||
}
|
}
|
||||||
|
|
||||||
void
|
void
|
||||||
skynet_free(void *ptr) {
|
skynet_free(void *ptr) {
|
||||||
if (ptr == NULL) return;
|
if (ptr == NULL) return;
|
||||||
void* rawptr = clean_prefix(ptr);
|
void* rawptr = clean_prefix(ptr);
|
||||||
je_free(rawptr);
|
je_free(rawptr);
|
||||||
}
|
}
|
||||||
|
|
||||||
void *
|
void *
|
||||||
skynet_calloc(size_t nmemb,size_t size) {
|
skynet_calloc(size_t nmemb,size_t size) {
|
||||||
void* ptr = je_calloc(nmemb + ((PREFIX_SIZE+size-1)/size), size );
|
void* ptr = je_calloc(nmemb + ((PREFIX_SIZE+size-1)/size), size );
|
||||||
if(!ptr) malloc_oom(size);
|
if(!ptr) malloc_oom(size);
|
||||||
return fill_prefix(ptr);
|
return fill_prefix(ptr);
|
||||||
|
}
|
||||||
|
|
||||||
|
#else
|
||||||
|
|
||||||
|
void
|
||||||
|
memory_info_dump(void) {
|
||||||
|
skynet_error(NULL, "No jemalloc");
|
||||||
|
}
|
||||||
|
|
||||||
|
size_t
|
||||||
|
mallctl_int64(const char* name, size_t* newval) {
|
||||||
|
skynet_error(NULL, "No jemalloc : mallctl_int64 %s.", name);
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
int
|
||||||
|
mallctl_opt(const char* name, int* newval) {
|
||||||
|
skynet_error(NULL, "No jemalloc : mallctl_opt %s.", name);
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
#endif
|
||||||
|
|
||||||
|
size_t
|
||||||
|
malloc_used_memory(void) {
|
||||||
|
return _used_memory;
|
||||||
|
}
|
||||||
|
|
||||||
|
size_t
|
||||||
|
malloc_memory_block(void) {
|
||||||
|
return _memory_block;
|
||||||
|
}
|
||||||
|
|
||||||
|
void
|
||||||
|
dump_c_mem() {
|
||||||
|
int i;
|
||||||
|
size_t total = 0;
|
||||||
|
printf("dump all service mem:\n");
|
||||||
|
for(i=0; i<SLOT_SIZE; i++) {
|
||||||
|
mem_data* data = &mem_stats[i];
|
||||||
|
if(data->handle != 0 && data->allocated != 0) {
|
||||||
|
total += data->allocated;
|
||||||
|
printf("0x%x -> %zdkb\n", data->handle, data->allocated >> 10);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
printf("+total: %zdkb\n",total >> 10);
|
||||||
}
|
}
|
||||||
|
|
||||||
char *
|
char *
|
||||||
@@ -198,3 +225,8 @@ skynet_lalloc(void *ud, void *ptr, size_t osize, size_t nsize) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
void
|
||||||
|
malloc_inithook(void) {
|
||||||
|
memset(mem_stats, 0, sizeof(mem_stats));
|
||||||
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -3,12 +3,12 @@
|
|||||||
|
|
||||||
#include <stddef.h>
|
#include <stddef.h>
|
||||||
|
|
||||||
#ifdef SKYNET_MALLOC_RENAME
|
#ifdef NOUSE_JEMALLOC
|
||||||
|
|
||||||
#define malloc skynet_malloc
|
#define skynet_malloc malloc
|
||||||
#define calloc skynet_calloc
|
#define skynet_calloc calloc
|
||||||
#define realloc skynet_realloc
|
#define skynet_realloc realloc
|
||||||
#define free skynet_free
|
#define skynet_free free
|
||||||
|
|
||||||
#endif
|
#endif
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user