Compare commits
1 Commits
main
...
erikj/fix_
| Author | SHA1 | Date | |
|---|---|---|---|
|
29037e3ca1 |
@@ -1,46 +0,0 @@
|
||||
name: Build and push docker images
|
||||
|
||||
on:
|
||||
push:
|
||||
tags: ["v*"]
|
||||
branches: [ main ]
|
||||
workflow_dispatch:
|
||||
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- name: Set up Docker Buildx
|
||||
uses: docker/setup-buildx-action@v2
|
||||
|
||||
- name: Log in to Gitea Container Registry
|
||||
uses: docker/login-action@v2
|
||||
with:
|
||||
registry: git.yongyuancv.cn
|
||||
username: ${{ gitea.repository_owner }}
|
||||
password: ${{ secrets.GITEA_TOKEN }}
|
||||
|
||||
- name: Calculate docker image tag
|
||||
id: set-tag
|
||||
uses: docker/metadata-action@master
|
||||
with:
|
||||
images: |
|
||||
git.yongyuancv.cn/${{ gitea.repository }}
|
||||
git.yongyuancv.cn/heimoshuiyu/${{ gitea.event.repository.name }}
|
||||
flavor: |
|
||||
latest=false
|
||||
tags: |
|
||||
type=raw,value=latest,enable=${{ gitea.ref == 'refs/heads/main' }}
|
||||
type=sha,prefix=,format=long
|
||||
type=semver,pattern=v{{version}}
|
||||
type=semver,pattern=v{{major}}.{{minor}}
|
||||
|
||||
- name: Build and push all platforms
|
||||
uses: docker/build-push-action@v4
|
||||
with:
|
||||
push: true
|
||||
labels: "gitsha1=${{ gitea.sha }}"
|
||||
tags: "${{ steps.set-tag.outputs.tags }}"
|
||||
platforms: linux/amd64,linux/arm64
|
||||
cache-from: type=registry,ref=git.yongyuancv.cn/${{ gitea.repository }}:buildcache
|
||||
cache-to: type=registry,ref=git.yongyuancv.cn/${{ gitea.repository }}:buildcache,mode=max
|
||||
58
.github/workflows/docker.yaml
vendored
58
.github/workflows/docker.yaml
vendored
@@ -1,58 +0,0 @@
|
||||
# GitHub actions workflow which builds and publishes the docker images.
|
||||
|
||||
name: Build and push docker images
|
||||
|
||||
on:
|
||||
push:
|
||||
tags: ["v*"]
|
||||
branches: [ main ]
|
||||
workflow_dispatch:
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
packages: write
|
||||
|
||||
jobs:
|
||||
build:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- name: Set up Docker Buildx
|
||||
uses: docker/setup-buildx-action@v2
|
||||
|
||||
- name: Log in to DockerHub
|
||||
uses: docker/login-action@v2
|
||||
with:
|
||||
username: ${{ secrets.DOCKER_HUB_USERNAME }}
|
||||
password: ${{ secrets.DOCKER_HUB_TOKEN }}
|
||||
|
||||
- name: Log in to GHCR
|
||||
uses: docker/login-action@v2
|
||||
with:
|
||||
registry: ghcr.io
|
||||
username: ${{ github.repository_owner }}
|
||||
password: ${{ secrets.GITHUB_TOKEN }}
|
||||
|
||||
- name: Calculate docker image tag
|
||||
id: set-tag
|
||||
uses: docker/metadata-action@master
|
||||
with:
|
||||
images: |
|
||||
ghcr.io/${{ github.repository }}
|
||||
docker.io/${{ secrets.DOCKER_HUB_USERNAME }}/${{ github.event.repository.name }}
|
||||
flavor: |
|
||||
latest=false
|
||||
tags: |
|
||||
type=raw,value=latest,enable=${{ github.ref == 'refs/heads/main' }}
|
||||
type=sha,prefix=,format=long
|
||||
type=semver,pattern=v{{version}}
|
||||
type=semver,pattern=v{{major}}.{{minor}}
|
||||
|
||||
- name: Build and push all platforms
|
||||
uses: docker/build-push-action@v4
|
||||
with:
|
||||
push: true
|
||||
labels: "gitsha1=${{ github.sha }}"
|
||||
tags: "${{ steps.set-tag.outputs.tags }}"
|
||||
platforms: linux/amd64,linux/arm64
|
||||
cache-from: type=registry,ref=ghcr.io/${{ github.repository }}:buildcache
|
||||
cache-to: type=registry,ref=ghcr.io/${{ github.repository }}:buildcache,mode=max
|
||||
28
.github/workflows/triage_incoming.yml
vendored
28
.github/workflows/triage_incoming.yml
vendored
@@ -1,28 +0,0 @@
|
||||
name: Move new issues into the issue triage board
|
||||
|
||||
on:
|
||||
issues:
|
||||
types: [ opened ]
|
||||
|
||||
jobs:
|
||||
add_new_issues:
|
||||
name: Add new issues to the triage board
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- uses: octokit/graphql-action@v2.x
|
||||
id: add_to_project
|
||||
with:
|
||||
headers: '{"GraphQL-Features": "projects_next_graphql"}'
|
||||
query: |
|
||||
mutation add_to_project($projectid:ID!,$contentid:ID!) {
|
||||
addProjectV2ItemById(input: {projectId: $projectid contentId: $contentid}) {
|
||||
item {
|
||||
id
|
||||
}
|
||||
}
|
||||
}
|
||||
projectid: ${{ env.PROJECT_ID }}
|
||||
contentid: ${{ github.event.issue.node_id }}
|
||||
env:
|
||||
PROJECT_ID: "PVT_kwDOAIB0Bs4AFDdZ"
|
||||
GITHUB_TOKEN: ${{ secrets.ELEMENT_BOT_TOKEN }}
|
||||
44
.github/workflows/triage_labelled.yml
vendored
44
.github/workflows/triage_labelled.yml
vendored
@@ -1,44 +0,0 @@
|
||||
name: Move labelled issues to correct projects
|
||||
|
||||
on:
|
||||
issues:
|
||||
types: [ labeled ]
|
||||
|
||||
jobs:
|
||||
move_needs_info:
|
||||
name: Move X-Needs-Info on the triage board
|
||||
runs-on: ubuntu-latest
|
||||
if: >
|
||||
contains(github.event.issue.labels.*.name, 'X-Needs-Info')
|
||||
steps:
|
||||
- uses: actions/add-to-project@main
|
||||
id: add_project
|
||||
with:
|
||||
project-url: "https://github.com/orgs/matrix-org/projects/67"
|
||||
github-token: ${{ secrets.ELEMENT_BOT_TOKEN }}
|
||||
- name: Set status
|
||||
env:
|
||||
GITHUB_TOKEN: ${{ secrets.ELEMENT_BOT_TOKEN }}
|
||||
run: |
|
||||
gh api graphql -f query='
|
||||
mutation(
|
||||
$project: ID!
|
||||
$item: ID!
|
||||
$fieldid: ID!
|
||||
$columnid: String!
|
||||
) {
|
||||
updateProjectV2ItemFieldValue(
|
||||
input: {
|
||||
projectId: $project
|
||||
itemId: $item
|
||||
fieldId: $fieldid
|
||||
value: {
|
||||
singleSelectOptionId: $columnid
|
||||
}
|
||||
}
|
||||
) {
|
||||
projectV2Item {
|
||||
id
|
||||
}
|
||||
}
|
||||
}' -f project="PVT_kwDOAIB0Bs4AFDdZ" -f item=${{ steps.add_project.outputs.itemId }} -f fieldid="PVTSSF_lADOAIB0Bs4AFDdZzgC6ZA4" -f columnid=ba22e43c --silent
|
||||
3
.vscode/settings.json
vendored
Normal file
3
.vscode/settings.json
vendored
Normal file
@@ -0,0 +1,3 @@
|
||||
{
|
||||
"rust-analyzer.checkOnSave.command": "clippy"
|
||||
}
|
||||
@@ -1,80 +0,0 @@
|
||||
# Contributing
|
||||
|
||||
## Sign off
|
||||
|
||||
In order to have a concrete record that your contribution is intentional
|
||||
and you agree to license it under the same terms as the project's license, we've adopted the
|
||||
same lightweight approach that the Linux Kernel
|
||||
[submitting patches process](
|
||||
https://www.kernel.org/doc/html/latest/process/submitting-patches.html#sign-your-work-the-developer-s-certificate-of-origin>),
|
||||
[Docker](https://github.com/docker/docker/blob/master/CONTRIBUTING.md), and many other
|
||||
projects use: the DCO ([Developer Certificate of Origin](http://developercertificate.org/)).
|
||||
This is a simple declaration that you wrote
|
||||
the contribution or otherwise have the right to contribute it to Matrix:
|
||||
|
||||
```
|
||||
Developer Certificate of Origin
|
||||
Version 1.1
|
||||
|
||||
Copyright (C) 2004, 2006 The Linux Foundation and its contributors.
|
||||
660 York Street, Suite 102,
|
||||
San Francisco, CA 94110 USA
|
||||
|
||||
Everyone is permitted to copy and distribute verbatim copies of this
|
||||
license document, but changing it is not allowed.
|
||||
|
||||
Developer's Certificate of Origin 1.1
|
||||
|
||||
By making a contribution to this project, I certify that:
|
||||
|
||||
(a) The contribution was created in whole or in part by me and I
|
||||
have the right to submit it under the open source license
|
||||
indicated in the file; or
|
||||
|
||||
(b) The contribution is based upon previous work that, to the best
|
||||
of my knowledge, is covered under an appropriate open source
|
||||
license and I have the right under that license to submit that
|
||||
work with modifications, whether created in whole or in part
|
||||
by me, under the same open source license (unless I am
|
||||
permitted to submit under a different license), as indicated
|
||||
in the file; or
|
||||
|
||||
(c) The contribution was provided directly to me by some other
|
||||
person who certified (a), (b) or (c) and I have not modified
|
||||
it.
|
||||
|
||||
(d) I understand and agree that this project and the contribution
|
||||
are public and that a record of the contribution (including all
|
||||
personal information I submit with it, including my sign-off) is
|
||||
maintained indefinitely and may be redistributed consistent with
|
||||
this project or the open source license(s) involved.
|
||||
```
|
||||
|
||||
If you agree to this for your contribution, then all that's needed is to
|
||||
include the line in your commit or pull request comment:
|
||||
|
||||
```
|
||||
Signed-off-by: Your Name <your@email.example.org>
|
||||
```
|
||||
|
||||
We accept contributions under a legally identifiable name, such as
|
||||
your name on government documentation or common-law names (names
|
||||
claimed by legitimate usage or repute). Unfortunately, we cannot
|
||||
accept anonymous contributions at this time.
|
||||
|
||||
Git allows you to add this signoff automatically when using the `-s`
|
||||
flag to `git commit`, which uses the name and email set in your
|
||||
`user.name` and `user.email` git configs.
|
||||
|
||||
### Private Sign off
|
||||
|
||||
If you would like to provide your legal name privately to the Matrix.org
|
||||
Foundation (instead of in a public commit or comment), you can do so
|
||||
by emailing your legal name and a link to the pull request to
|
||||
[dco@matrix.org](mailto:dco@matrix.org?subject=Private%20sign%20off).
|
||||
It helps to include "sign off" or similar in the subject line. You will then
|
||||
be instructed further.
|
||||
|
||||
Once private sign off is complete, doing so for future contributions will not
|
||||
be required.
|
||||
|
||||
1230
Cargo.lock
generated
1230
Cargo.lock
generated
File diff suppressed because it is too large
Load Diff
43
Cargo.toml
43
Cargo.toml
@@ -8,21 +8,19 @@ name = "synapse_compress_state"
|
||||
version = "0.1.0"
|
||||
edition = "2018"
|
||||
|
||||
[[bin]]
|
||||
name = "synapse_compress_state"
|
||||
required-features = ["clap"]
|
||||
|
||||
[dependencies]
|
||||
indicatif = "0.17.6"
|
||||
openssl = "0.10.60"
|
||||
postgres = "0.19.7"
|
||||
clap = "2.33.0"
|
||||
indicatif = "0.16.0"
|
||||
openssl = "0.10.32"
|
||||
postgres = "0.19.0"
|
||||
postgres-openssl = "0.5.0"
|
||||
rand = "0.8.5"
|
||||
rayon = "1.7.0"
|
||||
string_cache = "0.8.7"
|
||||
env_logger = "0.10.0"
|
||||
log = "0.4.20"
|
||||
log-panics = "2.1.0"
|
||||
rand = "0.8.0"
|
||||
rayon = "1.3.0"
|
||||
string_cache = "0.8.0"
|
||||
env_logger = "0.9.0"
|
||||
log = "0.4.14"
|
||||
pyo3-log = "0.4.0"
|
||||
log-panics = "2.0.0"
|
||||
|
||||
[dependencies.state-map]
|
||||
git = "https://github.com/matrix-org/rust-matrix-state-map"
|
||||
@@ -31,26 +29,15 @@ git = "https://github.com/matrix-org/rust-matrix-state-map"
|
||||
[lib]
|
||||
crate-type = ["cdylib", "rlib"]
|
||||
|
||||
[dependencies.clap]
|
||||
version = "4.4.2"
|
||||
features = ["cargo"]
|
||||
optional = true
|
||||
|
||||
[dependencies.pyo3]
|
||||
version = "0.19.2"
|
||||
features = ["extension-module"]
|
||||
optional = true
|
||||
|
||||
[dependencies.pyo3-log]
|
||||
version = "0.8.3"
|
||||
optional = true
|
||||
version = "0.14.1"
|
||||
features = ["extension-module","abi3-py36"]
|
||||
|
||||
[dependencies.tikv-jemallocator]
|
||||
version = "0.5.4"
|
||||
version = "0.4.1"
|
||||
optional = true
|
||||
|
||||
[features]
|
||||
default = ["clap", "jemalloc"]
|
||||
default = ["jemalloc"]
|
||||
jemalloc = ["tikv-jemallocator"]
|
||||
no-progress-bars = []
|
||||
pyo3 = ["dep:pyo3", "dep:pyo3-log"]
|
||||
|
||||
56
Dockerfile
56
Dockerfile
@@ -1,54 +1,22 @@
|
||||
# This uses the multi-stage build feature of Docker to build the binaries for multiple architectures without QEMU.
|
||||
# The first stage is responsible for building binaries for all the supported architectures (amd64 and arm64), and the
|
||||
# second stage only copies the binaries for the target architecture.
|
||||
# We leverage Zig and cargo-zigbuild for providing a cross-compilation-capable C compiler and linker.
|
||||
FROM rust:alpine AS builder
|
||||
|
||||
ARG RUSTC_VERSION=1.72.0
|
||||
ARG ZIG_VERSION=0.11.0
|
||||
ARG CARGO_ZIGBUILD_VERSION=0.17.1
|
||||
RUN apk add python3 musl-dev pkgconfig openssl-dev make
|
||||
|
||||
FROM --platform=${BUILDPLATFORM} docker.io/rust:${RUSTC_VERSION} AS builder
|
||||
|
||||
# Install cargo-zigbuild for cross-compilation
|
||||
ARG CARGO_ZIGBUILD_VERSION
|
||||
RUN cargo install --locked cargo-zigbuild@=${CARGO_ZIGBUILD_VERSION}
|
||||
|
||||
# Download zig compiler for cross-compilation
|
||||
ARG ZIG_VERSION
|
||||
RUN curl -L "https://ziglang.org/download/${ZIG_VERSION}/zig-linux-$(uname -m)-${ZIG_VERSION}.tar.xz" | tar -J -x -C /usr/local && \
|
||||
ln -s "/usr/local/zig-linux-$(uname -m)-${ZIG_VERSION}/zig" /usr/local/bin/zig
|
||||
|
||||
# Install all cross-compilation targets
|
||||
ARG RUSTC_VERSION
|
||||
RUN rustup target add \
|
||||
--toolchain "${RUSTC_VERSION}" \
|
||||
x86_64-unknown-linux-musl \
|
||||
aarch64-unknown-linux-musl
|
||||
ENV RUSTFLAGS="-C target-feature=-crt-static"
|
||||
|
||||
WORKDIR /opt/synapse-compressor/
|
||||
|
||||
COPY . .
|
||||
|
||||
# Build for all targets
|
||||
RUN cargo zigbuild \
|
||||
--release \
|
||||
--workspace \
|
||||
--bins \
|
||||
--features "openssl/vendored" \
|
||||
--target aarch64-unknown-linux-musl \
|
||||
--target x86_64-unknown-linux-musl
|
||||
RUN cargo build
|
||||
|
||||
# Move the binaries in a separate folder per architecture, so we can copy them using the TARGETARCH build arg
|
||||
RUN mkdir -p /opt/binaries/amd64 /opt/binaries/arm64
|
||||
RUN mv target/x86_64-unknown-linux-musl/release/synapse_compress_state \
|
||||
target/x86_64-unknown-linux-musl/release/synapse_auto_compressor \
|
||||
/opt/binaries/amd64
|
||||
RUN mv target/aarch64-unknown-linux-musl/release/synapse_compress_state \
|
||||
target/aarch64-unknown-linux-musl/release/synapse_auto_compressor \
|
||||
/opt/binaries/arm64
|
||||
WORKDIR /opt/synapse-compressor/synapse_auto_compressor/
|
||||
|
||||
FROM --platform=${TARGETPLATFORM} docker.io/alpine
|
||||
RUN cargo build
|
||||
|
||||
ARG TARGETARCH
|
||||
FROM alpine
|
||||
|
||||
COPY --from=builder /opt/binaries/${TARGETARCH}/synapse_compress_state /usr/local/bin/synapse_compress_state
|
||||
COPY --from=builder /opt/binaries/${TARGETARCH}/synapse_auto_compressor /usr/local/bin/synapse_auto_compressor
|
||||
RUN apk add --no-cache libgcc
|
||||
|
||||
COPY --from=builder /opt/synapse-compressor/target/debug/synapse_compress_state /usr/local/bin/synapse_compress_state
|
||||
COPY --from=builder /opt/synapse-compressor/target/debug/synapse_auto_compressor /usr/local/bin/synapse_auto_compressor
|
||||
|
||||
13
README.md
13
README.md
@@ -26,9 +26,6 @@ periodically.
|
||||
This tool requires `cargo` to be installed. See https://www.rust-lang.org/tools/install
|
||||
for instructions on how to do this.
|
||||
|
||||
This project follows the deprecation policy of [Synapse](https://matrix-org.github.io/synapse/latest/deprecation_policy.html)
|
||||
on Rust and will assume a recent stable version of Rust and the ability to fetch a more recent one if necessary.
|
||||
|
||||
To build `synapse_auto_compressor`, clone this repository and navigate to the
|
||||
`synapse_auto_compressor/` subdirectory. Then execute `cargo build`.
|
||||
|
||||
@@ -36,17 +33,9 @@ This will create an executable and store it in
|
||||
`synapse_auto_compressor/target/debug/synapse_auto_compressor`.
|
||||
|
||||
## Example usage
|
||||
|
||||
Compress 100 chunks of size 500 in a remote PostgreSQL database:
|
||||
```
|
||||
$ synapse_auto_compressor -p postgresql://user:pass@localhost/synapse -c 500 -n 100
|
||||
```
|
||||
|
||||
Compress 100 chunks of size 500 using local PostgreSQL socket:
|
||||
```
|
||||
$ sudo -u postgres synapse_auto_compressor -p "user=postgres dbname=matrix-synapse host=/var/run/postgresql" -c 500 -n 100
|
||||
```
|
||||
|
||||
## Running Options
|
||||
|
||||
- -p [POSTGRES_LOCATION] **Required**
|
||||
@@ -67,7 +56,7 @@ in) then the entire chunk is skipped.
|
||||
*CHUNKS_TO_COMPRESS* chunks of size *CHUNK_SIZE* will be compressed. The higher this
|
||||
number is set to, the longer the compressor will run for.
|
||||
|
||||
- -l [LEVELS]
|
||||
- -d [LEVELS]
|
||||
Sizes of each new level in the compression algorithm, as a comma-separated list.
|
||||
The first entry in the list is for the lowest, most granular level, with each
|
||||
subsequent entry being for the next highest level. The number of entries in the
|
||||
|
||||
@@ -6,16 +6,16 @@ edition = "2018"
|
||||
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
|
||||
|
||||
[dependencies]
|
||||
string_cache = "0.8.7"
|
||||
serial_test = "2.0.0"
|
||||
openssl = "0.10.60"
|
||||
postgres = "0.19.7"
|
||||
string_cache = "0.8.0"
|
||||
serial_test = "0.5.1"
|
||||
openssl = "0.10.32"
|
||||
postgres = "0.19.0"
|
||||
postgres-openssl = "0.5.0"
|
||||
rand = "0.8.5"
|
||||
rand = "0.8.0"
|
||||
synapse_compress_state = { path = "../", features = ["no-progress-bars"] }
|
||||
synapse_auto_compressor = { path = "../synapse_auto_compressor/" }
|
||||
env_logger = "0.10.0"
|
||||
log = "0.4.20"
|
||||
env_logger = "0.9.0"
|
||||
log = "0.4.14"
|
||||
|
||||
[dependencies.state-map]
|
||||
git = "https://github.com/matrix-org/rust-matrix-state-map"
|
||||
|
||||
@@ -4,12 +4,7 @@ use postgres::{fallible_iterator::FallibleIterator, Client};
|
||||
use postgres_openssl::MakeTlsConnector;
|
||||
use rand::{distributions::Alphanumeric, thread_rng, Rng};
|
||||
use state_map::StateMap;
|
||||
use std::{
|
||||
borrow::Cow,
|
||||
collections::BTreeMap,
|
||||
env,
|
||||
fmt::{self, Write as _},
|
||||
};
|
||||
use std::{borrow::Cow, collections::BTreeMap, env, fmt};
|
||||
use string_cache::DefaultAtom as Atom;
|
||||
|
||||
use synapse_compress_state::StateGroupEntry;
|
||||
@@ -28,48 +23,47 @@ pub fn add_contents_to_database(room_id: &str, state_group_map: &BTreeMap<i64, S
|
||||
let mut client = Client::connect(DB_URL, connector).unwrap();
|
||||
|
||||
// build up the query
|
||||
let mut sql = String::new();
|
||||
|
||||
let room_id = PGEscape(room_id);
|
||||
let event_id = PGEscape("left_blank");
|
||||
let mut sql = "".to_string();
|
||||
|
||||
for (sg, entry) in state_group_map {
|
||||
// create the entry for state_groups
|
||||
writeln!(
|
||||
sql,
|
||||
"INSERT INTO state_groups (id, room_id, event_id) \
|
||||
VALUES ({sg}, {room_id}, {event_id});",
|
||||
)
|
||||
.expect("Writing to a String cannot fail");
|
||||
sql.push_str(&format!(
|
||||
"INSERT INTO state_groups (id, room_id, event_id) VALUES ({},{},{});\n",
|
||||
sg,
|
||||
PGEscape(room_id),
|
||||
PGEscape("left_blank")
|
||||
));
|
||||
|
||||
// create the entry in state_group_edges IF exists
|
||||
if let Some(prev_sg) = entry.prev_state_group {
|
||||
writeln!(
|
||||
sql,
|
||||
"INSERT INTO state_group_edges (state_group, prev_state_group) \
|
||||
VALUES ({sg}, {prev_sg});",
|
||||
)
|
||||
.unwrap();
|
||||
sql.push_str(&format!(
|
||||
"INSERT INTO state_group_edges (state_group, prev_state_group) VALUES ({}, {});\n",
|
||||
sg, prev_sg
|
||||
));
|
||||
}
|
||||
|
||||
// write entry for each row in delta
|
||||
if !entry.state_map.is_empty() {
|
||||
sql.push_str(
|
||||
"INSERT INTO state_groups_state \
|
||||
(state_group, room_id, type, state_key, event_id) \
|
||||
VALUES\n",
|
||||
);
|
||||
sql.push_str("INSERT INTO state_groups_state (state_group, room_id, type, state_key, event_id) VALUES");
|
||||
|
||||
let mut first = true;
|
||||
for ((t, s), e) in entry.state_map.iter() {
|
||||
let t = PGEscape(t);
|
||||
let s = PGEscape(s);
|
||||
let e = PGEscape(e);
|
||||
|
||||
writeln!(sql, " ({sg}, {room_id}, {t}, {s}, {e}),").unwrap();
|
||||
if first {
|
||||
sql.push_str(" ");
|
||||
first = false;
|
||||
} else {
|
||||
sql.push_str(" ,");
|
||||
}
|
||||
sql.push_str(&format!(
|
||||
"({}, {}, {}, {}, {})",
|
||||
sg,
|
||||
PGEscape(room_id),
|
||||
PGEscape(t),
|
||||
PGEscape(s),
|
||||
PGEscape(e)
|
||||
));
|
||||
}
|
||||
|
||||
// Replace the last comma with a semicolon
|
||||
sql.replace_range((sql.len() - 2).., ";\n");
|
||||
sql.push_str(";\n");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -195,7 +189,7 @@ fn collapse_state_with_database(state_group: i64) -> StateMap<Atom> {
|
||||
|
||||
while let Some(sg) = next_group {
|
||||
// get predecessor from state_group_edges
|
||||
let mut pred = client.query_raw(query_pred, [sg]).unwrap();
|
||||
let mut pred = client.query_raw(query_pred, &[sg]).unwrap();
|
||||
|
||||
// set next_group to predecessor
|
||||
next_group = match pred.next().unwrap() {
|
||||
@@ -209,7 +203,7 @@ fn collapse_state_with_database(state_group: i64) -> StateMap<Atom> {
|
||||
}
|
||||
drop(pred);
|
||||
|
||||
let mut rows = client.query_raw(query_deltas, [sg]).unwrap();
|
||||
let mut rows = client.query_raw(query_deltas, &[sg]).unwrap();
|
||||
|
||||
while let Some(row) = rows.next().unwrap() {
|
||||
// Copy the single delta from the predecessor stored in this row
|
||||
|
||||
@@ -1,8 +1,4 @@
|
||||
[build-system]
|
||||
requires = ["maturin>=1.0,<2.0"]
|
||||
requires = ["maturin>=0.11,<0.12"]
|
||||
build-backend = "maturin"
|
||||
|
||||
[tool.maturin]
|
||||
profile = "release"
|
||||
features = ["pyo3"]
|
||||
no-default-features = true
|
||||
cargo-extra-args = "--no-default-features"
|
||||
@@ -30,7 +30,7 @@
|
||||
|
||||
use indicatif::{ProgressBar, ProgressStyle};
|
||||
use state_map::StateMap;
|
||||
use std::{collections::BTreeMap, time::Duration};
|
||||
use std::collections::BTreeMap;
|
||||
use string_cache::DefaultAtom as Atom;
|
||||
|
||||
use super::{collapse_state_maps, StateGroupEntry};
|
||||
@@ -156,7 +156,7 @@ impl<'a> Compressor<'a> {
|
||||
) -> Compressor<'a> {
|
||||
let levels = level_info
|
||||
.iter()
|
||||
.map(|l| Level::restore(l.max_length, l.current_chain_length, l.head))
|
||||
.map(|l| Level::restore((*l).max_length, (*l).current_chain_length, (*l).head))
|
||||
.collect();
|
||||
|
||||
let mut compressor = Compressor {
|
||||
@@ -181,18 +181,16 @@ impl<'a> Compressor<'a> {
|
||||
panic!("Can only call `create_new_tree` once");
|
||||
}
|
||||
|
||||
let pb = if cfg!(feature = "no-progress-bars") {
|
||||
let pb: ProgressBar = if cfg!(feature = "no-progress-bars") {
|
||||
ProgressBar::hidden()
|
||||
} else {
|
||||
ProgressBar::new(self.original_state_map.len() as u64)
|
||||
};
|
||||
pb.set_style(
|
||||
ProgressStyle::default_bar()
|
||||
.template("[{elapsed_precise}] {bar} {pos}/{len} {msg}")
|
||||
.unwrap(),
|
||||
ProgressStyle::default_bar().template("[{elapsed_precise}] {bar} {pos}/{len} {msg}"),
|
||||
);
|
||||
pb.set_message("state groups");
|
||||
pb.enable_steady_tick(Duration::from_millis(100));
|
||||
pb.enable_steady_tick(100);
|
||||
|
||||
for (&state_group, entry) in self.original_state_map {
|
||||
// Check whether this entry is in_range or is just present in the map due to being
|
||||
|
||||
@@ -93,10 +93,7 @@ fn create_new_tree_does_nothing_if_already_compressed() {
|
||||
|
||||
for i in 0i64..=13i64 {
|
||||
// edge from map
|
||||
let pred_group = initial_edges.get(&i);
|
||||
|
||||
// Need Option<i64> not Option<&i64>
|
||||
let prev = pred_group.copied();
|
||||
let prev = initial_edges.get(&i).copied();
|
||||
|
||||
// insert that edge into the initial map
|
||||
initial.insert(
|
||||
|
||||
@@ -142,10 +142,7 @@ fn stats_correct_if_no_changes() {
|
||||
|
||||
for i in 0i64..=13i64 {
|
||||
// edge from map
|
||||
let pred_group = initial_edges.get(&i);
|
||||
|
||||
// Need Option<i64> not Option<&i64>
|
||||
let prev = pred_group.copied();
|
||||
let prev = initial_edges.get(&i).copied();
|
||||
|
||||
// insert that edge into the initial map
|
||||
initial.insert(
|
||||
|
||||
@@ -18,7 +18,7 @@ use openssl::ssl::{SslConnector, SslMethod, SslVerifyMode};
|
||||
use postgres::{fallible_iterator::FallibleIterator, types::ToSql, Client};
|
||||
use postgres_openssl::MakeTlsConnector;
|
||||
use rand::{distributions::Alphanumeric, thread_rng, Rng};
|
||||
use std::{borrow::Cow, collections::BTreeMap, fmt, time::Duration};
|
||||
use std::{borrow::Cow, collections::BTreeMap, fmt};
|
||||
|
||||
use crate::{compressor::Level, generate_sql};
|
||||
|
||||
@@ -237,9 +237,15 @@ fn load_map_from_db(
|
||||
let mut missing_sgs: Vec<_> = state_group_map
|
||||
.iter()
|
||||
.filter_map(|(_sg, entry)| {
|
||||
entry
|
||||
.prev_state_group
|
||||
.filter(|&prev_sg| !state_group_map.contains_key(&prev_sg))
|
||||
if let Some(prev_sg) = entry.prev_state_group {
|
||||
if state_group_map.contains_key(&prev_sg) {
|
||||
None
|
||||
} else {
|
||||
Some(prev_sg)
|
||||
}
|
||||
} else {
|
||||
None
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
|
||||
@@ -366,17 +372,15 @@ fn get_initial_data_from_db(
|
||||
// Copy the data from the database into a map
|
||||
let mut state_group_map: BTreeMap<i64, StateGroupEntry> = BTreeMap::new();
|
||||
|
||||
let pb = if cfg!(feature = "no-progress-bars") {
|
||||
let pb: ProgressBar = if cfg!(feature = "no-progress-bars") {
|
||||
ProgressBar::hidden()
|
||||
} else {
|
||||
ProgressBar::new_spinner()
|
||||
};
|
||||
pb.set_style(
|
||||
ProgressStyle::default_spinner()
|
||||
.template("{spinner} [{elapsed}] {pos} rows retrieved")
|
||||
.unwrap(),
|
||||
ProgressStyle::default_spinner().template("{spinner} [{elapsed}] {pos} rows retrieved"),
|
||||
);
|
||||
pb.enable_steady_tick(Duration::from_millis(100));
|
||||
pb.enable_steady_tick(100);
|
||||
|
||||
while let Some(row) = rows.next().unwrap() {
|
||||
// The row in the map to copy the data to
|
||||
@@ -532,18 +536,16 @@ pub fn send_changes_to_db(
|
||||
debug!("Writing changes...");
|
||||
|
||||
// setup the progress bar
|
||||
let pb = if cfg!(feature = "no-progress-bars") {
|
||||
let pb: ProgressBar = if cfg!(feature = "no-progress-bars") {
|
||||
ProgressBar::hidden()
|
||||
} else {
|
||||
ProgressBar::new(old_map.len() as u64)
|
||||
};
|
||||
pb.set_style(
|
||||
ProgressStyle::default_bar()
|
||||
.template("[{elapsed_precise}] {bar} {pos}/{len} {msg}")
|
||||
.unwrap(),
|
||||
ProgressStyle::default_bar().template("[{elapsed_precise}] {bar} {pos}/{len} {msg}"),
|
||||
);
|
||||
pb.set_message("state groups");
|
||||
pb.enable_steady_tick(Duration::from_millis(100));
|
||||
pb.enable_steady_tick(100);
|
||||
|
||||
for sql_transaction in generate_sql(old_map, new_map, room_id) {
|
||||
if sql_transaction.is_empty() {
|
||||
|
||||
239
src/lib.rs
239
src/lib.rs
@@ -20,19 +20,14 @@
|
||||
// of arguments - this hopefully doesn't make the code unclear
|
||||
// #[allow(clippy::too_many_arguments)] is therefore used around some functions
|
||||
|
||||
use log::{info, warn};
|
||||
#[cfg(feature = "pyo3")]
|
||||
use log::{info, warn, LevelFilter};
|
||||
use pyo3::{exceptions, prelude::*};
|
||||
|
||||
#[cfg(feature = "clap")]
|
||||
use clap::{crate_authors, crate_description, crate_name, crate_version, Arg, Command};
|
||||
use clap::{crate_authors, crate_description, crate_name, crate_version, value_t, App, Arg};
|
||||
use indicatif::{ProgressBar, ProgressStyle};
|
||||
use rayon::prelude::*;
|
||||
use state_map::StateMap;
|
||||
use std::{
|
||||
collections::BTreeMap, convert::TryInto, fmt::Write as _, fs::File, io::Write, str::FromStr,
|
||||
time::Duration,
|
||||
};
|
||||
use std::{collections::BTreeMap, convert::TryInto, fs::File, io::Write, str::FromStr};
|
||||
use string_cache::DefaultAtom as Atom;
|
||||
|
||||
mod compressor;
|
||||
@@ -54,7 +49,7 @@ pub struct StateGroupEntry {
|
||||
}
|
||||
|
||||
/// Helper struct for parsing the `level_sizes` argument.
|
||||
#[derive(PartialEq, Debug, Clone)]
|
||||
#[derive(PartialEq, Debug)]
|
||||
struct LevelSizes(Vec<usize>);
|
||||
|
||||
impl FromStr for LevelSizes {
|
||||
@@ -119,17 +114,16 @@ pub struct Config {
|
||||
verify: bool,
|
||||
}
|
||||
|
||||
#[cfg(feature = "clap")]
|
||||
impl Config {
|
||||
/// Build up config from command line arguments
|
||||
pub fn parse_arguments() -> Config {
|
||||
let matches = Command::new(crate_name!())
|
||||
let matches = App::new(crate_name!())
|
||||
.version(crate_version!())
|
||||
.author(crate_authors!("\n"))
|
||||
.about(crate_description!())
|
||||
.arg(
|
||||
Arg::new("postgres-url")
|
||||
.short('p')
|
||||
Arg::with_name("postgres-url")
|
||||
.short("p")
|
||||
.value_name("POSTGRES_LOCATION")
|
||||
.help("The configruation for connecting to the postgres database.")
|
||||
.long_help(concat!(
|
||||
@@ -139,69 +133,64 @@ impl Config {
|
||||
"See https://docs.rs/tokio-postgres/0.7.2/tokio_postgres/config/struct.Config.html ",
|
||||
"for the full details."
|
||||
))
|
||||
.num_args(1)
|
||||
.takes_value(true)
|
||||
.required(true),
|
||||
).arg(
|
||||
Arg::new("room_id")
|
||||
.short('r')
|
||||
Arg::with_name("room_id")
|
||||
.short("r")
|
||||
.value_name("ROOM_ID")
|
||||
.help("The room to process")
|
||||
.long_help(concat!(
|
||||
"The room to process. This is the value found in the rooms table of the database",
|
||||
" not the common name for the room - is should look like: \"!wOlkWNmgkAZFxbTaqj:matrix.org\""
|
||||
))
|
||||
.num_args(1)
|
||||
.takes_value(true)
|
||||
.required(true),
|
||||
).arg(
|
||||
Arg::new("min_state_group")
|
||||
.short('b')
|
||||
Arg::with_name("min_state_group")
|
||||
.short("b")
|
||||
.value_name("MIN_STATE_GROUP")
|
||||
.value_parser(clap::value_parser!(i64))
|
||||
.help("The state group to start processing from (non inclusive)")
|
||||
.num_args(1)
|
||||
.takes_value(true)
|
||||
.required(false),
|
||||
).arg(
|
||||
Arg::new("min_saved_rows")
|
||||
.short('m')
|
||||
Arg::with_name("min_saved_rows")
|
||||
.short("m")
|
||||
.value_name("COUNT")
|
||||
.value_parser(clap::value_parser!(i32))
|
||||
.help("Abort if fewer than COUNT rows would be saved")
|
||||
.long_help("If the compressor cannot save this many rows from the database then it will stop early")
|
||||
.num_args(1)
|
||||
.takes_value(true)
|
||||
.required(false),
|
||||
).arg(
|
||||
Arg::new("groups_to_compress")
|
||||
.short('n')
|
||||
Arg::with_name("groups_to_compress")
|
||||
.short("n")
|
||||
.value_name("GROUPS_TO_COMPRESS")
|
||||
.value_parser(clap::value_parser!(i64))
|
||||
.help("How many groups to load into memory to compress")
|
||||
.long_help(concat!(
|
||||
"How many groups to load into memory to compress (starting from",
|
||||
" the 1st group in the room or the group specified by -s)"))
|
||||
.num_args(1)
|
||||
.takes_value(true)
|
||||
.required(false),
|
||||
).arg(
|
||||
Arg::new("output_file")
|
||||
.short('o')
|
||||
Arg::with_name("output_file")
|
||||
.short("o")
|
||||
.value_name("FILE")
|
||||
.help("File to output the changes to in SQL")
|
||||
.num_args(1),
|
||||
.takes_value(true),
|
||||
).arg(
|
||||
Arg::new("max_state_group")
|
||||
.short('s')
|
||||
Arg::with_name("max_state_group")
|
||||
.short("s")
|
||||
.value_name("MAX_STATE_GROUP")
|
||||
.value_parser(clap::value_parser!(i64))
|
||||
.help("The maximum state group to process up to")
|
||||
.long_help(concat!(
|
||||
"If a max_state_group is specified then only state groups with id's lower",
|
||||
" than this number are able to be compressed."))
|
||||
.num_args(1)
|
||||
.takes_value(true)
|
||||
.required(false),
|
||||
).arg(
|
||||
Arg::new("level_sizes")
|
||||
.short('l')
|
||||
Arg::with_name("level_sizes")
|
||||
.short("l")
|
||||
.value_name("LEVELS")
|
||||
.value_parser(clap::value_parser!(LevelSizes))
|
||||
.help("Sizes of each new level in the compression algorithm, as a comma separated list.")
|
||||
.long_help(concat!(
|
||||
"Sizes of each new level in the compression algorithm, as a comma separated list.",
|
||||
@@ -214,36 +203,32 @@ impl Config {
|
||||
" iterations needed to fetch a given set of state.",
|
||||
))
|
||||
.default_value("100,50,25")
|
||||
.num_args(1),
|
||||
.takes_value(true),
|
||||
).arg(
|
||||
Arg::new("transactions")
|
||||
.short('t')
|
||||
.action(clap::ArgAction::SetTrue)
|
||||
Arg::with_name("transactions")
|
||||
.short("t")
|
||||
.help("Whether to wrap each state group change in a transaction")
|
||||
.long_help(concat!("If this flag is set then then each change to a particular",
|
||||
" state group is wrapped in a transaction. This should be done if you wish to",
|
||||
" apply the changes while synapse is still running."))
|
||||
.requires("output_file"),
|
||||
).arg(
|
||||
Arg::new("graphs")
|
||||
.short('g')
|
||||
.action(clap::ArgAction::SetTrue)
|
||||
Arg::with_name("graphs")
|
||||
.short("g")
|
||||
.help("Output before and after graphs")
|
||||
.long_help(concat!("If this flag is set then output the node and edge information for",
|
||||
" the state_group directed graph built up from the predecessor state_group links.",
|
||||
" These can be looked at in something like Gephi (https://gephi.org)")),
|
||||
).arg(
|
||||
Arg::new("commit_changes")
|
||||
.short('c')
|
||||
.action(clap::ArgAction::SetTrue)
|
||||
Arg::with_name("commit_changes")
|
||||
.short("c")
|
||||
.help("Commit changes to the database")
|
||||
.long_help(concat!("If this flag is set then the changes the compressor makes will",
|
||||
" be committed to the database. This should be safe to use while synapse is running",
|
||||
" as it assumes by default that the transactions flag is set")),
|
||||
).arg(
|
||||
Arg::new("no_verify")
|
||||
.short('N')
|
||||
.action(clap::ArgAction::SetTrue)
|
||||
Arg::with_name("no_verify")
|
||||
.short("N")
|
||||
.help("Do not double-check that the compression was performed correctly")
|
||||
.long_help(concat!("If this flag is set then the verification of the compressed",
|
||||
" state groups, which compares them to the original groups, is skipped. This",
|
||||
@@ -251,27 +236,43 @@ impl Config {
|
||||
).get_matches();
|
||||
|
||||
let db_url = matches
|
||||
.get_one::<String>("postgres-url")
|
||||
.value_of("postgres-url")
|
||||
.expect("db url should be required");
|
||||
|
||||
let output_file = matches.get_one::<String>("output_file").map(|path| {
|
||||
let output_file = matches.value_of("output_file").map(|path| {
|
||||
File::create(path).unwrap_or_else(|e| panic!("Unable to create output file: {}", e))
|
||||
});
|
||||
|
||||
let room_id = matches
|
||||
.get_one::<String>("room_id")
|
||||
.value_of("room_id")
|
||||
.expect("room_id should be required since no file");
|
||||
|
||||
let min_state_group = matches.get_one("min_state_group").copied();
|
||||
let groups_to_compress = matches.get_one("groups_to_compress").copied();
|
||||
let min_saved_rows = matches.get_one("min_saved_rows").copied();
|
||||
let max_state_group = matches.get_one("max_state_group").copied();
|
||||
let level_sizes = matches.get_one("level_sizes").cloned().unwrap();
|
||||
let min_state_group = matches
|
||||
.value_of("min_state_group")
|
||||
.map(|s| s.parse().expect("min_state_group must be an integer"));
|
||||
|
||||
let transactions = matches.get_flag("transactions");
|
||||
let graphs = matches.get_flag("graphs");
|
||||
let commit_changes = matches.get_flag("commit_changes");
|
||||
let verify = !matches.get_flag("no_verify");
|
||||
let groups_to_compress = matches
|
||||
.value_of("groups_to_compress")
|
||||
.map(|s| s.parse().expect("groups_to_compress must be an integer"));
|
||||
|
||||
let min_saved_rows = matches
|
||||
.value_of("min_saved_rows")
|
||||
.map(|v| v.parse().expect("COUNT must be an integer"));
|
||||
|
||||
let max_state_group = matches
|
||||
.value_of("max_state_group")
|
||||
.map(|s| s.parse().expect("max_state_group must be an integer"));
|
||||
|
||||
let level_sizes = value_t!(matches, "level_sizes", LevelSizes)
|
||||
.unwrap_or_else(|e| panic!("Unable to parse level_sizes: {}", e));
|
||||
|
||||
let transactions = matches.is_present("transactions");
|
||||
|
||||
let graphs = matches.is_present("graphs");
|
||||
|
||||
let commit_changes = matches.is_present("commit_changes");
|
||||
|
||||
let verify = !matches.is_present("no_verify");
|
||||
|
||||
Config {
|
||||
db_url: String::from(db_url),
|
||||
@@ -422,7 +423,8 @@ fn generate_sql<'a>(
|
||||
new_map: &'a BTreeMap<i64, StateGroupEntry>,
|
||||
room_id: &'a str,
|
||||
) -> impl Iterator<Item = String> + 'a {
|
||||
old_map.iter().map(move |(sg, old_entry)| {
|
||||
old_map.iter().map(move |(sg,old_entry)| {
|
||||
|
||||
let new_entry = &new_map[sg];
|
||||
|
||||
// Check if the new map has a different entry for this state group
|
||||
@@ -432,50 +434,48 @@ fn generate_sql<'a>(
|
||||
let mut sql = String::new();
|
||||
|
||||
// remove the current edge
|
||||
writeln!(
|
||||
sql,
|
||||
"DELETE FROM state_group_edges WHERE state_group = {sg};",
|
||||
)
|
||||
.expect("Writing to a String cannot fail");
|
||||
sql.push_str(&format!(
|
||||
"DELETE FROM state_group_edges WHERE state_group = {};\n",
|
||||
sg
|
||||
));
|
||||
|
||||
// if the new entry has a predecessor then put that into state_group_edges
|
||||
if let Some(prev_sg) = new_entry.prev_state_group {
|
||||
writeln!(
|
||||
sql,
|
||||
"INSERT INTO state_group_edges (state_group, prev_state_group) \
|
||||
VALUES ({sg}, {prev_sg});",
|
||||
)
|
||||
.unwrap();
|
||||
sql.push_str(&format!("INSERT INTO state_group_edges (state_group, prev_state_group) VALUES ({}, {});\n", sg, prev_sg));
|
||||
}
|
||||
|
||||
// remove the current deltas for this state group
|
||||
writeln!(
|
||||
sql,
|
||||
"DELETE FROM state_groups_state WHERE state_group = {sg};",
|
||||
)
|
||||
.unwrap();
|
||||
sql.push_str(&format!(
|
||||
"DELETE FROM state_groups_state WHERE state_group = {};\n",
|
||||
sg
|
||||
));
|
||||
|
||||
if !new_entry.state_map.is_empty() {
|
||||
// place all the deltas for the state group in the new map into state_groups_state
|
||||
sql.push_str(
|
||||
"INSERT INTO state_groups_state \
|
||||
(state_group, room_id, type, state_key, event_id) \
|
||||
VALUES\n",
|
||||
);
|
||||
sql.push_str("INSERT INTO state_groups_state (state_group, room_id, type, state_key, event_id) VALUES\n");
|
||||
|
||||
let room_id = PGEscape(room_id);
|
||||
let mut first = true;
|
||||
for ((t, s), e) in new_entry.state_map.iter() {
|
||||
let t = PGEscape(t);
|
||||
let s = PGEscape(s);
|
||||
let e = PGEscape(e);
|
||||
// Add a comma at the start if not the first row to be inserted
|
||||
if first {
|
||||
sql.push_str(" ");
|
||||
first = false;
|
||||
} else {
|
||||
sql.push_str(" ,");
|
||||
}
|
||||
|
||||
// write the row to be inserted of the form:
|
||||
// write the row to be insterted of the form:
|
||||
// (state_group, room_id, type, state_key, event_id)
|
||||
writeln!(sql, " ({sg}, {room_id}, {t}, {s}, {e}),").unwrap();
|
||||
sql.push_str(&format!(
|
||||
"({}, {}, {}, {}, {})",
|
||||
sg,
|
||||
PGEscape(room_id),
|
||||
PGEscape(t),
|
||||
PGEscape(s),
|
||||
PGEscape(e)
|
||||
));
|
||||
}
|
||||
|
||||
// Replace the last comma with a semicolon
|
||||
sql.replace_range((sql.len() - 2).., ";\n");
|
||||
sql.push_str(";\n");
|
||||
}
|
||||
|
||||
sql
|
||||
@@ -507,18 +507,16 @@ fn output_sql(
|
||||
|
||||
info!("Writing changes...");
|
||||
|
||||
let pb = if cfg!(feature = "no-progress-bars") {
|
||||
let pb: ProgressBar = if cfg!(feature = "no-progress-bars") {
|
||||
ProgressBar::hidden()
|
||||
} else {
|
||||
ProgressBar::new(old_map.len() as u64)
|
||||
};
|
||||
pb.set_style(
|
||||
ProgressStyle::default_bar()
|
||||
.template("[{elapsed_precise}] {bar} {pos}/{len} {msg}")
|
||||
.unwrap(),
|
||||
ProgressStyle::default_bar().template("[{elapsed_precise}] {bar} {pos}/{len} {msg}"),
|
||||
);
|
||||
pb.set_message("state groups");
|
||||
pb.enable_steady_tick(Duration::from_millis(100));
|
||||
pb.enable_steady_tick(100);
|
||||
|
||||
if let Some(output) = &mut config.output_file {
|
||||
for mut sql_transaction in generate_sql(old_map, new_map, &config.room_id) {
|
||||
@@ -566,7 +564,7 @@ pub fn continue_run(
|
||||
let (state_group_map, max_group_found) =
|
||||
database::reload_data_from_db(db_url, room_id, start, Some(chunk_size), level_info)?;
|
||||
|
||||
let original_num_rows = state_group_map.values().map(|v| v.state_map.len()).sum();
|
||||
let original_num_rows = state_group_map.iter().map(|(_, v)| v.state_map.len()).sum();
|
||||
|
||||
// Now we actually call the compression algorithm.
|
||||
let compressor = Compressor::compress_from_save(&state_group_map, level_info);
|
||||
@@ -623,18 +621,16 @@ fn check_that_maps_match(
|
||||
) {
|
||||
info!("Checking that state maps match...");
|
||||
|
||||
let pb = if cfg!(feature = "no-progress-bars") {
|
||||
let pb: ProgressBar = if cfg!(feature = "no-progress-bars") {
|
||||
ProgressBar::hidden()
|
||||
} else {
|
||||
ProgressBar::new(old_map.len() as u64)
|
||||
};
|
||||
pb.set_style(
|
||||
ProgressStyle::default_bar()
|
||||
.template("[{elapsed_precise}] {bar} {pos}/{len} {msg}")
|
||||
.unwrap(),
|
||||
ProgressStyle::default_bar().template("[{elapsed_precise}] {bar} {pos}/{len} {msg}"),
|
||||
);
|
||||
pb.set_message("state groups");
|
||||
pb.enable_steady_tick(Duration::from_millis(100));
|
||||
pb.enable_steady_tick(100);
|
||||
|
||||
// Now let's iterate through and assert that the state for each group
|
||||
// matches between the two versions.
|
||||
@@ -750,31 +746,23 @@ impl Config {
|
||||
/// Default arguments are equivalent to using the command line tool
|
||||
/// No default's are provided for db_url or room_id since these arguments
|
||||
/// are compulsory (so that new() act's like parse_arguments())
|
||||
#[cfg(feature = "pyo3")]
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
#[pyfunction]
|
||||
#[pyo3(signature = (
|
||||
#[pyfunction(
|
||||
// db_url has no default
|
||||
db_url,
|
||||
|
||||
// room_id has no default
|
||||
room_id,
|
||||
|
||||
output_file = None,
|
||||
min_state_group = None,
|
||||
groups_to_compress = None,
|
||||
min_saved_rows = None,
|
||||
max_state_group = None,
|
||||
level_sizes = String::from("100,50,25"),
|
||||
|
||||
// room_id has no default
|
||||
output_file = "None",
|
||||
min_state_group = "None",
|
||||
groups_to_compress = "None",
|
||||
min_saved_rows = "None",
|
||||
max_state_group = "None",
|
||||
level_sizes = "String::from(\"100,50,25\")",
|
||||
// have this default to true as is much worse to not have it if you need it
|
||||
// than to have it and not need it
|
||||
transactions = true,
|
||||
|
||||
graphs = false,
|
||||
commit_changes = false,
|
||||
verify = true,
|
||||
))]
|
||||
)]
|
||||
fn run_compression(
|
||||
db_url: String,
|
||||
room_id: String,
|
||||
@@ -813,15 +801,14 @@ fn run_compression(
|
||||
}
|
||||
|
||||
/// Python module - "import synapse_compress_state" to use
|
||||
#[cfg(feature = "pyo3")]
|
||||
#[pymodule]
|
||||
fn synapse_compress_state(_py: Python, m: &PyModule) -> PyResult<()> {
|
||||
let _ = pyo3_log::Logger::default()
|
||||
// don't send out anything lower than a warning from other crates
|
||||
.filter(log::LevelFilter::Warn)
|
||||
.filter(LevelFilter::Warn)
|
||||
// don't log warnings from synapse_compress_state, the synapse_auto_compressor handles these
|
||||
// situations and provides better log messages
|
||||
.filter_target("synapse_compress_state".to_owned(), log::LevelFilter::Debug)
|
||||
.filter_target("synapse_compress_state".to_owned(), LevelFilter::Debug)
|
||||
.install();
|
||||
// ensure any panics produce error messages in the log
|
||||
log_panics::init();
|
||||
|
||||
@@ -1,15 +1,11 @@
|
||||
[package]
|
||||
name = "synapse_auto_compressor"
|
||||
authors = ["William Ashton"]
|
||||
version = "0.1.3"
|
||||
version = "0.1.2"
|
||||
edition = "2018"
|
||||
|
||||
[[bin]]
|
||||
name = "synapse_auto_compressor"
|
||||
required-features = ["clap"]
|
||||
|
||||
[package.metadata.maturin]
|
||||
requires-python = ">=3.7"
|
||||
requires-python = ">=3.6"
|
||||
project-url = {Source = "https://github.com/matrix-org/rust-synapse-compress-state"}
|
||||
classifier = [
|
||||
"Development Status :: 4 - Beta",
|
||||
@@ -17,40 +13,24 @@ classifier = [
|
||||
]
|
||||
|
||||
[dependencies]
|
||||
openssl = { version = "0.10.60", features = ["vendored"] }
|
||||
postgres = "0.19.7"
|
||||
clap = "2.33.0"
|
||||
openssl = "0.10.32"
|
||||
postgres = "0.19.0"
|
||||
postgres-openssl = "0.5.0"
|
||||
rand = "0.8.5"
|
||||
serial_test = "2.0.0"
|
||||
synapse_compress_state = { path = "../", features = ["no-progress-bars"], default-features = false }
|
||||
env_logger = "0.10.0"
|
||||
log = "0.4.20"
|
||||
log-panics = "2.1.0"
|
||||
anyhow = "1.0.75"
|
||||
tikv-jemallocator = "0.4.1"
|
||||
rand = "0.8.0"
|
||||
serial_test = "0.5.1"
|
||||
synapse_compress_state = { path = "../", features = ["no-progress-bars"] }
|
||||
env_logger = "0.9.0"
|
||||
log = "0.4.14"
|
||||
log-panics = "2.0.0"
|
||||
anyhow = "1.0.42"
|
||||
pyo3-log = "0.4.0"
|
||||
|
||||
# Needed for pyo3 support
|
||||
[lib]
|
||||
crate-type = ["cdylib", "rlib"]
|
||||
|
||||
[dependencies.clap]
|
||||
version = "4.4.2"
|
||||
features = ["cargo"]
|
||||
optional = true
|
||||
|
||||
[dependencies.pyo3]
|
||||
version = "0.19.2"
|
||||
features = ["extension-module"]
|
||||
optional = true
|
||||
|
||||
[dependencies.pyo3-log]
|
||||
version = "0.8.3"
|
||||
optional = true
|
||||
|
||||
[dependencies.tikv-jemallocator]
|
||||
version = "0.5.4"
|
||||
optional = true
|
||||
|
||||
[features]
|
||||
default = ["clap", "jemalloc"]
|
||||
jemalloc = ["tikv-jemallocator", "synapse_compress_state/jemalloc"]
|
||||
pyo3 = ["dep:pyo3", "dep:pyo3-log", "synapse_compress_state/pyo3"]
|
||||
version = "0.14.1"
|
||||
features = ["extension-module","abi3-py36"]
|
||||
|
||||
@@ -1,8 +0,0 @@
|
||||
[build-system]
|
||||
requires = ["maturin>=1.0,<2.0"]
|
||||
build-backend = "maturin"
|
||||
|
||||
[tool.maturin]
|
||||
profile = "release"
|
||||
features = ["pyo3"]
|
||||
no-default-features = true
|
||||
@@ -7,9 +7,7 @@
|
||||
//! on space reductions
|
||||
|
||||
use anyhow::Result;
|
||||
#[cfg(feature = "pyo3")]
|
||||
use log::{error, LevelFilter};
|
||||
#[cfg(feature = "pyo3")]
|
||||
use pyo3::{
|
||||
exceptions::PyRuntimeError, prelude::pymodule, types::PyModule, PyErr, PyResult, Python,
|
||||
};
|
||||
@@ -28,7 +26,7 @@ pub mod state_saving;
|
||||
///
|
||||
/// This is needed since FromStr cannot be implemented for structs
|
||||
/// that aren't defined in this scope
|
||||
#[derive(PartialEq, Eq, Debug, Clone)]
|
||||
#[derive(PartialEq, Debug)]
|
||||
pub struct LevelInfo(pub Vec<Level>);
|
||||
|
||||
// Implement FromStr so that an argument of the form "100,50,25"
|
||||
@@ -58,7 +56,6 @@ impl FromStr for LevelInfo {
|
||||
}
|
||||
|
||||
// PyO3 INTERFACE STARTS HERE
|
||||
#[cfg(feature = "pyo3")]
|
||||
#[pymodule]
|
||||
fn synapse_auto_compressor(_py: Python, m: &PyModule) -> PyResult<()> {
|
||||
let _ = pyo3_log::Logger::default()
|
||||
@@ -74,8 +71,7 @@ fn synapse_auto_compressor(_py: Python, m: &PyModule) -> PyResult<()> {
|
||||
// ensure any panics produce error messages in the log
|
||||
log_panics::init();
|
||||
|
||||
#[pyfn(m)]
|
||||
#[pyo3(name = "compress_largest_rooms")]
|
||||
#[pyfn(m, compress_largest_rooms)]
|
||||
fn compress_state_events_table(
|
||||
py: Python,
|
||||
db_url: String,
|
||||
|
||||
@@ -16,11 +16,10 @@
|
||||
//! the state_compressor_state table so that the compressor can seemlesly
|
||||
//! continue from where it left off.
|
||||
|
||||
#[cfg(feature = "jemalloc")]
|
||||
#[global_allocator]
|
||||
static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
|
||||
|
||||
use clap::{crate_authors, crate_description, crate_name, crate_version, Arg, Command};
|
||||
use clap::{crate_authors, crate_description, crate_name, crate_version, value_t, App, Arg};
|
||||
use log::LevelFilter;
|
||||
use std::env;
|
||||
use synapse_auto_compressor::{manager, state_saving, LevelInfo};
|
||||
@@ -51,13 +50,13 @@ fn main() {
|
||||
log::info!("synapse_auto_compressor started");
|
||||
|
||||
// parse the command line arguments using the clap crate
|
||||
let arguments = Command::new(crate_name!())
|
||||
let arguments = App::new(crate_name!())
|
||||
.version(crate_version!())
|
||||
.author(crate_authors!("\n"))
|
||||
.about(crate_description!())
|
||||
.arg(
|
||||
Arg::new("postgres-url")
|
||||
.short('p')
|
||||
Arg::with_name("postgres-url")
|
||||
.short("p")
|
||||
.value_name("POSTGRES_LOCATION")
|
||||
.help("The configruation for connecting to the postgres database.")
|
||||
.long_help(concat!(
|
||||
@@ -67,13 +66,12 @@ fn main() {
|
||||
"See https://docs.rs/tokio-postgres/0.7.2/tokio_postgres/config/struct.Config.html ",
|
||||
"for the full details."
|
||||
))
|
||||
.num_args(1)
|
||||
.takes_value(true)
|
||||
.required(true),
|
||||
).arg(
|
||||
Arg::new("chunk_size")
|
||||
.short('c')
|
||||
Arg::with_name("chunk_size")
|
||||
.short("c")
|
||||
.value_name("COUNT")
|
||||
.value_parser(clap::value_parser!(i64))
|
||||
.help("The maximum number of state groups to load into memroy at once")
|
||||
.long_help(concat!(
|
||||
"The number of state_groups to work on at once. All of the entries",
|
||||
@@ -84,13 +82,12 @@ fn main() {
|
||||
" chunk as a whole (which may well happen in rooms with lots",
|
||||
" of backfill in) then the entire chunk is skipped.)",
|
||||
))
|
||||
.num_args(1)
|
||||
.takes_value(true)
|
||||
.required(true),
|
||||
).arg(
|
||||
Arg::new("default_levels")
|
||||
.short('l')
|
||||
Arg::with_name("default_levels")
|
||||
.short("l")
|
||||
.value_name("LEVELS")
|
||||
.value_parser(clap::value_parser!(LevelInfo))
|
||||
.help("Sizes of each new level in the compression algorithm, as a comma separated list.")
|
||||
.long_help(concat!(
|
||||
"Sizes of each new level in the compression algorithm, as a comma separated list.",
|
||||
@@ -103,43 +100,40 @@ fn main() {
|
||||
" iterations needed to fetch a given set of state.",
|
||||
))
|
||||
.default_value("100,50,25")
|
||||
.num_args(1)
|
||||
.takes_value(true)
|
||||
.required(false),
|
||||
).arg(
|
||||
Arg::new("number_of_chunks")
|
||||
.short('n')
|
||||
Arg::with_name("number_of_chunks")
|
||||
.short("n")
|
||||
.value_name("CHUNKS_TO_COMPRESS")
|
||||
.value_parser(clap::value_parser!(i64))
|
||||
.help("The number of chunks to compress")
|
||||
.long_help(concat!(
|
||||
"This many chunks of the database will be compressed. The higher this number is set to, ",
|
||||
"the longer the compressor will run for."
|
||||
))
|
||||
.num_args(1)
|
||||
.takes_value(true)
|
||||
.required(true),
|
||||
).get_matches();
|
||||
|
||||
// The URL of the database
|
||||
let db_url = arguments
|
||||
.get_one::<String>("postgres-url")
|
||||
.value_of("postgres-url")
|
||||
.expect("A database url is required");
|
||||
|
||||
// The number of state groups to work on at once
|
||||
let chunk_size = arguments
|
||||
.get_one("chunk_size")
|
||||
.copied()
|
||||
.value_of("chunk_size")
|
||||
.map(|s| s.parse().expect("chunk_size must be an integer"))
|
||||
.expect("A chunk size is required");
|
||||
|
||||
// The default structure to use when compressing
|
||||
let default_levels = arguments
|
||||
.get_one::<LevelInfo>("default_levels")
|
||||
.cloned()
|
||||
.unwrap();
|
||||
let default_levels = value_t!(arguments, "default_levels", LevelInfo)
|
||||
.unwrap_or_else(|e| panic!("Unable to parse default levels: {}", e));
|
||||
|
||||
// The number of rooms to compress with this tool
|
||||
let number_of_chunks = arguments
|
||||
.get_one("number_of_chunks")
|
||||
.copied()
|
||||
.value_of("number_of_chunks")
|
||||
.map(|s| s.parse().expect("number_of_chunks must be an integer"))
|
||||
.expect("number_of_chunks is required");
|
||||
|
||||
// Connect to the database and create the 2 tables this tool needs
|
||||
|
||||
Reference in New Issue
Block a user