Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: support WasmEdge as an alternative engine #1

Draft
wants to merge 33 commits into
base: master
Choose a base branch
from
Draft
Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
33 commits
Select commit Hold shift + click to select a range
fb5b2fe
init work: make the public API clear
xxchan Feb 26, 2023
7a30006
implement filter transform
xxchan Feb 26, 2023
685ae36
wip
xxchan Apr 9, 2023
15686cc
make the public API clear
xxchan Feb 26, 2023
1937438
move wasmtime_engine to a different mod
xxchan Apr 15, 2023
6f1604e
fix wasi feature
xxchan Apr 15, 2023
6b88555
rename wasmtime-engine -> wasmtime
xxchan Apr 15, 2023
cd038a3
Merge branch 'master' into xxchan/clena
xxchan Apr 18, 2023
86326da
reorg
xxchan Apr 18, 2023
e959a9d
Merge branch 'xxchan/clena' into xxchan/wasmedge
xxchan Apr 18, 2023
f17b8ef
support init for wasmedge/common
xxchan Apr 18, 2023
8b11252
Merge branch 'master' into xxchan/wasmedge
xxchan Apr 18, 2023
11e1751
make SmartModuleInstance common
xxchan Apr 18, 2023
abd4fb6
make create_transform common & reorg trait imp to imp
xxchan Apr 18, 2023
66bcbdd
more movement
xxchan Apr 18, 2023
0132522
Merge branch 'master' into xxchan/wasmedge
xxchan Apr 29, 2023
da7a2e2
add other tests for wasmedge
xxchan Apr 29, 2023
26d9053
support agg for common/wasmedge
xxchan Apr 29, 2023
56de150
movement
xxchan Apr 29, 2023
14c4c3c
refactor wasmtime to use the common code
xxchan Apr 29, 2023
3d8f4bf
WasmTime -> Wasmtime
xxchan Apr 29, 2023
d0bb940
move transform unit tests to common
xxchan Apr 29, 2023
7924834
rename Wasmedge -> WasmEdge
xxchan Apr 29, 2023
f2f55e8
minor tweaks
xxchan Apr 29, 2023
e6f7e2e
change features
xxchan May 15, 2023
06d579f
rm wasmedge
xxchan May 31, 2023
6ff943b
Merge branch 'master' into xxchan/wasmedge
xxchan May 31, 2023
7e559c1
remove wasmedge
xxchan May 31, 2023
7d23e1a
fmt
xxchan May 31, 2023
f5650e3
fmt with group_imports = "StdExternalCrate"
xxchan May 31, 2023
fe98b8f
try to make diff smaller
xxchan May 31, 2023
136c572
make diff smaller
xxchan May 31, 2023
72f4ed7
clippy
xxchan May 31, 2023
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
make the public API clear
  • Loading branch information
xxchan committed Apr 15, 2023
commit 15686cc0630a25efeed7731c15f459fca821cc8a
4 changes: 2 additions & 2 deletions crates/fluvio-smartengine/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -12,11 +12,11 @@ description = "The official Fluvio SmartEngine"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

[features]
engine = ["wasmtime"]
engine = ["wasmtime-engine"]
wasi = ["wasmtime-wasi", "engine"]
transformation = ["serde_json", "serde_yaml"]
default = ["engine"]

wasmtime-engine = ["wasmtime", "wasmtime-wasi"]

[dependencies]
tracing ={ workspace = true }
Expand Down
73 changes: 73 additions & 0 deletions crates/fluvio-smartengine/src/engine/config.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,73 @@
use derive_builder::Builder;
use fluvio_smartmodule::dataplane::smartmodule::SmartModuleExtraParams;

const DEFAULT_SMARTENGINE_VERSION: i16 = 17;

/// Initial seed data to passed, this will be send back as part of the output
#[derive(Debug, Clone)]
#[non_exhaustive]
pub enum SmartModuleInitialData {
None,
Aggregate { accumulator: Vec<u8> },
}

impl SmartModuleInitialData {
pub fn with_aggregate(accumulator: Vec<u8>) -> Self {
Self::Aggregate { accumulator }
}
}

impl Default for SmartModuleInitialData {
fn default() -> Self {
Self::None
}
}

/// SmartModule configuration
#[derive(Builder)]
pub struct SmartModuleConfig {
#[builder(default, setter(strip_option))]
pub(crate) initial_data: SmartModuleInitialData,
#[builder(default)]
pub(crate) params: SmartModuleExtraParams,
// this will be deprecated in the future
#[builder(default, setter(into, strip_option))]
pub(crate) version: Option<i16>,
}

impl SmartModuleConfigBuilder {
/// add initial parameters
pub fn param(&mut self, key: impl Into<String>, value: impl Into<String>) -> &mut Self {
let mut new = self;
let mut params = new.params.take().unwrap_or_default();
params.insert(key.into(), value.into());
new.params = Some(params);
new
}
}

