Re: [PATCH v7 2/2] tests/functional: add memlock tests

[Thomas Huth <thuth@redhat.com>]

On 21/05/2025 15.55, Alexandr Moshkov wrote:
> Add new tests to check the correctness of the `-overcommit memlock`
> option (possible values: off, on, on-fault) by using
> `/proc/{qemu_pid}/smaps` file to check in Size, Rss and Locked fields of
> anonymous segments:
> 
> * if `memlock=off`, then Locked = 0 on every anonymous smaps;
> * if `memlock=on`, then Size, Rss and Locked values must be equal for
> every anon smaps where Rss is not 0;
> * if `memlock=on-fault`, then Rss and Locked must be equal on every anon
> smaps and anonymous segment with Rss < Size must exists.
> 
> Signed-off-by: Alexandr Moshkov <dtalexundeer@yandex-team.ru>
> ---
>   tests/functional/meson.build     |   1 +
>   tests/functional/test_memlock.py | 107 +++++++++++++++++++++++++++++++
>   2 files changed, 108 insertions(+)
>   create mode 100755 tests/functional/test_memlock.py
> 
> diff --git a/tests/functional/meson.build b/tests/functional/meson.build
> index 0f8be30fe2..339af7835f 100644
> --- a/tests/functional/meson.build
> +++ b/tests/functional/meson.build
> @@ -61,6 +61,7 @@ tests_generic_system = [
>     'empty_cpu_model',
>     'info_usernet',
>     'version',
> +  'memlock',
>   ]
>   
>   tests_generic_linuxuser = [
> diff --git a/tests/functional/test_memlock.py b/tests/functional/test_memlock.py
> new file mode 100755
> index 0000000000..a090e7f9ad
> --- /dev/null
> +++ b/tests/functional/test_memlock.py
> @@ -0,0 +1,107 @@
> +#!/usr/bin/env python3
> +#
> +# Functional test that check overcommit memlock options
> +#
> +# Copyright (c) Yandex Technologies LLC, 2025
> +#
> +# Author:
> +#  Alexandr Moshkov <dtalexundeer@yandex-team.ru>
> +#
> +#
> +# This work is licensed under the terms of the GNU GPL, version 2 or
> +# later.  See the COPYING file in the top-level directory.
> +
> +import re
> +
> +from typing import List, Dict
> +
> +from qemu_test import QemuSystemTest
> +from qemu_test import skipLockedMemoryTest
> +
> +
> +SMAPS_HEADER_PATTERN = re.compile(r'^\w+-\w+', re.MULTILINE)
> +SMAPS_VALUE_PATTERN = re.compile(r'^(\w+):\s+(\d+) kB', re.MULTILINE)
> +
> +
> +@skipLockedMemoryTest(2_097_152)  # 2GB
> +class MemlockTest(QemuSystemTest):
> +    """
> +    Boots a Linux system with memlock options.
> +    Then verify, that this options is working correctly
> +    by checking the smaps of the QEMU proccess.
> +    """
> +
> +    def common_vm_setup_with_memlock(self, memlock):
> +        self.vm.add_args('-overcommit', f'mem-lock={memlock}')
> +        self.vm.launch()
> +
> +    def get_anon_smaps_by_pid(self, pid):
> +        smaps_raw = self._get_raw_smaps_by_pid(pid)
> +        return self._parse_anonymous_smaps(smaps_raw)
> +
> +    def test_memlock_off(self):
> +        self.common_vm_setup_with_memlock('off')
> +
> +        anon_smaps = self.get_anon_smaps_by_pid(self.vm.get_pid())
> +
> +        # locked = 0 on every smap
> +        for smap in anon_smaps:
> +            self.assertEqual(smap['Locked'], 0)
> +
> +    def test_memlock_on(self):
> +        self.common_vm_setup_with_memlock('on')
> +
> +        anon_smaps = self.get_anon_smaps_by_pid(self.vm.get_pid())
> +
> +        # size = rss = locked on every smap where rss not 0
> +        for smap in anon_smaps:
> +            if smap['Rss'] == 0:
> +                continue
> +            self.assertTrue(smap['Size'] == smap['Rss'] == smap['Locked'])
> +
> +    def test_memlock_onfault(self):
> +        self.common_vm_setup_with_memlock('on-fault')
> +
> +        anon_smaps = self.get_anon_smaps_by_pid(self.vm.get_pid())
> +
> +        # rss = locked on every smap and segment with rss < size exists
> +        exists = False
> +        for smap in anon_smaps:
> +            self.assertTrue(smap['Rss'] == smap['Locked'])
> +            if smap['Rss'] < smap['Size']:
> +                exists = True
> +        self.assertTrue(exists)
> +
> +    def _parse_anonymous_smaps(self, smaps_raw: str) -> List[Dict[str, int]]:
> +        result_segments = []
> +        current_segment = {}
> +        is_anonymous = False
> +
> +        for line in smaps_raw.split('\n'):
> +            if SMAPS_HEADER_PATTERN.match(line):
> +                if current_segment and is_anonymous:
> +                    result_segments.append(current_segment)
> +                current_segment = {}
> +                # anonymous segment header looks like this:
> +                # 7f3b8d3f0000-7f3b8d3f3000 rw-s 00000000 00:0f 1052
> +                # and non anonymous header looks like this:
> +                # 7f3b8d3f0000-7f3b8d3f3000 rw-s 00000000 00:0f 1052   [stack]
> +                is_anonymous = len(line.split()) == 5
> +            elif m := SMAPS_VALUE_PATTERN.match(line):
> +                current_segment[m.group(1)] = int(m.group(2))
> +
> +        if current_segment and is_anonymous:
> +            result_segments.append(current_segment)
> +
> +        return result_segments
> +
> +    def _get_raw_smaps_by_pid(self, pid: int) -> str:
> +        try:
> +            with open(f'/proc/{pid}/smaps', 'r') as f:
> +                return f.read()
> +        except IOError:

FWIW, I think it should rather be FileNotFoundError in modern Python, but 
IOError still seems to work, too.

> +            self.skipTest("Can't open smaps file of the process")
> +
> +
> +if __name__ == '__main__':
> +    MemlockTest.main()

Thanks, I'll queue the patch for my next pull request with functional patches.

  Thomas