blob: 124f8bc55614f691e328b8c6dffd9bdebaf0290a [file] [log] [blame]
# Copyright 2022 The Fuchsia Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import collections
import json
import re
def re_from_string_or_list(input):
if input is None:
return None
re_str = input if isinstance(input, str) else '|'.join(input)
return re.compile(re_str)
class BucketMatch(object):
def __init__(self, name, process, vmo): = name
self.process = re_from_string_or_list(process)
self.vmo = re_from_string_or_list(vmo)
class Bucket(object):
def __init__(self, name): = name
self.processes = collections.defaultdict(list)
self.size = 0
class Digest(object):
def FromMatchSpecs(cls, snapshot, match_specs):
return Digest(
snapshot, [BucketMatch(m[0], m[1], m[2]) for m in match_specs])
def FromJSON(cls, snapshot, match_json):
return Digest(
snapshot, [
m["name"], m["process"] if "process" in m else None,
m["vmo"] if "vmo" in m else None) for m in match_json
def FromJSONFile(cls, snapshot, match_file):
return cls.FromJSON(snapshot, json.load(match_file))
def FromJSONString(cls, snapshot, match_string):
return cls.FromJSON(snapshot, json.loads(match_string))
def FromJSONFilename(cls, snapshot, match_filename):
with open(match_filename) as match_file:
return cls.FromJSONFile(snapshot, match_file)
def __init__(self, snapshot, bucket_matches):
self.buckets = { Bucket( for bm in bucket_matches}
undigested_vmos = snapshot.vmos.copy()
# Each VMO will be assigned to at most one bucket. Precedence follows
# the order of bucket_matches.
for process in snapshot.processes.values():
for bm in bucket_matches:
if bm.process and not bm.process.fullmatch(
for vmo in process.vmos:
if vmo.koid not in undigested_vmos:
if bm.vmo and not bm.vmo.fullmatch(
if vmo.committed_bytes == 0:
self.buckets[].size += vmo.committed_bytes
del undigested_vmos[vmo.koid]
undigested = Bucket("Undigested")
# For the Undigested bucket, we want to store VMOs by "sharing pools"
# That is, the keys to undigested.processes will be a set of process
# names that map to the list of VMOs they all share.
vmos_to_processes = collections.defaultdict(list)
for process in snapshot.processes.values():
for v in filter(lambda v: v.koid in undigested_vmos, process.vmos):
for vmo_koid, processes in vmos_to_processes.items():
processes.sort(key=lambda p: p.full_name)
vmo = snapshot.vmos[vmo_koid]
if vmo.committed_bytes == 0:
undigested.size += vmo.committed_bytes
self.buckets["Undigested"] = undigested
kernel = Bucket("Kernel")
kernel.size = (
snapshot.kernel.wired + snapshot.kernel.total_heap +
snapshot.kernel.mmu + snapshot.kernel.ipc + snapshot.kernel.other)
self.buckets["Kernel"] = kernel
orphaned = Bucket("Orphaned")
orphaned.size = snapshot.orphaned
self.buckets["Orphaned"] = orphaned