func TestCreateTimestampFile(t *testing.T) {
	realDir := filepath.Join(os.TempDir(), "util_test")
	util.Mkdir(realDir, 0755)
	defer util.RemoveAll(realDir)
	timestampFilePath := filepath.Join(realDir, TIMESTAMP_FILE_NAME)
	if err := CreateTimestampFile(realDir); err != nil {
		t.Errorf("Unexpected error: %s", err)
	}
	// Assert timestamp file exists.
	if _, err := os.Stat(timestampFilePath); err != nil {
		t.Errorf("Timestamp file %s was not created!", timestampFilePath)
	}
	// Assert timestamp file contains int64.
	fileContent, err := ioutil.ReadFile(timestampFilePath)
	if err != nil {
		t.Errorf("Unexpected error: %s", err)
	}
	if _, err := strconv.ParseInt(string(fileContent), 10, 64); err != nil {
		t.Errorf("Unexpected value in %s: %s", timestampFilePath, err)
	}

	// Assert error returned when specified dir does not exist.
	nonexistantDir := filepath.Join(os.TempDir(), "util_test_nonexistant")
	util.RemoveAll(nonexistantDir)
	if err := CreateTimestampFile(nonexistantDir); err != nil {
		// Expected error
	} else {
		t.Error("Unexpected lack of error")
	}
}

// Will need a local valid google_storage_token.data file with read write access
// to run the below test.
func Auth_TestDownloadWorkerArtifacts(t *testing.T) {
	testPagesetsDirName := filepath.Join("unit-tests", "util", "page_sets")
	client, _ := GetOAuthClient()
	gs, err := NewGsUtil(client)
	if err != nil {
		t.Errorf("Unexpected error: %s", err)
	}

	tmpDir := filepath.Join(os.TempDir(), "util_test")
	StorageDir = tmpDir
	defer util.RemoveAll(tmpDir)
	if err := gs.DownloadWorkerArtifacts(testPagesetsDirName, "10k", 1); err != nil {
		t.Errorf("Unexpected error: %s", err)
	}

	// Examine contents of the local directory.
	localDir := filepath.Join(tmpDir, testPagesetsDirName, "10k")
	files, err := ioutil.ReadDir(localDir)
	if err != nil {
		t.Errorf("Unexpected error: %s", err)
	}
	assert.Equal(t, 3, len(files))
	assert.Equal(t, "TIMESTAMP", files[0].Name())
	assert.Equal(t, "alexa1-1.py", files[1].Name())
	assert.Equal(t, "alexa2-2.py", files[2].Name())
}

func testRasterizer(t *testing.T, rasterizer Rasterizer, expectation string) {
	assert.True(t, rasterizer.Enabled(), "%s.Enabled() failed.", rasterizer.String())

	testDataDir, err := testutils.TestDataDir()
	assert.Nil(t, err, "TestDataDir missing: %v", err)

	tempDir, err := ioutil.TempDir("", "pdf_test_")
	assert.Nil(t, err, "ioutil.TempDir failed")
	defer util.RemoveAll(tempDir)

	pdfSrcPath := path.Join(testDataDir, "minimal.pdf")
	assert.True(t, fileutil.FileExists(pdfSrcPath), "Path '%s' does not exist", pdfSrcPath)
	pdfInputPath := path.Join(tempDir, "minimal.pdf")

	err = os.Symlink(pdfSrcPath, pdfInputPath)
	assert.Nil(t, err, "Symlink failed")
	assert.True(t, fileutil.FileExists(pdfInputPath), "Path '%s' does not exist", pdfInputPath)

	outputFileName := path.Join(tempDir, "test.png")

	badPath := path.Join(tempDir, "this_file_should_really_not_exist.pdf")

	if err := rasterizer.Rasterize(badPath, outputFileName); err == nil {
		t.Errorf(": Got '%v' Want '%v'", err, nil)
	}

	if err := rasterizer.Rasterize(pdfInputPath, outputFileName); err != nil {
		t.Errorf(": Got '%v' Want '!nil'", err)
	}

	expectedOutput := path.Join(testDataDir, expectation)
	assert.True(t, filesEqual(outputFileName, expectedOutput), "png output not correct")
}

// processResult rasterizes a single PDF result and returns a set of new results.
func (xformer *pdfXformer) processResult(res goldingester.Result) []*goldingester.Result {
	rasterizedResults := []*goldingester.Result{}
	resultMap, found := xformer.results[res.Digest]
	if found {
		// Skip rasterizion steps: big win.
		for index, rasterizer := range xformer.rasterizers {
			digest, ok := resultMap[index]
			if ok {
				rasterizedResults = append(rasterizedResults,
					newResult(res.Key, rasterizer.String(), digest))
			} else {
				glog.Errorf("missing rasterizer %s on %s", rasterizer.String(), res.Digest)
			}
		}
		return rasterizedResults
	}

	tempdir, err := xformer.makeTmpDir()
	if err != nil {
		glog.Errorf("error making temp directory: %s", err)
		return rasterizedResults
	}
	defer util.RemoveAll(tempdir)
	pdfPath := path.Join(tempdir, fmt.Sprintf("%s.pdf", res.Digest))
	objectName := fmt.Sprintf("%s/%s.pdf", *storageImagesDirectory, res.Digest)
	storageURL := fmt.Sprintf("gs://%s/%s", *storageBucket, objectName)
	object, err := xformer.client.storageService.Objects.Get(*storageBucket, objectName).Do()
	if err != nil {
		glog.Errorf("unable to find %s: %s", storageURL, err)
		return []*goldingester.Result{}
	}
	pdfData, _, err := gsFetch(object, xformer.client)
	if err != nil {
		glog.Errorf("unable to retrieve %s: %s", storageURL, err)
		return []*goldingester.Result{}
	}
	err = writeTo(pdfPath, &pdfData)
	if err != nil {
		glog.Errorf("unable to write file %s: %s", pdfPath, err)
		return []*goldingester.Result{}
	}
	if !isPDF(pdfPath) {
		glog.Errorf("%s is not a PDF", objectName)
		return []*goldingester.Result{}
	}
	resultMap = map[int]string{}
	for index, rasterizer := range xformer.rasterizers {
		digest, err := xformer.rasterizeOnce(pdfPath, index)
		if err != nil {
			glog.Errorf("rasterizer %s failed on %s.pdf: %s", rasterizer, res.Digest, err)
			continue
		}
		rasterizedResults = append(rasterizedResults,
			newResult(res.Key, rasterizer.String(), digest))
		resultMap[index] = digest
	}
	xformer.results[res.Digest] = resultMap
	return rasterizedResults
}

