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

Add socket_sendbuffer

Browse Source
This commit is contained in:
Cloud Wu
2019-11-05 16:23:11 +08:00
committed by 云风
parent 58291a2a48
commit 9d7ea09789
7 changed files with 225 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
skynet-src/skynet_socket.c
View File

@@ -117,13 +117,13 @@ skynet_socket_poll() {
}
int
skynet_socket_send(struct skynet_context *ctx, int id, void *buffer, int sz) {
return socket_server_send(SOCKET_SERVER, id, buffer, sz);
skynet_socket_sendbuffer(struct skynet_context *ctx, struct socket_sendbuffer *buffer) {
return socket_server_send(SOCKET_SERVER, buffer);
}
int
skynet_socket_send_lowpriority(struct skynet_context *ctx, int id, void *buffer, int sz) {
return socket_server_send_lowpriority(SOCKET_SERVER, id, buffer, sz);
skynet_socket_sendbuffer_lowpriority(struct skynet_context *ctx, struct socket_sendbuffer *buffer) {
return socket_server_send_lowpriority(SOCKET_SERVER, buffer);
}
int
@@ -179,8 +179,8 @@ skynet_socket_udp_connect(struct skynet_context *ctx, int id, const char * addr,
}
int
skynet_socket_udp_send(struct skynet_context *ctx, int id, const char * address, const void *buffer, int sz) {
return socket_server_udp_send(SOCKET_SERVER, id, (const struct socket_udp_address *)address, buffer, sz);
skynet_socket_udp_sendbuffer(struct skynet_context *ctx, const char * address, struct socket_sendbuffer *buffer) {
return socket_server_udp_send(SOCKET_SERVER, (const struct socket_udp_address *)address, buffer);
}
const char *

38
skynet-src/skynet_socket.h
View File

@@ -2,6 +2,7 @@
#define skynet_socket_h
#include "socket_info.h"
#include "socket_buffer.h"
struct skynet_context;
@@ -26,8 +27,8 @@ void skynet_socket_free();
int skynet_socket_poll();
void skynet_socket_updatetime();
int skynet_socket_send(struct skynet_context *ctx, int id, void *buffer, int sz);
int skynet_socket_send_lowpriority(struct skynet_context *ctx, int id, void *buffer, int sz);
int skynet_socket_sendbuffer(struct skynet_context *ctx, struct socket_sendbuffer *buffer);
int skynet_socket_sendbuffer_lowpriority(struct skynet_context *ctx, struct socket_sendbuffer *buffer);
int skynet_socket_listen(struct skynet_context *ctx, const char *host, int port, int backlog);
int skynet_socket_connect(struct skynet_context *ctx, const char *host, int port);
int skynet_socket_bind(struct skynet_context *ctx, int fd);
@@ -38,9 +39,40 @@ void skynet_socket_nodelay(struct skynet_context *ctx, int id);
int skynet_socket_udp(struct skynet_context *ctx, const char * addr, int port);
int skynet_socket_udp_connect(struct skynet_context *ctx, int id, const char * addr, int port);
int skynet_socket_udp_send(struct skynet_context *ctx, int id, const char * address, const void *buffer, int sz);
int skynet_socket_udp_sendbuffer(struct skynet_context *ctx, const char * address, struct socket_sendbuffer *buffer);
const char * skynet_socket_udp_address(struct skynet_socket_message *, int *addrsz);
struct socket_info * skynet_socket_info();
// legacy APIs
static inline void sendbuffer_init_(struct socket_sendbuffer *buf, int id, const void *buffer, int sz) {
buf->id = id;
buf->buffer = buffer;
if (sz < 0) {
buf->type = SOCKET_BUFFER_OBJECT;
} else {
buf->type = SOCKET_BUFFER_MEMORY;
}
buf->sz = (size_t)sz;
}
static inline int skynet_socket_send(struct skynet_context *ctx, int id, void *buffer, int sz) {
struct socket_sendbuffer tmp;
sendbuffer_init_(&tmp, id, buffer, sz);
return skynet_socket_sendbuffer(ctx, &tmp);
}
static inline int skynet_socket_send_lowpriority(struct skynet_context *ctx, int id, void *buffer, int sz) {
struct socket_sendbuffer tmp;
sendbuffer_init_(&tmp, id, buffer, sz);
return skynet_socket_sendbuffer_lowpriority(ctx, &tmp);
}
static inline int skynet_socket_udp_send(struct skynet_context *ctx, int id, const char * address, const void *buffer, int sz) {
struct socket_sendbuffer tmp;
sendbuffer_init_(&tmp, id, buffer, sz);
return skynet_socket_udp_sendbuffer(ctx, address, &tmp);
}
#endif

17
skynet-src/socket_buffer.h Normal file
View File

@@ -0,0 +1,17 @@
#ifndef socket_buffer_h
#define socket_buffer_h
#include <stdlib.h>
#define SOCKET_BUFFER_MEMORY 0
#define SOCKET_BUFFER_OBJECT 1
#define SOCKET_BUFFER_RAWPOINTER 2
struct socket_sendbuffer {
int id;
int type;
const void *buffer;
size_t sz;
};
#endif

144
skynet-src/socket_server.c
View File

@@ -58,11 +58,13 @@
#define WARNING_SIZE (1024*1024)
#define USEROBJECT ((size_t)(~0))
struct write_buffer {
struct write_buffer * next;
void *buffer;
const void *buffer;
char *ptr;
int sz;
size_t sz;
bool userobject;
uint8_t udp_address[UDP_ADDRESS_SIZE];
};
@@ -131,8 +133,8 @@ struct request_open {
struct request_send {
int id;
int sz;
char * buffer;
size_t sz;
const void * buffer;
};
struct request_send_udp {
@@ -225,8 +227,8 @@ union sockaddr_all {
};
struct send_object {
void * buffer;
int sz;
const void * buffer;
size_t sz;
void (*free_func)(void *);
};
@@ -272,8 +274,8 @@ socket_unlock(struct socket_lock *sl) {
}
static inline bool
send_object_init(struct socket_server *ss, struct send_object *so, void *object, int sz) {
if (sz < 0) {
send_object_init(struct socket_server *ss, struct send_object *so, const void *object, size_t sz) {
if (sz == USEROBJECT) {
so->buffer = ss->soi.buffer(object);
so->sz = ss->soi.size(object);
so->free_func = ss->soi.free;
@@ -286,12 +288,40 @@ send_object_init(struct socket_server *ss, struct send_object *so, void *object,
}
}
static void
dummy_free(void *ptr) {
(void)ptr;
}
static inline void
send_object_init_from_sendbuffer(struct socket_server *ss, struct send_object *so, struct socket_sendbuffer *buf) {
switch (buf->type) {
case SOCKET_BUFFER_MEMORY:
send_object_init(ss, so, (void *)buf->buffer, buf->sz);
break;
case SOCKET_BUFFER_OBJECT:
send_object_init(ss, so, (void *)buf->buffer, USEROBJECT);
break;
case SOCKET_BUFFER_RAWPOINTER:
so->buffer = (void *)buf->buffer;
so->sz = buf->sz;
so->free_func = dummy_free;
break;
default:
// never get here
so->buffer = NULL;
so->sz = 0;
so->free_func = NULL;
break;
}
}
static inline void
write_buffer_free(struct socket_server *ss, struct write_buffer *wb) {
if (wb->userobject) {
ss->soi.free(wb->buffer);
ss->soi.free((void *)wb->buffer);
} else {
FREE(wb->buffer);
FREE((void *)wb->buffer);
}
FREE(wb);
}
@@ -400,10 +430,39 @@ free_wb_list(struct socket_server *ss, struct wb_list *list) {
}
static void
free_buffer(struct socket_server *ss, const void * buffer, int sz) {
struct send_object so;
send_object_init(ss, &so, (void *)buffer, sz);
so.free_func((void *)buffer);
free_buffer(struct socket_server *ss, struct socket_sendbuffer *buf) {
void *buffer = (void *)buf->buffer;
switch (buf->type) {
case SOCKET_BUFFER_MEMORY:
FREE((void *)buffer);
break;
case SOCKET_BUFFER_OBJECT:
ss->soi.free(buffer);
break;
case SOCKET_BUFFER_RAWPOINTER:
break;
}
}
static const void *
clone_buffer(struct socket_sendbuffer *buf, size_t *sz) {
switch (buf->type) {
case SOCKET_BUFFER_MEMORY:
*sz = buf->sz;
return buf->buffer;
case SOCKET_BUFFER_OBJECT:
*sz = USEROBJECT;
return buf->buffer;
case SOCKET_BUFFER_RAWPOINTER:
// It's a raw pointer, we need make a copy
*sz = buf->sz;
void * tmp = MALLOC(*sz);
memcpy(tmp, buf->buffer, *sz);
return tmp;
}
// never get here
*sz = 0;
return NULL;
}
static void
@@ -429,7 +488,12 @@ force_close(struct socket_server *ss, struct socket *s, struct socket_lock *l, s
}
s->type = SOCKET_TYPE_INVALID;
if (s->dw_buffer) {
free_buffer(ss, s->dw_buffer, s->dw_size);
struct socket_sendbuffer tmp;
tmp.buffer = s->dw_buffer;
tmp.sz = s->dw_size;
tmp.id = s->id;
tmp.type = (tmp.sz == USEROBJECT) ? SOCKET_BUFFER_OBJECT : SOCKET_BUFFER_MEMORY;
free_buffer(ss, &tmp);
s->dw_buffer = NULL;
}
socket_unlock(l);
@@ -849,12 +913,12 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock
if (s->type == SOCKET_TYPE_INVALID || s->id != id
|| s->type == SOCKET_TYPE_HALFCLOSE
|| s->type == SOCKET_TYPE_PACCEPT) {
so.free_func(request->buffer);
so.free_func((void *)request->buffer);
return -1;
}
if (s->type == SOCKET_TYPE_PLISTEN || s->type == SOCKET_TYPE_LISTEN) {
fprintf(stderr, "socket-server: write to listen fd %d.\n", id);
so.free_func(request->buffer);
so.free_func((void *)request->buffer);
return -1;
}
if (send_buffer_empty(s) && s->type == SOCKET_TYPE_CONNECTED) {
@@ -870,7 +934,7 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock
if (sasz == 0) {
// udp type mismatch, just drop it.
fprintf(stderr, "socket-server: udp socket (%d) type mistach.\n", id);
so.free_func(request->buffer);
so.free_func((void *)request->buffer);
return -1;
}
int n = sendto(s->fd, so.buffer, so.sz, 0, &sa.s, sasz);
@@ -878,7 +942,7 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock
append_sendbuffer_udp(ss,s,priority,request,udp_address);
} else {
stat_write(ss,s,n);
so.free_func(request->buffer);
so.free_func((void *)request->buffer);
return -1;
}
}
@@ -1572,10 +1636,11 @@ can_direct_write(struct socket *s, int id) {
// return -1 when error, 0 when success
int
socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz) {
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) {
free_buffer(ss, buffer, sz);
free_buffer(ss, buf);
return -1;
}
@@ -1587,7 +1652,7 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
if (can_direct_write(s,id)) {
// send directly
struct send_object so;
send_object_init(ss, &so, (void *)buffer, sz);
send_object_init_from_sendbuffer(ss, &so, buf);
ssize_t n;
if (s->protocol == PROTOCOL_TCP) {
n = write(s->fd, so.buffer, so.sz);
@@ -1597,7 +1662,7 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
if (sasz == 0) {
fprintf(stderr, "socket-server : set udp (%d) address first.\n", id);
socket_unlock(&l);
so.free_func((void *)buffer);
so.free_func((void *)buf->buffer);
return -1;
}
n = sendto(s->fd, so.buffer, so.sz, 0, &sa.s, sasz);
@@ -1610,12 +1675,11 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
if (n == so.sz) {
// write done
socket_unlock(&l);
so.free_func((void *)buffer);
so.free_func((void *)buf->buffer);
return 0;
}
// write failed, put buffer into s->dw_* , and let socket thread send it. see send_buffer()
s->dw_buffer = buffer;
s->dw_size = sz;
s->dw_buffer = clone_buffer(buf, &s->dw_size);
s->dw_offset = n;
sp_write(ss->event_fd, s->fd, s, true);
@@ -1630,8 +1694,7 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
struct request_package request;
request.u.send.id = id;
request.u.send.sz = sz;
request.u.send.buffer = (char *)buffer;
request.u.send.buffer = clone_buffer(buf, &request.u.send.sz);
send_request(ss, &request, 'D', sizeof(request.u.send));
return 0;
@@ -1639,10 +1702,12 @@ socket_server_send(struct socket_server *ss, int id, const void * buffer, int sz
// return -1 when error, 0 when success
int
socket_server_send_lowpriority(struct socket_server *ss, int id, const void * buffer, int sz) {
socket_server_send_lowpriority(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) {
free_buffer(ss, buffer, sz);
free_buffer(ss, buf);
return -1;
}
@@ -1650,8 +1715,7 @@ socket_server_send_lowpriority(struct socket_server *ss, int id, const void * bu
struct request_package request;
request.u.send.id = id;
request.u.send.sz = sz;
request.u.send.buffer = (char *)buffer;
request.u.send.buffer = clone_buffer(buf, &request.u.send.sz);
send_request(ss, &request, 'P', sizeof(request.u.send));
return 0;
@@ -1836,10 +1900,11 @@ socket_server_udp(struct socket_server *ss, uintptr_t opaque, const char * addr,
}
int
socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp_address *addr, const void *buffer, int sz) {
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) {
free_buffer(ss, buffer, sz);
free_buffer(ss, buf);
return -1;
}
@@ -1853,7 +1918,7 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp
addrsz = 1+2+16; // 1 type, 2 port, 16 ipv6
break;
default:
free_buffer(ss, buffer, sz);
free_buffer(ss, buf);
return -1;
}
@@ -1865,12 +1930,12 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp
if (can_direct_write(s,id)) {
// send directly
struct send_object so;
send_object_init(ss, &so, (void *)buffer, sz);
send_object_init_from_sendbuffer(ss, &so, buf);
union sockaddr_all sa;
socklen_t sasz = udp_socket_address(s, udp_address, &sa);
if (sasz == 0) {
socket_unlock(&l);
so.free_func((void *)buffer);
so.free_func((void *)buf->buffer);
return -1;
}
int n = sendto(s->fd, so.buffer, so.sz, 0, &sa.s, sasz);
@@ -1878,7 +1943,7 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp
// sendto succ
stat_write(ss,s,n);
socket_unlock(&l);
so.free_func((void *)buffer);
so.free_func((void *)buf->buffer);
return 0;
}
}
@@ -1888,8 +1953,7 @@ socket_server_udp_send(struct socket_server *ss, int id, const struct socket_udp
struct request_package request;
request.u.send_udp.send.id = id;
request.u.send_udp.send.sz = sz;
request.u.send_udp.send.buffer = (char *)buffer;
request.u.send_udp.send.buffer = clone_buffer(buf, &request.u.send_udp.send.sz);
memcpy(request.u.send_udp.address, udp_address, addrsz);

13
skynet-src/socket_server.h
View File

@@ -3,6 +3,7 @@
#include <stdint.h>
#include "socket_info.h"
#include "socket_buffer.h"
#define SOCKET_DATA 0
#define SOCKET_CLOSE 1
@@ -33,8 +34,8 @@ void socket_server_shutdown(struct socket_server *, uintptr_t opaque, int id);
void socket_server_start(struct socket_server *, uintptr_t opaque, int id);
// return -1 when error
int socket_server_send(struct socket_server *, int id, const void * buffer, int sz);
int socket_server_send_lowpriority(struct socket_server *, int id, const void * buffer, int sz);
int socket_server_send(struct socket_server *, struct socket_sendbuffer *buffer);
int socket_server_send_lowpriority(struct socket_server *, struct socket_sendbuffer *buffer);
// ctrl command below returns id
int socket_server_listen(struct socket_server *, uintptr_t opaque, const char * addr, int port, int backlog);
@@ -53,17 +54,17 @@ int socket_server_udp(struct socket_server *, uintptr_t opaque, const char * add
int socket_server_udp_connect(struct socket_server *, int id, const char * addr, int port);
// If the socket_udp_address is NULL, use last call socket_server_udp_connect address instead
// You can also use socket_server_send
int socket_server_udp_send(struct socket_server *, int id, const struct socket_udp_address *, const void *buffer, int sz);
int socket_server_udp_send(struct socket_server *, const struct socket_udp_address *, struct socket_sendbuffer *buffer);
// extract the address of the message, struct socket_message * should be SOCKET_UDP
const struct socket_udp_address * socket_server_udp_address(struct socket_server *, struct socket_message *, int *addrsz);
struct socket_object_interface {
void * (*buffer)(void *);
int (*size)(void *);
const void * (*buffer)(const void *);
size_t (*size)(const void *);
void (*free)(void *);
};
// if you send package sz == -1, use soi.
// if you send package with type SOCKET_BUFFER_OBJECT, use soi.
void socket_server_userobject(struct socket_server *, struct socket_object_interface *soi);
struct socket_info * socket_server_info(struct socket_server *);