curl

Author: danielshaving

example/curl/curl.cc

@@ -0,0 +1,283 @@
+#include "httpcurl.h"
+
+static void dummy(const std::shared_ptr<Channel> &)
+{
+
+}
+
+Request::Request(Curl *owner, const char *url, const char *data, size_t len)
+: owner(owner),
+  curl(curl_easy_init())
+{
+	setopt(CURLOPT_URL, url);
+	setopt(CURLOPT_WRITEFUNCTION, &Request::writeData);
+	setopt(CURLOPT_WRITEDATA, this);
+	setopt(CURLOPT_HEADERFUNCTION, &Request::headerData);
+	setopt(CURLOPT_HEADERDATA, this);
+	setopt(CURLOPT_PRIVATE, this);
+	setopt(CURLOPT_USERAGENT, "curl");
+	setopt(CURLOPT_POST, 1L);
+	setopt(CURLOPT_POSTFIELDS, data);  
+	setopt(CURLOPT_POSTFIELDSIZE, len);
+	
+	LOG_DEBUG << curl << " " << url;
+	curl_multi_add_handle(owner->getCurlm(), curl);
+}
+
+Request::Request(Curl *owner, const char *url)
+  : owner(owner),
+    curl(curl_easy_init())
+{
+	setopt(CURLOPT_URL, url);
+	setopt(CURLOPT_WRITEFUNCTION, &Request::writeData);
+	setopt(CURLOPT_WRITEDATA, this);
+	setopt(CURLOPT_HEADERFUNCTION, &Request::headerData);
+	setopt(CURLOPT_HEADERDATA, this);
+	setopt(CURLOPT_PRIVATE, this);
+	setopt(CURLOPT_USERAGENT, "curl");
+
+	LOG_DEBUG << curl << " " << url;
+	curl_multi_add_handle(owner->getCurlm(), curl);
+}
+
+Request::~Request()
+{
+	assert(!channel || channel->isNoneEvent());
+	curl_multi_remove_handle(owner->getCurlm(), curl);
+	curl_easy_cleanup(curl);
+}
+
+void Request::headerOnly()
+{
+	setopt(CURLOPT_NOBODY, 1);
+}
+
+void Request::setRange(const std::string_view range)
+{
+	setopt(CURLOPT_RANGE, range.data());
+}
+
+const char *Request::getEffectiveUrl()
+{
+	const char *p = nullptr;
+	curl_easy_getinfo(curl, CURLINFO_EFFECTIVE_URL, &p);
+	return p;
+}
+
+const char *Request::getRedirectUrl()
+{
+	const char *p = nullptr;
+	curl_easy_getinfo(curl, CURLINFO_REDIRECT_URL, &p);
+	return p;
+}
+
+int Request::getResponseCode()
+{
+	long code = 0;
+	curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &code);
+	return static_cast<int>(code);
+}
+
+Channel *Request::setChannel(int fd)
+{
+	assert(channel.get() == nullptr);
+	channel.reset(new Channel(owner->getLoop(), fd));
+	channel->setTie(shared_from_this());
+	return channel.get();
+}
+
+void Request::removeChannel()
+{
+	channel->disableAll();
+	channel->remove();
+	owner->getLoop()->queueInLoop(std::bind(dummy, channel));
+	channel.reset();
+}
+
+void Request::done(int code)
+{
+	if (doneCb)
+	{
+		doneCb(this, code);
+	}
+}
+
+void Request::dataCallback(const char *buffer, int len)
+{
+	if (dataCb)
+	{
+		dataCb(this, buffer, len);
+	}
+}
+
+void Request::headerCallback(const char *buffer, int len)
+{
+	if (headerCb)
+	{
+		headerCb(buffer, len);
+	}
+}
+
+size_t Request::writeData(char *buffer, size_t size, size_t nmemb, void *userp)
+{
+	assert(size == 1);
+	Request *req = static_cast<Request*>(userp);
+	req->dataCallback(buffer, static_cast<int>(nmemb));
+	return nmemb;
+}
+
+size_t Request::headerData(char *buffer, size_t size, size_t nmemb, void *userp)
+{
+	assert(size == 1);
+	Request *req = static_cast<Request*>(userp);
+	req->headerCallback(buffer, static_cast<int>(nmemb));
+	return nmemb;
+}
+
+// ==================================================================
+
+void Curl::initialize(Option opt)
+{
+	curl_global_init(opt == kCURLnossl ? CURL_GLOBAL_NOTHING : CURL_GLOBAL_SSL);
+}
+
+int Curl::socketCallback(CURL *c, int fd, int what, void *userp, void *socketp)
+{
+	Curl *curl = static_cast<Curl*>(userp);
+	const char *whatstr[]={ "none", "IN", "OUT", "INOUT", "REMOVE" };
+	LOG_DEBUG << "Curl::socketCallback [" << curl << "] - fd = " << fd
+			<< " what = " << whatstr[what];
+	Request *req = nullptr;
+	curl_easy_getinfo(c, CURLINFO_PRIVATE, &req);
+	assert(req->getCurl() == c);
+	if (what == CURL_POLL_REMOVE)
+	{
+		Channel *ch = static_cast<Channel*>(socketp);
+		assert(req->getChannel() == ch);
+		req->removeChannel();
+		ch = nullptr;
+		curl_multi_assign(curl->curlm, fd, ch);
+	}
+	else
+	{
+		Channel *ch = static_cast<Channel*>(socketp);
+		if (!ch)
+		{
+		  ch = req->setChannel(fd);
+		  ch->setReadCallback(std::bind(&Curl::onRead, curl, fd));
+		  ch->setWriteCallback(std::bind(&Curl::onWrite, curl, fd));
+		  ch->enableReading();
+		  curl_multi_assign(curl->curlm, fd, ch);
+		  LOG_TRACE << "new channel for fd=" << fd;
+		}
+
+		assert(req->getChannel() == ch);
+		// update
+		if (what & CURL_POLL_OUT)
+		{
+			ch->enableWriting();
+		}
+		else
+		{
+			ch->disableWriting();
+		}
+	}
+	return 0;
+}
+
+int Curl::timerCallback(CURLM *curlm, long ms, void *userp)
+{
+	Curl *curl = static_cast<Curl*>(userp);
+	LOG_DEBUG << curl << " " << ms << " ms";
+	curl->loop->runAfter(static_cast<int>(ms)/1000.0, false, std::bind(&Curl::onTimer, curl));
+	return 0;
+}
+
+Curl::Curl(EventLoop *loop)
+  : loop(loop),
+    curlm(curl_multi_init()),
+    runningHandles(0),
+    prevRunningHandles(0)
+{
+	curl_multi_setopt(curlm, CURLMOPT_SOCKETFUNCTION, &Curl::socketCallback);
+	curl_multi_setopt(curlm, CURLMOPT_SOCKETDATA, this);
+	curl_multi_setopt(curlm, CURLMOPT_TIMERFUNCTION, &Curl::timerCallback);
+	curl_multi_setopt(curlm, CURLMOPT_TIMERDATA, this);
+}
+
+Curl::~Curl()
+{
+	curl_multi_cleanup(curlm);
+}
+
+RequestPtr Curl::getUrl(std::string_view url)
+{
+	RequestPtr req(new Request(this, url.data()));
+	return req;
+}
+
+RequestPtr Curl::getUrl(std::string_view url, std::string_view body)
+{
+	RequestPtr req(new Request(this, url.data(), body.data(), body.size()));
+	return req;
+}
+
+void Curl::onTimer()
+{
+	CURLMcode rc = CURLM_OK;
+	do
+	{
+		LOG_TRACE;
+		rc = curl_multi_socket_action(curlm, CURL_SOCKET_TIMEOUT, 0, &runningHandles);
+		LOG_TRACE << rc << " " << runningHandles;
+	} while (rc == CURLM_CALL_MULTI_PERFORM);
+	checkFinish();
+}
+
+void Curl::onRead(int fd)
+{
+	CURLMcode rc = CURLM_OK;
+	do
+	{
+		LOG_TRACE << fd;
+		rc = curl_multi_socket_action(curlm, fd, CURL_POLL_IN, &runningHandles);
+		LOG_TRACE << fd << " " << rc << " " << runningHandles;
+	} while (rc == CURLM_CALL_MULTI_PERFORM);
+	checkFinish();
+}
+
+void Curl::onWrite(int fd)
+{
+	CURLMcode rc = CURLM_OK;
+	do
+	{
+		LOG_TRACE << fd;
+		rc = curl_multi_socket_action(curlm, fd, CURL_POLL_OUT, &runningHandles);
+		LOG_TRACE << fd << " " << rc << " " << runningHandles;
+	} while (rc == CURLM_CALL_MULTI_PERFORM);
+	checkFinish();
+}
+
+void Curl::checkFinish()
+{
+	if (prevRunningHandles > runningHandles || runningHandles == 0)
+	{
+		CURLMsg *msg = nullptr;
+		int left = 0;
+		while ( (msg = curl_multi_info_read(curlm, &left)) != nullptr)
+		{
+			if (msg->msg == CURLMSG_DONE)
+			{
+				CURL *c = msg->easy_handle;
+				CURLcode res = msg->data.result;
+				Request *req = nullptr;
+				curl_easy_getinfo(c, CURLINFO_PRIVATE, &req);
+				assert(req->getCurl() == c);
+				LOG_TRACE << req << " done";
+				req->done(res);
+			}
+		}
+	}
+	prevRunningHandles = runningHandles;
+}
+

example/curl/curl.h

@@ -0,0 +1,135 @@
+#pragma once
+#include "all.h"
+#include "eventloop.h"
+#include "channel.h"
+#include "tcpserver.h"
+#include "log.h"
+#include <curl/curl.h>
+
+extern "C"
+{
+	typedef void CURLM;
+	typedef void CURL;
+}
+
+class Curl;
+class Request : public std::enable_shared_from_this<Request>
+{
+public:
+	typedef std::function<void(Request *, const char *, int)> DataCallback;
+	typedef std::function<void(const char *, int)> HeadCallback;
+	typedef std::function<void(Request *, int)> DoneCallback;
+
+	Request(Curl *, const char *url);
+	Request(Curl *, const char *url, const char *data, size_t len);
+	
+	~Request();
+
+	void setDataCallback(const DataCallback &cb)
+	{ dataCb = cb; }
+
+	void setDoneCallback(const DoneCallback &cb)
+	{ doneCb = cb; }
+
+	void setHeaderCallback(const HeadCallback &cb)
+	{ headerCb = cb; }
+
+	void headerOnly();
+
+	void setRange(const std::string_view range);
+
+	template<typename OPT>
+	int setopt(OPT opt, long p)
+	{
+		return curl_easy_setopt(curl, opt, p);
+	}
+
+	template<typename OPT>
+	int setopt(OPT opt, const char *p)
+	{
+		return curl_easy_setopt(curl, opt, p);
+	}
+
+	template<typename OPT>
+	int setopt(OPT opt, void *p)
+	{
+		return curl_easy_setopt(curl, opt, p);
+	}
+
+	template<typename OPT>
+	int setopt(OPT opt, size_t (*p)(char *, size_t , size_t , void *))
+	{
+		return curl_easy_setopt(curl, opt, p);
+	}
+
+	TcpConnectionPtr getTcpConnection() { return conn; }
+	void setTcpConnection(const TcpConnectionPtr &connection) { conn = connection; }
+	const char *getEffectiveUrl();
+	const char *getRedirectUrl();
+	int getResponseCode();
+
+	// internal
+	Channel *setChannel(int fd);
+	void removeChannel();
+	void done(int code);
+	CURL *getCurl() { return curl; }
+	Channel *getChannel() { return channel.get(); }
+private:
+	Request(const Request&);
+	void operator=(const Request&);
+
+	void dataCallback(const char *buffer, int len);
+	void headerCallback(const char *buffer, int len);
+	static size_t writeData(char *buffer, size_t size, size_t nmemb, void *userp);
+	static size_t headerData(char *buffer, size_t size, size_t nmemb, void *userp);
+	void doneCallback();
+
+	class Curl *owner;
+	CURL *curl;
+	std::shared_ptr<Channel> channel;
+	DataCallback dataCb;
+	HeadCallback headerCb;
+	DoneCallback doneCb;
+	TcpConnectionPtr conn;
+};
+
+typedef std::shared_ptr<Request> RequestPtr;
+class Curl
+{
+public:
+	enum Option
+	{
+		kCURLnossl = 0,
+		kCURLssl   = 1,
+	};
+
+	explicit Curl(EventLoop *loop);
+	~Curl();
+
+	RequestPtr getUrl(std::string_view url);
+	RequestPtr getUrl(std::string_view url, std::string_view body);
+
+	static void initialize(Option opt = kCURLnossl);
+
+	// internal
+	CURLM *getCurlm() { return curlm; }
+	EventLoop* getLoop() { return loop; }
+
+ private:
+	Curl(const Curl&);
+	void operator=(const Curl&);
+
+	void onTimer();
+	void onRead(int fd);
+	void onWrite(int fd);
+	void checkFinish();
+
+	static int socketCallback(CURL *, int, int, void*, void*);
+	static int timerCallback(CURLM *, long, void*);
+
+	EventLoop *loop;
+	CURLM *curlm;
+	int runningHandles;
+	int prevRunningHandles;
+};
+

