src/pagespeed/kernel/cache/purge_context.cc - incubator-pagespeed-debian - Git at Google

 /*
  * Copyright 2013 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: jmarantz@google.com (Joshua Marantz)

 #include "pagespeed/kernel/cache/purge_context.h"

 #include "base/logging.h"
 #include "pagespeed/kernel/base/abstract_mutex.h"
 #include "pagespeed/kernel/base/atomic_bool.h"
 #include "pagespeed/kernel/base/basictypes.h"
 #include "pagespeed/kernel/base/file_system.h"
 #include "pagespeed/kernel/base/function.h"
 #include "pagespeed/kernel/base/message_handler.h"
 #include "pagespeed/kernel/base/null_message_handler.h"
 #include "pagespeed/kernel/base/named_lock_manager.h"
 #include "pagespeed/kernel/base/statistics.h"
 #include "pagespeed/kernel/base/string.h"
 #include "pagespeed/kernel/base/string_util.h"
 #include "pagespeed/kernel/base/thread_system.h"
 #include "pagespeed/kernel/base/time_util.h"
 #include "pagespeed/kernel/cache/lru_cache_base.h"
 #include "pagespeed/kernel/thread/scheduler.h"
 #include "pagespeed/kernel/util/copy_on_write.h"

 namespace net_instaweb {

 namespace {

 const int64 kStealLockAfterMs = 2 * Timer::kSecondMs;
 const int64 kTimeoutMs = 3 * Timer::kSecondMs;
 const int kMaxContentionRetries = 2;

 }  // namespace

 const char PurgeContext::kCancellations[]        = "purge_cancellations";
 const char PurgeContext::kContentions[]          = "purge_contentions";
 const char PurgeContext::kFileParseFailures[]    = "purge_file_parse_failures";
 const char PurgeContext::kFileStats[]            = "purge_file_stats";
 const char PurgeContext::kFileWriteFailures[]    = "purge_file_write_failures";
 const char PurgeContext::kFileWrites[]           = "purge_file_writes";
 const char PurgeContext::kPurgeIndex[]           = "purge_index";

 // TODO(jmarantz): make it possible to avoid showing this implementation detail
 // in the statistics page.
 const char PurgeContext::kPurgePollTimestampMs[] = "_purge_poll_timestamp_ms";

 namespace {

 // Wrapper class around an UpDownCounter which makes it guaranteed to be
 // locally consistent, even if the backing UpDownCounter is non-functional
 // due to a shared-memory failure or statistics being turned off.
 class BackupUpDownCounter : public UpDownCounter {
  public:
   // Takes ownership of mutex.
   BackupUpDownCounter(UpDownCounter* counter, AbstractMutex* mutex)
       : counter_(counter),
         mutex_(mutex),
         local_counter_(0) {
   }

   virtual int64 Get() const {
     // Don't take the lock for local_counter unless we have actually set some
     // value in the past, yet we can't retrieve it from the real UpDownCounter.
     bool was_set = was_set_.value();
     int64 val = counter_->Get();
     if (was_set && (val == 0)) {
       ScopedMutex lock(mutex_.get());
       val = local_counter_;
     }
     return val;
   }

   virtual StringPiece GetName() const { return counter_->GetName(); }

   virtual void Set(int64 value) {
     counter_->Set(value);
     {
       ScopedMutex lock(mutex_.get());
       local_counter_ = value;
       was_set_.set_value(true);
     }
   }

   virtual int64 AddHelper(int64 delta) {
     LOG(DFATAL) << "AddHelper is not used in PurgeContext";
     return 0;
   }

  private:
   UpDownCounter* counter_;
   scoped_ptr<AbstractMutex> mutex_;
   int64 local_counter_ GUARDED_BY(mutex_);
   AtomicBool was_set_;
 };

 }  // namespace

 PurgeContext::PurgeContext(StringPiece filename,
                            FileSystem* file_system,
                            Timer* timer,
                            int max_bytes_in_cache,
                            ThreadSystem* thread_system,
                            NamedLockManager* lock_manager,
                            Scheduler* scheduler,
                            Statistics* statistics,
                            MessageHandler* handler)
     : filename_(filename.data(), filename.size()),
       interprocess_lock_(lock_manager->CreateNamedLock(LockName())),
       file_system_(file_system),
       timer_(timer),
       statistics_(statistics),
       mutex_(thread_system->NewMutex()),
       pending_purges_(max_bytes_in_cache),
       local_purge_index_(0),
       num_consecutive_failures_(0),
       waiting_for_interprocess_lock_(false),
       reading_(false),
       enable_purge_(true),
       max_bytes_in_cache_(max_bytes_in_cache),
       request_batching_delay_ms_(0),
       cancellations_(statistics->GetVariable(kCancellations)),
       contentions_(statistics->GetVariable(kContentions)),
       file_parse_failures_(statistics->GetVariable(kFileParseFailures)),
       file_stats_(statistics->GetVariable(kFileStats)),
       file_write_failures_(statistics->GetVariable(kFileWriteFailures)),
       file_writes_(statistics->GetVariable(kFileWrites)),
       purge_index_(statistics->GetVariable(kPurgeIndex)),
       purge_poll_timestamp_ms_(new BackupUpDownCounter(
           statistics->GetUpDownCounter(kPurgePollTimestampMs),
           thread_system->NewMutex())),
       scheduler_(scheduler),
       message_handler_(handler) {
   purge_set_.MakeWriteable()->set_max_size(max_bytes_in_cache_);
 }

 PurgeContext::~PurgeContext() {
 }

 void PurgeContext::InitStats(Statistics* statistics) {
   statistics->AddVariable(kCancellations);
   statistics->AddVariable(kContentions);
   statistics->AddVariable(kFileParseFailures);
   statistics->AddVariable(kFileStats);
   statistics->AddVariable(kFileWrites);
   statistics->AddVariable(kFileWriteFailures);
   statistics->AddVariable(kPurgeIndex);
   statistics->AddUpDownCounter(kPurgePollTimestampMs);
 }

 bool PurgeContext::ParseAndValidateTimestamp(
     StringPiece time_string, int64 now_ms, int64* timestamp_ms) {
   if (!StringToInt64(time_string, timestamp_ms)) {
     message_handler_->Info(filename_.c_str(), 1,
                            "Invalidation timestamp (%s) not parsed as int64",
                            time_string.as_string().c_str());
     return false;
   } else if ((*timestamp_ms != PurgeSet::kInitialTimestampMs) &&
              ((*timestamp_ms < 0) ||
               (*timestamp_ms > now_ms + PurgeSet::kClockSkewAllowanceMs))) {
     GoogleString converted_time_string;
     ConvertTimeToString(*timestamp_ms, &converted_time_string);
     message_handler_->Info(filename_.c_str(), 1,
                            "Invalidation timestamp (%s) in the future: %s",
                            time_string.as_string().c_str(),
                            converted_time_string.c_str());
     return false;
   }
   return true;
 }

 // Parses the cache purge file.
 void PurgeContext::ReadPurgeFile(PurgeSet* purges_from_file) {
   GoogleString buffer;

   file_stats_->Add(1);
   NullMessageHandler null_handler;

   // Prior to mod_pagespeed 1.8, the cache.flush file's contents were not
   // significant, and generally empty, and only the timestamp of the file
   // itself was important, meaning "wipe everything out of the cache predating
   // that timestamp."
   if (!enable_purge_) {
     int64 timestamp_sec;
     if (file_system_->Mtime(filename_, &timestamp_sec, &null_handler)) {
       int64 timestamp_ms = timestamp_sec * Timer::kSecondMs;
       purges_from_file->UpdateGlobalInvalidationTimestampMs(timestamp_ms);
     }
     return;
   }

   if (!file_system_->ReadFile(filename_.c_str(), &buffer, &null_handler)) {
     // If the file simply doesn't exist, that's a 'successful' read.  It's
     // fine for there to be no cache file and no invalidation data, and thus
     // we swallow file-not-found messages with NullMessageHandler.
     return;
   }
   StringPieceVector lines;
   SplitStringPieceToVector(buffer, "\n", &lines, true);
   int64 timestamp_ms = 0;
   int64 now_ms = timer_->NowMs();

   // The first line should contain the global invalidation timestamp,
   // though we'll just silently leave the invalidation timestamp unchanged
   // if the file was empty.
   if (lines.empty() || !ParseAndValidateTimestamp(lines[0], now_ms,
                                                   &timestamp_ms)) {
     file_parse_failures_->Add(1);
     return;
   }
   purges_from_file->UpdateGlobalInvalidationTimestampMs(timestamp_ms);

   for (int i = 1, n = lines.size(); i < n; ++i) {
     // Each line is in the form "TIMESTAMP_MS URL".  The url
     // has no character restrictions in this file format, and terminates at
     // newline.  The timestamp is a 64-bit decimal number measured in
     // milliseconds since 1970.
     StringPiece line = lines[i];
     if (!line.empty()) {  // skip empty lines but include them in line-count.
       stringpiece_ssize_type pos = line.find(' ');
       if ((pos == StringPiece::npos) ||
           !ParseAndValidateTimestamp(line.substr(0, pos), now_ms,
                                      &timestamp_ms)) {
         file_parse_failures_->Add(1);
       } else {
         StringPiece url = line.substr(pos + 1);
         purges_from_file->Put(url.as_string(), timestamp_ms);
       }
     }
   }
 }

 // While still holding the interprocess lock, verify that the bytes in
 // the file are the ones we wrote.  If another process stole the lock
 // and overwrote our bytes, we'll simply schedule another try, which will
 // merge in whatever results they had written.
 bool PurgeContext::Verify(const GoogleString& expected_purge_file_contents) {
   GoogleString verify;
   return (file_system_->ReadFile(filename_.c_str(), &verify,
                                  message_handler_) &&
           (verify == expected_purge_file_contents));
 }

 void PurgeContext::UpdateCachePurgeFile() {
   // Use a global lock to perform an atomic read/modify/write.
   //
   // Note that the lock is advisory, and if it gets stolen, then we'll
   // have a contended update to the file and potentially lose some
   // purge records, at least temporarily.
   DCHECK(interprocess_lock_->Held());
   DCHECK(waiting_for_interprocess_lock_);
   PurgeSet purges_from_file(max_bytes_in_cache_);
   PurgeSet return_purges(max_bytes_in_cache_);
   PurgeCallbackVector callbacks;
   bool lock_and_update = false;
   bool success = true;
   int failures = 0;

   // Initiate a read/modify/write/verify sequence while holding
   // interprocess_lock_.  Note that during 'modify' we need to
   // also grab mutex_, so we'll need to collect the serizlized
   // buffer and callback-list at the same time for atomicity.
   GoogleString buffer, verify;
   ReadPurgeFile(&purges_from_file);                                   // read
   ModifyPurgeSet(&purges_from_file, &buffer, &callbacks,
                  &return_purges, &failures);                          // modify
   if (!WritePurgeFile(buffer) ||                                      // write
       !Verify(buffer)) {                                              // verify
     contentions_->Add(1);
     success = false;
     HandleWriteFailure(failures, &callbacks, &return_purges, &lock_and_update);
   }

   interprocess_lock_->Unlock();

   if (!callbacks.empty()) {
     if (success) {
       // Induce a file-read the next time PollFileSystem() is called.
       // Note that there is a small chance we might read the same
       // version of the file twice if we get a PollFileSystem()
       // request with 5 seconds since the last check, after writing
       // the file and before this line.  However that redundant read
       // is not harmful.
       purge_index_->Add(1);
     }
     for (int i = 0, n = callbacks.size(); i < n; ++i) {
       callbacks[i]->Run(success, "");
     }
   } else if (lock_and_update) {
     GrabLockAndUpdate();
   }
 }

 void PurgeContext::HandleWriteFailure(int failures,
                                       PurgeCallbackVector* callbacks,
                                       PurgeSet* return_purges,
                                       bool* lock_and_update) {
   ScopedMutex lock(mutex_.get());

   num_consecutive_failures_ += failures + 1;
   if (num_consecutive_failures_ <= kMaxContentionRetries) {
     // Since we relinquished the lock prior to verifying, another
     // thread may have added more purge requests prior to re-acquiring
     // the lock.  In either case we need to restore the callbacks so
     // we can try again and notify interested parties.
     if (waiting_for_interprocess_lock_) {
       DCHECK(!pending_callbacks_.empty());
       pending_callbacks_.insert(pending_callbacks_.end(), callbacks->begin(),
                                 callbacks->end());
       pending_purges_.Merge(*return_purges);
       callbacks->clear();
     } else {
       DCHECK(pending_callbacks_.empty());
       DCHECK(!callbacks->empty());
       waiting_for_interprocess_lock_ = true;
       callbacks->swap(pending_callbacks_);
       pending_purges_.Swap(return_purges);
       *lock_and_update = true;
     }
   } else {
     // Either the file-write failed or the read-verify mismatched.
     //
     // This might be due to a permissions problem, in which case
     // we might never succeed, or it might be a transient issue
     // due to a broken lock contention.  In either case, we will
     // give up and report failure.
     //
     // Note: the file-system itself already reported Info messages on
     // whatever the error was so we don't need another error here, but we'll
     // return a 404 or whatever to the PURGE request.
     //
     // TODO(jmarantz): capture the file-system logs and return those
     // to PURGE clients as the response body.
     file_write_failures_->Add(callbacks->size());
     num_consecutive_failures_ = 0;
   }
 }

 void PurgeContext::ModifyPurgeSet(PurgeSet* purges_from_file,
                                   GoogleString* buffer,
                                   PurgeCallbackVector* return_callbacks,
                                   PurgeSet* return_purges,
                                   int* failures) {
   // Note that while were are reading the file, another Purge might
   // arrive in pending_purges_, protected by mutex_.  We avoid holding
   // the that mutex when reading/writing the file as that might create
   // contention, with IsValid requests in other threads.  But we must
   // hold the mutex during merges.
   ScopedMutex lock(mutex_.get());

   // TODO(jmarantz): as we do not currently merge in purge_set_ as
   // consider the Source Of Truth to be the contents of the file plus
   // the pending purges.  However I would like to add some resilience
   // to file corruption, which might be partially addressed by putting
   // in 'purges_from_file->Merge(purge_set_);' here.  However we
   // should have testcases for how the recovery would work and be
   // transmitted back to the file system and to all servers in
   // a bounded amount of time.

   purges_from_file->Merge(pending_purges_);
   return_purges->Swap(&pending_purges_);
   pending_purges_.Clear();
   waiting_for_interprocess_lock_ = false;
   // Now if another Purge arrives it will have to wait for the next lock
   // again.

   // Collect the write-buffer from the aggregated PurgeSet while we
   // have mutex_ held.
   StrAppend(buffer, Integer64ToString(
       purges_from_file->global_invalidation_timestamp_ms()),
             "\n");
   for (PurgeSet::Iterator p = purges_from_file->Begin(),
            e = purges_from_file->End();
        p != e; ++p) {
     StrAppend(buffer, Integer64ToString(p.Value()), " ", p.Key(), "\n");
   }

   // We must also collect up and return callbacks for the requests we have
   // now merged, since once the lock is released more Purge requests may
   // come in and we don't want to call their callbacks prematurely.
   //
   // Note we can't call these callbacks until we've written the serialized
   // form to disk and verified that the data indeed hit the disk.
   return_callbacks->swap(pending_callbacks_);

   // Same with the number of failures.  They need to be removed from
   // our mutex-protected world now and re-accumulated under mutex if
   // another failure ensues.
   *failures = num_consecutive_failures_;
   num_consecutive_failures_ = 0;
 }

 bool PurgeContext::WritePurgeFile(const GoogleString& buffer) {
   // Atomicly write file so we don't have to acquire the lock to read it.
   file_writes_->Add(1);
   return file_system_->WriteFileAtomic(filename_, buffer, message_handler_);
 }

 // There is a non-zero chance that this thread will be unable to
 // acquire the named-lock in the given time-limit.  We'll just bump up
 // the cancellation count.
 void PurgeContext::CancelCachePurgeFile() {
   PurgeCallbackVector callbacks;
   {
     ScopedMutex lock(mutex_.get());
     callbacks.swap(pending_callbacks_);

     // We are giving up on these pending invalidations rather than having
     // them take effect at some random time in the future.
     pending_purges_.Clear();

     waiting_for_interprocess_lock_ = false;
   }

   // All the purges in the queue failed :(.  Maybe the clients
   // will retry.
   cancellations_->Add(callbacks.size());
   for (int i = 0, n = callbacks.size(); i < n; ++i) {
     callbacks[i]->Run(false, "timeout");
   }
 }

 void PurgeContext::WaitForTimerAndGrabLock() {
   if (request_batching_delay_ms_ == 0) {
     GrabLockAndUpdate();
   } else {
     int64 alarm_time_us =
         timer_->NowUs() + request_batching_delay_ms_ * Timer::kMsUs;
     scheduler_->AddAlarmAtUs(alarm_time_us,
                              MakeFunction(this,
                                           &PurgeContext::GrabLockAndUpdate,
                                           &PurgeContext::CancelCachePurgeFile));
   }
 }

 void PurgeContext::GrabLockAndUpdate() {
   interprocess_lock_->LockTimedWaitStealOld(
       kTimeoutMs, kStealLockAfterMs,
       MakeFunction(this,
                    &PurgeContext::UpdateCachePurgeFile,
                    &PurgeContext::CancelCachePurgeFile));
 }

 void PurgeContext::SetCachePurgeGlobalTimestampMs(
     int64 timestamp_ms, PurgeCallback* callback) {
   bool grab_lock = false;
   {
     ScopedMutex lock(mutex_.get());
     pending_purges_.UpdateGlobalInvalidationTimestampMs(timestamp_ms);
     if (!waiting_for_interprocess_lock_) {
       waiting_for_interprocess_lock_ = true;
       grab_lock = true;
     }
     pending_callbacks_.push_back(callback);
   }
   if (grab_lock) {
     WaitForTimerAndGrabLock();
   }
 }

 void PurgeContext::AddPurgeUrl(StringPiece url, int64 timestamp_ms,
                                PurgeCallback* callback) {
   if (!enable_purge_) {
     callback->Run(false, "EnableCachePurge is off");
   } else {
     bool grab_lock = false;
     {
       ScopedMutex lock(mutex_.get());
       pending_purges_.Put(url.as_string(), timestamp_ms);
       if (!waiting_for_interprocess_lock_) {
         waiting_for_interprocess_lock_ = true;
         grab_lock = true;
       }
       pending_callbacks_.push_back(callback);
     }
     if (grab_lock) {
       WaitForTimerAndGrabLock();
     }
   }
 }

 void PurgeContext::PollFileSystem() {
   int64 now_ms = timer_->NowMs();
   int64 delta_ms = now_ms - purge_poll_timestamp_ms_->Get();
   int64 global_purge_index = purge_index_->Get();
   mutex_->Lock();
   bool needs_update = (local_purge_index_ < global_purge_index);
   if (!reading_ &&
       (needs_update || (delta_ms >= kCheckCacheIntervalMs))) {
     if (needs_update) {
       local_purge_index_ = global_purge_index;
     }
     reading_ = true;
     mutex_->Unlock();
     purge_poll_timestamp_ms_->Set(now_ms);
     ReadFileAndCallCallbackIfChanged(needs_update);
     mutex_->Lock();
     reading_ = false;
   }
   mutex_->Unlock();
 }

 void PurgeContext::ReadFileAndCallCallbackIfChanged(bool needs_update) {
   CopyOnWrite<PurgeSet> purges_from_file;
   PurgeSet* mutable_purges_from_file = purges_from_file.MakeWriteable();
   mutable_purges_from_file->set_max_size(max_bytes_in_cache_);
   bool call_callback = false;

   // Note that we don't hold the global lock while reading the file.
   // But under mutex we have set reading_ so another thread doesn't
   // try a concurrent read.
   DCHECK(reading_);
   ReadPurgeFile(mutable_purges_from_file);

   {
     ScopedMutex lock(mutex_.get());
     if (!purge_set_->Equals(*purges_from_file)) {
       if (!needs_update) {
         // This update was induced by a timeout in this process, rather
         // than by a signal from another process or an UpdateCachePurgeFile
         // in this process.  This signals to other child processes that their
         // local_purge_index_ state is now stale and they must re-read.  We
         // only do this if the purge-set from disk actually changed.
         purge_index_->Add(1);
       }
       purge_set_ = purges_from_file;
       if (update_callback_ != NULL) {
         // We don't want to call the update callback while holding the
         // lock.  Also note that even though we will release the lock
         // before calling the callback, purge_set_ will not be mutated
         // until our caller sets reading_ to false.
         call_callback = true;
       }
     }
   }
   if (call_callback) {
     update_callback_->Run(purges_from_file);
   }

   // Note that we don't merge in pending_purges_ in this flow.  Only
   // when we grab the lock and write the file do we merge in the
   // pending_purges_.
 }

 void PurgeContext::SetUpdateCallback(PurgeSetCallback* cb) {
   update_callback_.reset(cb);
 }

 }  // namespace net_instaweb
	/*
	* Copyright 2013 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: jmarantz@google.com (Joshua Marantz)

	#include "pagespeed/kernel/cache/purge_context.h"

	#include "base/logging.h"
	#include "pagespeed/kernel/base/abstract_mutex.h"
	#include "pagespeed/kernel/base/atomic_bool.h"
	#include "pagespeed/kernel/base/basictypes.h"
	#include "pagespeed/kernel/base/file_system.h"
	#include "pagespeed/kernel/base/function.h"
	#include "pagespeed/kernel/base/message_handler.h"
	#include "pagespeed/kernel/base/null_message_handler.h"
	#include "pagespeed/kernel/base/named_lock_manager.h"
	#include "pagespeed/kernel/base/statistics.h"
	#include "pagespeed/kernel/base/string.h"
	#include "pagespeed/kernel/base/string_util.h"
	#include "pagespeed/kernel/base/thread_system.h"
	#include "pagespeed/kernel/base/time_util.h"
	#include "pagespeed/kernel/cache/lru_cache_base.h"
	#include "pagespeed/kernel/thread/scheduler.h"
	#include "pagespeed/kernel/util/copy_on_write.h"

	namespace net_instaweb {

	namespace {

	const int64 kStealLockAfterMs = 2 * Timer::kSecondMs;
	const int64 kTimeoutMs = 3 * Timer::kSecondMs;
	const int kMaxContentionRetries = 2;

	} // namespace

	const char PurgeContext::kCancellations[] = "purge_cancellations";
	const char PurgeContext::kContentions[] = "purge_contentions";
	const char PurgeContext::kFileParseFailures[] = "purge_file_parse_failures";
	const char PurgeContext::kFileStats[] = "purge_file_stats";
	const char PurgeContext::kFileWriteFailures[] = "purge_file_write_failures";
	const char PurgeContext::kFileWrites[] = "purge_file_writes";
	const char PurgeContext::kPurgeIndex[] = "purge_index";

	// TODO(jmarantz): make it possible to avoid showing this implementation detail
	// in the statistics page.
	const char PurgeContext::kPurgePollTimestampMs[] = "_purge_poll_timestamp_ms";

	namespace {

	// Wrapper class around an UpDownCounter which makes it guaranteed to be
	// locally consistent, even if the backing UpDownCounter is non-functional
	// due to a shared-memory failure or statistics being turned off.
	class BackupUpDownCounter : public UpDownCounter {
	public:
	// Takes ownership of mutex.
	BackupUpDownCounter(UpDownCounter* counter, AbstractMutex* mutex)
	: counter_(counter),
	mutex_(mutex),
	local_counter_(0) {
	}

	virtual int64 Get() const {
	// Don't take the lock for local_counter unless we have actually set some
	// value in the past, yet we can't retrieve it from the real UpDownCounter.
	bool was_set = was_set_.value();
	int64 val = counter_->Get();
	if (was_set && (val == 0)) {
	ScopedMutex lock(mutex_.get());
	val = local_counter_;
	}
	return val;
	}

	virtual StringPiece GetName() const { return counter_->GetName(); }

	virtual void Set(int64 value) {
	counter_->Set(value);
	{
	ScopedMutex lock(mutex_.get());
	local_counter_ = value;
	was_set_.set_value(true);
	}
	}

	virtual int64 AddHelper(int64 delta) {
	LOG(DFATAL) << "AddHelper is not used in PurgeContext";
	return 0;
	}

	private:
	UpDownCounter* counter_;
	scoped_ptr<AbstractMutex> mutex_;
	int64 local_counter_ GUARDED_BY(mutex_);
	AtomicBool was_set_;
	};

	} // namespace

	PurgeContext::PurgeContext(StringPiece filename,
	FileSystem* file_system,
	Timer* timer,
	int max_bytes_in_cache,
	ThreadSystem* thread_system,
	NamedLockManager* lock_manager,
	Scheduler* scheduler,
	Statistics* statistics,
	MessageHandler* handler)
	: filename_(filename.data(), filename.size()),
	interprocess_lock_(lock_manager->CreateNamedLock(LockName())),
	file_system_(file_system),
	timer_(timer),
	statistics_(statistics),
	mutex_(thread_system->NewMutex()),
	pending_purges_(max_bytes_in_cache),
	local_purge_index_(0),
	num_consecutive_failures_(0),
	waiting_for_interprocess_lock_(false),
	reading_(false),
	enable_purge_(true),
	max_bytes_in_cache_(max_bytes_in_cache),
	request_batching_delay_ms_(0),
	cancellations_(statistics->GetVariable(kCancellations)),
	contentions_(statistics->GetVariable(kContentions)),
	file_parse_failures_(statistics->GetVariable(kFileParseFailures)),
	file_stats_(statistics->GetVariable(kFileStats)),
	file_write_failures_(statistics->GetVariable(kFileWriteFailures)),
	file_writes_(statistics->GetVariable(kFileWrites)),
	purge_index_(statistics->GetVariable(kPurgeIndex)),
	purge_poll_timestamp_ms_(new BackupUpDownCounter(
	statistics->GetUpDownCounter(kPurgePollTimestampMs),
	thread_system->NewMutex())),
	scheduler_(scheduler),
	message_handler_(handler) {
	purge_set_.MakeWriteable()->set_max_size(max_bytes_in_cache_);
	}

	PurgeContext::~PurgeContext() {
	}

	void PurgeContext::InitStats(Statistics* statistics) {
	statistics->AddVariable(kCancellations);
	statistics->AddVariable(kContentions);
	statistics->AddVariable(kFileParseFailures);
	statistics->AddVariable(kFileStats);
	statistics->AddVariable(kFileWrites);
	statistics->AddVariable(kFileWriteFailures);
	statistics->AddVariable(kPurgeIndex);
	statistics->AddUpDownCounter(kPurgePollTimestampMs);
	}

	bool PurgeContext::ParseAndValidateTimestamp(
	StringPiece time_string, int64 now_ms, int64* timestamp_ms) {
	if (!StringToInt64(time_string, timestamp_ms)) {
	message_handler_->Info(filename_.c_str(), 1,
	"Invalidation timestamp (%s) not parsed as int64",
	time_string.as_string().c_str());
	return false;
	} else if ((*timestamp_ms != PurgeSet::kInitialTimestampMs) &&
	((*timestamp_ms < 0) \|\|
	(*timestamp_ms > now_ms + PurgeSet::kClockSkewAllowanceMs))) {
	GoogleString converted_time_string;
	ConvertTimeToString(*timestamp_ms, &converted_time_string);
	message_handler_->Info(filename_.c_str(), 1,
	"Invalidation timestamp (%s) in the future: %s",
	time_string.as_string().c_str(),
	converted_time_string.c_str());
	return false;
	}
	return true;
	}

	// Parses the cache purge file.
	void PurgeContext::ReadPurgeFile(PurgeSet* purges_from_file) {
	GoogleString buffer;

	file_stats_->Add(1);
	NullMessageHandler null_handler;

	// Prior to mod_pagespeed 1.8, the cache.flush file's contents were not
	// significant, and generally empty, and only the timestamp of the file
	// itself was important, meaning "wipe everything out of the cache predating
	// that timestamp."
	if (!enable_purge_) {
	int64 timestamp_sec;
	if (file_system_->Mtime(filename_, &timestamp_sec, &null_handler)) {
	int64 timestamp_ms = timestamp_sec * Timer::kSecondMs;
	purges_from_file->UpdateGlobalInvalidationTimestampMs(timestamp_ms);
	}
	return;
	}

	if (!file_system_->ReadFile(filename_.c_str(), &buffer, &null_handler)) {
	// If the file simply doesn't exist, that's a 'successful' read. It's
	// fine for there to be no cache file and no invalidation data, and thus
	// we swallow file-not-found messages with NullMessageHandler.
	return;
	}
	StringPieceVector lines;
	SplitStringPieceToVector(buffer, "\n", &lines, true);
	int64 timestamp_ms = 0;
	int64 now_ms = timer_->NowMs();

	// The first line should contain the global invalidation timestamp,
	// though we'll just silently leave the invalidation timestamp unchanged
	// if the file was empty.
	if (lines.empty() \|\| !ParseAndValidateTimestamp(lines[0], now_ms,
	&timestamp_ms)) {
	file_parse_failures_->Add(1);
	return;
	}
	purges_from_file->UpdateGlobalInvalidationTimestampMs(timestamp_ms);

	for (int i = 1, n = lines.size(); i < n; ++i) {
	// Each line is in the form "TIMESTAMP_MS URL". The url
	// has no character restrictions in this file format, and terminates at
	// newline. The timestamp is a 64-bit decimal number measured in
	// milliseconds since 1970.
	StringPiece line = lines[i];
	if (!line.empty()) { // skip empty lines but include them in line-count.
	stringpiece_ssize_type pos = line.find(' ');
	if ((pos == StringPiece::npos) \|\|
	!ParseAndValidateTimestamp(line.substr(0, pos), now_ms,
	&timestamp_ms)) {
	file_parse_failures_->Add(1);
	} else {
	StringPiece url = line.substr(pos + 1);
	purges_from_file->Put(url.as_string(), timestamp_ms);
	}
	}
	}
	}

	// While still holding the interprocess lock, verify that the bytes in
	// the file are the ones we wrote. If another process stole the lock
	// and overwrote our bytes, we'll simply schedule another try, which will
	// merge in whatever results they had written.
	bool PurgeContext::Verify(const GoogleString& expected_purge_file_contents) {
	GoogleString verify;
	return (file_system_->ReadFile(filename_.c_str(), &verify,
	message_handler_) &&
	(verify == expected_purge_file_contents));
	}

	void PurgeContext::UpdateCachePurgeFile() {
	// Use a global lock to perform an atomic read/modify/write.
	//
	// Note that the lock is advisory, and if it gets stolen, then we'll
	// have a contended update to the file and potentially lose some
	// purge records, at least temporarily.
	DCHECK(interprocess_lock_->Held());
	DCHECK(waiting_for_interprocess_lock_);
	PurgeSet purges_from_file(max_bytes_in_cache_);
	PurgeSet return_purges(max_bytes_in_cache_);
	PurgeCallbackVector callbacks;
	bool lock_and_update = false;
	bool success = true;
	int failures = 0;

	// Initiate a read/modify/write/verify sequence while holding
	// interprocess_lock_. Note that during 'modify' we need to
	// also grab mutex_, so we'll need to collect the serizlized
	// buffer and callback-list at the same time for atomicity.
	GoogleString buffer, verify;
	ReadPurgeFile(&purges_from_file); // read
	ModifyPurgeSet(&purges_from_file, &buffer, &callbacks,
	&return_purges, &failures); // modify
	if (!WritePurgeFile(buffer) \|\| // write
	!Verify(buffer)) { // verify
	contentions_->Add(1);
	success = false;
	HandleWriteFailure(failures, &callbacks, &return_purges, &lock_and_update);
	}

	interprocess_lock_->Unlock();

	if (!callbacks.empty()) {
	if (success) {
	// Induce a file-read the next time PollFileSystem() is called.
	// Note that there is a small chance we might read the same
	// version of the file twice if we get a PollFileSystem()
	// request with 5 seconds since the last check, after writing
	// the file and before this line. However that redundant read
	// is not harmful.
	purge_index_->Add(1);
	}
	for (int i = 0, n = callbacks.size(); i < n; ++i) {
	callbacks[i]->Run(success, "");
	}
	} else if (lock_and_update) {
	GrabLockAndUpdate();
	}
	}

	void PurgeContext::HandleWriteFailure(int failures,
	PurgeCallbackVector* callbacks,
	PurgeSet* return_purges,
	bool* lock_and_update) {
	ScopedMutex lock(mutex_.get());

	num_consecutive_failures_ += failures + 1;
	if (num_consecutive_failures_ <= kMaxContentionRetries) {
	// Since we relinquished the lock prior to verifying, another
	// thread may have added more purge requests prior to re-acquiring
	// the lock. In either case we need to restore the callbacks so
	// we can try again and notify interested parties.
	if (waiting_for_interprocess_lock_) {
	DCHECK(!pending_callbacks_.empty());
	pending_callbacks_.insert(pending_callbacks_.end(), callbacks->begin(),
	callbacks->end());
	pending_purges_.Merge(*return_purges);
	callbacks->clear();
	} else {
	DCHECK(pending_callbacks_.empty());
	DCHECK(!callbacks->empty());
	waiting_for_interprocess_lock_ = true;
	callbacks->swap(pending_callbacks_);
	pending_purges_.Swap(return_purges);
	*lock_and_update = true;
	}
	} else {
	// Either the file-write failed or the read-verify mismatched.
	//
	// This might be due to a permissions problem, in which case
	// we might never succeed, or it might be a transient issue
	// due to a broken lock contention. In either case, we will
	// give up and report failure.
	//
	// Note: the file-system itself already reported Info messages on
	// whatever the error was so we don't need another error here, but we'll
	// return a 404 or whatever to the PURGE request.
	//
	// TODO(jmarantz): capture the file-system logs and return those
	// to PURGE clients as the response body.
	file_write_failures_->Add(callbacks->size());
	num_consecutive_failures_ = 0;
	}
	}

	void PurgeContext::ModifyPurgeSet(PurgeSet* purges_from_file,
	GoogleString* buffer,
	PurgeCallbackVector* return_callbacks,
	PurgeSet* return_purges,
	int* failures) {
	// Note that while were are reading the file, another Purge might
	// arrive in pending_purges_, protected by mutex_. We avoid holding
	// the that mutex when reading/writing the file as that might create
	// contention, with IsValid requests in other threads. But we must
	// hold the mutex during merges.
	ScopedMutex lock(mutex_.get());

	// TODO(jmarantz): as we do not currently merge in purge_set_ as
	// consider the Source Of Truth to be the contents of the file plus
	// the pending purges. However I would like to add some resilience
	// to file corruption, which might be partially addressed by putting
	// in 'purges_from_file->Merge(purge_set_);' here. However we
	// should have testcases for how the recovery would work and be
	// transmitted back to the file system and to all servers in
	// a bounded amount of time.

	purges_from_file->Merge(pending_purges_);
	return_purges->Swap(&pending_purges_);
	pending_purges_.Clear();
	waiting_for_interprocess_lock_ = false;
	// Now if another Purge arrives it will have to wait for the next lock
	// again.

	// Collect the write-buffer from the aggregated PurgeSet while we
	// have mutex_ held.
	StrAppend(buffer, Integer64ToString(
	purges_from_file->global_invalidation_timestamp_ms()),
	"\n");
	for (PurgeSet::Iterator p = purges_from_file->Begin(),
	e = purges_from_file->End();
	p != e; ++p) {
	StrAppend(buffer, Integer64ToString(p.Value()), " ", p.Key(), "\n");
	}

	// We must also collect up and return callbacks for the requests we have
	// now merged, since once the lock is released more Purge requests may
	// come in and we don't want to call their callbacks prematurely.
	//
	// Note we can't call these callbacks until we've written the serialized
	// form to disk and verified that the data indeed hit the disk.
	return_callbacks->swap(pending_callbacks_);

	// Same with the number of failures. They need to be removed from
	// our mutex-protected world now and re-accumulated under mutex if
	// another failure ensues.
	*failures = num_consecutive_failures_;
	num_consecutive_failures_ = 0;
	}

	bool PurgeContext::WritePurgeFile(const GoogleString& buffer) {
	// Atomicly write file so we don't have to acquire the lock to read it.
	file_writes_->Add(1);
	return file_system_->WriteFileAtomic(filename_, buffer, message_handler_);
	}

	// There is a non-zero chance that this thread will be unable to
	// acquire the named-lock in the given time-limit. We'll just bump up
	// the cancellation count.
	void PurgeContext::CancelCachePurgeFile() {
	PurgeCallbackVector callbacks;
	{
	ScopedMutex lock(mutex_.get());
	callbacks.swap(pending_callbacks_);

	// We are giving up on these pending invalidations rather than having
	// them take effect at some random time in the future.
	pending_purges_.Clear();

	waiting_for_interprocess_lock_ = false;
	}

	// All the purges in the queue failed :(. Maybe the clients
	// will retry.
	cancellations_->Add(callbacks.size());
	for (int i = 0, n = callbacks.size(); i < n; ++i) {
	callbacks[i]->Run(false, "timeout");
	}
	}

	void PurgeContext::WaitForTimerAndGrabLock() {
	if (request_batching_delay_ms_ == 0) {
	GrabLockAndUpdate();
	} else {
	int64 alarm_time_us =
	timer_->NowUs() + request_batching_delay_ms_ * Timer::kMsUs;
	scheduler_->AddAlarmAtUs(alarm_time_us,
	MakeFunction(this,
	&PurgeContext::GrabLockAndUpdate,
	&PurgeContext::CancelCachePurgeFile));
	}
	}

	void PurgeContext::GrabLockAndUpdate() {
	interprocess_lock_->LockTimedWaitStealOld(
	kTimeoutMs, kStealLockAfterMs,
	MakeFunction(this,
	&PurgeContext::UpdateCachePurgeFile,
	&PurgeContext::CancelCachePurgeFile));
	}

	void PurgeContext::SetCachePurgeGlobalTimestampMs(
	int64 timestamp_ms, PurgeCallback* callback) {
	bool grab_lock = false;
	{
	ScopedMutex lock(mutex_.get());
	pending_purges_.UpdateGlobalInvalidationTimestampMs(timestamp_ms);
	if (!waiting_for_interprocess_lock_) {
	waiting_for_interprocess_lock_ = true;
	grab_lock = true;
	}
	pending_callbacks_.push_back(callback);
	}
	if (grab_lock) {
	WaitForTimerAndGrabLock();
	}
	}

	void PurgeContext::AddPurgeUrl(StringPiece url, int64 timestamp_ms,
	PurgeCallback* callback) {
	if (!enable_purge_) {
	callback->Run(false, "EnableCachePurge is off");
	} else {
	bool grab_lock = false;
	{
	ScopedMutex lock(mutex_.get());
	pending_purges_.Put(url.as_string(), timestamp_ms);
	if (!waiting_for_interprocess_lock_) {
	waiting_for_interprocess_lock_ = true;
	grab_lock = true;
	}
	pending_callbacks_.push_back(callback);
	}
	if (grab_lock) {
	WaitForTimerAndGrabLock();
	}
	}
	}

	void PurgeContext::PollFileSystem() {
	int64 now_ms = timer_->NowMs();
	int64 delta_ms = now_ms - purge_poll_timestamp_ms_->Get();
	int64 global_purge_index = purge_index_->Get();
	mutex_->Lock();
	bool needs_update = (local_purge_index_ < global_purge_index);
	if (!reading_ &&
	(needs_update \|\| (delta_ms >= kCheckCacheIntervalMs))) {
	if (needs_update) {
	local_purge_index_ = global_purge_index;
	}
	reading_ = true;
	mutex_->Unlock();
	purge_poll_timestamp_ms_->Set(now_ms);
	ReadFileAndCallCallbackIfChanged(needs_update);
	mutex_->Lock();
	reading_ = false;
	}
	mutex_->Unlock();
	}

	void PurgeContext::ReadFileAndCallCallbackIfChanged(bool needs_update) {
	CopyOnWrite<PurgeSet> purges_from_file;
	PurgeSet* mutable_purges_from_file = purges_from_file.MakeWriteable();
	mutable_purges_from_file->set_max_size(max_bytes_in_cache_);
	bool call_callback = false;

	// Note that we don't hold the global lock while reading the file.
	// But under mutex we have set reading_ so another thread doesn't
	// try a concurrent read.
	DCHECK(reading_);
	ReadPurgeFile(mutable_purges_from_file);

	{
	ScopedMutex lock(mutex_.get());
	if (!purge_set_->Equals(*purges_from_file)) {
	if (!needs_update) {
	// This update was induced by a timeout in this process, rather
	// than by a signal from another process or an UpdateCachePurgeFile
	// in this process. This signals to other child processes that their
	// local_purge_index_ state is now stale and they must re-read. We
	// only do this if the purge-set from disk actually changed.
	purge_index_->Add(1);
	}
	purge_set_ = purges_from_file;
	if (update_callback_ != NULL) {
	// We don't want to call the update callback while holding the
	// lock. Also note that even though we will release the lock
	// before calling the callback, purge_set_ will not be mutated
	// until our caller sets reading_ to false.
	call_callback = true;
	}
	}
	}
	if (call_callback) {
	update_callback_->Run(purges_from_file);
	}

	// Note that we don't merge in pending_purges_ in this flow. Only
	// when we grab the lock and write the file do we merge in the
	// pending_purges_.
	}

	void PurgeContext::SetUpdateCallback(PurgeSetCallback* cb) {
	update_callback_.reset(cb);
	}

	} // namespace net_instaweb