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

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::AsyncContext::AsyncContext(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 destroyed_(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::AsyncContext::AsyncContext(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 destroyed_(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 void FileStream::AsyncContext::Destroy() {
74 destroyed_ = true;
75 CancelIo(file_);
willchan no longer on Chromium 2012/08/27 06:59:10 Is this correct? What happens when |file_| is inva
pivanof 2012/08/27 08:43:31 Any test working in sync mode should trip over thi
76 if (!async_in_progress_)
77 DeleteAbandoned();
78 }
136 79
137 void EnableErrorStatistics() { 80 void FileStream::AsyncContext::OpenAsync(
138 record_uma_ = true; 81 const FilePath& path,
139 } 82 int open_flags,
83 const CompletionCallback& callback) {
84 DCHECK(!async_in_progress_);
140 85
141 private: 86 BeginOpenEvent(path);
142 virtual void OnIOCompleted(MessageLoopForIO::IOContext* context,
143 DWORD bytes_read, DWORD error) OVERRIDE;
144 87
145 MessageLoopForIO::IOContext context_; 88 const bool posted = base::PostTaskAndReplyWithResult(
146 CompletionCallback callback_; 89 base::WorkerPool::GetTaskRunner(true /* task_is_slow */),
147 scoped_refptr<IOBuffer> in_flight_buf_; 90 FROM_HERE,
148 bool is_closing_; 91 base::Bind(&AsyncContext::OpenFileImpl,
149 bool record_uma_; 92 base::Unretained(this), path, open_flags),
150 const net::BoundNetLog bound_net_log_; 93 base::Bind(&AsyncContext::OnOpenCompleted,
151 FileErrorSource error_source_; 94 base::Unretained(this), callback));
152 }; 95 DCHECK(posted);
153 96
154 FileStreamWin::AsyncContext::~AsyncContext() { 97 async_in_progress_ = true;
155 is_closing_ = true; 98 }
156 bool waited = false; 99
157 base::TimeTicks start = base::TimeTicks::Now(); 100 int FileStream::AsyncContext::OpenSync(const FilePath& path, int open_flags) {
158 while (!callback_.is_null()) { 101 BeginOpenEvent(path);
159 waited = true; 102 int result = OpenFileImpl(path, open_flags);
160 MessageLoopForIO::current()->WaitForIOCompletion(INFINITE, this); 103 CheckForOpenError(&result);
161 } 104 // TODO(satorux): Remove this once all async clients are migrated to use
162 if (waited) { 105 // Open(). crbug.com/114783
163 // We want to see if we block the message loop for too long. 106 if (open_flags & base::PLATFORM_FILE_ASYNC)
164 UMA_HISTOGRAM_TIMES("AsyncIO.FileStreamClose", 107 RegisterInMessageLoop();
165 base::TimeTicks::Now() - start); 108 return result;
109 }
110
111 void FileStream::AsyncContext::CloseAsync(
112 const CompletionCallback& callback) {
113 DCHECK(!async_in_progress_);
114
115 bound_net_log_.AddEvent(net::NetLog::TYPE_FILE_STREAM_CLOSE);
116
117 if (file_ == base::kInvalidPlatformFileValue) {
118 MessageLoop::current()->PostTask(
119 FROM_HERE,
120 base::Bind(&AsyncContext::OnAsyncCompleted<int>,
121 base::Unretained(this), callback, OK));
122 } else {
123 const bool posted = base::WorkerPool::PostTaskAndReply(
124 FROM_HERE,
125 base::Bind(&AsyncContext::CloseFileImpl,
126 base::Unretained(this)),
127 base::Bind(&AsyncContext::OnCloseCompleted,
128 base::Unretained(this), callback),
129 true /* task_is_slow */);
130 DCHECK(posted);
131
132 async_in_progress_ = true;
166 } 133 }
167 } 134 }
168 135
169 void FileStreamWin::AsyncContext::IOCompletionIsPending( 136 void FileStream::AsyncContext::CloseSync() {
170 const CompletionCallback& callback, 137 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); 138 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) { 139 if (file_ != base::kInvalidPlatformFileValue) {
303 if (!base::ClosePlatformFile(file_)) 140 CloseFileImpl();
304 NOTREACHED();
305 file_ = base::kInvalidPlatformFileValue;
306
307 bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN); 141 bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
308 } 142 }
309 } 143 }
310 144
311 int FileStreamWin::Open(const FilePath& path, int open_flags, 145 void FileStream::AsyncContext::SeekAsync(
312 const CompletionCallback& callback) { 146 Whence whence,
313 if (IsOpen()) { 147 int64 offset,
314 DLOG(FATAL) << "File is already open!"; 148 const Int64CompletionCallback& callback) {
315 return ERR_UNEXPECTED; 149 DCHECK(!async_in_progress_);
316 }
317 150
318 open_flags_ = open_flags; 151 int64* result = new int64(-1);
319 DCHECK(open_flags_ & base::PLATFORM_FILE_ASYNC); 152 const bool posted = base::PostTaskAndReplyWithResult(
320 DCHECK(!weak_ptr_factory_.HasWeakPtrs()); 153 base::WorkerPool::GetTaskRunner(true /* task is slow */),
321 DCHECK(!on_io_complete_.get()); 154 FROM_HERE,
322 on_io_complete_.reset(new base::WaitableEvent( 155 base::Bind(&AsyncContext::SeekFileImpl,
323 false /* manual_reset */, false /* initially_signaled */)); 156 base::Unretained(this), whence, offset),
157 base::Bind(&AsyncContext::OnSeekCompleted,
158 base::Unretained(this), callback));
159 DCHECK(posted);
324 160
325 // Passing &file_ to a thread pool looks unsafe but it's safe here as the 161 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 } 162 }
342 163
343 int FileStreamWin::OpenSync(const FilePath& path, int open_flags) { 164 int64 FileStream::AsyncContext::SeekSync(Whence whence, int64 offset) {
344 if (IsOpen()) { 165 int64 result = SeekFileImpl(whence, offset);
345 DLOG(FATAL) << "File is already open!"; 166 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; 167 return result;
411 } 168 }
412 169
413 int64 FileStreamWin::Available() { 170 int64 FileStream::AsyncContext::GetFileSize() {
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; 171 LARGE_INTEGER file_size;
424 if (!GetFileSizeEx(file_, &file_size)) { 172 if (!GetFileSizeEx(file_, &file_size)) {
425 DWORD error = GetLastError(); 173 DWORD error = GetLastError();
426 LOG(WARNING) << "GetFileSizeEx failed: " << error; 174 LOG(WARNING) << "GetFileSizeEx failed: " << error;
427 return RecordAndMapError(error, 175 return RecordAndMapError(error, FILE_ERROR_SOURCE_GET_SIZE);
428 FILE_ERROR_SOURCE_GET_SIZE,
429 record_uma_,
430 bound_net_log_);
431 } 176 }
432 177
433 return file_size.QuadPart - cur_pos; 178 return file_size.QuadPart;
434 } 179 }
435 180
436 int FileStreamWin::Read( 181 int FileStream::AsyncContext::ReadAsync(
437 IOBuffer* buf, int buf_len, const CompletionCallback& callback) { 182 IOBuffer* buf,
438 DCHECK(async_context_.get()); 183 int buf_len,
439 184 const CompletionCallback& callback) {
440 if (!IsOpen()) 185 DCHECK(!async_in_progress_);
441 return ERR_UNEXPECTED; 186 error_source_ = FILE_ERROR_SOURCE_READ;
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::AsyncContext::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::AsyncContext::WriteAsync(
507 int to_read = buf_len; 231 IOBuffer* buf,
508 int bytes_total = 0; 232 int buf_len,
509 233 const CompletionCallback& callback) {
510 do { 234 error_source_ = FILE_ERROR_SOURCE_WRITE;
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 235
542 int rv = 0; 236 int rv = 0;
543 DWORD bytes_written = 0; 237 DWORD bytes_written = 0;
544 if (!WriteFile(file_, buf->data(), buf_len, &bytes_written, overlapped)) { 238 if (!WriteFile(file_, buf->data(), buf_len,
239 &bytes_written, &io_context_.overlapped)) {
545 DWORD error = GetLastError(); 240 DWORD error = GetLastError();
546 if (error == ERROR_IO_PENDING) { 241 if (error == ERROR_IO_PENDING) {
547 async_context_->IOCompletionIsPending(callback, buf); 242 IOCompletionIsPending(callback, buf);
548 rv = ERR_IO_PENDING; 243 rv = ERR_IO_PENDING;
549 } else { 244 } else {
550 LOG(WARNING) << "WriteFile failed: " << error; 245 LOG(WARNING) << "WriteFile failed: " << error;
551 rv = RecordAndMapError(error, 246 rv = RecordAndMapError(error, FILE_ERROR_SOURCE_WRITE);
552 FILE_ERROR_SOURCE_WRITE,
553 record_uma_,
554 bound_net_log_);
555 } 247 }
556 } else if (overlapped) { 248 } else {
557 async_context_->IOCompletionIsPending(callback, buf); 249 IOCompletionIsPending(callback, buf);
558 rv = ERR_IO_PENDING; 250 rv = ERR_IO_PENDING;
559 } else {
560 rv = static_cast<int>(bytes_written);
561 } 251 }
562 return rv; 252 return rv;
563 } 253 }
564 254
565 int FileStreamWin::WriteSync( 255 int FileStream::AsyncContext::WriteSync(const char* buf, int buf_len) {
566 const char* buf, int buf_len) {
567 DCHECK(!async_context_.get());
568 base::ThreadRestrictions::AssertIOAllowed(); 256 base::ThreadRestrictions::AssertIOAllowed();
569 257
570 if (!IsOpen())
571 return ERR_UNEXPECTED;
572
573 DCHECK(open_flags_ & base::PLATFORM_FILE_WRITE);
574
575 int rv = 0; 258 int rv = 0;
576 DWORD bytes_written = 0; 259 DWORD bytes_written = 0;
577 if (!WriteFile(file_, buf, buf_len, &bytes_written, NULL)) { 260 if (!WriteFile(file_, buf, buf_len, &bytes_written, NULL)) {
578 DWORD error = GetLastError(); 261 DWORD error = GetLastError();
579 LOG(WARNING) << "WriteFile failed: " << error; 262 LOG(WARNING) << "WriteFile failed: " << error;
580 rv = RecordAndMapError(error, 263 rv = RecordAndMapError(error, FILE_ERROR_SOURCE_WRITE);
581 FILE_ERROR_SOURCE_WRITE,
582 record_uma_,
583 bound_net_log_);
584 } else { 264 } else {
585 rv = static_cast<int>(bytes_written); 265 rv = static_cast<int>(bytes_written);
586 } 266 }
587 return rv; 267 return rv;
588 } 268 }
589 269
590 int FileStreamWin::Flush() { 270 int FileStream::AsyncContext::Flush() {
271 if (FlushFileBuffers(file_))
272 return OK;
273
274 return RecordAndMapError(GetLastError(), FILE_ERROR_SOURCE_FLUSH);
275 }
276
277 int FileStream::AsyncContext::Truncate(int64 bytes) {
278 BOOL result = SetEndOfFile(file_);
279 if (result)
280 return bytes;
281
282 DWORD error = GetLastError();
283 LOG(WARNING) << "SetEndOfFile failed: " << error;
284 return RecordAndMapError(error, FILE_ERROR_SOURCE_SET_EOF);
285 }
286
287 int FileStream::AsyncContext::RecordAndMapError(int error,
288 FileErrorSource source) {
289 // The following check is against incorrect use or bug. File descriptor
290 // shouldn't ever be closed outside of FileStream while it still tries to do
291 // something with it.
292 DCHECK(error != ERROR_INVALID_HANDLE);
293 net::Error net_error = MapSystemError(error);
294
295 if (!destroyed_) {
296 bound_net_log_.AddEvent(net::NetLog::TYPE_FILE_STREAM_ERROR,
297 base::Bind(&NetLogFileStreamErrorCallback,
298 source, error, net_error));
299 }
300 RecordFileError(error, source, record_uma_);
301 return net_error;
302 }
303
304 void FileStream::AsyncContext::BeginOpenEvent(const FilePath& path) {
305 std::string file_name = path.AsUTF8Unsafe();
306 bound_net_log_.BeginEvent(net::NetLog::TYPE_FILE_STREAM_OPEN,
307 NetLog::StringCallback("file_name", &file_name));
308 }
309
310 int FileStream::AsyncContext::OpenFileImpl(const FilePath& path,
311 int open_flags) {
312 if (destroyed_)
313 return OK;
314
315 file_ = base::CreatePlatformFile(path, open_flags, NULL, NULL);
316 if (file_ == base::kInvalidPlatformFileValue)
317 return GetLastError();
318
319 return OK;
320 }
321
322 void FileStream::AsyncContext::CheckForOpenError(int* result) {
323 if (file_ == base::kInvalidPlatformFileValue) {
324 bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
325 *result = RecordAndMapError(*result, FILE_ERROR_SOURCE_OPEN);
326 }
327 }
328
329 void FileStream::AsyncContext::OnOpenCompleted(
330 const CompletionCallback& callback,
331 int result) {
332 CheckForOpenError(&result);
333 if (!destroyed_)
334 RegisterInMessageLoop();
335 OnAsyncCompleted(callback, result);
336 }
337
338 void FileStream::AsyncContext::RegisterInMessageLoop() {
339 if (file_ != base::kInvalidPlatformFileValue)
340 MessageLoopForIO::current()->RegisterIOHandler(file_, this);
341 }
342
343 void FileStream::AsyncContext::CloseFileImpl() {
344 if (!base::ClosePlatformFile(file_))
345 NOTREACHED();
346 file_ = base::kInvalidPlatformFileValue;
347 }
348
349 void FileStream::AsyncContext::OnCloseCompleted(
350 const CompletionCallback& callback) {
351 if (!destroyed_)
352 bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
353 OnAsyncCompleted(callback, static_cast<int>(OK));
354 }
355
356 int64 FileStream::AsyncContext::SeekFileImpl(Whence whence, int64 offset) {
591 base::ThreadRestrictions::AssertIOAllowed(); 357 base::ThreadRestrictions::AssertIOAllowed();
592 358
593 if (!IsOpen()) 359 // If context has been already destroyed nobody waits for operation results.
594 return ERR_UNEXPECTED; 360 if (destroyed_)
willchan no longer on Chromium 2012/08/27 06:59:10 Remove this
595 361 return 0;
596 DCHECK(open_flags_ & base::PLATFORM_FILE_WRITE); 362
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; 363 LARGE_INTEGER distance, res;
677 distance.QuadPart = offset; 364 distance.QuadPart = offset;
678 DWORD move_method = static_cast<DWORD>(whence); 365 DWORD move_method = static_cast<DWORD>(whence);
679 if (!SetFilePointerEx(file_, distance, &res, move_method)) { 366 if (SetFilePointerEx(file_, distance, &res, move_method)) {
680 DWORD error = GetLastError(); 367 SetOffset(&io_context_.overlapped, res);
681 LOG(WARNING) << "SetFilePointerEx failed: " << error; 368 return res.QuadPart;
682 *result = RecordAndMapError(error, 369 }
683 FILE_ERROR_SOURCE_SEEK, 370
684 record_uma_, 371 return -static_cast<int>(GetLastError());
685 bound_net_log_); 372 }
373
374 void FileStream::AsyncContext::CheckForSeekError(int64* result) {
375 if (*result < 0) {
376 *result = RecordAndMapError(static_cast<int>(-(*result)),
377 FILE_ERROR_SOURCE_SEEK);
378 }
379 }
380
381 void FileStream::AsyncContext::OnSeekCompleted(
382 const CompletionCallback64& callback,
383 int64 result) {
384 CheckForSeekError(&result);
385 OnAsyncCompleted(callback, result);
386 }
387
388 void FileStream::AsyncContext::IOCompletionIsPending(
389 const CompletionCallback& callback,
390 IOBuffer* buf) {
391 DCHECK(callback_.is_null());
392 callback_ = callback;
393 in_flight_buf_ = buf; // Hold until the async operation ends.
394 async_in_progress_ = true;
395 }
396
397 void FileStream::AsyncContext::OnIOCompleted(
398 MessageLoopForIO::IOContext* context,
399 DWORD bytes_read,
400 DWORD error) {
401 DCHECK_EQ(&io_context_, context);
402 DCHECK(!callback_.is_null());
403
404 if (destroyed_) {
405 callback_.Reset();
406 in_flight_buf_ = NULL;
407 DeleteAbandoned();
686 return; 408 return;
687 } 409 }
688 if (async_context_.get()) { 410
689 async_context_->set_error_source(FILE_ERROR_SOURCE_SEEK); 411 int result = static_cast<int>(bytes_read);
690 SetOffset(async_context_->overlapped(), res); 412 if (error && error != ERROR_HANDLE_EOF)
691 } 413 result = RecordAndMapError(error, error_source_);
692 *result = res.QuadPart; 414
693 } 415 if (bytes_read)
694 416 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 417
704 // Reset this before Run() as Run() may issue a new async operation. 418 // Reset this before Run() as Run() may issue a new async operation.
705 ResetOnIOComplete(); 419 async_in_progress_ = false;
706 callback.Run(*result); 420 CompletionCallback temp_callback = callback_;
707 } 421 callback_.Reset();
708 422 scoped_refptr<IOBuffer> temp_buf = in_flight_buf_;
709 void FileStreamWin::OnSeeked( 423 in_flight_buf_ = NULL;
710 const Int64CompletionCallback& callback, 424 temp_callback.Run(result);
711 int64* result) { 425 }
712 // Reset this before Run() as Run() may issue a new async operation. 426
713 ResetOnIOComplete(); 427 template <typename R>
714 callback.Run(*result); 428 void FileStream::AsyncContext::OnAsyncCompleted(
715 } 429 const base::Callback<void(R)>& callback,
716 430 R result) {
717 void FileStreamWin::ResetOnIOComplete() { 431 if (destroyed_) {
718 on_io_complete_.reset(); 432 DeleteAbandoned();
719 weak_ptr_factory_.InvalidateWeakPtrs(); 433 } else {
720 } 434 // Reset this before Run() as Run() may issue a new async operation.
721 435 async_in_progress_ = false;
722 void FileStreamWin::WaitForIOCompletion() { 436 callback.Run(result);
723 // http://crbug.com/115067 437 }
724 base::ThreadRestrictions::ScopedAllowWait allow_wait; 438 }
725 if (on_io_complete_.get()) { 439
726 on_io_complete_->Wait(); 440 void FileStream::AsyncContext::DeleteAbandoned() {
727 on_io_complete_.reset(); 441 if (file_ != base::kInvalidPlatformFileValue) {
728 } 442 const bool posted = base::WorkerPool::PostTask(
729 } 443 FROM_HERE,
730 444 // Context should be deleted after closing, thus Owned().
445 base::Bind(&AsyncContext::CloseFileImpl, base::Owned(this)),
446 true /* task_is_slow */);
447 DCHECK(posted);
448 } else {
449 delete this;
450 }
451 }
452
731 } // namespace net 453 } // namespace net
OLDNEW

Powered by Google App Engine
This is Rietveld 408576698