example/curl/main.cc

@@ -0,0 +1,46 @@
+#include "httpcurl.h"
+
+
+EventLoop* g_loop = NULL;
+
+void onData(Request* c, const char* data, int len)
+{
+	printf("data %s\n", data);
+}
+
+void onHead(const char* data, int len)
+{
+	printf("data1 %s\n", data);
+}
+
+void done(Request* c, int code)
+{
+    printf("done %p %s %d\n", c, c->getEffectiveUrl(), code);
+}
+
+void done2(Request* c, int code)
+{
+	printf("done2 %p %s %d %d\n", c, c->getRedirectUrl(), c->getResponseCode(), code);
+  //g_loop->quit();
+}
+
+int main(int argc, char* argv[])
+{
+	EventLoop loop;
+	g_loop = &loop;
+	loop.runAfter(5.0, std::bind(&EventLoop::quit, &loop));
+	Curl::initialize(Curl::kCURLnossl);
+	Curl curl(&loop);
+	signal(SIGPIPE, SIG_IGN);
+
+	RequestPtr req = curl.getUrl("http://192.168.6.1:8000/");
+	req->setDataCallback(onData);
+	req->setDoneCallback(done2);
+	req->setHeaderCallback(onHead);
+
+	curl::RequestPtr req2 = curl.getUrl("https://github.com");
+	//req2->allowRedirect(5);
+	req2->setDataCallback(onData);
+	req2->setDoneCallback(done);
+	loop.run();
+}

example/evcpp/main.cpp

@@ -1,14 +1,14 @@
-#include "evcpp.h"
-
-void onConnect(evcpp::TcpConnectionPtr conn)
-{
-	printf("onConnect %zd\n",conn.use_count());
-}
-
-int main()
-{
-	evcpp::EventLoop loop;
-	evcpp::Listener listener(&loop,1234);
-	listener.setNewConnectionCallback(onConnect);
-	loop.loop();
-}
+#include "evcpp.h"
+
+void onConnect(evcpp::TcpConnectionPtr conn)
+{
+	printf("onConnect %zd\n",conn.use_count());
+}
+
+int main()
+{
+	evcpp::EventLoop loop;
+	evcpp::Listener listener(&loop,1234);
+	listener.setNewConnectionCallback(onConnect);
+	loop.loop();
+}

example/filesearch/client.cc

