Commit 65e0473a authored by Marek Vavruša's avatar Marek Vavruša

daemon: split worker to 'engine' and workers

The engine is responsible for central managemento of resources and work
distribution. This allows to spawn multiple loops.

The engine configuration / interface is based on Lua 5.1-5.2, including
a RPC and a simple CLI interface. This is going to be the core of configuration
later on.
parent 10103e6e
/* Copyright (C) 2015 CZ.NIC, z.s.p.o. <knot-dns@labs.nic.cz>
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <libknot/internal/mempool.h>
#include "daemon/cmd.h"
#include "lib/defines.h"
struct cmd {
const char *name;
int (*cb)(struct worker_ctx *, char *);
};
static int help(struct worker_ctx *worker, char *args)
{
printf("help:\n show this help\n");
printf("context:\n show context information\n");
printf("load:\n load module\n");
printf("unload:\n unload module\n");
struct kr_context *ctx = &worker->resolve;
for (unsigned i = 0; i < ctx->mod_loaded; ++i) {
struct kr_module *mod = &ctx->modules[i];
for (struct kr_prop *p = mod->props; p && p->name; ++p) {
printf("%s.%s:\n %s\n", mod->name, p->name, p->info);
}
}
return kr_ok();
}
static int context(struct worker_ctx *worker, char *args)
{
struct kr_context *ctx = &worker->resolve;
/* Modules */
printf("modules:\n");
for (unsigned i = 0; i < ctx->mod_loaded; ++i) {
struct kr_module *mod = &ctx->modules[i];
printf(" %s\n", mod->name);
}
/* Options */
printf("options: 0x%x\n", ctx->options);
return kr_ok();
}
static int mod_load(struct worker_ctx *worker, char *args)
{
struct kr_context *ctx = &worker->resolve;
char *saveptr = NULL;
char *prop_name = strtok_r(args, " \t\n\r", &saveptr);
return kr_context_register(ctx, prop_name);
}
static int mod_unload(struct worker_ctx *worker, char *args)
{
return kr_error(ENOTSUP);
}
static int cmd_exec_prop(struct worker_ctx *worker, char *name, char *prop, char *args)
{
struct kr_context *ctx = &worker->resolve;
for (unsigned i = 0; i < ctx->mod_loaded; ++i) {
struct kr_module *mod = &ctx->modules[i];
if (strncmp(mod->name, name, strlen(mod->name)) != 0) {
continue;
}
for (struct kr_prop *p = mod->props; p && p->name; ++p) {
if (strncmp(p->name, prop, strlen(p->name)) == 0) {
auto_free char *res = p->cb(ctx, mod, args);
printf("%s\n", res);
return kr_ok();
}
}
}
return kr_error(ENOENT);
}
int cmd_exec(struct worker_ctx *worker, char *cmd)
{
static struct cmd cmd_table[] = {
{ "help", &help },
{ "context", &context },
{ "load", &mod_load },
{ "unload", &mod_unload },
{ NULL, NULL }
};
int ret = kr_error(ENOENT);
char *args = strchr(cmd, ' ');
if (args != NULL) {
*args = '\0';
args += 1;
}
/* Search builtin namespace. */
for (struct cmd *c = cmd_table; c->name; ++c) {
if (strncmp(cmd, c->name, strlen(c->name)) == 0) {
return c->cb(worker, args);
}
}
/* Search module namespace. */
char *prop = strchr(cmd, '.');
if (prop != NULL) {
ret = cmd_exec_prop(worker, cmd, prop + 1, args);
}
return ret;
}
\ No newline at end of file
......@@ -2,10 +2,16 @@ kresolved_SOURCES := \
daemon/layer/query.c \
daemon/udp.c \
daemon/tcp.c \
daemon/cmd.c \
daemon/engine.c \
daemon/worker.c \
daemon/main.c
# Embed resources
daemon/engine.o: daemon/lua/init.inc
%.inc: %.lua
@$(call quiet,XXD,$<) -i < $< > $@
@echo ', 0x00' >> $@
# Dependencies
kresolved_DEPEND := $(libkresolve)
kresolved_LIBS := $(libkresolve_TARGET) $(libknot_LIBS) $(libuv_LIBS) $(lua_LIBS)
......@@ -19,4 +25,6 @@ endif
daemon: $(kresolved)
daemon-install: kresolved-install
daemon-clean: kresolved-clean
@$(RM) daemon/lua/*.inc
.PHONY: daemon daemon-install daemon-clean
/* Copyright (C) 2015 CZ.NIC, z.s.p.o. <knot-dns@labs.nic.cz>
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <uv.h>
#include <unistd.h>
#include <libknot/internal/mem.h>
#include "daemon/engine.h"
#include "lib/cache.h"
#include "lib/defines.h"
/*
* Defines
*/
#define CACHE_DEFAULT_PATH "/tmp/kresolved"
#define CACHE_DEFAULT_SIZE 10*(1024*1024)
/*
* Global bindings.
*/
/** Print help and available commands. */
static int l_help(lua_State *L)
{
static const char *help_str =
"help()\n show this help\n"
"quit()\n quit\n"
;
puts(help_str);
/* No results */
return 0;
}
/** Quit current executable. */
static int l_quit(lua_State *L)
{
/* Stop engine */
engine_stop(engine_luaget(L));
/* No results */
return 0;
}
/** Trampoline function for modules. */
static int l_trampoline(lua_State *L)
{
const char *name = lua_tostring(L, lua_upvalueindex(1));
const char *property = lua_tostring(L, lua_upvalueindex(2));
struct engine *engine = engine_luaget(L);
struct kr_context *ctx = &engine->resolver;
/* Find module. */
for (unsigned i = 0; i < engine->modules.len; ++i) {
struct kr_module *module = &engine->modules.at[i];
if (strcmp(module->name, name) != 0) {
continue;
}
/* Find property. */
for (struct kr_prop *p = module->props; p && p->name; ++p) {
if (strcmp(p->name, property) == 0) {
auto_free char *ret = p->cb(ctx, module, NULL);
lua_pushstring(L, ret);
return 1;
}
}
break;
}
/* No results */
return 0;
}
/*
* Engine API.
*/
static int init_resolver(struct engine *engine)
{
/* Open resolution context */
engine->resolver.modules = &engine->modules;
/* Load basic modules */
engine_register(engine, "iterate");
engine_register(engine, "itercache");
return kr_ok();
}
static int init_state(struct engine *engine)
{
/* Initialize Lua state */
engine->L = luaL_newstate();
if (engine->L == NULL) {
return kr_error(ENOMEM);
}
/* Initialize used libraries. */
luaL_openlibs(engine->L);
/* Global functions */
lua_pushcfunction(engine->L, l_help);
lua_setglobal(engine->L, "help");
lua_pushcfunction(engine->L, l_quit);
lua_setglobal(engine->L, "quit");
lua_pushlightuserdata(engine->L, engine);
lua_setglobal(engine->L, "__engine");
return kr_ok();
}
int engine_init(struct engine *engine, mm_ctx_t *pool)
{
if (engine == NULL) {
return kr_error(EINVAL);
}
memset(engine, 0, sizeof(*engine));
engine->pool = pool;
/* Initialize state */
int ret = init_state(engine);
if (ret != 0) {
engine_deinit(engine);
}
/* Initialize resolver */
ret = init_resolver(engine);
if (ret != 0) {
return ret;
}
return ret;
}
void engine_deinit(struct engine *engine)
{
if (engine == NULL) {
return;
}
/* Unload modules. */
for (size_t i = 0; i < engine->modules.len; ++i) {
kr_module_unload(&engine->modules.at[i]);
}
array_clear(engine->modules);
if (engine->L) {
lua_close(engine->L);
}
kr_cache_close(engine->resolver.cache);
}
int engine_cmd(struct engine *engine, const char *str)
{
if (engine == NULL || engine->L == NULL) {
return kr_error(ENOEXEC);
}
/* Evaluate results */
int ret = luaL_dostring(engine->L, str);
/* Print results. */
int nres = lua_gettop(engine->L);
for (int i = 0; i < nres; ++i) {
const char *out = lua_tostring(engine->L, -1);
if (out != NULL) {
printf("%s\n", out);
}
lua_pop(engine->L, 1);
}
/* Check result. */
if (ret != 0) {
return kr_error(EINVAL);
}
return kr_ok();
}
static int engine_loadconf(struct engine *engine)
{
/* Load config file */
int ret = 0;
if(access("config", F_OK ) != -1 ) {
ret = luaL_dofile(engine->L, "config");
} else {
/* Load defaults */
static const char config_init[] = {
#include "daemon/lua/init.inc"
};
ret = luaL_dostring(engine->L, config_init);
}
/* Evaluate */
if (ret != 0) {
fprintf(stderr, "error: %s\n", lua_tostring(engine->L, -1));
lua_pop(engine->L, 1);
return kr_error(EINVAL);
}
return kr_ok();
}
int engine_start(struct engine *engine)
{
/* Load configuration. */
int ret = engine_loadconf(engine);
if (ret != 0) {
return ret;
}
return uv_run(uv_default_loop(), UV_RUN_DEFAULT);
}
void engine_stop(struct engine *engine)
{
uv_stop(uv_default_loop());
}
int engine_register(struct engine *engine, const char *name)
{
if (engine == NULL || name == NULL) {
return kr_error(EINVAL);
}
/* Load module */
size_t next = engine->modules.len;
array_reserve(engine->modules, next + 1);
struct kr_module *module = &engine->modules.at[next];
int ret = kr_module_load(module, name, NULL);
if (ret != 0) {
return ret;
} else {
engine->modules.len += 1;
}
/* Register properties */
if (module->props) {
lua_newtable(engine->L);
for (struct kr_prop *p = module->props; p->name; ++p) {
lua_pushstring(engine->L, module->name);
lua_pushstring(engine->L, p->name);
lua_pushcclosure(engine->L, l_trampoline, 2);
lua_setfield(engine->L, -2, p->name);
}
lua_setglobal(engine->L, module->name);
}
return kr_ok();
}
int engine_unregister(struct engine *engine, const char *name)
{
/* Find matching module. */
modulelist_t *mod_list = &engine->modules;
size_t found = mod_list->len;
for (size_t i = 0; i < mod_list->len; ++i) {
if (strcmp(mod_list->at[i].name, name) == 0) {
found = i;
break;
}
}
/* Unregister module */
if (found < mod_list->len) {
kr_module_unload(&mod_list->at[found]);
array_del(*mod_list, found);
return kr_ok();
}
return kr_error(ENOENT);
}
void engine_lualib(struct engine *engine, const char *name, lua_CFunction lib_cb)
{
if (engine != NULL) {
#if LUA_VERSION_NUM < 502
lib_cb(engine->L);
#else
luaL_requiref(engine->L, name, lib_cb, 1);
lua_pop(engine->L, 1);
#endif
}
}
struct engine *engine_luaget(lua_State *L)
{
lua_getglobal(L, "__engine");
struct engine *engine = lua_touserdata(L, -1);
lua_pop(engine->L, 1);
return engine;
}
\ No newline at end of file
......@@ -16,7 +16,27 @@
#pragma once
#include <uv.h>
#include "daemon/worker.h"
#include <lua.h>
#include <lualib.h>
#include <lauxlib.h>
int cmd_exec(struct worker_ctx *worker, char *cmd);
\ No newline at end of file
#include "lib/resolve.h"
#include "lib/generic/array.h"
struct engine {
struct kr_context resolver;
modulelist_t modules;
mm_ctx_t *pool;
lua_State *L;
};
int engine_init(struct engine *engine, mm_ctx_t *pool);
void engine_deinit(struct engine *engine);
int engine_cmd(struct engine *engine, const char *str);
int engine_start(struct engine *engine);
void engine_stop(struct engine *engine);
int engine_register(struct engine *engine, const char *module);
int engine_unregister(struct engine *engine, const char *module);
/** Return engine light userdata. */
void engine_lualib(struct engine *engine, const char *name, lua_CFunction lib_cb);
struct engine *engine_luaget(lua_State *L);
\ No newline at end of file
-- Default configuration
cache.open('.', 10485760)
......@@ -24,7 +24,7 @@
#include "lib/resolve.h"
#include "daemon/udp.h"
#include "daemon/tcp.h"
#include "daemon/cmd.h"
#include "daemon/engine.h"
static void tty_read(uv_stream_t *stream, ssize_t nread, const uv_buf_t *buf)
{
......@@ -33,10 +33,7 @@ static void tty_read(uv_stream_t *stream, ssize_t nread, const uv_buf_t *buf)
char *cmd = buf->base;
cmd[nread - 1] = '\0';
/* Execute */
int ret = cmd_exec((struct worker_ctx *)stream->data, cmd);
if (ret != 0) {
printf("ret: %s\n", kr_strerror(ret));
}
engine_cmd((struct engine *)stream->data, cmd);
}
printf("> ");
......@@ -56,11 +53,13 @@ void signal_handler(uv_signal_t *handle, int signum)
static void help(int argc, char *argv[])
{
printf("Usage: %s [parameters]\n", argv[0]);
printf("Usage: %s [parameters] [rundir]\n", argv[0]);
printf("\nParameters:\n"
" -a, --addr=[addr] Server address (default localhost#53).\n"
" -V, --version Print version of the server.\n"
" -h, --help Print help and usage.\n");
" -a, --addr=[addr] Server address (default: localhost#53).\n"
" -v, --version Print version of the server.\n"
" -h, --help Print help and usage.\n"
"Options:\n"
" [rundir] Path to the working directory (default: .)\n");
}
static int set_addr(struct sockaddr_storage *ss, char *addr)
......@@ -89,21 +88,21 @@ int main(int argc, char **argv)
int c = 0, li = 0, ret = 0;
struct option opts[] = {
{"addr", required_argument, 0, 'a'},
{"version", no_argument, 0, 'V'},
{"version", no_argument, 0, 'v'},
{"help", no_argument, 0, 'h'},
{0, 0, 0, 0}
};
while ((c = getopt_long(argc, argv, "a:Vh", opts, &li)) != -1) {
while ((c = getopt_long(argc, argv, "a:vh", opts, &li)) != -1) {
switch (c)
{
case 'a':
ret = set_addr(&addr, optarg);
if (ret != 0) {
fprintf(stderr, "Address '%s': %s\n", optarg, knot_strerror(ret));
fprintf(stderr, "[system]: address '%s': %s\n", optarg, knot_strerror(ret));
return EXIT_FAILURE;
}
break;
case 'V':
case 'v':
printf("%s, version %s\n", "Knot DNS Resolver", PACKAGE_VERSION);
return EXIT_SUCCESS;
case 'h':
......@@ -116,19 +115,39 @@ int main(int argc, char **argv)
}
}
mm_ctx_t mm;
mm_ctx_init(&mm);
uv_loop_t *loop = uv_default_loop();
/* Switch to rundir. */
if (optind < argc) {
ret = chdir(argv[optind]);
if (ret != 0) {
fprintf(stderr, "[system] rundir '%s': %s\n", argv[optind], strerror(errno));
return EXIT_FAILURE;
}
printf("[system] rundir '%s'\n", argv[optind]);
}
/* Block signals. */
uv_loop_t *loop = uv_default_loop();
uv_signal_t sigint;
uv_signal_init(loop, &sigint);
uv_signal_start(&sigint, signal_handler, SIGINT);
/* Create a worker. */
struct worker_ctx worker;
worker_init(&worker, &mm);
/* Create a server engine. */
mm_ctx_t pool;
mm_ctx_mempool(&pool, 4096);
struct engine engine;
ret = engine_init(&engine, &pool);
if (ret != 0) {
fprintf(stderr, "[system] failed to initialize engine: %s\n", kr_strerror(ret));
return EXIT_FAILURE;
}
/* Create main worker. */
struct worker_ctx worker = {
.engine = &engine,
.loop = loop,
.mm = NULL
};
/* Bind to sockets. */
char addr_str[SOCKADDR_STRLEN] = {'\0'};
......@@ -139,36 +158,36 @@ int main(int argc, char **argv)
uv_tcp_init(loop, &tcp_sock);
printf("[system] listening on '%s/UDP'\n", addr_str);
ret = udp_bind((uv_handle_t *)&udp_sock, &worker, (struct sockaddr *)&addr);
if (ret == KNOT_EOK) {
if (ret == 0) {
printf("[system] listening on '%s/TCP'\n", addr_str);
ret = tcp_bind((uv_handle_t *)&tcp_sock, &worker, (struct sockaddr *)&addr);
}
/* Allocate TTY */
uv_pipe_t pipe;
uv_pipe_init(loop, &pipe, 0);
uv_pipe_open(&pipe, 0);
pipe.data = &worker;
/* Check results */
if (ret != KNOT_EOK) {
if (ret != 0) {
fprintf(stderr, "[system] bind to '%s' %s\n", addr_str, knot_strerror(ret));
ret = EXIT_FAILURE;
} else {
/* Allocate TTY */
uv_pipe_t pipe;
uv_pipe_init(loop, &pipe, 0);
uv_pipe_open(&pipe, 0);
pipe.data = &engine;
/* Interactive stdin */
if (!feof(stdin)) {
printf("[system] started in interactive mode, type 'help'\n");
printf("[system] started in interactive mode, type 'help()'\n");
tty_read(NULL, 0, NULL);
uv_read_start((uv_stream_t*) &pipe, tty_alloc, tty_read);
}
/* Run the event loop. */
ret = uv_run(loop, UV_RUN_DEFAULT);
ret = engine_start(&engine);
}
/* Cleanup. */
fprintf(stderr, "\n[system] quitting\n");
udp_unbind((uv_handle_t *)&udp_sock);
tcp_unbind((uv_handle_t *)&tcp_sock);
worker_deinit(&worker);
return ret;
}
......@@ -18,71 +18,28 @@
#include <libknot/packet/pkt.h>
#include <libknot/internal/net.h>
#include <libknot/errcode.h>
#include "daemon/worker.h"
#include "daemon/engine.h"
#include "daemon/layer/query.h"
/* Defines */
#define CACHE_DEFAULT_SIZE 10*1024*1024
int worker_init(struct worker_ctx *worker, mm_ctx_t *mm)
{
if (worker == NULL) {
return KNOT_EINVAL;
}
memset(worker, 0, sizeof(struct worker_ctx));
worker->pool = mm;
/* Open resolution context */
int ret = kr_context_init(&worker->resolve, mm);
if (ret != KNOT_EOK) {
return ret;
}
/* Open resolution context cache */
worker->resolve.cache = kr_cache_open("/tmp/kresolved", mm, CACHE_DEFAULT_SIZE);
if (worker->resolve.cache == NULL) {
fprintf(stderr, "Cache directory '/tmp/kresolved' not exists, exiting.\n");
kr_context_deinit(&worker->resolve);
return KNOT_ERROR;
}
/* Load basic modules */
kr_context_register(&worker->resolve, "iterate");
kr_context_register(&worker->resolve, "itercache");
kr_context_register(&worker->resolve, "hints");
return KNOT_EOK;
}
void worker_deinit(struct worker_ctx *worker)
{
if (worker == NULL) {
return;
}
kr_context_deinit(&worker->resolve);
}
int worker_exec(struct worker_ctx *worker, knot_pkt_t *answer, knot_pkt_t *query)
{
if (worker == NULL) {
return KNOT_EINVAL;
return kr_error(EINVAL);
}
/* Parse query packet. */
int ret = knot_pkt_parse(query, 0);
if (ret != KNOT_EOK) {
return ret; /* Ignore malformed query. */
return kr_error(EPROTO); /* Ignore malformed query. */
}
/* Process query packet. */
knot_layer_t proc;
memset(&proc, 0, sizeof(knot_layer_t));
proc.mm = worker->pool;
knot_layer_begin(&proc, LAYER_QUERY, &worker->resolve);
proc.mm = worker->mm;
knot_layer_begin(&proc, LAYER_QUERY, &worker->engine->resolver);
int state = knot_layer_consume(&proc, query);
/* Build an answer. */
......@@ -94,5 +51,5 @@ int worker_exec(struct worker_ctx *worker, knot_pkt_t *answer, knot_pkt_t *query
/* Cleanup. */
knot_layer_finish(&proc);
return KNOT_EOK;
return kr_ok();
}
......@@ -16,38 +16,25 @@
#pragma once
#include <libknot/packet/pkt.h>
#include <libknot/internal/mempattern.h>
#include "lib/resolve.h"
#include "daemon/engine.h"
/**
* Query resolution worker.
*/
struct worker_ctx {
struct kr_context resolve;
struct engine *engine;
uv_loop_t *loop;
mm_ctx_t *mm;
};