skynet abort

This commit is contained in:
云风
2012-11-08 14:57:51 +08:00
parent 37d8020155
commit b6314b537f
10 changed files with 84 additions and 22 deletions

View File

@@ -584,4 +584,8 @@ function skynet.endless()
return c.command("ENDLESS")~=nil
end
function skynet.abort()
c.command("ABORT")
end
return skynet

View File

@@ -181,12 +181,12 @@ static void
_hash_delete(struct hashmap *hash) {
int i;
for (i=0;i<HASH_SIZE;i++) {
struct keyvalue ** ptr = &hash->node[i];
while (*ptr) {
struct keyvalue * node = *ptr;
ptr = &node->next;
struct keyvalue * node = hash->node[i];
while (node) {
struct keyvalue * next = node->next;
_release_queue(node->queue);
free(node);
node = next;
}
}
free(hash);

View File

@@ -54,11 +54,11 @@ master_release(struct master * m) {
free(m->remote_addr[i]);
}
for (i=0;i<HASH_SIZE;i++) {
struct name ** ptr = &m->map.node[i];
while (*ptr) {
struct name * node = *ptr;
ptr = &node->next;
struct name * node = m->map.node[i];
while (node) {
struct name * next = node->next;
free(node);
node = next;
}
}
free(m);

3
service/abort.lua Normal file
View File

@@ -0,0 +1,3 @@
local skynet = require "skynet"
skynet.abort()

View File

@@ -93,6 +93,26 @@ skynet_handle_retire(uint32_t handle) {
rwlock_wunlock(&s->lock);
}
void
skynet_handle_retireall() {
struct handle_storage *s = H;
for (;;) {
int n=0;
int i;
for (i=0;i<s->slot_size;i++) {
rwlock_rlock(&s->lock);
struct skynet_context * ctx = s->slot[i];
rwlock_runlock(&s->lock);
if (ctx != NULL) {
++n;
skynet_handle_retire(skynet_context_handle(ctx));
}
}
if (n==0)
return;
}
}
struct skynet_context *
skynet_handle_grab(uint32_t handle) {
struct handle_storage *s = H;

View File

@@ -10,6 +10,7 @@ struct skynet_context;
uint32_t skynet_handle_register(struct skynet_context *);
void skynet_handle_retire(uint32_t handle);
struct skynet_context * skynet_handle_grab(uint32_t handle);
void skynet_handle_retireall();
uint32_t skynet_handle_findname(const char * name);
const char * skynet_handle_namehandle(uint32_t handle, const char *name);

View File

@@ -114,5 +114,7 @@ main(int argc, char *argv[]) {
skynet_start(&config);
printf("skynet exit\n");
return 0;
}

View File

@@ -49,6 +49,23 @@ struct skynet_context {
CHECKCALLING_DECL
};
static int g_total_context = 0;
int
skynet_context_total() {
return g_total_context;
}
static void
_context_inc() {
__sync_fetch_and_add(&g_total_context,1);
}
static void
_context_dec() {
__sync_fetch_and_sub(&g_total_context,1);
}
static void
_id_to_hex(char * str, uint32_t id) {
int i;
@@ -96,7 +113,10 @@ skynet_context_new(const char * name, const char *param) {
ctx->init = true;
}
skynet_mq_force_push(queue);
printf("[:%x] launch %s %s\n",ret->handle, name, param ? param : "");
if (ret) {
printf("[:%x] launch %s %s\n",ret->handle, name, param ? param : "");
}
_context_inc();
return ret;
} else {
skynet_context_release(ctx);
@@ -121,6 +141,7 @@ _delete_context(struct skynet_context *ctx) {
skynet_module_instance_release(ctx->mod, ctx->instance);
skynet_mq_mark_release(ctx->queue);
free(ctx);
_context_dec();
}
struct skynet_context *
@@ -480,6 +501,11 @@ skynet_command(struct skynet_context * context, const char * cmd , const char *
return NULL;
}
if (strcmp(cmd,"ABORT") == 0) {
skynet_handle_retireall();
return NULL;
}
return NULL;
}

View File

@@ -17,6 +17,7 @@ int skynet_context_push(uint32_t handle, struct skynet_message *message);
void skynet_context_send(struct skynet_context * context, void * msg, size_t sz, uint32_t source, int type, int session);
int skynet_context_newsession(struct skynet_context *);
int skynet_context_message_dispatch(struct skynet_monitor *); // return 1 when block
int skynet_context_total();
void skynet_context_endless(uint32_t handle); // for monitor

View File

@@ -20,6 +20,8 @@ struct monitor {
struct skynet_monitor ** m;
};
#define CHECK_ABORT if (skynet_context_total()==0) break;
static void *
_monitor(void *p) {
struct monitor * m = p;
@@ -29,8 +31,14 @@ _monitor(void *p) {
for (i=0;i<n;i++) {
skynet_monitor_check(m->m[i]);
}
CHECK_ABORT
sleep(5);
}
for (i=0;i<n;i++) {
skynet_monitor_delete(m->m[i]);
}
free(m->m);
free(m);
return NULL;
}
@@ -39,6 +47,7 @@ static void *
_timer(void *p) {
for (;;) {
skynet_updatetime();
CHECK_ABORT
usleep(2500);
}
return NULL;
@@ -49,6 +58,7 @@ _worker(void *p) {
struct skynet_monitor *sm = p;
for (;;) {
if (skynet_context_message_dispatch(sm)) {
CHECK_ABORT
usleep(1000);
}
}
@@ -59,29 +69,24 @@ static void
_start(int thread) {
pthread_t pid[thread+2];
struct monitor m;
m.count = thread;
m.m = malloc(thread * sizeof(struct skynet_monitor *));
struct monitor *m = malloc(sizeof(*m));
m->count = thread;
m->m = malloc(thread * sizeof(struct skynet_monitor *));
int i;
for (i=0;i<thread;i++) {
m.m[i] = skynet_monitor_new();
m->m[i] = skynet_monitor_new();
}
pthread_create(&pid[0], NULL, _timer, NULL);
pthread_create(&pid[1], NULL, _monitor, &m);
pthread_create(&pid[0], NULL, _monitor, m);
pthread_create(&pid[1], NULL, _timer, NULL);
for (i=0;i<thread;i++) {
pthread_create(&pid[i+2], NULL, _worker, m.m[i]);
pthread_create(&pid[i+2], NULL, _worker, m->m[i]);
}
for (i=0;i<thread+2;i++) {
for (i=1;i<thread+2;i++) {
pthread_join(pid[i], NULL);
}
for (i=0;i<thread;i++) {
skynet_monitor_delete(m.m[i]);
}
free(m.m);
}
static int