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

remove unused source

Browse Source
This commit is contained in:
云风
2013-08-23 21:29:19 +08:00
parent c99f9a12d8
commit 424c84915b
7 changed files with 0 additions and 1508 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
gate/README
View File

@@ -1,3 +0,0 @@
See https://github.com/cloudwu/mread for detail.
Support write buffer now.

347
gate/main.c
View File

@@ -1,347 +0,0 @@
#include "skynet.h"
#include "mread.h"
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <assert.h>
#include <stdint.h>
#include <stdio.h>
#include <stdarg.h>
struct connection {
uint32_t agent;
uint32_t client;
int connection_id;
int uid;
};
struct gate {
struct mread_pool * pool;
uint32_t watchdog;
uint32_t broker;
int id_index;
int cap;
int max_connection;
int client_tag;
int header_size;
struct connection ** agent;
struct connection * map;
};
struct gate *
gate_create(void) {
struct gate * g = malloc(sizeof(*g));
memset(g,0,sizeof(*g));
return g;
}
static inline struct connection *
_id_to_agent(struct gate *g,int uid) {
struct connection * agent = g->agent[uid & (g->cap - 1)];
if (agent && agent->uid == uid) {
return agent;
}
return NULL;
}
static void
_parm(char *msg, int sz, int command_sz) {
while (command_sz < sz) {
if (msg[command_sz] != ' ')
break;
++command_sz;
}
int i;
for (i=command_sz;i<sz;i++) {
msg[i-command_sz] = msg[i];
}
msg[i-command_sz] = '\0';
}
static void
_forward_agent(struct gate * g, int id, uint32_t agentaddr, uint32_t clientaddr) {
struct connection * agent = _id_to_agent(g,id);
if (agent) {
agent->agent = agentaddr;
agent->client = clientaddr;
}
}
static void
_ctrl(struct skynet_context * ctx, struct gate * g, const void * msg, int sz) {
char tmp[sz+1];
memcpy(tmp, msg, sz);
tmp[sz] = '\0';
char * command = tmp;
int i;
if (sz == 0)
return;
for (i=0;i<sz;i++) {
if (command[i]==' ') {
break;
}
}
if (memcmp(command,"kick",i)==0) {
_parm(tmp, sz, i);
int uid = strtol(command , NULL, 10);
struct connection * agent = _id_to_agent(g,uid);
if (agent) {
int connection_id = agent->connection_id;
mread_close_client(g->pool,connection_id);
}
return;
}
if (memcmp(command,"forward",i)==0) {
_parm(tmp, sz, i);
char * client = tmp;
char * idstr = strsep(&client, " ");
if (client == NULL) {
return;
}
int id = strtol(idstr , NULL, 10);
char * agent = strsep(&client, " ");
if (client == NULL) {
return;
}
uint32_t agent_handle = strtoul(agent+1, NULL, 16);
uint32_t client_handle = strtoul(client+1, NULL, 16);
_forward_agent(g, id, agent_handle, client_handle);
return;
}
if (memcmp(command,"broker",i)==0) {
_parm(tmp, sz, i);
g->broker = skynet_queryname(ctx, command);
return;
}
if (memcmp(command,"start",i) == 0) {
skynet_command(ctx,"TIMEOUT","0");
return;
}
if (memcmp(command, "close", i) == 0) {
mread_close_listen(g->pool);
return;
}
skynet_error(ctx, "[gate] Unkown command : %s", command);
}
static void
_report(struct gate *g, struct skynet_context * ctx, const char * data, ...) {
if (g->watchdog == 0) {
return;
}
va_list ap;
va_start(ap, data);
char tmp[1024];
int n = vsnprintf(tmp, sizeof(tmp), data, ap);
va_end(ap);
skynet_send(ctx, 0, g->watchdog, PTYPE_TEXT, 0, tmp, n);
}
static void
_forward(struct skynet_context * ctx,struct gate *g, int uid, void * data, size_t len) {
if (g->broker) {
skynet_send(ctx, 0, g->broker, g->client_tag, 0, data, len);
return;
}
struct connection * agent = _id_to_agent(g,uid);
if (agent == NULL)
return;
if (agent->agent) {
skynet_send(ctx, agent->client, agent->agent, g->client_tag, 0 , data, len);
} else if (g->watchdog) {
char * tmp = malloc(len + 32);
int n = snprintf(tmp,len+32,"%d data ",uid);
memcpy(tmp+n,data,len);
skynet_send(ctx, 0, g->watchdog, PTYPE_TEXT | PTYPE_TAG_DONTCOPY, 0, tmp, len + n);
}
}
static int
_gen_id(struct gate * g, int connection_id) {
int uid = ++g->id_index;
int i;
for (i=0;i<g->cap;i++) {
int hash = (uid + i) & (g->cap - 1);
if (g->agent[hash] == NULL) {
uid = uid + i;
struct connection * conn = &g->map[connection_id];
conn->uid = uid;
g->agent[hash] = conn;
g->id_index = uid;
return uid;
}
}
assert(0);
}
static void
_remove_id(struct gate *g, int uid) {
struct connection ** pconn = &g->agent[uid & (g->cap - 1)];
struct connection * conn = *pconn;
if (conn) {
assert(conn->uid == uid);
conn->uid = 0;
conn->agent = 0;
*pconn = NULL;
}
}
static int
_cb(struct skynet_context * ctx, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) {
struct gate *g = ud;
if (type == PTYPE_TEXT) {
_ctrl(ctx, g , msg , (int)sz);
return 0;
} else if (type == PTYPE_CLIENT) {
if (sz <=4 ) {
skynet_error(ctx, "Invalid client message from %x",source);
return 0;
}
struct mread_pool * m = g->pool;
// The first 4 bytes in msg are the id of socket, write following bytes to it
const uint8_t * data = msg;
uint32_t uid = data[0] | data[1] << 8 | data[2] << 16 | data[3] << 24;
struct connection * agent = _id_to_agent(g,uid);
if (agent) {
int id = agent->connection_id;
mread_push(m, id, (void *)(data+4), sz - 4, (void *)data);
return 1;
} else {
skynet_error(ctx, "Invalid client id %d from %x",(int)uid,source);
return 0;
}
}
assert(type == PTYPE_RESPONSE);
struct mread_pool * m = g->pool;
int connection_id = mread_poll(m,10); // timeout : 10 ms
if (connection_id >= 0) {
int id = g->map[connection_id].uid;
if (id == 0) {
id = _gen_id(g, connection_id);
int fd = mread_socket(m , connection_id);
struct sockaddr_in remote_addr;
socklen_t len = sizeof(struct sockaddr_in);
getpeername(fd, (struct sockaddr *)&remote_addr, &len);
_report(g, ctx, "%d open %d %s:%u",id,fd,inet_ntoa(remote_addr.sin_addr),ntohs(remote_addr.sin_port));
}
uint8_t * plen = mread_pull(m,g->header_size);
if (plen == NULL) {
if (mread_closed(m)) {
_remove_id(g,id);
_report(g, ctx, "%d close", id);
}
goto _break;
}
// big-endian
int len ;
if (g->header_size == 2) {
len = plen[0] << 8 | plen[1];
} else {
len = plen[0] << 24 | plen[1] << 16 | plen[2] << 8 | plen[3];
}
void * data = mread_pull(m, len);
if (data == NULL) {
if (mread_closed(m)) {
_remove_id(g,id);
_report(g, ctx, "%d close", id);
}
goto _break;
}
_forward(ctx, g, id, data, len);
mread_yield(m);
}
_break:
skynet_command(ctx, "TIMEOUT", "0");
return 0;
}
int
gate_init(struct gate *g , struct skynet_context * ctx, char * parm) {
int port = 0;
int max = 0;
int buffer = 0;
int sz = strlen(parm)+1;
char watchdog[sz];
char binding[sz];
int client_tag = 0;
char header;
int n = sscanf(parm, "%c %s %s %d %d %d",&header,watchdog, binding,&client_tag , &max,&buffer);
if (n<4) {
skynet_error(ctx, "Invalid gate parm %s",parm);
return 1;
}
if (max <=0 ) {
skynet_error(ctx, "Need max connection");
return 1;
}
if (header != 'S' && header !='L') {
skynet_error(ctx, "Invalid data header style");
return 1;
}
if (client_tag == 0) {
client_tag = PTYPE_CLIENT;
}
char * portstr = strchr(binding,':');
uint32_t addr = INADDR_ANY;
if (portstr == NULL) {
port = strtol(binding, NULL, 10);
if (port <= 0) {
skynet_error(ctx, "Invalid gate address %s",parm);
return 1;
}
} else {
port = strtol(portstr + 1, NULL, 10);
if (port <= 0) {
skynet_error(ctx, "Invalid gate address %s",parm);
return 1;
}
portstr[0] = '\0';
addr=inet_addr(binding);
}
struct mread_pool * pool = mread_create(addr, port, max, buffer);
if (pool == NULL) {
skynet_error(ctx, "Create gate %s failed",parm);
return 1;
}
if (watchdog[0] == '!') {
g->watchdog = 0;
} else {
g->watchdog = skynet_queryname(ctx, watchdog);
if (g->watchdog == 0) {
skynet_error(ctx, "Invalid watchdog %s",watchdog);
return 1;
}
}
g->pool = pool;
int cap = 1;
while (cap < max) {
cap *= 2;
}
g->cap = cap;
g->max_connection = max;
g->id_index = 0;
g->client_tag = client_tag;
g->header_size = header=='S' ? 2 : 4;
g->agent = malloc(cap * sizeof(struct connection *));
memset(g->agent, 0, cap * sizeof(struct connection *));
g->map = malloc(max * sizeof(struct connection));
memset(g->map, 0, max * sizeof(struct connection));
int i;
for (i=0;i<max;i++) {
g->map[i].connection_id = i;
}
skynet_callback(ctx,g,_cb);
return 0;
}

