DelayedCacheWriter must be run by the caller of an interface that uses delayed cache writing.
(wc *mgr.WorkerCtx)
| 11 | |
| 12 | // DelayedCacheWriter must be run by the caller of an interface that uses delayed cache writing. |
| 13 | func (i *Interface) DelayedCacheWriter(wc *mgr.WorkerCtx) error { |
| 14 | // Check if the DelayedCacheWriter should be run at all. |
| 15 | if i.options.CacheSize <= 0 || i.options.DelayCachedWrites == "" { |
| 16 | return errors.New("delayed cache writer is not applicable to this database interface") |
| 17 | } |
| 18 | |
| 19 | // Check if backend support the Batcher interface. |
| 20 | batchPut := i.PutMany(i.options.DelayCachedWrites) |
| 21 | // End batchPut immediately and check for an error. |
| 22 | err := batchPut(nil) |
| 23 | if err != nil { |
| 24 | return err |
| 25 | } |
| 26 | |
| 27 | // percentThreshold defines the minimum percentage of entries in the write cache in relation to the cache size that need to be present in order for flushing the cache to the database storage. |
| 28 | percentThreshold := 25 |
| 29 | thresholdWriteTicker := time.NewTicker(5 * time.Second) |
| 30 | forceWriteTicker := time.NewTicker(5 * time.Minute) |
| 31 | |
| 32 | for { |
| 33 | // Wait for trigger for writing the cache. |
| 34 | select { |
| 35 | case <-wc.Done(): |
| 36 | // The caller is shutting down, flush the cache to storage and exit. |
| 37 | i.flushWriteCache(0) |
| 38 | return nil |
| 39 | |
| 40 | case <-i.triggerCacheWrite: |
| 41 | // An entry from the cache was evicted that was also in the write cache. |
| 42 | // This makes it likely that other entries that are also present in the |
| 43 | // write cache will be evicted soon. Flush the write cache to storage |
| 44 | // immediately in order to reduce single writes. |
| 45 | i.flushWriteCache(0) |
| 46 | |
| 47 | case <-thresholdWriteTicker.C: |
| 48 | // Often check if the write cache has filled up to a certain degree and |
| 49 | // flush it to storage before we start evicting to-be-written entries and |
| 50 | // slow down the hot path again. |
| 51 | i.flushWriteCache(percentThreshold) |
| 52 | |
| 53 | case <-forceWriteTicker.C: |
| 54 | // Once in a while, flush the write cache to storage no matter how much |
| 55 | // it is filled. We don't want entries lingering around in the write |
| 56 | // cache forever. This also reduces the amount of data loss in the event |
| 57 | // of a total crash. |
| 58 | i.flushWriteCache(0) |
| 59 | } |
| 60 | } |
| 61 | } |
| 62 | |
| 63 | // ClearCache clears the read cache. |
| 64 | func (i *Interface) ClearCache() { |