report_queue_impl.cc 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334
  1. // Copyright 2020 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 "components/reporting/client/report_queue_impl.h"
  5. #include <memory>
  6. #include <queue>
  7. #include <string>
  8. #include <utility>
  9. #include "base/bind.h"
  10. #include "base/callback.h"
  11. #include "base/check.h"
  12. #include "base/memory/ptr_util.h"
  13. #include "base/memory/scoped_refptr.h"
  14. #include "base/notreached.h"
  15. #include "base/sequence_checker.h"
  16. #include "base/task/bind_post_task.h"
  17. #include "base/task/task_traits.h"
  18. #include "base/task/thread_pool.h"
  19. #include "base/time/time.h"
  20. #include "components/reporting/client/report_queue_configuration.h"
  21. #include "components/reporting/proto/synced/record.pb.h"
  22. #include "components/reporting/proto/synced/record_constants.pb.h"
  23. #include "components/reporting/storage/storage_module_interface.h"
  24. #include "components/reporting/util/status.h"
  25. #include "components/reporting/util/statusor.h"
  26. namespace reporting {
  27. namespace {
  28. // Calls |record_producer|, checks the result and in case of success, forwards
  29. // it to the storage. In production code should be invoked asynchronously, on a
  30. // thread pool (no synchronization expected).
  31. void AddRecordToStorage(scoped_refptr<StorageModuleInterface> storage,
  32. Priority priority,
  33. std::string dm_token,
  34. Destination destination,
  35. ReportQueue::RecordProducer record_producer,
  36. StorageModuleInterface::EnqueueCallback callback) {
  37. // Generate record data.
  38. auto record_result = std::move(record_producer).Run();
  39. if (!record_result.ok()) {
  40. std::move(callback).Run(record_result.status());
  41. return;
  42. }
  43. // Augment data.
  44. Record record;
  45. *record.mutable_data() = std::move(record_result.ValueOrDie());
  46. record.set_destination(destination);
  47. // |record| with no DM token is assumed to be associated with device DM token
  48. if (!dm_token.empty()) {
  49. *record.mutable_dm_token() = std::move(dm_token);
  50. }
  51. // Calculate timestamp in microseconds - to match Spanner expectations.
  52. const int64_t time_since_epoch_us =
  53. base::Time::Now().ToJavaTime() * base::Time::kMicrosecondsPerMillisecond;
  54. record.set_timestamp_us(time_since_epoch_us);
  55. if (!record_result.ok()) {
  56. std::move(callback).Run(record_result.status());
  57. return;
  58. }
  59. // Add resulting Record to the storage.
  60. storage->AddRecord(priority, std::move(record), std::move(callback));
  61. }
  62. } // namespace
  63. void ReportQueueImpl::Create(
  64. std::unique_ptr<ReportQueueConfiguration> config,
  65. scoped_refptr<StorageModuleInterface> storage,
  66. base::OnceCallback<void(StatusOr<std::unique_ptr<ReportQueue>>)> cb) {
  67. std::move(cb).Run(base::WrapUnique<ReportQueueImpl>(
  68. new ReportQueueImpl(std::move(config), storage)));
  69. }
  70. ReportQueueImpl::ReportQueueImpl(
  71. std::unique_ptr<ReportQueueConfiguration> config,
  72. scoped_refptr<StorageModuleInterface> storage)
  73. : config_(std::move(config)), storage_(storage) {}
  74. ReportQueueImpl::~ReportQueueImpl() = default;
  75. void ReportQueueImpl::AddProducedRecord(RecordProducer record_producer,
  76. Priority priority,
  77. EnqueueCallback callback) const {
  78. const Status status = config_->CheckPolicy();
  79. if (!status.ok()) {
  80. std::move(callback).Run(status);
  81. return;
  82. }
  83. if (priority == Priority::UNDEFINED_PRIORITY) {
  84. std::move(callback).Run(
  85. Status(error::INVALID_ARGUMENT, "Priority must be defined"));
  86. return;
  87. }
  88. // Execute |record_producer| on arbitrary thread, analyze the result and send
  89. // it to the Storage, returning with the callback.
  90. base::ThreadPool::PostTask(
  91. FROM_HERE, {base::TaskPriority::BEST_EFFORT},
  92. base::BindOnce(&AddRecordToStorage, storage_, priority,
  93. config_->dm_token(), config_->destination(),
  94. std::move(record_producer), std::move(callback)));
  95. }
  96. void ReportQueueImpl::Flush(Priority priority, FlushCallback callback) {
  97. storage_->Flush(priority, std::move(callback));
  98. }
  99. base::OnceCallback<void(StatusOr<std::unique_ptr<ReportQueue>>)>
  100. ReportQueueImpl::PrepareToAttachActualQueue() const {
  101. NOTREACHED();
  102. return base::BindOnce(
  103. [](StatusOr<std::unique_ptr<ReportQueue>>) { NOTREACHED(); });
  104. }
  105. // Implementation of SpeculativeReportQueueImpl::PendingRecordProducer
  106. SpeculativeReportQueueImpl::PendingRecordProducer::PendingRecordProducer(
  107. RecordProducer producer,
  108. Priority priority)
  109. : record_producer(std::move(producer)), record_priority(priority) {}
  110. SpeculativeReportQueueImpl::PendingRecordProducer::PendingRecordProducer(
  111. PendingRecordProducer&& other)
  112. : record_producer(std::move(other.record_producer)),
  113. record_priority(other.record_priority) {}
  114. SpeculativeReportQueueImpl::PendingRecordProducer::~PendingRecordProducer() =
  115. default;
  116. SpeculativeReportQueueImpl::PendingRecordProducer&
  117. SpeculativeReportQueueImpl::PendingRecordProducer::operator=(
  118. PendingRecordProducer&& other) {
  119. record_producer = std::move(other.record_producer);
  120. record_priority = other.record_priority;
  121. return *this;
  122. }
  123. // static
  124. std::unique_ptr<SpeculativeReportQueueImpl, base::OnTaskRunnerDeleter>
  125. SpeculativeReportQueueImpl::Create() {
  126. scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner =
  127. base::ThreadPool::CreateSequencedTaskRunner(
  128. {base::TaskPriority::BEST_EFFORT, base::MayBlock()});
  129. return std::unique_ptr<SpeculativeReportQueueImpl, base::OnTaskRunnerDeleter>(
  130. new SpeculativeReportQueueImpl(sequenced_task_runner),
  131. base::OnTaskRunnerDeleter(sequenced_task_runner));
  132. }
  133. SpeculativeReportQueueImpl::SpeculativeReportQueueImpl(
  134. scoped_refptr<base::SequencedTaskRunner> sequenced_task_runner)
  135. : sequenced_task_runner_(sequenced_task_runner) {
  136. DETACH_FROM_SEQUENCE(sequence_checker_);
  137. }
  138. SpeculativeReportQueueImpl::~SpeculativeReportQueueImpl() {
  139. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  140. }
  141. void SpeculativeReportQueueImpl::Flush(Priority priority,
  142. FlushCallback callback) {
  143. sequenced_task_runner_->PostTask(
  144. FROM_HERE,
  145. base::BindOnce(
  146. [](Priority priority, FlushCallback callback,
  147. base::WeakPtr<SpeculativeReportQueueImpl> self) {
  148. if (!self) {
  149. std::move(callback).Run(
  150. Status(error::UNAVAILABLE, "Queue has been destructed"));
  151. return;
  152. }
  153. DCHECK_CALLED_ON_VALID_SEQUENCE(self->sequence_checker_);
  154. if (!self->status_or_report_queue_.has_value()) {
  155. std::move(callback).Run(Status(error::FAILED_PRECONDITION,
  156. "ReportQueue is not ready yet."));
  157. return;
  158. }
  159. if (!self->status_or_report_queue_->ok()) {
  160. std::move(callback).Run(self->status_or_report_queue_->status());
  161. return;
  162. }
  163. const std::unique_ptr<ReportQueue>& report_queue =
  164. self->status_or_report_queue_->ValueOrDie();
  165. report_queue->Flush(priority, std::move(callback));
  166. },
  167. priority, std::move(callback), weak_ptr_factory_.GetWeakPtr()));
  168. }
  169. void SpeculativeReportQueueImpl::AddProducedRecord(
  170. RecordProducer record_producer,
  171. Priority priority,
  172. EnqueueCallback callback) const {
  173. // Invoke producer on a thread pool, then enqueue record on sequenced task
  174. // runner.
  175. sequenced_task_runner_->PostTask(
  176. FROM_HERE,
  177. base::BindOnce(&SpeculativeReportQueueImpl::MaybeEnqueueRecordProducer,
  178. weak_ptr_factory_.GetWeakPtr(), priority,
  179. std::move(callback), std::move(record_producer)));
  180. }
  181. void SpeculativeReportQueueImpl::MaybeEnqueueRecordProducer(
  182. Priority priority,
  183. EnqueueCallback callback,
  184. RecordProducer record_producer) const {
  185. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  186. if (!status_or_report_queue_.has_value()) {
  187. // Queue is not ready yet, store the record in the memory queue.
  188. pending_record_producers_.emplace(std::move(record_producer), priority);
  189. std::move(callback).Run(Status::StatusOK());
  190. return;
  191. }
  192. if (!status_or_report_queue_->ok()) {
  193. // Queue creation failed.
  194. std::move(callback).Run(status_or_report_queue_->status());
  195. return;
  196. }
  197. // Queue is ready. If memory queue is empty, just forward the record.
  198. if (pending_record_producers_.empty()) {
  199. const std::unique_ptr<ReportQueue>& report_queue =
  200. status_or_report_queue_->ValueOrDie();
  201. report_queue->AddProducedRecord(std::move(record_producer), priority,
  202. std::move(callback));
  203. return;
  204. }
  205. // If memory queue is not empty, attach the new record at the
  206. // end and initiate enqueuing of everything from there.
  207. pending_record_producers_.emplace(std::move(record_producer), priority);
  208. EnqueuePendingRecordProducers(std::move(callback));
  209. }
  210. void SpeculativeReportQueueImpl::EnqueuePendingRecordProducers(
  211. EnqueueCallback callback) const {
  212. DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
  213. DCHECK(status_or_report_queue_.has_value());
  214. DCHECK(status_or_report_queue_->ok());
  215. if (pending_record_producers_.empty()) {
  216. std::move(callback).Run(Status::StatusOK());
  217. return;
  218. }
  219. const std::unique_ptr<ReportQueue>& report_queue =
  220. status_or_report_queue_->ValueOrDie();
  221. auto head = std::move(pending_record_producers_.front());
  222. pending_record_producers_.pop();
  223. if (pending_record_producers_.empty()) {
  224. // Last of the pending records.
  225. report_queue->AddProducedRecord(std::move(head.record_producer),
  226. head.record_priority, std::move(callback));
  227. return;
  228. }
  229. report_queue->AddProducedRecord(
  230. std::move(head.record_producer), head.record_priority,
  231. base::BindPostTask(
  232. sequenced_task_runner_,
  233. base::BindOnce(
  234. [](base::WeakPtr<const SpeculativeReportQueueImpl> self,
  235. EnqueueCallback callback, Status status) {
  236. if (!status.ok()) {
  237. std::move(callback).Run(status);
  238. return;
  239. }
  240. if (!self) {
  241. std::move(callback).Run(
  242. Status(error::UNAVAILABLE, "Queue has been destructed"));
  243. return;
  244. }
  245. self->sequenced_task_runner_->PostTask(
  246. FROM_HERE, base::BindOnce(&SpeculativeReportQueueImpl::
  247. EnqueuePendingRecordProducers,
  248. self, std::move(callback)));
  249. },
  250. weak_ptr_factory_.GetWeakPtr(), std::move(callback))));
  251. }
  252. base::OnceCallback<void(StatusOr<std::unique_ptr<ReportQueue>>)>
  253. SpeculativeReportQueueImpl::PrepareToAttachActualQueue() const {
  254. return base::BindPostTask(
  255. sequenced_task_runner_,
  256. base::BindOnce(
  257. [](base::WeakPtr<SpeculativeReportQueueImpl> speculative_queue,
  258. StatusOr<std::unique_ptr<ReportQueue>> actual_queue_result) {
  259. if (!speculative_queue) {
  260. return; // Speculative queue was destructed in a meantime.
  261. }
  262. // Set actual queue for the speculative queue to use
  263. // (asynchronously).
  264. speculative_queue->AttachActualQueue(
  265. std::move(std::move(actual_queue_result)));
  266. },
  267. weak_ptr_factory_.GetWeakPtr()));
  268. }
  269. void SpeculativeReportQueueImpl::AttachActualQueue(
  270. StatusOr<std::unique_ptr<ReportQueue>> status_or_actual_queue) {
  271. sequenced_task_runner_->PostTask(
  272. FROM_HERE,
  273. base::BindOnce(
  274. [](base::WeakPtr<SpeculativeReportQueueImpl> self,
  275. StatusOr<std::unique_ptr<ReportQueue>> status_or_actual_queue) {
  276. if (!self) {
  277. return;
  278. }
  279. DCHECK_CALLED_ON_VALID_SEQUENCE(self->sequence_checker_);
  280. if (self->status_or_report_queue_.has_value()) {
  281. // Already attached, do nothing.
  282. return;
  283. }
  284. self->status_or_report_queue_ = std::move(status_or_actual_queue);
  285. // TODO(b/239583016): remove the ok status check once the enqueue
  286. // callbacks are stored along with the records in a pending queue
  287. // instead of only the records, and run the callbacks with the
  288. // failure status if creation failed.
  289. if (self->status_or_report_queue_->ok() &&
  290. !self->pending_record_producers_.empty()) {
  291. self->EnqueuePendingRecordProducers(
  292. base::BindOnce([](Status enqueue_status) {
  293. if (!enqueue_status.ok()) {
  294. LOG(ERROR) << "Pending records failed to enqueue, status="
  295. << enqueue_status;
  296. }
  297. }));
  298. }
  299. },
  300. weak_ptr_factory_.GetWeakPtr(), std::move(status_or_actual_queue)));
  301. }
  302. } // namespace reporting