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