Collaboration diagram for athenaEF_tdaq_infra.Infrastructure:

Public Member Functions
	__init__ (self, args)
	__del__ (self)
	register_handlers (self)
	start (self)
	start_webproxy (self)
	check_alive (self)
	copy_histograms (self)
	stop (self)

Public Attributes
	args = args
list	processes = []
	pid = os.getpid()
bool	ready = False
dict	prehandlers = {}

Static Public Attributes
list	sigs

Protected Member Functions
	_handle_quit (self, signum, frame)
	_implant_bomb (self)
	_launch (self, name, cmd)

Protected Attributes
	_handle_quit

Detailed Description

Manage the private TDAQ infrastructure (adapted from HLTMPPy.runner.Infrastructure)

Definition at line 88 of file athenaEF_tdaq_infra.py.

Constructor & Destructor Documentation

◆ init()

athenaEF_tdaq_infra.Infrastructure.__init__	(		self,
			args )

Definition at line 94 of file athenaEF_tdaq_infra.py.

    def __init__(self, args):
        self.args = args
        self.processes = []      # (name, subprocess.Popen)
        self.pid = os.getpid()   # Distinguish mother from children after forking
        self.ready = False       # True once all servers are up
        self.register_handlers()
 

◆ del()

athenaEF_tdaq_infra.Infrastructure.__del__ ( self )

Stop infrastructure in the mother process, in case program exits before stop()

Definition at line 101 of file athenaEF_tdaq_infra.py.

    def __del__(self):
        """Stop infrastructure in the mother process, in case program exits before stop()"""
        if os.getpid() == self.pid:
            self.stop()
 

Member Function Documentation

◆ _handle_quit()

athenaEF_tdaq_infra.Infrastructure._handle_quit	(	self,
		signum,
		frame )

protected

Definition at line 112 of file athenaEF_tdaq_infra.py.

    def _handle_quit(self, signum, frame):
        log.info('Caught signal %d. Cleaning up the infrastructure and exiting', signum)
        self.stop()
        prehandler = self.prehandlers.pop(signum, signal.SIG_DFL)
        signal.signal(signum, prehandler)
        sys.exit(0)
 

◆ _implant_bomb()

athenaEF_tdaq_infra.Infrastructure._implant_bomb ( self )

protected

preexec_fn ensuring infrastructure processes exit when this process dies

Definition at line 119 of file athenaEF_tdaq_infra.py.

    def _implant_bomb(self):
        """preexec_fn ensuring infrastructure processes exit when this process dies"""
        from ctypes import cdll
        PR_SET_PDEATHSIG = 1
        try:
            return lambda: cdll['libc.so.6'].prctl(PR_SET_PDEATHSIG, signal.SIGKILL)
        except Exception:
            log.error('Error setting PR_SET_PDEATHSIG for infrastructure processes. '
                      'Using a dummy function instead')
            return lambda: 1
 

◆ _launch()

athenaEF_tdaq_infra.Infrastructure._launch	(	self,
		name,
		cmd )

protected

Definition at line 130 of file athenaEF_tdaq_infra.py.

    def _launch(self, name, cmd):
        logbase = os.path.join(self.args.log_dir, f'{name}_{self.args.partition}')
        proc = subprocess.Popen(cmd, preexec_fn=self._implant_bomb(),
                                stdout=open(logbase + '.out', 'w'),
                                stderr=open(logbase + '.err', 'w'),
                                close_fds=True)
        log.info('Started %s (pid %d): %s', name, proc.pid, ' '.join(cmd))
        self.processes.append((name, proc))
        return proc
 

◆ check_alive()

athenaEF_tdaq_infra.Infrastructure.check_alive ( self )

Return False if any infrastructure process has exited

Definition at line 191 of file athenaEF_tdaq_infra.py.

    def check_alive(self):
        """Return False if any infrastructure process has exited"""
        for name, proc in self.processes:
            ret = proc.poll()
            if ret is not None:
                log.error('Infrastructure process %s (pid %d) exited with code %s',
                          name, proc.pid, ret)
                return False
        return True
 

◆ copy_histograms()

athenaEF_tdaq_infra.Infrastructure.copy_histograms ( self )

Definition at line 201 of file athenaEF_tdaq_infra.py.

    def copy_histograms(self):
        fname = f'r{self.args.run_number:010d}_{self.args.partition}_{self.args.oh_server}.root'
        log.info('Copying histograms into %s (oh_cp)', fname)
        subprocess.call(['oh_cp', '-p', self.args.partition, '-s', self.args.oh_server,
                         '-n', '.*', '-o', '.*', '-O',
                         '-r', str(self.args.run_number), '-f', fname])
 

◆ register_handlers()

athenaEF_tdaq_infra.Infrastructure.register_handlers ( self )

Definition at line 106 of file athenaEF_tdaq_infra.py.

    def register_handlers(self):
        self.prehandlers = {}
        for s in self.sigs:
            self.prehandlers[s] = signal.getsignal(s)
            signal.signal(s, self._handle_quit)
 

◆ start()

