You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2025/07/03 12:55:37 UTC
(iggy) branch io_uring_tpc_compio created (now 6184be0c)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a change to branch io_uring_tpc_compio
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 6184be0c compio works
This branch includes the following new commits:
new 6184be0c compio works
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 01/01: compio works
Posted by gk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch io_uring_tpc_compio
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 6184be0cd94c9a8366eb0f2dea83175702fcce2a
Author: numinex <g....@gmail.com>
AuthorDate: Thu Jul 3 14:50:37 2025 +0200
compio works
---
Cargo.lock | 418 ++++++++++++++-------
core/server/Cargo.toml | 4 +-
.../handlers/messages/poll_messages_handler.rs | 25 +-
core/server/src/binary/sender.rs | 17 +-
core/server/src/bootstrap.rs | 35 +-
.../src/compat/index_rebuilding/index_rebuilder.rs | 34 +-
core/server/src/io/file.rs | 117 ------
core/server/src/io/fs_utils.rs | 2 +-
core/server/src/io/mod.rs | 3 -
core/server/src/io/reader.rs | 44 ---
core/server/src/io/writer.rs | 57 ---
core/server/src/log/runtime.rs | 48 +++
core/server/src/main.rs | 26 +-
core/server/src/quic/quic_sender.rs | 5 +-
core/server/src/shard/mod.rs | 6 +-
core/server/src/shard/system/snapshot/mod.rs | 41 +-
core/server/src/shard/system/storage.rs | 9 +-
core/server/src/shard/system/streams.rs | 2 +-
core/server/src/shard/system/topics.rs | 12 +-
core/server/src/shard/task_registry.rs | 8 +-
core/server/src/shard/tasks/messages.rs | 2 +-
core/server/src/state/file.rs | 36 +-
core/server/src/streaming/partitions/storage.rs | 27 +-
core/server/src/streaming/persistence/persister.rs | 24 +-
core/server/src/streaming/persistence/task.rs | 10 +-
.../src/streaming/segments/indexes/index_reader.rs | 16 +-
.../src/streaming/segments/indexes/index_writer.rs | 15 +-
.../streaming/segments/messages/messages_reader.rs | 20 +-
.../streaming/segments/messages/messages_writer.rs | 19 +-
core/server/src/streaming/segments/messages/mod.rs | 31 +-
core/server/src/streaming/segments/segment.rs | 6 +-
core/server/src/streaming/streams/storage.rs | 6 +-
core/server/src/streaming/topics/storage.rs | 6 +-
core/server/src/streaming/utils/file.rs | 14 +-
core/server/src/streaming/utils/pooled_buffer.rs | 34 +-
core/server/src/tcp/connection_handler.rs | 1 +
core/server/src/tcp/sender.rs | 60 ++-
core/server/src/tcp/tcp_listener.rs | 12 +-
core/server/src/tcp/tcp_sender.rs | 9 +-
core/server/src/tcp/tcp_tls_listener.rs | 185 ++++-----
core/server/src/tcp/tcp_tls_sender.rs | 18 +-
41 files changed, 753 insertions(+), 711 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index ccecbd7b..d9383e8a 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -142,7 +142,7 @@ dependencies = [
"actix-utils",
"futures-core",
"futures-util",
- "mio 1.0.4",
+ "mio",
"socket2",
"tokio",
"tracing",
@@ -307,6 +307,15 @@ dependencies = [
"memchr",
]
+[[package]]
+name = "aligned-array"
+version = "1.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e05c92d086290f52938013f6242ac62bf7d401fab8ad36798a609faa65c3fd2c"
+dependencies = [
+ "generic-array",
+]
+
[[package]]
name = "alloc-no-stdlib"
version = "2.0.4"
@@ -578,6 +587,12 @@ dependencies = [
"syn 2.0.104",
]
+[[package]]
+name = "async-task"
+version = "4.7.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b75356056920673b02621b35afd0f7dda9306d03c79a30f5c56c44cf256e3de"
+
[[package]]
name = "async-trait"
version = "0.1.88"
@@ -643,17 +658,6 @@ dependencies = [
"webpki-roots 0.26.11",
]
-[[package]]
-name = "auto-const-array"
-version = "0.2.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fd73835ad7deb4bd2b389e6f10333b143f025d607c55ca04c66a0bcc6bb2fc6d"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.104",
-]
-
[[package]]
name = "autocfg"
version = "1.5.0"
@@ -1529,6 +1533,216 @@ dependencies = [
"static_assertions",
]
+[[package]]
+name = "compio"
+version = "0.15.0"
+source = "git+https://github.com/compio-rs/compio.git?rev=5504e02ec2e821e9b024006224c34c06646410cc#5504e02ec2e821e9b024006224c34c06646410cc"
+dependencies = [
+ "compio-buf",
+ "compio-dispatcher",
+ "compio-driver",
+ "compio-fs",
+ "compio-io",
+ "compio-log",
+ "compio-macros",
+ "compio-net",
+ "compio-process",
+ "compio-quic",
+ "compio-runtime",
+ "compio-signal",
+ "compio-tls",
+]
+
+[[package]]
+name = "compio-buf"
+version = "0.6.0"
+source = "git+https://github.com/compio-rs/compio.git?rev=5504e02ec2e821e9b024006224c34c06646410cc#5504e02ec2e821e9b024006224c34c06646410cc"
+dependencies = [
+ "arrayvec",
+ "bytes",
+ "libc",
+]
+
+[[package]]
+name = "compio-dispatcher"
+version = "0.7.0"
+source = "git+https://github.com/compio-rs/compio.git?rev=5504e02ec2e821e9b024006224c34c06646410cc#5504e02ec2e821e9b024006224c34c06646410cc"
+dependencies = [
+ "compio-driver",
+ "compio-runtime",
+ "flume",
+ "futures-channel",
+]
+
+[[package]]
+name = "compio-driver"
+version = "0.8.1"
+source = "git+https://github.com/compio-rs/compio.git?rev=5504e02ec2e821e9b024006224c34c06646410cc#5504e02ec2e821e9b024006224c34c06646410cc"
+dependencies = [
+ "aligned-array",
+ "cfg-if",
+ "cfg_aliases",
+ "compio-buf",
+ "compio-log",
+ "crossbeam-channel",
+ "crossbeam-queue",
+ "futures-util",
+ "io-uring",
+ "io_uring_buf_ring",
+ "libc",
+ "once_cell",
+ "paste",
+ "polling",
+ "slab",
+ "socket2",
+ "windows-sys 0.52.0",
+]
+
+[[package]]
+name = "compio-fs"
+version = "0.8.0"
+source = "git+https://github.com/compio-rs/compio.git?rev=5504e02ec2e821e9b024006224c34c06646410cc#5504e02ec2e821e9b024006224c34c06646410cc"
+dependencies = [
+ "cfg-if",
+ "cfg_aliases",
+ "compio-buf",
+ "compio-driver",
+ "compio-io",
+ "compio-runtime",
+ "libc",
+ "os_pipe",
+ "widestring",
+ "windows-sys 0.52.0",
+]
+
+[[package]]
+name = "compio-io"
+version = "0.7.0"
+source = "git+https://github.com/compio-rs/compio.git?rev=5504e02ec2e821e9b024006224c34c06646410cc#5504e02ec2e821e9b024006224c34c06646410cc"
+dependencies = [
+ "compio-buf",
+ "futures-util",
+ "paste",
+ "pin-project-lite",
+]
+
+[[package]]
+name = "compio-log"
+version = "0.1.0"
+source = "git+https://github.com/compio-rs/compio.git?rev=5504e02ec2e821e9b024006224c34c06646410cc#5504e02ec2e821e9b024006224c34c06646410cc"
+dependencies = [
+ "tracing",
+]
+
+[[package]]
+name = "compio-macros"
+version = "0.1.2"
+source = "git+https://github.com/compio-rs/compio.git?rev=5504e02ec2e821e9b024006224c34c06646410cc#5504e02ec2e821e9b024006224c34c06646410cc"
+dependencies = [
+ "proc-macro-crate 3.3.0",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.104",
+]
+
+[[package]]
+name = "compio-net"
+version = "0.8.0"
+source = "git+https://github.com/compio-rs/compio.git?rev=5504e02ec2e821e9b024006224c34c06646410cc#5504e02ec2e821e9b024006224c34c06646410cc"
+dependencies = [
+ "cfg-if",
+ "compio-buf",
+ "compio-driver",
+ "compio-io",
+ "compio-runtime",
+ "either",
+ "libc",
+ "once_cell",
+ "socket2",
+ "widestring",
+ "windows-sys 0.52.0",
+]
+
+[[package]]
+name = "compio-process"
+version = "0.5.0"
+source = "git+https://github.com/compio-rs/compio.git?rev=5504e02ec2e821e9b024006224c34c06646410cc#5504e02ec2e821e9b024006224c34c06646410cc"
+dependencies = [
+ "cfg-if",
+ "compio-buf",
+ "compio-driver",
+ "compio-io",
+ "compio-runtime",
+ "futures-util",
+ "windows-sys 0.52.0",
+]
+
+[[package]]
+name = "compio-quic"
+version = "0.4.0"
+source = "git+https://github.com/compio-rs/compio.git?rev=5504e02ec2e821e9b024006224c34c06646410cc#5504e02ec2e821e9b024006224c34c06646410cc"
+dependencies = [
+ "cfg_aliases",
+ "compio-buf",
+ "compio-io",
+ "compio-log",
+ "compio-net",
+ "compio-runtime",
+ "flume",
+ "futures-util",
+ "libc",
+ "quinn-proto",
+ "rustc-hash 2.1.1",
+ "rustls",
+ "thiserror 2.0.12",
+ "windows-sys 0.52.0",
+]
+
+[[package]]
+name = "compio-runtime"
+version = "0.8.1"
+source = "git+https://github.com/compio-rs/compio.git?rev=5504e02ec2e821e9b024006224c34c06646410cc#5504e02ec2e821e9b024006224c34c06646410cc"
+dependencies = [
+ "async-task",
+ "cfg-if",
+ "compio-buf",
+ "compio-driver",
+ "compio-log",
+ "crossbeam-queue",
+ "futures-util",
+ "libc",
+ "once_cell",
+ "scoped-tls",
+ "slab",
+ "socket2",
+ "windows-sys 0.52.0",
+]
+
+[[package]]
+name = "compio-signal"
+version = "0.6.0"
+source = "git+https://github.com/compio-rs/compio.git?rev=5504e02ec2e821e9b024006224c34c06646410cc#5504e02ec2e821e9b024006224c34c06646410cc"
+dependencies = [
+ "compio-buf",
+ "compio-driver",
+ "compio-runtime",
+ "libc",
+ "once_cell",
+ "os_pipe",
+ "slab",
+ "windows-sys 0.52.0",
+]
+
+[[package]]
+name = "compio-tls"
+version = "0.6.0"
+source = "git+https://github.com/compio-rs/compio.git?rev=5504e02ec2e821e9b024006224c34c06646410cc#5504e02ec2e821e9b024006224c34c06646410cc"
+dependencies = [
+ "compio-buf",
+ "compio-io",
+ "rustls",
+]
+
[[package]]
name = "concurrent-queue"
version = "2.5.0"
@@ -1696,6 +1910,17 @@ version = "0.8.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"
+[[package]]
+name = "core_affinity"
+version = "0.8.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a034b3a7b624016c6e13f5df875747cc25f884156aad2abd12b6c46797971342"
+dependencies = [
+ "libc",
+ "num_cpus",
+ "winapi",
+]
+
[[package]]
name = "cpufeatures"
version = "0.2.17"
@@ -1846,7 +2071,7 @@ version = "3.4.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "46f93780a459b7d656ef7f071fe699c4d3d2cb201c4b24d085b6ddc505276e73"
dependencies = [
- "nix 0.30.1",
+ "nix",
"windows-sys 0.59.0",
]
@@ -2682,15 +2907,6 @@ dependencies = [
"slab",
]
-[[package]]
-name = "fxhash"
-version = "0.2.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c31b6d751ae2c7f11320402d34e41349dd1016f8d5d45e48c4312bc8625af50c"
-dependencies = [
- "byteorder",
-]
-
[[package]]
name = "generator"
version = "0.8.5"
@@ -4143,14 +4359,26 @@ dependencies = [
[[package]]
name = "io-uring"
-version = "0.6.4"
+version = "0.7.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "595a0399f411a508feb2ec1e970a4a30c249351e30208960d58298de8660b0e5"
+checksum = "b86e202f00093dcba4275d4636b93ef9dd75d025ae560d2521b45ea28ab49013"
dependencies = [
- "bitflags 1.3.2",
+ "bitflags 2.9.1",
+ "cfg-if",
"libc",
]
+[[package]]
+name = "io_uring_buf_ring"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8a8867874ff5758b47c1dac069e6e86541432f9da8be9111c5e94154134f07d0"
+dependencies = [
+ "bytes",
+ "io-uring",
+ "rustix 1.0.7",
+]
+
[[package]]
name = "ipnet"
version = "2.11.0"
@@ -4657,15 +4885,6 @@ version = "2.7.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "32a282da65faaf38286cf3be983213fcf1d2e2a58700e808f83f4ea9a4804bc0"
-[[package]]
-name = "memoffset"
-version = "0.7.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5de893c32cde5f383baa4c04c5d6dbdd735cfd4a794b0debdb2bb1b421da5ff4"
-dependencies = [
- "autocfg",
-]
-
[[package]]
name = "mimalloc"
version = "0.1.47"
@@ -4715,18 +4934,6 @@ dependencies = [
"adler2",
]
-[[package]]
-name = "mio"
-version = "0.8.11"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a4a650543ca06a924e8b371db273b2756685faae30f8487da1b56505a8f78b0c"
-dependencies = [
- "libc",
- "log",
- "wasi 0.11.1+wasi-snapshot-preview1",
- "windows-sys 0.48.0",
-]
-
[[package]]
name = "mio"
version = "1.0.4"
@@ -4787,62 +4994,6 @@ dependencies = [
"uuid",
]
-[[package]]
-name = "monoio"
-version = "0.2.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3bd0f8bcde87b1949f95338b547543fcab187bc7e7a5024247e359a5e828ba6a"
-dependencies = [
- "auto-const-array",
- "bytes",
- "flume",
- "fxhash",
- "io-uring",
- "libc",
- "memchr",
- "mio 0.8.11",
- "monoio-macros",
- "nix 0.26.4",
- "once_cell",
- "pin-project-lite",
- "socket2",
- "threadpool",
- "windows-sys 0.48.0",
-]
-
-[[package]]
-name = "monoio-io-wrapper"
-version = "0.1.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4bcfaa76e5daf87cc4d31b4d1b6bc93c12db59c19df50b9200afdbde42077655"
-dependencies = [
- "monoio",
-]
-
-[[package]]
-name = "monoio-macros"
-version = "0.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "176a5f5e69613d9e88337cf2a65e11135332b4efbcc628404a7c555e4452084c"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.104",
-]
-
-[[package]]
-name = "monoio-rustls"
-version = "0.4.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6e31f422825bd7fb19957af6eaf89d7234ba143fcc0e515f5a2f526e332d1875"
-dependencies = [
- "bytes",
- "monoio",
- "monoio-io-wrapper",
- "rustls",
- "thiserror 1.0.69",
-]
-
[[package]]
name = "nanorand"
version = "0.7.0"
@@ -4858,19 +5009,6 @@ version = "6.6.666"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cf5a574dadd7941adeaa71823ecba5e28331b8313fb2e1c6a5c7e5981ea53ad6"
-[[package]]
-name = "nix"
-version = "0.26.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "598beaf3cc6fdd9a5dfb1630c2800c7acd31df7aaf0f565796fba2b53ca1af1b"
-dependencies = [
- "bitflags 1.3.2",
- "cfg-if",
- "libc",
- "memoffset",
- "pin-utils",
-]
-
[[package]]
name = "nix"
version = "0.30.1"
@@ -4935,7 +5073,7 @@ dependencies = [
"kqueue",
"libc",
"log",
- "mio 1.0.4",
+ "mio",
"notify-types",
"walkdir",
"windows-sys 0.59.0",
@@ -5351,6 +5489,16 @@ dependencies = [
"hashbrown 0.14.5",
]
+[[package]]
+name = "os_pipe"
+version = "1.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "db335f4760b14ead6290116f2427bf33a14d4f0617d49f78a246de10c1831224"
+dependencies = [
+ "libc",
+ "windows-sys 0.59.0",
+]
+
[[package]]
name = "overload"
version = "0.1.1"
@@ -5602,6 +5750,21 @@ version = "0.3.32"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c"
+[[package]]
+name = "polling"
+version = "3.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b53a684391ad002dd6a596ceb6c74fd004fdce75f4be2e3f615068abbea5fd50"
+dependencies = [
+ "cfg-if",
+ "concurrent-queue",
+ "hermit-abi",
+ "pin-project-lite",
+ "rustix 1.0.7",
+ "tracing",
+ "windows-sys 0.59.0",
+]
+
[[package]]
name = "polonius-the-crab"
version = "0.2.1"
@@ -6874,7 +7037,9 @@ dependencies = [
"bytes",
"chrono",
"clap",
+ "compio",
"console-subscriber",
+ "core_affinity",
"crossbeam",
"ctrlc",
"dashmap",
@@ -6894,9 +7059,7 @@ dependencies = [
"mimalloc",
"mockall",
"moka",
- "monoio",
- "monoio-rustls",
- "nix 0.30.1",
+ "nix",
"once_cell",
"opentelemetry",
"opentelemetry-appender-tracing",
@@ -7409,15 +7572,6 @@ dependencies = [
"cfg-if",
]
-[[package]]
-name = "threadpool"
-version = "1.8.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d050e60b33d41c19108b32cea32164033a9013fe3b46cbd4457559bfbf77afaa"
-dependencies = [
- "num_cpus",
-]
-
[[package]]
name = "time"
version = "0.3.41"
@@ -7494,7 +7648,7 @@ dependencies = [
"backtrace",
"bytes",
"libc",
- "mio 1.0.4",
+ "mio",
"parking_lot 0.12.4",
"pin-project-lite",
"signal-hook-registry",
@@ -8330,6 +8484,12 @@ dependencies = [
"safe_arch",
]
+[[package]]
+name = "widestring"
+version = "1.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dd7cf3379ca1aac9eea11fba24fd7e315d621f8dfe35c8d7d2be8b793726e07d"
+
[[package]]
name = "winapi"
version = "0.3.9"
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index f91ac15c..57ab996a 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -63,12 +63,12 @@ human-repr = { workspace = true }
iggy_common = { workspace = true }
jsonwebtoken = "9.3.1"
socket2 = "0.5.10"
+core_affinity = "0.8.0"
lending-iterator = "0.1.7"
hash32 = "1.0.0"
mimalloc = { workspace = true, optional = true }
moka = { version = "0.12.10", features = ["future"] }
-monoio = { version = "0.2.4", features = ["mkdirat", "unlinkat", "renameat", "sync"] }
-monoio-rustls = "0.4.0"
+compio = { git = "https://github.com/compio-rs/compio.git", rev = "5504e02ec2e821e9b024006224c34c06646410cc", features = ["runtime", "macros", "io-uring", "time", "rustls"] }
nix = { version = "0.30", features = ["fs"] }
once_cell = "1.21.3"
opentelemetry = { version = "0.30.0", features = ["trace", "logs"] }
diff --git a/core/server/src/binary/handlers/messages/poll_messages_handler.rs b/core/server/src/binary/handlers/messages/poll_messages_handler.rs
index dfc4a89f..2f02a039 100644
--- a/core/server/src/binary/handlers/messages/poll_messages_handler.rs
+++ b/core/server/src/binary/handlers/messages/poll_messages_handler.rs
@@ -27,8 +27,10 @@ use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardMessage, ShardRequest, ShardRequestPayload};
use crate::streaming::segments::IggyMessagesBatchSet;
use crate::streaming::session::Session;
+use crate::streaming::utils::PooledBuffer;
use crate::to_iovec;
use anyhow::Result;
+use bytes::BytesMut;
use error_set::ErrContext;
use iggy_common::{IggyError, PollMessages};
use std::io::IoSlice;
@@ -76,7 +78,7 @@ impl ServerCommandHandler for PollMessages {
let user_id = session.get_user_id();
let client_id = session.client_id;
- let (metadata, batch) = shard
+ let (metadata, mut batch) = shard
.poll_messages(
client_id,
user_id,
@@ -96,16 +98,19 @@ impl ServerCommandHandler for PollMessages {
let response_length = 4 + 8 + 4 + batch.size();
let response_length_bytes = response_length.to_le_bytes();
- let partition_id = metadata.partition_id.to_le_bytes();
- let current_offset = metadata.current_offset.to_le_bytes();
- let count = batch.count().to_le_bytes();
+ let mut bufs = Vec::with_capacity(batch.containers_count() + 5);
+ let mut partition_id_buf = PooledBuffer::with_capacity(4);
+ let mut current_offset_buf = PooledBuffer::with_capacity(8);
+ let mut count_buf = PooledBuffer::with_capacity(4);
+ partition_id_buf.put_u32_le(metadata.partition_id);
+ current_offset_buf.put_u64_le(metadata.current_offset);
+ count_buf.put_u32_le(batch.count());
- let mut iovecs = Vec::with_capacity(batch.containers_count() + 3);
- iovecs.push(to_iovec(&partition_id));
- iovecs.push(to_iovec(¤t_offset));
- iovecs.push(to_iovec(&count));
+ bufs.push(partition_id_buf);
+ bufs.push(current_offset_buf);
+ bufs.push(count_buf);
- iovecs.extend(batch.iter().map(|m| to_iovec(&m)));
+ batch.iter_mut().for_each(|m| bufs.push(m.take_messages()));
trace!(
"Sending {} messages to client ({} bytes) to client",
batch.count(),
@@ -113,7 +118,7 @@ impl ServerCommandHandler for PollMessages {
);
sender
- .send_ok_response_vectored(&response_length_bytes, iovecs)
+ .send_ok_response_vectored(&response_length_bytes, bufs)
.await?;
Ok(())
}
diff --git a/core/server/src/binary/sender.rs b/core/server/src/binary/sender.rs
index 45066c75..96bed3d5 100644
--- a/core/server/src/binary/sender.rs
+++ b/core/server/src/binary/sender.rs
@@ -19,17 +19,17 @@
use std::future::Future;
use std::io::IoSlice;
+use crate::streaming::utils::PooledBuffer;
use crate::tcp::tcp_sender::TcpSender;
use crate::tcp::tcp_tls_sender::TcpTlsSender;
use crate::{quic::quic_sender::QuicSender, server_error::ServerError};
use bytes::BytesMut;
+use compio::buf::{IoBuf, IoBufMut};
+use compio::io::{AsyncReadExt, AsyncWriteExt};
+use compio::net::TcpStream;
use iggy_common::IggyError;
-use monoio::buf::IoBufMut;
-use monoio::io::{AsyncReadRent, AsyncWriteRent};
-use monoio::net::TcpStream;
use nix::libc;
use quinn::{RecvStream, SendStream};
-use monoio_rustls::{ServerTlsStream, TlsStream};
macro_rules! forward_async_methods {
(
@@ -65,7 +65,7 @@ pub trait Sender {
fn send_ok_response_vectored(
&mut self,
length: &[u8],
- slices: Vec<libc::iovec>,
+ slices: Vec<PooledBuffer>,
) -> impl Future<Output = Result<(), IggyError>>;
fn send_error_response(
&mut self,
@@ -85,8 +85,9 @@ impl SenderKind {
Self::Tcp(TcpSender { stream })
}
- pub fn get_tcp_tls_sender(stream: ServerTlsStream<TcpStream>) -> Self {
- Self::TcpTls(TcpTlsSender { stream })
+ pub fn get_tcp_tls_sender(stream: ()) -> Self {
+ todo!();
+ //Self::TcpTls(TcpTlsSender { stream })
}
pub fn get_quic_sender(send_stream: SendStream, recv_stream: RecvStream) -> Self {
@@ -100,7 +101,7 @@ impl SenderKind {
async fn read<B: IoBufMut>(&mut self, buffer: B) -> (Result<usize, IggyError>, B);
async fn send_empty_ok_response(&mut self) -> Result<(), IggyError>;
async fn send_ok_response(&mut self, payload: &[u8]) -> Result<(), IggyError>;
- async fn send_ok_response_vectored(&mut self, length: &[u8], slices: Vec<libc::iovec>) -> Result<(), IggyError>;
+ async fn send_ok_response_vectored(&mut self, length: &[u8], slices: Vec<PooledBuffer>) -> Result<(), IggyError>;
async fn send_error_response(&mut self, error: IggyError) -> Result<(), IggyError>;
async fn shutdown(&mut self) -> Result<(), ServerError>;
}
diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs
index ddd0599c..6a2eb8af 100644
--- a/core/server/src/bootstrap.rs
+++ b/core/server/src/bootstrap.rs
@@ -1,3 +1,4 @@
+use compio::{fs::create_dir_all, runtime::Runtime};
use iggy_common::{
IggyError,
defaults::{
@@ -5,7 +6,6 @@ use iggy_common::{
MIN_PASSWORD_LENGTH, MIN_USERNAME_LENGTH,
},
};
-use monoio::{Buildable, Driver, Runtime, fs::create_dir_all, time::TimeDriver};
use tracing::info;
use crate::{
@@ -121,24 +121,21 @@ pub fn create_root_user() -> User {
user
}
-pub fn create_default_executor<D>() -> Runtime<D>
-where
- D: Driver + Buildable,
-{
- let builder = monoio::RuntimeBuilder::<D>::new();
- let rt = Buildable::build(builder).expect("Failed to create default runtime");
- rt
-}
-
-pub fn create_shard_executor() -> Runtime<TimeDriver<monoio::IoUringDriver>> {
- // TODO: Figure out what else we could tweak there
- // We for sure want to disable the userspace interrupts on new cq entry (set_coop_taskrun)
- // TODO: Shall we make the size of ring be configureable ?
- let builder = monoio::RuntimeBuilder::<monoio::IoUringDriver>::new()
- //.uring_builder(urb.setup_coop_taskrun()) // broken shit.
- .with_entries(1024) // Default size
- .enable_timer();
- let rt = Buildable::build(builder).expect("Failed to create default runtime");
+pub fn create_shard_executor() -> Runtime {
+ //TODO: The event intererval tick, could be configureed based on the fact
+ // How many clients we expect to have connected.
+ // This roughly estimates the number of tasks we will create.
+ let proactor = compio::driver::ProactorBuilder::new()
+ .capacity(4096)
+ .coop_taskrun(true)
+ .taskrun_flag(false) //TODO: Try enabling this.
+ .thread_pool_limit(0)
+ .to_owned();
+ let rt = compio::runtime::RuntimeBuilder::new()
+ .with_proactor(proactor)
+ .event_interval(69)
+ .build()
+ .unwrap();
rt
}
diff --git a/core/server/src/compat/index_rebuilding/index_rebuilder.rs b/core/server/src/compat/index_rebuilding/index_rebuilder.rs
index 1eebdacd..06204bf8 100644
--- a/core/server/src/compat/index_rebuilding/index_rebuilder.rs
+++ b/core/server/src/compat/index_rebuilding/index_rebuilder.rs
@@ -16,12 +16,11 @@
* under the License.
*/
-use crate::io::file::IggyFile;
-use crate::io::writer::IggyWriter;
use crate::streaming::utils::file;
-use crate::{io::reader::IggyReader, server_error::CompatError};
+use crate::server_error::CompatError;
+use async_zip::tokio::write;
use iggy_common::{IGGY_MESSAGE_HEADER_SIZE, IggyMessageHeader};
-use monoio::io::{AsyncReadRent, AsyncReadRentExt, AsyncWriteRent, AsyncWriteRentExt};
+use compio::{fs::File, io::{AsyncBufRead, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader, BufWriter}};
use std::io::{Seek, SeekFrom};
pub struct IndexRebuilder {
@@ -40,17 +39,17 @@ impl IndexRebuilder {
}
async fn read_message_header(
- reader: &mut IggyReader<IggyFile>,
+ reader: &mut BufReader<std::io::Cursor<File>>,
) -> Result<IggyMessageHeader, std::io::Error> {
let buf = [0u8; IGGY_MESSAGE_HEADER_SIZE];
- let (result, buf) = reader.read_exact(Box::new(buf)).await;
+ let (result, buf) = reader.read_exact(Box::new(buf)).await.into();
result?;
IggyMessageHeader::from_raw_bytes(&*buf)
.map_err(|_| std::io::Error::from(std::io::ErrorKind::InvalidData))
}
async fn write_index_entry(
- writer: &mut IggyWriter<IggyFile>,
+ writer: &mut BufWriter<std::io::Cursor<File>>,
header: &IggyMessageHeader,
position: usize,
start_offset: u64,
@@ -58,26 +57,29 @@ impl IndexRebuilder {
// Write offset (4 bytes) - base_offset + last_offset_delta - start_offset
let offset = start_offset - header.offset;
debug_assert!(offset <= u32::MAX as u64);
- let (result, _) = writer.write_all(Box::new(offset.to_le_bytes())).await;
+ let (result, _) = writer.write_all(Box::new(offset.to_le_bytes())).await.into();
result?;
// Write position (4 bytes)
- let (result, _) = writer.write_all(Box::new(position.to_le_bytes())).await;
+ let (result, _) = writer.write_all(Box::new(position.to_le_bytes())).await.into();
result?;
// Write timestamp (8 bytes)
let (result, _) = writer
.write_all(Box::new(header.timestamp.to_le_bytes()))
- .await;
+ .await.into();
result?;
Ok(())
}
pub async fn rebuild(&self) -> Result<(), CompatError> {
- let mut reader =
- IggyReader::new(IggyFile::new(file::open(&self.messages_file_path).await?));
- let mut writer = IggyWriter::new(IggyFile::new(file::overwrite(&self.index_path).await?));
+ let read_cursor = std::io::Cursor::new(file::open(&self.messages_file_path).await?);
+ let write_cursor = std::io::Cursor::new(
+ file::overwrite(&self.index_path).await?
+ );
+ let mut reader = BufReader::new(read_cursor);
+ let mut writer = BufWriter::new(write_cursor);
let mut position = 0;
let mut next_position;
@@ -93,9 +95,9 @@ impl IndexRebuilder {
.await?;
// Skip message payload and headers
- reader.seek(SeekFrom::Current(
- header.payload_length as i64 + header.user_headers_length as i64,
- ))?;
+ reader.consume(
+ header.payload_length as usize + header.user_headers_length as usize
+ );
// Update position for next iteration
position = next_position;
diff --git a/core/server/src/io/file.rs b/core/server/src/io/file.rs
deleted file mode 100644
index 95129aa8..00000000
--- a/core/server/src/io/file.rs
+++ /dev/null
@@ -1,117 +0,0 @@
-use std::io::SeekFrom;
-
-use monoio::{
- BufResult,
- buf::{IoBuf, IoBufMut, IoVecBuf, IoVecBufMut, IoVecWrapper},
- fs::{File, Metadata},
- io::{AsyncReadRent, AsyncWriteRent},
-};
-
-/// Wrapper around `monoio::fs::File` to provide a consistent API for reading and writing files
-/// in an asynchronous context. This struct maintains the current position in the file and provides
-/// methods to read and write data at specific positions.
-#[derive(Debug)]
-pub struct IggyFile {
- file: File,
- position: u64,
-}
-
-impl From<File> for IggyFile {
- fn from(file: File) -> Self {
- Self { file, position: 0 }
- }
-}
-
-impl std::io::Seek for IggyFile {
- /// This method doesn't do bound checking aswell as, the `End` variant is not supported.
- fn seek(&mut self, pos: SeekFrom) -> std::io::Result<u64> {
- self.position = match pos {
- SeekFrom::Start(n) => n as i64,
- SeekFrom::End(_) => unreachable!("End variant is not supported in IggyFile"),
- SeekFrom::Current(n) => self.position as i64 + n,
- } as u64;
- Ok(self.position)
- }
-}
-
-impl IggyFile {
- pub fn new(file: File) -> Self {
- Self { file, position: 0 }
- }
-
- pub async fn metadata(&self) -> std::io::Result<Metadata> {
- self.file.metadata().await
- }
-
- pub async fn write_at<T: IoBuf>(&self, buf: T, pos: usize) -> BufResult<usize, T> {
- self.file.write_at(buf, pos as u64).await
- }
-
- pub async fn write_all_at<T: IoBuf>(&self, buf: T, pos: u64) -> BufResult<(), T> {
- self.file.write_all_at(buf, pos).await
- }
-
- pub async fn read_exact_at<T: IoBufMut>(&self, buf: T, pos: u64) -> BufResult<(), T> {
- self.file.read_exact_at(buf, pos).await
- }
-
- pub async fn read_at<T: IoBufMut>(&self, buf: T, pos: usize) -> BufResult<usize, T> {
- self.file.read_at(buf, pos as u64).await
- }
-
- pub async fn sync_all(&self) -> std::io::Result<()> {
- self.file.sync_all().await
- }
-}
-
-impl AsyncReadRent for IggyFile {
- /// Reads `exactly` `buf.len()` bytes into the buffer.
- async fn read<T: IoBufMut>(&mut self, buf: T) -> BufResult<usize, T> {
- let (res, buf) = self.file.read_at(buf, self.position).await;
- let n = match res {
- Ok(n) => n,
- Err(e) => return (Err(e), buf),
- };
- self.position += n as u64;
- (Ok(n), buf)
- }
-
- //TODO(numinex) - maybe implement this ?
- fn readv<T: IoVecBufMut>(&mut self, buf: T) -> impl Future<Output = BufResult<usize, T>> {
- async move { (Ok(0), buf) }
- }
-}
-
-impl AsyncWriteRent for IggyFile {
- /// Writes entire buffer to the file at the current position.
- async fn write<B: IoBuf>(&mut self, buf: B) -> BufResult<usize, B> {
- let (res, buf) = self.file.write_at(buf, self.position).await;
- let n = match res {
- Ok(n) => n,
- Err(e) => return (Err(e), buf),
- };
- self.position += n as u64;
- (Ok(n), buf)
- }
-
- // This is bait!!!!
- async fn writev<T: IoVecBuf>(&mut self, buf_vec: T) -> BufResult<usize, T> {
- unimplemented!("writev is not implemented for IggyFile");
- }
-
- async fn flush(&mut self) -> std::io::Result<()> {
- self.file.sync_all().await
- }
-
- //TODO(numinex) - How to implement this ?
- async fn shutdown(&mut self) -> std::io::Result<()> {
- panic!("shutdown is not supported for IggyFile");
- /*
- self.file.sync_all().await;
- unsafe {
- let file = *self.file;
- file.close().await
- }
- */
- }
-}
diff --git a/core/server/src/io/fs_utils.rs b/core/server/src/io/fs_utils.rs
index d9be21e7..6bbc1b7f 100644
--- a/core/server/src/io/fs_utils.rs
+++ b/core/server/src/io/fs_utils.rs
@@ -16,7 +16,7 @@
* under the License.
*/
-use monoio::fs;
+use compio::fs;
use std::io;
use std::path::{Path, PathBuf};
diff --git a/core/server/src/io/mod.rs b/core/server/src/io/mod.rs
index 686717cd..c3140c9e 100644
--- a/core/server/src/io/mod.rs
+++ b/core/server/src/io/mod.rs
@@ -1,4 +1 @@
-pub mod file;
pub mod fs_utils;
-pub mod reader;
-pub mod writer;
diff --git a/core/server/src/io/reader.rs b/core/server/src/io/reader.rs
deleted file mode 100644
index 63443214..00000000
--- a/core/server/src/io/reader.rs
+++ /dev/null
@@ -1,44 +0,0 @@
-use std::io::{Seek, SeekFrom};
-
-use monoio::{
- BufResult,
- buf::{IoBufMut, IoVecBufMut},
- io::{AsyncReadRent, BufReader},
-};
-
-pub struct IggyReader<R: AsyncReadRent + Seek> {
- inner: BufReader<R>,
-}
-
-impl<R: AsyncReadRent + Seek> IggyReader<R> {
- pub fn new(reader: R) -> Self {
- Self {
- inner: BufReader::new(reader),
- }
- }
-
- pub fn with_capacity(capacity: usize, reader: R) -> Self {
- Self {
- inner: BufReader::with_capacity(capacity, reader),
- }
- }
-}
-
-impl<R: AsyncReadRent + Seek> Seek for IggyReader<R> {
- fn seek(&mut self, pos: SeekFrom) -> std::io::Result<u64> {
- self.inner.get_mut().seek(pos)
- }
-}
-
-impl<R> AsyncReadRent for IggyReader<R>
-where
- R: AsyncReadRent + Seek,
-{
- fn read<T: IoBufMut>(&mut self, buf: T) -> impl Future<Output = BufResult<usize, T>> {
- self.inner.read(buf)
- }
-
- fn readv<T: IoVecBufMut>(&mut self, buf: T) -> impl Future<Output = BufResult<usize, T>> {
- self.inner.readv(buf)
- }
-}
diff --git a/core/server/src/io/writer.rs b/core/server/src/io/writer.rs
deleted file mode 100644
index e249e405..00000000
--- a/core/server/src/io/writer.rs
+++ /dev/null
@@ -1,57 +0,0 @@
-use std::io::{Seek, SeekFrom};
-
-use monoio::{
- BufResult,
- io::{AsyncWriteRent, BufWriter},
-};
-
-pub struct IggyWriter<W: AsyncWriteRent + Seek> {
- inner: BufWriter<W>,
-}
-
-impl<W: AsyncWriteRent + Seek> IggyWriter<W> {
- pub fn new(writer: W) -> Self {
- Self {
- inner: BufWriter::new(writer),
- }
- }
-
- pub fn with_capacity(capacity: usize, writer: W) -> Self {
- Self {
- inner: BufWriter::with_capacity(capacity, writer),
- }
- }
-}
-
-impl<W: AsyncWriteRent + Seek> Seek for IggyWriter<W> {
- fn seek(&mut self, pos: SeekFrom) -> std::io::Result<u64> {
- self.inner.get_mut().seek(pos)
- }
-}
-
-impl<W> AsyncWriteRent for IggyWriter<W>
-where
- W: AsyncWriteRent + Seek,
-{
- fn write<T: monoio::buf::IoBuf>(
- &mut self,
- buf: T,
- ) -> impl Future<Output = BufResult<usize, T>> {
- self.inner.write(buf)
- }
-
- fn writev<T: monoio::buf::IoVecBuf>(
- &mut self,
- buf_vec: T,
- ) -> impl Future<Output = BufResult<usize, T>> {
- self.inner.writev(buf_vec)
- }
-
- fn flush(&mut self) -> impl Future<Output = std::io::Result<()>> {
- self.inner.flush()
- }
-
- fn shutdown(&mut self) -> impl Future<Output = std::io::Result<()>> {
- self.inner.shutdown()
- }
-}
diff --git a/core/server/src/log/runtime.rs b/core/server/src/log/runtime.rs
new file mode 100644
index 00000000..36bbdd80
--- /dev/null
+++ b/core/server/src/log/runtime.rs
@@ -0,0 +1,48 @@
+use std::{pin::Pin, time::Duration};
+
+use futures::{FutureExt, SinkExt, Stream};
+use opentelemetry_sdk::runtime::{Runtime, RuntimeChannel, TrySend};
+
+#[derive(Clone)]
+pub struct MonoioRuntime;
+
+impl Runtime for MonoioRuntime {
+ fn spawn<F>(&self, future: F)
+ where
+ F: Future<Output = ()> + Send + 'static,
+ {
+ // TODO: This wont' work, we init Opentelemetry in the main thread, when there is no instance of monoio runtime
+ // running yet....
+ monoio::spawn(future);
+ }
+
+ fn delay(&self, duration: Duration) -> impl Future<Output = ()> + Send + 'static {
+ let sleep = Sleep::new(duration);
+ sleep
+ }
+}
+
+
+pub struct Sleep {
+ pub inner: Pin<Box<monoio::time::Sleep>>,
+}
+
+impl Sleep {
+ pub fn new(duration: Duration) -> Self {
+ Self {
+ inner: Box::pin(monoio::time::sleep(duration)),
+ }
+ }
+}
+
+impl Future for Sleep {
+ type Output = ();
+
+ fn poll(mut self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> std::task::Poll<Self::Output> {
+ self.inner.as_mut().poll_unpin(cx)
+ }
+}
+
+// Safety: There is no way for `Sleep` future to be flipped like a burger to another thread,
+// because we create instance of OpenTelemetry SDK runtime in the main thread, and monoio futures don't require Send & Sync bounds.
+unsafe impl Send for Sleep {}
\ No newline at end of file
diff --git a/core/server/src/main.rs b/core/server/src/main.rs
index 53fa2d7b..53151abf 100644
--- a/core/server/src/main.rs
+++ b/core/server/src/main.rs
@@ -30,7 +30,7 @@ use iggy_common::defaults::DEFAULT_ROOT_USER_ID;
use iggy_common::{Aes256GcmEncryptor, EncryptorKind, IggyError};
use server::args::Args;
use server::bootstrap::{
- create_default_executor, create_directories, create_root_user, create_shard_connections,
+ create_directories, create_root_user, create_shard_connections,
create_shard_executor, load_config, resolve_persister,
};
use server::configs::config_provider::{self};
@@ -84,7 +84,8 @@ fn main() -> Result<(), ServerError> {
let config_provider = config_provider::resolve(&args.config_provider)?;
// Load config and create directories.
// Remove `local_data` directory if run with `--fresh` flag.
- let mut rt = create_default_executor::<monoio::IoUringDriver>();
+ // TODO: Replace this once we use `monoio::main` macro.
+ let rt = create_shard_executor();
let config = rt
.block_on(async {
let config = load_config(&config_provider)
@@ -94,7 +95,7 @@ fn main() -> Result<(), ServerError> {
})?;
if args.fresh {
let system_path = config.system.get_system_path();
- if monoio::fs::metadata(&system_path).await.is_ok() {
+ if compio::fs::metadata(&system_path).await.is_ok() {
println!(
"Removing system path at: {} because `--fresh` flag was set",
system_path
@@ -122,6 +123,7 @@ fn main() -> Result<(), ServerError> {
// TODO: Make this configurable from config as a range
// for example this instance of Iggy will use cores from 0..4
+ let core_ids = core_affinity::get_core_ids().unwrap();
let available_cpus = available_parallelism().expect("Failed to get num of cores");
let shards_count = available_cpus.into();
let shards_set = 0..shards_count;
@@ -136,14 +138,24 @@ fn main() -> Result<(), ServerError> {
let config = config.clone();
let state_persister = resolve_persister(config.system.state.enforce_fsync);
+ let core_ids = core_ids.clone();
let handle = std::thread::Builder::new()
.name(format!("shard-{id}"))
.spawn(move || {
+ let core_ids = core_ids.clone();
MemoryPool::init_pool(config.system.clone());
- monoio::utils::bind_to_cpu_set(Some(shard_id))
+ if core_ids.len() > shard_id {
+ core_affinity::set_for_current(core_ids[shard_id]);
+ }
+
+ // TODO: Fix this once this PR gets resolved.
+ // https://github.com/compio-rs/compio/pull/444
+ /*
+ compio::utils::bind_to_cpu_set(Some(shard_id))
.unwrap_or_else(|e| panic!("Failed to set CPU affinity for shard-{id}: {e}"));
+ */
- let mut rt = create_shard_executor();
+ let rt = create_shard_executor();
rt.block_on(async move {
let version = SemanticVersion::current().expect("Invalid version");
info!(
@@ -253,7 +265,8 @@ fn main() -> Result<(), ServerError> {
}
let shutdown_handles_for_signal = shutdown_handles.clone();
- ctrlc::set_handler(move || {
+ /*
+ ::set_handler(move || {
info!("Received shutdown signal (SIGTERM/SIGINT), initiating graceful shutdown...");
for (shard_id, stop_sender) in &shutdown_handles_for_signal {
@@ -267,6 +280,7 @@ fn main() -> Result<(), ServerError> {
}
})
.expect("Error setting Ctrl-C handler");
+*/
info!("Iggy server is running. Press Ctrl+C or send SIGTERM to shutdown.");
for (idx, handle) in handles.into_iter().enumerate() {
diff --git a/core/server/src/quic/quic_sender.rs b/core/server/src/quic/quic_sender.rs
index c6403349..cc10c658 100644
--- a/core/server/src/quic/quic_sender.rs
+++ b/core/server/src/quic/quic_sender.rs
@@ -17,11 +17,12 @@
*/
use crate::quic::COMPONENT;
+use crate::streaming::utils::PooledBuffer;
use crate::{binary::sender::Sender, server_error::ServerError};
use bytes::BytesMut;
use error_set::ErrContext;
use iggy_common::IggyError;
-use monoio::buf::IoBufMut;
+use compio::buf::{IoBuf, IoBufMut};
use nix::libc;
use quinn::{RecvStream, SendStream};
use std::io::IoSlice;
@@ -71,7 +72,7 @@ impl Sender for QuicSender {
async fn send_ok_response_vectored(
&mut self,
length: &[u8],
- slices: Vec<libc::iovec>,
+ slices: Vec<PooledBuffer>,
) -> Result<(), IggyError> {
debug!("Sending vectored response with status: {:?}...", STATUS_OK);
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index 8ffc6560..29a1d51b 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -193,7 +193,8 @@ impl IggyShard {
let stop_receiver = self.get_stop_receiver();
let shard_for_shutdown = self.clone();
- monoio::spawn(async move {
+ /*
+ compio::runtime::spawn(async move {
let _ = stop_receiver.recv().await;
info!("Shard {} received shutdown signal", shard_for_shutdown.id);
@@ -207,6 +208,7 @@ impl IggyShard {
);
}
});
+ */
let result = try_join_all(tasks).await;
result?;
@@ -506,7 +508,7 @@ impl IggyShard {
async fn handle_event(&self, event: Arc<ShardEvent>) -> Result<(), IggyError> {
match &*event {
ShardEvent::CreatedStream { stream_id, name } => {
- self.create_stream_bypass_auth(*stream_id, name).await
+ self.create_stream_bypass_auth(*stream_id, name)
}
ShardEvent::CreatedTopic {
stream_id,
diff --git a/core/server/src/shard/system/snapshot/mod.rs b/core/server/src/shard/system/snapshot/mod.rs
index eb726509..71fd771e 100644
--- a/core/server/src/shard/system/snapshot/mod.rs
+++ b/core/server/src/shard/system/snapshot/mod.rs
@@ -23,14 +23,14 @@ use crate::shard::IggyShard;
use crate::streaming::session::Session;
use async_zip::tokio::write::ZipFileWriter;
use async_zip::{Compression, ZipEntryBuilder};
+use compio::io::{AsyncReadAtExt, AsyncWriteAtExt};
use iggy_common::{IggyDuration, IggyError, Snapshot, SnapshotCompression, SystemSnapshotType};
-use monoio::fs::{File, OpenOptions};
+use compio::fs::{File, OpenOptions};
use std::io::Cursor;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Instant;
use tempfile::NamedTempFile;
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::process::Command;
use tokio_util::compat::TokioAsyncWriteCompatExt;
use tracing::{error, info};
@@ -80,13 +80,17 @@ impl IggyShard {
let filename = format!("{snapshot_type}.txt");
let entry = ZipEntryBuilder::new(filename.clone().into(), compression);
- let file = File::open(temp_file.path()).await.map_err(|e| {
- error!("Failed to open temporary file: {}", e);
- IggyError::SnapshotFileCompletionFailed
- })?;
+ let file = OpenOptions::new()
+ .read(true)
+ .open(temp_file.path())
+ .await
+ .map_err(|e| {
+ error!("Failed to open temporary file: {}", e);
+ IggyError::SnapshotFileCompletionFailed
+ })?;
let content = Vec::new();
- let (result, content) = file.read_exact_at(content, 0).await;
+ let (result, content) = file.read_exact_at(content, 0).await.into();
if let Err(e) = result {
error!("Failed to read temporary file: {}", e);
continue;
@@ -137,8 +141,11 @@ async fn write_command_output_to_temp_file(
) -> Result<NamedTempFile, std::io::Error> {
let output = command.output().await?;
let temp_file = NamedTempFile::new()?;
- let file = File::from_std(temp_file.as_file().try_clone()?).unwrap();
- let (result, _) = file.write_all_at(output.stdout, 0).await;
+ let mut file = OpenOptions::new()
+ .write(true)
+ .open(temp_file.path())
+ .await?;
+ let (result, _) = file.write_all_at(output.stdout, 0).await.into();
result?;
file.sync_all().await?;
Ok(temp_file)
@@ -150,33 +157,37 @@ async fn get_filesystem_overview() -> Result<NamedTempFile, std::io::Error> {
async fn get_process_info() -> Result<NamedTempFile, std::io::Error> {
let temp_file = NamedTempFile::new()?;
- let file = File::from_std(temp_file.as_file().try_clone()?).unwrap();
+ let mut file = OpenOptions::new()
+ .create(true)
+ .write(true)
+ .open(temp_file.path())
+ .await?;
let mut position = 0;
let ps_output = Command::new("ps").arg("aux").output().await?;
let (result, written) = file
.write_all_at(b"=== Process List (ps aux) ===\n", 0)
- .await;
+ .await.into();
result?;
position += written.len() as u64;
- let (result, written) = file.write_all_at(ps_output.stdout, position).await;
+ let (result, written) = file.write_all_at(ps_output.stdout, position).await.into();
result?;
position += written.len() as u64;
- let (result, written) = file.write_all_at(b"\n\n", position).await;
+ let (result, written) = file.write_all_at(b"\n\n", position).await.into();
result?;
position += written.len() as u64;
let (result, written) = file
.write_all_at(b"=== Detailed Process Information ===\n", position)
- .await;
+ .await.into();
result?;
position += written.len() as u64;
let proc_info = procdump::get_proc_info().await?;
let bytes = proc_info.as_bytes().to_owned();
- let (result, _) = file.write_all_at(bytes, position).await;
+ let (result, _) = file.write_all_at(bytes, position).await.into();
result?;
file.sync_all().await?;
diff --git a/core/server/src/shard/system/storage.rs b/core/server/src/shard/system/storage.rs
index 3d895d4c..75104cc2 100644
--- a/core/server/src/shard/system/storage.rs
+++ b/core/server/src/shard/system/storage.rs
@@ -17,18 +17,17 @@
*/
use super::COMPONENT;
-use crate::io::file::IggyFile;
use crate::shard::system::info::SystemInfo;
use crate::streaming::persistence::persister::PersisterKind;
use crate::streaming::storage::SystemInfoStorage;
use crate::streaming::utils::PooledBuffer;
use crate::streaming::utils::file;
use anyhow::Context;
+use compio::io::AsyncReadAtExt;
use error_set::ErrContext;
use iggy_common::IggyError;
-use monoio::io::AsyncReadRentExt;
use std::sync::Arc;
-use tokio::io::AsyncReadExt;
+use compio::io::AsyncReadExt;
use tracing::info;
#[derive(Debug)]
@@ -63,10 +62,10 @@ impl SystemInfoStorage for FileSystemInfoStorage {
.map_err(|_| IggyError::CannotReadFileMetadata)?
.len() as usize;
- let mut file = IggyFile::new(file);
+ let file = file::open(&self.path).await.map_err(|_| IggyError::CannotReadFile)?;
let mut buffer = PooledBuffer::with_capacity(file_size);
buffer.put_bytes(0, file_size);
- let (result, buffer) = file.read_exact(buffer).await;
+ let (result, buffer) = file.read_exact_at(buffer, 0).await.into();
result
.with_error_context(|error| {
format!(
diff --git a/core/server/src/shard/system/streams.rs b/core/server/src/shard/system/streams.rs
index b4921066..bc8fc2bb 100644
--- a/core/server/src/shard/system/streams.rs
+++ b/core/server/src/shard/system/streams.rs
@@ -186,7 +186,7 @@ impl IggyShard {
}))
}
- pub async fn create_stream_bypass_auth(
+ pub fn create_stream_bypass_auth(
&self,
stream_id: Option<u32>,
name: &str,
diff --git a/core/server/src/shard/system/topics.rs b/core/server/src/shard/system/topics.rs
index 0f8155fb..462f47a4 100644
--- a/core/server/src/shard/system/topics.rs
+++ b/core/server/src/shard/system/topics.rs
@@ -132,9 +132,9 @@ impl IggyShard {
.with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to get topic with ID: {topic_id} in stream with ID: {stream_id}")
})?;
- topic.persist().await.with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to persist topic: {topic}")
- })?;
+ topic.persist().await.with_error_context(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to persist topic: {topic}")
+ })?;
// TODO: Figure out a way how to distribute the shards table among different shards,
// without the need to do code from below, everytime we handle a `ShardEvent`.
@@ -204,9 +204,9 @@ impl IggyShard {
.with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to get topic with ID: {topic_id} in stream with ID: {stream_id}")
})?;
- topic.persist().await.with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to persist topic: {topic}")
- })?;
+ topic.persist().await.with_error_context(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to persist topic: {topic}")
+ })?;
let records = partition_ids.into_iter().map(|partition_id| {
let namespace = IggyNamespace::new(stream_id, topic_id, partition_id);
diff --git a/core/server/src/shard/task_registry.rs b/core/server/src/shard/task_registry.rs
index 1a201d3b..ff674a8e 100644
--- a/core/server/src/shard/task_registry.rs
+++ b/core/server/src/shard/task_registry.rs
@@ -17,8 +17,8 @@
*/
use async_channel::{Receiver, Sender, bounded};
+use compio::runtime::JoinHandle;
use futures::future::join_all;
-use monoio::task::JoinHandle;
use std::cell::RefCell;
use std::collections::HashMap;
use std::future::Future;
@@ -42,7 +42,7 @@ impl TaskRegistry {
where
F: Future<Output = ()> + 'static,
{
- let handle = monoio::spawn(future);
+ let handle = compio::runtime::spawn(future);
self.tasks.borrow_mut().push(handle);
}
@@ -85,8 +85,8 @@ impl TaskRegistry {
.into_iter()
.enumerate()
.map(|(idx, handle)| async move {
- match monoio::time::timeout(timeout, handle).await {
- Ok(()) => (idx, true),
+ match compio::time::timeout(timeout, handle).await {
+ Ok(_) => (idx, true),
Err(_) => {
warn!("Task {} did not complete within timeout", idx);
(idx, false)
diff --git a/core/server/src/shard/tasks/messages.rs b/core/server/src/shard/tasks/messages.rs
index ec4018c0..d573c058 100644
--- a/core/server/src/shard/tasks/messages.rs
+++ b/core/server/src/shard/tasks/messages.rs
@@ -14,7 +14,7 @@ async fn run_shard_messages_receiver(shard: Rc<IggyShard>) -> Result<(), IggyErr
if shard.is_shutting_down() {
return;
}
- monoio::time::sleep(Duration::from_millis(100)).await;
+ compio::time::sleep(Duration::from_millis(100)).await;
}
};
diff --git a/core/server/src/state/file.rs b/core/server/src/state/file.rs
index e8eb6ee6..85a280cb 100644
--- a/core/server/src/state/file.rs
+++ b/core/server/src/state/file.rs
@@ -16,28 +16,27 @@
* under the License.
*/
-use crate::io::file::IggyFile;
-use crate::io::reader::IggyReader;
use crate::state::command::EntryCommand;
use crate::state::{COMPONENT, State, StateEntry};
use crate::streaming::persistence::persister::PersisterKind;
use crate::streaming::utils::file;
use crate::versioning::SemanticVersion;
use bytes::{Buf, BufMut, Bytes, BytesMut};
+use compio::fs::File;
use error_set::ErrContext;
use iggy_common::BytesSerializable;
use iggy_common::EncryptorKind;
use iggy_common::IggyByteSize;
use iggy_common::IggyError;
use iggy_common::IggyTimestamp;
-use monoio::io::AsyncReadRentExt;
+use compio::io::{AsyncReadExt, AsyncWriteExt};
use std::fmt::Debug;
use std::path::Path;
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use tracing::{debug, error, info};
-pub const BUF_READER_CAPACITY_BYTES: usize = 512 * 1000;
+pub const BUF_cursor_CAPACITY_BYTES: usize = 512 * 1000;
const FILE_STATE_PARSE_ERROR: &str = "STATE - failed to parse file state";
#[derive(Debug)]
@@ -129,7 +128,6 @@ impl State for FileState {
.map_err(|_| IggyError::CannotReadFileMetadata)?
.len();
- let file = IggyFile::new(file);
if file_size == 0 {
info!("State file is empty");
return Ok(Vec::new());
@@ -141,11 +139,11 @@ impl State for FileState {
);
let mut entries = Vec::new();
let mut total_size: u64 = 0;
- let mut reader = IggyReader::with_capacity(BUF_READER_CAPACITY_BYTES, file);
+ let mut cursor = std::io::Cursor::new(file);
let mut current_index = 0;
let mut entries_count = 0;
loop {
- let index = reader
+ let index = cursor
.read_u64_le()
.await
.with_error_context(|error| format!("{FILE_STATE_PARSE_ERROR} index. {error}"))
@@ -163,32 +161,32 @@ impl State for FileState {
current_index = index;
entries_count += 1;
- let term = reader
+ let term = cursor
.read_u64_le()
.await
.with_error_context(|error| format!("{FILE_STATE_PARSE_ERROR} term. {error}"))
.map_err(|_| IggyError::InvalidNumberEncoding)?;
total_size += 8;
- let leader_id = reader
+ let leader_id = cursor
.read_u32_le()
.await
.with_error_context(|error| format!("{FILE_STATE_PARSE_ERROR} leader_id. {error}"))
.map_err(|_| IggyError::InvalidNumberEncoding)?;
total_size += 4;
- let version = reader
+ let version = cursor
.read_u32_le()
.await
.with_error_context(|error| format!("{FILE_STATE_PARSE_ERROR} version. {error}"))
.map_err(|_| IggyError::InvalidNumberEncoding)?;
total_size += 4;
- let flags = reader
+ let flags = cursor
.read_u64_le()
.await
.with_error_context(|error| format!("{FILE_STATE_PARSE_ERROR} flags. {error}"))
.map_err(|_| IggyError::InvalidNumberEncoding)?;
total_size += 8;
let timestamp = IggyTimestamp::from(
- reader
+ cursor
.read_u64_le()
.await
.with_error_context(|error| {
@@ -197,19 +195,19 @@ impl State for FileState {
.map_err(|_| IggyError::InvalidNumberEncoding)?,
);
total_size += 8;
- let user_id = reader
+ let user_id = cursor
.read_u32_le()
.await
.with_error_context(|error| format!("{FILE_STATE_PARSE_ERROR} user_id. {error}"))
.map_err(|_| IggyError::InvalidNumberEncoding)?;
total_size += 4;
- let checksum = reader
+ let checksum = cursor
.read_u32_le()
.await
.with_error_context(|error| format!("{FILE_STATE_PARSE_ERROR} checksum. {error}"))
.map_err(|_| IggyError::InvalidNumberEncoding)?;
total_size += 4;
- let context_length = reader
+ let context_length = cursor
.read_u32_le()
.await
.with_error_context(|error| {
@@ -220,20 +218,20 @@ impl State for FileState {
total_size += 4;
let mut context = BytesMut::with_capacity(context_length);
context.put_bytes(0, context_length);
- let (result, context) = reader.read_exact(context).await;
+ let (result, context) = cursor.read_exact(context).await.into();
result
.with_error_context(|error| format!("{FILE_STATE_PARSE_ERROR} code. {error}"))
.map_err(|_| IggyError::CannotReadFile)?;
let context = context.freeze();
total_size += context_length as u64;
- let code = reader
+ let code = cursor
.read_u32_le()
.await
.with_error_context(|error| format!("{FILE_STATE_PARSE_ERROR} code. {error}"))
.map_err(|_| IggyError::InvalidNumberEncoding)?;
total_size += 4;
- let mut command_length = reader
+ let mut command_length = cursor
.read_u32_le()
.await
.with_error_context(|error| {
@@ -244,7 +242,7 @@ impl State for FileState {
total_size += 4;
let mut command = BytesMut::with_capacity(command_length);
command.put_bytes(0, command_length);
- let (result, command) = reader.read_exact(command).await;
+ let (result, command) = cursor.read_exact(command).await.into();
result
.with_error_context(|error| format!("{FILE_STATE_PARSE_ERROR} command. {error}"))
.map_err(|_| IggyError::CannotReadFile)?;
diff --git a/core/server/src/streaming/partitions/storage.rs b/core/server/src/streaming/partitions/storage.rs
index 6621ea19..69d2cd32 100644
--- a/core/server/src/streaming/partitions/storage.rs
+++ b/core/server/src/streaming/partitions/storage.rs
@@ -18,7 +18,6 @@
use crate::compat::index_rebuilding::index_rebuilder::IndexRebuilder;
use crate::configs::cache_indexes::CacheIndexesConfig;
-use crate::io::file::IggyFile;
use crate::io::fs_utils;
use crate::state::system::PartitionState;
use crate::streaming::partitions::COMPONENT;
@@ -27,12 +26,12 @@ use crate::streaming::persistence::persister::PersisterKind;
use crate::streaming::segments::*;
use crate::streaming::storage::PartitionStorage;
use crate::streaming::utils::file;
+use compio::fs;
+use compio::fs::create_dir_all;
+use compio::io::AsyncReadExt;
use error_set::ErrContext;
use iggy_common::ConsumerKind;
use iggy_common::IggyError;
-use monoio::fs;
-use monoio::fs::create_dir_all;
-use monoio::io::AsyncReadRentExt;
use std::path::Path;
use std::sync::Arc;
use std::sync::atomic::Ordering;
@@ -117,8 +116,18 @@ impl PartitionStorage for FilePartitionStorage {
let messages_file_path = segment.messages_file_path().to_owned();
let time_index_path = index_path.replace(INDEX_EXTENSION, "timeindex");
- let index_path_exists = tokio::fs::try_exists(&index_path).await.unwrap();
- let time_index_path_exists = tokio::fs::try_exists(&time_index_path).await.unwrap();
+ // TODO: Move to fs_utils
+ async fn try_exists(index_path: &str) -> Result<bool, std::io::Error> {
+ match compio::fs::metadata(index_path).await {
+ Ok(_) => Ok(true),
+ Err(err) => match err.kind() {
+ std::io::ErrorKind::NotFound => Ok(false),
+ _ => Err(err),
+ },
+ }
+ }
+ let index_path_exists = try_exists(&index_path).await.unwrap();
+ let time_index_path_exists = try_exists(&time_index_path).await.unwrap();
let index_cache_enabled = matches!(
partition.config.segment.cache_indexes,
CacheIndexesConfig::All | CacheIndexesConfig::OpenSegment
@@ -416,6 +425,7 @@ impl PartitionStorage for FilePartitionStorage {
let mut consumer_offsets = Vec::new();
let mut dir_entries = dir_entries.unwrap();
+ // TODO: reimplement using the compio dir walk
while let Some(dir_entry) = dir_entries.next() {
let dir_entry = dir_entry.unwrap();
let metadata = dir_entry.metadata();
@@ -451,8 +461,9 @@ impl PartitionStorage for FilePartitionStorage {
)
})
.map_err(|_| IggyError::CannotReadFile)?;
- let mut file = IggyFile::new(file);
- let offset = file
+ // TODO: This is like awfull.
+ let mut cursor = std::io::Cursor::new(file);
+ let offset = cursor
.read_u64_le()
.await
.with_error_context(|error| {
diff --git a/core/server/src/streaming/persistence/persister.rs b/core/server/src/streaming/persistence/persister.rs
index 6ba5c87b..ca522b85 100644
--- a/core/server/src/streaming/persistence/persister.rs
+++ b/core/server/src/streaming/persistence/persister.rs
@@ -18,12 +18,13 @@
use crate::streaming::persistence::COMPONENT;
use crate::streaming::utils::file;
+use compio::buf::IoBuf;
+use compio::fs::remove_file;
+use compio::io::AsyncWriteAtExt;
use error_set::ErrContext;
use iggy_common::IggyError;
-use monoio::buf::IoBuf;
use std::fmt::Debug;
use std::future::Future;
-use tokio::fs;
#[cfg(test)]
use mockall::automock;
@@ -85,13 +86,13 @@ pub struct FileWithSyncPersister;
impl Persister for FilePersister {
async fn append<B: IoBuf>(&self, path: &str, bytes: B) -> Result<(), IggyError> {
- let file = file::append(path)
+ let (mut file, position) = file::append(path)
.await
.with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to append to file: {path}")
})
.map_err(|_| IggyError::CannotAppendToFile)?;
- file.write_all_at(bytes, 0)
+ file.write_all_at(bytes, position)
.await
.0
.with_error_context(|error| {
@@ -102,7 +103,7 @@ impl Persister for FilePersister {
}
async fn overwrite<B: IoBuf>(&self, path: &str, bytes: B) -> Result<(), IggyError> {
- let file = file::overwrite(path)
+ let mut file = file::overwrite(path)
.await
.with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to overwrite file: {path}")
@@ -120,7 +121,7 @@ impl Persister for FilePersister {
}
async fn delete(&self, path: &str) -> Result<(), IggyError> {
- fs::remove_file(path)
+ remove_file(path)
.await
.with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to delete file: {path}")
@@ -132,17 +133,12 @@ impl Persister for FilePersister {
impl Persister for FileWithSyncPersister {
async fn append<B: IoBuf>(&self, path: &str, bytes: B) -> Result<(), IggyError> {
- let file = file::append(path)
+ let (mut file, position) = file::append(path)
.await
.with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to append to file: {path}")
})
.map_err(|_| IggyError::CannotAppendToFile)?;
- let position = file
- .metadata()
- .await
- .map_err(|_| IggyError::CannotReadFileMetadata)?
- .len();
file.write_all_at(bytes, position)
.await
.0
@@ -162,7 +158,7 @@ impl Persister for FileWithSyncPersister {
}
async fn overwrite<B: IoBuf>(&self, path: &str, bytes: B) -> Result<(), IggyError> {
- let file = file::overwrite(path)
+ let mut file = file::overwrite(path)
.await
.with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to overwrite file: {path}")
@@ -188,7 +184,7 @@ impl Persister for FileWithSyncPersister {
}
async fn delete(&self, path: &str) -> Result<(), IggyError> {
- fs::remove_file(path)
+ remove_file(path)
.await
.with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to delete file: {path}")
diff --git a/core/server/src/streaming/persistence/task.rs b/core/server/src/streaming/persistence/task.rs
index d1c716aa..1b359ef7 100644
--- a/core/server/src/streaming/persistence/task.rs
+++ b/core/server/src/streaming/persistence/task.rs
@@ -18,18 +18,18 @@
use crate::streaming::persistence::COMPONENT;
use bytes::Bytes;
+use compio::runtime::Task;
use error_set::ErrContext;
use flume::{Receiver, Sender, unbounded};
use iggy_common::IggyError;
-use monoio::task;
-use std::{sync::Arc, time::Duration};
+use std::{any::Any, sync::Arc, time::Duration};
use tracing::error;
use super::persister::PersisterKind;
pub struct LogPersisterTask {
_sender: Option<Sender<Bytes>>,
- _task_handle: Option<task::JoinHandle<()>>,
+ _task_handle: Option<Task<Result<(), Box<dyn Any + Send>>>>,
}
impl std::fmt::Debug for LogPersisterTask {
@@ -50,7 +50,7 @@ impl LogPersisterTask {
) -> Self {
let (sender, receiver): (Sender<Bytes>, Receiver<Bytes>) = unbounded();
- let task_handle = monoio::spawn(async move {
+ let task_handle = compio::runtime::spawn(async move {
loop {
match receiver.recv_async().await {
Ok(data) => {
@@ -129,7 +129,7 @@ impl Drop for LogPersisterTask {
self._sender.take();
if let Some(handle) = self._task_handle.take() {
- monoio::spawn(async move { handle.await });
+ compio::runtime::spawn(async move { handle.await });
}
}
}
diff --git a/core/server/src/streaming/segments/indexes/index_reader.rs b/core/server/src/streaming/segments/indexes/index_reader.rs
index 8b09666a..bdb64279 100644
--- a/core/server/src/streaming/segments/indexes/index_reader.rs
+++ b/core/server/src/streaming/segments/indexes/index_reader.rs
@@ -17,13 +17,16 @@
*/
use super::IggyIndexesMut;
-use crate::{io::file::IggyFile, streaming::utils::PooledBuffer};
+use crate::{streaming::utils::PooledBuffer};
use bytes::BytesMut;
+use compio::{
+ BufResult,
+ fs::{File, OpenOptions},
+ io::AsyncReadAtExt,
+};
use error_set::ErrContext;
use iggy_common::{INDEX_SIZE, IggyError, IggyIndex, IggyIndexView};
-use monoio::fs::OpenOptions;
use std::{
- fs::File as StdFile,
io::ErrorKind,
os::unix::fs::FileExt,
sync::{
@@ -37,7 +40,7 @@ use tracing::{error, trace};
#[derive(Debug)]
pub struct IndexReader {
file_path: String,
- file: IggyFile,
+ file: File,
index_size_bytes: Arc<AtomicU64>,
}
@@ -51,7 +54,6 @@ impl IndexReader {
.with_error_context(|error| format!("Failed to open index file: {file_path}. {error}"))
.map_err(|_| IggyError::CannotReadFile)?;
- let file = IggyFile::new(file);
trace!(
"Opened index file for reading: {file_path}, size: {}",
index_size_bytes.load(Ordering::Acquire)
@@ -337,13 +339,13 @@ impl IndexReader {
if use_pool {
let mut buf = PooledBuffer::with_capacity(len as usize);
unsafe { buf.set_len(len as usize) };
- let (result, buf) = self.file.read_exact_at(buf, offset as u64).await;
+ let (result, buf) = self.file.read_exact_at(buf, offset as u64).await.into();
result?;
Ok(buf)
} else {
let mut buf = BytesMut::with_capacity(len as usize);
unsafe { buf.set_len(len as usize) };
- let (result, buf) = self.file.read_exact_at(buf, offset as u64).await;
+ let (result, buf) = self.file.read_exact_at(buf, offset as u64).await.into();
result?;
Ok(PooledBuffer::from_existing(buf))
}
diff --git a/core/server/src/streaming/segments/indexes/index_writer.rs b/core/server/src/streaming/segments/indexes/index_writer.rs
index 4b79f9d2..416cde21 100644
--- a/core/server/src/streaming/segments/indexes/index_writer.rs
+++ b/core/server/src/streaming/segments/indexes/index_writer.rs
@@ -16,25 +16,25 @@
* under the License.
*/
+use compio::fs::File;
+use compio::fs::OpenOptions;
+use compio::io::AsyncWriteAtExt;
use error_set::ErrContext;
use iggy_common::INDEX_SIZE;
use iggy_common::IggyError;
-use monoio::fs::OpenOptions;
-use monoio::io::AsyncWriteRentExt;
use std::sync::{
Arc,
atomic::{AtomicU64, Ordering},
};
use tracing::trace;
-use crate::io::file::IggyFile;
use crate::streaming::utils::PooledBuffer;
/// A dedicated struct for writing to the index file.
#[derive(Debug)]
pub struct IndexWriter {
file_path: String,
- file: IggyFile,
+ file: File,
index_size_bytes: Arc<AtomicU64>,
fsync: bool,
}
@@ -48,15 +48,13 @@ impl IndexWriter {
file_exists: bool,
) -> Result<Self, IggyError> {
let file = OpenOptions::new()
- .write(true)
- .append(true)
.create(true)
+ .write(true)
.open(file_path)
.await
.with_error_context(|error| format!("Failed to open index file: {file_path}. {error}"))
.map_err(|_| IggyError::CannotReadFile)?;
- let file = IggyFile::from(file);
if file_exists {
let _ = file.sync_all().await.with_error_context(|error| {
format!("Failed to fsync index file after creation: {file_path}. {error}",)
@@ -96,8 +94,9 @@ impl IndexWriter {
let count = indexes.len() / INDEX_SIZE;
let len = indexes.len();
+ let position = self.index_size_bytes.load(Ordering::Relaxed);
self.file
- .write_all(indexes)
+ .write_all_at(indexes, position)
.await
.0
.with_error_context(|error| {
diff --git a/core/server/src/streaming/segments/messages/messages_reader.rs b/core/server/src/streaming/segments/messages/messages_reader.rs
index 4091e8e3..e81eb793 100644
--- a/core/server/src/streaming/segments/messages/messages_reader.rs
+++ b/core/server/src/streaming/segments/messages/messages_reader.rs
@@ -16,13 +16,13 @@
* under the License.
*/
-use crate::io::file::IggyFile;
use crate::streaming::segments::{IggyIndexesMut, IggyMessagesBatchMut};
-use crate::streaming::utils::{PooledBuffer, file};
+use crate::streaming::utils::PooledBuffer;
use bytes::BytesMut;
+use compio::fs::{File, OpenOptions};
+use compio::io::AsyncReadAtExt;
use error_set::ErrContext;
use iggy_common::IggyError;
-use monoio::fs::File;
use std::{
io::ErrorKind,
sync::{
@@ -36,7 +36,7 @@ use tracing::{error, trace};
#[derive(Debug)]
pub struct MessagesReader {
file_path: String,
- file: IggyFile,
+ file: File,
messages_size_bytes: Arc<AtomicU64>,
}
@@ -46,9 +46,12 @@ impl MessagesReader {
file_path: &str,
messages_size_bytes: Arc<AtomicU64>,
) -> Result<Self, IggyError> {
- let file = file::open_std(file_path)
+ let file = OpenOptions::new()
+ .read(true)
+ .open(file_path)
+ .await
.with_error_context(|error| {
- format!("Failed to open messages file: {file_path}, error: {error}")
+ format!("Failed to open messages file: {file_path}. {error}")
})
.map_err(|_| IggyError::CannotReadFile)?;
@@ -67,7 +70,6 @@ impl MessagesReader {
)
});
}
- let file = File::from_std(file).unwrap();
trace!(
"Opened messages file for reading: {file_path}, size: {}",
@@ -177,13 +179,13 @@ impl MessagesReader {
if use_pool {
let mut buf = PooledBuffer::with_capacity(len as usize);
unsafe { buf.set_len(len as usize) };
- let (result, buf) = self.file.read_exact_at(buf, offset as u64).await;
+ let (result, buf) = self.file.read_exact_at(buf, offset as u64).await.into();
result?;
Ok(buf)
} else {
let mut buf = BytesMut::with_capacity(len as usize);
unsafe { buf.set_len(len as usize) };
- let (result, buf) = self.file.read_exact_at(buf, offset as u64).await;
+ let (result, buf) = self.file.read_exact_at(buf, offset as u64).await.into();
result?;
Ok(PooledBuffer::from_existing(buf))
}
diff --git a/core/server/src/streaming/segments/messages/messages_writer.rs b/core/server/src/streaming/segments/messages/messages_writer.rs
index 63e424a9..97b8a253 100644
--- a/core/server/src/streaming/segments/messages/messages_writer.rs
+++ b/core/server/src/streaming/segments/messages/messages_writer.rs
@@ -17,12 +17,11 @@
*/
use crate::{
- io::file::IggyFile,
streaming::segments::{IggyMessagesBatchSet, messages::write_batch},
};
+use compio::fs::{File, OpenOptions};
use error_set::ErrContext;
use iggy_common::{IggyByteSize, IggyError};
-use monoio::fs::{File, OpenOptions};
use std::sync::{
Arc,
atomic::{AtomicU64, Ordering},
@@ -34,7 +33,7 @@ use tracing::{error, trace};
pub struct MessagesWriter {
file_path: String,
/// Holds the file for synchronous writes; when asynchronous persistence is enabled, this will be None.
- file: Option<IggyFile>,
+ file: Option<File>,
/// When set, asynchronous writes are handled by this persister task.
messages_size_bytes: Arc<AtomicU64>,
fsync: bool,
@@ -53,11 +52,13 @@ impl MessagesWriter {
file_exists: bool,
) -> Result<Self, IggyError> {
let file = OpenOptions::new()
- .write(true)
.create(true)
- .append(true)
+ .write(true)
.open(file_path)
.await
+ .with_error_context(|err| {
+ format!("Failed to open messages file: {file_path}, error: {err}")
+ })
.map_err(|_| IggyError::CannotReadFile)?;
if file_exists {
@@ -82,7 +83,7 @@ impl MessagesWriter {
messages_size_bytes.load(Ordering::Acquire)
);
- let file = Some(IggyFile::new(file));
+ let file = Some(file);
Ok(Self {
file_path: file_path.to_string(),
file,
@@ -103,16 +104,16 @@ impl MessagesWriter {
"Saving batch set of size {messages_size} bytes ({containers_count} containers, {messages_count} messages) to messages file: {}",
self.file_path
);
+ let position = self.messages_size_bytes.load(Ordering::Relaxed);
if let Some(ref mut file) = self.file {
- write_batch(file, &self.file_path, batch_set)
+ write_batch(file, position, batch_set)
.await
.with_error_context(|error| {
format!(
"Failed to write batch to messages file: {}. {error}",
self.file_path
)
- })
- .map_err(|_| IggyError::CannotWriteToFile)?;
+ })?;
} else {
error!("File handle is not available for synchronous write.");
return Err(IggyError::CannotWriteToFile);
diff --git a/core/server/src/streaming/segments/messages/mod.rs b/core/server/src/streaming/segments/messages/mod.rs
index 9799608f..cdf25abd 100644
--- a/core/server/src/streaming/segments/messages/mod.rs
+++ b/core/server/src/streaming/segments/messages/mod.rs
@@ -19,37 +19,26 @@
mod messages_reader;
mod messages_writer;
-use crate::io::file::IggyFile;
use super::IggyMessagesBatchSet;
-use error_set::ErrContext;
+use compio::{fs::File, io::AsyncWriteAtExt};
use iggy_common::IggyError;
-use monoio::io::AsyncWriteRentExt;
pub use messages_reader::MessagesReader;
pub use messages_writer::MessagesWriter;
/// Vectored write a batches of messages to file
async fn write_batch(
- file: &mut IggyFile,
- file_path: &str,
+ file: &mut File,
+ position: u64,
mut batches: IggyMessagesBatchSet,
) -> Result<usize, IggyError> {
- //let mut slices = batches.iter().map(|b| to_iovec(&b)).collect::<Vec<iovec>>();
- let mut total_written = 0;
- // TODO: Fork monoio, piece of shit runtime.
- for batch in batches.iter_mut() {
- let messages = batch.take_messages();
- let writen = file
- .write_all(messages)
- .await
- .0
- .with_error_context(|error| {
- format!("Failed to write messages to file: {file_path}, error: {error}",)
- // TODO: Better error variant.
- })
- .map_err(|_| IggyError::CannotAppendMessage)?;
- total_written += writen;
- }
+ let total_written = batches.iter().map(|b| b.size() as usize).sum();
+ let batches = batches
+ .iter_mut()
+ .map(|b| b.take_messages())
+ .collect::<Vec<_>>();
+ let (result, _) = file.write_vectored_all_at(batches, position).await.into();
+ result.map_err(|_| IggyError::CannotWriteToFile)?;
Ok(total_written)
}
diff --git a/core/server/src/streaming/segments/segment.rs b/core/server/src/streaming/segments/segment.rs
index 9887df90..c13ce07a 100644
--- a/core/server/src/streaming/segments/segment.rs
+++ b/core/server/src/streaming/segments/segment.rs
@@ -21,6 +21,7 @@ use super::messages::*;
use super::messages_accumulator::MessagesAccumulator;
use crate::configs::system::SystemConfig;
use crate::streaming::segments::*;
+use compio::fs::remove_file;
use error_set::ErrContext;
use iggy_common::INDEX_SIZE;
use iggy_common::IggyByteSize;
@@ -30,7 +31,6 @@ use iggy_common::IggyTimestamp;
use std::rc::Rc;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
-use tokio::fs::remove_file;
use tracing::{info, warn};
const SIZE_16MB: usize = 16 * 1024 * 1024;
@@ -302,7 +302,7 @@ impl Segment {
pub async fn shutdown_writing(&mut self) {
if let Some(log_writer) = self.messages_writer.take() {
//TODO: Fixme not sure whether we should spawn a task here.
- monoio::spawn(async move {
+ compio::runtime::spawn(async move {
let _ = log_writer.fsync().await;
});
} else {
@@ -314,7 +314,7 @@ impl Segment {
if let Some(index_writer) = self.index_writer.take() {
//TODO: Fixme not sure whether we should spawn a task here.
- monoio::spawn(async move {
+ compio::runtime::spawn(async move {
let _ = index_writer.fsync().await;
drop(index_writer)
});
diff --git a/core/server/src/streaming/streams/storage.rs b/core/server/src/streaming/streams/storage.rs
index d9848f03..b2b9672c 100644
--- a/core/server/src/streaming/streams/storage.rs
+++ b/core/server/src/streaming/streams/storage.rs
@@ -23,12 +23,12 @@ use crate::streaming::streams::COMPONENT;
use crate::streaming::streams::stream::Stream;
use crate::streaming::topics::topic::Topic;
use ahash::AHashSet;
+use compio::fs;
+use compio::fs::create_dir_all;
use error_set::ErrContext;
use futures::future::join_all;
use iggy_common::IggyError;
use iggy_common::IggyTimestamp;
-use monoio::fs;
-use monoio::fs::create_dir_all;
use serde::{Deserialize, Serialize};
use std::path::Path;
use std::sync::Arc;
@@ -157,7 +157,7 @@ impl StreamStorage for FileStreamStorage {
for mut topic in unloaded_topics {
let loaded_topics = loaded_topics.clone();
let topic_state = state.topics.remove(&topic.topic_id).unwrap();
- let load_topic = monoio::spawn(async move {
+ let load_topic = compio::runtime::spawn(async move {
match topic.load(topic_state).await {
Ok(_) => loaded_topics.lock().await.push(topic),
Err(error) => error!(
diff --git a/core/server/src/streaming/topics/storage.rs b/core/server/src/streaming/topics/storage.rs
index 8d92b9d1..f7a1fe38 100644
--- a/core/server/src/streaming/topics/storage.rs
+++ b/core/server/src/streaming/topics/storage.rs
@@ -25,13 +25,13 @@ use crate::streaming::topics::consumer_group::ConsumerGroup;
use crate::streaming::topics::topic::Topic;
use ahash::AHashSet;
use anyhow::Context;
+use compio::fs;
+use compio::fs::create_dir_all;
use error_set::ErrContext;
use futures::future::join_all;
use iggy_common::IggyError;
use iggy_common::locking::IggyRwLock;
use iggy_common::locking::IggySharedMutFn;
-use monoio::fs;
-use monoio::fs::create_dir_all;
use serde::{Deserialize, Serialize};
use std::path::Path;
use std::sync::Arc;
@@ -188,7 +188,7 @@ impl TopicStorage for FileTopicStorage {
for mut partition in unloaded_partitions {
let loaded_partitions = loaded_partitions.clone();
let partition_state = state.partitions.remove(&partition.partition_id).unwrap();
- let load_partition = monoio::spawn(async move {
+ let load_partition = compio::runtime::spawn(async move {
match partition.load(partition_state).await {
Ok(_) => {
loaded_partitions.lock().await.push(partition);
diff --git a/core/server/src/streaming/utils/file.rs b/core/server/src/streaming/utils/file.rs
index 7a5bb524..de00aa5e 100644
--- a/core/server/src/streaming/utils/file.rs
+++ b/core/server/src/streaming/utils/file.rs
@@ -16,7 +16,7 @@
* under the License.
*/
-use monoio::fs::{File, OpenOptions, remove_file};
+use compio::fs::{File, OpenOptions, remove_file};
use std::path::Path;
pub fn open_std(path: &str) -> Result<std::fs::File, std::io::Error> {
@@ -27,8 +27,14 @@ pub async fn open(path: &str) -> Result<File, std::io::Error> {
OpenOptions::new().read(true).open(path).await
}
-pub async fn append(path: &str) -> Result<File, std::io::Error> {
- OpenOptions::new().read(true).append(true).open(path).await
+pub async fn append(path: &str) -> Result<(File, u64), std::io::Error> {
+ let file = OpenOptions::new()
+ .create(true)
+ .write(true)
+ .open(path)
+ .await?;
+ let position = file.metadata().await?.len();
+ Ok((file, position))
}
pub async fn overwrite(path: &str) -> Result<File, std::io::Error> {
@@ -45,7 +51,7 @@ pub async fn remove(path: &str) -> Result<(), std::io::Error> {
}
pub async fn rename(old_path: &str, new_path: &str) -> Result<(), std::io::Error> {
- monoio::fs::rename(Path::new(old_path), Path::new(new_path)).await
+ compio::fs::rename(Path::new(old_path), Path::new(new_path)).await
}
pub async fn exists(path: &str) -> Result<bool, std::io::Error> {
diff --git a/core/server/src/streaming/utils/pooled_buffer.rs b/core/server/src/streaming/utils/pooled_buffer.rs
index a8664325..39ef25be 100644
--- a/core/server/src/streaming/utils/pooled_buffer.rs
+++ b/core/server/src/streaming/utils/pooled_buffer.rs
@@ -18,7 +18,7 @@
use super::memory_pool::{BytesMutExt, memory_pool};
use bytes::{Buf, BufMut, BytesMut};
-use monoio::buf::{IoBuf, IoBufMut};
+use compio::buf::{IoBuf, IoBufMut, SetBufInit};
use std::ops::{Deref, DerefMut};
#[derive(Debug)]
@@ -217,26 +217,32 @@ impl Buf for PooledBuffer {
}
}
-unsafe impl IoBufMut for PooledBuffer {
- fn write_ptr(&mut self) -> *mut u8 {
- self.inner.write_ptr()
- }
-
- fn bytes_total(&mut self) -> usize {
- self.inner.bytes_total()
+impl SetBufInit for PooledBuffer {
+ unsafe fn set_buf_init(&mut self, len: usize) {
+ if self.inner.len() <= len {
+ unsafe {
+ self.inner.set_len(len);
+ }
+ }
}
+}
- unsafe fn set_init(&mut self, pos: usize) {
- unsafe { self.inner.set_init(pos) }
+unsafe impl IoBufMut for PooledBuffer {
+ fn as_buf_mut_ptr(&mut self) -> *mut u8 {
+ self.inner.as_buf_mut_ptr()
}
}
unsafe impl IoBuf for PooledBuffer {
- fn read_ptr(&self) -> *const u8 {
- self.inner.read_ptr()
+ fn as_buf_ptr(&self) -> *const u8 {
+ self.inner.as_buf_ptr()
+ }
+
+ fn buf_len(&self) -> usize {
+ self.inner.len()
}
- fn bytes_init(&self) -> usize {
- self.inner.bytes_init()
+ fn buf_capacity(&self) -> usize {
+ self.inner.capacity()
}
}
diff --git a/core/server/src/tcp/connection_handler.rs b/core/server/src/tcp/connection_handler.rs
index 418af387..9c6803b8 100644
--- a/core/server/src/tcp/connection_handler.rs
+++ b/core/server/src/tcp/connection_handler.rs
@@ -62,6 +62,7 @@ pub(crate) async fn handle_connection(
if error.as_code() == IggyError::ConnectionClosed.as_code() {
return Err(ConnectionError::from(error));
} else {
+ error!("got error: {:?}", error);
sender.send_error_response(error).await?;
continue;
}
diff --git a/core/server/src/tcp/sender.rs b/core/server/src/tcp/sender.rs
index 2f445d4f..61a6ce2b 100644
--- a/core/server/src/tcp/sender.rs
+++ b/core/server/src/tcp/sender.rs
@@ -16,30 +16,35 @@
* under the License.
*/
-use bytes::BytesMut;
-use iggy_common::IggyError;
-use monoio::{
- buf::IoBufMut,
- io::{AsyncReadRent, AsyncReadRentExt, AsyncWriteRent, AsyncWriteRentExt},
+use bytes::{Bytes, BytesMut};
+use compio::{
+ buf::{IoBuf, IoBufMut}, io::{AsyncRead, AsyncReadAtExt, AsyncReadExt, AsyncWriteExt}, BufResult
};
+use iggy_common::IggyError;
use nix::libc;
use std::io::IoSlice;
-use tracing::debug;
+use tracing::{debug, error};
+
+use crate::streaming::utils::PooledBuffer;
const STATUS_OK: &[u8] = &[0; 4];
pub(crate) async fn read<T, B>(stream: &mut T, buffer: B) -> (Result<usize, IggyError>, B)
where
- T: AsyncReadRent + AsyncWriteRent + Unpin,
+ T: AsyncReadExt + AsyncWriteExt + Unpin,
B: IoBufMut,
{
- match stream.read_exact(buffer).await {
- (Ok(0), buffer) => (Err(IggyError::ConnectionClosed), buffer),
- (Ok(read_bytes), buffer) => (Ok(read_bytes), buffer),
+ let BufResult(result, buffer) = stream.read_exact(buffer).await;
+ match (result, buffer) {
+ (Ok(_), buffer) => (Ok(buffer.buf_len()), buffer),
+ // TODO: How to handle this ?(Ok(0), buffer) => (Err(IggyError::ConnectionClosed), buffer),
+ // `read_exact` from compio doesn't return how many bytes it read.
(Err(error), buffer) => {
if error.kind() == std::io::ErrorKind::UnexpectedEof {
+ //error!("Got some error tho.. {}", error);
(Err(IggyError::ConnectionClosed), buffer)
} else {
+ //error!("Got some other error tho.. {}", error);
(Err(IggyError::TcpError), buffer)
}
}
@@ -48,14 +53,14 @@ where
pub(crate) async fn send_empty_ok_response<T>(stream: &mut T) -> Result<(), IggyError>
where
- T: AsyncReadRent + AsyncWriteRent + Unpin,
+ T: AsyncReadExt + AsyncWriteExt + Unpin,
{
send_ok_response(stream, &[]).await
}
pub(crate) async fn send_ok_response<T>(stream: &mut T, payload: &[u8]) -> Result<(), IggyError>
where
- T: AsyncReadRent + AsyncWriteRent + Unpin,
+ T: AsyncReadExt + AsyncWriteExt + Unpin,
{
send_response(stream, STATUS_OK, payload).await
}
@@ -63,10 +68,10 @@ where
pub(crate) async fn send_ok_response_vectored<T>(
stream: &mut T,
length: &[u8],
- slices: Vec<libc::iovec>,
+ slices: Vec<PooledBuffer>,
) -> Result<(), IggyError>
where
- T: AsyncReadRentExt + AsyncWriteRentExt + Unpin,
+ T: AsyncReadExt + AsyncWriteExt + Unpin,
{
send_response_vectored(stream, STATUS_OK, length, slices).await
}
@@ -76,7 +81,7 @@ pub(crate) async fn send_error_response<T>(
error: IggyError,
) -> Result<(), IggyError>
where
- T: AsyncReadRent + AsyncWriteRent + Unpin,
+ T: AsyncReadExt + AsyncWriteExt + Unpin,
{
send_response(stream, &error.as_code().to_le_bytes(), &[]).await
}
@@ -87,7 +92,7 @@ pub(crate) async fn send_response<T>(
payload: &[u8],
) -> Result<(), IggyError>
where
- T: AsyncReadRent + AsyncWriteRent + Unpin,
+ T: AsyncReadExt + AsyncWriteExt + Unpin,
{
debug!(
"Sending response of len: {} with status: {:?}...",
@@ -108,32 +113,25 @@ pub(crate) async fn send_response_vectored<T>(
stream: &mut T,
status: &[u8],
length: &[u8],
- mut slices: Vec<libc::iovec>,
+ mut slices: Vec<PooledBuffer>,
) -> Result<(), IggyError>
where
- T: AsyncReadRentExt + AsyncWriteRentExt + Unpin,
+ T: AsyncReadExt + AsyncWriteExt + Unpin,
{
+ let resp_status = u32::from_le_bytes(status.try_into().unwrap());
debug!(
"Sending vectored response of len: {} with status: {:?}...",
slices.len(),
- status
+ resp_status
);
- let prefix = [
- libc::iovec {
- iov_base: status.as_ptr() as _,
- iov_len: status.len(),
- },
- libc::iovec {
- iov_base: length.as_ptr() as _,
- iov_len: length.len(),
- },
- ];
- slices.splice(0..0, prefix);
+ let status = PooledBuffer::from(status);
+ let length = PooledBuffer::from(length);
+ slices.splice(0..0, [status, length]);
stream
.write_vectored_all(slices)
.await
.0
.map_err(|_| IggyError::TcpError)?;
- debug!("Sent response with status: {:?}", status);
+ debug!("Sent response with status: {:?}", resp_status);
Ok(())
}
diff --git a/core/server/src/tcp/tcp_listener.rs b/core/server/src/tcp/tcp_listener.rs
index 4f528016..b541305b 100644
--- a/core/server/src/tcp/tcp_listener.rs
+++ b/core/server/src/tcp/tcp_listener.rs
@@ -29,13 +29,19 @@ use std::rc::Rc;
use std::time::Duration;
use tracing::{error, info};
-pub async fn start(server_name: &'static str, addr: SocketAddr, socket: Socket, shard: Rc<IggyShard>) -> Result<(), IggyError> {
+pub async fn start(
+ server_name: &'static str,
+ addr: SocketAddr,
+ socket: Socket,
+ shard: Rc<IggyShard>,
+) -> Result<(), IggyError> {
socket
.bind(&addr.into())
.expect("Failed to bind TCP listener");
socket.listen(1024).unwrap();
+ //TODO: Fix it later to use `TcpOpts` from compio.
let listener: std::net::TcpListener = socket.into();
- let listener = monoio::net::TcpListener::from_std(listener).unwrap();
+ let listener = compio::net::TcpListener::from_std(listener).unwrap();
info!("{server_name} server has started on: {:?}", addr);
loop {
@@ -44,7 +50,7 @@ pub async fn start(server_name: &'static str, addr: SocketAddr, socket: Socket,
if shard.is_shutting_down() {
return;
}
- monoio::time::sleep(Duration::from_millis(100)).await;
+ compio::time::sleep(Duration::from_millis(100)).await;
}
};
diff --git a/core/server/src/tcp/tcp_sender.rs b/core/server/src/tcp/tcp_sender.rs
index cfa25aad..d0a1c27c 100644
--- a/core/server/src/tcp/tcp_sender.rs
+++ b/core/server/src/tcp/tcp_sender.rs
@@ -17,14 +17,15 @@
*/
use crate::binary::sender::Sender;
+use crate::streaming::utils::PooledBuffer;
use crate::tcp::COMPONENT;
use crate::{server_error::ServerError, tcp::sender};
use bytes::BytesMut;
+use compio::buf::{IoBuf, IoBufMut};
+use compio::io::AsyncWrite;
+use compio::net::TcpStream;
use error_set::ErrContext;
use iggy_common::IggyError;
-use monoio::buf::IoBufMut;
-use monoio::io::AsyncWriteRent;
-use monoio::net::TcpStream;
use nix::libc;
#[derive(Debug)]
@@ -62,7 +63,7 @@ impl Sender for TcpSender {
async fn send_ok_response_vectored(
&mut self,
length: &[u8],
- slices: Vec<libc::iovec>,
+ slices: Vec<PooledBuffer>,
) -> Result<(), IggyError> {
sender::send_ok_response_vectored(&mut self.stream, length, slices).await
}
diff --git a/core/server/src/tcp/tcp_tls_listener.rs b/core/server/src/tcp/tcp_tls_listener.rs
index 2c81eb60..00d7b5c4 100644
--- a/core/server/src/tcp/tcp_tls_listener.rs
+++ b/core/server/src/tcp/tcp_tls_listener.rs
@@ -24,7 +24,6 @@ use crate::streaming::clients::client_manager::Transport;
use crate::tcp::connection_handler::{handle_connection, handle_error};
use futures::FutureExt;
use iggy_common::IggyError;
-use monoio_rustls::TlsAcceptor;
use rustls::ServerConfig;
use rustls::pki_types::{CertificateDer, PrivateKeyDer};
use rustls_pemfile::{certs, private_key};
@@ -42,106 +41,108 @@ pub(crate) async fn start(
socket: Socket,
shard: Rc<IggyShard>,
) -> Result<(), IggyError> {
- let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
- let config = &shard.config.tcp.tls;
-
- let (certs, key) =
- if config.self_signed && !std::path::Path::new(&config.cert_file).exists() {
- info!("Generating self-signed certificate for TCP TLS server");
- generate_self_signed_cert()
- .unwrap_or_else(|e| panic!("Failed to generate self-signed certificate: {e}"))
- } else {
- load_certificates(&config.cert_file, &config.key_file)
- .unwrap_or_else(|e| panic!("Failed to load certificates: {e}"))
- };
-
- let server_config = ServerConfig::builder()
- .with_no_client_auth()
- .with_single_cert(certs, key)
- .unwrap_or_else(|e| panic!("Unable to create TLS server config: {e}"));
-
- let acceptor = TlsAcceptor::from(Arc::new(server_config));
-
- socket
- .bind(&addr.into())
- .unwrap_or_else(|e| panic!("Unable to bind socket to address '{addr}': {e}",));
-
- let listener: std::net::TcpListener = socket.into();
- let listener = monoio::net::TcpListener::from_std(listener).unwrap();
- info!("{server_name} server has started on: {:?}", addr);
-
- loop {
- let shutdown_check = async {
- loop {
- if shard.is_shutting_down() {
- return;
- }
- monoio::time::sleep(Duration::from_millis(100)).await;
+ /*
+ let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
+ let config = &shard.config.tcp.tls;
+
+ let (certs, key) =
+ if config.self_signed && !std::path::Path::new(&config.cert_file).exists() {
+ info!("Generating self-signed certificate for TCP TLS server");
+ generate_self_signed_cert()
+ .unwrap_or_else(|e| panic!("Failed to generate self-signed certificate: {e}"))
+ } else {
+ load_certificates(&config.cert_file, &config.key_file)
+ .unwrap_or_else(|e| panic!("Failed to load certificates: {e}"))
+ };
+
+ let server_config = ServerConfig::builder()
+ .with_no_client_auth()
+ .with_single_cert(certs, key)
+ .unwrap_or_else(|e| panic!("Unable to create TLS server config: {e}"));
+
+ let acceptor = TlsAcceptor::from(Arc::new(server_config));
+
+ socket
+ .bind(&addr.into())
+ .unwrap_or_else(|e| panic!("Unable to bind socket to address '{addr}': {e}",));
+
+ let listener: std::net::TcpListener = socket.into();
+ let listener = monoio::net::TcpListener::from_std(listener).unwrap();
+ info!("{server_name} server has started on: {:?}", addr);
+
+ loop {
+ let shutdown_check = async {
+ loop {
+ if shard.is_shutting_down() {
+ return;
}
- };
+ monoio::time::sleep(Duration::from_millis(100)).await;
+ }
+ };
- let accept_future = listener.accept();
- futures::select! {
- _ = shutdown_check.fuse() => {
- info!("TCP TLS server detected shutdown flag, no longer accepting connections");
- break;
- }
- result = accept_future.fuse() => {
- match result {
- Ok((stream, address)) => {
- if shard.is_shutting_down() {
- info!("Rejecting new TLS connection from {} during shutdown", address);
- continue;
- }
- let shard_clone = shard.clone();
- info!("Accepted new TCP TLS connection: {}", address);
- let transport = Transport::Tcp;
- let session = shard_clone.add_client(&address, transport);
- //TODO: Those can be shared with other shards.
- shard_clone.add_active_session(session.clone());
- // Broadcast session to all shards.
- let event = ShardEvent::NewSession { address, transport };
- // TODO: Fixme look inside of broadcast_event_to_all_shards method.
- let _responses = shard_clone.broadcast_event_to_all_shards(event.into());
-
- let client_id = session.client_id;
- info!("Created new session: {session}");
- let acceptor = acceptor.clone();
-
- let conn_stop_receiver = shard_clone.task_registry.add_connection(client_id);
-
- let shard_for_conn = shard_clone.clone();
- shard_clone.task_registry.spawn_tracked(async move {
- match acceptor.accept(stream).await {
- Ok(tls_stream) => {
- let mut sender = SenderKind::get_tcp_tls_sender(tls_stream.into());
- if let Err(error) = handle_connection(&session, &mut sender, &shard_for_conn, conn_stop_receiver).await {
- handle_error(error);
- }
- shard_for_conn.task_registry.remove_connection(&client_id);
-
- if let Err(error) = sender.shutdown().await {
- error!(
- "Failed to shutdown TCP TLS stream for client: {client_id}, address: {address}. {error}"
- );
- } else {
- info!(
- "Successfully closed TCP TLS stream for client: {client_id}, address: {address}."
- );
- }
+ let accept_future = listener.accept();
+ futures::select! {
+ _ = shutdown_check.fuse() => {
+ info!("TCP TLS server detected shutdown flag, no longer accepting connections");
+ break;
+ }
+ result = accept_future.fuse() => {
+ match result {
+ Ok((stream, address)) => {
+ if shard.is_shutting_down() {
+ info!("Rejecting new TLS connection from {} during shutdown", address);
+ continue;
+ }
+ let shard_clone = shard.clone();
+ info!("Accepted new TCP TLS connection: {}", address);
+ let transport = Transport::Tcp;
+ let session = shard_clone.add_client(&address, transport);
+ //TODO: Those can be shared with other shards.
+ shard_clone.add_active_session(session.clone());
+ // Broadcast session to all shards.
+ let event = ShardEvent::NewSession { address, transport };
+ // TODO: Fixme look inside of broadcast_event_to_all_shards method.
+ let _responses = shard_clone.broadcast_event_to_all_shards(event.into());
+
+ let client_id = session.client_id;
+ info!("Created new session: {session}");
+ let acceptor = acceptor.clone();
+
+ let conn_stop_receiver = shard_clone.task_registry.add_connection(client_id);
+
+ let shard_for_conn = shard_clone.clone();
+ shard_clone.task_registry.spawn_tracked(async move {
+ match acceptor.accept(stream).await {
+ Ok(tls_stream) => {
+ let mut sender = SenderKind::get_tcp_tls_sender(tls_stream.into());
+ if let Err(error) = handle_connection(&session, &mut sender, &shard_for_conn, conn_stop_receiver).await {
+ handle_error(error);
}
- Err(e) => {
- error!("Failed to accept TLS connection from '{address}': {e}");
- shard_for_conn.task_registry.remove_connection(&client_id);
+ shard_for_conn.task_registry.remove_connection(&client_id);
+
+ if let Err(error) = sender.shutdown().await {
+ error!(
+ "Failed to shutdown TCP TLS stream for client: {client_id}, address: {address}. {error}"
+ );
+ } else {
+ info!(
+ "Successfully closed TCP TLS stream for client: {client_id}, address: {address}."
+ );
}
}
- });
- }
- Err(error) => error!("Unable to accept TCP TLS socket. {error}"),
+ Err(e) => {
+ error!("Failed to accept TLS connection from '{address}': {e}");
+ shard_for_conn.task_registry.remove_connection(&client_id);
+ }
+ }
+ });
}
+ Err(error) => error!("Unable to accept TCP TLS socket. {error}"),
}
}
}
+ }
+ */
Ok(())
}
diff --git a/core/server/src/tcp/tcp_tls_sender.rs b/core/server/src/tcp/tcp_tls_sender.rs
index b214e5a6..8dbb3972 100644
--- a/core/server/src/tcp/tcp_tls_sender.rs
+++ b/core/server/src/tcp/tcp_tls_sender.rs
@@ -17,41 +17,46 @@
*/
use crate::binary::sender::Sender;
+use crate::streaming::utils::PooledBuffer;
use crate::tcp::COMPONENT;
use crate::{server_error::ServerError, tcp::sender};
use bytes::BytesMut;
+use compio::buf::IoBufMut;
+use compio::io::AsyncWrite;
+use compio::net::TcpStream;
use error_set::ErrContext;
use iggy_common::IggyError;
-use monoio::buf::IoBufMut;
-use monoio::io::AsyncWriteRent;
-use monoio::net::TcpStream;
-use monoio_rustls::ServerTlsStream;
//use tokio_rustls::server::TlsStream;
use nix::libc;
#[derive(Debug)]
pub struct TcpTlsSender {
- pub(crate) stream: ServerTlsStream<TcpStream>,
+ pub(crate) stream: TcpStream,
}
impl Sender for TcpTlsSender {
async fn read<B: IoBufMut>(&mut self, buffer: B) -> (Result<usize, IggyError>, B) {
+ todo!();
sender::read(&mut self.stream, buffer).await
}
async fn send_empty_ok_response(&mut self) -> Result<(), IggyError> {
+ todo!();
sender::send_empty_ok_response(&mut self.stream).await
}
async fn send_ok_response(&mut self, payload: &[u8]) -> Result<(), IggyError> {
+ todo!();
sender::send_ok_response(&mut self.stream, payload).await
}
async fn send_error_response(&mut self, error: IggyError) -> Result<(), IggyError> {
+ todo!();
sender::send_error_response(&mut self.stream, error).await
}
async fn shutdown(&mut self) -> Result<(), ServerError> {
+ todo!();
self.stream
.shutdown()
.await
@@ -64,8 +69,9 @@ impl Sender for TcpTlsSender {
async fn send_ok_response_vectored(
&mut self,
length: &[u8],
- slices: Vec<libc::iovec>,
+ slices: Vec<PooledBuffer>,
) -> Result<(), IggyError> {
+ todo!();
sender::send_ok_response_vectored(&mut self.stream, length, slices).await
}
}