void CDBThread::delImsiMsisdnByMsisdn(stMSISDN* pMsisdn)
{
    stImsiMsisdn numdata;
    memset(&numdata, 0, sizeof(stImsiMsisdn));
    //delete the old one from db by MSISDN
    numdata.MsisdnLen = pMsisdn->ucLen;
    memcpy(numdata.strMsisdn, pMsisdn->ucMSISDN, pMsisdn->ucLen);
    #ifdef __SYB_ESQL__
    int ret = ::delMsisdnImsiByMsisdn(&numdata, 0);
    if(ret < 0)
    {
        char buf[512] = {0};
        if(ERROR_WRITE(-ret))
        {
            LOG4CXX_WARN(dbLogger, "The DB connection has broken, reconnect ...");
            connect2db();
        }
        else
        {
            LOG4CXX_WARN(dbLogger, "DELETE MSISDN["<<pMsisdn->ucMSISDN<<"]<=>IMSI by MSISDN from DB failed: ", ::getLastError());
        }
    }
    #endif
    #ifdef __SYB_CTLIB__
    if(m_pSybClient->delMsisdnImsiByMsisdn(m_iDbConnId, "SapDB.dbo.TBL_ImsiMsisdn", &numdata) < 0)
    {
        char buf[512] = {0};
        snprintf(buf, sizeof(buf) - 1, "DELETE MSISDN[%s]<=>IMSI by MSISDN from DB failed: ", pMsisdn->ucMSISDN);
        if(checkSybCtlibError(buf) < 0)
        {
            _connect2db();
        }
    }
    #endif
}

void CDBThread::insertImsiMsisdn(stImsiMsisdn* pNum)
{
    #ifdef __SYB_ESQL__
    int ret = ::writeMsisdnImsi(pNum, 0);
    if(ret < 0)
    {
        if(ERROR_WRITE(-ret))
        {
            LOG4CXX_WARN(dbLogger, "The DB connection has broken, reconnect ...");
            connect2db();
        }
        else
        {
            LOG4CXX_WARN(dbLogger, "INSERT MSISDN["<<pNum->strMsisdn<<"]<=>IMSI["<<pNum->strImsi<<"] to DB failed: "<<::getLastError());
        }
    }
    #endif
    #ifdef __SYB_CTLIB__
    if(m_pSybClient->write_ImsiMsisdn(m_iDbConnId, "SapDB.dbo.TBL_ImsiMsisdn", pNum) < 0)
    {
        char buf[512] = {0};
        snprintf(buf, sizeof(buf) - 1, "INSERT MSISDN[%s]<=>IMSI[%s] to DB failed: ", pNum->strMsisdn, pNum->strImsi);
        if(checkSybCtlibError(buf) < 0)
        {
            _connect2db();
        }
    }
    #endif
}

void QueryService::AddService()
{
	std::list<std::string>::iterator iter = file_list_.begin();
	void* handler;
	QueryInfo* query;
	QueryObj result;

	for(;iter!=file_list_.end();++iter)
	{
		std::string temp_path = *iter;
		LOG4CXX_INFO(log_,temp_path);
		handler = dlopen(temp_path.c_str(),RTLD_LAZY);

		if(!handler)
		{
			//load failed
			LOG4CXX_WARN(log_,"Can not load so: "+*iter);
			continue;
		}
		dlerror();

		query = (QueryInfo *)dlsym(handler,g_query_name.c_str());
		const char* dlsym_error = dlerror();
		if(dlsym_error)
		{
			dlclose(handler);
			LOG4CXX_WARN(log_,"Can not find QueryInfo in "+*iter);
			continue;
		}
		
		result = query->handler();
		node_.service()->addService(result.name,result.object,false);
		LOG4CXX_INFO(log_,"add service "+result.name);
	}
}

