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

add worker dispatch weight

Browse Source
This commit is contained in:
Cloud Wu
2014-07-08 19:56:30 +08:00
parent e60fb1d722
commit 35775685c3
3 changed files with 33 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
skynet-src/skynet_server.c
View File

@@ -245,7 +245,7 @@ skynet_context_dispatchall(struct skynet_context * ctx) {
}
struct message_queue *
skynet_context_message_dispatch(struct skynet_monitor *sm, struct message_queue *q) {
skynet_context_message_dispatch(struct skynet_monitor *sm, struct message_queue *q, int weight) {
if (q == NULL) {
q = skynet_globalmq_pop();
if (q==NULL)
@@ -261,18 +261,27 @@ skynet_context_message_dispatch(struct skynet_monitor *sm, struct message_queue
return skynet_globalmq_pop();
}
int i,n=1;
struct skynet_message msg;
if (skynet_mq_pop(q,&msg)) {
skynet_context_release(ctx);
return skynet_globalmq_pop();
}
skynet_monitor_trigger(sm, msg.source , handle);
for (i=0;i<n;i++) {
if (skynet_mq_pop(q,&msg)) {
skynet_context_release(ctx);
return skynet_globalmq_pop();
} else if (i==0 && weight >= 0) {
n = skynet_mq_length(q);
n >>= weight;
}
if (ctx->cb == NULL) {
skynet_free(msg.data);
} else {
_dispatch_message(ctx, &msg);
skynet_monitor_trigger(sm, msg.source , handle);
if (ctx->cb == NULL) {
skynet_free(msg.data);
} else {
_dispatch_message(ctx, &msg);
}
skynet_monitor_trigger(sm, 0,0);
}
assert(q == ctx->queue);
@@ -285,8 +294,6 @@ skynet_context_message_dispatch(struct skynet_monitor *sm, struct message_queue
}
skynet_context_release(ctx);
skynet_monitor_trigger(sm, 0,0);
return q;
}

2
skynet-src/skynet_server.h
View File

@@ -16,7 +16,7 @@ void skynet_context_init(struct skynet_context *, uint32_t handle);
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 *);
struct message_queue * skynet_context_message_dispatch(struct skynet_monitor *, struct message_queue *); // return next queue
struct message_queue * skynet_context_message_dispatch(struct skynet_monitor *, struct message_queue *, int weight); // return next queue
int skynet_context_total();
void skynet_context_dispatchall(struct skynet_context * context); // for skynet_error output before exit

14
skynet-src/skynet_start.c
View File

@@ -27,6 +27,7 @@ struct monitor {
struct worker_parm {
struct monitor *m;
int id;
int weight;
};
#define CHECK_ABORT if (skynet_context_total()==0) break;
@@ -118,12 +119,13 @@ static void *
_worker(void *p) {
struct worker_parm *wp = p;
int id = wp->id;
int weight = wp->weight;
struct monitor *m = wp->m;
struct skynet_monitor *sm = m->m[id];
skynet_initthread(THREAD_WORKER);
struct message_queue * q = NULL;
for (;;) {
q = skynet_context_message_dispatch(sm, q);
q = skynet_context_message_dispatch(sm, q, weight);
if (q == NULL) {
CHECK_ABORT
if (pthread_mutex_lock(&m->mutex) == 0) {
@@ -169,10 +171,20 @@ _start(int thread) {
create_thread(&pid[1], _timer, m);
create_thread(&pid[2], _socket, m);
static int weight[] = {
-1, -1, -1, -1, 0, 0, 0, 0,
1, 1, 1, 1, 1, 1, 1, 1,
2, 2, 2, 2, 2, 2, 2, 2,
3, 3, 3, 3, 3, 3, 3, 3, };
struct worker_parm wp[thread];
for (i=0;i<thread;i++) {
wp[i].m = m;
wp[i].id = i;
if (i < sizeof(weight)/sizeof(weight[0])) {
wp[i].weight= weight[i];
} else {
wp[i].weight = 0;
}
create_thread(&pid[i+3], _worker, &wp[i]);
}