Skip to content

Commit 6b4b5ca

Browse files
hypfviehhypfvieh
authored
Merge pull request #291 from hypfvieh/virtual-threads
Virtual threads
2 parents 715518b + 560c163 commit 6b4b5ca

14 files changed

Lines changed: 300 additions & 99 deletions

File tree

‎README.md‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,10 @@ The library will remain open source and MIT licensed and can still be used, fork
8383
- with this change, you can also create Structs and use them as return value having something like `GetCurrentStateStruct` instead of using the long Tuple-name
8484
- Further details on this in [#285](https://github.com/hypfvieh/dbus-java/issues/285)
8585
- Added new commandline option `--disable-tuples` to `InterfaceCodeGenerator` to create `Struct` classes instead of `Tuple`s for multi value return (**Caution** the generated code will only work with dbus-java 6.0.0+)
86+
- Added support for Virtual-Threads in `ReceivingService`
87+
- This can be enabled using the `DBusConnectionBuilder`, example: `DBusConnection sessionConnection = DBusConnectionBuilder.forSystemBus().receivingThreadConfig().withAllVirtualThreads(true).connectionConfig().build()`
88+
- Virtual-Threads can be enabled/disabled for each of the different executor services used in `ReceivingService`: `SIGNAL`, `ERROR`, `METHODCALL`, `METHODRETURN`
89+
- default remains native threads on all executors
8690

8791
##### Changes in 5.2.0 (2025-12-21):
8892
- removed properties from dbus-java.version which causes issues with reproducable builds ([PR#279](https://github.com/hypfvieh/dbus-java/issues/279))

‎dbus-java-core/src/main/java/org/freedesktop/dbus/Marshalling.java‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -759,8 +759,7 @@ public static Object[] deSerializeParameters(Object[] _parameters, Type[] _types
759759
return new Object[] {o};
760760
} else if (!_methodCall && Struct.class.isAssignableFrom(clz)) {
761761
LOGGER.trace("(4) Deserializing Struct return");
762-
Object[] val = deSerializeParameters(_parameters, types, _conn, true);
763-
return val;
762+
return deSerializeParameters(_parameters, types, _conn, true);
764763
}
765764
}
766765

‎dbus-java-core/src/main/java/org/freedesktop/dbus/RemoteInvocationHandler.java‎

Lines changed: 3 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -58,8 +58,9 @@ public Object invoke(Object _proxy, Method _method, Object[] _args) throws Throw
5858
return _args[0] != null && remote.equals(((RemoteInvocationHandler) Proxy.getInvocationHandler(_args[0])).remote);
5959
}
6060
} catch (IllegalArgumentException _exIa) {
61-
return Boolean.FALSE;
61+
LOGGER.trace("Error calling equals method", _exIa);
6262
}
63+
return Boolean.FALSE;
6364
case "finalize":
6465
return null;
6566
case "getClass":
@@ -85,10 +86,8 @@ public Object invoke(Object _proxy, Method _method, Object[] _args) throws Throw
8586
} else if (_args.length == 2 && _args[0] instanceof Long l && _args[1] instanceof Integer i) {
8687
remote.wait(l, i);
8788
}
88-
if (_args.length <= 2) {
89-
return null;
90-
}
9189
}
90+
return null;
9291
case "toString":
9392
return remote.toString();
9493
default:

‎dbus-java-core/src/main/java/org/freedesktop/dbus/connections/base/ReceivingService.java‎

Lines changed: 25 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -9,13 +9,11 @@
99
import org.slf4j.Logger;
1010
import org.slf4j.LoggerFactory;
1111

12+
import java.util.Arrays;
1213
import java.util.Map;
1314
import java.util.Map.Entry;
1415
import java.util.Optional;
15-
import java.util.concurrent.ConcurrentHashMap;
16-
import java.util.concurrent.ExecutorService;
17-
import java.util.concurrent.Executors;
18-
import java.util.concurrent.TimeUnit;
16+
import java.util.concurrent.*;
1917

