Bump import RAM.

[lightning.git] / arvados.go

diff --git a/arvados.go b/arvados.go
index 5f78a36408bdbbb8e2aba29b4571fbfd87749c85..48cfad7ccdd6f37e3381f0bb54d1170031babc5c 100644 (file)
--- a/arvados.go
+++ b/arvados.go
@@ -1,18 +1,21 @@
-package main
+package lightning
 
 import (
        "context"
        "encoding/json"
        "errors"
        "fmt"
+       "io"
        "io/ioutil"
        "net/url"
        "os"
        "regexp"
+       "runtime"
        "strings"
        "sync"
        "time"
 
+       "git.arvados.org/arvados.git/lib/cmd"
        "git.arvados.org/arvados.git/sdk/go/arvados"
        "git.arvados.org/arvados.git/sdk/go/arvadosclient"
        "git.arvados.org/arvados.git/sdk/go/keepclient"
@@ -184,12 +187,14 @@ type arvadosContainerRunner struct {
        Name        string
        OutputName  string
        ProjectUUID string
+       APIAccess   bool
        VCPUs       int
        RAM         int64
        Prog        string // if empty, run /proc/self/exe
        Args        []string
        Mounts      map[string]map[string]interface{}
        Priority    int
+       KeepCache   int // cache buffers per VCPU (0 for default)
 }
 
 func (runner *arvadosContainerRunner) Run() (string, error) {
@@ -229,10 +234,15 @@ func (runner *arvadosContainerRunner) RunContext(ctx context.Context) (string, e
        if priority < 1 {
                priority = 500
        }
+       keepCache := runner.KeepCache
+       if keepCache < 1 {
+               keepCache = 2
+       }
        rc := arvados.RuntimeConstraints{
+               API:          &runner.APIAccess,
                VCPUs:        runner.VCPUs,
                RAM:          runner.RAM,
-               KeepCacheRAM: (1 << 26) * 2 * int64(runner.VCPUs),
+               KeepCacheRAM: (1 << 26) * int64(keepCache) * int64(runner.VCPUs),
        }
        outname := &runner.OutputName
        if *outname == "" {
@@ -252,6 +262,10 @@ func (runner *arvadosContainerRunner) RunContext(ctx context.Context) (string, e
                        "runtime_constraints": rc,
                        "priority":            runner.Priority,
                        "state":               arvados.ContainerRequestStateCommitted,
+                       "scheduling_parameters": arvados.SchedulingParameters{
+                               Preemptible: true,
+                               Partitions:  []string{},
+                       },
                },
        })
        if err != nil {
@@ -391,7 +405,7 @@ func (runner *arvadosContainerRunner) makeCommandCollection() (string, error) {
                return "", err
        }
        b2 := blake2b.Sum256(exe)
-       cname := fmt.Sprintf("lightning-%x", b2)
+       cname := "lightning " + cmd.Version.String() // must build with "make", not just "go install"
        var existing arvados.CollectionList
        err = runner.Client.RequestAndDecode(&existing, "GET", "arvados/v1/collections", nil, arvados.ListOptions{
                Limit: 1,
@@ -399,15 +413,16 @@ func (runner *arvadosContainerRunner) makeCommandCollection() (string, error) {
                Filters: []arvados.Filter{
                        {Attr: "name", Operator: "=", Operand: cname},
                        {Attr: "owner_uuid", Operator: "=", Operand: runner.ProjectUUID},
+                       {Attr: "properties.blake2b", Operator: "=", Operand: fmt.Sprintf("%x", b2)},
                },
        })
        if err != nil {
                return "", err
        }
        if len(existing.Items) > 0 {
-               uuid := existing.Items[0].UUID
-               log.Printf("using lightning binary in existing collection %s (name is %q; did not verify whether content matches)", uuid, cname)
-               return uuid, nil
+               coll := existing.Items[0]
+               log.Printf("using lightning binary in existing collection %s (name is %q, hash is %q; did not verify whether content matches)", coll.UUID, cname, coll.Properties["blake2b"])
+               return coll.UUID, nil
        }
        log.Printf("writing lightning binary to new collection %q", cname)
        ac, err := arvadosclient.New(runner.Client)
@@ -441,6 +456,9 @@ func (runner *arvadosContainerRunner) makeCommandCollection() (string, error) {
                        "owner_uuid":    runner.ProjectUUID,
                        "manifest_text": mtxt,
                        "name":          cname,
+                       "properties": map[string]interface{}{
+                               "blake2b": fmt.Sprintf("%x", b2),
+                       },
                },
        })
        if err != nil {
@@ -449,3 +467,45 @@ func (runner *arvadosContainerRunner) makeCommandCollection() (string, error) {
        log.Printf("stored lightning binary in new collection %s", coll.UUID)
        return coll.UUID, nil
 }
+
+var (
+       arvadosClientFromEnv = arvados.NewClientFromEnv()
+       siteFS               arvados.CustomFileSystem
+       siteFSMtx            sync.Mutex
+)
+
+func open(fnm string) (io.ReadCloser, error) {
+       if os.Getenv("ARVADOS_API_HOST") == "" {
+               return os.Open(fnm)
+       }
+       m := collectionInPathRe.FindStringSubmatch(fnm)
+       if m == nil {
+               return os.Open(fnm)
+       }
+       uuid := m[2]
+       mnt := "/mnt/" + uuid + "/"
+       if !strings.HasPrefix(fnm, mnt) {
+               return os.Open(fnm)
+       }
+
+       siteFSMtx.Lock()
+       defer siteFSMtx.Unlock()
+       if siteFS == nil {
+               log.Info("setting up Arvados client")
+               ac, err := arvadosclient.New(arvadosClientFromEnv)
+               if err != nil {
+                       return nil, err
+               }
+               ac.Client = arvados.DefaultSecureClient
+               kc := keepclient.New(ac)
+               // Don't use keepclient's default short timeouts.
+               kc.HTTPClient = arvados.DefaultSecureClient
+               // Guess max concurrent readers, hope to avoid cache
+               // thrashing.
+               kc.BlockCache = &keepclient.BlockCache{MaxBlocks: runtime.NumCPU() * 3}
+               siteFS = arvadosClientFromEnv.SiteFileSystem(kc)
+       }
+
+       log.Infof("reading %q from %s using Arvados client", fnm[len(mnt):], uuid)
+       return siteFS.Open("by_id/" + uuid + "/" + fnm[len(mnt):])
+}