Episode 4: When Something Fails Mid-Run

See what a node that raises mid-run leaves behind: a stopped pipeline, a condition to poll, a failure entry that names the node, and a file that ends early but intact.

Episode 2 dealt with a pipeline that could never run; that refusal reaches you from start(), because it is knowable before a single sample flows. This episode is about the other kind. The pipeline starts, runs for a while, and then a node raises on some frame. Nobody is waiting on a return value at that point, since the call came from a source thread, so the failure cannot be raised at you. It is recorded instead, and this episode shows where to find it.

A node built to break

It passes data through and raises on its twentieth frame. It is deliberately broken and exists only to break; writing nodes properly is Season 7.

17class Flaky(gp.IONode):
18    """Passes data through, then raises on the twentieth frame."""
19
20    def __init__(self, **kwargs):
21        super().__init__(**kwargs)
22        self._seen = 0
23
24    def step(self, data):
25        self._seen += 1
26        if self._seen == 20:
27            raise RuntimeError("the twentieth frame was too much for me")
28        return {gp.Constants.Defaults.PORT_OUT:
29                data[gp.Constants.Defaults.PORT_IN]}
Start, and wait for it

start() returns normally: the pipeline was viable and the run did start; the failure has not happened yet. The script then polls get_condition() until it reports the error condition, the same check a status indicator in an application would make.

34    p = gp.Pipeline()
35    source = gp.Generator(sampling_rate=250, channel_count=4)
36    flaky = Flaky(name="flaky")
37    p.connect(source, flaky)
38    p.connect(flaky, gp.CsvWriter(file_name=FILE))
39
40    # start() returns normally: the pipeline was viable and the run did
41    # start; the failure has not happened yet
42    p.start()
43    print("started; the pipeline was fine")
44
45    # Wait for it to go wrong, the same check a status indicator polls
46    deadline = time.time() + 10
47    while time.time() < deadline:
48        if p.get_condition() == gp.Constants.Conditions.ERROR:
49            break
50        time.sleep(0.05)
Read the failure

p.failure is the log entry of the failure: its message, the node’s instance name, and where it was raised.

52    print()
53    print("condition :", p.get_condition())
54    failure = p.failure
55    if failure:
56        # A log entry, subscripted like a dict
57        print("message   :", failure["message"])
58        print("node      :", failure["source"].get("instance"))
59        print("raised in :", failure["source"].get("summary"))
60
61    p.stop()
62    p.close()
What the file holds

The CsvWriter appended a timestamp to the name, and the stop closed the file properly: comment lines, one header line, then the rows that were written.

64    # The writer appended a timestamp to the name, and closed the file
65    # properly: comment lines, one header line, then the rows
66    written = sorted(glob.glob(FILE.replace(".csv", "_*.csv")))
67    rows = 0
68    if written:
69        with open(written[-1]) as handle:
70            rows = sum(1 for line in handle if not line.startswith("#")) - 1
71    print()
72    print("file      :", os.path.basename(written[-1]) if written else "-")
73    print("rows      :", rows)
Result

Run the script. The pipeline prints a banner of its own when it stops, naming the error, the node and the log file; the script’s lines are these:

started; the pipeline was fine

condition : Error
message   : RuntimeError: the twentieth frame was too much for me
node      : flaky
raised in : Flaky.step() [s6e4_when_something_fails_mid_run.py:27]

file      : s6e4_interrupted_<date>_<time>.csv
rows      : 76

Seventy-six rows are nineteen frames of four samples: every sample up to the frame that failed, and nothing after it.

Three things to take away

A failing node stops the whole pipeline. The first error ends the run. g.Pype does not carry on with the rest of the pipeline, because the nodes downstream would go on emitting results computed from data that stopped arriving.

The recording is short, not corrupt. The stop closes the file properly, so what is on disk holds every sample up to the failing frame and simply ends early. Check get_condition() before you trust a run that finished quietly; a file that ends early is the only other clue you get.

The entry names the node. That is what the four characters of name= buy you: flaky is findable in a log, and a pipeline with three unnamed nodes of the same class is not.

Summary

In this episode, you learned how to:

  • poll get_condition() for a failure the pipeline cannot raise at you

  • read p.failure for the message, the node and the place

  • trust a short recording, and distrust a quiet run you did not check

That closes Season 6. Proceed to Season 7, where you write nodes of your own.

Nodes in this episode: Generator, CsvWriter

File s6e4_when_something_fails_mid_run.py – View file on GitHub

 1"""Season 6, Episode 4: a node that fails mid-run, and what it leaves behind.
 2
 3The failing node is deliberately broken and exists only to break;
 4writing nodes properly is Season 7.
 5"""
 6
 7import glob
 8import os
 9import tempfile
10import time
11
12import gpype as gp
13
14FILE = os.path.join(tempfile.gettempdir(), "s6e4_interrupted.csv")
15
16
17class Flaky(gp.IONode):
18    """Passes data through, then raises on the twentieth frame."""
19
20    def __init__(self, **kwargs):
21        super().__init__(**kwargs)
22        self._seen = 0
23
24    def step(self, data):
25        self._seen += 1
26        if self._seen == 20:
27            raise RuntimeError("the twentieth frame was too much for me")
28        return {gp.Constants.Defaults.PORT_OUT:
29                data[gp.Constants.Defaults.PORT_IN]}
30
31
32if __name__ == "__main__":
33
34    p = gp.Pipeline()
35    source = gp.Generator(sampling_rate=250, channel_count=4)
36    flaky = Flaky(name="flaky")
37    p.connect(source, flaky)
38    p.connect(flaky, gp.CsvWriter(file_name=FILE))
39
40    # start() returns normally: the pipeline was viable and the run did
41    # start; the failure has not happened yet
42    p.start()
43    print("started; the pipeline was fine")
44
45    # Wait for it to go wrong, the same check a status indicator polls
46    deadline = time.time() + 10
47    while time.time() < deadline:
48        if p.get_condition() == gp.Constants.Conditions.ERROR:
49            break
50        time.sleep(0.05)
51
52    print()
53    print("condition :", p.get_condition())
54    failure = p.failure
55    if failure:
56        # A log entry, subscripted like a dict
57        print("message   :", failure["message"])
58        print("node      :", failure["source"].get("instance"))
59        print("raised in :", failure["source"].get("summary"))
60
61    p.stop()
62    p.close()
63
64    # The writer appended a timestamp to the name, and closed the file
65    # properly: comment lines, one header line, then the rows
66    written = sorted(glob.glob(FILE.replace(".csv", "_*.csv")))
67    rows = 0
68    if written:
69        with open(written[-1]) as handle:
70            rows = sum(1 for line in handle if not line.startswith("#")) - 1
71    print()
72    print("file      :", os.path.basename(written[-1]) if written else "-")
73    print("rows      :", rows)