test/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2024-12-02 11:59:00 +08:00
Code Issues Actions 12 Packages Projects Releases Wiki Activity

Add rocksmq comments (#8474)

Browse Source 
Signed-off-by: fishpenguin <kun.yu@zilliz.com>
This commit is contained in:
yukun 2021-09-24 21:45:56 +08:00 committed by GitHub
parent 3aabc1d3ee
commit 5e25041254
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 14 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
internal/util/rocksmq/server/rocksmq/global_rmq.go
View File

@ -24,16 +24,23 @@ import (
rocksdbkv "github.com/milvus-io/milvus/internal/kv/rocksdb"
)
// Global rocksmq instance that will be initialized only once
var Rmq *rocksmq
// once is used to init global rocksmq
var once sync.Once
// Params provide params that rocksmq needs
var params paramtable.BaseTable
// InitRmq is deprecate implementation of global rocksmq. will be removed later
func InitRmq(rocksdbName string, idAllocator allocator.GIDAllocator) error {
var err error
Rmq, err = NewRocksMQ(rocksdbName, idAllocator)
return err
}
// InitRocksMQ init global rocksmq single instance
func InitRocksMQ() error {
var err error
once.Do(func() {
@ -68,6 +75,7 @@ func InitRocksMQ() error {
return err
}
// CloseRocksMQ is used to close global rocksmq
func CloseRocksMQ() {
log.Debug("Close Rocksmq!")
if Rmq != nil && Rmq.store != nil {

6
internal/util/rocksmq/server/rocksmq/rocksmq_impl.go
View File

@ -28,10 +28,13 @@ import (
rocksdbkv "github.com/milvus-io/milvus/internal/kv/rocksdb"
)
// UniqueID is the type of message ID
type UniqueID = typeutil.UniqueID
// RocksmqPageSize is the size of a message page, default 2GB
var RocksmqPageSize int64 = 2 << 30
// Const variable that will be used in rocksmqs
const (
DefaultMessageID = "-1"
FixedChannelNameLen = 320
@ -109,6 +112,9 @@ type rocksmq struct {
retentionInfo *retentionInfo
}
// 1. New rocksmq instance based on rocksdb with name and rocksdbkv with kvname
// 2. Init retention info, load retention info to memory
// 3. Start retention goroutine
func NewRocksMQ(name string, idAllocator allocator.GIDAllocator) (*rocksmq, error) {
bbto := gorocksdb.NewDefaultBlockBasedTableOptions()
bbto.SetBlockCache(gorocksdb.NewLRUCache(RocksDBLRUCacheCapacity))