<s> """<STR_LIT>""" <EOL> __authors__ = [ '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> import json <EOL> import logging <EOL> import pprint <EOL> import sys <EOL> import time <EOL> import traceback <EOL> from collections import defaultdict <EOL> from copy import deepcopy <EOL> from functools import partial <EOL> from tornado import gen , stack_context <EOL> from tornado . ioloop import IOLoop <EOL> from viewfinder . backend . base import counters , message , util <EOL> from viewfinder . backend . base . exceptions import FailpointError , InvalidRequestError , LimitExceededError , PermissionError <EOL> from viewfinder . backend . base . exceptions import CannotWaitError , NotFoundError , LockFailedError , StopOperationError <EOL> from viewfinder . backend . db . lock import Lock <EOL> from viewfinder . backend . db . lock_resource_type import LockResourceType <EOL> from viewfinder . backend . db . operation import Operation <EOL> from viewfinder . backend . op . op_mgr_db_client import OpMgrDBClient <EOL> from viewfinder . backend . op . op_context import OpContext <EOL> _avg_op_time = counters . define_average ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> _ops_per_min = counters . define_rate ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT> ) <EOL> _retries_per_min = counters . define_rate ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT> ) <EOL> _aborts_per_min = counters . define_rate ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT> ) <EOL> _ABORTABLE_EXCEPTIONS = ( <EOL> PermissionError , <EOL> InvalidRequestError , <EOL> LimitExceededError , <EOL> NotFoundError , <EOL> ) <EOL> _SMALLER_RETRY_EXCEPTIONS = ( <EOL> LockFailedError , <EOL> ) <EOL> class UserOpManager ( object ) : <EOL> """<STR_LIT>""" <EOL> _INITIAL_BACKOFF_SECS = <NUM_LIT> <EOL> _SMALL_INITIAL_BACKOFF_SECS = <NUM_LIT> <EOL> _MAX_BACKOFF_STEPS = <NUM_LIT:10> <EOL> _MAX_OPERATION_SIZE = <NUM_LIT> <EOL> def __init__ ( self , client , op_map , user_id , callback ) : <EOL> """<STR_LIT>""" <EOL> self . _client = OpMgrDBClient ( client ) <EOL> self . _op_map = op_map <EOL> self . _user_id = user_id <EOL> self . _sync_cb_map = defaultdict ( list ) <EOL> self . _callback = stack_context . wrap ( callback ) <EOL> self . _is_executing = False <EOL> def Drain ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> self . _callback = stack_context . wrap ( callback ) <EOL> self . Execute ( ) <EOL> @ gen . engine <EOL> def Execute ( self , operation_id = None , wait_callback = None ) : <EOL> """<STR_LIT>""" <EOL> def _OnCompletedOp ( type = None , value = None , tb = None ) : <EOL> """<STR_LIT>""" <EOL> if ( type , value , tb ) != ( None , None , None ) : <EOL> raise type , value , tb <EOL> wait_callback ( ) <EOL> @ gen . engine <EOL> def _ExecuteAll ( ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . _is_executing = True <EOL> self . _requery = True <EOL> while self . _requery : <EOL> yield self . _ExecuteAll ( operation_id = operation_id ) <EOL> finally : <EOL> for cb_op_id in self . _sync_cb_map . keys ( ) : <EOL> self . _InvokeSyncCallbacks ( cb_op_id ) <EOL> self . _is_executing = False <EOL> self . _callback ( ) <EOL> if wait_callback is not None : <EOL> self . _sync_cb_map [ operation_id ] . append ( stack_context . wrap ( _OnCompletedOp ) ) <EOL> if not self . _is_executing : <EOL> with stack_context . StackContext ( OpContext ( ) ) : <EOL> _ExecuteAll ( ) <EOL> else : <EOL> self . _requery = True <EOL> @ gen . coroutine <EOL> def _ExecuteAll ( self , operation_id = None ) : <EOL> """<STR_LIT>""" <EOL> self . _requery = False <EOL> results = yield gen . Task ( Lock . TryAcquire , <EOL> self . _client , <EOL> LockResourceType . Operation , <EOL> str ( self . _user_id ) , <EOL> resource_data = operation_id , <EOL> detect_abandonment = True ) <EOL> self . _lock , status = results . args <EOL> if status == Lock . FAILED_TO_ACQUIRE_LOCK : <EOL> for operation_id in self . _sync_cb_map . keys ( ) : <EOL> self . _InvokeSyncCallbacks ( operation_id , CannotWaitError , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> return <EOL> try : <EOL> next_ops = None <EOL> if status == Lock . ACQUIRED_ABANDONED_LOCK and self . _lock . resource_data is not None : <EOL> op = yield gen . Task ( Operation . Query , <EOL> self . _client , <EOL> self . _user_id , <EOL> self . _lock . resource_data , <EOL> col_names = None , <EOL> must_exist = False , <EOL> consistent_read = True ) <EOL> next_ops = [ op ] <EOL> last_op_id = None <EOL> while True : <EOL> if next_ops is None : <EOL> next_ops = yield gen . Task ( Operation . RangeQuery , <EOL> self . _client , <EOL> self . _user_id , <EOL> range_desc = None , <EOL> limit = <NUM_LIT:10> , <EOL> col_names = None , <EOL> excl_start_key = last_op_id , <EOL> consistent_read = True ) <EOL> if len ( next_ops ) == <NUM_LIT:0> : <EOL> break <EOL> for op in next_ops : <EOL> if not op . quarantine or not op . IsBackedOff ( ) : <EOL> yield self . _ExecuteOp ( op ) <EOL> last_op_id = None <EOL> break <EOL> else : <EOL> logging . info ( '<STR_LIT>' % <EOL> ( op . operation_id , op . user_id , op . backoff - time . time ( ) ) ) <EOL> last_op_id = op . operation_id <EOL> next_ops = None <EOL> finally : <EOL> yield gen . Task ( self . _lock . Release , self . _client ) <EOL> if self . _lock . acquire_failures is not None : <EOL> logging . info ( '<STR_LIT>' % self . _lock ) <EOL> self . _requery = True <EOL> @ gen . coroutine <EOL> def _ExecuteOp ( self , op ) : <EOL> """<STR_LIT>""" <EOL> if op . backoff is not None : <EOL> yield gen . Task ( IOLoop . current ( ) . add_timeout , op . backoff ) <EOL> with OpContext . current ( ) . Enter ( op ) : <EOL> op_entry = self . _op_map [ op . method ] <EOL> op_args = json . loads ( op . json ) <EOL> if self . _lock . resource_data != op . operation_id : <EOL> self . _lock . resource_data = op . operation_id <EOL> yield gen . Task ( self . _lock . Update , self . _client ) <EOL> op_message = message . Message ( op_args ) <EOL> yield gen . Task ( op_message . Migrate , <EOL> self . _client , <EOL> migrate_version = message . MAX_MESSAGE_VERSION , <EOL> migrators = op_entry . migrators ) <EOL> try : <EOL> del op_args [ '<STR_LIT>' ] <EOL> scrubbed_op_args = op_args <EOL> if op_entry . scrubber is not None : <EOL> scrubbed_op_args = deepcopy ( op_args ) <EOL> op_entry . scrubber ( scrubbed_op_args ) <EOL> args_str = pprint . pformat ( scrubbed_op_args ) <EOL> logging . info ( '<STR_LIT>' % <EOL> ( op . user_id , op . device_id , op . operation_id , op . method , <EOL> ( '<STR_LIT:\n>' if args_str . find ( '<STR_LIT:\n>' ) != - <NUM_LIT:1> else '<STR_LIT:U+0020>' ) , args_str ) ) <EOL> _ops_per_min . increment ( ) <EOL> if op . attempts > <NUM_LIT:0> : <EOL> _retries_per_min . increment ( ) <EOL> self . _client . ResetDBModified ( ) <EOL> results = yield gen . Task ( op_entry . handler , self . _client , ** op_args ) <EOL> elapsed_secs = time . time ( ) - op . timestamp <EOL> logging . info ( '<STR_LIT>' % <EOL> ( op . user_id , op . device_id , op . operation_id , op . method , elapsed_secs , <EOL> ( '<STR_LIT>' % pprint . pformat ( results ) if results else '<STR_LIT>' ) ) ) <EOL> _avg_op_time . add ( elapsed_secs ) <EOL> self . _InvokeSyncCallbacks ( op . operation_id ) <EOL> yield self . _DeleteOp ( op ) <EOL> except StopOperationError : <EOL> pass <EOL> except FailpointError : <EOL> type , value , tb = sys . exc_info ( ) <EOL> logging . warning ( '<STR_LIT>' , value . filename , value . lineno ) <EOL> except Exception : <EOL> type , value , tb = sys . exc_info ( ) <EOL> self . _InvokeSyncCallbacks ( op . operation_id , type , value , tb ) <EOL> if op . attempts == <NUM_LIT:0> and issubclass ( type , _ABORTABLE_EXCEPTIONS ) : <EOL> yield self . _AbortOp ( op , type , value , tb ) <EOL> else : <EOL> initial_backoff = UserOpManager . _INITIAL_BACKOFF_SECS <EOL> if issubclass ( type , _SMALLER_RETRY_EXCEPTIONS ) : <EOL> initial_backoff = UserOpManager . _SMALL_INITIAL_BACKOFF_SECS <EOL> yield self . _FailOp ( op , type , value , tb , initial_backoff_secs = initial_backoff ) <EOL> @ gen . coroutine <EOL> def _AbortOp ( self , op , type , value , tb ) : <EOL> """<STR_LIT>""" <EOL> if self . _client . HasDBBeenModified ( ) : <EOL> stackDumpLines = '<STR_LIT>' . join ( traceback . format_list ( self . _client . GetModifiedDBStack ( ) ) ) <EOL> logging . error ( '<STR_LIT>' % stackDumpLines ) <EOL> yield self . _FailOp ( op , type , value , tb ) <EOL> else : <EOL> elapsed_secs = time . time ( ) - op . timestamp <EOL> logging . warning ( '<STR_LIT>' % <EOL> ( op . user_id , op . device_id , op . operation_id , op . method , elapsed_secs , value ) ) <EOL> _avg_op_time . add ( elapsed_secs ) <EOL> _aborts_per_min . increment ( ) <EOL> yield self . _DeleteOp ( op ) <EOL> @ gen . coroutine <EOL> def _FailOp ( self , op , type , value , tb , initial_backoff_secs = _INITIAL_BACKOFF_SECS ) : <EOL> """<STR_LIT>""" <EOL> elapsed_secs = time . time ( ) - op . timestamp <EOL> logging . error ( '<STR_LIT>' % <EOL> ( op . user_id , op . device_id , op . operation_id , op . method , elapsed_secs , value ) , <EOL> exc_info = ( type , value , tb ) ) <EOL> exc = '<STR_LIT>' . join ( traceback . format_exception ( type , value , tb ) ) <EOL> if op . first_failure : <EOL> op . last_failure = exc <EOL> else : <EOL> op . first_failure = exc <EOL> max_traceback_size = UserOpManager . _MAX_OPERATION_SIZE - len ( op . json ) <EOL> if ( len ( op . first_failure or '<STR_LIT>' ) + len ( op . last_failure or '<STR_LIT>' ) ) > max_traceback_size : <EOL> if op . first_failure and op . last_failure : <EOL> max_traceback_size = int ( max_traceback_size / <NUM_LIT:2> ) <EOL> if op . first_failure : <EOL> op . first_failure = op . first_failure [ - max_traceback_size : ] <EOL> if op . last_failure : <EOL> op . last_failure = op . last_failure [ - max_traceback_size : ] <EOL> op . backoff = time . time ( ) + ( <NUM_LIT:1> << min ( op . attempts , UserOpManager . _MAX_BACKOFF_STEPS ) ) * initial_backoff_secs <EOL> retry_after_secs = op . backoff - time . time ( ) <EOL> op . attempts = op . attempts + <NUM_LIT:1> <EOL> if op . attempts < <NUM_LIT:3> : <EOL> logging . warning ( '<STR_LIT>' % <EOL> ( op . attempts , util . Pluralize ( op . attempts ) , op . operation_id , op . user_id , retry_after_secs ) ) <EOL> else : <EOL> logging . warning ( '<STR_LIT>' <EOL> '<STR_LIT>' % <EOL> ( op . attempts , util . Pluralize ( op . attempts ) , op . operation_id , op . user_id , retry_after_secs ) ) <EOL> self . _last_op_id = op . operation_id <EOL> op . quarantine = <NUM_LIT:1> <EOL> yield gen . Task ( op . Update , self . _client ) <EOL> @ gen . coroutine <EOL> def _DeleteOp ( self , op ) : <EOL> """<STR_LIT>""" <EOL> self . _last_op_id = op . operation_id <EOL> try : <EOL> yield gen . Task ( op . Delete , self . _client ) <EOL> except Exception : <EOL> logging . warning ( '<STR_LIT>' % <EOL> ( op . method , op . operation_id ) , exc_info = True ) <EOL> def _InvokeSyncCallbacks ( self , operation_id , type = None , value = None , tb = None ) : <EOL> """<STR_LIT>""" <EOL> sync_cb_list = self . _sync_cb_map . pop ( operation_id , None ) <EOL> if sync_cb_list is not None : <EOL> for sync_cb in sync_cb_list : <EOL> IOLoop . current ( ) . add_callback ( partial ( sync_cb , type , value , tb ) ) </s>
