artem-v
commented
6 years ago Line of interest is
at reactor.core.scheduler.Schedulers.workerSchedule(Schedulers.java:708) ~[reactor-core-3.1.8.RELEASE.jar:3.1.8.RELEASE]When one calls Schedulers.fromExecutorService(exexcutorService) he creates DelegateServiceScheduler instance, which has method :
@Override
public Worker createWorker() {
return new ExecutorServiceWorker(executor);
}By turn, ExecutorServiceWorker has field Composite tasks = Disposables.composite() ,. So here it goes: in workerSchedule method we see following:
WorkerTask sr = new WorkerTask(task, tasks);
if (!tasks.add(sr)) {
throw Exceptions.failWithRejected();
}If look at .add method which by turn would lead to .addEntry method we can see following:
boolean addEntry(Disposable value) {
final Disposable[] a = disposables;
final int m = mask;
int pos = mix(value.hashCode()) & m;
Disposable curr = a[pos];
if (curr != null) {
if (curr.equals(value)) {
return false;
}
for (;;) {
pos = (pos + 1) & m;
curr = a[pos];
if (curr == null) {
break;
}
if (curr.equals(value)) {
return false;
}
}
}
a[pos] = value;
if (++size >= maxSize) {
rehash();
}
return true;
}So, looks like return false; block only called when curr.equals(value) .
To conclude all above it appears that we can't schedule same task again and again and again to the scheduler which was created via reactor.core.scheduler.Schedulers#fromExecutorService .
Just recently were about running c5.2xlarge perf test with 1 client 1 gw and 1 service. Test didn't start and failed with following in logs.
On the client:
On the gateway:
On the service: