[DData] Adding large number of values to ORMultiValueDictionary using SetItems takes too long

[Arkatufus]

Version Information
Version of Akka.NET?
Which Akka.NET Modules? DData

Describe the bug
Using distributed data ORMultiValueDictionary and adding a key say with 20000 values calling SetItems takes upto 2-3 mins to complete.

To Reproduce
Steps to reproduce the behavior:

r.Replicator.Tell(Dsl.Update(
    registryKey, 
    ORMultiValueDictionary<TKey, TVal>.EmptyWithValueDeltas, 
    WriteLocal.Instance,
    s => s.SetItems(c.SelfUniqueAddress, d.Key, d.Value)));

Links to working reproductions on Github / Gitlab are very much appreciated

Expected behavior
It should not take long to add 20,000 values

Actual behavior
Takes upto 2-3 mins to complete.

[Aaronontheweb]

So is the issue in this case that the number of items inside the SetItems call is 20000? Or is it that this method is being called 20000 times?

[huzaifak]

Number of items in the SetItems call is 20,000. Below is a unit test I used in ReplicatorSpecs class.

  `[Fact]
    public async Task ORMultiValueDictionary_WithValueDeltas_LargeDataSet()
    {
        await InitCluster();

        var changedProbe2 = CreateTestProbe(_sys2);
        _replicator2.Tell(Dsl.Subscribe(_keyJ, changedProbe2.Ref));

        var changedProbe3 = CreateTestProbe(_sys3);
        _replicator3.Tell(Dsl.Subscribe(_keyJ, changedProbe3.Ref));

        // Scenario 1 - add 1 entry with multiple values to all nodes
        var keyA = "A";
        var entryA = ImmutableHashSet<string>.Empty;
        for (int i =0; i < 20000; i++)
        {
            entryA = entryA.Add(Convert.ToString(i));
        }
        Stopwatch stopwatch = new Stopwatch();
        
        await AwaitAssertAsync(async () => {
            var m1 = await _replicator1.Ask<UpdateSuccess>(Dsl.Update(
                _keyJ,
                ORMultiValueDictionary<string, string>.EmptyWithValueDeltas,
                new WriteMajority(_timeOut),
                s => 
                {
                    stopwatch.Start();
                    var snew = s.SetItems(Cluster.Cluster.Get(_sys1), keyA, entryA);
                    stopwatch.Stop();
                    return snew;
                }));
           
            var node2EntriesA = changedProbe2.ExpectMsg<Changed>(g => Equals(g.Key, _keyJ)).Get(_keyJ).Entries;
            node2EntriesA[keyA].Should().BeEquivalentTo(entryA);

            var node3EntriesA = changedProbe3.ExpectMsg<Changed>(g => Equals(g.Key, _keyJ)).Get(_keyJ).Entries;
            node3EntriesA[keyA].Should().BeEquivalentTo(entryA);
        });

    }`

[Arkatufus]

@huzaifak thanks for the reproduction spec, we'll look into it.

[Arkatufus]

@huzaifak this should be fixed in #5686