Refactor the first-read logic for the in-mem server....

net/tools/flip_server/flip_in_mem_edsm_server.cc

 // 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 <dirent.h>
 #include <netinet/tcp.h>  // For TCP_NODELAY
 #include <sys/socket.h>
 #include <sys/types.h>
 #include <sys/file.h>
 #include <sys/stat.h>
@@ skipping 133 matching lines @@
 using spdy::SpdySynStreamControlFrame;
 
 FlipConfig g_proxy_config;
 
 ////////////////////////////////////////////////////////////////////////////////
 
 void PrintSslError() {
   char buf[128];  // this buffer must be at least 120 chars long.
   int error_num = ERR_get_error();
   while (error_num != 0) {
-    LOG(ERROR) << ERR_error_string(error_num, buf);
+    ERR_error_string_n(error_num, buf, sizeof(buf));
+    LOG(ERROR) << buf;
     error_num = ERR_get_error();
   }
 }
 
 static int ssl_set_npn_callback(SSL *s,
                                 const unsigned char **data,
                                 unsigned int *len,
                                 void *arg)
 {
   VLOG(1) <<  "SSL NPN callback: advertising protocols.";
@@ skipping 871 matching lines @@
     Cleanup("OnShutdown");
     return;
   }
 
   void Cleanup(const char* cleanup) {
     VLOG(2) << log_prefix_ << ACCEPTOR_CLIENT_IDENT << "Cleanup: " << cleanup;
     if (!initialized_) {
       return;
     }
     Reset();
-    if (connection_pool_) {
+    if (connection_pool_)
       connection_pool_->SMConnectionDone(this);
-    }
-    if (sm_interface_) {
+    if (sm_interface_)
       sm_interface_->ResetForNewConnection();
-    }
     last_read_time_ = 0;
   }
 
  private:
   void HandleEvents() {
     VLOG(2) << log_prefix_ << ACCEPTOR_CLIENT_IDENT << "Received: "
             << EpollServer::EventMaskToString(events_).c_str();
 
     if (events_ & EPOLLIN) {
       if (!DoRead())
@@ skipping 32 matching lines @@
     if (events_ & (EPOLLHUP | EPOLLERR)) {
       VLOG(1) << log_prefix_ << ACCEPTOR_CLIENT_IDENT << "!!! Got HUP or ERR";
       goto handle_close_or_error;
     }
     return;
 
     handle_close_or_error:
     Cleanup("HandleEvents");
   }
 
+  // Decide if SPDY was negotiated.
+  bool WasSpdyNegotiated() {
+    if (FLAGS_force_spdy)
+      return true;
+
+    // If this is an SSL connection, check if NPN specifies SPDY.
+    if (ssl_) {
+      const unsigned char *npn_proto;
+      unsigned int npn_proto_len;
+      SSL_get0_next_proto_negotiated(ssl_, &npn_proto, &npn_proto_len);
+      if (npn_proto_len > 0) {
+        string npn_proto_str((const char *)npn_proto, npn_proto_len);
+        VLOG(1) << log_prefix_ << ACCEPTOR_CLIENT_IDENT
+                << "NPN protocol detected: " << npn_proto_str;
+        if (!strncmp(reinterpret_cast<const char*>(npn_proto),
+                     "spdy/2", npn_proto_len))
+          return true;
+      }
+    }
+
+    return false;
+  }
+
+  // Initialize the protocol interfaces we'll need for this connection.
+  // Returns true if successful, false otherwise.
+  bool SetupProtocolInterfaces() {
+    DCHECK(!protocol_detected_);
+    protocol_detected_ = true;
+
+    bool spdy_negotiated = WasSpdyNegotiated();
+    bool using_ssl = ssl_ != NULL;
+
+    if (using_ssl)
+      VLOG(1) << (SSL_session_reused(ssl_) ? "Resumed" : "Renegotiated")
+              << " SSL Session.";
+
+    if (acceptor_->spdy_only_ && !spdy_negotiated) {
+      VLOG(1) << log_prefix_ << ACCEPTOR_CLIENT_IDENT
+              << "SPDY proxy only, closing HTTPS connection.";
+      return false;
+    }
+
+    switch (acceptor_->flip_handler_type_) {
+      case FLIP_HANDLER_HTTP_SERVER:
+        {
+          DCHECK(!spdy_negotiated);
+          VLOG(2) << log_prefix_ << ACCEPTOR_CLIENT_IDENT
+                  << (sm_http_interface_ ? "Creating" : "Reusing")
+                  << " HTTP interface.";
+          if (!sm_http_interface_)
+            sm_http_interface_ = NewHttpSM(this, NULL, epoll_server_,
+                                           memory_cache_, acceptor_);
+          sm_interface_ = sm_http_interface_;
+        }
+        break;
+      case FLIP_HANDLER_PROXY:
+        {
+          VLOG(2) << log_prefix_ << ACCEPTOR_CLIENT_IDENT
+                  << (sm_streamer_interface_ ? "Creating" : "Reusing")
+                  << " PROXY Streamer interface.";
+          if (!sm_streamer_interface_)
+            sm_streamer_interface_ = NewStreamerSM(this, NULL,
+                                                   epoll_server_,
+                                                   acceptor_);
+          sm_interface_ = sm_streamer_interface_;
+          // If spdy is not negotiated, the streamer interface will proxy all
+          // data to the origin server.
+          if (!spdy_negotiated)
+            break;
+        }
+        // Otherwise fall through into the case below.
+      case FLIP_HANDLER_SPDY_SERVER:
+        {
+          DCHECK(spdy_negotiated);
+          VLOG(2) << log_prefix_ << ACCEPTOR_CLIENT_IDENT
+                  << (sm_spdy_interface_ ? "Creating" : "Reusing")
+                  << " SPDY interface.";
+          if (!sm_spdy_interface_)
+            sm_spdy_interface_ = NewSpdySM(this, NULL, epoll_server_,
+                                           memory_cache_, acceptor_);
+          sm_interface_ = sm_spdy_interface_;
+        }
+        break;
+    }
+    if (!sm_interface_->PostAcceptHook())
+      return false;
+
+    return true;
+  }
+
   bool DoRead() {
     VLOG(2) << log_prefix_ << ACCEPTOR_CLIENT_IDENT << "DoRead()";
     while (!read_buffer_.Full()) {
       char* bytes;
       int size;
       if (fd_ == -1) {
         VLOG(2) << log_prefix_ << ACCEPTOR_CLIENT_IDENT
                 << "DoRead(): fd_ == -1. Invalid FD. Returning false";
         return false;
       }
       read_buffer_.GetWritablePtr(&bytes, &size);
       ssize_t bytes_read = 0;
       if (ssl_) {
         bytes_read = SSL_read(ssl_, bytes, size);
         if (bytes_read < 0) {
-          switch(SSL_get_error(ssl_, bytes_read)) {
+          int err = SSL_get_error(ssl_, bytes_read);
+          switch(err) {
             case SSL_ERROR_WANT_READ:
             case SSL_ERROR_WANT_WRITE:
             case SSL_ERROR_WANT_ACCEPT:
             case SSL_ERROR_WANT_CONNECT:
               events_ &= ~EPOLLIN;
+              VLOG(2) << log_prefix_ << ACCEPTOR_CLIENT_IDENT
+                      << "DoRead: SSL WANT_XXX: " << err;
               goto done;
             default:
               PrintSslError();
               goto error_or_close;
           }
         }
       } else {
         bytes_read = recv(fd_, bytes, size, MSG_DONTWAIT);
       }
       int stored_errno = errno;
@@ skipping 11 matching lines @@
           default:
             VLOG(1) << log_prefix_ << ACCEPTOR_CLIENT_IDENT
                     << "While calling recv, got error: "
                     << (ssl_?"(ssl error)":strerror(stored_errno));
             goto error_or_close;
         }
       } else if (bytes_read > 0) {
         VLOG(2) << log_prefix_ << ACCEPTOR_CLIENT_IDENT << "read " << bytes_read
                  << " bytes";
         last_read_time_ = time(NULL);
+        // If the protocol hasn't been detected yet, set up the handlers
+        // we'll need.
         if (!protocol_detected_) {
-          if (acceptor_->flip_handler_type_ == FLIP_HANDLER_HTTP_SERVER) {
-            // Http Server
-            protocol_detected_ = true;
-            if (!sm_http_interface_) {
-              VLOG(2) << log_prefix_ << ACCEPTOR_CLIENT_IDENT
-                      << "Created HTTP interface.";
-              sm_http_interface_ = NewHttpSM(this, NULL, epoll_server_,
-                                             memory_cache_, acceptor_);
-            } else {
-              VLOG(2) << log_prefix_ << ACCEPTOR_CLIENT_IDENT
-                      << "Reusing HTTP interface.";
-            }
-            sm_interface_ = sm_http_interface_;
-          } else if (ssl_) {
-            protocol_detected_ = true;
-            if (SSL_session_reused(ssl_) == 0) {
-              VLOG(1) << "Session status: renegotiated";
-            } else {
-              VLOG(1) << "Session status: resumed";
-            }
-            bool spdy_negotiated = FLAGS_force_spdy;
-            if (!spdy_negotiated) {
-              const unsigned char *npn_proto;
-              unsigned int npn_proto_len;
-              SSL_get0_next_proto_negotiated(ssl_, &npn_proto, &npn_proto_len);
-              if (npn_proto_len > 0) {
-                string npn_proto_str((const char *)npn_proto, npn_proto_len);
-                VLOG(1) << log_prefix_ << ACCEPTOR_CLIENT_IDENT
-                        << "NPN protocol detected: " << npn_proto_str;
-              } else {
-                VLOG(1) << log_prefix_ << ACCEPTOR_CLIENT_IDENT
-                        << "NPN protocol detected: none";
-                if (acceptor_->flip_handler_type_ == FLIP_HANDLER_SPDY_SERVER) {
-                  VLOG(1) << log_prefix_ << ACCEPTOR_CLIENT_IDENT
-                          << "NPN protocol: Could not negotiate SPDY protocol.";
-                  goto error_or_close;
-                }
-              }
-              if (npn_proto_len > 0 &&
-                  !strncmp(reinterpret_cast<const char*>(npn_proto),
-                           "spdy/2", npn_proto_len)) {
-                  spdy_negotiated = true;
-              }
-            }
-            if (spdy_negotiated) {
-              if (!sm_spdy_interface_) {
-                sm_spdy_interface_ = NewSpdySM(this, NULL, epoll_server_,
-                                               memory_cache_, acceptor_);
-                VLOG(2) << log_prefix_ << ACCEPTOR_CLIENT_IDENT
-                        << "Created SPDY interface.";
-              } else {
-                VLOG(2) << log_prefix_ << ACCEPTOR_CLIENT_IDENT
-                        << "Reusing SPDY interface.";
-              }
-              sm_interface_ = sm_spdy_interface_;
-            } else if (acceptor_->spdy_only_) {
-              VLOG(1) << log_prefix_ << ACCEPTOR_CLIENT_IDENT
-                      << "SPDY proxy only, closing HTTPS connection.";
-              goto error_or_close;
-            } else {
-              if (!sm_streamer_interface_) {
-                sm_streamer_interface_ = NewStreamerSM(this, NULL,
-                                                       epoll_server_,
-                                                       acceptor_);
-                VLOG(2) << log_prefix_ << ACCEPTOR_CLIENT_IDENT
-                        << "Created Streamer interface.";
-              } else {
-                VLOG(2) << log_prefix_ << ACCEPTOR_CLIENT_IDENT
-                        << "Reusing Streamer interface: ";
-              }
-              sm_interface_ = sm_streamer_interface_;
-            }
-          }
-          if (sm_interface_->PostAcceptHook() == 0) {
+          if (!SetupProtocolInterfaces()) {
+            LOG(ERROR) << "Error setting up protocol interfaces.";
             goto error_or_close;
           }
         }
         read_buffer_.AdvanceWritablePtr(bytes_read);
-        if (!DoConsumeReadData()) {
+        if (!DoConsumeReadData())
           goto error_or_close;
-        }
         continue;
       } else {  // bytes_read == 0
         VLOG(1) << log_prefix_ << ACCEPTOR_CLIENT_IDENT
                 << "0 bytes read with recv call.";
       }
       goto error_or_close;
     }
    done:
+    VLOG(2) << log_prefix_ << ACCEPTOR_CLIENT_IDENT << "DoRead done!";
     return true;
 
     error_or_close:
     VLOG(1) << log_prefix_ << ACCEPTOR_CLIENT_IDENT
             << "DoRead(): error_or_close. "
             << "Cleaning up, then returning false";
     Cleanup("DoRead");
     return false;
   }
 
@@ skipping 54 matching lines @@
       }
       if (output_list_.empty()) {
         events_ &= ~EPOLLOUT;
       }
     }
     while (!output_list_.empty()) {
       VLOG(2) << log_prefix_ << "DoWrite: Items in output list: "
               << output_list_.size();
       if (bytes_sent >= max_bytes_sent_per_dowrite_) {
         VLOG(2) << log_prefix_ << ACCEPTOR_CLIENT_IDENT
-                << " byte sent >= max bytes sent per write: Setting EPOLLOUT";
+                << " byte sent >= max bytes sent per write: Setting EPOLLOUT: "
+                << bytes_sent;
         events_ |= EPOLLOUT;
         break;
       }
       if (sm_interface_ && output_list_.size() < 2) {
         sm_interface_->GetOutput();
       }
       DataFrame* data_frame = output_list_.front();
       const char*  bytes = data_frame->data;
       int size = data_frame->size;
       bytes += data_frame->index;
@@ skipping 90 matching lines @@
     protocol_detected_ = false;
     events_ = 0;
     for (list<DataFrame*>::iterator i =
          output_list_.begin();
          i != output_list_.end();
          ++i) {
       delete *i;
     }
     output_list_.clear();
   }
