this repo has no description
feeds: do manual checkpointing
+19
+9
feeds/popular.py
+9
feeds/popular.py
···
1
1
import os
2
+
import sys
2
3
import math
3
4
import sqlite3
4
5
···
16
17
self.db_cnx.executescript(
17
18
"pragma journal_mode = WAL;"
18
19
"pragma synchronous = OFF;"
20
+
"pragma wal_autocheckpoint = 0;"
19
21
"create table if not exists posts (uri text, create_ts timestamp, update_ts timestamp, temperature int);"
20
22
"create unique index if not exists uri_idx on posts(uri);"
21
23
)
···
43
45
44
46
self.cleanup_checkpoint += 1
45
47
if self.cleanup_checkpoint % 1000 == 0:
48
+
sys.stdout.write('popular: running cleanup checkpoint\n')
49
+
sys.stdout.flush()
50
+
46
51
with self.db_cnx:
47
52
self.db_cnx.execute(
48
53
"delete from posts where temperature * exp( -1 * ( ( strftime( '%s', 'now' ) - strftime( '%s', create_ts ) ) / 1800.0 ) ) < 1.0 and strftime( '%s', create_ts ) < strftime( '%s', 'now', '-15 minutes' )"
49
54
)
55
+
56
+
self.db_cnx.execute(
57
+
"pragma wal_checkpoint(TRUNCATE)"
58
+
)
50
59
51
60
def serve(self, limit, offset, langs):
52
61
cur = self.db_cnx.execute((
+10
feeds/rapidfire.py
+10
feeds/rapidfire.py
···
1
1
import os
2
+
import sys
2
3
import sqlite3
3
4
4
5
MAX_TEXT_LENGTH = 140
···
15
16
self.db_cnx.executescript(
16
17
"pragma journal_mode = WAL;"
17
18
"pragma synchronous = OFF;"
19
+
"pragma wal_autocheckpoint = 0;"
18
20
"create table if not exists posts (uri text, create_ts timestamp, lang text);"
19
21
"create index if not exists create_ts_idx on posts(create_ts);"
20
22
)
21
23
24
+
self.checkpoint = 0
25
+
22
26
def process(self, commit):
23
27
op = commit['op']
24
28
if op['action'] != 'create':
···
40
44
path = op['path']
41
45
post_uri = f'at://{repo}/{path}'
42
46
ts = record['createdAt']
47
+
self.checkpoint += 1
43
48
44
49
with self.db_cnx:
45
50
langs = record.get('langs') or ['']
···
52
57
self.db_cnx.execute(
53
58
"delete from posts where strftime('%s', create_ts) < strftime('%s', 'now', '-15 minutes')"
54
59
)
60
+
61
+
if self.checkpoint % 100 == 0:
62
+
sys.stdout.write('rapidfire: checkpoint\n')
63
+
sys.stdout.flush()
64
+
self.db_cnx.execute("pragma wal_checkpoint(TRUNCATE)")
55
65
56
66
def serve(self, limit, offset, langs):
57
67
if '*' in langs: