Skip to content

Commit aeef8bb

Browse files
HotNoobHotNoob
authored
Merge pull request #94 from jaredmauch/main
2 parents 513bb05 + d6caf06 commit aeef8bb

6 files changed

Lines changed: 539 additions & 10 deletions

File tree

‎MULTIPROCESSING.md‎

Lines changed: 166 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,166 @@
1+
# Multiprocessing Support
2+
3+
## Overview
4+
5+
The Python Protocol Gateway now supports **automatic multiprocessing** when multiple transports are configured. This provides true concurrency and complete isolation between transports, solving the "transport busy" issues that can occur with single-threaded operation.
6+
7+
## How It Works
8+
9+
### Automatic Detection
10+
- **Single Transport**: Uses the original single-threaded approach
11+
- **Multiple Transports**: Automatically switches to multiprocessing mode
12+
13+
### Process Isolation
14+
Each transport runs in its own separate process, providing:
15+
- **Complete isolation** of resources and state
16+
- **True concurrent operation** - no waiting for other transports
17+
- **Independent error handling** - one transport failure doesn't affect others
18+
- **Automatic restart** of failed processes
19+
20+
### Transport Types
21+
22+
#### Input Transports (read_interval > 0)
23+
- Modbus RTU, TCP, etc.
24+
- Actively read data from devices
25+
- Send data to output transports via bridging
26+
27+
#### Output Transports (read_interval <= 0)
28+
- InfluxDB, MQTT, etc.
29+
- Receive data from input transports via bridging
30+
- Process and forward data to external systems
31+
32+
## Configuration Example
33+
34+
```ini
35+
[transport.0]
36+
transport = modbus_rtu
37+
protocol_version = eg4_v58
38+
address = 1
39+
port = /dev/ttyUSB0
40+
baudrate = 19200
41+
bridge = influxdb_output
42+
read_interval = 10
43+
44+
[transport.1]
45+
transport = modbus_rtu
46+
protocol_version = eg4_v58
47+
address = 1
48+
port = /dev/ttyUSB1
49+
baudrate = 19200
50+
bridge = influxdb_output
51+
read_interval = 10
52+
53+
[influxdb_output]
54+
transport = influxdb_out
55+
host = influxdb.example.com
56+
port = 8086
57+
database = solar
58+
measurement = eg4_data
59+
```
60+
61+
## Inter-Process Communication
62+
63+
### Bridging
64+
- Uses `multiprocessing.Queue` for communication
65+
- Automatic message routing between processes
66+
- Non-blocking communication
67+
- Source transport information preserved
68+
69+
### Message Format
70+
```python
71+
{
72+
'source_transport': 'transport.0',
73+
'target_transport': 'influxdb_output',
74+
'data': {...},
75+
'source_transport_info': {
76+
'transport_name': 'transport.0',
77+
'device_identifier': '...',
78+
'device_manufacturer': '...',
79+
'device_model': '...',
80+
'device_serial_number': '...'
81+
}
82+
}
83+
```
84+
85+
## Benefits
86+
87+
### Performance
88+
- **True concurrency** - no serialization delays
89+
- **Independent timing** - each transport runs at its own interval
90+
- **No resource contention** - each process has isolated resources
91+
92+
### Reliability
93+
- **Process isolation** - one transport failure doesn't affect others
94+
- **Automatic restart** - failed processes are automatically restarted
95+
- **Independent error handling** - each process handles its own errors
96+
97+
### Scalability
98+
- **Linear scaling** - performance scales with number of CPU cores
99+
- **Resource efficiency** - only uses multiprocessing when needed
100+
- **Memory isolation** - each process has its own memory space
101+
102+
## Troubleshooting
103+
104+
### Common Issues
105+
106+
#### "Register is Empty; transport busy?"
107+
- **Cause**: Shared state between transports in single-threaded mode
108+
- **Solution**: Use multiprocessing mode (automatic with multiple transports)
109+
110+
#### InfluxDB not receiving data
111+
- **Cause**: Output transport not properly configured or started
112+
- **Solution**: Ensure `influxdb_output` section has `transport = influxdb_out`
113+
114+
#### Process restarting frequently
115+
- **Cause**: Transport configuration error or device connection issue
116+
- **Solution**: Check logs for specific error messages
117+
118+
### Debugging
119+
120+
#### Enable Debug Logging
121+
```ini
122+
[general]
123+
log_level = DEBUG
124+
```
125+
126+
#### Monitor Process Status
127+
The gateway logs process creation and status:
128+
```
129+
[2025-06-22 19:30:45] Starting multiprocessing mode with 3 transports
130+
[2025-06-22 19:30:45] Input transports: 2, Output transports: 1
131+
[2025-06-22 19:30:45] Bridging detected - enabling inter-process communication
132+
[2025-06-22 19:30:45] Started process for transport.0 (PID: 12345)
133+
[2025-06-22 19:30:45] Started process for transport.1 (PID: 12346)
134+
[2025-06-22 19:30:45] Started process for influxdb_output (PID: 12347)
135+
```
136+
137+
## Testing
138+
139+
Run the test script to verify multiprocessing functionality:
140+
```bash
141+
python pytests/test_multiprocessing.py
142+
```
143+
144+
This will:
145+
- Load your configuration
146+
- Display transport information
147+
- Run for 30 seconds to verify operation
148+
- Show any errors or issues
149+
150+
## Limitations
151+
152+
1. **Memory Usage**: Each process uses additional memory
153+
2. **Startup Time**: Slight delay when starting multiple processes
154+
3. **Inter-Process Communication**: Bridge messages have small overhead
155+
4. **Debugging**: More complex debugging due to multiple processes
156+
157+
## Migration
158+
159+
No migration required! Existing configurations will automatically benefit from multiprocessing when multiple transports are present.
160+
161+
## Performance Tips
162+
163+
1. **Stagger Read Intervals**: Use different read intervals to avoid resource contention
164+
2. **Optimize Batch Sizes**: Adjust batch sizes for faster individual reads
165+
3. **Monitor Logs**: Watch for process restarts indicating issues
166+
4. **Resource Limits**: Ensure sufficient system resources for multiple processes

