-package main
+package lightning
import (
"context"
"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"
"runtime_constraints": rc,
"priority": runner.Priority,
"state": arvados.ContainerRequestStateCommitted,
+ "scheduling_parameters": arvados.SchedulingParameters{
+ Preemptible: true,
+ Partitions: []string{},
+ },
},
})
if err != nil {
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,
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)
"owner_uuid": runner.ProjectUUID,
"manifest_text": mtxt,
"name": cname,
+ "properties": map[string]interface{}{
+ "blake2b": fmt.Sprintf("%x", b2),
+ },
},
})
if err != nil {
return coll.UUID, nil
}
-var arvadosClientFromEnv = arvados.NewClientFromEnv()
+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)
}
- log.Infof("reading %q from %s using Arvados client library", fnm[len(mnt):], uuid)
- 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
- // Don't cache more than one block for this file.
- kc.BlockCache = &keepclient.BlockCache{MaxBlocks: 1}
-
- var coll arvados.Collection
- err = arvadosClientFromEnv.RequestAndDecode(&coll, "GET", "arvados/v1/collections/"+uuid, nil, arvados.GetOptions{Select: []string{"uuid", "manifest_text"}})
- if err != nil {
- return nil, err
- }
- fs, err := coll.FileSystem(arvadosClientFromEnv, kc)
- if err != nil {
- return nil, err
- }
- return fs.Open(fnm[len(mnt):])
+ 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):])
}