홍길동 / onos

Go to a project
Toggle navigation Toggle navigation pinning
Madan Jampani
Committed by Gerrit Code Review
0583387a 1 parent e18b5340

Implementation of Hybrid Logical Clock Service. 

Change-Id: I52e231433d044f9e6156db7e28bde9fd199118e8
Inline Side-by-side
Showing 10 changed files with 376 additions and 4 deletions
1 +/*
2 + * Copyright 2016-present Open Networking Laboratory
3 + *
4 + * Licensed under the Apache License, Version 2.0 (the "License");
5 + * you may not use this file except in compliance with the License.
6 + * You may obtain a copy of the License at
7 + *
8 + * http://www.apache.org/licenses/LICENSE-2.0
9 + *
10 + * Unless required by applicable law or agreed to in writing, software
11 + * distributed under the License is distributed on an "AS IS" BASIS,
12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 + * See the License for the specific language governing permissions and
14 + * limitations under the License.
15 + */
16 +package org.onosproject.core;
17 +
18 +import org.onosproject.store.service.WallClockTimestamp;
19 +
20 +/**
21 + * The <a href="http://www.cse.buffalo.edu/tech-reports/2014-04.pdf">hybrid logical time</a> keeper service.
22 + */
23 +public interface HybridLogicalClockService {
24 +
25 + /**
26 + * Returns the current hybrid logical time.
27 + * @return current hybrid logical time
28 + */
29 + HybridLogicalTime timeNow();
30 +
31 + /**
32 + * Records a (receive) event and accordingly makes adjustments to the hybrid logical time.
33 + * @param time received event time
34 + */
35 + void recordEventTime(HybridLogicalTime time);
36 +
37 + /**
38 + * Returns the current time derived from the hybrid logical time.
39 + * @return current system time
40 + */
41 + default long now() {
42 + return timeNow().time();
43 + }
44 +
45 + /**
46 + * Returns the current time as a {@code WallClockTimestamp}.
47 + * @return wall clock timestamp
48 + */
49 + default WallClockTimestamp wallClockTimestamp() {
50 + return new WallClockTimestamp(now());
51 + }
52 +}
1 +/*
2 + * Copyright 2016-present Open Networking Laboratory
3 + *
4 + * Licensed under the Apache License, Version 2.0 (the "License");
5 + * you may not use this file except in compliance with the License.
6 + * You may obtain a copy of the License at
7 + *
8 + * http://www.apache.org/licenses/LICENSE-2.0
9 + *
10 + * Unless required by applicable law or agreed to in writing, software
11 + * distributed under the License is distributed on an "AS IS" BASIS,
12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 + * See the License for the specific language governing permissions and
14 + * limitations under the License.
15 + */
16 +package org.onosproject.core;
17 +
18 +import com.google.common.base.MoreObjects;
19 +
20 +/**
21 + * Time provided by a Hybrid Logical Clock described in
22 + * this <a href="http://www.cse.buffalo.edu/tech-reports/2014-04.pdf">paper</a>.
23 + */
24 +public class HybridLogicalTime {
25 + private final long logicalTime;
26 + private final long logicalCounter;
27 +
28 + public HybridLogicalTime(long logicalTime, long logicalCounter) {
29 + this.logicalTime = logicalTime;
30 + this.logicalCounter = logicalCounter;
31 + }
32 +
33 + /**
34 + * Returns the logical time component of a HLT.
35 + * @return logical time
36 + */
37 + public long logicalTime() {
38 + return logicalTime;
39 + }
40 +
41 + /**
42 + * Returns the logical counter component of a HLT.
43 + * @return logical counter
44 + */
45 + public long logicalCounter() {
46 + return logicalCounter;
47 + }
48 +
49 + /**
50 + * Returns the real system time represented by this HLT.
51 + * @return real system time
52 + */
53 + public long time() {
54 + return (logicalTime >> 16 << 16) | (logicalCounter << 48 >> 48);
55 + }
56 +
57 + @Override
58 + public String toString() {
59 + return MoreObjects.toStringHelper(getClass())
60 + .add("logicalTime", logicalTime)
61 + .add("logicalCounter", logicalCounter)
62 + .toString();
63 + }
64 +}
1 +/*
2 + * Copyright 2016-present Open Networking Laboratory
3 + *
4 + * Licensed under the Apache License, Version 2.0 (the "License");
5 + * you may not use this file except in compliance with the License.
6 + * You may obtain a copy of the License at
7 + *
8 + * http://www.apache.org/licenses/LICENSE-2.0
9 + *
10 + * Unless required by applicable law or agreed to in writing, software
11 + * distributed under the License is distributed on an "AS IS" BASIS,
12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 + * See the License for the specific language governing permissions and
14 + * limitations under the License.
15 + */
16 +
17 +package org.onosproject.core.impl;
18 +
19 +import static org.slf4j.LoggerFactory.getLogger;
20 +
21 +import java.util.function.Supplier;
22 +
23 +import org.apache.felix.scr.annotations.Activate;
24 +import org.apache.felix.scr.annotations.Component;
25 +import org.apache.felix.scr.annotations.Deactivate;
26 +import org.apache.felix.scr.annotations.Service;
27 +import org.onosproject.core.HybridLogicalClockService;
28 +import org.onosproject.core.HybridLogicalTime;
29 +import org.slf4j.Logger;
30 +
31 +/**
32 + * Implementation of {@link HybridLogicalClockService}.
33 + * <p>
34 + * Implementation is based on HLT <a href="http://www.cse.buffalo.edu/tech-reports/2014-04.pdf">paper</a>.
35 + */
36 +@Component(immediate = true)
37 +@Service
38 +public class HybridLogicalClockManager implements HybridLogicalClockService {
39 +
40 + private final Logger log = getLogger(getClass());
41 +
42 + protected Supplier<Long> physicalTimeSource = System::currentTimeMillis;
43 +
44 + private long logicalTime = 0;
45 + private long logicalCounter = 0;
46 +
47 + @Activate
48 + public void activate() {
49 + log.info("Started");
50 + }
51 +
52 + @Deactivate
53 + public void deactivate() {
54 + log.info("Stopped");
55 + }
56 +
57 + @Override
58 + public synchronized HybridLogicalTime timeNow() {
59 + final long oldLogicalTime = logicalTime;
60 + logicalTime = Math.max(oldLogicalTime, physicalTimeSource.get());
61 + if (logicalTime == oldLogicalTime) {
62 + logicalCounter++;
63 + } else {
64 + logicalCounter = 0;
65 + }
66 + return new HybridLogicalTime(logicalTime, logicalCounter);
67 + }
68 +
69 + @Override
70 + public synchronized void recordEventTime(HybridLogicalTime eTime) {
71 + final long oldLogicalTime = logicalTime;
72 + logicalTime = Math.max(oldLogicalTime, Math.max(eTime.logicalTime(), physicalTimeSource.get()));
73 + if (logicalTime == oldLogicalTime && oldLogicalTime == eTime.logicalTime()) {
74 + logicalCounter = Math.max(logicalCounter, eTime.logicalCounter()) + 1;
75 + } else if (logicalTime == oldLogicalTime) {
76 + logicalCounter++;
77 + } else if (logicalTime == eTime.logicalTime()) {
78 + logicalCounter = eTime.logicalCounter() + 1;
79 + } else {
80 + logicalCounter = 0;
81 + }
82 + }
83 +
84 + protected long logicalTime() {
85 + return logicalTime;
86 + }
87 +
88 + protected long logicalCounter() {
89 + return logicalCounter;
90 + }
91 +}
1 +/*
2 + * Copyright 2016-present Open Networking Laboratory
3 + *
4 + * Licensed under the Apache License, Version 2.0 (the "License");
5 + * you may not use this file except in compliance with the License.
6 + * You may obtain a copy of the License at
7 + *
8 + * http://www.apache.org/licenses/LICENSE-2.0
9 + *
10 + * Unless required by applicable law or agreed to in writing, software
11 + * distributed under the License is distributed on an "AS IS" BASIS,
12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 + * See the License for the specific language governing permissions and
14 + * limitations under the License.
15 + */
16 +package org.onosproject.core.impl;
17 +
18 +import java.util.concurrent.atomic.AtomicLong;
19 +import java.util.function.Supplier;
20 +
21 +import org.junit.Assert;
22 +import org.junit.Test;
23 +import org.onosproject.core.HybridLogicalTime;
24 +
25 +/**
26 + * Unit tests for {@link HybridLogicalClockManager}.
27 + */
28 +public class HybridLogicalClockManagerTest {
29 +
30 + @Test
31 + public void testLocalEvents() {
32 + AtomicLong ticker = new AtomicLong();
33 + Supplier<Long> ptSource = ticker::get;
34 + HybridLogicalClockManager clockManager = new HybridLogicalClockManager();
35 + clockManager.physicalTimeSource = ptSource;
36 +
37 + HybridLogicalTime time1 = clockManager.timeNow();
38 + Assert.assertEquals(0, time1.logicalTime());
39 + Assert.assertEquals(1, time1.logicalCounter());
40 +
41 + HybridLogicalTime time2 = clockManager.timeNow();
42 + Assert.assertEquals(0, time2.logicalTime());
43 + Assert.assertEquals(2, time2.logicalCounter());
44 +
45 + ticker.incrementAndGet();
46 +
47 + HybridLogicalTime time3 = clockManager.timeNow();
48 + Assert.assertEquals(1, time3.logicalTime());
49 + Assert.assertEquals(0, time3.logicalCounter());
50 +
51 + HybridLogicalTime time4 = clockManager.timeNow();
52 + Assert.assertEquals(1, time4.logicalTime());
53 + Assert.assertEquals(1, time4.logicalCounter());
54 + }
55 +
56 + @Test
57 + public void testReceiveEvents() {
58 + AtomicLong ticker = new AtomicLong(1);
59 + Supplier<Long> ptSource = ticker::get;
60 + HybridLogicalClockManager clockManager = new HybridLogicalClockManager();
61 + clockManager.physicalTimeSource = ptSource;
62 +
63 + HybridLogicalTime time1 = clockManager.timeNow();
64 + Assert.assertEquals(1, time1.logicalTime());
65 + Assert.assertEquals(0, time1.logicalCounter());
66 +
67 + HybridLogicalTime eventTime1 = new HybridLogicalTime(1, 0);
68 + clockManager.recordEventTime(eventTime1);
69 +
70 + Assert.assertEquals(1, clockManager.logicalTime());
71 + Assert.assertEquals(1, clockManager.logicalCounter());
72 +
73 + HybridLogicalTime eventTime2 = new HybridLogicalTime(2, 0);
74 + clockManager.recordEventTime(eventTime2);
75 +
76 + Assert.assertEquals(2, clockManager.logicalTime());
77 + Assert.assertEquals(1, clockManager.logicalCounter());
78 +
79 + HybridLogicalTime eventTime3 = new HybridLogicalTime(2, 2);
80 + clockManager.recordEventTime(eventTime3);
81 +
82 + Assert.assertEquals(2, clockManager.logicalTime());
83 + Assert.assertEquals(3, clockManager.logicalCounter());
84 +
85 + HybridLogicalTime eventTime4 = new HybridLogicalTime(2, 1);
86 + clockManager.recordEventTime(eventTime4);
87 +
88 + Assert.assertEquals(2, clockManager.logicalTime());
89 + Assert.assertEquals(4, clockManager.logicalCounter());
90 +
91 + ticker.set(4);
92 +
93 + HybridLogicalTime eventTime5 = new HybridLogicalTime(3, 0);
94 + clockManager.recordEventTime(eventTime5);
95 +
96 + Assert.assertEquals(4, clockManager.logicalTime());
97 + Assert.assertEquals(0, clockManager.logicalCounter());
98 + }
99 +}
...@@ -20,6 +20,8 @@ package org.onosproject.store.cluster.messaging.impl; ...@@ -20,6 +20,8 @@ package org.onosproject.store.cluster.messaging.impl;
20 */ 20 */
21 public enum DecoderState { 21 public enum DecoderState {
22 READ_MESSAGE_PREAMBLE, 22 READ_MESSAGE_PREAMBLE,
23 + READ_LOGICAL_TIME,
24 + READ_LOGICAL_COUNTER,
23 READ_MESSAGE_ID, 25 READ_MESSAGE_ID,
24 READ_SENDER_IP_VERSION, 26 READ_SENDER_IP_VERSION,
25 READ_SENDER_IP, 27 READ_SENDER_IP,
......
...@@ -16,7 +16,9 @@ ...@@ -16,7 +16,9 @@
16 package org.onosproject.store.cluster.messaging.impl; 16 package org.onosproject.store.cluster.messaging.impl;
17 17
18 import com.google.common.base.MoreObjects; 18 import com.google.common.base.MoreObjects;
19 +
19 import org.onlab.util.ByteArraySizeHashPrinter; 20 import org.onlab.util.ByteArraySizeHashPrinter;
21 +import org.onosproject.core.HybridLogicalTime;
20 import org.onosproject.store.cluster.messaging.Endpoint; 22 import org.onosproject.store.cluster.messaging.Endpoint;
21 23
22 /** 24 /**
...@@ -55,18 +57,31 @@ public final class InternalMessage { ...@@ -55,18 +57,31 @@ public final class InternalMessage {
55 } 57 }
56 58
57 private final int preamble; 59 private final int preamble;
60 + private final HybridLogicalTime time;
58 private final long id; 61 private final long id;
59 private final Endpoint sender; 62 private final Endpoint sender;
60 private final String type; 63 private final String type;
61 private final byte[] payload; 64 private final byte[] payload;
62 private final Status status; 65 private final Status status;
63 66
64 - public InternalMessage(int preamble, long id, Endpoint sender, String type, byte[] payload) { 67 + public InternalMessage(int preamble,
65 - this(preamble, id, sender, type, payload, Status.OK); 68 + HybridLogicalTime time,
69 + long id,
70 + Endpoint sender,
71 + String type,
72 + byte[] payload) {
73 + this(preamble, time, id, sender, type, payload, Status.OK);
66 } 74 }
67 75
68 - public InternalMessage(int preamble, long id, Endpoint sender, String type, byte[] payload, Status status) { 76 + public InternalMessage(int preamble,
77 + HybridLogicalTime time,
78 + long id,
79 + Endpoint sender,
80 + String type,
81 + byte[] payload,
82 + Status status) {
69 this.preamble = preamble; 83 this.preamble = preamble;
84 + this.time = time;
70 this.id = id; 85 this.id = id;
71 this.sender = sender; 86 this.sender = sender;
72 this.type = type; 87 this.type = type;
...@@ -74,6 +89,10 @@ public final class InternalMessage { ...@@ -74,6 +89,10 @@ public final class InternalMessage {
74 this.status = status; 89 this.status = status;
75 } 90 }
76 91
92 + public HybridLogicalTime time() {
93 + return time;
94 + }
95 +
77 public int preamble() { 96 public int preamble() {
78 return preamble; 97 return preamble;
79 } 98 }
...@@ -101,6 +120,7 @@ public final class InternalMessage { ...@@ -101,6 +120,7 @@ public final class InternalMessage {
101 @Override 120 @Override
102 public String toString() { 121 public String toString() {
103 return MoreObjects.toStringHelper(this) 122 return MoreObjects.toStringHelper(this)
123 + .add("time", time)
104 .add("id", id) 124 .add("id", id)
105 .add("type", type) 125 .add("type", type)
106 .add("sender", sender) 126 .add("sender", sender)
......
...@@ -23,6 +23,7 @@ import io.netty.handler.codec.ReplayingDecoder; ...@@ -23,6 +23,7 @@ import io.netty.handler.codec.ReplayingDecoder;
23 23
24 import org.onlab.packet.IpAddress; 24 import org.onlab.packet.IpAddress;
25 import org.onlab.packet.IpAddress.Version; 25 import org.onlab.packet.IpAddress.Version;
26 +import org.onosproject.core.HybridLogicalTime;
26 import org.onosproject.store.cluster.messaging.Endpoint; 27 import org.onosproject.store.cluster.messaging.Endpoint;
27 import org.onosproject.store.cluster.messaging.impl.InternalMessage.Status; 28 import org.onosproject.store.cluster.messaging.impl.InternalMessage.Status;
28 import org.slf4j.Logger; 29 import org.slf4j.Logger;
...@@ -39,6 +40,8 @@ public class MessageDecoder extends ReplayingDecoder<DecoderState> { ...@@ -39,6 +40,8 @@ public class MessageDecoder extends ReplayingDecoder<DecoderState> {
39 40
40 private final Logger log = LoggerFactory.getLogger(getClass()); 41 private final Logger log = LoggerFactory.getLogger(getClass());
41 42
43 + private long logicalTime;
44 + private long logicalCounter;
42 private long messageId; 45 private long messageId;
43 private int preamble; 46 private int preamble;
44 private Version ipVersion; 47 private Version ipVersion;
...@@ -63,6 +66,12 @@ public class MessageDecoder extends ReplayingDecoder<DecoderState> { ...@@ -63,6 +66,12 @@ public class MessageDecoder extends ReplayingDecoder<DecoderState> {
63 switch (state()) { 66 switch (state()) {
64 case READ_MESSAGE_PREAMBLE: 67 case READ_MESSAGE_PREAMBLE:
65 preamble = buffer.readInt(); 68 preamble = buffer.readInt();
69 + checkpoint(DecoderState.READ_LOGICAL_TIME);
70 + case READ_LOGICAL_TIME:
71 + logicalTime = buffer.readLong();
72 + checkpoint(DecoderState.READ_LOGICAL_COUNTER);
73 + case READ_LOGICAL_COUNTER:
74 + logicalCounter = buffer.readLong();
66 checkpoint(DecoderState.READ_MESSAGE_ID); 75 checkpoint(DecoderState.READ_MESSAGE_ID);
67 case READ_MESSAGE_ID: 76 case READ_MESSAGE_ID:
68 messageId = buffer.readLong(); 77 messageId = buffer.readLong();
...@@ -102,6 +111,7 @@ public class MessageDecoder extends ReplayingDecoder<DecoderState> { ...@@ -102,6 +111,7 @@ public class MessageDecoder extends ReplayingDecoder<DecoderState> {
102 payload = new byte[0]; 111 payload = new byte[0];
103 } 112 }
104 InternalMessage message = new InternalMessage(preamble, 113 InternalMessage message = new InternalMessage(preamble,
114 + new HybridLogicalTime(logicalTime, logicalCounter),
105 messageId, 115 messageId,
106 new Endpoint(senderIp, senderPort), 116 new Endpoint(senderIp, senderPort),
107 messageType, 117 messageType,
......
...@@ -51,6 +51,10 @@ public class MessageEncoder extends MessageToByteEncoder<InternalMessage> { ...@@ -51,6 +51,10 @@ public class MessageEncoder extends MessageToByteEncoder<InternalMessage> {
51 51
52 out.writeInt(this.preamble); 52 out.writeInt(this.preamble);
53 53
54 + // write time
55 + out.writeLong(message.time().logicalTime());
56 + out.writeLong(message.time().logicalCounter());
57 +
54 // write message id 58 // write message id
55 out.writeLong(message.id()); 59 out.writeLong(message.id());
56 60
......
...@@ -53,6 +53,7 @@ import org.apache.felix.scr.annotations.Service; ...@@ -53,6 +53,7 @@ import org.apache.felix.scr.annotations.Service;
53 import org.onlab.util.Tools; 53 import org.onlab.util.Tools;
54 import org.onosproject.cluster.ClusterMetadataService; 54 import org.onosproject.cluster.ClusterMetadataService;
55 import org.onosproject.cluster.ControllerNode; 55 import org.onosproject.cluster.ControllerNode;
56 +import org.onosproject.core.HybridLogicalClockService;
56 import org.onosproject.store.cluster.messaging.Endpoint; 57 import org.onosproject.store.cluster.messaging.Endpoint;
57 import org.onosproject.store.cluster.messaging.MessagingException; 58 import org.onosproject.store.cluster.messaging.MessagingException;
58 import org.onosproject.store.cluster.messaging.MessagingService; 59 import org.onosproject.store.cluster.messaging.MessagingService;
...@@ -99,6 +100,9 @@ public class NettyMessagingManager implements MessagingService { ...@@ -99,6 +100,9 @@ public class NettyMessagingManager implements MessagingService {
99 100
100 private static final String REPLY_MESSAGE_TYPE = "NETTY_MESSAGING_REQUEST_REPLY"; 101 private static final String REPLY_MESSAGE_TYPE = "NETTY_MESSAGING_REQUEST_REPLY";
101 102
103 + @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
104 + protected HybridLogicalClockService clockService;
105 +
102 private Endpoint localEp; 106 private Endpoint localEp;
103 private int preamble; 107 private int preamble;
104 private final AtomicBoolean started = new AtomicBoolean(false); 108 private final AtomicBoolean started = new AtomicBoolean(false);
...@@ -218,6 +222,7 @@ public class NettyMessagingManager implements MessagingService { ...@@ -218,6 +222,7 @@ public class NettyMessagingManager implements MessagingService {
218 public CompletableFuture<Void> sendAsync(Endpoint ep, String type, byte[] payload) { 222 public CompletableFuture<Void> sendAsync(Endpoint ep, String type, byte[] payload) {
219 checkPermission(CLUSTER_WRITE); 223 checkPermission(CLUSTER_WRITE);
220 InternalMessage message = new InternalMessage(preamble, 224 InternalMessage message = new InternalMessage(preamble,
225 + clockService.timeNow(),
221 messageIdGenerator.incrementAndGet(), 226 messageIdGenerator.incrementAndGet(),
222 localEp, 227 localEp,
223 type, 228 type,
...@@ -264,7 +269,12 @@ public class NettyMessagingManager implements MessagingService { ...@@ -264,7 +269,12 @@ public class NettyMessagingManager implements MessagingService {
264 Callback callback = new Callback(response, executor); 269 Callback callback = new Callback(response, executor);
265 Long messageId = messageIdGenerator.incrementAndGet(); 270 Long messageId = messageIdGenerator.incrementAndGet();
266 callbacks.put(messageId, callback); 271 callbacks.put(messageId, callback);
267 - InternalMessage message = new InternalMessage(preamble, messageId, localEp, type, payload); 272 + InternalMessage message = new InternalMessage(preamble,
273 + clockService.timeNow(),
274 + messageId,
275 + localEp,
276 + type,
277 + payload);
268 return sendAsync(ep, message).whenComplete((r, e) -> { 278 return sendAsync(ep, message).whenComplete((r, e) -> {
269 if (e != null) { 279 if (e != null) {
270 callbacks.invalidate(messageId); 280 callbacks.invalidate(messageId);
...@@ -502,6 +512,7 @@ public class NettyMessagingManager implements MessagingService { ...@@ -502,6 +512,7 @@ public class NettyMessagingManager implements MessagingService {
502 log.debug("Received {} with invalid preamble from {}", message.type(), message.sender()); 512 log.debug("Received {} with invalid preamble from {}", message.type(), message.sender());
503 sendReply(message, Status.PROTOCOL_EXCEPTION, Optional.empty()); 513 sendReply(message, Status.PROTOCOL_EXCEPTION, Optional.empty());
504 } 514 }
515 + clockService.recordEventTime(message.time());
505 String type = message.type(); 516 String type = message.type();
506 if (REPLY_MESSAGE_TYPE.equals(type)) { 517 if (REPLY_MESSAGE_TYPE.equals(type)) {
507 try { 518 try {
...@@ -538,6 +549,7 @@ public class NettyMessagingManager implements MessagingService { ...@@ -538,6 +549,7 @@ public class NettyMessagingManager implements MessagingService {
538 549
539 private void sendReply(InternalMessage message, Status status, Optional<byte[]> responsePayload) { 550 private void sendReply(InternalMessage message, Status status, Optional<byte[]> responsePayload) {
540 InternalMessage response = new InternalMessage(preamble, 551 InternalMessage response = new InternalMessage(preamble,
552 + clockService.timeNow(),
541 message.id(), 553 message.id(),
542 localEp, 554 localEp,
543 REPLY_MESSAGE_TYPE, 555 REPLY_MESSAGE_TYPE,
......
...@@ -21,10 +21,12 @@ import java.util.concurrent.CountDownLatch; ...@@ -21,10 +21,12 @@ import java.util.concurrent.CountDownLatch;
21 import java.util.concurrent.ExecutorService; 21 import java.util.concurrent.ExecutorService;
22 import java.util.concurrent.Executors; 22 import java.util.concurrent.Executors;
23 import java.util.concurrent.atomic.AtomicBoolean; 23 import java.util.concurrent.atomic.AtomicBoolean;
24 +import java.util.concurrent.atomic.AtomicLong;
24 import java.util.concurrent.atomic.AtomicReference; 25 import java.util.concurrent.atomic.AtomicReference;
25 import java.util.function.BiFunction; 26 import java.util.function.BiFunction;
26 27
27 import com.google.common.collect.Sets; 28 import com.google.common.collect.Sets;
29 +
28 import org.junit.After; 30 import org.junit.After;
29 import org.junit.Before; 31 import org.junit.Before;
30 import org.junit.Test; 32 import org.junit.Test;
...@@ -34,6 +36,8 @@ import org.onosproject.cluster.ClusterMetadataEventListener; ...@@ -34,6 +36,8 @@ import org.onosproject.cluster.ClusterMetadataEventListener;
34 import org.onosproject.cluster.ClusterMetadataService; 36 import org.onosproject.cluster.ClusterMetadataService;
35 import org.onosproject.cluster.ControllerNode; 37 import org.onosproject.cluster.ControllerNode;
36 import org.onosproject.cluster.NodeId; 38 import org.onosproject.cluster.NodeId;
39 +import org.onosproject.core.HybridLogicalClockService;
40 +import org.onosproject.core.HybridLogicalTime;
37 import org.onosproject.net.provider.ProviderId; 41 import org.onosproject.net.provider.ProviderId;
38 import org.onosproject.store.cluster.messaging.Endpoint; 42 import org.onosproject.store.cluster.messaging.Endpoint;
39 43
...@@ -48,6 +52,18 @@ import static org.onlab.junit.TestTools.findAvailablePort; ...@@ -48,6 +52,18 @@ import static org.onlab.junit.TestTools.findAvailablePort;
48 */ 52 */
49 public class NettyMessagingManagerTest { 53 public class NettyMessagingManagerTest {
50 54
55 + HybridLogicalClockService testClockService = new HybridLogicalClockService() {
56 + AtomicLong counter = new AtomicLong();
57 + @Override
58 + public HybridLogicalTime timeNow() {
59 + return new HybridLogicalTime(counter.incrementAndGet(), 0);
60 + }
61 +
62 + @Override
63 + public void recordEventTime(HybridLogicalTime time) {
64 + }
65 + };
66 +
51 NettyMessagingManager netty1; 67 NettyMessagingManager netty1;
52 NettyMessagingManager netty2; 68 NettyMessagingManager netty2;
53 69
...@@ -63,11 +79,13 @@ public class NettyMessagingManagerTest { ...@@ -63,11 +79,13 @@ public class NettyMessagingManagerTest {
63 ep1 = new Endpoint(IpAddress.valueOf("127.0.0.1"), findAvailablePort(5001)); 79 ep1 = new Endpoint(IpAddress.valueOf("127.0.0.1"), findAvailablePort(5001));
64 netty1 = new NettyMessagingManager(); 80 netty1 = new NettyMessagingManager();
65 netty1.clusterMetadataService = dummyMetadataService(DUMMY_NAME, IP_STRING, ep1); 81 netty1.clusterMetadataService = dummyMetadataService(DUMMY_NAME, IP_STRING, ep1);
82 + netty1.clockService = testClockService;
66 netty1.activate(); 83 netty1.activate();
67 84
68 ep2 = new Endpoint(IpAddress.valueOf("127.0.0.1"), findAvailablePort(5003)); 85 ep2 = new Endpoint(IpAddress.valueOf("127.0.0.1"), findAvailablePort(5003));
69 netty2 = new NettyMessagingManager(); 86 netty2 = new NettyMessagingManager();
70 netty2.clusterMetadataService = dummyMetadataService(DUMMY_NAME, IP_STRING, ep2); 87 netty2.clusterMetadataService = dummyMetadataService(DUMMY_NAME, IP_STRING, ep2);
88 + netty2.clockService = testClockService;
71 netty2.activate(); 89 netty2.activate();
72 } 90 }
73 91
......