swift/proxy/controllers/obj.py
def _store_object(self, req, data_source, nodes, partition,
                      outgoing_headers):
        """
        Store an erasure coded object.
        """
        policy_index = int(req.headers.get('X-Backend-Storage-Policy-Index'))
        policy = POLICIES.get_by_index(policy_index)

        expected_frag_size = None
        if req.content_length:
            # TODO: PyECLib <= 1.2.0 looks to return the segment info
            # different from the input for aligned data efficiency but
            # Swift never does. So calculate the fragment length Swift
            # will actually send to object server by making two different
            # get_segment_info calls (until PyECLib fixed).
            # policy.fragment_size makes the call using segment size,
            # and the next call is to get info for the last segment

            # get number of fragments except the tail - use truncation //
            # req.content_length: object size in Byte
            # policy.ec_segment_size: configured in swfit.conf, 1048576 by default
            # a segment is a series of consecutive HTTP chunks buffered up
            # ...before performing an EC operation
            num_fragments = req.content_length // policy.ec_segment_size
            # fragment: Data and parity ‘fragments’ are generated
            # ...when erasure coding transformation is applied to a segment.
            expected_frag_size = policy.fragment_size * num_fragments

            # calculate the tail fragment_size by hand and add it to
            # expected_frag_size
            last_segment_size = req.content_length % policy.ec_segment_size
            if last_segment_size:
                # get_segment_info(data_len, segment_size):
                # returns a dict with the following keys:
                # fragment_size: the fragment size returned by encode()
                # ...
                last_info = policy.pyeclib_driver.get_segment_info(
                    last_segment_size, policy.ec_segment_size)
                expected_frag_size += last_info['fragment_size']
        for headers in outgoing_headers:
            headers['X-Backend-Obj-Content-Length'] = expected_frag_size
            # the object server will get different bytes, so these
            # values do not apply.
            headers.pop('Content-Length', None)
            headers.pop('Etag', None)

        # Since the request body sent from client -> proxy is not
        # the same as the request body sent proxy -> object, we
        # can't rely on the object-server to do the etag checking -
        # so we have to do it here.
        etag_hasher = md5()

        min_conns = policy.quorum
        putters = self._get_put_connections(
            req, nodes, partition, outgoing_headers, policy)

        try:
            # check that a minimum number of connections were established and
            # meet all the correct conditions set in the request
            self._check_failure_put_connections(putters, req, min_conns)

            self._transfer_data(req, policy, data_source, putters,
                                nodes, min_conns, etag_hasher)
            # The durable state will propagate in a replicated fashion; if
            # one fragment is durable then the reconstructor will spread the
            # durable status around.
            # In order to avoid successfully writing an object, but refusing
            # to serve it on a subsequent GET because don't have enough
            # durable data fragments - we require the same number of durable
            # writes as quorum fragment writes.  If object servers are in the
            # future able to serve their non-durable fragment archives we may
            # be able to reduce this quorum count if needed.
            # ignore response etags
            statuses, reasons, bodies, _etags = \
                self._get_put_responses(req, putters, len(nodes),
                                        final_phase=True,
                                        min_responses=min_conns)
        except HTTPException as resp:
            return resp
        finally:
            for putter in putters:
                putter.close()

        etag = etag_hasher.hexdigest()
        resp = self.best_response(req, statuses, reasons, bodies,
                                  _('Object PUT'), etag=etag,
                                  quorum_size=min_conns)
        resp.last_modified = math.ceil(
            float(Timestamp(req.headers['X-Timestamp'])))
        return resp

        Store an erasure coded object.
        
