-
Notifications
You must be signed in to change notification settings - Fork 107
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
versiondb streaming service checkpoint
- Loading branch information
Showing
12 changed files
with
1,316 additions
and
9 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,164 @@ | ||
| package app | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "strings" | ||
| "sync" | ||
|
|
||
| "github.com/cosmos/cosmos-sdk/baseapp" | ||
| "github.com/cosmos/cosmos-sdk/codec" | ||
| serverTypes "github.com/cosmos/cosmos-sdk/server/types" | ||
| "github.com/cosmos/cosmos-sdk/store/streaming" | ||
| "github.com/cosmos/cosmos-sdk/store/types" | ||
| sdk "github.com/cosmos/cosmos-sdk/types" | ||
|
|
||
| "github.com/public-awesome/stargaze/v11/versiondb" | ||
| "github.com/spf13/cast" | ||
| ) | ||
|
|
||
| // ServiceConstructor is used to construct a streaming service | ||
| type ServiceConstructor func(serverTypes.AppOptions, []types.StoreKey, codec.BinaryCodec) (baseapp.StreamingService, error) | ||
|
|
||
| // ServiceType enum for specifying the type of StreamingService | ||
| type ServiceType int | ||
|
|
||
| const ( | ||
| Unknown ServiceType = iota | ||
| File | ||
| VersionDb | ||
| ) | ||
|
|
||
| // Streaming option keys | ||
| const ( | ||
| OptStoreStreamers = "store.streamers" | ||
| ) | ||
|
|
||
| // ServiceTypeFromString returns the streaming.ServiceType corresponding to the | ||
| // provided name. | ||
| func ServiceTypeFromString(name string) ServiceType { | ||
| switch strings.ToLower(name) { | ||
| case "file", "f": | ||
| return File | ||
| case "versiondb", "v": | ||
| return VersionDb | ||
|
|
||
| default: | ||
| return Unknown | ||
| } | ||
| } | ||
|
|
||
| // String returns the string name of a streaming.ServiceType | ||
| func (sst ServiceType) String() string { | ||
| switch sst { | ||
| case File: | ||
| return "file" | ||
| case VersionDb: | ||
| return "versiondb" | ||
|
|
||
| default: | ||
| return "unknown" | ||
| } | ||
| } | ||
|
|
||
| // ServiceConstructorLookupTable is a mapping of streaming.ServiceTypes to | ||
| // streaming.ServiceConstructors types. | ||
| var ServiceConstructorLookupTable = map[ServiceType]ServiceConstructor{ | ||
| File: streaming.NewFileStreamingService, | ||
| VersionDb: versiondb.NewVersionDbStreamingService, | ||
| } | ||
|
|
||
| // NewServiceConstructor returns the streaming.ServiceConstructor corresponding | ||
| // to the provided name. | ||
| func NewServiceConstructor(name string) (ServiceConstructor, error) { | ||
| ssType := ServiceTypeFromString(name) | ||
| if ssType == Unknown { | ||
| return nil, fmt.Errorf("unrecognized streaming service name %s", name) | ||
| } | ||
|
|
||
| if constructor, ok := ServiceConstructorLookupTable[ssType]; ok && constructor != nil { | ||
| return constructor, nil | ||
| } | ||
|
|
||
| return nil, fmt.Errorf("streaming service constructor of type %s not found", ssType.String()) | ||
| } | ||
|
|
||
| // LoadStreamingServices is a function for loading StreamingServices onto the | ||
| // BaseApp using the provided AppOptions, codec, and keys. It returns the | ||
| // WaitGroup and quit channel used to synchronize with the streaming services | ||
| // and any error that occurs during the setup. | ||
| func LoadStreamingServices( | ||
| bApp *baseapp.BaseApp, | ||
| appOpts serverTypes.AppOptions, | ||
| appCodec codec.BinaryCodec, | ||
| keys map[string]*types.KVStoreKey, | ||
| ) ([]baseapp.StreamingService, *sync.WaitGroup, error) { | ||
| // waitgroup and quit channel for optional shutdown coordination of the streaming service(s) | ||
| wg := new(sync.WaitGroup) | ||
|
|
||
| // configure state listening capabilities using AppOptions | ||
| streamers := cast.ToStringSlice(appOpts.Get(OptStoreStreamers)) | ||
| activeStreamers := make([]baseapp.StreamingService, 0, len(streamers)) | ||
|
|
||
| for _, streamerName := range streamers { | ||
| var exposeStoreKeys []types.StoreKey | ||
|
|
||
| // get the store keys allowed to be exposed for this streaming service | ||
| exposeKeyStrs := cast.ToStringSlice(appOpts.Get(fmt.Sprintf("streamers.%s.keys", streamerName))) | ||
|
|
||
| // if list contains '*', expose all store keys | ||
| if sdk.SliceContains(exposeKeyStrs, "*") { | ||
| exposeStoreKeys = make([]types.StoreKey, 0, len(keys)) | ||
| for _, storeKey := range keys { | ||
| exposeStoreKeys = append(exposeStoreKeys, storeKey) | ||
| } | ||
| } else { | ||
| exposeStoreKeys = make([]types.StoreKey, 0, len(exposeKeyStrs)) | ||
| for _, keyStr := range exposeKeyStrs { | ||
| if storeKey, ok := keys[keyStr]; ok { | ||
| exposeStoreKeys = append(exposeStoreKeys, storeKey) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| if len(exposeStoreKeys) == 0 { | ||
| continue | ||
| } | ||
|
|
||
| constructor, err := NewServiceConstructor(streamerName) | ||
| if err != nil { | ||
| // Close any services we may have already spun up before hitting the error | ||
| // on this one. | ||
| for _, activeStreamer := range activeStreamers { | ||
| activeStreamer.Close() | ||
| } | ||
|
|
||
| return nil, nil, err | ||
| } | ||
|
|
||
| // Generate the streaming service using the constructor, appOptions, and the | ||
| // StoreKeys we want to expose. | ||
| streamingService, err := constructor(appOpts, exposeStoreKeys, appCodec) | ||
| if err != nil { | ||
| // Close any services we may have already spun up before hitting the error | ||
| // on this one. | ||
| for _, activeStreamer := range activeStreamers { | ||
| activeStreamer.Close() | ||
| } | ||
|
|
||
| return nil, nil, err | ||
| } | ||
|
|
||
| // register the streaming service with the BaseApp | ||
| bApp.SetStreamingService(streamingService) | ||
|
|
||
| // kick off the background streaming service loop | ||
| streamingService.Stream(wg) | ||
|
|
||
| // add to the list of active streamers | ||
| activeStreamers = append(activeStreamers, streamingService) | ||
| } | ||
|
|
||
| // If there are no active streamers, activeStreamers is empty (len == 0) and | ||
| // the waitGroup is not waiting on anything. | ||
| return activeStreamers, wg, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,116 @@ | ||
| package versiondb | ||
|
|
||
| import ( | ||
| "context" | ||
| "os" | ||
| "path/filepath" | ||
| "sort" | ||
| "strings" | ||
| "sync" | ||
|
|
||
| "github.com/public-awesome/stargaze/v11/versiondb/tsrocksdb" | ||
| abci "github.com/tendermint/tendermint/abci/types" | ||
|
|
||
| "github.com/cosmos/cosmos-sdk/baseapp" | ||
| "github.com/cosmos/cosmos-sdk/codec" | ||
| "github.com/cosmos/cosmos-sdk/store/types" | ||
| storetypes "github.com/cosmos/cosmos-sdk/store/types" | ||
| ) | ||
|
|
||
| var _ baseapp.StreamingService = &StreamingService{} | ||
|
|
||
| // StreamingService is a concrete implementation of StreamingService that accumulate the state changes in current block, | ||
| // writes the ordered changeset out to version storage. | ||
| type StreamingService struct { | ||
| listeners []*types.MemoryListener // the listeners that will be initialized with BaseApp | ||
| VersionStore VersionStore | ||
| CurrentBlockNumber int64 // the current block number | ||
| } | ||
|
|
||
| // NewFileStreamingService is the streaming.ServiceConstructor function for | ||
| // creating a FileStreamingService. | ||
| func NewVersionDbStreamingService( | ||
| homePath string, | ||
| keys []storetypes.StoreKey, | ||
| marshaller codec.BinaryCodec, | ||
| ) (*StreamingService, error) { | ||
| dataDir := filepath.Join(homePath, "data", "versiondb") | ||
| if err := os.MkdirAll(dataDir, os.ModePerm); err != nil { | ||
| return nil, err | ||
| } | ||
| versionDB, err := tsrocksdb.NewStore(dataDir) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| // default to exposing all | ||
| exposeStoreKeys := make([]storetypes.StoreKey, 0, len(keys)) | ||
| for _, storeKey := range keys { | ||
| exposeStoreKeys = append(exposeStoreKeys, storeKey) | ||
| } | ||
|
|
||
| service := NewStreamingService(versionDB, exposeStoreKeys) | ||
|
|
||
| return service, nil | ||
| } | ||
|
|
||
| // NewStreamingService creates a new StreamingService for the provided writeDir, (optional) filePrefix, and storeKeys | ||
| func NewStreamingService(versionStore VersionStore, storeKeys []types.StoreKey) *StreamingService { | ||
| // sort by the storeKeys first | ||
| sort.SliceStable(storeKeys, func(i, j int) bool { | ||
| return strings.Compare(storeKeys[i].Name(), storeKeys[j].Name()) < 0 | ||
| }) | ||
|
|
||
| listeners := make([]*types.MemoryListener, len(storeKeys)) | ||
| for i, key := range storeKeys { | ||
| listeners[i] = types.NewMemoryListener(key) | ||
| } | ||
| return &StreamingService{listeners, versionStore, 0} | ||
| } | ||
|
|
||
| // Listeners satisfies the baseapp.StreamingService interface | ||
| func (fss *StreamingService) Listeners() map[types.StoreKey][]types.WriteListener { | ||
| listeners := make(map[types.StoreKey][]types.WriteListener, len(fss.listeners)) | ||
| for _, listener := range fss.listeners { | ||
| listeners[listener.StoreKey()] = []types.WriteListener{listener} | ||
| } | ||
| return listeners | ||
| } | ||
|
|
||
| // ListenBeginBlock satisfies the baseapp.ABCIListener interface | ||
| // It sets the currentBlockNumber. | ||
| func (fss *StreamingService) ListenBeginBlock(ctx context.Context, req abci.RequestBeginBlock, res abci.ResponseBeginBlock) error { | ||
| fss.CurrentBlockNumber = req.GetHeader().Height | ||
| return nil | ||
| } | ||
|
|
||
| // ListenDeliverTx satisfies the baseapp.ABCIListener interface | ||
| func (fss *StreamingService) ListenDeliverTx(ctx context.Context, req abci.RequestDeliverTx, res abci.ResponseDeliverTx) error { | ||
| return nil | ||
| } | ||
|
|
||
| // ListenEndBlock satisfies the baseapp.ABCIListener interface | ||
| // It merge the state caches of all the listeners together, and write out to the versionStore. | ||
| func (fss *StreamingService) ListenEndBlock(ctx context.Context, req abci.RequestEndBlock, res abci.ResponseEndBlock) error { | ||
| return nil | ||
| } | ||
|
|
||
| func (fss *StreamingService) ListenCommit(ctx context.Context, res abci.ResponseCommit) error { | ||
| // concat the state caches | ||
| var changeSet []types.StoreKVPair | ||
| for _, listener := range fss.listeners { | ||
| changeSet = append(changeSet, listener.PopStateCache()...) | ||
| } | ||
|
|
||
| return fss.VersionStore.PutAtVersion(fss.CurrentBlockNumber, changeSet) | ||
| } | ||
|
|
||
| // Stream satisfies the baseapp.StreamingService interface | ||
| func (fss *StreamingService) Stream(wg *sync.WaitGroup) error { | ||
| return nil | ||
| } | ||
|
|
||
| // Close satisfies the io.Closer interface, which satisfies the baseapp.StreamingService interface | ||
| func (fss *StreamingService) Close() error { | ||
| return nil | ||
| } |
Oops, something went wrong.