<s> """<STR_LIT>""" <EOL> import datetime <EOL> import logging <EOL> import math <EOL> import os <EOL> import random <EOL> import sys <EOL> import time <EOL> import traceback <EOL> import zlib <EOL> try : <EOL> import json <EOL> except ImportError : <EOL> import simplejson as json <EOL> from google . appengine . ext import ndb <EOL> from google . appengine import runtime <EOL> from google . appengine . api import datastore_errors <EOL> from google . appengine . api import logservice <EOL> from google . appengine . api import taskqueue <EOL> from google . appengine . ext import db <EOL> from mapreduce import base_handler <EOL> from mapreduce import context <EOL> from mapreduce import errors <EOL> from mapreduce import input_readers <EOL> from mapreduce import map_job_context <EOL> from mapreduce import model <EOL> from mapreduce import operation <EOL> from mapreduce import output_writers <EOL> from mapreduce import parameters <EOL> from mapreduce import shard_life_cycle <EOL> from mapreduce import util <EOL> from mapreduce . api import map_job <EOL> from google . appengine . runtime import apiproxy_errors <EOL> try : <EOL> import cloudstorage <EOL> if hasattr ( cloudstorage , "<STR_LIT>" ) : <EOL> cloudstorage = None <EOL> except ImportError : <EOL> cloudstorage = None <EOL> _TEST_INJECTED_FAULTS = set ( ) <EOL> def _run_task_hook ( hooks , method , task , queue_name ) : <EOL> """<STR_LIT>""" <EOL> if hooks is not None : <EOL> try : <EOL> getattr ( hooks , method ) ( task , queue_name ) <EOL> except NotImplementedError : <EOL> return False <EOL> return True <EOL> return False <EOL> class MapperWorkerCallbackHandler ( base_handler . HugeTaskHandler ) : <EOL> """<STR_LIT>""" <EOL> _TASK_DIRECTIVE = util . _enum ( <EOL> PROCEED_TASK = "<STR_LIT>" , <EOL> RETRY_TASK = "<STR_LIT>" , <EOL> RETRY_SLICE = "<STR_LIT>" , <EOL> DROP_TASK = "<STR_LIT>" , <EOL> RECOVER_SLICE = "<STR_LIT>" , <EOL> RETRY_SHARD = "<STR_LIT>" , <EOL> FAIL_TASK = "<STR_LIT>" , <EOL> ABORT_SHARD = "<STR_LIT>" ) <EOL> def __init__ ( self , * args ) : <EOL> """<STR_LIT>""" <EOL> super ( MapperWorkerCallbackHandler , self ) . __init__ ( * args ) <EOL> self . _time = time . time <EOL> self . slice_context = None <EOL> self . shard_context = None <EOL> def _drop_gracefully ( self ) : <EOL> """<STR_LIT>""" <EOL> shard_id = self . request . headers [ util . _MR_SHARD_ID_TASK_HEADER ] <EOL> mr_id = self . request . headers [ util . _MR_ID_TASK_HEADER ] <EOL> shard_state , mr_state = db . get ( [ <EOL> model . ShardState . get_key_by_shard_id ( shard_id ) , <EOL> model . MapreduceState . get_key_by_job_id ( mr_id ) ] ) <EOL> if shard_state and shard_state . active : <EOL> shard_state . set_for_failure ( ) <EOL> config = util . create_datastore_write_config ( mr_state . mapreduce_spec ) <EOL> shard_state . put ( config = config ) <EOL> def _try_acquire_lease ( self , shard_state , tstate ) : <EOL> """<STR_LIT>""" <EOL> if not shard_state : <EOL> logging . warning ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> tstate . shard_id ) <EOL> return self . _TASK_DIRECTIVE . DROP_TASK <EOL> if not shard_state . active : <EOL> logging . warning ( "<STR_LIT>" <EOL> "<STR_LIT>" , tstate . shard_id ) <EOL> logging . warning ( str ( shard_state ) ) <EOL> return self . _TASK_DIRECTIVE . DROP_TASK <EOL> if shard_state . retries > tstate . retries : <EOL> logging . warning ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> tstate . shard_id , <EOL> tstate . retries ) <EOL> logging . warning ( str ( shard_state ) ) <EOL> return self . _TASK_DIRECTIVE . DROP_TASK <EOL> elif shard_state . retries < tstate . retries : <EOL> logging . warning ( <EOL> "<STR_LIT>" , <EOL> shard_state . shard_id ) <EOL> return self . _TASK_DIRECTIVE . RETRY_TASK <EOL> if shard_state . slice_id > tstate . slice_id : <EOL> logging . warning ( <EOL> "<STR_LIT>" "<STR_LIT>" , <EOL> tstate . shard_id , tstate . slice_id , shard_state . slice_id ) <EOL> return self . _TASK_DIRECTIVE . DROP_TASK <EOL> elif shard_state . slice_id < tstate . slice_id : <EOL> logging . warning ( <EOL> "<STR_LIT>" , <EOL> tstate . shard_id , tstate . slice_id , shard_state . slice_id ) <EOL> return self . _TASK_DIRECTIVE . RETRY_TASK <EOL> if shard_state . slice_start_time : <EOL> countdown = self . _wait_time ( shard_state , <EOL> parameters . _LEASE_DURATION_SEC ) <EOL> if countdown > <NUM_LIT:0> : <EOL> logging . warning ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , tstate . shard_id , tstate . slice_id , <EOL> countdown ) <EOL> time . sleep ( countdown ) <EOL> return self . _TASK_DIRECTIVE . RETRY_TASK <EOL> else : <EOL> if self . _wait_time ( shard_state , <EOL> parameters . _MAX_LEASE_DURATION_SEC ) : <EOL> if not self . _has_old_request_ended ( shard_state ) : <EOL> logging . warning ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , tstate . shard_id , tstate . slice_id , <EOL> shard_state . slice_request_id ) <EOL> return self . _TASK_DIRECTIVE . RETRY_TASK <EOL> else : <EOL> logging . warning ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> tstate . shard_id , tstate . slice_id , <EOL> parameters . _MAX_LEASE_DURATION_SEC ) <EOL> config = util . create_datastore_write_config ( tstate . mapreduce_spec ) <EOL> @ db . transactional ( retries = <NUM_LIT:5> ) <EOL> def _tx ( ) : <EOL> """<STR_LIT>""" <EOL> fresh_state = model . ShardState . get_by_shard_id ( tstate . shard_id ) <EOL> if not fresh_state : <EOL> logging . warning ( "<STR_LIT>" ) <EOL> raise db . Rollback ( ) <EOL> if ( fresh_state . active and <EOL> fresh_state . slice_id == shard_state . slice_id and <EOL> fresh_state . slice_start_time == shard_state . slice_start_time ) : <EOL> shard_state . slice_start_time = datetime . datetime . now ( ) <EOL> shard_state . slice_request_id = os . environ . get ( "<STR_LIT>" ) <EOL> shard_state . acquired_once = True <EOL> shard_state . put ( config = config ) <EOL> return self . _TASK_DIRECTIVE . PROCEED_TASK <EOL> else : <EOL> logging . warning ( <EOL> "<STR_LIT>" , <EOL> tstate . shard_id , tstate . slice_id ) <EOL> time . sleep ( random . randrange ( <NUM_LIT:1> , <NUM_LIT:5> ) ) <EOL> return self . _TASK_DIRECTIVE . RETRY_TASK <EOL> return _tx ( ) <EOL> def _has_old_request_ended ( self , shard_state ) : <EOL> """<STR_LIT>""" <EOL> assert shard_state . slice_start_time is not None <EOL> assert shard_state . slice_request_id is not None <EOL> request_ids = [ shard_state . slice_request_id ] <EOL> logs = None <EOL> try : <EOL> logs = list ( logservice . fetch ( request_ids = request_ids ) ) <EOL> except ( apiproxy_errors . FeatureNotEnabledError , <EOL> apiproxy_errors . CapabilityDisabledError ) as e : <EOL> logging . warning ( "<STR_LIT>" , e ) <EOL> if not logs or not logs [ <NUM_LIT:0> ] . finished : <EOL> return False <EOL> return True <EOL> def _wait_time ( self , shard_state , secs , now = datetime . datetime . now ) : <EOL> """<STR_LIT>""" <EOL> assert shard_state . slice_start_time is not None <EOL> delta = now ( ) - shard_state . slice_start_time <EOL> duration = datetime . timedelta ( seconds = secs ) <EOL> if delta < duration : <EOL> return util . total_seconds ( duration - delta ) <EOL> else : <EOL> return <NUM_LIT:0> <EOL> def _try_free_lease ( self , shard_state , slice_retry = False ) : <EOL> """<STR_LIT>""" <EOL> @ db . transactional <EOL> def _tx ( ) : <EOL> fresh_state = model . ShardState . get_by_shard_id ( shard_state . shard_id ) <EOL> if fresh_state and fresh_state . active : <EOL> fresh_state . slice_start_time = None <EOL> fresh_state . slice_request_id = None <EOL> if slice_retry : <EOL> fresh_state . slice_retries += <NUM_LIT:1> <EOL> fresh_state . put ( ) <EOL> try : <EOL> _tx ( ) <EOL> except Exception , e : <EOL> logging . warning ( e ) <EOL> logging . warning ( <EOL> "<STR_LIT>" , <EOL> shard_state . shard_id ) <EOL> def _maintain_LC ( self , obj , slice_id , last_slice = False , begin_slice = True , <EOL> shard_ctx = None , slice_ctx = None ) : <EOL> """<STR_LIT>""" <EOL> if obj is None or not isinstance ( obj , shard_life_cycle . _ShardLifeCycle ) : <EOL> return <EOL> shard_context = shard_ctx or self . shard_context <EOL> slice_context = slice_ctx or self . slice_context <EOL> if begin_slice : <EOL> if slice_id == <NUM_LIT:0> : <EOL> obj . begin_shard ( shard_context ) <EOL> obj . begin_slice ( slice_context ) <EOL> else : <EOL> obj . end_slice ( slice_context ) <EOL> if last_slice : <EOL> obj . end_shard ( shard_context ) <EOL> def _lc_start_slice ( self , tstate , slice_id ) : <EOL> self . _maintain_LC ( tstate . output_writer , slice_id ) <EOL> self . _maintain_LC ( tstate . input_reader , slice_id ) <EOL> self . _maintain_LC ( tstate . handler , slice_id ) <EOL> def _lc_end_slice ( self , tstate , slice_id , last_slice = False ) : <EOL> self . _maintain_LC ( tstate . handler , slice_id , last_slice = last_slice , <EOL> begin_slice = False ) <EOL> self . _maintain_LC ( tstate . input_reader , slice_id , last_slice = last_slice , <EOL> begin_slice = False ) <EOL> self . _maintain_LC ( tstate . output_writer , slice_id , last_slice = last_slice , <EOL> begin_slice = False ) <EOL> def handle ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _start_time = self . _time ( ) <EOL> shard_id = self . request . headers [ util . _MR_SHARD_ID_TASK_HEADER ] <EOL> mr_id = self . request . headers [ util . _MR_ID_TASK_HEADER ] <EOL> spec = model . MapreduceSpec . _get_mapreduce_spec ( mr_id ) <EOL> shard_state , control = db . get ( [ <EOL> model . ShardState . get_key_by_shard_id ( shard_id ) , <EOL> model . MapreduceControl . get_key_by_job_id ( mr_id ) , <EOL> ] ) <EOL> ctx = context . Context ( spec , shard_state , <EOL> task_retry_count = self . task_retry_count ( ) ) <EOL> context . Context . _set ( ctx ) <EOL> tstate = model . TransientShardState . from_request ( self . request ) <EOL> if shard_state : <EOL> is_this_a_retry = shard_state . acquired_once <EOL> task_directive = self . _try_acquire_lease ( shard_state , tstate ) <EOL> if task_directive in ( self . _TASK_DIRECTIVE . RETRY_TASK , <EOL> self . _TASK_DIRECTIVE . DROP_TASK ) : <EOL> return self . __return ( shard_state , tstate , task_directive ) <EOL> assert task_directive == self . _TASK_DIRECTIVE . PROCEED_TASK <EOL> if control and control . command == model . MapreduceControl . ABORT : <EOL> task_directive = self . _TASK_DIRECTIVE . ABORT_SHARD <EOL> return self . __return ( shard_state , tstate , task_directive ) <EOL> if ( is_this_a_retry and <EOL> parameters . config . TASK_MAX_DATA_PROCESSING_ATTEMPTS <= <NUM_LIT:1> ) : <EOL> task_directive = self . _TASK_DIRECTIVE . RETRY_SHARD <EOL> return self . __return ( shard_state , tstate , task_directive ) <EOL> util . _set_ndb_cache_policy ( ) <EOL> job_config = map_job . JobConfig . _to_map_job_config ( <EOL> spec , <EOL> os . environ . get ( "<STR_LIT>" ) ) <EOL> job_context = map_job_context . JobContext ( job_config ) <EOL> self . shard_context = map_job_context . ShardContext ( job_context , shard_state ) <EOL> self . slice_context = map_job_context . SliceContext ( self . shard_context , <EOL> shard_state , <EOL> tstate ) <EOL> try : <EOL> slice_id = tstate . slice_id <EOL> self . _lc_start_slice ( tstate , slice_id ) <EOL> if shard_state . is_input_finished ( ) : <EOL> self . _lc_end_slice ( tstate , slice_id , last_slice = True ) <EOL> if ( tstate . output_writer and <EOL> isinstance ( tstate . output_writer , output_writers . OutputWriter ) ) : <EOL> tstate . output_writer . finalize ( ctx , shard_state ) <EOL> shard_state . set_for_success ( ) <EOL> return self . __return ( shard_state , tstate , task_directive ) <EOL> if is_this_a_retry : <EOL> task_directive = self . _attempt_slice_recovery ( shard_state , tstate ) <EOL> if task_directive != self . _TASK_DIRECTIVE . PROCEED_TASK : <EOL> return self . __return ( shard_state , tstate , task_directive ) <EOL> last_slice = self . _process_inputs ( <EOL> tstate . input_reader , shard_state , tstate , ctx ) <EOL> self . _lc_end_slice ( tstate , slice_id ) <EOL> ctx . flush ( ) <EOL> if last_slice : <EOL> shard_state . set_input_finished ( ) <EOL> except Exception , e : <EOL> logging . warning ( "<STR_LIT>" , shard_state . shard_id ) <EOL> logging . error ( traceback . format_exc ( ) ) <EOL> if type ( e ) is errors . FailJobError : <EOL> logging . error ( "<STR_LIT>" ) <EOL> task_directive = self . _TASK_DIRECTIVE . FAIL_TASK <EOL> else : <EOL> task_directive = self . _TASK_DIRECTIVE . RETRY_SLICE <EOL> self . __return ( shard_state , tstate , task_directive ) <EOL> def __return ( self , shard_state , tstate , task_directive ) : <EOL> """<STR_LIT>""" <EOL> task_directive = self . _set_state ( shard_state , tstate , task_directive ) <EOL> self . _save_state_and_schedule_next ( shard_state , tstate , task_directive ) <EOL> context . Context . _set ( None ) <EOL> def _process_inputs ( self , <EOL> input_reader , <EOL> shard_state , <EOL> tstate , <EOL> ctx ) : <EOL> """<STR_LIT>""" <EOL> processing_limit = self . _processing_limit ( tstate . mapreduce_spec ) <EOL> if processing_limit == <NUM_LIT:0> : <EOL> return <EOL> finished_shard = True <EOL> iterator = iter ( input_reader ) <EOL> while True : <EOL> try : <EOL> entity = iterator . next ( ) <EOL> except StopIteration : <EOL> break <EOL> if isinstance ( entity , db . Model ) : <EOL> shard_state . last_work_item = repr ( entity . key ( ) ) <EOL> elif isinstance ( entity , ndb . Model ) : <EOL> shard_state . last_work_item = repr ( entity . key ) <EOL> else : <EOL> shard_state . last_work_item = repr ( entity ) [ : <NUM_LIT:100> ] <EOL> processing_limit -= <NUM_LIT:1> <EOL> if not self . _process_datum ( <EOL> entity , input_reader , ctx , tstate ) : <EOL> finished_shard = False <EOL> break <EOL> elif processing_limit == <NUM_LIT:0> : <EOL> finished_shard = False <EOL> break <EOL> self . slice_context . incr ( <EOL> context . COUNTER_MAPPER_WALLTIME_MS , <EOL> int ( ( self . _time ( ) - self . _start_time ) * <NUM_LIT:1000> ) ) <EOL> return finished_shard <EOL> def _process_datum ( self , data , input_reader , ctx , transient_shard_state ) : <EOL> """<STR_LIT>""" <EOL> if data is not input_readers . ALLOW_CHECKPOINT : <EOL> self . slice_context . incr ( context . COUNTER_MAPPER_CALLS ) <EOL> handler = transient_shard_state . handler <EOL> if isinstance ( handler , map_job . Mapper ) : <EOL> handler ( self . slice_context , data ) <EOL> else : <EOL> if input_reader . expand_parameters : <EOL> result = handler ( * data ) <EOL> else : <EOL> result = handler ( data ) <EOL> if util . is_generator ( result ) : <EOL> for output in result : <EOL> if isinstance ( output , operation . Operation ) : <EOL> output ( ctx ) <EOL> else : <EOL> output_writer = transient_shard_state . output_writer <EOL> if not output_writer : <EOL> logging . warning ( <EOL> "<STR_LIT>" , output ) <EOL> else : <EOL> output_writer . write ( output ) <EOL> if self . _time ( ) - self . _start_time >= parameters . config . _SLICE_DURATION_SEC : <EOL> return False <EOL> return True <EOL> def _set_state ( self , shard_state , tstate , task_directive ) : <EOL> """<STR_LIT>""" <EOL> if task_directive in ( self . _TASK_DIRECTIVE . RETRY_TASK , <EOL> self . _TASK_DIRECTIVE . DROP_TASK ) : <EOL> return task_directive <EOL> if task_directive == self . _TASK_DIRECTIVE . ABORT_SHARD : <EOL> shard_state . set_for_abort ( ) <EOL> return task_directive <EOL> if task_directive == self . _TASK_DIRECTIVE . PROCEED_TASK : <EOL> shard_state . advance_for_next_slice ( ) <EOL> tstate . advance_for_next_slice ( ) <EOL> return task_directive <EOL> if task_directive == self . _TASK_DIRECTIVE . RECOVER_SLICE : <EOL> tstate . advance_for_next_slice ( recovery_slice = True ) <EOL> shard_state . advance_for_next_slice ( recovery_slice = True ) <EOL> return task_directive <EOL> if task_directive == self . _TASK_DIRECTIVE . RETRY_SLICE : <EOL> task_directive = self . _attempt_slice_retry ( shard_state , tstate ) <EOL> if task_directive == self . _TASK_DIRECTIVE . RETRY_SHARD : <EOL> task_directive = self . _attempt_shard_retry ( shard_state , tstate ) <EOL> if task_directive == self . _TASK_DIRECTIVE . FAIL_TASK : <EOL> shard_state . set_for_failure ( ) <EOL> return task_directive <EOL> def _save_state_and_schedule_next ( self , shard_state , tstate , task_directive ) : <EOL> """<STR_LIT>""" <EOL> spec = tstate . mapreduce_spec <EOL> if task_directive == self . _TASK_DIRECTIVE . DROP_TASK : <EOL> return <EOL> if task_directive in ( self . _TASK_DIRECTIVE . RETRY_SLICE , <EOL> self . _TASK_DIRECTIVE . RETRY_TASK ) : <EOL> return self . retry_task ( ) <EOL> elif task_directive == self . _TASK_DIRECTIVE . ABORT_SHARD : <EOL> logging . info ( "<STR_LIT>" , <EOL> shard_state . shard_number , shard_state . mapreduce_id ) <EOL> task = None <EOL> elif task_directive == self . _TASK_DIRECTIVE . FAIL_TASK : <EOL> logging . critical ( "<STR_LIT>" , shard_state . shard_id ) <EOL> task = None <EOL> elif task_directive == self . _TASK_DIRECTIVE . RETRY_SHARD : <EOL> logging . warning ( "<STR_LIT>" , <EOL> shard_state . shard_id , <EOL> shard_state . retries + <NUM_LIT:1> ) <EOL> task = self . _state_to_task ( tstate , shard_state ) <EOL> elif task_directive == self . _TASK_DIRECTIVE . RECOVER_SLICE : <EOL> logging . warning ( "<STR_LIT>" , <EOL> shard_state . shard_id , <EOL> shard_state . slice_id ) <EOL> task = self . _state_to_task ( tstate , shard_state ) <EOL> else : <EOL> assert task_directive == self . _TASK_DIRECTIVE . PROCEED_TASK <EOL> countdown = self . _get_countdown_for_next_slice ( spec ) <EOL> task = self . _state_to_task ( tstate , shard_state , countdown = countdown ) <EOL> queue_name = os . environ . get ( "<STR_LIT>" , <EOL> "<STR_LIT:default>" ) <EOL> config = util . create_datastore_write_config ( spec ) <EOL> @ db . transactional ( retries = <NUM_LIT:5> ) <EOL> def _tx ( ) : <EOL> """<STR_LIT>""" <EOL> fresh_shard_state = model . ShardState . get_by_shard_id ( tstate . shard_id ) <EOL> if not fresh_shard_state : <EOL> raise db . Rollback ( ) <EOL> if ( not fresh_shard_state . active or <EOL> "<STR_LIT>" in _TEST_INJECTED_FAULTS ) : <EOL> logging . warning ( "<STR_LIT>" <EOL> "<STR_LIT>" , tstate . shard_id ) <EOL> logging . warning ( "<STR_LIT>" , str ( fresh_shard_state ) ) <EOL> logging . warning ( "<STR_LIT>" , str ( shard_state ) ) <EOL> return <EOL> fresh_shard_state . copy_from ( shard_state ) <EOL> fresh_shard_state . put ( config = config ) <EOL> if fresh_shard_state . active : <EOL> self . _add_task ( task , spec , queue_name ) <EOL> try : <EOL> _tx ( ) <EOL> except ( datastore_errors . Error , <EOL> taskqueue . Error , <EOL> runtime . DeadlineExceededError , <EOL> apiproxy_errors . Error ) , e : <EOL> logging . warning ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> tstate . shard_id , <EOL> tstate . slice_id , <EOL> self . task_retry_count ( ) + <NUM_LIT:1> ) <EOL> self . _try_free_lease ( shard_state ) <EOL> raise e <EOL> def _attempt_slice_recovery ( self , shard_state , tstate ) : <EOL> """<STR_LIT>""" <EOL> mapper_spec = tstate . mapreduce_spec . mapper <EOL> if not ( tstate . output_writer and <EOL> tstate . output_writer . _supports_slice_recovery ( mapper_spec ) ) : <EOL> return self . _TASK_DIRECTIVE . PROCEED_TASK <EOL> tstate . output_writer = tstate . output_writer . _recover ( <EOL> tstate . mapreduce_spec , shard_state . shard_number , <EOL> shard_state . retries + <NUM_LIT:1> ) <EOL> return self . _TASK_DIRECTIVE . RECOVER_SLICE <EOL> def _attempt_shard_retry ( self , shard_state , tstate ) : <EOL> """<STR_LIT>""" <EOL> shard_attempts = shard_state . retries + <NUM_LIT:1> <EOL> if shard_attempts >= parameters . config . SHARD_MAX_ATTEMPTS : <EOL> logging . warning ( <EOL> "<STR_LIT>" , <EOL> shard_attempts , parameters . config . SHARD_MAX_ATTEMPTS ) <EOL> return self . _TASK_DIRECTIVE . FAIL_TASK <EOL> if tstate . output_writer and ( <EOL> not tstate . output_writer . _supports_shard_retry ( tstate ) ) : <EOL> logging . warning ( "<STR_LIT>" , <EOL> tstate . output_writer . __class__ . __name__ ) <EOL> return self . _TASK_DIRECTIVE . FAIL_TASK <EOL> shard_state . reset_for_retry ( ) <EOL> logging . warning ( "<STR_LIT>" , <EOL> shard_state . shard_id , <EOL> shard_state . retries , <EOL> parameters . config . SHARD_MAX_ATTEMPTS ) <EOL> output_writer = None <EOL> if tstate . output_writer : <EOL> output_writer = tstate . output_writer . create ( <EOL> tstate . mapreduce_spec , shard_state . shard_number , shard_attempts + <NUM_LIT:1> ) <EOL> tstate . reset_for_retry ( output_writer ) <EOL> return self . _TASK_DIRECTIVE . RETRY_SHARD <EOL> def _attempt_slice_retry ( self , shard_state , tstate ) : <EOL> """<STR_LIT>""" <EOL> if ( shard_state . slice_retries + <NUM_LIT:1> < <EOL> parameters . config . TASK_MAX_DATA_PROCESSING_ATTEMPTS ) : <EOL> logging . warning ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> tstate . shard_id , <EOL> tstate . slice_id , <EOL> shard_state . slice_retries + <NUM_LIT:1> , <EOL> parameters . config . TASK_MAX_DATA_PROCESSING_ATTEMPTS , <EOL> self . task_retry_count ( ) + <NUM_LIT:1> , <EOL> parameters . config . TASK_MAX_ATTEMPTS ) <EOL> sys . exc_clear ( ) <EOL> self . _try_free_lease ( shard_state , slice_retry = True ) <EOL> return self . _TASK_DIRECTIVE . RETRY_SLICE <EOL> if parameters . config . TASK_MAX_DATA_PROCESSING_ATTEMPTS > <NUM_LIT:0> : <EOL> logging . warning ( "<STR_LIT>" , <EOL> self . task_retry_count ( ) + <NUM_LIT:1> , <EOL> parameters . config . TASK_MAX_DATA_PROCESSING_ATTEMPTS ) <EOL> return self . _TASK_DIRECTIVE . RETRY_SHARD <EOL> @ staticmethod <EOL> def get_task_name ( shard_id , slice_id , retry = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" % ( <EOL> shard_id , slice_id , retry ) <EOL> def _get_countdown_for_next_slice ( self , spec ) : <EOL> """<STR_LIT>""" <EOL> countdown = <NUM_LIT:0> <EOL> if self . _processing_limit ( spec ) != - <NUM_LIT:1> : <EOL> countdown = max ( <EOL> int ( parameters . config . _SLICE_DURATION_SEC - <EOL> ( self . _time ( ) - self . _start_time ) ) , <NUM_LIT:0> ) <EOL> return countdown <EOL> @ classmethod <EOL> def _state_to_task ( cls , <EOL> tstate , <EOL> shard_state , <EOL> eta = None , <EOL> countdown = None ) : <EOL> """<STR_LIT>""" <EOL> base_path = tstate . base_path <EOL> task_name = MapperWorkerCallbackHandler . get_task_name ( <EOL> tstate . shard_id , <EOL> tstate . slice_id , <EOL> tstate . retries ) <EOL> headers = util . _get_task_headers ( tstate . mapreduce_spec . mapreduce_id ) <EOL> headers [ util . _MR_SHARD_ID_TASK_HEADER ] = tstate . shard_id <EOL> worker_task = model . HugeTask ( <EOL> url = base_path + "<STR_LIT>" + tstate . shard_id , <EOL> params = tstate . to_dict ( ) , <EOL> name = task_name , <EOL> eta = eta , <EOL> countdown = countdown , <EOL> parent = shard_state , <EOL> headers = headers ) <EOL> return worker_task <EOL> @ classmethod <EOL> def _add_task ( cls , <EOL> worker_task , <EOL> mapreduce_spec , <EOL> queue_name ) : <EOL> """<STR_LIT>""" <EOL> if not _run_task_hook ( mapreduce_spec . get_hooks ( ) , <EOL> "<STR_LIT>" , <EOL> worker_task , <EOL> queue_name ) : <EOL> try : <EOL> worker_task . add ( queue_name ) <EOL> except ( taskqueue . TombstonedTaskError , <EOL> taskqueue . TaskAlreadyExistsError ) , e : <EOL> logging . warning ( "<STR_LIT>" , <EOL> worker_task . name , <EOL> e . __class__ , <EOL> e ) <EOL> def _processing_limit ( self , spec ) : <EOL> """<STR_LIT>""" <EOL> processing_rate = float ( spec . mapper . params . get ( "<STR_LIT>" , <NUM_LIT:0> ) ) <EOL> slice_processing_limit = - <NUM_LIT:1> <EOL> if processing_rate > <NUM_LIT:0> : <EOL> slice_processing_limit = int ( math . ceil ( <EOL> parameters . config . _SLICE_DURATION_SEC * processing_rate / <EOL> int ( spec . mapper . shard_count ) ) ) <EOL> return slice_processing_limit <EOL> @ classmethod <EOL> def _schedule_slice ( cls , <EOL> shard_state , <EOL> tstate , <EOL> queue_name = None , <EOL> eta = None , <EOL> countdown = None ) : <EOL> """<STR_LIT>""" <EOL> queue_name = queue_name or os . environ . get ( "<STR_LIT>" , <EOL> "<STR_LIT:default>" ) <EOL> task = cls . _state_to_task ( tstate , shard_state , eta , countdown ) <EOL> cls . _add_task ( task , tstate . mapreduce_spec , queue_name ) <EOL> class ControllerCallbackHandler ( base_handler . HugeTaskHandler ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * args ) : <EOL> """<STR_LIT>""" <EOL> super ( ControllerCallbackHandler , self ) . __init__ ( * args ) <EOL> self . _time = time . time <EOL> def _drop_gracefully ( self ) : <EOL> """<STR_LIT>""" <EOL> mr_id = self . request . headers [ util . _MR_ID_TASK_HEADER ] <EOL> state = model . MapreduceState . get_by_job_id ( mr_id ) <EOL> if not state or not state . active : <EOL> return <EOL> state . active = False <EOL> state . result_status = model . MapreduceState . RESULT_FAILED <EOL> config = util . create_datastore_write_config ( state . mapreduce_spec ) <EOL> puts = [ ] <EOL> for ss in model . ShardState . find_all_by_mapreduce_state ( state ) : <EOL> if ss . active : <EOL> ss . set_for_failure ( ) <EOL> puts . append ( ss ) <EOL> if len ( puts ) > model . ShardState . _MAX_STATES_IN_MEMORY : <EOL> db . put ( puts , config = config ) <EOL> puts = [ ] <EOL> db . put ( puts , config = config ) <EOL> db . put ( state , config = config ) <EOL> def handle ( self ) : <EOL> """<STR_LIT>""" <EOL> spec = model . MapreduceSpec . from_json_str ( <EOL> self . request . get ( "<STR_LIT>" ) ) <EOL> state , control = db . get ( [ <EOL> model . MapreduceState . get_key_by_job_id ( spec . mapreduce_id ) , <EOL> model . MapreduceControl . get_key_by_job_id ( spec . mapreduce_id ) , <EOL> ] ) <EOL> if not state : <EOL> logging . warning ( "<STR_LIT>" , <EOL> spec . mapreduce_id ) <EOL> return <EOL> if not state . active : <EOL> logging . warning ( <EOL> "<STR_LIT>" , <EOL> spec . mapreduce_id ) <EOL> self . _clean_up_mr ( spec ) <EOL> return <EOL> shard_states = model . ShardState . find_all_by_mapreduce_state ( state ) <EOL> self . _update_state_from_shard_states ( state , shard_states , control ) <EOL> if state . active : <EOL> ControllerCallbackHandler . reschedule ( <EOL> state , spec , self . serial_id ( ) + <NUM_LIT:1> ) <EOL> def _update_state_from_shard_states ( self , state , shard_states , control ) : <EOL> """<STR_LIT>""" <EOL> state . active_shards , state . aborted_shards , state . failed_shards = <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> <EOL> total_shards = <NUM_LIT:0> <EOL> processed_counts = [ ] <EOL> processed_status = [ ] <EOL> state . counters_map . clear ( ) <EOL> for s in shard_states : <EOL> total_shards += <NUM_LIT:1> <EOL> status = '<STR_LIT>' <EOL> if s . active : <EOL> state . active_shards += <NUM_LIT:1> <EOL> status = '<STR_LIT>' <EOL> if s . result_status == model . ShardState . RESULT_SUCCESS : <EOL> status = '<STR_LIT:success>' <EOL> elif s . result_status == model . ShardState . RESULT_ABORTED : <EOL> state . aborted_shards += <NUM_LIT:1> <EOL> status = '<STR_LIT>' <EOL> elif s . result_status == model . ShardState . RESULT_FAILED : <EOL> state . failed_shards += <NUM_LIT:1> <EOL> status = '<STR_LIT>' <EOL> state . counters_map . add_map ( s . counters_map ) <EOL> processed_counts . append ( s . counters_map . get ( context . COUNTER_MAPPER_CALLS ) ) <EOL> processed_status . append ( status ) <EOL> state . set_processed_counts ( processed_counts , processed_status ) <EOL> state . last_poll_time = datetime . datetime . utcfromtimestamp ( self . _time ( ) ) <EOL> spec = state . mapreduce_spec <EOL> if total_shards != spec . mapper . shard_count : <EOL> logging . error ( "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> total_shards , spec . mapper . shard_count , <EOL> spec . mapreduce_id ) <EOL> model . MapreduceControl . abort ( spec . mapreduce_id ) <EOL> state . active = bool ( state . active_shards ) <EOL> if not control and ( state . failed_shards or state . aborted_shards ) : <EOL> model . MapreduceControl . abort ( spec . mapreduce_id ) <EOL> if not state . active : <EOL> if state . failed_shards or not total_shards : <EOL> state . result_status = model . MapreduceState . RESULT_FAILED <EOL> elif state . aborted_shards : <EOL> state . result_status = model . MapreduceState . RESULT_ABORTED <EOL> else : <EOL> state . result_status = model . MapreduceState . RESULT_SUCCESS <EOL> self . _finalize_outputs ( spec , state ) <EOL> self . _finalize_job ( spec , state ) <EOL> else : <EOL> @ db . transactional ( retries = <NUM_LIT:5> ) <EOL> def _put_state ( ) : <EOL> """<STR_LIT>""" <EOL> fresh_state = model . MapreduceState . get_by_job_id ( spec . mapreduce_id ) <EOL> if not fresh_state . active : <EOL> logging . warning ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , spec . mapreduce_id ) <EOL> return <EOL> config = util . create_datastore_write_config ( spec ) <EOL> state . put ( config = config ) <EOL> _put_state ( ) <EOL> def serial_id ( self ) : <EOL> """<STR_LIT>""" <EOL> return int ( self . request . get ( "<STR_LIT>" ) ) <EOL> @ classmethod <EOL> def _finalize_outputs ( cls , mapreduce_spec , mapreduce_state ) : <EOL> """<STR_LIT>""" <EOL> if ( mapreduce_spec . mapper . output_writer_class ( ) and <EOL> mapreduce_state . result_status == model . MapreduceState . RESULT_SUCCESS ) : <EOL> mapreduce_spec . mapper . output_writer_class ( ) . finalize_job ( mapreduce_state ) <EOL> @ classmethod <EOL> def _finalize_job ( cls , mapreduce_spec , mapreduce_state ) : <EOL> """<STR_LIT>""" <EOL> config = util . create_datastore_write_config ( mapreduce_spec ) <EOL> queue_name = util . get_queue_name ( mapreduce_spec . params . get ( <EOL> model . MapreduceSpec . PARAM_DONE_CALLBACK_QUEUE ) ) <EOL> done_callback = mapreduce_spec . params . get ( <EOL> model . MapreduceSpec . PARAM_DONE_CALLBACK ) <EOL> done_task = None <EOL> if done_callback : <EOL> done_task = taskqueue . Task ( <EOL> url = done_callback , <EOL> headers = util . _get_task_headers ( mapreduce_spec . mapreduce_id , <EOL> util . CALLBACK_MR_ID_TASK_HEADER ) , <EOL> method = mapreduce_spec . params . get ( "<STR_LIT>" , "<STR_LIT:POST>" ) ) <EOL> @ db . transactional ( retries = <NUM_LIT:5> ) <EOL> def _put_state ( ) : <EOL> """<STR_LIT>""" <EOL> fresh_state = model . MapreduceState . get_by_job_id ( <EOL> mapreduce_spec . mapreduce_id ) <EOL> if not fresh_state . active : <EOL> logging . warning ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , mapreduce_spec . mapreduce_id ) <EOL> return <EOL> mapreduce_state . put ( config = config ) <EOL> if done_task and not _run_task_hook ( <EOL> mapreduce_spec . get_hooks ( ) , <EOL> "<STR_LIT>" , <EOL> done_task , <EOL> queue_name ) : <EOL> done_task . add ( queue_name , transactional = True ) <EOL> _put_state ( ) <EOL> logging . info ( "<STR_LIT>" , <EOL> mapreduce_spec . mapreduce_id , mapreduce_state . result_status ) <EOL> cls . _clean_up_mr ( mapreduce_spec ) <EOL> @ classmethod <EOL> def _clean_up_mr ( cls , mapreduce_spec ) : <EOL> FinalizeJobHandler . schedule ( mapreduce_spec ) <EOL> @ staticmethod <EOL> def get_task_name ( mapreduce_spec , serial_id ) : <EOL> """<STR_LIT>""" <EOL> return "<STR_LIT>" % ( <EOL> mapreduce_spec . mapreduce_id , serial_id ) <EOL> @ staticmethod <EOL> def controller_parameters ( mapreduce_spec , serial_id ) : <EOL> """<STR_LIT>""" <EOL> return { "<STR_LIT>" : mapreduce_spec . to_json_str ( ) , <EOL> "<STR_LIT>" : str ( serial_id ) } <EOL> @ classmethod <EOL> def reschedule ( cls , <EOL> mapreduce_state , <EOL> mapreduce_spec , <EOL> serial_id , <EOL> queue_name = None ) : <EOL> """<STR_LIT>""" <EOL> task_name = ControllerCallbackHandler . get_task_name ( <EOL> mapreduce_spec , serial_id ) <EOL> task_params = ControllerCallbackHandler . controller_parameters ( <EOL> mapreduce_spec , serial_id ) <EOL> if not queue_name : <EOL> queue_name = os . environ . get ( "<STR_LIT>" , "<STR_LIT:default>" ) <EOL> controller_callback_task = model . HugeTask ( <EOL> url = ( mapreduce_spec . params [ "<STR_LIT>" ] + "<STR_LIT>" + <EOL> mapreduce_spec . mapreduce_id ) , <EOL> name = task_name , params = task_params , <EOL> countdown = parameters . config . _CONTROLLER_PERIOD_SEC , <EOL> parent = mapreduce_state , <EOL> headers = util . _get_task_headers ( mapreduce_spec . mapreduce_id ) ) <EOL> if not _run_task_hook ( mapreduce_spec . get_hooks ( ) , <EOL> "<STR_LIT>" , <EOL> controller_callback_task , <EOL> queue_name ) : <EOL> try : <EOL> controller_callback_task . add ( queue_name ) <EOL> except ( taskqueue . TombstonedTaskError , <EOL> taskqueue . TaskAlreadyExistsError ) , e : <EOL> logging . warning ( "<STR_LIT>" , <EOL> task_name , task_params , e . __class__ , e ) <EOL> class KickOffJobHandler ( base_handler . TaskQueueHandler ) : <EOL> """<STR_LIT>""" <EOL> _SERIALIZED_INPUT_READERS_KEY = "<STR_LIT>" <EOL> def handle ( self ) : <EOL> """<STR_LIT>""" <EOL> mr_id = self . request . get ( "<STR_LIT>" ) <EOL> logging . info ( "<STR_LIT>" , mr_id ) <EOL> state = model . MapreduceState . get_by_job_id ( mr_id ) <EOL> if not self . _check_mr_state ( state , mr_id ) : <EOL> return <EOL> readers , serialized_readers_entity = self . _get_input_readers ( state ) <EOL> if readers is None : <EOL> logging . warning ( "<STR_LIT>" ) <EOL> state . active = False <EOL> state . result_status = model . MapreduceState . RESULT_SUCCESS <EOL> ControllerCallbackHandler . _finalize_job ( <EOL> state . mapreduce_spec , state ) <EOL> return False <EOL> self . _setup_output_writer ( state ) <EOL> result = self . _save_states ( state , serialized_readers_entity ) <EOL> if result is None : <EOL> readers , _ = self . _get_input_readers ( state ) <EOL> elif not result : <EOL> return <EOL> queue_name = self . request . headers . get ( "<STR_LIT>" ) <EOL> KickOffJobHandler . _schedule_shards ( state . mapreduce_spec , readers , <EOL> queue_name , <EOL> state . mapreduce_spec . params [ "<STR_LIT>" ] , <EOL> state ) <EOL> ControllerCallbackHandler . reschedule ( <EOL> state , state . mapreduce_spec , serial_id = <NUM_LIT:0> , queue_name = queue_name ) <EOL> def _drop_gracefully ( self ) : <EOL> """<STR_LIT>""" <EOL> mr_id = self . request . get ( "<STR_LIT>" ) <EOL> logging . error ( "<STR_LIT>" , mr_id ) <EOL> state = model . MapreduceState . get_by_job_id ( mr_id ) <EOL> if not self . _check_mr_state ( state , mr_id ) : <EOL> return <EOL> config = util . create_datastore_write_config ( state . mapreduce_spec ) <EOL> model . MapreduceControl . abort ( mr_id , config = config ) <EOL> state . active = False <EOL> state . result_status = model . MapreduceState . RESULT_FAILED <EOL> ControllerCallbackHandler . _finalize_job ( state . mapreduce_spec , state ) <EOL> def _get_input_readers ( self , state ) : <EOL> """<STR_LIT>""" <EOL> serialized_input_readers_key = ( self . _SERIALIZED_INPUT_READERS_KEY % <EOL> state . key ( ) . id_or_name ( ) ) <EOL> serialized_input_readers = model . _HugeTaskPayload . get_by_key_name ( <EOL> serialized_input_readers_key , parent = state ) <EOL> input_reader_class = state . mapreduce_spec . mapper . input_reader_class ( ) <EOL> split_param = state . mapreduce_spec . mapper <EOL> if issubclass ( input_reader_class , map_job . InputReader ) : <EOL> split_param = map_job . JobConfig . _to_map_job_config ( <EOL> state . mapreduce_spec , <EOL> os . environ . get ( "<STR_LIT>" ) ) <EOL> if serialized_input_readers is None : <EOL> readers = input_reader_class . split_input ( split_param ) <EOL> else : <EOL> readers = [ input_reader_class . from_json_str ( _json ) for _json in <EOL> json . loads ( zlib . decompress ( <EOL> serialized_input_readers . payload ) ) ] <EOL> if not readers : <EOL> return None , None <EOL> state . mapreduce_spec . mapper . shard_count = len ( readers ) <EOL> state . active_shards = len ( readers ) <EOL> if serialized_input_readers is None : <EOL> serialized_input_readers = model . _HugeTaskPayload ( <EOL> key_name = serialized_input_readers_key , parent = state ) <EOL> readers_json_str = [ i . to_json_str ( ) for i in readers ] <EOL> serialized_input_readers . payload = zlib . compress ( json . dumps ( <EOL> readers_json_str ) ) <EOL> return readers , serialized_input_readers <EOL> def _setup_output_writer ( self , state ) : <EOL> if not state . writer_state : <EOL> output_writer_class = state . mapreduce_spec . mapper . output_writer_class ( ) <EOL> if output_writer_class : <EOL> output_writer_class . init_job ( state ) <EOL> @ db . transactional <EOL> def _save_states ( self , state , serialized_readers_entity ) : <EOL> """<STR_LIT>""" <EOL> mr_id = state . key ( ) . id_or_name ( ) <EOL> fresh_state = model . MapreduceState . get_by_job_id ( mr_id ) <EOL> if not self . _check_mr_state ( fresh_state , mr_id ) : <EOL> return False <EOL> if fresh_state . active_shards != <NUM_LIT:0> : <EOL> logging . warning ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , mr_id ) <EOL> return None <EOL> config = util . create_datastore_write_config ( state . mapreduce_spec ) <EOL> db . put ( [ state , serialized_readers_entity ] , config = config ) <EOL> return True <EOL> @ classmethod <EOL> def _schedule_shards ( cls , <EOL> spec , <EOL> readers , <EOL> queue_name , <EOL> base_path , <EOL> mr_state ) : <EOL> """<STR_LIT>""" <EOL> shard_states = [ ] <EOL> for shard_number , input_reader in enumerate ( readers ) : <EOL> shard_state = model . ShardState . create_new ( spec . mapreduce_id , shard_number ) <EOL> shard_state . shard_description = str ( input_reader ) <EOL> shard_states . append ( shard_state ) <EOL> existing_shard_states = db . get ( shard . key ( ) for shard in shard_states ) <EOL> existing_shard_keys = set ( shard . key ( ) for shard in existing_shard_states <EOL> if shard is not None ) <EOL> db . put ( ( shard for shard in shard_states <EOL> if shard . key ( ) not in existing_shard_keys ) , <EOL> config = util . create_datastore_write_config ( spec ) ) <EOL> writer_class = spec . mapper . output_writer_class ( ) <EOL> writers = [ None ] * len ( readers ) <EOL> if writer_class : <EOL> for shard_number , shard_state in enumerate ( shard_states ) : <EOL> writers [ shard_number ] = writer_class . create ( <EOL> mr_state . mapreduce_spec , <EOL> shard_state . shard_number , shard_state . retries + <NUM_LIT:1> , <EOL> mr_state . writer_state ) <EOL> for shard_number , ( input_reader , output_writer ) in enumerate ( <EOL> zip ( readers , writers ) ) : <EOL> shard_id = model . ShardState . shard_id_from_number ( <EOL> spec . mapreduce_id , shard_number ) <EOL> task = MapperWorkerCallbackHandler . _state_to_task ( <EOL> model . TransientShardState ( <EOL> base_path , spec , shard_id , <NUM_LIT:0> , input_reader , input_reader , <EOL> output_writer = output_writer , <EOL> handler = spec . mapper . handler ) , <EOL> shard_states [ shard_number ] ) <EOL> MapperWorkerCallbackHandler . _add_task ( task , <EOL> spec , <EOL> queue_name ) <EOL> @ classmethod <EOL> def _check_mr_state ( cls , state , mr_id ) : <EOL> """<STR_LIT>""" <EOL> if state is None : <EOL> logging . warning ( <EOL> "<STR_LIT>" , <EOL> mr_id ) <EOL> return False <EOL> if not state . active : <EOL> logging . warning ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , mr_id ) <EOL> return False <EOL> return True <EOL> class StartJobHandler ( base_handler . PostJsonHandler ) : <EOL> """<STR_LIT>""" <EOL> def handle ( self ) : <EOL> """<STR_LIT>""" <EOL> mapreduce_name = self . _get_required_param ( "<STR_LIT:name>" ) <EOL> mapper_input_reader_spec = self . _get_required_param ( "<STR_LIT>" ) <EOL> mapper_handler_spec = self . _get_required_param ( "<STR_LIT>" ) <EOL> mapper_output_writer_spec = self . request . get ( "<STR_LIT>" ) <EOL> mapper_params = self . _get_params ( <EOL> "<STR_LIT>" , "<STR_LIT>" ) <EOL> params = self . _get_params ( <EOL> "<STR_LIT>" , "<STR_LIT>" ) <EOL> mr_params = map_job . JobConfig . _get_default_mr_params ( ) <EOL> mr_params . update ( params ) <EOL> if "<STR_LIT>" in mapper_params : <EOL> mr_params [ "<STR_LIT>" ] = mapper_params [ "<STR_LIT>" ] <EOL> mapper_params [ "<STR_LIT>" ] = int ( mapper_params . get ( <EOL> "<STR_LIT>" ) or parameters . config . PROCESSING_RATE_PER_SEC ) <EOL> mapper_spec = model . MapperSpec ( <EOL> mapper_handler_spec , <EOL> mapper_input_reader_spec , <EOL> mapper_params , <EOL> int ( mapper_params . get ( "<STR_LIT>" , parameters . config . SHARD_COUNT ) ) , <EOL> output_writer_spec = mapper_output_writer_spec ) <EOL> mapreduce_id = self . _start_map ( <EOL> mapreduce_name , <EOL> mapper_spec , <EOL> mr_params , <EOL> queue_name = mr_params [ "<STR_LIT>" ] , <EOL> _app = mapper_params . get ( "<STR_LIT>" ) ) <EOL> self . json_response [ "<STR_LIT>" ] = mapreduce_id <EOL> def _get_params ( self , validator_parameter , name_prefix ) : <EOL> """<STR_LIT>""" <EOL> params_validator = self . request . get ( validator_parameter ) <EOL> user_params = { } <EOL> for key in self . request . arguments ( ) : <EOL> if key . startswith ( name_prefix ) : <EOL> values = self . request . get_all ( key ) <EOL> adjusted_key = key [ len ( name_prefix ) : ] <EOL> if len ( values ) == <NUM_LIT:1> : <EOL> user_params [ adjusted_key ] = values [ <NUM_LIT:0> ] <EOL> else : <EOL> user_params [ adjusted_key ] = values <EOL> if params_validator : <EOL> resolved_validator = util . for_name ( params_validator ) <EOL> resolved_validator ( user_params ) <EOL> return user_params <EOL> def _get_required_param ( self , param_name ) : <EOL> """<STR_LIT>""" <EOL> value = self . request . get ( param_name ) <EOL> if not value : <EOL> raise errors . NotEnoughArgumentsError ( param_name + "<STR_LIT>" ) <EOL> return value <EOL> @ classmethod <EOL> def _start_map ( cls , <EOL> name , <EOL> mapper_spec , <EOL> mapreduce_params , <EOL> queue_name , <EOL> eta = None , <EOL> countdown = None , <EOL> hooks_class_name = None , <EOL> _app = None , <EOL> in_xg_transaction = False ) : <EOL> """<STR_LIT>""" <EOL> mapper_input_reader_class = mapper_spec . input_reader_class ( ) <EOL> mapper_input_reader_class . validate ( mapper_spec ) <EOL> mapper_output_writer_class = mapper_spec . output_writer_class ( ) <EOL> if mapper_output_writer_class : <EOL> mapper_output_writer_class . validate ( mapper_spec ) <EOL> mapreduce_id = model . MapreduceState . new_mapreduce_id ( ) <EOL> mapreduce_spec = model . MapreduceSpec ( <EOL> name , <EOL> mapreduce_id , <EOL> mapper_spec . to_json ( ) , <EOL> mapreduce_params , <EOL> hooks_class_name ) <EOL> ctx = context . Context ( mapreduce_spec , None ) <EOL> context . Context . _set ( ctx ) <EOL> try : <EOL> mapper_spec . handler <EOL> finally : <EOL> context . Context . _set ( None ) <EOL> if in_xg_transaction : <EOL> propagation = db . MANDATORY <EOL> else : <EOL> propagation = db . INDEPENDENT <EOL> @ db . transactional ( propagation = propagation ) <EOL> def _txn ( ) : <EOL> cls . _create_and_save_state ( mapreduce_spec , _app ) <EOL> cls . _add_kickoff_task ( mapreduce_params [ "<STR_LIT>" ] , mapreduce_spec , eta , <EOL> countdown , queue_name ) <EOL> _txn ( ) <EOL> return mapreduce_id <EOL> @ classmethod <EOL> def _create_and_save_state ( cls , mapreduce_spec , _app ) : <EOL> """<STR_LIT>""" <EOL> state = model . MapreduceState . create_new ( mapreduce_spec . mapreduce_id ) <EOL> state . mapreduce_spec = mapreduce_spec <EOL> state . active = True <EOL> state . active_shards = <NUM_LIT:0> <EOL> if _app : <EOL> state . app_id = _app <EOL> config = util . create_datastore_write_config ( mapreduce_spec ) <EOL> state . put ( config = config ) <EOL> return state <EOL> @ classmethod <EOL> def _add_kickoff_task ( cls , <EOL> base_path , <EOL> mapreduce_spec , <EOL> eta , <EOL> countdown , <EOL> queue_name ) : <EOL> """<STR_LIT>""" <EOL> params = { "<STR_LIT>" : mapreduce_spec . mapreduce_id } <EOL> kickoff_task = taskqueue . Task ( <EOL> url = base_path + "<STR_LIT>" + mapreduce_spec . mapreduce_id , <EOL> headers = util . _get_task_headers ( mapreduce_spec . mapreduce_id ) , <EOL> params = params , <EOL> eta = eta , <EOL> countdown = countdown ) <EOL> hooks = mapreduce_spec . get_hooks ( ) <EOL> if hooks is not None : <EOL> try : <EOL> hooks . enqueue_kickoff_task ( kickoff_task , queue_name ) <EOL> return <EOL> except NotImplementedError : <EOL> pass <EOL> kickoff_task . add ( queue_name , transactional = True ) <EOL> class FinalizeJobHandler ( base_handler . TaskQueueHandler ) : <EOL> """<STR_LIT>""" <EOL> def handle ( self ) : <EOL> mapreduce_id = self . request . get ( "<STR_LIT>" ) <EOL> mapreduce_state = model . MapreduceState . get_by_job_id ( mapreduce_id ) <EOL> if mapreduce_state : <EOL> config = ( <EOL> util . create_datastore_write_config ( mapreduce_state . mapreduce_spec ) ) <EOL> keys = [ model . MapreduceControl . get_key_by_job_id ( mapreduce_id ) ] <EOL> for ss in model . ShardState . find_all_by_mapreduce_state ( mapreduce_state ) : <EOL> keys . extend ( list ( <EOL> model . _HugeTaskPayload . all ( ) . ancestor ( ss ) . run ( keys_only = True ) ) ) <EOL> keys . extend ( list ( model . _HugeTaskPayload . all ( ) . ancestor ( <EOL> mapreduce_state ) . run ( keys_only = True ) ) ) <EOL> db . delete ( keys , config = config ) <EOL> @ classmethod <EOL> def schedule ( cls , mapreduce_spec ) : <EOL> """<STR_LIT>""" <EOL> task_name = mapreduce_spec . mapreduce_id + "<STR_LIT>" <EOL> finalize_task = taskqueue . Task ( <EOL> name = task_name , <EOL> url = ( mapreduce_spec . params [ "<STR_LIT>" ] + "<STR_LIT>" + <EOL> mapreduce_spec . mapreduce_id ) , <EOL> params = { "<STR_LIT>" : mapreduce_spec . mapreduce_id } , <EOL> headers = util . _get_task_headers ( mapreduce_spec . mapreduce_id ) ) <EOL> queue_name = util . get_queue_name ( None ) <EOL> if not _run_task_hook ( mapreduce_spec . get_hooks ( ) , <EOL> "<STR_LIT>" , <EOL> finalize_task , <EOL> queue_name ) : <EOL> try : <EOL> finalize_task . add ( queue_name ) <EOL> except ( taskqueue . TombstonedTaskError , <EOL> taskqueue . TaskAlreadyExistsError ) , e : <EOL> logging . warning ( "<STR_LIT>" , <EOL> task_name , e . __class__ , e ) <EOL> class CleanUpJobHandler ( base_handler . PostJsonHandler ) : <EOL> """<STR_LIT>""" <EOL> def handle ( self ) : <EOL> mapreduce_id = self . request . get ( "<STR_LIT>" ) <EOL> mapreduce_state = model . MapreduceState . get_by_job_id ( mapreduce_id ) <EOL> if mapreduce_state : <EOL> shard_keys = model . ShardState . calculate_keys_by_mapreduce_state ( <EOL> mapreduce_state ) <EOL> db . delete ( shard_keys ) <EOL> db . delete ( mapreduce_state ) <EOL> self . json_response [ "<STR_LIT:status>" ] = ( "<STR_LIT>" % <EOL> mapreduce_id ) <EOL> class AbortJobHandler ( base_handler . PostJsonHandler ) : <EOL> """<STR_LIT>""" <EOL> def handle ( self ) : <EOL> model . MapreduceControl . abort ( self . request . get ( "<STR_LIT>" ) ) <EOL> self . json_response [ "<STR_LIT:status>" ] = "<STR_LIT>" </s>
