Skip to content

GitLab

Commit 77a15c1b authored by hannahhoward's avatar hannahhoward
Browse files
Options

refactor(asyncloader): return single channel 

Async loader returns a single channel result type
parent 7a583a29
Show whitespace changes
Inline Side-by-side
Showing with 54 additions and 87 deletions
requestmanager/asyncloader/asyncloader.go
View file @ 77a15c1b
......@@ -12,8 +12,7 @@ import (
type loadRequest struct {
requestID gsmsg.GraphSyncRequestID
link ipld.Link
responseChan chan []byte
errChan chan error
resultChan chan AsyncLoadResult
}
var loadRequestPool = sync.Pool{
......@@ -24,13 +23,11 @@ var loadRequestPool = sync.Pool{
func newLoadRequest(requestID gsmsg.GraphSyncRequestID,
link ipld.Link,
responseChan chan []byte,
errChan chan error) *loadRequest {
resultChan chan AsyncLoadResult) *loadRequest {
lr := loadRequestPool.Get().(*loadRequest)
lr.requestID = requestID
lr.link = link
lr.responseChan = responseChan
lr.errChan = errChan
lr.resultChan = resultChan
return lr
}
......@@ -60,6 +57,12 @@ type finishRequestMessage struct {
// bytes nil, error nil = did not load, but try again later
type LoadAttempter func(gsmsg.GraphSyncRequestID, ipld.Link) ([]byte, error)
// AsyncLoadResult is sent once over the channel returned by an async load.
type AsyncLoadResult struct {
Data []byte
Err error
}
// AsyncLoader is used to make multiple attempts to load a blocks over the
// course of a request - as long as a request is in progress, it will make multiple
// attempts to load a block until it gets a definitive result of whether the block
......@@ -89,16 +92,15 @@ func New(ctx context.Context, loadAttempter LoadAttempter) *AsyncLoader {
// AsyncLoad asynchronously loads the given link for the given request ID. It returns a channel for data and a channel
// for errors -- only one message will be sent over either.
func (abl *AsyncLoader) AsyncLoad(requestID gsmsg.GraphSyncRequestID, link ipld.Link) (<-chan []byte, <-chan error) {
responseChan := make(chan []byte, 1)
errChan := make(chan error, 1)
lr := newLoadRequest(requestID, link, responseChan, errChan)
func (abl *AsyncLoader) AsyncLoad(requestID gsmsg.GraphSyncRequestID, link ipld.Link) <-chan AsyncLoadResult {
resultChan := make(chan AsyncLoadResult, 1)
lr := newLoadRequest(requestID, link, resultChan)
select {
case <-abl.ctx.Done():
abl.terminateWithError("Context Closed", responseChan, errChan)
abl.terminateWithError("Context Closed", resultChan)
case abl.incomingMessages <- lr:
}
return responseChan, errChan
return resultChan
}
// NewResponsesAvailable indicates that the async loader should make another attempt to load
......@@ -179,22 +181,20 @@ func (abl *AsyncLoader) messageQueueWorker() {
func (lr *loadRequest) handle(abl *AsyncLoader) {
_, ok := abl.activeRequests[lr.requestID]
if !ok {
abl.terminateWithError("No active request", lr.responseChan, lr.errChan)
abl.terminateWithError("No active request", lr.resultChan)
returnLoadRequest(lr)
return
}
response, err := abl.loadAttempter(lr.requestID, lr.link)
if err != nil {
lr.errChan <- err
close(lr.errChan)
close(lr.responseChan)
lr.resultChan <- AsyncLoadResult{nil, err}
close(lr.resultChan)
returnLoadRequest(lr)
return
}
if response != nil {
lr.responseChan <- response
close(lr.errChan)
close(lr.responseChan)
lr.resultChan <- AsyncLoadResult{response, nil}
close(lr.resultChan)
returnLoadRequest(lr)
return
}
......@@ -211,7 +211,7 @@ func (frm *finishRequestMessage) handle(abl *AsyncLoader) {
abl.pausedRequests = nil
for _, lr := range pausedRequests {
if lr.requestID == frm.requestID {
abl.terminateWithError("No active request", lr.responseChan, lr.errChan)
abl.terminateWithError("No active request", lr.resultChan)
returnLoadRequest(lr)
} else {
abl.pausedRequests = append(abl.pausedRequests, lr)
......@@ -232,8 +232,7 @@ func (nram *newResponsesAvailableMessage) handle(abl *AsyncLoader) {
}
}
func (abl *AsyncLoader) terminateWithError(errMsg string, responseChan chan<- []byte, errChan chan<- error) {
errChan <- errors.New(errMsg)
close(errChan)
close(responseChan)
func (abl *AsyncLoader) terminateWithError(errMsg string, resultChan chan<- AsyncLoadResult) {
resultChan <- AsyncLoadResult{nil, errors.New(errMsg)}
close(resultChan)
}
requestmanager/asyncloader/asyncloader_test.go
View file @ 77a15c1b
......@@ -27,24 +27,19 @@ func TestAsyncLoadWhenRequestNotInProgress(t *testing.T) {
link := testbridge.NewMockLink()
requestID := gsmsg.GraphSyncRequestID(rand.Int31())
responseChan, errChan := asyncLoader.AsyncLoad(requestID, link)
resultChan := asyncLoader.AsyncLoad(requestID, link)
select {
case _, ok := <-responseChan:
if ok {
case result := <-resultChan:
if result.Data != nil {
t.Fatal("should not have sent responses")
}
case <-ctx.Done():
t.Fatal("should have closed response channel")
}
select {
case _, ok := <-errChan:
if !ok {
if result.Err == nil {
t.Fatal("should have sent an error")
}
case <-ctx.Done():
t.Fatal("should have closed error channel")
t.Fatal("should have produced result")
}
if callCount > 0 {
......@@ -67,24 +62,18 @@ func TestAsyncLoadWhenInitialLoadSucceeds(t *testing.T) {
link := testbridge.NewMockLink()
requestID := gsmsg.GraphSyncRequestID(rand.Int31())
asyncLoader.StartRequest(requestID)
responseChan, errChan := asyncLoader.AsyncLoad(requestID, link)
resultChan := asyncLoader.AsyncLoad(requestID, link)
select {
case _, ok := <-responseChan:
if !ok {
case result := <-resultChan:
if result.Data == nil {
t.Fatal("should have sent a response")
}
case <-ctx.Done():
t.Fatal("should have closed response channel")
}
select {
case _, ok := <-errChan:
if ok {
if result.Err != nil {
t.Fatal("should not have sent an error")
}
case <-ctx.Done():
t.Fatal("should have closed error channel")
t.Fatal("should have closed response channel")
}
if callCount == 0 {
......@@ -107,24 +96,18 @@ func TestAsyncLoadInitialLoadFails(t *testing.T) {
link := testbridge.NewMockLink()
requestID := gsmsg.GraphSyncRequestID(rand.Int31())
asyncLoader.StartRequest(requestID)
responseChan, errChan := asyncLoader.AsyncLoad(requestID, link)
resultChan := asyncLoader.AsyncLoad(requestID, link)
select {
case _, ok := <-responseChan:
if ok {
case result := <-resultChan:
if result.Data != nil {
t.Fatal("should not have sent responses")
}
case <-ctx.Done():
t.Fatal("should have closed response channel")
}
select {
case _, ok := <-errChan:
if !ok {
if result.Err == nil {
t.Fatal("should have sent an error")
}
case <-ctx.Done():
t.Fatal("should have closed error channel")
t.Fatal("should have closed response channel")
}
if callCount == 0 {
......@@ -153,34 +136,26 @@ func TestAsyncLoadInitialLoadIndeterminateThenSucceeds(t *testing.T) {
link := testbridge.NewMockLink()
requestID := gsmsg.GraphSyncRequestID(rand.Int31())
asyncLoader.StartRequest(requestID)
responseChan, errChan := asyncLoader.AsyncLoad(requestID, link)
resultChan := asyncLoader.AsyncLoad(requestID, link)
select {
case <-called:
case <-responseChan:
case <-resultChan:
t.Fatal("Should not have sent message on response chan")
case <-errChan:
t.Fatal("Should not have sent messages on error chan")
case <-ctx.Done():
t.Fatal("should have attempted load once")
}
asyncLoader.NewResponsesAvailable()
select {
case _, ok := <-responseChan:
if !ok {
case result := <-resultChan:
if result.Data == nil {
t.Fatal("should have sent a response")
}
case <-ctx.Done():
t.Fatal("should have closed response channel")
}
select {
case _, ok := <-errChan:
if ok {
if result.Err != nil {
t.Fatal("should not have sent an error")
}
case <-ctx.Done():
t.Fatal("should have closed error channel")
t.Fatal("should have closed response channel")
}
if callCount < 2 {
......@@ -209,13 +184,11 @@ func TestAsyncLoadInitialLoadIndeterminateThenRequestFinishes(t *testing.T) {
link := testbridge.NewMockLink()
requestID := gsmsg.GraphSyncRequestID(rand.Int31())
asyncLoader.StartRequest(requestID)
responseChan, errChan := asyncLoader.AsyncLoad(requestID, link)
resultChan := asyncLoader.AsyncLoad(requestID, link)
select {
case <-called:
case <-responseChan:
case <-resultChan:
t.Fatal("Should not have sent message on response chan")
case <-errChan:
t.Fatal("Should not have sent messages on error chan")
case <-ctx.Done():
t.Fatal("should have attempted load once")
}
......@@ -223,22 +196,17 @@ func TestAsyncLoadInitialLoadIndeterminateThenRequestFinishes(t *testing.T) {
asyncLoader.NewResponsesAvailable()
select {
case _, ok := <-responseChan:
if ok {
case result := <-resultChan:
if result.Data != nil {
t.Fatal("should not have sent responses")
}
case <-ctx.Done():
t.Fatal("should have closed response channel")
}
select {
case _, ok := <-errChan:
if !ok {
if result.Err == nil {
t.Fatal("should have sent an error")
}
case <-ctx.Done():
t.Fatal("should have closed error channel")
t.Fatal("should have closed response channel")
}
if callCount > 1 {
t.Fatal("should only have attempted one call but attempted multiple")
}
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment