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