metastore

README

MetaStore

Store is an abstraction over a string map that supports get, set, delete, publish, and subscribe.

MetaStore is an abstraction over a Store that breaks the key space into buckets. By doing so, we get finer lock granularity when deployed in a concurrent environment, such as in Jack.

Store

A Store is, as mentioned, a wrapper over a string map. It provides safe access and tracks subscribers. Here is an overview:

Create a Store:

var s metastore.Store
s.Init()

Set a value to a key:

s.Set(key, value)

Get a value at a key:

value, ok := s.Get(key)

Delete a key:

s.Delete(key)

Delete everything (including subscriptions):

s.FlushAll()

Publish a value to a key:

s.Publish(key, value)

In terms of state change, Publish is equivalent to Set. However, Publish also updates subscribers with the new value, while Set does not.

Subscribe to changes on a key:

recv := make(chan string)
s.Subscribe(key, recv)
for {
    fmt.Println(<-recv)
}

As it stands, each subscription requires a unique channel for the receiver to know which key an update is associated with. This can introduce some overhead, but the alternative would be sending the key as well and parsing... I don't like parsing. Besides, if you really want to conserve resources, you can add an identifier to the value and do it yourself.

To unsubscribe, you can close the channel. Store will drop the channel from the subscriber list.

For finer control, call Unsubscribe:

S.Unsubscribe(key, recv)

Want to know how many are subscriptions a key has?

nInt := S.NumSubscribers(key)
fmt.Printf("%s has %d subscribers\n", key, nInt)

Check out this many-to-many pub/sub pattern:

var S ms.Store
S.Init()
	
// create five publishers on the same key
for pid := 0; pid < 5; pid++ {
    go func(pid int) {
        for i := 0; ; i++ {
            time.Sleep(time.Second)
				val := fmt.Sprintf("%d_%d", pid, i)
				S.Publish("k123", val)
			}
		}(pid)
	}

c1 := make(chan string)
c2 := make(chan string)
c3 := make(chan string)
	
S.Subscribe("k123", c1)
S.Subscribe("k123", c2)
S.Subscribe("k123", c3)
var v string
for {
	select {
    case v = <-c1:
    	fmt.Println("channel 1 got value:", v)
    case v = <-c2:
    	fmt.Println("channel 2 got value:", v)
    case v = <-c3:
    	fmt.Println("channel 3 got value:", v)
    }
}

MetaStore

MetaStore splits the key space into buckets and provides a means to determine which bucket a key-value pair resides in.

Create a MetaStore and split the key space over 1000 buckets:

var m metastore.MetaStore
m.Init(1000)

Get a hash function to determine which bucket a key goes in:

h := m.GetHasher()
bucketId := h([]byte("the key"))

Set a value to a key:

bucketId := h([]byte(key))
m.Bucket[bucketId].Set(key, value)

MetaStore also implements all of Store's public methods, so you don't need to go to the trouble of calculating the bucket index. In other words, this is valid code, too:

m.Set(key, value)
value, ok := m.Get(key)
m.Delete(key)
m.Publish(key, value)
m.Subscribe(key, recv)
m.Unsubscribe(key, recv)
nInt := m.NumSubscribers(key)
m.FlushAll()

Documentation

Index

• type MetaStore
  • func (m *MetaStore) Delete(key string)
  • func (m *MetaStore) FlushAll()
  • func (m *MetaStore) Get(key string) (string, bool)
  • func (m *MetaStore) GetHasher() func([]byte) uint
  • func (m *MetaStore) Init(n uint)
  • func (m *MetaStore) NumSubscribers(key string) int
  • func (m *MetaStore) Publish(key, value string)
  • func (m *MetaStore) Set(key, value string)
  • func (m *MetaStore) Subscribe(key string, outgoing chan<- string)
  • func (m *MetaStore) Unsubscribe(key string, outgoing chan<- string)
• type Store
  • func (s *Store) Delete(key string)
  • func (s *Store) FlushAll()
  • func (s *Store) Get(key string) (value string, ok bool)
  • func (s *Store) Init()
  • func (s *Store) NumSubscribers(key string) int
  • func (s *Store) Publish(key, value string)
  • func (s *Store) Set(key, value string)
  • func (s *Store) Subscribe(key string, outgoing chan<- string)
  • func (s *Store) Unsubscribe(key string, outgoing chan<- string)

Types

type MetaStore

type MetaStore struct {
	Bucket []Store
	sync.RWMutex
	// contains filtered or unexported fields
}

func (*MetaStore) Delete

func (m *MetaStore) Delete(key string)

func (*MetaStore) FlushAll

func (m *MetaStore) FlushAll()

func (*MetaStore) Get

func (m *MetaStore) Get(key string) (string, bool)

func (*MetaStore) GetHasher

func (m *MetaStore) GetHasher() func([]byte) uint

func (*MetaStore) Init

func (m *MetaStore) Init(n uint)

func (*MetaStore) NumSubscribers

func (m *MetaStore) NumSubscribers(key string) int

func (*MetaStore) Publish

func (m *MetaStore) Publish(key, value string)

func (*MetaStore) Set

func (m *MetaStore) Set(key, value string)

func (*MetaStore) Subscribe

func (m *MetaStore) Subscribe(key string, outgoing chan<- string)

func (*MetaStore) Unsubscribe

func (m *MetaStore) Unsubscribe(key string, outgoing chan<- string)

type Store

type Store struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

Store is the base storage structure.

func (*Store) Delete

func (s *Store) Delete(key string)

Delete removes a key from storage.

func (*Store) FlushAll

func (s *Store) FlushAll()

FlushAll deletes all keys and subscribers

func (*Store) Get

func (s *Store) Get(key string) (value string, ok bool)

Get returns the value associated with a key; also returns a boolean indicating success or failure.

func (*Store) Init

func (s *Store) Init()

Init initializes the Store.

func (*Store) NumSubscribers

func (s *Store) NumSubscribers(key string) int

NumSubscribers returns the number of subscribers on a key

func (*Store) Publish

func (s *Store) Publish(key, value string)

Publish associates a key with a value and updates subscribers.

func (*Store) Set

func (s *Store) Set(key, value string)

Set associates a key with a value.

func (*Store) Subscribe

func (s *Store) Subscribe(key string, outgoing chan<- string)

Subscribe associates an alert on an outgoing channel with a key.

func (*Store) Unsubscribe

func (s *Store) Unsubscribe(key string, outgoing chan<- string)

Unsubscribe removes a channel from a subscriber list