FSTMemoryPersistence.mm 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433
  1. /*
  2. * Copyright 2017 Google
  3. *
  4. * Licensed under the Apache License, Version 2.0 (the "License");
  5. * you may not use this file except in compliance with the License.
  6. * You may obtain a copy of the License at
  7. *
  8. * http://www.apache.org/licenses/LICENSE-2.0
  9. *
  10. * Unless required by applicable law or agreed to in writing, software
  11. * distributed under the License is distributed on an "AS IS" BASIS,
  12. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. * See the License for the specific language governing permissions and
  14. * limitations under the License.
  15. */
  16. #import "Firestore/Source/Local/FSTMemoryPersistence.h"
  17. #include <memory>
  18. #include <unordered_map>
  19. #include <unordered_set>
  20. #include <vector>
  21. #include "Firestore/core/src/firebase/firestore/auth/user.h"
  22. #include "Firestore/core/src/firebase/firestore/local/index_manager.h"
  23. #include "Firestore/core/src/firebase/firestore/local/listen_sequence.h"
  24. #include "Firestore/core/src/firebase/firestore/local/lru_garbage_collector.h"
  25. #include "Firestore/core/src/firebase/firestore/local/memory_index_manager.h"
  26. #include "Firestore/core/src/firebase/firestore/local/memory_mutation_queue.h"
  27. #include "Firestore/core/src/firebase/firestore/local/memory_query_cache.h"
  28. #include "Firestore/core/src/firebase/firestore/local/memory_remote_document_cache.h"
  29. #include "Firestore/core/src/firebase/firestore/local/proto_sizer.h"
  30. #include "Firestore/core/src/firebase/firestore/local/reference_set.h"
  31. #include "Firestore/core/src/firebase/firestore/model/document_key.h"
  32. #include "Firestore/core/src/firebase/firestore/util/hard_assert.h"
  33. #include "absl/memory/memory.h"
  34. using firebase::firestore::auth::HashUser;
  35. using firebase::firestore::auth::User;
  36. using firebase::firestore::local::ListenSequence;
  37. using firebase::firestore::local::LruGarbageCollector;
  38. using firebase::firestore::local::LruParams;
  39. using firebase::firestore::local::MemoryIndexManager;
  40. using firebase::firestore::local::MemoryMutationQueue;
  41. using firebase::firestore::local::MemoryQueryCache;
  42. using firebase::firestore::local::MemoryRemoteDocumentCache;
  43. using firebase::firestore::local::ProtoSizer;
  44. using firebase::firestore::local::QueryData;
  45. using firebase::firestore::local::ReferenceSet;
  46. using firebase::firestore::local::TargetCallback;
  47. using firebase::firestore::model::DocumentKey;
  48. using firebase::firestore::model::DocumentKeyHash;
  49. using firebase::firestore::model::ListenSequenceNumber;
  50. using firebase::firestore::model::TargetId;
  51. using firebase::firestore::util::Status;
  52. using MutationQueues = std::unordered_map<User, std::unique_ptr<MemoryMutationQueue>, HashUser>;
  53. NS_ASSUME_NONNULL_BEGIN
  54. @interface FSTMemoryPersistence ()
  55. - (MemoryQueryCache *)queryCache;
  56. - (MemoryRemoteDocumentCache *)remoteDocumentCache;
  57. - (MemoryIndexManager *)indexManager;
  58. - (MemoryMutationQueue *)mutationQueueForUser:(const User &)user;
  59. @property(nonatomic, readonly) MutationQueues &mutationQueues;
  60. @property(nonatomic, assign, getter=isStarted) BOOL started;
  61. // Make this property writable so we can wire up a delegate.
  62. @property(nonatomic, strong) id<FSTReferenceDelegate> referenceDelegate;
  63. @end
  64. @implementation FSTMemoryPersistence {
  65. /**
  66. * The QueryCache representing the persisted cache of queries.
  67. *
  68. * Note that this is retained here to make it easier to write tests affecting both the in-memory
  69. * and LevelDB-backed persistence layers. Tests can create a new FSTLocalStore wrapping this
  70. * FSTPersistence instance and this will make the in-memory persistence layer behave as if it
  71. * were actually persisting values.
  72. */
  73. std::unique_ptr<MemoryQueryCache> _queryCache;
  74. /** The RemoteDocumentCache representing the persisted cache of remote documents. */
  75. std::unique_ptr<MemoryRemoteDocumentCache> _remoteDocumentCache;
  76. MemoryIndexManager _indexManager;
  77. FSTTransactionRunner _transactionRunner;
  78. id<FSTReferenceDelegate> _referenceDelegate;
  79. }
  80. + (instancetype)persistenceWithEagerGC {
  81. FSTMemoryPersistence *persistence = [[FSTMemoryPersistence alloc] init];
  82. persistence.referenceDelegate =
  83. [[FSTMemoryEagerReferenceDelegate alloc] initWithPersistence:persistence];
  84. return persistence;
  85. }
  86. + (instancetype)persistenceWithLruParams:(firebase::firestore::local::LruParams)lruParams
  87. serializer:(FSTLocalSerializer *)serializer {
  88. FSTMemoryPersistence *persistence = [[FSTMemoryPersistence alloc] init];
  89. persistence.referenceDelegate =
  90. [[FSTMemoryLRUReferenceDelegate alloc] initWithPersistence:persistence
  91. serializer:serializer
  92. lruParams:lruParams];
  93. return persistence;
  94. }
  95. - (instancetype)init {
  96. if (self = [super init]) {
  97. _queryCache = absl::make_unique<MemoryQueryCache>(self);
  98. _remoteDocumentCache = absl::make_unique<MemoryRemoteDocumentCache>(self);
  99. self.started = YES;
  100. }
  101. return self;
  102. }
  103. - (void)setReferenceDelegate:(id<FSTReferenceDelegate>)referenceDelegate {
  104. _referenceDelegate = referenceDelegate;
  105. id delegate = _referenceDelegate;
  106. if ([delegate conformsToProtocol:@protocol(FSTTransactional)]) {
  107. _transactionRunner.SetBackingPersistence((id<FSTTransactional>)_referenceDelegate);
  108. }
  109. }
  110. - (void)shutdown {
  111. // No durable state to ensure is closed on shutdown.
  112. HARD_ASSERT(self.isStarted, "FSTMemoryPersistence shutdown without start!");
  113. self.started = NO;
  114. }
  115. - (id<FSTReferenceDelegate>)referenceDelegate {
  116. return _referenceDelegate;
  117. }
  118. - (ListenSequenceNumber)currentSequenceNumber {
  119. return [_referenceDelegate currentSequenceNumber];
  120. }
  121. - (const FSTTransactionRunner &)run {
  122. return _transactionRunner;
  123. }
  124. - (MemoryMutationQueue *)mutationQueueForUser:(const User &)user {
  125. const std::unique_ptr<MemoryMutationQueue> &existing = _mutationQueues[user];
  126. if (!existing) {
  127. _mutationQueues[user] = absl::make_unique<MemoryMutationQueue>(self);
  128. return _mutationQueues[user].get();
  129. } else {
  130. return existing.get();
  131. }
  132. }
  133. - (MemoryQueryCache *)queryCache {
  134. return _queryCache.get();
  135. }
  136. - (MemoryRemoteDocumentCache *)remoteDocumentCache {
  137. return _remoteDocumentCache.get();
  138. }
  139. - (MemoryIndexManager *)indexManager {
  140. return &_indexManager;
  141. }
  142. @end
  143. @implementation FSTMemoryLRUReferenceDelegate {
  144. local::LruDelegateBridge _delegateBridge;
  145. // This delegate should have the same lifetime as the persistence layer, but mark as
  146. // weak to avoid retain cycle.
  147. __weak FSTMemoryPersistence *_persistence;
  148. // Tracks sequence numbers of when documents are used. Equivalent to sentinel rows in
  149. // the leveldb implementation.
  150. std::unordered_map<DocumentKey, ListenSequenceNumber, DocumentKeyHash> _sequenceNumbers;
  151. ReferenceSet *_additionalReferences;
  152. std::unique_ptr<LruGarbageCollector> _gc;
  153. // PORTING NOTE: when this class is ported to C++, this does not need to be a pointer
  154. std::unique_ptr<ListenSequence> _listenSequence;
  155. ListenSequenceNumber _currentSequenceNumber;
  156. FSTLocalSerializer *_serializer;
  157. }
  158. - (instancetype)initWithPersistence:(FSTMemoryPersistence *)persistence
  159. serializer:(FSTLocalSerializer *)serializer
  160. lruParams:(firebase::firestore::local::LruParams)lruParams {
  161. if (self = [super init]) {
  162. _persistence = persistence;
  163. _delegateBridge = local::LruDelegateBridge(self);
  164. _gc = absl::make_unique<LruGarbageCollector>(&_delegateBridge, lruParams);
  165. _currentSequenceNumber = local::kListenSequenceNumberInvalid;
  166. // Theoretically this is always 0, since this is all in-memory...
  167. ListenSequenceNumber highestSequenceNumber =
  168. _persistence.queryCache->highest_listen_sequence_number();
  169. _listenSequence = absl::make_unique<ListenSequence>(highestSequenceNumber);
  170. _serializer = serializer;
  171. }
  172. return self;
  173. }
  174. - (local::LruGarbageCollector *)gc {
  175. return _gc.get();
  176. }
  177. - (ListenSequenceNumber)currentSequenceNumber {
  178. HARD_ASSERT(_currentSequenceNumber != local::kListenSequenceNumberInvalid,
  179. "Asking for a sequence number outside of a transaction");
  180. return _currentSequenceNumber;
  181. }
  182. - (void)addInMemoryPins:(ReferenceSet *)set {
  183. // Technically can't assert this, due to restartWithNoopGarbageCollector (for now...)
  184. // FSTAssert(_additionalReferences == nil, @"Overwriting additional references");
  185. _additionalReferences = set;
  186. }
  187. - (void)removeTarget:(const QueryData &)queryData {
  188. QueryData updated = queryData.Copy(queryData.snapshot_version(), queryData.resume_token(),
  189. _currentSequenceNumber);
  190. _persistence.queryCache->UpdateTarget(updated);
  191. }
  192. - (void)limboDocumentUpdated:(const DocumentKey &)key {
  193. _sequenceNumbers[key] = self.currentSequenceNumber;
  194. }
  195. - (void)startTransaction:(absl::string_view)label {
  196. _currentSequenceNumber = _listenSequence->Next();
  197. }
  198. - (void)commitTransaction {
  199. _currentSequenceNumber = local::kListenSequenceNumberInvalid;
  200. }
  201. - (void)enumerateTargetsUsingCallback:(const TargetCallback &)callback {
  202. return _persistence.queryCache->EnumerateTargets(callback);
  203. }
  204. - (void)enumerateMutationsUsingCallback:
  205. (const firebase::firestore::local::OrphanedDocumentCallback &)callback {
  206. for (const auto &entry : _sequenceNumbers) {
  207. ListenSequenceNumber sequenceNumber = entry.second;
  208. const DocumentKey &key = entry.first;
  209. // Pass in the exact sequence number as the upper bound so we know it won't be pinned by being
  210. // too recent.
  211. if (![self isPinnedAtSequenceNumber:sequenceNumber document:key]) {
  212. callback(key, sequenceNumber);
  213. }
  214. }
  215. }
  216. - (int)removeTargetsThroughSequenceNumber:(ListenSequenceNumber)sequenceNumber
  217. liveQueries:
  218. (const std::unordered_map<TargetId, QueryData> &)liveQueries {
  219. return _persistence.queryCache->RemoveTargets(sequenceNumber, liveQueries);
  220. }
  221. - (size_t)sequenceNumberCount {
  222. size_t totalCount = _persistence.queryCache->size();
  223. [self enumerateMutationsUsingCallback:[&totalCount](const DocumentKey &key,
  224. ListenSequenceNumber sequenceNumber) {
  225. totalCount++;
  226. }];
  227. return totalCount;
  228. }
  229. - (int)removeOrphanedDocumentsThroughSequenceNumber:(ListenSequenceNumber)upperBound {
  230. std::vector<DocumentKey> removed =
  231. _persistence.remoteDocumentCache->RemoveOrphanedDocuments(self, upperBound);
  232. for (const auto &key : removed) {
  233. _sequenceNumbers.erase(key);
  234. }
  235. return static_cast<int>(removed.size());
  236. }
  237. - (void)addReference:(const DocumentKey &)key {
  238. _sequenceNumbers[key] = self.currentSequenceNumber;
  239. }
  240. - (void)removeReference:(const DocumentKey &)key {
  241. _sequenceNumbers[key] = self.currentSequenceNumber;
  242. }
  243. - (BOOL)mutationQueuesContainKey:(const DocumentKey &)key {
  244. const MutationQueues &queues = [_persistence mutationQueues];
  245. for (const auto &entry : queues) {
  246. if (entry.second->ContainsKey(key)) {
  247. return YES;
  248. }
  249. }
  250. return NO;
  251. }
  252. - (void)removeMutationReference:(const DocumentKey &)key {
  253. _sequenceNumbers[key] = self.currentSequenceNumber;
  254. }
  255. - (BOOL)isPinnedAtSequenceNumber:(ListenSequenceNumber)upperBound
  256. document:(const DocumentKey &)key {
  257. if ([self mutationQueuesContainKey:key]) {
  258. return YES;
  259. }
  260. if (_additionalReferences->ContainsKey(key)) {
  261. return YES;
  262. }
  263. if (_persistence.queryCache->Contains(key)) {
  264. return YES;
  265. }
  266. auto it = _sequenceNumbers.find(key);
  267. if (it != _sequenceNumbers.end() && it->second > upperBound) {
  268. return YES;
  269. }
  270. return NO;
  271. }
  272. - (size_t)byteSize {
  273. // Note that this method is only used for testing because this delegate is only
  274. // used for testing. The algorithm here (loop through everything, serialize it
  275. // and count bytes) is inefficient and inexact, but won't run in production.
  276. size_t count = 0;
  277. ProtoSizer sizer(_serializer);
  278. count += _persistence.queryCache->CalculateByteSize(sizer);
  279. count += _persistence.remoteDocumentCache->CalculateByteSize(sizer);
  280. const MutationQueues &queues = [_persistence mutationQueues];
  281. for (const auto &entry : queues) {
  282. count += entry.second->CalculateByteSize(_serializer);
  283. }
  284. return count;
  285. }
  286. @end
  287. @implementation FSTMemoryEagerReferenceDelegate {
  288. std::unique_ptr<std::unordered_set<DocumentKey, DocumentKeyHash>> _orphaned;
  289. // This delegate should have the same lifetime as the persistence layer, but mark as
  290. // weak to avoid retain cycle.
  291. __weak FSTMemoryPersistence *_persistence;
  292. ReferenceSet *_additionalReferences;
  293. }
  294. - (instancetype)initWithPersistence:(FSTMemoryPersistence *)persistence {
  295. if (self = [super init]) {
  296. _persistence = persistence;
  297. }
  298. return self;
  299. }
  300. - (ListenSequenceNumber)currentSequenceNumber {
  301. return local::kListenSequenceNumberInvalid;
  302. }
  303. - (void)addInMemoryPins:(ReferenceSet *)set {
  304. // We should be able to assert that _additionalReferences is nil, but due to restarts in spec
  305. // tests it would fail.
  306. _additionalReferences = set;
  307. }
  308. - (void)removeTarget:(const QueryData &)queryData {
  309. for (const DocumentKey &docKey :
  310. _persistence.queryCache->GetMatchingKeys(queryData.target_id())) {
  311. _orphaned->insert(docKey);
  312. }
  313. _persistence.queryCache->RemoveTarget(queryData);
  314. }
  315. - (void)addReference:(const DocumentKey &)key {
  316. _orphaned->erase(key);
  317. }
  318. - (void)removeReference:(const DocumentKey &)key {
  319. _orphaned->insert(key);
  320. }
  321. - (void)removeMutationReference:(const DocumentKey &)key {
  322. _orphaned->insert(key);
  323. }
  324. - (BOOL)isReferenced:(const DocumentKey &)key {
  325. if (_persistence.queryCache->Contains(key)) {
  326. return YES;
  327. }
  328. if ([self mutationQueuesContainKey:key]) {
  329. return YES;
  330. }
  331. if (_additionalReferences->ContainsKey(key)) {
  332. return YES;
  333. }
  334. return NO;
  335. }
  336. - (void)limboDocumentUpdated:(const DocumentKey &)key {
  337. if ([self isReferenced:key]) {
  338. _orphaned->erase(key);
  339. } else {
  340. _orphaned->insert(key);
  341. }
  342. }
  343. - (void)startTransaction:(__unused absl::string_view)label {
  344. _orphaned = absl::make_unique<std::unordered_set<DocumentKey, DocumentKeyHash>>();
  345. }
  346. - (BOOL)mutationQueuesContainKey:(const DocumentKey &)key {
  347. const MutationQueues &queues = [_persistence mutationQueues];
  348. for (const auto &entry : queues) {
  349. if (entry.second->ContainsKey(key)) {
  350. return YES;
  351. }
  352. }
  353. return NO;
  354. }
  355. - (void)commitTransaction {
  356. for (const auto &key : *_orphaned) {
  357. if (![self isReferenced:key]) {
  358. _persistence.remoteDocumentCache->Remove(key);
  359. }
  360. }
  361. _orphaned.reset();
  362. }
  363. @end
  364. NS_ASSUME_NONNULL_END