git.meshlink.io Git - catta/blob - avahi-core/response-sched.c

   1 /* $Id$ */
   2
   3 /***
   4   This file is part of avahi.
   5
   6   avahi is free software; you can redistribute it and/or modify it
   7   under the terms of the GNU Lesser General Public License as
   8   published by the Free Software Foundation; either version 2.1 of the
   9   License, or (at your option) any later version.
  10
  11   avahi is distributed in the hope that it will be useful, but WITHOUT
  12   ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
  13   or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General
  14   Public License for more details.
  15
  16   You should have received a copy of the GNU Lesser General Public
  17   License along with avahi; if not, write to the Free Software
  18   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
  19   USA.
  20 ***/
  21
  22 #ifdef HAVE_CONFIG_H
  23 #include <config.h>
  24 #endif
  25
  26 #include <avahi-common/timeval.h>
  27 #include <avahi-common/malloc.h>
  28
  29 #include "response-sched.h"
  30 #include "log.h"
  31
  32 #define AVAHI_RESPONSE_HISTORY_MSEC 500
  33 #define AVAHI_RESPONSE_DEFER_MSEC 20
  34 #define AVAHI_RESPONSE_JITTER_MSEC 100
  35 #define AVAHI_RESPONSE_SUPPRESS_MSEC 700
  36
  37 typedef struct AvahiResponseJob AvahiResponseJob;
  38
  39 typedef enum {
  40     AVAHI_SCHEDULED,
  41     AVAHI_DONE,
  42     AVAHI_SUPPRESSED
  43 } AvahiResponseJobState;
  44
  45 struct AvahiResponseJob {
  46     AvahiResponseScheduler *scheduler;
  47     AvahiTimeEvent *time_event;
  48
  49     AvahiResponseJobState state;
  50     struct timeval delivery;
  51
  52     AvahiRecord *record;
  53     int flush_cache;
  54     AvahiAddress querier;
  55     int querier_valid;
  56
  57     AVAHI_LLIST_FIELDS(AvahiResponseJob, jobs);
  58 };
  59
  60 struct AvahiResponseScheduler {
  61     AvahiInterface *interface;
  62     AvahiTimeEventQueue *time_event_queue;
  63
  64     AVAHI_LLIST_HEAD(AvahiResponseJob, jobs);
  65     AVAHI_LLIST_HEAD(AvahiResponseJob, history);
  66     AVAHI_LLIST_HEAD(AvahiResponseJob, suppressed);
  67 };
  68
  69 static AvahiResponseJob* job_new(AvahiResponseScheduler *s, AvahiRecord *record, AvahiResponseJobState state) {
  70     AvahiResponseJob *rj;
  71
  72     assert(s);
  73     assert(record);
  74
  75     if (!(rj = avahi_new(AvahiResponseJob, 1))) {
  76         avahi_log_error(__FILE__": Out of memory");
  77         return NULL;
  78     }
  79
  80     rj->scheduler = s;
  81     rj->record = avahi_record_ref(record);
  82     rj->time_event = NULL;
  83     rj->flush_cache = 0;
  84     rj->querier_valid = 0;
  85
  86     if ((rj->state = state) == AVAHI_SCHEDULED)
  87         AVAHI_LLIST_PREPEND(AvahiResponseJob, jobs, s->jobs, rj);
  88     else if (rj->state == AVAHI_DONE)
  89         AVAHI_LLIST_PREPEND(AvahiResponseJob, jobs, s->history, rj);
  90     else  /* rj->state == AVAHI_SUPPRESSED */
  91         AVAHI_LLIST_PREPEND(AvahiResponseJob, jobs, s->suppressed, rj);
  92
  93     return rj;
  94 }
  95
  96 static void job_free(AvahiResponseScheduler *s, AvahiResponseJob *rj) {
  97     assert(s);
  98     assert(rj);
  99
 100     if (rj->time_event)
 101         avahi_time_event_free(rj->time_event);
 102
 103     if (rj->state == AVAHI_SCHEDULED)
 104         AVAHI_LLIST_REMOVE(AvahiResponseJob, jobs, s->jobs, rj);
 105     else if (rj->state == AVAHI_DONE)
 106         AVAHI_LLIST_REMOVE(AvahiResponseJob, jobs, s->history, rj);
 107     else /* rj->state == AVAHI_SUPPRESSED */
 108         AVAHI_LLIST_REMOVE(AvahiResponseJob, jobs, s->suppressed, rj);
 109
 110     avahi_record_unref(rj->record);
 111     avahi_free(rj);
 112 }
 113
 114 static void elapse_callback(AvahiTimeEvent *e, void* data);
 115
 116 static void job_set_elapse_time(AvahiResponseScheduler *s, AvahiResponseJob *rj, unsigned msec, unsigned jitter) {
 117     struct timeval tv;
 118
 119     assert(s);
 120     assert(rj);
 121
 122     avahi_elapse_time(&tv, msec, jitter);
 123
 124     if (rj->time_event)
 125         avahi_time_event_update(rj->time_event, &tv);
 126     else
 127         rj->time_event = avahi_time_event_new(s->time_event_queue, &tv, elapse_callback, rj);
 128 }
 129
 130 static void job_mark_done(AvahiResponseScheduler *s, AvahiResponseJob *rj) {
 131     assert(s);
 132     assert(rj);
 133
 134     assert(rj->state == AVAHI_SCHEDULED);
 135
 136     AVAHI_LLIST_REMOVE(AvahiResponseJob, jobs, s->jobs, rj);
 137     AVAHI_LLIST_PREPEND(AvahiResponseJob, jobs, s->history, rj);
 138
 139     rj->state = AVAHI_DONE;
 140
 141     job_set_elapse_time(s, rj, AVAHI_RESPONSE_HISTORY_MSEC, 0);
 142
 143     gettimeofday(&rj->delivery, NULL);
 144 }
 145
 146 AvahiResponseScheduler *avahi_response_scheduler_new(AvahiInterface *i) {
 147     AvahiResponseScheduler *s;
 148     assert(i);
 149
 150     if (!(s = avahi_new(AvahiResponseScheduler, 1))) {
 151         avahi_log_error(__FILE__": Out of memory");
 152         return NULL;
 153     }
 154
 155     s->interface = i;
 156     s->time_event_queue = i->monitor->server->time_event_queue;
 157
 158     AVAHI_LLIST_HEAD_INIT(AvahiResponseJob, s->jobs);
 159     AVAHI_LLIST_HEAD_INIT(AvahiResponseJob, s->history);
 160     AVAHI_LLIST_HEAD_INIT(AvahiResponseJob, s->suppressed);
 161
 162     return s;
 163 }
 164
 165 void avahi_response_scheduler_free(AvahiResponseScheduler *s) {
 166     assert(s);
 167
 168     avahi_response_scheduler_clear(s);
 169     avahi_free(s);
 170 }
 171
 172 void avahi_response_scheduler_clear(AvahiResponseScheduler *s) {
 173     assert(s);
 174
 175     while (s->jobs)
 176         job_free(s, s->jobs);
 177     while (s->history)
 178         job_free(s, s->history);
 179     while (s->suppressed)
 180         job_free(s, s->suppressed);
 181 }
 182
 183 static void enumerate_aux_records_callback(AvahiServer *s, AvahiRecord *r, int flush_cache, void* userdata) {
 184     AvahiResponseJob *rj = userdata;
 185
 186     assert(r);
 187     assert(rj);
 188
 189     avahi_response_scheduler_post(rj->scheduler, r, flush_cache, rj->querier_valid ? &rj->querier : NULL, 0);
 190 }
 191
 192 static int packet_add_response_job(AvahiResponseScheduler *s, AvahiDnsPacket *p, AvahiResponseJob *rj) {
 193     assert(s);
 194     assert(p);
 195     assert(rj);
 196
 197     /* Try to add this record to the packet */
 198     if (!avahi_dns_packet_append_record(p, rj->record, rj->flush_cache, 0))
 199         return 0;
 200
 201     /* Ok, this record will definitely be sent, so schedule the
 202      * auxilliary packets, too */
 203     avahi_server_enumerate_aux_records(s->interface->monitor->server, s->interface, rj->record, enumerate_aux_records_callback, rj);
 204     job_mark_done(s, rj);
 205
 206     return 1;
 207 }
 208
 209 static void send_response_packet(AvahiResponseScheduler *s, AvahiResponseJob *rj) {
 210     AvahiDnsPacket *p;
 211     unsigned n;
 212
 213     assert(s);
 214     assert(rj);
 215
 216     if (!(p = avahi_dns_packet_new_response(s->interface->hardware->mtu, 1)))
 217         return; /* OOM */
 218     n = 1;
 219
 220     /* Put it in the packet. */
 221     if (packet_add_response_job(s, p, rj)) {
 222
 223         /* Try to fill up packet with more responses, if available */
 224         while (s->jobs) {
 225
 226             if (!packet_add_response_job(s, p, s->jobs))
 227                 break;
 228
 229             n++;
 230         }
 231
 232     } else {
 233         size_t size;
 234
 235         avahi_dns_packet_free(p);
 236
 237         /* OK, the packet was too small, so create one that fits */
 238         size = avahi_record_get_estimate_size(rj->record) + AVAHI_DNS_PACKET_HEADER_SIZE;
 239
 240         if (size > AVAHI_DNS_PACKET_MAX_SIZE)
 241             size = AVAHI_DNS_PACKET_MAX_SIZE;
 242
 243         if (!(p = avahi_dns_packet_new_response(size, 1)))
 244             return; /* OOM */
 245
 246         if (!packet_add_response_job(s, p, rj)) {
 247             avahi_dns_packet_free(p);
 248
 249             avahi_log_warn("Record too large, cannot send");
 250             job_mark_done(s, rj);
 251             return;
 252         }
 253     }
 254
 255     avahi_dns_packet_set_field(p, AVAHI_DNS_FIELD_ANCOUNT, n);
 256     avahi_interface_send_packet(s->interface, p);
 257     avahi_dns_packet_free(p);
 258 }
 259
 260 static void elapse_callback(AvahiTimeEvent *e, void* data) {
 261     AvahiResponseJob *rj = data;
 262
 263     assert(rj);
 264
 265     if (rj->state == AVAHI_DONE || rj->state == AVAHI_SUPPRESSED)
 266         job_free(rj->scheduler, rj);         /* Lets drop this entry */
 267     else
 268         send_response_packet(rj->scheduler, rj);
 269 }
 270
 271 static AvahiResponseJob* find_scheduled_job(AvahiResponseScheduler *s, AvahiRecord *record) {
 272     AvahiResponseJob *rj;
 273
 274     assert(s);
 275     assert(record);
 276
 277     for (rj = s->jobs; rj; rj = rj->jobs_next) {
 278         assert(rj->state == AVAHI_SCHEDULED);
 279
 280         if (avahi_record_equal_no_ttl(rj->record, record))
 281             return rj;
 282     }
 283
 284     return NULL;
 285 }
 286
 287 static AvahiResponseJob* find_history_job(AvahiResponseScheduler *s, AvahiRecord *record) {
 288     AvahiResponseJob *rj;
 289
 290     assert(s);
 291     assert(record);
 292
 293     for (rj = s->history; rj; rj = rj->jobs_next) {
 294         assert(rj->state == AVAHI_DONE);
 295
 296         if (avahi_record_equal_no_ttl(rj->record, record)) {
 297             /* Check whether this entry is outdated */
 298
 299 /*             avahi_log_debug("history age: %u", (unsigned) (avahi_age(&rj->delivery)/1000)); */
 300
 301             if (avahi_age(&rj->delivery)/1000 > AVAHI_RESPONSE_HISTORY_MSEC) {
 302                 /* it is outdated, so let's remove it */
 303                 job_free(s, rj);
 304                 return NULL;
 305             }
 306
 307             return rj;
 308         }
 309     }
 310
 311     return NULL;
 312 }
 313
 314 static AvahiResponseJob* find_suppressed_job(AvahiResponseScheduler *s, AvahiRecord *record, const AvahiAddress *querier) {
 315     AvahiResponseJob *rj;
 316
 317     assert(s);
 318     assert(record);
 319     assert(querier);
 320
 321     for (rj = s->suppressed; rj; rj = rj->jobs_next) {
 322         assert(rj->state == AVAHI_SUPPRESSED);
 323         assert(rj->querier_valid);
 324
 325         if (avahi_record_equal_no_ttl(rj->record, record) &&
 326             avahi_address_cmp(&rj->querier, querier) == 0) {
 327             /* Check whether this entry is outdated */
 328
 329             if (avahi_age(&rj->delivery) > AVAHI_RESPONSE_SUPPRESS_MSEC*1000) {
 330                 /* it is outdated, so let's remove it */
 331                 job_free(s, rj);
 332                 return NULL;
 333             }
 334
 335             return rj;
 336         }
 337     }
 338
 339     return NULL;
 340 }
 341
 342 int avahi_response_scheduler_post(AvahiResponseScheduler *s, AvahiRecord *record, int flush_cache, const AvahiAddress *querier, int immediately) {
 343     AvahiResponseJob *rj;
 344     struct timeval tv;
 345 /*     char *t; */
 346
 347     assert(s);
 348     assert(record);
 349
 350     assert(!avahi_key_is_pattern(record->key));
 351
 352 /*     t = avahi_record_to_string(record); */
 353 /*     avahi_log_debug("post %i %s", immediately, t); */
 354 /*     avahi_free(t); */
 355
 356     /* Check whether this response is suppressed */
 357     if (querier &&
 358         (rj = find_suppressed_job(s, record, querier)) &&
 359         avahi_record_is_goodbye(record) == avahi_record_is_goodbye(rj->record) &&
 360         rj->record->ttl >= record->ttl/2) {
 361
 362 /*         avahi_log_debug("Response suppressed by known answer suppression.");  */
 363         return 0;
 364     }
 365
 366     /* Check if we already sent this response recently */
 367     if ((rj = find_history_job(s, record))) {
 368
 369         if (avahi_record_is_goodbye(record) == avahi_record_is_goodbye(rj->record) &&
 370             rj->record->ttl >= record->ttl/2 &&
 371             (rj->flush_cache || !flush_cache)) {
 372 /*             avahi_log_debug("Response suppressed by local duplicate suppression (history)");  */
 373             return 0;
 374         }
 375
 376         /* Outdated ... */
 377         job_free(s, rj);
 378     }
 379
 380     avahi_elapse_time(&tv, immediately ? 0 : AVAHI_RESPONSE_DEFER_MSEC, immediately ? 0 : AVAHI_RESPONSE_JITTER_MSEC);
 381
 382     if ((rj = find_scheduled_job(s, record))) {
 383 /*          avahi_log_debug("Response suppressed by local duplicate suppression (scheduled)"); */
 384
 385         /* Update a little ... */
 386
 387         /* Update the time if the new is prior to the old */
 388         if (avahi_timeval_compare(&tv, &rj->delivery) < 0) {
 389             rj->delivery = tv;
 390             avahi_time_event_update(rj->time_event, &rj->delivery);
 391         }
 392
 393         /* Update the flush cache bit */
 394         if (flush_cache)
 395             rj->flush_cache = 1;
 396
 397         /* Update the querier field */
 398         if (!querier || (rj->querier_valid && avahi_address_cmp(querier, &rj->querier) != 0))
 399             rj->querier_valid = 0;
 400
 401         /* Update record data (just for the TTL) */
 402         avahi_record_unref(rj->record);
 403         rj->record = avahi_record_ref(record);
 404
 405         return 1;
 406     } else {
 407 /*         avahi_log_debug("Accepted new response job.");  */
 408
 409         /* Create a new job and schedule it */
 410         if (!(rj = job_new(s, record, AVAHI_SCHEDULED)))
 411             return 0; /* OOM */
 412
 413         rj->delivery = tv;
 414         rj->time_event = avahi_time_event_new(s->time_event_queue, &rj->delivery, elapse_callback, rj);
 415         rj->flush_cache = flush_cache;
 416
 417         if ((rj->querier_valid = !!querier))
 418             rj->querier = *querier;
 419
 420         return 1;
 421     }
 422 }
 423
 424 void avahi_response_scheduler_incoming(AvahiResponseScheduler *s, AvahiRecord *record, int flush_cache) {
 425     AvahiResponseJob *rj;
 426     assert(s);
 427
 428     /* This function is called whenever an incoming response was
 429      * receieved. We drop scheduled responses which match here. The
 430      * keyword is "DUPLICATE ANSWER SUPPRESION". */
 431
 432     if ((rj = find_scheduled_job(s, record))) {
 433
 434         if ((!rj->flush_cache || flush_cache) &&    /* flush cache bit was set correctly */
 435             avahi_record_is_goodbye(record) == avahi_record_is_goodbye(rj->record) &&   /* both goodbye packets, or both not */
 436             record->ttl >= rj->record->ttl/2) {     /* sensible TTL */
 437
 438             /* A matching entry was found, so let's mark it done */
 439 /*             avahi_log_debug("Response suppressed by distributed duplicate suppression"); */
 440             job_mark_done(s, rj);
 441         }
 442
 443         return;
 444     }
 445
 446     if ((rj = find_history_job(s, record))) {
 447         /* Found a history job, let's update it */
 448         avahi_record_unref(rj->record);
 449         rj->record = avahi_record_ref(record);
 450     } else
 451         /* Found no existing history job, so let's create a new one */
 452         if (!(rj = job_new(s, record, AVAHI_DONE)))
 453             return; /* OOM */
 454
 455     rj->flush_cache = flush_cache;
 456     rj->querier_valid = 0;
 457
 458     gettimeofday(&rj->delivery, NULL);
 459     job_set_elapse_time(s, rj, AVAHI_RESPONSE_HISTORY_MSEC, 0);
 460 }
 461
 462 void avahi_response_scheduler_suppress(AvahiResponseScheduler *s, AvahiRecord *record, const AvahiAddress *querier) {
 463     AvahiResponseJob *rj;
 464
 465     assert(s);
 466     assert(record);
 467     assert(querier);
 468
 469     if ((rj = find_scheduled_job(s, record))) {
 470
 471         if (rj->querier_valid && avahi_address_cmp(querier, &rj->querier) == 0 && /* same originator */
 472             avahi_record_is_goodbye(record) == avahi_record_is_goodbye(rj->record) && /* both goodbye packets, or both not */
 473             record->ttl >= rj->record->ttl/2) {                                  /* sensible TTL */
 474
 475             /* A matching entry was found, so let's drop it */
 476 /*             avahi_log_debug("Known answer suppression active!"); */
 477             job_free(s, rj);
 478         }
 479     }
 480
 481     if ((rj = find_suppressed_job(s, record, querier))) {
 482
 483         /* Let's update the old entry */
 484         avahi_record_unref(rj->record);
 485         rj->record = avahi_record_ref(record);
 486
 487     } else {
 488
 489         /* Create a new entry */
 490         if (!(rj = job_new(s, record, AVAHI_SUPPRESSED)))
 491             return; /* OOM */
 492         rj->querier_valid = 1;
 493         rj->querier = *querier;
 494     }
 495
 496     gettimeofday(&rj->delivery, NULL);
 497     job_set_elapse_time(s, rj, AVAHI_RESPONSE_SUPPRESS_MSEC, 0);
 498 }
 499
 500 void avahi_response_scheduler_force(AvahiResponseScheduler *s) {
 501     assert(s);
 502
 503     /* Send all scheduled responses immediately */
 504     while (s->jobs)
 505         send_response_packet(s, s->jobs);
 506 }