From 3fa23e24d3dd396cd726026ae740f53f472afee1 Mon Sep 17 00:00:00 2001 From: Sahil Lenka <76817449+Sahil-u07@users.noreply.github.com> Date: Wed, 14 Oct 2026 00:55:41 +0530 Subject: [PATCH] fix: move simtime forward on the read after a write Since #431 write() no longer touches simtime, so a node only moves its clock when it reads. A plant that writes with delta=1 never reaches maxtime by itself any more. If its controller finishes first and writes its init values back to the port before the plant has read the last real value, the plant reads the reset value, stays one step short of maxtime and waits forever (#608). It shows up when the plant is a bit slower than the controller, e.g. on Windows or with MATLAB/Verilog. write() still leaves simtime alone, so writes in the same step keep the same timestamp (#385). It now remembers simtime + delta, and the next read moves simtime up to that before reading. If simtime was changed in between (initval, or set by hand) the pending advance is dropped. Done the same way in Python, C++ (concore.hpp, concoredocker.hpp), Java, MATLAB/Octave and Julia (concore.jl, concoredocker.jl), with a test in each. tests/test_end_of_run.py runs a controller and a slower plant as two processes and hangs without this change. Verilog needs #604 first since readdata still overwrites simtime there. Fixes #608 --- .github/workflows/ci.yml | 2 +- ConcoreJavaRuntimeCore.java | 29 +++++++++++- TestConcoreApi.java | 20 ++++++++ TestConcoreHpp.cpp | 31 ++++++++++++ concore.hpp | 23 +++++++++ concore.jl | 27 +++++++++++ concore_base.py | 16 +++++++ concore_read.m | 4 ++ concore_write.m | 6 +++ concoredocker.hpp | 23 +++++++++ concoredocker.jl | 27 +++++++++++ tests/julia/test_protocol.jl | 24 ++++++++++ tests/matlab/test_write_advance.m | 38 +++++++++++++++ tests/test_concore.py | 63 +++++++++++++++++++++++++ tests/test_end_of_run.py | 75 ++++++++++++++++++++++++++++++ tests/test_protocol_conformance.py | 2 + 16 files changed, 408 insertions(+), 2 deletions(-) create mode 100644 tests/matlab/test_write_advance.m create mode 100644 tests/test_end_of_run.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c297b79..07f2a03 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -126,7 +126,7 @@ jobs: - name: Run MATLAB interop and wire-format tests uses: matlab-actions/run-command@v3 with: - command: addpath(pwd); addpath(fullfile(pwd, 'tests', 'matlab')); test_julia_interop; test_literal_eval + command: addpath(pwd); addpath(fullfile(pwd, 'tests', 'matlab')); test_julia_interop; test_literal_eval; test_write_advance - name: Read MATLAB data from Julia run: julia --color=yes tests/julia/test_matlab_interop.jl read diff --git a/ConcoreJavaRuntimeCore.java b/ConcoreJavaRuntimeCore.java index 3120e1e..1c9b905 100644 --- a/ConcoreJavaRuntimeCore.java +++ b/ConcoreJavaRuntimeCore.java @@ -124,7 +124,26 @@ private static Map parseFile(String filename) throws IOException void setOutPath(String path) { outpath = path; } void setDelay(int ms) { delay = ms; } double getSimtime() { return simtime; } - void resetState() { s = ""; olds = ""; simtime = 0; } + void resetState() { s = ""; olds = ""; simtime = 0; hasWrite = false; } + + // A write moves simtime forward at the next read (issue #608). The write + // itself leaves simtime alone so writes in one step share a timestamp. + private boolean hasWrite = false; + private double writeBase = 0, writeTarget = 0; + + private void noteWrite(int delta) { + if (!hasWrite || writeBase != simtime) { + hasWrite = true; + writeBase = writeTarget = simtime; + } + writeTarget = Math.max(writeTarget, simtime + delta); + } + + private void applyWrite() { + if (hasWrite && writeBase == simtime) { + simtime = Math.max(simtime, writeTarget); + } + } boolean unchanged() { if (olds.equals(s)) { @@ -158,6 +177,7 @@ void defaultMaxTime(double defaultValue) { } ReadResult readFilePort(int port, String name, String initstr) { + applyWrite(); List defaultVal = new ArrayList<>(); try { List parsed = (List) literalEval(initstr); @@ -259,6 +279,9 @@ void writeFilePort(int port, String name, Object val, int delta) { return; } Files.write(pathObj, content.toString().getBytes()); + if (!(val instanceof String)) { + noteWrite(delta); + } } catch (InterruptedException e) { Thread.currentThread().interrupt(); System.out.println("skipping " + outpath + "/" + port + "/" + name); @@ -323,6 +346,7 @@ private static int zmqSocketTypeFromString(String s) { } ReadResult readZmqPort(String portName, String name, String initstr) { + applyWrite(); List defaultVal = new ArrayList<>(); try { List parsed = (List) literalEval(initstr); @@ -378,6 +402,9 @@ void writeZmqPort(String portName, String name, Object val, int delta) { return; } port.sendWithRetry(payload); + if (val instanceof List) { + noteWrite(delta); + } } private static String escapePythonString(String s) { diff --git a/TestConcoreApi.java b/TestConcoreApi.java index fd56dfc..9ff73ba 100644 --- a/TestConcoreApi.java +++ b/TestConcoreApi.java @@ -19,6 +19,7 @@ public static void main(String[] args) { testReadParsesFileAndStripsSimtime(); testReadWriteRoundtrip(); testSimtimeAdvancesWithDelta(); + testWriteAdvancesSimtimeOnNextRead(); testUnchangedReturnsFalseAfterRead(); testUnchangedReturnsTrueOnSameData(); testInitValExtractsSimtime(); @@ -165,6 +166,25 @@ static void testSimtimeAdvancesWithDelta() { check("simtime after iter 2", 2.0, concore.getSimtime()); } + // issue #608: the next read moves simtime forward even when it gets no newer timestamp + static void testWriteAdvancesSimtimeOnNextRead() { + Path tmp = makeTempDir(); + String base = basePath(tmp); + concore.resetState(); + concore.setInPath(base); + concore.setOutPath(base); + try { + Files.createDirectories(portDir(base, 1)); + } catch (IOException e) { + throw new RuntimeException(e); + } + + concore.write(1, "u", Collections.singletonList((Object) 1.0), 1); + check("write leaves simtime alone", 0.0, concore.getSimtime()); + concore.read(1, "missing", "[0.0, 5.0]"); + check("next read moves simtime to 1", 1.0, concore.getSimtime()); + } + static void testUnchangedReturnsFalseAfterRead() { Path tmp = makeTempDir(); String base = basePath(tmp); diff --git a/TestConcoreHpp.cpp b/TestConcoreHpp.cpp index 90d989e..97fb9b4 100644 --- a/TestConcoreHpp.cpp +++ b/TestConcoreHpp.cpp @@ -136,6 +136,35 @@ static void test_write_FM_creates_file() { // ------------- unchanged() ----------------------------------------------- +// issue #608: write() leaves simtime alone, the next read moves it forward +static void test_write_moves_simtime_forward_on_next_read() { + setup_dirs(); + rm("in1/no_port"); + + Concore c; + c.delay = 0; + c.simtime = 10.0; + + c.write_FM(1, "u", std::vector{1.0}, 1); + check("write_FM leaves simtime alone", approx(c.simtime, 10.0)); + c.read_FM(1, "no_port", "[0.0,5.0]"); + check("read after write moves simtime to 11", approx(c.simtime, 11.0)); +} + +static void test_writes_in_one_step_advance_once() { + setup_dirs(); + rm("in1/no_port"); + + Concore c; + c.delay = 0; + c.simtime = 10.0; + + c.write_FM(1, "u", std::vector{1.0}, 1); + c.write_FM(1, "v", std::vector{2.0}, 2); + c.read_FM(1, "no_port", "[0.0,5.0]"); + check("several writes advance simtime once to 12", approx(c.simtime, 12.0)); +} + static void test_unchanged_after_read_is_false() { setup_dirs(); write_file("in1/uc", "[1.0,0.1]"); @@ -312,6 +341,8 @@ int main() { test_read_FM_missing_file_uses_initstr(); test_read_result_missing_file_status(); test_write_FM_creates_file(); + test_write_moves_simtime_forward_on_next_read(); + test_writes_in_one_step_advance_once(); // unchanged() test_unchanged_after_read_is_false(); diff --git a/concore.hpp b/concore.hpp index ba676c0..fbd39bb 100644 --- a/concore.hpp +++ b/concore.hpp @@ -97,6 +97,23 @@ class Concore{ map params; ReadStatus last_read_status = ReadStatus::SUCCESS; + // A write moves simtime forward at the next read (issue #608). write() + // itself leaves simtime alone so writes in one step share a timestamp. + bool has_write = false; + double write_base = 0, write_target = 0; + void note_write(int delta) { + if (!has_write || write_base != simtime) { + has_write = true; + write_base = write_target = simtime; + } + if (simtime + delta > write_target) + write_target = simtime + delta; + } + void apply_write() { + if (has_write && write_base == simtime && write_target > simtime) + simtime = write_target; + } + /** * @brief Constructor for Concore class. * Initializes the iport and oport maps by parsing the respective files. @@ -566,6 +583,7 @@ class Concore{ * @return a string of file content */ vector read_FM(int port, string name, string initstr){ + apply_write(); chrono::milliseconds timespan((int)(1000*delay)); this_thread::sleep_for(timespan); string ins; @@ -647,6 +665,7 @@ class Concore{ * @return string of file content */ vector read_SM(int port, string name, string initstr){ + apply_write(); chrono::milliseconds timespan((int)(1000*delay)); this_thread::sleep_for(timespan); string ins = ""; @@ -773,6 +792,7 @@ class Concore{ outfile< read_ZMQ(string port_name, string name, string initstr) { + apply_write(); ReadStatus status = ReadStatus::SUCCESS; auto it = zmq_ports.find(port_name); if (it == zmq_ports.end()) { @@ -1010,6 +1032,7 @@ class Concore{ val.insert(val.begin(), simtime + delta); it->second->send_with_retry(val); // simtime must not be mutated here (issue #385). + note_write(delta); } /** diff --git a/concore.jl b/concore.jl index 04a2987..af47901 100644 --- a/concore.jl +++ b/concore.jl @@ -76,6 +76,29 @@ global retrycount::Int = 0 """Current simulation time.""" global simtime::Float64 = 0.0 +# A write moves simtime forward at the next read (issue #608). The write itself +# leaves simtime alone so writes in one step share a timestamp. +global write_base::Union{Nothing,Float64} = nothing +global write_target::Float64 = 0.0 + +function _note_write(delta::Real) + global write_base, write_target + if write_base != simtime + write_base = simtime + write_target = simtime + end + write_target = max(write_target, simtime + Float64(delta)) + return nothing +end + +function _apply_write() + global simtime + if write_base == simtime + simtime = max(simtime, write_target) + end + return nothing +end + """Maximum simulation time.""" global maxtime::Float64 = 100.0 @@ -252,6 +275,7 @@ function _zmq_read(port_name::AbstractString, initstr::AbstractString)::Vector{F port === nothing && error("ZMQ port '$port_name' not registered") global simtime + _apply_write() try msg = _zmq_recv(port.socket) vals = safe_parse_list(msg) @@ -273,6 +297,7 @@ function _zmq_write( payload = _zmq_payload(val, delta) _zmq_send(port.socket, payload) + val isa AbstractString || _note_write(delta) return nothing end @@ -644,6 +669,7 @@ function concore_read( initstr::AbstractString, )::Vector{Float64} global s, simtime, retrycount + _apply_write() sleep(delay) @@ -726,6 +752,7 @@ function concore_write( else _write_wire_file(filepath, wire) end + _note_write(delta) return nothing end diff --git a/concore_base.py b/concore_base.py index 4362424..24503ca 100644 --- a/concore_base.py +++ b/concore_base.py @@ -262,6 +262,12 @@ def read(mod, port_identifier, name, initstr_val): """ global last_read_status + # A write since the last read moves simtime forward now. Without this a + # node that only advances through its own writes can miss the end of the + # run when its peer resets the port (issue #608). + if getattr(mod, "_write_base", None) == mod.simtime: + mod.simtime = max(mod.simtime, mod._write_target) + # Default return default_return_val = initstr_val if isinstance(initstr_val, str): @@ -373,6 +379,14 @@ def read(mod, port_identifier, name, initstr_val): return default_return_val, False +def _note_write(mod, delta): + """Remember how far this write moves time, read() applies it (#608). + Writes in the same step share one base, so they advance it only once.""" + if getattr(mod, "_write_base", None) != mod.simtime: + mod._write_base = mod._write_target = mod.simtime + mod._write_target = max(mod._write_target, mod.simtime + delta) + + def write(mod, port_identifier, name, val, delta=0): """ Write data either to ZMQ port or file. @@ -390,6 +404,7 @@ def write(mod, port_identifier, name, val, delta=0): zmq_p.send_json_with_retry(payload) # simtime must not be mutated here. # Mutation breaks cross-language determinism (see issue #385). + _note_write(mod, delta) else: zmq_p.send_json_with_retry(zmq_val) except TimeoutError as e: @@ -427,6 +442,7 @@ def write(mod, port_identifier, name, val, delta=0): outfile.write(str(data_to_write)) # simtime must not be mutated here. # Mutation breaks cross-language determinism (see issue #385). + _note_write(mod, delta) else: outfile.write(val) except Exception as e: diff --git a/concore_read.m b/concore_read.m index 9e03214..3ed0996 100644 --- a/concore_read.m +++ b/concore_read.m @@ -1,5 +1,9 @@ function [result] = concore_read(port, name, inistr) global concore; + % a write since the last read moves simtime forward now (issue #608) + if isfield(concore, 'write_base') && concore.write_base == concore.simtime + concore.simtime = max(concore.simtime, concore.write_target); + end pause(concore.delay); try input1 = fopen(strcat(concore.inpath,num2str(port),'/',name)); diff --git a/concore_write.m b/concore_write.m index f026e8e..8beee1d 100644 --- a/concore_write.m +++ b/concore_write.m @@ -53,6 +53,12 @@ function concore_write(port, name, val, delta) fprintf(output1, '%s', outstr); fclose(output1); % simtime must not be mutated here (issue #385). + % the next read moves simtime forward to this (issue #608) + if ~isfield(concore, 'write_base') || concore.write_base ~= concore.simtime + concore.write_base = concore.simtime; + concore.write_target = concore.simtime; + end + concore.write_target = max(concore.write_target, concore.simtime + delta); catch exc disp(['skipping ' concore.outpath num2str(port) '/' name]); end diff --git a/concoredocker.hpp b/concoredocker.hpp index 108ba5f..33fc741 100644 --- a/concoredocker.hpp +++ b/concoredocker.hpp @@ -76,6 +76,23 @@ class Concore { double maxtime = 100; std::unordered_map params; ReadStatus last_read_status = ReadStatus::SUCCESS; + + // A write moves simtime forward at the next read (issue #608). write() + // itself leaves simtime alone so writes in one step share a timestamp. + bool has_write = false; + double write_base = 0, write_target = 0; + void note_write(int delta) { + if (!has_write || write_base != simtime) { + has_write = true; + write_base = write_target = simtime; + } + if (simtime + delta > write_target) + write_target = simtime + delta; + } + void apply_write() { + if (has_write && write_base == simtime && write_target > simtime) + simtime = write_target; + } #ifdef CONCORE_USE_ZMQ std::map zmq_ports; #endif @@ -423,6 +440,7 @@ class Concore { } std::vector read(int port, const std::string& name, const std::string& initstr) { + apply_write(); #ifdef __linux__ if (communication_iport == 1) return read_SM(port, name, initstr); @@ -500,6 +518,7 @@ class Concore { } std::vector read_SM(int port, const std::string& name, const std::string& initstr) { + apply_write(); ReadStatus status = ReadStatus::SUCCESS; std::this_thread::sleep_for(std::chrono::seconds(delay)); std::string ins; @@ -573,6 +592,7 @@ class Concore { } outfile << "]"; // simtime must not be mutated here (issue #385). + note_write(delta); } } @@ -608,6 +628,7 @@ class Concore { (void)__atomic_fetch_add(seqp, uint64_t{1}, __ATOMIC_ACQ_REL); // even = ready } shm_sem_release(semId_create); + note_write(delta); } catch (...) { std::cerr << "skipping +" << outpath << port << "/" << name << "\n"; } @@ -627,6 +648,7 @@ class Concore { } std::vector read_ZMQ(const std::string& port_name, const std::string& name, const std::string& initstr) { + apply_write(); ReadStatus status = ReadStatus::SUCCESS; auto it = zmq_ports.find(port_name); if (it == zmq_ports.end()) { @@ -662,6 +684,7 @@ class Concore { val.insert(val.begin(), simtime + delta); it->second->send_with_retry(val); // simtime must not be mutated here. + note_write(delta); } std::vector read(const std::string& port_name, const std::string& name, const std::string& initstr) { diff --git a/concoredocker.jl b/concoredocker.jl index c27672c..e2c03c1 100644 --- a/concoredocker.jl +++ b/concoredocker.jl @@ -77,6 +77,29 @@ global retrycount::Int = 0 """Current simulation time.""" global simtime::Float64 = 0.0 +# A write moves simtime forward at the next read (issue #608). The write itself +# leaves simtime alone so writes in one step share a timestamp. +global write_base::Union{Nothing,Float64} = nothing +global write_target::Float64 = 0.0 + +function _note_write(delta::Real) + global write_base, write_target + if write_base != simtime + write_base = simtime + write_target = simtime + end + write_target = max(write_target, simtime + Float64(delta)) + return nothing +end + +function _apply_write() + global simtime + if write_base == simtime + simtime = max(simtime, write_target) + end + return nothing +end + """Maximum simulation time.""" global maxtime::Float64 = 100.0 @@ -253,6 +276,7 @@ function _zmq_read(port_name::AbstractString, initstr::AbstractString)::Vector{F port === nothing && error("ZMQ port '$port_name' not registered") global simtime + _apply_write() try msg = _zmq_recv(port.socket) vals = safe_parse_list(msg) @@ -274,6 +298,7 @@ function _zmq_write( payload = _zmq_payload(val, delta) _zmq_send(port.socket, payload) + val isa AbstractString || _note_write(delta) return nothing end @@ -645,6 +670,7 @@ function concore_read( initstr::AbstractString, )::Vector{Float64} global s, simtime, retrycount + _apply_write() sleep(delay) @@ -727,6 +753,7 @@ function concore_write( else _write_wire_file(filepath, wire) end + _note_write(delta) return nothing end diff --git a/tests/julia/test_protocol.jl b/tests/julia/test_protocol.jl index 2cb915b..cea6a23 100644 --- a/tests/julia/test_protocol.jl +++ b/tests/julia/test_protocol.jl @@ -505,4 +505,28 @@ end + # issue #608: a write moves simtime forward at the next read + @testset "write advances simtime on next read" begin + mktempdir() do dir + reset_concore_state!() + old_in, old_out = Concore.inpath, Concore.outpath + Concore.inpath = joinpath(dir, "in") + Concore.outpath = joinpath(dir, "out") + + Concore.simtime = 10.0 + Concore.concore_write(1, "u", [1.0]; delta=1) + @test Concore.simtime == 10.0 + Concore.concore_read(1, "missing", "[0.0, 5.0]") + @test Concore.simtime == 11.0 + + Concore.simtime = 20.0 + Concore.concore_write(1, "u", [1.0]; delta=1) + Concore.concore_write(1, "v", [2.0]; delta=2) + Concore.concore_read(1, "missing", "[0.0, 5.0]") + @test Concore.simtime == 22.0 + + Concore.inpath, Concore.outpath = old_in, old_out + end + end + end diff --git a/tests/matlab/test_write_advance.m b/tests/matlab/test_write_advance.m new file mode 100644 index 0000000..f612545 --- /dev/null +++ b/tests/matlab/test_write_advance.m @@ -0,0 +1,38 @@ +function test_write_advance + % issue #608: concore_write leaves simtime alone, the next read moves it + % forward so a node whose own writes are its clock still reaches maxtime + global concore; + import_concore; + + dir = tempname; + mkdir(dir); + mkdir(fullfile(dir, 'in1')); + mkdir(fullfile(dir, 'out1')); + concore.inpath = [dir '/in']; + concore.outpath = [dir '/out']; + concore.delay = 0; + + % 1. one write, then a read that gets no newer timestamp + concore.simtime = 10; + concore_write(1, 'u', [1.0], 1); + assert(concore.simtime == 10); + concore_read(1, 'missing', '[0.0, 5.0]'); + assert(concore.simtime == 11); + + % 2. several writes in one step advance it once, to the largest delta + concore.simtime = 20; + concore_write(1, 'u', [1.0], 1); + concore_write(1, 'v', [2.0], 2); + concore_read(1, 'missing', '[0.0, 5.0]'); + assert(concore.simtime == 22); + + % 3. simtime set by hand after the write wins + concore.simtime = 30; + concore_write(1, 'u', [1.0], 1); + concore.simtime = 3; + concore_read(1, 'missing', '[0.0, 5.0]'); + assert(concore.simtime == 3); + + rmdir(dir, 's'); + disp('MATLAB write advance tests passed'); +end diff --git a/tests/test_concore.py b/tests/test_concore.py index c3096a6..c9f6d2d 100644 --- a/tests/test_concore.py +++ b/tests/test_concore.py @@ -488,6 +488,69 @@ def test_write_timestamp_matches_cpp_semantics(self, temp_dir): ) +class TestWriteAdvancesSimtimeOnNextRead: + """Issue #608: write() leaves simtime alone (#385), but the next read() + moves it up to simtime + delta. Otherwise a node that only advances + through its own writes can miss the end of the run.""" + + @pytest.fixture(autouse=True) + def restore_state(self, temp_dir): + import concore + + saved = (concore.simtime, concore.delay, concore.inpath, concore.outpath) + concore.delay = 0 + concore.inpath = os.path.join(temp_dir, "in") + concore.outpath = os.path.join(temp_dir, "out") + os.makedirs(os.path.join(temp_dir, "in1")) + os.makedirs(os.path.join(temp_dir, "out1")) + yield + concore.simtime, concore.delay, concore.inpath, concore.outpath = saved + + def test_read_after_write_moves_simtime_forward(self): + import concore + + concore.simtime = 10 + concore.write(1, "u", [1.0], delta=1) + assert concore.simtime == 10 + concore.read(1, "missing", "[0.0, 5.0]") + assert concore.simtime == 11 + + def test_several_writes_in_one_step_advance_once(self): + import concore + + concore.simtime = 10 + concore.write(1, "u", [1.0], delta=1) + concore.write(1, "v", [2.0], delta=2) + concore.write(1, "w", [3.0], delta=1) + concore.read(1, "missing", "[0.0, 5.0]") + assert concore.simtime == 12 + + def test_zmq_write_moves_simtime_forward_on_next_read(self): + import concore + + class DummyPort: + def send_json_with_retry(self, msg): + self.sent = msg + + concore.zmq_ports["advance_test"] = DummyPort() + try: + concore.simtime = 10 + concore.write("advance_test", "u", [1.0], delta=2) + concore.read(1, "missing", "[0.0, 5.0]") + finally: + del concore.zmq_ports["advance_test"] + assert concore.simtime == 12 + + def test_simtime_set_by_hand_after_write_is_kept(self): + import concore + + concore.simtime = 10 + concore.write(1, "u", [1.0], delta=1) + concore.simtime = 3 + concore.read(1, "missing", "[0.0, 5.0]") + assert concore.simtime == 3 + + class TestZMQRetryExhaustion: """Issue #393 – recv_json_with_retry / send_json_with_retry must raise TimeoutError instead of silently returning None when retries are diff --git a/tests/test_end_of_run.py b/tests/test_end_of_run.py new file mode 100644 index 0000000..fa6b317 --- /dev/null +++ b/tests/test_end_of_run.py @@ -0,0 +1,75 @@ +import os +import subprocess +import sys +import time +from pathlib import Path + +REPO = Path(__file__).resolve().parents[1] + +# Same shape as testsou/cpymat.py and testsou/pmpymat.py: the controller writes +# its init values back to u right after its loop. The plant is slower than the +# controller, so it can miss the last real u (see #608). +CONTROLLER = """ +import concore +concore.inpath, concore.outpath = {ym!r}, {u!r} +concore.delay = 0.01 +init = "[0.0,0.0,0.0]" +u = concore.initval(init) +while concore.simtime < 20: + while concore.unchanged(): + ym = concore.read(1, "ym", init) + u[0] = ym[0] + 1 + concore.write(1, "u", u) +concore.write(1, "u", init) +print("done") +""" + +PLANT = """ +import concore +concore.inpath, concore.outpath = {u!r}, {ym!r} +concore.delay = 0.05 +init = "[0.0,0.0,0.0]" +ym = concore.initval(init) +while concore.simtime < 20: + while concore.unchanged(): + u = concore.read(1, "u", init) + ym[0] = u[0] + 10000 + concore.write(1, "ym", ym, delta=1) +concore.write(1, "ym", init) +print("done") +""" + + +def test_slow_plant_still_finishes(tmp_path): + ports = {"u": str(tmp_path / "u"), "ym": str(tmp_path / "ym")} + for base in ports.values(): + os.makedirs(base + "1") + env = dict(os.environ, PYTHONPATH=str(REPO)) + + procs = {} + for name, source in (("controller", CONTROLLER), ("plant", PLANT)): + node = tmp_path / name + node.mkdir() + (node / "node.py").write_text(source.format(**ports)) + procs[name] = subprocess.Popen( + [sys.executable, "node.py"], + cwd=node, + env=env, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + ) + + deadline = time.time() + 60 + while any(p.poll() is None for p in procs.values()) and time.time() < deadline: + time.sleep(0.2) + hung = [name for name, p in procs.items() if p.poll() is None] + for p in procs.values(): + if p.poll() is None: + p.kill() + for p in procs.values(): + p.wait() + + assert not hung, f"{', '.join(hung)} never reached the end of its loop" + for name, p in procs.items(): + assert "done" in p.stdout.read(), name diff --git a/tests/test_protocol_conformance.py b/tests/test_protocol_conformance.py index fcb6fcd..a6f2737 100644 --- a/tests/test_protocol_conformance.py +++ b/tests/test_protocol_conformance.py @@ -102,6 +102,8 @@ def _run_read_file_case(case): try: with tempfile.TemporaryDirectory() as temp_dir: concore.simtime = case["input"]["initial_simtime"] + # fresh node, nothing written before this read (#608) + concore._write_base = None concore.inpath = os.path.join(temp_dir, "in") concore.delay = 0