+++ /dev/null
-
-// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
-// vim: ts=8 sw=2 smarttab
-
-#include "rgw_compression.h"
-
-#define dout_subsys ceph_subsys_rgw
-
-//------------RGWPutObj_Compress---------------
-
-int RGWPutObj_Compress::handle_data(bufferlist& bl, off_t ofs, void **phandle, rgw_raw_obj *pobj, bool *again)
-{
- bufferlist in_bl;
- if (*again) {
- return next->handle_data(in_bl, ofs, phandle, pobj, again);
- }
- if (bl.length() > 0) {
- // compression stuff
- if ((ofs > 0 && compressed) || // if previous part was compressed
- (ofs == 0)) { // or it's the first part
- ldout(cct, 10) << "Compression for rgw is enabled, compress part " << bl.length() << dendl;
- int cr = compressor->compress(bl, in_bl);
- if (cr < 0) {
- if (ofs > 0) {
- lderr(cct) << "Compression failed with exit code " << cr
- << " for next part, compression process failed" << dendl;
- return -EIO;
- }
- compressed = false;
- ldout(cct, 5) << "Compression failed with exit code " << cr
- << " for first part, storing uncompressed" << dendl;
- in_bl.claim(bl);
- } else {
- compressed = true;
-
- compression_block newbl;
- size_t bs = blocks.size();
- newbl.old_ofs = ofs;
- newbl.new_ofs = bs > 0 ? blocks[bs-1].len + blocks[bs-1].new_ofs : 0;
- newbl.len = in_bl.length();
- blocks.push_back(newbl);
- }
- } else {
- compressed = false;
- in_bl.claim(bl);
- }
- // end of compression stuff
- }
- return next->handle_data(in_bl, ofs, phandle, pobj, again);
-}
-
-//----------------RGWGetObj_Decompress---------------------
-RGWGetObj_Decompress::RGWGetObj_Decompress(CephContext* cct_,
- RGWCompressionInfo* cs_info_,
- bool partial_content_,
- RGWGetDataCB* next): RGWGetObj_Filter(next),
- cct(cct_),
- cs_info(cs_info_),
- partial_content(partial_content_),
- q_ofs(0),
- q_len(0),
- cur_ofs(0)
-{
- compressor = Compressor::create(cct, cs_info->compression_type);
- if (!compressor.get())
- lderr(cct) << "Cannot load compressor of type " << cs_info->compression_type << dendl;
-}
-
-int RGWGetObj_Decompress::handle_data(bufferlist& bl, off_t bl_ofs, off_t bl_len)
-{
- ldout(cct, 10) << "Compression for rgw is enabled, decompress part "
- << "bl_ofs="<< bl_ofs << bl_len << dendl;
-
- if (!compressor.get()) {
- // if compressor isn't available - error, because cannot return decompressed data?
- lderr(cct) << "Cannot load compressor of type " << cs_info->compression_type << dendl;
- return -EIO;
- }
- bufferlist out_bl, in_bl, temp_in_bl;
- bl.copy(bl_ofs, bl_len, temp_in_bl);
- bl_ofs = 0;
- int r;
- if (waiting.length() != 0) {
- in_bl.append(waiting);
- in_bl.append(temp_in_bl);
- waiting.clear();
- } else {
- in_bl.claim(temp_in_bl);
- }
- bl_len = in_bl.length();
-
- while (first_block <= last_block) {
- bufferlist tmp;
- off_t ofs_in_bl = first_block->new_ofs - cur_ofs;
- if (ofs_in_bl + (off_t)first_block->len > bl_len) {
- // not complete block, put it to waiting
- unsigned tail = bl_len - ofs_in_bl;
- in_bl.copy(ofs_in_bl, tail, waiting);
- cur_ofs -= tail;
- break;
- }
- in_bl.copy(ofs_in_bl, first_block->len, tmp);
- int cr = compressor->decompress(tmp, out_bl);
- if (cr < 0) {
- lderr(cct) << "Compression failed with exit code " << cr << dendl;
- return cr;
- }
- ++first_block;
- while (out_bl.length() - q_ofs >= cct->_conf->rgw_max_chunk_size)
- {
- off_t ch_len = std::min<off_t>(cct->_conf->rgw_max_chunk_size, q_len);
- q_len -= ch_len;
- r = next->handle_data(out_bl, q_ofs, ch_len);
- if (r < 0) {
- lderr(cct) << "handle_data failed with exit code " << r << dendl;
- return r;
- }
- out_bl.splice(0, q_ofs + ch_len);
- q_ofs = 0;
- }
- }
-
- cur_ofs += bl_len;
-
- off_t ch_len = std::min<off_t>(out_bl.length() - q_ofs, q_len);
- r = next->handle_data(out_bl, q_ofs, ch_len);
- if (r < 0) {
- lderr(cct) << "handle_data failed with exit code " << r << dendl;
- return r;
- }
- q_len -= ch_len;
- q_ofs = 0;
- return r;
-}
-
-int RGWGetObj_Decompress::fixup_range(off_t& ofs, off_t& end)
-{
- if (partial_content) {
- // if user set range, we need to calculate it in decompressed data
- first_block = cs_info->blocks.begin(); last_block = cs_info->blocks.begin();
- if (cs_info->blocks.size() > 1) {
- vector<compression_block>::iterator fb, lb;
- // not bad to use auto for lambda, I think
- auto cmp_u = [] (off_t ofs, const compression_block& e) { return (uint64_t)ofs < e.old_ofs; };
- auto cmp_l = [] (const compression_block& e, off_t ofs) { return e.old_ofs <= (uint64_t)ofs; };
- fb = upper_bound(cs_info->blocks.begin()+1,
- cs_info->blocks.end(),
- ofs,
- cmp_u);
- first_block = fb - 1;
- lb = lower_bound(fb,
- cs_info->blocks.end(),
- end,
- cmp_l);
- last_block = lb - 1;
- }
- } else {
- first_block = cs_info->blocks.begin(); last_block = cs_info->blocks.end() - 1;
- }
-
- q_ofs = ofs - first_block->old_ofs;
- q_len = end + 1 - ofs;
-
- ofs = first_block->new_ofs;
- end = last_block->new_ofs + last_block->len - 1;
-
- cur_ofs = ofs;
- waiting.clear();
-
- return next->fixup_range(ofs, end);
-}
Code Review / stor4nfv.git / blobdiff