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/domain.h>
27 #include <catta/timeval.h>
28 #include <catta/malloc.h>
30 #include "probe-sched.h"
31 #include <catta/log.h>
34 #define CATTA_PROBE_HISTORY_MSEC 150
35 #define CATTA_PROBE_DEFER_MSEC 50
37 typedef struct CattaProbeJob CattaProbeJob;
39 struct CattaProbeJob {
40 CattaProbeScheduler *scheduler;
41 CattaTimeEvent *time_event;
43 int chosen; /* Use for packet assembling */
45 struct timeval delivery;
49 CATTA_LLIST_FIELDS(CattaProbeJob, jobs);
52 struct CattaProbeScheduler {
53 CattaInterface *iface;
54 CattaTimeEventQueue *time_event_queue;
56 CATTA_LLIST_HEAD(CattaProbeJob, jobs);
57 CATTA_LLIST_HEAD(CattaProbeJob, history);
60 static CattaProbeJob* job_new(CattaProbeScheduler *s, CattaRecord *record, int done) {
66 if (!(pj = catta_new(CattaProbeJob, 1))) {
67 catta_log_error(__FILE__": Out of memory");
68 return NULL; /* OOM */
72 pj->record = catta_record_ref(record);
73 pj->time_event = NULL;
76 if ((pj->done = done))
77 CATTA_LLIST_PREPEND(CattaProbeJob, jobs, s->history, pj);
79 CATTA_LLIST_PREPEND(CattaProbeJob, jobs, s->jobs, pj);
84 static void job_free(CattaProbeScheduler *s, CattaProbeJob *pj) {
88 catta_time_event_free(pj->time_event);
91 CATTA_LLIST_REMOVE(CattaProbeJob, jobs, s->history, pj);
93 CATTA_LLIST_REMOVE(CattaProbeJob, jobs, s->jobs, pj);
95 catta_record_unref(pj->record);
99 static void elapse_callback(CattaTimeEvent *e, void* data);
101 static void job_set_elapse_time(CattaProbeScheduler *s, CattaProbeJob *pj, unsigned msec, unsigned jitter) {
107 catta_elapse_time(&tv, msec, jitter);
110 catta_time_event_update(pj->time_event, &tv);
112 pj->time_event = catta_time_event_new(s->time_event_queue, &tv, elapse_callback, pj);
115 static void job_mark_done(CattaProbeScheduler *s, CattaProbeJob *pj) {
121 CATTA_LLIST_REMOVE(CattaProbeJob, jobs, s->jobs, pj);
122 CATTA_LLIST_PREPEND(CattaProbeJob, jobs, s->history, pj);
126 job_set_elapse_time(s, pj, CATTA_PROBE_HISTORY_MSEC, 0);
127 gettimeofday(&pj->delivery, NULL);
130 CattaProbeScheduler *catta_probe_scheduler_new(CattaInterface *i) {
131 CattaProbeScheduler *s;
135 if (!(s = catta_new(CattaProbeScheduler, 1))) {
136 catta_log_error(__FILE__": Out of memory");
141 s->time_event_queue = i->monitor->server->time_event_queue;
143 CATTA_LLIST_HEAD_INIT(CattaProbeJob, s->jobs);
144 CATTA_LLIST_HEAD_INIT(CattaProbeJob, s->history);
149 void catta_probe_scheduler_free(CattaProbeScheduler *s) {
152 catta_probe_scheduler_clear(s);
156 void catta_probe_scheduler_clear(CattaProbeScheduler *s) {
160 job_free(s, s->jobs);
162 job_free(s, s->history);
165 static int packet_add_probe_query(CattaProbeScheduler *s, CattaDnsPacket *p, CattaProbeJob *pj) {
176 /* Estimate the size for this record */
178 catta_key_get_estimate_size(pj->record->key) +
179 catta_record_get_estimate_size(pj->record);
182 if (size > catta_dns_packet_space(p))
185 /* Create the probe query */
186 if (!(k = catta_key_new(pj->record->key->name, pj->record->key->clazz, CATTA_DNS_TYPE_ANY)))
189 b = !!catta_dns_packet_append_key(p, k, 0);
192 /* Mark this job for addition to the packet */
195 /* Scan for more jobs whith matching key pattern */
196 for (pj = s->jobs; pj; pj = pj->jobs_next) {
200 /* Does the record match the probe? */
201 if (k->clazz != pj->record->key->clazz || !catta_domain_equal(k->name, pj->record->key->name))
204 /* This job wouldn't fit in */
205 if (catta_record_get_estimate_size(pj->record) > catta_dns_packet_space(p))
208 /* Mark this job for addition to the packet */
217 static void elapse_callback(CATTA_GCC_UNUSED CattaTimeEvent *e, void* data) {
218 CattaProbeJob *pj = data, *next;
219 CattaProbeScheduler *s;
227 /* Lets remove it from the history */
232 if (!(p = catta_dns_packet_new_query(s->iface->hardware->mtu)))
236 /* Add the import probe */
237 if (!packet_add_probe_query(s, p, pj)) {
242 catta_dns_packet_free(p);
244 /* The probe didn't fit in the package, so let's allocate a larger one */
247 catta_key_get_estimate_size(pj->record->key) +
248 catta_record_get_estimate_size(pj->record) +
249 CATTA_DNS_PACKET_HEADER_SIZE;
251 if (!(p = catta_dns_packet_new_query(size + CATTA_DNS_PACKET_EXTRA_SIZE)))
254 if (!(k = catta_key_new(pj->record->key->name, pj->record->key->clazz, CATTA_DNS_TYPE_ANY))) {
255 catta_dns_packet_free(p);
259 b = catta_dns_packet_append_key(p, k, 0) && catta_dns_packet_append_record(p, pj->record, 0, 0);
263 catta_dns_packet_set_field(p, CATTA_DNS_FIELD_NSCOUNT, 1);
264 catta_dns_packet_set_field(p, CATTA_DNS_FIELD_QDCOUNT, 1);
265 catta_interface_send_packet(s->iface, p);
267 catta_log_warn("Probe record too large, cannot send");
269 catta_dns_packet_free(p);
270 job_mark_done(s, pj);
275 /* Try to fill up packet with more probes, if available */
276 for (pj = s->jobs; pj; pj = pj->jobs_next) {
281 if (!packet_add_probe_query(s, p, pj))
287 catta_dns_packet_set_field(p, CATTA_DNS_FIELD_QDCOUNT, n);
291 /* Now add the chosen records to the authorative section */
292 for (pj = s->jobs; pj; pj = next) {
294 next = pj->jobs_next;
299 if (!catta_dns_packet_append_record(p, pj->record, 0, 0)) {
300 /* catta_log_warn("Bad probe size estimate!"); */
302 /* Unmark all following jobs */
303 for (; pj; pj = pj->jobs_next)
309 job_mark_done(s, pj);
314 catta_dns_packet_set_field(p, CATTA_DNS_FIELD_NSCOUNT, n);
317 catta_interface_send_packet(s->iface, p);
318 catta_dns_packet_free(p);
321 static CattaProbeJob* find_scheduled_job(CattaProbeScheduler *s, CattaRecord *record) {
327 for (pj = s->jobs; pj; pj = pj->jobs_next) {
330 if (catta_record_equal_no_ttl(pj->record, record))
337 static CattaProbeJob* find_history_job(CattaProbeScheduler *s, CattaRecord *record) {
343 for (pj = s->history; pj; pj = pj->jobs_next) {
346 if (catta_record_equal_no_ttl(pj->record, record)) {
347 /* Check whether this entry is outdated */
349 if (catta_age(&pj->delivery) > CATTA_PROBE_HISTORY_MSEC*1000) {
350 /* it is outdated, so let's remove it */
362 int catta_probe_scheduler_post(CattaProbeScheduler *s, CattaRecord *record, int immediately) {
368 assert(!catta_key_is_pattern(record->key));
370 if ((pj = find_history_job(s, record)))
373 catta_elapse_time(&tv, immediately ? 0 : CATTA_PROBE_DEFER_MSEC, 0);
375 if ((pj = find_scheduled_job(s, record))) {
377 if (catta_timeval_compare(&tv, &pj->delivery) < 0) {
378 /* If the new entry should be scheduled earlier, update the old entry */
380 catta_time_event_update(pj->time_event, &pj->delivery);
385 /* Create a new job and schedule it */
386 if (!(pj = job_new(s, record, 0)))
390 pj->time_event = catta_time_event_new(s->time_event_queue, &pj->delivery, elapse_callback, pj);
393 /* catta_log_debug("Accepted new probe job."); */