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

Redesign harbor/master/dummy service

Browse Source
This commit is contained in:
Cloud Wu
2014-06-21 17:01:59 +08:00
parent 7137850eb2
commit 9937081854
16 changed files with 793 additions and 1084 deletions
Download Patch File Download Diff File Expand all files Collapse all files

319
service-src/service_dummy.c
View File

@@ -1,319 +0,0 @@
#include "skynet.h"
#include "skynet_harbor.h"
#include <string.h>
#include <assert.h>
#define HASH_SIZE 4096
#define DEFAULT_QUEUE_SIZE 1024
struct msg {
uint8_t * buffer;
size_t size;
};
struct msg_queue {
int size;
int head;
int tail;
struct msg * data;
};
struct keyvalue {
struct keyvalue * next;
char key[GLOBALNAME_LENGTH];
uint32_t hash;
uint32_t value;
struct msg_queue * queue;
};
struct hashmap {
struct keyvalue *node[HASH_SIZE];
};
/*
message type (8bits) is in destination high 8bits
harbor id (8bits) is also in that place , but remote message doesn't need harbor id.
*/
struct remote_message_header {
uint32_t source;
uint32_t destination;
uint32_t session;
};
// 12 is sizeof(struct remote_message_header)
#define HEADER_COOKIE_LENGTH 12
struct dummy {
struct skynet_context *ctx;
struct hashmap * map;
};
// hash table
static void
_push_queue(struct msg_queue * queue, const void * buffer, size_t sz, struct remote_message_header * header) {
// If there is only 1 free slot which is reserved to distinguish full/empty
// of circular buffer, expand it.
if (((queue->tail + 1) % queue->size) == queue->head) {
struct msg * new_buffer = skynet_malloc(queue->size * 2 * sizeof(struct msg));
int i;
for (i=0;i<queue->size-1;i++) {
new_buffer[i] = queue->data[(i+queue->head) % queue->size];
}
skynet_free(queue->data);
queue->data = new_buffer;
queue->head = 0;
queue->tail = queue->size - 1;
queue->size *= 2;
}
struct msg * slot = &queue->data[queue->tail];
queue->tail = (queue->tail + 1) % queue->size;
slot->buffer = skynet_malloc(sz + sizeof(*header));
memcpy(slot->buffer, buffer, sz);
memcpy(slot->buffer + sz, header, sizeof(*header));
slot->size = sz + sizeof(*header);
}
static struct msg *
_pop_queue(struct msg_queue * queue) {
if (queue->head == queue->tail) {
return NULL;
}
struct msg * slot = &queue->data[queue->head];
queue->head = (queue->head + 1) % queue->size;
return slot;
}
static struct msg_queue *
_new_queue() {
struct msg_queue * queue = skynet_malloc(sizeof(*queue));
queue->size = DEFAULT_QUEUE_SIZE;
queue->head = 0;
queue->tail = 0;
queue->data = skynet_malloc(DEFAULT_QUEUE_SIZE * sizeof(struct msg));
return queue;
}
static void
_release_queue(struct msg_queue *queue) {
if (queue == NULL)
return;
struct msg * m = _pop_queue(queue);
while (m) {
skynet_free(m->buffer);
m = _pop_queue(queue);
}
skynet_free(queue->data);
skynet_free(queue);
}
static struct keyvalue *
_hash_search(struct hashmap * hash, const char name[GLOBALNAME_LENGTH]) {
uint32_t *ptr = (uint32_t*) name;
uint32_t h = ptr[0] ^ ptr[1] ^ ptr[2] ^ ptr[3];
struct keyvalue * node = hash->node[h % HASH_SIZE];
while (node) {
if (node->hash == h && strncmp(node->key, name, GLOBALNAME_LENGTH) == 0) {
return node;
}
node = node->next;
}
return NULL;
}
static struct keyvalue *
_hash_insert(struct hashmap * hash, const char name[GLOBALNAME_LENGTH]) {
uint32_t *ptr = (uint32_t *)name;
uint32_t h = ptr[0] ^ ptr[1] ^ ptr[2] ^ ptr[3];
struct keyvalue ** pkv = &hash->node[h % HASH_SIZE];
struct keyvalue * node = skynet_malloc(sizeof(*node));
memcpy(node->key, name, GLOBALNAME_LENGTH);
node->next = *pkv;
node->queue = NULL;
node->hash = h;
node->value = 0;
*pkv = node;
return node;
}
static struct hashmap *
_hash_new() {
struct hashmap * h = skynet_malloc(sizeof(struct hashmap));
memset(h,0,sizeof(*h));
return h;
}
static void
_hash_delete(struct hashmap *hash) {
int i;
for (i=0;i<HASH_SIZE;i++) {
struct keyvalue * node = hash->node[i];
while (node) {
struct keyvalue * next = node->next;
_release_queue(node->queue);
skynet_free(node);
node = next;
}
}
skynet_free(hash);
}
///////////////
struct dummy *
dummy_create(void) {
struct dummy * d = skynet_malloc(sizeof(*d));
d->map = _hash_new();
return d;
}
void
dummy_release(struct dummy *d) {
_hash_delete(d->map);
skynet_free(d);
}
static inline void
to_bigendian(uint8_t *buffer, uint32_t n) {
buffer[0] = (n >> 24) & 0xff;
buffer[1] = (n >> 16) & 0xff;
buffer[2] = (n >> 8) & 0xff;
buffer[3] = n & 0xff;
}
static inline void
_header_to_message(const struct remote_message_header * header, uint8_t * message) {
to_bigendian(message , header->source);
to_bigendian(message+4 , header->destination);
to_bigendian(message+8 , header->session);
}
static inline uint32_t
from_bigendian(uint32_t n) {
union {
uint32_t big;
uint8_t bytes[4];
} u;
u.big = n;
return u.bytes[0] << 24 | u.bytes[1] << 16 | u.bytes[2] << 8 | u.bytes[3];
}
static inline void
_message_to_header(const uint32_t *message, struct remote_message_header *header) {
header->source = from_bigendian(message[0]);
header->destination = from_bigendian(message[1]);
header->session = from_bigendian(message[2]);
}
static void
_dispatch_queue(struct dummy *h, struct msg_queue * queue, uint32_t handle, const char name[GLOBALNAME_LENGTH] ) {
struct msg * m = _pop_queue(queue);
while (m) {
struct remote_message_header cookie;
uint8_t *ptr = m->buffer + m->size - sizeof(cookie);
memcpy(&cookie, ptr, sizeof(cookie));
int type = cookie.destination >> HANDLE_REMOTE_SHIFT;
skynet_send(h->ctx, cookie.source, handle , type | PTYPE_TAG_DONTCOPY, cookie.session, m->buffer, m->size - sizeof(cookie));
m = _pop_queue(queue);
}
}
static void
_update_name(struct dummy *h, const char name[GLOBALNAME_LENGTH], uint32_t handle) {
struct keyvalue * node = _hash_search(h->map, name);
if (node == NULL) {
node = _hash_insert(h->map, name);
}
node->value = handle;
if (node->queue) {
_dispatch_queue(h, node->queue, handle, name);
_release_queue(node->queue);
node->queue = NULL;
}
}
static void
_send_name(struct dummy *h, uint32_t source, const char name[GLOBALNAME_LENGTH], int type, int session, const char * msg, size_t sz) {
struct keyvalue * node = _hash_search(h->map, name);
if (node == NULL) {
node = _hash_insert(h->map, name);
}
if (node->value == 0) {
if (node->queue == NULL) {
node->queue = _new_queue();
}
struct remote_message_header header;
header.source = source;
header.destination = type << HANDLE_REMOTE_SHIFT;
header.session = (uint32_t)session;
_push_queue(node->queue, msg, sz, &header);
} else {
// local message
skynet_send(h->ctx, source, node->value , type | PTYPE_TAG_DONTCOPY, session, (void *)msg, sz);
}
}
static void
dummy_command(struct dummy * h, const char * msg, size_t sz, int session, uint32_t source) {
switch(msg[0]) {
case 'R' : {
// register global name
const char * name = msg + 2;
int s = (int)sz;
s -= 2;
if (s <=0 || s>= GLOBALNAME_LENGTH) {
skynet_error(h->ctx, "Invalid global name %s", name);
return;
}
struct remote_name rn;
memset(&rn, 0, sizeof(rn));
memcpy(rn.name, name, s);
rn.handle = source;
_update_name(h, rn.name, rn.handle);
break;
}
case 'C' :
case 'M' :
skynet_error(h->ctx, "Don't support harbor monitor in cluster dummy mode");
skynet_send(h->ctx, 0, source, PTYPE_ERROR, session, NULL, 0);
break;
default:
skynet_error(h->ctx, "Unknown command %s", msg);
return;
}
}
static int
_mainloop(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) {
struct dummy * h = ud;
switch (type) {
case PTYPE_SYSTEM: {
dummy_command(h, msg, sz, session, source);
return 0;
}
default: {
// remote message out
const struct remote_message *rmsg = msg;
if (rmsg->destination.handle == 0) {
_send_name(h, source , rmsg->destination.name, type, session, rmsg->message, rmsg->sz);
} else {
// local message
skynet_send(context, source, rmsg->destination.handle , type | PTYPE_TAG_DONTCOPY, session, (void *)rmsg->message, rmsg->sz);
}
return 0;
}
}
}
int
dummy_init(struct dummy *d, struct skynet_context *ctx, const char * args) {
d->ctx = ctx;
skynet_harbor_start(ctx);
skynet_callback(ctx, d, _mainloop);
return 0;
}

