Skip to content

Commit 2cd6a53

Browse files
menachermenacher
committed
Initial commit of Lane Feature
1 parent 75a97fc commit 2cd6a53

6 files changed

Lines changed: 224 additions & 2 deletions

File tree

‎jetserver/src/main/java/org/menacheri/jetserver/concurrent/DefaultLane.java‎

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,65 @@
1+
package org.menacheri.jetserver.concurrent;
2+
3+
import java.util.concurrent.ExecutorService;
4+
5+
public class DefaultLane implements Lane<String,ExecutorService>
6+
{
7+
8+
final String laneName;
9+
final ExecutorService exec;
10+
11+
public DefaultLane(String threadName, ExecutorService exec)
12+
{
13+
this.laneName = threadName;
14+
this.exec = exec;
15+
}
16+
17+
@Override
18+
public boolean isOnSameLane(String currentThread)
19+
{
20+
return currentThread.equals(laneName);
21+
}
22+
23+
@Override
24+
public ExecutorService getUnderlyingLane()
25+
{
26+
return exec;
27+
}
28+
29+
@Override
30+
public int hashCode()
31+
{
32+
final int prime = 31;
33+
int result = 1;
34+
result = prime * result
35+
+ ((laneName == null) ? 0 : laneName.hashCode());
36+
return result;
37+
}
38+
39+
@Override
40+
public boolean equals(Object obj)
41+
{
42+
if (this == obj)
43+
return true;
44+
if (obj == null)
45+
return false;
46+
if (getClass() != obj.getClass())
47+
return false;
48+
DefaultLane other = (DefaultLane) obj;
49+
if (laneName == null)
50+
{
51+
if (other.laneName != null)
52+
return false;
53+
}
54+
else if (!laneName.equals(other.laneName))
55+
return false;
56+
return true;
57+
}
58+
59+
@Override
60+
public String getId()
61+
{
62+
return laneName;
63+
}
64+
65+
}

‎jetserver/src/main/java/org/menacheri/jetserver/concurrent/Lane.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
package org.menacheri.jetserver.concurrent;
2+
3+
public interface Lane<I,T>
4+
{
5+
boolean isOnSameLane(I currentLane);
6+
I getId();
7+
T getUnderlyingLane();
8+
}

‎jetserver/src/main/java/org/menacheri/jetserver/concurrent/LaneStrategy.java‎

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,69 @@
1+
package org.menacheri.jetserver.concurrent;
2+
3+
import java.util.ArrayList;
4+
import java.util.HashMap;
5+
import java.util.List;
6+
import java.util.Map;
7+
import java.util.concurrent.ExecutorService;
8+
import java.util.concurrent.atomic.AtomicInteger;
9+
10+
import org.menacheri.jetserver.app.GameRoom;
11+
12+
public interface LaneStrategy
13+
{
14+
<I, T, O> Lane<I, T> chooseLane(O group);
15+
16+
public enum LaneStrategies implements LaneStrategy
17+
{
18+
19+
ROUND_ROBIN
20+
{
21+
final AtomicInteger currentLane = new AtomicInteger(0);
22+
23+
@SuppressWarnings("unchecked")
24+
@Override
25+
public Lane<String, ExecutorService> chooseLane(Object group)
26+
{
27+
synchronized (currentLane)
28+
{
29+
if (currentLane.get() == lanes.size())
30+
{
31+
currentLane.set(0);
32+
}
33+
}
34+
return lanes.get(currentLane.getAndIncrement());
35+
}
36+
},
37+
GAME_ROOM
38+
{
39+
@SuppressWarnings("rawtypes")
40+
Map laneRoomMap = new HashMap();
41+
42+
@SuppressWarnings({ "hiding", "unchecked" })
43+
@Override
44+
public <String, ExecutorService, GameRoom> Lane<String, ExecutorService> chooseLane(
45+
GameRoom group)
46+
{
47+
synchronized (laneRoomMap)
48+
{
49+
if (laneRoomMap.isEmpty())
50+
{
51+
for (@SuppressWarnings("rawtypes")
52+
Lane lane : lanes)
53+
{
54+
List<GameRoom> roomList = new ArrayList<GameRoom>();
55+
laneRoomMap.put(lane, roomList);
56+
}
57+
}
58+
}
59+
60+
return null;
61+
}
62+
63+
};
64+
65+
final List<Lane<String, ExecutorService>> lanes = Lanes.LANES
66+
.getJetLanes();
67+
}
68+
69+
}

‎jetserver/src/main/java/org/menacheri/jetserver/concurrent/Lanes.java‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
package org.menacheri.jetserver.concurrent;
2+
3+
import java.util.ArrayList;
4+
import java.util.List;
5+
import java.util.concurrent.ExecutorService;
6+
import java.util.concurrent.ThreadFactory;
7+
8+
public enum Lanes
9+
{
10+
LANES;
11+
final String serverCores = System.getProperty("jet.cores");
12+
final int numOfCores;
13+
final List<Lane<String, ExecutorService>> jetLanes;
14+
15+
Lanes()
16+
{
17+
int cores = 1;
18+
if (null != serverCores)
19+
{
20+
try
21+
{
22+
cores = Integer.parseInt(serverCores);
23+
}
24+
catch (NumberFormatException e)
25+
{
26+
// ignore;
27+
}
28+
}
29+
numOfCores = cores;
30+
jetLanes = new ArrayList<Lane<String, ExecutorService>>();
31+
ThreadFactory threadFactory = new NamedThreadFactory("Lane");
32+
for (int i = 0; i <= cores; i++)
33+
{
34+
DefaultLane defaultLane = new DefaultLane("Lane[" + (i + 1) + "]",
35+
ManagedExecutor.newSingleThreadExecutor(threadFactory));
36+
jetLanes.add(defaultLane);
37+
}
38+
}
39+
40+
public List<Lane<String, ExecutorService>> getJetLanes()
41+
{
42+
return jetLanes;
43+
}
44+
45+
public int getNumOfCores()
46+
{
47+
return numOfCores;
48+
}
49+
}

‎jetserver/src/main/java/org/menacheri/jetserver/concurrent/ManagedExecutor.java‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
package org.menacheri.jetserver.concurrent;
2+
3+
import java.util.ArrayList;
4+
import java.util.List;
5+
import java.util.concurrent.ExecutorService;
6+
import java.util.concurrent.Executors;
7+
import java.util.concurrent.ThreadFactory;
8+
9+
public class ManagedExecutor
10+
{
11+
public static final List<ExecutorService> EXECUTOR_SERVICES = new ArrayList<ExecutorService>();
12+
13+
static{
14+
Runtime.getRuntime().addShutdownHook(new Thread(){
15+
@Override
16+
public void run()
17+
{
18+
for(ExecutorService service: EXECUTOR_SERVICES){
19+
service.shutdown();
20+
}
21+
}
22+
});
23+
}
24+
25+
public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory)
26+
{
27+
final ExecutorService exec = Executors.newSingleThreadExecutor(threadFactory);
28+
EXECUTOR_SERVICES.add(exec);
29+
return exec;
30+
}
31+
}

‎jetserver/src/main/java/org/menacheri/jetserver/concurrent/NamedThreadFactory.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,8 +5,8 @@
55

66
public class NamedThreadFactory implements ThreadFactory
77
{
8-
private AtomicInteger counter = new AtomicInteger(0);
9-
private String name;
8+
private static AtomicInteger counter = new AtomicInteger(1);
9+
private String name = "Lane";
1010
private boolean daemon;
1111
private int priority;
1212

0 commit comments

Comments
 (0)