boost之无锁队列使用实例

#ifndef	SEARCH_ENGINE_H
#define SEARCH_ENGINE_H

#include "boost_comm.h"
#include "message_header.h"

namespace tspace
{
	class search_engine
	{
	public:
		search_engine();
		~search_engine();
	public:
		static search_engine *instance()
		{
			static search_engine ins_;
			return &ins_;
		}
	public:

		static void print()
		{
			cout car_id_, p->json_basic_, (char *)p->buffer_ctx_, p->buffer_size_);
				file_system::fstore_file(p->car_id_, p->json_basic_, (char *)ptr_bin->bin_value_, ptr_bin->bin_value_len_);
				buffer_pool::bin_value_free(p->buffer_ctx_);
				delete ptr_bin;
				delete p;
			}
			else
			{
				std::ostringstream oslog;
				oslog container_bin_.front();
				search_engine::instance()->map_lock_.lock();
				if(search_engine::instance()->m_curr_task_list.find(p->task_id_)==search_engine::instance()->m_curr_task_list.end())
				{
					auto itr = search_engine::instance()->m_curr_task_list.insert(make_pair(p->task_id_, p));
					search_engine::instance()->map_lock_.unlock();
					file_system::search_file(p->task_id_, p->json_condition_, p->json_detail_, 
						(char *)ptr_bin->bin_value_, ptr_bin->bin_value_len_);
					search_engine::instance()->map_lock_.lock();
					if(search_engine::instance()->m_curr_task_list.find(p->task_id_) !=search_engine::instance()->m_curr_task_list.end())
						search_engine::instance()->m_curr_task_list.erase(itr.first);
					search_engine::instance()->map_lock_.unlock();
				}
				else 
				{
					std::ostringstream oslog;
					oslog second->task_state_ = task_sigal_interrupt;
					search_engine::instance()->m_curr_task_list.erase(itr);
				}
				search_engine::instance()->map_lock_.unlock();

				delete p;
			}
		}
		catch (...)
		{
			std::ostringstream oslog;
			oslog task_id_, str_json, 100);
				if (p->user_ctx_)
				{
					unsigned char *buff = buffer_pool::bin_value_alloc();
					int len = BUFFER_POOL_BIN_CHUNK_SIZE;
					bool ret = protocol_functions::construct_response((char *)buff, len, str_json.c_str());
					(*((tcp_client *)p->user_ctx_))((char *)buff, len);
					buffer_pool::bin_value_free(buff);
				}
				delete p;
			}
		}
		catch (...)
		{
			std::ostringstream oslog;
			oslog mtx1_);
		bool ret = instance()->add_feature_.push(src);
		if (ret)
		{
			instance()->m_add_feature_counts_++;
			thread_pool::post_task(add_feature_run);
			if (instance()->m_total_feature_counts_++ % 400 == 1)
			{
				std::ostringstream oslog;
				oslog mtx2_);
		bool ret = instance()->start_task_.push(src);
		if (ret)
		{
			instance()->m_start_task_counts_++;
			thread_pool::post_task(start_task_run);
		}
		else
		{
			std::ostringstream oslog;
			oslog stop_task_.push(src);
		if (ret)
		{
			instance()->m_stop_task_counts_++;
			thread_pool::post_task(stop_task_run);
		}
		else
		{
			std::ostringstream oslog;
			oslog mtx2_);
		bool ret = instance()->start_task_.pop(ctx);
		if (ret)
		{
			instance()->m_start_task_counts_--;
		}
		return ret;
	}
	bool search_engine::mt_pop_stop_task(ptr_stop_task & ctx)
	{
		boost::unique_lock wlock(instance()->mtx4_);
		bool ret = instance()->stop_task_.pop(ctx);
		if (ret)
		{
			instance()->m_stop_task_counts_++;
		}
		return ret;
	}
	bool search_engine::mt_push_get_result(get_result *src)
	{
		boost::unique_lock wlock(instance()->mtx3_);
		bool ret = instance()->get_result_.push(src);
		if (ret)
		{
			instance()->m_get_result_counts_++;
			thread_pool::post_task(get_result_run);
		}
		else
		{
			std::ostringstream oslog;
			oslog mtx3_);
		bool ret = instance()->get_result_.pop(ctx);
		if (ret)
		{
			instance()->m_get_result_counts_++;
		}
		return ret;
	}

	int search_engine::mt_push_task_result(const string taskid, result_data *data, int max/*=100*/)
	{
		bool ret = false;
		try
		{
			boost::unique_lock wlock(instance()->mtx_);
			//LOG(INFO)ret_result_.size() < 100)
				{
					itr->second->ret_result_.insert(data);
					//LOG(INF) second->ret_result_.insert(data);
						object_pool::free_data(b_data);
						itr->second->ret_result_.erase(itr->second->ret_result_.begin());
					}
					else
					{
						object_pool::free_data(data);
					}
				}
			}
			else
			{
				ret_result *ptr_res = object_pool::alloc_result("0", "success");
				ptr_res->ret_result_.insert(data);
				ret = instance()->taskid_result_.insert(make_pair(taskid, ptr_res)).second;
			}
		}
		catch (...)
		{
			std::ostringstream oslog;
			oslog mtx_);
		taskid_result_itr itr_result = instance()->taskid_result_.find(taskid);
		std::string strSql = "";
		std::ostringstream ostrstream;
		if (itr_result != instance()->taskid_result_.end())
		{
			//char db_buff[1024];
			ostrstreamret_result_.begin();
				itr != itr_result->second->ret_result_.end(); )
			{
				string carPlateNumber = (*itr)->carPlateNumber_;
				//char outbuf[1024];
			//	int outlen = 1024;
#ifdef _WIN32
#else
				//extern int code_convert(char *from_charset, char *to_charset, char *inbuf, int inlen, char *outbuf, int outlen);
				//code_convert("gb2312", "utf-8", (char *)carPlateNumber.c_str(), carPlateNumber.size(), outbuf, outlen);
#endif
				//taskId,carId,matchingScore,snapShotTime,carPlateNumber,searchType
				if(count++>0)
					ostrstream
			str_json = "{\"returnCode\":2,\"msg\":\"can't find taskid\",\"data\":[]}";
		}
		return 0;
	}
#endif
}