file_stream_context_win.cc 6.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230
  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
  3. // found in the LICENSE file.
  4. #include "net/base/file_stream_context.h"
  5. #include <windows.h>
  6. #include <utility>
  7. #include "base/bind.h"
  8. #include "base/files/file_path.h"
  9. #include "base/location.h"
  10. #include "base/logging.h"
  11. #include "base/message_loop/message_pump_for_io.h"
  12. #include "base/task/current_thread.h"
  13. #include "base/task/single_thread_task_runner.h"
  14. #include "base/task/task_runner.h"
  15. #include "base/threading/thread_task_runner_handle.h"
  16. #include "net/base/io_buffer.h"
  17. #include "net/base/net_errors.h"
  18. namespace net {
  19. namespace {
  20. void SetOffset(OVERLAPPED* overlapped, const LARGE_INTEGER& offset) {
  21. overlapped->Offset = offset.LowPart;
  22. overlapped->OffsetHigh = offset.HighPart;
  23. }
  24. void IncrementOffset(OVERLAPPED* overlapped, DWORD count) {
  25. LARGE_INTEGER offset;
  26. offset.LowPart = overlapped->Offset;
  27. offset.HighPart = overlapped->OffsetHigh;
  28. offset.QuadPart += static_cast<LONGLONG>(count);
  29. SetOffset(overlapped, offset);
  30. }
  31. } // namespace
  32. FileStream::Context::Context(scoped_refptr<base::TaskRunner> task_runner)
  33. : Context(base::File(), std::move(task_runner)) {}
  34. FileStream::Context::Context(base::File file,
  35. scoped_refptr<base::TaskRunner> task_runner)
  36. : base::MessagePumpForIO::IOHandler(FROM_HERE),
  37. file_(std::move(file)),
  38. task_runner_(std::move(task_runner)) {
  39. if (file_.IsValid()) {
  40. DCHECK(file_.async());
  41. OnFileOpened();
  42. }
  43. }
  44. FileStream::Context::~Context() = default;
  45. int FileStream::Context::Read(IOBuffer* buf,
  46. int buf_len,
  47. CompletionOnceCallback callback) {
  48. DCHECK(!async_in_progress_);
  49. DCHECK(!async_read_initiated_);
  50. DCHECK(!async_read_completed_);
  51. DCHECK(!io_complete_for_read_received_);
  52. IOCompletionIsPending(std::move(callback), buf);
  53. async_read_initiated_ = true;
  54. result_ = 0;
  55. task_runner_->PostTask(
  56. FROM_HERE,
  57. base::BindOnce(&FileStream::Context::ReadAsync, base::Unretained(this),
  58. file_.GetPlatformFile(), base::WrapRefCounted(buf),
  59. buf_len, &io_context_.overlapped,
  60. base::ThreadTaskRunnerHandle::Get()));
  61. return ERR_IO_PENDING;
  62. }
  63. int FileStream::Context::Write(IOBuffer* buf,
  64. int buf_len,
  65. CompletionOnceCallback callback) {
  66. DCHECK(!async_in_progress_);
  67. result_ = 0;
  68. DWORD bytes_written = 0;
  69. if (!WriteFile(file_.GetPlatformFile(), buf->data(), buf_len,
  70. &bytes_written, &io_context_.overlapped)) {
  71. IOResult error = IOResult::FromOSError(GetLastError());
  72. if (error.os_error == ERROR_IO_PENDING) {
  73. IOCompletionIsPending(std::move(callback), buf);
  74. } else {
  75. LOG(WARNING) << "WriteFile failed: " << error.os_error;
  76. }
  77. return static_cast<int>(error.result);
  78. }
  79. IOCompletionIsPending(std::move(callback), buf);
  80. return ERR_IO_PENDING;
  81. }
  82. FileStream::Context::IOResult FileStream::Context::SeekFileImpl(
  83. int64_t offset) {
  84. LARGE_INTEGER result;
  85. result.QuadPart = offset;
  86. SetOffset(&io_context_.overlapped, result);
  87. return IOResult(result.QuadPart, 0);
  88. }
  89. void FileStream::Context::OnFileOpened() {
  90. HRESULT hr = base::CurrentIOThread::Get()->RegisterIOHandler(
  91. file_.GetPlatformFile(), this);
  92. if (!SUCCEEDED(hr))
  93. file_.Close();
  94. }
  95. void FileStream::Context::IOCompletionIsPending(CompletionOnceCallback callback,
  96. IOBuffer* buf) {
  97. DCHECK(callback_.is_null());
  98. callback_ = std::move(callback);
  99. in_flight_buf_ = buf; // Hold until the async operation ends.
  100. async_in_progress_ = true;
  101. }
  102. void FileStream::Context::OnIOCompleted(
  103. base::MessagePumpForIO::IOContext* context,
  104. DWORD bytes_read,
  105. DWORD error) {
  106. DCHECK_EQ(&io_context_, context);
  107. DCHECK(!callback_.is_null());
  108. DCHECK(async_in_progress_);
  109. if (!async_read_initiated_)
  110. async_in_progress_ = false;
  111. if (orphaned_) {
  112. io_complete_for_read_received_ = true;
  113. // If we are called due to a pending read and the asynchronous read task
  114. // has not completed we have to keep the context around until it completes.
  115. if (async_read_initiated_ && !async_read_completed_)
  116. return;
  117. DeleteOrphanedContext();
  118. return;
  119. }
  120. if (error == ERROR_HANDLE_EOF) {
  121. result_ = 0;
  122. } else if (error) {
  123. IOResult error_result = IOResult::FromOSError(error);
  124. result_ = static_cast<int>(error_result.result);
  125. } else {
  126. if (result_)
  127. DCHECK_EQ(result_, static_cast<int>(bytes_read));
  128. result_ = bytes_read;
  129. IncrementOffset(&io_context_.overlapped, bytes_read);
  130. }
  131. if (async_read_initiated_)
  132. io_complete_for_read_received_ = true;
  133. InvokeUserCallback();
  134. }
  135. void FileStream::Context::InvokeUserCallback() {
  136. // For an asynchonous Read operation don't invoke the user callback until
  137. // we receive the IO completion notification and the asynchronous Read
  138. // completion notification.
  139. if (async_read_initiated_) {
  140. if (!io_complete_for_read_received_ || !async_read_completed_)
  141. return;
  142. async_read_initiated_ = false;
  143. io_complete_for_read_received_ = false;
  144. async_read_completed_ = false;
  145. async_in_progress_ = false;
  146. }
  147. scoped_refptr<IOBuffer> temp_buf = in_flight_buf_;
  148. in_flight_buf_ = nullptr;
  149. std::move(callback_).Run(result_);
  150. }
  151. void FileStream::Context::DeleteOrphanedContext() {
  152. async_in_progress_ = false;
  153. callback_.Reset();
  154. in_flight_buf_ = nullptr;
  155. CloseAndDelete();
  156. }
  157. // static
  158. void FileStream::Context::ReadAsync(
  159. FileStream::Context* context,
  160. HANDLE file,
  161. scoped_refptr<IOBuffer> buf,
  162. int buf_len,
  163. OVERLAPPED* overlapped,
  164. scoped_refptr<base::SingleThreadTaskRunner> origin_thread_task_runner) {
  165. DWORD bytes_read = 0;
  166. BOOL ret = ::ReadFile(file, buf->data(), buf_len, &bytes_read, overlapped);
  167. origin_thread_task_runner->PostTask(
  168. FROM_HERE, base::BindOnce(&FileStream::Context::ReadAsyncResult,
  169. base::Unretained(context), ret, bytes_read,
  170. ::GetLastError()));
  171. }
  172. void FileStream::Context::ReadAsyncResult(BOOL read_file_ret,
  173. DWORD bytes_read,
  174. DWORD os_error) {
  175. // If the context is orphaned and we already received the io completion
  176. // notification then we should delete the context and get out.
  177. if (orphaned_ && io_complete_for_read_received_) {
  178. DeleteOrphanedContext();
  179. return;
  180. }
  181. async_read_completed_ = true;
  182. if (read_file_ret) {
  183. result_ = bytes_read;
  184. InvokeUserCallback();
  185. return;
  186. }
  187. IOResult error = IOResult::FromOSError(os_error);
  188. if (error.os_error == ERROR_IO_PENDING) {
  189. InvokeUserCallback();
  190. } else {
  191. OnIOCompleted(&io_context_, 0, error.os_error);
  192. }
  193. }
  194. } // namespace net