Skip to content

naneos.cloud.backlog

Data waiting for the upload, kept in RAM, capped in bytes.

While the network is down the manager keeps what it gathers here and the sender takes it out again, oldest first. The frames are the ones the upload reads (see upload.prepare_frames: one row per second, only the columns that go on the wire), so they are small, and they stay frames so that the sender can merge them into as large a request as the backend is used to.

Snapshots are merged into chunks of up to CHUNK_SECONDS of data. A pandas frame has a fixed overhead of about 5 KB, which is more than the data of a 30 s snapshot of one device: kept one by one, a day would cost about 80 % on top of the data, in chunks of 600 s less than 10 %.

Chunk dataclass

The frames of up to CHUNK_SECONDS of data, indexed by unix seconds.

Source code in src/naneos/cloud/backlog.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
@dataclass
class Chunk:
    """The frames of up to CHUNK_SECONDS of data, indexed by unix seconds."""

    frames: dict[int, pd.DataFrame]
    first: int  # unix seconds of the first row of any device
    last: int
    snapshots: int = 1  # how many snapshots were merged into it
    sealed: bool = False  # taken once: nothing is merged into it any more
    nbytes: int = field(init=False, default=0)

    def __post_init__(self) -> None:
        self.nbytes = _size_of(self.frames)

    @property
    def seconds(self) -> int:
        return self.last - self.first + 1

    @property
    def rows(self) -> int:
        """The rows of all devices: what the backend has to write for this chunk."""
        return sum(len(df) for df in self.frames.values())

rows property

The rows of all devices: what the backend has to write for this chunk.

Eviction

Bases: NamedTuple

What a full buffer threw away.

Source code in src/naneos/cloud/backlog.py 
30
31
32
33
34
class Eviction(NamedTuple):
    """What a full buffer threw away."""

    snapshots: int
    seconds: int  # of data, counted per chunk (devices in parallel are not added up)

UploadBacklog

A FIFO of chunks, thread-safe: the manager adds, the sender takes.

Parameters:

Name Type Description Default
max_bytes int

the cap. It is soft by one chunk: the newest chunk is never evicted, so a cap below the size of one chunk still keeps the latest data.

 required
chunk_seconds int

the most data merged into one chunk.

 CHUNK_SECONDS
Source code in src/naneos/cloud/backlog.py 
 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
class UploadBacklog:
    """A FIFO of chunks, thread-safe: the manager adds, the sender takes.

    Args:
        max_bytes: the cap. It is soft by one chunk: the newest chunk is never
            evicted, so a cap below the size of one chunk still keeps the latest data.
        chunk_seconds: the most data merged into one chunk.
    """

    def __init__(self, max_bytes: int, chunk_seconds: int = CHUNK_SECONDS) -> None:
        self._max_bytes = max_bytes
        self._chunk_seconds = chunk_seconds
        self._chunks: deque[Chunk] = deque()
        self._bytes = 0
        self._lock = threading.Lock()

    def add(self, frames: dict[int, pd.DataFrame]) -> Eviction | None:
        """Keep the frames of one snapshot. Returns what had to go to stay under the cap."""
        frames = {sn: df for sn, df in frames.items() if not df.empty}
        if not frames:
            return None
        first = min(int(df.index.min()) for df in frames.values())
        last = max(int(df.index.max()) for df in frames.values())

        with self._lock:
            tail = self._chunks[-1] if self._chunks else None
            if tail is not None and not tail.sealed and last - tail.first < self._chunk_seconds:
                self._bytes -= tail.nbytes
                self._merge(tail, frames, last)
                self._bytes += tail.nbytes
            else:
                chunk = Chunk(frames, first, last)
                self._chunks.append(chunk)
                self._bytes += chunk.nbytes
            return self._evict()

    def take(self, max_seconds: int | None = None, max_rows: int | None = None) -> Chunk | None:
        """Remove and return the oldest chunk, or the first part of it.

        The part is at most `max_seconds` long and holds about `max_rows` rows (all
        devices together; the rows are taken to be spread evenly over the time).
        What the caller could not send goes back with restore().
        """
        with self._lock:
            if not self._chunks:
                return None
            chunk = self._chunks.popleft()
            self._bytes -= chunk.nbytes
            chunk.sealed = True

            limit = chunk.seconds if max_seconds is None else max_seconds
            if max_rows is not None and chunk.rows > max_rows:
                limit = min(limit, max(1, int(chunk.seconds * max_rows / chunk.rows)))
            if chunk.seconds > limit:
                chunk, rest = _split(chunk, chunk.first + limit)
                self._chunks.appendleft(rest)
                self._bytes += rest.nbytes
            return chunk

    def restore(self, chunk: Chunk) -> Eviction | None:
        """Put a chunk that could not be sent back at the head."""
        with self._lock:
            chunk.sealed = True
            self._chunks.appendleft(chunk)
            self._bytes += chunk.nbytes
            return self._evict()

    @property
    def size_bytes(self) -> int:
        return self._bytes

    @property
    def snapshots(self) -> int:
        """The number of snapshots waiting."""
        with self._lock:
            return sum(chunk.snapshots for chunk in self._chunks)

    @property
    def seconds(self) -> int:
        """Seconds of data waiting, counted per chunk."""
        with self._lock:
            return sum(chunk.seconds for chunk in self._chunks)

    def __len__(self) -> int:
        return len(self._chunks)

    @staticmethod
    def _merge(tail: Chunk, frames: dict[int, pd.DataFrame], last: int) -> None:
        for sn, df in frames.items():
            known = tail.frames.get(sn)
            if known is None:
                tail.frames[sn] = df
            else:
                merged = pd.concat([known, df])
                # A 10 or 100 Hz device can put rows of two snapshots into the same second.
                tail.frames[sn] = aggregate_duplicate_index(merged)
        tail.last = max(tail.last, last)
        tail.snapshots += 1
        tail.nbytes = _size_of(tail.frames)

    def _evict(self) -> Eviction | None:
        """Drop the oldest chunks while over the cap; the caller holds the lock."""
        snapshots = seconds = 0
        while self._bytes > self._max_bytes and len(self._chunks) > 1:
            dropped = self._chunks.popleft()
            self._bytes -= dropped.nbytes
            snapshots += dropped.snapshots
            seconds += dropped.seconds
        return Eviction(snapshots, seconds) if snapshots else None

