tdlight-team/tdlight-java
tdlight-team/tdlight-java
1
2
Fork 0
Code Issues Releases Wiki Activity

Fix race condition

Browse Source
This commit is contained in:
Andrea Cavalli 2020-10-13 18:33:06 +02:00
parent 2647bd0d70
commit fe8bfd17e0
3 changed files with 90 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/main/java/it/tdlight/common/InternalClient.java
View File

@ -10,7 +10,7 @@ import java.util.concurrent.ConcurrentHashMap;
public class InternalClient implements ClientEventsHandler, TelegramClient { public class InternalClient implements ClientEventsHandler, TelegramClient {
private static final java.lang.Object CLIENT_CREATION_LOCK = new java.lang.Object(); static final java.lang.Object CLIENT_CREATION_LOCK = new java.lang.Object();
private final ConcurrentHashMap<Long, Handler> handlers = new ConcurrentHashMap<Long, Handler>(); private final ConcurrentHashMap<Long, Handler> handlers = new ConcurrentHashMap<Long, Handler>();
private final int clientId; private final int clientId;
@ -30,9 +30,8 @@ public class InternalClient implements ClientEventsHandler, TelegramClient {
this.updatesHandler = null; this.updatesHandler = null;
this.defaultExceptionHandler = defaultExceptionHandler; this.defaultExceptionHandler = defaultExceptionHandler;
this.clientManager = clientManager; this.clientManager = clientManager;
clientManager.preregisterClient(this);
this.clientId = NativeClientAccess.create(); this.clientId = NativeClientAccess.create();
clientManager.registerClient(clientId, this); clientManager.registerClient(clientId, this);
} }
} }
@ -46,9 +45,8 @@ public class InternalClient implements ClientEventsHandler, TelegramClient {
this.updatesHandler = new MultiHandler(updatesHandler, updateExceptionHandler); this.updatesHandler = new MultiHandler(updatesHandler, updateExceptionHandler);
this.clientManager = clientManager; this.clientManager = clientManager;
this.defaultExceptionHandler = defaultExceptionHandler; this.defaultExceptionHandler = defaultExceptionHandler;
clientManager.preregisterClient(this);
this.clientId = NativeClientAccess.create(); this.clientId = NativeClientAccess.create();
clientManager.registerClient(clientId, this); clientManager.registerClient(clientId, this);
} }
} }

22
src/main/java/it/tdlight/common/InternalClientManager.java
View File

