scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 17:15:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1440 from jaffee/even-more-unexport

Browse source 
Even more unexport
This commit is contained in:
Matthew Jaffee 2018-07-02 09:57:27 -05:00 • committed by GitHub
commit f84f98268f
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
13 changed files with 88 additions and 116 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
api.go
View file

@ -104,7 +104,7 @@ func (api *API) Query(ctx context.Context, req *QueryRequest) (QueryResponse, er
if err != nil {
return resp, errors.Wrap(err, "parsing")
}
execOpts := &ExecOptions{
execOpts := &execOptions{
Remote: req.Remote,
ExcludeRowAttrs: req.ExcludeRowAttrs,
ExcludeColumns: req.ExcludeColumns,
@ -788,8 +788,8 @@ func (api *API) ResizeAbort() error {
return errors.Wrap(err, "complete current job")
}
// TranslateStoreBufferSize is the buffer size used for streaming data.
const TranslateStoreBufferSize = 65536
// translateStoreBufferSize is the buffer size used for streaming data.
const translateStoreBufferSize = 65536
func (api *API) GetTranslateData(ctx context.Context, w io.WriteCloser, offset int64) error {
rc, err := api.server.primaryTranslateStore.Reader(ctx, offset)
@ -804,7 +804,7 @@ func (api *API) GetTranslateData(ctx context.Context, w io.WriteCloser, offset i
defer rc.Close()
defer w.Close()
buf := make([]byte, TranslateStoreBufferSize)
buf := make([]byte, translateStoreBufferSize)
// Copy from reader to client until store or client disconnect.
for {

44
attr.go
View file

@ -42,57 +42,39 @@ type AttrStore interface {
BlockData(i uint64) (map[uint64]map[string]interface{}, error)
}
func init() {
NopAttrStore = &nopAttrStore{}
}
// nopStore represents an AttrStore that doesn't do anything.
var nopStore AttrStore = nopAttrStore{}
// NopAttrStore represents an AttrStore that doesn't do anything.
var NopAttrStore AttrStore
func NewNopAttrStore(string) AttrStore {
return &nopAttrStore{}
}
// newNopAttrStore returns an attr store which does nothing. It returns a global
// object to avoid unecessary allocations.
func newNopAttrStore(string) AttrStore { return nopStore }
// nopAttrStore represents a no-op implementation of the AttrStore interface.
type nopAttrStore struct{}
// Path is a no-op implementation of AttrStore Path method.
func (s *nopAttrStore) Path() string { return "" }
func (s nopAttrStore) Path() string { return "" }
// Open is a no-op implementation of AttrStore Open method.
func (s *nopAttrStore) Open() error {
return nil
}
func (s nopAttrStore) Open() error { return nil }
// Close is a no-op implementation of AttrStore Close method.
func (s *nopAttrStore) Close() error {
return nil
}
func (s nopAttrStore) Close() error { return nil }
// Attrs is a no-op implementation of AttrStore Attrs method.
func (s *nopAttrStore) Attrs(id uint64) (m map[string]interface{}, err error) {
return nil, nil
}
func (s nopAttrStore) Attrs(id uint64) (m map[string]interface{}, err error) { return nil, nil }
// SetAttrs is a no-op implementation of AttrStore SetAttrs method.
func (s *nopAttrStore) SetAttrs(id uint64, m map[string]interface{}) error {
return nil
}
func (s nopAttrStore) SetAttrs(id uint64, m map[string]interface{}) error { return nil }
// SetBulkAttrs is a no-op implementation of AttrStore SetBulkAttrs method.
func (s *nopAttrStore) SetBulkAttrs(m map[uint64]map[string]interface{}) error {
return nil
}
func (s nopAttrStore) SetBulkAttrs(m map[uint64]map[string]interface{}) error { return nil }
// Blocks is a no-op implementation of AttrStore Blocks method.
func (s *nopAttrStore) Blocks() ([]AttrBlock, error) {
return nil, nil
}
func (s nopAttrStore) Blocks() ([]AttrBlock, error) { return nil, nil }
// BlockData is a no-op implementation of AttrStore BlockData method.
func (s *nopAttrStore) BlockData(i uint64) (map[uint64]map[string]interface{}, error) {
return nil, nil
}
func (s nopAttrStore) BlockData(i uint64) (map[uint64]map[string]interface{}, error) { return nil, nil }
// AttrBlock represents a checksummed block of the attribute store.
type AttrBlock struct {

18
broadcast.go
View file

@ -23,8 +23,8 @@ import (
"github.com/pkg/errors"
)
// Broadcaster is an interface for broadcasting messages.
type Broadcaster interface {
// broadcaster is an interface for broadcasting messages.
type broadcaster interface {
SendSync(pb proto.Message) error
SendAsync(pb proto.Message) error
SendTo(to *Node, pb proto.Message) error
@ -35,24 +35,18 @@ func init() {
}
// NopBroadcaster represents a Broadcaster that doesn't do anything.
var NopBroadcaster Broadcaster
var NopBroadcaster broadcaster
type nopBroadcaster struct{}
// SendSync A no-op implementation of Broadcaster SendSync method.
func (n *nopBroadcaster) SendSync(pb proto.Message) error {
return nil
}
func (n nopBroadcaster) SendSync(pb proto.Message) error { return nil }
// SendAsync A no-op implementation of Broadcaster SendAsync method.
func (n *nopBroadcaster) SendAsync(pb proto.Message) error {
return nil
}
func (n nopBroadcaster) SendAsync(pb proto.Message) error { return nil }
// SendTo is a no-op implementation of Broadcaster SendTo method.
func (c *nopBroadcaster) SendTo(to *Node, pb proto.Message) error {
return nil
}
func (c nopBroadcaster) SendTo(to *Node, pb proto.Message) error { return nil }
// Broadcast message types.
const (

12
cluster.go
View file

@ -240,7 +240,7 @@ type cluster struct {
state string
Coordinator string
holder *Holder
broadcaster Broadcaster
broadcaster broadcaster
joiningLeavingNodes chan nodeAction
@ -262,8 +262,8 @@ type cluster struct {
InternalClient InternalClient
}
// NewCluster returns a new instance of Cluster with defaults.
func NewCluster() *cluster {
// newCluster returns a new instance of Cluster with defaults.
func newCluster() *cluster {
return &cluster{
Hasher: &jmphasher{},
partitionN: DefaultPartitionN,
@ -708,7 +708,7 @@ func (c *cluster) fragSources(to *cluster, idx *Index) (map[string][]*internal.R
// require that a replica fragment be the source data.
srcCluster := c
if action == resizeJobActionAdd && c.ReplicaN > 1 {
srcCluster = NewCluster()
srcCluster = newCluster()
srcCluster.Nodes = Nodes(c.Nodes).Clone()
srcCluster.Hasher = c.Hasher
srcCluster.partitionN = c.partitionN
@ -1109,7 +1109,7 @@ func (c *cluster) generateResizeJobByAction(nodeAction nodeAction) (*resizeJob,
j.Broadcaster = c.broadcaster
// toCluster is a clone of Cluster with the new node added/removed for comparison.
toCluster := NewCluster()
toCluster := newCluster()
toCluster.Nodes = Nodes(c.Nodes).Clone()
toCluster.Hasher = c.Hasher
toCluster.partitionN = c.partitionN
@ -1310,7 +1310,7 @@ type resizeJob struct {
ID int64
IDs map[string]bool
Instructions []*internal.ResizeInstruction
Broadcaster Broadcaster
Broadcaster broadcaster
action string
result chan string

18
cluster_internal_test.go
View file

@ -43,7 +43,7 @@ func TestFragCombos(t *testing.T) {
node0 := &Node{ID: "node0", URI: *uri0}
node1 := &Node{ID: "node1", URI: *uri1}
c := NewCluster()
c := newCluster()
c.addNodeBasicSorted(node0)
c.addNodeBasicSorted(node1)
@ -120,29 +120,29 @@ func TestFragSources(t *testing.T) {
node2 := &Node{ID: "node2", URI: *uri2}
node3 := &Node{ID: "node3", URI: *uri3}
c1 := NewCluster()
c1 := newCluster()
c1.ReplicaN = 1
c1.addNodeBasicSorted(node0)
c1.addNodeBasicSorted(node1)
c2 := NewCluster()
c2 := newCluster()
c2.ReplicaN = 1
c2.addNodeBasicSorted(node0)
c2.addNodeBasicSorted(node1)
c2.addNodeBasicSorted(node2)
c3 := NewCluster()
c3 := newCluster()
c3.ReplicaN = 2
c3.addNodeBasicSorted(node0)
c3.addNodeBasicSorted(node1)
c4 := NewCluster()
c4 := newCluster()
c4.ReplicaN = 2
c4.addNodeBasicSorted(node0)
c4.addNodeBasicSorted(node1)
c4.addNodeBasicSorted(node2)
c5 := NewCluster()
c5 := newCluster()
c5.ReplicaN = 2
c5.addNodeBasicSorted(node0)
c5.addNodeBasicSorted(node1)
@ -340,7 +340,7 @@ func TestCluster_Owners(t *testing.T) {
// Ensure the partitioner can assign a fragment to a partition.
func TestCluster_Partition(t *testing.T) {
if err := quick.Check(func(index string, shard uint64, partitionN int) bool {
c := NewCluster()
c := newCluster()
c.partitionN = partitionN
partitionID := c.partition(index, shard)
@ -457,10 +457,10 @@ func TestCluster_Coordinator(t *testing.T) {
node1 := &Node{ID: "node1", URI: uri1}
node2 := &Node{ID: "node2", URI: uri2}
c1 := *NewCluster()
c1 := *newCluster()
c1.Node = node1
c1.Coordinator = node1.ID
c2 := *NewCluster()
c2 := *newCluster()
c2.Node = node2
c2.Coordinator = node1.ID

48
executor.go
View file

@ -80,7 +80,7 @@ func newExecutor(opts ...executorOption) *executor {
}
// Execute executes a PQL query.
func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shards []uint64, opt *ExecOptions) ([]interface{}, error) {
func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shards []uint64, opt *execOptions) ([]interface{}, error) {
// Verify that an index is set.
if index == "" {
return nil, ErrIndexRequired
@ -98,7 +98,7 @@ func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shar
// Default options.
if opt == nil {
opt = &ExecOptions{}
opt = &execOptions{}
}
// Translate query keys to ids, if necessary.
@ -123,7 +123,7 @@ func (e *executor) Execute(ctx context.Context, index string, q *pql.Query, shar
return results, nil
}
func (e *executor) execute(ctx context.Context, index string, q *pql.Query, shards []uint64, opt *ExecOptions) ([]interface{}, error) {
func (e *executor) execute(ctx context.Context, index string, q *pql.Query, shards []uint64, opt *execOptions) ([]interface{}, error) {
// Don't bother calculating shards for query types that don't require it.
needsShards := needsShards(q.Calls)
@ -162,7 +162,7 @@ func (e *executor) execute(ctx context.Context, index string, q *pql.Query, shar
}
// executeCall executes a call.
func (e *executor) executeCall(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *ExecOptions) (interface{}, error) {
func (e *executor) executeCall(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (interface{}, error) {
if err := e.validateCallArgs(c); err != nil {
return nil, errors.Wrap(err, "validating args")
}
@ -220,7 +220,7 @@ func (e *executor) validateCallArgs(c *pql.Call) error {
}
// executeSum executes a Sum() call.
func (e *executor) executeSum(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *ExecOptions) (ValCount, error) {
func (e *executor) executeSum(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (ValCount, error) {
if field := c.Args["field"]; field == "" {
return ValCount{}, errors.New("Sum(): field required")
}
@ -253,7 +253,7 @@ func (e *executor) executeSum(ctx context.Context, index string, c *pql.Call, sh
}
// executeMin executes a Min() call.
func (e *executor) executeMin(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *ExecOptions) (ValCount, error) {
func (e *executor) executeMin(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (ValCount, error) {
if field := c.Args["field"]; field == "" {
return ValCount{}, errors.New("Min(): field required")
}
@ -286,7 +286,7 @@ func (e *executor) executeMin(ctx context.Context, index string, c *pql.Call, sh
}
// executeMax executes a Max() call.
func (e *executor) executeMax(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *ExecOptions) (ValCount, error) {
func (e *executor) executeMax(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (ValCount, error) {
if field := c.Args["field"]; field == "" {
return ValCount{}, errors.New("Max(): field required")
}
@ -319,7 +319,7 @@ func (e *executor) executeMax(ctx context.Context, index string, c *pql.Call, sh
}
// executeBitmapCall executes a call that returns a bitmap.
func (e *executor) executeBitmapCall(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *ExecOptions) (*Row, error) {
func (e *executor) executeBitmapCall(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (*Row, error) {
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
return e.executeBitmapCallShard(ctx, index, c, shard)
@ -521,7 +521,7 @@ func (e *executor) executeMaxShard(ctx context.Context, index string, c *pql.Cal
// executeTopN executes a TopN() call.
// This first performs the TopN() to determine the top results and then
// requeries to retrieve the full counts for each of the top results.
func (e *executor) executeTopN(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *ExecOptions) ([]Pair, error) {
func (e *executor) executeTopN(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) ([]Pair, error) {
idsArg, _, err := c.UintSliceArg("ids")
if err != nil {
return nil, fmt.Errorf("executeTopN: %v", err)
@ -560,7 +560,7 @@ func (e *executor) executeTopN(ctx context.Context, index string, c *pql.Call, s
return trimmedList, nil
}
func (e *executor) executeTopNShards(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *ExecOptions) ([]Pair, error) {
func (e *executor) executeTopNShards(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) ([]Pair, error) {
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
return e.executeTopNShard(ctx, index, c, shard)
@ -964,7 +964,7 @@ func (e *executor) executeXorShard(ctx context.Context, index string, c *pql.Cal
}
// executeCount executes a count() call.
func (e *executor) executeCount(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *ExecOptions) (uint64, error) {
func (e *executor) executeCount(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (uint64, error) {
if len(c.Children) == 0 {
return 0, errors.New("Count() requires an input bitmap")
} else if len(c.Children) > 1 {
@ -996,7 +996,7 @@ func (e *executor) executeCount(ctx context.Context, index string, c *pql.Call,
}
// executeClearBit executes a Clear() call.
func (e *executor) executeClearBit(ctx context.Context, index string, c *pql.Call, opt *ExecOptions) (bool, error) {
func (e *executor) executeClearBit(ctx context.Context, index string, c *pql.Call, opt *execOptions) (bool, error) {
fieldName, err := c.FieldArg()
if err != nil {
return false, errors.New("Clear() argument required: field")
@ -1031,7 +1031,7 @@ func (e *executor) executeClearBit(ctx context.Context, index string, c *pql.Cal
}
// executeClearBitField executes a Clear() call for a single view.
func (e *executor) executeClearBitField(ctx context.Context, index string, c *pql.Call, f *Field, colID, rowID uint64, opt *ExecOptions) (bool, error) {
func (e *executor) executeClearBitField(ctx context.Context, index string, c *pql.Call, f *Field, colID, rowID uint64, opt *execOptions) (bool, error) {
shard := colID / ShardWidth
ret := false
for _, node := range e.Cluster.shardNodes(index, shard) {
@ -1061,7 +1061,7 @@ func (e *executor) executeClearBitField(ctx context.Context, index string, c *pq
}
// executeSetBit executes a Set() call.
func (e *executor) executeSetBit(ctx context.Context, index string, c *pql.Call, opt *ExecOptions) (bool, error) {
func (e *executor) executeSetBit(ctx context.Context, index string, c *pql.Call, opt *execOptions) (bool, error) {
fieldName, err := c.FieldArg()
if err != nil {
return false, errors.New("Set() argument required: field")
@ -1106,7 +1106,7 @@ func (e *executor) executeSetBit(ctx context.Context, index string, c *pql.Call,
}
// executeSetBitField executes a Set() call for a specific view.
func (e *executor) executeSetBitField(ctx context.Context, index string, c *pql.Call, f *Field, colID, rowID uint64, timestamp *time.Time, opt *ExecOptions) (bool, error) {
func (e *executor) executeSetBitField(ctx context.Context, index string, c *pql.Call, f *Field, colID, rowID uint64, timestamp *time.Time, opt *execOptions) (bool, error) {
shard := colID / ShardWidth
ret := false
@ -1138,7 +1138,7 @@ func (e *executor) executeSetBitField(ctx context.Context, index string, c *pql.
}
// executeSetValue executes a SetValue() call.
func (e *executor) executeSetValue(ctx context.Context, index string, c *pql.Call, opt *ExecOptions) error {
func (e *executor) executeSetValue(ctx context.Context, index string, c *pql.Call, opt *execOptions) error {
// Parse labels.
columnID, ok, err := c.UintArg(columnLabel)
if err != nil {
@ -1198,7 +1198,7 @@ func (e *executor) executeSetValue(ctx context.Context, index string, c *pql.Cal
}
// executeSetRowAttrs executes a SetRowAttrs() call.
func (e *executor) executeSetRowAttrs(ctx context.Context, index string, c *pql.Call, opt *ExecOptions) error {
func (e *executor) executeSetRowAttrs(ctx context.Context, index string, c *pql.Call, opt *execOptions) error {
fieldName, ok := c.Args["_field"].(string)
if !ok {
return errors.New("SetRowAttrs() field required")
@ -1255,7 +1255,7 @@ func (e *executor) executeSetRowAttrs(ctx context.Context, index string, c *pql.
}
// executeBulkSetRowAttrs executes a set of SetRowAttrs() calls.
func (e *executor) executeBulkSetRowAttrs(ctx context.Context, index string, calls []*pql.Call, opt *ExecOptions) ([]interface{}, error) {
func (e *executor) executeBulkSetRowAttrs(ctx context.Context, index string, calls []*pql.Call, opt *execOptions) ([]interface{}, error) {
// Collect attributes by field/id.
m := make(map[string]map[uint64]map[string]interface{})
for _, c := range calls {
@ -1342,7 +1342,7 @@ func (e *executor) executeBulkSetRowAttrs(ctx context.Context, index string, cal
}
// executeSetColumnAttrs executes a SetColumnAttrs() call.
func (e *executor) executeSetColumnAttrs(ctx context.Context, index string, c *pql.Call, opt *ExecOptions) error {
func (e *executor) executeSetColumnAttrs(ctx context.Context, index string, c *pql.Call, opt *execOptions) error {
// Retrieve index.
idx := e.Holder.Index(index)
if idx == nil {
@ -1390,7 +1390,7 @@ func (e *executor) executeSetColumnAttrs(ctx context.Context, index string, c *p
}
// exec executes a PQL query remotely for a set of shards on a node.
func (e *executor) remoteExec(ctx context.Context, node *Node, index string, q *pql.Query, shards []uint64, opt *ExecOptions) (results []interface{}, err error) {
func (e *executor) remoteExec(ctx context.Context, node *Node, index string, q *pql.Query, shards []uint64, opt *execOptions) (results []interface{}, err error) {
// Encode request object.
pbreq := &internal.QueryRequest{
Query: q.String(),
@ -1461,7 +1461,7 @@ loop:
//
// If a mapping of shards to a node fails then the shards are resplit across
// secondary nodes and retried. This continues to occur until all nodes are exhausted.
func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64, c *pql.Call, opt *ExecOptions, mapFn mapFunc, reduceFn reduceFunc) (interface{}, error) {
func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64, c *pql.Call, opt *execOptions, mapFn mapFunc, reduceFn reduceFunc) (interface{}, error) {
ch := make(chan mapResponse)
// Wrap context with a cancel to kill goroutines on exit.
@ -1520,7 +1520,7 @@ func (e *executor) mapReduce(ctx context.Context, index string, shards []uint64,
}
}
func (e *executor) mapper(ctx context.Context, ch chan mapResponse, nodes []*Node, index string, shards []uint64, c *pql.Call, opt *ExecOptions, mapFn mapFunc, reduceFn reduceFunc) error {
func (e *executor) mapper(ctx context.Context, ch chan mapResponse, nodes []*Node, index string, shards []uint64, c *pql.Call, opt *execOptions, mapFn mapFunc, reduceFn reduceFunc) error {
// Group shards together by nodes.
m, err := e.shardsByNode(nodes, index, shards)
if err != nil {
@ -1710,8 +1710,8 @@ type mapResponse struct {
err error
}
// ExecOptions represents an execution context for a single Execute() call.
type ExecOptions struct {
// execOptions represents an execution context for a single Execute() call.
type execOptions struct {
Remote bool
ExcludeRowAttrs bool
ExcludeColumns bool

4
field.go
View file

@ -64,7 +64,7 @@ type Field struct {
// Row attribute storage and cache
rowAttrStore AttrStore
broadcaster Broadcaster
broadcaster broadcaster
Stats StatsClient
// Field options.
@ -133,7 +133,7 @@ func NewField(path, index, name string, options FieldOptions) (*Field, error) {
views: make(map[string]*View),
rowAttrStore: NopAttrStore,
rowAttrStore: nopStore,
broadcaster: NopBroadcaster,
Stats: NopStatsClient,

4
holder.go
View file

@ -50,7 +50,7 @@ type Holder struct {
// opened channel is closed once Open() completes.
opened chan struct{}
Broadcaster Broadcaster
Broadcaster broadcaster
NewAttrStore func(string) AttrStore
@ -81,7 +81,7 @@ func NewHolder() *Holder {
Broadcaster: NopBroadcaster,
Stats: NopStatsClient,
NewAttrStore: NewNopAttrStore,
NewAttrStore: newNopAttrStore,
CacheFlushInterval: defaultCacheFlushInterval,

6
index.go
View file

@ -46,7 +46,7 @@ type Index struct {
// Column attribute storage and cache.
columnAttrStore AttrStore
broadcaster Broadcaster
broadcaster broadcaster
Stats StatsClient
Logger Logger
@ -66,8 +66,8 @@ func NewIndex(path, name string) (*Index, error) {
remoteMaxShard: 0,
NewAttrStore: NewNopAttrStore,
columnAttrStore: NopAttrStore,
NewAttrStore: newNopAttrStore,
columnAttrStore: nopStore,
broadcaster: NopBroadcaster,
Stats: NopStatsClient,

4
server.go
View file

@ -40,7 +40,7 @@ const (
)
// Ensure Server implements interfaces.
var _ Broadcaster = &Server{}
var _ broadcaster = &Server{}
var _ MemberServer = &Server{}
// Server represents a holder wrapped by a running HTTP server.
@ -228,7 +228,7 @@ func OptServerClusterHasher(h Hasher) ServerOption {
func NewServer(opts ...ServerOption) (*Server, error) {
s := &Server{
closing: make(chan struct{}),
cluster: NewCluster(),
cluster: newCluster(),
holder: NewHolder(),
diagnostics: NewDiagnosticsCollector(DefaultDiagnosticServer),
systemInfo: NewNopSystemInfo(),

32
translate.go
View file

@ -24,11 +24,7 @@ const (
)
const (
DefaultReplicationRetryInterval = 1 * time.Second
)
const (
ReplicationBufferSize = 65536
defaultReplicationRetryInterval = 1 * time.Second
)
var (
@ -90,7 +86,7 @@ func NewTranslateFile() *TranslateFile {
MapSize: DefaultMapSize,
ReplicationRetryInterval: DefaultReplicationRetryInterval,
ReplicationRetryInterval: defaultReplicationRetryInterval,
}
}
@ -216,7 +212,7 @@ func (s *TranslateFile) applyEntry(entry *LogEntry, offset int64) error {
key := entry.Keys[i]
// Determine key offset based on ID size.
sz := int64(UvarintSize(id))
sz := int64(uVarintSize(id))
idx.insert(id, offset+sz)
// Move sequence forward.
@ -225,7 +221,7 @@ func (s *TranslateFile) applyEntry(entry *LogEntry, offset int64) error {
}
// Move offset forward.
offset += sz + int64(UvarintSize(uint64(len(key)))) + int64(len(key))
offset += sz + int64(uVarintSize(uint64(len(key)))) + int64(len(key))
}
return nil
@ -564,11 +560,11 @@ type LogEntry struct {
// HeaderSize returns the number of bytes required for size, type, index, frame, & pair count.
func (e *LogEntry) HeaderSize() int64 {
sz := UvarintSize(e.Length) + // total entry length
sz := uVarintSize(e.Length) + // total entry length
1 + // type
UvarintSize(uint64(len(e.Index))) + len(e.Index) + // Index length and data
UvarintSize(uint64(len(e.Frame))) + len(e.Frame) + // Frame length and data
UvarintSize(uint64(len(e.IDs))) // ID/Key pair count
uVarintSize(uint64(len(e.Index))) + len(e.Index) + // Index length and data
uVarintSize(uint64(len(e.Frame))) + len(e.Frame) + // Frame length and data
uVarintSize(uint64(len(e.IDs))) // ID/Key pair count
return int64(sz)
}
@ -578,13 +574,13 @@ func (e *LogEntry) ReadFrom(r io.Reader) (_ int64, err error) {
// Read the entry length.
if e.Length, err = binary.ReadUvarint(br); err != nil {
return int64(UvarintSize(e.Length)), err
return int64(uVarintSize(e.Length)), err
}
// Slurp entire entry and replace reader.
buf := make([]byte, e.Length)
n, err := io.ReadFull(r, buf)
n64 := int64(n + UvarintSize(e.Length))
n64 := int64(n + uVarintSize(e.Length))
if err != nil {
return n64, err
}
@ -710,8 +706,8 @@ func (e *LogEntry) WriteTo(w io.Writer) (_ int64, err error) {
return int64(sz) + n, err
}
// ValidLogEntriesLen returns the maximum length of p that contains valid entries.
func ValidLogEntriesLen(p []byte) (n int) {
// validLogEntriesLen returns the maximum length of p that contains valid entries.
func validLogEntriesLen(p []byte) (n int) {
r := bytes.NewReader(p)
for {
if sz, err := binary.ReadUvarint(r); err != nil {
@ -988,13 +984,13 @@ func (r *TranslateFileReader) read(p []byte) (n int, err error) {
// Read data from file at offset.
// Limit the number of bytes read to only whole entries.
n, err = r.file.ReadAt(p, r.offset)
n = ValidLogEntriesLen(p[:n])
n = validLogEntriesLen(p[:n])
r.offset += int64(n)
return n, err
}
// Copied & modified from encoding/binary.
func UvarintSize(x uint64) (i int) {
func uVarintSize(x uint64) (i int) {
for x >= 0x80 {
x >>= 7
i++

4
utils_internal_test.go
View file

@ -34,7 +34,7 @@ func NewTestCluster(n int) *cluster {
panic(err)
}
c := NewCluster()
c := newCluster()
c.ReplicaN = 1
c.Hasher = NewTestModHasher()
c.Path = path
@ -222,7 +222,7 @@ func (t *ClusterCluster) addCluster(i int, saveTopology bool) (*cluster, error)
h.Path = path
// cluster
c := NewCluster()
c := newCluster()
c.ReplicaN = 1
c.Hasher = NewTestModHasher()
c.Path = path

2
view.go
View file

@ -52,7 +52,7 @@ type View struct {
// prevent sending multiple `CreateShardMessage` messages
maxShard uint64
broadcaster Broadcaster
broadcaster broadcaster
stats StatsClient
RowAttrStore AttrStore
Logger Logger