1#!/usr/bin/env python3 2# Copyright (c) Facebook, Inc. and its affiliates. All Rights Reserved. 3 4import gc 5import heapq 6import random 7import sys 8import time 9from collections import OrderedDict 10from os import path 11 12import numpy as np 13 14 15kSampleSize = 64 # The sample size used when performing eviction. 16kMicrosInSecond = 1000000 17kSecondsInMinute = 60 18kSecondsInHour = 3600 19 20 21class TraceRecord: 22 """ 23 A trace record represents a block access. 24 It holds the same struct as BlockCacheTraceRecord in 25 trace_replay/block_cache_tracer.h 26 """ 27 28 def __init__( 29 self, 30 access_time, 31 block_id, 32 block_type, 33 block_size, 34 cf_id, 35 cf_name, 36 level, 37 fd, 38 caller, 39 no_insert, 40 get_id, 41 key_id, 42 kv_size, 43 is_hit, 44 referenced_key_exist_in_block, 45 num_keys_in_block, 46 table_id, 47 seq_number, 48 block_key_size, 49 key_size, 50 block_offset_in_file, 51 next_access_seq_no, 52 ): 53 self.access_time = access_time 54 self.block_id = block_id 55 self.block_type = block_type 56 self.block_size = block_size + block_key_size 57 self.cf_id = cf_id 58 self.cf_name = cf_name 59 self.level = level 60 self.fd = fd 61 self.caller = caller 62 if no_insert == 1: 63 self.no_insert = True 64 else: 65 self.no_insert = False 66 self.get_id = get_id 67 self.key_id = key_id 68 self.kv_size = kv_size 69 if is_hit == 1: 70 self.is_hit = True 71 else: 72 self.is_hit = False 73 if referenced_key_exist_in_block == 1: 74 self.referenced_key_exist_in_block = True 75 else: 76 self.referenced_key_exist_in_block = False 77 self.num_keys_in_block = num_keys_in_block 78 self.table_id = table_id 79 self.seq_number = seq_number 80 self.block_key_size = block_key_size 81 self.key_size = key_size 82 self.block_offset_in_file = block_offset_in_file 83 self.next_access_seq_no = next_access_seq_no 84 85 86class CacheEntry: 87 """A cache entry stored in the cache.""" 88 89 def __init__( 90 self, 91 value_size, 92 cf_id, 93 level, 94 block_type, 95 table_id, 96 access_number, 97 time_s, 98 num_hits=0, 99 ): 100 self.value_size = value_size 101 self.last_access_number = access_number 102 self.num_hits = num_hits 103 self.cf_id = 0 104 self.level = level 105 self.block_type = block_type 106 self.last_access_time = time_s 107 self.insertion_time = time_s 108 self.table_id = table_id 109 110 def __repr__(self): 111 """Debug string.""" 112 return "(s={},last={},hits={},cf={},l={},bt={})\n".format( 113 self.value_size, 114 self.last_access_number, 115 self.num_hits, 116 self.cf_id, 117 self.level, 118 self.block_type, 119 ) 120 121 def cost_class(self, cost_class_label): 122 if cost_class_label == "table_bt": 123 return "{}-{}".format(self.table_id, self.block_type) 124 elif cost_class_label == "table": 125 return "{}".format(self.table_id) 126 elif cost_class_label == "bt": 127 return "{}".format(self.block_type) 128 elif cost_class_label == "cf": 129 return "{}".format(self.cf_id) 130 elif cost_class_label == "cf_bt": 131 return "{}-{}".format(self.cf_id, self.block_type) 132 elif cost_class_label == "table_level_bt": 133 return "{}-{}-{}".format(self.table_id, self.level, self.block_type) 134 assert False, "Unknown cost class label {}".format(cost_class_label) 135 return None 136 137 138class HashEntry: 139 """A hash entry stored in a hash table.""" 140 141 def __init__(self, key, hash, value): 142 self.key = key 143 self.hash = hash 144 self.value = value 145 146 def __repr__(self): 147 return "k={},h={},v=[{}]".format(self.key, self.hash, self.value) 148 149 150class HashTable: 151 """ 152 A custom implementation of hash table to support fast random sampling. 153 It is closed hashing and uses chaining to resolve hash conflicts. 154 It grows/shrinks the hash table upon insertion/deletion to support 155 fast lookups and random samplings. 156 """ 157 158 def __init__(self): 159 self.initial_size = 32 160 self.table = [None] * self.initial_size 161 self.elements = 0 162 163 def random_sample(self, sample_size): 164 """Randomly sample 'sample_size' hash entries from the table.""" 165 samples = [] 166 index = random.randint(0, len(self.table) - 1) 167 pos = index 168 # Starting from index, adding hash entries to the sample list until 169 # sample_size is met or we ran out of entries. 170 while True: 171 if self.table[pos] is not None: 172 for i in range(len(self.table[pos])): 173 if self.table[pos][i] is None: 174 continue 175 samples.append(self.table[pos][i]) 176 if len(samples) == sample_size: 177 break 178 pos += 1 179 pos = pos % len(self.table) 180 if pos == index or len(samples) == sample_size: 181 break 182 assert len(samples) <= sample_size 183 return samples 184 185 def __repr__(self): 186 all_entries = [] 187 for i in range(len(self.table)): 188 if self.table[i] is None: 189 continue 190 for j in range(len(self.table[i])): 191 if self.table[i][j] is not None: 192 all_entries.append(self.table[i][j]) 193 return "{}".format(all_entries) 194 195 def values(self): 196 all_values = [] 197 for i in range(len(self.table)): 198 if self.table[i] is None: 199 continue 200 for j in range(len(self.table[i])): 201 if self.table[i][j] is not None: 202 all_values.append(self.table[i][j].value) 203 return all_values 204 205 def __len__(self): 206 return self.elements 207 208 def insert(self, key, hash, value): 209 """ 210 Insert a hash entry in the table. Replace the old entry if it already 211 exists. 212 """ 213 self.grow() 214 inserted = False 215 index = hash % len(self.table) 216 if self.table[index] is None: 217 self.table[index] = [] 218 # Search for the entry first. 219 for i in range(len(self.table[index])): 220 if self.table[index][i] is None: 221 continue 222 if self.table[index][i].hash == hash and self.table[index][i].key == key: 223 # The entry already exists in the table. 224 self.table[index][i] = HashEntry(key, hash, value) 225 return 226 227 # Find an empty slot. 228 for i in range(len(self.table[index])): 229 if self.table[index][i] is None: 230 self.table[index][i] = HashEntry(key, hash, value) 231 inserted = True 232 break 233 if not inserted: 234 self.table[index].append(HashEntry(key, hash, value)) 235 self.elements += 1 236 237 def resize(self, new_size): 238 if new_size == len(self.table): 239 return 240 if new_size < self.initial_size: 241 return 242 if self.elements < 100: 243 return 244 new_table = [None] * new_size 245 # Copy 'self.table' to new_table. 246 for i in range(len(self.table)): 247 entries = self.table[i] 248 if entries is None: 249 continue 250 for j in range(len(entries)): 251 if entries[j] is None: 252 continue 253 index = entries[j].hash % new_size 254 if new_table[index] is None: 255 new_table[index] = [] 256 new_table[index].append(entries[j]) 257 self.table = new_table 258 del new_table 259 # Manually call python gc here to free the memory as 'self.table' 260 # might be very large. 261 gc.collect() 262 263 def grow(self): 264 if self.elements < 4 * len(self.table): 265 return 266 new_size = int(len(self.table) * 1.5) 267 self.resize(new_size) 268 269 def delete(self, key, hash): 270 index = hash % len(self.table) 271 deleted = False 272 deleted_entry = None 273 if self.table[index] is None: 274 return 275 for i in range(len(self.table[index])): 276 if ( 277 self.table[index][i] is not None 278 and self.table[index][i].hash == hash 279 and self.table[index][i].key == key 280 ): 281 deleted_entry = self.table[index][i] 282 self.table[index][i] = None 283 self.elements -= 1 284 deleted = True 285 break 286 if deleted: 287 self.shrink() 288 return deleted_entry 289 290 def shrink(self): 291 if self.elements * 2 >= len(self.table): 292 return 293 new_size = int(len(self.table) * 0.7) 294 self.resize(new_size) 295 296 def lookup(self, key, hash): 297 index = hash % len(self.table) 298 if self.table[index] is None: 299 return None 300 for i in range(len(self.table[index])): 301 if ( 302 self.table[index][i] is not None 303 and self.table[index][i].hash == hash 304 and self.table[index][i].key == key 305 ): 306 return self.table[index][i].value 307 return None 308 309 310class MissRatioStats: 311 def __init__(self, time_unit): 312 self.num_misses = 0 313 self.num_accesses = 0 314 self.time_unit = time_unit 315 self.time_misses = {} 316 self.time_miss_bytes = {} 317 self.time_accesses = {} 318 319 def update_metrics(self, access_time, is_hit, miss_bytes): 320 access_time /= kMicrosInSecond * self.time_unit 321 self.num_accesses += 1 322 if access_time not in self.time_accesses: 323 self.time_accesses[access_time] = 0 324 self.time_accesses[access_time] += 1 325 if not is_hit: 326 self.num_misses += 1 327 if access_time not in self.time_misses: 328 self.time_misses[access_time] = 0 329 self.time_miss_bytes[access_time] = 0 330 self.time_misses[access_time] += 1 331 self.time_miss_bytes[access_time] += miss_bytes 332 333 def reset_counter(self): 334 self.num_misses = 0 335 self.num_accesses = 0 336 self.time_miss_bytes.clear() 337 self.time_misses.clear() 338 self.time_accesses.clear() 339 340 def compute_miss_bytes(self): 341 miss_bytes = [] 342 for at in self.time_miss_bytes: 343 miss_bytes.append(self.time_miss_bytes[at]) 344 miss_bytes = sorted(miss_bytes) 345 avg_miss_bytes = 0 346 p95_miss_bytes = 0 347 for i in range(len(miss_bytes)): 348 avg_miss_bytes += float(miss_bytes[i]) / float(len(miss_bytes)) 349 350 p95_index = min(int(0.95 * float(len(miss_bytes))), len(miss_bytes) - 1) 351 p95_miss_bytes = miss_bytes[p95_index] 352 return avg_miss_bytes, p95_miss_bytes 353 354 def miss_ratio(self): 355 return float(self.num_misses) * 100.0 / float(self.num_accesses) 356 357 def write_miss_timeline( 358 self, cache_type, cache_size, target_cf_name, result_dir, start, end 359 ): 360 start /= kMicrosInSecond * self.time_unit 361 end /= kMicrosInSecond * self.time_unit 362 header_file_path = "{}/header-ml-miss-timeline-{}-{}-{}-{}".format( 363 result_dir, self.time_unit, cache_type, cache_size, target_cf_name 364 ) 365 if not path.exists(header_file_path): 366 with open(header_file_path, "w+") as header_file: 367 header = "time" 368 for trace_time in range(start, end): 369 header += ",{}".format(trace_time) 370 header_file.write(header + "\n") 371 file_path = "{}/data-ml-miss-timeline-{}-{}-{}-{}".format( 372 result_dir, self.time_unit, cache_type, cache_size, target_cf_name 373 ) 374 with open(file_path, "w+") as file: 375 row = "{}".format(cache_type) 376 for trace_time in range(start, end): 377 row += ",{}".format(self.time_misses.get(trace_time, 0)) 378 file.write(row + "\n") 379 380 def write_miss_ratio_timeline( 381 self, cache_type, cache_size, target_cf_name, result_dir, start, end 382 ): 383 start /= kMicrosInSecond * self.time_unit 384 end /= kMicrosInSecond * self.time_unit 385 header_file_path = "{}/header-ml-miss-ratio-timeline-{}-{}-{}-{}".format( 386 result_dir, self.time_unit, cache_type, cache_size, target_cf_name 387 ) 388 if not path.exists(header_file_path): 389 with open(header_file_path, "w+") as header_file: 390 header = "time" 391 for trace_time in range(start, end): 392 header += ",{}".format(trace_time) 393 header_file.write(header + "\n") 394 file_path = "{}/data-ml-miss-ratio-timeline-{}-{}-{}-{}".format( 395 result_dir, self.time_unit, cache_type, cache_size, target_cf_name 396 ) 397 with open(file_path, "w+") as file: 398 row = "{}".format(cache_type) 399 for trace_time in range(start, end): 400 naccesses = self.time_accesses.get(trace_time, 0) 401 miss_ratio = 0 402 if naccesses > 0: 403 miss_ratio = float( 404 self.time_misses.get(trace_time, 0) * 100.0 405 ) / float(naccesses) 406 row += ",{0:.2f}".format(miss_ratio) 407 file.write(row + "\n") 408 409 410class PolicyStats: 411 def __init__(self, time_unit, policies): 412 self.time_selected_polices = {} 413 self.time_accesses = {} 414 self.policy_names = {} 415 self.time_unit = time_unit 416 for i in range(len(policies)): 417 self.policy_names[i] = policies[i].policy_name() 418 419 def update_metrics(self, access_time, selected_policy): 420 access_time /= kMicrosInSecond * self.time_unit 421 if access_time not in self.time_accesses: 422 self.time_accesses[access_time] = 0 423 self.time_accesses[access_time] += 1 424 if access_time not in self.time_selected_polices: 425 self.time_selected_polices[access_time] = {} 426 policy_name = self.policy_names[selected_policy] 427 if policy_name not in self.time_selected_polices[access_time]: 428 self.time_selected_polices[access_time][policy_name] = 0 429 self.time_selected_polices[access_time][policy_name] += 1 430 431 def write_policy_timeline( 432 self, cache_type, cache_size, target_cf_name, result_dir, start, end 433 ): 434 start /= kMicrosInSecond * self.time_unit 435 end /= kMicrosInSecond * self.time_unit 436 header_file_path = "{}/header-ml-policy-timeline-{}-{}-{}-{}".format( 437 result_dir, self.time_unit, cache_type, cache_size, target_cf_name 438 ) 439 if not path.exists(header_file_path): 440 with open(header_file_path, "w+") as header_file: 441 header = "time" 442 for trace_time in range(start, end): 443 header += ",{}".format(trace_time) 444 header_file.write(header + "\n") 445 file_path = "{}/data-ml-policy-timeline-{}-{}-{}-{}".format( 446 result_dir, self.time_unit, cache_type, cache_size, target_cf_name 447 ) 448 with open(file_path, "w+") as file: 449 for policy in self.policy_names: 450 policy_name = self.policy_names[policy] 451 row = "{}-{}".format(cache_type, policy_name) 452 for trace_time in range(start, end): 453 row += ",{}".format( 454 self.time_selected_polices.get(trace_time, {}).get( 455 policy_name, 0 456 ) 457 ) 458 file.write(row + "\n") 459 460 def write_policy_ratio_timeline( 461 self, cache_type, cache_size, target_cf_name, file_path, start, end 462 ): 463 start /= kMicrosInSecond * self.time_unit 464 end /= kMicrosInSecond * self.time_unit 465 header_file_path = "{}/header-ml-policy-ratio-timeline-{}-{}-{}-{}".format( 466 result_dir, self.time_unit, cache_type, cache_size, target_cf_name 467 ) 468 if not path.exists(header_file_path): 469 with open(header_file_path, "w+") as header_file: 470 header = "time" 471 for trace_time in range(start, end): 472 header += ",{}".format(trace_time) 473 header_file.write(header + "\n") 474 file_path = "{}/data-ml-policy-ratio-timeline-{}-{}-{}-{}".format( 475 result_dir, self.time_unit, cache_type, cache_size, target_cf_name 476 ) 477 with open(file_path, "w+") as file: 478 for policy in self.policy_names: 479 policy_name = self.policy_names[policy] 480 row = "{}-{}".format(cache_type, policy_name) 481 for trace_time in range(start, end): 482 naccesses = self.time_accesses.get(trace_time, 0) 483 ratio = 0 484 if naccesses > 0: 485 ratio = float( 486 self.time_selected_polices.get(trace_time, {}).get( 487 policy_name, 0 488 ) 489 * 100.0 490 ) / float(naccesses) 491 row += ",{0:.2f}".format(ratio) 492 file.write(row + "\n") 493 494 495class Policy(object): 496 """ 497 A policy maintains a set of evicted keys. It returns a reward of one to 498 itself if it has not evicted a missing key. Otherwise, it gives itself 0 499 reward. 500 """ 501 502 def __init__(self): 503 self.evicted_keys = {} 504 505 def evict(self, key, max_size): 506 self.evicted_keys[key] = 0 507 508 def delete(self, key): 509 self.evicted_keys.pop(key, None) 510 511 def prioritize_samples(self, samples, auxilliary_info): 512 raise NotImplementedError 513 514 def policy_name(self): 515 raise NotImplementedError 516 517 def generate_reward(self, key): 518 if key in self.evicted_keys: 519 return 0 520 return 1 521 522 523class LRUPolicy(Policy): 524 def prioritize_samples(self, samples, auxilliary_info): 525 return sorted( 526 samples, 527 cmp=lambda e1, e2: e1.value.last_access_number 528 - e2.value.last_access_number, 529 ) 530 531 def policy_name(self): 532 return "lru" 533 534 535class MRUPolicy(Policy): 536 def prioritize_samples(self, samples, auxilliary_info): 537 return sorted( 538 samples, 539 cmp=lambda e1, e2: e2.value.last_access_number 540 - e1.value.last_access_number, 541 ) 542 543 def policy_name(self): 544 return "mru" 545 546 547class LFUPolicy(Policy): 548 def prioritize_samples(self, samples, auxilliary_info): 549 return sorted(samples, cmp=lambda e1, e2: e1.value.num_hits - e2.value.num_hits) 550 551 def policy_name(self): 552 return "lfu" 553 554 555class HyperbolicPolicy(Policy): 556 """ 557 An implementation of Hyperbolic caching. 558 559 Aaron Blankstein, Siddhartha Sen, and Michael J. Freedman. 2017. 560 Hyperbolic caching: flexible caching for web applications. In Proceedings 561 of the 2017 USENIX Conference on Usenix Annual Technical Conference 562 (USENIX ATC '17). USENIX Association, Berkeley, CA, USA, 499-511. 563 """ 564 565 def compare(self, e1, e2, now): 566 e1_duration = max(0, (now - e1.value.insertion_time) / kMicrosInSecond) * float( 567 e1.value.value_size 568 ) 569 e2_duration = max(0, (now - e2.value.insertion_time) / kMicrosInSecond) * float( 570 e2.value.value_size 571 ) 572 if e1_duration == e2_duration: 573 return e1.value.num_hits - e2.value.num_hits 574 if e1_duration == 0: 575 return 1 576 if e2_duration == 0: 577 return 1 578 diff = (float(e1.value.num_hits) / (float(e1_duration))) - ( 579 float(e2.value.num_hits) / float(e2_duration) 580 ) 581 if diff == 0: 582 return 0 583 elif diff > 0: 584 return 1 585 else: 586 return -1 587 588 def prioritize_samples(self, samples, auxilliary_info): 589 assert len(auxilliary_info) == 3 590 now = auxilliary_info[0] 591 return sorted(samples, cmp=lambda e1, e2: self.compare(e1, e2, now)) 592 593 def policy_name(self): 594 return "hb" 595 596 597class CostClassPolicy(Policy): 598 """ 599 We calculate the hit density of a cost class as 600 number of hits / total size in cache * average duration in the cache. 601 602 An entry has a higher priority if its class's hit density is higher. 603 """ 604 605 def compare(self, e1, e2, now, cost_classes, cost_class_label): 606 e1_class = e1.value.cost_class(cost_class_label) 607 e2_class = e2.value.cost_class(cost_class_label) 608 609 assert e1_class in cost_classes 610 assert e2_class in cost_classes 611 612 e1_entry = cost_classes[e1_class] 613 e2_entry = cost_classes[e2_class] 614 e1_density = e1_entry.density(now) 615 e2_density = e2_entry.density(now) 616 e1_hits = cost_classes[e1_class].hits 617 e2_hits = cost_classes[e2_class].hits 618 619 if e1_density == e2_density: 620 return e1_hits - e2_hits 621 622 if e1_entry.num_entries_in_cache == 0: 623 return -1 624 if e2_entry.num_entries_in_cache == 0: 625 return 1 626 627 if e1_density == 0: 628 return 1 629 if e2_density == 0: 630 return -1 631 diff = (float(e1_hits) / float(e1_density)) - ( 632 float(e2_hits) / float(e2_density) 633 ) 634 if diff == 0: 635 return 0 636 elif diff > 0: 637 return 1 638 else: 639 return -1 640 641 def prioritize_samples(self, samples, auxilliary_info): 642 assert len(auxilliary_info) == 3 643 now = auxilliary_info[0] 644 cost_classes = auxilliary_info[1] 645 cost_class_label = auxilliary_info[2] 646 return sorted( 647 samples, 648 cmp=lambda e1, e2: self.compare( 649 e1, e2, now, cost_classes, cost_class_label 650 ), 651 ) 652 653 def policy_name(self): 654 return "cc" 655 656 657class Cache(object): 658 """ 659 This is the base class for the implementations of alternative cache 660 replacement policies. 661 """ 662 663 def __init__(self, cache_size, enable_cache_row_key): 664 self.cache_size = cache_size 665 self.used_size = 0 666 self.per_second_miss_ratio_stats = MissRatioStats(1) 667 self.miss_ratio_stats = MissRatioStats(kSecondsInMinute) 668 self.per_hour_miss_ratio_stats = MissRatioStats(kSecondsInHour) 669 # 0: disabled. 1: enabled. Insert both row and the refereneced data block. 670 # 2: enabled. Insert only the row but NOT the referenced data block. 671 self.enable_cache_row_key = enable_cache_row_key 672 self.get_id_row_key_map = {} 673 self.max_seen_get_id = 0 674 self.retain_get_id_range = 100000 675 676 def block_key(self, trace_record): 677 return "b{}".format(trace_record.block_id) 678 679 def row_key(self, trace_record): 680 return "g{}-{}".format(trace_record.fd, trace_record.key_id) 681 682 def _lookup(self, trace_record, key, hash): 683 """ 684 Look up the key in the cache. 685 Returns true upon a cache hit, false otherwise. 686 """ 687 raise NotImplementedError 688 689 def _evict(self, trace_record, key, hash, value_size): 690 """ 691 Evict entries in the cache until there is enough room to insert the new 692 entry with 'value_size'. 693 """ 694 raise NotImplementedError 695 696 def _insert(self, trace_record, key, hash, value_size): 697 """ 698 Insert the new entry into the cache. 699 """ 700 raise NotImplementedError 701 702 def _should_admit(self, trace_record, key, hash, value_size): 703 """ 704 A custom admission policy to decide whether we should admit the new 705 entry upon a cache miss. 706 Returns true if the new entry should be admitted, false otherwise. 707 """ 708 raise NotImplementedError 709 710 def cache_name(self): 711 """ 712 The name of the replacement policy. 713 """ 714 raise NotImplementedError 715 716 def is_ml_cache(self): 717 return False 718 719 def _update_stats(self, access_time, is_hit, miss_bytes): 720 self.per_second_miss_ratio_stats.update_metrics(access_time, is_hit, miss_bytes) 721 self.miss_ratio_stats.update_metrics(access_time, is_hit, miss_bytes) 722 self.per_hour_miss_ratio_stats.update_metrics(access_time, is_hit, miss_bytes) 723 724 def access(self, trace_record): 725 """ 726 Access a trace record. The simulator calls this function to access a 727 trace record. 728 """ 729 assert self.used_size <= self.cache_size 730 if ( 731 self.enable_cache_row_key > 0 732 and trace_record.caller == 1 733 and trace_record.key_id != 0 734 and trace_record.get_id != 0 735 ): 736 # This is a get request. 737 self._access_row(trace_record) 738 return 739 is_hit = self._access_kv( 740 trace_record, 741 self.block_key(trace_record), 742 trace_record.block_id, 743 trace_record.block_size, 744 trace_record.no_insert, 745 ) 746 self._update_stats( 747 trace_record.access_time, is_hit=is_hit, miss_bytes=trace_record.block_size 748 ) 749 750 def _access_row(self, trace_record): 751 row_key = self.row_key(trace_record) 752 self.max_seen_get_id = max(self.max_seen_get_id, trace_record.get_id) 753 self.get_id_row_key_map.pop( 754 self.max_seen_get_id - self.retain_get_id_range, None 755 ) 756 if trace_record.get_id not in self.get_id_row_key_map: 757 self.get_id_row_key_map[trace_record.get_id] = {} 758 self.get_id_row_key_map[trace_record.get_id]["h"] = False 759 if self.get_id_row_key_map[trace_record.get_id]["h"]: 760 # We treat future accesses as hits since this get request 761 # completes. 762 # print("row hit 1") 763 self._update_stats(trace_record.access_time, is_hit=True, miss_bytes=0) 764 return 765 if row_key not in self.get_id_row_key_map[trace_record.get_id]: 766 # First time seen this key. 767 is_hit = self._access_kv( 768 trace_record, 769 key=row_key, 770 hash=trace_record.key_id, 771 value_size=trace_record.kv_size, 772 no_insert=False, 773 ) 774 inserted = False 775 if trace_record.kv_size > 0: 776 inserted = True 777 self.get_id_row_key_map[trace_record.get_id][row_key] = inserted 778 self.get_id_row_key_map[trace_record.get_id]["h"] = is_hit 779 if self.get_id_row_key_map[trace_record.get_id]["h"]: 780 # We treat future accesses as hits since this get request 781 # completes. 782 # print("row hit 2") 783 self._update_stats(trace_record.access_time, is_hit=True, miss_bytes=0) 784 return 785 # Access its blocks. 786 no_insert = trace_record.no_insert 787 if ( 788 self.enable_cache_row_key == 2 789 and trace_record.kv_size > 0 790 and trace_record.block_type == 9 791 ): 792 no_insert = True 793 is_hit = self._access_kv( 794 trace_record, 795 key=self.block_key(trace_record), 796 hash=trace_record.block_id, 797 value_size=trace_record.block_size, 798 no_insert=no_insert, 799 ) 800 self._update_stats( 801 trace_record.access_time, is_hit, miss_bytes=trace_record.block_size 802 ) 803 if ( 804 trace_record.kv_size > 0 805 and not self.get_id_row_key_map[trace_record.get_id][row_key] 806 ): 807 # Insert the row key-value pair. 808 self._access_kv( 809 trace_record, 810 key=row_key, 811 hash=trace_record.key_id, 812 value_size=trace_record.kv_size, 813 no_insert=False, 814 ) 815 # Mark as inserted. 816 self.get_id_row_key_map[trace_record.get_id][row_key] = True 817 818 def _access_kv(self, trace_record, key, hash, value_size, no_insert): 819 # Sanity checks. 820 assert self.used_size <= self.cache_size 821 if self._lookup(trace_record, key, hash): 822 # A cache hit. 823 return True 824 if no_insert or value_size <= 0: 825 return False 826 # A cache miss. 827 if value_size > self.cache_size: 828 # The block is too large to fit into the cache. 829 return False 830 self._evict(trace_record, key, hash, value_size) 831 if self._should_admit(trace_record, key, hash, value_size): 832 self._insert(trace_record, key, hash, value_size) 833 self.used_size += value_size 834 return False 835 836 837class CostClassEntry: 838 """ 839 A cost class maintains aggregated statistics of cached entries in a class. 840 For example, we may define block type as a class. Then, cached blocks of the 841 same type will share one cost class entry. 842 """ 843 844 def __init__(self): 845 self.hits = 0 846 self.num_entries_in_cache = 0 847 self.size_in_cache = 0 848 self.sum_insertion_times = 0 849 self.sum_last_access_time = 0 850 851 def insert(self, trace_record, key, value_size): 852 self.size_in_cache += value_size 853 self.num_entries_in_cache += 1 854 self.sum_insertion_times += trace_record.access_time / kMicrosInSecond 855 self.sum_last_access_time += trace_record.access_time / kMicrosInSecond 856 857 def remove(self, insertion_time, last_access_time, key, value_size, num_hits): 858 self.hits -= num_hits 859 self.num_entries_in_cache -= 1 860 self.sum_insertion_times -= insertion_time / kMicrosInSecond 861 self.size_in_cache -= value_size 862 self.sum_last_access_time -= last_access_time / kMicrosInSecond 863 864 def update_on_hit(self, trace_record, last_access_time): 865 self.hits += 1 866 self.sum_last_access_time -= last_access_time / kMicrosInSecond 867 self.sum_last_access_time += trace_record.access_time / kMicrosInSecond 868 869 def avg_lifetime_in_cache(self, now): 870 avg_insertion_time = self.sum_insertion_times / self.num_entries_in_cache 871 return now / kMicrosInSecond - avg_insertion_time 872 873 def avg_last_access_time(self): 874 if self.num_entries_in_cache == 0: 875 return 0 876 return float(self.sum_last_access_time) / float(self.num_entries_in_cache) 877 878 def avg_size(self): 879 if self.num_entries_in_cache == 0: 880 return 0 881 return float(self.sum_last_access_time) / float(self.num_entries_in_cache) 882 883 def density(self, now): 884 avg_insertion_time = self.sum_insertion_times / self.num_entries_in_cache 885 in_cache_duration = now / kMicrosInSecond - avg_insertion_time 886 return self.size_in_cache * in_cache_duration 887 888 889class MLCache(Cache): 890 """ 891 MLCache is the base class for implementations of alternative replacement 892 policies using reinforcement learning. 893 """ 894 895 def __init__(self, cache_size, enable_cache_row_key, policies, cost_class_label): 896 super(MLCache, self).__init__(cache_size, enable_cache_row_key) 897 self.table = HashTable() 898 self.policy_stats = PolicyStats(kSecondsInMinute, policies) 899 self.per_hour_policy_stats = PolicyStats(kSecondsInHour, policies) 900 self.policies = policies 901 self.cost_classes = {} 902 self.cost_class_label = cost_class_label 903 904 def is_ml_cache(self): 905 return True 906 907 def _lookup(self, trace_record, key, hash): 908 value = self.table.lookup(key, hash) 909 if value is not None: 910 # Update the entry's cost class statistics. 911 if self.cost_class_label is not None: 912 cost_class = value.cost_class(self.cost_class_label) 913 assert cost_class in self.cost_classes 914 self.cost_classes[cost_class].update_on_hit( 915 trace_record, value.last_access_time 916 ) 917 # Update the entry's last access time. 918 self.table.insert( 919 key, 920 hash, 921 CacheEntry( 922 value_size=value.value_size, 923 cf_id=value.cf_id, 924 level=value.level, 925 block_type=value.block_type, 926 table_id=value.table_id, 927 access_number=self.miss_ratio_stats.num_accesses, 928 time_s=trace_record.access_time, 929 num_hits=value.num_hits + 1, 930 ), 931 ) 932 return True 933 return False 934 935 def _evict(self, trace_record, key, hash, value_size): 936 # Select a policy, random sample kSampleSize keys from the cache, then 937 # evict keys in the sample set until we have enough room for the new 938 # entry. 939 policy_index = self._select_policy(trace_record, key) 940 assert policy_index < len(self.policies) and policy_index >= 0 941 self.policies[policy_index].delete(key) 942 self.policy_stats.update_metrics(trace_record.access_time, policy_index) 943 self.per_hour_policy_stats.update_metrics( 944 trace_record.access_time, policy_index 945 ) 946 while self.used_size + value_size > self.cache_size: 947 # Randomly sample n entries. 948 samples = self.table.random_sample(kSampleSize) 949 samples = self.policies[policy_index].prioritize_samples( 950 samples, 951 [trace_record.access_time, self.cost_classes, self.cost_class_label], 952 ) 953 for hash_entry in samples: 954 assert self.table.delete(hash_entry.key, hash_entry.hash) is not None 955 self.used_size -= hash_entry.value.value_size 956 self.policies[policy_index].evict( 957 key=hash_entry.key, max_size=self.table.elements 958 ) 959 # Update the entry's cost class statistics. 960 if self.cost_class_label is not None: 961 cost_class = hash_entry.value.cost_class(self.cost_class_label) 962 assert cost_class in self.cost_classes 963 self.cost_classes[cost_class].remove( 964 hash_entry.value.insertion_time, 965 hash_entry.value.last_access_time, 966 key, 967 hash_entry.value.value_size, 968 hash_entry.value.num_hits, 969 ) 970 if self.used_size + value_size <= self.cache_size: 971 break 972 973 def _insert(self, trace_record, key, hash, value_size): 974 assert self.used_size + value_size <= self.cache_size 975 entry = CacheEntry( 976 value_size, 977 trace_record.cf_id, 978 trace_record.level, 979 trace_record.block_type, 980 trace_record.table_id, 981 self.miss_ratio_stats.num_accesses, 982 trace_record.access_time, 983 ) 984 # Update the entry's cost class statistics. 985 if self.cost_class_label is not None: 986 cost_class = entry.cost_class(self.cost_class_label) 987 if cost_class not in self.cost_classes: 988 self.cost_classes[cost_class] = CostClassEntry() 989 self.cost_classes[cost_class].insert(trace_record, key, value_size) 990 self.table.insert(key, hash, entry) 991 992 def _should_admit(self, trace_record, key, hash, value_size): 993 return True 994 995 def _select_policy(self, trace_record, key): 996 raise NotImplementedError 997 998 999class ThompsonSamplingCache(MLCache): 1000 """ 1001 An implementation of Thompson Sampling for the Bernoulli Bandit. 1002 1003 Daniel J. Russo, Benjamin Van Roy, Abbas Kazerouni, Ian Osband, 1004 and Zheng Wen. 2018. A Tutorial on Thompson Sampling. Found. 1005 Trends Mach. Learn. 11, 1 (July 2018), 1-96. 1006 DOI: https://doi.org/10.1561/2200000070 1007 """ 1008 1009 def __init__( 1010 self, 1011 cache_size, 1012 enable_cache_row_key, 1013 policies, 1014 cost_class_label, 1015 init_a=1, 1016 init_b=1, 1017 ): 1018 super(ThompsonSamplingCache, self).__init__( 1019 cache_size, enable_cache_row_key, policies, cost_class_label 1020 ) 1021 self._as = {} 1022 self._bs = {} 1023 for _i in range(len(policies)): 1024 self._as = [init_a] * len(self.policies) 1025 self._bs = [init_b] * len(self.policies) 1026 1027 def _select_policy(self, trace_record, key): 1028 if len(self.policies) == 1: 1029 return 0 1030 samples = [ 1031 np.random.beta(self._as[x], self._bs[x]) for x in range(len(self.policies)) 1032 ] 1033 selected_policy = max(range(len(self.policies)), key=lambda x: samples[x]) 1034 reward = self.policies[selected_policy].generate_reward(key) 1035 assert reward <= 1 and reward >= 0 1036 self._as[selected_policy] += reward 1037 self._bs[selected_policy] += 1 - reward 1038 return selected_policy 1039 1040 def cache_name(self): 1041 if self.enable_cache_row_key: 1042 return "Hybrid ThompsonSampling with cost class {} (ts_hybrid)".format( 1043 self.cost_class_label 1044 ) 1045 return "ThompsonSampling with cost class {} (ts)".format(self.cost_class_label) 1046 1047 1048class LinUCBCache(MLCache): 1049 """ 1050 An implementation of LinUCB with disjoint linear models. 1051 1052 Lihong Li, Wei Chu, John Langford, and Robert E. Schapire. 2010. 1053 A contextual-bandit approach to personalized news article recommendation. 1054 In Proceedings of the 19th international conference on World wide web 1055 (WWW '10). ACM, New York, NY, USA, 661-670. 1056 DOI=http://dx.doi.org/10.1145/1772690.1772758 1057 """ 1058 1059 def __init__(self, cache_size, enable_cache_row_key, policies, cost_class_label): 1060 super(LinUCBCache, self).__init__( 1061 cache_size, enable_cache_row_key, policies, cost_class_label 1062 ) 1063 self.nfeatures = 4 # Block type, level, cf. 1064 self.th = np.zeros((len(self.policies), self.nfeatures)) 1065 self.eps = 0.2 1066 self.b = np.zeros_like(self.th) 1067 self.A = np.zeros((len(self.policies), self.nfeatures, self.nfeatures)) 1068 self.A_inv = np.zeros((len(self.policies), self.nfeatures, self.nfeatures)) 1069 for i in range(len(self.policies)): 1070 self.A[i] = np.identity(self.nfeatures) 1071 self.th_hat = np.zeros_like(self.th) 1072 self.p = np.zeros(len(self.policies)) 1073 self.alph = 0.2 1074 1075 def _select_policy(self, trace_record, key): 1076 if len(self.policies) == 1: 1077 return 0 1078 x_i = np.zeros(self.nfeatures) # The current context vector 1079 x_i[0] = trace_record.block_type 1080 x_i[1] = trace_record.level 1081 x_i[2] = trace_record.cf_id 1082 p = np.zeros(len(self.policies)) 1083 for a in range(len(self.policies)): 1084 self.th_hat[a] = self.A_inv[a].dot(self.b[a]) 1085 ta = x_i.dot(self.A_inv[a]).dot(x_i) 1086 a_upper_ci = self.alph * np.sqrt(ta) 1087 a_mean = self.th_hat[a].dot(x_i) 1088 p[a] = a_mean + a_upper_ci 1089 p = p + (np.random.random(len(p)) * 0.000001) 1090 selected_policy = p.argmax() 1091 reward = self.policies[selected_policy].generate_reward(key) 1092 assert reward <= 1 and reward >= 0 1093 self.A[selected_policy] += np.outer(x_i, x_i) 1094 self.b[selected_policy] += reward * x_i 1095 self.A_inv[selected_policy] = np.linalg.inv(self.A[selected_policy]) 1096 del x_i 1097 return selected_policy 1098 1099 def cache_name(self): 1100 if self.enable_cache_row_key: 1101 return "Hybrid LinUCB with cost class {} (linucb_hybrid)".format( 1102 self.cost_class_label 1103 ) 1104 return "LinUCB with cost class {} (linucb)".format(self.cost_class_label) 1105 1106 1107class OPTCacheEntry: 1108 """ 1109 A cache entry for the OPT algorithm. The entries are sorted based on its 1110 next access sequence number in reverse order, i.e., the entry which next 1111 access is the furthest in the future is ordered before other entries. 1112 """ 1113 1114 def __init__(self, key, next_access_seq_no, value_size): 1115 self.key = key 1116 self.next_access_seq_no = next_access_seq_no 1117 self.value_size = value_size 1118 self.is_removed = False 1119 1120 def __cmp__(self, other): 1121 if other.next_access_seq_no != self.next_access_seq_no: 1122 return other.next_access_seq_no - self.next_access_seq_no 1123 return self.value_size - other.value_size 1124 1125 def __repr__(self): 1126 return "({} {} {} {})".format( 1127 self.key, self.next_access_seq_no, self.value_size, self.is_removed 1128 ) 1129 1130 1131class PQTable: 1132 """ 1133 A hash table with a priority queue. 1134 """ 1135 1136 def __init__(self): 1137 # A list of entries arranged in a heap sorted based on the entry custom 1138 # implementation of __cmp__ 1139 self.pq = [] 1140 self.table = {} 1141 1142 def pqinsert(self, entry): 1143 "Add a new key or update the priority of an existing key" 1144 # Remove the entry from the table first. 1145 removed_entry = self.table.pop(entry.key, None) 1146 if removed_entry: 1147 # Mark as removed since there is no 'remove' API in heappq. 1148 # Instead, an entry in pq is removed lazily when calling pop. 1149 removed_entry.is_removed = True 1150 self.table[entry.key] = entry 1151 heapq.heappush(self.pq, entry) 1152 return removed_entry 1153 1154 def pqpop(self): 1155 while self.pq: 1156 entry = heapq.heappop(self.pq) 1157 if not entry.is_removed: 1158 del self.table[entry.key] 1159 return entry 1160 return None 1161 1162 def pqpeek(self): 1163 while self.pq: 1164 entry = self.pq[0] 1165 if not entry.is_removed: 1166 return entry 1167 heapq.heappop(self.pq) 1168 return 1169 1170 def __contains__(self, k): 1171 return k in self.table 1172 1173 def __getitem__(self, k): 1174 return self.table[k] 1175 1176 def __len__(self): 1177 return len(self.table) 1178 1179 def values(self): 1180 return self.table.values() 1181 1182 1183class OPTCache(Cache): 1184 """ 1185 An implementation of the Belady MIN algorithm. OPTCache evicts an entry 1186 in the cache whose next access occurs furthest in the future. 1187 1188 Note that Belady MIN algorithm is optimal assuming all blocks having the 1189 same size and a missing entry will be inserted in the cache. 1190 These are NOT true for the block cache trace since blocks have different 1191 sizes and we may not insert a block into the cache upon a cache miss. 1192 However, it is still useful to serve as a "theoretical upper bound" on the 1193 lowest miss ratio we can achieve given a cache size. 1194 1195 L. A. Belady. 1966. A Study of Replacement Algorithms for a 1196 Virtual-storage Computer. IBM Syst. J. 5, 2 (June 1966), 78-101. 1197 DOI=http://dx.doi.org/10.1147/sj.52.0078 1198 """ 1199 1200 def __init__(self, cache_size): 1201 super(OPTCache, self).__init__(cache_size, enable_cache_row_key=0) 1202 self.table = PQTable() 1203 1204 def _lookup(self, trace_record, key, hash): 1205 if key not in self.table: 1206 return False 1207 # A cache hit. Update its next access time. 1208 assert ( 1209 self.table.pqinsert( 1210 OPTCacheEntry( 1211 key, trace_record.next_access_seq_no, self.table[key].value_size 1212 ) 1213 ) 1214 is not None 1215 ) 1216 return True 1217 1218 def _evict(self, trace_record, key, hash, value_size): 1219 while self.used_size + value_size > self.cache_size: 1220 evict_entry = self.table.pqpop() 1221 assert evict_entry is not None 1222 self.used_size -= evict_entry.value_size 1223 1224 def _insert(self, trace_record, key, hash, value_size): 1225 assert ( 1226 self.table.pqinsert( 1227 OPTCacheEntry(key, trace_record.next_access_seq_no, value_size) 1228 ) 1229 is None 1230 ) 1231 1232 def _should_admit(self, trace_record, key, hash, value_size): 1233 return True 1234 1235 def cache_name(self): 1236 return "Belady MIN (opt)" 1237 1238 1239class GDSizeEntry: 1240 """ 1241 A cache entry for the greedy dual size replacement policy. 1242 """ 1243 1244 def __init__(self, key, value_size, priority): 1245 self.key = key 1246 self.value_size = value_size 1247 self.priority = priority 1248 self.is_removed = False 1249 1250 def __cmp__(self, other): 1251 if other.priority != self.priority: 1252 return self.priority - other.priority 1253 return self.value_size - other.value_size 1254 1255 def __repr__(self): 1256 return "({} {} {} {})".format( 1257 self.key, self.next_access_seq_no, self.value_size, self.is_removed 1258 ) 1259 1260 1261class GDSizeCache(Cache): 1262 """ 1263 An implementation of the greedy dual size algorithm. 1264 We define cost as an entry's size. 1265 1266 See https://www.usenix.org/legacy/publications/library/proceedings/usits97/full_papers/cao/cao_html/node8.html 1267 and N. Young. The k-server dual and loose competitiveness for paging. 1268 Algorithmica,June 1994, vol. 11,(no.6):525-41. 1269 Rewritten version of ''On-line caching as cache size varies'', 1270 in The 2nd Annual ACM-SIAM Symposium on Discrete Algorithms, 241-250, 1991. 1271 """ 1272 1273 def __init__(self, cache_size, enable_cache_row_key): 1274 super(GDSizeCache, self).__init__(cache_size, enable_cache_row_key) 1275 self.table = PQTable() 1276 self.L = 0.0 1277 1278 def cache_name(self): 1279 if self.enable_cache_row_key: 1280 return "Hybrid GreedyDualSize (gdsize_hybrid)" 1281 return "GreedyDualSize (gdsize)" 1282 1283 def _lookup(self, trace_record, key, hash): 1284 if key not in self.table: 1285 return False 1286 # A cache hit. Update its priority. 1287 entry = self.table[key] 1288 assert ( 1289 self.table.pqinsert( 1290 GDSizeEntry(key, entry.value_size, self.L + entry.value_size) 1291 ) 1292 is not None 1293 ) 1294 return True 1295 1296 def _evict(self, trace_record, key, hash, value_size): 1297 while self.used_size + value_size > self.cache_size: 1298 evict_entry = self.table.pqpop() 1299 assert evict_entry is not None 1300 self.L = evict_entry.priority 1301 self.used_size -= evict_entry.value_size 1302 1303 def _insert(self, trace_record, key, hash, value_size): 1304 assert ( 1305 self.table.pqinsert(GDSizeEntry(key, value_size, self.L + value_size)) 1306 is None 1307 ) 1308 1309 def _should_admit(self, trace_record, key, hash, value_size): 1310 return True 1311 1312 1313class Deque(object): 1314 """A Deque class facilitates the implementation of LRU and ARC.""" 1315 1316 def __init__(self): 1317 self.od = OrderedDict() 1318 1319 def appendleft(self, k): 1320 if k in self.od: 1321 del self.od[k] 1322 self.od[k] = None 1323 1324 def pop(self): 1325 item = self.od.popitem(last=False) if self.od else None 1326 if item is not None: 1327 return item[0] 1328 return None 1329 1330 def remove(self, k): 1331 del self.od[k] 1332 1333 def __len__(self): 1334 return len(self.od) 1335 1336 def __contains__(self, k): 1337 return k in self.od 1338 1339 def __iter__(self): 1340 return reversed(self.od) 1341 1342 def __repr__(self): 1343 return "Deque(%r)" % (list(self),) 1344 1345 1346class ARCCache(Cache): 1347 """ 1348 An implementation of ARC. ARC assumes that all blocks are having the 1349 same size. The size of index and filter blocks are variable. To accommodate 1350 this, we modified ARC as follows: 1351 1) We use 16 KB as the average block size and calculate the number of blocks 1352 (c) in the cache. 1353 2) When we insert an entry, the cache evicts entries in both t1 and t2 1354 queues until it has enough space for the new entry. This also requires 1355 modification of the algorithm to maintain a maximum of 2*c blocks. 1356 1357 Nimrod Megiddo and Dharmendra S. Modha. 2003. ARC: A Self-Tuning, Low 1358 Overhead Replacement Cache. In Proceedings of the 2nd USENIX Conference on 1359 File and Storage Technologies (FAST '03). USENIX Association, Berkeley, CA, 1360 USA, 115-130. 1361 """ 1362 1363 def __init__(self, cache_size, enable_cache_row_key): 1364 super(ARCCache, self).__init__(cache_size, enable_cache_row_key) 1365 self.table = {} 1366 self.c = cache_size / 16 * 1024 # Number of elements in the cache. 1367 self.p = 0 # Target size for the list T1 1368 # L1: only once recently 1369 self.t1 = Deque() # T1: recent cache entries 1370 self.b1 = Deque() # B1: ghost entries recently evicted from the T1 cache 1371 # L2: at least twice recently 1372 self.t2 = Deque() # T2: frequent entries 1373 self.b2 = Deque() # B2: ghost entries recently evicted from the T2 cache 1374 1375 def _replace(self, key, value_size): 1376 while self.used_size + value_size > self.cache_size: 1377 if self.t1 and ((key in self.b2) or (len(self.t1) > self.p)): 1378 old = self.t1.pop() 1379 self.b1.appendleft(old) 1380 else: 1381 if self.t2: 1382 old = self.t2.pop() 1383 self.b2.appendleft(old) 1384 else: 1385 old = self.t1.pop() 1386 self.b1.appendleft(old) 1387 self.used_size -= self.table[old].value_size 1388 del self.table[old] 1389 1390 def _lookup(self, trace_record, key, hash): 1391 # Case I: key is in T1 or T2. 1392 # Move key to MRU position in T2. 1393 if key in self.t1: 1394 self.t1.remove(key) 1395 self.t2.appendleft(key) 1396 return True 1397 1398 if key in self.t2: 1399 self.t2.remove(key) 1400 self.t2.appendleft(key) 1401 return True 1402 return False 1403 1404 def _evict(self, trace_record, key, hash, value_size): 1405 # Case II: key is in B1 1406 # Move x from B1 to the MRU position in T2 (also fetch x to the cache). 1407 if key in self.b1: 1408 self.p = min(self.c, self.p + max(len(self.b2) / len(self.b1), 1)) 1409 self._replace(key, value_size) 1410 self.b1.remove(key) 1411 self.t2.appendleft(key) 1412 return 1413 1414 # Case III: key is in B2 1415 # Move x from B2 to the MRU position in T2 (also fetch x to the cache). 1416 if key in self.b2: 1417 self.p = max(0, self.p - max(len(self.b1) / len(self.b2), 1)) 1418 self._replace(key, value_size) 1419 self.b2.remove(key) 1420 self.t2.appendleft(key) 1421 return 1422 1423 # Case IV: key is not in (T1 u B1 u T2 u B2) 1424 self._replace(key, value_size) 1425 while len(self.t1) + len(self.b1) >= self.c and self.b1: 1426 self.b1.pop() 1427 1428 total = len(self.t1) + len(self.b1) + len(self.t2) + len(self.b2) 1429 while total >= (2 * self.c) and self.b2: 1430 self.b2.pop() 1431 total -= 1 1432 # Finally, move it to MRU position in T1. 1433 self.t1.appendleft(key) 1434 return 1435 1436 def _insert(self, trace_record, key, hash, value_size): 1437 self.table[key] = CacheEntry( 1438 value_size, 1439 trace_record.cf_id, 1440 trace_record.level, 1441 trace_record.block_type, 1442 trace_record.table_id, 1443 0, 1444 trace_record.access_time, 1445 ) 1446 1447 def _should_admit(self, trace_record, key, hash, value_size): 1448 return True 1449 1450 def cache_name(self): 1451 if self.enable_cache_row_key: 1452 return "Hybrid Adaptive Replacement Cache (arc_hybrid)" 1453 return "Adaptive Replacement Cache (arc)" 1454 1455 1456class LRUCache(Cache): 1457 """ 1458 A strict LRU queue. 1459 """ 1460 1461 def __init__(self, cache_size, enable_cache_row_key): 1462 super(LRUCache, self).__init__(cache_size, enable_cache_row_key) 1463 self.table = {} 1464 self.lru = Deque() 1465 1466 def cache_name(self): 1467 if self.enable_cache_row_key: 1468 return "Hybrid LRU (lru_hybrid)" 1469 return "LRU (lru)" 1470 1471 def _lookup(self, trace_record, key, hash): 1472 if key not in self.table: 1473 return False 1474 # A cache hit. Update LRU queue. 1475 self.lru.remove(key) 1476 self.lru.appendleft(key) 1477 return True 1478 1479 def _evict(self, trace_record, key, hash, value_size): 1480 while self.used_size + value_size > self.cache_size: 1481 evict_key = self.lru.pop() 1482 self.used_size -= self.table[evict_key].value_size 1483 del self.table[evict_key] 1484 1485 def _insert(self, trace_record, key, hash, value_size): 1486 self.table[key] = CacheEntry( 1487 value_size, 1488 trace_record.cf_id, 1489 trace_record.level, 1490 trace_record.block_type, 1491 trace_record.table_id, 1492 0, 1493 trace_record.access_time, 1494 ) 1495 self.lru.appendleft(key) 1496 1497 def _should_admit(self, trace_record, key, hash, value_size): 1498 return True 1499 1500 1501class TraceCache(Cache): 1502 """ 1503 A trace cache. Lookup returns true if the trace observes a cache hit. 1504 It is used to maintain cache hits observed in the trace. 1505 """ 1506 1507 def __init__(self, cache_size): 1508 super(TraceCache, self).__init__(cache_size, enable_cache_row_key=0) 1509 1510 def _lookup(self, trace_record, key, hash): 1511 return trace_record.is_hit 1512 1513 def _evict(self, trace_record, key, hash, value_size): 1514 pass 1515 1516 def _insert(self, trace_record, key, hash, value_size): 1517 pass 1518 1519 def _should_admit(self, trace_record, key, hash, value_size): 1520 return False 1521 1522 def cache_name(self): 1523 return "Trace" 1524 1525 1526def parse_cache_size(cs): 1527 cs = cs.replace("\n", "") 1528 if cs[-1] == "M": 1529 return int(cs[: len(cs) - 1]) * 1024 * 1024 1530 if cs[-1] == "G": 1531 return int(cs[: len(cs) - 1]) * 1024 * 1024 * 1024 1532 if cs[-1] == "T": 1533 return int(cs[: len(cs) - 1]) * 1024 * 1024 * 1024 * 1024 1534 return int(cs) 1535 1536 1537def create_cache(cache_type, cache_size, downsample_size): 1538 cache_size = cache_size / downsample_size 1539 enable_cache_row_key = 0 1540 if "hybridn" in cache_type: 1541 enable_cache_row_key = 2 1542 cache_type = cache_type[:-8] 1543 if "hybrid" in cache_type: 1544 enable_cache_row_key = 1 1545 cache_type = cache_type[:-7] 1546 if cache_type == "ts": 1547 return ThompsonSamplingCache( 1548 cache_size, 1549 enable_cache_row_key, 1550 [LRUPolicy(), LFUPolicy(), HyperbolicPolicy()], 1551 cost_class_label=None, 1552 ) 1553 elif cache_type == "linucb": 1554 return LinUCBCache( 1555 cache_size, 1556 enable_cache_row_key, 1557 [LRUPolicy(), LFUPolicy(), HyperbolicPolicy()], 1558 cost_class_label=None, 1559 ) 1560 elif cache_type == "pylru": 1561 return ThompsonSamplingCache( 1562 cache_size, enable_cache_row_key, [LRUPolicy()], cost_class_label=None 1563 ) 1564 elif cache_type == "pymru": 1565 return ThompsonSamplingCache( 1566 cache_size, enable_cache_row_key, [MRUPolicy()], cost_class_label=None 1567 ) 1568 elif cache_type == "pylfu": 1569 return ThompsonSamplingCache( 1570 cache_size, enable_cache_row_key, [LFUPolicy()], cost_class_label=None 1571 ) 1572 elif cache_type == "pyhb": 1573 return ThompsonSamplingCache( 1574 cache_size, 1575 enable_cache_row_key, 1576 [HyperbolicPolicy()], 1577 cost_class_label=None, 1578 ) 1579 elif cache_type == "pycctbbt": 1580 return ThompsonSamplingCache( 1581 cache_size, 1582 enable_cache_row_key, 1583 [CostClassPolicy()], 1584 cost_class_label="table_bt", 1585 ) 1586 elif cache_type == "pycccf": 1587 return ThompsonSamplingCache( 1588 cache_size, enable_cache_row_key, [CostClassPolicy()], cost_class_label="cf" 1589 ) 1590 elif cache_type == "pycctblevelbt": 1591 return ThompsonSamplingCache( 1592 cache_size, 1593 enable_cache_row_key, 1594 [CostClassPolicy()], 1595 cost_class_label="table_level_bt", 1596 ) 1597 elif cache_type == "pycccfbt": 1598 return ThompsonSamplingCache( 1599 cache_size, 1600 enable_cache_row_key, 1601 [CostClassPolicy()], 1602 cost_class_label="cf_bt", 1603 ) 1604 elif cache_type == "pycctb": 1605 return ThompsonSamplingCache( 1606 cache_size, 1607 enable_cache_row_key, 1608 [CostClassPolicy()], 1609 cost_class_label="table", 1610 ) 1611 elif cache_type == "pyccbt": 1612 return ThompsonSamplingCache( 1613 cache_size, enable_cache_row_key, [CostClassPolicy()], cost_class_label="bt" 1614 ) 1615 elif cache_type == "opt": 1616 if enable_cache_row_key: 1617 print("opt does not support hybrid mode.") 1618 assert False 1619 return OPTCache(cache_size) 1620 elif cache_type == "trace": 1621 if enable_cache_row_key: 1622 print("trace does not support hybrid mode.") 1623 assert False 1624 return TraceCache(cache_size) 1625 elif cache_type == "lru": 1626 return LRUCache(cache_size, enable_cache_row_key) 1627 elif cache_type == "arc": 1628 return ARCCache(cache_size, enable_cache_row_key) 1629 elif cache_type == "gdsize": 1630 return GDSizeCache(cache_size, enable_cache_row_key) 1631 else: 1632 print("Unknown cache type {}".format(cache_type)) 1633 assert False 1634 return None 1635 1636 1637class BlockAccessTimeline: 1638 """ 1639 BlockAccessTimeline stores all accesses of a block. 1640 """ 1641 1642 def __init__(self): 1643 self.accesses = [] 1644 self.current_access_index = 1 1645 1646 def get_next_access(self): 1647 if self.current_access_index == len(self.accesses): 1648 return sys.maxsize 1649 next_access_seq_no = self.accesses[self.current_access_index] 1650 self.current_access_index += 1 1651 return next_access_seq_no 1652 1653 1654def percent(e1, e2): 1655 if e2 == 0: 1656 return -1 1657 return float(e1) * 100.0 / float(e2) 1658 1659 1660def is_target_cf(access_cf, target_cf_name): 1661 if target_cf_name == "all": 1662 return True 1663 return access_cf == target_cf_name 1664 1665 1666def run( 1667 trace_file_path, 1668 cache_type, 1669 cache, 1670 warmup_seconds, 1671 max_accesses_to_process, 1672 target_cf_name, 1673): 1674 warmup_complete = False 1675 trace_miss_ratio_stats = MissRatioStats(kSecondsInMinute) 1676 access_seq_no = 0 1677 time_interval = 1 1678 start_time = time.time() 1679 trace_start_time = 0 1680 trace_duration = 0 1681 is_opt_cache = False 1682 if cache.cache_name() == "Belady MIN (opt)": 1683 is_opt_cache = True 1684 1685 block_access_timelines = {} 1686 num_no_inserts = 0 1687 num_blocks_with_no_size = 0 1688 num_inserts_block_with_no_size = 0 1689 1690 if is_opt_cache: 1691 # Read all blocks in memory and stores their access times so that OPT 1692 # can use this information to evict the cached key which next access is 1693 # the furthest in the future. 1694 print("Preprocessing block traces.") 1695 with open(trace_file_path, "r") as trace_file: 1696 for line in trace_file: 1697 if ( 1698 max_accesses_to_process != -1 1699 and access_seq_no > max_accesses_to_process 1700 ): 1701 break 1702 ts = line.split(",") 1703 timestamp = int(ts[0]) 1704 cf_name = ts[5] 1705 if not is_target_cf(cf_name, target_cf_name): 1706 continue 1707 if trace_start_time == 0: 1708 trace_start_time = timestamp 1709 trace_duration = timestamp - trace_start_time 1710 block_id = int(ts[1]) 1711 block_size = int(ts[3]) 1712 no_insert = int(ts[9]) 1713 if block_id not in block_access_timelines: 1714 block_access_timelines[block_id] = BlockAccessTimeline() 1715 if block_size == 0: 1716 num_blocks_with_no_size += 1 1717 block_access_timelines[block_id].accesses.append(access_seq_no) 1718 access_seq_no += 1 1719 if no_insert == 1: 1720 num_no_inserts += 1 1721 if no_insert == 0 and block_size == 0: 1722 num_inserts_block_with_no_size += 1 1723 if access_seq_no % 100 != 0: 1724 continue 1725 now = time.time() 1726 if now - start_time > time_interval * 10: 1727 print( 1728 "Take {} seconds to process {} trace records with trace " 1729 "duration of {} seconds. Throughput: {} records/second.".format( 1730 now - start_time, 1731 access_seq_no, 1732 trace_duration / 1000000, 1733 access_seq_no / (now - start_time), 1734 ) 1735 ) 1736 time_interval += 1 1737 print( 1738 "Trace contains {0} blocks, {1}({2:.2f}%) blocks with no size." 1739 "{3} accesses, {4}({5:.2f}%) accesses with no_insert," 1740 "{6}({7:.2f}%) accesses that want to insert but block size is 0.".format( 1741 len(block_access_timelines), 1742 num_blocks_with_no_size, 1743 percent(num_blocks_with_no_size, len(block_access_timelines)), 1744 access_seq_no, 1745 num_no_inserts, 1746 percent(num_no_inserts, access_seq_no), 1747 num_inserts_block_with_no_size, 1748 percent(num_inserts_block_with_no_size, access_seq_no), 1749 ) 1750 ) 1751 1752 access_seq_no = 0 1753 time_interval = 1 1754 start_time = time.time() 1755 trace_start_time = 0 1756 trace_duration = 0 1757 print("Running simulated {} cache on block traces.".format(cache.cache_name())) 1758 with open(trace_file_path, "r") as trace_file: 1759 for line in trace_file: 1760 if ( 1761 max_accesses_to_process != -1 1762 and access_seq_no > max_accesses_to_process 1763 ): 1764 break 1765 if access_seq_no % 1000000 == 0: 1766 # Force a python gc periodically to reduce memory usage. 1767 gc.collect() 1768 ts = line.split(",") 1769 timestamp = int(ts[0]) 1770 cf_name = ts[5] 1771 if not is_target_cf(cf_name, target_cf_name): 1772 continue 1773 if trace_start_time == 0: 1774 trace_start_time = timestamp 1775 trace_duration = timestamp - trace_start_time 1776 if ( 1777 not warmup_complete 1778 and warmup_seconds > 0 1779 and trace_duration > warmup_seconds * 1000000 1780 ): 1781 cache.miss_ratio_stats.reset_counter() 1782 warmup_complete = True 1783 next_access_seq_no = 0 1784 block_id = int(ts[1]) 1785 if is_opt_cache: 1786 next_access_seq_no = block_access_timelines[block_id].get_next_access() 1787 record = TraceRecord( 1788 access_time=int(ts[0]), 1789 block_id=int(ts[1]), 1790 block_type=int(ts[2]), 1791 block_size=int(ts[3]), 1792 cf_id=int(ts[4]), 1793 cf_name=ts[5], 1794 level=int(ts[6]), 1795 fd=int(ts[7]), 1796 caller=int(ts[8]), 1797 no_insert=int(ts[9]), 1798 get_id=int(ts[10]), 1799 key_id=int(ts[11]), 1800 kv_size=int(ts[12]), 1801 is_hit=int(ts[13]), 1802 referenced_key_exist_in_block=int(ts[14]), 1803 num_keys_in_block=int(ts[15]), 1804 table_id=int(ts[16]), 1805 seq_number=int(ts[17]), 1806 block_key_size=int(ts[18]), 1807 key_size=int(ts[19]), 1808 block_offset_in_file=int(ts[20]), 1809 next_access_seq_no=next_access_seq_no, 1810 ) 1811 trace_miss_ratio_stats.update_metrics( 1812 record.access_time, is_hit=record.is_hit, miss_bytes=record.block_size 1813 ) 1814 cache.access(record) 1815 access_seq_no += 1 1816 del record 1817 del ts 1818 if access_seq_no % 100 != 0: 1819 continue 1820 # Report progress every 10 seconds. 1821 now = time.time() 1822 if now - start_time > time_interval * 10: 1823 print( 1824 "Take {} seconds to process {} trace records with trace " 1825 "duration of {} seconds. Throughput: {} records/second. " 1826 "Trace miss ratio {}".format( 1827 now - start_time, 1828 access_seq_no, 1829 trace_duration / 1000000, 1830 access_seq_no / (now - start_time), 1831 trace_miss_ratio_stats.miss_ratio(), 1832 ) 1833 ) 1834 time_interval += 1 1835 print( 1836 "{},0,0,{},{},{}".format( 1837 cache_type, 1838 cache.cache_size, 1839 cache.miss_ratio_stats.miss_ratio(), 1840 cache.miss_ratio_stats.num_accesses, 1841 ) 1842 ) 1843 now = time.time() 1844 print( 1845 "Take {} seconds to process {} trace records with trace duration of {} " 1846 "seconds. Throughput: {} records/second. Trace miss ratio {}".format( 1847 now - start_time, 1848 access_seq_no, 1849 trace_duration / 1000000, 1850 access_seq_no / (now - start_time), 1851 trace_miss_ratio_stats.miss_ratio(), 1852 ) 1853 ) 1854 print( 1855 "{},0,0,{},{},{}".format( 1856 cache_type, 1857 cache.cache_size, 1858 cache.miss_ratio_stats.miss_ratio(), 1859 cache.miss_ratio_stats.num_accesses, 1860 ) 1861 ) 1862 return trace_start_time, trace_duration 1863 1864 1865def report_stats( 1866 cache, 1867 cache_type, 1868 cache_size, 1869 target_cf_name, 1870 result_dir, 1871 trace_start_time, 1872 trace_end_time, 1873): 1874 cache_label = "{}-{}-{}".format(cache_type, cache_size, target_cf_name) 1875 with open("{}/data-ml-mrc-{}".format(result_dir, cache_label), "w+") as mrc_file: 1876 mrc_file.write( 1877 "{},0,0,{},{},{}\n".format( 1878 cache_type, 1879 cache_size, 1880 cache.miss_ratio_stats.miss_ratio(), 1881 cache.miss_ratio_stats.num_accesses, 1882 ) 1883 ) 1884 1885 cache_stats = [ 1886 cache.per_second_miss_ratio_stats, 1887 cache.miss_ratio_stats, 1888 cache.per_hour_miss_ratio_stats, 1889 ] 1890 for i in range(len(cache_stats)): 1891 avg_miss_bytes, p95_miss_bytes = cache_stats[i].compute_miss_bytes() 1892 1893 with open( 1894 "{}/data-ml-avgmb-{}-{}".format( 1895 result_dir, cache_stats[i].time_unit, cache_label 1896 ), 1897 "w+", 1898 ) as mb_file: 1899 mb_file.write( 1900 "{},0,0,{},{}\n".format(cache_type, cache_size, avg_miss_bytes) 1901 ) 1902 1903 with open( 1904 "{}/data-ml-p95mb-{}-{}".format( 1905 result_dir, cache_stats[i].time_unit, cache_label 1906 ), 1907 "w+", 1908 ) as mb_file: 1909 mb_file.write( 1910 "{},0,0,{},{}\n".format(cache_type, cache_size, p95_miss_bytes) 1911 ) 1912 1913 cache_stats[i].write_miss_timeline( 1914 cache_type, 1915 cache_size, 1916 target_cf_name, 1917 result_dir, 1918 trace_start_time, 1919 trace_end_time, 1920 ) 1921 cache_stats[i].write_miss_ratio_timeline( 1922 cache_type, 1923 cache_size, 1924 target_cf_name, 1925 result_dir, 1926 trace_start_time, 1927 trace_end_time, 1928 ) 1929 1930 if not cache.is_ml_cache(): 1931 return 1932 1933 policy_stats = [cache.policy_stats, cache.per_hour_policy_stats] 1934 for i in range(len(policy_stats)): 1935 policy_stats[i].write_policy_timeline( 1936 cache_type, 1937 cache_size, 1938 target_cf_name, 1939 result_dir, 1940 trace_start_time, 1941 trace_end_time, 1942 ) 1943 policy_stats[i].write_policy_ratio_timeline( 1944 cache_type, 1945 cache_size, 1946 target_cf_name, 1947 result_dir, 1948 trace_start_time, 1949 trace_end_time, 1950 ) 1951 1952 1953if __name__ == "__main__": 1954 if len(sys.argv) <= 8: 1955 print( 1956 "Must provide 8 arguments.\n" 1957 "1) Cache type (ts, linucb, arc, lru, opt, pylru, pymru, pylfu, " 1958 "pyhb, gdsize, trace). One may evaluate the hybrid row_block cache " 1959 "by appending '_hybrid' to a cache_type, e.g., ts_hybrid. " 1960 "Note that hybrid is not supported with opt and trace. \n" 1961 "2) Cache size (xM, xG, xT).\n" 1962 "3) The sampling frequency used to collect the trace. (The " 1963 "simulation scales down the cache size by the sampling frequency).\n" 1964 "4) Warmup seconds (The number of seconds used for warmup).\n" 1965 "5) Trace file path.\n" 1966 "6) Result directory (A directory that saves generated results)\n" 1967 "7) Max number of accesses to process\n" 1968 "8) The target column family. (The simulation will only run " 1969 "accesses on the target column family. If it is set to all, " 1970 "it will run against all accesses.)" 1971 ) 1972 exit(1) 1973 print("Arguments: {}".format(sys.argv)) 1974 cache_type = sys.argv[1] 1975 cache_size = parse_cache_size(sys.argv[2]) 1976 downsample_size = int(sys.argv[3]) 1977 warmup_seconds = int(sys.argv[4]) 1978 trace_file_path = sys.argv[5] 1979 result_dir = sys.argv[6] 1980 max_accesses_to_process = int(sys.argv[7]) 1981 target_cf_name = sys.argv[8] 1982 cache = create_cache(cache_type, cache_size, downsample_size) 1983 trace_start_time, trace_duration = run( 1984 trace_file_path, 1985 cache_type, 1986 cache, 1987 warmup_seconds, 1988 max_accesses_to_process, 1989 target_cf_name, 1990 ) 1991 trace_end_time = trace_start_time + trace_duration 1992 report_stats( 1993 cache, 1994 cache_type, 1995 cache_size, 1996 target_cf_name, 1997 result_dir, 1998 trace_start_time, 1999 trace_end_time, 2000 ) 2001