git.meshlink.io Git - catta/blob - src/query-sched.c

   1 /***
   2   This file is part of catta.
   3
   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.
   8
   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.
  13
  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
  17   USA.
  18 ***/
  19
  20 #ifdef HAVE_CONFIG_H
  21 #include <config.h>
  22 #endif
  23
  24 #include <stdlib.h>
  25
  26 #include <catta/timeval.h>
  27 #include <catta/malloc.h>
  28
  29 #include "query-sched.h"
  30 #include <catta/log.h>
  31
  32 #define CATTA_QUERY_HISTORY_MSEC 100
  33 #define CATTA_QUERY_DEFER_MSEC 100
  34
  35 typedef struct CattaQueryJob CattaQueryJob;
  36 typedef struct CattaKnownAnswer CattaKnownAnswer;
  37
  38 struct CattaQueryJob {
  39     unsigned id;
  40     int n_posted;
  41
  42     CattaQueryScheduler *scheduler;
  43     CattaTimeEvent *time_event;
  44
  45     int done;
  46     struct timeval delivery;
  47
  48     CattaKey *key;
  49
  50     /* Jobs are stored in a simple linked list. It might turn out in
  51      * the future that this list grows too long and we must switch to
  52      * some other kind of data structure. This needs further
  53      * investigation. I expect the list to be very short (< 20
  54      * entries) most of the time, but this might be a wrong
  55      * assumption, especially on setups where traffic reflection is
  56      * involved. */
  57
  58     CATTA_LLIST_FIELDS(CattaQueryJob, jobs);
  59 };
  60
  61 struct CattaKnownAnswer {
  62     CattaQueryScheduler *scheduler;
  63     CattaRecord *record;
  64
  65     CATTA_LLIST_FIELDS(CattaKnownAnswer, known_answer);
  66 };
  67
  68 struct CattaQueryScheduler {
  69     CattaInterface *interface;
  70     CattaTimeEventQueue *time_event_queue;
  71
  72     unsigned next_id;
  73
  74     CATTA_LLIST_HEAD(CattaQueryJob, jobs);
  75     CATTA_LLIST_HEAD(CattaQueryJob, history);
  76     CATTA_LLIST_HEAD(CattaKnownAnswer, known_answers);
  77 };
  78
  79 static CattaQueryJob* job_new(CattaQueryScheduler *s, CattaKey *key, int done) {
  80     CattaQueryJob *qj;
  81
  82     assert(s);
  83     assert(key);
  84
  85     if (!(qj = catta_new(CattaQueryJob, 1))) {
  86         catta_log_error(__FILE__": Out of memory");
  87         return NULL;
  88     }
  89
  90     qj->scheduler = s;
  91     qj->key = catta_key_ref(key);
  92     qj->time_event = NULL;
  93     qj->n_posted = 1;
  94     qj->id = s->next_id++;
  95
  96     if ((qj->done = done))
  97         CATTA_LLIST_PREPEND(CattaQueryJob, jobs, s->history, qj);
  98     else
  99         CATTA_LLIST_PREPEND(CattaQueryJob, jobs, s->jobs, qj);
 100
 101     return qj;
 102 }
 103
 104 static void job_free(CattaQueryScheduler *s, CattaQueryJob *qj) {
 105     assert(s);
 106     assert(qj);
 107
 108     if (qj->time_event)
 109         catta_time_event_free(qj->time_event);
 110
 111     if (qj->done)
 112         CATTA_LLIST_REMOVE(CattaQueryJob, jobs, s->history, qj);
 113     else
 114         CATTA_LLIST_REMOVE(CattaQueryJob, jobs, s->jobs, qj);
 115
 116     catta_key_unref(qj->key);
 117     catta_free(qj);
 118 }
 119
 120 static void elapse_callback(CattaTimeEvent *e, void* data);
 121
 122 static void job_set_elapse_time(CattaQueryScheduler *s, CattaQueryJob *qj, unsigned msec, unsigned jitter) {
 123     struct timeval tv;
 124
 125     assert(s);
 126     assert(qj);
 127
 128     catta_elapse_time(&tv, msec, jitter);
 129
 130     if (qj->time_event)
 131         catta_time_event_update(qj->time_event, &tv);
 132     else
 133         qj->time_event = catta_time_event_new(s->time_event_queue, &tv, elapse_callback, qj);
 134 }
 135
 136 static void job_mark_done(CattaQueryScheduler *s, CattaQueryJob *qj) {
 137     assert(s);
 138     assert(qj);
 139
 140     assert(!qj->done);
 141
 142     CATTA_LLIST_REMOVE(CattaQueryJob, jobs, s->jobs, qj);
 143     CATTA_LLIST_PREPEND(CattaQueryJob, jobs, s->history, qj);
 144
 145     qj->done = 1;
 146
 147     job_set_elapse_time(s, qj, CATTA_QUERY_HISTORY_MSEC, 0);
 148     gettimeofday(&qj->delivery, NULL);
 149 }
 150
 151 CattaQueryScheduler *catta_query_scheduler_new(CattaInterface *i) {
 152     CattaQueryScheduler *s;
 153     assert(i);
 154
 155     if (!(s = catta_new(CattaQueryScheduler, 1))) {
 156         catta_log_error(__FILE__": Out of memory");
 157         return NULL; /* OOM */
 158     }
 159
 160     s->interface = i;
 161     s->time_event_queue = i->monitor->server->time_event_queue;
 162     s->next_id = 0;
 163
 164     CATTA_LLIST_HEAD_INIT(CattaQueryJob, s->jobs);
 165     CATTA_LLIST_HEAD_INIT(CattaQueryJob, s->history);
 166     CATTA_LLIST_HEAD_INIT(CattaKnownAnswer, s->known_answers);
 167
 168     return s;
 169 }
 170
 171 void catta_query_scheduler_free(CattaQueryScheduler *s) {
 172     assert(s);
 173
 174     assert(!s->known_answers);
 175     catta_query_scheduler_clear(s);
 176     catta_free(s);
 177 }
 178
 179 void catta_query_scheduler_clear(CattaQueryScheduler *s) {
 180     assert(s);
 181
 182     while (s->jobs)
 183         job_free(s, s->jobs);
 184     while (s->history)
 185         job_free(s, s->history);
 186 }
 187
 188 static void* known_answer_walk_callback(CattaCache *c, CattaKey *pattern, CattaCacheEntry *e, void* userdata) {
 189     CattaQueryScheduler *s = userdata;
 190     CattaKnownAnswer *ka;
 191
 192     assert(c);
 193     assert(pattern);
 194     assert(e);
 195     assert(s);
 196
 197     if (catta_cache_entry_half_ttl(c, e))
 198         return NULL;
 199
 200     if (!(ka = catta_new0(CattaKnownAnswer, 1))) {
 201         catta_log_error(__FILE__": Out of memory");
 202         return NULL;
 203     }
 204
 205     ka->scheduler = s;
 206     ka->record = catta_record_ref(e->record);
 207
 208     CATTA_LLIST_PREPEND(CattaKnownAnswer, known_answer, s->known_answers, ka);
 209     return NULL;
 210 }
 211
 212 static int packet_add_query_job(CattaQueryScheduler *s, CattaDnsPacket *p, CattaQueryJob *qj) {
 213     assert(s);
 214     assert(p);
 215     assert(qj);
 216
 217     if (!catta_dns_packet_append_key(p, qj->key, 0))
 218         return 0;
 219
 220     /* Add all matching known answers to the list */
 221     catta_cache_walk(s->interface->cache, qj->key, known_answer_walk_callback, s);
 222
 223     job_mark_done(s, qj);
 224
 225     return 1;
 226 }
 227
 228 static void append_known_answers_and_send(CattaQueryScheduler *s, CattaDnsPacket *p) {
 229     CattaKnownAnswer *ka;
 230     unsigned n;
 231     assert(s);
 232     assert(p);
 233
 234     n = 0;
 235
 236     while ((ka = s->known_answers)) {
 237         int too_large = 0;
 238
 239         while (!catta_dns_packet_append_record(p, ka->record, 0, 0)) {
 240
 241             if (catta_dns_packet_is_empty(p)) {
 242                 /* The record is too large to fit into one packet, so
 243                    there's no point in sending it. Better is letting
 244                    the owner of the record send it as a response. This
 245                    has the advantage of a cache refresh. */
 246
 247                 too_large = 1;
 248                 break;
 249             }
 250
 251             catta_dns_packet_set_field(p, CATTA_DNS_FIELD_FLAGS, catta_dns_packet_get_field(p, CATTA_DNS_FIELD_FLAGS) | CATTA_DNS_FLAG_TC);
 252             catta_dns_packet_set_field(p, CATTA_DNS_FIELD_ANCOUNT, n);
 253             catta_interface_send_packet(s->interface, p);
 254             catta_dns_packet_free(p);
 255
 256             p = catta_dns_packet_new_query(s->interface->hardware->mtu);
 257             n = 0;
 258         }
 259
 260         CATTA_LLIST_REMOVE(CattaKnownAnswer, known_answer, s->known_answers, ka);
 261         catta_record_unref(ka->record);
 262         catta_free(ka);
 263
 264         if (!too_large)
 265             n++;
 266     }
 267
 268     catta_dns_packet_set_field(p, CATTA_DNS_FIELD_ANCOUNT, n);
 269     catta_interface_send_packet(s->interface, p);
 270     catta_dns_packet_free(p);
 271 }
 272
 273 static void elapse_callback(CATTA_GCC_UNUSED CattaTimeEvent *e, void* data) {
 274     CattaQueryJob *qj = data;
 275     CattaQueryScheduler *s;
 276     CattaDnsPacket *p;
 277     unsigned n;
 278     int b;
 279
 280     assert(qj);
 281     s = qj->scheduler;
 282
 283     if (qj->done) {
 284         /* Lets remove it  from the history */
 285         job_free(s, qj);
 286         return;
 287     }
 288
 289     assert(!s->known_answers);
 290
 291     if (!(p = catta_dns_packet_new_query(s->interface->hardware->mtu)))
 292         return; /* OOM */
 293
 294     b = packet_add_query_job(s, p, qj);
 295     assert(b); /* An query must always fit in */
 296     n = 1;
 297
 298     /* Try to fill up packet with more queries, if available */
 299     while (s->jobs) {
 300
 301         if (!packet_add_query_job(s, p, s->jobs))
 302             break;
 303
 304         n++;
 305     }
 306
 307     catta_dns_packet_set_field(p, CATTA_DNS_FIELD_QDCOUNT, n);
 308
 309     /* Now add known answers */
 310     append_known_answers_and_send(s, p);
 311 }
 312
 313 static CattaQueryJob* find_scheduled_job(CattaQueryScheduler *s, CattaKey *key) {
 314     CattaQueryJob *qj;
 315
 316     assert(s);
 317     assert(key);
 318
 319     for (qj = s->jobs; qj; qj = qj->jobs_next) {
 320         assert(!qj->done);
 321
 322         if (catta_key_equal(qj->key, key))
 323             return qj;
 324     }
 325
 326     return NULL;
 327 }
 328
 329 static CattaQueryJob* find_history_job(CattaQueryScheduler *s, CattaKey *key) {
 330     CattaQueryJob *qj;
 331
 332     assert(s);
 333     assert(key);
 334
 335     for (qj = s->history; qj; qj = qj->jobs_next) {
 336         assert(qj->done);
 337
 338         if (catta_key_equal(qj->key, key)) {
 339             /* Check whether this entry is outdated */
 340
 341             if (catta_age(&qj->delivery) > CATTA_QUERY_HISTORY_MSEC*1000) {
 342                 /* it is outdated, so let's remove it */
 343                 job_free(s, qj);
 344                 return NULL;
 345             }
 346
 347             return qj;
 348         }
 349     }
 350
 351     return NULL;
 352 }
 353
 354 int catta_query_scheduler_post(CattaQueryScheduler *s, CattaKey *key, int immediately, unsigned *ret_id) {
 355     struct timeval tv;
 356     CattaQueryJob *qj;
 357
 358     assert(s);
 359     assert(key);
 360
 361     if ((qj = find_history_job(s, key)))
 362         return 0;
 363
 364     catta_elapse_time(&tv, immediately ? 0 : CATTA_QUERY_DEFER_MSEC, 0);
 365
 366     if ((qj = find_scheduled_job(s, key))) {
 367         /* Duplicate questions suppression */
 368
 369         if (catta_timeval_compare(&tv, &qj->delivery) < 0) {
 370             /* If the new entry should be scheduled earlier,
 371              * update the old entry */
 372             qj->delivery = tv;
 373             catta_time_event_update(qj->time_event, &qj->delivery);
 374         }
 375
 376         qj->n_posted++;
 377
 378     } else {
 379
 380         if (!(qj = job_new(s, key, 0)))
 381             return 0; /* OOM */
 382
 383         qj->delivery = tv;
 384         qj->time_event = catta_time_event_new(s->time_event_queue, &qj->delivery, elapse_callback, qj);
 385     }
 386
 387     if (ret_id)
 388         *ret_id = qj->id;
 389
 390     return 1;
 391 }
 392
 393 void catta_query_scheduler_incoming(CattaQueryScheduler *s, CattaKey *key) {
 394     CattaQueryJob *qj;
 395
 396     assert(s);
 397     assert(key);
 398
 399     /* This function is called whenever an incoming query was
 400      * received. We drop scheduled queries that match. The keyword is
 401      * "DUPLICATE QUESTION SUPPRESION". */
 402
 403     if ((qj = find_scheduled_job(s, key))) {
 404         job_mark_done(s, qj);
 405         return;
 406     }
 407
 408     /* Look if there's a history job for this key. If there is, just
 409      * update the elapse time */
 410     if (!(qj = find_history_job(s, key)))
 411         if (!(qj = job_new(s, key, 1)))
 412             return; /* OOM */
 413
 414     gettimeofday(&qj->delivery, NULL);
 415     job_set_elapse_time(s, qj, CATTA_QUERY_HISTORY_MSEC, 0);
 416 }
 417
 418 int catta_query_scheduler_withdraw_by_id(CattaQueryScheduler *s, unsigned id) {
 419     CattaQueryJob *qj;
 420
 421     assert(s);
 422
 423     /* Very short lived queries can withdraw an already scheduled item
 424      * from the queue using this function, simply by passing the id
 425      * returned by catta_query_scheduler_post(). */
 426
 427     for (qj = s->jobs; qj; qj = qj->jobs_next) {
 428         assert(!qj->done);
 429
 430         if (qj->id == id) {
 431             /* Entry found */
 432
 433             assert(qj->n_posted >= 1);
 434
 435             if (--qj->n_posted <= 0) {
 436
 437                 /* We withdraw this job only if the calling object was
 438                  * the only remaining poster. (Usually this is the
 439                  * case since there should exist only one querier per
 440                  * key, but there are exceptions, notably reflected
 441                  * traffic.) */
 442
 443                 job_free(s, qj);
 444                 return 1;
 445             }
 446         }
 447     }
 448
 449     return 0;
 450 }