void
ReplicationOptimizer::notifySendFailure(DataObjectRef dObj, NodeRef node)
{
    if (utilFunction) {
        utilFunction->notifySendFailure(dObj, node);
    }

    string dobj_id = DataObject::idString(dObj);
    string node_id = Node::idString(node);
    string key = getKey(dobj_id, node_id);

    meta_map_t::iterator it = meta.find(key);
    if (it == meta.end()) {
        errorCount++;
        HAGGLE_ERR("Missing entry.\n");
        return;
    }

    ReplicationDataObjectUtilityMetadataRef do_info = (*it).second;
    if (!do_info) {
        errorCount++;
        HAGGLE_ERR("Missing metadata.\n");
        return;
    }

    Timeval now = Timeval::now();
    double delta = now.getTimeAsMilliSecondsDouble() - do_info->getComputedTime().getTimeAsMilliSecondsDouble();

    if (delta > computePeriodMs && utilFunction) {
        double new_utility = utilFunction->compute(do_info->getId(), do_info->getNodeId());
        do_info->setUtility(new_utility, now);
    }
}

long Timeval::delta(const Timeval& other) const
{
	// 2^31 milliseconds is just over 4 years.
	int32_t deltaS = other.sec() - sec();
	int32_t deltaUs = other.usec() - usec();
	return 1000*deltaS + deltaUs/1000;
}

int32_t Clock::FN() const
{
	ScopedLock lock(mLock);
	Timeval now;
	int32_t deltaSec = now.sec() - mBaseTime.sec();
	int32_t deltaUSec = now.usec() - mBaseTime.usec();
	int64_t elapsedUSec = 1000000LL*deltaSec + deltaUSec;
	int64_t elapsedFrames = elapsedUSec / gFrameMicroseconds;
	int32_t currentFN = (mBaseFN + elapsedFrames) % gHyperframe;
	return currentFN;
}

int		Timeval::cmp(Timeval const &toCmp)
{
  if (this->_tv.tv_sec > toCmp.getSec())
    return (1);
  if (this->_tv.tv_sec < toCmp.getSec())
    return (-1);
  if (this->_tv.tv_usec > toCmp.getUSec())
    return (1);
  if (this->_tv.tv_usec < toCmp.getUSec())
    return (-1);
  return (0);
}

DelayInterval operator-(const Timeval& arg1, const Timeval& arg2) {
  time_base_seconds secs = arg1.seconds() - arg2.seconds();
  time_base_seconds usecs = arg1.useconds() - arg2.useconds();

  if ((int)usecs < 0) {
    usecs += MILLION;
    --secs;
  }
  if ((int)secs < 0)
    return DELAY_ZERO;
  else
    return DelayInterval(secs, usecs);
}

void NewPager::serviceLoop()
{
	Timeval nextTime;
	while (! gBTS.btsShutdown()) {
		// Wait for pending activity to clear PCH.
		if (unsigned load = gPagingQ.getPagingLoad()) {
			LOG(DEBUG) << "Pager waiting with load " << load;
			sleepFrames(51); // There could be multiple paging channels, in which case this is longer than necessary, but no matter.
			continue;
		}
		// nextTime controls how quickly we resend pages.
		if (! nextTime.passed()) {
			sleepFrames(51);
			continue;
		}
		newPageAll();
		nextTime.future(5000); // Wait 5 seconds between paging sets.
	}
}

void
ReplicationOptimizer::computeUtilities()
{
    for (meta_map_t::iterator it = meta.begin(); it != meta.end(); it++) {
        ReplicationDataObjectUtilityMetadataRef do_info = (*it).second;
        if (!do_info) {
            errorCount++;
            HAGGLE_ERR("NULL DO in cache\n");
            continue;
        }
        Timeval now = Timeval::now();
        // was the utility recently computed?
        double delta = now.getTimeAsMilliSecondsDouble() - do_info->getComputedTime().getTimeAsMilliSecondsDouble();
        if (delta > computePeriodMs && utilFunction) {
            double new_utility = utilFunction->compute(do_info->getId(), do_info->getNodeId());
HAGGLE_DBG("do:node %s:%s has a utility value of %f\n", do_info->getId().c_str(), do_info->getNodeId().c_str());
            do_info->setUtility(new_utility, now);
        }
    }
}

