Skip to content

Commit 1329d2c

Browse files
zix99zix99
authored
Update exit signals to respect interrupts (#166)
* Update exit signals to respect interrupts * Add interrupt detection across accumulators
1 parent e210a9d commit 1329d2c

11 files changed

Lines changed: 78 additions & 49 deletions

File tree

‎cmd/analyze.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -65,15 +65,15 @@ func analyzeFunction(c *cli.Context) error {
6565
batcher := helpers.BuildBatcherFromArguments(c)
6666
ext := helpers.BuildExtractorFromArguments(c, batcher)
6767

68-
helpers.RunAggregationLoop(ext, aggr, func() {
68+
interrupt := helpers.RunAggregationLoop(ext, aggr, func() {
6969
line := writeAggrOutput(writer, aggr, extra, quantiles)
7070
writer.WriteForLine(line+1, helpers.BuildExtractorSummary(ext, aggr.ParseErrors()))
7171
writer.WriteForLine(line+2, batcher.StatusString())
7272
})
7373

7474
writer.Close()
7575

76-
return helpers.DetermineErrorState(batcher, ext, aggr)
76+
return helpers.DetermineErrorState(interrupt, batcher, ext, aggr)
7777
}
7878

7979
func analyzeCommand() *cli.Command {

‎cmd/bargraph.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,7 @@ func bargraphFunction(c *cli.Context) error {
4343
ext := helpers.BuildExtractorFromArguments(c, batcher)
4444
sorter := helpers.BuildSorterOrFail(sortName)
4545

46-
helpers.RunAggregationLoop(ext, counter, func() {
46+
interrupt := helpers.RunAggregationLoop(ext, counter, func() {
4747
line := 0
4848

4949
writer.SetKeys(counter.SubKeys()...)
@@ -62,7 +62,7 @@ func bargraphFunction(c *cli.Context) error {
6262
return err
6363
}
6464

65-
return helpers.DetermineErrorState(batcher, ext, counter)
65+
return helpers.DetermineErrorState(interrupt, batcher, ext, counter)
6666
}
6767

6868
func bargraphCommand() *cli.Command {

‎cmd/filter.go‎

Lines changed: 45 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package cmd
33
import (
44
"bufio"
55
"os"
6+
"os/signal"
67
"unicode/utf8"
78

89
"github.com/zix99/rare/cmd/helpers"
@@ -27,40 +28,56 @@ func filterFunction(c *cli.Context, fileGlobs ...string) error {
2728

2829
stdout := bufio.NewWriter(os.Stdout)
2930

31+
exitSignal := make(chan os.Signal, 1)
32+
signal.Notify(exitSignal, os.Interrupt)
33+
interrupted := false
34+
35+
readChan := extractor.ReadFull()
36+
3037
OUTER_LOOP:
31-
for matchBatch := range extractor.ReadFull() {
32-
for _, match := range matchBatch {
33-
if writeLines {
34-
color.WriteString(stdout, color.BrightGreen, match.Source)
35-
stdout.WriteByte(' ')
36-
color.WriteUint64(stdout, color.BrightYellow, match.LineNumber)
37-
stdout.WriteString(": ")
38+
for {
39+
select {
40+
case <-exitSignal:
41+
interrupted = true
42+
break OUTER_LOOP
43+
case matchBatch, more := <-readChan:
44+
if !more {
45+
break OUTER_LOOP
3846
}
3947

40-
switch {
41-
case customExtractor:
42-
stdout.WriteString(match.Extracted)
43-
case onlyText && !utf8.ValidString(match.Line):
44-
color.WriteString(stdout, color.BrightBlue, "Binary Match")
45-
case len(match.Indices) == 2:
46-
// Single match, highlight entire phrase
47-
color.WrapIndices(stdout, match.Line, match.Indices)
48-
default:
49-
// Multi-match groups, highlight individual groups
50-
color.WrapIndices(stdout, match.Line, match.Indices[2:])
48+
for _, match := range matchBatch {
49+
if writeLines {
50+
color.WriteString(stdout, color.BrightGreen, match.Source)
51+
stdout.WriteByte(' ')
52+
color.WriteUint64(stdout, color.BrightYellow, match.LineNumber)
53+
stdout.WriteString(": ")
54+
}
55+
56+
switch {
57+
case customExtractor:
58+
stdout.WriteString(match.Extracted)
59+
case onlyText && !utf8.ValidString(match.Line):
60+
color.WriteString(stdout, color.BrightBlue, "Binary Match")
61+
case len(match.Indices) == 2:
62+
// Single match, highlight entire phrase
63+
color.WrapIndices(stdout, match.Line, match.Indices)
64+
default:
65+
// Multi-match groups, highlight individual groups
66+
color.WrapIndices(stdout, match.Line, match.Indices[2:])
67+
}
68+
stdout.WriteByte('\n')
69+
70+
readLines++
71+
if numLineLimit > 0 && readLines >= numLineLimit {
72+
break OUTER_LOOP
73+
}
5174
}
52-
stdout.WriteByte('\n')
5375

54-
readLines++
55-
if numLineLimit > 0 && readLines >= numLineLimit {
56-
break OUTER_LOOP
76+
// Flush after each batch to make file-following work as expected
77+
if err := stdout.Flush(); err != nil {
78+
logger.Fatal(helpers.ExitCodeOutputError, err)
5779
}
5880
}
59-
60-
// Flush after each batch to make file-following work as expected
61-
if err := stdout.Flush(); err != nil {
62-
logger.Fatal(helpers.ExitCodeOutputError, err)
63-
}
6481
}
6582

6683
// Final flush
@@ -80,7 +97,7 @@ OUTER_LOOP:
8097
}
8198
os.Stderr.WriteString("\n")
8299

83-
return helpers.DetermineErrorState(batcher, extractor, nil)
100+
return helpers.DetermineErrorState(interrupted, batcher, extractor, nil)
84101
}
85102

86103
func getFilterArgs(isSearch bool) []cli.Flag {

‎cmd/heatmap.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -47,7 +47,7 @@ func heatmapFunction(c *cli.Context) error {
4747
writer.Scaler = helpers.BuildScalerOrFail(scalerName)
4848
writer.Formatter = helpers.BuildFormatterOrFail(formatName)
4949

50-
helpers.RunAggregationLoop(ext, counter, func() {
50+
interrupt := helpers.RunAggregationLoop(ext, counter, func() {
5151
writer.WriteTable(counter, rowSorter, colSorter)
5252
writer.WriteFooter(0, helpers.BuildExtractorSummary(ext, counter.ParseErrors(),
5353
fmt.Sprintf("(R: %v; C: %v)", color.Wrapi(color.Yellow, counter.RowCount()), color.Wrapi(color.BrightBlue, counter.ColumnCount()))))
@@ -60,7 +60,7 @@ func heatmapFunction(c *cli.Context) error {
6060
return err
6161
}
6262

63-
return helpers.DetermineErrorState(batcher, ext, counter)
63+
return helpers.DetermineErrorState(interrupt, batcher, ext, counter)
6464
}
6565

6666
func heatmapCommand() *cli.Command {

‎cmd/helpers/exitCodes.go‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ const (
99
ExitCodeInvalidUsage = 2
1010
ExitCodeReadError = 3
1111
ExitCodeOutputError = 4
12+
ExitCodeSigInt = 128 + 2 // 2 is SIGINT
1213
)
1314

1415
type (
@@ -23,7 +24,10 @@ type (
2324
}
2425
)
2526

26-
func DetermineErrorState(b BatcherErrors, e ExtractorSummary, agg AggregationErrors) error {
27+
func DetermineErrorState(interrupt bool, b BatcherErrors, e ExtractorSummary, agg AggregationErrors) error {
28+
if interrupt {
29+
return cli.Exit("", ExitCodeSigInt)
30+
}
2731
if b.ReadErrors() > 0 {
2832
return cli.Exit("Read errors", ExitCodeReadError)
2933
}

‎cmd/helpers/exitCodes_test.go‎

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -24,14 +24,17 @@ func (s *mockExitState) MatchedLines() uint64 {
2424

2525
func TestDetermineErrorState(t *testing.T) {
2626
s := mockExitState{0, 0, 1}
27-
assert.NoError(t, DetermineErrorState(&s, &s, &s))
27+
assert.NoError(t, DetermineErrorState(false, &s, &s, &s))
2828

2929
s = mockExitState{0, 0, 0}
30-
assert.Error(t, DetermineErrorState(&s, &s, &s))
30+
assert.Error(t, DetermineErrorState(false, &s, &s, &s))
3131

3232
s = mockExitState{0, 1, 1}
33-
assert.Error(t, DetermineErrorState(&s, &s, &s))
33+
assert.Error(t, DetermineErrorState(false, &s, &s, &s))
3434

3535
s = mockExitState{1, 0, 1}
36-
assert.Error(t, DetermineErrorState(&s, &s, &s))
36+
assert.Error(t, DetermineErrorState(false, &s, &s, &s))
37+
38+
s = mockExitState{0, 0, 1}
39+
assert.Error(t, DetermineErrorState(true, &s, &s, &s))
3740
}

‎cmd/helpers/updatingAggregator.go‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@ import (
1818
// writeOutput - triggered after a delay, only if there's an update
1919
//
2020
// The two functions are guaranteed to never happen at the same time
21-
func RunAggregationLoop(ext *extractor.Extractor, aggregator aggregation.Aggregator, writeOutput func()) {
21+
func RunAggregationLoop(ext *extractor.Extractor, aggregator aggregation.Aggregator, writeOutput func()) (interrupt bool) {
2222
logger.DeferLogs()
2323

2424
// Updater sync variables
@@ -47,6 +47,7 @@ PROCESSING_LOOP:
4747
for {
4848
select {
4949
case <-exitSignal:
50+
interrupt = true
5051
break PROCESSING_LOOP
5152
case matchBatch, more := <-reader:
5253
if !more {
@@ -62,4 +63,6 @@ PROCESSING_LOOP:
6263
outputDone <- true
6364

6465
writeOutput()
66+
67+
return
6568
}

‎cmd/histo.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,7 @@ func histoFunction(c *cli.Context) error {
5757
fmt.Sprintf("(Groups: %s)", color.Wrapi(color.BrightBlue, counter.GroupCount())))
5858
}
5959

60-
helpers.RunAggregationLoop(ext, counter, func() {
60+
interrupt := helpers.RunAggregationLoop(ext, counter, func() {
6161
writeHistoOutput(writer, counter, topItems, sorter, atLeast)
6262
writer.WriteFooter(0, progressString())
6363
writer.WriteFooter(1, batcher.StatusString())
@@ -81,7 +81,7 @@ func histoFunction(c *cli.Context) error {
8181
return err
8282
}
8383

84-
return helpers.DetermineErrorState(batcher, ext, counter)
84+
return helpers.DetermineErrorState(interrupt, batcher, ext, counter)
8585
}
8686

8787
// HistogramCommand Exported command

‎cmd/reduce.go‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -72,6 +72,8 @@ func reduceFunction(c *cli.Context) error {
7272
formatters := buildFormatterSetOrFail(aggr, formatNames...)
7373

7474
// run the aggregation
75+
interrupted := false
76+
7577
if aggr.GroupColCount() > 0 || table {
7678
// Table output
7779
table := termrenderers.NewTable(vt, colCount, rowCount)
@@ -88,7 +90,7 @@ func reduceFunction(c *cli.Context) error {
8890
table.WriteRow(0, rowBuf...)
8991
}
9092

91-
helpers.RunAggregationLoop(extractor, aggr, func() {
93+
interrupted = helpers.RunAggregationLoop(extractor, aggr, func() {
9294
// write data
9395
for i, group := range aggr.Groups(sorter) {
9496
rowBuf := make([]string, aggr.ColCount())
@@ -110,7 +112,7 @@ func reduceFunction(c *cli.Context) error {
110112
})
111113
} else {
112114
// Simple output
113-
helpers.RunAggregationLoop(extractor, aggr, func() {
115+
interrupted = helpers.RunAggregationLoop(extractor, aggr, func() {
114116
items := aggr.Data("")
115117
colNames := aggr.DataCols()
116118
for idx, expr := range items {
@@ -127,7 +129,7 @@ func reduceFunction(c *cli.Context) error {
127129
return err
128130
}
129131

130-
return helpers.DetermineErrorState(batcher, extractor, aggr)
132+
return helpers.DetermineErrorState(interrupted, batcher, extractor, aggr)
131133
}
132134

133135
func parseKeyValInitial(s, defaultInitial string) (key, initial, val string) {

‎cmd/spark.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@ func sparkFunction(c *cli.Context) error {
3838
writer.Scaler = helpers.BuildScalerOrFail(scalerName)
3939
writer.Formatter = helpers.BuildFormatterOrFail(formatName)
4040

41-
helpers.RunAggregationLoop(ext, counter, func() {
41+
interrupt := helpers.RunAggregationLoop(ext, counter, func() {
4242

4343
// Trim unused data from the data store (keep memory tidy!)
4444
if !noTruncate {
@@ -69,7 +69,7 @@ func sparkFunction(c *cli.Context) error {
6969
return err
7070
}
7171

72-
return helpers.DetermineErrorState(batcher, ext, counter)
72+
return helpers.DetermineErrorState(interrupt, batcher, ext, counter)
7373
}
7474

7575
func sparkCommand() *cli.Command {

0 commit comments

Comments
 (0)