package main
import (
+ "context"
"encoding/json"
"errors"
"fmt"
}
func (runner *arvadosContainerRunner) Run() (string, error) {
+ return runner.RunContext(context.Background())
+}
+
+func (runner *arvadosContainerRunner) RunContext(ctx context.Context) (string, error) {
if runner.ProjectUUID == "" {
return "", errors.New("cannot run arvados container: ProjectUUID not provided")
}
}
var reCrunchstat = regexp.MustCompile(`mem .* rss`)
+waitctr:
for cr.State != arvados.ContainerRequestStateFinal {
select {
+ case <-ctx.Done():
+ err := runner.Client.RequestAndDecode(&cr, "PATCH", "arvados/v1/container_requests/"+cr.UUID, nil, map[string]interface{}{
+ "container_request": map[string]interface{}{
+ "priority": 0,
+ },
+ })
+ if err != nil {
+ log.Errorf("error while trying to cancel container request %s: %s", cr.UUID, err)
+ }
+ break waitctr
case <-ticker.C:
refreshCR()
case msg := <-logch:
}
fmt.Fprint(os.Stderr, neednewline)
+ if err := ctx.Err(); err != nil {
+ return "", err
+ }
+
var c arvados.Container
err = runner.Client.RequestAndDecode(&c, "GET", "arvados/v1/containers/"+cr.ContainerUUID, nil, nil)
if err != nil {
return nil
}
+var mtxMakeCommandCollection sync.Mutex
+
func (runner *arvadosContainerRunner) makeCommandCollection() (string, error) {
+ mtxMakeCommandCollection.Lock()
+ defer mtxMakeCommandCollection.Unlock()
exe, err := ioutil.ReadFile("/proc/self/exe")
if err != nil {
return "", err
import (
"bufio"
"compress/gzip"
+ "context"
"encoding/gob"
"encoding/json"
"errors"
refFile string
outputFile string
projectUUID string
+ loglevel string
+ priority int
runLocal bool
skipOOO bool
outputTiles bool
outputStats string
matchChromosome *regexp.Regexp
encoder *gob.Encoder
+ batchArgs
}
func (cmd *importer) RunCommand(prog string, args []string, stdin io.Reader, stdout, stderr io.Writer) int {
flags.BoolVar(&cmd.outputTiles, "output-tiles", false, "include tile variant sequences in output file")
flags.BoolVar(&cmd.saveIncompleteTiles, "save-incomplete-tiles", false, "treat tiles with no-calls as regular tiles")
flags.StringVar(&cmd.outputStats, "output-stats", "", "output stats to `file` (json)")
+ cmd.batchArgs.Flags(flags)
matchChromosome := flags.String("match-chromosome", "^(chr)?([0-9]+|X|Y|MT?)$", "import chromosomes that match the given `regexp`")
- priority := flags.Int("priority", 500, "container request priority")
+ flags.IntVar(&cmd.priority, "priority", 500, "container request priority")
pprof := flags.String("pprof", "", "serve Go profile data at http://`[addr]:port`")
- loglevel := flags.String("loglevel", "info", "logging threshold (trace, debug, info, warn, error, fatal, or panic)")
+ flags.StringVar(&cmd.loglevel, "loglevel", "info", "logging threshold (trace, debug, info, warn, error, fatal, or panic)")
err = flags.Parse(args)
if err == flag.ErrHelp {
err = nil
}()
}
- lvl, err := log.ParseLevel(*loglevel)
+ lvl, err := log.ParseLevel(cmd.loglevel)
if err != nil {
return 2
}
}
if !cmd.runLocal {
- runner := arvadosContainerRunner{
- Name: "lightning import",
- Client: arvados.NewClientFromEnv(),
- ProjectUUID: cmd.projectUUID,
- RAM: 80000000000,
- VCPUs: 32,
- Priority: *priority,
- }
- err = runner.TranslatePaths(&cmd.tagLibraryFile, &cmd.refFile, &cmd.outputFile)
- if err != nil {
- return 1
- }
- inputs := flags.Args()
- for i := range inputs {
- err = runner.TranslatePaths(&inputs[i])
- if err != nil {
- return 1
- }
- }
- if cmd.outputFile == "-" {
- cmd.outputFile = "/mnt/output/library.gob.gz"
- } else {
- // Not yet implemented, but this should write
- // the collection to an existing collection,
- // possibly even an in-place update.
- err = errors.New("cannot specify output file in container mode: not implemented")
- return 1
- }
- runner.Args = append([]string{"import",
- "-local=true",
- "-loglevel=" + *loglevel,
- fmt.Sprintf("-skip-ooo=%v", cmd.skipOOO),
- fmt.Sprintf("-output-tiles=%v", cmd.outputTiles),
- fmt.Sprintf("-save-incomplete-tiles=%v", cmd.saveIncompleteTiles),
- "-match-chromosome", cmd.matchChromosome.String(),
- "-output-stats", "/mnt/output/stats.json",
- "-tag-library", cmd.tagLibraryFile,
- "-ref", cmd.refFile,
- "-o", cmd.outputFile,
- }, inputs...)
- var output string
- output, err = runner.Run()
+ err = cmd.runBatches(stdout, flags.Args())
if err != nil {
return 1
}
- fmt.Fprintln(stdout, output+"/library.gob.gz")
return 0
}
if err != nil {
return 1
}
+ infiles = cmd.batchArgs.Slice(infiles)
taglib, err := cmd.loadTagLibrary()
if err != nil {
return 0
}
+func (cmd *importer) runBatches(stdout io.Writer, inputs []string) error {
+ if cmd.outputFile != "-" {
+ // Not yet implemented, but this should write
+ // the collection to an existing collection,
+ // possibly even an in-place update.
+ return errors.New("cannot specify output file in container mode: not implemented")
+ }
+ client := arvados.NewClientFromEnv()
+ runner := arvadosContainerRunner{
+ Name: "lightning import",
+ Client: client,
+ ProjectUUID: cmd.projectUUID,
+ RAM: 80000000000,
+ VCPUs: 32,
+ Priority: cmd.priority,
+ }
+ err := runner.TranslatePaths(&cmd.tagLibraryFile, &cmd.refFile, &cmd.outputFile)
+ if err != nil {
+ return err
+ }
+ for i := range inputs {
+ err = runner.TranslatePaths(&inputs[i])
+ if err != nil {
+ return err
+ }
+ }
+
+ outputs, err := cmd.batchArgs.RunBatches(context.Background(), func(ctx context.Context, batch int) (string, error) {
+ runner := runner
+ if cmd.batches > 1 {
+ runner.Name += fmt.Sprintf(" (batch %d of %d)", batch, cmd.batches)
+ }
+ runner.Args = []string{"import",
+ "-local=true",
+ "-loglevel=" + cmd.loglevel,
+ fmt.Sprintf("-skip-ooo=%v", cmd.skipOOO),
+ fmt.Sprintf("-output-tiles=%v", cmd.outputTiles),
+ fmt.Sprintf("-save-incomplete-tiles=%v", cmd.saveIncompleteTiles),
+ "-match-chromosome", cmd.matchChromosome.String(),
+ "-output-stats", "/mnt/output/stats.json",
+ "-tag-library", cmd.tagLibraryFile,
+ "-ref", cmd.refFile,
+ "-o", "/mnt/output/library.gob.gz",
+ }
+ runner.Args = append(runner.Args, cmd.batchArgs.Args(batch)...)
+ runner.Args = append(runner.Args, inputs...)
+ return runner.RunContext(ctx)
+ })
+ if err != nil {
+ return err
+ }
+ var outfiles []string
+ for _, o := range outputs {
+ outfiles = append(outfiles, o+"/library.gob.gz")
+ }
+ fmt.Fprintln(stdout, strings.Join(outfiles, " "))
+ return nil
+}
+
func (cmd *importer) tileFasta(tilelib *tileLibrary, infile string) (tileSeq, []importStats, error) {
var input io.ReadCloser
input, err := os.Open(infile)
outputDir string
runLocal bool
vcpus int
+ batchArgs
stderr io.Writer
}
flags.StringVar(&cmd.outputDir, "output-dir", "", "output directory")
flags.IntVar(&cmd.vcpus, "vcpus", 0, "number of VCPUs to request for arvados container (default: 2*number of input files, max 32)")
flags.BoolVar(&cmd.runLocal, "local", false, "run on local host (default: run in an arvados container)")
+ cmd.batchArgs.Flags(flags)
priority := flags.Int("priority", 500, "container request priority")
pprof := flags.String("pprof", "", "serve Go profile data at http://`[addr]:port`")
err = flags.Parse(args)
if err != nil {
return 1
}
- if cmd.vcpus = len(infiles) * 2; cmd.vcpus > 32 {
+ if cmd.vcpus = len(cmd.batchArgs.Slice(infiles)) * 2; cmd.vcpus > 32 {
cmd.vcpus = 32
}
}
+ client := arvados.NewClientFromEnv()
runner := arvadosContainerRunner{
Name: "lightning vcf2fasta",
- Client: arvados.NewClientFromEnv(),
+ Client: client,
ProjectUUID: cmd.projectUUID,
RAM: 2<<30 + int64(cmd.vcpus)<<28,
VCPUs: cmd.vcpus,
},
},
}
- if cmd.mask {
- runner.RAM += int64(cmd.vcpus) << 31
- }
err = runner.TranslatePaths(&cmd.refFile, &cmd.genomeFile)
if err != nil {
return 1
return 1
}
}
- runner.Args = append([]string{"vcf2fasta",
- "-local=true",
- "-ref", cmd.refFile, fmt.Sprintf("-mask=%v", cmd.mask),
- "-genome", cmd.genomeFile,
- "-gvcf-regions.py", "/gvcf_regions.py",
- "-gvcf-type", cmd.gvcfType,
- "-output-dir", "/mnt/output"}, inputs...)
- var output string
- output, err = runner.Run()
+ var outputs []string
+ outputs, err = cmd.batchArgs.RunBatches(context.Background(), func(ctx context.Context, batch int) (string, error) {
+ runner := runner
+ if cmd.mask {
+ runner.RAM += int64(cmd.vcpus) << 31
+ }
+ runner.Args = []string{"vcf2fasta",
+ "-local=true",
+ "-ref", cmd.refFile, fmt.Sprintf("-mask=%v", cmd.mask),
+ "-genome", cmd.genomeFile,
+ "-gvcf-regions.py", "/gvcf_regions.py",
+ "-gvcf-type", cmd.gvcfType,
+ "-output-dir", "/mnt/output",
+ }
+ runner.Args = append(runner.Args, cmd.batchArgs.Args(batch)...)
+ runner.Args = append(runner.Args, inputs...)
+ log.Printf("batch %d: %v", batch, runner.Args)
+ return runner.RunContext(ctx)
+ })
if err != nil {
return 1
}
- fmt.Fprintln(stdout, output)
+ fmt.Fprintln(stdout, strings.Join(outputs, " "))
return 0
}
if err != nil {
return 1
}
+ infiles = cmd.batchArgs.Slice(infiles)
type job struct {
vcffile string
projects / lightning.git / commitdiff