Change in vdsm[master]: lib: pthread: {get,set}affinity support

Francesco Romani has uploaded a new change for review.

Change subject: lib: pthread: {get,set}affinity support
......................................................................

lib: pthread: {get,set}affinity support

To improve the performance of VDSM, it can be worth
pinning all pyrhon threads on one core.
This makes sense because of the cpython's GIL, if the
threads jumps from one core to another, we will get
all the disadvantages (cache invalidations) without
any benefit.

This patch prepares the ground exposing get/set affinity
functions on VDSM.

https://bugzilla.redhat.com/1247075
Change-Id: Ie1544ec372095599b39b9ef124a4426a6e73fd4e
Signed-off-by: Francesco Romani <fromani(a)redhat.com&gt;
---
M lib/vdsm/pthread.py
M tests/pthreadTests.py
2 files changed, 203 insertions(+), 0 deletions(-)


  git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/82/45282/1

diff --git a/lib/vdsm/pthread.py b/lib/vdsm/pthread.py
index 851afd1..059d632 100644
--- a/lib/vdsm/pthread.py
+++ b/lib/vdsm/pthread.py
@@ -20,7 +20,9 @@
 
 from __future__ import absolute_import
 import ctypes
+import itertools
 import logging
+import multiprocessing
 import threading
 
 
@@ -50,6 +52,31 @@
     logging.warning(
         'pthread_{set,get}name_np unavailable. '
         'System thread names will not be set.')
+
+
+_pthread_setaffinity_np_proto = ctypes.CFUNCTYPE(
+    ctypes.c_int, ctypes.c_ulong, ctypes.c_size_t, ctypes.c_void_p)
+_pthread_getaffinity_np_proto = ctypes.CFUNCTYPE(
+    ctypes.c_int, ctypes.c_ulong, ctypes.c_size_t, ctypes.c_void_p)
+
+
+try:
+    _pthread_setaffinity_np = _pthread_setaffinity_np_proto(
+        ('pthread_setaffinity_np', _LIBPTHREAD))
+
+    _pthread_getaffinity_np = _pthread_getaffinity_np_proto(
+        ('pthread_getaffinity_np', _LIBPTHREAD))
+
+except AttributeError:
+    def _pthread_setaffinity_np(ident, cpusetsize, cpuset):
+        raise RuntimeError("setaffinity not available")
+
+    def _pthread_getaffinity_np(ident, cpusetsize, cpuset):
+        raise RuntimeError("getaffinity not available")
+
+    logging.warning(
+        'pthread_{set,get}affinity_np unavailable. '
+        'CPU affinity for threads will not be set.')
 
 
 def setname(name):
@@ -88,3 +115,149 @@
     thread = threading.current_thread()
     _pthread_getname_np(thread.ident, buf, bufsize)
     return buf.value
