Partial rewrite, adding WebSockets, threads, pool.
parent
dda2236e4b
commit
5b7aa50e62
@ -0,0 +1,126 @@
|
|||||||
|
#include "pool.h"
|
||||||
|
#include "worker.h"
|
||||||
|
#include "conf.h"
|
||||||
|
#include "server.h"
|
||||||
|
|
||||||
|
#include <stdlib.h>
|
||||||
|
#include <event.h>
|
||||||
|
#include <hiredis/adapters/libevent.h>
|
||||||
|
|
||||||
|
struct pool *
|
||||||
|
pool_new(struct worker *w, int count) {
|
||||||
|
|
||||||
|
struct pool *p = calloc(1, sizeof(struct pool));
|
||||||
|
|
||||||
|
p->count = count;
|
||||||
|
p->ac = calloc(count, sizeof(redisAsyncContext*));
|
||||||
|
|
||||||
|
p->w = w;
|
||||||
|
p->cfg = w->s->cfg;
|
||||||
|
|
||||||
|
return p;
|
||||||
|
}
|
||||||
|
|
||||||
|
static void
|
||||||
|
pool_on_connect(const redisAsyncContext *c) {
|
||||||
|
struct pool *p = c->data;
|
||||||
|
int i = 0;
|
||||||
|
|
||||||
|
printf("Connected to redis\n");
|
||||||
|
if(!p) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* add to pool */
|
||||||
|
for(i = 0; i < p->count; ++i) {
|
||||||
|
if(p->ac[i] == NULL) {
|
||||||
|
p->ac[i] = c;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
static void
|
||||||
|
pool_on_disconnect(const redisAsyncContext *c, int status) {
|
||||||
|
|
||||||
|
struct pool *p = c->data;
|
||||||
|
int i = 0;
|
||||||
|
if (status != REDIS_OK) {
|
||||||
|
fprintf(stderr, "Error: %s\n", c->errstr);
|
||||||
|
}
|
||||||
|
|
||||||
|
if(p == NULL) { /* no need to clean anything here. */
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* remove from the pool */
|
||||||
|
for(i = 0; i < p->count; ++i) {
|
||||||
|
if(p->ac[i] == c) {
|
||||||
|
p->ac[i] = NULL;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/* reconnect */
|
||||||
|
pool_connect(p, 1);
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Create new connection.
|
||||||
|
*/
|
||||||
|
redisAsyncContext *
|
||||||
|
pool_connect(struct pool *p, int attach) {
|
||||||
|
|
||||||
|
struct redisAsyncContext *ac;
|
||||||
|
if(p->cfg->redis_host[0] == '/') { /* unix socket */
|
||||||
|
ac = redisAsyncConnectUnix(p->cfg->redis_host);
|
||||||
|
} else {
|
||||||
|
ac = redisAsyncConnect(p->cfg->redis_host, p->cfg->redis_port);
|
||||||
|
}
|
||||||
|
|
||||||
|
if(attach) {
|
||||||
|
ac->data = p;
|
||||||
|
} else {
|
||||||
|
ac->data = NULL;
|
||||||
|
}
|
||||||
|
|
||||||
|
if(ac->err) {
|
||||||
|
/*
|
||||||
|
const char err[] = "Connection failed";
|
||||||
|
slog(s, WEBDIS_ERROR, err, sizeof(err)-1);
|
||||||
|
*/
|
||||||
|
fprintf(stderr, "Error: %s\n", ac->errstr);
|
||||||
|
redisAsyncFree(ac);
|
||||||
|
return NULL;
|
||||||
|
}
|
||||||
|
|
||||||
|
redisLibeventAttach(ac, p->w->base);
|
||||||
|
redisAsyncSetConnectCallback(ac, pool_on_connect);
|
||||||
|
redisAsyncSetDisconnectCallback(ac, pool_on_disconnect);
|
||||||
|
|
||||||
|
if (p->cfg->redis_auth) { /* authenticate. */
|
||||||
|
redisAsyncCommand(ac, NULL, NULL, "AUTH %s", p->cfg->redis_auth);
|
||||||
|
}
|
||||||
|
if (p->cfg->database) { /* change database. */
|
||||||
|
redisAsyncCommand(ac, NULL, NULL, "SELECT %d", p->cfg->database);
|
||||||
|
}
|
||||||
|
return ac;
|
||||||
|
}
|
||||||
|
|
||||||
|
const redisAsyncContext *
|
||||||
|
pool_get_context(struct pool *p) {
|
||||||
|
|
||||||
|
int orig = p->cur++;
|
||||||
|
|
||||||
|
do {
|
||||||
|
p->cur++;
|
||||||
|
p->cur %= p->count;
|
||||||
|
if(p->ac[p->cur] != NULL) {
|
||||||
|
return p->ac[p->cur];
|
||||||
|
}
|
||||||
|
} while(p->cur != orig);
|
||||||
|
|
||||||
|
return NULL;
|
||||||
|
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,30 @@
|
|||||||
|
#ifndef POOL_H
|
||||||
|
#define POOL_H
|
||||||
|
|
||||||
|
#include <hiredis/async.h>
|
||||||
|
|
||||||
|
struct conf;
|
||||||
|
struct worker;
|
||||||
|
|
||||||
|
struct pool {
|
||||||
|
|
||||||
|
struct worker *w;
|
||||||
|
struct conf *cfg;
|
||||||
|
|
||||||
|
const redisAsyncContext **ac;
|
||||||
|
int count;
|
||||||
|
int cur;
|
||||||
|
|
||||||
|
};
|
||||||
|
|
||||||
|
|
||||||
|
struct pool *
|
||||||
|
pool_new(struct worker *w, int count);
|
||||||
|
|
||||||
|
redisAsyncContext *
|
||||||
|
pool_connect(struct pool *p, int attach);
|
||||||
|
|
||||||
|
const redisAsyncContext *
|
||||||
|
pool_get_context(struct pool *p);
|
||||||
|
|
||||||
|
#endif
|
@ -1,42 +1,30 @@
|
|||||||
#ifndef SERVER_H
|
#ifndef SERVER_H
|
||||||
#define SERVER_H
|
#define SERVER_H
|
||||||
|
|
||||||
#include <hiredis/async.h>
|
|
||||||
#include <time.h>
|
|
||||||
#include <sys/queue.h>
|
|
||||||
#include <event.h>
|
#include <event.h>
|
||||||
|
#include <hiredis/async.h>
|
||||||
|
|
||||||
struct server {
|
struct worker;
|
||||||
|
struct conf;
|
||||||
|
|
||||||
struct conf *cfg;
|
struct server {
|
||||||
struct event_base *base;
|
|
||||||
redisAsyncContext *ac;
|
|
||||||
|
|
||||||
/* server socket and event struct */
|
|
||||||
int fd;
|
int fd;
|
||||||
struct event ev;
|
struct event ev;
|
||||||
|
struct event_base *base;
|
||||||
|
|
||||||
struct event ev_reconnect;
|
struct conf *cfg;
|
||||||
struct timeval tv_reconnect;
|
|
||||||
};
|
|
||||||
|
|
||||||
void
|
|
||||||
webdis_connect(struct server *s);
|
|
||||||
|
|
||||||
struct server *
|
|
||||||
server_new(const char *filename);
|
|
||||||
|
|
||||||
void
|
/* worker threads */
|
||||||
server_free(struct server *s);
|
struct worker **w;
|
||||||
|
int next_worker;
|
||||||
|
};
|
||||||
|
|
||||||
struct server *
|
struct server *
|
||||||
server_copy(const struct server *s);
|
server_new(const char *cfg_file);
|
||||||
|
|
||||||
void
|
int
|
||||||
server_start(struct server *s);
|
server_start(struct server *s);
|
||||||
|
|
||||||
void
|
|
||||||
webdis_log(struct server *s, int level, const char *body);
|
|
||||||
|
|
||||||
#endif
|
#endif
|
||||||
|
|
||||||
|
@ -0,0 +1,15 @@
|
|||||||
|
OUT=websocket
|
||||||
|
CFLAGS=-O3 -Wall -Wextra
|
||||||
|
LDFLAGS=-levent -lpthread
|
||||||
|
|
||||||
|
all: $(OUT) Makefile
|
||||||
|
|
||||||
|
%: %.o Makefile
|
||||||
|
$(CC) $(LDFLAGS) -o $@ $<
|
||||||
|
|
||||||
|
%.o: %.c Makefile
|
||||||
|
$(CC) -c $(CFLAGS) -o $@ $<
|
||||||
|
|
||||||
|
clean:
|
||||||
|
rm -f *.o $(OUT)
|
||||||
|
|
@ -0,0 +1,181 @@
|
|||||||
|
#!/usr/bin/python
|
||||||
|
import urllib2, unittest, json
|
||||||
|
from functools import wraps
|
||||||
|
try:
|
||||||
|
import bson
|
||||||
|
except:
|
||||||
|
bson = None
|
||||||
|
|
||||||
|
|
||||||
|
host = '127.0.0.1'
|
||||||
|
port = 7379
|
||||||
|
|
||||||
|
class TestWebdis(unittest.TestCase):
|
||||||
|
|
||||||
|
def wrap(self,url):
|
||||||
|
return 'http://%s:%d/%s' % (host, port, url)
|
||||||
|
|
||||||
|
def query(self, url):
|
||||||
|
r = urllib2.Request(self.wrap(url))
|
||||||
|
return urllib2.urlopen(r)
|
||||||
|
|
||||||
|
class TestBasics(TestWebdis):
|
||||||
|
|
||||||
|
def test_crossdomain(self):
|
||||||
|
f = self.query('crossdomain.xml')
|
||||||
|
self.assertTrue(f.headers.getheader('Content-Type') == 'application/xml')
|
||||||
|
self.assertTrue("allow-access-from domain" in f.read())
|
||||||
|
|
||||||
|
def test_options(self):
|
||||||
|
pass
|
||||||
|
# not sure if OPTIONS is supported by urllib2...
|
||||||
|
# f = self.query('') # TODO: call with OPTIONS.
|
||||||
|
# self.assertTrue(f.headers.getheader('Content-Type') == 'text/html')
|
||||||
|
# self.assertTrue(f.headers.getheader('Allow') == 'GET,POST,PUT,OPTIONS')
|
||||||
|
# self.assertTrue(f.headers.getheader('Content-Length') == '0')
|
||||||
|
# self.assertTrue(f.headers.getheader('Access-Control-Allow-Origin') == '*')
|
||||||
|
|
||||||
|
|
||||||
|
class TestJSON(TestWebdis):
|
||||||
|
|
||||||
|
def test_set(self):
|
||||||
|
"success type (+OK)"
|
||||||
|
self.query('DEL/hello')
|
||||||
|
f = self.query('SET/hello/world')
|
||||||
|
self.assertTrue(f.headers.getheader('Content-Type') == 'application/json')
|
||||||
|
self.assertTrue(f.headers.getheader('ETag') == '"0db1124cf79ffeb80aff6d199d5822f8"')
|
||||||
|
self.assertTrue(f.read() == '{"SET":[true,"OK"]}')
|
||||||
|
|
||||||
|
def test_get(self):
|
||||||
|
"string type"
|
||||||
|
self.query('SET/hello/world')
|
||||||
|
f = self.query('GET/hello')
|
||||||
|
self.assertTrue(f.headers.getheader('Content-Type') == 'application/json')
|
||||||
|
self.assertTrue(f.headers.getheader('ETag') == '"8cf38afc245b7a6a88696566483d1390"')
|
||||||
|
self.assertTrue(f.read() == '{"GET":"world"}')
|
||||||
|
|
||||||
|
def test_incr(self):
|
||||||
|
"integer type"
|
||||||
|
self.query('DEL/hello')
|
||||||
|
f = self.query('INCR/hello')
|
||||||
|
self.assertTrue(f.headers.getheader('Content-Type') == 'application/json')
|
||||||
|
self.assertTrue(f.headers.getheader('ETag') == '"500e9bcdcbb1e98f25c1fbb880a96c99"')
|
||||||
|
self.assertTrue(f.read() == '{"INCR":1}')
|
||||||
|
|
||||||
|
def test_list(self):
|
||||||
|
"list type"
|
||||||
|
self.query('DEL/hello')
|
||||||
|
self.query('RPUSH/hello/abc')
|
||||||
|
self.query('RPUSH/hello/def')
|
||||||
|
f = self.query('LRANGE/hello/0/-1')
|
||||||
|
self.assertTrue(f.headers.getheader('Content-Type') == 'application/json')
|
||||||
|
self.assertTrue(f.headers.getheader('ETag') == '"622e51f547a480bef7cf5452fb7782db"')
|
||||||
|
self.assertTrue(f.read() == '{"LRANGE":["abc","def"]}')
|
||||||
|
|
||||||
|
def test_error(self):
|
||||||
|
"error return type"
|
||||||
|
f = self.query('UNKNOWN/COMMAND')
|
||||||
|
self.assertTrue(f.headers.getheader('Content-Type') == 'application/json')
|
||||||
|
try:
|
||||||
|
obj = json.loads(f.read())
|
||||||
|
except:
|
||||||
|
self.assertTrue(False)
|
||||||
|
return
|
||||||
|
|
||||||
|
self.assertTrue(len(obj) == 1)
|
||||||
|
self.assertTrue('UNKNOWN' in obj)
|
||||||
|
self.assertTrue(isinstance(obj['UNKNOWN'], list))
|
||||||
|
self.assertTrue(obj['UNKNOWN'][0] == False)
|
||||||
|
self.assertTrue(isinstance(obj['UNKNOWN'][1], unicode))
|
||||||
|
|
||||||
|
class TestRaw(TestWebdis):
|
||||||
|
|
||||||
|
def test_set(self):
|
||||||
|
"success type (+OK)"
|
||||||
|
self.query('DEL/hello')
|
||||||
|
f = self.query('SET/hello/world.raw')
|
||||||
|
self.assertTrue(f.headers.getheader('Content-Type') == 'binary/octet-stream')
|
||||||
|
self.assertTrue(f.read() == "+OK\r\n")
|
||||||
|
|
||||||
|
def test_get(self):
|
||||||
|
"string type"
|
||||||
|
self.query('SET/hello/world')
|
||||||
|
f = self.query('GET/hello.raw')
|
||||||
|
self.assertTrue(f.read() == '$5\r\nworld\r\n')
|
||||||
|
|
||||||
|
def test_incr(self):
|
||||||
|
"integer type"
|
||||||
|
self.query('DEL/hello')
|
||||||
|
f = self.query('INCR/hello.raw')
|
||||||
|
self.assertTrue(f.read() == ':1\r\n')
|
||||||
|
|
||||||
|
def test_list(self):
|
||||||
|
"list type"
|
||||||
|
self.query('DEL/hello')
|
||||||
|
self.query('RPUSH/hello/abc')
|
||||||
|
self.query('RPUSH/hello/def')
|
||||||
|
f = self.query('LRANGE/hello/0/-1.raw')
|
||||||
|
self.assertTrue(f.read() == "*2\r\n$3\r\nabc\r\n$3\r\ndef\r\n")
|
||||||
|
|
||||||
|
def test_error(self):
|
||||||
|
"error return type"
|
||||||
|
f = self.query('UNKNOWN/COMMAND.raw')
|
||||||
|
self.assertTrue(f.read().startswith("-ERR "))
|
||||||
|
|
||||||
|
def need_bson(fn):
|
||||||
|
def wrapper(self):
|
||||||
|
if bson:
|
||||||
|
fn(self)
|
||||||
|
return wrapper
|
||||||
|
|
||||||
|
class TestBSon(TestWebdis):
|
||||||
|
|
||||||
|
@need_bson
|
||||||
|
def test_set(self):
|
||||||
|
"success type (+OK)"
|
||||||
|
self.query('DEL/hello')
|
||||||
|
f = self.query('SET/hello/world.bson')
|
||||||
|
self.assertTrue(f.headers.getheader('Content-Type') == 'application/bson')
|
||||||
|
obj = bson.decode_all(f.read())
|
||||||
|
self.assertTrue(obj == [{u'SET': [True, bson.Binary('OK', 0)]}])
|
||||||
|
|
||||||
|
@need_bson
|
||||||
|
def test_get(self):
|
||||||
|
"string type"
|
||||||
|
self.query('SET/hello/world')
|
||||||
|
f = self.query('GET/hello.bson')
|
||||||
|
obj = bson.decode_all(f.read())
|
||||||
|
self.assertTrue(obj == [{u'GET': bson.Binary('world', 0)}])
|
||||||
|
|
||||||
|
@need_bson
|
||||||
|
def test_incr(self):
|
||||||
|
"integer type"
|
||||||
|
self.query('DEL/hello')
|
||||||
|
f = self.query('INCR/hello.bson')
|
||||||
|
obj = bson.decode_all(f.read())
|
||||||
|
self.assertTrue(obj == [{u'INCR': 1L}])
|
||||||
|
|
||||||
|
@need_bson
|
||||||
|
def test_list(self):
|
||||||
|
"list type"
|
||||||
|
self.query('DEL/hello')
|
||||||
|
self.query('RPUSH/hello/abc')
|
||||||
|
self.query('RPUSH/hello/def')
|
||||||
|
f = self.query('LRANGE/hello/0/-1.bson')
|
||||||
|
obj = bson.decode_all(f.read())
|
||||||
|
self.assertTrue(obj == [{u'LRANGE': [bson.Binary('abc', 0), bson.Binary('def', 0)]}])
|
||||||
|
|
||||||
|
@need_bson
|
||||||
|
def test_error(self):
|
||||||
|
"error return type"
|
||||||
|
f = self.query('UNKNOWN/COMMAND.bson')
|
||||||
|
obj = bson.decode_all(f.read())
|
||||||
|
self.assertTrue(len(obj) == 1)
|
||||||
|
self.assertTrue(u'UNKNOWN' in obj[0])
|
||||||
|
self.assertTrue(isinstance(obj[0], dict))
|
||||||
|
self.assertTrue(isinstance(obj[0][u'UNKNOWN'], list))
|
||||||
|
self.assertTrue(obj[0]['UNKNOWN'][0] == False)
|
||||||
|
self.assertTrue(isinstance(obj[0]['UNKNOWN'][1], bson.Binary))
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
unittest.main()
|
@ -0,0 +1,334 @@
|
|||||||
|
/* http://tools.ietf.org/html/draft-hixie-thewebsocketprotocol-76 */
|
||||||
|
|
||||||
|
#include <stdlib.h>
|
||||||
|
#define _GNU_SOURCE
|
||||||
|
#include <stdio.h>
|
||||||
|
#include <string.h>
|
||||||
|
#include <unistd.h>
|
||||||
|
#include <pthread.h>
|
||||||
|
#include <arpa/inet.h>
|
||||||
|
#include <errno.h>
|
||||||
|
|
||||||
|
|
||||||
|
#include <sys/types.h>
|
||||||
|
#include <sys/socket.h>
|
||||||
|
|
||||||
|
|
||||||
|
#include <event.h>
|
||||||
|
|
||||||
|
struct host_info {
|
||||||
|
char *host;
|
||||||
|
short port;
|
||||||
|
};
|
||||||
|
|
||||||
|
/* worker_thread, with counter of remaining messages */
|
||||||
|
struct worker_thread {
|
||||||
|
struct host_info *hi;
|
||||||
|
struct event_base *base;
|
||||||
|
|
||||||
|
int msg_target;
|
||||||
|
int msg_received;
|
||||||
|
int msg_sent;
|
||||||
|
int byte_count;
|
||||||
|
pthread_t thread;
|
||||||
|
|
||||||
|
struct evbuffer *buffer;
|
||||||
|
int got_header;
|
||||||
|
|
||||||
|
int verbose;
|
||||||
|
struct event ev_w;
|
||||||
|
};
|
||||||
|
|
||||||
|
void
|
||||||
|
process_message(struct worker_thread *wt, size_t sz) {
|
||||||
|
|
||||||
|
// printf("process_message\n");
|
||||||
|
if(wt->msg_received % 10000 == 0) {
|
||||||
|
printf("thread %u: %8d messages left (got %9d bytes so far).\n",
|
||||||
|
(unsigned int)wt->thread,
|
||||||
|
wt->msg_target - wt->msg_received, wt->byte_count);
|
||||||
|
}
|
||||||
|
wt->byte_count += sz;
|
||||||
|
|
||||||
|
/* decrement read count, and stop receiving when we reach zero. */
|
||||||
|
wt->msg_received++;
|
||||||
|
if(wt->msg_received == wt->msg_target) {
|
||||||
|
event_base_loopexit(wt->base, NULL);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
void
|
||||||
|
websocket_write(int fd, short event, void *ptr) {
|
||||||
|
int ret;
|
||||||
|
struct worker_thread *wt = ptr;
|
||||||
|
|
||||||
|
if(event != EV_WRITE) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
char message[] = "\x00[\"SET\",\"key\",\"value\"]\xff\x00[\"GET\",\"key\"]\xff";
|
||||||
|
ret = write(fd, message, sizeof(message)-1);
|
||||||
|
if(ret != sizeof(message)-1) {
|
||||||
|
fprintf(stderr, "write on %d failed: %s\n", fd, strerror(errno));
|
||||||
|
close(fd);
|
||||||
|
}
|
||||||
|
|
||||||
|
wt->msg_sent += 2;
|
||||||
|
if(wt->msg_sent < wt->msg_target) {
|
||||||
|
event_set(&wt->ev_w, fd, EV_WRITE, websocket_write, wt);
|
||||||
|
event_base_set(wt->base, &wt->ev_w);
|
||||||
|
ret = event_add(&wt->ev_w, NULL);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
static void
|
||||||
|
websocket_read(int fd, short event, void *ptr) {
|
||||||
|
char packet[2048], *pos;
|
||||||
|
int ret, success = 1;
|
||||||
|
|
||||||
|
struct worker_thread *wt = ptr;
|
||||||
|
|
||||||
|
if(event != EV_READ) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* read message */
|
||||||
|
ret = read(fd, packet, sizeof(packet));
|
||||||
|
pos = packet;
|
||||||
|
if(ret > 0) {
|
||||||
|
char *data, *last;
|
||||||
|
int sz, msg_sz;
|
||||||
|
|
||||||
|
/*
|
||||||
|
printf("Received %d bytes: \n", ret);
|
||||||
|
write(1, packet, ret);
|
||||||
|
printf("\n");
|
||||||
|
*/
|
||||||
|
|
||||||
|
if(wt->got_header == 0) { /* first response */
|
||||||
|
char *frame_start = strstr(packet, "MH"); /* end of the handshake */
|
||||||
|
if(frame_start == NULL) {
|
||||||
|
return; /* not yet */
|
||||||
|
} else { /* start monitoring possible writes */
|
||||||
|
printf("start monitoring possible writes\n");
|
||||||
|
evbuffer_add(wt->buffer, frame_start + 2, ret - (frame_start + 2 - packet));
|
||||||
|
|
||||||
|
wt->got_header = 1;
|
||||||
|
event_set(&wt->ev_w, fd, EV_WRITE,
|
||||||
|
websocket_write, wt);
|
||||||
|
event_base_set(wt->base, &wt->ev_w);
|
||||||
|
ret = event_add(&wt->ev_w, NULL);
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
/* we've had the header already, now bufffer data. */
|
||||||
|
evbuffer_add(wt->buffer, packet, ret);
|
||||||
|
}
|
||||||
|
|
||||||
|
while(1) {
|
||||||
|
data = (char*)EVBUFFER_DATA(wt->buffer);
|
||||||
|
sz = EVBUFFER_LENGTH(wt->buffer);
|
||||||
|
|
||||||
|
if(sz == 0) { /* no data */
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
if(*data != 0) { /* missing frame start */
|
||||||
|
success = 0;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
last = memchr(data, 0xff, sz); /* look for frame end */
|
||||||
|
if(!last) {
|
||||||
|
/* no end of frame in sight. */
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
msg_sz = last - data - 1;
|
||||||
|
process_message(ptr, msg_sz); /* record packet */
|
||||||
|
|
||||||
|
/* drain including frame delimiters (+2 bytes) */
|
||||||
|
evbuffer_drain(wt->buffer, msg_sz + 2);
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
printf("ret=%d\n", ret);
|
||||||
|
success = 0;
|
||||||
|
}
|
||||||
|
if(success == 0) {
|
||||||
|
shutdown(fd, SHUT_RDWR);
|
||||||
|
close(fd);
|
||||||
|
event_base_loopexit(wt->base, NULL);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
void*
|
||||||
|
worker_main(void *ptr) {
|
||||||
|
|
||||||
|
char ws_template[] = "GET /.json HTTP/1.1\r\n"
|
||||||
|
"Host: %s:%d\r\n"
|
||||||
|
"Connection: Upgrade\r\n"
|
||||||
|
"Upgrade: WebSocket\r\n"
|
||||||
|
"Origin: http://%s:%d\r\n"
|
||||||
|
"Sec-WebSocket-Key1: 18x 6]8vM;54 *(5: { U1]8 z [ 8\r\n"
|
||||||
|
"Sec-WebSocket-Key2: 1_ tx7X d < nw 334J702) 7]o}` 0\r\n"
|
||||||
|
"\r\n"
|
||||||
|
"Tm[K T2u";
|
||||||
|
|
||||||
|
struct worker_thread *wt = ptr;
|
||||||
|
|
||||||
|
int ret;
|
||||||
|
int fd;
|
||||||
|
struct sockaddr_in addr;
|
||||||
|
char *ws_handshake;
|
||||||
|
size_t ws_handshake_sz;
|
||||||
|
|
||||||
|
/* connect socket */
|
||||||
|
fd = socket(AF_INET, SOCK_STREAM, 0);
|
||||||
|
addr.sin_family = AF_INET;
|
||||||
|
addr.sin_port = htons(wt->hi->port);
|
||||||
|
memset(&(addr.sin_addr), 0, sizeof(addr.sin_addr));
|
||||||
|
addr.sin_addr.s_addr = inet_addr(wt->hi->host);
|
||||||
|
|
||||||
|
ret = connect(fd, (struct sockaddr*)&addr, sizeof(struct sockaddr));
|
||||||
|
if(ret != 0) {
|
||||||
|
fprintf(stderr, "connect: ret=%d: %s\n", ret, strerror(errno));
|
||||||
|
return NULL;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* initialize worker thread */
|
||||||
|
wt->base = event_base_new();
|
||||||
|
wt->buffer = evbuffer_new();
|
||||||
|
wt->byte_count = 0;
|
||||||
|
wt->got_header = 0;
|
||||||
|
|
||||||
|
/* send handshake */
|
||||||
|
ws_handshake_sz = sizeof(ws_handshake)
|
||||||
|
+ 2*strlen(wt->hi->host) + 500;
|
||||||
|
ws_handshake = calloc(ws_handshake_sz, 1);
|
||||||
|
ws_handshake_sz = (size_t)sprintf(ws_handshake, ws_template,
|
||||||
|
wt->hi->host, wt->hi->port,
|
||||||
|
wt->hi->host, wt->hi->port);
|
||||||
|
ret = write(fd, ws_handshake, ws_handshake_sz);
|
||||||
|
|
||||||
|
struct event ev_r;
|
||||||
|
event_set(&ev_r, fd, EV_READ | EV_PERSIST, websocket_read, wt);
|
||||||
|
event_base_set(wt->base, &ev_r);
|
||||||
|
event_add(&ev_r, NULL);
|
||||||
|
|
||||||
|
/* go! */
|
||||||
|
event_base_dispatch(wt->base);
|
||||||
|
event_base_free(wt->base);
|
||||||
|
free(ws_handshake);
|
||||||
|
return NULL;
|
||||||
|
}
|
||||||
|
|
||||||
|
void
|
||||||
|
usage(const char* argv0, char *host_default, short port_default,
|
||||||
|
int thread_count_default, int messages_default) {
|
||||||
|
|
||||||
|
printf("Usage: %s [options]\n"
|
||||||
|
"Options are:\n"
|
||||||
|
"\t-h host\t\t(default = \"%s\")\n"
|
||||||
|
"\t-p port\t\t(default = %d)\n"
|
||||||
|
"\t-c threads\t(default = %d)\n"
|
||||||
|
"\t-n count\t(number of messages per thread, default = %d)\n"
|
||||||
|
"\t-v\t\t(verbose)\n",
|
||||||
|
argv0, host_default, (int)port_default,
|
||||||
|
thread_count_default, messages_default);
|
||||||
|
}
|
||||||
|
|
||||||
|
int
|
||||||
|
main(int argc, char *argv[]) {
|
||||||
|
|
||||||
|
struct timespec t0, t1;
|
||||||
|
|
||||||
|
int messages_default = 100000;
|
||||||
|
int thread_count_default = 4;
|
||||||
|
short port_default = 7379;
|
||||||
|
char *host_default = "127.0.0.1";
|
||||||
|
|
||||||
|
int msg_target = messages_default;
|
||||||
|
int thread_count = thread_count_default;
|
||||||
|
int i, opt;
|
||||||
|
char *colon;
|
||||||
|
double total = 0, total_bytes = 0;
|
||||||
|
int verbose = 0;
|
||||||
|
|
||||||
|
struct host_info hi = {host_default, port_default};
|
||||||
|
|
||||||
|
struct worker_thread *workers;
|
||||||
|
|
||||||
|
/* getopt */
|
||||||
|
while ((opt = getopt(argc, argv, "h:p:c:n:v")) != -1) {
|
||||||
|
switch (opt) {
|
||||||
|
case 'h':
|
||||||
|
colon = strchr(optarg, ':');
|
||||||
|
if(!colon) {
|
||||||
|
size_t sz = strlen(optarg);
|
||||||
|
hi.host = calloc(1 + sz, 1);
|
||||||
|
strncpy(hi.host, optarg, sz);
|
||||||
|
} else {
|
||||||
|
hi.host = calloc(1+colon-optarg, 1);
|
||||||
|
strncpy(hi.host, optarg, colon-optarg);
|
||||||
|
hi.port = (short)atol(colon+1);
|
||||||
|
}
|
||||||
|
break;
|
||||||
|
|
||||||
|
case 'p':
|
||||||
|
hi.port = (short)atol(optarg);
|
||||||
|
break;
|
||||||
|
|
||||||
|
case 'c':
|
||||||
|
thread_count = atoi(optarg);
|
||||||
|
break;
|
||||||
|
|
||||||
|
case 'n':
|
||||||
|
msg_target = atoi(optarg);
|
||||||
|
break;
|
||||||
|
|
||||||
|
case 'v':
|
||||||
|
verbose = 1;
|
||||||
|
break;
|
||||||
|
default: /* '?' */
|
||||||
|
usage(argv[0], host_default, port_default,
|
||||||
|
thread_count_default,
|
||||||
|
messages_default);
|
||||||
|
exit(EXIT_FAILURE);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/* run threads */
|
||||||
|
workers = calloc(sizeof(struct worker_thread), thread_count);
|
||||||
|
|
||||||
|
clock_gettime(CLOCK_MONOTONIC, &t0);
|
||||||
|
for(i = 0; i < thread_count; ++i) {
|
||||||
|
workers[i].msg_target = msg_target;
|
||||||
|
workers[i].hi = &hi;
|
||||||
|
workers[i].verbose = verbose;
|
||||||
|
|
||||||
|
pthread_create(&workers[i].thread, NULL,
|
||||||
|
worker_main, &workers[i]);
|
||||||
|
}
|
||||||
|
|
||||||
|
/* wait for threads to finish */
|
||||||
|
for(i = 0; i < thread_count; ++i) {
|
||||||
|
pthread_join(workers[i].thread, NULL);
|
||||||
|
total += workers[i].msg_received;
|
||||||
|
total_bytes += workers[i].byte_count;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* timing */
|
||||||
|
clock_gettime(CLOCK_MONOTONIC, &t1);
|
||||||
|
float mili0 = t0.tv_sec * 1000 + t0.tv_nsec / 1000000;
|
||||||
|
float mili1 = t1.tv_sec * 1000 + t1.tv_nsec / 1000000;
|
||||||
|
|
||||||
|
if(total != 0) {
|
||||||
|
printf("Read %ld messages in %0.2f sec: %0.2f msg/sec (%d MB/sec, %d KB/sec)\n",
|
||||||
|
(long)total,
|
||||||
|
(mili1-mili0)/1000.0,
|
||||||
|
1000*total/(mili1-mili0),
|
||||||
|
(int)(total_bytes / (1000*(mili1-mili0))),
|
||||||
|
(int)(total_bytes / (mili1-mili0)));
|
||||||
|
} else {
|
||||||
|
printf("No message was read.\n");
|
||||||
|
}
|
||||||
|
|
||||||
|
return EXIT_SUCCESS;
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,234 @@
|
|||||||
|
#include "md5/md5.h"
|
||||||
|
#include "websocket.h"
|
||||||
|
#include "client.h"
|
||||||
|
#include "formats/json.h"
|
||||||
|
#include "cmd.h"
|
||||||
|
#include "worker.h"
|
||||||
|
#include "pool.h"
|
||||||
|
|
||||||
|
#include <stdlib.h>
|
||||||
|
#include <stdio.h>
|
||||||
|
#include <string.h>
|
||||||
|
#include <unistd.h>
|
||||||
|
#include <errno.h>
|
||||||
|
|
||||||
|
static uint32_t
|
||||||
|
ws_read_key(const char *s) {
|
||||||
|
|
||||||
|
uint32_t ret = 0, spaces = 0;
|
||||||
|
const char *p;
|
||||||
|
size_t sz;
|
||||||
|
|
||||||
|
if(!s) {
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
sz = strlen(s);
|
||||||
|
|
||||||
|
for(p = s; p < s+sz; ++p) {
|
||||||
|
if(*p >= '0' && *p <= '9') {
|
||||||
|
ret *= 10;
|
||||||
|
ret += (*p) - '0';
|
||||||
|
} else if (*p == ' ') {
|
||||||
|
spaces++;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return htonl(ret / spaces);
|
||||||
|
}
|
||||||
|
|
||||||
|
static int
|
||||||
|
ws_compute_handshake(struct http_client *c, unsigned char *out) {
|
||||||
|
|
||||||
|
char buffer[16];
|
||||||
|
md5_state_t ctx;
|
||||||
|
|
||||||
|
// websocket handshake
|
||||||
|
uint32_t number_1 = ws_read_key(client_get_header(c, "Sec-WebSocket-Key1"));
|
||||||
|
uint32_t number_2 = ws_read_key(client_get_header(c, "Sec-WebSocket-Key2"));
|
||||||
|
|
||||||
|
if(c->body_sz < 8) { /* we need at least 8 bytes */
|
||||||
|
return -1;
|
||||||
|
}
|
||||||
|
|
||||||
|
memcpy(buffer, &number_1, sizeof(uint32_t));
|
||||||
|
memcpy(buffer + sizeof(uint32_t), &number_2, sizeof(uint32_t));
|
||||||
|
memcpy(buffer + 2 * sizeof(uint32_t), c->body + c->body_sz - 8, 8); /* last 8 bytes */
|
||||||
|
|
||||||
|
md5_init(&ctx);
|
||||||
|
md5_append(&ctx, (const md5_byte_t *)buffer, sizeof(buffer));
|
||||||
|
md5_finish(&ctx, out);
|
||||||
|
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
int
|
||||||
|
ws_handshake_reply(struct http_client *c) {
|
||||||
|
|
||||||
|
int ret;
|
||||||
|
unsigned char md5_handshake[16];
|
||||||
|
char *buffer = NULL, *p;
|
||||||
|
const char *origin = NULL, *host = NULL;
|
||||||
|
size_t origin_sz = 0, host_sz = 0, sz;
|
||||||
|
|
||||||
|
char template0[] = "HTTP/1.1 101 Websocket Protocol Handshake\r\n"
|
||||||
|
"Upgrade: WebSocket\r\n"
|
||||||
|
"Connection: Upgrade\r\n"
|
||||||
|
"Sec-WebSocket-Origin: "; /* %s */
|
||||||
|
char template1[] = "\r\n"
|
||||||
|
"Sec-WebSocket-Location: ws://"; /* %s%s */
|
||||||
|
char template2[] = "\r\n"
|
||||||
|
"Origin: http://"; /* %s */
|
||||||
|
char template3[] = "\r\n\r\n";
|
||||||
|
|
||||||
|
if((origin = client_get_header(c, "Origin"))) {
|
||||||
|
origin_sz = strlen(origin);
|
||||||
|
}
|
||||||
|
if((host = client_get_header(c, "Host"))) {
|
||||||
|
host_sz = strlen(host);
|
||||||
|
}
|
||||||
|
|
||||||
|
/* need those headers */
|
||||||
|
if(!origin || !origin_sz || !host || !host_sz || !c->path || !c->path_sz) {
|
||||||
|
return -1;
|
||||||
|
}
|
||||||
|
|
||||||
|
if(ws_compute_handshake(c, &md5_handshake[0]) != 0) {
|
||||||
|
printf("failed to compute handshake.\n");
|
||||||
|
return -1;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* This code uses the WebSocket specification from May 23, 2010.
|
||||||
|
* The latest copy is available at http://www.whatwg.org/specs/web-socket-protocol/
|
||||||
|
*/
|
||||||
|
|
||||||
|
sz = sizeof(template0)-1 + origin_sz
|
||||||
|
+ sizeof(template1)-1 + host_sz + c->path_sz
|
||||||
|
+ sizeof(template2)-1 + host_sz
|
||||||
|
+ sizeof(template3)-1 + sizeof(md5_handshake);
|
||||||
|
|
||||||
|
p = buffer = malloc(sz);
|
||||||
|
|
||||||
|
/* Concat all */
|
||||||
|
|
||||||
|
/* template0 */
|
||||||
|
memcpy(p, template0, sizeof(template0)-1);
|
||||||
|
p += sizeof(template0)-1;
|
||||||
|
memcpy(p, origin, origin_sz);
|
||||||
|
p += origin_sz;
|
||||||
|
|
||||||
|
/* template1 */
|
||||||
|
memcpy(p, template1, sizeof(template1)-1);
|
||||||
|
p += sizeof(template1)-1;
|
||||||
|
memcpy(p, host, host_sz);
|
||||||
|
p += host_sz;
|
||||||
|
memcpy(p, c->path, c->path_sz);
|
||||||
|
p += c->path_sz;
|
||||||
|
|
||||||
|
/* template2 */
|
||||||
|
memcpy(p, template2, sizeof(template2)-1);
|
||||||
|
p += sizeof(template2)-1;
|
||||||
|
memcpy(p, host, host_sz);
|
||||||
|
p += host_sz;
|
||||||
|
|
||||||
|
/* template3 */
|
||||||
|
memcpy(p, template3, sizeof(template3)-1);
|
||||||
|
p += sizeof(template3)-1;
|
||||||
|
memcpy(p, &md5_handshake[0], sizeof(md5_handshake));
|
||||||
|
|
||||||
|
ret = write(c->fd, buffer, sz);
|
||||||
|
free(buffer);
|
||||||
|
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
static int
|
||||||
|
ws_execute(struct http_client *c, const char *frame, size_t frame_len) {
|
||||||
|
|
||||||
|
struct cmd*(*fun_extract)(struct http_client *, const char *, size_t) = NULL;
|
||||||
|
|
||||||
|
if(strncmp(c->path, "/.json", 6) == 0) {
|
||||||
|
fun_extract = json_ws_extract;
|
||||||
|
}
|
||||||
|
|
||||||
|
if(fun_extract) {
|
||||||
|
struct cmd *cmd = fun_extract(c, frame, frame_len);
|
||||||
|
if(cmd) {
|
||||||
|
cmd->is_websocket = 1;
|
||||||
|
cmd->fd = c->fd;
|
||||||
|
|
||||||
|
/* TODO: clean this mess */
|
||||||
|
redisAsyncContext *ac = (redisAsyncContext*)pool_get_context(c->w->pool);
|
||||||
|
cmd_send(ac, json_reply, cmd);
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return -1;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Process some data just received on the socket.
|
||||||
|
*/
|
||||||
|
enum ws_read_action
|
||||||
|
ws_add_data(struct http_client *c) {
|
||||||
|
const char *frame_start, *frame_end;
|
||||||
|
char *tmp;
|
||||||
|
|
||||||
|
while(1) {
|
||||||
|
/* look for frame start */
|
||||||
|
if(!c->sz || c->buffer[0] != '\x00') {
|
||||||
|
/* printf("frame start fail\n"); */
|
||||||
|
return WS_READ_FAIL;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* look for frame end */
|
||||||
|
int ret;
|
||||||
|
size_t frame_len;
|
||||||
|
frame_start = c->buffer;
|
||||||
|
frame_end = memchr(frame_start, '\xff', c->sz);
|
||||||
|
if(frame_end == NULL) {
|
||||||
|
/* continue reading */
|
||||||
|
return WS_READ_MORE;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* parse and execute frame. */
|
||||||
|
frame_len = frame_end - frame_start - 1;
|
||||||
|
|
||||||
|
ret = ws_execute(c, frame_start + 1, frame_len);
|
||||||
|
if(ret != 0) {
|
||||||
|
return WS_READ_FAIL;
|
||||||
|
}
|
||||||
|
|
||||||
|
/* remove frame from buffer */
|
||||||
|
c->sz -= (2 + frame_len);
|
||||||
|
tmp = malloc(c->sz);
|
||||||
|
memcpy(tmp, c->buffer + 2 + frame_len, c->sz);
|
||||||
|
free(c->buffer);
|
||||||
|
c->buffer = tmp;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
int
|
||||||
|
ws_reply(struct cmd *cmd, const char *p, size_t sz) {
|
||||||
|
|
||||||
|
int ret;
|
||||||
|
char *buffer = malloc(sz + 2);
|
||||||
|
|
||||||
|
/* create frame */
|
||||||
|
buffer[0] = '\x00';
|
||||||
|
memcpy(buffer + 1, p, sz);
|
||||||
|
buffer[sz + 1] = '\xff';
|
||||||
|
|
||||||
|
/* send WS frame */
|
||||||
|
ret = write(cmd->fd, buffer, sz+2);
|
||||||
|
free(buffer);
|
||||||
|
|
||||||
|
if(ret == (int)sz + 2) {
|
||||||
|
/* http_client_serve(c); */
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
/* printf("WRITE FAIL on fd=%d, ret=%d (%s)\n", c->fd, ret, strerror(errno)); */
|
||||||
|
|
||||||
|
return -1;
|
||||||
|
}
|
@ -0,0 +1,23 @@
|
|||||||
|
#ifndef WEBSOCKET_H
|
||||||
|
#define WEBSOCKET_H
|
||||||
|
|
||||||
|
#include <stdlib.h>
|
||||||
|
|
||||||
|
struct http_client;
|
||||||
|
struct cmd;
|
||||||
|
|
||||||
|
enum ws_read_action {
|
||||||
|
WS_READ_FAIL,
|
||||||
|
WS_READ_MORE,
|
||||||
|
WS_READ_EXEC};
|
||||||
|
|
||||||
|
int
|
||||||
|
ws_handshake_reply(struct http_client *c);
|
||||||
|
|
||||||
|
enum ws_read_action
|
||||||
|
ws_add_data(struct http_client *c);
|
||||||
|
|
||||||
|
int
|
||||||
|
ws_reply(struct cmd *cmd, const char *p, size_t sz);
|
||||||
|
|
||||||
|
#endif
|
@ -0,0 +1,196 @@
|
|||||||
|
#include "worker.h"
|
||||||
|
#include "client.h"
|
||||||
|
#include "http.h"
|
||||||
|
#include "cmd.h"
|
||||||
|
#include "pool.h"
|
||||||
|
#include "slog.h"
|
||||||
|
#include "websocket.h"
|
||||||
|
|
||||||
|
#include <stdlib.h>
|
||||||
|
#include <stdio.h>
|
||||||
|
#include <unistd.h>
|
||||||
|
#include <event.h>
|
||||||
|
#include <string.h>
|
||||||
|
|
||||||
|
|
||||||
|
struct worker *
|
||||||
|
worker_new(struct server *s) {
|
||||||
|
|
||||||
|
struct worker *w = calloc(1, sizeof(struct worker));
|
||||||
|
w->s = s;
|
||||||
|
|
||||||
|
/* setup communication link */
|
||||||
|
pipe(w->link);
|
||||||
|
|
||||||
|
/* Redis connection pool */
|
||||||
|
w->pool = pool_new(w, 8); /* FIXME: change the number? use conf? */
|
||||||
|
|
||||||
|
return w;
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
void
|
||||||
|
worker_can_read(int fd, short event, void *p) {
|
||||||
|
|
||||||
|
struct http_client *c = p;
|
||||||
|
int ret, nparsed;
|
||||||
|
|
||||||
|
(void)fd;
|
||||||
|
(void)event;
|
||||||
|
|
||||||
|
ret = http_client_read(c);
|
||||||
|
if(ret <= 0) {
|
||||||
|
printf("client disconnected\n");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if(c->is_websocket) {
|
||||||
|
/* printf("Got websocket data! (%d bytes)\n", ret); */
|
||||||
|
ws_add_data(c);
|
||||||
|
} else {
|
||||||
|
/* run parser */
|
||||||
|
nparsed = http_client_execute(c);
|
||||||
|
|
||||||
|
if(c->is_websocket) {
|
||||||
|
/* we need to use the remaining (unparsed) data as the body. */
|
||||||
|
if(nparsed < ret) {
|
||||||
|
http_client_set_body(c, c->buffer + nparsed + 1, c->sz - nparsed - 1);
|
||||||
|
ws_handshake_reply(c);
|
||||||
|
} else {
|
||||||
|
c->broken = 1;
|
||||||
|
}
|
||||||
|
free(c->buffer);
|
||||||
|
c->buffer = NULL;
|
||||||
|
c->sz = 0;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if(c->broken) { /* terminate client */
|
||||||
|
/* printf("terminate client\n"); */
|
||||||
|
http_client_free(c);
|
||||||
|
} else {
|
||||||
|
/* printf("start monitoring input again.\n"); */
|
||||||
|
worker_monitor_input(c);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
void
|
||||||
|
worker_monitor_input(struct http_client *c) {
|
||||||
|
|
||||||
|
event_set(&c->ev, c->fd, EV_READ, worker_can_read, c);
|
||||||
|
event_base_set(c->w->base, &c->ev);
|
||||||
|
event_add(&c->ev, NULL);
|
||||||
|
}
|
||||||
|
|
||||||
|
static void
|
||||||
|
worker_can_accept(int pipefd, short event, void *ptr) {
|
||||||
|
|
||||||
|
struct http_client *c;
|
||||||
|
unsigned long addr;
|
||||||
|
|
||||||
|
(void)event;
|
||||||
|
(void)ptr;
|
||||||
|
|
||||||
|
int ret = read(pipefd, &addr, sizeof(addr));
|
||||||
|
if(ret == sizeof(addr)) {
|
||||||
|
c = (struct http_client*)addr;
|
||||||
|
/* create client, monitor fd for input */
|
||||||
|
worker_monitor_input(c);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
static void
|
||||||
|
worker_pool_connect(struct worker *w) {
|
||||||
|
|
||||||
|
int i;
|
||||||
|
/* create connections */
|
||||||
|
for(i = 0; i < w->pool->count; ++i) {
|
||||||
|
pool_connect(w->pool, 1);
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
static void*
|
||||||
|
worker_main(void *p) {
|
||||||
|
|
||||||
|
struct worker *w = p;
|
||||||
|
struct event ev;
|
||||||
|
|
||||||
|
/* setup libevent */
|
||||||
|
w->base = event_base_new();
|
||||||
|
|
||||||
|
/* monitor pipe link */
|
||||||
|
event_set(&ev, w->link[0], EV_READ | EV_PERSIST, worker_can_accept, w);
|
||||||
|
event_base_set(w->base, &ev);
|
||||||
|
event_add(&ev, NULL);
|
||||||
|
|
||||||
|
/* connect to Redis */
|
||||||
|
worker_pool_connect(w);
|
||||||
|
|
||||||
|
/* loop */
|
||||||
|
event_base_dispatch(w->base);
|
||||||
|
|
||||||
|
return NULL;
|
||||||
|
}
|
||||||
|
|
||||||
|
void
|
||||||
|
worker_start(struct worker *w) {
|
||||||
|
|
||||||
|
pthread_create(&w->thread, NULL, worker_main, w);
|
||||||
|
}
|
||||||
|
|
||||||
|
/* queue new client to process */
|
||||||
|
void
|
||||||
|
worker_add_client(struct worker *w, struct http_client *c) {
|
||||||
|
|
||||||
|
/* write into pipe link */
|
||||||
|
unsigned long addr = (unsigned long)c;
|
||||||
|
int ret = write(w->link[1], &addr, sizeof(addr));
|
||||||
|
(void)ret;
|
||||||
|
/* printf("[for worker %p] write: %lu, c=%p (ret=%d)\n", (void*)w, addr, (void*)c, ret); */
|
||||||
|
}
|
||||||
|
|
||||||
|
/* Called when a client has finished reading input and is ready to be executed. */
|
||||||
|
void
|
||||||
|
worker_process_client(struct http_client *c) {
|
||||||
|
|
||||||
|
/* printf("worker_process_client\n"); */
|
||||||
|
/* check that the command can be executed */
|
||||||
|
struct worker *w = c->w;
|
||||||
|
int ret = -1;
|
||||||
|
switch(c->parser.method) {
|
||||||
|
case HTTP_GET:
|
||||||
|
if(c->path_sz == 16 && memcmp(c->path, "/crossdomain.xml", 16) == 0) {
|
||||||
|
http_crossdomain(c);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
slog(w->s, WEBDIS_DEBUG, c->path, c->path_sz);
|
||||||
|
ret = cmd_run(c->w, c, 1+c->path, c->path_sz-1, NULL, 0);
|
||||||
|
break;
|
||||||
|
|
||||||
|
case HTTP_POST:
|
||||||
|
slog(w->s, WEBDIS_DEBUG, c->path, c->path_sz);
|
||||||
|
ret = cmd_run(c->w, c, c->body, c->body_sz, NULL, 0);
|
||||||
|
break;
|
||||||
|
|
||||||
|
case HTTP_PUT:
|
||||||
|
slog(w->s, WEBDIS_DEBUG, c->path, c->path_sz);
|
||||||
|
ret = cmd_run(c->w, c, 1+c->path, c->path_sz-1,
|
||||||
|
c->body, c->body_sz);
|
||||||
|
break;
|
||||||
|
|
||||||
|
case HTTP_OPTIONS:
|
||||||
|
http_send_options(c);
|
||||||
|
|
||||||
|
default:
|
||||||
|
slog(w->s, WEBDIS_DEBUG, "405", 3);
|
||||||
|
http_send_error(c, 405, "Method Not Allowed");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if(ret < 0) {
|
||||||
|
http_send_error(c, 403, "Forbidden");
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,41 @@
|
|||||||
|
#ifndef WORKER_H
|
||||||
|
#define WORKER_H
|
||||||
|
|
||||||
|
#include <pthread.h>
|
||||||
|
|
||||||
|
struct http_client;
|
||||||
|
struct pool;
|
||||||
|
|
||||||
|
struct worker {
|
||||||
|
|
||||||
|
/* self */
|
||||||
|
pthread_t thread;
|
||||||
|
struct event_base *base;
|
||||||
|
|
||||||
|
/* connection dispatcher */
|
||||||
|
struct server *s;
|
||||||
|
int link[2];
|
||||||
|
|
||||||
|
/* Redis connection pool */
|
||||||
|
struct pool *pool;
|
||||||
|
};
|
||||||
|
|
||||||
|
struct worker *
|
||||||
|
worker_new(struct server *s);
|
||||||
|
|
||||||
|
void
|
||||||
|
worker_start(struct worker *w);
|
||||||
|
|
||||||
|
void
|
||||||
|
worker_add_client(struct worker *w, struct http_client *c);
|
||||||
|
|
||||||
|
void
|
||||||
|
worker_monitor_input(struct http_client *c);
|
||||||
|
|
||||||
|
void
|
||||||
|
worker_can_read(int fd, short event, void *p);
|
||||||
|
|
||||||
|
void
|
||||||
|
worker_process_client(struct http_client *c);
|
||||||
|
|
||||||
|
#endif
|
Loading…
Reference in New Issue