Update hook expiration logic

tidwall · Sep 12, 2021 · 83094b2 · 83094b2
1 parent c686b87
commit 83094b2
Show file tree

Hide file tree

Showing 6 changed files with 89 additions and 266 deletions.
diff --git a/internal/expire/expire.go b/internal/expire/expire.go
diff --git a/internal/expire/expire_test.go b/internal/expire/expire_test.go
diff --git a/internal/server/crud.go b/internal/server/crud.go
@@ -508,6 +508,7 @@ func (server *Server) cmdFlushDB(msg *Message) (res resp.Value, d commandDetails
 	server.cols = btree.NewNonConcurrent(byCollectionKey)
 	server.groupHooks = btree.NewNonConcurrent(byGroupHook)
 	server.groupObjects = btree.NewNonConcurrent(byGroupObject)
+	server.hookExpires = btree.NewNonConcurrent(byHookExpires)
 	server.hooks = make(map[string]*Hook)
 	server.hooksOut = make(map[string]*Hook)
 	server.hookTree = &rtree.RTree{}

diff --git a/internal/server/expire.go b/internal/server/expire.go
@@ -18,32 +18,70 @@ func (s *Server) backgroundExpiring() {
 		func() {
 			s.mu.Lock()
 			defer s.mu.Unlock()
-			now := time.Now().UnixNano()
-			var ids []string
-			var msgs []*Message
-			s.cols.Ascend(nil, func(v interface{}) bool {
-				col := v.(*collectionKeyContainer)
-				ids = col.col.Expired(now, ids[:0])
-				for _, id := range ids {
-					msgs = append(msgs, &Message{
-						Args: []string{"del", col.key, id},
-					})
-				}
-				return true
-			})
-			for _, msg := range msgs {
-				_, d, err := s.cmdDel(msg)
-				if err != nil {
-					log.Fatal(err)
-				}
-				if err := s.writeAOF(msg.Args, &d); err != nil {
-					log.Fatal(err)
-				}
-			}
-			if len(msgs) > 0 {
-				log.Debugf("Expired %d items\n", len(msgs))
-			}
+			now := time.Now()
+			s.backgroundExpireObjects(now)
+			s.backgroundExpireHooks(now)
 		}()
 		time.Sleep(bgExpireDelay)
 	}
 }
+
+func (s *Server) backgroundExpireObjects(now time.Time) {
+	nano := now.UnixNano()
+	var ids []string
+	var msgs []*Message
+	s.cols.Ascend(nil, func(v interface{}) bool {
+		col := v.(*collectionKeyContainer)
+		ids = col.col.Expired(nano, ids[:0])
+		for _, id := range ids {
+			msgs = append(msgs, &Message{
+				Args: []string{"del", col.key, id},
+			})
+		}
+		return true
+	})
+	for _, msg := range msgs {
+		_, d, err := s.cmdDel(msg)
+		if err != nil {
+			log.Fatal(err)
+		}
+		if err := s.writeAOF(msg.Args, &d); err != nil {
+			log.Fatal(err)
+		}
+	}
+	if len(msgs) > 0 {
+		log.Debugf("Expired %d objects\n", len(msgs))
+	}
+
+}
+
+func (s *Server) backgroundExpireHooks(now time.Time) {
+	var msgs []*Message
+	s.hookExpires.Ascend(nil, func(v interface{}) bool {
+		h := v.(*Hook)
+		if h.expires.After(now) {
+			return false
+		}
+		msg := &Message{}
+		if h.channel {
+			msg.Args = []string{"delchan", h.Name}
+		} else {
+			msg.Args = []string{"delhook", h.Name}
+		}
+		msgs = append(msgs, msg)
+		return true
+	})
+
+	for _, msg := range msgs {
+		_, d, err := s.cmdDelHook(msg, msg.Args[0] == "delchan")
+		if err != nil {
+			log.Fatal(err)
+		}
+		if err := s.writeAOF(msg.Args, &d); err != nil {
+			log.Fatal(err)
+		}
+	}
+	if len(msgs) > 0 {
+		log.Debugf("Expired %d hooks\n", len(msgs))
+	}
+}