gapid/replay/manager.go

// 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"
	"sync"

	"android.googlesource.com/platform/tools/gpu/client/gapir"
	"android.googlesource.com/platform/tools/gpu/framework/log"
	"android.googlesource.com/platform/tools/gpu/gapid/database"
	"android.googlesource.com/platform/tools/gpu/gapid/service"
	"android.googlesource.com/platform/tools/gpu/gapid/service/path"
)

// Manager is used discover replay devices and to send replay requests to those
// discovered devices.
type Manager struct {
	database  database.Database
	discovery *gapir.Discovery
	batchers  map[batcherContext]*batcher
	mutex     sync.Mutex // guards batchers
}

func (m *Manager) getBatchStream(ctx log.Context, bContext batcherContext) (chan<- job, error) {
	m.mutex.Lock()
	defer m.mutex.Unlock()

	pc := &path.Capture{ID: bContext.Capture}
	ctx = service.WithCaptureAndDatabase(ctx, pc, m.database)

	// TODO: This accumulates batchers with running go-routines forever.
	// Rework to free the batcher after execution.
	b, found := m.batchers[bContext]
	if !found {
		ctx.Info().V("capture", bContext.Capture).Log("New batch stream")
		device := m.discovery.Device(bContext.Device)
		if device == nil {
			return nil, fmt.Errorf("Unknown device %v", bContext.Device)
		}
		b = &batcher{
			feed:     make(chan job, 64),
			context:  bContext,
			database: m.database,
			device:   device,
		}
		m.batchers[bContext] = b
		go b.run(ctx)
	}
	return b.feed, nil
}

// New returns a new Manager instance using the database db.
func New(ctx log.Context, d database.Database) *Manager {
	return &Manager{
		database:  d,
		discovery: &gapir.Discovery{},
		batchers:  make(map[batcherContext]*batcher),
	}
}

// Replay requests that req is to be performed on the device described by intent,
// using the capture described by intent. Replay requests made with configs that
// have equality (==) will likely be batched into the same replay pass.
func (m *Manager) Replay(ctx log.Context, intent Intent, cfg Config, req Request, generator Generator) error {
	ctx.Info().Log("Replay request")
	batch, err := m.getBatchStream(ctx, batcherContext{
		Intent:    intent,
		Generator: generator,
		Config:    cfg,
	})
	if err != nil {
		return err
	}
	res := make(chan error, 1)
	batch <- job{request: req, result: res}
	return <-res
}

// Discovery returns the device discovery being used by this manager.
func (m *Manager) Discovery() *gapir.Discovery {
	return m.discovery
}