src/pagespeed/kernel/cache/async_cache.cc

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

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

#include "base/logging.h"
#include "pagespeed/kernel/base/atomic_bool.h"
#include "pagespeed/kernel/base/atomic_int32.h"
#include "pagespeed/kernel/base/function.h"
#include "pagespeed/kernel/base/shared_string.h"
#include "pagespeed/kernel/base/string.h"
#include "pagespeed/kernel/base/string_util.h"
#include "pagespeed/kernel/cache/cache_interface.h"
#include "pagespeed/kernel/cache/key_value_codec.h"
#include "pagespeed/kernel/thread/queued_worker_pool.h"

namespace net_instaweb {

AsyncCache::AsyncCache(CacheInterface* cache, QueuedWorkerPool* pool)
    : cache_(cache) {
  CHECK(cache->IsBlocking());
  sequence_ = pool->NewSequence();
  sequence_->set_max_queue_size(kMaxQueueSize);
}

AsyncCache::~AsyncCache() {
  DCHECK_EQ(0, outstanding_operations());
}

GoogleString AsyncCache::FormatName(StringPiece cache) {
  return StrCat("Async(", cache, ")");
}

void AsyncCache::Get(const GoogleString& key, Callback* callback) {
  if (IsHealthy()) {
    outstanding_operations_.NoBarrierIncrement(1);
    sequence_->Add(MakeFunction(this, &AsyncCache::DoGet,
                                &AsyncCache::CancelGet,
                                new GoogleString(key), callback));
  } else {
    ValidateAndReportResult(key, CacheInterface::kNotFound, callback);
  }
}

void AsyncCache::MultiGet(MultiGetRequest* request) {
  outstanding_operations_.NoBarrierIncrement(1);
  if (IsHealthy()) {
    sequence_->Add(MakeFunction(this, &AsyncCache::DoMultiGet,
                                &AsyncCache::CancelMultiGet, request));
  } else {
    CancelMultiGet(request);
  }
}

void AsyncCache::DoGet(GoogleString* key, Callback* callback) {
  if (IsHealthy()) {
    cache_->Get(*key, callback);
    delete key;
    outstanding_operations_.NoBarrierIncrement(-1);
  } else {
    CancelGet(key, callback);
  }
}

void AsyncCache::CancelGet(GoogleString* key, Callback* callback) {
  ValidateAndReportResult(*key, CacheInterface::kNotFound, callback);
  delete key;
  outstanding_operations_.NoBarrierIncrement(-1);
}

void AsyncCache::DoMultiGet(MultiGetRequest* request) {
  if (IsHealthy()) {
    cache_->MultiGet(request);
    outstanding_operations_.NoBarrierIncrement(-1);
  } else {
    CancelMultiGet(request);
  }
}

void AsyncCache::CancelMultiGet(MultiGetRequest* request) {
  ReportMultiGetNotFound(request);
  outstanding_operations_.NoBarrierIncrement(-1);
}

void AsyncCache::Put(const GoogleString& key, SharedString* value) {
  if (IsHealthy()) {
    // If the cache will encode the key into the value during Put,
    // then instead do it inline now, not in sequence_, as
    // SharedString::Append can mutate the shared value storage in a
    // thread-unsafe way.
    if (cache_->MustEncodeKeyInValueOnPut()) {
      SharedString* encoded_value = new SharedString;
      if (!key_value_codec::Encode(key, value, encoded_value)) {
        delete encoded_value;
        return;
      }
      value = encoded_value;
    } else {
      value = new SharedString(*value);
    }

    outstanding_operations_.NoBarrierIncrement(1);
    sequence_->Add(
        MakeFunction(this, &AsyncCache::DoPut, &AsyncCache::CancelPut,
                     new GoogleString(key), value));
  }
}

void AsyncCache::DoPut(GoogleString* key, SharedString* value) {
  if (IsHealthy()) {
    // TODO(jmarantz): Start timers at the beginning of each operation,
    // particularly this one, and use long delays as a !IsHealthy signal.
    if (cache_->MustEncodeKeyInValueOnPut()) {
      cache_->PutWithKeyInValue(*key, value);
    } else {
      cache_->Put(*key, value);
    }
  }
  delete key;
  delete value;
  outstanding_operations_.NoBarrierIncrement(-1);
}

void AsyncCache::CancelPut(GoogleString* key, SharedString* value) {
  delete key;
  delete value;
  outstanding_operations_.NoBarrierIncrement(-1);
}

void AsyncCache::Delete(const GoogleString& key) {
  if (IsHealthy()) {
    outstanding_operations_.NoBarrierIncrement(1);
    sequence_->Add(MakeFunction(this, &AsyncCache::DoDelete,
                                &AsyncCache::CancelDelete,
                                new GoogleString(key)));
  }
}

void AsyncCache::DoDelete(GoogleString* key) {
  if (IsHealthy()) {
    cache_->Delete(*key);
  }
  delete key;
  outstanding_operations_.NoBarrierIncrement(-1);
}

void AsyncCache::CancelDelete(GoogleString* key) {
  outstanding_operations_.NoBarrierIncrement(-1);
  delete key;
}

void AsyncCache::ShutDown() {
  stopped_.set_value(true);
  sequence_->CancelPendingFunctions();

  // Note that though we've canceled pending functions, the cache might be
  // still be busy with a function -- say if it's blocked on a wedged memcached.
  //
  // So we can't Disable it until it quiesces.  The only way out from a
  // completely wedged system is kill -9.  Other solutions likely cause
  // core dumps.
  sequence_->Add(MakeFunction(cache_, &CacheInterface::ShutDown));
}

}  // namespace net_instaweb