@@ -0,0 +1,356 @@
+#include "all.h"
+#include "tcpconnection.h"
+#include "tcpclient.h"
+#include "eventloop.h"
+#include "log.h"
+#include "table.h"
+#include "posix.h"
+#include "tablebuilder.h"
+#include "memtable.h"
+#include "option.h"
+
+const size_t kMaxSize = 10 * 1000 * 1000;
+enum SaverState
+{
+	kNotFound,
+	kFound,
+	kDeleted,
+	kCorrupt,
+};
+
+class Sharder
+{
+public:
+	explicit Sharder(int32_t nbuckets)
+	{
+		for (int32_t i = 0; i < nbuckets; ++i)
+		{
+			char buf[256];
+			snprintf(buf, sizeof buf, "shard-%05d-of-%05d", i, nbuckets);
+			std::shared_ptr<std::ofstream> in(new std::ofstream(buf));
+			buckets.push_back(in);
+		}
+		assert(buckets.size() == static_cast<size_t>(nbuckets));
+	}
+
+	void output(const std::string &key, const std::string &value)
+	{
+		size_t idx = std::hash<std::string>()(key) % buckets.size();
+		*buckets[idx] << key << '\t' << value << '\n';
+	}
+
+protected:
+	std::vector<std::shared_ptr<std::ofstream>> buckets;
+private:
+	Sharder(const Sharder&);
+	void operator=(const Sharder&);
+};
+// ======= merge =======
+
+class Source  // copyable
+{
+public:
+	explicit Source(const std::shared_ptr<std::ifstream> &in)
+		: in(in)
+	{
+
+	}
+
+	bool next()
+	{
+		std::string line;
+		if (getline(*(in.get()), line))
+		{
+			size_t tab = line.find('\t');
+			if (tab != std::string::npos)
+			{
+				std::string key(line.c_str(), line.c_str() + tab);
+				this->key = key;
+				std::string(line.c_str() + tab + 1, line.c_str() + line.size());
+				this->value = value;
+				return true;
+			}
+		}
+		return false;
+	}
+
+	bool operator<(const Source &rhs) const
+	{
+		return key < rhs.key;
+	}
+	
+	std::string getKey() { MD5(key).toStr(); }
+	std::string getValue() { return value; }
+	void outputTo(std::ostream &out)
+	{
+		out << key << '\t' << value << '\n';
+	}
+
+	std::shared_ptr<std::ifstream> in;
+	std::string key;
+	std::string value;
+};
+
+class Client
+{
+public:
+	Client(const char *ip, uint16_t port, int32_t argc, char *argv[])
+	:ip(ip),
+	port(port),
+	nbuckets(10)
+	{
+		EventLoop loop;
+		this->loop = &loop;
+		TcpClient client(&loop, ip, port, nullptr);
+		client.setConnectionCallback(std::bind(&Client::connectionCallback, this, std::placeholders::_1));
+		client.setMessageCallback(std::bind(&Client::messageCallback, this, std::placeholders::_1, std::placeholders::_2));
+		client.enableRetry();
+		client.connect(true);
+		this->client = &client;
+		
+		shard(nbuckets, argc, argv);
+		sortShareds(nbuckets);
+		merge(nbuckets);
+	}
+	
+	void writeCompleteCallback(const TcpConnectionPtr &conn)
+	{
+		conn->startRead();
+		conn->setWriteCompleteCallback(WriteCompleteCallback());
+	}
+
+	void highWaterCallback(const TcpConnectionPtr &conn, size_t bytesToSent)
+	{
+		LOG_INFO << " bytes " << bytesToSent;
+		if (conn->outputBuffer()->readableBytes() > 0)
+		{
+			conn->stopRead();
+			conn->setWriteCompleteCallback(
+				std::bind(&Client::writeCompleteCallback, this, std::placeholders::_1));
+		}
+	}
+
+	void connectionCallback(const TcpConnectionPtr &conn)
+	{
+		conn->setHighWaterMarkCallback(
+			std::bind(&Client::highWaterCallback, this, std::placeholders::_1, std::placeholders::_2),
+			1024 * 1024);
+	}
+	
+	void cmp(const std::any &arg, const std::string_view &ikey, const std::string_view &v)
+	{
+		auto c = ikey.compare(v);
+		if (c == 0)
+		{
+			state = kCorrupt;
+		}
+		else
+		{
+			state = kFound;
+		}
+	}
+
+	void messageCallback(const TcpConnectionPtr &conn, Buffer *buffer)
+	{
+		const char *crlf;
+		while ((crlf = buffer->findCRLF()) != nullptr)
+		{
+			// string request(buffer->peek(), crlf);
+			// printf("%s\n", request.c_str());
+			const char *tab = std::find(buffer->peek(), crlf, '\t');
+			if (tab != crlf)
+			{
+				std::string key(buffer->peek(), tab);
+				std::string value(tab + 1, crlf);
+
+				Status s = table->internalGet(ReadOptions(), key, nullptr, std::bind(&Client::cmp, this,
+						std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
+
+				if (!s.ok())
+				{
+					return s;
+				}
+
+				switch (state)
+				{
+					case kNotFound:
+						files[key] = value;
+					  break;      // Keep searching in other files
+					case kFound:
+					  break;
+				}
+			}
+			else
+			{
+				LOG_WARN << "Wrong format, no tab found";
+				conn->shutdown();
+			}
+			buffer->retrieveUntil(crlf + 2);
+		}
+	}
+	
+	void merge(const int32_t nbuckets)
+	{
+		std::vector<std::shared_ptr<std::ifstream>> inputs;
+		std::vector<Source> keys;
+
+		for (int32_t i = 0; i < nbuckets; ++i)
+		{
+			char buf[256];
+			snprintf(buf, sizeof buf, "count-%05d-of-%05d", i, nbuckets);
+			std::shared_ptr<std::ifstream> in(new std::ifstream(buf));
+			inputs.push_back(in);
+			Source rec(in);
+			if (rec.next())
+			{
+				keys.push_back(rec);
+			}
+			::unlink(buf);
+		}
+
+		Options options;
+		std::shared_ptr<WritableFile> wfile;
+		s = options.env->newWritableFile("output", wfile);
+		assert(s.ok());
+		TableBuilder builder(options, wfile);
+
+		std::make_heap(keys.begin(), keys.end());
+		while (!keys.empty())
+		{
+			std::pop_heap(keys.begin(), keys.end());
+			builder.add(keys.back().getKey(), keys.back().getValue());
+			assert(builder.status().ok());
+
+			if (keys.back().next())
+			{
+				std::push_heap(keys.begin(), keys.end());
+			}
+			else
+			{
+				keys.pop_back();
+			}
+		}
+
+		s = builder.finish();
+		assert(s.ok());
+		s = wfile->sync();
+		assert(s.ok());
+		s = wfile->close();
+		assert(s.ok());
+
+		uint64_t size;
+		s = options.env->getFileSize(filename,&size);
+		assert (s.ok());
+		assert(size == builder.fileSize());
+
+		std::shared_ptr<RandomAccessFile> mfile = nullptr;
+		s = options.env->newRandomAccessFile(filename, mfile);
+		assert(s.ok());
+		s = Table::open(options, mfile, builder.fileSize(), table);
+		assert(s.ok());
+	}
+	
+	void shard(int32_t nbuckets, int32_t argc, char *argv[])
+	{
+		Sharder sharder(nbuckets);
+		for (int32_t i = 1; i < argc; ++i)
+		{
+			std::cout << "  processing input file " << argv[i] << std::endl;
+			std::map<std::string, std::string> shareds;
+			std::ifstream in(argv[i]);
+			while (in && !in.eof())
+			{
+				shareds.clear();
+				std::string str;
+				while (in >> str)
+				{
+					shareds.insert(std::make_pair(str, str));
+					if (shareds.size() > kMaxSize)
+					{
+						std::cout << "    split" << std::endl;
+						break;
+					}
+				}
+
+				for (const auto &kv : shareds)
+				{
+					sharder.output(kv.first, kv.second);
+				}
+			}
+		}
+		std::cout << "shuffling done" << std::endl;
+	}
+
+	std::map<std::string, std::string> readShard(int32_t idx, int32_t nbuckets)
+	{
+		std::map<std::string, std::string> shareds;
+
+		char buf[256];
+		snprintf(buf, sizeof buf, "shard-%05d-of-%05d", idx, nbuckets);
+		std::cout << "  reading " << buf << std::endl;
+		{
+			std::ifstream in(buf);
+			std::string line;
+
+			while (getline(in, line))
+			{
+				size_t tab = line.find('\t');
+				assert(tab != std::string::npos);
+				std::string key(line.c_str(), line.c_str() + tab);
+				std::string value(line.c_str() + tab + 1, line.c_str() + line.size());
+				shareds.insert(std::make_pair(key, value));
+			}
+		}
+
+		::unlink(buf);
+		return shareds;
+	}
+	
+	void sortShareds(const int32_t nbuckets)
+	{
+		for (int32_t i = 0; i < nbuckets; ++i)
+		{
+			// std::cout << "  sorting " << std::endl;
+			std::map<std::string, std::string> shareds;
+			for (const auto &entry : readShard(i, nbuckets))
+			{
+				shareds.insert(std::make_pair(entry.first, entry.second));
+			}
+
+			char buf[256];
+			snprintf(buf, sizeof buf, "count-%05d-of-%05d", i, nbuckets);
+			std::ofstream out(buf);
+			std::cout << "  writing " << buf << std::endl;
+			for (auto &it : shareds)
+			{
+				out << it.first << '\t' << it.second << '\n';
+			}
+		}
+		std::cout << "reducing done" << std::endl;
+	}
+	
+	void run()
+	{
+		loop->run();
+	}
+private:
+	Client(const Client&);
+	void operator=(const Client&);
+	
+	EventLoop *loop;
+	TcpClient *client;
+	const char *ip;
+	uint16_t port;
+	int32_t nbuckets;
+	std::shared_ptr<Table> table;
+	std::map<std::string, std::string> files;
+	Status state;
+
+};
+
+int main(int argc, char* argv[])
+{
+	Client client("127.0.0.1", 8888, argc, argv); 
+	client.run();
+	return 0;
+}

example/filesearch/filesort.cc

@@ -0,0 +1,218 @@
+#include <unistd.h>
+#include <memory>
+#include <fstream>
+#include <iostream>
+#include <unordered_map>
+#include <map>
+#include <vector>
+#include <set>
+#include <assert.h>
+#include <algorithm>
+
+int32_t index = 0;
+const size_t kMaxSize = 10 * 1000 * 1000;
+
+class Sharder
+{
+public:
+	explicit Sharder(int32_t nbuckets)
+	{
+		for (int32_t i = 0; i < nbuckets; ++i)
+		{
+			char buf[256];
+			snprintf(buf, sizeof buf, "shard-%05d-of-%05d", i, nbuckets);
+			std::shared_ptr<std::ofstream> in(new std::ofstream(buf));
+			buckets.push_back(in);
+		}
+		assert(buckets.size() == static_cast<size_t>(nbuckets));
+	}
+
+	void output(const std::string &key, const std::string &value)
+	{
+		size_t idx = std::hash<std::string>()(key) % buckets.size();
+		*buckets[idx] << key << '\t' << value << '\n';
+	}
+
+protected:
+	std::vector<std::shared_ptr<std::ofstream>> buckets;
+private:
+	Sharder(const Sharder&);
+	void operator=(const Sharder&);
+};
+
+void shard(int32_t nbuckets, int32_t argc, char *argv[])
+{
+	Sharder sharder(nbuckets);
+	for (int32_t i = 1; i < argc; ++i)
+	{
+		std::cout << "  processing input file " << argv[i] << std::endl;
+		std::map<std::string, std::string> shareds;
+		std::ifstream in(argv[i]);
+		while (in && !in.eof())
+		{
+			shareds.clear();
+			std::string str;
+			while (in >> str)
+			{
+				shareds.insert(std::make_pair(str, str));
+				if (shareds.size() > kMaxSize)
+				{
+					std::cout << "    split" << std::endl;
+					break;
+				}
+			}
+
+			for (const auto &kv : shareds)
+			{
+				sharder.output(kv.first, kv.second);
+			}
+		}
+	}
+	std::cout << "shuffling done" << std::endl;
+}
+
+// ======= sortShareds =======
+
+std::map<std::string, std::string> readShard(int32_t idx, int32_t nbuckets)
+{
+	std::map<std::string, std::string> shareds;
+
+	char buf[256];
+	snprintf(buf, sizeof buf, "shard-%05d-of-%05d", idx, nbuckets);
+	std::cout << "  reading " << buf << std::endl;
+	{
+		std::ifstream in(buf);
+		std::string line;
+
+		while (getline(in, line))
+		{
+			size_t tab = line.find('\t');
+			assert(tab != std::string::npos);
+			std::string key(line.c_str(), line.c_str() + tab);
+			std::string value(line.c_str() + tab + 1, line.c_str() + line.size());
+			shareds.insert(std::make_pair(key, value));
+		}
+	}
+
+	::unlink(buf);
+	return shareds;
+}
+
+void sortShareds(const int32_t nbuckets)
+{
+	for (int32_t i = 0; i < nbuckets; ++i)
+	{
+		// std::cout << "  sorting " << std::endl;
+		std::map<std::string, std::string> shareds;
+		for (const auto &entry : readShard(i, nbuckets))
+		{
+			shareds.insert(std::make_pair(entry.first, entry.second));
+		}
+
+		char buf[256];
+		snprintf(buf, sizeof buf, "count-%05d-of-%05d", i, nbuckets);
+		std::ofstream out(buf);
+		std::cout << "  writing " << buf << std::endl;
+		for (auto &it : shareds)
+		{
+			out << it.first << '\t' << it.second << '\n';
+		}
+	}
+	std::cout << "reducing done" << std::endl;
+}
+
+// ======= merge =======
+
+class Source  // copyable
+{
+public:
+	explicit Source(const std::shared_ptr<std::ifstream> &in)
+		: in(in)
+	{
+
+	}
+
+	bool next()
+	{
+		std::string line;
+		if (getline(*(in.get()), line))
+		{
+			size_t tab = line.find('\t');
+			if (tab != std::string::npos)
+			{
+				std::string key(line.c_str(), line.c_str() + tab);
+				this->key = key;
+				std::string(line.c_str() + tab + 1, line.c_str() + line.size());
+				this->value = value;
+				return true;
+			}
+		}
+		return false;
+	}
+
+	bool operator<(const Source &rhs) const
+	{
+		return key < rhs.key;
+	}
+	
+	void retrieve()
+	{
+		index += key.size() + 3 + value.size() + sizeof(int32_t)
+	}
+
+	void outputTo(std::ostream &out, int32_t index) const
+	{
+		out << key << '\t' << value << '\t' << std::to_string(index) << '\n';
+		retrieve();
+	}
+
+	std::shared_ptr<std::ifstream> in;
+	std::string key;
+	std::string value;
+};
+
+void merge(const int32_t nbuckets)
+{
+	std::vector<std::shared_ptr<std::ifstream>> inputs;
+	std::vector<Source> keys;
+
+	for (int32_t i = 0; i < nbuckets; ++i)
+	{
+		char buf[256];
+		snprintf(buf, sizeof buf, "count-%05d-of-%05d", i, nbuckets);
+		std::shared_ptr<std::ifstream> in(new std::ifstream(buf));
+		inputs.push_back(in);
+		Source rec(in);
+		if (rec.next())
+		{
+			keys.push_back(rec);
+		}
+		::unlink(buf);
+	}
+
+	std::ofstream out("output");
+	std::make_heap(keys.begin(), keys.end());
+
+	while (!keys.empty())
+	{
+		std::pop_heap(keys.begin(), keys.end());
+		keys.back().outputTo(out, index);
+	
+		if (keys.back().next())
+		{
+			std::push_heap(keys.begin(), keys.end());
+		}
+		else
+		{
+			keys.pop_back();
+		}
+	}
+}
+
+int32_t main(int32_t argc, char *argv[])
+{
+	int32_t nbuckets = 10;
+	shard(nbuckets, argc, argv);
+	sortShareds(nbuckets);
+	merge(nbuckets);
+}

example/filesearch/gen.py

@@ -0,0 +1,14 @@
+#!/usr/bin/python
+
+import random
+
+words = 100
+word_len = 10
+alphabet = 'ABCDEFGHIJKLMNOPQRSTUVWXYZABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-'
+
+output = open('random_words', 'w')
+for x in xrange(words):
+	arr = [random.choice(alphabet) for i in range(word_len)]
+	word = ''.join(arr)
+	output.write(word)
+	output.write('\n')

example/filesearch/md5.cc

@@ -0,0 +1,210 @@
+#include "md5.h"
+
+/* Define the static member of MD5. */
+const byte MD5::PADDING[64] = { 0x80 };
+const char MD5::HEX_NUMBERS[16] = {
+  '0', '1', '2', '3',
+  '4', '5', '6', '7',
+  '8', '9', 'a', 'b',
+  'c', 'd', 'e', 'f'
+};
+
+MD5::MD5(const string& message) {
+	finished = false;
+	/* Reset number of bits. */
+	count[0] = count[1] = 0;
+	/* Initialization constants. */
+	state[0] = 0x67452301;
+	state[1] = 0xefcdab89;
+	state[2] = 0x98badcfe;
+	state[3] = 0x10325476;
+
+	/* Initialization the object according to message. */
+	init((const byte*)message.c_str(), message.length());
+}
+
+const byte* MD5::getDigest() {
+	if (!finished) {
+		finished = true;
+
+		byte bits[8];
+		bit32 oldState[4];
+		bit32 oldCount[2];
+		bit32 index, padLen;
+
+		/* Save current state and count. */
+		memcpy(oldState, state, 16);
+		memcpy(oldCount, count, 8);
+
+		/* Save number of bits */
+		encode(count, bits, 8);
+
+		/* Pad out to 56 mod 64. */
+		index = (bit32)((count[0] >> 3) & 0x3f);
+		padLen = (index < 56) ? (56 - index) : (120 - index);
+		init(PADDING, padLen);
+
+		/* Append length (before padding) */
+		init(bits, 8);
+
+		/* Store state in digest */
+		encode(state, digest, 16);
+
+		/* Restore current state and count. */
+		memcpy(state, oldState, 16);
+		memcpy(count, oldCount, 8);
+	}
+	return digest;
+}
+
+void MD5::init(const byte* input, size_t len) {
+
+	bit32 i, index, partLen;
+
+	finished = false;
+
+	/* Compute number of bytes mod 64 */
+	index = (bit32)((count[0] >> 3) & 0x3f);
+
+	/* update number of bits */
+	if ((count[0] += ((bit32)len << 3)) < ((bit32)len << 3)) {
+		++count[1];
+	}
+	count[1] += ((bit32)len >> 29);
+
+	partLen = 64 - index;
+
+	/* transform as many times as possible. */
+	if (len >= partLen) {
+
+		memcpy(&buffer[index], input, partLen);
+		transform(buffer);
+
+		for (i = partLen; i + 63 < len; i += 64) {
+			transform(&input[i]);
+		}
+		index = 0;
+
+	}
+	else {
+		i = 0;
+	}
+
+	/* Buffer remaining input */
+	memcpy(&buffer[index], &input[i], len - i);
+}
+
+void MD5::transform(const byte block[64]) {
+
+	bit32 a = state[0], b = state[1], c = state[2], d = state[3], x[16];
+
+	decode(block, x, 64);
+
+	/* Round 1 */
+	FF(a, b, c, d, x[0], s11, 0xd76aa478);
+	FF(d, a, b, c, x[1], s12, 0xe8c7b756);
+	FF(c, d, a, b, x[2], s13, 0x242070db);
+	FF(b, c, d, a, x[3], s14, 0xc1bdceee);
+	FF(a, b, c, d, x[4], s11, 0xf57c0faf);
+	FF(d, a, b, c, x[5], s12, 0x4787c62a);
+	FF(c, d, a, b, x[6], s13, 0xa8304613);
+	FF(b, c, d, a, x[7], s14, 0xfd469501);
+	FF(a, b, c, d, x[8], s11, 0x698098d8);
+	FF(d, a, b, c, x[9], s12, 0x8b44f7af);
+	FF(c, d, a, b, x[10], s13, 0xffff5bb1);
+	FF(b, c, d, a, x[11], s14, 0x895cd7be);
+	FF(a, b, c, d, x[12], s11, 0x6b901122);
+	FF(d, a, b, c, x[13], s12, 0xfd987193);
+	FF(c, d, a, b, x[14], s13, 0xa679438e);
+	FF(b, c, d, a, x[15], s14, 0x49b40821);
+
+	/* Round 2 */
+	GG(a, b, c, d, x[1], s21, 0xf61e2562);
+	GG(d, a, b, c, x[6], s22, 0xc040b340);
+	GG(c, d, a, b, x[11], s23, 0x265e5a51);
+	GG(b, c, d, a, x[0], s24, 0xe9b6c7aa);
+	GG(a, b, c, d, x[5], s21, 0xd62f105d);
+	GG(d, a, b, c, x[10], s22, 0x2441453);
+	GG(c, d, a, b, x[15], s23, 0xd8a1e681);
+	GG(b, c, d, a, x[4], s24, 0xe7d3fbc8);
+	GG(a, b, c, d, x[9], s21, 0x21e1cde6);
+	GG(d, a, b, c, x[14], s22, 0xc33707d6);
+	GG(c, d, a, b, x[3], s23, 0xf4d50d87);
+	GG(b, c, d, a, x[8], s24, 0x455a14ed);
+	GG(a, b, c, d, x[13], s21, 0xa9e3e905);
+	GG(d, a, b, c, x[2], s22, 0xfcefa3f8);
+	GG(c, d, a, b, x[7], s23, 0x676f02d9);
+	GG(b, c, d, a, x[12], s24, 0x8d2a4c8a);
+
+	/* Round 3 */
+	HH(a, b, c, d, x[5], s31, 0xfffa3942);
+	HH(d, a, b, c, x[8], s32, 0x8771f681);
+	HH(c, d, a, b, x[11], s33, 0x6d9d6122);
+	HH(b, c, d, a, x[14], s34, 0xfde5380c);
+	HH(a, b, c, d, x[1], s31, 0xa4beea44);
+	HH(d, a, b, c, x[4], s32, 0x4bdecfa9);
+	HH(c, d, a, b, x[7], s33, 0xf6bb4b60);
+	HH(b, c, d, a, x[10], s34, 0xbebfbc70);
+	HH(a, b, c, d, x[13], s31, 0x289b7ec6);
+	HH(d, a, b, c, x[0], s32, 0xeaa127fa);
+	HH(c, d, a, b, x[3], s33, 0xd4ef3085);
+	HH(b, c, d, a, x[6], s34, 0x4881d05);
+	HH(a, b, c, d, x[9], s31, 0xd9d4d039);
+	HH(d, a, b, c, x[12], s32, 0xe6db99e5);
+	HH(c, d, a, b, x[15], s33, 0x1fa27cf8);
+	HH(b, c, d, a, x[2], s34, 0xc4ac5665);
+
+	/* Round 4 */
+	II(a, b, c, d, x[0], s41, 0xf4292244);
+	II(d, a, b, c, x[7], s42, 0x432aff97);
+	II(c, d, a, b, x[14], s43, 0xab9423a7);
+	II(b, c, d, a, x[5], s44, 0xfc93a039);
+	II(a, b, c, d, x[12], s41, 0x655b59c3);
+	II(d, a, b, c, x[3], s42, 0x8f0ccc92);
+	II(c, d, a, b, x[10], s43, 0xffeff47d);
+	II(b, c, d, a, x[1], s44, 0x85845dd1);
+	II(a, b, c, d, x[8], s41, 0x6fa87e4f);
+	II(d, a, b, c, x[15], s42, 0xfe2ce6e0);
+	II(c, d, a, b, x[6], s43, 0xa3014314);
+	II(b, c, d, a, x[13], s44, 0x4e0811a1);
+	II(a, b, c, d, x[4], s41, 0xf7537e82);
+	II(d, a, b, c, x[11], s42, 0xbd3af235);
+	II(c, d, a, b, x[2], s43, 0x2ad7d2bb);
+	II(b, c, d, a, x[9], s44, 0xeb86d391);
+
+	state[0] += a;
+	state[1] += b;
+	state[2] += c;
+	state[3] += d;
+}
+
+void MD5::encode(const bit32* input, byte* output, size_t length) {
+
+	for (size_t i = 0, j = 0; j < length; ++i, j += 4) {
+		output[j] = (byte)(input[i] & 0xff);
+		output[j + 1] = (byte)((input[i] >> 8) & 0xff);
+		output[j + 2] = (byte)((input[i] >> 16) & 0xff);
+		output[j + 3] = (byte)((input[i] >> 24) & 0xff);
+	}
+}
+
+void MD5::decode(const byte* input, bit32* output, size_t length) {
+	for (size_t i = 0, j = 0; j < length; ++i, j += 4) {
+		output[i] = ((bit32)input[j]) | (((bit32)input[j + 1]) << 8) |
+			(((bit32)input[j + 2]) << 16) | (((bit32)input[j + 3]) << 24);
+	}
+}
+
+string MD5::toStr() {
+	const byte* digest_ = getDigest();
+	string str;
+	str.reserve(16 << 1);
+	for (size_t i = 0; i < 16; ++i) {
+		int t = digest_[i];
+		int a = t / 16;
+		int b = t % 16;
+		str.append(1, HEX_NUMBERS[a]);
+		str.append(1, HEX_NUMBERS[b]);
+	}
+	return str;
+}

example/filesearch/md5.h

@@ -0,0 +1,103 @@
+#define s11 7
+#define s12 12
+#define s13 17
+#define s14 22
+#define s21 5
+#define s22 9
+#define s23 14
+#define s24 20
+#define s31 4
+#define s32 11
+#define s33 16
+#define s34 23
+#define s41 6
+#define s42 10
+#define s43 15
+#define s44 21
+
+#define F(x, y, z) (((x) & (y)) | ((~x) & (z)))
+#define G(x, y, z) (((x) & (z)) | ((y) & (~z)))
+#define H(x, y, z) ((x) ^ (y) ^ (z))
+#define I(x, y, z) ((y) ^ ((x) | (~z)))
+
+#define ROTATELEFT(num, n) (((num) << (n)) | ((num) >> (32-(n))))
+
+#define FF(a, b, c, d, x, s, ac) { \
+  (a) += F ((b), (c), (d)) + (x) + ac; \
+  (a) = ROTATELEFT ((a), (s)); \
+  (a) += (b); \
+}
+#define GG(a, b, c, d, x, s, ac) { \
+  (a) += G ((b), (c), (d)) + (x) + ac; \
+  (a) = ROTATELEFT ((a), (s)); \
+  (a) += (b); \
+}
+#define HH(a, b, c, d, x, s, ac) { \
+  (a) += H ((b), (c), (d)) + (x) + ac; \
+  (a) = ROTATELEFT ((a), (s)); \
+  (a) += (b); \
+}
+#define II(a, b, c, d, x, s, ac) { \
+  (a) += I ((b), (c), (d)) + (x) + ac; \
+  (a) = ROTATELEFT ((a), (s)); \
+  (a) += (b); \
+}
+
+#include <string>
+#include <cstring>
+
+using std::string;
+
+/* Define of btye.*/
+typedef unsigned char byte;
+/* Define of byte. */
+typedef unsigned int bit32;
+
+class MD5 {
+public:
+	/* Construct a MD5 object with a string. */
+	MD5(const string& message);
+
+	/* Generate md5 digest. */
+	const byte* getDigest();
+
+	/* Convert digest to string value */
+	string toStr();
+
+private:
+	/* Initialization the md5 object, processing another message block,
+	 * and updating the context.*/
+	void init(const byte* input, size_t len);
+
+	/* MD5 basic transformation. Transforms state based on block. */
+	void transform(const byte block[64]);
+
+	/* Encodes input (usigned long) into output (byte). */
+	void encode(const bit32* input, byte* output, size_t length);
+
+	/* Decodes input (byte) into output (usigned long). */
+	void decode(const byte* input, bit32* output, size_t length);
+
+private:
+	/* Flag for mark whether calculate finished. */
+	bool finished;
+
+	/* state (ABCD). */
+	bit32 state[4];
+
+	/* number of bits, low-order word first. */
+	bit32 count[2];
+
+	/* input buffer. */
+	byte buffer[64];
+
+	/* message digest. */
+	byte digest[16];
+
+	/* padding for calculate. */
+	static const byte PADDING[64];
+
+	/* Hex numbers. */
+	static const char HEX_NUMBERS[16];
+};
+