void IRCNetworkPlugin::handleJoinRoomRequest(const std::string &user, const std::string &room, const std::string &nickname, const std::string &password) {
	std::string session = getSessionName(user, room);
	std::string target = getTargetName(room);

	LOG4CXX_INFO(logger, user << ": Session name: " << session << ", Joining room " << target);
	if (m_sessions[session] == NULL) {
		if (m_servers.empty()) {
			// in gateway mode we want to login this user to network according to legacyName
			if (room.find("@") != std::string::npos) {
				// suffix is %irc.freenode.net to let MyIrcSession return #room%irc.freenode.net
				m_sessions[session] = createSession(user, room.substr(room.find("@") + 1), nickname, "", room.substr(room.find("@")));
			}
			else {
				LOG4CXX_WARN(logger, user << ": There's no proper server defined in room to which this user wants to join: " << room);
				return;
			}
		}
		else {
			LOG4CXX_WARN(logger, user << ": Join room requested for unconnected user");
			return;
		}
	}

	m_sessions[session]->addAutoJoinChannel(target, password);
	m_sessions[session]->sendCommand(IrcCommand::createJoin(FROM_UTF8(target), FROM_UTF8(password)));
	m_sessions[session]->rooms += 1;

	// update nickname, because we have nickname per session, no nickname per room.
	handleRoomNicknameChanged(user, target, TO_UTF8(m_sessions[session]->nickName()));
}

void QueryService::GetFile()
{
	if(!boost::filesystem::exists(path_))
	{
		std::ostringstream oss;
		oss << "path do not exists!!";
		LOG4CXX_WARN(log_,oss.str());
	}
	else if(path_.empty())
	{
		std::ostringstream oss;
		oss << "path is empty!";
		LOG4CXX_WARN(log_,oss.str());
	}
	else
	{
		boost::filesystem::directory_iterator end_it;
		for(boost::filesystem::directory_iterator it(path_);
				it!=end_it;++it)
		{
			boost::filesystem::path temp(*it);
			//only load so file
			if(temp.extension() == ".so")
			{
				file_list_.push_back(temp.string());
				std::ostringstream oss;
				oss << "find " << temp.filename();
				LOG4CXX_DEBUG(log_,oss.str());
			}
		}
	}
}

bool Collector::TryToConnect(ACE_SOCK_Connector& con,ACE_SOCK_Stream& peer)
{
	if(data_source_.node_list.size() == 0)
	{
		std::ostringstream oss;
		oss << data_source_.name << " have no legal IP";
		LOG4CXX_WARN(log_,oss.str());

		return false;
	}
	
	for(int i=0;i<data_source_.node_list.size();i++)
	{
		std::string ip_temp = data_source_.node_list[i].ip;
		unsigned int port = data_source_.node_list[i].port;
		ACE_INET_Addr addr(port,ip_temp.c_str());

		if(con.connect(peer,addr)==0)
		{
			return true;
		}
	}

	std::ostringstream oss;
	oss << "can not connect every host in cluster:" << data_source_.name;
	LOG4CXX_WARN(log_,oss.str());

	return false;
}

 void MpiLauncher::handleKillTimeout(boost::shared_ptr<boost::asio::deadline_timer>& killTimer,
                                     const boost::system::error_code& error)
 {
     ScopedMutexLock lock(_mutex);

     if (error == boost::asio::error::operation_aborted) {
         assert(_pid < 0);
         LOG4CXX_TRACE(logger, " MPI launcher kill timer cancelled");
         return;
     }
     if (error) {
         assert(false);
         LOG4CXX_WARN(logger, "MPI launcher kill timer encountered error"<<error);
     }
     if (_pid <= 0) {
         LOG4CXX_WARN(logger, "MPI launcher kill timer cannot kill pid="<<_pid);
         return;
     }
     if (!_waiting) {
         assert(false);
         LOG4CXX_ERROR(logger, "MPI launcher kill timer cannot kill pid="<<_pid);
         throw InvalidStateException(REL_FILE, __FUNCTION__, __LINE__)
             << " MPI launcher process cannot be killed";
     }
     LOG4CXX_WARN(logger, "MPI launcher is about to kill group pid="<<_pid);

     // kill launcher's proc group
     MpiErrorHandler::killProc(_installPath, -_pid);
 }

