ospd_openvas.lock.LockFile._acquire_lock() - Code Metrics - Inspection of "Merge pull request #459 from greenbone/mergify/bp/..." - greenbone/ospd-openvas - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 9b3e23...2a9884 )

by Björn

created 2021-07-01 14:25 UTC

ospd_openvas.lock.LockFile._acquire_lock() C

↳ Parent: ospd_openvas.lock

Complexity

Conditions

Size

Total Lines	64
Code Lines	46

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	8
eloc	46
nop	1
dl	0
loc	64
rs	6.9006
c	0
b	0
f	0

How to fix Long Method

# -*- coding: utf-8 -*-
# Copyright (C) 2014-2021 Greenbone Networks GmbH
#
# SPDX-License-Identifier: AGPL-3.0-or-later
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.


import logging
import time
import fcntl

from pathlib import Path

logger = logging.getLogger(__name__)


class LockFile:
    def __init__(self, path: Path):
        self._lock_file_path = path
        self._has_lock = False
        self._fd = None

    def has_lock(self) -> bool:
        return self._has_lock

    def _acquire_lock(self) -> "LockFile":
        """Acquire a lock by creating a lock file."""
        if self.has_lock():
            return self

        parent_dir = self._lock_file_path.parent

        try:
            # create parent directories recursively
            parent_dir.mkdir(parents=True, mode=0o770, exist_ok=True)
        except OSError as e:
            logger.error(
                "Could not create parent dir %s for lock file. %s",
                str(parent_dir),
                e,
            )
            return self

        try:
            # Open the fd with append flag to create the file
            # if not exists and to avoid deleting the content
            # something else wrote in it.
            self._fd = self._lock_file_path.open('a')
        except Exception as e:  # pylint: disable=broad-except
            logger.error(
                "Failed to open lock file %s. %s",
                str(self._lock_file_path),
                e,
            )
            try:
                self._fd.close()
                self._fd = None
            except Exception:  # pylint: disable=broad-except
                pass
            return self

        try:
            self._lock_file_path.chmod(0o660)
        except OSError as e:
            # ignore error because it is very likely that the file exists, has
            # the correct permissions but we are not the owner
            logger.debug(
                "Could not change permissions of lock file %s",
                str(self._lock_file_path),
            )

        # Try to acquire the lock.
        try:
            fcntl.flock(self._fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
            self._has_lock = True
            logger.debug("Created lock file %s.", str(self._lock_file_path))
        except BlockingIOError as e:
            logger.debug(
                "Failed to lock the file %s. %s",
                str(self._lock_file_path),
                e,
            )
            try:
                self._fd.close()
                self._fd = None
            except Exception:  # pylint: disable=broad-except
                pass

        return self

    def wait_for_lock(self):
        while not self.has_lock():
            self._acquire_lock()
            time.sleep(10)

        return self

    def _release_lock(self) -> None:
        """Release the lock by deleting the lock file"""
        if self.has_lock() and self._fd:
            fcntl.flock(self._fd, fcntl.LOCK_UN)
            self._fd.close()
            self._fd = None
            self._has_lock = False
            logger.debug(
                "Removed lock from file %s.", str(self._lock_file_path)
            )

    def __enter__(self):
        self._acquire_lock()
        return self

    def __exit__(self, exc_type, exc_value, exc_tb):
        self._release_lock()


1			# -- coding: utf-8 --
2			# Copyright (C) 2014-2021 Greenbone Networks GmbH
3			#
4			# SPDX-License-Identifier: AGPL-3.0-or-later
5			#
6			# This program is free software: you can redistribute it and/or modify
7			# it under the terms of the GNU Affero General Public License as
8			# published by the Free Software Foundation, either version 3 of the
9			# License, or (at your option) any later version.
10			#
11			# This program is distributed in the hope that it will be useful,
12			# but WITHOUT ANY WARRANTY; without even the implied warranty of
13			# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14			# GNU Affero General Public License for more details.
15			#
16			# You should have received a copy of the GNU Affero General Public License
17			# along with this program. If not, see <http://www.gnu.org/licenses/>.
18
19
20			import logging
21			import time
22			import fcntl
23
24			from pathlib import Path
25
26			logger = logging.getLogger(__name__)
27
28
29			class LockFile:
30			def __init__(self, path: Path):
31			self._lock_file_path = path
32			self._has_lock = False
33			self._fd = None
34
35			def has_lock(self) -> bool:
36			return self._has_lock
37
38			def _acquire_lock(self) -> "LockFile":
39			"""Acquire a lock by creating a lock file."""
40			if self.has_lock():
41			return self
42
43			parent_dir = self._lock_file_path.parent
44
45			try:
46			# create parent directories recursively
47			parent_dir.mkdir(parents=True, mode=0o770, exist_ok=True)
48			except OSError as e:
49			logger.error(
50			"Could not create parent dir %s for lock file. %s",
51			str(parent_dir),
52			e,
53			)
54			return self
55
56			try:
57			# Open the fd with append flag to create the file
58			# if not exists and to avoid deleting the content
59			# something else wrote in it.
60			self._fd = self._lock_file_path.open('a')
61			except Exception as e: # pylint: disable=broad-except
62			logger.error(
63			"Failed to open lock file %s. %s",
64			str(self._lock_file_path),
65			e,
66			)
67			try:
68			self._fd.close()
69			self._fd = None
70			except Exception: # pylint: disable=broad-except
71			pass
72			return self
73
74			try:
75			self._lock_file_path.chmod(0o660)
76			except OSError as e:
77			# ignore error because it is very likely that the file exists, has
78			# the correct permissions but we are not the owner
79			logger.debug(
80			"Could not change permissions of lock file %s",
81			str(self._lock_file_path),
82			)
83
84			# Try to acquire the lock.
85			try:
86			fcntl.flock(self._fd, fcntl.LOCK_EX \| fcntl.LOCK_NB)
87			self._has_lock = True
88			logger.debug("Created lock file %s.", str(self._lock_file_path))
89			except BlockingIOError as e:
90			logger.debug(
91			"Failed to lock the file %s. %s",
92			str(self._lock_file_path),
93			e,
94			)
95			try:
96			self._fd.close()
97			self._fd = None
98			except Exception: # pylint: disable=broad-except
99			pass
100
101			return self
102
103			def wait_for_lock(self):
104			while not self.has_lock():
105			self._acquire_lock()
106			time.sleep(10)
107
108			return self
109
110			def _release_lock(self) -> None:
111			"""Release the lock by deleting the lock file"""
112			if self.has_lock() and self._fd:
113			fcntl.flock(self._fd, fcntl.LOCK_UN)
114			self._fd.close()
115			self._fd = None
116			self._has_lock = False
117			logger.debug(
118			"Removed lock from file %s.", str(self._lock_file_path)
119			)
120
121			def __enter__(self):
122			self._acquire_lock()
123			return self
124
125			def __exit__(self, exc_type, exc_value, exc_tb):
126			self._release_lock()
127

greenbone / ospd-openvas

Push — master ( 9b3e23...2a9884 )

ospd_openvas.lock.LockFile._acquire_lock() C

Complexity

Size

Duplication

Importance

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like