bool ReportingTable::commit()
{
	ReportBatch oustanding;
	ReportBatch::iterator mp;
	unsigned oustandingCount = 0;

	// copy out to free up access to mBatch as quickly as possible
	mLock.lock();
	mp = mBatch.begin();
	while (mp != mBatch.end()) {
		if (mp->second > 0) {
			oustanding[mp->first] = mp->second;
			mBatch[mp->first] = 0;
			oustandingCount++;
		}
		mp++;
	}
	mLock.unlock();

	// now actually write them into the db if needed
	if (oustandingCount > 0) {
		Timeval timer;
		char cmd[200];

		// TODO : could wrap this in a BEGIN; COMMIT; pair to transactionize these X UPDATEs
		mp = oustanding.begin();
		while (mp != oustanding.end()) {
			sprintf(cmd,"UPDATE REPORTING SET VALUE=VALUE+%u, UPDATETIME=%ld WHERE NAME=\"%s\"", mp->second, time(NULL), mp->first.c_str());
			if (!sqlite3_command(mDB,cmd)) {
				LOG(CRIT) << "could not increment reporting parameter " << mp->first << ", error message: " << sqlite3_errmsg(mDB);
			}
			mp++;
		}

		LOG(INFO) << "wrote " << oustandingCount << " entries in " << timer.elapsed() << "ms";
	}

	return true;
}

void EvictStrategyLRFU::updateInfoDataObject(DataObjectRef &dObj, unsigned int count, Timeval time) 
{

    ScratchpadManager *pad = getKernel()->getDataStore()->getScratchpadManager();
    //create scratchpad keys
    string paramNameF0=name;
    paramNameF0.append("_LRU_F0");
    double c_now = 0.0;
    double c_old = 0.0;
    double lastTime = 0.0;
    double paramValueF0 = 0.0;
    double paramValueLastk = 0.0;
    string paramName_c_now=name;
    paramName_c_now.append("_c_now");
    string paramNameLastk=name;
    paramNameLastk.append("_LRU_last_k");
    
    if (countType == EVICT_STRAT_LRFU_COUNT_TYPE_TIME) {
	paramValueLastk = time.getTimeAsMilliSecondsDouble();
    } else {  //otherwise count
	paramValueLastk = (double) count; 
    }

    bool has_attr = pad->hasScratchpadAttributeDouble(dObj, paramNameF0);
    if (has_attr) {
        paramValueF0 = pad->getScratchpadAttributeDouble(dObj, paramNameF0);
	c_now = pad->getScratchpadAttributeDouble(dObj, paramName_c_now);
	c_old = c_now;
       	lastTime = pad->getScratchpadAttributeDouble(dObj, paramNameLastk);
        c_now *= fx_calc(paramValueLastk-lastTime);
        c_now += paramValueF0; 
       	HAGGLE_DBG("%s f(x) = %f + G(%f - %f)*%f = %f\n", dObj->getIdStr() ,paramValueF0 , (float) count, lastTime, c_old,  c_now);

    } else {
	c_now = 1.0; //fx_calc(paramValueLastk);
       	pad->setScratchpadAttributeDouble(dObj, paramNameF0, c_now);
       	HAGGLE_DBG("%s f(0) = g(%f) = %f\n", dObj->getIdStr(), paramValueLastk, c_now);
    }

   //set current values
   pad->setScratchpadAttributeDouble(dObj, paramName_c_now, c_now);
   pad->setScratchpadAttributeDouble(dObj, paramNameLastk, paramValueLastk);
   

}