+
+
+def setaffinity(cpuset):
+    cpu_bitset = cpuset.to_bits()
+
+    thread = threading.current_thread()
+    ret = _pthread_setaffinity_np(
+        thread.ident,
+        ctypes.sizeof(cpu_bitset),
+        ctypes.byref(cpu_bitset))
+    if ret < 0:
+        raise RuntimeError("pthread_set_affinity_np failed err=%i", ret)
+
+
+def getaffinity():
+    cpu_bitset = _CPUBitSet()
+
+    thread = threading.current_thread()
+    ret = _pthread_getaffinity_np(
+        thread.ident,
+        ctypes.sizeof(cpu_bitset),
+        ctypes.byref(cpu_bitset))
+    if ret < 0:
+        raise RuntimeError("pthread_get_affinity_np failed err=%i", ret)
+
+    return CPUSet.from_bits(cpu_bitset)
+
+
+class CPUSet(object):
+
+    @classmethod
+    def from_bits(cls, cpu_bitset):
+        obj = cls()
+        obj._cpus = _cpu_bitset_to_set(cpu_bitset)
+        return obj
+
+    @classmethod
+    def from_str(cls, indexes):
+        obj = cls()
+        obj._cpus = _indexes_str_to_set(indexes)
+        return obj
+
+    def __init__(self):
+        """
+        Do not call directly, use from_* functions instead!
+        """
+        self._cpus = set()
+
+    def to_bits(self):
+        return _indexes_set_to_bitset(self.cpus)
+
+    def to_str(self):
+        return _indexes_set_to_str(self._cpus)
+
+    __str__ = to_str
+
+    # for test/debug purposes
+    def to_set(self):
+        return self._cpus
+
+
+# namedtuple is unfit: we need mutability
+class _Group(object):
+
+    __slots__ = ('first', 'last')
+
+    def __init__(self, first, last):
+        self.first = first
+        self.last = last
+
+    def __str__(self):
+        if self.first == self.last:
+            return '%d' % (self.first)
+        else:
+            return '%d-%d' % (self.first, self.last)
+
+    def adjacent(self, value):
+        return self.last == value - 1
+
+    def expand_right(self, value):
+        self.last = value
+
+
+# CAUTION: on x86_64 the CPU_SET macro is mapped on
+# unsigned long, which is uint64 under linux ABI.
+# hence, we need at least one c_uint64 here.
+
+class _CPUBitSet(ctypes.Structure):
+    #  TODO: adjust the size dinamycally?
+    #  TODO: check all the cpus are addressable?
+    _fields_ = [("bits", ctypes.c_uint64 * 8)]
+
+
+def _group_consecutive(indexes):
+    # TODO: improve algorhytm. This one is bad.
+    items = []
+    for index in sorted(indexes):
+        if items and items[-1].adjacent(index):
+            items[-1].expand_right(index)
+        else:
+            items.append(_Group(index, index))
+    return items
+
+
+def _cpu_bitset_to_set(cpu_bitset):
+    # TODO: improve algorhytm. This one is bad.
+    cpu_idx, cpus = 0, set()
+
+    for i in xrange(len(cpu_bitset.bits)):
+        mask = cpu_bitset.bits[i]
+        while mask:
+            if mask & 1:
+                cpus.add(cpu_idx)
+            cpu_idx += 1
+            mask = mask >> 1
+
+    return cpus
+
+
+def _indexes_set_to_str(cpus):
+    return ','.join(str(pair)
+                    for pair in _group_consecutive(cpus))
+
+
+def _indexes_set_to_bitset(cpus):
+    bitset = _CPUBitSet()
+    segment_size = ctypes.sizeof(bitset[0]) * 8  # TODO
+    for cpu_idx in cpus:
+        bitset[cpu_idx / segment_size] |= (
+            1 << (cpu_idx % segment_size))
+    return bitset
+
+
+def _indexes_str_to_set(indexes):
+    cpus = set()
+
+    for cpu in indexes.split(','):
+        if '-' in cpu:
+            begin_str, end_str = cpu.split('-')
+            begin, end = int(begin_str), int(end_str)
+            for idx in range(begin, end+1):
+                cpus.add(idx)
+        else:
+            cpus.add(int(cpu))
+
+    return cpus
diff --git a/tests/pthreadTests.py b/tests/pthreadTests.py
index 598e544..a487350 100644
--- a/tests/pthreadTests.py
+++ b/tests/pthreadTests.py
@@ -22,6 +22,7 @@
 
 from vdsm import pthread
 
+from testlib import expandPermutations, permutations
 from testlib import VdsmTestCase as TestCaseBase
 
 
@@ -68,3 +69,32 @@
         self.assertEqual(parent_name, pthread.getname())
 
         self.done.set()
+
+
+@expandPermutations
+class CPUSetTests(TestCaseBase):
+
+    @permutations([['0', set((0,))],
+                   ['0,1', set((0,1))],
+                   ['0,1,2,3', set((0,1,2,3))],
+                   ['0-1', set((0,1))],
+                   ['0-3', set((0,1,2,3))],
+                   ['0,2-3', set((0,2,3))],
+                   ['0,1-3', set((0,1,2,3))],
+                   ['0-2,3', set((0,1,2,3))]])
+    def test_from_str(self, given, expected):
+        cpus = pthread.CPUSet.from_str(given)
+        self.assertEqual(cpus.to_set(), expected)
+
+
+    @permutations([['0', '0'],
+                   ['0,1', '0-1'],
+                   ['0,1,2,3', '0-3'],
+                   ['0-1', '0-1'],
+                   ['0-3', '0-3'],
+                   ['0,2-3', '0,2-3'],
+                   ['0,1-3', '0-3'],
+                   ['0-2,3', '0-3']])
+    def test_str_round_trip(self, given, expected):
+        cpus = pthread.CPUSet.from_str(given)
+        self.assertEqual(cpus.to_str(), expected)


-- 
To view, visit https://gerrit.ovirt.org/45282
To unsubscribe, visit https://gerrit.ovirt.org/settings

Gerrit-MessageType: newchange
Gerrit-Change-Id: Ie1544ec372095599b39b9ef124a4426a6e73fd4e
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Francesco Romani <fromani(a)redhat.com&gt;