Repository navigation
Expand file tree
/
Copy pathroot.go
More file actions
946 lines (839 loc) · 31.4 KB
/
Copy pathroot.go
File metadata and controls
946 lines (839 loc) · 31.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
// Copyright © 2017-2023 Wei Shen <shenwei356@gmail.com>
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.
package main
import (
"bufio"
"bytes"
"context"
"crypto/sha256"
"fmt"
"os"
"os/signal"
"regexp"
"runtime"
"strings"
"sync/atomic"
"time"
"github.com/pkg/errors"
pb "github.com/schollz/progressbar/v3"
"github.com/shenwei356/rush/internal/runstate"
"github.com/shenwei356/rush/process"
"github.com/shenwei356/util/stringutil"
"github.com/shenwei356/xopen"
"github.com/spf13/cobra"
)
var writeSuccessfulCommandFile = func(w *bufio.Writer, value string) (int, error) { return w.WriteString(value) }
var flushSuccessfulCommandFile = func(w *bufio.Writer) error { return w.Flush() }
var truncateSuccessfulCommandFile = func(f *os.File, size int64) error { return f.Truncate(size) }
var seekSuccessfulCommandFile = func(f *os.File, offset int64, whence int) (int64, error) {
return f.Seek(offset, whence)
}
func appendSuccessfulCommand(w *bufio.Writer, command string) error {
if _, err := writeSuccessfulCommandFile(w, command+endMarkOfCMD); err != nil {
return err
}
return flushSuccessfulCommandFile(w)
}
func rollbackSuccessfulCommands(w *bufio.Writer, f *os.File, size int64) error {
w.Reset(f)
if err := truncateSuccessfulCommandFile(f, size); err != nil {
return err
}
_, err := seekSuccessfulCommandFile(f, 0, os.SEEK_END)
return err
}
// RootCmd represents the base command when called without any subcommands
var RootCmd = &cobra.Command{
Use: "rush",
Short: "a cross-platform command-line tool for executing jobs in parallel",
Long: fmt.Sprintf(`rush -- a cross-platform command-line tool for executing jobs in parallel
Version: %s
Author: Wei Shen <shenwei356@gmail.com>
Homepage: https://github.com/shenwei356/rush
Input:
- Input could be a list of strings or numbers, e.g., file paths.
- Input can be given either from the STDIN or file(s) via the option -i/--infile.
- Some options could be used to defined how the input records are parsed:
-d, --field-delimiter field delimiter in records (default "\s+")
-D, --record-delimiter record delimiter (default "\n")
-n, --nrecords number of records sent to a command (default 1)
-J, --records-join-sep record separator for joining multi-records (default "\n")
--pipe send each group of records to the command's standard input
-T, --trim trim white space (" \t\r\n") in input
Output:
- Outputs of all commands are written to STDOUT by default,
you can also use -o/--out-file to specify a output file.
- Outputs of all commands are random, you can use the flag -k/--keep-order
to keep output in order of input.
- Outputs of all commands are buffered, you can use the flag -I/--immediate-output
to print output immediately and interleaved.
Replacement strings in commands:
{} full data
{n} nth field in delimiter-delimited data
{/} dirname
{%%} basename
{.} remove the last file extension
{:} remove all file extensions.
{^suffix} remove suffix
{@regexp} capture submatch using regular expression.
Limitation: curly brackets can't be used in the regexp.
{#} job ID
{?} a value computed as $cpus / $jobs, which can be used as the number of
threads for each command. This value is dynamically adjusted according
to the number of jobs (-j/--jobs).
With --continue, {#} and {?} are kept stable in the successful-command file,
so changing input order or job count does not rerun otherwise unchanged jobs.
Escaping curly brackets "{}":
{{}} {}
{{1}} {1}
{{1,}} {1,}
{{a}} {a}
Combinations:
{%%.}, {%%:} basename without extension
{2.}, {2/}, {2%%.} manipulate nth field
{file:}, {file:^_1} remove all extensions of a preset variable (see below)
Preset variable (macro):
1. You can pass variables to the command like awk via the option -v. E.g.,
$ seq 3 | rush -v p=prefix_ -v s=_suffix 'echo {p}{}{s}'
prefix_3_suffix
prefix_1_suffix
prefix_2_suffix
2. A variable name should start with a letter and be followed by letters, digits, or underscores.
A regular expression is used to check them: ^[a-zA-Z][A-Za-z0-9_]*$
3. The value could also contain replacement strings.
# {p} will be replaced with {%%:}, which computes the basename and remove all file extensions.
$ echo a/b/c.txt.gz | rush -v 'p={%%:}' 'echo {p} {p}.csv'
c c.csv
`, VERSION),
Run: func(cmd *cobra.Command, args []string) {
process.Log = log
var err error
config := getConfigs(cmd)
if len(config.Infiles) == 0 {
config.Infiles = append(config.Infiles, "-")
}
if config.Version {
checkVersion()
return
}
if len(config.Infiles) == 1 && isStdin(config.Infiles[0]) && !xopen.IsStdin() {
checkError(fmt.Errorf(`STDIN not detected. type "rush -h" for help`))
}
j := config.Jobs + 1
if j > runtime.NumCPU() {
j = runtime.NumCPU()
}
runtime.GOMAXPROCS(j)
config.reFieldDelimiter, err = regexp.Compile(config.FieldDelimiter)
checkError(errors.Wrap(err, "compile field delimiter"))
command0 := strings.Join(args, " ")
if command0 == "" {
if runtime.GOOS == "windows" {
command0 = `echo {}`
} else {
command0 = `echo "{}"`
}
}
// -----------------------------------------------------------------
// out file handler
var outfh *os.File = nil
var errfh *os.File = nil
if isStdin(config.OutFile) {
// use unbuffered file for writing, so we don't lose output on .Exit()
outfh = os.Stdout
errfh = os.Stderr
} else {
// use unbuffered file for writing, so we don't lose output on .Exit()
outfh, err = os.Create(config.OutFile)
checkError(err)
defer outfh.Close()
errfh = outfh
}
// TmpOutputDataBuffer = config.BufferSize
if config.DryRun {
config.Verbose = false
}
var succCmds = make(map[string]struct{})
var bfhSuccCmds *bufio.Writer
var fhSuccCmds *os.File
var succCmdOriginalSize int64
if config.Continue {
var existed bool
existed, err = exists(config.SuccCmdFile)
checkError(err)
if existed {
succCmds = readSuccCmds(config.SuccCmdFile)
fhSuccCmds, err = os.OpenFile(config.SuccCmdFile, os.O_APPEND|os.O_WRONLY, 0664)
} else {
fhSuccCmds, err = os.Create(config.SuccCmdFile)
}
checkError(err)
info, statErr := fhSuccCmds.Stat()
checkError(statErr)
succCmdOriginalSize = info.Size()
defer fhSuccCmds.Close()
bfhSuccCmds = bufio.NewWriter(fhSuccCmds)
}
opts := &process.Options{
DryRun: config.DryRun,
Jobs: config.Jobs,
ETA: config.ETA,
KeepOrder: config.KeepOrder,
Retries: config.Retries,
RetryInterval: time.Duration(config.RetryInterval * float64(time.Second)),
OutFileHandle: outfh,
ErrFileHandle: errfh,
ImmediateOutput: config.ImmediateOutput,
PrintRetryOutput: config.PrintRetryOutput,
Timeout: time.Duration(config.Timeout) * time.Second,
StopOnErr: config.StopOnErr,
NoStopExes: config.NoStopExes,
NoKillExes: config.NoKillExes,
CleanupTime: time.Duration(config.CleanupTime) * time.Second,
StartDelay: config.StartDelay,
MaxLoad: config.MaxLoad,
MinFreeMemory: config.MinFreeMemory,
// Always collect statuses here so timeout retains its documented 124
// even when ordinary child-status propagation is disabled.
PropExitStatus: true,
Verbose: config.Verbose,
RecordSuccessfulCmd: config.Continue,
}
// split function for scanner
recordDelimiter := []byte(config.RecordDelimiter)
split := func(data []byte, atEOF bool) (advance int, token []byte, err error) {
if atEOF && len(data) == 0 {
return 0, nil, nil
}
if len(recordDelimiter) == 0 {
return 1, data[0:1], nil
}
i := bytes.Index(data, recordDelimiter)
if i >= 0 {
return i + len(recordDelimiter), data[0 : i+len(recordDelimiter)], nil
}
if atEOF {
return len(data), data, nil
}
return 0, nil, nil
}
// ---------------------------------------------------------------
state, runCtx := runstate.New(context.TODO())
cancelRun := state.Cancel
defer cancelRun()
chExitSignalMonitor := make(chan struct{})
signalChan := make(chan os.Signal, 2)
cleanupDone := make(chan int)
signal.Notify(signalChan, terminationSignals()...)
go func() {
interrupted := false
for {
select {
case sig := <-signalChan:
if interrupted {
log.Criticalf("received another interrupt, killing unfinished commands...")
state.Force()
opts.ForceStop()
continue
}
interrupted = true
status := terminationStatus(sig)
log.Criticalf("received an interrupt, stopping unfinished commands...")
state.Stop(runstate.Cause{Kind: runstate.Signal, Status: status})
case <-chExitSignalMonitor:
signal.Stop(signalChan)
cleanupDone <- 1
return
}
}
}()
donePreprocessFiles := make(chan int)
// channel of command
chCmdStr := make(chan process.Job, 1)
// run
chOutput, chSuccessfulCmd, doneSendOutput, chExitStatus := process.Run4OutputContextJobs(opts, runCtx, cancelRun, chCmdStr)
doneCheckSuccCmd := make(chan int)
var nSuccCmds atomic.Int64
var succCmdWriteErr error
go func() {
for c := range chSuccessfulCmd {
nSuccCmds.Add(1)
if config.Continue {
if succCmdWriteErr != nil {
continue
}
succCmdWriteErr = appendSuccessfulCommand(bfhSuccCmds, c)
if succCmdWriteErr != nil {
log.Error(errors.Wrap(succCmdWriteErr, "write successful-command file"))
state.Stop(runstate.Cause{Kind: runstate.Internal, Status: 1})
}
}
}
doneCheckSuccCmd <- 1
}()
// generate commands
anyCommands := false
var inputlines []inputRecord
READ_INPUT:
for _, file := range config.Infiles {
if runCtx.Err() != nil {
break
}
// input file handler
var infh *os.File
if isStdin(file) {
infh = os.Stdin
} else {
infh, err = openInputFile(file)
checkError(err)
defer infh.Close()
}
scanner := bufio.NewScanner(newInputReader(infh, runCtx))
inputCloserDone := make(chan struct{})
inputCloserExited := make(chan struct{})
go func(f *os.File) {
defer close(inputCloserExited)
select {
case <-runCtx.Done():
_ = f.Close()
case <-inputCloserDone:
}
}(infh)
// 2147483647: max int32
scanner.Buffer(make([]byte, 0, 16384), 2147483647)
scanner.Split(split)
for scanner.Scan() {
select {
case <-runCtx.Done():
if config.Verbose {
log.Warningf("cancel reading file: %s", file)
}
close(inputCloserDone)
<-inputCloserExited
break READ_INPUT
default:
}
record := scanner.Text()
terminated := false
if len(recordDelimiter) > 0 && strings.HasSuffix(record, config.RecordDelimiter) {
record = record[:len(record)-len(config.RecordDelimiter)]
terminated = true
}
inputlines = append(inputlines, inputRecord{data: record, terminated: terminated})
}
close(inputCloserDone)
<-inputCloserExited
if scanErr := scanner.Err(); scanErr != nil && runCtx.Err() == nil {
state.Stop(runstate.Cause{Kind: runstate.Internal, Status: 1})
checkError(errors.Wrap(scanErr, "read input data"))
}
}
if opts.ETA {
jobcount := len(inputlines) / config.NRecords
if len(inputlines)%config.NRecords > 0 {
jobcount++
}
opts.ETABar = pb.NewOptions(jobcount,
pb.OptionSetWriter(os.Stderr),
pb.OptionSetDescription("processed jobs: "),
pb.OptionShowCount(),
pb.OptionShowIts(),
pb.OptionSetItsString("jobs"),
pb.OptionSetRenderBlankState(true),
pb.OptionSetTheme(pb.Theme{
Saucer: "=",
SaucerHead: ">",
SaucerPadding: "_",
BarStart: "[",
BarEnd: "]",
}))
}
// read data and generate command
go func() {
defer close(chCmdStr)
defer close(donePreprocessFiles)
sendCommand := func(cmdStr, recordCmd, stdin string) bool {
select {
case chCmdStr <- process.Job{Cmd: cmdStr, RecordCmd: recordCmd, Stdin: stdin}:
anyCommands = true
return true
case <-runCtx.Done():
return false
}
}
wasSuccessful := func(recordCmd, cmdStr string) bool {
if _, ok := succCmds[recordCmd]; ok {
return true
}
if config.Pipe {
return false
}
// Older successful-command files contain the expanded {#} value.
_, ok := succCmds[cmdStr]
return ok
}
n := config.NRecords
var id uint64 = 1
var records []string
records = make([]string, 0, n)
pipeRecords := make([]inputRecord, 0, n)
var cmdStr, recordCmd string
var stdin string
var runned bool
nJobs := (len(inputlines) + n - 1) / n
for _, input := range inputlines {
record := input.data
if record == "" {
continue
}
records = append(records, record)
pipeRecords = append(pipeRecords, input)
if len(records) == n {
cmdStr, err = fillCommand(config, command0, Chunk{ID: id, Data: records}, nJobs-int(nSuccCmds.Load()))
checkError(errors.Wrap(err, "fill command"))
recordCmd = cmdStr
if config.Continue {
recordCmd, err = fillCommandForContinue(config, command0, Chunk{ID: id, Data: records}, nJobs-int(nSuccCmds.Load()))
checkError(errors.Wrap(err, "fill command for continue"))
}
if config.Escape {
cmdStr = stringutil.EscapeSymbols(cmdStr, config.EscapeSymbols)
recordCmd = stringutil.EscapeSymbols(recordCmd, config.EscapeSymbols)
}
stdin = ""
if config.Pipe {
stdin = buildPipeInput(pipeRecords, config.RecordDelimiter)
if config.Continue {
recordCmd = pipeContinueKey(recordCmd, stdin)
}
}
if len(cmdStr) > 0 {
if config.Continue {
if runned = wasSuccessful(recordCmd, cmdStr); runned {
if config.Verbose {
log.Infof("ignore cmd: %s", cmdStr)
}
if opts.ETA {
opts.ETABar.Add(1)
fmt.Fprintln(os.Stderr)
}
// bfhSuccCmds.WriteString(cmdStr + endMarkOfCMD)
// bfhSuccCmds.Flush()
} else {
if !sendCommand(cmdStr, recordCmd, stdin) {
return
}
}
} else {
if !sendCommand(cmdStr, recordCmd, stdin) {
return
}
}
id++
}
records = make([]string, 0, n)
pipeRecords = make([]inputRecord, 0, n)
}
}
if len(records) > 0 {
cmdStr, err = fillCommand(config, command0, Chunk{ID: id, Data: records}, nJobs-int(nSuccCmds.Load()))
checkError(errors.Wrap(err, "fill command"))
recordCmd = cmdStr
if config.Continue {
recordCmd, err = fillCommandForContinue(config, command0, Chunk{ID: id, Data: records}, nJobs-int(nSuccCmds.Load()))
checkError(errors.Wrap(err, "fill command for continue"))
}
if config.Escape {
cmdStr = stringutil.EscapeSymbols(cmdStr, config.EscapeSymbols)
recordCmd = stringutil.EscapeSymbols(recordCmd, config.EscapeSymbols)
}
stdin = ""
if config.Pipe {
stdin = buildPipeInput(pipeRecords, config.RecordDelimiter)
if config.Continue {
recordCmd = pipeContinueKey(recordCmd, stdin)
}
}
if len(cmdStr) > 0 {
if config.Continue {
if runned = wasSuccessful(recordCmd, cmdStr); runned {
if config.Verbose {
log.Infof("ignore cmd: %s", cmdStr)
}
// bfhSuccCmds.WriteString(cmdStr + endMarkOfCMD)
// bfhSuccCmds.Flush()
} else {
if !sendCommand(cmdStr, recordCmd, stdin) {
return
}
}
} else {
if !sendCommand(cmdStr, recordCmd, stdin) {
return
}
}
}
}
}()
// ---------------------------------------------------------------
// read from chOutput and print
doneOutput := make(chan int)
var outputWriteErr error
go func() {
last := time.Now().Add(2 * time.Second)
for c := range chOutput {
if outputWriteErr == nil {
_, outputWriteErr = outfh.WriteString(c)
if outputWriteErr != nil {
log.Error(errors.Wrap(outputWriteErr, "write command output"))
state.Stop(runstate.Cause{Kind: runstate.Internal, Status: 1})
}
}
if t := time.Now(); t.After(last) {
last = t.Add(2 * time.Second)
}
}
doneOutput <- 1
}()
var pToolExitStatus *int = nil
var doneExitStatus chan int
doneExitStatus = make(chan int)
toolExitStatus := 0
go func() {
for childCode := range chExitStatus {
setPointer := false
setCode := false
if pToolExitStatus == nil {
setPointer = true
setCode = true
} else {
// use the code from the first error we received
if *pToolExitStatus == 0 && childCode != 0 {
setCode = true
}
}
if setPointer {
pToolExitStatus = &toolExitStatus
}
if setCode {
*pToolExitStatus = childCode
}
}
doneExitStatus <- 1
}()
// ---------------------------------------------------------------
// the order is very important!
<-donePreprocessFiles // finish read data and send command
<-doneSendOutput // finish send output
<-doneOutput // finish print output
<-doneExitStatus
if config.Continue {
<-doneCheckSuccCmd
}
close(chExitSignalMonitor)
<-cleanupDone
cause := state.Cause()
if opts.ETA {
// Only call Finish() if the run completed normally (not interrupted)
if cause.Status == 0 {
opts.ETABar.Finish()
}
os.Stderr.WriteString("\n")
}
if config.Continue && (cause.Kind == runstate.Internal || outputWriteErr != nil || succCmdWriteErr != nil) {
// Discard buffered bytes before restoring the exact pre-run length.
if rollbackErr := rollbackSuccessfulCommands(bfhSuccCmds, fhSuccCmds, succCmdOriginalSize); rollbackErr != nil {
log.Error(errors.Wrap(rollbackErr, "rollback successful-command file"))
cause = runstate.Cause{Kind: runstate.Internal, Status: 1}
}
}
if cause.Status != 0 {
os.Exit(cause.Status)
}
if anyCommands {
if pToolExitStatus != nil {
if *pToolExitStatus == 124 || config.PropExitStatus {
os.Exit(*pToolExitStatus)
}
} else {
checkError(fmt.Errorf(`did not get an exit status int from any child process)`))
}
}
},
}
type inputRecord struct {
data string
terminated bool
}
func buildPipeInput(records []inputRecord, delimiter string) string {
var buf strings.Builder
for i, record := range records {
buf.WriteString(record.data)
if delimiter != "" && (record.terminated || i+1 < len(records)) {
buf.WriteString(delimiter)
}
}
return buf.String()
}
func pipeContinueKey(command, stdin string) string {
digest := sha256.Sum256([]byte(stdin))
return fmt.Sprintf("%s\n# rush --pipe stdin sha256: %x", command, digest)
}
// Chunk contains input data records sent to a command
type Chunk struct {
ID uint64
Data []string
}
// Execute adds all child commands to the root command sets flags appropriately.
// This is called by main.main(). It only needs to happen once to the rootCmd.
func Execute() {
if err := RootCmd.Execute(); err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(-1)
}
}
func init() {
RootCmd.Flags().BoolP("verbose", "", false, "print verbose information")
RootCmd.Flags().BoolP("version", "V", false, `print version information and check for update`)
RootCmd.Flags().BoolP("eta", "", false, `show ETA progress bar`)
RootCmd.Flags().IntP("jobs", "j", runtime.NumCPU(), "run n jobs in parallel (default value depends on your device)")
RootCmd.Flags().Float64("delay", 0, "minimum seconds between starting jobs (supports fractions)")
RootCmd.Flags().String("load", "", "start jobs only while system load is below this value (number or percent of CPUs)")
RootCmd.Flags().String("memfree", "", "minimum available memory before starting jobs (bytes or K/M/G/T/P suffix)")
RootCmd.Flags().StringP("out-file", "o", "-", `out file ("-" for stdout)`)
RootCmd.Flags().StringSliceP("infile", "i", []string{}, "input data file, multi-values supported")
RootCmd.Flags().StringP("record-delimiter", "D", "\n", `record delimiter (default is "\n")`)
RootCmd.Flags().StringP("records-join-sep", "J", "\n", `record separator for joining multi-records (default is "\n")`)
RootCmd.Flags().IntP("nrecords", "n", 1, "number of records sent to a command")
RootCmd.Flags().StringP("field-delimiter", "d", `\s+`, "field delimiter in records, support regular expression")
RootCmd.Flags().Bool("pipe", false, "send each group of records to the command's standard input")
RootCmd.Flags().IntP("retries", "r", 0, "maximum retries (default 0)")
RootCmd.Flags().Float64P("retry-interval", "", 0, "retry interval (unit: second, supports fractions like 0.5) (default 0)")
RootCmd.Flags().BoolP("immediate-output", "I", false, "print output immediately and interleaved, to aid debugging")
RootCmd.Flags().BoolP("print-retry-output", "", true, "print output from retry commands")
RootCmd.Flags().IntP("timeout", "t", 0, "timeout of a command (unit: seconds, 0 for no timeout) (default 0)")
RootCmd.Flags().BoolP("keep-order", "k", false, "keep output in order of input")
RootCmd.Flags().BoolP("stop-on-error", "e", false, "stop scheduling and clean up active child processes on first error")
RootCmd.Flags().BoolP("propagate-exit-status", "", true, "propagate child exit status up to the exit status of rush")
RootCmd.Flags().StringSliceP("no-stop-exes", "", []string{}, "exe names to exclude from stop signal, example: mspdbsrv.exe; or use all for all exes (default none)")
RootCmd.Flags().StringSliceP("no-kill-exes", "", []string{}, "exe names to exclude from kill signal, example: mspdbsrv.exe; or use all for all exes (default none)")
RootCmd.Flags().IntP("cleanup-time", "", 3, "time to allow child processes to clean up between stop / kill signals (unit: seconds, 0 for no time) (default 3)")
RootCmd.Flags().BoolP("dry-run", "", false, "print command but not run")
RootCmd.Flags().BoolP("continue", "c", false, `continue jobs.`+
` NOTES: 1) successful commands are saved in file (given by flag -C/--succ-cmd-file);`+
` 2) if the file does not exist, rush saves data so we can continue jobs next time;`+
` 3) if the file exists, rush ignores jobs in it and update the file;`+
` 4) skipped jobs are silent unless --verbose is used`)
RootCmd.Flags().StringP("succ-cmd-file", "C", "successful_cmds.rush", `file for saving successful commands`)
// RootCmd.Flags().IntP("buffer-size", "", 1, "buffer size for output of a command before saving to tmpfile (unit: Mb)")
RootCmd.Flags().StringSliceP("assign", "v", []string{}, "assign the value val to the variable var (format: var=val, val also supports replacement strings)")
RootCmd.Flags().StringP("trim", "T", "", `trim white space (" \t\r\n") in input (available values: "l" for left, "r" for right, "lr", "rl", "b" for both side)`)
// RootCmd.Flags().BoolP("greedy", "g", false, `greedy replacement strings (replace again), useful for preset variable, e.g., rush -v p={:^_1} 'echo {p}'`)
RootCmd.Flags().BoolP("escape", "q", false, `escape special symbols like $ which you can customize by flag -Q/--escape-symbols`)
RootCmd.Flags().StringP("escape-symbols", "Q", "$#&`", "symbols to escape")
// RootCmd.Flags().BoolP("escape-curly-brackets", "B", false, `escape curly brackets "{}" in the, e.g., "text{}" or "attr{href}"`)
RootCmd.Example = ` 1. simple run, quoting is not necessary
$ seq 1 10 | rush echo {}
2. keep order
$ seq 1 10 | rush 'echo {}' -k
3. timeout
$ seq 1 | rush 'sleep 2; echo {}' -t 1
4. retry
$ seq 1 | rush 'python script.py' -r 3
5. dirname & basename & remove suffix
$ echo dir/file_1.txt.gz | rush 'echo {/} {%} {^_1.txt.gz}'
dir file.txt.gz dir/file
6. basename without the last or any extension
$ echo dir.d/file.txt.gz | rush 'echo {.} {:} {%.} {%:}'
dir.d/file.txt dir.d/file file.txt file
7. job ID, combine fields and other replacement strings
$ echo 12 file.txt dir/s_1.fq.gz | rush 'echo job {#}: {2} {2.} {3%:^_1}'
job 1: file.txt file s
8. capture submatch using regular expression
$ echo read_1.fq.gz | rush 'echo {@(.+)_\d}'
read
9. custom field delimiter
$ echo a=b=c | rush 'echo {1} {2} {3}' -d =
a b c
10. custom record delimiter
$ echo a=b=c | rush -D "=" -k 'echo {}'
a
b
c
$ echo abc | rush -D "" -k 'echo {}'
a
b
c
11. assign value to variable, like "awk -v"
# seq 1 | rush 'echo Hello, {fname} {lname}!' -v fname=Wei,lname=Shen
$ seq 1 | rush 'echo Hello, {fname} {lname}!' -v fname=Wei -v lname=Shen
Hello, Wei Shen!
# preset variables support extra operations as well.
echo read_1.fq.gz | ./rush -v 'p={:^_1}' -v 'f=a.s-10.txt' 'echo {} {p} {f:} {f@s\-(\d+)}'
read_1.fq.gz read a 10
12. preset variable (Macro)
# equal to: echo sample_1.fq.gz | rush 'echo {:^_1} {} {:^_1}_2.fq.gz'
$ echo sample_1.fq.gz | rush -v p={:^_1} 'echo {p} {} {p}_2.fq.gz'
sample sample_1.fq.gz sample_2.fq.gz
13. save successful commands to continue in NEXT run
$ seq 1 3 | rush 'sleep {}; echo {}' -c -t 2
[ERRO] run cmd #1: sleep 2; echo 2: time out
[ERRO] run cmd #2: sleep 3; echo 3: time out
14. escape special symbols
$ seq 1 | rush 'echo -e "a\tb" | awk "{print $1}"' -q
a
15. escape curly brackets "{}"
$ echo aaa bbb ccc | sed -E "s/(\S){3,}/\1/g"
a b c
$ echo 1 | rush 'echo aaa bbb ccc | sed -E "s/(\S){{3,}}/\1/g"' --dry-run
echo aaa bbb ccc | sed -E "s/(\S){3,}/\1/g"
16. run a command with relative paths in Windows, please use backslash as the separator.
# "brename -l -R" is used to search paths recursively
$ brename -l -q -R -i -p "\.go$" | rush "bin\app.exe {}"
17. send a fixed number of records to each command's standard input
$ seq 10000 | rush --pipe -n 1000 -j 4 'wc -l'
More examples: https://github.com/shenwei356/rush`
RootCmd.SetUsageTemplate(`Usage:{{if .Runnable}}
{{if .HasAvailableFlags}}{{appendIfNotPresent .UseLine "[flags]"}}{{else}}{{.UseLine}}{{end}}{{end}}{{if .HasAvailableSubCommands}}
{{ .CommandPath}} [command]{{end}} [command] {{if gt .Aliases 0}}
Aliases:
{{.NameAndAliases}}
{{end}}{{if .HasExample}}
Examples:
{{ .Example }}{{end}}{{ if .HasAvailableSubCommands}}
Available Commands:{{range .Commands}}{{if .IsAvailableCommand}}
{{rpad .Name .NamePadding }} {{.Short}}{{end}}{{end}}{{end}}{{ if .HasAvailableLocalFlags}}
Flags:
{{.LocalFlags.FlagUsagesWrapped 110 | trimTrailingWhitespaces}}{{end}}{{if .HasAvailableInheritedFlags}}
Global Flags:
{{.InheritedFlags.FlagUsagesWrapped 110 | trimTrailingWhitespaces}}{{end}}{{if .HasHelpSubCommands}}
Additional help topics:{{range .Commands}}{{if .IsHelpCommand}}
{{rpad .CommandPath .CommandPathPadding}} {{.Short}}{{end}}{{end}}{{end}}{{ if .HasAvailableSubCommands }}
Use "{{.CommandPath}} [command] --help" for more information about a command.{{end}}
`)
}
// Config is the struct containing all global flags
type Config struct {
Verbose bool
Version bool
ETA bool
Jobs int
StartDelay time.Duration
MaxLoad float64
MinFreeMemory uint64
OutFile string
Infiles []string
RecordDelimiter string
RecordsJoinSeparator string
NRecords int
FieldDelimiter string
reFieldDelimiter *regexp.Regexp
Pipe bool
Retries int
RetryInterval float64
ImmediateOutput bool
PrintRetryOutput bool
Timeout int
KeepOrder bool
StopOnErr bool
NoStopExes []string
NoKillExes []string
CleanupTime int
PropExitStatus bool
DryRun bool
Continue bool
SuccCmdFile string
AssignMap map[string]string
Trim string
Greedy bool
GreedyCount int
Escape bool
EscapeSymbols string
EscapeCurlyBrackets bool
}
// var=value
var reAssign = regexp.MustCompile(`\s*([a-zA-Z][A-Za-z0-9_]*)\s*=(.+)`)
func getConfigs(cmd *cobra.Command) Config {
trim := getFlagString(cmd, "trim")
if trim != "" {
trim = strings.ToLower(trim)
switch trim {
case "l":
case "r":
case "lr", "rl", "b":
default:
checkError(fmt.Errorf(`illegal value for flag --trim: %s. (available values: "l" for left, "r" for right, "lr", "rl", "b" for both side)`, trim))
}
}
assignStrs := getFlagStringSlice(cmd, "assign")
assignMap := make(map[string]string)
for _, s := range assignStrs {
if reAssign.MatchString(s) {
found := reAssign.FindStringSubmatch(s)
assignMap[found[1]] = found[2]
} else {
checkError(fmt.Errorf(`illegal value for flag -v/--assign (format: "var=value", type "rush -h" for more details): %s`, s))
}
}
startDelay, err := parseStartDelay(getFlagNonNegativeFloat64(cmd, "delay"))
checkError(err)
maxLoad, err := parseMaxLoad(getFlagString(cmd, "load"), runtime.NumCPU())
checkError(err)
minFreeMemory, err := parseMemorySize(getFlagString(cmd, "memfree"))
checkError(err)
return Config{
Verbose: getFlagBool(cmd, "verbose"),
Version: getFlagBool(cmd, "version"),
ETA: getFlagBool(cmd, "eta"),
Jobs: getFlagPositiveInt(cmd, "jobs"),
StartDelay: startDelay,
MaxLoad: maxLoad,
MinFreeMemory: minFreeMemory,
OutFile: getFlagString(cmd, "out-file"),
Infiles: getFlagStringSlice(cmd, "infile"),
RecordDelimiter: getFlagString(cmd, "record-delimiter"),
RecordsJoinSeparator: getFlagString(cmd, "records-join-sep"),
NRecords: getFlagPositiveInt(cmd, "nrecords"),
FieldDelimiter: getFlagString(cmd, "field-delimiter"),
Pipe: getFlagBool(cmd, "pipe"),
Retries: getFlagNonNegativeInt(cmd, "retries"),
RetryInterval: getFlagNonNegativeFloat64(cmd, "retry-interval"),
ImmediateOutput: getFlagBool(cmd, "immediate-output"),
PrintRetryOutput: getFlagBool(cmd, "print-retry-output"),
Timeout: getFlagNonNegativeInt(cmd, "timeout"),
KeepOrder: getFlagBool(cmd, "keep-order"),
StopOnErr: getFlagBool(cmd, "stop-on-error"),
NoStopExes: getFlagStringSlice(cmd, "no-stop-exes"),
NoKillExes: getFlagStringSlice(cmd, "no-kill-exes"),
CleanupTime: getFlagNonNegativeInt(cmd, "cleanup-time"),
PropExitStatus: getFlagBool(cmd, "propagate-exit-status"),
DryRun: getFlagBool(cmd, "dry-run"),
Continue: getFlagBool(cmd, "continue"),
SuccCmdFile: getFlagString(cmd, "succ-cmd-file"),
Trim: trim,
AssignMap: assignMap,
Greedy: true, // getFlagBool(cmd, "greedy"),
GreedyCount: 1,
Escape: getFlagBool(cmd, "escape"),
EscapeSymbols: getFlagString(cmd, "escape-symbols"),
EscapeCurlyBrackets: true, // getFlagBool(cmd, "escape-curly-brackets"),
}
}
var reCurlyBrackets = regexp.MustCompile(`\{([^\{\}]*?)\}`)