

from __future__ import absolute_import

import logging
import re
import threading

from vdsm.common import constants
from vdsm.common import cmdutils
from vdsm.common import concurrent
from vdsm.common.compat import subprocess
from vdsm.storage import asyncevent
from vdsm.storage import asyncutils
from vdsm.storage import exception

EXEC_ERROR = 127

_log = logging.getLogger("storage.check")


class CheckService(object):

    def __init__(self):
        self._lock = threading.Lock()
        self._loop = asyncevent.EventLoop()
        self._thread = concurrent.thread(self._loop.run_forever,
                                         name="check/loop")
        self._checkers = {}

    def start(self):
        _log.info("Starting check service")
        self._thread.start()

    def stop(self):
        if not self._thread.is_alive():
            return
        _log.info("Stopping check service")
        with self._lock:
            for checker in self._checkers.values():
                self._loop.call_soon_threadsafe(checker.stop)
            self._checkers.clear()
            self._loop.call_soon_threadsafe(self._loop.stop)
            self._thread.join()
            self._loop.close()

    def start_checking(self, path, complete, interval=10.0):
        _log.info("Start checking %r", path)
        with self._lock:
            if path in self._checkers:
                raise RuntimeError("Already checking path %r" % path)
            checker = DirectioChecker(self._loop, path, complete,
                                      interval=interval)
            self._checkers[path] = checker
        self._loop.call_soon_threadsafe(checker.start)

    def stop_checking(self, path, timeout=None):
        _log.info("Stop checking %r", path)
        with self._lock:
            checker = self._checkers.pop(path)
        self._loop.call_soon_threadsafe(checker.stop)
        if timeout:
            return checker.wait(timeout)

    def is_checking(self, path):
        return path in self._checkers


IDLE = "idle"
RUNNING = "running"
STOPPING = "stopping"


class DirectioChecker(object):

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

    def __init__(self, loop, path, complete, interval=10.0):
        self._loop = loop
        self._path = path
        self._complete = complete
        self._interval = interval
        self._looper = asyncutils.LoopingCall(loop, self._check)
        self._check_time = None
        self._proc = None
        self._reader = None
        self._reaper = None
        self._err = None
        self._state = IDLE
        self._stopped = threading.Event()
        self._completed = False

    def start(self):
        if self._state is not IDLE:
            raise RuntimeError("Checker is %s", self._state)
        self._state = RUNNING
        _log.debug("Checker %r started", self._path)
        self._stopped.clear()
        self._looper.start(self._interval)

    def stop(self):
        if self._state is not RUNNING:
            return
        _log.debug("Checker %r stopping", self._path)
        self._state = STOPPING
        self._looper.stop()
        if self._proc is None:
            self._stop_completed()

    def wait(self, timeout=None):
        return self._stopped.wait(timeout)

    def is_running(self):
        return self._state is not IDLE

    def _stop_completed(self):
        self._state = IDLE
        _log.debug("Checker %r stopped", self._path)
        self._stopped.set()

    def _check(self):
        assert self._state is RUNNING
        if self._proc:
            if self._completed:
                _log.warning("Checker %r is blocked for %.2f seconds",
                             self._path, self._loop.time() - self._check_time)
            else:
                self._read_timeout()
            return
        self._completed = False
        self._check_time = self._loop.time()
        _log.debug("START check %r (delay=%.2f)",
                   self._path, self._check_time - self._looper.deadline)
        try:
            self._start_process()
        except Exception as e:
            self._err = "Error starting process: %s" % e
            self._check_completed(EXEC_ERROR)

    def _start_process(self):
        cmd = [constants.EXT_DD, "if=%s" % self._path, "of=/dev/null",
               "bs=4096", "count=1", "iflag=direct"]
        cmd = cmdutils.wrap_command(cmd)
        self._proc = subprocess.Popen(
            cmd, stdin=None, stdout=None,
            stderr=subprocess.PIPE)
        self._reader = self._loop.create_dispatcher(
            asyncevent.BufferedReader, self._proc.stderr, self._read_completed)

    def _read_timeout(self):
        assert self._state is not IDLE
        self._completed = True
        elapsed = self._loop.time() - self._check_time
        _log.debug("FINISH check %r timeout (elapsed=%.02f)",
                   self._path, elapsed)
        result = CheckResult(self._path, 1, "Read timeout", self._check_time,
                             elapsed)
        try:
            self._complete(result)
        except Exception:
            _log.exception("Unhandled error in complete callback")

    def _read_completed(self, data):
        assert self._state is not IDLE
        self._reader = None
        self._err = data
        rc = self._proc.poll()
        if rc is None:
            self._reaper = asyncevent.Reaper(self._loop, self._proc,
                                             self._check_completed)
            return
        self._check_completed(rc)

    def _check_completed(self, rc):
        assert self._state is not IDLE
        self._reaper = None
        self._proc = None
        if self._state is STOPPING:
            self._stop_completed()
            return
        if self._completed:
            return
        self._completed = True
        elapsed = self._loop.time() - self._check_time
        _log.debug("FINISH check %r (rc=%s, elapsed=%.02f)",
                   self._path, rc, elapsed)
        result = CheckResult(self._path, rc, self._err, self._check_time,
                             elapsed)
        try:
            self._complete(result)
        except Exception:
            _log.exception("Unhandled error in complete callback")

    def __repr__(self):
        info = [self.__class__.__name__,
                self._path,
                self._state]
        if self._state is RUNNING:
            info.append("next_check=%.2f" % self._looper.deadline)
        return "<%s at 0x%x>" % (" ".join(info), id(self))


class CheckResult(object):

    _PATTERN = re.compile(br".*, ([\de\-.]+) s,[^,]+")

    def __init__(self, path, rc, err, time, elapsed):
        self.path = path
        self.rc = rc
        self.err = err
        self.time = time
        self.elapsed = elapsed

    def delay(self):
        if self.rc != 0:
            raise exception.MiscFileReadException(self.path, self.rc, self.err)
        if not self.err:
            raise exception.MiscFileReadException(self.path, "no stats")
        stats = self.err.splitlines()[-1]
        match = self._PATTERN.match(stats)
        if not match:
            raise exception.MiscFileReadException(self.path,
                                                  "no match: %r" % stats)
        seconds = match.group(1)
        try:
            return float(seconds)
        except ValueError as e:
            raise exception.MiscFileReadException(self.path, e)

    def __repr__(self):
        return "<%s path=%s rc=%d err=%r time=%.2f elapsed=%.2f at 0x%x>" % (
            self.__class__.__name__, self.path, self.rc, self.err, self.time,
            self.elapsed, id(self))
