Concurrent Search Bug

[whyfate]

Describe the bug
Such as the title

To Reproduce
it's my test code

static void ConcurrentSearchTest()
{
    var client = new FhirClient("https://spark.incendi.no/fhir");
    // This request will return two resource: Encounter and Patient
    var hasResourceURL = "https://spark.incendi.no/fhir/Encounter?_id=2&_include=Encounter:subject";
    // This request has no resource returned
    var noResourceURL = "https://spark.incendi.no/fhir/Encounter?_id=22222222&_include=Encounter:subject";
    for (int i = 0; i < 500; i++)
    {
        var j = i;
        System.Threading.Tasks.Task.Run(async () =>
        {
            if (j % 2 == 0)
            {
                var bundle = (Bundle)await client.GetAsync(hasResourceURL);
                if (bundle.Entry.Count != 2)
                {
                    Console.WriteLine($"{j}---search has resource error");
                }
                else
                {
                    Console.WriteLine($"{j}---succeed");
                }
            }
            else
            {
                var bundle = (Bundle)await client.GetAsync(noResourceURL);
                if (bundle.Entry.Count > 0)
                {
                    Console.WriteLine($"{j}---search no resource error");
                }
                else
                {
                    Console.WriteLine($"{j}---success");
                }
            }
        });
    }
}

Expected behavior
all print {j}---success

Screenshots

[whyfate]

I debugged the source code and found that the cause of the error was the concurrency problem in the SnapshotPaginationProvider

spark/src/Spark.Engine/Service/FhirServiceExtensions/SnapshotPaginationProvider.cs

