cujojs/most

you can think of Stream objects as the nozzle of a whipped cream can

Micah

@micahscopes

@TylorS I tried it out and it worked great! Except for the timing issues caused by doing timing in the main thread where all the heavy react stuff and game animation were also being computed. So my next step is to attempt using the @most stuff in an audioworklet processor... stay tuned, I'll share more when I've made some progress. I'm really pumped to be able to do generative music stuff with @most and lodash or whatever, it's a long time dream come true

If I'm successful, it seems like it'd be useful to share any extra utilities I make for doing that in my little fork

Micah

@micahscopes

I have a periodic stream and I'd like to make a new periodic stream derived from it that only includes every Nth event. How can I do this?

Tylor Steinberger

@TylorS

Hey @micahscopes you might be able to do what you need using loop and switchLatest/chain/mergeConcurrently depending on the concurrency you need. Something like this to do the nth of a Stream might help

const nth = <A>(n: number, stream: Stream<A>) => 
  switchLatest(loop(
    (acc, a) => acc === n  
      ? { seed: 0, value: now(a) } 
      : { seed: acc + 1, value: empty() }, 
    0, 
    stream
  ))

Tylor Steinberger

@TylorS

From there you should be able to construct a higher-order stream for the periodic portions

const stream = chain(
  () => nth(n2, periodic(n3)),
  periodic(n1)
)

Micah

@micahscopes

awesome, thanks @TylorS

Micah

@micahscopes

by the way, I got my scheduler working inside of an audioworklet processor, sending MIDI events to another audioworklet processor! it's awesome

Micah

@micahscopes

just wanted to share that I've pretty much finished prototyping using @most/core inside of an AudioWorkletProcessor to load, generate, schedule and transform MIDI events that get played by this synth

I'm planning on releasing this stuff as a little toolkit once I get a chance to clean it up and organize it!

Micah

@micahscopes

Is there a way to manually clean up references from old events in a scheduler during a runEffects of a perpetual stream? I'm using this big giant aggregate stream that I just pass into runEffects and am leaking memory somehow, at a rate of about ~3MB/hour, maybe because of how I'm generating random values using patterns like

