summaryrefslogtreecommitdiffstats
path: root/src/osmo_ms_driver/location_update_test.py
blob: 5ff219974321336caf6f1fe423007311d10080f9 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
# osmo_ms_driver: Location Update Test
# Create MS's and wait for the Location Update to succeed.
#
# Copyright (C) 2018 by Holger Hans Peter Freyther
#
# 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/>.

from copy import copy
from osmo_gsm_tester import log
from .starter import OsmoVirtPhy, OsmoMobile
from .test_support import Results

from datetime import timedelta

import collections
import time

class LUResult(Results):
    """Representation of a Location Updating Result."""

    def __init__(self, name):
        super().__init__(name)
        self._time_of_lu = None

    def set_lu_time(self, time):
        assert self._time_of_lu is None
        self._time_of_lu = time

    def has_lu_time(self):
        return self._time_of_lu is not None

    def lu_time(self):
        return self._time_of_lu or 0

    def lu_delay(self):
        return self.lu_time() - self.start_time()


LUStats = collections.namedtuple("LUStats", ["num_attempted", "num_completed",
                                 "min_latency", "max_latency"])

class MassUpdateLocationTest(log.Origin):
    """
    A test to launch a configurable amount of MS and make them
    execute a Location Updating Procedure.

    Configure the number of MS to be tested and a function that
    decides how quickly to start them and a timeout.
    """

    TEMPLATE_LUA = "osmo-mobile-lu.lua"
    TEMPLATE_CFG = "osmo-mobile.cfg"

    def __init__(self, name, options, cdf_function,
                 event_server, tmp_dir, suite_run=None):
        super().__init__(log.C_RUN, name)
        self._binary_options = options
        self._cdf = cdf_function
        self._suite_run = suite_run
        self._tmp_dir = tmp_dir
        self._unstarted = []
        self._mobiles = []
        self._phys = []
        self._results = {}

        self._event_server = event_server
        self._event_server.register(self.handle_msg)
        self._started = []
        self._subscribers = []

    def subscriber_add(self, subscriber):
        """
        Adds a subscriber to the list of subscribers.

        Must be called before starting the testcase.
        """
        self._subscribers.append(subscriber)

    def configure_tasks(self):
        """Sets up the test run."""

        self._cdf.set_target(len(self._subscribers))
        self._outstanding = len(self._subscribers)
        for i in range(0, self._outstanding):
            ms_name = "%.5d" % i

            phy = OsmoVirtPhy(self._binary_options.virtphy,
                              self._binary_options.env,
                              ms_name, self._tmp_dir)
            self._phys.append(phy)

            launcher = OsmoMobile(self._binary_options.mobile,
                                self._binary_options.env,
                                ms_name, self._tmp_dir, self.TEMPLATE_LUA,
                                self.TEMPLATE_CFG, self._subscribers[i],
                                phy.phy_filename(),
                                self._event_server.server_path())
            self._results[ms_name] = LUResult(ms_name)
            self._mobiles.append(launcher)
        self._unstarted = copy(self._mobiles)

    def pre_launch(self, loop):
        """
        We need the virtphy's be ready when the lua script in the
        mobile comes and kicks-off the test. In lua we don't seem to
        be able to just stat/check if a file/socket exists so we need
        to do this from here.
        """
        self.log("Pre-launching all virtphy's")
        for phy in self._phys:
            phy.start(loop, self._suite_run)

        self.log("Checking if sockets are in the filesystem")
        for phy in self._phys:
            phy.verify_ready()

    def prepare(self, loop):
        self.log("Starting testcase")

        self.configure_tasks()
        self.pre_launch(loop)

        self._start_time = time.clock_gettime(time.CLOCK_MONOTONIC)
        self._end_time = self._start_time + \
                            self._cdf.duration().total_seconds() + \
                            timedelta(seconds=120).total_seconds()

        self._started = []
        self._too_slow = 0

    def step_once(self, loop, current_time):
        if len(self._unstarted) <= 0:
            return current_time, None

        step_size = self._cdf.step_size().total_seconds()

        # Start
        self._cdf.step_once()

        # Check for timeout
        # start pending MS
        while len(self._started) < self._cdf.current_scaled_value() and len(self._unstarted) > 0:
            ms = self._unstarted.pop(0)
            ms.start(loop, self._suite_run)
            launch_time = time.clock_gettime(time.CLOCK_MONOTONIC)
            self._results[ms.name_number()].set_launch_time(launch_time)
            self._started.append(ms)

        now_time = time.clock_gettime(time.CLOCK_MONOTONIC)
        sleep_time = (current_time + step_size) - now_time
        if sleep_time <= 0:
            self.log("Starting too slowly. Moving on",
		    target=(current_time + step_size), now=now_time, sleep=sleep_time)
            self._too_slow += 1
            sleep_time = 0

        if len(self._unstarted) == 0:
            end_time = time.clock_gettime(time.CLOCK_MONOTONIC)
            self.log("All started...", too_slow=self._too_slow, duration=end_time - self._start_time)
            return current_time, None

        return current_time + step_size, sleep_time

    def run_test(self, loop, test_duration):
        self.prepare(loop)

        to_complete_time = self._start_time + test_duration.total_seconds()
        tick_time = self._start_time

        while not self.all_completed():
            tick_time, sleep_time = self.step_once(loop, tick_time)
            now_time = time.clock_gettime(time.CLOCK_MONOTONIC)
            if sleep_time is None:
                sleep_time = to_complete_time - now_time
            if sleep_time < 0:
                break
            loop.schedule_timeout(sleep_time)
            loop.select()

    def stop_all(self):
        for launcher in self._started:
            launcher.terminate()

    def handle_msg(self, _data, addr, time):
        import json
        data = json.loads(_data.decode())

        if data['type'] == 'register':
            ms = self._results[data['ms']]
            ms.set_start_time(time)
            launch_delay = ms.start_time() - ms.launch_time()
            self.log("MS start registered ", ms=ms, at=time, delay=launch_delay)
        elif data['type'] == 'event':
            if data['data']['lu_done'] == 1:
                ms = self._results[data['ms']]
                if not ms.has_lu_time():
                    self._outstanding = self._outstanding - 1
                ms.set_lu_time(time)
                self.log("MS performed LU ", ms=ms, at=time, lu_delay=ms.lu_delay())
        else:
            print(time, data)
            raise Exception("Unknown event type..:" + _data.decode())


    def all_completed(self):
        return self._outstanding == 0

    def find_min_max(self, results):
        min_value = max_value = None
        for result in results:
            if min_value is None or result.lu_delay() < min_value:
                min_value = result.lu_delay()
            if max_value is None or result.lu_delay() > max_value:
                max_value = result.lu_delay()
        return min_value, max_value

    def get_result_values(self):
        """
        Returns the raw result values of the test run in any order.
        """
        return self._results.values()

    def get_stats(self):
        """
        Returns a statistical summary of the test.
        """
        attempted = len(self._subscribers)
        completed = attempted - self._outstanding
        min_latency, max_latency = self.find_min_max(filter(lambda x: x.has_lu_time(), self._results.values()))
        return LUStats(attempted, completed, min_latency, max_latency)

    def print_stats(self):
        stats = self.get_stats()
        all_completed = stats.num_attempted == stats.num_completed

        self.log("Tests done", all_completed=all_completed,
                    min=stats.min_latency, max=stats.max_latency)