Skip to content

Commit 5d7290d

Browse files
belabanbelaban
committed
Made MessageCache add/drain/iteration thread-safe (https://issues.redhat.com/browse/JGRP-2747)
1 parent 8fdac01 commit 5d7290d

File tree

3 files changed

+118
-48
lines changed

3 files changed

+118
-48
lines changed

‎src/org/jgroups/protocols/UNICAST3.java

Lines changed: 7 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -139,6 +139,8 @@ public class UNICAST3 extends Protocol implements AgeOutCache.Handler<Address> {
139139
/** Keep track of when a SEND_FIRST_SEQNO message was sent to a given sender */
140140
protected ExpiryCache<Address> last_sync_sent=null;
141141

142+
// Queues messages until a {@link ReceiverEntry} has been created. Queued messages are then removed from
143+
// the cache and added to the ReceiverEntry
142144
protected final MessageCache msg_cache=new MessageCache();
143145

144146
protected static final Message DUMMY_OOB_MSG=new Message().setFlag(Message.Flag.OOB);
@@ -485,7 +487,7 @@ public void up(MessageBatch batch) {
485487
if(hdr.first)
486488
entry=getReceiverEntry(sender, hdr.seqno(), hdr.first, hdr.connId());
487489
else if(entry == null) {
488-
msg_cache.cache(sender, msg);
490+
msg_cache.add(sender, msg);
489491
log.trace("%s: cached %s#%d", local_addr, sender, hdr.seqno());
490492
}
491493
}
@@ -495,7 +497,7 @@ else if(entry == null) {
495497
sendRequestForFirstSeqno(sender);
496498
else {
497499
if(!msg_cache.isEmpty()) { // quick and dirty check
498-
List<Message> queued_msgs=msg_cache.drain(sender);
500+
Collection<Message> queued_msgs=msg_cache.drain(sender);
499501
if(queued_msgs != null)
500502
addQueuedMessages(sender, entry, queued_msgs);
501503
}
@@ -745,12 +747,12 @@ public void expired(Address key) {
745747
protected void handleDataReceived(final Address sender, long seqno, short conn_id, boolean first, final Message msg) {
746748
ReceiverEntry entry=getReceiverEntry(sender, seqno, first, conn_id);
747749
if(entry == null) {
748-
msg_cache.cache(sender, msg);
750+
msg_cache.add(sender, msg);
749751
log.trace("%s: cached %s#%d", local_addr, sender, seqno);
750752
return;
751753
}
752754
if(!msg_cache.isEmpty()) { // quick and dirty check
753-
List<Message> queued_msgs=msg_cache.drain(sender);
755+
Collection<Message> queued_msgs=msg_cache.drain(sender);
754756
if(queued_msgs != null)
755757
addQueuedMessages(sender, entry, queued_msgs);
756758
}
@@ -781,7 +783,7 @@ protected void addMessage(ReceiverEntry entry, Address sender, long seqno, Messa
781783
}
782784
}
783785

784-
protected void addQueuedMessages(final Address sender, final ReceiverEntry entry, List<Message> queued_msgs) {
786+
protected void addQueuedMessages(final Address sender, final ReceiverEntry entry, Collection<Message> queued_msgs) {
785787
for(Message msg: queued_msgs) {
786788
UnicastHeader3 hdr=msg.getHeader(this.id);
787789
if(hdr.conn_id != entry.conn_id) {
@@ -1444,47 +1446,4 @@ public String toString() {
14441446
}
14451447
}
14461448

1447-
/**
1448-
* Used to queue messages until a {@link ReceiverEntry} has been created. Queued messages are then removed from
1449-
* the cache and added to the ReceiverEntry
1450-
*/
1451-
protected class MessageCache {
1452-
private final Map<Address,List<Message>> map=new ConcurrentHashMap<>();
1453-
private volatile boolean is_empty=true;
1454-
1455-
protected MessageCache cache(Address sender, Message msg) {
1456-
List<Message> list=map.computeIfAbsent(sender, addr -> new ArrayList<>());
1457-
list.add(msg);
1458-
is_empty=false;
1459-
return this;
1460-
}
1461-
1462-
protected List<Message> drain(Address sender) {
1463-
List<Message> list=map.remove(sender);
1464-
if(map.isEmpty())
1465-
is_empty=true;
1466-
return list;
1467-
}
1468-
1469-
protected MessageCache clear() {
1470-
map.clear();
1471-
is_empty=true;
1472-
return this;
1473-
}
1474-
1475-
/** Returns a count of all messages */
1476-
protected int size() {
1477-
return map.values().stream().mapToInt(Collection::size).sum();
1478-
}
1479-
1480-
protected boolean isEmpty() {
1481-
return is_empty;
1482-
}
1483-
1484-
public String toString() {
1485-
return String.format("%d message(s)", size());
1486-
}
1487-
}
1488-
1489-
14901449
}

‎src/org/jgroups/util/MessageCache.java

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
1+
package org.jgroups.util;
2+
3+
import org.jgroups.Address;
4+
import org.jgroups.Message;
5+
6+
import java.util.Collection;
7+
import java.util.Map;
8+
import java.util.Queue;
9+
import java.util.concurrent.ConcurrentHashMap;
10+
import java.util.concurrent.ConcurrentLinkedQueue;
11+
12+
/**
13+
* A cache associating members and messages
14+
* @author Bela Ban
15+
* @since 5.3.2
16+
*/
17+
public class MessageCache {
18+
protected final Map<Address,Queue<Message>> map=new ConcurrentHashMap<>();
19+
protected volatile boolean is_empty=true;
20+
21+
public MessageCache add(Address sender, Message msg) {
22+
Queue<Message> list=map.computeIfAbsent(sender, addr -> new ConcurrentLinkedQueue<>());
23+
list.add(msg);
24+
is_empty=false;
25+
return this;
26+
}
27+
28+
public Collection<Message> drain(Address sender) {
29+
if(sender == null)
30+
return null;
31+
Queue<Message> queue=map.remove(sender);
32+
if(map.isEmpty())
33+
is_empty=true;
34+
return queue;
35+
}
36+
37+
public MessageCache clear() {
38+
map.clear();
39+
is_empty=true;
40+
return this;
41+
}
42+
43+
/** Returns a count of all messages */
44+
public int size() {
45+
return map.values().stream().mapToInt(Collection::size).sum();
46+
}
47+
48+
public boolean isEmpty() {
49+
return is_empty;
50+
}
51+
52+
public String toString() {
53+
return String.format("%d message(s)", size());
54+
}
55+
}

‎tests/junit-functional/org/jgroups/tests/MessageCacheTest.java

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
package org.jgroups.tests;
2+
3+
import org.jgroups.Address;
4+
import org.jgroups.Global;
5+
import org.jgroups.Message;
6+
import org.jgroups.ObjectMessage;
7+
import org.jgroups.util.MessageCache;
8+
import org.jgroups.util.Util;
9+
import org.testng.annotations.BeforeMethod;
10+
import org.testng.annotations.Test;
11+
12+
import java.util.Collection;
13+
14+
/**
15+
* Tests {@link org.jgroups.util.MessageCache}
16+
* @author Bela Ban
17+
* @since 5.3.2
18+
*/
19+
@Test(groups= Global.FUNCTIONAL,singleThreaded=true)
20+
public class MessageCacheTest {
21+
protected MessageCache cache;
22+
protected static final Address A=Util.createRandomAddress("A"), B=Util.createRandomAddress("B"),
23+
C=Util.createRandomAddress("C");
24+
25+
@BeforeMethod protected void setup() {
26+
cache=new MessageCache();
27+
}
28+
29+
public void testCreation() {
30+
assert cache.isEmpty();
31+
}
32+
33+
public void testAdd() {
34+
for(int i=1; i <= 5; i++) {
35+
cache.add(A, new ObjectMessage(A, i));
36+
cache.add(B, new ObjectMessage(B, i+10));
37+
}
38+
assert !cache.isEmpty();
39+
assert cache.size() == 10;
40+
}
41+
42+
public void testDrain() {
43+
testAdd();
44+
Collection<Message> list=cache.drain(null);
45+
assert list == null;
46+
list=cache.drain(C);
47+
assert list == null;
48+
list=cache.drain(B);
49+
assert list.size() == 5;
50+
assert cache.size() == 5;
51+
assert !cache.isEmpty();
52+
list=cache.drain(A);
53+
assert list.size() == 5;
54+
assert cache.isEmpty();
55+
}
56+
}

0 commit comments

Comments
 (0)