#!/usr/bin/env python3
'''
t2_block_cache.py - this file is part of S3QL.

Copyright © 2008 Nikolaus Rath <Nikolaus@rath.org>

This work can be distributed under the terms of the GNU GPLv3.
'''

if __name__ == '__main__':
    import pytest
    import sys
    sys.exit(pytest.main([__file__] + sys.argv[1:]))

from contextlib import contextmanager
from s3ql.backends import local
from s3ql.backends.common import AbstractBackend
from s3ql.backends.pool import BackendPool
from s3ql.block_cache import BlockCache, QuitSentinel
from s3ql.mkfs import init_tables
from s3ql.metadata import create_tables
from s3ql.database import Connection
from s3ql.common import AsyncFn, time_ns
from common import safe_sleep
from pytest_checklogs import assert_logs
import llfuse
import errno
import os
import logging
import shutil
import stat
import tempfile
import threading
import unittest
import queue
import pytest

# A dummy removal queue to monkeypatch around the need for removal and upload
# threads
class DummyQueue:
    def __init__(self, cache):
        self.obj = None
        self.cache = cache

    def get_nowait(self):
        return self.get(block=False)

    def put(self, obj, timeout=None):
        self.obj = obj
        self.cache._removal_loop()
        return True

    def get(self, block=True):
        if self.obj is None:
            raise queue.Empty()
        elif self.obj is QuitSentinel:
            self.obj = None
            return QuitSentinel
        else:
            tmp = self.obj
            self.obj = QuitSentinel
            return tmp

    def qsize(self):
        return 0