void CDBThread::connect2db()
{
    #ifdef __SYB_ESQL__
    int ret = -1;
    while((ret = ::connectDb(sysPtr->getSysCfg().szDbUser, sysPtr->getSysCfg().szDbPassword, sysPtr->getSysCfg().szDbName)) < 0 && !EXIT())
    {
		LOG4CXX_WARN(dbLogger, "Connect to DB failed: "<<::getLastError()<<", retry after 3s ...");
        ::disconnectDb(0);
        Delay2(3, 0);
    }
    if(ret >= 0) LOG4CXX_DEBUG(dbLogger, "Connect to db OK");
    #endif
    #ifdef __SYB_CTLIB__
    while(m_pSybClient->open() < 0 && !EXIT())
    {
        LOG4CXX_WARN(dbLogger, "SYBASE CTLIB initialize failed, retry after 3s ...");
        Delay2(3, 0);
    }
    while((m_iDbConnId = m_pSybClient->connect(sysPtr->getSysCfg().szDbName, sysPtr->getSysCfg().szDbUser, sysPtr->getSysCfg().szDbPassword)) < 0 && !EXIT())
    {
        LOG4CXX_WARN(dbLogger, "Connect to db failed, retry after 3s ...");
        Delay2(3, 0);
    }
    if(m_iDbConnId >= 0) LOG4CXX_DEBUG(dbLogger, "Connect to db OK");
    #endif
}

	bool doMsg(std::string& reply, const std::string& msg)
	{
		Cmd::Set setmsg;
		if (setmsg.ParseFromString(msg))
		{
			const std::string& key = setmsg.key();
			const std::string& value = setmsg.value();

			LOG4CXX_INFO(logger, "<" << pthread_self() << "> set : " << setmsg.key() << ", " << setmsg.value().size());


			RedisConnectionByKey redis(key, worker->cachePool());
			if (redis)
			{
				redis->hmset("%s value %b flag %u touchtime %u", key.c_str(),
						value.data(), value.size(), 0, (unsigned int)time(0));
			}
			else
			{
				LOG4CXX_WARN(logger, "set " << key << " .. redis is not ready");
			}

			if (0 == s_cacheonly)
			{
				MySqlConnectionByKey mysql(key, worker->dbPool());
				if (mysql)
				{
					try {
						mysqlpp::Query query = mysql->query();
						query.reset();
						query << "replace into " 
						      << getKVSTableName(key) 
						      << "(`key`, value, flag, touchtime) values(" 
							  << mysqlpp::quote << key << ","
							  << mysqlpp::quote << value << ","
							  << 0 << ","
							  << time(0) << ")";
						query.execute();

						LOG4CXX_INFO(logger, "mysql " << key << "," << mysql->shard() << "," << getKVSTableName(key));
					}
					catch (std::exception& err)
					{
						LOG4CXX_WARN(logger, "set " << key << " .. mysql error:" << err.what());
						return false;
					}
				}
				else
				{
					LOG4CXX_WARN(logger, "set " << key << " .. mysql is not ready");
				}
			}
		}

		
		reply = "OK";
		return true;
	}

