<commit_before>import os
import sys
import json
import re
from shlex import quote as shq
import logging
import asyncio
import random
import traceback
import base64
import uuid
import shutil
import aiohttp
import aiohttp.client_exceptions
from aiohttp import web
import async_timeout
import concurrent
import aiodocker
from aiodocker.exceptions import DockerError
import google.oauth2.service_account
from hailtop.utils import time_msecs, request_retry_transient_errors, RETRY_FUNCTION_SCRIPT, \
    sleep_and_backoff, retry_all_errors

# import uvloop

from hailtop.config import DeployConfig
from gear import configure_logging

from .utils import parse_cpu_in_mcpu, parse_image_tag, parse_memory_in_bytes, \
    adjust_cores_for_memory_request, cores_mcpu_to_memory_bytes
from .semaphore import FIFOWeightedSemaphore, NullWeightedSemaphore
from .log_store import LogStore
from .globals import HTTP_CLIENT_MAX_SIZE
from .batch_format_version import BatchFormatVersion

# uvloop.install()

configure_logging()
log = logging.getLogger('batch-worker')

MAX_IDLE_TIME_MSECS = 30 * 1000
MAX_DOCKER_IMAGE_PULL_SECS = 20 * 60
MAX_DOCKER_WAIT_SECS = 5 * 60
MAX_DOCKER_OTHER_OPERATION_SECS = 1 * 60

CORES = int(os.environ['CORES'])
NAME = os.environ['NAME']
NAMESPACE = os.environ['NAMESPACE']
# ACTIVATION_TOKEN
IP_ADDRESS = os.environ['IP_ADDRESS']
BUCKET_NAME = os.environ['BUCKET_NAME']
INSTANCE_ID = os.environ['INSTANCE_ID']
PROJECT = os.environ['PROJECT']
WORKER_TYPE = os.environ['WORKER_TYPE']

log.info(f'CORES {CORES}')
log.info(f'NAME {NAME}')
log.info(f'NAMESPACE {NAMESPACE}')
# ACTIVATION_TOKEN
log.info(f'IP_ADDRESS {IP_ADDRESS}')
log.info(f'WORKER_TYPE {WORKER_TYPE}')
log.info(f'BUCKET_NAME {BUCKET_NAME}')
log.info(f'INSTANCE_ID {INSTANCE_ID}')
log.info(f'PROJECT {PROJECT}')

deploy_config = DeployConfig('gce', NAMESPACE, {})

docker = None

port_allocator = None

worker = None


class PortAllocator:
    def __init__(self):
        self.ports = asyncio.Queue()
        port_base = 46572
        for port in range(port_base, port_base + 10):
            self.ports.put_nowait(port)

    async def allocate(self):
        return await self.ports.get()

    def free(self, port):
        self.ports.put_nowait(port)