example/filesearch/server.cc

@@ -0,0 +1,114 @@
+#include "all.h"
+#include "tcpconnection.h"
+#include "tcpserver.h"
+#include "eventloop.h"
+#include "log.h"
+#include "md5.h"
+namespace fs = std::experimental::filesystem;
+
+class Server
+{
+public:
+	Server(const char *ip, uint16_t port)
+	:ip(ip),
+	port(port),
+	path("filedata/")
+	{
+		EventLoop loop;
+		this->loop = &loop;
+		TcpServer server(&loop, ip, port, nullptr);
+		server.setConnectionCallback(std::bind(&Server::connectionCallback, this, std::placeholders::_1));
+		server.setMessageCallback(std::bind(&Server::messageCallback, this, std::placeholders::_1, std::placeholders::_2));
+		server.start();
+	}
+	
+	void readAllFile(const TcpConnectionPtr &conn)
+	{
+		std::string fileName;
+		for (auto &iter : fs::directory_iterator(path))
+		{
+			auto fe = iter.path();
+
+			fileName.clear();
+			fileName += path;
+			fileName += fe.filename().string();
+
+			if (!fs::exists(fileName))
+			{
+				exit(1);
+			}
+
+			std::ifstream fin(fileName.c_str());
+			if (fin.eof())
+			{
+				continue;
+			}
+
+			int i = 0;
+			std::string line;
+			while(std::getline(fin,line, '\n'))
+			{
+				Buffer *buffer = conn->outputBuffer();
+				buffer->append(MD5(line).toStr());
+				buffer->append("\r\n", 2);
+				conn->sendPipe();
+			}
+			fin.close();
+		}
+	}
+
+	void writeCompleteCallback(const TcpConnectionPtr &conn)
+	{
+		conn->startRead();
+		conn->setWriteCompleteCallback(WriteCompleteCallback());
+	}
+
+	void highWaterCallback(const TcpConnectionPtr &conn, size_t bytesToSent)
+	{
+		LOG_INFO << " bytes " << bytesToSent;
+		if (conn->outputBuffer()->readableBytes() > 0)
+		{
+			conn->stopRead();
+			conn->setWriteCompleteCallback(
+				std::bind(&Server::writeCompleteCallback, this, std::placeholders::_1));
+		}
+	}
+
+	void connectionCallback(const TcpConnectionPtr &conn)
+	{
+		conn->setHighWaterMarkCallback(
+			std::bind(&Server::highWaterCallback, this, std::placeholders::_1, std::placeholders::_2),
+			1024 * 1024);
+		readAllFile(conn);
+	}
+	
+	void messageCallback(const TcpConnectionPtr &conn, Buffer *buffer)
+	{
+		buffer->retrieveAll();
+	}
+	
+	void readFileData(const TcpConnectionPtr &conn)
+	{
+		conn->shutdown();
+	}
+	
+	void run()
+	{
+		loop->run();
+	}
+private:
+	Server(const Server&);
+	void operator=(const Server&);
+	
+	EventLoop *loop;
+	const char *ip;
+	uint16_t port;
+	const char *path;
+};
+
+int main(int argc, char* argv[])
+{
+	Server server("127.0.0.1", 8888);
+	server.run();
+	return 0;
+}

example/http/httptest.cc

@@ -50,8 +50,7 @@ void onMessage(const HttpRequest &req, HttpResponse *resp)
 		}
 		else if (req.getPath() == "/test.txt")
 		{
-			std::string filename = "attachment;filename=";
-			filename += "test.txt";
+			std::string filename = "attachment;filename=test.txt";
 			resp->setStatusCode(HttpResponse::k200k);
 			resp->setStatusMessage("OK");
 			resp->setContentType("text/plain");

example/leveldb/.DS_Store

@@ -1,108 +1,108 @@
-#include "dbimpl.h"
-#include "option.h"
-#include "cache.h"
-#include "coding.h"
-#include <vector>
-#include <functional>
-
-// Conversions between numeric keys/values and the types expected by Cache.
-static std::string encodeKey(int k)
-{
-	std::string result;
-	putFixed32(&result, k);
-	return result;
-}
-
-static int decodeKey(const std::string_view &k)
-{
-	assert(k.size() == 4);
-	return decodeFixed32(k.data());
-}
-
-static int decodeValue(const std::any &v)
-{
-	assert(v.has_value());
-	return std::any_cast<int>(v);
-}
-
-class CacheTest
-{
-public:
-	void deleterCallBack(const std::string_view &key, const std::any &v)
-	{
-		deletedKeys.push_back(decodeKey(key));
-		deletedValues.push_back(decodeValue(v));
-	}
-
-	CacheTest()
-		:cache(new ShardedLRUCache(kCacheSize))
-	{
-
-	}
-
-	int lookup(int key)
-	{
-		auto handle = cache->lookup(encodeKey(key));
-		const int r = (handle == nullptr) ? -1 : decodeValue(cache->value(handle));
-		if (handle != nullptr)
-		{
-			cache->erase(encodeKey(key));
-		}
-		return r;
-	}
-
-	void insert(int key, int value, int charge = 1)
-	{
-		cache->insert(encodeKey(key), value, charge,
-			std::bind(&CacheTest::deleterCallBack, this,
-				std::placeholders::_1, std::placeholders::_2));
-	}
-
-	void erase(int key)
-	{
-		cache->erase(encodeKey(key));
-	}
-
-	void hitAndMiss()
-	{
-		assert(lookup(100) == -1);
-		insert(100, 101);
-		assert(lookup(100) == 101);
-		assert(lookup(200) == -1);
-		assert(lookup(300) == -1);
-
-		insert(200, 201);
-		assert(lookup(100) == 101);
-		assert(lookup(200) == 201);
-		assert(lookup(300) == -1);
-
-		insert(100, 102);
-		assert(lookup(100) == 102);
-		assert(lookup(200) == 201);
-		assert(lookup(300) == -1);
-
-		assert(deletedKeys.size() == 1);
-		assert(deletedKeys[0] == 100);
-		assert(deletedValues[0] == 101);
-	}
-
-	void erase()
-	{
-
-	}
-
-private:
-	static const int kCacheSize = 1000;
-	std::vector<int> deletedKeys;
-	std::vector<int> deletedValues;
-	std::shared_ptr<ShardedLRUCache> cache;
-};
-
-int main(int argc, char *argv[])
-{
-	CacheTest test;
-	test.hitAndMiss();
-	test.erase();
-	return 0;
-
-}
+#include "dbimpl.h"
+#include "option.h"
+#include "cache.h"
+#include "coding.h"
+#include <vector>
+#include <functional>
+
+// Conversions between numeric keys/values and the types expected by Cache.
+static std::string encodeKey(int k)
+{
+	std::string result;
+	putFixed32(&result, k);
+	return result;
+}
+
+static int decodeKey(const std::string_view &k)
+{
+	assert(k.size() == 4);
+	return decodeFixed32(k.data());
+}
+
+static int decodeValue(const std::any &v)
+{
+	assert(v.has_value());
+	return std::any_cast<int>(v);
+}
+
+class CacheTest
+{
+public:
+	void deleterCallBack(const std::string_view &key, const std::any &v)
+	{
+		deletedKeys.push_back(decodeKey(key));
+		deletedValues.push_back(decodeValue(v));
+	}
+
+	CacheTest()
+		:cache(new ShardedLRUCache(kCacheSize))
+	{
+
+	}
+
+	int lookup(int key)
+	{
+		auto handle = cache->lookup(encodeKey(key));
+		const int r = (handle == nullptr) ? -1 : decodeValue(cache->value(handle));
+		if (handle != nullptr)
+		{
+			cache->erase(encodeKey(key));
+		}
+		return r;
+	}
+
+	void insert(int key, int value, int charge = 1)
+	{
+		cache->insert(encodeKey(key), value, charge,
+			std::bind(&CacheTest::deleterCallBack, this,
+				std::placeholders::_1, std::placeholders::_2));
+	}
+
+	void erase(int key)
+	{
+		cache->erase(encodeKey(key));
+	}
+
+	void hitAndMiss()
+	{
+		assert(lookup(100) == -1);
+		insert(100, 101);
+		assert(lookup(100) == 101);
+		assert(lookup(200) == -1);
+		assert(lookup(300) == -1);
+
+		insert(200, 201);
+		assert(lookup(100) == 101);
+		assert(lookup(200) == 201);
+		assert(lookup(300) == -1);
+
+		insert(100, 102);
+		assert(lookup(100) == 102);
+		assert(lookup(200) == 201);
+		assert(lookup(300) == -1);
+
+		assert(deletedKeys.size() == 1);
+		assert(deletedKeys[0] == 100);
+		assert(deletedValues[0] == 101);
+	}
+
+	void erase()
+	{
+
+	}
+
+private:
+	static const int kCacheSize = 1000;
+	std::vector<int> deletedKeys;
+	std::vector<int> deletedValues;
+	std::shared_ptr<ShardedLRUCache> cache;
+};
+
+int main(int argc, char *argv[])
+{
+	CacheTest test;
+	test.hitAndMiss();
+	test.erase();
+	return 0;
+
+}

example/leveldb/example/cachetest.cc

@@ -0,0 +1,3 @@
+#pragma once
+#include "dbimpl.h"
+

example/leveldb/example/tabletest.cc

@@ -134,11 +134,13 @@ class BytewiseComparatorImpl : public Comparator
 	{
 
 	}
+
 	virtual const char *name() const { return "leveldb.BytewiseComparator"; }
 	virtual int compare(const std::string_view &a, const std::string_view &b) const
 	{
 		return a.compare(b);
 	}
+
 	virtual void findShortestSeparator(std::string *start, const std::string_view &limit) const;
 	virtual void findShortSuccessor(std::string *key) const;
 };