func TestIngester(t *testing.T) {
	defer util.RemoveAll(LOCAL_STATUS_DIR)

	now := time.Now()
	beginningOfTime := now.Add(-time.Hour * 24 * 10).Unix()
	const totalCommits = 100

	// Instantiate mock VCS and the source.
	vcs := getVCS(beginningOfTime, now.Unix(), totalCommits)
	hashes := vcs.From(time.Unix(0, 0))
	assert.Equal(t, totalCommits, len(hashes))
	for _, h := range hashes {
		assert.NotEqual(t, "", h)
	}

	sources := []Source{MockSource(t, vcs)}

	// Instantiate the mock processor.
	collected := map[string]int{}
	var mutex sync.Mutex

	processFn := func(result ResultFileLocation) error {
		mutex.Lock()
		defer mutex.Unlock()
		collected[result.Name()] += 1
		return nil
	}

	finishFn := func() error { return nil }
	processor := MockProcessor(processFn, finishFn)

	// Instantiate ingesterConf
	conf := &sharedconfig.IngesterConfig{
		RunEvery:  sharedconfig.TomlDuration{Duration: 1 * time.Second},
		NCommits:  50,
		MinDays:   3,
		StatusDir: LOCAL_STATUS_DIR,
	}

	// Instantiate ingester and start it.
	ingester, err := NewIngester("test-ingester", conf, vcs, sources, processor)
	assert.Nil(t, err)
	ingester.Start()

	// Give it enough time to run a few ingestion cycles and to shut down.
	time.Sleep(5 * time.Second)
	ingester.stop()
	time.Sleep(5 * time.Second)

	assert.Equal(t, totalCommits/2, len(collected))
	for _, count := range collected {
		assert.Equal(t, 1, count)
	}
}

func TestAreTimestampsEqual(t *testing.T) {
	gs, err := NewGsUtil(util.NewTimeoutClient())
	if err != nil {
		t.Errorf("Unexpected error: %s", err)
	}

	tmpDir := filepath.Join(os.TempDir(), "util_test")
	util.Mkdir(tmpDir, 0777)
	defer util.RemoveAll(tmpDir)

	f, err := os.Create(filepath.Join(tmpDir, TIMESTAMP_FILE_NAME))
	if err != nil {
		t.Errorf("Unexpected error: %s", err)
	}
	defer util.Close(f)

	// Test with matching timestamps.
	if _, err := f.WriteString(GS_TEST_TIMESTAMP_VALUE); err != nil {
		t.Errorf("Unexpected error: %s", err)
	}
	result1, err := gs.AreTimeStampsEqual(tmpDir, "unit-tests/util/")
	if err != nil {
		t.Errorf("Unexpected error: %s", err)
	}
	assert.True(t, result1)

	// Test with differing timestamps.
	if _, err := f.WriteString(GS_TEST_TIMESTAMP_VALUE); err != nil {
		t.Errorf("Unexpected error: %s", err)
	}
	result2, err := gs.AreTimeStampsEqual(tmpDir, "unit-tests/util/")
	if err != nil {
		t.Errorf("Unexpected error: %s", err)
	}
	assert.False(t, result2)

	// Test with Google Storage timestamp missing.
	result3, err := gs.AreTimeStampsEqual(tmpDir, "unit-tests/util/dummy_name/")
	if err == nil {
		t.Error("Expected an error")
	}
	assert.False(t, result3)

	// Test with local timestamp missing.
	result4, err := gs.AreTimeStampsEqual(tmpDir+"dummy_name", "unit-tests/util/")
	if err == nil {
		t.Error("Expected an error")
	}
	assert.False(t, result4)
}

// downloadAndExecPython downloads a Python script from the Repo, executes it, and
// parses its output in JSON format into the given destination.
func downloadAndExecPython(r *gitiles.Repo, dst interface{}, srcPath string, workdir string) error {
	destDir, err := ioutil.TempDir(workdir, "gitiles")
	if err != nil {
		return err
	}
	defer util.RemoveAll(destDir)
	destPath := path.Join(destDir, path.Base(srcPath))
	if err := r.DownloadFile(srcPath, destPath); err != nil {
		return err
	}
	output, err := exec.Command("python", destPath).Output()
	if err != nil {
		return err
	}
	if err := json.Unmarshal(output, dst); err != nil {
		return err
	}
	return nil
}

