git.sesse.net Git - nageru/blob - nageru/experiments/queue_drop_policy.cpp

   1 /*
   2  * A program to simulate various queue-drop strategies, using real frame
   3  * arrival data as input. Contains various anchors, as well as parametrized
   4  * values of the real algorithms that have been used in Nageru over time.
   5  *
   6  * Expects a log of frame arrivals (in and out). This isn't included in the
   7  * git repository because it's quite large, but there's one available
   8  * in compressed form at
   9  *
  10  *   https://storage.sesse.net/nageru-latency-log.txt.xz
  11  *
  12  * The data set in question contains a rather difficult case, with two 50 Hz
  13  * clocks slowly drifting from each other (at the rate of about a frame an hour).
  14  * This means they are very nearly in sync for a long time, where rare bursts
  15  * of jitter can make it hard for the algorithm to find the right level of
  16  * conservatism.
  17  *
  18  * This is not meant to be production-quality code.
  19  */
  20
  21 #include <assert.h>
  22 #include <getopt.h>
  23 #include <math.h>
  24 #include <stdio.h>
  25 #include <locale.h>
  26 #include <string.h>
  27 #include <stdlib.h>
  28 #include <algorithm>
  29 #include <vector>
  30 #include <deque>
  31 #include <memory>
  32 #include <string>
  33 #include <limits>
  34
  35 using namespace std;
  36
  37 size_t max_drops = numeric_limits<size_t>::max();
  38 size_t max_underruns = numeric_limits<size_t>::max();
  39 double max_latency_ms = numeric_limits<double>::max();
  40
  41 struct Event {
  42         enum { IN, OUT } direction;
  43         double t;
  44 };
  45
  46 class Queue {
  47 public:
  48         void add_frame(double t);
  49         void get_frame(double now);
  50         void drop_frame();
  51         void eval(const string &name);
  52         size_t queue_len() const { return frames_in_queue.size(); }
  53         bool should_abort() const { return num_underruns > max_underruns || num_drops > max_drops; }
  54
  55 private:
  56         deque<double> frames_in_queue;
  57         size_t num_underruns = 0;
  58         size_t num_drops = 0;
  59         size_t frames_since_underrun = 0;
  60         size_t num_drops_on_first = 0;
  61
  62         double latency_sum = 0.0;
  63         size_t latency_count = 0;
  64 };
  65
  66 void Queue::add_frame(double t)
  67 {
  68         frames_in_queue.push_back(t);
  69 }
  70
  71 void Queue::get_frame(double now)
  72 {
  73         if (frames_in_queue.empty()) {
  74                 ++num_underruns;
  75                 frames_since_underrun = 0;
  76                 return;
  77         }
  78         double t = frames_in_queue.front();
  79         frames_in_queue.pop_front();
  80         assert(now >= t);
  81         latency_sum += (now - t);
  82         ++latency_count;
  83         ++frames_since_underrun;
  84 }
  85
  86 void Queue::drop_frame()
  87 {
  88         assert(!frames_in_queue.empty());
  89         frames_in_queue.pop_front();
  90         ++num_drops;
  91         if (frames_since_underrun <= 1) {
  92                 ++num_drops_on_first;
  93         }
  94 }
  95
  96 void Queue::eval(const string &name)
  97 {
  98         double latency_ms = 1e3 * latency_sum / latency_count;
  99         if (num_underruns > max_underruns) return;
 100         if (num_drops > max_drops) return;
 101         if (latency_ms > max_latency_ms) return;
 102         printf("%-50s: %2lu frames left in queue at end, %5lu underruns, %5lu drops (%5lu immediate), %6.2f ms avg latency\n",
 103                 name.c_str(), frames_in_queue.size(), num_underruns, num_drops, num_drops_on_first, latency_ms);
 104 }
 105
 106 // A strategy that never drops; low anchor for drops and underruns, high anchor for latency.
 107 void test_nodrop(const vector<Event> &events)
 108 {
 109         Queue q;
 110         for (const Event &event : events) {
 111                 if (event.direction == Event::IN) {
 112                         q.add_frame(event.t);
 113                 } else {
 114                         q.get_frame(event.t);
 115                 }
 116         }
 117         q.eval("no-drop");
 118 }
 119
 120 // A strategy that accepts only one element in the queue; low anchor for latency.
 121 void test_limit_to_1(const vector<Event> &events)
 122 {
 123         Queue q;
 124         for (const Event &event : events) {
 125                 if (event.direction == Event::IN) {
 126                         q.add_frame(event.t);
 127                         while (q.queue_len() > 1) q.drop_frame();
 128                 } else {
 129                         q.get_frame(event.t);
 130                 }
 131         }
 132         q.eval("limit-to-1");
 133 }
 134
 135 // A strategy that accepts one or two elements in the queue.
 136 void test_limit_to_2(const vector<Event> &events)
 137 {
 138         Queue q;
 139         for (const Event &event : events) {
 140                 if (event.direction == Event::IN) {
 141                         q.add_frame(event.t);
 142                         while (q.queue_len() > 2) q.drop_frame();
 143                 } else {
 144                         q.get_frame(event.t);
 145                 }
 146         }
 147         q.eval("limit-to-2");
 148 }
 149
 150 // The algorithm used from Nageru 1.2.0 to 1.6.0; raise the ceiling by 1 every time
 151 // we underrun, drop it if the ceiling hasn't been needed for 1000 frames.
 152 void test_nageru_1_2_0(const vector<Event> &events)
 153 {
 154         Queue q;
 155         unsigned safe_queue_length = 1;
 156         unsigned frames_with_at_least_one = 0;
 157         bool been_at_safe_point_since_last_starvation = false;
 158         for (const Event &event : events) {
 159                 if (event.direction == Event::IN) {
 160                         q.add_frame(event.t);
 161                 } else {
 162                         unsigned queue_length = q.queue_len();
 163                         if (queue_length == 0) {  // Starvation.
 164                                 if (been_at_safe_point_since_last_starvation /*&& safe_queue_length < unsigned(global_flags.max_input_queue_frames)*/) {
 165                                         ++safe_queue_length;
 166                                 }
 167                                 frames_with_at_least_one = 0;
 168                                 been_at_safe_point_since_last_starvation = false;
 169                                 q.get_frame(event.t);  // mark it
 170                                 continue;
 171                         }
 172                         if (queue_length >= safe_queue_length) {
 173                                 been_at_safe_point_since_last_starvation = true;
 174                         }
 175                         if (++frames_with_at_least_one >= 1000 && safe_queue_length > 1) {
 176                                 --safe_queue_length;
 177                                 frames_with_at_least_one = 0;
 178                         }
 179                         while (q.queue_len() > safe_queue_length) {
 180                                 q.drop_frame();
 181                         }
 182                         q.get_frame(event.t);
 183                 }
 184         }
 185         q.eval("nageru-1.2.0");
 186 }
 187
 188 class Jitter {
 189         const double multiplier, alpha;
 190         double expected_timestamp = -1.0;
 191         double max_jitter_seconds = 0.0;
 192
 193 public:
 194         Jitter(double multiplier, double alpha)
 195                 : multiplier(multiplier), alpha(alpha) {}
 196
 197         void update(double timestamp, double frame_duration, size_t dropped_frames)
 198         {
 199                 if (expected_timestamp >= 0.0) {
 200                         expected_timestamp += dropped_frames * frame_duration;
 201                         double jitter_seconds = fabs(expected_timestamp - timestamp);
 202                         max_jitter_seconds = max(multiplier * jitter_seconds, alpha * max_jitter_seconds);  // About two seconds half-time.
 203
 204                         // Cap at 100 ms.
 205                         max_jitter_seconds = min(max_jitter_seconds, 0.1);
 206                 }
 207                 expected_timestamp = timestamp + frame_duration;
 208         }
 209
 210         double get_expected() const
 211         {
 212                 return expected_timestamp;
 213         }
 214
 215         double get_jitter() const
 216         {
 217                 return max_jitter_seconds;
 218         }
 219 };
 220
 221 // Keep a running estimate of k times max jitter seen, decreasing by a factor alpha every frame.
 222 void test_jitter_filter(const vector<Event> &events, double multiplier, double alpha, double margin)
 223 {
 224         Queue q;
 225         Jitter input_jitter(multiplier, alpha);
 226         Jitter output_jitter(multiplier, alpha);
 227
 228         for (const Event &event : events) {
 229                 if (event.direction == Event::IN) {
 230                         input_jitter.update(event.t, 0.020, 0);
 231                         q.add_frame(event.t);
 232                 } else {
 233                         double now = event.t;
 234                         output_jitter.update(event.t, 0.020, 0);
 235                         q.get_frame(event.t);
 236
 237                         double seconds_until_next_frame = max(input_jitter.get_expected() - now + input_jitter.get_jitter(), 0.0);
 238                         double master_frame_length_seconds = 0.020;
 239
 240                         seconds_until_next_frame += margin;  // Hack.
 241
 242                         size_t safe_queue_length = max<int>(floor((seconds_until_next_frame + output_jitter.get_jitter()) / master_frame_length_seconds), 0);
 243                         while (q.queue_len() > safe_queue_length) {
 244                                 q.drop_frame();
 245                         }
 246                 }
 247                 if (q.should_abort()) return;
 248         }
 249
 250         char name[256];
 251         snprintf(name, sizeof(name), "jitter-filter[mul=%.1f,alpha=%.4f,margin=%.1f]", multiplier, alpha, 1e3 * margin);
 252         q.eval(name);
 253 }
 254
 255 // Implements an unbalanced binary search tree that can also satisfy order queries
 256 // (e.g. “give me the 86th largest entry”).
 257 class HistoryJitter {
 258         const size_t history_length;
 259         const double multiplier, percentile;
 260         double expected_timestamp = 0.0;
 261         double max_jitter_seconds = 0.0;
 262         size_t num_updates = 0;
 263
 264         deque<double> history;
 265         struct TreeNode {
 266                 double val;
 267                 size_t children = 0;
 268                 unique_ptr<TreeNode> left, right;
 269         };
 270         unique_ptr<TreeNode> root;
 271
 272         unique_ptr<TreeNode> alloc_cache;  // Holds the last freed value, for fast reallocation.
 273
 274         TreeNode *alloc_node()
 275         {
 276                 if (alloc_cache == nullptr) {
 277                         return new TreeNode;
 278                 }
 279                 alloc_cache->children = 0;
 280                 return alloc_cache.release();
 281         }
 282
 283         void insert(double val)
 284         {
 285                 if (root == nullptr) {
 286                         root.reset(alloc_node());
 287                         root->val = val;
 288                         return;
 289                 } else {
 290                         insert(root.get(), val);
 291                 }
 292         }
 293
 294         void insert(TreeNode *node, double val)
 295         {
 296                 ++node->children;
 297                 if (val <= node->val) {
 298                         // Goes into left.
 299                         if (node->left == nullptr) {
 300                                 node->left.reset(alloc_node());
 301                                 node->left->val = val;
 302                         } else {
 303                                 insert(node->left.get(), val);
 304                         }
 305                 } else {
 306                         // Goes into right.
 307                         if (node->right == nullptr) {
 308                                 node->right.reset(alloc_node());
 309                                 node->right->val = val;
 310                         } else {
 311                                 insert(node->right.get(), val);
 312                         }
 313                 }
 314         }
 315
 316         void remove(double val)
 317         {
 318                 assert(root != nullptr);
 319                 if (root->children == 0) {
 320                         assert(root->val == val);
 321                         alloc_cache = move(root);
 322                 } else {
 323                         remove(root.get(), val);
 324                 }
 325         }
 326
 327         void remove(TreeNode *node, double val)
 328         {
 329                 //printf("Down into %p looking for %f [left=%p right=%p]\n", node, val, node->left.get(), node->right.get());
 330                 if (node->val == val) {
 331                         remove(node);
 332                 } else if (val < node->val) {
 333                         assert(node->left != nullptr);
 334                         --node->children;
 335                         if (node->left->children == 0) {
 336                                 assert(node->left->val == val);
 337                                 alloc_cache = move(node->left);
 338                         } else {
 339                                 remove(node->left.get(), val);
 340                         }
 341                 } else {
 342                         assert(node->right != nullptr);
 343                         --node->children;
 344                         if (node->right->children == 0) {
 345                                 assert(node->right->val == val);
 346                                 alloc_cache = move(node->right);
 347                         } else {
 348                                 remove(node->right.get(), val);
 349                         }
 350                 }
 351         }
 352
 353         // Declares a node to be empty, so it should pull up the value of one of its children.
 354         // The node must be an interior node (ie., have at least one child).
 355         void remove(TreeNode *node)
 356         {
 357                 //printf("Decided that %p must be removed\n", node);
 358                 assert(node->children > 0);
 359                 --node->children;
 360
 361                 bool remove_left;
 362                 if (node->right == nullptr) {
 363                         remove_left = true;
 364                 } else if (node->left == nullptr) {
 365                         remove_left = false;
 366                 } else {
 367                         remove_left = (node->left->children >= node->right->children);
 368                 }
 369                 if (remove_left) {
 370                         if (node->left->children == 0) {
 371                                 node->val = node->left->val;
 372                                 alloc_cache = move(node->left);
 373                         } else {
 374                                 // Move maximum value up to this node.
 375                                 node->val = elem_at(node->left.get(), node->left->children);
 376                                 remove(node->left.get(), node->val);
 377                         }
 378                 } else {
 379                         if (node->right->children == 0) {
 380                                 node->val = node->right->val;
 381                                 alloc_cache = move(node->right);
 382                         } else {
 383                                 // Move minimum value up to this node.
 384                                 node->val = elem_at(node->right.get(), 0);
 385                                 remove(node->right.get(), node->val);
 386                         }
 387                 }
 388         }
 389
 390         double elem_at(size_t elem_idx)
 391         {
 392                 return elem_at(root.get(), elem_idx);
 393         }
 394
 395         double elem_at(TreeNode *node, size_t elem_idx)
 396         {
 397                 //printf("Looking for %lu in node %p [%lu children]\n", elem_idx, node, node->children);
 398                 assert(node != nullptr);
 399                 assert(elem_idx <= node->children);
 400                 if (node->left != nullptr) {
 401                         if (elem_idx <= node->left->children) {
 402                                 return elem_at(node->left.get(), elem_idx);
 403                         } else {
 404                                 elem_idx -= node->left->children + 1;
 405                         }
 406                 }
 407                 if (elem_idx == 0) {
 408                         return node->val;
 409                 }
 410                 return elem_at(node->right.get(), elem_idx - 1);
 411         }
 412
 413         void print_tree(TreeNode *node, size_t indent, double min, double max)
 414         {
 415                 if (node == nullptr) return;
 416                 if (!(node->val >= min && node->val <= max)) {
 417                         //printf("node %p is outside range [%f,%f]\n", node, min, max);
 418                         assert(false);
 419                 }
 420                 for (size_t i = 0; i < indent * 2; ++i) putchar(' ');
 421                 printf("%f [%p, %lu children]\n", node->val, node, node->children);
 422                 print_tree(node->left.get(), indent + 1, min, node->val);
 423                 print_tree(node->right.get(), indent + 1, node->val, max);
 424         }
 425
 426 public:
 427         HistoryJitter(size_t history_length, double multiplier, double percentile)
 428                 : history_length(history_length), multiplier(multiplier), percentile(percentile) {}
 429
 430         void update(double timestamp, double frame_duration, size_t dropped_frames)
 431         {
 432                 //if (++num_updates % 1000 == 0) {
 433                 //      printf("%d... [%lu in tree %p]\n", num_updates, root->children + 1, root.get());
 434                 //}
 435
 436                 if (expected_timestamp >= 0.0) {
 437                         expected_timestamp += dropped_frames * frame_duration;
 438                         double jitter_seconds = fabs(expected_timestamp - timestamp);
 439
 440                         history.push_back(jitter_seconds);
 441                         insert(jitter_seconds);
 442                         //printf("\nTree %p after insert of %f:\n", root.get(), jitter_seconds);
 443                         //print_tree(root.get(), 0, -HUGE_VAL, HUGE_VAL);
 444                         while (history.size() > history_length) {
 445                         //      printf("removing %f, because %p has %lu elements and history has %lu elements\n", history.front(), root.get(), root->children + 1, history.size());
 446                                 remove(history.front());
 447                                 history.pop_front();
 448                         }
 449
 450                         size_t elem_idx = lrint(percentile * (history.size() - 1));
 451 //                      printf("Searching for element %lu in %p, which has %lu elements (history has %lu elements)\n", elem_idx, root.get(), root->children + 1, history.size());
 452 //                      fflush(stdout);
 453 //
 454                         // Cap at 100 ms.
 455                         max_jitter_seconds = min(elem_at(elem_idx), 0.1);
 456                 }
 457                 expected_timestamp = timestamp + frame_duration;
 458         }
 459
 460         double get_expected() const
 461         {
 462                 return expected_timestamp;
 463         }
 464
 465         double get_jitter() const
 466         {
 467                 return max_jitter_seconds * multiplier;
 468         }
 469 };
 470
 471 void test_jitter_history(const vector<Event> &events, size_t history_length, double multiplier, double percentile, double margin)
 472 {
 473         Queue q;
 474         HistoryJitter input_jitter(history_length, multiplier, percentile);
 475         HistoryJitter output_jitter(history_length, multiplier, percentile);
 476
 477         for (const Event &event : events) {
 478                 if (event.direction == Event::IN) {
 479                         input_jitter.update(event.t, 0.020, 0);
 480                         q.add_frame(event.t);
 481                 } else {
 482                         double now = event.t;
 483                         output_jitter.update(event.t, 0.020, 0);
 484                         q.get_frame(event.t);
 485
 486                         double seconds_until_next_frame = max(input_jitter.get_expected() - now + input_jitter.get_jitter(), 0.0);
 487                         double master_frame_length_seconds = 0.020;
 488
 489                         seconds_until_next_frame += margin;  // Hack.
 490
 491                         size_t safe_queue_length = max<int>(floor((seconds_until_next_frame + output_jitter.get_jitter()) / master_frame_length_seconds), 0);
 492                         while (q.queue_len() > safe_queue_length) {
 493                                 q.drop_frame();
 494                         }
 495                 }
 496                 if (q.should_abort()) return;
 497         }
 498         char name[256];
 499         snprintf(name, sizeof(name), "history[len=%lu,mul=%.1f,pct=%.4f,margin=%.1f]", history_length, multiplier, percentile, 1e3 * margin);
 500         q.eval(name);
 501 }
 502
 503 int main(int argc, char **argv)
 504 {
 505         static const option long_options[] = {
 506                 { "max-drops", required_argument, 0, 'd' },
 507                 { "max-underruns", required_argument, 0, 'u' },
 508                 { "max-latency-ms", required_argument, 0, 'l' },
 509                 { 0, 0, 0, 0 }
 510         };
 511         for ( ;; ) {
 512                 int option_index = 0;
 513                 int c = getopt_long(argc, argv, "d:u:l:", long_options, &option_index);
 514
 515                 if (c == -1) {
 516                         break;
 517                 }
 518                 switch (c) {
 519                 case 'd':
 520                         max_drops = atof(optarg);
 521                         break;
 522                 case 'u':
 523                         max_underruns = atof(optarg);
 524                         break;
 525                 case 'l':
 526                         max_latency_ms = atof(optarg);
 527                         break;
 528                 default:
 529                         fprintf(stderr, "Usage: simul [--max-drops NUM] [--max-underruns NUM] [--max-latency-ms TIME]\n");
 530                         exit(1);
 531                 }
 532         }
 533
 534         vector<Event> events;
 535
 536         const char *filename = (optind < argc) ? argv[optind] : "nageru-latency-log.txt";
 537         FILE *fp = fopen(filename, "r");
 538         if (fp == nullptr) {
 539                 perror(filename);
 540                 exit(1);
 541         }
 542         while (!feof(fp)) {
 543                 char dir[256];
 544                 double t;
 545
 546                 if (fscanf(fp, "%s %lf", dir, &t) != 2) {
 547                         break;
 548                 }
 549                 if (dir[0] == 'I') {
 550                         events.push_back(Event{Event::IN, t});
 551                 } else if (dir[0] == 'O') {
 552                         events.push_back(Event{Event::OUT, t});
 553                 } else {
 554                         fprintf(stderr, "ERROR: Unreadable line\n");
 555                         exit(1);
 556                 }
 557         }
 558         fclose(fp);
 559
 560         sort(events.begin(), events.end(), [](const Event &a, const Event &b) { return a.t < b.t; });
 561
 562         test_nodrop(events);
 563         test_limit_to_1(events);
 564         test_limit_to_2(events);
 565         test_nageru_1_2_0(events);
 566         for (double multiplier : { 0.0, 0.5, 1.0, 2.0, 3.0, 5.0 }) {
 567                 for (double alpha : { 0.5, 0.9, 0.99, 0.995, 0.999, 0.9999 }) {
 568                         for (double margin_ms : { -1.0, 0.0, 1.0, 2.0, 5.0, 10.0, 20.0 }) {
 569                                 test_jitter_filter(events, multiplier, alpha, 1e-3 * margin_ms);
 570                         }
 571                 }
 572         }
 573         for (size_t history_samples : { 10, 100, 500, 1000, 5000, 10000, 25000 }) {
 574                 for (double multiplier : { 0.5, 1.0, 2.0, 3.0, 5.0, 10.0 }) {
 575                         for (double percentile : { 0.5, 0.75, 0.9, 0.99, 0.995, 0.999, 1.0 }) {
 576                                 if (lrint(percentile * (history_samples - 1)) == int(history_samples - 1) && percentile != 1.0) {
 577                                         // Redundant.
 578                                         continue;
 579                                 }
 580
 581                                 //for (double margin_ms : { -1.0, 0.0, 1.0, 2.0, 5.0, 10.0, 20.0 }) {
 582                                 for (double margin_ms : { 0.0 }) {
 583                                         test_jitter_history(events, history_samples, multiplier, percentile, 1e-3 * margin_ms);
 584                                 }
 585                         }
 586                 }
 587         }
 588 }