blackbeard420/kore
1
0
Fork 0
mirror of https://github.com/jorisvink/kore synced 2025-03-09 04:29:02 -04:00
Code Issues Projects Releases Wiki Activity

Add BSD kqueue(2) support. Compile with make bsd (or make linux for linux)

Browse Source
This commit is contained in:
Joris Vink 2013-06-17 23:39:17 +02:00
parent 98de763632
commit 25f1ab9865
9 changed files with 696 additions and 396 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
Makefile
View File

@ -3,20 +3,29 @@
CC=gcc
BIN=kore
S_SRC= src/kore.c src/buf.c src/config.c src/net.c src/spdy.c src/http.c \
S_SRC+= src/kore.c src/buf.c src/config.c src/net.c src/spdy.c src/http.c \
src/module.c src/utils.c src/zlib_dict.c
S_OBJS= $(S_SRC:.c=.o)
CFLAGS+=-I/usr/local/ssl/include
CFLAGS+=-Wall -Wstrict-prototypes -Wmissing-prototypes
CFLAGS+=-Wmissing-declarations -Wshadow -Wpointer-arith -Wcast-qual
CFLAGS+=-D_GNU_SOURCE=1 -Wsign-compare -Iincludes -g
LDFLAGS=-rdynamic -Llibs -lssl -lcrypto -ldl -lz
CFLAGS+=-Wsign-compare -Iincludes -g
LDFLAGS+=-rdynamic -Llibs -lssl -lcrypto -lz
default:
@echo "Please specify a build target [linux | bsd]"
linux:
@LDFLAGS=-ldl CFLAGS=-D_GNU_SOURCE=1 S_SRC=src/linux.c make kore
bsd:
@S_SRC=src/bsd.c make kore
kore: $(S_OBJS)
$(CC) $(CFLAGS) $(S_OBJS) $(LDFLAGS) -o $(BIN)
.c.o: $<
.c.o:
$(CC) $(CFLAGS) -c $< -o $@
clean:

2
example.conf
View File

@ -3,7 +3,7 @@
# Server configuration.
bind 10.211.55.3 443
certfile cert/server.crt
certkey certe/server.key
certkey cert/server.key
# The path worker processes will chroot too after starting.
chroot /home/joris/src/kore

2
example/Makefile
View File

@ -37,7 +37,7 @@ html_inject: tools/html_inject.c
.css.c:
tools/html_inject $< `basename $<` > $@
.c.o: $<
.c.o:
$(CC) -fPIC $(CFLAGS) -c $< -o $@
clean:

1
example/src/example.c
View File

@ -18,7 +18,6 @@
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/queue.h>
#include <sys/epoll.h>
#include <netinet/in.h>
#include <arpa/inet.h>

24
includes/kore.h
View File

@ -111,6 +111,8 @@ struct kore_worker {
TAILQ_ENTRY(kore_worker) list;
};
TAILQ_HEAD(kore_worker_h, kore_worker);
#define KORE_BUF_INITIAL 128
#define KORE_BUF_INCREMENT KORE_BUF_INITIAL
@ -125,6 +127,7 @@ struct buf_vec {
u_int32_t length;
};
extern pid_t mypid;
extern int kore_debug;
extern int server_port;
extern char *server_ip;
@ -134,8 +137,25 @@ extern char *kore_module_onload;
extern char *kore_pidfile;
extern char *kore_certfile;
extern char *kore_certkey;
extern u_int8_t worker_count;
extern pid_t mypid;
extern u_int16_t cpu_count;
extern u_int8_t worker_count;
extern struct listener server;
extern struct kore_worker_h kore_workers;
void kore_init(void);
void kore_worker_init(void);
void kore_worker_wait(int);
void kore_event_init(void);
void kore_event_wait(int);
void kore_set_proctitle(char *);
void kore_worker_spawn(u_int16_t);
void kore_worker_entry(struct kore_worker *);
void kore_worker_setcpu(struct kore_worker *);
void kore_event_schedule(int, int, int, void *);
int kore_connection_handle(struct connection *);
int kore_server_accept(struct listener *, struct connection **);
void kore_log_init(void);
void *kore_malloc(size_t);

206
src/bsd.c Normal file
View File

