X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/8734a7391a5672eebcdf572d93bae1b3ed1179c9..a41c0f6aa41b658c8f2947c46cb90778894f5cf3:/services/crunch-dispatch-local/crunch-dispatch-local.go diff --git a/services/crunch-dispatch-local/crunch-dispatch-local.go b/services/crunch-dispatch-local/crunch-dispatch-local.go index bb3c05c7eb..279327ba18 100644 --- a/services/crunch-dispatch-local/crunch-dispatch-local.go +++ b/services/crunch-dispatch-local/crunch-dispatch-local.go @@ -1,12 +1,15 @@ +// Copyright (C) The Arvados Authors. All rights reserved. +// +// SPDX-License-Identifier: AGPL-3.0 + package main // Dispatcher service for Crunch that runs containers locally. import ( + "context" "flag" - "git.curoverse.com/arvados.git/sdk/go/arvados" - "git.curoverse.com/arvados.git/sdk/go/arvadosclient" - "git.curoverse.com/arvados.git/sdk/go/dispatch" + "fmt" "log" "os" "os/exec" @@ -14,8 +17,14 @@ import ( "sync" "syscall" "time" + + "git.curoverse.com/arvados.git/sdk/go/arvados" + "git.curoverse.com/arvados.git/sdk/go/arvadosclient" + "git.curoverse.com/arvados.git/sdk/go/dispatch" ) +var version = "dev" + func main() { err := doMain() if err != nil { @@ -43,9 +52,22 @@ func doMain() error { "/usr/bin/crunch-run", "Crunch command to run container") + getVersion := flags.Bool( + "version", + false, + "Print version information and exit.") + // Parse args; omit the first arg which is the command name flags.Parse(os.Args[1:]) + // Print version information if requested + if *getVersion { + fmt.Printf("crunch-dispatch-local %s\n", version) + return nil + } + + log.Printf("crunch-dispatch-local %s started", version) + runningCmds = make(map[string]*exec.Cmd) arv, err := arvadosclient.MakeArvadosClient() @@ -61,7 +83,8 @@ func doMain() error { PollPeriod: time.Duration(*pollInterval) * time.Second, } - err = dispatcher.Run() + ctx, cancel := context.WithCancel(context.Background()) + err = dispatcher.Run(ctx) if err != nil { return err } @@ -72,7 +95,7 @@ func doMain() error { log.Printf("Received %s, shutting down", sig) signal.Stop(c) - dispatcher.Stop() + cancel() runningCmdsMutex.Lock() // Finished dispatching; interrupt any crunch jobs that are still running @@ -103,7 +126,7 @@ var startCmd = startFunc // crunch-run terminates, mark the container as Cancelled. func run(dispatcher *dispatch.Dispatcher, container arvados.Container, - status chan arvados.Container) { + status <-chan arvados.Container) { uuid := container.UUID @@ -170,8 +193,7 @@ func run(dispatcher *dispatch.Dispatcher, if err != nil { log.Printf("Error getting final container state: %v", err) } - if container.LockedByUUID == dispatcher.Auth.UUID && - (container.State == dispatch.Locked || container.State == dispatch.Running) { + if container.State == dispatch.Locked || container.State == dispatch.Running { log.Printf("After %s process termination, container state for %v is %q. Updating it to %q", *crunchRunCommand, container.State, uuid, dispatch.Cancelled) dispatcher.UpdateState(uuid, dispatch.Cancelled)