example/leveldb/example/writebatchtest.cc

@@ -13,6 +13,16 @@
 #include "posix.h"
 #include "tablecache.h"
 
+// A range of keys
+struct Range
+{
+  std::string_view start;          // Included in the range
+  std::string_view limit;          // Not included in the range
+
+  Range() { }
+  Range(const std::string_view &s, const std::string_view &l) : start(s), limit(l) { }
+};
+
 class DBImpl
 {
 public:
@@ -21,12 +31,32 @@ class DBImpl
 
 	Status open();
 	Status newDB();
+	// Set the database entry for "key" to "value".  Returns OK on success,
+	// and a non-OK status on error.
+	// Note: consider setting options.sync = true.
 	Status put(const WriteOptions&, const std::string_view &key, const std::string_view &value);
+	// Remove the database entry (if any) for "key".  Returns OK on
+	// success, and a non-OK status on error.  It is not an error if "key"
+	// did not exist in the database.
+	// Note: consider setting options.sync = true.
 	Status del(const WriteOptions&, const std::string_view &key);
+	// Apply the specified updates to the database.
+	// Returns OK on success, non-OK on failure.
+	// Note: consider setting options.sync = true.
 	Status write(const WriteOptions &options, WriteBatch *updates);
+	// If the database contains an entry for "key" store the
+	// corresponding value in *value and return OK.
+	//
+	// If there is no entry for "key" leave *value unchanged and return
+	// a status for which Status::IsNotFound() returns true.
+	//
+	// May return some other Status on an error.
 	Status get(const ReadOptions &options, const std::string_view &key, std::string *value);
 	Status destroyDB(const std::string &dbname, const Options &options);
 	void deleteObsoleteFiles();
+	// Recover the descriptor from persistent storage.  May do a significant
+	// amount of work to recover recently logged updates.  Any changes to
+	// be made to the descriptor are added to *edit.
 	Status recover(VersionEdit *edit, bool *saveManifest);
 	Status recoverLogFile(uint64_t logNumber, bool lastLog,
 		bool *saveManifest, VersionEdit *edit, uint64_t *maxSequence);
@@ -37,10 +67,42 @@ class DBImpl
 	void compactMemTable();
 	std::shared_ptr<Iterator> newInternalIterator(const ReadOptions &options,
 		uint64_t *latestSnapshot, uint32_t *seed); 
+	// Return a heap-allocated iterator over the contents of the database.
+	// The result of NewIterator() is initially invalid (caller must
+	// call one of the Seek methods on the iterator before using it).
+	//
+	// Caller should delete the iterator when it is no longer needed.
+	// The returned iterator should be deleted before this db is deleted.
 	std::shared_ptr<Iterator> newIterator(const ReadOptions &options);
 	Options sanitizeOptions(const std::string &dbname,
 	                    const InternalKeyComparator *icmp,
 	                    const Options &src);
+	// DB implementations can export properties about their state
+	// via this method.  If "property" is a valid property understood by this
+	// DB implementation, fills "*value" with its current value and returns
+	// true.  Otherwise returns false.
+	//
+	//
+	// Valid property names include:
+	//
+	//  "leveldb.num-files-at-level<N>" - return the number of files at level <N>,
+	//     where <N> is an ASCII representation of a level number (e.g. "0").
+	//  "leveldb.stats" - returns a multi-line string that describes statistics
+	//     about the internal operation of the DB.
+	//  "leveldb.sstables" - returns a multi-line string that describes all
+	//     of the sstables that make up the db contents.
+	//  "leveldb.approximate-memory-usage" - returns the approximate number of
+	//     bytes of memory in use by the DB.
+	bool getProperty(const std::string_view &property, std::string *value);
+	// For each i in [0,n-1], store in "sizes[i]", the approximate
+	// file system space used by keys in "[range[i].start .. range[i].limit)".
+	//
+	// Note that the returned sizes measure file system space usage, so
+	// if the user data compresses by a factor of ten, the returned
+	// sizes will be one-tenth the size of the corresponding user data size.
+	//
+	// The results may not include the sizes of recently written data.
+	void getApproximateSizes(const Range *range, int n, uint64_t* sizes);
 
 	 // Force current memtable contents to be compacted.
 	Status testCompactMemTable();

example/leveldb/src/dbformat.h

@@ -0,0 +1,98 @@
+#include <stdint.h>
+#include <stdio.h>
+
+#include <algorithm>
+#include <set>
+#include <string>
+#include <vector>
+#include <assert.h>
+#include <random>
+
+#include "dbimpl.h"
+#include "filename.h"
+#include "logwriter.h"
+#include "logreader.h"
+#include "tablecache.h"
+#include "versionset.h"
+#include "status.h"
+#include "table.h"
+#include "tablebuilder.h"
+#include "block.h"
+#include "merger.h"
+#include "option.h"
+
+const int kNumNonTableCacheFiles = 10;
+
+std::default_random_engine e;
+
+std::string makeKey(unsigned int num) 
+{
+	char buf[30];
+	snprintf(buf, sizeof(buf), "%016u", num);
+	return std::string(buf);
+}
+
+class DBTest
+{
+public:
+	std::string dbname;
+	DBImpl db;
+};
+
+void bmLogAndApply(int iters, int numbasefiles)
+{
+	Options opts;
+	opts.createIfMissing = true;
+	std::string dbname = "./leveldb_test_benchmark";
+	DBImpl db(opts, dbname);	 
+	db.destroyDB(dbname, opts);
+
+	Status s = db.open();
+	assert(s.ok());
+	
+	BytewiseComparatorImpl byteImpl;
+	InternalKeyComparator cmp(&byteImpl);
+	Options options;
+	VersionSet vset(dbname, options, nullptr, &cmp);
+	bool manifest;
+	assert(vset.recover(&manifest).ok());
+	
+	VersionEdit vbase;
+	uint64_t fnum = 1;
+	for (int i = 0; i < numbasefiles; i++) 
+	{
+		InternalKey start(makeKey(2 * fnum), 1, kTypeValue);
+		InternalKey limit(makeKey(2 * fnum + 1), 1, kTypeDeletion);
+		vbase.addFile(2, fnum++, 1 /* file size */, start, limit);
+	}
+  
+	assert(vset.logAndApply(&vbase).ok());
+	uint64_t startMicros = options.env->nowMicros();
+	
+	for (int i = 0; i < iters; i++) 
+	{
+		VersionEdit vedit;
+		vedit.deleteFile(2, fnum);
+		InternalKey start(makeKey(2 * fnum), 1, kTypeValue);
+		InternalKey limit(makeKey(2 * fnum + 1), 1, kTypeDeletion);
+		vedit.addFile(2, fnum++, 1 /* file size */, start, limit);
+		vset.logAndApply(&vedit);
+	}
+ 
+	uint64_t stopMicros = options.env->nowMicros();
+	unsigned int us = stopMicros - startMicros;
+	char buf[16];
+	snprintf(buf, sizeof(buf), "%d", numbasefiles);
+	fprintf(stderr,
+	  "BM_LogAndApply/%-6s   %8d iters : %9u us (%7.0f us / iter)\n",
+	  buf, iters, us, ((float)us) / iters);
+}
+
+int main()
+{
+	bmLogAndApply(1000, 1);
+	bmLogAndApply(1000, 100);
+	bmLogAndApply(1000, 10000);
+	bmLogAndApply(100, 100000);
+	return 0;
+}

example/leveldb/src/dbimpl.cc

@@ -26,4 +26,7 @@ Status readFileToString(PosixEnv *env, const std::string &fname, std::string *da
 
 Status writeStringToFileSync(PosixEnv *env, const std::string &data, const std::string &fname);
 
+// Return a human-readable version of "value".
+// Escapes any non-printable characters found in "value".
 std::string escapeString(const std::string &value);
+

example/leveldb/src/dbimpl.h

@@ -0,0 +1,21 @@
+#include "option.h"
+
+// Create an Options object with default values for all fields.
+
+static BytewiseComparatorImpl byteImpl;
+Options::Options()
+	:comparator(&byteImpl),
+	createIfMissing(false),
+	errorIfExists(false),
+	paranoidChecks(false),
+	env(new PosixEnv()),
+	writeBufferSize(4 << 20),
+	maxOpenFiles(1000),
+	blockSize(4096),
+	blockRestartInterval(16),
+	maxFileSize(2 << 20),
+	compression(kNoCompression),
+	reuseLogs(false)
+{
+
+}	

example/leveldb/src/dbtest.cc

@@ -93,7 +93,7 @@ Status PosixWritableFile::syncDirIfManifest()
 		}
 		else
 		{
-			if (fsync(fd) < 0)
+			if (::fsync(fd) < 0)
 			{
 				s = posixError(dir, errno);
 			}
@@ -112,14 +112,19 @@ Status PosixWritableFile::sync()
 		return s;
 	}
 
+#ifdef __APPLE__
 	s = flushBuffered();
-	if (s.ok())
+	if (s.ok() && ::fsync(fd) != 0)
 	{
-		if (::fsync(fd) != 0)
-		{
-			s = posixError(filename, errno);
-		}
+		s = posixError(filename, errno);
+	}
+#else
+	s = flushBuffered();
+	if (s.ok() && ::fdatasync(fd) != 0)
+	{
+		s = posixError(filename, errno);
 	}
+#endif
 	return s;
 }
 

example/leveldb/src/logging.h

