Use fine-grained crossbeam dependencies (#688)

Author: djc

core/Cargo.toml

@@ -27,7 +27,8 @@ arc-swap = "1.3"
 async-trait = "0.1"
 base64 = "0.21"
 console-subscriber = { version = "0.1", optional = true }
-crossbeam = "0.8"
+crossbeam-channel = "0.5"
+crossbeam-queue = "0.3"
 dashmap = "5.5"
 derive_builder = "0.12"
 derive_more = "0.99"

core/src/core_tests/local_activities.rs

@@ -8,7 +8,7 @@ use crate::{
     worker::{client::mocks::mock_workflow_client, LEGACY_QUERY_ID},
 };
 use anyhow::anyhow;
-use crossbeam::queue::SegQueue;
+use crossbeam_queue::SegQueue;
 use futures::{future::join_all, FutureExt};
 use std::{
     collections::HashMap,

core/src/telemetry/metrics.rs

@@ -632,14 +632,14 @@ pub struct MetricsCallBuffer<I>
 where
     I: BufferInstrumentRef,
 {
-    calls_rx: crossbeam::channel::Receiver<MetricEvent<I>>,
+    calls_rx: crossbeam_channel::Receiver<MetricEvent<I>>,
     calls_tx: LogErrOnFullSender<MetricEvent<I>>,
 }
 #[derive(Clone, Debug)]
-struct LogErrOnFullSender<I>(crossbeam::channel::Sender<I>);
+struct LogErrOnFullSender<I>(crossbeam_channel::Sender<I>);
 impl<I> LogErrOnFullSender<I> {
     fn send(&self, v: I) {
-        if let Err(crossbeam::channel::TrySendError::Full(_)) = self.0.try_send(v) {
+        if let Err(crossbeam_channel::TrySendError::Full(_)) = self.0.try_send(v) {
             error!(
                 "Core's metrics buffer is full! Dropping call to record metrics. \
                  Make sure you drain the metric buffer often!"
@@ -654,7 +654,7 @@ where
 {
     /// Create a new buffer with the given capacity
     pub fn new(buffer_size: usize) -> Self {
-        let (calls_tx, calls_rx) = crossbeam::channel::bounded(buffer_size);
+        let (calls_tx, calls_rx) = crossbeam_channel::bounded(buffer_size);
         MetricsCallBuffer {
             calls_rx,
             calls_tx: LogErrOnFullSender(calls_tx),

core/src/worker/workflow/workflow_stream/saved_wf_inputs.rs

@@ -9,7 +9,7 @@ use crate::{
         LocalActRequest, LocalActivityResolution,
     },
 };
-use crossbeam::queue::SegQueue;
+use crossbeam_queue::SegQueue;
 use futures::Stream;
 use futures_util::StreamExt;
 use serde::{Deserialize, Serialize};

sdk/Cargo.toml

@@ -17,7 +17,7 @@ async-trait = "0.1"
 thiserror = "1.0"
 anyhow = "1.0"
 base64 = "0.21"
-crossbeam = "0.8"
+crossbeam-channel = "0.5"
 derive_more = "0.99"
 futures = "0.3"
 once_cell = "1.10"

sdk/src/workflow_context.rs

@@ -11,7 +11,7 @@ use crate::{
     IntoUpdateValidatorFunc, RustWfCmd, SignalExternalWfResult, TimerResult, UnblockEvent,
     Unblockable, UpdateFunctions,
 };
-use crossbeam::channel::{Receiver, Sender};
+use crossbeam_channel::{Receiver, Sender};
 use futures::{task::Context, FutureExt, Stream, StreamExt};
 use parking_lot::RwLock;
 use std::{
@@ -117,7 +117,7 @@ impl WfContext {
         am_cancelled: watch::Receiver<bool>,
     ) -> (Self, Receiver<RustWfCmd>) {
         // We need to use a normal std channel since our receiving side is non-async
-        let (chan, rx) = crossbeam::channel::unbounded();
+        let (chan, rx) = crossbeam_channel::unbounded();
         (
             Self {
                 namespace,

sdk/src/workflow_future.rs

@@ -4,7 +4,7 @@ use crate::{
     WorkflowResult,
 };
 use anyhow::{anyhow, bail, Context as AnyhowContext, Error};
-use crossbeam::channel::Receiver;
+use crossbeam_channel::Receiver;
 use futures::{future::BoxFuture, FutureExt};
 use std::{
     collections::{hash_map::Entry, HashMap},