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