
from __future__ import absolute_import

import asyncore
import collections
import errno
import heapq
import logging
import os
import select

import six

from vdsm.common import filecontrol
from vdsm.common import osutils
from vdsm.common import time

log = logging.getLogger("storage.asyncevent")


class EventLoop(object):

    def __init__(self):
        self._channels = {}
        self._scheduled = []
        self._ready = collections.deque()
        self._running = False
        self._closed = False
        self._waker = self.create_dispatcher(Waker)


    def run_forever(self):
        self._check_closed()
        if self._running:
            raise RuntimeError('Event loop is running.')
        self._running = True
        try:
            log.info("Starting %s", self)
            while True:
                try:
                    self._run_once()
                except _StopError:
                    break
        finally:
            self._running = False
            log.debug("%s stopped", self)

    def _run_once(self):
        while self._scheduled and self._scheduled[0]._cancelled:
            heapq.heappop(self._scheduled)

        timeout = None
        if self._ready:
            timeout = 0
        elif self._scheduled:
            when = self._scheduled[0]._when
            timeout = max(0, when - self.time())

        events = poll(timeout, map=self._channels)
        self._process_events(events)

        now = self.time()
        while self._scheduled:
            handle = self._scheduled[0]
            if handle._when > now:
                break
            heapq.heappop(self._scheduled)
            self._ready.append(handle)

        ntodo = len(self._ready)
        for i in range(ntodo):
            handle = self._ready.popleft()
            if handle._cancelled:
                continue
            handle._run()


    def _process_events(self, events):
        for fd, obj, flags in events:
            handle = Handle(self._readwrite, (fd, obj, flags))
            self._ready.append(handle)

    def _readwrite(self, fd, obj, flags):
        if self._channels.get(fd) is obj:
            asyncore.readwrite(obj, flags)

    def stop(self):
        self._check_closed()
        log.info("Stopping %r", self)
        self.call_soon(_raise_stop_error)

    def close(self):
        if self._running:
            raise RuntimeError("Cannot close a running event loop")
        if self._closed:
            return
        log.debug("Closing %r", self)
        self._closed = True
        self._waker.close()
        self._ready.clear()
        del self._scheduled[:]
        asyncore.close_all(map=self._channels)


    def call_soon(self, callback, *args):
        self._check_closed()
        return self._call_soon(callback, args)

    def call_soon_threadsafe(self, callback, *args):
        handle = self._call_soon(callback, args)
        self._write_to_self()
        return handle

    def _call_soon(self, callback, args):
        handle = Handle(callback, args)
        self._ready.append(handle)
        return handle

    def _write_to_self(self):
        self._waker.wakeup()


    def call_later(self, delay, callback, *args):
        return self.call_at(self.time() + delay, callback, *args)

    def call_at(self, when, callback, *args):
        self._check_closed()
        timer = Timer(when, callback, args)
        heapq.heappush(self._scheduled, timer)
        return timer


    def create_dispatcher(self, dispatcher_class, *args, **kwargs):
        assert "map" not in kwargs, "map is set by the event loop"
        return dispatcher_class(*args, map=self._channels, **kwargs)


    def is_running(self):
        return self._running

    def is_closed(self):
        return self._closed


    def time(self):
        return time.monotonic_time()


    def _check_closed(self):
        if self._closed:
            raise RuntimeError('Event loop is closed')


    def __repr__(self):
        return ("<{self.__class__.__name__} "
                "running={self._running} "
                "closed={self._closed} "
                "at 0x{addr}>").format(self=self, addr=id(self))


class _StopError(BaseException):
    """ Raised for stopping the event loop """


def _raise_stop_error():
    raise _StopError


def poll(timeout=0.0, map=None):

    if not map:
        return []

    if timeout is not None:
        timeout = int(timeout * 1000)

    pollster = select.poll()

    for fd, obj in six.iteritems(map):
        flags = 0
        if obj.readable():
            flags |= select.POLLIN | select.POLLPRI
        if obj.writable() and not obj.accepting:
            flags |= select.POLLOUT
        if flags:
            pollster.register(fd, flags)

    try:
        r = pollster.poll(timeout)
    except select.error as e:
        if e[0] != errno.EINTR:
            raise
        return []



