<commit_before>import os
import copy
import functools
import time
import imp
import traceback
import platform

from asf.asf_context import ASFError
#TODO: migrate ASFError out of ASF to a more root location
import gevent.pool
import gevent.socket
import gevent.threadpool
import gevent.greenlet

import pp_crypt
import context

import ll

ml = ll.LLogger()


@functools.wraps(gevent.spawn)
def spawn(*a, **kw):
    if a:
        f = a[0]
    else:
        f = kw['run']
    # NOTE: tested functools.wraps to take 3.827 microseconds, which is okay
    gr = gevent.spawn(
        functools.wraps(f)(_exception_catcher), *a, **kw)
    context.get_context().greenlet_ancestors[gr] = gevent.getcurrent()
    return gr


sleep = gevent.sleep  # alias gevent.sleep here so user doesn't have to know/worry about gevent


def _exception_catcher(f, *a, **kw):
    try:
        return f(*a, **kw)
    except Exception as e:  # NOTE: would rather do this with weakrefs,
        if not hasattr(e, '__greenlet_traces'):  # but Exceptions are not weakref-able
            e.__greenlet_traces = []
        traces = e.__greenlet_traces
        traces.append(traceback.format_exc())
        traces.append(repr(gevent.getcurrent()))
        raise


def get_parent(greenlet=None):
    return context.get_context().greenlet_ancestors.get(greenlet or gevent.getcurrent())


def get_cur_correlation_id():
    ctx = context.get_context()
    ancestors = ctx.greenlet_ancestors
    corr_ids = ctx.greenlet_correlation_ids
    cur = gevent.getcurrent()
    #walk ancestors looking for a correlation id
    while cur not in corr_ids and cur in ancestors:
        cur = ancestors[cur]
    #if no correlation id found, create a new one at highest level
    if cur not in corr_ids:
        #this is reproducing CalUtility.cpp
        #TODO: where do different length correlation ids come from in CAL logs?
        t = time.time()
        corr_val = "{0}{1}{2}{3}".format(gevent.socket.gethostname(),
                                         os.getpid(), int(t), int(t % 1 * 10 ** 6))
        corr_id = "{0:x}{1:x}".format(pp_crypt.fnv_hash(corr_val), int(t % 1 * 10 ** 6))
        corr_ids[cur] = corr_id
    return corr_ids[cur]


def set_cur_correlation_id(corr_id):
    context.get_context().greenlet_correlation_ids[gevent.getcurrent()] = corr_id


def cpu_bound(f):
    '''
    decorator to mark a function as cpu-heavy; will be executed in a separate
    thread to avoid blocking any socket communication
    '''
    q_depth = [0]   # just 1 depth for the 1 IO thread we think

    @functools.wraps(f)
    def g(*a, **kw):
        q_depth[0] = q_depth[0] + 1
        enqueued = curtime()  # better than microsecond precision
        ctx = context.get_context()
        ctx.stats['cpu_bound.depth'].add(q_depth[0])  # always count this
        started = []

        def in_thread(*a, **kw):
            ml.ld3("In thread {0}", f.__name__)
            started.append(curtime())
            q_depth[0] = q_depth[0] - 1
            ctx.stats['cpu_bound.depth'].add(q_depth[0])
            return f(*a, **kw)
        #some modules import things lazily; it is too dangerous to run a function
        #in another thread if the import lock is held by the current thread
        #(this happens rarely -- only if the cpu_bound function is being executed
        #at the import time of a module)
        if not ctx.cpu_thread_enabled or imp.lock_held():
            ret = in_thread(*a, **kw)
        else:
            tlocals = ctx.thread_locals
            if getattr(tlocals, 'in_cpu_thread', False):
                ret = in_thread(*a, **kw)
            else:
                if not hasattr(tlocals, 'cpu_thread'):
                    tlocals.cpu_thread = gevent.threadpool.ThreadPool(1)
                    ml.ld2("Getting new thread pool for CPU bound {0}", id(tlocals.cpu_thread))

                    def set_flag():
                        tlocals.in_cpu_thread = True
                    tlocals.cpu_thread.apply_e((Exception,), set_flag, (), {})
                    ml.ld3("Enqueued to thread {0}", f.__name__)
                ret = tlocals.cpu_thread.apply_e((Exception,), in_thread, a, kw)
        start = started[0]
        duration = curtime() - start
        queued = start - enqueued
        ctx.stats['cpu_bound.' + f.__name__ + '.queued(ms)'].add(queued * 1000)<commit_msg>use built in Q depth<commit_after>import os
