diff --git a/.DS_Store b/.DS_Store new file mode 100644 index 00000000..833e2ea3 Binary files /dev/null and b/.DS_Store differ diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 00000000..3ae8281a --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,110 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Build Commands + +```bash +# Full installation (installs deps, creates venv, compiles runtime) +sudo ./install.sh + +# Manual build (C/C++ runtime core) +mkdir -p build && cd build && cmake .. && make -j$(nproc) + +# Start the runtime (requires root for real-time scheduling) +sudo ./start_openplc.sh + +# Run web server only (for development) +source venvs/runtime/bin/activate +sudo python3 -m webserver.app + +# Run PLC runtime only +sudo ./build/plc_main --print-logs +``` + +## Testing + +```bash +# Setup test environment and run tests +sudo bash scripts/setup-tests-env.sh +pytest tests/ + +# Or use the test script +bash scripts/run-pytest.sh +``` + +## Linting and Formatting + +Pre-commit hooks handle formatting. Install with: +```bash +pip install pre-commit +pre-commit install +pre-commit run --all-files +``` + +- **C/C++**: Clang-Format (LLVM style, 4-space indent, 100 char limit) +- **Python**: Black + isort + Ruff (100 char line length) + +## Architecture Overview + +OpenPLC Runtime v4 is a **dual-process industrial PLC runtime**: + +### Process 1: REST API Server (Python/Flask) +- **Location**: `webserver/` +- **Port**: 8443 (HTTPS with self-signed TLS) +- **Purpose**: REST API for OpenPLC Editor, WebSocket debug interface, compilation orchestration +- **Entry point**: `webserver/app.py` + +### Process 2: PLC Runtime Core (C/C++) +- **Location**: `core/src/plc_app/` +- **Executable**: `build/plc_main` +- **Purpose**: Real-time PLC execution with SCHED_FIFO priority +- **Entry point**: `core/src/plc_app/plc_main.c` + +### Inter-Process Communication +- **Command socket**: `/run/runtime/plc_runtime.socket` (text protocol for start/stop/status) +- **Log socket**: `/run/runtime/log_runtime.socket` (real-time log streaming) +- **Client**: `webserver/unixclient.py` +- **Server**: `core/src/plc_app/unix_socket.c` + +### PLC State Machine +``` +EMPTY -> INIT -> RUNNING <-> STOPPED -> ERROR +``` +State management: `core/src/plc_app/plc_state_manager.c` + +### Plugin System +- **Config**: `plugins.conf` +- **Types**: Python (type=0) and Native C/C++ (type=1) +- **Driver code**: `core/src/drivers/` +- **Plugin examples**: `core/src/drivers/plugins/python/` and `core/src/drivers/plugins/native/` + +### Key Subsystems +- **Scan cycle manager**: `core/src/plc_app/scan_cycle_manager.c` - deterministic timing +- **Debug handler**: `core/src/plc_app/debug_handler.c` - WebSocket debug protocol +- **Watchdog**: `core/src/plc_app/utils/watchdog.c` - health monitoring +- **Image tables**: `core/src/plc_app/image_tables.c` - I/O buffer management + +## Code Style + +- **C/C++**: 4-space indent, no tabs, `snake_case` functions, `snake_case_t` types, `UPPER_CASE` macros +- **Python**: PEP 8, type hints, 100 char lines +- **No emojis** anywhere in code, comments, or documentation (project standard) + +## Key Directories + +- `webserver/` - Flask REST API and WebSocket debug interface +- `core/src/plc_app/` - C/C++ real-time PLC runtime +- `core/src/drivers/` - Plugin driver system +- `core/generated/` - Generated PLC code from uploaded programs +- `scripts/` - Build, compile, and management scripts +- `build/` - CMake output (`plc_main` executable, `libplc_*.so` libraries) +- `venvs/` - Python virtual environments (runtime + per-plugin) +- `docs/` - Detailed documentation + +## Compilation Flow + +1. OpenPLC Editor uploads `program.zip` to `/api/upload-file` +2. Runtime validates, extracts to `core/generated/` +3. `scripts/compile.sh` compiles to `build/libplc_*.so` +4. Runtime loads shared library dynamically via `plcapp_manager.c` diff --git a/Dockerfile b/Dockerfile index 5528a3ff..5b93974c 100644 --- a/Dockerfile +++ b/Dockerfile @@ -17,6 +17,9 @@ RUN rm -rf build/ venvs/ .venv/ 2>/dev/null || true # Run installation script RUN ./install.sh +# Clean up apt cache to reduce image size (Docker-specific optimization) +RUN rm -rf /var/lib/apt/lists/* + # Expose webserver port EXPOSE 8443 diff --git a/Dockerfile.dev b/Dockerfile.dev index 3ef85bb7..a5620bbc 100644 --- a/Dockerfile.dev +++ b/Dockerfile.dev @@ -14,6 +14,9 @@ RUN rm -rf build/ venvs/ 2>/dev/null || true RUN chmod +x install.sh scripts/* start_openplc.sh RUN ./install.sh +# Clean up apt cache to reduce image size (Docker-specific optimization) +RUN rm -rf /var/lib/apt/lists/* + EXPOSE 8443 CMD ["bash", "./scripts/run-pytest.sh"] diff --git a/core/.DS_Store b/core/.DS_Store new file mode 100644 index 00000000..639efdfd Binary files /dev/null and b/core/.DS_Store differ diff --git a/core/src/.DS_Store b/core/src/.DS_Store new file mode 100644 index 00000000..51e56e1e Binary files /dev/null and b/core/src/.DS_Store differ diff --git a/core/src/CMakeLists.txt b/core/src/CMakeLists.txt index ba00b5ba..7cd2d893 100644 --- a/core/src/CMakeLists.txt +++ b/core/src/CMakeLists.txt @@ -34,6 +34,7 @@ add_executable(plc_main ${CMAKE_SOURCE_DIR}/core/src/plc_app/utils/utils.c ${CMAKE_SOURCE_DIR}/core/src/plc_app/utils/watchdog.c ${CMAKE_SOURCE_DIR}/core/src/plc_app/image_tables.c + ${CMAKE_SOURCE_DIR}/core/src/plc_app/journal_buffer.c ${CMAKE_SOURCE_DIR}/core/src/plc_app/plc_state_manager.c ${CMAKE_SOURCE_DIR}/core/src/plc_app/plcapp_manager.c ${CMAKE_SOURCE_DIR}/core/src/plc_app/scan_cycle_manager.c diff --git a/core/src/drivers/.DS_Store b/core/src/drivers/.DS_Store new file mode 100644 index 00000000..fbeefcd4 Binary files /dev/null and b/core/src/drivers/.DS_Store differ diff --git a/core/src/drivers/plugin_driver.c b/core/src/drivers/plugin_driver.c index eb7ce1d3..5bc20ae3 100644 --- a/core/src/drivers/plugin_driver.c +++ b/core/src/drivers/plugin_driver.c @@ -14,6 +14,7 @@ #endif #include "../plc_app/image_tables.h" +#include "../plc_app/journal_buffer.h" #include "../plc_app/utils/log.h" #include "plugin_config.h" #include "plugin_driver.h" @@ -105,6 +106,36 @@ int plugin_mutex_give(pthread_mutex_t *mutex) return pthread_mutex_unlock(mutex); } +// Journal write wrapper functions for plugins +// These match the function pointer signatures in plugin_types.h and delegate +// to the journal_buffer.h API with proper type casting. + +static int plugin_journal_write_bool(int type, int index, int bit, int value) +{ + return journal_write_bool((journal_buffer_type_t)type, (uint16_t)index, (uint8_t)bit, + value != 0); +} + +static int plugin_journal_write_byte(int type, int index, int value) +{ + return journal_write_byte((journal_buffer_type_t)type, (uint16_t)index, (uint8_t)value); +} + +static int plugin_journal_write_int(int type, int index, int value) +{ + return journal_write_int((journal_buffer_type_t)type, (uint16_t)index, (uint16_t)value); +} + +static int plugin_journal_write_dint(int type, int index, unsigned int value) +{ + return journal_write_dint((journal_buffer_type_t)type, (uint16_t)index, (uint32_t)value); +} + +static int plugin_journal_write_lint(int type, int index, unsigned long long value) +{ + return journal_write_lint((journal_buffer_type_t)type, (uint16_t)index, (uint64_t)value); +} + // Python capsule destructor for runtime args // Breakpoint here to debug capsule issues static void plugin_runtime_args_capsule_destructor(PyObject *capsule) @@ -670,6 +701,13 @@ void *generate_structured_args_with_driver(plugin_type_t type, plugin_driver_t * args->log_warn = log_warn; args->log_error = log_error; + // Initialize journal write functions for race-condition-free buffer writes + args->journal_write_bool = plugin_journal_write_bool; + args->journal_write_byte = plugin_journal_write_byte; + args->journal_write_int = plugin_journal_write_int; + args->journal_write_dint = plugin_journal_write_dint; + args->journal_write_lint = plugin_journal_write_lint; + // printf("[PLUGIN]: Runtime args initialized:\n"); // printf("[PLUGIN]: buffer_size = %d\n", args->buffer_size); // printf("[PLUGIN]: bits_per_buffer = %d\n", args->bits_per_buffer); diff --git a/core/src/drivers/plugin_types.h b/core/src/drivers/plugin_types.h index 0c8d9aca..ea5933ac 100644 --- a/core/src/drivers/plugin_types.h +++ b/core/src/drivers/plugin_types.h @@ -30,6 +30,26 @@ typedef void (*plugin_log_debug_func_t)(const char *fmt, ...); typedef void (*plugin_log_warn_func_t)(const char *fmt, ...); typedef void (*plugin_log_error_func_t)(const char *fmt, ...); +/** + * @brief Journal write function pointer types + * + * These function pointers allow plugins to write to I/O buffers through + * the journal buffer system, ensuring race-condition-free writes. + * All writes are applied atomically at the start of the next PLC scan cycle. + * + * Buffer type values (matching journal_buffer_type_t): + * 0=BOOL_INPUT, 1=BOOL_OUTPUT, 2=BOOL_MEMORY + * 3=BYTE_INPUT, 4=BYTE_OUTPUT + * 5=INT_INPUT, 6=INT_OUTPUT, 7=INT_MEMORY + * 8=DINT_INPUT, 9=DINT_OUTPUT, 10=DINT_MEMORY + * 11=LINT_INPUT, 12=LINT_OUTPUT, 13=LINT_MEMORY + */ +typedef int (*plugin_journal_write_bool_func_t)(int type, int index, int bit, int value); +typedef int (*plugin_journal_write_byte_func_t)(int type, int index, int value); +typedef int (*plugin_journal_write_int_func_t)(int type, int index, int value); +typedef int (*plugin_journal_write_dint_func_t)(int type, int index, unsigned int value); +typedef int (*plugin_journal_write_lint_func_t)(int type, int index, unsigned long long value); + /** * @brief Runtime buffer access structure for plugins * @@ -79,6 +99,13 @@ typedef struct plugin_log_debug_func_t log_debug; plugin_log_warn_func_t log_warn; plugin_log_error_func_t log_error; + + /* Journal write functions - race-condition-free buffer writes */ + plugin_journal_write_bool_func_t journal_write_bool; + plugin_journal_write_byte_func_t journal_write_byte; + plugin_journal_write_int_func_t journal_write_int; + plugin_journal_write_dint_func_t journal_write_dint; + plugin_journal_write_lint_func_t journal_write_lint; } plugin_runtime_args_t; #endif /* PLUGIN_TYPES_H */ diff --git a/core/src/drivers/plugins/.DS_Store b/core/src/drivers/plugins/.DS_Store new file mode 100644 index 00000000..65b83b40 Binary files /dev/null and b/core/src/drivers/plugins/.DS_Store differ diff --git a/core/src/drivers/plugins/native/.DS_Store b/core/src/drivers/plugins/native/.DS_Store new file mode 100644 index 00000000..0a78a9b7 Binary files /dev/null and b/core/src/drivers/plugins/native/.DS_Store differ diff --git a/core/src/drivers/plugins/native/s7comm/.DS_Store b/core/src/drivers/plugins/native/s7comm/.DS_Store new file mode 100644 index 00000000..923d7d47 Binary files /dev/null and b/core/src/drivers/plugins/native/s7comm/.DS_Store differ diff --git a/core/src/drivers/plugins/native/s7comm/s7comm_plugin.cpp b/core/src/drivers/plugins/native/s7comm/s7comm_plugin.cpp index ff5554c0..1ac1cec2 100644 --- a/core/src/drivers/plugins/native/s7comm/s7comm_plugin.cpp +++ b/core/src/drivers/plugins/native/s7comm/s7comm_plugin.cpp @@ -5,21 +5,16 @@ * This plugin implements a Siemens S7 communication server using the Snap7 library. * It allows S7-compatible HMIs and SCADA systems to read/write OpenPLC I/O buffers. * - * Buffer Sync Strategy: - * - S7 buffers: What Snap7 clients read/write (accessed asynchronously) - * - Shadow buffers: Used for sync with OpenPLC (minimizes mutex hold time) - * - S7 mutex: Protects S7 buffers during brief memcpy operations + * Buffer Sync Strategy (using journal buffer system): + * - S7 buffers: What Snap7 clients read/write (managed by Snap7 server) + * - Journal writes: Race-condition-free writes to OpenPLC buffers * - * Data flow (full bidirectional sync for ALL types): - * - cycle_start: Copy ENTIRE S7 buffer -> OpenPLC buffer - * S7 client writes become visible to the PLC program + * Data flow (on-demand via Snap7 RWArea callback): + * - S7 client READ: Callback acquires OpenPLC mutex, copies fresh data to S7 buffer + * - S7 client WRITE: Callback uses journal writes (thread-safe, no mutex needed) * - * - cycle_end: Copy ENTIRE OpenPLC buffer -> S7 buffer - * PLC program outputs become visible to S7 clients - * - * This approach allows S7 clients to write to any location. Values that are - * actively driven by the PLC program or I/O drivers will be overwritten - * during the scan cycle, but S7 writes to other locations will persist. + * This approach allows the S7 server thread to run independently without + * requiring cycle_start/cycle_end hooks for data synchronization. */ #include @@ -30,6 +25,7 @@ /* Snap7 includes */ #include "snap7_libmain.h" #include "s7_types.h" +#include "s7_server.h" /* For OperationRead, OperationWrite, TS7Tag */ /* Plugin includes */ extern "C" { @@ -48,7 +44,7 @@ extern "C" { /* * ============================================================================= - * Data Block Runtime Structure (with double-buffering) + * Data Block Runtime Structure * ============================================================================= */ typedef struct { @@ -58,12 +54,11 @@ typedef struct { int size_bytes; /* Size in bytes */ bool bit_addressing; /* Bit-level access enabled */ uint8_t *s7_buffer; /* S7 buffer (registered with Snap7) */ - uint8_t *shadow_buffer; /* Shadow buffer for sync with OpenPLC */ } s7comm_db_runtime_t; /* * ============================================================================= - * System Area Runtime Structure (with double-buffering) + * System Area Runtime Structure * ============================================================================= */ typedef struct { @@ -72,7 +67,6 @@ typedef struct { s7comm_buffer_type_t type; int start_buffer; uint8_t *s7_buffer; /* S7 buffer (registered with Snap7) */ - uint8_t *shadow_buffer; /* Shadow buffer for sync with OpenPLC */ } s7comm_area_runtime_t; /* @@ -90,14 +84,13 @@ static bool g_config_loaded = false; /* Snap7 server handle (S7Object is uintptr_t, use 0 for null) */ static S7Object g_server = 0; -/* S7 buffer mutex - protects S7 buffers during sync */ -static pthread_mutex_t g_s7_mutex = PTHREAD_MUTEX_INITIALIZER; +/* No S7 buffer mutex needed - reads use OpenPLC mutex, writes use journal */ /* Runtime data blocks (dynamically allocated based on config) */ static s7comm_db_runtime_t g_db_runtime[S7COMM_MAX_DATA_BLOCKS]; static int g_num_db_runtime = 0; -/* System area runtime (with double-buffering) */ +/* System area runtime */ static s7comm_area_runtime_t g_pe_runtime; static s7comm_area_runtime_t g_pa_runtime; static s7comm_area_runtime_t g_mk_runtime; @@ -108,11 +101,15 @@ static s7comm_area_runtime_t g_mk_runtime; * ============================================================================= */ static void s7comm_event_callback(void *usrPtr, PSrvEvent PEvent, int Size); +static int s7comm_rw_area_callback(void *usrPtr, int Sender, int Operation, PS7Tag PTag, void *pUsrData); static int allocate_buffers(void); static void free_buffers(void); static int register_all_areas(void); -static void sync_shadow_to_openplc_by_type(uint8_t *shadow, int size, s7comm_buffer_type_t type, int start_buffer); -static void sync_openplc_to_shadow_by_type(uint8_t *shadow, int size, s7comm_buffer_type_t type, int start_buffer); +static void read_openplc_to_buffer(uint8_t *dest, int size, s7comm_buffer_type_t type, int start_buffer); +static void write_buffer_to_openplc_journal(uint8_t *src, int size, s7comm_buffer_type_t type, int start_buffer); +static s7comm_db_runtime_t* find_db_runtime(int db_number); +static s7comm_area_runtime_t* find_area_runtime(int area); +static int get_type_size(s7comm_buffer_type_t type); /* * ============================================================================= @@ -152,7 +149,7 @@ static inline uint64_t swap64(uint64_t val) */ /** - * @brief Allocate a system area with double-buffering + * @brief Allocate a system area buffer */ static int allocate_area(s7comm_area_runtime_t *area, const s7comm_system_area_t *config) { @@ -174,19 +171,11 @@ static int allocate_area(s7comm_area_runtime_t *area, const s7comm_system_area_t return -1; } - /* Allocate shadow buffer (for sync with OpenPLC) */ - area->shadow_buffer = (uint8_t *)calloc(1, config->size_bytes); - if (area->shadow_buffer == NULL) { - free(area->s7_buffer); - area->s7_buffer = NULL; - return -1; - } - return 0; } /** - * @brief Free a system area's buffers + * @brief Free a system area's buffer */ static void free_area(s7comm_area_runtime_t *area) { @@ -194,37 +183,33 @@ static void free_area(s7comm_area_runtime_t *area) free(area->s7_buffer); area->s7_buffer = NULL; } - if (area->shadow_buffer != NULL) { - free(area->shadow_buffer); - area->shadow_buffer = NULL; - } area->enabled = false; } /** - * @brief Allocate all buffers (S7 + shadow) based on configuration + * @brief Allocate all S7 buffers based on configuration */ static int allocate_buffers(void) { g_num_db_runtime = 0; - /* Allocate system areas with double-buffering */ + /* Allocate system areas */ if (allocate_area(&g_pe_runtime, &g_config.pe_area) != 0) { - plugin_logger_error(&g_logger, "Failed to allocate PE area buffers"); + plugin_logger_error(&g_logger, "Failed to allocate PE area buffer"); return -1; } if (allocate_area(&g_pa_runtime, &g_config.pa_area) != 0) { - plugin_logger_error(&g_logger, "Failed to allocate PA area buffers"); + plugin_logger_error(&g_logger, "Failed to allocate PA area buffer"); return -1; } if (allocate_area(&g_mk_runtime, &g_config.mk_area) != 0) { - plugin_logger_error(&g_logger, "Failed to allocate MK area buffers"); + plugin_logger_error(&g_logger, "Failed to allocate MK area buffer"); return -1; } - /* Allocate data blocks with double-buffering */ + /* Allocate data blocks */ for (int i = 0; i < g_config.num_data_blocks; i++) { const s7comm_data_block_t *db_cfg = &g_config.data_blocks[i]; @@ -248,17 +233,8 @@ static int allocate_buffers(void) return -1; } - /* Allocate shadow buffer */ - db_rt->shadow_buffer = (uint8_t *)calloc(1, db_cfg->size_bytes); - if (db_rt->shadow_buffer == NULL) { - plugin_logger_error(&g_logger, "Failed to allocate DB%d shadow buffer", db_cfg->db_number); - free(db_rt->s7_buffer); - db_rt->s7_buffer = NULL; - return -1; - } - g_num_db_runtime++; - plugin_logger_debug(&g_logger, "Allocated DB%d: %d bytes (double-buffered), type=%s", + plugin_logger_debug(&g_logger, "Allocated DB%d: %d bytes, type=%s", db_cfg->db_number, db_cfg->size_bytes, s7comm_buffer_type_name(db_cfg->mapping.type)); } @@ -282,10 +258,6 @@ static void free_buffers(void) free(g_db_runtime[i].s7_buffer); g_db_runtime[i].s7_buffer = NULL; } - if (g_db_runtime[i].shadow_buffer != NULL) { - free(g_db_runtime[i].shadow_buffer); - g_db_runtime[i].shadow_buffer = NULL; - } } g_num_db_runtime = 0; } @@ -297,7 +269,7 @@ static int register_all_areas(void) { int result; - /* Register system areas (using S7 buffers, not shadow) */ + /* Register system areas (using S7 buffers) */ if (g_pe_runtime.enabled && g_pe_runtime.s7_buffer != NULL) { result = Srv_RegisterArea(g_server, srvAreaPE, 0, g_pe_runtime.s7_buffer, g_pe_runtime.size_bytes); if (result != 0) { @@ -352,7 +324,7 @@ extern "C" int init(void *args) { /* Initialize logger first (before we have runtime_args) */ plugin_logger_init(&g_logger, "S7COMM", NULL); - plugin_logger_info(&g_logger, "Initializing S7Comm plugin (double-buffered)..."); + plugin_logger_info(&g_logger, "Initializing S7Comm plugin (journal-buffered)..."); if (!args) { plugin_logger_error(&g_logger, "init args is NULL"); @@ -367,9 +339,6 @@ extern "C" int init(void *args) plugin_logger_info(&g_logger, "Buffer size: %d", g_runtime_args.buffer_size); - /* Initialize S7 buffer mutex */ - pthread_mutex_init(&g_s7_mutex, NULL); - /* Parse configuration file */ const char *config_path = g_runtime_args.plugin_specific_config_file_path; if (config_path == NULL || config_path[0] == '\0') { @@ -401,7 +370,7 @@ extern "C" int init(void *args) plugin_logger_info(&g_logger, "PLC identity: %s (%s)", g_config.identity.name, g_config.identity.module_type); plugin_logger_info(&g_logger, "Data blocks configured: %d", g_config.num_data_blocks); - /* Allocate all buffers (S7 + shadow for double-buffering) */ + /* Allocate S7 buffers */ if (allocate_buffers() != 0) { plugin_logger_error(&g_logger, "Failed to allocate buffers"); free_buffers(); @@ -450,11 +419,14 @@ extern "C" int init(void *args) /* Set event callback for logging */ Srv_SetEventsCallback(g_server, s7comm_event_callback, NULL); + /* Set RWArea callback for on-demand data synchronization */ + Srv_SetRWAreaCallback(g_server, s7comm_rw_area_callback, NULL); + /* Register all S7 areas with the server */ register_all_areas(); g_initialized = true; - plugin_logger_info(&g_logger, "S7Comm plugin initialized successfully (double-buffered mode)"); + plugin_logger_info(&g_logger, "S7Comm plugin initialized successfully (journal-buffered mode)"); /* Log registered areas summary */ if (g_pe_runtime.enabled) { @@ -564,7 +536,6 @@ extern "C" void cleanup(void) } free_buffers(); - pthread_mutex_destroy(&g_s7_mutex); g_initialized = false; g_config_loaded = false; @@ -574,131 +545,23 @@ extern "C" void cleanup(void) /** * @brief Called at the start of each PLC scan cycle * - * Copy ENTIRE S7 buffer -> OpenPLC buffer for ALL types. - * This allows S7 clients to write to any location. Values that are actively - * driven by the PLC program or I/O drivers will be overwritten during the - * scan cycle, but S7 writes to other locations will persist. + * With the journal-based approach, data synchronization happens on-demand + * via the RWArea callback. No action needed here. */ extern "C" void cycle_start(void) { - if (!g_initialized || !g_running || !g_config.enabled) { - return; - } - - /* - * Lock S7 mutex and copy S7 -> shadow for ALL areas - * This captures everything S7 clients have written - */ - pthread_mutex_lock(&g_s7_mutex); - - /* Copy all system areas S7 -> shadow */ - if (g_pe_runtime.enabled) { - memcpy(g_pe_runtime.shadow_buffer, g_pe_runtime.s7_buffer, g_pe_runtime.size_bytes); - } - if (g_pa_runtime.enabled) { - memcpy(g_pa_runtime.shadow_buffer, g_pa_runtime.s7_buffer, g_pa_runtime.size_bytes); - } - if (g_mk_runtime.enabled) { - memcpy(g_mk_runtime.shadow_buffer, g_mk_runtime.s7_buffer, g_mk_runtime.size_bytes); - } - - /* Copy all data blocks S7 -> shadow */ - for (int i = 0; i < g_num_db_runtime; i++) { - s7comm_db_runtime_t *db = &g_db_runtime[i]; - memcpy(db->shadow_buffer, db->s7_buffer, db->size_bytes); - } - - pthread_mutex_unlock(&g_s7_mutex); - - /* - * Sync shadow -> OpenPLC for ALL areas - * S7 client writes become visible to the PLC program - */ - - /* Sync all system areas shadow -> OpenPLC */ - if (g_pe_runtime.enabled) { - sync_shadow_to_openplc_by_type(g_pe_runtime.shadow_buffer, g_pe_runtime.size_bytes, - g_pe_runtime.type, g_pe_runtime.start_buffer); - } - if (g_pa_runtime.enabled) { - sync_shadow_to_openplc_by_type(g_pa_runtime.shadow_buffer, g_pa_runtime.size_bytes, - g_pa_runtime.type, g_pa_runtime.start_buffer); - } - if (g_mk_runtime.enabled) { - sync_shadow_to_openplc_by_type(g_mk_runtime.shadow_buffer, g_mk_runtime.size_bytes, - g_mk_runtime.type, g_mk_runtime.start_buffer); - } - - /* Sync all data blocks shadow -> OpenPLC */ - for (int i = 0; i < g_num_db_runtime; i++) { - s7comm_db_runtime_t *db = &g_db_runtime[i]; - sync_shadow_to_openplc_by_type(db->shadow_buffer, db->size_bytes, db->type, db->start_buffer); - } + /* Data sync is handled on-demand via RWArea callback */ } /** * @brief Called at the end of each PLC scan cycle * - * Copy ENTIRE OpenPLC buffer -> S7 buffer for ALL types. - * This makes all PLC values visible to S7 clients. Any values driven by - * the PLC program or I/O drivers during the scan cycle will overwrite - * whatever S7 clients wrote at cycle_start. + * With the journal-based approach, data synchronization happens on-demand + * via the RWArea callback. No action needed here. */ extern "C" void cycle_end(void) { - if (!g_initialized || !g_running || !g_config.enabled) { - return; - } - - /* - * Sync OpenPLC -> shadow for ALL areas - * This captures the final state after PLC execution - */ - - /* Sync all system areas OpenPLC -> shadow */ - if (g_pe_runtime.enabled) { - sync_openplc_to_shadow_by_type(g_pe_runtime.shadow_buffer, g_pe_runtime.size_bytes, - g_pe_runtime.type, g_pe_runtime.start_buffer); - } - if (g_pa_runtime.enabled) { - sync_openplc_to_shadow_by_type(g_pa_runtime.shadow_buffer, g_pa_runtime.size_bytes, - g_pa_runtime.type, g_pa_runtime.start_buffer); - } - if (g_mk_runtime.enabled) { - sync_openplc_to_shadow_by_type(g_mk_runtime.shadow_buffer, g_mk_runtime.size_bytes, - g_mk_runtime.type, g_mk_runtime.start_buffer); - } - - /* Sync all data blocks OpenPLC -> shadow */ - for (int i = 0; i < g_num_db_runtime; i++) { - s7comm_db_runtime_t *db = &g_db_runtime[i]; - sync_openplc_to_shadow_by_type(db->shadow_buffer, db->size_bytes, db->type, db->start_buffer); - } - - /* - * Lock S7 mutex and copy shadow -> S7 for ALL areas - * This makes PLC values visible to S7 clients - */ - pthread_mutex_lock(&g_s7_mutex); - - /* Copy all system areas shadow -> S7 */ - if (g_pe_runtime.enabled) { - memcpy(g_pe_runtime.s7_buffer, g_pe_runtime.shadow_buffer, g_pe_runtime.size_bytes); - } - if (g_pa_runtime.enabled) { - memcpy(g_pa_runtime.s7_buffer, g_pa_runtime.shadow_buffer, g_pa_runtime.size_bytes); - } - if (g_mk_runtime.enabled) { - memcpy(g_mk_runtime.s7_buffer, g_mk_runtime.shadow_buffer, g_mk_runtime.size_bytes); - } - - /* Copy all data blocks shadow -> S7 */ - for (int i = 0; i < g_num_db_runtime; i++) { - s7comm_db_runtime_t *db = &g_db_runtime[i]; - memcpy(db->s7_buffer, db->shadow_buffer, db->size_bytes); - } - - pthread_mutex_unlock(&g_s7_mutex); + /* Data sync is handled on-demand via RWArea callback */ } /* @@ -761,53 +624,118 @@ static void s7comm_event_callback(void *usrPtr, PSrvEvent PEvent, int Size) /* * ============================================================================= - * Buffer Synchronization Functions (Shadow <-> OpenPLC) + * On-Demand Data Synchronization (via RWArea Callback) + * + * S7 client READs: Acquire OpenPLC mutex, copy to S7 buffer, release mutex + * S7 client WRITEs: Use journal writes (thread-safe, no mutex needed) * ============================================================================= */ /** - * @brief Sync a bool buffer from shadow to OpenPLC + * @brief Map s7comm buffer type to journal buffer type * - * Copies S7 client writes to OpenPLC buffers. Used at cycle_start for ALL types. - * The PLC program and I/O drivers will overwrite values they actively control. + * Journal buffer types (from plugin_types.h): + * 0=BOOL_INPUT, 1=BOOL_OUTPUT, 2=BOOL_MEMORY + * 3=BYTE_INPUT, 4=BYTE_OUTPUT + * 5=INT_INPUT, 6=INT_OUTPUT, 7=INT_MEMORY + * 8=DINT_INPUT, 9=DINT_OUTPUT, 10=DINT_MEMORY + * 11=LINT_INPUT, 12=LINT_OUTPUT, 13=LINT_MEMORY */ -static void sync_shadow_bool_to_openplc(uint8_t *shadow, int size, s7comm_buffer_type_t type, int start_buffer) +static int map_to_journal_type(s7comm_buffer_type_t type) { - IEC_BOOL *(*buffer)[8] = NULL; + switch (type) { + case BUFFER_TYPE_BOOL_INPUT: return 0; + case BUFFER_TYPE_BOOL_OUTPUT: return 1; + case BUFFER_TYPE_BOOL_MEMORY: return 2; + case BUFFER_TYPE_INT_INPUT: return 5; + case BUFFER_TYPE_INT_OUTPUT: return 6; + case BUFFER_TYPE_INT_MEMORY: return 7; + case BUFFER_TYPE_DINT_INPUT: return 8; + case BUFFER_TYPE_DINT_OUTPUT: return 9; + case BUFFER_TYPE_DINT_MEMORY: return 10; + case BUFFER_TYPE_LINT_INPUT: return 11; + case BUFFER_TYPE_LINT_OUTPUT: return 12; + case BUFFER_TYPE_LINT_MEMORY: return 13; + default: return -1; + } +} + +/** + * @brief Find DB runtime structure by DB number + */ +static s7comm_db_runtime_t* find_db_runtime(int db_number) +{ + for (int i = 0; i < g_num_db_runtime; i++) { + if (g_db_runtime[i].db_number == db_number) { + return &g_db_runtime[i]; + } + } + return NULL; +} +/** + * @brief Find system area runtime structure by S7 protocol area code + * + * Note: The callback receives S7 protocol area codes (S7AreaPE=0x81, S7AreaPA=0x82, etc.) + * not server area codes (srvAreaPE=0, srvAreaPA=1, etc.) + */ +static s7comm_area_runtime_t* find_area_runtime(int area) +{ + switch (area) { + case S7AreaPE: + return g_pe_runtime.enabled ? &g_pe_runtime : NULL; + case S7AreaPA: + return g_pa_runtime.enabled ? &g_pa_runtime : NULL; + case S7AreaMK: + return g_mk_runtime.enabled ? &g_mk_runtime : NULL; + default: + return NULL; + } +} + +/** + * @brief Get size of a single element for a buffer type + */ +static int get_type_size(s7comm_buffer_type_t type) +{ switch (type) { case BUFFER_TYPE_BOOL_INPUT: - buffer = g_runtime_args.bool_input; - break; case BUFFER_TYPE_BOOL_OUTPUT: - buffer = g_runtime_args.bool_output; - break; case BUFFER_TYPE_BOOL_MEMORY: - buffer = g_runtime_args.bool_memory; - break; - default: - return; - } + return 1; /* 1 byte per bool group */ - int max_bytes = g_runtime_args.buffer_size - start_buffer; - if (max_bytes > size) max_bytes = size; + case BUFFER_TYPE_INT_INPUT: + case BUFFER_TYPE_INT_OUTPUT: + case BUFFER_TYPE_INT_MEMORY: + return 2; /* 2 bytes per INT */ - for (int byte_idx = 0; byte_idx < max_bytes; byte_idx++) { - uint8_t byte_val = shadow[byte_idx]; - int plc_idx = start_buffer + byte_idx; - for (int bit_idx = 0; bit_idx < 8; bit_idx++) { - IEC_BOOL *ptr = buffer[plc_idx][bit_idx]; - if (ptr != NULL) { - *ptr = (byte_val >> bit_idx) & 0x01; - } - } + case BUFFER_TYPE_DINT_INPUT: + case BUFFER_TYPE_DINT_OUTPUT: + case BUFFER_TYPE_DINT_MEMORY: + return 4; /* 4 bytes per DINT */ + + case BUFFER_TYPE_LINT_INPUT: + case BUFFER_TYPE_LINT_OUTPUT: + case BUFFER_TYPE_LINT_MEMORY: + return 8; /* 8 bytes per LINT */ + + default: + return 1; } } +/* + * ============================================================================= + * Read Functions: OpenPLC -> S7 Buffer (for S7 client READs) + * These functions copy data from OpenPLC image tables to the S7 buffer. + * Called with OpenPLC mutex held. + * ============================================================================= + */ + /** - * @brief Sync OpenPLC bool buffer to shadow + * @brief Read OpenPLC bool buffer to destination (mutex must be held) */ -static void sync_openplc_bool_to_shadow(uint8_t *shadow, int size, s7comm_buffer_type_t type, int start_buffer) +static void read_openplc_bool_to_buffer(uint8_t *dest, int size, s7comm_buffer_type_t type, int start_buffer) { IEC_BOOL *(*buffer)[8] = NULL; @@ -825,6 +753,10 @@ static void sync_openplc_bool_to_shadow(uint8_t *shadow, int size, s7comm_buffer return; } + if (buffer == NULL) { + return; + } + int max_bytes = g_runtime_args.buffer_size - start_buffer; if (max_bytes > size) max_bytes = size; @@ -837,17 +769,14 @@ static void sync_openplc_bool_to_shadow(uint8_t *shadow, int size, s7comm_buffer byte_val |= (1 << bit_idx); } } - shadow[byte_idx] = byte_val; + dest[byte_idx] = byte_val; } } /** - * @brief Sync shadow int buffer to OpenPLC (with endian conversion) - * - * Copies S7 client writes to OpenPLC buffers. Used at cycle_start for ALL types. - * The PLC program and I/O drivers will overwrite values they actively control. + * @brief Read OpenPLC int buffer to destination with endian conversion (mutex must be held) */ -static void sync_shadow_int_to_openplc(uint8_t *shadow, int size, s7comm_buffer_type_t type, int start_buffer) +static void read_openplc_int_to_buffer(uint8_t *dest, int size, s7comm_buffer_type_t type, int start_buffer) { IEC_UINT **buffer = NULL; @@ -865,7 +794,7 @@ static void sync_shadow_int_to_openplc(uint8_t *shadow, int size, s7comm_buffer_ return; } - uint16_t *shadow_words = (uint16_t *)shadow; + uint16_t *s7_words = (uint16_t *)dest; int num_words = size / 2; int max_words = g_runtime_args.buffer_size - start_buffer; if (max_words > num_words) max_words = num_words; @@ -873,52 +802,15 @@ static void sync_shadow_int_to_openplc(uint8_t *shadow, int size, s7comm_buffer_ for (int i = 0; i < max_words; i++) { IEC_UINT *ptr = buffer[start_buffer + i]; if (ptr != NULL) { - *ptr = swap16(shadow_words[i]); + s7_words[i] = swap16(*ptr); } } } /** - * @brief Sync OpenPLC int buffer to shadow (with endian conversion) + * @brief Read OpenPLC dint buffer to destination with endian conversion (mutex must be held) */ -static void sync_openplc_int_to_shadow(uint8_t *shadow, int size, s7comm_buffer_type_t type, int start_buffer) -{ - IEC_UINT **buffer = NULL; - - switch (type) { - case BUFFER_TYPE_INT_INPUT: - buffer = g_runtime_args.int_input; - break; - case BUFFER_TYPE_INT_OUTPUT: - buffer = g_runtime_args.int_output; - break; - case BUFFER_TYPE_INT_MEMORY: - buffer = g_runtime_args.int_memory; - break; - default: - return; - } - - uint16_t *shadow_words = (uint16_t *)shadow; - int num_words = size / 2; - int max_words = g_runtime_args.buffer_size - start_buffer; - if (max_words > num_words) max_words = num_words; - - for (int i = 0; i < max_words; i++) { - IEC_UINT *ptr = buffer[start_buffer + i]; - if (ptr != NULL) { - shadow_words[i] = swap16(*ptr); - } - } -} - -/** - * @brief Sync shadow dint buffer to OpenPLC (with endian conversion) - * - * Copies S7 client writes to OpenPLC buffers. Used at cycle_start for ALL types. - * The PLC program and I/O drivers will overwrite values they actively control. - */ -static void sync_shadow_dint_to_openplc(uint8_t *shadow, int size, s7comm_buffer_type_t type, int start_buffer) +static void read_openplc_dint_to_buffer(uint8_t *dest, int size, s7comm_buffer_type_t type, int start_buffer) { IEC_UDINT **buffer = NULL; @@ -936,7 +828,7 @@ static void sync_shadow_dint_to_openplc(uint8_t *shadow, int size, s7comm_buffer return; } - uint32_t *shadow_dwords = (uint32_t *)shadow; + uint32_t *s7_dwords = (uint32_t *)dest; int num_dwords = size / 4; int max_dwords = g_runtime_args.buffer_size - start_buffer; if (max_dwords > num_dwords) max_dwords = num_dwords; @@ -944,52 +836,15 @@ static void sync_shadow_dint_to_openplc(uint8_t *shadow, int size, s7comm_buffer for (int i = 0; i < max_dwords; i++) { IEC_UDINT *ptr = buffer[start_buffer + i]; if (ptr != NULL) { - *ptr = swap32(shadow_dwords[i]); + s7_dwords[i] = swap32(*ptr); } } } /** - * @brief Sync OpenPLC dint buffer to shadow (with endian conversion) + * @brief Read OpenPLC lint buffer to destination with endian conversion (mutex must be held) */ -static void sync_openplc_dint_to_shadow(uint8_t *shadow, int size, s7comm_buffer_type_t type, int start_buffer) -{ - IEC_UDINT **buffer = NULL; - - switch (type) { - case BUFFER_TYPE_DINT_INPUT: - buffer = g_runtime_args.dint_input; - break; - case BUFFER_TYPE_DINT_OUTPUT: - buffer = g_runtime_args.dint_output; - break; - case BUFFER_TYPE_DINT_MEMORY: - buffer = g_runtime_args.dint_memory; - break; - default: - return; - } - - uint32_t *shadow_dwords = (uint32_t *)shadow; - int num_dwords = size / 4; - int max_dwords = g_runtime_args.buffer_size - start_buffer; - if (max_dwords > num_dwords) max_dwords = num_dwords; - - for (int i = 0; i < max_dwords; i++) { - IEC_UDINT *ptr = buffer[start_buffer + i]; - if (ptr != NULL) { - shadow_dwords[i] = swap32(*ptr); - } - } -} - -/** - * @brief Sync shadow lint buffer to OpenPLC (with endian conversion) - * - * Copies S7 client writes to OpenPLC buffers. Used at cycle_start for ALL types. - * The PLC program and I/O drivers will overwrite values they actively control. - */ -static void sync_shadow_lint_to_openplc(uint8_t *shadow, int size, s7comm_buffer_type_t type, int start_buffer) +static void read_openplc_lint_to_buffer(uint8_t *dest, int size, s7comm_buffer_type_t type, int start_buffer) { IEC_ULINT **buffer = NULL; @@ -1007,7 +862,7 @@ static void sync_shadow_lint_to_openplc(uint8_t *shadow, int size, s7comm_buffer return; } - uint64_t *shadow_lwords = (uint64_t *)shadow; + uint64_t *s7_lwords = (uint64_t *)dest; int num_lwords = size / 8; int max_lwords = g_runtime_args.buffer_size - start_buffer; if (max_lwords > num_lwords) max_lwords = num_lwords; @@ -1015,76 +870,39 @@ static void sync_shadow_lint_to_openplc(uint8_t *shadow, int size, s7comm_buffer for (int i = 0; i < max_lwords; i++) { IEC_ULINT *ptr = buffer[start_buffer + i]; if (ptr != NULL) { - *ptr = swap64(shadow_lwords[i]); + s7_lwords[i] = swap64(*ptr); } } } /** - * @brief Sync OpenPLC lint buffer to shadow (with endian conversion) + * @brief Dispatch read from OpenPLC to buffer based on buffer type */ -static void sync_openplc_lint_to_shadow(uint8_t *shadow, int size, s7comm_buffer_type_t type, int start_buffer) -{ - IEC_ULINT **buffer = NULL; - - switch (type) { - case BUFFER_TYPE_LINT_INPUT: - buffer = g_runtime_args.lint_input; - break; - case BUFFER_TYPE_LINT_OUTPUT: - buffer = g_runtime_args.lint_output; - break; - case BUFFER_TYPE_LINT_MEMORY: - buffer = g_runtime_args.lint_memory; - break; - default: - return; - } - - uint64_t *shadow_lwords = (uint64_t *)shadow; - int num_lwords = size / 8; - int max_lwords = g_runtime_args.buffer_size - start_buffer; - if (max_lwords > num_lwords) max_lwords = num_lwords; - - for (int i = 0; i < max_lwords; i++) { - IEC_ULINT *ptr = buffer[start_buffer + i]; - if (ptr != NULL) { - shadow_lwords[i] = swap64(*ptr); - } - } -} - -/** - * @brief Dispatch sync from shadow to OpenPLC based on buffer type - * - * Used by cycle_start() for ALL types. Copies S7 client writes to OpenPLC buffers. - * Values driven by the PLC program or I/O drivers will be overwritten during the scan cycle. - */ -static void sync_shadow_to_openplc_by_type(uint8_t *shadow, int size, s7comm_buffer_type_t type, int start_buffer) +static void read_openplc_to_buffer(uint8_t *dest, int size, s7comm_buffer_type_t type, int start_buffer) { switch (type) { case BUFFER_TYPE_BOOL_INPUT: case BUFFER_TYPE_BOOL_OUTPUT: case BUFFER_TYPE_BOOL_MEMORY: - sync_shadow_bool_to_openplc(shadow, size, type, start_buffer); + read_openplc_bool_to_buffer(dest, size, type, start_buffer); break; case BUFFER_TYPE_INT_INPUT: case BUFFER_TYPE_INT_OUTPUT: case BUFFER_TYPE_INT_MEMORY: - sync_shadow_int_to_openplc(shadow, size, type, start_buffer); + read_openplc_int_to_buffer(dest, size, type, start_buffer); break; case BUFFER_TYPE_DINT_INPUT: case BUFFER_TYPE_DINT_OUTPUT: case BUFFER_TYPE_DINT_MEMORY: - sync_shadow_dint_to_openplc(shadow, size, type, start_buffer); + read_openplc_dint_to_buffer(dest, size, type, start_buffer); break; case BUFFER_TYPE_LINT_INPUT: case BUFFER_TYPE_LINT_OUTPUT: case BUFFER_TYPE_LINT_MEMORY: - sync_shadow_lint_to_openplc(shadow, size, type, start_buffer); + read_openplc_lint_to_buffer(dest, size, type, start_buffer); break; default: @@ -1092,37 +910,196 @@ static void sync_shadow_to_openplc_by_type(uint8_t *shadow, int size, s7comm_buf } } +/* + * ============================================================================= + * Write Functions: S7 Buffer -> OpenPLC via Journal (for S7 client WRITEs) + * These functions write data from S7 buffer to OpenPLC via journal. + * No mutex needed - journal writes are thread-safe. + * ============================================================================= + */ + +/** + * @brief Write bool buffer to OpenPLC via journal + */ +static void write_bool_to_openplc_journal(uint8_t *src, int size, s7comm_buffer_type_t type, int start_buffer) +{ + int journal_type = map_to_journal_type(type); + if (journal_type < 0) return; + + int max_bytes = g_runtime_args.buffer_size - start_buffer; + if (max_bytes > size) max_bytes = size; + + for (int byte_idx = 0; byte_idx < max_bytes; byte_idx++) { + uint8_t byte_val = src[byte_idx]; + int plc_idx = start_buffer + byte_idx; + for (int bit_idx = 0; bit_idx < 8; bit_idx++) { + int bit_val = (byte_val >> bit_idx) & 0x01; + g_runtime_args.journal_write_bool(journal_type, plc_idx, bit_idx, bit_val); + } + } +} + +/** + * @brief Write int buffer to OpenPLC via journal with endian conversion + */ +static void write_int_to_openplc_journal(uint8_t *src, int size, s7comm_buffer_type_t type, int start_buffer) +{ + int journal_type = map_to_journal_type(type); + if (journal_type < 0) return; + + uint16_t *s7_words = (uint16_t *)src; + int num_words = size / 2; + int max_words = g_runtime_args.buffer_size - start_buffer; + if (max_words > num_words) max_words = num_words; + + for (int i = 0; i < max_words; i++) { + uint16_t value = swap16(s7_words[i]); + g_runtime_args.journal_write_int(journal_type, start_buffer + i, value); + } +} + +/** + * @brief Write dint buffer to OpenPLC via journal with endian conversion + */ +static void write_dint_to_openplc_journal(uint8_t *src, int size, s7comm_buffer_type_t type, int start_buffer) +{ + int journal_type = map_to_journal_type(type); + if (journal_type < 0) return; + + uint32_t *s7_dwords = (uint32_t *)src; + int num_dwords = size / 4; + int max_dwords = g_runtime_args.buffer_size - start_buffer; + if (max_dwords > num_dwords) max_dwords = num_dwords; + + for (int i = 0; i < max_dwords; i++) { + uint32_t value = swap32(s7_dwords[i]); + g_runtime_args.journal_write_dint(journal_type, start_buffer + i, value); + } +} + /** - * @brief Dispatch sync from OpenPLC to shadow based on buffer type + * @brief Write lint buffer to OpenPLC via journal with endian conversion */ -static void sync_openplc_to_shadow_by_type(uint8_t *shadow, int size, s7comm_buffer_type_t type, int start_buffer) +static void write_lint_to_openplc_journal(uint8_t *src, int size, s7comm_buffer_type_t type, int start_buffer) +{ + int journal_type = map_to_journal_type(type); + if (journal_type < 0) return; + + uint64_t *s7_lwords = (uint64_t *)src; + int num_lwords = size / 8; + int max_lwords = g_runtime_args.buffer_size - start_buffer; + if (max_lwords > num_lwords) max_lwords = num_lwords; + + for (int i = 0; i < max_lwords; i++) { + uint64_t value = swap64(s7_lwords[i]); + g_runtime_args.journal_write_lint(journal_type, start_buffer + i, value); + } +} + +/** + * @brief Dispatch write from buffer to OpenPLC journal based on buffer type + */ +static void write_buffer_to_openplc_journal(uint8_t *src, int size, s7comm_buffer_type_t type, int start_buffer) { switch (type) { case BUFFER_TYPE_BOOL_INPUT: case BUFFER_TYPE_BOOL_OUTPUT: case BUFFER_TYPE_BOOL_MEMORY: - sync_openplc_bool_to_shadow(shadow, size, type, start_buffer); + write_bool_to_openplc_journal(src, size, type, start_buffer); break; case BUFFER_TYPE_INT_INPUT: case BUFFER_TYPE_INT_OUTPUT: case BUFFER_TYPE_INT_MEMORY: - sync_openplc_int_to_shadow(shadow, size, type, start_buffer); + write_int_to_openplc_journal(src, size, type, start_buffer); break; case BUFFER_TYPE_DINT_INPUT: case BUFFER_TYPE_DINT_OUTPUT: case BUFFER_TYPE_DINT_MEMORY: - sync_openplc_dint_to_shadow(shadow, size, type, start_buffer); + write_dint_to_openplc_journal(src, size, type, start_buffer); break; case BUFFER_TYPE_LINT_INPUT: case BUFFER_TYPE_LINT_OUTPUT: case BUFFER_TYPE_LINT_MEMORY: - sync_openplc_lint_to_shadow(shadow, size, type, start_buffer); + write_lint_to_openplc_journal(src, size, type, start_buffer); break; default: break; } } + +/* + * ============================================================================= + * Snap7 RWArea Callback - On-Demand Data Synchronization + * ============================================================================= + */ + +/** + * @brief Snap7 RWArea callback for on-demand data synchronization + * + * Called by Snap7 when an S7 client reads or writes data. + * - On READ: Acquire OpenPLC mutex, copy fresh data to S7 buffer, release mutex + * - On WRITE: Use journal writes (thread-safe, no mutex needed) + * + * @param usrPtr User pointer (unused) + * @param Sender Client identifier + * @param Operation OperationRead or OperationWrite + * @param PTag S7 tag with Area, DBNumber, Start, Size + * @param pUsrData Pointer to data buffer + * @return 0 to accept operation, non-zero to reject + */ +static int s7comm_rw_area_callback(void *usrPtr, int Sender, int Operation, PS7Tag PTag, void *pUsrData) +{ + (void)usrPtr; + (void)Sender; + + if (pUsrData == NULL || PTag == NULL) { + return -1; + } + + s7comm_buffer_type_t type; + int start_buffer; + int size = PTag->Size; + + /* Determine mapping based on S7 protocol area code */ + if (PTag->Area == S7AreaDB) { + /* Data block - look up configuration */ + s7comm_db_runtime_t *db = find_db_runtime(PTag->DBNumber); + if (db == NULL) { + /* DB not configured - return zeros for read, ignore write */ + return 0; + } + type = db->type; + start_buffer = db->start_buffer + (PTag->Start / get_type_size(type)); + } else { + /* System area (PE, PA, MK) */ + s7comm_area_runtime_t *area = find_area_runtime(PTag->Area); + if (area == NULL) { + /* Area not configured - return zeros for read, ignore write */ + return 0; + } + type = area->type; + start_buffer = area->start_buffer + (PTag->Start / get_type_size(type)); + } + + if (Operation == OperationRead) { + /* + * S7 client is READing - provide fresh data from OpenPLC + * Acquire mutex, copy data, release mutex + */ + g_runtime_args.mutex_take(g_runtime_args.buffer_mutex); + read_openplc_to_buffer((uint8_t *)pUsrData, size, type, start_buffer); + g_runtime_args.mutex_give(g_runtime_args.buffer_mutex); + } else if (Operation == OperationWrite) { + /* + * S7 client is WRITing - journal the changes + * Journal writes are thread-safe, no mutex needed + */ + write_buffer_to_openplc_journal((uint8_t *)pUsrData, size, type, start_buffer); + } + + return 0; /* Accept operation */ +} diff --git a/core/src/drivers/plugins/python/modbus_slave/simple_modbus.py b/core/src/drivers/plugins/python/modbus_slave/simple_modbus.py index b467c953..6c6c6f12 100644 --- a/core/src/drivers/plugins/python/modbus_slave/simple_modbus.py +++ b/core/src/drivers/plugins/python/modbus_slave/simple_modbus.py @@ -119,7 +119,11 @@ def getValues(self, address, count=1): return values def setValues(self, address, values): - """Set coil values to OpenPLC bool_output using SafeBufferAccess""" + """Set coil values to OpenPLC bool_output using SafeBufferAccess. + + Note: Writes go through the journal buffer system which is internally + thread-safe, so no mutex is needed for write operations. + """ address = address - 1 # Modbus addresses are 0-based if not self.safe_buffer_access.is_valid: @@ -129,9 +133,7 @@ def setValues(self, address, values): ) return - # Ensure thread-safe access - self.safe_buffer_access.acquire_mutex() - + # Journal writes are thread-safe, no mutex needed for i, value in enumerate(values): coil_addr = address + i @@ -141,15 +143,12 @@ def setValues(self, address, values): bit_idx = coil_addr % MAX_BITS # bit within buffer _, error_msg = self.safe_buffer_access.write_bool_output( - buffer_idx, bit_idx, bool(value), thread_safe=False + buffer_idx, bit_idx, bool(value) ) if error_msg != "Success": if logger: logger.error(f"Error setting coil {coil_addr}: {error_msg}") - # Release mutex after access - self.safe_buffer_access.release_mutex() - class OpenPLCDiscreteInputsDataBlock(ModbusSparseDataBlock): """Custom Modbus discrete inputs data block that mirrors OpenPLC bool_input.""" @@ -330,7 +329,11 @@ def getValues(self, address, count=1): return values def setValues(self, address, values): - """Set holding register values to OpenPLC int_output using SafeBufferAccess""" + """Set holding register values to OpenPLC int_output using SafeBufferAccess. + + Note: Writes go through the journal buffer system which is internally + thread-safe, so no mutex is needed for write operations. + """ address = address - 1 # Modbus addresses are 0-based if not self.safe_buffer_access.is_valid: @@ -340,23 +343,16 @@ def setValues(self, address, values): ) return - # Ensure buffer mutex - self.safe_buffer_access.acquire_mutex() - + # Journal writes are thread-safe, no mutex needed for i, value in enumerate(values): reg_addr = address + i if reg_addr < self.num_registers: - _, error_msg = self.safe_buffer_access.write_int_output( - reg_addr, value, thread_safe=False - ) + _, error_msg = self.safe_buffer_access.write_int_output(reg_addr, value) if error_msg != "Success": if logger: logger.error(f"Error setting holding register {reg_addr}: {error_msg}") - # Release mutex after access - self.safe_buffer_access.release_mutex() - class OpenPLCSegmentedCoilsDataBlock(ModbusSparseDataBlock): """ @@ -459,7 +455,11 @@ def getValues(self, address, count=1): self.safe_buffer_access.release_mutex() def setValues(self, address, values): - """Set coil values to appropriate OpenPLC buffer based on address segmentation""" + """Set coil values to appropriate OpenPLC buffer based on address segmentation. + + Note: Writes go through the journal buffer system which is internally + thread-safe, so no mutex is needed for write operations. + """ address = address - 1 # Modbus addresses are 1-based if not self.safe_buffer_access.is_valid: @@ -469,29 +469,26 @@ def setValues(self, address, values): ) return - self.safe_buffer_access.acquire_mutex() - try: - for i, value in enumerate(values): - coil_addr = address + i - segment, buffer_idx, bit_idx = self._get_segment_info(coil_addr) + # Journal writes are thread-safe, no mutex needed + for i, value in enumerate(values): + coil_addr = address + i + segment, buffer_idx, bit_idx = self._get_segment_info(coil_addr) - if segment == "qx": - _, error_msg = self.safe_buffer_access.write_bool_output( - buffer_idx, bit_idx, bool(value), thread_safe=False - ) - if error_msg != "Success": - if logger: - logger.error(f"Error setting coil %QX{coil_addr}: {error_msg}") - elif segment == "mx": - mx_addr = coil_addr - self.qx_bits - _, error_msg = self.safe_buffer_access.write_bool_memory( - buffer_idx, bit_idx, bool(value), thread_safe=False - ) - if error_msg != "Success": - if logger: - logger.error(f"Error setting coil %MX{mx_addr}: {error_msg}") - finally: - self.safe_buffer_access.release_mutex() + if segment == "qx": + _, error_msg = self.safe_buffer_access.write_bool_output( + buffer_idx, bit_idx, bool(value) + ) + if error_msg != "Success": + if logger: + logger.error(f"Error setting coil %QX{coil_addr}: {error_msg}") + elif segment == "mx": + mx_addr = coil_addr - self.qx_bits + _, error_msg = self.safe_buffer_access.write_bool_memory( + buffer_idx, bit_idx, bool(value) + ) + if error_msg != "Success": + if logger: + logger.error(f"Error setting coil %MX{mx_addr}: {error_msg}") class OpenPLCSegmentedHoldingRegistersDataBlock(ModbusSparseDataBlock): @@ -704,7 +701,13 @@ def getValues(self, address, count=1): self.safe_buffer_access.release_mutex() def setValues(self, address, values): - """Set holding register values to appropriate OpenPLC buffer based on address segmentation""" + """Set holding register values to appropriate OpenPLC buffer based on address segmentation. + + Note: While journal writes are thread-safe, this method still uses mutex because + partial DINT/LINT updates require read-modify-write (RMW) operations. The mutex + ensures consistency between the read and subsequent write for multi-word values. + Simple QW/MW writes don't strictly need the mutex, but we keep it for RMW safety. + """ address = address - 1 # Modbus addresses are 1-based if not self.safe_buffer_access.is_valid: @@ -714,18 +717,20 @@ def setValues(self, address, values): ) return + # Mutex needed for read-modify-write consistency on partial DINT/LINT updates self.safe_buffer_access.acquire_mutex() try: # For multi-word values, we need to handle partial writes carefully # Build a map of pending multi-word updates - pending_dint = {} # value_idx -> {word_offset: value} - pending_lint = {} # value_idx -> {word_offset: value} + pending_dint = {} # value_idx -> [words] + pending_lint = {} # value_idx -> [words] for i, value in enumerate(values): reg_addr = address + i segment, value_idx, word_offset = self._get_segment_info(reg_addr) if segment == "qw": + # Simple 16-bit write - journal handles thread safety _, error_msg = self.safe_buffer_access.write_int_output( value_idx, value & 0xFFFF, thread_safe=False ) @@ -734,6 +739,7 @@ def setValues(self, address, values): logger.error(f"Error setting %QW{value_idx}: {error_msg}") elif segment == "mw": + # Simple 16-bit write - journal handles thread safety _, error_msg = self.safe_buffer_access.write_int_memory( value_idx, value & 0xFFFF, thread_safe=False ) @@ -742,7 +748,7 @@ def setValues(self, address, values): logger.error(f"Error setting %MW{value_idx}: {error_msg}") elif segment == "md": - # Collect words for this DINT + # Partial 32-bit write - need RMW for consistency if value_idx not in pending_dint: # Read current value to preserve unchanged words current, _ = self.safe_buffer_access.read_dint_memory( @@ -754,7 +760,7 @@ def setValues(self, address, values): pending_dint[value_idx][word_offset] = value & 0xFFFF elif segment == "ml": - # Collect words for this LINT + # Partial 64-bit write - need RMW for consistency if value_idx not in pending_lint: # Read current value to preserve unchanged words current, _ = self.safe_buffer_access.read_lint_memory( @@ -765,7 +771,7 @@ def setValues(self, address, values): ) pending_lint[value_idx][word_offset] = value & 0xFFFF - # Write pending DINT values + # Write pending DINT values (RMW complete, write to journal) for value_idx, words in pending_dint.items(): dint_value = self._combine_words_to_dint(words) _, error_msg = self.safe_buffer_access.write_dint_memory( @@ -775,7 +781,7 @@ def setValues(self, address, values): if logger: logger.error(f"Error setting %MD{value_idx}: {error_msg}") - # Write pending LINT values + # Write pending LINT values (RMW complete, write to journal) for value_idx, words in pending_lint.items(): lint_value = self._combine_words_to_lint(words) _, error_msg = self.safe_buffer_access.write_lint_memory( diff --git a/core/src/drivers/plugins/python/shared/batch_processor.py b/core/src/drivers/plugins/python/shared/batch_processor.py index 332acfc7..0f0d09de 100644 --- a/core/src/drivers/plugins/python/shared/batch_processor.py +++ b/core/src/drivers/plugins/python/shared/batch_processor.py @@ -3,6 +3,12 @@ This module handles batch operations for optimized buffer access. It processes multiple read/write operations with a single mutex acquisition. + +Note on thread safety: +- Read operations require mutex to ensure consistent reads from image tables +- Write operations go through the journal buffer system which is internally + thread-safe. The mutex is still acquired for batch writes to maintain API + consistency and to support mixed read/write operations atomically. """ from typing import List, Tuple, Dict, Any diff --git a/core/src/drivers/plugins/python/shared/buffer_accessor.py b/core/src/drivers/plugins/python/shared/buffer_accessor.py index 685c87d8..58222e7d 100644 --- a/core/src/drivers/plugins/python/shared/buffer_accessor.py +++ b/core/src/drivers/plugins/python/shared/buffer_accessor.py @@ -33,8 +33,29 @@ class GenericBufferAccessor(IBufferAccessor): This class encapsulates the complex ctypes buffer access logic and provides a clean, type-agnostic interface for buffer operations. It eliminates the massive code duplication that existed in the original SafeBufferAccess class. + + Write operations use the journal buffer system for race-condition-free writes. + Read operations access buffers directly (reads are always safe). """ + # Journal buffer type mapping (matches journal_buffer_type_t enum) + JOURNAL_TYPE_MAP = { + "bool_input": 0, # JOURNAL_BOOL_INPUT + "bool_output": 1, # JOURNAL_BOOL_OUTPUT + "bool_memory": 2, # JOURNAL_BOOL_MEMORY + "byte_input": 3, # JOURNAL_BYTE_INPUT + "byte_output": 4, # JOURNAL_BYTE_OUTPUT + "int_input": 5, # JOURNAL_INT_INPUT + "int_output": 6, # JOURNAL_INT_OUTPUT + "int_memory": 7, # JOURNAL_INT_MEMORY + "dint_input": 8, # JOURNAL_DINT_INPUT + "dint_output": 9, # JOURNAL_DINT_OUTPUT + "dint_memory": 10, # JOURNAL_DINT_MEMORY + "lint_input": 11, # JOURNAL_LINT_INPUT + "lint_output": 12, # JOURNAL_LINT_OUTPUT + "lint_memory": 13, # JOURNAL_LINT_MEMORY + } + def __init__(self, runtime_args, validator: BufferValidator, mutex_manager: MutexManager): """ Initialize the generic buffer accessor. @@ -97,12 +118,16 @@ def write_buffer( """ Generic buffer write operation. + Writes go through the journal buffer system for race-condition-free operation. + The journal is internally thread-safe, so the thread_safe parameter is kept + for backward compatibility but writes no longer require explicit mutex. + Args: buffer_type: Buffer type name (e.g., 'bool_output', 'int_output') buffer_idx: Buffer index value: Value to write bit_idx: Bit index (required for boolean operations) - thread_safe: Whether to use mutex protection + thread_safe: Kept for backward compatibility (journal is always thread-safe) Returns: Tuple[bool, str]: (success, error_message) @@ -117,18 +142,11 @@ def write_buffer( # Get buffer type info buffer_type_obj, direction = self.buffer_types.get_buffer_info(buffer_type) - # Define the write operation - def do_write(): - return self._perform_write( - buffer_type, buffer_type_obj, direction, buffer_idx, value, bit_idx - ) - - # Execute with or without mutex - if thread_safe: - result = self.mutex.with_mutex(do_write) - return result if isinstance(result, tuple) else (result, "Success") - else: - return do_write() + # Journal writes are thread-safe internally, no mutex needed + # The thread_safe parameter is ignored but kept for backward compatibility + return self._perform_write( + buffer_type, buffer_type_obj, direction, buffer_idx, value, bit_idx + ) def get_buffer_pointer(self, buffer_type: str) -> Optional[ctypes.POINTER]: """ @@ -215,27 +233,61 @@ def _perform_write( ) -> Tuple[bool, str]: """ Internal method to perform the actual buffer write operation. + + Uses the journal buffer system for race-condition-free writes. + All writes go through the journal and are applied atomically at the + start of the next PLC scan cycle. """ try: - # Get the appropriate buffer pointer - buffer_ptr = self.get_buffer_pointer(buffer_type) - if buffer_ptr is None or buffer_ptr.contents is None: - return False, f"Buffer pointer not available for {buffer_type}" + # Get journal buffer type + journal_type = self.JOURNAL_TYPE_MAP.get(buffer_type) + if journal_type is None: + return False, f"Unknown buffer type: {buffer_type}" # Handle boolean operations (require bit indexing) if buffer_type_obj.name == "bool": if bit_idx is None: return False, "Bit index required for boolean operations" - # Set the specific bit within the buffer - buffer_ptr[buffer_idx][bit_idx].contents.value = 1 if value else 0 + # Write through journal + result = self.args.journal_write_bool( + journal_type, buffer_idx, bit_idx, 1 if value else 0 + ) + if result != 0: + return False, f"Journal write failed with code {result}" return True, "Success" - # Handle other buffer types (direct value assignment) - else: - buffer_ptr[buffer_idx].contents.value = value + # Handle byte operations + elif buffer_type_obj.name == "byte": + result = self.args.journal_write_byte(journal_type, buffer_idx, int(value) & 0xFF) + if result != 0: + return False, f"Journal write failed with code {result}" return True, "Success" + # Handle int operations (16-bit) + elif buffer_type_obj.name == "int": + result = self.args.journal_write_int(journal_type, buffer_idx, int(value) & 0xFFFF) + if result != 0: + return False, f"Journal write failed with code {result}" + return True, "Success" + + # Handle dint operations (32-bit) + elif buffer_type_obj.name == "dint": + result = self.args.journal_write_dint(journal_type, buffer_idx, int(value) & 0xFFFFFFFF) + if result != 0: + return False, f"Journal write failed with code {result}" + return True, "Success" + + # Handle lint operations (64-bit) + elif buffer_type_obj.name == "lint": + result = self.args.journal_write_lint(journal_type, buffer_idx, int(value)) + if result != 0: + return False, f"Journal write failed with code {result}" + return True, "Success" + + else: + return False, f"Unsupported buffer type: {buffer_type_obj.name}" + except (AttributeError, TypeError, ValueError, OSError, MemoryError) as e: return False, f"Buffer write error: {e}" diff --git a/core/src/drivers/plugins/python/shared/plugin_runtime_args.py b/core/src/drivers/plugins/python/shared/plugin_runtime_args.py index 13c44257..a091e055 100644 --- a/core/src/drivers/plugins/python/shared/plugin_runtime_args.py +++ b/core/src/drivers/plugins/python/shared/plugin_runtime_args.py @@ -50,6 +50,13 @@ class PluginRuntimeArgs(ctypes.Structure): ("log_debug", ctypes.CFUNCTYPE(None, ctypes.c_char_p)), ("log_warn", ctypes.CFUNCTYPE(None, ctypes.c_char_p)), ("log_error", ctypes.CFUNCTYPE(None, ctypes.c_char_p)), + # Journal write function pointers for race-condition-free buffer writes + # int (*func)(int type, int index, int bit/value, int value) + ("journal_write_bool", ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_int, ctypes.c_int, ctypes.c_int, ctypes.c_int)), + ("journal_write_byte", ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_int, ctypes.c_int, ctypes.c_int)), + ("journal_write_int", ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_int, ctypes.c_int, ctypes.c_int)), + ("journal_write_dint", ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_int, ctypes.c_int, ctypes.c_uint)), + ("journal_write_lint", ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_int, ctypes.c_int, ctypes.c_ulonglong)), ] def validate_pointers(self): diff --git a/core/src/plc_app/journal_buffer.c b/core/src/plc_app/journal_buffer.c new file mode 100644 index 00000000..7358f75e --- /dev/null +++ b/core/src/plc_app/journal_buffer.c @@ -0,0 +1,506 @@ +/** + * @file journal_buffer.c + * @brief Journal Buffer Implementation for Race-Condition-Free Plugin Writes + * + * This implementation provides: + * - Static journal buffer with configurable max entries + * - Thread-safe write operations with mutex protection + * - Atomic application of journal entries at cycle start + * - Emergency flush when buffer is full + * - Last-writer-wins conflict resolution via sequence numbers + */ + +#include "journal_buffer.h" +#include +#include + +/* + * ============================================================================= + * Static State + * ============================================================================= + */ + +/* Journal entries buffer */ +static journal_entry_t g_entries[JOURNAL_MAX_ENTRIES]; + +/* Current number of entries in journal */ +static size_t g_count = 0; + +/* Next sequence number to assign (auto-increment) */ +static uint32_t g_next_sequence = 0; + +/* Journal mutex - protects g_entries, g_count, g_next_sequence */ +static pthread_mutex_t g_journal_mutex = PTHREAD_MUTEX_INITIALIZER; + +/* Buffer pointers for applying entries */ +static journal_buffer_ptrs_t g_buffer_ptrs; + +/* Initialization flag */ +static bool g_initialized = false; + +/* + * ============================================================================= + * Forward Declarations + * ============================================================================= + */ +static void apply_entry(const journal_entry_t *entry); +static void emergency_flush_locked(void); + +/* + * ============================================================================= + * Initialization and Cleanup + * ============================================================================= + */ + +int journal_init(const journal_buffer_ptrs_t *buffer_ptrs) +{ + if (buffer_ptrs == NULL) { + fprintf(stderr, "[JOURNAL] Error: buffer_ptrs is NULL\n"); + return -1; + } + + if (buffer_ptrs->image_mutex == NULL) { + fprintf(stderr, "[JOURNAL] Error: image_mutex is NULL\n"); + return -1; + } + + pthread_mutex_lock(&g_journal_mutex); + + /* Copy buffer pointers */ + memcpy(&g_buffer_ptrs, buffer_ptrs, sizeof(journal_buffer_ptrs_t)); + + /* Reset journal state */ + g_count = 0; + g_next_sequence = 0; + memset(g_entries, 0, sizeof(g_entries)); + + g_initialized = true; + + pthread_mutex_unlock(&g_journal_mutex); + + return 0; +} + +void journal_cleanup(void) +{ + pthread_mutex_lock(&g_journal_mutex); + + g_initialized = false; + g_count = 0; + g_next_sequence = 0; + memset(&g_buffer_ptrs, 0, sizeof(g_buffer_ptrs)); + + pthread_mutex_unlock(&g_journal_mutex); +} + +bool journal_is_initialized(void) +{ + bool result; + pthread_mutex_lock(&g_journal_mutex); + result = g_initialized; + pthread_mutex_unlock(&g_journal_mutex); + return result; +} + +/* + * ============================================================================= + * Write Functions + * ============================================================================= + */ + +/** + * @brief Internal function to add an entry to the journal + * + * Must be called with g_journal_mutex held. + * Handles emergency flush if buffer is full. + * + * @return Pointer to the new entry, or NULL on failure + */ +static journal_entry_t *add_entry_locked(void) +{ + /* Check if buffer is full */ + if (g_count >= JOURNAL_MAX_ENTRIES) { + /* Emergency flush: apply all entries and clear */ + emergency_flush_locked(); + } + + /* Add new entry */ + journal_entry_t *entry = &g_entries[g_count]; + entry->sequence = g_next_sequence++; + g_count++; + + return entry; +} + +int journal_write_bool(journal_buffer_type_t type, uint16_t index, + uint8_t bit, bool value) +{ + if (!g_initialized) { + return -1; + } + + /* Validate type */ + if (type != JOURNAL_BOOL_INPUT && + type != JOURNAL_BOOL_OUTPUT && + type != JOURNAL_BOOL_MEMORY) { + return -1; + } + + /* Validate bit index */ + if (bit > 7) { + return -1; + } + + pthread_mutex_lock(&g_journal_mutex); + + journal_entry_t *entry = add_entry_locked(); + if (entry == NULL) { + pthread_mutex_unlock(&g_journal_mutex); + return -1; + } + + entry->buffer_type = (uint8_t)type; + entry->index = index; + entry->bit_index = bit; + entry->value = value ? 1 : 0; + + pthread_mutex_unlock(&g_journal_mutex); + return 0; +} + +int journal_write_byte(journal_buffer_type_t type, uint16_t index, + uint8_t value) +{ + if (!g_initialized) { + return -1; + } + + /* Validate type */ + if (type != JOURNAL_BYTE_INPUT && type != JOURNAL_BYTE_OUTPUT) { + return -1; + } + + pthread_mutex_lock(&g_journal_mutex); + + journal_entry_t *entry = add_entry_locked(); + if (entry == NULL) { + pthread_mutex_unlock(&g_journal_mutex); + return -1; + } + + entry->buffer_type = (uint8_t)type; + entry->index = index; + entry->bit_index = 0xFF; /* Not applicable for non-bool types */ + entry->value = value; + + pthread_mutex_unlock(&g_journal_mutex); + return 0; +} + +int journal_write_int(journal_buffer_type_t type, uint16_t index, + uint16_t value) +{ + if (!g_initialized) { + return -1; + } + + /* Validate type */ + if (type != JOURNAL_INT_INPUT && + type != JOURNAL_INT_OUTPUT && + type != JOURNAL_INT_MEMORY) { + return -1; + } + + pthread_mutex_lock(&g_journal_mutex); + + journal_entry_t *entry = add_entry_locked(); + if (entry == NULL) { + pthread_mutex_unlock(&g_journal_mutex); + return -1; + } + + entry->buffer_type = (uint8_t)type; + entry->index = index; + entry->bit_index = 0xFF; + entry->value = value; + + pthread_mutex_unlock(&g_journal_mutex); + return 0; +} + +int journal_write_dint(journal_buffer_type_t type, uint16_t index, + uint32_t value) +{ + if (!g_initialized) { + return -1; + } + + /* Validate type */ + if (type != JOURNAL_DINT_INPUT && + type != JOURNAL_DINT_OUTPUT && + type != JOURNAL_DINT_MEMORY) { + return -1; + } + + pthread_mutex_lock(&g_journal_mutex); + + journal_entry_t *entry = add_entry_locked(); + if (entry == NULL) { + pthread_mutex_unlock(&g_journal_mutex); + return -1; + } + + entry->buffer_type = (uint8_t)type; + entry->index = index; + entry->bit_index = 0xFF; + entry->value = value; + + pthread_mutex_unlock(&g_journal_mutex); + return 0; +} + +int journal_write_lint(journal_buffer_type_t type, uint16_t index, + uint64_t value) +{ + if (!g_initialized) { + return -1; + } + + /* Validate type */ + if (type != JOURNAL_LINT_INPUT && + type != JOURNAL_LINT_OUTPUT && + type != JOURNAL_LINT_MEMORY) { + return -1; + } + + pthread_mutex_lock(&g_journal_mutex); + + journal_entry_t *entry = add_entry_locked(); + if (entry == NULL) { + pthread_mutex_unlock(&g_journal_mutex); + return -1; + } + + entry->buffer_type = (uint8_t)type; + entry->index = index; + entry->bit_index = 0xFF; + entry->value = value; + + pthread_mutex_unlock(&g_journal_mutex); + return 0; +} + +/* + * ============================================================================= + * Apply and Clear + * ============================================================================= + */ + +/** + * @brief Apply a single journal entry to the image tables + * + * @param entry The entry to apply + */ +static void apply_entry(const journal_entry_t *entry) +{ + uint16_t idx = entry->index; + + /* Bounds check */ + if (idx >= (uint16_t)g_buffer_ptrs.buffer_size) { + return; + } + + switch ((journal_buffer_type_t)entry->buffer_type) { + case JOURNAL_BOOL_INPUT: { + IEC_BOOL *ptr = g_buffer_ptrs.bool_input[idx][entry->bit_index]; + if (ptr != NULL) { + *ptr = (IEC_BOOL)(entry->value & 1); + } + break; + } + case JOURNAL_BOOL_OUTPUT: { + IEC_BOOL *ptr = g_buffer_ptrs.bool_output[idx][entry->bit_index]; + if (ptr != NULL) { + *ptr = (IEC_BOOL)(entry->value & 1); + } + break; + } + case JOURNAL_BOOL_MEMORY: { + IEC_BOOL *ptr = g_buffer_ptrs.bool_memory[idx][entry->bit_index]; + if (ptr != NULL) { + *ptr = (IEC_BOOL)(entry->value & 1); + } + break; + } + case JOURNAL_BYTE_INPUT: { + IEC_BYTE *ptr = g_buffer_ptrs.byte_input[idx]; + if (ptr != NULL) { + *ptr = (IEC_BYTE)(entry->value & 0xFF); + } + break; + } + case JOURNAL_BYTE_OUTPUT: { + IEC_BYTE *ptr = g_buffer_ptrs.byte_output[idx]; + if (ptr != NULL) { + *ptr = (IEC_BYTE)(entry->value & 0xFF); + } + break; + } + case JOURNAL_INT_INPUT: { + IEC_UINT *ptr = g_buffer_ptrs.int_input[idx]; + if (ptr != NULL) { + *ptr = (IEC_UINT)(entry->value & 0xFFFF); + } + break; + } + case JOURNAL_INT_OUTPUT: { + IEC_UINT *ptr = g_buffer_ptrs.int_output[idx]; + if (ptr != NULL) { + *ptr = (IEC_UINT)(entry->value & 0xFFFF); + } + break; + } + case JOURNAL_INT_MEMORY: { + IEC_UINT *ptr = g_buffer_ptrs.int_memory[idx]; + if (ptr != NULL) { + *ptr = (IEC_UINT)(entry->value & 0xFFFF); + } + break; + } + case JOURNAL_DINT_INPUT: { + IEC_UDINT *ptr = g_buffer_ptrs.dint_input[idx]; + if (ptr != NULL) { + *ptr = (IEC_UDINT)(entry->value & 0xFFFFFFFF); + } + break; + } + case JOURNAL_DINT_OUTPUT: { + IEC_UDINT *ptr = g_buffer_ptrs.dint_output[idx]; + if (ptr != NULL) { + *ptr = (IEC_UDINT)(entry->value & 0xFFFFFFFF); + } + break; + } + case JOURNAL_DINT_MEMORY: { + IEC_UDINT *ptr = g_buffer_ptrs.dint_memory[idx]; + if (ptr != NULL) { + *ptr = (IEC_UDINT)(entry->value & 0xFFFFFFFF); + } + break; + } + case JOURNAL_LINT_INPUT: { + IEC_ULINT *ptr = g_buffer_ptrs.lint_input[idx]; + if (ptr != NULL) { + *ptr = (IEC_ULINT)entry->value; + } + break; + } + case JOURNAL_LINT_OUTPUT: { + IEC_ULINT *ptr = g_buffer_ptrs.lint_output[idx]; + if (ptr != NULL) { + *ptr = (IEC_ULINT)entry->value; + } + break; + } + case JOURNAL_LINT_MEMORY: { + IEC_ULINT *ptr = g_buffer_ptrs.lint_memory[idx]; + if (ptr != NULL) { + *ptr = (IEC_ULINT)entry->value; + } + break; + } + default: + /* Unknown type - ignore */ + break; + } +} + +void journal_apply_and_clear(void) +{ + if (!g_initialized) { + return; + } + + pthread_mutex_lock(&g_journal_mutex); + + /* Apply all entries in sequence order (they're already in order) */ + for (size_t i = 0; i < g_count; i++) { + apply_entry(&g_entries[i]); + } + + /* Clear journal */ + g_count = 0; + g_next_sequence = 0; + + pthread_mutex_unlock(&g_journal_mutex); +} + +/* + * ============================================================================= + * Emergency Flush + * ============================================================================= + */ + +/** + * @brief Emergency flush - apply all entries when buffer is full + * + * Called when a new write is attempted but the journal buffer is full. + * Must be called with g_journal_mutex already held. + * + * Lock ordering to prevent deadlock: + * 1. Release journal mutex + * 2. Acquire image mutex + * 3. Re-acquire journal mutex + * 4. Apply entries + * 5. Release image mutex + * (Continue holding journal mutex for the pending write) + */ +static void emergency_flush_locked(void) +{ + /* Release journal mutex to respect lock ordering */ + pthread_mutex_unlock(&g_journal_mutex); + + /* Acquire image mutex first (lock ordering) */ + pthread_mutex_lock(g_buffer_ptrs.image_mutex); + + /* Re-acquire journal mutex */ + pthread_mutex_lock(&g_journal_mutex); + + /* Apply all entries */ + for (size_t i = 0; i < g_count; i++) { + apply_entry(&g_entries[i]); + } + + /* Clear journal */ + g_count = 0; + g_next_sequence = 0; + + /* Release image mutex (keep journal mutex for the pending write) */ + pthread_mutex_unlock(g_buffer_ptrs.image_mutex); + + /* Note: We return with g_journal_mutex still held */ +} + +/* + * ============================================================================= + * Diagnostics + * ============================================================================= + */ + +size_t journal_pending_count(void) +{ + size_t count; + pthread_mutex_lock(&g_journal_mutex); + count = g_count; + pthread_mutex_unlock(&g_journal_mutex); + return count; +} + +uint32_t journal_get_sequence(void) +{ + uint32_t seq; + pthread_mutex_lock(&g_journal_mutex); + seq = g_next_sequence; + pthread_mutex_unlock(&g_journal_mutex); + return seq; +} diff --git a/core/src/plc_app/journal_buffer.h b/core/src/plc_app/journal_buffer.h new file mode 100644 index 00000000..b819eeb1 --- /dev/null +++ b/core/src/plc_app/journal_buffer.h @@ -0,0 +1,234 @@ +/** + * @file journal_buffer.h + * @brief Journal Buffer System for Race-Condition-Free Plugin Writes + * + * This module provides a journaled write mechanism for plugins to write to + * OpenPLC image tables without race conditions. All plugin writes go through + * the journal buffer, which is applied atomically at the start of each PLC + * scan cycle. + * + * Key design principles: + * - Writes go to journal: All plugin writes are recorded with a sequence number + * - Reads are direct: Plugins read directly from image tables + * - Atomic application: Journal is applied at cycle_start before plugin hooks + * - Last writer wins: Entries applied in sequence order + * + * Usage: + * // In plugin code, instead of: + * // *bool_output[5][3] = true; + * // Use: + * journal_write_bool(JOURNAL_BOOL_OUTPUT, 5, 3, true); + * + * @note Thread-safe. Uses internal mutex for journal operations. + */ + +#ifndef JOURNAL_BUFFER_H +#define JOURNAL_BUFFER_H + +#include +#include +#include +#include +#include "../lib/iec_types.h" + +#ifdef __cplusplus +extern "C" { +#endif + +/** + * @brief Maximum number of journal entries per cycle + * + * If this limit is reached, an emergency flush is triggered to apply + * pending entries and make room for new ones. + */ +#define JOURNAL_MAX_ENTRIES 1024 + +/** + * @brief Buffer type enumeration for journal entries + * + * Matches the OpenPLC image table types. + */ +typedef enum { + JOURNAL_BOOL_INPUT = 0, + JOURNAL_BOOL_OUTPUT, + JOURNAL_BOOL_MEMORY, + JOURNAL_BYTE_INPUT, + JOURNAL_BYTE_OUTPUT, + JOURNAL_INT_INPUT, + JOURNAL_INT_OUTPUT, + JOURNAL_INT_MEMORY, + JOURNAL_DINT_INPUT, + JOURNAL_DINT_OUTPUT, + JOURNAL_DINT_MEMORY, + JOURNAL_LINT_INPUT, + JOURNAL_LINT_OUTPUT, + JOURNAL_LINT_MEMORY, + JOURNAL_TYPE_COUNT +} journal_buffer_type_t; + +/** + * @brief Journal entry structure + * + * Each entry represents a single write operation to be applied. + * Entries are applied in sequence order during journal_apply_and_clear(). + */ +typedef struct { + uint32_t sequence; /**< Auto-increment, determines apply order */ + uint8_t buffer_type; /**< journal_buffer_type_t enum */ + uint8_t bit_index; /**< For bool types: 0-7, for others: 0xFF */ + uint16_t index; /**< Buffer array index */ + uint64_t value; /**< Value to write (sized for largest type) */ +} journal_entry_t; + +/** + * @brief Buffer pointers structure for journal initialization + * + * Contains pointers to all image table arrays and the image table mutex. + */ +typedef struct { + /* Boolean buffers (pointer to 2D array) */ + IEC_BOOL *(*bool_input)[8]; + IEC_BOOL *(*bool_output)[8]; + IEC_BOOL *(*bool_memory)[8]; + + /* Byte buffers */ + IEC_BYTE **byte_input; + IEC_BYTE **byte_output; + + /* Integer buffers (16-bit) */ + IEC_UINT **int_input; + IEC_UINT **int_output; + IEC_UINT **int_memory; + + /* Double integer buffers (32-bit) */ + IEC_UDINT **dint_input; + IEC_UDINT **dint_output; + IEC_UDINT **dint_memory; + + /* Long integer buffers (64-bit) */ + IEC_ULINT **lint_input; + IEC_ULINT **lint_output; + IEC_ULINT **lint_memory; + + /* Buffer size (number of elements in each array) */ + int buffer_size; + + /* Image table mutex (for emergency flush and apply operations) */ + pthread_mutex_t *image_mutex; +} journal_buffer_ptrs_t; + +/** + * @brief Initialize the journal buffer system + * + * Must be called during runtime initialization, after image tables are set up. + * + * @param buffer_ptrs Pointer to structure containing image table pointers + * @return 0 on success, -1 on failure + */ +int journal_init(const journal_buffer_ptrs_t *buffer_ptrs); + +/** + * @brief Cleanup journal buffer resources + * + * Should be called during runtime shutdown. + */ +void journal_cleanup(void); + +/** + * @brief Check if journal buffer is initialized + * + * @return true if initialized and ready for use + */ +bool journal_is_initialized(void); + +/** + * @brief Write a boolean value to the journal + * + * @param type Buffer type (JOURNAL_BOOL_INPUT, JOURNAL_BOOL_OUTPUT, JOURNAL_BOOL_MEMORY) + * @param index Buffer array index (0 to buffer_size-1) + * @param bit Bit index within the byte (0-7) + * @param value Boolean value to write + * @return 0 on success, -1 on failure + */ +int journal_write_bool(journal_buffer_type_t type, uint16_t index, + uint8_t bit, bool value); + +/** + * @brief Write a byte value to the journal + * + * @param type Buffer type (JOURNAL_BYTE_INPUT, JOURNAL_BYTE_OUTPUT) + * @param index Buffer array index (0 to buffer_size-1) + * @param value Byte value to write + * @return 0 on success, -1 on failure + */ +int journal_write_byte(journal_buffer_type_t type, uint16_t index, + uint8_t value); + +/** + * @brief Write a 16-bit integer value to the journal + * + * @param type Buffer type (JOURNAL_INT_INPUT, JOURNAL_INT_OUTPUT, JOURNAL_INT_MEMORY) + * @param index Buffer array index (0 to buffer_size-1) + * @param value 16-bit value to write + * @return 0 on success, -1 on failure + */ +int journal_write_int(journal_buffer_type_t type, uint16_t index, + uint16_t value); + +/** + * @brief Write a 32-bit integer value to the journal + * + * @param type Buffer type (JOURNAL_DINT_INPUT, JOURNAL_DINT_OUTPUT, JOURNAL_DINT_MEMORY) + * @param index Buffer array index (0 to buffer_size-1) + * @param value 32-bit value to write + * @return 0 on success, -1 on failure + */ +int journal_write_dint(journal_buffer_type_t type, uint16_t index, + uint32_t value); + +/** + * @brief Write a 64-bit integer value to the journal + * + * @param type Buffer type (JOURNAL_LINT_INPUT, JOURNAL_LINT_OUTPUT, JOURNAL_LINT_MEMORY) + * @param index Buffer array index (0 to buffer_size-1) + * @param value 64-bit value to write + * @return 0 on success, -1 on failure + */ +int journal_write_lint(journal_buffer_type_t type, uint16_t index, + uint64_t value); + +/** + * @brief Apply all pending journal entries to image tables and clear the journal + * + * This function should be called at the start of each PLC scan cycle, + * BEFORE plugin cycle_start hooks are called. + * + * @note The image table mutex (image_mutex) MUST be held by the caller. + * This function acquires the journal mutex internally. + */ +void journal_apply_and_clear(void); + +/** + * @brief Get the number of pending journal entries + * + * Useful for diagnostics and monitoring. + * + * @return Number of entries waiting to be applied + */ +size_t journal_pending_count(void); + +/** + * @brief Get the current sequence number + * + * Useful for diagnostics. The sequence number increments with each write + * and resets to 0 when the journal is cleared. + * + * @return Current sequence number + */ +uint32_t journal_get_sequence(void); + +#ifdef __cplusplus +} +#endif + +#endif /* JOURNAL_BUFFER_H */ diff --git a/core/src/plc_app/plc_state_manager.c b/core/src/plc_app/plc_state_manager.c index c9f84be7..678eb586 100644 --- a/core/src/plc_app/plc_state_manager.c +++ b/core/src/plc_app/plc_state_manager.c @@ -3,6 +3,7 @@ #include "../drivers/plugin_driver.h" #include "image_tables.h" +#include "journal_buffer.h" #include "plc_state_manager.h" #include "scan_cycle_manager.h" #include "utils/log.h" @@ -36,6 +37,31 @@ void *plc_cycle_thread(void *arg) image_tables_fill_null_pointers(); plugin_mutex_give(&plugin_driver->buffer_mutex); + // Initialize journal buffer for race-condition-free plugin writes + journal_buffer_ptrs_t journal_ptrs = { + .bool_input = bool_input, + .bool_output = bool_output, + .bool_memory = bool_memory, + .byte_input = byte_input, + .byte_output = byte_output, + .int_input = int_input, + .int_output = int_output, + .int_memory = int_memory, + .dint_input = dint_input, + .dint_output = dint_output, + .dint_memory = dint_memory, + .lint_input = lint_input, + .lint_output = lint_output, + .lint_memory = lint_memory, + .buffer_size = BUFFER_SIZE, + .image_mutex = &plugin_driver->buffer_mutex + }; + if (journal_init(&journal_ptrs) != 0) { + log_error("Failed to initialize journal buffer"); + } else { + log_info("Journal buffer initialized"); + } + log_info("Starting main loop"); pthread_mutex_lock(&state_mutex); @@ -53,6 +79,10 @@ void *plc_cycle_thread(void *arg) scan_cycle_time_start(); plugin_mutex_take(&plugin_driver->buffer_mutex); + // Apply pending journal entries before plugin hooks run + // This ensures all plugin writes from the previous cycle are visible + journal_apply_and_clear(); + // Call cycle_start for all active native plugins that registered the hook plugin_driver_cycle_start(plugin_driver); @@ -160,6 +190,10 @@ int unload_plc_program(PluginManager *pm) // Wait for the PLC thread to finish pthread_join(plc_thread, NULL); + // Cleanup journal buffer before clearing image tables + journal_cleanup(); + log_info("Journal buffer cleaned up"); + // Clear temporary pointers from image tables before unloading // This ensures clean state for the next program load plugin_mutex_take(&plugin_driver->buffer_mutex); diff --git a/docs/JOURNAL_BUFFER_ARCHITECTURE.md b/docs/JOURNAL_BUFFER_ARCHITECTURE.md new file mode 100644 index 00000000..9f02af85 --- /dev/null +++ b/docs/JOURNAL_BUFFER_ARCHITECTURE.md @@ -0,0 +1,596 @@ +# Journal Buffer Architecture + +## Overview + +This document describes the architecture for a journaled buffer system to handle plugin writes to OpenPLC image tables. The journal buffer provides a race-condition-free mechanism for multiple plugins (both native and Python) to write to shared I/O buffers with deterministic conflict resolution. + +## Problem Statement + +### Current Architecture Issues + +The current plugin architecture has several limitations when multiple plugins need to write to the same image table locations: + +1. **Race Conditions**: Plugins run in separate threads and can write to buffers at arbitrary times, causing race conditions. + +2. **Zero vs. Uninitialized Ambiguity**: When a plugin copies its buffer to the core image tables, there's no way to distinguish between "intentionally wrote zero" and "never touched this location." + +3. **Asynchronous Plugin Timing**: Python plugins run in their own threads without `cycle_start`/`cycle_end` hooks, so their writes can occur at any point relative to the PLC scan cycle. + +4. **No Conflict Resolution**: When two plugins write to the same location, the outcome depends on thread timing, leading to unpredictable behavior. + +### Example Scenario + +Consider a system with both S7Comm (native) and Modbus Master (Python) plugins: + +``` +Timeline: +|---- PLC Scan Cycle N ----|--- Sleep ---|---- PLC Scan Cycle N+1 ----| + cycle_start cycle_end cycle_start cycle_end + │ │ │ │ + S7→OpenPLC OpenPLC→S7 S7→OpenPLC OpenPLC→S7 + ↑ + Overwrites! + │ + Modbus Master writes + new input value here +``` + +1. **cycle_end N**: OpenPLC inputs (value X) → S7 buffer +2. **Sleep period**: Modbus Master reads remote I/O, writes new value Y to OpenPLC inputs +3. **cycle_start N+1**: S7 buffer (still has old value X) → OpenPLC inputs (overwrites Y with X!) + +## Proposed Solution: Journal Buffer + +### Core Concept + +Instead of plugins writing directly to image tables, all writes go through a journal buffer. The journal is applied atomically at the start of each PLC scan cycle, with "last writer wins" semantics based on write sequence. + +``` +┌─────────────────────────────────────────────────────────────────┐ +│ PLC Scan Cycle │ +├─────────────────────────────────────────────────────────────────┤ +│ │ +│ ┌─────────────┐ ┌─────────────┐ │ +│ │ JOURNAL │ ──── Apply ────────► │ CORE │ │ +│ │ BUFFER │ (cycle_start) │ IMAGE │ │ +│ │ │ │ TABLES │ │ +│ │ seq=1: ... │ │ │ │ +│ │ seq=2: ... │ │ bool_input │ │ +│ │ seq=3: ... │ │ bool_output │ │ +│ │ seq=4: ... │ │ int_input │ │ +│ └──────▲──────┘ │ ... │ │ +│ │ └──────┬──────┘ │ +│ │ │ │ +│ WRITES READS │ +│ (journal) (direct) │ +│ │ │ │ +│ ┌──────┴────────────────────────────────────────▼──────┐ │ +│ │ │ │ +│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐│ │ +│ │ │Plugin A │ │Plugin B │ │Plugin C │ │Plugin D ││ │ +│ │ │(S7Comm) │ │(Modbus) │ │(Python) │ │(Native) ││ │ +│ │ └─────────┘ └─────────┘ └─────────┘ └─────────┘│ │ +│ │ │ │ +│ └───────────────────────────────────────────────────────┘ │ +└─────────────────────────────────────────────────────────────────┘ +``` + +### Key Design Principles + +1. **Writes Go to Journal**: All plugin writes are recorded in a journal buffer with a sequence number. + +2. **Reads Are Direct**: Plugins read directly from image tables, getting the most recent applied values. + +3. **Atomic Application**: The entire journal is applied at `cycle_start`, before plugin hooks run. + +4. **Last Writer Wins**: Entries are applied in sequence order. If multiple plugins write to the same location, the one with the highest sequence number wins. + +5. **Sequence Reset Per Cycle**: The sequence counter resets to zero when the journal is cleared after application. + +## Detailed Architecture + +### Journal Entry Structure + +```c +typedef struct { + uint32_t sequence; /* Auto-increment, determines apply order */ + uint8_t buffer_type; /* journal_buffer_type_t enum */ + uint8_t bit_index; /* For bool types: 0-7, for others: 0xFF */ + uint16_t index; /* Buffer array index */ + uint64_t value; /* Value to write (sized for largest type) */ +} journal_entry_t; +``` + +**Size**: 16 bytes per entry (with padding for alignment, may be 20 bytes) + +### Buffer Type Enumeration + +```c +typedef enum { + JOURNAL_BOOL_INPUT = 0, + JOURNAL_BOOL_OUTPUT, + JOURNAL_BOOL_MEMORY, + JOURNAL_BYTE_INPUT, + JOURNAL_BYTE_OUTPUT, + JOURNAL_INT_INPUT, + JOURNAL_INT_OUTPUT, + JOURNAL_INT_MEMORY, + JOURNAL_DINT_INPUT, + JOURNAL_DINT_OUTPUT, + JOURNAL_DINT_MEMORY, + JOURNAL_LINT_INPUT, + JOURNAL_LINT_OUTPUT, + JOURNAL_LINT_MEMORY, + JOURNAL_TYPE_COUNT +} journal_buffer_type_t; +``` + +### Static Journal Buffer + +```c +#define JOURNAL_MAX_ENTRIES 1024 + +static journal_entry_t g_entries[JOURNAL_MAX_ENTRIES]; +static size_t g_count = 0; +static uint32_t g_next_sequence = 0; +static pthread_mutex_t g_journal_mutex = PTHREAD_MUTEX_INITIALIZER; +``` + +**Memory Usage**: ~20KB for 1024 entries (fixed, no dynamic allocation) + +### Thread Safety Model + +Two mutexes are involved: + +1. **Journal Mutex** (`g_journal_mutex`): Protects journal buffer state (entries, count, sequence) +2. **Image Table Mutex** (`image_mutex`): Protects core image tables (existing runtime mutex) + +**Lock Ordering** (to prevent deadlock): Always acquire `image_mutex` before `journal_mutex` when both are needed. + +### Write Flow + +``` +Plugin calls journal_write_*() + │ + ▼ +┌─────────────────────────┐ +│ Acquire journal_mutex │ +└───────────┬─────────────┘ + │ + ▼ + ┌───────────────┐ + │ Journal Full? │ + └───────┬───────┘ + │ + Yes │ No + ┌───────┴───────┐ + │ │ + ▼ │ +┌─────────────┐ │ +│ Emergency │ │ +│ Flush │ │ +│ (see below) │ │ +└──────┬──────┘ │ + │ │ + └─────┬──────┘ + │ + ▼ +┌─────────────────────────┐ +│ Add entry with │ +│ next sequence number │ +└───────────┬─────────────┘ + │ + ▼ +┌─────────────────────────┐ +│ Release journal_mutex │ +└─────────────────────────┘ +``` + +### Apply Flow (at cycle_start) + +``` +PLC Cycle Manager calls journal_apply_and_clear() +(image_mutex already held) + │ + ▼ +┌─────────────────────────┐ +│ Acquire journal_mutex │ +└───────────┬─────────────┘ + │ + ▼ +┌─────────────────────────┐ +│ For each entry in order:│ +│ Apply to image table │ +└───────────┬─────────────┘ + │ + ▼ +┌─────────────────────────┐ +│ Reset count = 0 │ +│ Reset sequence = 0 │ +└───────────┬─────────────┘ + │ + ▼ +┌─────────────────────────┐ +│ Release journal_mutex │ +└─────────────────────────┘ +``` + +### Emergency Flush + +When the journal buffer is full and a new write is attempted: + +``` +(Already holding journal_mutex) + │ + ▼ +┌─────────────────────────┐ +│ Release journal_mutex │ ← Prevent deadlock +└───────────┬─────────────┘ + │ + ▼ +┌─────────────────────────┐ +│ Acquire image_mutex │ ← Lock ordering +└───────────┬─────────────┘ + │ + ▼ +┌─────────────────────────┐ +│ Acquire journal_mutex │ +└───────────┬─────────────┘ + │ + ▼ +┌─────────────────────────┐ +│ Apply all entries │ +│ Clear journal │ +└───────────┬─────────────┘ + │ + ▼ +┌─────────────────────────┐ +│ Release image_mutex │ +└───────────┬─────────────┘ + │ + ▼ +(Continue with write, still holding journal_mutex) +``` + +## Public API + +### C API (for native plugins) + +```c +/* Initialize the journal buffer system */ +int journal_init(const journal_buffer_ptrs_t* buffer_ptrs); + +/* Cleanup journal buffer resources */ +void journal_cleanup(void); + +/* Write functions */ +int journal_write_bool(journal_buffer_type_t type, uint16_t index, + uint8_t bit, bool value); +int journal_write_byte(journal_buffer_type_t type, uint16_t index, + uint8_t value); +int journal_write_int(journal_buffer_type_t type, uint16_t index, + uint16_t value); +int journal_write_dint(journal_buffer_type_t type, uint16_t index, + uint32_t value); +int journal_write_lint(journal_buffer_type_t type, uint16_t index, + uint64_t value); + +/* Apply pending writes (called at cycle_start, image_mutex must be held) */ +void journal_apply_and_clear(void); + +/* Diagnostics */ +size_t journal_pending_count(void); +bool journal_is_initialized(void); +``` + +### Python API (for Python plugins) + +The existing `SafeBufferAccess` class will be extended to route writes through the journal: + +```python +class SafeBufferAccess: + def write_bool(self, buf_type: int, index: int, bit: int, value: bool) -> bool: + """Write boolean value to journal buffer.""" + return self._journal_write_bool(buf_type, index, bit, value) + + def write_int(self, buf_type: int, index: int, value: int) -> bool: + """Write 16-bit integer value to journal buffer.""" + return self._journal_write_int(buf_type, index, value) + + # ... similar for other types + + def read_bool(self, buf_type: int, index: int, bit: int) -> Optional[bool]: + """Read boolean value directly from image table (unchanged).""" + return self._direct_read_bool(buf_type, index, bit) +``` + +## Integration Points + +### 1. PLC State Manager Integration + +In `plc_state_manager.c`, add journal application at cycle start: + +```c +void plc_cycle_start(void) { + /* Acquire image table mutex */ + pthread_mutex_lock(&image_mutex); + + /* Apply journal entries FIRST, before any plugin hooks */ + journal_apply_and_clear(); + + /* Call plugin cycle_start hooks */ + plugin_driver_cycle_start(); + + /* ... rest of cycle start logic */ +} +``` + +### 2. Runtime Initialization + +In runtime initialization, set up journal buffer pointers: + +```c +int runtime_init(void) { + /* ... existing init code ... */ + + /* Initialize journal buffer */ + journal_buffer_ptrs_t ptrs = { + .bool_input = bool_input, + .bool_output = bool_output, + .bool_memory = bool_memory, + .byte_input = byte_input, + .byte_output = byte_output, + .int_input = int_input, + .int_output = int_output, + .int_memory = int_memory, + .dint_input = dint_input, + .dint_output = dint_output, + .dint_memory = dint_memory, + .lint_input = lint_input, + .lint_output = lint_output, + .lint_memory = lint_memory, + .buffer_size = BUFFER_SIZE, + .image_mutex = &bufferLock /* existing buffer mutex */ + }; + + if (journal_init(&ptrs) != 0) { + log_error("Failed to initialize journal buffer"); + return -1; + } + + /* ... rest of init ... */ +} +``` + +### 3. Native Plugin Migration + +Native plugins using direct buffer access need to migrate to journal writes: + +**Before (direct write):** +```c +void sync_to_openplc(void) { + IEC_BOOL** arr = runtime_args->bool_output; + if (arr[index][bit] != NULL) { + *arr[index][bit] = value; /* Direct write */ + } +} +``` + +**After (journal write):** +```c +void sync_to_openplc(void) { + journal_write_bool(JOURNAL_BOOL_OUTPUT, index, bit, value); +} +``` + +### 4. Python Plugin Migration + +Python plugins using `SafeBufferAccess` will automatically use the journal when the `SafeBufferAccess` implementation is updated. No changes required in plugin code. + +## Conflict Resolution Examples + +### Example 1: Two Plugins Write Same Location + +``` +Time 0ms: Plugin A writes %QX0.0 = TRUE → seq=1 +Time 5ms: Plugin B writes %QX0.0 = FALSE → seq=2 +Time 10ms: cycle_start - apply journal + +Apply order: + 1. seq=1: %QX0.0 = TRUE + 2. seq=2: %QX0.0 = FALSE ← This wins (last writer) + +Final value: %QX0.0 = FALSE +``` + +### Example 2: Modbus and S7 Conflict + +``` +Time 0ms: S7 client writes %IX0.0 = TRUE → seq=1 +Time 50ms: Modbus Master reads remote I/O, writes %IX0.0 = FALSE → seq=2 +Time 100ms: cycle_start - apply journal + +Result: %IX0.0 = FALSE (Modbus Master wins because it wrote later) +``` + +### Example 3: Emergency Flush + +``` +Time 0ms: Writes 1-1023 accumulate in journal +Time 50ms: Write 1024 triggers emergency flush + - Apply seq 1-1023 to image tables + - Clear journal + - Add write 1024 as seq=0 +Time 100ms: Writes 1-500 accumulate +Time 200ms: cycle_start - apply journal (seq 0-500) +``` + +## Performance Considerations + +### Memory Usage + +- **Journal Buffer**: 1024 entries × ~20 bytes = ~20KB (static allocation) +- **No per-plugin buffers needed**: Single shared journal +- **Total overhead**: Minimal compared to current architecture + +### Latency + +- **Write latency**: Acquiring journal mutex (~microseconds) +- **Apply latency**: Iterating 1024 entries maximum (~milliseconds) +- **Read latency**: Unchanged (direct buffer access) + +### Throughput + +- **Maximum writes per cycle**: 1024 (configurable via `JOURNAL_MAX_ENTRIES`) +- **Emergency flush**: Handles overflow gracefully without data loss + +## Implementation Phases + +### Phase 1: Core Journal Buffer (C Implementation) + +**Files to create:** +- `core/src/plc_app/journal_buffer.h` - Public API header +- `core/src/plc_app/journal_buffer.c` - Implementation + +**Tasks:** +1. Implement journal entry structure and static buffer +2. Implement write functions with mutex protection +3. Implement `journal_apply_and_clear()` function +4. Implement emergency flush logic +5. Add unit tests for journal operations + +**Estimated effort**: 2-3 days + +### Phase 2: Runtime Integration + +**Files to modify:** +- `core/src/plc_app/plc_state_manager.c` - Add journal apply at cycle_start +- `core/src/plc_app/main.c` or equivalent - Initialize journal at startup + +**Tasks:** +1. Initialize journal buffer during runtime startup +2. Call `journal_apply_and_clear()` at cycle_start +3. Cleanup journal buffer at runtime shutdown +4. Integration testing with PLC scan cycle + +**Estimated effort**: 1-2 days + +### Phase 3: Python API Extension + +**Files to modify:** +- `core/src/drivers/plugins/python/shared/safe_buffer_access.py` - Route writes to journal +- `core/src/drivers/plugins/python/shared/__init__.py` - Export new functions + +**Tasks:** +1. Add ctypes bindings for journal write functions +2. Modify `SafeBufferAccess` write methods to use journal +3. Keep read methods unchanged (direct buffer access) +4. Update documentation + +**Estimated effort**: 1-2 days + +### Phase 4: Native Plugin Migration + +**Files to modify:** +- `core/src/drivers/plugins/native/s7comm/s7comm_plugin.cpp` - Use journal writes +- Any other native plugins with direct buffer writes + +**Tasks:** +1. Replace direct buffer writes with `journal_write_*()` calls +2. Remove plugin-specific double-buffering logic (no longer needed) +3. Simplify `cycle_start`/`cycle_end` hooks +4. Integration testing + +**Estimated effort**: 1-2 days per plugin + +### Phase 5: Python Plugin Migration + +**Files to modify:** +- `core/src/drivers/plugins/python/modbus_master/modbus_master_plugin.py` +- `core/src/drivers/plugins/python/modbus_slave/simple_modbus.py` +- Any other Python plugins + +**Tasks:** +1. Verify plugins work with new `SafeBufferAccess` (should be automatic) +2. Integration testing +3. Performance testing + +**Estimated effort**: 1-2 days + +### Phase 6: Testing and Documentation + +**Tasks:** +1. End-to-end testing with multiple plugins +2. Stress testing (high write volume, emergency flush scenarios) +3. Performance benchmarking +4. Update plugin development documentation +5. Update architecture documentation + +**Estimated effort**: 2-3 days + +## Total Implementation Estimate + +| Phase | Effort | +|-------|--------| +| Phase 1: Core Journal Buffer | 2-3 days | +| Phase 2: Runtime Integration | 1-2 days | +| Phase 3: Python API Extension | 1-2 days | +| Phase 4: Native Plugin Migration | 1-2 days per plugin | +| Phase 5: Python Plugin Migration | 1-2 days | +| Phase 6: Testing and Documentation | 2-3 days | +| **Total** | **~10-15 days** | + +## Future Enhancements (Out of Scope) + +The following features are explicitly NOT part of this design but could be added later: + +1. **Write Coalescing**: Merge multiple writes to the same location within a cycle +2. **Read-After-Write Consistency**: Allow plugins to read their own pending writes +3. **Journal Persistence**: Save journal to disk for crash recovery +4. **Journal Replay**: Replay journal for debugging/simulation +5. **Per-Plugin Statistics**: Track write counts per plugin for diagnostics + +## Appendix A: Comparison with Alternative Approaches + +### Alternative 1: Per-Plugin Shadow Buffers + +Each plugin gets its own copy of image tables. Merge at cycle boundaries. + +**Pros**: Complete isolation +**Cons**: High memory usage (N copies), complex merge logic, priority configuration needed + +### Alternative 2: Timestamp-Based Conflict Resolution + +Each write includes a timestamp. Latest timestamp wins. + +**Pros**: Most recent data always wins +**Cons**: Clock synchronization issues, storage overhead, complexity + +### Alternative 3: Priority-Based System + +Each plugin has a priority. Higher priority wins conflicts. + +**Pros**: Deterministic, configurable +**Cons**: Requires priority configuration, may not reflect actual "freshness" of data + +### Why Journal Buffer? + +The journal buffer approach was chosen because: + +1. **No configuration needed**: "Last writer wins" is intuitive +2. **Memory efficient**: Single shared buffer +3. **Simple API**: Just replace write calls +4. **Deterministic**: Same sequence = same result +5. **Debuggable**: Can log/inspect journal contents + +## Appendix B: Glossary + +| Term | Definition | +|------|------------| +| **Image Table** | Core OpenPLC buffer arrays (bool_input, int_output, etc.) | +| **Journal Entry** | Single write record with sequence, type, index, and value | +| **Journal Buffer** | Static array holding pending write entries | +| **Sequence Number** | Auto-incrementing counter determining apply order | +| **Emergency Flush** | Immediate journal application when buffer is full | +| **Apply** | Process of writing journal entries to image tables | +| **Cycle Start** | Beginning of PLC scan cycle, when journal is applied | diff --git a/install.sh b/install.sh index 918cfcf8..e2c3b043 100755 --- a/install.sh +++ b/install.sh @@ -186,8 +186,7 @@ install_deps_apt() { gcc \ make \ cmake \ - pkg-config \ - && rm -rf /var/lib/apt/lists/* + pkg-config } # For yum-based distros (RHEL 7, CentOS 7, Amazon Linux)