-
 };
 
 ////////////////////////////////////////////////////////////////////////////////
 
 class OutputOrdering {
  public:
   typedef list<MemCacheIter> PriorityRing;
 
   typedef map<uint32, PriorityRing> PriorityMap;
 
@@ skipping 175 matching lines @@
   }
 };
 
 
 ////////////////////////////////////////////////////////////////////////////////
 
 class SpdySM : public SpdyFramerVisitorInterface, public SMInterface {
  private:
   uint64 seq_num_;
   SpdyFramer* spdy_framer_;
+  bool valid_spdy_session_;  // True if we have seen valid data on this session.
+                             // Use this to fail fast when junk is sent to our
+                             // port.
 
   SMConnection* connection_;
   OutputList* client_output_list_;
   OutputOrdering client_output_ordering_;
   uint32 next_outgoing_stream_id_;
   EpollServer* epoll_server_;
   FlipAcceptor* acceptor_;
   MemoryCache* memory_cache_;
   vector<SMInterface*> server_interface_list;
   vector<int32> unused_server_interface_list;
   typedef map<uint32,SMInterface*> StreamToSmif;
   StreamToSmif stream_to_smif_;
  public:
   SpdySM(SMConnection* connection,
          SMInterface* sm_http_interface,
          EpollServer* epoll_server,
          MemoryCache* memory_cache,
          FlipAcceptor* acceptor)
          : seq_num_(0),
            spdy_framer_(new SpdyFramer),
+           valid_spdy_session_(false),
            connection_(connection),
            client_output_list_(connection->output_list()),
            client_output_ordering_(connection),
            next_outgoing_stream_id_(2),
            epoll_server_(epoll_server),
            acceptor_(acceptor),
            memory_cache_(memory_cache) {
     spdy_framer_->set_visitor(this);
   }
 
@@ skipping 145 matching lines @@
         const SpdySynStreamControlFrame* syn_stream =
             reinterpret_cast<const SpdySynStreamControlFrame*>(frame);
 
           string http_data;
           bool is_https_scheme;
           int ret = SpdyHandleNewStream(frame, http_data, &is_https_scheme);
           if (!ret) {
             LOG(ERROR) << "SpdySM: Could not convert spdy into http.";
             break;
           }
+          // We've seen a valid looking SYN_STREAM, consider this to have
+          // been a real spdy session.
+          valid_spdy_session_ = true;
 
           if (acceptor_->flip_handler_type_ == FLIP_HANDLER_PROXY) {
             string server_ip;
             string server_port;
             if (is_https_scheme) {
               server_ip = acceptor_->https_server_ip_;
               server_port = acceptor_->https_server_port_;
             } else {
               server_ip = acceptor_->http_server_ip_;
               server_port = acceptor_->http_server_port_;
@@ skipping 25 matching lines @@
         break;
 
       default:
         LOG(ERROR) << "SpdySM: Unknown control frame type";
     }
   }
   virtual void OnStreamFrameData(SpdyStreamId stream_id,
                                  const char* data, size_t len) {
     VLOG(2) << ACCEPTOR_CLIENT_IDENT << "SpdySM: StreamData(" << stream_id
             << ", [" << len << "])";
-    if (acceptor_->flip_handler_type_ == FLIP_HANDLER_PROXY) {
-          stream_to_smif_[stream_id]->ProcessWriteInput(data, len);
+    StreamToSmif::iterator it = stream_to_smif_.find(stream_id);
+    if (it == stream_to_smif_.end()) {
+      VLOG(2) << "Dropping frame from unknown stream " << stream_id;
+      if (!valid_spdy_session_)
+        connection_->Cleanup("invalid");
+      return;
     }
+
+    SMInterface* interface = it->second;
+    if (acceptor_->flip_handler_type_ == FLIP_HANDLER_PROXY)
+      interface->ProcessWriteInput(data, len);
   }
 
  public:
   size_t ProcessReadInput(const char* data, size_t len) {
     return spdy_framer_->ProcessInput(data, len);
   }
 
   size_t ProcessWriteInput(const char* data, size_t len) {
     return 0;
   }
@@ skipping 20 matching lines @@
     VLOG(2) << ACCEPTOR_CLIENT_IDENT << "SpdySM: Reset for new interface: "
             << "server_idx: " << server_idx;
     unused_server_interface_list.push_back(server_idx);
   }
 
   void ResetForNewConnection() {
     // seq_num is not cleared, intentionally.
     delete spdy_framer_;
     spdy_framer_ = new SpdyFramer;
     spdy_framer_->set_visitor(this);
+    valid_spdy_session_ = false;
     client_output_ordering_.Reset();
     next_outgoing_stream_id_ = 2;
   }
 
   // SMInterface's Cleanup is currently only called by SMConnection after a
   // protocol message as been fully read. Spdy's SMInterface does not need
   // to do any cleanup at this time.
   // TODO (klindsay) This method is probably not being used properly and
   // some logic review and method renaming is probably in order.
   void Cleanup() {}
@@ skipping 505 matching lines @@
     seq_num_ = 0;
     output_ordering_.Reset();
     http_framer_->Reset();
     if (sm_spdy_interface_) {
       sm_spdy_interface_->ResetForNewInterface(server_idx_);
     }
   }
 
   void Cleanup() {
     if (!(acceptor_->flip_handler_type_ == FLIP_HANDLER_HTTP_SERVER)) {
-      connection_->Cleanup("HttpSM Request Fully Read: stream_id " +
-                           stream_id_);
+      VLOG(2) << "HttpSM Request Fully Read; stream_id: " << stream_id_;
+      connection_->Cleanup("request complete");
     }
   }
 
   int PostAcceptHook() {
     return 1;
   }
 
   void NewStream(uint32 stream_id, uint32 priority, const string& filename) {
     MemCacheIter mci;
     mci.stream_id = stream_id;
@@ skipping 376 matching lines @@
   SMAcceptorThread(FlipAcceptor *acceptor,
                    MemoryCache* memory_cache) :
       SimpleThread("SMAcceptorThread"),
       acceptor_(acceptor),
       ssl_state_(NULL),
       use_ssl_(false),
       quitting_(false),
       memory_cache_(memory_cache)
   {
     if (!acceptor->ssl_cert_filename_.empty() &&
-        !acceptor->ssl_cert_filename_.empty()) {
+        !acceptor->ssl_key_filename_.empty()) {
       ssl_state_ = new SSLState;
       bool use_npn = true;
       if (acceptor_->flip_handler_type_ == FLIP_HANDLER_HTTP_SERVER) {
         use_npn = false;
       }
       spdy_init_ssl(ssl_state_, acceptor_->ssl_cert_filename_,
                     acceptor_->ssl_key_filename_, use_npn);
       use_ssl_ = true;
     }
   }
@@ skipping 535 matching lines @@
       }
       break;
     }
     usleep(1000*10);  // 10 ms
   }
 
   unlink(PIDFILE);
   close(pidfile_fd);
   return 0;
 }