forked from appscode/g2
-
Notifications
You must be signed in to change notification settings - Fork 5
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Sync/Wait for processing handler (#14)
Use a key-specific wait for delayed additions into the handler map.
- Loading branch information
Showing
4 changed files
with
225 additions
and
13 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,115 @@ | ||
| package client | ||
|
|
||
| import ( | ||
| "container/list" | ||
| "golang.org/x/net/context" | ||
|
|
||
| "sync" | ||
| "time" | ||
| ) | ||
|
|
||
| type HandlerMap struct { | ||
| mu sync.Mutex | ||
| innerMap map[string]ResponseHandler | ||
| waitersMap map[string]*list.List | ||
| } | ||
|
|
||
| type waiter struct { | ||
| ready chan<- struct{} // Closed when semaphore acquired. | ||
| } | ||
|
|
||
| func NewHandlerMap() *HandlerMap { | ||
| return &HandlerMap{sync.Mutex{}, | ||
| make(map[string]ResponseHandler, 100), | ||
| make(map[string]*list.List, 100), | ||
| } | ||
| } | ||
|
|
||
| func (m *HandlerMap) GetCounts() (counts int, waiters int) { | ||
| m.mu.Lock() | ||
| defer m.mu.Unlock() | ||
| return len(m.innerMap), len(m.waitersMap) | ||
| } | ||
|
|
||
| func (m *HandlerMap) Put(key string, value ResponseHandler) { | ||
| m.mu.Lock() | ||
| defer m.mu.Unlock() | ||
| m.innerMap[key] = value | ||
| // signal to any waiters here | ||
| if waiters, ok := m.waitersMap[key]; ok { | ||
| for { | ||
| next := waiters.Front() | ||
| if next == nil { | ||
| break // No more waiters blocked. | ||
| } | ||
| w := next.Value.(waiter) | ||
| waiters.Remove(next) | ||
| close(w.ready) | ||
| } | ||
| delete(m.waitersMap, key) | ||
| } | ||
| } | ||
|
|
||
| func (m *HandlerMap) Delete(key string) { | ||
| m.mu.Lock() | ||
| defer m.mu.Unlock() | ||
| delete(m.innerMap, key) | ||
| } | ||
|
|
||
| func (m *HandlerMap) Get(key string, timeoutMs int) (value ResponseHandler, ok bool) { | ||
| m.mu.Lock() | ||
|
|
||
| // optimistic check first | ||
| value, ok = m.innerMap[key] | ||
| if ok { | ||
| m.mu.Unlock() | ||
| return | ||
| } | ||
|
|
||
| // let's remember the current time | ||
| curTime := time.Now() | ||
| maxTime := curTime.Add(time.Duration(timeoutMs) * time.Millisecond) | ||
|
|
||
| for time.Now().Before(maxTime) && !ok { | ||
| value, ok = m.innerMap[key] | ||
| if !ok { | ||
| nsLeft := maxTime.Sub(time.Now()).Nanoseconds() | ||
| ctx, _ := context.WithTimeout(context.Background(), time.Duration(nsLeft)*time.Nanosecond) | ||
|
|
||
| waiters, wok := m.waitersMap[key] | ||
| if !wok { | ||
| waiters = &list.List{} | ||
| m.waitersMap[key] = waiters | ||
| } | ||
| ready := make(chan struct{}) | ||
| w := waiter{ready: ready} | ||
| elem := waiters.PushBack(w) | ||
| m.mu.Unlock() // unlock before we start waiting on stuff | ||
|
|
||
| select { | ||
| case <-ctx.Done(): | ||
| m.mu.Lock() | ||
| select { | ||
| case <-ready: | ||
| // in case we got signalled during cancellation | ||
| continue | ||
| default: | ||
| // we got timeout, let's remove | ||
| waiters.Remove(elem) | ||
| if waiters.Len() == 0 { | ||
| delete(m.waitersMap, key) | ||
| } | ||
| } | ||
| m.mu.Unlock() | ||
| return | ||
|
|
||
| case <-ready: | ||
| m.mu.Lock() // going back to the loop, gotta lock | ||
| continue | ||
| } | ||
| } | ||
| } | ||
|
|
||
| m.mu.Unlock() | ||
| return | ||
| } |
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,100 @@ | ||
| package client | ||
|
|
||
| import ( | ||
| "github.com/stretchr/testify/assert" | ||
| "math" | ||
| "testing" | ||
| "time" | ||
| ) | ||
|
|
||
| const ( | ||
| testKey = "test_key" | ||
| timeoutMs = 200 | ||
| marginErrorPct = 10 | ||
| ) | ||
|
|
||
| func getMsSince(startTime time.Time) int { | ||
| return int(time.Now().Sub(startTime).Nanoseconds() / 1e6) | ||
| } | ||
|
|
||
| func TestHandlerMapEarlyStoreRetrieve(t *testing.T) { | ||
|
|
||
| handler_map := NewHandlerMap() | ||
| var handler ResponseHandler = func(*Response) { | ||
| t.Logf("test: got a response \n") | ||
| } | ||
| handler_map.Put(testKey, handler) | ||
| myHandler, ok := handler_map.Get(testKey, timeoutMs) | ||
| if !ok { | ||
| t.Error("Failed to get test key") | ||
| } | ||
| myHandler(nil) | ||
|
|
||
| } | ||
|
|
||
| func TestHandlerMapDelayedPutRetrieve(t *testing.T) { | ||
|
|
||
| handler_map := NewHandlerMap() | ||
| startTime := time.Now() | ||
| expectedResponseMs := timeoutMs / 2 | ||
|
|
||
| go func() { | ||
| time.Sleep(time.Duration(expectedResponseMs) * time.Millisecond) | ||
|
|
||
| // at this point the Get would be waiting for the response. | ||
| counts, waiters := handler_map.GetCounts() | ||
| assert.Equal(t, 0, counts, "Map Elements") | ||
| assert.Equal(t, 1, waiters, "Waiter groups") | ||
|
|
||
| var handler ResponseHandler = func(*Response) { | ||
| t.Logf("test: got a response at time %d ms after start\n", getMsSince(startTime)) | ||
| } | ||
| handler_map.Put(testKey, handler) | ||
| }() | ||
|
|
||
| t.Logf("test: started waiting for key at %d ms after start\n", getMsSince(startTime)) | ||
| myHandler, ok := handler_map.Get(testKey, timeoutMs) | ||
| if !ok { | ||
| t.Error("Failed to get test key") | ||
| } else { | ||
| myHandler(nil) | ||
| actualResponseMs := getMsSince(startTime) | ||
| var comp assert.Comparison = func() (success bool) { | ||
| return math.Abs(float64(actualResponseMs-expectedResponseMs))/float64(expectedResponseMs) < float64(marginErrorPct)/100 | ||
| } | ||
| assert.Condition(t, comp, "Response did not arrive within %d%% margin, expected time %d ms", marginErrorPct, expectedResponseMs) | ||
|
|
||
| } | ||
|
|
||
| } | ||
|
|
||
| func TestHandlerMapTimeoutPutTooLate(t *testing.T) { | ||
|
|
||
| handler_map := NewHandlerMap() | ||
| startTime := time.Now() | ||
|
|
||
| go func() { | ||
| time.Sleep(2 * timeoutMs * time.Millisecond) | ||
| handler_map.Put(testKey, func(*Response) {}) | ||
| }() | ||
|
|
||
| t.Logf("test: started waiting for key at %d ms after start\n", getMsSince(startTime)) | ||
| _, ok := handler_map.Get(testKey, timeoutMs) | ||
| if ok { | ||
| t.Error("Should have timed out when getting the key") | ||
| return | ||
| } else { | ||
| actualTimeoutMs := getMsSince(startTime) | ||
| t.Logf("test: timed out waiting for key at %d ms after start\n", actualTimeoutMs) | ||
| var comp assert.Comparison = func() (success bool) { | ||
| return math.Abs(float64(actualTimeoutMs-timeoutMs))/timeoutMs < float64(marginErrorPct)/100 | ||
| } | ||
| assert.Condition(t, comp, "Timeout did not occur within %d%% margin, expected timeout ms: %d", marginErrorPct, timeoutMs) | ||
| // wait till producer has added the element | ||
| time.Sleep(3 * timeoutMs * time.Millisecond) | ||
| counts, waiters := handler_map.GetCounts() | ||
| assert.Equal(t, 1, counts, "Map elements") | ||
| assert.Equal(t, 0, waiters, "Waiter groups") | ||
| } | ||
|
|
||
| } |
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