utils: merge per-platform utils into `platform.py`

This allows us to simplify running mypy, as we can just mypy the entire
utils module regardless of the platform we're using. As an added bonus,
this means we open one less module.

1 files changed, 79 insertions, 5 deletions

diff --git a/mesonbuild/utils/platform.py b/mesonbuild/utils/platform.py
index f9eaa47ed..538c10890 100644
--- a/mesonbuild/utils/platform.py
+++ b/mesonbuild/utils/platform.py
@@ -2,17 +2,19 @@
 # Copyright 2012-2021 The Meson development team
 # Copyright © 2021-2023 Intel Corporation
 
-from __future__ import annotations
+"""Utility functions with platform specific implementations."""
 
-"""base classes providing no-op functionality.."""
+from __future__ import annotations
 
 import enum
 import os
+import sys
 import typing as T
 
 from .. import mlog
+from .core import MesonException
 
-__all__ = ['DirectoryLock', 'DirectoryLockAction', 'DirectoryLockBase']
+__all__ = ['DirectoryLock', 'DirectoryLockAction']
 
 class DirectoryLockAction(enum.Enum):
     IGNORE = 0
@@ -34,5 +36,77 @@ class DirectoryLockBase:
     def __exit__(self, *args: T.Any) -> None:
         pass
 
-class DirectoryLock(DirectoryLockBase):
-    pass
+
+if sys.platform == 'win32':
+    import msvcrt
+
+    class DirectoryLock(DirectoryLockBase):
+
+        def __enter__(self) -> None:
+            try:
+                self.lockfile = open(self.lockpath, 'w+', encoding='utf-8')
+            except (FileNotFoundError, IsADirectoryError):
+                # For FileNotFoundError, there is nothing to lock.
+                # For IsADirectoryError, something is seriously wrong.
+                raise
+            except OSError:
+                if self.action == DirectoryLockAction.IGNORE or self.optional:
+                    return
+
+            try:
+                mode = msvcrt.LK_LOCK
+                if self.action != DirectoryLockAction.WAIT:
+                    mode = msvcrt.LK_NBLCK
+                msvcrt.locking(self.lockfile.fileno(), mode, 1)
+            except BlockingIOError:
+                self.lockfile.close()
+                if self.action == DirectoryLockAction.IGNORE:
+                    return
+                raise MesonException(self.err)
+            except PermissionError:
+                self.lockfile.close()
+                raise MesonException(self.err)
+
+        def __exit__(self, *args: T.Any) -> None:
+            if self.lockfile is None or self.lockfile.closed:
+                return
+            msvcrt.locking(self.lockfile.fileno(), msvcrt.LK_UNLCK, 1)
+            self.lockfile.close()
+else:
+    import fcntl
+
+    class DirectoryLock(DirectoryLockBase):
+
+        def __enter__(self) -> None:
+            try:
+                self.lockfile = open(self.lockpath, 'w+', encoding='utf-8')
+            except (FileNotFoundError, IsADirectoryError):
+                # For FileNotFoundError, there is nothing to lock.
+                # For IsADirectoryError, something is seriously wrong.
+                raise
+            except OSError:
+                if self.action == DirectoryLockAction.IGNORE or self.optional:
+                    return
+
+            try:
+                flags = fcntl.LOCK_EX
+                if self.action != DirectoryLockAction.WAIT:
+                    flags = flags | fcntl.LOCK_NB
+                fcntl.flock(self.lockfile, flags)
+            except BlockingIOError:
+                self.lockfile.close()
+                if self.action == DirectoryLockAction.IGNORE:
+                    return
+                raise MesonException(self.err)
+            except PermissionError:
+                self.lockfile.close()
+                raise MesonException(self.err)
+            except OSError as e:
+                self.lockfile.close()
+                raise MesonException(f'Failed to lock directory {self.lockpath}: {e.strerror}')
+
+        def __exit__(self, *args: T.Any) -> None:
+            if self.lockfile is None or self.lockfile.closed:
+                return
+            fcntl.flock(self.lockfile, fcntl.LOCK_UN)
+            self.lockfile.close()