<s> import logging <EOL> import traceback <EOL> import time <EOL> import sys <EOL> import uuid <EOL> from abc import ABCMeta <EOL> from abc import abstractmethod <EOL> from collections import OrderedDict <EOL> import boto <EOL> import venusian <EOL> import pyswfaws . decorators <EOL> from boto . swf . layer1_decisions import Layer1Decisions <EOL> from models import * <EOL> from serializers import * <EOL> from datastores import * <EOL> from promise import Timer as PTimer , Marker as PMarker <EOL> from pyswfaws . exceptions import UnfulfilledPromiseException <EOL> class DistributedDecisionWorker : <EOL> """<STR_LIT>""" <EOL> logger = logging . getLogger ( __name__ ) <EOL> def __init__ ( self , decision_function , decision_object = None , swf_domain = None , swf_task_list = None , workflow_type = None , <EOL> workflow_version = None , aws_access_key_id = None , aws_secret_access_key = None ) : <EOL> """<STR_LIT>""" <EOL> self . _decision_object = decision_object <EOL> self . _swf = boto . connect_swf ( aws_access_key_id = aws_access_key_id , <EOL> aws_secret_access_key = aws_secret_access_key ) <EOL> self . _decision_function = decision_function <EOL> registry = Registry ( ) <EOL> self . _scanner = venusian . Scanner ( registry = registry , mode = '<STR_LIT>' , caller = '<STR_LIT>' ) <EOL> self . _scanner . parent_decision_worker = decision_function <EOL> self . _scanner . scan ( sys . modules [ decision_function . orig . __module__ ] , categories = ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) ) <EOL> PTimer . is_remote_mode = True <EOL> PMarker . is_remote_mode = True <EOL> if hasattr ( self . _decision_function , '<STR_LIT>' ) : <EOL> self . _swf_domain = self . choose_first_not_none ( '<STR_LIT>' <EOL> '<STR_LIT>' , swf_domain , <EOL> decision_function . swf_options [ '<STR_LIT>' ] ) <EOL> self . _swf_task_list = self . choose_first_not_none ( '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> swf_task_list , decision_function . swf_options [ '<STR_LIT>' ] ) <EOL> self . _wf_type = self . choose_first_not_none ( '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> workflow_type , decision_function . swf_options [ '<STR_LIT>' ] ) <EOL> self . _wf_version = self . choose_first_not_none ( '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> workflow_version , decision_function . swf_options [ <EOL> '<STR_LIT>' ] ) <EOL> self . _input_serializer = decision_function . serialization_options [ '<STR_LIT>' ] <EOL> self . _input_data_store = decision_function . serialization_options [ '<STR_LIT>' ] <EOL> self . _result_serializer = decision_function . serialization_options [ '<STR_LIT>' ] <EOL> self . _result_data_store = decision_function . serialization_options [ '<STR_LIT>' ] <EOL> else : <EOL> raise Exception ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def handle_no_op ( self , decision_task ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def handle_exception ( self , exception , decision_context ) : <EOL> """<STR_LIT>""" <EOL> self . logger . exception ( '<STR_LIT>' ) <EOL> decision_context . decisions = Layer1Decisions ( ) <EOL> decision_context . decisions . fail_workflow_execution ( reason = '<STR_LIT>' , details = exception . message [ : <NUM_LIT> ] ) <EOL> return False <EOL> def start ( self ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> try : <EOL> decision_task = self . _swf . poll_for_decision_task ( domain = self . _swf_domain , task_list = self . _swf_task_list ) <EOL> if '<STR_LIT>' not in decision_task : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> self . handle_no_op ( decision_task ) <EOL> continue <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> history = decision_task [ '<STR_LIT>' ] <EOL> next_page_token = decision_task . get ( '<STR_LIT>' ) <EOL> while next_page_token : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> additional_history = self . _swf . poll_for_decision_task ( domain = self . _swf_domain , <EOL> task_list = self . _swf_task_list , <EOL> next_page_token = next_page_token ) <EOL> if '<STR_LIT>' in additional_history : <EOL> history . extend ( additional_history [ '<STR_LIT>' ] ) <EOL> decision_task [ '<STR_LIT>' ] = list ( ) <EOL> next_page_token = additional_history . get ( '<STR_LIT>' ) <EOL> except Exception as e : <EOL> self . logger . exception ( '<STR_LIT>' ) <EOL> raise e <EOL> try : <EOL> decision_context = self . _populate_decision_context ( decision_task , history ) <EOL> self . _scanner . decision_context = decision_context <EOL> PTimer . decision_context = decision_context <EOL> PMarker . decision_context = decision_context <EOL> except Exception as e : <EOL> self . logger . exception ( '<STR_LIT>' ) <EOL> message = str ( e ) <EOL> if message : <EOL> message = message [ : <NUM_LIT> ] <EOL> decision_context . decisions = Layer1Decisions ( ) <EOL> decision_context . decisions . fail_workflow_execution ( reason = '<STR_LIT>' , <EOL> details = message ) <EOL> self . _swf . respond_decision_task_completed ( decision_task [ '<STR_LIT>' ] , <EOL> decision_context . decisions . _data ) <EOL> return <EOL> args = ( list ( ) , dict ( ) ) <EOL> if decision_context . workflow . input : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> serialized_args = self . _input_data_store . get ( decision_context . workflow . input ) <EOL> args = self . _input_serializer . deserialize_input ( serialized_args ) <EOL> if self . _decision_object : <EOL> args [ <NUM_LIT:0> ] . insert ( <NUM_LIT:0> , self . _decision_object ) <EOL> try : <EOL> finished = False <EOL> result = self . _decision_function ( * args [ <NUM_LIT:0> ] , ** args [ <NUM_LIT:1> ] ) <EOL> finished = True <EOL> except UnfulfilledPromiseException : <EOL> finished = False <EOL> except Exception as e : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> should_exit = self . handle_exception ( e , decision_context ) <EOL> if should_exit : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> return <EOL> if finished is True : <EOL> swf_result = None <EOL> if result : <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> serialized_result = self . _result_serializer . serialize_result ( result ) <EOL> key = '<STR_LIT>' . format ( decision_context . workflow . run_id ) <EOL> swf_result = self . _result_data_store . put ( serialized_result , key ) <EOL> decision_context . decisions . complete_workflow_execution ( result = swf_result ) <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> try : <EOL> self . _swf . respond_decision_task_completed ( decision_task [ '<STR_LIT>' ] , <EOL> decision_context . decisions . _data ) <EOL> except Exception : <EOL> self . logger . exception ( '<STR_LIT>' ) <EOL> @ staticmethod <EOL> def choose_first_not_none ( exception_message , * args ) : <EOL> for arg in args : <EOL> if arg is not None : <EOL> return arg <EOL> raise Exception ( exception_message ) <EOL> @ staticmethod <EOL> def _populate_decision_context ( decision_task , history ) : <EOL> """<STR_LIT>""" <EOL> activities = OrderedDict ( ) <EOL> signals = list ( ) <EOL> timers = OrderedDict ( ) <EOL> child_workflows = OrderedDict ( ) <EOL> workflow = Workflow ( ) <EOL> cache_markers = list ( ) <EOL> user_markers = list ( ) <EOL> id_count = <NUM_LIT:1> <EOL> workflow . run_id = decision_task [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> for e in history : <EOL> et = e [ '<STR_LIT>' ] <EOL> if et == '<STR_LIT>' : <EOL> workflow . continued_execution_run_id = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> workflow . input = e [ '<STR_LIT>' ] . get ( '<STR_LIT:input>' ) <EOL> workflow . child_policy = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> workflow . continued_execution_run_id = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> workflow . execution_start_to_close_timeout = e [ '<STR_LIT>' ] . get ( <EOL> '<STR_LIT>' ) <EOL> workflow . lambda_role = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> wf_type = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> if wf_type : <EOL> workflow . parent_workflow_id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> workflow . parent_run_id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> workflow . tag_list = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> workflow . task_list = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> workflow . task_priority = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> workflow . type = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT:name>' ] <EOL> workflow . version = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT:version>' ] <EOL> workflow . state = '<STR_LIT>' <EOL> elif et == '<STR_LIT>' : <EOL> workflow . state = '<STR_LIT>' <EOL> workflow . termination_cause = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> workflow . termination_detail = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> workflow . termination_reason = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> elif et == '<STR_LIT>' : <EOL> workflow . state = '<STR_LIT>' <EOL> workflow . result = e [ '<STR_LIT>' ] . get ( '<STR_LIT:result>' ) <EOL> elif et == '<STR_LIT>' : <EOL> workflow . state = '<STR_LIT>' <EOL> workflow . failure_details = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> workflow . failure_reason = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> elif et == '<STR_LIT>' : <EOL> workflow . state == '<STR_LIT>' <EOL> elif et == '<STR_LIT>' : <EOL> workflow . state = '<STR_LIT>' <EOL> workflow . cancellation_cause = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> elif et == '<STR_LIT>' : <EOL> workflow . state == '<STR_LIT>' <EOL> workflow . cancellation_details = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> elif et == '<STR_LIT>' : <EOL> activity_task = Activity ( ) <EOL> id_count += <NUM_LIT:1> <EOL> activity_task . id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> activity_task . type = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT:name>' ] <EOL> activity_task . version = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT:version>' ] <EOL> activity_task . control = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> activity_task . input = e [ '<STR_LIT>' ] . get ( '<STR_LIT:input>' ) <EOL> activity_task . task_list = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] . get ( '<STR_LIT:name>' ) <EOL> activity_task . task_priority = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> activity_task . state = '<STR_LIT>' <EOL> activities [ activity_task . id ] = activity_task <EOL> elif et == '<STR_LIT>' : <EOL> id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> activity_task = activities [ id ] <EOL> activity_task . state = '<STR_LIT>' <EOL> activity_task . schedule_failure_cause = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> elif et == '<STR_LIT>' : <EOL> started_id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> activity_id = history [ started_id - <NUM_LIT:1> ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> activities [ activity_id ] . state = '<STR_LIT>' <EOL> elif et == '<STR_LIT>' : <EOL> started_id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> activity_id = history [ started_id - <NUM_LIT:1> ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> activities [ activity_id ] . state = '<STR_LIT>' <EOL> activities [ activity_id ] . result = e [ '<STR_LIT>' ] . get ( '<STR_LIT:result>' ) <EOL> elif et == '<STR_LIT>' : <EOL> started_id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> activity_id = history [ started_id - <NUM_LIT:1> ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> activities [ activity_id ] . state = '<STR_LIT>' <EOL> activities [ activity_id ] . failure_reason = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> activities [ activity_id ] . failure_details = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> elif et == '<STR_LIT>' : <EOL> started_id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> activity_id = history [ started_id - <NUM_LIT:1> ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> activities [ activity_id ] . state = '<STR_LIT>' <EOL> activities [ activity_id ] . time_out_details = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> activities [ activity_id ] . time_out_type = e [ '<STR_LIT>' ] . get ( <EOL> '<STR_LIT>' ) <EOL> elif et == '<STR_LIT>' : <EOL> started_id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> activity_id = history [ started_id - <NUM_LIT:1> ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> activities [ activity_id ] . state = '<STR_LIT>' <EOL> activities [ activity_id ] . cancellation_details = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> elif et == '<STR_LIT>' : <EOL> pass <EOL> elif et == '<STR_LIT>' : <EOL> pass <EOL> elif et == '<STR_LIT>' : <EOL> signal = Signal ( ) <EOL> signal . input = e [ '<STR_LIT>' ] . get ( '<STR_LIT:input>' ) <EOL> signal . name = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> signals . append ( signal ) <EOL> elif et == '<STR_LIT>' : <EOL> timer = Timer ( ) <EOL> id_count += <NUM_LIT:1> <EOL> timer . id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> timer . control = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> timer . state = '<STR_LIT>' <EOL> timers [ timer . id ] = timer <EOL> elif et == '<STR_LIT>' : <EOL> id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> timers [ id ] . state = '<STR_LIT>' <EOL> timers [ id ] . failure_cause = [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> elif et == '<STR_LIT>' : <EOL> id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> timers [ id ] . state = '<STR_LIT>' <EOL> elif et == '<STR_LIT>' : <EOL> id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> timers [ id ] . state = '<STR_LIT>' <EOL> elif et == '<STR_LIT>' : <EOL> pass <EOL> elif et == '<STR_LIT>' : <EOL> id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> child_workflow = child_workflows [ id ] <EOL> child_workflow . state = '<STR_LIT>' <EOL> elif et == '<STR_LIT>' : <EOL> id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> child_workflow = child_workflows [ id ] <EOL> child_workflow . state = '<STR_LIT>' <EOL> child_workflow . result = e [ '<STR_LIT>' ] [ '<STR_LIT:result>' ] <EOL> elif et == '<STR_LIT>' : <EOL> id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> child_workflow = child_workflows [ id ] <EOL> child_workflow . state = '<STR_LIT>' <EOL> child_workflow . failure_details = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> child_workflow . failure_reason = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> elif et == '<STR_LIT>' : <EOL> id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> child_workflow = child_workflows [ id ] <EOL> child_workflow . continued_execution_run_id = e [ '<STR_LIT>' ] . get ( <EOL> '<STR_LIT>' ) <EOL> wf_type = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> if wf_type : <EOL> child_workflow . parent_workflow_id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> child_workflow . parent_run_id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> child_workflow . run_id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> child_workflow . state = '<STR_LIT>' <EOL> elif et == '<STR_LIT>' : <EOL> id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> child_workflow = child_workflows [ id ] <EOL> child_workflow . state = '<STR_LIT>' <EOL> elif et == '<STR_LIT>' : <EOL> id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> child_workflow = child_workflows [ id ] <EOL> child_workflow . state = '<STR_LIT>' <EOL> elif et == '<STR_LIT>' : <EOL> id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> child_workflow = child_workflows [ id ] <EOL> child_workflow . failure_cause = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> elif et == '<STR_LIT>' : <EOL> child_workflow = Workflow ( ) <EOL> id = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> child_workflows [ id ] = child_workflow <EOL> child_workflow . workflow_id = id <EOL> child_workflow . child_policy = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> child_workflow . control = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> child_workflow . execution_start_to_close_timeout = e [ '<STR_LIT>' ] . get ( <EOL> '<STR_LIT>' ) <EOL> child_workflow . input = e [ '<STR_LIT>' ] . get ( '<STR_LIT:input>' ) <EOL> child_workflow . lambda_role = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> child_workflow . tag_list = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> child_workflow . task_list = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> child_workflow . task_priority = e [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> child_workflow . type = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT:name>' ] <EOL> child_workflow . version = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT:version>' ] <EOL> child_workflow . state = '<STR_LIT>' <EOL> elif et == '<STR_LIT>' : <EOL> marker_name = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> marker_details = e [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> marker = Marker ( name = marker_name , details = marker_details ) <EOL> if marker_name == '<STR_LIT>' : <EOL> cache_markers . append ( marker ) <EOL> else : <EOL> user_markers . append ( marker ) <EOL> swf_decision_context = SwfDecisionContext ( ) <EOL> activities_with_retries = OrderedDict ( activities ) <EOL> json_serializer = JsonSerializer ( ) <EOL> ids_to_delete = list ( ) <EOL> for activity in activities_with_retries . itervalues ( ) : <EOL> if activity . control : <EOL> control_data = json_serializer . deserialize_result ( activity . control ) <EOL> original_task_id = control_data [ '<STR_LIT>' ] <EOL> if activity . id != original_task_id : <EOL> activities_with_retries [ original_task_id ] = activity <EOL> ids_to_delete . append ( activity . id ) <EOL> for id_to_delete in ids_to_delete : <EOL> del activities_with_retries [ id_to_delete ] <EOL> swf_decision_context . activities = activities <EOL> swf_decision_context . activities_iter = iter ( activities_with_retries . itervalues ( ) ) <EOL> swf_decision_context . child_workflows = child_workflows <EOL> swf_decision_context . child_workflows_iter = iter ( child_workflows . itervalues ( ) ) <EOL> swf_decision_context . decision_task = decision_task <EOL> swf_decision_context . signals = signals <EOL> swf_decision_context . signals_iter = iter ( signals ) <EOL> swf_decision_context . timers = timers <EOL> swf_decision_context . timers_iter = iter ( timers . itervalues ( ) ) <EOL> swf_decision_context . swf_history = history <EOL> swf_decision_context . cache_markers = cache_markers <EOL> swf_decision_context . cache_markers_iter = iter ( cache_markers ) <EOL> swf_decision_context . user_markers = user_markers <EOL> swf_decision_context . user_markers_iter = iter ( user_markers ) <EOL> swf_decision_context . workflow = workflow <EOL> swf_decision_context . _id_generator = id_count <EOL> return swf_decision_context <EOL> class LocalDecisionWorker ( object ) : <EOL> """<STR_LIT>""" <EOL> logger = logging . getLogger ( __name__ ) <EOL> def __init__ ( self , decision_function , decision_object = None ) : <EOL> self . _decision_object = decision_object <EOL> self . _decision_function = decision_function <EOL> registry = Registry ( ) <EOL> self . _scanner = venusian . Scanner ( registry = registry , mode = '<STR_LIT>' , caller = '<STR_LIT>' ) <EOL> self . _scanner . parent_decision_worker = decision_function <EOL> self . _scanner . scan ( sys . modules [ decision_function . orig . __module__ ] , categories = ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) ) <EOL> PTimer . is_remote_mode = False <EOL> PMarker . is_remote_mode = False <EOL> def start ( self , * args , ** kwargs ) : <EOL> if self . _decision_object : <EOL> return self . _decision_function ( self . _decision_object , * args , ** kwargs ) <EOL> else : <EOL> return self . _decision_function ( * args , ** kwargs ) <EOL> class Registry ( object ) : <EOL> __slots__ = [ '<STR_LIT>' ] <EOL> def __init__ ( self ) : <EOL> self . registered = dict ( ) <EOL> def add ( self , orig_func , modified_func ) : <EOL> self . registered [ orig_func ] = modified_func </s>
