# Copyright (c) 2010-2012 OpenStack Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import asyncio
from asyncio.exceptions import TimeoutError as AsyncioTimeoutError
import errno
import fcntl
import importlib.metadata
import logging
from types import TracebackType
from typing import Self, TYPE_CHECKING

import eventlet
import eventlet.asyncio
import eventlet.debug
import eventlet.greenthread
import eventlet.hubs
import eventlet.hubs.asyncio
import eventlet.patcher

if TYPE_CHECKING:
    # Needed until we bump our minimum to Python 3.11
    #
    # https://github.com/python/typeshed/issues/7855
    from _typeshed import SupportsWrite

    _StreamHandler = logging.StreamHandler[SupportsWrite[str]]
else:
    _StreamHandler = logging.StreamHandler

# We want the blocking APIs, because we set file descriptors to non-blocking.
os = eventlet.patcher.original("os")
# Used to communicate between real threads:
SimpleQueue = eventlet.patcher.original("queue").SimpleQueue
# Real OS-level threads:
threading = eventlet.patcher.original("threading")


class _BaseMutex:
    """Shared code for different mutex implementations."""

    def __init__(self) -> None:
        self.owner: int | None = None
        self.recursion_depth = 0

    def acquire(self, blocking: bool = True) -> bool:
        """Acquire the mutex.

        If called with blocking=False, returns True if the mutex was
        acquired and False if it wasn't. Otherwise, blocks until the mutex
        is acquired and returns True.
        This lock is recursive; the same greenthread may acquire it as many
        times as it wants to, though it must then release it that many times
        too.
        """
        current_greenthread_id = id(eventlet.greenthread.getcurrent())
        if self.owner == current_greenthread_id:
            self.recursion_depth += 1
            return True

        return self._acquire_eventlet(blocking, current_greenthread_id)

    def _acquire_eventlet(
        self, blocking: bool, current_greenthread_id: int
    ) -> bool:
        raise NotImplementedError()

    def release(self) -> None:
        """Release the mutex."""
        current_greenthread_id = id(eventlet.greenthread.getcurrent())
        if self.owner != current_greenthread_id:
            raise RuntimeError("cannot release un-acquired lock")

        if self.recursion_depth > 0:
            self.recursion_depth -= 1
            return

        self.owner = None
        self._release_eventlet()

    def _release_eventlet(self) -> None:
        raise NotImplementedError()

    def close(self) -> None:
        """Close the mutex.

        This releases its file descriptors.
        You can't use a mutex after it's been closed.
        """
        self.owner = None
        self.recursion_depth = 0

    def __enter__(self) -> 'Self':
        self.acquire()
        return self

    def __exit__(
        self,
        type_: type[BaseException] | None,
        value: BaseException | None,
        traceback: TracebackType | None,
    ) -> None:
        self.release()


class _ReallyPipeMutex(_BaseMutex):
    """Mutex using a pipe.

    Works across both greenlets and real threads, even at the same time.

    Class code copied from Swift's swift/common/utils.py
    Related eventlet bug: https://github.com/eventlet/eventlet/issues/432
    """

    def __init__(self) -> None:
        super().__init__()

        self.rfd, self.wfd = os.pipe()

        # You can't create a pipe in non-blocking mode; you must set it
        # later.
        rflags = fcntl.fcntl(self.rfd, fcntl.F_GETFL)
        fcntl.fcntl(self.rfd, fcntl.F_SETFL, rflags | os.O_NONBLOCK)
        os.write(self.wfd, b'-')  # start unlocked

        # Usually, it's an error to have multiple greenthreads all waiting
        # to read the same file descriptor. It's often a sign of inadequate
        # concurrency control; for example, if you have two greenthreads
        # trying to use the same memcache connection, they'll end up writing
        # interleaved garbage to the socket or stealing part of each others'
        # responses.
        #
        # In this case, we have multiple greenthreads waiting on the same
        # file descriptor by design. This lets greenthreads in real thread A
        # wait with greenthreads in real thread B for the same mutex.
        # Therefore, we must turn off eventlet's multiple-reader detection.
        #
        # It would be better to turn off multiple-reader detection for only
        # our calls to trampoline(), but eventlet does not support that.
        eventlet.debug.hub_prevent_multiple_readers(False)

    def _acquire_eventlet(
        self, blocking: bool, current_greenthread_id: int
    ) -> bool:
        while True:
            try:
                # If there is a byte available, this will read it and remove
                # it from the pipe. If not, this will raise OSError with
                # errno=EAGAIN.
                os.read(self.rfd, 1)
                self.owner = current_greenthread_id
                return True
            except OSError as err:
                if err.errno != errno.EAGAIN:
                    raise

                if not blocking:
                    return False

                # Tell eventlet to suspend the current greenthread until
                # self.rfd becomes readable. This will happen when someone
                # else writes to self.wfd.
                eventlet.hubs.trampoline(self.rfd, read=True)

    def _release_eventlet(self) -> None:
        os.write(self.wfd, b'X')

    def close(self) -> None:
        """Close the mutex.

        This releases its file descriptors.
        You can't use a mutex after it's been closed.
        """
        if self.wfd is not None:
            os.close(self.wfd)
            self.wfd = None
        if self.rfd is not None:
            os.close(self.rfd)
            self.rfd = None
        super().close()

    def __del__(self) -> None:
        # We need this so we don't leak file descriptors. Otherwise, if you
        # call get_logger() and don't explicitly dispose of it by calling
        # logger.logger.handlers[0].lock.close() [1], the pipe file
        # descriptors are leaked.
        #
        # This only really comes up in tests. Service processes tend to call
        # get_logger() once and then hang on to it until they exit, but the
        # test suite calls get_logger() a lot.
        #
        # [1] and that's a completely ridiculous thing to expect callers to
        # do, so nobody does it and that's okay.
        self.close()