692
gate/mread.c
View File

@@ -1,692 +0,0 @@
#include "mread.h"
#include "ringbuffer.h"
#include "event.h"
#include <netinet/in.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <errno.h>
#include <stdlib.h>
#include <stdint.h>
#include <string.h>
#include <assert.h>
#include <stdio.h>
#include <fcntl.h>
#define BACKLOG 32
#define READBLOCKSIZE 2048
#define RINGBUFFER_DEFAULT 1024 * 1024
#define SOCKET_INVALID 0
#define SOCKET_CLOSED 1
#define SOCKET_SUSPEND 2
#define SOCKET_READ 3
#define SOCKET_POLLIN 4
#define SOCKET_HALFCLOSE 5
#define SOCKET_ALIVE SOCKET_SUSPEND
#define LISTENSOCKET (void *)((intptr_t)~0)
struct send_buffer {
struct send_buffer * next;
int size;
char * buff;
char * ptr;
};
struct send_client {
struct send_buffer * head;
struct send_buffer * tail;
};
struct socket {
int fd;
struct ringbuffer_block * node;
struct ringbuffer_block * temp;
struct send_client client;
int enablewrite;
int status;
};
struct mread_pool {
int listen_fd;
int efd;
int max_connection;
int closed;
int active;
int skip;
struct socket * sockets;
struct socket * free_socket;
int queue_len;
int queue_head;
struct event ev[MAX_EVENT];
struct ringbuffer * rb;
};
// send begin
static void
turn_on(struct mread_pool *self, struct socket * s) {
if (s->status < SOCKET_ALIVE || s->enablewrite) {
return;
}
event_write(self->efd, s->fd, s, true);
s->enablewrite = 1;
}
static void
turn_off(struct mread_pool *self, struct socket * s) {
if (s->status < SOCKET_ALIVE || ! s->enablewrite) {
return;
}
s->enablewrite = 0;
event_write(self->efd, s->fd, s, false);
}
static void
free_buffer(struct send_client * sc) {
struct send_buffer * sb = sc->head;
while (sb) {
struct send_buffer * tmp = sb;
sb = sb->next;
free(tmp->ptr);
free(tmp);
}
sc->head = sc->tail = NULL;
}
static void
client_send(struct send_client *c, int fd) {
while (c->head) {
struct send_buffer * tmp = c->head;
for (;;) {
int sz = write(fd, tmp->buff, tmp->size);
if (sz < 0) {
switch(errno) {
case EINTR:
continue;
case EWOULDBLOCK:
return;
}
free_buffer(c);
return;
}
if (sz != tmp->size) {
assert(sz < tmp->size);
tmp->buff += sz;
tmp->size -= sz;
return;
}
break;
}
c->head = tmp->next;
free(tmp->ptr);
free(tmp);
}
c->tail = NULL;
}
static void
client_push(struct send_client *c, void * buf, int sz, void * ptr) {
struct send_buffer * sb = malloc(sizeof(*sb));
sb->next = NULL;
sb->buff = buf;
sb->size = sz;
sb->ptr = ptr;
if (c->head) {
c->tail->next = sb;
c->tail = sb;
} else {
c->head = c->tail = sb;
}
}
static void
try_close(struct mread_pool * self, struct socket * s) {
if (s->client.head == NULL) {
turn_off(self, s);
}
if (s->status != SOCKET_HALFCLOSE) {
return;
}
if (s->client.head == NULL) {
s->status = SOCKET_CLOSED;
s->node = NULL;
s->temp = NULL;
event_del(self->efd, s->fd);
close(s->fd);
// printf("MREAD close %d (fd=%d)\n",id,s->fd);
++self->closed;
}
}
void
mread_push(struct mread_pool *self, int id, void * buffer, int size, void * ptr) {
struct socket * s = &self->sockets[id];
switch(s->status) {
case SOCKET_INVALID:
case SOCKET_CLOSED:
case SOCKET_HALFCLOSE:
free(ptr);
return;
}
if (s->client.head == NULL) {
for (;;) {
int sz = write(s->fd, buffer, size);
if (sz < 0) {
switch(errno) {
case EINTR:
continue;
case EWOULDBLOCK:
break;
default:
// write error : close fd
s->status = SOCKET_HALFCLOSE;
try_close(self, s);
return;
}
break;
}
if (sz == size) {
// ptr is malloc by
// service-src/service-client.c : 18
// uint8_t *tmp = malloc(sz + 4 + 2);
free(ptr);
return;
}
buffer = (char *)buffer + sz;
size -= sz;
}
}
client_push(&s->client,buffer, size, ptr);
turn_on(self, s);
}
// send end
static struct socket *
_create_sockets(int max) {
int i;
struct socket * s = malloc(max * sizeof(struct socket));
for (i=0;i<max;i++) {
s[i].fd = i+1;
s[i].node = NULL;
s[i].temp = NULL;
s[i].status = SOCKET_INVALID;
s[i].enablewrite = 0;
s[i].client.head = s[i].client.tail = NULL;
}
s[max-1].fd = -1;
return s;
}
static struct ringbuffer *
_create_rb(int size) {
size = (size + 3) & ~3;
if (size < READBLOCKSIZE * 2) {
size = READBLOCKSIZE * 2;
}
struct ringbuffer * rb = ringbuffer_new(size);
return rb;
}
static void
_release_rb(struct ringbuffer * rb) {
ringbuffer_delete(rb);
}
static int
_set_nonblocking(int fd)
{
int flag = fcntl(fd, F_GETFL, 0);
if ( -1 == flag ) {
return -1;
}
return fcntl(fd, F_SETFL, flag | O_NONBLOCK);
}
struct mread_pool *
mread_create(uint32_t addr, int port , int max , int buffer_size) {
int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
if (listen_fd == -1) {
return NULL;
}
if ( -1 == _set_nonblocking(listen_fd) ) {
close(listen_fd);
return NULL;
}
int reuse = 1;
setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(int));
struct sockaddr_in my_addr;
memset(&my_addr, 0, sizeof(struct sockaddr_in));
my_addr.sin_family = AF_INET;
my_addr.sin_port = htons(port);
my_addr.sin_addr.s_addr = addr;
// printf("MREAD bind %s:%u\n",inet_ntoa(my_addr.sin_addr),ntohs(my_addr.sin_port));
if (bind(listen_fd, (struct sockaddr *)&my_addr, sizeof(struct sockaddr)) == -1) {
close(listen_fd);
return NULL;
}
if (listen(listen_fd, BACKLOG) == -1) {
close(listen_fd);
return NULL;
}
int efd = event_init(max+1);
if (efd == -1) {
close(listen_fd);
return NULL;
}
if (event_add(efd, listen_fd, LISTENSOCKET)) {
close(listen_fd);
close(efd);
return NULL;
}
struct mread_pool * self = malloc(sizeof(*self));
self->listen_fd = listen_fd;
self->efd = efd;
self->max_connection = max;
self->closed = 0;
self->active = -1;
self->skip = 0;
self->sockets = _create_sockets(max);
self->free_socket = &self->sockets[0];
self->queue_len = 0;
self->queue_head = 0;
if (buffer_size == 0) {
self->rb = _create_rb(RINGBUFFER_DEFAULT);
} else {
self->rb = _create_rb(buffer_size);
}
return self;
}
void
mread_close(struct mread_pool *self) {
if (self == NULL)
return;
int i;
struct socket * s = self->sockets;
for (i=0;i<self->max_connection;i++) {
if (s[i].status >= SOCKET_ALIVE) {
close(s[i].fd);
}
}
free_buffer(&s->client);
free(s);
mread_close_listen(self);
close(self->efd);
_release_rb(self->rb);
free(self);
}
void
mread_close_listen(struct mread_pool *self) {
if (self == NULL)
return;
if (self->listen_fd >= 0) {
close(self->listen_fd);
self->listen_fd = 0;
}
}
static int
_read_queue(struct mread_pool * self, int timeout) {
self->queue_head = 0;
int n = event_wait(self->efd, self->ev, timeout);
if (n == -1) {
self->queue_len = 0;
return -1;
}
self->queue_len = n;
return n;
}
inline static struct socket *
_read_one(struct mread_pool * self) {
for (;;) {
if (self->queue_head >= self->queue_len) {
return NULL;
}
struct socket * ret = self->ev[self->queue_head].s;
bool writeflag = self->ev[self->queue_head].write;
bool readflag = self->ev[self->queue_head].read;
++ self->queue_head;
if (ret == LISTENSOCKET) {
return ret;
}
if (writeflag && ret->status != SOCKET_CLOSED) {
client_send(&ret->client, ret->fd);
try_close(self, ret);
}
if (readflag && ret->status != SOCKET_HALFCLOSE &&
ret->status != SOCKET_CLOSED)
return ret;
}
}
static struct socket *
_alloc_socket(struct mread_pool * self) {
if (self->free_socket == NULL) {
return NULL;
}
struct socket * s = self->free_socket;
int next_free = s->fd;
if (next_free < 0 ) {
self->free_socket = NULL;
} else {
self->free_socket = &self->sockets[next_free];
}
return s;
}
static struct socket *
_add_client(struct mread_pool * self, int fd) {
struct socket * s = _alloc_socket(self);
if (s == NULL) {
close(fd);
return NULL;
}
if (event_add(self->efd, fd, s)) {
close(fd);
return NULL;
}
s->fd = fd;
s->node = NULL;
s->status = SOCKET_SUSPEND;
s->enablewrite = 0;
return s;
}
static int
_report_closed(struct mread_pool * self) {
int i;
for (i=0;i<self->max_connection;i++) {
if (self->sockets[i].status == SOCKET_CLOSED) {
self->active = i;
return i;
}
}
assert(0);
return -1;
}
int
mread_poll(struct mread_pool * self , int timeout) {
self->skip = 0;
if (self->active >= 0) {
struct socket * s = &self->sockets[self->active];
if (s->status == SOCKET_READ) {
return self->active;
}
}
if (self->queue_head >= self->queue_len) {
if (self->closed > 0 ) {
return _report_closed(self);
}
if (_read_queue(self, timeout) == -1) {
self->active = -1;
return -1;
}
}
for (;;) {
struct socket * s = _read_one(self);
if (s == NULL) {
self->active = -1;
return -1;
}
if (s == LISTENSOCKET) {
struct sockaddr_in remote_addr;
socklen_t len = sizeof(struct sockaddr_in);
int client_fd = accept(self->listen_fd , (struct sockaddr *)&remote_addr , &len);
if (client_fd >= 0) {
_set_nonblocking(client_fd);
// printf("MREAD connect %s:%u (fd=%d)\n",inet_ntoa(remote_addr.sin_addr),ntohs(remote_addr.sin_port), client_fd);
struct socket * s = _add_client(self, client_fd);
if (s) {
self->active = -1;
return s - self->sockets;
}
}
} else {
int index = s - self->sockets;
assert(index >=0 && index < self->max_connection);
self->active = index;
s->status = SOCKET_POLLIN;
return index;
}
}
}
int
mread_socket(struct mread_pool * self, int index) {
return self->sockets[index].fd;
}
static void
_link_node(struct ringbuffer * rb, int id, struct socket * s , struct ringbuffer_block * blk) {
if (s->node) {
ringbuffer_link(rb, s->node , blk);
} else {
blk->id = id;
s->node = blk;
}
}
void
mread_close_client(struct mread_pool * self, int id) {
struct socket * s = &self->sockets[id];
if (s->status >= SOCKET_ALIVE && s->status != SOCKET_HALFCLOSE) {
s->status = SOCKET_HALFCLOSE;
try_close(self, s);
}
}
static void
force_close_client(struct mread_pool * self, int id) {
struct socket * s = &self->sockets[id];
free_buffer(&s->client);
if (s->status >= SOCKET_ALIVE && s->status != SOCKET_HALFCLOSE) {
s->status = SOCKET_HALFCLOSE;
try_close(self, s);
}
}
static void
_close_active(struct mread_pool * self) {
int id = self->active;
struct socket * s = &self->sockets[id];
ringbuffer_free(self->rb, s->temp);
ringbuffer_free(self->rb, s->node);
force_close_client(self, id);
}
static char *
_ringbuffer_read(struct mread_pool * self, int *size) {
struct socket * s = &self->sockets[self->active];
if (s->node == NULL) {
*size = 0;
return NULL;
}
int sz = *size;
void * ret;
*size = ringbuffer_data(self->rb, s->node, sz , self->skip, &ret);
return ret;
}
static void
skip_all(struct socket *s) {
char tmp[1024];
for (;;) {
int bytes = read(s->fd, tmp, sizeof(tmp));
if (bytes == 0) {
free_buffer(&s->client);
} else if (bytes != sizeof(tmp)) {
return;
}
}
}
void *
mread_pull(struct mread_pool * self , int size) {
if (self->active == -1) {
return NULL;
}
struct socket *s = &self->sockets[self->active];
int rd_size = size;
char * buffer = _ringbuffer_read(self, &rd_size);
if (buffer) {
self->skip += size;
return buffer;
}
switch (s->status) {
case SOCKET_READ:
s->status = SOCKET_SUSPEND;
case SOCKET_CLOSED:
case SOCKET_SUSPEND:
return NULL;
case SOCKET_HALFCLOSE:
skip_all(s);
return NULL;
default:
assert(s->status == SOCKET_POLLIN);
break;
}
int sz = size - rd_size;
int rd = READBLOCKSIZE;
if (rd < sz) {
rd = sz;
}
int id = self->active;
struct ringbuffer * rb = self->rb;
struct ringbuffer_block * blk = ringbuffer_alloc(rb , rd);
while (blk == NULL) {
int collect_id = ringbuffer_collect(rb);
force_close_client(self , collect_id);
if (id == collect_id) {
return NULL;
}
blk = ringbuffer_alloc(rb , rd);
}
buffer = (char *)(blk + 1);
for (;;) {
int bytes = read(s->fd, buffer, rd);
if (bytes > 0) {
ringbuffer_shrink(rb, blk , bytes);
if (bytes < sz) {
_link_node(rb, self->active, s , blk);
s->status = SOCKET_SUSPEND;
return NULL;
}
s->status = SOCKET_READ;
break;
}
if (bytes == 0) {
ringbuffer_shrink(rb, blk, 0);
_close_active(self);
return NULL;
}
if (bytes == -1) {
switch(errno) {
case EWOULDBLOCK:
ringbuffer_shrink(rb, blk, 0);
s->status = SOCKET_SUSPEND;
return NULL;
case EINTR:
continue;
default:
ringbuffer_shrink(rb, blk, 0);
_close_active(self);
return NULL;
}
}
}
_link_node(rb, self->active , s , blk);
void * ret;
int real_rd = ringbuffer_data(rb, s->node , size , self->skip, &ret);
if (ret) {
self->skip += size;
return ret;
}
assert(real_rd == size);
struct ringbuffer_block * temp = ringbuffer_alloc(rb, size);
while (temp == NULL) {
int collect_id = ringbuffer_collect(rb);
force_close_client(self , collect_id);
if (id == collect_id) {
return NULL;
}
temp = ringbuffer_alloc(rb , size);
}
temp->id = id;
if (s->temp) {
ringbuffer_link(rb, temp, s->temp);
}
s->temp = temp;
ret = ringbuffer_copy(rb, s->node, self->skip, temp);
assert(ret);
self->skip += size;
return ret;
}
void
mread_yield(struct mread_pool * self) {
if (self->active == -1) {
return;
}
struct socket *s = &self->sockets[self->active];
ringbuffer_free(self->rb , s->temp);
s->temp = NULL;
if (s->status == SOCKET_CLOSED && s->node == NULL) {
--self->closed;
s->status = SOCKET_INVALID;
if (self->free_socket) {
s->fd = self->free_socket - self->sockets;
} else {
s->fd = -1;
}
self->free_socket = s;
self->skip = 0;
self->active = -1;
} else {
if (s->node) {
s->node = ringbuffer_yield(self->rb, s->node, self->skip);
}
self->skip = 0;
if (s->node == NULL) {
self->active = -1;
}
}
}
int
mread_closed(struct mread_pool * self) {
if (self->active == -1) {
return 0;
}
struct socket * s = &self->sockets[self->active];
if (s->status == SOCKET_CLOSED && s->node == NULL) {
mread_yield(self);
return 1;
}
return 0;
}

