mirror of
https://github.com/cloudwu/skynet.git
synced 2026-07-24 20:23:06 +00:00
Rewrite harbor and don't need zeromq :)
This commit is contained in:
云风
272
service-src/service_master.c
Normal file
272
service-src/service_master.c
Normal file
@@ -0,0 +1,272 @@
|
||||
#include "skynet.h"
|
||||
#include "skynet_harbor.h"
|
||||
|
||||
#include <sys/types.h>
|
||||
#include <sys/socket.h>
|
||||
#include <arpa/inet.h>
|
||||
#include <unistd.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <assert.h>
|
||||
#include <stdint.h>
|
||||
#include <errno.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 {
|
||||
int remote_fd[REMOTE_MAX];
|
||||
char * remote_addr[REMOTE_MAX];
|
||||
struct namemap map;
|
||||
};
|
||||
|
||||
struct master *
|
||||
master_create() {
|
||||
struct master *m = malloc(sizeof(*m));
|
||||
int i;
|
||||
for (i=0;i<REMOTE_MAX;i++) {
|
||||
m->remote_fd[i] = -1;
|
||||
m->remote_addr[i] = NULL;
|
||||
}
|
||||
memset(&m->map, 0, sizeof(m->map));
|
||||
return m;
|
||||
}
|
||||
|
||||
void
|
||||
master_release(struct master * m) {
|
||||
int i;
|
||||
for (i=0;i<REMOTE_MAX;i++) {
|
||||
int fd = m->remote_fd[i];
|
||||
if (fd < 0) {
|
||||
close(fd);
|
||||
}
|
||||
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;
|
||||
free(node);
|
||||
}
|
||||
}
|
||||
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 = 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 int
|
||||
_connect_to(const char *ipaddress) {
|
||||
int fd = socket(AF_INET,SOCK_STREAM,0);
|
||||
struct sockaddr_in my_addr;
|
||||
char * port = strchr(ipaddress,':');
|
||||
if (port==NULL) {
|
||||
return -1;
|
||||
}
|
||||
int sz = port - ipaddress;
|
||||
char tmp[sz + 1];
|
||||
memcpy(tmp,ipaddress,sz);
|
||||
tmp[sz] = '\0';
|
||||
|
||||
my_addr.sin_addr.s_addr=inet_addr(tmp);
|
||||
my_addr.sin_family=AF_INET;
|
||||
my_addr.sin_port=htons(strtol(port+1,NULL,10));
|
||||
|
||||
int r = connect(fd,(struct sockaddr *)&my_addr,sizeof(struct sockaddr_in));
|
||||
|
||||
if (r == -1) {
|
||||
close(fd);
|
||||
return -1;
|
||||
}
|
||||
|
||||
return fd;
|
||||
}
|
||||
|
||||
static int
|
||||
_send_to(int fd, const void * buf, size_t sz, uint32_t handle) {
|
||||
char buffer[2 + sz + 12];
|
||||
uint16_t header = htons(sz+12);
|
||||
memcpy(buffer, &header, 2);
|
||||
memcpy(buffer+2, buf, sz);
|
||||
uint32_t u32 = 0;
|
||||
memcpy(buffer+2+sz,&u32,4);
|
||||
u32 = htonl(handle);
|
||||
memcpy(buffer+2+sz+4,&u32,4);
|
||||
u32 = 0;
|
||||
memcpy(buffer+2+sz+8,&u32,4);
|
||||
|
||||
sz += 2 + 12;
|
||||
|
||||
for (;;) {
|
||||
int err = send(fd, buffer, sz, 0);
|
||||
if (err < 0) {
|
||||
switch (errno) {
|
||||
case EAGAIN:
|
||||
case EINTR:
|
||||
continue;
|
||||
}
|
||||
}
|
||||
if (err != sz) {
|
||||
return 1;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
static void
|
||||
_broadcast(struct skynet_context * context, 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)
|
||||
continue;
|
||||
int err = _send_to(fd, name, sz, handle);
|
||||
if (err) {
|
||||
close(fd);
|
||||
fd = _connect_to(m->remote_addr[i]);
|
||||
if (fd < 0) {
|
||||
m->remote_fd[i] = -1;
|
||||
skynet_error(context, "Reconnect to harbor %d : %s faild", i, m->remote_addr[i]);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
static void
|
||||
_request_name(struct skynet_context * context, 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(context, m, name, GLOBALNAME_LENGTH, n->value);
|
||||
}
|
||||
|
||||
static void
|
||||
_update_name(struct skynet_context * context, 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(context, m,name,GLOBALNAME_LENGTH, handle);
|
||||
}
|
||||
|
||||
static void
|
||||
_update_address(struct skynet_context * context, struct master *m, int harbor_id, const char * buffer, size_t sz) {
|
||||
if (m->remote_fd[harbor_id] >= 0) {
|
||||
close(m->remote_fd[harbor_id]);
|
||||
m->remote_fd[harbor_id] = -1;
|
||||
}
|
||||
free(m->remote_addr[harbor_id]);
|
||||
char * addr = malloc(sz+1);
|
||||
memcpy(addr, buffer, sz);
|
||||
addr[sz] = '\0';
|
||||
m->remote_addr[harbor_id] = addr;
|
||||
int fd = _connect_to(addr);
|
||||
if (fd<0) {
|
||||
skynet_error(context, "Can't connect to harbor %d : %s", harbor_id, addr);
|
||||
return;
|
||||
}
|
||||
m->remote_fd[harbor_id] = fd;
|
||||
_broadcast(context, m, addr, sz, harbor_id);
|
||||
}
|
||||
|
||||
|
||||
/*
|
||||
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 session, const char * addr, const void * msg, size_t sz) {
|
||||
struct master *m = ud;
|
||||
assert(session == SESSION_CLIENT);
|
||||
uint32_t handle = 0;
|
||||
memcpy(&handle, msg, 4);
|
||||
handle = ntohl(handle);
|
||||
sz -= 4;
|
||||
const char * name = msg;
|
||||
name += 4;
|
||||
|
||||
if (handle == 0) {
|
||||
_request_name(context, m , name, sz);
|
||||
} else if (handle < REMOTE_MAX) {
|
||||
_update_address(context, m , handle, name, sz);
|
||||
} else {
|
||||
_update_name(context, 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 ! %s %d 0",args,REMOTE_MAX);
|
||||
const char * self_addr = skynet_command(ctx, "REG", NULL);
|
||||
const char * gate_addr = skynet_command(ctx, "LAUNCH", tmp);
|
||||
if (gate_addr == NULL) {
|
||||
skynet_error(ctx, "Master : launch gate failed");
|
||||
return 1;
|
||||
}
|
||||
int n = sprintf(tmp,"broker %s",self_addr);
|
||||
skynet_send(ctx, NULL, gate_addr, 0, tmp, n, 0);
|
||||
skynet_send(ctx, NULL, gate_addr, 0, "start", 5, 0);
|
||||
|
||||
skynet_callback(ctx, m, _mainloop);
|
||||
|
||||
return 0;
|
||||
}
|
||||
Reference in New Issue
Block a user