flip_in_mem_edsm_server.cc revision c407dc5cd9bdc5668497f21b26b09d988ab439de 
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 <dirent.h>
6#include <linux/tcp.h>  // For TCP_NODELAY
7#include <sys/socket.h>
8#include <sys/types.h>
9#include <unistd.h>
10#include <openssl/err.h>
11#include <openssl/ssl.h>
12
13#include <deque>
14#include <iostream>
15#include <limits>
16#include <vector>
17#include <list>
18
19#include "base/logging.h"
20#include "base/simple_thread.h"
21#include "base/timer.h"
22#include "base/lock.h"
23#include "net/spdy/spdy_frame_builder.h"
24#include "net/spdy/spdy_framer.h"
25#include "net/spdy/spdy_protocol.h"
26#include "net/tools/flip_server/balsa_enums.h"
27#include "net/tools/flip_server/balsa_frame.h"
28#include "net/tools/flip_server/balsa_headers.h"
29#include "net/tools/flip_server/balsa_visitor_interface.h"
30#include "net/tools/flip_server/buffer_interface.h"
31#include "net/tools/flip_server/create_listener.h"
32#include "net/tools/flip_server/epoll_server.h"
33#include "net/tools/flip_server/other_defines.h"
34#include "net/tools/flip_server/ring_buffer.h"
35#include "net/tools/flip_server/simple_buffer.h"
36#include "net/tools/flip_server/split.h"
37#include "net/tools/flip_server/url_to_filename_encoder.h"
38#include "net/tools/flip_server/url_utilities.h"
39
40////////////////////////////////////////////////////////////////////////////////
41
42using std::cerr;
43using std::deque;
44using std::list;
45using std::map;
46using std::ostream;
47using std::pair;
48using std::string;
49using std::vector;
50
51////////////////////////////////////////////////////////////////////////////////
52
53
54//         If set to true, then the server will act as an SSL server for both
55//          HTTP and SPDY);
56bool FLAGS_use_ssl = true;
57
58// The name of the cert .pem file);
59string FLAGS_ssl_cert_name = "cert.pem";
60
61// The name of the key .pem file);
62string FLAGS_ssl_key_name = "key.pem";
63
64// The number of responses given before the server closes the
65//  connection);
66int32 FLAGS_response_count_until_close = 1000*1000;
67
68// If true, then disables the nagle algorithm);
69bool FLAGS_no_nagle = true;
70
71// The number of times that accept() will be called when the
72//  alarm goes off when the accept_using_alarm flag is set to true.
73//  If set to 0, accept() will be performed until the accept queue
74//  is completely drained and the accept() call returns an error);
75int32 FLAGS_accepts_per_wake = 0;
76
77// The port on which the spdy server listens);
78int32 FLAGS_spdy_port = 10040;
79
80// The port on which the http server listens);
81int32 FLAGS_port = 16002;
82
83// The size of the TCP accept backlog);
84int32 FLAGS_accept_backlog_size = 1024;
85
86// The directory where cache locates);
87string FLAGS_cache_base_dir = ".";
88
89// If true, then encode url to filename);
90bool FLAGS_need_to_encode_url = true;
91
92// If set to false a single socket will be used. If set to true
93//  then a new socket will be created for each accept thread.
94//  Note that this only works with kernels that support
95//  SO_REUSEPORT);
96bool FLAGS_reuseport = false;
97
98// The amount of time the server delays before sending back the
99//  reply);
100double FLAGS_server_think_time_in_s = 0;
101
102// Does the server send X-Subresource headers);
103bool FLAGS_use_xsub = false;
104
105// Does the server send X-Associated-Content headers);
106bool FLAGS_use_xac = false;
107
108// Does the server advance cwnd by sending no-op packets);
109bool FLAGS_use_cwnd_opener = false;
110
111// Does the server compress data frames);
112bool FLAGS_use_compression = false;
113
114////////////////////////////////////////////////////////////////////////////////
115
116using base::StringPiece;
117using base::SimpleThread;
118// using base::Lock;  // heh, this isn't in base namespace?!
119// using base::AutoLock;  // ditto!
120using net::BalsaFrame;
121using net::BalsaFrameEnums;
122using net::BalsaHeaders;
123using net::BalsaHeadersEnums;
124using net::BalsaVisitorInterface;
125using net::EpollAlarmCallbackInterface;
126using net::EpollCallbackInterface;
127using net::EpollEvent;
128using net::EpollServer;
129using net::RingBuffer;
130using net::SimpleBuffer;
131using net::SplitStringPieceToVector;
132using net::UrlUtilities;
133using spdy::CONTROL_FLAG_NONE;
134using spdy::DATA_FLAG_COMPRESSED;
135using spdy::DATA_FLAG_FIN;
136using spdy::RST_STREAM;
137using spdy::SYN_REPLY;
138using spdy::SYN_STREAM;
139using spdy::SpdyControlFrame;
140using spdy::SpdyDataFlags;
141using spdy::SpdyDataFrame;
142using spdy::SpdyRstStreamControlFrame;
143using spdy::SpdyFrame;
144using spdy::SpdyFrameBuilder;
145using spdy::SpdyFramer;
146using spdy::SpdyFramerVisitorInterface;
147using spdy::SpdyHeaderBlock;
148using spdy::SpdyStreamId;
149using spdy::SpdySynReplyControlFrame;
150using spdy::SpdySynStreamControlFrame;
151
152
153////////////////////////////////////////////////////////////////////////////////
154
155void PrintSslError() {
156  char buf[128];  // this buffer must be at least 120 chars long.
157  int error_num = ERR_get_error();
158  while (error_num != 0) {
159    LOG(INFO)<< ERR_error_string(error_num, buf);
160    error_num = ERR_get_error();
161  }
162}
163
164////////////////////////////////////////////////////////////////////////////////
165
166// Creates a socket with domain, type and protocol parameters.
167// Assigns the return value of socket() to *fd.
168// Returns errno if an error occurs, else returns zero.
169int CreateSocket(int domain, int type, int protocol, int *fd) {
170  CHECK(fd != NULL);
171  *fd = ::socket(domain, type, protocol);
172  return (*fd == -1) ? errno : 0;
173}
174
175////////////////////////////////////////////////////////////////////////////////
176
177// Sets an FD to be nonblocking.
178void SetNonBlocking(int fd) {
179  DCHECK(fd >= 0);
180
181  int fcntl_return = fcntl(fd, F_GETFL, 0);
182  CHECK_NE(fcntl_return, -1)
183    << "error doing fcntl(fd, F_GETFL, 0) fd: " << fd
184    << " errno=" << errno;
185
186  if (fcntl_return & O_NONBLOCK)
187    return;
188
189  fcntl_return = fcntl(fd, F_SETFL, fcntl_return | O_NONBLOCK);
190  CHECK_NE(fcntl_return, -1)
191    << "error doing fcntl(fd, F_SETFL, fcntl_return) fd: " << fd
192    << " errno=" << errno;
193}
194
195// Encode the URL.
196string EncodeURL(string uri, string host, string method) {
197  if (!FLAGS_need_to_encode_url) {
198    // TODO(mbelshe): if uri is fully qualified, need to strip protocol/host.
199    return string(method + "_" + uri);
200  }
201
202  string filename;
203  if (uri[0] == '/') {
204    // uri is not fully qualified.
205    filename = net::UrlToFilenameEncoder::Encode(
206        "http://" + host + uri, method + "_/");
207  } else {
208    filename = net::UrlToFilenameEncoder::Encode(uri, method + "_/");
209  }
210  return filename;
211}
212
213////////////////////////////////////////////////////////////////////////////////
214
215
216struct GlobalSSLState {
217  SSL_METHOD* ssl_method;
218  SSL_CTX* ssl_ctx;
219};
220
221////////////////////////////////////////////////////////////////////////////////
222
223GlobalSSLState* global_ssl_state = NULL;
224
225////////////////////////////////////////////////////////////////////////////////
226
227// SSL stuff
228void spdy_init_ssl(GlobalSSLState* state) {
229  SSL_library_init();
230  PrintSslError();
231
232  SSL_load_error_strings();
233  PrintSslError();
234
235  state->ssl_method = SSLv23_method();
236  state->ssl_ctx = SSL_CTX_new(state->ssl_method);
237  if (!state->ssl_ctx) {
238    PrintSslError();
239    LOG(FATAL) << "Unable to create SSL context";
240  }
241  // Disable SSLv2 support.
242  SSL_CTX_set_options(state->ssl_ctx, SSL_OP_NO_SSLv2);
243  if (SSL_CTX_use_certificate_file(state->ssl_ctx,
244                                   FLAGS_ssl_cert_name.c_str(),
245                                   SSL_FILETYPE_PEM) <= 0) {
246    PrintSslError();
247    LOG(FATAL) << "Unable to use cert.pem as SSL cert.";
248  }
249  if (SSL_CTX_use_PrivateKey_file(state->ssl_ctx,
250                                  FLAGS_ssl_key_name.c_str(),
251                                  SSL_FILETYPE_PEM) <= 0) {
252    PrintSslError();
253    LOG(FATAL) << "Unable to use key.pem as SSL key.";
254  }
255  if (!SSL_CTX_check_private_key(state->ssl_ctx)) {
256    PrintSslError();
257    LOG(FATAL) << "The cert.pem and key.pem files don't match";
258  }
259}
260
261SSL* spdy_new_ssl(SSL_CTX* ssl_ctx) {
262  SSL* ssl = SSL_new(ssl_ctx);
263  PrintSslError();
264
265  SSL_set_accept_state(ssl);
266  PrintSslError();
267  return ssl;
268}
269
270////////////////////////////////////////////////////////////////////////////////
271
272const int kMSS = 1460;
273const int kInitialDataSendersThreshold = (2 * kMSS) - SpdyFrame::size();
274const int kNormalSegmentSize = (2 * kMSS) - SpdyFrame::size();
275
276////////////////////////////////////////////////////////////////////////////////
277
278class DataFrame {
279 public:
280  const char* data;
281  size_t size;
282  bool delete_when_done;
283  size_t index;
284  DataFrame() : data(NULL), size(0), delete_when_done(false), index(0) {}
285  void MaybeDelete() {
286    if (delete_when_done) {
287      delete[] data;
288    }
289  }
290};
291
292////////////////////////////////////////////////////////////////////////////////
293
294class StoreBodyAndHeadersVisitor: public BalsaVisitorInterface {
295 public:
296  BalsaHeaders headers;
297  string body;
298  bool error_;
299
300  virtual void ProcessBodyInput(const char *input, size_t size) {}
301  virtual void ProcessBodyData(const char *input, size_t size) {
302    body.append(input, size);
303  }
304  virtual void ProcessHeaderInput(const char *input, size_t size) {}
305  virtual void ProcessTrailerInput(const char *input, size_t size) {}
306  virtual void ProcessHeaders(const BalsaHeaders& headers) {
307    // nothing to do here-- we're assuming that the BalsaFrame has
308    // been handed our headers.
309  }
310  virtual void ProcessRequestFirstLine(const char* line_input,
311                                       size_t line_length,
312                                       const char* method_input,
313                                       size_t method_length,
314                                       const char* request_uri_input,
315                                       size_t request_uri_length,
316                                       const char* version_input,
317                                       size_t version_length) {}
318  virtual void ProcessResponseFirstLine(const char *line_input,
319                                        size_t line_length,
320                                        const char *version_input,
321                                        size_t version_length,
322                                        const char *status_input,
323                                        size_t status_length,
324                                        const char *reason_input,
325                                        size_t reason_length) {}
326  virtual void ProcessChunkLength(size_t chunk_length) {}
327  virtual void ProcessChunkExtensions(const char *input, size_t size) {}
328  virtual void HeaderDone() {}
329  virtual void MessageDone() {}
330  virtual void HandleHeaderError(BalsaFrame* framer) { HandleError(); }
331  virtual void HandleHeaderWarning(BalsaFrame* framer) { HandleError(); }
332  virtual void HandleChunkingError(BalsaFrame* framer) { HandleError(); }
333  virtual void HandleBodyError(BalsaFrame* framer) { HandleError(); }
334
335  void HandleError() { error_ = true; }
336};
337
338////////////////////////////////////////////////////////////////////////////////
339
340struct FileData {
341  void CopyFrom(const FileData& file_data) {
342    headers = new BalsaHeaders;
343    headers->CopyFrom(*(file_data.headers));
344    filename = file_data.filename;
345    related_files = file_data.related_files;
346    body = file_data.body;
347  }
348  FileData(BalsaHeaders* h, const string& b) : headers(h), body(b) {}
349  FileData() {}
350  BalsaHeaders* headers;
351  string filename;
352  vector< pair<int, string> > related_files;   // priority, filename
353  string body;
354};
355
356////////////////////////////////////////////////////////////////////////////////
357
358class MemCacheIter {
359 public:
360  MemCacheIter() :
361      file_data(NULL),
362      priority(0),
363      transformed_header(false),
364      body_bytes_consumed(0),
365      stream_id(0),
366      max_segment_size(kInitialDataSendersThreshold),
367      bytes_sent(0) {}
368  explicit MemCacheIter(FileData* fd) :
369      file_data(fd),
370      priority(0),
371      transformed_header(false),
372      body_bytes_consumed(0),
373      stream_id(0),
374      max_segment_size(kInitialDataSendersThreshold),
375      bytes_sent(0) {}
376  FileData* file_data;
377  int priority;
378  bool transformed_header;
379  size_t body_bytes_consumed;
380  uint32 stream_id;
381  uint32 max_segment_size;
382  size_t bytes_sent;
383};
384
385////////////////////////////////////////////////////////////////////////////////
386
387class MemoryCache {
388 public:
389  typedef map<string, FileData> Files;
390
391 public:
392  Files files_;
393  string cwd_;
394
395  void CloneFrom(const MemoryCache& mc) {
396    for (Files::const_iterator i = mc.files_.begin();
397         i != mc.files_.end();
398         ++i) {
399      Files::iterator out_i =
400        files_.insert(make_pair(i->first, FileData())).first;
401      out_i->second.CopyFrom(i->second);
402      cwd_ = mc.cwd_;
403    }
404  }
405
406  void AddFiles() {
407    LOG(INFO) << "Adding files!";
408    deque<string> paths;
409    cwd_ = FLAGS_cache_base_dir;
410    paths.push_back(cwd_ + "/GET_");
411    DIR* current_dir = NULL;
412    while (!paths.empty()) {
413      while (current_dir == NULL && !paths.empty()) {
414        string current_dir_name = paths.front();
415        VLOG(1) << "Attempting to open dir: \"" << current_dir_name << "\"";
416        current_dir = opendir(current_dir_name.c_str());
417        paths.pop_front();
418
419        if (current_dir == NULL) {
420          perror("Unable to open directory. ");
421          current_dir_name.clear();
422          continue;
423        }
424
425        if (current_dir) {
426          VLOG(1) << "Succeeded opening";
427          for (struct dirent* dir_data = readdir(current_dir);
428               dir_data != NULL;
429               dir_data = readdir(current_dir)) {
430            string current_entry_name =
431              current_dir_name + "/" + dir_data->d_name;
432            if (dir_data->d_type == DT_REG) {
433              VLOG(1) << "Found file: " << current_entry_name;
434              ReadAndStoreFileContents(current_entry_name.c_str());
435            } else if (dir_data->d_type == DT_DIR) {
436              VLOG(1) << "Found subdir: " << current_entry_name;
437              if (string(dir_data->d_name) != "." &&
438                  string(dir_data->d_name) != "..") {
439                VLOG(1) << "Adding to search path: " << current_entry_name;
440                paths.push_front(current_entry_name);
441              }
442            }
443          }
444          VLOG(1) << "Oops, no data left. Closing dir.";
445          closedir(current_dir);
446          current_dir = NULL;
447        }
448      }
449    }
450  }
451
452  void ReadToString(const char* filename, string* output) {
453    output->clear();
454    int fd = open(filename, 0, "r");
455    if (fd == -1)
456      return;
457    char buffer[4096];
458    ssize_t read_status = read(fd, buffer, sizeof(buffer));
459    while (read_status > 0) {
460      output->append(buffer, static_cast<size_t>(read_status));
461      do {
462        read_status = read(fd, buffer, sizeof(buffer));
463      } while (read_status <= 0 && errno == EINTR);
464    }
465    close(fd);
466  }
467
468  void ReadAndStoreFileContents(const char* filename) {
469    StoreBodyAndHeadersVisitor visitor;
470    BalsaFrame framer;
471    framer.set_balsa_visitor(&visitor);
472    framer.set_balsa_headers(&(visitor.headers));
473    string filename_contents;
474    ReadToString(filename, &filename_contents);
475
476    // Ugly hack to make everything look like 1.1.
477    if (filename_contents.find("HTTP/1.0") == 0)
478      filename_contents[7] = '1';
479
480    size_t pos = 0;
481    size_t old_pos = 0;
482    while (true) {
483      old_pos = pos;
484      pos += framer.ProcessInput(filename_contents.data() + pos,
485                                 filename_contents.size() - pos);
486      if (framer.Error() || pos == old_pos) {
487        LOG(ERROR) << "Unable to make forward progress, or error"
488          " framing file: " << filename;
489        if (framer.Error()) {
490          LOG(INFO) << "********************************************ERROR!";
491          return;
492        }
493        return;
494      }
495      if (framer.MessageFullyRead()) {
496        // If no Content-Length or Transfer-Encoding was captured in the
497        // file, then the rest of the data is the body.  Many of the captures
498        // from within Chrome don't have content-lengths.
499        if (!visitor.body.length())
500          visitor.body = filename_contents.substr(pos);
501        break;
502      }
503    }
504    visitor.headers.RemoveAllOfHeader("content-length");
505    visitor.headers.RemoveAllOfHeader("transfer-encoding");
506    visitor.headers.RemoveAllOfHeader("connection");
507    visitor.headers.AppendHeader("transfer-encoding", "chunked");
508    visitor.headers.AppendHeader("connection", "keep-alive");
509
510    // Experiment with changing headers for forcing use of cached
511    // versions of content.
512    // TODO(mbelshe) REMOVE ME
513#if 0
514    // TODO(mbelshe) append current date.
515    visitor.headers.RemoveAllOfHeader("date");
516    if (visitor.headers.HasHeader("expires")) {
517      visitor.headers.RemoveAllOfHeader("expires");
518      visitor.headers.AppendHeader("expires",
519                                 "Fri, 30 Aug, 2019 12:00:00 GMT");
520    }
521#endif
522    BalsaHeaders* headers = new BalsaHeaders;
523    headers->CopyFrom(visitor.headers);
524    string filename_stripped = string(filename).substr(cwd_.size() + 1);
525//    LOG(INFO) << "Adding file (" << visitor.body.length() << " bytes): "
526//              << filename_stripped;
527    files_[filename_stripped] = FileData();
528    FileData& fd = files_[filename_stripped];
529    fd = FileData(headers, visitor.body);
530    fd.filename = string(filename_stripped,
531                         filename_stripped.find_first_of('/'));
532    if (headers->HasHeader("X-Associated-Content")) {
533      string content = headers->GetHeader("X-Associated-Content").as_string();
534      vector<StringPiece> urls_and_priorities;
535      SplitStringPieceToVector(content, "||", &urls_and_priorities, true);
536      VLOG(1) << "Examining X-Associated-Content header";
537      for (unsigned int i = 0; i < urls_and_priorities.size(); ++i) {
538        const StringPiece& url_and_priority_pair = urls_and_priorities[i];
539        vector<StringPiece> url_and_priority;
540        SplitStringPieceToVector(url_and_priority_pair, "??",
541                                 &url_and_priority, true);
542        if (url_and_priority.size() >= 2) {
543          string priority_string(url_and_priority[0].data(),
544                                 url_and_priority[0].size());
545          string filename_string(url_and_priority[1].data(),
546                                 url_and_priority[1].size());
547          long priority;
548          char* last_eaten_char;
549          priority = strtol(priority_string.c_str(), &last_eaten_char, 0);
550          if (last_eaten_char ==
551              priority_string.c_str() + priority_string.size()) {
552            pair<int, string> entry(priority, filename_string);
553            VLOG(1) << "Adding associated content: " << filename_string;
554            fd.related_files.push_back(entry);
555          }
556        }
557      }
558    }
559  }
560
561  // Called at runtime to update learned headers
562  // |url| is a url which contains a referrer header.
563  // |referrer| is the referring URL
564  // Adds an X-Subresource or X-Associated-Content to |referer| for |url|
565  void UpdateHeaders(string referrer, string file_url) {
566    if (!FLAGS_use_xac && !FLAGS_use_xsub)
567      return;
568
569    string referrer_host_path =
570      net::UrlToFilenameEncoder::Encode(referrer, "GET_/");
571
572    FileData* fd1 = GetFileData(string("GET_") + file_url);
573    if (!fd1) {
574      LOG(ERROR) << "Updating headers for unknown url: " << file_url;
575      return;
576    }
577    string url = fd1->headers->GetHeader("X-Original-Url").as_string();
578    string content_type = fd1->headers->GetHeader("Content-Type").as_string();
579    if (content_type.length() == 0) {
580      LOG(ERROR) << "Skipping subresource with unknown content-type";
581      return;
582    }
583
584    // Now, lets see if this is the same host or not
585    bool same_host = (UrlUtilities::GetUrlHost(referrer) ==
586                      UrlUtilities::GetUrlHost(url));
587
588    // This is a hacked algorithm for figuring out what priority
589    // to use with pushed content.
590    int priority = 4;
591    if (content_type.find("css") != string::npos)
592      priority = 1;
593    else if (content_type.find("cript") != string::npos)
594      priority = 1;
595    else if (content_type.find("html") != string::npos)
596      priority = 2;
597
598    LOG(ERROR) << "Attempting update for " << referrer_host_path;
599
600    FileData* fd2 = GetFileData(referrer_host_path);
601    if (fd2 != NULL) {
602      // If they are on the same host, we'll use X-Associated-Content
603      string header_name;
604      string new_value;
605      string delimiter;
606      bool related_files = false;
607      if (same_host && FLAGS_use_xac) {
608        header_name = "X-Associated-Content";
609        char pri_ch = priority + '0';
610        new_value = pri_ch + string("??") + url;
611        delimiter = "||";
612        related_files = true;
613      } else {
614        if (!FLAGS_use_xsub)
615          return;
616        header_name = "X-Subresource";
617        new_value = content_type + "!!" + url;
618        delimiter = "!!";
619      }
620
621      if (fd2->headers->HasNonEmptyHeader(header_name)) {
622        string existing_header =
623            fd2->headers->GetHeader(header_name).as_string();
624        if (existing_header.find(url) != string::npos)
625          return;  // header already recorded
626
627        // Don't let these lists grow too long for low pri stuff.
628        // TODO(mbelshe) We need better algorithms for this.
629        if (existing_header.length() > 256 && priority > 2)
630          return;
631
632        new_value = existing_header + delimiter + new_value;
633      }
634
635      LOG(INFO) << "Recording " << header_name << " for " << new_value;
636      fd2->headers->ReplaceOrAppendHeader(header_name, new_value);
637
638      // Add it to the related files so that it will actually get sent out.
639      if (related_files) {
640        pair<int, string> entry(4, file_url);
641        fd2->related_files.push_back(entry);
642      }
643    } else {
644      LOG(ERROR) << "Failed to update headers:";
645      LOG(ERROR) << "FAIL url: " << url;
646      LOG(ERROR) << "FAIL ref: " << referrer_host_path;
647    }
648  }
649
650  FileData* GetFileData(const string& filename) {
651    Files::iterator fi = files_.end();
652    if (filename.compare(filename.length() - 5, 5, ".html", 5) == 0) {
653      string new_filename(filename.data(), filename.size() - 5);
654      new_filename += ".http";
655      fi = files_.find(new_filename);
656    }
657    if (fi == files_.end())
658      fi = files_.find(filename);
659
660    if (fi == files_.end()) {
661      return NULL;
662    }
663    return &(fi->second);
664  }
665
666  bool AssignFileData(const string& filename, MemCacheIter* mci) {
667    mci->file_data = GetFileData(filename);
668    if (mci->file_data == NULL) {
669      LOG(ERROR) << "Could not find file data for " << filename;
670      return false;
671    }
672    return true;
673  }
674};
675
676////////////////////////////////////////////////////////////////////////////////
677
678class NotifierInterface {
679 public:
680  virtual ~NotifierInterface() {}
681  virtual void Notify() = 0;
682};
683
684////////////////////////////////////////////////////////////////////////////////
685
686class SMInterface {
687 public:
688  virtual size_t ProcessInput(const char* data, size_t len) = 0;
689  virtual bool MessageFullyRead() const = 0;
690  virtual bool Error() const = 0;
691  virtual const char* ErrorAsString() const = 0;
692  virtual void Reset() = 0;
693  virtual void ResetForNewConnection() = 0;
694
695  virtual void PostAcceptHook() = 0;
696
697  virtual void NewStream(uint32 stream_id, uint32 priority,
698                         const string& filename) = 0;
699  virtual void SendEOF(uint32 stream_id) = 0;
700  virtual void SendErrorNotFound(uint32 stream_id) = 0;
701  virtual size_t SendSynStream(uint32 stream_id,
702                              const BalsaHeaders& headers) = 0;
703  virtual size_t SendSynReply(uint32 stream_id,
704                              const BalsaHeaders& headers) = 0;
705  virtual void SendDataFrame(uint32 stream_id, const char* data, int64 len,
706                             uint32 flags, bool compress) = 0;
707  virtual void GetOutput() = 0;
708
709  virtual ~SMInterface() {}
710};
711
712////////////////////////////////////////////////////////////////////////////////
713
714class SMServerConnection;
715typedef SMInterface*(SMInterfaceFactory)(SMServerConnection*);
716
717////////////////////////////////////////////////////////////////////////////////
718
719typedef list<DataFrame> OutputList;
720
721////////////////////////////////////////////////////////////////////////////////
722
723class SMServerConnection;
724
725class SMServerConnectionPoolInterface {
726 public:
727  virtual ~SMServerConnectionPoolInterface() {}
728  // SMServerConnections will use this:
729  virtual void SMServerConnectionDone(SMServerConnection* connection) = 0;
730};
731
732////////////////////////////////////////////////////////////////////////////////
733
734class SMServerConnection: public EpollCallbackInterface,
735                          public NotifierInterface {
736 private:
737  SMServerConnection(SMInterfaceFactory* sm_interface_factory,
738                     MemoryCache* memory_cache,
739                     EpollServer* epoll_server) :
740      fd_(-1),
741      events_(0),
742
743      registered_in_epoll_server_(false),
744      initialized_(false),
745
746      connection_pool_(NULL),
747      epoll_server_(epoll_server),
748
749      read_buffer_(4096*10),
750      memory_cache_(memory_cache),
751      sm_interface_(sm_interface_factory(this)),
752
753      max_bytes_sent_per_dowrite_(4096),
754
755      ssl_(NULL) {}
756
757  int fd_;
758  int events_;
759
760  bool registered_in_epoll_server_;
761  bool initialized_;
762
763  SMServerConnectionPoolInterface* connection_pool_;
764  EpollServer* epoll_server_;
765
766  RingBuffer read_buffer_;
767
768  OutputList output_list_;
769  MemoryCache* memory_cache_;
770  SMInterface* sm_interface_;
771
772  size_t max_bytes_sent_per_dowrite_;
773
774  SSL* ssl_;
775 public:
776  EpollServer* epoll_server() { return epoll_server_; }
777  OutputList* output_list() { return &output_list_; }
778  MemoryCache* memory_cache() { return memory_cache_; }
779  void ReadyToSend() {
780    epoll_server_->SetFDReady(fd_, EPOLLIN | EPOLLOUT);
781  }
782  void EnqueueDataFrame(const DataFrame& df) {
783    output_list_.push_back(df);
784    VLOG(2) << "EnqueueDataFrame. Setting FD ready.";
785    ReadyToSend();
786  }
787
788 public:
789  ~SMServerConnection() {
790    if (initialized()) {
791      Reset();
792    }
793  }
794  static SMServerConnection* NewSMServerConnection(SMInterfaceFactory* smif,
795                                                   MemoryCache* memory_cache,
796                                                   EpollServer* epoll_server) {
797    return new SMServerConnection(smif, memory_cache, epoll_server);
798  }
799
800  bool initialized() const { return initialized_; }
801
802  void InitSMServerConnection(SMServerConnectionPoolInterface* connection_pool,
803                              EpollServer* epoll_server,
804                              int fd) {
805    if (initialized_) {
806      LOG(FATAL) << "Attempted to initialize already initialized server";
807      return;
808    }
809    if (epoll_server_ && registered_in_epoll_server_ && fd_ != -1) {
810      epoll_server_->UnregisterFD(fd_);
811    }
812    if (fd_ != -1) {
813      VLOG(2) << "Closing pre-existing fd";
814      close(fd_);
815      fd_ = -1;
816    }
817
818    fd_ = fd;
819
820    registered_in_epoll_server_ = false;
821    initialized_ = true;
822
823    connection_pool_ = connection_pool;
824    epoll_server_ = epoll_server;
825
826    sm_interface_->Reset();
827    read_buffer_.Clear();
828
829    epoll_server_->RegisterFD(fd_, this, EPOLLIN | EPOLLOUT | EPOLLET);
830
831    if (global_ssl_state) {
832      ssl_ = spdy_new_ssl(global_ssl_state->ssl_ctx);
833      SSL_set_fd(ssl_, fd_);
834      PrintSslError();
835    }
836    sm_interface_->PostAcceptHook();
837  }
838
839  int Send(const char* bytes, int len, int flags) {
840    return send(fd_, bytes, len, flags);
841  }
842
843  // the following are from the EpollCallbackInterface
844  virtual void OnRegistration(EpollServer* eps, int fd, int event_mask) {
845    registered_in_epoll_server_ = true;
846  }
847  virtual void OnModification(int fd, int event_mask) { }
848  virtual void OnEvent(int fd, EpollEvent* event) {
849    events_ |= event->in_events;
850    HandleEvents();
851    if (events_) {
852      event->out_ready_mask = events_;
853      events_ = 0;
854    }
855  }
856  virtual void OnUnregistration(int fd, bool replaced) {
857    registered_in_epoll_server_ = false;
858  }
859  virtual void OnShutdown(EpollServer* eps, int fd) {
860    Cleanup("OnShutdown");
861    return;
862  }
863
864 private:
865  void HandleEvents() {
866    VLOG(1) << "Received: " << EpollServer::EventMaskToString(events_);
867    if (events_ & EPOLLIN) {
868      if (!DoRead())
869        goto handle_close_or_error;
870    }
871
872    if (events_ & EPOLLOUT) {
873      if (!DoWrite())
874        goto handle_close_or_error;
875    }
876
877    if (events_ & (EPOLLHUP | EPOLLERR)) {
878      VLOG(2) << "!!!! Got HUP or ERR";
879      goto handle_close_or_error;
880    }
881    return;
882
883 handle_close_or_error:
884    Cleanup("HandleEvents");
885  }
886
887  bool DoRead() {
888    VLOG(2) << "DoRead()";
889    if (fd_ == -1) {
890      VLOG(2) << "DoRead(): fd_ == -1. Invalid FD. Returning false";
891      return false;
892    }
893    while (!read_buffer_.Full()) {
894      char* bytes;
895      int size;
896      read_buffer_.GetWritablePtr(&bytes, &size);
897      ssize_t bytes_read = 0;
898      if (ssl_) {
899        bytes_read = SSL_read(ssl_, bytes, size);
900        PrintSslError();
901      } else {
902        bytes_read = recv(fd_, bytes, size, MSG_DONTWAIT);
903      }
904      int stored_errno = errno;
905      if (bytes_read == -1) {
906        switch (stored_errno) {
907          case EAGAIN:
908            events_ &= ~EPOLLIN;
909            VLOG(2) << "Got EAGAIN while reading";
910            goto done;
911          case EINTR:
912            VLOG(2) << "Got EINTR while reading";
913            continue;
914          default:
915            VLOG(2) << "While calling recv, got error: " << stored_errno
916              << " " << strerror(stored_errno);
917            goto error_or_close;
918        }
919      } else if (bytes_read > 0) {
920        VLOG(2) << "Read: " << bytes_read << " bytes from fd: " << fd_;
921        read_buffer_.AdvanceWritablePtr(bytes_read);
922        if (!DoConsumeReadData()) {
923          goto error_or_close;
924        }
925        continue;
926      } else {  // bytes_read == 0
927        VLOG(2) << "0 bytes read with recv call.";
928      }
929      goto error_or_close;
930    }
931   done:
932    return true;
933
934   error_or_close:
935    VLOG(2) << "DoRead(): error_or_close. Cleaning up, then returning false";
936    Cleanup("DoRead");
937    return false;
938  }
939
940  bool DoConsumeReadData() {
941    char* bytes;
942    int size;
943    read_buffer_.GetReadablePtr(&bytes, &size);
944    while (size != 0) {
945      size_t bytes_consumed = sm_interface_->ProcessInput(bytes, size);
946      VLOG(2) << "consumed: " << bytes_consumed << " from socket fd: " << fd_;
947      if (bytes_consumed == 0) {
948        break;
949      }
950      read_buffer_.AdvanceReadablePtr(bytes_consumed);
951      if (sm_interface_->MessageFullyRead()) {
952        VLOG(2) << "HandleRequestFullyRead";
953        HandleRequestFullyRead();
954        sm_interface_->Reset();
955        events_ |= EPOLLOUT;
956      } else if (sm_interface_->Error()) {
957        LOG(ERROR) << "Framer error detected: "
958                   << sm_interface_->ErrorAsString();
959        // this causes everything to be closed/cleaned up.
960        events_ |= EPOLLOUT;
961        return false;
962      }
963      read_buffer_.GetReadablePtr(&bytes, &size);
964    }
965    return true;
966  }
967
968  void WriteResponse() {
969    // this happens asynchronously from separate threads
970    // feeding files into the output buffer.
971  }
972
973  void HandleRequestFullyRead() {
974  }
975
976  void Notify() {
977  }
978
979  bool DoWrite() {
980    size_t bytes_sent = 0;
981    int flags = MSG_NOSIGNAL | MSG_DONTWAIT;
982    if (fd_ == -1) {
983      VLOG(2) << "DoWrite: fd == -1. Returning false.";
984      return false;
985    }
986    if (output_list_.empty()) {
987      sm_interface_->GetOutput();
988      if (output_list_.empty())
989        events_ &= ~EPOLLOUT;
990    }
991    while (!output_list_.empty()) {
992      if (bytes_sent >= max_bytes_sent_per_dowrite_) {
993        events_ |= EPOLLOUT;
994        break;
995      }
996      if (output_list_.size() < 2) {
997        sm_interface_->GetOutput();
998      }
999      DataFrame& data_frame = output_list_.front();
1000      const char*  bytes = data_frame.data;
1001      int size = data_frame.size;
1002      bytes += data_frame.index;
1003      size -= data_frame.index;
1004      DCHECK_GE(size, 0);
1005      if (size <= 0) {
1006        data_frame.MaybeDelete();
1007        output_list_.pop_front();
1008        continue;
1009      }
1010
1011      flags = MSG_NOSIGNAL | MSG_DONTWAIT;
1012      if (output_list_.size() > 1) {
1013        flags |= MSG_MORE;
1014      }
1015      ssize_t bytes_written = 0;
1016      if (ssl_) {
1017        bytes_written = SSL_write(ssl_, bytes, size);
1018        PrintSslError();
1019      } else {
1020        bytes_written = send(fd_, bytes, size, flags);
1021      }
1022      int stored_errno = errno;
1023      if (bytes_written == -1) {
1024        switch (stored_errno) {
1025          case EAGAIN:
1026            events_ &= ~EPOLLOUT;
1027            VLOG(2) << " Got EAGAIN while writing";
1028            goto done;
1029          case EINTR:
1030            VLOG(2) << " Got EINTR while writing";
1031            continue;
1032          default:
1033            VLOG(2) << "While calling send, got error: " << stored_errno
1034              << " " << strerror(stored_errno);
1035            goto error_or_close;
1036        }
1037      } else if (bytes_written > 0) {
1038        VLOG(1) << "Wrote: " << bytes_written  << " bytes to socket fd: "
1039          << fd_;
1040        data_frame.index += bytes_written;
1041        bytes_sent += bytes_written;
1042        continue;
1043      }
1044      VLOG(2) << "0 bytes written to socket " << fd_ << " with send call.";
1045      goto error_or_close;
1046    }
1047   done:
1048    return true;
1049
1050   error_or_close:
1051    VLOG(2) << "DoWrite: error_or_close. Returning false after cleaning up";
1052    Cleanup("DoWrite");
1053    return false;
1054  }
1055
1056  friend ostream& operator<<(ostream& os, const SMServerConnection& c) {
1057    os << &c << "\n";
1058    return os;
1059  }
1060
1061  void Reset() {
1062    VLOG(2) << "Resetting";
1063    if (ssl_) {
1064      SSL_shutdown(ssl_);
1065      PrintSslError();
1066      SSL_free(ssl_);
1067      PrintSslError();
1068    }
1069    if (registered_in_epoll_server_) {
1070      epoll_server_->UnregisterFD(fd_);
1071      registered_in_epoll_server_ = false;
1072    }
1073    if (fd_ >= 0) {
1074      VLOG(2) << "Closing connection";
1075      close(fd_);
1076      fd_ = -1;
1077    }
1078    sm_interface_->ResetForNewConnection();
1079    read_buffer_.Clear();
1080    initialized_ = false;
1081    events_ = 0;
1082    output_list_.clear();
1083  }
1084
1085  void Cleanup(const char* cleanup) {
1086    VLOG(2) << "Cleaning up: " << cleanup;
1087    if (!initialized_) {
1088      return;
1089    }
1090    Reset();
1091    connection_pool_->SMServerConnectionDone(this);
1092  }
1093};
1094
1095////////////////////////////////////////////////////////////////////////////////
1096
1097class OutputOrdering {
1098 public:
1099  typedef list<MemCacheIter> PriorityRing;
1100
1101  typedef map<uint32, PriorityRing> PriorityMap;
1102
1103  struct PriorityMapPointer {
1104    PriorityMapPointer(): ring(NULL), alarm_enabled(false) {}
1105    PriorityRing* ring;
1106    PriorityRing::iterator it;
1107    bool alarm_enabled;
1108    EpollServer::AlarmRegToken alarm_token;
1109  };
1110  typedef map<uint32, PriorityMapPointer> StreamIdToPriorityMap;
1111
1112  StreamIdToPriorityMap stream_ids_;
1113  PriorityMap priority_map_;
1114  PriorityRing first_data_senders_;
1115  uint32 first_data_senders_threshold_;  // when you've passed this, you're no
1116                                         // longer a first_data_sender...
1117  SMServerConnection* connection_;
1118  EpollServer* epoll_server_;
1119
1120  explicit OutputOrdering(SMServerConnection* connection) :
1121      first_data_senders_threshold_(kInitialDataSendersThreshold),
1122      connection_(connection),
1123      epoll_server_(connection->epoll_server()) {
1124  }
1125
1126  void Reset() {
1127    while (!stream_ids_.empty()) {
1128      StreamIdToPriorityMap::iterator sitpmi = stream_ids_.begin();
1129      PriorityMapPointer& pmp = sitpmi->second;
1130      if (pmp.alarm_enabled) {
1131        epoll_server_->UnregisterAlarm(pmp.alarm_token);
1132      }
1133      stream_ids_.erase(sitpmi);
1134    }
1135    priority_map_.clear();
1136    first_data_senders_.clear();
1137  }
1138
1139  bool ExistsInPriorityMaps(uint32 stream_id) {
1140    StreamIdToPriorityMap::iterator sitpmi = stream_ids_.find(stream_id);
1141    return sitpmi != stream_ids_.end();
1142  }
1143
1144  struct BeginOutputtingAlarm : public EpollAlarmCallbackInterface {
1145   public:
1146    BeginOutputtingAlarm(OutputOrdering* oo,
1147                         OutputOrdering::PriorityMapPointer* pmp,
1148                         const MemCacheIter& mci) :
1149        output_ordering_(oo), pmp_(pmp), mci_(mci), epoll_server_(NULL) {}
1150
1151    int64 OnAlarm() {
1152      OnUnregistration();
1153      output_ordering_->MoveToActive(pmp_, mci_);
1154      VLOG(1) << "ON ALARM! Should now start to output...";
1155      delete this;
1156      return 0;
1157    }
1158    void OnRegistration(const EpollServer::AlarmRegToken& tok,
1159                        EpollServer* eps) {
1160      epoll_server_ = eps;
1161      pmp_->alarm_token = tok;
1162      pmp_->alarm_enabled = true;
1163    }
1164    void OnUnregistration() {
1165      pmp_->alarm_enabled = false;
1166    }
1167    void OnShutdown(EpollServer* eps) {
1168      OnUnregistration();
1169    }
1170    ~BeginOutputtingAlarm() {
1171      if (epoll_server_ && pmp_->alarm_enabled)
1172        epoll_server_->UnregisterAlarm(pmp_->alarm_token);
1173    }
1174   private:
1175    OutputOrdering* output_ordering_;
1176    OutputOrdering::PriorityMapPointer* pmp_;
1177    MemCacheIter mci_;
1178    EpollServer* epoll_server_;
1179  };
1180
1181  void MoveToActive(PriorityMapPointer* pmp, MemCacheIter mci) {
1182    VLOG(1) <<"Moving to active!";
1183    first_data_senders_.push_back(mci);
1184    pmp->ring = &first_data_senders_;
1185    pmp->it = first_data_senders_.end();
1186    --pmp->it;
1187    connection_->ReadyToSend();
1188  }
1189
1190  void AddToOutputOrder(const MemCacheIter& mci) {
1191    if (ExistsInPriorityMaps(mci.stream_id))
1192      LOG(FATAL) << "OOps, already was inserted here?!";
1193
1194    double think_time_in_s = FLAGS_server_think_time_in_s;
1195    string x_server_latency =
1196      mci.file_data->headers->GetHeader("X-Server-Latency").as_string();
1197    if (x_server_latency.size() != 0) {
1198      char* endp;
1199      double tmp_think_time_in_s = strtod(x_server_latency.c_str(), &endp);
1200      if (endp != x_server_latency.c_str() + x_server_latency.size()) {
1201        LOG(ERROR) << "Unable to understand X-Server-Latency of: "
1202          << x_server_latency << " for resource: " << mci.file_data->filename;
1203      } else {
1204        think_time_in_s = tmp_think_time_in_s;
1205      }
1206    }
1207    StreamIdToPriorityMap::iterator sitpmi;
1208    sitpmi = stream_ids_.insert(
1209        pair<uint32, PriorityMapPointer>(mci.stream_id,
1210                                         PriorityMapPointer())).first;
1211    PriorityMapPointer& pmp = sitpmi->second;
1212
1213    BeginOutputtingAlarm* boa = new BeginOutputtingAlarm(this, &pmp, mci);
1214    VLOG(2) << "Server think time: " << think_time_in_s;
1215    epoll_server_->RegisterAlarmApproximateDelta(
1216        think_time_in_s * 1000000, boa);
1217  }
1218
1219  void SpliceToPriorityRing(PriorityRing::iterator pri) {
1220    MemCacheIter& mci = *pri;
1221    PriorityMap::iterator pmi = priority_map_.find(mci.priority);
1222    if (pmi == priority_map_.end()) {
1223      pmi = priority_map_.insert(
1224          pair<uint32, PriorityRing>(mci.priority, PriorityRing())).first;
1225    }
1226
1227    pmi->second.splice(pmi->second.end(),
1228                       first_data_senders_,
1229                       pri);
1230    StreamIdToPriorityMap::iterator sitpmi = stream_ids_.find(mci.stream_id);
1231    sitpmi->second.ring = &(pmi->second);
1232  }
1233
1234  MemCacheIter* GetIter() {
1235    while (!first_data_senders_.empty()) {
1236      MemCacheIter& mci = first_data_senders_.front();
1237      if (mci.bytes_sent >= first_data_senders_threshold_) {
1238        SpliceToPriorityRing(first_data_senders_.begin());
1239      } else {
1240        first_data_senders_.splice(first_data_senders_.end(),
1241                                  first_data_senders_,
1242                                  first_data_senders_.begin());
1243        mci.max_segment_size = kInitialDataSendersThreshold;
1244        return &mci;
1245      }
1246    }
1247    while (!priority_map_.empty()) {
1248      PriorityRing& first_ring = priority_map_.begin()->second;
1249      if (first_ring.empty()) {
1250        priority_map_.erase(priority_map_.begin());
1251        continue;
1252      }
1253      MemCacheIter& mci = first_ring.front();
1254      first_ring.splice(first_ring.end(),
1255                        first_ring,
1256                        first_ring.begin());
1257      mci.max_segment_size = kNormalSegmentSize;
1258      return &mci;
1259    }
1260    return NULL;
1261  }
1262
1263  void RemoveStreamId(uint32 stream_id) {
1264    StreamIdToPriorityMap::iterator sitpmi = stream_ids_.find(stream_id);
1265    if (sitpmi == stream_ids_.end())
1266      return;
1267    PriorityMapPointer& pmp = sitpmi->second;
1268    if (pmp.alarm_enabled) {
1269      epoll_server_->UnregisterAlarm(pmp.alarm_token);
1270    } else {
1271      pmp.ring->erase(pmp.it);
1272    }
1273
1274    stream_ids_.erase(sitpmi);
1275  }
1276};
1277
1278////////////////////////////////////////////////////////////////////////////////
1279
1280class SpdySM : public SpdyFramerVisitorInterface, public SMInterface {
1281 private:
1282  uint64 seq_num_;
1283  SpdyFramer* framer_;
1284
1285  SMServerConnection* connection_;
1286  OutputList* output_list_;
1287  OutputOrdering output_ordering_;
1288  MemoryCache* memory_cache_;
1289  uint32 next_outgoing_stream_id_;
1290 public:
1291  explicit SpdySM(SMServerConnection* connection) :
1292      seq_num_(0),
1293      framer_(new SpdyFramer),
1294      connection_(connection),
1295      output_list_(connection->output_list()),
1296      output_ordering_(connection),
1297      memory_cache_(connection->memory_cache()),
1298      next_outgoing_stream_id_(2) {
1299    framer_->set_visitor(this);
1300  }
1301 private:
1302  virtual void OnError(SpdyFramer* framer) {
1303    /* do nothing with this right now */
1304  }
1305
1306  virtual void OnControl(const SpdyControlFrame* frame) {
1307    SpdyHeaderBlock headers;
1308    bool parsed_headers = false;
1309    switch (frame->type()) {
1310      case SYN_STREAM:
1311        {
1312        const SpdySynStreamControlFrame* syn_stream =
1313            reinterpret_cast<const SpdySynStreamControlFrame*>(frame);
1314        parsed_headers = framer_->ParseHeaderBlock(frame, &headers);
1315        VLOG(2) << "OnSyn(" << syn_stream->stream_id() << ")";
1316        VLOG(2) << "headers parsed?: " << (parsed_headers? "yes": "no");
1317        if (parsed_headers) {
1318          VLOG(2) << "# headers: " << headers.size();
1319        }
1320        for (SpdyHeaderBlock::iterator i = headers.begin();
1321             i != headers.end();
1322             ++i) {
1323          VLOG(2) << i->first << ": " << i->second;
1324        }
1325
1326        SpdyHeaderBlock::iterator method = headers.find("method");
1327        SpdyHeaderBlock::iterator url = headers.find("url");
1328        if (url == headers.end() || method == headers.end()) {
1329          VLOG(2) << "didn't find method or url or method. Not creating stream";
1330          break;
1331        }
1332
1333        SpdyHeaderBlock::iterator referer = headers.find("referer");
1334        if (referer != headers.end() && method->second == "GET") {
1335          memory_cache_->UpdateHeaders(referer->second, url->second);
1336        }
1337        string uri = UrlUtilities::GetUrlPath(url->second);
1338        string host = UrlUtilities::GetUrlHost(url->second);
1339
1340        string filename = EncodeURL(uri, host, method->second);
1341        NewStream(syn_stream->stream_id(),
1342                  reinterpret_cast<const SpdySynStreamControlFrame*>(frame)->
1343                    priority(),
1344                  filename);
1345        }
1346        break;
1347
1348      case SYN_REPLY:
1349        parsed_headers = framer_->ParseHeaderBlock(frame, &headers);
1350        VLOG(2) << "OnSynReply("
1351                << reinterpret_cast<const SpdySynReplyControlFrame*>(
1352                    frame)->stream_id() << ")";
1353        break;
1354      case RST_STREAM:
1355        {
1356        const SpdyRstStreamControlFrame* rst_stream =
1357            reinterpret_cast<const SpdyRstStreamControlFrame*>(frame);
1358        VLOG(2) << "OnRst(" << rst_stream->stream_id() << ")";
1359        output_ordering_.RemoveStreamId(rst_stream ->stream_id());
1360        }
1361        break;
1362
1363      default:
1364        LOG(DFATAL) << "Unknown control frame type";
1365    }
1366  }
1367  virtual void OnStreamFrameData(
1368    SpdyStreamId stream_id,
1369    const char* data, size_t len) {
1370    VLOG(2) << "StreamData(" << stream_id << ", [" << len << "])";
1371    /* do nothing with this right now */
1372  }
1373  virtual void OnLameDuck() {
1374    /* do nothing with this right now */
1375  }
1376
1377 public:
1378  ~SpdySM() {
1379    Reset();
1380  }
1381  size_t ProcessInput(const char* data, size_t len) {
1382    return framer_->ProcessInput(data, len);
1383  }
1384
1385  bool MessageFullyRead() const {
1386    return framer_->MessageFullyRead();
1387  }
1388
1389  bool Error() const {
1390    return framer_->HasError();
1391  }
1392
1393  const char* ErrorAsString() const {
1394    return SpdyFramer::ErrorCodeToString(framer_->error_code());
1395  }
1396
1397  void Reset() {}
1398  void ResetForNewConnection() {
1399    // seq_num is not cleared, intentionally.
1400    delete framer_;
1401    framer_ = new SpdyFramer;
1402    framer_->set_visitor(this);
1403    output_ordering_.Reset();
1404    next_outgoing_stream_id_ = 2;
1405  }
1406
1407  // Send a couple of NOOP packets to force opening of cwnd.
1408  void PostAcceptHook() {
1409    if (!FLAGS_use_cwnd_opener)
1410      return;
1411
1412    // We send 2 because that is the initial cwnd, and also because
1413    // we have to in order to get an ACK back from the client due to
1414    // delayed ACK.
1415    const int kPkts = 2;
1416
1417    LOG(ERROR) << "Sending NOP FRAMES";
1418
1419    scoped_ptr<SpdyControlFrame> frame(SpdyFramer::CreateNopFrame());
1420    for (int i = 0; i < kPkts; ++i) {
1421      char* bytes = frame->data();
1422      size_t size = SpdyFrame::size();
1423      ssize_t bytes_written = connection_->Send(bytes, size, MSG_DONTWAIT);
1424      if (static_cast<size_t>(bytes_written) != size) {
1425        LOG(ERROR) << "Trouble sending Nop packet! (" << errno << ")";
1426        if (errno == EAGAIN)
1427          break;
1428      }
1429    }
1430  }
1431
1432  void AddAssociatedContent(FileData* file_data) {
1433    for (unsigned int i = 0; i < file_data->related_files.size(); ++i) {
1434      pair<int, string>& related_file = file_data->related_files[i];
1435      MemCacheIter mci;
1436      string filename  = "GET_";
1437      filename += related_file.second;
1438      if (!memory_cache_->AssignFileData(filename, &mci)) {
1439        VLOG(1) << "Unable to find associated content for: " << filename;
1440        continue;
1441      }
1442      VLOG(1) << "Adding associated content: " << filename;
1443      mci.stream_id = next_outgoing_stream_id_;
1444      next_outgoing_stream_id_ += 2;
1445      mci.priority =  related_file.first;
1446      AddToOutputOrder(mci);
1447    }
1448  }
1449
1450  void NewStream(uint32 stream_id, uint32 priority, const string& filename) {
1451    MemCacheIter mci;
1452    mci.stream_id = stream_id;
1453    mci.priority = priority;
1454    if (!memory_cache_->AssignFileData(filename, &mci)) {
1455      // error creating new stream.
1456      VLOG(2) << "Sending ErrorNotFound";
1457      SendErrorNotFound(stream_id);
1458    } else {
1459      AddToOutputOrder(mci);
1460      if (FLAGS_use_xac) {
1461        AddAssociatedContent(mci.file_data);
1462      }
1463    }
1464  }
1465
1466  void AddToOutputOrder(const MemCacheIter& mci) {
1467    output_ordering_.AddToOutputOrder(mci);
1468  }
1469
1470  void SendEOF(uint32 stream_id) {
1471    SendEOFImpl(stream_id);
1472  }
1473
1474  void SendErrorNotFound(uint32 stream_id) {
1475    SendErrorNotFoundImpl(stream_id);
1476  }
1477
1478  void SendOKResponse(uint32 stream_id, string* output) {
1479    SendOKResponseImpl(stream_id, output);
1480  }
1481
1482  size_t SendSynStream(uint32 stream_id, const BalsaHeaders& headers) {
1483    return SendSynStreamImpl(stream_id, headers);
1484  }
1485
1486  size_t SendSynReply(uint32 stream_id, const BalsaHeaders& headers) {
1487    return SendSynReplyImpl(stream_id, headers);
1488  }
1489
1490  void SendDataFrame(uint32 stream_id, const char* data, int64 len,
1491                     uint32 flags, bool compress) {
1492    SpdyDataFlags spdy_flags = static_cast<SpdyDataFlags>(flags);
1493    SendDataFrameImpl(stream_id, data, len, spdy_flags, compress);
1494  }
1495
1496  SpdyFramer* spdy_framer() { return framer_; }
1497
1498 private:
1499  void SendEOFImpl(uint32 stream_id) {
1500    SendDataFrame(stream_id, NULL, 0, DATA_FLAG_FIN, false);
1501    VLOG(2) << "Sending EOF: " << stream_id;
1502    KillStream(stream_id);
1503  }
1504
1505  void SendErrorNotFoundImpl(uint32 stream_id) {
1506    BalsaHeaders my_headers;
1507    my_headers.SetFirstlineFromStringPieces("HTTP/1.1", "404", "Not Found");
1508    SendSynReplyImpl(stream_id, my_headers);
1509    SendDataFrame(stream_id, "wtf?", 4, DATA_FLAG_FIN, false);
1510    output_ordering_.RemoveStreamId(stream_id);
1511  }
1512
1513  void SendOKResponseImpl(uint32 stream_id, string* output) {
1514    BalsaHeaders my_headers;
1515    my_headers.SetFirstlineFromStringPieces("HTTP/1.1", "200", "OK");
1516    SendSynReplyImpl(stream_id, my_headers);
1517    SendDataFrame(
1518        stream_id, output->c_str(), output->size(), DATA_FLAG_FIN, false);
1519    output_ordering_.RemoveStreamId(stream_id);
1520  }
1521
1522  void KillStream(uint32 stream_id) {
1523    output_ordering_.RemoveStreamId(stream_id);
1524  }
1525
1526  void CopyHeaders(SpdyHeaderBlock& dest, const BalsaHeaders& headers) {
1527    for (BalsaHeaders::const_header_lines_iterator hi =
1528         headers.header_lines_begin();
1529         hi != headers.header_lines_end();
1530         ++hi) {
1531      SpdyHeaderBlock::iterator fhi = dest.find(hi->first.as_string());
1532      if (fhi == dest.end()) {
1533        dest[hi->first.as_string()] = hi->second.as_string();
1534      } else {
1535        dest[hi->first.as_string()] = (
1536            string(fhi->second.data(), fhi->second.size()) + "," +
1537            string(hi->second.data(), hi->second.size()));
1538      }
1539    }
1540
1541    // These headers have no value
1542    dest.erase("X-Associated-Content");  // TODO(mbelshe): case-sensitive
1543    dest.erase("X-Original-Url");  // TODO(mbelshe): case-sensitive
1544  }
1545
1546  size_t SendSynStreamImpl(uint32 stream_id, const BalsaHeaders& headers) {
1547    SpdyHeaderBlock block;
1548    block["method"] = headers.request_method().as_string();
1549    if (!headers.HasHeader("status"))
1550      block["status"] = headers.response_code().as_string();
1551    if (!headers.HasHeader("version"))
1552      block["version"] =headers.response_version().as_string();
1553    if (headers.HasHeader("X-Original-Url")) {
1554      string original_url = headers.GetHeader("X-Original-Url").as_string();
1555      block["path"] = UrlUtilities::GetUrlPath(original_url);
1556    } else {
1557      block["path"] = headers.request_uri().as_string();
1558    }
1559    CopyHeaders(block, headers);
1560
1561    SpdySynStreamControlFrame* fsrcf =
1562      framer_->CreateSynStream(stream_id, 0, 0, CONTROL_FLAG_NONE, true,
1563                               &block);
1564    DataFrame df;
1565    df.size = fsrcf->length() + SpdyFrame::size();
1566    size_t df_size = df.size;
1567    df.data = fsrcf->data();
1568    df.delete_when_done = true;
1569    EnqueueDataFrame(df);
1570
1571    VLOG(2) << "Sending SynStreamheader " << stream_id;
1572    return df_size;
1573  }
1574
1575  size_t SendSynReplyImpl(uint32 stream_id, const BalsaHeaders& headers) {
1576    SpdyHeaderBlock block;
1577    CopyHeaders(block, headers);
1578    block["status"] = headers.response_code().as_string() + " " +
1579                      headers.response_reason_phrase().as_string();
1580    block["version"] = headers.response_version().as_string();
1581
1582    SpdySynReplyControlFrame* fsrcf =
1583      framer_->CreateSynReply(stream_id, CONTROL_FLAG_NONE, true, &block);
1584    DataFrame df;
1585    df.size = fsrcf->length() + SpdyFrame::size();
1586    size_t df_size = df.size;
1587    df.data = fsrcf->data();
1588    df.delete_when_done = true;
1589    EnqueueDataFrame(df);
1590
1591    VLOG(2) << "Sending SynReplyheader " << stream_id;
1592    return df_size;
1593  }
1594
1595  void SendDataFrameImpl(uint32 stream_id, const char* data, int64 len,
1596                         SpdyDataFlags flags, bool compress) {
1597    // Force compression off if disabled via command line.
1598    if (!FLAGS_use_compression)
1599      flags = static_cast<SpdyDataFlags>(flags & ~DATA_FLAG_COMPRESSED);
1600
1601    // TODO(mbelshe):  We can't compress here - before going into the
1602    //                 priority queue.  Compression needs to be done
1603    //                 with late binding.
1604    SpdyDataFrame* fdf = framer_->CreateDataFrame(stream_id, data, len,
1605                                                  flags);
1606    DataFrame df;
1607    df.size = fdf->length() + SpdyFrame::size();
1608    df.data = fdf->data();
1609    df.delete_when_done = true;
1610    EnqueueDataFrame(df);
1611
1612    VLOG(2) << "Sending data frame" << stream_id << " [" << len << "]"
1613            << " shrunk to " << fdf->length();
1614  }
1615
1616  void EnqueueDataFrame(const DataFrame& df) {
1617    connection_->EnqueueDataFrame(df);
1618  }
1619
1620  void GetOutput() {
1621    while (output_list_->size() < 2) {
1622      MemCacheIter* mci = output_ordering_.GetIter();
1623      if (mci == NULL) {
1624        VLOG(2) << "GetOutput: nothing to output!?";
1625        return;
1626      }
1627      if (!mci->transformed_header) {
1628        mci->transformed_header = true;
1629        VLOG(2) << "GetOutput transformed header stream_id: ["
1630          << mci->stream_id << "]";
1631        if ((mci->stream_id % 2) == 0) {
1632          // this is a server initiated stream.
1633          // Ideally, we'd do a 'syn-push' here, instead of a syn-reply.
1634          BalsaHeaders headers;
1635          headers.CopyFrom(*(mci->file_data->headers));
1636          headers.ReplaceOrAppendHeader("status", "200");
1637          headers.ReplaceOrAppendHeader("version", "http/1.1");
1638          headers.SetRequestFirstlineFromStringPieces("PUSH",
1639                                                      mci->file_data->filename,
1640                                                      "");
1641          mci->bytes_sent = SendSynStream(mci->stream_id, headers);
1642        } else {
1643          BalsaHeaders headers;
1644          headers.CopyFrom(*(mci->file_data->headers));
1645          mci->bytes_sent = SendSynReply(mci->stream_id, headers);
1646        }
1647        return;
1648      }
1649      if (mci->body_bytes_consumed >= mci->file_data->body.size()) {
1650        VLOG(2) << "GetOutput remove_stream_id: [" << mci->stream_id << "]";
1651        SendEOF(mci->stream_id);
1652        return;
1653      }
1654      size_t num_to_write =
1655        mci->file_data->body.size() - mci->body_bytes_consumed;
1656      if (num_to_write > mci->max_segment_size)
1657        num_to_write = mci->max_segment_size;
1658
1659      bool should_compress = false;
1660      if (!mci->file_data->headers->HasHeader("content-encoding")) {
1661        if (mci->file_data->headers->HasHeader("content-type")) {
1662          string content_type =
1663              mci->file_data->headers->GetHeader("content-type").as_string();
1664          if (content_type.find("image") == content_type.npos)
1665            should_compress = true;
1666        }
1667      }
1668
1669      SendDataFrame(mci->stream_id,
1670                    mci->file_data->body.data() + mci->body_bytes_consumed,
1671                    num_to_write, 0, should_compress);
1672      VLOG(2) << "GetOutput SendDataFrame[" << mci->stream_id
1673        << "]: " << num_to_write;
1674      mci->body_bytes_consumed += num_to_write;
1675      mci->bytes_sent += num_to_write;
1676    }
1677  }
1678};
1679
1680////////////////////////////////////////////////////////////////////////////////
1681
1682class HTTPSM : public BalsaVisitorInterface, public SMInterface {
1683 private:
1684  uint64 seq_num_;
1685  BalsaFrame* framer_;
1686  BalsaHeaders headers_;
1687  uint32 stream_id_;
1688
1689  SMServerConnection* connection_;
1690  OutputList* output_list_;
1691  OutputOrdering output_ordering_;
1692  MemoryCache* memory_cache_;
1693 public:
1694  explicit HTTPSM(SMServerConnection* connection) :
1695      seq_num_(0),
1696      framer_(new BalsaFrame),
1697      stream_id_(1),
1698      connection_(connection),
1699      output_list_(connection->output_list()),
1700      output_ordering_(connection),
1701      memory_cache_(connection->memory_cache()) {
1702    framer_->set_balsa_visitor(this);
1703    framer_->set_balsa_headers(&headers_);
1704  }
1705 private:
1706  typedef map<string, uint32> ClientTokenMap;
1707 private:
1708    virtual void ProcessBodyInput(const char *input, size_t size) {
1709    }
1710    virtual void ProcessBodyData(const char *input, size_t size) {
1711      // ignoring this.
1712    }
1713    virtual void ProcessHeaderInput(const char *input, size_t size) {
1714    }
1715    virtual void ProcessTrailerInput(const char *input, size_t size) {}
1716    virtual void ProcessHeaders(const BalsaHeaders& headers) {
1717      VLOG(2) << "Got new request!";
1718      string host = UrlUtilities::GetUrlHost(
1719          headers.GetHeader("Host").as_string());
1720      string method = headers.request_method().as_string();
1721      string filename = EncodeURL(headers.request_uri().as_string(), host,
1722          method);
1723      NewStream(stream_id_, 0, filename);
1724      stream_id_ += 2;
1725    }
1726    virtual void ProcessRequestFirstLine(const char* line_input,
1727                                         size_t line_length,
1728                                         const char* method_input,
1729                                         size_t method_length,
1730                                         const char* request_uri_input,
1731                                         size_t request_uri_length,
1732                                         const char* version_input,
1733                                         size_t version_length) {}
1734    virtual void ProcessResponseFirstLine(const char *line_input,
1735                                          size_t line_length,
1736                                          const char *version_input,
1737                                          size_t version_length,
1738                                          const char *status_input,
1739                                          size_t status_length,
1740                                          const char *reason_input,
1741                                          size_t reason_length) {}
1742    virtual void ProcessChunkLength(size_t chunk_length) {}
1743    virtual void ProcessChunkExtensions(const char *input, size_t size) {}
1744    virtual void HeaderDone() {}
1745    virtual void MessageDone() {
1746      VLOG(2) << "MessageDone!";
1747    }
1748    virtual void HandleHeaderError(BalsaFrame* framer) {
1749      HandleError();
1750    }
1751    virtual void HandleHeaderWarning(BalsaFrame* framer) {}
1752    virtual void HandleChunkingError(BalsaFrame* framer) {
1753      HandleError();
1754    }
1755    virtual void HandleBodyError(BalsaFrame* framer) {
1756      HandleError();
1757    }
1758
1759    void HandleError() {
1760      VLOG(2) << "Error detected";
1761    }
1762
1763 public:
1764  ~HTTPSM() {
1765    Reset();
1766  }
1767  size_t ProcessInput(const char* data, size_t len) {
1768    return framer_->ProcessInput(data, len);
1769  }
1770
1771  bool MessageFullyRead() const {
1772    return framer_->MessageFullyRead();
1773  }
1774
1775  bool Error() const {
1776    return framer_->Error();
1777  }
1778
1779  const char* ErrorAsString() const {
1780    return BalsaFrameEnums::ErrorCodeToString(framer_->ErrorCode());
1781  }
1782
1783  void Reset() {
1784    framer_->Reset();
1785  }
1786
1787  void ResetForNewConnection() {
1788    seq_num_ = 0;
1789    output_ordering_.Reset();
1790    framer_->Reset();
1791  }
1792
1793  void PostAcceptHook() {
1794  }
1795
1796  void NewStream(uint32 stream_id, uint32 priority, const string& filename) {
1797    MemCacheIter mci;
1798    mci.stream_id = stream_id;
1799    mci.priority = priority;
1800    if (!memory_cache_->AssignFileData(filename, &mci)) {
1801      SendErrorNotFound(stream_id);
1802    } else {
1803      AddToOutputOrder(mci);
1804    }
1805  }
1806
1807  void AddToOutputOrder(const MemCacheIter& mci) {
1808    output_ordering_.AddToOutputOrder(mci);
1809  }
1810
1811  void SendEOF(uint32 stream_id) {
1812    SendEOFImpl(stream_id);
1813  }
1814
1815  void SendErrorNotFound(uint32 stream_id) {
1816    SendErrorNotFoundImpl(stream_id);
1817  }
1818
1819  void SendOKResponse(uint32 stream_id, string* output) {
1820    SendOKResponseImpl(stream_id, output);
1821  }
1822
1823  size_t SendSynStream(uint32 stream_id, const BalsaHeaders& headers) {
1824    return 0;
1825  }
1826
1827  size_t SendSynReply(uint32 stream_id, const BalsaHeaders& headers) {
1828    return SendSynReplyImpl(stream_id, headers);
1829  }
1830
1831  void SendDataFrame(uint32 stream_id, const char* data, int64 len,
1832                     uint32 flags, bool compress) {
1833    SendDataFrameImpl(stream_id, data, len, flags, compress);
1834  }
1835
1836  BalsaFrame* spdy_framer() { return framer_; }
1837
1838 private:
1839  void SendEOFImpl(uint32 stream_id) {
1840    DataFrame df;
1841    df.data = "0\r\n\r\n";
1842    df.size = 5;
1843    df.delete_when_done = false;
1844    EnqueueDataFrame(df);
1845  }
1846
1847  void SendErrorNotFoundImpl(uint32 stream_id) {
1848    BalsaHeaders my_headers;
1849    my_headers.SetFirstlineFromStringPieces("HTTP/1.1", "404", "Not Found");
1850    my_headers.RemoveAllOfHeader("content-length");
1851    my_headers.AppendHeader("transfer-encoding", "chunked");
1852    SendSynReplyImpl(stream_id, my_headers);
1853    SendDataFrame(stream_id, "wtf?", 4, 0, false);
1854    SendEOFImpl(stream_id);
1855    output_ordering_.RemoveStreamId(stream_id);
1856  }
1857
1858  void SendOKResponseImpl(uint32 stream_id, string* output) {
1859    BalsaHeaders my_headers;
1860    my_headers.SetFirstlineFromStringPieces("HTTP/1.1", "200", "OK");
1861    my_headers.RemoveAllOfHeader("content-length");
1862    my_headers.AppendHeader("transfer-encoding", "chunked");
1863    SendSynReplyImpl(stream_id, my_headers);
1864    SendDataFrame(stream_id, output->c_str(), output->size(), 0, false);
1865    SendEOFImpl(stream_id);
1866    output_ordering_.RemoveStreamId(stream_id);
1867  }
1868
1869  size_t SendSynReplyImpl(uint32 stream_id, const BalsaHeaders& headers) {
1870    SimpleBuffer sb;
1871    headers.WriteHeaderAndEndingToBuffer(&sb);
1872    DataFrame df;
1873    df.size = sb.ReadableBytes();
1874    char* buffer = new char[df.size];
1875    df.data = buffer;
1876    df.delete_when_done = true;
1877    sb.Read(buffer, df.size);
1878    VLOG(2) << "******************Sending HTTP Reply header " << stream_id;
1879    size_t df_size = df.size;
1880    EnqueueDataFrame(df);
1881    return df_size;
1882  }
1883
1884  size_t SendSynStreamImpl(uint32 stream_id, const BalsaHeaders& headers) {
1885    SimpleBuffer sb;
1886    headers.WriteHeaderAndEndingToBuffer(&sb);
1887    DataFrame df;
1888    df.size = sb.ReadableBytes();
1889    char* buffer = new char[df.size];
1890    df.data = buffer;
1891    df.delete_when_done = true;
1892    sb.Read(buffer, df.size);
1893    VLOG(2) << "******************Sending HTTP Reply header " << stream_id;
1894    size_t df_size = df.size;
1895    EnqueueDataFrame(df);
1896    return df_size;
1897  }
1898
1899  void SendDataFrameImpl(uint32 stream_id, const char* data, int64 len,
1900                         uint32 flags, bool compress) {
1901    char chunk_buf[128];
1902    snprintf(chunk_buf, sizeof(chunk_buf), "%x\r\n", (unsigned int)len);
1903    string chunk_description(chunk_buf);
1904    DataFrame df;
1905    df.size = chunk_description.size() + len + 2;
1906    char* buffer = new char[df.size];
1907    df.data = buffer;
1908    df.delete_when_done = true;
1909    memcpy(buffer, chunk_description.data(), chunk_description.size());
1910    memcpy(buffer + chunk_description.size(), data, len);
1911    memcpy(buffer + chunk_description.size() + len, "\r\n", 2);
1912    EnqueueDataFrame(df);
1913  }
1914
1915  void EnqueueDataFrame(const DataFrame& df) {
1916    connection_->EnqueueDataFrame(df);
1917  }
1918
1919  void GetOutput() {
1920    MemCacheIter* mci = output_ordering_.GetIter();
1921    if (mci == NULL) {
1922      VLOG(2) << "GetOutput: nothing to output!?";
1923      return;
1924    }
1925    if (!mci->transformed_header) {
1926      mci->bytes_sent = SendSynReply(mci->stream_id,
1927                                     *(mci->file_data->headers));
1928      mci->transformed_header = true;
1929      VLOG(2) << "GetOutput transformed header stream_id: ["
1930        << mci->stream_id << "]";
1931      return;
1932    }
1933    if (mci->body_bytes_consumed >= mci->file_data->body.size()) {
1934      SendEOF(mci->stream_id);
1935      output_ordering_.RemoveStreamId(mci->stream_id);
1936      VLOG(2) << "GetOutput remove_stream_id: [" << mci->stream_id << "]";
1937      return;
1938    }
1939    size_t num_to_write =
1940      mci->file_data->body.size() - mci->body_bytes_consumed;
1941    if (num_to_write > mci->max_segment_size)
1942      num_to_write = mci->max_segment_size;
1943    SendDataFrame(mci->stream_id,
1944                  mci->file_data->body.data() + mci->body_bytes_consumed,
1945                  num_to_write, 0, true);
1946    VLOG(2) << "GetOutput SendDataFrame[" << mci->stream_id
1947      << "]: " << num_to_write;
1948    mci->body_bytes_consumed += num_to_write;
1949    mci->bytes_sent += num_to_write;
1950  }
1951};
1952
1953////////////////////////////////////////////////////////////////////////////////
1954
1955class Notification {
1956 public:
1957  explicit Notification(bool value) : value_(value) {}
1958
1959  void Notify() {
1960    AutoLock al(lock_);
1961    value_ = true;
1962  }
1963  bool HasBeenNotified() {
1964    AutoLock al(lock_);
1965    return value_;
1966  }
1967  bool value_;
1968  Lock lock_;
1969};
1970
1971////////////////////////////////////////////////////////////////////////////////
1972
1973class SMAcceptorThread : public SimpleThread,
1974                         public EpollCallbackInterface,
1975                         public SMServerConnectionPoolInterface {
1976  EpollServer epoll_server_;
1977  int listen_fd_;
1978  int accepts_per_wake_;
1979
1980  vector<SMServerConnection*> unused_server_connections_;
1981  vector<SMServerConnection*> tmp_unused_server_connections_;
1982  vector<SMServerConnection*> allocated_server_connections_;
1983  Notification quitting_;
1984  SMInterfaceFactory* sm_interface_factory_;
1985  MemoryCache* memory_cache_;
1986 public:
1987
1988  SMAcceptorThread(int listen_fd,
1989                   int accepts_per_wake,
1990                   SMInterfaceFactory* smif,
1991                   MemoryCache* memory_cache) :
1992      SimpleThread("SMAcceptorThread"),
1993      listen_fd_(listen_fd),
1994      accepts_per_wake_(accepts_per_wake),
1995      quitting_(false),
1996      sm_interface_factory_(smif),
1997      memory_cache_(memory_cache) {
1998  }
1999
2000  ~SMAcceptorThread() {
2001    for (vector<SMServerConnection*>::iterator i =
2002           allocated_server_connections_.begin();
2003         i != allocated_server_connections_.end();
2004         ++i) {
2005      delete *i;
2006    }
2007  }
2008
2009  SMServerConnection* NewConnection() {
2010    SMServerConnection* server =
2011      SMServerConnection::NewSMServerConnection(sm_interface_factory_,
2012                                                memory_cache_,
2013                                                &epoll_server_);
2014    allocated_server_connections_.push_back(server);
2015    VLOG(3) << "Making new server: " << server;
2016    return server;
2017  }
2018
2019  SMServerConnection* FindOrMakeNewSMServerConnection() {
2020    if (unused_server_connections_.empty()) {
2021      return NewConnection();
2022    }
2023    SMServerConnection* retval = unused_server_connections_.back();
2024    unused_server_connections_.pop_back();
2025    return retval;
2026  }
2027
2028
2029  void InitWorker() {
2030    epoll_server_.RegisterFD(listen_fd_, this, EPOLLIN | EPOLLET);
2031  }
2032
2033  void HandleConnection(int client_fd) {
2034    SMServerConnection* server_connection = FindOrMakeNewSMServerConnection();
2035    if (server_connection == NULL) {
2036      VLOG(2) << "Closing " << client_fd;
2037      close(client_fd);
2038      return;
2039    }
2040    server_connection->InitSMServerConnection(this,
2041                                            &epoll_server_,
2042                                            client_fd);
2043  }
2044
2045  void AcceptFromListenFD() {
2046    if (accepts_per_wake_ > 0) {
2047      for (int i = 0; i < accepts_per_wake_; ++i) {
2048        struct sockaddr address;
2049        socklen_t socklen = sizeof(address);
2050        int fd = accept(listen_fd_, &address, &socklen);
2051        if (fd == -1) {
2052          VLOG(2) << "accept fail(" << listen_fd_ << "): " << errno;
2053          break;
2054        }
2055        VLOG(2) << "********************Accepted fd: " << fd << "\n\n\n";
2056        HandleConnection(fd);
2057      }
2058    } else {
2059      while (true) {
2060        struct sockaddr address;
2061        socklen_t socklen = sizeof(address);
2062        int fd = accept(listen_fd_, &address, &socklen);
2063        if (fd == -1) {
2064          VLOG(2) << "accept fail(" << listen_fd_ << "): " << errno;
2065          break;
2066        }
2067        VLOG(2) << "********************Accepted fd: " << fd << "\n\n\n";
2068        HandleConnection(fd);
2069      }
2070    }
2071  }
2072
2073  // EpollCallbackInteface virtual functions.
2074  virtual void OnRegistration(EpollServer* eps, int fd, int event_mask) { }
2075  virtual void OnModification(int fd, int event_mask) { }
2076  virtual void OnEvent(int fd, EpollEvent* event) {
2077    if (event->in_events | EPOLLIN) {
2078      VLOG(2) << "Accepting based upon epoll events";
2079      AcceptFromListenFD();
2080    }
2081  }
2082  virtual void OnUnregistration(int fd, bool replaced) { }
2083  virtual void OnShutdown(EpollServer* eps, int fd) { }
2084
2085  void Quit() {
2086    quitting_.Notify();
2087  }
2088
2089  void Run() {
2090    while (!quitting_.HasBeenNotified()) {
2091      epoll_server_.set_timeout_in_us(10 * 1000);  // 10 ms
2092      epoll_server_.WaitForEventsAndExecuteCallbacks();
2093      unused_server_connections_.insert(unused_server_connections_.end(),
2094                                        tmp_unused_server_connections_.begin(),
2095                                        tmp_unused_server_connections_.end());
2096      tmp_unused_server_connections_.clear();
2097    }
2098  }
2099
2100  // SMServerConnections will use this:
2101  virtual void SMServerConnectionDone(SMServerConnection* sc) {
2102    VLOG(3) << "Done with server connection: " << sc;
2103    tmp_unused_server_connections_.push_back(sc);
2104  }
2105};
2106
2107////////////////////////////////////////////////////////////////////////////////
2108
2109SMInterface* NewSpdySM(SMServerConnection* connection) {
2110  return new SpdySM(connection);
2111}
2112
2113SMInterface* NewHTTPSM(SMServerConnection* connection) {
2114  return new HTTPSM(connection);
2115}
2116
2117////////////////////////////////////////////////////////////////////////////////
2118
2119int CreateListeningSocket(int port, int backlog_size,
2120                          bool reuseport, bool no_nagle) {
2121  int listening_socket = 0;
2122  char port_buf[256];
2123  snprintf(port_buf, sizeof(port_buf), "%d", port);
2124  cerr <<" Attempting to listen on port: " << port_buf << "\n";
2125  cerr <<" input port: " << port << "\n";
2126  net::CreateListeningSocket("",
2127                              port_buf,
2128                              true,
2129                              backlog_size,
2130                              &listening_socket,
2131                              true,
2132                              reuseport,
2133                              &cerr);
2134  SetNonBlocking(listening_socket);
2135  if (no_nagle) {
2136    // set SO_REUSEADDR on the listening socket.
2137    int on = 1;
2138    int rc;
2139    rc = setsockopt(listening_socket, IPPROTO_TCP,  TCP_NODELAY,
2140                    reinterpret_cast<char*>(&on), sizeof(on));
2141    if (rc < 0) {
2142      close(listening_socket);
2143      LOG(FATAL) << "setsockopt() failed fd=" << listening_socket << "\n";
2144    }
2145  }
2146  return listening_socket;
2147}
2148
2149////////////////////////////////////////////////////////////////////////////////
2150
2151bool GotQuitFromStdin() {
2152  // Make stdin nonblocking. Yes this is done each time. Oh well.
2153  fcntl(0, F_SETFL, O_NONBLOCK);
2154  char c;
2155  string maybequit;
2156  while (read(0, &c, 1) > 0) {
2157    maybequit += c;
2158  }
2159  if (maybequit.size()) {
2160    VLOG(2) << "scanning string: \"" << maybequit << "\"";
2161  }
2162  return (maybequit.size() > 1 &&
2163          (maybequit.c_str()[0] == 'q' ||
2164           maybequit.c_str()[0] == 'Q'));
2165}
2166
2167
2168////////////////////////////////////////////////////////////////////////////////
2169
2170const char* BoolToStr(bool b) {
2171  if (b)
2172    return "true";
2173  return "false";
2174}
2175
2176////////////////////////////////////////////////////////////////////////////////
2177
2178int main(int argc, char**argv) {
2179  bool use_ssl = FLAGS_use_ssl;
2180  int response_count_until_close = FLAGS_response_count_until_close;
2181  int spdy_port = FLAGS_spdy_port;
2182  int port = FLAGS_port;
2183  int backlog_size = FLAGS_accept_backlog_size;
2184  bool reuseport = FLAGS_reuseport;
2185  bool no_nagle = FLAGS_no_nagle;
2186  double server_think_time_in_s = FLAGS_server_think_time_in_s;
2187  int accepts_per_wake = FLAGS_accepts_per_wake;
2188  int num_threads = 1;
2189
2190
2191  MemoryCache spdy_memory_cache;
2192  spdy_memory_cache.AddFiles();
2193
2194  MemoryCache http_memory_cache;
2195  http_memory_cache.CloneFrom(spdy_memory_cache);
2196
2197  LOG(INFO) <<
2198    "Starting up with the following state: \n"
2199    "                      use_ssl: " << use_ssl << "\n"
2200    "   response_count_until_close: " << response_count_until_close << "\n"
2201    "                         port: " << port << "\n"
2202    "                    spdy_port: " << spdy_port << "\n"
2203    "                 backlog_size: " << backlog_size << "\n"
2204    "                    reuseport: " << BoolToStr(reuseport) << "\n"
2205    "                     no_nagle: " << BoolToStr(no_nagle) << "\n"
2206    "       server_think_time_in_s: " << server_think_time_in_s << "\n"
2207    "             accepts_per_wake: " << accepts_per_wake << "\n"
2208    "                  num_threads: " << num_threads << "\n"
2209    "                     use_xsub: " << BoolToStr(FLAGS_use_xsub) << "\n"
2210    "                      use_xac: " << BoolToStr(FLAGS_use_xac) << "\n";
2211
2212  if (use_ssl) {
2213    global_ssl_state = new GlobalSSLState;
2214    spdy_init_ssl(global_ssl_state);
2215  } else {
2216    global_ssl_state = NULL;
2217  }
2218  EpollServer epoll_server;
2219  vector<SMAcceptorThread*> sm_worker_threads_;
2220
2221  {
2222    // spdy
2223    int listen_fd = -1;
2224
2225    if (reuseport || listen_fd == -1) {
2226      listen_fd = CreateListeningSocket(spdy_port, backlog_size,
2227                                        reuseport, no_nagle);
2228      if (listen_fd < 0) {
2229        LOG(FATAL) << "Unable to open listening socket on spdy_port: "
2230          << spdy_port;
2231      } else {
2232        LOG(INFO) << "Listening for spdy on port: " << spdy_port;
2233      }
2234    }
2235    sm_worker_threads_.push_back(
2236        new SMAcceptorThread(listen_fd,
2237                             accepts_per_wake,
2238                             &NewSpdySM,
2239                             &spdy_memory_cache));
2240    // Note that spdy_memory_cache is not threadsafe, it is merely
2241    // thread compatible. Thus, if ever we are to spawn multiple threads,
2242    // we either must make the MemoryCache threadsafe, or use
2243    // a separate MemoryCache for each thread.
2244    //
2245    // The latter is what is currently being done as we spawn
2246    // two threads (one for spdy, one for http).
2247    sm_worker_threads_.back()->InitWorker();
2248    sm_worker_threads_.back()->Start();
2249  }
2250
2251  {
2252    // http
2253    int listen_fd = -1;
2254    if (reuseport || listen_fd == -1) {
2255      listen_fd = CreateListeningSocket(port, backlog_size,
2256                                        reuseport, no_nagle);
2257      if (listen_fd < 0) {
2258        LOG(FATAL) << "Unable to open listening socket on port: " << port;
2259      } else {
2260        LOG(INFO) << "Listening for HTTP on port: " << port;
2261      }
2262    }
2263    sm_worker_threads_.push_back(
2264        new SMAcceptorThread(listen_fd,
2265                             accepts_per_wake,
2266                             &NewHTTPSM,
2267                             &http_memory_cache));
2268    // Note that spdy_memory_cache is not threadsafe, it is merely
2269    // thread compatible. Thus, if ever we are to spawn multiple threads,
2270    // we either must make the MemoryCache threadsafe, or use
2271    // a separate MemoryCache for each thread.
2272    //
2273    // The latter is what is currently being done as we spawn
2274    // two threads (one for spdy, one for http).
2275    sm_worker_threads_.back()->InitWorker();
2276    sm_worker_threads_.back()->Start();
2277  }
2278
2279  while (true) {
2280    if (GotQuitFromStdin()) {
2281      for (unsigned int i = 0; i < sm_worker_threads_.size(); ++i) {
2282        sm_worker_threads_[i]->Quit();
2283      }
2284      for (unsigned int i = 0; i < sm_worker_threads_.size(); ++i) {
2285        sm_worker_threads_[i]->Join();
2286      }
2287      return 0;
2288    }
2289    usleep(1000*10);  // 10 ms
2290  }
2291  return 0;
2292}
2293
2294