Commit 6f10cd99 authored by Vladimír Čunát's avatar Vladimír Čunát

Merge !335: predict: fixes for refreshing expiring RRs

parents 537220bd d5c165e0
Pipeline #10712 canceled with stages
in 75 minutes and 34 seconds
......@@ -1188,7 +1188,10 @@ static int wrk_resolve(lua_State *L)
lua_error(L);
}
uint8_t dname[KNOT_DNAME_MAXLEN];
knot_dname_from_str(dname, lua_tostring(L, 1), sizeof(dname));
if (!knot_dname_from_str(dname, lua_tostring(L, 1), sizeof(dname))) {
lua_pushstring(L, "invalid qname");
lua_error(L);
};
/* Check class and type */
uint16_t rrtype = lua_tointeger(L, 2);
if (!lua_isnumber(L, 2)) {
......
......@@ -88,7 +88,6 @@ static int loot_rr(struct kr_cache *cache, knot_pkt_t *pkt, const knot_dname_t *
return kr_error(ENOENT);
}
/* Mark as expiring if it has less than 1% TTL (or less than 5s) */
if (is_expiring(&cache_rr, drift)) {
qry->flags |= QUERY_EXPIRING;
}
......@@ -301,7 +300,19 @@ static int commit_rr(const char *key, void *val, void *data)
* This way they would have to hit the first answer (whenever TTL expires). */
if (cached_rank >= 0) {
VERBOSE_MSG(baton->qry, "=> orig. rank: 0%0.2o\n", cached_rank);
if (cached_rank >= rank) {
bool accept = rank > cached_rank;
/* Additionally accept equal rank if the cached RR is expiring.
* This is primarily for prefetching from predict module. */
if (rank == cached_rank) {
uint32_t drift = baton->timestamp;
knot_rrset_t cache_rr;
knot_rrset_init(&cache_rr, rr->owner, rr->type, rr->rclass);
int ret = kr_cache_peek_rr(baton->cache, &cache_rr, NULL, NULL, &drift);
if (ret != kr_ok() || is_expiring(&cache_rr, drift)) {
accept = true;
}
}
if (!accept) {
return kr_ok();
}
}
......
......@@ -3,7 +3,7 @@
Prefetching records
-------------------
The module tracks expiring records (having less than 5% of original TTL) and batches them for predict.
The module refreshes records that are about to expire when they're used (having less than 1% of original TTL).
This improves latency for frequently used records, as they are fetched in advance.
It is also able to learn usage patterns and repetitive queries that the server makes. For example, if
......@@ -29,6 +29,7 @@ Example configuration
Defaults are 15 minutes window, 6 hours period.
.. tip:: Use period 0 to turn off prediction and just do prefetching of expiring records.
That works even without the 'stats' module.
Exported metrics
^^^^^^^^^^^^^^^^
......@@ -47,4 +48,4 @@ Properties
Reconfigure the predictor to given tracking window and period length. Both parameters are optional.
Window length is in minutes, period is a number of windows that can be kept in memory.
e.g. if a ``window`` is 15 minutes, a ``period`` of "24" means 6 hours.
\ No newline at end of file
e.g. if a ``window`` is 15 minutes, a ``period`` of "24" means 6 hours.
......@@ -31,7 +31,7 @@ function predict.drain(ev)
local deleted = 0
for key, val in pairs(predict.queue) do
local qtype, qname = key:match('(%S*)%s(.*)')
worker.resolve(qname, kres.type[qtype], 1, kres.query.NO_CACHE)
worker.resolve(qname, kres.type[qtype], kres.class.IN, kres.query.NO_CACHE)
predict.queue[key] = nil
deleted = deleted + 1
if deleted >= predict.batch then
......@@ -76,13 +76,6 @@ local function enqueue_from_log(current)
return queued
end
-- Prefetch soon-to-expire records
function predict.prefetch()
local queries = stats.expiring()
stats.clear_expiring()
return enqueue(queries)
end
-- Sample current epoch, return number of sampled queries
function predict.sample(epoch_now)
if not epoch_now then return 0, 0 end
......@@ -119,7 +112,9 @@ local function generate(epoch_now)
end
function predict.process(ev)
if not stats then error("'stats' module required") end
if (predict.period or 0) ~= 0 and not stats then
error("'stats' module required")
end
-- Start a new epoch, or continue sampling
predict.ev_sample = nil
local epoch_now = current_epoch()
......@@ -140,8 +135,6 @@ function predict.process(ev)
-- Sample current epoch
local nr_learned = predict.sample(epoch_now)
-- Prefetch expiring records
nr_queued = nr_queued + predict.prefetch()
-- Dispatch predicted queries
if nr_queued > 0 then
predict.queue_len = predict.queue_len + nr_queued
......@@ -151,8 +144,10 @@ function predict.process(ev)
end
end
predict.ev_sample = event.after(next_event(), predict.process)
stats['predict.queue'] = predict.queue_len
stats['predict.learned'] = nr_learned
if stats then
stats['predict.queue'] = predict.queue_len
stats['predict.learned'] = nr_learned
end
collectgarbage()
end
......@@ -184,4 +179,19 @@ function predict.config(config)
predict.init()
end
predict.layer = {
-- Prefetch all expiring (sub-)queries immediately after the request finishes.
-- Doing that immediately is simplest and avoids creating (new) large bursts of activity.
finish = function (state, req)
req = kres.request_t(req)
local qrys = req.rplan.resolved
for i = 0, (tonumber(qrys.len) - 1) do -- size_t doesn't work for some reason
local qry = qrys.at[i]
if bit.band(qry.flags, kres.query.EXPIRING) ~= 0 then
worker.resolve(kres.dname2str(qry.sname), qry.stype, qry.sclass, kres.query.NO_CACHE)
end
end
end
}
return predict
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment