Skip to content

Commit 6a7883d

Browse files
nicktindallnicktindall
authored
Reproduction of and fix for OpenHFT#81
* Reproduction of OpenHFT#81 * Make lock acquisition atomic, Fixes OpenHFT#81
1 parent 0bdef18 commit 6a7883d

11 files changed

Lines changed: 495 additions & 390 deletions

‎affinity/src/main/java/net/openhft/affinity/LockCheck.java‎

Lines changed: 12 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@
2828
/**
2929
* @author Rob Austin.
3030
*/
31-
enum LockCheck {
31+
public enum LockCheck {
3232
; // none
3333

3434
private static final Logger LOGGER = LoggerFactory.getLogger(LockCheck.class);
@@ -38,7 +38,7 @@ enum LockCheck {
3838

3939
private static final LockChecker lockChecker = FileLockBasedLockChecker.getInstance();
4040

41-
static long getPID() {
41+
public static long getPID() {
4242
String processName =
4343
java.lang.management.ManagementFactory.getRuntimeMXBean().getName();
4444
return Long.parseLong(processName.split("@")[0]);
@@ -52,30 +52,14 @@ public static boolean isCpuFree(int cpu) {
5252
if (!canOSSupportOperation())
5353
return true;
5454

55-
if (isLockFree(cpu)) {
56-
return true;
57-
} else {
58-
int currentProcess = 0;
59-
try {
60-
currentProcess = getProcessForCpu(cpu);
61-
} catch (RuntimeException | IOException e) {
62-
LOGGER.warn("Failed to determine process on cpu " + cpu, e);
63-
e.printStackTrace();
64-
return true;
65-
}
66-
if (!isProcessRunning(currentProcess)) {
67-
lockChecker.releaseLock(cpu);
68-
return true;
69-
}
70-
return false;
71-
}
55+
return isLockFree(cpu);
7256
}
7357

74-
static void replacePid(int cpu, long processID) throws IOException {
75-
storePid(processID, cpu);
58+
static boolean replacePid(int cpu, long processID) throws IOException {
59+
return storePid(processID, cpu);
7660
}
7761

78-
static boolean isProcessRunning(long pid) {
62+
public static boolean isProcessRunning(long pid) {
7963
if (canOSSupportOperation())
8064
return new File("/proc/" + pid).exists();
8165
else
@@ -86,17 +70,15 @@ static boolean isProcessRunning(long pid) {
8670
* stores the pid in a file, named by the core, the pid is written to the file with the date
8771
* below
8872
*/
89-
private synchronized static void storePid(long processID, int cpu) throws IOException {
90-
if (!lockChecker.obtainLock(cpu, Long.toString(processID))) {
91-
throw new IOException(String.format("Cannot obtain file lock for cpu %d", cpu));
92-
}
73+
private synchronized static boolean storePid(long processID, int cpu) throws IOException {
74+
return lockChecker.obtainLock(cpu, Long.toString(processID));
9375
}
9476

9577
private synchronized static boolean isLockFree(int id) {
9678
return lockChecker.isLockFree(id);
9779
}
9880

99-
static int getProcessForCpu(int core) throws IOException {
81+
public static int getProcessForCpu(int core) throws IOException {
10082
String meta = lockChecker.getMetaInfo(core);
10183

10284
if (meta != null && !meta.isEmpty()) {
@@ -109,15 +91,10 @@ static int getProcessForCpu(int core) throws IOException {
10991
return EMPTY_PID;
11092
}
11193

112-
static void updateCpu(int cpu) {
94+
static boolean updateCpu(int cpu) throws IOException {
11395
if (!canOSSupportOperation())
114-
return;
115-
try {
116-
replacePid(cpu, getPID());
117-
} catch (IOException e) {
118-
LOGGER.warn("Failed to update lock file for cpu " + cpu, e);
119-
e.printStackTrace();
120-
}
96+
return true;
97+
return replacePid(cpu, getPID());
12198
}
12299

123100
public static void releaseLock(int cpu) {

‎affinity/src/main/java/net/openhft/affinity/LockInventory.java‎

Lines changed: 30 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import org.slf4j.Logger;
2323
import org.slf4j.LoggerFactory;
2424

25+
import java.io.IOException;
2526
import java.util.NavigableMap;
2627
import java.util.TreeMap;
2728

@@ -69,9 +70,24 @@ private static boolean isAnyCpu(final int cpuId) {
6970
return cpuId == AffinityLock.ANY_CPU;
7071
}
7172

72-
private static void updateLockForCurrentThread(final boolean bind, final AffinityLock al, final boolean b) {
73-
al.assignCurrentThread(bind, b);
74-
LockCheck.updateCpu(al.cpuId());
73+
/**
74+
* Update the lock for the current thread
75+
*
76+
* @param bind Whether to also bind the thread to the core
77+
* @param al The lock to update
78+
* @param wholeCore Whether to bind the whole core
79+
* @return true if the lock was acquired, false otherwise
80+
*/
81+
private static boolean updateLockForCurrentThread(final boolean bind, final AffinityLock al, final boolean wholeCore) {
82+
try {
83+
if (LockCheck.updateCpu(al.cpuId())) {
84+
al.assignCurrentThread(bind, wholeCore);
85+
return true;
86+
}
87+
} catch (IOException e) {
88+
LOGGER.warn("Error occurred acquiring lock", e);
89+
}
90+
return false;
7591
}
7692

7793
public final synchronized CpuLayout getCpuLayout() {
@@ -106,8 +122,9 @@ public final synchronized AffinityLock acquireLock(boolean bind, int cpuId, Affi
106122
final boolean specificCpuRequested = !isAnyCpu(cpuId);
107123
if (specificCpuRequested && cpuId != 0) {
108124
final AffinityLock required = logicalCoreLocks[cpuId];
109-
if (required.canReserve(true) && anyStrategyMatches(cpuId, cpuId, strategies)) {
110-
updateLockForCurrentThread(bind, required, false);
125+
if (required.canReserve(true)
126+
&& anyStrategyMatches(cpuId, cpuId, strategies)
127+
&& updateLockForCurrentThread(bind, required, false)) {
111128
return required;
112129
}
113130
LOGGER.warn("Unable to acquire lock on CPU {} for thread {}, trying to find another CPU",
@@ -119,8 +136,9 @@ public final synchronized AffinityLock acquireLock(boolean bind, int cpuId, Affi
119136
// if you have only one core, this library is not appropriate in any case.
120137
for (int i = logicalCoreLocks.length - 1; i > 0; i--) {
121138
AffinityLock al = logicalCoreLocks[i];
122-
if (al.canReserve(false) && (isAnyCpu(cpuId) || strategy.matches(cpuId, al.cpuId()))) {
123-
updateLockForCurrentThread(bind, al, false);
139+
if (al.canReserve(false)
140+
&& (isAnyCpu(cpuId) || strategy.matches(cpuId, al.cpuId()))
141+
&& updateLockForCurrentThread(bind, al, false)) {
124142
return al;
125143
}
126144
}
@@ -136,8 +154,8 @@ public final synchronized AffinityLock tryAcquireLock(boolean bind, int cpuId) {
136154
return null;
137155

138156
final AffinityLock required = logicalCoreLocks[cpuId];
139-
if (required.canReserve(true)) {
140-
updateLockForCurrentThread(bind, required, false);
157+
if (required.canReserve(true)
158+
&& updateLockForCurrentThread(bind, required, false)) {
141159
return required;
142160
}
143161

@@ -156,8 +174,9 @@ public final synchronized AffinityLock acquireCore(boolean bind, int cpuId, Affi
156174
continue LOOP;
157175

158176
final AffinityLock al = als[0];
159-
updateLockForCurrentThread(bind, al, true);
160-
return al;
177+
if (updateLockForCurrentThread(bind, al, true)) {
178+
return al;
179+
}
161180
}
162181
}
163182

‎affinity/src/main/java/net/openhft/affinity/lockchecker/FileBasedLockChecker.java‎

Lines changed: 0 additions & 83 deletions
This file was deleted.

0 commit comments

Comments
 (0)