athenaEF_tdaq_infra.Infrastructure Class Reference

Collaboration diagram for athenaEF_tdaq_infra.Infrastructure:

Public Member Functions
	__init__ (self, args)
	__del__ (self)
	register_handlers (self)
	start (self)
	start_rdb (self)
	start_webproxy (self)
	check_alive (self)
	copy_histograms (self)
	dump_is_content (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 95 of file athenaEF_tdaq_infra.py.

Constructor & Destructor Documentation

__init__()

athenaEF_tdaq_infra.Infrastructure.__init__	(		self,
			args )

Definition at line 101 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 108 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 119 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 126 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 137 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 217 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 227 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])
 

dump_is_content()

athenaEF_tdaq_infra.Infrastructure.dump_is_content

(

self

)

Definition at line 234 of file athenaEF_tdaq_infra.py.

    def dump_is_content(self):
        fname = f'r{self.args.run_number:010d}_{self.args.partition}_DF.txt'
        log.info('Writing content of DF IS server to %s', fname)
        with open(fname, "w") as f:
            subprocess.call(['is_ls', '-p', self.args.partition, '-n', 'DF',
                             '-R', '.*', '-TNv'],
                            stdout=f, stderr=subprocess.STDOUT, text=True)
 

register_handlers()

athenaEF_tdaq_infra.Infrastructure.register_handlers

(

self

)

Definition at line 113 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 147 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_rdb()
        self.start_webproxy()
        self.ready = True
 

start_rdb()

athenaEF_tdaq_infra.Infrastructure.start_rdb

(

self

)

Start rdb_server serving the IS type schema

Definition at line 176 of file athenaEF_tdaq_infra.py.

    def start_rdb(self):
        """Start rdb_server serving the IS type schema"""
        if not self.args.schema:
            log.info('No --schema given, not starting rdb_server '
                     '(IS publication will be rejected with HTTP 400)')
            return
 
        missing = [f for f in self.args.schema if not os.path.exists(f)]
        if missing:
            log.error('IS schema file(s) not found: %s', ', '.join(missing))
            self.stop()
            sys.exit(1)
 
        log.info('Starting rdb_server with IS schema: %s', ', '.join(self.args.schema))
        self._launch('rdb', ['rdb_server', '-p', self.args.partition,
                             '-d', 'ISRepository', '-s',
                             '-D'] + self.args.schema)
 

start_webproxy()

athenaEF_tdaq_infra.Infrastructure.start_webproxy

(

self

)

Start the webproxy REST server

Definition at line 194 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 242 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()
            self.dump_is_content()
 
        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 117 of file athenaEF_tdaq_infra.py.

args

athenaEF_tdaq_infra.Infrastructure.args = args

Definition at line 102 of file athenaEF_tdaq_infra.py.

pid

athenaEF_tdaq_infra.Infrastructure.pid = os.getpid()

Definition at line 104 of file athenaEF_tdaq_infra.py.

prehandlers

dict athenaEF_tdaq_infra.Infrastructure.prehandlers = {}

Definition at line 114 of file athenaEF_tdaq_infra.py.

processes

list athenaEF_tdaq_infra.Infrastructure.processes = []

Definition at line 103 of file athenaEF_tdaq_infra.py.

ready

bool athenaEF_tdaq_infra.Infrastructure.ready = False

Definition at line 105 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 98 of file athenaEF_tdaq_infra.py.

---

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

• athenaEF_tdaq_infra.py