majjo/Pumpkin
1
0
Fork 0
mirror of https://github.com/Pumpkin-MC/Pumpkin.git synced 2026-08-30 20:14:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

This should prevent some dead locks

Browse Source
This commit is contained in:
Alexander Medvedev
2025-11-23 20:08:47 +01:00
parent 577f51dc1b
commit 03b3ff5e15
7 changed files with 104 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
Cargo.lock generated
View File

@@ -125,9 +125,9 @@ checksum = "a23eb6b1614318a8071c9b2521f36b424b2c83db5eb3a0fead4a6c0809af6e61"
[[package]]
name = "async-compression"
version = "0.4.33"
version = "0.4.34"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "93c1f86859c1af3d514fa19e8323147ff10ea98684e6c7b307912509f50e67b2"
checksum = "0e86f6d3dc9dc4352edeea6b8e499e13e3f5dc3b964d7ca5fd411415a3498473"
dependencies = [
"compression-codecs",
"compression-core",
@@ -416,9 +416,9 @@ dependencies = [
[[package]]
name = "compression-codecs"
version = "0.4.32"
version = "0.4.33"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "680dc087785c5230f8e8843e2e57ac7c1c90488b6a91b88caa265410568f441b"
checksum = "302266479cb963552d11bd042013a58ef1adc56768016c8b82b4199488f2d4ad"
dependencies = [
"compression-core",
"flate2",
@@ -426,9 +426,9 @@ dependencies = [
[[package]]
name = "compression-core"
version = "0.4.30"
version = "0.4.31"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3a9b614a5787ef0c8802a55766480563cb3a93b435898c422ed2a359cf811582"
checksum = "75984efb6ed102a0d42db99afb6c1948f0380d1d91808d5529916e6c08b49d8d"
[[package]]
name = "console-api"
@@ -655,13 +655,13 @@ checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28"
[[package]]
name = "crossfire"
version = "2.1.6"
version = "2.1.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a6b5ad0a8e719cef020e136986d534068ffd35df287ab1635ff242585c02213e"
checksum = "121161e240fcedaef95067636249fd1200e76921a2f0b101dca1648ba6d7ffa5"
dependencies = [
"crossbeam-queue",
"crossbeam-utils",
"futures",
"futures-core",
"parking_lot",
]

2
pumpkin-protocol/Cargo.toml
View File

@@ -27,7 +27,7 @@ aes = "0.8"
cfb8 = "0.8"
# compression
async-compression = { version = "0.4.33", features = ["tokio", "zlib"] }
async-compression = { version = "0.4.34", features = ["tokio", "zlib"] }
take_mut = "0.2.2"
bitflags = "2.10.0"

29
pumpkin-protocol/src/codec/var_int.rs
View File

@@ -38,15 +38,15 @@ impl VarInt {
}
pub fn encode(&self, write: &mut impl Write) -> Result<(), WritingError> {
let mut val = self.0;
loop {
let b: u8 = val as u8 & 0x7F;
// Must cast to u32 to prevent infinite loops on negative i32s
let mut val = self.0 as u32;
while val > 0x7F {
write.write_u8((val as u8) | 0x80)?;
val >>= 7;
write.write_u8(if val == 0 { b } else { b | 0x80 })?;
if val == 0 {
break;
}
}
write.write_u8(val as u8)?;
Ok(())
}
@@ -201,17 +201,14 @@ impl<'de> Deserialize<'de> for VarInt {
impl PacketWrite for VarInt {
fn write<W: Write>(&self, writer: &mut W) -> Result<(), Error> {
let mut val = (self.0 << 1) ^ (self.0 >> 31);
loop {
let b: u8 = val as u8 & 0b01111111;
let mut val = ((self.0 << 1) ^ (self.0 >> 31)) as u32;
while val > 0x7F {
((val as u8 & 0x7F) | 0x80).write(writer)?;
val >>= 7;
if val == 0 {
b.write(writer)?;
break;
} else {
(b | 0b10000000).write(writer)?;
};
}
(val as u8).write(writer)?;
Ok(())
}
}

43
pumpkin-protocol/src/codec/var_long.rs
View File

@@ -27,27 +27,15 @@ impl VarLong {
/// The maximum number of bytes a `VarLong` can occupy.
const MAX_SIZE: NonZeroUsize = NonZeroUsize::new(10).unwrap();
/// Returns the exact number of bytes this VarLong will write when
/// [`Encode::encode`] is called, assuming no error occurs.
pub fn written_size(&self) -> usize {
match self.0 {
0 => 1,
n => (31 - n.leading_zeros() as usize) / 7 + 1,
}
}
pub fn encode(&self, write: &mut impl Write) -> Result<(), WritingError> {
let mut x = self.0;
loop {
let byte = (x & 0x7F) as u8;
x >>= 7;
if x == 0 {
write.write_u8(byte)?;
break;
}
write.write_u8(byte | 0x80)?;
let mut val = self.0 as u64;
while val > 0x7F {
write.write_u8((val as u8) | 0x80)?;
val >>= 7;
}
write.write_u8(val as u8)?;
Ok(())
}
@@ -56,8 +44,8 @@ impl VarLong {
let mut val = 0;
for i in 0..Self::MAX_SIZE.get() {
let byte = read.get_u8()?;
val |= (i64::from(byte) & 0b01111111) << (i * 7);
if byte & 0b10000000 == 0 {
val |= (i64::from(byte) & 0x7F) << (i * 7);
if byte & 0x80 == 0 {
return Ok(VarLong(val));
}
}
@@ -158,15 +146,14 @@ impl<'de> Deserialize<'de> for VarLong {
impl PacketWrite for VarLong {
fn write<W: Write>(&self, writer: &mut W) -> Result<(), Error> {
let mut value = (self.0 << 1) ^ (self.0 >> 63);
loop {
let b: u8 = value as u8 & 127;
value >>= 7;
writer.write_all(&if value == 0 { [b] } else { [b | 128] })?;
if value == 0 {
break;
}
let mut val = ((self.0 << 1) ^ (self.0 >> 63)) as u64;
while val > 0x7F {
((val as u8 & 0x7F) | 0x80).write(writer)?;
val >>= 7;
}
(val as u8).write(writer)?;
Ok(())
}
}

2
pumpkin-world/Cargo.toml
View File

@@ -49,7 +49,7 @@ rand = "=0.10.0-rc.5"
num_cpus = "1.17.0"
rustc-hash = "2.1.1"
slotmap = "1.0"
crossfire = "2.1.6"
crossfire = "2.1.7"
[dev-dependencies]
criterion = { version = "0.7", default-features = false, features = ["html_reports", "async_tokio"] }

106
pumpkin/src/logging.rs
View File

@@ -4,8 +4,12 @@ use rustyline_async::Readline;
use simplelog::{CombinedLogger, Config, SharedLogger, WriteLogger};
use std::fmt::format;
use std::fs::File;
use std::io::BufWriter;
use std::path::Path;
use std::io::{self, BufWriter};
use std::path::PathBuf;
use time::{Duration, OffsetDateTime, UtcOffset};
const LOG_DIR: &str = "logs";
const MAX_ATTEMPTS: u32 = 100;
/// A wrapper for our logger to hold the terminal input while no input is expected in order to
/// properly flush logs to the output while they happen instead of batched
@@ -34,65 +38,85 @@ impl GzipRollingLogger {
filename: String,
) -> Result<Box<Self>, Box<dyn std::error::Error>> {
let now = time::OffsetDateTime::now_utc();
std::fs::create_dir_all("logs")?;
std::fs::create_dir_all(LOG_DIR)?;
let latest_path = PathBuf::from(LOG_DIR).join(&filename);
// If latest.log exists, we will gzip it
if Path::new(&format!("logs/{filename}")).exists() {
let new_filename = Self::new_filename(false);
let mut file = File::open(format!("logs/{filename}"))?;
if latest_path.exists() {
eprintln!(
"Found existing log file at '{}', gzipping it now...",
latest_path.display()
);
let new_gz_path = Self::new_filename(true)?;
let mut file = File::open(&latest_path)?;
let mut encoder = GzEncoder::new(
BufWriter::new(File::create(&new_filename)?),
BufWriter::new(File::create(&new_gz_path)?),
flate2::Compression::best(),
);
std::io::copy(&mut file, &mut encoder)?;
io::copy(&mut file, &mut encoder)?;
encoder.finish()?;
std::fs::remove_file(&latest_path)?;
}
let new_logger = WriteLogger::new(log_level, config.clone(), File::create(&latest_path)?);
Ok(Box::new(Self {
log_level,
data: std::sync::Mutex::new(GzipRollingLoggerData {
current_day_of_month: now.day(),
last_rotate_time: now,
latest_filename: filename.clone(),
latest_logger: *WriteLogger::new(
log_level,
config.clone(),
File::create(format!("logs/{filename}")).unwrap(),
),
latest_filename: filename,
latest_logger: *new_logger,
}),
config,
}))
}
pub fn new_filename(yesterday: bool) -> String {
let mut now = time::OffsetDateTime::now_utc()
.to_offset(time::UtcOffset::current_local_offset().unwrap_or(time::UtcOffset::UTC));
if yesterday {
now -= time::Duration::days(1)
}
let base_filename = format!("{}-{:02}-{:02}", now.year(), now.month() as u8, now.day());
pub fn new_filename(yesterday: bool) -> Result<PathBuf, Box<dyn std::error::Error>> {
let local_offset = UtcOffset::current_local_offset().unwrap_or(UtcOffset::UTC);
let mut now = OffsetDateTime::now_utc().to_offset(local_offset);
let mut id = 1;
loop {
let filename = format!("logs/{base_filename}-{id}.log.gz");
if !Path::new(&filename).exists() {
return filename;
}
id += 1;
if yesterday {
now -= Duration::days(1);
}
let date_format = format!("{}-{:02}-{:02}", now.year(), now.month() as u8, now.day());
let log_path = PathBuf::from(LOG_DIR);
for id in 1..=MAX_ATTEMPTS {
let filename = log_path.join(format!("{}-{}.log.gz", date_format, id));
if !filename.exists() {
return Ok(filename);
}
}
Err(format!(
"Failed to find a unique log filename for date {} after {} attempts.",
date_format, MAX_ATTEMPTS
)
.into())
}
fn rotate_log(&self) -> Result<(), Box<dyn std::error::Error>> {
let now = time::OffsetDateTime::now_utc();
let mut data = self.data.lock().unwrap();
let new_filename = Self::new_filename(true);
let mut file = File::open(format!("logs/{}", data.latest_filename))?;
let new_gz_path = Self::new_filename(true)?;
let latest_path = PathBuf::from(LOG_DIR).join(&data.latest_filename);
let mut file = File::open(&latest_path)?;
let mut encoder = GzEncoder::new(
BufWriter::new(File::create(format!("logs/{new_filename}"))?),
BufWriter::new(File::create(&new_gz_path)?),
flate2::Compression::best(),
);
std::io::copy(&mut file, &mut encoder)?;
io::copy(&mut file, &mut encoder)?;
encoder.finish()?;
data.current_day_of_month = now.day();
@@ -100,7 +124,7 @@ impl GzipRollingLogger {
data.latest_logger = *WriteLogger::new(
self.log_level,
self.config.clone(),
File::create(format!("logs/{}", data.latest_filename)).unwrap(),
File::create(&latest_path)?,
);
Ok(())
}
@@ -108,21 +132,19 @@ impl GzipRollingLogger {
fn remove_ansi_color_code(s: &str) -> String {
let mut result = String::with_capacity(s.len());
let mut in_escape_sequence = false;
let mut it = s.chars();
for c in s.chars() {
if in_escape_sequence {
if c.is_ascii_alphabetic() {
// This broadly covers 'm', 'J', 'H', etc.
in_escape_sequence = false;
while let Some(c) = it.next() {
if c == '\x1b' {
for c_seq in it.by_ref() {
if c_seq.is_ascii_alphabetic() {
break;
}
}
} else if c == '\x1B' {
in_escape_sequence = true;
} else {
result.push(c);
}
}
result
}

5
pumpkin/src/net/java/mod.rs
View File

@@ -159,10 +159,7 @@ impl JavaClient {
///
/// * `server`: A reference to the `Server` instance.
pub async fn process_packets(self: &Arc<Self>, server: &Arc<Server>) {
loop {
let packet = self.get_packet().await;
let Some(packet) = packet else { break };
while let Some(packet) = self.get_packet().await {
if let Err(error) = self.handle_packet(server, &packet).await {
let text = format!("Error while reading incoming packet {error}");
log::error!(