@@ -1,104 +1,104 @@
-#include "tablecache.h"
-#include "filename.h"
-#include "coding.h"
-#include "table.h"
-
-struct TableAndFile 
-{
-	std::shared_ptr<RandomAccessFile> file;
-	std::shared_ptr<Table> table;
-};
-
-static void deleteEntry(const std::string_view &key, std::any &value) 
-{
-	
-}
-
-TableCache::TableCache(const std::string &dbname, const Options &options, int entries)
-:options(options),
- dbname(dbname),
- cache(new ShardedLRUCache(entries))
-{
-
-}
-
-TableCache::~TableCache()
-{
-
-}
-
-Status TableCache::findTable(uint64_t fileNumber, uint64_t fileSize,
-		  std::shared_ptr<LRUHandle> &handle)
-{
-	Status s;
-	char buf[sizeof(fileNumber)];
-	encodeFixed64(buf, fileNumber);
-	std::string_view key(buf, sizeof(buf));
-	handle = cache->lookup(key);
-	if (handle == nullptr)
-	{
-		std::string fname = tableFileName(dbname, fileNumber);
-		std::shared_ptr<RandomAccessFile> file = nullptr;
-		std::shared_ptr<Table> table = nullptr;
-		s = options.env->newRandomAccessFile(fname, file);
-		if (s.ok()) 
-		{
-			s = Table::open(options, file, fileSize, table);
-		}
-
-		if (!s.ok()) 
-		{
-			assert(table == nullptr);
-			// We do not cache error results so that if the error is transient,
-			// or somebody repairs the file, we recover automatically.
-		}
-		else 
-		{
-			std::shared_ptr<TableAndFile> tf(new TableAndFile);
-			tf->file = file;
-			tf->table = table;
-			handle = cache->insert(key, tf, 1, nullptr);
-			printf("Table cache is open %s\n", fname.c_str());
-		}
-	}
-	return s;
-}
-
-std::shared_ptr<Iterator> TableCache::newIterator(const ReadOptions &options,
-	uint64_t fileNumber,
-	uint64_t fileSize,
-	std::shared_ptr<Table> tableptr)
-{
-	std::shared_ptr<LRUHandle> handle;
-	Status s = findTable(fileNumber, fileSize, handle);
-	if (!s.ok())
-	{
-		return newErrorIterator(s);
-	}
-
-	std::shared_ptr<Table> table = std::any_cast<std::shared_ptr<TableAndFile>>(handle->value)->table;
-	std::shared_ptr<Iterator> result = table->newIterator(options);
-	result->registerCleanup(handle);
-	tableptr = table;
-	return result;
-}
-
-Status TableCache::get(const ReadOptions &options,
-		 uint64_t fileNumber,
-		 uint64_t fileSize,
-		 const std::string_view &k,
-		 const std::any &arg,
-		 std::function<void( const std::any &,
-		 const std::string_view &,  const std::string_view &)> &&callback)
-{
-	std::shared_ptr<LRUHandle> handle;
-	Status s = findTable(fileNumber, fileSize, handle);
-	if (s.ok())
-	{
-		//printf("Table cache get file number :%d bytes %lld\n", fileNumber, fileSize);
-		std::shared_ptr<Table> table = std::any_cast<std::shared_ptr<TableAndFile>>(handle->value)->table;
-		s = table->internalGet(options, k, arg, callback);
-	}
-	
-	return s;
-}
+#include "tablecache.h"
+#include "filename.h"
+#include "coding.h"
+#include "table.h"
+
+struct TableAndFile 
+{
+	std::shared_ptr<RandomAccessFile> file;
+	std::shared_ptr<Table> table;
+};
+
+static void deleteEntry(const std::string_view &key, std::any &value) 
+{
+	
+}
+
+TableCache::TableCache(const std::string &dbname, const Options &options, int entries)
+:options(options),
+ dbname(dbname),
+ cache(new ShardedLRUCache(entries))
+{
+
+}
+
+TableCache::~TableCache()
+{
+
+}
+
+Status TableCache::findTable(uint64_t fileNumber, uint64_t fileSize,
+		  std::shared_ptr<LRUHandle> &handle)
+{
+	Status s;
+	char buf[sizeof(fileNumber)];
+	encodeFixed64(buf, fileNumber);
+	std::string_view key(buf, sizeof(buf));
+	handle = cache->lookup(key);
+	if (handle == nullptr)
+	{
+		std::string fname = tableFileName(dbname, fileNumber);
+		std::shared_ptr<RandomAccessFile> file = nullptr;
+		std::shared_ptr<Table> table = nullptr;
+		s = options.env->newRandomAccessFile(fname, file);
+		if (s.ok()) 
+		{
+			s = Table::open(options, file, fileSize, table);
+		}
+
+		if (!s.ok()) 
+		{
+			assert(table == nullptr);
+			// We do not cache error results so that if the error is transient,
+			// or somebody repairs the file, we recover automatically.
+		}
+		else 
+		{
+			std::shared_ptr<TableAndFile> tf(new TableAndFile);
+			tf->file = file;
+			tf->table = table;
+			handle = cache->insert(key, tf, 1, nullptr);
+			printf("Table cache is open %s\n", fname.c_str());
+		}
+	}
+	return s;
+}
+
+std::shared_ptr<Iterator> TableCache::newIterator(const ReadOptions &options,
+	uint64_t fileNumber,
+	uint64_t fileSize,
+	std::shared_ptr<Table> tableptr)
+{
+	std::shared_ptr<LRUHandle> handle;
+	Status s = findTable(fileNumber, fileSize, handle);
+	if (!s.ok())
+	{
+		return newErrorIterator(s);
+	}
+
+	std::shared_ptr<Table> table = std::any_cast<std::shared_ptr<TableAndFile>>(handle->value)->table;
+	std::shared_ptr<Iterator> result = table->newIterator(options);
+	result->registerCleanup(handle);
+	tableptr = table;
+	return result;
+}
+
+Status TableCache::get(const ReadOptions &options,
+		 uint64_t fileNumber,
+		 uint64_t fileSize,
+		 const std::string_view &k,
+		 const std::any &arg,
+		 std::function<void( const std::any &,
+		 const std::string_view &,  const std::string_view &)> &&callback)
+{
+	std::shared_ptr<LRUHandle> handle;
+	Status s = findTable(fileNumber, fileSize, handle);
+	if (s.ok())
+	{
+		//printf("Table cache get file number :%d bytes %lld\n", fileNumber, fileSize);
+		std::shared_ptr<Table> table = std::any_cast<std::shared_ptr<TableAndFile>>(handle->value)->table;
+		s = table->internalGet(options, k, arg, callback);
+	}
+	
+	return s;
+}

example/leveldb/src/option.cc

@@ -1,49 +1,51 @@
-#pragma once
-#include <string>
-#include <stdint.h>
-#include "dbformat.h"
-#include "cache.h"
-#include "table.h"
-#include "option.h"
-
-//, std::bind(&Version::saveValue, this, 
-//				std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)
-//std::function<void(const std::any &arg, 
-//				 const std::string_view &k, const std::any &value)> &&callback
-class TableCache
-{
-public:
-	 TableCache(const std::string &dbname, const Options &options, int entries);
-	  ~TableCache();
-
-	// Return an iterator for the specified file number (the corresponding
-	// file length must be exactly "file_size" bytes).  If "tableptr" is
-	// non-null, also sets "*tableptr" to point to the Table object
-	// underlying the returned iterator, or to nullptr if no Table object
-	// underlies the returned iterator.  The returned "*tableptr" object is owned
-	// by the cache and should not be deleted, and is valid for as long as the
-	// returned iterator is live.
-	std::shared_ptr<Iterator> newIterator(const ReadOptions &options,
-	                uint64_t fileNumber,
-	                uint64_t fileSize,
-	                std::shared_ptr<Table> tableptr = nullptr);
-	  
-	  // If a seek to internal key "k" in specified file finds an entry,
-	  // call (*handle_result)(arg, found_key, found_value).
-	  Status get(const ReadOptions &options,
-	             uint64_t fileNumber,
-	             uint64_t fileSize,
-	             const std::string_view &k,
-	             const std::any &arg,
-				 std::function<void( const std::any &,
-				 const std::string_view &,  const std::string_view &)> &&callback);
-	  Status findTable(uint64_t fileNumber, uint64_t fileSize,
-			  std::shared_ptr<LRUHandle> &handle);
-private:
-	TableCache(const TableCache&) = delete;
-	void operator=(const TableCache&) = delete;
-
-	std::string dbname;
-	const Options options;
-	std::shared_ptr<ShardedLRUCache> cache;
-};
+#pragma once
+#include <string>
+#include <stdint.h>
+#include "dbformat.h"
+#include "cache.h"
+#include "table.h"
+#include "option.h"
+
+//, std::bind(&Version::saveValue, this, 
+//				std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)
+//std::function<void(const std::any &arg, 
+//				 const std::string_view &k, const std::any &value)> &&callback
+class TableCache
+{
+public:
+	 TableCache(const std::string &dbname, const Options &options, int entries);
+	  ~TableCache();
+
+	// Return an iterator for the specified file number (the corresponding
+	// file length must be exactly "file_size" bytes).  If "tableptr" is
+	// non-null, also sets "*tableptr" to point to the Table object
+	// underlying the returned iterator, or to nullptr if no Table object
+	// underlies the returned iterator.  The returned "*tableptr" object is owned
+	// by the cache and should not be deleted, and is valid for as long as the
+	// returned iterator is live.
+	std::shared_ptr<Iterator> newIterator(const ReadOptions &options,
+	                uint64_t fileNumber,
+	                uint64_t fileSize,
+	                std::shared_ptr<Table> tableptr = nullptr);
+	  
+	  // If a seek to internal key "k" in specified file finds an entry,
+	  // call (*handle_result)(arg, found_key, found_value).
+	  Status get(const ReadOptions &options,
+	             uint64_t fileNumber,
+	             uint64_t fileSize,
+	             const std::string_view &k,
+	             const std::any &arg,
+				 std::function<void( const std::any &,
+				 const std::string_view &,  const std::string_view &)> &&callback);
+	  Status findTable(uint64_t fileNumber, uint64_t fileSize,
+			  std::shared_ptr<LRUHandle> &handle);
+	std::shared_ptr<ShardedLRUCache> getCache() { return cache; }
+
+private:
+	TableCache(const TableCache&) = delete;
+	void operator=(const TableCache&) = delete;
+
+	std::string dbname;
+	const Options options;
+	std::shared_ptr<ShardedLRUCache> cache;
+};

example/leveldb/src/posix.cc

@@ -511,6 +511,34 @@ void Version::getOverlappingInputs(
 	}
 }
 
+std::string Version::debugString() const
+{
+	std::string r;
+	for (int level = 0; level < kNumLevels; level++)
+	{
+		// E.g.,
+		//   --- level 1 ---
+		//   17:123['a' .. 'd']
+		//   20:43['e' .. 'g']
+		r.append("--- level ");
+		appendNumberTo(&r, level);
+		r.append(" ---\n");
+		const std::vector<std::shared_ptr<FileMetaData>> &f = files[level];
+		for (size_t i = 0; i < f.size(); i++) {
+			r.push_back(' ');
+			appendNumberTo(&r, f[i]->number);
+			r.push_back(':');
+			appendNumberTo(&r, f[i]->fileSize);
+			r.append("[");
+			r.append(f[i]->smallest.debugString());
+			r.append(" .. ");
+			r.append(f[i]->largest.debugString());
+			r.append("]\n");
+		}
+	}
+	return r;
+}
+
 int Version::pickLevelForMemTableOutput(const std::string_view &smallestUserKey,
 	const std::string_view &largestUserKey)
 {
@@ -866,6 +894,13 @@ void VersionSet::markFileNumberUsed(uint64_t number)
 	}
 }
 
+int64_t VersionSet::numLevelBytes(int level) const
+{
+	assert(level >= 0);
+	assert(level < kNumLevels);
+	return totalFileSize(current()->files[level]);
+}
+
 int VersionSet::numLevelFiles(int level) const
 {
 	assert(level >= 0);
@@ -930,6 +965,10 @@ Status VersionSet::logAndApply(VersionEdit *edit)
 			descriptorLog.reset(new LogWriter(descriptorFile.get()));
 			//s = WriteSnapshot(descriptor_log_);
 		}
+		else
+		{
+
+		}
 	}
 
 	// Write new record to MANIFEST log
@@ -949,6 +988,8 @@ Status VersionSet::logAndApply(VersionEdit *edit)
 		}
 	}
 
+	descriptorLog.reset();
+	descriptorFile.reset();
 	// If we just created a new descriptor file, install it by writing a
 	// new CURRENT file that points to it.
 	if (s.ok() && !newManifestFile.empty())

example/leveldb/src/tablecache.cc

@@ -67,6 +67,9 @@ class Version
 	// REQUIRES: lock is held
 	bool updateStats(const GetStats &stats);
 
+	// Return a human readable string that describes this version's contents.
+	std::string debugString() const;
+
 	// No copying allowed
 	Version(const Version&);
 	void operator=(const Version&);
@@ -197,7 +200,10 @@ class VersionSet
 
 	void appendVersion(const std::shared_ptr<Version> &v);
 	bool reuseManifest(const std::string &dscname, const std::string &dscbase);
+	// Return the number of Table files at the specified level.
 	int numLevelFiles(int level) const;
+	// Return the combined file size of all files at the specified level.
+	int64_t numLevelBytes(int level) const;
 
 	std::shared_ptr<Iterator> makeInputIterator(Compaction *c);
 

example/leveldb/src/tablecache.h

