<commit_before># Copyright (C) 2015-2016 Regents of the University of California
#
# 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.

# python 2/3 compatibility
from __future__ import absolute_import
from builtins import range

# standard library
from contextlib import contextmanager
import logging
import random
import shutil
import os
import re
import tempfile
import stat
import errno
import time
import traceback
import psutil
try:
    import cPickle as pickle
except ImportError:
    import pickle

# toil and bd2k dependencies
from toil.fileStore import FileID
from toil.lib.bioio import absSymPath
from toil.jobStores.abstractJobStore import (AbstractJobStore,
                                             NoSuchJobException,
                                             NoSuchFileException,
                                             JobStoreExistsException,
                                             NoSuchJobStoreException)
from toil.jobGraph import JobGraph

logger = logging.getLogger( __name__ )


class FileJobStore(AbstractJobStore):
    """
    A job store that uses a directory on a locally attached file system. To be compatible with
    distributed batch systems, that file system must be shared by all worker nodes.
    """

    # Valid chars for the creation of temporary directories
    validDirs = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789"
    # Depth of temporary subdirectories created
    levels = 2

    # 10Mb RAM chunks when reading/writing files
    BUFFER_SIZE = 10485760 # 10Mb

    def __init__(self, path):
        """
        :param str path: Path to directory holding the job store
        """
        super(FileJobStore, self).__init__()
        self.jobStoreDir = absSymPath(path)
        logger.debug("Path to job store directory is '%s'.", self.jobStoreDir)
        # Directory where temporary files go
        self.tempFilesDir = os.path.join(self.jobStoreDir, 'tmp')
        self.linkImports = None

    def initialize(self, config):
        try:
            os.mkdir(self.jobStoreDir)
        except OSError as e:
            if e.errno == errno.EEXIST:
                raise JobStoreExistsException(self.jobStoreDir)
            else:
                raise
        os.mkdir(self.tempFilesDir)
        self.linkImports = config.linkImports
        super(FileJobStore, self).initialize(config)

    def resume(self):
        if not os.path.isdir(self.jobStoreDir):
            raise NoSuchJobStoreException(self.jobStoreDir)
        super(FileJobStore, self).resume()

    def robust_rmtree(self, path, max_retries=3):
        """Robustly tries to delete paths.

        Retries several times (with increasing delays) if an OSError
        occurs.  If the final attempt fails, the Exception is propagated
        to the caller.

        Borrowed and slightly modified from:
        https://github.com/hashdist/hashdist
        """
        
        
        
        def handle_error(function, path, excinfo):
            logger.error('Error removing {}'.format(path))
            
            logger.error('We are PID {}'.format(os.getpid()))
            
            if function == os.remove:
                # This is a stuck file
                
                # Go hunting for who might have it open.
                for process in psutil.process_iter():
                    try:
                        # Some processes we are not allowed to look at, so we ignore errors
                        for open_file in process.open_files():
                            if open_file.path == path:
                                # This process has this file open
                                logger.error('Process {} ({}) has {} open as FD {}'.format(process.pid, process.cmdline(), path, open_file.fd))
                                
                                if process.pid != os.getpid():
                                    # Try killing it
                                    #logger.error('Kill {} to free up file'.format(process.pid))
                                    #os.kill(process.pid, 9)
                                    pass
                    except psutil.AccessDenied:
                        pass
                                
                            
            # See what we have open
            our_process = psutil.Process()
            for open_file in our_process.open_files():
                logger.error('We opened: {}'.format(open_file))
            
            raise OSError()
        
        dt = 1
        for _ in range(max_retries):
            try:
                shutil.rmtree(path, False, handle_error)
                return
            except OSError:
                logger.error('Unable to remove path: {}.  Retrying in {} seconds.'.format(path, dt))
                time.sleep(dt)
                dt *= 2

        # Final attempt, pass any Exceptions up to caller.
        shutil.rmtree(path, False, handle_error)

    def destroy(self):
        if os.path.exists(self.jobStoreDir):
            self.robust_rmtree(self.jobStoreDir)

    ##########################################
    # The following methods deal with creating/loading/updating/writing/checking for the
    # existence of jobs
    ##########################################<commit_msg>Forcibly kill processes and close file descriptors to get the job store cleaned up<commit_after># Copyright (C) 2015-2016 Regents of the University of California
#
# 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.

# python 2/3 compatibility
from __future__ import absolute_import
from builtins import range

# standard library
from contextlib import contextmanager
import logging
import random
import shutil
import os
import re
import tempfile
import stat
import errno
import time
import traceback
import psutil
try:
    import cPickle as pickle
except ImportError:
    import pickle

# toil and bd2k dependencies
from toil.fileStore import FileID
from toil.lib.bioio import absSymPath
from toil.jobStores.abstractJobStore import (AbstractJobStore,
                                             NoSuchJobException,
                                             NoSuchFileException,
                                             JobStoreExistsException,
                                             NoSuchJobStoreException)