21
gate/mread.h
View File

@@ -1,21 +0,0 @@
#ifndef MREAD_H
#define MREAD_H
#include <stdint.h>
struct mread_pool;
struct mread_pool * mread_create(uint32_t addr, int port , int max , int buffer);
void mread_close(struct mread_pool *m);
void mread_close_listen(struct mread_pool *self);
int mread_poll(struct mread_pool *m , int timeout);
void * mread_pull(struct mread_pool *m , int size);
void mread_push(struct mread_pool *m, int id, void * buffer, int size, void * ptr);
void mread_yield(struct mread_pool *m);
int mread_closed(struct mread_pool *m);
void mread_close_client(struct mread_pool *m, int id);
int mread_socket(struct mread_pool *m , int index);
void mread_close_listen(struct mread_pool *self);
#endif

281
gate/ringbuffer.c
View File

@@ -1,281 +0,0 @@
#include "ringbuffer.h"
#include <stdlib.h>
#include <assert.h>
#include <string.h>
#include <stdio.h>
#define ALIGN(s) (((s) + 3 ) & ~3)
struct ringbuffer {
int size;
int head;
};
static inline int
block_offset(struct ringbuffer * rb, struct ringbuffer_block * blk) {
char * start = (char *)(rb + 1);
return (char *)blk - start;
}
static inline struct ringbuffer_block *
block_ptr(struct ringbuffer * rb, int offset) {
char * start = (char *)(rb + 1);
return (struct ringbuffer_block *)(start + offset);
}
static inline struct ringbuffer_block *
block_next(struct ringbuffer * rb, struct ringbuffer_block * blk) {
int align_length = ALIGN(blk->length);
int head = block_offset(rb, blk);
if (align_length + head == rb->size) {
return NULL;
}
assert(align_length + head < rb->size);
return block_ptr(rb, head + align_length);
}
struct ringbuffer *
ringbuffer_new(int size) {
struct ringbuffer * rb = malloc(sizeof(*rb) + size);
rb->size = size;
rb->head = 0;
struct ringbuffer_block * blk = block_ptr(rb, 0);
blk->length = size;
blk->id = -1;
return rb;
}
void
ringbuffer_delete(struct ringbuffer * rb) {
free(rb);
}
void
ringbuffer_link(struct ringbuffer *rb , struct ringbuffer_block * head, struct ringbuffer_block * next) {
while (head->next >=0) {
head = block_ptr(rb, head->next);
}
next->id = head->id;
head->next = block_offset(rb, next);
}
static struct ringbuffer_block *
_alloc(struct ringbuffer * rb, int total_size , int size) {
struct ringbuffer_block * blk = block_ptr(rb, rb->head);
int align_length = ALIGN(sizeof(struct ringbuffer_block) + size);
blk->length = sizeof(struct ringbuffer_block) + size;
blk->offset = 0;
blk->next = -1;
blk->id = -1;
struct ringbuffer_block * next = block_next(rb, blk);
if (next) {
rb->head = block_offset(rb, next);
if (align_length < total_size) {
next->length = total_size - align_length;
if (next->length >= sizeof(struct ringbuffer_block)) {
next->id = -1;
}
}
} else {
rb->head = 0;
}
return blk;
}
struct ringbuffer_block *
ringbuffer_alloc(struct ringbuffer * rb, int size) {
int align_length = ALIGN(sizeof(struct ringbuffer_block) + size);
int i;
for (i=0;i<2;i++) {
int free_size = 0;
struct ringbuffer_block * blk = block_ptr(rb, rb->head);
do {
if (blk->length >= sizeof(struct ringbuffer_block) && blk->id >= 0)
return NULL;
free_size += ALIGN(blk->length);
if (free_size >= align_length) {
return _alloc(rb, free_size , size);
}
blk = block_next(rb, blk);
} while(blk);
rb->head = 0;
}
return NULL;
}
static int
_last_id(struct ringbuffer * rb) {
int i;
for (i=0;i<2;i++) {
struct ringbuffer_block * blk = block_ptr(rb, rb->head);
do {
if (blk->length >= sizeof(struct ringbuffer_block) && blk->id >= 0)
return blk->id;
blk = block_next(rb, blk);
} while(blk);
rb->head = 0;
}
return -1;
}
int
ringbuffer_collect(struct ringbuffer * rb) {
int id = _last_id(rb);
struct ringbuffer_block *blk = block_ptr(rb, 0);
do {
if (blk->length >= sizeof(struct ringbuffer_block) && blk->id == id) {
blk->id = -1;
}
blk = block_next(rb, blk);
} while(blk);
return id;
}
void
ringbuffer_shrink(struct ringbuffer * rb, struct ringbuffer_block * blk, int size) {
if (size == 0) {
rb->head = block_offset(rb, blk);
return;
}
int align_length = ALIGN(sizeof(struct ringbuffer_block) + size);
int old_length = ALIGN(blk->length);
assert(align_length <= old_length);
blk->length = size + sizeof(struct ringbuffer_block);
if (align_length == old_length) {
return;
}
blk = block_next(rb, blk);
blk->length = old_length - align_length;
if (blk->length >= sizeof(struct ringbuffer_block)) {
blk->id = -1;
}
rb->head = block_offset(rb, blk);
}
static int
_block_id(struct ringbuffer_block * blk) {
assert(blk->length >= sizeof(struct ringbuffer_block));
int id = blk->id;
assert(id>=0);
return id;
}
void
ringbuffer_free(struct ringbuffer * rb, struct ringbuffer_block * blk) {
if (blk == NULL)
return;
int id = _block_id(blk);
blk->id = -1;
while (blk->next >= 0) {
blk = block_ptr(rb, blk->next);
assert(_block_id(blk) == id);
blk->id = -1;
}
}
int
ringbuffer_data(struct ringbuffer * rb, struct ringbuffer_block * blk, int size, int skip, void **ptr) {
int length = blk->length - sizeof(struct ringbuffer_block) - blk->offset;
for (;;) {
if (length > skip) {
if (length - skip >= size) {
char * start = (char *)(blk + 1);
*ptr = (start + blk->offset + skip);
return size;
}
*ptr = NULL;
int ret = length - skip;
while (blk->next >= 0) {
blk = block_ptr(rb, blk->next);
ret += blk->length - sizeof(struct ringbuffer_block);
if (ret >= size)
return size;
}
return ret;
}
if (blk->next < 0) {
assert(length == skip);
*ptr = NULL;
return 0;
}
blk = block_ptr(rb, blk->next);
assert(blk->offset == 0);
skip -= length;
length = blk->length - sizeof(struct ringbuffer_block);
}
}
void *
ringbuffer_copy(struct ringbuffer * rb, struct ringbuffer_block * from, int skip, struct ringbuffer_block * to) {
int size = to->length - sizeof(struct ringbuffer_block);
int length = from->length - sizeof(struct ringbuffer_block) - from->offset;
char * ptr = (char *)(to+1);
for (;;) {
if (length > skip) {
char * src = (char *)(from + 1);
src += from->offset + skip;
length -= skip;
while (length < size) {
memcpy(ptr, src, length);
assert(from->next >= 0);
from = block_ptr(rb , from->next);
assert(from->offset == 0);
ptr += length;
size -= length;
length = from->length - sizeof(struct ringbuffer_block);
src = (char *)(from + 1);
}
memcpy(ptr, src , size);
to->id = from->id;
return (char *)(to + 1);
}
assert(from->next >= 0);
from = block_ptr(rb, from->next);
assert(from->offset == 0);
skip -= length;
length = from->length - sizeof(struct ringbuffer_block);
}
}
struct ringbuffer_block *
ringbuffer_yield(struct ringbuffer * rb, struct ringbuffer_block *blk, int skip) {
int length = blk->length - sizeof(struct ringbuffer_block) - blk->offset;
for (;;) {
if (length > skip) {
blk->offset += skip;
return blk;
}
blk->id = -1;
if (blk->next < 0) {
return NULL;
}
blk = block_ptr(rb, blk->next);
assert(blk->offset == 0);
skip -= length;
length = blk->length - sizeof(struct ringbuffer_block);
}
}
void
ringbuffer_dump(struct ringbuffer * rb) {
struct ringbuffer_block *blk = block_ptr(rb,0);
int i=0;
printf("total size= %d\n",rb->size);
while (blk) {
++i;
if (i>10)
break;
if (blk->length >= sizeof(*blk)) {
printf("[%u : %d]", (unsigned)(blk->length - sizeof(*blk)), block_offset(rb,blk));
printf(" id=%d",blk->id);
if (blk->id >=0) {
printf(" offset=%d next=%d",blk->offset, blk->next);
}
} else {
printf("<%u : %d>", blk->length, block_offset(rb,blk));
}
printf("\n");
blk = block_next(rb, blk);
}
}