seconds property

Seconds of data waiting, counted per chunk.

snapshots property

The number of snapshots waiting.

add(frames)

Keep the frames of one snapshot. Returns what had to go to stay under the cap.

Source code in src/naneos/cloud/backlog.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def add(self, frames: dict[int, pd.DataFrame]) -> Eviction | None:
    """Keep the frames of one snapshot. Returns what had to go to stay under the cap."""
    frames = {sn: df for sn, df in frames.items() if not df.empty}
    if not frames:
        return None
    first = min(int(df.index.min()) for df in frames.values())
    last = max(int(df.index.max()) for df in frames.values())

    with self._lock:
        tail = self._chunks[-1] if self._chunks else None
        if tail is not None and not tail.sealed and last - tail.first < self._chunk_seconds:
            self._bytes -= tail.nbytes
            self._merge(tail, frames, last)
            self._bytes += tail.nbytes
        else:
            chunk = Chunk(frames, first, last)
            self._chunks.append(chunk)
            self._bytes += chunk.nbytes
        return self._evict()

restore(chunk)

Put a chunk that could not be sent back at the head.

Source code in src/naneos/cloud/backlog.py 
126
127
128
129
130
131
132
def restore(self, chunk: Chunk) -> Eviction | None:
    """Put a chunk that could not be sent back at the head."""
    with self._lock:
        chunk.sealed = True
        self._chunks.appendleft(chunk)
        self._bytes += chunk.nbytes
        return self._evict()

take(max_seconds=None, max_rows=None)

Remove and return the oldest chunk, or the first part of it.

The part is at most max_seconds long and holds about max_rows rows (all devices together; the rows are taken to be spread evenly over the time). What the caller could not send goes back with restore().

Source code in src/naneos/cloud/backlog.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def take(self, max_seconds: int | None = None, max_rows: int | None = None) -> Chunk | None:
    """Remove and return the oldest chunk, or the first part of it.

    The part is at most `max_seconds` long and holds about `max_rows` rows (all
    devices together; the rows are taken to be spread evenly over the time).
    What the caller could not send goes back with restore().
    """
    with self._lock:
        if not self._chunks:
            return None
        chunk = self._chunks.popleft()
        self._bytes -= chunk.nbytes
        chunk.sealed = True

        limit = chunk.seconds if max_seconds is None else max_seconds
        if max_rows is not None and chunk.rows > max_rows:
            limit = min(limit, max(1, int(chunk.seconds * max_rows / chunk.rows)))
        if chunk.seconds > limit:
            chunk, rest = _split(chunk, chunk.first + limit)
            self._chunks.appendleft(rest)
            self._bytes += rest.nbytes
        return chunk