1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.newsclub.net.unix.memory;
19
20 import java.io.IOException;
21 import java.lang.foreign.MemorySegment;
22 import java.lang.foreign.ValueLayout;
23 import java.lang.foreign.ValueLayout.OfInt;
24 import java.lang.invoke.VarHandle;
25 import java.util.concurrent.atomic.AtomicBoolean;
26
27 import org.newsclub.net.unix.MemoryImplUtilInternal;
28
29 final class Futex32 implements Futex {
30 private static final VarHandle VH_INT = ValueLayout.OfInt.JAVA_INT.varHandle();
31
32 private static final int MUTEX_UNLOCKED = 0;
33 private static final int MUTEX_LOCKED = 1;
34 private static final int MUTEX_LOCKED_WAITING = 2;
35
36 private final MemorySegment ms;
37 private final boolean zeroOnClose;
38 private final AtomicBoolean closed = new AtomicBoolean(false);
39
40 Futex32(MemorySegment ms, boolean zeroOnClose) throws IOException {
41 if (!ms.isAccessibleBy(Thread.currentThread())) {
42 throw new IllegalStateException("Cannot access this MemorySegment from the current thread");
43 }
44 if ((ms.address() & 3) != 0) {
45 throw new IOException("Not aligned");
46 }
47 if (ms.byteSize() != SharedMemory.FUTEX32_SEGMENT_SIZE) {
48 throw new IOException("MemorySegment must be exactly 4 bytes long");
49 }
50 this.ms = ms;
51 this.zeroOnClose = zeroOnClose;
52
53
54 SharedMemory.UTIL.madvise(ms.address(), SharedMemory.FUTEX32_SEGMENT_SIZE,
55 MemoryImplUtilInternal.MADV_WILLNEED, true);
56 }
57
58 @Override
59 public void close() {
60 if (!closed.getAndSet(true)) {
61 if (zeroOnClose && ms != null) {
62 ms.set(OfInt.JAVA_INT, 0, 0);
63 }
64 }
65 }
66
67 @Override
68 public boolean tryWait(int ifValue, int timeoutMillis) throws IOException {
69 return SharedMemory.UTIL.futexWait(ms.address(), ifValue, timeoutMillis);
70 }
71
72 @Override
73 public boolean tryWake(boolean wakeAll) throws IOException {
74 return SharedMemory.UTIL.futexWake(ms.address(), wakeAll);
75 }
76
77 @Override
78 public boolean isClosed() {
79 return closed.get();
80 }
81
82 MemorySegment getMemorySegment() {
83 return ms;
84 }
85
86 @Override
87 public boolean isInterProcess() {
88 return SharedMemory.UTIL.futexIsInterProcess();
89 }
90
91 private final class Mutex32 implements SharedMutex {
92 @Override
93 public void close() throws IOException {
94 Futex32.this.close();
95 }
96
97 @Override
98 public boolean tryLock(int timeoutMillis) throws IOException {
99 int w = (int) VH_INT.compareAndExchange(ms, 0, MUTEX_UNLOCKED, MUTEX_LOCKED);
100 if (w == MUTEX_UNLOCKED) {
101 return true;
102 }
103
104 if (w != MUTEX_LOCKED_WAITING) {
105 w = (int) VH_INT.getAndSet(ms, 0, MUTEX_LOCKED_WAITING);
106 }
107
108 if (w == MUTEX_UNLOCKED) {
109 return true;
110 }
111
112 long start = System.currentTimeMillis();
113 while (!Thread.interrupted()) {
114 if (!Futex32.this.tryWait(MUTEX_LOCKED_WAITING, timeoutMillis)) {
115 if (isClosed()) {
116 return false;
117 }
118 }
119 w = (int) VH_INT.getAndSet(ms, 0, MUTEX_LOCKED_WAITING);
120 if (w == MUTEX_UNLOCKED) {
121 return true;
122 }
123 if (timeoutMillis != 0) {
124 timeoutMillis -= (int) (System.currentTimeMillis() - start);
125 if (timeoutMillis <= 0) {
126 return false;
127 }
128 }
129 }
130 return false;
131 }
132
133 @Override
134 public void unlock() throws IOException {
135 int c = (int) VH_INT.getAndAdd(ms, 0, -1);
136 switch (c) {
137 case MUTEX_UNLOCKED:
138 case MUTEX_LOCKED:
139 break;
140 default:
141 VH_INT.set(ms, 0, MUTEX_UNLOCKED);
142 Futex32.this.tryWake(false);
143 }
144 }
145
146 @Override
147 public boolean isReentrant() {
148 return false;
149 }
150
151 @Override
152 public boolean isInterProcess() {
153 return Futex32.this.isInterProcess();
154 }
155 }
156
157 SharedMutex mutex() {
158 return new Mutex32();
159 }
160 }