@@ -0,0 +1,218 @@
+#include <unistd.h>
+#include <memory>
+#include <fstream>
+#include <iostream>
+#include <unordered_map>
+#include <map>
+#include <vector>
+#include <set>
+#include <assert.h>
+#include <algorithm>
+
+int32_t index = 0;
+const size_t kMaxSize = 10 * 1000 * 1000;
+
+class Sharder
+{
+public:
+	explicit Sharder(int32_t nbuckets)
+	{
+		for (int32_t i = 0; i < nbuckets; ++i)
+		{
+			char buf[256];
+			snprintf(buf, sizeof buf, "shard-%05d-of-%05d", i, nbuckets);
+			std::shared_ptr<std::ofstream> in(new std::ofstream(buf));
+			buckets.push_back(in);
+		}
+		assert(buckets.size() == static_cast<size_t>(nbuckets));
+	}
+
+	void output(const std::string &key, const std::string &value)
+	{
+		size_t idx = std::hash<std::string>()(key) % buckets.size();
+		*buckets[idx] << key << '\t' << value << '\n';
+	}
+
+protected:
+	std::vector<std::shared_ptr<std::ofstream>> buckets;
+private:
+	Sharder(const Sharder&);
+	void operator=(const Sharder&);
+};
+
+void shard(int32_t nbuckets, int32_t argc, char *argv[])
+{
+	Sharder sharder(nbuckets);
+	for (int32_t i = 1; i < argc; ++i)
+	{
+		std::cout << "  processing input file " << argv[i] << std::endl;
+		std::map<std::string, std::string> shareds;
+		std::ifstream in(argv[i]);
+		while (in && !in.eof())
+		{
+			shareds.clear();
+			std::string str;
+			while (in >> str)
+			{
+				shareds.insert(std::make_pair(str, str));
+				if (shareds.size() > kMaxSize)
+				{
+					std::cout << "    split" << std::endl;
+					break;
+				}
+			}
+
+			for (const auto &kv : shareds)
+			{
+				sharder.output(kv.first, kv.second);
+			}
+		}
+	}
+	std::cout << "shuffling done" << std::endl;
+}
+
+// ======= sortShareds =======
+
+std::map<std::string, std::string> readShard(int32_t idx, int32_t nbuckets)
+{
+	std::map<std::string, std::string> shareds;
+
+	char buf[256];
+	snprintf(buf, sizeof buf, "shard-%05d-of-%05d", idx, nbuckets);
+	std::cout << "  reading " << buf << std::endl;
+	{
+		std::ifstream in(buf);
+		std::string line;
+
+		while (getline(in, line))
+		{
+			size_t tab = line.find('\t');
+			assert(tab != std::string::npos);
+			std::string key(line.c_str(), line.c_str() + tab);
+			std::string value(line.c_str() + tab + 1, line.c_str() + line.size());
+			shareds.insert(std::make_pair(key, value));
+		}
+	}
+
+	::unlink(buf);
+	return shareds;
+}
+
+void sortShareds(const int32_t nbuckets)
+{
+	for (int32_t i = 0; i < nbuckets; ++i)
+	{
+		// std::cout << "  sorting " << std::endl;
+		std::map<std::string, std::string> shareds;
+		for (const auto &entry : readShard(i, nbuckets))
+		{
+			shareds.insert(std::make_pair(entry.first, entry.second));
+		}
+
+		char buf[256];
+		snprintf(buf, sizeof buf, "count-%05d-of-%05d", i, nbuckets);
+		std::ofstream out(buf);
+		std::cout << "  writing " << buf << std::endl;
+		for (auto &it : shareds)
+		{
+			out << it.first << '\t' << it.second << '\n';
+		}
+	}
+	std::cout << "reducing done" << std::endl;
+}
+
+// ======= merge =======
+
+class Source  // copyable
+{
+public:
+	explicit Source(const std::shared_ptr<std::ifstream> &in)
+		: in(in)
+	{
+
+	}
+
+	bool next()
+	{
+		std::string line;
+		if (getline(*(in.get()), line))
+		{
+			size_t tab = line.find('\t');
+			if (tab != std::string::npos)
+			{
+				std::string key(line.c_str(), line.c_str() + tab);
+				this->key = key;
+				std::string(line.c_str() + tab + 1, line.c_str() + line.size());
+				this->value = value;
+				return true;
+			}
+		}
+		return false;
+	}
+
+	bool operator<(const Source &rhs) const
+	{
+		return key < rhs.key;
+	}
+	
+	void retrieve()
+	{
+		index += key.size() + 3 + value.size() + sizeof(int32_t)
+	}
+
+	void outputTo(std::ostream &out, int32_t index) const
+	{
+		out << key << '\t' << value << '\t' << std::to_string(index) << '\n';
+		retrieve();
+	}
+
+	std::shared_ptr<std::ifstream> in;
+	std::string key;
+	std::string value;
+};
+
+void merge(const int32_t nbuckets)
+{
+	std::vector<std::shared_ptr<std::ifstream>> inputs;
+	std::vector<Source> keys;
+
+	for (int32_t i = 0; i < nbuckets; ++i)
+	{
+		char buf[256];
+		snprintf(buf, sizeof buf, "count-%05d-of-%05d", i, nbuckets);
+		std::shared_ptr<std::ifstream> in(new std::ifstream(buf));
+		inputs.push_back(in);
+		Source rec(in);
+		if (rec.next())
+		{
+			keys.push_back(rec);
+		}
+		::unlink(buf);
+	}
+
+	std::ofstream out("output");
+	std::make_heap(keys.begin(), keys.end());
+
+	while (!keys.empty())
+	{
+		std::pop_heap(keys.begin(), keys.end());
+		keys.back().outputTo(out, index);
+	
+		if (keys.back().next())
+		{
+			std::push_heap(keys.begin(), keys.end());
+		}
+		else
+		{
+			keys.pop_back();
+		}
+	}
+}
+
+int32_t main(int32_t argc, char *argv[])
+{
+	int32_t nbuckets = 10;
+	shard(nbuckets, argc, argv);
+	sortShareds(nbuckets);
+	merge(nbuckets);
+}

example/leveldb/src/versionset.cc