athenaEF_tdaq_infra.Infrastructure.start ( self )

Definition at line 140 of file athenaEF_tdaq_infra.py.

    def start(self):
        from ispy import IPCPartition
 
        partition = self.args.partition
 
        # Private IPC domain: reference file local to this job's directory
        ipc_ref = 'file:' + os.path.join(os.getcwd(), 'ipc_init.ref')
        log.info('Setting TDAQ_IPC_INIT_REF: %s', ipc_ref)
        os.environ['TDAQ_IPC_INIT_REF'] = ipc_ref
 
        log.info('Initializing OH monitoring infrastructure for partition %s', partition)
 
        self._launch('ipc_initial', ['ipc_server'])
        while not IPCPartition('initial').isValid():
            log.info('Waiting until initial partition is available...')
            time.sleep(1)
 
        self._launch('ipc_partition', ['ipc_server', '-p', partition])
        while not IPCPartition(partition).isValid():
            log.info('Waiting until partition %s is available...', partition)
            time.sleep(1)
 
        for server in ['DF', 'RunParams', self.args.oh_server]:
            self._launch(f'is_{server}', ['is_server', '-p', partition, '-n', server])
 
        self.start_webproxy()
        self.ready = True
 

◆ start_webproxy()

athenaEF_tdaq_infra.Infrastructure.start_webproxy ( self )

Start the webproxy REST server

Definition at line 168 of file athenaEF_tdaq_infra.py.

    def start_webproxy(self):
        """Start the webproxy REST server""" 
        port = self.args.webdaq_port
        webproxy = self._launch('webproxy', ['webproxy', '-p', str(port)])
 
        timeout = 60
        for _ in range(timeout):
            if webproxy.poll() is not None:
                log.error('webproxy exited early (code %s); see webproxy_%s.err',
                          webproxy.returncode, self.args.partition)
                self.stop()
                sys.exit(1)
            try:
                with socket.create_connection(('localhost', port), timeout=1):
                    log.info('webproxy listening on http://localhost:%d', port)
                    return
            except OSError:
                time.sleep(1)
 
        log.error('webproxy is not listening on localhost:%d after %d s', port, timeout)
        self.stop()
        sys.exit(1)
 

◆ stop()

athenaEF_tdaq_infra.Infrastructure.stop ( self )

Definition at line 208 of file athenaEF_tdaq_infra.py.

    def stop(self):
        if not self.processes:
            return
 
        if self.ready:   # Nothing was ever published if we did not fully start
            self.copy_histograms()
 
        log.info('Finalizing OH monitoring infrastructure')
        for part in [self.args.partition, 'initial']:
            log.info('Destroying partition: %s', part)
            subprocess.call(['ipc_rm', '-f', '-p', part, '-i', '".*"', '-n', '".*"'],
                            stdout=subprocess.PIPE, stderr=subprocess.PIPE)
 
        for name, proc in self.processes:
            while proc.poll() is None:
                proc.kill()
                time.sleep(0.1)
        self.processes = []
        log.info('Terminated all infrastructure processes')
 
 

Member Data Documentation

◆ _handle_quit

athenaEF_tdaq_infra.Infrastructure._handle_quit

protected

Definition at line 110 of file athenaEF_tdaq_infra.py.

◆ args

athenaEF_tdaq_infra.Infrastructure.args = args

Definition at line 95 of file athenaEF_tdaq_infra.py.

◆ pid

athenaEF_tdaq_infra.Infrastructure.pid = os.getpid()

Definition at line 97 of file athenaEF_tdaq_infra.py.

◆ prehandlers

dict athenaEF_tdaq_infra.Infrastructure.prehandlers = {}

Definition at line 107 of file athenaEF_tdaq_infra.py.

◆ processes

list athenaEF_tdaq_infra.Infrastructure.processes = []

Definition at line 96 of file athenaEF_tdaq_infra.py.

◆ ready

bool athenaEF_tdaq_infra.Infrastructure.ready = False

Definition at line 98 of file athenaEF_tdaq_infra.py.

◆ sigs

list athenaEF_tdaq_infra.Infrastructure.sigs

static

Initial value:

= [signal.SIGFPE, signal.SIGHUP, signal.SIGQUIT, signal.SIGSEGV,

signal.SIGTERM, signal.SIGINT]

Definition at line 91 of file athenaEF_tdaq_infra.py.

The documentation for this class was generated from the following file:

athenaEF_tdaq_infra.py

Public Member Functions

Public Attributes

Static Public Attributes

Protected Member Functions

Protected Attributes

Detailed Description

Constructor & Destructor Documentation

◆ __init__()

◆ __del__()

Member Function Documentation

◆ _handle_quit()

◆ _implant_bomb()

◆ _launch()

◆ check_alive()

◆ copy_histograms()

◆ register_handlers()

◆ start()

◆ start_webproxy()

◆ stop()

Member Data Documentation

◆ _handle_quit

◆ args

◆ pid

◆ prehandlers

◆ processes

◆ ready

◆ sigs

◆ init()

◆ del()