oss/conduit
oss/conduit
1
0
Fork 0
mirror of https://gitlab.com/famedly/conduit.git synced 2025-07-02 16:38:36 +00:00
Code Wiki Activity

remove eldrich being and install good being

Browse source
This commit is contained in:
Jonathan de Jong 2021-07-04 01:18:06 +02:00
parent 9df86c2c1e
commit 14e6afc45e
34 changed files with 371 additions and 315 deletions
Download patch file Download diff file Expand all files Collapse all files

44
src/database/abstraction/sqlite.rs
View file

@ -3,7 +3,7 @@ use std::{
ops::Deref,
path::{Path, PathBuf},
pin::Pin,
sync::{Arc, Weak},
sync::Arc,
thread,
time::{Duration, Instant},
};
@ -36,7 +36,6 @@ use tokio::sync::oneshot::Sender;
struct Pool {
writer: Mutex<Connection>,
readers: Vec<Mutex<Connection>>,
reader_rwlock: RwLock<()>,
path: PathBuf,
}
@ -71,7 +70,6 @@ impl Pool {
Ok(Self {
writer,
readers,
reader_rwlock: RwLock::new(()),
path: path.as_ref().to_path_buf(),
})
}
@ -95,16 +93,12 @@ impl Pool {
}
fn read_lock(&self) -> HoldingConn<'_> {
let _guard = self.reader_rwlock.read();
for r in &self.readers {
if let Some(reader) = r.try_lock() {
return HoldingConn::FromGuard(reader);
}
}
drop(_guard);
log::warn!("all readers locked, creating spillover reader...");
let spilled = Self::prepare_conn(&self.path).unwrap();
@ -115,7 +109,6 @@ impl Pool {
pub struct SqliteEngine {
pool: Pool,
iterator_lock: RwLock<()>,
}
impl DatabaseEngine for SqliteEngine {
@ -128,36 +121,7 @@ impl DatabaseEngine for SqliteEngine {
pool.write_lock()
.execute("CREATE TABLE IF NOT EXISTS _noop (\"key\" INT)", params![])?;
let arc = Arc::new(SqliteEngine {
pool,
iterator_lock: RwLock::new(()),
});
let weak: Weak<SqliteEngine> = Arc::downgrade(&arc);
thread::spawn(move || {
let r = crossbeam::channel::tick(Duration::from_secs(60));
let weak = weak;
loop {
let _ = r.recv();
if let Some(arc) = Weak::upgrade(&weak) {
log::warn!("wal-trunc: locking...");
let iterator_guard = arc.iterator_lock.write();
let read_guard = arc.pool.reader_rwlock.write();
log::warn!("wal-trunc: locked, flushing...");
let start = Instant::now();
arc.flush_wal().unwrap();
log::warn!("wal-trunc: locked, flushed in {:?}", start.elapsed());
drop(read_guard);
drop(iterator_guard);
} else {
break;
}
}
});
let arc = Arc::new(SqliteEngine { pool });
Ok(arc)
}
@ -190,7 +154,7 @@ impl DatabaseEngine for SqliteEngine {
}
impl SqliteEngine {
fn flush_wal(self: &Arc<Self>) -> Result<()> {
pub fn flush_wal(self: &Arc<Self>) -> Result<()> {
self.pool
.write_lock()
.execute_batch(
@ -244,9 +208,7 @@ impl SqliteTable {
let engine = self.engine.clone();
thread::spawn(move || {
let guard = engine.iterator_lock.read();
let _ = f(&engine.pool.read_lock(), s);
drop(guard);
});
Box::new(r.into_iter())

54
src/database/admin.rs
View file

@ -10,6 +10,7 @@ use ruma::{
events::{room::message, EventType},
UserId,
};
use tokio::sync::RwLock;
pub enum AdminCommand {
RegisterAppservice(serde_yaml::Value),
@ -25,20 +26,22 @@ pub struct Admin {
impl Admin {
pub fn start_handler(
&self,
db: Arc<Database>,
db: Arc<RwLock<Database>>,
mut receiver: mpsc::UnboundedReceiver<AdminCommand>,
) {
tokio::spawn(async move {
// TODO: Use futures when we have long admin commands
//let mut futures = FuturesUnordered::new();
let conduit_user = UserId::try_from(format!("@conduit:{}", db.globals.server_name()))
let guard = db.read().await;
let conduit_user = UserId::try_from(format!("@conduit:{}", guard.globals.server_name()))
.expect("@conduit:server_name is valid");
let conduit_room = db
let conduit_room = guard
.rooms
.id_from_alias(
&format!("#admins:{}", db.globals.server_name())
&format!("#admins:{}", guard.globals.server_name())
.try_into()
.expect("#admins:server_name is a valid room alias"),
)
@ -48,24 +51,10 @@ impl Admin {
warn!("Conduit instance does not have an #admins room. Logging to that room will not work. Restart Conduit after creating a user to fix this.");
}
drop(guard);
let send_message = |message: message::MessageEventContent| {
if let Some(conduit_room) = &conduit_room {
db.rooms
.build_and_append_pdu(
PduBuilder {
event_type: EventType::RoomMessage,
content: serde_json::to_value(message)
.expect("event is valid, we just created it"),
unsigned: None,
state_key: None,
redacts: None,
},
&conduit_user,
&conduit_room,
&db,
)
.unwrap();
}
};
loop {
@ -73,10 +62,10 @@ impl Admin {
Some(event) = receiver.next() => {
match event {
AdminCommand::RegisterAppservice(yaml) => {
db.appservice.register_appservice(yaml).unwrap(); // TODO handle error
db.read().await.appservice.register_appservice(yaml).unwrap(); // TODO handle error
}
AdminCommand::ListAppservices => {
if let Ok(appservices) = db.appservice.iter_ids().map(|ids| ids.collect::<Vec<_>>()) {
if let Ok(appservices) = db.read().await.appservice.iter_ids().map(|ids| ids.collect::<Vec<_>>()) {
let count = appservices.len();
let output = format!(
"Appservices ({}): {}",
@ -89,7 +78,24 @@ impl Admin {
}
}
AdminCommand::SendMessage(message) => {
send_message(message);
if let Some(conduit_room) = &conduit_room {
let guard = db.read().await;
guard.rooms
.build_and_append_pdu(
PduBuilder {
event_type: EventType::RoomMessage,
content: serde_json::to_value(message)
.expect("event is valid, we just created it"),
unsigned: None,
state_key: None,
redacts: None,
},
&conduit_user,
&conduit_room,
&guard,
)
.unwrap();
}
}
}
}

43
src/database/sending.rs
View file

@ -30,7 +30,7 @@ use ruma::{
receipt::ReceiptType,
MilliSecondsSinceUnixEpoch, ServerName, UInt, UserId,
};
use tokio::{select, sync::Semaphore};
use tokio::{select, sync::{Semaphore, RwLock}};
use super::abstraction::Tree;
@ -90,7 +90,7 @@ enum TransactionStatus {
}
impl Sending {
pub fn start_handler(&self, db: Arc<Database>, mut receiver: mpsc::UnboundedReceiver<Vec<u8>>) {
pub fn start_handler(&self, db: Arc<RwLock<Database>>, mut receiver: mpsc::UnboundedReceiver<Vec<u8>>) {
tokio::spawn(async move {
let mut futures = FuturesUnordered::new();
@ -98,8 +98,11 @@ impl Sending {
// Retry requests we could not finish yet
let mut initial_transactions = HashMap::<OutgoingKind, Vec<SendingEventType>>::new();
let guard = db.read().await;
for (key, outgoing_kind, event) in
db.sending
guard.sending
.servercurrentevents
.iter()
.filter_map(|(key, _)| {
@ -117,17 +120,19 @@ impl Sending {
"Dropping some current events: {:?} {:?} {:?}",
key, outgoing_kind, event
);
db.sending.servercurrentevents.remove(&key).unwrap();
guard.sending.servercurrentevents.remove(&key).unwrap();
continue;
}
entry.push(event);
}
drop(guard);
for (outgoing_kind, events) in initial_transactions {
current_transaction_status
.insert(outgoing_kind.get_prefix(), TransactionStatus::Running);
futures.push(Self::handle_events(outgoing_kind.clone(), events, &db));
futures.push(Self::handle_events(outgoing_kind.clone(), events, Arc::clone(&db)));
}
loop {
@ -135,15 +140,17 @@ impl Sending {
Some(response) = futures.next() => {
match response {
Ok(outgoing_kind) => {
let guard = db.read().await;
let prefix = outgoing_kind.get_prefix();
for (key, _) in db.sending.servercurrentevents
for (key, _) in guard.sending.servercurrentevents
.scan_prefix(prefix.clone())
{
db.sending.servercurrentevents.remove(&key).unwrap();
guard.sending.servercurrentevents.remove(&key).unwrap();
}
// Find events that have been added since starting the last request
let new_events = db.sending.servernamepduids
let new_events = guard.sending.servernamepduids
.scan_prefix(prefix.clone())
.map(|(k, _)| {
SendingEventType::Pdu(k[prefix.len()..].to_vec())
@ -161,17 +168,19 @@ impl Sending {
SendingEventType::Pdu(b) |
SendingEventType::Edu(b) => {
current_key.extend_from_slice(&b);
db.sending.servercurrentevents.insert(&current_key, &[]).unwrap();
db.sending.servernamepduids.remove(&current_key).unwrap();
guard.sending.servercurrentevents.insert(&current_key, &[]).unwrap();
guard.sending.servernamepduids.remove(&current_key).unwrap();
}
}
}
drop(guard);
futures.push(
Self::handle_events(
outgoing_kind.clone(),
new_events,
&db,
Arc::clone(&db),
)
);
} else {
@ -192,13 +201,15 @@ impl Sending {
},
Some(key) = receiver.next() => {
if let Ok((outgoing_kind, event)) = Self::parse_servercurrentevent(&key) {
let guard = db.read().await;
if let Ok(Some(events)) = Self::select_events(
&outgoing_kind,
vec![(event, key)],
&mut current_transaction_status,
&db
&guard
) {
futures.push(Self::handle_events(outgoing_kind, events, &db));
futures.push(Self::handle_events(outgoing_kind, events, Arc::clone(&db)));
}
}
}
@ -403,8 +414,10 @@ impl Sending {
async fn handle_events(
kind: OutgoingKind,
events: Vec<SendingEventType>,
db: &Database,
db: Arc<RwLock<Database>>,
) -> std::result::Result<OutgoingKind, (OutgoingKind, Error)> {
let db = db.read().await;
match &kind {
OutgoingKind::Appservice(server) => {
let mut pdu_jsons = Vec::new();
@ -543,7 +556,7 @@ impl Sending {
&pusher,
rules_for_user,
&pdu,
db,
&db,
)
.await
.map(|_response| kind.clone())