13457957190

Committed 21 Feb 2025 01:35PM CUT coverage: 78.516%. Remained the same

Build # 13457957190

Build Type

Pull #6312

github

Committed by

Vincent-lau

Commit Message

CA-407033: Call `receive_finalize2` synchronously

`Remote.receive_finalize2` is called at the end of SXM to clean things
up and compose the base and leaf images together. The compose operation
should only be called while the VDI is deactivated. Currently a thread
is created to call `receive_finalize2`, which could caused problems
where the VM itself gets started while the `receive_finalize2`/`VDI.compose`
is still in progress. This is not a safe operation to do.

The fix here is to simply remove the thread and make the whole operation
sequential.

Signed-off-by: Vincent Liu <shuntian.liu2@cloud.com>

Pull Request Pull Request #6312: CA-407033: Call `receive_finalize2` synchronously

Run Details

3512 of 4473 relevant lines covered (78.52%)

0.79 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

64.58

/ocaml/xenopsd/scripts/igmp_query_injector.py

#!/usr/bin/env python3
import argparse
import threading
import logging
import subprocess
from scapy.all import Ether, IP, sendp
from scapy.contrib.igmp import IGMP
from xcp import logger as log
import sys
import os
from xen.lowlevel.xs import xs


# global xenstore handler
g_xs_handler = xs()
VIF_CONNECTED_STATE = '4'


class XSWatcher(object):
    """Tool for watching xenstore
    """
    def __init__(self):
        self.watches = dict()

    def watch(self, path, token):
        self.watches[path] = token
        return g_xs_handler.watch(path, token)

    def unwatch(self, path, token):
        self.watches.pop(path)
        return g_xs_handler.unwatch(path, token)

    def read_watch(self):
        return g_xs_handler.read_watch()


