15 replies

How to reuse outputs from effects with unbounded concurrency?

Hi, I've started using Effect a few days ago and I'm stuck trying to do something...
The image shows what I'm trying to do. I have a few tasks: task 1, task 2 (which needs the result from task 1), and task 3 (which also needs the result from task 1). I also only want each task to execute once.

My first try was to use

Effect.do

Effect.do

Effect.do

Effect.do and the code looked something like

const task1 = Effect.fn("task1")(function* () {
    console.log("task1");
    return 1;
})

const task2 = Effect.fn("task2")(function* (result1: number) {
    console.log(`task2 [result1=${result1}]`);
    yield* Effect.sleep("1 second");
    return 2;
})

const task3 = Effect.fn("task3")(function* (result1: number) {
    console.log(`task3 [result1=${result1}]`);
    return 3;
})

const task4 = Effect.fn("task4")(function* (result2: number, result3: number) {
    console.log(`task4 [result2=${result2}, result3=${result3}]`);
    return 4;
})

const program = Effect.Do.pipe(
    Effect.bind("result1", () => task1()),
    Effect.bind("result2", ({ result1 }) => task2(result1)),
    Effect.bind("result3", ({ result1 }) => task3(result1)),
    Effect.bind("result4", ({ result2, result3 }) => task4(result2, result3)),
    Effect.andThen(Effect.log),
)

NodeRuntime.runMain(program)

const task1 = Effect.fn("task1")(function* () {
    console.log("task1");
    return 1;
})

const task2 = Effect.fn("task2")(function* (result1: number) {
    console.log(`task2 [result1=${result1}]`);
    yield* Effect.sleep("1 second");
    return 2;
})

const task3 = Effect.fn("task3")(function* (result1: number) {
    console.log(`task3 [result1=${result1}]`);
    return 3;
})

const task4 = Effect.fn("task4")(function* (result2: number, result3: number) {
    console.log(`task4 [result2=${result2}, result3=${result3}]`);
    return 4;
})

const program = Effect.Do.pipe(
    Effect.bind("result1", () => task1()),
    Effect.bind("result2", ({ result1 }) => task2(result1)),
    Effect.bind("result3", ({ result1 }) => task3(result1)),
    Effect.bind("result4", ({ result2, result3 }) => task4(result2, result3)),
    Effect.andThen(Effect.log),
)

NodeRuntime.runMain(program)

const task1 = Effect.fn("task1")(function* () {
    console.log("task1");
    return 1;
})

const task2 = Effect.fn("task2")(function* (result1: number) {
    console.log(`task2 [result1=${result1}]`);
    yield* Effect.sleep("1 second");
    return 2;
})

const task3 = Effect.fn("task3")(function* (result1: number) {
    console.log(`task3 [result1=${result1}]`);
    return 3;
})

const task4 = Effect.fn("task4")(function* (result2: number, result3: number) {
    console.log(`task4 [result2=${result2}, result3=${result3}]`);
    return 4;
})

const program = Effect.Do.pipe(
    Effect.bind("result1", () => task1()),
    Effect.bind("result2", ({ result1 }) => task2(result1)),
    Effect.bind("result3", ({ result1 }) => task3(result1)),
    Effect.bind("result4", ({ result2, result3 }) => task4(result2, result3)),
    Effect.andThen(Effect.log),
)

NodeRuntime.runMain(program)

const task1 = Effect.fn("task1")(function* () {
    console.log("task1");
    return 1;
})

const task2 = Effect.fn("task2")(function* (result1: number) {
    console.log(`task2 [result1=${result1}]`);
    yield* Effect.sleep("1 second");
    return 2;
})

const task3 = Effect.fn("task3")(function* (result1: number) {
    console.log(`task3 [result1=${result1}]`);
    return 3;
})

const task4 = Effect.fn("task4")(function* (result2: number, result3: number) {
    console.log(`task4 [result2=${result2}, result3=${result3}]`);
    return 4;
})

const program = Effect.Do.pipe(
    Effect.bind("result1", () => task1()),
    Effect.bind("result2", ({ result1 }) => task2(result1)),
    Effect.bind("result3", ({ result1 }) => task3(result1)),
    Effect.bind("result4", ({ result2, result3 }) => task4(result2, result3)),
    Effect.andThen(Effect.log),
)

NodeRuntime.runMain(program)

This approach works, but I couldn't find to make

task2

task2

task2

task2 and

task3

task3

task3

task3 execute in parallel with

Effect.Do

Effect.Do

Effect.Do

Effect.Do. So I tried

Effect.gen

Effect.gen

Effect.gen

Effect.gen:

const program = Effect.gen(function* () {
    const result1 = yield* task1();

    const [result2, result3] = yield* Effect.all([
        task2(result1),
        task3(result1),
    ], { concurrency: "unbounded" })

    const result4 = yield* task4(result2, result3);

    return { result1, result2, result3, result4 };
}).pipe(
    Effect.andThen(Effect.log),
)

const program = Effect.gen(function* () {
    const result1 = yield* task1();

    const [result2, result3] = yield* Effect.all([
        task2(result1),
        task3(result1),
    ], { concurrency: "unbounded" })

    const result4 = yield* task4(result2, result3);

    return { result1, result2, result3, result4 };
}).pipe(
    Effect.andThen(Effect.log),
)

const program = Effect.gen(function* () {
    const result1 = yield* task1();

    const [result2, result3] = yield* Effect.all([
        task2(result1),
        task3(result1),
    ], { concurrency: "unbounded" })

    const result4 = yield* task4(result2, result3);

    return { result1, result2, result3, result4 };
}).pipe(
    Effect.andThen(Effect.log),
)

const program = Effect.gen(function* () {
    const result1 = yield* task1();

    const [result2, result3] = yield* Effect.all([
        task2(result1),
        task3(result1),
    ], { concurrency: "unbounded" })

    const result4 = yield* task4(result2, result3);

    return { result1, result2, result3, result4 };
}).pipe(
    Effect.andThen(Effect.log),
)

This works, but I fear it would get more complicated as more tasks are needed and dependencies become more complex.
Ideally, I would want something like

Effect.Do

Effect.Do

Effect.Do

Effect.Do, but with concurrent evaluation of the effects.

How to reuse outputs from effects with unbounded concurrency?

Similar Threads

Similar Threads

Similar Threads