
from __future__ import absolute_import

import io
import logging
import mmap
import os
import struct
import time

from collections import namedtuple
from contextlib import contextmanager

import six

from vdsm import utils
from vdsm.common import cmdutils
from vdsm.common import commands
from vdsm.common import constants
from vdsm.common import errors
from vdsm.common.osutils import uninterruptible
from vdsm.storage import constants as sc
from vdsm.storage import exception as se
from vdsm.storage import fsutils
from vdsm.storage.compat import sanlock

RESERVED_SLOTS = 3

METADATA_SIZE = 512

RECORD_BASE = METADATA_SIZE

RECORD_SIZE = 64

INDEX_SIZE = 62 * 4096

MAX_RECORDS = (INDEX_SIZE - METADATA_SIZE) // RECORD_SIZE

INDEX_VERSION = 1

META_STRUCT = struct.Struct(">i x 4s x 48s x 10s x c 440x c")

INDEX_MAGIC = 0x12152016

RECORD_STRUCT = struct.Struct("48s x 11s x 3c")

LOOKUP_STRUCT = struct.Struct("48s x")

RECORD_TERM = b"\n"

FLAG_NONE = b"-"
FLAG_UPDATING = b"u"


SANLK_LEADER_MAGIC = -223


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



class Error(errors.Base):

    def __init__(self, lease_id):
        self.lease_id = lease_id


class LeaseExists(Error):
    msg = "Lease {self.lease_id} exists"


class LeaseUpdating(Error):
    msg = "Lease {self.lease_id} is updating"


class NoSpace(Error):
    msg = "No space to add lease {self.lease_id}"


class InvalidRecord(Error):
    msg = "Invalid record ({self.reason}): {self.record}"

    def __init__(self, reason, record):
        self.reason = reason
        self.record = record


class NoSuchResource(Error):
    msg = "No such resource {self.path, self.offset}"

    def __init__(self, path, offset):
        self.path = path
        self.offset = offset


class InvalidIndex(Error):
    """
    Base class for errors aboout unusable index.
    """


class IndexIsUpdating(InvalidIndex):
    msg = ("Index is updating or an update operation was aborted, the index "
           "must be formatted or rebuilt from storage: {self.metadata}")

    def __init__(self, metadata):
        self.metadata = metadata


class InvalidMetadata(InvalidIndex):
    msg = ("Invalid index metadata ({self.reason}), the index must be "
           "formatted or rebuilt from storage: {self.data!r}")

    def __init__(self, reason, data):
        self.reason = reason
        self.data = data


class TruncatedIndex(InvalidIndex):
    msg = ("Couuld not read index, expected {self.expected} bytes, available "
           "{self.available} bytes")

    def __init__(self, expected, available):
        self.expected = expected
        self.available = available


LeaseInfo = namedtuple("LeaseInfo", (
))


ResourceInfo = namedtuple("ResourceInfo", (
))