@ -12,7 +12,6 @@ public class InternalClientManager implements AutoCloseable {
private final String implementationName; private final String implementationName;
private final ResponseReceiver responseReceiver = new ResponseReceiver(this::handleClientEvents); private final ResponseReceiver responseReceiver = new ResponseReceiver(this::handleClientEvents);
private final ConcurrentHashMap<Integer, ClientEventsHandler> registeredClientEventHandlers = new ConcurrentHashMap<>(); private final ConcurrentHashMap<Integer, ClientEventsHandler> registeredClientEventHandlers = new ConcurrentHashMap<>();
private final ConcurrentHashMap<ClientEventsHandler, java.lang.Object> unregisteredClientEventHandlers = new ConcurrentHashMap<>();
private final AtomicLong currentQueryId = new AtomicLong(); private final AtomicLong currentQueryId = new AtomicLong();
private InternalClientManager(String implementationName) { private InternalClientManager(String implementationName) {
@ -32,15 +31,6 @@ public class InternalClientManager implements AutoCloseable {
private void handleClientEvents(int clientId, boolean isClosed, long[] clientEventIds, Object[] clientEvents) { private void handleClientEvents(int clientId, boolean isClosed, long[] clientEventIds, Object[] clientEvents) {
ClientEventsHandler handler = registeredClientEventHandlers.get(clientId); ClientEventsHandler handler = registeredClientEventHandlers.get(clientId);
if (handler == null) {
handler = unregisteredClientEventHandlers
.keySet()
.stream()
.filter(item -> item.getClientId() == clientId)
.findAny()
.orElse(null);
}
if (handler != null) { if (handler != null) {
handler.handleEvents(isClosed, clientEventIds, clientEvents); handler.handleEvents(isClosed, clientEventIds, clientEvents);
} else { } else {
@ -48,17 +38,17 @@ public class InternalClientManager implements AutoCloseable {
} }
if (isClosed) { if (isClosed) {
System.err.println("Unregister client " + clientId);
registeredClientEventHandlers.remove(clientId); registeredClientEventHandlers.remove(clientId);
} }
} }
public void preregisterClient(ClientEventsHandler client) {
this.unregisteredClientEventHandlers.put(client, new java.lang.Object());
}
public void registerClient(int clientId, InternalClient internalClient) { public void registerClient(int clientId, InternalClient internalClient) {
registeredClientEventHandlers.put(clientId, internalClient); System.err.println("Register client " + clientId + ", " + internalClient);
unregisteredClientEventHandlers.remove(internalClient); boolean replaced = registeredClientEventHandlers.put(clientId, internalClient) != null;
if (replaced) {
throw new IllegalStateException("Client " + clientId + " already registered");
}
} }
public String getImplementationName() { public String getImplementationName() {

106
src/main/java/it/tdlight/common/ResponseReceiver.java
View File

@ -1,11 +1,22 @@
package it.tdlight.common; package it.tdlight.common;
import static it.tdlight.common.InternalClient.CLIENT_CREATION_LOCK;
import it.tdlight.jni.TdApi; import it.tdlight.jni.TdApi;
import it.tdlight.jni.TdApi.Object;
import java.util.ArrayList;
import java.util.Arrays; import java.util.Arrays;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CountDownLatch; import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;
public class ResponseReceiver extends Thread implements AutoCloseable { public class ResponseReceiver extends Thread implements AutoCloseable {
private static final boolean USE_OPTIMIZED_DISPATCHER = Boolean.parseBoolean(System.getProperty(
"tdlight.dispatcher.use_optimized_dispatcher",
"true"
));
private static final int MAX_EVENTS = 1000; private static final int MAX_EVENTS = 1000;
private static final int[] originalSortingSource = new int[MAX_EVENTS]; private static final int[] originalSortingSource = new int[MAX_EVENTS];
static { static {
@ -37,50 +48,95 @@ public class ResponseReceiver extends Thread implements AutoCloseable {
int[] sortIndex; int[] sortIndex;
try { try {
while(true) { while(true) {
int resultsCount = NativeClientAccess.receive(clientIds, eventIds, events, 2.0 /*seconds*/); int resultsCount;
synchronized (CLIENT_CREATION_LOCK) {
resultsCount = NativeClientAccess.receive(clientIds, eventIds, events, 2.0 /*seconds*/);
}
if (resultsCount <= 0) if (resultsCount <= 0)
continue; continue;
// Generate a list of indices sorted by client id, from 0 to resultsCount if (USE_OPTIMIZED_DISPATCHER) {
sortIndex = generateSortIndex(0, resultsCount, clientIds); // Generate a list of indices sorted by client id, from 0 to resultsCount
sortIndex = generateSortIndex(0, resultsCount, clientIds);
int lastClientId = clientIds[sortIndex[0]]; int lastClientId = clientIds[sortIndex[0]];
int lastClientIdEventsCount = 0; int lastClientIdEventsCount = 0;
boolean lastClientClosed = false; boolean lastClientClosed = false;
for (int i = 0; i <= resultsCount; i++) { for (int i = 0; i <= resultsCount; i++) {
if (i == resultsCount || (i != 0 && clientIds[sortIndex[i]] != lastClientId)) { if (i == resultsCount || (i != 0 && clientIds[sortIndex[i]] != lastClientId)) {
if (lastClientIdEventsCount > 0) { if (lastClientIdEventsCount > 0) {
int clientId = lastClientId; int clientId = lastClientId;
long[] clientEventIds = new long[lastClientIdEventsCount]; long[] clientEventIds = new long[lastClientIdEventsCount];
TdApi.Object[] clientEvents = new TdApi.Object[lastClientIdEventsCount]; TdApi.Object[] clientEvents = new TdApi.Object[lastClientIdEventsCount];
for (int j = 0; j < lastClientIdEventsCount; j++) { for (int j = 0; j < lastClientIdEventsCount; j++) {
clientEventIds[j] = eventIds[sortIndex[i - lastClientIdEventsCount + j]]; clientEventIds[j] = eventIds[sortIndex[i - lastClientIdEventsCount + j]];
clientEvents[j] = events[sortIndex[i - lastClientIdEventsCount + j]]; clientEvents[j] = events[sortIndex[i - lastClientIdEventsCount + j]];
if (clientEventIds[j] == 0 && clientEvents[j] instanceof TdApi.UpdateAuthorizationState) { if (clientEventIds[j] == 0 && clientEvents[j] instanceof TdApi.UpdateAuthorizationState) {
TdApi.AuthorizationState authorizationState = ((TdApi.UpdateAuthorizationState) clientEvents[j]).authorizationState; TdApi.AuthorizationState authorizationState = ((TdApi.UpdateAuthorizationState) clientEvents[j]).authorizationState;
if (authorizationState instanceof TdApi.AuthorizationStateClosed) { if (authorizationState instanceof TdApi.AuthorizationStateClosed) {
lastClientClosed = true; lastClientClosed = true;
}
} }
} }
eventsHandler.handleClientEvents(clientId, lastClientClosed, clientEventIds, clientEvents);
} }
eventsHandler.handleClientEvents(clientId, lastClientClosed, clientEventIds, clientEvents); if (i < resultsCount) {
lastClientId = clientIds[sortIndex[i]];
lastClientIdEventsCount = 0;
lastClientClosed = false;
}
} }
if (i < resultsCount) { if (i < resultsCount) {
lastClientId = clientIds[sortIndex[i]]; lastClientIdEventsCount++;
lastClientIdEventsCount = 0; }
lastClientClosed = false; }
} else {
class Event {
public final int clientId;
public final long eventId;
public final Object event;
public Event(int clientId, long eventId, Object event) {
this.clientId = clientId;
this.eventId = eventId;
this.event = event;
} }
} }
if (i < resultsCount) { List<Event> eventsList = new ArrayList<>(resultsCount);
lastClientIdEventsCount++; for (int i = 0; i < resultsCount; i++) {
eventsList.add(new Event(clientIds[i], eventIds[i], events[i]));
}
Set<Integer> clientIds = eventsList.stream().map(e -> e.clientId).collect(Collectors.toSet());
for (int clientId : clientIds) {
List<Event> clientEventsList = eventsList.stream().filter(e -> e.clientId == clientId).collect(Collectors.toList());
long[] clientEventIds = new long[clientEventsList.size()];
Object[] clientEvents = new Object[clientEventsList.size()];
boolean closed = false;
for (int i = 0; i < clientEventsList.size(); i++) {
Event e = clientEventsList.get(i);
clientEventIds[i] = e.eventId;
clientEvents[i] = e.event;
if (e.eventId == 0 && e.event instanceof TdApi.UpdateAuthorizationState) {
TdApi.AuthorizationState authorizationState = ((TdApi.UpdateAuthorizationState) e.event).authorizationState;
if (authorizationState instanceof TdApi.AuthorizationStateClosed) {
closed = true;
}
}
}
eventsHandler.handleClientEvents(clientId, closed, clientEventIds, clientEvents);
} }
} }
Arrays.fill(clientIds, 0);
Arrays.fill(eventIds, 0);
Arrays.fill(events, null); Arrays.fill(events, null);
} }
} finally { } finally {