blob: f207e0d325ed2737c415c20b30cd4afd9dab63c8 [file] [log] [blame]
// Copyright 2017 The Chromium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#include "mojo/public/cpp/system/file_data_pipe_producer.h"
#include <algorithm>
#include <limits>
#include <memory>
#include <utility>
#include "base/bind.h"
#include "base/callback.h"
#include "base/location.h"
#include "base/memory/ref_counted_delete_on_sequence.h"
#include "base/numerics/safe_conversions.h"
#include "base/sequenced_task_runner.h"
#include "base/synchronization/lock.h"
#include "base/task/post_task.h"
#include "base/threading/sequenced_task_runner_handle.h"
#include "mojo/public/cpp/system/simple_watcher.h"
namespace mojo {
namespace {
// No good reason not to attempt very large pipe transactions in case the data
// pipe in use has a very large capacity available, so we default to trying
// 64 MB chunks whenever a producer is writable.
constexpr uint32_t kDefaultMaxReadSize = 64 * 1024 * 1024;
MojoResult FileErrorToMojoResult(base::File::Error error) {
switch (error) {
case base::File::FILE_OK:
return MOJO_RESULT_OK;
case base::File::FILE_ERROR_NOT_FOUND:
return MOJO_RESULT_NOT_FOUND;
case base::File::FILE_ERROR_SECURITY:
case base::File::FILE_ERROR_ACCESS_DENIED:
return MOJO_RESULT_PERMISSION_DENIED;
case base::File::FILE_ERROR_TOO_MANY_OPENED:
case base::File::FILE_ERROR_NO_MEMORY:
return MOJO_RESULT_RESOURCE_EXHAUSTED;
case base::File::FILE_ERROR_ABORT:
return MOJO_RESULT_ABORTED;
default:
return MOJO_RESULT_UNKNOWN;
}
}
} // namespace
class FileDataPipeProducer::FileSequenceState
: public base::RefCountedDeleteOnSequence<FileSequenceState> {
public:
using CompletionCallback =
base::OnceCallback<void(ScopedDataPipeProducerHandle producer,
MojoResult result)>;
FileSequenceState(
ScopedDataPipeProducerHandle producer_handle,
scoped_refptr<base::SequencedTaskRunner> file_task_runner,
CompletionCallback callback,
scoped_refptr<base::SequencedTaskRunner> callback_task_runner,
std::unique_ptr<Observer> observer)
: base::RefCountedDeleteOnSequence<FileSequenceState>(
std::move(file_task_runner)),
callback_task_runner_(std::move(callback_task_runner)),
producer_handle_(std::move(producer_handle)),
callback_(std::move(callback)),
observer_(std::move(observer)) {}
void Cancel() {
base::AutoLock lock(lock_);
is_cancelled_ = true;
}
void StartFromFile(base::File file, size_t max_bytes) {
owning_task_runner()->PostTask(
FROM_HERE,
base::BindOnce(&FileSequenceState::StartFromFileOnFileSequence, this,
std::move(file), max_bytes));
}
void StartFromPath(const base::FilePath& path) {
owning_task_runner()->PostTask(
FROM_HERE,
base::BindOnce(&FileSequenceState::StartFromPathOnFileSequence, this,
path));
}
private:
friend class base::DeleteHelper<FileSequenceState>;
friend class base::RefCountedDeleteOnSequence<FileSequenceState>;
~FileSequenceState() = default;
void StartFromFileOnFileSequence(base::File file, size_t max_bytes) {
if (file.error_details() != base::File::FILE_OK) {
Finish(FileErrorToMojoResult(file.error_details()));
return;
}
file_ = std::move(file);
max_bytes_ = max_bytes;
TransferSomeBytes();
if (producer_handle_.is_valid()) {
// If we didn't nail it all on the first transaction attempt, setup a
// watcher and complete the read asynchronously.
watcher_ = std::make_unique<SimpleWatcher>(
FROM_HERE, SimpleWatcher::ArmingPolicy::AUTOMATIC,
base::SequencedTaskRunnerHandle::Get());
watcher_->Watch(producer_handle_.get(), MOJO_HANDLE_SIGNAL_WRITABLE,
MOJO_WATCH_CONDITION_SATISFIED,
base::Bind(&FileSequenceState::OnHandleReady, this));
}
}
void StartFromPathOnFileSequence(const base::FilePath& path) {
StartFromFileOnFileSequence(
base::File(path, base::File::FLAG_OPEN | base::File::FLAG_READ),
std::numeric_limits<size_t>::max());
}
void OnHandleReady(MojoResult result, const HandleSignalsState& state) {
{
// Stop ourselves from doing redundant work if we've been cancelled from
// another thread. Note that we do not rely on this for any kind of thread
// safety concerns.
base::AutoLock lock(lock_);
if (is_cancelled_)
return;
}
if (result != MOJO_RESULT_OK) {
// Either the consumer pipe has been closed or something terrible
// happened. In any case, we'll never be able to write more data.
Finish(result);
return;
}
TransferSomeBytes();
}
void TransferSomeBytes() {
while (true) {
// Lock as much of the pipe as we can.
void* pipe_buffer;
uint32_t size = kDefaultMaxReadSize;
MojoResult result = producer_handle_->BeginWriteData(
&pipe_buffer, &size, MOJO_WRITE_DATA_FLAG_NONE);
if (result == MOJO_RESULT_SHOULD_WAIT)
return;
if (result != MOJO_RESULT_OK) {
Finish(result);
return;
}
// Attempt to read that many bytes from the file, directly into the data
// pipe. Note that while |max_bytes_remaining| may be very large, the
// length we attempt read is bounded by the much smaller
// |kDefaultMaxReadSize| via |size|.
DCHECK(base::IsValueInRangeForNumericType<int>(size));
const size_t max_bytes_remaining = max_bytes_ - bytes_transferred_;
int attempted_read_size = static_cast<int>(
std::min(static_cast<size_t>(size), max_bytes_remaining));
int read_size = file_.ReadAtCurrentPos(static_cast<char*>(pipe_buffer),
attempted_read_size);
base::File::Error read_error;
if (read_size < 0) {
read_error = base::File::GetLastFileError();
DCHECK_NE(base::File::FILE_OK, read_error);
if (observer_)
observer_->OnBytesRead(pipe_buffer, 0u, read_error);
} else {
read_error = base::File::FILE_OK;
if (observer_) {
observer_->OnBytesRead(pipe_buffer, static_cast<size_t>(read_size),
base::File::FILE_OK);
}
}
producer_handle_->EndWriteData(
read_size >= 0 ? static_cast<uint32_t>(read_size) : 0);
if (read_size < 0) {
Finish(FileErrorToMojoResult(read_error));
return;
}
bytes_transferred_ += read_size;
DCHECK_LE(bytes_transferred_, max_bytes_);
if (read_size < attempted_read_size) {
// ReadAtCurrentPos makes a best effort to read all requested bytes. We
// reasonably assume if it fails to read what we ask for, we've hit EOF.
Finish(MOJO_RESULT_OK);
return;
}
if (bytes_transferred_ == max_bytes_) {
// We've read as much as we were asked to read.
Finish(MOJO_RESULT_OK);
return;
}
}
}
void Finish(MojoResult result) {
if (observer_) {
observer_->OnDoneReading();
observer_ = nullptr;
}
watcher_.reset();
callback_task_runner_->PostTask(
FROM_HERE, base::BindOnce(std::move(callback_),
std::move(producer_handle_), result));
}
const scoped_refptr<base::SequencedTaskRunner> callback_task_runner_;
// State which is effectively owned and used only on the file sequence.
ScopedDataPipeProducerHandle producer_handle_;
base::File file_;
size_t max_bytes_ = 0;
size_t bytes_transferred_ = 0;
CompletionCallback callback_;
std::unique_ptr<SimpleWatcher> watcher_;
// Guards |is_cancelled_|.
base::Lock lock_;
bool is_cancelled_ = false;
std::unique_ptr<Observer> observer_;
DISALLOW_COPY_AND_ASSIGN(FileSequenceState);
};
FileDataPipeProducer::FileDataPipeProducer(
ScopedDataPipeProducerHandle producer,
std::unique_ptr<Observer> observer)
: producer_(std::move(producer)),
observer_(std::move(observer)),
weak_factory_(this) {}
FileDataPipeProducer::~FileDataPipeProducer() {
if (file_sequence_state_)
file_sequence_state_->Cancel();
}
void FileDataPipeProducer::WriteFromFile(base::File file,
CompletionCallback callback) {
WriteFromFile(std::move(file), std::numeric_limits<size_t>::max(),
std::move(callback));
}
void FileDataPipeProducer::WriteFromFile(base::File file,
size_t max_bytes,
CompletionCallback callback) {
InitializeNewRequest(std::move(callback));
file_sequence_state_->StartFromFile(std::move(file), max_bytes);
}
void FileDataPipeProducer::WriteFromPath(const base::FilePath& path,
CompletionCallback callback) {
InitializeNewRequest(std::move(callback));
file_sequence_state_->StartFromPath(path);
}
void FileDataPipeProducer::InitializeNewRequest(CompletionCallback callback) {
DCHECK(!file_sequence_state_);
// TODO(crbug.com/924416): Re-evaluate how TaskPriority is set here and in
// other file URL-loading-related code. Some callers require USER_VISIBLE
// (i.e., BEST_EFFORT is not enough).
auto file_task_runner = base::CreateSequencedTaskRunnerWithTraits(
{base::MayBlock(), base::TaskPriority::USER_VISIBLE});
file_sequence_state_ = new FileSequenceState(
std::move(producer_), file_task_runner,
base::BindOnce(&FileDataPipeProducer::OnWriteComplete,
weak_factory_.GetWeakPtr(), std::move(callback)),
base::SequencedTaskRunnerHandle::Get(), std::move(observer_));
}
void FileDataPipeProducer::OnWriteComplete(
CompletionCallback callback,
ScopedDataPipeProducerHandle producer,
MojoResult ready_result) {
producer_ = std::move(producer);
file_sequence_state_ = nullptr;
std::move(callback).Run(ready_result);
}
} // namespace mojo