You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@openmeetings.apache.org by so...@apache.org on 2018/03/26 10:40:38 UTC
[openmeetings] branch master updated: [OPENMEETINGS-1836] another
attempt to fix performance
This is an automated email from the ASF dual-hosted git repository.
solomax pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/openmeetings.git
The following commit(s) were added to refs/heads/master by this push:
new 6eac1a6 [OPENMEETINGS-1836] another attempt to fix performance
6eac1a6 is described below
commit 6eac1a67205cd12fa831bd18a120a6ecc7fb0abd
Author: Maxim Solodovnik <so...@gmail.com>
AuthorDate: Mon Mar 26 17:15:31 2018 +0700
[OPENMEETINGS-1836] another attempt to fix performance
---
.../openmeetings/core/util/WebSocketHelper.java | 21 +++-
.../openmeetings/db/entity/basic/Client.java | 20 ++++
.../openmeetings/db/manager/IClientManager.java | 3 +-
.../apache/openmeetings/web/app/ClientManager.java | 119 ++++++++++++++++-----
.../openmeetings/web/app/WhiteboardManager.java | 48 ++++++++-
.../src/main/webapp/WEB-INF/classes/hazelcast.xml | 37 ++++++-
6 files changed, 215 insertions(+), 33 deletions(-)
diff --git a/openmeetings-core/src/main/java/org/apache/openmeetings/core/util/WebSocketHelper.java b/openmeetings-core/src/main/java/org/apache/openmeetings/core/util/WebSocketHelper.java
index ea6b9c1..f7f02a1 100644
--- a/openmeetings-core/src/main/java/org/apache/openmeetings/core/util/WebSocketHelper.java
+++ b/openmeetings-core/src/main/java/org/apache/openmeetings/core/util/WebSocketHelper.java
@@ -22,6 +22,7 @@ import static org.apache.openmeetings.core.remote.ScopeApplicationAdapter.getApp
import static org.apache.openmeetings.db.util.FormatHelper.getDisplayName;
import java.io.IOException;
+import java.util.Collection;
import java.util.List;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
@@ -229,7 +230,7 @@ public class WebSocketHelper {
Application app = (Application)getApp();
WebSocketSettings settings = WebSocketSettings.Holder.get(app);
IWebSocketConnectionRegistry reg = settings.getConnectionRegistry();
- Executor executor = settings.getWebSocketPushMessageExecutor();
+ Executor executor = settings.getWebSocketPushMessageExecutor(); // new NewThreadExecutor();
for (IWebSocketConnection c : reg.getConnections(app)) {
executor.run(() -> {
try {
@@ -243,7 +244,10 @@ public class WebSocketHelper {
}
protected static void publish(IClusterWsMessage m) {
- getApp().publishWsTopic(m);
+ IApplication app = getApp();
+ new Thread(() -> {
+ app.publishWsTopic(m);
+ }).start();
}
protected static void sendRoom(final Long roomId, final JSONObject m, Predicate<Client> check, BiFunction<JSONObject, Client, String> func) {
@@ -262,7 +266,7 @@ public class WebSocketHelper {
}
private static void send(
- final Function<Application, List<Client>> func
+ final Function<Application, Collection<Client>> func
, BiConsumer<IWebSocketConnection, Client> consumer
, Predicate<Client> check)
{
@@ -270,7 +274,7 @@ public class WebSocketHelper {
Application app = (Application)getApp();
WebSocketSettings settings = WebSocketSettings.Holder.get(app);
IWebSocketConnectionRegistry reg = settings.getConnectionRegistry();
- Executor executor = settings.getWebSocketPushMessageExecutor();
+ Executor executor = settings.getWebSocketPushMessageExecutor(); //new NewThreadExecutor();
for (Client c : func.apply(app)) {
if (check == null || check.test(c)) {
final IWebSocketConnection wc = reg.getConnection(app, c.getSessionId(), new PageIdKey(c.getPageId()));
@@ -281,4 +285,13 @@ public class WebSocketHelper {
}
}).start();
}
+
+ public static class NewThreadExecutor implements Executor {
+ @Override
+ public void run(Runnable command) {
+ new Thread(() -> {
+ command.run();
+ }).start();
+ }
+ }
}
diff --git a/openmeetings-db/src/main/java/org/apache/openmeetings/db/entity/basic/Client.java b/openmeetings-db/src/main/java/org/apache/openmeetings/db/entity/basic/Client.java
index c386c3d..8f1cfbf 100644
--- a/openmeetings-db/src/main/java/org/apache/openmeetings/db/entity/basic/Client.java
+++ b/openmeetings-db/src/main/java/org/apache/openmeetings/db/entity/basic/Client.java
@@ -438,6 +438,26 @@ public class Client implements IClient, IWsClient {
return toJson(self).put("sid", _sid).put("streams", _streams);
}
+ public void merge(Client c) {
+ user = c.user;
+ room = c.room;
+ Set<Right> rr = new HashSet<>(c.rights);
+ rights.clear();
+ rights.addAll(rr);
+ Set<Activity> aa = new HashSet<>(c.activities);
+ activities.clear();
+ activities.addAll(aa);
+ Set<String> ss = new HashSet<>(c.streams);
+ streams.clear();
+ streams.addAll(ss);
+ pod = c.pod;
+ cam = c.cam;
+ mic = c.mic;
+ width = c.width;
+ height = c.height;
+ recordingId = c.recordingId;
+ }
+
@Override
public int hashCode() {
final int prime = 31;
diff --git a/openmeetings-db/src/main/java/org/apache/openmeetings/db/manager/IClientManager.java b/openmeetings-db/src/main/java/org/apache/openmeetings/db/manager/IClientManager.java
index cab82c6..037e4e6 100644
--- a/openmeetings-db/src/main/java/org/apache/openmeetings/db/manager/IClientManager.java
+++ b/openmeetings-db/src/main/java/org/apache/openmeetings/db/manager/IClientManager.java
@@ -18,6 +18,7 @@
*/
package org.apache.openmeetings.db.manager;
+import java.util.Collection;
import java.util.List;
import java.util.Set;
@@ -28,7 +29,7 @@ public interface IClientManager {
Client get(String uid);
Client getBySid(String sid);
List<Client> listByRoom(Long roomId);
- List<Client> listByUser(Long userId);
+ Collection<Client> listByUser(Long userId);
Client update(Client c);
void exit(IClient c);
diff --git a/openmeetings-web/src/main/java/org/apache/openmeetings/web/app/ClientManager.java b/openmeetings-web/src/main/java/org/apache/openmeetings/web/app/ClientManager.java
index 749f9b6..794375f 100644
--- a/openmeetings-web/src/main/java/org/apache/openmeetings/web/app/ClientManager.java
+++ b/openmeetings-web/src/main/java/org/apache/openmeetings/web/app/ClientManager.java
@@ -22,13 +22,17 @@ import static org.apache.openmeetings.core.util.WebSocketHelper.sendRoom;
import static org.apache.openmeetings.web.app.Application.getHazelcast;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Predicate;
+import javax.annotation.PostConstruct;
+
import org.apache.openmeetings.db.dao.log.ConferenceLogDao;
import org.apache.openmeetings.db.entity.basic.Client;
import org.apache.openmeetings.db.entity.basic.IClient;
@@ -41,7 +45,12 @@ import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
+import com.hazelcast.core.EntryEvent;
import com.hazelcast.core.IMap;
+import com.hazelcast.map.listener.EntryAddedListener;
+import com.hazelcast.map.listener.EntryRemovedListener;
+import com.hazelcast.map.listener.EntryUpdatedListener;
+import com.hazelcast.query.Predicates;
@Component
public class ClientManager implements IClientManager {
@@ -49,37 +58,48 @@ public class ClientManager implements IClientManager {
private static final String ROOMS_KEY = "ROOMS_KEY";
private static final String ONLINE_USERS_KEY = "ONLINE_USERS_KEY";
private static final String UID_BY_SID_KEY = "UID_BY_SID_KEY";
+ private final Map<String, Client> onlineClients = new ConcurrentHashMap<>();
+ private final Map<Long, Set<String>> onlineRooms = new ConcurrentHashMap<>();
@Autowired
private ConferenceLogDao confLogDao;
- private static Map<String, Client> map() {
+ private static IMap<String, Client> map() {
return getHazelcast().getMap(ONLINE_USERS_KEY);
}
- private static Map<String, String> mapUidBySid() {
+ private static Map<String, String> mapBySid() {
return getHazelcast().getMap(UID_BY_SID_KEY);
}
- private static IMap<Long, Set<String>> getRooms() {
+ private static IMap<Long, Set<String>> rooms() {
return getHazelcast().getMap(ROOMS_KEY);
}
+ @PostConstruct
+ void init() {
+ map().addEntryListener(new ClientListener(), true);
+ rooms().addEntryListener(new RoomListener(), true);
+ }
+
public void add(Client c) {
log.debug("Adding online client: {}, room: {}", c.getUid(), c.getRoom());
c.setServerId(Application.get().getServerId());
map().put(c.getUid(), c);
- mapUidBySid().put(c.getSid(), c.getUid());
+ onlineClients.put(c.getUid(), c);
+ mapBySid().put(c.getSid(), c.getUid());
}
@Override
public Client update(Client c) {
- map().put(c.getUid(), c); // update in storage
+ onlineClients.get(c.getUid()).merge(c);
+ map().put(c.getUid(), c);
return c;
}
+
@Override
public Client get(String uid) {
- return uid == null ? null : map().get(uid);
+ return uid == null ? null : onlineClients.get(uid);
}
@Override
@@ -87,8 +107,8 @@ public class ClientManager implements IClientManager {
if (sid == null) {
return null;
}
- String uid = mapUidBySid().get(sid);
- return uid == null ? null : map().get(uid);
+ String uid = mapBySid().get(sid);
+ return uid == null ? null : get(uid);
}
public void exitRoom(IClient c) {
@@ -110,7 +130,8 @@ public class ClientManager implements IClientManager {
exitRoom(c);
log.debug("Removing online client: {}, roomId: {}", c.getUid(), c.getRoomId());
map().remove(c.getUid());
- mapUidBySid().remove(c.getSid());
+ onlineClients.remove(c.getUid());
+ mapBySid().remove(c.getSid());
}
}
@@ -125,7 +146,7 @@ public class ClientManager implements IClientManager {
@Override
public Set<Long> getActiveRoomIds() {
- return getRooms().keySet();
+ return onlineRooms.keySet();
}
/**
@@ -137,13 +158,14 @@ public class ClientManager implements IClientManager {
public int addToRoom(Client c) {
Long roomId = c.getRoom().getId();
log.debug("Adding online room client: {}, room: {}", c.getUid(), roomId);
- IMap<Long, Set<String>> rooms = getRooms();
+ IMap<Long, Set<String>> rooms = rooms();
rooms.lock(roomId);
rooms.putIfAbsent(roomId, new ConcurrentHashSet<String>());
Set<String> set = rooms.get(roomId);
set.add(c.getUid());
final int count = set.size();
rooms.put(roomId, set);
+ onlineRooms.put(roomId, set);
rooms.unlock(roomId);
update(c);
return count;
@@ -153,12 +175,15 @@ public class ClientManager implements IClientManager {
Long roomId = _c.getRoomId();
log.debug("Removing online room client: {}, room: {}", _c.getUid(), roomId);
if (roomId != null) {
- Map<Long, Set<String>> rooms = getRooms();
+ IMap<Long, Set<String>> rooms = rooms();
+ rooms.lock(roomId);
Set<String> clients = rooms.get(roomId);
if (clients != null) {
clients.remove(_c.getUid());
rooms.put(roomId, clients);
+ onlineRooms.put(roomId, clients);
}
+ rooms.unlock(roomId);
/* FIXME TODO KurentoHandler
if (_c instanceof StreamClient) {
StreamClient sc = (StreamClient)_c;
@@ -200,15 +225,8 @@ public class ClientManager implements IClientManager {
}
@Override
- public List<Client> listByUser(Long userId) {
- List<Client> result = new ArrayList<>();
- for (Map.Entry<String, Client> e : map().entrySet()) {
- if (e.getValue().getUserId().equals(userId)) {
- result.add(e.getValue());
- break;
- }
- }
- return result;
+ public Collection<Client> listByUser(Long userId) {
+ return map().values(Predicates.equal("userId", userId));
}
@Override
@@ -219,7 +237,7 @@ public class ClientManager implements IClientManager {
public List<Client> listByRoom(Long roomId, Predicate<Client> filter) {
List<Client> clients = new ArrayList<>();
if (roomId != null) {
- Set<String> uids = getRooms().get(roomId);
+ Set<String> uids = onlineRooms.get(roomId);
if (uids != null) {
for (String uid : uids) {
Client c = get(uid);
@@ -234,7 +252,7 @@ public class ClientManager implements IClientManager {
public Set<Long> listRoomIds(Long userId) {
Set<Long> result = new HashSet<>();
- for (Entry<Long, Set<String>> me : getRooms().entrySet()) {
+ for (Entry<Long, Set<String>> me : onlineRooms.entrySet()) {
for (String uid : me.getValue()) {
Client c = get(uid);
if (c != null && c.getUserId().equals(userId)) {
@@ -246,7 +264,7 @@ public class ClientManager implements IClientManager {
}
public boolean isInRoom(long roomId, long userId) {
- Set<String> clients = getRooms().get(roomId);
+ Set<String> clients = onlineRooms.get(roomId);
if (clients != null) {
for (String uid : clients) {
Client c = get(uid);
@@ -280,4 +298,57 @@ public class ClientManager implements IClientManager {
}
}
}
+
+ public class ClientListener implements
+ EntryAddedListener<String, Client>
+ , EntryUpdatedListener<String, Client>
+ , EntryRemovedListener<String, Client>
+ {
+ @Override
+ public void entryAdded(EntryEvent<String, Client> event) {
+ log.trace("ClientListener::Add");
+ final String uid = event.getKey();
+ if (onlineClients.containsKey(uid)) {
+ onlineClients.get(uid).merge(event.getValue());
+ } else {
+ onlineClients.put(uid, event.getValue());
+ }
+ }
+
+ @Override
+ public void entryUpdated(EntryEvent<String, Client> event) {
+ log.trace("ClientListener::Update");
+ onlineClients.get(event.getKey()).merge(event.getValue());
+ }
+
+ @Override
+ public void entryRemoved(EntryEvent<String, Client> event) {
+ log.trace("ClientListener::Remove");
+ onlineClients.remove(event.getKey());
+ }
+ }
+
+ public class RoomListener implements
+ EntryAddedListener<Long, Set<String>>
+ , EntryUpdatedListener<Long, Set<String>>
+ , EntryRemovedListener<Long, Set<String>>
+ {
+ @Override
+ public void entryAdded(EntryEvent<Long, Set<String>> event) {
+ log.trace("RoomListener::Add");
+ onlineRooms.put(event.getKey(), event.getValue());
+ }
+
+ @Override
+ public void entryUpdated(EntryEvent<Long, Set<String>> event) {
+ log.trace("RoomListener::Update");
+ onlineRooms.put(event.getKey(), event.getValue());
+ }
+
+ @Override
+ public void entryRemoved(EntryEvent<Long, Set<String>> event) {
+ log.trace("RoomListener::Remove");
+ onlineRooms.remove(event.getKey(), event.getValue());
+ }
+ }
}
diff --git a/openmeetings-web/src/main/java/org/apache/openmeetings/web/app/WhiteboardManager.java b/openmeetings-web/src/main/java/org/apache/openmeetings/web/app/WhiteboardManager.java
index 52b9994..d12ff86 100644
--- a/openmeetings-web/src/main/java/org/apache/openmeetings/web/app/WhiteboardManager.java
+++ b/openmeetings-web/src/main/java/org/apache/openmeetings/web/app/WhiteboardManager.java
@@ -21,16 +21,26 @@ package org.apache.openmeetings.web.app;
import static org.apache.openmeetings.util.OpenmeetingsVariables.getDefaultLang;
import static org.apache.openmeetings.web.app.Application.getHazelcast;
+import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+import javax.annotation.PostConstruct;
import org.apache.openmeetings.db.dao.label.LabelDao;
import org.apache.openmeetings.db.dto.room.Whiteboard;
import org.apache.openmeetings.db.dto.room.Whiteboards;
import org.apache.openmeetings.db.manager.IWhiteboardManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
+import com.hazelcast.core.EntryEvent;
import com.hazelcast.core.IMap;
+import com.hazelcast.map.listener.EntryAddedListener;
+import com.hazelcast.map.listener.EntryRemovedListener;
+import com.hazelcast.map.listener.EntryUpdatedListener;
/**
* Hazelcast based Whiteboard manager
@@ -40,12 +50,19 @@ import com.hazelcast.core.IMap;
*/
@Component
public class WhiteboardManager implements IWhiteboardManager {
+ private static final Logger log = LoggerFactory.getLogger(WhiteboardManager.class);
+ private final Map<Long, Whiteboards> onlineWbs = new ConcurrentHashMap<>();
private static final String WBS_KEY = "WBS_KEY";
private static IMap<Long, Whiteboards> map() {
return getHazelcast().getMap(WBS_KEY);
}
+ @PostConstruct
+ void init() {
+ map().addEntryListener(new WbListener(), true);
+ }
+
public boolean tryLock(Long roomId) {
return map().tryLock(roomId);
}
@@ -75,7 +92,7 @@ public class WhiteboardManager implements IWhiteboardManager {
if (roomId == null) {
return null;
}
- Whiteboards wbs = map().get(roomId);
+ Whiteboards wbs = onlineWbs.get(roomId);
if (wbs == null) {
wbs = new Whiteboards(roomId);
Whiteboard wb = add(wbs, langId);
@@ -132,7 +149,32 @@ public class WhiteboardManager implements IWhiteboardManager {
update(wbs);
}
- private static void update(Whiteboards wbs) {
- map().put(wbs.getRoomId(), wbs);
+ private void update(Whiteboards wbs) {
+ onlineWbs.put(wbs.getRoomId(), wbs);
+ new Thread(() -> map().put(wbs.getRoomId(), wbs)).start();
+ }
+
+ public class WbListener implements
+ EntryAddedListener<Long, Whiteboards>
+ , EntryUpdatedListener<Long, Whiteboards>
+ , EntryRemovedListener<Long, Whiteboards>
+ {
+ @Override
+ public void entryAdded(EntryEvent<Long, Whiteboards> event) {
+ log.trace("WbListener::Add");
+ onlineWbs.put(event.getKey(), event.getValue());
+ }
+
+ @Override
+ public void entryUpdated(EntryEvent<Long, Whiteboards> event) {
+ log.trace("WbListener::Update");
+ onlineWbs.put(event.getKey(), event.getValue());
+ }
+
+ @Override
+ public void entryRemoved(EntryEvent<Long, Whiteboards> event) {
+ log.trace("WbListener::Remove");
+ onlineWbs.remove(event.getKey());
+ }
}
}
diff --git a/openmeetings-web/src/main/webapp/WEB-INF/classes/hazelcast.xml b/openmeetings-web/src/main/webapp/WEB-INF/classes/hazelcast.xml
index 3734821..65cc426 100644
--- a/openmeetings-web/src/main/webapp/WEB-INF/classes/hazelcast.xml
+++ b/openmeetings-web/src/main/webapp/WEB-INF/classes/hazelcast.xml
@@ -21,8 +21,43 @@
<hazelcast
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://www.hazelcast.com/schema/config"
- xsi:schemaLocation="http://www.hazelcast.com/schema/config https://hazelcast.com/schema/config/hazelcast-config-3.8.xsd"
+ xsi:schemaLocation="http://www.hazelcast.com/schema/config https://hazelcast.com/schema/config/hazelcast-config-3.9.xsd"
>
+ <map name="ONLINE_USERS_KEY">
+ <near-cache>
+ <eviction-policy>NONE</eviction-policy>
+ <in-memory-format>OBJECT</in-memory-format>
+ <cache-local-entries>true</cache-local-entries>
+ </near-cache>
+ </map>
+ <map name="ROOMS_KEY">
+ <near-cache>
+ <eviction-policy>NONE</eviction-policy>
+ <in-memory-format>OBJECT</in-memory-format>
+ <cache-local-entries>true</cache-local-entries>
+ </near-cache>
+ </map>
+ <map name="UID_BY_SID_KEY">
+ <near-cache>
+ <eviction-policy>NONE</eviction-policy>
+ <in-memory-format>OBJECT</in-memory-format>
+ <cache-local-entries>true</cache-local-entries>
+ </near-cache>
+ </map>
+ <map name="WBS_KEY">
+ <near-cache>
+ <eviction-policy>NONE</eviction-policy>
+ <in-memory-format>OBJECT</in-memory-format>
+ <cache-local-entries>true</cache-local-entries>
+ </near-cache>
+ </map>
+ <map name="STREAM_CLIENT_KEY">
+ <near-cache>
+ <eviction-policy>NONE</eviction-policy>
+ <in-memory-format>OBJECT</in-memory-format>
+ <cache-local-entries>true</cache-local-entries>
+ </near-cache>
+ </map>
<instance-name>server-1</instance-name><!-- MAKE SURE THIS ONE IS UNIQUE -->
<network>
<join>
--
To stop receiving notification emails like this one, please contact
solomax@apache.org.