Fix -max-tag filter.

[lightning.git] / arvados.go

diff --git a/arvados.go b/arvados.go
index 50ab375f816b52632c7d45913b3565a91fbaa185..8965a0c43f9dffdc9a95ed162fe6d88ffa7365fc 100644 (file)
--- a/arvados.go
+++ b/arvados.go
@@ -15,6 +15,7 @@ import (
        "net/url"
        "os"
        "regexp"
+       "strconv"
        "strings"
        "sync"
        "time"
@@ -177,7 +178,7 @@ reconnect:
                                }
                                client.mtx.Lock()
                                for ch := range client.notifying[msg.ObjectUUID] {
-                                       ch <- msg
+                                       go func() { ch <- msg }()
                                }
                                client.mtx.Unlock()
                        }
@@ -268,7 +269,7 @@ func (runner *arvadosContainerRunner) RunContext(ctx context.Context) (string, e
                        "priority":            runner.Priority,
                        "state":               arvados.ContainerRequestStateCommitted,
                        "scheduling_parameters": arvados.SchedulingParameters{
-                               Preemptible: true,
+                               Preemptible: false,
                                Partitions:  []string{},
                        },
                        "environment": map[string]string{
@@ -297,14 +298,18 @@ func (runner *arvadosContainerRunner) RunContext(ctx context.Context) (string, e
 
        lastState := cr.State
        refreshCR := func() {
-               err = runner.Client.RequestAndDecode(&cr, "GET", "arvados/v1/container_requests/"+cr.UUID, nil, nil)
+               ctx, cancel := context.WithDeadline(ctx, time.Now().Add(time.Minute))
+               defer cancel()
+               err = runner.Client.RequestAndDecodeContext(ctx, &cr, "GET", "arvados/v1/container_requests/"+cr.UUID, nil, nil)
                if err != nil {
                        fmt.Fprint(os.Stderr, neednewline)
+                       neednewline = ""
                        log.Printf("error getting container request: %s", err)
                        return
                }
                if lastState != cr.State {
                        fmt.Fprint(os.Stderr, neednewline)
+                       neednewline = ""
                        log.Printf("container request state: %s", cr.State)
                        lastState = cr.State
                }
@@ -321,7 +326,7 @@ func (runner *arvadosContainerRunner) RunContext(ctx context.Context) (string, e
                }
        }
 
-       var reCrunchstat = regexp.MustCompile(`mem .* rss`)
+       var reCrunchstat = regexp.MustCompile(`mem .* (\d+) rss`)
 waitctr:
        for cr.State != arvados.ContainerRequestStateFinal {
                select {
@@ -351,9 +356,10 @@ waitctr:
                                }
                        case "crunchstat":
                                for _, line := range strings.Split(msg.Properties.Text, "\n") {
-                                       mem := reCrunchstat.FindString(line)
-                                       if mem != "" {
-                                               fmt.Fprintf(os.Stderr, "%s               \r", mem)
+                                       m := reCrunchstat.FindStringSubmatch(line)
+                                       if m != nil {
+                                               rss, _ := strconv.ParseInt(m[1], 10, 64)
+                                               fmt.Fprintf(os.Stderr, "%s rss %.3f GB           \r", cr.UUID, float64(rss)/1e9)
                                                neednewline = "\n"
                                        }
                                }
@@ -392,16 +398,20 @@ func (runner *arvadosContainerRunner) TranslatePaths(paths ...*string) error {
                if m == nil {
                        return fmt.Errorf("cannot find uuid in path: %q", *path)
                }
-               uuid := m[2]
-               mnt, ok := runner.Mounts["/mnt/"+uuid]
+               collID := m[2]
+               mnt, ok := runner.Mounts["/mnt/"+collID]
                if !ok {
                        mnt = map[string]interface{}{
                                "kind": "collection",
-                               "uuid": uuid,
                        }
-                       runner.Mounts["/mnt/"+uuid] = mnt
+                       if len(collID) == 27 {
+                               mnt["uuid"] = collID
+                       } else {
+                               mnt["portable_data_hash"] = collID
+                       }
+                       runner.Mounts["/mnt/"+collID] = mnt
                }
-               *path = "/mnt/" + uuid + m[3]
+               *path = "/mnt/" + collID + m[3]
        }
        return nil
 }
@@ -533,7 +543,7 @@ func open(fnm string) (file, error) {
                return os.Open(fnm)
        }
        collectionUUID := m[2]
-       collectionPath := fnm[strings.Index(fnm, collectionUUID)+len(collectionUUID):]
+       collectionPath := m[3]
 
        siteFSMtx.Lock()
        defer siteFSMtx.Unlock()