Adds high-level restate-types Scope, ServiceName, and LockName

ServiceName uses the new `InternedReString` for thread-local checked interning when it makes sense.

LockName will be used for representing the new service locking mechanism for vqueues that will be introduced in the next commit(s).

Author: AhmedSoliman

crates/types/Cargo.toml

@@ -30,6 +30,7 @@ restate-memory = { workspace = true }
 restate-serde-util = { workspace = true }
 restate-test-util = { workspace = true, optional = true }
 restate-time-util = { workspace = true, features = ["serde", "serde_with"] }
+restate-util-string = { workspace = true, features = ["serde", "bilrost"] }
 restate-utoipa = { workspace = true }
 
 adaptive-timeout = { workspace = true, features = ["serde"] }
@@ -89,7 +90,6 @@ serde_path_to_error = { version = "0.1" }
 serde_with = { workspace = true, features = ["json"] }
 sha2 = { workspace = true }
 smallvec = { workspace = true }
-restate-util-string = { workspace = true, features = ["serde"] }
 static_assertions = { workspace = true }
 strum = { workspace = true }
 tempfile = { workspace = true, optional = true }

crates/types/src/identifiers.rs

@@ -10,6 +10,7 @@
 
 //! Restate uses many identifiers to uniquely identify its services and entities.
 
+use std::cell::RefCell;
 use std::fmt::{self, Display, Formatter};
 use std::hash::Hash;
 use std::mem::size_of;
@@ -34,6 +35,12 @@ use crate::invocation::{InvocationTarget, InvocationTargetType, WorkflowHandlerT
 use crate::journal_v2::SignalId;
 use crate::time::MillisSinceEpoch;
 
+thread_local! {
+    // a thread-local xxh3 hashing state to reuse its allocation since its internal buffer is quite
+    // large (~500 bytes)
+    static PARTITION_KEY_HASHER: RefCell<xxhash_rust::xxh3::Xxh3> = const { RefCell::new(xxhash_rust::xxh3::Xxh3::new()) };
+}
+
 /// Identifying the leader epoch of a partition processor
 #[derive(
     PartialEq,
@@ -711,7 +718,7 @@ impl WithPartitionKey for IdempotencyId {
 pub type ServiceRevision = u32;
 
 pub mod partitioner {
-    use super::PartitionKey;
+    use super::{PARTITION_KEY_HASHER, PartitionKey};
 
     use std::hash::{Hash, Hasher};
 
@@ -720,9 +727,12 @@ pub mod partitioner {
 
     impl HashPartitioner {
         pub fn compute_partition_key(value: impl Hash) -> PartitionKey {
-            let mut hasher = xxhash_rust::xxh3::Xxh3::default();
-            value.hash(&mut hasher);
-            hasher.finish()
+            PARTITION_KEY_HASHER.with_borrow_mut(|hasher| {
+                value.hash(hasher);
+                let key = hasher.finish();
+                hasher.reset();
+                key
+            })
         }
     }
 }

crates/types/src/invocation/mod.rs

@@ -12,28 +12,31 @@
 
 pub mod client;
 
-use crate::errors::InvocationError;
-use crate::identifiers::{
-    DeploymentId, EntryIndex, IdempotencyId, InvocationId, PartitionKey,
-    PartitionProcessorRpcRequestId, ServiceId, SubscriptionId, WithInvocationId, WithPartitionKey,
-};
-use crate::journal_v2::{CompletionId, GetInvocationOutputResult, Signal};
-use crate::time::MillisSinceEpoch;
-use crate::{GenerationalNodeId, RestateVersion};
-
-use bytes::Bytes;
-use bytestring::ByteString;
-use opentelemetry::trace::{SpanContext, SpanId, TraceFlags, TraceState};
-use serde_with::{DisplayFromStr, FromInto, serde_as};
 use std::borrow::Cow;
 use std::hash::Hash;
 use std::ops::Deref;
 use std::str::FromStr;
 use std::time::Duration;
 use std::{cmp, fmt};
 
+use bytes::Bytes;
+use bytestring::ByteString;
+use opentelemetry::trace::{SpanContext, SpanId, TraceFlags, TraceState};
 // Re-exporting opentelemetry [`TraceId`] to avoid having to import opentelemetry in all crates.
 pub use opentelemetry::trace::TraceId;
+use serde_with::{DisplayFromStr, FromInto, serde_as};
+
+use restate_util_string::ReString;
+
+use crate::Scope;
+use crate::errors::InvocationError;
+use crate::identifiers::{
+    DeploymentId, EntryIndex, IdempotencyId, InvocationId, PartitionKey,
+    PartitionProcessorRpcRequestId, ServiceId, SubscriptionId, WithInvocationId, WithPartitionKey,
+};
+use crate::journal_v2::{CompletionId, GetInvocationOutputResult, Signal};
+use crate::time::MillisSinceEpoch;
+use crate::{GenerationalNodeId, LockName, RestateVersion, ServiceName};
 
 #[derive(Eq, Hash, PartialEq, Clone, Copy, Debug, serde::Serialize, serde::Deserialize)]
 #[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
@@ -220,6 +223,25 @@ impl InvocationTarget {
         }
     }
 
+    pub fn scope(&self) -> Option<Scope> {
+        // todo: This needs to be filled
+        None
+    }
+
+    pub fn lock_name(&self) -> Option<LockName> {
+        match self {
+            InvocationTarget::Service { .. } => None,
+            InvocationTarget::VirtualObject { name, key, .. } => Some(LockName::new(
+                ServiceName::new(name.as_ref()),
+                ReString::from(key.as_ref()),
+            )),
+            InvocationTarget::Workflow { name, key, .. } => Some(LockName::new(
+                ServiceName::new(name.as_ref()),
+                ReString::from(key.as_ref()),
+            )),
+        }
+    }
+
     pub fn key(&self) -> Option<&ByteString> {
         match self {
             InvocationTarget::Service { .. } => None,

crates/types/src/lib.rs

@@ -12,6 +12,7 @@
 
 mod base62_util;
 mod id_util;
+mod locking;
 mod macros;
 mod node_id;
 mod restate_version;
@@ -58,12 +59,18 @@ pub mod timer;
 pub mod vqueue;
 
 pub use id_util::IdResourceType;
+pub use locking::*;
 pub use node_id::*;
+use restate_encoding::BilrostNewType;
+use restate_util_string::InternedReString;
 pub use restate_version::*;
 pub use version::*;
 
 // Re-export of the old time module by delegating to the restate-clock crate.
 pub use restate_clock::time;
+
+use self::identifiers::partitioner::HashPartitioner;
+use self::identifiers::{PartitionKey, WithPartitionKey};
 pub mod clock {
     pub use restate_clock::*;
 }
@@ -76,6 +83,151 @@ pub mod memory {
 // Re-export metrics' SharedString (Space-efficient Cow + RefCounted variant)
 pub type SharedString = metrics::SharedString;
 
+/// An interned service name
+#[derive(
+    derive_more::Display,
+    derive_more::Debug,
+    derive_more::AsRef,
+    derive_more::From,
+    Clone,
+    Eq,
+    PartialEq,
+    Ord,
+    PartialOrd,
+    Hash,
+    BilrostNewType,
+)]
+#[debug("{}", _0)]
+#[display("{}", _0)]
+#[repr(transparent)]
+pub struct ServiceName(InternedReString);
+
+impl ServiceName {
+    #[inline]
+    pub fn new(value: &str) -> Self {
+        assert!(!value.is_empty());
+        Self(InternedReString::new(value))
+    }
+
+    #[inline]
+    pub const fn from_static(value: &'static str) -> Self {
+        assert!(!value.is_empty());
+        Self(InternedReString::from_static(value))
+    }
+
+    #[inline]
+    pub fn as_str(&self) -> &str {
+        self.0.as_str()
+    }
+
+    #[inline]
+    pub fn as_bytes(&self) -> &[u8] {
+        self.0.as_bytes()
+    }
+
+    #[inline]
+    pub fn len(&self) -> usize {
+        self.0.len()
+    }
+
+    #[inline]
+    pub fn is_empty(&self) -> bool {
+        self.0.is_empty()
+    }
+}
+
+impl AsRef<str> for ServiceName {
+    #[inline]
+    fn as_ref(&self) -> &str {
+        self.0.as_ref()
+    }
+}
+
+impl std::borrow::Borrow<str> for ServiceName {
+    #[inline]
+    fn borrow(&self) -> &str {
+        self.0.as_ref()
+    }
+}
+
+/// An interned scope
+///
+/// A scope defines the partitioning boundary for sets of service instances and
+/// invocations.
+#[derive(
+    derive_more::Display,
+    derive_more::Debug,
+    derive_more::AsRef,
+    derive_more::From,
+    Clone,
+    Eq,
+    PartialEq,
+    Ord,
+    PartialOrd,
+    Hash,
+    BilrostNewType,
+)]
+#[debug("{}", _0)]
+#[display("{}", _0)]
+#[repr(transparent)]
+pub struct Scope(InternedReString);
+
+impl Scope {
+    #[inline]
+    pub fn new(value: &str) -> Self {
+        assert!(!value.is_empty());
+        Self(InternedReString::new(value))
+    }
+
+    #[inline]
+    pub const fn from_static(value: &'static str) -> Self {
+        assert!(!value.is_empty());
+        Self(InternedReString::from_static(value))
+    }
+
+    #[inline]
+    pub fn as_str(&self) -> &str {
+        self.0.as_str()
+    }
+
+    #[inline]
+    pub fn as_bytes(&self) -> &[u8] {
+        self.0.as_bytes()
+    }
+
+    #[inline]
+    pub fn len(&self) -> usize {
+        self.0.len()
+    }
+
+    #[inline]
+    pub fn is_empty(&self) -> bool {
+        self.0.is_empty()
+    }
+}
+
+impl AsRef<str> for Scope {
+    #[inline]
+    fn as_ref(&self) -> &str {
+        self.0.as_ref()
+    }
+}
+
+impl std::borrow::Borrow<str> for Scope {
+    #[inline]
+    fn borrow(&self) -> &str {
+        self.0.as_ref()
+    }
+}
+
+impl WithPartitionKey for Scope {
+    #[inline]
+    fn partition_key(&self) -> PartitionKey {
+        // the partition key is calculated directly from the scope value
+        HashPartitioner::compute_partition_key(&self.0)
+    }
+}
+
 /// Trait for merging two attributes
 pub trait Merge {
     /// Return true if the value was mutated as a result of the merge