CppCMS/tcp_cache.cpp

#include "config.h"
#ifdef __CYGWIN__
// Cygwin ASIO works only with win32 sockets
#define _WIN32_WINNT 1
#define __USE_W32_SOCKETS 1
#endif

#ifdef USE_BOOST_ASIO
#include <boost/asio.hpp>
namespace aio = boost::asio;
using boost::system::error_code;
using boost::system::system_error;
#else
#include <asio.hpp>
namespace aio = asio;
using asio::error_code;
using asio::system_error;
#endif

#include "tcp_cache.h"
#include "tcp_cache_protocol.h"
#include "archive.h"

using aio::ip::tcp;

namespace cppcms {

class messenger : boost::noncopyable {
	aio::io_service srv_;
	tcp::socket socket_;
	string ip_;
	int port_;
public:
	void connect(string ip,int port) {
		ip_=ip;
		port_=port;
		error_code e;
		socket_.connect(tcp::endpoint(aio::ip::address::from_string(ip),port),e);
		if(e) throw cppcms_error("connect:"+e.message());
		tcp::no_delay nd(true);
		socket_.set_option(nd);
	}
	messenger(string ip,int port) :
		socket_(srv_)
	{
		connect(ip,port);
	}
	messenger() : socket_(srv_) { };

	void transmit(tcp_operation_header &h,string &data)
	{
		bool done=false;
		int times=0;
		do {
			try {
				aio::write(socket_,aio::buffer(&h,sizeof(h)));
				if(h.size>0) {
					aio::write(socket_,aio::buffer(data,h.size));
				}
				aio::read(socket_,aio::buffer(&h,sizeof(h)));
				if(h.size>0) {
					vector<char> d(h.size);
					aio::read(socket_,aio::buffer(d,h.size));
					data.assign(d.begin(),d.begin()+h.size);
				}
				done=true;
			}
			catch(system_error const &e) {
				if(times) {
					throw cppcms_error(string("tcp_cache:")+e.what());
				}
				socket_.close();
				error_code er;
				socket_.connect(
					tcp::endpoint(
						aio::ip::address::from_string(ip_),port_),er);
				if(er) throw cppcms_error("reconnect:"+er.message());
				times++;
			}
		}while(!done);
	}
	
};

tcp_cache::tcp_cache(vector<string> const& ip,vector<long> const &port)
{
	if(ip.size()<1 || port.size()!=ip.size()) {
		throw cppcms_error("Incorrect parameters for tcp cache");
	}
	conns=ip.size();
	tcp=new messenger[conns];
	try {
		for(int i=0;i<conns;i++) {
			tcp[i].connect(ip[i],port[i]);
		}
	}
	catch(...) {
		delete [] tcp;
		tcp=NULL;
		throw;
	}
}

tcp_cache::~tcp_cache()
{
	delete [] tcp;
}

void tcp_cache::broadcast(tcp_operation_header &h,string &data)
{
	int i;
	for(i=0;i<conns;i++) {
		tcp_operation_header ht=h;
		string dt=data;
		tcp[i].transmit(ht,data);
	}
}

void tcp_cache::rise(string const &trigger)
{
	tcp_operation_header h={0};
	h.opcode=opcodes::rise;
	h.size=trigger.size();
	string data=trigger;
	h.operations.rise.trigger_len=trigger.size();
	broadcast(h,data);
}

void tcp_cache::clear()
{
	tcp_operation_header h={0};
	h.opcode=opcodes::clear;
	h.size=0;
	string empty;
	broadcast(h,empty);
}

bool tcp_cache::fetch_page(string const  &key,string &output,bool gzip)
{
	string data=key;
	tcp_operation_header h={0};
	h.opcode=opcodes::fetch_page;
	h.size=data.size();
	h.operations.fetch_page.gzip=gzip;
	h.operations.fetch_page.strlen=data.size();
	get(key).transmit(h,data);
	if(h.opcode==opcodes::page_data) {
		output=data;
		return true;
	}
	return false;
}

bool tcp_cache::fetch(string const &key,archive &a,set<string> &tags)
{
	string data=key;
	tcp_operation_header h={0};
	h.opcode=opcodes::fetch;
	h.size=data.size();
	h.operations.fetch.key_len=data.size();
	get(key).transmit(h,data);
	if(h.opcode!=opcodes::data)
		return false;
	char const *ptr=data.c_str();
	a.set(ptr,h.operations.data.data_len);
	ptr+=h.operations.data.data_len;
	int len=h.operations.data.triggers_len;
	while(len>0) {
		string tag;
		unsigned tmp_len=strlen(ptr);
		tag.assign(ptr,tmp_len);
		ptr+=tmp_len+1;
		len-=tmp_len+1;
		tags.insert(tag);
	}
	return true;
}

void tcp_cache::stats(unsigned &keys,unsigned &triggers)
{
	keys=0; triggers=0;
	for(int i=0;i<conns;i++) {
		tcp_operation_header h={0};
		string data;
		h.opcode=opcodes::stats;
		tcp[i].transmit(h,data);
		if(h.opcode==opcodes::out_stats) {
			keys+=h.operations.out_stats.keys;
			triggers+=h.operations.out_stats.triggers;
		}
	}
}

void tcp_cache::store(string const &key,set<string> const &triggers,time_t timeout,archive const &a)
{
	tcp_operation_header h={0};
	string data;
	h.opcode=opcodes::store;
	data.append(key);
	h.operations.store.key_len=key.size();
	data.append(a.get());
	h.operations.store.data_len=a.get().size();
	time_t now;
	time(&now);
	h.operations.store.timeout=timeout-now > 0 ? timeout-now : 0;
	unsigned tlen=0;
	for(set<string>::const_iterator p=triggers.begin(),e=triggers.end();p!=e;++p) {
		tlen+=p->size()+1;
		data.append(p->c_str(),p->size()+1);
	}
	h.operations.store.triggers_len=tlen;
	h.size=data.size();
	get(key).transmit(h,data);
}

messenger &tcp_cache::get(string const &key)
{
	if(conns==1) return tcp[0];
	unsigned val=0,i;
	for(i=0;i<key.size();i++) {
		val+=251*key[i]+103 % 307;
	}
	return tcp[val % conns];
}


}