27
gate/ringbuffer.h
View File

@@ -1,27 +0,0 @@
#ifndef MREAD_RINGBUFFER_H
#define MREAD_RINGBUFFER_H
struct ringbuffer;
struct ringbuffer_block {
int length;
int offset;
int id;
int next;
};
struct ringbuffer * ringbuffer_new(int size);
void ringbuffer_delete(struct ringbuffer * rb);
void ringbuffer_link(struct ringbuffer *rb , struct ringbuffer_block * prev, struct ringbuffer_block * next);
struct ringbuffer_block * ringbuffer_alloc(struct ringbuffer * rb, int size);
int ringbuffer_collect(struct ringbuffer * rb);
void ringbuffer_shrink(struct ringbuffer * rb, struct ringbuffer_block * blk, int size);
void ringbuffer_free(struct ringbuffer * rb, struct ringbuffer_block * blk);
int ringbuffer_data(struct ringbuffer * rb, struct ringbuffer_block * blk, int size, int skip, void **ptr);
void * ringbuffer_copy(struct ringbuffer * rb, struct ringbuffer_block * from, int skip, struct ringbuffer_block * to);
struct ringbuffer_block * ringbuffer_yield(struct ringbuffer * rb, struct ringbuffer_block *blk, int skip);
void ringbuffer_dump(struct ringbuffer * rb);
#endif