impl SmartModuleConfig {
pub fn builder() -> SmartModuleConfigBuilder {
SmartModuleConfigBuilder::default()
}

pub(crate) fn version(&self) -> i16 {
self.version.unwrap_or(DEFAULT_SMARTENGINE_VERSION)
}
}

#[cfg(feature = "transformation")]
impl From<crate::transformation::TransformationStep> for SmartModuleConfig {
fn from(step: crate::transformation::TransformationStep) -> Self {
Self {
initial_data: SmartModuleInitialData::None,
params: step
.with
.into_iter()
.map(|(k, v)| (k, v.into()))
.collect::<std::collections::BTreeMap<String, String>>()
.into(),
version: None,
}
}
}
Original file line number Diff line number Diff line change
@@ -1,28 +1,23 @@
use std::fmt::{self, Debug};

use anyhow::Result;
use derive_builder::Builder;
use tracing::debug;
use wasmtime::{Engine, Module};

use fluvio_smartmodule::dataplane::smartmodule::{
SmartModuleExtraParams, SmartModuleInput, SmartModuleOutput,
};
use fluvio_smartmodule::dataplane::smartmodule::{SmartModuleInput, SmartModuleOutput};

use crate::init::SmartModuleInit;
use crate::instance::{SmartModuleInstance, SmartModuleInstanceContext};
use crate::{SmartEngine, SmartModuleChainInstance};

use crate::metrics::SmartModuleChainMetrics;
use crate::state::WasmState;
use crate::transforms::create_transform;
use super::init::SmartModuleInit;
use super::instance::{SmartModuleInstance, SmartModuleInstanceContext};

const DEFAULT_SMARTENGINE_VERSION: i16 = 17;
use super::metrics::SmartModuleChainMetrics;
use super::state::WasmState;
use super::transforms::create_transform;

#[derive(Clone)]
pub struct SmartEngine(Engine);
pub struct SmartEngineImp(Engine);

#[allow(clippy::new_without_default)]
impl SmartEngine {
impl SmartEngineImp {
pub fn new() -> Self {
let mut config = wasmtime::Config::default();
config.consume_fuel(true);
Expand All @@ -34,71 +29,39 @@ impl SmartEngine {
}
}

impl Debug for SmartEngine {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "SmartModuleEngine")
}
}

/// Building SmartModule
#[derive(Default)]
pub struct SmartModuleChainBuilder {
smart_modules: Vec<(SmartModuleConfig, Vec<u8>)>,
}

