Blocking event loop during database operation

E. Ulrich Kriegel

unread,

Jan 30, 2017, 6:08:04 AM1/30/17

to vert.x

Hi there,

we have to read a bunch of record from an input database, process these records and write the results to an output database, which is empty and has no indices at the moment of writing. The reactive paradigm is used.

However, for the processing, java crypto libraries are used, i.e. the event loop could be blocked. Therefore, in our first attempt, these operations has been enclosed in a vertx.executeBlockingObservable - but we got a lot of event-loop-blocked warnings. After removing of the vertx.executeBlockingObservable, the number of warnings in a small use case with 20000 input records was reduced. We tried using onBackPressureBuffer - without success. The only solution we found is to place a delay of 10 ms in the pipe. With that we were able to process 2million records without getting a event-loop-blocking warning.

Is there a adequate solution - placing a delay is IMHO a little bit susceptive.

Here comes our code

		JDBCClient inJdbcClient = JDBCClient.createNonShared(vertx, inDbConfig);
		// abstract from database specific peculiarities
		inDbHelper = new DatabaseHelper4Postgres();
		// drop indices on optbut database
		Observable.using(DisposableVertxSQLConnection::new, f -> f.create(jdbcClient), f -> f.dispose()).flatMap(conn-> outDbHelper.dropIndices(conn)) 
		.flatMap(dummy ->Observable.using(DisposableVertxSQLConnection::new, f -> f.create(inJdbcClient), f -> f.dispose()))
		// read records from input database
		.flatMap(inConn-> inDbHelper.readDatabase(inConn, "inTable"))
		.map(rs -> rs.getResults())
		.delay(10, TimeUnit.MILLISECONDS)
		.flatMapIterable(li -> li)
		// transform input records
		.flatMap(input -> inputToPreparedInsertQueryObservable(input))
		// collect chunks to use batch datbase operations
		.buffer(1000)
		.flatMap(params -> {
			Observable<Integer> resultObservable = Observable.using(DisposableVertxSQLConnection::new, f -> f.create(jdbcClient), f -> f.dispose())
					.flatMap(conn-> {
						Observable<List<Integer>> li =  outDbHelper.batchInsertInput(conn, (List<JsonArray>)(Object)params);
						return li;}).flatMapIterable(i->i);
			// merge batch output with input record id's to generate error messages
			return Observable.zip(resultObservable, Observable.from(params), (first, second) -> Observable.just(new Tuple(first, ((JsonArray) second).getString(0)))).flatMap(x->x).collect(()->new ArrayList<Tuple>(), (data, item)-> data.add(item));
			})
		.map(li ->{
			// count batch content
			Observable<Integer> totalCountObservable = Observable.from(li).count(); 
			// filter not written records
			Observable<JsonArray> errorListObservable = Observable.from(li).filter(tuple -> tuple.getFirst() == 0).collect(()-> new JsonArray(), (data, item)->data.add(item.getSecond()));
			return Observable.zip(totalCountObservable, errorListObservable, (first, second) -> {
				int s = second.size();
				if(s > 0){
					logger.warn(String.format("Not inserted ids: %s", second.toString()));
				}
				return first - s;});			
		})
		.flatMap(l -> l)
		.collect(()-> { 
			long [] counter = new long [1];
			counter[0] = 0;
			return counter;
		}, (counter, item)->{
			counter[0] += item; })
		// finally create indices on output database
		.map(counter ->{
			Observable<SQLConnection> indexObservable = jdbcClient.getConnectionObservable()
					.flatMap(conn-> outDbHelper.createIndices(conn));
			return Observable.zip(Observable.just(counter[0]), indexObservable, (count, conn)->{conn.close(); return count;});
			
		})
		.flatMap(i -> i)
		.subscribe(insertedRecords->{

Thanks in advance

--Ulrich

Thomas SEGISMONT

unread,

Jan 30, 2017, 12:42:07 PM1/30/17

to ve...@googlegroups.com

The delay operator, as you used it, operates on the computation scheduler. That's why you no longer see the warnings. You shouldn't block the computation scheduler threads either BTW.

Can you simplify your code sample to point to the part where you use blocking/heavy computational APIs? And your attempt to solve it with executeBlocking?

--
You received this message because you are subscribed to the Google Groups "vert.x" group.
To unsubscribe from this group and stop receiving emails from it, send an email to vertx+unsubscribe@googlegroups.com.
Visit this group at https://groups.google.com/group/vertx.
To view this discussion on the web, visit https://groups.google.com/d/msgid/vertx/833c7afd-ade6-44d6-a6a5-8784724c6d72%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

E. Ulrich Kriegel

unread,

Jan 31, 2017, 3:46:54 AM1/31/17

to vert.x

Hi Thomes, here comes the code

                JDBCClient inJdbcClient = JDBCClient.createNonShared(vertx, inDbConfig);

                // abstract from database specific peculiarities

inDbHelper = new DatabaseHelper4Postgres(testFlag);


                // drop indices on optbut database

                Observable.using(DisposableVertxSQLConnection::new, f -> f.create(jdbcClient), f -> f.dispose())

                .flatMap(dummy ->Observable.using(DisposableVertxSQLConnection::new, f -> f.create(inJdbcClient), f -> f.dispose()))

                // read records from input database

                .flatMap(inConn-> inDbHelper.readPersonFromPlainDatabase(inConn, inDbConfig.getString("inTable")))

                .map(rs -> rs.getResults())


                .flatMapIterable(li -> li)

                // transform input records

                .flatMap(input -> vertx.executeBlockingObservable(_future ->{  //EXECUTEBLOCKING

                        _future.complete(inputPreparedInsertQuery(input));


                }))

                // collect chunks to use batch datbase operations

                .buffer(Constants.MAX_BATCH_SIZE)

                .flatMap(params ->

                        Observable.using(DisposableVertxSQLConnection::new, f -> f.create(jdbcClient), f -> f.dispose()).flatMap(conn-> outDbHelper.batchInsert(conn,  (List<JsonArray>(Object)params)))

                .map(li -> li.size())


                .collect(()-> { 

                        long [] counter = new long [1];

                        counter[0] = 0;

                        return counter;

                }, (counter, item)->{

                        counter[0] += item; })

                .subscribe(insertedRecords->{
....

Thomas SEGISMONT

unread,

Feb 1, 2017, 4:23:54 PM2/1/17

to ve...@googlegroups.com

Your usage of executeBlocking looks, good to you have any stacktrace which could help to determine where the thread is blocked?

--
You received this message because you are subscribed to the Google Groups "vert.x" group.
To unsubscribe from this group and stop receiving emails from it, send an email to vertx+unsubscribe@googlegroups.com.
Visit this group at https://groups.google.com/group/vertx.

To view this discussion on the web, visit https://groups.google.com/d/msgid/vertx/5a2e727a-aea1-4a9a-adc4-857e859cba9a%40googlegroups.com.

Ulrich Kriegel

unread,

Feb 2, 2017, 8:28:55 AM2/2/17

to ve...@googlegroups.com

Hi Thomas,

Am 01.02.2017 um 22:23 schrieb Thomas SEGISMONT <tsegi...@gmail.com>:

Your usage of executeBlocking looks, good to you have any stacktrace which could help to determine where the thread is blocked?

How can i do that. Vertx logs only, that thread is blocked. Is ther a hook or something else i can use?

Regards

Ulrich Kriegel

ulrich....@gmail.com

Thomas SEGISMONT

unread,

Feb 2, 2017, 10:10:53 AM2/2/17

to ve...@googlegroups.com

After some time, Vert.x prints a stack trace. You can change this time to be much shorter than the default (5 seconds):

vertxOptions.setWarningExceptionTime(xxx)

The unit in nanoseconds.

--
You received this message because you are subscribed to the Google Groups "vert.x" group.
To unsubscribe from this group and stop receiving emails from it, send an email to vertx+unsubscribe@googlegroups.com.
Visit this group at https://groups.google.com/group/vertx.

To view this discussion on the web, visit https://groups.google.com/d/msgid/vertx/15EA0A1D-ABDE-47A6-81F6-25457527DA78%40gmail.com.

Jochen Mader

unread,

Feb 2, 2017, 11:10:45 AM2/2/17

to ve...@googlegroups.com

Send a 'kill 3' to the process running your application. That will cause the JVM to emit a threaddump which should show you where the thread gets stuck.

If you have trouble reading them put them in a Gist and share the link with us.

To view this discussion on the web, visit https://groups.google.com/d/msgid/vertx/CACiEr_T1qFJQ%3DZY%3DiTiYZOOqq-kc7nKAKfbanOjVT6z5C%2Bn_%2BA%40mail.gmail.com.

For more options, visit https://groups.google.com/d/optout.

--

Jochen Mader | Lead IT Consultant

E. Ulrich Kriegel

unread,

Feb 3, 2017, 5:39:21 AM2/3/17

to vert.x

I first with 'kill -3' as proposed by Jochen, but i got 3 times different dumps. then i found out, that my logging configuration prevented me to get the stack traces.

Here they are

MLog initialization issue: slf4j found no binding or threatened to use its (dangerously silent) NOPLogger. We consider the slf4j library not found.

MLog clients using java 1.4+ standard logging.

Initializing c3p0-0.9.5.2 [built 08-December-2015 22:06:04 -0800; debug? true; trace: 10]

{"subject":"de.fraunhofer.fokus.adep.BatchMgmtVerticle","status":"success","action":"initialize"}

Succeeded in deploying verticle

Initializing c3p0 pool... com.mchange.v2.c3p0.ComboPooledDataSource [ acquireIncrement -> 3, acquireRetryAttempts -> 30, acquireRetryDelay -> 1000, autoCommitOnClose -> false, automaticTestTable -> null, breakAfterAcquireFailure -> false, checkoutTimeout -> 0, connectionCustomizerClassName -> null, connectionTesterClassName -> com.mchange.v2.c3p0.impl.DefaultConnectionTester, contextClassLoaderSource -> caller, dataSourceName -> 2xmt4d9mzc9er6ko2cn|310ecab0, debugUnreturnedConnectionStackTraces -> false, description -> null, driverClass -> org.postgresql.Driver, extensions -> {}, factoryClassLocation -> null, forceIgnoreUnresolvedTransactions -> false, forceSynchronousCheckins -> false, forceUseNamedDriverClass -> false, identityToken -> 2xmt4d9mzc9er6ko2cn|310ecab0, idleConnectionTestPeriod -> 0, initialPoolSize -> 3, jdbcUrl -> jdbc:postgresql://localhost:5432/adep-fr-new, maxAdministrativeTaskTime -> 0, maxConnectionAge -> 0, maxIdleTime -> 0, maxIdleTimeExcessConnections -> 0, maxPoolSize -> 1000, maxStatements -> 0, maxStatementsPerConnection -> 0, minPoolSize -> 3, numHelperThreads -> 3, preferredTestQuery -> null, privilegeSpawnedThreads -> false, properties -> {user=******, password=******}, propertyCycle -> 0, statementCacheNumDeferredCloseThreads -> 0, testConnectionOnCheckin -> false, testConnectionOnCheckout -> false, unreturnedConnectionTimeout -> 0, userOverrides -> {}, usesTraditionalReflectiveProxies -> false ]

Initializing c3p0 pool... com.mchange.v2.c3p0.ComboPooledDataSource [ acquireIncrement -> 3, acquireRetryAttempts -> 30, acquireRetryDelay -> 1000, autoCommitOnClose -> false, automaticTestTable -> null, breakAfterAcquireFailure -> false, checkoutTimeout -> 0, connectionCustomizerClassName -> null, connectionTesterClassName -> com.mchange.v2.c3p0.impl.DefaultConnectionTester, contextClassLoaderSource -> caller, dataSourceName -> 2xmt4d9mzc9er6ko2cn|16e20167, debugUnreturnedConnectionStackTraces -> false, description -> null, driverClass -> org.postgresql.Driver, extensions -> {}, factoryClassLocation -> null, forceIgnoreUnresolvedTransactions -> false, forceSynchronousCheckins -> false, forceUseNamedDriverClass -> false, identityToken -> 2xmt4d9mzc9er6ko2cn|16e20167, idleConnectionTestPeriod -> 0, initialPoolSize -> 3, jdbcUrl -> jdbc:postgresql://localhost:5432/input-test-1m, maxAdministrativeTaskTime -> 0, maxConnectionAge -> 0, maxIdleTime -> 0, maxIdleTimeExcessConnections -> 0, maxPoolSize -> 15, maxStatements -> 0, maxStatementsPerConnection -> 0, minPoolSize -> 3, numHelperThreads -> 3, preferredTestQuery -> null, privilegeSpawnedThreads -> false, properties -> {user=******, password=******}, propertyCycle -> 0, statementCacheNumDeferredCloseThreads -> 0, testConnectionOnCheckin -> false, testConnectionOnCheckout -> false, unreturnedConnectionTimeout -> 0, userOverrides -> {}, usesTraditionalReflectiveProxies -> false ]

Thread Thread[vert.x-eventloop-thread-0,5,main] has been blocked for 4047 ms, time limit is 2000

Thread Thread[vert.x-eventloop-thread-0,5,main] has been blocked for 5049 ms, time limit is 2000

io.vertx.core.VertxException: Thread blocked

at rx.internal.operators.OperatorMerge$MergeSubscriber.emit(OperatorMerge.java:568)

at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:249)

at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:148)

at rx.internal.operators.OnSubscribeMap$MapSubscriber.onNext(OnSubscribeMap.java:77)

at rx.internal.operators.OnSubscribeFlattenIterable$FlattenIterableSubscriber.drain(OnSubscribeFlattenIterable.java:236)

at rx.internal.operators.OnSubscribeFlattenIterable$FlattenIterableSubscriber.onNext(OnSubscribeFlattenIterable.java:128)

at rx.internal.operators.OnSubscribeMap$MapSubscriber.onNext(OnSubscribeMap.java:77)

at rx.internal.operators.OperatorMerge$MergeSubscriber.emitScalar(OperatorMerge.java:395)

at rx.internal.operators.OperatorMerge$MergeSubscriber.tryEmit(OperatorMerge.java:355)

at rx.internal.operators.OperatorMerge$InnerSubscriber.onNext(OperatorMerge.java:846)

at io.vertx.rx.java.SingleOnSubscribeAdapter.fireNext(SingleOnSubscribeAdapter.java:53)

at io.vertx.rx.java.ObservableFuture$1.dispatch(ObservableFuture.java:58)

at io.vertx.rx.java.ObservableFuture$HandlerAdapter.handle(ObservableFuture.java:32)

at io.vertx.rx.java.ObservableFuture$HandlerAdapter.handle(ObservableFuture.java:12)

at io.vertx.core.impl.FutureImpl.checkCallHandler(FutureImpl.java:158)

at io.vertx.core.impl.FutureImpl.setHandler(FutureImpl.java:100)

at io.vertx.core.impl.ContextImpl.lambda$null$0(ContextImpl.java:271)

at io.vertx.core.impl.ContextImpl$$Lambda$38/1095017261.handle(Unknown Source)

at io.vertx.core.impl.ContextImpl.lambda$wrapTask$2(ContextImpl.java:316)

at io.vertx.core.impl.ContextImpl$$Lambda$25/1307904972.run(Unknown Source)

at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163)

at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:418)

at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:440)

at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:873)

at java.lang.Thread.run(Thread.java:745)

Thread Thread[vert.x-eventloop-thread-0,5,main] has been blocked for 6052 ms, time limit is 2000