seaweedfs/weed/messaging/broker/topic_manager.go

124 lines
2.8 KiB
Go
Raw Normal View History

2020-04-18 16:12:01 +08:00
package broker
import (
2020-04-19 15:18:32 +08:00
"fmt"
2020-04-18 16:12:01 +08:00
"sync"
"time"
2020-04-19 15:18:32 +08:00
"github.com/chrislusf/seaweedfs/weed/filer2"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/messaging_pb"
2020-04-18 16:12:01 +08:00
"github.com/chrislusf/seaweedfs/weed/util/log_buffer"
)
type TopicPartition struct {
Namespace string
Topic string
Partition int32
}
2020-05-10 18:50:30 +08:00
2020-05-08 17:47:22 +08:00
const (
TopicPartitionFmt = "%s/%s_%02d"
2020-05-08 17:47:22 +08:00
)
2020-05-10 18:50:30 +08:00
2020-05-08 17:47:22 +08:00
func (tp *TopicPartition) String() string {
return fmt.Sprintf(TopicPartitionFmt, tp.Namespace, tp.Topic, tp.Partition)
}
2020-05-13 13:55:55 +08:00
type TopicControl struct {
2020-04-18 16:12:01 +08:00
sync.Mutex
2020-04-21 08:48:06 +08:00
cond *sync.Cond
2020-04-18 16:12:01 +08:00
subscriberCount int
publisherCount int
logBuffer *log_buffer.LogBuffer
}
2020-05-12 23:48:00 +08:00
type TopicManager struct {
2020-04-18 16:12:01 +08:00
sync.Mutex
2020-05-16 12:38:42 +08:00
topicControls map[TopicPartition]*TopicControl
broker *MessageBroker
2020-04-18 16:12:01 +08:00
}
2020-05-12 23:48:00 +08:00
func NewTopicManager(messageBroker *MessageBroker) *TopicManager {
return &TopicManager{
2020-05-16 12:38:42 +08:00
topicControls: make(map[TopicPartition]*TopicControl),
broker: messageBroker,
2020-04-18 16:12:01 +08:00
}
}
2020-05-16 12:38:42 +08:00
func (tm *TopicManager) buildLogBuffer(tl *TopicControl, tp TopicPartition, topicConfig *messaging_pb.TopicConfiguration) *log_buffer.LogBuffer {
2020-04-18 16:12:01 +08:00
2020-04-19 15:18:32 +08:00
flushFn := func(startTime, stopTime time.Time, buf []byte) {
if topicConfig.IsTransient {
2020-04-30 18:05:34 +08:00
// return
}
2020-04-20 14:37:40 +08:00
// fmt.Printf("flushing with topic config %+v\n", topicConfig)
2020-04-19 15:18:32 +08:00
targetFile := fmt.Sprintf(
"%s/%s/%s/%04d-%02d-%02d/%02d-%02d.part%02d",
filer2.TopicsDir, tp.Namespace, tp.Topic,
startTime.Year(), startTime.Month(), startTime.Day(), startTime.Hour(), startTime.Minute(),
tp.Partition,
)
2020-05-12 23:48:00 +08:00
if err := tm.broker.appendToFile(targetFile, topicConfig, buf); err != nil {
2020-04-19 15:18:32 +08:00
glog.V(0).Infof("log write failed %s: %v", targetFile, err)
}
2020-04-18 16:12:01 +08:00
}
2020-04-19 15:18:32 +08:00
logBuffer := log_buffer.NewLogBuffer(time.Minute, flushFn, func() {
2020-05-16 12:38:42 +08:00
tl.cond.Broadcast()
2020-04-19 15:18:32 +08:00
})
2020-04-18 16:12:01 +08:00
2020-04-19 15:18:32 +08:00
return logBuffer
2020-04-18 16:12:01 +08:00
}
2020-05-13 13:55:55 +08:00
func (tm *TopicManager) RequestLock(partition TopicPartition, topicConfig *messaging_pb.TopicConfiguration, isPublisher bool) *TopicControl {
2020-05-12 23:48:00 +08:00
tm.Lock()
defer tm.Unlock()
2020-04-18 16:12:01 +08:00
2020-05-16 12:38:42 +08:00
tc, found := tm.topicControls[partition]
2020-04-18 16:12:01 +08:00
if !found {
2020-05-13 13:55:55 +08:00
tc = &TopicControl{}
2020-05-16 12:38:42 +08:00
tc.cond = sync.NewCond(&tc.Mutex)
tm.topicControls[partition] = tc
2020-05-13 13:55:55 +08:00
tc.logBuffer = tm.buildLogBuffer(tc, partition, topicConfig)
2020-04-18 16:12:01 +08:00
}
2020-04-19 15:18:32 +08:00
if isPublisher {
2020-05-13 13:55:55 +08:00
tc.publisherCount++
2020-04-19 15:18:32 +08:00
} else {
2020-05-13 13:55:55 +08:00
tc.subscriberCount++
2020-04-19 15:18:32 +08:00
}
2020-05-13 13:55:55 +08:00
return tc
2020-04-18 16:12:01 +08:00
}
2020-05-12 23:48:00 +08:00
func (tm *TopicManager) ReleaseLock(partition TopicPartition, isPublisher bool) {
tm.Lock()
defer tm.Unlock()
2020-04-18 16:12:01 +08:00
2020-05-16 12:38:42 +08:00
lock, found := tm.topicControls[partition]
2020-04-18 16:12:01 +08:00
if !found {
return
}
if isPublisher {
lock.publisherCount--
} else {
lock.subscriberCount--
}
if lock.subscriberCount <= 0 && lock.publisherCount <= 0 {
2020-05-16 12:38:42 +08:00
delete(tm.topicControls, partition)
lock.logBuffer.Shutdown()
2020-04-18 16:12:01 +08:00
}
}
2020-05-08 17:47:22 +08:00
2020-05-12 23:48:00 +08:00
func (tm *TopicManager) ListTopicPartitions() (tps []TopicPartition) {
tm.Lock()
defer tm.Unlock()
2020-05-08 17:47:22 +08:00
2020-05-16 12:38:42 +08:00
for k := range tm.topicControls {
2020-05-08 17:47:22 +08:00
tps = append(tps, k)
}
return
}