func main() {
	defer common.LogPanic()
	worker_common.Init()
	defer util.TimeTrack(time.Now(), "Fixing archives")
	defer glog.Flush()

	// Create the task file so that the master knows this worker is still busy.
	skutil.LogErr(util.CreateTaskFile(util.ACTIVITY_FIXING_ARCHIVES))
	defer util.DeleteTaskFile(util.ACTIVITY_FIXING_ARCHIVES)

	if *pagesetType == "" {
		glog.Error("Must specify --pageset_type")
		return
	}
	if *chromiumBuild == "" {
		glog.Error("Must specify --chromium_build")
		return
	}
	if *runID == "" {
		glog.Error("Must specify --run_id")
		return
	}

	// Reset the local chromium checkout.
	if err := util.ResetCheckout(util.ChromiumSrcDir); err != nil {
		glog.Errorf("Could not reset %s: %s", util.ChromiumSrcDir, err)
		return
	}
	// Sync the local chromium checkout.
	if err := util.SyncDir(util.ChromiumSrcDir); err != nil {
		glog.Errorf("Could not gclient sync %s: %s", util.ChromiumSrcDir, err)
		return
	}

	// Instantiate GsUtil object.
	gs, err := util.NewGsUtil(nil)
	if err != nil {
		glog.Error(err)
		return
	}

	// Download the specified chromium build.
	if err := gs.DownloadChromiumBuild(*chromiumBuild); err != nil {
		glog.Error(err)
		return
	}
	// Delete the chromium build to save space when we are done.
	defer skutil.RemoveAll(filepath.Join(util.ChromiumBuildsDir, *chromiumBuild))
	chromiumBinary := filepath.Join(util.ChromiumBuildsDir, *chromiumBuild, util.BINARY_CHROME)

	// Download pagesets if they do not exist locally.
	if err := gs.DownloadWorkerArtifacts(util.PAGESETS_DIR_NAME, *pagesetType, *workerNum); err != nil {
		glog.Error(err)
		return
	}
	pathToPagesets := filepath.Join(util.PagesetsDir, *pagesetType)

	// Download archives if they do not exist locally.
	if err := gs.DownloadWorkerArtifacts(util.WEB_ARCHIVES_DIR_NAME, *pagesetType, *workerNum); err != nil {
		glog.Error(err)
		return
	}

	// Establish output paths.
	localOutputDir := filepath.Join(util.StorageDir, util.FixArchivesRunsDir, *runID)
	skutil.RemoveAll(localOutputDir)
	skutil.MkdirAll(localOutputDir, 0700)
	defer skutil.RemoveAll(localOutputDir)

	// Construct path to the ct_run_benchmark python script.
	_, currentFile, _, _ := runtime.Caller(0)
	pathToPyFiles := filepath.Join(
		filepath.Dir((filepath.Dir(filepath.Dir(filepath.Dir(currentFile))))),
		"py")

	timeoutSecs := util.PagesetTypeToInfo[*pagesetType].RunChromiumPerfTimeoutSecs
	fileInfos, err := ioutil.ReadDir(pathToPagesets)
	if err != nil {
		glog.Errorf("Unable to read the pagesets dir %s: %s", pathToPagesets, err)
		return
	}

	// Location of the WPR logs.
	wprLogs := filepath.Join(util.ChromiumSrcDir, "webpagereplay_logs", "logs.txt")
	// Slice that will contain
	inconsistentArchives := []string{}

	// Loop through all pagesets.
	for _, fileInfo := range fileInfos {
		benchmarkResults := []float64{}
		resourceMissingCounts := []int{}
		pagesetBaseName := filepath.Base(fileInfo.Name())
		if pagesetBaseName == util.TIMESTAMP_FILE_NAME || filepath.Ext(pagesetBaseName) == ".pyc" {
			// Ignore timestamp files and .pyc files.
			continue
		}

		// Convert the filename into a format consumable by the run_benchmarks
		// binary.
		pagesetName := strings.TrimSuffix(pagesetBaseName, filepath.Ext(pagesetBaseName))
//.........这里部分代码省略.........

func main() {
	defer common.LogPanic()
	common.Init()
	defer util.TimeTrack(time.Now(), "Creating Pagesets")
	defer glog.Flush()
	// Create the task file so that the master knows this worker is still busy.
	skutil.LogErr(util.CreateTaskFile(util.ACTIVITY_CREATING_PAGESETS))
	defer util.DeleteTaskFile(util.ACTIVITY_CREATING_PAGESETS)

	// Delete and remake the local pagesets directory.
	pathToPagesets := filepath.Join(util.PagesetsDir, *pagesetType)
	skutil.RemoveAll(pathToPagesets)
	skutil.MkdirAll(pathToPagesets, 0700)

	// Get info about the specified pageset type.
	pagesetTypeInfo := util.PagesetTypeToInfo[*pagesetType]
	csvSource := pagesetTypeInfo.CSVSource
	numPages := pagesetTypeInfo.NumPages
	userAgent := pagesetTypeInfo.UserAgent

	// Download the CSV file from Google Storage to a tmp location.
	gs, err := util.NewGsUtil(nil)
	if err != nil {
		glog.Error(err)
		return
	}
	respBody, err := gs.GetRemoteFileContents(csvSource)
	if err != nil {
		glog.Error(err)
		return
	}
	defer skutil.Close(respBody)
	csvFile := filepath.Join(os.TempDir(), filepath.Base(csvSource))
	out, err := os.Create(csvFile)
	if err != nil {
		glog.Errorf("Unable to create file %s: %s", csvFile, err)
		return
	}
	defer skutil.Close(out)
	defer skutil.Remove(csvFile)
	if _, err = io.Copy(out, respBody); err != nil {
		glog.Error(err)
		return
	}

	// Figure out which pagesets this worker should generate.
	numPagesPerSlave := numPages / util.NUM_WORKERS
	startNum := (*workerNum-1)*numPagesPerSlave + 1
	endNum := *workerNum * numPagesPerSlave

	// Construct path to the create_page_set.py python script.
	_, currentFile, _, _ := runtime.Caller(0)
	createPageSetScript := filepath.Join(
		filepath.Dir((filepath.Dir(filepath.Dir(filepath.Dir(currentFile))))),
		"py", "create_page_set.py")

	// Execute the create_page_set.py python script.
	timeoutSecs := util.PagesetTypeToInfo[*pagesetType].CreatePagesetsTimeoutSecs
	for currNum := startNum; currNum <= endNum; currNum++ {
		args := []string{
			createPageSetScript,
			"-s", strconv.Itoa(currNum),
			"-e", strconv.Itoa(currNum),
			"-c", csvFile,
			"-p", *pagesetType,
			"-u", userAgent,
			"-o", pathToPagesets,
		}
		if err := util.ExecuteCmd("python", args, []string{}, time.Duration(timeoutSecs)*time.Second, nil, nil); err != nil {
			glog.Error(err)
			return
		}
	}
	// Write timestamp to the pagesets dir.
	skutil.LogErr(util.CreateTimestampFile(pathToPagesets))

	// Upload pagesets dir to Google Storage.
	if err := gs.UploadWorkerArtifacts(util.PAGESETS_DIR_NAME, *pagesetType, *workerNum); err != nil {
		glog.Error(err)
		return
	}
}

// runID is the unique id of the current run (typically requester + timestamp).
// targetPlatform is the platform the benchmark will run on (Android / Linux ).
// chromiumHash is the hash the checkout should be synced to. If not specified then
// Chromium's Tot hash is used.
// skiaHash is the hash the checkout should be synced to. If not specified then
// Skia's LKGR hash is used (the hash in Chromium's DEPS file).
// applyPatches if true looks for Chromium/Blink/Skia patches in the temp dir and
// runs once with the patch applied and once without the patch applied.
func CreateChromiumBuild(runID, targetPlatform, chromiumHash, skiaHash string, applyPatches bool) (string, string, error) {
	// Determine which build dir and fetch target to use.
	var chromiumBuildDir, fetchTarget string
	if targetPlatform == "Android" {
		chromiumBuildDir = filepath.Join(ChromiumBuildsDir, "android_base")
		fetchTarget = "android"
	} else if targetPlatform == "Linux" {
		chromiumBuildDir = filepath.Join(ChromiumBuildsDir, "linux_base")
		fetchTarget = "chromium"
	} else {
		return "", "", fmt.Errorf("Unrecognized target_platform %s", targetPlatform)
	}
	util.MkdirAll(chromiumBuildDir, 0700)

	// Find which Chromium commit hash should be used.
	var err error
	if chromiumHash == "" {
		chromiumHash, err = getChromiumHash()
		if err != nil {
			return "", "", fmt.Errorf("Error while finding Chromium's Hash: %s", err)
		}
	}

	// Find which Skia commit hash should be used.
	if skiaHash == "" {
		skiaHash, err = getSkiaHash()
		if err != nil {
			return "", "", fmt.Errorf("Error while finding Skia's Hash: %s", err)
		}
	}

	// Run chromium sync command using the above commit hashes.
	// Construct path to the sync_skia_in_chrome python script.
	_, currentFile, _, _ := runtime.Caller(0)
	pathToPyFiles := filepath.Join(
		filepath.Dir((filepath.Dir(filepath.Dir(currentFile)))),
		"py")
	syncArgs := []string{
		filepath.Join(pathToPyFiles, "sync_skia_in_chrome.py"),
		"--destination=" + chromiumBuildDir,
		"--fetch_target=" + fetchTarget,
		"--chrome_revision=" + chromiumHash,
		"--skia_revision=" + skiaHash,
	}
	if err := ExecuteCmd("python", syncArgs, []string{}, 2*time.Hour, nil, nil); err != nil {
		glog.Warning("There was an error. Deleting base directory and trying again.")
		util.RemoveAll(chromiumBuildDir)
		util.MkdirAll(chromiumBuildDir, 0700)
		if err := ExecuteCmd("python", syncArgs, []string{}, 2*time.Hour, nil, nil); err != nil {
			return "", "", fmt.Errorf("There was an error checking out chromium %s + skia %s: %s", chromiumHash, skiaHash, err)
		}
	}

	// Make sure we are starting from a clean slate.
	if err := resetChromiumCheckout(filepath.Join(chromiumBuildDir, "src")); err != nil {
		return "", "", fmt.Errorf("Could not reset the chromium checkout in %s: %s", chromiumBuildDir, err)
	}
	googleStorageDirName := fmt.Sprintf("%s-%s-%s", getTruncatedHash(chromiumHash), getTruncatedHash(skiaHash), runID)
	if runID == "" {
		// Do not include the runID in the dir name if it is not specified.
		googleStorageDirName = fmt.Sprintf("%s-%s", getTruncatedHash(chromiumHash), getTruncatedHash(skiaHash))
	}
	if applyPatches {
		if err := applyRepoPatches(filepath.Join(chromiumBuildDir, "src"), runID); err != nil {
			return "", "", fmt.Errorf("Could not apply patches in the chromium checkout in %s: %s", chromiumBuildDir, err)
		}
		// Add "try" prefix and "withpatch" suffix.
		googleStorageDirName = fmt.Sprintf("try-%s-withpatch", googleStorageDirName)
	}
	// Build chromium.
	if err := buildChromium(chromiumBuildDir, targetPlatform); err != nil {
		return "", "", fmt.Errorf("There was an error building chromium %s + skia %s: %s", chromiumHash, skiaHash, err)
	}

	// Upload to Google Storage.
	gs, err := NewGsUtil(nil)
	if err != nil {
		return "", "", fmt.Errorf("Could not create GS object: %s", err)
	}
	if err := uploadChromiumBuild(filepath.Join(chromiumBuildDir, "src", "out", "Release"), filepath.Join(CHROMIUM_BUILDS_DIR_NAME, googleStorageDirName), targetPlatform, gs); err != nil {
		return "", "", fmt.Errorf("There was an error uploaded the chromium build dir %s: %s", filepath.Join(chromiumBuildDir, "src", "out", "Release"), err)
	}

	// Check for the applypatch flag and reset and then build again and copy to
	// google storage.
	if applyPatches {
		// Now build chromium without the patches and upload it to Google Storage.

		// Make sure we are starting from a clean slate.
		if err := resetChromiumCheckout(filepath.Join(chromiumBuildDir, "src")); err != nil {
			return "", "", fmt.Errorf("Could not reset the chromium checkout in %s: %s", chromiumBuildDir, err)
		}
//.........这里部分代码省略.........

func main() {
	common.Init()
	webhook.MustInitRequestSaltFromFile(util.WebhookRequestSaltPath)

	// Send start email.
	emailsArr := util.ParseEmails(*emails)
	emailsArr = append(emailsArr, util.CtAdmins...)
	if len(emailsArr) == 0 {
		glog.Error("At least one email address must be specified")
		return
	}
	skutil.LogErr(util.SendTaskStartEmail(emailsArr, "Skia correctness"))
	// Ensure webapp is updated and email is sent even if task fails.
	defer updateWebappTask()
	defer sendEmail(emailsArr)
	// Cleanup tmp files after the run.
	defer util.CleanTmpDir()
	// Finish with glog flush and how long the task took.
	defer util.TimeTrack(time.Now(), "Running skia correctness task on workers")
	defer glog.Flush()

	if *pagesetType == "" {
		glog.Error("Must specify --pageset_type")
		return
	}
	if *chromiumBuild == "" {
		glog.Error("Must specify --chromium_build")
		return
	}
	if *runID == "" {
		glog.Error("Must specify --run_id")
		return
	}

	// Instantiate GsUtil object.
	gs, err := util.NewGsUtil(nil)
	if err != nil {
		glog.Error(err)
		return
	}
	remoteOutputDir := filepath.Join(util.SkiaCorrectnessRunsDir, *runID)

	// Copy the patch to Google Storage.
	patchName := *runID + ".patch"
	patchRemoteDir := filepath.Join(remoteOutputDir, "patches")
	if err := gs.UploadFile(patchName, os.TempDir(), patchRemoteDir); err != nil {
		glog.Errorf("Could not upload %s to %s: %s", patchName, patchRemoteDir, err)
		return
	}
	skiaPatchLink = util.GS_HTTP_LINK + filepath.Join(util.GS_BUCKET_NAME, patchRemoteDir, patchName)

	// Run the run_skia_correctness script on all workers.
	runSkiaCorrCmdTemplate := "DISPLAY=:0 run_skia_correctness --worker_num={{.WorkerNum}} --log_dir={{.LogDir}} " +
		"--pageset_type={{.PagesetType}} --chromium_build={{.ChromiumBuild}} --run_id={{.RunID}} " +
		"--render_pictures_args=\"{{.RenderPicturesArgs}}\" --gpu_nopatch_run={{.GpuNoPatchRun}} " +
		"--gpu_withpatch_run={{.GpuWithPatchRun}};"
	runSkiaCorrTemplateParsed := template.Must(template.New("run_skia_correctness_cmd").Parse(runSkiaCorrCmdTemplate))
	runSkiaCorrCmdBytes := new(bytes.Buffer)
	if err := runSkiaCorrTemplateParsed.Execute(runSkiaCorrCmdBytes, struct {
		WorkerNum          string
		LogDir             string
		PagesetType        string
		ChromiumBuild      string
		RunID              string
		RenderPicturesArgs string
		GpuNoPatchRun      string
		GpuWithPatchRun    string
	}{
		WorkerNum:          util.WORKER_NUM_KEYWORD,
		LogDir:             util.GLogDir,
		PagesetType:        *pagesetType,
		ChromiumBuild:      *chromiumBuild,
		RunID:              *runID,
		RenderPicturesArgs: *renderPicturesArgs,
		GpuNoPatchRun:      strconv.FormatBool(*gpuNoPatchRun),
		GpuWithPatchRun:    strconv.FormatBool(*gpuWithPatchRun),
	}); err != nil {
		glog.Errorf("Failed to execute template: %s", err)
		return
	}
	cmd := []string{
		fmt.Sprintf("cd %s;", util.CtTreeDir),
		"git pull;",
		"make all;",
		// The main command that runs run_skia_correctness on all workers.
		runSkiaCorrCmdBytes.String(),
	}
	if _, err := util.SSH(strings.Join(cmd, " "), util.Slaves, 4*time.Hour); err != nil {
		glog.Errorf("Error while running cmd %s: %s", cmd, err)
		return
	}

	localOutputDir := filepath.Join(util.StorageDir, util.SkiaCorrectnessRunsDir, *runID)
	localSummariesDir := filepath.Join(localOutputDir, "summaries")
	skutil.MkdirAll(localSummariesDir, 0700)
	defer skutil.RemoveAll(filepath.Join(util.StorageDir, util.SkiaCorrectnessRunsDir))
	// Copy outputs from all slaves locally.
	for i := 0; i < util.NUM_WORKERS; i++ {
		workerNum := i + 1
		workerLocalOutputPath := filepath.Join(localSummariesDir, fmt.Sprintf("slave%d", workerNum)+".json")
//.........这里部分代码省略.........

func (p *pdfProcessor) rasterizeAndUpload(dmResultName string, dmResults *goldingestion.DMResults, pdfResults []*goldingestion.Result) error {
	processedResults := make([]*goldingestion.Result, 0, len(pdfResults)*len(p.rasterizers))

	// Create a temporary directory to hold the rastered images.
	tempDir, err := ioutil.TempDir(p.pdfCacheDir, "pdfingestion")
	if err != nil {
		return err
	}
	defer util.RemoveAll(tempDir)

	// Go through all results that generated PDF files.
	for resultIdx, result := range pdfResults {
		// Fetch the PDF file if it's not in the cache.
		pdfFileName := fmt.Sprintf("%s.%s", result.Digest, PDF_EXT)
		pdfPath := filepath.Join(p.pdfCacheDir, pdfFileName)
		if !fileutil.FileExists(pdfPath) {
			if err = p.download(p.inImagesBucket, p.inImagesDir, pdfFileName, pdfPath); err != nil {
				glog.Errorf("Unable to retrieve image: %s. Error: %s", pdfFileName, err)
				continue
			}
		}

		// Generate an image for each rasterizer.
		for rasterIdx, rasterizer := range p.rasterizers {
			tempName := filepath.Join(tempDir, fmt.Sprintf("rastering_%d_%d.%s", resultIdx, rasterIdx, PNG_EXT))
			err := rasterizer.Rasterize(pdfPath, tempName)
			if err != nil {
				glog.Errorf("Rasterizing %s with %s failed: %s", filepath.Base(pdfPath), rasterizer.String(), err)
				continue
			}

			// Open the generated image and calculate the MD5.
			file, err := os.Open(tempName)
			if err != nil {
				glog.Errorf("Unable to open generated image: %s", err)
				continue
			}

			var buf bytes.Buffer
			md5, err := util.MD5FromReader(file, &buf)
			if err != nil {
				glog.Errorf("Unable to calculate MD5 hash of file %s. Got error: %s", tempName, err)
				continue
			}
			digest := hex.EncodeToString(md5)
			uploadFileName := fmt.Sprintf("%s.%s", digest, PNG_EXT)
			if err := p.upload(p.outImagesBucket, p.outImagesDir, uploadFileName, bytes.NewBuffer(buf.Bytes())); err != nil {
				glog.Errorf("Unable to upload file %s. Error: %s", uploadFileName, err)
				continue
			}

			// Update the result and add it to the successfully processed results.
			result.Key["rasterizer"] = rasterizer.String()
			result.Digest = digest
			result.Options["ext"] = PNG_EXT
			processedResults = append(processedResults, result)
		}
	}

	// If we have no processed results we consider it an error.
	if len(processedResults) == 0 {
		return fmt.Errorf("No input image was processed successfully.")
	}

	// Replace the old results in the original result and write it to the cloud.
	dmResults.Results = processedResults
	jsonBytes, err := json.MarshalIndent(dmResults, "", "    ")
	if err != nil {
		return fmt.Errorf("Unable to encode JSON: %s", err)
	}

	return p.upload(p.outJsonBucket, p.outJsonDir, dmResultName, bytes.NewBuffer(jsonBytes))
}

func main() {
	common.Init()
	defer util.TimeTrack(time.Now(), "Running Skia Correctness")
	defer glog.Flush()

	if *chromiumBuild == "" {
		glog.Error("Must specify --chromium_build")
		return
	}
	if *runID == "" {
		glog.Error("Must specify --run_id")
		return
	}

	// Create the task file so that the master knows this worker is still busy.
	skutil.LogErr(util.CreateTaskFile(util.ACTIVITY_RUNNING_SKIA_CORRECTNESS))
	defer util.DeleteTaskFile(util.ACTIVITY_RUNNING_SKIA_CORRECTNESS)

	// Establish output paths.
	localOutputDir := filepath.Join(util.StorageDir, util.SkiaCorrectnessRunsDir, *runID)
	skutil.RemoveAll(filepath.Join(util.StorageDir, util.SkiaCorrectnessRunsDir))
	skutil.MkdirAll(localOutputDir, 0700)
	defer skutil.RemoveAll(localOutputDir)
	remoteOutputDir := filepath.Join(util.SkiaCorrectnessRunsDir, *runID, fmt.Sprintf("slave%d", *workerNum))

	// Instantiate GsUtil object.
	gs, err := util.NewGsUtil(nil)
	if err != nil {
		glog.Error(err)
		return
	}

	// Download SKPs if they do not exist locally.
	if err := gs.DownloadWorkerArtifacts(util.SKPS_DIR_NAME, filepath.Join(*pagesetType, *chromiumBuild), *workerNum); err != nil {
		glog.Error(err)
		return
	}
	localSkpsDir := filepath.Join(util.SkpsDir, *pagesetType, *chromiumBuild)

	// Download the Skia patch for this run from Google storage.
	patchName := *runID + ".patch"
	patchLocalPath := filepath.Join(os.TempDir(), patchName)
	remoteDir := filepath.Join(util.SkiaCorrectnessRunsDir, *runID)
	patchRemotePath := filepath.Join(remoteDir, "patches", patchName)
	respBody, err := gs.GetRemoteFileContents(patchRemotePath)
	if err != nil {
		glog.Errorf("Could not fetch %s: %s", patchRemotePath, err)
		return
	}
	defer skutil.Close(respBody)
	out, err := os.Create(patchLocalPath)
	if err != nil {
		glog.Errorf("Unable to create file %s: %s", patchLocalPath, err)
		return
	}
	defer skutil.Close(out)
	defer skutil.Remove(patchLocalPath)
	if _, err = io.Copy(out, respBody); err != nil {
		glog.Error(err)
		return
	}

	// Apply the patch to a clean checkout and run render_pictures.
	// Reset Skia tree.
	skutil.LogErr(util.ResetCheckout(util.SkiaTreeDir))
	// Sync Skia tree.
	skutil.LogErr(util.SyncDir(util.SkiaTreeDir))
	// Apply Skia patch.
	file, _ := os.Open(patchLocalPath)
	fileInfo, _ := file.Stat()
	// It is a valid patch only if it is more than 10 bytes.
	if fileInfo.Size() > 10 {
		glog.Info("Attempting to apply %s to %s", patchLocalPath, util.SkiaTreeDir)
		if err := util.ApplyPatch(patchLocalPath, util.SkiaTreeDir); err != nil {
			glog.Errorf("Could not apply patch %s to %s: %s", patchLocalPath, util.SkiaTreeDir, err)
			return
		}
		glog.Info("Patch successfully applied")
	} else {
		glog.Info("Patch is empty or invalid. Skipping the patch.")
	}
	// Build tools.
	skutil.LogErr(util.BuildSkiaTools())
	// Run render_pictures.
	if err := runRenderPictures(localSkpsDir, filepath.Join(localOutputDir, "withpatch"), filepath.Join(remoteOutputDir, "withpatch"), *gpuWithPatchRun); err != nil {
		glog.Errorf("Error while running withpatch render_pictures: %s", err)
		return
	}

	// Remove the patch and run render_pictures.
	// Reset Skia tree.
	skutil.LogErr(util.ResetCheckout(util.SkiaTreeDir))
	// Build tools.
	skutil.LogErr(util.BuildSkiaTools())
	// Run render_pictures.
	if err := runRenderPictures(localSkpsDir, filepath.Join(localOutputDir, "nopatch"), filepath.Join(remoteOutputDir, "nopatch"), *gpuNoPatchRun); err != nil {
		glog.Errorf("Error while running nopatch render_pictures: %s", err)
		return
	}

//.........这里部分代码省略.........

// downloadRemoteDir downloads the specified Google Storage dir to the specified
// local dir. The local dir will be emptied and recreated. Handles multiple levels
// of directories.
func (gs *GsUtil) downloadRemoteDir(localDir, gsDir string) error {
	// Empty the local dir.
	util.RemoveAll(localDir)
	// Create the local dir.
	util.MkdirAll(localDir, 0700)
	// The channel where the storage objects to be deleted will be sent to.
	chStorageObjects := make(chan filePathToStorageObject, MAX_CHANNEL_SIZE)
	req := gs.service.Objects.List(GS_BUCKET_NAME).Prefix(gsDir + "/")
	for req != nil {
		resp, err := req.Do()
		if err != nil {
			return fmt.Errorf("Error occured while listing %s: %s", gsDir, err)
		}
		for _, result := range resp.Items {
			fileName := filepath.Base(result.Name)
			// If downloading from subdir then add it to the fileName.
			fileGsDir := filepath.Dir(result.Name)
			subDirs := strings.TrimPrefix(fileGsDir, gsDir)
			if subDirs != "" {
				dirTokens := strings.Split(subDirs, "/")
				for i := range dirTokens {
					fileName = filepath.Join(dirTokens[len(dirTokens)-i-1], fileName)
				}
				// Create the local directory.
				util.MkdirAll(filepath.Join(localDir, filepath.Dir(fileName)), 0700)
			}
			chStorageObjects <- filePathToStorageObject{storageObject: result, filePath: fileName}
		}
		if len(resp.NextPageToken) > 0 {
			req.PageToken(resp.NextPageToken)
		} else {
			req = nil
		}
	}
	close(chStorageObjects)

	// Kick off goroutines to download the storage objects.
	var wg sync.WaitGroup
	for i := 0; i < GOROUTINE_POOL_SIZE; i++ {
		wg.Add(1)
		go func(goroutineNum int) {
			defer wg.Done()
			for obj := range chStorageObjects {
				result := obj.storageObject
				filePath := obj.filePath
				respBody, err := getRespBody(result, gs.client)
				if err != nil {
					glog.Errorf("Could not fetch %s: %s", result.MediaLink, err)
					return
				}
				defer util.Close(respBody)
				outputFile := filepath.Join(localDir, filePath)
				out, err := os.Create(outputFile)
				if err != nil {
					glog.Errorf("Unable to create file %s: %s", outputFile, err)
					return
				}
				defer util.Close(out)
				if _, err = io.Copy(out, respBody); err != nil {
					glog.Error(err)
					return
				}
				glog.Infof("Downloaded gs://%s/%s to %s with goroutine#%d", GS_BUCKET_NAME, result.Name, outputFile, goroutineNum)
				// Sleep for a second after downloading file to avoid bombarding Cloud
				// storage.
				time.Sleep(time.Second)
			}
		}(i + 1)
	}
	wg.Wait()
	return nil
}

func TestInfo(t *testing.T) {
	levelDBDir := filepath.Join(os.TempDir(), "android-leveldb")
	defer util.RemoveAll(levelDBDir)

	db, err := leveldb.OpenFile(levelDBDir, nil)
	if err != nil {
		t.Fatalf("Failed to create test leveldb: %s", err)
	}

	i := &info{
		db:      db,
		commits: mockCommits{},
	}
	if got, want := i.branchtargets(), []string{}; !util.SSliceEqual(got, want) {
		t.Errorf("Wrong targets: Got %v Want %v", got, want)
	}
	i.single_poll()

	// The first time we Get on an unknown target well get nil, err.
	commit, err := i.Get("git_master-skia", "razor-userdebug", "100")
	assert.Nil(t, commit)
	assert.NotNil(t, err)

	// After the first time we'll get an error.
	commit, err = i.Get("git_master-skia", "razor-userdebug", "100")
	assert.Nil(t, commit)
	assert.NotNil(t, err)

	// Now poll.
	i.single_poll()

	// Now the commits should be populated so the Get should succeed.
	commit, err = i.Get("git_master-skia", "razor-userdebug", "100")
	assert.Nil(t, err)
	assert.NotNil(t, commit)
	if got, want := commit.Hash, "1234567890"; got != want {
		t.Errorf("Wrong commit returned: Got %v Want %v", got, want)
	}

	commit, err = i.Get("git_master-skia", "razor-userdebug", "101")
	assert.Nil(t, err)
	assert.NotNil(t, commit)
	if got, want := commit.Hash, "1234567890"; got != want {
		t.Errorf("Wrong commit returned: Got %v Want %v", got, want)
	}

	commit, err = i.Get("git_master-skia", "razor-userdebug", "103")
	assert.Nil(t, err)
	assert.NotNil(t, commit)
	if got, want := commit.Hash, "987654321"; got != want {
		t.Errorf("Wrong commit returned: Got %v Want %v", got, want)
	}

	commit, err = i.Get("git_master-skia", "razor-userdebug", "99")
	assert.Nil(t, commit)
	assert.NotNil(t, err)

	// Now add another target.

	// The first time we Get on an unknown target well get nil, err.
	commit, err = i.Get("git_master-skia", "volantis-userdebug", "100")
	assert.Nil(t, commit)
	assert.NotNil(t, err)

	// After the first time we'll get an error.
	commit, err = i.Get("git_master-skia", "volantis-userdebug", "100")
	assert.Nil(t, commit)
	assert.NotNil(t, err)

	// Now poll.
	i.single_poll()

	// Now the commits should be populated so the Get should succeed.
	commit, err = i.Get("git_master-skia", "volantis-userdebug", "100")
	assert.Nil(t, err)
	assert.NotNil(t, commit)
	if got, want := commit.Hash, "1234567890"; got != want {
		t.Errorf("Wrong commit returned: Got %v Want %v", got, want)
	}

	commit, err = i.Get("git_master-skia", "volantis-userdebug", "101")
	assert.Nil(t, err)
	assert.NotNil(t, commit)
	if got, want := commit.Hash, "1234567890"; got != want {
		t.Errorf("Wrong commit returned: Got %v Want %v", got, want)
	}

	commit, err = i.Get("git_master-skia", "volantis-userdebug", "105")
	assert.Nil(t, err)
	assert.NotNil(t, commit)
	if got, want := commit.Hash, "987654321"; got != want {
		t.Errorf("Wrong commit returned: Got %v Want %v", got, want)
	}

	commit, err = i.Get("git_master-skia", "volantis-userdebug", "99")
	assert.Nil(t, commit)
	assert.NotNil(t, err)
}

// CleanTmpDir deletes all tmp files from the caller because telemetry tends to
// generate a lot of temporary artifacts there and they take up root disk space.
func CleanTmpDir() {
	files, _ := ioutil.ReadDir(os.TempDir())
	for _, f := range files {
		util.RemoveAll(filepath.Join(os.TempDir(), f.Name()))
	}
}

func TestPDFProcessor(t *testing.T) {
	testutils.SkipIfShort(t)

	// Get the service account client from meta data or a local config file.
	client, err := auth.NewJWTServiceAccountClient("", auth.DEFAULT_JWT_FILENAME, nil, storage.ScopeFullControl)
	assert.Nil(t, err)

	cacheDir, err := fileutil.EnsureDirExists(CACHE_DIR)
	assert.Nil(t, err)

	// Clean up after the test.
	defer func() {
		defer util.RemoveAll(cacheDir)
		deleteFolderContent(t, TEST_BUCKET, IMAGES_OUT_DIR, client)
		deleteFolderContent(t, TEST_BUCKET, JSON_OUT_DIR, client)
	}()

	// Configure the processor.
	ingesterConf := &sharedconfig.IngesterConfig{
		ExtraParams: map[string]string{
			CONFIG_INPUT_IMAGES_BUCKET:  TEST_BUCKET,
			CONFIG_INPUT_IMAGES_DIR:     IMAGES_IN_DIR,
			CONFIG_OUTPUT_JSON_BUCKET:   TEST_BUCKET,
			CONFIG_OUTPUT_JSON_DIR:      JSON_OUT_DIR,
			CONFIG_OUTPUT_IMAGES_BUCKET: TEST_BUCKET,
			CONFIG_OUTPUT_IMAGES_DIR:    IMAGES_OUT_DIR,
			CONFIG_PDF_CACHEDIR:         cacheDir,
		},
	}
	processor, err := newPDFProcessor(nil, ingesterConf, client)
	assert.Nil(t, err)

	// Load the example file and process it.
	fsResult, err := ingestion.FileSystemResult(TEST_INGESTION_FILE, "./")
	assert.Nil(t, err)

	err = processor.Process(fsResult)
	assert.Nil(t, err)

	// Fetch the json output and parse it.
	pProcessor := processor.(*pdfProcessor)

	// download the result.
	resultFileName := filepath.Join(CACHE_DIR, "result-file.json")
	assert.Nil(t, pProcessor.download(TEST_BUCKET, JSON_OUT_DIR, fsResult.Name(), resultFileName))

	// Make sure we get the expected result.
	fsResult, err = ingestion.FileSystemResult(TEST_INGESTION_FILE, "./")
	assert.Nil(t, err)
	r, err := fsResult.Open()
	assert.Nil(t, err)
	fsDMResults, err := goldingestion.ParseDMResultsFromReader(r)
	assert.Nil(t, err)

	foundResult, err := ingestion.FileSystemResult(resultFileName, "./")
	assert.Nil(t, err)
	r, err = foundResult.Open()
	assert.Nil(t, err)
	foundDMResults, err := goldingestion.ParseDMResultsFromReader(r)
	assert.Nil(t, err)

	dmResult1 := *fsDMResults
	dmResult2 := *foundDMResults
	dmResult1.Results = nil
	dmResult2.Results = nil
	assert.Equal(t, dmResult1, dmResult2)

	foundIdx := 0
	srcResults := fsDMResults.Results
	tgtResults := foundDMResults.Results
	for _, result := range srcResults {
		assert.True(t, foundIdx < len(tgtResults))
		if result.Options["ext"] == "pdf" {
			for ; (foundIdx < len(tgtResults)) && (result.Key["name"] == tgtResults[foundIdx].Key["name"]); foundIdx++ {
				assert.True(t, tgtResults[found