class IGMPQueryInjector(object):
    def __init__(self, max_resp_time, vifs, vif_connected_timeout=0):
        self.max_resp_time = max_resp_time
        self.vifs = vifs
        self.vif_connected_timeout = vif_connected_timeout

    def inject_to_vif(self, vif):
        mac = get_vif_mac(vif)
        try:
            self.inject_packet(vif, mac)
            log.info('Inject IGMP query to vif:%s, mac:%s' % (vif, mac))
        except Exception as e:
            log.error('Inject IGMP query to vif:%s, mac:%s failed' % (vif, mac))
            log.logException(e)

    def inject_packet(self, iface, dst_mac):
        ether_part = Ether(src='00:00:00:00:00:00', dst=dst_mac)
        ip_part = IP(ttl=1, src='0.0.0.0', dst='224.0.0.1')
        igmp_part = IGMP(type=0x11)
        # Should use integer division // in python 3
        igmp_part.mrcode = (self.max_resp_time // 100) & 0xff
        igmp_part.igmpize()
        # Make this IGMP query packet as an unicast packet
        ether_part.dst = dst_mac
        sendp(ether_part / ip_part / igmp_part, iface=iface, verbose=False)

    def inject(self):
        if not self.vifs:
            return

        if self.vif_connected_timeout > 0:
            # should check connection state
            log.info('Inject IGMP query with connection state check')
            self._inject_with_connection_state_check()
        else:
            log.info('Inject IGMP query without connection state check')
            self._inject_without_connection_state_check()

    def check_and_inject(self, watcher):
        vif_connected_set = set()
        while watcher.watches:
            path, vif = watcher.read_watch()
            if g_xs_handler.read('', path) == VIF_CONNECTED_STATE:
                # should ensure both backend and frontend connected before injection
                if vif not in vif_connected_set:
                    vif_connected_set.add(vif)
                else:
                    vif_connected_set.remove(vif)
                    self.inject_to_vif(vif)
                watcher.unwatch(path, vif)

    def _inject_with_connection_state_check(self):
        watcher = XSWatcher()
        for vif in self.vifs:
            state_path, backend_state_path = get_vif_state_path(vif)
            # watch both frontend and backend vif state
            watcher.watch(state_path, vif)
            watcher.watch(backend_state_path, vif)

        # We have 2 options to handle the blocking function `xs.read_watch`
        # 1. Single thread with Unix alarm signal
        #    The signal handler cannot be properly invoked because when signal received the process context is in
        #    C extension but not python interpreter, so the interpreter cannot call the signal handler.
        #    So we should not apply this option.
        # 2. Multi thread with threading.event or timeout associated join
        #    Create a new thread to invoke the blocking methon `xs.read_watch`. In the main thread using a timeout
        #    associated join method for waiting. We will apply this option.
        t = threading.Thread(target=self.check_and_inject, args=(watcher,))
        t.daemon = True
        t.start()
        t.join(self.vif_connected_timeout)
        if watcher.watches:
            log.warning('Wait vif state change timeout')
            for vif in watcher.watches.values():
                log.warning("Vif:%s state did not change to '%s', don't inject IGMP query to mac: %s" %
                            (vif, VIF_CONNECTED_STATE, get_vif_mac(vif)))

    def _inject_without_connection_state_check(self):
        for vif in self.vifs:
            self.inject_to_vif(vif)


def domid_vifid_of_vif(vif):
    return [int(x) for x in vif.split('vif')[1].split('.')]


def vif_frontend_path(vif):
    domid, vifid = domid_vifid_of_vif(vif)
    dompath = g_xs_handler.get_domain_path(domid)
    return os.path.join(os.path.join(dompath, 'device', 'vif', '%d' % vifid))


def vif_backend_path(vif):
    dompath = g_xs_handler.get_domain_path(0)
    domid, vifid = domid_vifid_of_vif(vif)
    return os.path.join(os.path.join(dompath, 'backend', 'vif', '%d' % domid, '%d' % vifid))


def get_vif_mac(vif):
    return g_xs_handler.read('', os.path.join(vif_frontend_path(vif), 'mac'))


def get_vif_state_path(vif):
    """return frontend and backend vif state path
    """
    return os.path.join(vif_frontend_path(vif), 'state'), os.path.join(vif_backend_path(vif), 'state')


def get_parent_bridge(bridge):
    return subprocess.check_output(['/usr/bin/ovs-vsctl', 'br-to-parent', bridge], universal_newlines=True).strip()


def network_backend_is_openvswitch():
    bridge_type = subprocess.check_output(['/opt/xensource/bin/xe-get-network-backend'], universal_newlines=True).strip()
    return bridge_type == 'openvswitch'


def memodict(f):
    """ Memoization decorator for a function taking a single argument """
    class Memodict(dict):
        def __missing__(self, key):
            ret = self[key] = f(key)
            return ret
    return Memodict().__getitem__


@memodict
def igmp_snooping_is_enabled_on_bridge(bridge):
    vlan = subprocess.check_output(['/usr/bin/ovs-vsctl', 'br-to-vlan', bridge], universal_newlines=True).strip()
    if vlan != '0':
        # this br is a fake br, should get its parent
        bridge = get_parent_bridge(bridge)
    return _igmp_snooping_is_enabled_on_bridge(bridge)


@memodict
def _igmp_snooping_is_enabled_on_bridge(bridge):
    enabled = subprocess.check_output(['/usr/bin/ovs-vsctl', 'get', 'bridge', bridge, 'mcast_snooping_enable'], universal_newlines=True).strip()
    return enabled == 'true'


def igmp_snooping_is_enabled_on_bridge_of_vif(vif):
    bridge = subprocess.check_output(['/usr/bin/ovs-vsctl', 'iface-to-br', vif], universal_newlines=True).strip()
    return igmp_snooping_is_enabled_on_bridge(bridge)


def inject_to_vifs(args):
    log.debug('Entry point: Inject IGMP query per pif')
    if args.no_check_snooping_toggle:
        vifs = args.vifs
    else:
        vifs = [vif for vif in args.vifs if igmp_snooping_is_enabled_on_bridge_of_vif(vif)]
    injector = IGMPQueryInjector(args.max_resp_time, vifs, args.vif_connected_timeout)
    return injector.inject()


def build_parser():
    parser = argparse.ArgumentParser(prog='igmp_query_injector.py', description='Tool for injecting IGMP query packet')
    parser.add_argument('--max-resp-time', dest='max_resp_time', required=False, metavar='max_resp_time', type=int,
                        default=100, help='max response time of IGMP query, unit is millisecond')
    parser.add_argument('--verbose', dest='verbose', required=False, action='store_true',
                        help='print verbose log')
    parser.add_argument('--wait-vif-connected', dest='vif_connected_timeout', metavar='timeout', type=int,
                        default=0, help='timeout value for waiting vif connected, unit is second')
    parser.add_argument('--no-check-snooping-toggle', dest='no_check_snooping_toggle', required=False,
                        action='store_true', help='do not need to check IGMP snooping toggle')
    parser.add_argument('vifs', metavar='vif_name', nargs='+', help='vif interface name in Dom0')
    return parser


def main():
    args = build_parser().parse_args()

    logging_lvl = logging.INFO
    if args.verbose:
        logging_lvl = logging.DEBUG

    log.logToSyslog(level=logging_lvl)

    if not network_backend_is_openvswitch():
        log.info('Network backend type is not openvswitch, no need to inject query')
        sys.exit(0)

    inject_to_vifs(args)


if __name__ == '__main__':
    main()

1	#!/usr/bin/env python3
2	import argparse	1✔
3	import threading	1✔
4	import logging	1✔
5	import subprocess	1✔
6	from scapy.all import Ether, IP, sendp	1✔
7	from scapy.contrib.igmp import IGMP	1✔
8	from xcp import logger as log	1✔
9	import sys	1✔
10	import os	1✔
11	from xen.lowlevel.xs import xs	1✔
12
13
14	# global xenstore handler
15	g_xs_handler = xs()	1✔
16	VIF_CONNECTED_STATE = '4'	1✔
17
18
19	class XSWatcher(object):	1✔
20	"""Tool for watching xenstore
21	"""
22	def __init__(self):	1✔
23	self.watches = dict()	1✔
24
25	def watch(self, path, token):	1✔
26	self.watches[path] = token	1✔
27	return g_xs_handler.watch(path, token)	1✔
28
29	def unwatch(self, path, token):	1✔
30	self.watches.pop(path)	1✔
31	return g_xs_handler.unwatch(path, token)	1✔
32
33	def read_watch(self):	1✔
34	return g_xs_handler.read_watch()	1✔
35
36
37	class IGMPQueryInjector(object):	1✔
38	def __init__(self, max_resp_time, vifs, vif_connected_timeout=0):	1✔
39	self.max_resp_time = max_resp_time	1✔
40	self.vifs = vifs	1✔
41	self.vif_connected_timeout = vif_connected_timeout	1✔
42
43	def inject_to_vif(self, vif):	1✔
44	mac = get_vif_mac(vif)	×
45	try:	×
46	self.inject_packet(vif, mac)	×
47	log.info('Inject IGMP query to vif:%s, mac:%s' % (vif, mac))	×
48	except Exception as e:	×
49	log.error('Inject IGMP query to vif:%s, mac:%s failed' % (vif, mac))	×
50	log.logException(e)	×
51
52	def inject_packet(self, iface, dst_mac):	1✔
53	ether_part = Ether(src='00:00:00:00:00:00', dst=dst_mac)	×
54	ip_part = IP(ttl=1, src='0.0.0.0', dst='224.0.0.1')	×
55	igmp_part = IGMP(type=0x11)	×
56	# Should use integer division // in python 3
57	igmp_part.mrcode = (self.max_resp_time // 100) & 0xff	×
58	igmp_part.igmpize()	×
59	# Make this IGMP query packet as an unicast packet
60	ether_part.dst = dst_mac	×
61	sendp(ether_part / ip_part / igmp_part, iface=iface, verbose=False)	×
62
63	def inject(self):	1✔
64	if not self.vifs:	1✔
65	return	×
66
67	if self.vif_connected_timeout > 0:	1✔
68	# should check connection state
69	log.info('Inject IGMP query with connection state check')	1✔
70	self._inject_with_connection_state_check()	1✔
71	else:
72	log.info('Inject IGMP query without connection state check')	1✔
73	self._inject_without_connection_state_check()	1✔
74
75	def check_and_inject(self, watcher):	1✔
76	vif_connected_set = set()	1✔
77	while watcher.watches:	1✔
78	path, vif = watcher.read_watch()	1✔
79	if g_xs_handler.read('', path) == VIF_CONNECTED_STATE:	1✔
80	# should ensure both backend and frontend connected before injection
81	if vif not in vif_connected_set:	1✔
82	vif_connected_set.add(vif)	1✔
83	else:
84	vif_connected_set.remove(vif)	1✔
85	self.inject_to_vif(vif)	1✔
86	watcher.unwatch(path, vif)	1✔
87
88	def _inject_with_connection_state_check(self):	1✔
89	watcher = XSWatcher()	1✔
90	for vif in self.vifs:	1✔
91	state_path, backend_state_path = get_vif_state_path(vif)	1✔
92	# watch both frontend and backend vif state
93	watcher.watch(state_path, vif)	1✔
94	watcher.watch(backend_state_path, vif)	1✔
95
96	# We have 2 options to handle the blocking function `xs.read_watch`
97	# 1. Single thread with Unix alarm signal
98	# The signal handler cannot be properly invoked because when signal received the process context is in
99	# C extension but not python interpreter, so the interpreter cannot call the signal handler.
100	# So we should not apply this option.
101	# 2. Multi thread with threading.event or timeout associated join
102	# Create a new thread to invoke the blocking methon `xs.read_watch`. In the main thread using a timeout
103	# associated join method for waiting. We will apply this option.
104	t = threading.Thread(target=self.check_and_inject, args=(watcher,))	1✔
105	t.daemon = True	1✔
106	t.start()	1✔
107	t.join(self.vif_connected_timeout)	1✔
108	if watcher.watches:	1✔
109	log.warning('Wait vif state change timeout')	1✔
110	for vif in watcher.watches.values():	1✔
111	log.warning("Vif:%s state did not change to '%s', don't inject IGMP query to mac: %s" %	1✔
112	(vif, VIF_CONNECTED_STATE, get_vif_mac(vif)))
113
114	def _inject_without_connection_state_check(self):	1✔
115	for vif in self.vifs:	1✔
116	self.inject_to_vif(vif)	1✔
117
118
119	def domid_vifid_of_vif(vif):	1✔
120	return [int(x) for x in vif.split('vif')[1].split('.')]	1✔
121
122
123	def vif_frontend_path(vif):	1✔
124	domid, vifid = domid_vifid_of_vif(vif)	1✔
125	dompath = g_xs_handler.get_domain_path(domid)	1✔
126	return os.path.join(os.path.join(dompath, 'device', 'vif', '%d' % vifid))	1✔
127
128
129	def vif_backend_path(vif):	1✔
130	dompath = g_xs_handler.get_domain_path(0)	1✔
131	domid, vifid = domid_vifid_of_vif(vif)	1✔
132	return os.path.join(os.path.join(dompath, 'backend', 'vif', '%d' % domid, '%d' % vifid))	1✔
133
134
135	def get_vif_mac(vif):	1✔
136	return g_xs_handler.read('', os.path.join(vif_frontend_path(vif), 'mac'))	1✔
137
138
139	def get_vif_state_path(vif):	1✔
140	"""return frontend and backend vif state path
141	"""
142	return os.path.join(vif_frontend_path(vif), 'state'), os.path.join(vif_backend_path(vif), 'state')	1✔
143
144
145	def get_parent_bridge(bridge):	1✔
146	return subprocess.check_output(['/usr/bin/ovs-vsctl', 'br-to-parent', bridge], universal_newlines=True).strip()	×
147
148
149	def network_backend_is_openvswitch():	1✔
150	bridge_type = subprocess.check_output(['/opt/xensource/bin/xe-get-network-backend'], universal_newlines=True).strip()	×
151	return bridge_type == 'openvswitch'	×
152
153
154	def memodict(f):	1✔
155	""" Memoization decorator for a function taking a single argument """
156	class Memodict(dict):	1✔
157	def __missing__(self, key):	1✔
158	ret = self[key] = f(key)	×
159	return ret	×
160	return Memodict().__getitem__	1✔
161
162
163	@memodict	1✔
164	def igmp_snooping_is_enabled_on_bridge(bridge):	1✔
165	vlan = subprocess.check_output(['/usr/bin/ovs-vsctl', 'br-to-vlan', bridge], universal_newlines=True).strip()	×
166	if vlan != '0':	×
167	# this br is a fake br, should get its parent
168	bridge = get_parent_bridge(bridge)	×
169	return _igmp_snooping_is_enabled_on_bridge(bridge)	×
170
171
172	@memodict	1✔
173	def _igmp_snooping_is_enabled_on_bridge(bridge):	1✔
174	enabled = subprocess.check_output(['/usr/bin/ovs-vsctl', 'get', 'bridge', bridge, 'mcast_snooping_enable'], universal_newlines=True).strip()	×
175	return enabled == 'true'	×
176
177
178	def igmp_snooping_is_enabled_on_bridge_of_vif(vif):	1✔
179	bridge = subprocess.check_output(['/usr/bin/ovs-vsctl', 'iface-to-br', vif], universal_newlines=True).strip()	×
180	return igmp_snooping_is_enabled_on_bridge(bridge)	×
181
182
183	def inject_to_vifs(args):	1✔
184	log.debug('Entry point: Inject IGMP query per pif')	×
185	if args.no_check_snooping_toggle:	×
186	vifs = args.vifs	×
187	else:
188	vifs = [vif for vif in args.vifs if igmp_snooping_is_enabled_on_bridge_of_vif(vif)]	×
189	injector = IGMPQueryInjector(args.max_resp_time, vifs, args.vif_connected_timeout)	×
190	return injector.inject()	×
191
192
193	def build_parser():	1✔
194	parser = argparse.ArgumentParser(prog='igmp_query_injector.py', description='Tool for injecting IGMP query packet')	×
195	parser.add_argument('--max-resp-time', dest='max_resp_time', required=False, metavar='max_resp_time', type=int,	×
196	default=100, help='max response time of IGMP query, unit is millisecond')
197	parser.add_argument('--verbose', dest='verbose', required=False, action='store_true',	×
198	help='print verbose log')
199	parser.add_argument('--wait-vif-connected', dest='vif_connected_timeout', metavar='timeout', type=int,	×
200	default=0, help='timeout value for waiting vif connected, unit is second')
201	parser.add_argument('--no-check-snooping-toggle', dest='no_check_snooping_toggle', required=False,	×
202	action='store_true', help='do not need to check IGMP snooping toggle')
203	parser.add_argument('vifs', metavar='vif_name', nargs='+', help='vif interface name in Dom0')	×
204	return parser	×
205
206
207	def main():	1✔
208	args = build_parser().parse_args()	×
209
210	logging_lvl = logging.INFO	×
211	if args.verbose:	×
212	logging_lvl = logging.DEBUG	×
213
214	log.logToSyslog(level=logging_lvl)	×
215
216	if not network_backend_is_openvswitch():	×
217	log.info('Network backend type is not openvswitch, no need to inject query')	×
218	sys.exit(0)	×
219
220	inject_to_vifs(args)	×
221
222
223	if __name__ == '__main__':	1✔
224	main()	×

xapi-project / xen-api / 13457957190

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous