add Log.cacheLock.RLock(), fix influxdata#4369 and influxdata#4368

add Log.cacheLock.RLock(),  fix influxdata#4369 and influxdata#4368

tsdb/engine/tsm1/wal.go

@@ -557,13 +557,18 @@ func (l *Log) flush(flush flushType) error {
 	if flush == idleFlush {
 		if l.currentSegmentFile != nil {
 			if err := l.currentSegmentFile.Close(); err != nil {
+				l.cacheLock.Unlock()
+				l.writeLock.Unlock()
 				return err
 			}
 			l.currentSegmentFile = nil
 			l.currentSegmentSize = 0
 		}
 	} else {
 		if err := l.newSegmentFile(); err != nil {
+			l.cacheLock.Unlock()
+			l.writeLock.Unlock()
+
 			// there's no recovering from this, fail hard
 			panic(fmt.Sprintf("error creating new wal file: %s", err.Error()))
 		}
@@ -597,9 +602,11 @@ func (l *Log) flush(flush flushType) error {
 	l.seriesToCreateCache = nil
 
 	l.cacheLock.Unlock()
+	l.cacheLock.RLock()
 
 	// exit if there's nothing to flush to the index
 	if len(l.flushCache) == 0 && len(mfc) == 0 && len(scc) == 0 && flush != startupFlush {
+		l.cacheLock.RUnlock()
 		return nil
 	}
 
@@ -615,8 +622,11 @@ func (l *Log) flush(flush flushType) error {
 
 	startTime := time.Now()
 	if err := l.Index.Write(l.flushCache, mfc, scc); err != nil {
+		l.cacheLock.RUnlock()
 		return err
 	}
+	l.cacheLock.RUnlock()
+
 	if l.LoggingEnabled {
 		l.logger.Printf("%s flush to index took %s\n", l.path, time.Since(startTime))
 	}