class IndexMetadata(object):

    @classmethod
    def fromebytes(cls, data):
        try:
            magic, version, lockspace, mtime, updating, _ = \
                META_STRUCT.unpack(data)
        except struct.error as e:
            raise InvalidMetadata("cannot unpack: %s" % e, data)

        if magic != INDEX_MAGIC:
            raise InvalidMetadata("invalid magic: %s" % magic, data)

        try:
            version = int(version)
        except ValueError:
            raise InvalidMetadata("invalid version: %s" % version, data)

        if version != INDEX_VERSION:
            raise InvalidMetadata("unsupported version %s" % version, data)

        lockspace = lockspace.rstrip(b"\0")
        try:
            lockspace = lockspace.decode("ascii")
        except UnicodeDecodeError:
            raise InvalidMetadata("cannot decode lockspace %r" %
                                  lockspace, data)

        try:
            mtime = int(mtime)
        except ValueError:
            raise InvalidMetadata("cannot parse mtime %r" %
                                  mtime, data)

        updating = (updating == FLAG_UPDATING)

        return cls(version, lockspace, mtime=mtime, updating=updating)

    def __init__(self, version, lockspace, mtime=None, updating=False):
        if mtime is None:
            mtime = int(time.time())
        self._version = version
        self._lockspace = lockspace
        self._mtime = mtime
        self._updating = updating

    @property
    def version(self):
        return self._version

    @property
    def lockspace(self):
        return self._lockspace

    @property
    def mtime(self):
        return self._mtime

    @property
    def updating(self):
        return self._updating

    def bytes(self):
        return META_STRUCT.pack(
            INDEX_MAGIC,
            b"%04d" % self._version,
            self._lockspace.encode("ascii"),
            b"%010d" % self._mtime,
            FLAG_UPDATING if self._updating else FLAG_NONE,
            RECORD_TERM,
        )

    def __repr__(self):
        return ("<IndexMetadata version={self.version}, "
                "lockspace={self.lockspace!r}, "
                "mtime={self.mtime}, "
                "updating={self.updating} "


class Record(object):

    @classmethod
    def frombytes(cls, record):
        try:
            resource, offset, updating, _, _ = RECORD_STRUCT.unpack(record)
        except struct.error as e:
            raise InvalidRecord("cannot unpack: %s" % e, record)

        resource = resource.rstrip(b"\0")
        try:
            resource = resource.decode("ascii")
        except UnicodeDecodeError:
            raise InvalidRecord("cannot decode resource %r" % resource, record)

        updating = (updating == FLAG_UPDATING)

        try:
            offset = int(offset)
        except ValueError:
            raise InvalidRecord("cannot parse offset %r" % offset, record)

        return cls(resource, offset, updating=updating)

    def __init__(self, resource, offset, updating=False):
        self._resource = resource
        self._offset = offset
        self._updating = updating

    def bytes(self):
        return RECORD_STRUCT.pack(
            self._resource.encode("ascii"),
            b"%011d" % self._offset,
            FLAG_UPDATING if self.updating else FLAG_NONE,
            FLAG_NONE,
            RECORD_TERM,
        )

    @property
    def resource(self):
        return self._resource

    @property
    def offset(self):
        return self._offset

    @property
    def updating(self):
        return self._updating


EMPTY_RECORD = Record("", 0)


class LeasesVolume(object):

    def __init__(
            self, file, alignment=sc.ALIGNMENT_1M,
            block_size=sc.BLOCK_SIZE_512):
        log.debug("Loading index from %r", file.name)
        self._file = file
        self._alignment = alignment
        self._block_size = block_size
        self._index = VolumeIndex(alignment, block_size)
        try:
            self._index.load(file)
            self._md = self._index.read_metadata()
            if self._md.updating:
                raise IndexIsUpdating(self._md)
        except:
            self._index.close()
            raise
        log.debug("Loaded %s", self._md)

    @property
    def path(self):
        return self._file.name

    @property
    def lockspace(self):
        return self._md.lockspace

    @property
    def version(self):
        return self._md.version

    @property
    def mtime(self):
        return self._md.mtime

    def lookup(self, lease_id):
        log.debug("Looking up lease %r in lockspace %r",
                  lease_id, self.lockspace)
        recnum = self._index.find_record(lease_id)
        if recnum == -1:
            raise se.NoSuchLease(lease_id)

        record = self._index.read_record(recnum)
        if record.updating:
            raise LeaseUpdating(lease_id)

        offset = lease_offset(recnum, self._alignment)
        return LeaseInfo(self.lockspace, lease_id, self._file.name, offset)

    def add(self, lease_id):
        log.info("Adding lease %r in lockspace %r",
                 lease_id, self.lockspace)
        recnum = self._index.find_record(lease_id)
        if recnum != -1:
            record = self._index.read_record(recnum)
            if record.updating:
                raise LeaseUpdating(lease_id)
            else:
                raise LeaseExists(lease_id)

        recnum = self._index.find_free_record()
        if recnum == -1:
            raise NoSpace(lease_id)

        offset = lease_offset(recnum, self._alignment)
        record = Record(lease_id, offset, updating=True)
        self._write_record(recnum, record)

        sanlock.write_resource(
            self.lockspace,
            lease_id,
            [(self._file.name, offset)],
            align=self._alignment,
            sector=self._block_size)

        record = Record(lease_id, offset)
        self._write_record(recnum, record)

        return LeaseInfo(self.lockspace, lease_id, self._file.name, offset)

    def remove(self, lease_id):
        log.info("Removing lease %r in lockspace %r",
                 lease_id, self.lockspace)
        recnum = self._index.find_record(lease_id)
        if recnum == -1:
            raise se.NoSuchLease(lease_id)

        offset = lease_offset(recnum, self._alignment)
        record = Record(lease_id, offset, updating=True)
        self._write_record(recnum, record)

        sanlock.write_resource(
            "",
            "",
            [(self._file.name, offset)],
            align=self._alignment,
            sector=self._block_size)

        self._write_record(recnum, EMPTY_RECORD)

    def leases(self):
        log.debug("Getting all leases for lockspace %r", self.lockspace)
        leases = {}
        for recnum in range(MAX_RECORDS):
            record = self._index.read_record(recnum)
            if record.resource:
                leases[record.resource] = {
                    "offset": lease_offset(recnum, self._alignment),
                    "updating": record.updating,
                }
        return leases

    def close(self):
        log.debug("Closing index for lockspace %r", self.lockspace)
        self._index.close()

    def _write_record(self, recnum, record):
        block = self._index.copy_record_block(recnum)
        with utils.closing(block):
            block.write_record(recnum, record)
            block.dump(self._file)
        self._index.write_record(recnum, record)


def format_index(lockspace, file, alignment=sc.ALIGNMENT_1M,
                 block_size=sc.BLOCK_SIZE_512):
    log.info("Formatting index for lockspace %r (version=%d)",
             lockspace, INDEX_VERSION)
    index = VolumeIndex(alignment, block_size)
    with utils.closing(index):
        with index.updating(lockspace, file):
            for recnum in range(MAX_RECORDS):
                index.write_record(recnum, EMPTY_RECORD)
            index.dump(file)


def rebuild_index(
        lockspace, file, alignment=sc.ALIGNMENT_1M,
        block_size=sc.BLOCK_SIZE_512):
    log.info("Rebuilding index for lockspace %r (version=%d)",
             lockspace, INDEX_VERSION)
    index = VolumeIndex(alignment, block_size)
    with utils.closing(index):
        with index.updating(lockspace, file):
            max_offset = file.size() - alignment
            for recnum in range(MAX_RECORDS):
                offset = lease_offset(recnum, alignment)
                if offset > max_offset:
                    index.write_record(recnum, EMPTY_RECORD)
                    continue
                try:
                    res = read_resource(
                        file.name,
                        offset,
                        alignment=alignment,
                        block_size=block_size)
                except NoSuchResource:
                    record = EMPTY_RECORD
                else:
                    log.debug("Restoring lease %s", res)
                    record = Record(res.resource, offset)
                index.write_record(recnum, record)

            log.debug("Writing new index")
            index.dump(file)


def read_resource(
        path, offset, alignment=sc.ALIGNMENT_1M, block_size=sc.BLOCK_SIZE_512):
    try:
        res = sanlock.read_resource(
            path, offset, align=alignment, sector=block_size)
    except sanlock.SanlockException as e:
        if e.errno != SANLK_LEADER_MAGIC:
            raise
        raise NoSuchResource(path, offset)
    if res["resource"] == "":
        raise NoSuchResource(path, offset)
    return ResourceInfo(res["lockspace"], res["resource"], res["version"])


def lease_offset(recnum, alignment):
    return (RESERVED_SLOTS + recnum) * alignment


class VolumeIndex(object):

    def __init__(self, offset, block_size):
        self._offset = offset
        self._block_size = block_size
        self._buf = mmap.mmap(-1, INDEX_SIZE, mmap.MAP_SHARED)

    def find_record(self, lease_id):
        prefix = LOOKUP_STRUCT.pack(lease_id.encode("ascii"))

        offset = self._buf.find(prefix, RECORD_BASE)
        if offset == -1:
            return -1

        return self._record_number(offset)

    def find_free_record(self):
        offset = self._buf.find(EMPTY_RECORD.bytes(), RECORD_BASE)
        if offset == -1:
            return -1

        return self._record_number(offset)

    def read_record(self, recnum):
        offset = self._record_offset(recnum)
        self._buf.seek(offset)
        data = self._buf.read(RECORD_SIZE)
        return Record.frombytes(data)

    def write_record(self, recnum, record):
        offset = self._record_offset(recnum)
        self._buf.seek(offset)
        self._buf.write(record.bytes())

    def read_metadata(self):
        self._buf.seek(0)
        data = self._buf.read(METADATA_SIZE)
        return IndexMetadata.fromebytes(data)

    def write_metadata(self, metadata):
        self._buf.seek(0)
        self._buf.write(metadata.bytes())

    def load(self, file):
        nread = file.pread(self._offset, self._buf)
        if nread < len(self._buf):
            raise TruncatedIndex(len(self._buf), nread)

    def dump(self, file):
        file.pwrite(self._offset, self._buf)

    def copy_record_block(self, recnum):
        offset = self._record_offset(recnum)
        block_start = offset - (offset % self._block_size)
        return ChangeBlock(
            self._offset, self._buf, block_start, self._block_size)

    @contextmanager
    def updating(self, lockspace, file):
        metadata = IndexMetadata(INDEX_VERSION, lockspace, updating=True)
        self.write_metadata(metadata)

        yield

        metadata = IndexMetadata(INDEX_VERSION, lockspace)
        self.write_metadata(metadata)

        block = ChangeBlock(self._offset, self._buf, 0, self._block_size)
        with utils.closing(block):
            block.dump(file)

    def close(self):
        self._buf.close()

    def _record_offset(self, recnum):
        return RECORD_BASE + recnum * RECORD_SIZE

    def _record_number(self, offset):
        return (offset - RECORD_BASE) // RECORD_SIZE


class ChangeBlock(object):

    def __init__(self, index_offset, index_buf, offset, size):
        self._index_offset = index_offset
        self._offset = offset
        self._size = size
        self._buf = mmap.mmap(-1, size, mmap.MAP_SHARED)
        self._buf[:] = index_buf[offset:offset + size]

    def write_record(self, recnum, record):
        offset = self._record_offset(recnum)
        self._buf.seek(offset)
        self._buf.write(record.bytes())

    def dump(self, file):
        file.pwrite(self._index_offset + self._offset, self._buf)

    def close(self):
        self._buf.close()

    def _record_offset(self, recnum):
        offset = RECORD_BASE + recnum * RECORD_SIZE - self._offset
        last_offset = self._size - RECORD_SIZE
        if not 0 <= offset <= last_offset:
            raise ValueError("recnum %s out of range for this block" % recnum)
        return offset


class DirectFile(object):

    def __init__(self, path):
        self._path = path
        fd = os.open(path, os.O_RDWR | os.O_DIRECT)
        self._file = io.FileIO(fd, "r+", closefd=True)

    @property
    def name(self):
        return self._path

    def pread(self, offset, buf):
        self._file.seek(offset, os.SEEK_SET)
        pos = 0
        if six.PY2:
            rbuf = mmap.mmap(-1, len(buf), mmap.MAP_SHARED)
            with utils.closing(rbuf, log=log.name):
                while pos < len(buf):
                    nread = uninterruptible(self._file.readinto, rbuf)
                    if nread == 0:
                    buf.write(rbuf[:nread])
                    pos += nread
        else:
            while pos < len(buf):
                rbuf = memoryview(buf)[pos:]
                nread = uninterruptible(self._file.readinto, rbuf)
                if nread == 0:
                pos += nread
        return pos

    def pwrite(self, offset, buf):
        self._file.seek(offset, os.SEEK_SET)
        pos = 0
        while pos < len(buf):
            if six.PY2:
                wbuf = buffer(buf, pos)
            else:
                wbuf = memoryview(buf)[pos:]
            pos += uninterruptible(self._file.write, wbuf)
        os.fsync(self._file.fileno())

    def size(self):
        return fsutils.size(self._path)

    def close(self):
        self._file.close()


class InterruptibleDirectFile(object):

    def __init__(self, path, oop):
        self._path = path
        self._oop = oop

    @property
    def name(self):
        return self._path

    def pread(self, offset, buf):
        args = [
            constants.EXT_DD,
            "if=%s" % self._path,
            "iflag=direct,skip_bytes",
            "skip=%d" % offset,
            "bs=%d" % len(buf),
            "count=1",
        ]
        out = self._run(args)
        buf.write(out)
        return len(out)

    def pwrite(self, offset, buf):
        args = [
            constants.EXT_DD,
            "iflag=fullblock",
            "of=%s" % self._path,
            "oflag=direct,seek_bytes",
            "seek=%d" % offset,
            "bs=%d" % len(buf),
            "count=1",
            "conv=notrunc,nocreat,fsync",
        ]
        self._run(args, data=buf[:])

    def size(self):
        return self._oop.os.stat(self._path).st_size

    def close(self):
        pass

    def _run(self, args, data=None):
        rc, out, err = commands.execCmd(
            args,
            data=data,
            raw=True,
            resetCpuAffinity=False)
        if rc != 0:
            raise cmdutils.Error(args, rc, "[suppressed]", err)
        return out