#ifdef TCP_CACHE_UNIT_TEST

#include <assert.h>
#include <iostream>
#include <cstdlib>
int main(int argc,char **argv)
{
	using namespace cppcms;
	using namespace std;
	if(argc!=3) {
		cerr<<"Usage IP port"<<endl;
		return 1;
	}
	try {
		archive a;
		set<string> s;
		tcp_cache tcp(argv[1],atoi(argv[2]));
		assert(tcp.fetch("something",a,s)==false);
		time_t t;
		time(&t);
		t+=2;
		a.set("data",4);
		tcp.store("key",s,t,a);
		unsigned keys,triggers;
		tcp.stats(keys,triggers);
		assert(keys==1);
		assert(triggers==1);
		s.clear();
		a.set("");
		assert(tcp.fetch("key",a,s)==true);
		assert(s.size()==1);
		assert(*(s.begin())=="key");
		assert(a.get()=="data");
		sleep(3);
		assert(tcp.fetch("key",a,s)==false);
		a.set("");
		a<<string("msg1");
		a<<string("msg2");
		time(&t);
		t+=50;
		s.clear();
		s.insert("a");
		s.insert("b");
		tcp.store("k",s,t,a);
		string x;
		assert(tcp.fetch_page("k",x,true)==true);
		assert(x=="msg2");
		assert(tcp.fetch_page("k",x,false)==true);
		assert(x=="msg1");
		a.set("");
		s.clear();
		assert(tcp.fetch("k",a,s)==true);
		assert(s.size()==3);
		set<string>::iterator ptr=s.begin();
		assert(*ptr++=="a");
		assert(*ptr++=="b");
		assert(*ptr++=="k");
		tcp.rise("a");
		assert(tcp.fetch("k",a,s)==false);
		a.set("Something");
		s.clear();
		tcp.store("bb",s,t,a);
		assert(tcp.fetch("xx",a,s)==false);
		assert(tcp.fetch("bb",a,s)==true);
		tcp.clear();
		assert(tcp.fetch("bb",a,s)==false);
		cout<<"Done... OK!\n";
	}
	catch(std::exception const &e) {
		cerr<<e.what()<<endl;
	}
}

#endif