Lines 15 to 35 in 6ac316e

	public class SnapshotPaginationProvider : ISnapshotPaginationProvider, ISnapshotPagination
	{
	private IFhirStore _fhirStore;
	private readonly ITransfer _transfer;
	private readonly ILocalhost _localhost;
	private readonly ISnapshotPaginationCalculator _snapshotPaginationCalculator;
	private Snapshot _snapshot;
	public SnapshotPaginationProvider(IFhirStore fhirStore, ITransfer transfer, ILocalhost localhost, ISnapshotPaginationCalculator snapshotPaginationCalculator)
	{
	_fhirStore = fhirStore;
	_transfer = transfer;
	_localhost = localhost;
	_snapshotPaginationCalculator = snapshotPaginationCalculator;
	}
	public ISnapshotPagination StartPagination(Snapshot snapshot)
	{
	_snapshot = snapshot;
	return this;
	}

the Call chain is:
AsyncFhirService -> PagingService -> SnapshotPaginationProvider
Although SnapshotPaginationProvider is injected Transient, because AsyncFhirService is a Singleton, SnapshotPaginationProvider is also a Singleton.

spark/src/Spark.Engine/Service/AsyncFhirService.cs

Lines 370 to 390 in 212d071

	private async Task<FhirResponse> CreateSnapshotResponseAsync(Snapshot snapshot, int pageIndex = 0)
	{
	var pagingExtension = FindExtension<IPagingService>();
	if (pagingExtension == null)
	{
	var bundle = new Bundle
	{
	Type = snapshot.Type,
	Total = snapshot.Count
	};
	var resourceStorage = FindExtension<IResourceStorageService>();
	bundle.Append(await resourceStorage.GetAsync(snapshot.Keys).ConfigureAwait(false));
	return _responseFactory.GetFhirResponse(bundle);
	}
	else
	{
	var pagination = await pagingExtension.StartPaginationAsync(snapshot).ConfigureAwait(false);
	var bundle = await pagination.GetPageAsync(pageIndex).ConfigureAwait(false);
	return _responseFactory.GetFhirResponse(bundle);
	}
	}

At present, I have changed AsyncFhirService to Transient to solve the problem temporarily, but I don't think this is the best way.Maybe need to refactor FhirService.

[kennethmyhra]

It was singleton in good old classic Spark (.NET FW), and probably therefore it is singleton in the .NET Core/Standard version of Spark. Actually I do not see a good reason for AsyncFhirService to be a singleton and I think we should consider making it transient instead.

[whyfate]

But creating an AsyncFhirService is time consuming.

spark/src/Spark.Engine/Service/AsyncFhirService.cs

Lines 28 to 43 in 212d071

	public AsyncFhirService(
	IFhirServiceExtension[] extensions,
	IFhirResponseFactory responseFactory, // TODO: can we remove this dependency?
	ITransfer transfer,
	ICompositeServiceListener serviceListener = null) // TODO: can we remove this dependency? - CCR
	{
	_responseFactory = responseFactory ?? throw new ArgumentNullException(nameof(responseFactory));
	_transfer = transfer ?? throw new ArgumentNullException(nameof(transfer));
	_serviceListener = serviceListener ?? throw new ArgumentNullException(nameof(serviceListener));
	foreach (var serviceExtension in extensions)
	{
	AddExtension(serviceExtension);
	}
	}

This has a effect on Performance.

[kennethmyhra]

Yes, that is a concern. I have not measured the effect it would have on performance, but yes that would be the concern.

[one0410]

Hi,
I have the same problem.
If I query API at almost the same time, sometimes I got weird response different from what I expected.
So that I couldn't put this project into a production environment.

For example, I have 20 Device resource in DB.
If I query identifier 000000000001, I expect a result with same identifier.
Please see the following code, sometimes I got a random result.
Please help.

  testFhir(): void {
    for (let i = 0; i <20; i++) {
      this.http.get('https://esp32fhir.azurewebsites.net/fhir/Device?identifier=' + i.toString().padStart(12, '0'))
      .toPromise()
      .then((res: any) => {
        console.log(i, res.entry[0].resource);
      });
    }
  }

Andy

[one0410]

Hi @whyfate,

Could you tell me how you changed AsyncFhirService to Transient to solve the problem temporarily ?

Andy

[whyfate]

@one0410 I modified the source code.

spark/src/Spark.Engine/Extensions/NetCore/IServiceCollectionExtensions.cs

Line 85 in e130c88

services.TryAddSingleton<IAsyncFhirService, AsyncFhirService>();

services.TryAddTransient<IAsyncFhirService, AsyncFhirService>();

[kennethmyhra]

@whyfate how did that affect the performance?

[one0410]

@whyfate You are right, the performance is terrible.
@kennethmyhra Is this what you want ?

singleton:

transient:

[kennethmyhra]

@one0410, absolutely not. I haven't yet had time to look into the root cause of this. If any of you have time to do that it would be much appreciated. Anyways, I'll put it into the current sprint for the next release

[whyfate]

I have an immature suggestion, which is to pass snapshot into the parameter of this method.

spark/src/Spark.Engine/Service/FhirServiceExtensions/ISnapshotPagination.cs

Line 19 in e130c88

Bundle GetPage(int? index = null, Action<Entry> transformElement = null);

Bundle GetPage(Snapshot snapshot,int? index = null, Action<Entry> transformElement = null); 

Replacing SnapshotPaginationProvider internal variables _snapshot with parameters

spark/src/Spark.Engine/Service/FhirServiceExtensions/SnapshotPaginationProvider.cs

Lines 29 to 58 in e130c88

	private Snapshot _snapshot;
	public SnapshotPaginationProvider(IFhirStore fhirStore, ITransfer transfer, ILocalhost localhost, ISnapshotPaginationCalculator snapshotPaginationCalculator)
	{
	_fhirStore = fhirStore;
	_transfer = transfer;
	_localhost = localhost;
	_snapshotPaginationCalculator = snapshotPaginationCalculator;
	}
	public ISnapshotPagination StartPagination(Snapshot snapshot)
	{
	_snapshot = snapshot;
	return this;
	}
	public Bundle GetPage(int? index = null, Action<Entry> transformElement = null)
	{
	if (_snapshot == null)
	throw Error.NotFound("There is no paged snapshot");
	if (!_snapshot.InRange(index ?? 0))
	{
	throw Error.NotFound(
	"The specified index lies outside the range of available results ({0}) in snapshot {1}",
	_snapshot.Keys.Count(), _snapshot.Id);
	}
	return CreateBundle(index);
	}

But emmm...bad code warning!!!

[kennethmyhra]

I've done some research and can confirm that I am seeing the error as well. I tried putting a SemaphoreSlim around the code in AsyncFhirService.SearchAsync (as a hot fix), it helps, but it slows it down since it forces the requests to run synchronously one-by-one.

I need to try and reproduce this with FhirServiceService to see if we had this problem before we added AsyncFhirService

[whyfate]

This problem is Snapshot design, which has existed for a long time, not async.

[kennethmyhra]

This is fixed in both STU3 and R4. If you are compiling from source, please verify by pulling. If you are using the NuGet packages I will create a new release today
STU3 - https://www.nuget.org/packages/Spark.Engine.STU3
R4 - https://www.nuget.org/packages/Spark.Engine.R4

Summarized I split ISnapshotPaginationProvider and ISnapshotPagination into two implementations making sure as @whyfate pointed out that Snapshot is not affected by the singleton behavior of ISnapshotPaginationProvider. There is a more lengthy explanation in the commit/PR

[one0410]

Thank you very much. This does help me a lot. You are really amazing.

[whyfate]

@kennethmyhra Hi,I saw your implementation,I don't think it's good to directly a new SnapshotPaginationService.
How about this?

services.AddTransient<ISnapshotPaginationProvider, SnapshotPaginationProvider>();

public class PagingService : IPagingService
{
    private readonly ISnapshotStore _snapshotstore;
    private readonly IServiceProvider _serviceProvider;

    public PagingService(ISnapshotStore snapshotstore, IServiceProvider serviceProvider)
    {
        _snapshotstore = snapshotstore;
        _serviceProvider = serviceProvider;
    }

    public ISnapshotPagination StartPagination(Snapshot snapshot)
    {
        return Task.Run(() => StartPaginationAsync(snapshot)).GetAwaiter().GetResult();
    }

    public ISnapshotPagination StartPagination(string snapshotkey)
    {
        return Task.Run(() => StartPaginationAsync(snapshotkey)).GetAwaiter().GetResult();
    }

    public async Task<ISnapshotPagination> StartPaginationAsync(Snapshot snapshot)
    {
        if (_snapshotstore != null)
        {
            await _snapshotstore.AddSnapshotAsync(snapshot).ConfigureAwait(false);
        }
        else
        {
            snapshot.Id = null;
        }

        var snapshotPaginationProvider = _serviceProvider.GetRequiredService<ISnapshotPaginationProvider>();
        return snapshotPaginationProvider.StartPagination(snapshot);
    }
    public async Task<ISnapshotPagination> StartPaginationAsync(string snapshotkey)
    {
        if (_snapshotstore == null)
        {
            throw new NotSupportedException("Stateful pagination is not currently supported.");
        }

        var snapshotPaginationProvider = _serviceProvider.GetRequiredService<ISnapshotPaginationProvider>();
        return snapshotPaginationProvider.StartPagination(await _snapshotstore.GetSnapshotAsync(snapshotkey).ConfigureAwait(false));
    }
}

[kennethmyhra]

@whyfate the plan is to do this in the next version. While I agree with you - right now the only downside is that you can't make your own implementation of the interface ISnapshotPagination which was the interface touched in this iteration

I also see now that you are referring to some other interface (IPagingService) and implementation (PagingService) which was not touched in this iteration. The whole infrastructure around snapshot pagination seems messy and unfinished which I also noticed when I started touching this part and therefore did not put too much effort into the infrastructure part at this point

[whyfate]

I see,thanks.