1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
|
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab
#include "compression_onwire.h"
#include "compression_meta.h"
#include "common/dout.h"
#define dout_subsys ceph_subsys_ms
namespace ceph::compression::onwire {
rxtx_t rxtx_t::create_handler_pair(
CephContext* ctx,
const CompConnectionMeta& comp_meta,
std::uint64_t compress_min_size)
{
if (comp_meta.is_compress()) {
CompressorRef compressor = Compressor::create(ctx, comp_meta.get_method());
if (compressor) {
return {std::make_unique<RxHandler>(ctx, compressor),
std::make_unique<TxHandler>(ctx, compressor,
comp_meta.get_mode(),
compress_min_size)};
}
}
return {};
}
std::optional<ceph::bufferlist> TxHandler::compress(const ceph::bufferlist &input)
{
if (m_init_onwire_size < m_min_size) {
ldout(m_cct, 20) << __func__
<< " discovered frame that is smaller than threshold, aborting compression"
<< dendl;
return {};
}
m_compress_potential -= input.length();
ceph::bufferlist out;
if (input.length() == 0) {
ldout(m_cct, 20) << __func__
<< " discovered an empty segment, skipping compression without aborting"
<< dendl;
out.clear();
return out;
}
std::optional<int32_t> compressor_message;
if (m_compressor->compress(input, out, compressor_message)) {
return {};
} else {
ldout(m_cct, 20) << __func__ << " uncompressed.length()=" << input.length()
<< " compressed.length()=" << out.length() << dendl;
m_onwire_size += out.length();
return out;
}
}
std::optional<ceph::bufferlist> RxHandler::decompress(const ceph::bufferlist &input)
{
ceph::bufferlist out;
if (input.length() == 0) {
ldout(m_cct, 20) << __func__
<< " discovered an empty segment, skipping decompression without aborting"
<< dendl;
out.clear();
return out;
}
std::optional<int32_t> compressor_message;
if (m_compressor->decompress(input, out, compressor_message)) {
return {};
} else {
ldout(m_cct, 20) << __func__ << " compressed.length()=" << input.length()
<< " uncompressed.length()=" << out.length() << dendl;
return out;
}
}
void TxHandler::done()
{
ldout(m_cct, 25) << __func__ << " compression ratio=" << get_ratio() << dendl;
}
} // namespace ceph::compression::onwire
|
