Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(64)

Unified Diff: net/base/file_stream_posix.cc

Issue 10701050: net: Implement canceling of all async operations in FileStream. (Closed) Base URL: https://src.chromium.org/chrome/trunk/src/
Patch Set: Created 8 years, 5 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View side-by-side diff with in-line comments
Download patch
Index: net/base/file_stream_posix.cc
===================================================================
--- net/base/file_stream_posix.cc (revision 145483)
+++ net/base/file_stream_posix.cc (working copy)
@@ -20,12 +20,12 @@
#include "base/eintr_wrapper.h"
#include "base/file_path.h"
#include "base/logging.h"
+#include "base/memory/ref_counted.h"
#include "base/message_loop.h"
#include "base/metrics/histogram.h"
#include "base/string_util.h"
#include "base/threading/thread_restrictions.h"
#include "base/threading/worker_pool.h"
-#include "base/synchronization/waitable_event.h"
#include "net/base/file_stream_metrics.h"
#include "net/base/file_stream_net_log_parameters.h"
#include "net/base/io_buffer.h"
@@ -49,320 +49,486 @@
FROM_CURRENT == SEEK_CUR &&
FROM_END == SEEK_END, whence_matches_system);
-namespace {
+class FileStreamPosix::AsyncContext {
+ public:
+ explicit AsyncContext(const BoundNetLog& bound_net_log);
+ AsyncContext(base::PlatformFile file, const BoundNetLog& bound_net_log);
-int RecordAndMapError(int error,
- FileErrorSource source,
- bool record_uma,
- const net::BoundNetLog& bound_net_log) {
- net::Error net_error = MapSystemError(error);
+ // Destroys the context. It can be deleted in the method or deletion can be
+ // deferred to WorkerPool if some asynchronous operation is now in progress
+ // or if auto-closing is needed.
+ void Destroy();
- bound_net_log.AddEvent(
- net::NetLog::TYPE_FILE_STREAM_ERROR,
- base::Bind(&NetLogFileStreamErrorCallback,
- source, error, net_error));
+ bool record_uma() { return record_uma_; }
+ void set_record_uma(bool value) { record_uma_ = value; }
+ base::PlatformFile file() { return file_; }
+ bool auto_closed() { return auto_closed_; }
+ void set_auto_closed(bool value) { auto_closed_ = value; }
+ bool async_in_progress() { return async_in_progress_; }
- RecordFileError(error, source, record_uma);
+ int RecordAndMapError(int error, FileErrorSource source);
- return net_error;
+ // Sync and async versions of all operations
+ void OpenAsync(const FilePath& path,
+ int open_flags,
+ const CompletionCallback& callback);
+ int OpenSync(const FilePath& path, int open_flags);
+
+ void CloseAsync(const CompletionCallback& callback);
+ void CloseSync();
+
+ void SeekAsync(Whence whence,
+ int64 offset,
+ const Int64CompletionCallback& callback);
+ int64 SeekSync(Whence whence, int64 offset);
+
+ void ReadAsync(IOBuffer* buf,
+ int buf_len,
+ const CompletionCallback& callback);
+ int ReadSync(char* buf, int buf_len);
+
+ void WriteAsync(IOBuffer* in_buf,
+ int buf_len,
+ const CompletionCallback& callback);
+ int WriteSync(const char* in_buf, int buf_len);
+
+ private:
+ // Map system error into network error code and log it with |bound_net_log_|.
+ // Method should be called with |net_log_lock_| locked.
+ int MapAndLogError(int error, FileErrorSource source);
+
+ // Opens a file with some network logging.
+ // The result code is written to |result|.
+ void OpenFileImpl(const FilePath& path, int open_flags, int* result);
+
+ // Closes a file with some network logging.
+ void CloseFileImpl();
+
+ // Adjusts the position from where the data is read.
+ void SeekFileImpl(Whence whence, int64 offset, int64* result);
+
+ // ReadFile() is a simple wrapper around read() that handles EINTR signals
+ // and calls RecordAndMapError() to map errno to net error codes.
+ void ReadFileImpl(scoped_refptr<IOBuffer> buf, int buf_len, int* result);
+
+ // WriteFile() is a simple wrapper around write() that handles EINTR signals
+ // and calls MapSystemError() to map errno to net error codes. It tries
+ // to write to completion.
+ void WriteFileImpl(scoped_refptr<IOBuffer> buf, int buf_len, int* result);
+
+ // Called when asynchronous Close(), Open(), Read(), Write() or Seek()
+ // is completed. |result| contains the result or a network error code.
+ template <typename R>
+ void OnAsyncCompleted(const base::Callback<void(R)>& callback, R* result);
+
+ // Delete the context with asynchronous closing if necessary.
+ void DeleteAbandoned();
+
+ base::PlatformFile file_;
+ bool auto_closed_;
+ bool record_uma_;
+ bool async_in_progress_;
+ bool destroyed_;
+ base::Lock net_log_lock_;
+ BoundNetLog bound_net_log_;
+};
+
+
+FileStreamPosix::AsyncContext::AsyncContext(const BoundNetLog& bound_net_log)
+ : file_(base::kInvalidPlatformFileValue),
+ auto_closed_(true),
+ record_uma_(false),
+ async_in_progress_(false),
+ destroyed_(false),
+ bound_net_log_(bound_net_log) {
}
-// Opens a file with some network logging.
-// The opened file and the result code are written to |file| and |result|.
-void OpenFile(const FilePath& path,
- int open_flags,
- bool record_uma,
- base::PlatformFile* file,
- int* result,
- const net::BoundNetLog& bound_net_log) {
- std::string file_name = path.AsUTF8Unsafe();
- bound_net_log.BeginEvent(
- net::NetLog::TYPE_FILE_STREAM_OPEN,
- NetLog::StringCallback("file_name", &file_name));
+FileStreamPosix::AsyncContext::AsyncContext(base::PlatformFile file,
+ const BoundNetLog& bound_net_log)
+ : file_(file),
+ auto_closed_(false),
+ record_uma_(false),
+ async_in_progress_(false),
+ destroyed_(false),
+ bound_net_log_(bound_net_log) {
+}
- *result = OK;
- *file = base::CreatePlatformFile(path, open_flags, NULL, NULL);
- if (*file == base::kInvalidPlatformFileValue) {
- bound_net_log.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
- *result = RecordAndMapError(errno, FILE_ERROR_SOURCE_OPEN, record_uma,
- bound_net_log);
+void FileStreamPosix::AsyncContext::Destroy() {
+ {
+ // By locking we don't allow any operation with |bound_net_log_| to be
+ // in progress while this method is executed. Attempt to do something
+ // with |bound_net_log_| will be done either before this method or after
+ // we switch |context_->destroyed_| which will prohibit any operation on
+ // |bound_net_log_|.
+ base::AutoLock locked(net_log_lock_);
+ destroyed_ = true;
}
+ if (!async_in_progress_)
+ DeleteAbandoned();
}
-// Opens a file using OpenFile() and then signals the completion.
-void OpenFileAndSignal(const FilePath& path,
- int open_flags,
- bool record_uma,
- base::PlatformFile* file,
- int* result,
- base::WaitableEvent* on_io_complete,
- const net::BoundNetLog& bound_net_log) {
- OpenFile(path, open_flags, record_uma, file, result, bound_net_log);
- on_io_complete->Signal();
+int FileStreamPosix::AsyncContext::RecordAndMapError(int error,
+ FileErrorSource source) {
+ int net_error;
+ {
+ base::AutoLock locked(net_log_lock_);
+ net_error = MapAndLogError(error, source);
+ }
+ RecordFileError(error, source, record_uma_);
+ return net_error;
}
-// Closes a file with some network logging.
-void CloseFile(base::PlatformFile file,
- const net::BoundNetLog& bound_net_log) {
- bound_net_log.AddEvent(net::NetLog::TYPE_FILE_STREAM_CLOSE);
- if (file == base::kInvalidPlatformFileValue)
- return;
+void FileStreamPosix::AsyncContext::OpenAsync(
+ const FilePath& path,
+ int open_flags,
+ const CompletionCallback& callback) {
+ DCHECK(!async_in_progress_);
- if (!base::ClosePlatformFile(file))
- NOTREACHED();
- bound_net_log.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
+ int* result = new int(OK);
+ const bool posted = base::WorkerPool::PostTaskAndReply(
+ FROM_HERE,
+ base::Bind(&AsyncContext::OpenFileImpl, base::Unretained(this),
+ path, open_flags, result),
+ base::Bind(&AsyncContext::OnAsyncCompleted<int>,
+ base::Unretained(this),
+ callback, base::Owned(result)),
+ true /* task_is_slow */);
+ DCHECK(posted);
+
+ async_in_progress_ = true;
}
-// Closes a file with CloseFile() and signals the completion.
-void CloseFileAndSignal(base::PlatformFile* file,
- base::WaitableEvent* on_io_complete,
- const net::BoundNetLog& bound_net_log) {
- CloseFile(*file, bound_net_log);
- *file = base::kInvalidPlatformFileValue;
- on_io_complete->Signal();
+int FileStreamPosix::AsyncContext::OpenSync(const FilePath& path,
+ int open_flags) {
+ int result = OK;
+ OpenFileImpl(path, open_flags, &result);
+ return result;
}
-// Adjusts the position from where the data is read.
-void SeekFile(base::PlatformFile file,
- Whence whence,
- int64 offset,
- int64* result,
- bool record_uma,
- const net::BoundNetLog& bound_net_log) {
- off_t res = lseek(file, static_cast<off_t>(offset),
- static_cast<int>(whence));
- if (res == static_cast<off_t>(-1)) {
- *result = RecordAndMapError(errno,
- FILE_ERROR_SOURCE_SEEK,
- record_uma,
- bound_net_log);
- return;
- }
- *result = res;
+void FileStreamPosix::AsyncContext::CloseAsync(
+ const CompletionCallback& callback) {
+ DCHECK(!async_in_progress_);
+
+ // Value OK will never be changed in AsyncContext::CloseFile() and is needed
+ // here just to use the same AsyncContext::OnAsyncCompleted().
+ int* result = new int(OK);
+ const bool posted = base::WorkerPool::PostTaskAndReply(
+ FROM_HERE,
+ base::Bind(&AsyncContext::CloseFileImpl, base::Unretained(this)),
+ base::Bind(&AsyncContext::OnAsyncCompleted<int>,
+ base::Unretained(this),
+ callback, base::Owned(result)),
+ true /* task_is_slow */);
+ DCHECK(posted);
+
+ async_in_progress_ = true;
}
-// Seeks a file by calling SeekSync() and signals the completion.
-void SeekFileAndSignal(base::PlatformFile file,
- Whence whence,
- int64 offset,
- int64* result,
- bool record_uma,
- base::WaitableEvent* on_io_complete,
- const net::BoundNetLog& bound_net_log) {
- SeekFile(file, whence, offset, result, record_uma, bound_net_log);
- on_io_complete->Signal();
+void FileStreamPosix::AsyncContext::CloseSync() {
+ DCHECK(!async_in_progress_);
+ CloseFileImpl();
}
-// ReadFile() is a simple wrapper around read() that handles EINTR signals and
-// calls MapSystemError() to map errno to net error codes.
-void ReadFile(base::PlatformFile file,
- char* buf,
- int buf_len,
- bool record_uma,
- int* result,
- const net::BoundNetLog& bound_net_log) {
- base::ThreadRestrictions::AssertIOAllowed();
- // read(..., 0) returns 0 to indicate end-of-file.
+void FileStreamPosix::AsyncContext::SeekAsync(
+ Whence whence,
+ int64 offset,
+ const Int64CompletionCallback& callback) {
+ DCHECK(!async_in_progress_);
- // Loop in the case of getting interrupted by a signal.
- ssize_t res = HANDLE_EINTR(read(file, buf, static_cast<size_t>(buf_len)));
- if (res == -1) {
- res = RecordAndMapError(errno, FILE_ERROR_SOURCE_READ,
- record_uma, bound_net_log);
+ int64* result = new int64(-1);
+ const bool posted = base::WorkerPool::PostTaskAndReply(
+ FROM_HERE,
+ base::Bind(&AsyncContext::SeekFileImpl, base::Unretained(this),
+ whence, offset, result),
+ base::Bind(&AsyncContext::OnAsyncCompleted<int64>,
+ base::Unretained(this),
+ callback, base::Owned(result)),
+ true /* task is slow */);
+ DCHECK(posted);
+
+ async_in_progress_ = true;
+}
+
+int64 FileStreamPosix::AsyncContext::SeekSync(Whence whence, int64 offset) {
+ off_t result = -1;
+ SeekFileImpl(whence, offset, &result);
+ return result;
+}
+
+void FileStreamPosix::AsyncContext::ReadAsync(
+ IOBuffer* in_buf,
+ int buf_len,
+ const CompletionCallback& callback) {
+ DCHECK(!async_in_progress_);
+
+ int* result = new int(OK);
+ scoped_refptr<IOBuffer> buf = in_buf;
+ const bool posted = base::WorkerPool::PostTaskAndReply(
+ FROM_HERE,
+ base::Bind(&AsyncContext::ReadFileImpl, base::Unretained(this),
+ buf, buf_len, result),
+ base::Bind(&AsyncContext::OnAsyncCompleted<int>,
+ base::Unretained(this),
+ callback, base::Owned(result)),
+ true /* task is slow */);
+ DCHECK(posted);
+
+ async_in_progress_ = true;
+}
+
+int FileStreamPosix::AsyncContext::ReadSync(char* in_buf, int buf_len) {
+ int result = OK;
+ scoped_refptr<IOBuffer> buf = new WrappedIOBuffer(in_buf);
+ ReadFileImpl(buf, buf_len, &result);
+ return result;
+}
+
+void FileStreamPosix::AsyncContext::WriteAsync(
+ IOBuffer* in_buf,
+ int buf_len,
+ const CompletionCallback& callback) {
+ DCHECK(!async_in_progress_);
+
+ int* result = new int(OK);
+ scoped_refptr<IOBuffer> buf = in_buf;
+ const bool posted = base::WorkerPool::PostTaskAndReply(
+ FROM_HERE,
+ base::Bind(&AsyncContext::WriteFileImpl, base::Unretained(this),
+ buf, buf_len, result),
+ base::Bind(&AsyncContext::OnAsyncCompleted<int>,
+ base::Unretained(this),
+ callback, base::Owned(result)),
+ true /* task is slow */);
+ DCHECK(posted);
+
+ async_in_progress_ = true;
+}
+
+int FileStreamPosix::AsyncContext::WriteSync(const char* in_buf, int buf_len) {
+ int result = OK;
+ scoped_refptr<IOBuffer> buf = new WrappedIOBuffer(in_buf);
+ WriteFileImpl(buf, buf_len, &result);
+ return result;
+}
+
+int FileStreamPosix::AsyncContext::MapAndLogError(int error,
+ FileErrorSource source) {
+ net_log_lock_.AssertAcquired();
+ // The following check is against incorrect use or bug. File descriptor
+ // shouldn't ever be closed outside of FileStream while it still tries to do
+ // something with it.
+ DCHECK(error != EBADF);
+ net::Error net_error = MapSystemError(error);
+
+ if (!destroyed_) {
+ bound_net_log_.AddEvent(
+ net::NetLog::TYPE_FILE_STREAM_ERROR,
+ base::Bind(&NetLogFileStreamErrorCallback,
+ source, error, net_error));
}
- *result = res;
+
+ return net_error;
}
-// Reads a file using ReadFile() and signals the completion.
-void ReadFileAndSignal(base::PlatformFile file,
- scoped_refptr<IOBuffer> buf,
- int buf_len,
- bool record_uma,
- int* result,
- base::WaitableEvent* on_io_complete,
- const net::BoundNetLog& bound_net_log) {
- ReadFile(file, buf->data(), buf_len, record_uma, result, bound_net_log);
- on_io_complete->Signal();
+void FileStreamPosix::AsyncContext::OpenFileImpl(const FilePath& path,
+ int open_flags,
+ int* result) {
+ std::string file_name = path.AsUTF8Unsafe();
+ {
+ base::AutoLock locked(net_log_lock_);
+ // Bail out quickly if operation was already canceled
+ if (destroyed_)
+ return;
+
+ bound_net_log_.BeginEvent(
+ net::NetLog::TYPE_FILE_STREAM_OPEN,
+ NetLog::StringCallback("file_name", &file_name));
+ }
+
+ *result = OK;
+ file_ = base::CreatePlatformFile(path, open_flags, NULL, NULL);
+ if (file_ == base::kInvalidPlatformFileValue) {
+ int error = errno;
+ {
+ base::AutoLock locked(net_log_lock_);
+ if (!destroyed_)
+ bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
+ *result = MapAndLogError(error, FILE_ERROR_SOURCE_OPEN);
+ }
+ RecordFileError(error, FILE_ERROR_SOURCE_OPEN, record_uma_);
+ }
}
-// WriteFile() is a simple wrapper around write() that handles EINTR signals and
-// calls MapSystemError() to map errno to net error codes. It tries to write to
-// completion.
-void WriteFile(base::PlatformFile file,
- const char* buf,
- int buf_len,
- bool record_uma,
- int* result,
- const net::BoundNetLog& bound_net_log) {
+void FileStreamPosix::AsyncContext::CloseFileImpl() {
+ {
+ base::AutoLock locked(net_log_lock_);
+ if (!destroyed_)
+ bound_net_log_.AddEvent(net::NetLog::TYPE_FILE_STREAM_CLOSE);
+ }
+
+ if (file_ == base::kInvalidPlatformFileValue)
+ return;
+ if (!base::ClosePlatformFile(file_))
+ NOTREACHED();
+ file_ = base::kInvalidPlatformFileValue;
+
+ {
+ base::AutoLock locked(net_log_lock_);
+ if (!destroyed_)
+ bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
+ }
+}
+
+void FileStreamPosix::AsyncContext::SeekFileImpl(Whence whence,
+ int64 offset,
+ int64* result) {
base::ThreadRestrictions::AssertIOAllowed();
- ssize_t res = HANDLE_EINTR(write(file, buf, buf_len));
- if (res == -1) {
- res = RecordAndMapError(errno, FILE_ERROR_SOURCE_WRITE, record_uma,
- bound_net_log);
- }
+ // If context has been already destroyed nobody waits for operation results.
+ if (destroyed_)
+ return;
+
+ off_t res = lseek(file_, static_cast<off_t>(offset),
+ static_cast<int>(whence));
+ if (res == static_cast<off_t>(-1))
+ res = RecordAndMapError(errno, FILE_ERROR_SOURCE_SEEK);
*result = res;
}
-// Writes a file using WriteFile() and signals the completion.
-void WriteFileAndSignal(base::PlatformFile file,
- scoped_refptr<IOBuffer> buf,
- int buf_len,
- bool record_uma,
- int* result,
- base::WaitableEvent* on_io_complete,
- const net::BoundNetLog& bound_net_log) {
- WriteFile(file, buf->data(), buf_len, record_uma, result, bound_net_log);
- on_io_complete->Signal();
+void FileStreamPosix::AsyncContext::ReadFileImpl(scoped_refptr<IOBuffer> buf,
+ int buf_len,
+ int* result) {
+ base::ThreadRestrictions::AssertIOAllowed();
+
+ // If context has been already destroyed nobody waits for operation results.
+ if (destroyed_)
+ return;
+
+ // Loop in the case of getting interrupted by a signal.
+ ssize_t res = HANDLE_EINTR(read(file_, buf->data(),
+ static_cast<size_t>(buf_len)));
+ if (res == -1)
+ res = RecordAndMapError(errno, FILE_ERROR_SOURCE_READ);
+ *result = res;
}
-// FlushFile() is a simple wrapper around fsync() that handles EINTR signals and
-// calls MapSystemError() to map errno to net error codes. It tries to flush to
-// completion.
-int FlushFile(base::PlatformFile file,
- bool record_uma,
- const net::BoundNetLog& bound_net_log) {
+void FileStreamPosix::AsyncContext::WriteFileImpl(scoped_refptr<IOBuffer> buf,
+ int buf_len,
+ int* result) {
base::ThreadRestrictions::AssertIOAllowed();
- ssize_t res = HANDLE_EINTR(fsync(file));
- if (res == -1) {
- res = RecordAndMapError(errno, FILE_ERROR_SOURCE_FLUSH, record_uma,
- bound_net_log);
- }
- return res;
+
+ // If context has been already destroyed nobody waits for operation results.
+ if (destroyed_)
+ return;
+
+ ssize_t res = HANDLE_EINTR(write(file_, buf->data(), buf_len));
+ if (res == -1)
+ res = RecordAndMapError(errno, FILE_ERROR_SOURCE_WRITE);
+ *result = res;
}
-// Called when Read(), Write() or Seek() is completed.
-// |result| contains the result or a network error code.
template <typename R>
-void OnIOComplete(const base::WeakPtr<FileStreamPosix>& stream,
- const base::Callback<void(R)>& callback,
- R* result) {
- if (!stream.get())
- return;
+void FileStreamPosix::AsyncContext::OnAsyncCompleted(
+ const base::Callback<void(R)>& callback,
+ R* result) {
+ if (destroyed_) {
+ DeleteAbandoned();
+ } else {
+ // Reset this before Run() as Run() may issue a new async operation.
+ async_in_progress_ = false;
+ callback.Run(*result);
+ }
+}
- // Reset this before Run() as Run() may issue a new async operation.
- stream->ResetOnIOComplete();
- callback.Run(*result);
+void FileStreamPosix::AsyncContext::DeleteAbandoned() {
+ if (auto_closed_ && file_ != base::kInvalidPlatformFileValue) {
+ const bool posted = base::WorkerPool::PostTask(
+ FROM_HERE,
+ // Context should be deleted after closing, thus Owned().
+ base::Bind(&AsyncContext::CloseFileImpl, base::Owned(this)),
+ true /* task_is_slow */);
+ DCHECK(posted);
+ } else {
+ delete this;
+ }
}
-} // namespace
// FileStreamPosix ------------------------------------------------------------
FileStreamPosix::FileStreamPosix(net::NetLog* net_log)
- : file_(base::kInvalidPlatformFileValue),
+ : context_(NULL),
open_flags_(0),
- auto_closed_(true),
- record_uma_(false),
bound_net_log_(net::BoundNetLog::Make(net_log,
- net::NetLog::SOURCE_FILESTREAM)),
- weak_ptr_factory_(ALLOW_THIS_IN_INITIALIZER_LIST(this)) {
+ net::NetLog::SOURCE_FILESTREAM)) {
+ context_ = new AsyncContext(bound_net_log_);
+
bound_net_log_.BeginEvent(net::NetLog::TYPE_FILE_STREAM_ALIVE);
}
-FileStreamPosix::FileStreamPosix(
- base::PlatformFile file, int flags, net::NetLog* net_log)
- : file_(file),
+FileStreamPosix::FileStreamPosix(base::PlatformFile file,
+ int flags,
+ net::NetLog* net_log)
+ : context_(NULL),
open_flags_(flags),
- auto_closed_(false),
- record_uma_(false),
bound_net_log_(net::BoundNetLog::Make(net_log,
- net::NetLog::SOURCE_FILESTREAM)),
- weak_ptr_factory_(ALLOW_THIS_IN_INITIALIZER_LIST(this)) {
+ net::NetLog::SOURCE_FILESTREAM)) {
+ context_ = new AsyncContext(file, bound_net_log_);
+
bound_net_log_.BeginEvent(net::NetLog::TYPE_FILE_STREAM_ALIVE);
}
FileStreamPosix::~FileStreamPosix() {
- if (open_flags_ & base::PLATFORM_FILE_ASYNC) {
- // Block until the last open/close/read/write operation is complete.
- // TODO(satorux): Ideally we should not block. crbug.com/115067
- WaitForIOCompletion();
- }
+ if (IsOpen() && context_->auto_closed() && !is_async())
+ CloseSync();
+ context_->Destroy();
- if (auto_closed_) {
- if (open_flags_ & base::PLATFORM_FILE_ASYNC) {
- // Close the file in the background.
- if (IsOpen()) {
- const bool posted = base::WorkerPool::PostTask(
- FROM_HERE,
- base::Bind(&CloseFile, file_, bound_net_log_),
- true /* task_is_slow */);
- DCHECK(posted);
- }
- } else {
- CloseSync();
- }
- }
-
bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_ALIVE);
}
void FileStreamPosix::Close(const CompletionCallback& callback) {
- DCHECK(open_flags_ & base::PLATFORM_FILE_ASYNC);
-
- DCHECK(!weak_ptr_factory_.HasWeakPtrs());
- DCHECK(!on_io_complete_.get());
- on_io_complete_.reset(new base::WaitableEvent(
- false /* manual_reset */, false /* initially_signaled */));
-
- // Passing &file_ to a thread pool looks unsafe but it's safe here as the
- // destructor ensures that the close operation is complete with
- // WaitForIOCompletion(). See also the destructor.
- const bool posted = base::WorkerPool::PostTaskAndReply(
- FROM_HERE,
- base::Bind(&CloseFileAndSignal, &file_, on_io_complete_.get(),
- bound_net_log_),
- base::Bind(&FileStreamPosix::OnClosed,
- weak_ptr_factory_.GetWeakPtr(),
- callback),
- true /* task_is_slow */);
-
- DCHECK(posted);
+ DCHECK(is_async());
+ context_->CloseAsync(callback);
}
void FileStreamPosix::CloseSync() {
+ // CloseSync() should be called on the correct thread even if it eventually
+ // ends up inside CloseAndCancelAsync().
+ base::ThreadRestrictions::AssertIOAllowed();
+
// TODO(satorux): Replace the following async stuff with a
- // DCHECK(open_flags & ASYNC) once once all async clients are migrated to
+ // DCHECK(open_flags & ASYNC) once all async clients are migrated to
// use Close(). crbug.com/114783
+ if (context_->async_in_progress())
+ CloseAndCancelAsync();
+ else
+ context_->CloseSync();
+}
- // Abort any existing asynchronous operations.
- weak_ptr_factory_.InvalidateWeakPtrs();
- // Block until the last open/read/write operation is complete.
- // TODO(satorux): Ideally we should not block. crbug.com/115067
- WaitForIOCompletion();
+void FileStreamPosix::CloseAndCancelAsync() {
+ if (!IsOpen())
+ return;
- CloseFile(file_, bound_net_log_);
- file_ = base::kInvalidPlatformFileValue;
+ DCHECK(is_async());
+
+ AsyncContext* old_ctx = context_;
+ context_ = new AsyncContext(bound_net_log_);
+ context_->set_record_uma(old_ctx->record_uma());
+ old_ctx->set_auto_closed(true);
+ old_ctx->Destroy();
}
-int FileStreamPosix::Open(const FilePath& path, int open_flags,
- const CompletionCallback& callback) {
+int FileStreamPosix::Open(const FilePath& path,
+ int open_flags,
+ const CompletionCallback& callback) {
if (IsOpen()) {
DLOG(FATAL) << "File is already open!";
return ERR_UNEXPECTED;
}
open_flags_ = open_flags;
- DCHECK(open_flags_ & base::PLATFORM_FILE_ASYNC);
-
- DCHECK(!weak_ptr_factory_.HasWeakPtrs());
- DCHECK(!on_io_complete_.get());
- on_io_complete_.reset(new base::WaitableEvent(
- false /* manual_reset */, false /* initially_signaled */));
-
- // Passing &file_ to a thread pool looks unsafe but it's safe here as the
- // destructor ensures that the open operation is complete with
- // WaitForIOCompletion(). See also the destructor.
- int* result = new int(OK);
- const bool posted = base::WorkerPool::PostTaskAndReply(
- FROM_HERE,
- base::Bind(&OpenFileAndSignal,
- path, open_flags, record_uma_, &file_, result,
- on_io_complete_.get(), bound_net_log_),
- base::Bind(&OnIOComplete<int>, weak_ptr_factory_.GetWeakPtr(),
- callback, base::Owned(result)),
- true /* task_is_slow */);
- DCHECK(posted);
+ DCHECK(is_async());
+ context_->OpenAsync(path, open_flags, callback);
return ERR_IO_PENDING;
}
@@ -373,59 +539,36 @@
}
open_flags_ = open_flags;
- // TODO(satorux): Put a DCHECK once once all async clients are migrated
+ // TODO(satorux): Put a DCHECK once all async clients are migrated
// to use Open(). crbug.com/114783
//
- // DCHECK(!(open_flags_ & base::PLATFORM_FILE_ASYNC));
-
- int result = OK;
- OpenFile(path, open_flags_, record_uma_, &file_, &result, bound_net_log_);
- return result;
+ // DCHECK(!is_async());
+ return context_->OpenSync(path, open_flags_);
}
bool FileStreamPosix::IsOpen() const {
- return file_ != base::kInvalidPlatformFileValue;
+ return context_->file() != base::kInvalidPlatformFileValue;
}
-int FileStreamPosix::Seek(Whence whence, int64 offset,
+int FileStreamPosix::Seek(Whence whence,
+ int64 offset,
const Int64CompletionCallback& callback) {
if (!IsOpen())
return ERR_UNEXPECTED;
- // Make sure we're async and we have no other in-flight async operations.
- DCHECK(open_flags_ & base::PLATFORM_FILE_ASYNC);
- DCHECK(!weak_ptr_factory_.HasWeakPtrs());
- DCHECK(!on_io_complete_.get());
-
- on_io_complete_.reset(new base::WaitableEvent(
- false /* manual_reset */, false /* initially_signaled */));
-
- int64* result = new int64(-1);
- const bool posted = base::WorkerPool::PostTaskAndReply(
- FROM_HERE,
- base::Bind(&SeekFileAndSignal, file_, whence, offset, result,
- record_uma_, on_io_complete_.get(), bound_net_log_),
- base::Bind(&OnIOComplete<int64>,
- weak_ptr_factory_.GetWeakPtr(),
- callback, base::Owned(result)),
- true /* task is slow */);
- DCHECK(posted);
+ // Make sure we're async.
+ DCHECK(is_async());
+ context_->SeekAsync(whence, offset, callback);
return ERR_IO_PENDING;
}
int64 FileStreamPosix::SeekSync(Whence whence, int64 offset) {
- base::ThreadRestrictions::AssertIOAllowed();
-
if (!IsOpen())
return ERR_UNEXPECTED;
// If we're in async, make sure we don't have a request in flight.
- DCHECK(!(open_flags_ & base::PLATFORM_FILE_ASYNC) ||
- !on_io_complete_.get());
-
- off_t result = -1;
- SeekFile(file_, whence, offset, &result, record_uma_, bound_net_log_);
- return result;
+ DCHECK(!is_async() || !context_->async_in_progress());
+ return context_->SeekSync(whence, offset);
}
int64 FileStreamPosix::Available() {
@@ -439,12 +582,8 @@
return cur_pos;
struct stat info;
- if (fstat(file_, &info) != 0) {
- return RecordAndMapError(errno,
- FILE_ERROR_SOURCE_GET_SIZE,
- record_uma_,
- bound_net_log_);
- }
+ if (fstat(context_->file(), &info) != 0)
+ return context_->RecordAndMapError(errno, FILE_ERROR_SOURCE_GET_SIZE);
int64 size = static_cast<int64>(info.st_size);
DCHECK_GT(size, cur_pos);
@@ -452,34 +591,18 @@
return size - cur_pos;
}
-int FileStreamPosix::Read(
- IOBuffer* in_buf, int buf_len, const CompletionCallback& callback) {
+int FileStreamPosix::Read(IOBuffer* buf,
+ int buf_len,
+ const CompletionCallback& callback) {
if (!IsOpen())
return ERR_UNEXPECTED;
// read(..., 0) will return 0, which indicates end-of-file.
DCHECK_GT(buf_len, 0);
DCHECK(open_flags_ & base::PLATFORM_FILE_READ);
- DCHECK(open_flags_ & base::PLATFORM_FILE_ASYNC);
+ DCHECK(is_async());
- // Make sure we don't have a request in flight.
- DCHECK(!weak_ptr_factory_.HasWeakPtrs());
- DCHECK(!on_io_complete_.get());
-
- on_io_complete_.reset(new base::WaitableEvent(
- false /* manual_reset */, false /* initially_signaled */));
-
- int* result = new int(OK);
- scoped_refptr<IOBuffer> buf = in_buf;
- const bool posted = base::WorkerPool::PostTaskAndReply(
- FROM_HERE,
- base::Bind(&ReadFileAndSignal, file_, buf, buf_len,
- record_uma_, result, on_io_complete_.get(), bound_net_log_),
- base::Bind(&OnIOComplete<int>,
- weak_ptr_factory_.GetWeakPtr(),
- callback, base::Owned(result)),
- true /* task is slow */);
- DCHECK(posted);
+ context_->ReadAsync(buf, buf_len, callback);
return ERR_IO_PENDING;
}
@@ -487,14 +610,12 @@
if (!IsOpen())
return ERR_UNEXPECTED;
- DCHECK(!(open_flags_ & base::PLATFORM_FILE_ASYNC));
+ DCHECK(!is_async());
// read(..., 0) will return 0, which indicates end-of-file.
DCHECK_GT(buf_len, 0);
DCHECK(open_flags_ & base::PLATFORM_FILE_READ);
- int result = OK;
- ReadFile(file_, buf, buf_len, record_uma_, &result, bound_net_log_);
- return result;
+ return context_->ReadSync(buf, buf_len);
}
int FileStreamPosix::ReadUntilComplete(char *buf, int buf_len) {
@@ -518,49 +639,31 @@
return bytes_total;
}
-int FileStreamPosix::Write(
- IOBuffer* in_buf, int buf_len, const CompletionCallback& callback) {
+int FileStreamPosix::Write(IOBuffer* buf,
+ int buf_len,
+ const CompletionCallback& callback) {
if (!IsOpen())
return ERR_UNEXPECTED;
- DCHECK(open_flags_ & base::PLATFORM_FILE_ASYNC);
+ DCHECK(is_async());
DCHECK(open_flags_ & base::PLATFORM_FILE_WRITE);
// write(..., 0) will return 0, which indicates end-of-file.
DCHECK_GT(buf_len, 0);
- // Make sure we don't have a request in flight.
- DCHECK(!weak_ptr_factory_.HasWeakPtrs());
- DCHECK(!on_io_complete_.get());
- on_io_complete_.reset(new base::WaitableEvent(
- false /* manual_reset */, false /* initially_signaled */));
-
- int* result = new int(OK);
- scoped_refptr<IOBuffer> buf = in_buf;
- const bool posted = base::WorkerPool::PostTaskAndReply(
- FROM_HERE,
- base::Bind(&WriteFileAndSignal, file_, buf, buf_len,
- record_uma_, result, on_io_complete_.get(), bound_net_log_),
- base::Bind(&OnIOComplete<int>,
- weak_ptr_factory_.GetWeakPtr(),
- callback, base::Owned(result)),
- true /* task is slow */);
- DCHECK(posted);
+ context_->WriteAsync(buf, buf_len, callback);
return ERR_IO_PENDING;
}
-int FileStreamPosix::WriteSync(
- const char* buf, int buf_len) {
+int FileStreamPosix::WriteSync(const char* buf, int buf_len) {
if (!IsOpen())
return ERR_UNEXPECTED;
- DCHECK(!(open_flags_ & base::PLATFORM_FILE_ASYNC));
+ DCHECK(!is_async());
DCHECK(open_flags_ & base::PLATFORM_FILE_WRITE);
// write(..., 0) will return 0, which indicates end-of-file.
DCHECK_GT(buf_len, 0);
- int result = OK;
- WriteFile(file_, buf, buf_len, record_uma_, &result, bound_net_log_);
- return result;
+ return context_->WriteSync(buf, buf_len);
}
int64 FileStreamPosix::Truncate(int64 bytes) {
@@ -578,25 +681,27 @@
return ERR_UNEXPECTED;
// And truncate the file.
- int result = ftruncate(file_, bytes);
+ int result = ftruncate(context_->file(), bytes);
if (result == 0)
return seek_position;
- return RecordAndMapError(errno,
- FILE_ERROR_SOURCE_SET_EOF,
- record_uma_,
- bound_net_log_);
+ return context_->RecordAndMapError(errno, FILE_ERROR_SOURCE_SET_EOF);
}
int FileStreamPosix::Flush() {
+ base::ThreadRestrictions::AssertIOAllowed();
+
if (!IsOpen())
return ERR_UNEXPECTED;
- return FlushFile(file_, record_uma_, bound_net_log_);
+ ssize_t res = HANDLE_EINTR(fsync(context_->file()));
+ if (res == -1)
+ res = context_->RecordAndMapError(errno, FILE_ERROR_SOURCE_FLUSH);
+ return res;
}
void FileStreamPosix::EnableErrorStatistics() {
- record_uma_ = true;
+ context_->set_record_uma(true);
}
void FileStreamPosix::SetBoundNetLogSource(
@@ -620,29 +725,7 @@
}
base::PlatformFile FileStreamPosix::GetPlatformFileForTesting() {
- return file_;
+ return context_->file();
}
-void FileStreamPosix::ResetOnIOComplete() {
- on_io_complete_.reset();
- weak_ptr_factory_.InvalidateWeakPtrs();
-}
-
-void FileStreamPosix::OnClosed(const CompletionCallback& callback) {
- file_ = base::kInvalidPlatformFileValue;
-
- // Reset this before Run() as Run() may issue a new async operation.
- ResetOnIOComplete();
- callback.Run(OK);
-}
-
-void FileStreamPosix::WaitForIOCompletion() {
- // http://crbug.com/115067
- base::ThreadRestrictions::ScopedAllowWait allow_wait;
- if (on_io_complete_.get()) {
- on_io_complete_->Wait();
- on_io_complete_.reset();
- }
-}
-
} // namespace net

Powered by Google App Engine
This is Rietveld 408576698