import copy
import functools
import time
import imp
import traceback
import platform

from asf.asf_context import ASFError
#TODO: migrate ASFError out of ASF to a more root location
import gevent.pool
import gevent.socket
import gevent.threadpool
import gevent.greenlet

import pp_crypt
import context

import ll

ml = ll.LLogger()


@functools.wraps(gevent.spawn)
def spawn(*a, **kw):
    if a:
        f = a[0]
    else:
        f = kw['run']
    # NOTE: tested functools.wraps to take 3.827 microseconds, which is okay
    gr = gevent.spawn(
        functools.wraps(f)(_exception_catcher), *a, **kw)
    context.get_context().greenlet_ancestors[gr] = gevent.getcurrent()
    return gr


sleep = gevent.sleep  # alias gevent.sleep here so user doesn't have to know/worry about gevent


def _exception_catcher(f, *a, **kw):
    try:
        return f(*a, **kw)
    except Exception as e:  # NOTE: would rather do this with weakrefs,
        if not hasattr(e, '__greenlet_traces'):  # but Exceptions are not weakref-able
            e.__greenlet_traces = []
        traces = e.__greenlet_traces
        traces.append(traceback.format_exc())
        traces.append(repr(gevent.getcurrent()))
        raise


def get_parent(greenlet=None):
    return context.get_context().greenlet_ancestors.get(greenlet or gevent.getcurrent())


def get_cur_correlation_id():
    ctx = context.get_context()
    ancestors = ctx.greenlet_ancestors
    corr_ids = ctx.greenlet_correlation_ids
    cur = gevent.getcurrent()
    #walk ancestors looking for a correlation id
    while cur not in corr_ids and cur in ancestors:
        cur = ancestors[cur]
    #if no correlation id found, create a new one at highest level
    if cur not in corr_ids:
        #this is reproducing CalUtility.cpp
        #TODO: where do different length correlation ids come from in CAL logs?
        t = time.time()
        corr_val = "{0}{1}{2}{3}".format(gevent.socket.gethostname(),
                                         os.getpid(), int(t), int(t % 1 * 10 ** 6))
        corr_id = "{0:x}{1:x}".format(pp_crypt.fnv_hash(corr_val), int(t % 1 * 10 ** 6))
        corr_ids[cur] = corr_id
    return corr_ids[cur]


def set_cur_correlation_id(corr_id):
    context.get_context().greenlet_correlation_ids[gevent.getcurrent()] = corr_id


def cpu_bound(f):
    '''
    decorator to mark a function as cpu-heavy; will be executed in a separate
    thread to avoid blocking any socket communication
    '''

    @functools.wraps(f)
    def g(*a, **kw):
        enqueued = curtime()  # better than microsecond precision
        ctx = context.get_context()
        tlocals = ctx.thread_locals
        started = []

        def in_thread(*a, **kw):
            ml.ld3("In thread {0}", f.__name__)
            started.append(curtime())
            if hasattr(tlocals, 'cpu_thread') and tlocals.cpu_thread:
                ctx.stats['cpu_bound.depth'].add(tlocals.cpu_thread.task_queue.qsize())
            return f(*a, **kw)
        #some modules import things lazily; it is too dangerous to run a function
        #in another thread if the import lock is held by the current thread
        #(this happens rarely -- only if the cpu_bound function is being executed
        #at the import time of a module)
        if not ctx.cpu_thread_enabled or imp.lock_held():
            ret = in_thread(*a, **kw)
        else:
            if getattr(tlocals, 'in_cpu_thread', False):
                ret = in_thread(*a, **kw)
            else:
                if not hasattr(tlocals, 'cpu_thread'):
                    tlocals.cpu_thread = gevent.threadpool.ThreadPool(1)
                    ml.ld2("Getting new thread pool for CPU bound {0}", id(tlocals.cpu_thread))

                    def set_flag():
                        tlocals.in_cpu_thread = True
                    tlocals.cpu_thread.apply_e((Exception,), set_flag, (), {})
                ret = tlocals.cpu_thread.apply_e((Exception,), in_thread, a, kw)
                ml.ld3("Enqueued to thread {0}", f.__name__)
        start = started[0]
        duration = curtime() - start
        queued = start - enqueued
        ctx.stats['cpu_bound.' + f.__name__ + '.queued(ms)'].add(queued * 1000)
{'fraction_non_alphanumeric': 0.10560595487322633, 'fraction_numerical': 0.008490346592230752, 'mean_word_length': 2.5069331158238173, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '543285', 'n_tokens_mistral': 2767, 'n_tokens_neox': 2684, 'n_words': 762}