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.

Sign up for GitHub

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

Jump to bottom

add Stream.share api #3080

Merged
merged 17 commits into from
Sep 9, 2024
Merged

add Stream.share api #3080

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
17 commits
Select commit Hold shift click to select a range
9775cb4
feat(Stream/share): initial implementation
dilame Jun 24, 2024
25f614b
feat(Stream/share): add config and tests
dilame Jul 5, 2024
e61d2f5
feat(Stream/share): fix test
dilame Jul 5, 2024
cccdd87
feat(Stream/share): fix test
dilame Jul 5, 2024
650b825
feat(Stream/share): fix test
dilame Jul 6, 2024
fe3dc3d
feat(Stream): split share and shareRefCount
dilame Jul 7, 2024
4f3f1d3
feat(Stream/share): use unbounded pubsub by default
dilame Jul 7, 2024
7da06ee
refactor(Stream/shareRefCount): rename internal var `consumersCount` …
dilame Jul 7, 2024
12a71b8
refactor(Stream/shareRefCount): fix public declaration
dilame Jul 7, 2024
db839f5
test(Stream/shareRefCount): implement dead-lock test if use `forkScop…
dilame Jul 7, 2024
ecb9e8e
fix(Stream/share): use PubSub.bounded(16) by default
dilame Jul 7, 2024
7953e66
fix(Stream/share): fork in local scope
dilame Jul 8, 2024
2e60d91
fix(Stream/share): reset fiber ref only when refCount===0
dilame Jul 8, 2024
4849549
fix(Stream/share): prevent potential race condition
dilame Jul 8, 2024
b6f27be
broadcast ref count and replay
dilame Jul 9, 2024
f9800cb
add some tests
dilame Jul 9, 2024
c14e39f
rename & update changeset
tim-smart Sep 9, 2024
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
fix(Stream/share): fork in local scope
  • Loading branch information
@dilame @tim-smart
dilame authored and tim-smart committed Sep 9, 2024
commit 7953e66176d76ebd7f8346b6240cdef486e41242
4 changes: 2 additions & 2 deletions packages/effect/src/Stream.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4407,10 4407,10 @@ export const share: {
export const shareRefCount: {
<A, E>(config: {
readonly connector?: Effect.Effect<PubSub.PubSub<Take.Take<A, E>>>
}): <R>(self: Stream<A, E, R>) => Stream<A, E, R>
}): <R>(self: Stream<A, E, R>) => Effect.Effect<Stream<A, E, R>, never, R | Scope.Scope>
<A, E, R>(self: Stream<A, E, R>, config: {
readonly connector?: Effect.Effect<PubSub.PubSub<Take.Take<A, E>>>
}): Stream<A, E, R>
}): Effect.Effect<Stream<A, E, R>, never, R | Scope.Scope>
} = internal.shareRefCount

/**
Expand Down
51 changes: 29 additions & 22 deletions packages/effect/src/internal/stream.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8637,42 8637,49 @@ export const shareRefCount = dual<
readonly connector?: Effect.Effect<PubSub.PubSub<Take.Take<A, E>>>
}) => <R>(
self: Stream.Stream<A, E, R>
) => Stream.Stream<A, E, R>,
) => Effect.Effect<Stream.Stream<A, E, R>, never, R | Scope.Scope>,
<A, E, R>(
self: Stream.Stream<A, E, R>,
config: {
readonly connector?: Effect.Effect<PubSub.PubSub<Take.Take<A, E>>>
}
) => Stream.Stream<A, E, R>
) => Effect.Effect<Stream.Stream<A, E, R>, never, R | Scope.Scope>
>(
2,
<A, E, R>(
self: Stream.Stream<A, E, R>,
config: {
readonly connector?: Effect.Effect<PubSub.PubSub<Take.Take<A, E>>>
}
): Stream.Stream<A, E, R> => {
let refCount = 0
let connector: PubSub.PubSub<Take.Take<A, E>> | null = null
let fiber: Fiber.RuntimeFiber<void> | null = null
): Effect.Effect<Stream.Stream<A, E, R>, never, R | Scope.Scope> => {
return Effect.gen(function*() {
refCount
connector ??= yield* (config.connector ?? PubSub.bounded<Take.Take<A, E>>(16))
fiber ??= yield* Effect.forkDaemon(runIntoPubSub(self, connector))
return flattenTake(fromPubSub(connector))
}).pipe(
unwrap,
ensuring(
Effect.suspend(() => {
refCount--
const cleanup = fiber
fiber = null
return refCount === 0 && cleanup
? Fiber.interrupt(cleanup)
: Effect.void
})
let refCount = 0
let connector: PubSub.PubSub<Take.Take<A, E>> | null = null
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
let connector: PubSub.PubSub<Take.Take<A, E>> | null = null
const connector = yield* (config.connector ?? PubSub.bounded<Take.Take<A, E>>(16))

let fiber: Fiber.RuntimeFiber<void> | null = null
const scope = yield* Effect.scope
const context = yield* Effect.context<R>()
return Effect.gen(function*() {
refCount
connector ??= yield* (config.connector ?? PubSub.bounded<Take.Take<A, E>>(16))
fiber ??= yield* runIntoPubSub(self, connector).pipe(
tim-smart marked this conversation as resolved.
Show resolved Hide resolved
Effect.locally(FiberRef.currentContext, context),
Effect.forkIn(scope)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more. 

declare const stream: Stream<number, never, Foo>
Effect.gen(function* () {
  const shared = yield* sharedRefCount(stream)
  yield* doSomethingWith(shared).pipe(
    Effect.provide(Foo1Layer),
    Effect.forkScoped,
  )
  yield* doSomethingWith(shared).pipe(
    Effect.provide(Foo2Layer),
    Effect.forkScoped
  )
})

@tim-smart if I understand correctly, both instances would use the Foo1 version. No?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

They would use whatever was supplied to the sharedRefCount effect.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

despite this effect being ran as part of the stream evaluation? How?

)
return flattenTake(fromPubSub(connector))
}).pipe(
unwrap,
ensuring(
Effect.suspend(() => {
refCount--
const cleanup = fiber
fiber = null
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You are clearing the fiber even if the ref count is not 0.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it doesn't matter what is the ref count, theres a new pubsub and new forked fiber for every evaluation of listening

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I use Nullish coalescing assignment, so i don't think it's the case here

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You're right, only a new forked fiber on every eval, not new PubSub, I glazed over the ??=

You should still instantiate the PubSub earlier imo. Theres no reason that a pubsub should be instantiated as a hidden side-effect of listening to a stream

return refCount === 0 && cleanup
? Fiber.interrupt(cleanup)
: Effect.void
})
)
)
)
})
}
)

Expand Down
12 changes: 4 additions & 8 deletions packages/effect/test/Stream/sharing.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 1,3 @@
import * as Deferred from "effect/Deferred"
import * as Effect from "effect/Effect"
import * as Fiber from "effect/Fiber"
import * as Schedule from "effect/Schedule"
Expand Down Expand Up @@ -38,14 37,11 @@ describe("Stream", () => {
assert.deepStrictEqual(second, [2])
}))

it.effect("shareRefCount sequenced", () =>
it.scoped("shareRefCount sequenced", () =>
Effect.gen(function*() {
const sharedStream = Stream.fromSchedule(
const sharedStream = yield* Stream.fromSchedule(
Schedule.spaced("1 seconds")
).pipe(
Stream.ensuringWith(() => {
return Effect.void
}),
Stream.shareRefCount({})
)

Expand Down Expand Up @@ -74,9 70,9 @@ describe("Stream", () => {
assert.deepStrictEqual(second, [0])
}))

it.effect("shareRefCount parallel", () =>
it.scoped("shareRefCount parallel", () =>
Effect.gen(function*() {
const sharedStream = Stream.fromSchedule(
const sharedStream = yield* Stream.fromSchedule(
Schedule.spaced("1 seconds")
).pipe(Stream.shareRefCount({}))

Expand Down