‎classes/protocol_settings.py‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -265,7 +265,7 @@ class protocol_settings:
265265
_log : logging.Logger = None
266266

267267

268-
def __init__(self, protocol : str, transport_settings : "SectionProxy" = None, settings_dir : str = "protocols"):
268+
def __init__(self, protocol : str, transport_settings : "SectionProxy" = None, settings_dir : str = "protocols", unique_id : str = None):
269269

270270
#apply log level to logger
271271
self._log_level = getattr(logging, logging.getLevelName(logging.getLogger().getEffectiveLevel()), logging.INFO)
@@ -275,6 +275,7 @@ def __init__(self, protocol : str, transport_settings : "SectionProxy" = None, s
275275
self.protocol = protocol
276276
self.settings_dir = settings_dir
277277
self.transport_settings = transport_settings
278+
self.unique_id = unique_id # Store unique identifier for this instance
278279

279280
#load variable mask
280281
self.variable_mask = []

‎classes/transports/influxdb_out.py‎

Lines changed: 24 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
from configparser import SectionProxy
33
from typing import TextIO
44
import time
5+
import logging
56

67
from defs.common import strtobool
78

@@ -21,6 +22,7 @@ class influxdb_out(transport_base):
2122
include_device_info: bool = True
2223
batch_size: int = 100
2324
batch_timeout: float = 10.0
25+
force_float: bool = True # Force all numeric fields to be floats to avoid InfluxDB type conflicts
2426

2527
client = None
2628
batch_points = []
@@ -37,6 +39,7 @@ def __init__(self, settings: SectionProxy):
3739
self.include_device_info = strtobool(settings.get("include_device_info", fallback=self.include_device_info))
3840
self.batch_size = settings.getint("batch_size", fallback=self.batch_size)
3941
self.batch_timeout = settings.getfloat("batch_timeout", fallback=self.batch_timeout)
42+
self.force_float = strtobool(settings.get("force_float", fallback=self.force_float))
4043

4144
self.write_enabled = True # InfluxDB output is always write-enabled
4245
super().__init__(settings)
@@ -103,14 +106,17 @@ def write_data(self, data: dict[str, str], from_transport: transport_base):
103106
for key, value in data.items():
104107
# Check if we should force float formatting based on protocol settings
105108
should_force_float = False
109+
unit_mod_found = None
106110

107111
# Try to get registry entry from protocol settings to check unit_mod
108112
if hasattr(from_transport, 'protocolSettings') and from_transport.protocolSettings:
109113
# Check both input and holding registries
110114
for registry_type in [Registry_Type.INPUT, Registry_Type.HOLDING]:
111115
registry_map = from_transport.protocolSettings.get_registry_map(registry_type)
112116
for entry in registry_map:
113-
if entry.variable_name == key:
117+
# Match by variable_name (which is lowercase)
118+
if entry.variable_name.lower() == key.lower():
119+
unit_mod_found = entry.unit_mod
114120
# If unit_mod is not 1.0, this value should be treated as float
115121
if entry.unit_mod != 1.0:
116122
should_force_float = True
@@ -124,14 +130,28 @@ def write_data(self, data: dict[str, str], from_transport: transport_base):
124130
# Try to convert to float first
125131
float_val = float(value)
126132

127-
# If it's an integer but should be forced to float, or if it's already a float
128-
if should_force_float or not float_val.is_integer():
133+
# Always use float for InfluxDB to avoid type conflicts
134+
# InfluxDB is strict about field types - once a field is created as integer,
135+
# it must always be integer. Using float avoids this issue.
136+
if self.force_float:
129137
fields[key] = float_val
130138
else:
131-
fields[key] = int(float_val)
139+
# Only use integer if it's actually an integer and we're not forcing floats
140+
if float_val.is_integer():
141+
fields[key] = int(float_val)
142+
else:
143+
fields[key] = float_val
144+
145+
# Log data type conversion for debugging
146+
if self._log.isEnabledFor(logging.DEBUG):
147+
original_type = type(value).__name__
148+
final_type = type(fields[key]).__name__
149+
self._log.debug(f"Field {key}: {value} ({original_type}) -> {fields[key]} ({final_type}) [unit_mod: {unit_mod_found}]")
150+
132151
except (ValueError, TypeError):
133152
# If conversion fails, store as string
134153
fields[key] = str(value)
154+
self._log.debug(f"Field {key}: {value} -> string (conversion failed)")
135155

136156
# Create InfluxDB point
137157
point = {

‎classes/transports/transport_base.py‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -112,7 +112,10 @@ def __init__(self, settings : "SectionProxy") -> None:
112112
#must load after settings
113113
self.protocol_version = settings.get("protocol_version")
114114
if self.protocol_version:
115-
self.protocolSettings = protocol_settings(self.protocol_version, transport_settings=settings)
115+
# Create a unique protocol settings instance for each transport to avoid shared state
116+
unique_id = f"{self.transport_name}_{self.protocol_version}"
117+
self._log.debug(f"Creating protocol settings with unique_id: {unique_id}")
118+
self.protocolSettings = protocol_settings(self.protocol_version, transport_settings=settings, unique_id=unique_id)
116119

117120
if self.protocolSettings:
118121
self.protocol_version = self.protocolSettings.protocol

0 commit comments

Comments
 (0)