1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
|
// Copyright 2021 The Chromium Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#include "chrome/browser/safe_browsing/cloud_content_scanning/file_opening_job.h"
#include "base/command_line.h"
#include "base/functional/bind.h"
#include "base/strings/string_number_conversions.h"
#include "base/task/task_traits.h"
#include "components/safe_browsing/core/common/safebrowsing_switches.h"
namespace safe_browsing {
namespace {
constexpr size_t kDefaultMaxFileOpeningThreads = 5;
} // namespace
// static
size_t FileOpeningJob::GetMaxFileOpeningThreads() {
base::CommandLine* command_line = base::CommandLine::ForCurrentProcess();
if (command_line->HasSwitch(switches::kWpMaxFileOpeningThreads)) {
int parsed_max;
if (base::StringToInt(command_line->GetSwitchValueASCII(
switches::kWpMaxFileOpeningThreads),
&parsed_max) &&
parsed_max > 0) {
return parsed_max;
} else {
LOG(ERROR) << switches::kWpMaxFileOpeningThreads << " had invalid value";
}
}
return kDefaultMaxFileOpeningThreads;
}
FileOpeningJob::FileOpeningTask::FileOpeningTask() = default;
FileOpeningJob::FileOpeningTask::~FileOpeningTask() = default;
FileOpeningJob::FileOpeningJob(std::vector<FileOpeningTask> tasks)
: tasks_(std::move(tasks)), max_threads_(GetMaxFileOpeningThreads()) {
num_unopened_files_ = tasks_.size();
// The base::Unratained calls are safe because `file_opening_job_handle_` is
// destroyed when `this` is.
file_opening_job_handle_ =
base::PostJob(FROM_HERE,
{base::TaskPriority::USER_VISIBLE, base::MayBlock(),
base::ThreadPolicy::PREFER_BACKGROUND},
base::BindRepeating(&FileOpeningJob::ProcessNextTask,
base::Unretained(this)),
base::BindRepeating(&FileOpeningJob::MaxConcurrentThreads,
base::Unretained(this)));
}
FileOpeningJob::~FileOpeningJob() {
if (file_opening_job_handle_)
file_opening_job_handle_.Cancel();
}
void FileOpeningJob::ProcessNextTask(base::JobDelegate* job_delegate) {
// Loop over `tasks_` until one can safely be taken by this thread.
for (size_t i = 0; i < tasks_.size() && num_unopened_files() != 0 &&
!job_delegate->ShouldYield();
++i) {
// The task's `taken` value is atomic, so exchanging it to find it used to
// be true indicates we were the not the thread that took it.
// std::memory_order_relaxed is safe here since `taken` is not synchronized
// with other state.
if (tasks_[i].taken.exchange(true, std::memory_order_relaxed))
continue;
// Since we know we now have taken `tasks_[i]`, we can do the file opening
// work safely.
tasks_[i].request->OpenFile();
// Now that the file opening work is done, `num_unopened_files_` is
// decremented atomically and we return to free the thread.
num_unopened_files_.fetch_sub(1, std::memory_order_relaxed);
return;
}
}
size_t FileOpeningJob::num_unopened_files() {
return num_unopened_files_.load(std::memory_order_relaxed);
}
size_t FileOpeningJob::MaxConcurrentThreads(size_t /*worker_count*/) {
return std::min(num_unopened_files(), max_threads_);
}
} // namespace safe_browsing
|