Store an erasure coded object.
Store an erasure coded object .
def _store_object(self, req, data_source, nodes, partition,
                      outgoing_headers):
        policy_index = int(req.headers.get('X-Backend-Storage-Policy-Index'))
        policy = POLICIES.get_by_index(policy_index)
        expected_frag_size = None
        if req.content_length:
            num_fragments = req.content_length // policy.ec_segment_size
            expected_frag_size = policy.fragment_size * num_fragments
            last_segment_size = req.content_length % policy.ec_segment_size
            if last_segment_size:
                last_info = policy.pyeclib_driver.get_segment_info(
                    last_segment_size, policy.ec_segment_size)
                expected_frag_size += last_info['fragment_size']
        for headers in outgoing_headers:
            headers['X-Backend-Obj-Content-Length'] = expected_frag_size
            headers.pop('Content-Length', None)
            headers.pop('Etag', None)
        etag_hasher = md5()
        min_conns = policy.quorum
        putters = self._get_put_connections(
            req, nodes, partition, outgoing_headers, policy)
        try:
            self._check_failure_put_connections(putters, req, min_conns)
            self._transfer_data(req, policy, data_source, putters,
                                nodes, min_conns, etag_hasher)
            statuses, reasons, bodies, _etags = \
                self._get_put_responses(req, putters, len(nodes),
                                        final_phase=True,
                                        min_responses=min_conns)
        except HTTPException as resp:
            return resp
        finally:
            for putter in putters:
                putter.close()
        etag = etag_hasher.hexdigest()
        resp = self.best_response(req, statuses, reasons, bodies,
                                  _('Object PUT'), etag=etag,
                                  quorum_size=min_conns)
        resp.last_modified = math.ceil(
            float(Timestamp(req.headers['X-Timestamp'])))
        return resp
def _store_object ( self , req , data_source , nodes , partition , outgoing_headers ) : policy_index = int ( req . headers . get ( 'X-Backend-Storage-Policy-Index' ) ) policy = POLICIES . get_by_index ( policy_index ) expected_frag_size = None if req . content_length : num_fragments = req . content_length // policy . ec_segment_size expected_frag_size = policy . fragment_size * num_fragments last_segment_size = req . content_length % policy . ec_segment_size if last_segment_size : last_info = policy . pyeclib_driver . get_segment_info ( last_segment_size , policy . ec_segment_size ) expected_frag_size += last_info [ 'fragment_size' ] for headers in outgoing_headers : headers [ 'X-Backend-Obj-Content-Length' ] = expected_frag_size headers . pop ( 'Content-Length' , None ) headers . pop ( 'Etag' , None ) etag_hasher = md5 ( ) min_conns = policy . quorum putters = self . _get_put_connections ( req , nodes , partition , outgoing_headers , policy ) try : self . _check_failure_put_connections ( putters , req , min_conns ) self . _transfer_data ( req , policy , data_source , putters , nodes , min_conns , etag_hasher ) statuses , reasons , bodies , _etags = self . _get_put_responses ( req , putters , len ( nodes ) , final_phase = True , min_responses = min_conns ) except HTTPException as resp : return resp finally : for putter in putters : putter . close ( ) etag = etag_hasher . hexdigest ( ) resp = self . best_response ( req , statuses , reasons , bodies , _ ( 'Object PUT' ) , etag = etag , quorum_size = min_conns ) resp . last_modified = math . ceil ( float ( Timestamp ( req . headers [ 'X-Timestamp' ] ) ) ) return resp
Store an erasure coded object.
Store an erasure coded object .
"""
        Store an erasure coded object.
        """ # TODO: PyECLib <= 1.2.0 looks to return the segment info # different from the input for aligned data efficiency but # Swift never does. So calculate the fragment length Swift # will actually send to object server by making two different # get_segment_info calls (until PyECLib fixed). # policy.fragment_size makes the call using segment size, # and the next call is to get info for the last segment # get number of fragments except the tail - use truncation // # req.content_length: object size in Byte # policy.ec_segment_size: configured in swfit.conf, 1048576 by default # a segment is a series of consecutive HTTP chunks buffered up # ...before performing an EC operation # fragment: Data and parity ‘fragments’ are generated # ...when erasure coding transformation is applied to a segment. # calculate the tail fragment_size by hand and add it to # expected_frag_size # get_segment_info(data_len, segment_size): # returns a dict with the following keys: # fragment_size: the fragment size returned by encode() # ... # the object server will get different bytes, so these # values do not apply. # Since the request body sent from client -> proxy is not # the same as the request body sent proxy -> object, we # can't rely on the object-server to do the etag checking - # so we have to do it here. # check that a minimum number of connections were established and # meet all the correct conditions set in the request # The durable state will propagate in a replicated fashion; if # one fragment is durable then the reconstructor will spread the # durable status around. # In order to avoid successfully writing an object, but refusing # to serve it on a subsequent GET because don't have enough # durable data fragments - we require the same number of durable # writes as quorum fragment writes.  If object servers are in the # future able to serve their non-durable fragment archives we may # be able to reduce this quorum count if needed. # ignore response etags
self
req
data_source
nodes
partition
outgoing_headers