bool RosterResponder::handleSetRequest(const Swift::JID& from, const Swift::JID& to, const std::string& id, boost::shared_ptr<Swift::RosterPayload> payload) {
	sendResponse(from, id, boost::shared_ptr<RosterPayload>(new RosterPayload()));

	User *user = m_userManager->getUser(from.toBare().toString());
	if (!user) {
		LOG4CXX_WARN(logger, from.toBare().toString() << ": User is not logged in");
		return true;
	}

	if (payload->getItems().size() == 0) {
		LOG4CXX_WARN(logger, from.toBare().toString() << ": Roster push with no item");
		return true;
	}

	Swift::RosterItemPayload item = payload->getItems()[0];

	if (item.getJID().getNode().empty()) {
		return true;
	}

	Buddy *buddy = user->getRosterManager()->getBuddy(Buddy::JIDToLegacyName(item.getJID()));
	if (buddy) {
		if (item.getSubscription() == Swift::RosterItemPayload::Remove) {
			LOG4CXX_INFO(logger, from.toBare().toString() << ": Removing buddy " << buddy->getName());
			onBuddyRemoved(buddy);

			// send roster push here
			Swift::SetRosterRequest::ref request = Swift::SetRosterRequest::create(payload, user->getJID().toBare(), user->getComponent()->getIQRouter());
			request->send();
		}
		else {
			LOG4CXX_INFO(logger, from.toBare().toString() << ": Updating buddy " << buddy->getName());
			onBuddyUpdated(buddy, item);
		}
	}
	else if (item.getSubscription() != Swift::RosterItemPayload::Remove) {
		// Roster push for this new buddy is sent by RosterManager
		BuddyInfo buddyInfo;
		buddyInfo.id = -1;
		buddyInfo.alias = item.getName();
		buddyInfo.legacyName = Buddy::JIDToLegacyName(item.getJID());
		buddyInfo.subscription = "both";
		buddyInfo.flags = Buddy::buddyFlagsFromJID(item.getJID());
		LOG4CXX_INFO(logger, from.toBare().toString() << ": Adding buddy " << buddyInfo.legacyName);

		buddy = user->getComponent()->getFactory()->createBuddy(user->getRosterManager(), buddyInfo);
		user->getRosterManager()->setBuddy(buddy);
		onBuddyAdded(buddy, item);
	}

	return true;
}

                        void LoggerUtil::warning(HLoggerPtr logger, const char* msg){
#ifdef HAVE_LOG4CXX
                            LOG4CXX_WARN(static_cast<Logger*>(logger), msg);
#else
                            stdLogOut(logger, msg);
#endif
                        }

void ConversationManager::handleMessageReceived(Swift::Message::ref message) {
// 	std::string name = message->getTo().getUnescapedNode();
// 	if (name.find_last_of("%") != std::string::npos) { // OK when commented
// 		name.replace(name.find_last_of("%"), 1, "@"); // OK when commented
// 	}
	std::string name = Buddy::JIDToLegacyName(message->getTo());
	if (name.empty()) {
		LOG4CXX_WARN(logger, m_user->getJID().toString() << ": Tried to create empty conversation");
		return;
	}

	// create conversation if it does not exist.
	if (!m_convs[name]) {
		Conversation *conv = m_component->getFactory()->createConversation(this, name);
		addConversation(conv);
	}
	// if it exists and it's MUC, but this message is PM, get PM conversation or create new one.
	else if (m_convs[name]->isMUC() && message->getType() != Swift::Message::Groupchat) {
		std::string room_name = name;
		name = room_name + "/" + message->getTo().getResource();
		if (m_convs.find(name) == m_convs.end()) {
			Conversation *conv = m_component->getFactory()->createConversation(this, message->getTo().getResource());
			conv->setRoom(room_name);
			conv->setNickname(name);
			addConversation(conv);
		}
	}

	// update resource and send the message
	m_convs[name]->setJID(message->getFrom());
	m_convs[name]->sendMessage(message);
}

static bool
SetVorbisCommentDouble(FLAC__StreamMetadata *block, const char *key, CFNumberRef value, CFStringRef format = NULL)
{
	assert(NULL != block);
	assert(NULL != key);
	
	CFStringRef numberString = NULL;
	
	if(NULL != value) {
		double f;
		if(!CFNumberGetValue(value, kCFNumberDoubleType, &f)) {
			log4cxx::LoggerPtr logger = log4cxx::Logger::getLogger("org.sbooth.AudioEngine.AudioMetadata.FLAC");
			LOG4CXX_WARN(logger, "CFNumberGetValue() failed");
			return false;
		}

		numberString = CFStringCreateWithFormat(kCFAllocatorDefault, NULL, NULL == format ? CFSTR("%f") : format, f);
	}
	
	bool result = SetVorbisComment(block, key, numberString);
	
	if(numberString)
		CFRelease(numberString), numberString = NULL;
	
	return result;
}

