<commit_before>import datetime
import errno
import uuid
import os
import tempfile
from cStringIO import StringIO

from gevent import with_timeout, Timeout, spawn_later
from gevent.event import Event, AsyncResult
from spinoff.actor import Actor, Unhandled
from spinoff.contrib.filetransfer.util import read_file_async
from spinoff.util.logging import dbg, err
from spinoff.util.pattern_matching import ANY, IN
from spinoff.util.lockfile import lock_file


DEFAULT_CHUNK_SIZE = 10 * 1024 * 1024
OPEN_FILE_TIMEOUT = 30 * 60
FILE_MAX_LIFETIME = 60 * 60
# MAX_FILE_SIZE_IN_MEMORY = 10 * 1024 * 1024


class FileAlreadyPublished(Exception):
    pass


class _Sender(Actor):
    def run(self, service, pub_id, file, send_to):
        self.watch(send_to)
        seek_ptr = 0

        while True:
            try:
                msg = with_timeout(OPEN_FILE_TIMEOUT, self.get)
            except Timeout:
                err("Sending of file at %r timed out" % (file,))
                break

            if ('next-chunk', ANY) == msg:
                service << ('touch-file', pub_id)
                _, chunk_size = msg
                chunk = read_file_async(file, start=seek_ptr, end=seek_ptr + chunk_size)
                seek_ptr += len(chunk)
                more_coming = len(chunk) > 0 if chunk_size > 0 else True
                send_to << ('chunk', chunk, more_coming)
                if not more_coming:
                    break
            elif ('terminated', send_to) == msg:
                break
            else:
                self.unhandled(msg)


class FilePublisher(Actor):
    _instances = {}

    @classmethod
    def get(cls, node=None):
        if node not in cls._instances:
            cls._instances[node] = node.spawn(cls.using())
        return cls._instances[node]

    def pre_start(self):
        self.published = {}  # <pub_id> => (<local file path>, <time added>)
        self.senders = {}  # <sender> => <pub_id>

        self << 'purge-old'

    def _touch_file(self, pub_id):
        file_path, time_added = self.published[pub_id]
        self.published[pub_id] = (file_path, datetime.datetime.now())

    def receive(self, msg):
        if ('publish', ANY, ANY) == msg:
            _, file_path, pub_id = msg
            if not os.path.exists(file_path) and not os.path.isdir(file_path):
                err("attempt to publish a file that does not exist")
                return
            if pub_id in self.published:
                raise FileAlreadyPublished("Attempt to publish %r with ID %r but a file already exists with that ID" % (file_path, pub_id))
            else:
                self.published[pub_id] = (file_path, datetime.datetime.now())
        elif ('get-file', ANY, ANY) == msg:
            _, pub_id, send_to = msg
            if pub_id not in self.published:
                err("attempt to get a file with ID %r which has not been published or is not available anymore" % (pub_id,))
                return
            self._touch_file(pub_id)
            file_path, time_added = self.published[pub_id]
            sender = self.watch(_Sender.using(service=self.ref, pub_id=pub_id, file=file_path, send_to=send_to))
            self.senders[sender] = pub_id
            # self.senders[sender] = pub_id
            send_to << ('take-file', sender)
        elif ('touch-file', ANY) == msg:
            _, pub_id = msg
            if pub_id not in self.published:
                err("attempt to touch a file with ID %r which has not been published or is not available anymore" % (pub_id,))
                return
            _, pub_id = msg
            self._touch_file(pub_id)
        elif 'purge-old' == msg:
            spawn_later(60.0, self.send, 'purge-old')
            t = datetime.datetime.now()
            for pub_id, (file_path, time_added) in self.published.items():
                if (t - time_added).total_seconds() > FILE_MAX_LIFETIME and pub_id not in self.senders.values():
                    dbg("purging file %r at %r" % (pub_id, file_path))
                    del self.published[pub_id]
        elif ('terminated', IN(self.senders)) == msg:
            _, sender = msg
            del self.senders[sender]

    def post_stop(self):
        del self._instances[self.node]


class _Receiver(Actor):
    def run(self, pub_id, file_service):
        self.watch(file_service)

        file_service << ('get-file', pub_id, self.ref)

        msg = self.get(('take-file', ANY), ('terminated', file_service))

        if ('take-file', ANY) == msg:
            _, sender = msg
        elif ('terminated', file_service) == msg:
            _, _, d = self.get(('next-chunk', ANY, ANY))
            d.set_exception(Exception("file sender died prematurely"))
            return
        else:
            raise Unhandled(msg)

        self.watch(sender)

        while True:
            msg = self.get(('next-chunk', ANY, ANY), ('terminated', sender))

            if ('next-chunk', ANY, ANY) == msg:
                _, size, d = msg
                sender << ('next-chunk', size)
                _, chunk, more_coming = self.get(('chunk', ANY, ANY))
                d.set((chunk, more_coming))

            elif ('terminated', sender) == msg:
                break

            else:
                assert False


class FileRef(object):
    _fetching = {}

    def __init__(self, pub_id, file_service, abstract_path, mtime, size):
        """Private; see File.publish instead"""
        self.pub_id = pub_id
        self.file_service = file_service
        self.abstract_path = abstract_path
        self.mtime = mtime
        self.size = size

    @classmethod
    def publish(cls, path, node, abstract_path=None):
        if not os.path.exists(path):<commit_msg>Cleaned up contrib.filetransfer _Receiver and properly report failed transfers<commit_after>import datetime
import errno
import uuid
import os
import tempfile
from cStringIO import StringIO

