Disk cache: Update the disk cache tools and tests to use...

net/tools/dump_cache/upgrade.cc

 // Copyright (c) 2010 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 "base/file_path.h"
 #include "base/logging.h"
 #include "base/message_loop.h"
 #include "base/scoped_ptr.h"
 #include "base/string_util.h"
+#include "base/thread.h"
 #include "googleurl/src/gurl.h"
 #include "net/base/io_buffer.h"
+#include "net/base/test_completion_callback.h"
 #include "net/disk_cache/backend_impl.h"
 #include "net/disk_cache/entry_impl.h"
 #include "net/http/http_cache.h"
 #include "net/http/http_response_headers.h"
 #include "net/http/http_response_info.h"
 #include "net/tools/dump_cache/cache_dumper.h"
 
 namespace {
 
 const wchar_t kPipePrefix[] = L"\\\\.\\pipe\\dump_cache_";
@@ skipping 80 matching lines @@
   RESULT_OK = 0,
   RESULT_UNKNOWN_COMMAND,
   RESULT_INVALID_PARAMETER,
   RESULT_NAME_OVERFLOW
 };
 
 // -----------------------------------------------------------------------
 
 class BaseSM : public MessageLoopForIO::IOHandler {
  public:
-  BaseSM(HANDLE channel);
+  explicit BaseSM(HANDLE channel);
   virtual ~BaseSM();
 
  protected:
   bool SendMsg(const Message& msg);
   bool ReceiveMsg();
   bool ConnectChannel();
   bool IsPending();
 
   MessageLoopForIO::IOContext in_context_;
   MessageLoopForIO::IOContext out_context_;
   disk_cache::EntryImpl* entry_;
   HANDLE channel_;
   int state_;
   int pending_count_;
   scoped_array<char> in_buffer_;
   scoped_array<char> out_buffer_;
   IoBuffer* input_;
   IoBuffer* output_;
+  base::Thread cache_thread_;
+
   DISALLOW_COPY_AND_ASSIGN(BaseSM);
 };
 
 BaseSM::BaseSM(HANDLE channel)
