swift/proxy/controllers/obj.py
def _transfer_data(self, req, policy, data_source, putters, nodes,
                       min_conns, etag_hasher):
        """
        Transfer data for an erasure coded object.

        This method was added in the PUT method extraction change
        """
        bytes_transferred = 0
        # 生成一个迭代器 chunk_transform
        # 第一次调用 chunk_transformer()并不会执行改函数而是生成一个迭代器
        chunk_transform = chunk_transformer(policy)
        chunk_transform.send(None)
        frag_hashers = collections.defaultdict(md5)

        def send_chunk(chunk):
            # Note: there's two different hashers in here. etag_hasher is
            # hashing the original object so that we can validate the ETag
            # that the client sent (and etag_hasher is None if the client
            # didn't send one). The hasher in frag_hashers is hashing the
            # fragment archive being sent to the client; this lets us guard
            # against data corruption on the network between proxy and
            # object server.
            if etag_hasher:
                etag_hasher.update(chunk)  # 对原始数据进行hash
            # 调用 chunk_transform.send() 对数据进行编码
            # backend_chunks 中的每一个元素都对应存到一个 node 上
            backend_chunks = chunk_transform.send(chunk)
            if backend_chunks is None:
                # If there's not enough bytes buffered for erasure-encoding
                # or whatever we're doing, the transform will give us None.
                return

            updated_frag_indexes = set()
            for putter in list(putters):
                frag_index = putter_to_frag_index[putter]
                backend_chunk = backend_chunks[frag_index]
                if not putter.failed:
                    # N.B. same frag_index will appear when using
                    # ec_duplication_factor >= 2. So skip to feed the chunk
                    # to hasher if the frag was updated already.
                    if frag_index not in updated_frag_indexes:
                        frag_hashers[frag_index].update(backend_chunk) # hash编码后的数据
                        updated_frag_indexes.add(frag_index)
                    putter.send_chunk(backend_chunk)  # put 数据到 putter 的队列中
                else:
                    putter.close()
                    putters.remove(putter)
            self._check_min_conn(
                req, putters, min_conns,
                msg=_('Object PUT exceptions during send, '
                      '%(conns)s/%(nodes)s required connections'))

        try:
            # ContexPool 继承了 GreenPool，实现了方法 __enter__() 和 __exit__()
            # 紧跟 with 后面的语句被执行后，调用 __enter__() 方法，返回值赋值给 as 后面的变量
            # 当 with 后面的代码块全部被执行完之后，将调用 __exit__() 方法
            with ContextPool(len(putters)) as pool:

                # build our putter_to_frag_index dict to place handoffs in the
                # same part nodes index as the primaries they are covering
                putter_to_frag_index = self._determine_chunk_destinations(
                    putters, policy)

                # 调用 _send_file 从 putter 的队列中读取数据发送出去
                for putter in putters:
                    putter.spawn_sender_greenthread(
                        pool, self.app.put_queue_depth, self.app.node_timeout,
                        self.app.exception_occurred)
                while True:
                    with ChunkReadTimeout(self.app.client_timeout):
                        try:
                            # 每次读取 client_chunk_size 字节的数据到chunk中
                            chunk = next(data_source)
                        except StopIteration:
                            break
                    bytes_transferred += len(chunk)
                    if bytes_transferred > constraints.MAX_FILE_SIZE:
                        raise HTTPRequestEntityTooLarge(request=req)

                    send_chunk(chunk)

                if req.content_length and (
                        bytes_transferred < req.content_length):
                    req.client_disconnect = True
                    self.app.logger.warning(
                        _('Client disconnected without sending enough data'))
                    self.app.logger.increment('client_disconnects')
                    raise HTTPClientDisconnect(request=req)

                computed_etag = (etag_hasher.hexdigest()
                                 if etag_hasher else None)
                received_etag = req.headers.get(
                    'etag', '').strip('"')
                if (computed_etag and received_etag and
                   computed_etag != received_etag):
                    raise HTTPUnprocessableEntity(request=req)

                send_chunk('')  # flush out any buffered data

                footers = self._get_footers(req)
                received_etag = footers.get(
                    'etag', '').strip('"')
                if (computed_etag and received_etag and
                   computed_etag != received_etag):
                    raise HTTPUnprocessableEntity(request=req)

                # Remove any EC reserved metadata names from footers
                footers = {(k, v) for k, v in footers.items()
                           if not k.lower().startswith('x-object-sysmeta-ec-')}
                for putter in putters:
                    frag_index = putter_to_frag_index[putter]
                    # Update any footers set by middleware with EC footers
                    trail_md = trailing_metadata(
                        policy, etag_hasher,
                        bytes_transferred, frag_index)
                    trail_md.update(footers)
                    # Etag footer must always be hash of what we sent
                    trail_md['Etag'] = frag_hashers[frag_index].hexdigest()
                    putter.end_of_object_data(footer_metadata=trail_md)

                for putter in putters:
                    putter.wait()

                # for storage policies requiring 2-phase commit (e.g.
                # erasure coding), enforce >= 'quorum' number of
                # 100-continue responses - this indicates successful
                # object data and metadata commit and is a necessary
                # condition to be met before starting 2nd PUT phase
                final_phase = False
                statuses, reasons, bodies, _junk = \
                    self._get_put_responses(
                        req, putters, len(nodes), final_phase=final_phase,
                        min_responses=min_conns)
                if not self.have_quorum(
                        statuses, len(nodes), quorum=min_conns):
                    self.app.logger.error(
                        _('Not enough object servers ack\'ed (got %d)'),
                        statuses.count(HTTP_CONTINUE))
                    raise HTTPServiceUnavailable(request=req)

                elif not self._have_adequate_informational(
                        statuses, min_conns):
                    resp = self.best_response(req, statuses, reasons, bodies,
                                              _('Object PUT'),
                                              quorum_size=min_conns)
                    if is_client_error(resp.status_int):
                        # if 4xx occurred in this state it is absolutely
                        # a bad conversation between proxy-server and
                        # object-server (even if it's
                        # HTTP_UNPROCESSABLE_ENTITY) so we should regard this
                        # as HTTPServiceUnavailable.
                        raise HTTPServiceUnavailable(request=req)
                    else:
                        # Other errors should use raw best_response
                        raise resp

                # quorum achieved, start 2nd phase - send commit
                # confirmation to participating object servers
                # so they write a .durable state file indicating
                # a successful PUT
                for putter in putters:
                    putter.send_commit_confirmation()
                for putter in putters:
                    putter.wait()
        except ChunkReadTimeout as err:
            self.app.logger.warning(
                _('ERROR Client read timeout (%ss)'), err.seconds)
            self.app.logger.increment('client_timeouts')
            raise HTTPRequestTimeout(request=req)
        except ChunkReadError:
            req.client_disconnect = True
            self.app.logger.warning(
                _('Client disconnected without sending last chunk'))
            self.app.logger.increment('client_disconnects')
            raise HTTPClientDisconnect(request=req)
        except HTTPException:
            raise
        except Timeout:
            self.app.logger.exception(
                _('ERROR Exception causing client disconnect'))
            raise HTTPClientDisconnect(request=req)
        except Exception:
            self.app.logger.exception(
                _('ERROR Exception transferring data to object servers %s'),
                {'path': req.path})
            raise HTTPInternalServerError(request=req)

        Transfer data for an erasure coded object.

        This method was added in the PUT method extraction change
        
