// Copyright (c) 2009 The Chromium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#include "net/tools/flip_server/output_ordering.h"
#include "net/tools/flip_server/flip_config.h"
#include "net/tools/flip_server/sm_connection.h"
namespace net {
// static
double OutputOrdering::server_think_time_in_s_ = 0.0;
OutputOrdering::OutputOrdering(SMConnectionInterface* connection)
: first_data_senders_threshold_(kInitialDataSendersThreshold),
connection_(connection) {
if (connection)
epoll_server_ = connection->epoll_server();
}
OutputOrdering::~OutputOrdering() {}
void OutputOrdering::Reset() {
while (!stream_ids_.empty()) {
StreamIdToPriorityMap::iterator sitpmi = stream_ids_.begin();
PriorityMapPointer& pmp = sitpmi->second;
if (pmp.alarm_enabled) {
epoll_server_->UnregisterAlarm(pmp.alarm_token);
}
stream_ids_.erase(sitpmi);
}
priority_map_.clear();
first_data_senders_.clear();
}
bool OutputOrdering::ExistsInPriorityMaps(uint32 stream_id) {
StreamIdToPriorityMap::iterator sitpmi = stream_ids_.find(stream_id);
return sitpmi != stream_ids_.end();
}
OutputOrdering::BeginOutputtingAlarm::BeginOutputtingAlarm(
OutputOrdering* oo,
OutputOrdering::PriorityMapPointer* pmp,
const MemCacheIter& mci)
: output_ordering_(oo),
pmp_(pmp),
mci_(mci),
epoll_server_(NULL) {
}
OutputOrdering::BeginOutputtingAlarm::~BeginOutputtingAlarm() {
if (epoll_server_ && pmp_->alarm_enabled)
epoll_server_->UnregisterAlarm(pmp_->alarm_token);
}
int64 OutputOrdering::BeginOutputtingAlarm::OnAlarm() {
OnUnregistration();
output_ordering_->MoveToActive(pmp_, mci_);
VLOG(2) << "ON ALARM! Should now start to output...";
delete this;
return 0;
}
void OutputOrdering::BeginOutputtingAlarm::OnRegistration(
const EpollServer::AlarmRegToken& tok,
EpollServer* eps) {
epoll_server_ = eps;
pmp_->alarm_token = tok;
pmp_->alarm_enabled = true;
}
void OutputOrdering::BeginOutputtingAlarm::OnUnregistration() {
pmp_->alarm_enabled = false;
}
void OutputOrdering::BeginOutputtingAlarm::OnShutdown(EpollServer* eps) {
OnUnregistration();
}
void OutputOrdering::MoveToActive(PriorityMapPointer* pmp, MemCacheIter mci) {
VLOG(2) << "Moving to active!";
first_data_senders_.push_back(mci);
pmp->ring = &first_data_senders_;
pmp->it = first_data_senders_.end();
--pmp->it;
connection_->ReadyToSend();
}
void OutputOrdering::AddToOutputOrder(const MemCacheIter& mci) {
if (ExistsInPriorityMaps(mci.stream_id))
LOG(ERROR) << "OOps, already was inserted here?!";
double think_time_in_s = server_think_time_in_s_;
std::string x_server_latency =
mci.file_data->headers->GetHeader("X-Server-Latency").as_string();
if (!x_server_latency.empty()) {
char* endp;
double tmp_think_time_in_s = strtod(x_server_latency.c_str(), &endp);
if (endp != x_server_latency.c_str() + x_server_latency.size()) {
LOG(ERROR) << "Unable to understand X-Server-Latency of: "
<< x_server_latency << " for resource: "
<< mci.file_data->filename.c_str();
} else {
think_time_in_s = tmp_think_time_in_s;
}
}
StreamIdToPriorityMap::iterator sitpmi;
sitpmi = stream_ids_.insert(
std::pair<uint32, PriorityMapPointer>(mci.stream_id,
PriorityMapPointer())).first;
PriorityMapPointer& pmp = sitpmi->second;
BeginOutputtingAlarm* boa = new BeginOutputtingAlarm(this, &pmp, mci);
VLOG(1) << "Server think time: " << think_time_in_s;
epoll_server_->RegisterAlarmApproximateDelta(
think_time_in_s * 1000000, boa);
}
void OutputOrdering::SpliceToPriorityRing(PriorityRing::iterator pri) {
MemCacheIter& mci = *pri;
PriorityMap::iterator pmi = priority_map_.find(mci.priority);
if (pmi == priority_map_.end()) {
pmi = priority_map_.insert(
std::pair<uint32, PriorityRing>(mci.priority, PriorityRing())).first;
}
pmi->second.splice(pmi->second.end(),
first_data_senders_,
pri);
StreamIdToPriorityMap::iterator sitpmi = stream_ids_.find(mci.stream_id);
sitpmi->second.ring = &(pmi->second);
}
MemCacheIter* OutputOrdering::GetIter() {
while (!first_data_senders_.empty()) {
MemCacheIter& mci = first_data_senders_.front();
if (mci.bytes_sent >= first_data_senders_threshold_) {
SpliceToPriorityRing(first_data_senders_.begin());
} else {
first_data_senders_.splice(first_data_senders_.end(),
first_data_senders_,
first_data_senders_.begin());
mci.max_segment_size = kInitialDataSendersThreshold;
return &mci;
}
}
while (!priority_map_.empty()) {
PriorityRing& first_ring = priority_map_.begin()->second;
if (first_ring.empty()) {
priority_map_.erase(priority_map_.begin());
continue;
}
MemCacheIter& mci = first_ring.front();
first_ring.splice(first_ring.end(),
first_ring,
first_ring.begin());
mci.max_segment_size = kSpdySegmentSize;
return &mci;
}
return NULL;
}
void OutputOrdering::RemoveStreamId(uint32 stream_id) {
StreamIdToPriorityMap::iterator sitpmi = stream_ids_.find(stream_id);
if (sitpmi == stream_ids_.end())
return;
PriorityMapPointer& pmp = sitpmi->second;
if (pmp.alarm_enabled)
epoll_server_->UnregisterAlarm(pmp.alarm_token);
else
pmp.ring->erase(pmp.it);
stream_ids_.erase(sitpmi);
}
} // namespace net