lock parsing per sharecode

csgo/demo_loader.go

@@ -52,6 +52,7 @@ type DemoMatchLoader struct {
 	lock       *sync.RWMutex
 	dp         *DemoParser
 	parseDemo  chan *Demo
+	parseMap   map[string]bool
 }
 
 func AccountId2SteamId(accId uint32) uint64 {
@@ -329,9 +330,17 @@ func (d *DemoMatchLoader) gcWorker(apiKey string, rl ratelimit.Limiter) {
 	for {
 		select {
 		case demo := <-d.parseDemo:
+			if _, ok := d.parseMap[demo.ShareCode]; ok {
+				log.Infof("[DL] Skipping %s: parsing in progress", demo.ShareCode)
+				continue
+			} else {
+				d.parseMap[demo.ShareCode] = true
+			}
+
 			if !d.GCReady {
 				log.Infof("[DL] Postponing match %d (%s): GC not ready", demo.MatchId, demo.ShareCode)
 				time.Sleep(5 * time.Second)
+				delete(d.parseMap, demo.ShareCode)
 				d.parseDemo <- demo
 				continue
 			}
@@ -339,6 +348,7 @@ func (d *DemoMatchLoader) gcWorker(apiKey string, rl ratelimit.Limiter) {
 			matchId, _, _, err := DecodeSharecode(demo.ShareCode)
 			if err != nil || matchId == 0 {
 				log.Warningf("[DL] Can't parse match with sharecode %s: %v", demo.ShareCode, err)
+				delete(d.parseMap, demo.ShareCode)
 				continue
 			}
 
@@ -351,6 +361,8 @@ func (d *DemoMatchLoader) gcWorker(apiKey string, rl ratelimit.Limiter) {
 					break
 				default:
 					log.Errorf("[DL] Failure trying to lookup match %d in db: %v", matchId, e)
+					delete(d.parseMap, demo.ShareCode)
+					continue
 				}
 			} else {
 				if iMatch.DemoParsed == false && iMatch.Date.After(time.Now().UTC().AddDate(0, 0, -30)) {
@@ -361,16 +373,19 @@ func (d *DemoMatchLoader) gcWorker(apiKey string, rl ratelimit.Limiter) {
 					if err != nil {
 						log.Warningf("[DL] Parsing demo from match %d failed: %v", demo.MatchId, err)
 					}
+					delete(d.parseMap, demo.ShareCode)
 					continue
 				}
 
 				log.Debugf("[DL] Skipped match %d: already parsed", matchId)
+				delete(d.parseMap, demo.ShareCode)
 				continue
 			}
 
 			matchDetails, err := d.getMatchDetails(demo.ShareCode)
 			if err != nil {
 				log.Warningf("[DL] Failure to get match-details for %d from GC: %v", demo.MatchId, err)
+				delete(d.parseMap, demo.ShareCode)
 				continue
 			}
 
@@ -409,6 +424,7 @@ func (d *DemoMatchLoader) gcWorker(apiKey string, rl ratelimit.Limiter) {
 			d.lock.Unlock()
 			if err != nil {
 				log.Warningf("[DL] Unable to create match %d: %v", matchZero.GetMatchid(), err)
+				delete(d.parseMap, demo.ShareCode)
 				continue
 			}
 
@@ -473,6 +489,7 @@ func (d *DemoMatchLoader) gcWorker(apiKey string, rl ratelimit.Limiter) {
 			if err != nil {
 				log.Warningf("[DL] Can't queue demo %d for parsing: %v", demo.MatchId, err)
 			}
+			delete(d.parseMap, demo.ShareCode)
 		}
 	}
 }