Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Minor enhancement on RedisRegistrationStore #402

Merged
merged 4 commits into from
Nov 10, 2017
Merged

Minor enhancement on RedisRegistrationStore #402

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
e581476
Do not remove registration by endpoint, if no registration by Id.
sbernard31 Oct 9, 2017
6cad580
Deserialize registration in protected block (after lock)
sbernard31 Oct 9, 2017
02d5190
Add clemency for registration expiration to redisStore
sbernard31 Oct 9, 2017
3b8e7a9
remove isAlive test on getRegistrationByEndpoint in redisStore
sbernard31 Oct 9, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
91 changes: 50 additions & 41 deletions ...erver-cluster/src/main/java/org/eclipse/leshan/server/cluster/RedisRegistrationStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,11 @@
*/
public class RedisRegistrationStore implements CaliforniumRegistrationStore, Startable, Stoppable {

/** Default time in seconds between 2 cleaning tasks (used to remove expired registration). */
public static final long DEFAULT_CLEAN_PERIOD = 60;
/** Defaut Extra time for registration lifetime in seconds */
public static final long DEFAULT_GRACE_PERIOD = 0;

private static final Logger LOG = LoggerFactory.getLogger(RedisRegistrationStore.class);

// Redis key prefixes
Expand All @@ -78,21 +83,24 @@ public class RedisRegistrationStore implements CaliforniumRegistrationStore, Sta

private final ScheduledExecutorService schedExecutor;
private final long cleanPeriod; // in seconds
private final long gracePeriod; // in seconds

public RedisRegistrationStore(Pool<Jedis> p) {
this(p, 60); // default clean period 60s
this(p, DEFAULT_CLEAN_PERIOD, DEFAULT_GRACE_PERIOD); // default clean period 60s
}

public RedisRegistrationStore(Pool<Jedis> p, long cleanPeriodInSec) {
public RedisRegistrationStore(Pool<Jedis> p, long cleanPeriodInSec, long lifetimeGracePeriodInSec) {
this(p, Executors.newScheduledThreadPool(1,
new NamedThreadFactory(String.format("RedisRegistrationStore Cleaner (%ds)", cleanPeriodInSec))),
cleanPeriodInSec);
cleanPeriodInSec, lifetimeGracePeriodInSec);
}

public RedisRegistrationStore(Pool<Jedis> p, ScheduledExecutorService schedExecutor, long cleanPeriodInSec) {
public RedisRegistrationStore(Pool<Jedis> p, ScheduledExecutorService schedExecutor, long cleanPeriodInSec,
long lifetimeGracePeriodInSec) {
this.pool = p;
this.schedExecutor = schedExecutor;
this.cleanPeriod = cleanPeriodInSec;
this.gracePeriod = lifetimeGracePeriodInSec;
}

/* *************** Redis Key utility function **************** */
Expand Down Expand Up @@ -163,19 +171,19 @@ public UpdatedRegistration updateRegistration(RegistrationUpdate update) {
return null;
}

// fetch the client
byte[] data = j.get(toEndpointKey(ep));
if (data == null) {
return null;
}

Registration r = deserializeReg(data);

byte[] lockValue = null;
byte[] lockKey = toLockKey(r.getEndpoint());
byte[] lockKey = toLockKey(ep);
try {
lockValue = RedisLock.acquire(j, lockKey);

// fetch the client
byte[] data = j.get(toEndpointKey(ep));
if (data == null) {
return null;
}

Registration r = deserializeReg(data);

Registration updatedRegistration = update.update(r);

// store the new client
Expand Down Expand Up @@ -204,8 +212,7 @@ public Registration getRegistrationByEndpoint(String endpoint) {
if (data == null) {
return null;
}
Registration r = deserializeReg(data);
return r.isAlive() ? r : null;
return deserializeReg(data);
}
}

Expand Down Expand Up @@ -292,37 +299,38 @@ public void remove() {
@Override
public Deregistration removeRegistration(String registrationId) {
try (Jedis j = pool.getResource()) {
return removeRegistration(j, registrationId, false);
}
}

byte[] regKey = toRegIdKey(registrationId);
private Deregistration removeRegistration(Jedis j, String registrationId, boolean removeOnlyIfNotAlive) {
// fetch the client ep by registration ID index
byte[] ep = j.get(toRegIdKey(registrationId));
if (ep == null) {
return null;
}

// fetch the client ep by registration ID index
byte[] ep = j.get(regKey);
if (ep == null) {
return null;
}
byte[] lockValue = null;
byte[] lockKey = toLockKey(ep);
try {
lockValue = RedisLock.acquire(j, lockKey);

// fetch the client
byte[] data = j.get(toEndpointKey(ep));
if (data == null) {
return null;
}

Registration r = deserializeReg(data);
deleteRegistration(j, r);
Collection<Observation> obsRemoved = unsafeRemoveAllObservations(j, r.getId());
return new Deregistration(r, obsRemoved);
}
}

private void deleteRegistration(Jedis j, Registration r) {
byte[] lockValue = null;
byte[] lockKey = toLockKey(r.getEndpoint());
try {
lockValue = RedisLock.acquire(j, lockKey);

// delete all entries
j.del(toRegIdKey(r.getId()));
j.del(toEndpointKey(r.getEndpoint()));

if (!removeOnlyIfNotAlive || !r.isAlive(gracePeriod)) {
long nbRemoved = j.del(toRegIdKey(r.getId()));
if (nbRemoved > 0) {
j.del(toEndpointKey(r.getEndpoint()));
Collection<Observation> obsRemoved = unsafeRemoveAllObservations(j, r.getId());
return new Deregistration(r, obsRemoved);
}
}
return null;
} finally {
RedisLock.release(j, lockKey, lockValue);
}
Expand Down Expand Up @@ -669,13 +677,14 @@ public void run() {
ScanParams params = new ScanParams().match(REG_EP + "*").count(100);
String cursor = "0";
do {
// TODO we probably need a lock here
ScanResult<byte[]> res = j.scan(cursor.getBytes(), params);
for (byte[] key : res.getResult()) {
Registration r = deserializeReg(j.get(key));
if (!r.isAlive()) {
deleteRegistration(j, r);
expirationListener.registrationExpired(r, new ArrayList<Observation>());
if (!r.isAlive(gracePeriod)) {
Deregistration dereg = removeRegistration(j, r.getId(), true);
if (dereg != null)
expirationListener.registrationExpired(dereg.getRegistration(),
dereg.getObservations());
}
}
cursor = res.getStringCursor();
Expand Down
14 changes: 13 additions & 1 deletion leshan-server-core/src/main/java/org/eclipse/leshan/server/registration/Registration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -261,8 +261,20 @@ public Date getLastUpdate() {
return lastUpdate;
}

/**
* @return true if the last registration update was done less than lifetime seconds ago.
*/
public boolean isAlive() {
return lastUpdate.getTime() + lifeTimeInSec * 1000 > System.currentTimeMillis();
return isAlive(0);
}

/**
* This is the same idea than {@link Registration#isAlive()} but with a grace period. <br/>
*
* @return true if the last registration update was done less than lifetime+gracePeriod seconds ago.
*/
public boolean isAlive(long gracePeriodInSec) {
return lastUpdate.getTime() + lifeTimeInSec * 1000 + gracePeriodInSec * 1000 > System.currentTimeMillis();
}

public Map<String, String> getAdditionalRegistrationAttributes() {
Expand Down