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