gnuradio zmq broker refactoring
First step towards support everal ENBs and installing the remote script at runtime: * The gnuradio broker class is moved to its own file, to decouple it from RFemu. * The ENBs are registered earlier in the test so that the GrBroker knows when to start (delay start + setup until all ENBs have been configured). * Handle ENBs internally as a list. Change-Id: I4f1095bbc7ed0a816fe47caef44f7becadd9d737
This commit is contained in:
parent
c230efcb19
commit
fbb8611381
|
|
@ -21,7 +21,7 @@ from abc import ABCMeta, abstractmethod
|
|||
from ..core import log, config
|
||||
from ..core import schema
|
||||
from . import run_node
|
||||
from .rfemu_gnuradio_zmq import GrBroker
|
||||
from .gnuradio_zmq_broker import GrBroker
|
||||
|
||||
def on_register_schemas():
|
||||
resource_schema = {
|
||||
|
|
@ -91,7 +91,9 @@ class eNodeB(log.Origin, metaclass=ABCMeta):
|
|||
self._num_cells = None
|
||||
self._epc = None
|
||||
self.gen_conf = None
|
||||
self.gr_broker = None
|
||||
self.gr_broker = GrBroker.ref()
|
||||
self.gr_broker.register_enb(self)
|
||||
self._use_gr_broker = False
|
||||
|
||||
def using_grbroker(self, cfg_values):
|
||||
# whether we are to use Grbroker in between ENB and UE.
|
||||
|
|
@ -189,7 +191,8 @@ class eNodeB(log.Origin, metaclass=ABCMeta):
|
|||
enb_bind_port = resourcep.next_zmq_port_range(self, num_ports)
|
||||
self.assign_enb_zmq_ports(values, 'zmq_enb_bind_port', enb_bind_port)
|
||||
# If we are to use a GrBroker, then initialize here to have remote zmq ports available:
|
||||
if self.using_grbroker(values):
|
||||
self._use_gr_broker = self.using_grbroker(values)
|
||||
if self._use_gr_broker:
|
||||
zmq_enb_peer_port = resourcep.next_zmq_port_range(self, num_ports)
|
||||
self.assign_enb_zmq_ports(values, 'zmq_enb_peer_port', zmq_enb_peer_port) # These are actually bound to GrBroker
|
||||
self.assign_enb_zmq_ports_joined_earfcn(values, 'zmq_ue_bind_port', ue_bind_port) # This is were GrBroker binds on the UE side
|
||||
|
|
@ -197,8 +200,7 @@ class eNodeB(log.Origin, metaclass=ABCMeta):
|
|||
self.assign_enb_zmq_ports_joined_earfcn(values, 'zmq_ue_peer_port', zmq_ue_peer_port) # This is were GrBroker binds on the UE side
|
||||
# Already set gen_conf here in advance since gr_broker needs the cell list
|
||||
self.gen_conf = values
|
||||
self.gr_broker = GrBroker.ref()
|
||||
self.gr_broker.handle_enb(self)
|
||||
self.gr_broker.start()
|
||||
else:
|
||||
self.assign_enb_zmq_ports(values, 'zmq_enb_peer_port', ue_bind_port)
|
||||
self.assign_enb_zmq_ports(values, 'zmq_ue_bind_port', ue_bind_port) #If no broker we need to match amount of ports
|
||||
|
|
@ -223,6 +225,7 @@ class eNodeB(log.Origin, metaclass=ABCMeta):
|
|||
def cleanup(self):
|
||||
'Nothing to do by default. Subclass can override if required.'
|
||||
if self.gr_broker:
|
||||
self.gr_broker.unregister_enb(self)
|
||||
GrBroker.unref()
|
||||
self.gr_broker = None
|
||||
|
||||
|
|
@ -247,7 +250,7 @@ class eNodeB(log.Origin, metaclass=ABCMeta):
|
|||
def get_zmq_rf_dev_args(self, cfg_values):
|
||||
base_srate = self.num_prb2base_srate(self.num_prb())
|
||||
|
||||
if self.gr_broker:
|
||||
if self._use_gr_broker:
|
||||
ul_rem_addr = self.addr()
|
||||
else:
|
||||
ul_rem_addr = self.ue.addr()
|
||||
|
|
@ -274,7 +277,7 @@ class eNodeB(log.Origin, metaclass=ABCMeta):
|
|||
idx = 0
|
||||
earfcns_done = []
|
||||
for cell in cell_list:
|
||||
if self.gr_broker:
|
||||
if self._use_gr_broker:
|
||||
if cell['dl_earfcn'] in earfcns_done:
|
||||
continue
|
||||
earfcns_done.append(cell['dl_earfcn'])
|
||||
|
|
|
|||
|
|
@ -0,0 +1,161 @@
|
|||
# osmo_gsm_tester: class defining a RF emulation object implemented using SRS ENB stdin interface
|
||||
#
|
||||
# Copyright (C) 2020 by sysmocom - s.f.m.c. GmbH
|
||||
#
|
||||
# Author: Pau Espin Pedrol <pespin@sysmocom.de>
|
||||
#
|
||||
# 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 3 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, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import json
|
||||
import socket
|
||||
|
||||
from ..core import log
|
||||
from ..core import util
|
||||
from ..core import process
|
||||
from ..core import remote
|
||||
from ..core.event_loop import MainLoop
|
||||
|
||||
class GrBroker(log.Origin):
|
||||
|
||||
# static fields:
|
||||
refcount = 0
|
||||
instance = None
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(log.C_RUN, 'gr_zmq_broker')
|
||||
self.process = None
|
||||
self.ctrl_port = 5005
|
||||
self.run_dir = None
|
||||
self.rem_host = None
|
||||
self.enb_li = []
|
||||
self.addr = None
|
||||
self.ctrl_sk = None
|
||||
self.num_enb_started = 0
|
||||
|
||||
@staticmethod
|
||||
def ref():
|
||||
if GrBroker.refcount == 0:
|
||||
GrBroker.instance = GrBroker()
|
||||
GrBroker.refcount = GrBroker.refcount + 1
|
||||
return GrBroker.instance
|
||||
|
||||
@staticmethod
|
||||
def unref():
|
||||
GrBroker.refcount = GrBroker.refcount - 1
|
||||
if GrBroker.refcount == 0:
|
||||
GrBroker.instance.cleanup()
|
||||
GrBroker.instance = None
|
||||
|
||||
|
||||
def cleanup(self):
|
||||
if self.ctrl_sk is not None:
|
||||
self.cmd_exit()
|
||||
self.ctrl_sk.close()
|
||||
self.ctrl_sk = None
|
||||
self.enb_li = []
|
||||
self.testenv = None
|
||||
|
||||
def register_enb(self, enb):
|
||||
self.enb_li.append(enb)
|
||||
|
||||
def unregister_enb(self, enb):
|
||||
self.enb_li.remove(enb)
|
||||
|
||||
def gen_json_enb(self, enb):
|
||||
res = []
|
||||
cell_list = enb.gen_conf['enb']['cell_list']
|
||||
for cell in cell_list:
|
||||
# TODO: probably add enb_id, cell_id to support several ENB
|
||||
data = {'earfcn': int(cell['dl_earfcn']),
|
||||
'bind_port': int(cell['zmq_enb_peer_port']),
|
||||
'peer_addr': enb.addr(),
|
||||
'peer_port': int(cell['zmq_enb_bind_port']),
|
||||
'use_mimo': True if enb.num_ports() > 1 else False
|
||||
}
|
||||
res.append(data)
|
||||
return res
|
||||
|
||||
def gen_json_ue(self, enb):
|
||||
res = {}
|
||||
res = []
|
||||
earfcns_done = []
|
||||
cell_list = enb.gen_conf['enb']['cell_list']
|
||||
for cell in cell_list:
|
||||
data = {}
|
||||
if int(cell['dl_earfcn']) in earfcns_done:
|
||||
continue
|
||||
earfcns_done.append(int(cell['dl_earfcn']))
|
||||
data = {'earfcn': int(cell['dl_earfcn']),
|
||||
'bind_port': int(cell['zmq_ue_peer_port']),
|
||||
'peer_addr': enb.ue.addr(),
|
||||
'peer_port': int(cell['zmq_ue_bind_port']),
|
||||
'use_mimo': True if enb.num_ports() > 1 else False
|
||||
}
|
||||
res.append(data)
|
||||
return res
|
||||
|
||||
def gen_json(self):
|
||||
res = {'enb': [self.gen_json_enb(enb) for enb in self.enb_li],
|
||||
'ue': [self.gen_json_ue(self.enb_li[0])]}
|
||||
return res
|
||||
|
||||
def configure(self):
|
||||
self.addr = self.enb_li[0].addr()
|
||||
self.testenv = self.enb_li[0].testenv
|
||||
self.run_dir = util.Dir(self.testenv.test().get_run_dir().new_dir(self.name()))
|
||||
if not self.enb_li[0]._run_node.is_local():
|
||||
self.rem_host = remote.RemoteHost(self.run_dir, self.enb_li[0]._run_node.ssh_user(), self.enb_li[0]._run_node.ssh_addr())
|
||||
|
||||
def start(self):
|
||||
self.num_enb_started += 1
|
||||
self.dbg('start(%d/%d)' % (self.num_enb_started, len(self.enb_li)))
|
||||
if self.num_enb_started == 1:
|
||||
self.configure()
|
||||
args = ('osmo-gsm-tester_zmq_broker.py',
|
||||
'-c', str(self.ctrl_port),
|
||||
'-b', self.addr)
|
||||
if self.enb_li[0]._run_node.is_local():
|
||||
self.process = process.Process(self.name(), self.run_dir, args)
|
||||
else:
|
||||
self.process = self.rem_host.RemoteProcessSafeExit('zmq_gr_broker', util.Dir('/tmp/ogt_%s' % self.name()), args, wait_time_sec=7)
|
||||
self.testenv.remember_to_stop(self.process)
|
||||
self.process.launch()
|
||||
# Wait until all ENBs are configured/started:
|
||||
if self.num_enb_started == len(self.enb_li):
|
||||
self.dbg('waiting for gr script to be available...')
|
||||
MainLoop.sleep(5)
|
||||
self.ctrl_sk = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
self.cmd_setup()
|
||||
|
||||
def send_cmd(self, str_buf):
|
||||
self.dbg('sending cmd: "%s"' % str_buf)
|
||||
self.ctrl_sk.sendto(str_buf.encode('utf-8'), (self.addr, self.ctrl_port))
|
||||
|
||||
def cmd_setup(self):
|
||||
cfg = self.gen_json()
|
||||
buf = json.dumps(cfg)
|
||||
self.send_cmd(buf)
|
||||
|
||||
def cmd_set_relative_gain_on_local_port(self, port, rel_gain):
|
||||
d = { 'action': 'set_relative_gain',
|
||||
'port': port,
|
||||
'rel_gain': rel_gain
|
||||
}
|
||||
buf = json.dumps(d)
|
||||
self.send_cmd(buf)
|
||||
|
||||
def cmd_exit(self):
|
||||
d = { 'action': 'exit' }
|
||||
buf = json.dumps(d)
|
||||
self.send_cmd(buf)
|
||||
|
|
@ -17,142 +17,9 @@
|
|||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import json
|
||||
import socket
|
||||
|
||||
from ..core import log
|
||||
from ..core import util
|
||||
from ..core import process
|
||||
from ..core import remote
|
||||
from ..core.event_loop import MainLoop
|
||||
from .rfemu import RFemulation
|
||||
|
||||
|
||||
class GrBroker(log.Origin):
|
||||
|
||||
# static fields:
|
||||
refcount = 0
|
||||
instance = None
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(log.C_RUN, 'zmq_gr_broker')
|
||||
self.process = None
|
||||
self.ctrl_port = 5005
|
||||
self.run_dir = None
|
||||
self.rem_host = None
|
||||
self.cfg = None
|
||||
self.enb = None
|
||||
self.addr = None
|
||||
self.ctrl_sk = None
|
||||
|
||||
@staticmethod
|
||||
def ref():
|
||||
if GrBroker.refcount == 0:
|
||||
GrBroker.instance = GrBroker()
|
||||
GrBroker.refcount = GrBroker.refcount + 1
|
||||
return GrBroker.instance
|
||||
|
||||
@staticmethod
|
||||
def unref():
|
||||
GrBroker.refcount = GrBroker.refcount - 1
|
||||
if GrBroker.refcount == 0:
|
||||
GrBroker.instance.cleanup()
|
||||
GrBroker.instance = None
|
||||
|
||||
|
||||
def cleanup(self):
|
||||
if self.ctrl_sk is not None:
|
||||
self.cmd_exit()
|
||||
self.ctrl_sk.close()
|
||||
self.ctrl_sk = None
|
||||
self.enb = None
|
||||
self.testenv = None
|
||||
|
||||
def handle_enb(self, enb):
|
||||
self.enb = enb
|
||||
self.addr = self.enb.addr()
|
||||
self.testenv = self.enb.testenv
|
||||
self.cfg = self.gen_json(enb)
|
||||
# FIXME: we may need to delay this somehow if we want to support several ENBs
|
||||
self.start()
|
||||
self.setup()
|
||||
|
||||
def gen_json_enb(self, enb):
|
||||
res = []
|
||||
cell_list = enb.gen_conf['enb']['cell_list']
|
||||
for cell in cell_list:
|
||||
# TODO: probably add enb_id, cell_id to support several ENB
|
||||
data = {'earfcn': int(cell['dl_earfcn']),
|
||||
'bind_port': int(cell['zmq_enb_peer_port']),
|
||||
'peer_addr': enb.addr(),
|
||||
'peer_port': int(cell['zmq_enb_bind_port']),
|
||||
'use_mimo': True if enb.num_ports() > 1 else False
|
||||
}
|
||||
res.append(data)
|
||||
return res
|
||||
|
||||
def gen_json_ue(self, enb):
|
||||
res = {}
|
||||
res = []
|
||||
earfcns_done = []
|
||||
cell_list = enb.gen_conf['enb']['cell_list']
|
||||
for cell in cell_list:
|
||||
data = {}
|
||||
if int(cell['dl_earfcn']) in earfcns_done:
|
||||
continue
|
||||
earfcns_done.append(int(cell['dl_earfcn']))
|
||||
data = {'earfcn': int(cell['dl_earfcn']),
|
||||
'bind_port': int(cell['zmq_ue_peer_port']),
|
||||
'peer_addr': enb.ue.addr(),
|
||||
'peer_port': int(cell['zmq_ue_bind_port']),
|
||||
'use_mimo': True if enb.num_ports() > 1 else False
|
||||
}
|
||||
res.append(data)
|
||||
return res
|
||||
|
||||
def gen_json(self, enb):
|
||||
res = {'enb': [self.gen_json_enb(enb)],
|
||||
'ue': [self.gen_json_ue(enb)]}
|
||||
return res
|
||||
|
||||
def start(self):
|
||||
self.run_dir = util.Dir(self.testenv.test().get_run_dir().new_dir(self.name()))
|
||||
|
||||
args = ('osmo-gsm-tester_zmq_broker.py',
|
||||
'-c', str(self.ctrl_port),
|
||||
'-b', self.enb.addr())
|
||||
|
||||
if self.enb._run_node.is_local():
|
||||
self.process = process.Process(self.name(), self.run_dir, args)
|
||||
else:
|
||||
self.rem_host = remote.RemoteHost(self.run_dir, self.enb._run_node.ssh_user(), self.enb._run_node.ssh_addr())
|
||||
self.process = self.rem_host.RemoteProcessSafeExit('zmq_gr_broker', util.Dir('/tmp/ogt_%s' % self.name()), args, wait_time_sec=7)
|
||||
self.testenv.remember_to_stop(self.process)
|
||||
self.process.launch()
|
||||
|
||||
def setup(self):
|
||||
self.dbg('waiting for gr script to be available...')
|
||||
MainLoop.sleep(5)
|
||||
self.ctrl_sk = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
buf = json.dumps(self.cfg)
|
||||
self.send_cmd(buf)
|
||||
|
||||
def send_cmd(self, str_buf):
|
||||
self.dbg('sending cmd: "%s"' % str_buf)
|
||||
self.ctrl_sk.sendto(str_buf.encode('utf-8'), (self.addr, self.ctrl_port))
|
||||
|
||||
def cmd_set_relative_gain_on_local_port(self, port, rel_gain):
|
||||
d = { 'action': 'set_relative_gain',
|
||||
'port': port,
|
||||
'rel_gain': rel_gain
|
||||
}
|
||||
buf = json.dumps(d)
|
||||
self.send_cmd(buf)
|
||||
|
||||
def cmd_exit(self):
|
||||
d = { 'action': 'exit' }
|
||||
buf = json.dumps(d)
|
||||
self.send_cmd(buf)
|
||||
from .gnuradio_zmq_broker import GrBroker
|
||||
|
||||
class RFemulationGnuradioZmq(RFemulation):
|
||||
##############
|
||||
|
|
|
|||
Loading…
Reference in New Issue