/**
 * The main function to purge the cache and recompute utilities. 
 * If HAGGLE_DEBUG2 is set, this will add a new debug printout per utility computation,
 * which wills how the results of each individual utility value per data object.
 * Useful for verifying new utilities are acting correctly, not recommended for production
 * run.   To use this feature, pass any non empty string to getUtilityFunction()->compute(id, string).
 * The compute() routines will append current values and status for the data object identified by 'id'
 * only if the string being passed is non empty.   If the string is empty, the routines ignore it.
 *
 */
void
CacheStrategyUtility::_purgeCache(string doid, bool *o_was_deleted)
{
    List<DataObjectUtilityMetadata *> process_dos;
    List<DataObjectUtilityMetadata *> delete_dos;
    long long total_size = 0;
    // iterate across data objects and compute utilities for
    // DOS that have not been computed recently
    for (do_util_metadata_t::iterator it = utilMetadata.begin(); it != utilMetadata.end(); it++) {
        DataObjectUtilityMetadata *do_info = (*it).second;
        Timeval now = Timeval::now();
        if (!do_info) {
            HAGGLE_ERR("NULL DO in cache\n");
            continue;
        }
        total_size += do_info->getCost();
        // was the utility recently computed?
        double delta = now.getTimeAsMilliSecondsDouble() - do_info->getComputedTime().getTimeAsMilliSecondsDouble();
        if (delta > computePeriodMs) {
            string strResults="";
            if (Trace::trace.getTraceType() == TRACE_TYPE_DEBUG2) {
              strResults.append(do_info->getId());
              strResults.append("[T]=");
            }
            double new_utility = getUtilityFunction()->compute(do_info->getId(), strResults);
            do_info->setUtility(new_utility, now);
            HAGGLE_DBG2("%s --> %f\n", strResults.c_str(), new_utility);
        }

        // does the utility meet the threshold?
        if (do_info->getUtility() < getGlobalOptimizer()->getMinimumThreshold()) {
            delete_dos.push_front(do_info);
        }
        else {
            process_dos.push_front(do_info);
        }
    }

    if (total_size != current_size) {
        HAGGLE_ERR("Fatal error: count mismatch of cache sizes! (%lld != %lld)\n", total_size, current_size);
        return;
    }

    if (total_size > (watermark_capacity_kb*1024)) {
        CacheKnapsackOptimizerResults results = getKnapsackOptimizer()->solve(&process_dos, &delete_dos, watermark_capacity_kb*1024 );
        if (results.getHadError()) {
            HAGGLE_ERR("Knapsack optimizer failed\n");
        }
    }
    else {
        //HAGGLE_DBG("High water mark short-circuit, bypassing knapsack...\n");
    }


    int num_deleted = 0;
    long long bytes_freed = 0;

    while (!delete_dos.empty()) {
        DataObjectUtilityMetadata *to_delete = delete_dos.front();
        delete_dos.pop_front();
        if (!to_delete) {
            HAGGLE_ERR("NULL utility\n");
            continue;
        }
        DataObjectId_t id;
        DataObject::idStrToId(to_delete->getId(), id);
		bool deletionEnabled = to_delete->getEnableDeletion();

        if (to_delete->getId() == doid && o_was_deleted) {
            *o_was_deleted = true;
        }

        bytes_freed += to_delete->getCost();

        num_deleted++;

        // some DOs should not be deleted immedaitely 
        // (they may not even be inserted yet)
        if (deletionEnabled) {
            getManager()->getKernel()->getDataStore()->deleteDataObject(id, true);
        }

        DataObjectRef stale_dobj = to_delete->getDataObject();
        // SW: this will delete "to_delete"
        _handleDeletedDataObject(stale_dobj);
    }

    // sanity check
    if (current_size > (watermark_capacity_kb*1024)) {
        HAGGLE_ERR("Optimizer failed (did not remove enough bytes)\n");
//.........这里部分代码省略.........

ProtocolEvent ProtocolUDP::receiveDataObject()
{
	size_t len = 0;
	string haggleTag = "</Haggle>";
	DataObjectRef dObj;
        char buf[SOCKADDR_SIZE];
        struct sockaddr *peer_addr = (struct sockaddr *)buf;
	ProtocolEvent pEvent;
        unsigned short port;
        Address *addr = NULL;
	struct sockaddr_in *sa = NULL;

#ifdef OS_WINDOWS
	pEvent = receiveData(buffer, bufferSize, peer_addr, 0, &len);
#else
	pEvent = receiveData(buffer, bufferSize, peer_addr, MSG_DONTWAIT, &len);
#endif

	if (pEvent != PROT_EVENT_SUCCESS)
		return pEvent;

        if (peer_addr->sa_family == AF_INET) {
                sa = (struct sockaddr_in *)peer_addr;
                port = ntohs(sa->sin_port);
                addr = new IPv4Address(sa->sin_addr, TransportUDP(port));
	}
#if defined(ENABLE_IPv6) 
        else if (peer_addr->sa_family == AF_INET6) {
                struct sockaddr_in6 *sa6 = (struct sockaddr_in6 *)peer_addr;
                port = ntohs(sa6->sin6_port);
                addr = new IPv6Address(sa6->sin6_addr, TransportUDP(port));
        }
#endif

        if (addr == NULL)
                return PROT_EVENT_ERROR;

	if (peerIface) {
		HAGGLE_ERR("%s UDP peer interface was not null\n", getName());          
                delete addr;
		return PROT_EVENT_ERROR;
	}

        peerIface = new ApplicationPortInterface(port, "Application", addr, IFFLAG_UP);
        peerNode = getKernel()->getNodeStore()->retrieve(peerIface);

	delete addr;

	if (!peerNode) {
		peerNode = Node::create(Node::TYPE_APPLICATION, "Unknown application");

		if (!peerNode) {      
			HAGGLE_ERR("%s Could not create application node\n", getName());
			return PROT_EVENT_ERROR;
		}
	}

	dObj = DataObject::create(buffer, len, localIface, peerIface);

	if (!dObj) {
                HAGGLE_DBG("%s Could not create data object\n", getName());
		return PROT_EVENT_ERROR;
	}

        // Haggle doesn't own files that applications have put in: // MOS - not true - maybe some code missing here

	Timeval ct = dObj->getCreateTime();
	Timeval now = Timeval::now();
	if(!ct.isValid()) dObj->setCreateTime(now); // MOS - in case it was not set by application
	dObj->setReceiveTime(now);

	dataObjectsIncoming += 1; // MOS
	dataObjectBytesIncoming += len; // MOS
	if(!dObj->isControlMessage()) dataObjectsIncomingNonControl += 1; // MOS

        // We must release the peer interface reference after
        // the data object is created as the next incoming
        // data might be from another peer
        peerIface = NULL;

	// MOS - no need to ignore objects from local application (avoid risk of blocking control messages)
	// if (getKernel()->getThisNode()->getBloomfilter()->has(dObj)) {
	//	HAGGLE_DBG("Data object [%s] from interface %s:%u has already been received, ignoring.\n", 
	//		dObj->getIdStr(), sa ? ip_to_str(sa->sin_addr) : "undefined", port);
	//	return PROT_EVENT_SUCCESS;
	// }

	// Generate first an incoming event to conform with the base Protocol class
	getKernel()->addEvent(new Event(EVENT_TYPE_DATAOBJECT_INCOMING, dObj, peerNode));
	
	dataObjectsReceived += 1; // MOS
	dataObjectBytesReceived += len; // MOS

	HAGGLE_DBG("%s Received data object [%s] from interface %s:%u\n", getName(), 
		dObj->getIdStr(), sa ? ip_to_str(sa->sin_addr) : "undefined", port);

	// Since there is no data following, we generate the received event immediately 
	// following the incoming one
	getKernel()->addEvent(new Event(EVENT_TYPE_DATAOBJECT_RECEIVED, dObj, peerNode));

//.........这里部分代码省略.........

//.........这里部分代码省略.........
	// LAPDm operation, from GSM 04.11, Annex F:
	// """
	// Case A: Mobile originating short message transfer, no parallel call:
	// The mobile station side will initiate SAPI 3 establishment by a SABM command
	// on the SDCCH after the cipher mode has been set. If no hand over occurs, the
	// SAPI 3 link will stay up until the last CP-ACK is received by the MSC, and
	// the clearing procedure is invoked.
	// """

	// FIXME: check provisioning

	// Let the phone know we're going ahead with the transaction.
	LOG(INFO) << "sending CMServiceAccept";
	LCH->send(L3CMServiceAccept());
	// Wait for SAP3 to connect.
	// The first read on SAP3 is the ESTABLISH primitive.
	delete getFrameSMS(LCH,ESTABLISH);

	// Step 1
	// Now get the first message.
	// Should be CP-DATA, containing RP-DATA.
	L3Frame *CM = getFrameSMS(LCH);
	LOG(DEBUG) << "data from MS " << *CM;
	if (CM->MTI()!=CPMessage::DATA) {
		LOG(NOTICE) << "unexpected SMS CP message with TI=" << CM->MTI();
		throw UnexpectedMessage();
	}
	unsigned TI = CM->TIValue();

	// Step 2
	// Respond with CP-ACK.
	// This just means that we got the message.
	LOG(INFO) << "sending CPAck";
	LCH->send(CPAck(1,TI),3);

	// Parse the message in CM and process RP part.
	// This is where we actually parse the message and send it out.
	// FIXME -- We need to set the message ref correctly,
	// even if the parsing fails.
	// The compiler gives a warning here.  Let it.  It will remind someone to fix it.
	unsigned ref;
	bool success = false;
	try {
		CPData data;
		data.parse(*CM);
		delete CM;
		LOG(INFO) << "CPData " << data;
		// Transfer out the RPDU -> TPDU -> delivery.
		ref = data.RPDU().reference();
		// This handler invokes higher-layer parsers, too.
		success = handleRPDU(mobileIdentity,data.RPDU());
	}
	catch (SMSReadError) {
		LOG(WARN) << "SMS parsing failed (above L3)";
		// Cause 95, "semantically incorrect message".
		LCH->send(CPData(1,TI,RPError(95,ref)),3);
		throw UnexpectedMessage();
	}
	catch (L3ReadError) {
		LOG(WARN) << "SMS parsing failed (in L3)";
		throw UnsupportedMessage();
	}

	// Step 3
	// Send CP-DATA containing RP-ACK and message reference.
	if (success) {
		LOG(INFO) << "sending RPAck in CPData";
		LCH->send(CPData(1,TI,RPAck(ref)),3);
	} else {
		LOG(INFO) << "sending RPError in CPData";
		// Cause 127 is "internetworking error, unspecified".
		// See GSM 04.11 Table 8.4.
		LCH->send(CPData(1,TI,RPError(127,ref)),3);
	}

	// Step 4
	// Get CP-ACK from the MS.
	CM = getFrameSMS(LCH);
	if (CM->MTI()!=CPMessage::ACK) {
		LOG(NOTICE) << "unexpected SMS CP message with TI=" << CM->MTI();
		throw UnexpectedMessage();
	}
	LOG(DEBUG) << "ack from MS: " << *CM;
	CPAck ack;
	ack.parse(*CM);
	LOG(INFO) << "CPAck " << ack;

	// RRLP Here if enabled
	if (gConfig.defines("GSM.RRLP") && gConfig.getNum("GSM.RRLP") == 1 &&
		gConfig.defines("RRLP.LocationUpdate") && gConfig.getNum("RRLP.LocationUpdate") == 1 /* RRLP? */)
	{
		Timeval start;
		RRLP::collectMSInfo(mobileIdentity, LCH, true /* DO RRLP */);
		LOG(INFO) << "submitSMS with RRLP took " << start.elapsed() << " for IMSI " << mobileIdentity;
	}

	// Done.
	LOG(INFO) << "closing";
	LCH->send(L3ChannelRelease());
}

int Timeval::operator>=(const Timeval& arg2) const {
  return seconds() > arg2.seconds()
    || (seconds() == arg2.seconds()
	&& useconds() >= arg2.useconds());
}

bool HaggleKernel::init()
{
	char hostname[HOSTNAME_LEN];

#if defined(DEBUG) && (defined(OS_WINDOWS_MOBILE) || defined(OS_ANDROID) || defined(OS_LINUX)) // MOS - added linux
	if (!Trace::trace.enableFileTrace()) {
	  // HAGGLE_ERR("Could not enable file tracing\n"); // MOS - enableFileTrace generates accurate message
	}
#endif

// MOS - disable trace file by default for better performance
#if defined(LOGTRACE)
	if (!LogTrace::init()) {
		HAGGLE_ERR("Could not open trace file\n");
	}
#endif

#ifdef OS_WINDOWS
	WSADATA wsaData;
	int iResult;

	// Initialize Winsock
	iResult = WSAStartup(MAKEWORD(2, 2), &wsaData);

	if (iResult != 0) {
		HAGGLE_ERR("WSAStartup failed: %d\n", iResult);
                return false;
	}
#endif

	HAGGLE_LOG("Git reference is %s\n", GIT_REF);

	Timeval startupTime = Timeval::now(); // MOS
	HAGGLE_LOG("Startup Time is %s\n", startupTime.getAsString().c_str());

	HAGGLE_LOG("Storage path is %s\n", storagepath.c_str());
	HAGGLE_LOG("Datastore path is %s\n", DEFAULT_DATASTORE_PATH);
	
#if defined(OS_ANDROID)
	int num = getgroups(0, NULL);

	HAGGLE_LOG("Haggle has the permissions of the following groups:\n");

	if (num > 0) {
		gid_t *groups = new gid_t[num];
		
		if (groups) { 
			if (getgroups(num, groups) == num) {
				for (int i = 0; i < num; i++) {
					struct group *g = getgrgid(groups[i]);
					if (g) {
						HAGGLE_LOG("group: %u %s\n", g->gr_gid, g->gr_name);
					}
				}
			}
			delete [] groups;
		}
	}

	struct group *g = getgrnam("bluetooth");

	if (g) {
		HAGGLE_LOG("Haggle needs the permissions of group %u %s, but is currently not granted\n", g->gr_gid, g->gr_name);
	}
#endif

	if (dataStore) {
		dataStore->kernel = this;
		if (!dataStore->init()) {
			HAGGLE_ERR("Data store could not be initialized\n");
			return false;
		}
	} else {
		HAGGLE_ERR("No data store!!!\n");
		return false;
	}

	// The interfaces on this node will be discovered when the
	// ConnectivityManager is started. Hence, they will not be
	// part of thisNode when we insert it in the datastore. This
	// is a potential problem.
	//dataStore->insertNode(&thisNode);

	int res = gethostname(hostname, HOSTNAME_LEN);

	if (res != 0) {
		HAGGLE_ERR("Could not get hostname\n");
		return false;
	}
	
	// Always zero terminate in case the hostname didn't fit
	hostname[HOSTNAME_LEN-1] = '\0';
        
        HAGGLE_LOG("Hostname is %s\n", hostname);

	thisNode = nodeStore.add(Node::create(Node::TYPE_LOCAL_DEVICE, hostname));

	if (!thisNode) {
		HAGGLE_ERR("Could not create this node\n");
		return false;
//.........这里部分代码省略.........

Timeval::Timeval(Timeval const &cpy)
{
  this->_tv.tv_sec = cpy.getSec();
  this->_tv.tv_usec = cpy.getUSec();
}