Transfer data for an erasure coded object.
This method was added in the PUT method extraction change
Transfer data for an erasure coded object . This method was added in the PUT method extraction change
def _transfer_data(self, req, policy, data_source, putters, nodes,
                       min_conns, etag_hasher):
        bytes_transferred = 0
        chunk_transform = chunk_transformer(policy)
        chunk_transform.send(None)
        frag_hashers = collections.defaultdict(md5)
        def send_chunk(chunk):
            if etag_hasher:
                etag_hasher.update(chunk)  
            backend_chunks = chunk_transform.send(chunk)
            if backend_chunks is None:
                return
            updated_frag_indexes = set()
            for putter in list(putters):
                frag_index = putter_to_frag_index[putter]
                backend_chunk = backend_chunks[frag_index]
                if not putter.failed:
                    if frag_index not in updated_frag_indexes:
                        frag_hashers[frag_index].update(backend_chunk) 
                        updated_frag_indexes.add(frag_index)
                    putter.send_chunk(backend_chunk)  
                else:
                    putter.close()
                    putters.remove(putter)
            self._check_min_conn(
                req, putters, min_conns,
                msg=_('Object PUT exceptions during send, '
                      '%(conns)s/%(nodes)s required connections'))
        try:
            with ContextPool(len(putters)) as pool:
                putter_to_frag_index = self._determine_chunk_destinations(
                    putters, policy)
                for putter in putters:
                    putter.spawn_sender_greenthread(
                        pool, self.app.put_queue_depth, self.app.node_timeout,
                        self.app.exception_occurred)
                while True:
                    with ChunkReadTimeout(self.app.client_timeout):
                        try:
                            chunk = next(data_source)
                        except StopIteration:
                            break
                    bytes_transferred += len(chunk)
                    if bytes_transferred > constraints.MAX_FILE_SIZE:
                        raise HTTPRequestEntityTooLarge(request=req)
                    send_chunk(chunk)
                if req.content_length and (
                        bytes_transferred < req.content_length):
                    req.client_disconnect = True
                    self.app.logger.warning(
                        _('Client disconnected without sending enough data'))
                    self.app.logger.increment('client_disconnects')
                    raise HTTPClientDisconnect(request=req)
                computed_etag = (etag_hasher.hexdigest()
                                 if etag_hasher else None)
                received_etag = req.headers.get(
                    'etag', '').strip('"')
                if (computed_etag and received_etag and
                   computed_etag != received_etag):
                    raise HTTPUnprocessableEntity(request=req)
                send_chunk('')  
                footers = self._get_footers(req)
                received_etag = footers.get(
                    'etag', '').strip('"')
                if (computed_etag and received_etag and
                   computed_etag != received_etag):
                    raise HTTPUnprocessableEntity(request=req)
                footers = {(k, v) for k, v in footers.items()
                           if not k.lower().startswith('x-object-sysmeta-ec-')}
                for putter in putters:
                    frag_index = putter_to_frag_index[putter]
                    trail_md = trailing_metadata(
                        policy, etag_hasher,
                        bytes_transferred, frag_index)
                    trail_md.update(footers)
                    trail_md['Etag'] = frag_hashers[frag_index].hexdigest()
                    putter.end_of_object_data(footer_metadata=trail_md)
                for putter in putters:
                    putter.wait()
                final_phase = False
                statuses, reasons, bodies, _junk = \
                    self._get_put_responses(
                        req, putters, len(nodes), final_phase=final_phase,
                        min_responses=min_conns)
                if not self.have_quorum(
                        statuses, len(nodes), quorum=min_conns):
                    self.app.logger.error(
                        _('Not enough object servers ack\'ed (got %d)'),
                        statuses.count(HTTP_CONTINUE))
                    raise HTTPServiceUnavailable(request=req)
                elif not self._have_adequate_informational(
                        statuses, min_conns):
                    resp = self.best_response(req, statuses, reasons, bodies,
                                              _('Object PUT'),
                                              quorum_size=min_conns)
                    if is_client_error(resp.status_int):
                        raise HTTPServiceUnavailable(request=req)
                    else:
                        raise resp
                for putter in putters:
                    putter.send_commit_confirmation()
                for putter in putters:
                    putter.wait()
        except ChunkReadTimeout as err:
            self.app.logger.warning(
                _('ERROR Client read timeout (%ss)'), err.seconds)
            self.app.logger.increment('client_timeouts')
            raise HTTPRequestTimeout(request=req)
        except ChunkReadError:
            req.client_disconnect = True
            self.app.logger.warning(
                _('Client disconnected without sending last chunk'))
            self.app.logger.increment('client_disconnects')
            raise HTTPClientDisconnect(request=req)
        except HTTPException:
            raise
        except Timeout:
            self.app.logger.exception(
                _('ERROR Exception causing client disconnect'))
            raise HTTPClientDisconnect(request=req)
        except Exception:
            self.app.logger.exception(
                _('ERROR Exception transferring data to object servers %s'),
                {'path': req.path})
            raise HTTPInternalServerError(request=req)