727
service-src/service_harbor.c
View File

@@ -2,6 +2,16 @@
#include "skynet_harbor.h"
#include "skynet_socket.h"
/*
harbor listen the PTYPE_HARBOR (in text)
N name : update the global name
S fd id: connect to new harbor , we should send self_id to fd first , and then recv a id (check it), and at last send queue.
A fd id: accept new harbor , we should send self_id to fd , and then send queue.
If the fd is disconnected, send message to slave in PTYPE_TEXT. D id
If we don't known a globalname, send message to slave in PTYPE_TEXT. Q name
*/
#include <stdio.h>
#include <stdlib.h>
#include <stdbool.h>
@@ -13,8 +23,22 @@
#define HASH_SIZE 4096
#define DEFAULT_QUEUE_SIZE 1024
// 12 is sizeof(struct remote_message_header)
#define HEADER_COOKIE_LENGTH 12
/*
message type (8bits) is in destination high 8bits
harbor id (8bits) is also in that place , but remote message doesn't need harbor id.
*/
struct remote_message_header {
uint32_t source;
uint32_t destination;
uint32_t session;
};
struct msg {
uint8_t * buffer;
struct remote_message_header header;
void * buffer;
size_t size;
};
@@ -37,99 +61,34 @@ struct hashmap {
struct keyvalue *node[HASH_SIZE];
};
/*
message type (8bits) is in destination high 8bits
harbor id (8bits) is also in that place , but remote message doesn't need harbor id.
*/
struct remote_message_header {
uint32_t source;
uint32_t destination;
uint32_t session;
};
#define STATUS_WAIT 0
#define STATUS_HANDSHAKE 1
#define STATUS_HEADER 2
#define STATUS_CONTENT 3
#define STATUS_DOWN 4
struct monitor_response {
uint32_t addr;
int session;
struct slave {
int fd;
struct msg_queue *queue;
int status;
int length;
int read;
uint8_t size[4];
char * recv_buffer;
};
struct monitor_set {
int cap;
int n;
struct monitor_response * resp;
};
// 12 is sizeof(struct remote_message_header)
#define HEADER_COOKIE_LENGTH 12
struct harbor {
struct skynet_context *ctx;
char * local_addr;
int id;
uint32_t slave;
struct hashmap * map;
int master_fd;
char * master_addr;
int remote_fd[REMOTE_MAX];
bool connected[REMOTE_MAX];
char * remote_addr[REMOTE_MAX];
struct monitor_set * monitor[REMOTE_MAX];
struct slave s[REMOTE_MAX];
};
static void
monitor_free(struct harbor *h) {
int i;
for (i=0;i<REMOTE_MAX;i++) {
struct monitor_set * m = h->monitor[i];
if (m) {
skynet_free(m->resp);
skynet_free(m);
h->monitor[i] = NULL;
}
}
}
static void
monitor_add(struct harbor *h, int id, uint32_t addr, int session) {
struct monitor_set * m = h->monitor[id];
if (m == NULL) {
m = skynet_malloc(sizeof(*m));
m->cap = 4;
m->n = 0;
m->resp = skynet_malloc(m->cap * sizeof(struct monitor_response));
h->monitor[id] = m;
}
if (m->n >= m->cap) {
assert(m->n == m->cap);
struct monitor_response * resp = skynet_malloc(m->cap * 2 * sizeof(struct monitor_response));
int i;
for (i=0;i<m->n;i++) {
resp[i] = m->resp[i];
}
m->cap *= 2;
skynet_free(m->resp);
m->resp = resp;
}
struct monitor_response * resp = &m->resp[m->n++];
resp->addr = addr;
resp->session = session;
}
static void
monitor_clear(struct harbor *h, int id) {
struct monitor_set * m = h->monitor[id];
if (m) {
int i;
for (i=0;i<m->n;i++) {
struct monitor_response * resp = &m->resp[i];
skynet_send(h->ctx, 0, resp->addr, PTYPE_RESPONSE, resp->session, NULL, 0);
}
m->n = 0;
}
}
// hash table
static void
_push_queue(struct msg_queue * queue, const void * buffer, size_t sz, struct remote_message_header * header) {
push_queue_msg(struct msg_queue * queue, struct msg * m) {
// If there is only 1 free slot which is reserved to distinguish full/empty
// of circular buffer, expand it.
if (((queue->tail + 1) % queue->size) == queue->head) {
@@ -145,16 +104,21 @@ _push_queue(struct msg_queue * queue, const void * buffer, size_t sz, struct rem
queue->size *= 2;
}
struct msg * slot = &queue->data[queue->tail];
*slot = *m;
queue->tail = (queue->tail + 1) % queue->size;
}
slot->buffer = skynet_malloc(sz + sizeof(*header));
memcpy(slot->buffer, buffer, sz);
memcpy(slot->buffer + sz, header, sizeof(*header));
slot->size = sz + sizeof(*header);
static void
push_queue(struct msg_queue * queue, void * buffer, size_t sz, struct remote_message_header * header) {
struct msg m;
m.header = *header;
m.buffer = buffer;
m.size = sz;
push_queue_msg(queue, &m);
}
static struct msg *
_pop_queue(struct msg_queue * queue) {
pop_queue(struct msg_queue * queue) {
if (queue->head == queue->tail) {
return NULL;
}
@@ -164,7 +128,7 @@ _pop_queue(struct msg_queue * queue) {
}
static struct msg_queue *
_new_queue() {
new_queue() {
struct msg_queue * queue = skynet_malloc(sizeof(*queue));
queue->size = DEFAULT_QUEUE_SIZE;
queue->head = 0;
@@ -175,20 +139,19 @@ _new_queue() {
}
static void
_release_queue(struct msg_queue *queue) {
release_queue(struct msg_queue *queue) {
if (queue == NULL)
return;
struct msg * m = _pop_queue(queue);
while (m) {
struct msg * m;
while ((m=pop_queue(queue)) != NULL) {
skynet_free(m->buffer);
m = _pop_queue(queue);
}
skynet_free(queue->data);
skynet_free(queue);
}
static struct keyvalue *
_hash_search(struct hashmap * hash, const char name[GLOBALNAME_LENGTH]) {
hash_search(struct hashmap * hash, const char name[GLOBALNAME_LENGTH]) {
uint32_t *ptr = (uint32_t*) name;
uint32_t h = ptr[0] ^ ptr[1] ^ ptr[2] ^ ptr[3];
struct keyvalue * node = hash->node[h % HASH_SIZE];
@@ -206,7 +169,7 @@ _hash_search(struct hashmap * hash, const char name[GLOBALNAME_LENGTH]) {
// Don't support erase name yet
static struct void
_hash_erase(struct hashmap * hash, char name[GLOBALNAME_LENGTH) {
hash_erase(struct hashmap * hash, char name[GLOBALNAME_LENGTH) {
uint32_t *ptr = name;
uint32_t h = ptr[0] ^ ptr[1] ^ ptr[2] ^ ptr[3];
struct keyvalue ** ptr = &hash->node[h % HASH_SIZE];
@@ -224,7 +187,7 @@ _hash_erase(struct hashmap * hash, char name[GLOBALNAME_LENGTH) {
*/
static struct keyvalue *
_hash_insert(struct hashmap * hash, const char name[GLOBALNAME_LENGTH]) {
hash_insert(struct hashmap * hash, const char name[GLOBALNAME_LENGTH]) {
uint32_t *ptr = (uint32_t *)name;
uint32_t h = ptr[0] ^ ptr[1] ^ ptr[2] ^ ptr[3];
struct keyvalue ** pkv = &hash->node[h % HASH_SIZE];
@@ -240,20 +203,20 @@ _hash_insert(struct hashmap * hash, const char name[GLOBALNAME_LENGTH]) {
}
static struct hashmap *
_hash_new() {
hash_new() {
struct hashmap * h = skynet_malloc(sizeof(struct hashmap));
memset(h,0,sizeof(*h));
return h;
}
static void
_hash_delete(struct hashmap *hash) {
hash_delete(struct hashmap *hash) {
int i;
for (i=0;i<HASH_SIZE;i++) {
struct keyvalue * node = hash->node[i];
while (node) {
struct keyvalue * next = node->next;
_release_queue(node->queue);
release_queue(node->queue);
skynet_free(node);
node = next;
}
@@ -263,62 +226,49 @@ _hash_delete(struct hashmap *hash) {
///////////////
static void
close_harbor(struct harbor *h, int id) {
struct slave *s = &h->s[id];
s->status = STATUS_DOWN;
if (s->fd) {
skynet_socket_close(h->ctx, s->fd);
}
if (s->queue) {
release_queue(s->queue);
s->queue = NULL;
}
}
static void
report_harbor_down(struct harbor *h, int id) {
char down[64];
int n = sprintf(down, "D %d",id);
skynet_send(h->ctx, 0, h->slave, PTYPE_TEXT, 0, down, n);
}
struct harbor *
harbor_create(void) {
struct harbor * h = skynet_malloc(sizeof(*h));
h->ctx = NULL;
h->id = 0;
h->master_fd = -1;
h->master_addr = NULL;
int i;
for (i=0;i<REMOTE_MAX;i++) {
h->remote_fd[i] = -1;
h->connected[i] = false;
h->remote_addr[i] = NULL;
h->monitor[i] = NULL;
}
h->map = _hash_new();
memset(h,0,sizeof(*h));
h->map = hash_new();
return h;
}
void
harbor_release(struct harbor *h) {
struct skynet_context *ctx = h->ctx;
if (h->master_fd >= 0) {
skynet_socket_close(ctx, h->master_fd);
}
skynet_free(h->master_addr);
skynet_free(h->local_addr);
int i;
for (i=0;i<REMOTE_MAX;i++) {
if (h->remote_fd[i] >= 0) {
skynet_socket_close(ctx, h->remote_fd[i]);
skynet_free(h->remote_addr[i]);
for (i=1;i<REMOTE_MAX;i++) {
struct slave *s = &h->s[i];
if (s->fd && s->status != STATUS_DOWN) {
close_harbor(h,i);
report_harbor_down(h,i);
}
}
_hash_delete(h->map);
monitor_free(h);
hash_delete(h->map);
skynet_free(h);
}
static int
_connect_to(struct harbor *h, const char *ipaddress) {
char * port = strchr(ipaddress,':');
if (port==NULL) {
return -1;
}
int sz = port - ipaddress;
char tmp[sz + 1];
memcpy(tmp,ipaddress,sz);
tmp[sz] = '\0';
int portid = strtol(port+1, NULL,10);
skynet_error(h->ctx, "Harbor(%d) connect to %s:%d", h->id, tmp, portid);
return skynet_socket_connect(h->ctx, tmp, portid);
}
static inline void
to_bigendian(uint8_t *buffer, uint32_t n) {
buffer[0] = (n >> 24) & 0xff;
@@ -328,7 +278,7 @@ to_bigendian(uint8_t *buffer, uint32_t n) {
}
static inline void
_header_to_message(const struct remote_message_header * header, uint8_t * message) {
header_to_message(const struct remote_message_header * header, uint8_t * message) {
to_bigendian(message , header->source);
to_bigendian(message+4 , header->destination);
to_bigendian(message+8 , header->session);
@@ -345,122 +295,200 @@ from_bigendian(uint32_t n) {
}
static inline void
_message_to_header(const uint32_t *message, struct remote_message_header *header) {
message_to_header(const uint32_t *message, struct remote_message_header *header) {
header->source = from_bigendian(message[0]);
header->destination = from_bigendian(message[1]);
header->session = from_bigendian(message[2]);
}
static void
_send_package(struct skynet_context *ctx, int fd, const void * buffer, size_t sz) {
uint8_t * sendbuf = skynet_malloc(sz+4);
to_bigendian(sendbuf, sz);
memcpy(sendbuf+4, buffer, sz);
// socket package
if (skynet_socket_send(ctx, fd, sendbuf, sz+4)) {
skynet_error(ctx, "Send to %d error", fd);
}
static void
forward_local_messsage(struct harbor *h, void *msg, int sz) {
const char * cookie = msg;
cookie += sz - HEADER_COOKIE_LENGTH;
struct remote_message_header header;
message_to_header((const uint32_t *)cookie, &header);
uint32_t destination = header.destination;
int type = (destination >> HANDLE_REMOTE_SHIFT) | PTYPE_TAG_DONTCOPY;
destination = (destination & HANDLE_MASK) | ((uint32_t)h->id << HANDLE_REMOTE_SHIFT);
skynet_send(h->ctx, header.source, destination, type, (int)header.session, (void *)msg, sz-HEADER_COOKIE_LENGTH);
}
static void
_send_remote(struct skynet_context * ctx, int fd, const char * buffer, size_t sz, struct remote_message_header * cookie) {
send_remote(struct skynet_context * ctx, int fd, const char * buffer, size_t sz, struct remote_message_header * cookie) {
uint32_t sz_header = sz+sizeof(*cookie);
uint8_t * sendbuf = skynet_malloc(sz_header+4);
to_bigendian(sendbuf, sz_header);
memcpy(sendbuf+4, buffer, sz);
_header_to_message(cookie, sendbuf+4+sz);
header_to_message(cookie, sendbuf+4+sz);
if (skynet_socket_send(ctx, fd, sendbuf, sz_header+4)) {
skynet_error(ctx, "Remote send to %d error", fd);
}
// ignore send error, because if the connection is broken, the mainloop will recv a message.
skynet_socket_send(ctx, fd, sendbuf, sz_header+4);
}
static void
response_close(struct harbor *h, int id) {
if (h->connected[id]) {
monitor_clear(h, id);
}
h->connected[id] = false;
}
static void
_update_remote_address(struct harbor *h, int harbor_id, const char * ipaddr) {
if (harbor_id == h->id) {
return;
}
assert(harbor_id > 0 && harbor_id< REMOTE_MAX);
struct skynet_context * context = h->ctx;
if (h->remote_fd[harbor_id] >=0) {
skynet_socket_close(context, h->remote_fd[harbor_id]);
skynet_free(h->remote_addr[harbor_id]);
h->remote_addr[harbor_id] = NULL;
}
h->remote_fd[harbor_id] = _connect_to(h, ipaddr);
response_close(h, harbor_id);
}
static void
_dispatch_queue(struct harbor *h, struct msg_queue * queue, uint32_t handle, const char name[GLOBALNAME_LENGTH] ) {
dispatch_name_queue(struct harbor *h, struct keyvalue * node) {
struct msg_queue * queue = node->queue;
uint32_t handle = node->value;
int harbor_id = handle >> HANDLE_REMOTE_SHIFT;
assert(harbor_id != 0);
struct skynet_context * context = h->ctx;
int fd = h->remote_fd[harbor_id];
if (fd < 0) {
char tmp [GLOBALNAME_LENGTH+1];
memcpy(tmp, name , GLOBALNAME_LENGTH);
tmp[GLOBALNAME_LENGTH] = '\0';
skynet_error(context, "Drop message to %s (in harbor %d)",tmp,harbor_id);
struct slave *s = &h->s[harbor_id];
int fd = s->fd;
if (fd == 0) {
if (s->status == STATUS_DOWN) {
char tmp [GLOBALNAME_LENGTH+1];
memcpy(tmp, node->key, GLOBALNAME_LENGTH);
tmp[GLOBALNAME_LENGTH] = '\0';
skynet_error(context, "Drop message to %s (in harbor %d)",tmp,harbor_id);
} else {
if (s->queue == NULL) {
s->queue = node->queue;
node->queue = NULL;
} else {
struct msg * m;
while ((m = pop_queue(queue))!=NULL) {
push_queue_msg(s->queue, m);
}
}
}
return;
}
struct msg * m = _pop_queue(queue);
while (m) {
struct remote_message_header cookie;
uint8_t *ptr = m->buffer + m->size - sizeof(cookie);
memcpy(&cookie, ptr, sizeof(cookie));
cookie.destination |= (handle & HANDLE_MASK);
_header_to_message(&cookie, ptr);
_send_package(context, fd, m->buffer, m->size);
m = _pop_queue(queue);
struct msg * m;
while ((m = pop_queue(queue)) != NULL) {
m->header.destination |= (handle & HANDLE_MASK);
send_remote(context, fd, m->buffer, m->size, &m->header);
}
}
static void
_update_remote_name(struct harbor *h, const char name[GLOBALNAME_LENGTH], uint32_t handle) {
struct keyvalue * node = _hash_search(h->map, name);
dispatch_queue(struct harbor *h, int id) {
struct slave *s = &h->s[id];
int fd = s->fd;
assert(fd != 0);
struct msg_queue *queue = s->queue;
if (queue == NULL)
return;
struct msg * m;
while ((m = pop_queue(queue)) != NULL) {
send_remote(h->ctx, fd, m->buffer, m->size, &m->header);
}
release_queue(queue);
s->queue = NULL;
}
static void
push_socket_data(struct harbor *h, const struct skynet_socket_message * message) {
assert(message->type == SKYNET_SOCKET_TYPE_DATA);
int fd = message->id;
int i;
int id = 0;
struct slave * s = NULL;
for (i=1;i<REMOTE_MAX;i++) {
if (h->s[i].fd == fd) {
s = &h->s[i];
id = i;
break;
}
}
if (s == NULL) {
skynet_free(message->buffer);
skynet_error(h->ctx, "Invalid socket fd (%d) data", fd);
return;
}
uint8_t * buffer = (uint8_t *)message->buffer;
int size = message->ud;
for (;;) {
switch(s->status) {
case STATUS_HANDSHAKE: {
// check id
uint8_t remote_id = buffer[0];
if (remote_id != id) {
skynet_error(h->ctx, "Invalid shakehand id (%d) from fd = %d , harbor = %d", id, fd, remote_id);
close_harbor(h,id);
return;
}
++buffer;
--size;
s->status = STATUS_HEADER;
dispatch_queue(h, id);
if (size == 0) {
break;
}
// go though
}
case STATUS_HEADER:
if (size < 4) {
s->length = size;
memcpy(s->size, buffer, size);
return;
} else {
// big endian 4 bytes length
if (buffer[0] != 0) {
skynet_error(h->ctx, "Message is too long from harbor %d", id);
close_harbor(h,id);
return;
}
s->length = buffer[1] << 16 | buffer[2] << 8 | buffer[3];
s->read = 0;
s->recv_buffer = skynet_malloc(s->length);
s->status = STATUS_CONTENT;
buffer += 4;
size -= 4;
if (size == 0) {
return;
}
}
// go though
case STATUS_CONTENT:
if (size < s->length - s->read) {
memcpy(s->recv_buffer + s->read, buffer, size);
s->read += size;
return;
}
int need = s->length - s->read;
memcpy(s->recv_buffer + s->read, buffer, need);
forward_local_messsage(h, s->recv_buffer, s->length);
s->length = 0;
s->read = 0;
s->recv_buffer = NULL;
size -= need;
buffer += need;
s->status = STATUS_HEADER;
if (size == 0)
return;
break;
default:
return;
}
}
}
static void
update_name(struct harbor *h, const char name[GLOBALNAME_LENGTH], uint32_t handle) {
struct keyvalue * node = hash_search(h->map, name);
if (node == NULL) {
node = _hash_insert(h->map, name);
node = hash_insert(h->map, name);
}
node->value = handle;
if (node->queue) {
_dispatch_queue(h, node->queue, handle, name);
_release_queue(node->queue);
dispatch_name_queue(h, node);
release_queue(node->queue);
node->queue = NULL;
}
}
static void
_request_master(struct harbor *h, const char name[GLOBALNAME_LENGTH], size_t i, uint32_t handle) {
uint8_t buffer[4+i];
to_bigendian(buffer, handle);
memcpy(buffer+4,name,i);
if (h->master_fd >= 0) {
_send_package(h->ctx, h->master_fd, buffer, 4+i);
}
}
/*
update global name to master
2 bytes (size)
4 bytes (handle) (handle == 0 for request)
n bytes string (name)
*/
static int
_remote_send_handle(struct harbor *h, uint32_t source, uint32_t destination, int type, int session, const char * msg, size_t sz) {
remote_send_handle(struct harbor *h, uint32_t source, uint32_t destination, int type, int session, const char * msg, size_t sz) {
int harbor_id = destination >> HANDLE_REMOTE_SHIFT;
assert(harbor_id != 0);
struct skynet_context * context = h->ctx;
if (harbor_id == h->id) {
// local message
@@ -468,95 +496,63 @@ _remote_send_handle(struct harbor *h, uint32_t source, uint32_t destination, int
return 1;
}
int fd = h->remote_fd[harbor_id];
if (fd >= 0 && h->connected[harbor_id]) {
struct slave * s = &h->s[harbor_id];
if (s->fd == 0 || s->status == STATUS_HANDSHAKE) {
if (s->status == STATUS_DOWN) {
// throw an error return to source
// report the destination is dead
skynet_send(context, destination, source, PTYPE_ERROR, 0 , NULL, 0);
skynet_error(context, "Drop message to harbor %d from %x to %x (session = %d, msgsz = %d)",harbor_id, source, destination,session,(int)sz);
} else {
struct remote_message_header header;
header.source = source;
header.destination = type << HANDLE_REMOTE_SHIFT;
header.session = (uint32_t)session;
push_queue(s->queue, (void *)msg, sz, &header);
return 1;
}
} else {
struct remote_message_header cookie;
cookie.source = source;
cookie.destination = (destination & HANDLE_MASK) | ((uint32_t)type << HANDLE_REMOTE_SHIFT);
cookie.session = (uint32_t)session;
_send_remote(context, fd, msg,sz,&cookie);
} else {
// throw an error return to source
// report the destination is dead
skynet_send(context, destination, source, PTYPE_ERROR, 0 , NULL, 0);
skynet_error(context, "Drop message to harbor %d from %x to %x (session = %d, msgsz = %d)",harbor_id, source, destination,session,(int)sz);
send_remote(context, s->fd, msg,sz,&cookie);
}
return 0;
}
static void
_remote_register_name(struct harbor *h, const char name[GLOBALNAME_LENGTH], uint32_t handle) {
int i;
for (i=0;i<GLOBALNAME_LENGTH;i++) {
if (name[i] == '\0')
break;
}
if (handle != 0) {
_update_remote_name(h, name, handle);
}
_request_master(h, name,i,handle);
}
static int
_remote_send_name(struct harbor *h, uint32_t source, const char name[GLOBALNAME_LENGTH], int type, int session, const char * msg, size_t sz) {
struct keyvalue * node = _hash_search(h->map, name);
remote_send_name(struct harbor *h, uint32_t source, const char name[GLOBALNAME_LENGTH], int type, int session, const char * msg, size_t sz) {
struct keyvalue * node = hash_search(h->map, name);
if (node == NULL) {
node = _hash_insert(h->map, name);
node = hash_insert(h->map, name);
}
if (node->value == 0) {
if (node->queue == NULL) {
node->queue = _new_queue();
node->queue = new_queue();
}
struct remote_message_header header;
header.source = source;
header.destination = type << HANDLE_REMOTE_SHIFT;
header.session = (uint32_t)session;
_push_queue(node->queue, msg, sz, &header);
// 0 for request
_remote_register_name(h, name, 0);
push_queue(node->queue, (void *)msg, sz, &header);
char query[2+GLOBALNAME_LENGTH+1] = "Q ";
query[2+GLOBALNAME_LENGTH] = 0;
memcpy(query+2, name, GLOBALNAME_LENGTH);
skynet_send(h->ctx, 0, h->slave, PTYPE_TEXT, 0, query, strlen(query));
return 1;
} else {
return _remote_send_handle(h, source, node->value, type, session, msg, sz);
return remote_send_handle(h, source, node->value, type, session, msg, sz);
}
}
static int
harbor_id(struct harbor *h, int fd) {
int i;
for (i=1;i<REMOTE_MAX;i++) {
if (h->remote_fd[i] == fd)
return i;
}
return 0;
}
static void
close_harbor(struct harbor *h, int fd) {
if (fd == h->master_fd) {
skynet_socket_close(h->ctx, fd);
skynet_error(h->ctx, "Master disconnected");
h->master_fd = -1;
return;
}
int id = harbor_id(h,fd);
if (id == 0)
return;
skynet_error(h->ctx, "Harbor %d closed",id);
skynet_socket_close(h->ctx, fd);
h->remote_fd[id] = -1;
response_close(h,id);
}
static void
open_harbor(struct harbor *h, int fd) {
int id = harbor_id(h,fd);
if (id == 0)
return;
assert(h->connected[id] == false);
monitor_clear(h, id);
h->connected[id] = true;
skynet_error(h->ctx, "Harbor(%d) connected", id);
handshake(struct harbor *h, int id) {
struct slave *s = &h->s[id];
uint8_t * handshake = skynet_malloc(1);
handshake[0] = (uint8_t)h->id;
skynet_socket_send(h->ctx, s->fd, handshake, 1);
}
static void
@@ -565,8 +561,7 @@ harbor_command(struct harbor * h, const char * msg, size_t sz, int session, uint
int s = (int)sz;
s -= 2;
switch(msg[0]) {
case 'R' : {
// register global name
case 'N' : {
if (s <=0 || s>= GLOBALNAME_LENGTH) {
skynet_error(h->ctx, "Invalid global name %s", name);
return;
@@ -575,35 +570,35 @@ harbor_command(struct harbor * h, const char * msg, size_t sz, int session, uint
memset(&rn, 0, sizeof(rn));
memcpy(rn.name, name, s);
rn.handle = source;
_remote_register_name(h, rn.name, rn.handle);
update_name(h, rn.name, rn.handle);
break;
}
case 'C' :
case 'M' : {
if (s <= 0) {
skynet_error(h->ctx, "Invalid harbor montior");
skynet_send(h->ctx, 0, source, PTYPE_ERROR, session, NULL, 0);
case 'S' :
case 'A' : {
char buffer[s+1];
memcpy(buffer, name, s);
buffer[s] = 0;
int fd=0, id=0;
sscanf(buffer, "%d %d",&fd,&id);
if (fd == 0 || id <= 0 || id>=REMOTE_MAX) {
skynet_error(h->ctx, "Invalid command %c %s", msg[0], buffer);
return;
}
int hid = strtol(name, NULL, 10);
if (hid <= 0 || hid >= REMOTE_MAX) {
skynet_error(h->ctx, "Invalid harbor montior id : %s", name);
skynet_send(h->ctx, 0, source, PTYPE_ERROR, session, NULL, 0);
struct slave * slave = &h->s[id];
if (slave->fd != 0) {
skynet_error(h->ctx, "Harbor %d alreay exist", id);
return;
}
if (msg[0] == 'M') {
if (!h->connected[hid]) {
skynet_send(h->ctx, 0, source, PTYPE_RESPONSE, session, NULL, 0);
return;
}
slave->fd = fd;
skynet_socket_start(h->ctx, fd);
handshake(h, id);
if (msg[0] == 'S') {
slave->status = STATUS_HANDSHAKE;
} else {
assert(msg[0] == 'C');
if (h->connected[hid]) {
skynet_send(h->ctx, 0, source, PTYPE_RESPONSE, session, NULL, 0);
return;
}
slave->status = STATUS_HEADER;
dispatch_queue(h,id);
}
monitor_add(h, hid, source, session);
break;
}
default:
@@ -613,63 +608,48 @@ harbor_command(struct harbor * h, const char * msg, size_t sz, int session, uint
}
static int
_mainloop(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) {
harbor_id(struct harbor *h, int fd) {
int i;
for (i=1;i<REMOTE_MAX;i++) {
struct slave *s = &h->s[i];
if (s->fd == fd) {
return i;
}
}
return 0;
}
static int
mainloop(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) {
struct harbor * h = ud;
switch (type) {
case PTYPE_SOCKET: {
const struct skynet_socket_message * message = msg;
switch(message->type) {
case SKYNET_SOCKET_TYPE_DATA:
push_socket_data(h, message);
skynet_free(message->buffer);
skynet_error(context, "recv invalid socket message (size=%d)", message->ud);
break;
case SKYNET_SOCKET_TYPE_ACCEPT:
skynet_error(context, "recv invalid socket accept message");
break;
case SKYNET_SOCKET_TYPE_ERROR:
case SKYNET_SOCKET_TYPE_CLOSE:
close_harbor(h, message->id);
case SKYNET_SOCKET_TYPE_CLOSE: {
int id = harbor_id(h, message->id);
if (id) {
report_harbor_down(h,id);
} else {
skynet_error(context, "Unkown fd (%d) closed", message->id);
}
break;
}
case SKYNET_SOCKET_TYPE_CONNECT:
open_harbor(h, message->id);
// fd forward to this service
break;
default:
skynet_error(context, "recv invalid socket message type %d", type);
break;
}
return 0;
}
case PTYPE_HARBOR: {
// remote message in
const char * cookie = msg;
cookie += sz - HEADER_COOKIE_LENGTH;
struct remote_message_header header;
_message_to_header((const uint32_t *)cookie, &header);
if (header.source == 0) {
if (header.destination < REMOTE_MAX) {
// 1 byte harbor id (0~255)
// update remote harbor address
char ip [sz - HEADER_COOKIE_LENGTH + 1];
memcpy(ip, msg, sz-HEADER_COOKIE_LENGTH);
ip[sz-HEADER_COOKIE_LENGTH] = '\0';
_update_remote_address(h, header.destination, ip);
} else {
// update global name
if (sz - HEADER_COOKIE_LENGTH > GLOBALNAME_LENGTH) {
char name[sz-HEADER_COOKIE_LENGTH+1];
memcpy(name, msg, sz-HEADER_COOKIE_LENGTH);
name[sz-HEADER_COOKIE_LENGTH] = '\0';
skynet_error(context, "Global name is too long %s", name);
}
_update_remote_name(h, msg, header.destination);
}
} else {
uint32_t destination = header.destination;
int type = (destination >> HANDLE_REMOTE_SHIFT) | PTYPE_TAG_DONTCOPY;
destination = (destination & HANDLE_MASK) | ((uint32_t)h->id << HANDLE_REMOTE_SHIFT);
skynet_send(context, header.source, destination, type, (int)header.session, (void *)msg, sz-HEADER_COOKIE_LENGTH);
return 1;
}
return 0;
}
case PTYPE_SYSTEM: {
harbor_command(h, msg,sz,session,source);
return 0;
}
@@ -677,11 +657,11 @@ _mainloop(struct skynet_context * context, void * ud, int type, int session, uin
// remote message out
const struct remote_message *rmsg = msg;
if (rmsg->destination.handle == 0) {
if (_remote_send_name(h, source , rmsg->destination.name, type, session, rmsg->message, rmsg->sz)) {
if (remote_send_name(h, source , rmsg->destination.name, type, session, rmsg->message, rmsg->sz)) {
return 0;
}
} else {
if (_remote_send_handle(h, source , rmsg->destination.handle, type, session, rmsg->message, rmsg->sz)) {
if (remote_send_handle(h, source , rmsg->destination.handle, type, session, rmsg->message, rmsg->sz)) {
return 0;
}
}
@@ -691,48 +671,19 @@ _mainloop(struct skynet_context * context, void * ud, int type, int session, uin
}
}
static void
_launch_gate(struct skynet_context * ctx, const char * local_addr) {
char tmp[128];
sprintf(tmp,"gate L ! %s %d %d 0",local_addr, PTYPE_HARBOR, REMOTE_MAX);
const char * gate_addr = skynet_command(ctx, "LAUNCH", tmp);
if (gate_addr == NULL) {
fprintf(stderr, "Harbor : launch gate failed\n");
exit(1);
}
uint32_t gate = strtoul(gate_addr+1 , NULL, 16);
if (gate == 0) {
fprintf(stderr, "Harbor : launch gate invalid %s", gate_addr);
exit(1);
}
const char * self_addr = skynet_command(ctx, "REG", NULL);
int n = sprintf(tmp,"broker %s",self_addr);
skynet_send(ctx, 0, gate, PTYPE_TEXT, 0, tmp, n);
skynet_send(ctx, 0, gate, PTYPE_TEXT, 0, "start", 5);
}
int
harbor_init(struct harbor *h, struct skynet_context *ctx, const char * args) {
h->ctx = ctx;
int sz = strlen(args)+1;
char master_addr[sz];
char local_addr[sz];
int harbor_id = 0;
sscanf(args,"%s %s %d",master_addr, local_addr, &harbor_id);
h->master_addr = skynet_strdup(master_addr);
h->id = harbor_id;
h->master_fd = _connect_to(h, master_addr);
if (h->master_fd == -1) {
fprintf(stderr, "Harbor: Connect to master failed\n");
exit(1);
uint32_t slave = 0;
sscanf(args,"%d %u", &harbor_id, &slave);
if (slave == 0) {
return 1;
}
h->id = harbor_id;
h->slave = slave;
skynet_callback(ctx, h, mainloop);
skynet_harbor_start(ctx);
h->local_addr = skynet_strdup(local_addr);
_launch_gate(ctx, local_addr);
skynet_callback(ctx, h, _mainloop);
_request_master(h, local_addr, strlen(local_addr), harbor_id);
return 0;
}

314
service-src/service_master.c
View File

@@ -1,314 +0,0 @@
#include "skynet.h"
#include "skynet_harbor.h"
#include "skynet_socket.h"
#include <stdio.h>
#include <stdlib.h>
#include <stdbool.h>
#include <string.h>
#include <assert.h>
#include <stdint.h>
#define HASH_SIZE 4096
struct name {
struct name * next;
char key[GLOBALNAME_LENGTH];
uint32_t hash;
uint32_t value;
};
struct namemap {
struct name *node[HASH_SIZE];
};
struct master {
struct skynet_context *ctx;
int remote_fd[REMOTE_MAX];
bool connected[REMOTE_MAX];
char * remote_addr[REMOTE_MAX];
struct namemap map;
};
struct master *
master_create() {
struct master *m = skynet_malloc(sizeof(*m));
int i;
for (i=0;i<REMOTE_MAX;i++) {
m->remote_fd[i] = -1;
m->remote_addr[i] = NULL;
m->connected[i] = false;
}
memset(&m->map, 0, sizeof(m->map));
return m;
}
void
master_release(struct master * m) {
int i;
struct skynet_context *ctx = m->ctx;
for (i=0;i<REMOTE_MAX;i++) {
int fd = m->remote_fd[i];
if (fd >= 0) {
assert(ctx);
skynet_socket_close(ctx, fd);
}
skynet_free(m->remote_addr[i]);
}
for (i=0;i<HASH_SIZE;i++) {
struct name * node = m->map.node[i];
while (node) {
struct name * next = node->next;
skynet_free(node);
node = next;
}
}
skynet_free(m);
}
static struct name *
_search_name(struct master *m, char name[GLOBALNAME_LENGTH]) {
uint32_t *ptr = (uint32_t *) name;
uint32_t h = ptr[0] ^ ptr[1] ^ ptr[2] ^ ptr[3];
struct name * node = m->map.node[h % HASH_SIZE];
while (node) {
if (node->hash == h && strncmp(node->key, name, GLOBALNAME_LENGTH) == 0) {
return node;
}
node = node->next;
}
return NULL;
}
static struct name *
_insert_name(struct master *m, char name[GLOBALNAME_LENGTH]) {
uint32_t *ptr = (uint32_t *)name;
uint32_t h = ptr[0] ^ ptr[1] ^ ptr[2] ^ ptr[3];
struct name **pname = &m->map.node[h % HASH_SIZE];
struct name * node = skynet_malloc(sizeof(*node));
memcpy(node->key, name, GLOBALNAME_LENGTH);
node->next = *pname;
node->hash = h;
node->value = 0;
*pname = node;
return node;
}
static void
_copy_name(char *name, const char * buffer, size_t sz) {
if (sz < GLOBALNAME_LENGTH) {
memcpy(name, buffer, sz);
memset(name+sz, 0 , GLOBALNAME_LENGTH - sz);
} else {
memcpy(name, buffer, GLOBALNAME_LENGTH);
}
}
static void
_connect_to(struct master *m, int id) {
assert(m->connected[id] == false);
struct skynet_context * ctx = m->ctx;
const char *ipaddress = m->remote_addr[id];
char * portstr = strchr(ipaddress,':');
if (portstr==NULL) {
skynet_error(ctx, "Harbor %d : address invalid (%s)",id, ipaddress);
return;
}
int sz = portstr - ipaddress;
char tmp[sz + 1];
memcpy(tmp,ipaddress,sz);
tmp[sz] = '\0';
int port = strtol(portstr+1,NULL,10);
skynet_error(ctx, "Master connect to harbor(%d) %s:%d", id, tmp, port);
m->remote_fd[id] = skynet_socket_connect(ctx, tmp, port);
m->connected[id] = true;
}
static inline void
to_bigendian(uint8_t *buffer, uint32_t n) {
buffer[0] = (n >> 24) & 0xff;
buffer[1] = (n >> 16) & 0xff;
buffer[2] = (n >> 8) & 0xff;
buffer[3] = n & 0xff;
}
static void
_send_to(struct master *m, int id, const void * buf, int sz, uint32_t handle) {
uint8_t * buffer= (uint8_t *)skynet_malloc(4 + sz + 12);
to_bigendian(buffer, sz+12);
memcpy(buffer+4, buf, sz);
to_bigendian(buffer+4+sz, 0);
to_bigendian(buffer+4+sz+4, handle);
to_bigendian(buffer+4+sz+8, 0);
sz += 4 + 12;
if (skynet_socket_send(m->ctx, m->remote_fd[id], buffer, sz)) {
skynet_error(m->ctx, "Harbor %d : send error", id);
}
}
static void
_broadcast(struct master *m, const char *name, size_t sz, uint32_t handle) {
int i;
for (i=1;i<REMOTE_MAX;i++) {
int fd = m->remote_fd[i];
if (fd < 0 || m->connected[i]==false)
continue;
_send_to(m, i , name, sz, handle);
}
}
static void
_request_name(struct master *m, const char * buffer, size_t sz) {
char name[GLOBALNAME_LENGTH];
_copy_name(name, buffer, sz);
struct name * n = _search_name(m, name);
if (n == NULL) {
return;
}
_broadcast(m, name, GLOBALNAME_LENGTH, n->value);
}
static void
_update_name(struct master *m, uint32_t handle, const char * buffer, size_t sz) {
char name[GLOBALNAME_LENGTH];
_copy_name(name, buffer, sz);
struct name * n = _search_name(m, name);
if (n==NULL) {
n = _insert_name(m,name);
}
n->value = handle;
_broadcast(m,name,GLOBALNAME_LENGTH, handle);
}
static void
close_harbor(struct master *m, int harbor_id) {
if (m->connected[harbor_id]) {
struct skynet_context * context = m->ctx;
skynet_socket_close(context, m->remote_fd[harbor_id]);
m->remote_fd[harbor_id] = -1;
m->connected[harbor_id] = false;
}
}
static void
_update_address(struct master *m, int harbor_id, const char * buffer, size_t sz) {
if (m->remote_fd[harbor_id] >= 0) {
close_harbor(m, harbor_id);
}
skynet_free(m->remote_addr[harbor_id]);
char * addr = skynet_malloc(sz+1);
memcpy(addr, buffer, sz);
addr[sz] = '\0';
m->remote_addr[harbor_id] = addr;
_connect_to(m, harbor_id);
}
static int
socket_id(struct master *m, int id) {
int i;
for (i=1;i<REMOTE_MAX;i++) {
if (m->remote_fd[i] == id)
return i;
}
return 0;
}
static void
on_connected(struct master *m, int id) {
_broadcast(m, m->remote_addr[id], strlen(m->remote_addr[id]), id);
// m->connected[id] = true;
int i;
for (i=1;i<REMOTE_MAX;i++) {
if (i == id)
continue;
const char * addr = m->remote_addr[i];
if (addr == NULL || m->connected[i] == false) {
continue;
}
_send_to(m, id , addr, strlen(addr), i);
}
}
static void
dispatch_socket(struct master *m, const struct skynet_socket_message *msg, int sz) {
int id = socket_id(m, msg->id);
switch(msg->type) {
case SKYNET_SOCKET_TYPE_CONNECT:
assert(id);
on_connected(m, id);
break;
case SKYNET_SOCKET_TYPE_ERROR:
skynet_error(m->ctx, "socket error on harbor %d", id);
// go though, close socket
case SKYNET_SOCKET_TYPE_CLOSE:
close_harbor(m, id);
break;
default:
skynet_error(m->ctx, "Invalid socket message type %d", msg->type);
break;
}
}
/*
update global name to master
4 bytes (handle) (handle == 0 for request)
n bytes string (name)
*/
static int
_mainloop(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) {
if (type == PTYPE_SOCKET) {
dispatch_socket(ud, msg, (int)sz);
return 0;
}
if (type != PTYPE_HARBOR) {
skynet_error(context, "None harbor message recv from %x (type = %d)", source, type);
return 0;
}
assert(sz >= 4);
struct master *m = ud;
const uint8_t *handlen = msg;
uint32_t handle = handlen[0]<<24 | handlen[1]<<16 | handlen[2]<<8 | handlen[3];
sz -= 4;
const char * name = msg;
name += 4;
if (handle == 0) {
_request_name(m , name, sz);
} else if (handle < REMOTE_MAX) {
_update_address(m , handle, name, sz);
} else {
_update_name(m , handle, name, sz);
}
return 0;
}
int
master_init(struct master *m, struct skynet_context *ctx, const char * args) {
char tmp[strlen(args) + 32];
sprintf(tmp,"gate L ! %s %d %d 0",args,PTYPE_HARBOR,REMOTE_MAX);
const char * gate_addr = skynet_command(ctx, "LAUNCH", tmp);
if (gate_addr == NULL) {
skynet_error(ctx, "Master : launch gate failed");
return 1;
}
uint32_t gate = strtoul(gate_addr+1, NULL, 16);
if (gate == 0) {
skynet_error(ctx, "Master : launch gate invalid %s", gate_addr);
return 1;
}
const char * self_addr = skynet_command(ctx, "REG", NULL);
int n = sprintf(tmp,"broker %s",self_addr);
skynet_send(ctx, 0, gate, PTYPE_TEXT, 0, tmp, n);
skynet_send(ctx, 0, gate, PTYPE_TEXT, 0, "start", 5);
skynet_callback(ctx, m, _mainloop);
m->ctx = ctx;
return 0;
}