Switch MessagePumpForIO to use completion ports on Windows.

Cleanup the separation between MessagePumpForUI and 
MessagePumpForIO, and convert the latter to use Completion
Ports instead of MsgWaitForMultipleobjects to sleep
when idle.

Remove all traces of Windows messages from MessagePumpForIO,
remove the transitional API of completion port notifications
and remove WatchObject API.

Modify all callers of RegisterIOHandler so that they are no
longer using RegisterIOContext, and also handle properly
the new semantics of completion ports (notifications even when
the IO completes immediately).

Add a new interface to allow proper cleanup of disk cache (to
replace code that was waiting for pending APCs from the destructor).

Add a way for the message pump to perform cleanup of abandoned IO.


BUG=B/1344358, 3497, 3630
TESt=unit tests
R=darin


Review URL: http://codereview.chromium.org/8156

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@5021 0039d316-1c4b-4281-b951-d872f2087c98

7 files changed, 150 insertions, 199 deletions

diff --git a/net/disk_cache/cache_util_win.cc b/net/disk_cache/cache_util_win.cc
index 377cbbf..adb9d8a5 100644
--- a/net/disk_cache/cache_util_win.cc
+++ b/net/disk_cache/cache_util_win.cc
@@ -7,6 +7,7 @@
 #include <windows.h>
 
 #include "base/logging.h"
+#include "base/message_loop.h"
 #include "base/scoped_handle.h"
 #include "base/file_util.h"
 
