mirror of
https://github.com/cloudwu/skynet.git
synced 2026-07-24 12:20:41 +00:00
@@ -37,19 +37,20 @@ skynet_handle_register(struct skynet_context *ctx) {
|
|||||||
for (;;) {
|
for (;;) {
|
||||||
int i;
|
int i;
|
||||||
for (i=0;i<s->slot_size;i++) {
|
for (i=0;i<s->slot_size;i++) {
|
||||||
int hash = (i+s->handle_index) & (s->slot_size-1);
|
uint32_t handle = (i+s->handle_index) & HANDLE_MASK;
|
||||||
|
int hash = handle & (s->slot_size-1);
|
||||||
if (s->slot[hash] == NULL) {
|
if (s->slot[hash] == NULL) {
|
||||||
s->slot[hash] = ctx;
|
s->slot[hash] = ctx;
|
||||||
uint32_t handle = s->handle_index + i ;
|
s->handle_index = handle + 1;
|
||||||
skynet_context_init(ctx, handle);
|
|
||||||
|
|
||||||
rwlock_wunlock(&s->lock);
|
rwlock_wunlock(&s->lock);
|
||||||
|
|
||||||
s->handle_index = handle + 1;
|
handle |= s->harbor;
|
||||||
|
skynet_context_init(ctx, handle);
|
||||||
return (handle & HANDLE_MASK) | s->harbor;
|
return handle;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
assert((s->slot_size*2 - 1) <= HANDLE_MASK);
|
||||||
struct skynet_context ** new_slot = malloc(s->slot_size * 2 * sizeof(struct skynet_context *));
|
struct skynet_context ** new_slot = malloc(s->slot_size * 2 * sizeof(struct skynet_context *));
|
||||||
memset(new_slot, 0, s->slot_size * 2 * sizeof(struct skynet_context *));
|
memset(new_slot, 0, s->slot_size * 2 * sizeof(struct skynet_context *));
|
||||||
for (i=0;i<s->slot_size;i++) {
|
for (i=0;i<s->slot_size;i++) {
|
||||||
@@ -76,18 +77,17 @@ skynet_handle_retire(uint32_t handle) {
|
|||||||
s->slot[hash] = NULL;
|
s->slot[hash] = NULL;
|
||||||
|
|
||||||
int i;
|
int i;
|
||||||
int j=0;
|
int j=0, n=s->name_count;
|
||||||
for (i=0;i<s->name_count;i++,j++) {
|
for (i=0; i<n; ++i) {
|
||||||
if (s->name[i].handle == handle) {
|
if (s->name[i].handle == handle) {
|
||||||
free(s->name[i].name);
|
free(s->name[i].name);
|
||||||
++j;
|
continue;
|
||||||
--s->name_count;
|
} else if (i!=j) {
|
||||||
} else {
|
s->name[j] = s->name[i];
|
||||||
if (i!=j) {
|
|
||||||
s->name[i] = s->name[j];
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
++j;
|
||||||
}
|
}
|
||||||
|
s->name_count = j;
|
||||||
}
|
}
|
||||||
|
|
||||||
rwlock_wunlock(&s->lock);
|
rwlock_wunlock(&s->lock);
|
||||||
@@ -128,7 +128,6 @@ skynet_handle_findname(const char * name) {
|
|||||||
int c = strcmp(n->name, name);
|
int c = strcmp(n->name, name);
|
||||||
if (c==0) {
|
if (c==0) {
|
||||||
handle = n->handle;
|
handle = n->handle;
|
||||||
handle |= s->harbor;
|
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
if (c<0) {
|
if (c<0) {
|
||||||
|
|||||||
@@ -313,6 +313,36 @@ _report_zmq_error(int rc) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
static int
|
||||||
|
_isdecimal(int c) {
|
||||||
|
return c>='0' && c<='9';
|
||||||
|
}
|
||||||
|
|
||||||
|
// Name-updating protocols:
|
||||||
|
//
|
||||||
|
// 1) harbor_id=harbor_address
|
||||||
|
// 2) context_name=context_handle
|
||||||
|
static int
|
||||||
|
_split_name(uint8_t *buf, int len, int *np) {
|
||||||
|
uint8_t *sep;
|
||||||
|
if (len > 0 && _isdecimal(buf[0])) {
|
||||||
|
int i=0;
|
||||||
|
int n=0;
|
||||||
|
do {
|
||||||
|
n = n*10 + (buf[i]-'0');
|
||||||
|
} while(++i<len && _isdecimal(buf[i]));
|
||||||
|
if (i < len && buf[i] == '=') {
|
||||||
|
buf[i] = '\0';
|
||||||
|
*np = n;
|
||||||
|
return i;
|
||||||
|
}
|
||||||
|
} else if ((sep = memchr(buf, '=', len)) != NULL) {
|
||||||
|
*sep = '\0';
|
||||||
|
return (int)(sep-buf);
|
||||||
|
}
|
||||||
|
return -1;
|
||||||
|
}
|
||||||
|
|
||||||
// Always in main harbor thread
|
// Always in main harbor thread
|
||||||
static void
|
static void
|
||||||
_name_update() {
|
_name_update() {
|
||||||
@@ -321,17 +351,11 @@ _name_update() {
|
|||||||
int rc = zmq_recv(Z->zmq_local,&content,0);
|
int rc = zmq_recv(Z->zmq_local,&content,0);
|
||||||
_report_zmq_error(rc);
|
_report_zmq_error(rc);
|
||||||
int sz = zmq_msg_size(&content);
|
int sz = zmq_msg_size(&content);
|
||||||
int i;
|
|
||||||
int n = 0;
|
|
||||||
uint8_t * buffer = zmq_msg_data(&content);
|
uint8_t * buffer = zmq_msg_data(&content);
|
||||||
for (i=0;i<sz;i++) {
|
|
||||||
if (buffer[i] == '=') {
|
int n = 0;
|
||||||
buffer[i] = '\0';
|
int i = _split_name(buffer, sz, &n);
|
||||||
break;
|
if (i == -1) {
|
||||||
}
|
|
||||||
n = n * 10 + (buffer[i] - '0');
|
|
||||||
}
|
|
||||||
if (i==sz) {
|
|
||||||
char tmp[sz+1];
|
char tmp[sz+1];
|
||||||
memcpy(tmp,buffer,sz);
|
memcpy(tmp,buffer,sz);
|
||||||
tmp[sz] = '\0';
|
tmp[sz] = '\0';
|
||||||
@@ -526,6 +550,7 @@ _goback:
|
|||||||
// double check
|
// double check
|
||||||
if (!skynet_remotemq_pop(Z->queue,&msg)) {
|
if (!skynet_remotemq_pop(Z->queue,&msg)) {
|
||||||
printf("goback %x\n",msg.destination);
|
printf("goback %x\n",msg.destination);
|
||||||
|
__sync_lock_test_and_set(&Z->notice_event, 1);
|
||||||
goto _goback;
|
goto _goback;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -158,12 +158,18 @@ skynet_context_message_dispatch(void) {
|
|||||||
|
|
||||||
assert(ctx->in_global_queue);
|
assert(ctx->in_global_queue);
|
||||||
|
|
||||||
|
int re_q = 1;
|
||||||
struct skynet_message msg;
|
struct skynet_message msg;
|
||||||
if (skynet_mq_pop(q,&msg)) {
|
if (skynet_mq_pop(q,&msg)) {
|
||||||
// empty queue
|
// empty queue
|
||||||
__sync_lock_release(&ctx->in_global_queue);
|
__sync_lock_release(&ctx->in_global_queue);
|
||||||
skynet_context_release(ctx);
|
if (skynet_mq_pop(q,&msg)) {
|
||||||
return 0;
|
skynet_context_release(ctx);
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
if (__sync_lock_test_and_set(&ctx->in_global_queue, 1)) {
|
||||||
|
re_q = 0;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if (ctx->cb == NULL) {
|
if (ctx->cb == NULL) {
|
||||||
@@ -178,7 +184,9 @@ skynet_context_message_dispatch(void) {
|
|||||||
|
|
||||||
skynet_context_release(ctx);
|
skynet_context_release(ctx);
|
||||||
|
|
||||||
skynet_globalmq_push(q);
|
if (re_q) {
|
||||||
|
skynet_globalmq_push(q);
|
||||||
|
}
|
||||||
|
|
||||||
return 0;
|
return 0;
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user