projects
/
lightning.git
/ blobdiff
summary
|
shortlog
|
log
|
commit
|
commitdiff
|
tree
raw
|
inline
| side by side
Bump slice-numpy memory.
[lightning.git]
/
slicenumpy.go
diff --git
a/slicenumpy.go
b/slicenumpy.go
index a8d217fefa664c3d566a93554f01cf5224ef5367..5aa97ac5a7c44707b03720ac6631eaa03599ce54 100644
(file)
--- a/
slicenumpy.go
+++ b/
slicenumpy.go
@@
-13,6
+13,7
@@
import (
"net/http"
_ "net/http/pprof"
"os"
"net/http"
_ "net/http/pprof"
"os"
+ "regexp"
"runtime"
"sort"
"strings"
"runtime"
"sort"
"strings"
@@
-21,13
+22,13
@@
import (
"git.arvados.org/arvados.git/sdk/go/arvados"
"github.com/arvados/lightning/hgvs"
"github.com/kshedden/gonpy"
"git.arvados.org/arvados.git/sdk/go/arvados"
"github.com/arvados/lightning/hgvs"
"github.com/kshedden/gonpy"
- "github.com/sirupsen/logrus"
log "github.com/sirupsen/logrus"
"golang.org/x/crypto/blake2b"
)
type sliceNumpy struct {
log "github.com/sirupsen/logrus"
"golang.org/x/crypto/blake2b"
)
type sliceNumpy struct {
- filter filter
+ filter filter
+ threads int
}
func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, stdout, stderr io.Writer) int {
}
func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, stdout, stderr io.Writer) int {
@@
-48,6
+49,7
@@
func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
ref := flags.String("ref", "", "reference name (if blank, choose last one that appears in input)")
regionsFilename := flags.String("regions", "", "only output columns/annotations that intersect regions in specified bed `file`")
expandRegions := flags.Int("expand-regions", 0, "expand specified regions by `N` base pairs on each side`")
ref := flags.String("ref", "", "reference name (if blank, choose last one that appears in input)")
regionsFilename := flags.String("regions", "", "only output columns/annotations that intersect regions in specified bed `file`")
expandRegions := flags.Int("expand-regions", 0, "expand specified regions by `N` base pairs on each side`")
+ flags.IntVar(&cmd.threads, "threads", 16, "number of memory-hungry assembly threads")
cmd.filter.Flags(flags)
err = flags.Parse(args)
if err == flag.ErrHelp {
cmd.filter.Flags(flags)
err = flags.Parse(args)
if err == flag.ErrHelp {
@@
-68,8
+70,8
@@
func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
Name: "lightning slice-numpy",
Client: arvados.NewClientFromEnv(),
ProjectUUID: *projectUUID,
Name: "lightning slice-numpy",
Client: arvados.NewClientFromEnv(),
ProjectUUID: *projectUUID,
- RAM:
2
50000000000,
- VCPUs:
32
,
+ RAM:
6
50000000000,
+ VCPUs:
96
,
Priority: *priority,
KeepCache: 2,
APIAccess: true,
Priority: *priority,
KeepCache: 2,
APIAccess: true,
@@
-82,6
+84,7
@@
func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
"-pprof", ":6060",
"-input-dir", *inputDir,
"-output-dir", "/mnt/output",
"-pprof", ":6060",
"-input-dir", *inputDir,
"-output-dir", "/mnt/output",
+ "-threads", fmt.Sprintf("%d", cmd.threads),
"-regions", *regionsFilename,
"-expand-regions", fmt.Sprintf("%d", *expandRegions),
}
"-regions", *regionsFilename,
"-expand-regions", fmt.Sprintf("%d", *expandRegions),
}
@@
-107,10
+110,18
@@
func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
var cgnames []string
var refseq map[string][]tileLibRef
var cgnames []string
var refseq map[string][]tileLibRef
+ var reftiledata = make(map[tileLibRef][]byte, 11000000)
in0, err := open(infiles[0])
if err != nil {
return 1
}
in0, err := open(infiles[0])
if err != nil {
return 1
}
+
+ matchGenome, err := regexp.Compile(cmd.filter.MatchGenome)
+ if err != nil {
+ err = fmt.Errorf("-match-genome: invalid regexp: %q", cmd.filter.MatchGenome)
+ return 1
+ }
+
taglen := -1
DecodeLibrary(in0, strings.HasSuffix(infiles[0], ".gz"), func(ent *LibraryEntry) error {
if len(ent.TagSet) > 0 {
taglen := -1
DecodeLibrary(in0, strings.HasSuffix(infiles[0], ".gz"), func(ent *LibraryEntry) error {
if len(ent.TagSet) > 0 {
@@
-122,7
+133,14
@@
func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
}
}
for _, cg := range ent.CompactGenomes {
}
}
for _, cg := range ent.CompactGenomes {
- cgnames = append(cgnames, cg.Name)
+ if matchGenome.MatchString(cg.Name) {
+ cgnames = append(cgnames, cg.Name)
+ }
+ }
+ for _, tv := range ent.TileVariants {
+ if tv.Ref {
+ reftiledata[tileLibRef{tv.Tag, tv.Variant}] = tv.Sequence
+ }
}
return nil
})
}
return nil
})
@@
-138,6
+156,10
@@
func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
err = fmt.Errorf("tagset not found")
return 1
}
err = fmt.Errorf("tagset not found")
return 1
}
+ if len(cgnames) == 0 {
+ err = fmt.Errorf("no genomes found matching regexp %q", cmd.filter.MatchGenome)
+ return 1
+ }
sort.Strings(cgnames)
{
sort.Strings(cgnames)
{
@@
-163,7
+185,7
@@
func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
}
}
}
}
- log.Info("
building list of reference tiles to load") // TODO: more efficient if we had saved all ref tiles in slice0
+ log.Info("
indexing reference tiles")
type reftileinfo struct {
variant tileVariantID
seqname string // chr1
type reftileinfo struct {
variant tileVariantID
seqname string // chr1
@@
-173,55
+195,43
@@
func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
reftile := map[tagID]*reftileinfo{}
for seqname, cseq := range refseq {
for _, libref := range cseq {
reftile := map[tagID]*reftileinfo{}
for seqname, cseq := range refseq {
for _, libref := range cseq {
- reftile[libref.Tag] = &reftileinfo{seqname: seqname, variant: libref.Variant}
- }
- }
- log.Info("loading reference tiles from all slices")
- throttle1 := throttle{Max: runtime.GOMAXPROCS(0)}
- for _, infile := range infiles {
- infile := infile
- throttle1.Go(func() error {
- defer log.Infof("%s: done", infile)
- f, err := open(infile)
- if err != nil {
- return err
+ reftile[libref.Tag] = &reftileinfo{
+ seqname: seqname,
+ variant: libref.Variant,
+ tiledata: reftiledata[libref],
}
}
- defer f.Close()
- return DecodeLibrary(f, strings.HasSuffix(infile, ".gz"), func(ent *LibraryEntry) error {
- for _, tv := range ent.TileVariants {
- if dst, ok := reftile[tv.Tag]; ok && dst.variant == tv.Variant {
- dst.tiledata = tv.Sequence
- }
- }
- return nil
- })
- })
+ }
}
}
- throttle1.Wait()
+ throttleCPU := throttle{Max: runtime.GOMAXPROCS(0)}
log.Info("reconstructing reference sequences")
for seqname, cseq := range refseq {
seqname, cseq := seqname, cseq
log.Info("reconstructing reference sequences")
for seqname, cseq := range refseq {
seqname, cseq := seqname, cseq
- throttle
1
.Go(func() error {
+ throttle
CPU
.Go(func() error {
defer log.Printf("... %s done", seqname)
pos := 0
for _, libref := range cseq {
rt := reftile[libref.Tag]
rt.pos = pos
defer log.Printf("... %s done", seqname)
pos := 0
for _, libref := range cseq {
rt := reftile[libref.Tag]
rt.pos = pos
+ if len(rt.tiledata) == 0 {
+ return fmt.Errorf("missing tiledata for tag %d variant %d in %s in ref", libref.Tag, libref.Variant, seqname)
+ }
pos += len(rt.tiledata) - taglen
}
return nil
})
}
pos += len(rt.tiledata) - taglen
}
return nil
})
}
- throttle
1
.Wait()
+ throttle
CPU
.Wait()
log.Info("TODO: determining which tiles intersect given regions")
log.Info("TODO: determining which tiles intersect given regions")
+ throttleMem := throttle{Max: cmd.threads} // TODO: estimate using mem and data size
+ throttleNumpyMem := throttle{Max: cmd.threads/2 + 1}
log.Info("generating annotations and numpy matrix for each slice")
var done int64
for infileIdx, infile := range infiles {
infileIdx, infile := infileIdx, infile
log.Info("generating annotations and numpy matrix for each slice")
var done int64
for infileIdx, infile := range infiles {
infileIdx, infile := infileIdx, infile
- throttle
1
.Go(func() error {
+ throttle
Mem
.Go(func() error {
seq := make(map[tagID][]TileVariant, 50000)
cgs := make(map[string]CompactGenome, len(cgnames))
f, err := open(infile)
seq := make(map[tagID][]TileVariant, 50000)
cgs := make(map[string]CompactGenome, len(cgnames))
f, err := open(infile)
@@
-229,9
+239,12
@@
func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
return err
}
defer f.Close()
return err
}
defer f.Close()
- log.Infof("
reading %s"
, infile)
+ log.Infof("
%04d: reading %s", infileIdx
, infile)
err = DecodeLibrary(f, strings.HasSuffix(infile, ".gz"), func(ent *LibraryEntry) error {
for _, tv := range ent.TileVariants {
err = DecodeLibrary(f, strings.HasSuffix(infile, ".gz"), func(ent *LibraryEntry) error {
for _, tv := range ent.TileVariants {
+ if tv.Ref {
+ continue
+ }
variants := seq[tv.Tag]
if len(variants) == 0 {
variants = make([]TileVariant, 100)
variants := seq[tv.Tag]
if len(variants) == 0 {
variants = make([]TileVariant, 100)
@@
-243,7
+256,9
@@
func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
seq[tv.Tag] = variants
}
for _, cg := range ent.CompactGenomes {
seq[tv.Tag] = variants
}
for _, cg := range ent.CompactGenomes {
- cgs[cg.Name] = cg
+ if matchGenome.MatchString(cg.Name) {
+ cgs[cg.Name] = cg
+ }
}
return nil
})
}
return nil
})
@@
-255,14
+270,14
@@
func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
// TODO: filters
// TODO: filters
- log.Infof("
renumber/dedup variants for tags %d-%d"
, tagstart, tagend)
+ log.Infof("
%04d: renumber/dedup variants for tags %d-%d", infileIdx
, tagstart, tagend)
variantRemap := make([][]tileVariantID, tagend-tagstart)
variantRemap := make([][]tileVariantID, tagend-tagstart)
- throttle
2
:= throttle{Max: runtime.GOMAXPROCS(0)}
+ throttle
CPU
:= throttle{Max: runtime.GOMAXPROCS(0)}
for tag, variants := range seq {
tag, variants := tag, variants
for tag, variants := range seq {
tag, variants := tag, variants
- throttle
2
.Acquire()
+ throttle
CPU
.Acquire()
go func() {
go func() {
- defer throttle
2
.Release()
+ defer throttle
CPU
.Release()
count := make(map[[blake2b.Size256]byte]int, len(variants))
for _, cg := range cgs {
idx := (tag - tagstart) * 2
count := make(map[[blake2b.Size256]byte]int, len(variants))
for _, cg := range cgs {
idx := (tag - tagstart) * 2
@@
-303,10
+318,10
@@
func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
variantRemap[tag-tagstart] = remap
}()
}
variantRemap[tag-tagstart] = remap
}()
}
- throttle
2
.Wait()
+ throttle
CPU
.Wait()
annotationsFilename := fmt.Sprintf("%s/matrix.%04d.annotations.csv", *outputDir, infileIdx)
annotationsFilename := fmt.Sprintf("%s/matrix.%04d.annotations.csv", *outputDir, infileIdx)
- log.Infof("
writing %s"
, annotationsFilename)
+ log.Infof("
%04d: writing %s", infileIdx
, annotationsFilename)
annof, err := os.Create(annotationsFilename)
if err != nil {
return err
annof, err := os.Create(annotationsFilename)
if err != nil {
return err
@@
-334,7
+349,7
@@
func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
diffs, _ := hgvs.Diff(reftilestr, strings.ToUpper(string(tv.Sequence)), 0)
for _, diff := range diffs {
diff.Position += rt.pos
diffs, _ := hgvs.Diff(reftilestr, strings.ToUpper(string(tv.Sequence)), 0)
for _, diff := range diffs {
diff.Position += rt.pos
- fmt.Fprintf(annow, "%d,%d,%d,%s:g.%s
\n", tag, outcol, remap[v], rt.seqname, diff.String()
)
+ fmt.Fprintf(annow, "%d,%d,%d,%s:g.%s
,%s,%d,%s,%s\n", tag, outcol, remap[v], rt.seqname, diff.String(), rt.seqname, diff.Position, diff.Ref, diff.New
)
}
}
}
}
}
}
@@
-347,7
+362,8
@@
func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
return err
}
return err
}
- log.Infof("%s: preparing numpy", infile)
+ throttleNumpyMem.Acquire()
+ log.Infof("%04d: preparing numpy", infileIdx)
rows := len(cgnames)
cols := 2 * int(tagend-tagstart)
out := make([]int16, rows*cols)
rows := len(cgnames)
cols := 2 * int(tagend-tagstart)
out := make([]int16, rows*cols)
@@
-361,6
+377,8
@@
func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
}
}
}
}
}
}
+ seq = nil
+ throttleNumpyMem.Release()
fnm := fmt.Sprintf("%s/matrix.%04d.npy", *outputDir, infileIdx)
output, err := os.Create(fnm)
fnm := fmt.Sprintf("%s/matrix.%04d.npy", *outputDir, infileIdx)
output, err := os.Create(fnm)
@@
-373,11
+391,11
@@
func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
if err != nil {
return err
}
if err != nil {
return err
}
- log.WithFields(log
rus
.Fields{
+ log.WithFields(log.Fields{
"filename": fnm,
"rows": rows,
"cols": cols,
"filename": fnm,
"rows": rows,
"cols": cols,
- }).Info
("writing numpy"
)
+ }).Info
f("%04d: writing numpy", infileIdx
)
npw.Shape = []int{rows, cols}
npw.WriteInt16(out)
err = bufw.Flush()
npw.Shape = []int{rows, cols}
npw.WriteInt16(out)
err = bufw.Flush()
@@
-392,7
+410,7
@@
func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
return nil
})
}
return nil
})
}
- if err = throttle
1
.Wait(); err != nil {
+ if err = throttle
Mem
.Wait(); err != nil {
return 1
}
return 0
return 1
}
return 0