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

Side by Side Diff: net/base/file_stream_win.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, 3 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 unified diff | Download patch
OLDNEW
1 // Copyright (c) 2012 The Chromium Authors. All rights reserved. 1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
2 // Use of this source code is governed by a BSD-style license that can be 2 // Use of this source code is governed by a BSD-style license that can be
3 // found in the LICENSE file. 3 // found in the LICENSE file.
4 4
5 #include "net/base/file_stream.h" 5 #include "net/base/file_stream_win.h"
6 6
7 #include <windows.h> 7 #include <windows.h>
8 8
9 #include "base/file_path.h" 9 #include "base/file_path.h"
10 #include "base/logging.h" 10 #include "base/logging.h"
11 #include "base/message_loop.h" 11 #include "base/memory/ref_counted.h"
12 #include "base/metrics/histogram.h" 12 #include "base/metrics/histogram.h"
13 #include "base/synchronization/waitable_event.h" 13 #include "base/task_runner_util.h"
14 #include "base/threading/thread_restrictions.h" 14 #include "base/threading/thread_restrictions.h"
15 #include "base/threading/worker_pool.h" 15 #include "base/threading/worker_pool.h"
16 #include "net/base/file_stream_metrics.h"
17 #include "net/base/file_stream_net_log_parameters.h" 16 #include "net/base/file_stream_net_log_parameters.h"
18 #include "net/base/io_buffer.h" 17 #include "net/base/io_buffer.h"
19 #include "net/base/net_errors.h" 18 #include "net/base/net_errors.h"
20 19
21 namespace net { 20 namespace net {
22 21
23 // Ensure that we can just use our Whence values directly. 22 // Ensure that we can just use our Whence values directly.
24 COMPILE_ASSERT(FROM_BEGIN == FILE_BEGIN, bad_whence_begin); 23 COMPILE_ASSERT(FROM_BEGIN == FILE_BEGIN, bad_whence_begin);
25 COMPILE_ASSERT(FROM_CURRENT == FILE_CURRENT, bad_whence_current); 24 COMPILE_ASSERT(FROM_CURRENT == FILE_CURRENT, bad_whence_current);
26 COMPILE_ASSERT(FROM_END == FILE_END, bad_whence_end); 25 COMPILE_ASSERT(FROM_END == FILE_END, bad_whence_end);
27 26
28 namespace { 27 namespace {
29 28
30 void SetOffset(OVERLAPPED* overlapped, const LARGE_INTEGER& offset) { 29 void SetOffset(OVERLAPPED* overlapped, const LARGE_INTEGER& offset) {
31 overlapped->Offset = offset.LowPart; 30 overlapped->Offset = offset.LowPart;
32 overlapped->OffsetHigh = offset.HighPart; 31 overlapped->OffsetHigh = offset.HighPart;
33 } 32 }
34 33
35 void IncrementOffset(OVERLAPPED* overlapped, DWORD count) { 34 void IncrementOffset(OVERLAPPED* overlapped, DWORD count) {
36 LARGE_INTEGER offset; 35 LARGE_INTEGER offset;
37 offset.LowPart = overlapped->Offset; 36 offset.LowPart = overlapped->Offset;
38 offset.HighPart = overlapped->OffsetHigh; 37 offset.HighPart = overlapped->OffsetHigh;
39 offset.QuadPart += static_cast<LONGLONG>(count); 38 offset.QuadPart += static_cast<LONGLONG>(count);
40 SetOffset(overlapped, offset); 39 SetOffset(overlapped, offset);
41 } 40 }
42 41
43 int RecordAndMapError(int error,
44 FileErrorSource source,
45 bool record_uma,
46 const net::BoundNetLog& bound_net_log) {
47 net::Error net_error = MapSystemError(error);
48
49 bound_net_log.AddEvent(
50 net::NetLog::TYPE_FILE_STREAM_ERROR,
51 base::Bind(&NetLogFileStreamErrorCallback,
52 source, error, net_error));
53
54 RecordFileError(error, source, record_uma);
55
56 return net_error;
57 }
58
59 // Opens a file with some network logging.
60 // The opened file and the result code are written to |file| and |result|.
61 void OpenFile(const FilePath& path,
62 int open_flags,
63 bool record_uma,
64 base::PlatformFile* file,
65 int* result,
66 const net::BoundNetLog& bound_net_log) {
67 std::string file_name = path.AsUTF8Unsafe();
68 bound_net_log.BeginEvent(
69 net::NetLog::TYPE_FILE_STREAM_OPEN,
70 NetLog::StringCallback("file_name", &file_name));
71
72 *file = base::CreatePlatformFile(path, open_flags, NULL, NULL);
73 if (*file == base::kInvalidPlatformFileValue) {
74 DWORD error = GetLastError();
75 LOG(WARNING) << "Failed to open file: " << error;
76 *result = RecordAndMapError(error,
77 FILE_ERROR_SOURCE_OPEN,
78 record_uma,
79 bound_net_log);
80 bound_net_log.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
81 return;
82 }
83 }
84
85 // Closes a file with some network logging.
86 void CloseFile(base::PlatformFile file,
87 const net::BoundNetLog& bound_net_log) {
88 bound_net_log.AddEvent(net::NetLog::TYPE_FILE_STREAM_CLOSE);
89 if (file == base::kInvalidPlatformFileValue)
90 return;
91
92 CancelIo(file);
93
94 if (!base::ClosePlatformFile(file))
95 NOTREACHED();
96 bound_net_log.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
97 }
98
99 // Closes a file with CloseFile() and signals the completion.
100 void CloseFileAndSignal(base::PlatformFile* file,
101 base::WaitableEvent* on_io_complete,
102 const net::BoundNetLog& bound_net_log) {
103 CloseFile(*file, bound_net_log);
104 *file = base::kInvalidPlatformFileValue;
105 on_io_complete->Signal();
106 }
107
108 // Invokes a given closure and signals the completion.
109 void InvokeAndSignal(const base::Closure& closure,
110 base::WaitableEvent* on_io_complete) {
111 closure.Run();
112 on_io_complete->Signal();
113 }
114
115 } // namespace 42 } // namespace
116 43
117 // FileStreamWin::AsyncContext ---------------------------------------------- 44 // FileStream::AsyncContext ----------------------------------------------
118 45
119 class FileStreamWin::AsyncContext : public MessageLoopForIO::IOHandler {
120 public:
121 explicit AsyncContext(const net::BoundNetLog& bound_net_log)
122 : context_(), is_closing_(false),
123 record_uma_(false), bound_net_log_(bound_net_log),
124 error_source_(FILE_ERROR_SOURCE_COUNT) {
125 context_.handler = this;
126 }
127 ~AsyncContext();
128 46
129 void IOCompletionIsPending(const CompletionCallback& callback, 47 FileStream::Context::Context(const BoundNetLog& bound_net_log)
130 IOBuffer* buf); 48 : io_context_(),
49 file_(base::kInvalidPlatformFileValue),
50 record_uma_(false),
51 async_in_progress_(false),
52 orphaned_(false),
53 bound_net_log_(bound_net_log),
54 error_source_(FILE_ERROR_SOURCE_COUNT) {
55 io_context_.handler = this;
56 }
131 57
132 OVERLAPPED* overlapped() { return &context_.overlapped; } 58 FileStream::Context::Context(base::PlatformFile file,
133 const CompletionCallback& callback() const { return callback_; } 59 const BoundNetLog& bound_net_log,
60 int open_flags)
61 : io_context_(),
62 file_(file),
63 record_uma_(false),
64 async_in_progress_(false),
65 orphaned_(false),
66 bound_net_log_(bound_net_log),
67 error_source_(FILE_ERROR_SOURCE_COUNT) {
68 io_context_.handler = this;
69 if (open_flags & base::PLATFORM_FILE_ASYNC)
70 RegisterInMessageLoop();
71 }
134 72
135 void set_error_source(FileErrorSource source) { error_source_ = source; } 73 FileStream::Context::~Context() {
74 }
136 75
137 void EnableErrorStatistics() { 76 void FileStream::Context::Orphan() {
138 record_uma_ = true; 77 orphaned_ = true;
139 } 78 if (file_ != base::kInvalidPlatformFileValue)
79 CancelIo(file_);
80 if (!async_in_progress_)
81 CloseAsync(CompletionCallback());
82 }
140 83
141 private: 84 void FileStream::Context::OpenAsync(const FilePath& path,
142 virtual void OnIOCompleted(MessageLoopForIO::IOContext* context, 85 int open_flags,
143 DWORD bytes_read, DWORD error) OVERRIDE; 86 const CompletionCallback& callback) {
87 DCHECK(!async_in_progress_);
144 88
145 MessageLoopForIO::IOContext context_; 89 BeginOpenEvent(path);
146 CompletionCallback callback_;
147 scoped_refptr<IOBuffer> in_flight_buf_;
148 bool is_closing_;
149 bool record_uma_;
150 const net::BoundNetLog bound_net_log_;
151 FileErrorSource error_source_;
152 };
153 90
154 FileStreamWin::AsyncContext::~AsyncContext() { 91 const bool posted = base::PostTaskAndReplyWithResult(
155 is_closing_ = true; 92 base::WorkerPool::GetTaskRunner(true /* task_is_slow */),
156 bool waited = false; 93 FROM_HERE,
157 base::TimeTicks start = base::TimeTicks::Now(); 94 base::Bind(&Context::OpenFileImpl,
158 while (!callback_.is_null()) { 95 base::Unretained(this), path, open_flags),
159 waited = true; 96 base::Bind(&Context::OnOpenCompleted,
160 MessageLoopForIO::current()->WaitForIOCompletion(INFINITE, this); 97 base::Unretained(this), callback));
161 } 98 DCHECK(posted);
162 if (waited) { 99
163 // We want to see if we block the message loop for too long. 100 async_in_progress_ = true;
164 UMA_HISTOGRAM_TIMES("AsyncIO.FileStreamClose", 101 }
165 base::TimeTicks::Now() - start); 102
103 int FileStream::Context::OpenSync(const FilePath& path, int open_flags) {
104 BeginOpenEvent(path);
105 int result = OpenFileImpl(path, open_flags);
106 CheckForOpenError(&result);
107 // TODO(satorux): Remove this once all async clients are migrated to use
108 // Open(). crbug.com/114783
109 if (open_flags & base::PLATFORM_FILE_ASYNC)
110 RegisterInMessageLoop();
111 return result;
112 }
113
114 void FileStream::Context::CloseAsync(const CompletionCallback& callback) {
115 DCHECK(!async_in_progress_);
116
117 bound_net_log_.AddEvent(net::NetLog::TYPE_FILE_STREAM_CLOSE);
118
119 if (file_ == base::kInvalidPlatformFileValue) {
120 MessageLoop::current()->PostTask(
121 FROM_HERE,
122 base::Bind(&Context::OnCloseCompleted,
123 base::Unretained(this), callback));
124 } else {
125 const bool posted = base::WorkerPool::PostTaskAndReply(
126 FROM_HERE,
127 base::Bind(&Context::CloseFileImpl,
128 base::Unretained(this)),
129 base::Bind(&Context::OnCloseCompleted,
130 base::Unretained(this), callback),
131 true /* task_is_slow */);
132 DCHECK(posted);
133
134 async_in_progress_ = true;
166 } 135 }
167 } 136 }
168 137
169 void FileStreamWin::AsyncContext::IOCompletionIsPending( 138 void FileStream::Context::CloseSync() {
170 const CompletionCallback& callback, 139 DCHECK(!async_in_progress_);
171 IOBuffer* buf) {
172 DCHECK(callback_.is_null());
173 callback_ = callback;
174 in_flight_buf_ = buf; // Hold until the async operation ends.
175 }
176
177 void FileStreamWin::AsyncContext::OnIOCompleted(
178 MessageLoopForIO::IOContext* context, DWORD bytes_read, DWORD error) {
179 DCHECK_EQ(&context_, context);
180 DCHECK(!callback_.is_null());
181
182 if (is_closing_) {
183 callback_.Reset();
184 in_flight_buf_ = NULL;
185 return;
186 }
187
188 int result = static_cast<int>(bytes_read);
189 if (error && error != ERROR_HANDLE_EOF) {
190 result = RecordAndMapError(error, error_source_, record_uma_,
191 bound_net_log_);
192 }
193
194 if (bytes_read)
195 IncrementOffset(&context->overlapped, bytes_read);
196
197 CompletionCallback temp_callback = callback_;
198 callback_.Reset();
199 scoped_refptr<IOBuffer> temp_buf = in_flight_buf_;
200 in_flight_buf_ = NULL;
201 temp_callback.Run(result);
202 }
203
204 // FileStream ------------------------------------------------------------
205
206 FileStreamWin::FileStreamWin(net::NetLog* net_log)
207 : file_(base::kInvalidPlatformFileValue),
208 open_flags_(0),
209 auto_closed_(true),
210 record_uma_(false),
211 bound_net_log_(net::BoundNetLog::Make(net_log,
212 net::NetLog::SOURCE_FILESTREAM)),
213 weak_ptr_factory_(ALLOW_THIS_IN_INITIALIZER_LIST(this)) {
214 bound_net_log_.BeginEvent(net::NetLog::TYPE_FILE_STREAM_ALIVE);
215 }
216
217 FileStreamWin::FileStreamWin(
218 base::PlatformFile file, int flags, net::NetLog* net_log)
219 : file_(file),
220 open_flags_(flags),
221 auto_closed_(false),
222 record_uma_(false),
223 bound_net_log_(net::BoundNetLog::Make(net_log,
224 net::NetLog::SOURCE_FILESTREAM)),
225 weak_ptr_factory_(ALLOW_THIS_IN_INITIALIZER_LIST(this)) {
226 bound_net_log_.BeginEvent(net::NetLog::TYPE_FILE_STREAM_ALIVE);
227
228 // If the file handle is opened with base::PLATFORM_FILE_ASYNC, we need to
229 // make sure we will perform asynchronous File IO to it.
230 if (flags & base::PLATFORM_FILE_ASYNC) {
231 async_context_.reset(new AsyncContext(bound_net_log_));
232 MessageLoopForIO::current()->RegisterIOHandler(file_,
233 async_context_.get());
234 }
235 }
236
237 FileStreamWin::~FileStreamWin() {
238 if (open_flags_ & base::PLATFORM_FILE_ASYNC) {
239 // Block until the in-flight open/close operation is complete.
240 // TODO(satorux): Ideally we should not block. crbug.com/115067
241 WaitForIOCompletion();
242
243 // Block until the last read/write operation is complete.
244 async_context_.reset();
245 }
246
247 if (auto_closed_) {
248 if (open_flags_ & base::PLATFORM_FILE_ASYNC) {
249 // Close the file in the background.
250 if (IsOpen()) {
251 const bool posted = base::WorkerPool::PostTask(
252 FROM_HERE,
253 base::Bind(&CloseFile, file_, bound_net_log_),
254 true /* task_is_slow */);
255 DCHECK(posted);
256 }
257 } else {
258 CloseSync();
259 }
260 }
261
262 bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_ALIVE);
263 }
264
265 void FileStreamWin::Close(const CompletionCallback& callback) {
266 DCHECK(open_flags_ & base::PLATFORM_FILE_ASYNC);
267 DCHECK(!weak_ptr_factory_.HasWeakPtrs());
268 DCHECK(!on_io_complete_.get());
269 on_io_complete_.reset(new base::WaitableEvent(
270 false /* manual_reset */, false /* initially_signaled */));
271
272 // Passing &file_ to a thread pool looks unsafe but it's safe here as the
273 // destructor ensures that the close operation is complete with
274 // WaitForIOCompletion(). See also the destructor.
275 const bool posted = base::WorkerPool::PostTaskAndReply(
276 FROM_HERE,
277 base::Bind(&CloseFileAndSignal, &file_, on_io_complete_.get(),
278 bound_net_log_),
279 base::Bind(&FileStreamWin::OnClosed,
280 weak_ptr_factory_.GetWeakPtr(),
281 callback),
282 true /* task_is_slow */);
283 DCHECK(posted);
284 }
285
286 void FileStreamWin::CloseSync() {
287 // The logic here is similar to CloseFile() but async_context_.reset() is
288 // caled in this function.
289
290 // Block until the in-flight open operation is complete.
291 // TODO(satorux): Replace this with a DCHECK(open_flags & ASYNC) once this
292 // once all async clients are migrated to use Close(). crbug.com/114783
293 WaitForIOCompletion();
294
295 bound_net_log_.AddEvent(net::NetLog::TYPE_FILE_STREAM_CLOSE); 140 bound_net_log_.AddEvent(net::NetLog::TYPE_FILE_STREAM_CLOSE);
296 if (file_ != base::kInvalidPlatformFileValue)
297 CancelIo(file_);
298
299 // Block until the last read/write operation is complete.
300 async_context_.reset();
301
302 if (file_ != base::kInvalidPlatformFileValue) { 141 if (file_ != base::kInvalidPlatformFileValue) {
303 if (!base::ClosePlatformFile(file_)) 142 CloseFileImpl();
304 NOTREACHED();
305 file_ = base::kInvalidPlatformFileValue;
306
307 bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN); 143 bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
308 } 144 }
309 } 145 }
310 146
311 int FileStreamWin::Open(const FilePath& path, int open_flags, 147 void FileStream::Context::SeekAsync(Whence whence,
312 const CompletionCallback& callback) { 148 int64 offset,
313 if (IsOpen()) { 149 const Int64CompletionCallback& callback) {
314 DLOG(FATAL) << "File is already open!"; 150 DCHECK(!async_in_progress_);
315 return ERR_UNEXPECTED;
316 }
317 151
318 open_flags_ = open_flags; 152 int64* result = new int64(-1);
319 DCHECK(open_flags_ & base::PLATFORM_FILE_ASYNC); 153 const bool posted = base::PostTaskAndReplyWithResult(
320 DCHECK(!weak_ptr_factory_.HasWeakPtrs()); 154 base::WorkerPool::GetTaskRunner(true /* task is slow */),
321 DCHECK(!on_io_complete_.get()); 155 FROM_HERE,
322 on_io_complete_.reset(new base::WaitableEvent( 156 base::Bind(&Context::SeekFileImpl,
323 false /* manual_reset */, false /* initially_signaled */)); 157 base::Unretained(this), whence, offset),
158 base::Bind(&Context::OnSeekCompleted,
159 base::Unretained(this), callback));
160 DCHECK(posted);
324 161
325 // Passing &file_ to a thread pool looks unsafe but it's safe here as the 162 async_in_progress_ = true;
326 // destructor ensures that the open operation is complete with
327 // WaitForIOCompletion(). See also the destructor.
328 int* result = new int(OK);
329 const bool posted = base::WorkerPool::PostTaskAndReply(
330 FROM_HERE,
331 base::Bind(&InvokeAndSignal,
332 base::Bind(&OpenFile, path, open_flags, record_uma_, &file_,
333 result, bound_net_log_),
334 on_io_complete_.get()),
335 base::Bind(&FileStreamWin::OnOpened,
336 weak_ptr_factory_.GetWeakPtr(),
337 callback, base::Owned(result)),
338 true /* task_is_slow */);
339 DCHECK(posted);
340 return ERR_IO_PENDING;
341 } 163 }
342 164
343 int FileStreamWin::OpenSync(const FilePath& path, int open_flags) { 165 int64 FileStream::Context::SeekSync(Whence whence, int64 offset) {
344 if (IsOpen()) { 166 int64 result = SeekFileImpl(whence, offset);
345 DLOG(FATAL) << "File is already open!"; 167 CheckForSeekError(&result);
346 return ERR_UNEXPECTED;
347 }
348
349 open_flags_ = open_flags;
350
351 int result = OK;
352 OpenFile(path, open_flags_, record_uma_, &file_, &result, bound_net_log_);
353 if (result != OK)
354 return result;
355
356 // TODO(satorux): Remove this once all async clients are migrated to use
357 // Open(). crbug.com/114783
358 if (open_flags_ & base::PLATFORM_FILE_ASYNC) {
359 async_context_.reset(new AsyncContext(bound_net_log_));
360 if (record_uma_)
361 async_context_->EnableErrorStatistics();
362 MessageLoopForIO::current()->RegisterIOHandler(file_,
363 async_context_.get());
364 }
365
366 return OK;
367 }
368
369 bool FileStreamWin::IsOpen() const {
370 return file_ != base::kInvalidPlatformFileValue;
371 }
372
373 int FileStreamWin::Seek(Whence whence, int64 offset,
374 const Int64CompletionCallback& callback) {
375 if (!IsOpen())
376 return ERR_UNEXPECTED;
377
378 // Make sure we're async and we have no other in-flight async operations.
379 DCHECK(open_flags_ & base::PLATFORM_FILE_ASYNC);
380 DCHECK(!weak_ptr_factory_.HasWeakPtrs());
381 DCHECK(!on_io_complete_.get());
382
383 int64* result = new int64(-1);
384 on_io_complete_.reset(new base::WaitableEvent(
385 false /* manual_reset */, false /* initially_signaled */));
386
387 const bool posted = base::WorkerPool::PostTaskAndReply(
388 FROM_HERE,
389 base::Bind(&InvokeAndSignal,
390 // Unretained should be fine as we wait for a signal on
391 // on_io_complete_ at the destructor.
392 base::Bind(&FileStreamWin::SeekFile, base::Unretained(this),
393 whence, offset, result),
394 on_io_complete_.get()),
395 base::Bind(&FileStreamWin::OnSeeked,
396 weak_ptr_factory_.GetWeakPtr(),
397 callback, base::Owned(result)),
398 true /* task is slow */);
399 DCHECK(posted);
400 return ERR_IO_PENDING;
401 }
402
403 int64 FileStreamWin::SeekSync(Whence whence, int64 offset) {
404 if (!IsOpen())
405 return ERR_UNEXPECTED;
406
407 DCHECK(!async_context_.get() || async_context_->callback().is_null());
408 int64 result = -1;
409 SeekFile(whence, offset, &result);
410 return result; 168 return result;
411 } 169 }
412 170
413 int64 FileStreamWin::Available() { 171 int64 FileStream::Context::GetFileSize() const {
414 base::ThreadRestrictions::AssertIOAllowed();
415
416 if (!IsOpen())
417 return ERR_UNEXPECTED;
418
419 int64 cur_pos = SeekSync(FROM_CURRENT, 0);
420 if (cur_pos < 0)
421 return cur_pos;
422
423 LARGE_INTEGER file_size; 172 LARGE_INTEGER file_size;
424 if (!GetFileSizeEx(file_, &file_size)) { 173 if (!GetFileSizeEx(file_, &file_size)) {
425 DWORD error = GetLastError(); 174 DWORD error = GetLastError();
426 LOG(WARNING) << "GetFileSizeEx failed: " << error; 175 LOG(WARNING) << "GetFileSizeEx failed: " << error;
427 return RecordAndMapError(error, 176 return RecordAndMapError(error, FILE_ERROR_SOURCE_GET_SIZE);
428 FILE_ERROR_SOURCE_GET_SIZE,
429 record_uma_,
430 bound_net_log_);
431 } 177 }
432 178
433 return file_size.QuadPart - cur_pos; 179 return file_size.QuadPart;
434 } 180 }
435 181
436 int FileStreamWin::Read( 182 int FileStream::Context::ReadAsync(IOBuffer* buf,
437 IOBuffer* buf, int buf_len, const CompletionCallback& callback) { 183 int buf_len,
438 DCHECK(async_context_.get()); 184 const CompletionCallback& callback) {
439 185 DCHECK(!async_in_progress_);
440 if (!IsOpen()) 186 error_source_ = FILE_ERROR_SOURCE_READ;
441 return ERR_UNEXPECTED;
442
443 DCHECK(open_flags_ & base::PLATFORM_FILE_READ);
444
445 OVERLAPPED* overlapped = NULL;
446 DCHECK(!callback.is_null());
447 DCHECK(async_context_->callback().is_null());
448 overlapped = async_context_->overlapped();
449 async_context_->set_error_source(FILE_ERROR_SOURCE_READ);
450 187
451 int rv = 0; 188 int rv = 0;
452 189
453 DWORD bytes_read; 190 DWORD bytes_read;
454 if (!ReadFile(file_, buf->data(), buf_len, &bytes_read, overlapped)) { 191 if (!ReadFile(file_, buf->data(), buf_len,
192 &bytes_read, &io_context_.overlapped)) {
455 DWORD error = GetLastError(); 193 DWORD error = GetLastError();
456 if (error == ERROR_IO_PENDING) { 194 if (error == ERROR_IO_PENDING) {
457 async_context_->IOCompletionIsPending(callback, buf); 195 IOCompletionIsPending(callback, buf);
458 rv = ERR_IO_PENDING; 196 rv = ERR_IO_PENDING;
459 } else if (error == ERROR_HANDLE_EOF) { 197 } else if (error == ERROR_HANDLE_EOF) {
460 rv = 0; // Report EOF by returning 0 bytes read. 198 rv = 0; // Report EOF by returning 0 bytes read.
461 } else { 199 } else {
462 LOG(WARNING) << "ReadFile failed: " << error; 200 LOG(WARNING) << "ReadFile failed: " << error;
463 rv = RecordAndMapError(error, 201 rv = RecordAndMapError(error, FILE_ERROR_SOURCE_READ);
464 FILE_ERROR_SOURCE_READ,
465 record_uma_,
466 bound_net_log_);
467 } 202 }
468 } else if (overlapped) { 203 } else {
469 async_context_->IOCompletionIsPending(callback, buf); 204 IOCompletionIsPending(callback, buf);
470 rv = ERR_IO_PENDING; 205 rv = ERR_IO_PENDING;
471 } else {
472 rv = static_cast<int>(bytes_read);
473 } 206 }
474 return rv; 207 return rv;
475 } 208 }
476 209
477 int FileStreamWin::ReadSync(char* buf, int buf_len) { 210 int FileStream::Context::ReadSync(char* buf, int buf_len) {
478 DCHECK(!async_context_.get());
479 base::ThreadRestrictions::AssertIOAllowed(); 211 base::ThreadRestrictions::AssertIOAllowed();
480 212
481 if (!IsOpen())
482 return ERR_UNEXPECTED;
483
484 DCHECK(open_flags_ & base::PLATFORM_FILE_READ);
485
486 int rv = 0; 213 int rv = 0;
487 214
488 DWORD bytes_read; 215 DWORD bytes_read;
489 if (!ReadFile(file_, buf, buf_len, &bytes_read, NULL)) { 216 if (!ReadFile(file_, buf, buf_len, &bytes_read, NULL)) {
490 DWORD error = GetLastError(); 217 DWORD error = GetLastError();
491 if (error == ERROR_HANDLE_EOF) { 218 if (error == ERROR_HANDLE_EOF) {
492 rv = 0; // Report EOF by returning 0 bytes read. 219 rv = 0; // Report EOF by returning 0 bytes read.
493 } else { 220 } else {
494 LOG(WARNING) << "ReadFile failed: " << error; 221 LOG(WARNING) << "ReadFile failed: " << error;
495 rv = RecordAndMapError(error, 222 rv = RecordAndMapError(error, FILE_ERROR_SOURCE_READ);
496 FILE_ERROR_SOURCE_READ,
497 record_uma_,
498 bound_net_log_);
499 } 223 }
500 } else { 224 } else {
501 rv = static_cast<int>(bytes_read); 225 rv = static_cast<int>(bytes_read);
502 } 226 }
503 return rv; 227 return rv;
504 } 228 }
505 229
506 int FileStreamWin::ReadUntilComplete(char *buf, int buf_len) { 230 int FileStream::Context::WriteAsync(IOBuffer* buf,
507 int to_read = buf_len; 231 int buf_len,
508 int bytes_total = 0; 232 const CompletionCallback& callback) {
509 233 error_source_ = FILE_ERROR_SOURCE_WRITE;
510 do {
511 int bytes_read = ReadSync(buf, to_read);
512 if (bytes_read <= 0) {
513 if (bytes_total == 0)
514 return bytes_read;
515
516 return bytes_total;
517 }
518
519 bytes_total += bytes_read;
520 buf += bytes_read;
521 to_read -= bytes_read;
522 } while (bytes_total < buf_len);
523
524 return bytes_total;
525 }
526
527 int FileStreamWin::Write(
528 IOBuffer* buf, int buf_len, const CompletionCallback& callback) {
529 DCHECK(async_context_.get());
530
531 if (!IsOpen())
532 return ERR_UNEXPECTED;
533
534 DCHECK(open_flags_ & base::PLATFORM_FILE_WRITE);
535
536 OVERLAPPED* overlapped = NULL;
537 DCHECK(!callback.is_null());
538 DCHECK(async_context_->callback().is_null());
539 overlapped = async_context_->overlapped();
540 async_context_->set_error_source(FILE_ERROR_SOURCE_WRITE);
541 234
542 int rv = 0; 235 int rv = 0;
543 DWORD bytes_written = 0; 236 DWORD bytes_written = 0;
544 if (!WriteFile(file_, buf->data(), buf_len, &bytes_written, overlapped)) { 237 if (!WriteFile(file_, buf->data(), buf_len,
238 &bytes_written, &io_context_.overlapped)) {
545 DWORD error = GetLastError(); 239 DWORD error = GetLastError();
546 if (error == ERROR_IO_PENDING) { 240 if (error == ERROR_IO_PENDING) {
547 async_context_->IOCompletionIsPending(callback, buf); 241 IOCompletionIsPending(callback, buf);
548 rv = ERR_IO_PENDING; 242 rv = ERR_IO_PENDING;
549 } else { 243 } else {
550 LOG(WARNING) << "WriteFile failed: " << error; 244 LOG(WARNING) << "WriteFile failed: " << error;
551 rv = RecordAndMapError(error, 245 rv = RecordAndMapError(error, FILE_ERROR_SOURCE_WRITE);
552 FILE_ERROR_SOURCE_WRITE,
553 record_uma_,
554 bound_net_log_);
555 } 246 }
556 } else if (overlapped) { 247 } else {
557 async_context_->IOCompletionIsPending(callback, buf); 248 IOCompletionIsPending(callback, buf);
558 rv = ERR_IO_PENDING; 249 rv = ERR_IO_PENDING;
559 } else {
560 rv = static_cast<int>(bytes_written);
561 } 250 }
562 return rv; 251 return rv;
563 } 252 }
564 253
565 int FileStreamWin::WriteSync( 254 int FileStream::Context::WriteSync(const char* buf, int buf_len) {
566 const char* buf, int buf_len) {
567 DCHECK(!async_context_.get());
568 base::ThreadRestrictions::AssertIOAllowed(); 255 base::ThreadRestrictions::AssertIOAllowed();
569 256
570 if (!IsOpen())
571 return ERR_UNEXPECTED;
572
573 DCHECK(open_flags_ & base::PLATFORM_FILE_WRITE);
574
575 int rv = 0; 257 int rv = 0;
576 DWORD bytes_written = 0; 258 DWORD bytes_written = 0;
577 if (!WriteFile(file_, buf, buf_len, &bytes_written, NULL)) { 259 if (!WriteFile(file_, buf, buf_len, &bytes_written, NULL)) {
578 DWORD error = GetLastError(); 260 DWORD error = GetLastError();
579 LOG(WARNING) << "WriteFile failed: " << error; 261 LOG(WARNING) << "WriteFile failed: " << error;
580 rv = RecordAndMapError(error, 262 rv = RecordAndMapError(error, FILE_ERROR_SOURCE_WRITE);
581 FILE_ERROR_SOURCE_WRITE,
582 record_uma_,
583 bound_net_log_);
584 } else { 263 } else {
585 rv = static_cast<int>(bytes_written); 264 rv = static_cast<int>(bytes_written);
586 } 265 }
587 return rv; 266 return rv;
588 } 267 }
589 268
590 int FileStreamWin::Flush() { 269 int FileStream::Context::Flush() {
270 if (FlushFileBuffers(file_))
271 return OK;
272
273 return RecordAndMapError(GetLastError(), FILE_ERROR_SOURCE_FLUSH);
274 }
275
276 int FileStream::Context::Truncate(int64 bytes) {
277 BOOL result = SetEndOfFile(file_);
278 if (result)
279 return bytes;
280
281 DWORD error = GetLastError();
282 LOG(WARNING) << "SetEndOfFile failed: " << error;
283 return RecordAndMapError(error, FILE_ERROR_SOURCE_SET_EOF);
284 }
285
286 int FileStream::Context::RecordAndMapError(int error,
287 FileErrorSource source) const {
288 // The following check is against incorrect use or bug. File descriptor
289 // shouldn't ever be closed outside of FileStream while it still tries to do
290 // something with it.
291 DCHECK(error != ERROR_INVALID_HANDLE);
292 net::Error net_error = MapSystemError(error);
293
294 if (!orphaned_) {
295 bound_net_log_.AddEvent(net::NetLog::TYPE_FILE_STREAM_ERROR,
296 base::Bind(&NetLogFileStreamErrorCallback,
297 source, error, net_error));
298 }
299 RecordFileError(error, source, record_uma_);
300 return net_error;
301 }
302
303 void FileStream::Context::BeginOpenEvent(const FilePath& path) {
304 std::string file_name = path.AsUTF8Unsafe();
305 bound_net_log_.BeginEvent(net::NetLog::TYPE_FILE_STREAM_OPEN,
306 NetLog::StringCallback("file_name", &file_name));
307 }
308
309 int FileStream::Context::OpenFileImpl(const FilePath& path, int open_flags) {
310 file_ = base::CreatePlatformFile(path, open_flags, NULL, NULL);
311 if (file_ == base::kInvalidPlatformFileValue)
312 return GetLastError();
313
314 return OK;
315 }
316
317 void FileStream::Context::CheckForOpenError(int* result) {
318 if (file_ == base::kInvalidPlatformFileValue) {
319 bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
320 *result = RecordAndMapError(*result, FILE_ERROR_SOURCE_OPEN);
321 }
322 }
323
324 void FileStream::Context::OnOpenCompleted(const CompletionCallback& callback,
325 int result) {
326 CheckForOpenError(&result);
327 if (!orphaned_)
328 RegisterInMessageLoop();
329 OnAsyncCompleted(callback, result);
330 }
331
332 void FileStream::Context::RegisterInMessageLoop() {
333 if (file_ != base::kInvalidPlatformFileValue)
334 MessageLoopForIO::current()->RegisterIOHandler(file_, this);
335 }
336
337 void FileStream::Context::CloseFileImpl() {
338 if (!base::ClosePlatformFile(file_))
339 NOTREACHED();
340 file_ = base::kInvalidPlatformFileValue;
341 }
342
343 void FileStream::Context::OnCloseCompleted(const CompletionCallback& callback) {
344 if (!orphaned_) {
345 bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
346 // Reset this before Run() as Run() may issue a new async operation.
347 async_in_progress_ = false;
348 callback.Run(OK);
349 } else {
350 delete this;
351 }
352 }
353
354 int64 FileStream::Context::SeekFileImpl(Whence whence, int64 offset) {
591 base::ThreadRestrictions::AssertIOAllowed(); 355 base::ThreadRestrictions::AssertIOAllowed();
592 356
593 if (!IsOpen())
594 return ERR_UNEXPECTED;
595
596 DCHECK(open_flags_ & base::PLATFORM_FILE_WRITE);
597 if (FlushFileBuffers(file_)) {
598 return OK;
599 }
600
601 return RecordAndMapError(GetLastError(),
602 FILE_ERROR_SOURCE_FLUSH,
603 record_uma_,
604 bound_net_log_);
605 }
606
607 int64 FileStreamWin::Truncate(int64 bytes) {
608 base::ThreadRestrictions::AssertIOAllowed();
609
610 if (!IsOpen())
611 return ERR_UNEXPECTED;
612
613 // We'd better be open for writing.
614 DCHECK(open_flags_ & base::PLATFORM_FILE_WRITE);
615
616 // Seek to the position to truncate from.
617 int64 seek_position = SeekSync(FROM_BEGIN, bytes);
618 if (seek_position != bytes)
619 return ERR_UNEXPECTED;
620
621 // And truncate the file.
622 BOOL result = SetEndOfFile(file_);
623 if (!result) {
624 DWORD error = GetLastError();
625 LOG(WARNING) << "SetEndOfFile failed: " << error;
626 return RecordAndMapError(error,
627 FILE_ERROR_SOURCE_SET_EOF,
628 record_uma_,
629 bound_net_log_);
630 }
631
632 // Success.
633 return seek_position;
634 }
635
636 void FileStreamWin::EnableErrorStatistics() {
637 record_uma_ = true;
638
639 if (async_context_.get())
640 async_context_->EnableErrorStatistics();
641 }
642
643 void FileStreamWin::SetBoundNetLogSource(
644 const net::BoundNetLog& owner_bound_net_log) {
645 if ((owner_bound_net_log.source().id == net::NetLog::Source::kInvalidId) &&
646 (bound_net_log_.source().id == net::NetLog::Source::kInvalidId)) {
647 // Both |BoundNetLog|s are invalid.
648 return;
649 }
650
651 // Should never connect to itself.
652 DCHECK_NE(bound_net_log_.source().id, owner_bound_net_log.source().id);
653
654 bound_net_log_.AddEvent(
655 net::NetLog::TYPE_FILE_STREAM_BOUND_TO_OWNER,
656 owner_bound_net_log.source().ToEventParametersCallback());
657
658 owner_bound_net_log.AddEvent(
659 net::NetLog::TYPE_FILE_STREAM_SOURCE,
660 bound_net_log_.source().ToEventParametersCallback());
661 }
662
663 base::PlatformFile FileStreamWin::GetPlatformFileForTesting() {
664 return file_;
665 }
666
667 void FileStreamWin::OnClosed(const CompletionCallback& callback) {
668 file_ = base::kInvalidPlatformFileValue;
669
670 // Reset this before Run() as Run() may issue a new async operation.
671 ResetOnIOComplete();
672 callback.Run(OK);
673 }
674
675 void FileStreamWin::SeekFile(Whence whence, int64 offset, int64* result) {
676 LARGE_INTEGER distance, res; 357 LARGE_INTEGER distance, res;
677 distance.QuadPart = offset; 358 distance.QuadPart = offset;
678 DWORD move_method = static_cast<DWORD>(whence); 359 DWORD move_method = static_cast<DWORD>(whence);
679 if (!SetFilePointerEx(file_, distance, &res, move_method)) { 360 if (SetFilePointerEx(file_, distance, &res, move_method)) {
680 DWORD error = GetLastError(); 361 SetOffset(&io_context_.overlapped, res);
681 LOG(WARNING) << "SetFilePointerEx failed: " << error; 362 return res.QuadPart;
682 *result = RecordAndMapError(error, 363 }
683 FILE_ERROR_SOURCE_SEEK, 364
684 record_uma_, 365 return -static_cast<int>(GetLastError());
685 bound_net_log_); 366 }
367
368 void FileStream::Context::CheckForSeekError(int64* result) {
369 if (*result < 0) {
370 *result = RecordAndMapError(static_cast<int>(-(*result)),
371 FILE_ERROR_SOURCE_SEEK);
372 }
373 }
374
375 void FileStream::Context::OnSeekCompleted(
376 const Int64CompletionCallback& callback,
377 int64 result) {
378 CheckForSeekError(&result);
379 OnAsyncCompleted(callback, result);
380 }
381
382 void FileStream::Context::IOCompletionIsPending(
383 const CompletionCallback& callback,
384 IOBuffer* buf) {
385 DCHECK(callback_.is_null());
386 callback_ = callback;
387 in_flight_buf_ = buf; // Hold until the async operation ends.
388 async_in_progress_ = true;
389 }
390
391 void FileStream::Context::OnIOCompleted(MessageLoopForIO::IOContext* context,
392 DWORD bytes_read,
393 DWORD error) {
394 DCHECK_EQ(&io_context_, context);
395 DCHECK(!callback_.is_null());
396
397 if (orphaned_) {
398 callback_.Reset();
399 in_flight_buf_ = NULL;
400 CloseAsync(CompletionCallback());
686 return; 401 return;
687 } 402 }
688 if (async_context_.get()) { 403
689 async_context_->set_error_source(FILE_ERROR_SOURCE_SEEK); 404 int result = static_cast<int>(bytes_read);
690 SetOffset(async_context_->overlapped(), res); 405 if (error && error != ERROR_HANDLE_EOF)
691 } 406 result = RecordAndMapError(error, error_source_);
692 *result = res.QuadPart; 407
693 } 408 if (bytes_read)
694 409 IncrementOffset(&io_context_.overlapped, bytes_read);
695 void FileStreamWin::OnOpened(const CompletionCallback& callback, int* result) {
696 if (*result == OK) {
697 async_context_.reset(new AsyncContext(bound_net_log_));
698 if (record_uma_)
699 async_context_->EnableErrorStatistics();
700 MessageLoopForIO::current()->RegisterIOHandler(file_,
701 async_context_.get());
702 }
703 410
704 // Reset this before Run() as Run() may issue a new async operation. 411 // Reset this before Run() as Run() may issue a new async operation.
705 ResetOnIOComplete(); 412 async_in_progress_ = false;
706 callback.Run(*result); 413 CompletionCallback temp_callback = callback_;
707 } 414 callback_.Reset();
708 415 scoped_refptr<IOBuffer> temp_buf = in_flight_buf_;
709 void FileStreamWin::OnSeeked( 416 in_flight_buf_ = NULL;
710 const Int64CompletionCallback& callback, 417 temp_callback.Run(result);
711 int64* result) { 418 }
712 // Reset this before Run() as Run() may issue a new async operation. 419
713 ResetOnIOComplete(); 420 template <typename R>
714 callback.Run(*result); 421 void FileStream::Context::OnAsyncCompleted(
715 } 422 const base::Callback<void(R)>& callback,
716 423 R result) {
717 void FileStreamWin::ResetOnIOComplete() { 424 // Reset this before Run() as Run() may issue a new async operation. Also it
718 on_io_complete_.reset(); 425 // should be reset before CloseAsync() because it shouldn't run if any async
719 weak_ptr_factory_.InvalidateWeakPtrs(); 426 // operation is in progress.
720 } 427 async_in_progress_ = false;
721 428 if (orphaned_)
722 void FileStreamWin::WaitForIOCompletion() { 429 CloseAsync(CompletionCallback());
723 // http://crbug.com/115067 430 else
724 base::ThreadRestrictions::ScopedAllowWait allow_wait; 431 callback.Run(result);
725 if (on_io_complete_.get()) {
726 on_io_complete_->Wait();
727 on_io_complete_.reset();
728 }
729 } 432 }
730 433
731 } // namespace net 434 } // namespace net
OLDNEW

Powered by Google App Engine
This is Rietveld 408576698