gapid/replay/batcher.go - platform/tools/gpu - Git at Google

 // Copyright (C) 2016 The Android Open Source Project
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //      http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.

 package replay

 import (
 	"fmt"
 	"time"

 	"android.googlesource.com/platform/tools/gpu/client/gapir"
 	"android.googlesource.com/platform/tools/gpu/framework/benchmark"
 	"android.googlesource.com/platform/tools/gpu/framework/device"
 	"android.googlesource.com/platform/tools/gpu/framework/log"
 	"android.googlesource.com/platform/tools/gpu/gapid/atom"
 	"android.googlesource.com/platform/tools/gpu/gapid/config"
 	"android.googlesource.com/platform/tools/gpu/gapid/database"
 	"android.googlesource.com/platform/tools/gpu/gapid/gfxapi"
 	"android.googlesource.com/platform/tools/gpu/gapid/gfxapi/state"
 	"android.googlesource.com/platform/tools/gpu/gapid/replay/builder"
 	"android.googlesource.com/platform/tools/gpu/gapid/replay/executor"
 	"android.googlesource.com/platform/tools/gpu/gapid/service"
 )

 const maxBatchDelay = 250 * time.Millisecond

 type batcherContext struct {
 	Intent
 	Generator Generator
 	Config    Config
 }

 type job struct {
 	request Request
 	result  chan<- error
 }

 type batcher struct {
 	feed     chan job
 	context  batcherContext
 	database database.Database
 	device   gapir.Device
 }

 var (
 	generatorReplayCounter       = benchmark.GlobalCounters.Duration("batcher.send.generatorReplayTotalDuration")
 	builderBuildCounter          = benchmark.GlobalCounters.Duration("batcher.send.builderBuildTotalDuration")
 	executorExecuteCounter       = benchmark.GlobalCounters.Duration("batcher.send.executorExecuteTotalDuration")
 	batcherSendInvocationCounter = benchmark.GlobalCounters.Integer("batcher.send.invocations")
 )

 func (b *batcher) run(ctx log.Context) {
 	ctx = ctx.V("capture", b.context.Capture)

 	// Gather all the batchEntries that are added to feed within maxBatchDelay.
 	for j := range b.feed {
 		jobs := []job{j}
 		timeout := time.After(maxBatchDelay)
 	inner:
 		for {
 			select {
 			case j, ok := <-b.feed:
 				if !ok {
 					break inner
 				}
 				jobs = append(jobs, j)
 			case <-timeout:
 				break inner
 			}
 		}

 		// Batch formed. Trigger the replay.
 		requests := make([]Request, len(jobs))
 		for i, job := range jobs {
 			requests[i] = job.request
 		}

 		ctx.Info().Log("Replay batch")
 		err := b.send(ctx, requests)
 		for _, job := range jobs {
 			job.result <- err
 		}
 	}
 }

 // captureMemoryLayout returns the device memory layout of the capture from the
 // atoms. This function assumes there's an architecture atom at the beginning of
 // the capture. TODO: Replace this with a proper capture header containing
 // device and process information.
 func captureMemoryLayout(ctx log.Context, list *atom.List, db database.Database) *device.MemoryLayout {
 	s := state.New(ctx)
 	if len(list.Atoms) > 0 {
 		list.Atoms[0].Mutate(ctx, s, db, nil)
 	}
 	return &s.MemoryLayout
 }

 // findABI looks for the ABI with the matching memory layout, retuning it if an
 // exact match is found. If no matching ABI is found then nil is returned.
 func findABI(ml *device.MemoryLayout, abis []device.ABI) *device.ABI {
 	for _, abi := range abis {
 		if abi.MemoryLayout == *ml {
 			return &abi
 		}
 	}
 	return nil
 }

 func (b *batcher) send(ctx log.Context, requests []Request) (err error) {
 	batcherSendInvocationCounter.Increment()

 	device, err := b.device.Info(ctx)
 	if err != nil {
 		return ctx.WrapError(err, "Failed to get device information")
 	}
 	ctx = ctx.S("device", device.Name)

 	ctx.Info().Logf("Replaying...")

 	c, err := service.ResolveCapture(ctx, b.context.Capture, b.database)
 	if err != nil {
 		return ctx.WrapError(err, "Failed to load capture")
 	}

 	list, err := service.ResolveAtomList(ctx, c.Atoms, b.database)
 	if err != nil {
 		return ctx.WrapError(err, "Failed to load atom stream")
 	}

 	cml := captureMemoryLayout(ctx, list, b.database)
 	ctx = ctx.V("capture memory layout", cml)

 	if len(device.Configuration.ABIs) == 0 {
 		return ctx.AsError("Replay device doesn't list any ABIs")
 	}

 	replayABI := findABI(cml, device.Configuration.ABIs)
 	if replayABI == nil {
 		ctx.Info().Log("Replay device does not have a memory layout matching device used to trace")
 		replayABI = &device.Configuration.ABIs[0]
 	}
 	ctx = ctx.V("replay target ABI", replayABI)

 	builder := builder.New(replayABI.MemoryLayout)

 	out := &adapter{
 		state:   state.New(ctx),
 		db:      b.database,
 		builder: builder,
 	}

 	t0 := generatorReplayCounter.Start()
 	if err := b.context.Generator.Replay(
 		ctx,
 		b.context.Intent,
 		b.context.Config,
 		requests,
 		device,
 		*list,
 		out,
 		b.database); err != nil {

 		return ctx.WrapError(err, "Replay returned error")
 	}
 	generatorReplayCounter.Stop(t0)

 	if config.DebugReplay {
 		ctx.Print("Building payload...")
 	}

 	t0 = builderBuildCounter.Start()
 	payload, decoder, err := builder.Build(ctx)
 	if err != nil {
 		return ctx.WrapError(err, "Failed to build replay payload")
 	}
 	builderBuildCounter.Stop(t0)

 	defer func() {
 		caught := recover()
 		if err == nil && caught != nil {
 			err, _ = caught.(error)
 			if err == nil {
 				// If we are panicing, we always want an error to send.
 				err = fmt.Errorf("%s", caught)
 			}
 		}
 		if err != nil {
 			// An error was returned or thrown after the replay postbacks were requested.
 			// Inform each postback handler that they're not going to get data,
 			// to avoid chans blocking forever.
 			decoder(nil, err)
 		}
 		if caught != nil {
 			panic(caught)
 		}
 	}()

 	connection, err := b.device.Connect(ctx, replayABI)
 	if err != nil {
 		return ctx.WrapError(err, "Failed to connect to device")
 	}
 	defer connection.Close()

 	if config.DebugReplay {
 		ctx.Info().Log("Sending payload")
 	}

 	if Events.OnReplay != nil {
 		Events.OnReplay(b.device, b.context.Intent, b.context.Config, requests)
 	}

 	t0 = executorExecuteCounter.Start()
 	err = executor.Execute(
 		ctx,
 		payload,
 		decoder,
 		connection,
 		b.database,
 		replayABI.MemoryLayout,
 	)
 	executorExecuteCounter.Stop(t0)
 	return err
 }

 // adapter conforms to the the atom Writer interface, performing replay writes
 // on each atom.
 type adapter struct {
 	state   *gfxapi.State
 	db      database.Database
 	builder *builder.Builder
 }

 func (w *adapter) Write(ctx log.Context, i atom.ID, a atom.Atom) {
 	w.builder.BeginAtom(uint64(i))
 	if err := a.Mutate(ctx, w.state, w.db, w.builder); err == nil {
 		w.builder.CommitAtom()
 	} else {
 		w.builder.RevertAtom(err)
 		ctx.Warning().Logf("Failed to write atom %d (%T) for replay: %v", i, a, err)
 	}
 }
	// Copyright (C) 2016 The Android Open Source Project
	//
	// Licensed under the Apache License, Version 2.0 (the "License");
	// you may not use this file except in compliance with the License.
	// You may obtain a copy of the License at
	//
	// http://www.apache.org/licenses/LICENSE-2.0
	//
	// Unless required by applicable law or agreed to in writing, software
	// distributed under the License is distributed on an "AS IS" BASIS,
	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	// See the License for the specific language governing permissions and
	// limitations under the License.

	package replay

	import (
	"fmt"
	"time"

	"android.googlesource.com/platform/tools/gpu/client/gapir"
	"android.googlesource.com/platform/tools/gpu/framework/benchmark"
	"android.googlesource.com/platform/tools/gpu/framework/device"
	"android.googlesource.com/platform/tools/gpu/framework/log"
	"android.googlesource.com/platform/tools/gpu/gapid/atom"
	"android.googlesource.com/platform/tools/gpu/gapid/config"
	"android.googlesource.com/platform/tools/gpu/gapid/database"
	"android.googlesource.com/platform/tools/gpu/gapid/gfxapi"
	"android.googlesource.com/platform/tools/gpu/gapid/gfxapi/state"
	"android.googlesource.com/platform/tools/gpu/gapid/replay/builder"
	"android.googlesource.com/platform/tools/gpu/gapid/replay/executor"
	"android.googlesource.com/platform/tools/gpu/gapid/service"
	)

	const maxBatchDelay = 250 * time.Millisecond

	type batcherContext struct {
	Intent
	Generator Generator
	Config Config
	}

	type job struct {
	request Request
	result chan<- error
	}

	type batcher struct {
	feed chan job
	context batcherContext
	database database.Database
	device gapir.Device
	}

	var (
	generatorReplayCounter = benchmark.GlobalCounters.Duration("batcher.send.generatorReplayTotalDuration")
	builderBuildCounter = benchmark.GlobalCounters.Duration("batcher.send.builderBuildTotalDuration")
	executorExecuteCounter = benchmark.GlobalCounters.Duration("batcher.send.executorExecuteTotalDuration")
	batcherSendInvocationCounter = benchmark.GlobalCounters.Integer("batcher.send.invocations")
	)

	func (b *batcher) run(ctx log.Context) {
	ctx = ctx.V("capture", b.context.Capture)

	// Gather all the batchEntries that are added to feed within maxBatchDelay.
	for j := range b.feed {
	jobs := []job{j}
	timeout := time.After(maxBatchDelay)
	inner:
	for {
	select {
	case j, ok := <-b.feed:
	if !ok {
	break inner
	}
	jobs = append(jobs, j)
	case <-timeout:
	break inner
	}
	}

	// Batch formed. Trigger the replay.
	requests := make([]Request, len(jobs))
	for i, job := range jobs {
	requests[i] = job.request
	}

	ctx.Info().Log("Replay batch")
	err := b.send(ctx, requests)
	for _, job := range jobs {
	job.result <- err
	}
	}
	}

	// captureMemoryLayout returns the device memory layout of the capture from the
	// atoms. This function assumes there's an architecture atom at the beginning of
	// the capture. TODO: Replace this with a proper capture header containing
	// device and process information.
	func captureMemoryLayout(ctx log.Context, list atom.List, db database.Database) device.MemoryLayout {
	s := state.New(ctx)
	if len(list.Atoms) > 0 {
	list.Atoms[0].Mutate(ctx, s, db, nil)
	}
	return &s.MemoryLayout
	}

	// findABI looks for the ABI with the matching memory layout, retuning it if an
	// exact match is found. If no matching ABI is found then nil is returned.
	func findABI(ml device.MemoryLayout, abis []device.ABI) device.ABI {
	for _, abi := range abis {
	if abi.MemoryLayout == *ml {
	return &abi
	}
	}
	return nil
	}

	func (b *batcher) send(ctx log.Context, requests []Request) (err error) {
	batcherSendInvocationCounter.Increment()

	device, err := b.device.Info(ctx)
	if err != nil {
	return ctx.WrapError(err, "Failed to get device information")
	}
	ctx = ctx.S("device", device.Name)

	ctx.Info().Logf("Replaying...")

	c, err := service.ResolveCapture(ctx, b.context.Capture, b.database)
	if err != nil {
	return ctx.WrapError(err, "Failed to load capture")
	}

	list, err := service.ResolveAtomList(ctx, c.Atoms, b.database)
	if err != nil {
	return ctx.WrapError(err, "Failed to load atom stream")
	}

	cml := captureMemoryLayout(ctx, list, b.database)
	ctx = ctx.V("capture memory layout", cml)

	if len(device.Configuration.ABIs) == 0 {
	return ctx.AsError("Replay device doesn't list any ABIs")
	}

	replayABI := findABI(cml, device.Configuration.ABIs)
	if replayABI == nil {
	ctx.Info().Log("Replay device does not have a memory layout matching device used to trace")
	replayABI = &device.Configuration.ABIs[0]
	}
	ctx = ctx.V("replay target ABI", replayABI)

	builder := builder.New(replayABI.MemoryLayout)

	out := &adapter{
	state: state.New(ctx),
	db: b.database,
	builder: builder,
	}

	t0 := generatorReplayCounter.Start()
	if err := b.context.Generator.Replay(
	ctx,
	b.context.Intent,
	b.context.Config,
	requests,
	device,
	*list,
	out,
	b.database); err != nil {

	return ctx.WrapError(err, "Replay returned error")
	}
	generatorReplayCounter.Stop(t0)

	if config.DebugReplay {
	ctx.Print("Building payload...")
	}

	t0 = builderBuildCounter.Start()
	payload, decoder, err := builder.Build(ctx)
	if err != nil {
	return ctx.WrapError(err, "Failed to build replay payload")
	}
	builderBuildCounter.Stop(t0)

	defer func() {
	caught := recover()
	if err == nil && caught != nil {
	err, _ = caught.(error)
	if err == nil {
	// If we are panicing, we always want an error to send.
	err = fmt.Errorf("%s", caught)
	}
	}
	if err != nil {
	// An error was returned or thrown after the replay postbacks were requested.
	// Inform each postback handler that they're not going to get data,
	// to avoid chans blocking forever.
	decoder(nil, err)
	}
	if caught != nil {
	panic(caught)
	}
	}()

	connection, err := b.device.Connect(ctx, replayABI)
	if err != nil {
	return ctx.WrapError(err, "Failed to connect to device")
	}
	defer connection.Close()

	if config.DebugReplay {
	ctx.Info().Log("Sending payload")
	}

	if Events.OnReplay != nil {
	Events.OnReplay(b.device, b.context.Intent, b.context.Config, requests)
	}

	t0 = executorExecuteCounter.Start()
	err = executor.Execute(
	ctx,
	payload,
	decoder,
	connection,
	b.database,
	replayABI.MemoryLayout,
	)
	executorExecuteCounter.Stop(t0)
	return err
	}

	// adapter conforms to the the atom Writer interface, performing replay writes
	// on each atom.
	type adapter struct {
	state *gfxapi.State
	db database.Database
	builder *builder.Builder
	}

	func (w *adapter) Write(ctx log.Context, i atom.ID, a atom.Atom) {
	w.builder.BeginAtom(uint64(i))
	if err := a.Mutate(ctx, w.state, w.db, w.builder); err == nil {
	w.builder.CommitAtom()
	} else {
	w.builder.RevertAtom(err)
	ctx.Warning().Logf("Failed to write atom %d (%T) for replay: %v", i, a, err)
	}
	}