1// Copyright (c) 2009 The Chromium Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5#include "net/tools/flip_server/output_ordering.h"
6
7#include <utility>
8
9#include "net/tools/flip_server/flip_config.h"
10#include "net/tools/flip_server/sm_connection.h"
11
12namespace net {
13
14OutputOrdering::PriorityMapPointer::PriorityMapPointer()
15    : ring(NULL), alarm_enabled(false) {}
16
17OutputOrdering::PriorityMapPointer::~PriorityMapPointer() {}
18
19// static
20double OutputOrdering::server_think_time_in_s_ = 0.0;
21
22OutputOrdering::OutputOrdering(SMConnectionInterface* connection)
23    : first_data_senders_threshold_(kInitialDataSendersThreshold),
24      connection_(connection) {
25  if (connection)
26    epoll_server_ = connection->epoll_server();
27}
28
29OutputOrdering::~OutputOrdering() { Reset(); }
30
31void OutputOrdering::Reset() {
32  while (!stream_ids_.empty()) {
33    StreamIdToPriorityMap::iterator sitpmi = stream_ids_.begin();
34    PriorityMapPointer& pmp = sitpmi->second;
35    if (pmp.alarm_enabled) {
36      epoll_server_->UnregisterAlarm(pmp.alarm_token);
37    }
38    stream_ids_.erase(sitpmi);
39  }
40  priority_map_.clear();
41  first_data_senders_.clear();
42}
43
44bool OutputOrdering::ExistsInPriorityMaps(uint32 stream_id) const {
45  StreamIdToPriorityMap::const_iterator sitpmi = stream_ids_.find(stream_id);
46  return sitpmi != stream_ids_.end();
47}
48
49OutputOrdering::BeginOutputtingAlarm::BeginOutputtingAlarm(
50    OutputOrdering* oo,
51    OutputOrdering::PriorityMapPointer* pmp,
52    const MemCacheIter& mci)
53    : output_ordering_(oo), pmp_(pmp), mci_(mci), epoll_server_(NULL) {}
54
55OutputOrdering::BeginOutputtingAlarm::~BeginOutputtingAlarm() {
56  if (epoll_server_ && pmp_->alarm_enabled)
57    epoll_server_->UnregisterAlarm(pmp_->alarm_token);
58}
59
60int64 OutputOrdering::BeginOutputtingAlarm::OnAlarm() {
61  OnUnregistration();
62  output_ordering_->MoveToActive(pmp_, mci_);
63  VLOG(2) << "ON ALARM! Should now start to output...";
64  delete this;
65  return 0;
66}
67
68void OutputOrdering::BeginOutputtingAlarm::OnRegistration(
69    const EpollServer::AlarmRegToken& tok,
70    EpollServer* eps) {
71  epoll_server_ = eps;
72  pmp_->alarm_token = tok;
73  pmp_->alarm_enabled = true;
74}
75
76void OutputOrdering::BeginOutputtingAlarm::OnUnregistration() {
77  pmp_->alarm_enabled = false;
78  delete this;
79}
80
81void OutputOrdering::BeginOutputtingAlarm::OnShutdown(EpollServer* eps) {
82  OnUnregistration();
83}
84
85void OutputOrdering::MoveToActive(PriorityMapPointer* pmp, MemCacheIter mci) {
86  VLOG(2) << "Moving to active!";
87  first_data_senders_.push_back(mci);
88  pmp->ring = &first_data_senders_;
89  pmp->it = first_data_senders_.end();
90  --pmp->it;
91  connection_->ReadyToSend();
92}
93
94void OutputOrdering::AddToOutputOrder(const MemCacheIter& mci) {
95  if (ExistsInPriorityMaps(mci.stream_id))
96    LOG(ERROR) << "OOps, already was inserted here?!";
97
98  double think_time_in_s = server_think_time_in_s_;
99  std::string x_server_latency =
100      mci.file_data->headers()->GetHeader("X-Server-Latency").as_string();
101  if (!x_server_latency.empty()) {
102    char* endp;
103    double tmp_think_time_in_s = strtod(x_server_latency.c_str(), &endp);
104    if (endp != x_server_latency.c_str() + x_server_latency.size()) {
105      LOG(ERROR) << "Unable to understand X-Server-Latency of: "
106                 << x_server_latency
107                 << " for resource: " << mci.file_data->filename().c_str();
108    } else {
109      think_time_in_s = tmp_think_time_in_s;
110    }
111  }
112  StreamIdToPriorityMap::iterator sitpmi;
113  sitpmi = stream_ids_.insert(std::pair<uint32, PriorityMapPointer>(
114                                  mci.stream_id, PriorityMapPointer())).first;
115  PriorityMapPointer& pmp = sitpmi->second;
116
117  BeginOutputtingAlarm* boa = new BeginOutputtingAlarm(this, &pmp, mci);
118  VLOG(1) << "Server think time: " << think_time_in_s;
119  epoll_server_->RegisterAlarmApproximateDelta(think_time_in_s * 1000000, boa);
120}
121
122void OutputOrdering::SpliceToPriorityRing(PriorityRing::iterator pri) {
123  MemCacheIter& mci = *pri;
124  PriorityMap::iterator pmi = priority_map_.find(mci.priority);
125  if (pmi == priority_map_.end()) {
126    pmi = priority_map_.insert(std::pair<uint32, PriorityRing>(
127                                   mci.priority, PriorityRing())).first;
128  }
129
130  pmi->second.splice(pmi->second.end(), first_data_senders_, pri);
131  StreamIdToPriorityMap::iterator sitpmi = stream_ids_.find(mci.stream_id);
132  sitpmi->second.ring = &(pmi->second);
133}
134
135MemCacheIter* OutputOrdering::GetIter() {
136  while (!first_data_senders_.empty()) {
137    MemCacheIter& mci = first_data_senders_.front();
138    if (mci.bytes_sent >= first_data_senders_threshold_) {
139      SpliceToPriorityRing(first_data_senders_.begin());
140    } else {
141      first_data_senders_.splice(first_data_senders_.end(),
142                                 first_data_senders_,
143                                 first_data_senders_.begin());
144      mci.max_segment_size = kInitialDataSendersThreshold;
145      return &mci;
146    }
147  }
148  while (!priority_map_.empty()) {
149    PriorityRing& first_ring = priority_map_.begin()->second;
150    if (first_ring.empty()) {
151      priority_map_.erase(priority_map_.begin());
152      continue;
153    }
154    MemCacheIter& mci = first_ring.front();
155    first_ring.splice(first_ring.end(), first_ring, first_ring.begin());
156    mci.max_segment_size = kSpdySegmentSize;
157    return &mci;
158  }
159  return NULL;
160}
161
162void OutputOrdering::RemoveStreamId(uint32 stream_id) {
163  StreamIdToPriorityMap::iterator sitpmi = stream_ids_.find(stream_id);
164  if (sitpmi == stream_ids_.end())
165    return;
166
167  PriorityMapPointer& pmp = sitpmi->second;
168  if (pmp.alarm_enabled)
169    epoll_server_->UnregisterAlarm(pmp.alarm_token);
170  else
171    pmp.ring->erase(pmp.it);
172  stream_ids_.erase(sitpmi);
173}
174
175}  // namespace net
176