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