Refactor BackgroundExecutor add submit() methods returning Future (#2121)

- Adds submit() methods that return Future
- Refactor internals to use DaemonScheduleThreadPool
- Delete the now unused DaemonExecutorService
- Tidy internals using wrapMDC() methods
This commit is contained in:
Rob Bygrave authored and GitHub committed 2020-12-03 13:16:51 +13:00
1 parent 9ab26cfb9d
commit 786444a6b9
11 files changed
+254 -267

No files matched your search

@@ -2,12 +2,64 @@ package io.ebeaninternal.server.core;
import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.MDC;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import static org.assertj.core.api.Assertions.assertThat;
public class DefaultBackgroundExecutorTest {
@Test
public void submit_callable() throws Exception {
DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 2, "test");
final Future<String> future0 = es.submit(() -> "Hello");
final Future<String> future1 = es.submit(() -> "There");
final Future<String> future2 = es.submit(() -> {
try {
Thread.sleep(100);
return "Slow";
} catch (InterruptedException e) {
e.printStackTrace();
return "Interrupted";
}
});
es.shutdown();
assertThat(future0.get()).isEqualTo("Hello");
assertThat(future1.get(1, TimeUnit.SECONDS)).isEqualTo("There");
assertThat(future2.get()).isEqualTo("Slow");
}
@Test
public void shutdown_slowCallable_expect_interrupted() throws Exception {
int shutdownWaitSecs = 1;
DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, shutdownWaitSecs, "test");
final Future<String> future2 = es.submit(() -> {
try {
Thread.sleep(1500); // longer than shutdown wait
return "Slow";
} catch (InterruptedException e) {
// expected for this test
Thread.currentThread().interrupt();
return "Interrupted";
}
});
// shutdown waits max shutdownWaitSecs seconds for active tasks
es.shutdown();
assertThat(future2.get()).isEqualTo("Interrupted");
}
@Test
@Ignore("test takes long time")
public void shutdown_when_running_expect_waitAndNiceShutdown() throws Exception {
public void shutdown_when_running_expect_waitAndNiceShutdown() {
DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 20, "test");
@@ -20,7 +72,7 @@ public class DefaultBackgroundExecutorTest {
@Test
@Ignore("test takes long time")
public void shutdown_when_rougeRunnable_expect_InterruptedException() throws Exception {
public void shutdown_when_rougeRunnable_expect_InterruptedException() {
DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 10, "test");
@@ -31,8 +83,44 @@ public class DefaultBackgroundExecutorTest {
es.shutdown();
}
@Test
public void wrapWithNoMDC() {
DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 10, "test");
assertThat(MDC.getCopyOfContextMap()).isNull();
es.wrapMDC(() -> {
assertThat(MDC.getCopyOfContextMap()).isNull();
});
es.wrapMDC(() -> {
assertThat(MDC.getCopyOfContextMap()).isNull();
return "Callable";
});
es.shutdown();
}
class RunFor implements Runnable {
@Test
public void wrapWithMDC_expect_() {
DefaultBackgroundExecutor es = new DefaultBackgroundExecutor(1, 10, "test");
MDC.clear();
MDC.put("hello", "there");
es.wrapMDC(() -> {
assertThat(MDC.get("hello")).isEqualTo("there");
});
es.wrapMDC(() -> {
assertThat(MDC.get("hello")).isEqualTo("there");
return "Callable";
});
es.execute(() -> {
assertThat(MDC.get("hello")).isEqualTo("there");
});
es.submit(() -> {
assertThat(MDC.get("hello")).isEqualTo("there");
return "Callable";
});
MDC.clear();
es.shutdown();
}
private static class RunFor implements Runnable {
final long wait;
final String id;
@@ -1,67 +0,0 @@
package io.ebeaninternal.server.lib.sql;
import io.ebean.BaseTestCase;
import io.ebean.Ebean;
import io.ebean.EbeanServer;
import io.ebeaninternal.server.core.DefaultBackgroundExecutor;
import org.tests.model.basic.Customer;
import org.junit.Test;
public class TestDataSourceMaxWithEntity extends BaseTestCase {
@Test
public void test() {
boolean skipThisTest = true;
if (skipThisTest) {
return;
}
EbeanServer server = Ebean.getServer(null);
DefaultBackgroundExecutor bg = new DefaultBackgroundExecutor(1, 30, "testDs");
try {
for (int i = 0; i < 12; i++) {
// Thread.sleep(10*i);
bg.execute(new ConnRunner(server, 4000, i));
}
Thread.sleep(30000);
server.shutdown(true, false);
} catch (Exception e) {
e.printStackTrace();
}
}
private static class ConnRunner implements Runnable {
final EbeanServer server;
final long sleepMillis;
final int position;
ConnRunner(EbeanServer server, long sleepMillis, int position) {
this.server = server;
this.sleepMillis = sleepMillis;
this.position = position;
}
@Override
public void run() {
server.find(Customer.class).findCount();
try {
System.out.println(position + " sleep " + sleepMillis);
Thread.sleep(sleepMillis);
System.out.println(position + " sleep done");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
@@ -0,0 +1,57 @@
package io.ebeaninternal.server.lib.sql;
import io.ebean.BackgroundExecutor;
import io.ebean.BaseTestCase;
import io.ebean.DB;
import io.ebean.Database;
import org.junit.Ignore;
import org.junit.Test;
import org.tests.model.basic.Customer;
public class TestShutdownWithBackgroundTasks extends BaseTestCase {
@Test
@Ignore
public void test() {
Database server = DB.getDefault();
final BackgroundExecutor bg = server.getBackgroundExecutor();
try {
for (int i = 0; i < 12; i++) {
bg.execute(new Job(server, 500, i));
}
Thread.sleep(1000);
server.shutdown();
} catch (Exception e) {
e.printStackTrace();
}
}
private static class Job implements Runnable {
final Database server;
final long sleepMillis;
final int position;
Job(Database server, long sleepMillis, int position) {
this.server = server;
this.sleepMillis = sleepMillis;
this.position = position;
}
@Override
public void run() {
try {
System.out.println(position + " sleep " + sleepMillis);
Thread.sleep(sleepMillis);
server.find(Customer.class).findCount();
System.out.println(position + " sleep done");
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
e.printStackTrace();
}
}
}
}