23 replies

Implementing a Rendezvous Channel in Rust

trying to implement a 'rendezvous' channel (https://doc.rust-lang.org/std/sync/mpsc/)

Note that a bound of 0 is allowed, causing the channel to become a “rendezvous” channel where each sender atomically hands off a message to a receiver.

basically a channel where both

take

take

and

offer

offer

suspend until there is a corresponding action on the other end to complete the transfer

Deferred

Deferred

doesn't work here because all of the 'complete' functions send immediately, whether or not anyone is

await

await

ing

But similarly to the rust implementation, I would expect this to be possible with a 0 bounded

Queue

Queue

, so I tried it out:

in the case where

take

take

is called first, everything works as expected:

const program = Effect.gen(function* (_) {
  const rendevous = yield* _(Queue.bounded<string>(0));

  const fiber = yield* _(
    Effect.fork(
      Effect.gen(function* (_) {
        yield* _(Effect.sleep("100 millis"));
        yield* _(Effect.log("sending message"));
        yield* _(Queue.offer(rendevous, "Hello World"));
        yield* _(Effect.log("sent message"));
      })
    )
  );

  const fiber2 = yield* _(
    Effect.fork(
      Effect.gen(function* (_) {
        yield* _(Effect.sleep("50 millis"));
        yield* _(Effect.log("receiving message"));
        const message = yield* _(Queue.take(rendevous));
        yield* _(Effect.log("received message"));
        yield* _(Effect.log(message));
      })
    )
  );

  yield* _(Fiber.join(Fiber.zip(fiber, fiber2)));
});

await Effect.runPromise(program);

const program = Effect.gen(function* (_) {
  const rendevous = yield* _(Queue.bounded<string>(0));

  const fiber = yield* _(
    Effect.fork(
      Effect.gen(function* (_) {
        yield* _(Effect.sleep("100 millis"));
        yield* _(Effect.log("sending message"));
        yield* _(Queue.offer(rendevous, "Hello World"));
        yield* _(Effect.log("sent message"));
      })
    )
  );

  const fiber2 = yield* _(
    Effect.fork(
      Effect.gen(function* (_) {
        yield* _(Effect.sleep("50 millis"));
        yield* _(Effect.log("receiving message"));
        const message = yield* _(Queue.take(rendevous));
        yield* _(Effect.log("received message"));
        yield* _(Effect.log(message));
      })
    )
  );

  yield* _(Fiber.join(Fiber.zip(fiber, fiber2)));
});

await Effect.runPromise(program);

timestamp=2023-10-06T04:36:09.072Z level=INFO fiber=#2 message="receiving message"
timestamp=2023-10-06T04:36:09.118Z level=INFO fiber=#1 message="sending message"
timestamp=2023-10-06T04:36:09.119Z level=INFO fiber=#1 message="sent message"
timestamp=2023-10-06T04:36:09.121Z level=INFO fiber=#2 message="received message"
timestamp=2023-10-06T04:36:09.122Z level=INFO fiber=#2 message="Hello World"

timestamp=2023-10-06T04:36:09.072Z level=INFO fiber=#2 message="receiving message"
timestamp=2023-10-06T04:36:09.118Z level=INFO fiber=#1 message="sending message"
timestamp=2023-10-06T04:36:09.119Z level=INFO fiber=#1 message="sent message"
timestamp=2023-10-06T04:36:09.121Z level=INFO fiber=#2 message="received message"
timestamp=2023-10-06T04:36:09.122Z level=INFO fiber=#2 message="Hello World"

but if we swap the sleep times so

offer

offer

is called first, it deadlocks

const program = Effect.gen(function* (_) {
  const rendevous = yield* _(Queue.bounded<string>(0));

  const fiber = yield* _(
    Effect.fork(
      Effect.gen(function* (_) {
        yield* _(Effect.sleep("50 millis"));
        yield* _(Effect.log("sending message"));
        yield* _(Queue.offer(rendevous, "Hello World"));
        yield* _(Effect.log("sent message"));
      })
    )
  );

  const fiber2 = yield* _(
    Effect.fork(
      Effect.gen(function* (_) {
        yield* _(Effect.sleep("100 millis"));
        yield* _(Effect.log("receiving message"));
        const message = yield* _(Queue.take(rendevous));
        yield* _(Effect.log("received message"));
        yield* _(Effect.log(message));
      })
    )
  );

  yield* _(Fiber.join(Fiber.zip(fiber, fiber2)));
});

await Effect.runPromise(program);

const program = Effect.gen(function* (_) {
  const rendevous = yield* _(Queue.bounded<string>(0));

  const fiber = yield* _(
    Effect.fork(
      Effect.gen(function* (_) {
        yield* _(Effect.sleep("50 millis"));
        yield* _(Effect.log("sending message"));
        yield* _(Queue.offer(rendevous, "Hello World"));
        yield* _(Effect.log("sent message"));
      })
    )
  );

  const fiber2 = yield* _(
    Effect.fork(
      Effect.gen(function* (_) {
        yield* _(Effect.sleep("100 millis"));
        yield* _(Effect.log("receiving message"));
        const message = yield* _(Queue.take(rendevous));
        yield* _(Effect.log("received message"));
        yield* _(Effect.log(message));
      })
    )
  );

  yield* _(Fiber.join(Fiber.zip(fiber, fiber2)));
});

await Effect.runPromise(program);

timestamp=2023-10-06T04:37:09.794Z level=INFO fiber=#1 message="sending message"
timestamp=2023-10-06T04:37:09.841Z level=INFO fiber=#2 message="receiving message"
<hangs>

timestamp=2023-10-06T04:37:09.794Z level=INFO fiber=#1 message="sending message"
timestamp=2023-10-06T04:37:09.841Z level=INFO fiber=#2 message="receiving message"
<hangs>

in this second case I would expect

offer

offer

to suspend until

take

take

is called, and at that point unblock and send the data through
but that does not seem to be the case

---

Is this intended behavior?
If so, how would one create a 'rendezvous' channel in Effect?

Implementing a Rendezvous Channel in Rust

trying to implement a 'rendezvous' channel (https://doc.rust-lang.org/std/sync/mpsc/)

Note that a bound of 0 is allowed, causing the channel to become a “rendezvous” channel where each sender atomically hands off a message to a receiver.

basically a channel where both

take

take

and

offer

offer

suspend until there is a corresponding action on the other end to complete the transfer

Deferred

Deferred

doesn't work here because all of the 'complete' functions send immediately, whether or not anyone is

await

await

ing

But similarly to the rust implementation, I would expect this to be possible with a 0 bounded

Queue

Queue

, so I tried it out:

in the case where

take

take

is called first, everything works as expected:

const program = Effect.gen(function* (_) {
  const rendevous = yield* _(Queue.bounded<string>(0));

  const fiber = yield* _(
    Effect.fork(
      Effect.gen(function* (_) {
        yield* _(Effect.sleep("100 millis"));
        yield* _(Effect.log("sending message"));
        yield* _(Queue.offer(rendevous, "Hello World"));
        yield* _(Effect.log("sent message"));
      })
    )
  );

  const fiber2 = yield* _(
    Effect.fork(
      Effect.gen(function* (_) {
        yield* _(Effect.sleep("50 millis"));
        yield* _(Effect.log("receiving message"));
        const message = yield* _(Queue.take(rendevous));
        yield* _(Effect.log("received message"));
        yield* _(Effect.log(message));
      })
    )
  );

  yield* _(Fiber.join(Fiber.zip(fiber, fiber2)));
});

await Effect.runPromise(program);

const program = Effect.gen(function* (_) {
  const rendevous = yield* _(Queue.bounded<string>(0));

  const fiber = yield* _(
    Effect.fork(
      Effect.gen(function* (_) {
        yield* _(Effect.sleep("100 millis"));
        yield* _(Effect.log("sending message"));
        yield* _(Queue.offer(rendevous, "Hello World"));
        yield* _(Effect.log("sent message"));
      })
    )
  );

  const fiber2 = yield* _(
    Effect.fork(
      Effect.gen(function* (_) {
        yield* _(Effect.sleep("50 millis"));
        yield* _(Effect.log("receiving message"));
        const message = yield* _(Queue.take(rendevous));
        yield* _(Effect.log("received message"));
        yield* _(Effect.log(message));
      })
    )
  );

  yield* _(Fiber.join(Fiber.zip(fiber, fiber2)));
});

await Effect.runPromise(program);

timestamp=2023-10-06T04:36:09.072Z level=INFO fiber=#2 message="receiving message"
timestamp=2023-10-06T04:36:09.118Z level=INFO fiber=#1 message="sending message"
timestamp=2023-10-06T04:36:09.119Z level=INFO fiber=#1 message="sent message"
timestamp=2023-10-06T04:36:09.121Z level=INFO fiber=#2 message="received message"
timestamp=2023-10-06T04:36:09.122Z level=INFO fiber=#2 message="Hello World"

timestamp=2023-10-06T04:36:09.072Z level=INFO fiber=#2 message="receiving message"
timestamp=2023-10-06T04:36:09.118Z level=INFO fiber=#1 message="sending message"
timestamp=2023-10-06T04:36:09.119Z level=INFO fiber=#1 message="sent message"
timestamp=2023-10-06T04:36:09.121Z level=INFO fiber=#2 message="received message"
timestamp=2023-10-06T04:36:09.122Z level=INFO fiber=#2 message="Hello World"

but if we swap the sleep times so

offer

offer

is called first, it deadlocks

const program = Effect.gen(function* (_) {
  const rendevous = yield* _(Queue.bounded<string>(0));

  const fiber = yield* _(
    Effect.fork(
      Effect.gen(function* (_) {
        yield* _(Effect.sleep("50 millis"));
        yield* _(Effect.log("sending message"));
        yield* _(Queue.offer(rendevous, "Hello World"));
        yield* _(Effect.log("sent message"));
      })
    )
  );

  const fiber2 = yield* _(
    Effect.fork(
      Effect.gen(function* (_) {
        yield* _(Effect.sleep("100 millis"));
        yield* _(Effect.log("receiving message"));
        const message = yield* _(Queue.take(rendevous));
        yield* _(Effect.log("received message"));
        yield* _(Effect.log(message));
      })
    )
  );

  yield* _(Fiber.join(Fiber.zip(fiber, fiber2)));
});

await Effect.runPromise(program);

const program = Effect.gen(function* (_) {
  const rendevous = yield* _(Queue.bounded<string>(0));

  const fiber = yield* _(
    Effect.fork(
      Effect.gen(function* (_) {
        yield* _(Effect.sleep("50 millis"));
        yield* _(Effect.log("sending message"));
        yield* _(Queue.offer(rendevous, "Hello World"));
        yield* _(Effect.log("sent message"));
      })
    )
  );

  const fiber2 = yield* _(
    Effect.fork(
      Effect.gen(function* (_) {
        yield* _(Effect.sleep("100 millis"));
        yield* _(Effect.log("receiving message"));
        const message = yield* _(Queue.take(rendevous));
        yield* _(Effect.log("received message"));
        yield* _(Effect.log(message));
      })
    )
  );

  yield* _(Fiber.join(Fiber.zip(fiber, fiber2)));
});

await Effect.runPromise(program);

timestamp=2023-10-06T04:37:09.794Z level=INFO fiber=#1 message="sending message"
timestamp=2023-10-06T04:37:09.841Z level=INFO fiber=#2 message="receiving message"
<hangs>

timestamp=2023-10-06T04:37:09.794Z level=INFO fiber=#1 message="sending message"
timestamp=2023-10-06T04:37:09.841Z level=INFO fiber=#2 message="receiving message"
<hangs>

in this second case I would expect

offer

offer

to suspend until

take

take

is called, and at that point unblock and send the data through
but that does not seem to be the case

---

Is this intended behavior?
If so, how would one create a 'rendezvous' channel in Effect?

Implementing a Rendezvous Channel in Rust

Implementing a Rendezvous Channel in Rust

Similar Threads

Similar Threads

Similar Threads