def _transfer_data ( self , req , policy , data_source , putters , nodes , min_conns , etag_hasher ) : bytes_transferred = 0 chunk_transform = chunk_transformer ( policy ) chunk_transform . send ( None ) frag_hashers = collections . defaultdict ( md5 ) def send_chunk ( chunk ) : if etag_hasher : etag_hasher . update ( chunk ) backend_chunks = chunk_transform . send ( chunk ) if backend_chunks is None : return updated_frag_indexes = set ( ) for putter in list ( putters ) : frag_index = putter_to_frag_index [ putter ] backend_chunk = backend_chunks [ frag_index ] if not putter . failed : if frag_index not in updated_frag_indexes : frag_hashers [ frag_index ] . update ( backend_chunk ) updated_frag_indexes . add ( frag_index ) putter . send_chunk ( backend_chunk ) else : putter . close ( ) putters . remove ( putter ) self . _check_min_conn ( req , putters , min_conns , msg = _ ( 'Object PUT exceptions during send, ' '%(conns)s/%(nodes)s required connections' ) ) try : with ContextPool ( len ( putters ) ) as pool : putter_to_frag_index = self . _determine_chunk_destinations ( putters , policy ) for putter in putters : putter . spawn_sender_greenthread ( pool , self . app . put_queue_depth , self . app . node_timeout , self . app . exception_occurred ) while True : with ChunkReadTimeout ( self . app . client_timeout ) : try : chunk = next ( data_source ) except StopIteration : break bytes_transferred += len ( chunk ) if bytes_transferred > constraints . MAX_FILE_SIZE : raise HTTPRequestEntityTooLarge ( request = req ) send_chunk ( chunk ) if req . content_length and ( bytes_transferred < req . content_length ) : req . client_disconnect = True self . app . logger . warning ( _ ( 'Client disconnected without sending enough data' ) ) self . app . logger . increment ( 'client_disconnects' ) raise HTTPClientDisconnect ( request = req ) computed_etag = ( etag_hasher . hexdigest ( ) if etag_hasher else None ) received_etag = req . headers . get ( 'etag' , '' ) . strip ( '"' ) if ( computed_etag and received_etag and computed_etag != received_etag ) : raise HTTPUnprocessableEntity ( request = req ) send_chunk ( '' ) footers = self . _get_footers ( req ) received_etag = footers . get ( 'etag' , '' ) . strip ( '"' ) if ( computed_etag and received_etag and computed_etag != received_etag ) : raise HTTPUnprocessableEntity ( request = req ) footers = { ( k , v ) for k , v in footers . items ( ) if not k . lower ( ) . startswith ( 'x-object-sysmeta-ec-' ) } for putter in putters : frag_index = putter_to_frag_index [ putter ] trail_md = trailing_metadata ( policy , etag_hasher , bytes_transferred , frag_index ) trail_md . update ( footers ) trail_md [ 'Etag' ] = frag_hashers [ frag_index ] . hexdigest ( ) putter . end_of_object_data ( footer_metadata = trail_md ) for putter in putters : putter . wait ( ) final_phase = False statuses , reasons , bodies , _junk = self . _get_put_responses ( req , putters , len ( nodes ) , final_phase = final_phase , min_responses = min_conns ) if not self . have_quorum ( statuses , len ( nodes ) , quorum = min_conns ) : self . app . logger . error ( _ ( 'Not enough object servers ack\'ed (got %d)' ) , statuses . count ( HTTP_CONTINUE ) ) raise HTTPServiceUnavailable ( request = req ) elif not self . _have_adequate_informational ( statuses , min_conns ) : resp = self . best_response ( req , statuses , reasons , bodies , _ ( 'Object PUT' ) , quorum_size = min_conns ) if is_client_error ( resp . status_int ) : raise HTTPServiceUnavailable ( request = req ) else : raise resp for putter in putters : putter . send_commit_confirmation ( ) for putter in putters : putter . wait ( ) except ChunkReadTimeout as err : self . app . logger . warning ( _ ( 'ERROR Client read timeout (%ss)' ) , err . seconds ) self . app . logger . increment ( 'client_timeouts' ) raise HTTPRequestTimeout ( request = req ) except ChunkReadError : req . client_disconnect = True self . app . logger . warning ( _ ( 'Client disconnected without sending last chunk' ) ) self . app . logger . increment ( 'client_disconnects' ) raise HTTPClientDisconnect ( request = req ) except HTTPException : raise except Timeout : self . app . logger . exception ( _ ( 'ERROR Exception causing client disconnect' ) ) raise HTTPClientDisconnect ( request = req ) except Exception : self . app . logger . exception ( _ ( 'ERROR Exception transferring data to object servers %s' ) , { 'path' : req . path } ) raise HTTPInternalServerError ( request = req )
Transfer data for an erasure coded object.
Transfer data for an erasure coded object .
"""
        Transfer data for an erasure coded object.

        This method was added in the PUT method extraction change
        """ # 生成一个迭代器 chunk_transform # 第一次调用 chunk_transformer()并不会执行改函数而是生成一个迭代器 # Note: there's two different hashers in here. etag_hasher is # hashing the original object so that we can validate the ETag # that the client sent (and etag_hasher is None if the client # didn't send one). The hasher in frag_hashers is hashing the # fragment archive being sent to the client; this lets us guard # against data corruption on the network between proxy and # object server. # 对原始数据进行hash # 调用 chunk_transform.send() 对数据进行编码 # backend_chunks 中的每一个元素都对应存到一个 node 上 # If there's not enough bytes buffered for erasure-encoding # or whatever we're doing, the transform will give us None. # N.B. same frag_index will appear when using # ec_duplication_factor >= 2. So skip to feed the chunk # to hasher if the frag was updated already. # hash编码后的数据 # put 数据到 putter 的队列中 # ContexPool 继承了 GreenPool，实现了方法 __enter__() 和 __exit__() # 紧跟 with 后面的语句被执行后，调用 __enter__() 方法，返回值赋值给 as 后面的变量 # 当 with 后面的代码块全部被执行完之后，将调用 __exit__() 方法 # build our putter_to_frag_index dict to place handoffs in the # same part nodes index as the primaries they are covering # 调用 _send_file 从 putter 的队列中读取数据发送出去 # 每次读取 client_chunk_size 字节的数据到chunk中 # flush out any buffered data # Remove any EC reserved metadata names from footers # Update any footers set by middleware with EC footers # Etag footer must always be hash of what we sent # for storage policies requiring 2-phase commit (e.g. # erasure coding), enforce >= 'quorum' number of # 100-continue responses - this indicates successful # object data and metadata commit and is a necessary # condition to be met before starting 2nd PUT phase # if 4xx occurred in this state it is absolutely # a bad conversation between proxy-server and # object-server (even if it's # HTTP_UNPROCESSABLE_ENTITY) so we should regard this # as HTTPServiceUnavailable. # Other errors should use raw best_response # quorum achieved, start 2nd phase - send commit # confirmation to participating object servers # so they write a .durable state file indicating # a successful PUT
self
req
policy
data_source
putters
nodes
min_conns
etag_hasher