@@ -7,11 +7,11 @@ import (
7
7
"os/signal"
8
8
"os/user"
9
9
"strings"
10
- "sync"
11
10
"syscall"
12
11
"time"
13
12
14
13
ps "github.com/mitchellh/go-ps"
14
+ "github.com/sourcegraph/conc/pool"
15
15
"github.com/spf13/cobra"
16
16
)
17
17
@@ -46,6 +46,7 @@ const (
46
46
_partWorkDir = "wd"
47
47
_partWorkDirShort = "wd_trim"
48
48
49
+ _partPid = "pid"
49
50
_partPidShell = "pid_shell"
50
51
_partPidShellExec = "pid_shell_exec"
51
52
_partPidParent = "pid_parent"
@@ -82,16 +83,16 @@ func handleQUIT() context.CancelFunc {
82
83
sig := make (chan os.Signal , 1 )
83
84
signal .Notify (sig , os .Interrupt , os .Kill , syscall .SIGTERM )
84
85
85
- // defer flog(" terminating")
86
+ defer debugLog ( "quit: terminating" )
86
87
87
88
// Stdout watchdog
88
89
go func () {
89
- // flog ("start watchdog " + fmt.Sprintf("%d", os.Getppid()))
90
+ // debugLog ("start watchdog " + fmt.Sprintf("%d", os.Getppid()))
90
91
defer bgctxCancel ()
91
92
92
93
for {
93
94
if _ , err := os .Stdout .Stat (); err != nil {
94
- // flog(" terminating early")
95
+ debugLog ( "quit: terminating early" )
95
96
return
96
97
}
97
98
@@ -100,7 +101,7 @@ func handleQUIT() context.CancelFunc {
100
101
case <- tick :
101
102
continue
102
103
case <- sig :
103
- // flog(" terminating early")
104
+ debugLog ( "quit: terminating early" )
104
105
return
105
106
case <- bgctx .Done ():
106
107
return
@@ -112,6 +113,7 @@ func handleQUIT() context.CancelFunc {
112
113
}
113
114
114
115
func cmdQueryRun (_ * cobra.Command , _ []string ) error {
116
+ debugLog ("query: start" )
115
117
defer bgctxCancel ()
116
118
117
119
printerStop , printPart := startPrinter ()
@@ -132,7 +134,7 @@ func cmdQueryRun(_ *cobra.Command, _ []string) error {
132
134
}()
133
135
}
134
136
135
- tasks := new ( AsyncTaskDispatcher )
137
+ tasks := pool . New (). WithContext ( bgctx )
136
138
defer func () {
137
139
tasks .Wait ()
138
140
printPart ("done" , "ok" )
@@ -145,7 +147,7 @@ func cmdQueryRun(_ *cobra.Command, _ []string) error {
145
147
printPart (_partStatus , fmt .Sprintf ("%#v" , * flgQCmdStatus ))
146
148
}
147
149
148
- tasks .Dispatch (func () {
150
+ tasks .Go (func (ctx context. Context ) error {
149
151
homeDir := os .Getenv ("HOME" )
150
152
151
153
if wd , err := os .Getwd (); err == nil {
@@ -173,12 +175,14 @@ func cmdQueryRun(_ *cobra.Command, _ []string) error {
173
175
printPart (_partDuration , diff )
174
176
}
175
177
}
178
+
179
+ return nil
176
180
})
177
181
178
- tasks .Dispatch (func () {
182
+ tasks .Go (func (_ context. Context ) error {
179
183
psChain , err := moduleFindProcessChain ()
180
184
if err != nil {
181
- return
185
+ return nil
182
186
}
183
187
184
188
if len (psChain ) > 3 {
@@ -206,16 +210,18 @@ func cmdQueryRun(_ *cobra.Command, _ []string) error {
206
210
printPart (_partPidRemote , pidRemote .Pid ())
207
211
printPart (_partPidRemoteExec , pidShellRemoteExecName )
208
212
}
213
+
214
+ return nil
209
215
})
210
216
211
- tasks .Dispatch (func () {
217
+ tasks .Go (func (context. Context ) error {
212
218
subTasks := new (AsyncTaskDispatcher )
213
219
defer subTasks .Wait ()
214
220
215
221
if _ , err := stringExec ("git" , "rev-parse" , "--show-toplevel" ); err == nil {
216
222
printPart (_partVcs , "git" )
217
223
} else {
218
- return
224
+ return nil
219
225
}
220
226
221
227
subTasks .Dispatch (func () {
@@ -288,9 +294,11 @@ func cmdQueryRun(_ *cobra.Command, _ []string) error {
288
294
printPart (_partVcsLogBehind , parts [1 ])
289
295
}
290
296
})
297
+
298
+ return nil
291
299
})
292
300
293
- tasks .Dispatch (func () {
301
+ tasks .Go (func (context. Context ) error {
294
302
var err error
295
303
296
304
subTasks := new (AsyncTaskDispatcher )
@@ -301,7 +309,7 @@ func cmdQueryRun(_ *cobra.Command, _ []string) error {
301
309
printPart (_partVcsStg , "1" )
302
310
printPart (_partVcsStgQlen , stgSeriesLen )
303
311
} else {
304
- return
312
+ return nil
305
313
}
306
314
307
315
subTasks .Dispatch (func () {
@@ -314,7 +322,7 @@ func cmdQueryRun(_ *cobra.Command, _ []string) error {
314
322
if stgPatchTop , err = stringExec ("stg" , "top" ); err == nil {
315
323
printPart (_partVcsStgTop , stgPatchTop )
316
324
} else {
317
- return
325
+ return nil
318
326
}
319
327
320
328
subTasks .Dispatch (func () {
@@ -327,29 +335,13 @@ func cmdQueryRun(_ *cobra.Command, _ []string) error {
327
335
printPart (_partVcsStgDirty , 0 )
328
336
}
329
337
})
338
+
339
+ return nil
330
340
})
331
341
332
342
return nil
333
343
}
334
344
335
- func startPrinter () (func (), func (name string , value interface {})) {
336
- printCH := make (chan shellKV )
337
- printerWG := new (sync.WaitGroup )
338
- printerWG .Add (1 )
339
- go func () {
340
- defer printerWG .Done ()
341
- shellKVStaggeredPrinter (printCH , 20 * time .Millisecond , 600 * time .Millisecond )
342
- }()
343
- printerStop := func () {
344
- close (printCH )
345
- printerWG .Wait ()
346
- }
347
- printPart := func (name string , value interface {}) {
348
- printCH <- shellKV {name , value }
349
- }
350
- return printerStop , printPart
351
- }
352
-
353
345
func moduleFindProcessChain () ([]ps.Process , error ) {
354
346
psPTR := os .Getpid ()
355
347
var pidChain []ps.Process
@@ -368,3 +360,27 @@ func moduleFindProcessChain() ([]ps.Process, error) {
368
360
369
361
return pidChain , nil
370
362
}
363
+
364
+
365
+ func startPrinter () (func (), func (name string , value interface {})) {
366
+ debugLog ("query-printer: start" )
367
+ defer debugLog ("query-printer: stop" )
368
+
369
+ printCH := make (chan shellKV )
370
+ doneSIG := make (chan struct {})
371
+ go func () {
372
+ defer close (doneSIG )
373
+ shellKVStaggeredPrinter (printCH , 20 * time .Millisecond , 100 * time .Millisecond )
374
+ }()
375
+
376
+ printerStop := func () {
377
+ close (printCH )
378
+ <- doneSIG
379
+ }
380
+ printPart := func (name string , value interface {}) {
381
+ printCH <- shellKV {name , value }
382
+ }
383
+
384
+ printPart (_partPid , os .Getpid ())
385
+ return printerStop , printPart
386
+ }
0 commit comments