class cache_tests(unittest.TestCase):

    def setUp(self):

        self.backend_dir = tempfile.mkdtemp(prefix='s3ql-backend-')
        self.backend_pool = BackendPool(lambda: local.Backend('local://' + self.backend_dir,
                                                           None, None))

        self.cachedir = tempfile.mkdtemp(prefix='s3ql-cache-')
        self.max_obj_size = 1024

        # Destructors are not guaranteed to run, and we can't unlink
        # the file immediately because apsw refers to it by name.
        # Therefore, we unlink the file manually in tearDown()
        self.dbfile = tempfile.NamedTemporaryFile(delete=False)
        self.db = Connection(self.dbfile.name)
        create_tables(self.db)
        init_tables(self.db)

        # Create an inode we can work with
        self.inode = 42
        now_ns = time_ns()
        self.db.execute("INSERT INTO inodes (id,mode,uid,gid,mtime_ns,atime_ns,ctime_ns,refcount,size) "
                        "VALUES (?,?,?,?,?,?,?,?,?)",
                        (self.inode, stat.S_IFREG | stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR
                         | stat.S_IRGRP | stat.S_IXGRP | stat.S_IROTH | stat.S_IXOTH,
                         os.getuid(), os.getgid(), now_ns, now_ns, now_ns, 1, 32))

        cache = BlockCache(self.backend_pool, self.db, self.cachedir + "/cache",
                           self.max_obj_size * 100)
        self.cache = cache

        # Monkeypatch around the need for removal and upload threads
        cache.to_remove = DummyQueue(cache)

        class DummyDistributor:
            def put(self, arg, timeout=None):
                cache._do_upload(*arg)
                return True
        cache.to_upload = DummyDistributor()

        # Tested methods assume that they are called from
        # file system request handler
        llfuse.lock.acquire()

    def tearDown(self):
        llfuse.lock.release()
        self.cache.backend_pool = self.backend_pool
        self.cache.destroy()
        shutil.rmtree(self.cachedir)
        shutil.rmtree(self.backend_dir)
        self.dbfile.close()
        os.unlink(self.dbfile.name)

    def test_thread_hang(self):
        # Make sure that we don't deadlock if uploads threads or removal
        # threads have died and we try to expire or terminate

        # Monkeypatch to avoid error messages about uncaught exceptions
        # in other threads
        upload_exc = False
        removal_exc = False
        def _upload_loop(*a, fn=self.cache._upload_loop):
            try:
                return fn(*a)
            except NotADirectoryError:
                nonlocal upload_exc
                upload_exc = True
        def _removal_loop(*a, fn=self.cache._removal_loop):
            try:
                return fn(*a)
            except NotADirectoryError:
                nonlocal removal_exc
                removal_exc = True
        self.cache._upload_loop = _upload_loop
        self.cache._removal_loop = _removal_loop

        # Start threads
        self.cache.init(threads=3)

        # Create first object (we'll try to remove that)
        with self.cache.get(self.inode, 0) as fh:
            fh.write(b'bar wurfz!')
        self.cache.commit()
        self.cache.wait()

        # Make sure that upload and removal will fail
        os.rename(self.backend_dir, self.backend_dir + '-tmp')
        open(self.backend_dir, 'w').close()

        # Create second object (we'll try to upload that)
        with self.cache.get(self.inode, 1) as fh:
            fh.write(b'bar wurfz number two!')

        # Schedule a removal
        self.cache.remove(self.inode, 0)

        try:
            # Try to clean-up (implicitly calls expire)
            with llfuse.lock_released, \
                assert_logs('Unable to flush cache, no upload threads left alive',
                              level=logging.ERROR, count=1):
                with pytest.raises(OSError) as exc_info:
                     self.cache.destroy()
                assert exc_info.value.errno == errno.ENOTEMPTY
            assert upload_exc
            assert removal_exc
        finally:
            # Fix backend dir
            os.unlink(self.backend_dir)
            os.rename(self.backend_dir + '-tmp', self.backend_dir)

            # Remove objects from cache and make final destroy
            # call into no-op.
            self.cache.remove(self.inode, 1)
            self.cache.destroy = lambda: None


    @staticmethod
    def random_data(len_):
        with open("/dev/urandom", "rb") as fh:
            return fh.read(len_)

    def test_get(self):
        inode = self.inode
        blockno = 11
        data = self.random_data(int(0.5 * self.max_obj_size))

        # Case 1: Object does not exist yet
        with self.cache.get(inode, blockno) as fh:
            fh.seek(0)
            fh.write(data)

        # Case 2: Object is in cache
        with self.cache.get(inode, blockno) as fh:
            fh.seek(0)
            self.assertEqual(data, fh.read(len(data)))

        # Case 3: Object needs to be downloaded
        self.cache.clear()
        with self.cache.get(inode, blockno) as fh:
            fh.seek(0)
            self.assertEqual(data, fh.read(len(data)))

    def test_expire(self):
        inode = self.inode

        # Define the 4 most recently accessed ones
        most_recent = [7, 11, 10, 8]
        for i in most_recent:
            safe_sleep(0.2)
            with self.cache.get(inode, i) as fh:
                fh.write(('%d' % i).encode())

        # And some others
        for i in range(20):
            if i in most_recent:
                continue
            with self.cache.get(inode, i) as fh:
                fh.write(('%d' % i).encode())

        # Flush the 2 most recently accessed ones
        commit(self.cache, inode, most_recent[-2])
        commit(self.cache, inode, most_recent[-3])

        # We want to expire 4 entries, 2 of which are already flushed
        self.cache.cache.max_entries = 16
        self.cache.backend_pool = TestBackendPool(self.backend_pool, no_write=2)
        self.cache.expire()
        self.cache.backend_pool.verify()
        self.assertEqual(len(self.cache.cache), 16)

        for i in range(20):
            if i in most_recent:
                self.assertTrue((inode, i) not in self.cache.cache)
            else:
                self.assertTrue((inode, i) in self.cache.cache)

    def test_upload(self):
        inode = self.inode
        datalen = int(0.1 * self.cache.cache.max_size)
        blockno1 = 21
        blockno2 = 25
        blockno3 = 7

        data1 = self.random_data(datalen)
        data2 = self.random_data(datalen)
        data3 = self.random_data(datalen)

        # Case 1: create new object
        self.cache.backend_pool = TestBackendPool(self.backend_pool, no_write=1)
        with self.cache.get(inode, blockno1) as fh:
            fh.seek(0)
            fh.write(data1)
            el1 = fh
        self.cache.upload(el1)
        self.cache.backend_pool.verify()

        # Case 2: Link new object
        self.cache.backend_pool = TestBackendPool(self.backend_pool)
        with self.cache.get(inode, blockno2) as fh:
            fh.seek(0)
            fh.write(data1)
            el2 = fh
        self.cache.upload(el2)
        self.cache.backend_pool.verify()

        # Case 3: Upload old object, still has references
        self.cache.backend_pool = TestBackendPool(self.backend_pool, no_write=1)
        with self.cache.get(inode, blockno1) as fh:
            fh.seek(0)
            fh.write(data2)
        self.cache.upload(el1)
        self.cache.backend_pool.verify()

        # Case 4: Upload old object, no references left
        self.cache.backend_pool = TestBackendPool(self.backend_pool, no_del=1, no_write=1)
        with self.cache.get(inode, blockno2) as fh:
            fh.seek(0)
            fh.write(data3)
        self.cache.upload(el2)
        self.cache.backend_pool.verify()

        # Case 5: Link old object, no references left
        self.cache.backend_pool = TestBackendPool(self.backend_pool, no_del=1)
        with self.cache.get(inode, blockno2) as fh:
            fh.seek(0)
            fh.write(data2)
        self.cache.upload(el2)
        self.cache.backend_pool.verify()

        # Case 6: Link old object, still has references
        # (Need to create another object first)
        self.cache.backend_pool = TestBackendPool(self.backend_pool, no_write=1)
        with self.cache.get(inode, blockno3) as fh:
            fh.seek(0)
            fh.write(data1)
            el3 = fh
        self.cache.upload(el3)
        self.cache.backend_pool.verify()

        self.cache.backend_pool = TestBackendPool(self.backend_pool)
        with self.cache.get(inode, blockno1) as fh:
            fh.seek(0)
            fh.write(data1)
        self.cache.upload(el1)
        self.cache.clear()
        self.cache.backend_pool.verify()

    def test_remove_referenced(self):
        inode = self.inode
        datalen = int(0.1 * self.cache.cache.max_size)
        blockno1 = 21
        blockno2 = 24
        data = self.random_data(datalen)

        self.cache.backend_pool = TestBackendPool(self.backend_pool, no_write=1)
        with self.cache.get(inode, blockno1) as fh:
            fh.seek(0)
            fh.write(data)
        with self.cache.get(inode, blockno2) as fh:
            fh.seek(0)
            fh.write(data)
        self.cache.clear()
        self.cache.backend_pool.verify()

        self.cache.backend_pool = TestBackendPool(self.backend_pool)
        self.cache.remove(inode, blockno1)
        self.cache.backend_pool.verify()

    def test_remove_cache(self):
        inode = self.inode
        data1 = self.random_data(int(0.4 * self.max_obj_size))

        # Case 1: Elements only in cache
        with self.cache.get(inode, 1) as fh:
            fh.seek(0)
            fh.write(data1)
        self.cache.remove(inode, 1)
        with self.cache.get(inode, 1) as fh:
            fh.seek(0)
            self.assertEqual(fh.read(42), b'')

    def test_upload_race(self):
        inode = self.inode
        blockno = 1
        data1 = self.random_data(int(0.4 * self.max_obj_size))
        with self.cache.get(inode, blockno) as fh:
            fh.seek(0)
            fh.write(data1)

        # Remove it
        self.cache.remove(inode, blockno)

        # Try to upload it, may happen if CommitThread is interrupted
        self.cache.upload(fh)

    def test_expire_race(self):
        # Create element
        inode = self.inode
        blockno = 1
        data1 = self.random_data(int(0.4 * self.max_obj_size))
        with self.cache.get(inode, blockno) as fh:
            fh.seek(0)
            fh.write(data1)
        self.cache.upload(fh)

        # Make sure entry will be expired
        self.cache.cache.max_entries = 0
        def e_w_l():
            with llfuse.lock:
                self.cache.expire()

        # Lock it
        self.cache._lock_entry(inode, blockno, release_global=True)

        try:
            # Start expiration, will block on lock
            t1 = AsyncFn(e_w_l)
            t1.start()

            # Start second expiration, will block
            t2 = AsyncFn(e_w_l)
            t2.start()

            # Release lock
            with llfuse.lock_released:
                safe_sleep(0.1)
                self.cache._unlock_entry(inode, blockno)
                t1.join_and_raise()
                t2.join_and_raise()

            assert len(self.cache.cache) == 0
        finally:
                self.cache._unlock_entry(inode, blockno, release_global=True,
                                         noerror=True)


    def test_parallel_expire(self):
        # Create elements
        inode = self.inode
        for i in range(5):
            data1 = self.random_data(int(0.4 * self.max_obj_size))
            with self.cache.get(inode, i) as fh:
                fh.write(data1)

        # We want to expire just one element, but have
        # several threads running expire() simultaneously
        self.cache.cache.max_entries = 4
        def e_w_l():
            with llfuse.lock:
                self.cache.expire()

        # Lock first element so that we have time to start threads
        self.cache._lock_entry(inode, 0, release_global=True)

        try:
            # Start expiration, will block on lock
            t1 = AsyncFn(e_w_l)
            t1.start()

            # Start second expiration, will block
            t2 = AsyncFn(e_w_l)
            t2.start()

            # Release lock
            with llfuse.lock_released:
                safe_sleep(0.1)
                self.cache._unlock_entry(inode, 0)
                t1.join_and_raise()
                t2.join_and_raise()

            assert len(self.cache.cache) == 4
        finally:
                self.cache._unlock_entry(inode, 0, release_global=True,
                                         noerror=True)


    def test_remove_cache_db(self):
        inode = self.inode
        data1 = self.random_data(int(0.4 * self.max_obj_size))

        # Case 2: Element in cache and db
        with self.cache.get(inode, 1) as fh:
            fh.seek(0)
            fh.write(data1)
        self.cache.backend_pool = TestBackendPool(self.backend_pool, no_write=1)
        commit(self.cache, inode)
        self.cache.backend_pool.verify()
        self.cache.backend_pool = TestBackendPool(self.backend_pool, no_del=1)
        self.cache.remove(inode, 1)
        self.cache.backend_pool.verify()

        with self.cache.get(inode, 1) as fh:
            fh.seek(0)
            self.assertEqual(fh.read(42), b'')


    def test_remove_db(self):
        inode = self.inode
        data1 = self.random_data(int(0.4 * self.max_obj_size))

        # Case 3: Element only in DB
        with self.cache.get(inode, 1) as fh:
            fh.seek(0)
            fh.write(data1)
        self.cache.backend_pool = TestBackendPool(self.backend_pool, no_write=1)
        self.cache.clear()
        self.cache.backend_pool.verify()
        self.cache.backend_pool = TestBackendPool(self.backend_pool, no_del=1)
        self.cache.remove(inode, 1)
        self.cache.backend_pool.verify()
        with self.cache.get(inode, 1) as fh:
            fh.seek(0)
            self.assertEqual(fh.read(42), b'')