def docker_call_retry(timeout, name):
    async def wrapper(f, *args, **kwargs):
        delay = 0.1
        while True:
            try:
                return await asyncio.wait_for(f(*args, **kwargs), timeout)
            except DockerError as e:
                # 408 request timeout, 503 service unavailable
                if e.status == 408 or e.status == 503:
                    log.exception(f'in docker call to {f.__name__} for {name}, retrying', stack_info=True)
                # DockerError(500, 'Get https://registry-1.docker.io/v2/: net/http: request canceled while waiting for connection (Client.Timeout exceeded while awaiting headers)
                # DockerError(500, 'error creating overlay mount to /var/lib/docker/overlay2/545a1337742e0292d9ed197b06fe900146c85ab06e468843cd0461c3f34df50d/merged: device or resource busy'
                elif e.status == 500 and ("request canceled while waiting for connection" in e.message or
                                          re.match("error creating overlay mount.*device or resource busy", e.message)):
                    log.exception(f'in docker call to {f.__name__} for {name}, retrying', stack_info=True)
                else:
                    raise
            except asyncio.CancelledError:  # pylint: disable=try-except-raise
                raise<commit_msg>[batch] add transient error (#8189)<commit_after>import os
import sys
import json
import re
from shlex import quote as shq
import logging
import asyncio
import random
import traceback
import base64
import uuid
import shutil
import aiohttp
import aiohttp.client_exceptions
from aiohttp import web
import async_timeout
import concurrent
import aiodocker
from aiodocker.exceptions import DockerError
import google.oauth2.service_account
from hailtop.utils import time_msecs, request_retry_transient_errors, RETRY_FUNCTION_SCRIPT, \
    sleep_and_backoff, retry_all_errors

# import uvloop

from hailtop.config import DeployConfig
from gear import configure_logging

from .utils import parse_cpu_in_mcpu, parse_image_tag, parse_memory_in_bytes, \
    adjust_cores_for_memory_request, cores_mcpu_to_memory_bytes
from .semaphore import FIFOWeightedSemaphore, NullWeightedSemaphore
from .log_store import LogStore
from .globals import HTTP_CLIENT_MAX_SIZE
from .batch_format_version import BatchFormatVersion

# uvloop.install()

configure_logging()
log = logging.getLogger('batch-worker')

MAX_IDLE_TIME_MSECS = 30 * 1000
MAX_DOCKER_IMAGE_PULL_SECS = 20 * 60
MAX_DOCKER_WAIT_SECS = 5 * 60
MAX_DOCKER_OTHER_OPERATION_SECS = 1 * 60

CORES = int(os.environ['CORES'])
NAME = os.environ['NAME']
NAMESPACE = os.environ['NAMESPACE']
# ACTIVATION_TOKEN
IP_ADDRESS = os.environ['IP_ADDRESS']
BUCKET_NAME = os.environ['BUCKET_NAME']
INSTANCE_ID = os.environ['INSTANCE_ID']
PROJECT = os.environ['PROJECT']
WORKER_TYPE = os.environ['WORKER_TYPE']

log.info(f'CORES {CORES}')
log.info(f'NAME {NAME}')
log.info(f'NAMESPACE {NAMESPACE}')
# ACTIVATION_TOKEN
log.info(f'IP_ADDRESS {IP_ADDRESS}')
log.info(f'WORKER_TYPE {WORKER_TYPE}')
log.info(f'BUCKET_NAME {BUCKET_NAME}')
log.info(f'INSTANCE_ID {INSTANCE_ID}')
log.info(f'PROJECT {PROJECT}')

deploy_config = DeployConfig('gce', NAMESPACE, {})

docker = None

port_allocator = None

worker = None


class PortAllocator:
    def __init__(self):
        self.ports = asyncio.Queue()
        port_base = 46572
        for port in range(port_base, port_base + 10):
            self.ports.put_nowait(port)

    async def allocate(self):
        return await self.ports.get()

    def free(self, port):
        self.ports.put_nowait(port)


def docker_call_retry(timeout, name):
    async def wrapper(f, *args, **kwargs):
        delay = 0.1
        while True:
            try:
                return await asyncio.wait_for(f(*args, **kwargs), timeout)
            except DockerError as e:
                # 408 request timeout, 503 service unavailable
                if e.status == 408 or e.status == 503:
                    log.exception(f'in docker call to {f.__name__} for {name}, retrying', stack_info=True)
                # DockerError(500, 'Get https://registry-1.docker.io/v2/: net/http: request canceled while waiting for connection (Client.Timeout exceeded while awaiting headers)
                # DockerError(500, 'error creating overlay mount to /var/lib/docker/overlay2/545a1337742e0292d9ed197b06fe900146c85ab06e468843cd0461c3f34df50d/merged: device or resource busy'
                # DockerError(500, 'Get https://gcr.io/v2/: dial tcp: lookup gcr.io: Temporary failure in name resolution')
                elif e.status == 500 and ("request canceled while waiting for connection" in e.message or
                                          re.match("error creating overlay mount.*device or resource busy", e.message) or
                                          "Temporary failure in name resolution" in e.message):
                    log.exception(f'in docker call to {f.__name__} for {name}, retrying', stack_info=True)
                else:
                    raise
            except asyncio.CancelledError:  # pylint: disable=try-except-raise
                raise
{'fraction_non_alphanumeric': 0.08968974393713589, 'fraction_numerical': 0.027367565370545997, 'mean_word_length': 3.5427692307692307, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 3, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5285430', 'n_tokens_mistral': 2493, 'n_tokens_neox': 2321, 'n_words': 585}