class _AsyncioMutex(_BaseMutex):
    """Alternative implementation of mutex for eventlet asyncio hub.

    When using the eventlet asyncio hub, multiple file descriptors as readers
    aren't supported.  So instead, we use two levels of locking:

        1. A normal RLock, for locking across OS threads.

        2. An ``asyncio.Lock`` for locking across greenlets within a single OS
           thread (each OS thread running greenlets has its own asyncio loop)
    """

    def __init__(self) -> None:
        super().__init__()
        self._asyncio_lock = asyncio.Lock()
        self._threading_lock = threading.RLock()

    async def _asyncio_acquire(
        self, blocking: bool, current_greenthread_id: int
    ) -> bool:
        if blocking:
            timeout = None
        else:
            timeout = 0.000001
        try:
            await asyncio.wait_for(
                self._asyncio_lock.acquire(), timeout=timeout
            )
        except AsyncioTimeoutError:
            return False
        else:
            self.owner = current_greenthread_id
            return True

    def _acquire_eventlet(
        self, blocking: bool, current_greenthread_id: int
    ) -> bool:
        return eventlet.asyncio.spawn_for_awaitable(  # type: ignore
            self._asyncio_acquire(blocking, current_greenthread_id)
        ).wait()

    def acquire(self, blocking: bool = True) -> bool:
        """Acquire the mutex.

        If called with blocking=False, returns True if the mutex was
        acquired and False if it wasn't. Otherwise, blocks until the mutex
        is acquired and returns True.
        This lock is recursive; the same greenthread may acquire it as many
        times as it wants to, though it must then release it that many times
        too.
        """
        # First, acquire the RLock:
        rlock_acquired = self._threading_lock.acquire(blocking=False)
        if not rlock_acquired and not blocking:
            return False
        # Simulate blocking while allowing other greenlets to run:
        while not rlock_acquired:
            rlock_acquired = self._threading_lock.acquire(blocking=False)
            # The chance of hitting this path is usually quite low (it requires
            # multiple _OS_ threads having a conflict) and sleeping for too
            # short a time leads to very slow results in one of the tests,
            # perhaps due to lots of context switching overhead. So compromise
            # on a short but not too-short sleep.
            eventlet.sleep(0.0001)
        # Then, do the eventlet locking:
        return super().acquire(blocking=blocking)

    def _release_eventlet(self) -> None:
        self._asyncio_lock.release()

    def release(self) -> None:
        """Release the mutex."""
        # We release in reverse order from acquire(), first eventlet and then
        # the RLock:
        super().release()
        self._threading_lock.release()

    def close(self) -> None:
        """Close the mutex."""
        del self._asyncio_lock
        del self._threading_lock
        super().close()


_HUB = eventlet.hubs.get_hub()
if isinstance(_HUB, eventlet.hubs.asyncio.Hub):
    major, minor, patch = map(
        int, importlib.metadata.version("eventlet").split(".")[:3]
    )
    if (major, minor, patch) < (0, 38, 2):
        raise RuntimeError(
            "eventlet 0.38.2 or later is required when using asyncio hub"
        )
    PipeMutex: type[_AsyncioMutex | _ReallyPipeMutex] = _AsyncioMutex
else:
    PipeMutex = _ReallyPipeMutex


def pipe_createLock(self: _StreamHandler) -> None:
    """Replacement for logging.Handler.createLock method."""
    self.lock = PipeMutex()  # type: ignore