Skip to content

local_process_memory_consumption

mlte/measurement/memory/local_process_memory_consumption.py

Memory consumption measurement for local training processes.

LocalProcessMemoryConsumption

Bases: ProcessMeasurement

Measure memory consumption for a local training process.

Source code in mlte/measurement/memory/local_process_memory_consumption.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
class LocalProcessMemoryConsumption(ProcessMeasurement):
    """Measure memory consumption for a local training process."""

    def __init__(self, identifier: str):
        """
        Initialize a LocalProcessMemoryConsumption instance.

        :param identifier: A unique identifier for the measurement
        """
        super().__init__(self, identifier)

    def __call__(self, pid: int, poll_interval: int = 1) -> MemoryStatistics:
        """
        Monitor memory consumption of process at `pid` until exit.

        :param pid: The process identifier
        :param poll_interval: The poll interval, in seconds
        :return: The captured statistics
        """
        stats = []
        while True:
            kb = _get_memory_usage_psutil(pid)
            if kb == 0:
                break
            stats.append(kb)
            time.sleep(poll_interval)

        return MemoryStatistics(
            self.metadata,
            avg=int(sum(stats) / len(stats)),
            min=min(stats),
            max=max(stats),
        )

    @classmethod
    def value(self) -> Type[MemoryStatistics]:
        """Returns the class type object for the Value produced by the Measurement."""
        return MemoryStatistics

__call__(pid, poll_interval=1)

Monitor memory consumption of process at pid until exit.

Parameters:

Name Type Description Default
pid int

The process identifier

 required
poll_interval int

The poll interval, in seconds

 1

Returns:

Type Description
MemoryStatistics

The captured statistics
Source code in mlte/measurement/memory/local_process_memory_consumption.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def __call__(self, pid: int, poll_interval: int = 1) -> MemoryStatistics:
    """
    Monitor memory consumption of process at `pid` until exit.

    :param pid: The process identifier
    :param poll_interval: The poll interval, in seconds
    :return: The captured statistics
    """
    stats = []
    while True:
        kb = _get_memory_usage_psutil(pid)
        if kb == 0:
            break
        stats.append(kb)
        time.sleep(poll_interval)

    return MemoryStatistics(
        self.metadata,
        avg=int(sum(stats) / len(stats)),
        min=min(stats),
        max=max(stats),
    )

__init__(identifier)

Initialize a LocalProcessMemoryConsumption instance.

Parameters:

Name Type Description Default
identifier str

A unique identifier for the measurement

 required
Source code in mlte/measurement/memory/local_process_memory_consumption.py 
134
135
136
137
138
139
140
def __init__(self, identifier: str):
    """
    Initialize a LocalProcessMemoryConsumption instance.

    :param identifier: A unique identifier for the measurement
    """
    super().__init__(self, identifier)

value() classmethod

Returns the class type object for the Value produced by the Measurement.

Source code in mlte/measurement/memory/local_process_memory_consumption.py 
165
166
167
168
@classmethod
def value(self) -> Type[MemoryStatistics]:
    """Returns the class type object for the Value produced by the Measurement."""
    return MemoryStatistics

MemoryStatistics

Bases: ValueBase

The MemoryStatistics class encapsulates data and functionality for tracking and updating memory consumption statistics for a running process.

Source code in mlte/measurement/memory/local_process_memory_consumption.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
class MemoryStatistics(ValueBase):
    """
    The MemoryStatistics class encapsulates data
    and functionality for tracking and updating memory
    consumption statistics for a running process.
    """

    def __init__(
        self,
        evidence_metadata: EvidenceMetadata,
        avg: int,
        min: int,
        max: int,
    ):
        """
        Initialize a MemoryStatistics instance.

        :param evidence_metadata: The generating measurement's metadata
        :param avg: The average memory consumption, in KB
        :param min: The minimum memory consumption, in KB
        :param max: The maximum memory consumption, in KB
        """
        super().__init__(self, evidence_metadata)

        self.avg = avg
        """The average memory consumption (KB)."""

        self.min = min
        """The minimum memory consumption (KB)."""

        self.max = max
        """The maximum memory consumption (KB)."""

    def serialize(self) -> Dict[str, Any]:
        """
        Serialize an MemoryStatistics to a JSON object.

        :return: The JSON object
        """
        return {"avg": self.avg, "min": self.min, "max": self.max}

    @staticmethod
    def deserialize(
        evidence_metadata: EvidenceMetadata, data: Dict[str, Any]
    ) -> MemoryStatistics:
        """
        Deserialize an MemoryStatistics from a JSON object.

        :param evidence_metadata: The generating measurement's metadata
        :param data: The JSON object

        :return: The deserialized instance
        """
        return MemoryStatistics(
            evidence_metadata,
            avg=data["avg"],
            min=data["min"],
            max=data["max"],
        )

    def __str__(self) -> str:
        """Return a string representation of MemoryStatistics."""
        s = ""
        s += f"Average: {self.avg}\n"
        s += f"Minimum: {self.min}\n"
        s += f"Maximum: {self.max}"
        return s

    @classmethod
    def max_consumption_less_than(cls, threshold: int) -> Condition:
        """
        Construct and invoke a condition for maximum memory consumption.

        :param threshold: The threshold value for maximum consumption, in KB

        :return: The Condition that can be used to validate a Value.
        """
        condition: Condition = Condition.build_condition(
            bool_exp=lambda stats: stats.max < threshold,
            success=f"Maximum consumption below threshold {threshold}",
            failure=f"Maximum consumption exceeds threshold {threshold}",
        )
        return condition

    @classmethod
    def average_consumption_less_than(cls, threshold: float) -> Condition:
        """
        Construct and invoke a condition for average memory consumption.

        :param threshold: The threshold value for average consumption, in KB

        :return: The Condition that can be used to validate a Value.
        """
        condition: Condition = Condition.build_condition(
            bool_exp=lambda stats: stats.avg < threshold,
            success=f"Average consumption below threshold {threshold}",
            failure=f"Average consumption exceeds threshold {threshold}",
        )
        return condition

