From b164e3a8a9cb1ce8d3a9988fba59d4e73241cfc1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BA=91=E9=A3=8E?= Date: Mon, 11 Jan 2021 10:54:34 +0800 Subject: [PATCH] use stdatomic (#1317) --- 3rd/lua/lstring.c | 4 +- lualib-src/lua-multicast.c | 10 ++-- lualib-src/lua-sharedata.c | 14 ++--- lualib-src/lua-stm.c | 12 ++-- service-src/service_snlua.c | 24 ++++---- skynet-src/atomic.h | 38 ++++++++++-- skynet-src/malloc_hook.c | 20 +++---- skynet-src/rwlock.h | 32 +++++----- skynet-src/skynet_monitor.c | 4 +- skynet-src/skynet_server.c | 38 ++++++------ skynet-src/socket_server.c | 115 +++++++++++++++++++----------------- skynet-src/spinlock.h | 28 +++++++-- 12 files changed, 201 insertions(+), 138 deletions(-) diff --git a/3rd/lua/lstring.c b/3rd/lua/lstring.c index 195534ac..89672afe 100644 --- a/3rd/lua/lstring.c +++ b/3rd/lua/lstring.c @@ -23,7 +23,7 @@ #include "atomic.h" static unsigned int STRSEED; -static size_t STRID = 0; +static ATOM_SIZET STRID = 0; /* ** Maximum size for string table. @@ -60,7 +60,7 @@ void luaS_share (TString *ts) { if (ts == NULL || isshared(ts)) return; makeshared(ts); - ts->id = ATOM_DEC(&STRID); + ts->id = ATOM_FDEC(&STRID) - 1; } unsigned int luaS_hash (const char *str, size_t l, unsigned int seed) { diff --git a/lualib-src/lua-multicast.c b/lualib-src/lua-multicast.c index dab90ee2..1c845dc8 100644 --- a/lualib-src/lua-multicast.c +++ b/lualib-src/lua-multicast.c @@ -10,7 +10,7 @@ #include "atomic.h" struct mc_package { - int reference; + ATOM_INT reference; uint32_t size; void *data; }; @@ -18,7 +18,7 @@ struct mc_package { static int pack(lua_State *L, void *data, size_t size) { struct mc_package * pack = skynet_malloc(sizeof(struct mc_package)); - pack->reference = 0; + ATOM_INIT(&pack->reference, 0); pack->size = (uint32_t)size; pack->data = data; struct mc_package ** ret = skynet_malloc(sizeof(*ret)); @@ -91,10 +91,10 @@ static int mc_bindrefer(lua_State *L) { struct mc_package ** pack = lua_touserdata(L,1); int ref = luaL_checkinteger(L,2); - if ((*pack)->reference != 0) { + if (ATOM_LOAD(&(*pack)->reference) != 0) { return luaL_error(L, "Can't bind a multicast package more than once"); } - (*pack)->reference = ref; + ATOM_STORE(&(*pack)->reference , ref); lua_pushlightuserdata(L, *pack); @@ -110,7 +110,7 @@ static int mc_closelocal(lua_State *L) { struct mc_package *pack = lua_touserdata(L,1); - int ref = ATOM_DEC(&pack->reference); + int ref = ATOM_FDEC(&pack->reference)-1; if (ref <= 0) { skynet_free(pack->data); skynet_free(pack); diff --git a/lualib-src/lua-sharedata.c b/lualib-src/lua-sharedata.c index e96bf407..8e329f9d 100644 --- a/lualib-src/lua-sharedata.c +++ b/lualib-src/lua-sharedata.c @@ -40,7 +40,7 @@ struct node { struct state { int dirty; - int ref; + ATOM_INT ref; struct table * root; }; @@ -383,7 +383,7 @@ convert_stringmap(struct context *ctx, struct table *tbl) { lua_pushvalue(L, 1); struct state * s = lua_newuserdatauv(L, sizeof(*s), 1); s->dirty = 0; - s->ref = 0; + ATOM_INIT(&s->ref , 0); s->root = tbl; lua_replace(L, 1); lua_replace(L, -2); @@ -665,7 +665,7 @@ releaseobj(lua_State *L) { struct ctrl *c = lua_touserdata(L, 1); struct table *tbl = c->root; struct state *s = lua_touserdata(tbl->L, 1); - ATOM_DEC(&s->ref); + ATOM_FDEC(&s->ref); c->root = NULL; c->update = NULL; @@ -676,7 +676,7 @@ static int lboxconf(lua_State *L) { struct table * tbl = get_table(L,1); struct state * s = lua_touserdata(tbl->L, 1); - ATOM_INC(&s->ref); + ATOM_FINC(&s->ref); struct ctrl * c = lua_newuserdatauv(L, sizeof(*c), 1); c->root = tbl; @@ -712,7 +712,7 @@ static int lgetref(lua_State *L) { struct table *tbl = get_table(L,1); struct state * s = lua_touserdata(tbl->L, 1); - lua_pushinteger(L , s->ref); + lua_pushinteger(L , ATOM_LOAD(&s->ref)); return 1; } @@ -721,7 +721,7 @@ static int lincref(lua_State *L) { struct table *tbl = get_table(L,1); struct state * s = lua_touserdata(tbl->L, 1); - int ref = ATOM_INC(&s->ref); + int ref = ATOM_FINC(&s->ref)+1; lua_pushinteger(L , ref); return 1; @@ -731,7 +731,7 @@ static int ldecref(lua_State *L) { struct table *tbl = get_table(L,1); struct state * s = lua_touserdata(tbl->L, 1); - int ref = ATOM_DEC(&s->ref); + int ref = ATOM_FDEC(&s->ref)-1; lua_pushinteger(L , ref); return 1; diff --git a/lualib-src/lua-stm.c b/lualib-src/lua-stm.c index ecfa1e57..86502c91 100644 --- a/lualib-src/lua-stm.c +++ b/lualib-src/lua-stm.c @@ -13,7 +13,7 @@ struct stm_object { struct rwlock lock; - int reference; + ATOM_INT reference; struct stm_copy * copy; }; @@ -27,7 +27,7 @@ struct stm_copy { static struct stm_copy * stm_newcopy(void * msg, int32_t sz) { struct stm_copy * copy = skynet_malloc(sizeof(*copy)); - copy->reference = 1; + ATOM_INIT(©->reference, 1); copy->sz = sz; copy->msg = msg; @@ -38,7 +38,7 @@ static struct stm_object * stm_new(void * msg, int32_t sz) { struct stm_object * obj = skynet_malloc(sizeof(*obj)); rwlock_init(&obj->lock); - obj->reference = 1; + ATOM_INIT(&obj->reference , 1); obj->copy = stm_newcopy(msg, sz); return obj; @@ -48,7 +48,7 @@ static void stm_releasecopy(struct stm_copy *copy) { if (copy == NULL) return; - if (ATOM_DEC(©->reference) == 0) { + if (ATOM_FDEC(©->reference) <= 1) { skynet_free(copy->msg); skynet_free(copy); } @@ -61,7 +61,7 @@ stm_release(struct stm_object *obj) { // writer release the stm object, so release the last copy . stm_releasecopy(obj->copy); obj->copy = NULL; - if (--obj->reference > 0) { + if (ATOM_FDEC(&obj->reference) > 1) { // stm object grab by readers, reset the copy to NULL. rwlock_wunlock(&obj->lock); return; @@ -73,7 +73,7 @@ stm_release(struct stm_object *obj) { static void stm_releasereader(struct stm_object *obj) { rwlock_rlock(&obj->lock); - if (ATOM_DEC(&obj->reference) == 0) { + if (ATOM_FDEC(&obj->reference) == 1) { // last reader, no writer. so no need to unlock assert(obj->copy == NULL); skynet_free(obj); diff --git a/service-src/service_snlua.c b/service-src/service_snlua.c index 3ef4f0e7..f3ed03ee 100644 --- a/service-src/service_snlua.c +++ b/service-src/service_snlua.c @@ -30,7 +30,7 @@ struct snlua { size_t mem_report; size_t mem_limit; lua_State * activeL; - volatile int trap; + volatile ATOM_INT trap; }; // LUA_CACHELIB may defined in patched lua for shared proto @@ -67,8 +67,8 @@ signal_hook(lua_State *L, lua_Debug *ar) { struct snlua *l = (struct snlua *)ud; lua_sethook (L, NULL, 0, 0); - if (l->trap) { - l->trap = 0; + if (ATOM_LOAD(&l->trap)) { + ATOM_STORE(&l->trap , 0); luaL_error(L, "signal 0"); } } @@ -76,7 +76,7 @@ signal_hook(lua_State *L, lua_Debug *ar) { static void switchL(lua_State *L, struct snlua *l) { l->activeL = L; - if (l->trap) { + if (ATOM_LOAD(&l->trap)) { lua_sethook(L, signal_hook, LUA_MASKCOUNT, 1); } } @@ -88,9 +88,9 @@ lua_resumeX(lua_State *L, lua_State *from, int nargs, int *nresults) { struct snlua *l = (struct snlua *)ud; switchL(L, l); int err = lua_resume(L, from, nargs, nresults); - if (l->trap) { + if (ATOM_LOAD(&l->trap)) { // wait for lua_sethook. (l->trap == -1) - while (l->trap >= 0) ; + while (ATOM_LOAD(&l->trap) >= 0) ; } switchL(from, l); return err; @@ -506,7 +506,7 @@ snlua_create(void) { l->mem_limit = 0; l->L = lua_newstate(lalloc, l); l->activeL = NULL; - l->trap = 0; + ATOM_INIT(&l->trap , 0); return l; } @@ -520,13 +520,17 @@ void snlua_signal(struct snlua *l, int signal) { skynet_error(l->ctx, "recv a signal %d", signal); if (signal == 0) { - if (l->trap == 0) { + if (ATOM_LOAD(&l->trap) == 0) { + ATOM_INT zero; + ATOM_INIT(&zero, 0); // only one thread can set trap ( l->trap 0->1 ) - if (!ATOM_CAS(&l->trap, 0, 1)) + if (!ATOM_CAS(&l->trap, zero, 1)) return; + ATOM_INT one; + ATOM_INIT(&one, 1); lua_sethook (l->activeL, signal_hook, LUA_MASKCOUNT, 1); // finish set ( l->trap 1 -> -1 ) - ATOM_CAS(&l->trap, 1, -1); + ATOM_CAS(&l->trap, one, -1); } } else if (signal == 1) { skynet_error(l->ctx, "Current Memory %.3fK", (float)l->mem / 1024); diff --git a/skynet-src/atomic.h b/skynet-src/atomic.h index f79c7148..dea6fbef 100644 --- a/skynet-src/atomic.h +++ b/skynet-src/atomic.h @@ -1,14 +1,42 @@ #ifndef SKYNET_ATOMIC_H #define SKYNET_ATOMIC_H +#ifdef NO_STDATOMIC + +#define ATOM_BYTE unsigned char +#define ATOM_INT int +#define ATOM_POINTER void * +#define ATOM_SIZET size_t +#define ATOM_INIT(ptr, v) (*(ptr) = v) +#define ATOM_LOAD(ptr) (*(ptr)) +#define ATOM_STORE(ptr, v) (*(ptr) = v) #define ATOM_CAS(ptr, oval, nval) __sync_bool_compare_and_swap(ptr, oval, nval) #define ATOM_CAS_POINTER(ptr, oval, nval) __sync_bool_compare_and_swap(ptr, oval, nval) -#define ATOM_INC(ptr) __sync_add_and_fetch(ptr, 1) #define ATOM_FINC(ptr) __sync_fetch_and_add(ptr, 1) -#define ATOM_DEC(ptr) __sync_sub_and_fetch(ptr, 1) #define ATOM_FDEC(ptr) __sync_fetch_and_sub(ptr, 1) -#define ATOM_ADD(ptr,n) __sync_add_and_fetch(ptr, n) -#define ATOM_SUB(ptr,n) __sync_sub_and_fetch(ptr, n) -#define ATOM_AND(ptr,n) __sync_and_and_fetch(ptr, n) +#define ATOM_FADD(ptr,n) __sync_fetch_and_add(ptr, n) +#define ATOM_FSUB(ptr,n) __sync_fetch_and_sub(ptr, n) +#define ATOM_FAND(ptr,n) __sync_fetch_and_and(ptr, n) + +#else + +#include + +#define ATOM_BYTE atomic_uchar +#define ATOM_INT atomic_int +#define ATOM_POINTER atomic_uintptr_t +#define ATOM_SIZET atomic_size_t +#define ATOM_INIT(ref, v) atomic_init(ref, v) +#define ATOM_LOAD(ptr) atomic_load(ptr) +#define ATOM_STORE(ptr, v) atomic_store(ptr, v) +#define ATOM_CAS(ptr, oval, nval) atomic_compare_exchange_weak(ptr, &(oval), nval) +#define ATOM_CAS_POINTER(ptr, oval, nval) atomic_compare_exchange_weak(ptr, &(oval), nval) +#define ATOM_FINC(ptr) atomic_fetch_add(ptr, 1) +#define ATOM_FDEC(ptr) atomic_fetch_sub(ptr, 1) +#define ATOM_FADD(ptr,n) atomic_fetch_add(ptr, n) +#define ATOM_FSUB(ptr,n) atomic_fetch_sub(ptr, n) +#define ATOM_FAND(ptr,n) atomic_fetch_and(ptr, n) + +#endif #endif diff --git a/skynet-src/malloc_hook.c b/skynet-src/malloc_hook.c index a3fed30e..d02113cf 100644 --- a/skynet-src/malloc_hook.c +++ b/skynet-src/malloc_hook.c @@ -15,8 +15,8 @@ #define MEMORY_ALLOCTAG 0x20140605 #define MEMORY_FREETAG 0x0badf00d -static size_t _used_memory = 0; -static size_t _memory_block = 0; +static ATOM_SIZET _used_memory = 0; +static ATOM_SIZET _memory_block = 0; struct mem_data { uint32_t handle; @@ -67,21 +67,21 @@ get_allocated_field(uint32_t handle) { inline static void update_xmalloc_stat_alloc(uint32_t handle, size_t __n) { - ATOM_ADD(&_used_memory, __n); - ATOM_INC(&_memory_block); + ATOM_FADD(&_used_memory, __n); + ATOM_FINC(&_memory_block); ssize_t* allocated = get_allocated_field(handle); if(allocated) { - ATOM_ADD(allocated, __n); + ATOM_FADD(allocated, __n); } } inline static void update_xmalloc_stat_free(uint32_t handle, size_t __n) { - ATOM_SUB(&_used_memory, __n); - ATOM_DEC(&_memory_block); + ATOM_FSUB(&_used_memory, __n); + ATOM_FDEC(&_memory_block); ssize_t* allocated = get_allocated_field(handle); if(allocated) { - ATOM_SUB(allocated, __n); + ATOM_FSUB(allocated, __n); } } @@ -273,12 +273,12 @@ mallctl_cmd(const char* name) { size_t malloc_used_memory(void) { - return _used_memory; + return ATOM_LOAD(&_used_memory); } size_t malloc_memory_block(void) { - return _memory_block; + return ATOM_LOAD(&_memory_block); } void diff --git a/skynet-src/rwlock.h b/skynet-src/rwlock.h index 5a995918..56feebf7 100644 --- a/skynet-src/rwlock.h +++ b/skynet-src/rwlock.h @@ -3,26 +3,26 @@ #ifndef USE_PTHREAD_LOCK +#include "atomic.h" + struct rwlock { - int write; - int read; + ATOM_INT write; + ATOM_INT read; }; static inline void rwlock_init(struct rwlock *lock) { - lock->write = 0; - lock->read = 0; + ATOM_INIT(&lock->write, 0); + ATOM_INIT(&lock->read, 0); } static inline void rwlock_rlock(struct rwlock *lock) { for (;;) { - while(lock->write) { - __sync_synchronize(); - } - __sync_add_and_fetch(&lock->read,1); - if (lock->write) { - __sync_sub_and_fetch(&lock->read,1); + while(ATOM_LOAD(&lock->write)) {} + ATOM_FINC(&lock->read); + if (ATOM_LOAD(&lock->write)) { + ATOM_FDEC(&lock->read); } else { break; } @@ -31,20 +31,20 @@ rwlock_rlock(struct rwlock *lock) { static inline void rwlock_wlock(struct rwlock *lock) { - while (__sync_lock_test_and_set(&lock->write,1)) {} - while(lock->read) { - __sync_synchronize(); - } + ATOM_INT clear; + ATOM_INIT(&clear, 0); + while (!ATOM_CAS(&lock->write,clear,1)) {} + while(ATOM_LOAD(&lock->read)) {} } static inline void rwlock_wunlock(struct rwlock *lock) { - __sync_lock_release(&lock->write); + ATOM_STORE(&lock->write, 0); } static inline void rwlock_runlock(struct rwlock *lock) { - __sync_sub_and_fetch(&lock->read,1); + ATOM_FDEC(&lock->read); } #else diff --git a/skynet-src/skynet_monitor.c b/skynet-src/skynet_monitor.c index 8d47fc89..48def95b 100644 --- a/skynet-src/skynet_monitor.c +++ b/skynet-src/skynet_monitor.c @@ -9,7 +9,7 @@ #include struct skynet_monitor { - int version; + ATOM_INT version; int check_version; uint32_t source; uint32_t destination; @@ -31,7 +31,7 @@ void skynet_monitor_trigger(struct skynet_monitor *sm, uint32_t source, uint32_t destination) { sm->source = source; sm->destination = destination; - ATOM_INC(&sm->version); + ATOM_FINC(&sm->version); } void diff --git a/skynet-src/skynet_server.c b/skynet-src/skynet_server.c index 9ee42300..d0cc3fdd 100644 --- a/skynet-src/skynet_server.c +++ b/skynet-src/skynet_server.c @@ -52,7 +52,7 @@ struct skynet_context { char result[32]; uint32_t handle; int session_id; - int ref; + ATOM_INT ref; int message_count; bool init; bool endless; @@ -62,7 +62,7 @@ struct skynet_context { }; struct skynet_node { - int total; + ATOM_INT total; int init; uint32_t monitor_exit; pthread_key_t handle_key; @@ -73,17 +73,17 @@ static struct skynet_node G_NODE; int skynet_context_total() { - return G_NODE.total; + return ATOM_LOAD(&G_NODE.total); } static void context_inc() { - ATOM_INC(&G_NODE.total); + ATOM_FINC(&G_NODE.total); } static void context_dec() { - ATOM_DEC(&G_NODE.total); + ATOM_FDEC(&G_NODE.total); } uint32_t @@ -137,11 +137,11 @@ skynet_context_new(const char * name, const char *param) { ctx->mod = mod; ctx->instance = inst; - ctx->ref = 2; + ATOM_INIT(&ctx->ref , 2); ctx->cb = NULL; ctx->cb_ud = NULL; ctx->session_id = 0; - ctx->logfile = NULL; + ATOM_INIT(&ctx->logfile, NULL); ctx->init = false; ctx->endless = false; @@ -194,7 +194,7 @@ skynet_context_newsession(struct skynet_context *ctx) { void skynet_context_grab(struct skynet_context *ctx) { - ATOM_INC(&ctx->ref); + ATOM_FINC(&ctx->ref); } void @@ -207,8 +207,9 @@ skynet_context_reserve(struct skynet_context *ctx) { static void delete_context(struct skynet_context *ctx) { - if (ctx->logfile) { - fclose(ctx->logfile); + FILE *f = ATOM_LOAD(&ctx->logfile); + if (f) { + fclose(f); } skynet_module_instance_release(ctx->mod, ctx->instance); skynet_mq_mark_release(ctx->queue); @@ -219,7 +220,7 @@ delete_context(struct skynet_context *ctx) { struct skynet_context * skynet_context_release(struct skynet_context *ctx) { - if (ATOM_DEC(&ctx->ref) == 0) { + if (ATOM_FDEC(&ctx->ref) == 1) { delete_context(ctx); return NULL; } @@ -264,8 +265,9 @@ dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) { pthread_setspecific(G_NODE.handle_key, (void *)(uintptr_t)(ctx->handle)); int type = msg->sz >> MESSAGE_TYPE_SHIFT; size_t sz = msg->sz & MESSAGE_TYPE_MASK; - if (ctx->logfile) { - skynet_log_output(ctx->logfile, msg->source, type, msg->session, msg->data, sz); + FILE *f = ATOM_LOAD(&ctx->logfile); + if (f) { + skynet_log_output(f, msg->source, type, msg->session, msg->data, sz); } ++ctx->message_count; int reserve_msg; @@ -589,11 +591,13 @@ cmd_logon(struct skynet_context * context, const char * param) { if (ctx == NULL) return NULL; FILE *f = NULL; - FILE * lastf = ctx->logfile; + FILE * lastf = ATOM_LOAD(&ctx->logfile); if (lastf == NULL) { f = skynet_log_open(context, handle); if (f) { - if (!ATOM_CAS_POINTER(&ctx->logfile, NULL, f)) { + ATOM_POINTER exp; + ATOM_INIT(&exp, 0); + if (!ATOM_CAS_POINTER(&ctx->logfile, exp, f)) { // logfile opens in other thread, close this one. fclose(f); } @@ -611,7 +615,7 @@ cmd_logoff(struct skynet_context * context, const char * param) { struct skynet_context * ctx = skynet_handle_grab(handle); if (ctx == NULL) return NULL; - FILE * f = ctx->logfile; + FILE * f = ATOM_LOAD(&ctx->logfile); if (f) { // logfile may close in other thread if (ATOM_CAS_POINTER(&ctx->logfile, f, NULL)) { @@ -806,7 +810,7 @@ skynet_context_send(struct skynet_context * ctx, void * msg, size_t sz, uint32_t void skynet_globalinit(void) { - G_NODE.total = 0; + ATOM_INIT(&G_NODE.total , 0); G_NODE.monitor_exit = 0; G_NODE.init = 1; if (pthread_key_create(&G_NODE.handle_key, NULL)) { diff --git a/skynet-src/socket_server.c b/skynet-src/socket_server.c index c3a221d1..c57e26bc 100644 --- a/skynet-src/socket_server.c +++ b/skynet-src/socket_server.c @@ -95,7 +95,7 @@ struct socket { int fd; int id; uint8_t protocol; - uint8_t type; + ATOM_BYTE type; bool reading; bool writing; int udpconnecting; @@ -116,7 +116,7 @@ struct socket_server { int sendctrl_fd; int checkctrl; poll_fd event_fd; - int alloc_id; + ATOM_INT alloc_id; int event_n; int event_index; struct socket_object_interface soi; @@ -276,6 +276,11 @@ socket_unlock(struct socket_lock *sl) { } } +static inline int +socket_invalid(struct socket *s, int id) { + return (s->id != id || ATOM_LOAD(&s->type) == SOCKET_TYPE_INVALID); +} + static inline bool send_object_init(struct socket_server *ss, struct send_object *so, const void *object, size_t sz) { if (sz == USEROBJECT) { @@ -339,13 +344,14 @@ static int reserve_id(struct socket_server *ss) { int i; for (i=0;ialloc_id)); + int id = ATOM_FINC(&(ss->alloc_id))+1; if (id < 0) { - id = ATOM_AND(&(ss->alloc_id), 0x7fffffff); + id = ATOM_FAND(&(ss->alloc_id), 0x7fffffff) & 0x7fffffff; } struct socket *s = &ss->slot[HASH_ID(id)]; - if (s->type == SOCKET_TYPE_INVALID) { - if (ATOM_CAS(&s->type, SOCKET_TYPE_INVALID, SOCKET_TYPE_RESERVE)) { + unsigned char type_invalid = ATOM_LOAD(&s->type); + if (type_invalid == SOCKET_TYPE_INVALID) { + if (ATOM_CAS(&s->type, type_invalid, SOCKET_TYPE_RESERVE)) { s->id = id; s->protocol = PROTOCOL_UNKNOWN; // socket_server_udp_connect may inc s->udpconncting directly (from other thread, before new_fd), @@ -400,12 +406,12 @@ socket_server_create(uint64_t time) { for (i=0;islot[i]; - s->type = SOCKET_TYPE_INVALID; + ATOM_INIT(&s->type, SOCKET_TYPE_INVALID); clear_wb_list(&s->high); clear_wb_list(&s->low); spinlock_init(&s->dw_lock); } - ss->alloc_id = 0; + ATOM_INIT(&ss->alloc_id , 0); ss->event_n = 0; ss->event_index = 0; memset(&ss->soi, 0, sizeof(ss->soi)); @@ -474,20 +480,21 @@ force_close(struct socket_server *ss, struct socket *s, struct socket_lock *l, s result->ud = 0; result->data = NULL; result->opaque = s->opaque; - if (s->type == SOCKET_TYPE_INVALID) { + uint8_t type = ATOM_LOAD(&s->type); + if (type == SOCKET_TYPE_INVALID) { return; } - assert(s->type != SOCKET_TYPE_RESERVE); + assert(type != SOCKET_TYPE_RESERVE); free_wb_list(ss,&s->high); free_wb_list(ss,&s->low); sp_del(ss->event_fd, s->fd); socket_lock(l); - if (s->type != SOCKET_TYPE_BIND) { + if (type != SOCKET_TYPE_BIND) { if (close(s->fd) < 0) { perror("close socket:"); } } - s->type = SOCKET_TYPE_INVALID; + ATOM_STORE(&s->type, SOCKET_TYPE_INVALID); if (s->dw_buffer) { struct socket_sendbuffer tmp; tmp.buffer = s->dw_buffer; @@ -508,7 +515,7 @@ socket_server_release(struct socket_server *ss) { struct socket *s = &ss->slot[i]; struct socket_lock l; socket_lock_init(s, &l); - if (s->type != SOCKET_TYPE_RESERVE) { + if (ATOM_LOAD(&s->type) != SOCKET_TYPE_RESERVE) { force_close(ss, s, &l, &dummy); } spinlock_destroy(&s->dw_lock); @@ -546,10 +553,10 @@ enable_read(struct socket_server *ss, struct socket *s, bool enable) { static struct socket * new_fd(struct socket_server *ss, int id, int fd, int protocol, uintptr_t opaque, bool reading) { struct socket * s = &ss->slot[HASH_ID(id)]; - assert(s->type == SOCKET_TYPE_RESERVE); + assert(ATOM_LOAD(&s->type) == SOCKET_TYPE_RESERVE); if (sp_add(ss->event_fd, fd, s)) { - s->type = SOCKET_TYPE_INVALID; + ATOM_STORE(&s->type, SOCKET_TYPE_INVALID); return NULL; } @@ -569,7 +576,7 @@ new_fd(struct socket_server *ss, int id, int fd, int protocol, uintptr_t opaque, s->dw_size = 0; memset(&s->stat, 0, sizeof(s->stat)); if (enable_read(ss, s, reading)) { - s->type = SOCKET_TYPE_INVALID; + ATOM_STORE(&s->type , SOCKET_TYPE_INVALID); return NULL; } return s; @@ -642,7 +649,7 @@ open_socket(struct socket_server *ss, struct request_open * request, struct sock } if(status == 0) { - ns->type = SOCKET_TYPE_CONNECTED; + ATOM_STORE(&ns->type , SOCKET_TYPE_CONNECTED); struct sockaddr * addr = ai_ptr->ai_addr; void * sin_addr = (ai_ptr->ai_family == AF_INET) ? (void*)&((struct sockaddr_in *)addr)->sin_addr : (void*)&((struct sockaddr_in6 *)addr)->sin6_addr; if (inet_ntop(ai_ptr->ai_family, sin_addr, ss->buffer, sizeof(ss->buffer))) { @@ -651,7 +658,7 @@ open_socket(struct socket_server *ss, struct request_open * request, struct sock freeaddrinfo( ai_list ); return SOCKET_OPEN; } else { - ns->type = SOCKET_TYPE_CONNECTING; + ATOM_STORE(&ns->type , SOCKET_TYPE_CONNECTING); if (enable_write(ss, ns, true)) { result->data = "enable write failed"; goto _failed; @@ -838,7 +845,7 @@ send_buffer_(struct socket_server *ss, struct socket *s, struct socket_lock *l, assert(send_buffer_empty(s) && s->wb_size == 0); int err = enable_write(ss, s, false); - if (s->type == SOCKET_TYPE_HALFCLOSE) { + if (ATOM_LOAD(&s->type) == SOCKET_TYPE_HALFCLOSE) { force_close(ss, s, l, result); return SOCKET_CLOSE; } @@ -936,7 +943,7 @@ static int trigger_write(struct socket_server *ss, struct request_send * request, struct socket_message *result) { int id = request->id; struct socket * s = &ss->slot[HASH_ID(id)]; - if (s->type == SOCKET_TYPE_INVALID || s->id != id) + if (socket_invalid(s, id)) return -1; if (enable_write(ss, s, true)) { result->opaque = s->opaque; @@ -961,18 +968,19 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock struct socket * s = &ss->slot[HASH_ID(id)]; struct send_object so; send_object_init(ss, &so, request->buffer, request->sz); - if (s->type == SOCKET_TYPE_INVALID || s->id != id - || s->type == SOCKET_TYPE_HALFCLOSE - || s->type == SOCKET_TYPE_PACCEPT) { + uint8_t type = ATOM_LOAD(&s->type); + if (type == SOCKET_TYPE_INVALID || s->id != id + || type == SOCKET_TYPE_HALFCLOSE + || type == SOCKET_TYPE_PACCEPT) { so.free_func((void *)request->buffer); return -1; } - if (s->type == SOCKET_TYPE_PLISTEN || s->type == SOCKET_TYPE_LISTEN) { + if (type == SOCKET_TYPE_PLISTEN || type == SOCKET_TYPE_LISTEN) { skynet_error(NULL, "socket-server: write to listen fd %d.", id); so.free_func((void *)request->buffer); return -1; } - if (send_buffer_empty(s) && s->type == SOCKET_TYPE_CONNECTED) { + if (send_buffer_empty(s) && type == SOCKET_TYPE_CONNECTED) { if (s->protocol == PROTOCOL_TCP) { append_sendbuffer(ss, s, request); // add to high priority list, even priority == PRIORITY_LOW } else { @@ -1037,7 +1045,7 @@ listen_socket(struct socket_server *ss, struct request_listen * request, struct if (s == NULL) { goto _failed; } - s->type = SOCKET_TYPE_PLISTEN; + ATOM_STORE(&s->type , SOCKET_TYPE_PLISTEN); return -1; _failed: close(listen_fd); @@ -1059,7 +1067,7 @@ static int close_socket(struct socket_server *ss, struct request_close *request, struct socket_message *result) { int id = request->id; struct socket * s = &ss->slot[HASH_ID(id)]; - if (s->type == SOCKET_TYPE_INVALID || s->id != id) { + if (socket_invalid(s, id)) { result->id = id; result->opaque = request->opaque; result->ud = 0; @@ -1081,7 +1089,7 @@ close_socket(struct socket_server *ss, struct request_close *request, struct soc result->opaque = request->opaque; return SOCKET_CLOSE; } - s->type = SOCKET_TYPE_HALFCLOSE; + ATOM_STORE(&s->type , SOCKET_TYPE_HALFCLOSE); shutdown(s->fd, SHUT_RD); return -1; @@ -1099,7 +1107,7 @@ bind_socket(struct socket_server *ss, struct request_bind *request, struct socke return SOCKET_ERR; } sp_nonblocking(request->fd); - s->type = SOCKET_TYPE_BIND; + ATOM_STORE(&s->type , SOCKET_TYPE_BIND); result->data = "binding"; return SOCKET_OPEN; } @@ -1112,7 +1120,7 @@ resume_socket(struct socket_server *ss, struct request_resumepause *request, str result->ud = 0; result->data = NULL; struct socket *s = &ss->slot[HASH_ID(id)]; - if (s->type == SOCKET_TYPE_INVALID || s->id !=id) { + if (socket_invalid(s, id)) { result->data = "invalid socket"; return SOCKET_ERR; } @@ -1122,12 +1130,13 @@ resume_socket(struct socket_server *ss, struct request_resumepause *request, str result->data = "enable read failed"; return SOCKET_ERR; } - if (s->type == SOCKET_TYPE_PACCEPT || s->type == SOCKET_TYPE_PLISTEN) { - s->type = (s->type == SOCKET_TYPE_PACCEPT) ? SOCKET_TYPE_CONNECTED : SOCKET_TYPE_LISTEN; + uint8_t type = ATOM_LOAD(&s->type); + if (type == SOCKET_TYPE_PACCEPT || type == SOCKET_TYPE_PLISTEN) { + ATOM_STORE(&s->type , (type == SOCKET_TYPE_PACCEPT) ? SOCKET_TYPE_CONNECTED : SOCKET_TYPE_LISTEN); s->opaque = request->opaque; result->data = "start"; return SOCKET_OPEN; - } else if (s->type == SOCKET_TYPE_CONNECTED) { + } else if (type == SOCKET_TYPE_CONNECTED) { // todo: maybe we should send a message SOCKET_TRANSFER to s->opaque s->opaque = request->opaque; result->data = "transfer"; @@ -1141,7 +1150,7 @@ static int pause_socket(struct socket_server *ss, struct request_resumepause *request, struct socket_message *result) { int id = request->id; struct socket *s = &ss->slot[HASH_ID(id)]; - if (s->type == SOCKET_TYPE_INVALID || s->id !=id) { + if (socket_invalid(s, id)) { return -1; } if (enable_read(ss, s, false)) { @@ -1158,7 +1167,7 @@ static void setopt_socket(struct socket_server *ss, struct request_setopt *request) { int id = request->id; struct socket *s = &ss->slot[HASH_ID(id)]; - if (s->type == SOCKET_TYPE_INVALID || s->id !=id) { + if (socket_invalid(s, id)) { return; } int v = request->value; @@ -1210,7 +1219,7 @@ add_udp_socket(struct socket_server *ss, struct request_udp *udp) { ss->slot[HASH_ID(id)].type = SOCKET_TYPE_INVALID; return; } - ns->type = SOCKET_TYPE_CONNECTED; + ATOM_STORE(&ns->type , SOCKET_TYPE_CONNECTED); memset(ns->p.udp_address, 0, sizeof(ns->p.udp_address)); } @@ -1218,7 +1227,7 @@ static int set_udp_address(struct socket_server *ss, struct request_setudp *request, struct socket_message *result) { int id = request->id; struct socket *s = &ss->slot[HASH_ID(id)]; - if (s->type == SOCKET_TYPE_INVALID || s->id !=id) { + if (socket_invalid(s, id)) { return -1; } int type = request->address[0]; @@ -1236,7 +1245,7 @@ set_udp_address(struct socket_server *ss, struct request_setudp *request, struct } else { memcpy(s->p.udp_address, request->address, 1+2+16); // 1 type, 2 port, 16 ipv6 } - ATOM_DEC(&s->udpconnecting); + ATOM_FDEC(&s->udpconnecting); return -1; } @@ -1268,7 +1277,7 @@ dec_sending_ref(struct socket_server *ss, int id) { // Notice: udp may inc sending while type == SOCKET_TYPE_RESERVE if (s->id == id && s->protocol == PROTOCOL_TCP) { assert((s->sending & 0xffff) != 0); - ATOM_DEC(&s->sending); + ATOM_FDEC(&s->sending); } } @@ -1361,13 +1370,13 @@ forward_message_tcp(struct socket_server *ss, struct socket *s, struct socket_lo force_close(ss,s,l,result); return SOCKET_CLOSE; } else { - s->type = SOCKET_TYPE_HALFCLOSE; + ATOM_STORE(&s->type , SOCKET_TYPE_HALFCLOSE); shutdown(s->fd, SHUT_RD); return -1; } } - if (s->type == SOCKET_TYPE_HALFCLOSE) { + if (ATOM_LOAD(&s->type) == SOCKET_TYPE_HALFCLOSE) { // discard recv data FREE(buffer); return -1; @@ -1461,7 +1470,7 @@ report_connect(struct socket_server *ss, struct socket *s, struct socket_lock *l result->data = strerror(errno); return SOCKET_ERR; } else { - s->type = SOCKET_TYPE_CONNECTED; + ATOM_STORE(&s->type , SOCKET_TYPE_CONNECTED); result->opaque = s->opaque; result->id = s->id; result->ud = 0; @@ -1532,7 +1541,7 @@ report_accept(struct socket_server *ss, struct socket *s, struct socket_message // accept new one connection stat_read(ss,s,1); - ns->type = SOCKET_TYPE_PACCEPT; + ATOM_STORE(&ns->type , SOCKET_TYPE_PACCEPT); result->opaque = s->opaque; result->id = s->id; result->ud = id; @@ -1554,7 +1563,7 @@ clear_closed_event(struct socket_server *ss, struct socket_message * result, int struct event *e = &ss->ev[i]; struct socket *s = e->s; if (s) { - if (s->type == SOCKET_TYPE_INVALID && s->id == id) { + if (socket_invalid(s, id)) { e->s = NULL; break; } @@ -1602,7 +1611,7 @@ socket_server_poll(struct socket_server *ss, struct socket_message * result, int } struct socket_lock l; socket_lock_init(s, &l); - switch (s->type) { + switch (ATOM_LOAD(&s->type)) { case SOCKET_TYPE_CONNECTING: return report_connect(ss, s, &l, result); case SOCKET_TYPE_LISTEN: { @@ -1720,7 +1729,7 @@ socket_server_connect(struct socket_server *ss, uintptr_t opaque, const char * a static inline int can_direct_write(struct socket *s, int id) { - return s->id == id && nomore_sending_data(s) && s->type == SOCKET_TYPE_CONNECTED && s->udpconnecting == 0; + return s->id == id && nomore_sending_data(s) && ATOM_LOAD(&s->type) == SOCKET_TYPE_CONNECTED && s->udpconnecting == 0; } // return -1 when error, 0 when success @@ -1728,7 +1737,7 @@ int socket_server_send(struct socket_server *ss, struct socket_sendbuffer *buf) { int id = buf->id; struct socket * s = &ss->slot[HASH_ID(id)]; - if (s->id != id || s->type == SOCKET_TYPE_INVALID) { + if (socket_invalid(s, id)) { free_buffer(ss, buf); return -1; } @@ -1804,7 +1813,7 @@ socket_server_send_lowpriority(struct socket_server *ss, struct socket_sendbuffe int id = buf->id; struct socket * s = &ss->slot[HASH_ID(id)]; - if (s->id != id || s->type == SOCKET_TYPE_INVALID) { + if (socket_invalid(s, id)) { free_buffer(ss, buf); return -1; } @@ -2009,7 +2018,7 @@ int socket_server_udp_send(struct socket_server *ss, const struct socket_udp_address *addr, struct socket_sendbuffer *buf) { int id = buf->id; struct socket * s = &ss->slot[HASH_ID(id)]; - if (s->id != id || s->type == SOCKET_TYPE_INVALID) { + if (socket_invalid(s, id)) { free_buffer(ss, buf); return -1; } @@ -2070,17 +2079,17 @@ socket_server_udp_send(struct socket_server *ss, const struct socket_udp_address int socket_server_udp_connect(struct socket_server *ss, int id, const char * addr, int port) { struct socket * s = &ss->slot[HASH_ID(id)]; - if (s->id != id || s->type == SOCKET_TYPE_INVALID) { + if (socket_invalid(s, id)) { return -1; } struct socket_lock l; socket_lock_init(s, &l); socket_lock(&l); - if (s->id != id || s->type == SOCKET_TYPE_INVALID) { + if (socket_invalid(s, id)) { socket_unlock(&l); return -1; } - ATOM_INC(&s->udpconnecting); + ATOM_FINC(&s->udpconnecting); socket_unlock(&l); int status; @@ -2158,7 +2167,7 @@ static int query_info(struct socket *s, struct socket_info *si) { union sockaddr_all u; socklen_t slen = sizeof(u); - switch (s->type) { + switch (ATOM_LOAD(&s->type)) { case SOCKET_TYPE_BIND: si->type = SOCKET_INFO_BIND; si->name[0] = '\0'; diff --git a/skynet-src/spinlock.h b/skynet-src/spinlock.h index 514546f4..27ef462a 100644 --- a/skynet-src/spinlock.h +++ b/skynet-src/spinlock.h @@ -8,28 +8,46 @@ #ifndef USE_PTHREAD_LOCK +#ifdef NO_STDATOMIC + +#define atomic_flag_ int +#define ATOMIC_FLAG_INIT_ 0 +#define atomic_flag_test_and_set_(ptr) __sync_lock_test_and_set(ptr, 1) +#define atomic_flag_clear_(ptr) __sync_lock_release(ptr) + +#else + +#include +#define atomic_flag_ atomic_flag +#define ATOMIC_FLAG_INIT_ ATOMIC_FLAG_INIT +#define atomic_flag_test_and_set_ atomic_flag_test_and_set +#define atomic_flag_clear_ atomic_flag_clear + +#endif + struct spinlock { - int lock; + atomic_flag_ lock; }; static inline void spinlock_init(struct spinlock *lock) { - lock->lock = 0; + atomic_flag_ v = ATOMIC_FLAG_INIT_; + lock->lock = v; } static inline void spinlock_lock(struct spinlock *lock) { - while (__sync_lock_test_and_set(&lock->lock,1)) {} + while (atomic_flag_test_and_set_(&lock->lock)) {} } static inline int spinlock_trylock(struct spinlock *lock) { - return __sync_lock_test_and_set(&lock->lock,1) == 0; + return atomic_flag_test_and_set_(&lock->lock) == 0; } static inline void spinlock_unlock(struct spinlock *lock) { - __sync_lock_release(&lock->lock); + atomic_flag_clear_(&lock->lock); } static inline void