Issue 336560043: Adds bandwidth control to copy tasks (Closed)
|
Created:
8 years, 5 months ago by joshuablome Modified:
8 years, 2 months ago CC:
opi-crs_google.com Visibility:
Public. |
DescriptionAdds bandwidth control to copy tasks
Patch Set 1 #
Total comments: 20
Patch Set 2 : Addressed first round of comments #Patch Set 3 : Removed commented out line. #Patch Set 4 : No longer checking bandwidth usage asynchronously #
Total comments: 7
Patch Set 5 : Added SLeep to mock clock and added a clock to CopyHandler #
Total comments: 3
Patch Set 6 : Changed Sleep parameter name to be consistent #
Total comments: 3
Patch Set 7 : Now using builtin rate.Limiter instead of manual rate limiting #
Total comments: 1
Patch Set 8 : Added error handling #
Total comments: 2
Patch Set 9 : Punctuation fix and removed unnecessary error check #MessagesTotal messages: 17
https://codereview.appspot.com/336560043/diff/1/agent/bandwidth_controller.go File agent/bandwidth_controller.go (right): https://codereview.appspot.com/336560043/diff/1/agent/bandwidth_controller.go... agent/bandwidth_controller.go:24: type BandwidthController struct { Exported types and functions should have comments describing what they do, of the form: // BandwidthController ... Aside: does this type need to be exported? https://codereview.appspot.com/336560043/diff/1/agent/bandwidth_controller.go... agent/bandwidth_controller.go:30: desiredCopyTime := time.Duration(float64(bytesWritten) / float64(bc.bandwidth) * float64(time.Second)) This can potentially divide by zero https://codereview.appspot.com/336560043/diff/1/agent/copy.go File agent/copy.go (right): https://codereview.appspot.com/336560043/diff/1/agent/copy.go#newcode155 agent/copy.go:155: go h.bwController.Start(&totalBytesWritten, &sleepDuration, time.Now()) Why do we need a goroutine for this? It seems like all we're doing is comparing start time with how much has been written. The clock will tick in the background regardless, so would we be better off using a mock clock here and avoiding the ticker approach? https://codereview.appspot.com/336560043/diff/1/agent/copy.go#newcode165 agent/copy.go:165: _, err = w.Write(buffer[:n]) This will write bufferSize bytes, which could vastly exceed our bandwidth for a long period of time. Do we need to break up the bufferSize (or at least how much we send over the network) to ensure this doesn't happen?
Sign in to reply to this message.
https://codereview.appspot.com/336560043/diff/1/agent/bandwidth_controller.go File agent/bandwidth_controller.go (right): https://codereview.appspot.com/336560043/diff/1/agent/bandwidth_controller.go... agent/bandwidth_controller.go:30: desiredCopyTime := time.Duration(float64(bytesWritten) / float64(bc.bandwidth) * float64(time.Second)) On 2018/02/16 22:30:54, thobrla wrote: > This can potentially divide by zero It's also likely that a bandwidth of zero (or negative) will be what we use to indicate "no bandwidth control", or "unlimited bandwidth". https://codereview.appspot.com/336560043/diff/1/agent/bandwidth_controller.go... agent/bandwidth_controller.go:39: bytesWrittenP *int, sleepDurationP *time.Duration, startTime time.Time) { Unless you need to be able to write to the int, it's better to pass-by-value instead of using a pointer here. https://codereview.appspot.com/336560043/diff/1/agent/copy.go File agent/copy.go (right): https://codereview.appspot.com/336560043/diff/1/agent/copy.go#newcode59 agent/copy.go:59: bwController BandwidthController I don't think the BandwidthController is something we want to be part of the CopyHandler struct. The general idea is that each task will have it's own bandwidth assignment, so the bandwidth control should be tied to the individual tasks. This means it will likely have to be something that is instantiated with ever new CopyHandler Do(...) call, and apply to that specific task. https://codereview.appspot.com/336560043/diff/1/agent/copy.go#newcode165 agent/copy.go:165: _, err = w.Write(buffer[:n]) On 2018/02/16 22:30:54, thobrla wrote: > This will write bufferSize bytes, which could vastly exceed our bandwidth for a > long period of time. Do we need to break up the bufferSize (or at least how > much we send over the network) to ensure this doesn't happen? This chunk size right now is set to 32MB (See the chunkSize flag in agentmain.go). We should figure out what the right level of granularity is, since there's a trade-off between how smooth we can make the bandwidth control, and how large or small we make our upload chunks. (There's more complexity going on here than what this code implies, but the gist is that this is breaking up the transfer into 32MB HTTP payloads). I'm not advocating for any particular level of granularity, but you should think about it, and document/comment whatever your conclusions are.
Sign in to reply to this message.
https://codereview.appspot.com/336560043/diff/1/agent/bandwidth_controller.go File agent/bandwidth_controller.go (right): https://codereview.appspot.com/336560043/diff/1/agent/bandwidth_controller.go... agent/bandwidth_controller.go:24: type BandwidthController struct { On 2018/02/16 22:30:54, thobrla wrote: > Exported types and functions should have comments describing what they do, of > the form: > > // BandwidthController ... > > Aside: does this type need to be exported? Ah, probably not. I misunderstood how exporting names worked. Thought it had to be exported to be used outside the file, instead of outside the package. https://codereview.appspot.com/336560043/diff/1/agent/bandwidth_controller.go... agent/bandwidth_controller.go:30: desiredCopyTime := time.Duration(float64(bytesWritten) / float64(bc.bandwidth) * float64(time.Second)) On 2018/02/16 23:56:10, akaiser wrote: > On 2018/02/16 22:30:54, thobrla wrote: > > This can potentially divide by zero > > It's also likely that a bandwidth of zero (or negative) will be what we use to > indicate "no bandwidth control", or "unlimited bandwidth". Done. https://codereview.appspot.com/336560043/diff/1/agent/bandwidth_controller.go... agent/bandwidth_controller.go:39: bytesWrittenP *int, sleepDurationP *time.Duration, startTime time.Time) { On 2018/02/16 23:56:10, akaiser wrote: > Unless you need to be able to write to the int, it's better to pass-by-value > instead of using a pointer here. It doesn't need to write, but this periodically accesses the value of bytesWritten. The alternative is having a method in copy.go that periodically passes/updates the value. https://codereview.appspot.com/336560043/diff/1/agent/copy.go File agent/copy.go (right): https://codereview.appspot.com/336560043/diff/1/agent/copy.go#newcode59 agent/copy.go:59: bwController BandwidthController On 2018/02/16 23:56:10, akaiser wrote: > I don't think the BandwidthController is something we want to be part of the > CopyHandler struct. The general idea is that each task will have it's own > bandwidth assignment, so the bandwidth control should be tied to the individual > tasks. This means it will likely have to be something that is instantiated with > ever new CopyHandler Do(...) call, and apply to that specific task. Yup, good call. Not sure if it should go in the taskParams or as its own parameter. https://codereview.appspot.com/336560043/diff/1/agent/copy.go#newcode155 agent/copy.go:155: go h.bwController.Start(&totalBytesWritten, &sleepDuration, time.Now()) On 2018/02/16 22:30:54, thobrla wrote: > Why do we need a goroutine for this? It seems like all we're doing is comparing > start time with how much has been written. The clock will tick in the > background regardless, so would we be better off using a mock clock here and > avoiding the ticker approach? I'm not sure what you mean. We need to start some function that looks at the bandwidth used. Without a goroutine, a persistent function would block forever. We could put the check in the for loop that does the copy. That would actually be simpler, but I thought that check would be more granular/often than we want. https://codereview.appspot.com/336560043/diff/1/agent/copy.go#newcode165 agent/copy.go:165: _, err = w.Write(buffer[:n]) On 2018/02/16 23:56:10, akaiser wrote: > On 2018/02/16 22:30:54, thobrla wrote: > > This will write bufferSize bytes, which could vastly exceed our bandwidth for > a > > long period of time. Do we need to break up the bufferSize (or at least how > > much we send over the network) to ensure this doesn't happen? > > This chunk size right now is set to 32MB (See the chunkSize flag in > agentmain.go). We should figure out what the right level of granularity is, > since there's a trade-off between how smooth we can make the bandwidth control, > and how large or small we make our upload chunks. (There's more complexity going > on here than what this code implies, but the gist is that this is breaking up > the transfer into 32MB HTTP payloads). > > I'm not advocating for any particular level of granularity, but you should think > about it, and document/comment whatever your conclusions are. So should I set a limit on the bufferSize/how much is written on each loop? Or do we think the 32MB chunk size is a good granularity right now?
Sign in to reply to this message.
More review to come later https://codereview.appspot.com/336560043/diff/1/agent/copy.go File agent/copy.go (right): https://codereview.appspot.com/336560043/diff/1/agent/copy.go#newcode155 agent/copy.go:155: go h.bwController.Start(&totalBytesWritten, &sleepDuration, time.Now()) On 2018/02/20 22:31:07, Josh B wrote: > On 2018/02/16 22:30:54, thobrla wrote: > > Why do we need a goroutine for this? It seems like all we're doing is > comparing > > start time with how much has been written. The clock will tick in the > > background regardless, so would we be better off using a mock clock here and > > avoiding the ticker approach? > > I'm not sure what you mean. We need to start some function that looks at the > bandwidth used. Without a goroutine, a persistent function would block forever. > We could put the check in the for loop that does the copy. That would actually > be simpler, but I thought that check would be more granular/often than we want. The check is inexpensive, right? That seems far simpler than trying to do it asynchronously.
Sign in to reply to this message.
https://codereview.appspot.com/336560043/diff/1/agent/copy.go File agent/copy.go (right): https://codereview.appspot.com/336560043/diff/1/agent/copy.go#newcode155 agent/copy.go:155: go h.bwController.Start(&totalBytesWritten, &sleepDuration, time.Now()) On 2018/02/21 00:30:58, thobrla wrote: > On 2018/02/20 22:31:07, Josh B wrote: > > On 2018/02/16 22:30:54, thobrla wrote: > > > Why do we need a goroutine for this? It seems like all we're doing is > > comparing > > > start time with how much has been written. The clock will tick in the > > > background regardless, so would we be better off using a mock clock here and > > > avoiding the ticker approach? > > > > I'm not sure what you mean. We need to start some function that looks at the > > bandwidth used. Without a goroutine, a persistent function would block > forever. > > We could put the check in the for loop that does the copy. That would actually > > be simpler, but I thought that check would be more granular/often than we > want. > > The check is inexpensive, right? That seems far simpler than trying to do it > asynchronously. Yup, then I think it makes more sense for the check to just live in copy.go too.
Sign in to reply to this message.
https://codereview.appspot.com/336560043/diff/1/agent/copy.go File agent/copy.go (right): https://codereview.appspot.com/336560043/diff/1/agent/copy.go#newcode153 agent/copy.go:153: time.Sleep(h.bwController.WaitTime(totalBytesWritten, time.Since(startTime))) Deferred functions' arguments are evaluated immediately: https://tour.golang.org/flowcontrol/12 So totalBytesWritten and time.Since(startTime) will be 0 and not-much-time, respectively, which I don't think is what you want. At a higher level, what's the incentive for using defer here? Is it to make sure that we wait in all cases (including errors)? https://codereview.appspot.com/336560043/diff/1/agent/copy.go#newcode155 agent/copy.go:155: go h.bwController.Start(&totalBytesWritten, &sleepDuration, time.Now()) On 2018/02/21 18:13:28, Josh B wrote: > On 2018/02/21 00:30:58, thobrla wrote: > > On 2018/02/20 22:31:07, Josh B wrote: > > > On 2018/02/16 22:30:54, thobrla wrote: > > > > Why do we need a goroutine for this? It seems like all we're doing is > > > comparing > > > > start time with how much has been written. The clock will tick in the > > > > background regardless, so would we be better off using a mock clock here > and > > > > avoiding the ticker approach? > > > > > > I'm not sure what you mean. We need to start some function that looks at the > > > bandwidth used. Without a goroutine, a persistent function would block > > forever. > > > We could put the check in the for loop that does the copy. That would > actually > > > be simpler, but I thought that check would be more granular/often than we > > want. > > > > The check is inexpensive, right? That seems far simpler than trying to do it > > asynchronously. > > Yup, then I think it makes more sense for the check to just live in copy.go too. Expanding on this a bit, a core computer science principle is that multi-threading is helpful only when you're blocked on IO. In this case you're not blocked on IO (and actually you want to do the blocking yourself!) https://codereview.appspot.com/336560043/diff/1/agent/copy.go#newcode165 agent/copy.go:165: _, err = w.Write(buffer[:n]) On 2018/02/20 22:31:07, Josh B wrote: > On 2018/02/16 23:56:10, akaiser wrote: > > On 2018/02/16 22:30:54, thobrla wrote: > > > This will write bufferSize bytes, which could vastly exceed our bandwidth > for > > a > > > long period of time. Do we need to break up the bufferSize (or at least how > > > much we send over the network) to ensure this doesn't happen? > > > > This chunk size right now is set to 32MB (See the chunkSize flag in > > agentmain.go). We should figure out what the right level of granularity is, > > since there's a trade-off between how smooth we can make the bandwidth > control, > > and how large or small we make our upload chunks. (There's more complexity > going > > on here than what this code implies, but the gist is that this is breaking up > > the transfer into 32MB HTTP payloads). > > > > I'm not advocating for any particular level of granularity, but you should > think > > about it, and document/comment whatever your conclusions are. > > So should I set a limit on the bufferSize/how much is written on each loop? Or > do we think the 32MB chunk size is a good granularity right now? Per our offline discussion, unless we decouple the chunk size from the send size, then then chunk size has an impact on our choice of minimum bandwidth per task. https://codereview.appspot.com/336560043/diff/60001/agent/copy.go File agent/copy.go (right): https://codereview.appspot.com/336560043/diff/60001/agent/copy.go#newcode148 agent/copy.go:148: bandwidth = taskParams["bandwidth"].(int) I'm not familiar with .(int), can you help me understand what this does? https://codereview.appspot.com/336560043/diff/60001/agent/copy_test.go File agent/copy_test.go (right): https://codereview.appspot.com/336560043/diff/60001/agent/copy_test.go#newcod... agent/copy_test.go:196: }{{10000}, {1000}} With such a small amount of content transferred, does it really matter to test multiple values? https://codereview.appspot.com/336560043/diff/60001/agent/copy_test.go#newcod... agent/copy_test.go:229: startTime := time.Now() Shouldn't we use a mock ticker or mock clock for this test? Otherwise the test itself will wait for a number of seconds (and potentially be flaky).
Sign in to reply to this message.
https://codereview.appspot.com/336560043/diff/1/agent/copy.go File agent/copy.go (right): https://codereview.appspot.com/336560043/diff/1/agent/copy.go#newcode153 agent/copy.go:153: time.Sleep(h.bwController.WaitTime(totalBytesWritten, time.Since(startTime))) On 2018/02/21 19:33:59, thobrla wrote: > Deferred functions' arguments are evaluated immediately: > https://tour.golang.org/flowcontrol/12 > > So totalBytesWritten and time.Since(startTime) will be 0 and not-much-time, > respectively, which I don't think is what you want. > > At a higher level, what's the incentive for using defer here? Is it to make > sure that we wait in all cases (including errors)? In this case, they aren't arguments to the deferred function, so they would be evaluated when the function executes. But either way, this was needed when we might copy a whole file before the task was told to wait, so the defer would still delay the task's completion. With the delay always happening in the read/write loop, this *should* never be necessary. https://codereview.appspot.com/336560043/diff/60001/agent/copy.go File agent/copy.go (right): https://codereview.appspot.com/336560043/diff/60001/agent/copy.go#newcode148 agent/copy.go:148: bandwidth = taskParams["bandwidth"].(int) On 2018/02/21 19:33:59, thobrla wrote: > I'm not familiar with .(int), can you help me understand what this does? It's a type assertion to get the value from the task param because taskParams is a map[string] interface{}. I changed it to be safer if the value isn't an int. https://codereview.appspot.com/336560043/diff/60001/agent/copy_test.go File agent/copy_test.go (right): https://codereview.appspot.com/336560043/diff/60001/agent/copy_test.go#newcod... agent/copy_test.go:196: }{{10000}, {1000}} On 2018/02/21 19:33:59, thobrla wrote: > With such a small amount of content transferred, does it really matter to test > multiple values? Not anymore, the multiple values were useful when there was a particular time threshold we wanted to pass. https://codereview.appspot.com/336560043/diff/60001/agent/copy_test.go#newcod... agent/copy_test.go:229: startTime := time.Now() On 2018/02/21 19:33:59, thobrla wrote: > Shouldn't we use a mock ticker or mock clock for this test? Otherwise the test > itself will wait for a number of seconds (and potentially be flaky). Now using a mock clock to control time.
Sign in to reply to this message.
LGTM https://codereview.appspot.com/336560043/diff/60001/agent/copy.go File agent/copy.go (right): https://codereview.appspot.com/336560043/diff/60001/agent/copy.go#newcode148 agent/copy.go:148: bandwidth = taskParams["bandwidth"].(int) On 2018/02/21 23:55:47, Josh B wrote: > On 2018/02/21 19:33:59, thobrla wrote: > > I'm not familiar with .(int), can you help me understand what this does? > > It's a type assertion to get the value from the task param because taskParams is > a map[string] interface{}. I changed it to be safer if the value isn't an int. Acknowledged. https://codereview.appspot.com/336560043/diff/80001/agent/copy_test.go File agent/copy_test.go (right): https://codereview.appspot.com/336560043/diff/80001/agent/copy_test.go#newcod... agent/copy_test.go:194: mockCtrl := gomock.NewController(t) Should there also be a test that spans multiple resumable chunks? https://codereview.appspot.com/336560043/diff/80001/helpers/clock.go File helpers/clock.go (right): https://codereview.appspot.com/336560043/diff/80001/helpers/clock.go#newcode19 helpers/clock.go:19: func (realClock) Sleep(dur time.Duration) { Nit: argument is named "d" in the mock but "dur" here
Sign in to reply to this message.
https://codereview.appspot.com/336560043/diff/80001/helpers/clock.go File helpers/clock.go (right): https://codereview.appspot.com/336560043/diff/80001/helpers/clock.go#newcode19 helpers/clock.go:19: func (realClock) Sleep(dur time.Duration) { On 2018/02/22 01:40:02, thobrla wrote: > Nit: argument is named "d" in the mock but "dur" here Done.
Sign in to reply to this message.
https://codereview.appspot.com/336560043/diff/100001/agent/copy.go File agent/copy.go (right): https://codereview.appspot.com/336560043/diff/100001/agent/copy.go#newcode40 agent/copy.go:40: defaultBandwidthCap int = 16 * 1024 * 1024 // Default bandwidth cap is 16 MB / second. Before you submit this, we should probably raise this value to "infinite". This isn't meaningful unless we have information from the DCP, so there's no reason to artificially slow down the system.
Sign in to reply to this message.
Reply to view. On Fri, Feb 23, 2018 at 10:05 AM, <thobrla@google.com> wrote: > > https://codereview.appspot.com/336560043/diff/100001/agent/copy.go > File agent/copy.go (right): > > https://codereview.appspot.com/336560043/diff/100001/agent/ > copy.go#newcode40 > agent/copy.go:40: defaultBandwidthCap int = 16 * 1024 * 1024 // > Default bandwidth cap is 16 MB / second. > Before you submit this, we should probably raise this value to > "infinite". This isn't meaningful unless we have information from the > DCP, so there's no reason to artificially slow down the system. > > https://codereview.appspot.com/336560043/ > > -- > You received this message because you are subscribed to the Google Groups > "opi-crs" group. > To unsubscribe from this group and stop receiving emails from it, send an > email to opi-crs+unsubscribe@google.com. > To post to this group, send email to opi-crs@google.com. > To view this discussion on the web visit https://groups.google.com/a/go > ogle.com/d/msgid/opi-crs/001a113ee8680584990565e50098%40google.com. >
Sign in to reply to this message.
One Suggestion you may consider below. Thanks, Mahmoud https://codereview.appspot.com/336560043/diff/100001/agent/copy.go File agent/copy.go (right): https://codereview.appspot.com/336560043/diff/100001/agent/copy.go#newcode174 agent/copy.go:174: h.clock.Sleep(waitTime(bandwidth, totalBytesWritten, h.clock.Now().Sub(startTime))) Have you considered to use rate.Limiter? https://godoc.org/golang.org/x/time/rate#Limiter If that work, it will reduce the amount of code we generate and maintain.
Sign in to reply to this message.
https://codereview.appspot.com/336560043/diff/100001/agent/copy.go File agent/copy.go (right): https://codereview.appspot.com/336560043/diff/100001/agent/copy.go#newcode174 agent/copy.go:174: h.clock.Sleep(waitTime(bandwidth, totalBytesWritten, h.clock.Now().Sub(startTime))) On 2018/02/23 18:37:13, mbassiouny wrote: > Have you considered to use rate.Limiter? > https://godoc.org/golang.org/x/time/rate#Limiter > > If that work, it will reduce the amount of code we generate and maintain. Yea, I hadn't seen that. It does greatly simplify things. Thanks Mahmoud!
Sign in to reply to this message.
Still LGTM
Sign in to reply to this message.
Thanks Josh, LGTM with minor nit. Thanks, Mahmoud https://codereview.appspot.com/336560043/diff/120001/agent/copy.go File agent/copy.go (right): https://codereview.appspot.com/336560043/diff/120001/agent/copy.go#newcode153 agent/copy.go:153: limiter.WaitN(ctx, maxBucketSize) This may return err, please check on error before moving forward.
Sign in to reply to this message.
LGTM 2 nits https://codereview.appspot.com/336560043/diff/140001/agent/copy.go File agent/copy.go (right): https://codereview.appspot.com/336560043/diff/140001/agent/copy.go#newcode152 agent/copy.go:152: // The rate limiter starts with a full token bucket, we need to empty it before copying copying. Blah blah, punctuation. :) https://codereview.appspot.com/336560043/diff/140001/agent/copy.go#newcode179 agent/copy.go:179: if err != nil { Is this second "if err != nil" block necessary? It doesn't look like it to me.
Sign in to reply to this message.
|
