| /* |
| * Copyright 2012 Google Inc. |
| * |
| * Licensed under the Apache License, Version 2.0 (the "License"); |
| * you may not use this file except in compliance with the License. |
| * You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| // Author: x.dinic@gmail.com (Junmin Xiong) |
| |
| extern "C" { |
| #include <ngx_http.h> |
| #include <ngx_core.h> |
| } |
| |
| #include "ngx_url_async_fetcher.h" |
| #include "ngx_fetch.h" |
| |
| #include <vector> |
| #include <algorithm> |
| #include <string> |
| #include <list> |
| #include <map> |
| #include <set> |
| |
| #include "net/instaweb/http/public/async_fetch.h" |
| #include "net/instaweb/http/public/inflating_fetch.h" |
| #include "net/instaweb/public/version.h" |
| #include "pagespeed/kernel/base/basictypes.h" |
| #include "pagespeed/kernel/base/condvar.h" |
| #include "pagespeed/kernel/base/message_handler.h" |
| #include "pagespeed/kernel/base/pool.h" |
| #include "pagespeed/kernel/base/pool_element.h" |
| #include "pagespeed/kernel/base/statistics.h" |
| #include "pagespeed/kernel/base/string_util.h" |
| #include "pagespeed/kernel/base/thread_system.h" |
| #include "pagespeed/kernel/base/timer.h" |
| #include "pagespeed/kernel/base/writer.h" |
| #include "pagespeed/kernel/http/request_headers.h" |
| #include "pagespeed/kernel/http/response_headers.h" |
| #include "pagespeed/kernel/http/response_headers_parser.h" |
| |
| namespace net_instaweb { |
| |
| NgxUrlAsyncFetcher::NgxUrlAsyncFetcher(const char* proxy, |
| ngx_log_t* log, |
| ngx_msec_t resolver_timeout, |
| ngx_msec_t fetch_timeout, |
| ngx_resolver_t* resolver, |
| int max_keepalive_requests, |
| ThreadSystem* thread_system, |
| MessageHandler* handler) |
| : fetchers_count_(0), |
| shutdown_(false), |
| track_original_content_length_(false), |
| byte_count_(0), |
| thread_system_(thread_system), |
| message_handler_(handler), |
| mutex_(NULL), |
| max_keepalive_requests_(max_keepalive_requests), |
| event_connection_(NULL) { |
| resolver_timeout_ = resolver_timeout; |
| fetch_timeout_ = fetch_timeout; |
| ngx_memzero(&proxy_, sizeof(proxy_)); |
| if (proxy != NULL && *proxy != '\0') { |
| proxy_.url.data = reinterpret_cast<u_char*>(const_cast<char*>(proxy)); |
| proxy_.url.len = ngx_strlen(proxy); |
| } |
| mutex_ = thread_system_->NewMutex(); |
| log_ = log; |
| pool_ = NULL; |
| resolver_ = resolver; |
| // If init fails, set shutdown_ so no fetches will be attempted. |
| if (!Init(const_cast<ngx_cycle_t*>(ngx_cycle))) { |
| shutdown_ = true; |
| message_handler_->Message( |
| kError, "NgxUrlAsyncFetcher failed to init, fetching disabled."); |
| } |
| } |
| |
| NgxUrlAsyncFetcher::~NgxUrlAsyncFetcher() { |
| DCHECK(shutdown_) << "Shut down before destructing NgxUrlAsyncFetcher."; |
| message_handler_->Message( |
| kInfo, |
| "Destruct NgxUrlAsyncFetcher with [%d] active fetchers", |
| ApproximateNumActiveFetches()); |
| |
| CancelActiveFetches(); |
| active_fetches_.DeleteAll(); |
| NgxConnection::Terminate(); |
| |
| if (pool_ != NULL) { |
| ngx_destroy_pool(pool_); |
| pool_ = NULL; |
| } |
| if (mutex_ != NULL) { |
| delete mutex_; |
| mutex_ = NULL; |
| } |
| } |
| |
| |
| bool NgxUrlAsyncFetcher::ParseUrl(ngx_url_t* url, ngx_pool_t* pool) { |
| size_t scheme_offset; |
| u_short port; |
| if (ngx_strncasecmp(url->url.data, reinterpret_cast<u_char*>( |
| const_cast<char*>("http://")), 7) == 0) { |
| scheme_offset = 7; |
| port = 80; |
| } else if (ngx_strncasecmp(url->url.data, reinterpret_cast<u_char*>( |
| const_cast<char*>("https://")), 8) == 0) { |
| scheme_offset = 8; |
| port = 443; |
| } else { |
| scheme_offset = 0; |
| port = 80; |
| } |
| |
| url->url.data += scheme_offset; |
| url->url.len -= scheme_offset; |
| url->default_port = port; |
| // See: http://lxr.evanmiller.org/http/source/core/ngx_inet.c#L875 |
| url->no_resolve = 0; |
| url->uri_part = 1; |
| |
| if (ngx_parse_url(pool, url) == NGX_OK) { |
| return true; |
| } |
| return false; |
| } |
| |
| // If there are still active requests, cancel them. |
| void NgxUrlAsyncFetcher::CancelActiveFetches() { |
| // TODO(oschaaf): this seems tricky, this may end up calling |
| // FetchComplete, modifying the active fetches while we are looping |
| // it |
| for (NgxFetchPool::const_iterator p = active_fetches_.begin(), |
| e = active_fetches_.end(); p != e; ++p) { |
| NgxFetch* fetch = *p; |
| fetch->CallbackDone(false); |
| } |
| } |
| |
| // Create the pool for fetcher, create the pipe, add the read event for main |
| // thread. It should be called in the worker process. |
| bool NgxUrlAsyncFetcher::Init(ngx_cycle_t* cycle) { |
| log_ = cycle->log; |
| CHECK(event_connection_ == NULL) << "event connection already set"; |
| event_connection_ = new NgxEventConnection(ReadCallback); |
| if (!event_connection_->Init(cycle)) { |
| return false; |
| } |
| if (pool_ == NULL) { |
| pool_ = ngx_create_pool(4096, log_); |
| if (pool_ == NULL) { |
| ngx_log_error(NGX_LOG_ERR, log_, 0, |
| "NgxUrlAsyncFetcher::Init create pool failed"); |
| return false; |
| } |
| } |
| |
| if (proxy_.url.len == 0) { |
| return true; |
| } |
| |
| // TODO(oschaaf): shouldn't we do this earlier? Do we need to clean |
| // up when parsing the url failed? |
| if (!ParseUrl(&proxy_, pool_)) { |
| ngx_log_error(NGX_LOG_ERR, log_, 0, |
| "NgxUrlAsyncFetcher::Init parse proxy[%V] failed", &proxy_.url); |
| return false; |
| } |
| return true; |
| } |
| |
| void NgxUrlAsyncFetcher::ShutDown() { |
| shutdown_ = true; |
| if (!pending_fetches_.empty()) { |
| for (Pool<NgxFetch>::iterator p = pending_fetches_.begin(), |
| e = pending_fetches_.end(); p != e; p++) { |
| NgxFetch* fetch = *p; |
| fetch->CallbackDone(false); |
| } |
| pending_fetches_.DeleteAll(); |
| } |
| |
| if (!active_fetches_.empty()) { |
| for (Pool<NgxFetch>::iterator p = active_fetches_.begin(), |
| e = active_fetches_.end(); p != e; p++) { |
| NgxFetch* fetch = *p; |
| fetch->CallbackDone(false); |
| } |
| active_fetches_.Clear(); |
| } |
| if (event_connection_ != NULL) { |
| event_connection_->Shutdown(); |
| delete event_connection_; |
| event_connection_ = NULL; |
| } |
| } |
| |
| // It's called in the rewrite thread. All the fetches are started at |
| // this function. It will notify the main thread to start the fetch job. |
| void NgxUrlAsyncFetcher::Fetch(const GoogleString& url, |
| MessageHandler* message_handler, |
| AsyncFetch* async_fetch) { |
| // Don't accept new fetches when shut down. This flow is also entered when |
| // we did not initialize properly in ::Init(). |
| if (shutdown_) { |
| async_fetch->Done(false); |
| return; |
| } |
| async_fetch = EnableInflation(async_fetch); |
| NgxFetch* fetch = new NgxFetch(url, async_fetch, |
| message_handler, log_); |
| ScopedMutex lock(mutex_); |
| pending_fetches_.Add(fetch); |
| |
| // TODO(oschaaf): thread safety on written vs shutdown. |
| // It is possible that shutdown() is called after writing an event? In that |
| // case, this could (rarely) fail when it shouldn't. |
| bool written = event_connection_->WriteEvent(this); |
| CHECK(written || shutdown_) << "NgxUrlAsyncFetcher: event write failure"; |
| } |
| |
| // This is the read event which is called in the main thread. |
| // It will do the real work. Add the work event and start the fetch. |
| void NgxUrlAsyncFetcher::ReadCallback(const ps_event_data& data) { |
| std::vector<NgxFetch*> to_start; |
| NgxUrlAsyncFetcher* fetcher = reinterpret_cast<NgxUrlAsyncFetcher*>( |
| data.sender); |
| |
| fetcher->mutex_->Lock(); |
| fetcher->completed_fetches_.DeleteAll(); |
| |
| for (Pool<NgxFetch>::iterator p = fetcher->pending_fetches_.begin(), |
| e = fetcher->pending_fetches_.end(); p != e; p++) { |
| NgxFetch* fetch = *p; |
| to_start.push_back(fetch); |
| } |
| |
| fetcher->pending_fetches_.Clear(); |
| fetcher->mutex_->Unlock(); |
| |
| for (size_t i = 0; i < to_start.size(); i++) { |
| fetcher->StartFetch(to_start[i]); |
| } |
| |
| return; |
| } |
| |
| // TODO(oschaaf): return value is ignored. |
| bool NgxUrlAsyncFetcher::StartFetch(NgxFetch* fetch) { |
| mutex_->Lock(); |
| active_fetches_.Add(fetch); |
| fetchers_count_++; |
| mutex_->Unlock(); |
| |
| // Don't initiate the fetch when we are shutting down |
| if (shutdown_) { |
| fetch->CallbackDone(false); |
| return false; |
| } |
| |
| bool started = fetch->Start(this); |
| |
| if (!started) { |
| message_handler_->Message(kWarning, "Fetch failed to start: %s", |
| fetch->str_url()); |
| fetch->CallbackDone(false); |
| } |
| |
| return started; |
| } |
| |
| void NgxUrlAsyncFetcher::FetchComplete(NgxFetch* fetch) { |
| ScopedMutex lock(mutex_); |
| byte_count_ += fetch->bytes_received(); |
| fetchers_count_--; |
| active_fetches_.Remove(fetch); |
| completed_fetches_.Add(fetch); |
| } |
| |
| void NgxUrlAsyncFetcher::PrintActiveFetches(MessageHandler* handler) const { |
| for (NgxFetchPool::const_iterator p = active_fetches_.begin(), |
| e = active_fetches_.end(); p != e; ++p) { |
| NgxFetch* fetch = *p; |
| handler->Message(kInfo, "Active fetch: %s", fetch->str_url()); |
| } |
| } |
| } // namespace net_instaweb |