from gevent import with_timeout, Timeout, spawn_later
from gevent.event import Event, AsyncResult
from spinoff.actor import Actor, Unhandled
from spinoff.contrib.filetransfer.util import read_file_async
from spinoff.util.logging import dbg, err
from spinoff.util.pattern_matching import ANY, IN
from spinoff.util.lockfile import lock_file


DEFAULT_CHUNK_SIZE = 10 * 1024 * 1024
OPEN_FILE_TIMEOUT = 30 * 60
FILE_MAX_LIFETIME = 60 * 60
# MAX_FILE_SIZE_IN_MEMORY = 10 * 1024 * 1024


class FileAlreadyPublished(Exception):
    pass


class _Sender(Actor):
    def run(self, service, pub_id, file, send_to):
        self.watch(send_to)
        seek_ptr = 0

        while True:
            try:
                msg = with_timeout(OPEN_FILE_TIMEOUT, self.get)
            except Timeout:
                err("Sending of file at %r timed out" % (file,))
                break

            if ('next-chunk', ANY) == msg:
                service << ('touch-file', pub_id)
                _, chunk_size = msg
                chunk = read_file_async(file, start=seek_ptr, end=seek_ptr + chunk_size)
                seek_ptr += len(chunk)
                more_coming = len(chunk) > 0 if chunk_size > 0 else True
                send_to << ('chunk', chunk, more_coming)
                if not more_coming:
                    break
            elif ('terminated', send_to) == msg:
                break
            else:
                self.unhandled(msg)


class FilePublisher(Actor):
    _instances = {}

    @classmethod
    def get(cls, node=None):
        if node not in cls._instances:
            cls._instances[node] = node.spawn(cls.using())
        return cls._instances[node]

    def pre_start(self):
        self.published = {}  # <pub_id> => (<local file path>, <time added>)
        self.senders = {}  # <sender> => <pub_id>

        self << 'purge-old'

    def _touch_file(self, pub_id):
        file_path, time_added = self.published[pub_id]
        self.published[pub_id] = (file_path, datetime.datetime.now())

    def receive(self, msg):
        if ('publish', ANY, ANY) == msg:
            _, file_path, pub_id = msg
            if not os.path.exists(file_path) and not os.path.isdir(file_path):
                err("attempt to publish a file that does not exist")
                return
            if pub_id in self.published:
                raise FileAlreadyPublished("Attempt to publish %r with ID %r but a file already exists with that ID" % (file_path, pub_id))
            else:
                self.published[pub_id] = (file_path, datetime.datetime.now())
        elif ('get-file', ANY, ANY) == msg:
            _, pub_id, send_to = msg
            if pub_id not in self.published:
                err("attempt to get a file with ID %r which has not been published or is not available anymore" % (pub_id,))
                return
            self._touch_file(pub_id)
            file_path, time_added = self.published[pub_id]
            sender = self.watch(_Sender.using(service=self.ref, pub_id=pub_id, file=file_path, send_to=send_to))
            self.senders[sender] = pub_id
            # self.senders[sender] = pub_id
            send_to << ('take-file', sender)
        elif ('touch-file', ANY) == msg:
            _, pub_id = msg
            if pub_id not in self.published:
                err("attempt to touch a file with ID %r which has not been published or is not available anymore" % (pub_id,))
                return
            _, pub_id = msg
            self._touch_file(pub_id)
        elif 'purge-old' == msg:
            spawn_later(60.0, self.send, 'purge-old')
            t = datetime.datetime.now()
            for pub_id, (file_path, time_added) in self.published.items():
                if (t - time_added).total_seconds() > FILE_MAX_LIFETIME and pub_id not in self.senders.values():
                    dbg("purging file %r at %r" % (pub_id, file_path))
                    del self.published[pub_id]
        elif ('terminated', IN(self.senders)) == msg:
            _, sender = msg
            del self.senders[sender]

    def post_stop(self):
        del self._instances[self.node]


class _Receiver(Actor):
    def run(self, pub_id, file_service):
        self.watch(file_service)

        file_service << ('get-file', pub_id, self.ref)

        msg = self.get(('take-file', ANY), ('terminated', file_service))

        if ('take-file', ANY) == msg:
            _, sender = msg
            self.watch(sender)
            while True:
                msg = self.get(('next-chunk', ANY, ANY), ('terminated', sender))
                if ('next-chunk', ANY, ANY) == msg:
                    _, size, d = msg
                    sender << ('next-chunk', size)
                    _, chunk, more_coming = self.get(('chunk', ANY, ANY))
                    d.set((chunk, more_coming))
                elif ('terminated', sender) == msg:
                    d.set_exception(Exception("file sender died prematurely"))
                else:
                    assert False
        elif ('terminated', file_service) == msg:
            _, _, d = self.get(('next-chunk', ANY, ANY))
            d.set_exception(Exception("file service died prematurely"))


class FileRef(object):
    _fetching = {}

    def __init__(self, pub_id, file_service, abstract_path, mtime, size):
        """Private; see File.publish instead"""
        self.pub_id = pub_id
        self.file_service = file_service
        self.abstract_path = abstract_path
        self.mtime = mtime
        self.size = size

    @classmethod
    def publish(cls, path, node, abstract_path=None):
        if not os.path.exists(path):
{'fraction_non_alphanumeric': 0.10735294117647058, 'fraction_numerical': 0.0058823529411764705, 'mean_word_length': 2.0256477361947134, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1877603', 'n_tokens_mistral': 3537, 'n_tokens_neox': 3346, 'n_words': 920}