src/server/pps/server/api_server.go
func (a *apiServer) hardStopPipeline(pachClient *client.APIClient, pipelineInfo *pps.PipelineInfo) error {
	// Remove the output branch's provenance so that no new jobs can be created
	if err := pachClient.CreateBranch(
		pipelineInfo.Pipeline.Name,
		pipelineInfo.OutputBranch,
		pipelineInfo.OutputBranch,
		nil,
	); err != nil && !isNotFoundErr(err) {
		return fmt.Errorf("could not recreate original output branch: %v", err)
	}

	// Now that new commits won't be created on the master branch, enumerate
	// existing commits and close any open ones.
	iter, err := pachClient.ListCommitStream(pachClient.Ctx(), &pfs.ListCommitRequest{
		Repo: client.NewRepo(pipelineInfo.Pipeline.Name),
		To:   client.NewCommit(pipelineInfo.Pipeline.Name, pipelineInfo.OutputBranch),
	})
	if err != nil {
		return fmt.Errorf("couldn't get open commits on '%s': %v", pipelineInfo.OutputBranch, err)
	}
	// Finish all open commits, most recent first (so that we finish the
	// current job's output commit--the oldest--last, and unblock the master
	// only after all other commits are also finished, preventing any new jobs)
	for {
		ci, err := iter.Recv()
		if err == io.EOF {
			break
		} else if err != nil {
			return err
		}
		if ci.Finished == nil {
			// Finish the commit and don't pass a tree
			pachClient.PfsAPIClient.FinishCommit(pachClient.Ctx(), &pfs.FinishCommitRequest{
				Commit: ci.Commit,
				Empty:  true,
			})
		}
	}
	return nil
}
func (a *apiServer) hardStopPipeline(pachClient *client.APIClient, pipelineInfo *pps.PipelineInfo) error {
	// Remove the output branch's provenance so that no new jobs can be created
	if err := pachClient.CreateBranch(
		pipelineInfo.Pipeline.Name,
		pipelineInfo.OutputBranch,
		pipelineInfo.OutputBranch,
		nil,
	); err != nil && !isNotFoundErr(err) {
		return fmt.Errorf("could not recreate original output branch: %v", err)
	}

	// Now that new commits won't be created on the master branch, enumerate
	// existing commits and close any open ones.
	iter, err := pachClient.ListCommitStream(pachClient.Ctx(), &pfs.ListCommitRequest{
		Repo: client.NewRepo(pipelineInfo.Pipeline.Name),
		To:   client.NewCommit(pipelineInfo.Pipeline.Name, pipelineInfo.OutputBranch),
	})
	if err != nil {
		return fmt.Errorf("couldn't get open commits on '%s': %v", pipelineInfo.OutputBranch, err)
	}
	// Finish all open commits, most recent first (so that we finish the
	// current job's output commit--the oldest--last, and unblock the master
	// only after all other commits are also finished, preventing any new jobs)
	for {
		ci, err := iter.Recv()
		if err == io.EOF {
			break
		} else if err != nil {
			return err
		}
		if ci.Finished == nil {
			// Finish the commit and don't pass a tree
			pachClient.PfsAPIClient.FinishCommit(pachClient.Ctx(), &pfs.FinishCommitRequest{
				Commit: ci.Commit,
				Empty:  true,
			})
		}
	}
	return nil
}
func ( a * apiServer ) hardStopPipeline ( pachClient * client . APIClient , pipelineInfo * pps . PipelineInfo ) error { // Remove the output branch's provenance so that no new jobs can be created if err := pachClient . CreateBranch ( pipelineInfo . Pipeline . Name , pipelineInfo . OutputBranch , pipelineInfo . OutputBranch , nil , ) ; err != nil && ! isNotFoundErr ( err ) { return fmt . Errorf ( " " , err ) } // Now that new commits won't be created on the master branch, enumerate // existing commits and close any open ones. iter , err := pachClient . ListCommitStream ( pachClient . Ctx ( ) , & pfs . ListCommitRequest { Repo : client . NewRepo ( pipelineInfo . Pipeline . Name ) , To : client . NewCommit ( pipelineInfo . Pipeline . Name , pipelineInfo . OutputBranch ) , } ) if err != nil { return fmt . Errorf ( " " , pipelineInfo . OutputBranch , err ) } // Finish all open commits, most recent first (so that we finish the // current job's output commit--the oldest--last, and unblock the master // only after all other commits are also finished, preventing any new jobs) for { ci , err := iter . Recv ( ) if err == io . EOF { break } else if err != nil { return err } if ci . Finished == nil { // Finish the commit and don't pass a tree pachClient . PfsAPIClient . FinishCommit ( pachClient . Ctx ( ) , & pfs . FinishCommitRequest { Commit : ci . Commit , Empty : true , } ) } } return nil }
// hardStopPipeline does essentially the same thing as StopPipeline (deletes the
// pipeline's branch provenance, deletes any open commits, deletes any k8s
// workers), but does it immediately. This is to avoid races between operations
// that will do subsequent work (e.g. UpdatePipeline and DeletePipeline) and the
// PPS master
hardStopPipeline does essentially the same thing as StopPipeline ( deletes the pipeline s branch provenance deletes any open commits deletes any k8s workers ) but does it immediately . This is to avoid races between operations that will do subsequent work ( e . g . UpdatePipeline and DeletePipeline ) and the PPS master