@ -0,0 +1,206 @@
/*
* Copyright (c) 2013 Joris Vink <joris@coders.se>
*
* Permission to use, copy, modify, and distribute this software for any
* purpose with or without fee is hereby granted, provided that the above
* copyright notice and this permission notice appear in all copies.
*
* THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
* WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
* MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
* ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
* WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
* ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
* OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
#include <sys/param.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/queue.h>
#include <sys/event.h>
#include <sys/wait.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <openssl/err.h>
#include <openssl/ssl.h>
#include <errno.h>
#include <fcntl.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <syslog.h>
#include <unistd.h>
#include <time.h>
#include <regex.h>
#include <zlib.h>
#include <unistd.h>
#include "spdy.h"
#include "kore.h"
#include "http.h"
#define KQUEUE_EVENTS 500
static int kfd = -1;
static struct kevent *events;
static int nchanges;
static struct kevent *changelist;
void
kore_init(void)
{
cpu_count = 0;
}
void
kore_worker_init(void)
{
u_int16_t i;
if (worker_count == 0)
fatal("no workers specified");
kore_debug("kore_worker_init(): starting %d workers", worker_count);
TAILQ_INIT(&kore_workers);
for (i = 0; i < worker_count; i++)
kore_worker_spawn(0);
}
void
kore_worker_wait(int final)
{
pid_t pid;
int status;
struct kore_worker k, *kw, *next;
if (final)
pid = waitpid(WAIT_ANY, &status, 0);
else
pid = waitpid(WAIT_ANY, &status, WNOHANG);
if (pid == -1) {
kore_debug("waitpid(): %s", errno_s);
return;
}
if (pid == 0)
return;
for (kw = TAILQ_FIRST(&kore_workers); kw != NULL; kw = next) {
next = TAILQ_NEXT(kw, list);
if (kw->pid != pid)
continue;
k = *kw;
TAILQ_REMOVE(&kore_workers, kw, list);
kore_log(LOG_NOTICE, "worker %d (%d)-> status %d",
kw->id, pid, status);
free(kw);
if (final)
continue;
if (WEXITSTATUS(status) || WTERMSIG(status) ||
WCOREDUMP(status)) {
kore_log(LOG_NOTICE,
"worker %d (pid: %d) gone, respawning new one",
k.id, k.pid);
kore_worker_spawn(0);
}
}
}
void
kore_worker_setcpu(struct kore_worker *kw)
{
}
void
kore_event_init(void)
{
if ((kfd = kqueue()) == -1)
fatal("kqueue(): %s", errno_s);
nchanges = 0;
events = kore_calloc(KQUEUE_EVENTS, sizeof(struct kevent));
changelist = kore_calloc(KQUEUE_EVENTS, sizeof(struct kevent));
kore_event_schedule(server.fd, EVFILT_READ, EV_ADD, &server);
}
void
kore_event_wait(int quit)
{
struct connection *c;
int n, i, *fd;
n = kevent(kfd, changelist, nchanges, events, KQUEUE_EVENTS, NULL);
if (n == -1) {
if (errno == EINTR)
return;
fatal("kevent(): %s", errno_s);
}
nchanges = 0;
if (n > 0)
kore_debug("main(): %d sockets available", n);
for (i = 0; i < n; i++) {
fd = (int *)events[i].udata;
if (events[i].flags & EV_EOF ||
events[i].flags & EV_ERROR) {
if (*fd == server.fd)
fatal("error on server socket");
c = (struct connection *)events[i].udata;
kore_server_disconnect(c);
continue;
}
if (*fd == server.fd) {
if (!quit) {
kore_server_accept(&server, &c);
if (c == NULL)
continue;
kore_event_schedule(c->fd, EVFILT_READ,
EV_ADD, c);
kore_event_schedule(c->fd, EVFILT_WRITE,
EV_ADD | EV_ONESHOT, c);
}
} else {
c = (struct connection *)events[i].udata;
if (events[i].filter == EVFILT_READ)
c->flags |= CONN_READ_POSSIBLE;
if (events[i].filter == EVFILT_WRITE)
c->flags |= CONN_WRITE_POSSIBLE;
if (!kore_connection_handle(c)) {
kore_server_disconnect(c);
} else {
if (!TAILQ_EMPTY(&(c->send_queue))) {
kore_event_schedule(c->fd, EVFILT_WRITE,
EV_ADD | EV_ONESHOT, c);
}
}
}
}
}
void
kore_event_schedule(int fd, int type, int flags, void *data)
{
EV_SET(&changelist[nchanges], fd, type, flags, 0, 0, data);
nchanges++;
}
void
kore_set_proctitle(char *title)
{
setproctitle("%s", title);
}

609
src/kore.c
View File

@ -18,8 +18,6 @@
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/queue.h>
#include <sys/epoll.h>
#include <sys/prctl.h>
#include <sys/wait.h>
#include <netinet/in.h>
@ -48,21 +46,18 @@
#include "kore.h"
#include "http.h"
#define EPOLL_EVENTS 500
static int efd = -1;
static SSL_CTX *ssl_ctx = NULL;
volatile sig_atomic_t sig_recv;
static struct listener server;
static TAILQ_HEAD(, connection) disconnected;
static TAILQ_HEAD(, connection) worker_clients;
static TAILQ_HEAD(, kore_worker) kore_workers;
static struct passwd *pw = NULL;
static u_int16_t workerid = 0;
static u_int16_t cpu_count = 1;
struct listener server;
pid_t mypid = -1;
u_int16_t cpu_count = 1;
struct kore_worker_h kore_workers;
int kore_debug = 0;
int server_port = 0;
u_int8_t worker_count = 0;
@ -75,17 +70,9 @@ char *kore_pidfile = KORE_PIDFILE_DEFAULT;
static void usage(void);
static void kore_signal(int);
static void kore_worker_wait(int);
static void kore_worker_init(void);
static void kore_write_mypid(void);
static int kore_socket_nonblock(int);
static int kore_server_sslstart(void);
static void kore_event(int, int, void *);
static void kore_worker_spawn(u_int16_t);
static int kore_server_accept(struct listener *);
static void kore_worker_entry(struct kore_worker *);
static void kore_worker_setcpu(struct kore_worker *);
static int kore_connection_handle(struct connection *, int);
static void kore_server_final_disconnect(struct connection *);
static int kore_server_bind(struct listener *, const char *, int);
static int kore_ssl_npn_cb(SSL *, const u_char **, unsigned int *, void *);
@ -146,10 +133,7 @@ main(int argc, char *argv[])
if (kore_certfile == NULL || kore_certkey == NULL)
fatal("missing certificate information");
if ((cpu_count = sysconf(_SC_NPROCESSORS_ONLN)) == -1) {
kore_debug("could not get number of cpu's falling back to 1");
cpu_count = 1;
}
kore_init();
if (!kore_server_sslstart())
fatal("cannot initiate SSL");
@ -163,9 +147,7 @@ main(int argc, char *argv[])
kore_log(LOG_NOTICE, "kore is starting up");
kore_worker_init();
if (prctl(PR_SET_NAME, "kore [main]"))
kore_debug("cannot set process title");
kore_set_proctitle("kore [parent]");
sig_recv = 0;
signal(SIGHUP, kore_signal);
@ -213,6 +195,47 @@ main(int argc, char *argv[])
return (0);
}
int
kore_server_accept(struct listener *l, struct connection **out)
{
socklen_t len;
struct connection *c;
kore_debug("kore_server_accept(%p)", l);
*out = NULL;
len = sizeof(struct sockaddr_in);
c = (struct connection *)kore_malloc(sizeof(*c));
if ((c->fd = accept(l->fd, (struct sockaddr *)&(c->sin), &len)) == -1) {
free(c);
kore_debug("accept(): %s", errno_s);
return (KORE_RESULT_ERROR);
}
if (!kore_socket_nonblock(c->fd)) {
close(c->fd);
free(c);
return (KORE_RESULT_ERROR);
}
c->owner = l;
c->ssl = NULL;
c->flags = 0;
c->inflate_started = 0;
c->deflate_started = 0;
c->client_stream_id = 0;
c->proto = CONN_PROTO_UNKNOWN;
c->state = CONN_STATE_SSL_SHAKE;
TAILQ_INIT(&(c->send_queue));
TAILQ_INIT(&(c->recv_queue));
TAILQ_INIT(&(c->spdy_streams));
TAILQ_INSERT_TAIL(&worker_clients, c, list);
*out = c;
return (KORE_RESULT_OK);
}
void
kore_server_disconnect(struct connection *c)
{
@ -224,6 +247,183 @@ kore_server_disconnect(struct connection *c)
}
}
int
kore_connection_handle(struct connection *c)
{
int r;
u_int32_t len;
const u_char *data;
kore_debug("kore_connection_handle(%p)", c);
switch (c->state) {
case CONN_STATE_SSL_SHAKE:
if (c->ssl == NULL) {
c->ssl = SSL_new(ssl_ctx);
if (c->ssl == NULL) {
kore_debug("SSL_new(): %s", ssl_errno_s);
return (KORE_RESULT_ERROR);
}
SSL_set_fd(c->ssl, c->fd);
}
r = SSL_accept(c->ssl);
if (r <= 0) {
r = SSL_get_error(c->ssl, r);
switch (r) {
case SSL_ERROR_WANT_READ:
case SSL_ERROR_WANT_WRITE:
return (KORE_RESULT_OK);
default:
kore_debug("SSL_accept(): %s", ssl_errno_s);
return (KORE_RESULT_ERROR);
}
}
r = SSL_get_verify_result(c->ssl);
if (r != X509_V_OK) {
kore_debug("SSL_get_verify_result(): %s", ssl_errno_s);
return (KORE_RESULT_ERROR);
}
SSL_get0_next_proto_negotiated(c->ssl, &data, &len);
if (data) {
if (!memcmp(data, "spdy/3", 6))
kore_debug("using SPDY/3");
c->proto = CONN_PROTO_SPDY;
net_recv_queue(c, SPDY_FRAME_SIZE, 0,
NULL, spdy_frame_recv);
} else {
kore_debug("using HTTP/1.1");
c->proto = CONN_PROTO_HTTP;
net_recv_queue(c, HTTP_HEADER_MAX_LEN,
NETBUF_CALL_CB_ALWAYS, NULL,
http_header_recv);
}
c->state = CONN_STATE_ESTABLISHED;
/* FALLTHROUGH */
case CONN_STATE_ESTABLISHED:
if (c->flags & CONN_READ_POSSIBLE) {
if (!net_recv_flush(c))
return (KORE_RESULT_ERROR);
}
if (c->flags & CONN_WRITE_POSSIBLE) {
if (!net_send_flush(c))
return (KORE_RESULT_ERROR);
}
break;
case CONN_STATE_DISCONNECTING:
break;
default:
kore_debug("unknown state on %d (%d)", c->fd, c->state);
break;
}
return (KORE_RESULT_OK);
}
void
kore_worker_spawn(u_int16_t cpu)
{
struct kore_worker *kw;
kw = (struct kore_worker *)kore_malloc(sizeof(*kw));
kw->id = workerid++;
kw->cpu = cpu;
kw->pid = fork();
if (kw->pid == -1)
fatal("could not spawn worker child: %s", errno_s);
if (kw->pid == 0) {
kw->pid = getpid();
kore_worker_entry(kw);
/* NOTREACHED */
}
TAILQ_INSERT_TAIL(&kore_workers, kw, list);
}
void
kore_worker_entry(struct kore_worker *kw)
{
int quit;
char buf[16];
struct connection *c, *cnext;
struct kore_worker *k, *next;
if (chroot(chroot_path) == -1)
fatal("cannot chroot(): %s", errno_s);
if (chdir("/") == -1)
fatal("cannot chdir(): %s", errno_s);
if (setgroups(1, &pw->pw_gid) || setresgid(pw->pw_gid, pw->pw_gid,
pw->pw_gid) || setresuid(pw->pw_uid, pw->pw_uid, pw->pw_uid))
fatal("unable to drop privileges");
snprintf(buf, sizeof(buf), "kore [wrk %d]", kw->id);
kore_set_proctitle(buf);
kore_worker_setcpu(kw);
for (k = TAILQ_FIRST(&kore_workers); k != NULL; k = next) {
next = TAILQ_NEXT(k, list);
TAILQ_REMOVE(&kore_workers, k, list);
free(k);
}
mypid = kw->pid;
sig_recv = 0;
signal(SIGHUP, kore_signal);
signal(SIGQUIT, kore_signal);
http_init();
TAILQ_INIT(&disconnected);
TAILQ_INIT(&worker_clients);
quit = 0;
kore_event_init();
kore_log(LOG_NOTICE, "worker %d going to work (CPU: %d)",
kw->id, kw->cpu);
for (;;) {
if (sig_recv != 0) {
if (sig_recv == SIGHUP)
kore_module_reload();
else if (sig_recv == SIGQUIT)
quit = 1;
sig_recv = 0;
}
kore_event_wait(quit);
http_process();
for (c = TAILQ_FIRST(&disconnected); c != NULL; c = cnext) {
cnext = TAILQ_NEXT(c, list);
kore_server_final_disconnect(c);
}
if (quit && http_request_count == 0)
break;
}
for (c = TAILQ_FIRST(&worker_clients); c != NULL; c = cnext) {
cnext = TAILQ_NEXT(c, list);
net_send_flush(c);
kore_server_final_disconnect(c);
}
for (c = TAILQ_FIRST(&disconnected); c != NULL; c = cnext) {
cnext = TAILQ_NEXT(c, list);
net_send_flush(c);
kore_server_final_disconnect(c);
}
kore_debug("worker %d shutting down", kw->id);
exit(0);
}
static int
kore_server_sslstart(void)
{
@ -301,47 +501,6 @@ kore_server_bind(struct listener *l, const char *ip, int port)
return (KORE_RESULT_OK);
}
static int
kore_server_accept(struct listener *l)
{
socklen_t len;
struct connection *c;
kore_debug("kore_server_accept(%p)", l);
len = sizeof(struct sockaddr_in);
c = (struct connection *)kore_malloc(sizeof(*c));
if ((c->fd = accept(l->fd, (struct sockaddr *)&(c->sin), &len)) == -1) {
free(c);
kore_debug("accept(): %s", errno_s);
return (KORE_RESULT_ERROR);
}
if (!kore_socket_nonblock(c->fd)) {
close(c->fd);
free(c);
return (KORE_RESULT_ERROR);
}
c->owner = l;
c->ssl = NULL;
c->flags = 0;
c->inflate_started = 0;
c->deflate_started = 0;
c->client_stream_id = 0;
c->proto = CONN_PROTO_UNKNOWN;
c->state = CONN_STATE_SSL_SHAKE;
TAILQ_INIT(&(c->send_queue));
TAILQ_INIT(&(c->recv_queue));
TAILQ_INIT(&(c->spdy_streams));;
TAILQ_INSERT_TAIL(&worker_clients, c, list);
kore_event(c->fd, EPOLLIN | EPOLLOUT | EPOLLET, c);
return (KORE_RESULT_OK);
}
static void
kore_server_final_disconnect(struct connection *c)
{
@ -390,309 +549,6 @@ kore_server_final_disconnect(struct connection *c)
free(c);
}
static int
kore_connection_handle(struct connection *c, int flags)
{
int r;
u_int32_t len;
const u_char *data;
kore_debug("kore_connection_handle(%p, %d)", c, flags);
if (flags & EPOLLIN)
c->flags |= CONN_READ_POSSIBLE;
if (flags & EPOLLOUT)
c->flags |= CONN_WRITE_POSSIBLE;
switch (c->state) {
case CONN_STATE_SSL_SHAKE:
if (c->ssl == NULL) {
c->ssl = SSL_new(ssl_ctx);
if (c->ssl == NULL) {
kore_debug("SSL_new(): %s", ssl_errno_s);
return (KORE_RESULT_ERROR);
}
SSL_set_fd(c->ssl, c->fd);
}
r = SSL_accept(c->ssl);
if (r <= 0) {
r = SSL_get_error(c->ssl, r);
switch (r) {
case SSL_ERROR_WANT_READ:
case SSL_ERROR_WANT_WRITE:
return (KORE_RESULT_OK);
default:
kore_debug("SSL_accept(): %s", ssl_errno_s);
return (KORE_RESULT_ERROR);
}
}
r = SSL_get_verify_result(c->ssl);
if (r != X509_V_OK) {
kore_debug("SSL_get_verify_result(): %s", ssl_errno_s);
return (KORE_RESULT_ERROR);
}
SSL_get0_next_proto_negotiated(c->ssl, &data, &len);
if (data) {
if (!memcmp(data, "spdy/3", 6))
kore_debug("using SPDY/3");
c->proto = CONN_PROTO_SPDY;
net_recv_queue(c, SPDY_FRAME_SIZE, 0,
NULL, spdy_frame_recv);
} else {
kore_debug("using HTTP/1.1");
c->proto = CONN_PROTO_HTTP;
net_recv_queue(c, HTTP_HEADER_MAX_LEN,
NETBUF_CALL_CB_ALWAYS, NULL,
http_header_recv);
}
c->state = CONN_STATE_ESTABLISHED;
/* FALLTHROUGH */
case CONN_STATE_ESTABLISHED:
if (c->flags & CONN_READ_POSSIBLE) {
if (!net_recv_flush(c))
return (KORE_RESULT_ERROR);
}
if (c->flags & CONN_WRITE_POSSIBLE) {
if (!net_send_flush(c))
return (KORE_RESULT_ERROR);
}
break;
case CONN_STATE_DISCONNECTING:
break;
default:
kore_debug("unknown state on %d (%d)", c->fd, c->state);
break;
}
return (KORE_RESULT_OK);
}
static void
kore_worker_init(void)
{
u_int16_t i, cpu;
if (worker_count == 0)
fatal("no workers specified");
kore_debug("kore_worker_init(): system has %d cpu's", cpu_count);
kore_debug("kore_worker_init(): starting %d workers", worker_count);
if (worker_count > cpu_count)
kore_debug("kore_worker_init(): more workers then cpu's");
cpu = 0;
TAILQ_INIT(&kore_workers);
for (i = 0; i < worker_count; i++) {
kore_worker_spawn(cpu++);
if (cpu == cpu_count)
cpu = 0;
}
}
static void
kore_worker_spawn(u_int16_t cpu)
{
struct kore_worker *kw;
kw = (struct kore_worker *)kore_malloc(sizeof(*kw));
kw->id = workerid++;
kw->cpu = cpu;
kw->pid = fork();
if (kw->pid == -1)
fatal("could not spawn worker child: %s", errno_s);
if (kw->pid == 0) {
kw->pid = getpid();
kore_worker_entry(kw);
/* NOTREACHED */
}
TAILQ_INSERT_TAIL(&kore_workers, kw, list);
}
static void
kore_worker_wait(int final)
{
int r;
siginfo_t info;
struct kore_worker k, *kw, *next;
memset(&info, 0, sizeof(info));
if (final)
r = waitid(P_ALL, 0, &info, WEXITED);
else
r = waitid(P_ALL, 0, &info, WEXITED | WNOHANG);
if (r == -1) {
kore_debug("waitid(): %s", errno_s);
return;
}
if (info.si_pid == 0)
return;
for (kw = TAILQ_FIRST(&kore_workers); kw != NULL; kw = next) {
next = TAILQ_NEXT(kw, list);
if (kw->pid != info.si_pid)
continue;
k = *kw;
TAILQ_REMOVE(&kore_workers, kw, list);
kore_log(LOG_NOTICE, "worker %d (%d)-> status %d (%d)",
kw->id, info.si_pid, info.si_status, info.si_code);
free(kw);
if (final)
continue;
if (info.si_code == CLD_EXITED ||
info.si_code == CLD_KILLED ||
info.si_code == CLD_DUMPED) {
kore_log(LOG_NOTICE,
"worker %d (pid: %d) gone, respawning new one",
k.id, k.pid);
kore_worker_spawn(k.cpu);
}
}
}
static void
kore_worker_setcpu(struct kore_worker *kw)
{
cpu_set_t cpuset;
CPU_ZERO(&cpuset);
CPU_SET(kw->cpu, &cpuset);
if (sched_setaffinity(0, sizeof(cpu_set_t), &cpuset) == -1) {
kore_debug("kore_worker_setcpu(): %s", errno_s);
} else {
kore_debug("kore_worker_setcpu(): worker %d on cpu %d",
kw->id, kw->cpu);
}
}
static void
kore_worker_entry(struct kore_worker *kw)
{
char buf[16];
struct epoll_event *events;
struct connection *c, *cnext;
struct kore_worker *k, *next;
int n, i, *fd, quit;
snprintf(buf, sizeof(buf), "kore [wrk %d]", kw->id);
if (prctl(PR_SET_NAME, buf) == -1)
kore_debug("cannot set process title");
if (chroot(chroot_path) == -1)
fatal("cannot chroot(): %s", errno_s);
if (chdir("/") == -1)
fatal("cannot chdir(): %s", errno_s);
if (setgroups(1, &pw->pw_gid) || setresgid(pw->pw_gid, pw->pw_gid,
pw->pw_gid) || setresuid(pw->pw_uid, pw->pw_uid, pw->pw_uid))
fatal("unable to drop privileges");
kore_worker_setcpu(kw);
for (k = TAILQ_FIRST(&kore_workers); k != NULL; k = next) {
next = TAILQ_NEXT(k, list);
TAILQ_REMOVE(&kore_workers, k, list);
free(k);
}
mypid = kw->pid;
if ((efd = epoll_create(1000)) == -1)
fatal("epoll_create(): %s", errno_s);
sig_recv = 0;
signal(SIGHUP, kore_signal);
signal(SIGQUIT, kore_signal);
http_init();
TAILQ_INIT(&disconnected);
TAILQ_INIT(&worker_clients);
quit = 0;
kore_event(server.fd, EPOLLIN, &server);
events = kore_calloc(EPOLL_EVENTS, sizeof(struct epoll_event));
kore_log(LOG_NOTICE, "worker %d going to work (CPU: %d)",
kw->id, kw->cpu);
for (;;) {
if (sig_recv != 0) {
if (sig_recv == SIGHUP)
kore_module_reload();
else if (sig_recv == SIGQUIT)
quit = 1;
sig_recv = 0;
}
n = epoll_wait(efd, events, EPOLL_EVENTS, 100);
if (n == -1) {
if (errno == EINTR)
continue;
fatal("epoll_wait(): %s", errno_s);
}
if (n > 0)
kore_debug("main(): %d sockets available", n);
for (i = 0; i < n; i++) {
fd = (int *)events[i].data.ptr;
if (events[i].events & EPOLLERR ||
events[i].events & EPOLLHUP) {
if (*fd == server.fd)
fatal("error on server socket");
c = (struct connection *)events[i].data.ptr;
kore_server_disconnect(c);
continue;
}
if (*fd == server.fd) {
if (!quit)
kore_server_accept(&server);
} else {
c = (struct connection *)events[i].data.ptr;
if (!kore_connection_handle(c,
events[i].events))
kore_server_disconnect(c);
}
}
http_process();
for (c = TAILQ_FIRST(&disconnected); c != NULL; c = cnext) {
cnext = TAILQ_NEXT(c, list);
kore_server_final_disconnect(c);
}
if (quit && http_request_count == 0)
break;
}
for (c = TAILQ_FIRST(&worker_clients); c != NULL; c = cnext) {
cnext = TAILQ_NEXT(c, list);
net_send_flush(c);
kore_server_final_disconnect(c);
}
for (c = TAILQ_FIRST(&disconnected); c != NULL; c = cnext) {
cnext = TAILQ_NEXT(c, list);
net_send_flush(c);
kore_server_final_disconnect(c);
}
kore_debug("worker %d shutting down", kw->id);
exit(0);
}
static int
kore_socket_nonblock(int fd)
{
@ -725,25 +581,6 @@ kore_ssl_npn_cb(SSL *ssl, const u_char **data, unsigned int *len, void *arg)
return (SSL_TLSEXT_ERR_OK);
}
static void
kore_event(int fd, int flags, void *udata)
{
struct epoll_event evt;
kore_debug("kore_event(%d, %d, %p)", fd, flags, udata);
evt.events = flags;
evt.data.ptr = udata;
if (epoll_ctl(efd, EPOLL_CTL_ADD, fd, &evt) == -1) {
if (errno == EEXIST) {
if (epoll_ctl(efd, EPOLL_CTL_MOD, fd, &evt) == -1)
fatal("epoll_ctl() MOD: %s", errno_s);
} else {
fatal("epoll_ctl() ADD: %s", errno_s);
}
}
}
static void
kore_write_mypid(void)
{

230
src/linux.c Normal file
View File

@ -0,0 +1,230 @@
/*
* Copyright (c) 2013 Joris Vink <joris@coders.se>
*
* Permission to use, copy, modify, and distribute this software for any
* purpose with or without fee is hereby granted, provided that the above
* copyright notice and this permission notice appear in all copies.
*
* THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
* WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
* MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
* ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
* WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
* ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
* OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
#include <sys/param.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/queue.h>
#include <sys/epoll.h>
#include <sys/prctl.h>
#include <sys/wait.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <openssl/err.h>
#include <openssl/ssl.h>
#include <errno.h>
#include <fcntl.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sched.h>
#include <syslog.h>
#include <unistd.h>
#include <time.h>
#include <regex.h>
#include <zlib.h>
#include <unistd.h>
#include "spdy.h"
#include "kore.h"
#include "http.h"
#define EPOLL_EVENTS 500
static int efd = -1;
static struct epoll_event *events = NULL;
void
kore_init(void)
{
if ((cpu_count = sysconf(_SC_NPROCESSORS_ONLN)) == -1) {
kore_debug("could not get number of cpu's falling back to 1");
cpu_count = 1;
}
}
void
kore_worker_init(void)
{
u_int16_t i, cpu;
if (worker_count == 0)
fatal("no workers specified");
kore_debug("kore_worker_init(): system has %d cpu's", cpu_count);
kore_debug("kore_worker_init(): starting %d workers", worker_count);
if (worker_count > cpu_count)
kore_debug("kore_worker_init(): more workers then cpu's");
cpu = 0;
TAILQ_INIT(&kore_workers);
for (i = 0; i < worker_count; i++) {
kore_worker_spawn(cpu++);
if (cpu == cpu_count)
cpu = 0;
}
}
void
kore_worker_wait(int final)
{
int r;
siginfo_t info;
struct kore_worker k, *kw, *next;
memset(&info, 0, sizeof(info));
if (final)
r = waitid(P_ALL, 0, &info, WEXITED);
else
r = waitid(P_ALL, 0, &info, WEXITED | WNOHANG);
if (r == -1) {
kore_debug("waitid(): %s", errno_s);
return;
}
if (info.si_pid == 0)
return;
for (kw = TAILQ_FIRST(&kore_workers); kw != NULL; kw = next) {
next = TAILQ_NEXT(kw, list);
if (kw->pid != info.si_pid)
continue;
k = *kw;
TAILQ_REMOVE(&kore_workers, kw, list);
kore_log(LOG_NOTICE, "worker %d (%d)-> status %d (%d)",
kw->id, info.si_pid, info.si_status, info.si_code);
free(kw);
if (final)
continue;
if (info.si_code == CLD_EXITED ||
info.si_code == CLD_KILLED ||
info.si_code == CLD_DUMPED) {
kore_log(LOG_NOTICE,
"worker %d (pid: %d) gone, respawning new one",
k.id, k.pid);
kore_worker_spawn(k.cpu);
}
}
}
void
kore_worker_setcpu(struct kore_worker *kw)
{
cpu_set_t cpuset;
CPU_ZERO(&cpuset);
CPU_SET(kw->cpu, &cpuset);
if (sched_setaffinity(0, sizeof(cpu_set_t), &cpuset) == -1) {
kore_debug("kore_worker_setcpu(): %s", errno_s);
} else {
kore_debug("kore_worker_setcpu(): worker %d on cpu %d",
kw->id, kw->cpu);
}
}
void
kore_event_init(void)
{
if ((efd = epoll_create(1000)) == -1)
fatal("epoll_create(): %s", errno_s);
events = kore_calloc(EPOLL_EVENTS, sizeof(struct epoll_event));
kore_event_schedule(server.fd, EPOLLIN, 0, &server);
}
void
kore_event_wait(int quit)
{
struct connection *c;
int n, i, *fd;
n = epoll_wait(efd, events, EPOLL_EVENTS, 100);
if (n == -1) {
if (errno == EINTR)
return;
fatal("epoll_wait(): %s", errno_s);
}
if (n > 0)
kore_debug("main(): %d sockets available", n);
for (i = 0; i < n; i++) {
fd = (int *)events[i].data.ptr;
if (events[i].events & EPOLLERR ||
events[i].events & EPOLLHUP) {
if (*fd == server.fd)
fatal("error on server socket");
c = (struct connection *)events[i].data.ptr;
kore_server_disconnect(c);
continue;
}
if (*fd == server.fd) {
if (!quit) {
kore_server_accept(&server, &c);
if (c == NULL)
continue;
kore_event_schedule(c->fd,
EPOLLIN | EPOLLOUT | EPOLLET, 0, c);
}
} else {
c = (struct connection *)events[i].data.ptr;
if (events[i].events & EPOLLIN)
c->flags |= CONN_READ_POSSIBLE;
if (events[i].events & EPOLLOUT)
c->flags |= CONN_WRITE_POSSIBLE;
if (!kore_connection_handle(c))
kore_server_disconnect(c);
}
}
}
void
kore_event_schedule(int fd, int type, int flags, void *udata)
{
struct epoll_event evt;
kore_debug("kore_event(%d, %d, %d, %p)", fd, type, flags, udata);
evt.events = type;
evt.data.ptr = udata;
if (epoll_ctl(efd, EPOLL_CTL_ADD, fd, &evt) == -1) {
if (errno == EEXIST) {
if (epoll_ctl(efd, EPOLL_CTL_MOD, fd, &evt) == -1)
fatal("epoll_ctl() MOD: %s", errno_s);
} else {
fatal("epoll_ctl() ADD: %s", errno_s);
}
}
}
void
kore_set_proctitle(char *title)
{
if (prctl(PR_SET_NAME, title) == -1)
kore_debug("prctl(): %s", errno_s);
}

1
src/net.c
View File

@ -18,7 +18,6 @@
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/queue.h>
#include <sys/epoll.h>
#include <netinet/in.h>
#include <arpa/inet.h>