Discussion at https://golang.org/issue/18802
As it stands, Go programs do not have a good way to avoid contention when combining highly concurrent code with shared memory locations that frequently require mutation. This proposal describes a new package and type to satisfy this need.
There are several scenarios in which a Go program might want to have shared data that is mutated frequently. We will briefly discuss some of these scenarios, such that we can evaluate the proposed API against these concrete use-cases.
In RPC servers or in scientific computing code, there is often a need for global
counters. For instance, in an RPC server, a global counter might count the
number of requests received by the server, or the number of bytes received by
the server. Go makes it easy to write RPC servers which are inherently
concurrent, often processing each connection and each request on a concurrent
goroutine. This means that in the context of a multicore machine, several
goroutines can be incrementing the same global counter in parallel. Using an API
like atomic.AddInt64 will ensure that such a counter is lock-free, but
parallel goroutines will be contending for the same cache line, so this counter
will not experience linear scalability as the number of cores is increased.
(Indeed, one might even expect scalability to unexpectedly decrease due to
increased core-to-core communication).
It's probably also worth noting that there are other similar use-cases in this space (e.g. types that record distributions rather than just sums, max-value trackers, etc).
It is common in programs to have data that is read frequently, but written very
rarely. In these cases, a common synchronization primitive is sync.RWMutex,
which offers an RLock/RUnlock API for readers. When no writers are
interacting with an RWMutex, an arbitrary number of goroutines can use the
read-side of the RWMutex without blocking.
However, in order to correctly pair calls to RLock with calls to RUnlock,
RWMutex internally maintains a counter, which is incremented during RLock
and decremented during RUnlock. There is also other shared mutable state that
is atomically updated inside the RWMutex during these calls (and during
calls to Lock and Unlock). For reasons similar to the previous example, if
many goroutines are acquiring and releasing read-locks concurrently and the
program is running on a multicore machine, then it is likely that the
performance of RWMutex.RLock/RWMutex.RUnlock will not scale linearly with
the number of cores given to the program.
In programs that make many RPC calls in parallel, there can be contention on shared mutable state stored inside the RPC or HTTP clients. For instance, an RPC client might support connecting to a pool of servers, and implements a configurable load balancing policy to select a connection to use for a given RPC; these load balancing policies often need to maintain state for each connection in the pool of managed connections. For instance, an implementation of the "Least-Loaded" policy needs to maintain a counter of active requests per connection, such that a new request can select the connection with the least number of active requests. In scenarios where a client is performing a large number of requests in parallel (perhaps enqueueing many RPCs before finally waiting on them at a later point in the program), then contention on this internal state can start to affect the rate at which the requests can be dispatched.
In data-processing pipelines, code running in a particular stage may want to
'batch' its output, such that it only sends data downstream in N-element
batches, rather than sending single elements through the pipeline at a time. In
the single-goroutine case and where the element type is 'byte', then the
familiar type bufio.Writer implements this pattern. Indeed, one
option for the general data-processing pipeline, is to have a single goroutine
run every stage in the pipeline from end-to-end, and then instantiate a small
number of parallel pipeline instances. This strategy effectively handles
pipelines composed solely of stages dominated by CPU-time. However, if a
pipeline contains any IO (e.g. initially reading the input from a distributed
file system, making RPCs, or writing the result back to a distributed file
system), then this setup will not be efficient, as a single stall in IO will
take out a significant chunk of your throughput.
To mitigate this problem, IO bound stages need to run many goroutines. Indeed, a clever framework (like Apache Beam) can detect these sorts of situations dynamically, by measuring the rate of stage input compared to the rate of stage output; they can even reactively increase or decrease the "concurrency level" of a stage in response to these measurements. In Beam's case, it might do this by dynamically changing the number of threads-per-binary, or number of workers-per-stage.
When stages have varying concurrency levels, but are connected to each other in a pipeline structure, it is important to place a concurrency-safe abstraction between the stages to buffer elements waiting to be processed by the next stage. Ideally, this structure would minimize the contention experienced by the caller.
To solve these problems, we propose an API with a single new type
percpu.Sharded. Here is an outline of the proposed API.
// Package percpu provides low-level utilities for avoiding contention on
// multicore machines.
package percpu
// A Sharded is a container of homogenously typed values.
//
// On a best effort basis, the runtime will strongly associate a given value
// with a CPU core. That is to say, retrieving a value twice on the same CPU
// core will return the same value with high probablity. Note that the runtime
// cannot guarantee this fact, and clients must assume that retrieved values
// can be shared between concurrently executing goroutines.
//
// Once a value is placed in a Sharded, the Sharded will retain a reference to
// this value permanently. Clients can control the maximum number of distinct
// values created using the SetMaxShards API.
//
// A Sharded must not be copied after first use.
//
// All methods are safe to call from multiple goroutines.
type Sharded struct {
// contains unexported fields
}
// SetMaxShards sets a limit on the maximum number of elements stored in the
// Sharded.
//
// It will not apply retroactively, any elements already created will remain
// inside the Sharded.
//
// If maxShards is less than 1, Sharded will panic.
func (s *Sharded) SetMaxShards(maxShards int)
// GetOrCreate retrieves a value roughly associated with the current CPU. If
// there is no such value, then createFn is called to create a value, store it
// in the Sharded, and return it.
//
// All calls to createFn are serialized; this means that one must complete
// before the next one is started.
//
// createFn should not return nil, or Sharded will panic.
//
// If createFn is called with a ShardInfo.ShardIndex equal to X, no future call
// to GetOrCreate will call createFn again with a ShardInfo.ShardIndex equal to
// X.
func (s *Sharded) GetOrCreate(createFn func(ShardInfo) interface{}) interface{}
// Get retrieves any preexisting value associated with the current CPU. If
// there is no such value, nil is returned.
func (s *Sharded) Get() interface{}
// Do iterates over a snapshot of all elements stored in the Sharded, and calls
// fn once for each element.
//
// If more elements are created during the iteration itself, they may be
// visible to the iteration, but this is not guaranteed. For stronger
// guarantees, see DoLocked.
func (s *Sharded) Do(fn func(interface{}))
// DoLocked iterates over all the elements stored in the Sharded, and calls fn
// once for each element.
//
// DoLocked will observe a consistent snapshot of the elements in the Sharded;
// any previous creations will complete before the iteration begins, and all
// subsequent creations will wait until the iteration ends.
func (s *Sharded) DoLocked(fn func(interface{}))
// ShardInfo contains information about a CPU core.
type ShardInfo struct {
// ShardIndex is strictly less than any call to any prior call to SetMaxShards.
ShardIndex int
}
Here, we evaluate the proposed API in light of the use-cases described above.
A counter API can be fairly easily built on top of percpu.Sharded.
Specifically, it would offer two methods IncrementBy(int64), and Sum() int64.
The former would only allow positive increments (if required, clients can build
negative increments by composing two counters of additions and subtractions).
The implementation of IncrementBy, would call GetOrCreate, passing a
function that returned an *int64. To avoid false sharing between cache lines,
it would probably return it as an interior pointer into a struct with
appropriate padding. Once the pointer is retrieved from GetOrCreate, the
function would then use atomic.AddInt64 on that pointer with the value passed
to IncrementBy.
The implementation of Sum would call Do to retrieve a snapshot of all
previously created values, then sum up their values using atomic.LoadInt64.
If the application is managing many long-lived counters, then one possible
optimization would be to implement the Counter type in terms of a
counterBatch (which logically encapsulates N independent counters). This can
drastically limit the padding required to fix false sharing between cache lines.
It is a little tricky to implement a drop-in replacement for sync.RWMutex on
top of percpu.Sharded. Naively, one could imagine a sharded lock composed of
many internal sync.RWMutex instances. Calling RLock() on the aggregate lock
would grab a single sync.RWMutex instance using GetOrCreate and then call
RLock() on that instance. Unfortunately, because there is no state passed
between RLock() and RUnlock() (something we should probably consider fixing
for Go 2), we cannot implement RUnlock() efficiently, as the percpu.Sharded
might have migrated to a different shard and therefore we've lost the
association to the original RLock().
That said, since such a sharded lock would be considerably more memory-hungry
than a normal sync.RWMutex, callers should only replace truly contended
mutexes with a sharded lock, so requiring them to make minor API changes should
not be too onerous (particularly for mutexes, which should always be private
implementation details, and therefore not cross API boundaries). In particular,
one could have RLock() on the sharded lock return a RLockHandle type, which
has a RUnlock() method. This RLockHandle could keep an internal pointer to
the sync.RWMutex that was initially chosen, and it can then RUnlock() that
specific instance.
It's worth noting that it's also possible to drastically change the standard
library's sync.RWMutex implementation itself to be scalable by default using
percpu.Sharded; this is why the implementation sketch below is careful not not
use the sync package to avoid circular dependencies. See Facebook's
SharedMutex
class to get a sense of how this could be done. However, that requires
significant research and deserves a proposal of its own.
It's straightforward to use percpu.Sharded to implement a sharded RPC client.
This is a case where its likely that the default implementation will continue to
be unsharded, and a program will need to explicitly say something like
grpc.Dial("some-server", grpc.ShardedClient(4)) (where the "4" might come from
an application flag). This kind of client-contrallable sharding is one place
where the SetMaxShards API can be useful.
This can be implemented using percpu.Sharded. For instance, a writer would
call GetOrCreate to retrieve a shard-local buffer, they would acquire a lock,
and insert the element into the buffer. If the buffer became full, they would
flush it downstream.
A watchdog goroutine could walk the buffers periodically using AppendAll, and
flush partially-full buffers to ensure that elements are flushed fairly
promptly. If it finds no elements to flush, it could start incrementing a
counter of "useless" scans, and stop scanning after it reaches a threshold. If a
writer is enqueuing the first element in a buffer, and it sees the counter over
the threshold, it could reset the counter, and wake the watchdog.
What follows is a rough sketch of an implementation of percpu.Sharded. This is
to show that this is implementable, and to give some context to the discussion
of performance below.
First, a sketch of an implementation for percpu.sharder, an internal helper
type for percpu.Sharded.
const (
defaultUserDefinedMaxShards = 32
)
type sharder struct {
maxShards int32
}
func (s *sharder) SetMaxShards(maxShards int) {
if maxShards < 1 {
panic("maxShards < 1")
}
atomic.StoreInt32(&s.maxShards, roundDownToPowerOf2(int32(maxShards)))
}
func (s *sharder) userDefinedMaxShards() int32 {
s := atomic.LoadInt32(&s.maxShards)
if s == 0 {
return defaultUserDefinedMaxShards
}
return s
}
func (s *sharder) shardInfo() ShardInfo {
shardId := runtime_getShardIndex()
// If we're in race mode, then all bets are off. Half the time, randomize the
// shardId completely, the rest of the time, use shardId 0.
//
// If we're in a test but not in race mode, then we want an implementation
// that keeps cache contention to a minimum so benchmarks work properly, but
// we still want to flush out any assumption of a stable mapping to shardId.
// So half the time, we double the id. This catches fewer problems than what
// we get in race mode, but it should still catch one class of issue (clients
// assuming that two sequential calls to Get() will return the same value).
if raceEnabled {
rnd := runtime_fastRand()
if rnd%2 == 0 {
shardId = 0
} else {
shardId += rnd / 2
}
} else if testing {
if runtime_fastRand()%2 == 0 {
shardId *= 2
}
}
shardId &= runtimeDefinedMaxShards()-1
shardId &= userDefinedMaxShards()-1
return ShardInfo{ShardIndex: shardId}
}
func runtimeDefinedMaxShards() int32 {
max := runtime_getMaxShards()
if (testing || raceEnabled) && max < 4 {
max = 4
}
return max
}
// Implemented in the runtime, should effectively be
// roundUpToPowerOf2(min(GOMAXPROCS, NumCPU)).
// (maybe caching that periodically in the P).
func runtime_getMaxShards() int32 {
return 4
}
// Implemented in the runtime, should effectively be the result of the getcpu
// syscall, or similar. The returned index should densified if possible (i.e.
// if binary is locked to cores 2 and 4), they should return 0 and 1
// respectively, not 2 and 4.
//
// Densification can be best-effort, and done with a process-wide mapping table
// maintained by sysmon periodically.
//
// Does not have to be bounded by runtime_getMaxShards(), or indeed by
// anything.
func runtime_getShardIndex() int32 {
return 0
}
// Implemented in the runtime. Only technically needs an implementation for
// raceEnabled and tests. Should be scalable (e.g. using a per-P seed and
// state).
func runtime_fastRand() int32 {
return 0
}
Next, a sketch of percpu.Sharded itself.
type Sharded struct {
sharder
lock uintptr
data atomic.Value // *shardedData
typ unsafe.Pointer
}
func (s *Sharded) loadData() *shardedData {
return s.data.Load().(*shardedData)
}
func (s *Sharded) getFastPath(createFn func(ShardInfo) interface{}) (out interface{}) {
shardInfo := s.shardInfo()
curData := s.loadData()
if curData == nil || shardInfo.ShardIndex >= len(curData.elems) {
if createFn == nil {
return nil
}
return s.getSlowPath(shardInfo, createFn)
}
existing := curData.load(shardInfo.ShardIndex)
if existing == nil {
if createFn == nil {
return nil
}
return s.getSlowPath(shardInfo, createFn)
}
outp := (*ifaceWords)(unsafe.Pointer(&out))
outp.typ = s.typ
outp.data = existing
return
}
func (s *Sharded) getSlowPath(shardInfo ShardInfo, createFn func(ShardInfo) interface{}) (out interface{}) {
runtime_lock(&s.lock)
defer runtime_unlock(&s.lock)
curData := s.loadData()
if curData == nil || shardInfo.ShardIndex >= len(curData.elems) {
curData = allocShardedData(curData, shardInfo)
s.data.Store(curData)
}
existing := curData.load(shardInfo.ShardIndex)
if existing != nil {
outp := (*ifaceWords)(unsafe.Pointer(&out))
outp.typ = s.typ
outp.data = existing
return
}
newElem := createFn(shardInfo)
if newElem == nil {
panic("createFn returned nil value")
}
newElemP := *(*ifaceWords)(unsafe.Pointer(&newElem))
// If this is the first call to createFn, then stash the type-pointer for
// later verification.
//
// Otherwise, verify its the same as the previous.
if s.typ == nil {
s.typ = newElemP.typ
} else if s.typ != newElemP.typ {
panic("percpu: GetOrCreate was called with function that returned inconsistently typed value")
}
// Store back the new value.
curData.store(shardInfo.ShardIndex, newElemP.val)
// Return it.
outp := (*ifaceWords)(unsafe.Pointer(&out))
outp.typ = s.typ
outp.data = newElemP.val
}
func (s *Sharded) loadData() *shardedData {
return s.data.Load().(*shardedData)
}
func (s *Sharded) GetOrCreate(createFn func(ShardInfo) interface{}) interface{} {
if createFn == nil {
panic("createFn nil")
}
return s.getFastPath(createFn)
}
func (s *Sharded) Get() interface{} {
return s.getFastPath(nil)
}
func (s *Sharded) Do(fn func(interface{})) {
curData := s.loadData()
if curData == nil {
return nil
}
for i := range curData.elems {
elem := curData.load(i)
if elem == nil {
continue
}
var next interface{}
nextP := (*ifaceWords)(unsafe.Pointer(&next))
nextP.typ = s.typ
nextP.val = elem
fn(next)
}
return elems
}
func (s *Sharded) DoLocked(fn func(interface{})) {
runtime_lock(&s.lock)
defer runtime_unlock(&s.lock)
s.Do(fn)
}
type shardedData struct {
elems []unsafe.Pointer
}
percpu.sharderAs presented, calling shardInfo on a percpu.sharder makes two calls to the
runtime, and does a single atomic load.
However, both of the calls to the runtime would be satisfied with stale values.
So, an obvious avenue of optimization is to squeeze these two pieces of
information (effectively "current shard", and "max shards") into a single word,
and cache it on the P when first calling the shardInfo API. To accommodate
changes in the underlying values, a P can store a timestamp when it last
computed these values, and clear the cache when the value is older than X and
the P is in the process of switching goroutines.
This means that effectively, shardInfo will consist of 2 atomic loads, and a
little bit of math on the resulting values.
percpu.ShardedIn the get-for-current-shard path, percpu.Sharded will call shardInfo, and
then perform 2 atomic loads (to retrieve the list of elements and to retrieve
the specific element for the current shard, respectively). If either of these
loads fails, it might fall back to a much-slower slow path. In the fast path,
there's no allocation.
In the get-all path, percpu.Sharded will perform a single atomic followed by a
O(n) atomic loads, proportional to the number of elements stored in the
percpu.Sharded. It will not allocate.
percpu.ShardedWith the given API, if GOMAXPROCS is temporarily increased (or the CPUs
assigned to the given program), and then decreased to its original value, a
percpu.Sharded might have allocated additional elements to satisfy the
additional CPUs. These additional elements would not be eligible for garbage
collection, as the percpu.Sharded would retain an internal reference.
First, its worth noting that we cannot unilaterally shrink the number of
elements stored in a percpu.Sharded, because this might affect program
correctness. For instance, this could result in counters losing values, or in
breaking the invariants of sharded locks.
The presented solution just sidesteps this problem by defaulting to a fairly low
value of MaxShards. This can be overridden by the user with explicit action
(though the runtime has the freedom to bound the number more strictly than the
user's value, e.g. to limit the size of internal data-structures to reasonable
levels.).
One thing to keep in mind, clients who require garbage collection of stale
values can build this on top of percpu.Sharded. For instance, one could
imagine a design where clients would maintain a counter recording each use. A
watchdog goroutine can then scan the elements and if a particular value has not
been used for some period of time, swap in a nil pointer, and then gracefully
tear down the value (potentially transferring the logical data encapsulated to
other elements in the percpu.Sharded).
Requiring clients to implement their own GC in this way seems kinda gross, but on the other hand, its unclear to me how to generically solve this problem without knowledge of the client use-case. One could imagine some sort of reference-counting design, but again, without knowing the semantics of the use-case, its hard to know if its safe to clear the reference to the type.
Also, for a performance-oriented type, like percpu.Sharded, it seems
unfortunate to add unnecessary synchronization to the fast path of the type (and
I don't see how to implement something in this direction without adding
synchronization).
ShardInfo a struct and not just an int?This is mostly to retain the ability to extend the API in a compatible manner. One concrete avenue is to add additional details to allow clients to optimize their code for the NUMA architecture of the machine. For instance, for a sharded buffering scheme (i.e. the "Order-independent accumulator" above), it might make sense to have multiple levels of buffering in play, with another level at the NUMA-node layer.
ShardInfo.ShardIndex returning an id for the CPU, or the P executing the goroutine?This is left unspecified, but the name of the package seems to imply the former. In practice, I think we want a combination.
That is to say, we would prefer that a program running on a 2-core machine with
GOMAXPROCS set to 100 should use 2 shards, not 100. On the other hand, we
would also prefer that a program running on a 100-core machine with GOMAXPROCS
set to 2 should also use 2 shards, not 100.
This ideal state should be achievable on systems that provide reasonable APIs to retrieve the id of the current CPU.
That said, any implementation effort would likely start with a simple portable
implementation which uses the id of the local P. This will allow us to get a
sense of the performance of the type, and to serve as a fallback implementation
for platforms where the necessary APIs are either not available, or require
privileged execution.
percpu.Sharded to behave differently during tests?This is a good question; I am not certain of the answer here. I am confident
that during race mode, we should definitely randomize the behaviour of
percpu.Sharded significantly (and the implementation sketch above does that).
However, for tests, the answer seems less clear to me.
As presented, the implementation sketch above randomizes the value by flipping
randomly between two values for every CPU. That seems like it will catch bugs
where the client assumes that sequential calls to Get/GetOrCreate will return
the same values. That amount of randomness seems warranted to me, though I'd
understand if folks would prefer to avoid it in favor of keeping non-test code
and test code behaving identically.
On a more mundane note: I'm not entirely sure if this is implementable with
zero-cost. One fairly efficient strategy would be an internal package that
exposes an "IsTesting bool", which is set by the testing package and read by
the percpu package. But ideally, this could be optimized away at compile time;
I don't believe we have any mechanism to do this now.
ShardInfo.ShardIndex at all?I think so. Even if we don't, clients can retrieve an effectively equivalent
value by just incrementing an atomic integer inside the createFn passed to
GetOrCreate. For pre-allocated use-cases (e.g. see the Facebook SharedMutex
linked above), it seems important to let clients index into pre-allocated
memory.
We could define GetOrCreate to behave like Get if the passed createFn is
nil. This is less API (and might be more efficient, until mid-stack inlining
works), but seems less semantically clean to me. It seems better to just have
clients say what they want explicitly.
If we had to choose one of those, then I would say we should expose Do. This
is because it is the higher performance, minimal-synchronization version, and
DoLocked can be implemented on top. That said, I do think we should just
provide both. The implementation is simple, and implementing it on top feels
odd.
Of the 4 use-cases presented above, 2 would probably use Do (counters and
order-independent accumulators), and 2 would probably use DoLocked (read-write
locks, and RPC clients (for the latter, probably just for implementing
Close())).
I'm not particularly wedded to any of the names in the API sketch above, so I'm happy to see it changed to whatever people prefer.
The API presented above is straightforward to implement without any runtime
support; in particular, this could be implemented as a thin wrapper around a
sync.Once. This will not effectively reduce contention, but it would still be
a correct implementation. It's probably a good idea to implement such a shim,
and put it in the x/sync repo, with appropriate build tags and type-aliases to
allow clients to immediately start using the new type.