3 // Dispatcher service for Crunch that runs containers locally.
7 "git.curoverse.com/arvados.git/sdk/go/arvados"
8 "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
9 "git.curoverse.com/arvados.git/sdk/go/dispatch"
27 runningCmds map[string]*exec.Cmd
28 runningCmdsMutex sync.Mutex
29 waitGroup sync.WaitGroup
30 crunchRunCommand *string
34 flags := flag.NewFlagSet("crunch-dispatch-local", flag.ExitOnError)
36 pollInterval := flags.Int(
39 "Interval in seconds to poll for queued containers")
41 crunchRunCommand = flags.String(
43 "/usr/bin/crunch-run",
44 "Crunch command to run container")
46 // Parse args; omit the first arg which is the command name
47 flags.Parse(os.Args[1:])
49 runningCmds = make(map[string]*exec.Cmd)
51 arv, err := arvadosclient.MakeArvadosClient()
53 log.Printf("Error making Arvados client: %v", err)
58 dispatcher := dispatch.Dispatcher{
61 PollPeriod: time.Duration(*pollInterval) * time.Second,
64 err = dispatcher.Run()
69 c := make(chan os.Signal, 1)
70 signal.Notify(c, os.Interrupt, syscall.SIGTERM, syscall.SIGQUIT)
72 log.Printf("Received %s, shutting down", sig)
77 runningCmdsMutex.Lock()
78 // Finished dispatching; interrupt any crunch jobs that are still running
79 for _, cmd := range runningCmds {
80 cmd.Process.Signal(os.Interrupt)
82 runningCmdsMutex.Unlock()
84 // Wait for all running crunch jobs to complete / terminate
90 func startFunc(container arvados.Container, cmd *exec.Cmd) error {
94 var startCmd = startFunc
98 // If the container is Locked, start a new crunch-run process and wait until
99 // crunch-run completes. If the priority is set to zero, set an interrupt
100 // signal to the crunch-run process.
102 // If the container is in any other state, or is not Complete/Cancelled after
103 // crunch-run terminates, mark the container as Cancelled.
104 func run(dispatcher *dispatch.Dispatcher,
105 container arvados.Container,
106 status chan arvados.Container) {
108 uuid := container.UUID
110 if container.State == dispatch.Locked {
113 cmd := exec.Command(*crunchRunCommand, uuid)
115 cmd.Stderr = os.Stderr
116 cmd.Stdout = os.Stderr
118 log.Printf("Starting container %v", uuid)
120 // Add this crunch job to the list of runningCmds only if we
121 // succeed in starting crunch-run.
123 runningCmdsMutex.Lock()
124 if err := startCmd(container, cmd); err != nil {
125 runningCmdsMutex.Unlock()
126 log.Printf("Error starting %v for %v: %q", *crunchRunCommand, uuid, err)
127 dispatcher.UpdateState(uuid, dispatch.Cancelled)
129 runningCmds[uuid] = cmd
130 runningCmdsMutex.Unlock()
132 // Need to wait for crunch-run to exit
133 done := make(chan struct{})
136 if _, err := cmd.Process.Wait(); err != nil {
137 log.Printf("Error while waiting for crunch job to finish for %v: %q", uuid, err)
139 log.Printf("sending done")
149 // Interrupt the child process if priority changes to 0
150 if (c.State == dispatch.Locked || c.State == dispatch.Running) && c.Priority == 0 {
151 log.Printf("Sending SIGINT to pid %d to cancel container %v", cmd.Process.Pid, uuid)
152 cmd.Process.Signal(os.Interrupt)
158 log.Printf("Finished container run for %v", uuid)
160 // Remove the crunch job from runningCmds
161 runningCmdsMutex.Lock()
162 delete(runningCmds, uuid)
163 runningCmdsMutex.Unlock()
168 // If the container is not finalized, then change it to "Cancelled".
169 err := dispatcher.Arv.Get("containers", uuid, nil, &container)
171 log.Printf("Error getting final container state: %v", err)
173 if container.LockedByUUID == dispatcher.Auth.UUID &&
174 (container.State == dispatch.Locked || container.State == dispatch.Running) {
175 log.Printf("After %s process termination, container state for %v is %q. Updating it to %q",
176 *crunchRunCommand, container.State, uuid, dispatch.Cancelled)
177 dispatcher.UpdateState(uuid, dispatch.Cancelled)
180 // drain any subsequent status changes
184 log.Printf("Finalized container %v", uuid)