<commit_before># 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.

from asyncio.futures import TimeoutError
from collections import defaultdict
from concurrent.futures import as_completed
from operator import itemgetter
import asyncio
import contextlib
import logging
import math
import random
import threading

from bndl.net.serialize import attach, attachment
from bndl.util.collection import split
from bndl.util.exceptions import catch


logger = logging.getLogger(__name__)


AVAILABILITY_TIMEOUT = 1


class BlockSpec(object):
    def __init__(self, seeder, name, num_blocks):
        self.seeder = seeder
        self.name = name
        self.num_blocks = num_blocks


class Block(object):
    '''
    Helper class for exchanging blocks between peers. It uses the attach and
    attachment utilities from bndl.net.serialize to optimize sending the already
    serialized data.
    '''

    def __init__(self, block_id, data):
        self.id = block_id
        self.data = data


    def __getstate__(self):
        @contextlib.contextmanager
        def _attacher(loop, writer):
            @asyncio.coroutine
            def sender():
                writer.write(self.data)
            yield len(self.data), sender
        attach(str(self.id).encode(), _attacher)
        state = dict(self.__dict__)
        del state['data']
        return state


    def __setstate__(self, state):
        self.__dict__.update(state)
        self.data = attachment(str(self.id).encode())


def _batch_blocks(data, block_size):
    length = len(data)
    if isinstance(block_size, tuple):
        block_count, min_block_size, max_block_size = block_size
        assert block_count > 0
        assert min_block_size >= 0
        assert max_block_size > 0
        block_size = math.ceil(length / block_count)
        if block_size > max_block_size:
            block_size = max_block_size
        elif block_size < min_block_size:
            block_size = min_block_size
        return _batch_blocks(data, block_size)<commit_msg>work around race with to_disk<commit_after># 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.

from asyncio.futures import TimeoutError
from collections import defaultdict
from concurrent.futures import as_completed
from operator import itemgetter
import asyncio
import contextlib
import logging
import math
import random
import threading

from bndl.net.serialize import attach, attachment
from bndl.util.collection import split
from bndl.util.exceptions import catch


logger = logging.getLogger(__name__)


AVAILABILITY_TIMEOUT = 1


class BlockSpec(object):
    def __init__(self, seeder, name, num_blocks):
        self.seeder = seeder
        self.name = name
        self.num_blocks = num_blocks


class Block(object):
    '''
    Helper class for exchanging blocks between peers. It uses the attach and
    attachment utilities from bndl.net.serialize to optimize sending the already
    serialized data.
    '''

    def __init__(self, block_id, data):
        self.id = block_id
        self.data = data


    def __getstate__(self):
        data = self.data
        @contextlib.contextmanager
        def _attacher(loop, writer):
            @asyncio.coroutine
            def sender():
                writer.write(data)
            yield len(data), sender
        attach(str(self.id).encode(), _attacher)
        state = dict(self.__dict__)
        state.pop('data', None)
        return state


    def __setstate__(self, state):
        self.__dict__.update(state)
        self.data = attachment(str(self.id).encode())


def _batch_blocks(data, block_size):
    length = len(data)
    if isinstance(block_size, tuple):
        block_count, min_block_size, max_block_size = block_size
        assert block_count > 0
        assert min_block_size >= 0
        assert max_block_size > 0
        block_size = math.ceil(length / block_count)
        if block_size > max_block_size:
            block_size = max_block_size
        elif block_size < min_block_size:
            block_size = min_block_size
        return _batch_blocks(data, block_size)
{'fraction_non_alphanumeric': 0.07065324649694099, 'fraction_numerical': 0.0031576869942766925, 'mean_word_length': 3.041467304625199, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '1166355', 'n_tokens_mistral': 1413, 'n_tokens_neox': 1373, 'n_words': 473}