--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2015 Haomai Wang <haomaiwang@gmail.com>
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation. See file COPYING.
+ *
+ */
+
+#ifndef CEPH_ASYNCCOMPRESSOR_H
+#define CEPH_ASYNCCOMPRESSOR_H
+
+#include <deque>
+#include <vector>
+#include <atomic>
+
+#include "include/str_list.h"
+
+#include "Compressor.h"
+#include "common/WorkQueue.h"
+
+class AsyncCompressor {
+ private:
+ CompressorRef compressor;
+ CephContext *cct;
+ std::atomic<uint64_t> job_id { 0 };
+ vector<int> coreids;
+ ThreadPool compress_tp;
+
+ enum class status_t {
+ WAIT,
+ WORKING,
+ DONE,
+ ERROR
+ };
+
+ struct Job {
+ uint64_t id;
+ std::atomic<status_t> status { status_t::WAIT };
+ bool is_compress;
+ bufferlist data;
+ Job(uint64_t i, bool compress): id(i), is_compress(compress) {}
+ Job(const Job &j): id(j.id), status(j.status.load()), is_compress(j.is_compress), data(j.data) {}
+ };
+ Mutex job_lock;
+ // only when job.status == DONE && with job_lock holding, we can insert/erase element in jobs
+ // only when job.status == WAIT && with pool_lock holding, you can change its status and modify element's info later
+ unordered_map<uint64_t, Job> jobs;
+
+ struct CompressWQ : public ThreadPool::WorkQueue<Job> {
+ typedef AsyncCompressor::Job Job;
+ AsyncCompressor *async_compressor;
+ deque<Job*> job_queue;
+
+ CompressWQ(AsyncCompressor *ac, time_t timeout, time_t suicide_timeout, ThreadPool *tp)
+ : ThreadPool::WorkQueue<Job>("AsyncCompressor::CompressWQ", timeout, suicide_timeout, tp), async_compressor(ac) {}
+
+ bool _enqueue(Job *item) override {
+ job_queue.push_back(item);
+ return true;
+ }
+ void _dequeue(Job *item) override {
+ ceph_abort();
+ }
+ bool _empty() override {
+ return job_queue.empty();
+ }
+ Job* _dequeue() override {
+ if (job_queue.empty())
+ return NULL;
+ Job *item = NULL;
+ while (!job_queue.empty()) {
+ item = job_queue.front();
+ job_queue.pop_front();
+
+ auto expected = status_t::WAIT;
+ if (item->status.compare_exchange_strong(expected, status_t::WORKING)) {
+ break;
+ } else {
+ Mutex::Locker l(async_compressor->job_lock);
+ async_compressor->jobs.erase(item->id);
+ item = NULL;
+ }
+ }
+ return item;
+ }
+ void _process(Job *item, ThreadPool::TPHandle &) override {
+ assert(item->status == status_t::WORKING);
+ bufferlist out;
+ int r;
+ if (item->is_compress)
+ r = async_compressor->compressor->compress(item->data, out);
+ else
+ r = async_compressor->compressor->decompress(item->data, out);
+ if (!r) {
+ item->data.swap(out);
+ auto expected = status_t::WORKING;
+ assert(item->status.compare_exchange_strong(expected, status_t::DONE));
+ } else {
+ item->status = status_t::ERROR;
+ }
+ }
+ void _process_finish(Job *item) override {}
+ void _clear() override {}
+ } compress_wq;
+ friend class CompressWQ;
+ void _compress(bufferlist &in, bufferlist &out);
+ void _decompress(bufferlist &in, bufferlist &out);
+
+ public:
+ explicit AsyncCompressor(CephContext *c);
+ virtual ~AsyncCompressor() {}
+
+ int get_cpuid(int id) {
+ if (coreids.empty())
+ return -1;
+ return coreids[id % coreids.size()];
+ }
+
+ void init();
+ void terminate();
+ uint64_t async_compress(bufferlist &data);
+ uint64_t async_decompress(bufferlist &data);
+ int get_compress_data(uint64_t compress_id, bufferlist &data, bool blocking, bool *finished);
+ int get_decompress_data(uint64_t decompress_id, bufferlist &data, bool blocking, bool *finished);
+};
+
+#endif