Resolved Performance issues with consumer method and how to resolve it

[timotech]

Hi guys,
Based on my last thread: Await blocking other consumers in a for loop
In which @Skydiver recommend getting more knowledge in concurrency in c#, well I've been able to resolve most of the issues i have, but I have one that is troubling me.
I have multiple producers and they are working fine. I'm using one consumer to get data from the producers.
The challenge I have is that when I use the code below, the program hangs

//multiple producers
            for (int i = 0; i < pages.Count; i++)
            {
                realtimeSources[i] = new BlockingCollection<AudioSamples>();
                //producers
                Listener listener = new(cbBoxes[i].SelectedIndex, sampleRate, realtimeSources[i]);
                listeners.Add(listener);               
            }

            //consumer
            BlockingCollection<AudioSamples> audioSamples = new BlockingCollection<AudioSamples>();

            _ = BlockingCollection<AudioSamples>.TryTakeFromAny(realtimeSources, out AudioSamples item);
            audioSamples.Add(item);
            _ = GetBestMatchForStream(realtimeSources[0], modelService, tokenSource.Token, "Wazobia");

but when I use only one item e.g producer[0], I get responses, but responses from only one producer

//multiple producers
            for (int i = 0; i < pages.Count; i++)
            {
                realtimeSources[i] = new BlockingCollection<AudioSamples>();
                //producers
                Listener listener = new(cbBoxes[i].SelectedIndex, sampleRate, realtimeSources[i]);
                listeners.Add(listener);               
            }

            //consumer (consuming only one producer realtimeSources[0])
            _ = GetBestMatchForStream(realtimeSources[0], modelService, tokenSource.Token, "Wazobia");

Could it be that i'm implementing the design wrongly or my quad core processor cannot handle it?.
The producers are BlockingCollection<AudioSamples> i.e realtimeSources, 8 in number
Please advice.
Thanks

 

[Skydiver]

In your first code chunk, why are you not passing on item as the first parameter to GetBestMatchForStream()?

What does the code inside GetBestMatchForStream() look like?

 

[timotech]

In your first code chunk, why are you not passing on item as the first parameter to GetBestMatchForStream()?

What does the code inside GetBestMatchForStream() look like?

Hi @Skydiver, thanks for always responding.
I will try that suggestion, however, please check the GetBestMatchForStream Code below. Thanks

public async Task<double> GetBestMatchForStream(BlockingCollection<AudioSamples> realtimeSource, IModelService modelService, CancellationToken token, string tabName)
        {
            double seconds = await QueryCommandBuilder.Instance
                .BuildRealtimeQueryCommand()
                .From(new BlockingRealtimeCollection<AudioSamples>(realtimeSource))
                .WithRealtimeQueryConfig(config =>
                {
                    // match only those entries got at least 5 seconds of query match
                    config.ResultEntryFilter = new TrackMatchLengthEntryFilter(5d);
                    //config.QueryConfiguration.Audio.MaxTracksToReturn = 1; //May remove this after observation

                    // provide a success callback that will be invoked for matches that pass the result entry filter
                    config.SuccessCallback = result =>
                    {
                        foreach (var entry in result.ResultEntries)
                        {
                            Debug.WriteLine($"Successfully matched {entry.TrackId}");
                            Debug.WriteLine($"Successfully matched {entry.Audio.Track.Title} at {DateTime.Now} on {tabName}");
                        }
                    };

                    config.DidNotPassFilterCallback = (queryResult) =>
                    {
                        foreach (var result in queryResult.ResultEntries)
                        {
                            //Debug.WriteLine($"Did not pass filter {result.TrackId}");
                            Debug.WriteLine($"Did not pass filter {result.Audio.Track.Title} at {DateTime.Now}");
                        }
                    };

                    return config;
                })
                .UsingServices(modelService)
                .Query(token);

            //Debug.WriteLine($"Realtime query stopped. Issued {seconds} seconds of query.");
            return seconds;
        }

its from this library: Realtime Query Command · AddictedCS/soundfingerprinting Wiki
I'm just modifying it.
Edit: I forgot to mention, when I pass in item directly, it brings this error:

Cannot convert from 'Soundfingerprinting.Audio.AudioSamples' to 'System.Collections.Concurrent.BlockingCollection<Soundfingerprinting.Audio.AudioSamples>'

That was why I had to add it to a new instance
Thanks

 

[timotech]

Ok, so after going through a course on Asynchronous programming in C#, I was introduced to a life saver called
Task.Run().
It removes processing from the UI thread and creates a separate task for the process making your UI to be free from any form of freezing. And it solved my problem.
All I needed to do was call GetBestMatch method inside it.
Some thing like this:

Task.Run(() =>
            {
                BlockingCollection<AudioSamples> audioSamples = new BlockingCollection<AudioSamples>();

                _ = BlockingCollection<AudioSamples>.TakeFromAny(realtimeSources.ToArray(), out AudioSamples item);
                audioSamples.Add(item);
                _ = GetBestMatchForStream(audioSamples, modelService, tokenSource.Token, "Wazobia");
            });

Thanks to @Skydiver for pushing me further