Is it recomended to publishOn after every database call?

[kurt-jadson]

Hello, I tried to ask this on spring-data-r2dbc project, but since there is no discussion tab opened there I cannot get find a good place to ask this, hope that someone could clarify it to me because I do not find good answers in stackoverflow for that.

So, here is my Controller endpoint

  @GetMapping
  public Mono<Page<EmployeeResponse>> findAll(
      @Valid PageFilter pageFilter,
      @Valid EmployeeCriteria employeeCriteria,
      final ServerWebExchange exchange) {
    return Mono.just("This line is here only to print first thread entrance.")
        .doOnNext(x -> System.out.println("Starting in thread: " + Thread.currentThread().getName()))
        .flatMap(s ->
            service
                .findAll(pageFilter.toPageable(), mapper.toEntity(employeeCriteria))
                .map(page -> page.map(mapper::toResponse))
                .doOnNext(x -> System.out.println("toResponse on " + Thread.currentThread().getName()))
                .doOnSuccess(employee -> exchange.getResponse().setStatusCode(HttpStatus.OK)));
  }

and here is the Service class method:

  public Mono<Page<Employee>> findAll(Pageable pageable, Employee employee) {
    return Mono.deferContextual(
        ctx -> {
          final UUID companyId = ctx.get(COMPANY_ID);
          return repository
              .findAll(companyId, pageable, employee)
              // nothing test #1
              //.publishOn(Schedulers.boundedElastic()) // Test #2
              .doOnNext(x -> System.out.println("findAll on " + Thread.currentThread().getName()))
              .collectList()
              .zipWith(repository.count(companyId, pageable, employee)
                  // nothing test #1
                  //.publishOn(Schedulers.boundedElastic()) // Test #2
                  .doOnNext(x -> System.out.println("count on " + Thread.currentThread().getName())))
              .map(
                  entityTuples ->
                      new PageImpl<>(entityTuples.getT1(), pageable, entityTuples.getT2()))
              .doOnNext(x -> System.out.println("Map on " + Thread.currentThread().getName()));
        });

In test #1 I just ran it without any publishing to other Scheduler. So the result is:

Starting in thread: lettuce-nioEventLoop-5-1
2021-06-17 08:48:40 [reactor-tcp-nio-1] DEBUG DefaultDatabaseClient:329 - Executing SQL statement [SELECT employee.* FROM employee WHERE employee.company_id = $1 ORDER BY code ASC, name ASC LIMIT 10]
findAll on reactor-tcp-nio-1
findAll on reactor-tcp-nio-1
2021-06-17 08:48:41 [reactor-tcp-nio-1] DEBUG DefaultDatabaseClient:329 - Executing SQL statement [SELECT COUNT(employee.employee_id) FROM employee WHERE employee.company_id = $1]
count on reactor-tcp-nio-1
Map on reactor-tcp-nio-1
toResponse on reactor-tcp-nio-1

So, what I am understanding is that r2dbc driver is switching to a reactor-tcp-nio thread to execute the SQL statement. Once it gets result back, then all the consecutive operations such as map gets executed in the same thread.

In test #2 I intencionally published the results of the query to a boundedElastic scheduler. Now you can see:

Starting in thread: lettuce-nioEventLoop-5-1
2021-06-17 08:55:58 [reactor-tcp-nio-1] DEBUG DefaultDatabaseClient:329 - Executing SQL statement [SELECT employee.* FROM employee WHERE employee.company_id = $1 ORDER BY code ASC, name ASC LIMIT 10]
findAll on boundedElastic-3
findAll on boundedElastic-3
2021-06-17 08:55:59 [reactor-tcp-nio-1] DEBUG DefaultDatabaseClient:329 - Executing SQL statement [SELECT COUNT(employee.employee_id) FROM employee WHERE employee.company_id = $1]
count on boundedElastic-3
Map on boundedElastic-3
toResponse on boundedElastic-3

R2DBC executes the SQL statement in a reactor-tcp-nio thread, but the map and other operations takes place in boundedElastic.

So, my question is:
Is it a recomendation that we always publishOn the results of the database call to another scheduler?
Here is a sample that I found in my researchs but using JDBC instead.

[mp911de]

What's the motivation for switching threads?
In general, the R2DBC driver comes with its own threading infrastructure and since thread affinity in reactive programming is less of a concern than in imperative programming, I'm wondering about the use case.

[kurt-jadson]

Maybe it could be a wrong interpretation on my side, but since I see reactor-tcp-nio is bounded to the number of CPUs on my machine I could imagine that consecutive operations would hold on this thread until its complete, so that means r2dbc would be awaiting for the next available reactor-tcp-nio thread to perform other SQL statement, that could decrease some throughput on multi-requests enviroments. Is it correct?

[mp911de]

There's indeed a relationship between thread workload and connection throughput. Any downstream work (JSON deserialization, encryption, computation) keeps the I/O thread busy and connections sharing the same event loop thread are hindered in progressing. This is a specific scenario where you need to pay special attention. On the other side, typical arrangements make use of more than one connection. If you offload work to another thread eventually your CPU is saturated with work and switching between threads actually accumulates overhead which isn't ideal either.

The ultimate answer is running proper benchmarks to find a balance between connection count, I/O thread count, and the use of occasional thread switching.

Let me know whether that helps.