avg = avg instance-attribute

The average memory consumption (KB).

max = max instance-attribute

The maximum memory consumption (KB).

min = min instance-attribute

The minimum memory consumption (KB).

__init__(evidence_metadata, avg, min, max)

Initialize a MemoryStatistics instance.

Parameters:

Name Type Description Default
evidence_metadata EvidenceMetadata

The generating measurement's metadata

 required
avg int

The average memory consumption, in KB

 required
min int

The minimum memory consumption, in KB

 required
max int

The maximum memory consumption, in KB

 required
Source code in mlte/measurement/memory/local_process_memory_consumption.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def __init__(
    self,
    evidence_metadata: EvidenceMetadata,
    avg: int,
    min: int,
    max: int,
):
    """
    Initialize a MemoryStatistics instance.

    :param evidence_metadata: The generating measurement's metadata
    :param avg: The average memory consumption, in KB
    :param min: The minimum memory consumption, in KB
    :param max: The maximum memory consumption, in KB
    """
    super().__init__(self, evidence_metadata)

    self.avg = avg
    """The average memory consumption (KB)."""

    self.min = min
    """The minimum memory consumption (KB)."""

    self.max = max
    """The maximum memory consumption (KB)."""

__str__()

Return a string representation of MemoryStatistics.

Source code in mlte/measurement/memory/local_process_memory_consumption.py 
85
86
87
88
89
90
91
def __str__(self) -> str:
    """Return a string representation of MemoryStatistics."""
    s = ""
    s += f"Average: {self.avg}\n"
    s += f"Minimum: {self.min}\n"
    s += f"Maximum: {self.max}"
    return s

average_consumption_less_than(threshold) classmethod

Construct and invoke a condition for average memory consumption.

Parameters:

Name Type Description Default
threshold float

The threshold value for average consumption, in KB

 required

Returns:

Type Description
Condition

The Condition that can be used to validate a Value.
Source code in mlte/measurement/memory/local_process_memory_consumption.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
@classmethod
def average_consumption_less_than(cls, threshold: float) -> Condition:
    """
    Construct and invoke a condition for average memory consumption.

    :param threshold: The threshold value for average consumption, in KB

    :return: The Condition that can be used to validate a Value.
    """
    condition: Condition = Condition.build_condition(
        bool_exp=lambda stats: stats.avg < threshold,
        success=f"Average consumption below threshold {threshold}",
        failure=f"Average consumption exceeds threshold {threshold}",
    )
    return condition

deserialize(evidence_metadata, data) staticmethod

Deserialize an MemoryStatistics from a JSON object.

Parameters:

Name Type Description Default
evidence_metadata EvidenceMetadata

The generating measurement's metadata

 required
data Dict[str, Any]

The JSON object

 required

Returns:

Type Description
MemoryStatistics

The deserialized instance
Source code in mlte/measurement/memory/local_process_memory_consumption.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
@staticmethod
def deserialize(
    evidence_metadata: EvidenceMetadata, data: Dict[str, Any]
) -> MemoryStatistics:
    """
    Deserialize an MemoryStatistics from a JSON object.

    :param evidence_metadata: The generating measurement's metadata
    :param data: The JSON object

    :return: The deserialized instance
    """
    return MemoryStatistics(
        evidence_metadata,
        avg=data["avg"],
        min=data["min"],
        max=data["max"],
    )

max_consumption_less_than(threshold) classmethod

Construct and invoke a condition for maximum memory consumption.

Parameters:

Name Type Description Default
threshold int

The threshold value for maximum consumption, in KB

 required

Returns:

Type Description
Condition

The Condition that can be used to validate a Value.
Source code in mlte/measurement/memory/local_process_memory_consumption.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@classmethod
def max_consumption_less_than(cls, threshold: int) -> Condition:
    """
    Construct and invoke a condition for maximum memory consumption.

    :param threshold: The threshold value for maximum consumption, in KB

    :return: The Condition that can be used to validate a Value.
    """
    condition: Condition = Condition.build_condition(
        bool_exp=lambda stats: stats.max < threshold,
        success=f"Maximum consumption below threshold {threshold}",
        failure=f"Maximum consumption exceeds threshold {threshold}",
    )
    return condition

serialize()

Serialize an MemoryStatistics to a JSON object.

Returns:

Type Description
Dict[str, Any]

The JSON object
Source code in mlte/measurement/memory/local_process_memory_consumption.py 
58
59
60
61
62
63
64
def serialize(self) -> Dict[str, Any]:
    """
    Serialize an MemoryStatistics to a JSON object.

    :return: The JSON object
    """
    return {"avg": self.avg, "min": self.min, "max": self.max}