# ebpfcat, A Python-based EBPF generator and EtherCAT master
# Copyright (C) 2021 Martin Teichmann <martin.teichmann@xfel.eu>
# Copyright (C) 2026 European XFEL GmbH
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
"""\
:mod:`!ebpfcat.ebpfcat` --- The high-level API for EtherCAT loops
=================================================================
"""
import asyncio
import os
import shutil
import tempfile
from asyncio import (
CancelledError, TimeoutError, ensure_future, gather, get_event_loop, sleep,
wait_for)
from collections import defaultdict
from contextlib import AsyncExitStack, asynccontextmanager, contextmanager
from enum import Enum
from random import randrange
from struct import calcsize, pack, unpack_from
from time import monotonic
from .arraymap import ArrayGlobalVarDesc, ArrayMap
from .bpf import (
MapType, ProgType, create_map, delete_elem, lookup_elem, obj_get, obj_pin,
prog_test_run, update_elem)
from .ebpf import (
EBPFBase, FuncId, MemoryDesc, SubProgram, prandom)
from .ethercat import (
ECCmd, EtherCat, EtherCatError, MachineState, Packet, SyncManager,
Terminal)
from .lock import LockFile, ParallelMailboxLock
from .simulated import ProcessEBPF
from .util import logger
from .xdp import XDP
from .xdp import PacketVar as XDPPacketVar
from .xdp import XDPExitCode
[docs]
class PacketDesc:
"""A single value in a process data
This describes some data in the process data coming from or sent to
a terminal. This is the low-level version of :class:`ProcessDesc`, which
can be used if the terminal's self-desciption is lacking.
:param sm: the sync manager, either :attr:`SyncManager.IN` or
:attr:`SyncManager.OUT`.
:param position: the byte position in the process data
:param size: either a :mod:`python:struct` definition of a data type,
or an integer denoting the bit within a byte to be adressed.
"""
def __init__(self, sm, position, size):
self.sm = sm
self.position = position
self.size = size
def __get__(self, instance, owner):
if instance is None:
return self
offset = instance.position_offset[self.sm]
if isinstance(instance, Struct):
terminal = instance.terminal
device = instance.device
else:
terminal = instance
device = None
ret = PacketVar(terminal, self.sm, self.position + offset, self.size)
if device is None:
return ret
else:
return ret.get(device)
[docs]
class ProcessDesc:
"""A process variable as described in the current mapping
This describes a process variable as found in the current
PDO mapping read from the terminal.
:param index: the index of the process variable, usually found
in the terminal's documentation
:param subindex: the subindex, also found in the documentation
:param size: usually the size is taken from the PDO mapping. A
different size as in a :mod:`python:struct` definition may be
given here, or the number of a bit for a bit field.
"""
def __init__(self, index, subindex, size=None):
self.index = index
self.subindex = subindex
self.size = size
def __get__(self, instance, owner):
if instance is None:
return self
index = self.index + instance.position_offset[None]
if isinstance(instance, Struct):
terminal = instance.terminal
device = instance.device
else:
terminal = instance
device = None
sm, offset, size = terminal.pdos[index, self.subindex]
if self.size is not None:
size = self.size
ret = PacketVar(terminal, sm, offset, size)
if device is None:
return ret
else:
return ret.get(device)
class PacketVar(MemoryDesc):
base_register = 9
def fmt(self):
if isinstance(self.size, int):
return "B"
else:
return self.size
def __init__(self, terminal, sm, position, size):
self.terminal = terminal
self.sm = sm
self.position = position
self.size = size
if isinstance(size, int):
self.struct = None
else:
self.struct = '<' + self.size
self.structsize = calcsize(self.struct)
def set(self, device, value):
if device.sync_group.current_data is None:
super().__set__(device, value)
else:
start = self._start(device)
data = device.sync_group.current_data
if self.struct is None:
mask = 1 << self.size
if value:
data[start] |= mask
else:
data[start] &= ~mask
else:
data[start : start+self.structsize] = pack(self.struct, value)
def get(self, device):
if device.sync_group.current_data is None:
return super().__get__(device, None)
else:
start = self._start(device)
data = device.sync_group.current_data
if self.struct is None:
return bool(data[start] & (1 << self.size))
else:
return unpack_from(self.struct, data, start)[0]
def _start(self, device):
return device.sync_group.pdo_assign[self.terminal][self.sm] \
+ self.position
def fmt_addr(self, device):
return ((self.size, 1) if isinstance(self.size, int) else self.size,
self._start(device) + Packet.ETHERNET_HEADER)
[docs]
class TerminalVar:
"""a device variable to be linked to a process variable
Whithin a :class:`Device`, one can refer to process variables that should
later be linked to process variables of a terminal. Within the device, one
can access the process variable generically. Upon instantiation one would
then assign a :class:`ProcessDesc` (or :class:`PacketDesc`) to it to link
the variable to an actual terminal.
For example::
class MyDevice(Device):
the_output = TerminalVar()
def program(self):
self.the_output = 5 # write 5 to whatever variable linked
terminal = MyTerminal()
device = MyDevice()
device.the_output = terminal.output5 # link the_output to output5
"""
def __set__(self, instance, value):
if isinstance(value, PacketVar):
instance.__dict__[self.name] = value
elif isinstance(value, Struct):
instance.__dict__[self.name] = value
value.device = instance
else:
return instance.__dict__[self.name].set(instance, value)
def __get__(self, instance, owner):
if instance is None:
return self
var = instance.__dict__.get(self.name)
if var is None:
return None
elif isinstance(var, Struct):
return var
else:
return instance.__dict__[self.name].get(instance)
def __set_name__(self, owner, name):
self.name = name
[docs]
class DeviceVar(ArrayGlobalVarDesc):
"""A variable in a device for higher-level use
define a variable within a device which the device's user can
access. This is especially important for fast devices, this is the
way data is communicated to and from the EBPF program.
For non-fast devices, this acts like normal Python variables.
:param size: the size of a variable in :mod:`python:struct` letters
:param write: whether the variable will be written to by the user
For example::
class MyDevice(Device):
my_data = DeviceVar()
def program(self):
self.my_data = 7
device = MyDevice()
print(self.my_data) # should print 7 once the program is running
"""
def __init__(self, size="I", write=False):
super().__init__(SyncGroupBase.properties, size)
self.write = write
def __get__(self, instance, owner):
if instance is None:
return self
elif instance.sync_group is None:
return instance, self.name
elif isinstance(instance.sync_group, EBPFBase):
return super().__get__(instance, owner)
else:
return instance.__dict__.get(self.name, 0)
def __set__(self, instance, value):
if isinstance(instance.sync_group, EBPFBase):
super().__set__(instance, value)
else:
instance.__dict__[self.name] = value
[docs]
class Device(SubProgram):
"""A device is a functional unit in an EtherCAT loop
A device aggregates data coming in and going to terminals
to serve a common goal. A terminal may be used by several
devices.
"""
sync_group = None
[docs]
def get_terminals(self):
"""return the terminals used by this device
return a dictionary of terminal vs. a boolean indicating
whether access is read-write.
"""
ret = defaultdict(lambda: False)
for pv in self.__dict__.values():
if isinstance(pv, (PacketVar, Struct)):
ret[pv.terminal] |= pv.sm is SyncManager.OUT
return ret
[docs]
def update(self):
"""overwrite this method for slow devices
This method gets called regularly to actually implement the control
loop, for slow loops. """
[docs]
def program(self):
"""overwrite this method with the EPBF generation code
The EBPF generated by this method is injected into the Linux kernel
to do the actual control loop. It is advisable to only put the part
the code here which actually needs to be fast. Anything else goes
into :meth:`fast_update`.
"""
[docs]
def fast_update(self):
"""overwrite this method for running Python code in fast devices
The parts of the control loop which do not need to run at full speed
go here. This method gets regularly called and is usually doing
housekeeping tasks which do not need to run as EBPF.
Sometimes it is useful to have a device that can run both as fast
and slow device. In this case one might leave :meth:`program` empty,
and let :meth:`update` and :meth:`fast_update` do the same.
"""
[docs]
def initialize(self):
"""overwrite this method with code to initialize upon startup"""
class ServiceDesc:
def __init__(self, index, subidx):
self.index = index
self.subidx = subidx
[docs]
class Struct:
"""Define repetitive structures in CoE objects
Some terminals, especially multi-channel terminals, have repetitive
structures in their CoE. Inherit from this class to create a structure for
them. Each instance will then define one channel. It takes one parameter,
which is the offset in the CoE address space from the template structure to
the one of the channel.
"""
device = None
def __new__(cls, *args, **kwargs):
return StructDesc(cls, *args, **kwargs)
class StructDesc:
def __init__(self, struct, sm3=0, sm2=None, coe=None):
self.struct = struct
if sm2 is None:
sm2 = sm3
if coe is None:
coe = sm3
self.position_offset = {SyncManager.OUT: sm2, SyncManager.IN: sm3,
None: coe}
def __get__(self, instance, owner):
if instance is None:
return self
if (ret := instance.__dict__.get(self.name)) is not None:
return ret
ret = object.__new__(self.struct)
ret.position_offset = self.position_offset
ret.terminal = instance
instance.__dict__[self.name] = ret
return ret
def __set_name__(self, owner, name):
self.name = name
[docs]
class EBPFTerminal(Terminal):
"""This is the base class for all supported terminal types
inheriting classes should define a ``compatibility`` class variable
which is a set of tuples, each of which is a pair of Ethercat vendor and
product id of all supported terminal types.
"""
compatibility = None
position_offset = {SyncManager.OUT: 0, SyncManager.IN: 0, None: 0}
use_fmmu = True
out_pdos = None
in_pdos = None
async def apply_eeprom(self):
await super().apply_eeprom()
if (self.compatibility is not None and
(self.vendorId, self.productCode) not in self.compatibility):
raise EtherCatError(
f"Incompatible Terminal: {self.vendorId}:{self.productCode}")
await self.set_state(MachineState.PRE_OPERATIONAL)
if self.out_pdos is not None:
await self.write_pdos(0x1c12, self.out_pdos)
if self.in_pdos is not None:
await self.write_pdos(0x1c13, self.in_pdos)
# we used to set self.pdos = {}, but that seems no good for safety term
outbits, inbits = await self.parse_pdos()
self.pdo_out_sz = int((outbits + 7) // 8)
assert not self.pdo_out_sz or self.pdo_out_off
self.pdo_in_sz = int((inbits + 7) // 8)
assert not self.pdo_in_sz or self.pdo_in_off
await self.write_pdo_sm()
# going to SAFE_OPERATIONAL checks the correctness of our setup.
# we could do that later, but doing it here we get a better clue
# where the problem could be.
await self.set_state(MachineState.SAFE_OPERATIONAL)
await self.parse_sdos()
async def write_pdos(self, index, values):
await self.sdo_write(pack('B', 0), index, 0)
for i, v in enumerate(values, 1):
await self.sdo_write(pack('<H', v), index, i)
await self.sdo_write(pack('B', len(values)), index, 0)
[docs]
def allocate(self, packet, readwrite):
"""allocate space in packet for the pdos of this terminal
return a dict that contains the datagram number and
starting offset therein for each sync manager.
Negative datagram numbers are for the future FMMU
datagrams."""
bases = {}
if self.use_fmmu:
if self.pdo_in_sz:
bases[SyncManager.IN] = (BaseType.FMMU_IN, packet.fmmu_in_size)
packet.fmmu_in_size += self.pdo_in_sz
packet.fmmu_in_count += 1
if readwrite and self.pdo_out_sz:
bases[SyncManager.OUT] = (BaseType.FMMU_OUT,
packet.fmmu_out_size)
packet.fmmu_out_size += self.pdo_out_sz
packet.fmmu_out_count += 1
else:
if self.pdo_in_sz:
bases[SyncManager.IN] = (BaseType.NO_FMMU, packet.size)
packet.append(ECCmd.FPRD, b"\0" * self.pdo_in_sz, 0,
self.position, self.pdo_in_off)
if readwrite and self.pdo_out_sz:
bases[SyncManager.OUT] = (BaseType.NO_FMMU, packet.size)
packet.append_writer(ECCmd.FPWR, b"\0" * self.pdo_out_sz, 0,
self.position, self.pdo_out_off)
return bases
async def parse_sdos(self):
async def parse_sdo(obj, offset):
for cls in obj.__class__.__mro__:
for k, v in cls.__dict__.items():
if isinstance(v, ServiceDesc):
setattr(obj, k, await self.read_object_entry(
v.index + offset, v.subidx))
elif isinstance(v, StructDesc):
struct = getattr(self, k)
await parse_sdo(struct, struct.position_offset[None])
await parse_sdo(self, 0)
def update(self, data):
pass
[docs]
class EtherXDP(XDP):
"""The EtherCat packet dispatcher
This class creates an EBPF program that receives EtherCAT packet
from the network and dispatches them to the EBPF program of the fast
sync group they belong to, or passes them on to user space if they
do not belong to any fast sync group.
The additional information needed is put into a first, internal
datagram in the EtherCAT packet, marked as no-op. It also contains
an ethertype that should be used once the packet is handed over to
user space, so it can be dispatched to the correct listener.
For each fast sync group, there are always two packets on the wire,
one that only reads value from the terminals, the other one also
writes. Usually only the read-write packet is handed over to the
sync group's program. If, however, that packet gets lost, the next
read-only packet is handed over.
User space is supposed to constantly feed in new packets, and the
then-superfluous packets are sent back to user space. This way user
space can constantly read data independent of the EBPF program. It
cannot write, however, as this would cause priority issues.
"""
license = "GPL"
minimumPacketSize = 30
variables = ArrayMap()
dropcounter = variables.globalVar("I")
counters = variables.globalVar("64I")
rate = 0
INDEX0 = 17
ethertype = XDPPacketVar(12, "!H")
addr0 = XDPPacketVar(18, "I") # indicates the fast sync group number
cmd0 = XDPPacketVar(16, "B") # 0 is a noop, internal datagram
index0 = XDPPacketVar(INDEX0, "B") # the loop counter
data0 = XDPPacketVar(26, "H") # the ethertype to use
def program(self):
self.r8 = self.r1
with prandom(self.ebpf) & 0xffff < self.rate:
self.dropcounter += 1
self.ebpf.exit(XDPExitCode.DROP)
with self.ethertype == 0x88A4, self.cmd0 == 0:
self.r3 = self.addr0 # use r3 for tail_call
with self.counters.get_address(None, False, False) as (dst, _), \
self.r3 < FastEtherCat.MAX_PROGS:
self.r[dst] += 4 * self.r3
self.r4 = self.mB[self.r[dst]]
# we lost a packet
with self.index0 == self.r4 as Else:
self.mI[self.r[dst]] += 1 + (self.r4 & 1)
# normal case: two packets on the wire
with Else, ((self.index0 + 1 & 0xff) == self.r4) \
| (self.index0 == 0) as Else:
self.mI[self.r[dst]] += 1
with self.r4 & 1: # last one was active
self.index0 = self.mB[self.r[dst]]
self.exit(XDPExitCode.TX)
with Else:
self.ethertype = self.data0
self.exit(XDPExitCode.PASS)
self.index0 = self.mB[self.r[dst]]
self.r2 = self.get_fd(self.programs)
self.r7 = self.r9
self.call(FuncId.tail_call)
self.r3 = self.r7
self.r1 = self.r8
with self.packetSize > self.minimumPacketSize:
self.ethertype = self.data0
self.exit(XDPExitCode.PASS)
class SimpleEtherCat(EtherCat):
pass
[docs]
class FastEtherCat(SimpleEtherCat):
"""An EtherCAT driver class for fast and slow sync groups"""
MAX_PROGS = 64
def __init__(self, network):
super().__init__(network)
self.sync_groups = {}
@contextmanager
def register_sync_group(self, sg):
sg.load()
while True:
index = randrange(self.MAX_PROGS)
key = pack("<I", index)
try:
ret = lookup_elem(self.programs, key, '<I')
except KeyError:
break
update_elem(self.programs, key,
pack("<I", sg.file_descriptor))
sg.close()
self.sync_groups[index] = sg
try:
yield index
finally:
delete_elem(self.programs, pack("<I", index))
del self.sync_groups[index]
async def connect(self):
await super().connect()
self.ebpf = EtherXDP()
self.ebpf.programs = self.programs = \
create_map(MapType.PROG_ARRAY, 4, 4, self.MAX_PROGS)
self.ebpf.attach(self.addr[0])
@asynccontextmanager
async def run(self):
await super().connect()
self.ebpf = EtherXDP()
self.ebpf.programs = self.programs
async with self.ebpf.run(self.addr[0]):
try:
yield
finally:
await gather(*[v.cancel_and_wait()
for v in self.sync_groups.values()])
[docs]
class ParallelEtherCat(FastEtherCat):
"""A multi-processing EtherCAT loop
If several programs want to access an EtherCAT loop at the same time, they
need to negotiate where the packets go. This class installs an XDP program
that dispatches the packets to the right consumer. The dispatch is done by
modifying the ethertype of the packet, as this is what we can
:meth:`~socket.socket.bind` to.
The first program connecting to the loop installs the XDP program, the last
one leaving uninstalls it. We put lock files into ``/run/lock`` to
synchronize that, and put a map of XDP programs into ``/sys/fs/bpf``, where
all participants can put their programs.
"""
def get_ethertype(self, lockdir):
while True:
self.ethertype = randrange(0x3001, 0x3100)
try:
lockfile = f'{self.ethertype}.lock'
with open(f'{lockdir}/{lockfile}', 'x') as lf:
lf.write(f'{os.getpid():10}\n')
self.next_logical_addr = (self.ethertype - 0x3000) << 24
return lockfile
except FileExistsError:
continue
def get_mbx_lock(self, no):
return ParallelMailboxLock(self.mbx_lock_file, no)
@asynccontextmanager
async def run(self):
lockdir = f'/run/lock/ebpf.{self.addr[0]}.lock'
programs = f'/sys/fs/bpf/{self.addr[0]}'
os.makedirs(programs, exist_ok=True)
programs += '/programs'
tmpdir = tempfile.mkdtemp(dir='/run/lock')
lockfile = self.get_ethertype(tmpdir)
try:
os.rename(tmpdir, lockdir)
except OSError:
logger.info('other EPBFCat users found, use existing lock files')
shutil.rmtree(tmpdir)
lockfile = self.get_ethertype(lockdir)
try:
await super(FastEtherCat, self).connect()
self.ebpf = EtherXDP()
try:
self.ebpf.programs = self.programs = obj_get(programs)
except FileNotFoundError:
await sleep(0.1)
self.ebpf.programs = self.programs = obj_get(programs)
except Exception:
os.remove(f'{lockdir}/{lockfile}')
raise
else:
logger.info('no other EBPFCat user found, create lock files')
try:
await super(FastEtherCat, self).connect()
self.ebpf = EtherXDP()
self.ebpf.programs = self.programs = \
create_map(MapType.PROG_ARRAY, 4, 4, self.MAX_PROGS)
try:
os.remove(programs)
except OSError:
pass
else:
logger.error('an old programs file was still at %s',
programs)
self.ebpf.attach(self.addr[0])
self.ebpf.close()
obj_pin(programs, self.programs)
await self.clear_fmmus()
except Exception:
shutil.rmtree(lockdir)
raise
self.mbx_lock_file = LockFile(f'/run/ebpf/{self.addr[0]}',
*self.terminal_addr_range)
try:
yield
finally:
self.transport.close()
self.transport = None
await gather(*[v.cancel_and_wait()
for v in self.sync_groups.values()])
for entry in os.scandir(lockdir):
with open(entry.path, 'r') as fin:
pid = int(fin.read())
if not os.path.exists(f'/proc/{pid}'):
logger.warning('lock file for pid %i exists, '
'but no process, removing', pid)
os.remove(entry)
os.remove(f'{lockdir}/{lockfile}')
try:
os.rmdir(lockdir)
except OSError:
logger.info('other EPBFCat users still alive, keep locks')
else:
logger.info('we are last EPBFCat user, remove locks')
self.ebpf.detach(self.addr[0])
os.remove(programs)
self.mbx_lock_file.remove()
def __getstate__(self):
return self.addr[0]
def __setstate__(self, network):
self.__init__(network)
[docs]
class SterilePacket(Packet):
"""a sterile packet has all its sets exchanged by NOPs"""
logical_addr_inc = 0x800
def __init__(self):
super().__init__()
self.on_the_fly = [] # list of sterilized positions
self.fmmu_out_size = self.fmmu_in_size = 0
self.fmmu_out_count = self.fmmu_in_count = 0
self.counters = {}
def append_writer(self, cmd, *args, **kwargs):
start = self.size
self.append(cmd, *args, **kwargs)
self.on_the_fly.append((start, self.size, cmd))
def append(self, cmd, *args, counter=1):
super().append(cmd, *args, wkc=counter)
self.counters[self.size - 2] = counter
def sterile(self, index, ethertype=0x88A4):
ret = bytearray(self.assemble(index, ethertype))
for pos, _, cmd in self.on_the_fly:
ret[pos] = ECCmd.NOP.value
return ret
def append_fmmu(self, logical_in, logical_out):
fmmu_in_pos = self.size
if self.fmmu_in_size:
self.append(ECCmd.LRD, b"\0" * self.fmmu_in_size, 0,
logical_in, counter=self.fmmu_in_count)
fmmu_out_pos = self.size
if self.fmmu_out_size:
self.append_writer(ECCmd.LWR, b"\0" * self.fmmu_out_size, 0,
logical_out, counter=self.fmmu_out_count)
return fmmu_in_pos, fmmu_out_pos
[docs]
def activate(self, ebpf):
"""generate the EBPF program that re-activates a sterile packet"""
with ebpf.ebpf.wkc_errors == 0:
ebpf.ebpf.exit(XDPExitCode.TX)
for start, stop, cmd in self.on_the_fly:
ebpf.pB[start + self.ETHERNET_HEADER] = cmd.value
with ebpf.pH[stop + self.ETHERNET_HEADER - 2] \
!= self.counters[stop - 2]:
ebpf.ebpf.wkc_errors += 1
ebpf.pH[stop + self.ETHERNET_HEADER - 2] = 0
class BaseType(Enum):
NO_FMMU = 0
FMMU_IN = 1
FMMU_OUT = 2
class SyncGroupBase:
missed_counter = 0
cycletime = 0.01 # cycle time of the PLC loop
task = None
current_data = None
logical_in = logical_out = None
name = 'No Name'
properties = ArrayMap()
wkc_errors = properties.globalVar('I')
def __init__(self, ec, devices, **kwargs):
super().__init__(**kwargs)
self.ec = ec
self.devices = devices
self.terminals = {}
for dev in self.devices:
dev.sync_group = self
for t, rw in dev.get_terminals().items():
self.terminals[t] = self.terminals.get(t, False) | rw
async def run(self):
async with AsyncExitStack() as stack:
for terminal, bases in self.fmmu_maps.items():
try:
base = bases.get(SyncManager.OUT)
if base is not None:
await stack.enter_async_context(
terminal.map_fmmu(base, True))
base = bases.get(SyncManager.IN)
if base is not None:
await stack.enter_async_context(
terminal.map_fmmu(base, False))
except Exception as e:
e.add_note(f'while fmmu-mapping {terminal.name}')
raise
await self.inner_loop()
async def clear_safe_errors(self):
"""clear the error why we went to safe operational"""
await sleep(0.2) # assure the watchdog kicks in
await gather(*[t.set_state(MachineState.SAFE_OPERATIONAL,
clear_error=True)
for t, rw in self.terminals.items() if rw])
async def inner_loop(self):
data = self.asm_packet
self.wkc_errors = 0
lasttime = monotonic()
await gather(*[t.set_state(MachineState.SAFE_OPERATIONAL, clear_error=True)
for t, rw in self.terminals.items() if rw])
future = self.ec.roundtrip_packet(data, self.packet_index)
await gather(*[t.set_state(MachineState.OPERATIONAL)
for t, rw in self.terminals.items() if rw])
self.wkc_errors = 1 # write actions are ignored before
while self.running:
try:
data = await wait_for(future, timeout=0.02)
except TimeoutError:
self.missed_counter += 1
logger.warning(
"%s: did not receive Ethercat response in time %i",
self.name, self.missed_counter)
future = self.ec.roundtrip_packet(data,
self.packet_index)
continue
data = self.update_devices(data)
newtime = monotonic()
if newtime - lasttime > self.cycletime:
logger.warning('%s: response time exceeded (%.0f ms)',
self.name, (newtime - lasttime) * 1000)
await sleep(self.cycletime - (newtime - lasttime))
newtime = monotonic()
if newtime - lasttime > 0.05:
logger.warning('%s: excessive cycle time (%.0f ms)',
self.name, (newtime - lasttime) * 1000)
lasttime = newtime
future = self.ec.roundtrip_packet(data, self.packet_index)
def allocate(self):
self.packet = SterilePacket()
terminals = {t: t.allocate(self.packet, rw)
for t, rw in self.terminals.items()}
logical_in = self.ec.get_fmmu_addr()
logical_out = logical_in + self.packet.logical_addr_inc
in_pos, out_pos = self.packet.append_fmmu(logical_in, logical_out)
offsets = {BaseType.FMMU_IN: in_pos, BaseType.FMMU_OUT: out_pos}
self.pdo_assign = {t: {sm: offsets.get(base, 0) + off + Packet.DATAGRAM_HEADER
for sm, (base, off) in d.items()}
for t, d in terminals.items()}
offsets = {BaseType.FMMU_IN: logical_in,
BaseType.FMMU_OUT: logical_out}
self.fmmu_maps = {t: {sm: offsets.get(base, base) + off
for sm, (base, off) in d.items()
if base is not BaseType.NO_FMMU}
for t, d in terminals.items()}
[docs]
class SyncGroup(SyncGroupBase):
"""A group of devices communicating at the same time"""
packet_index = 1000
def update_devices(self, data):
self.current_data[:] = data
for pos, counts in self.packet.counters.items():
if data[pos] != counts:
logger.warning(
'EtherCAT datagram "%s" processed %i times, should be %i',
self.name, data[pos], counts)
self.wkc_errors += 1
self.current_data[pos] = 0
for dev in self.devices:
dev.update()
return self.current_data
def start(self):
self.allocate()
self.packet_index = SyncGroup.packet_index
SyncGroup.packet_index += 1
self.asm_packet = self.packet.assemble(self.packet_index,
self.ec.ethertype)
self.current_data = bytearray(self.asm_packet)
self.running = True
task = ensure_future(self.run())
for dev in self.devices:
dev.initialize()
return task
[docs]
class ProcessSyncGroup(SyncGroup, ProcessEBPF):
"""A :class:`SyncGroup` running in a separate process
In order to lower latency, one may run a sync group in a different
process. All the :meth:`~Device.update` methods are run in that
separate process.
From the calling process, one may read or write :class:`DeviceVar`\\ s,
or read (but not write) :class:`TerminalVar`\\ s.
"""
def __init__(self, ec, devices, **kwargs):
super().__init__(ec, devices, subprograms=devices, **kwargs)
async def subprocess_loop(self):
async with self.ec.run():
self.asm_packet = self.packet.assemble(self.packet_index,
self.ec.ethertype)
for dev in self.devices:
dev.initialize()
await super().inner_loop()
@property
def current_data(self):
return memoryview(self._current_data.get_obj()).cast('B')
def start(self):
assert isinstance(self.ec, ParallelEtherCat)
self.allocate()
self.packet_index = SyncGroup.packet_index
SyncGroup.packet_index += 1
self._current_data = self.ctx.Array('B', max(46, self.packet.size))
task = ensure_future(self.run())
return task
async def inner_loop(self):
await ProcessEBPF.start(self)
[docs]
class FastSyncGroup(SyncGroupBase, XDP):
"""A :class:`SyncGroup` where all devices are EBPF programs"""
license = "GPL"
debug = False
def __init__(self, ec, devices, **kwargs):
super().__init__(ec, devices, subprograms=devices, **kwargs)
def load(self, *args, **kwargs):
if self.debug:
return super().load(*args, log_level=1, **kwargs)
else:
return super().load()
def program(self):
with self.packetSize >= self.packet.size + Packet.ETHERNET_HEADER \
as self.packet_access:
self.packet.activate(self.packet_access)
for dev in self.devices:
dev.program()
self.exit(XDPExitCode.TX)
async def run(self):
with self.ec.register_sync_group(self) as self.packet_index:
for dev in self.devices:
dev.initialize()
self.wkc_errors = 0
self.asm_packet = self.packet.sterile(self.packet_index,
self.ec.ethertype)
# prime the pump: two packets to get things going
try:
await wait_for(self.ec.roundtrip_packet(
self.asm_packet, self.packet_index), timeout=0.1)
raise EtherCatError(f'ebpf not installed? {self.name}')
except TimeoutError:
pass # all fine!
try:
await wait_for(self.ec.roundtrip_packet(
self.asm_packet, self.packet_index), timeout=0.1)
raise EtherCatError(f'ebpf not installed (2nd) ? {self.name}')
except TimeoutError:
pass # all fine!
await super().run()
def update_devices(self, data):
if data[EtherXDP.INDEX0 - Packet.ETHERNET_HEADER] & 1:
self.current_data = data
elif self.current_data is None:
return self.asm_packet
for dev in self.devices:
dev.fast_update()
return self.asm_packet
def start(self):
self.allocate()
self.running = True
self.task = ensure_future(self.run())
return self.task
async def cancel_and_wait(self):
if not self.task.done():
self.task.cancel()
try:
await self.task
except CancelledError:
pass