periodic(t)
  |> filter(() => Math.random() < 0.4)
  |> delay(d)
  |> map(() => fnThatGeneratesLotsOfObjects()
  |> ... // and more stuff that generates objects

waiting around idle.heapprofile

^ there's the heap profile, you can see that most of the pileup is in SchedulerImpl.scheduleTask, so I assume maybe it's the scheduler's Timeline growing unboundedly

Maybe the simplest way to deal with this would be to just call runEffects periodically

Micah

@micahscopes

I did do a test and found that there's nothing sticking around after runEffects finishes, so it's airtight on that level

(a screenshot of the heapprofile shared above):

Tylor Steinberger

@TylorS

Hey @micahscopes :wave:

I'm glad to hear you've got your prototype working, congrats!

There is 1 method for cleanup up a Scheduled task from the Timeline here on the scheduler, but oftentimes you won't have direct access to those scheduled events https://github.com/mostjs/core/blob/master/packages/types/index.d.ts#L50

I'm not familiar enough with the heap profile to decipher and help from there, but only a few operators make heavy use of the scheduler

Tylor Steinberger

@TylorS

now, at, and some of the other synchronous constructors use it only once to schedule the start of the stream, periodic being the expeception which schedules over and over again. On the transformation side though, delay, debounce, and throttle utilize the scheduler pretty aggressively but it'll only ever have a single scheduled task at a time per execution context

Tylor Steinberger

@TylorS

Actually, that's not true of delay - it creates n number of tasks per n number of events. So if you're encountering a lot of memory usage it's probably most likely from delay

Micah

@micahscopes

@TylorS the heap chart seems to agree with your observations, if I'm reading it right. DelaySink.event has the most scheduleTask memory usage, followed by TapSink.event and then FilterSink.event.

The thing I don't get is just exactly how streams that have ended get cleaned up. If I schedule the streams in the exact same way, except that I give them an end and run them consecutively, the heap allocations get garbage collected.

The only thing I can think is happening is that maybe scheduled events are hanging around in the Timeline and this is why memory usage is accumulating? Maybe somehow running cancel for past events would solve my problem?

Tylor Steinberger

@TylorS

It's pretty much built into runEffects via this Sink instance. When sink.end or sink.error is called it will dispose of the Disposablereturned by calling Stream.run(sink, scheduler)

As you construct a stream graph, many combinators will call stream.run(sink, scheduler) and collect disposables into their own graph of sorts, and that allows ultimately collecting them into a single Disposableinstance when running the resulting Stream

Those ScheduledTask instances that are returned by the scheduler implement the same Disposable interface, so they also get kept in that Disposable graph as well

Micah

@micahscopes

Mmm, I see. So when I'm using mergeArray to collect all my little streams into one giant stream, I'm also collecting their Disposables into a single Disposable instance

Something I've thought about doing instead of that mega merged stream is using higher order streams that get passed to tap(stream => runEffects(stream, scheduler))

Seems like that should solve my issue if it means dispose is getting called more regularly!

Micah

@micahscopes

So for example if I wanted to loop a little melody over 12 bars I might do something like:

periodic(12*BAR)
|> map(() => melody$ |> until(12*BAR)
|> tap(stream => runEffects(stream, scheduler))

(currently I'm using switchLatest to get that effect, but I'm seeing now how that's probably deferring dispose indefinitely!)

@TylorS wow... okay thank you! I think I'm starting to understand how this Disposable stuff works in practice

Micah

@micahscopes

K so I have verified that this ^ is saving me at least 4x on accumulated heap allocations over ~10 hours of simulated music play!

That's huge

matt penrice

@elmpp

Hi there. I'd love someone's opinion on this mostjs question - cujojs/most#551

jarkkoskyttala

@jarkkoskyttala

Hi,

We have been struggling with some strange behaviour regarding combineArray and sample. Could you take a look at this test and let me know if you have any idea why result$ is not triggering? pipe function is from fp-ts. The test passes if I remove the hold from s2$

it('should trigger with result', () => {
        const scheduler = defaultScheduler();
        const s$ = at(100, 'test');

        const s2$ = pipe(
          map(() => 'test-2', s$),
          hold
        );

        const result$ = sample(
          combineArray(
            (x, y) => ({ x, y, }),
            [s2$, s$]
          ), s2$);

        const expected = {
          x: 'test-2',
          y: 'test',
        };

        const test_X$ = tap((x) => {
            expect(x).to.eql(expected);
          }, result$);

        runEffects(test_X$, scheduler);
      })

jarkkoskyttala

@jarkkoskyttala

I can also make the test pass by switching s2$ and s$ locations in the combineArray array, so it will take [s$, s2$].

Micah

@micahscopes

nothing like a couple of well placed multicast calls to fix all the bugs with side effects happening multiple times

Galileo Sanchez

@galileopy

Intuitively I sense that event streams should lead to a natural pattern for rate limiting, but I haven't give this too much though yet. Has anybody ever tried this?

Micah

@micahscopes

@galileopy @most/core has both debounce and throttle stream modifiers, have you tried these out?

https://mostcore.readthedocs.io/en/latest/api.html#rate-limiting

sourcevault

@sourcevault

.chain ($) -> $

is the same as :

.join()

sourcevault

@sourcevault

have anybody created a UI libary using most yet ?

sourcevault

@sourcevault

stream$       <--      has a (outer) dispose function

stream$
.takeWhile    <---     can end here, and (outer) dispose called here.
.continueWith <--- [1] logic to deal with early termination
.chain        <---     logic here ...
.takeWhile    <--- [2] ... can be used to end stream$ and call (outer) dispose.

would want it to look like this

stream$       <--      has a (outer) dispose function

stream$
.takeWhile    <---     can end here, and (outer) dispose called here.
.chain        <---     logic here ...
.takeWhile    <--- [2] ... can be used to end stream$ and call (outer) dispose.

.subscribe (complete:....) <--- would want logic in [1] to go here, but [2] gets in the way

sourcevault

@sourcevault

there are not a lot of resources on structuring most apps.

one thing that is important to do is separate the logic form the sideEffects

but what if the sideEffects result in change in logic ?

real world applications is not so clean

if I put the logic in [2] to the subscribe function, it won't be able to call the outer dispose in stream$

sourcevault

@sourcevault

would it be possible to abstract out stream processing with Symbols ? and handle things in the driver / sideEffect / 'observe' region ?

for example takeWhile returns a symbol tw1 and the second takeWhile returns tw2.

sourcevault

@sourcevault

ok, i think i found the answer to my question

somethings it helps to try to write down the problem !

Where communities thrive

People

Repo info

Activity