| 414 | } |
| 415 | |
| 416 | void TupleCacheMgr::CompleteWrite(UniqueHandle handle, size_t size) { |
| 417 | DCHECK(enabled_); |
| 418 | DCHECK(handle != nullptr && handle->cache_handle != nullptr); |
| 419 | DCHECK(handle->is_writer); |
| 420 | DCHECK_LE(size, MaxSize()); |
| 421 | DCHECK_GE(size, 0); |
| 422 | if (sync_pool_size_ > 0 && |
| 423 | sync_thread_pool_->GetQueueSize() >= sync_pool_queue_depth_) { |
| 424 | // The sync_thread_pool_ has reached its max queue size. This should almost never |
| 425 | // happen, as the outstanding writes limit should kick in before this is overwhelmed. |
| 426 | // If it does happen, bail out. |
| 427 | AbortWrite(move(handle), false); |
| 428 | tuple_cache_dropped_sync_->Increment(1); |
| 429 | return; |
| 430 | } |
| 431 | VLOG_FILE << "Tuple Cache: Complete " << GetPath(handle) << " (" << size << ")"; |
| 432 | UpdateWriteSize(handle.get(), size); |
| 433 | CHECK(UpdateState(handle.get(), |
| 434 | TupleCacheState::IN_PROGRESS, TupleCacheState::COMPLETE_UNSYNCED)); |
| 435 | tuple_cache_entries_in_use_bytes_->Increment(size); |
| 436 | tuple_cache_entry_size_stats_->Update(size); |
| 437 | // When the sync_pool_size_ is 0, there is no thread pool and this does the sync |
| 438 | // directly. This is used for backend tests to avoid race conditions. |
| 439 | if (sync_pool_size_ > 0) { |
| 440 | // Offer the cache key to the thread pool. |
| 441 | bool success = sync_thread_pool_->Offer(cache_->Key(handle->cache_handle).ToString()); |
| 442 | if (!success) { |
| 443 | // The queue is full, so evict this entry |
| 444 | VLOG_FILE << "Tuple Cache: Sync thread pool queue full. Evicting " |
| 445 | << GetPath(handle); |
| 446 | cache_->Erase(cache_->Key(handle->cache_handle)); |
| 447 | tuple_cache_dropped_sync_->Increment(1); |
| 448 | } |
| 449 | } else { |
| 450 | SyncFileToDisk(cache_->Key(handle->cache_handle).ToString()); |
| 451 | } |
| 452 | } |
| 453 | |
| 454 | void TupleCacheMgr::AbortWrite(UniqueHandle handle, bool tombstone) { |
| 455 | DCHECK(enabled_); |