137
service-src/event.h
View File

@@ -1,137 +0,0 @@
#ifndef skynet_event_h
#define skynet_event_h
#include <stdbool.h>
#include <stdlib.h>
/* Test for polling API */
#ifdef __linux__
#define HAVE_EPOLL 1
#endif
#if defined(__APPLE__) || defined(__FreeBSD__) || defined(__OpenBSD__) || defined (__NetBSD__)
#define HAVE_KQUEUE 1
#endif
#if !defined(HAVE_EPOLL) && !defined(HAVE_KQUEUE)
#error "system does not support epoll or kqueue API"
#endif
/* ! Test for polling API */
#include <sys/types.h>
#ifdef HAVE_EPOLL
#include <sys/epoll.h>
#elif HAVE_KQUEUE
#include <time.h>
#include <sys/event.h>
#endif
#define MAX_EVENT 32
struct event {
void * s;
bool read;
bool write;
};
#ifdef HAVE_EPOLL
static int
event_init(int max) {
return epoll_create(max);
}
static int
event_add(int efd, int sock, void *ud) {
struct epoll_event ev;
ev.events = EPOLLIN;
ev.data.ptr = ud;
if (epoll_ctl(efd, EPOLL_CTL_ADD, sock, &ev) == -1) {
return 1;
}
return 0;
}
static void
event_del(int efd, int fd) {
epoll_ctl(efd, EPOLL_CTL_DEL, fd , NULL);
}
static int
event_wait(int efd, struct event * e, int timeout) {
struct epoll_event ev[MAX_EVENT];
int n = epoll_wait(efd , ev, MAX_EVENT, timeout);
int i;
for (i=0;i<n;i++) {
e[i].s = ev[i].data.ptr;
unsigned flag = ev[i].events;
e[i].write = (flag & EPOLLOUT) != 0;
e[i].read = (flag & EPOLLIN) != 0;
}
return n;
}
static void
event_write(int efd, int fd, void *ud, bool enable) {
struct epoll_event ev;
ev.events = EPOLLIN | (enable ? EPOLLOUT : 0);
ev.data.ptr = ud;
epoll_ctl(efd, EPOLL_CTL_MOD, fd, &ev);
}
#elif HAVE_KQUEUE
static int
event_init(int max) {
return kqueue();
}
static int
event_add(int efd, int sock, void *ud) {
struct kevent ke;
EV_SET(&ke, sock, EVFILT_READ, EV_ADD, 0, 0, ud);
if (kevent(efd, &ke, 1, NULL, 0, NULL) == -1) {
return 1;
}
return 0;
}
static void
event_del(int efd, int fd) {
struct kevent ke;
EV_SET(&ke, fd, EVFILT_READ | EVFILT_WRITE, EV_DELETE, 0, 0, NULL);
kevent(efd, &ke, 1, NULL, 0, NULL);
}
static int
event_wait(int kfd, struct event *e, int timeout) {
struct timespec timeoutspec;
timeoutspec.tv_sec = timeout / 1000;
timeoutspec.tv_nsec = (timeout % 1000) * 1000000;
struct kevent ev[MAX_EVENT];
int n = kevent(kfd, NULL, 0, ev, MAX_EVENT, &timeoutspec);
int i;
for (i=0;i<n;i++) {
e[i].s = ev[i].udata;
unsigned flag = ev[i].filter;
e[i].write = (flag & EVFILT_WRITE) != 0;
e[i].read = (flag & EVFILT_READ) != 0;
}
return n;
}
static void
event_write(int efd, int fd, void *ud, bool enable) {
struct kevent ke;
EV_SET(&ke, fd, EVFILT_WRITE, enable ? EV_ENABLE : EV_DISABLE, 0, 0, ud);
kevent(efd, &ke, 1, NULL, 0, NULL);
}
#endif
#endif