<s> import six <EOL> from six . moves . urllib . parse import unquote , quote <EOL> import collections <EOL> import itertools <EOL> import json <EOL> import mimetypes <EOL> import time <EOL> import math <EOL> import random <EOL> from hashlib import md5 <EOL> from swift import gettext_ as _ <EOL> from greenlet import GreenletExit <EOL> from eventlet import GreenPile <EOL> from eventlet . queue import Queue <EOL> from eventlet . timeout import Timeout <EOL> from swift . common . utils import ( <EOL> clean_content_type , config_true_value , ContextPool , csv_append , <EOL> GreenAsyncPile , GreenthreadSafeIterator , Timestamp , <EOL> normalize_delete_at_timestamp , public , get_expirer_container , <EOL> document_iters_to_http_response_body , parse_content_range , <EOL> quorum_size , reiterate , close_if_possible ) <EOL> from swift . common . bufferedhttp import http_connect <EOL> from swift . common . constraints import check_metadata , check_object_creation , check_copy_from_header , check_destination_header , check_account_format <EOL> from swift . common import constraints <EOL> from swift . common . exceptions import ChunkReadTimeout , ChunkWriteTimeout , ConnectionTimeout , ResponseTimeout , InsufficientStorage , FooterNotSupported , MultiphasePUTNotSupported , PutterConnectError , ChunkReadError <EOL> from swift . common . header_key_dict import HeaderKeyDict <EOL> from swift . common . http import ( <EOL> is_informational , is_success , is_client_error , is_server_error , <EOL> is_redirection , HTTP_CONTINUE , HTTP_CREATED , HTTP_MULTIPLE_CHOICES , <EOL> HTTP_INTERNAL_SERVER_ERROR , HTTP_SERVICE_UNAVAILABLE , <EOL> HTTP_INSUFFICIENT_STORAGE , HTTP_PRECONDITION_FAILED , HTTP_CONFLICT , <EOL> HTTP_UNPROCESSABLE_ENTITY , HTTP_REQUESTED_RANGE_NOT_SATISFIABLE ) <EOL> from swift . common . storage_policy import ( POLICIES , REPL_POLICY , EC_POLICY , <EOL> ECDriverError , PolicyError ) <EOL> from swift . proxy . controllers . base import Controller , delay_denial , cors_validation , ResumingGetter <EOL> from swift . common . swob import HTTPAccepted , HTTPBadRequest , HTTPNotFound , HTTPPreconditionFailed , HTTPRequestEntityTooLarge , HTTPRequestTimeout , HTTPServerError , HTTPServiceUnavailable , Request , HTTPClientDisconnect , HTTPUnprocessableEntity , Response , HTTPException , HTTPRequestedRangeNotSatisfiable , Range , HTTPInternalServerError <EOL> from swift . common . request_helpers import is_sys_or_user_meta , is_sys_meta , remove_items , copy_header_subset <EOL> def copy_headers_into ( from_r , to_r ) : <EOL> """<STR_LIT>""" <EOL> pass_headers = [ '<STR_LIT>' ] <EOL> for k , v in from_r . headers . items ( ) : <EOL> if is_sys_or_user_meta ( '<STR_LIT:object>' , k ) or k . lower ( ) in pass_headers : <EOL> to_r . headers [ k ] = v <EOL> def check_content_type ( req ) : <EOL> if not req . environ . get ( '<STR_LIT>' ) and '<STR_LIT:;>' in req . headers . get ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> for param in req . headers [ '<STR_LIT>' ] . split ( '<STR_LIT:;>' ) [ <NUM_LIT:1> : ] : <EOL> if param . lstrip ( ) . startswith ( '<STR_LIT>' ) : <EOL> return HTTPBadRequest ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> return None <EOL> class ObjectControllerRouter ( object ) : <EOL> policy_type_to_controller_map = { } <EOL> @ classmethod <EOL> def register ( cls , policy_type ) : <EOL> """<STR_LIT>""" <EOL> def register_wrapper ( controller_cls ) : <EOL> if policy_type in cls . policy_type_to_controller_map : <EOL> raise PolicyError ( <EOL> '<STR_LIT>' % ( <EOL> cls . policy_type_to_controller_map [ policy_type ] , <EOL> policy_type ) ) <EOL> cls . policy_type_to_controller_map [ policy_type ] = controller_cls <EOL> controller_cls . policy_type = policy_type <EOL> return controller_cls <EOL> return register_wrapper <EOL> def __init__ ( self ) : <EOL> self . policy_to_controller_cls = { } <EOL> for policy in POLICIES : <EOL> self . policy_to_controller_cls [ policy ] = self . policy_type_to_controller_map [ policy . policy_type ] <EOL> def __getitem__ ( self , policy ) : <EOL> return self . policy_to_controller_cls [ policy ] <EOL> class BaseObjectController ( Controller ) : <EOL> """<STR_LIT>""" <EOL> server_type = '<STR_LIT>' <EOL> def __init__ ( self , app , account_name , container_name , object_name , <EOL> ** kwargs ) : <EOL> Controller . __init__ ( self , app ) <EOL> self . account_name = unquote ( account_name ) <EOL> self . container_name = unquote ( container_name ) <EOL> self . object_name = unquote ( object_name ) <EOL> def iter_nodes_local_first ( self , ring , partition ) : <EOL> """<STR_LIT>""" <EOL> is_local = self . app . write_affinity_is_local_fn <EOL> if is_local is None : <EOL> return self . app . iter_nodes ( ring , partition ) <EOL> primary_nodes = ring . get_part_nodes ( partition ) <EOL> num_locals = self . app . write_affinity_node_count ( len ( primary_nodes ) ) <EOL> all_nodes = itertools . chain ( primary_nodes , <EOL> ring . get_more_nodes ( partition ) ) <EOL> first_n_local_nodes = list ( itertools . islice ( <EOL> six . moves . filter ( is_local , all_nodes ) , num_locals ) ) <EOL> all_nodes = itertools . chain ( primary_nodes , <EOL> ring . get_more_nodes ( partition ) ) <EOL> local_first_node_iter = itertools . chain ( <EOL> first_n_local_nodes , <EOL> six . moves . filter ( lambda node : node not in first_n_local_nodes , <EOL> all_nodes ) ) <EOL> return self . app . iter_nodes ( <EOL> ring , partition , node_iter = local_first_node_iter ) <EOL> def GETorHEAD ( self , req ) : <EOL> """<STR_LIT>""" <EOL> container_info = self . container_info ( <EOL> self . account_name , self . container_name , req ) <EOL> req . acl = container_info [ '<STR_LIT>' ] <EOL> policy_index = req . headers . get ( '<STR_LIT>' , <EOL> container_info [ '<STR_LIT>' ] ) <EOL> policy = POLICIES . get_by_index ( policy_index ) <EOL> obj_ring = self . app . get_object_ring ( policy_index ) <EOL> req . headers [ '<STR_LIT>' ] = policy_index <EOL> if '<STR_LIT>' in req . environ : <EOL> aresp = req . environ [ '<STR_LIT>' ] ( req ) <EOL> if aresp : <EOL> return aresp <EOL> partition = obj_ring . get_part ( <EOL> self . account_name , self . container_name , self . object_name ) <EOL> node_iter = self . app . iter_nodes ( obj_ring , partition ) <EOL> resp = self . _reroute ( policy ) . _get_or_head_response ( <EOL> req , node_iter , partition , policy ) <EOL> if '<STR_LIT:;>' in resp . headers . get ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> resp . content_type = clean_content_type ( <EOL> resp . headers [ '<STR_LIT>' ] ) <EOL> return resp <EOL> @ public <EOL> @ cors_validation <EOL> @ delay_denial <EOL> def GET ( self , req ) : <EOL> """<STR_LIT>""" <EOL> return self . GETorHEAD ( req ) <EOL> @ public <EOL> @ cors_validation <EOL> @ delay_denial <EOL> def HEAD ( self , req ) : <EOL> """<STR_LIT>""" <EOL> return self . GETorHEAD ( req ) <EOL> @ public <EOL> @ cors_validation <EOL> @ delay_denial <EOL> def POST ( self , req ) : <EOL> """<STR_LIT>""" <EOL> if self . app . object_post_as_copy : <EOL> req . method = '<STR_LIT>' <EOL> req . path_info = '<STR_LIT>' % ( <EOL> self . account_name , self . container_name , self . object_name ) <EOL> req . headers [ '<STR_LIT>' ] = <NUM_LIT:0> <EOL> req . headers [ '<STR_LIT>' ] = quote ( '<STR_LIT>' % ( self . container_name , <EOL> self . object_name ) ) <EOL> req . environ [ '<STR_LIT>' ] = True <EOL> req . environ [ '<STR_LIT>' ] = True <EOL> resp = self . PUT ( req ) <EOL> if resp . status_int != HTTP_CREATED : <EOL> return resp <EOL> return HTTPAccepted ( request = req ) <EOL> else : <EOL> error_response = check_metadata ( req , '<STR_LIT:object>' ) <EOL> if error_response : <EOL> return error_response <EOL> container_info = self . container_info ( <EOL> self . account_name , self . container_name , req ) <EOL> container_partition = container_info [ '<STR_LIT>' ] <EOL> containers = container_info [ '<STR_LIT>' ] <EOL> req . acl = container_info [ '<STR_LIT>' ] <EOL> if '<STR_LIT>' in req . environ : <EOL> aresp = req . environ [ '<STR_LIT>' ] ( req ) <EOL> if aresp : <EOL> return aresp <EOL> if not containers : <EOL> return HTTPNotFound ( request = req ) <EOL> req , delete_at_container , delete_at_part , delete_at_nodes = self . _config_obj_expiration ( req ) <EOL> policy_index = req . headers . get ( '<STR_LIT>' , <EOL> container_info [ '<STR_LIT>' ] ) <EOL> obj_ring = self . app . get_object_ring ( policy_index ) <EOL> req . headers [ '<STR_LIT>' ] = policy_index <EOL> partition , nodes = obj_ring . get_nodes ( <EOL> self . account_name , self . container_name , self . object_name ) <EOL> req . headers [ '<STR_LIT>' ] = Timestamp ( time . time ( ) ) . internal <EOL> headers = self . _backend_requests ( <EOL> req , len ( nodes ) , container_partition , containers , <EOL> delete_at_container , delete_at_part , delete_at_nodes ) <EOL> return self . _post_object ( req , obj_ring , partition , headers ) <EOL> def _backend_requests ( self , req , n_outgoing , <EOL> container_partition , containers , <EOL> delete_at_container = None , delete_at_partition = None , <EOL> delete_at_nodes = None ) : <EOL> policy_index = req . headers [ '<STR_LIT>' ] <EOL> policy = POLICIES . get_by_index ( policy_index ) <EOL> headers = [ self . generate_request_headers ( req , additional = req . headers ) <EOL> for _junk in range ( n_outgoing ) ] <EOL> def set_container_update ( index , container ) : <EOL> headers [ index ] [ '<STR_LIT>' ] = container_partition <EOL> headers [ index ] [ '<STR_LIT>' ] = csv_append ( <EOL> headers [ index ] . get ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' % container ) <EOL> headers [ index ] [ '<STR_LIT>' ] = csv_append ( <EOL> headers [ index ] . get ( '<STR_LIT>' ) , <EOL> container [ '<STR_LIT>' ] ) <EOL> for i , container in enumerate ( containers ) : <EOL> i = i % len ( headers ) <EOL> set_container_update ( i , container ) <EOL> n_updates_needed = min ( policy . quorum + <NUM_LIT:1> , n_outgoing ) <EOL> container_iter = itertools . cycle ( containers ) <EOL> existing_updates = len ( containers ) <EOL> while existing_updates < n_updates_needed : <EOL> set_container_update ( existing_updates , next ( container_iter ) ) <EOL> existing_updates += <NUM_LIT:1> <EOL> for i , node in enumerate ( delete_at_nodes or [ ] ) : <EOL> i = i % len ( headers ) <EOL> headers [ i ] [ '<STR_LIT>' ] = delete_at_container <EOL> headers [ i ] [ '<STR_LIT>' ] = delete_at_partition <EOL> headers [ i ] [ '<STR_LIT>' ] = csv_append ( <EOL> headers [ i ] . get ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' % node ) <EOL> headers [ i ] [ '<STR_LIT>' ] = csv_append ( <EOL> headers [ i ] . get ( '<STR_LIT>' ) , <EOL> node [ '<STR_LIT>' ] ) <EOL> return headers <EOL> def _await_response ( self , conn , ** kwargs ) : <EOL> with Timeout ( self . app . node_timeout ) : <EOL> if conn . resp : <EOL> return conn . resp <EOL> else : <EOL> return conn . getresponse ( ) <EOL> def _get_conn_response ( self , conn , req , logger_thread_locals , ** kwargs ) : <EOL> self . app . logger . thread_locals = logger_thread_locals <EOL> try : <EOL> resp = self . _await_response ( conn , ** kwargs ) <EOL> return ( conn , resp ) <EOL> except ( Exception , Timeout ) : <EOL> self . app . exception_occurred ( <EOL> conn . node , _ ( '<STR_LIT>' ) , <EOL> _ ( '<STR_LIT>' ) % req . path ) <EOL> return ( None , None ) <EOL> def _get_put_responses ( self , req , conns , nodes , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> statuses = [ ] <EOL> reasons = [ ] <EOL> bodies = [ ] <EOL> etags = set ( ) <EOL> pile = GreenAsyncPile ( len ( conns ) ) <EOL> for conn in conns : <EOL> pile . spawn ( self . _get_conn_response , conn , <EOL> req , self . app . logger . thread_locals ) <EOL> def _handle_response ( conn , response ) : <EOL> statuses . append ( response . status ) <EOL> reasons . append ( response . reason ) <EOL> bodies . append ( response . read ( ) ) <EOL> if response . status == HTTP_INSUFFICIENT_STORAGE : <EOL> self . app . error_limit ( conn . node , <EOL> _ ( '<STR_LIT>' ) ) <EOL> elif response . status >= HTTP_INTERNAL_SERVER_ERROR : <EOL> self . app . error_occurred ( <EOL> conn . node , <EOL> _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) % <EOL> { '<STR_LIT:status>' : response . status , <EOL> '<STR_LIT:body>' : bodies [ - <NUM_LIT:1> ] [ : <NUM_LIT> ] , '<STR_LIT:path>' : req . path } ) <EOL> elif is_success ( response . status ) : <EOL> etags . add ( response . getheader ( '<STR_LIT>' ) . strip ( '<STR_LIT:">' ) ) <EOL> for ( conn , response ) in pile : <EOL> if response : <EOL> _handle_response ( conn , response ) <EOL> if self . have_quorum ( statuses , len ( nodes ) ) : <EOL> break <EOL> finished_quickly = pile . waitall ( self . app . post_quorum_timeout ) <EOL> for ( conn , response ) in finished_quickly : <EOL> if response : <EOL> _handle_response ( conn , response ) <EOL> while len ( statuses ) < len ( nodes ) : <EOL> statuses . append ( HTTP_SERVICE_UNAVAILABLE ) <EOL> reasons . append ( '<STR_LIT>' ) <EOL> bodies . append ( '<STR_LIT>' ) <EOL> return statuses , reasons , bodies , etags <EOL> def _config_obj_expiration ( self , req ) : <EOL> delete_at_container = None <EOL> delete_at_part = None <EOL> delete_at_nodes = None <EOL> req = constraints . check_delete_headers ( req ) <EOL> if '<STR_LIT>' in req . headers : <EOL> x_delete_at = int ( normalize_delete_at_timestamp ( <EOL> int ( req . headers [ '<STR_LIT>' ] ) ) ) <EOL> req . environ . setdefault ( '<STR_LIT>' , [ ] ) . append ( <EOL> '<STR_LIT>' % x_delete_at ) <EOL> delete_at_container = get_expirer_container ( <EOL> x_delete_at , self . app . expiring_objects_container_divisor , <EOL> self . account_name , self . container_name , self . object_name ) <EOL> delete_at_part , delete_at_nodes = self . app . container_ring . get_nodes ( <EOL> self . app . expiring_objects_account , delete_at_container ) <EOL> return req , delete_at_container , delete_at_part , delete_at_nodes <EOL> def _handle_copy_request ( self , req ) : <EOL> """<STR_LIT>""" <EOL> if req . environ . get ( '<STR_LIT>' , req . method ) != '<STR_LIT:POST>' : <EOL> req . environ . setdefault ( '<STR_LIT>' , [ ] ) . append ( <EOL> '<STR_LIT>' % req . headers [ '<STR_LIT>' ] ) <EOL> ver , acct , _rest = req . split_path ( <NUM_LIT:2> , <NUM_LIT:3> , True ) <EOL> src_account_name = req . headers . get ( '<STR_LIT>' , None ) <EOL> if src_account_name : <EOL> src_account_name = check_account_format ( req , src_account_name ) <EOL> else : <EOL> src_account_name = acct <EOL> src_container_name , src_obj_name = check_copy_from_header ( req ) <EOL> source_header = '<STR_LIT>' % ( <EOL> ver , src_account_name , src_container_name , src_obj_name ) <EOL> source_req = req . copy_get ( ) <EOL> source_req . headers . pop ( '<STR_LIT>' , None ) <EOL> source_req . path_info = source_header <EOL> source_req . headers [ '<STR_LIT>' ] = '<STR_LIT:true>' <EOL> if '<STR_LIT>' in req . environ : <EOL> source_req . environ [ '<STR_LIT>' ] = lambda req : None <EOL> source_req . environ [ '<STR_LIT>' ] = True <EOL> orig_obj_name = self . object_name <EOL> orig_container_name = self . container_name <EOL> orig_account_name = self . account_name <EOL> sink_req = Request . blank ( req . path_info , <EOL> environ = req . environ , headers = req . headers ) <EOL> self . object_name = src_obj_name <EOL> self . container_name = src_container_name <EOL> self . account_name = src_account_name <EOL> source_resp = self . GET ( source_req ) <EOL> hook = req . environ . get ( <EOL> '<STR_LIT>' , <EOL> ( lambda source_req , source_resp , sink_req : source_resp ) ) <EOL> source_resp = hook ( source_req , source_resp , sink_req ) <EOL> self . object_name = orig_obj_name <EOL> self . container_name = orig_container_name <EOL> self . account_name = orig_account_name <EOL> if source_resp . status_int >= HTTP_MULTIPLE_CHOICES : <EOL> return source_resp , None , None , None <EOL> if source_resp . content_length is None : <EOL> raise HTTPRequestEntityTooLarge ( request = req ) <EOL> if source_resp . content_length > constraints . MAX_FILE_SIZE : <EOL> raise HTTPRequestEntityTooLarge ( request = req ) <EOL> data_source = iter ( source_resp . app_iter ) <EOL> sink_req . content_length = source_resp . content_length <EOL> sink_req . etag = source_resp . etag <EOL> del sink_req . headers [ '<STR_LIT>' ] <EOL> if '<STR_LIT>' in sink_req . headers : <EOL> del sink_req . headers [ '<STR_LIT>' ] <EOL> if not req . content_type_manually_set : <EOL> sink_req . headers [ '<STR_LIT:Content-Type>' ] = source_resp . headers [ '<STR_LIT:Content-Type>' ] <EOL> fresh_meta_flag = config_true_value ( <EOL> sink_req . headers . get ( '<STR_LIT>' , '<STR_LIT:false>' ) ) <EOL> if fresh_meta_flag or '<STR_LIT>' in sink_req . environ : <EOL> condition = lambda k : is_sys_meta ( '<STR_LIT:object>' , k ) <EOL> remove_items ( sink_req . headers , condition ) <EOL> copy_header_subset ( source_resp , sink_req , condition ) <EOL> else : <EOL> copy_headers_into ( source_resp , sink_req ) <EOL> copy_headers_into ( req , sink_req ) <EOL> if '<STR_LIT>' in source_resp . headers and ( req . params . get ( '<STR_LIT>' ) == '<STR_LIT>' or <EOL> '<STR_LIT>' in req . environ ) : <EOL> sink_req . headers [ '<STR_LIT>' ] = source_resp . headers [ '<STR_LIT>' ] <EOL> req = sink_req <EOL> def update_response ( req , resp ) : <EOL> acct , path = source_resp . environ [ '<STR_LIT>' ] . split ( '<STR_LIT:/>' , <NUM_LIT:3> ) [ <NUM_LIT:2> : <NUM_LIT:4> ] <EOL> resp . headers [ '<STR_LIT>' ] = quote ( acct ) <EOL> resp . headers [ '<STR_LIT>' ] = quote ( path ) <EOL> if '<STR_LIT>' in source_resp . headers : <EOL> resp . headers [ '<STR_LIT>' ] = source_resp . headers [ '<STR_LIT>' ] <EOL> copy_headers_into ( req , resp ) <EOL> return resp <EOL> return None , req , data_source , update_response <EOL> def _update_content_type ( self , req ) : <EOL> req . content_type_manually_set = True <EOL> detect_content_type = config_true_value ( req . headers . get ( '<STR_LIT>' ) ) <EOL> if detect_content_type or not req . headers . get ( '<STR_LIT>' ) : <EOL> guessed_type , _junk = mimetypes . guess_type ( req . path_info ) <EOL> req . headers [ '<STR_LIT:Content-Type>' ] = guessed_type or '<STR_LIT>' <EOL> if detect_content_type : <EOL> req . headers . pop ( '<STR_LIT>' ) <EOL> else : <EOL> req . content_type_manually_set = False <EOL> def _update_x_timestamp ( self , req ) : <EOL> if '<STR_LIT>' in req . headers : <EOL> try : <EOL> req_timestamp = Timestamp ( req . headers [ '<STR_LIT>' ] ) <EOL> except ValueError : <EOL> raise HTTPBadRequest ( <EOL> request = req , content_type = '<STR_LIT>' , <EOL> body = '<STR_LIT>' <EOL> '<STR_LIT>' % req . headers [ '<STR_LIT>' ] ) <EOL> req . headers [ '<STR_LIT>' ] = req_timestamp . internal <EOL> else : <EOL> req . headers [ '<STR_LIT>' ] = Timestamp ( time . time ( ) ) . internal <EOL> return None <EOL> def _check_failure_put_connections ( self , conns , req , nodes , min_conns ) : <EOL> """<STR_LIT>""" <EOL> if req . if_none_match is not None and '<STR_LIT:*>' in req . if_none_match : <EOL> statuses = [ conn . resp . status for conn in conns if conn . resp ] <EOL> if HTTP_PRECONDITION_FAILED in statuses : <EOL> self . app . logger . debug ( <EOL> _ ( '<STR_LIT>' ) , <EOL> { '<STR_LIT>' : statuses } ) <EOL> raise HTTPPreconditionFailed ( request = req ) <EOL> if any ( conn for conn in conns if conn . resp and <EOL> conn . resp . status == HTTP_CONFLICT ) : <EOL> status_times = [ '<STR_LIT>' % { <EOL> '<STR_LIT:status>' : conn . resp . status , <EOL> '<STR_LIT>' : HeaderKeyDict ( <EOL> conn . resp . getheaders ( ) ) . get ( <EOL> '<STR_LIT>' , '<STR_LIT>' ) <EOL> } for conn in conns if conn . resp ] <EOL> self . app . logger . debug ( <EOL> _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> { '<STR_LIT>' : req . timestamp . internal , <EOL> '<STR_LIT>' : '<STR_LIT:U+002CU+0020>' . join ( status_times ) } ) <EOL> raise HTTPAccepted ( request = req ) <EOL> self . _check_min_conn ( req , conns , min_conns ) <EOL> def _connect_put_node ( self , nodes , part , path , headers , <EOL> logger_thread_locals ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( ) <EOL> def _get_put_connections ( self , req , nodes , partition , outgoing_headers , <EOL> policy , expect ) : <EOL> """<STR_LIT>""" <EOL> obj_ring = policy . object_ring <EOL> node_iter = GreenthreadSafeIterator ( <EOL> self . iter_nodes_local_first ( obj_ring , partition ) ) <EOL> pile = GreenPile ( len ( nodes ) ) <EOL> for nheaders in outgoing_headers : <EOL> if expect : <EOL> nheaders [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> pile . spawn ( self . _connect_put_node , node_iter , partition , <EOL> req . swift_entity_path , nheaders , <EOL> self . app . logger . thread_locals ) <EOL> conns = [ conn for conn in pile if conn ] <EOL> return conns <EOL> def _check_min_conn ( self , req , conns , min_conns , msg = None ) : <EOL> msg = msg or '<STR_LIT>' '<STR_LIT>' <EOL> if len ( conns ) < min_conns : <EOL> self . app . logger . error ( ( msg ) , <EOL> { '<STR_LIT>' : len ( conns ) , '<STR_LIT>' : min_conns } ) <EOL> raise HTTPServiceUnavailable ( request = req ) <EOL> def _store_object ( self , req , data_source , nodes , partition , <EOL> outgoing_headers ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( ) <EOL> def _delete_object ( self , req , obj_ring , partition , headers ) : <EOL> """<STR_LIT>""" <EOL> status_overrides = { <NUM_LIT> : <NUM_LIT> } <EOL> resp = self . make_requests ( req , obj_ring , <EOL> partition , '<STR_LIT>' , req . swift_entity_path , <EOL> headers , overrides = status_overrides ) <EOL> return resp <EOL> def _post_object ( self , req , obj_ring , partition , headers ) : <EOL> """<STR_LIT>""" <EOL> resp = self . make_requests ( req , obj_ring , partition , <EOL> '<STR_LIT:POST>' , req . swift_entity_path , headers ) <EOL> return resp <EOL> @ public <EOL> @ cors_validation <EOL> @ delay_denial <EOL> def PUT ( self , req ) : <EOL> """<STR_LIT>""" <EOL> if req . if_none_match is not None and '<STR_LIT:*>' not in req . if_none_match : <EOL> return HTTPBadRequest ( request = req , content_type = '<STR_LIT>' , <EOL> body = '<STR_LIT>' ) <EOL> container_info = self . container_info ( <EOL> self . account_name , self . container_name , req ) <EOL> policy_index = req . headers . get ( '<STR_LIT>' , <EOL> container_info [ '<STR_LIT>' ] ) <EOL> obj_ring = self . app . get_object_ring ( policy_index ) <EOL> container_nodes = container_info [ '<STR_LIT>' ] <EOL> container_partition = container_info [ '<STR_LIT>' ] <EOL> partition , nodes = obj_ring . get_nodes ( <EOL> self . account_name , self . container_name , self . object_name ) <EOL> req . headers [ '<STR_LIT>' ] = policy_index <EOL> req . acl = container_info [ '<STR_LIT>' ] <EOL> req . environ [ '<STR_LIT>' ] = container_info [ '<STR_LIT>' ] <EOL> if '<STR_LIT>' in req . environ : <EOL> aresp = req . environ [ '<STR_LIT>' ] ( req ) <EOL> if aresp : <EOL> return aresp <EOL> if not container_info [ '<STR_LIT>' ] : <EOL> return HTTPNotFound ( request = req ) <EOL> self . _update_content_type ( req ) <EOL> error_response = check_object_creation ( req , self . object_name ) or check_content_type ( req ) <EOL> if error_response : <EOL> return error_response <EOL> self . _update_x_timestamp ( req ) <EOL> source_header = req . headers . get ( '<STR_LIT>' ) <EOL> if source_header : <EOL> error_response , req , data_source , update_response = self . _handle_copy_request ( req ) <EOL> if error_response : <EOL> return error_response <EOL> else : <EOL> def reader ( ) : <EOL> try : <EOL> return req . environ [ '<STR_LIT>' ] . read ( <EOL> self . app . client_chunk_size ) <EOL> except ( ValueError , IOError ) as e : <EOL> raise ChunkReadError ( str ( e ) ) <EOL> data_source = iter ( reader , '<STR_LIT>' ) <EOL> update_response = lambda req , resp : resp <EOL> req , delete_at_container , delete_at_part , delete_at_nodes = self . _config_obj_expiration ( req ) <EOL> outgoing_headers = self . _backend_requests ( <EOL> req , len ( nodes ) , container_partition , container_nodes , <EOL> delete_at_container , delete_at_part , delete_at_nodes ) <EOL> resp = self . _store_object ( <EOL> req , data_source , nodes , partition , outgoing_headers ) <EOL> return update_response ( req , resp ) <EOL> @ public <EOL> @ cors_validation <EOL> @ delay_denial <EOL> def DELETE ( self , req ) : <EOL> """<STR_LIT>""" <EOL> container_info = self . container_info ( <EOL> self . account_name , self . container_name , req ) <EOL> policy_index = req . headers . get ( '<STR_LIT>' , <EOL> container_info [ '<STR_LIT>' ] ) <EOL> obj_ring = self . app . get_object_ring ( policy_index ) <EOL> req . headers [ '<STR_LIT>' ] = policy_index <EOL> container_partition = container_info [ '<STR_LIT>' ] <EOL> containers = container_info [ '<STR_LIT>' ] <EOL> req . acl = container_info [ '<STR_LIT>' ] <EOL> req . environ [ '<STR_LIT>' ] = container_info [ '<STR_LIT>' ] <EOL> if '<STR_LIT>' in req . environ : <EOL> aresp = req . environ [ '<STR_LIT>' ] ( req ) <EOL> if aresp : <EOL> return aresp <EOL> if not containers : <EOL> return HTTPNotFound ( request = req ) <EOL> partition , nodes = obj_ring . get_nodes ( <EOL> self . account_name , self . container_name , self . object_name ) <EOL> if '<STR_LIT>' in req . headers : <EOL> try : <EOL> req_timestamp = Timestamp ( req . headers [ '<STR_LIT>' ] ) <EOL> except ValueError : <EOL> return HTTPBadRequest ( <EOL> request = req , content_type = '<STR_LIT>' , <EOL> body = '<STR_LIT>' <EOL> '<STR_LIT>' % req . headers [ '<STR_LIT>' ] ) <EOL> req . headers [ '<STR_LIT>' ] = req_timestamp . internal <EOL> else : <EOL> req . headers [ '<STR_LIT>' ] = Timestamp ( time . time ( ) ) . internal <EOL> headers = self . _backend_requests ( <EOL> req , len ( nodes ) , container_partition , containers ) <EOL> return self . _delete_object ( req , obj_ring , partition , headers ) <EOL> def _reroute ( self , policy ) : <EOL> """<STR_LIT>""" <EOL> if not policy : <EOL> raise HTTPServiceUnavailable ( '<STR_LIT>' ) <EOL> if policy . policy_type != self . policy_type : <EOL> controller = self . app . obj_controller_router [ policy ] ( <EOL> self . app , self . account_name , self . container_name , <EOL> self . object_name ) <EOL> else : <EOL> controller = self <EOL> return controller <EOL> @ public <EOL> @ cors_validation <EOL> @ delay_denial <EOL> def COPY ( self , req ) : <EOL> """<STR_LIT>""" <EOL> if not req . headers . get ( '<STR_LIT>' ) : <EOL> return HTTPPreconditionFailed ( request = req , <EOL> body = '<STR_LIT>' ) <EOL> dest_account = self . account_name <EOL> if '<STR_LIT>' in req . headers : <EOL> dest_account = req . headers . get ( '<STR_LIT>' ) <EOL> dest_account = check_account_format ( req , dest_account ) <EOL> req . headers [ '<STR_LIT>' ] = self . account_name <EOL> self . account_name = dest_account <EOL> del req . headers [ '<STR_LIT>' ] <EOL> dest_container , dest_object = check_destination_header ( req ) <EOL> source = '<STR_LIT>' % ( self . container_name , self . object_name ) <EOL> self . container_name = dest_container <EOL> self . object_name = dest_object <EOL> req . method = '<STR_LIT>' <EOL> req . path_info = '<STR_LIT>' % ( dest_account , dest_container , dest_object ) <EOL> req . headers [ '<STR_LIT>' ] = <NUM_LIT:0> <EOL> req . headers [ '<STR_LIT>' ] = quote ( source ) <EOL> del req . headers [ '<STR_LIT>' ] <EOL> container_info = self . container_info ( <EOL> dest_account , dest_container , req ) <EOL> dest_policy = POLICIES . get_by_index ( container_info [ '<STR_LIT>' ] ) <EOL> return self . _reroute ( dest_policy ) . PUT ( req ) <EOL> @ ObjectControllerRouter . register ( REPL_POLICY ) <EOL> class ReplicatedObjectController ( BaseObjectController ) : <EOL> def _get_or_head_response ( self , req , node_iter , partition , policy ) : <EOL> concurrency = self . app . get_object_ring ( policy . idx ) . replica_count if self . app . concurrent_gets else <NUM_LIT:1> <EOL> resp = self . GETorHEAD_base ( <EOL> req , _ ( '<STR_LIT>' ) , node_iter , partition , <EOL> req . swift_entity_path , concurrency ) <EOL> return resp <EOL> def _connect_put_node ( self , nodes , part , path , headers , <EOL> logger_thread_locals ) : <EOL> """<STR_LIT>""" <EOL> self . app . logger . thread_locals = logger_thread_locals <EOL> for node in nodes : <EOL> try : <EOL> start_time = time . time ( ) <EOL> with ConnectionTimeout ( self . app . conn_timeout ) : <EOL> conn = http_connect ( <EOL> node [ '<STR_LIT>' ] , node [ '<STR_LIT:port>' ] , node [ '<STR_LIT>' ] , part , '<STR_LIT>' , <EOL> path , headers ) <EOL> self . app . set_node_timing ( node , time . time ( ) - start_time ) <EOL> with Timeout ( self . app . node_timeout ) : <EOL> resp = conn . getexpect ( ) <EOL> if resp . status == HTTP_CONTINUE : <EOL> conn . resp = None <EOL> conn . node = node <EOL> return conn <EOL> elif ( is_success ( resp . status ) <EOL> or resp . status in ( HTTP_CONFLICT , <EOL> HTTP_UNPROCESSABLE_ENTITY ) ) : <EOL> conn . resp = resp <EOL> conn . node = node <EOL> return conn <EOL> elif headers [ '<STR_LIT>' ] is not None and resp . status == HTTP_PRECONDITION_FAILED : <EOL> conn . resp = resp <EOL> conn . node = node <EOL> return conn <EOL> elif resp . status == HTTP_INSUFFICIENT_STORAGE : <EOL> self . app . error_limit ( node , _ ( '<STR_LIT>' ) ) <EOL> elif is_server_error ( resp . status ) : <EOL> self . app . error_occurred ( <EOL> node , <EOL> _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) % { <EOL> '<STR_LIT:status>' : resp . status } ) <EOL> except ( Exception , Timeout ) : <EOL> self . app . exception_occurred ( <EOL> node , _ ( '<STR_LIT>' ) , <EOL> _ ( '<STR_LIT>' ) % path ) <EOL> def _send_file ( self , conn , path ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> chunk = conn . queue . get ( ) <EOL> if not conn . failed : <EOL> try : <EOL> with ChunkWriteTimeout ( self . app . node_timeout ) : <EOL> conn . send ( chunk ) <EOL> except ( Exception , ChunkWriteTimeout ) : <EOL> conn . failed = True <EOL> self . app . exception_occurred ( <EOL> conn . node , _ ( '<STR_LIT>' ) , <EOL> _ ( '<STR_LIT>' ) % path ) <EOL> conn . queue . task_done ( ) <EOL> def _transfer_data ( self , req , data_source , conns , nodes ) : <EOL> """<STR_LIT>""" <EOL> min_conns = quorum_size ( len ( nodes ) ) <EOL> bytes_transferred = <NUM_LIT:0> <EOL> try : <EOL> with ContextPool ( len ( nodes ) ) as pool : <EOL> for conn in conns : <EOL> conn . failed = False <EOL> conn . queue = Queue ( self . app . put_queue_depth ) <EOL> pool . spawn ( self . _send_file , conn , req . path ) <EOL> while True : <EOL> with ChunkReadTimeout ( self . app . client_timeout ) : <EOL> try : <EOL> chunk = next ( data_source ) <EOL> except StopIteration : <EOL> if req . is_chunked : <EOL> for conn in conns : <EOL> conn . queue . put ( '<STR_LIT>' ) <EOL> break <EOL> bytes_transferred += len ( chunk ) <EOL> if bytes_transferred > constraints . MAX_FILE_SIZE : <EOL> raise HTTPRequestEntityTooLarge ( request = req ) <EOL> for conn in list ( conns ) : <EOL> if not conn . failed : <EOL> conn . queue . put ( <EOL> '<STR_LIT>' % ( len ( chunk ) , chunk ) <EOL> if req . is_chunked else chunk ) <EOL> else : <EOL> conn . close ( ) <EOL> conns . remove ( conn ) <EOL> self . _check_min_conn ( <EOL> req , conns , min_conns , <EOL> msg = '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> for conn in conns : <EOL> if conn . queue . unfinished_tasks : <EOL> conn . queue . join ( ) <EOL> conns = [ conn for conn in conns if not conn . failed ] <EOL> self . _check_min_conn ( <EOL> req , conns , min_conns , <EOL> msg = '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> except ChunkReadTimeout as err : <EOL> self . app . logger . warning ( <EOL> _ ( '<STR_LIT>' ) , err . seconds ) <EOL> self . app . logger . increment ( '<STR_LIT>' ) <EOL> raise HTTPRequestTimeout ( request = req ) <EOL> except HTTPException : <EOL> raise <EOL> except ChunkReadError : <EOL> req . client_disconnect = True <EOL> self . app . logger . warning ( <EOL> _ ( '<STR_LIT>' ) ) <EOL> self . app . logger . increment ( '<STR_LIT>' ) <EOL> raise HTTPClientDisconnect ( request = req ) <EOL> except Timeout : <EOL> self . app . logger . exception ( <EOL> _ ( '<STR_LIT>' ) ) <EOL> raise HTTPClientDisconnect ( request = req ) <EOL> except Exception : <EOL> self . app . logger . exception ( <EOL> _ ( '<STR_LIT>' ) , <EOL> { '<STR_LIT:path>' : req . path } ) <EOL> raise HTTPInternalServerError ( request = req ) <EOL> if req . content_length and bytes_transferred < req . content_length : <EOL> req . client_disconnect = True <EOL> self . app . logger . warning ( <EOL> _ ( '<STR_LIT>' ) ) <EOL> self . app . logger . increment ( '<STR_LIT>' ) <EOL> raise HTTPClientDisconnect ( request = req ) <EOL> def _store_object ( self , req , data_source , nodes , partition , <EOL> outgoing_headers ) : <EOL> """<STR_LIT>""" <EOL> policy_index = req . headers . get ( '<STR_LIT>' ) <EOL> policy = POLICIES . get_by_index ( policy_index ) <EOL> if not nodes : <EOL> return HTTPNotFound ( ) <EOL> if ( req . content_length > <NUM_LIT:0> ) or req . is_chunked : <EOL> expect = True <EOL> else : <EOL> expect = False <EOL> conns = self . _get_put_connections ( req , nodes , partition , <EOL> outgoing_headers , policy , expect ) <EOL> min_conns = quorum_size ( len ( nodes ) ) <EOL> try : <EOL> self . _check_failure_put_connections ( conns , req , nodes , min_conns ) <EOL> self . _transfer_data ( req , data_source , conns , nodes ) <EOL> statuses , reasons , bodies , etags = self . _get_put_responses ( <EOL> req , conns , nodes ) <EOL> except HTTPException as resp : <EOL> return resp <EOL> finally : <EOL> for conn in conns : <EOL> conn . close ( ) <EOL> if len ( etags ) > <NUM_LIT:1> : <EOL> self . app . logger . error ( <EOL> _ ( '<STR_LIT>' ) , len ( etags ) ) <EOL> return HTTPServerError ( request = req ) <EOL> etag = etags . pop ( ) if len ( etags ) else None <EOL> resp = self . best_response ( req , statuses , reasons , bodies , <EOL> _ ( '<STR_LIT>' ) , etag = etag ) <EOL> resp . last_modified = math . ceil ( <EOL> float ( Timestamp ( req . headers [ '<STR_LIT>' ] ) ) ) <EOL> return resp <EOL> class ECAppIter ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , path , policy , internal_parts_iters , range_specs , <EOL> fa_length , obj_length , logger ) : <EOL> self . path = path <EOL> self . policy = policy <EOL> self . internal_parts_iters = internal_parts_iters <EOL> self . range_specs = range_specs <EOL> self . fa_length = fa_length <EOL> self . obj_length = obj_length if obj_length is not None else <NUM_LIT:0> <EOL> self . boundary = '<STR_LIT>' <EOL> self . logger = logger <EOL> self . mime_boundary = None <EOL> self . learned_content_type = None <EOL> self . stashed_iter = None <EOL> def close ( self ) : <EOL> if self . stashed_iter : <EOL> self . stashed_iter . close ( ) <EOL> for it in self . internal_parts_iters : <EOL> close_if_possible ( it ) <EOL> def kickoff ( self , req , resp ) : <EOL> """<STR_LIT>""" <EOL> self . mime_boundary = resp . boundary <EOL> self . stashed_iter = reiterate ( self . _real_iter ( req , resp . headers ) ) <EOL> if self . learned_content_type is not None : <EOL> resp . content_type = self . learned_content_type <EOL> resp . content_length = self . obj_length <EOL> def _next_range ( self ) : <EOL> frag_iters = [ ] <EOL> headers = None <EOL> for parts_iter in self . internal_parts_iters : <EOL> part_info = next ( parts_iter ) <EOL> frag_iters . append ( part_info [ '<STR_LIT>' ] ) <EOL> headers = part_info [ '<STR_LIT>' ] <EOL> headers = HeaderKeyDict ( headers ) <EOL> return headers , frag_iters <EOL> def _actual_range ( self , req_start , req_end , entity_length ) : <EOL> try : <EOL> rng = Range ( "<STR_LIT>" % ( <EOL> req_start if req_start is not None else '<STR_LIT>' , <EOL> req_end if req_end is not None else '<STR_LIT>' ) ) <EOL> except ValueError : <EOL> return ( None , None ) <EOL> rfl = rng . ranges_for_length ( entity_length ) <EOL> if not rfl : <EOL> return ( None , None ) <EOL> else : <EOL> return ( rfl [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] , rfl [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] - <NUM_LIT:1> ) <EOL> def _fill_out_range_specs_from_obj_length ( self , range_specs ) : <EOL> for spec in range_specs : <EOL> cstart , cend = self . _actual_range ( <EOL> spec [ '<STR_LIT>' ] , <EOL> spec [ '<STR_LIT>' ] , <EOL> self . obj_length ) <EOL> spec [ '<STR_LIT>' ] = cstart <EOL> spec [ '<STR_LIT>' ] = cend <EOL> spec [ '<STR_LIT>' ] = ( cstart is not None and cend is not None ) <EOL> sstart , send = self . _actual_range ( <EOL> spec [ '<STR_LIT>' ] , <EOL> spec [ '<STR_LIT>' ] , <EOL> self . obj_length ) <EOL> seg_size = self . policy . ec_segment_size <EOL> if spec [ '<STR_LIT>' ] is None and sstart % seg_size != <NUM_LIT:0> : <EOL> sstart += seg_size - ( sstart % seg_size ) <EOL> spec [ '<STR_LIT>' ] = sstart <EOL> spec [ '<STR_LIT>' ] = send <EOL> def _fill_out_range_specs_from_fa_length ( self , fa_length , range_specs ) : <EOL> for spec in range_specs : <EOL> fstart , fend = self . _actual_range ( <EOL> spec [ '<STR_LIT>' ] , <EOL> spec [ '<STR_LIT>' ] , <EOL> fa_length ) <EOL> spec [ '<STR_LIT>' ] = fstart <EOL> spec [ '<STR_LIT>' ] = fend <EOL> def __iter__ ( self ) : <EOL> if self . stashed_iter is not None : <EOL> return iter ( self . stashed_iter ) <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> def _real_iter ( self , req , resp_headers ) : <EOL> if not self . range_specs : <EOL> client_asked_for_range = False <EOL> range_specs = [ { <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : ( None if self . obj_length is None <EOL> else self . obj_length - <NUM_LIT:1> ) , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : ( None if self . obj_length is None <EOL> else self . obj_length - <NUM_LIT:1> ) , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : ( None if self . obj_length is None <EOL> else self . obj_length - <NUM_LIT:1> ) , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : ( None if self . obj_length is None <EOL> else self . obj_length - <NUM_LIT:1> ) , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : self . fa_length - <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : self . fa_length - <NUM_LIT:1> , <EOL> '<STR_LIT>' : self . obj_length > <NUM_LIT:0> , <EOL> } ] <EOL> else : <EOL> client_asked_for_range = True <EOL> range_specs = self . range_specs <EOL> self . _fill_out_range_specs_from_obj_length ( range_specs ) <EOL> multipart = ( len ( [ rs for rs in range_specs if rs [ '<STR_LIT>' ] ] ) > <NUM_LIT:1> ) <EOL> def convert_ranges_iter ( ) : <EOL> seen_first_headers = False <EOL> ranges_for_resp = { } <EOL> while True : <EOL> next_range = self . _next_range ( ) <EOL> headers , frag_iters = next_range <EOL> content_type = headers [ '<STR_LIT:Content-Type>' ] <EOL> content_range = headers . get ( '<STR_LIT>' ) <EOL> if content_range is not None : <EOL> fa_start , fa_end , fa_length = parse_content_range ( <EOL> content_range ) <EOL> elif self . fa_length <= <NUM_LIT:0> : <EOL> fa_start = None <EOL> fa_end = None <EOL> fa_length = <NUM_LIT:0> <EOL> else : <EOL> fa_start = <NUM_LIT:0> <EOL> fa_end = self . fa_length - <NUM_LIT:1> <EOL> fa_length = self . fa_length <EOL> if not seen_first_headers : <EOL> self . _fill_out_range_specs_from_fa_length ( <EOL> fa_length , range_specs ) <EOL> satisfiable = False <EOL> for range_spec in range_specs : <EOL> satisfiable |= range_spec [ '<STR_LIT>' ] <EOL> key = ( range_spec [ '<STR_LIT>' ] , <EOL> range_spec [ '<STR_LIT>' ] ) <EOL> ranges_for_resp . setdefault ( key , [ ] ) . append ( range_spec ) <EOL> if client_asked_for_range and not satisfiable : <EOL> req . environ [ <EOL> '<STR_LIT>' ] = True <EOL> raise HTTPRequestedRangeNotSatisfiable ( <EOL> request = req , headers = resp_headers ) <EOL> self . learned_content_type = content_type <EOL> seen_first_headers = True <EOL> range_spec = ranges_for_resp [ ( fa_start , fa_end ) ] . pop ( <NUM_LIT:0> ) <EOL> seg_iter = self . _decode_segments_from_fragments ( frag_iters ) <EOL> if not range_spec [ '<STR_LIT>' ] : <EOL> for x in seg_iter : <EOL> pass <EOL> continue <EOL> byterange_iter = self . _iter_one_range ( range_spec , seg_iter ) <EOL> converted = { <EOL> "<STR_LIT>" : range_spec [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : range_spec [ "<STR_LIT>" ] , <EOL> "<STR_LIT>" : content_type , <EOL> "<STR_LIT>" : byterange_iter } <EOL> if self . obj_length is not None : <EOL> converted [ "<STR_LIT>" ] = self . obj_length <EOL> yield converted <EOL> return document_iters_to_http_response_body ( <EOL> convert_ranges_iter ( ) , self . mime_boundary , multipart , self . logger ) <EOL> def _iter_one_range ( self , range_spec , segment_iter ) : <EOL> client_start = range_spec [ '<STR_LIT>' ] <EOL> client_end = range_spec [ '<STR_LIT>' ] <EOL> segment_start = range_spec [ '<STR_LIT>' ] <EOL> segment_end = range_spec [ '<STR_LIT>' ] <EOL> segment_end = ( min ( segment_end , self . obj_length - <NUM_LIT:1> ) <EOL> if segment_end is not None <EOL> else self . obj_length - <NUM_LIT:1> ) <EOL> client_end = ( min ( client_end , self . obj_length - <NUM_LIT:1> ) <EOL> if client_end is not None <EOL> else self . obj_length - <NUM_LIT:1> ) <EOL> num_segments = int ( <EOL> math . ceil ( float ( segment_end + <NUM_LIT:1> - segment_start ) <EOL> / self . policy . ec_segment_size ) ) <EOL> start_overrun = client_start - segment_start <EOL> end_overrun = segment_end - client_end <EOL> for i , next_seg in enumerate ( segment_iter ) : <EOL> if start_overrun > <NUM_LIT:0> : <EOL> seglen = len ( next_seg ) <EOL> if seglen <= start_overrun : <EOL> start_overrun -= seglen <EOL> continue <EOL> else : <EOL> next_seg = next_seg [ start_overrun : ] <EOL> start_overrun = <NUM_LIT:0> <EOL> if i == ( num_segments - <NUM_LIT:1> ) and end_overrun : <EOL> next_seg = next_seg [ : - end_overrun ] <EOL> yield next_seg <EOL> def _decode_segments_from_fragments ( self , fragment_iters ) : <EOL> queues = [ Queue ( <NUM_LIT:1> ) for _junk in range ( len ( fragment_iters ) ) ] <EOL> def put_fragments_in_queue ( frag_iter , queue ) : <EOL> try : <EOL> for fragment in frag_iter : <EOL> if fragment . startswith ( '<STR_LIT:U+0020>' ) : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> queue . put ( fragment ) <EOL> except GreenletExit : <EOL> pass <EOL> except ChunkReadTimeout : <EOL> self . logger . exception ( "<STR_LIT>" % <EOL> self . path ) <EOL> except : <EOL> self . logger . exception ( "<STR_LIT>" % <EOL> self . path ) <EOL> finally : <EOL> queue . resize ( <NUM_LIT:2> ) <EOL> queue . put ( None ) <EOL> frag_iter . close ( ) <EOL> with ContextPool ( len ( fragment_iters ) ) as pool : <EOL> for frag_iter , queue in zip ( fragment_iters , queues ) : <EOL> pool . spawn ( put_fragments_in_queue , frag_iter , queue ) <EOL> while True : <EOL> fragments = [ ] <EOL> for queue in queues : <EOL> fragment = queue . get ( ) <EOL> queue . task_done ( ) <EOL> fragments . append ( fragment ) <EOL> if not all ( fragments ) : <EOL> break <EOL> try : <EOL> segment = self . policy . pyeclib_driver . decode ( fragments ) <EOL> except ECDriverError : <EOL> self . logger . exception ( "<STR_LIT>" % <EOL> self . path ) <EOL> raise <EOL> yield segment <EOL> def app_iter_range ( self , start , end ) : <EOL> return self <EOL> def app_iter_ranges ( self , ranges , content_type , boundary , content_size ) : <EOL> return self <EOL> def client_range_to_segment_range ( client_start , client_end , segment_size ) : <EOL> """<STR_LIT>""" <EOL> segment_start = ( <EOL> int ( client_start // segment_size ) <EOL> * segment_size ) if client_start is not None else None <EOL> segment_end = ( <EOL> None if client_end is None else <EOL> ( ( ( int ( client_end // segment_size ) + <NUM_LIT:1> ) <EOL> * segment_size ) - <NUM_LIT:1> ) if client_start is not None else <EOL> ( int ( math . ceil ( ( <EOL> float ( client_end ) / segment_size ) + <NUM_LIT:1> ) ) <EOL> * segment_size ) ) <EOL> return ( segment_start , segment_end ) <EOL> def segment_range_to_fragment_range ( segment_start , segment_end , segment_size , <EOL> fragment_size ) : <EOL> """<STR_LIT>""" <EOL> fragment_start = ( ( <EOL> segment_start / segment_size * fragment_size ) <EOL> if segment_start is not None else None ) <EOL> fragment_end = ( <EOL> None if segment_end is None else <EOL> ( ( segment_end + <NUM_LIT:1> ) / segment_size <EOL> * fragment_size ) if segment_start is None else <EOL> ( ( segment_end + <NUM_LIT:1> ) / segment_size * fragment_size ) - <NUM_LIT:1> ) <EOL> return ( fragment_start , fragment_end ) <EOL> NO_DATA_SENT = <NUM_LIT:1> <EOL> SENDING_DATA = <NUM_LIT:2> <EOL> DATA_SENT = <NUM_LIT:3> <EOL> DATA_ACKED = <NUM_LIT:4> <EOL> COMMIT_SENT = <NUM_LIT:5> <EOL> class ECPutter ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , conn , node , resp , path , connect_duration , <EOL> mime_boundary ) : <EOL> self . conn = conn <EOL> self . node = node <EOL> self . resp = resp <EOL> self . path = path <EOL> self . connect_duration = connect_duration <EOL> self . node_index = node . get ( '<STR_LIT:index>' ) <EOL> self . mime_boundary = mime_boundary <EOL> self . chunk_hasher = md5 ( ) <EOL> self . failed = False <EOL> self . queue = None <EOL> self . state = NO_DATA_SENT <EOL> def await_response ( self , timeout , informational = False ) : <EOL> """<STR_LIT>""" <EOL> conn = self . conn <EOL> with Timeout ( timeout ) : <EOL> if not conn . resp : <EOL> if informational : <EOL> self . resp = conn . getexpect ( ) <EOL> else : <EOL> self . resp = conn . getresponse ( ) <EOL> return self . resp <EOL> def spawn_sender_greenthread ( self , pool , queue_depth , write_timeout , <EOL> exception_handler ) : <EOL> """<STR_LIT>""" <EOL> self . queue = Queue ( queue_depth ) <EOL> pool . spawn ( self . _send_file , write_timeout , exception_handler ) <EOL> def wait ( self ) : <EOL> if self . queue . unfinished_tasks : <EOL> self . queue . join ( ) <EOL> def _start_mime_doc_object_body ( self ) : <EOL> self . queue . put ( "<STR_LIT>" % <EOL> ( self . mime_boundary , ) ) <EOL> def send_chunk ( self , chunk ) : <EOL> if not chunk : <EOL> return <EOL> elif self . state == DATA_SENT : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if self . state == NO_DATA_SENT and self . mime_boundary : <EOL> self . _start_mime_doc_object_body ( ) <EOL> self . state = SENDING_DATA <EOL> self . queue . put ( chunk ) <EOL> def end_of_object_data ( self , footer_metadata ) : <EOL> """<STR_LIT>""" <EOL> if self . state == DATA_SENT : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> elif self . state == NO_DATA_SENT and self . mime_boundary : <EOL> self . _start_mime_doc_object_body ( ) <EOL> footer_body = json . dumps ( footer_metadata ) <EOL> footer_md5 = md5 ( footer_body ) . hexdigest ( ) <EOL> tail_boundary = ( "<STR_LIT>" % ( self . mime_boundary , ) ) <EOL> message_parts = [ <EOL> ( "<STR_LIT>" % self . mime_boundary ) , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" % footer_md5 , <EOL> "<STR_LIT:\r\n>" , <EOL> footer_body , "<STR_LIT:\r\n>" , <EOL> tail_boundary , "<STR_LIT:\r\n>" , <EOL> ] <EOL> self . queue . put ( "<STR_LIT>" . join ( message_parts ) ) <EOL> self . queue . put ( '<STR_LIT>' ) <EOL> self . state = DATA_SENT <EOL> def send_commit_confirmation ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . state == COMMIT_SENT : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> self . state = DATA_ACKED <EOL> if self . mime_boundary : <EOL> body = "<STR_LIT>" <EOL> tail_boundary = ( "<STR_LIT>" % ( self . mime_boundary , ) ) <EOL> message_parts = [ <EOL> "<STR_LIT>" , <EOL> "<STR_LIT:\r\n>" , <EOL> body , "<STR_LIT:\r\n>" , <EOL> tail_boundary , <EOL> ] <EOL> self . queue . put ( "<STR_LIT>" . join ( message_parts ) ) <EOL> self . queue . put ( '<STR_LIT>' ) <EOL> self . state = COMMIT_SENT <EOL> def _send_file ( self , write_timeout , exception_handler ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> chunk = self . queue . get ( ) <EOL> if not self . failed : <EOL> to_send = "<STR_LIT>" % ( len ( chunk ) , chunk ) <EOL> try : <EOL> with ChunkWriteTimeout ( write_timeout ) : <EOL> self . conn . send ( to_send ) <EOL> except ( Exception , ChunkWriteTimeout ) : <EOL> self . failed = True <EOL> exception_handler ( self . conn . node , _ ( '<STR_LIT>' ) , <EOL> _ ( '<STR_LIT>' ) % self . path ) <EOL> self . queue . task_done ( ) <EOL> @ classmethod <EOL> def connect ( cls , node , part , path , headers , conn_timeout , node_timeout , <EOL> chunked = False , expected_frag_archive_size = None ) : <EOL> """<STR_LIT>""" <EOL> mime_boundary = "<STR_LIT>" % random . randint ( <NUM_LIT:0> , <NUM_LIT:16> ** <NUM_LIT:64> ) <EOL> headers = HeaderKeyDict ( headers ) <EOL> headers [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> headers [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> headers . pop ( '<STR_LIT>' ) <EOL> headers [ '<STR_LIT>' ] = expected_frag_archive_size <EOL> headers [ '<STR_LIT>' ] = mime_boundary <EOL> headers [ '<STR_LIT>' ] = '<STR_LIT:yes>' <EOL> headers [ '<STR_LIT>' ] = '<STR_LIT:yes>' <EOL> start_time = time . time ( ) <EOL> with ConnectionTimeout ( conn_timeout ) : <EOL> conn = http_connect ( node [ '<STR_LIT>' ] , node [ '<STR_LIT:port>' ] , node [ '<STR_LIT>' ] , <EOL> part , '<STR_LIT>' , path , headers ) <EOL> connect_duration = time . time ( ) - start_time <EOL> with ResponseTimeout ( node_timeout ) : <EOL> resp = conn . getexpect ( ) <EOL> if resp . status == HTTP_INSUFFICIENT_STORAGE : <EOL> raise InsufficientStorage <EOL> if is_server_error ( resp . status ) : <EOL> raise PutterConnectError ( resp . status ) <EOL> if is_informational ( resp . status ) : <EOL> continue_headers = HeaderKeyDict ( resp . getheaders ( ) ) <EOL> can_send_metadata_footer = config_true_value ( <EOL> continue_headers . get ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> can_handle_multiphase_put = config_true_value ( <EOL> continue_headers . get ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> if not can_send_metadata_footer : <EOL> raise FooterNotSupported ( ) <EOL> if not can_handle_multiphase_put : <EOL> raise MultiphasePUTNotSupported ( ) <EOL> conn . node = node <EOL> conn . resp = None <EOL> if is_success ( resp . status ) or resp . status == HTTP_CONFLICT : <EOL> conn . resp = resp <EOL> elif ( headers . get ( '<STR_LIT>' , None ) is not None and <EOL> resp . status == HTTP_PRECONDITION_FAILED ) : <EOL> conn . resp = resp <EOL> return cls ( conn , node , resp , path , connect_duration , mime_boundary ) <EOL> def chunk_transformer ( policy , nstreams ) : <EOL> segment_size = policy . ec_segment_size <EOL> buf = collections . deque ( ) <EOL> total_buf_len = <NUM_LIT:0> <EOL> chunk = yield <EOL> while chunk : <EOL> buf . append ( chunk ) <EOL> total_buf_len += len ( chunk ) <EOL> if total_buf_len >= segment_size : <EOL> chunks_to_encode = [ ] <EOL> while total_buf_len >= segment_size : <EOL> to_take = segment_size <EOL> pieces = [ ] <EOL> while to_take > <NUM_LIT:0> : <EOL> piece = buf . popleft ( ) <EOL> if len ( piece ) > to_take : <EOL> buf . appendleft ( piece [ to_take : ] ) <EOL> piece = piece [ : to_take ] <EOL> pieces . append ( piece ) <EOL> to_take -= len ( piece ) <EOL> total_buf_len -= len ( piece ) <EOL> chunks_to_encode . append ( '<STR_LIT>' . join ( pieces ) ) <EOL> frags_by_byte_order = [ ] <EOL> for chunk_to_encode in chunks_to_encode : <EOL> frags_by_byte_order . append ( <EOL> policy . pyeclib_driver . encode ( chunk_to_encode ) ) <EOL> obj_data = [ '<STR_LIT>' . join ( frags ) <EOL> for frags in zip ( * frags_by_byte_order ) ] <EOL> chunk = yield obj_data <EOL> else : <EOL> chunk = yield None <EOL> last_bytes = '<STR_LIT>' . join ( buf ) <EOL> if last_bytes : <EOL> last_frags = policy . pyeclib_driver . encode ( last_bytes ) <EOL> yield last_frags <EOL> else : <EOL> yield [ '<STR_LIT>' ] * nstreams <EOL> def trailing_metadata ( policy , client_obj_hasher , <EOL> bytes_transferred_from_client , <EOL> fragment_archive_index ) : <EOL> return { <EOL> '<STR_LIT>' : client_obj_hasher . hexdigest ( ) , <EOL> '<STR_LIT>' : <EOL> str ( bytes_transferred_from_client ) , <EOL> '<STR_LIT>' : <EOL> client_obj_hasher . hexdigest ( ) , <EOL> '<STR_LIT>' : <EOL> str ( bytes_transferred_from_client ) , <EOL> '<STR_LIT>' : str ( fragment_archive_index ) , <EOL> '<STR_LIT>' : policy . ec_scheme_description , <EOL> '<STR_LIT>' : str ( policy . ec_segment_size ) , <EOL> } <EOL> @ ObjectControllerRouter . register ( EC_POLICY ) <EOL> class ECObjectController ( BaseObjectController ) : <EOL> def _fragment_GET_request ( self , req , node_iter , partition , policy ) : <EOL> """<STR_LIT>""" <EOL> backend_headers = self . generate_request_headers ( <EOL> req , additional = req . headers ) <EOL> getter = ResumingGetter ( self . app , req , '<STR_LIT>' , node_iter , <EOL> partition , req . swift_entity_path , <EOL> backend_headers , <EOL> client_chunk_size = policy . fragment_size , <EOL> newest = False ) <EOL> return ( getter , getter . response_parts_iter ( req ) ) <EOL> def _convert_range ( self , req , policy ) : <EOL> """<STR_LIT>""" <EOL> segment_size = policy . ec_segment_size <EOL> fragment_size = policy . fragment_size <EOL> range_specs = [ ] <EOL> new_ranges = [ ] <EOL> for client_start , client_end in req . range . ranges : <EOL> segment_start , segment_end = client_range_to_segment_range ( <EOL> client_start , client_end , segment_size ) <EOL> fragment_start , fragment_end = segment_range_to_fragment_range ( <EOL> segment_start , segment_end , <EOL> segment_size , fragment_size ) <EOL> new_ranges . append ( ( fragment_start , fragment_end ) ) <EOL> range_specs . append ( { '<STR_LIT>' : client_start , <EOL> '<STR_LIT>' : client_end , <EOL> '<STR_LIT>' : segment_start , <EOL> '<STR_LIT>' : segment_end , <EOL> '<STR_LIT>' : fragment_start , <EOL> '<STR_LIT>' : fragment_end } ) <EOL> req . range = "<STR_LIT>" + "<STR_LIT:U+002C>" . join ( <EOL> "<STR_LIT>" % ( s if s is not None else "<STR_LIT>" , <EOL> e if e is not None else "<STR_LIT>" ) <EOL> for s , e in new_ranges ) <EOL> return range_specs <EOL> def _get_or_head_response ( self , req , node_iter , partition , policy ) : <EOL> req . headers . setdefault ( "<STR_LIT>" , <EOL> "<STR_LIT>" ) <EOL> if req . method == '<STR_LIT>' : <EOL> concurrency = policy . ec_ndata if self . app . concurrent_gets else <NUM_LIT:1> <EOL> resp = self . GETorHEAD_base ( <EOL> req , _ ( '<STR_LIT>' ) , node_iter , partition , <EOL> req . swift_entity_path , concurrency ) <EOL> else : <EOL> orig_range = None <EOL> range_specs = [ ] <EOL> if req . range : <EOL> orig_range = req . range <EOL> range_specs = self . _convert_range ( req , policy ) <EOL> safe_iter = GreenthreadSafeIterator ( node_iter ) <EOL> with ContextPool ( policy . ec_ndata ) as pool : <EOL> pile = GreenAsyncPile ( pool ) <EOL> for _junk in range ( policy . ec_ndata ) : <EOL> pile . spawn ( self . _fragment_GET_request , <EOL> req , safe_iter , partition , <EOL> policy ) <EOL> bad_gets = [ ] <EOL> etag_buckets = collections . defaultdict ( list ) <EOL> best_etag = None <EOL> for get , parts_iter in pile : <EOL> if is_success ( get . last_status ) : <EOL> etag = HeaderKeyDict ( <EOL> get . last_headers ) [ '<STR_LIT>' ] <EOL> etag_buckets [ etag ] . append ( ( get , parts_iter ) ) <EOL> if etag != best_etag and ( <EOL> len ( etag_buckets [ etag ] ) > <EOL> len ( etag_buckets [ best_etag ] ) ) : <EOL> best_etag = etag <EOL> else : <EOL> bad_gets . append ( ( get , parts_iter ) ) <EOL> matching_response_count = max ( <EOL> len ( etag_buckets [ best_etag ] ) , len ( bad_gets ) ) <EOL> if ( policy . ec_ndata - matching_response_count > <EOL> pile . _pending ) and node_iter . nodes_left > <NUM_LIT:0> : <EOL> pile . spawn ( self . _fragment_GET_request , req , <EOL> safe_iter , partition , policy ) <EOL> req . range = orig_range <EOL> if len ( etag_buckets [ best_etag ] ) >= policy . ec_ndata : <EOL> resp_headers = HeaderKeyDict ( <EOL> etag_buckets [ best_etag ] [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] . source_headers [ - <NUM_LIT:1> ] ) <EOL> resp_headers . pop ( '<STR_LIT>' , None ) <EOL> eccl = resp_headers . get ( '<STR_LIT>' ) <EOL> obj_length = int ( eccl ) if eccl is not None else None <EOL> fa_length = int ( resp_headers [ '<STR_LIT>' ] ) <EOL> app_iter = ECAppIter ( <EOL> req . swift_entity_path , <EOL> policy , <EOL> [ iterator for getter , iterator in etag_buckets [ best_etag ] ] , <EOL> range_specs , fa_length , obj_length , <EOL> self . app . logger ) <EOL> resp = Response ( <EOL> request = req , <EOL> headers = resp_headers , <EOL> conditional_response = True , <EOL> app_iter = app_iter ) <EOL> try : <EOL> app_iter . kickoff ( req , resp ) <EOL> except HTTPException as err_resp : <EOL> resp = err_resp <EOL> else : <EOL> statuses = [ ] <EOL> reasons = [ ] <EOL> bodies = [ ] <EOL> headers = [ ] <EOL> for getter , body_parts_iter in bad_gets : <EOL> statuses . extend ( getter . statuses ) <EOL> reasons . extend ( getter . reasons ) <EOL> bodies . extend ( getter . bodies ) <EOL> headers . extend ( getter . source_headers ) <EOL> resp = self . best_response ( <EOL> req , statuses , reasons , bodies , '<STR_LIT>' , <EOL> headers = headers ) <EOL> self . _fix_response ( resp ) <EOL> return resp <EOL> def _fix_response ( self , resp ) : <EOL> resp . headers [ '<STR_LIT>' ] = resp . headers . get ( '<STR_LIT>' ) <EOL> if ( is_success ( resp . status_int ) or is_redirection ( resp . status_int ) or <EOL> resp . status_int == HTTP_REQUESTED_RANGE_NOT_SATISFIABLE ) : <EOL> resp . accept_ranges = '<STR_LIT>' <EOL> if is_success ( resp . status_int ) : <EOL> resp . headers [ '<STR_LIT>' ] = resp . headers . get ( <EOL> '<STR_LIT>' ) <EOL> resp . fix_conditional_response ( ) <EOL> def _connect_put_node ( self , node_iter , part , path , headers , <EOL> logger_thread_locals ) : <EOL> """<STR_LIT>""" <EOL> client_cl = headers . pop ( '<STR_LIT>' , None ) <EOL> headers . pop ( '<STR_LIT>' , None ) <EOL> expected_frag_size = None <EOL> if client_cl : <EOL> policy_index = int ( headers . get ( '<STR_LIT>' ) ) <EOL> policy = POLICIES . get_by_index ( policy_index ) <EOL> num_fragments = int ( client_cl ) // policy . ec_segment_size <EOL> expected_frag_size = policy . fragment_size * num_fragments <EOL> last_segment_size = int ( client_cl ) % policy . ec_segment_size <EOL> if last_segment_size : <EOL> last_info = policy . pyeclib_driver . get_segment_info ( <EOL> last_segment_size , policy . ec_segment_size ) <EOL> expected_frag_size += last_info [ '<STR_LIT>' ] <EOL> self . app . logger . thread_locals = logger_thread_locals <EOL> for node in node_iter : <EOL> try : <EOL> putter = ECPutter . connect ( <EOL> node , part , path , headers , <EOL> conn_timeout = self . app . conn_timeout , <EOL> node_timeout = self . app . node_timeout , <EOL> expected_frag_archive_size = expected_frag_size ) <EOL> self . app . set_node_timing ( node , putter . connect_duration ) <EOL> return putter <EOL> except InsufficientStorage : <EOL> self . app . error_limit ( node , _ ( '<STR_LIT>' ) ) <EOL> except PutterConnectError as e : <EOL> self . app . error_occurred ( <EOL> node , _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) % { <EOL> '<STR_LIT:status>' : e . status } ) <EOL> except ( Exception , Timeout ) : <EOL> self . app . exception_occurred ( <EOL> node , _ ( '<STR_LIT>' ) , <EOL> _ ( '<STR_LIT>' ) % path ) <EOL> def _determine_chunk_destinations ( self , putters ) : <EOL> """<STR_LIT>""" <EOL> handoff_conns = [ ] <EOL> chunk_index = { } <EOL> for p in putters : <EOL> if p . node_index is not None : <EOL> chunk_index [ p ] = p . node_index <EOL> else : <EOL> handoff_conns . append ( p ) <EOL> holes = [ x for x in range ( len ( putters ) ) <EOL> if x not in chunk_index . values ( ) ] <EOL> for hole , p in zip ( holes , handoff_conns ) : <EOL> chunk_index [ p ] = hole <EOL> return chunk_index <EOL> def _transfer_data ( self , req , policy , data_source , putters , nodes , <EOL> min_conns , etag_hasher ) : <EOL> """<STR_LIT>""" <EOL> bytes_transferred = <NUM_LIT:0> <EOL> chunk_transform = chunk_transformer ( policy , len ( nodes ) ) <EOL> chunk_transform . send ( None ) <EOL> def send_chunk ( chunk ) : <EOL> if etag_hasher : <EOL> etag_hasher . update ( chunk ) <EOL> backend_chunks = chunk_transform . send ( chunk ) <EOL> if backend_chunks is None : <EOL> return <EOL> for putter in list ( putters ) : <EOL> backend_chunk = backend_chunks [ chunk_index [ putter ] ] <EOL> if not putter . failed : <EOL> putter . chunk_hasher . update ( backend_chunk ) <EOL> putter . send_chunk ( backend_chunk ) <EOL> else : <EOL> putters . remove ( putter ) <EOL> self . _check_min_conn ( <EOL> req , putters , min_conns , msg = '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> try : <EOL> with ContextPool ( len ( putters ) ) as pool : <EOL> chunk_index = self . _determine_chunk_destinations ( putters ) <EOL> for putter in putters : <EOL> putter . spawn_sender_greenthread ( <EOL> pool , self . app . put_queue_depth , self . app . node_timeout , <EOL> self . app . exception_occurred ) <EOL> while True : <EOL> with ChunkReadTimeout ( self . app . client_timeout ) : <EOL> try : <EOL> chunk = next ( data_source ) <EOL> except StopIteration : <EOL> break <EOL> bytes_transferred += len ( chunk ) <EOL> if bytes_transferred > constraints . MAX_FILE_SIZE : <EOL> raise HTTPRequestEntityTooLarge ( request = req ) <EOL> send_chunk ( chunk ) <EOL> if req . content_length and ( <EOL> bytes_transferred < req . content_length ) : <EOL> req . client_disconnect = True <EOL> self . app . logger . warning ( <EOL> _ ( '<STR_LIT>' ) ) <EOL> self . app . logger . increment ( '<STR_LIT>' ) <EOL> raise HTTPClientDisconnect ( request = req ) <EOL> computed_etag = ( etag_hasher . hexdigest ( ) <EOL> if etag_hasher else None ) <EOL> received_etag = req . headers . get ( <EOL> '<STR_LIT>' , '<STR_LIT>' ) . strip ( '<STR_LIT:">' ) <EOL> if ( computed_etag and received_etag and <EOL> computed_etag != received_etag ) : <EOL> raise HTTPUnprocessableEntity ( request = req ) <EOL> send_chunk ( '<STR_LIT>' ) <EOL> for putter in putters : <EOL> trail_md = trailing_metadata ( <EOL> policy , etag_hasher , <EOL> bytes_transferred , <EOL> chunk_index [ putter ] ) <EOL> trail_md [ '<STR_LIT>' ] = putter . chunk_hasher . hexdigest ( ) <EOL> putter . end_of_object_data ( trail_md ) <EOL> for putter in putters : <EOL> putter . wait ( ) <EOL> final_phase = False <EOL> need_quorum = True <EOL> statuses , reasons , bodies , _junk , quorum = self . _get_put_responses ( <EOL> req , putters , len ( nodes ) , final_phase , <EOL> min_conns , need_quorum = need_quorum ) <EOL> if not quorum : <EOL> self . app . logger . error ( <EOL> _ ( '<STR_LIT>' ) , <EOL> statuses . count ( HTTP_CONTINUE ) ) <EOL> raise HTTPServiceUnavailable ( request = req ) <EOL> elif not self . _have_adequate_informational ( <EOL> statuses , min_conns ) : <EOL> resp = self . best_response ( req , statuses , reasons , bodies , <EOL> _ ( '<STR_LIT>' ) , <EOL> quorum_size = min_conns ) <EOL> if is_client_error ( resp . status_int ) : <EOL> raise HTTPServiceUnavailable ( request = req ) <EOL> else : <EOL> raise resp <EOL> for putter in putters : <EOL> putter . send_commit_confirmation ( ) <EOL> for putter in putters : <EOL> putter . wait ( ) <EOL> except ChunkReadTimeout as err : <EOL> self . app . logger . warning ( <EOL> _ ( '<STR_LIT>' ) , err . seconds ) <EOL> self . app . logger . increment ( '<STR_LIT>' ) <EOL> raise HTTPRequestTimeout ( request = req ) <EOL> except ChunkReadError : <EOL> req . client_disconnect = True <EOL> self . app . logger . warning ( <EOL> _ ( '<STR_LIT>' ) ) <EOL> self . app . logger . increment ( '<STR_LIT>' ) <EOL> raise HTTPClientDisconnect ( request = req ) <EOL> except HTTPException : <EOL> raise <EOL> except Timeout : <EOL> self . app . logger . exception ( <EOL> _ ( '<STR_LIT>' ) ) <EOL> raise HTTPClientDisconnect ( request = req ) <EOL> except Exception : <EOL> self . app . logger . exception ( <EOL> _ ( '<STR_LIT>' ) , <EOL> { '<STR_LIT:path>' : req . path } ) <EOL> raise HTTPInternalServerError ( request = req ) <EOL> def _have_adequate_responses ( <EOL> self , statuses , min_responses , conditional_func ) : <EOL> """<STR_LIT>""" <EOL> if sum ( <NUM_LIT:1> for s in statuses if ( conditional_func ( s ) ) ) >= min_responses : <EOL> return True <EOL> return False <EOL> def _have_adequate_successes ( self , statuses , min_responses ) : <EOL> """<STR_LIT>""" <EOL> return self . _have_adequate_responses ( <EOL> statuses , min_responses , is_success ) <EOL> def _have_adequate_informational ( self , statuses , min_responses ) : <EOL> """<STR_LIT>""" <EOL> return self . _have_adequate_responses ( <EOL> statuses , min_responses , is_informational ) <EOL> def _await_response ( self , conn , final_phase ) : <EOL> return conn . await_response ( <EOL> self . app . node_timeout , not final_phase ) <EOL> def _get_conn_response ( self , conn , req , logger_thread_locals , <EOL> final_phase , ** kwargs ) : <EOL> self . app . logger . thread_locals = logger_thread_locals <EOL> try : <EOL> resp = self . _await_response ( conn , final_phase = final_phase , <EOL> ** kwargs ) <EOL> except ( Exception , Timeout ) : <EOL> resp = None <EOL> if final_phase : <EOL> status_type = '<STR_LIT>' <EOL> else : <EOL> status_type = '<STR_LIT>' <EOL> self . app . exception_occurred ( <EOL> conn . node , _ ( '<STR_LIT>' ) , <EOL> _ ( '<STR_LIT>' ) % <EOL> { '<STR_LIT>' : status_type , '<STR_LIT:path>' : req . path } ) <EOL> return ( conn , resp ) <EOL> def _get_put_responses ( self , req , putters , num_nodes , final_phase , <EOL> min_responses , need_quorum = True ) : <EOL> """<STR_LIT>""" <EOL> statuses = [ ] <EOL> reasons = [ ] <EOL> bodies = [ ] <EOL> etags = set ( ) <EOL> pile = GreenAsyncPile ( len ( putters ) ) <EOL> for putter in putters : <EOL> if putter . failed : <EOL> continue <EOL> pile . spawn ( self . _get_conn_response , putter , req , <EOL> self . app . logger . thread_locals , final_phase = final_phase ) <EOL> def _handle_response ( putter , response ) : <EOL> statuses . append ( response . status ) <EOL> reasons . append ( response . reason ) <EOL> if final_phase : <EOL> body = response . read ( ) <EOL> else : <EOL> body = '<STR_LIT>' <EOL> bodies . append ( body ) <EOL> if response . status == HTTP_INSUFFICIENT_STORAGE : <EOL> putter . failed = True <EOL> self . app . error_limit ( putter . node , <EOL> _ ( '<STR_LIT>' ) ) <EOL> elif response . status >= HTTP_INTERNAL_SERVER_ERROR : <EOL> putter . failed = True <EOL> self . app . error_occurred ( <EOL> putter . node , <EOL> _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) % <EOL> { '<STR_LIT:status>' : response . status , <EOL> '<STR_LIT:body>' : body [ : <NUM_LIT> ] , '<STR_LIT:path>' : req . path } ) <EOL> elif is_success ( response . status ) : <EOL> etags . add ( response . getheader ( '<STR_LIT>' ) . strip ( '<STR_LIT:">' ) ) <EOL> quorum = False <EOL> for ( putter , response ) in pile : <EOL> if response : <EOL> _handle_response ( putter , response ) <EOL> if self . _have_adequate_successes ( statuses , min_responses ) : <EOL> break <EOL> else : <EOL> putter . failed = True <EOL> finished_quickly = pile . waitall ( self . app . post_quorum_timeout ) <EOL> for ( putter , response ) in finished_quickly : <EOL> if response : <EOL> _handle_response ( putter , response ) <EOL> if need_quorum : <EOL> if final_phase : <EOL> while len ( statuses ) < num_nodes : <EOL> statuses . append ( HTTP_SERVICE_UNAVAILABLE ) <EOL> reasons . append ( '<STR_LIT>' ) <EOL> bodies . append ( '<STR_LIT>' ) <EOL> else : <EOL> if self . have_quorum ( statuses , num_nodes , quorum = min_responses ) : <EOL> quorum = True <EOL> return statuses , reasons , bodies , etags , quorum <EOL> def _store_object ( self , req , data_source , nodes , partition , <EOL> outgoing_headers ) : <EOL> """<STR_LIT>""" <EOL> policy_index = int ( req . headers . get ( '<STR_LIT>' ) ) <EOL> policy = POLICIES . get_by_index ( policy_index ) <EOL> etag_hasher = md5 ( ) <EOL> min_conns = policy . quorum <EOL> putters = self . _get_put_connections ( <EOL> req , nodes , partition , outgoing_headers , <EOL> policy , expect = True ) <EOL> try : <EOL> self . _check_failure_put_connections ( putters , req , nodes , min_conns ) <EOL> self . _transfer_data ( req , policy , data_source , putters , <EOL> nodes , min_conns , etag_hasher ) <EOL> final_phase = True <EOL> need_quorum = False <EOL> min_conns = policy . quorum <EOL> putters = [ p for p in putters if not p . failed ] <EOL> statuses , reasons , bodies , _etags , _quorum = self . _get_put_responses ( req , putters , len ( nodes ) , <EOL> final_phase , min_conns , <EOL> need_quorum = need_quorum ) <EOL> except HTTPException as resp : <EOL> return resp <EOL> etag = etag_hasher . hexdigest ( ) <EOL> resp = self . best_response ( req , statuses , reasons , bodies , <EOL> _ ( '<STR_LIT>' ) , etag = etag , <EOL> quorum_size = min_conns ) <EOL> resp . last_modified = math . ceil ( <EOL> float ( Timestamp ( req . headers [ '<STR_LIT>' ] ) ) ) <EOL> return resp </s>