impl SmartModuleChainBuilder {
/// Add SmartModule with a single transform and init
pub fn add_smart_module(&mut self, config: SmartModuleConfig, bytes: Vec<u8>) {
self.smart_modules.push((config, bytes))
/// stop adding smartmodule and return SmartModuleChain that can be executed
pub fn initialize_imp(
builder: super::SmartModuleChainBuilder,
engine: &SmartEngine,
) -> Result<SmartModuleChainInstance> {
let mut instances = Vec::with_capacity(builder.smart_modules.len());
let mut state = engine.inner.new_state();
for (config, bytes) in builder.smart_modules {
let module = Module::new(&engine.inner.0, bytes)?;
let version = config.version();
let ctx =
SmartModuleInstanceContext::instantiate(&mut state, module, config.params, version)?;
let init = SmartModuleInit::try_instantiate(&ctx, &mut state)?;
let transform = create_transform(&ctx, config.initial_data, &mut state)?;
let mut instance = SmartModuleInstance::new(ctx, init, transform);
instance.init(&mut state)?;
instances.push(instance);
}

/// stop adding smartmodule and return SmartModuleChain that can be executed
pub fn initialize(self, engine: &SmartEngine) -> Result<SmartModuleChainInstance> {
let mut instances = Vec::with_capacity(self.smart_modules.len());
let mut state = engine.new_state();
for (config, bytes) in self.smart_modules {
let module = Module::new(&engine.0, bytes)?;
let version = config.version();
let ctx = SmartModuleInstanceContext::instantiate(
&mut state,
module,
config.params,
version,
)?;
let init = SmartModuleInit::try_instantiate(&ctx, &mut state)?;
let transform = create_transform(&ctx, config.initial_data, &mut state)?;
let mut instance = SmartModuleInstance::new(ctx, init, transform);
instance.init(&mut state)?;
instances.push(instance);
}
Ok(SmartModuleChainInstance {
Ok(SmartModuleChainInstance {
inner: SmartModuleChainInstanceImp {
store: state,
instances,
})
}
}

impl<T: Into<Vec<u8>>> From<(SmartModuleConfig, T)> for SmartModuleChainBuilder {
fn from(pair: (SmartModuleConfig, T)) -> Self {
let mut result = Self::default();
result.add_smart_module(pair.0, pair.1.into());
result
}
},
})
}

/// SmartModule Chain Instance that can be executed
pub struct SmartModuleChainInstance {
pub struct SmartModuleChainInstanceImp {
store: WasmState,
instances: Vec<SmartModuleInstance>,
}

impl Debug for SmartModuleChainInstance {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "SmartModuleChainInstance")
}
}

impl SmartModuleChainInstance {
impl SmartModuleChainInstanceImp {
#[cfg(test)]
pub(crate) fn instances(&self) -> &Vec<SmartModuleInstance> {
&self.instances
Expand Down Expand Up @@ -155,79 +118,10 @@ impl SmartModuleChainInstance {
}
}

/// Initial seed data to passed, this will be send back as part of the output
#[derive(Debug, Clone)]
#[non_exhaustive]
pub enum SmartModuleInitialData {
None,
Aggregate { accumulator: Vec<u8> },
}

impl SmartModuleInitialData {
pub fn with_aggregate(accumulator: Vec<u8>) -> Self {
Self::Aggregate { accumulator }
}
}

impl Default for SmartModuleInitialData {
fn default() -> Self {
Self::None
}
}

/// SmartModule configuration
#[derive(Builder)]
pub struct SmartModuleConfig {
#[builder(default, setter(strip_option))]
initial_data: SmartModuleInitialData,
#[builder(default)]
params: SmartModuleExtraParams,
// this will be deprecated in the future
#[builder(default, setter(into, strip_option))]
version: Option<i16>,
}

impl SmartModuleConfigBuilder {
/// add initial parameters
pub fn param(&mut self, key: impl Into<String>, value: impl Into<String>) -> &mut Self {
let mut new = self;
let mut params = new.params.take().unwrap_or_default();
params.insert(key.into(), value.into());
new.params = Some(params);
new
}
}

impl SmartModuleConfig {
pub fn builder() -> SmartModuleConfigBuilder {
SmartModuleConfigBuilder::default()
}

pub(crate) fn version(&self) -> i16 {
self.version.unwrap_or(DEFAULT_SMARTENGINE_VERSION)
}
}

#[cfg(feature = "transformation")]
impl From<crate::transformation::TransformationStep> for SmartModuleConfig {
fn from(step: crate::transformation::TransformationStep) -> Self {
Self {
initial_data: SmartModuleInitialData::None,
params: step
.with
.into_iter()
.map(|(k, v)| (k, v.into()))
.collect::<std::collections::BTreeMap<String, String>>()
.into(),
version: None,
}
}
}

#[cfg(test)]
mod test {

use super::SmartModuleConfig;
use crate::SmartModuleConfig;

#[test]
fn test_param() {
Expand All @@ -247,15 +141,15 @@ mod chaining_test {

use fluvio_smartmodule::{dataplane::smartmodule::SmartModuleInput, Record};

use crate::{
use super::super::{
SmartEngine, SmartModuleChainBuilder, SmartModuleConfig, SmartModuleInitialData,
metrics::SmartModuleChainMetrics,
};

const SM_FILTER_INIT: &str = "fluvio_smartmodule_filter_init";
const SM_MAP: &str = "fluvio_smartmodule_map";

use crate::fixture::read_wasm_module;
use super::super::fixture::read_wasm_module;

#[ignore]
#[test]
Expand All @@ -279,7 +173,8 @@ mod chaining_test {

let mut chain = chain_builder
.initialize(&engine)
.expect("failed to build chain");
.expect("failed to build chain")
.inner;
assert_eq!(chain.instances().len(), 2);

let input = vec![Record::new("hello world")];
Expand Down Expand Up @@ -333,7 +228,8 @@ mod chaining_test {

let mut chain = chain_builder
.initialize(&engine)
.expect("failed to build chain");
.expect("failed to build chain")
.inner;
assert_eq!(chain.instances().len(), 2);

let input = vec![
Expand Down Expand Up @@ -372,7 +268,8 @@ mod chaining_test {
let chain_builder = SmartModuleChainBuilder::default();
let mut chain = chain_builder
.initialize(&engine)
.expect("failed to build chain");
.expect("failed to build chain")
.inner;

assert_eq!(chain.store.get_used_fuel(), 0);

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@ use fluvio_smartmodule::dataplane::smartmodule::{
};
use wasmtime::{AsContextMut, TypedFunc};

use crate::instance::SmartModuleInstanceContext;
use super::instance::SmartModuleInstanceContext;

pub(crate) const INIT_FN_NAME: &str = "init";
type WasmInitFn = TypedFunc<(i32, i32, u32), i32>;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -12,10 +12,10 @@ use fluvio_smartmodule::dataplane::smartmodule::{
SmartModuleExtraParams, SmartModuleInput, SmartModuleOutput, SmartModuleInitInput,
};

use crate::error::EngineError;
use crate::init::SmartModuleInit;
use crate::{WasmSlice, memory};
use crate::state::WasmState;
use super::error::EngineError;
use super::init::SmartModuleInit;
use super::{WasmSlice, memory};
use super::state::WasmState;

pub(crate) struct SmartModuleInstance {
ctx: SmartModuleInstanceContext,
Expand Down
Loading