void ThriftAsyncClientPool::DiscoverServicesDelegate::process(CallbackResponse response,
        const ::std::vector< ::std::string>& services) {
    if ((response != ServiceDiscoveryCallback::OK) || (services.empty())) {
        LOG4CXX_WARN(_poolRef.LOG, "Failed to get application services."
                "  This might be okay if no application services have been registered");
    } else if (services.empty()) {
        //done discovering all services for this iteration. Invoke next callback
        _nextCb->operator()();
        return;
    }

    ::boost::optional< ::std::string> appName_opt;

    if ((_nextCb != _ownerDispatch) &&
        (_poolRef._applicationName.find_first_not_of(' ') != ::std::string::npos)) {
        //not looking for common serivces. If application name has been set,
        //set the optional agrument for the endpoint discovery delegate
        appName_opt = _poolRef._applicationName;
    }

    ::boost::shared_ptr<DiscoverEndpointsDelegate> delegate =
            ::boost::make_shared<DiscoverEndpointsDelegate>(_poolRef, ::boost::bind(*_nextCb),
                    appName_opt, services);

    //save a boost shared ptr to this instance inorder not to prematurely
    //destory the instance before the endpoint delegate completes
    delegate->_servicesDelegate = shared_from_this();

    //use a shared_ptr to invoke the discoverEndpoint method as the discoverEndpoint method uses the
    //shared_ptr to hold a reference to the delegate object event after this current function
    //exits.
    delegate->discoverEndpoints();
}

 rgb_classification_dataset::rgb_classification_dataset(const std::string& filename, int pattern_size, int n_crops)
     :m_n_crops(n_crops)
     ,m_pattern_size(pattern_size)
     ,m_filename(filename){
         std::ifstream ifs(filename.c_str());
         unsigned int n_cls;
         ifs >> n_cls;
         ifs.get(); // consume '\n'
         for(unsigned int klass = 0; klass < n_cls; klass++){
             std::string line;
             std::getline(ifs, line);
             if(ifs)
                 m_class_names.push_back(line);
         }
         cuvAssert(m_class_names.size() == n_cls);
         while(ifs){
             meta_t m;
             ifs >> m.rgb_filename;
             //ifs >> m.klass; // dummy value for historical reasons
             ifs >> m.klass;
             if(ifs)
                 m_meta.push_back(m);
         }
         cuvAssert(m_meta.size() > 0);
         LOG4CXX_WARN(g_log, "read `"<< filename<<"', n_classes: "<<n_cls<<", size: "<<m_meta.size());
         m_predictions.resize(this->size());
         shuffle(false);
     }

void Collector::PollData()
{
	ACE_SOCK_Stream peer;
	ACE_SOCK_Connector connector;
	try
	{
		if(TryToConnect(connector,peer))
		{
			std::ostringstream oss;
			oss << "poll data form cluster: " << data_source_.name;
			LOG4CXX_INFO(log_,oss.str());

			char buf[T_SIZE] = "request";
			if((peer.send(buf,sizeof(buf)))<=0)
				throw "KSN POLLER COLLECTOR SEND REQUEST ERROR!!";
			char* ch = new char[T_SIZE]();
			if((peer.recv(ch,T_SIZE)<=0))
				throw "KSN POLLER COLLECTOR RECV ERROR!!";

			msg_ = new ACE_Message_Block(ch,sizeof(*ch));
			sender_->putq(msg_);
		}
		else
		{
			std::ostringstream oss;
			oss << "Can not poll data form " + data_source_.name;
			LOG4CXX_WARN(log_,oss.str());
		}
	}
	catch(const char* ch)
	{
		LOG4CXX_ERROR(log_,ch);
	}
}

bool TapeMsg::IsValid() {
	if(m_recId.length() == 0) {
		LOG4CXX_WARN(LOG.messaging,"Ignoring tape message with empty recId");
		return false;
	}
	return true;
}

