fix: move simtime forward on the read after a write

.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

ConcoreJavaRuntimeCore.java

@@ -124,7 +124,26 @@ private static Map<String, Object> 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<Object> 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<Object> 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) {

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);

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<double>{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<double>{1.0}, 1);
+    c.write_FM(1, "v", std::vector<double>{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();

concore.hpp

@@ -97,6 +97,23 @@ class Concore{
     map <string, string> 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<double> 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<double> 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<<val[val.size()-1]<<']';
                 outfile.close();
                 // simtime must not be mutated here (issue #385).
+                note_write(delta);
                 }
             else{
                 throw 505;
@@ -875,6 +895,7 @@ class Concore{
             shm_sem_release(semId_create);
 #endif
             // simtime must not be mutated here (issue #385).
+            note_write(delta);
         }
 
         catch (const std::exception &e) {
@@ -968,6 +989,7 @@ class Concore{
      * @return a vector of double values
      */
     vector<double> 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);
     }
 
     /**

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

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:

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));

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