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

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"
12 #include "base/metrics/histogram.h" 11 #include "base/metrics/histogram.h"
13 #include "base/synchronization/waitable_event.h" 12 #include "base/task_runner_util.h"
14 #include "base/threading/thread_restrictions.h"
15 #include "base/threading/worker_pool.h" 13 #include "base/threading/worker_pool.h"
16 #include "net/base/file_stream_metrics.h"
17 #include "net/base/file_stream_net_log_parameters.h"
18 #include "net/base/io_buffer.h" 14 #include "net/base/io_buffer.h"
19 #include "net/base/net_errors.h" 15 #include "net/base/net_errors.h"
20 16
21 namespace net { 17 namespace net {
22 18
23 // Ensure that we can just use our Whence values directly. 19 // Ensure that we can just use our Whence values directly.
24 COMPILE_ASSERT(FROM_BEGIN == FILE_BEGIN, bad_whence_begin); 20 COMPILE_ASSERT(FROM_BEGIN == FILE_BEGIN, bad_whence_begin);
25 COMPILE_ASSERT(FROM_CURRENT == FILE_CURRENT, bad_whence_current); 21 COMPILE_ASSERT(FROM_CURRENT == FILE_CURRENT, bad_whence_current);
26 COMPILE_ASSERT(FROM_END == FILE_END, bad_whence_end); 22 COMPILE_ASSERT(FROM_END == FILE_END, bad_whence_end);
27 23
28 namespace { 24 namespace {
29 25
30 void SetOffset(OVERLAPPED* overlapped, const LARGE_INTEGER& offset) { 26 void SetOffset(OVERLAPPED* overlapped, const LARGE_INTEGER& offset) {
31 overlapped->Offset = offset.LowPart; 27 overlapped->Offset = offset.LowPart;
32 overlapped->OffsetHigh = offset.HighPart; 28 overlapped->OffsetHigh = offset.HighPart;
33 } 29 }
34 30
35 void IncrementOffset(OVERLAPPED* overlapped, DWORD count) { 31 void IncrementOffset(OVERLAPPED* overlapped, DWORD count) {
36 LARGE_INTEGER offset; 32 LARGE_INTEGER offset;
37 offset.LowPart = overlapped->Offset; 33 offset.LowPart = overlapped->Offset;
38 offset.HighPart = overlapped->OffsetHigh; 34 offset.HighPart = overlapped->OffsetHigh;
39 offset.QuadPart += static_cast<LONGLONG>(count); 35 offset.QuadPart += static_cast<LONGLONG>(count);
40 SetOffset(overlapped, offset); 36 SetOffset(overlapped, offset);
41 } 37 }
42 38
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 39 } // namespace
116 40
117 // FileStreamWin::AsyncContext ---------------------------------------------- 41 FileStream::Context::Context(const BoundNetLog& bound_net_log)
42 : io_context_(),
43 file_(base::kInvalidPlatformFileValue),
44 record_uma_(false),
45 async_in_progress_(false),
46 orphaned_(false),
47 bound_net_log_(bound_net_log),
48 error_source_(FILE_ERROR_SOURCE_COUNT) {
49 io_context_.handler = this;
50 }
118 51
119 class FileStreamWin::AsyncContext : public MessageLoopForIO::IOHandler { 52 FileStream::Context::Context(base::PlatformFile file,
120 public: 53 const BoundNetLog& bound_net_log,
121 explicit AsyncContext(const net::BoundNetLog& bound_net_log) 54 int open_flags)
122 : context_(), is_closing_(false), 55 : io_context_(),
123 record_uma_(false), bound_net_log_(bound_net_log), 56 file_(file),
124 error_source_(FILE_ERROR_SOURCE_COUNT) { 57 record_uma_(false),
125 context_.handler = this; 58 async_in_progress_(false),
126 } 59 orphaned_(false),
127 ~AsyncContext(); 60 bound_net_log_(bound_net_log),
128 61 error_source_(FILE_ERROR_SOURCE_COUNT) {
129 void IOCompletionIsPending(const CompletionCallback& callback, 62 io_context_.handler = this;
130 IOBuffer* buf); 63 if (file_ != base::kInvalidPlatformFileValue &&
131 64 (open_flags & base::PLATFORM_FILE_ASYNC)) {
132 OVERLAPPED* overlapped() { return &context_.overlapped; } 65 OnAsyncFileOpened();
133 const CompletionCallback& callback() const { return callback_; }
134
135 void set_error_source(FileErrorSource source) { error_source_ = source; }
136
137 void EnableErrorStatistics() {
138 record_uma_ = true;
139 }
140
141 private:
142 virtual void OnIOCompleted(MessageLoopForIO::IOContext* context,
143 DWORD bytes_read, DWORD error) OVERRIDE;
144
145 MessageLoopForIO::IOContext context_;
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
154 FileStreamWin::AsyncContext::~AsyncContext() {
155 is_closing_ = true;
156 bool waited = false;
157 base::TimeTicks start = base::TimeTicks::Now();
158 while (!callback_.is_null()) {
159 waited = true;
160 MessageLoopForIO::current()->WaitForIOCompletion(INFINITE, this);
161 }
162 if (waited) {
163 // We want to see if we block the message loop for too long.
164 UMA_HISTOGRAM_TIMES("AsyncIO.FileStreamClose",
165 base::TimeTicks::Now() - start);
166 } 66 }
167 } 67 }
168 68
169 void FileStreamWin::AsyncContext::IOCompletionIsPending( 69 FileStream::Context::~Context() {
170 const CompletionCallback& callback,
171 IOBuffer* buf) {
172 DCHECK(callback_.is_null());
173 callback_ = callback;
174 in_flight_buf_ = buf; // Hold until the async operation ends.
175 } 70 }
176 71
177 void FileStreamWin::AsyncContext::OnIOCompleted( 72 int64 FileStream::Context::GetFileSize() const {
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);
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) {
303 if (!base::ClosePlatformFile(file_))
304 NOTREACHED();
305 file_ = base::kInvalidPlatformFileValue;
306
307 bound_net_log_.EndEvent(net::NetLog::TYPE_FILE_STREAM_OPEN);
308 }
309 }
310
311 int FileStreamWin::Open(const FilePath& path, int open_flags,
312 const CompletionCallback& callback) {
313 if (IsOpen()) {
314 DLOG(FATAL) << "File is already open!";
315 return ERR_UNEXPECTED;
316 }
317
318 open_flags_ = open_flags;
319 DCHECK(open_flags_ & base::PLATFORM_FILE_ASYNC);
320 DCHECK(!weak_ptr_factory_.HasWeakPtrs());
321 DCHECK(!on_io_complete_.get());
322 on_io_complete_.reset(new base::WaitableEvent(
323 false /* manual_reset */, false /* initially_signaled */));
324
325 // Passing &file_ to a thread pool looks unsafe but it's safe here as the
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 }
342
343 int FileStreamWin::OpenSync(const FilePath& path, int open_flags) {
344 if (IsOpen()) {
345 DLOG(FATAL) << "File is already open!";
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;
411 }
412
413 int64 FileStreamWin::Available() {
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; 73 LARGE_INTEGER file_size;
424 if (!GetFileSizeEx(file_, &file_size)) { 74 if (!GetFileSizeEx(file_, &file_size)) {
425 DWORD error = GetLastError(); 75 DWORD error = GetLastError();
426 LOG(WARNING) << "GetFileSizeEx failed: " << error; 76 LOG(WARNING) << "GetFileSizeEx failed: " << error;
427 return RecordAndMapError(error, 77 return RecordAndMapError(error, FILE_ERROR_SOURCE_GET_SIZE);
428 FILE_ERROR_SOURCE_GET_SIZE,
429 record_uma_,
430 bound_net_log_);
431 } 78 }
432 79
433 return file_size.QuadPart - cur_pos; 80 return file_size.QuadPart;
434 } 81 }
435 82
436 int FileStreamWin::Read( 83 int FileStream::Context::ReadAsync(IOBuffer* buf,
437 IOBuffer* buf, int buf_len, const CompletionCallback& callback) { 84 int buf_len,
438 DCHECK(async_context_.get()); 85 const CompletionCallback& callback) {
439 86 DCHECK(!async_in_progress_);
440 if (!IsOpen()) 87 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 88
451 int rv = 0; 89 int rv = 0;
452 90
453 DWORD bytes_read; 91 DWORD bytes_read;
454 if (!ReadFile(file_, buf->data(), buf_len, &bytes_read, overlapped)) { 92 if (!ReadFile(file_, buf->data(), buf_len,
93 &bytes_read, &io_context_.overlapped)) {
455 DWORD error = GetLastError(); 94 DWORD error = GetLastError();
456 if (error == ERROR_IO_PENDING) { 95 if (error == ERROR_IO_PENDING) {
457 async_context_->IOCompletionIsPending(callback, buf); 96 IOCompletionIsPending(callback, buf);
458 rv = ERR_IO_PENDING; 97 rv = ERR_IO_PENDING;
459 } else if (error == ERROR_HANDLE_EOF) { 98 } else if (error == ERROR_HANDLE_EOF) {
460 rv = 0; // Report EOF by returning 0 bytes read. 99 rv = 0; // Report EOF by returning 0 bytes read.
461 } else { 100 } else {
462 LOG(WARNING) << "ReadFile failed: " << error; 101 LOG(WARNING) << "ReadFile failed: " << error;
463 rv = RecordAndMapError(error, 102 rv = RecordAndMapError(error, FILE_ERROR_SOURCE_READ);
464 FILE_ERROR_SOURCE_READ,
465 record_uma_,
466 bound_net_log_);
467 } 103 }
468 } else if (overlapped) { 104 } else {
469 async_context_->IOCompletionIsPending(callback, buf); 105 IOCompletionIsPending(callback, buf);
470 rv = ERR_IO_PENDING; 106 rv = ERR_IO_PENDING;
471 } else {
472 rv = static_cast<int>(bytes_read);
473 } 107 }
474 return rv; 108 return rv;
475 } 109 }
476 110
477 int FileStreamWin::ReadSync(char* buf, int buf_len) { 111 int FileStream::Context::ReadSync(char* buf, int buf_len) {
478 DCHECK(!async_context_.get());
479 base::ThreadRestrictions::AssertIOAllowed();
480
481 if (!IsOpen())
482 return ERR_UNEXPECTED;
483
484 DCHECK(open_flags_ & base::PLATFORM_FILE_READ);
485
486 int rv = 0; 112 int rv = 0;
487 113
488 DWORD bytes_read; 114 DWORD bytes_read;
489 if (!ReadFile(file_, buf, buf_len, &bytes_read, NULL)) { 115 if (!ReadFile(file_, buf, buf_len, &bytes_read, NULL)) {
490 DWORD error = GetLastError(); 116 DWORD error = GetLastError();
491 if (error == ERROR_HANDLE_EOF) { 117 if (error == ERROR_HANDLE_EOF) {
492 rv = 0; // Report EOF by returning 0 bytes read. 118 rv = 0; // Report EOF by returning 0 bytes read.
493 } else { 119 } else {
494 LOG(WARNING) << "ReadFile failed: " << error; 120 LOG(WARNING) << "ReadFile failed: " << error;
495 rv = RecordAndMapError(error, 121 rv = RecordAndMapError(error, FILE_ERROR_SOURCE_READ);
496 FILE_ERROR_SOURCE_READ,
497 record_uma_,
498 bound_net_log_);
499 } 122 }
500 } else { 123 } else {
501 rv = static_cast<int>(bytes_read); 124 rv = static_cast<int>(bytes_read);
502 } 125 }
503 return rv; 126 return rv;
504 } 127 }
505 128
506 int FileStreamWin::ReadUntilComplete(char *buf, int buf_len) { 129 int FileStream::Context::WriteAsync(IOBuffer* buf,
507 int to_read = buf_len; 130 int buf_len,
508 int bytes_total = 0; 131 const CompletionCallback& callback) {
509 132 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 133
542 int rv = 0; 134 int rv = 0;
543 DWORD bytes_written = 0; 135 DWORD bytes_written = 0;
544 if (!WriteFile(file_, buf->data(), buf_len, &bytes_written, overlapped)) { 136 if (!WriteFile(file_, buf->data(), buf_len,
137 &bytes_written, &io_context_.overlapped)) {
545 DWORD error = GetLastError(); 138 DWORD error = GetLastError();
546 if (error == ERROR_IO_PENDING) { 139 if (error == ERROR_IO_PENDING) {
547 async_context_->IOCompletionIsPending(callback, buf); 140 IOCompletionIsPending(callback, buf);
548 rv = ERR_IO_PENDING; 141 rv = ERR_IO_PENDING;
549 } else { 142 } else {
550 LOG(WARNING) << "WriteFile failed: " << error; 143 LOG(WARNING) << "WriteFile failed: " << error;
551 rv = RecordAndMapError(error, 144 rv = RecordAndMapError(error, FILE_ERROR_SOURCE_WRITE);
552 FILE_ERROR_SOURCE_WRITE,
553 record_uma_,
554 bound_net_log_);
555 } 145 }
556 } else if (overlapped) { 146 } else {
557 async_context_->IOCompletionIsPending(callback, buf); 147 IOCompletionIsPending(callback, buf);
558 rv = ERR_IO_PENDING; 148 rv = ERR_IO_PENDING;
149 }
150 return rv;
151 }
152
153 int FileStream::Context::WriteSync(const char* buf, int buf_len) {
154 int rv = 0;
155 DWORD bytes_written = 0;
156 if (!WriteFile(file_, buf, buf_len, &bytes_written, NULL)) {
157 DWORD error = GetLastError();
158 LOG(WARNING) << "WriteFile failed: " << error;
159 rv = RecordAndMapError(error, FILE_ERROR_SOURCE_WRITE);
559 } else { 160 } else {
560 rv = static_cast<int>(bytes_written); 161 rv = static_cast<int>(bytes_written);
561 } 162 }
562 return rv; 163 return rv;
563 } 164 }
564 165
565 int FileStreamWin::WriteSync( 166 int FileStream::Context::Flush() {
566 const char* buf, int buf_len) { 167 if (FlushFileBuffers(file_))
567 DCHECK(!async_context_.get()); 168 return OK;
568 base::ThreadRestrictions::AssertIOAllowed();
569 169
570 if (!IsOpen()) 170 return RecordAndMapError(GetLastError(), FILE_ERROR_SOURCE_FLUSH);
571 return ERR_UNEXPECTED;
572
573 DCHECK(open_flags_ & base::PLATFORM_FILE_WRITE);
574
575 int rv = 0;
576 DWORD bytes_written = 0;
577 if (!WriteFile(file_, buf, buf_len, &bytes_written, NULL)) {
578 DWORD error = GetLastError();
579 LOG(WARNING) << "WriteFile failed: " << error;
580 rv = RecordAndMapError(error,
581 FILE_ERROR_SOURCE_WRITE,
582 record_uma_,
583 bound_net_log_);
584 } else {
585 rv = static_cast<int>(bytes_written);
586 }
587 return rv;
588 } 171 }
589 172
590 int FileStreamWin::Flush() { 173 int FileStream::Context::Truncate(int64 bytes) {
591 base::ThreadRestrictions::AssertIOAllowed(); 174 BOOL result = SetEndOfFile(file_);
175 if (result)
176 return bytes;
592 177
593 if (!IsOpen()) 178 DWORD error = GetLastError();
594 return ERR_UNEXPECTED; 179 LOG(WARNING) << "SetEndOfFile failed: " << error;
180 return RecordAndMapError(error, FILE_ERROR_SOURCE_SET_EOF);
181 }
595 182
596 DCHECK(open_flags_ & base::PLATFORM_FILE_WRITE); 183 void FileStream::Context::OnAsyncFileOpened() {
597 if (FlushFileBuffers(file_)) { 184 MessageLoopForIO::current()->RegisterIOHandler(file_, this);
598 return OK; 185 }
186
187 int64 FileStream::Context::SeekFileImpl(Whence whence, int64 offset) {
188 LARGE_INTEGER distance, res;
189 distance.QuadPart = offset;
190 DWORD move_method = static_cast<DWORD>(whence);
191 if (SetFilePointerEx(file_, distance, &res, move_method)) {
192 SetOffset(&io_context_.overlapped, res);
193 return res.QuadPart;
599 } 194 }
600 195
601 return RecordAndMapError(GetLastError(), 196 return -static_cast<int>(GetLastError());
602 FILE_ERROR_SOURCE_FLUSH,
603 record_uma_,
604 bound_net_log_);
605 } 197 }
606 198
607 int64 FileStreamWin::Truncate(int64 bytes) { 199 void FileStream::Context::IOCompletionIsPending(
608 base::ThreadRestrictions::AssertIOAllowed(); 200 const CompletionCallback& callback,
609 201 IOBuffer* buf) {
610 if (!IsOpen()) 202 DCHECK(callback_.is_null());
611 return ERR_UNEXPECTED; 203 callback_ = callback;
612 204 in_flight_buf_ = buf; // Hold until the async operation ends.
613 // We'd better be open for writing. 205 async_in_progress_ = true;
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 } 206 }
635 207
636 void FileStreamWin::EnableErrorStatistics() { 208 void FileStream::Context::OnIOCompleted(MessageLoopForIO::IOContext* context,
637 record_uma_ = true; 209 DWORD bytes_read,
210 DWORD error) {
211 DCHECK_EQ(&io_context_, context);
212 DCHECK(!callback_.is_null());
638 213
639 if (async_context_.get()) 214 if (orphaned_) {
640 async_context_->EnableErrorStatistics(); 215 callback_.Reset();
641 } 216 in_flight_buf_ = NULL;
642 217 CloseAsync(CompletionCallback());
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; 218 return;
649 } 219 }
650 220
651 // Should never connect to itself. 221 int result = static_cast<int>(bytes_read);
652 DCHECK_NE(bound_net_log_.source().id, owner_bound_net_log.source().id); 222 if (error && error != ERROR_HANDLE_EOF)
223 result = RecordAndMapError(error, error_source_);
653 224
654 bound_net_log_.AddEvent( 225 if (bytes_read)
655 net::NetLog::TYPE_FILE_STREAM_BOUND_TO_OWNER, 226 IncrementOffset(&io_context_.overlapped, bytes_read);
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 227
670 // Reset this before Run() as Run() may issue a new async operation. 228 // Reset this before Run() as Run() may issue a new async operation.
671 ResetOnIOComplete(); 229 async_in_progress_ = false;
672 callback.Run(OK); 230 CompletionCallback temp_callback = callback_;
673 } 231 callback_.Reset();
674 232 scoped_refptr<IOBuffer> temp_buf = in_flight_buf_;
675 void FileStreamWin::SeekFile(Whence whence, int64 offset, int64* result) { 233 in_flight_buf_ = NULL;
676 LARGE_INTEGER distance, res; 234 temp_callback.Run(result);
677 distance.QuadPart = offset;
678 DWORD move_method = static_cast<DWORD>(whence);
679 if (!SetFilePointerEx(file_, distance, &res, move_method)) {
680 DWORD error = GetLastError();
681 LOG(WARNING) << "SetFilePointerEx failed: " << error;
682 *result = RecordAndMapError(error,
683 FILE_ERROR_SOURCE_SEEK,
684 record_uma_,
685 bound_net_log_);
686 return;
687 }
688 if (async_context_.get()) {
689 async_context_->set_error_source(FILE_ERROR_SOURCE_SEEK);
690 SetOffset(async_context_->overlapped(), res);
691 }
692 *result = res.QuadPart;
693 }
694
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
704 // Reset this before Run() as Run() may issue a new async operation.
705 ResetOnIOComplete();
706 callback.Run(*result);
707 }
708
709 void FileStreamWin::OnSeeked(
710 const Int64CompletionCallback& callback,
711 int64* result) {
712 // Reset this before Run() as Run() may issue a new async operation.
713 ResetOnIOComplete();
714 callback.Run(*result);
715 }
716
717 void FileStreamWin::ResetOnIOComplete() {
718 on_io_complete_.reset();
719 weak_ptr_factory_.InvalidateWeakPtrs();
720 }
721
722 void FileStreamWin::WaitForIOCompletion() {
723 // http://crbug.com/115067
724 base::ThreadRestrictions::ScopedAllowWait allow_wait;
725 if (on_io_complete_.get()) {
726 on_io_complete_->Wait();
727 on_io_complete_.reset();
728 }
729 } 235 }
730 236
731 } // namespace net 237 } // namespace net
OLDNEW

Powered by Google App Engine
This is Rietveld 408576698