Program for MT call load testing with rtpsource

This program (derived from is used to start MT calls
via the MNCC interface of OsmoMSC.  Every calls RTP is connected
to the new 'rtpsource' process, which generates a realistic RTP
flow in terms of number of packets (20ms interval) and payload size.

Change-Id: I9e5d799aaeeff5188dc97061f0d6e1873d9bf653
Harald Welte 3 years ago
parent 09462c3b04
commit 41af37de9f
  1. 70
  2. 188

@ -0,0 +1,70 @@
# Simple class for synchronous/blocking interface with a remote
# Osmocom program via the CTRL interface
# (C) 2020 by Harald Welte <>
# Licensed under GNU General Public License, Version 2 or at your
# option, any later version.
from osmopy.osmo_ipa import Ctrl
import socket
class OsmoCtrlSimple:
'''Simple class for synchronous/blocking interface with a remote
Osmocom program via the CTRL interface'''
remote_host = ""
remote_port = 0
sock = None
def __init__(self, host=None, port=None):
self.remote_host = host
self.remote_port = port
def connect(self, host=None, port=None):
if host:
self.remote_host = host
if port:
self.remote_port = port
self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.sock.connect((self.remote_host, self.remote_port))
return self.sock
def disconnect(self):
self.sock = None
def _leftovers(self, fl):
Read outstanding data if any according to flags
data = self.sock.recv(1024, fl)
except socket.error as _:
return False
if len(data) != 0:
tail = data
while True:
(head, tail) = Ctrl().split_combined(tail)
print("Got message:", Ctrl().rem_header(head))
if len(tail) == 0:
return True
return False
def do_set_get(self, var, value = None):
(r, c) = Ctrl().cmd(var, value)
while True:
ret = self.sock.recv(4096)
# handle multiple messages, ignore TRAPs
ret = Ctrl().skip_traps(ret)
if ret != None:
(i, k, v) = Ctrl().parse(ret)
return (Ctrl().rem_header(ret),) + Ctrl().verify(ret, r, var, value)
def set_var(self, var, val):
(a, _, _) = self.do_set_get(var, val)
return a

@ -0,0 +1,188 @@
#!/usr/bin/env python2
# Python testing tool for establishing mobile-terminated calls using
# the OsmoMSC MNCC interface. It wants to communicate via CTRL with a
# 'rtpsoucre' program which will take care of generating RtP flows
# for the MT calls that are established.
# (C) 2015, 2020 by Harald Welte <>
# Licensed under GNU General Public License, Version 2 or at your
# option, any later version.
from gsm_call_fsm import GsmCallFsm, GSM48
from mncc_sock import MnccSocket
from thread import start_new_thread
from ctrl import OsmoCtrlSimple
import pykka
import logging as log
import signal, sys, time
import readline, code
import socket
import struct
import mncc
from mncc_sock import mncc_rtp_msg
def int2ipstr(addr):
'''Convert from an integer to a strinf-formatted IPv4 address'''
return socket.inet_ntoa(struct.pack("!I", addr))
def ipstr2int(addr):
'''Convert from a string-formatted IPv4 address to integer'''
return struct.unpack("!I", socket.inet_aton(addr))[0]
class MnccActor(pykka.ThreadingActor):
'''MnccActor provides an interface for GsmCallFsm to send MNCC messages'''
def __init__(self, mncc_sock):
super(MnccActor, self).__init__()
self.mncc_sock = mncc_sock
def on_receive(self, message):
if message['type'] == 'send':
msg = message['msg']
log.debug('MnccActor TxMNCC %s' % msg)
raise Exception('mncc', 'MnccActor Received unhandled %s' % message)
def mncc_rx_thread(mncc_sock):
'''MNCC receive thread, broadcasting received MNCC packets to GsmCallFsm'''
while 1:
msg = mncc_sock.recv()
if msg == None:
raise Exception('mncc', 'MnccSocket disconnected')
if msg.is_frame():
log.warning("Dropping traffic frame: %s" % msg)
log.debug("MnccActor RxMNCC %s, broadcasting to Call FSMs" % msg)
# we simply broadcast to all calls
pykka.ActorRegistry.broadcast({'type': 'mncc', 'msg': msg}, GsmCallFsm)
class RtpSourceCtrlActor(pykka.ThreadingActor):
'''Actor for talking with the rtpsource program via an osmocom CTRL interface'''
def __init__(self, ctrl_host, ctrl_port=11111):
super(RtpSourceCtrlActor, self).__init__()
self.ctrl = OsmoCtrlSimple(ctrl_host, ctrl_port)
def _set_var(self, var, val):
(res, var, val) = self.ctrl.do_set_get(var, val)
log.debug('RtpSourceCtrlActor result=%s var=%s val=%s' % (res, var, val))
return (res, var, val)
def on_receive(self, message):
if message['type'] == 'rtp_create':
(res, var, val) = self._set_var('rtp_create', message['cname'])
v = val.split(',') # input looks like '1,,37723'
return {'cname': v[0], 'remote_host': v[1], 'remote_port': int(v[2])}
elif message['type'] == 'rtp_connect':
val = '%s,%s,%u,%u' % (message['cname'], message['remote_host'],
message['remote_port'], message['payload_type'])
(res, var, val) = self._set_var('rtp_connect', val)
return res
elif message['type'] == 'rtp_delete':
(res, var, val) = self._set_var('rtp_delete', message['cname'])
return res
raise Exception('ctrl', 'RtpSourceCtrlActor Received unhandled %s' % message)
class MTCallRtpsource(pykka.ThreadingActor):
'''Actor to start a network-initiated MT (mobile terminated) call to a given MSISDN,
connecting the RTP flow to an extenal rtpsource program'''
def __init__(self, mncc_act, ctrl_act, codecs_permitted = GSM48.AllCodecs):
super(MTCallRtpsource, self).__init__()
self.mncc_act = mncc_act
self.ctrl_act = ctrl_act
self.codecs_permitted = codecs_permitted = GsmCallFsm.start(self.mncc_act, self.actor_ref, True, self.codecs_permitted)
self.callref ={'type':'get_callref'})
self.state = 'NULL'
self.rtp_msc = None
def start_call(self, msisdn_called, msisdn_calling):
'''Start a MT call from given [external] calling party to given mobile party MSISDN'''
self.msisdn_called = msisdn_called
self.msisdn_calling = msisdn_calling
# allocate a RTP connection @ rtpsource
r = self.ctrl_act.ask({'type':'rtp_create', 'cname':self.callref})
self.ext_rtp_host = r['remote_host']
self.ext_rtp_port = r['remote_port']
# start the MNCC call FSM{'type':'start_mt_call',
'calling':self.msisdn_calling, 'called':self.msisdn_called})
def on_stop(self):
# Attempt to do a graceful shutdown by deleting the RTP connection from rtpsource
self.ctrl_act.ask({'type':'rtp_delete', 'cname':self.callref})
def on_failure(self, exception_type, exception_value, traceback):
# on_stop() is not called in case of failure; fix that
def on_receive(self, message):
if message['type'] == 'rtp_create_ind':
# MSC+MGW informs us of the PLMN side IP/port
mncc_rtp_ind = message['rtp']
# tell external rtpsourc to connect to this address
r = self.ctrl_act.ask({'type':'rtp_connect', 'cname':self.callref,
'remote_host': int2ipstr(mncc_rtp_ind.ip),
'remote_port': mncc_rtp_ind.port,
# tell MSC to connect to ip/port @ rtpsource
mncc_rtp_conn = mncc_rtp_msg(msg_type=mncc.MNCC_RTP_CONNECT, callref=self.callref,
ip=ipstr2int(self.ext_rtp_host), port=self.ext_rtp_port)
self.mncc_act.tell({'type': 'send', 'msg': mncc_rtp_conn})
elif message['type'] == 'call_state_change':
if message['new_state'] == 'NULL':
# Call FSM has reached the NULL state again (call terminated)
# on_stop() will clean up the RTP connection at rtpsource
# capture SIGINT (Ctrl+C) and stop all actors
def sigint_handler(signum, frame):
log.basicConfig(level = log.DEBUG,
format = "%(levelname)s %(filename)s:%(lineno)d %(message)s")
signal.signal(signal.SIGINT, sigint_handler)
# start the MnccSocket and associated pykka actor + rx thread
mncc_sock = MnccSocket()
mncc_act = MnccActor.start(mncc_sock)
start_new_thread(mncc_rx_thread, (mncc_sock,))
# connect via CTRL to rtpsource
rtpctrl_act = RtpSourceCtrlActor.start(RTPSOURCE_CTRL_IP, RTPSOURCE_CTRL_PORT)
# convenience wrapper
def mt_call(msisdn_called, msisdn_calling = '123456789', codecs = GSM48.AllCodecs):
call_conn = MTCallRtpsource.start(mncc_act, rtpctrl_act, codecs).proxy()
call_conn.start_call(msisdn_called, msisdn_calling)
return call_conn
# start one call automatically
# start a shell to enable the user to add more calls as needed
vars = globals().copy()
shell = code.InteractiveConsole(vars)
except SystemExit:
# clan up after user leaves interactive shell
sigint_handler(signal.SIGINT, None)