int CGsmASythesize::sendPowerOff(SPowerOff& data)
{
    //@zh 判断是否是本地号码，否则不发送
    //stNumPrefix* prefix = findNumPrefix((unsigned char*)(&data.msisdn[0]));
    stNumPrefix* prefix = sysPtr->findNumPrefix((unsigned char*)(&data.msisdn[0]));
    if(!prefix) return 0;
    //

    WEvent evt;
    evt.PtlId = PTL_SAP;
    evt.PtlVer = 0; // 不填亦可，程序不做判断。
    evt.dmid = LMI_SCP << 4;
    evt.what = evControl;
    evt.evCode = SYT_SCP_MOFF;
    evt.msgLen = sizeof(SPowerOff);
    evt.msgPtr = NULL;
    evt.PktLength = sizeof(WEvent) + evt.msgLen;

    //
    evt.msgPtr = (char*)&data;
    int ret = ipcPtr->SendMsg(evt);
    if(ret < 0)
    {
        LOG4CXX_WARN(rootlog, "Send POWER ON message to SCP failed!");
    }
#ifdef SYTHESIZE_DEBUG
    else
    {
        printf("$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$\n"
               "Send power off to scp:\nMSISDN: %s\n"
               "$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$\n", data.msisdn);
    }
#endif
    return ret;
}

void CoreBoundQueuesScheduler::pushToQueue(std::shared_ptr<Task> task)
{
  // check if task should be scheduled on specific core
  int core = task->getPreferredCore();

  if (core >= 0 && core < static_cast<int>(this->_queues)) {
    //potentially assigns task to a queue blocked by long running task
    this->_taskQueues[core]->push(task);
    LOG4CXX_DEBUG(this->_logger,  "Task " << std::hex << (void *)task.get() << std::dec << " pushed to queue " << core);
  } else if (core == Task::NO_PREFERRED_CORE || core >= static_cast<int>(this->_queues)) {

    if (core < Task::NO_PREFERRED_CORE || core >= static_cast<int>(this->_queues))
      // Tried to assign task to core which is not assigned to scheduler; assigned to other core, log warning
      LOG4CXX_WARN(this->_logger, "Tried to assign task " << std::hex << (void *)task.get() << std::dec << " to core " << std::to_string(core) << " which is not assigned to scheduler; assigned it to next available core");

    // lock queuesMutex to sync pushing to queue and incrementing next queue
    {
      std::lock_guard<lock_t> lk2(this->_queuesMutex);
      // simple strategy to avoid blocking of queues; check if queue is blocked - try a couple of times, otherwise schedule on next queue
      size_t retries = 0;
      while (static_cast<CoreBoundQueue *>(this->_taskQueues[this->_nextQueue])->blocked() && retries < 100) {
        this->_nextQueue = (this->_nextQueue + 1) % this->_queues;
        ++retries;
      }

      this->_taskQueues[this->_nextQueue]->push(task);
      //std::cout << "Task " <<  task->vname() << "; hex " << std::hex << &task << std::dec << " pushed to queue " << this->_nextQueue << std::endl;
      //round robin on cores
      this->_nextQueue = (this->_nextQueue + 1) % this->_queues;
    }
  }
}

int SYSTEM::sendCDR     (SCallDetailRecord& data)
{
    WEvent evt;
    evt.PtlId = PTL_SAP;
    evt.PtlVer = 0; // 不填亦可，程序不做判断。
    evt.dmid = LMI_SCP << 4;
    evt.what = evControl;
    evt.evCode = SYT_SCP_CDR;
    evt.msgLen = sizeof(SCallDetailRecord);
    evt.msgPtr = NULL;
    evt.PktLength = sizeof(WEvent) + evt.msgLen;

    //
    evt.msgPtr = (char*)&data;
    int ret = ipcPtr->SendMsg(evt);
    if(ret < 0)
    {
        LOG4CXX_WARN(rootlog, "Send CDR message to SCP failed!");
    }
#ifdef SYTHESIZE_DEBUG
    else
    {
        printf("$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$\n"
               "Send cdr to scp:\ncaller MSISDN: %s\ncallee MSISDN: %s\n"
               "$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$\n", data.caller, data.called);
    }
#endif
    return ret;
}