diff options
Diffstat (limited to 'net/disk_cache/file_posix.cc')
| -rw-r--r-- | net/disk_cache/file_posix.cc | 273 |
1 files changed, 258 insertions, 15 deletions
diff --git a/net/disk_cache/file_posix.cc b/net/disk_cache/file_posix.cc index a7d1c24..b080f53 100644 --- a/net/disk_cache/file_posix.cc +++ b/net/disk_cache/file_posix.cc @@ -6,9 +6,239 @@ #include <fcntl.h> +#include <set> + #include "base/logging.h" +#include "base/message_loop.h" +#include "base/singleton.h" +#include "base/waitable_event.h" +#include "base/worker_pool.h" #include "net/disk_cache/disk_cache.h" +namespace { + +class InFlightIO; + +// This class represents a single asynchronous IO operation while it is being +// bounced between threads. +class BackgroundIO : public base::RefCountedThreadSafe<BackgroundIO> { + public: + // Other than the actual parameters for the IO operation (including the + // |callback| that must be notified at the end), we need the controller that + // is keeping track of all operations. When done, we notify the controller + // (we do NOT invoke the callback), in the worker thead that completed the + // operation. + BackgroundIO(disk_cache::File* file, const void* buf, size_t buf_len, + size_t offset, disk_cache::FileIOCallback* callback, + InFlightIO* controller) + : io_completed_(true, false), callback_(callback), file_(file), buf_(buf), + buf_len_(buf_len), offset_(offset), controller_(controller), + bytes_(0) {} + + ~BackgroundIO() {} + + // Read and Write are the operations that can be performed asynchronously. + // The actual parameters for the operation are setup in the constructor of + // the object, with the exception of |delete_buffer|, that allows a write + // without a callback. Both methods should be called from a worker thread, by + // posting a task to the WorkerPool (they are RunnableMethods). When finished, + // controller->OnIOComplete() is called. + void Read(); + void Write(bool delete_buffer); + + // This method signals the controller that this operation is finished, in the + // original thread (presumably the IO-Thread). In practice, this is a + // RunableMethod that allows cancellation. + void OnIOSignalled(); + + // Allows the cancellation of the task to notify the controller (step number 7 + // in the diagram below). In practice, if the controller waits for the + // operation to finish it doesn't have to wait for the final task to be + // processed by the message loop so calling this method prevents its delivery. + // Note that this method is not intended to cancel the actual IO operation or + // to prevent the first notification to take place (OnIOComplete). + void Cancel(); + + // Retrieves the number of bytes transfered. + int Result(); + + base::WaitableEvent* io_completed() { + return &io_completed_; + } + + disk_cache::FileIOCallback* callback() { + return callback_; + } + + private: + // An event to signal when the operation completes, and the user callback tha + // has to be invoked. These members are accessed directly by the controller. + base::WaitableEvent io_completed_; + disk_cache::FileIOCallback* callback_; + + scoped_refptr<disk_cache::File> file_; + const void* buf_; + size_t buf_len_; + size_t offset_; + InFlightIO* controller_; // The controller that tracks all operations. + int bytes_; // Final operation result. + + DISALLOW_COPY_AND_ASSIGN(BackgroundIO); +}; + +// This class keeps track of every asynchronous IO operation. A single instance +// of this class is meant to be used to start an asynchronous operation (using +// PostRead/PostWrite). This class will post the operation to a worker thread, +// hanlde the notification when the operation finishes and perform the callback +// on the same thread that was used to start the operation. +// +// The regular sequence of calls is: +// Thread_1 Worker_thread +// 1. InFlightIO::PostRead() +// 2. -> PostTask -> +// 3. BackgroundIO::Read() +// 4. IO operation completes +// 5. InFlightIO::OnIOComplete() +// 6. <- PostTask <- +// 7. BackgroundIO::OnIOSignalled() +// 8. InFlightIO::InvokeCallback() +// 9. invoke callback +// +// Shutdown is a special case that is handled though WaitForPendingIO() instead +// of just waiting for step 7. +class InFlightIO { + public: + InFlightIO() : callback_thread_(MessageLoop::current()) {} + ~InFlightIO() {} + + // These methods start an asynchronous operation. The arguments have the same + // semantics of the File asynchronous operations, with the exception that the + // operation never finishes synchronously. + void PostRead(disk_cache::File* file, void* buf, size_t buf_len, + size_t offset, disk_cache::FileIOCallback* callback); + void PostWrite(disk_cache::File* file, const void* buf, size_t buf_len, + size_t offset, disk_cache::FileIOCallback* callback, + bool delete_buffer); + + // Blocks the current thread until all IO operations tracked by this object + // complete. + void WaitForPendingIO(); + + // Called on a worker thread when |operation| completes. + void OnIOComplete(BackgroundIO* operation); + + // Invokes the users' completion callback at the end of the IO operation. + // |cancel_task| is true if the actual task posted to the thread is still + // queued (because we are inside WaitForPendingIO), and false if said task is + // the one performing the call. + void InvokeCallback(BackgroundIO* operation, bool cancel_task); + + private: + typedef std::set<scoped_refptr<BackgroundIO> > IOList; + + IOList io_list_; // List of pending io operations. + MessageLoop* callback_thread_; +}; + +// --------------------------------------------------------------------------- + +// Runs on a worker thread. +void BackgroundIO::Read() { + if (file_->Read(const_cast<void*>(buf_), buf_len_, offset_)) { + bytes_ = static_cast<int>(buf_len_); + } else { + bytes_ = -1; + } + controller_->OnIOComplete(this); +} + +int BackgroundIO::Result() { + return bytes_; +} + +void BackgroundIO::Cancel() { + DCHECK(controller_); + controller_ = NULL; +} + +// Runs on a worker thread. +void BackgroundIO::Write(bool delete_buffer) { + bool rv = file_->Write(buf_, buf_len_, offset_); + if (delete_buffer) { + // TODO(rvargas): remove or update this code. + delete[] reinterpret_cast<const char*>(buf_); + } + + bytes_ = rv ? static_cast<int>(buf_len_) : -1; + controller_->OnIOComplete(this); +} + +// Runs on the IO thread. +void BackgroundIO::OnIOSignalled() { + if (controller_) + controller_->InvokeCallback(this, false); +} + +// --------------------------------------------------------------------------- + +void InFlightIO::PostRead(disk_cache::File *file, void* buf, size_t buf_len, + size_t offset, disk_cache::FileIOCallback *callback) { + scoped_refptr<BackgroundIO> operation = + new BackgroundIO(file, buf, buf_len, offset, callback, this); + io_list_.insert(operation.get()); + + WorkerPool::PostTask(FROM_HERE, + NewRunnableMethod(operation.get(), &BackgroundIO::Read), + true); +} + +void InFlightIO::PostWrite(disk_cache::File* file, const void* buf, + size_t buf_len, size_t offset, + disk_cache::FileIOCallback* callback, + bool delete_buffer) { + scoped_refptr<BackgroundIO> operation = + new BackgroundIO(file, buf, buf_len, offset, callback, this); + io_list_.insert(operation.get()); + + WorkerPool::PostTask(FROM_HERE, + NewRunnableMethod(operation.get(), &BackgroundIO::Write, + delete_buffer), + true); +} + +void InFlightIO::WaitForPendingIO() { + while (!io_list_.empty()) { + // Block the current thread until all pending IO completes. + IOList::iterator it = io_list_.begin(); + InvokeCallback(*it, true); + } +} + +// Runs on a worker thread. +void InFlightIO::OnIOComplete(BackgroundIO* operation) { + callback_thread_->PostTask(FROM_HERE, + NewRunnableMethod(operation, + &BackgroundIO::OnIOSignalled)); + operation->io_completed()->Signal(); +} + +// Runs on the IO thread. +void InFlightIO::InvokeCallback(BackgroundIO* operation, bool cancel_task) { + operation->io_completed()->Wait(); + + if (cancel_task) + operation->Cancel(); + + disk_cache::FileIOCallback* callback = operation->callback(); + int bytes = operation->Result(); + + // Release the reference acquired in PostRead / PostWrite. + io_list_.erase(operation); + callback->OnFileIOComplete(bytes); +} + +} // namespace + namespace disk_cache { File::File(base::PlatformFile file) @@ -71,19 +301,31 @@ 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) { + if (completed) + *completed = true; + return Read(buffer, buffer_len, offset); + } + if (buffer_len > ULONG_MAX || offset > ULONG_MAX) return false; - // TODO: Implement async IO. - bool ret = Read(buffer, buffer_len, offset); - if (ret && completed) - *completed = true; - return ret; + InFlightIO* io_operations = Singleton<InFlightIO>::get(); + io_operations->PostRead(this, buffer, buffer_len, offset, callback); + + *completed = false; + return true; } bool File::Write(const void* buffer, size_t buffer_len, size_t offset, FileIOCallback* callback, bool* completed) { DCHECK(init_); + if (!callback) { + if (completed) + *completed = true; + return Write(buffer, buffer_len, offset); + } + return AsyncWrite(buffer, buffer_len, offset, true, callback, completed); } @@ -98,17 +340,12 @@ bool File::AsyncWrite(const void* buffer, size_t buffer_len, size_t offset, if (buffer_len > ULONG_MAX || offset > ULONG_MAX) return false; - // TODO: Implement async IO. - bool ret = Write(buffer, buffer_len, offset); - if (ret && completed) - *completed = true; - - // If we supply our own async callback, and the caller is not asking to be - // notified when completed, we are supposed to delete the buffer. - if (ret && !callback && !notify) - delete[] reinterpret_cast<const char*>(buffer); + InFlightIO* io_operations = Singleton<InFlightIO>::get(); + io_operations->PostWrite(this, buffer, buffer_len, offset, callback, !notify); - return ret; + if (completed) + *completed = false; + return true; } bool File::SetLength(size_t length) { @@ -125,4 +362,10 @@ size_t File::GetLength() { return ret; } +// Static. +void File::WaitForPendingIO(int* num_pending_io) { + if (*num_pending_io) + Singleton<InFlightIO>::get()->WaitForPendingIO(); +} + } // namespace disk_cache |