(
self, workload_config: WorkloadConfig
)
| 301 | class UserSessionManager: |
| 302 | |
| 303 | def __init__( |
| 304 | self, workload_config: WorkloadConfig |
| 305 | ): |
| 306 | self.workload_config = workload_config |
| 307 | self.sessions = [] |
| 308 | |
| 309 | gap_between_requests_per_user = workload_config.user_request_interval |
| 310 | self.gap_between_users = workload_config.new_user_interval |
| 311 | |
| 312 | logger.info( |
| 313 | f"Gap between users: {self.gap_between_users} secs.\n" |
| 314 | f"Gap between user reqs: {gap_between_requests_per_user} secs." |
| 315 | ) |
| 316 | |
| 317 | self.user_id = 0 |
| 318 | self.last_user_join = 0 |
| 319 | self.session_summaries = [] |
| 320 | self.start_time = None |
| 321 | |
| 322 | self.traces = [] |
| 323 | for usr_id, trace_file in enumerate(self.workload_config.trace_file): |
| 324 | self.traces.append([]) |
| 325 | with open(trace_file, "r", encoding="utf-8") as f: |
| 326 | for round_id, line in enumerate(f, start=1): |
| 327 | line = line.strip() |
| 328 | if not line: |
| 329 | continue |
| 330 | try: |
| 331 | record = json.loads(line) |
| 332 | except json.JSONDecodeError as e: |
| 333 | continue |
| 334 | self.traces[usr_id].append(record) |
| 335 | |
| 336 | self.continue_flag = True |
| 337 | |
| 338 | def _create_user_session(self): |
| 339 | self.user_id += 1 |
nothing calls this directly
no outgoing calls
no test coverage detected