"net/url"
"os"
"regexp"
+ "strconv"
"strings"
"sync"
"time"
}
client.mtx.Lock()
for ch := range client.notifying[msg.ObjectUUID] {
- ch <- msg
+ go func() { ch <- msg }()
}
client.mtx.Unlock()
}
"priority": runner.Priority,
"state": arvados.ContainerRequestStateCommitted,
"scheduling_parameters": arvados.SchedulingParameters{
- Preemptible: true,
+ Preemptible: false,
Partitions: []string{},
},
"environment": map[string]string{
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
}
}
}
- var reCrunchstat = regexp.MustCompile(`mem .* rss`)
+ var reCrunchstat = regexp.MustCompile(`mem .* (\d+) rss`)
waitctr:
for cr.State != arvados.ContainerRequestStateFinal {
select {
}
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"
}
}
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
}
return os.Open(fnm)
}
collectionUUID := m[2]
- collectionPath := fnm[strings.Index(fnm, collectionUUID)+len(collectionUUID):]
+ collectionPath := m[3]
siteFSMtx.Lock()
defer siteFSMtx.Unlock()