Change in vdsm[master]: netlink: Introduce socket pool

asegurap at redhat.com asegurap at redhat.com
Tue Feb 18 00:05:17 UTC 2014


Antoni Segura Puimedon has uploaded a new change for review.

Change subject: netlink: Introduce socket pool
......................................................................

netlink: Introduce socket pool

If a lot of netlink requests happened _nl_connect would fail at
the C level, more concretely when doing bind, which would return
EADDRINUSE.

This patch addresses that issue as well as reduces the the resource
consumption of the module by reusing the netlink sockets. The reusing
is done by a semaphore protected socket pool.

Change-Id: I657ac3d3e0c2661ce73bdef9aa807ead888a42eb
Signed-off-by: Antoni S. Puimedon <asegurap at redhat.com>
---
M lib/vdsm/netlink.py
1 file changed, 84 insertions(+), 30 deletions(-)


  git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/03/24603/1

diff --git a/lib/vdsm/netlink.py b/lib/vdsm/netlink.py
index eeb8e95..e50feba 100644
--- a/lib/vdsm/netlink.py
+++ b/lib/vdsm/netlink.py
@@ -21,6 +21,7 @@
 from ctypes import (CDLL, CFUNCTYPE, c_char, c_char_p, c_int, c_void_p,
                     c_size_t, get_errno, sizeof)
 from functools import partial
+from threading import BoundedSemaphore, Lock
 import errno
 
 NETLINK_ROUTE = 0
@@ -78,25 +79,82 @@
     'rtnl_scope2str', LIBNL))
 
 
+class NLSocketPoolError(Exception):
+    pass
+
+
+class NLSocketPool(object):
+    """Pool of netlink sockets."""
+    def __init__(self, size=3):
+        assert size > 0
+        self._size = size
+        self._nl_sockets = set()
+        self._semaphore = BoundedSemaphore(size)
+        self._population_lock = Lock()
+        self._populated = False
+
+    def _populate(self):
+        """Allocates size amount of netlink sockets."""
+        if self._nl_sockets:
+            raise NLSocketPoolError('Pool already populated')
+
+        allocated = []
+        try:
+            for _ in range(self._size):
+                allocated.append(_get_nl_socket())
+        except IOError:
+            for sock in allocated:
+                _nl_handle_destroy(sock)  # Free handles so caller can retry
+            raise
+
+        self._nl_sockets.update(allocated)
+        self._populated = True
+
+    @contextmanager
+    def socket(self):
+        """Takes a netlink socket from the pool and returns it afterwards."""
+        if not self._populated:
+            if self._population_lock.acquire(False):
+                try:
+                    self._populate()
+                finally:
+                    self._population_lock.release()
+            else:  # We weren't first. Let's wait for population to finish
+                with self._population_lock:
+                    pass
+
+        with self._semaphore:
+            socket = self._nl_sockets.pop()
+            try:
+                yield socket
+            finally:
+                self._nl_sockets.add(socket)
+
+
+_nl_socket_pool = NLSocketPool()
+
+
 def iter_links():
     """Generator that yields an information dictionary for each link of the
     system."""
-    with _nl_link_cache() as cache:
-        link = _nl_cache_get_first(cache)
-        while link:
-            yield _link_info(cache, link)
-            link = _nl_cache_get_next(link)
+    with _nl_socket_pool.socket() as sock:
+        with _nl_link_cache(sock) as cache:
+            link = _nl_cache_get_first(cache)
+            while link:
+                yield _link_info(cache, link)
+                link = _nl_cache_get_next(link)
 
 
 def iter_addrs():
     """Generator that yields an information dictionary for each network address
     in the system."""
-    with _nl_addr_cache() as addr_cache:
-        with _nl_link_cache() as link_cache:  # For index to label resolution
-            addr = _nl_cache_get_first(addr_cache)
-            while addr:
-                yield _addr_info(link_cache, addr)
-                addr = _nl_cache_get_next(addr)
+    with _nl_socket_pool.socket() as sock:
+        with _nl_addr_cache(sock) as addr_cache:
+            with _nl_link_cache(sock) as link_cache:  # for index to label
+                addr = _nl_cache_get_first(addr_cache)
+                while addr:
+                    yield _addr_info(link_cache, addr)
+                    addr = _nl_cache_get_next(addr)
 
 
 def get_link(name):
@@ -109,34 +167,30 @@
         return _link_info(cache, link)
 
 
- at contextmanager
-def _open_nl_socket():
-    """Provides a Netlink socket and closes and destroys it upon exit."""
+def _get_nl_socket():
+    """Returns an open netlink socket."""
     handle = _nl_handle_alloc()
     if handle is None:
         raise IOError(get_errno(), 'Failed to allocate netlink handle')
-    try:
-        err = _nl_connect(handle, NETLINK_ROUTE)
-        if err:
-            raise IOError(-err, 'Failed to connect to netlink socket.')
-        yield handle
-    finally:
-        # handle is automatically disconnected on destroy.
+
+    err = _nl_connect(handle, NETLINK_ROUTE)
+    if err:
         _nl_handle_destroy(handle)
+        raise IOError(-err, 'Failed to connect to netlink socket.')
+    return handle
 
 
 @contextmanager
-def _cache_manager(cache_allocator):
+def _cache_manager(cache_allocator, sock):
     """Provides a cache using cache_allocator and frees it and its links upon
     exit."""
-    with _open_nl_socket() as sock:
-        cache = cache_allocator(sock)
-        if cache is None:
-            raise IOError(get_errno(), 'Failed to allocate the cache.')
-        try:
-            yield cache
-        finally:
-            _nl_cache_free(cache)
+    cache = cache_allocator(sock)
+    if cache is None:
+        raise IOError(get_errno(), 'Failed to allocate the cache.')
+    try:
+        yield cache
+    finally:
+        _nl_cache_free(cache)
 
 
 _nl_link_cache = partial(_cache_manager, _rtnl_link_alloc_cache)


-- 
To view, visit http://gerrit.ovirt.org/24603
To unsubscribe, visit http://gerrit.ovirt.org/settings

Gerrit-MessageType: newchange
Gerrit-Change-Id: I657ac3d3e0c2661ce73bdef9aa807ead888a42eb
Gerrit-PatchSet: 1
Gerrit-Project: vdsm
Gerrit-Branch: master
Gerrit-Owner: Antoni Segura Puimedon <asegurap at redhat.com>


More information about the vdsm-patches mailing list