git.meshlink.io Git - catta/blob - src/response-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 "response-sched.h"
  30 #include <catta/log.h>
  31 #include "rr-util.h"
  32
  33 /* Local packets are supressed this long after sending them */
  34 #define CATTA_RESPONSE_HISTORY_MSEC 500
  35
  36 /* Local packets are deferred this long before sending them */
  37 #define CATTA_RESPONSE_DEFER_MSEC 20
  38
  39 /* Additional jitter for deferred packets */
  40 #define CATTA_RESPONSE_JITTER_MSEC 100
  41
  42 /* Remote packets can suppress local traffic as long as this value */
  43 #define CATTA_RESPONSE_SUPPRESS_MSEC 700
  44
  45 typedef struct CattaResponseJob CattaResponseJob;
  46
  47 typedef enum {
  48     CATTA_SCHEDULED,
  49     CATTA_DONE,
  50     CATTA_SUPPRESSED
  51 } CattaResponseJobState;
  52
  53 struct CattaResponseJob {
  54     CattaResponseScheduler *scheduler;
  55     CattaTimeEvent *time_event;
  56
  57     CattaResponseJobState state;
  58     struct timeval delivery;
  59
  60     CattaRecord *record;
  61     int flush_cache;
  62     CattaAddress querier;
  63     int querier_valid;
  64
  65     CATTA_LLIST_FIELDS(CattaResponseJob, jobs);
  66 };
  67
  68 struct CattaResponseScheduler {
  69     CattaInterface *interface;
  70     CattaTimeEventQueue *time_event_queue;
  71
  72     CATTA_LLIST_HEAD(CattaResponseJob, jobs);
  73     CATTA_LLIST_HEAD(CattaResponseJob, history);
  74     CATTA_LLIST_HEAD(CattaResponseJob, suppressed);
  75 };
  76
  77 static CattaResponseJob* job_new(CattaResponseScheduler *s, CattaRecord *record, CattaResponseJobState state) {
  78     CattaResponseJob *rj;
  79
  80     assert(s);
  81     assert(record);
  82
  83     if (!(rj = catta_new(CattaResponseJob, 1))) {
  84         catta_log_error(__FILE__": Out of memory");
  85         return NULL;
  86     }
  87
  88     rj->scheduler = s;
  89     rj->record = catta_record_ref(record);
  90     rj->time_event = NULL;
  91     rj->flush_cache = 0;
  92     rj->querier_valid = 0;
  93
  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);
 100
 101     return rj;
 102 }
 103
 104 static void job_free(CattaResponseScheduler *s, CattaResponseJob *rj) {
 105     assert(s);
 106     assert(rj);
 107
 108     if (rj->time_event)
 109         catta_time_event_free(rj->time_event);
 110
 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);
 117
 118     catta_record_unref(rj->record);
 119     catta_free(rj);
 120 }
 121
 122 static void elapse_callback(CattaTimeEvent *e, void* data);
 123
 124 static void job_set_elapse_time(CattaResponseScheduler *s, CattaResponseJob *rj, unsigned msec, unsigned jitter) {
 125     struct timeval tv;
 126
 127     assert(s);
 128     assert(rj);
 129
 130     catta_elapse_time(&tv, msec, jitter);
 131
 132     if (rj->time_event)
 133         catta_time_event_update(rj->time_event, &tv);
 134     else
 135         rj->time_event = catta_time_event_new(s->time_event_queue, &tv, elapse_callback, rj);
 136 }
 137
 138 static void job_mark_done(CattaResponseScheduler *s, CattaResponseJob *rj) {
 139     assert(s);
 140     assert(rj);
 141
 142     assert(rj->state == CATTA_SCHEDULED);
 143
 144     CATTA_LLIST_REMOVE(CattaResponseJob, jobs, s->jobs, rj);
 145     CATTA_LLIST_PREPEND(CattaResponseJob, jobs, s->history, rj);
 146
 147     rj->state = CATTA_DONE;
 148
 149     job_set_elapse_time(s, rj, CATTA_RESPONSE_HISTORY_MSEC, 0);
 150
 151     gettimeofday(&rj->delivery, NULL);
 152 }
 153
 154 CattaResponseScheduler *catta_response_scheduler_new(CattaInterface *i) {
 155     CattaResponseScheduler *s;
 156     assert(i);
 157
 158     if (!(s = catta_new(CattaResponseScheduler, 1))) {
 159         catta_log_error(__FILE__": Out of memory");
 160         return NULL;
 161     }
 162
 163     s->interface = i;
 164     s->time_event_queue = i->monitor->server->time_event_queue;
 165
 166     CATTA_LLIST_HEAD_INIT(CattaResponseJob, s->jobs);
 167     CATTA_LLIST_HEAD_INIT(CattaResponseJob, s->history);
 168     CATTA_LLIST_HEAD_INIT(CattaResponseJob, s->suppressed);
 169
 170     return s;
 171 }
 172
 173 void catta_response_scheduler_free(CattaResponseScheduler *s) {
 174     assert(s);
 175
 176     catta_response_scheduler_clear(s);
 177     catta_free(s);
 178 }
 179
 180 void catta_response_scheduler_clear(CattaResponseScheduler *s) {
 181     assert(s);
 182
 183     while (s->jobs)
 184         job_free(s, s->jobs);
 185     while (s->history)
 186         job_free(s, s->history);
 187     while (s->suppressed)
 188         job_free(s, s->suppressed);
 189 }
 190
 191 static void enumerate_aux_records_callback(CATTA_GCC_UNUSED CattaServer *s, CattaRecord *r, int flush_cache, void* userdata) {
 192     CattaResponseJob *rj = userdata;
 193
 194     assert(r);
 195     assert(rj);
 196
 197     catta_response_scheduler_post(rj->scheduler, r, flush_cache, rj->querier_valid ? &rj->querier : NULL, 0);
 198 }
 199
 200 static int packet_add_response_job(CattaResponseScheduler *s, CattaDnsPacket *p, CattaResponseJob *rj) {
 201     assert(s);
 202     assert(p);
 203     assert(rj);
 204
 205     /* Try to add this record to the packet */
 206     if (!catta_dns_packet_append_record(p, rj->record, rj->flush_cache, 0))
 207         return 0;
 208
 209     /* Ok, this record will definitely be sent, so schedule the
 210      * auxilliary packets, too */
 211     catta_server_enumerate_aux_records(s->interface->monitor->server, s->interface, rj->record, enumerate_aux_records_callback, rj);
 212     job_mark_done(s, rj);
 213
 214     return 1;
 215 }
 216
 217 static void send_response_packet(CattaResponseScheduler *s, CattaResponseJob *rj) {
 218     CattaDnsPacket *p;
 219     unsigned n;
 220
 221     assert(s);
 222     assert(rj);
 223
 224     if (!(p = catta_dns_packet_new_response(s->interface->hardware->mtu, 1)))
 225         return; /* OOM */
 226     n = 1;
 227
 228     /* Put it in the packet. */
 229     if (packet_add_response_job(s, p, rj)) {
 230
 231         /* Try to fill up packet with more responses, if available */
 232         while (s->jobs) {
 233
 234             if (!packet_add_response_job(s, p, s->jobs))
 235                 break;
 236
 237             n++;
 238         }
 239
 240     } else {
 241         size_t size;
 242
 243         catta_dns_packet_free(p);
 244
 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;
 247
 248         if (!(p = catta_dns_packet_new_response(size + CATTA_DNS_PACKET_EXTRA_SIZE, 1)))
 249             return; /* OOM */
 250
 251         if (!packet_add_response_job(s, p, rj)) {
 252             catta_dns_packet_free(p);
 253
 254             catta_log_warn("Record too large, cannot send");
 255             job_mark_done(s, rj);
 256             return;
 257         }
 258     }
 259
 260     catta_dns_packet_set_field(p, CATTA_DNS_FIELD_ANCOUNT, n);
 261     catta_interface_send_packet(s->interface, p);
 262     catta_dns_packet_free(p);
 263 }
 264
 265 static void elapse_callback(CATTA_GCC_UNUSED CattaTimeEvent *e, void* data) {
 266     CattaResponseJob *rj = data;
 267
 268     assert(rj);
 269
 270     if (rj->state == CATTA_DONE || rj->state == CATTA_SUPPRESSED)
 271         job_free(rj->scheduler, rj);         /* Lets drop this entry */
 272     else
 273         send_response_packet(rj->scheduler, rj);
 274 }
 275
 276 static CattaResponseJob* find_scheduled_job(CattaResponseScheduler *s, CattaRecord *record) {
 277     CattaResponseJob *rj;
 278
 279     assert(s);
 280     assert(record);
 281
 282     for (rj = s->jobs; rj; rj = rj->jobs_next) {
 283         assert(rj->state == CATTA_SCHEDULED);
 284
 285         if (catta_record_equal_no_ttl(rj->record, record))
 286             return rj;
 287     }
 288
 289     return NULL;
 290 }
 291
 292 static CattaResponseJob* find_history_job(CattaResponseScheduler *s, CattaRecord *record) {
 293     CattaResponseJob *rj;
 294
 295     assert(s);
 296     assert(record);
 297
 298     for (rj = s->history; rj; rj = rj->jobs_next) {
 299         assert(rj->state == CATTA_DONE);
 300
 301         if (catta_record_equal_no_ttl(rj->record, record)) {
 302             /* Check whether this entry is outdated */
 303
 304 /*             catta_log_debug("history age: %u", (unsigned) (catta_age(&rj->delivery)/1000)); */
 305
 306             if (catta_age(&rj->delivery)/1000 > CATTA_RESPONSE_HISTORY_MSEC) {
 307                 /* it is outdated, so let's remove it */
 308                 job_free(s, rj);
 309                 return NULL;
 310             }
 311
 312             return rj;
 313         }
 314     }
 315
 316     return NULL;
 317 }
 318
 319 static CattaResponseJob* find_suppressed_job(CattaResponseScheduler *s, CattaRecord *record, const CattaAddress *querier) {
 320     CattaResponseJob *rj;
 321
 322     assert(s);
 323     assert(record);
 324     assert(querier);
 325
 326     for (rj = s->suppressed; rj; rj = rj->jobs_next) {
 327         assert(rj->state == CATTA_SUPPRESSED);
 328         assert(rj->querier_valid);
 329
 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 */
 333
 334             if (catta_age(&rj->delivery) > CATTA_RESPONSE_SUPPRESS_MSEC*1000) {
 335                 /* it is outdated, so let's remove it */
 336                 job_free(s, rj);
 337                 return NULL;
 338             }
 339
 340             return rj;
 341         }
 342     }
 343
 344     return NULL;
 345 }
 346
 347 int catta_response_scheduler_post(CattaResponseScheduler *s, CattaRecord *record, int flush_cache, const CattaAddress *querier, int immediately) {
 348     CattaResponseJob *rj;
 349     struct timeval tv;
 350 /*     char *t; */
 351
 352     assert(s);
 353     assert(record);
 354
 355     assert(!catta_key_is_pattern(record->key));
 356
 357 /*     t = catta_record_to_string(record); */
 358 /*     catta_log_debug("post %i %s", immediately, t); */
 359 /*     catta_free(t); */
 360
 361     /* Check whether this response is suppressed */
 362     if (querier &&
 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) {
 366
 367 /*         catta_log_debug("Response suppressed by known answer suppression.");  */
 368         return 0;
 369     }
 370
 371     /* Check if we already sent this response recently */
 372     if ((rj = find_history_job(s, record))) {
 373
 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)");  */
 378             return 0;
 379         }
 380
 381         /* Outdated ... */
 382         job_free(s, rj);
 383     }
 384
 385     catta_elapse_time(&tv, immediately ? 0 : CATTA_RESPONSE_DEFER_MSEC, immediately ? 0 : CATTA_RESPONSE_JITTER_MSEC);
 386
 387     if ((rj = find_scheduled_job(s, record))) {
 388 /*          catta_log_debug("Response suppressed by local duplicate suppression (scheduled)"); */
 389
 390         /* Update a little ... */
 391
 392         /* Update the time if the new is prior to the old */
 393         if (catta_timeval_compare(&tv, &rj->delivery) < 0) {
 394             rj->delivery = tv;
 395             catta_time_event_update(rj->time_event, &rj->delivery);
 396         }
 397
 398         /* Update the flush cache bit */
 399         if (flush_cache)
 400             rj->flush_cache = 1;
 401
 402         /* Update the querier field */
 403         if (!querier || (rj->querier_valid && catta_address_cmp(querier, &rj->querier) != 0))
 404             rj->querier_valid = 0;
 405
 406         /* Update record data (just for the TTL) */
 407         catta_record_unref(rj->record);
 408         rj->record = catta_record_ref(record);
 409
 410         return 1;
 411     } else {
 412 /*         catta_log_debug("Accepted new response job.");  */
 413
 414         /* Create a new job and schedule it */
 415         if (!(rj = job_new(s, record, CATTA_SCHEDULED)))
 416             return 0; /* OOM */
 417
 418         rj->delivery = tv;
 419         rj->time_event = catta_time_event_new(s->time_event_queue, &rj->delivery, elapse_callback, rj);
 420         rj->flush_cache = flush_cache;
 421
 422         if ((rj->querier_valid = !!querier))
 423             rj->querier = *querier;
 424
 425         return 1;
 426     }
 427 }
 428
 429 void catta_response_scheduler_incoming(CattaResponseScheduler *s, CattaRecord *record, int flush_cache) {
 430     CattaResponseJob *rj;
 431     assert(s);
 432
 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". */
 436
 437     if ((rj = find_scheduled_job(s, record))) {
 438
 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 */
 442
 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);
 446         }
 447
 448         return;
 449     }
 450
 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);
 455     } else
 456         /* Found no existing history job, so let's create a new one */
 457         if (!(rj = job_new(s, record, CATTA_DONE)))
 458             return; /* OOM */
 459
 460     rj->flush_cache = flush_cache;
 461     rj->querier_valid = 0;
 462
 463     gettimeofday(&rj->delivery, NULL);
 464     job_set_elapse_time(s, rj, CATTA_RESPONSE_HISTORY_MSEC, 0);
 465 }
 466
 467 void catta_response_scheduler_suppress(CattaResponseScheduler *s, CattaRecord *record, const CattaAddress *querier) {
 468     CattaResponseJob *rj;
 469
 470     assert(s);
 471     assert(record);
 472     assert(querier);
 473
 474     if ((rj = find_scheduled_job(s, record))) {
 475
 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 */
 479
 480             /* A matching entry was found, so let's drop it */
 481 /*             catta_log_debug("Known answer suppression active!"); */
 482             job_free(s, rj);
 483         }
 484     }
 485
 486     if ((rj = find_suppressed_job(s, record, querier))) {
 487
 488         /* Let's update the old entry */
 489         catta_record_unref(rj->record);
 490         rj->record = catta_record_ref(record);
 491
 492     } else {
 493
 494         /* Create a new entry */
 495         if (!(rj = job_new(s, record, CATTA_SUPPRESSED)))
 496             return; /* OOM */
 497         rj->querier_valid = 1;
 498         rj->querier = *querier;
 499     }
 500
 501     gettimeofday(&rj->delivery, NULL);
 502     job_set_elapse_time(s, rj, CATTA_RESPONSE_SUPPRESS_MSEC, 0);
 503 }
 504
 505 void catta_response_scheduler_force(CattaResponseScheduler *s) {
 506     assert(s);
 507
 508     /* Send all scheduled responses immediately */
 509     while (s->jobs)
 510         send_response_packet(s, s->jobs);
 511 }