lihao/skynet
1
0
Fork 0
mirror of https://github.com/cloudwu/skynet.git synced 2026-07-25 12:43:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use _Atomic type for socket.sending and socket.udpconnecting. See #1317

Browse Source
This commit is contained in:
Cloud Wu
2021-01-11 16:57:25 +08:00
parent c0e3421462
commit 1ed4fdf94d
4 changed files with 13 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
service-src/service_snlua.c
View File

@@ -30,7 +30,7 @@ struct snlua {
size_t mem_report; size_t mem_report;
size_t mem_limit; size_t mem_limit;
lua_State * activeL; lua_State * activeL;
volatile ATOM_INT trap; ATOM_INT trap;
}; };
// LUA_CACHELIB may defined in patched lua for shared proto // LUA_CACHELIB may defined in patched lua for shared proto

4
skynet-src/atomic.h
View File

@@ -1,12 +1,13 @@
#ifndef SKYNET_ATOMIC_H #ifndef SKYNET_ATOMIC_H
#define SKYNET_ATOMIC_H #define SKYNET_ATOMIC_H
#ifdef NO_STDATOMIC #ifdef __STDC_NO_ATOMICS__
#define ATOM_BYTE unsigned char #define ATOM_BYTE unsigned char
#define ATOM_INT int #define ATOM_INT int
#define ATOM_POINTER void * #define ATOM_POINTER void *
#define ATOM_SIZET size_t #define ATOM_SIZET size_t
#define ATOM_ULONG unsigned long
#define ATOM_INIT(ptr, v) (*(ptr) = v) #define ATOM_INIT(ptr, v) (*(ptr) = v)
#define ATOM_LOAD(ptr) (*(ptr)) #define ATOM_LOAD(ptr) (*(ptr))
#define ATOM_STORE(ptr, v) (*(ptr) = v) #define ATOM_STORE(ptr, v) (*(ptr) = v)
@@ -26,6 +27,7 @@
#define ATOM_INT atomic_int #define ATOM_INT atomic_int
#define ATOM_POINTER atomic_uintptr_t #define ATOM_POINTER atomic_uintptr_t
#define ATOM_SIZET atomic_size_t #define ATOM_SIZET atomic_size_t
#define ATOM_ULONG atomic_ulong
#define ATOM_INIT(ref, v) atomic_init(ref, v) #define ATOM_INIT(ref, v) atomic_init(ref, v)
#define ATOM_LOAD(ptr) atomic_load(ptr) #define ATOM_LOAD(ptr) atomic_load(ptr)
#define ATOM_STORE(ptr, v) atomic_store(ptr, v) #define ATOM_STORE(ptr, v) atomic_store(ptr, v)

16
skynet-src/socket_server.c
View File

@@ -91,14 +91,14 @@ struct socket {
struct wb_list low; struct wb_list low;
int64_t wb_size; int64_t wb_size;
struct socket_stat stat; struct socket_stat stat;
volatile uint32_t sending; ATOM_ULONG sending;
int fd; int fd;
int id; int id;
uint8_t protocol; uint8_t protocol;
ATOM_BYTE type; ATOM_BYTE type;
bool reading; bool reading;
bool writing; bool writing;
int udpconnecting; ATOM_INT udpconnecting;
int64_t warn_size; int64_t warn_size;
union { union {
int size; int size;
@@ -356,7 +356,7 @@ reserve_id(struct socket_server *ss) {
s->protocol = PROTOCOL_UNKNOWN; s->protocol = PROTOCOL_UNKNOWN;
// socket_server_udp_connect may inc s->udpconncting directly (from other thread, before new_fd), // socket_server_udp_connect may inc s->udpconncting directly (from other thread, before new_fd),
// so reset it to 0 here rather than in new_fd. // so reset it to 0 here rather than in new_fd.
s->udpconnecting = 0; ATOM_INIT(&s->udpconnecting, 0);
s->fd = -1; s->fd = -1;
return id; return id;
} else { } else {
@@ -564,7 +564,7 @@ new_fd(struct socket_server *ss, int id, int fd, int protocol, uintptr_t opaque,
s->fd = fd; s->fd = fd;
s->reading = true; s->reading = true;
s->writing = false; s->writing = false;
s->sending = ID_TAG16(id) << 16 | 0; ATOM_INIT(&s->sending , ID_TAG16(id) << 16 | 0);
s->protocol = protocol; s->protocol = protocol;
s->p.size = MIN_READ_BUFFER; s->p.size = MIN_READ_BUFFER;
s->opaque = opaque; s->opaque = opaque;
@@ -1060,7 +1060,7 @@ _failed:
static inline int static inline int
nomore_sending_data(struct socket *s) { nomore_sending_data(struct socket *s) {
return send_buffer_empty(s) && s->dw_buffer == NULL && (s->sending & 0xffff) == 0; return send_buffer_empty(s) && s->dw_buffer == NULL && (ATOM_LOAD(&s->sending) & 0xffff) == 0;
} }
static int static int
@@ -1254,7 +1254,7 @@ inc_sending_ref(struct socket *s, int id) {
if (s->protocol != PROTOCOL_TCP) if (s->protocol != PROTOCOL_TCP)
return; return;
for (;;) { for (;;) {
uint32_t sending = s->sending; unsigned long sending = ATOM_LOAD(&s->sending);
if ((sending >> 16) == ID_TAG16(id)) { if ((sending >> 16) == ID_TAG16(id)) {
if ((sending & 0xffff) == 0xffff) { if ((sending & 0xffff) == 0xffff) {
// s->sending may overflow (rarely), so busy waiting here for socket thread dec it. see issue #794 // s->sending may overflow (rarely), so busy waiting here for socket thread dec it. see issue #794
@@ -1276,7 +1276,7 @@ dec_sending_ref(struct socket_server *ss, int id) {
struct socket * s = &ss->slot[HASH_ID(id)]; struct socket * s = &ss->slot[HASH_ID(id)];
// Notice: udp may inc sending while type == SOCKET_TYPE_RESERVE // Notice: udp may inc sending while type == SOCKET_TYPE_RESERVE
if (s->id == id && s->protocol == PROTOCOL_TCP) { if (s->id == id && s->protocol == PROTOCOL_TCP) {
assert((s->sending & 0xffff) != 0); assert((ATOM_LOAD(&s->sending) & 0xffff) != 0);
ATOM_FDEC(&s->sending); ATOM_FDEC(&s->sending);
} }
} }
@@ -1729,7 +1729,7 @@ socket_server_connect(struct socket_server *ss, uintptr_t opaque, const char * a
static inline int static inline int
can_direct_write(struct socket *s, int id) { can_direct_write(struct socket *s, int id) {
return s->id == id && nomore_sending_data(s) && ATOM_LOAD(&s->type) == SOCKET_TYPE_CONNECTED && s->udpconnecting == 0; return s->id == id && nomore_sending_data(s) && ATOM_LOAD(&s->type) == SOCKET_TYPE_CONNECTED && ATOM_LOAD(&s->udpconnecting) == 0;
} }
// return -1 when error, 0 when success // return -1 when error, 0 when success

2
skynet-src/spinlock.h
View File

@@ -8,7 +8,7 @@
#ifndef USE_PTHREAD_LOCK #ifndef USE_PTHREAD_LOCK
#ifdef NO_STDATOMIC #ifdef __STDC_NO_ATOMICS__
#define atomic_flag_ int #define atomic_flag_ int
#define ATOMIC_FLAG_INIT_ 0 #define ATOMIC_FLAG_INIT_ 0