class TestBackendPool(AbstractBackend):
    has_native_rename = False

    def __init__(self, backend_pool, no_read=0, no_write=0, no_del=0):
        super().__init__()
        self.no_read = no_read
        self.no_write = no_write
        self.no_del = no_del
        self.backend_pool = backend_pool
        self.backend = backend_pool.pop_conn()
        self.lock = threading.Lock()

    def __del__(self):
        self.backend_pool.push_conn(self.backend)

    def verify(self):
        if self.no_read != 0:
            raise RuntimeError('Got too few open_read calls')
        if self.no_write != 0:
            raise RuntimeError('Got too few open_write calls')
        if self.no_del != 0:
            raise RuntimeError('Got too few delete calls')

    @contextmanager
    def __call__(self):
        '''Provide connection from pool (context manager)'''

        with self.lock:
            yield self

    def lookup(self, key):
        return self.backend.lookup(key)

    def open_read(self, key):
        self.no_read -= 1
        if self.no_read < 0:
            raise RuntimeError('Got too many open_read calls')

        return self.backend.open_read(key)

    def open_write(self, key, metadata=None, is_compressed=False):
        self.no_write -= 1
        if self.no_write < 0:
            raise RuntimeError('Got too many open_write calls')

        return self.backend.open_write(key, metadata, is_compressed)

    def is_temp_failure(self, exc):
        return self.backend.is_temp_failure(exc)

    def clear(self):
        return self.backend.clear()

    def contains(self, key):
        return self.backend.contains(key)

    def delete(self, key, force=False):
        self.no_del -= 1
        if self.no_del < 0:
            raise RuntimeError('Got too many delete calls')

        return self.backend.delete(key, force)

    def list(self, prefix=''):
        '''List keys in backend

        Returns an iterator over all keys in the backend.
        '''
        return self.backend.list(prefix)

    def copy(self, src, dest, metadata=None):
        return self.backend.copy(src, dest, metadata)

    def rename(self, src, dest, metadata=None):
        return self.backend.rename(src, dest, metadata)

    def update_meta(self, key, metadata):
        return self.backend.update_meta(key, metadata)

    def get_size(self, key):
        '''Return size of object stored under *key*'''

        return self.backend.get_size(key)

def commit(cache, inode, block=None):
    """Upload data for `inode`

    This is only for testing purposes, since the method blocks until all current
    uploads have been completed.
    """

    for el in cache.cache.values():
        if el.inode != inode:
            continue
        if not el.dirty:
            continue

        if block is not None and el.blockno != block:
            continue

        cache.upload(el)