2 This file is part of catta.
4 catta is free software; you can redistribute it and/or modify it
5 under the terms of the GNU Lesser General Public License as
6 published by the Free Software Foundation; either version 2.1 of the
7 License, or (at your option) any later version.
9 catta is distributed in the hope that it will be useful, but WITHOUT
10 ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
11 or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General
12 Public License for more details.
14 You should have received a copy of the GNU Lesser General Public
15 License along with catta; if not, write to the Free Software
16 Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
26 #include <catta/timeval.h>
27 #include <catta/malloc.h>
29 #include "response-sched.h"
30 #include <catta/log.h>
33 /* Local packets are supressed this long after sending them */
34 #define CATTA_RESPONSE_HISTORY_MSEC 500
36 /* Local packets are deferred this long before sending them */
37 #define CATTA_RESPONSE_DEFER_MSEC 20
39 /* Additional jitter for deferred packets */
40 #define CATTA_RESPONSE_JITTER_MSEC 100
42 /* Remote packets can suppress local traffic as long as this value */
43 #define CATTA_RESPONSE_SUPPRESS_MSEC 700
45 typedef struct CattaResponseJob CattaResponseJob;
51 } CattaResponseJobState;
53 struct CattaResponseJob {
54 CattaResponseScheduler *scheduler;
55 CattaTimeEvent *time_event;
57 CattaResponseJobState state;
58 struct timeval delivery;
65 CATTA_LLIST_FIELDS(CattaResponseJob, jobs);
68 struct CattaResponseScheduler {
69 CattaInterface *iface;
70 CattaTimeEventQueue *time_event_queue;
72 CATTA_LLIST_HEAD(CattaResponseJob, jobs);
73 CATTA_LLIST_HEAD(CattaResponseJob, history);
74 CATTA_LLIST_HEAD(CattaResponseJob, suppressed);
77 static CattaResponseJob* job_new(CattaResponseScheduler *s, CattaRecord *record, CattaResponseJobState state) {
83 if (!(rj = catta_new(CattaResponseJob, 1))) {
84 catta_log_error(__FILE__": Out of memory");
89 rj->record = catta_record_ref(record);
90 rj->time_event = NULL;
92 rj->querier_valid = 0;
94 if ((rj->state = state) == CATTA_SCHEDULED)
95 CATTA_LLIST_PREPEND(CattaResponseJob, jobs, s->jobs, rj);
96 else if (rj->state == CATTA_DONE)
97 CATTA_LLIST_PREPEND(CattaResponseJob, jobs, s->history, rj);
98 else /* rj->state == CATTA_SUPPRESSED */
99 CATTA_LLIST_PREPEND(CattaResponseJob, jobs, s->suppressed, rj);
104 static void job_free(CattaResponseScheduler *s, CattaResponseJob *rj) {
109 catta_time_event_free(rj->time_event);
111 if (rj->state == CATTA_SCHEDULED)
112 CATTA_LLIST_REMOVE(CattaResponseJob, jobs, s->jobs, rj);
113 else if (rj->state == CATTA_DONE)
114 CATTA_LLIST_REMOVE(CattaResponseJob, jobs, s->history, rj);
115 else /* rj->state == CATTA_SUPPRESSED */
116 CATTA_LLIST_REMOVE(CattaResponseJob, jobs, s->suppressed, rj);
118 catta_record_unref(rj->record);
122 static void elapse_callback(CattaTimeEvent *e, void* data);
124 static void job_set_elapse_time(CattaResponseScheduler *s, CattaResponseJob *rj, unsigned msec, unsigned jitter) {
130 catta_elapse_time(&tv, msec, jitter);
133 catta_time_event_update(rj->time_event, &tv);
135 rj->time_event = catta_time_event_new(s->time_event_queue, &tv, elapse_callback, rj);
138 static void job_mark_done(CattaResponseScheduler *s, CattaResponseJob *rj) {
142 assert(rj->state == CATTA_SCHEDULED);
144 CATTA_LLIST_REMOVE(CattaResponseJob, jobs, s->jobs, rj);
145 CATTA_LLIST_PREPEND(CattaResponseJob, jobs, s->history, rj);
147 rj->state = CATTA_DONE;
149 job_set_elapse_time(s, rj, CATTA_RESPONSE_HISTORY_MSEC, 0);
151 gettimeofday(&rj->delivery, NULL);
154 CattaResponseScheduler *catta_response_scheduler_new(CattaInterface *i) {
155 CattaResponseScheduler *s;
158 if (!(s = catta_new(CattaResponseScheduler, 1))) {
159 catta_log_error(__FILE__": Out of memory");
164 s->time_event_queue = i->monitor->server->time_event_queue;
166 CATTA_LLIST_HEAD_INIT(CattaResponseJob, s->jobs);
167 CATTA_LLIST_HEAD_INIT(CattaResponseJob, s->history);
168 CATTA_LLIST_HEAD_INIT(CattaResponseJob, s->suppressed);
173 void catta_response_scheduler_free(CattaResponseScheduler *s) {
176 catta_response_scheduler_clear(s);
180 void catta_response_scheduler_clear(CattaResponseScheduler *s) {
184 job_free(s, s->jobs);
186 job_free(s, s->history);
187 while (s->suppressed)
188 job_free(s, s->suppressed);
191 static void enumerate_aux_records_callback(CATTA_GCC_UNUSED CattaServer *s, CattaRecord *r, int flush_cache, void* userdata) {
192 CattaResponseJob *rj = userdata;
197 catta_response_scheduler_post(rj->scheduler, r, flush_cache, rj->querier_valid ? &rj->querier : NULL, 0);
200 static int packet_add_response_job(CattaResponseScheduler *s, CattaDnsPacket *p, CattaResponseJob *rj) {
205 /* Try to add this record to the packet */
206 if (!catta_dns_packet_append_record(p, rj->record, rj->flush_cache, 0))
209 /* Ok, this record will definitely be sent, so schedule the
210 * auxilliary packets, too */
211 catta_server_enumerate_aux_records(s->iface->monitor->server, s->iface, rj->record, enumerate_aux_records_callback, rj);
212 job_mark_done(s, rj);
217 static void send_response_packet(CattaResponseScheduler *s, CattaResponseJob *rj) {
224 if (!(p = catta_dns_packet_new_response(s->iface->hardware->mtu, 1)))
228 /* Put it in the packet. */
229 if (packet_add_response_job(s, p, rj)) {
231 /* Try to fill up packet with more responses, if available */
234 if (!packet_add_response_job(s, p, s->jobs))
243 catta_dns_packet_free(p);
245 /* OK, the packet was too small, so create one that fits */
246 size = catta_record_get_estimate_size(rj->record) + CATTA_DNS_PACKET_HEADER_SIZE;
248 if (!(p = catta_dns_packet_new_response(size + CATTA_DNS_PACKET_EXTRA_SIZE, 1)))
251 if (!packet_add_response_job(s, p, rj)) {
252 catta_dns_packet_free(p);
254 catta_log_warn("Record too large, cannot send");
255 job_mark_done(s, rj);
260 catta_dns_packet_set_field(p, CATTA_DNS_FIELD_ANCOUNT, n);
261 catta_interface_send_packet(s->iface, p);
262 catta_dns_packet_free(p);
265 static void elapse_callback(CATTA_GCC_UNUSED CattaTimeEvent *e, void* data) {
266 CattaResponseJob *rj = data;
270 if (rj->state == CATTA_DONE || rj->state == CATTA_SUPPRESSED)
271 job_free(rj->scheduler, rj); /* Lets drop this entry */
273 send_response_packet(rj->scheduler, rj);
276 static CattaResponseJob* find_scheduled_job(CattaResponseScheduler *s, CattaRecord *record) {
277 CattaResponseJob *rj;
282 for (rj = s->jobs; rj; rj = rj->jobs_next) {
283 assert(rj->state == CATTA_SCHEDULED);
285 if (catta_record_equal_no_ttl(rj->record, record))
292 static CattaResponseJob* find_history_job(CattaResponseScheduler *s, CattaRecord *record) {
293 CattaResponseJob *rj;
298 for (rj = s->history; rj; rj = rj->jobs_next) {
299 assert(rj->state == CATTA_DONE);
301 if (catta_record_equal_no_ttl(rj->record, record)) {
302 /* Check whether this entry is outdated */
304 /* catta_log_debug("history age: %u", (unsigned) (catta_age(&rj->delivery)/1000)); */
306 if (catta_age(&rj->delivery)/1000 > CATTA_RESPONSE_HISTORY_MSEC) {
307 /* it is outdated, so let's remove it */
319 static CattaResponseJob* find_suppressed_job(CattaResponseScheduler *s, CattaRecord *record, const CattaAddress *querier) {
320 CattaResponseJob *rj;
326 for (rj = s->suppressed; rj; rj = rj->jobs_next) {
327 assert(rj->state == CATTA_SUPPRESSED);
328 assert(rj->querier_valid);
330 if (catta_record_equal_no_ttl(rj->record, record) &&
331 catta_address_cmp(&rj->querier, querier) == 0) {
332 /* Check whether this entry is outdated */
334 if (catta_age(&rj->delivery) > CATTA_RESPONSE_SUPPRESS_MSEC*1000) {
335 /* it is outdated, so let's remove it */
347 int catta_response_scheduler_post(CattaResponseScheduler *s, CattaRecord *record, int flush_cache, const CattaAddress *querier, int immediately) {
348 CattaResponseJob *rj;
355 assert(!catta_key_is_pattern(record->key));
357 /* t = catta_record_to_string(record); */
358 /* catta_log_debug("post %i %s", immediately, t); */
361 /* Check whether this response is suppressed */
363 (rj = find_suppressed_job(s, record, querier)) &&
364 catta_record_is_goodbye(record) == catta_record_is_goodbye(rj->record) &&
365 rj->record->ttl >= record->ttl/2) {
367 /* catta_log_debug("Response suppressed by known answer suppression."); */
371 /* Check if we already sent this response recently */
372 if ((rj = find_history_job(s, record))) {
374 if (catta_record_is_goodbye(record) == catta_record_is_goodbye(rj->record) &&
375 rj->record->ttl >= record->ttl/2 &&
376 (rj->flush_cache || !flush_cache)) {
377 /* catta_log_debug("Response suppressed by local duplicate suppression (history)"); */
385 catta_elapse_time(&tv, immediately ? 0 : CATTA_RESPONSE_DEFER_MSEC, immediately ? 0 : CATTA_RESPONSE_JITTER_MSEC);
387 if ((rj = find_scheduled_job(s, record))) {
388 /* catta_log_debug("Response suppressed by local duplicate suppression (scheduled)"); */
390 /* Update a little ... */
392 /* Update the time if the new is prior to the old */
393 if (catta_timeval_compare(&tv, &rj->delivery) < 0) {
395 catta_time_event_update(rj->time_event, &rj->delivery);
398 /* Update the flush cache bit */
402 /* Update the querier field */
403 if (!querier || (rj->querier_valid && catta_address_cmp(querier, &rj->querier) != 0))
404 rj->querier_valid = 0;
406 /* Update record data (just for the TTL) */
407 catta_record_unref(rj->record);
408 rj->record = catta_record_ref(record);
412 /* catta_log_debug("Accepted new response job."); */
414 /* Create a new job and schedule it */
415 if (!(rj = job_new(s, record, CATTA_SCHEDULED)))
419 rj->time_event = catta_time_event_new(s->time_event_queue, &rj->delivery, elapse_callback, rj);
420 rj->flush_cache = flush_cache;
422 if ((rj->querier_valid = !!querier))
423 rj->querier = *querier;
429 void catta_response_scheduler_incoming(CattaResponseScheduler *s, CattaRecord *record, int flush_cache) {
430 CattaResponseJob *rj;
433 /* This function is called whenever an incoming response was
434 * receieved. We drop scheduled responses which match here. The
435 * keyword is "DUPLICATE ANSWER SUPPRESION". */
437 if ((rj = find_scheduled_job(s, record))) {
439 if ((!rj->flush_cache || flush_cache) && /* flush cache bit was set correctly */
440 catta_record_is_goodbye(record) == catta_record_is_goodbye(rj->record) && /* both goodbye packets, or both not */
441 record->ttl >= rj->record->ttl/2) { /* sensible TTL */
443 /* A matching entry was found, so let's mark it done */
444 /* catta_log_debug("Response suppressed by distributed duplicate suppression"); */
445 job_mark_done(s, rj);
451 if ((rj = find_history_job(s, record))) {
452 /* Found a history job, let's update it */
453 catta_record_unref(rj->record);
454 rj->record = catta_record_ref(record);
456 /* Found no existing history job, so let's create a new one */
457 if (!(rj = job_new(s, record, CATTA_DONE)))
460 rj->flush_cache = flush_cache;
461 rj->querier_valid = 0;
463 gettimeofday(&rj->delivery, NULL);
464 job_set_elapse_time(s, rj, CATTA_RESPONSE_HISTORY_MSEC, 0);
467 void catta_response_scheduler_suppress(CattaResponseScheduler *s, CattaRecord *record, const CattaAddress *querier) {
468 CattaResponseJob *rj;
474 if ((rj = find_scheduled_job(s, record))) {
476 if (rj->querier_valid && catta_address_cmp(querier, &rj->querier) == 0 && /* same originator */
477 catta_record_is_goodbye(record) == catta_record_is_goodbye(rj->record) && /* both goodbye packets, or both not */
478 record->ttl >= rj->record->ttl/2) { /* sensible TTL */
480 /* A matching entry was found, so let's drop it */
481 /* catta_log_debug("Known answer suppression active!"); */
486 if ((rj = find_suppressed_job(s, record, querier))) {
488 /* Let's update the old entry */
489 catta_record_unref(rj->record);
490 rj->record = catta_record_ref(record);
494 /* Create a new entry */
495 if (!(rj = job_new(s, record, CATTA_SUPPRESSED)))
497 rj->querier_valid = 1;
498 rj->querier = *querier;
501 gettimeofday(&rj->delivery, NULL);
502 job_set_elapse_time(s, rj, CATTA_RESPONSE_SUPPRESS_MSEC, 0);
505 void catta_response_scheduler_force(CattaResponseScheduler *s) {
508 /* Send all scheduled responses immediately */
510 send_response_packet(s, s->jobs);