sync2

Documentation

Overview

Package sync2 provides extra functionality along the same lines as sync.

Index

• type Batcher
  • func NewBatcher(interval time.Duration) *Batcher
  • func (b *Batcher) Wait() int
• type Consolidator
  • func NewConsolidator() *Consolidator
  • func (co *Consolidator) Create(query string) (r *Result, created bool)
• type ConsolidatorCache
  • func NewConsolidatorCache(capacity int64) *ConsolidatorCache
  • func (cc *ConsolidatorCache) Items() []ConsolidatorCacheItem
  • func (cc *ConsolidatorCache) Record(query string)
• type ConsolidatorCacheItem
• type Result
  • func (rs *Result) Broadcast()
  • func (rs *Result) Wait()
• type Semaphore
  • func NewSemaphore(count int, timeout time.Duration) *Semaphore
  • func (sem *Semaphore) Acquire() bool
  • func (sem *Semaphore) Release()
  • func (sem *Semaphore) Size() int
  • func (sem *Semaphore) TryAcquire() bool

Types

type Batcher

type Batcher struct {
	// contains filtered or unexported fields
}

Batcher delays concurrent operations for a configurable interval in order to batch them up or otherwise clock their operation to run concurrently.

It is implemented as a channel of int32s. Each waiter blocks on the channel from which it gets a sequentially increasing batch ID when the timer elapses.

Hence a waiter is delayed for at most the batch interval.

func NewBatcher

func NewBatcher(interval time.Duration) *Batcher

NewBatcher returns a new Batcher

func (*Batcher) Wait

func (b *Batcher) Wait() int

Wait adds a new waiter to the queue and blocks until the next batch

type Consolidator

type Consolidator struct {
	*ConsolidatorCache
	// contains filtered or unexported fields
}

Consolidator consolidates duplicate queries from executing simulaneously and shares results between them.

func NewConsolidator

func NewConsolidator() *Consolidator

NewConsolidator creates a new Consolidator

func (*Consolidator) Create

func (co *Consolidator) Create(query string) (r *Result, created bool)

Create adds a query to currently executing queries and acquires a lock on its Result if it is not already present. If the query is a duplicate, Create returns false.

type ConsolidatorCache

type ConsolidatorCache struct {
	*cache.LRUCache
}

ConsolidatorCache is a thread-safe object used for counting how often recent queries have been consolidated. It is also used by the txserializer package to count how often transactions have been queued and had to wait because they targeted the same row (range).

func NewConsolidatorCache

func NewConsolidatorCache(capacity int64) *ConsolidatorCache

NewConsolidatorCache creates a new cache with the given capacity.

func (*ConsolidatorCache) Items

func (cc *ConsolidatorCache) Items() []ConsolidatorCacheItem

Items returns the items in the cache as an array of String, int64 structs

func (*ConsolidatorCache) Record

func (cc *ConsolidatorCache) Record(query string)

Record increments the count for "query" by 1. If it's not in the cache yet, it will be added.

type ConsolidatorCacheItem

type ConsolidatorCacheItem struct {
	Query string
	Count int64
}

ConsolidatorCacheItem is a wrapper for the items in the consolidator cache

type Result

type Result struct {
	Result interface{}
	Err    error
	// contains filtered or unexported fields
}

Result is a wrapper for result of a query.

func (*Result) Broadcast

func (rs *Result) Broadcast()

Broadcast removes the entry from current queries and releases the lock on its Result. Broadcast should be invoked when original query completes execution.

func (*Result) Wait

func (rs *Result) Wait()

Wait waits for the original query to complete execution. Wait should be invoked for duplicate queries.

type Semaphore

type Semaphore struct {
	// contains filtered or unexported fields
}

Semaphore is a counting semaphore with the option to specify a timeout.

func NewSemaphore

func NewSemaphore(count int, timeout time.Duration) *Semaphore

NewSemaphore creates a Semaphore. The count parameter must be a positive number. A timeout of zero means that there is no timeout.

func (*Semaphore) Acquire

func (sem *Semaphore) Acquire() bool

Acquire returns true on successful acquisition, and false on a timeout.

func (*Semaphore) Release

func (sem *Semaphore) Release()

Release releases the acquired semaphore. You must not release more than the number of semaphores you've acquired.

func (*Semaphore) Size

func (sem *Semaphore) Size() int

Size returns the current number of available slots.

func (*Semaphore) TryAcquire

func (sem *Semaphore) TryAcquire() bool

TryAcquire acquires a semaphore if it's immediately available. It returns false otherwise.