-      : entry_(NULL), channel_(channel), state_(0), pending_count_(0) {
+      : entry_(NULL), channel_(channel), state_(0), pending_count_(0),
+        cache_thread_("cache") {
   in_buffer_.reset(new char[kChannelSize]);
   out_buffer_.reset(new char[kChannelSize]);
   input_ = reinterpret_cast<IoBuffer*>(in_buffer_.get());
   output_ = reinterpret_cast<IoBuffer*>(out_buffer_.get());
 
   memset(&in_context_, 0, sizeof(in_context_));
   memset(&out_context_, 0, sizeof(out_context_));
   in_context_.handler = this;
   out_context_.handler = this;
   MessageLoopForIO::current()->RegisterIOHandler(channel_, this);
+  CHECK(cache_thread_.StartWithOptions(
+            base::Thread::Options(MessageLoop::TYPE_IO, 0)));
 }
 
 BaseSM::~BaseSM() {
   if (entry_)
     entry_->Close();
 }
 
 bool BaseSM::SendMsg(const Message& msg) {
   // Only one command will be in-flight at a time. Let's start the Read IO here
   // when we know that it will be pending.
@@ skipping 37 matching lines @@
 }
 
 bool BaseSM::IsPending() {
   return pending_count_ != 0;
 }
 
 // -----------------------------------------------------------------------
 
 class MasterSM : public BaseSM {
  public:
-   MasterSM(const std::wstring& path, HANDLE channel, bool dump_to_disk)
-      : BaseSM(channel), path_(path), dump_to_disk_(dump_to_disk) {
+  MasterSM(const std::wstring& path, HANDLE channel, bool dump_to_disk)
+      : BaseSM(channel), path_(path), dump_to_disk_(dump_to_disk),
+        ALLOW_THIS_IN_INITIALIZER_LIST(
+            write_callback_(this, &MasterSM::DoReadDataComplete)) {
   }
   virtual ~MasterSM() {
     delete writer_;
   }
 
   bool DoInit();
   virtual void OnIOCompleted(MessageLoopForIO::IOContext* context,
                              DWORD bytes_transfered, DWORD error);
 
  private:
@@ skipping 11 matching lines @@
 
   void SendGetPrevEntry();
   void DoGetEntry();
   void DoGetKey(int bytes_read);
   void DoGetUseTimes();
   void SendGetDataSize();
   void DoGetDataSize();
   void CloseEntry();
   void SendReadData();
   void DoReadData(int bytes_read);
+  void DoReadDataComplete(int ret);
   void SendQuit();
   void DoEnd();
   void Fail();
 
   base::Time last_used_;
   base::Time last_modified_;
   int64 remote_entry_;
   int stream_;
   int bytes_remaining_;
   int offset_;
   int copied_entries_;
-  scoped_ptr<disk_cache::BackendImpl> cache_;
+  int read_size_;
+  scoped_ptr<disk_cache::Backend> cache_;
   CacheDumpWriter* writer_;
   const std::wstring& path_;
   bool dump_to_disk_;
+  net::CompletionCallbackImpl<MasterSM> write_callback_;
 };
 
 void MasterSM::OnIOCompleted(MessageLoopForIO::IOContext* context,
                              DWORD bytes_transfered, DWORD error) {
   pending_count_--;
   if (context == &out_context_) {
     if (!error)
       return;
     return Fail();
   }
@@ skipping 34 matching lines @@
   }
 }
 
 bool MasterSM::DoInit() {
   DEBUGMSG("Master DoInit\n");
   DCHECK(state_ == MASTER_INITIAL);
 
   if (dump_to_disk_) {
     writer_ = new DiskDumper(path_);
   } else {
-    cache_.reset(new disk_cache::BackendImpl(FilePath::FromWStringHack(path_)));
-    if (!cache_->Init()) {
+    disk_cache::Backend* cache;
+    TestCompletionCallback cb;
+    int rv = disk_cache::CreateCacheBackend(net::DISK_CACHE,
+                                            FilePath::FromWStringHack(path_), 0,
+                                            false,
+                                            cache_thread_.message_loop_proxy(),
+                                            &cache, &cb);
+    if (cb.GetResult(rv) != net::OK) {
       printf("Unable to initialize new files\n");
       return false;
     }
+    cache_.reset(cache);
     writer_ = new CacheDumper(cache_.get());
   }
   if (!writer_)
     return false;
 
   copied_entries_ = 0;
   remote_entry_ = 0;
 
   if (ConnectChannel()) {
     SendGetPrevEntry();
@@ skipping 147 matching lines @@
 
   int read_size = input_->msg.buffer_bytes;
   if (!read_size) {
     printf("Read failed, entry \"%s\" truncated!\n", entry_->GetKey().c_str());
     bytes_remaining_ = 0;
     return SendReadData();
   }
 
   scoped_refptr<net::WrappedIOBuffer> buf =
       new net::WrappedIOBuffer(input_->buffer);
-  if (!writer_->WriteEntry(entry_, stream_, offset_, buf, read_size))
+  int rv = writer_->WriteEntry(entry_, stream_, offset_, buf, read_size,
+                               &write_callback_);
+  if (rv == net::ERR_IO_PENDING) {
+    // We'll continue in DoReadDataComplete.
+    read_size_ = read_size;
+    return;
+  }
+
+  if (rv <= 0)
     return Fail();
 
   offset_ += read_size;
   bytes_remaining_ -= read_size;
   // Read some more.
   SendReadData();
 }
 
+void MasterSM::DoReadDataComplete(int ret) {
+  if (ret != read_size_)
+    return Fail();
+
+  offset_ += ret;
+  bytes_remaining_ -= ret;
+  // Read some more.
+  SendReadData();
+}
+
 void MasterSM::SendQuit() {
   DEBUGMSG("Master SendQuit\n");
   state_ = MASTER_END;
   Message msg;
   msg.command = QUIT;
   SendMsg(msg);
   if (!IsPending())
     DoEnd();
 }
 
 void MasterSM::DoEnd() {
   DEBUGMSG("Master DoEnd\n");
   MessageLoop::current()->PostTask(FROM_HERE, new MessageLoop::QuitTask());
 }
 
 void MasterSM::Fail() {
   DEBUGMSG("Master Fail\n");
   printf("Unexpected failure\n");
   SendQuit();
 }
 
 // -----------------------------------------------------------------------
 
 class SlaveSM : public BaseSM {
  public:
-  SlaveSM(const std::wstring& path, HANDLE channel)
-      : BaseSM(channel), iterator_(NULL) {
-    cache_.reset(new disk_cache::BackendImpl(FilePath::FromWStringHack(path)));
-    if (!cache_->Init()) {
-      printf("Unable to open cache files\n");
-      return;
-    }
-    cache_->SetUpgradeMode();
-  }
+  SlaveSM(const std::wstring& path, HANDLE channel);
   virtual ~SlaveSM();
 
   bool DoInit();
   virtual void OnIOCompleted(MessageLoopForIO::IOContext* context,
                              DWORD bytes_transfered, DWORD error);
 
  private:
   enum {
     SLAVE_INITIAL = 0,
     SLAVE_WAITING,
     SLAVE_END
   };
 
   void DoGetNextEntry();
   void DoGetPrevEntry();
   int32 GetEntryFromList();
   void DoCloseEntry();
   void DoGetKey();
   void DoGetUseTimes();
   void DoGetDataSize();
   void DoReadData();
+  void DoReadDataComplete(int ret);
   void DoEnd();
   void Fail();
 
   void* iterator_;
+  Message msg_;  // Only used for DoReadDataComplete.
 
+  net::CompletionCallbackImpl<SlaveSM> read_callback_;
   scoped_ptr<disk_cache::BackendImpl> cache_;
 };
 
+SlaveSM::SlaveSM(const std::wstring& path, HANDLE channel)
+    : BaseSM(channel), iterator_(NULL),
+      ALLOW_THIS_IN_INITIALIZER_LIST(
+          read_callback_(this, &SlaveSM::DoReadDataComplete)) {
+  disk_cache::Backend* cache;
+  TestCompletionCallback cb;
+  int rv = disk_cache::CreateCacheBackend(net::DISK_CACHE,
+                                          FilePath::FromWStringHack(path), 0,
+                                          false,
+                                          cache_thread_.message_loop_proxy(),
+                                          &cache, &cb);
+  if (cb.GetResult(rv) != net::OK) {
+    printf("Unable to open cache files\n");
+    return;
+  }
+  cache_.reset(reinterpret_cast<disk_cache::BackendImpl*>(cache));
+  cache_->SetUpgradeMode();
+}
+
 SlaveSM::~SlaveSM() {
   if (iterator_)
     cache_->EndEnumeration(&iterator_);
 }
 
 void SlaveSM::OnIOCompleted(MessageLoopForIO::IOContext* context,
                             DWORD bytes_transfered, DWORD error) {
   pending_count_--;
   if (state_ == SLAVE_END) {
     if (IsPending())
@@ skipping 187 matching lines @@
   msg.command = READ_DATA;
 
   int stream = input_->msg.arg1;
   int size = input_->msg.arg2;
   if (!entry_ || input_->msg.long_arg1 != reinterpret_cast<int64>(entry_) ||
       stream < 0 || stream > 1 || size > kBufferSize) {
     msg.result =  RESULT_INVALID_PARAMETER;
   } else {
     scoped_refptr<net::WrappedIOBuffer> buf =
         new net::WrappedIOBuffer(output_->buffer);
-    int ret = entry_->ReadData(stream, input_->msg.arg3, buf, size, NULL);
+    int ret = entry_->ReadData(stream, input_->msg.arg3, buf, size,
+                               &read_callback_);
+    if (ret == net::ERR_IO_PENDING) {
+      // Save the message so we can continue were we left.
+      msg_ = msg;
+      return;
+    }
 
     msg.buffer_bytes = (ret < 0) ? 0 : ret;
     msg.result = RESULT_OK;
   }
   SendMsg(msg);
 }
 
+void SlaveSM::DoReadDataComplete(int ret) {
+  DEBUGMSG("\t\t\tSlave DoReadDataComplete\n");
+  DCHECK_EQ(READ_DATA, msg_.command);
+  msg_.buffer_bytes = (ret < 0) ? 0 : ret;
+  msg_.result = RESULT_OK;
+  SendMsg(msg_);
+}
+
 void SlaveSM::DoEnd() {
   DEBUGMSG("\t\t\tSlave DoEnd\n");
   MessageLoop::current()->PostTask(FROM_HERE, new MessageLoop::QuitTask());
 }
 
 void SlaveSM::Fail() {
   DEBUGMSG("\t\t\tSlave Fail\n");
   printf("Unexpected failure\n");
   state_ = SLAVE_END;
   if (IsPending()) {
@@ skipping 46 matching lines @@
 
   SlaveSM slave(input_path, pipe);
   if (!slave.DoInit()) {
     printf("Unable to talk with the main process\n");
     return -1;
   }
 
   loop.Run();
   return 0;
 }