检测事件流中的超时

Question

我正试图为这个面试问题找到一个解决方案：

您有一个RPC请求流进入正在被记录的服务器。每个日志条目都是表单id，时间戳，键入(“开始”或“结束”)。给定一个日志条目序列和一个超时值，如果RPC调用超时，您需要尽早计算出来(例如，一旦检测到这种情况，就打印消息)。例子: Timeout =3 id - time - type 0-0- Start 1-1- Start 0-2- End 2-6- Start #计算出id 1在时间6，1-7结束时超时。

我相信我的解决方案是O(NlogN)，有什么改进的方法吗？

from sortedcontainers import SortedList


def process_log(log: list[list], timeout=3):
    rpc_ids = {}
    rpcs = SortedList()

    for rpc_id, timestamp, action in log:
        if action == 'Start':
            rpcs.add([timestamp, rpc_id])
            rpc_ids[rpc_id] = timestamp
        else:
            if rpc_id in rpc_ids:
                entry = [rpc_ids[rpc_id], rpc_id]
                if timestamp - rpc_ids[rpc_id] > timeout:
                    report_rpc(rpc_id, rpc_ids[rpc_id], timestamp)
                del rpc_ids[rpc_id]
                rpcs.remove(entry)

        idx = rpcs.bisect_left([timestamp-timeout, float('inf')])

        if idx > 0:
            for i in range(idx):
                start_time, rpc_id = rpcs[i]
                report_rpc(rpc_id, start_time, timestamp)
                del rpc_ids[rpc_id]
            del rpcs[:idx]


def report_rpc(rpc_id, start_time, timestamp):
    print(f'RPC #{rpc_id} started at {start_time} has timed out (@{timestamp})')


process_log([  # RPC #1 times out at timestamp 6
    [0, 0, 'Start'],
    [1, 1, 'Start'],
    [0, 2, 'End'],
    [2, 6, 'Start'],
    [1, 7, 'End'],
])

Answer 1

下面是一个只使用字典的O(n)解决方案：

from dataclasses import dataclass


@dataclass(frozen=True, slots=True)
class LogEntry:
    rpc_id: int
    timestamp: int
    action: str


class LogProcessor:
    def __init__(self, timeout: int):
        self.entries: dict[int, LogEntry] = {}
        self.timeout = timeout

    def process_log_entry(self, entry: LogEntry) -> list[LogEntry]:
        if entry.action == 'Start':
            self.entries[entry.rpc_id] = entry
        elif entry.rpc_id in self.entries and entry.timestamp - self.entries[entry.rpc_id].timestamp <= self.timeout:
            del self.entries[entry.rpc_id]

        timed_out = []

        for e in self.entries.values():
            if entry.timestamp - e.timestamp <= self.timeout:
                break
            timed_out.append(e)

        for e in timed_out:
            del self.entries[e.rpc_id]

        return timed_out


log_processor = LogProcessor(timeout=3)

assert log_processor.process_log_entry(LogEntry(1, 0, 'Start')) == []
assert log_processor.process_log_entry(LogEntry(2, 1, 'Start')) == []
assert log_processor.process_log_entry(LogEntry(1, 2, 'End')) == []
assert log_processor.process_log_entry(LogEntry(3, 6, 'Start')) == [LogEntry(2, 1, 'Start')]

由于Python字典是按插入顺序排列的，因此我们可以通过从字典开始迭代一次来检查每次调用process_log_entry()时是否有超时条目。因为我们在迭代之后立即删除超时条目，所以总的时间复杂度仍然是O(n)。

还对代码进行了重构，以使用日志条目的数据块和日志处理器的类，这两个类应该更具可读性和可维护性。