class Handle(object):

    def __init__(self, callback, args):
        self._callback = callback
        self._args = args

    def cancel(self):
        self._callback = _CANCELLED
        self._args = None

    @property
    def _cancelled(self):
        return self._callback is _CANCELLED

    def _run(self):
        try:
            self._callback(*self._args)
        except Exception:
            log.exception("Unhandled error in %s", self)
        self._callback = self._args = None

    def _repr_info(self):
        info = [self.__class__.__name__]
        if self._cancelled:
            info.append("cancelled")
        else:
            info.append("callback=%s" % self._callback)
            if self._args:
                info.append("args=%s" % self._args)
        return info

    def __repr__(self):
        return "<%s at 0x%x>" % (" ".join(self._repr_info()), id(self))


class Timer(Handle):

    def __init__(self, when, callback, args):
        super(Timer, self).__init__(callback, args)
        self._when = when


    def __lt__(self, other):
        return self._when < other._when

    def __le__(self, other):
        return self._when <= other._when

    def __gt__(self, other):
        return self._when > other._when

    def __ge__(self, other):
        return self._when >= other._when


    def _repr_info(self):
        info = super(Timer, self)._repr_info()
        info.insert(1, "when=%.6f" % self._when)
        return info


def _CANCELLED(*args):
    pass


class Waker(asyncore.file_dispatcher):

    def __init__(self, map):
        rfd, wfd = os.pipe()
        asyncore.file_dispatcher.__init__(self, rfd, map=map)
        filecontrol.set_close_on_exec(self._fileno)
        filecontrol.set_close_on_exec(wfd)
        filecontrol.set_non_blocking(wfd)
        self._wfd = wfd

    def wakeup(self):
        try:
            osutils.uninterruptible(os.write, self._wfd, b"\0")
        except OSError as e:
            if self.closing:
                return
            if e.errno == errno.EAGAIN:
                return
            raise

    def handle_read(self):
        osutils.uninterruptible(self.socket.read, 1024)

    def handle_close(self):
        log.error("Wakeup read end was closed")
        self.close()

    def writable(self):
        return False

    def close(self):
        if self.closing:
            return
        self.closing = True
        wfd = self._wfd
        self._wfd = -1
        os.close(wfd)
        asyncore.file_dispatcher.close(self)


class BufferedReader(asyncore.file_dispatcher):

    def __init__(self, fd, complete, bufsize=4096, map=None):
        asyncore.file_dispatcher.__init__(self, fd, map=map)
        filecontrol.set_close_on_exec(self._fileno)
        self._complete = complete
        self._bufsize = bufsize
        self._data = bytearray()

    def handle_read(self):
        chunk = self.socket.read(self._bufsize)
        if not chunk:
            self.handle_close()
            return
        self._data += chunk

    def handle_close(self):
        if self._complete:
            complete = self._complete
            self._complete = None
            complete(self._data)
        self.close()

    def handle_error(self):
        log.exception("Unhandled error in %s", self)
        self.handle_close()

    def close(self):
        if self.closing:
            return
        self.closing = True
        self._complete = None
        asyncore.file_dispatcher.close(self)

    def writable(self):
        return False


class Reaper(object):

    def __init__(self, loop, proc, complete, min_interval=2**-5,
                 max_interval=1.0):
        self._loop = loop
        self._proc = proc
        self._complete = complete
        self._interval = min_interval
        self._max_interval = max_interval
        self._count = 0
        self._loop.call_later(self._interval, self.reap)

    def reap(self):
        self._count += 1
        rc = self._proc.poll()
        if rc is None:
            if self._interval < self._max_interval:
                self._interval *= 2
            self._loop.call_later(self._interval, self.reap)
            return
        log.debug("Process %s terminated (count=%d)", self._proc, self._count)
        self._complete(rc)
        self._complete = None