@@ -143,7 +143,7 @@ class Source  // copyable
 	bool next()
 	{
 		std::string line;
-		if (getline(*in, line))
+		if (getline(*(in.get()), line))
 		{
 			size_t tab = line.find('\t');
 			if (tab != std::string::npos)
@@ -184,7 +184,7 @@ void merge(const int32_t nbuckets)
 		snprintf(buf, sizeof buf, "count-%05d-of-%05d", i, nbuckets);
 		std::shared_ptr<std::ifstream> in(new std::ifstream(buf));
 		inputs.push_back(in);
-		Source rec(inputs.back());
+		Source rec(in);
 		if (rec.next())
 		{
 			keys.push_back(rec);
@@ -208,6 +208,7 @@ void merge(const int32_t nbuckets)
 		if (++cnt >= topK)
 		{
 			keys.pop_back();
+			break;
 		}
 
 		if (keys.back().next())

example/leveldb/src/versionset.h

@@ -33,7 +33,7 @@ void Item::output(Buffer *out, bool needCas) const
 {
 	out->append("VALUE ");
 	out->append(data, keyLen);
-	xLogStream buf;
+	LogStream buf;
 	buf << ' ' << getFlags() << ' ' << valueLen - 2;
 	if (needCas)
 	{
@@ -581,17 +581,17 @@ bool MemcacheServer::storeItem(const ItemPtr &item, Item::UpdatePolicy policy, b
 
 ConstItemPtr MemcacheServer::getItem(const ConstItemPtr &key) const
 {
-	std::mutex & mutex = shards[key->getHash() % kShards].mutex;
-	const ItemMap& items = shards[key->getHash() % kShards].items;
+	std::mutex &mutex = shards[key->getHash() % kShards].mutex;
+	const ItemMap &items = shards[key->getHash() % kShards].items;
 	std::unique_lock <std::mutex> lck(mutex);
 	ItemMap::const_iterator it = items.find(key);
 	return it != items.end() ? *it : ConstItemPtr();
 }
 
 bool MemcacheServer::deleteItem(const ConstItemPtr &key)
 {
-	std::mutex & mutex = shards[key->getHash() % kShards].mutex;
-	ItemMap& items = shards[key->getHash() % kShards].items;
+	std::mutex &mutex = shards[key->getHash() % kShards].mutex;
+	ItemMap &items = shards[key->getHash() % kShards].items;
 	std::unique_lock <std::mutex> lck(mutex);
 	return items.erase(key) == 1;
 }

example/mapreduce/sort/filesort.cc

@@ -98,10 +98,7 @@ void RedisProxy::initRedisTimer()
 	{
 		auto it = threadHiredis.find(pools[i]->getThreadId());
 		assert(it != threadHiredis.end());
-		if (clusterEnabled)
-		{
-			pools[i]->runAfter(1.0, true, std::bind(&Hiredis::redisContextTimer, it->second.get()));
-		}
+		pools[i]->runAfter(1.0, true, std::bind(&Hiredis::redisContextTimer, it->second.get()));
 	}
 }
 
@@ -335,12 +332,13 @@ void RedisProxy::proxyCallback(const RedisAsyncContextPtr &c,
 	RedisCluster proxy = std::any_cast<RedisCluster>(privdata);
 
 	int64_t proxyCount = proxy.proxyCount;
-	int32_t commandCount = proxy.commandCount;
+	int64_t commandCount = proxy.commandCount;
 	RedisObjectPtr command = proxy.command;
 	WeakTcpConnectionPtr weakConn = proxy.conn;
 	TcpConnectionPtr conn = weakConn.lock();
 	if (conn == nullptr)
 	{	
+		LOG_WARN << "proxyCallback client disconnect err: ";
 		return;
 	}
 
@@ -602,26 +600,26 @@ RedisAsyncContextPtr RedisProxy::checkReply(const RedisObjectPtr &command, const
 	auto it = threadHiredis.find(conn->getLoop()->getThreadId());
 	assert(it != threadHiredis.end());
 
-	RedisAsyncContextPtr redis = nullptr;
+	RedisAsyncContextPtr redisAsync = nullptr;
 	if (clusterEnabled)
 	{
-		redis = it->second->getRedisAsyncContext(command,
+		redisAsync = it->second->getRedisAsyncContext(command,
 				conn->getLoop()->getThreadId(), conn->getSockfd());
 	}
 	else
 	{
-		redis = it->second->getRedisAsyncContext(
+		redisAsync = it->second->getRedisAsyncContext(
 				conn->getLoop()->getThreadId(), conn->getSockfd());
 	}
 
-	if (redis == nullptr)
+	if (redisAsync == nullptr)
 	{
 		clearProxy(conn->getLoop()->getThreadId(), conn->getSockfd());
 		std::string reply = it->second->getTcpClientInfo(
 			conn->getLoop()->getThreadId(), conn->getSockfd());
 		conn->sendPipe(reply.c_str(), reply.size());
 	}
-	return redis;
+	return redisAsync;
 }
 
 bool RedisProxy::debugCommand(const RedisObjectPtr &command, const std::vector<RedisObjectPtr> &commands,

example/mapreduce/sort/shardsbasic.cc

@@ -44,6 +44,7 @@
 #include <experimental/filesystem>
 #include <ratio>
 #include <chrono>
+#include <random>
 #ifdef _LUA
 #include <lua.hpp>
 #endif
@@ -114,9 +115,19 @@ typedef SSIZE_T ssize_t;
 #include <sys/utsname.h>
 #endif
 
-#define REDIS_CONNECT_RETRIES  10
+#define LUA_TNONE		(-1)
 
+#define LUA_TNIL		0
+#define LUA_TBOOLEAN		1
+#define LUA_TLIGHTUSERDATA	2
+#define LUA_TNUMBER		3
+#define LUA_TSTRING		4
+#define LUA_TTABLE		5
+#define LUA_TFUNCTION		6
+#define LUA_TUSERDATA		7
+#define LUA_TTHREAD		8
 
+#define REDIS_CONNECT_RETRIES  10
 /* Flag specific to the async API which means that the context should be clean
  * up as soon as possible. */
 #define REDIS_FREEING	10

example/memcached/server/server.cc

@@ -156,7 +156,7 @@ class Buffer
 
 	void append(const char *data)
 	{
-		int32_t len = strlen(data) + 1;
+		int32_t len = strlen(data);
 		ensureWritableBytes(len);
 		std::copy(data, data + len, beginWrite());
 		hasWritten(len);

example/redisproxy/redisproxy.cc

@@ -676,7 +676,8 @@ int32_t RedisReader::redisReaderGetReply(RedisReplyPtr &reply)
 
 RedisAsyncCallback::RedisAsyncCallback()
 	:data(nullptr),
-	len(0)
+	len(0),
+	type(false)
 {
 
 }
@@ -685,7 +686,14 @@ RedisAsyncCallback::~RedisAsyncCallback()
 {
 	if (data != nullptr)
 	{
-		zfree(data);
+		if (type)
+		{
+			sdsfree(data);
+		}
+		else
+		{
+			zfree(data);
+		}
 	}
 }
 
@@ -2030,6 +2038,30 @@ uint32_t Hiredis::keyHashSlot(char *key, int32_t keylen)
 	return crc16(key + s + 1, e - s - 1) & 0x3FFF;
 }
 
+int32_t RedisAsyncContext::redisAsyncCommandArgv(const RedisCallbackFn &fn,
+		const std::any &privdata, int32_t argc, const char **argv, const int32_t *argvlen)
+{
+	sds cmd;
+	int32_t len;
+	int32_t status;
+	len = redisFormatSdsCommandArgv(&cmd, argc, argv, argvlen);
+	RedisCallback cb;
+	cb.fn = std::move(fn);
+	cb.privdata = std::move(privdata);
+
+	RedisAsyncCallbackPtr asyncCallback(new RedisAsyncCallback());
+	asyncCallback->data = cmd;
+	asyncCallback->len = len;
+	asyncCallback->cb = std::move(cb);
+	asyncCallback->type = true;
+
+	TcpConnectionPtr conn = weakRedisConn.lock();
+	assert(conn != nullptr);
+	conn->getLoop()->runInLoop(std::bind(&RedisAsyncContext::__redisAsyncCommand,
+		shared_from_this(), asyncCallback));
+	return REDIS_OK;
+}
+
 RedisAsyncContextPtr Hiredis::getRedisAsyncContext(const RedisObjectPtr &command,
 		const std::thread::id &threadId, int32_t sockfd)
 {
@@ -2126,6 +2158,16 @@ RedisAsyncContextPtr Hiredis::getRedisAsyncContext(const std::thread::id &thread
 	}
 }
 
+RedisContextPtr Hiredis::getRedisContext(const int32_t sockfd)
+{
+	std::unique_lock<std::mutex> lk(mutex);
+	if (redisContexts.empty())
+	{
+		return nullptr;
+	}
+	return redisContexts[(redisContexts.size() - 1) % sockfd];
+}
+
 std::vector<RedisContextPtr> Hiredis::getRedisContext(const std::thread::id &threadId)
 {
 	std::unique_lock<std::mutex> lk(mutex);
@@ -2159,13 +2201,7 @@ RedisAsyncContextPtr Hiredis::getRedisAsyncContext(int32_t sockfd)
 		return nullptr;
 	}
 
-	int32_t index = (tcpClients.size() % sockfd) - 1;
-	if (index < 0 )
-	{
-		index = 0;
-	}
-	
-	const TcpConnectionPtr &conn = tcpClients[index]->getConnection();
+	const TcpConnectionPtr &conn = tcpClients[(tcpClients.size() - 1) % sockfd]->getConnection();
 	if (conn == nullptr)
 	{
 		return nullptr;
@@ -2651,104 +2687,103 @@ TcpConnectionPtr Hiredis::redirectySlot(int32_t sockfd, EventLoop *loop, const c
 void Hiredis::redisContextTimer()
 {
 	loop->assertInLoopThread();
-	std::map<int16_t, std::string> ports;
 
+	std::unique_lock<std::mutex> lk(mutex);
+	auto it = threadRedisContexts.find(loop->getThreadId());
+	assert(it != threadRedisContexts.end());
+
+	for (auto &iter : it->second)
 	{
-		std::unique_lock<std::mutex> lk(mutex);
-		auto it = threadRedisContexts.find(loop->getThreadId());
-		assert(it != threadRedisContexts.end());
+		RedisReplyPtr reply = iter->redisCommand("PING");
+		if (reply == nullptr ||
+				!(strcmp(reply->str, "PONG") == 0)
+				|| reply->type != REDIS_REPLY_STATUS)
+		{
+			iter = redisConnect(iter->ip, iter->port);
+		}
 
-		for (auto &iter : it->second)
+		if (!proxyMode)
 		{
-			RedisReplyPtr reply = iter->redisCommand("PING");
-			if (reply != nullptr)
-			{
-				assert(strcmp(reply->str, "PONG") == 0);
-				assert(reply->type == REDIS_REPLY_STATUS);
-			}
-			else
-			{
-				iter = redisConnect(iter->ip, iter->port);
-			}
+			return ;
+		}
 
-			clusterNodes.clear();
-			reply = iter->redisCommand("cluster nodes");
-			if (reply != nullptr && reply->type != REDIS_REPLY_ERROR)
+		clusterNodes.clear();
+		reply = iter->redisCommand("cluster nodes");
+		if (reply != nullptr && reply->type != REDIS_REPLY_ERROR)
+		{
+			int32_t index = 0;
+			while (index < sdslen(reply->str))
 			{
-				int32_t index = 0;
-				while (index < sdslen(reply->str))
+				std::shared_ptr<ClusterNode> node(new ClusterNode());
+				const char *id = strchr(reply->str + index, ' ');
+				assert(id != nullptr);
+
+				node->id = std::string(reply->str + index, id - (reply->str + index));
+				const char *ip = strchr(id + 1, ':');
+				assert(ip != nullptr);
+
+				node->ip = std::string(id + 1, ip - id - 1);
+				const char *port = strchr(ip + 1, '@');
+				assert(port != nullptr);
+
+				int16_t p = atoi(ip + 1);
+				node->port = p;
+				const char *p1 = strchr(port + 1, ' ');
+				assert(p1 != nullptr);
+
+				const char *myself = strchr(p1 + 1, ' ');
+				std::string m = std::string(p1 + 1, myself - p1 - 1);
+				if (m == "myself,master" || m == "master")
 				{
-					std::shared_ptr<ClusterNode> node(new ClusterNode());
-					const char *id = strchr(reply->str + index, ' ');
-					assert(id != nullptr);
-
-					node->id = std::string(reply->str + index, id - (reply->str + index));
-					const char *ip = strchr(id + 1, ':');
-					assert(ip != nullptr);
-
-					node->ip = std::string(id + 1, ip - id - 1);
-					const char *port = strchr(ip + 1, '@');
-					assert(port != nullptr);
-
-					int16_t p = atoi(ip + 1);
-					node->port = p;
-					const char *p1 = strchr(port + 1, ' ');
+					node->master = "master";
+					const char *p1 = strchr(myself + 1, ' ');
 					assert(p1 != nullptr);
-
-					const char *myself = strchr(p1 + 1, ' ');
-					std::string m = std::string(p1 + 1, myself - p1 - 1);
-					if (m == "myself,master" || m == "master")
+					const char *p2 = strchr(p1 + 1, ' ');
+					assert(p2 != nullptr);
+					const char *p3 = strchr(p2 + 1, ' ');
+					assert(p3 != nullptr);
+					const char *p4 = strchr(p3 + 1, ' ');
+					assert(p4 != nullptr);
+					const char *p5 = strchr(p4 + 1, ' ');
+					assert(p5 != nullptr);
+					std::string str = std::string(p4 + 1, p5 - p4 - 1);
+
+					if (str == "connected")
 					{
-						node->master = "master";
-						const char *p1 = strchr(myself + 1, ' ');
-						assert(p1 != nullptr);
-						const char *p2 = strchr(p1 + 1, ' ');
-						assert(p2 != nullptr);
-						const char *p3 = strchr(p2 + 1, ' ');
-						assert(p3 != nullptr);
-						const char *p4 = strchr(p3 + 1, ' ');
-						assert(p4 != nullptr);
-						const char *p5 = strchr(p4 + 1, ' ');
-						assert(p5 != nullptr);
-						std::string str = std::string(p4 + 1, p5 - p4 - 1);
-
-						if (str == "connected")
-						{
-							node->status = 1;
-						}
-						else
-						{
-							node->status = 0;
-						}
-
-						int16_t startSlot = atoi(p5 + 1);
-						node->startSlot = startSlot;
-						const char *p6 = strchr(p5 + 1, '-');
-						assert(p6 != nullptr);
-						const char *p7 = strchr(p6 + 1, '\n');
-						assert(p7 != nullptr);
-						int16_t endSlot = atoi(p6 + 1);
-						node->endSlot = endSlot;
-						index = p7 + 1 - reply->str;
+						node->status = 1;
 					}
 					else
 					{
-						node->slave = "slave";
-						const char *p1 = strchr(myself + 1, ' ');
-						assert(p1 != nullptr);
-						std::string master = std::string(myself + 1, p1 - myself - 1);
-						const char *p2 = strchr(p1 + 1, ' ');
-						assert(p2 != nullptr);
-						const char *p3 = strchr(p2 + 1, ' ');
-						assert(p3 != nullptr);
-						const char *p4 = strchr(p3 + 1, ' ');
-						assert(p4 != nullptr);
-						const char *p5 = strchr(p4 + 1, '\n');
-						assert(p5 != nullptr);
-						index = p5 + 1 - reply->str;
+						node->status = 0;
 					}
-					clusterNodes.push_back(node);
+
+					int16_t startSlot = atoi(p5 + 1);
+					node->startSlot = startSlot;
+					const char *p6 = strchr(p5 + 1, '-');
+					assert(p6 != nullptr);
+					const char *p7 = strchr(p6 + 1, '\n');
+					assert(p7 != nullptr);
+					int16_t endSlot = atoi(p6 + 1);
+					node->endSlot = endSlot;
+					index = p7 + 1 - reply->str;
+				}
+				else
+				{
+					node->slave = "slave";
+					const char *p1 = strchr(myself + 1, ' ');
+					assert(p1 != nullptr);
+					std::string master = std::string(myself + 1, p1 - myself - 1);
+					const char *p2 = strchr(p1 + 1, ' ');
+					assert(p2 != nullptr);
+					const char *p3 = strchr(p2 + 1, ' ');
+					assert(p3 != nullptr);
+					const char *p4 = strchr(p3 + 1, ' ');
+					assert(p4 != nullptr);
+					const char *p5 = strchr(p4 + 1, '\n');
+					assert(p5 != nullptr);
+					index = p5 + 1 - reply->str;
 				}
+				clusterNodes.push_back(node);
 			}
 		}
 
@@ -2770,14 +2805,14 @@ void Hiredis::redisContextTimer()
 
 					if (!mark)
 					{
-						ports[it->port] = it->ip;
+						consul[it->port] = it->ip;
 					}
 				}
 			}
 		}
 	}
 
-	for (auto &it : ports)
+	for (auto &it : consul)
 	{
 		start(loop, 0, it.second.c_str(), it.first);
 	}

src/redis/all.h

@@ -169,7 +169,8 @@ struct RedisAsyncCallback
 
 	RedisCallback cb;
 	int32_t len;
-	char *data;
+	sds data;
+	bool type;
 };
 
 /* Subscription callbacks */
@@ -209,13 +210,16 @@ class RedisAsyncContext : public std::enable_shared_from_this<RedisAsyncContext>
 	RedisAsyncContext(Buffer *buffer, const TcpConnectionPtr &conn);
 	~RedisAsyncContext();
 
+	int32_t redisAsyncCommandArgv(const RedisCallbackFn &fn,
+		const std::any &privdata, int32_t argc,
+		const char **argv, const int32_t *argvlen);
 	int32_t __redisAsyncCommand(const RedisAsyncCallbackPtr &asyncCallback);
 	int32_t redisvAsyncCommand(const RedisCallbackFn &fn,
 		const std::any &privdata, const char *format, va_list ap);
 	int32_t redisAsyncCommand(const RedisCallbackFn &fn,
 		const std::any &privdata, const char *format, ...);
-	int32_t threadProxyRedisvAsyncCommand(const RedisCallbackFn &fn, const char *data,
-		int32_t len, const std::any &privdata);
+	int32_t threadProxyRedisvAsyncCommand(const RedisCallbackFn &fn,
+		const char *data, int32_t len, const std::any &privdata);
 
 	int32_t proxyAsyncCommand(const RedisAsyncCallbackPtr &asyncCallback);
 	int32_t processCommand(const RedisCallbackFn &fn,
@@ -292,7 +296,7 @@ class Hiredis
 	RedisAsyncContextPtr getRedisAsyncContext(int32_t sockfd);
 	RedisAsyncContextPtr getRedisAsyncContext();
 	std::vector<RedisContextPtr> getRedisContext(const std::thread::id &threadId);
-	
+	RedisContextPtr getRedisContext(const int32_t sockfd);
 	RedisAsyncContextPtr getClusterRedisAsyncContext(const std::thread::id &threadId);
 	std::string getTcpClientInfo(const std::thread::id &threadId, int32_t sockfd);
 	std::string setTcpClientInfo(const char *ip, int16_t port);
@@ -312,7 +316,7 @@ class Hiredis
 	std::unordered_map<std::thread::id, std::vector<RedisContextPtr>> threadRedisContexts;
 	std::vector<TcpConnectionPtr> moveAskClients;
 	std::vector<std::shared_ptr<ClusterNode>> clusterNodes;
-
+	std::map<int16_t, std::string> consul;
 	ConnectionCallback connectionCallback;
 	DisConnectionCallback disConnectionCallback;
 	EventLoop *loop;

src/redis/buffer.h

@@ -525,7 +525,6 @@ void defaultOutput(const char* msg, int32_t len)
 	size_t n = ::fwrite(msg, 1, len, stdout);
 	//FIXME check n
 	(void)n;
-	printf("%s\n", msg);
 }
 
 void defaultFlush()