1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
2 // vim: ts=8 sw=2 smarttab
4 * Ceph - scalable distributed file system
6 * Copyright (C) 2015 Haomai Wang <haomaiwang@gmail.com>
8 * This is free software; you can redistribute it and/or
9 * modify it under the terms of the GNU Lesser General Public
10 * License version 2.1, as published by the Free Software
11 * Foundation. See file COPYING.
15 #include "common/dout.h"
16 #include "common/errno.h"
17 #include "AsyncCompressor.h"
19 #define dout_subsys ceph_subsys_compressor
21 #define dout_prefix *_dout << "compressor "
23 AsyncCompressor::AsyncCompressor(CephContext *c):
24 compressor(Compressor::create(c, c->_conf->async_compressor_type)), cct(c),
25 compress_tp(cct, "AsyncCompressor::compressor_tp", "tp_async_compr", cct->_conf->async_compressor_threads, "async_compressor_threads"),
26 job_lock("AsyncCompressor::job_lock"),
27 compress_wq(this, c->_conf->async_compressor_thread_timeout, c->_conf->async_compressor_thread_suicide_timeout, &compress_tp) {
30 void AsyncCompressor::init()
32 ldout(cct, 10) << __func__ << dendl;
36 void AsyncCompressor::terminate()
38 ldout(cct, 10) << __func__ << dendl;
42 uint64_t AsyncCompressor::async_compress(bufferlist &data)
44 uint64_t id = ++job_id;
45 pair<unordered_map<uint64_t, Job>::iterator, bool> it;
47 Mutex::Locker l(job_lock);
48 it = jobs.insert(make_pair(id, Job(id, true)));
49 it.first->second.data = data;
51 compress_wq.queue(&it.first->second);
52 ldout(cct, 10) << __func__ << " insert async compress job id=" << id << dendl;
56 uint64_t AsyncCompressor::async_decompress(bufferlist &data)
58 uint64_t id = ++job_id;
59 pair<unordered_map<uint64_t, Job>::iterator, bool> it;
61 Mutex::Locker l(job_lock);
62 it = jobs.insert(make_pair(id, Job(id, false)));
63 it.first->second.data = data;
65 compress_wq.queue(&it.first->second);
66 ldout(cct, 10) << __func__ << " insert async decompress job id=" << id << dendl;
70 int AsyncCompressor::get_compress_data(uint64_t compress_id, bufferlist &data, bool blocking, bool *finished)
73 Mutex::Locker l(job_lock);
74 unordered_map<uint64_t, Job>::iterator it = jobs.find(compress_id);
75 if (it == jobs.end() || !it->second.is_compress) {
76 ldout(cct, 10) << __func__ << " missing to get compress job id=" << compress_id << dendl;
81 auto status = it->second.status.load();
82 if (status == status_t::DONE) {
83 ldout(cct, 20) << __func__ << " successfully getting compressed data, job id=" << compress_id << dendl;
85 data.swap(it->second.data);
87 } else if (status == status_t::ERROR) {
88 ldout(cct, 20) << __func__ << " compressed data failed, job id=" << compress_id << dendl;
91 } else if (blocking) {
92 auto expected = status_t::WAIT;
93 if (it->second.status.compare_exchange_strong(expected, status_t::DONE)) {
94 ldout(cct, 10) << __func__ << " compress job id=" << compress_id << " hasn't finished, abort!"<< dendl;
95 if (compressor->compress(it->second.data, data)) {
96 ldout(cct, 1) << __func__ << " compress job id=" << compress_id << " failed!"<< dendl;
97 it->second.status = status_t::ERROR;
108 ldout(cct, 10) << __func__ << " compress job id=" << compress_id << " hasn't finished."<< dendl;
114 int AsyncCompressor::get_decompress_data(uint64_t decompress_id, bufferlist &data, bool blocking, bool *finished)
117 Mutex::Locker l(job_lock);
118 unordered_map<uint64_t, Job>::iterator it = jobs.find(decompress_id);
119 if (it == jobs.end() || it->second.is_compress) {
120 ldout(cct, 10) << __func__ << " missing to get decompress job id=" << decompress_id << dendl;
124 auto status = it->second.status.load();
125 if (status == status_t::DONE) {
126 ldout(cct, 20) << __func__ << " successfully getting decompressed data, job id=" << decompress_id << dendl;
128 data.swap(it->second.data);
130 } else if (status == status_t::ERROR) {
131 ldout(cct, 20) << __func__ << " compressed data failed, job id=" << decompress_id << dendl;
134 } else if (blocking) {
135 auto expected = status_t::WAIT;
136 if (it->second.status.compare_exchange_strong(expected, status_t::DONE)) {
137 ldout(cct, 10) << __func__ << " decompress job id=" << decompress_id << " hasn't started, abort!"<< dendl;
138 if (compressor->decompress(it->second.data, data)) {
139 ldout(cct, 1) << __func__ << " decompress job id=" << decompress_id << " failed!"<< dendl;
140 it->second.status = status_t::ERROR;
151 ldout(cct, 10) << __func__ << " decompress job id=" << decompress_id << " hasn't finished."<< dendl;