Skip to content

Commit 7d51ac9

Browse files
committed
keepalived: T9256: keep incomplete FIFO lines between reads
pipe_wait() reads at most 500 bytes per iteration and keeps no residual buffer between them. A VRRP transition involving enough instances emits more than that in a single burst, so a read lands mid-line: the tail of one chunk is queued as an incomplete fragment and the head of the next chunk as another. The dispatcher then fails to match the notify regex on both halves and silently runs no transition script. Observed on a pair with 14 instances in one sync group, where "GROUP" arrived as "ROUP" because the leading character ended the previous read. Hold the incomplete trailing line and prepend it to the next read, so only whole lines reach the queue. Only signal the processing thread when at least one complete line was queued.
1 parent b7e9483 commit 7d51ac9

1 file changed

Lines changed: 17 additions & 4 deletions

File tree

src/system/keepalived-fifo.py

Lines changed: 17 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -150,18 +150,31 @@ def pipe_process(self):
150150
def pipe_wait(self):
151151
logger.debug('Message reading start')
152152
self.pipe_read = os.open(self.pipe_path, os.O_RDONLY | os.O_NONBLOCK)
153+
# keepalived may write more than we read in one call, and a read can
154+
# land in the middle of a line. Hold the incomplete trailing line here
155+
# and prepend it to the next read, so only whole lines are queued.
156+
buffer = ''
153157
while self.stopme.is_set() is False:
154158
# sleep a bit to not produce 100% CPU load
155159
time.sleep(0.250)
156160
try:
157161
# try to read a message from PIPE
158162
message = os.read(self.pipe_read, 500)
159163
if message:
160-
# split PIPE content by lines and put them into queue
161-
for line in message.decode().strip().splitlines():
162-
self.message_queue.put(line)
164+
buffer += message.decode()
165+
# split PIPE content by lines and put them into queue,
166+
# keeping the last (possibly incomplete) line for later
167+
lines = buffer.split('\n')
168+
buffer = lines.pop()
169+
queued = False
170+
for line in lines:
171+
line = line.strip()
172+
if line:
173+
self.message_queue.put(line)
174+
queued = True
163175
# set new message flag to start processing
164-
self.message_event.set()
176+
if queued:
177+
self.message_event.set()
165178
except Exception as err:
166179
# ignore the "Resource temporarily unavailable" error
167180
if err.errno != 11:

0 commit comments

Comments
 (0)