from toil.jobGraph import JobGraph

logger = logging.getLogger( __name__ )


class FileJobStore(AbstractJobStore):
    """
    A job store that uses a directory on a locally attached file system. To be compatible with
    distributed batch systems, that file system must be shared by all worker nodes.
    """

    # Valid chars for the creation of temporary directories
    validDirs = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789"
    # Depth of temporary subdirectories created
    levels = 2

    # 10Mb RAM chunks when reading/writing files
    BUFFER_SIZE = 10485760 # 10Mb

    def __init__(self, path):
        """
        :param str path: Path to directory holding the job store
        """
        super(FileJobStore, self).__init__()
        self.jobStoreDir = absSymPath(path)
        logger.debug("Path to job store directory is '%s'.", self.jobStoreDir)
        # Directory where temporary files go
        self.tempFilesDir = os.path.join(self.jobStoreDir, 'tmp')
        self.linkImports = None

    def initialize(self, config):
        try:
            os.mkdir(self.jobStoreDir)
        except OSError as e:
            if e.errno == errno.EEXIST:
                raise JobStoreExistsException(self.jobStoreDir)
            else:
                raise
        os.mkdir(self.tempFilesDir)
        self.linkImports = config.linkImports
        super(FileJobStore, self).initialize(config)

    def resume(self):
        if not os.path.isdir(self.jobStoreDir):
            raise NoSuchJobStoreException(self.jobStoreDir)
        super(FileJobStore, self).resume()

    def robust_rmtree(self, path, max_retries=3):
        """Robustly tries to delete paths.

        Retries several times (with increasing delays) if an OSError
        occurs.  If the final attempt fails, the Exception is propagated
        to the caller.

        Borrowing patterns from:
        https://github.com/hashdist/hashdist
        """
        
        
        # First list all our open files
        us = psutil.Process()
        paths_and_fds = []
        for open_file in us.open_files():
            paths_and_fds.append((open_file.path, open_file.fd))
        logger.error('We have {} files open'.format(len(paths_and_fds)))
        for path, fd in paths_and_fds:
            logger.error('We have {} open as FD {}'.format(path, fd))
            
            with open(path, 'r') as sampler:
                # Show a sample of what is in the file
                logger.error('Sample data: "{}"\n\n'.format(sampler.read(50)))
        
        
        def handle_error(function, path, excinfo):
            logger.error('Error in {} on {}'.format(function, path))
            logger.error('Exception: {}'.format(excinfo))
            
            logger.error('We are PID {}'.format(os.getpid()))
            
            if function == os.remove:
                # This is a stuck file
                
                # Go hunting for who might have it open.
                for process in psutil.process_iter():
                    try:
                        # Some processes we are not allowed to look at, so we ignore errors
                        for open_file in process.open_files():
                            if open_file.path == path:
                                # This process has this file open
                                logger.error('Process {} ({}) has {} open as FD {}'.format(process.pid, process.cmdline(), path, open_file.fd))
                                
                                if process.pid == os.getpid():
                                    # Manually close the relevant file descriptor we have open
                                    # TODO: Work out how we could possibly have an FD open
                                    logger.error('We have it open. Close FD {}'.format(open_file.fd))
                                    os.close(open_file.fd)
                                else:
                                    # Try killing it
                                    logger.error('Kill {} to free up file'.format(process.pid))
                                    os.kill(process.pid, 9)
                    except psutil.AccessDenied:
                        pass
                                
                            
            # See what we have open
            our_process = psutil.Process()
            for open_file in our_process.open_files():
                logger.error('We opened: {}'.format(open_file))
            
            raise OSError()
        
        delay = 1
        for _ in range(max_retries):
            try:
                if not os.path.isdir(path):
                    # Remove the given normal file
                    os.remove(path)
                else:
                    # Remove the given directory
                    shutil.rmtree(path, False, handle_error)
                return
            except OSError:
                logger.error('Unable to remove path: {}.  Retrying in {} seconds.'.format(path, delay))
                time.sleep(delay)
                delay *= 2

        # Final attempt, pass any Exceptions up to caller.
        shutil.rmtree(path, False, handle_error)

    def destroy(self):
        if os.path.exists(self.jobStoreDir):
            self.robust_rmtree(self.jobStoreDir)

    ##########################################
    # The following methods deal with creating/loading/updating/writing/checking for the
    # existence of jobs
    ##########################################
{'fraction_non_alphanumeric': 0.08339822457561127, 'fraction_numerical': 0.006696776203083632, 'mean_word_length': 1.760154738878143, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 31, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1538725', 'n_tokens_mistral': 3293, 'n_tokens_neox': 3095, 'n_words': 1051}