Skip to content

Add a que for multimessages #64

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
RaulSMS merged 9 commits into from
Aug 28, 2026
Merged

Add a que for multimessages #64

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
fe32450
Add a que for multimessages when buffer is already full atm
kellergoech Jul 14, 2026
6fced92
revert order of buffer and sending
kellergoech Jul 14, 2026
1196287
Fix missing variable
kellergoech Jul 14, 2026
8935305
Fix findings from copilot
kellergoech Jul 24, 2026
9180c8e
Merge branch 'master' into patch-3
RaulSMS Aug 12, 2026
1e194db
Merge branch 'master' into patch-3
RaulSMS Aug 19, 2026
ecbdc72
Correct interlock of vuffers and return true when package was put in …
kellergoech Aug 25, 2026
9603d91
Merge branch 'master' into patch-3
kellergoech Aug 25, 2026
0c8bdc8
Adjust tests for send que handling
kellergoech Aug 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
130 changes: 77 additions & 53 deletions j1939/j1939_21.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,8 @@ def __init__(self, send_message, job_thread_wakeup, notify_subscribers, max_cmdt
self._rcv_buffer = {}
# Send buffers
self._snd_buffer = {}
# send que
self._snd_que = {}

# List of ControllerApplication
self._cas = []
Expand Down Expand Up @@ -102,6 +104,62 @@ def _buffer_hash(self, src_address, dest_address):
"""
return ((src_address & 0xFF) << 8) | (dest_address & 0xFF)

def _put_multi_msg(self, data, dest_address, src_address,priority, pgn, buffer_hash,pdu_specific):
message_size = len(data)
num_packets = (
int(message_size / 7)
if (message_size % 7 == 0)
else int(message_size / 7) + 1
)

# if the PF is between 240 and 255, the message can only be broadcast
if dest_address == ParameterGroupNumber.Address.GLOBAL:
# send BAM
self.__send_tp_bam(
src_address, priority, pgn.value, message_size, num_packets
)
# init new buffer for this connection
with self._buffer_lock:
self._snd_buffer[buffer_hash] = {
"pgn": pgn.value,
"priority": priority,
"message_size": message_size,
"num_packages": num_packets,
"data": data,
"state": self.SendBufferState.SENDING_BM,
"deadline": time.monotonic() + self._minimum_tp_bam_dt_interval,
"src_address": src_address,
"dest_address": ParameterGroupNumber.Address.GLOBAL,
"next_packet_to_send": 0,
}
else:
# send RTS/CTS
pgn.pdu_specific = 0 # this is 0 for peer-to-peer transfer
self.__send_tp_rts(
src_address,
pdu_specific,
priority,
pgn.value,
message_size,
num_packets,
min(self._max_cmdt_packets, num_packets),
)
# init new buffer for this connection
with self._buffer_lock:
self._snd_buffer[buffer_hash] = {
"pgn": pgn.value,
"priority": priority,
"message_size": message_size,
"num_packages": num_packets,
"data": data,
"state": self.SendBufferState.WAITING_CTS,
"deadline": time.monotonic() + self.Timeout.T3,
"src_address": src_address,
"dest_address": pdu_specific,
"next_packet_to_send": 0,
"next_wait_on_cts": 0,
}

def send_pgn(self, data_page, pdu_format, pdu_specific, priority, src_address, data, time_limit, frame_format):
pgn = ParameterGroupNumber(data_page, pdu_format, pdu_specific)
if len(data) <= 8:
Expand All @@ -119,59 +177,19 @@ def send_pgn(self, data_page, pdu_format, pdu_specific, priority, src_address, d
# init sequence
# known limitation: only one BAM can be sent in parallel to a destination node
buffer_hash = self._buffer_hash(src_address, dest_address)
message_size = len(data)
num_packets = int(message_size / 7) if (message_size % 7 == 0) else int(message_size / 7) + 1

# if the PF is between 240 and 255, the message can only be broadcast
if dest_address == ParameterGroupNumber.Address.GLOBAL:
# send BAM before acquiring the lock — CAN I/O must not be
# held under _buffer_lock to avoid priority inversion with the
# protocol thread.
with self._buffer_lock:
if buffer_hash in self._snd_buffer:
# There is already a sequence active for this pair
return False
self.__send_tp_bam(src_address, priority, pgn.value, message_size, num_packets)

# init new buffer for this connection
with self._buffer_lock:
self._snd_buffer[buffer_hash] = {
"pgn": pgn.value,
"priority": priority,
"message_size": message_size,
"num_packages": num_packets,
"data": data,
"state": self.SendBufferState.SENDING_BM,
"deadline": time.monotonic() + self._minimum_tp_bam_dt_interval,
'src_address' : src_address,
'dest_address' : ParameterGroupNumber.Address.GLOBAL,
'next_packet_to_send' : 0,
}
else:
# send RTS/CTS
pgn.pdu_specific = 0 # this is 0 for peer-to-peer transfer
with self._buffer_lock:
if buffer_hash in self._snd_buffer:
# There is already a sequence active for this pair
return False
self.__send_tp_rts(src_address, pdu_specific, priority, pgn.value, message_size, num_packets, min(self._max_cmdt_packets, num_packets))

# init new buffer for this connection
with self._buffer_lock:
self._snd_buffer[buffer_hash] = {
"pgn": pgn.value,
"priority": priority,
"message_size": message_size,
"num_packages": num_packets,
"data": data,
"state": self.SendBufferState.WAITING_CTS,
"deadline": time.monotonic() + self.Timeout.T3,
'src_address' : src_address,
'dest_address' : pdu_specific,
'next_packet_to_send' : 0,
'next_wait_on_cts': 0,
}

with self._buffer_lock:
if buffer_hash in self._snd_buffer:
# There is already a sequence active for this pair
# put in que
self._snd_que[buffer_hash] = {'buffer_hash': buffer_hash,
'pdu_specific': pdu_specific,
'priority': priority,
'src_address':src_address,
'pgn': pgn,
'data': data,
'dest_address': dest_address}
return True
self._put_multi_msg(data, dest_address, src_address, priority, pgn, buffer_hash, pdu_specific)
self.__job_thread_wakeup()

return True
Expand All @@ -181,6 +199,12 @@ def async_job_thread(self, now):

next_wakeup = now + 5.0 # wakeup in 5 seconds


# get from que if buffer is empty
if not self._snd_buffer and self._snd_que:
key = next(iter(self._snd_que))
self._put_multi_msg(**self._snd_que.pop(key))

with self._buffer_lock:
# check receive buffers for timeout
for bufid in list(self._rcv_buffer):
Expand Down
12 changes: 7 additions & 5 deletions test/test_threading.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -520,7 +520,8 @@ def spam_send_pgn():
def test_send_pgn_j1939_21_buffer_lock_no_race():
"""send_pgn check-then-write on _snd_buffer must be atomic: two threads
sending to the same src/dst pair must not both succeed and overwrite each
other's buffer entry."""
other's buffer entry.
with the new send que one of the threads will putthe package to the que and the message will be send later"""
results = []
errors = []

Expand Down Expand Up @@ -550,10 +551,11 @@ def send_once():
ecu.stop()

assert not errors, f"Exceptions: {errors}"
# Exactly one should succeed (True) and one should be rejected (False)
# because both target the same src/dst hash.
assert sorted(results) == [False, True], (
f"Expected one success and one rejection, got: {results}"
# both should succeed (True)
# because both target the same src/dst hash and one of the
# messages will be set to the que and send later.
assert sorted(results) == [True, True], (
f"Expected two times success, got: {results}"
)


Expand Down