file_writer_delegate.cc 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328
  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 "storage/browser/file_system/file_writer_delegate.h"
  5. #include <stdint.h>
  6. #include <memory>
  7. #include <utility>
  8. #include "base/bind.h"
  9. #include "base/callback.h"
  10. #include "base/location.h"
  11. #include "base/task/sequenced_task_runner.h"
  12. #include "base/task/single_thread_task_runner.h"
  13. #include "base/threading/thread_restrictions.h"
  14. #include "base/threading/thread_task_runner_handle.h"
  15. #include "net/base/net_errors.h"
  16. #include "storage/browser/file_system/file_stream_writer.h"
  17. #include "storage/browser/file_system/file_system_context.h"
  18. #include "storage/common/file_system/file_system_mount_option.h"
  19. #include "storage/common/file_system/file_system_util.h"
  20. namespace storage {
  21. static const int kReadBufSize = 32768;
  22. FileWriterDelegate::FileWriterDelegate(
  23. std::unique_ptr<FileStreamWriter> file_stream_writer,
  24. FlushPolicy flush_policy)
  25. : file_stream_writer_(std::move(file_stream_writer)),
  26. writing_started_(false),
  27. flush_policy_(flush_policy),
  28. bytes_written_backlog_(0),
  29. bytes_written_(0),
  30. bytes_read_(0),
  31. io_buffer_(base::MakeRefCounted<net::IOBufferWithSize>(kReadBufSize)),
  32. data_pipe_watcher_(FROM_HERE, mojo::SimpleWatcher::ArmingPolicy::MANUAL) {
  33. }
  34. FileWriterDelegate::~FileWriterDelegate() = default;
  35. void FileWriterDelegate::Start(std::unique_ptr<BlobReader> blob_reader,
  36. DelegateWriteCallback write_callback) {
  37. write_callback_ = std::move(write_callback);
  38. if (!blob_reader) {
  39. OnReadError(base::File::FILE_ERROR_FAILED);
  40. return;
  41. }
  42. blob_reader_ = std::move(blob_reader);
  43. BlobReader::Status status = blob_reader_->CalculateSize(base::BindOnce(
  44. &FileWriterDelegate::OnDidCalculateSize, weak_factory_.GetWeakPtr()));
  45. switch (status) {
  46. case BlobReader::Status::NET_ERROR:
  47. OnDidCalculateSize(blob_reader_->net_error());
  48. return;
  49. case BlobReader::Status::DONE:
  50. OnDidCalculateSize(net::OK);
  51. return;
  52. case BlobReader::Status::IO_PENDING:
  53. // Do nothing.
  54. return;
  55. }
  56. NOTREACHED();
  57. }
  58. void FileWriterDelegate::Start(mojo::ScopedDataPipeConsumerHandle data_pipe,
  59. DelegateWriteCallback write_callback) {
  60. write_callback_ = std::move(write_callback);
  61. if (!data_pipe) {
  62. OnReadError(base::File::FILE_ERROR_FAILED);
  63. return;
  64. }
  65. data_pipe_ = std::move(data_pipe);
  66. data_pipe_watcher_.Watch(
  67. data_pipe_.get(),
  68. MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
  69. MOJO_TRIGGER_CONDITION_SIGNALS_SATISFIED,
  70. base::BindRepeating(&FileWriterDelegate::OnDataPipeReady,
  71. weak_factory_.GetWeakPtr()));
  72. data_pipe_watcher_.ArmOrNotify();
  73. }
  74. void FileWriterDelegate::Cancel() {
  75. // Destroy the reader and invalidate weak ptrs to prevent pending callbacks.
  76. blob_reader_ = nullptr;
  77. data_pipe_watcher_.Cancel();
  78. data_pipe_.reset();
  79. weak_factory_.InvalidateWeakPtrs();
  80. const int status = file_stream_writer_->Cancel(base::BindOnce(
  81. &FileWriterDelegate::OnWriteCancelled, weak_factory_.GetWeakPtr()));
  82. // Return true to finish immediately if we have no pending writes.
  83. // Otherwise we'll do the final cleanup in the Cancel callback.
  84. if (status != net::ERR_IO_PENDING) {
  85. write_callback_.Run(base::File::FILE_ERROR_ABORT, 0,
  86. GetCompletionStatusOnError());
  87. }
  88. }
  89. void FileWriterDelegate::OnDidCalculateSize(int net_error) {
  90. DCHECK_NE(net::ERR_IO_PENDING, net_error);
  91. if (net_error != net::OK) {
  92. OnReadError(NetErrorToFileError(net_error));
  93. return;
  94. }
  95. Read();
  96. }
  97. void FileWriterDelegate::OnReadCompleted(int bytes_read) {
  98. DCHECK_NE(net::ERR_IO_PENDING, bytes_read);
  99. if (bytes_read < 0) {
  100. OnReadError(NetErrorToFileError(bytes_read));
  101. return;
  102. }
  103. OnDataReceived(bytes_read);
  104. }
  105. void FileWriterDelegate::Read() {
  106. bytes_written_ = 0;
  107. if (blob_reader_) {
  108. BlobReader::Status status =
  109. blob_reader_->Read(io_buffer_.get(), io_buffer_->size(), &bytes_read_,
  110. base::BindOnce(&FileWriterDelegate::OnReadCompleted,
  111. weak_factory_.GetWeakPtr()));
  112. switch (status) {
  113. case BlobReader::Status::NET_ERROR:
  114. OnReadCompleted(blob_reader_->net_error());
  115. return;
  116. case BlobReader::Status::DONE:
  117. base::ThreadTaskRunnerHandle::Get()->PostTask(
  118. FROM_HERE, base::BindOnce(&FileWriterDelegate::OnReadCompleted,
  119. weak_factory_.GetWeakPtr(), bytes_read_));
  120. return;
  121. case BlobReader::Status::IO_PENDING:
  122. // Do nothing.
  123. return;
  124. }
  125. NOTREACHED();
  126. return;
  127. }
  128. DCHECK(data_pipe_);
  129. uint32_t num_bytes = io_buffer_->size();
  130. MojoResult result = data_pipe_->ReadData(io_buffer_->data(), &num_bytes,
  131. MOJO_READ_DATA_FLAG_NONE);
  132. if (result == MOJO_RESULT_SHOULD_WAIT) {
  133. data_pipe_watcher_.ArmOrNotify();
  134. return;
  135. }
  136. if (result == MOJO_RESULT_OK) {
  137. bytes_read_ = num_bytes;
  138. OnReadCompleted(bytes_read_);
  139. return;
  140. }
  141. if (result == MOJO_RESULT_FAILED_PRECONDITION) {
  142. // Pipe closed, done reading.
  143. OnReadCompleted(0);
  144. return;
  145. }
  146. // Some unknown error, this shouldn't happen.
  147. NOTREACHED();
  148. OnReadError(base::File::FILE_ERROR_FAILED);
  149. }
  150. void FileWriterDelegate::OnDataReceived(int bytes_read) {
  151. bytes_read_ = bytes_read;
  152. if (bytes_read == 0) { // We're done.
  153. OnProgress(0, true);
  154. } else {
  155. // This could easily be optimized to rotate between a pool of buffers, so
  156. // that we could read and write at the same time. It's not yet clear that
  157. // it's necessary.
  158. cursor_ =
  159. base::MakeRefCounted<net::DrainableIOBuffer>(io_buffer_, bytes_read_);
  160. Write();
  161. }
  162. }
  163. void FileWriterDelegate::Write() {
  164. writing_started_ = true;
  165. int64_t bytes_to_write = bytes_read_ - bytes_written_;
  166. int write_response = file_stream_writer_->Write(
  167. cursor_.get(), static_cast<int>(bytes_to_write),
  168. base::BindOnce(&FileWriterDelegate::OnDataWritten,
  169. weak_factory_.GetWeakPtr()));
  170. if (write_response > 0) {
  171. base::ThreadTaskRunnerHandle::Get()->PostTask(
  172. FROM_HERE, base::BindOnce(&FileWriterDelegate::OnDataWritten,
  173. weak_factory_.GetWeakPtr(), write_response));
  174. } else if (net::ERR_IO_PENDING != write_response) {
  175. OnWriteError(NetErrorToFileError(write_response));
  176. } else {
  177. async_write_in_progress_ = true;
  178. }
  179. }
  180. void FileWriterDelegate::OnDataWritten(int write_response) {
  181. async_write_in_progress_ = false;
  182. if (saved_read_error_ != base::File::FILE_OK) {
  183. OnReadError(saved_read_error_);
  184. return;
  185. }
  186. if (write_response > 0) {
  187. OnProgress(write_response, false);
  188. cursor_->DidConsume(write_response);
  189. bytes_written_ += write_response;
  190. if (bytes_written_ == bytes_read_)
  191. Read();
  192. else
  193. Write();
  194. } else {
  195. OnWriteError(NetErrorToFileError(write_response));
  196. }
  197. }
  198. FileWriterDelegate::WriteProgressStatus
  199. FileWriterDelegate::GetCompletionStatusOnError() const {
  200. return writing_started_ ? ERROR_WRITE_STARTED : ERROR_WRITE_NOT_STARTED;
  201. }
  202. void FileWriterDelegate::OnReadError(base::File::Error error) {
  203. if (async_write_in_progress_) {
  204. // Error signaled by the URLRequest while writing. This will be processed
  205. // when the write completes.
  206. saved_read_error_ = error;
  207. return;
  208. }
  209. // Destroy the reader and invalidate weak ptrs to prevent pending callbacks.
  210. blob_reader_.reset();
  211. data_pipe_watcher_.Cancel();
  212. data_pipe_.reset();
  213. weak_factory_.InvalidateWeakPtrs();
  214. if (writing_started_)
  215. MaybeFlushForCompletion(error, 0, ERROR_WRITE_STARTED);
  216. else
  217. write_callback_.Run(error, 0, ERROR_WRITE_NOT_STARTED);
  218. }
  219. void FileWriterDelegate::OnWriteError(base::File::Error error) {
  220. // Destroy the reader and invalidate weak ptrs to prevent pending callbacks.
  221. blob_reader_.reset();
  222. data_pipe_watcher_.Cancel();
  223. data_pipe_.reset();
  224. weak_factory_.InvalidateWeakPtrs();
  225. // Errors when writing are not recoverable, so don't bother flushing.
  226. write_callback_.Run(
  227. error, 0,
  228. writing_started_ ? ERROR_WRITE_STARTED : ERROR_WRITE_NOT_STARTED);
  229. }
  230. void FileWriterDelegate::OnProgress(int bytes_written, bool done) {
  231. DCHECK(bytes_written + bytes_written_backlog_ >= bytes_written_backlog_);
  232. static const int kMinProgressDelayMS = 200;
  233. base::Time currentTime = base::Time::Now();
  234. if (done || last_progress_event_time_.is_null() ||
  235. (currentTime - last_progress_event_time_).InMilliseconds() >
  236. kMinProgressDelayMS) {
  237. bytes_written += bytes_written_backlog_;
  238. last_progress_event_time_ = currentTime;
  239. bytes_written_backlog_ = 0;
  240. if (done) {
  241. MaybeFlushForCompletion(base::File::FILE_OK, bytes_written,
  242. SUCCESS_COMPLETED);
  243. } else {
  244. write_callback_.Run(base::File::FILE_OK, bytes_written,
  245. SUCCESS_IO_PENDING);
  246. }
  247. return;
  248. }
  249. bytes_written_backlog_ += bytes_written;
  250. }
  251. void FileWriterDelegate::OnWriteCancelled(int status) {
  252. write_callback_.Run(base::File::FILE_ERROR_ABORT, 0,
  253. GetCompletionStatusOnError());
  254. }
  255. void FileWriterDelegate::MaybeFlushForCompletion(
  256. base::File::Error error,
  257. int bytes_written,
  258. WriteProgressStatus progress_status) {
  259. if (flush_policy_ == FlushPolicy::NO_FLUSH_ON_COMPLETION) {
  260. write_callback_.Run(error, bytes_written, progress_status);
  261. return;
  262. }
  263. // DCHECK_EQ on enum classes is not supported.
  264. DCHECK(flush_policy_ == FlushPolicy::FLUSH_ON_COMPLETION);
  265. int flush_error = file_stream_writer_->Flush(
  266. base::BindOnce(&FileWriterDelegate::OnFlushed, weak_factory_.GetWeakPtr(),
  267. error, bytes_written, progress_status));
  268. if (flush_error != net::ERR_IO_PENDING)
  269. OnFlushed(error, bytes_written, progress_status, flush_error);
  270. }
  271. void FileWriterDelegate::OnFlushed(base::File::Error error,
  272. int bytes_written,
  273. WriteProgressStatus progress_status,
  274. int flush_error) {
  275. if (error == base::File::FILE_OK && flush_error != net::OK) {
  276. // If the Flush introduced an error, overwrite the status.
  277. // Otherwise, keep the original error status.
  278. error = NetErrorToFileError(flush_error);
  279. progress_status = GetCompletionStatusOnError();
  280. }
  281. write_callback_.Run(error, bytes_written, progress_status);
  282. }
  283. void FileWriterDelegate::OnDataPipeReady(
  284. MojoResult result,
  285. const mojo::HandleSignalsState& state) {
  286. Read();
  287. }
  288. } // namespace storage