Skip to content

Commit

Permalink
maintner: fix netsource Update offset errors.
Browse files Browse the repository at this point in the history
When resuming from a point, the reclog reading code would double check
the record headers offsets but because netsource MutationSource would
seek on the *os.File, what reclog saw and expected didn't match what
it read.

Also move temporary 5 second sleep to 2 seconds and adjust some
logging.

Updates golang/go#19866

Change-Id: I66d1f9df8bb36cf028b715ddd284cb10bc74b45b
Reviewed-on: https://go-review.googlesource.com/42184
Reviewed-by: Brad Fitzpatrick <[email protected]>
  • Loading branch information
bradfitz committed May 1, 2017
1 parent 43b7628 commit ef7ac5c
Show file tree
Hide file tree
Showing 4 changed files with 13 additions and 10 deletions.
3 changes: 3 additions & 0 deletions cmd/gopherbot/gopherbot.go
Original file line number Diff line number Diff line change
Expand Up @@ -71,10 +71,13 @@ func main() {

ctx := context.Background()
for {
t0 := time.Now()
err := bot.doTasks(ctx)
if err != nil {
log.Print(err)
}
botDur := time.Since(t0)
log.Printf("gopherbot ran in %v", botDur)
if !*daemon {
if err != nil {
os.Exit(1)
Expand Down
2 changes: 1 addition & 1 deletion maintner/maintnerd/gcslog.go
Original file line number Diff line number Diff line change
Expand Up @@ -396,7 +396,7 @@ func (gl *gcsLog) GetMutations(ctx context.Context) <-chan maintner.MutationStre
ch := make(chan maintner.MutationStreamEvent, 50) // buffered: overlap gunzip/unmarshal with loading
go func() {
err := gl.foreachSegmentReader(ctx, func(r io.Reader) error {
return reclog.ForeachRecord(r, func(off int64, hdr, rec []byte) error {
return reclog.ForeachRecord(r, 0, func(off int64, hdr, rec []byte) error {
m := new(maintpb.Mutation)
if err := proto.Unmarshal(rec, m); err != nil {
return err
Expand Down
11 changes: 5 additions & 6 deletions maintner/netsource.go
Original file line number Diff line number Diff line change
Expand Up @@ -74,21 +74,20 @@ func (ns *netMutSource) waitForServerSegmentUpdate(ctx context.Context) error {
return fn(ctx)
}

// TODO: 5 second sleep is dumb. make it
// TODO: few second sleep is dumb. make it
// subscribe to pubsubhelper? maybe the
// server's response header should reference
// its pubsubhelper server URL. but then we
// can't assume activity means it'll be picked
// up right away. so maybe wait for activity,
// and then poll every second for 10 seconds
// or so, or until there's changes, and then
// go back to every 5 second polling or
// go back to every 2 second polling or
// something. or maybe the maintnerd server should
// have its own long poll functionality.
// for now, just 5 second polling:
log.Printf("sleeping for 5s...")
// for now, just 2 second polling:
select {
case <-time.After(5 * time.Second):
case <-time.After(2 * time.Second):
return nil
case <-ctx.Done():
return ctx.Err()
Expand Down Expand Up @@ -247,7 +246,7 @@ func (ns *netMutSource) sendMutations(ctx context.Context, ch chan<- MutationStr
return err
}
}
return reclog.ForeachRecord(io.LimitReader(f, seg.size), func(off int64, hdr, rec []byte) error {
return reclog.ForeachRecord(io.LimitReader(f, seg.size-seg.skip), seg.skip, func(off int64, hdr, rec []byte) error {
m := new(maintpb.Mutation)
if err := proto.Unmarshal(rec, m); err != nil {
return err
Expand Down
7 changes: 4 additions & 3 deletions maintner/reclog/reclog.go
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,7 @@ func ForeachFileRecord(path string, fn RecordCallback) error {
return err
}
defer f.Close()
if err := ForeachRecord(f, fn); err != nil {
if err := ForeachRecord(f, 0, fn); err != nil {
return fmt.Errorf("error in %s: %v", path, err)
}
return nil
Expand All @@ -64,8 +64,9 @@ func ForeachFileRecord(path string, fn RecordCallback) error {
// ForeachRecord calls fn for each record in r.
// Calls to fn are made serially.
// If fn returns an error, iteration ends and that error is returned.
func ForeachRecord(r io.Reader, fn RecordCallback) error {
var off int64
// The startOffset be 0 if reading from the beginning of a file.
func ForeachRecord(r io.Reader, startOffset int64, fn RecordCallback) error {
off := startOffset
br := bufio.NewReader(r)
var buf bytes.Buffer
var hdrBuf bytes.Buffer
Expand Down

0 comments on commit ef7ac5c

Please sign in to comment.