@@ -38,6 +39,9 @@ void DeleteFiles(const wchar_t* path, const wchar_t* search_name) {
 
 namespace disk_cache {
 
+// Implemented in file_win.cc.
+MessageLoopForIO::IOHandler* GetFileIOHandler();
+
 bool MoveCache(const std::wstring& from_path, const std::wstring& to_path) {
   // I don't want to use the shell version of move because if something goes
   // wrong, that version will attempt to move file by file and fail at the end.
@@ -63,12 +67,8 @@ bool DeleteCacheFile(const std::wstring& name) {
 void WaitForPendingIO(int* num_pending_io) {
   while (*num_pending_io) {
     // Asynchronous IO operations may be in flight and the completion may end
-    // up calling us back so let's wait for them (we need an alertable wait).
-    // The idea is to let other threads do usefull work and at the same time
-    // allow more than one IO to finish... 20 mS later, we process all queued
-    // APCs and see if we have to repeat the wait.
-    Sleep(20);
-    SleepEx(0, TRUE);
+    // up calling us back so let's wait for them.
+    MessageLoopForIO::current()->WaitForIOCompletion(100, GetFileIOHandler());
   }
 }
 
diff --git a/net/disk_cache/disk_cache_perftest.cc b/net/disk_cache/disk_cache_perftest.cc
index c160677..33cad54 100644
--- a/net/disk_cache/disk_cache_perftest.cc
+++ b/net/disk_cache/disk_cache_perftest.cc
@@ -9,6 +9,7 @@
 #include "base/basictypes.h"
 #include "base/file_util.h"
 #include "base/perftimer.h"
+#include "base/platform_test.h"
 #if defined(OS_WIN)
 #include "base/scoped_handle.h"
 #endif
@@ -17,7 +18,6 @@
 #include "net/base/net_errors.h"
 #include "net/disk_cache/block_files.h"
 #include "net/disk_cache/disk_cache.h"
-#include "net/disk_cache/disk_cache_test_base.h"
 #include "net/disk_cache/disk_cache_test_util.h"
 #include "net/disk_cache/hash.h"
 #include "testing/gtest/include/gtest/gtest.h"
@@ -28,6 +28,8 @@ extern int g_cache_tests_max_id;
 extern volatile int g_cache_tests_received;
 extern volatile bool g_cache_tests_error;
 
+typedef PlatformTest DiskCacheTest;
+
 namespace {
 
 bool EvictFileFromSystemCache(const wchar_t* name) {
@@ -226,6 +228,7 @@ TEST_F(DiskCacheTest, CacheBackendPerformance) {
   int ret = TimeWrite(num_entries, cache, &entries);
   EXPECT_EQ(ret, g_cache_tests_received);
 
+  MessageLoop::current()->RunAllPending();
   delete cache;
 
   std::wstring filename(path);
@@ -257,6 +260,7 @@ TEST_F(DiskCacheTest, CacheBackendPerformance) {
   ret = TimeRead(num_entries, cache, entries, false);
   EXPECT_EQ(ret, g_cache_tests_received);
 
+  MessageLoop::current()->RunAllPending();
   delete cache;
 }
 
@@ -266,6 +270,7 @@ TEST_F(DiskCacheTest, CacheBackendPerformance) {
 // fragmented, or if we have multiple files. This test measures that scenario,
 // by using multiple, highly fragmented files.
 TEST_F(DiskCacheTest, BlockFilesPerformance) {
+  MessageLoopForIO message_loop;
   std::wstring path = GetCachePath();
   ASSERT_TRUE(DeleteCache(path.c_str()));
 
@@ -303,4 +308,5 @@ TEST_F(DiskCacheTest, BlockFilesPerformance) {
   }
 
   timer2.Done();
+  MessageLoop::current()->RunAllPending();
 }
diff --git a/net/disk_cache/disk_cache_test_base.cc b/net/disk_cache/disk_cache_test_base.cc
index 417792f..0a3e07b 100644
--- a/net/disk_cache/disk_cache_test_base.cc
+++ b/net/disk_cache/disk_cache_test_base.cc
@@ -8,6 +8,10 @@
 #include "net/disk_cache/disk_cache_test_util.h"
 #include "net/disk_cache/mem_backend_impl.h"
 
+void DiskCacheTest::TearDown() {
+  MessageLoop::current()->RunAllPending();
+}
+
 void DiskCacheTestWithCache::SetMaxSize(int size) {
   size_ = size;
   if (cache_impl_)
@@ -73,6 +77,7 @@ void DiskCacheTestWithCache::InitDiskCache() {
 
 
 void DiskCacheTestWithCache::TearDown() {
+  MessageLoop::current()->RunAllPending();
   delete cache_;
 
   if (!memory_only_) {
diff --git a/net/disk_cache/disk_cache_test_base.h b/net/disk_cache/disk_cache_test_base.h
index ca56be1f..6b75361 100644
--- a/net/disk_cache/disk_cache_test_base.h
+++ b/net/disk_cache/disk_cache_test_base.h
@@ -9,12 +9,6 @@
 #include "base/platform_test.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
-// These tests can use the path service, which uses autoreleased objects on the
-// Mac, so this needs to be a PlatformTest.  Even tests that do not require a
-// cache (and that do not need to be a DiskCacheTestWithCache) are susceptible
-// to this problem; all such tests should use TEST_F(DiskCacheTest, ...).
-typedef PlatformTest DiskCacheTest;
-
 namespace disk_cache {
 
 class Backend;
@@ -23,6 +17,14 @@ class MemBackendImpl;
 
 }  // namespace disk_cache
 
+// These tests can use the path service, which uses autoreleased objects on the
+// Mac, so this needs to be a PlatformTest.  Even tests that do not require a
+// cache (and that do not need to be a DiskCacheTestWithCache) are susceptible
+// to this problem; all such tests should use TEST_F(DiskCacheTest, ...).
+class DiskCacheTest : public PlatformTest {
+  virtual void TearDown();
+};
+
 // Provides basic support for cache related tests.
 class DiskCacheTestWithCache : public DiskCacheTest {
  protected:
diff --git a/net/disk_cache/entry_impl.cc b/net/disk_cache/entry_impl.cc
index 844ff40..2821540 100644
--- a/net/disk_cache/entry_impl.cc
+++ b/net/disk_cache/entry_impl.cc
@@ -16,25 +16,8 @@ using base::TimeDelta;
 
 namespace {
 
-// This is a simple Task to execute the callback (from the message loop instead
-// of the APC).
-class InvokeCallback : public Task {
- public:
-  InvokeCallback(net::CompletionCallback* callback, int argument)
-      : callback_(callback), argument_(argument) {}
-
-  virtual void Run() {
-    callback_->Run(argument_);
-  }
-
- private:
-  net::CompletionCallback* callback_;
-  int argument_;
-  DISALLOW_EVIL_CONSTRUCTORS(InvokeCallback);
-};
-
-// This class implements FileIOCallback to buffer the callback from an IO
-// operation from the actual IO class.
+// This class implements FileIOCallback to buffer the callback from a file IO
+// operation from the actual net class.
 class SyncCallback: public disk_cache::FileIOCallback {
  public:
   SyncCallback(disk_cache::EntryImpl* entry,
@@ -57,10 +40,8 @@ class SyncCallback: public disk_cache::FileIOCallback {
 void SyncCallback::OnFileIOComplete(int bytes_copied) {
   entry_->DecrementIoCount();
   entry_->Release();
-  if (callback_) {
-    InvokeCallback* task = new InvokeCallback(callback_, bytes_copied);
-    MessageLoop::current()->PostTask(FROM_HERE, task);
-  }
+  if (callback_)
+    callback_->Run(bytes_copied);
   delete this;
 }
 
@@ -556,9 +537,11 @@ void EntryImpl::DeleteData(Addr address, int index) {
     if (files_[index])
       files_[index] = NULL;  // Releases the object.
 
-    if (!DeleteCacheFile(backend_->GetFileName(address)))
+    if (!DeleteCacheFile(backend_->GetFileName(address))) {
+      UMA_HISTOGRAM_COUNTS(L"DiskCache.DeleteFailed", 1);
       LOG(ERROR) << "Failed to delete " << backend_->GetFileName(address) <<
                     " from the cache.";
+    }
   } else {
     backend_->DeleteBlock(address, true);
   }
@@ -711,7 +694,6 @@ bool EntryImpl::ImportSeparateFile(int index, int offset, int buf_len) {
   return true;
 }
 
-
 // The common scenario is that this is called from the destructor of the entry,
 // to write to disk what we have buffered. We don't want to hold the destructor
 // until the actual IO finishes, so we'll send an asynchronous write that will
@@ -744,6 +726,13 @@ bool EntryImpl::Flush(int index, int size, bool async) {
   if (!file)
     return false;
 
+  // TODO(rvargas): figure out if it's worth to re-enable posting operations.
+  // Right now it is only used from GrowUserBuffer, not the destructor, and
+  // it is not accounted for from the point of view of the total number of
+  // pending operations of the cache. It is also racing with the actual write
+  // on the GrowUserBuffer path because there is no code to exclude the range
+  // that is going to be written.
+  async = false;
   if (async) {
     if (!file->PostWrite(user_buffers_[index].get(), len, offset))
       return false;
diff --git a/net/disk_cache/file_win.cc b/net/disk_cache/file_win.cc
index 0e8e3f3..bc89975 100644
--- a/net/disk_cache/file_win.cc
+++ b/net/disk_cache/file_win.cc
@@ -4,62 +4,39 @@
 
 #include "net/disk_cache/file.h"
 
+#include "base/message_loop.h"
+#include "base/singleton.h"
 #include "net/disk_cache/disk_cache.h"
 
 namespace {
 
-// This class implements FileIOCallback to perform IO operations
-// when the callback parameter of the operation is NULL.
-class SyncCallback: public disk_cache::FileIOCallback {
- public:
-  SyncCallback() : called_(false) {}
-  ~SyncCallback() {}
-
-  virtual void OnFileIOComplete(int bytes_copied);
-  void WaitForResult(int* bytes_copied);
- private:
-  bool called_;
-  int actual_;
-};
-
-void SyncCallback::OnFileIOComplete(int bytes_copied) {
-  actual_ = bytes_copied;
-  called_ = true;
-}
-
-// Waits for the IO operation to complete.
-void SyncCallback::WaitForResult(int* bytes_copied) {
-  for (;;) {
-    SleepEx(INFINITE, TRUE);
-    if (called_)
-      break;
-  }
-  *bytes_copied = actual_;
-}
-
 // Structure used for asynchronous operations.
 struct MyOverlapped {
-  OVERLAPPED overlapped;
-  disk_cache::File* file;
-  disk_cache::FileIOCallback* callback;
-  const void* buffer;
-  DWORD actual_bytes;
-  bool async;  // Invoke the callback form the completion.
-  bool called;  // Completion received.
-  bool delete_buffer;  // Delete the user buffer at completion.
-};
+  MyOverlapped(disk_cache::File* file, size_t offset,
+               disk_cache::FileIOCallback* callback);
+  ~MyOverlapped();
+  OVERLAPPED* overlapped() {
+    return &context_.overlapped;
+  }
 
-COMPILE_ASSERT(!offsetof(MyOverlapped, overlapped), starts_with_overlapped);
+  MessageLoopForIO::IOContext context_;
+  scoped_refptr<disk_cache::File> file_;
+  disk_cache::FileIOCallback* callback_;
+  const void* buffer_;
+  bool delete_buffer_;  // Delete the user buffer at completion.
+};
 
-}  // namespace
+COMPILE_ASSERT(!offsetof(MyOverlapped, context_), starts_with_overlapped);
 
-namespace disk_cache {
+// Helper class to handle the IO completion notifications from the message loop.
+class CompletionHandler : public MessageLoopForIO::IOHandler {
+  virtual void OnIOCompleted(MessageLoopForIO::IOContext* context,
+                             DWORD actual_bytes, DWORD error);
+};
 
-// SyncCallback to be invoked as an APC when the asynchronous operation
-// completes.
-void CALLBACK IoCompletion(DWORD error, DWORD actual_bytes,
-                           OVERLAPPED* overlapped) {
-  MyOverlapped* data = reinterpret_cast<MyOverlapped*>(overlapped);
+void CompletionHandler::OnIOCompleted(MessageLoopForIO::IOContext* context,
+                                      DWORD actual_bytes, DWORD error) {
+  MyOverlapped* data = reinterpret_cast<MyOverlapped*>(context);
 
   if (error) {
     DCHECK(!actual_bytes);
@@ -67,29 +44,39 @@ void CALLBACK IoCompletion(DWORD error, DWORD actual_bytes,
     NOTREACHED();
   }
 
-  if (data->delete_buffer) {
-    DCHECK(!data->callback);
-    data->file->Release();
-    delete data->buffer;
-    delete data;
-    return;
-  }
+  if (data->callback_)
+    data->callback_->OnFileIOComplete(static_cast<int>(actual_bytes));
 
-  if (data->async) {
-    data->callback->OnFileIOComplete(static_cast<int>(actual_bytes));
-    data->file->Release();
-    delete data;
-  } else {
-    // Somebody is waiting for this so don't delete data and instead notify
-    // that we were called.
-    data->actual_bytes = actual_bytes;
-    data->file->Release();
-    data->called = true;
+  delete data;
+}
+
+MyOverlapped::MyOverlapped(disk_cache::File* file, size_t offset,
+                           disk_cache::FileIOCallback* callback) {
+  memset(this, 0, sizeof(*this));
+  context_.handler = Singleton<CompletionHandler>::get();
+  context_.overlapped.Offset = static_cast<DWORD>(offset);
+  file_ = file;
+  callback_ = callback;
+}
+
+MyOverlapped::~MyOverlapped() {
+  if (delete_buffer_) {
+    DCHECK(!callback_);
+    delete buffer_;
   }
 }
 
+}  // namespace
+
+namespace disk_cache {
+
+// Used from WaitForPendingIO() when the cache is being destroyed.
+MessageLoopForIO::IOHandler* GetFileIOHandler() {
+  return Singleton<CompletionHandler>::get();
+}
+
 File::File(base::PlatformFile file)
-    : init_(true), mixed_(true), platform_file_(INVALID_HANDLE_VALUE),
+    : init_(true), platform_file_(INVALID_HANDLE_VALUE),
       sync_platform_file_(file) {
 }
 
@@ -99,23 +86,22 @@ bool File::Init(const std::wstring& name) {
     return false;
 
   platform_file_ = CreateFile(name.c_str(), GENERIC_READ | GENERIC_WRITE,
-                        FILE_SHARE_READ | FILE_SHARE_WRITE, NULL, OPEN_EXISTING,
-                        FILE_FLAG_OVERLAPPED, NULL);
+                              FILE_SHARE_READ | FILE_SHARE_WRITE, NULL,
+                              OPEN_EXISTING, FILE_FLAG_OVERLAPPED, NULL);
 
   if (INVALID_HANDLE_VALUE == platform_file_)
     return false;
 
+  MessageLoopForIO::current()->RegisterIOHandler(
+      platform_file_, Singleton<CompletionHandler>::get());
+
   init_ = true;
-  if (mixed_) {
-    sync_platform_file_ = CreateFile(name.c_str(), GENERIC_READ | GENERIC_WRITE,
-                                FILE_SHARE_READ | FILE_SHARE_WRITE, NULL,
-                                OPEN_EXISTING, 0, NULL);
-
-    if (INVALID_HANDLE_VALUE == sync_platform_file_)
-      return false;
-  } else {
-    sync_platform_file_ = INVALID_HANDLE_VALUE;
-  }
+  sync_platform_file_  = CreateFile(name.c_str(), GENERIC_READ | GENERIC_WRITE,
+                                    FILE_SHARE_READ | FILE_SHARE_WRITE, NULL,
+                                    OPEN_EXISTING, 0, NULL);
+
+  if (INVALID_HANDLE_VALUE == sync_platform_file_)
+    return false;
 
   return true;
 }
@@ -126,13 +112,13 @@ File::~File() {
 
   if (INVALID_HANDLE_VALUE != platform_file_)
     CloseHandle(platform_file_);
-  if (mixed_ && INVALID_HANDLE_VALUE != sync_platform_file_)
+  if (INVALID_HANDLE_VALUE != sync_platform_file_)
     CloseHandle(sync_platform_file_);
 }
 
 base::PlatformFile File::platform_file() const {
   DCHECK(init_);
-  return (INVALID_HANDLE_VALUE == platform_file_) ? sync_platform_file_ : 
+  return (INVALID_HANDLE_VALUE == platform_file_) ? sync_platform_file_ :
                                                     platform_file_;
 }
 
@@ -145,13 +131,11 @@ bool File::IsValid() const {
 
 bool File::Read(void* buffer, size_t buffer_len, size_t offset) {
   DCHECK(init_);
-  if (!mixed_ || buffer_len > ULONG_MAX || offset > LONG_MAX)
+  if (buffer_len > ULONG_MAX || offset > LONG_MAX)
     return false;
 
-  DWORD ret = SetFilePointer(sync_platform_file_, 
-                             static_cast<LONG>(offset), 
-                             NULL,
-                             FILE_BEGIN);
+  DWORD ret = SetFilePointer(sync_platform_file_, static_cast<LONG>(offset),
+                             NULL, FILE_BEGIN);
   if (INVALID_SET_FILE_POINTER == ret)
     return false;
 
@@ -164,13 +148,11 @@ bool File::Read(void* buffer, size_t buffer_len, size_t offset) {
 
 bool File::Write(const void* buffer, size_t buffer_len, size_t offset) {
   DCHECK(init_);
-  if (!mixed_ || buffer_len > ULONG_MAX || offset > ULONG_MAX)
+  if (buffer_len > ULONG_MAX || offset > ULONG_MAX)
     return false;
 
-  DWORD ret = SetFilePointer(sync_platform_file_, 
-                             static_cast<LONG>(offset), 
-                             NULL,
-                             FILE_BEGIN);
+  DWORD ret = SetFilePointer(sync_platform_file_, static_cast<LONG>(offset),
+                             NULL, FILE_BEGIN);
   if (INVALID_SET_FILE_POINTER == ret)
     return false;
 
@@ -187,55 +169,38 @@ bool File::Write(const void* buffer, size_t buffer_len, size_t offset) {
 bool File::Read(void* buffer, size_t buffer_len, size_t offset,
                 FileIOCallback* callback, bool* completed) {
   DCHECK(init_);
+  if (!callback)
+    return Read(buffer, buffer_len, offset);
+
   if (buffer_len > ULONG_MAX || offset > ULONG_MAX)
     return false;
 
-  MyOverlapped* data = new MyOverlapped;
-  memset(data, 0, sizeof(*data));
-
-  SyncCallback local_callback;
-  data->overlapped.Offset = static_cast<DWORD>(offset);
-  data->callback = callback ? callback : &local_callback;
-  data->file = this;
-
+  MyOverlapped* data = new MyOverlapped(this, offset, callback);
   DWORD size = static_cast<DWORD>(buffer_len);
-  AddRef();
 
-  if (!ReadFileEx(platform_file_, buffer, size, &data->overlapped, 
-                  &IoCompletion)) {
-    Release();
+  DWORD actual;
+  if (!ReadFile(platform_file_, buffer, size, &actual, data->overlapped())) {
+    *completed = false;
+    if (GetLastError() == ERROR_IO_PENDING)
+      return true;
     delete data;
     return false;
   }
 
-  if (callback) {
-    *completed = false;
-    // Let's check if the operation is already finished.
-    SleepEx(0, TRUE);
-    if (data->called) {
-      *completed = (data->actual_bytes == size);
-      DCHECK(data->actual_bytes == size);
-      delete data;
-      return *completed;
-    }
-    data->async = true;
-  } else {
-    // Invoke the callback and perform cleanup on the APC.
-    data->async = true;
-    int bytes_copied;
-    local_callback.WaitForResult(&bytes_copied);
-    if (static_cast<int>(buffer_len) != bytes_copied) {
-      NOTREACHED();
-      return false;
-    }
-  }
-
-  return true;
+  // The operation completed already. We'll be called back anyway.
+  *completed = (actual == size);
+  DCHECK(actual == size);
+  data->callback_ = NULL;
+  data->file_ = NULL;  // There is no reason to hold on to this anymore.
+  return *completed;
 }
 
 bool File::Write(const void* buffer, size_t buffer_len, size_t offset,
                  FileIOCallback* callback, bool* completed) {
   DCHECK(init_);
+  if (!callback)
+    return Write(buffer, buffer_len, offset);
+
   return AsyncWrite(buffer, buffer_len, offset, true, callback, completed);
 }
 
@@ -250,50 +215,32 @@ bool File::AsyncWrite(const void* buffer, size_t buffer_len, size_t offset,
   if (buffer_len > ULONG_MAX || offset > ULONG_MAX)
     return false;
 
-  MyOverlapped* data = new MyOverlapped;
-  memset(data, 0, sizeof(*data));
-
-  SyncCallback local_callback;
-  data->overlapped.Offset = static_cast<DWORD>(offset);
-  data->callback = callback ? callback : &local_callback;
-  data->file = this;
-  if (!callback && !notify) {
-    data->delete_buffer = true;
-    data->callback = NULL;
-    data->buffer = buffer;
+  MyOverlapped* data = new MyOverlapped(this, offset, callback);
+  bool dummy_completed;
+  if (!callback) {
+    DCHECK(!notify);
+    data->delete_buffer_ = true;
+    data->buffer_ = buffer;
+    completed = &dummy_completed;
   }
 
   DWORD size = static_cast<DWORD>(buffer_len);
-  AddRef();
 
-  if (!WriteFileEx(platform_file_, buffer, size, &data->overlapped,
-                   &IoCompletion)) {
-    Release();
+  DWORD actual;
+  if (!WriteFile(platform_file_, buffer, size, &actual, data->overlapped())) {
+    *completed = false;
+    if (GetLastError() == ERROR_IO_PENDING)
+      return true;
     delete data;
     return false;
   }
 
-  if (callback) {
-    *completed = false;
-    SleepEx(0, TRUE);
-    if (data->called) {
-      *completed = (data->actual_bytes == size);
-      DCHECK(data->actual_bytes == size);
-      delete data;
-      return *completed;
-    }
-    data->async = true;
-  } else if (notify) {
-    data->async = true;
-    int bytes_copied;
-    local_callback.WaitForResult(&bytes_copied);
-    if (static_cast<int>(buffer_len) != bytes_copied) {
-      NOTREACHED();
-      return false;
-    }
-  }
-
-  return true;
+  // The operation completed already. We'll be called back anyway.
+  *completed = (actual == size);
+  DCHECK(actual == size);
+  data->callback_ = NULL;
+  data->file_ = NULL;  // There is no reason to hold on to this anymore.
+  return *completed;
 }
 
 bool File::SetLength(size_t length) {
diff --git a/net/disk_cache/mapped_file_unittest.cc b/net/disk_cache/mapped_file_unittest.cc
index fe090e5..28eeff5 100644
--- a/net/disk_cache/mapped_file_unittest.cc
+++ b/net/disk_cache/mapped_file_unittest.cc
@@ -95,6 +95,8 @@ TEST_F(DiskCacheTest, MappedFile_AsyncIO) {
   g_cache_tests_max_id = 0;
   g_cache_tests_received = 0;
 
+  MessageLoopHelper helper;
+
   char buffer1[20];
   char buffer2[20];
   CacheTestFillBuffer(buffer1, sizeof(buffer1), false);
@@ -105,14 +107,14 @@ TEST_F(DiskCacheTest, MappedFile_AsyncIO) {
   int expected = completed ? 0 : 1;
 
   g_cache_tests_max_id = 1;
-  WaitForCallbacks(expected);
+  helper.WaitUntilCacheIoFinished(expected);
 
   EXPECT_TRUE(file->Read(buffer2, sizeof(buffer2), 1024 * 1024, &callback,
               &completed));
   if (!completed)
     expected++;
 
-  WaitForCallbacks(expected);
+  helper.WaitUntilCacheIoFinished(expected);
 
   EXPECT_EQ(expected, g_cache_tests_received);
   EXPECT_FALSE(g_cache_tests_error);