2018
/**
2119
* Service providing threads for every type of message expected to be received by DBus.
@@ -42,20 +40,34 @@ public class ReceivingService {
4240
ReceivingService(String _namePrefix, ReceivingServiceConfig _rsCfg) {
4341
String prefix = _namePrefix == null ? "" : _namePrefix;
4442
ReceivingServiceConfig rsCfg = Optional.ofNullable(_rsCfg).orElse(ReceivingServiceConfigBuilder.getDefaultConfig());
45-
executors.put(ExecutorNames.SIGNAL,
46-
Executors.newFixedThreadPool(rsCfg.getSignalThreadPoolSize(), new NameableThreadFactory(prefix + "DBus-Signal-Receiver-", true, rsCfg.getSignalThreadPriority())));
47-
executors.put(ExecutorNames.ERROR,
48-
Executors.newFixedThreadPool(rsCfg.getErrorThreadPoolSize(), new NameableThreadFactory(prefix + "DBus-Error-Receiver-", true, rsCfg.getErrorThreadPriority())));
4943

50-
// we need multiple threads here so recursive method calls are possible
51-
executors.put(ExecutorNames.METHODCALL,
52-
Executors.newFixedThreadPool(rsCfg.getMethodCallThreadPoolSize(), new NameableThreadFactory(prefix + "DBus-MethodCall-Receiver-", true, rsCfg.getMethodCallThreadPriority())));
53-
executors.put(ExecutorNames.METHODRETURN,
54-
Executors.newFixedThreadPool(rsCfg.getMethodReturnThreadPoolSize(), new NameableThreadFactory(prefix + "DBus-MethodReturn-Receiver-", true, rsCfg.getMethodReturnThreadPriority())));
44+
Arrays.stream(ExecutorNames.values())
45+
.forEach(t -> {
46+
executors.put(t,
47+
Executors.newFixedThreadPool(rsCfg.getPoolSize(t), createFactory(prefix + t.getThreadName() + "-", _rsCfg.isVirtual(t), _rsCfg.getPriority(t))));
48+
});
5549

5650
retryHandler = rsCfg.getRetryHandler();
5751
}
5852

53+
/**
54+
* Create thread factory for creating native or virtual threads.
55+
*
56+
* @param _name name for created threads
57+
* @param _virtualThreads true to create virtual threads
58+
* @param _priority thread priority (only used for non-virtual threads)
59+
* @return {@link ThreadFactory}
60+
*
61+
* @since 6.0.0 - 2026-01-10
62+
*/
63+
private ThreadFactory createFactory(String _name, boolean _virtualThreads, int _priority) {
64+
if (_virtualThreads) {
65+
return Thread.ofVirtual().name(_name, 1L).factory();
66+
} else {
67+
return new NameableThreadFactory(_name, true, _priority);
68+
}
69+
}
70+
5971
/**
6072
* Execute a runnable which handles a signal.
6173
*

‎dbus-java-core/src/main/java/org/freedesktop/dbus/connections/config/ReceivingServiceConfig.java‎

Lines changed: 72 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -1,98 +1,146 @@
11
package org.freedesktop.dbus.connections.config;
22

3+
import org.freedesktop.dbus.connections.shared.ExecutorNames;
34
import org.freedesktop.dbus.connections.shared.IThreadPoolRetryHandler;
45

6+
import java.util.EnumMap;
7+
import java.util.Map;
8+
import java.util.Optional;
9+
510
/**
611
* Bean which holds configuration for {@link org.freedesktop.dbus.connections.base.ReceivingService}.
712
*
813
* @author hypfvieh
914
* @since 4.2.0 - 2022-07-14
1015
*/
1116
public final class ReceivingServiceConfig {
12-
private int signalThreadPoolSize = 1;
13-
private int errorThreadPoolSize = 1;
14-
private int methodCallThreadPoolSize = 4;
15-
private int methodReturnThreadPoolSize = 1;
16-
private int signalThreadPriority = Thread.NORM_PRIORITY;
17-
private int methodCallThreadPriority = Thread.NORM_PRIORITY;
18-
private int errorThreadPriority = Thread.NORM_PRIORITY;
19-
private int methodReturnThreadPriority = Thread.NORM_PRIORITY;
17+
18+
private final Map<ExecutorNames, ThreadCfg> threadConfigs = new EnumMap<>(ExecutorNames.class);
2019

2120
private IThreadPoolRetryHandler retryHandler;
2221

2322
ReceivingServiceConfig() {
23+
threadConfigs.put(ExecutorNames.SIGNAL, new ThreadCfg(1, Thread.NORM_PRIORITY, false));
24+
threadConfigs.put(ExecutorNames.ERROR, new ThreadCfg(1, Thread.NORM_PRIORITY, false));
25+
threadConfigs.put(ExecutorNames.METHODCALL, new ThreadCfg(4, Thread.NORM_PRIORITY, false));
26+
threadConfigs.put(ExecutorNames.METHODRETURN, new ThreadCfg(1, Thread.NORM_PRIORITY, false));
2427
}
2528

2629
public int getSignalThreadPoolSize() {
27-
return signalThreadPoolSize;
30+
return threadConfigs.get(ExecutorNames.SIGNAL).poolSize();
2831
}
2932

3033
public int getErrorThreadPoolSize() {
31-
return errorThreadPoolSize;
34+
return threadConfigs.get(ExecutorNames.ERROR).poolSize();
3235
}
3336

3437
public int getMethodCallThreadPoolSize() {
35-
return methodCallThreadPoolSize;
38+
return threadConfigs.get(ExecutorNames.METHODCALL).poolSize();
3639
}
3740

3841
public int getMethodReturnThreadPoolSize() {
39-
return methodReturnThreadPoolSize;
42+
return threadConfigs.get(ExecutorNames.METHODRETURN).poolSize();
4043
}
4144

4245
public int getSignalThreadPriority() {
43-
return signalThreadPriority;
46+
return threadConfigs.get(ExecutorNames.SIGNAL).priority();
4447
}
4548

4649
public int getMethodCallThreadPriority() {
47-
return methodCallThreadPriority;
50+
return threadConfigs.get(ExecutorNames.METHODCALL).priority();
4851
}
4952

5053
public int getErrorThreadPriority() {
51-
return errorThreadPriority;
54+
return threadConfigs.get(ExecutorNames.ERROR).priority();
5255
}
5356

5457
public int getMethodReturnThreadPriority() {
55-
return methodReturnThreadPriority;
58+
return threadConfigs.get(ExecutorNames.METHODRETURN).priority();
5659
}
5760

5861
public IThreadPoolRetryHandler getRetryHandler() {
5962
return retryHandler;
6063
}
6164

65+
public boolean isSignalVirtualThreads() {
66+
return threadConfigs.get(ExecutorNames.SIGNAL).virtual();
67+
}
68+
69+
public boolean isErrorVirtualThreads() {
70+
return threadConfigs.get(ExecutorNames.ERROR).virtual();
71+
}
72+
73+
public boolean isMethodCallVirtualThreads() {
74+
return threadConfigs.get(ExecutorNames.METHODCALL).virtual();
75+
}
76+
77+
public boolean isMethodReturnVirtualThreads() {
78+
return threadConfigs.get(ExecutorNames.METHODRETURN).virtual();
79+
}
80+
81+
public boolean isVirtual(ExecutorNames _type) {
82+
return Optional.ofNullable(_type).map(p -> threadConfigs.get(p).virtual()).orElseThrow();
83+
}
84+
85+
public int getPoolSize(ExecutorNames _type) {
86+
return Optional.ofNullable(_type).map(p -> threadConfigs.get(p).poolSize()).orElseThrow();
87+
}
88+
89+
public int getPriority(ExecutorNames _type) {
90+
return Optional.ofNullable(_type).map(p -> threadConfigs.get(p).priority()).orElseThrow();
91+
}
92+
6293
void setSignalThreadPoolSize(int _signalThreadPoolSize) {
63-
signalThreadPoolSize = _signalThreadPoolSize;
94+
threadConfigs.compute(ExecutorNames.SIGNAL, (t, u) -> new ThreadCfg(_signalThreadPoolSize, u.priority(), u.virtual()));
6495
}
6596

6697
void setErrorThreadPoolSize(int _errorThreadPoolSize) {
67-
errorThreadPoolSize = _errorThreadPoolSize;
98+
threadConfigs.compute(ExecutorNames.ERROR, (t, u) -> new ThreadCfg(_errorThreadPoolSize, u.priority(), u.virtual()));
6899
}
69100

70101
void setMethodCallThreadPoolSize(int _methodCallThreadPoolSize) {
71-
methodCallThreadPoolSize = _methodCallThreadPoolSize;
102+
threadConfigs.compute(ExecutorNames.METHODCALL, (t, u) -> new ThreadCfg(_methodCallThreadPoolSize, u.priority(), u.virtual()));
72103
}
73104

74105
void setMethodReturnThreadPoolSize(int _methodReturnThreadPoolSize) {
75-
methodReturnThreadPoolSize = _methodReturnThreadPoolSize;
106+
threadConfigs.compute(ExecutorNames.METHODRETURN, (t, u) -> new ThreadCfg(_methodReturnThreadPoolSize, u.priority(), u.virtual()));
76107
}
77108

78109
void setSignalThreadPriority(int _signalThreadPriority) {
79-
signalThreadPriority = _signalThreadPriority;
110+
threadConfigs.compute(ExecutorNames.SIGNAL, (t, u) -> new ThreadCfg(u.poolSize(), _signalThreadPriority, u.virtual()));
80111
}
81112

82113
void setMethodCallThreadPriority(int _methodCallThreadPriority) {
83-
methodCallThreadPriority = _methodCallThreadPriority;
114+
threadConfigs.compute(ExecutorNames.METHODCALL, (t, u) -> new ThreadCfg(u.poolSize(), _methodCallThreadPriority, u.virtual()));
84115
}
85116

86117
void setErrorThreadPriority(int _errorThreadPriority) {
87-
errorThreadPriority = _errorThreadPriority;
118+
threadConfigs.compute(ExecutorNames.ERROR, (t, u) -> new ThreadCfg(u.poolSize(), _errorThreadPriority, u.virtual()));
88119
}
89120

90121
void setMethodReturnThreadPriority(int _methodReturnThreadPriority) {
91-
methodReturnThreadPriority = _methodReturnThreadPriority;
122+
threadConfigs.compute(ExecutorNames.METHODRETURN, (t, u) -> new ThreadCfg(u.poolSize(), _methodReturnThreadPriority, u.virtual()));
92123
}
93124

94125
void setRetryHandler(IThreadPoolRetryHandler _retryHandler) {
95126
retryHandler = _retryHandler;
96127
}
97128

129+
void setSignalVirtualThreads(boolean _signalVirtualThreads) {
130+
threadConfigs.compute(ExecutorNames.SIGNAL, (t, u) -> new ThreadCfg(u.poolSize(), u.priority(), _signalVirtualThreads));
131+
}
132+
133+
void setErrorVirtualThreads(boolean _errorVirtualThreads) {
134+
threadConfigs.compute(ExecutorNames.ERROR, (t, u) -> new ThreadCfg(u.poolSize(), u.priority(), _errorVirtualThreads));
135+
}
136+
137+
void setMethodCallVirtualThreads(boolean _methodCallVirtualThreads) {
138+
threadConfigs.compute(ExecutorNames.METHODCALL, (t, u) -> new ThreadCfg(u.poolSize(), u.priority(), _methodCallVirtualThreads));
139+
}
140+
141+
void setMethodReturnVirtualThreads(boolean _methodReturnVirtualThreads) {
142+
threadConfigs.compute(ExecutorNames.METHODRETURN, (t, u) -> new ThreadCfg(u.poolSize(), u.priority(), _methodReturnVirtualThreads));
143+
}
144+
145+
private record ThreadCfg(int poolSize, int priority, boolean virtual) {}
98146
}

0 commit comments

Comments
 (0)